TaskRunner.h

Go to the documentation of this file.

// 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 QC_CORE_TASKRUNNER_H
#define QC_CORE_TASKRUNNER_H

#include <boost/property_tree/ptree.hpp>

// O2
#include <Common/Timer.h>
#include <Framework/Task.h>
#include <Framework/DataProcessorSpec.h>
#include <Framework/CompletionPolicy.h>
#include <Framework/EndOfStreamContext.h>
#include <Headers/DataHeader.h>
#include <Framework/InitContext.h>
// QC
#include "QualityControl/TaskConfig.h"
#include "QualityControl/TaskInterface.h"

namespace o2::configuration
{
class ConfigurationInterface;
}

namespace o2::monitoring
{
class Monitoring;
}

namespace o2::quality_control::core
{

class TaskRunner : public framework::Task
{
 public:
  TaskRunner(const std::string& taskName, const std::string& configurationSource, size_t id = 0);
  ~TaskRunner() override = default;

  void init(framework::InitContext& iCtx) override;
  void run(framework::ProcessingContext& pCtx) override;

  static framework::CompletionPolicy::CompletionOp completionPolicyCallback(o2::framework::CompletionPolicy::InputSet inputs);

  std::string getDeviceName() { return mDeviceName; };
  const framework::Inputs& getInputsSpecs() { return mInputSpecs; };
  const framework::OutputSpec getOutputSpec() { return mMonitorObjectsSpec; };
  const framework::Options getOptions() { return mOptions; };

  void setResetAfterPublish(bool);

  static std::string createTaskRunnerIdString();
  static header::DataOrigin createTaskDataOrigin();
  static header::DataDescription createTaskDataDescription(const std::string& taskName);

  void endOfStream(framework::EndOfStreamContext& eosContext) override;

 private:
  void start(const framework::ConfigParamRegistry& options);
  void stop();
  void reset();

  std::tuple<bool /*data ready*/, bool /*timer ready*/> validateInputs(const framework::InputRecord&);
  void loadTaskConfig();
  void loadTopologyConfig();
  void startOfActivity();
  void endOfActivity();
  void startCycle();
  void finishCycle(framework::DataAllocator& outputs);
  int publish(framework::DataAllocator& outputs);
  void publishCycleStats();
  void saveToFile();

 private:
  std::string mDeviceName;
  TaskConfig mTaskConfig;
  std::shared_ptr<configuration::ConfigurationInterface> mConfigFile; // used in init only
  std::shared_ptr<monitoring::Monitoring> mCollector;
  std::shared_ptr<TaskInterface> mTask;
  bool mResetAfterPublish = false;
  std::shared_ptr<ObjectsManager> mObjectsManager;
  int mRunNumber;

  std::string validateDetectorName(std::string name) const;
  boost::property_tree::ptree getTaskConfigTree() const;
  void updateMonitoringStats(framework::ProcessingContext& pCtx);

  // consider moving these to TaskConfig
  framework::Inputs mInputSpecs;
  framework::OutputSpec mMonitorObjectsSpec;
  framework::Options mOptions;

  bool mCycleOn = false;
  bool mNoMoreCycles = false;
  int mCycleNumber = 0;

  // stats
  int mNumberMessagesReceivedInCycle = 0;
  int mNumberObjectsPublishedInCycle = 0;
  int mTotalNumberObjectsPublished = 0; // over a run
  double mLastPublicationDuration = 0;
  int mDataReceivedInCycle = 0;
  AliceO2::Common::Timer mTimerTotalDurationActivity;
  AliceO2::Common::Timer mTimerDurationCycle;
};

} // namespace o2::quality_control::core

#endif // QC_CORE_TASKRUNNER_H