MultiStoreSvc.cpp

Go to the documentation of this file.

//====================================================================
//  MultiStoreSvc.cpp
//--------------------------------------------------------------------
//
//  Package    : System ( The LHCb Offline System)
//
//  Description: implementation of the Transient event data service.
//
//  Author     : M.Frank
//  History    :
// +---------+----------------------------------------------+---------
// |    Date |                 Comment                      | Who
// +---------+----------------------------------------------+---------
// | 29/10/98| Initial version                              | MF
// +---------+----------------------------------------------+---------
//
//====================================================================
#define DATASVC_MULTISTORESVC_CPP

// Include files
#include "GaudiKernel/AttribStringParser.h"
#include "GaudiKernel/DataObject.h"
#include "GaudiKernel/IAddressCreator.h"
#include "GaudiKernel/IConversionSvc.h"
#include "GaudiKernel/IDataManagerSvc.h"
#include "GaudiKernel/IDataProviderSvc.h"
#include "GaudiKernel/IOpaqueAddress.h"
#include "GaudiKernel/IPartitionControl.h"
#include "GaudiKernel/ISvcLocator.h"
#include "GaudiKernel/ISvcManager.h"
#include "GaudiKernel/MsgStream.h"
#include "GaudiKernel/Service.h"
#include "GaudiKernel/SmartIF.h"
#include "GaudiKernel/TypeNameString.h"
#include "GaudiKernel/compose.h"
#include "boost/variant.hpp"
#include <map>

// Forward declarations
// This class
class MultiStoreSvc;

typedef const std::string CSTR;
typedef IDataStoreAgent   AGENT;
typedef DataObject        OBJECT;
typedef IOpaqueAddress    ADDRESS;

namespace
{
  struct Partition final {
    SmartIF<IDataProviderSvc> dataProvider;
    SmartIF<IDataManagerSvc>  dataManager;
    std::string               name;

    template <typename T>
    T* get();
  };
  template <>
  IDataProviderSvc* Partition::get<IDataProviderSvc>()
  {
    return dataProvider.get();
  }
  template <>
  IDataManagerSvc* Partition::get<IDataManagerSvc>()
  {
    return dataManager.get();
  }

  namespace detail
  {
    template <typename lambda>
    struct arg_helper : public arg_helper<decltype( &lambda::operator() )> {
    };
    template <typename T, typename Ret, typename Arg>
    struct arg_helper<Ret ( T::* )( Arg ) const> {
      using type = Arg;
    };

    // given a unary lambda whose argument is of type Arg_t,
    // argument_t<lambda> will be equal to Arg_t
    template <typename lambda>
    using argument_t = typename arg_helper<lambda>::type;
  }
  auto dispatch_variant = []( auto&& variant, auto&&... lambdas ) -> decltype( auto ) {
    return boost::apply_visitor( compose( std::forward<decltype( lambdas )>( lambdas )... ),
                                 std::forward<decltype( variant )>( variant ) );
  };
}

class MultiStoreSvc : public extends<Service, IDataProviderSvc, IDataManagerSvc, IPartitionControl>
{
protected:
  typedef std::vector<std::string> PartitionDefs;
  typedef std::map<std::string, Partition> Partitions;

  Gaudi::Property<CLID>          m_rootCLID{this, "RootCLID", 110, "CLID of root entry"};
  Gaudi::Property<std::string>   m_rootName{this, "RootName", "/Event", "name of root entry"};
  Gaudi::Property<PartitionDefs> m_partitionDefs{this, "Partitions", {}, "datastore partition definitions"};
  Gaudi::Property<std::string>   m_loader{this, "DataLoader", "EventPersistencySvc", "data loader name"};
  Gaudi::Property<std::string>   m_defaultPartition{this, "DefaultPartition", "Default", "default partition name"};

  SmartIF<IConversionSvc> m_dataLoader;
  SmartIF<IAddressCreator> m_addrCreator;
  struct tagROOT {
    std::string path;
    boost::variant<boost::blank, ADDRESS*, OBJECT*> root;
  } m_root;
  Partition m_current;
  Partitions m_partitions;

  // member templates to help writing the function calls
  template <typename Fun>
  StatusCode fwd( Fun f )
  {
    auto* svc = m_current.get<std::decay_t<detail::argument_t<Fun>>>();
    return svc ? f( *svc ) : IDataProviderSvc::Status::INVALID_ROOT;
  }

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

  StatusCode registerAddress( boost::string_ref path, ADDRESS* pAddr ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.registerAddress( path, pAddr ); } );
  }
  StatusCode registerAddress( OBJECT* parent, boost::string_ref path, ADDRESS* pAddr ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.registerAddress( parent, path, pAddr ); } );
  }
  StatusCode registerAddress( IRegistry* parent, boost::string_ref path, ADDRESS* pAddr ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.registerAddress( parent, path, pAddr ); } );
  }
  StatusCode unregisterAddress( boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.unregisterAddress( path ); } );
  }
  StatusCode unregisterAddress( OBJECT* pParent, boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.unregisterAddress( pParent, path ); } );
  }
  StatusCode unregisterAddress( IRegistry* pParent, boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.unregisterAddress( pParent, path ); } );
  }
  StatusCode objectLeaves( const OBJECT* pObject, std::vector<IRegistry*>& leaves ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.objectLeaves( pObject, leaves ); } );
  }
  StatusCode objectLeaves( const IRegistry* pObject, std::vector<IRegistry*>& leaves ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.objectLeaves( pObject, leaves ); } );
  }
  StatusCode objectParent( const OBJECT* pObject, IRegistry*& refpParent ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.objectParent( pObject, refpParent ); } );
  }
  StatusCode objectParent( const IRegistry* pObject, IRegistry*& refpParent ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.objectParent( pObject, refpParent ); } );
  }
  StatusCode clearSubTree( boost::string_ref path ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.clearSubTree( path ); } );
  }
  StatusCode clearSubTree( OBJECT* pObject ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.clearSubTree( pObject ); } );
  }
  StatusCode clearStore() override
  {
    for ( auto& i : m_partitions ) {
      i.second.dataManager->clearStore().ignore();
    }
    dispatch_variant( m_root.root,
                      []( auto* p ) {
                        if ( p ) p->release();
                      },
                      []( boost::blank ) {} );
    m_root.root = {};
    m_root.path.clear();
    return StatusCode::SUCCESS;
  }
  StatusCode traverseSubTree( boost::string_ref path, AGENT* pAgent ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.traverseSubTree( path, pAgent ); } );
  }
  StatusCode traverseSubTree( OBJECT* pObject, AGENT* pAgent ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.traverseSubTree( pObject, pAgent ); } );
  }
  StatusCode traverseTree( AGENT* pAgent ) override
  {
    return fwd( [&]( IDataManagerSvc& svc ) { return svc.traverseTree( pAgent ); } );
  }
  StatusCode setRoot( std::string path, OBJECT* pObj ) override
  {
    dispatch_variant( m_root.root,
                      []( auto* p ) {
                        if ( p ) p->release();
                      },
                      []( boost::blank ) {} );
    m_root.path = std::move( path );
    m_root.root = pObj;
    preparePartitions();
    return activate( m_defaultPartition );
  }

  StatusCode setRoot( std::string path, ADDRESS* pAddr ) override
  {
    dispatch_variant( m_root.root,
                      []( auto* p ) {
                        if ( p ) p->release();
                      },
                      []( boost::blank ) {} );
    m_root.path = std::move( path );
    m_root.root = pAddr;
    if ( !pAddr ) return StatusCode::FAILURE;
    pAddr->addRef();
    preparePartitions();
    return activate( m_defaultPartition );
  }
  StatusCode setDataLoader( IConversionSvc* pDataLoader, IDataProviderSvc* dpsvc = nullptr ) override
  {
    m_dataLoader = pDataLoader;
    if ( m_dataLoader ) m_dataLoader->setDataProvider( dpsvc ? dpsvc : this );
    for ( auto& i : m_partitions ) {
      i.second.dataManager->setDataLoader( m_dataLoader.get() ).ignore();
    }
    return StatusCode::SUCCESS;
  }
  StatusCode addPreLoadItem( const DataStoreItem& item ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.addPreLoadItem( item ); } );
  }
  StatusCode addPreLoadItem( std::string item ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.addPreLoadItem( std::move( item ) ); } );
  }
  StatusCode removePreLoadItem( const DataStoreItem& item ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.removePreLoadItem( item ); } );
  }
  StatusCode removePreLoadItem( std::string item ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.removePreLoadItem( std::move( item ) ); } );
  }
  StatusCode resetPreLoad() override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.resetPreLoad(); } );
  }
  StatusCode preLoad() override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.preLoad(); } );
  }
  StatusCode registerObject( boost::string_ref path, OBJECT* pObj ) override
  {
    return registerObject( nullptr, path, pObj );
  }
  StatusCode registerObject( boost::string_ref parent, boost::string_ref obj, OBJECT* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.registerObject( parent, obj, pObj ); } );
  }
  StatusCode registerObject( boost::string_ref parent, int item, OBJECT* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.registerObject( parent, item, pObj ); } );
  }
  StatusCode registerObject( OBJECT* parent, boost::string_ref obj, OBJECT* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.registerObject( parent, obj, pObj ); } );
  }
  StatusCode registerObject( OBJECT* parent, int obj, OBJECT* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.registerObject( parent, obj, pObj ); } );
  }
  StatusCode unregisterObject( boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unregisterObject( path ); } );
  }
  StatusCode unregisterObject( boost::string_ref parent, boost::string_ref obj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unregisterObject( parent, obj ); } );
  }
  StatusCode unregisterObject( boost::string_ref parent, int obj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unregisterObject( parent, obj ); } );
  }
  StatusCode unregisterObject( OBJECT* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unregisterObject( pObj ); } );
  }
  StatusCode unregisterObject( OBJECT* pObj, boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unregisterObject( pObj, path ); } );
  }
  StatusCode unregisterObject( OBJECT* pObj, int item ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unregisterObject( pObj, item ); } );
  }
  StatusCode retrieveObject( IRegistry* parent, boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( parent, path, pObj ); } );
  }
  StatusCode retrieveObject( boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( path, pObj ); } );
  }
  StatusCode retrieveObject( boost::string_ref parent, boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( parent, path, pObj ); } );
  }
  StatusCode retrieveObject( boost::string_ref parent, int item, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( parent, item, pObj ); } );
  }
  StatusCode retrieveObject( OBJECT* parent, boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( parent, path, pObj ); } );
  }
  StatusCode retrieveObject( OBJECT* parent, int item, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( parent, item, pObj ); } );
  }
  StatusCode findObject( boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( path, pObj ); } );
  }
  StatusCode findObject( IRegistry* parent, boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.findObject( parent, path, pObj ); } );
  }
  StatusCode findObject( boost::string_ref parent, boost::string_ref path, OBJECT*& pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.retrieveObject( parent, path, pObj ); } );
  }
  StatusCode findObject( boost::string_ref parent, int item, OBJECT*& pObject ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.findObject( parent, item, pObject ); } );
  }
  StatusCode findObject( OBJECT* parent, boost::string_ref path, OBJECT*& pObject ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.findObject( parent, path, pObject ); } );
  }
  StatusCode findObject( OBJECT* parent, int item, OBJECT*& pObject ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.findObject( parent, item, pObject ); } );
  }
  StatusCode linkObject( IRegistry* from, boost::string_ref objPath, OBJECT* to ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.linkObject( from, objPath, to ); } );
  }
  StatusCode linkObject( boost::string_ref from, boost::string_ref objPath, OBJECT* to ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.linkObject( from, objPath, to ); } );
  }
  StatusCode linkObject( OBJECT* from, boost::string_ref objPath, OBJECT* to ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.linkObject( from, objPath, to ); } );
  }
  StatusCode linkObject( boost::string_ref fullPath, OBJECT* to ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.linkObject( fullPath, to ); } );
  }
  StatusCode unlinkObject( IRegistry* from, boost::string_ref objPath ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unlinkObject( from, objPath ); } );
  }
  StatusCode unlinkObject( boost::string_ref from, boost::string_ref objPath ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unlinkObject( from, objPath ); } );
  }
  StatusCode unlinkObject( OBJECT* from, boost::string_ref objPath ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unlinkObject( from, objPath ); } );
  }
  StatusCode unlinkObject( boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.unlinkObject( path ); } );
  }
  StatusCode updateObject( IRegistry* pDirectory ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.updateObject( pDirectory ); } );
  }
  StatusCode updateObject( boost::string_ref path ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.updateObject( path ); } );
  }
  StatusCode updateObject( OBJECT* pObj ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.updateObject( pObj ); } );
  }
  StatusCode updateObject( boost::string_ref parent, boost::string_ref updatePath ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.updateObject( parent, updatePath ); } );
  }
  StatusCode updateObject( OBJECT* parent, boost::string_ref updatePath ) override
  {
    return fwd( [&]( IDataProviderSvc& svc ) { return svc.updateObject( parent, updatePath ); } );
  }

  StatusCode create( CSTR& nam, CSTR& typ ) override
  {
    IInterface* pPartition = nullptr;
    return create( nam, typ, pPartition );
  }
  StatusCode create( CSTR& nam, CSTR& typ, IInterface*& pPartition ) override
  {
    if ( get( nam, pPartition ).isSuccess() ) return IPartitionControl::Status::PARTITION_EXISTS;
    auto isvc = serviceLocator()->service<IService>( typ );
    if ( !isvc ) return IInterface::Status::NO_INTERFACE;
    auto dataMgr  = isvc.as<IDataManagerSvc>();
    auto dataProv = isvc.as<IDataProviderSvc>();
    if ( !dataMgr || !dataProv ) return IInterface::Status::NO_INTERFACE;
    m_partitions.emplace( nam, Partition{dataProv, dataMgr, nam} );
    return StatusCode::SUCCESS;
  }

  StatusCode drop( CSTR& nam ) override
  {
    auto i = m_partitions.find( nam );
    if ( i == m_partitions.end() ) return IPartitionControl::Status::PARTITION_NOT_PRESENT;
    if ( i->second.dataManager == m_current.dataManager ) {
      m_current = Partition();
    }
    i->second.dataManager->clearStore().ignore();
    m_partitions.erase( i );
    return StatusCode::SUCCESS;
  }

  StatusCode drop( IInterface* pPartition ) override
  {
    auto provider = SmartIF<IDataProviderSvc>( pPartition );
    if ( !provider ) return IInterface::Status::NO_INTERFACE;
    auto i = std::find_if( std::begin( m_partitions ), std::end( m_partitions ),
                           [&]( Partitions::const_reference p ) { return p.second.dataProvider == provider; } );
    if ( i == std::end( m_partitions ) ) return IPartitionControl::Status::PARTITION_NOT_PRESENT;
    i->second.dataManager->clearStore().ignore();
    m_partitions.erase( i );
    return StatusCode::SUCCESS;
  }

  StatusCode activate( CSTR& nam ) override
  {
    auto i = m_partitions.find( nam );
    if ( i != m_partitions.end() ) {
      m_current = i->second;
      return StatusCode::SUCCESS;
    }
    m_current = {};
    return IPartitionControl::Status::PARTITION_NOT_PRESENT;
  }

  StatusCode activate( IInterface* pPartition ) override
  {
    auto provider = SmartIF<IDataProviderSvc>( pPartition );
    m_current     = Partition();
    if ( !provider ) return IInterface::Status::NO_INTERFACE;
    auto i = std::find_if( std::begin( m_partitions ), std::end( m_partitions ),
                           [&]( Partitions::const_reference p ) { return p.second.dataProvider == provider; } );
    if ( i == std::end( m_partitions ) ) return IPartitionControl::Status::PARTITION_NOT_PRESENT;
    m_current = i->second;
    return StatusCode::SUCCESS;
  }

  StatusCode get( CSTR& nam, IInterface*& pPartition ) const override
  {
    auto i = m_partitions.find( nam );
    if ( i != m_partitions.end() ) {
      pPartition = i->second.dataProvider;
      return StatusCode::SUCCESS;
    }
    pPartition = nullptr;
    return IPartitionControl::Status::PARTITION_NOT_PRESENT;
  }

  StatusCode activePartition( std::string& nam, IInterface*& pPartition ) const override
  {
    if ( m_current.dataProvider ) {
      nam        = m_current.name;
      pPartition = m_current.dataProvider;
      return StatusCode::SUCCESS;
    }
    nam.clear();
    pPartition = nullptr;
    return IPartitionControl::Status::NO_ACTIVE_PARTITION;
  }

  StatusCode attachServices()
  {
    // Attach address creator facility
    m_addrCreator = service( m_loader, true );
    if ( !m_addrCreator ) {
      error() << "Failed to retrieve data loader "
              << "\"" << m_loader << "\"" << endmsg;
      return StatusCode::FAILURE;
    }
    // Attach data loader facility
    auto dataLoader = service<IConversionSvc>( m_loader, true );
    if ( !dataLoader ) {
      error() << "Failed to retrieve data loader "
              << "\"" << m_loader << "\"" << endmsg;
      return StatusCode::FAILURE;
    }
    auto sc = setDataLoader( dataLoader.get() );
    if ( !sc.isSuccess() ) {
      error() << "Failed to set data loader "
              << "\"" << m_loader << "\"" << endmsg;
    }
    return sc;
  }

  StatusCode detachServices()
  {
    m_addrCreator.reset();
    m_dataLoader.reset();
    return StatusCode::SUCCESS;
  }

  StatusCode initialize() override
  {
    // Nothing to do: just call base class initialisation
    StatusCode sc = Service::initialize();
    if ( !sc.isSuccess() ) return sc;
    sc = makePartitions();
    if ( !sc.isSuccess() ) {
      error() << "Failed to connect to all store partitions." << endmsg;
      return sc;
    }
    return attachServices();
  }

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

  StatusCode finalize() override
  {
    setDataLoader( nullptr ).ignore();
    clearStore().ignore();
    clearPartitions().ignore();
    m_current = Partition();
    detachServices();
    return Service::finalize();
  }

  // protected:

  using extends::extends;

  ~MultiStoreSvc() override
  {
    setDataLoader( nullptr ).ignore();
    resetPreLoad().ignore();
    clearStore().ignore();
    clearPartitions().ignore();
  }

  StatusCode preparePartitions()
  {
    StatusCode iret = StatusCode::SUCCESS;
    for ( auto& i : m_partitions ) {
      StatusCode sc =
          dispatch_variant( m_root.root,
                            [&]( ADDRESS* address ) -> StatusCode {
                              if ( !address ) return StatusCode::FAILURE;
                              ADDRESS* pAdd = nullptr;
                              ADDRESS* p    = address;
                              auto     sc =
                                  m_addrCreator->createAddress( p->svcType(), p->clID(), p->par(), p->ipar(), pAdd );
                              return sc.isSuccess() ? i.second.dataManager->setRoot( m_root.path, pAdd ) : sc;
                            },
                            [&]( OBJECT* object ) -> StatusCode {
                              if ( object && object->clID() == CLID_DataObject ) {
                                return i.second.dataManager->setRoot( m_root.path, new DataObject() );
                              }
                              return StatusCode::FAILURE;
                            },
                            []( boost::blank ) -> StatusCode { return StatusCode::FAILURE; } );
      if ( !sc.isSuccess() ) iret = sc;
    }
    return iret;
  }

  StatusCode clearPartitions()
  {
    for ( auto& i : m_partitions ) i.second.dataManager->clearStore().ignore();
    m_partitions.clear();
    return StatusCode::SUCCESS;
  }

  StatusCode makePartitions()
  {
    using Parser = Gaudi::Utils::AttribStringParser;
    std::string typ, nam;
    clearPartitions().ignore();
    for ( auto part : m_partitionDefs ) {
      for ( auto attrib : Parser( std::move( part ) ) ) {
        switch (::toupper( attrib.tag[0] ) ) {
        case 'N':
          nam = std::move( attrib.value );
          break;
        case 'T':
          typ = std::move( attrib.value );
          break;
        }
      }
      StatusCode sc = create( nam, typ );
      if ( !sc.isSuccess() ) return sc;
      if ( m_defaultPartition.empty() ) m_defaultPartition = nam;
    }
    return StatusCode::SUCCESS;
  }
};

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