Merge pull request #108 from DUNE-DAQ/ron/bad_omen

add sleep and send timeout checking in order to potentially log "bad …

Author: eflumerf

test/apps/zmq_recv.cpp

@@ -9,17 +9,17 @@ using namespace dunedaq::ipm;
 
 int main(int argc, char* argv[]){
   std::string conString="tcp://127.0.0.1:12345";
-  int npackets=1;
+  size_t npackets=1;
   int nthreads=1;
   int timeout = 10;
 
   namespace po = boost::program_options;
   po::options_description desc("Simple test program for ZmqReceiver");
   desc.add_options()(
-    "connection,c", po::value<std::string>(&conString), "Connection to listen on")(
-    "threads,t", po::value<int>(&nthreads), "Number of ZMQ threads")(
-    "packets,p", po::value<int>(&npackets), "Number of packets per group for reporting")(
-    "timeout,o", po::value<int>(&timeout), "Timeout, in seconds");
+    "connection,c", po::value<std::string>(&conString)->default_value(conString), "Connection to listen on")(
+    "threads,t", po::value<int>(&nthreads)->default_value(nthreads), "Number of ZMQ threads")(
+    "packets,p", po::value<size_t>(&npackets)->default_value(npackets), "Number of packets per group for reporting")(
+    "timeout,o", po::value<int>(&timeout)->default_value(timeout), "Timeout, in seconds");
   try {
     po::variables_map vm;
     po::store(po::parse_command_line(argc, argv, desc), vm);
@@ -30,6 +30,7 @@ int main(int argc, char* argv[]){
     return 0;
   }
 
+  dunedaq::logging::Logging::setup("ZMQ Test", "zmq_recv");
   if (nthreads > 1) {
     dunedaq::ipm::ZmqContext::instance().set_context_threads(nthreads);
   }
@@ -38,20 +39,36 @@ int main(int argc, char* argv[]){
   std::shared_ptr<Receiver> receiver=make_ipm_receiver("ZmqReceiver");
   receiver->connect_for_receives({ {"connection_string", conString} });
 
+  std::map<uint32_t, uint32_t> last_received_sequence;
+  int64_t first_latency = 0;
   try {
     while (true) {
       // Last arg is receive timeout
       auto start=std::chrono::steady_clock::now();
       float bytesReceived=0;
-      for (int p=0;p<npackets;p++) {
+      for (size_t p=0;p<npackets;p++) {
         Receiver::Response resp=receiver->receive(std::chrono::seconds(timeout));
+        int64_t recvd_ts = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
         bytesReceived+=resp.data.size();
+        auto this_id = *(reinterpret_cast<uint32_t*>(resp.data.data()));
+        auto this_sequence = *(reinterpret_cast<uint32_t*>(resp.data.data()) + 1);
+        auto this_ts = *(reinterpret_cast<uint64_t*>(resp.data.data()) + 1);
+
+        if(this_sequence < last_received_sequence[this_id] + 1) {
+          TLOG() << "Received sequence ID " << this_sequence << " < expected sequence " << (last_received_sequence[this_id] + 1) << " from sender " << this_id;
+        }
+        int64_t this_latency = recvd_ts - this_ts;
+        if(first_latency == 0) {
+          first_latency = this_latency;
+        }
+        TLOG_DEBUG(6) << "Received message " << this_sequence << " from sender " << this_id << ", latency= " << this_latency << " us (diff= " << (this_latency - first_latency) << " us)";
+        last_received_sequence[this_id] = this_sequence;
       }
       auto elapsed=std::chrono::steady_clock::now()-start;
       auto nano=std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count();
       float bw=bytesReceived/nano;
-      std::cout << "Received " << bytesReceived << " bytes in "
-                << nano << " ns " << bw << " GB/s" << std::endl;
+      TLOG() << "Received " << bytesReceived << " bytes in "
+                << nano << " ns " << bw << " GB/s";
       // std::cout << "resp.data=";
       // for (auto d: resp.data) {
       //   std::cout << d << std::endl;

test/apps/zmq_send.cpp

@@ -1,26 +1,31 @@
 #include "ipm/Sender.hpp"
 #include "ipm/ZmqContext.hpp"
-
+#include "logging/Logging.hpp"
 #include "boost/program_options.hpp"
 
 #include <memory>
 #include <chrono>
 #include <cstdlib>
 #include <cerrno>
+#include <unistd.h>				// sleep
 
 int main(int argc, char* argv[]){
-  int npackets=1;
-  int packetSize=100;
+  uint32_t npackets=1;
+  size_t packetSize=100;
+  size_t interval = 0;
   std::string conString="tcp://127.0.0.1:12345";
   int nthreads=1;
+  uint32_t id = 0;
 
   namespace po = boost::program_options;
   po::options_description desc("Simple test program for ZmqSender");
   desc.add_options()(
-    "connection,c", po::value<std::string>(&conString), "Connection to listen on")(
-    "threads,t", po::value<int>(&nthreads), "Number of ZMQ threads")(
-    "packets,p", po::value<int>(&npackets), "Number of packets to send")(
-    "packetSize,s", po::value<int>(&packetSize), "Number of bytes per packet");
+    "connection,c", po::value<std::string>(&conString)->default_value(conString), "Connection to listen on")(
+    "threads,t", po::value<int>(&nthreads)->default_value(nthreads), "Number of ZMQ threads")(
+    "packets,p", po::value<uint32_t>(&npackets)->default_value(npackets), "Number of packets to send")(
+    "packetSize,s", po::value<size_t>(&packetSize)->default_value(packetSize), "Number of bytes per packet")(
+    "interval,i", po::value<size_t>(&interval)->default_value(interval), "Microseconds to sleep between messages")(
+    "id", po::value<uint32_t>(&id)->default_value(id), "Identifier for this Sender");
   try {
     po::variables_map vm;
     po::store(po::parse_command_line(argc, argv, desc), vm);
@@ -31,25 +36,50 @@ int main(int argc, char* argv[]){
     return 0;
   }
 
+  dunedaq::logging::Logging::setup("ZMQ Test", "zmq_send");
   if (nthreads > 1) {
     dunedaq::ipm::ZmqContext::instance().set_context_threads(nthreads);
   }
+
+  // Set the minimum packet size to 16 bytes, 8 bytes for sequence number and 8 bytes for current time
+  if(packetSize < 16) { 
+    packetSize = 16;
+  }
 
   std::shared_ptr<dunedaq::ipm::Sender> sender=dunedaq::ipm::make_ipm_sender("ZmqSender");
   sender->connect_for_sends({ {"connection_string", conString} });
 
-  std::vector message(packetSize,0);
+  std::vector<char> message(packetSize,0);
+  *reinterpret_cast<uint32_t*>(message.data()) = id;
 
   auto start=std::chrono::steady_clock::now();
-  for (int p=0; p<npackets;p++) {
-    // Last arg is send timeout
-    sender->send((void*)message.data(), packetSize, std::chrono::milliseconds(100));
+  for (uint64_t p=0; p<npackets;p++) {
+    TLOG_DEBUG(3) << "Preparing Message " << p;
+	  // Last arg is send timeout
+	  int attempt=0;
+	  bool success;
+
+    *(reinterpret_cast<uint32_t*>(message.data()) + 1)= p;
+    *(reinterpret_cast<uint64_t*>(message.data()) + 1) = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
+
+    TLOG_DEBUG(4) << "Sending Message " << p;
+	  do {
+		  success = sender->send((void*)message.data(), packetSize, std::chrono::milliseconds(2),"",true);
+		  if (success == false && ++attempt == 1)
+			  TLOG() << "bad omen";
+	  } while (success == false);
+
+    TLOG_DEBUG(5) << "Message sent " << p;
+    if(interval > 0) {
+      usleep(interval);
+    }
   }
 
   auto elapsed=std::chrono::steady_clock::now()-start;
   auto nano=std::chrono::duration_cast<std::chrono::nanoseconds>(elapsed).count();
   float bw=((float)packetSize*npackets)/nano;
-  std::cout << "Sent " << packetSize*npackets << " bytes in "
-                << nano << " ns " << bw << " GB/s" << std::endl;
-  
+  TLOG() << "Sent " << packetSize*npackets << " bytes in "
+                << nano << " ns " << bw << " GB/s";
+  TLOG() << "Sleep 2 to allow finish???";
+  sleep(2); // Sleep for 2
 }