DataProcessingHelpers.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_DATAPROCESSINGHELPERS_H_
#define O2_FRAMEWORK_DATAPROCESSINGHELPERS_H_
 
#include <cstddef>
#include "Framework/TimesliceSlot.h"
#include "Framework/TimesliceIndex.h"
#include <fairmq/FwdDecls.h>
#include <vector>
#include <span>
 
namespace o2::framework
{
struct ServiceRegistryRef;
struct ForwardChannelInfo;
struct ForwardChannelState;
struct OutputChannelInfo;
struct OutputChannelSpec;
struct OutputChannelState;
struct ProcessingPolicies;
struct DeviceSpec;
struct FairMQDeviceProxy;
struct MessageSet;
struct ChannelIndex;
enum struct StreamingState;
enum struct TransitionHandlingState;
 
struct DataProcessingHelpers {
  static void sendEndOfStream(ServiceRegistryRef const& ref, OutputChannelSpec const& channel);
  static bool sendOldestPossibleTimeframe(ServiceRegistryRef const& ref, ForwardChannelInfo const& info, ForwardChannelState& state, size_t timeslice);
  static bool sendOldestPossibleTimeframe(ServiceRegistryRef const& ref, OutputChannelInfo const& info, OutputChannelState& state, size_t timeslice);
  static void broadcastOldestPossibleTimeslice(ServiceRegistryRef const& ref, size_t timeslice);
  static void switchState(ServiceRegistryRef const& ref, StreamingState newState);
  static bool hasOnlyGenerated(DeviceSpec const& spec);
  static TransitionHandlingState updateStateTransition(ServiceRegistryRef const& ref, ProcessingPolicies const& policies);
  static std::vector<fair::mq::Parts> routeForwardedMessageSet(FairMQDeviceProxy& proxy, std::vector<MessageSet>& currentSetOfInputs,
                                                               bool copy, bool consume);
  static void routeForwardedMessages(FairMQDeviceProxy& proxy, std::span<fair::mq::MessagePtr>& currentSetOfInputs, std::vector<fair::mq::Parts>& forwardedParts,
                                     bool copy, bool consume);
 
  static void cleanForwardedMessages(std::span<fair::mq::MessagePtr>& currentSetOfInputs, bool consume);
};
} // namespace o2::framework
#endif // O2_FRAMEWORK_DATAPROCESSINGHELPERS_H_