AthenaRootSharedWriterSvc Class Reference

This class provides an example for writing event data objects to Pool. More...

#include <AthenaRootSharedWriterSvc.h>

Inheritance diagram for AthenaRootSharedWriterSvc:

Collaboration diagram for AthenaRootSharedWriterSvc:

Public Member Functions
	AthenaRootSharedWriterSvc (const std::string &name, ISvcLocator *pSvcLocator)
	Standard Service Constructor.
virtual	~AthenaRootSharedWriterSvc ()=default
	Destructor.
virtual StatusCode	initialize () override
	Gaudi Service Interface method implementations:
virtual StatusCode	stop () override
virtual StatusCode	finalize () override
virtual StatusCode	share (int numClients=0, bool motherClient=false) override

Private Attributes
ServiceHandle< AthenaPoolSharedIOCnvSvc >	m_cnvSvc {this,"AthenaPoolSharedIOCnvSvc","AthenaPoolSharedIOCnvSvc"}
TServerSocket *	m_rootServerSocket
TMonitor *	m_rootMonitor
std::string	m_socketPath
	Path of the UNIX domain socket file, if used, removed on finalize.
THashTable	m_rootMergers
std::unordered_map< TClass *, void * >	m_cachedObjects
int	m_rootClientIndex
int	m_rootClientCount
int	m_numberOfStreams

Friends
class	SvcFactory< AthenaRootSharedWriterSvc >

Detailed Description

This class provides an example for writing event data objects to Pool.

Definition at line 26 of file AthenaRootSharedWriterSvc.h.

Constructor & Destructor Documentation

AthenaRootSharedWriterSvc()

AthenaRootSharedWriterSvc::AthenaRootSharedWriterSvc	(	const std::string &	name,
		ISvcLocator *	pSvcLocator )

Standard Service Constructor.

Definition at line 151 of file AthenaRootSharedWriterSvc.cxx.

  : base_class(name, pSvcLocator)
  , m_rootServerSocket(nullptr), m_rootMonitor(nullptr), m_rootMergers(), m_rootClientIndex(0), m_rootClientCount(0), m_numberOfStreams(0) {
}

~AthenaRootSharedWriterSvc()

virtual AthenaRootSharedWriterSvc::~AthenaRootSharedWriterSvc ( )

virtualdefault

Destructor.

Member Function Documentation

finalize()

StatusCode AthenaRootSharedWriterSvc::finalize ( )

overridevirtual

Definition at line 354 of file AthenaRootSharedWriterSvc.cxx.

                                               {
   ATH_MSG_INFO("in finalize()");
   delete m_rootMonitor; m_rootMonitor = nullptr;
   delete m_rootServerSocket; m_rootServerSocket = nullptr;
   if (!m_socketPath.empty()) {
      //coverity[CHECKED_RETURN]
      std::remove(m_socketPath.c_str()); //returns 0 on success.
   }
   for (auto& [cl, ptr] : m_cachedObjects) {
      if (cl && ptr) {
         cl->Destructor(ptr, false);
      }
   }
   m_cachedObjects.clear();
   return StatusCode::SUCCESS;
}

initialize()

StatusCode AthenaRootSharedWriterSvc::initialize ( )

overridevirtual

Gaudi Service Interface method implementations:

Definition at line 156 of file AthenaRootSharedWriterSvc.cxx.

                                                 {
   ATH_MSG_INFO("in initialize()");
 
   // Initialize IConversionSvc
   ATH_CHECK(m_cnvSvc.retrieve());
   IProperty* propertyServer = dynamic_cast<IProperty*>(m_cnvSvc.get());
   if (propertyServer == nullptr) {
      ATH_MSG_ERROR("Unable to cast conversion service to IProperty");
      return StatusCode::FAILURE;
   } else {
      std::string propertyName = "ParallelCompression";
      bool parallelCompression(false);
      BooleanProperty parallelCompressionProp(propertyName, parallelCompression);
      if (propertyServer->getProperty(&parallelCompressionProp).isFailure()) {
         ATH_MSG_INFO("Conversion service does not have ParallelCompression property");
      } else if (parallelCompressionProp.value()) {
         propertyName = "StreamPortString";
         std::string streamPortString("");
         StringProperty streamPortStringProp(propertyName, streamPortString);
         if (propertyServer->getProperty(&streamPortStringProp).isFailure()) {
            ATH_MSG_INFO("Conversion service does not have StreamPortString property, using default TCP port: 0");
            streamPortStringProp.setValue("?pmerge=localhost:0");
         }
         const std::string& pmergeProperty = streamPortStringProp.value();
         const std::size_t eqPos = pmergeProperty.find('=');
         if (eqPos == std::string::npos) {
            ATH_MSG_FATAL("Malformed StreamPortString property (missing '='): " << pmergeProperty);
            return StatusCode::FAILURE;
         }
         const std::string pmergePrefix = pmergeProperty.substr(0, eqPos + 1);
         const std::string pmergeArg = pmergeProperty.substr(eqPos + 1);
         if (pmergeArg.empty()) {
            ATH_MSG_FATAL("Malformed StreamPortString property (empty value after '='): " << pmergeProperty);
            return StatusCode::FAILURE;
         }
         std::string newStreamPortString;
         // "?pmerge=<host>:<port>" (default) selects a TCP socket.
         // Anything else (e.g. "?pmerge=<prefix>") is passed to gSystem->TempFileName() as a
         // (possibly relative) prefix for a UNIX domain socket file, created as $TMPDIR/<prefix>XXXXXX.
         if (pmergeArg.find(':') == std::string::npos) {
            TString socketPath = pmergeArg.c_str();
            FILE* reservedFile = gSystem->TempFileName(socketPath);
            if (reservedFile == nullptr) {
               ATH_MSG_FATAL("Could not create temporary file for UNIX domain socket: " << pmergeArg);
               return StatusCode::FAILURE;
            }
            m_socketPath = socketPath.Data();
            //coverity[CHECKED_RETURN]
            std::remove(m_socketPath.c_str()); //returns 0 if ok.
            std::fclose(reservedFile);
            m_rootServerSocket = new TServerSocket(socketPath);
            if (m_rootServerSocket == nullptr || !m_rootServerSocket->IsValid()) {
               ATH_MSG_FATAL("Could not create ROOT TServerSocket (UNIX domain socket): " << m_socketPath);
               return StatusCode::FAILURE;
            }
            newStreamPortString = pmergePrefix + m_socketPath;
            ATH_MSG_DEBUG("Successfully created ROOT TServerSocket (UNIX domain socket) and added it to TMonitor: ready to accept connections, " << m_socketPath);
         } else {
            const std::size_t colonPos = pmergeArg.find(':');
            int streamPort = atoi(pmergeArg.substr(colonPos + 1).c_str());
            m_rootServerSocket = new TServerSocket(streamPort, (streamPort == 0 ? false : true), 100, -1, ESocketBindOption::kInaddrLoopback);
            if (m_rootServerSocket == nullptr || !m_rootServerSocket->IsValid()) {
               ATH_MSG_FATAL("Could not create ROOT TServerSocket: " << streamPort);
               return StatusCode::FAILURE;
            }
            streamPort = m_rootServerSocket->GetLocalPort();
            newStreamPortString = pmergePrefix + pmergeArg.substr(0, colonPos + 1) + std::to_string(streamPort);
            ATH_MSG_DEBUG("Successfully created ROOT TServerSocket and added it to TMonitor: ready to accept connections, " << streamPort);
         }
         if (propertyServer->setProperty(propertyName,newStreamPortString).isFailure()) {
            ATH_MSG_FATAL("Could not set Conversion Service property " << propertyName << " from " << streamPortString << " to " << newStreamPortString);
            return StatusCode::FAILURE;
         }
         m_rootMonitor = new TMonitor;
         m_rootMonitor->Add(m_rootServerSocket);
      }
   }
   // Count the number of output streams
   const IAlgManager* algMgr = Gaudi::svcLocator()->as<IAlgManager>();
   for (const auto& alg : algMgr->getAlgorithms()) {
      if (alg->type() == "AthenaOutputStream") {
         ATH_MSG_DEBUG("Counting " << alg->name() << " as an output stream algorithm");
         m_numberOfStreams++;
      }
   }
   if (m_numberOfStreams == 0) {
      ATH_MSG_WARNING("No output stream algorithm found, setting the number of streams to 1");
      m_numberOfStreams = 1;
   } else {
      ATH_MSG_INFO("Found a total of " << m_numberOfStreams << " output streams");
   }
 
   return StatusCode::SUCCESS;
}

share()

StatusCode AthenaRootSharedWriterSvc::share ( int numClients = 0, bool motherClient = false )

overridevirtual

Definition at line 251 of file AthenaRootSharedWriterSvc.cxx.

                                                                             {
   ATH_MSG_DEBUG("Start commitOutput loop");
   StatusCode sc = m_cnvSvc->commitOutput("", false);
 
   // Allow ROOT clients to start up (by setting active clients)
   // and wait to stop the ROOT server until all clients are done and metadata is written (commitOutput fail).
   bool anyActiveClients = (m_rootServerSocket != nullptr);
   while (sc.isSuccess() || sc.isRecoverable() || anyActiveClients) {
      if (sc.isSuccess()) {
         ATH_MSG_VERBOSE("Success in commitOutput loop");
      } else if (m_rootMonitor != nullptr) {
         TSocket* socket = m_rootMonitor->Select(1);
         if (socket != nullptr && socket != (TSocket*)-1) {
            ATH_MSG_DEBUG("ROOT Monitor got: " << socket);
            if (socket->IsA() == TServerSocket::Class()) {
               TSocket* client = (static_cast<TServerSocket*>(socket))->Accept();
               client->Send(m_rootClientIndex, 0);
               client->Send(1, 1);
               ++m_rootClientIndex;
               ++m_rootClientCount;
               if (m_rootClientCount < (numClients-1)*m_numberOfStreams + 1) {
                  m_rootMonitor->Add(client);
                  ATH_MSG_INFO("ROOT Monitor add client: " << m_rootClientIndex << ", " << client);
               } else {
                  ATH_MSG_WARNING("ROOT Monitor do NOT add client: " << m_rootClientIndex << ", " << client);
                  client->Close("force");
                  --m_rootClientCount;
               }
            } else {
               TMessage* message = nullptr;
               Int_t result = socket->Recv(message);
               if (result < 0) {
                  ATH_MSG_ERROR("ROOT Monitor got an error while receiving the message from the socket: " << result);
                  return StatusCode::FAILURE;
               }
               if (message == nullptr) {
                  ATH_MSG_WARNING("ROOT Monitor got no message from socket: " << socket);
               } else if (message->What() == kMESS_STRING) {
                  char str[64];
                  message->ReadString(str, 64);
                  ATH_MSG_INFO("ROOT Monitor client: " << socket << ", " << str);
                  m_rootMonitor->Remove(socket);
                  ATH_MSG_DEBUG("ROOT Monitor client: " << socket << ", " << socket->GetBytesRecv() << ", " << socket->GetBytesSent());
                  socket->Close();
                  --m_rootClientCount;
                  if (m_rootMonitor->GetActive() == 0 || m_rootClientCount == 0) {
                     if (!motherClient) {
                        anyActiveClients = false;
                        ATH_MSG_INFO("ROOT Monitor: No more active clients...");
                     } else {
                        motherClient = false;
                        ATH_MSG_INFO("ROOT Monitor: Mother process is done...");
                        if (!m_cnvSvc->commitCatalog().isSuccess()) {
                           ATH_MSG_FATAL("Failed to commit file catalog.");
                           return StatusCode::FAILURE;
                        }
                     }
                  }
               } else if (message->What() == kMESS_ANY) {
                  long long length;
                  TString filename;
                  int clientId;
                  message->ReadInt(clientId);
                  message->ReadTString(filename);
                  message->ReadLong64(length);
                  ATH_MSG_DEBUG("ROOT Monitor client: " << socket << ", " << clientId << ": " << filename << ", " << length);
                  std::unique_ptr<TMemFile> transient(new TMemFile(filename, message->Buffer() + message->Length(), length, "UPDATE"));
                  message->SetBufferOffset(message->Length() + length);
                  ParallelFileMerger* info = static_cast<ParallelFileMerger*>(m_rootMergers.FindObject(filename));
                  if (!info) {
                     info = new ParallelFileMerger(filename, &m_cachedObjects, transient->GetCompressionSettings());
                     m_rootMergers.Add(info);
                     ATH_MSG_INFO("ROOT Monitor ParallelFileMerger: " << info << ", for: " << filename);
                  }
                  info->MergeTrees(transient.get());
               }
               delete message; message = nullptr;
            }
         }
      } else if (m_rootMonitor == nullptr) {
         usleep(100);
      }
      // Once commitOutput failed all legacy clients are finished (writing metadata), do not call again.
      if (sc.isSuccess() || sc.isRecoverable()) {
         sc = m_cnvSvc->commitOutput("", false);
         if (sc.isFailure() && !sc.isRecoverable()) {
            ATH_MSG_INFO("commitOutput failed, metadata done.");
            if (anyActiveClients && m_rootClientCount == 0) {
              ATH_MSG_INFO("ROOT Monitor: No clients, terminating the loop...");
              anyActiveClients = false;
            }
         }
      }
   }
   ATH_MSG_INFO("End commitOutput loop");
   return StatusCode::SUCCESS;
}

stop()

StatusCode AthenaRootSharedWriterSvc::stop ( )

overridevirtual

Definition at line 349 of file AthenaRootSharedWriterSvc.cxx.

                                           {
   m_rootMergers.Delete();
   return StatusCode::SUCCESS;
}

Friends And Related Symbol Documentation

SvcFactory< AthenaRootSharedWriterSvc >

friend class SvcFactory< AthenaRootSharedWriterSvc >

friend

Definition at line 1 of file AthenaRootSharedWriterSvc.h.

Member Data Documentation

m_cachedObjects

std::unordered_map<TClass*, void*> AthenaRootSharedWriterSvc::m_cachedObjects

private

Definition at line 52 of file AthenaRootSharedWriterSvc.h.

m_cnvSvc

ServiceHandle<AthenaPoolSharedIOCnvSvc> AthenaRootSharedWriterSvc::m_cnvSvc {this,"AthenaPoolSharedIOCnvSvc","AthenaPoolSharedIOCnvSvc"}

private

Definition at line 45 of file AthenaRootSharedWriterSvc.h.

{this,"AthenaPoolSharedIOCnvSvc","AthenaPoolSharedIOCnvSvc"};

m_numberOfStreams

int AthenaRootSharedWriterSvc::m_numberOfStreams

private

Definition at line 55 of file AthenaRootSharedWriterSvc.h.

m_rootClientCount

int AthenaRootSharedWriterSvc::m_rootClientCount

private

Definition at line 54 of file AthenaRootSharedWriterSvc.h.

m_rootClientIndex

int AthenaRootSharedWriterSvc::m_rootClientIndex

private

Definition at line 53 of file AthenaRootSharedWriterSvc.h.

m_rootMergers

THashTable AthenaRootSharedWriterSvc::m_rootMergers

private

Definition at line 51 of file AthenaRootSharedWriterSvc.h.

m_rootMonitor

TMonitor* AthenaRootSharedWriterSvc::m_rootMonitor

private

Definition at line 48 of file AthenaRootSharedWriterSvc.h.

m_rootServerSocket

TServerSocket* AthenaRootSharedWriterSvc::m_rootServerSocket

private

Definition at line 47 of file AthenaRootSharedWriterSvc.h.

m_socketPath

std::string AthenaRootSharedWriterSvc::m_socketPath

private

Path of the UNIX domain socket file, if used, removed on finalize.

Definition at line 50 of file AthenaRootSharedWriterSvc.h.

---

The documentation for this class was generated from the following files:

• AthenaRootSharedWriterSvc.h
• AthenaRootSharedWriterSvc.cxx