Aggregator

changes to aggregator

Fix logic in case of multiple slots

Fix in OutputSpec

Tool to run EPN + aggregator simulation

Remove aggregator since raw-proxy can be used

forgotten from the CMakeLists

clang-format

more info

Author: chiarazampolli

Detectors/Calibration/CMakeLists.txt

@@ -41,6 +41,5 @@ o2_add_executable(ccdb-populator-workflow
 	                                O2::DetectorsCalibration
 					O2::DataFormatsTOF
 					O2::CCDB)
-		
 add_subdirectory(workflow)
 add_subdirectory(testMacros)

Detectors/Calibration/testMacros/CMakeLists.txt

@@ -15,6 +15,9 @@ install(FILES populateCCDB.C
 install(FILES cdbSizeV0.txt
         DESTINATION share/Detectors/Calibration/data)	    
 
+install(FILES runEPNsimulation.sh
+        DESTINATION share/Detectors/Calibration/scripts)	    
+
 o2_add_test_root_macro(populateCCDB.C
                        PUBLIC_LINK_LIBRARIES O2::CCDB
                                              O2::Framework)

Detectors/Calibration/testMacros/README.md

@@ -0,0 +1,34 @@
+<!-- doxy
+\page refDetectorsCalibration/testMacros Module 'Detectors/Calibration/testMacros'
+/doxy -->
+
+# Simulation of sending of calibration data from EPNs to an aggregator
+
+To be used when calibrations produced by several EPNs have to be sent to a single node, the aggregator.
+On the aggregator, the devices producing calibration will run and send the output to the CCDB.
+
+* In order to populate the CCDB, the CCDB local server should be started, if the exercise is not meant to
+upload the official or test CCDB. In a terminal, run:
+```cpp
+java -jar local.jar
+```
+which will start a CCDB server on port 8080. See [instructions](https://github.com/AliceO2Group/AliceO2/tree/dev/CCDB#central-and-local-instances-of-the-ccdb).
+
+* To run the calibration and aggregator, open a terminal and start the aggregator, which is just the `o2-dpl-raw-proxy`:
+```cpp
+o2-dpl-raw-proxy --dataspec A:TOF/CALIBDATA/0 --channel-config "name=readout-proxy,type=pull,method=bind,address=tcp://localhost:30453,rateLogging=1,transport=zeromq"
+```
+where, as you can see, you can overwrite the configurations of the `readout-proxy` channel, which is the channel used by the aggregator.
+The aggregator should be started **before the other devices**, since it will listen for data (it is the channel that is *binding*).
+
+* The following example will pass the data arriving to the aggregator to the `o2-calibration-lhc-clockphase-workflow` device, and from there
+to the `o2-calibration-ccdb-populator-workflow` to update the CCDB. 
+In another terminal, run:
+```cpp
+source runEPNsimulation.sh 3
+```
+
+where the argument (`3` above) is the number of EPNs to be simulated. This will send the data to the aggregator process. 
+
+**N.B.**: the aggregator and calibration devices will need to use a different port from localHost:8080 in case the local
+CCDB server is used in its default configuration, since 8080 is used by CCDB. In the example above, port 30453 is used.

Detectors/Calibration/testMacros/runEPNsimulation.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+iEPN=0
+xpos_start=100
+while [[ $iEPN -lt $1 ]]
+do
+    echo $iEPN
+    xpos=$((xpos_start+1000*$iEPN))
+    xterm -hold -geometry 150x41+$xpos+300 -e bash -c "o2-calibration-data-generator-workflow --lanes 7 --gen-slot $iEPN --gen-norm 3 --mean-latency 100000 --max-timeframes 500 -b | o2-dpl-output-proxy --channel-config "name=downstream,method=connect,address=tcp://localhost:30453,type=push,transport=zeromq" --dataspec downstream:TOF/CALIBDATA -b ; exec bash" &
+    ((iEPN = iEPN +1 ))
+done

Detectors/TOF/calibration/testWorkflow/DataGeneratorSpec.h

@@ -49,9 +49,12 @@ class TFDispatcher : public o2::framework::Task
         LOG(INFO) << "Data generator reached TF " << tfid << ", stopping";
         pc.services().get<o2::framework::ControlService>().endOfStream();
         pc.services().get<o2::framework::ControlService>().readyToQuit(o2::framework::QuitRequest::Me);
+        if (!acceptTF(tfid)) {
+          return;
+        }
         break;
       }
-      if (((tfid / mNLanes) % mNGen) != mSlot) {
+      if (!acceptTF(tfid)) {
         return;
       }
     }
@@ -60,6 +63,20 @@ class TFDispatcher : public o2::framework::Task
     pc.outputs().snapshot(o2::framework::OutputRef{"output", 0}, size);
   }
 
+  bool acceptTF(int tfid)
+  {
+
+    // check if the current TF should be processed by this instance of the generator
+
+    int targetSlot = (tfid / mNLanes) % mNGen;
+    if (targetSlot != mSlot) {
+      LOG(INFO) << "tfid = " << tfid << ", mNLanes = " << mNLanes << ", mNGen = " << mNGen << ", mSlot = " << mSlot << " target slot = " << targetSlot << ": discarded";
+      return false;
+    }
+    LOG(INFO) << "tfid = " << tfid << ", mNLanes = " << mNLanes << ", mNGen = " << mNGen << ", mSlot = " << mSlot << " target slot = " << targetSlot << ": accepted";
+    return true;
+  }
+
  private:
   uint64_t mMaxTF = 1;
   int mSlot = 1;