Parallel processing and mergin of pedestal calibration in DPL

o Possibility to run processing of sector selections
o Possibility to run parallel processes
o Add merger for CalDet objects, which can send data to ccdb-populator

Author: wiechula

Detectors/TPC/base/src/TPCBaseLinkDef.h

@@ -27,6 +27,7 @@
 #pragma link C++ class o2::tpc::CalDet < short> + ;
 #pragma link C++ class o2::tpc::CalDet < bool> + ;
 #pragma link C++ class std::vector < o2::tpc::CalDet < float>> + ;
+#pragma link C++ class std::vector < o2::tpc::CalDet < float>*> + ;
 #pragma link C++ class std::unordered_map < std::string, o2::tpc::CalDet < float>> + ;
 #pragma link C++ class o2::tpc::CDBInterface;
 #pragma link C++ class o2::tpc::ContainerFactory;

Detectors/TPC/calibration/include/TPCCalibration/CalibPedestal.h

@@ -105,6 +105,9 @@ class CalibPedestal : public CalibRawBase
   /// \return noise calibration object
   const CalPad& getNoise() const { return mNoise; }
 
+  /// return all pad clibrations as vector
+  const std::vector<const o2::tpc::CalDet<float>*> getCalDets() const { return std::vector<const o2::tpc::CalDet<float>*>{&mPedestal, &mNoise}; }
+
   /// Get the statistics type
   StatisticsType getStatisticsType() const { return mStatisticsType; }
 

Detectors/TPC/workflow/CMakeLists.txt

@@ -26,6 +26,7 @@ o2_add_library(TPCWorkflow
                        src/CalibProcessingHelper.cxx
                        src/ClusterSharingMapSpec.cxx
                        src/FileReaderWorkflow.cxx
+                       src/CalDetMergerPublisherSpec.cxx
                TARGETVARNAME targetName
                PUBLIC_LINK_LIBRARIES O2::Framework O2::DataFormatsTPC
                                      O2::DPLUtils O2::TPCReconstruction

Detectors/TPC/workflow/README.md

@@ -152,10 +152,12 @@ bz=     magnetic field
 ### Pedestal calibration
 #### Options
 ```bash
---direct-file-dump write final calibration to local root file
---max-events       maximum number of events to process
---no-calib-output  don't send the calibration data via DPL (required in case the calibration write is not attached)
---use-old-subspec  Subspecification is built from RDH like 
+--direct-file-dump     write final calibration to local root file
+--max-events           maximum number of events to process
+--no-write-ccdb        don't send the calibration data via DPL (required in case the calibration write is not attached)
+--use-old-subspec      use old subspec definition (CruId << 16) | ((LinkId + 1) << (CruEndPoint == 1 ? 8 : 0))
+--lanes arg (=1)       number of parallel processes
+--sectors arg (=0-35)  list of TPC sectors, comma separated ranges, e.g. 0-3,7,9-15
 ```
 
 #### Running with data distribution
@@ -183,7 +185,7 @@ o2-raw-file-reader-workflow  --input-conf raw-reader.cfg --nocheck-hbf-per-tf --
 ```
 
 #### Send data to CCDB
-Remove the `--no-calib-output` option and add
+Remove the `--no-write-ccdb` option and add
 ```bash
 | o2-calibration-ccdb-populator-workflow
 ```

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

@@ -0,0 +1,30 @@
+// Copyright CERN and copyright holders of ALICE O2. This software is
+// distributed under the terms of the GNU General Public License v3 (GPL
+// Version 3), copied verbatim in the file "COPYING".
+//
+// See http://alice-o2.web.cern.ch/license for full licensing information.
+//
+// In applying this license CERN does not waive the privileges and immunities
+// granted to it by virtue of its status as an Intergovernmental Organization
+// or submit itself to any jurisdiction.
+
+#ifndef O2_TPC_CalDetMergerPublisherSpec_H
+#define O2_TPC_CalDetMergerPublisherSpec_H
+
+/// @file   CalDetMergerPublisherSpec.h
+/// @brief  TPC CalDet merger and CCDB publisher
+/// @author Jens Wiechula, Jens.Wiechula@ikf.uni-frankfurt.de
+
+#include "Framework/DataProcessorSpec.h"
+
+namespace o2
+{
+namespace tpc
+{
+
+o2::framework::DataProcessorSpec getCalDetMergerPublisherSpec(bool skipCCDB);
+
+} // namespace tpc
+} // namespace o2
+
+#endif

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

@@ -28,7 +28,7 @@ class RawReaderCRU;
 namespace calib_processing_helper
 {
 
-uint64_t processRawData(o2::framework::InputRecord& inputs, std::unique_ptr<RawReaderCRU>& reader, bool useOldSubspec = false);
+uint64_t processRawData(o2::framework::InputRecord& inputs, std::unique_ptr<RawReaderCRU>& reader, bool useOldSubspec = false, const std::vector<int>& sectors = {});
 } // namespace calib_processing_helper
 } // namespace tpc
 } // namespace o2

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

@@ -17,14 +17,16 @@
 #include <vector>
 #include <string>
 #include <chrono>
+#include <fmt/format.h>
 
 #include "Framework/Task.h"
 #include "Framework/ControlService.h"
 #include "Framework/Logger.h"
 #include "Framework/ConfigParamRegistry.h"
+#include "Framework/DataProcessorSpec.h"
 
+#include "CommonUtils/MemFileHelper.h"
 #include "Headers/DataHeader.h"
-#include "CCDB/CcdbApi.h"
 #include "DetectorsCalibration/Utils.h"
 
 #include "TPCBase/CDBInterface.h"
@@ -38,13 +40,13 @@ using clbUtils = o2::calibration::Utils;
 
 namespace o2
 {
-namespace calibration
+namespace tpc
 {
 
 class TPCCalibPedestalDevice : public o2::framework::Task
 {
  public:
-  TPCCalibPedestalDevice(bool skipCalib) : mSkipCalib(skipCalib) {}
+  TPCCalibPedestalDevice(int lane, const std::vector<int>& sectors) : mLane{lane}, mSectors(sectors) {}
 
   void init(o2::framework::InitContext& ic) final
   {
@@ -74,7 +76,7 @@ class TPCCalibPedestalDevice : public o2::framework::Task
     }
 
     auto& reader = mRawReader.getReaders()[0];
-    calib_processing_helper::processRawData(pc.inputs(), reader, mUseOldSubspec);
+    calib_processing_helper::processRawData(pc.inputs(), reader, mUseOldSubspec, mSectors);
 
     mCalibPedestal.incrementNEvents();
     LOGP(info, "Number of processed events: {} ({})", mCalibPedestal.getNumberOfProcessedEvents(), mMaxEvents);
@@ -96,51 +98,37 @@ class TPCCalibPedestalDevice : public o2::framework::Task
   {
     LOGP(info, "endOfStream");
     dumpCalibData();
-    if (!mSkipCalib) {
-      sendOutput(ec.outputs());
-    }
+    sendOutput(ec.outputs());
     ec.services().get<ControlService>().readyToQuit(QuitRequest::Me);
   }
 
  private:
   CalibPedestal mCalibPedestal;
   rawreader::RawReaderCRUManager mRawReader;
-  uint32_t mMaxEvents{100};
-  bool mReadyToQuit{false};
-  bool mCalibDumped{false};
-  bool mUseOldSubspec{false};
-  bool mForceQuit{false};
-  bool mDirectFileDump{false};
-  bool mSkipCalib{false};
+  uint32_t mMaxEvents{100};    ///< maximum number of events to process
+  int 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
+  bool mUseOldSubspec{false};  ///< use the old subspec definition
+  bool mForceQuit{false};      ///< for quit after processing finished
+  bool mDirectFileDump{false}; ///< directly dump the calibration data to file
 
   //____________________________________________________________________________
   void sendOutput(DataAllocator& output)
   {
-    CDBStorage::MetaData_t md;
-
-    // perhaps should be changed to time of the run
-    const auto now = std::chrono::system_clock::now();
-    long timeStart = std::chrono::duration_cast<std::chrono::milliseconds>(now.time_since_epoch()).count();
-    long timeEnd = 99999999999999;
 
     std::array<const CalDet<float>*, 2> data = {&mCalibPedestal.getPedestal(), &mCalibPedestal.getNoise()};
     std::array<CDBType, 2> dataType = {CDBType::CalPedestal, CDBType::CalNoise};
 
     for (size_t i = 0; i < data.size(); ++i) {
       auto cal = data[i];
-      o2::ccdb::CcdbObjectInfo w;
-      auto image = o2::ccdb::CcdbApi::createObjectImage(cal, &w);
-
-      w.setPath(CDBTypeMap.at(dataType[i]));
-      w.setStartValidityTimestamp(timeStart);
-      w.setEndValidityTimestamp(timeEnd);
-
-      LOG(INFO) << "Sending object " << w.getPath() << "/" << w.getFileName() << " of size " << image->size()
-                << " bytes, valid for " << w.getStartValidityTimestamp() << " : " << w.getEndValidityTimestamp();
+      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};
-      output.snapshot(Output{clbUtils::gDataOriginCLB, clbUtils::gDataDescriptionCLBPayload, subSpec}, *image.get());
-      output.snapshot(Output{clbUtils::gDataOriginCLB, clbUtils::gDataDescriptionCLBInfo, subSpec}, w);
+      output.snapshot(Output{clbUtils::gDataOriginCLB, "TPCCLBPART", subSpec}, *image.get());
+      output.snapshot(Output{clbUtils::gDataOriginCLB, "TPCCLBPARTINFO", subSpec}, type);
     }
   }
 
@@ -150,32 +138,24 @@ class TPCCalibPedestalDevice : public o2::framework::Task
     if (mDirectFileDump && !mCalibDumped) {
       LOGP(info, "Dumping output");
       mCalibPedestal.analyse();
-      mCalibPedestal.dumpToFile("pedestals.root");
+      mCalibPedestal.dumpToFile(fmt::format("pedestals_{:02}.root", mLane));
       mCalibDumped = true;
     }
   }
 };
 
-} // namespace calibration
-
-namespace framework
+DataProcessorSpec getTPCCalibPedestalSpec(const std::string inputSpec, int ilane = 0, std::vector<int> sectors = {})
 {
+  std::vector<o2::framework::OutputSpec> outputs;
+  outputs.emplace_back(ConcreteDataTypeMatcher{clbUtils::gDataOriginCLB, "TPCCLBPART"});
+  outputs.emplace_back(ConcreteDataTypeMatcher{clbUtils::gDataOriginCLB, "TPCCLBPARTINFO"});
 
-DataProcessorSpec getTPCCalibPedestalSpec(const std::string inputSpec, bool skipCalib)
-{
-  using device = o2::calibration::TPCCalibPedestalDevice;
-
-  std::vector<OutputSpec> outputs;
-  if (!skipCalib) {
-    outputs.emplace_back(ConcreteDataTypeMatcher{clbUtils::gDataOriginCLB, clbUtils::gDataDescriptionCLBPayload});
-    outputs.emplace_back(ConcreteDataTypeMatcher{clbUtils::gDataOriginCLB, clbUtils::gDataDescriptionCLBInfo});
-  }
-
+  const auto id = fmt::format("calib-tpc-pedestal-{:02}", ilane);
   return DataProcessorSpec{
-    "calib-tpc-pedestal",
+    id.data(),
     select(inputSpec.data()),
     outputs,
-    AlgorithmSpec{adaptFromTask<device>(skipCalib)},
+    AlgorithmSpec{adaptFromTask<TPCCalibPedestalDevice>(ilane, sectors)},
     Options{
       {"max-events", VariantType::Int, 100, {"maximum number of events to process"}},
       {"use-old-subspec", VariantType::Bool, false, {"use old subsecifiation definition"}},
@@ -185,7 +165,7 @@ DataProcessorSpec getTPCCalibPedestalSpec(const std::string inputSpec, bool skip
   };  // end DataProcessorSpec
 }
 
-} // namespace framework
+} // namespace tpc
 } // namespace o2
 
 #endif