Merge pull request #10 from sy-c/master

updating latest features

Author: sy-c

CMakeLists.txt

@@ -68,7 +68,7 @@ include_directories(
         ${Monitoring_INCLUDE_DIRS}
         ${FAIRROOT_INCLUDE_DIR}
         ${FAIRROOT_INCLUDE_DIR}/fairmq
-#        ${ZeroMQ_INCLUDE_DIR}
+        ${ZeroMQ_INCLUDE_DIR}
         ${Common_INCLUDE_DIRS}
         ${Configuration_INCLUDE_DIRS}
         ${InfoLogger_INCLUDE_DIRS}
@@ -142,6 +142,7 @@ O2_GENERATE_EXECUTABLE(
         BUCKET_NAME ${BUCKET_NAME}
 )
 
+if (FAIRROOT_FOUND)
 O2_GENERATE_EXECUTABLE(
         EXE_NAME testTxFMQ.exe
         SOURCES src/testTxFMQ.cxx $<TARGET_OBJECTS:objReadoutUtils>
@@ -153,6 +154,7 @@ O2_GENERATE_EXECUTABLE(
         SOURCES src/testRxFMQ.cxx $<TARGET_OBJECTS:objReadoutUtils>
         BUCKET_NAME ${BUCKET_NAME}
 )
+endif ()
 
 O2_GENERATE_EXECUTABLE(
         EXE_NAME readRaw.exe
@@ -166,6 +168,12 @@ O2_GENERATE_EXECUTABLE(
         BUCKET_NAME ${BUCKET_NAME}
 )
 
+O2_GENERATE_EXECUTABLE(
+        EXE_NAME testMonitor.exe
+        SOURCES src/testMonitor.cxx
+        BUCKET_NAME ${BUCKET_NAME}
+)
+
 # Install some extra files
 install(
         FILES readout.cfg

cmake/ReadoutDependencies.cmake

@@ -7,7 +7,7 @@ find_package(Common REQUIRED)
 find_package(InfoLogger REQUIRED)
 find_package(ReadoutCard REQUIRED)
 find_package(DataSampling REQUIRED)
-#find_package(ZeroMQ REQUIRED)
+find_package(ZeroMQ REQUIRED)
 find_package(Numa)
 
 if (FAIRROOT_FOUND)

src/ConsumerStats.cxx

@@ -89,6 +89,7 @@ class ConsumerStats: public Consumer {
       // https://alice.its.cern.ch/jira/browse/FLPPROT-69
 
       monitoringCollector->send({counterBlocks, "readout.Blocks"});
+      monitoringCollector->send({counterBytesTotal, "readout.BytesTotal"});
       monitoringCollector->send({counterBytesTotal, "readout.BytesTotal"}, DerivedMetricMode::RATE);
       monitoringCollector->send({counterBytesDiff, "readout.BytesInterval"});
 //      monitoringCollector->send({(counterBytesTotal/(1024*1024)), "readout.MegaBytesTotal"});

src/mainReadout.cxx

@@ -339,10 +339,11 @@ int main(int argc, char* argv[])
   theLog.log("Creating aggregator");
   AliceO2::Common::Fifo<DataSetReference> agg_output(1000);
   int nEquipmentsAggregated=0;
-  DataBlockAggregator agg(&agg_output,"Aggregator");
+  auto agg=std::make_unique<DataBlockAggregator>(&agg_output,"Aggregator");
+  
   for (auto && readoutDevice : readoutDevices) {
       //theLog.log("Adding equipment: %s",readoutDevice->getName().c_str());
-      agg.addInput(readoutDevice->dataOut);
+      agg->addInput(readoutDevice->dataOut);
       nEquipmentsAggregated++;
   }
   theLog.log("Aggregator: %d equipments", nEquipmentsAggregated);
@@ -368,7 +369,7 @@ int main(int argc, char* argv[])
 
 
   theLog.log("Starting aggregator");
-  agg.start();
+  agg->start();
 
   // notify consumers of imminent data flow start
   for (auto& c : dataConsumers) {
@@ -450,14 +451,15 @@ int main(int argc, char* argv[])
 
 
   theLog.log("Stopping aggregator");
-  agg.stop();
+  agg->stop();
 
   theLog.log("Stopping consumers");
   // close consumers before closing readout equipments (owner of data blocks)
   dataConsumers.clear();
 
   agg_output.clear();
-
+  agg=nullptr;  // destroy aggregator, and release blocks it may still own.
+   
   // todo: check nothing in the input pipeline
   // flush & stop equipments
   for (auto && readoutDevice : readoutDevices) {
@@ -472,7 +474,7 @@ int main(int argc, char* argv[])
   for (size_t i = 0, size = readoutDevices.size(); i != size; ++i) {
     readoutDevices[i]=nullptr;  // effectively deletes the device
   }
-  readoutDevices.clear(); // to do it all in one go
+  //readoutDevices.clear(); // to do it all in one go
 
   if (latencyFd>=0) {
     close(latencyFd);

src/testMonitor.cxx

@@ -0,0 +1,27 @@
+// simple test program used to evaluate overhead of monitoring lib
+
+#include <stdint.h>
+#include <stdio.h>
+
+#include <Monitoring/MonitoringFactory.h>
+using namespace o2::monitoring;
+
+int main() {
+
+  std::unique_ptr<Monitoring> monitoringCollector;  
+  monitoringCollector=MonitoringFactory::Get("influxdb-udp://aido2mon-gpn.cern.ch:8088");  
+  monitoringCollector->enableProcessMonitoring(1);
+  
+  uint64_t bytesTotal=0;
+
+  for(;;){
+    bytesTotal+=1000000000;
+    monitoringCollector->send({bytesTotal, "readout.BytesTotal"});
+    monitoringCollector->send({bytesTotal, "readout.BytesTotal"}, DerivedMetricMode::RATE);
+    printf(".");
+    fflush(stdout);
+    sleep(1);
+  }
+
+  return 0;
+}