TsDataSvc.cpp

Go to the documentation of this file.

//====================================================================
//  TsDataSvc.cpp
//--------------------------------------------------------------------
//
//  Package    : System ( The LHCb Offline System)
//
//  Description: implementation of the Transient data service: TsDataSvc
//
//  Author     : M.Frank
//  History    :
// +---------+----------------------------------------------+---------
// |    Date |                 Comment                      | Who
// +---------+----------------------------------------------+---------
// | 29/10/98| Initial version                              | M.Frank
// | 20/2/99 | Automatic data preloading introduced.        | M.Frank
// | 25/4/13 | Locking for concurrent access introduced.    | D.Piparo
// +---------+----------------------------------------------+---------
//
//====================================================================
#define TSDATASVC_DATASVC_CPP

// Framework include files
#include "GaudiKernel/IConversionSvc.h"
#include "GaudiKernel/IConverter.h"
#include "GaudiKernel/IOpaqueAddress.h"

#include "GaudiKernel/DataObject.h"
#include "GaudiKernel/GaudiException.h"
#include "GaudiKernel/SvcFactory.h"
#include "GaudiKernel/xtoa.h"

#include "GaudiKernel/DataIncident.h"
#include "GaudiKernel/IIncidentSvc.h"
#include "GaudiKernel/RegistryEntry.h"
#include "GaudiKernel/TsDataSvc.h"

// Include files
#include <algorithm>
#include <cassert>
#include <cstdlib>
#include <mutex>
#include <sstream>
#include <vector>

// Macro to lock a scope
#define STD_LOCK_GUARD_MACRO std::lock_guard<tsDataSvcMutex> lock( m_accessMutex );

namespace
{
  std::string itemToPath( int item ) { return '/' + std::to_string( item ); }

  std::string operator+( char c, boost::string_ref sr )
  {
    std::string s{c};
    s.append( sr.data(), sr.size() );
    return s;
  }

  boost::string_ref::size_type find( boost::string_ref s, char c, size_t o )
  {
    if ( !s.empty() ) s.remove_prefix( o );
    auto r = s.find( c );
    return r == boost::string_ref::npos ? r : ( r + o );
  }

  std::string to_string( boost::string_ref sr ) { return {sr.data(), sr.size()}; }
}

// If you absolutely need optimization: switch off dynamic_cast.
// This improves access to the data store roughly by 10 %
// for balanced trees.
//
// M.Frank
#define CAST_REGENTRY( x, y ) dynamic_cast<x>( y )
//#define CAST_REGENTRY(x,y) (x)(y)
typedef DataSvcHelpers::RegistryEntry RegEntry;

#define ON_DEBUG if ( msgLevel( MSG::DEBUG ) )
#define ON_VERBOSE if ( msgLevel( MSG::VERBOSE ) )

#define DEBMSG ON_DEBUG debug()
#define VERMSG ON_VERBOSE verbose()

StatusCode TsDataSvc::clearSubTree( boost::string_ref sub_tree_path )
{
  DataObject* pObject = nullptr;
  StatusCode status   = findObject( sub_tree_path, pObject );
  if ( !status.isSuccess() ) return status;
  RegEntry* node_entry = CAST_REGENTRY( RegEntry*, pObject->registry() );
  if ( !node_entry ) return INVALID_OBJECT;
  RegEntry* parent = node_entry->parentEntry();
  if ( !parent ) return INVALID_PARENT;
  parent->remove( node_entry );
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::clearSubTree( DataObject* pObject )
{
  if ( !checkRoot() ) return INVALID_ROOT;
  RegEntry* entry = CAST_REGENTRY( RegEntry*, pObject->registry() );
  if ( !entry ) return INVALID_OBJECT;
  RegEntry* parent = entry->parentEntry();
  if ( !parent ) return INVALID_PARENT;
  parent->remove( entry );
  return SUCCESS;
}

StatusCode TsDataSvc::clearStore()
{
  if ( !checkRoot() ) return INVALID_ROOT;
  m_root->release();
  m_root = nullptr;
  return SUCCESS;
}

StatusCode TsDataSvc::traverseSubTree( boost::string_ref sub_tree_path, IDataStoreAgent* pAgent )
{
  STD_LOCK_GUARD_MACRO
  DataObject* pO    = nullptr;
  StatusCode status = findObject( sub_tree_path, pO );
  return status.isSuccess() ? traverseSubTree( pO, pAgent ) : status;
}

StatusCode TsDataSvc::traverseSubTree( DataObject* pObject, IDataStoreAgent* pAgent )
{
  STD_LOCK_GUARD_MACRO
  if ( !checkRoot() ) return INVALID_ROOT;
  RegEntry* entry = CAST_REGENTRY( RegEntry*, pObject->registry() );
  if ( !entry ) return INVALID_OBJECT;
  return entry->traverseTree( pAgent );
}

StatusCode TsDataSvc::traverseTree( IDataStoreAgent* pAgent )
{
  STD_LOCK_GUARD_MACRO
  if ( !checkRoot() ) return INVALID_ROOT;
  return m_root->traverseTree( pAgent );
}

StatusCode TsDataSvc::setRoot( std::string root_path, DataObject* pRootObj )
{
  clearStore().ignore();
  return i_setRoot( std::move( root_path ), pRootObj );
}

StatusCode TsDataSvc::i_setRoot( std::string root_path, DataObject* pRootObj )
{
  if ( pRootObj ) {
    m_root = new RegEntry( std::move( root_path ) );
    m_root->makeHard( pRootObj );
    m_root->setDataSvc( this );
    // No done with GaudiHive. preLoad().ignore();
  }
  return SUCCESS;
}

StatusCode TsDataSvc::setRoot( std::string root_path, IOpaqueAddress* pRootAddr )
{
  clearStore().ignore();
  return i_setRoot( std::move( root_path ), pRootAddr );
}

StatusCode TsDataSvc::i_setRoot( std::string root_path, IOpaqueAddress* pRootAddr )
{
  if ( pRootAddr ) {
    m_root = new RegEntry( std::move( root_path ) );
    m_root->makeHard( pRootAddr );
    m_root->setDataSvc( this );
    // Not done with GaudiHive. preLoad().ignore();
  }
  return SUCCESS;
}

StatusCode TsDataSvc::setDataLoader( IConversionSvc* pDataLoader, IDataProviderSvc* dpsvc )
{
  if ( pDataLoader ) pDataLoader->addRef();
  if ( m_dataLoader ) m_dataLoader->release();
  if ( pDataLoader ) {
    pDataLoader->setDataProvider( dpsvc == nullptr ? this : dpsvc ).ignore();
  }
  m_dataLoader = pDataLoader;
  return SUCCESS;
}

StatusCode TsDataSvc::objectParent( const DataObject* pObject, IRegistry*& refpParent )
{
  STD_LOCK_GUARD_MACRO
  if ( !pObject ) return INVALID_OBJECT;
  return objectParent( pObject->registry(), refpParent );
}
StatusCode TsDataSvc::objectParent( const IRegistry* pRegistry, IRegistry*& refpParent )
{
  STD_LOCK_GUARD_MACRO
  if ( !checkRoot() ) return INVALID_ROOT;
  const RegEntry* node_entry = CAST_REGENTRY( const RegEntry*, pRegistry );
  if ( !node_entry ) return INVALID_OBJECT;
  refpParent = node_entry->parent();
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::objectLeaves( const DataObject* pObject, std::vector<IRegistry*>& leaves )
{
  STD_LOCK_GUARD_MACRO
  if ( !pObject ) return INVALID_OBJECT;
  return objectLeaves( pObject->registry(), leaves );
}

StatusCode TsDataSvc::objectLeaves( const IRegistry* pRegistry, std::vector<IRegistry*>& leaves )
{
  STD_LOCK_GUARD_MACRO
  if ( !checkRoot() ) return INVALID_ROOT;
  const RegEntry* node_entry = CAST_REGENTRY( const RegEntry*, pRegistry );
  if ( !node_entry ) return INVALID_OBJECT;
  leaves.insert( leaves.end(), node_entry->leaves().begin(), node_entry->leaves().end() );
  // leaves = node_entry->leaves();
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::registerAddress( boost::string_ref fullPath, IOpaqueAddress* pAddress )
{
  if ( fullPath.empty() ) return INVALID_OBJ_PATH;
  if ( fullPath.front() != SEPARATOR ) {
    return registerAddress( m_root, fullPath, pAddress );
  }
  IRegistry* pRegistry = nullptr;
  return registerAddress( pRegistry, fullPath, pAddress );
}

StatusCode TsDataSvc::registerAddress( DataObject* parentObj, boost::string_ref objectPath, IOpaqueAddress* pAddress )
{
  IRegistry* pRegistry = ( parentObj ? parentObj->registry() : nullptr );
  return registerAddress( pRegistry, objectPath, pAddress );
}

StatusCode TsDataSvc::registerAddress( IRegistry* parentObj, boost::string_ref objPath, IOpaqueAddress* pAddress )
{
  if ( !checkRoot() ) return INVALID_ROOT;
  if ( objPath.empty() ) return INVALID_OBJ_PATH;
  if ( !parentObj ) {
    if ( objPath.front() != SEPARATOR ) {
      return registerAddress( m_root, objPath, pAddress );
    }
    auto sep = find( objPath, SEPARATOR, 1 );
    if ( sep == boost::string_ref::npos || objPath.substr( 0, sep ) != m_rootName.value() ) {
      return INVALID_PARENT;
    }
    return registerAddress( m_root, objPath.substr( sep ), pAddress );
  }
  if ( objPath.front() != SEPARATOR ) {
    return registerAddress( parentObj, SEPARATOR + objPath, pAddress );
  }
  RegEntry* par_entry = CAST_REGENTRY( RegEntry*, parentObj );
  if ( !par_entry ) return INVALID_PARENT;
  auto sep = objPath.rfind( SEPARATOR );
  if ( sep > 0 && sep != boost::string_ref::npos ) {
    auto p_path       = objPath.substr( 0, sep );
    auto o_path       = objPath.substr( sep );
    RegEntry* p_entry = par_entry->findLeaf( p_path );
    // Create default object leafs if the
    // intermediate nodes are not present
    if ( !p_entry && m_forceLeaves ) {
      DataObject* pLeaf = createDefaultObject();
      StatusCode sc     = registerObject( par_entry->identifier(), p_path, pLeaf );
      if ( !sc.isSuccess() ) delete pLeaf;
      p_entry = par_entry->findLeaf( p_path );
    }
    if ( !p_entry ) return INVALID_PARENT;
    return registerAddress( p_entry, o_path, pAddress );
  }
  StatusCode status = par_entry->add( to_string( objPath ), pAddress );
  return status.isSuccess() ? status : DOUBL_OBJ_PATH;
}

StatusCode TsDataSvc::unregisterAddress( boost::string_ref fullPath )
{
  if ( fullPath.empty() ) return INVALID_OBJ_PATH;
  IRegistry* pRegistry = nullptr;
  if ( fullPath.front() != SEPARATOR ) {
    return unregisterAddress( m_root, fullPath );
  }
  return unregisterAddress( pRegistry, fullPath );
}

StatusCode TsDataSvc::unregisterAddress( DataObject* pParent, boost::string_ref objPath )
{
  IRegistry* pRegistry = ( pParent ? pParent->registry() : nullptr );
  return unregisterAddress( pRegistry, objPath );
}

StatusCode TsDataSvc::unregisterAddress( IRegistry* pParent, boost::string_ref objPath )
{
  if ( !checkRoot() ) return INVALID_ROOT;

  if ( objPath.empty() ) return INVALID_OBJ_PATH;
  if ( !pParent ) {
    if ( objPath.front() != SEPARATOR ) {
      return unregisterAddress( m_root, objPath );
    }
    auto sep = find( objPath, SEPARATOR, 1 );
    if ( sep == boost::string_ref::npos || objPath.substr( 0, sep ) != m_rootName.value() ) {
      return INVALID_PARENT;
    }
    return unregisterAddress( m_root, objPath.substr( sep ) );
  }
  if ( objPath.front() != SEPARATOR ) {
    return unregisterAddress( pParent, SEPARATOR + objPath );
  }
  RegEntry* node_entry = CAST_REGENTRY( RegEntry*, pParent );
  if ( node_entry ) {
    RegEntry* leaf_entry = node_entry->findLeaf( objPath );
    if ( leaf_entry ) {
      auto sep = objPath.rfind( SEPARATOR );
      if ( sep > 0 && sep != boost::string_ref::npos ) {
        return unregisterAddress( leaf_entry->parent(), objPath.substr( sep ) );
      }
      StatusCode status = node_entry->remove( objPath );
      if ( status.isSuccess() ) return status;
    }
  }
  return INVALID_PARENT;
}

StatusCode TsDataSvc::registerObject( boost::string_ref fullPath, DataObject* pObject )
{
  return registerObject( nullptr, fullPath, pObject );
}

StatusCode TsDataSvc::registerObject( boost::string_ref parentPath, boost::string_ref objPath, DataObject* pObject )
{
  DataObject* pO    = nullptr;
  StatusCode status = retrieveObject( parentPath, pO );
  if ( !status.isSuccess() && m_forceLeaves ) {
    pO     = createDefaultObject();
    status = registerObject( parentPath, pO );
    if ( !status.isSuccess() ) pO->release();
  }
  return status.isSuccess() ? registerObject( pO, objPath, pObject ) : status;
}

StatusCode TsDataSvc::registerObject( boost::string_ref parentPath, int item, DataObject* pObject )
{
  return registerObject( parentPath, itemToPath( item ), pObject );
}

StatusCode TsDataSvc::registerObject( DataObject* parentObj, int item, DataObject* pObject )
{
  return registerObject( parentObj, itemToPath( item ), pObject );
}

StatusCode TsDataSvc::registerObject( DataObject* parentObj, boost::string_ref objPath, DataObject* pObject )
{
  if ( !checkRoot() ) return INVALID_ROOT;
  if ( !parentObj ) {
    if ( !objPath.empty() ) {
      if ( objPath.front() != SEPARATOR ) {
        return registerObject( m_rootName.value(), objPath, pObject );
      }
      auto sep = find( objPath, SEPARATOR, 1 );
      if ( sep != boost::string_ref::npos ) {
        return registerObject( objPath.substr( 0, sep ), objPath.substr( sep ), pObject );
      }
    }
    return INVALID_OBJ_PATH;
  }
  RegEntry* node_entry = CAST_REGENTRY( RegEntry*, parentObj->registry() );
  if ( node_entry ) {
    StatusCode status = INVALID_PARENT;
    auto sep          = find( objPath, SEPARATOR, 1 );
    if ( sep != boost::string_ref::npos ) {
      auto p_path         = objPath.substr( 0, sep );
      auto o_path         = objPath.substr( sep );
      RegEntry* par_entry = node_entry->findLeaf( p_path );
      // Create default object leafs if the
      // intermediate nodes are not present
      if ( !par_entry && m_forceLeaves ) {
        DataObject* pLeaf = createDefaultObject();
        StatusCode sc     = registerObject( parentObj, p_path, pLeaf );
        if ( !sc.isSuccess() ) delete pLeaf;
        par_entry = node_entry->findLeaf( p_path );
      } else if ( par_entry && !par_entry->object() ) {
        status = retrieveEntry( node_entry, p_path, par_entry );
        if ( !status.isSuccess() && !par_entry->address() && m_forceLeaves ) {
          DataObject* pLeaf = createDefaultObject();
          StatusCode sc     = registerObject( parentObj, p_path, pLeaf );
          if ( !sc.isSuccess() ) delete pLeaf;
          par_entry = node_entry->findLeaf( p_path );
        }
      }
      node_entry = par_entry;
      if ( node_entry ) {
        DataObject* obj   = node_entry->object();
        if ( obj ) status = registerObject( obj, o_path, pObject );
      }
    } else {
      RegEntry* leaf = node_entry->findLeaf( objPath );
      if ( !leaf ) {
        status = node_entry->add( to_string( objPath ), pObject );
      } else {
        DataObject* obj = leaf->object();
        if ( !obj ) {
          if ( !pObject ) {
            error() << "registerObject: trying to register null DataObject" << endmsg;
            return StatusCode::FAILURE;
          } else {
            pObject->setRegistry( leaf );
          }
          leaf->setAddress( nullptr );
          leaf->setObject( pObject );
          status = StatusCode::SUCCESS;
        } else {
          status = DOUBL_OBJ_PATH;
        }
      }
    }
    return status;
  }
  return INVALID_PARENT;
}

StatusCode TsDataSvc::unregisterObject( boost::string_ref fullPath )
{
  DataObject* pObject = nullptr;
  StatusCode status   = findObject( fullPath, pObject );
  if ( status.isFailure() ) return status;
  RegEntry* pEntry = CAST_REGENTRY( RegEntry*, pObject->registry() );
  if ( !pEntry ) return INVALID_ROOT;
  if ( !pEntry->isEmpty() ) return DIR_NOT_EMPTY;
  RegEntry* pParent = pEntry->parentEntry();
  if ( !pParent ) return INVALID_PARENT;
  if ( pObject ) pObject->addRef();
  pParent->remove( pEntry );
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::unregisterObject( boost::string_ref parentPath, boost::string_ref objPath )
{
  DataObject* pO    = nullptr;
  StatusCode status = findObject( parentPath, pO );
  return status.isSuccess() ? unregisterObject( pO, objPath ) : status;
}

StatusCode TsDataSvc::unregisterObject( boost::string_ref parentPath, int item )
{
  return unregisterObject( parentPath, itemToPath( item ) );
}

StatusCode TsDataSvc::unregisterObject( DataObject* pObject )
{
  if ( !checkRoot() ) return INVALID_ROOT;
  RegEntry* entry = m_root->findLeaf( pObject );
  if ( !entry ) return INVALID_OBJECT;
  RegEntry* parent = entry->parentEntry();
  if ( !parent ) return INVALID_PARENT;
  if ( !entry->isEmpty() ) return DIR_NOT_EMPTY;
  if ( entry->object() ) entry->object()->addRef();
  if ( parent ) parent->remove( entry );
  return SUCCESS;
}

StatusCode TsDataSvc::unregisterObject( DataObject* pParentObj, boost::string_ref objectPath )
{
  if ( !checkRoot() ) return INVALID_ROOT;
  try {
    RegEntry* parent = CAST_REGENTRY( RegEntry*, pParentObj->registry() );
    if ( parent ) {
      RegEntry* entry = parent->findLeaf( objectPath );
      if ( !entry ) return INVALID_OBJECT;
      if ( !entry->isEmpty() ) return DIR_NOT_EMPTY;
      if ( entry->object() ) entry->object()->addRef();
      parent->remove( entry );
      return SUCCESS;
    }
  } catch ( ... ) {
  }
  return INVALID_PARENT;
}

StatusCode TsDataSvc::unregisterObject( DataObject* pParentObj, int item )
{
  return unregisterObject( pParentObj, itemToPath( item ) );
}

DataObject* TsDataSvc::handleDataFault( IRegistry* pReg, boost::string_ref path )
{
  if ( m_enableFaultHdlr ) {
    IRegistry* pLeaf = nullptr;
    if ( pReg && path.empty() ) {
      DataIncident incident( name(), m_faultName, pReg->identifier() );
      m_incidentSvc->fireIncident( incident );
      return pReg->object();
    } else if ( pReg ) {
      std::string p = pReg->identifier();
      if ( path.front() != SEPARATOR ) p += SEPARATOR;
      p.append( path.data(), path.size() );
      DataIncident incident( name(), m_faultName, p );
      m_incidentSvc->fireIncident( incident );
      pLeaf = m_root->findLeaf( p );
    } else {
      std::string p = m_root->identifier();
      if ( path.front() != SEPARATOR ) p += SEPARATOR;
      p.append( path.data(), path.size() );
      DataIncident incident( name(), m_faultName, p );
      m_incidentSvc->fireIncident( incident );
      pLeaf = m_root->findLeaf( p );
    }
    if ( pLeaf ) return pLeaf->object();
  }
  return nullptr;
}

StatusCode TsDataSvc::loadObject( IRegistry* pRegistry )
{
  IConversionSvc* pLoader = getDataLoader( pRegistry );
  return loadObject( pLoader, pRegistry );
}

StatusCode TsDataSvc::loadObject( IConversionSvc* pLoader, IRegistry* pRegistry )
{
  STD_LOCK_GUARD_MACRO
  StatusCode status   = INVALID_OBJ_ADDR;
  DataObject* pObject = nullptr;
  if ( !pLoader ) { // Precondition: Data loader must be present
    if ( handleDataFault( pRegistry ) )
      return SUCCESS;
    else
      return NO_DATA_LOADER;
  }
  if ( !pRegistry ) { // Precondition: Directory must be valid
    if ( handleDataFault( pRegistry ) )
      return SUCCESS;
    else
      return INVALID_OBJ_ADDR;
  }

  VERMSG << "Requested object " << pRegistry->identifier() << endmsg;

  if ( m_enableAccessHdlr ) {
    // Fire data access incident
    DataIncident incident( name(), m_accessName, pRegistry->identifier() );
    m_incidentSvc->fireIncident( incident );
  }
  if ( !m_inhibitPathes.empty() ) {
    const auto& ident = pRegistry->identifier();
    auto inhibit      = std::find( m_inhibitPathes.begin(), m_inhibitPathes.end(), ident );
    if ( inhibit != m_inhibitPathes.end() ) return NO_ACCESS;
  }
  IOpaqueAddress* pAddress = pRegistry->address();
  if ( !pAddress ) {         // Precondition:
    return INVALID_OBJ_ADDR; // Address must be valid
  }
  try {
    status = pLoader->createObj( pAddress, pObject ); // Call data loader
    if ( status.isSuccess() ) {

      VERMSG << "Object " << pRegistry->identifier() << " created" << endmsg;

      RegEntry* pEntry = CAST_REGENTRY( RegEntry*, pRegistry );
      pEntry->setObject( pObject );

      VERMSG << "Filling object " << pRegistry->identifier() << endmsg;
      status = pLoader->fillObjRefs( pAddress, pObject );
    }
  } catch ( const GaudiException& exc ) {
    if ( handleDataFault( pRegistry ) ) return SUCCESS;
    throw GaudiException( "GaudiException in loadObject() " + pRegistry->identifier(), name(), StatusCode::FAILURE,
                          exc );
  } catch ( const std::exception& x ) {
    if ( handleDataFault( pRegistry ) ) return SUCCESS;
    throw GaudiException( "std::exception in loadObject() " + pRegistry->identifier() + ": " +
                              System::typeinfoName( typeid( x ) ) + ", " + x.what(),
                          name(), StatusCode::FAILURE );
  } catch ( ... ) {
    if ( handleDataFault( pRegistry ) ) return SUCCESS;
    throw GaudiException( "UNKN exception in loadObject() " + pRegistry->identifier(), name(), StatusCode::FAILURE );
  }
  if ( !status.isSuccess() ) {
    if ( handleDataFault( pRegistry ) ) return StatusCode::SUCCESS;
  }
  ON_VERBOSE if ( status.isSuccess() )
  {
    verbose() << "Object " << pRegistry->identifier() << " successfully loaded" << endmsg;
  }
  return status;
}

StatusCode TsDataSvc::retrieveEntry( RegEntry* parentObj, boost::string_ref path, RegEntry*& pEntry )
{
  STD_LOCK_GUARD_MACRO
  auto sep          = find( path, SEPARATOR, 1 );
  StatusCode status = StatusCode( INVALID_ROOT, true );
  pEntry            = nullptr;
  // A.Valassi 16.08.2001 avoid core dump if store is empty
  if ( checkRoot() ) {
    if ( !parentObj ) {
      if ( path.empty() || path == m_rootName ) return retrieveEntry( m_root, "", pEntry );
      if ( path.front() != SEPARATOR ) return retrieveEntry( m_root, path, pEntry );
      if ( sep == boost::string_ref::npos ) return INVALID_OBJ_PATH;
      if ( !m_root->object() ) {
        RegEntry* r = nullptr;
        status      = retrieveEntry( m_root, "", r );
        if ( !status.isSuccess() ) return status;
      }
      return retrieveEntry( m_root, path.substr( sep ), pEntry );
    }
    if ( sep != boost::string_ref::npos ) { // the string contains a separator (after pos 0)
      auto p_path = path.substr( 0, sep );
      auto o_path = path.substr( sep );
      if ( !parentObj->object() ) { // if the parent object has not been loaded yet, load it now
        status = loadObject( parentObj );
        if ( !status.isSuccess() ) return status;
      }
      RegEntry* root_entry = parentObj->findLeaf( p_path );
      if ( !root_entry && m_enableFaultHdlr ) {
        // If not even the parent is there, an incident
        // to load the parent must be fired...
        handleDataFault( parentObj, p_path );
        root_entry = parentObj->findLeaf( p_path );
      }
      if ( root_entry ) {
        DataObject* pO = root_entry->object();
        if ( !pO ) {
          // Object is not loaded: load the object if at all possible
          status = loadObject( root_entry );
          if ( !status.isSuccess() ) return status;
        }
        if ( root_entry->isSoft() ) {
          root_entry = CAST_REGENTRY( RegEntry*, pO->registry() );
        }
        return retrieveEntry( root_entry, o_path, pEntry );
      }
      return status;
    } else if ( path.empty() ) {
      pEntry = parentObj;
    } else {
      if ( !parentObj->object() ) { // if the parent object has not been loaded yet, load it now
        status = loadObject( parentObj );
        if ( !status.isSuccess() ) return status;
      }
      // last leave in search: find leaf and load
      pEntry = parentObj->findLeaf( path );
      // If no registry entry was found, trigger incident for action-on-demand
      if ( !pEntry && m_enableFaultHdlr ) {
        handleDataFault( parentObj, path );
        pEntry = ( path.empty() ? parentObj : parentObj->findLeaf( path ) );
      }
    }
    // Check results and return
    if ( !pEntry ) {
      status = INVALID_OBJ_PATH;
    } else if ( !pEntry->object() ) {
      status = loadObject( pEntry );
    } else if ( m_enableAccessHdlr ) {
      // Fire data access incident
      // I do not know if this is a good idea....
      // This fires too often!
      //
      // DataIncident incident(name(), m_accessName, pEntry->identifier());
      // m_incidentSvc->fireIncident(incident);
      status = SUCCESS;
    } else {
      status = SUCCESS;
    }
  }
  return status;
}

StatusCode TsDataSvc::retrieveObject( IRegistry* pRegistry, boost::string_ref path, DataObject*& pObject )
{
  STD_LOCK_GUARD_MACRO
  pObject          = nullptr;
  RegEntry *result = nullptr, *parent = CAST_REGENTRY( RegEntry *, pRegistry );
  StatusCode status                 = retrieveEntry( parent, path, result );
  if ( status.isSuccess() ) pObject = result->object();
  return status;
}

StatusCode TsDataSvc::retrieveObject( boost::string_ref fullPath, DataObject*& pObject )
{
  IRegistry* nullDir = nullptr;
  return retrieveObject( nullDir, fullPath, pObject );
}

StatusCode TsDataSvc::retrieveObject( boost::string_ref parentPath, boost::string_ref objectPath, DataObject*& pObject )
{
  DataObject* parent = nullptr;
  StatusCode status  = retrieveObject( parentPath, parent );
  return status.isSuccess() ? retrieveObject( parent, objectPath, pObject ) : status;
}

StatusCode TsDataSvc::retrieveObject( boost::string_ref parentPath, int item, DataObject*& pObject )
{
  return retrieveObject( parentPath, itemToPath( item ), pObject );
}

StatusCode TsDataSvc::retrieveObject( DataObject* parentObj, boost::string_ref path, DataObject*& pObject )
{
  IRegistry* pRegistry = ( parentObj ? parentObj->registry() : nullptr );
  return retrieveObject( pRegistry, path, pObject );
}

StatusCode TsDataSvc::retrieveObject( DataObject* parentObj, int item, DataObject*& pObject )
{
  return retrieveObject( parentObj, itemToPath( item ), pObject );
}

StatusCode TsDataSvc::findObject( IRegistry* pRegistry, boost::string_ref path, DataObject*& pObject )
{
  STD_LOCK_GUARD_MACRO
  pObject              = nullptr;
  IRegistry* pReg      = ( pRegistry ? pRegistry : m_root );
  RegEntry* root_entry = CAST_REGENTRY( RegEntry*, pReg );
  if ( root_entry ) {
    if ( !path.empty() ) pReg = root_entry->find( path );
    if ( !pReg ) return INVALID_OBJ_PATH;
    pObject = pReg->object();
  }
  return pObject ? IDataProviderSvc_NO_ERROR : OBJ_NOT_LOADED;
}

StatusCode TsDataSvc::findObject( boost::string_ref path, DataObject*& pObject )
{
  STD_LOCK_GUARD_MACRO
  pObject = nullptr;
  if ( !checkRoot() ) return INVALID_ROOT;
  if ( path.empty() || path == m_rootName ) {
    pObject = m_root->object();
    return !pObject ? OBJ_NOT_LOADED : IDataProviderSvc_NO_ERROR;
  }
  if ( path.front() != SEPARATOR ) {
    return findObject( m_rootName.value(), path, pObject );
  }
  return findObject( static_cast<IRegistry*>( nullptr ), path, pObject );
}

StatusCode TsDataSvc::findObject( boost::string_ref parentPath, boost::string_ref objectPath, DataObject*& pObject )
{
  DataObject* parent = nullptr;
  StatusCode status  = findObject( parentPath, parent );
  return status.isSuccess() ? findObject( parent, objectPath, pObject ) : status;
}

StatusCode TsDataSvc::findObject( boost::string_ref parentPath, int item, DataObject*& pObject )
{
  return findObject( parentPath, itemToPath( item ), pObject );
}

StatusCode TsDataSvc::findObject( DataObject* parentObj, int item, DataObject*& pObject )
{
  return findObject( parentObj, itemToPath( item ), pObject );
}

StatusCode TsDataSvc::findObject( DataObject* parentObj, boost::string_ref path, DataObject*& pObject )
{
  return findObject( parentObj ? parentObj->registry() : nullptr, path, pObject );
}

StatusCode TsDataSvc::updateObject( boost::string_ref updatePath )
{
  DataObject* pO    = nullptr;
  StatusCode status = findObject( updatePath, pO );
  return status.isSuccess() ? updateObject( pO ) : retrieveObject( updatePath, pO );
}

StatusCode TsDataSvc::updateObject( IRegistry* pRegistry )
{
  if ( !pRegistry ) {        // Precondition:
    return INVALID_OBJ_ADDR; // Addres must be valid
  }
  DataObject* toUpdate = pRegistry->object();
  if ( !toUpdate ) { // Try first to load
    return loadObject( pRegistry );
  }
  return updateObject( toUpdate );
}

StatusCode TsDataSvc::updateObject( DataObject* toUpdate )
{
  STD_LOCK_GUARD_MACRO
  StatusCode status = INVALID_OBJ_ADDR;
  if ( !toUpdate ) {       // Precondition:
    return INVALID_OBJECT; // Address must be valid
  }
  IRegistry* pRegistry = toUpdate->registry(); // Precondition:
  if ( !pRegistry ) {                          // Need valid registry
    return INVALID_OBJECT;
  }
  IOpaqueAddress* pAddress = pRegistry->address(); // Precondition:
  if ( !pAddress ) {                               // Need valid address
    return INVALID_OBJ_ADDR;
  }
  IConversionSvc* pLoader = getDataLoader( pRegistry );
  if ( !pLoader ) {        // Precondition:
    return NO_DATA_LOADER; // Data loader must be present
  }
  if ( !m_inhibitPathes.empty() ) {
    auto& ident  = pRegistry->identifier();
    auto inhibit = std::find( m_inhibitPathes.begin(), m_inhibitPathes.end(), ident );
    if ( inhibit != m_inhibitPathes.end() ) return NO_ACCESS;
  }
  try {
    status = pLoader->updateObj( pAddress, toUpdate ); // Call data loader
    if ( status.isSuccess() ) {
      status = pLoader->updateObjRefs( pAddress, toUpdate );
    }
  } catch ( const GaudiException& exc ) {
    throw GaudiException( "GaudiException in updateObject() " + pRegistry->name(), name(), StatusCode::FAILURE, exc );
  } catch ( const std::exception& x ) {
    throw GaudiException( "std::exception in updateObject() " + pRegistry->name() + ": " +
                              System::typeinfoName( typeid( x ) ) + ", " + x.what(),
                          name(), StatusCode::FAILURE );
  } catch ( ... ) {
    throw GaudiException( "UNKN exception in updateObject() " + pRegistry->name(), name(), StatusCode::FAILURE );
  }
  return status;
}

StatusCode TsDataSvc::updateObject( boost::string_ref parentPath, boost::string_ref updatePath )
{
  DataObject* pParent = nullptr;
  StatusCode status   = findObject( parentPath, pParent );
  return status.isSuccess() ? updateObject( pParent, updatePath ) : status;
}

StatusCode TsDataSvc::updateObject( DataObject* parent, boost::string_ref updatePath )
{
  DataObject* pObject = nullptr;
  StatusCode status   = findObject( parent, updatePath, pObject );
  return status.isSuccess() ? updateObject( pObject ) : status;
}

// Link object
StatusCode TsDataSvc::linkObject( IRegistry* from, boost::string_ref objPath, DataObject* to )
{
  STD_LOCK_GUARD_MACRO
  if ( !checkRoot() ) return INVALID_ROOT;
  try {
    RegEntry* from_entry = CAST_REGENTRY( RegEntry*, from );
    if ( from_entry ) {
      // First check if both objects are already registered to the store
      RegEntry* to_entry = m_root->findLeaf( to );
      if ( !to_entry ) return INVALID_OBJECT;
      auto sep = objPath.rfind( SEPARATOR );
      if ( sep > 0 && sep != boost::string_ref::npos ) { // in case the objPath is a sub-directory itself
        DataObject* pO = nullptr;
        StatusCode sc  = retrieveObject( from, objPath.substr( 0, sep ), pO );
        return sc.isSuccess() ? linkObject( pO->registry(), objPath.substr( sep ), to ) : sc;
      }
      // Now register the soft link
      StatusCode status = from_entry->add( to_string( objPath ), to, true );
      return status.isSuccess() ? IDataProviderSvc_NO_ERROR : DOUBL_OBJ_PATH;
    }
  } catch ( ... ) {
  }
  return INVALID_PARENT;
}

StatusCode TsDataSvc::linkObject( boost::string_ref fullPath, DataObject* to )
{
  STD_LOCK_GUARD_MACRO
  if ( fullPath.empty() ) return INVALID_OBJ_PATH;
  if ( fullPath.front() != SEPARATOR ) return linkObject( m_rootName.value(), fullPath, to );
  auto sep = fullPath.rfind( SEPARATOR );
  return linkObject( fullPath.substr( 0, sep ), fullPath.substr( sep ), to );
}

StatusCode TsDataSvc::linkObject( boost::string_ref from, boost::string_ref objPath, DataObject* to )
{
  STD_LOCK_GUARD_MACRO
  DataObject* pO    = nullptr;
  StatusCode status = retrieveObject( from, pO );
  return status.isSuccess() ? linkObject( pO->registry(), objPath, to ) : status;
}

StatusCode TsDataSvc::linkObject( DataObject* from, boost::string_ref objPath, DataObject* to )
{
  STD_LOCK_GUARD_MACRO
  if ( from ) {
    IRegistry* from_entry = from->registry();
    if ( from_entry ) return linkObject( from_entry, objPath, to );
  }
  return INVALID_PARENT;
}

StatusCode TsDataSvc::unlinkObject( IRegistry* from, boost::string_ref objPath )
{
  STD_LOCK_GUARD_MACRO
  if ( !checkRoot() ) return INVALID_ROOT;
  try {
    RegEntry* from_entry = CAST_REGENTRY( RegEntry*, from );
    if ( from_entry ) {
      auto sep = objPath.rfind( SEPARATOR );
      if ( sep > 0 && sep != boost::string_ref::npos ) { // in case the objPath is a sub-directory itself
        DataObject* pO = nullptr;
        StatusCode sc  = findObject( from, objPath.substr( 0, sep ), pO );
        return sc.isSuccess() ? unlinkObject( pO->registry(), objPath.substr( sep ) ) : sc;
      }
      StatusCode status = from_entry->remove( objPath );
      return status.isSuccess() ? status : INVALID_OBJ_PATH;
    }
  } catch ( ... ) {
  }
  return INVALID_PARENT;
}

StatusCode TsDataSvc::unlinkObject( boost::string_ref fullPath )
{
  if ( fullPath.empty() ) return INVALID_OBJ_PATH;
  if ( fullPath.front() != SEPARATOR ) {
    return unlinkObject( m_rootName.value(), fullPath );
  }
  auto sep = fullPath.rfind( SEPARATOR );
  return unlinkObject( fullPath.substr( 0, sep ), fullPath.substr( sep ) );
}

StatusCode TsDataSvc::unlinkObject( boost::string_ref from, boost::string_ref objPath )
{
  DataObject* pObject = nullptr;
  StatusCode status   = findObject( from, pObject );
  return status.isSuccess() ? unlinkObject( pObject->registry(), objPath ) : status;
}

StatusCode TsDataSvc::unlinkObject( DataObject* from, boost::string_ref objPath )
{
  if ( !checkRoot() ) return INVALID_ROOT;
  return unlinkObject( m_root->findLeaf( from ), objPath );
}

StatusCode TsDataSvc::addPreLoadItem( const DataStoreItem& item )
{
  auto i = std::find( m_preLoads.begin(), m_preLoads.end(), item );
  if ( i == m_preLoads.end() ) m_preLoads.push_back( item );
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::addPreLoadItem( std::string itemPath )
{
  return addPreLoadItem( DataStoreItem( std::move( itemPath ), 1 ) );
}

StatusCode TsDataSvc::removePreLoadItem( const DataStoreItem& item )
{
  auto i = std::remove( m_preLoads.begin(), m_preLoads.end(), item );
  m_preLoads.erase( i, m_preLoads.end() );
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::removePreLoadItem( std::string itemPath )
{
  return removePreLoadItem( DataStoreItem( std::move( itemPath ), 1 ) );
}

StatusCode TsDataSvc::resetPreLoad()
{
  m_preLoads.clear();
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::preLoad( int depth, int load_depth, DataObject* pObject )
{
  // unused:  StatusCode sc = StatusCode::FAILURE;
  if ( pObject && depth++ < load_depth ) {
    RegEntry* dir = CAST_REGENTRY( RegEntry*, pObject->registry() );
    if ( dir ) {
      for ( const auto& i : *dir ) {
        DataObject* pObj  = nullptr;
        StatusCode status = retrieveObject( pObject, i->name(), pObj );
        if ( status.isSuccess() && depth < load_depth ) {
          preLoad( depth, load_depth, pObj ).ignore();
        }
      }
    }
  }
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::preLoad()
{
  DataObject* pObj = nullptr;
  for ( const auto& i : m_preLoads ) {
    StatusCode sc  = retrieveObject( i.path(), pObj );
    int load_depth = i.depth();
    if ( sc.isSuccess() && load_depth > 1 ) {
      preLoad( 1, load_depth, pObj ).ignore();
    }
  }
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::initialize()
{
  // Nothing to do: just call base class initialisation
  StatusCode sc = Service::initialize();
  if ( !sc.isSuccess() ) return sc;
  sc = service( "IncidentSvc", m_incidentSvc, true );
  if ( UNLIKELY( !sc.isSuccess() ) ) {
    error() << "Failed to access incident service." << endmsg;
  }
  return sc;
}

StatusCode TsDataSvc::reinitialize()
{
  StatusCode sc;
  // the finalize part is copied here
  setDataLoader( nullptr ).ignore();
  resetPreLoad().ignore();
  clearStore().ignore();
  if ( m_incidentSvc ) {
    m_incidentSvc->release();
    m_incidentSvc = nullptr;
  }
  // re-initialize the base class
  sc = Service::reinitialize();
  if ( UNLIKELY( !sc.isSuccess() ) ) {
    error() << "Unable to reinitialize base class" << endmsg;
    return sc;
  }
  // the initialize part is copied here
  sc = service( "IncidentSvc", m_incidentSvc, true );
  if ( UNLIKELY( !sc.isSuccess() ) ) {
    error() << "Failed to access incident service." << endmsg;
    return sc;
  }
  // return
  return StatusCode::SUCCESS;
}

StatusCode TsDataSvc::finalize()
{
  // Nothing to do: just call base class initialisation
  setDataLoader( nullptr ).ignore();
  resetPreLoad().ignore();
  clearStore().ignore();
  if ( m_incidentSvc ) {
    m_incidentSvc->release();
    m_incidentSvc = nullptr;
  }
  return Service::finalize();
}

CLID TsDataSvc::rootCLID() const { return ( (CLID)m_rootCLID ); }

const std::string& TsDataSvc::rootName() const { return ( m_rootName ); }

DataObject* TsDataSvc::createDefaultObject() const { return new DataObject(); }

IConversionSvc* TsDataSvc::getDataLoader( IRegistry* /* pReg */ ) { return m_dataLoader; }

TsDataSvc::~TsDataSvc()
{
  setDataLoader( nullptr ).ignore();
  resetPreLoad().ignore();
  clearStore().ignore();
}