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. More...
virtual	~AthenaRootSharedWriterSvc ()=default
	Destructor. More...
virtual StatusCode	initialize () override
	Gaudi Service Interface method implementations: More...
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
THashTable	m_rootMergers
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 25 of file AthenaRootSharedWriterSvc.h.

Constructor & Destructor Documentation

AthenaRootSharedWriterSvc()

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

Standard Service Constructor.

Definition at line 157 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 324 of file AthenaRootSharedWriterSvc.cxx.

                                               {
   ATH_MSG_INFO("in finalize()");
   delete m_rootMonitor; m_rootMonitor = nullptr;
   delete m_rootServerSocket; m_rootServerSocket = nullptr;
   return StatusCode::SUCCESS;
}

initialize()

StatusCode AthenaRootSharedWriterSvc::initialize ( )

overridevirtual

Gaudi Service Interface method implementations:

Definition at line 162 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()) {
         int streamPort = 0;
         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: " << streamPort);
         } else {
            streamPort = atoi(streamPortStringProp.value().substr(streamPortStringProp.value().find(':') + 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();
         const std::string newStreamPortString{streamPortStringProp.value().substr(0,streamPortStringProp.value().find(':')+1) + std::to_string(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);
         ATH_MSG_DEBUG("Successfully created ROOT TServerSocket and added it to TMonitor: ready to accept connections, " << streamPort);
      }
   }
   // 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 221 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, 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 319 of file AthenaRootSharedWriterSvc.cxx.

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

Friends And Related Function Documentation

SvcFactory< AthenaRootSharedWriterSvc >

friend class SvcFactory< AthenaRootSharedWriterSvc >

friend

Definition at line 1 of file AthenaRootSharedWriterSvc.h.

Member Data Documentation

m_cnvSvc

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

private

Definition at line 44 of file AthenaRootSharedWriterSvc.h.

m_numberOfStreams

int AthenaRootSharedWriterSvc::m_numberOfStreams

private

Definition at line 51 of file AthenaRootSharedWriterSvc.h.

m_rootClientCount

int AthenaRootSharedWriterSvc::m_rootClientCount

private

Definition at line 50 of file AthenaRootSharedWriterSvc.h.

m_rootClientIndex

int AthenaRootSharedWriterSvc::m_rootClientIndex

private

Definition at line 49 of file AthenaRootSharedWriterSvc.h.

m_rootMergers

THashTable AthenaRootSharedWriterSvc::m_rootMergers

private

Definition at line 48 of file AthenaRootSharedWriterSvc.h.

m_rootMonitor

TMonitor* AthenaRootSharedWriterSvc::m_rootMonitor

private

Definition at line 47 of file AthenaRootSharedWriterSvc.h.

m_rootServerSocket

TServerSocket* AthenaRootSharedWriterSvc::m_rootServerSocket

private

Definition at line 46 of file AthenaRootSharedWriterSvc.h.

---

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

• AthenaRootSharedWriterSvc.h
• AthenaRootSharedWriterSvc.cxx