HiveWhiteBoard.cpp

Go to the documentation of this file.

//====================================================================
//  WhiteBoard (Concurrent Event Data Store)
//--------------------------------------------------------------------
//
//====================================================================
#define  WHITEBOARD_CPP

// Include files
#include "GaudiKernel/Service.h"
#include "GaudiKernel/SmartIF.h"
#include "GaudiKernel/TypeNameString.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/SvcFactory.h"
#include "GaudiKernel/DataObject.h"
#include "GaudiKernel/DataSvc.h"
#include "GaudiKernel/ThreadLocalPtr.h"
#include "tbb/spin_mutex.h"
#include "tbb/recursive_mutex.h"


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

// Forward declarations

typedef  tbb::recursive_mutex wbMutex;
//typedef  tbb::spin_mutex wbMutex;

namespace {
  struct Partition  {

    SmartIF<IDataProviderSvc>        dataProvider;
    SmartIF<IDataManagerSvc>         dataManager;
    wbMutex               storeMutex;
    std::vector<std::string> newDataObjects;
    int                      eventNumber;
    Partition() : dataProvider(0), dataManager(0), eventNumber(-1) {}
    Partition(IDataProviderSvc* dp, IDataManagerSvc* dm) : dataProvider(dp), dataManager(dm), eventNumber(-1) {}
    Partition(const Partition& entry) : dataProvider(entry.dataProvider), dataManager(entry.dataManager), eventNumber(entry.eventNumber) {}
    Partition& operator=(const Partition& entry)  {
      dataProvider = entry.dataProvider;
      dataManager = entry.dataManager;
      return *this;
    }
  };
  class DataAgent : virtual public IDataStoreAgent  {
  private:
    std::vector<std::string>& m_dataObjects;
  public:
    DataAgent(std::vector<std::string>& objs) : m_dataObjects(objs) {}
    virtual ~DataAgent() {}
    virtual bool analyse(IRegistry* pReg, int )   {
      if (0 != pReg->object()) {
        m_dataObjects.emplace_back(pReg->identifier());
        return true;
      }
      else {
        return false;
      }
    }
  };
}

THREAD_LOCAL_PTR Partition* s_current(0);

class HiveWhiteBoard: public extends3<Service,
                                     IDataProviderSvc,
                                     IDataManagerSvc,
                                     IHiveWhiteBoard>
{
protected:
  typedef std::vector<Partition> Partitions;

  CLID                m_rootCLID;
  std::string         m_rootName;
  std::string         m_loader;
  IConversionSvc*     m_dataLoader;
  IAddressCreator*    m_addrCreator;
  Partitions          m_partitions;
  int              m_slots;
  bool                m_forceLeaves;
  bool m_enableFaultHdlr;

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

// macro to help writing the function calls
#define _CALL(P,F,ARGS) if(s_current) { \
wbMutex::scoped_lock lock; lock.acquire(s_current->storeMutex);\
return s_current->P ? s_current->P->F ARGS : IDataProviderSvc::INVALID_ROOT; }\
return IDataProviderSvc::INVALID_ROOT;

  StatusCode registerAddress(const std::string& path, IOpaqueAddress* pAddr) override   {
    _CALL(dataManager, registerAddress, (path, pAddr));
  }
  StatusCode registerAddress(DataObject* parent, const std::string& path, IOpaqueAddress* pAddr) override  {
    _CALL(dataManager, registerAddress, (parent, path, pAddr));
  }
  StatusCode registerAddress(IRegistry* parent, const std::string& path, IOpaqueAddress* pAdd) override  {
    _CALL(dataManager, registerAddress, (parent, path, pAdd));
  }
  StatusCode unregisterAddress(const std::string& path) override  {
    _CALL(dataManager, unregisterAddress, (path));
  }
  StatusCode unregisterAddress(DataObject* pParent, const std::string& path) override  {
    _CALL(dataManager, unregisterAddress, (pParent, path));
  }
  StatusCode unregisterAddress(IRegistry* pParent, const std::string& path) override  {
    _CALL(dataManager, unregisterAddress, (pParent, path));
  }
  StatusCode objectLeaves(const DataObject*  pObject, std::vector<IRegistry*>& leaves) override  {
    _CALL(dataManager, objectLeaves, (pObject, leaves));
  }
  StatusCode objectLeaves(const IRegistry* pObject, std::vector<IRegistry*>& leaves) override  {
    _CALL(dataManager, objectLeaves, (pObject, leaves));
  }
  StatusCode objectParent(const DataObject* pObject, IRegistry*& refpParent) override  {
    _CALL(dataManager, objectParent, (pObject, refpParent));
  }
  StatusCode objectParent(const IRegistry* pObject, IRegistry*& refpParent) override  {
    _CALL(dataManager, objectParent, (pObject, refpParent));
  }
  StatusCode clearSubTree(const std::string& path) override  {
    _CALL(dataManager, clearSubTree, (path));
  }
  StatusCode clearSubTree(DataObject* pObject) override  {
    _CALL(dataManager, clearSubTree, (pObject));
  }
  StatusCode clearStore() override  {
    for(auto& p: m_partitions) p.dataManager->clearStore().ignore();
    return StatusCode::SUCCESS;
  }

  StatusCode traverseSubTree(const std::string& path, IDataStoreAgent* pAgent) override  {
    _CALL(dataManager, traverseSubTree, (path, pAgent));
  }
  StatusCode traverseSubTree(DataObject* pObject, IDataStoreAgent* pAgent) override  {
    _CALL(dataManager, traverseSubTree, (pObject, pAgent));
  }
  StatusCode traverseTree( IDataStoreAgent* pAgent ) override  {
    _CALL(dataManager, traverseTree, (pAgent));
  }
  StatusCode setRoot(std::string path, DataObject* pObj) override {
    _CALL(dataManager, setRoot, (path, pObj));
  }

  StatusCode setRoot (std::string path, IOpaqueAddress* pAddr) override {
    _CALL(dataManager, setRoot, (path, pAddr));
  }

  StatusCode setDataLoader(IConversionSvc* pDataLoader,
                           IDataProviderSvc* dpsvc __attribute__((unused)) = nullptr) override {
    if ( 0 != pDataLoader  ) pDataLoader->addRef();
    if ( 0 != m_dataLoader ) m_dataLoader->release();
    if ( 0 != pDataLoader  )    {
      pDataLoader->setDataProvider(this);
    }
    m_dataLoader = pDataLoader;
    for(auto& p: m_partitions) {
      p.dataManager->setDataLoader(m_dataLoader,this).ignore();
    }
    return SUCCESS;
  }
  StatusCode addPreLoadItem(const DataStoreItem& item) override    {
    for(auto& p: m_partitions) p.dataProvider->addPreLoadItem(item);
    return StatusCode::SUCCESS;
  }
  StatusCode addPreLoadItem(const std::string& item) override   {
    for(auto& p: m_partitions) p.dataProvider->addPreLoadItem(item);
    return StatusCode::SUCCESS;
  }
  StatusCode removePreLoadItem(const DataStoreItem& item) override  {
    for(auto& p: m_partitions) p.dataProvider->removePreLoadItem(item);
    return StatusCode::SUCCESS;
  }
  StatusCode removePreLoadItem(const std::string& item) override  {
    for(auto& p: m_partitions) p.dataProvider->removePreLoadItem(item);
    return StatusCode::SUCCESS;
  }
  StatusCode resetPreLoad() override {
    for(auto& p: m_partitions) p.dataProvider->resetPreLoad();
    return StatusCode::SUCCESS;
  }
  StatusCode preLoad() override  {
    wbMutex::scoped_lock lock; lock.acquire(s_current->storeMutex);
    StatusCode sc = s_current->dataProvider->preLoad();
    DataAgent da(s_current->newDataObjects);
    s_current->dataManager->traverseTree(&da);
    return sc;
  }
  StatusCode registerObject(const std::string& path, DataObject* pObj) override  {
    wbMutex::scoped_lock lock; lock.acquire(s_current->storeMutex);
    StatusCode sc = s_current->dataProvider->registerObject(path, pObj);
    if( sc.isSuccess()) {
      s_current->newDataObjects.push_back(path);
    }
    return sc;
  }
  StatusCode registerObject(const std::string& parent, const std::string& obj, DataObject* pObj) override  {
    _CALL(dataProvider, registerObject, (parent, obj, pObj));
  }
  StatusCode registerObject(const std::string& parent, int item, DataObject* pObj) override  {
    _CALL(dataProvider, registerObject, (parent, item, pObj));
  }
  StatusCode registerObject(DataObject* parent, const std::string& obj, DataObject* pObj) override  {
    _CALL(dataProvider, registerObject, (parent, obj, pObj));
  }
  StatusCode registerObject(DataObject* parent, int obj, DataObject* pObj) override  {
    _CALL(dataProvider, registerObject, (parent, obj, pObj));
  }
  StatusCode unregisterObject(const std::string& path) override   {
    _CALL(dataProvider, unregisterObject, (path));
  }
  StatusCode unregisterObject(const std::string& parent, const std::string& obj) override  {
    _CALL(dataProvider, unregisterObject, (parent, obj));
  }
  StatusCode unregisterObject(const std::string& parent, int obj) override  {
    _CALL(dataProvider, unregisterObject, (parent, obj));
  }
  StatusCode unregisterObject(DataObject* pObj) override  {
    _CALL(dataProvider, unregisterObject, (pObj));
  }
  StatusCode unregisterObject(DataObject* pObj, const std::string& path) override  {
    _CALL(dataProvider, unregisterObject, (pObj, path));
  }
  StatusCode unregisterObject(DataObject* pObj, int item ) override  {
    _CALL(dataProvider, unregisterObject, (pObj, item));
  }
  StatusCode retrieveObject(IRegistry* parent, const std::string& path, DataObject*& pObj ) override  {
    _CALL(dataProvider, retrieveObject, (parent, path, pObj));
  }
  StatusCode retrieveObject(const std::string& path, DataObject*& pObj) override  {
    _CALL(dataProvider, retrieveObject, (path, pObj));
  }
  StatusCode retrieveObject(const std::string& parent, const std::string& path, DataObject*& pObj ) override  {
    _CALL(dataProvider, retrieveObject, (parent, path, pObj));
  }
  StatusCode retrieveObject(const std::string& parent, int item, DataObject*& pObj) override  {
    _CALL(dataProvider, retrieveObject, (parent, item, pObj));
  }
  StatusCode retrieveObject(DataObject* parent, const std::string& path, DataObject*& pObj ) override  {
    _CALL(dataProvider, retrieveObject, (parent, path, pObj));
  }
  StatusCode retrieveObject(DataObject* parent, int item, DataObject*& pObj ) override  {
    _CALL(dataProvider, retrieveObject, (parent, item, pObj));
  }
  StatusCode findObject(const std::string& path, DataObject*& pObj) override  {
    _CALL(dataProvider, retrieveObject, (path, pObj));
  }
  StatusCode findObject(IRegistry* parent, const std::string& path, DataObject*& pObj) override  {
    _CALL(dataProvider, retrieveObject, (parent, path, pObj));
  }
  StatusCode findObject(const std::string& parent, const std::string& path, DataObject*& pObj) override  {
    _CALL(dataProvider, retrieveObject, (parent, path, pObj));
  }
  StatusCode findObject(const std::string& parent, int item, DataObject*& pObject ) override {
    _CALL(dataProvider, findObject, (parent, item, pObject));
  }
  StatusCode findObject(DataObject* parent, const std::string& path, DataObject*& pObject) override  {
    _CALL(dataProvider, findObject, (parent, path, pObject));
  }
  StatusCode findObject(DataObject* parent, int item, DataObject*& pObject) override  {
    _CALL(dataProvider, findObject, (parent, item, pObject));
  }
  StatusCode linkObject(IRegistry* from, const std::string& objPath, DataObject* to) override  {
    _CALL(dataProvider, linkObject, (from, objPath, to));
  }
  StatusCode linkObject(const std::string& from, const std::string& objPath, DataObject* to) override   {
    _CALL(dataProvider, linkObject, (from, objPath, to));
  }
  StatusCode linkObject(DataObject* from, const std::string& objPath, DataObject* to) override  {
    _CALL(dataProvider, linkObject, (from, objPath, to));
  }
  StatusCode linkObject(const std::string& fullPath, DataObject* to) override  {
    _CALL(dataProvider, linkObject, (fullPath, to));
  }
  StatusCode unlinkObject(IRegistry* from, const std::string& objPath) override  {
    _CALL(dataProvider, unlinkObject, (from, objPath));
  }
  StatusCode unlinkObject(const std::string& from, const std::string& objPath) override  {
    _CALL(dataProvider, unlinkObject, (from, objPath));
  }
  StatusCode unlinkObject(DataObject* from, const std::string& objPath) override  {
    _CALL(dataProvider, unlinkObject, (from, objPath));
  }
  StatusCode unlinkObject(const std::string& path) override {
    _CALL(dataProvider, unlinkObject, (path));
  }
  StatusCode updateObject(IRegistry* pDirectory ) override  {
    _CALL(dataProvider, updateObject, (pDirectory));
  }
  StatusCode updateObject(const std::string& path) override  {
    _CALL(dataProvider, updateObject, (path));
  }
  StatusCode updateObject(DataObject* pObj ) override  {
    _CALL(dataProvider, updateObject, (pObj));
  }
  StatusCode updateObject(const std::string& parent, const std::string& updatePath ) override  {
    _CALL(dataProvider, updateObject, (parent, updatePath));
  }
  StatusCode updateObject(DataObject* parent, const std::string& updatePath) override  {
    _CALL(dataProvider, updateObject, (parent, updatePath));
  }

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

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

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

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

  size_t getNumberOfStores() override {
    return m_slots;
  }

  StatusCode getNewDataObjects(std::vector<std::string>& products) override {
    wbMutex::scoped_lock lock; lock.acquire(s_current->storeMutex);
    products = s_current->newDataObjects;
    s_current->newDataObjects.clear();
    return StatusCode::SUCCESS;
  }

  bool newDataObjectsPresent() override {
    wbMutex::scoped_lock lock; lock.acquire(s_current->storeMutex);
    return s_current->newDataObjects.size()!=0;
  }

  size_t allocateStore( int evtnumber ) override {
    //size_t free = std::string::npos;
    size_t index = 0;
    for (auto& p : m_partitions ) {
      if( p.eventNumber == evtnumber) {
        error() << "Attempt to allocate a store partition for an event that is still active" << endmsg;
        return std::string::npos;
      } else if (p.eventNumber == -1) {
        p.eventNumber = evtnumber;
        //info() << "Got allocated slot..." << index << endmsg;
        return index;
      }
      index++;
    }
    return std::string::npos;
  }

  StatusCode freeStore( size_t partition ) override {
    m_partitions[partition].eventNumber = -1;
    //info() << "Freed slot..." << partition << endmsg;
    return StatusCode::SUCCESS;
  }


  size_t getPartitionNumber(int eventnumber) const override {
    size_t index{};
    for (auto& p : m_partitions) {
      if (p.eventNumber == eventnumber) return index;
      ++index;
    }
    return 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 = 0;
    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 = 0;
    m_dataLoader = 0;
    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 (1 > m_slots ){
      error() << "Invalid number of slots (" << m_slots << ")" << endmsg;
      return StatusCode::FAILURE;
    }

    for( int i = 0; i< m_slots; i++) {
      std::ostringstream oss;
      oss << name() << "_" << i;
      DataSvc* svc = new DataSvc(oss.str(), serviceLocator());
      // Percolate properties
      svc->setProperty("RootCLID", std::to_string(m_rootCLID));
      svc->setProperty("RootName", m_rootName);
      svc->setProperty("ForceLeaves", std::to_string(m_forceLeaves));
      svc->setProperty("EnableFaultHandler", std::to_string(m_enableFaultHdlr));

      sc = svc->initialize();
      if (!sc.isSuccess()) {
        error() << "Failed to instantiate DataSvc as store partition" << endmsg;
        return sc;
      }
      m_partitions.push_back(Partition(svc, svc));
    }
    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();
  }


//protected:

  HiveWhiteBoard( const std::string& name, ISvcLocator* svc )
  : base_class(name,svc), m_rootCLID(110), m_rootName("/Event"),
    m_dataLoader(0), m_addrCreator(0)
  {
    m_dataLoader = 0;
    declareProperty("RootCLID",           m_rootCLID);
    declareProperty("RootName",           m_rootName);
    declareProperty("DataLoader",         m_loader="EventPersistencySvc");
    declareProperty("EventSlots",         m_slots = 1);
    declareProperty("ForceLeaves",        m_forceLeaves = 1);
    declareProperty("EnableFaultHandler", m_enableFaultHdlr = 0);
  }

  virtual ~HiveWhiteBoard()  {
    setDataLoader(0).ignore();
    resetPreLoad().ignore();
    clearStore().ignore();

    for(Partitions::iterator i = m_partitions.begin(); i != m_partitions.end(); ++i) {
      (*i).dataManager->release();
      (*i).dataProvider->release();
    }
    m_partitions.clear();
  }

};

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