DataRelayer.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_DATARELAYER_H_
#define O2_FRAMEWORK_DATARELAYER_H_
 
#include "Framework/RootSerializationSupport.h"
#include "Framework/InputRoute.h"
#include "Framework/DataDescriptorMatcher.h"
#include "Framework/ForwardRoute.h"
#include "Framework/CompletionPolicy.h"
#include <fairmq/Message.h>
#include "Framework/TimesliceIndex.h"
#include "Framework/Tracing.h"
#include "Framework/TimesliceSlot.h"
#include "Framework/ServiceRegistryRef.h"
 
#include <cstddef>
#include <mutex>
#include <span>
#include <vector>
#include <functional>
 
#include <fairmq/FwdDecls.h>
 
namespace o2::monitoring
{
class Monitoring;
}
 
namespace o2::framework
{
 
enum struct CacheEntryStatus : int {
  EMPTY,
  PENDING,
  RUNNING,
  DONE
};
 
class DataRelayer
{
 public:
  constexpr static ServiceKind service_kind = ServiceKind::Global;
  struct RelayChoice {
    enum struct Type {
      WillRelay,     
      Invalid,       
      Backpressured, 
      Dropped        
    };
    Type type;
    // The timeslice affected by the given operation.
    TimesliceId timeslice;
  };
 
  struct ActivityStats {
    int newSlots = 0;
    int expiredSlots = 0;
  };
 
  struct PruneOp {
    TimesliceSlot slot = {-1ULL};
  };
 
  struct RecordAction {
    TimesliceSlot slot;
    TimesliceId timeslice;
    CompletionPolicy::CompletionOp op;
  };
 
  enum struct InputType : int {
    Invalid = 0,
    Data = 1,
    SourceInfo = 2,
    DomainInfo = 3
  };
 
  struct InputInfo {
    InputInfo(size_t p, size_t s, InputType t, ChannelIndex i)
      : position(p), size(s), type(t), index(i)
    {
    }
    size_t position;
    size_t size;
    InputType type;
    ChannelIndex index;
  };
 
  DataRelayer(CompletionPolicy const&,
              std::vector<InputRoute> const& routes,
              TimesliceIndex&,
              ServiceRegistryRef,
              int);
 
  ActivityStats processDanglingInputs(std::vector<ExpirationHandler> const&,
                                      ServiceRegistryRef context, bool createNew);
 
  using OnDropCallback = std::function<void(TimesliceSlot, std::vector<std::span<fair::mq::MessagePtr>>&, TimesliceIndex::OldestOutputInfo info)>;
 
  // Callback for when some messages are about to be owned by the the DataRelayer
  using OnInsertionCallback = std::function<void(ServiceRegistryRef&, std::span<fair::mq::MessagePtr>&)>;
 
  void prunePending(OnDropCallback);
  void pruneCache(TimesliceSlot slot, OnDropCallback onDrop = nullptr);
 
  RelayChoice relay(void const* rawHeader,
                    std::unique_ptr<fair::mq::Message>* messages,
                    InputInfo const& info,
                    size_t nMessages,
                    size_t nPayloads = 1,
                    OnInsertionCallback onInsertion = nullptr,
                    OnDropCallback onDrop = nullptr);
 
  void setOldestPossibleInput(TimesliceId timeslice, ChannelIndex channel);
 
  [[nodiscard]] TimesliceIndex::OldestOutputInfo getOldestPossibleOutput() const;
 
  void getReadyToProcess(std::vector<RecordAction>& completed);
 
  std::vector<std::vector<fair::mq::MessagePtr>> consumeAllInputsForTimeslice(TimesliceSlot id);
  std::vector<std::vector<fair::mq::MessagePtr>> consumeExistingInputsForTimeslice(TimesliceSlot id);
 
  [[nodiscard]] size_t getParallelTimeslices() const;
 
  void setPipelineLength(size_t s);
 
  void sendContextState();
  void publishMetrics();
 
  TimesliceId getTimesliceForSlot(TimesliceSlot slot);
 
  void updateCacheStatus(TimesliceSlot slot, CacheEntryStatus oldStatus, CacheEntryStatus newStatus);
  uint32_t getFirstTFOrbitForSlot(TimesliceSlot slot);
  uint32_t getFirstTFCounterForSlot(TimesliceSlot slot);
  uint32_t getRunNumberForSlot(TimesliceSlot slot);
  uint64_t getCreationTimeForSlot(TimesliceSlot slot);
  void clear();
 
  void rescan() { mTimesliceIndex.rescan(); };
 
  [[nodiscard]] size_t getCacheSize() const { return mCache.size(); }
  [[nodiscard]] size_t getNumberOfTimeslices() const { return mTimesliceIndex.size(); }
  [[nodiscard]] size_t getNumberOfUniqueInputs() const { return mDistinctRoutesIndex.size(); }
 
 private:
  ServiceRegistryRef mContext;
 
  std::vector<std::vector<fair::mq::MessagePtr>> mCache;
 
  TimesliceIndex& mTimesliceIndex;
 
  CompletionPolicy mCompletionPolicy;
  std::vector<size_t> mDistinctRoutesIndex;
  std::vector<InputSpec> mInputs;
  std::vector<data_matcher::DataDescriptorMatcher> mInputMatchers;
  std::vector<data_matcher::VariableContext> mVariableContextes;
  std::vector<CacheEntryStatus> mCachedStateMetrics;
  std::vector<PruneOp> mPruneOps;
  size_t mMaxLanes;
 
  O2_LOCKABLE_NAMED(std::recursive_mutex, mMutex, "data relayer mutex");
};
 
} // namespace o2::framework
 
#endif // O2_FRAMEWORK_DATARELAYER_H_