ServiceRegistry.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_SERVICEREGISTRY_H_
#define O2_FRAMEWORK_SERVICEREGISTRY_H_
 
#include "Framework/ThreadSafetyAnalysis.h"
#include "Framework/ServiceHandle.h"
#include "Framework/ServiceSpec.h"
#include "Framework/CompilerBuiltins.h"
#include "Framework/TypeIdHelpers.h"
 
#include <array>
#include <string>
#include <type_traits>
#include <typeinfo>
#include <atomic>
#include <mutex>
 
namespace o2::framework
{
 
struct DeviceState;
 
struct ServiceMeta {
  ServiceKind kind = ServiceKind::Serial;
  uint64_t threadId = 0;
};
 
struct NoLocking {
  void lock() {}
  void unlock() {}
};
 
struct O2_DPL_CAPABILITY("mutex") MutexLock {
  void lock() O2_DPL_ACQUIRE() { mutex.lock(); }
  void unlock() O2_DPL_RELEASE() { mutex.unlock(); }
  std::mutex& mutex;
};
 
// A pointer to a service. Includes the locking policy
// for that service.
template <typename T, typename LOCKING = NoLocking>
class service_ptr : LOCKING
{
 public:
  service_ptr(T* ptr, LOCKING policy) : LOCKING(policy), mPtr{ptr} { this->lock(); }
  ~service_ptr() { this->unlock(); }
  service_ptr(service_ptr<T, LOCKING> const&) = delete;
  service_ptr& operator=(service_ptr<T, LOCKING> const&) = delete;
  T& operator*() { return *mPtr; }
  T* operator->() { return mPtr; }
 
 private:
  T* mPtr;
};
 
template <typename, typename = void>
struct ServiceKindExtractor {
  constexpr static ServiceKind kind = ServiceKind::Serial;
};
 
template <typename T>
struct ServiceKindExtractor<T, std::void_t<decltype(T::service_kind)>> : std::is_same<decltype(T::service_kind), enum ServiceKind> {
  constexpr static ServiceKind kind = T::service_kind;
};
 
template <typename T>
inline constexpr ServiceKind service_kind_v = ServiceKindExtractor<T>::kind;
 
struct ServiceRegistry {
  struct Salt {
    short streamId = 0;
    short dataProcessorId = 0;
  };
 
  enum struct SpecialStreamId : short {
    Global = 0,
    Callback = -1,
    Invalid = -2
  };
 
  enum struct SpecialDataProcessorId : short {
    Device = -1,
    Invalid = -2
  };
 
  static constexpr Salt GLOBAL_CONTEXT_SALT{0, 0};
 
  struct InstanceId {
    uint32_t id = 0;
  };
 
  struct Index {
    int32_t index = -1;
  };
 
  struct SpecIndex {
    int index = -1;
  };
 
  // Metadata about the service. This
  // might be interesting for debugging purposes.
  // however it's not used to uniquely identify
  // the service.
  struct Meta {
    ServiceKind kind = ServiceKind::Serial;
    char const* name = nullptr;
    // The index in the
    SpecIndex specIndex{};
  };
 
  // Unique identifier for a service.
  // While we use the salted hash to find the bucket
  // in the hashmap, the service can be uniquely identified
  // only by this 64 bit value.
  struct Key {
    ServiceTypeHash typeHash;
    Salt salt;
  };
 
  static constexpr int32_t valueFromSalt(Salt salt) { return ((int32_t)salt.streamId) << 16 | salt.dataProcessorId; }
  static constexpr uint64_t valueFromKey(Key key) { return ((uint64_t)key.typeHash.hash) << 32 | ((uint64_t)valueFromSalt(key.salt)); }
 
  constexpr static int32_t MAX_DISTANCE = 8;
  constexpr static uint32_t MAX_SERVICES = 256;
  constexpr static uint32_t MAX_SERVICES_MASK = MAX_SERVICES - 1;
 
  static Salt globalDeviceSalt()
  {
    return GLOBAL_CONTEXT_SALT;
  }
 
  static Salt globalStreamSalt(short streamId)
  {
    return {streamId, 0};
  }
 
  static Salt dataProcessorSalt(short /* dataProcessorId */)
  {
    // FIXME: old behaviour for now
    // return {0, dataProcessorId};
    return GLOBAL_CONTEXT_SALT;
  }
 
  static Salt streamSalt(short streamId, short dataProcessorId)
  {
    // FIXME: old behaviour for now
    // return {streamId, dataProcessorId};
    return {streamId, dataProcessorId};
  }
 
  constexpr InstanceId instanceFromTypeSalt(ServiceTypeHash type, Salt salt) const
  {
    // Fold the whole salt down into the low bits. The slot is the low bits of
    // this (see indexFromInstance) while streamId sits at bit 16 of
    // valueFromSalt, so using that directly gives every stream the same slot
    // for a given service: they pile into one probe window, and once it is
    // MAX_DISTANCE deep the next registration is refused -- reported against
    // whichever service happened to lose, not the one which filled it.
    //
    // Widening the table does not help on its own: the mask stays below bit 16
    // until MAX_SERVICES passes 65536.
    uint32_t mixed = static_cast<uint32_t>(static_cast<uint16_t>(salt.streamId)) * 0x9E3779B9u ^
                     static_cast<uint32_t>(static_cast<uint16_t>(salt.dataProcessorId));
    return InstanceId{type.hash ^ mixed};
  }
 
  constexpr Index indexFromInstance(InstanceId id) const
  {
    static_assert(MAX_SERVICES_MASK < 0x7FFFFFFF, "MAX_SERVICES_MASK must be smaller than 0x7FFFFFFF");
    return Index{static_cast<int32_t>(id.id & MAX_SERVICES_MASK)};
  }
 
  mutable std::vector<ServicePostRenderGUIHandle> mPostRenderGUIHandles;
 
  void throwError(const char* name, int64_t hash, int64_t streamId, int64_t dataprocessorId) const;
 
 public:
  using hash_type = decltype(TypeIdHelpers::uniqueId<void>());
  ServiceRegistry();
 
  ServiceRegistry(ServiceRegistry const& other);
  ServiceRegistry& operator=(ServiceRegistry const& other);
 
  void preExitCallbacks();
 
  void postRenderGUICallbacks();
 
  void declareService(ServiceSpec const& spec, DeviceState& state, fair::mq::ProgOptions& options, ServiceRegistry::Salt salt = ServiceRegistry::globalDeviceSalt());
 
  void bindService(ServiceRegistry::Salt salt, ServiceSpec const& spec, void* service) const;
 
  void lateBindStreamServices(DeviceState& state, fair::mq::ProgOptions& options, ServiceRegistry::Salt salt);
 
  void registerService(ServiceTypeHash typeHash, void* service, ServiceKind kind, Salt salt, char const* name = nullptr, ServiceRegistry::SpecIndex specIndex = SpecIndex{-1}) const;
 
  // Lookup a given @a typeHash for a given @a threadId at
  // a unique (per typeHash) location. There might
  // be other typeHash which sit in the same place, but
  // the if statement will rule them out. As long as
  // only one thread writes in a given i + id location
  // as guaranteed by the atomic, mServicesKey[i + id] will
  // either be 0 or the final value.
  // This method should NEVER register a new service, event when requested.
  int getPos(ServiceTypeHash typeHash, Salt salt) const;
 
  // Basic, untemplated API. This will require explicitly
  // providing the @a typeHash for the Service type,
  // there @a threadId of the thread asking the service
  // and the @a kind of service. This
  // method might trigger the registering of a new service
  // if the service is not a stream service and the global
  // zero service is available.
  // Use this API only if you know what you are doing.
  void* get(ServiceTypeHash typeHash, Salt salt, ServiceKind kind, char const* name = nullptr) const;
 
  void registerService(ServiceHandle handle, Salt salt = ServiceRegistry::globalDeviceSalt())
  {
    ServiceRegistry::registerService({handle.hash}, handle.instance, handle.kind, salt, handle.name.c_str());
  }
 
  mutable std::vector<ServiceSpec> mSpecs;
  mutable std::array<std::atomic<Key>, MAX_SERVICES + MAX_DISTANCE> mServicesKey;
  mutable std::array<void*, MAX_SERVICES + MAX_DISTANCE> mServicesValue;
  mutable std::array<Meta, MAX_SERVICES + MAX_DISTANCE> mServicesMeta;
  mutable std::array<std::atomic<bool>, MAX_SERVICES + MAX_DISTANCE> mServicesBooked;
  mutable std::recursive_mutex mMutex;
  mutable int64_t mLastLock = -1;
  mutable std::atomic<int> lockCounter = {0};
 
  template <class I, class C, enum ServiceKind K = ServiceKind::Serial>
    requires std::is_base_of_v<I, C>
  void registerService(C* service, Salt salt = ServiceRegistry::globalDeviceSalt())
  {
    // This only works for concrete implementations of the type T.
    // We need type elision as we do not want to know all the services in
    // advance
    constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<I>()};
    ServiceRegistry::registerService(typeHash, reinterpret_cast<void*>(service), K, salt, typeid(C).name());
  }
 
  template <class I, class C, enum ServiceKind K = ServiceKind::Serial>
    requires std::is_base_of_v<I, C>
  void registerService(C const* service, Salt salt = ServiceRegistry::globalDeviceSalt())
  {
    // This only works for concrete implementations of the type T.
    // We need type elision as we do not want to know all the services in
    // advance
    constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<I const>()};
    this->registerService(typeHash, reinterpret_cast<void*>(const_cast<C*>(service)), K, salt, typeid(C).name());
  }
 
  template <typename T>
    requires(std::is_const_v<T> == false)
  bool active(Salt salt) const
  {
    constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<T>()};
    if (this->getPos(typeHash, GLOBAL_CONTEXT_SALT) != -1) {
      return true;
    }
    auto result = this->getPos(typeHash, salt) != -1;
    return result;
  }
 
  template <typename T>
  T& get(Salt salt) const
  {
    constexpr ServiceTypeHash typeHash{TypeIdHelpers::uniqueId<T>()};
    auto ptr = this->get(typeHash, salt, ServiceKindExtractor<T>::kind, typeid(T).name());
    if (O2_BUILTIN_LIKELY(ptr != nullptr)) {
      if constexpr (std::is_const_v<T>) {
        return *reinterpret_cast<T const*>(ptr);
      } else {
        return *reinterpret_cast<T*>(ptr);
      }
    }
    throwError(typeid(T).name(), typeHash.hash, salt.streamId, salt.dataProcessorId);
    O2_BUILTIN_UNREACHABLE();
  }
 
  void postRenderGUICallbacks(ServiceRegistryRef ref);
  void lock(Salt salt) const O2_DPL_ACQUIRE(mMutex);
 
  void unlock(Salt salt) const O2_DPL_RELEASE(mMutex);
};
 
} // namespace o2::framework
 
#endif // O2_FRAMEWORK_SERVICEREGISTRY_H_