Prototype devices updates

CCDB/src/runConditionsServer.cxx

@@ -49,9 +49,8 @@ int main(int argc, char** argv)
             ("output-type", value<string>(&outputType)->default_value("ROOT"), "Output file type");
 
         config.AddToCmdLineOptions(serverOptions);
+        config.ParseAll(argc, argv);
 
-       config.ParseAll(argc, argv);
-
         string file = config.GetValue<string>("config-json-file");
         string id = config.GetValue<string>("id");
 

Examples/flp2epn-distributed/CMakeLists.txt

@@ -51,20 +51,33 @@ if(FAIRMQ_DEPENDENCIES)
     ${DEPENDENCIES}
     ${CMAKE_THREAD_LIBS_INIT}
     ${FAIRMQ_DEPENDENCIES}
+    ${Boost_RANDOM_LIBRARY}
+    ${Boost_CHRONO_LIBRARY}
+    ${Boost_REGEX_LIBRARY}
     FairMQ
   )
 else(FAIRMQ_DEPENDENCIES)
   set(DEPENDENCIES
     ${DEPENDENCIES}
     ${CMAKE_THREAD_LIBS_INIT}
-    ${Boost_DATE_TIME_LIBRARY} ${Boost_THREAD_LIBRARY} ${Boost_THREAD_LIBRARY} ${Boost_SYSTEM_LIBRARY} ${Boost_PROGRAM_OPTIONS_LIBRARY} ${Boost_CHRONO_LIBRARY} FairMQ
+    ${Boost_DATE_TIME_LIBRARY}
+    ${Boost_THREAD_LIBRARY}
+    ${Boost_THREAD_LIBRARY}
+    ${Boost_SYSTEM_LIBRARY}
+    ${Boost_PROGRAM_OPTIONS_LIBRARY}
+    ${Boost_CHRONO_LIBRARY}
+    ${Boost_RANDOM_LIBRARY}
+    ${Boost_REGEX_LIBRARY}
+    FairMQ
   )
 endif(FAIRMQ_DEPENDENCIES)
 
 if(DDS_FOUND)
   set(DEPENDENCIES
     ${DEPENDENCIES}
-    dds-key-value-lib
+    ${DDS_INTERCOM_LIBRARY_SHARED}
+    ${DDS_PROTOCOL_LIBRARY_SHARED} # also link the two DDS dependency libraries to avoid linking issues on some osx systems
+    ${DDS_USER_DEFAULTS_LIBRARY_SHARED}
   )
 endif()
 

Examples/flp2epn-distributed/EPNReceiver.cxx

@@ -51,7 +51,7 @@ void EPNReceiver::PrintBuffer(const unordered_map<uint16_t, TFBuffer>& buffer) c
 
   for (auto& it : buffer) {
     string stars = "";
-    for (unsigned int j = 1; j <= (it.second).parts.size(); ++j) {
+    for (unsigned int j = 1; j <= (it.second).parts.Size(); ++j) {
       stars += "*";
     }
     LOG(INFO) << setw(4) << it.first << ": " << stars;
@@ -65,10 +65,6 @@ void EPNReceiver::DiscardIncompleteTimeframes()
     if ((boost::posix_time::microsec_clock::local_time() - (it->second).startTime).total_milliseconds() > fBufferTimeoutInMs) {
       LOG(WARN) << "Timeframe #" << it->first << " incomplete after " << fBufferTimeoutInMs << " milliseconds, discarding";
       fDiscardedSet.insert(it->first);
-      for (unsigned int i = 0; i < (it->second).parts.size(); ++i) {
-        (it->second).parts.at(i).reset();
-      }
-      it->second.parts.clear();
       fTimeframeBuffer.erase(it++);
       LOG(WARN) << "Number of discarded timeframes: " << fDiscardedSet.size();
     } else {
@@ -92,19 +88,17 @@ void EPNReceiver::Run()
   // f2eHeader* header; // holds the header of the currently arrived message.
   uint16_t id = 0; // holds the timeframe id of the currently arrived sub-timeframe.
 
-  FairMQChannel& dataInputChannel = fChannels.at("data-in").at(0);
-  FairMQChannel& dataOutChannel = fChannels.at("data-out").at(0);
   FairMQChannel& ackOutChannel = fChannels.at("ack-out").at(0);
 
   while (CheckCurrentState(RUNNING)) {
     poller->Poll(100);
 
     if (poller->CheckInput(0)) {
-      unique_ptr<FairMQMessage> headerPart(fTransportFactory->CreateMessage());
+      FairMQParts parts;
 
-      if (dataInputChannel.Receive(headerPart) > 0) {
+      if (Receive(parts, "data-in") > 0) {
         // store the received ID
-        f2eHeader& header = *(static_cast<f2eHeader*>(headerPart->GetData()));
+        f2eHeader& header = *(static_cast<f2eHeader*>(parts.At(0)->GetData()));
         id = header.timeFrameId;
         // LOG(INFO) << "Received sub-time frame #" << id << " from FLP" << header.flpIndex;
 
@@ -119,60 +113,41 @@ void EPNReceiver::Run()
         // }
         // end DEBUG
 
-        unique_ptr<FairMQMessage> dataPart(fTransportFactory->CreateMessage());
-
-        // receive the data part
-        if (dataInputChannel.Receive(dataPart) > 0)
+        if (fDiscardedSet.find(id) == fDiscardedSet.end())
         {
-          if (fDiscardedSet.find(id) == fDiscardedSet.end())
-          {
-            if (fTimeframeBuffer.find(id) == fTimeframeBuffer.end())
-            {
-              // if this is the first part with this ID, save the receive time.
-              fTimeframeBuffer[id].startTime = boost::posix_time::microsec_clock::local_time();
-            }
-            // if the received ID has not previously been discarded,
-            // store the data part in the buffer
-            fTimeframeBuffer[id].parts.push_back(move(dataPart));
-            // PrintBuffer(fTimeframeBuffer);
-          }
-          else
+          if (fTimeframeBuffer.find(id) == fTimeframeBuffer.end())
           {
-            // if received ID has been previously discarded.
-            LOG(WARN) << "Received part from an already discarded timeframe with id " << id;
+            // if this is the first part with this ID, save the receive time.
+            fTimeframeBuffer[id].startTime = boost::posix_time::microsec_clock::local_time();
           }
+          // if the received ID has not previously been discarded,
+          // store the data part in the buffer
+          fTimeframeBuffer[id].parts.AddPart(move(parts.At(1)));
+          // PrintBuffer(fTimeframeBuffer);
         }
         else
         {
-          LOG(ERROR) << "no data received from input socket";
+          // if received ID has been previously discarded.
+          LOG(WARN) << "Received part from an already discarded timeframe with id " << id;
         }
 
-        if (fTimeframeBuffer[id].parts.size() == fNumFLPs) {
+        if (fTimeframeBuffer[id].parts.Size() == fNumFLPs) {
           // LOG(INFO) << "Collected all parts for timeframe #" << id;
-          // when all parts are collected send all except last one with 'snd-more' flag, and last one without the flag.
-          for (int i = 0; i < fNumFLPs - 1; ++i) {
-            dataOutChannel.SendPart(fTimeframeBuffer[id].parts.at(i));
-          }
-          dataOutChannel.Send(fTimeframeBuffer[id].parts.at(fNumFLPs - 1));
+          // when all parts are collected send then to the output channel
+          Send(fTimeframeBuffer[id].parts, "data-out");
 
           if (fTestMode > 0) {
             // Send an acknowledgement back to the sampler to measure the round trip time
-            unique_ptr<FairMQMessage> ack(fTransportFactory->CreateMessage(sizeof(uint16_t)));
+            unique_ptr<FairMQMessage> ack(NewMessage(sizeof(uint16_t)));
             memcpy(ack->GetData(), &id, sizeof(uint16_t));
 
             if (ackOutChannel.SendAsync(ack) <= 0) {
               LOG(ERROR) << "Could not send acknowledgement without blocking";
             }
           }
 
-          // let transport know that the data is no longer needed. transport will clean up after it is sent out.
-          for (unsigned int i = 0; i < fTimeframeBuffer[id].parts.size(); ++i) {
-            fTimeframeBuffer[id].parts.at(i).reset();
-          }
-          fTimeframeBuffer[id].parts.clear();
-
           // fTimeframeBuffer[id].endTime = boost::posix_time::microsec_clock::local_time();
-          // do something with time here ...
+
           fTimeframeBuffer.erase(id);
         }
 
@@ -209,7 +184,7 @@ void EPNReceiver::sendHeartbeats()
   while (CheckCurrentState(RUNNING)) {
     try {
       for (int i = 0; i < fNumFLPs; ++i) {
-        unique_ptr<FairMQMessage> heartbeatMsg(fTransportFactory->CreateMessage(ownAddressSize));
+        unique_ptr<FairMQMessage> heartbeatMsg(NewMessage(ownAddressSize));
         memcpy(heartbeatMsg->GetData(), ownAddress.c_str(), ownAddressSize);
 
         fChannels.at("heartbeat-out").at(i).Send(heartbeatMsg);

Examples/flp2epn-distributed/EPNReceiver.h

@@ -23,7 +23,7 @@ namespace Devices {
 
 struct TFBuffer
 {
-  std::vector<std::unique_ptr<FairMQMessage>> parts;
+  FairMQParts parts;
   boost::posix_time::ptime startTime;
   boost::posix_time::ptime endTime;
 };

Examples/flp2epn-distributed/FLPSender.cxx

@@ -28,8 +28,7 @@ struct f2eHeader {
 };
 
 FLPSender::FLPSender()
-  : fHeaderBuffer()
-  , fDataBuffer()
+  : fSTFBuffer()
   , fArrivalTime()
   , fNumEPNs(0)
   , fIndex(0)
@@ -64,10 +63,10 @@ void FLPSender::receiveHeartbeats()
 
   while (CheckCurrentState(RUNNING)) {
     try {
-      unique_ptr<FairMQMessage> hbMsg(fTransportFactory->CreateMessage());
+      unique_ptr<FairMQMessage> heartbeat(NewMessage());
 
-      if (hbChannel.Receive(hbMsg) > 0) {
-        string address = string(static_cast<char*>(hbMsg->GetData()), hbMsg->GetSize());
+      if (hbChannel.Receive(heartbeat) > 0) {
+        string address = string(static_cast<char*>(heartbeat->GetData()), heartbeat->GetSize());
 
         if (fHeartbeats.find(address) != fHeartbeats.end()) {
             ptime now = boost::posix_time::microsec_clock::local_time();
@@ -97,7 +96,7 @@ void FLPSender::Run()
   // boost::thread heartbeatReceiver(boost::bind(&FLPSender::receiveHeartbeats, this));
 
   // base buffer, to be copied from for every timeframe body (zero-copy)
-  unique_ptr<FairMQMessage> baseMsg(fTransportFactory->CreateMessage(fEventSize));
+  unique_ptr<FairMQMessage> baseMsg(NewMessage(fEventSize));
 
   uint16_t timeFrameId = 0;
 
@@ -106,59 +105,55 @@ void FLPSender::Run()
 
   while (CheckCurrentState(RUNNING)) {
     // initialize f2e header
-    f2eHeader* h = new f2eHeader;
+    f2eHeader* header = new f2eHeader;
 
     if (fTestMode > 0) {
       // test-mode: receive and store id part in the buffer.
-      unique_ptr<FairMQMessage> idPart(fTransportFactory->CreateMessage());
-      if (dataInChannel.Receive(idPart) > 0) {
-        h->timeFrameId = *(static_cast<uint16_t*>(idPart->GetData()));
-        h->flpIndex = fIndex;
+      unique_ptr<FairMQMessage> id(NewMessage());
+      if (dataInChannel.Receive(id) > 0) {
+        header->timeFrameId = *(static_cast<uint16_t*>(id->GetData()));
+        header->flpIndex = fIndex;
       } else {
         // if nothing was received, try again
-        delete h;
+        delete header;
         continue;
       }
     } else {
       // regular mode: use the id generated locally
-      h->timeFrameId = timeFrameId;
-      h->flpIndex = fIndex;
+      header->timeFrameId = timeFrameId;
+      header->flpIndex = fIndex;
 
       if (++timeFrameId == UINT16_MAX - 1) {
         timeFrameId = 0;
       }
     }
 
-   // unique_ptr<FairMQMessage> headerPart(fTransportFactory->CreateMessage(sizeof(f2eHeader)));
-    unique_ptr<FairMQMessage> headerPart(fTransportFactory->CreateMessage(h, sizeof(f2eHeader), [](void* data, void* hint){ delete static_cast<f2eHeader*>(hint); }, h));
-    unique_ptr<FairMQMessage> dataPart(fTransportFactory->CreateMessage());
+    FairMQParts parts;
+
+    parts.AddPart(NewMessage(header, sizeof(f2eHeader), [](void* data, void* hint){ delete static_cast<f2eHeader*>(hint); }, header));
+    parts.AddPart(NewMessage());
 
     // save the arrival time of the message.
     fArrivalTime.push(boost::posix_time::microsec_clock::local_time());
 
     if (fTestMode > 0) {
       // test-mode: initialize and store data part in the buffer.
-      dataPart->Copy(baseMsg);
-      fHeaderBuffer.push(move(headerPart));
-      fDataBuffer.push(move(dataPart));
+      parts.At(1)->Copy(baseMsg);
+      fSTFBuffer.push(move(parts));
     } else {
       // regular mode: receive data part from input
-      if (dataInChannel.Receive(dataPart) >= 0) {
-        fHeaderBuffer.push(move(headerPart));
-        fDataBuffer.push(move(dataPart));
+      if (dataInChannel.Receive(parts.At(1)) >= 0) {
+        fSTFBuffer.push(move(parts));
       } else {
         // if nothing was received, try again
         continue;
       }
     }
 
-    // LOG(INFO) << fDataBuffer.size();
-
     // if offset is 0 - send data out without staggering.
-    if (fSendOffset == 0 && fDataBuffer.size() > 0) {
+    if (fSendOffset == 0 && fSTFBuffer.size() > 0) {
       sendFrontData();
-    } else if (fDataBuffer.size() > 0) {
-      // size_t dataSize = fDataBuffer.front()->GetSize();
+    } else if (fSTFBuffer.size() > 0) {
       ptime now = boost::posix_time::microsec_clock::local_time();
       if ((now - fArrivalTime.front()).total_milliseconds() >= (fSendDelay * fSendOffset)) {
         sendFrontData();
@@ -174,8 +169,8 @@ void FLPSender::Run()
 
 inline void FLPSender::sendFrontData()
 {
-  f2eHeader h = *(static_cast<f2eHeader*>(fHeaderBuffer.front()->GetData()));
-  uint16_t currentTimeframeId = h.timeFrameId;
+  f2eHeader header = *(static_cast<f2eHeader*>(fSTFBuffer.front().At(0)->GetData()));
+  uint16_t currentTimeframeId = header.timeFrameId;
 
   // for which EPN is the message?
   int direction = currentTimeframeId % fNumEPNs;
@@ -193,21 +188,14 @@ inline void FLPSender::sendFrontData()
   // if (to_simple_string(storedHeartbeat) == "not-a-date-time" ||
   //     (currentTime - storedHeartbeat).total_milliseconds() > fHeartbeatTimeoutInMs) {
   //   LOG(WARN) << "Heartbeat too old for EPN#" << direction << ", discarding message.";
-  //   fHeaderBuffer.pop();
+  //   fSTFBuffer.pop();
   //   fArrivalTime.pop();
-  //   fDataBuffer.pop();
   // } else { // if the heartbeat from the corresponding EPN is within timeout period, send the data.
-    if (fChannels.at("data-out").at(direction).SendPart(fHeaderBuffer.front()) < 0) {
-      // TODO: replace SendPart() with SendPartAsync() after nov15 fairroot release
-      LOG(ERROR) << "Failed to queue ID part of event #" << currentTimeframeId;
-    } else {
-      if (fChannels.at("data-out").at(direction).SendAsync(fDataBuffer.front()) < 0) {
-        LOG(ERROR) << "Could not send message with event #" << currentTimeframeId << " without blocking";
-      }
+    if (SendAsync(fSTFBuffer.front(), "data-out", direction) < 0) {
+      LOG(ERROR) << "Failed to queue sub-timeframe #" << currentTimeframeId;
     }
-    fHeaderBuffer.pop();
+    fSTFBuffer.pop();
     fArrivalTime.pop();
-    fDataBuffer.pop();
   // }
 }
 

Examples/flp2epn-distributed/FLPSender.h

@@ -75,8 +75,7 @@ class FLPSender : public FairMQDevice
     /// Sends the "oldest" element from the sub-timeframe container
     void sendFrontData();
 
-    std::queue<std::unique_ptr<FairMQMessage>> fHeaderBuffer; ///< Stores sub-timeframe headers
-    std::queue<std::unique_ptr<FairMQMessage>> fDataBuffer; ///< Stores sub-timeframe bodies
+    std::queue<FairMQParts> fSTFBuffer; ///< Buffer for sub-timeframes
     std::queue<boost::posix_time::ptime> fArrivalTime; ///< Stores arrival times of sub-timeframes
 
     int fNumEPNs; ///< Number of epnReceivers