Merge branch 'ironMann-shm__multipart_send'

Author: sy-c

src/ConsumerFMQchannel.cxx

@@ -55,6 +55,8 @@ class ConsumerFMQchannel: public Consumer {
     int memoryPoolNumberOfPages;
 
   public:
+    std::vector<FairMQMessagePtr> messagesToSend; // collect HBF messages of each update
+
 
 
   ConsumerFMQchannel(ConfigFile &cfg, std::string cfgEntryPoint) : Consumer(cfg,cfgEntryPoint) {
@@ -225,8 +227,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));
 
 
@@ -252,8 +254,8 @@ class ConsumerFMQchannel: public Consumer {
         //printf("allocating blockRef %p for %p\n",pf.blockRef,br);
         pendingFrames.push_back(pf);
     };
-
-    auto pendingFramesSend = [&]() {
+    
+    auto pendingFramesCollect = [&]() {
       int nFrames=pendingFrames.size();
 
       if (nFrames==0) {
@@ -271,8 +273,10 @@ 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);
 
@@ -319,10 +323,9 @@ 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);
       }
@@ -346,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;
@@ -361,8 +364,14 @@ class ConsumerFMQchannel: public Consumer {
     }
 
     // purge pendingFrames
-    pendingFramesSend();
-
+    pendingFramesCollect();
+
+    // send all the messages
+    if (sendingChannel->Send(messagesToSend) >= 0)
+      messagesToSend.clear();
+    else
+      LOG(ERROR) << "Sending failed!";
+    
     /*
 
       std::unique_ptr<FairMQMessage> msgBody(transportFactory->CreateMessage((void *)(&b->data[HBstart]), (size_t)(HBlength), cleanupCallback, (void *)(ptr)));

src/testRxFMQ.cxx

@@ -2,6 +2,7 @@
 #include <fairmq/FairMQMessage.h>
 #include <fairmq/FairMQTransportFactory.h>
 #include <memory>
+#include <vector>
 
 int main() {
 
@@ -13,18 +14,22 @@ int main() {
   auto factory = FairMQTransportFactory::CreateTransportFactory(cfgTransportType);
   auto pull = FairMQChannel{cfgChannelName, cfgChannelType, factory};
   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

@@ -16,7 +16,7 @@ int main() {
   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) {
@@ -26,13 +26,23 @@ int main() {
   );
   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;