InputRecord.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_INPUTRECORD_H_
#define O2_FRAMEWORK_INPUTRECORD_H_
 
#include "Framework/DataRef.h"
#include "Framework/DataRefUtils.h"
#include "Framework/InputRoute.h"
#include "Framework/TypeTraits.h"
#include "Framework/TableConsumer.h"
#include "Framework/Traits.h"
#include "Framework/RuntimeError.h"
#include "Framework/Logger.h"
#include "Framework/ObjectCache.h"
#include "Framework/CallbackService.h"
 
#include "Headers/DataHeader.h"
 
#include <gsl/gsl>
 
#include <iterator>
#include <string>
#include <vector>
#include <cstring>
#include <cassert>
#include <memory>
#include <type_traits>
#include <concepts>
 
#include <fairmq/FwdDecls.h>
 
namespace o2::framework
{
 
// Wrapper class to get CCDB metadata
struct CCDBMetadataExtractor {
};
 
struct InputSpec;
class InputSpan;
class CallbackService;
 
class InputRecord
{
 public:
  using DataHeader = o2::header::DataHeader;
 
  // Typesafe position inside a record of an input.
  // Multiple routes by which the input gets in this
  // position are multiplexed.
  struct InputPos {
    size_t index;
    constexpr static size_t INVALID = -1LL;
  };
 
  InputRecord(std::vector<InputRoute> const& inputs,
              InputSpan& span,
              ServiceRegistryRef);
 
  template <typename T>
  class Deleter : public std::default_delete<T>
  {
   public:
    enum struct OwnershipProperty : short {
      Unknown = -1,
      NotOwning = 0, 
      Owning = 1     
    };
 
    using base = std::default_delete<T>;
    using self_type = Deleter<T>;
    // using pointer = typename base::pointer;
 
    constexpr Deleter() = default;
    constexpr Deleter(bool isOwning)
      : base::default_delete(), mProperty(isOwning ? OwnershipProperty::Owning : OwnershipProperty::NotOwning)
    {
    }
 
    // copy constructor is needed in the setup of unique_ptr
    // check that assignments happen only to uninitialized instances
    constexpr Deleter(const self_type& other) : base::default_delete(other), mProperty{OwnershipProperty::Unknown}
    {
      if (mProperty == OwnershipProperty::Unknown) {
        mProperty = other.mProperty;
      } else if (mProperty != other.mProperty) {
        throw runtime_error("Attemp to change resource control");
      }
    }
 
    // copy constructor for the default delete which simply sets the
    // resource ownership control to 'Owning'
    constexpr Deleter(const base& other) : base::default_delete(other), mProperty{OwnershipProperty::Owning} {}
 
    // allow assignment operator only for pristine or matching resource control property
    self_type& operator=(const self_type& other)
    {
      // the default_deleter does not have any state, so this could be skipped, but keep the call to
      // the base for completeness, and the (small) chance for changing the base
      base::operator=(other);
      if (mProperty == OwnershipProperty::Unknown) {
        mProperty = other.mProperty;
      } else if (mProperty != other.mProperty) {
        throw runtime_error("Attemp to change resource control");
      }
      return *this;
    }
 
    void operator()(T* ptr) const
    {
      if (mProperty == OwnershipProperty::NotOwning) {
        // nothing done if resource is not owned
        return;
      }
      base::operator()(ptr);
    }
 
   private:
    OwnershipProperty mProperty = OwnershipProperty::Unknown;
  };
 
  int getPos(const char* name) const;
  [[nodiscard]] static InputPos getPos(std::vector<InputRoute> const& routes, ConcreteDataMatcher matcher);
  [[nodiscard]] static DataRef getByPos(std::vector<InputRoute> const& routes, InputSpan const& span, int pos, int part = 0);
 
  [[nodiscard]] int getPos(const std::string& name) const;
 
  [[nodiscard]] DataRef getByPos(int pos, int part = 0) const;
 
  [[nodiscard]] DataRef getFirstValid(bool throwOnFailure = false) const;
 
  [[nodiscard]] size_t getNofParts(int pos) const;
 
  // Given a binding by string, return the associated DataRef
  DataRef getDataRefByString(const char* bindingName, int part = 0) const
  {
    int pos = getPos(bindingName);
    if (pos < 0) {
      auto msg = describeAvailableInputs();
      throw runtime_error_f("InputRecord::get: no input with binding %s found. %s", bindingName, msg.c_str());
    }
    return this->getByPos(pos, part);
  }
 
  template <typename R>
    requires std::is_convertible_v<R, char const*>
  DataRef getRef(R binding, int part = 0) const
  {
    return getDataRefByString(binding, part);
  }
 
  template <typename R>
    requires requires(R r) { r.c_str(); }
  DataRef getRef(R binding, int part = 0) const
  {
    return getDataRefByString(binding.c_str(), part);
  }
 
  template <typename R>
    requires std::is_convertible_v<R, DataRef>
  DataRef getRef(R ref, int part = 0) const
  {
    return ref;
  }
 
  template <typename T = DataRef, typename R>
  decltype(auto) get(R binding, int part = 0) const
  {
    DataRef ref = getRef(binding, part);
 
    using PointerLessValueT = std::remove_pointer_t<T>;
 
    if constexpr (std::is_same_v<std::decay_t<T>, DataRef>) {
      return ref;
    } else if constexpr (std::is_same<T, std::string>::value) {
      // substitution for std::string
      // If we ask for a string, we need to duplicate it because we do not want
      // the buffer to be deleted when it goes out of scope. The string is built
      // from the data and its lengh, null-termination is not necessary.
      // return std::string object
      return std::string(ref.payload, DataRefUtils::getPayloadSize(ref));
 
      // implementation (c)
    } else if constexpr (std::is_same<T, char const*>::value) {
      // substitution for const char*
      // If we ask for a char const *, we simply point to the payload. Notice this
      // is meant for C-style strings which are expected to be null terminated.
      // If you want to actually get hold of the buffer, use gsl::span<char> as that will
      // give you the size as well.
      // return pointer to payload content
      return reinterpret_cast<char const*>(ref.payload);
 
      // implementation (d)
    } else if constexpr (std::is_same<T, TableConsumer>::value) {
      // substitution for TableConsumer
      // For the moment this is dummy, as it requires proper support to
      // create the RDataSource from the arrow buffer.
      auto data = reinterpret_cast<uint8_t const*>(ref.payload);
      return std::make_unique<TableConsumer>(data, DataRefUtils::getPayloadSize(ref));
 
      // implementation (f)
    } else if constexpr (is_span<T>::value) {
      // substitution for span of messageable objects
      // FIXME: there will be std::span in C++20
      static_assert(is_messageable<typename T::value_type>::value, "span can only be created for messageable types");
      auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
      assert(header);
      if (sizeof(typename T::value_type) > 1 && header->payloadSerializationMethod != o2::header::gSerializationMethodNone) {
        throw runtime_error("Inconsistent serialization method for extracting span");
      }
      using ValueT = typename T::value_type;
      auto payloadSize = DataRefUtils::getPayloadSize(ref);
      if (payloadSize % sizeof(ValueT)) {
        throw runtime_error(("Inconsistent type and payload size at " + std::string(ref.spec->binding) + "(" + DataSpecUtils::describe(*ref.spec) + ")" +
                             ": type size " + std::to_string(sizeof(ValueT)) +
                             "  payload size " + std::to_string(payloadSize))
                              .c_str());
      }
      return gsl::span<ValueT const>(reinterpret_cast<ValueT const*>(ref.payload), payloadSize / sizeof(ValueT));
 
      // implementation (g)
    } else if constexpr (is_container<T>::value) {
      // currently implemented only for vectors
      if constexpr (is_specialization_v<std::remove_const_t<T>, std::vector>) {
        auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
        auto payloadSize = DataRefUtils::getPayloadSize(ref);
        auto method = header->payloadSerializationMethod;
        if (method == o2::header::gSerializationMethodNone) {
          // TODO: construct a vector spectator
          // this is a quick solution now which makes a copy of the plain vector data
          auto* start = reinterpret_cast<typename T::value_type const*>(ref.payload);
          auto* end = start + payloadSize / sizeof(typename T::value_type);
          T result(start, end);
          return result;
        } else if (method == o2::header::gSerializationMethodROOT) {
          using NonConstT = typename std::remove_const<T>::type;
          if constexpr (is_specialization_v<T, ROOTSerialized> == true || has_root_dictionary<T>::value == true) {
            // we expect the unique_ptr to hold an object, exception should have been thrown
            // otherwise
            auto object = DataRefUtils::as<NonConstT>(ref);
            // need to swap the content of the deserialized container to a local variable to force return
            // value optimization
            T container;
            std::swap(const_cast<NonConstT&>(container), *object);
            return container;
          } else {
            throw runtime_error("No supported conversion function for ROOT serialized message");
          }
        } else {
          throw runtime_error("Attempt to extract object from message with unsupported serialization type");
        }
      } else {
        static_assert(always_static_assert_v<T>, "unsupported code path");
      }
 
      // implementation (h)
    } else if constexpr (is_messageable<T>::value) {
      // extract a messageable type by reference
      // Cast content of payload bound by @a binding to known type.
      // we need to check the serialization type, the cast makes only sense for
      // unserialized objects
 
      auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
      auto method = header->payloadSerializationMethod;
      if (method != o2::header::gSerializationMethodNone) {
        // FIXME: we could in principle support serialized content here as well if we
        // store all extracted objects internally and provide cleanup
        throw runtime_error("Can not extract a plain object from serialized message");
      }
      return *reinterpret_cast<T const*>(ref.payload);
 
      // implementation (i)
    } else if constexpr (std::is_pointer_v<T> &&
                         (is_messageable<PointerLessValueT>::value ||
                          has_root_dictionary<PointerLessValueT>::value ||
                          (is_specialization_v<PointerLessValueT, std::vector> && has_messageable_value_type<PointerLessValueT>::value) ||
                          (has_root_dictionary_mapped_type<PointerLessValueT>::value))) {
      // extract a messageable type or object with ROOT dictionary by pointer
      // return unique_ptr to message content with custom deleter
      using ValueT = PointerLessValueT;
 
      auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
      auto payloadSize = DataRefUtils::getPayloadSize(ref);
      auto method = header->payloadSerializationMethod;
      if (method == o2::header::gSerializationMethodNone) {
        if constexpr (is_messageable<ValueT>::value) {
          auto const* ptr = reinterpret_cast<ValueT const*>(ref.payload);
          // return type with non-owning Deleter instance
          std::unique_ptr<ValueT const, Deleter<ValueT const>> result(ptr, Deleter<ValueT const>(false));
          return result;
        } else if constexpr (is_specialization_v<ValueT, std::vector> && has_messageable_value_type<ValueT>::value) {
          // TODO: construct a vector spectator
          // this is a quick solution now which makes a copy of the plain vector data
          auto* start = reinterpret_cast<typename ValueT::value_type const*>(ref.payload);
          auto* end = start + payloadSize / sizeof(typename ValueT::value_type);
          auto container = std::make_unique<ValueT>(start, end);
          std::unique_ptr<ValueT const, Deleter<ValueT const>> result(container.release(), Deleter<ValueT const>(true));
          return result;
        }
        throw runtime_error("unsupported code path");
      } else if (method == o2::header::gSerializationMethodROOT) {
        // This supports the common case of retrieving a root object and getting pointer.
        // Notice that this will return a copy of the actual contents of the buffer, because
        // the buffer is actually serialised, for this reason we return a unique_ptr<T>.
        // FIXME: does it make more sense to keep ownership of all the deserialised
        // objects in a single place so that we can avoid duplicate deserializations?
        // explicitely specify serialization method to ROOT-serialized because type T
        // is messageable and a different method would be deduced in DataRefUtils
        // return type with owning Deleter instance, forwarding to default_deleter
        std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<ROOTSerialized<ValueT>>(ref).release());
        return result;
      } else if (method == o2::header::gSerializationMethodCCDB) {
        // This is to support deserialising objects from CCDB. Contrary to what happens for
        // other objects, those objects are most likely long lived, so we
        // keep around an instance of the associated object and deserialise it only when
        // it's updated.
        // FIXME: add ability to apply callbacks to deserialised objects.
        auto id = ObjectCache::Id::fromRef(ref);
        ConcreteDataMatcher matcher{header->dataOrigin, header->dataDescription, header->subSpecification};
        // If the matcher does not have an entry in the cache, deserialise it
        // and cache the deserialised object at the given id.
        auto path = fmt::format("{}", DataSpecUtils::describe(matcher));
        LOGP(debug, "{}", path);
        auto& cache = mRegistry.get<ObjectCache>();
        auto& callbacks = mRegistry.get<CallbackService>();
        auto cacheEntry = cache.matcherToId.find(path);
        if (cacheEntry == cache.matcherToId.end()) {
          cache.matcherToId.insert(std::make_pair(path, id));
          std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release(), false);
          void* obj = (void*)result.get();
          callbacks.call<CallbackService::Id::CCDBDeserialised>((ConcreteDataMatcher&)matcher, (void*)obj);
          cache.idToObject[id] = obj;
          LOGP(info, "Caching in {} ptr to {} ({})", id.value, path, obj);
          return result;
        }
        auto& oldId = cacheEntry->second;
        // The id in the cache is the same, let's simply return it.
        if (oldId.value == id.value) {
          std::unique_ptr<ValueT const, Deleter<ValueT const>> result((ValueT const*)cache.idToObject[id], false);
          LOGP(debug, "Returning cached entry {} for {} ({})", id.value, path, (void*)result.get());
          return result;
        }
        // The id in the cache is different. Let's destroy the old cached entry
        // and create a new one.
        delete reinterpret_cast<ValueT*>(cache.idToObject[oldId]);
        std::unique_ptr<ValueT const, Deleter<ValueT const>> result(DataRefUtils::as<CCDBSerialized<ValueT>>(ref).release(), false);
        void* obj = (void*)result.get();
        callbacks.call<CallbackService::Id::CCDBDeserialised>((ConcreteDataMatcher&)matcher, (void*)obj);
        cache.idToObject[id] = obj;
        LOGP(info, "Replacing cached entry {} with {} for {} ({})", oldId.value, id.value, path, obj);
        oldId.value = id.value;
        return result;
      } else {
        throw runtime_error("Attempt to extract object from message with unsupported serialization type");
      }
    } else if constexpr (std::is_pointer_v<T>) {
      static_assert(always_static_assert<T>::value, "T is not a supported type");
    } else if constexpr (has_root_dictionary<T>::value) {
      // retrieving ROOT objects follows the pointer approach, i.e. T* has to be specified
      // as template parameter and a unique_ptr will be returned, std vectors of ROOT serializable
      // objects can be retrieved by move, this is handled above in the "container" code branch
      static_assert(always_static_assert_v<T>, "ROOT objects need to be retrieved by pointer");
    } else {
      // non-messageable objects for which serialization method can not be derived by type,
      // the operation depends on the transmitted serialization method
      auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
      auto method = header->payloadSerializationMethod;
      if (method == o2::header::gSerializationMethodNone) {
        // this code path is only selected if the type is non-messageable
        throw runtime_error(
          "Type mismatch: attempt to extract a non-messagable object "
          "from message with unserialized data");
      } else if (method == o2::header::gSerializationMethodROOT) {
        // explicitely specify serialization method to ROOT-serialized because type T
        // is messageable and a different method would be deduced in DataRefUtils
        // return type with owning Deleter instance, forwarding to default_deleter
        std::unique_ptr<T const, Deleter<T const>> result(DataRefUtils::as<ROOTSerialized<T>>(ref).release());
        return result;
      } else {
        throw runtime_error("Attempt to extract object from message with unsupported serialization type");
      }
    }
  }
 
  template <typename T = DataRef, typename R>
  std::map<std::string, std::string>& get(R binding, int part = 0) const
    requires std::same_as<T, CCDBMetadataExtractor>
  {
    auto ref = getRef(binding, part);
    auto header = DataRefUtils::getHeader<header::DataHeader*>(ref);
    auto payloadSize = DataRefUtils::getPayloadSize(ref);
    auto method = header->payloadSerializationMethod;
    if (method != header::gSerializationMethodCCDB) {
      throw runtime_error("Attempt to extract metadata from a non-CCDB serialised message");
    }
    // This is to support deserialising objects from CCDB. Contrary to what happens for
    // other objects, those objects are most likely long lived, so we
    // keep around an instance of the associated object and deserialise it only when
    // it's updated.
    auto id = ObjectCache::Id::fromRef(ref);
    ConcreteDataMatcher matcher{header->dataOrigin, header->dataDescription, header->subSpecification};
    // If the matcher does not have an entry in the cache, deserialise it
    // and cache the deserialised object at the given id.
    auto path = fmt::format("{}", DataSpecUtils::describe(matcher));
    LOGP(debug, "{}", path);
    auto& cache = mRegistry.get<ObjectCache>();
    auto cacheEntry = cache.matcherToMetadataId.find(path);
    if (cacheEntry == cache.matcherToMetadataId.end()) {
      cache.matcherToMetadataId.insert(std::make_pair(path, id));
      cache.idToMetadata[id] = DataRefUtils::extractCCDBHeaders(ref);
      LOGP(info, "Caching CCDB metadata {}: {}", id.value, path);
      return cache.idToMetadata[id];
    }
    auto& oldId = cacheEntry->second;
    // The id in the cache is the same, let's simply return it.
    if (oldId.value == id.value) {
      LOGP(debug, "Returning cached CCDB metatada {}: {}", id.value, path);
      return cache.idToMetadata[id];
    }
    // The id in the cache is different. Let's destroy the old cached entry
    // and create a new one.
    LOGP(info, "Replacing cached entry {} with {} for {}", oldId.value, id.value, path);
    cache.idToMetadata[id] = DataRefUtils::extractCCDBHeaders(ref);
    oldId.value = id.value;
    return cache.idToMetadata[id];
  }
 
  [[nodiscard]] bool isValid(std::string const& s) const
  {
    return isValid(s.c_str());
  }
 
  bool isValid(char const* s) const;
  [[nodiscard]] bool isValid(int pos) const;
 
  [[nodiscard]] size_t size() const;
 
  [[nodiscard]] size_t countValidInputs() const;
 
  template <typename ParentT, typename T>
  class Iterator
  {
   public:
    using ParentType = ParentT;
    using SelfType = Iterator;
    using iterator_category = std::forward_iterator_tag;
    using value_type = T;
    using reference = T&;
    using pointer = T*;
    using difference_type = std::ptrdiff_t;
    using ElementType = typename std::remove_const<value_type>::type;
 
    Iterator() = delete;
 
    Iterator(ParentType const* parent, size_t position = 0, size_t size = 0)
      : mPosition(position), mSize(size > position ? size : position), mParent(parent), mElement{nullptr, nullptr, nullptr}
    {
      if (mPosition < mSize) {
        if (mParent->isValid(mPosition)) {
          mElement = mParent->getByPos(mPosition);
        } else {
          ++(*this);
        }
      }
    }
 
    ~Iterator() = default;
 
    // prefix increment
    SelfType& operator++()
    {
      while (mPosition < mSize && ++mPosition < mSize) {
        if (!mParent->isValid(mPosition)) {
          continue;
        }
        mElement = mParent->getByPos(mPosition);
        break;
      }
      if (mPosition >= mSize) {
        // reset the element to the default value of the type
        mElement = ElementType{};
      }
      return *this;
    }
    // postfix increment
    SelfType operator++(int /*unused*/)
    {
      SelfType copy(*this);
      operator++();
      return copy;
    }
    // return reference
    reference operator*() const
    {
      return mElement;
    }
    // comparison
    bool operator==(const SelfType& rh) const
    {
      return mPosition == rh.mPosition;
    }
    // comparison
    bool operator!=(const SelfType& rh) const
    {
      return mPosition != rh.mPosition;
    }
 
    [[nodiscard]] bool matches(o2::header::DataHeader matcher) const
    {
      if (mPosition >= mSize || mElement.header == nullptr) {
        return false;
      }
      // at this point there must be a DataHeader, this has been checked by the DPL
      // input cache
      const auto* dh = DataRefUtils::getHeader<o2::header::DataHeader*>(mElement);
      return *dh == matcher;
    }
 
    [[nodiscard]] bool matches(o2::header::DataOrigin origin, o2::header::DataDescription description = o2::header::gDataDescriptionInvalid) const
    {
      if (mPosition >= mSize || mElement.header == nullptr) {
        return false;
      }
      // at this point there must be a DataHeader, this has been checked by the DPL
      // input cache
      const auto* dh = DataRefUtils::getHeader<o2::header::DataHeader*>(mElement);
      return dh->dataOrigin == origin && (description == o2::header::gDataDescriptionInvalid || dh->dataDescription == description);
    }
 
    [[nodiscard]] bool matches(o2::header::DataOrigin origin, o2::header::DataDescription description, o2::header::DataHeader::SubSpecificationType subspec) const
    {
      return matches(o2::header::DataHeader{description, origin, subspec});
    }
 
    [[nodiscard]] ParentType const* parent() const
    {
      return mParent;
    }
 
    [[nodiscard]] size_t position() const
    {
      return mPosition;
    }
 
   private:
    size_t mPosition;
    size_t mSize;
    ParentType const* mParent;
    ElementType mElement;
  };
 
  template <typename T>
  class InputRecordIterator : public Iterator<InputRecord, T>
  {
   public:
    using SelfType = InputRecordIterator;
    using BaseType = Iterator<InputRecord, T>;
    using value_type = typename BaseType::value_type;
    using reference = typename BaseType::reference;
    using pointer = typename BaseType::pointer;
    using ElementType = typename std::remove_const<value_type>::type;
    using iterator = Iterator<SelfType, T>;
    using const_iterator = Iterator<SelfType, const T>;
 
    InputRecordIterator(InputRecord const* parent, size_t position = 0, size_t size = 0)
      : BaseType(parent, position, size)
    {
    }
 
    [[nodiscard]] ElementType getByPos(size_t pos) const
    {
      return this->parent()->getByPos(this->position(), pos);
    }
 
    [[nodiscard]] bool isValid(size_t = 0) const
    {
      if (this->position() < this->parent()->size()) {
        return this->parent()->isValid(this->position());
      }
      return false;
    }
 
    [[nodiscard]] size_t size() const
    {
      return this->parent()->getNofParts(this->position());
    }
 
    [[nodiscard]] const_iterator begin() const
    {
      return const_iterator(this, 0, size());
    }
 
    [[nodiscard]] const_iterator end() const
    {
      return const_iterator(this, size());
    }
  };
 
  using iterator = InputRecordIterator<DataRef>;
  using const_iterator = InputRecordIterator<const DataRef>;
 
  [[nodiscard]] const_iterator begin() const
  {
    return {this, 0, size()};
  }
 
  [[nodiscard]] const_iterator end() const
  {
    return {this, size()};
  }
 
  InputSpan& span()
  {
    return mSpan;
  }
 
 private:
  // Produce a string describing the available inputs.
  [[nodiscard]] std::string describeAvailableInputs() const;
 
  ServiceRegistryRef mRegistry;
  std::vector<InputRoute> const& mInputsSchema;
  InputSpan& mSpan;
};
 
} // namespace o2::framework
 
#endif // O2_FRAMEWORK_INPUTREGISTRY_H_