DataProcessingDevice.h

Go to the documentation of this file.

// 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 O2_FRAMEWORK_DATAPROCESSINGDEVICE_H_
#define O2_FRAMEWORK_DATAPROCESSINGDEVICE_H_
 
#include "Framework/AlgorithmSpec.h"
#include "Framework/ComputingQuotaOffer.h"
#include "Framework/ConfigParamRegistry.h"
#include "Framework/DataAllocator.h"
#include "Framework/DataRelayer.h"
#include "Framework/DeviceSpec.h"
#include "Framework/DataProcessingStats.h"
#include "Framework/ExpirationHandler.h"
#include "Framework/ServiceRegistry.h"
#include "Framework/InputRoute.h"
#include "Framework/ForwardRoute.h"
#include "Framework/TimingInfo.h"
#include "Framework/ProcessingPolicies.h"
#include "Framework/Tracing.h"
#include "Framework/RunningWorkflowInfo.h"
#include "Framework/ObjectCache.h"
 
#include <fairmq/Device.h>
#include <fairmq/Parts.h>
 
#include <memory>
#include <mutex>
#include <uv.h>
 
namespace o2::framework
{
 
struct InputChannelInfo;
struct DeviceState;
struct ComputingQuotaEvaluator;
 
class DataProcessingDevice;
struct DataProcessorContext;
struct DeviceContext;
 
struct TaskStreamRef {
  int index = -1;
};
 
struct TaskStreamInfo {
  TaskStreamRef id;
  ServiceRegistry* registry;
  uv_work_t* task;
  bool running = false;
};
 
struct DeviceConfigurationHelpers {
  static std::unique_ptr<ConfigParamStore> getConfiguration(ServiceRegistryRef registry, const char* name, std::vector<ConfigParamSpec> const& options);
};
 
class DataProcessingDevice : public fair::mq::Device
{
 public:
  DataProcessingDevice(RunningDeviceRef ref, ServiceRegistry&, ProcessingPolicies& policies);
  void Init() final;
  void InitTask() final;
  void PreRun() final;
  void PostRun() final;
  void Reset() final;
  void ResetTask() final;
  void Run() final;
 
  // Processing functions are now renetrant
  static void doRun(ServiceRegistryRef);
  static void doPrepare(ServiceRegistryRef);
  static void handleData(ServiceRegistryRef, InputChannelInfo&);
  static bool tryDispatchComputation(ServiceRegistryRef ref, std::vector<DataRelayer::RecordAction>& completed);
 
 protected:
  void error(const char* msg);
  void fillContext(DataProcessorContext& context, DeviceContext& deviceContext);
 
 private:
  void initPollers();
  void startPollers();
  void stopPollers();
  RunningDeviceRef mRunningDevice;
 
  std::unique_ptr<ConfigParamRegistry> mConfigRegistry;
  ServiceRegistry& mServiceRegistry;
 
  uint64_t mLastSlowMetricSentTimestamp = 0;             
  uint64_t mLastMetricFlushedTimestamp = 0;              
  uint64_t mBeginIterationTimestamp = 0;                 
  std::vector<fair::mq::RegionInfo> mPendingRegionInfos; 
  std::mutex mRegionInfoMutex;
  ProcessingPolicies mProcessingPolicies; 
  std::vector<uv_work_t> mHandles;        
  std::vector<TaskStreamInfo> mStreams;   
  uv_async_t* mAwakeHandle = nullptr;
};
 
} // namespace o2::framework
#endif // O2_FRAMEWORK_DATAPROCESSINGDEVICE_H_