ATLAS Offline Software
Public Member Functions | Protected Types | Protected Member Functions | Protected Attributes | Private Member Functions | Private Attributes | List of all members
SharedWriterTool Class Referencefinalabstract

#include <SharedWriterTool.h>

Inheritance diagram for SharedWriterTool:
Collaboration diagram for SharedWriterTool:

Public Member Functions

 SharedWriterTool (const std::string &type, const std::string &name, const IInterface *parent)
 
virtual ~SharedWriterTool () override
 
virtual StatusCode initialize () override
 
virtual StatusCode finalize () override
 
virtual int makePool ATLAS_NOT_THREAD_SAFE (int maxevt, int nprocs, const std::string &topdir) override
 
virtual StatusCode exec ATLAS_NOT_THREAD_SAFE () override
 
virtual void subProcessLogs (std::vector< std::string > &) override
 
virtual AthenaMP::AllWorkerOutputs_ptr generateOutputReport () override
 
virtual std::unique_ptr< AthenaInterprocess::ScheduledWorkbootstrap_func () override
 
virtual std::unique_ptr< AthenaInterprocess::ScheduledWorkexec_func () override
 
virtual std::unique_ptr< AthenaInterprocess::ScheduledWorkfin_func () override
 
virtual StatusCode wait_once ATLAS_NOT_THREAD_SAFE (pid_t &pid) override
 
virtual void reportSubprocessStatuses () override
 
virtual void useFdsRegistry (std::shared_ptr< AthenaInterprocess::FdsRegistry >) override
 
virtual void setRandString (const std::string &randStr) override
 
virtual void setMaxEvt (int maxEvt) override
 
virtual void setMPRunStop (const AthenaInterprocess::IMPRunStop *runStop) override
 
virtual void killChildren () override
 
virtual std::unique_ptr< AthenaInterprocess::ScheduledWork > virtual operator() ATLAS_NOT_THREAD_SAFE(const AthenaInterprocess std::unique_ptr< AthenaInterprocess::ScheduledWorkbootstrap_func ()=0
 
virtual std::unique_ptr< ScheduledWork > operator () ATLAS_NOT_THREAD_SAFE(const ScheduledWork &)=0
 

Protected Types

enum  ESRange_Status {
  ESRANGE_SUCCESS, ESRANGE_NOTFOUND, ESRANGE_SEEKFAILED, ESRANGE_PROCFAILED,
  ESRANGE_FILENOTMADE, ESRANGE_BADINPFILE
}
 
enum  Func_Flag { FUNC_BOOTSTRAP, FUNC_EXEC, FUNC_FIN }
 

Protected Member Functions

int mapAsyncFlag ATLAS_NOT_THREAD_SAFE (Func_Flag flag, pid_t pid=0)
 
int redirectLog (const std::string &rundir, bool addTimeStamp=true)
 
int updateIoReg (const std::string &rundir)
 
std::string fmterror (int errnum)
 
int reopenFds ()
 
int handleSavedPfc (const std::filesystem::path &dest_path)
 
void waitForSignal ()
 
IEvtSelector * evtSelector ()
 

Protected Attributes

int m_nprocs {-1}
 Number of workers spawned by the master process. More...
 
int m_maxEvt {-1}
 Maximum number of events assigned to the job. More...
 
std::string m_subprocTopDir
 Top run directory for subprocesses. More...
 
std::string m_subprocDirPrefix
 For ex. "worker__". More...
 
std::string m_evtSelName
 Name of the event selector. More...
 
AthenaInterprocess::ProcessGroupm_processGroup {nullptr}
 
const AthenaInterprocess::IMPRunStopm_mpRunStop {nullptr}
 
ServiceHandle< IEventProcessor > m_evtProcessor
 
ServiceHandle< IAppMgrUI > m_appMgr
 
ServiceHandle< IFileMgr > m_fileMgr
 
ServiceHandle< IIoComponentMgr > m_ioMgr
 
SmartIF< IEvtSelector > m_evtSelector
 
std::string m_fileMgrLog
 
std::shared_ptr< AthenaInterprocess::FdsRegistrym_fdsRegistry
 
std::string m_randStr
 
Gaudi::Property< bool > m_isPileup {this, "IsPileup", false, "Flag for configuring PileUpEventLoopMgr"}
 

Private Member Functions

 SharedWriterTool ()
 
 SharedWriterTool (const SharedWriterTool &)
 
SharedWriterTooloperator= (const SharedWriterTool &)
 
int reopenFd (int fd, const std::string &name)
 

Private Attributes

Gaudi::Property< bool > m_nMotherProcess
 
Gaudi::Property< bool > m_debug
 
int m_rankId
 
AthenaInterprocess::SharedQueuem_sharedRankQueue
 
SmartIF< IConversionSvc > m_cnvSvc
 

Detailed Description

Definition at line 12 of file SharedWriterTool.h.

Member Enumeration Documentation

◆ ESRange_Status

enum AthenaMPToolBase::ESRange_Status
protectedinherited
Enumerator
ESRANGE_SUCCESS 
ESRANGE_NOTFOUND 
ESRANGE_SEEKFAILED 
ESRANGE_PROCFAILED 
ESRANGE_FILENOTMADE 
ESRANGE_BADINPFILE 

Definition at line 58 of file AthenaMPToolBase.h.

◆ Func_Flag

enum AthenaMPToolBase::Func_Flag
protectedinherited
Enumerator
FUNC_BOOTSTRAP 
FUNC_EXEC 
FUNC_FIN 

Definition at line 67 of file AthenaMPToolBase.h.

67  {
69  , FUNC_EXEC
70  , FUNC_FIN
71  };

Constructor & Destructor Documentation

◆ SharedWriterTool() [1/3]

SharedWriterTool::SharedWriterTool ( const std::string &  type,
const std::string &  name,
const IInterface *  parent 
)

Definition at line 19 of file SharedWriterTool.cxx.

23  , m_rankId(0)
24  , m_sharedRankQueue(nullptr)
25 {
26  m_subprocDirPrefix = "shared_writer";
27 }

◆ ~SharedWriterTool()

SharedWriterTool::~SharedWriterTool ( )
overridevirtual

Definition at line 29 of file SharedWriterTool.cxx.

30 {
31 }

◆ SharedWriterTool() [2/3]

SharedWriterTool::SharedWriterTool ( )
private

◆ SharedWriterTool() [3/3]

SharedWriterTool::SharedWriterTool ( const SharedWriterTool )
private

Member Function Documentation

◆ ATLAS_NOT_THREAD_SAFE() [1/4]

virtual StatusCode exec SharedWriterTool::ATLAS_NOT_THREAD_SAFE ( )
overridevirtual

◆ ATLAS_NOT_THREAD_SAFE() [2/4]

int mapAsyncFlag AthenaMPToolBase::ATLAS_NOT_THREAD_SAFE ( Func_Flag  flag,
pid_t  pid = 0 
)
protectedinherited

◆ ATLAS_NOT_THREAD_SAFE() [3/4]

virtual int makePool SharedWriterTool::ATLAS_NOT_THREAD_SAFE ( int  maxevt,
int  nprocs,
const std::string &  topdir 
)
overridevirtual

◆ ATLAS_NOT_THREAD_SAFE() [4/4]

virtual StatusCode wait_once AthenaMPToolBase::ATLAS_NOT_THREAD_SAFE ( pid_t pid)
overridevirtualinherited

◆ bootstrap_func() [1/2]

std::unique_ptr< AthenaInterprocess::ScheduledWork > SharedWriterTool::bootstrap_func ( )
overridevirtual

Definition at line 135 of file SharedWriterTool.cxx.

136 {
137  // It's possible to debug SharedWriter just like any other AthenaMP worker.
138  // The following procedure provides a minimal explanation on how this can be achieved:
139  //
140  // Terminal #1:
141  // * Run athena w/ debugging enabled, e.g. athena.py --debugWorker --stdcmalloc --nprocs=8 [...]
142  // * In this mode, workers will be stopped after fork(), waiting for SIGUSR1 to be resumed
143  // * Find the PID of the worker to be debugged (printed by the job in stdout)
144  //
145  // Terminal #2:
146  // * Attach gdb to the relevant worker, i.e. gdb python PID
147  // * Once the symbols are loaded, one can perform any gdb action such as setting breakpoints etc.
148  // * Once ready, send SIGUSR1 to the worker to resume work, i.e. signal SIGUSR1 (in gdb)
149  //
150  // Terminal #3:
151  // * Send SIGUSR1 to the remaining workers (easiest to use htop)
152  //
153  // However, note that sometimes Shared I/O infrastructure struggles with timing problems,
154  // such as server/client(s) starting/stopping too early/later. Debugging can change this
155  // behavior so please keep this in mind.
156  if(m_debug) waitForSignal();
157 
158  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
159  outwork->data = CxxUtils::xmalloc(sizeof(int));
160  *(int*)(outwork->data) = 1; // Error code: for now use 0 success, 1 failure
161  outwork->size = sizeof(int);
162 
163  // ...
164  // (possible) TODO: extend outwork with some error message, which will be eventually
165  // reported in the master proces
166  // ...
167 
168  // ________________________ Get RankID ________________________
169  //
171  ATH_MSG_ERROR("Unable to get rank ID!");
172  return outwork;
173  }
174  // Writer dir: mkdir
175  std::filesystem::path writer_rundir(m_subprocTopDir);
176  writer_rundir /= std::filesystem::path(m_subprocDirPrefix);
177 
178  if(mkdir(writer_rundir.string().c_str(),S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH)==-1) {
179  ATH_MSG_ERROR("Unable to make writer run directory: " << writer_rundir.string() << ". " << fmterror(errno));
180  return outwork;
181  }
182 
183  // __________ Redirect logs unless we want to attach debugger ____________
184  if(!m_debug) {
185  if(redirectLog(writer_rundir.string()))
186  return outwork;
187 
188  ATH_MSG_INFO("Logs redirected in the AthenaMP Shared Writer PID=" << getpid());
189  }
190 
191  // Update Io Registry
192  if(updateIoReg(writer_rundir.string()))
193  return outwork;
194 
195  ATH_MSG_INFO("Io registry updated in the AthenaMP Shared Writer PID=" << getpid());
196 
197  // _______________________ Handle saved PFC (if any) ______________________
198  std::filesystem::path abs_writer_rundir = std::filesystem::absolute(writer_rundir);
199  if(handleSavedPfc(abs_writer_rundir))
200  return outwork;
201 
202  // Reopen file descriptors
203  if(reopenFds())
204  return outwork;
205 
206  ATH_MSG_INFO("File descriptors re-opened in the AthenaMP Shared Writer PID=" << getpid());
207 
208  // Try to initialize AthenaRootSharedWriterSvc early on
209  SmartIF<IAthenaSharedWriterSvc> sharedWriterSvc(serviceLocator()->service("AthenaRootSharedWriterSvc"));
210  if(!sharedWriterSvc) {
211  ATH_MSG_WARNING("Error retrieving AthenaRootSharedWriterSvc from SharedWriterTool::bootstrap_func()");
212  }
213 
214  // Use IDataShare to make ConversionSvc a Share Server
215  SmartIF<IDataShare> cnvSvc(m_cnvSvc);
216  if (!cnvSvc || !cnvSvc->makeServer(-m_nprocs - 1 - 1024 * m_rankId).isSuccess()) {
217  ATH_MSG_ERROR("Failed to make the conversion service a share server");
218  return outwork;
219  }
220  else {
221  ATH_MSG_DEBUG("Successfully made the conversion service a share server");
222  }
223 
224  // ________________________ I/O reinit ________________________
225  if(!m_ioMgr->io_reinitialize().isSuccess()) {
226  ATH_MSG_ERROR("Failed to reinitialize I/O");
227  return outwork;
228  } else {
229  ATH_MSG_DEBUG("Successfully reinitialized I/O");
230  }
231 
232  // Writer dir: chdir
233  if(chdir(writer_rundir.string().c_str())==-1) {
234  ATH_MSG_ERROR("Failed to chdir to " << writer_rundir.string());
235  return outwork;
236  }
237 
238  // Declare success and return
239  *(int*)(outwork->data) = 0;
240  return outwork;
241 }

◆ bootstrap_func() [2/2]

virtual std::unique_ptr<AthenaInterprocess::ScheduledWork> virtual operator () ATLAS_NOT_THREAD_SAFE ( const AthenaInterprocess std::unique_ptr<AthenaInterprocess::ScheduledWork> AthenaMPToolBase::bootstrap_func ( )
pure virtualinherited

◆ evtSelector()

IEvtSelector* AthenaMPToolBase::evtSelector ( )
inlineprotectedinherited

Definition at line 83 of file AthenaMPToolBase.h.

83 { return m_evtSelector; }

◆ exec_func()

std::unique_ptr< AthenaInterprocess::ScheduledWork > SharedWriterTool::exec_func ( )
overridevirtual

Implements AthenaMPToolBase.

Definition at line 243 of file SharedWriterTool.cxx.

244 {
245  ATH_MSG_INFO("Exec function in the AthenaMP Shared Writer PID=" << getpid());
246  bool all_ok=true;
247 
248  SmartIF<IAthenaSharedWriterSvc> sharedWriterSvc(serviceLocator()->service("AthenaRootSharedWriterSvc"));
249  if(!sharedWriterSvc) {
250  ATH_MSG_ERROR("Error retrieving AthenaRootSharedWriterSvc");
251  all_ok=false;
252  } else if(!sharedWriterSvc->share(m_nprocs, m_nMotherProcess.value()).isSuccess()) {
253  ATH_MSG_ERROR("Exec function could not share data");
254  all_ok=false;
255  }
256  AthCnvSvc* cnvSvc = dynamic_cast<AthCnvSvc*>(m_cnvSvc.get());
257  if (cnvSvc == 0 || !cnvSvc->disconnectOutput("").isSuccess()) {
258  ATH_MSG_ERROR("Exec function could not disconnectOutput");
259  all_ok=false;
260  }
261 
262  if(m_appMgr->stop().isFailure()) {
263  ATH_MSG_ERROR("Unable to stop AppMgr");
264  all_ok=false;
265  }
266  else {
267  if(m_appMgr->finalize().isFailure()) {
268  std::cerr << "Unable to finalize AppMgr" << std::endl;
269  all_ok=false;
270  }
271  }
272 
273  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
274  outwork->data = CxxUtils::xmalloc(sizeof(int));
275  *(int*)(outwork->data) = (all_ok?0:1); // Error code: for now use 0 success, 1 failure
276  outwork->size = sizeof(int);
277 
278  // ...
279  // (possible) TODO: extend outwork with some error message, which will be eventually
280  // reported in the master proces
281  // ...
282  return outwork;
283 }

◆ fin_func()

std::unique_ptr< AthenaInterprocess::ScheduledWork > SharedWriterTool::fin_func ( )
overridevirtual

Implements AthenaMPToolBase.

Definition at line 285 of file SharedWriterTool.cxx.

286 {
287  // Dummy
288  std::unique_ptr<AthenaInterprocess::ScheduledWork> outwork(new AthenaInterprocess::ScheduledWork);
289  outwork->data = CxxUtils::xmalloc(sizeof(int));
290  *(int*)(outwork->data) = 0; // Error code: for now use 0 success, 1 failure
291  outwork->size = sizeof(int);
292  return outwork;
293 }

◆ finalize()

StatusCode SharedWriterTool::finalize ( )
overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 44 of file SharedWriterTool.cxx.

45 {
46  ATH_MSG_DEBUG("In finalize");
47 
48  delete m_sharedRankQueue;
49  return StatusCode::SUCCESS;
50 }

◆ fmterror()

std::string AthenaMPToolBase::fmterror ( int  errnum)
protectedinherited

Definition at line 333 of file AthenaMPToolBase.cxx.

334 {
335  char buf[256];
336  strerror_r(errnum, buf, sizeof(buf));
337  return std::string(buf);
338 }

◆ generateOutputReport()

AthenaMP::AllWorkerOutputs_ptr SharedWriterTool::generateOutputReport ( )
overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 129 of file SharedWriterTool.cxx.

130 {
132  return jobOutputs;
133 }

◆ handleSavedPfc()

int AthenaMPToolBase::handleSavedPfc ( const std::filesystem::path &  dest_path)
protectedinherited

Definition at line 396 of file AthenaMPToolBase.cxx.

397 {
398  if(std::filesystem::is_regular_file("PoolFileCatalog.xml.AthenaMP-saved"))
399  COPY_FILE_HACK("PoolFileCatalog.xml.AthenaMP-saved",dest_path.string()+"/PoolFileCatalog.xml");
400  return 0;
401 }

◆ initialize()

StatusCode SharedWriterTool::initialize ( )
overridevirtual

Reimplemented from AthenaMPToolBase.

Definition at line 33 of file SharedWriterTool.cxx.

34 {
35  ATH_MSG_DEBUG("In initialize");
36 
38  m_cnvSvc = serviceLocator()->service("AthenaPoolSharedIOCnvSvc");
39  ATH_CHECK(m_cnvSvc.isValid());
40 
41  return StatusCode::SUCCESS;
42 }

◆ killChildren()

void AthenaMPToolBase::killChildren ( )
overridevirtualinherited

Definition at line 201 of file AthenaMPToolBase.cxx.

202 {
204  kill(child.getProcessID(),SIGKILL);
205  }
206 }

◆ operator()

virtual std::unique_ptr<ScheduledWork> AthenaInterprocess::IMessageDecoder::operator ( ) const &
pure virtualinherited

◆ operator=()

◆ redirectLog()

int AthenaMPToolBase::redirectLog ( const std::string &  rundir,
bool  addTimeStamp = true 
)
protectedinherited

Definition at line 269 of file AthenaMPToolBase.cxx.

270 {
271  // Redirect both stdout and stderr to the same file AthenaMP.log
272  int dup2result1(0), dup2result2(0);
273 
274  int newout = open(std::string(rundir+"/AthenaMP.log").c_str(),O_CREAT | O_RDWR, S_IRWXU|S_IRGRP|S_IXGRP|S_IROTH|S_IXOTH);
275  if(newout==-1) {
276  ATH_MSG_ERROR("Unable to open log file in the run directory. " << fmterror(errno));
277  return -1;
278  }
279  dup2result1 = dup2(newout, STDOUT_FILENO);
280  dup2result2 = dup2(newout, STDERR_FILENO);
281  TEMP_FAILURE_RETRY(close(newout));
282  if(dup2result1==-1) {
283  ATH_MSG_ERROR("Unable to redirect standard output. " << fmterror(errno));
284  return -1;
285  }
286  if(dup2result2==-1) {
287  ATH_MSG_ERROR("Unable to redirect standard error. " << fmterror(errno));
288  return -1;
289  }
290 
291  if(addTimeStamp) {
292  SmartIF<IProperty> propertyServer(msgSvc());
293  if(propertyServer==0) {
294  ATH_MSG_ERROR("Unable to cast message svc to IProperty");
295  return -1;
296  }
297 
298  std::string propertyName("Format");
299  std::string oldFormat("");
300  StringProperty formatProp(propertyName,oldFormat);
301  StatusCode sc = propertyServer->getProperty(&formatProp);
302  if(sc.isFailure()) {
303  ATH_MSG_WARNING("Message Service does not have Format property");
304  }
305  else {
306  oldFormat = formatProp.value();
307  if(oldFormat.find("%t")==std::string::npos) {
308  // Add time stamps
309  std::string newFormat("%t " + oldFormat);
310  StringProperty newFormatProp(std::move(propertyName),newFormat);
311  ATH_CHECK(propertyServer->setProperty(newFormatProp), -1);
312  }
313  else {
314  ATH_MSG_DEBUG("MsgSvc format already contains timestamps. Nothing to be done");
315  }
316  }
317  }
318 
319  return 0;
320 }

◆ reopenFd()

int AthenaMPToolBase::reopenFd ( int  fd,
const std::string &  name 
)
privateinherited

Definition at line 419 of file AthenaMPToolBase.cxx.

420 {
421  ATH_MSG_DEBUG("Attempting to reopen descriptor for " << name);
422  int old_openflags = fcntl(fd,F_GETFL,0);
423  switch(old_openflags & O_ACCMODE) {
424  case O_RDONLY: {
425  ATH_MSG_DEBUG("The File Access Mode is RDONLY");
426  break;
427  }
428  case O_WRONLY: {
429  ATH_MSG_DEBUG("The File Access Mode is WRONLY");
430  break;
431  }
432  case O_RDWR: {
433  ATH_MSG_DEBUG("The File Access Mode is RDWR");
434  break;
435  }
436  }
437 
438  int old_descflags = fcntl(fd,F_GETFD,0);
439  off_t oldpos = lseek(fd,0,SEEK_CUR);
440  if(oldpos==-1) {
441  if(errno==ESPIPE) {
442  ATH_MSG_WARNING("Dealing with PIPE. Skipping ... (FIXME!)");
443  }
444  else {
445  ATH_MSG_ERROR("When re-opening file descriptors lseek failed on " << name << ". " << fmterror(errno));
446  return -1;
447  }
448  }
449  else {
450  Io::Fd newfd = open(name.c_str(),old_openflags);
451  if(newfd==-1) {
452  ATH_MSG_ERROR("When re-opening file descriptors unable to open " << name << " for reading. " << fmterror(errno));
453  return -1;
454  }
455  if(lseek(newfd,oldpos,SEEK_SET)==-1){
456  ATH_MSG_ERROR("When re-opening file descriptors lseek failed on the newly opened " << name << ". " << fmterror(errno));
457  TEMP_FAILURE_RETRY(close(newfd));
458  return -1;
459  }
460  TEMP_FAILURE_RETRY(close(fd));
461  if(dup2(newfd,fd)==-1) {
462  ATH_MSG_ERROR("When re-opening file descriptors unable to duplicate descriptor for " << name << ". " << fmterror(errno));
463  TEMP_FAILURE_RETRY(close(newfd));
464  return -1;
465  }
466  if(fcntl(fd,F_SETFD,old_descflags)==-1) {
467  ATH_MSG_ERROR("When re-opening file descriptors unable to set descriptor flags for " << name << ". " << fmterror(errno));
468  TEMP_FAILURE_RETRY(close(newfd));
469  return -1;
470  }
471  TEMP_FAILURE_RETRY(close(newfd));
472  }
473  return 0;
474 }

◆ reopenFds()

int AthenaMPToolBase::reopenFds ( )
protectedinherited

Definition at line 340 of file AthenaMPToolBase.cxx.

341 {
342  // Reopen file descriptors.
343  // First go over all open files, which have been registered with the FileMgr
344  // Then also check the FdsRegistry, in case it contains some files not registered with the FileMgr
345  std::set<int> fdLog;
346 
347  // Query the FileMgr contents
348  std::vector<const Io::FileAttr*> filemgrFiles;
349  std::vector<const Io::FileAttr*>::const_iterator itFile;
350  unsigned filenum = m_fileMgr->getFiles(filemgrFiles); // Get attributes for open files only. We don't care about closed ones at this point
351  if(filenum!=filemgrFiles.size())
352  ATH_MSG_WARNING("getFiles returned " << filenum << " while vector size is " << filemgrFiles.size());
353 
354  for(itFile=filemgrFiles.begin();itFile!=filemgrFiles.end();++itFile) {
355  ATH_MSG_DEBUG("* " << **itFile);
356  const std::string& filename = (**itFile).name();
357  Io::Fd fd = (**itFile).fd();
358 
359  if(fd==-1) {
360  // It is legal to have fd=-1 for remote inputs
361  // On the other hand, these inputs should not remain open after fork. The issue being tracked at ATEAM-434.
362  // So, this hopefully is a temporary patch
363  ATH_MSG_WARNING("FD=-1 detected on an open file retrieved from FileMgr. Skip FD reopening. File name: " << filename);
364  continue;
365  }
366 
367  if(reopenFd(fd,filename))
368  return -1;
369 
370  fdLog.insert(fd);
371  }
372 
373  // Check the FdsRegistry
374  for(const AthenaInterprocess::FdsRegistryEntry& regEntry : *m_fdsRegistry) {
375  if(fdLog.find(regEntry.fd)!=fdLog.end()) {
376  ATH_MSG_DEBUG("The file from FdsRegistry " << regEntry.name << " was registered with FileMgr. Skip reopening");
377  }
378  else {
379  ATH_MSG_WARNING("The file " << regEntry.name << " has not been registered with the FileMgr!");
380 
381  if(regEntry.fd==-1) {
382  // Same protection as the one above
383  ATH_MSG_WARNING("FD=-1 detected on an open file retrieved from FD Registry. Skip FD reopening. File name: " << regEntry.name);
384  continue;
385  }
386 
387  if(reopenFd(regEntry.fd,regEntry.name))
388  return -1;
389 
390  fdLog.insert(regEntry.fd);
391  }
392  }
393  return 0;
394 }

◆ reportSubprocessStatuses()

void AthenaMPToolBase::reportSubprocessStatuses ( )
overridevirtualinherited

Reimplemented in EvtRangeProcessor, SharedEvtQueueConsumer, and SharedHiveEvtQueueConsumer.

Definition at line 111 of file AthenaMPToolBase.cxx.

112 {
113  ATH_MSG_INFO("Statuses of sub-processes");
114  const std::vector<AthenaInterprocess::ProcessStatus>& statuses = m_processGroup->getStatuses();
115  for(size_t i=0; i<statuses.size(); ++i)
116  ATH_MSG_INFO("*** Process PID=" << statuses[i].pid << ". Status " << ((statuses[i].exitcode)?"FAILURE":"SUCCESS"));
117 }

◆ setMaxEvt()

virtual void AthenaMPToolBase::setMaxEvt ( int  maxEvt)
inlineoverridevirtualinherited

Definition at line 44 of file AthenaMPToolBase.h.

44 {m_maxEvt=maxEvt;}

◆ setMPRunStop()

virtual void AthenaMPToolBase::setMPRunStop ( const AthenaInterprocess::IMPRunStop runStop)
inlineoverridevirtualinherited

Definition at line 45 of file AthenaMPToolBase.h.

45 {m_mpRunStop=runStop;}

◆ setRandString()

void AthenaMPToolBase::setRandString ( const std::string &  randStr)
overridevirtualinherited

Definition at line 196 of file AthenaMPToolBase.cxx.

197 {
198  m_randStr = randStr;
199 }

◆ subProcessLogs()

void SharedWriterTool::subProcessLogs ( std::vector< std::string > &  filenames)
overridevirtual

Definition at line 121 of file SharedWriterTool.cxx.

122 {
123  filenames.clear();
124  std::filesystem::path writer_rundir(m_subprocTopDir);
125  writer_rundir/= std::filesystem::path(m_subprocDirPrefix);
126  filenames.push_back(writer_rundir.string()+std::string("/AthenaMP.log"));
127 }

◆ updateIoReg()

int AthenaMPToolBase::updateIoReg ( const std::string &  rundir)
protectedinherited

Definition at line 322 of file AthenaMPToolBase.cxx.

323 {
324  ATH_CHECK(m_ioMgr.retrieve(), -1);
325 
326  // update the IoRegistry for the new workdir - make sure we use absolute path
328  ATH_CHECK(m_ioMgr->io_update_all(abs_rundir.string()), -1);
329 
330  return 0;
331 }

◆ useFdsRegistry()

void AthenaMPToolBase::useFdsRegistry ( std::shared_ptr< AthenaInterprocess::FdsRegistry registry)
overridevirtualinherited

Definition at line 191 of file AthenaMPToolBase.cxx.

192 {
193  m_fdsRegistry = std::move(registry);
194 }

◆ waitForSignal()

void AthenaMPToolBase::waitForSignal ( )
protectedinherited

Definition at line 403 of file AthenaMPToolBase.cxx.

404 {
405  ATH_MSG_INFO("Bootstrap worker PID " << getpid() << " - waiting for SIGUSR1");
406  sigset_t mask, oldmask;
407 
409 
410  sigemptyset (&mask);
411  sigaddset (&mask, SIGUSR1);
412 
413  sigprocmask (SIG_BLOCK, &mask, &oldmask);
415  sigsuspend (&oldmask);
416  sigprocmask (SIG_UNBLOCK, &mask, NULL);
417 }

Member Data Documentation

◆ m_appMgr

ServiceHandle<IAppMgrUI> AthenaMPToolBase::m_appMgr
protectedinherited

Definition at line 95 of file AthenaMPToolBase.h.

◆ m_cnvSvc

SmartIF<IConversionSvc> SharedWriterTool::m_cnvSvc
private

Definition at line 52 of file SharedWriterTool.h.

◆ m_debug

Gaudi::Property<bool> SharedWriterTool::m_debug
private
Initial value:
{
this, "Debug", false,
"Are we running in debug mode? The default is false"}

Definition at line 45 of file SharedWriterTool.h.

◆ m_evtProcessor

ServiceHandle<IEventProcessor> AthenaMPToolBase::m_evtProcessor
protectedinherited

Definition at line 94 of file AthenaMPToolBase.h.

◆ m_evtSelector

SmartIF<IEvtSelector> AthenaMPToolBase::m_evtSelector
protectedinherited

Definition at line 98 of file AthenaMPToolBase.h.

◆ m_evtSelName

std::string AthenaMPToolBase::m_evtSelName
protectedinherited

Name of the event selector.

Definition at line 89 of file AthenaMPToolBase.h.

◆ m_fdsRegistry

std::shared_ptr<AthenaInterprocess::FdsRegistry> AthenaMPToolBase::m_fdsRegistry
protectedinherited

Definition at line 100 of file AthenaMPToolBase.h.

◆ m_fileMgr

ServiceHandle<IFileMgr> AthenaMPToolBase::m_fileMgr
protectedinherited

Definition at line 96 of file AthenaMPToolBase.h.

◆ m_fileMgrLog

std::string AthenaMPToolBase::m_fileMgrLog
protectedinherited

Definition at line 99 of file AthenaMPToolBase.h.

◆ m_ioMgr

ServiceHandle<IIoComponentMgr> AthenaMPToolBase::m_ioMgr
protectedinherited

Definition at line 97 of file AthenaMPToolBase.h.

◆ m_isPileup

Gaudi::Property<bool> AthenaMPToolBase::m_isPileup {this, "IsPileup", false, "Flag for configuring PileUpEventLoopMgr"}
protectedinherited

Definition at line 103 of file AthenaMPToolBase.h.

◆ m_maxEvt

int AthenaMPToolBase::m_maxEvt {-1}
protectedinherited

Maximum number of events assigned to the job.

Definition at line 86 of file AthenaMPToolBase.h.

◆ m_mpRunStop

const AthenaInterprocess::IMPRunStop* AthenaMPToolBase::m_mpRunStop {nullptr}
protectedinherited

Definition at line 92 of file AthenaMPToolBase.h.

◆ m_nMotherProcess

Gaudi::Property<bool> SharedWriterTool::m_nMotherProcess
private
Initial value:
{
this, "MotherProcess", false,
"Expect mother process to write event data. The default is false."}

Definition at line 42 of file SharedWriterTool.h.

◆ m_nprocs

int AthenaMPToolBase::m_nprocs {-1}
protectedinherited

Number of workers spawned by the master process.

Definition at line 85 of file AthenaMPToolBase.h.

◆ m_processGroup

AthenaInterprocess::ProcessGroup* AthenaMPToolBase::m_processGroup {nullptr}
protectedinherited

Definition at line 91 of file AthenaMPToolBase.h.

◆ m_randStr

std::string AthenaMPToolBase::m_randStr
protectedinherited

Definition at line 101 of file AthenaMPToolBase.h.

◆ m_rankId

int SharedWriterTool::m_rankId
private

Definition at line 49 of file SharedWriterTool.h.

◆ m_sharedRankQueue

AthenaInterprocess::SharedQueue* SharedWriterTool::m_sharedRankQueue
private

Definition at line 51 of file SharedWriterTool.h.

◆ m_subprocDirPrefix

std::string AthenaMPToolBase::m_subprocDirPrefix
protectedinherited

For ex. "worker__".

Definition at line 88 of file AthenaMPToolBase.h.

◆ m_subprocTopDir

std::string AthenaMPToolBase::m_subprocTopDir
protectedinherited

Top run directory for subprocesses.

Definition at line 87 of file AthenaMPToolBase.h.


The documentation for this class was generated from the following files:
python.Dso.registry
registry
Definition: Control/AthenaServices/python/Dso.py:158
python.DQPostProcessMod.rundir
def rundir(fname)
Definition: DQPostProcessMod.py:115
AthenaInterprocess::ProcessGroup::getStatuses
const std::vector< ProcessStatus > & getStatuses() const
Definition: ProcessGroup.cxx:204
SharedWriterTool::m_sharedRankQueue
AthenaInterprocess::SharedQueue * m_sharedRankQueue
Definition: SharedWriterTool.h:51
AthenaMPToolBase::waitForSignal
void waitForSignal()
Definition: AthenaMPToolBase.cxx:403
athena.path
path
python interpreter configuration --------------------------------------—
Definition: athena.py:128
AthenaMPToolBase::ESRANGE_BADINPFILE
@ ESRANGE_BADINPFILE
Definition: AthenaMPToolBase.h:64
ATH_MSG_INFO
#define ATH_MSG_INFO(x)
Definition: AthMsgStreamMacros.h:31
plot_material.mkdir
def mkdir(path, recursive=True)
Definition: plot_material.py:15
AthenaMPToolBase::m_processGroup
AthenaInterprocess::ProcessGroup * m_processGroup
Definition: AthenaMPToolBase.h:91
AthenaInterprocess::ScheduledWork
Definition: IMessageDecoder.h:12
AthenaMPToolBase::m_randStr
std::string m_randStr
Definition: AthenaMPToolBase.h:101
AthenaMPToolBase::m_nprocs
int m_nprocs
Number of workers spawned by the master process.
Definition: AthenaMPToolBase.h:85
AthenaMPToolBase::FUNC_FIN
@ FUNC_FIN
Definition: AthenaMPToolBase.h:70
AthenaInterprocess::ProcessGroup::getChildren
const std::vector< Process > & getChildren() const
Definition: ProcessGroup.cxx:197
plotting.yearwise_luminosity.absolute
absolute
Definition: yearwise_luminosity.py:29
AthenaMPToolBase::fmterror
std::string fmterror(int errnum)
Definition: AthenaMPToolBase.cxx:333
DeMoUpdate.statuses
list statuses
Definition: DeMoUpdate.py:568
athena.exitcode
int exitcode
Definition: athena.py:161
AthenaMPToolBase::ESRANGE_SEEKFAILED
@ ESRANGE_SEEKFAILED
Definition: AthenaMPToolBase.h:61
SharedWriterTool::m_rankId
int m_rankId
Definition: SharedWriterTool.h:49
python.CaloAddPedShiftConfig.type
type
Definition: CaloAddPedShiftConfig.py:42
python.utils.AtlRunQueryLookup.mask
string mask
Definition: AtlRunQueryLookup.py:459
Fd
IIoSvc::Fd Fd
Definition: IoSvc.cxx:22
sigaddset
#define sigaddset(x, y)
Definition: SealSignal.h:84
AthenaPoolTestRead.sc
sc
Definition: AthenaPoolTestRead.py:27
AthenaMPToolBase::FUNC_EXEC
@ FUNC_EXEC
Definition: AthenaMPToolBase.h:69
COPY_FILE_HACK
#define COPY_FILE_HACK(_src, _dest)
Definition: copy_file_icc_hack.h:15
sigset_t
int sigset_t
Definition: SealSignal.h:80
AthenaMPToolBase::m_evtSelector
SmartIF< IEvtSelector > m_evtSelector
Definition: AthenaMPToolBase.h:98
AthenaMPToolBase::m_appMgr
ServiceHandle< IAppMgrUI > m_appMgr
Definition: AthenaMPToolBase.h:95
ATH_MSG_ERROR
#define ATH_MSG_ERROR(x)
Definition: AthMsgStreamMacros.h:33
StdJOSetup.msgSvc
msgSvc
Provide convenience handles for various services.
Definition: StdJOSetup.py:36
lumiFormat.i
int i
Definition: lumiFormat.py:85
python.DecayParser.buf
buf
print ("=> [%s]"cmd)
Definition: DecayParser.py:26
EL::StatusCode
::StatusCode StatusCode
StatusCode definition for legacy code.
Definition: PhysicsAnalysis/D3PDTools/EventLoop/EventLoop/StatusCode.h:22
ATH_MSG_DEBUG
#define ATH_MSG_DEBUG(x)
Definition: AthMsgStreamMacros.h:29
AthenaMPToolBase::reopenFd
int reopenFd(int fd, const std::string &name)
Definition: AthenaMPToolBase.cxx:419
AthenaMPToolBase::AthenaMPToolBase
AthenaMPToolBase()
ParticleGun_EoverP_Config.pid
pid
Definition: ParticleGun_EoverP_Config.py:62
test_pyathena.parent
parent
Definition: test_pyathena.py:15
AthenaMPToolBase_d::pauseForDebug
void pauseForDebug(int)
Definition: AthenaMPToolBase.cxx:28
AthenaMPToolBase::ESRANGE_SUCCESS
@ ESRANGE_SUCCESS
Definition: AthenaMPToolBase.h:59
ATH_CHECK
#define ATH_CHECK
Definition: AthCheckMacros.h:40
AthenaMPToolBase::updateIoReg
int updateIoReg(const std::string &rundir)
Definition: AthenaMPToolBase.cxx:322
AthenaMPToolBase::initialize
virtual StatusCode initialize() override
Definition: AthenaMPToolBase.cxx:48
AthenaInterprocess::SharedQueue::receive_basic
bool receive_basic(T &)
Definition: SharedQueue.h:124
AthenaMPToolBase::m_mpRunStop
const AthenaInterprocess::IMPRunStop * m_mpRunStop
Definition: AthenaMPToolBase.h:92
SharedWriterTool::m_nMotherProcess
Gaudi::Property< bool > m_nMotherProcess
Definition: SharedWriterTool.h:42
ReadFromCoolCompare.fd
fd
Definition: ReadFromCoolCompare.py:196
SharedWriterTool::m_debug
Gaudi::Property< bool > m_debug
Definition: SharedWriterTool.h:45
AthenaMPToolBase::ESRANGE_PROCFAILED
@ ESRANGE_PROCFAILED
Definition: AthenaMPToolBase.h:62
name
std::string name
Definition: Control/AthContainers/Root/debug.cxx:240
AthenaMPToolBase::m_ioMgr
ServiceHandle< IIoComponentMgr > m_ioMgr
Definition: AthenaMPToolBase.h:97
AthenaMPToolBase::redirectLog
int redirectLog(const std::string &rundir, bool addTimeStamp=true)
Definition: AthenaMPToolBase.cxx:269
AthenaInterprocess::Process
Definition: Process.h:17
Cut::signal
@ signal
Definition: SUSYToolsAlg.cxx:67
grepfile.filenames
list filenames
Definition: grepfile.py:34
Trk::open
@ open
Definition: BinningType.h:40
AthenaMPToolBase::m_maxEvt
int m_maxEvt
Maximum number of events assigned to the job.
Definition: AthenaMPToolBase.h:86
python.CaloAddPedShiftConfig.int
int
Definition: CaloAddPedShiftConfig.py:45
ATH_MSG_WARNING
#define ATH_MSG_WARNING(x)
Definition: AthMsgStreamMacros.h:32
AthenaInterprocess::FdsRegistryEntry
Definition: FdsRegistry.h:13
AthenaMP::AllWorkerOutputs_ptr
std::unique_ptr< AllWorkerOutputs > AllWorkerOutputs_ptr
Definition: IAthenaMPTool.h:32
CaloCellTimeCorrFiller.filename
filename
Definition: CaloCellTimeCorrFiller.py:23
AthCnvSvc::disconnectOutput
virtual StatusCode disconnectOutput(const std::string &output)
Disconnect output files from the service.
Definition: AthCnvSvc.cxx:408
AthenaMPToolBase::m_subprocDirPrefix
std::string m_subprocDirPrefix
For ex. "worker__".
Definition: AthenaMPToolBase.h:88
AthenaMPToolBase::m_subprocTopDir
std::string m_subprocTopDir
Top run directory for subprocesses.
Definition: AthenaMPToolBase.h:87
AthenaMPToolBase::ESRANGE_NOTFOUND
@ ESRANGE_NOTFOUND
Definition: AthenaMPToolBase.h:60
CxxUtils::xmalloc
void * xmalloc(size_t size)
Trapping version of malloc.
Definition: xmalloc.cxx:31
SharedWriterTool::m_cnvSvc
SmartIF< IConversionSvc > m_cnvSvc
Definition: SharedWriterTool.h:52
AthenaMPToolBase::handleSavedPfc
int handleSavedPfc(const std::filesystem::path &dest_path)
Definition: AthenaMPToolBase.cxx:396
AthenaMPToolBase::m_fileMgr
ServiceHandle< IFileMgr > m_fileMgr
Definition: AthenaMPToolBase.h:96
AthenaMP::AllWorkerOutputs
std::map< std::string, SingleWorkerOutputs > AllWorkerOutputs
Definition: IAthenaMPTool.h:29
AthenaMPToolBase::FUNC_BOOTSTRAP
@ FUNC_BOOTSTRAP
Definition: AthenaMPToolBase.h:68
AthCnvSvc
Definition: AthCnvSvc.h:66
AthenaMPToolBase::reopenFds
int reopenFds()
Definition: AthenaMPToolBase.cxx:340
sigemptyset
#define sigemptyset(x)
Definition: SealSignal.h:82
AthenaMPToolBase::ESRANGE_FILENOTMADE
@ ESRANGE_FILENOTMADE
Definition: AthenaMPToolBase.h:63
AthenaMPToolBase::m_fdsRegistry
std::shared_ptr< AthenaInterprocess::FdsRegistry > m_fdsRegistry
Definition: AthenaMPToolBase.h:100
AthenaMPToolBase_d::sig_done
std::atomic< bool > sig_done
Definition: AthenaMPToolBase.cxx:27