PoolSvc.cxx

Go to the documentation of this file.

/*
  Copyright (C) 2002-2026 CERN for the benefit of the ATLAS collaboration
*/

 
#include "PoolSvc.h"
#include "MicroSessionManager.h"
 
#include "GaudiKernel/IIoComponentMgr.h"
#include "GaudiKernel/ConcurrencyFlags.h"
#include "GaudiKernel/Bootstrap.h"
#include "GaudiKernel/ISvcLocator.h"
#include "GaudiKernel/IMessageSvc.h"
 
#include "PathResolver/PathResolver.h"
#include "AthenaBaseComps/AthAlgTool.h"
 
#include "PersistentDataModel/Placement.h"
#include "PersistentDataModel/Token.h"
 
#include "PoolSvc/ISessionTool.h"
#include "PoolSvc/IDatabase.h"
#include "PoolSvc/IContainer.h"
#include "PoolSvc/ITokenIterator.h"
#include "PoolSvc/FileCatalogUtils.h"
#include "SessionTool.h"
 
#include "StorageSvc/DbType.h"
#include "StorageSvc/DbPrint.h"
 
#include <cstdlib>
#include <cstring>
#include <algorithm>
#include <cstdio>
#include <cctype>
#include <exception>  // for runtime_error
 
bool isNumber(const std::string& s) {
   return !s.empty() && (std::isdigit(s[0]) || s[0] == '+' || s[0] == '-');
}
 
//__________________________________________________________________________
StatusCode PoolSvc::initialize() {
   ATH_CHECK(::AthService::initialize());
 
   // Register this service for 'I/O' events
   ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
   ATH_CHECK(iomgr.retrieve());
   ATH_CHECK(iomgr->io_register(this));
   // Register input file's names with the I/O manager, unless in SharedWrite mode, set by AthenaPoolCnvSvc
   bool allGood = true;
   for (const auto& catalog : m_readCatalog.value()) {
      if (catalog.starts_with("xmlcatalog_file:")) {
         const std::string fileName = catalog.substr(16);
         if (!iomgr->io_register(this, IIoComponentMgr::IoMode::READ, fileName, fileName).isSuccess()) {
            ATH_MSG_FATAL("could not register [" << catalog << "] for input !");
            allGood = false;
         } else {
            ATH_MSG_INFO("io_register[" << this->name() << "](" << catalog << ") [ok]");
         }
      }
   }
   if (m_writeCatalog.value().starts_with("xmlcatalog_file:")) {
      const std::string fileName = m_writeCatalog.value().substr(16);
      if (!iomgr->io_register(this, IIoComponentMgr::IoMode::WRITE, fileName, fileName).isSuccess()) {
         ATH_MSG_FATAL("could not register [" << m_writeCatalog.value() << "] for input !");
         allGood = false;
      } else {
         ATH_MSG_INFO("io_register[" << this->name() << "](" << m_writeCatalog.value() << ") [ok]");
      }
   }
   if (!allGood) {
      return(StatusCode::FAILURE);
   }
 
   MSG::Level athLvl = msg().level();
   ATH_MSG_DEBUG("OutputLevel is " << (int)athLvl);
   pool::DbPrintLvl::setLevel(athLvl);
   return(setupPersistencySvc());
}
 
//__________________________________________________________________________
StatusCode PoolSvc::io_reinit() {
   ATH_MSG_INFO("I/O reinitialization...");
   ServiceHandle<IIoComponentMgr> iomgr("IoComponentMgr", name());
   if (!iomgr.retrieve().isSuccess()) {
      ATH_MSG_FATAL("Could not retrieve IoComponentMgr !");
      return(StatusCode::FAILURE);
   }
   if (!iomgr->io_hasitem(this)) {
      ATH_MSG_FATAL("IoComponentMgr does not know about myself !");
      return(StatusCode::FAILURE);
   }
   std::vector<std::string> readcat = m_readCatalog.value();
   for (std::size_t icat = 0, imax = readcat.size(); icat < imax; icat++) {
      if (readcat[icat].starts_with("xmlcatalog_file:")) {
         std::string fileName = readcat[icat].substr(16);
         if (iomgr->io_contains(this, fileName)) {
            if (!iomgr->io_retrieve(this, fileName).isSuccess()) {
               ATH_MSG_FATAL("Could not retrieve new value for [" << fileName << "] !");
               return(StatusCode::FAILURE);
            }
            readcat[icat] = "xmlcatalog_file:" + fileName;
         }
      }
   }
   // all good... copy over.
   m_readCatalog = readcat;
   if (m_writeCatalog.value().starts_with("xmlcatalog_file:")) {
      std::string fileName = m_writeCatalog.value().substr(16);
      if (iomgr->io_contains(this, fileName)) {
         if (!iomgr->io_retrieve(this, fileName).isSuccess()) {
            ATH_MSG_FATAL("Could not retrieve new value for [" << fileName << "] !");
            return(StatusCode::FAILURE);
         }
         if (!m_shareCat) {
            m_writeCatalog.setValue("xmlcatalog_file:" + fileName);
         }
      }
   }
   return(setupPersistencySvc());
}
//__________________________________________________________________________
StatusCode PoolSvc::setupPersistencySvc() {
   clearState();
   ATH_MSG_INFO("Setting up FileCatalog and Streams");
   if (!createCatalog()) {
      ATH_MSG_FATAL("Failed to setup POOL File Catalog.");
      return(StatusCode::FAILURE);
   }
   m_catalog->init();
   // Setup a persistency services
   ATH_CHECK( createSession("ReadSession") ); // Add Read Session
   m_pers_mut.push_back(new CallMutex);
   const bool& atttibuteValue = true;
   if (!m_dbSessionVec[IPoolSvc::kInputStream]->microSessionManager(pool::ROOT_StorageType.type()).setAttributeOfType("ENABLE_THREADSAFETY", static_cast<const void*>(&atttibuteValue), typeid(bool), "")) {
      ATH_MSG_FATAL("Failed to enable thread safety in ROOT via PersistencySvc.");
      return(StatusCode::FAILURE);
   }
   m_contextMaxFile.try_emplace(IPoolSvc::kInputStream, m_dbAgeLimit.value());
   if (!connect(Io::READ, IPoolSvc::kInputStream).isSuccess()) {
      ATH_MSG_FATAL("Failed to connect Input PersistencySvc.");
      return(StatusCode::FAILURE);
   }
   ATH_CHECK( createSession("WriteSession") ); // Add Write Session
   m_pers_mut.push_back(new CallMutex);
 
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode PoolSvc::start() {
   // Switiching on ROOT implicit multi threading for AthenaMT
   if (m_useROOTIMT && Gaudi::Concurrency::ConcurrencyFlags::numThreads() > 1) {
      const int& atttibuteValue = Gaudi::Concurrency::ConcurrencyFlags::numThreads() - 1;
      if (!m_dbSessionVec[IPoolSvc::kInputStream]->microSessionManager(pool::ROOT_StorageType.type()).setAttributeOfType("ENABLE_IMPLICITMT", static_cast<const void*>(&atttibuteValue), typeid(int), "")) {
         ATH_MSG_FATAL("Failed to enable implicit multithreading in ROOT via PersistencySvc.");
         return(StatusCode::FAILURE);
      }
      ATH_MSG_INFO("Enabled implicit multithreading in ROOT via PersistencySvc to: " << Gaudi::Concurrency::ConcurrencyFlags::numThreads() - 1);
   }
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode PoolSvc::stop() {
   ATH_MSG_VERBOSE("stop()");
   bool retError = false;
   for (unsigned int contextId = 0, imax = m_dbSessionVec.size(); contextId < imax; contextId++) {
      if (!disconnect(contextId).isSuccess()) {
         ATH_MSG_FATAL("Cannot disconnect Stream: " << contextId);
         retError = true;
      }
   }
   return(retError ? StatusCode::FAILURE : StatusCode::SUCCESS);
}
 
//__________________________________________________________________________
void PoolSvc::clearState() {
   std::lock_guard<CallMutex> lock(m_pool_mut);
   // Cleanup persistency service
   // Tools are kept forever, hopefully closing sessions will be enough
   for( auto& session : m_dbSessionVec ) {
      session->closeSession();
   }
   m_dbSessionVec.clear();
   for (const auto& persistencyMutex : m_pers_mut) {
      delete persistencyMutex;
   }
   m_mainOutputLabel.clear();
   m_inputContextLabel.clear();
   m_outputContextLabel.clear();
   m_pers_mut.clear();
   if (m_catalog.isValid()) {
      m_catalog->commit();
      m_catalog = SmartIF<Gaudi::IFileCatalog>();
      m_catalogMgr = SmartIF<Gaudi::IFileCatalogMgr>();
   }
}
//__________________________________________________________________________
StatusCode PoolSvc::finalize() {
   clearState();
   return(::AthService::finalize());
}
//__________________________________________________________________________
StatusCode PoolSvc::io_finalize() {
   ATH_MSG_INFO("I/O finalization...");
   for (size_t i = 0; i < m_dbSessionVec.size(); i++) {
      if ((m_dbSessionVec[i]->openMode() == Io::WRITE || m_dbSessionVec[i]->openMode() == Io::APPEND) &&
          !disconnect(i).isSuccess()) {
         ATH_MSG_WARNING("Cannot disconnect output Stream " << i);
      }
   }
   clearState();
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
Token* PoolSvc::registerForWrite(const Placement* placement,
                                 const void* obj,
                                 const RootType& classDesc) {
   unsigned int contextId = IPoolSvc::kOutputStream;
   const std::string& auxString = placement->auxString();
   if (!auxString.empty()) {
      if (auxString.starts_with("[CTXT=")) {
         ::sscanf(auxString.c_str(), "[CTXT=%08X]", &contextId);
      } else if (auxString.starts_with("[CLABEL=")) {
         contextId = this->getOutputContext(auxString);
      }
      if (contextId >= m_dbSessionVec.size()) {
         ATH_MSG_WARNING("registerForWrite: Using default output Stream instead of id = " << contextId);
         contextId = IPoolSvc::kOutputStream;
      }
   }
   std::lock_guard<CallMutex> lock(*m_pers_mut[contextId]);
   Token* token = m_dbSessionVec[contextId]->registerForWrite(*placement, obj, classDesc);
   if (token == nullptr) {
      ATH_MSG_WARNING("Cannot write object: " << placement->containerName());
   }
   return(token);
}
//__________________________________________________________________________
void PoolSvc::setObjPtr(void*& obj, const Token* token) {
   unsigned int contextId = IPoolSvc::kInputStream;
   const std::string& auxString = token->auxString();
   bool allowAging = true;
   if (!auxString.empty()) {
      if (auxString.starts_with("[CTXT=")) {
         ::sscanf(auxString.c_str(), "[CTXT=%08X]", &contextId);
         if (contextId == m_dbSessionVec.size()) {
            ATH_MSG_DEBUG("setObjPtr: On-demand creating new input Stream id = " << contextId);
            contextId = this->getInputContext("");
         }
      } else if (auxString.starts_with("[CLABEL=")) {
         contextId = this->getInputContext(auxString);
      }
      if (contextId >= m_dbSessionVec.size()) {
         ATH_MSG_WARNING("setObjPtr: Using default input Stream instead of id = " << contextId);
         contextId = IPoolSvc::kInputStream;
         allowAging = false;
      }
   }
   ATH_MSG_VERBOSE("setObjPtr: token=" << token->toString() << ", auxString=" << auxString << ", contextID=" << contextId);
   // Get Context ID/label from Token
   std::lock_guard<CallMutex> lock(*m_pers_mut[contextId]);
   obj = m_dbSessionVec[contextId]->readObject(*token, obj);
   std::map<unsigned int, unsigned int>::const_iterator maxFileIter = m_contextMaxFile.find(contextId);
   if (allowAging && maxFileIter != m_contextMaxFile.end() && maxFileIter->second > 0) {
      m_guidLists[contextId].remove(token->dbID());
      m_guidLists[contextId].push_back(token->dbID());
      while (m_guidLists[contextId].size() > maxFileIter->second) {
         this->disconnectDb("FID:" + m_guidLists[contextId].begin()->toString(), contextId).ignore();
      }
   }
}
//__________________________________________________________________________
unsigned int PoolSvc::getOutputContext(const std::string& label) {
   std::lock_guard<CallMutex> lock(m_pool_mut);
   if (m_mainOutputLabel.empty()) {
      m_mainOutputLabel = label;
      m_outputContextLabel.try_emplace(label, IPoolSvc::kOutputStream);
   }
   if (label == m_mainOutputLabel || label.empty()) {
      return(IPoolSvc::kOutputStream);
   }
   std::map<std::string, unsigned int>::const_iterator contextIter = m_outputContextLabel.find(label);
   if (contextIter != m_outputContextLabel.end()) {
      return(contextIter->second);
   }
   const unsigned int id = m_dbSessionVec.size();
   if( !createSession(label).isSuccess() ) {
      ATH_MSG_WARNING("getOutputContext: Failed to create new output Stream id = " << id << ". Using default output Stream instead.");
      return(IPoolSvc::kOutputStream);
   }
   m_pers_mut.push_back(new CallMutex);
   m_outputContextLabel.try_emplace(label, id);
   return(id);
}
//__________________________________________________________________________
unsigned int PoolSvc::getInputContext(const std::string& label, unsigned int maxFile) {
   std::lock_guard<CallMutex> lock(m_pool_mut);
   if (!label.empty()) {
      std::map<std::string, unsigned int>::const_iterator contextIter = m_inputContextLabel.find(label);
      if (contextIter != m_inputContextLabel.end()) {
         if (maxFile > 0) {
            m_contextMaxFile[contextIter->second] = maxFile;
         }
         return(contextIter->second);
      }
   }
   const unsigned int id = m_dbSessionVec.size();
   if( !createSession(label, maxFile).isSuccess() ) {
      ATH_MSG_WARNING("getInputContext: Failed to create new input Stream id = " << id << ". Using default input Stream instead.");
      return(IPoolSvc::kInputStream);
   }
   m_pers_mut.push_back(new CallMutex);
   if (!connect(Io::READ, id).isSuccess()) {
      ATH_MSG_WARNING("Failed to connect Input PersistencySvc: " << id);
      return(IPoolSvc::kInputStream);
   }
   if (!label.empty()) {
      m_inputContextLabel.try_emplace(label, id);
   }
   m_contextMaxFile.try_emplace(id, maxFile);
   return(id);
}
//__________________________________________________________________________
std::map<std::string, unsigned int> PoolSvc::getInputContextMap() const {
   std::lock_guard<CallMutex> lock(m_pool_mut);
   return(m_inputContextLabel);
}
//__________________________________________________________________________
unsigned int PoolSvc::getInputContextMapSize() const {
   std::lock_guard<CallMutex> lock(m_pool_mut);
   return(m_inputContextLabel.size());
}
//__________________________________________________________________________
pool::FileDescriptor PoolSvc::getInputFileDescriptor(unsigned int contextId,
                                                     const std::string& connection) {
   if (contextId == m_dbSessionVec.size()) {
      ATH_MSG_DEBUG("getInputFileDescriptor: On-demand creating new input Stream id = " << contextId);
      contextId = this->getInputContext("");
   }
   if (contextId >= m_dbSessionVec.size()) {
      ATH_MSG_WARNING("getInputFileDescriptor: Using default input Stream instead of id = " << contextId);
      contextId = IPoolSvc::kInputStream;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(contextId, connection);
   if (!dbH) {
      throw std::runtime_error("Could not retrieve a database handle to '" + connection + "'");
   }
   pool::FileDescriptor* fd = dbH->fileDescriptor();
   if (!fd) {
      throw std::runtime_error("Could not retrieve connection info from DB '" + connection + "'");
   }
   return *fd;
}
//__________________________________________________________________________
void PoolSvc::setShareMode(bool shareCat) {
   m_shareCat = shareCat;
}
//__________________________________________________________________________
void PoolSvc::startCatalog() {
   if (m_catalog.isValid()) {
      m_catalog->init();
   }
}
//__________________________________________________________________________
void PoolSvc::commitCatalog() {
   if (m_catalog.isValid()) {
      m_catalog->commit();
   }
}
//__________________________________________________________________________
void PoolSvc::lookupBestPfn(const std::string& dbID, std::string& pfn, std::string& type) const {
   FileCatalogUtils::getFirstPFN(*m_catalog, dbID, pfn, type); // FID -> best PFN
}
//__________________________________________________________________________
StatusCode PoolSvc::connectCollection(const std::string& connection,
        const std::string& collectionName,
        unsigned int contextId)
{
   ATH_MSG_DEBUG("connectCollection() connection=" << connection
                 << ", name=" << collectionName << ", contextID=" << contextId);
   if (contextId >= m_dbSessionVec.size()) {
      ATH_MSG_WARNING("connectCollection: Using default input Stream instead of id = " << contextId);
      contextId = IPoolSvc::kInputStream;
   }
 
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   // Check POOL FileCatalog entry.
   bool insertFile = false;
   if (connection.starts_with("PFN:")) {
      std::string fid, fileType;
      FileCatalogUtils::lookupFileByPFN(*m_catalog, connection.substr(4), fid, fileType);
      if (fid.empty()) { // No entry in file catalog
         insertFile = true;
         ATH_MSG_INFO("File is not in Catalog! Attempt to open it anyway.");
      }
   }
 
   // Check whether Collection Container exists.
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(contextId, connection);
   if( !dbH ) {
      ATH_MSG_INFO("connectCollection(): Failed to get Database Handle for: " << connection);
      return StatusCode::FAILURE;
   }
   try {
      if (dbH->openMode() == Io::INVALID) {
         dbH->connectForRead();
      }
      std::map<unsigned int, unsigned int>::const_iterator maxFileIter = m_contextMaxFile.find(contextId);
      if (maxFileIter != m_contextMaxFile.end() && maxFileIter->second > 0 && !dbH->fid().empty()) {
         const Guid guid(dbH->fid());
         m_guidLists[contextId].remove(guid);
         m_guidLists[contextId].push_back(guid);
         while (m_guidLists[contextId].size() > maxFileIter->second + 1) {
            this->disconnectDb("FID:" + m_guidLists[contextId].begin()->toString(), contextId).ignore();
         }
      }
   } catch (std::exception& e) {
      ATH_MSG_INFO("connectCollection() failed to open '" << connection << "' - trying.");
   }
   // For multithreaded processing (with multiple events in flight),
   // increase virtual tree size to accomodate back reads
   if (m_useROOTMaxTree && contextId == IPoolSvc::kInputStream && Gaudi::Concurrency::ConcurrencyFlags::numConcurrentEvents() > 1) {
      ATH_MSG_DEBUG("connectCollection(): Increasing virtual TTree size for: " << connection);
      if (!this->setAttribute("TREE_MAX_VIRTUAL_SIZE", "-1", pool::ROOT_StorageType.type(), connection.substr(4), "CollectionTree", contextId).isSuccess()) {
         ATH_MSG_DEBUG("Failed to increase maximum virtual TTree size.");
      }
   }
   if (insertFile) return(StatusCode::RECOVERABLE);
 
   return(StatusCode::SUCCESS);
}
//__________________________________________________________________________
StatusCode PoolSvc::checkCollection(const std::string& connection,
        unsigned int contextId,
                bool noContainer) {
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(contextId, connection);
   if (dbH != nullptr && !dbH->fid().empty()) {
      if (noContainer) {
         return(StatusCode::SUCCESS); // no events
      }
      if (m_attemptCatalogPatch.value()) {
         patchCatalog(connection.substr(4), *dbH);
      }
      return(StatusCode::SUCCESS);
   }
   return(StatusCode::FAILURE);
}
//__________________________________________________________________________
void PoolSvc::patchCatalog(const std::string& pfn, pool::IDatabase& dbH) const {
   std::scoped_lock lock(m_pool_mut);
   dbH.setTechnology(pool::ROOT_StorageType.type());
   std::string fid = dbH.fid();
   Gaudi::IFileCatalog* catalog_locked ATLAS_THREAD_SAFE = m_catalog.get();
   FileCatalogUtils::registerPFN(*catalog_locked, pfn, "ROOT_All", fid);
}
//__________________________________________________________________________
Token* PoolSvc::getToken(const std::string& connection,
                          const std::string& collection,
                          const unsigned long ientry) {
   std::lock_guard<CallMutex> lock(*m_pers_mut[IPoolSvc::kInputStream]);
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(IPoolSvc::kInputStream, connection);
   if (dbH == nullptr) {
      return(nullptr);
   }
   if (dbH->openMode() == Io::INVALID) {
      dbH->connectForRead();
   }
   std::unique_ptr<pool::IContainer> contH = getContainerHandle(dbH.get(), collection);
   if (contH == nullptr) {
      return(nullptr);
   }
   auto tokenIter = std::unique_ptr<pool::ITokenIterator>(contH->tokens());
   // the Token returned by the iterator has the refCount already increased
   return tokenIter->seek(ientry)? tokenIter->next() : nullptr;
}
//__________________________________________________________________________
StatusCode PoolSvc::connect(Io::IoFlag type, unsigned int contextId) {
   if (type != Io::READ) {
      if (contextId >= m_dbSessionVec.size()) {
         ATH_MSG_WARNING("connect: Using default output Stream instead of id = " << contextId);
         contextId = IPoolSvc::kOutputStream;
      }
   } else {
      if (contextId > m_dbSessionVec.size()) {
         ATH_MSG_WARNING("connect: Using default input Stream instead of id = " << contextId);
         contextId = IPoolSvc::kInputStream;
      } else if (contextId == m_dbSessionVec.size()) {
         ATH_MSG_INFO("Connecting to InputStream for: " << contextId);
         contextId = this->getInputContext("");
      }
   }
   if (contextId >= m_dbSessionVec.size()) {
      return StatusCode::FAILURE;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   auto& session = m_dbSessionVec[contextId];
   // Connect to a logical database using the pre-defined technology and dbID
   if( session->openMode() == Io::INVALID && !session->start(type) ) {
      ATH_MSG_ERROR("connect failed session = " << session << " type = " << type);
      return StatusCode::FAILURE;
   }
   return StatusCode::SUCCESS;
}
//__________________________________________________________________________
StatusCode PoolSvc::commit(unsigned int contextId) {
   if (contextId >= m_dbSessionVec.size()) {
      return StatusCode::FAILURE;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   auto& session = m_dbSessionVec[contextId];
   if( session->openMode() != Io::INVALID && !session->commit() ) {
      ATH_MSG_ERROR("POOL commit failed " << session);
      return StatusCode::FAILURE;
   }
   if (session->openMode() == Io::READ) {
      session->disconnectAll();
   }
   return StatusCode::SUCCESS;
}
//__________________________________________________________________________
StatusCode PoolSvc::commitAndHold(unsigned int contextId) {
   if (contextId >= m_dbSessionVec.size()) {
      return StatusCode::FAILURE;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   auto& session = m_dbSessionVec[contextId];
   if( session->openMode() != Io::INVALID && !session->commitAndHold() ) {
      ATH_MSG_ERROR("POOL commitAndHold failed " << session);
      return StatusCode::FAILURE;
   }
   return StatusCode::SUCCESS;
}
//__________________________________________________________________________
StatusCode PoolSvc::disconnect(unsigned int contextId) {
   ATH_MSG_DEBUG("Disconnect request for contextId=" << contextId);
   if (contextId >= m_dbSessionVec.size()) {
      return StatusCode::SUCCESS;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   auto& session = m_dbSessionVec[contextId];
   if( session->openMode() != Io::INVALID ) {
      if( !commit(contextId).isSuccess() ) {
         ATH_MSG_ERROR("disconnect failed to commit " << session);
         return StatusCode::FAILURE;
      }  
      if( session->disconnectAll() ) {
         ATH_MSG_DEBUG("Disconnected PersistencySvc session");
      } else {
         ATH_MSG_ERROR("disconnect failed to diconnect PersistencySvc");
         return StatusCode::FAILURE;
      }
   }
   return StatusCode::SUCCESS;
}
//__________________________________________________________________________
StatusCode PoolSvc::disconnectDb(const std::string& connection, unsigned int contextId) {
   if (contextId >= m_dbSessionVec.size()) {
      return (StatusCode::SUCCESS);
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(contextId, connection);
   if (dbH == nullptr) {
      ATH_MSG_ERROR("Failed to get Session/DatabaseHandle.");
      return(StatusCode::FAILURE);
   }
   std::map<unsigned int, unsigned int>::const_iterator maxFileIter = m_contextMaxFile.find(contextId);
   if (maxFileIter != m_contextMaxFile.end() && maxFileIter->second > 0) {
      m_guidLists[contextId].remove(Guid(dbH->fid()));
   }
   dbH->disconnect();
   return(StatusCode::SUCCESS);
}
//_______________________________________________________________________
StatusCode PoolSvc::getAttribute(const std::string& optName,
        std::string& data,
        long tech,
        unsigned int contextId) {
   if (contextId >= m_dbSessionVec.size()) {
      ATH_MSG_WARNING("getAttribute: Using default input Stream instead of id = " << contextId);
      contextId = IPoolSvc::kInputStream;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   auto& sesH = m_dbSessionVec[contextId];
   std::ostringstream oss;
   if (data == "DbLonglong") {
      long long int attr_data;
      oss << std::dec << sesH->microSessionManager(tech).attributeOfType(optName, static_cast<void*>(&attr_data), typeid(long long int), "");
   } else if (data == "double") {
      double attr_data;
      oss << std::dec << sesH->microSessionManager(tech).attributeOfType(optName, static_cast<void*>(&attr_data), typeid(double), "");
   } else {
      int attr_data;
      oss << std::dec << sesH->microSessionManager(tech).attributeOfType(optName, static_cast<void*>(&attr_data), typeid(int), "");
   }
   data = oss.str();
   ATH_MSG_INFO("Domain attribute [" << optName << "]" << ": " << data);
   return(StatusCode::SUCCESS);
}
//_______________________________________________________________________
StatusCode PoolSvc::getAttribute(const std::string& optName,
        std::string& data,
        long tech,
        const std::string& dbName,
        const std::string& contName,
        unsigned int contextId) {
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(contextId, dbName);
   if (dbH == nullptr) {
      ATH_MSG_DEBUG("getAttribute: Failed to get Session/DatabaseHandle to get POOL property.");
      return(StatusCode::FAILURE);
   }
   if (dbH->openMode() == Io::INVALID) {
      if (m_dbSessionVec[contextId]->openMode() == Io::WRITE || m_dbSessionVec[contextId]->openMode() == Io::APPEND) {
         dbH->setTechnology(tech);
         dbH->connectForWrite();
      } else {
         dbH->connectForRead();
      }
   }
   std::ostringstream oss;
   if (contName.empty()) {
      if (data == "DbLonglong") {
         long long int attr_data;
         oss << std::dec << dbH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(long long int), "");
      } else if (data == "double") {
         double attr_data;
         oss << std::dec << dbH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(double), "");
      } else if (data == "string") {
         char* attr_data;
         oss << std::dec << dbH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(char*), "");
      } else {
         int attr_data;
         oss << std::dec << dbH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(int), "");
      }
      ATH_MSG_INFO("Database (" << dbH->pfn() << ") attribute [" << optName << "]" << ": " << oss.str());
   } else {
      std::unique_ptr<pool::IContainer> contH = getContainerHandle(dbH.get(), contName);
      if (contH == nullptr) {
         ATH_MSG_DEBUG("Failed to get ContainerHandle to get POOL property.");
         return(StatusCode::FAILURE);
      }
      if (data == "DbLonglong") {
         long long int attr_data;
         oss << std::dec << contH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(long long int), "");
      } else if (data == "double") {
         double attr_data;
         oss << std::dec << contH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(double), "");
      } else {
         int attr_data;
         oss << std::dec << contH->attributeOfType(optName, static_cast<void*>(&attr_data), typeid(int), "");
      }
      ATH_MSG_INFO("Container attribute [" << contName << "." << optName << "]: " << oss.str());
   }
   data = oss.str();
   return(StatusCode::SUCCESS);
}
//_______________________________________________________________________
StatusCode PoolSvc::setAttribute(const std::string& optName,
        const std::string& data,
        long tech,
        unsigned int contextId) {
   if (contextId >= m_dbSessionVec.size()) {
      ATH_MSG_WARNING("setAttribute: Using default output Stream instead of id = " << contextId);
      contextId = IPoolSvc::kOutputStream;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   auto& sesH = m_dbSessionVec[contextId];
   if (data[data.size() - 1] == 'L') {
      const long long int& atttibuteValue = atoll(data.c_str());
      if (!sesH->microSessionManager(tech).setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(long long int), "")) {
         ATH_MSG_DEBUG("Failed to set POOL property, " << optName << " to " << data);
         return(StatusCode::FAILURE);
      }
   } else {
      const int& atttibuteValue = atoi(data.c_str());
      if (!sesH->microSessionManager(tech).setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(int), "")) {
         ATH_MSG_DEBUG("Failed to set POOL property, " << optName << " to " << data);
         return(StatusCode::FAILURE);
      }
   }
   return(StatusCode::SUCCESS);
}
//_______________________________________________________________________
StatusCode PoolSvc::setAttribute(const std::string& optName,
        const std::string& data,
        long tech,
        const std::string& dbName,
        const std::string& contName,
        unsigned int contextId) {
   if (contextId >= m_dbSessionVec.size()) {
      ATH_MSG_WARNING("setAttribute: Using default output Stream instead of id = " << contextId);
      contextId = IPoolSvc::kOutputStream;
   }
   ContextLock lock(contextId, m_pool_mut, m_pers_mut);
   std::unique_ptr<pool::IDatabase> dbH = getDbHandle(contextId, dbName);
   if (dbH == nullptr) {
      ATH_MSG_DEBUG("Failed to get Session/DatabaseHandle to set POOL property.");
      return(StatusCode::FAILURE);
   }
   if (dbH->openMode() == Io::INVALID) {
      if (m_dbSessionVec[contextId]->openMode() == Io::WRITE || m_dbSessionVec[contextId]->openMode() == Io::APPEND) {
         dbH->setTechnology(tech);
         dbH->connectForWrite();
      } else {
         dbH->connectForRead();
      }
   }
   bool retError = false;
   std::string objName;
   bool hasTTreeName = contName.starts_with("TTree=");
   if (contName.empty() || hasTTreeName || m_dbSessionVec[contextId]->openMode() == Io::READ) {
      objName = hasTTreeName ? contName.substr(6) : contName;
      if( !isNumber(data) ) {
         const char* atttibuteValue = data.c_str();
         retError = dbH->setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(char*), objName);
      } else if( data[data.size() - 1] == 'L' ) {
         const long long int& atttibuteValue = atoll(data.c_str());
         retError = dbH->setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(long long int), objName);
      } else {
         const int& atttibuteValue = atoi(data.c_str());
         retError = dbH->setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(int), objName);
      }
      if (!retError) {
         ATH_MSG_DEBUG("Failed to set POOL property, " << optName << " to " << data);
         return(StatusCode::FAILURE);
      }
   } else {
      std::unique_ptr<pool::IContainer> contH = getContainerHandle(dbH.get(), contName);
      if (contH == nullptr) {
         ATH_MSG_DEBUG("Failed to get ContainerHandle to set POOL property.");
         return(StatusCode::FAILURE);
      }
      if (auto p = contName.find('('); p != std::string::npos) {
         objName = contName.substr(p + 1); // Get BranchName between parenthesis
         objName.erase(objName.find(')'));
      } else if (auto p = contName.find("::"); p != std::string::npos) {
         objName = contName.substr(p + 2); // Split off Tree name
      } else if (auto p = contName.find('_'); p != std::string::npos) {
         objName = contName.substr(p + 1); // Split off "POOLContainer"
         objName.erase(objName.find('/')); // Split off key
      }
      std::string::size_type off = 0;
      while ((off = objName.find_first_of("<>/")) != std::string::npos) {
         objName[off] = '_'; // Replace special chars (e.g. templates)
      }
      if (data[data.size() - 1] == 'L') {
         const long long int& atttibuteValue = atoll(data.c_str());
         retError = contH->setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(long long int), objName);
      } else {
         const int& atttibuteValue = atoi(data.c_str());
         retError = contH->setAttributeOfType(optName, static_cast<const void*>(&atttibuteValue), typeid(int), objName);
      }
      if (!retError) {
         ATH_MSG_DEBUG("Failed to set POOL container property, " << optName << " for " << contName << " : " << objName << " to " << data);
         return(StatusCode::FAILURE);
      }
   }
   return(StatusCode::SUCCESS);
}
 
//__________________________________________________________________________
bool PoolSvc::createCatalog() {
   m_catalogMgr = Gaudi::svcLocator()->service<Gaudi::IFileCatalogMgr>("Gaudi::MultiFileCatalog");
   m_catalog = SmartIF<Gaudi::IFileCatalog>(m_catalogMgr);
   if (!m_catalogMgr.isValid() || !m_catalog.isValid()) {
      return false;
   }
   // set the output level of the XMLCatalog component - works only if the Gaudi AppMgr was initialized
   Gaudi::svcLocator()->service<IMessageSvc>("MessageSvc")
      ->setOutputLevel("XMLCatalog", pool::DbPrint::getOutputLvl());
   m_catalogMgr->removeCatalog("*");
   for (auto& catalog : m_readCatalog.value()) {
      ATH_MSG_DEBUG("POOL ReadCatalog is " << catalog);
      if (catalog.starts_with("apcfile:") || catalog.starts_with("prfile:")) {
         std::string::size_type cpos = catalog.find(':');
         // check for file accessed via ATLAS_POOLCOND_PATH
         std::string file = poolCondPath(catalog.substr(cpos + 1));
         if (!file.empty()) {
            ATH_MSG_INFO("Resolved path (via ATLAS_POOLCOND_PATH) is " << file);
            FileCatalogUtils::addCatalog(*m_catalogMgr, "file:" + file);
         } else {
            // As backup, check for file accessed via PathResolver
            file = PathResolver::find_file(catalog.substr(cpos + 1), "DATAPATH");
            if (!file.empty()) {
               ATH_MSG_INFO("Resolved path (via DATAPATH) is " << file);
               FileCatalogUtils::addCatalog(*m_catalogMgr, "file:" + file);
            } else {
               ATH_MSG_INFO("Unable find catalog "
                   << catalog
                   << " in $ATLAS_POOLCOND_PATH and $DATAPATH");
            }
         }
      } else {
         FileCatalogUtils::addCatalog(*m_catalogMgr, catalog);
      }
   }
   try {
      ATH_MSG_INFO("POOL WriteCatalog is " << m_writeCatalog.value());
      FileCatalogUtils::addCatalog(*m_catalogMgr, m_writeCatalog.value(), true);
   } catch(std::exception& e) {
      ATH_MSG_ERROR("setWriteCatalog - caught exception: " << e.what());
      return false; // This catalog is not setup properly!
   }
   return true;
}
 
//__________________________________________________________________________
PoolSvc::~PoolSvc() {
}
//__________________________________________________________________________
std::unique_ptr<pool::IDatabase> PoolSvc::getDbHandle(unsigned int contextId, const std::string& dbName) {
   if (contextId >= m_dbSessionVec.size()) {
      ATH_MSG_WARNING("getDbHandle: Using default input Stream instead of id = " << contextId);
      contextId = IPoolSvc::kInputStream;
   }
   auto& sesH = m_dbSessionVec[contextId];
   if (!sesH){
     ATH_MSG_ERROR("Session pointer is null.");
     return nullptr;
   }
   if (sesH->openMode() == Io::INVALID) {
      Io::IoFlag transMode = Io::READ;
      ATH_MSG_DEBUG("Start transaction, type = " << transMode);
      if (!sesH->start(transMode)) {
         ATH_MSG_WARNING("Failed to start transaction, type = " << transMode);
         return nullptr;
      }
   }
   if (dbName.starts_with("PFN:")) {
      return sesH->databaseHandle(dbName.substr(4), pool::DatabaseSpecification::PFN);
   } else if (dbName.starts_with("LFN:")) {
      return sesH->databaseHandle(dbName.substr(4), pool::DatabaseSpecification::LFN);
   } else if (dbName.starts_with("FID:")) {
      return sesH->databaseHandle(dbName.substr(4), pool::DatabaseSpecification::FID);
   }
   return sesH->databaseHandle(dbName, pool::DatabaseSpecification::PFN);
}
//__________________________________________________________________________
std::unique_ptr<pool::IContainer> PoolSvc::getContainerHandle(pool::IDatabase* dbH, const std::string& contName) {
   pool::IContainer* contH = nullptr;
   if (dbH == nullptr) {
      ATH_MSG_DEBUG("No DatabaseHandle to get Container.");
      return(nullptr);
   }
   if (contName.find("DataHeader") != std::string::npos) {
      contH = dbH->containerHandle(contName.substr(0, contName.find("_p")));
   } else {
      contH = dbH->containerHandle(contName);
   }
   return(std::unique_ptr<pool::IContainer>(contH));
}
//__________________________________________________________________________
std::string PoolSvc::poolCondPath(const std::string& leaf) {
   // look for files at $ATLAS_POOLCOND_PATH/<leaf>
   // return full filename if exists, or empty string if not
   const char* cpath = std::getenv("ATLAS_POOLCOND_PATH");
   if (cpath && strcmp(cpath, "") != 0) {
      const std::string testpath = std::string(cpath) + "/" + leaf;
 
      // Try to open file for reading. Note that a simple stat call may return
      // a wrong result if the file is residing on an auto-mounted FS (ATR-28801).
      if (FILE* fp = std::fopen(testpath.c_str(), "r")) {
         std::fclose(fp);
         return testpath;
      }
   }
   return {};
}

StatusCode
PoolSvc::createSession( const std::string& sessionName, int dbAgeLimit )
{
   ToolHandle<SessionTool> session_tool = ToolHandle<SessionTool>("SessionTool/" + sessionName, this);
   ATH_CHECK(session_tool.retrieve());
   session_tool->setupFC( *m_catalog, dbAgeLimit );
   m_dbSessionVec.push_back(session_tool);
   return StatusCode::SUCCESS;
}