Refactor Late Tasks by adopting Actor

Author: knopers8

…de/QualityControl/LateTaskRunnerConfig.h …/include/QualityControl/LateTaskConfig.hFramework/include/QualityControl/LateTaskRunnerConfig.h renamed to Framework/include/QualityControl/LateTaskConfig.h Framework/include/QualityControl/LateTaskRunnerConfig.h renamed to Framework/include/QualityControl/LateTaskConfig.h

@@ -13,18 +13,18 @@
 #define QC_CORE_LATETASKRUNNERCONFIG_H
 
 #include "QualityControl/UserCodeConfig.h"
+#include "QualityControl/DataSourceSpec.h"
 
 #include <Framework/DataProcessorSpec.h>
 
+
 namespace o2::quality_control::core {
 
 // fixme: should it be LateTaskConfig instead?
-struct LateTaskRunnerConfig : public UserCodeConfig {
-  std::string taskName;
-  std::string deviceName;
-  framework::Inputs inputSpecs;
-  framework::OutputSpec moSpec;
-  framework::Options options;
+struct LateTaskConfig : public UserCodeConfig {
+
+  // fixme: we should think what should happen when a user wants a critical late task relying on a non-critical QC task.
+  //  should it be marked as resilient?
   bool critical = true;
 };
 

Framework/include/QualityControl/LateTaskFactory.h

@@ -30,7 +30,7 @@ class LateTaskFactory
 
   /// \brief Create a new instance of a LateTaskInterface.
   /// The LateTaskInterface actual class is decided based on the parameters passed.
-  static LateTaskInterface* create(const LateTaskRunnerConfig& taskConfig, std::shared_ptr<ObjectsManager> objectsManager)
+  static LateTaskInterface* create(const LateTaskConfig& taskConfig, std::shared_ptr<ObjectsManager> objectsManager)
   {
     auto* result = root_class_factory::create<LateTaskInterface>(taskConfig.moduleName, taskConfig.className);
     result->setName(taskConfig.name);

Framework/include/QualityControl/LateTaskRunner.h

@@ -17,83 +17,47 @@
 /// \author Piotr Konopka
 ///
 
-#include <Framework/Task.h>
-#include <Framework/DataProcessorSpec.h>
-#include <Framework/CompletionPolicy.h>
+#include "QualityControl/LateTaskConfig.h"
+#include "QualityControl/LateTaskRunnerTraits.h"
 
-#include "QualityControl/LateTaskRunnerConfig.h"
+#include "QualityControl/ActorTraits.h"
+#include "QualityControl/Actor.h"
+#include <string_view>
+#include <memory>
 
 namespace o2::quality_control::core {
 
 class LateTaskInterface;
 class ObjectsManager;
 
-class LateTaskRunner : public framework::Task
+class LateTaskRunner : public Actor<LateTaskRunner>
 {
-  /// \brief Number of bytes in data description used for hashing of Task names. See HashDataDescription.h for details
-  static constexpr size_t taskDescriptionHashLength = 4;
-  static_assert(taskDescriptionHashLength <= o2::header::DataDescription::size);
-
  public:
-  explicit LateTaskRunner(const LateTaskRunnerConfig& config);
-  ~LateTaskRunner() override = default;
-
-  /// \brief LateTaskRunner's init callback
-  void init(framework::InitContext& iCtx) override;
-  /// \brief LateTaskRunner's process callback
-  void run(framework::ProcessingContext& pCtx) override;
-  /// \brief LateTaskRunner's finaliseCCDB callback
-  // void finaliseCCDB(framework::ConcreteDataMatcher& matcher, void* obj) override;
-
-  std::string getDeviceName() const { return mTaskConfig.deviceName; };
-  const framework::Inputs& getInputsSpecs() const { return mTaskConfig.inputSpecs; };
-  const framework::OutputSpec& getOutputSpec() const { return mTaskConfig.moSpec; };
-  const framework::Options& getOptions() const { return mTaskConfig.options; };
+  LateTaskRunner(const ServicesConfig& servicesConfig, const LateTaskConfig& config);
+  ~LateTaskRunner() = default;
 
-  /// \brief Data Processor Label to identify all Late Task Runners
-  static framework::DataProcessorLabel getLabel() { return { "qc-late-task" }; }
-  /// \brief ID string for all LateTaskRunner devices
-  static std::string createIdString();
-  /// \brief Unified DataOrigin for Quality Control tasks
-  static header::DataOrigin createDataOrigin(const std::string& detectorCode);
-  /// \brief Unified DataDescription naming scheme for all tasks
-  static header::DataDescription createDataDescription(const std::string& taskName);
+  void onStart(framework::ServiceRegistryRef services);
+  void onInit(framework::InitContext& iCtx);
+  void onProcess(framework::ProcessingContext& pCtx);
 
-  /// \brief Callback for CallbackService::Id::EndOfStream
-  // void endOfStream(framework::EndOfStreamContext& eosContext) override;
+  std::string_view getDetectorName() const { return mTaskConfig.detectorName; }
+  std::string_view getUserCodeName() const { return mTaskConfig.name; }
+  bool isCritical() const { return mTaskConfig.critical; }
 
  private:
-  /// \brief Callback for CallbackService::Id::Start (DPL) a.k.a. RUN transition (FairMQ)
-  // void start(framework::ServiceRegistryRef services);
-  /// \brief Callback for CallbackService::Id::Stop (DPL) a.k.a. STOP transition (FairMQ)
-  // void stop(framework::ServiceRegistryRef services);
-  /// \brief Callback for CallbackService::Id::Reset (DPL) a.k.a. RESET DEVICE transition (FairMQ)
-  // void reset();
 
   void startOfActivity();
   void endOfActivity();
   int publish(framework::DataAllocator& outputs);
-  // void publishCycleStats();
-  // void updateMonitoringStats(framework::ProcessingContext& pCtx);
-  // void registerToBookkeeping();
 
  private:
-  LateTaskRunnerConfig mTaskConfig;
-  // std::shared_ptr<monitoring::Monitoring> mCollector;
+  LateTaskConfig mTaskConfig;
   std::shared_ptr<LateTaskInterface> mTask;
   std::shared_ptr<ObjectsManager> mObjectsManager;
-  // std::shared_ptr<Timekeeper> mTimekeeper;
-  Activity mActivity;
   ValidityInterval mValidity;
-
-  // stats
-  // int mNumberMessagesReceivedInCycle = 0;
-  // int mNumberObjectsPublishedInCycle = 0;
-  // int mTotalNumberObjectsPublished = 0; // over a run
-  // double mLastPublicationDuration = 0;
-  // uint64_t mDataReceivedInCycle = 0;
 };
 
 
+
 }
 #endif //QC_CORE_LATETASKRUNNER_H

Framework/include/QualityControl/LateTaskRunnerFactory.h

@@ -20,7 +20,7 @@
 #include <vector>
 #include <optional>
 
-#include "QualityControl/LateTaskRunnerConfig.h"
+#include "QualityControl/LateTaskConfig.h"
 #include <Framework/DataProcessorSpec.h>
 
 namespace o2::framework
@@ -32,8 +32,9 @@ namespace o2::quality_control::core
 {
 
 struct LateTaskSpec;
-struct LateTaskRunnerConfig;
+struct LateTaskConfig;
 struct CommonSpec;
+struct ServicesConfig;
 
 /// \brief Factory in charge of creating DataProcessorSpec of LateTaskRunner
 class LateTaskRunnerFactory
@@ -45,10 +46,10 @@ class LateTaskRunnerFactory
   /// \brief Creates LateTaskRunner
   ///
   /// \param taskConfig
-  static o2::framework::DataProcessorSpec create(const LateTaskRunnerConfig&);
+  static o2::framework::DataProcessorSpec create(const ServicesConfig& ServicesConfig, const LateTaskConfig& taskConfig);
 
   /// \brief Knows how to create LateTaskConfig from Specs
-  static LateTaskRunnerConfig extractConfig(const CommonSpec&, const LateTaskSpec&);
+  static LateTaskConfig extractConfig(const CommonSpec&, const LateTaskSpec&);
 
   /// \brief Provides necessary customization of the LateTaskRunners.
   ///

Framework/include/QualityControl/LateTaskRunnerTraits.h

@@ -0,0 +1,51 @@
+// Copyright 2019-2020 CERN and copyright holders of ALICE O2.
+// See https://alice-o2.web.cern.ch/copyright for details of the copyright holders.
+// All rights not expressly granted are reserved.
+//
+// This software is distributed under the terms of the GNU General Public
+// License v3 (GPL Version 3), copied verbatim in the file "COPYING".
+//
+// 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 QUALITYCONTROL_LATETASKRUNNERTRAITS_H
+#define QUALITYCONTROL_LATETASKRUNNERTRAITS_H
+
+#include "QualityControl/ActorTraits.h"
+
+#include <string_view>
+#include <array>
+#include "QualityControl/DataSourceSpec.h"
+
+namespace o2::quality_control::core {
+
+class LateTaskRunner;
+
+template <>
+struct ActorTraits<LateTaskRunner>
+{
+  constexpr static std::string_view sActorTypeShort{"late"};
+  constexpr static std::string_view sActorTypeKebabCase{"qc-late-task"};
+  constexpr static std::string_view sActorTypeUpperCamelCase{"LateTaskRunner"};
+
+  constexpr static size_t sDataDescriptionHashLength{4};
+
+  constexpr static std::array<DataSourceType, 5> sConsumedDataSources{
+    DataSourceType::Task, DataSourceType::TaskMovingWindow, DataSourceType::Check, DataSourceType::Aggregator, DataSourceType::LateTask
+   };
+  constexpr static std::array<DataSourceType, 1> sPublishedDataSources{DataSourceType::LateTask};
+
+  constexpr static std::array<Service, 2> sRequiredServices{Service::InfoLogger, Service::Monitoring};
+  constexpr static bkp::DplProcessType sDplProcessType{bkp::DplProcessType::QC_POSTPROCESSING}; // todo at some point we should add a new type in BKP
+
+  constexpr static UserCodeInstanceCardinality sUserCodeInstanceCardinality{UserCodeInstanceCardinality::One};
+  constexpr static bool sDetectorSpecific{true};
+  constexpr static Criticality sCriticality{Criticality::UserDefined};
+};
+
+static_assert(ValidActorTraits<ActorTraits<LateTaskRunner>>);
+
+}
+
+#endif //QUALITYCONTROL_LATETASKRUNNERTRAITS_H

Framework/src/InfrastructureGenerator.cxx

@@ -52,6 +52,9 @@
 #include <vector>
 #include <ranges>
 
+#include "QualityControl/ActorHelpers.h"
+#include "QualityControl/InputUtils.h"
+
 using namespace o2::framework;
 using namespace o2::configuration;
 using namespace o2::mergers;
@@ -657,10 +660,7 @@ void InfrastructureGenerator::generateCheckRunners(framework::WorkflowSpec& work
   }
 
   for (const auto& lateTaskSpec : infrastructureSpec.lateTasks | std::views::filter(&LateTaskSpec::active)) {
-    InputSpec lateTaskOutput{ lateTaskSpec.taskName,
-                            LateTaskRunner::createDataOrigin(lateTaskSpec.detectorName),
-                            LateTaskRunner::createDataDescription(lateTaskSpec.taskName),
-                            Lifetime::Sporadic };
+    InputSpec lateTaskOutput = createUserInputSpec<LateTaskRunner, DataSourceType::LateTask>(lateTaskSpec.detectorName, lateTaskSpec.taskName);
     tasksOutputMap.insert({ DataSpecUtils::label(lateTaskOutput), lateTaskOutput });
   }
 
@@ -841,7 +841,10 @@ void InfrastructureGenerator::generateLateTasks(framework::WorkflowSpec& workflo
   }
 
   for (const auto& lateTaskSpec : infrastructureSpec.lateTasks | std::views::filter(&LateTaskSpec::active)) {
-    DataProcessorSpec spec = LateTaskRunnerFactory::create(LateTaskRunnerFactory::extractConfig(infrastructureSpec.common, lateTaskSpec));
+    auto servicesConfig = actor_helpers::extractConfig(infrastructureSpec.common);
+    auto lateTaskConfig = LateTaskRunnerFactory::extractConfig(infrastructureSpec.common, lateTaskSpec);
+
+    DataProcessorSpec spec = LateTaskRunnerFactory::create(servicesConfig, lateTaskConfig);
     workflow.emplace_back(std::move(spec));
   }
 }

Framework/src/InfrastructureSpecReader.cxx

@@ -15,11 +15,13 @@
 
 #include "QualityControl/InfrastructureSpecReader.h"
 #include "QualityControl/QcInfoLogger.h"
+#include "QualityControl/LateTaskRunnerTraits.h"
 #include "QualityControl/TaskRunner.h"
 #include "QualityControl/PostProcessingDevice.h"
 #include "QualityControl/Check.h"
 #include "QualityControl/AggregatorRunner.h"
 #include "QualityControl/LateTaskRunner.h"
+#include "QualityControl/InputUtils.h"
 
 #include <DataSampling/DataSampling.h>
 #include <Framework/DataDescriptorQueryBuilder.h>
@@ -301,7 +303,7 @@ DataSourceSpec InfrastructureSpecReader::readSpecEntry<DataSourceSpec>(const std
       dss.name = wholeTree.get<std::string>("qc.lateTasks." + dss.id + ".taskName", dss.id);
       auto detectorName = wholeTree.get<std::string>("qc.lateTasks." + dss.id + ".detectorName");
 
-      dss.inputs = { { dss.name, LateTaskRunner::createDataOrigin(detectorName), LateTaskRunner::createDataDescription(dss.name), 0, Lifetime::Sporadic } };
+      dss.inputs = { createUserInputSpec<LateTaskRunner, DataSourceType::LateTask>(detectorName, dss.name) };
       if (dataSourceTree.count("MOs") > 0) {
         for (const auto& moName : dataSourceTree.get_child("MOs")) {
           dss.subInputs.push_back(moName.second.get_value<std::string>());

Framework/src/LateTaskRunner.cxx

@@ -56,17 +56,19 @@ using namespace o2::monitoring;
 namespace o2::quality_control::core
 {
 
-LateTaskRunner::LateTaskRunner(const LateTaskRunnerConfig& config)
-  : mTaskConfig(config)
+LateTaskRunner::LateTaskRunner(const ServicesConfig& servicesConfig, const LateTaskConfig& config)
+: Actor<LateTaskRunner>(servicesConfig), mTaskConfig(config)
 {
+  // fixme: this should be moved to Actor
   o2::ccdb::BasicCCDBManager::instance().setFatalWhenNull(false);
 }
 
-void LateTaskRunner::init(InitContext& iCtx)
+void LateTaskRunner::onInit(InitContext& iCtx)
 {
+  // fixme: move exception handling to Actor
   try {
     // setup publisher
-    mObjectsManager = std::make_shared<ObjectsManager>(mTaskConfig.taskName, mTaskConfig.className, mTaskConfig.detectorName, 0);
+    mObjectsManager = std::make_shared<ObjectsManager>(mTaskConfig.name, mTaskConfig.className, mTaskConfig.detectorName, 0);
 
     // setup user's task
     mTask.reset(LateTaskFactory::create(mTaskConfig, mObjectsManager));
@@ -78,14 +80,15 @@ void LateTaskRunner::init(InitContext& iCtx)
     mValidity = gInvalidValidityInterval;
 
     // todo move to start
-    mTask->startOfActivity({});
+    mTask->startOfActivity(getActivity());
+    mObjectsManager->setActivity(getActivity());
   } catch (boost::exception& e) {
     ILOG(Info, Devel) << "exception during init " << diagnostic_information(e) << ENDM;
     throw;
   }
 }
 
-void LateTaskRunner::run(ProcessingContext& pCtx)
+void LateTaskRunner::onProcess(ProcessingContext& pCtx)
 {
   // todo: derive from received objects
   mValidity.update(getCurrentTimestamp());
@@ -101,7 +104,7 @@ void LateTaskRunner::run(ProcessingContext& pCtx)
 
     // fixme: come up with an elegant way for this. LateTask should be probably aware of the requested user inputs aside from just InputSpecs
     //  also, we should filter only expect objects from the received collections
-    if (dataOrigin.str[0] == 'Q' || dataOrigin.str[0] == 'W') { // main MOs and moving windows from QC tasks
+    if (dataOrigin.str[0] == 'Q' || dataOrigin.str[0] == 'W' || dataOrigin.str[0] == 'L') { // main MOs and moving windows from QC tasks, and outputs of late tasks
       auto moc = DataRefUtils::as<MonitorObjectCollection>(ref);
       moc->postDeserialization();
 
@@ -124,51 +127,15 @@ void LateTaskRunner::run(ProcessingContext& pCtx)
     }
   }
 
-
-
   // run the task
   mTask->process(taskInputs);
 
   // publish objects
   mObjectsManager->setValidity(mValidity);
   std::unique_ptr<MonitorObjectCollection> array(mObjectsManager->getNonOwningArray());
-  auto concreteOutput = framework::DataSpecUtils::asConcreteDataMatcher(mTaskConfig.moSpec);
-  pCtx.outputs().snapshot(Output{ concreteOutput.origin, concreteOutput.description, concreteOutput.subSpec }, *array);
+  pCtx.outputs().snapshot(mTaskConfig.name, *array);
   mObjectsManager->stopPublishing(PublicationPolicy::Once);
 }
 
-/// \brief ID string for all LateTaskRunner devices
-std::string LateTaskRunner::createIdString()
-{
-  return { "qc-late-task" };
-}
-/// \brief Unified DataOrigin for Quality Control tasks
-header::DataOrigin LateTaskRunner::createDataOrigin(const std::string& detectorCode)
-{
-  std::string originStr = "L";
-  if (detectorCode.empty()) {
-    ILOG(Warning, Support) << "empty detector code for a task data origin, trying to survive with: DET" << ENDM;
-    originStr += "DET";
-  } else if (detectorCode.size() > 3) {
-    ILOG(Warning, Support) << "too long detector code for a task data origin: " + detectorCode + ", trying to survive with: " + detectorCode.substr(0, 3) << ENDM;
-    originStr += detectorCode.substr(0, 3);
-  } else {
-    originStr += detectorCode;
-  }
-  o2::header::DataOrigin origin;
-  origin.runtimeInit(originStr.c_str());
-  return origin;
-}
-
-/// \brief Unified DataDescription naming scheme for all tasks
-header::DataDescription LateTaskRunner::createDataDescription(const std::string& lateTaskName)
-{
-  if (lateTaskName.empty()) {
-    BOOST_THROW_EXCEPTION(FatalException() << errinfo_details("Empty lateTaskName for task's data description"));
-  }
-
-  return quality_control::core::createDataDescription(lateTaskName, LateTaskRunner::taskDescriptionHashLength);
-}
-
 
 } // namespace o2::quality_control::core