From c049e906e550a5318a38c73dfd6316d98d0142d2 Mon Sep 17 00:00:00 2001 From: Alexey Rybalchenko Date: Tue, 22 Mar 2016 11:43:50 +0100 Subject: [PATCH] Update prototype devices - Use the new Multipart API for the prototype devices. - Update the FLP/EPN examples to use the new configuration system. - Update the DDS calls with the API (dds::intercom_api instead of dds::key_value). - Remove obsolete Merger device. --- CCDB/src/runConditionsServer.cxx | 3 +- Examples/flp2epn-distributed/CMakeLists.txt | 17 +- Examples/flp2epn-distributed/EPNReceiver.cxx | 65 ++--- Examples/flp2epn-distributed/EPNReceiver.h | 2 +- Examples/flp2epn-distributed/FLPSender.cxx | 70 +++--- Examples/flp2epn-distributed/FLPSender.h | 3 +- .../run/flp2epn-prototype.json | 229 ++++++++++++++++++ .../run/runEPNReceiver.cxx | 11 +- .../flp2epn-distributed/run/runFLPSender.cxx | 9 +- .../run/runFLPSyncSampler.cxx | 11 +- .../run/startFLP2EPN-distributed.sh.in | 25 +- .../runO2Prototype/runEPNReceiver.cxx | 6 +- .../runO2Prototype/runFLPSender.cxx | 8 +- .../runO2Prototype/runFLPSyncSampler.cxx | 4 +- Examples/flp2epn/CMakeLists.txt | 22 +- Examples/flp2epn/O2EPNex.cxx | 17 +- Examples/flp2epn/O2EpnMerger.cxx | 48 ---- Examples/flp2epn/O2EpnMerger.h | 23 -- Examples/flp2epn/O2FLPExContent.h | 13 + Examples/flp2epn/O2FLPex.cxx | 32 +-- Examples/flp2epn/O2FLPex.h | 6 +- Examples/flp2epn/O2Merger.cxx | 50 ---- Examples/flp2epn/O2Merger.h | 24 -- Examples/flp2epn/O2Proxy.cxx | 47 ---- Examples/flp2epn/O2Proxy.h | 24 -- Examples/flp2epn/README.md | 15 ++ Examples/flp2epn/run/flp2epn.json | 65 +++++ Examples/flp2epn/run/runEPN.cxx | 108 ++------- Examples/flp2epn/run/runEPN_M.cxx | 121 --------- Examples/flp2epn/run/runFLP.cxx | 117 ++------- Examples/flp2epn/run/runMerger.cxx | 148 ----------- Examples/flp2epn/run/runProxy.cxx | 149 ++---------- Examples/flp2epn/run/startFLP2EPN.sh.in | 28 ++- Examples/flp2epn/run/startMerger.sh.in | 57 ----- Utilities/aliceHLTwrapper/CMakeLists.txt | 11 +- Utilities/aliceHLTwrapper/EventSampler.cxx | 25 +- Utilities/aliceHLTwrapper/WrapperDevice.cxx | 54 ++--- Utilities/aliceHLTwrapper/WrapperDevice.h | 2 +- .../aliceHLTwrapper/aliceHLTEventSampler.cxx | 10 +- Utilities/aliceHLTwrapper/aliceHLTWrapper.cxx | 10 +- 40 files changed, 585 insertions(+), 1104 deletions(-) create mode 100644 Examples/flp2epn-distributed/run/flp2epn-prototype.json delete mode 100644 Examples/flp2epn/O2EpnMerger.cxx delete mode 100644 Examples/flp2epn/O2EpnMerger.h create mode 100644 Examples/flp2epn/O2FLPExContent.h delete mode 100644 Examples/flp2epn/O2Merger.cxx delete mode 100644 Examples/flp2epn/O2Merger.h delete mode 100644 Examples/flp2epn/O2Proxy.cxx delete mode 100644 Examples/flp2epn/O2Proxy.h create mode 100644 Examples/flp2epn/README.md create mode 100644 Examples/flp2epn/run/flp2epn.json delete mode 100644 Examples/flp2epn/run/runEPN_M.cxx delete mode 100644 Examples/flp2epn/run/runMerger.cxx delete mode 100755 Examples/flp2epn/run/startMerger.sh.in diff --git a/CCDB/src/runConditionsServer.cxx b/CCDB/src/runConditionsServer.cxx index b184d2cd10fa5..02e66e84a3ad0 100644 --- a/CCDB/src/runConditionsServer.cxx +++ b/CCDB/src/runConditionsServer.cxx @@ -49,9 +49,8 @@ int main(int argc, char** argv) ("output-type", value(&outputType)->default_value("ROOT"), "Output file type"); config.AddToCmdLineOptions(serverOptions); + config.ParseAll(argc, argv); - config.ParseAll(argc, argv); - string file = config.GetValue("config-json-file"); string id = config.GetValue("id"); diff --git a/Examples/flp2epn-distributed/CMakeLists.txt b/Examples/flp2epn-distributed/CMakeLists.txt index 46a0b7e3675d7..f250f5979573d 100644 --- a/Examples/flp2epn-distributed/CMakeLists.txt +++ b/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() diff --git a/Examples/flp2epn-distributed/EPNReceiver.cxx b/Examples/flp2epn-distributed/EPNReceiver.cxx index 769c07aef57ce..1be729480a61a 100644 --- a/Examples/flp2epn-distributed/EPNReceiver.cxx +++ b/Examples/flp2epn-distributed/EPNReceiver.cxx @@ -51,7 +51,7 @@ void EPNReceiver::PrintBuffer(const unordered_map& 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 headerPart(fTransportFactory->CreateMessage()); + FairMQParts parts; - if (dataInputChannel.Receive(headerPart) > 0) { + if (Receive(parts, "data-in") > 0) { // store the received ID - f2eHeader& header = *(static_cast(headerPart->GetData())); + f2eHeader& header = *(static_cast(parts.At(0)->GetData())); id = header.timeFrameId; // LOG(INFO) << "Received sub-time frame #" << id << " from FLP" << header.flpIndex; @@ -119,45 +113,32 @@ void EPNReceiver::Run() // } // end DEBUG - unique_ptr 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 ack(fTransportFactory->CreateMessage(sizeof(uint16_t))); + unique_ptr ack(NewMessage(sizeof(uint16_t))); memcpy(ack->GetData(), &id, sizeof(uint16_t)); if (ackOutChannel.SendAsync(ack) <= 0) { @@ -165,14 +146,8 @@ void EPNReceiver::Run() } } - // 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 heartbeatMsg(fTransportFactory->CreateMessage(ownAddressSize)); + unique_ptr heartbeatMsg(NewMessage(ownAddressSize)); memcpy(heartbeatMsg->GetData(), ownAddress.c_str(), ownAddressSize); fChannels.at("heartbeat-out").at(i).Send(heartbeatMsg); diff --git a/Examples/flp2epn-distributed/EPNReceiver.h b/Examples/flp2epn-distributed/EPNReceiver.h index 7236a2eea5eac..a508c63d50587 100644 --- a/Examples/flp2epn-distributed/EPNReceiver.h +++ b/Examples/flp2epn-distributed/EPNReceiver.h @@ -23,7 +23,7 @@ namespace Devices { struct TFBuffer { - std::vector> parts; + FairMQParts parts; boost::posix_time::ptime startTime; boost::posix_time::ptime endTime; }; diff --git a/Examples/flp2epn-distributed/FLPSender.cxx b/Examples/flp2epn-distributed/FLPSender.cxx index dfe6fd65fa28f..2fb1d734b562b 100644 --- a/Examples/flp2epn-distributed/FLPSender.cxx +++ b/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 hbMsg(fTransportFactory->CreateMessage()); + unique_ptr heartbeat(NewMessage()); - if (hbChannel.Receive(hbMsg) > 0) { - string address = string(static_cast(hbMsg->GetData()), hbMsg->GetSize()); + if (hbChannel.Receive(heartbeat) > 0) { + string address = string(static_cast(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 baseMsg(fTransportFactory->CreateMessage(fEventSize)); + unique_ptr 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 idPart(fTransportFactory->CreateMessage()); - if (dataInChannel.Receive(idPart) > 0) { - h->timeFrameId = *(static_cast(idPart->GetData())); - h->flpIndex = fIndex; + unique_ptr id(NewMessage()); + if (dataInChannel.Receive(id) > 0) { + header->timeFrameId = *(static_cast(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 headerPart(fTransportFactory->CreateMessage(sizeof(f2eHeader))); - unique_ptr headerPart(fTransportFactory->CreateMessage(h, sizeof(f2eHeader), [](void* data, void* hint){ delete static_cast(hint); }, h)); - unique_ptr dataPart(fTransportFactory->CreateMessage()); + FairMQParts parts; + + parts.AddPart(NewMessage(header, sizeof(f2eHeader), [](void* data, void* hint){ delete static_cast(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(fHeaderBuffer.front()->GetData())); - uint16_t currentTimeframeId = h.timeFrameId; + f2eHeader header = *(static_cast(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(); // } } diff --git a/Examples/flp2epn-distributed/FLPSender.h b/Examples/flp2epn-distributed/FLPSender.h index 91fe98057a35d..cf9a65aad10c1 100644 --- a/Examples/flp2epn-distributed/FLPSender.h +++ b/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> fHeaderBuffer; ///< Stores sub-timeframe headers - std::queue> fDataBuffer; ///< Stores sub-timeframe bodies + std::queue fSTFBuffer; ///< Buffer for sub-timeframes std::queue fArrivalTime; ///< Stores arrival times of sub-timeframes int fNumEPNs; ///< Number of epnReceivers diff --git a/Examples/flp2epn-distributed/run/flp2epn-prototype.json b/Examples/flp2epn-distributed/run/flp2epn-prototype.json new file mode 100644 index 0000000000000..a4b49d09e2ef2 --- /dev/null +++ b/Examples/flp2epn-distributed/run/flp2epn-prototype.json @@ -0,0 +1,229 @@ +{ + "fairMQOptions": + { + "devices": + [{ + "id": "flpSyncSampler", + "channels": + [{ + "name": "data", + "type": "pub", + "method": "bind", + "address": "tcp://127.0.0.1:5550", + "rateLogging": "0" + }, + { + "name": "ack", + "type": "pull", + "method": "bind", + "address": "tcp://127.0.0.1:5990", + "rateLogging": "0" + }] + }, + + + { + "id": "flpSender1", + "channels": + [{ + "name": "data-in", + "type": "sub", + "method": "connect", + "address": "tcp://127.0.0.1:5550", + "rcvBufSize": "10" + }, + { + "name": "hb-in", + "type": "sub", + "method": "bind", + "address": "tcp://127.0.0.1:5581", + "rateLogging": "0" + }, + { + "name": "data-out", + "type": "push", + "method": "connect", + "sockets": + [ + { "address": "tcp://127.0.0.1:5561" }, + { "address": "tcp://127.0.0.1:5562" }, + { "address": "tcp://127.0.0.1:5563" } + ], + "sndBufSize": "10" + }] + }, + { + "id": "flpSender2", + "channels": + [{ + "name": "data-in", + "type": "sub", + "method": "connect", + "address": "tcp://127.0.0.1:5550", + "rcvBufSize": "10" + }, + { + "name": "hb-in", + "type": "sub", + "method": "bind", + "address": "tcp://127.0.0.1:5582", + "rateLogging": "0" + }, + { + "name": "data-out", + "type": "push", + "method": "connect", + "sockets": + [ + { "address": "tcp://127.0.0.1:5561" }, + { "address": "tcp://127.0.0.1:5562" }, + { "address": "tcp://127.0.0.1:5563" } + ], + "sndBufSize": "10" + }] + }, + { + "id": "flpSender3", + "channels": + [{ + "name": "data-in", + "type": "sub", + "method": "connect", + "address": "tcp://127.0.0.1:5550", + "rcvBufSize": "10" + }, + { + "name": "hb-in", + "type": "sub", + "method": "bind", + "address": "tcp://127.0.0.1:5583", + "rateLogging": "0" + }, + { + "name": "data-out", + "type": "push", + "method": "connect", + "sockets": + [ + { "address": "tcp://127.0.0.1:5561" }, + { "address": "tcp://127.0.0.1:5562" }, + { "address": "tcp://127.0.0.1:5563" } + ], + "sndBufSize": "10" + }] + }, + + + { + "id": "epnReceiver1", + "channels": + [{ + "name": "data-in", + "type": "pull", + "method": "connect", + "address": "tcp://127.0.0.1:5561", + "rcvBufSize": "10" + }, + { + "name": "data-out", + "type": "pub", + "method": "bind", + "address": "tcp://127.0.0.1:5591", + "sndBufSize": "10" + }, + { + "name": "hb-out", + "type": "pub", + "method": "connect", + "sockets": + [ + { "address": "tcp://127.0.0.1:5581" }, + { "address": "tcp://127.0.0.1:5582" }, + { "address": "tcp://127.0.0.1:5583" } + ], + "rateLogging": "0" + }, + { + "name": "ack", + "type": "push", + "method": "connect", + "address": "tcp://127.0.0.1:5990", + "rateLogging": "0" + }] + }, + { + "id": "epnReceiver2", + "channels": + [{ + "name": "data-in", + "type": "pull", + "method": "connect", + "address": "tcp://127.0.0.1:5562", + "rcvBufSize": "10" + }, + { + "name": "data-out", + "type": "pub", + "method": "bind", + "address": "tcp://127.0.0.1:5592", + "sndBufSize": "10" + }, + { + "name": "hb-out", + "type": "pub", + "method": "connect", + "sockets": + [ + { "address": "tcp://127.0.0.1:5581" }, + { "address": "tcp://127.0.0.1:5582" }, + { "address": "tcp://127.0.0.1:5583" } + ], + "rateLogging": "0" + }, + { + "name": "ack", + "type": "push", + "method": "connect", + "address": "tcp://127.0.0.1:5990", + "rateLogging": "0" + }] + }, + { + "id": "epnReceiver3", + "channels": + [{ + "name": "data-in", + "type": "pull", + "method": "connect", + "address": "tcp://127.0.0.1:5563", + "rcvBufSize": "10" + }, + { + "name": "data-out", + "type": "pub", + "method": "bind", + "address": "tcp://127.0.0.1:5593", + "sndBufSize": "10" + }, + { + "name": "hb-out", + "type": "pub", + "method": "connect", + "sockets": + [ + { "address": "tcp://127.0.0.1:5581" }, + { "address": "tcp://127.0.0.1:5582" }, + { "address": "tcp://127.0.0.1:5583" } + ], + "rateLogging": "0" + }, + { + "name": "ack", + "type": "push", + "method": "connect", + "address": "tcp://127.0.0.1:5990", + "rateLogging": "0" + }] + }] + } +} diff --git a/Examples/flp2epn-distributed/run/runEPNReceiver.cxx b/Examples/flp2epn-distributed/run/runEPNReceiver.cxx index 80e5eee52a0eb..5bf8d0f82c46f 100644 --- a/Examples/flp2epn-distributed/run/runEPNReceiver.cxx +++ b/Examples/flp2epn-distributed/run/runEPNReceiver.cxx @@ -10,7 +10,6 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" -#include "FairMQTransportFactoryZMQ.h" #include "EPNReceiver.h" @@ -33,6 +32,7 @@ struct DeviceOptions int numFLPs; int testMode; int interactive; + string transport; string dataInSocketType; int dataInBufSize; @@ -74,6 +74,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) ("num-flps", bpo::value()->required(), "Number of FLPs") ("test-mode", bpo::value()->default_value(0), "Run in test mode") ("interactive", bpo::value()->default_value(1), "Run in interactive mode (1/0)") + ("transport", bpo::value()->default_value("zeromq"), "Transport (zeromq/nanomsg)") ("data-in-socket-type", bpo::value()->default_value("pull"), "Data input socket type: sub/pull") ("data-in-buff-size", bpo::value()->default_value(10), "Data input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") @@ -88,13 +89,13 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) ("data-out-rate-logging", bpo::value()->default_value(1), "Log output rate on data socket, 1/0") ("hb-out-socket-type", bpo::value()->default_value("pub"), "Heartbeat output socket type: pub/push") - ("hb-out-buff-size", bpo::value()->default_value(100), "Heartbeat output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") + ("hb-out-buff-size", bpo::value()->default_value(10), "Heartbeat output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") ("hb-out-method", bpo::value()->default_value("connect"), "Heartbeat output method: bind/connect") ("hb-out-address", bpo::value>()->required(), "Heartbeat output address, e.g.: \"tcp://localhost:5555\"") ("hb-out-rate-logging", bpo::value()->default_value(0), "Log output rate on heartbeat socket, 1/0") ("ack-out-socket-type", bpo::value()->default_value("push"), "Acknowledgement output socket type: pub/push") - ("ack-out-buff-size", bpo::value()->default_value(100), "Acknowledgement output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") + ("ack-out-buff-size", bpo::value()->default_value(10), "Acknowledgement output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") ("ack-out-method", bpo::value()->default_value("connect"), "Acknowledgement output method: bind/connect") ("ack-out-address", bpo::value()->required(), "Acknowledgement output address, e.g.: \"tcp://localhost:5555\"") ("ack-out-rate-logging", bpo::value()->default_value(0), "Log output rate on acknowledgement socket, 1/0") @@ -118,6 +119,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) if (vm.count("num-flps")) { _options->numFLPs = vm["num-flps"].as(); } if (vm.count("test-mode")) { _options->testMode = vm["test-mode"].as(); } if (vm.count("interactive")) { _options->interactive = vm["interactive"].as(); } + if (vm.count("transport")) { _options->transport = vm["transport"].as(); } if (vm.count("data-in-socket-type")) { _options->dataInSocketType = vm["data-in-socket-type"].as(); } if (vm.count("data-in-buff-size")) { _options->dataInBufSize = vm["data-in-buff-size"].as(); } @@ -173,8 +175,7 @@ int main(int argc, char** argv) LOG(INFO) << "EPN Receiver, ID: " << options.id << " (PID: " << getpid() << ")"; // configure the transport interface - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); - epn.SetTransport(transportFactory); + epn.SetTransport(options.transport); // set device properties epn.SetProperty(EPNReceiver::Id, options.id); diff --git a/Examples/flp2epn-distributed/run/runFLPSender.cxx b/Examples/flp2epn-distributed/run/runFLPSender.cxx index 9926f5d5762d4..233ec0443e997 100644 --- a/Examples/flp2epn-distributed/run/runFLPSender.cxx +++ b/Examples/flp2epn-distributed/run/runFLPSender.cxx @@ -10,7 +10,6 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" -#include "FairMQTransportFactoryZMQ.h" #include "FLPSender.h" @@ -35,6 +34,7 @@ struct DeviceOptions int interactive; int sendOffset; int sendDelay; + string transport; string dataInSocketType; int dataInBufSize; @@ -72,6 +72,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) ("heartbeat-timeout", bpo::value()->default_value(20000), "Heartbeat timeout in milliseconds") ("test-mode", bpo::value()->default_value(0), "Run in test mode") ("interactive", bpo::value()->default_value(1), "Run in interactive mode (1/0)") + ("transport", bpo::value()->default_value("zeromq"), "Transport (zeromq/nanomsg)") ("send-offset", bpo::value()->default_value(0), "Offset for staggered sending") ("send-delay", bpo::value()->default_value(8), "Delay for staggered sending") @@ -88,7 +89,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) ("data-out-rate-logging", bpo::value()->default_value(1), "Log output rate on socket, 1/0") ("hb-in-socket-type", bpo::value()->default_value("sub"), "Heartbeat in socket type: sub/pull") - ("hb-in-buff-size", bpo::value()->default_value(100), "Heartbeat in buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") + ("hb-in-buff-size", bpo::value()->default_value(10), "Heartbeat in buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") ("hb-in-method", bpo::value()->default_value("bind"), "Heartbeat in method: bind/connect") ("hb-in-address", bpo::value()->required(), "Heartbeat in address, e.g.: \"tcp://localhost:5555\"") ("hb-in-rate-logging", bpo::value()->default_value(0), "Log heartbeat in rate on socket, 1/0") @@ -115,6 +116,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) if (vm.count("heartbeat-timeout")) { _options->heartbeatTimeoutInMs = vm["heartbeat-timeout"].as(); } if (vm.count("test-mode")) { _options->testMode = vm["test-mode"].as(); } if (vm.count("interactive")) { _options->interactive = vm["interactive"].as(); } + if (vm.count("transport")) { _options->transport = vm["transport"].as(); } if (vm.count("send-offset")) { _options->sendOffset = vm["send-offset"].as(); } if (vm.count("send-delay")) { _options->sendDelay = vm["send-delay"].as(); } @@ -167,8 +169,7 @@ int main(int argc, char** argv) LOG(INFO) << "FLP Sender, ID: " << options.id << " (PID: " << getpid() << ")"; // configure the transport interface - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); - flp.SetTransport(transportFactory); + flp.SetTransport(options.transport); // set device properties flp.SetProperty(FLPSender::Id, options.id); diff --git a/Examples/flp2epn-distributed/run/runFLPSyncSampler.cxx b/Examples/flp2epn-distributed/run/runFLPSyncSampler.cxx index 226e05d196681..c1d71e56680fc 100644 --- a/Examples/flp2epn-distributed/run/runFLPSyncSampler.cxx +++ b/Examples/flp2epn-distributed/run/runFLPSyncSampler.cxx @@ -10,7 +10,6 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" -#include "FairMQTransportFactoryZMQ.h" #include "FLPSyncSampler.h" @@ -30,6 +29,7 @@ struct DeviceOptions int ioThreads; int interactive; int storeRTTinFile; + string transport; string dataOutSocketType; int dataOutBufSize; @@ -57,16 +57,17 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) ("max-events", bpo::value()->default_value(0), "Maximum number of events to send (0 - unlimited)") ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") ("interactive", bpo::value()->default_value(1), "Run in interactive mode (1/0)") + ("transport", bpo::value()->default_value("zeromq"), "Transport (zeromq/nanomsg)") ("store-rtt-in-file", bpo::value()->default_value(0), "Store round trip time measurements in a file (1/0)") ("data-out-socket-type", bpo::value()->default_value("pub"), "Data output socket type: pub/push") - ("data-out-buff-size", bpo::value()->default_value(100), "Data output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") + ("data-out-buff-size", bpo::value()->default_value(10), "Data output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") ("data-out-method", bpo::value()->default_value("bind"), "Data output method: bind/connect") ("data-out-address", bpo::value()->required(), "Data output address, e.g.: \"tcp://localhost:5555\"") ("data-out-rate-logging", bpo::value()->default_value(0), "Log output rate on data socket, 1/0") ("ack-in-socket-type", bpo::value()->default_value("pull"), "Acknowledgement Input socket type: sub/pull") - ("ack-in-buff-size", bpo::value()->default_value(100), "Acknowledgement Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") + ("ack-in-buff-size", bpo::value()->default_value(10), "Acknowledgement Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") ("ack-in-method", bpo::value()->default_value("bind"), "Acknowledgement Input method: bind/connect") ("ack-in-address", bpo::value()->required(), "Acknowledgement Input address, e.g.: \"tcp://localhost:5555\"") ("ack-in-rate-logging", bpo::value()->default_value(0), "Log input rate on Acknowledgement socket, 1/0") @@ -88,6 +89,7 @@ inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) if (vm.count("max-events")) { _options->maxEvents = vm["max-events"].as(); } if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as(); } if (vm.count("interactive")) { _options->interactive = vm["interactive"].as(); } + if (vm.count("transport")) { _options->transport = vm["transport"].as(); } if (vm.count("store-rtt-in-file")) { _options->storeRTTinFile = vm["store-rtt-in-file"].as(); } if (vm.count("data-out-socket-type")) { _options->dataOutSocketType = vm["data-out-socket-type"].as(); } @@ -126,8 +128,7 @@ int main(int argc, char** argv) LOG(INFO) << "FLP Sync Sampler, ID: " << options.id << " (PID: " << getpid() << ")"; // configure the transport interface - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); - sampler.SetTransport(transportFactory); + sampler.SetTransport(options.transport); // set device properties sampler.SetProperty(FLPSyncSampler::Id, options.id); diff --git a/Examples/flp2epn-distributed/run/startFLP2EPN-distributed.sh.in b/Examples/flp2epn-distributed/run/startFLP2EPN-distributed.sh.in index 6e51cfd38e75d..c22fd750df368 100755 --- a/Examples/flp2epn-distributed/run/startFLP2EPN-distributed.sh.in +++ b/Examples/flp2epn-distributed/run/startFLP2EPN-distributed.sh.in @@ -1,18 +1,18 @@ #!/bin/bash -buffSize="2" # zeromq high-water mark is in messages -#buffSize="50000000" # nanomsg buffer size is in bytes -signalBuffSize="100" +cfg="@CMAKE_BINARY_DIR@/bin/config/flp2epn-prototype.json" SAMPLER="flpSyncSampler" -SAMPLER+=" --id 101" +SAMPLER+=" --id flpSyncSampler" +# SAMPLER+=" --mq-config $cfg" SAMPLER+=" --event-rate 100" SAMPLER+=" --ack-in-address tcp://*:5990" SAMPLER+=" --data-out-address tcp://*:5550" xterm -geometry 80x25+0+0 -hold -e @CMAKE_BINARY_DIR@/bin/$SAMPLER & FLP0="flpSender" -FLP0+=" --id FLP1" +FLP0+=" --id flpSender1" +# FLP0+=" --mq-config $cfg" FLP0+=" --flp-index 0" FLP0+=" --event-size 1000000" FLP0+=" --num-epns 3" @@ -27,7 +27,8 @@ FLP0+=" --data-out-address tcp://127.0.0.1:5562" xterm -geometry 80x25+500+0 -hold -e @CMAKE_BINARY_DIR@/bin/$FLP0 & FLP1="flpSender" -FLP1+=" --id FLP2" +FLP1+=" --id flpSender2" +# FLP1+=" --mq-config $cfg" FLP1+=" --flp-index 1" FLP1+=" --event-size 1000000" FLP1+=" --num-epns 3" @@ -42,7 +43,8 @@ FLP1+=" --data-out-address tcp://127.0.0.1:5562" xterm -geometry 80x25+500+350 -hold -e @CMAKE_BINARY_DIR@/bin/$FLP1 & FLP2="flpSender" -FLP2+=" --id FLP3" +FLP2+=" --id flpSender3" +# FLP2+=" --mq-config $cfg" FLP2+=" --flp-index 2" FLP2+=" --event-size 1000000" FLP2+=" --num-epns 3" @@ -57,7 +59,8 @@ FLP2+=" --data-out-address tcp://127.0.0.1:5562" xterm -geometry 80x25+500+700 -hold -e @CMAKE_BINARY_DIR@/bin/$FLP2 & EPN0="epnReceiver" -EPN0+=" --id EPN1" +EPN0+=" --id epnReceiver1" +# EPN0+=" --mq-config $cfg" EPN0+=" --heartbeat-interval 5000" EPN0+=" --num-flps 3" EPN0+=" --test-mode 1" @@ -70,7 +73,8 @@ EPN0+=" --ack-out-address tcp://127.0.0.1:5990" xterm -geometry 80x25+1000+0 -hold -e @CMAKE_BINARY_DIR@/bin/$EPN0 & EPN1="epnReceiver" -EPN1+=" --id EPN2" +EPN1+=" --id epnReceiver2" +# EPN1+=" --mq-config $cfg" EPN1+=" --heartbeat-interval 5000" EPN1+=" --num-flps 3" EPN1+=" --test-mode 1" @@ -83,7 +87,8 @@ EPN1+=" --ack-out-address tcp://127.0.0.1:5990" xterm -geometry 80x25+1000+350 -hold -e @CMAKE_BINARY_DIR@/bin/$EPN1 & EPN2="epnReceiver" -EPN2+=" --id EPN3" +EPN2+=" --id epnReceiver3" +# EPN2+=" --mq-config $cfg" EPN2+=" --heartbeat-interval 5000" EPN2+=" --num-flps 3" EPN2+=" --test-mode 1" diff --git a/Examples/flp2epn-distributed/runO2Prototype/runEPNReceiver.cxx b/Examples/flp2epn-distributed/runO2Prototype/runEPNReceiver.cxx index ee99173e4ff8a..e42e56322a014 100644 --- a/Examples/flp2epn-distributed/runO2Prototype/runEPNReceiver.cxx +++ b/Examples/flp2epn-distributed/runO2Prototype/runEPNReceiver.cxx @@ -20,7 +20,7 @@ #include "EPNReceiver.h" -#include "KeyValue.h" // DDS +#include "dds_intercom.h" // DDS using namespace std; using namespace AliceO2::Devices; @@ -244,7 +244,7 @@ int main(int argc, char** argv) epn.WaitForInitialValidation(); // create DDS key value store - dds::key_value::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue ddsKeyValue; // Advertise the bound data input address via DDS. ddsKeyValue.putValue("EPNReceiverInputAddress", epn.fChannels["data-in"].at(0).GetAddress()); @@ -254,7 +254,7 @@ int main(int argc, char** argv) } // Initialize DDS store to receive properties - dds::key_value::CKeyValue::valuesMap_t values; + dds::intercom_api::CKeyValue::valuesMap_t values; { mutex keyMutex; condition_variable keyCondition; diff --git a/Examples/flp2epn-distributed/runO2Prototype/runFLPSender.cxx b/Examples/flp2epn-distributed/runO2Prototype/runFLPSender.cxx index 4b1cdbe27823e..1292c22ae829b 100644 --- a/Examples/flp2epn-distributed/runO2Prototype/runFLPSender.cxx +++ b/Examples/flp2epn-distributed/runO2Prototype/runFLPSender.cxx @@ -20,7 +20,7 @@ #include "FLPSender.h" -#include "KeyValue.h" // DDS +#include "dds_intercom.h" // DDS using namespace std; using namespace AliceO2::Devices; @@ -195,7 +195,7 @@ int main(int argc, char** argv) flp.SetTransport(transportFactory); // initialize DDS key value store - dds::key_value::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue ddsKeyValue; // set device properties flp.SetProperty(FLPSender::Id, options.id); @@ -231,7 +231,7 @@ int main(int argc, char** argv) if (options.testMode == 1) { // in test mode, retreive the output address of FLPSyncSampler to connect to and assign it to device - dds::key_value::CKeyValue::valuesMap_t values; + dds::intercom_api::CKeyValue::valuesMap_t values; { mutex keyMutex; condition_variable keyCondition; @@ -262,7 +262,7 @@ int main(int argc, char** argv) // advertise the heartbeat input address via DDS ddsKeyValue.putValue("FLPSenderHeartbeatInputAddress", flp.fChannels["heartbeat-in"].at(0).GetAddress()); - dds::key_value::CKeyValue::valuesMap_t epn_addr_values; + dds::intercom_api::CKeyValue::valuesMap_t epn_addr_values; // receive the EPNReceiver input addresses from DDS. { diff --git a/Examples/flp2epn-distributed/runO2Prototype/runFLPSyncSampler.cxx b/Examples/flp2epn-distributed/runO2Prototype/runFLPSyncSampler.cxx index e610e81b6f48d..799b9da73a232 100644 --- a/Examples/flp2epn-distributed/runO2Prototype/runFLPSyncSampler.cxx +++ b/Examples/flp2epn-distributed/runO2Prototype/runFLPSyncSampler.cxx @@ -20,7 +20,7 @@ #include "FLPSyncSampler.h" -#include "KeyValue.h" // DDS +#include "dds_intercom.h" // DDS using namespace std; using namespace AliceO2::Devices; @@ -178,7 +178,7 @@ int main(int argc, char** argv) sampler.WaitForInitialValidation(); // Advertise the bound addresses via DDS properties - dds::key_value::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue ddsKeyValue; ddsKeyValue.putValue("FLPSyncSamplerOutputAddress", sampler.fChannels["data-out"].at(0).GetAddress()); ddsKeyValue.putValue("FLPSyncSamplerInputAddress", sampler.fChannels["ack-in"].at(0).GetAddress()); diff --git a/Examples/flp2epn/CMakeLists.txt b/Examples/flp2epn/CMakeLists.txt index cbafe788fdbc0..8ef05ae223279 100644 --- a/Examples/flp2epn/CMakeLists.txt +++ b/Examples/flp2epn/CMakeLists.txt @@ -13,8 +13,8 @@ set(SYSTEM_INCLUDE_DIRECTORIES include_directories(${INCLUDE_DIRECTORIES}) include_directories(SYSTEM ${SYSTEM_INCLUDE_DIRECTORIES}) -configure_file( ${CMAKE_SOURCE_DIR}/Examples/flp2epn/run/startFLP2EPN.sh.in ${CMAKE_BINARY_DIR}/bin/startFLP2EPN.sh ) -configure_file( ${CMAKE_SOURCE_DIR}/Examples/flp2epn/run/startMerger.sh.in ${CMAKE_BINARY_DIR}/bin/startMerger.sh ) +configure_file(${CMAKE_SOURCE_DIR}/Examples/flp2epn/run/startFLP2EPN.sh.in ${CMAKE_BINARY_DIR}/bin/startFLP2EPN.sh) +configure_file(${CMAKE_SOURCE_DIR}/Examples/flp2epn/run/flp2epn.json ${CMAKE_BINARY_DIR}/bin/config/flp2epn.json) set(LINK_DIRECTORIES ${Boost_LIBRARY_DIRS} @@ -27,9 +27,6 @@ link_directories(${LINK_DIRECTORIES}) set(SRCS O2FLPex.cxx O2EPNex.cxx - O2Proxy.cxx - O2Merger.cxx - O2EpnMerger.cxx ) if(FAIRMQ_DEPENDENCIES) @@ -38,12 +35,21 @@ if(FAIRMQ_DEPENDENCIES) ${CMAKE_THREAD_LIBS_INIT} ${FAIRMQ_DEPENDENCIES} FairMQ + ${Boost_RANDOM_LIBRARY} + ${Boost_CHRONO_LIBRARY} + ${Boost_REGEX_LIBRARY} ) 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} + FairMQ ) endif(FAIRMQ_DEPENDENCIES) @@ -55,17 +61,13 @@ Set(Exe_Names ${Exe_Names} testFLP testEPN - testEPN_M testProxy - testMerger ) set(Exe_Source run/runFLP.cxx run/runEPN.cxx - run/runEPN_M.cxx run/runProxy.cxx - run/runMerger.cxx ) list(LENGTH Exe_Names _length) diff --git a/Examples/flp2epn/O2EPNex.cxx b/Examples/flp2epn/O2EPNex.cxx index 5b2917454504d..54d7f6d7b74a5 100644 --- a/Examples/flp2epn/O2EPNex.cxx +++ b/Examples/flp2epn/O2EPNex.cxx @@ -10,14 +10,9 @@ #include "FairMQLogger.h" #include "O2EPNex.h" +#include "O2FLPExContent.h" -struct Content { - double a; - double b; - int x; - int y; - int z; -}; +using namespace std; O2EPNex::O2EPNex() { @@ -26,12 +21,12 @@ O2EPNex::O2EPNex() void O2EPNex::Run() { while (CheckCurrentState(RUNNING)) { - std::unique_ptr msg(fTransportFactory->CreateMessage()); + unique_ptr msg(NewMessage()); - fChannels.at("data-in").at(0).Receive(msg); + fChannels.at("data").at(0).Receive(msg); - // int numInput = msg->GetSize() / sizeof(Content); - // Content* input = static_cast(msg->GetData()); + // int numInput = msg->GetSize() / sizeof(O2FLPExContent); + // O2FLPExContent* input = static_cast(msg->GetData()); // for (int i = 0; i < numInput; ++i) { // LOG(INFO) << (&input[i])->x << " " << (&input[i])->y << " " << (&input[i])->z << " " << (&input[i])->a << " " << (&input[i])->b; diff --git a/Examples/flp2epn/O2EpnMerger.cxx b/Examples/flp2epn/O2EpnMerger.cxx deleted file mode 100644 index c29598dcf13e8..0000000000000 --- a/Examples/flp2epn/O2EpnMerger.cxx +++ /dev/null @@ -1,48 +0,0 @@ -/** - * O2EpnMerger.cxx - * - * @since 2013-01-09 - * @author D. Klein, A. Rybalchenko, M.Al-Turany - */ - -#include -#include - -#include "O2EpnMerger.h" -#include "FairMQLogger.h" - -O2EpnMerger::O2EpnMerger() -{ -} - -void O2EpnMerger::Run() -{ - std::unique_ptr poller(fTransportFactory->CreatePoller(fChannels.at("data-in"))); - - int numParts = fChannels.at("data-in").size() - 1; - - while (CheckCurrentState(RUNNING)) { - std::unique_ptr msg(fTransportFactory->CreateMessage()); - - poller->Poll(100); - - for (unsigned int i = 0; i < fChannels.at("data-in").size(); i++) { - if (poller->CheckInput(i)) { - if (fChannels.at("data-in").at(i).Receive(msg)) { - // LOG(INFO) << "------ recieve Msg from " << i ; - if (i < numParts) { - fChannels.at("data-out").at(0).SendPart(msg); - // LOG(INFO) << "------ Send Msg Part " << i ; - } else { - fChannels.at("data-out").at(0).Send(msg); - // LOG(INFO) << "------ Send last Msg Part " << i ; - } - } - } - } - } -} - -O2EpnMerger::~O2EpnMerger() -{ -} diff --git a/Examples/flp2epn/O2EpnMerger.h b/Examples/flp2epn/O2EpnMerger.h deleted file mode 100644 index 6cca142822b9d..0000000000000 --- a/Examples/flp2epn/O2EpnMerger.h +++ /dev/null @@ -1,23 +0,0 @@ -/** - * O2EpnMerger.h - * - * @since 2013-01-09 - * @author D. Klein, A. Rybalchenko, M.Al-Turany - */ - -#ifndef O2EpnMerger_H_ -#define O2EpnMerger_H_ - -#include "FairMQDevice.h" - -class O2EpnMerger: public FairMQDevice -{ - public: - O2EpnMerger(); - virtual ~O2EpnMerger(); - - protected: - virtual void Run(); -}; - -#endif /* O2EpnMerger_H_ */ diff --git a/Examples/flp2epn/O2FLPExContent.h b/Examples/flp2epn/O2FLPExContent.h new file mode 100644 index 0000000000000..0912f0d4e4993 --- /dev/null +++ b/Examples/flp2epn/O2FLPExContent.h @@ -0,0 +1,13 @@ +#ifndef O2FLPEXCONTENT_H_ +#define O2FLPEXCONTENT_H_ + +struct O2FLPExContent { + int id; + double a; + double b; + int x; + int y; + int z; +}; + +#endif /* O2FLPEXCONTENT_H_ */ diff --git a/Examples/flp2epn/O2FLPex.cxx b/Examples/flp2epn/O2FLPex.cxx index 6eed815b2daaa..d0bc3f5f1e003 100644 --- a/Examples/flp2epn/O2FLPex.cxx +++ b/Examples/flp2epn/O2FLPex.cxx @@ -14,20 +14,12 @@ #include "FairMQLogger.h" #include "O2FLPex.h" +#include "O2FLPExContent.h" using namespace std; -struct Content { - int id; - double a; - double b; - int x; - int y; - int z; -}; - O2FLPex::O2FLPex() : - fEventSize(10000) + fNumContent(10000) { } @@ -39,14 +31,14 @@ void O2FLPex::Run() { srand(time(NULL)); - FairMQChannel& outChannel = fChannels.at("data-out").at(0); + FairMQChannel& outChannel = fChannels.at("data").at(0); - LOG(DEBUG) << "Message size: " << fEventSize * sizeof(Content) << " bytes."; + LOG(DEBUG) << "Message size: " << fNumContent * sizeof(O2FLPExContent) << " bytes."; while (CheckCurrentState(RUNNING)) { - vector payload(fEventSize); + vector payload(fNumContent); - for (int i = 0; i < fEventSize; ++i) { + for (int i = 0; i < fNumContent; ++i) { payload.at(i).x = rand() % 100 + 1; payload.at(i).y = rand() % 100 + 1; payload.at(i).z = rand() % 100 + 1; @@ -55,8 +47,8 @@ void O2FLPex::Run() // LOG(INFO) << (&payload[i])->x << " " << (&payload[i])->y << " " << (&payload[i])->z << " " << (&payload[i])->a << " " << (&payload[i])->b; } - unique_ptr msg(fTransportFactory->CreateMessage(fEventSize * sizeof(Content))); - memcpy(msg->GetData(), payload.data(), fEventSize * sizeof(Content)); + unique_ptr msg(NewMessage(fNumContent * sizeof(O2FLPExContent))); + memcpy(msg->GetData(), payload.data(), fNumContent * sizeof(O2FLPExContent)); outChannel.Send(msg); } @@ -82,8 +74,8 @@ string O2FLPex::GetProperty(const int key, const string& default_/*= ""*/) void O2FLPex::SetProperty(const int key, const int value) { switch (key) { - case EventSize: - fEventSize = value; + case NumContent: + fNumContent = value; break; default: FairMQDevice::SetProperty(key, value); @@ -94,8 +86,8 @@ void O2FLPex::SetProperty(const int key, const int value) int O2FLPex::GetProperty(const int key, const int default_/*= 0*/) { switch (key) { - case EventSize: - return fEventSize; + case NumContent: + return fNumContent; default: return FairMQDevice::GetProperty(key, default_); } diff --git a/Examples/flp2epn/O2FLPex.h b/Examples/flp2epn/O2FLPex.h index bef7be48f2a90..04bae34e67b8a 100644 --- a/Examples/flp2epn/O2FLPex.h +++ b/Examples/flp2epn/O2FLPex.h @@ -16,13 +16,11 @@ class O2FLPex : public FairMQDevice { public: enum { - InputFile = FairMQDevice::Last, - EventSize, + NumContent = FairMQDevice::Last, Last }; O2FLPex(); virtual ~O2FLPex(); - void Log(int intervalInMs); virtual void SetProperty(const int key, const std::string& value); virtual std::string GetProperty(const int key, const std::string& default_ = ""); @@ -30,7 +28,7 @@ class O2FLPex : public FairMQDevice virtual int GetProperty(const int key, const int default_ = 0); protected: - int fEventSize; + int fNumContent; virtual void Run(); }; diff --git a/Examples/flp2epn/O2Merger.cxx b/Examples/flp2epn/O2Merger.cxx deleted file mode 100644 index 590ba7da151b3..0000000000000 --- a/Examples/flp2epn/O2Merger.cxx +++ /dev/null @@ -1,50 +0,0 @@ -/** - * O2Merger.cxx - * - * @since 2012-12-06 - * @author D. Klein, A. Rybalchenko, M. Al-Turany - */ - -#include -#include - -#include "FairMQLogger.h" -#include "FairMQPoller.h" -#include "O2Merger.h" - -O2Merger::O2Merger() -{ -} - -O2Merger::~O2Merger() -{ -} - -void O2Merger::Run() -{ - std::unique_ptr poller(fTransportFactory->CreatePoller(fChannels.at("data-in"))); - - int numParts = fChannels.at("data-in").size() - 1; - - while (CheckCurrentState(RUNNING)) { - std::unique_ptr msg(fTransportFactory->CreateMessage()); - - poller->Poll(100); - - for (unsigned int i = 0; i < fChannels.at("data-in").size(); i++) { - if (poller->CheckInput(i)) { - if (fChannels.at("data-in").at(i).Receive(msg) >= 0) { - // LOG(INFO) << "------ recieve Msg from " << i ; - if (i < numParts) { - fChannels.at("data-out").at(0).SendPart(msg); - // LOG(INFO) << "------ Send Msg Part " << i ; - } else { - fChannels.at("data-out").at(0).Send(msg); - // LOG(INFO) << "------ Send last Msg Part " << i ; - } - } - } - } - } -} - diff --git a/Examples/flp2epn/O2Merger.h b/Examples/flp2epn/O2Merger.h deleted file mode 100644 index 02d9a0e9a834d..0000000000000 --- a/Examples/flp2epn/O2Merger.h +++ /dev/null @@ -1,24 +0,0 @@ -/** - * O2Merger.h - * - * @since 2012-12-06 - * @author D. Klein, A. Rybalchenko, M. Al-Turany - */ - -#ifndef O2Merger_H_ -#define O2Merger_H_ - -#include "FairMQDevice.h" - - -class O2Merger: public FairMQDevice -{ - public: - O2Merger(); - virtual ~O2Merger(); - - protected: - virtual void Run(); -}; - -#endif /* O2Merger_H_ */ diff --git a/Examples/flp2epn/O2Proxy.cxx b/Examples/flp2epn/O2Proxy.cxx deleted file mode 100644 index cb75d8f267ecd..0000000000000 --- a/Examples/flp2epn/O2Proxy.cxx +++ /dev/null @@ -1,47 +0,0 @@ -/** - * O2Proxy.cxx - * - * @since 2013-10-02 - * @author A. Rybalchenko, M.Al-Turany - */ - -#include -#include - -#include "FairMQLogger.h" -#include "O2Proxy.h" - -O2Proxy::O2Proxy() -{ -} - -O2Proxy::~O2Proxy() -{ -} - -void O2Proxy::Run() -{ - FairMQChannel& inChannel = fChannels.at("data-in").at(0); - FairMQChannel& outChannel = fChannels.at("data-out").at(0); - - while (CheckCurrentState(RUNNING)) { - // int i = 0; - bool more = false; - - do { - /* Create an empty message to hold the message part */ - std::unique_ptr part(fTransportFactory->CreateMessage()); - /* Block until a message is available to be received from socket */ - inChannel.Receive(part); - /* Determine if more message parts are to follow */ - more = inChannel.ExpectsAnotherPart(); - // LOG(INFO) << "------ Get Msg Part "<< " more = " << more << " counter " << i++ ; - if (more) { - outChannel.SendPart(part); - } else { - outChannel.Send(part); - } - } while (more); - // i = 0; - } -} diff --git a/Examples/flp2epn/O2Proxy.h b/Examples/flp2epn/O2Proxy.h deleted file mode 100644 index c3b7faef70d61..0000000000000 --- a/Examples/flp2epn/O2Proxy.h +++ /dev/null @@ -1,24 +0,0 @@ -/** - * O2Proxy.h - * - * @since 2013-10-02 - * @author A. Rybalchenko, , M.Al-Turany - */ - -#ifndef O2Proxy_H_ -#define O2Proxy_H_ - -#include "FairMQDevice.h" - - -class O2Proxy: public FairMQDevice -{ - public: - O2Proxy(); - virtual ~O2Proxy(); - - protected: - virtual void Run(); -}; - -#endif /* O2Proxy_H_ */ diff --git a/Examples/flp2epn/README.md b/Examples/flp2epn/README.md new file mode 100644 index 0000000000000..2aa8bf63cd390 --- /dev/null +++ b/Examples/flp2epn/README.md @@ -0,0 +1,15 @@ +#### Example devices - testFLP and testEPN +-------------------------------------------------------------- + +#### General + +Two simple devices - testFLP and testEPN, showing basic FairMQ usage. +For more complex scenario, refer to the prototype flp2epn topology in `devices/flp2epn-prototype`. + +For more basic FairMQ examples, take a look at the MQ examples in the [FairRoot repository](https://github.com/FairRootGroup/FairRoot). + +#### Device configuration + +The devices are started by the `startFLP2EPN.sh` script, which configures the devices via command line options and their communication channels via a JSON configuration file. + +To list *all* available device command line options, run the executable with `--help`. diff --git a/Examples/flp2epn/run/flp2epn.json b/Examples/flp2epn/run/flp2epn.json new file mode 100644 index 0000000000000..7ab9f4eb6fd02 --- /dev/null +++ b/Examples/flp2epn/run/flp2epn.json @@ -0,0 +1,65 @@ +{ + "fairMQOptions": + { + "devices": + [{ + "id": "flpEx", + "channel": + { + "name": "data", + "socket": + { + "type": "push", + "method": "connect", + "address": "tcp://127.0.0.1:5565", + "sndBufSize": "1000", + "rcvBufSize": "1000", + "rateLogging": "1" + } + } + }, + { + "id": "proxy", + "channels": + [{ + "name": "data-in", + "socket": + { + "type": "pull", + "method": "bind", + "address": "tcp://*:5565", + "sndBufSize": "1000", + "rcvBufSize": "1000", + "rateLogging": "1" + } + },{ + "name": "data-out", + "socket": + { + "type": "push", + "method": "bind", + "address": "tcp://*:5566", + "sndBufSize": "1000", + "rcvBufSize": "1000", + "rateLogging": "1" + } + }] + }, + { + "id": "epnEx", + "channel": + { + "name": "data", + "socket": + { + "type": "pull", + "method": "connect", + "address": "tcp://127.0.0.1:5566", + "sndBufSize": "1000", + "rcvBufSize": "1000", + "rateLogging": "1" + } + } + }] + } +} diff --git a/Examples/flp2epn/run/runEPN.cxx b/Examples/flp2epn/run/runEPN.cxx index e543cef2ce3d4..aef5e466da176 100644 --- a/Examples/flp2epn/run/runEPN.cxx +++ b/Examples/flp2epn/run/runEPN.cxx @@ -7,105 +7,23 @@ #include -#include "boost/program_options.hpp" - #include "FairMQLogger.h" +#include "FairMQParser.h" +#include "FairMQProgOptions.h" #include "O2EPNex.h" -#ifdef NANOMSG - #include "FairMQTransportFactoryNN.h" -#else - #include "FairMQTransportFactoryZMQ.h" -#endif - using namespace std; -struct DeviceOptions -{ - DeviceOptions() : - id(), ioThreads(1), - inputSocketType(), inputBufSize(1000), inputMethod(), inputAddress() {} - - string id; - int ioThreads; - string inputSocketType; - int inputBufSize; - string inputMethod; - string inputAddress; -}; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw std::runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("input-socket-type", bpo::value()->required(), "Input socket type: sub/pull") - ("input-buff-size", bpo::value()->required(), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("input-method", bpo::value()->required(), "Input method: bind/connect") - ("input-address", bpo::value()->required(), "Input address, e.g.: \"tcp://localhost:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if (vm.count("help")) - { - LOG(INFO) << "EPN" << endl << desc; - return false; - } - - bpo::notify(vm); - - if (vm.count("id")) { _options->id = vm["id"].as(); } - if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as(); } - if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as(); } - if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as(); } - if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as(); } - if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as(); } - - return true; -} - int main(int argc, char** argv) { - O2EPNex epn; - epn.CatchSignals(); - - DeviceOptions options; - try - { - if (!parse_cmd_line(argc, argv, &options)) - return 0; - } - catch (exception& e) - { - LOG(ERROR) << e.what(); - return 1; - } + O2EPNex epn; + epn.CatchSignals(); - LOG(INFO) << "PID: " << getpid(); + FairMQProgOptions config; -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif - - epn.SetTransport(transportFactory); - - epn.SetProperty(O2EPNex::Id, options.id); - epn.SetProperty(O2EPNex::NumIoThreads, options.ioThreads); - - FairMQChannel inputChannel(options.inputSocketType, options.inputMethod, options.inputAddress); - inputChannel.UpdateSndBufSize(options.inputBufSize); - inputChannel.UpdateRcvBufSize(options.inputBufSize); - - epn.fChannels["data-in"].push_back(inputChannel); + try { + config.ParseAll(argc, argv); + epn.SetConfig(config); epn.ChangeState("INIT_DEVICE"); epn.WaitForEndOfState("INIT_DEVICE"); @@ -115,6 +33,12 @@ int main(int argc, char** argv) epn.ChangeState("RUN"); epn.InteractiveStateLoop(); - - return 0; + } catch (std::exception& e) { + LOG(ERROR) << e.what(); + LOG(INFO) << "Command line options are the following: "; + config.PrintHelp(); + return 1; + } + + return 0; } diff --git a/Examples/flp2epn/run/runEPN_M.cxx b/Examples/flp2epn/run/runEPN_M.cxx deleted file mode 100644 index 5850941dfe3bc..0000000000000 --- a/Examples/flp2epn/run/runEPN_M.cxx +++ /dev/null @@ -1,121 +0,0 @@ -/** - * runEPN.cxx - * - * @since 2013-01-21 - * @author D. Klein, A. Rybalchenko, M.Al-Turany - */ - -#include - -#include "boost/program_options.hpp" - -#include "FairMQLogger.h" -#include "O2EpnMerger.h" - -#ifdef NANOMSG - #include "FairMQTransportFactoryNN.h" -#else - #include "FairMQTransportFactoryZMQ.h" -#endif - -using namespace std; - -struct DeviceOptions -{ - DeviceOptions() : - id(), ioThreads(1), - inputSocketType(), inputBufSize(1000), inputMethod(), inputAddress() {} - - string id; - int ioThreads; - string inputSocketType; - int inputBufSize; - string inputMethod; - string inputAddress; -}; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw std::runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("input-socket-type", bpo::value()->required(), "Input socket type: sub/pull") - ("input-buff-size", bpo::value()->required(), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("input-method", bpo::value()->required(), "Input method: bind/connect") - ("input-address", bpo::value()->required(), "Input address, e.g.: \"tcp://localhost:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if (vm.count("help")) - { - LOG(INFO) << "EPN Merger" << endl << desc; - return false; - } - - bpo::notify(vm); - - if (vm.count("id")) { _options->id = vm["id"].as(); } - if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as(); } - if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as(); } - if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as(); } - if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as(); } - if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as(); } - - return true; -} - -int main(int argc, char** argv) -{ - O2EpnMerger epn; - epn.CatchSignals(); - - DeviceOptions options; - try - { - if (!parse_cmd_line(argc, argv, &options)) - return 0; - } - catch (exception& e) - { - LOG(ERROR) << e.what(); - return 1; - } - - LOG(INFO) << "PID: " << getpid(); - -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif - - epn.SetTransport(transportFactory); - - epn.SetProperty(O2EpnMerger::Id, options.id); - epn.SetProperty(O2EpnMerger::NumIoThreads, options.ioThreads); - - FairMQChannel inputChannel(options.inputSocketType, options.inputMethod, options.inputAddress); - inputChannel.UpdateSndBufSize(options.inputBufSize); - inputChannel.UpdateRcvBufSize(options.inputBufSize); - - epn.fChannels["data-in"].push_back(inputChannel); - - epn.ChangeState("INIT_DEVICE"); - epn.WaitForEndOfState("INIT_DEVICE"); - - epn.ChangeState("INIT_TASK"); - epn.WaitForEndOfState("INIT_TASK"); - - epn.ChangeState("RUN"); - epn.InteractiveStateLoop(); - - return 0; -} - diff --git a/Examples/flp2epn/run/runFLP.cxx b/Examples/flp2epn/run/runFLP.cxx index a3830e3dbc6b3..14d76756d1149 100644 --- a/Examples/flp2epn/run/runFLP.cxx +++ b/Examples/flp2epn/run/runFLP.cxx @@ -10,106 +10,31 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" +#include "FairMQParser.h" +#include "FairMQProgOptions.h" #include "O2FLPex.h" -#ifdef NANOMSG - #include "FairMQTransportFactoryNN.h" -#else - #include "FairMQTransportFactoryZMQ.h" -#endif - -using namespace std; - -struct DeviceOptions -{ - DeviceOptions() : - id(), ioThreads(1), - outputSocketType(), outputBufSize(1000), outputMethod(), outputAddress() {} - - string id; - int eventSize; - int ioThreads; - string outputSocketType; - int outputBufSize; - string outputMethod; - string outputAddress; -}; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw std::runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("event-size", bpo::value()->default_value(1000), "Event size in bytes") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("output-socket-type", bpo::value()->required(), "Output socket type: pub/push") - ("output-buff-size", bpo::value()->required(), "Output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("output-method", bpo::value()->required(), "Output method: bind/connect") - ("output-address", bpo::value()->required(), "Output address, e.g.: \"tcp://localhost:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if (vm.count("help")) - { - LOG(INFO) << "FLP" << endl << desc; - return false; - } - - bpo::notify(vm); - - if (vm.count("id")) { _options->id = vm["id"].as(); } - if (vm.count("event-size")) { _options->eventSize = vm["event-size"].as(); } - if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as(); } - if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as(); } - if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as(); } - if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as(); } - if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as(); } - - return true; -} +using namespace boost::program_options; int main(int argc, char** argv) { - O2FLPex flp; - flp.CatchSignals(); + O2FLPex flp; + flp.CatchSignals(); - DeviceOptions options; - try - { - if (!parse_cmd_line(argc, argv, &options)) - return 0; - } - catch (exception& e) - { - LOG(ERROR) << e.what(); - return 1; - } + FairMQProgOptions config; - LOG(INFO) << "PID: " << getpid(); + try { + int numContent; -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif + options_description flpOptions("FLPex options"); + flpOptions.add_options() + ("num-content", value(&numContent)->default_value(1000), "Event size in bytes"); - flp.SetTransport(transportFactory); + config.AddToCmdLineOptions(flpOptions); - flp.SetProperty(O2FLPex::Id, options.id); - flp.SetProperty(O2FLPex::NumIoThreads, options.ioThreads); - flp.SetProperty(O2FLPex::EventSize, options.eventSize); - - FairMQChannel outputChannel(options.outputSocketType, options.outputMethod, options.outputAddress); - outputChannel.UpdateSndBufSize(options.outputBufSize); - outputChannel.UpdateRcvBufSize(options.outputBufSize); - - flp.fChannels["data-out"].push_back(outputChannel); + config.ParseAll(argc, argv); + flp.SetConfig(config); + flp.SetProperty(O2FLPex::NumContent, numContent); flp.ChangeState("INIT_DEVICE"); flp.WaitForEndOfState("INIT_DEVICE"); @@ -119,6 +44,12 @@ int main(int argc, char** argv) flp.ChangeState("RUN"); flp.InteractiveStateLoop(); - - return 0; -} + } catch (std::exception& e) { + LOG(ERROR) << e.what(); + LOG(INFO) << "Command line options are the following: "; + config.PrintHelp(); + return 1; + } + + return 0; +} \ No newline at end of file diff --git a/Examples/flp2epn/run/runMerger.cxx b/Examples/flp2epn/run/runMerger.cxx deleted file mode 100644 index a6a74db1587f4..0000000000000 --- a/Examples/flp2epn/run/runMerger.cxx +++ /dev/null @@ -1,148 +0,0 @@ -/** - * runMerger.cxx - * - * @since 2012-12-06 - * @author D. Klein, A. Rybalchenko, M.Al-Turany - */ - -#include - -#include "boost/program_options.hpp" - -#include "FairMQLogger.h" -#include "O2Merger.h" - -#ifdef NANOMSG - #include "FairMQTransportFactoryNN.h" -#else - #include "FairMQTransportFactoryZMQ.h" -#endif - -using namespace std; - -struct DeviceOptions -{ - DeviceOptions() : - id(), ioThreads(1), numInputs(0), - inputSocketType(), inputBufSize(), inputMethod(), inputAddress(), - outputSocketType(), outputBufSize(1000), outputMethod(), outputAddress() {} - - string id; - int ioThreads; - int numInputs; - vector inputSocketType; - vector inputBufSize; - vector inputMethod; - vector inputAddress; - string outputSocketType; - int outputBufSize; - string outputMethod; - string outputAddress; -}; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw std::runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("num-inputs", bpo::value()->required(), "Number of input sockets") - ("input-socket-type", bpo::value< vector >()->required(), "Input socket type: sub/pull") - ("input-buff-size", bpo::value< vector >()->required(), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("input-method", bpo::value< vector >()->required(), "Input method: bind/connect") - ("input-address", bpo::value< vector >()->required(), "Input address, e.g.: \"tcp://localhost:5555\"") - ("output-socket-type", bpo::value()->required(), "Output socket type: pub/push") - ("output-buff-size", bpo::value()->required(), "Output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("output-method", bpo::value()->required(), "Output method: bind/connect") - ("output-address", bpo::value()->required(), "Output address, e.g.: \"tcp://localhost:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if (vm.count("help")) - { - LOG(INFO) << "MERGER" << endl << desc; - return false; - } - - bpo::notify(vm); - - if (vm.count("id")) { _options->id = vm["id"].as(); } - if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as(); } - if (vm.count("num-inputs")) { _options->numInputs = vm["num-inputs"].as(); } - if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as< vector >(); } - if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as< vector >(); } - if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as< vector >(); } - if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as< vector >(); } - if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as(); } - if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as(); } - if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as(); } - if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as(); } - - return true; -} - -int main(int argc, char** argv) -{ - O2Merger merger; - merger.CatchSignals(); - - DeviceOptions options; - try - { - if (!parse_cmd_line(argc, argv, &options)) - return 0; - } - catch (exception& e) - { - LOG(ERROR) << e.what(); - return 1; - } - - LOG(INFO) << "PID: " << getpid(); - -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif - - merger.SetTransport(transportFactory); - - merger.SetProperty(O2Merger::Id, options.id); - merger.SetProperty(O2Merger::NumIoThreads, options.ioThreads); - - for (int i = 0; i < options.inputAddress.size(); ++i) - { - FairMQChannel inputChannel(options.inputSocketType.at(i), options.inputMethod.at(i), options.inputAddress.at(i)); - inputChannel.UpdateSndBufSize(options.inputBufSize.at(i)); - inputChannel.UpdateRcvBufSize(options.inputBufSize.at(i)); - inputChannel.UpdateRateLogging(1); - - merger.fChannels["data-in"].push_back(inputChannel); - } - - FairMQChannel outputChannel(options.outputSocketType, options.outputMethod, options.outputAddress); - outputChannel.UpdateSndBufSize(options.outputBufSize); - outputChannel.UpdateRcvBufSize(options.outputBufSize); - outputChannel.UpdateRateLogging(1); - - merger.fChannels["data-out"].push_back(outputChannel); - - merger.ChangeState("INIT_DEVICE"); - merger.WaitForEndOfState("INIT_DEVICE"); - - merger.ChangeState("INIT_TASK"); - merger.WaitForEndOfState("INIT_TASK"); - - merger.ChangeState("RUN"); - merger.InteractiveStateLoop(); - - return 0; -} - diff --git a/Examples/flp2epn/run/runProxy.cxx b/Examples/flp2epn/run/runProxy.cxx index f105f0735b6a4..277308e08e1bb 100644 --- a/Examples/flp2epn/run/runProxy.cxx +++ b/Examples/flp2epn/run/runProxy.cxx @@ -10,133 +10,32 @@ #include "boost/program_options.hpp" #include "FairMQLogger.h" -#include "O2Proxy.h" +#include "FairMQProgOptions.h" +#include "FairMQProxy.h" +#include "runSimpleMQStateMachine.h" -#ifdef NANOMSG - #include "FairMQTransportFactoryNN.h" -#else - #include "FairMQTransportFactoryZMQ.h" -#endif - -using namespace std; - -struct DeviceOptions -{ - DeviceOptions() : - id(), ioThreads(1), - inputSocketType(), inputBufSize(1000), inputMethod(), inputAddress(), - outputSocketType(), outputBufSize(1000), outputMethod(), outputAddress() {} - - string id; - int ioThreads; - string inputSocketType; - int inputBufSize; - string inputMethod; - string inputAddress; - string outputSocketType; - int outputBufSize; - string outputMethod; - string outputAddress; -}; - -inline bool parse_cmd_line(int _argc, char* _argv[], DeviceOptions* _options) -{ - if (_options == NULL) - throw std::runtime_error("Internal error: options' container is empty."); - - namespace bpo = boost::program_options; - bpo::options_description desc("Options"); - desc.add_options() - ("id", bpo::value()->required(), "Device ID") - ("io-threads", bpo::value()->default_value(1), "Number of I/O threads") - ("input-socket-type", bpo::value()->required(), "Input socket type: sub/pull") - ("input-buff-size", bpo::value()->required(), "Input buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("input-method", bpo::value()->required(), "Input method: bind/connect") - ("input-address", bpo::value()->required(), "Input address, e.g.: \"tcp://localhost:5555\"") - ("output-socket-type", bpo::value()->required(), "Output socket type: pub/push") - ("output-buff-size", bpo::value()->required(), "Output buffer size in number of messages (ZeroMQ)/bytes(nanomsg)") - ("output-method", bpo::value()->required(), "Output method: bind/connect") - ("output-address", bpo::value()->required(), "Output address, e.g.: \"tcp://localhost:5555\"") - ("help", "Print help messages"); - - bpo::variables_map vm; - bpo::store(bpo::parse_command_line(_argc, _argv, desc), vm); - - if (vm.count("help")) - { - LOG(INFO) << "Proxy" << endl << desc; - return false; - } - - bpo::notify(vm); - - if (vm.count("id")) { _options->id = vm["id"].as(); } - if (vm.count("io-threads")) { _options->ioThreads = vm["io-threads"].as(); } - if (vm.count("input-socket-type")) { _options->inputSocketType = vm["input-socket-type"].as(); } - if (vm.count("input-buff-size")) { _options->inputBufSize = vm["input-buff-size"].as(); } - if (vm.count("input-method")) { _options->inputMethod = vm["input-method"].as(); } - if (vm.count("input-address")) { _options->inputAddress = vm["input-address"].as(); } - if (vm.count("output-socket-type")) { _options->outputSocketType = vm["output-socket-type"].as(); } - if (vm.count("output-buff-size")) { _options->outputBufSize = vm["output-buff-size"].as(); } - if (vm.count("output-method")) { _options->outputMethod = vm["output-method"].as(); } - if (vm.count("output-address")) { _options->outputAddress = vm["output-address"].as(); } - - return true; -} +using namespace boost::program_options; int main(int argc, char** argv) { - O2Proxy proxy; - proxy.CatchSignals(); - - DeviceOptions options; - try - { - if (!parse_cmd_line(argc, argv, &options)) - return 0; - } - catch (exception& e) - { - LOG(ERROR) << e.what(); - return 1; - } - - LOG(INFO) << "PID: " << getpid(); - -#ifdef NANOMSG - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryNN(); -#else - FairMQTransportFactory* transportFactory = new FairMQTransportFactoryZMQ(); -#endif - - proxy.SetTransport(transportFactory); - - proxy.SetProperty(O2Proxy::Id, options.id); - proxy.SetProperty(O2Proxy::NumIoThreads, options.ioThreads); - - FairMQChannel inputChannel(options.inputSocketType, options.inputMethod, options.inputAddress); - inputChannel.UpdateSndBufSize(options.inputBufSize); - inputChannel.UpdateRcvBufSize(options.inputBufSize); - inputChannel.UpdateRateLogging(1); - - proxy.fChannels["data-in"].push_back(inputChannel); - - FairMQChannel outputChannel(options.outputSocketType, options.outputMethod, options.outputAddress); - outputChannel.UpdateSndBufSize(options.outputBufSize); - outputChannel.UpdateRcvBufSize(options.outputBufSize); - outputChannel.UpdateRateLogging(1); - - proxy.fChannels["data-out"].push_back(outputChannel); - - proxy.ChangeState("INIT_DEVICE"); - proxy.WaitForEndOfState("INIT_DEVICE"); - - proxy.ChangeState("INIT_TASK"); - proxy.WaitForEndOfState("INIT_TASK"); - - proxy.ChangeState("RUN"); - proxy.InteractiveStateLoop(); - - return 0; + try { + int multipart; + + options_description proxyOptions("Proxy options"); + proxyOptions.add_options() + ("multipart", value(&multipart)->default_value(1), "Handle multipart payloads"); + + FairMQProgOptions config; + config.AddToCmdLineOptions(proxyOptions); + config.ParseAll(argc, argv); + + FairMQProxy proxy; + proxy.SetProperty(FairMQProxy::Multipart, multipart); + runStateMachine(proxy, config); + } catch (std::exception& e) { + LOG(ERROR) << "Unhandled Exception reached the top of main: " << e.what() << ", application will now exit"; + return 1; + } + + return 0; } - diff --git a/Examples/flp2epn/run/startFLP2EPN.sh.in b/Examples/flp2epn/run/startFLP2EPN.sh.in index f36ada84d6516..0abd7db5f39d7 100755 --- a/Examples/flp2epn/run/startFLP2EPN.sh.in +++ b/Examples/flp2epn/run/startFLP2EPN.sh.in @@ -1,21 +1,23 @@ #!/bin/bash -buffSize="1000" # zeromq high-water mark is in messages -#buffSize="50000000" # nanomsg buffer size is in bytes +flp2epnEx1config="@CMAKE_BINARY_DIR@/bin/config/flp2epn.json" FLP0="testFLP" -FLP0+=" --id 101" -FLP0+=" --event-size 10000" -FLP0+=" --output-socket-type push --output-buff-size $buffSize --output-method connect --output-address tcp://127.0.0.1:5565" -xterm -e @CMAKE_BINARY_DIR@/bin/$FLP0 & +FLP0+=" --id flpEx" +# FLP0+=" --transport nanomsg" +FLP0+=" --num-content 10000" +FLP0+=" --mq-config $flp2epnEx1config" +xterm -geometry 80x23+0+0 -hold -e @CMAKE_BINARY_DIR@/bin/$FLP0 & PROXY="testProxy" -PROXY+=" --id 201" -PROXY+=" --input-socket-type pull --input-buff-size $buffSize --input-method bind --input-address tcp://*:5565" -PROXY+=" --output-socket-type push --output-buff-size $buffSize --output-method bind --output-address tcp://*:5566" -xterm -e @CMAKE_BINARY_DIR@/bin/$PROXY & +PROXY+=" --id proxy" +# PROXY+=" --transport nanomsg" +# PROXY+=" --multipart 0" +PROXY+=" --mq-config $flp2epnEx1config" +xterm -geometry 80x23+500+0 -hold -e @CMAKE_BINARY_DIR@/bin/$PROXY & EPN0="testEPN" -EPN0+=" --id 301" -EPN0+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://127.0.0.1:5566" -xterm -e @CMAKE_BINARY_DIR@/bin/$EPN0 & +EPN0+=" --id epnEx" +# EPN0+=" --transport nanomsg" +EPN0+=" --mq-config $flp2epnEx1config" +xterm -geometry 80x23+1000+0 -hold -e @CMAKE_BINARY_DIR@/bin/$EPN0 & diff --git a/Examples/flp2epn/run/startMerger.sh.in b/Examples/flp2epn/run/startMerger.sh.in deleted file mode 100755 index 57456400f4d60..0000000000000 --- a/Examples/flp2epn/run/startMerger.sh.in +++ /dev/null @@ -1,57 +0,0 @@ -#!/bin/bash - -buffSize="1000" # zeromq high-water mark is in messages -#buffSize="50000000" # nanomsg buffer size is in bytes - -FLP0="testFLP" -FLP0+=" --id 101" -FLP0+=" --event-size 10000" -FLP0+=" --output-socket-type push --output-buff-size $buffSize --output-method bind --output-address tcp://*:5565" -xterm -e @CMAKE_BINARY_DIR@/bin/$FLP0 & - -FLP1="testFLP" -FLP1+=" --id 102" -FLP1+=" --event-size 10000" -FLP1+=" --output-socket-type push --output-buff-size $buffSize --output-method bind --output-address tcp://*:5566" -xterm -e @CMAKE_BINARY_DIR@/bin/$FLP1 & - -FLP2="testFLP" -FLP2+=" --id 103" -FLP2+=" --event-size 10000" -FLP2+=" --output-socket-type push --output-buff-size $buffSize --output-method bind --output-address tcp://*:5567" -xterm -e @CMAKE_BINARY_DIR@/bin/$FLP2 & - -MERGER="testMerger" -MERGER+=" --id 501" -MERGER+=" --num-inputs 3" -MERGER+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5565" -MERGER+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5566" -MERGER+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5567" -MERGER+=" --output-socket-type push --output-buff-size $buffSize --output-method bind --output-address tcp://*:5581" -xterm -e @CMAKE_BINARY_DIR@/bin/$MERGER & - -PROXY="testProxy" -PROXY+=" --id 201" -PROXY+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5581" -PROXY+=" --output-socket-type push --output-buff-size $buffSize --output-method bind --output-address tcp://*:5582" -xterm -e @CMAKE_BINARY_DIR@/bin/$PROXY & - -EPN0="testEPN" -EPN0+=" --id 301" -EPN0+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5582" -xterm -e @CMAKE_BINARY_DIR@/bin/$EPN0 & - -EPN1="testEPN" -EPN1+=" --id 302" -EPN1+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5582" -xterm -e @CMAKE_BINARY_DIR@/bin/$EPN1 & - -EPN2="testEPN" -EPN2+=" --id 303" -EPN2+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5582" -xterm -e @CMAKE_BINARY_DIR@/bin/$EPN2 & - -EPN3="testEPN" -EPN3+=" --id 304" -EPN3+=" --input-socket-type pull --input-buff-size $buffSize --input-method connect --input-address tcp://localhost:5582" -xterm -e @CMAKE_BINARY_DIR@/bin/$EPN3 & diff --git a/Utilities/aliceHLTwrapper/CMakeLists.txt b/Utilities/aliceHLTwrapper/CMakeLists.txt index bd69a78d711e1..b1f2bfd58c6e3 100644 --- a/Utilities/aliceHLTwrapper/CMakeLists.txt +++ b/Utilities/aliceHLTwrapper/CMakeLists.txt @@ -50,7 +50,7 @@ set(SRCS if(DDS_FOUND) set(DEPENDENCIES ${DEPENDENCIES} - dds-key-value-lib + dds_intercom_lib ) endif() @@ -64,13 +64,20 @@ if(FAIRMQ_DEPENDENCIES) ${CMAKE_THREAD_LIBS_INIT} ${FAIRMQ_DEPENDENCIES} ${Boost_CHRONO_LIBRARY} + ${Boost_RANDOM_LIBRARY} + ${Boost_REGEX_LIBRARY} FairMQ ) else(FAIRMQ_DEPENDENCIES) set(DEPENDENCIES ${DEPENDENCIES} ${CMAKE_THREAD_LIBS_INIT} - ${Boost_CHRONO_LIBRARY} ${Boost_DATE_TIME_LIBRARY} ${Boost_THREAD_LIBRARY} ${Boost_THREAD_LIBRARY} ${Boost_SYSTEM_LIBRARY} ${Boost_PROGRAM_OPTIONS_LIBRARY} FairMQ + ${Boost_CHRONO_LIBRARY} + ${Boost_DATE_TIME_LIBRARY} + ${Boost_THREAD_LIBRARY} + ${Boost_SYSTEM_LIBRARY} + ${Boost_PROGRAM_OPTIONS_LIBRARY} + FairMQ ) endif(FAIRMQ_DEPENDENCIES) diff --git a/Utilities/aliceHLTwrapper/EventSampler.cxx b/Utilities/aliceHLTwrapper/EventSampler.cxx index 81833833dd28f..958821587b0d2 100644 --- a/Utilities/aliceHLTwrapper/EventSampler.cxx +++ b/Utilities/aliceHLTwrapper/EventSampler.cxx @@ -68,7 +68,7 @@ void EventSampler::Run() boost::thread samplerThread(boost::bind(&EventSampler::samplerLoop, this)); - FairMQPoller* poller = fTransportFactory->CreatePoller(fChannels["data-in"]); + unique_ptr poller(fTransportFactory->CreatePoller(fChannels["data-in"])); bool received = false; @@ -80,7 +80,7 @@ void EventSampler::Run() int errorCount=0; const int maxError=10; - vector inputMessages; + vector> inputMessages; vector inputMessageCntPerSocket(numInputs, 0); int nReadCycles=0; @@ -92,17 +92,12 @@ void EventSampler::Run() poller->Poll(mPollingTimeout); for(int i = 0; i < numInputs; i++) { if (poller->CheckInput(i)) { - do { - unique_ptr msg(fTransportFactory->CreateMessage()); - received = fChannels.at("data-in").at(i).Receive(msg.get()); - if (received) { - inputMessages.push_back(msg.release()); - inputMessageCntPerSocket[i]++; - if (mVerbosity > 3) { - LOG(INFO) << " |---- receive Msg from socket " << i; - } + if (fChannels.at("data-in").at(i).Receive(inputMessages)) { + inputMessageCntPerSocket[i] = inputMessages.size(); + if (mVerbosity > 3) { + LOG(INFO) << " |---- receive Msgs from socket " << i; } - } while (fChannels.at("data-in").at(i).ExpectsAnotherPart()); + } if (mVerbosity > 2) { LOG(INFO) << "------ received " << inputMessageCntPerSocket[i] << " message(s) from socket " << i; } @@ -113,7 +108,7 @@ void EventSampler::Run() auto seconds = std::chrono::duration_cast(timestamp - dayref); auto useconds = std::chrono::duration_cast(timestamp - dayref - seconds); - for (vector::iterator mit=inputMessages.begin(); + for (vector>::iterator mit=inputMessages.begin(); mit!=inputMessages.end(); mit++) { AliHLTComponentEventData* evtData=reinterpret_cast((*mit)->GetData()); if ((*mit)->GetSize() >= sizeof(AliHLTComponentEventData) && @@ -147,8 +142,6 @@ void EventSampler::Run() latencyLog << evtData->fEventID << " " << latencyUSeconds << endl; } } - - delete *mit; } inputMessages.clear(); for (vector::iterator mcit=inputMessageCntPerSocket.begin(); @@ -161,8 +154,6 @@ void EventSampler::Run() latencyLog.close(); } - delete poller; - samplerThread.interrupt(); samplerThread.join(); } diff --git a/Utilities/aliceHLTwrapper/WrapperDevice.cxx b/Utilities/aliceHLTwrapper/WrapperDevice.cxx index b6711f16c187e..995ca890cfa71 100644 --- a/Utilities/aliceHLTwrapper/WrapperDevice.cxx +++ b/Utilities/aliceHLTwrapper/WrapperDevice.cxx @@ -112,7 +112,7 @@ void WrapperDevice::Run() static system_clock::time_point refTime = system_clock::now(); #endif // USE_CHRONO - FairMQPoller* poller = fTransportFactory->CreatePoller(fChannels["data-in"]); + unique_ptr poller(fTransportFactory->CreatePoller(fChannels["data-in"])); // inherited variables of FairMQDevice: // fChannels @@ -122,7 +122,7 @@ void WrapperDevice::Run() int errorCount=0; const int maxError=10; - vector inputMessages; + vector> inputMessages; vector inputMessageCntPerSocket(numInputs, 0); int nReadCycles=0; while (CheckCurrentState(RUNNING)) { @@ -137,19 +137,14 @@ void WrapperDevice::Run() continue; } if (poller->CheckInput(i)) { - do { - unique_ptr msg(fTransportFactory->CreateMessage()); - if (fChannels.at("data-in").at(i).Receive(msg.get())) { - receivedAtLeastOneMessage = true; - inputMessages.push_back(msg.release()); - if (inputMessageCntPerSocket[i] == 0) - inputsReceived++; // count only the first message on that socket - inputMessageCntPerSocket[i]++; - if (mVerbosity > 3) { - LOG(INFO) << " |---- receive Msg from socket " << i; - } + if (fChannels.at("data-in").at(i).Receive(inputMessages)) { + receivedAtLeastOneMessage = true; + inputsReceived++; // count only the first message on that socket + inputMessageCntPerSocket[i] = inputMessages.size(); + if (mVerbosity > 3) { + LOG(INFO) << " |---- receive Msgs from socket " << i; } - } while (fChannels.at("data-in").at(i).ExpectsAnotherPart()); + } if (mVerbosity > 2) { LOG(INFO) << "------ received " << inputMessageCntPerSocket[i] << " message(s) from socket " << i; } @@ -199,7 +194,7 @@ void WrapperDevice::Run() if (!mSkipProcessing) { // prepare input from messages vector dataArray; - for (vector::iterator msg=inputMessages.begin(); + for (vector>::iterator msg=inputMessages.begin(); msg!=inputMessages.end(); msg++) { void* buffer=(*msg)->GetData(); dataArray.push_back(AliceO2::AliceHLT::MessageFormat::BufferDesc_t(reinterpret_cast(buffer), (*msg)->GetSize())); @@ -227,7 +222,7 @@ void WrapperDevice::Run() for (auto premsg = begin(mMessages); premsg != end(mMessages); premsg++) { if ((*premsg)->GetData() == opayload.mP && (*premsg)->GetSize() == opayload.mSize) { - omsg=*premsg; + omsg=(*premsg).get(); if (mVerbosity > 2) { LOG(DEBUG) << "using pre-allocated message of size " << opayload.mSize; } @@ -247,7 +242,7 @@ void WrapperDevice::Run() } AliHLTUInt8_t* pTarget = reinterpret_cast(msg->GetData()); memcpy(pTarget, opayload.mP, opayload.mSize); - mMessages.push_back(msg.release()); + mMessages.push_back(move(msg)); } else { if (errorCount == maxError && errorCount++ > 0) LOG(ERROR) << "persistent error, suppressing further output"; @@ -261,19 +256,9 @@ void WrapperDevice::Run() if (mMessages.size()>0) { if (fChannels.find("data-out") != fChannels.end() && fChannels["data-out"].size() > 0) { - for (auto sendmsg = begin(mMessages); sendmsg != end(mMessages); sendmsg++) { - if (sendmsg + 1 == end(mMessages)) { - // this is the last data block - if (mVerbosity > 2) { - LOG(DEBUG) << "sending last message, size " << (*sendmsg)->GetSize(); - } - fChannels["data-out"].at(0).Send(*sendmsg); - } else { - if (mVerbosity > 2) { - LOG(DEBUG) << "sending multipart message, size " << (*sendmsg)->GetSize(); - } - fChannels["data-out"].at(0).Send(*sendmsg, "snd-more"); - } + fChannels["data-out"].at(0).Send(mMessages); + if (mVerbosity > 2) { + LOG(DEBUG) << "sending multipart message with " << mMessages.size() << " parts"; } } else { if (errorCount == maxError && errorCount++ > 0) @@ -282,24 +267,17 @@ void WrapperDevice::Run() LOG(ERROR) << "no output slot available (" << (fChannels.find("data-out") == fChannels.end() ? "uninitialized" : "0 slots") << ")"; } - for (auto sendmsg : mMessages) delete sendmsg; mMessages.clear(); } } // cleanup - for (vector::iterator mit=inputMessages.begin(); - mit!=inputMessages.end(); mit++) { - delete *mit; - } inputMessages.clear(); for (vector::iterator mcit=inputMessageCntPerSocket.begin(); mcit!=inputMessageCntPerSocket.end(); mcit++) { *mcit=0; } } - - delete poller; } void WrapperDevice::Pause() @@ -365,6 +343,6 @@ unsigned char* WrapperDevice::createMessageBuffer(unsigned size) if (mVerbosity > 2) { LOG(DEBUG) << "allocating message of size " << size; } - mMessages.push_back(msg.release()); + mMessages.push_back(move(msg)); return reinterpret_cast(mMessages.back()->GetData()); } diff --git a/Utilities/aliceHLTwrapper/WrapperDevice.h b/Utilities/aliceHLTwrapper/WrapperDevice.h index 49dbad53ed777..0dcdc9eb4ee94 100644 --- a/Utilities/aliceHLTwrapper/WrapperDevice.h +++ b/Utilities/aliceHLTwrapper/WrapperDevice.h @@ -83,7 +83,7 @@ class WrapperDevice : public FairMQDevice { Component* mComponent; // component instance std::vector mArgv; // array of arguments for the component - std::vector mMessages; // array of output messages + std::vector> mMessages; // array of output messages int mPollingPeriod; // period of polling on input sockets in ms int mSkipProcessing; // skip component processing diff --git a/Utilities/aliceHLTwrapper/aliceHLTEventSampler.cxx b/Utilities/aliceHLTwrapper/aliceHLTEventSampler.cxx index 7e6fb65665b3e..bbf400187a6d8 100644 --- a/Utilities/aliceHLTwrapper/aliceHLTEventSampler.cxx +++ b/Utilities/aliceHLTwrapper/aliceHLTEventSampler.cxx @@ -38,7 +38,7 @@ #ifdef ENABLE_DDS #include #include -#include "KeyValue.h" // DDS +#include "dds_intercom.h" // DDS #include // boost::lock #endif @@ -522,7 +522,7 @@ int sendSocketPropertiesDDS(vector& sockets) ddsmsg << sit->address; #ifdef ENABLE_DDS - dds::key_value::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue ddsKeyValue; ddsKeyValue.putValue(sit->ddsprop, ddsmsg.str()); #endif @@ -540,8 +540,8 @@ int readSocketPropertiesDDS(vector& sockets) if (sit->ddscount==0) continue; // the previously inserted duplicates #ifdef ENABLE_DDS - dds::key_value::CKeyValue ddsKeyValue; - dds::key_value::CKeyValue::valuesMap_t values; + dds::intercom_api::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue::valuesMap_t values; std::string hostaddress=sit->address; vector::iterator workit=sit; @@ -561,7 +561,7 @@ int readSocketPropertiesDDS(vector& sockets) ddsKeyValue.getValues(sit->ddsprop.c_str(), &values); cout << "Info: DDS getValues received " << values.size() << " value(s) of property " << sit->ddsprop << " " << sit->ddscount-socketPropertiesToRead << " of " << sit->ddscount << " sockets processed" << endl; - for (dds::key_value::CKeyValue::valuesMap_t::const_iterator vit = values.begin(); + for (dds::intercom_api::CKeyValue::valuesMap_t::const_iterator vit = values.begin(); vit!=values.end(); vit++) { if (usedProperties.find(vit->first)!=usedProperties.end()) continue; // already processed cout << "Info: processing property " << vit->first << ", value " << vit->second << " on host " << hostaddress << endl; diff --git a/Utilities/aliceHLTwrapper/aliceHLTWrapper.cxx b/Utilities/aliceHLTwrapper/aliceHLTWrapper.cxx index 6caf171db7937..425c2eb7345fe 100644 --- a/Utilities/aliceHLTwrapper/aliceHLTWrapper.cxx +++ b/Utilities/aliceHLTwrapper/aliceHLTWrapper.cxx @@ -38,7 +38,7 @@ #ifdef ENABLE_DDS #include #include -#include "KeyValue.h" // DDS +#include "dds_intercom.h" // DDS #include // boost::lock #endif @@ -589,7 +589,7 @@ int sendSocketPropertiesDDS(vector& sockets) ddsmsg << sit->address; #ifdef ENABLE_DDS - dds::key_value::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue ddsKeyValue; ddsKeyValue.putValue(sit->ddsprop, ddsmsg.str()); #endif @@ -607,8 +607,8 @@ int readSocketPropertiesDDS(vector& sockets) if (sit->ddscount==0) continue; // the previously inserted duplicates #ifdef ENABLE_DDS - dds::key_value::CKeyValue ddsKeyValue; - dds::key_value::CKeyValue::valuesMap_t values; + dds::intercom_api::CKeyValue ddsKeyValue; + dds::intercom_api::CKeyValue::valuesMap_t values; std::string hostaddress=sit->address; vector::iterator workit=sit; @@ -628,7 +628,7 @@ int readSocketPropertiesDDS(vector& sockets) ddsKeyValue.getValues(sit->ddsprop.c_str(), &values); cout << "Info: DDS getValues received " << values.size() << " value(s) of property " << sit->ddsprop << " " << sit->ddscount-socketPropertiesToRead << " of " << sit->ddscount << " sockets processed" << endl; - for (dds::key_value::CKeyValue::valuesMap_t::const_iterator vit = values.begin(); + for (dds::intercom_api::CKeyValue::valuesMap_t::const_iterator vit = values.begin(); vit!=values.end(); vit++) { if (usedProperties.find(vit->first)!=usedProperties.end()) continue; // already processed cout << "Info: processing property " << vit->first << ", value " << vit->second << " on host " << hostaddress << endl;