Bugfix for the Event Service

Bugfix in the mechanism of handling failed MP workers in the Event Service.
This patch covers the case when a worker fails to transition from the event
processing state into the finalization state after receiving "No more events"
message from the range scatterer. The idea here is that instead of releasing
a fixed number of workers (nprocs), the scatterer will keep releasing the
workers until it gets a signal that all of them have finished.

Control/AthenaMPTools/src/EvtRangeProcessor.cxx

 /*
-  Copyright (C) 2002-2017 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
 */
 
 #include "EvtRangeProcessor.h"
@@ -47,6 +47,7 @@ EvtRangeProcessor::EvtRangeProcessor(const std::string& type
   , m_isPileup(false)
   , m_rankId(-1)
   , m_nEventsBeforeFork(0)
+  , m_activeWorkers(0)
   , m_inpFile("")
   , m_chronoStatSvc("ChronoStatSvc", name)
   , m_incidentSvc("IncidentSvc", name)
@@ -120,6 +121,7 @@ int EvtRangeProcessor::makePool(int, int nprocs, const std::string& topdir)
   }
 
   m_nprocs = (nprocs==-1?sysconf(_SC_NPROCESSORS_ONLN):nprocs);
+  m_activeWorkers = m_nprocs;
   m_subprocTopDir = topdir;
 
   // Create rank queue and fill it
@@ -236,6 +238,15 @@ StatusCode EvtRangeProcessor::wait_once(pid_t& pid)
 	return StatusCode::FAILURE;
       }
     }
+    else {
+      // The worker finished successfully and it was the last worker. Release the Event Range Scatterer
+      if(--m_activeWorkers==0
+	 && !m_sharedFailedPidQueue->send_basic<pid_t>(-1)) {
+	// To Do: how to report this error to the pilot?
+	ATH_MSG_ERROR("Failed to release the Event Range Scatterer");
+	return StatusCode::FAILURE;
+      }
+    }
 
     // Erase the pid from m_procStates map
     m_procStates.erase(itProcState);

Control/AthenaMPTools/src/EvtRangeProcessor.h

 /*
-  Copyright (C) 2002-2019 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
 */
 
 #ifndef ATHENAMPTOOLS_EVTRANGEPROCESSOR_H
@@ -66,6 +66,7 @@ class EvtRangeProcessor final : public AthenaMPToolBase
   bool m_isPileup;        // Are we doing pile-up digitization?
   int  m_rankId;          // Each worker has its own unique RankID from the range (0,...,m_nprocs-1) 
   int  m_nEventsBeforeFork;
+  int  m_activeWorkers;   // Keep track of the number of workers
   std::string m_inpFile;  // Cached name of the input file. To avoid reopening
 
   ServiceHandle<IChronoStatSvc>     m_chronoStatSvc;

Control/AthenaMPTools/src/EvtRangeScatterer.cxx

 /*
-  Copyright (C) 2002-2017 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
 */
 
 #include "EvtRangeScatterer.h"
@@ -23,8 +23,8 @@
 #include <cstdlib>
 
 EvtRangeScatterer::EvtRangeScatterer(const std::string& type
-			       , const std::string& name
-			       , const IInterface* parent)
+				     , const std::string& name
+				     , const IInterface* parent)
   : AthenaMPToolBase(type,name,parent)
   , m_processorChannel("")
   , m_eventRangeChannel("")
@@ -212,6 +212,7 @@ std::unique_ptr<AthenaInterprocess::ScheduledWork> EvtRangeScatterer::exec_func(
   std::string strReady("Ready for events");
   std::string strStopProcessing("No more events");
   std::string processorWaitRequest("");
+  int workerPid{-1};
 
   ATH_MSG_INFO("Starting main loop");
 
@@ -225,6 +226,12 @@ std::unique_ptr<AthenaInterprocess::ScheduledWork> EvtRangeScatterer::exec_func(
 	usleep(1000);
       }
       ATH_MSG_INFO("One of the processors is ready for the next range");
+      // Get PID from the request and Update m_pid2RangeID
+      workerPid = std::atoi(processorWaitRequest.c_str());
+      auto it = m_pid2RangeID.find(workerPid);
+      if(it!=m_pid2RangeID.end()) {
+	m_pid2RangeID.erase(it);
+      }
     }    
 
     // Signal the Pilot that AthenaMP is ready for event processing
@@ -239,7 +246,7 @@ std::unique_ptr<AthenaInterprocess::ScheduledWork> EvtRangeScatterer::exec_func(
       eventRange = eventRange.substr(0,carRet);
 
     // Break the loop if no more ranges are expected
-    if(eventRange.compare(strStopProcessing)==0) {
+    if(eventRange.find(strStopProcessing)!=std::string::npos) {
       ATH_MSG_INFO("Stopped the loop. Last message from the Event Range Channel: " << eventRange);
       break;
     }
@@ -335,6 +342,12 @@ std::unique_ptr<AthenaInterprocess::ScheduledWork> EvtRangeScatterer::exec_func(
 	usleep(1000);
       }
       ATH_MSG_INFO("One of the processors is ready for the next range");
+      // Get PID from the request and Update m_pid2RangeID
+      workerPid = std::atoi(processorWaitRequest.c_str());
+      auto it = m_pid2RangeID.find(workerPid);
+      if(it!=m_pid2RangeID.end()) {
+	m_pid2RangeID.erase(it);
+      }
     }
     
     // Send to the Processor: RangeID,evtToken[,evtToken] 
@@ -344,8 +357,7 @@ std::unique_ptr<AthenaInterprocess::ScheduledWork> EvtRangeScatterer::exec_func(
     procReportPending++;
 
     // Get PID from the request and Update m_pid2RangeID
-    int pid = std::atoi(processorWaitRequest.c_str());
-    m_pid2RangeID[pid] = rangeID;
+    m_pid2RangeID[workerPid] = rangeID;
     processorWaitRequest.clear();
 
     ATH_MSG_INFO("Sent response to the processor : " << message2ProcessorStr);
@@ -360,29 +372,32 @@ std::unique_ptr<AthenaInterprocess::ScheduledWork> EvtRangeScatterer::exec_func(
       // We already have one processor waiting for the answer
       emptyMess4Processor = malloc(1);
       socket2Processor->send(emptyMess4Processor,1);
-      ATH_MSG_INFO("Set one processor free");
+      ATH_MSG_INFO("Set worker PID=" << workerPid << " free");
+      processorWaitRequest.clear();
     }
-    for(int i(0); i<(processorWaitRequest.empty()?m_nprocs:m_nprocs-1); ++i) {
+    bool endLoop{false};
+    while(true) {
       ATH_MSG_DEBUG("Going to set another processor free");
-      while(getNewRangeRequest(socket2Processor,socket2Pilot,procReportPending).empty()) {
-	pollFailedPidQueue(sharedFailedPidQueue,socket2Pilot,procReportPending);
+      while(processorWaitRequest.empty()) {
+	processorWaitRequest = getNewRangeRequest(socket2Processor,socket2Pilot,procReportPending);
+	if(pollFailedPidQueue(sharedFailedPidQueue,socket2Pilot,procReportPending)==-1) {
+	  endLoop = true;
+	  break;
+	}
 	usleep(1000);
       }
+      if(endLoop) break;
+      // Remove worker from m_pid2RangeID
+      workerPid = std::atoi(processorWaitRequest.c_str());
+      auto it = m_pid2RangeID.find(workerPid);
+      if(it!=m_pid2RangeID.end()) {
+	m_pid2RangeID.erase(it);
+      }
       emptyMess4Processor = malloc(1);
       socket2Processor->send(emptyMess4Processor,1);
-      ATH_MSG_INFO("Set one processor free");
-    }
-
-    ATH_MSG_INFO("Still " << procReportPending << " pending reports");
-    
-    // Final round of colecting output file names from processors
-    while(procReportPending>0) {
-      std::string strProcessorRequest = getNewRangeRequest(socket2Processor,socket2Pilot,procReportPending);
-      if(!strProcessorRequest.empty()) {
-	ATH_MSG_WARNING("Unexpected message received from a processor at this stage : " << strProcessorRequest);
-      }
-      pollFailedPidQueue(sharedFailedPidQueue,socket2Pilot,procReportPending);
-      usleep(1000);
+      ATH_MSG_INFO("Set worker PID=" << workerPid << " free");
+      ATH_MSG_INFO("Still " << procReportPending << " pending reports");
+      processorWaitRequest.clear();
     }
   }
 
@@ -509,22 +524,26 @@ std::string EvtRangeScatterer::getNewRangeRequest(yampl::ISocket* socket2Process
   return strProcessorRequest;
 }
 
-void EvtRangeScatterer::pollFailedPidQueue(AthenaInterprocess::SharedQueue*  sharedFailedPidQueue
-					, yampl::ISocket* socket2Pilot
-					, int& procReportPending)
+pid_t EvtRangeScatterer::pollFailedPidQueue(AthenaInterprocess::SharedQueue* sharedFailedPidQueue
+					    , yampl::ISocket* socket2Pilot
+					    , int& procReportPending)
 {
-  pid_t pid;
-  if(sharedFailedPidQueue->try_receive_basic<pid_t>(pid)) {
+  pid_t pid{0};
+  if(sharedFailedPidQueue->try_receive_basic<pid_t>(pid)
+     && pid!=-1) {
     ATH_MSG_INFO("Procesor with PID=" << pid << " has failed!");
-    if(m_pid2RangeID.find(pid)!=m_pid2RangeID.end()) {
+    auto itPid = m_pid2RangeID.find(pid);
+    if(itPid!=m_pid2RangeID.end()) {
       ATH_MSG_WARNING("The failed RangeID = " << m_pid2RangeID[pid] << " will be reported to Pilot");
 
       std::string errorStr("ERR_ATHENAMP_PROCESS " + m_pid2RangeID[pid] + ": Failed to process event range");
       void* errorMessage = malloc(errorStr.size());
       memcpy(errorMessage,errorStr.data(),errorStr.size());
       socket2Pilot->send(errorMessage,errorStr.size());
+      --procReportPending;
+      m_pid2RangeID.erase(pid);
     }
-    procReportPending--;
     ATH_MSG_INFO("Reports pending: " << procReportPending);
   } 
+  return pid;
 }

Control/AthenaMPTools/src/EvtRangeScatterer.h

 /*
-  Copyright (C) 2002-2019 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
 */
 
 #ifndef ATHENAMPTOOLS_TOKENSCATTERER_H
@@ -56,9 +56,9 @@ class EvtRangeScatterer final : public AthenaMPToolBase
 				 , int& procReportPending);
 
   // Poll the failed PID queue to see if any of the Processors has failed
-  void pollFailedPidQueue(AthenaInterprocess::SharedQueue*  sharedFailedPidQueue
-			  , yampl::ISocket* socket2Pilot
-			  , int& procReportPending);
+  pid_t pollFailedPidQueue(AthenaInterprocess::SharedQueue*  sharedFailedPidQueue
+			   , yampl::ISocket* socket2Pilot
+			   , int& procReportPending);
 
   StringProperty           m_processorChannel;
   StringProperty           m_eventRangeChannel;