HiveWhiteBoard.cpp

Go to the documentation of this file.

//====================================================================
//  WhiteBoard (Concurrent Event Data Store)
//--------------------------------------------------------------------
//
//====================================================================
// Include files
#include "GaudiKernel/ConcurrencyFlags.h"
#include "GaudiKernel/DataObjID.h"
#include "GaudiKernel/DataObject.h"
#include "GaudiKernel/DataSvc.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/Service.h"
#include "GaudiKernel/SmartIF.h"
#include "GaudiKernel/TypeNameString.h"
#include "Rtypes.h"
#include "ThreadLocalStorage.h"
#include "boost/callable_traits.hpp"
#include "tbb/concurrent_queue.h"
#include "tbb/mutex.h"
#include "tbb/recursive_mutex.h"
#include <utility>

// Interfaces
#include "GaudiKernel/IAddressCreator.h"
#include "GaudiKernel/IConversionSvc.h"
#include "GaudiKernel/IDataManagerSvc.h"
#include "GaudiKernel/IDataProviderSvc.h"
#include "GaudiKernel/IDataStoreAgent.h"
#include "GaudiKernel/IHiveWhiteBoard.h"
#include "GaudiKernel/IOpaqueAddress.h"
#include "GaudiKernel/IRegistry.h"
#include "GaudiKernel/ISvcLocator.h"
#include "GaudiKernel/ISvcManager.h"

namespace
{
  struct Partition final {
    SmartIF<IDataProviderSvc> dataProvider;
    SmartIF<IDataManagerSvc>  dataManager;
    int                       eventNumber = -1;

    // allow acces 'by type' -- used in fwd
    template <typename T>
    T* get();
  };
  template <>
  IDataProviderSvc* Partition::get<IDataProviderSvc>()
  {
    return dataProvider.get();
  }
  template <>
  IDataManagerSvc* Partition::get<IDataManagerSvc>()
  {
    return dataManager.get();
  }

  // C++20: replace with http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2017/p0290r2.html
  //         http://www.open-std.org/jtc1/sc22/wg21/docs/papers/2014/n4033.html

  template <typename T, typename Mutex = tbb::recursive_mutex, typename ReadLock = typename Mutex::scoped_lock,
            typename WriteLock = ReadLock>
  class Synced
  {
    T             m_obj;
    mutable Mutex m_mtx;

  public:
    template <typename F>
    decltype( auto ) with_lock( F&& f )
    {
      WriteLock lock{m_mtx};
      return f( m_obj );
    }
    template <typename F>
    decltype( auto ) with_lock( F&& f ) const
    {
      ReadLock lock{m_mtx};
      return f( m_obj );
    }
  };
  // transform an f(T) into an f(Synced<T>)
  template <typename Fun>
  auto with_lock( Fun&& f )
  {
    return [f = std::forward<Fun>( f )]( auto& p )->decltype( auto ) { return p.with_lock( f ); };
  }
  // call f(T) for each element in a container of Synced<T>
  template <typename ContainerOfSynced, typename Fun>
  void for_( ContainerOfSynced& c, Fun&& f )
  {
    std::for_each( begin( c ), end( c ), with_lock( std::forward<Fun>( f ) ) );
  }
}

TTHREAD_TLS( Synced<Partition>* ) s_current = nullptr;

namespace
{
  namespace detail
  {
    // given a callable F(Arg_t,...)
    // argument_t<F> will be equal to Arg_t
    template <typename F>
    using argument_t = std::tuple_element_t<0, boost::callable_traits::args_t<F>>;
  }

  template <typename Fun>
  StatusCode fwd( Fun f )
  {
    if ( !s_current ) return IDataProviderSvc::Status::INVALID_ROOT;
    return s_current->with_lock( [&]( Partition& p ) {
      auto* svc = p.get<std::decay_t<detail::argument_t<Fun>>>();
      return svc ? f( *svc ) : IDataProviderSvc::Status::INVALID_ROOT;
    } );
  }
}

class HiveWhiteBoard : public extends<Service, IDataProviderSvc, IDataManagerSvc, IHiveWhiteBoard>
{
protected:
  Gaudi::Property<CLID>        m_rootCLID{this, "RootCLID", 110 /*CLID_Event*/, "CLID of root entry"};
  Gaudi::Property<std::string> m_rootName{this, "RootName", "/Event", "name of root entry"};
  Gaudi::Property<std::string> m_loader{this, "DataLoader", "EventPersistencySvc", ""};
  Gaudi::Property<size_t>      m_slots{this, "EventSlots", 1, "number of event slots"};
  Gaudi::Property<bool> m_forceLeaves{this, "ForceLeaves", false, "force creation of default leaves on registerObject"};
  Gaudi::Property<bool> m_enableFaultHdlr{this, "EnableFaultHandler", false,
                                          "enable incidents on data creation requests"};

  IConversionSvc* m_dataLoader = nullptr;
  IAddressCreator* m_addrCreator = nullptr;
  std::vector<Synced<Partition>> m_partitions;
  tbb::concurrent_queue<size_t> m_freeSlots;

public:
  using extends::extends;

  ~HiveWhiteBoard() override
  {
    setDataLoader( 0 ).ignore();
    resetPreLoad().ignore();
    clearStore().ignore();
    for_( m_partitions, []( Partition& p ) {
      p.dataManager->release();
      p.dataProvider->release();
    } );
    m_partitions.clear();
  }

  size_t freeSlots() override { return m_freeSlots.unsafe_size(); }

  CLID rootCLID() const override { return (CLID)m_rootCLID; }
  const std::string& rootName() const override { return m_rootName; }

  StatusCode registerAddress( boost::string_ref path, IOpaqueAddress* pAddr ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.registerAddress( path, pAddr ); } );
  }
  StatusCode registerAddress( IRegistry* parent, boost::string_ref path, IOpaqueAddress* pAdd ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.registerAddress( parent, path, pAdd ); } );
  }
  StatusCode unregisterAddress( boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.unregisterAddress( path ); } );
  }
  StatusCode unregisterAddress( IRegistry* pParent, boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.unregisterAddress( pParent, path ); } );
  }
  StatusCode objectLeaves( const DataObject* pObject, std::vector<IRegistry*>& leaves ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.objectLeaves( pObject, leaves ); } );
  }
  StatusCode objectLeaves( const IRegistry* pObject, std::vector<IRegistry*>& leaves ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.objectLeaves( pObject, leaves ); } );
  }
  StatusCode objectParent( const DataObject* pObject, IRegistry*& refpParent ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.objectParent( pObject, refpParent ); } );
  }
  StatusCode objectParent( const IRegistry* pObject, IRegistry*& refpParent ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.objectParent( pObject, refpParent ); } );
  }
  StatusCode clearSubTree( boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.clearSubTree( path ); } );
  }
  StatusCode clearSubTree( DataObject* pObject ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.clearSubTree( pObject ); } );
  }
  StatusCode clearStore() override
  {
    for_( m_partitions, []( Partition& p ) { p.dataManager->clearStore().ignore(); } );
    return StatusCode::SUCCESS;
  }

  StatusCode traverseSubTree( boost::string_ref path, IDataStoreAgent* pAgent ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.traverseSubTree( path, pAgent ); } );
  }
  StatusCode traverseSubTree( DataObject* pObject, IDataStoreAgent* pAgent ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.traverseSubTree( pObject, pAgent ); } );
  }
  StatusCode traverseTree( IDataStoreAgent* pAgent ) override
  {
    return fwd( [&]( IDataManagerSvc& p ) { return p.traverseTree( pAgent ); } );
  }
  StatusCode setRoot( std::string path, DataObject* pObj ) override
  {
    return fwd(
        [ pObj, path = std::move( path ) ]( IDataManagerSvc & p ) { return p.setRoot( std::move( path ), pObj ); } );
  }

  StatusCode setRoot( std::string path, IOpaqueAddress* pAddr ) override
  {
    return fwd(
        [ pAddr, path = std::move( path ) ]( IDataManagerSvc & p ) { return p.setRoot( std::move( path ), pAddr ); } );
  }

  StatusCode setDataLoader( IConversionSvc*   pDataLoader,
                            IDataProviderSvc* dpsvc __attribute__( ( unused ) ) = nullptr ) override
  {
    if ( pDataLoader ) pDataLoader->addRef();
    if ( m_dataLoader ) m_dataLoader->release();
    if ( pDataLoader ) pDataLoader->setDataProvider( this );
    m_dataLoader = pDataLoader;
    for_( m_partitions, [&]( Partition& p ) { p.dataManager->setDataLoader( m_dataLoader, this ).ignore(); } );
    return StatusCode::SUCCESS;
  }
  StatusCode addPreLoadItem( const DataStoreItem& item ) override
  {
    for_( m_partitions, [&]( Partition& p ) { p.dataProvider->addPreLoadItem( item ); } );
    return StatusCode::SUCCESS;
  }
  StatusCode removePreLoadItem( const DataStoreItem& item ) override
  {
    for_( m_partitions, [&]( Partition& p ) { p.dataProvider->removePreLoadItem( item ); } );
    return StatusCode::SUCCESS;
  }
  StatusCode resetPreLoad() override
  {
    for_( m_partitions, [&]( Partition& p ) { p.dataProvider->resetPreLoad(); } );
    return StatusCode::SUCCESS;
  }
  StatusCode preLoad() override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.preLoad(); } );
  }
  StatusCode registerObject( boost::string_ref parent, boost::string_ref obj, DataObject* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.registerObject( parent, obj, pObj ); } );
  }
  StatusCode registerObject( DataObject* parent, boost::string_ref obj, DataObject* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.registerObject( parent, obj, pObj ); } );
  }
  StatusCode unregisterObject( boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.unregisterObject( path ); } );
  }
  StatusCode unregisterObject( DataObject* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.unregisterObject( pObj ); } );
  }
  StatusCode unregisterObject( DataObject* pObj, boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.unregisterObject( pObj, path ); } );
  }
  StatusCode retrieveObject( IRegistry* parent, boost::string_ref path, DataObject*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.retrieveObject( parent, path, pObj ); } );
  }
  StatusCode findObject( boost::string_ref path, DataObject*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.retrieveObject( path, pObj ); } );
  }
  StatusCode findObject( IRegistry* parent, boost::string_ref path, DataObject*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.retrieveObject( parent, path, pObj ); } );
  }
  StatusCode linkObject( IRegistry* from, boost::string_ref objPath, DataObject* to ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.linkObject( from, objPath, to ); } );
  }
  StatusCode linkObject( boost::string_ref fullPath, DataObject* to ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.linkObject( fullPath, to ); } );
  }
  StatusCode unlinkObject( IRegistry* from, boost::string_ref objPath ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.unlinkObject( from, objPath ); } );
  }
  StatusCode unlinkObject( DataObject* from, boost::string_ref objPath ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.unlinkObject( from, objPath ); } );
  }
  StatusCode unlinkObject( boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.unlinkObject( path ); } );
  }
  StatusCode updateObject( IRegistry* pDirectory ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.updateObject( pDirectory ); } );
  }
  StatusCode updateObject( DataObject* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& p ) { return p.updateObject( pObj ); } );
  }

  //
  //---IHiveWhiteBard implemenation--------------------------------------------------
  //

  StatusCode clearStore( size_t partition ) override
  {
    return m_partitions[partition].with_lock( []( Partition& p ) { return p.dataManager->clearStore(); } );
  }

  StatusCode selectStore( size_t partition ) override
  {
    s_current = &m_partitions[partition];
    return StatusCode::SUCCESS;
  }

  StatusCode setNumberOfStores( size_t slots ) override
  {
    if ( FSMState() == Gaudi::StateMachine::INITIALIZED || FSMState() == Gaudi::StateMachine::RUNNING ) {
      warning() << "Too late to change the number of slots!" << endmsg;
      return StatusCode::FAILURE;
    }
    m_slots = slots;
    Gaudi::Concurrency::ConcurrencyFlags::setNumConcEvents( slots );
    return StatusCode::SUCCESS;
  }

  size_t getNumberOfStores() const override { return m_slots; }

  bool exists( const DataObjID& id ) override
  {
    DataObject* pObject{nullptr};
    return findObject( id.fullKey(), pObject ).isSuccess();
  }

  size_t allocateStore( int evtnumber ) override
  {
    // take next free slot in the list
    size_t slot = std::string::npos;
    if ( m_freeSlots.try_pop( slot ) ) {
      assert( slot != std::string::npos );
      assert( slot < m_partitions.size() );
      m_partitions[slot].with_lock( [evtnumber]( Partition& p ) {
        assert( p.eventNumber == -1 ); // or whatever value represents 'free'
        p.eventNumber = evtnumber;
      } );
    }
    return slot;
  }

  StatusCode freeStore( size_t partition ) override
  {
    assert( partition < m_partitions.size() );
    auto prev = m_partitions[partition].with_lock( []( Partition& p ) { return std::exchange( p.eventNumber, -1 ); } );
    if ( UNLIKELY( prev == -1 ) ) return StatusCode::FAILURE; // double free -- should never happen!
    m_freeSlots.push( partition );
    return StatusCode::SUCCESS;
  }

  size_t getPartitionNumber( int eventnumber ) const override
  {
    auto i = std::find_if( begin( m_partitions ), end( m_partitions ),
                           with_lock( [eventnumber]( const Partition& p ) { return p.eventNumber == eventnumber; } ) );
    return i != end( m_partitions ) ? std::distance( begin( m_partitions ), i ) : std::string::npos;
  }

  StatusCode attachServices()
  {
    StatusCode sc = service( m_loader, m_addrCreator, true );
    if ( !sc.isSuccess() ) {
      error() << "Failed to retrieve data loader "
              << "\"" << m_loader << "\"" << endmsg;
      return sc;
    }
    IConversionSvc* dataLoader = nullptr;
    sc                         = service( m_loader, dataLoader, true );
    if ( !sc.isSuccess() ) {
      error() << MSG::ERROR << "Failed to retrieve data loader "
              << "\"" << m_loader << "\"" << endmsg;
      return sc;
    }
    sc = setDataLoader( dataLoader );
    dataLoader->release();
    if ( !sc.isSuccess() ) {
      error() << MSG::ERROR << "Failed to set data loader "
              << "\"" << m_loader << "\"" << endmsg;
      return sc;
    }
    return sc;
  }

  StatusCode detachServices()
  {
    if ( m_addrCreator ) m_addrCreator->release();
    if ( m_dataLoader ) m_dataLoader->release();
    m_addrCreator = nullptr;
    m_dataLoader  = nullptr;
    return StatusCode::SUCCESS;
  }

  //
  //---IService implemenation---------------------------------------------------------
  //

  StatusCode initialize() override
  {
    StatusCode sc = Service::initialize();
    if ( !sc.isSuccess() ) {
      error() << "Unable to initialize base class" << endmsg;
      return sc;
    }
    if ( m_slots < (size_t)1 ) {
      error() << "Invalid number of slots (" << m_slots << ")" << endmsg;
      return StatusCode::FAILURE;
    }

    if ( !setNumberOfStores( m_slots ).isSuccess() ) {
      error() << "Cannot set number of slots" << endmsg;
      return StatusCode::FAILURE;
    }

    m_partitions = std::vector<Synced<Partition>>( m_slots );
    for ( size_t i = 0; i < m_slots; i++ ) {
      DataSvc* svc = new DataSvc( name() + "_" + std::to_string( i ), serviceLocator() );
      // Percolate properties
      svc->setProperty( m_rootCLID ).ignore();
      svc->setProperty( m_rootName ).ignore();
      svc->setProperty( m_forceLeaves ).ignore();
      svc->setProperty( m_enableFaultHdlr ).ignore();
      // make sure that CommonMessaging is initialized
      svc->setProperty( m_outputLevel ).ignore();

      sc = svc->initialize();
      if ( !sc.isSuccess() ) {
        error() << "Failed to instantiate DataSvc as store partition" << endmsg;
        return sc;
      }
      m_partitions[i].with_lock( [&]( Partition& p ) {
        p.dataProvider = svc;
        p.dataManager  = svc;
      } );
      m_freeSlots.push( i );
    }
    selectStore( 0 ).ignore();
    return attachServices();
  }

  StatusCode reinitialize() override
  {
    StatusCode sc = Service::reinitialize();
    if ( !sc.isSuccess() ) {
      error() << "Unable to reinitialize base class" << endmsg;
      return sc;
    }
    detachServices();
    sc = attachServices();
    if ( !sc.isSuccess() ) {
      error() << "Failed to attach necessary services." << endmsg;
      return sc;
    }
    return StatusCode::SUCCESS;
  }

  StatusCode finalize() override
  {
    setDataLoader( 0 ).ignore();
    clearStore().ignore();
    return Service::finalize();
  }
};

// Instantiation of a static factory class used by clients to create
// instances of this service
DECLARE_COMPONENT( HiveWhiteBoard )