Add option to publish after a certain amount of processed TFs

Author: wiechula

Detectors/TPC/workflow/include/TPCWorkflow/CalDetMergerPublisherSpec.h

@@ -22,7 +22,7 @@ namespace o2
 namespace tpc
 {
 
-o2::framework::DataProcessorSpec getCalDetMergerPublisherSpec(bool skipCCDB);
+o2::framework::DataProcessorSpec getCalDetMergerPublisherSpec(uint32_t lanes, bool skipCCDB, bool dumpAfterComplete = false);
 
 } // namespace tpc
 } // namespace o2

Detectors/TPC/workflow/include/TPCWorkflow/TPCCalibPedestalSpec.h

@@ -46,7 +46,7 @@ namespace tpc
 class TPCCalibPedestalDevice : public o2::framework::Task
 {
  public:
-  TPCCalibPedestalDevice(int lane, const std::vector<int>& sectors) : mLane{lane}, mSectors(sectors) {}
+  TPCCalibPedestalDevice(uint32_t lane, const std::vector<int>& sectors, uint32_t publishAfterTFs) : mLane{lane}, mSectors(sectors), mPublishAfter(publishAfterTFs) {}
 
   void init(o2::framework::InitContext& ic) final
   {
@@ -87,11 +87,20 @@ class TPCCalibPedestalDevice : public o2::framework::Task
     calib_processing_helper::processRawData(pc.inputs(), reader, mUseOldSubspec, mSectors);
 
     mCalibPedestal.incrementNEvents();
-    LOGP(info, "Number of processed events: {} ({})", mCalibPedestal.getNumberOfProcessedEvents(), mMaxEvents);
+    const auto nTFs = mCalibPedestal.getNumberOfProcessedEvents();
+    LOGP(info, "Number of processed TFs: {} ({})", nTFs, mMaxEvents);
 
-    if ((mCalibPedestal.getNumberOfProcessedEvents() >= mMaxEvents) && !mCalibDumped) {
-      LOGP(info, "Maximm number of events reached ({}), no more processing will be done", mMaxEvents);
+    if ((mPublishAfter && (nTFs % mPublishAfter) == 0)) {
+      LOGP(info, "Publishing after {} TFs", nTFs);
+      mCalibPedestal.analyse();
+      dumpCalibData();
+      sendOutput(pc.outputs());
+    }
+
+    if (mMaxEvents && (mCalibPedestal.getNumberOfProcessedEvents() >= mMaxEvents) && !mCalibDumped) {
+      LOGP(info, "Maximm number of TFs reached ({}), no more processing will be done", mMaxEvents);
       mReadyToQuit = true;
+      mCalibPedestal.analyse();
       dumpCalibData();
       if (mForceQuit) {
         pc.services().get<ControlService>().endOfStream();
@@ -105,6 +114,7 @@ class TPCCalibPedestalDevice : public o2::framework::Task
   void endOfStream(o2::framework::EndOfStreamContext& ec) final
   {
     LOGP(info, "endOfStream");
+    mCalibPedestal.analyse();
     dumpCalibData();
     sendOutput(ec.outputs());
     ec.services().get<ControlService>().readyToQuit(QuitRequest::Me);
@@ -113,8 +123,9 @@ class TPCCalibPedestalDevice : public o2::framework::Task
  private:
   CalibPedestal mCalibPedestal;
   rawreader::RawReaderCRUManager mRawReader;
-  uint32_t mMaxEvents{100};    ///< maximum number of events to process
-  int mLane{0};                ///< lane number of processor
+  uint32_t mMaxEvents{0};      ///< maximum number of events to process
+  uint32_t mPublishAfter{0};   ///< number of events after which to dump the calibration
+  uint32_t mLane{0};           ///< lane number of processor
   std::vector<int> mSectors{}; ///< sectors to process in this instance
   bool mReadyToQuit{false};    ///< if processor is ready to quit
   bool mCalibDumped{false};    ///< if calibration object already dumped
@@ -134,7 +145,7 @@ class TPCCalibPedestalDevice : public o2::framework::Task
       auto image = o2::utils::MemFileHelper::createFileImage(cal, typeid(*cal), cal->getName(), "data");
       int type = int(dataType[i]);
 
-      header::DataHeader::SubSpecificationType subSpec{(header::DataHeader::SubSpecificationType)i};
+      header::DataHeader::SubSpecificationType subSpec{(header::DataHeader::SubSpecificationType)((mLane << 4) + i)};
       output.snapshot(Output{clbUtils::gDataOriginCLB, "TPCCLBPART", subSpec}, *image.get());
       output.snapshot(Output{clbUtils::gDataOriginCLB, "TPCCLBPARTINFO", subSpec}, type);
     }
@@ -145,14 +156,13 @@ class TPCCalibPedestalDevice : public o2::framework::Task
   {
     if (mDirectFileDump && !mCalibDumped) {
       LOGP(info, "Dumping output");
-      mCalibPedestal.analyse();
       mCalibPedestal.dumpToFile(fmt::format("pedestals_{:02}.root", mLane));
       mCalibDumped = true;
     }
   }
 };
 
-DataProcessorSpec getTPCCalibPedestalSpec(const std::string inputSpec, int ilane = 0, std::vector<int> sectors = {})
+DataProcessorSpec getTPCCalibPedestalSpec(const std::string inputSpec, uint32_t ilane = 0, std::vector<int> sectors = {}, uint32_t publishAfterTFs = 0)
 {
   std::vector<o2::framework::OutputSpec> outputs;
   outputs.emplace_back(ConcreteDataTypeMatcher{clbUtils::gDataOriginCLB, "TPCCLBPART"});
@@ -163,9 +173,9 @@ DataProcessorSpec getTPCCalibPedestalSpec(const std::string inputSpec, int ilane
     id.data(),
     select(inputSpec.data()),
     outputs,
-    AlgorithmSpec{adaptFromTask<TPCCalibPedestalDevice>(ilane, sectors)},
+    AlgorithmSpec{adaptFromTask<TPCCalibPedestalDevice>(ilane, sectors, publishAfterTFs)},
     Options{
-      {"max-events", VariantType::Int, 100, {"maximum number of events to process"}},
+      {"max-events", VariantType::Int, 0, {"maximum number of events to process"}},
       {"use-old-subspec", VariantType::Bool, false, {"use old subsecifiation definition"}},
       {"force-quit", VariantType::Bool, false, {"force quit after max-events have been reached"}},
       {"direct-file-dump", VariantType::Bool, false, {"directly dump calibration to file"}},

Detectors/TPC/workflow/src/CalDetMergerPublisherSpec.cxx

@@ -12,6 +12,7 @@
 /// @brief  TPC CalDet merger and CCDB publisher
 /// @author Jens Wiechula, Jens.Wiechula@ikf.uni-frankfurt.de
 
+#include <bitset>
 #include <unordered_map>
 #include <vector>
 #include <string>
@@ -45,7 +46,7 @@ class CalDetMergerPublisherSpec : public o2::framework::Task
   using CcdbObjectInfo = o2::ccdb::CcdbObjectInfo;
 
  public:
-  CalDetMergerPublisherSpec(bool skipCCDB) : mSkipCCDB(skipCCDB) {}
+  CalDetMergerPublisherSpec(uint32_t lanes, bool skipCCDB, bool dumpAfterComplete = false) : mLanesToExpect(lanes), mSkipCCDB(skipCCDB), mPublishAfterComplete(dumpAfterComplete) {}
 
   void init(o2::framework::InitContext& ic) final
   {
@@ -58,10 +59,13 @@ class CalDetMergerPublisherSpec : public o2::framework::Task
     int nSlots = pc.inputs().getNofParts(0);
     assert(pc.inputs().getNofParts(1) == nSlots);
 
-    LOGP(info, "CalDetMergerPublisherSpec run");
     for (int isl = 0; isl < nSlots; isl++) {
       const auto type = pc.inputs().get<int>("clbInfo", isl);
       const auto pld = pc.inputs().get<gsl::span<char>>("clbPayload", isl); // this is actually an image of TMemFile
+      const auto* dh = DataRefUtils::getHeader<o2::header::DataHeader*>(pc.inputs().get("clbInfo", isl));
+      const auto subSpec = dh->subSpecification;
+      const int lane = subSpec >> 4;
+      const int calibType = subSpec & 0xf;
 
       //const auto& path = wrp->getPath();
       TMemFile f("file", (char*)&pld[0], pld.size(), "READ");
@@ -77,13 +81,28 @@ class CalDetMergerPublisherSpec : public o2::framework::Task
       }
       f.Close();
 
-      LOGP(info, "getting slot {}", isl);
+      LOGP(info, "getting slot {}, subspec {:#8x}, lane {}, type {}", isl, subSpec, lane, calibType);
+      //if (mReceivedLanes.test(lane)) {
+      //LOGP(warning, "lane {} received multiple times", lane);
+      //}
+      mReceivedLanes.set(lane);
+    }
+
+    if (mReceivedLanes.count() == mLanesToExpect) {
+      LOGP(info, "data of all lanes received");
+      if (mPublishAfterComplete) {
+        LOGP(info, "publishing after all data was received");
+        dumpCalibData();
+        sendOutput(pc.outputs());
+        mCalibDumped = false;
+      }
+      mReceivedLanes.reset();
     }
   }
 
   void endOfStream(o2::framework::EndOfStreamContext& ec) final
   {
-    LOGP(info, "CalDetMergerPublisherSpec endOfStream");
+    LOGP(info, "endOfStream");
 
     dumpCalibData();
     sendOutput(ec.outputs());
@@ -92,9 +111,12 @@ class CalDetMergerPublisherSpec : public o2::framework::Task
 
  private:
   using dataType = o2::tpc::CalDet<float>;
+  std::bitset<128> mReceivedLanes;                  ///< counter for received lanes
   std::unordered_map<int, dataType> mMergedCalDets; ///< calibration data to merge
+  uint32_t mLanesToExpect{0};                       ///< number of expected lanes sending data
   bool mForceQuit{false};                           ///< for quit after processing finished
   bool mDirectFileDump{false};                      ///< directly dump the calibration data to file
+  bool mPublishAfterComplete{false};                ///< dump calibration directly after data from all lanes received
   bool mCalibDumped{false};                         ///< if calibration object already dumped
   bool mSkipCCDB{false};                            ///< skip sending of calibration data
 
@@ -129,7 +151,7 @@ class CalDetMergerPublisherSpec : public o2::framework::Task
   void dumpCalibData()
   {
     if (mDirectFileDump && !mCalibDumped) {
-      LOGP(info, "CalDetMergerPublisherSpec Dumping output");
+      LOGP(info, "Dumping output to file");
       TFile f("merged_CalDet.root", "recreate");
       for (auto& [type, object] : mMergedCalDets) {
         f.WriteObject(&object, object.getName().data());
@@ -139,7 +161,7 @@ class CalDetMergerPublisherSpec : public o2::framework::Task
   }
 };
 
-o2::framework::DataProcessorSpec o2::tpc::getCalDetMergerPublisherSpec(bool skipCCDB)
+o2::framework::DataProcessorSpec o2::tpc::getCalDetMergerPublisherSpec(uint32_t lanes, bool skipCCDB, bool dumpAfterComplete)
 {
   std::vector<OutputSpec> outputs;
   if (!skipCCDB) {
@@ -157,7 +179,7 @@ o2::framework::DataProcessorSpec o2::tpc::getCalDetMergerPublisherSpec(bool skip
     id.data(),
     inputs,
     outputs,
-    AlgorithmSpec{adaptFromTask<CalDetMergerPublisherSpec>(skipCCDB)},
+    AlgorithmSpec{adaptFromTask<CalDetMergerPublisherSpec>(lanes, skipCCDB, dumpAfterComplete)},
     Options{
       {"force-quit", VariantType::Bool, false, {"force quit after max-events have been reached"}},
       {"direct-file-dump", VariantType::Bool, false, {"directly dump calibration to file"}},

Detectors/TPC/workflow/src/tpc-calib-pedestal.cxx

@@ -46,10 +46,11 @@ void customize(std::vector<o2::framework::CompletionPolicy>& policies)
 void customize(std::vector<ConfigParamSpec>& workflowOptions)
 {
   std::string sectorDefault = "0-" + std::to_string(o2::tpc::Sector::MAXSECTOR - 1);
-  int defaultlanes = 1; //std::max(1u, std::thread::hardware_concurrency() / 2);
+  int defaultlanes = std::max(1u, std::thread::hardware_concurrency() / 2);
 
   std::vector<ConfigParamSpec> options{
     {"input-spec", VariantType::String, "A:TPC/RAWDATA", {"selection string input specs"}},
+    {"publish-after-tfs", VariantType::Int, 0, {"number of time frames after which to force publishing the objects"}},
     {"configKeyValues", VariantType::String, "", {"Semicolon separated key=value strings (e.g.: 'TPCCalibPedestal.FirstTimeBin=10;...')"}},
     {"configFile", VariantType::String, "", {"configuration file for configurable parameters"}},
     {"no-write-ccdb", VariantType::Bool, false, {"skip sending the calibration output to CCDB"}},
@@ -75,10 +76,11 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
 
   const std::string inputSpec = config.options().get<std::string>("input-spec");
   const auto skipCCDB = config.options().get<bool>("no-write-ccdb");
+  const auto publishAfterTFs = (uint32_t)config.options().get<int>("publish-after-tfs");
 
   const auto tpcsectors = o2::RangeTokenizer::tokenize<int>(config.options().get<std::string>("sectors"));
-  const auto nSectors = (int)tpcsectors.size();
-  const auto nLanes = std::min(config.options().get<int>("lanes"), nSectors);
+  const auto nSectors = (uint32_t)tpcsectors.size();
+  const auto nLanes = std::min((uint32_t)config.options().get<int>("lanes"), nSectors);
   const auto sectorsPerLane = nSectors / nLanes + ((nSectors % nLanes) != 0);
 
   WorkflowSpec workflow;
@@ -94,10 +96,10 @@ WorkflowSpec defineDataProcessing(ConfigContext const& config)
     }
     auto last = std::min(tpcsectors.end(), first + sectorsPerLane);
     std::vector<int> range(first, last);
-    workflow.emplace_back(getTPCCalibPedestalSpec(inputSpec, ilane, range));
+    workflow.emplace_back(getTPCCalibPedestalSpec(inputSpec, ilane, range, publishAfterTFs));
   }
 
-  workflow.emplace_back(getCalDetMergerPublisherSpec(skipCCDB));
+  workflow.emplace_back(getCalDetMergerPublisherSpec(nLanes, skipCCDB, publishAfterTFs > 0));
 
   return workflow;
 }