FairMQDeviceProxy.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_FAIRMQDEVICEPROXY_H_
#define O2_FRAMEWORK_FAIRMQDEVICEPROXY_H_
 
#include <memory>
 
#include "Framework/ChannelInfo.h"
#include "Framework/RoutingIndices.h"
#include "Framework/RouteState.h"
#include "Framework/OutputRoute.h"
#include "Framework/InputRoute.h"
#include "Framework/ForwardRoute.h"
#include <fairmq/FwdDecls.h>
#include <fairmq/Version.h>
#if (FAIRMQ_VERSION_DEC >= 111000)
#include <fairmq/shmem/Common.h>
#endif
#include <vector>
 
namespace o2::header
{
struct DataHeader;
};
 
namespace o2::framework
{
#if (FAIRMQ_VERSION_DEC >= 111000)
using PointerReconstructor = std::function<std::byte*(fair::mq::shmem::MetaHeader&&)>;
#endif
class FairMQDeviceProxy
{
 public:
  FairMQDeviceProxy() = default;
  FairMQDeviceProxy(FairMQDeviceProxy const&) = delete;
  void bind(std::vector<OutputRoute> const& outputs, std::vector<InputRoute> const& inputs,
            std::vector<ForwardRoute> const& forwards,
            std::function<fair::mq::Channel&(std::string const&)> bindChannelByName,
            std::function<bool(void)> newStateRequestedCallback);
 
  [[nodiscard]] OutputRoute const& getOutputRoute(RouteIndex routeIndex) const { return mOutputs.at(routeIndex.value); }
  [[nodiscard]] fair::mq::TransportFactory* getOutputTransport(RouteIndex routeIndex) const;
  [[nodiscard]] fair::mq::TransportFactory* getInputTransport(RouteIndex routeIndex) const;
  [[nodiscard]] fair::mq::TransportFactory* getForwardTransport(RouteIndex routeIndex) const;
  [[nodiscard]] ChannelIndex getOutputChannelIndexByName(std::string const& channelName) const;
  [[nodiscard]] ChannelIndex getInputChannelIndexByName(std::string const& channelName) const;
  [[nodiscard]] ChannelIndex getForwardChannelIndexByName(std::string const& channelName) const;
  [[nodiscard]] ChannelIndex getOutputChannelIndex(OutputSpec const& spec, size_t timeslice) const;
#if (FAIRMQ_VERSION_DEC >= 111000)
  [[nodiscard]] PointerReconstructor getShmPointerReconstructor(InputSpec const& spec, size_t timeslice);
#endif
  void getMatchingForwardChannelIndexes(std::vector<ChannelIndex>& result, header::DataHeader const& header, size_t timeslice) const;
  [[nodiscard]] ChannelIndex getOutputChannelIndex(RouteIndex routeIndex) const;
  [[nodiscard]] ChannelIndex getInputChannelIndex(RouteIndex routeIndex) const;
  [[nodiscard]] ChannelIndex getForwardChannelIndex(RouteIndex routeIndex) const;
  [[nodiscard]] fair::mq::Channel* getInputChannel(ChannelIndex channelIndex) const;
  [[nodiscard]] fair::mq::Channel* getOutputChannel(ChannelIndex channelIndex) const;
  [[nodiscard]] fair::mq::Channel* getForwardChannel(ChannelIndex channelIndex) const;
 
  [[nodiscard]] OutputChannelInfo const& getOutputChannelInfo(ChannelIndex channelIndex) const;
  [[nodiscard]] ForwardChannelInfo const& getForwardChannelInfo(ChannelIndex channelIndex) const;
 
  OutputChannelState& getOutputChannelState(ChannelIndex channelIndex);
  ForwardChannelState& getForwardChannelState(ChannelIndex channelIndex);
 
  [[nodiscard]] std::unique_ptr<fair::mq::Message> createOutputMessage(RouteIndex routeIndex) const;
  [[nodiscard]] std::unique_ptr<fair::mq::Message> createOutputMessage(RouteIndex routeIndex, const size_t size) const;
 
  [[nodiscard]] std::unique_ptr<fair::mq::Message> createInputMessage(RouteIndex routeIndex) const;
  [[nodiscard]] std::unique_ptr<fair::mq::Message> createInputMessage(RouteIndex routeIndex, const size_t size) const;
 
  [[nodiscard]] std::unique_ptr<fair::mq::Message> createForwardMessage(RouteIndex routeIndex) const;
 
  [[nodiscard]] size_t getNumOutputChannels() const { return mOutputChannelInfos.size(); }
  [[nodiscard]] size_t getNumInputChannels() const { return mInputChannels.size(); }
  [[nodiscard]] size_t getNumForwardChannels() const { return mForwardChannelInfos.size(); }
  [[nodiscard]] size_t getNumForwards() const { return mForwards.size(); }
 
  [[nodiscard]] bool newStateRequested() const { return mStateChangeCallback(); }
 
 private:
  std::vector<OutputRoute> mOutputs;
  std::vector<RouteState> mOutputRoutes;
  std::vector<OutputChannelInfo> mOutputChannelInfos;
  std::vector<OutputChannelState> mOutputChannelStates;
 
  std::vector<InputRoute> mInputs;
  std::vector<RouteState> mInputRoutes;
  std::vector<fair::mq::Channel*> mInputChannels;
  std::vector<std::string> mInputChannelNames;
 
  std::vector<ForwardRoute> mForwards;
  std::vector<RouteState> mForwardRoutes;
  std::vector<ForwardChannelInfo> mForwardChannelInfos;
  std::vector<ForwardChannelState> mForwardChannelStates;
 
  std::function<bool()> mStateChangeCallback;
};
 
} // namespace o2::framework
 
#endif // O2_FRAMEWORK_FAIRMQDEVICEPROXY_H_