Switch to FairMQ SHM multi-part to implement the O2 interface

Performance optimization change only, the interface is unchanged

Author: ironMann

src/ConsumerFMQchannel.cxx

@@ -55,6 +55,8 @@ class ConsumerFMQchannel: public Consumer {
     int memoryPoolPageSize;
     int memoryPoolNumberOfPages;
 
+    std::vector<FairMQMessagePtr> messagesToSend; // collect HBF messages of each update
+
   public: 
 
 
@@ -226,8 +228,8 @@ class ConsumerFMQchannel: public Consumer {
 
     // create a header message
     //std::unique_ptr<FairMQMessage> msgHeader(transportFactory->CreateMessage((void *)stfHeader, sizeof(SubTimeframe), cleanupCallback, (void *)(blockRef)));
-    std::unique_ptr<FairMQMessage> msgHeader(transportFactory->CreateMessage(memoryBuffer,(void *)stfHeader, sizeof(SubTimeframe), (void *)(blockRef)));
-    sendingChannel->Send(msgHeader);
+    assert(messagesToSend.empty());
+    messagesToSend.emplace_back(std::move(sendingChannel->NewMessage(memoryBuffer,(void *)stfHeader, sizeof(SubTimeframe), (void *)(blockRef))));
     //printf("sent header %d bytes\n",(int)sizeof(SubTimeframe));
 
 
@@ -254,7 +256,7 @@ class ConsumerFMQchannel: public Consumer {
         pendingFrames.push_back(pf);       
     };
 
-    auto pendingFramesSend = [&]() {
+    auto pendingFramesCollect = [&]() {
       int nFrames=pendingFrames.size();
 
       if (nFrames==0) {
@@ -272,8 +274,9 @@ class ConsumerFMQchannel: public Consumer {
         void *hint=(void *)pendingFrames[0].blockRef;
         //printf("block %p ix = %d : %d hint=%p\n",(void *)(&(b->data[ix])),ix,l,hint);
         //std::cout << typeid(pendingFrames[0].blockRef).name() << std::endl;
-        std::unique_ptr<FairMQMessage> msgBody(transportFactory->CreateMessage(memoryBuffer,(void *)(&(b->data[ix])),(size_t)(l),hint));
-        sendingChannel->Send(msgBody);        
+
+        // create and queue a fmq message
+        messagesToSend.emplace_back(std::move(sendingChannel->NewMessage(memoryBuffer,(void *)(&(b->data[ix])),(size_t)(l),hint)));
         //printf("sent single HB %d = %d bytes\n",pendingFrames[0].HBid,l);
         //printf("left to FMQ: %p\n",pendingFrames[0].blockRef);
 
@@ -321,9 +324,8 @@ class ConsumerFMQchannel: public Consumer {
         //std::unique_ptr<FairMQMessage> msgBody(transportFactory->CreateMessage((void *)newBlock, totalSize, cleanupCallbackForMalloc, (void *)(newBlock)));
         //sendingChannel->Send(msgBody);
 
-        // create a fmq message
-        std::unique_ptr<FairMQMessage> msgCopy(transportFactory->CreateMessage(memoryBuffer,(void *)newBlock, totalSize, (void *)(blockRef)));
-        sendingChannel->Send(msgCopy);
+        // create and queue a fmq message
+        messagesToSend.emplace_back(std::move(sendingChannel->NewMessage(memoryBuffer,(void *)newBlock, totalSize, (void *)(blockRef))));
 
         //printf("sent reallocated HB %d (originally %d blocks) = %d bytes\n",pendingFrames[0].HBid,nFrames,totalSize);
       }
@@ -347,7 +349,7 @@ class ConsumerFMQchannel: public Consumer {
             pendingFramesAppend(HBstart,HBlength,lastHBid,br);
           }
           // send pending frames, if any
-          pendingFramesSend();
+          pendingFramesCollect();
 
           // update new HB frame
           HBstart=offset;
@@ -362,7 +364,13 @@ class ConsumerFMQchannel: public Consumer {
     }
 
     // purge pendingFrames
-    pendingFramesSend();
+    pendingFramesCollect();
+
+    // send all the messages
+    if (sendingChannel->Send(messagesToSend) >= 0)
+      messagesToSend.clear();
+    else
+      LOG(ERROR) << "Sending failed!";
 
     /*
 

src/testRxFMQ.cxx

@@ -3,6 +3,7 @@
 #include <fairmq/FairMQTransportFactory.h>
 #include <fairmq/zeromq/FairMQTransportFactoryZMQ.h>
 #include <memory>
+#include <vector>
 
 int main() {
 
@@ -13,19 +14,23 @@ int main() {
 
   auto factory = FairMQTransportFactory::CreateTransportFactory(cfgTransportType);
   auto pull = FairMQChannel{cfgChannelName, cfgChannelType, factory};
-  pull.Connect(cfgChannelAddress);  
+  pull.Connect(cfgChannelAddress);
+  int64_t ret;
 
   for (;;) {
-    auto msg = pull.NewMessage();
-    if (pull.ReceiveAsync(msg)>0) {
-      if (msg->GetSize()==0) {continue;}
-      int sz=(int)msg->GetSize();
-      void *data=msg->GetData();
-      printf("Received message %p size %d\n",data,sz);
-      sleep(1);
-      printf("Releasing message %p size %d\n",data,sz);
+    std::vector<FairMQMessagePtr> msgs;
+
+    ret = pull.Receive(msgs);
+    if (ret > 0) {
+      for (auto &msg : msgs) {
+        int sz=(int)msg->GetSize();
+        void *data=msg->GetData();
+        printf("Received message %p size %d\n",data,sz);
+        printf("Releasing message %p size %d\n",data,sz);
+      }
     } else {
-      usleep(1000);
+      printf("Error while receiving messages %d\n", (int)ret);
+      return -1;
     }
   }
 

src/testTxFMQ.cxx

@@ -11,29 +11,39 @@ int main() {
   std::string cfgChannelType="pair";
   std::string cfgChannelAddress="ipc:///tmp/test-pipe";
 
-       
+
   auto transportFactory=FairMQTransportFactory::CreateTransportFactory(cfgTransportType);
   auto channel=FairMQChannel{cfgChannelName, cfgChannelType, transportFactory};
   channel.Bind(cfgChannelAddress);
   if (!channel.ValidateChannel()) { return -1; }
-    
-  int bufferSize=100*1024*1024;
+
+  const size_t bufferSize=100*1024*1024;
   auto memoryBuffer=channel.Transport()->CreateUnmanagedRegion(
     bufferSize,
     [](void* data, size_t size, void* hint) {
       // cleanup callback
-      printf("ack %p (size %d) hint=%p\n",data,(int)size,hint);                                                           
+      printf("ack %p (size %d) hint=%p\n",data,(int)size,hint);
     }
   );
   printf("Created buffer %p size %ld\n",memoryBuffer->GetData(),memoryBuffer->GetSize());
   size_t msgSize=100;
-  for (int ix=0;ix<bufferSize;ix+=msgSize) {
-    void *dataPtr=(void *)(&((char *)memoryBuffer->GetData())[ix]);
-    void *hint=(void *)ix;
-    std::unique_ptr<FairMQMessage> msg(transportFactory->CreateMessage(memoryBuffer,dataPtr,msgSize,hint));
-    printf("send %p : %ld bytes hint=%p\n",dataPtr,msgSize,hint);
-    channel.Send(msg);
-    usleep(3000000);
+
+  size_t ix=0;
+  while (ix<(bufferSize-msgSize)) {
+    // send random number of messages in one multipart [1-50]
+    int nmsgs = (rand()%50)+1;
+    nmsgs = std::min(nmsgs, (int)((bufferSize-ix)/msgSize));
+
+    std::vector<FairMQMessagePtr> msgs;
+    for (int im=0; im < nmsgs; im++, ix+=msgSize) {
+      void *dataPtr=(void *)(&((char *)memoryBuffer->GetData())[ix]);
+      void *hint=(void *)ix;
+      msgs.emplace_back(transportFactory->CreateMessage(memoryBuffer,dataPtr,msgSize,hint));
+      printf("send %p : %ld bytes hint=%p\n",dataPtr,msgSize,hint);
+    }
+    printf("* sending %lu messages\n",msgs.size());
+    channel.Send(msgs);
+    sleep(2);
   }
 
   return 0;