Calorimeter/CaloLocalHadCalib/share/GetH6LCSinglePionsPerf_cbt.py

@@ -126,17 +126,6 @@ if doTopoClusterAgain:
                                                              "TileExt0", "TileExt1", "TileExt2",
                                                              "TileGap1", "TileGap2", "TileGap3",
                                                              "FCAL0", "FCAL1"] 
-  # simple noise tool
-  SimpleNoiseFile = "sigmas_all.dat"
-  from CaloTools.CaloToolsConf import SimpleNoiseToolFromTextFile
-  SimpleNoiseTool = SimpleNoiseToolFromTextFile("SimpleNoiseTool")
-  SimpleNoiseTool.CellNoiseFileName = SimpleNoiseFile
-  ToolSvc += SimpleNoiseTool
-
-  topSequence.CaloTopoCluster.TopoMaker.CaloNoiseTool           = SimpleNoiseTool
-  topSequence.CaloTopoCluster.LocalCalib.LCWeight.CaloNoiseTool = SimpleNoiseTool
-  topSequence.CaloTopoCluster.DMCalib.LCDeadMaterial.CaloNoiseTool = SimpleNoiseTool
-
   # remove annoying BadChannel tool
   tlist=[]
   for x in topSequence.CaloTopoCluster.ClusterCorrectionTools:

Calorimeter/CaloLocalHadCalib/share/cl_w_ooc_H6.py

@@ -142,16 +142,6 @@ if doTopoClusterAgain:
                                                              "TileExt0", "TileExt1", "TileExt2",
                                                              "TileGap1", "TileGap2", "TileGap3",
                                                              "FCAL0", "FCAL1"] 
-  # simple noise tool
-  SimpleNoiseFile = "sigmas_ave.dat"
-  from CaloTools.CaloToolsConf import SimpleNoiseToolFromTextFile
-  SimpleNoiseTool = SimpleNoiseToolFromTextFile("SimpleNoiseTool")
-  SimpleNoiseTool.CellNoiseFileName = SimpleNoiseFile
-  ToolSvc += SimpleNoiseTool
-
-  topSequence.CaloTopoCluster.TopoMaker.CaloNoiseTool           = SimpleNoiseTool
-  topSequence.CaloTopoCluster.LocalCalib.LCWeight.CaloNoiseTool = SimpleNoiseTool
-  topSequence.CaloTopoCluster.DMCalib.LCDeadMaterial.CaloNoiseTool = SimpleNoiseTool
 
   # remove annoying BadChannel tool
   tlist=[]

Calorimeter/CaloLocalHadCalib/share/deadmaterialtree_ctb.py

@@ -137,16 +137,6 @@ if doTopoClusterAgain:
                                                              "TileExt0", "TileExt1", "TileExt2",
                                                              "TileGap1", "TileGap2", "TileGap3",
                                                              "FCAL0", "FCAL1"] 
-  # simple noise tool
-  SimpleNoiseFile = "sigmas_ave.dat"
-  from CaloTools.CaloToolsConf import SimpleNoiseToolFromTextFile
-  SimpleNoiseTool = SimpleNoiseToolFromTextFile("SimpleNoiseTool")
-  SimpleNoiseTool.CellNoiseFileName = SimpleNoiseFile
-  ToolSvc += SimpleNoiseTool
-
-  topSequence.CaloTopoCluster.TopoMaker.CaloNoiseTool           = SimpleNoiseTool
-  topSequence.CaloTopoCluster.LocalCalib.LCWeight.CaloNoiseTool = SimpleNoiseTool
-  topSequence.CaloTopoCluster.DMCalib.LCDeadMaterial.CaloNoiseTool = SimpleNoiseTool
 
   # remove annoying BadChannel tool
   tlist=[]

Calorimeter/CaloRec/ispellwords

@@ -654,7 +654,6 @@ TileCalibrationCellHitCnt
 TileCalibHitActiveCell
 TileCalibHitInactiveCell
 TileCalibHitDeadMaterial
-CellDetailsNoiseTool
 DetStore
 AuditInitialize
 doLArHitToCellDirect
@@ -908,12 +907,10 @@ const
 CPUTimeRef
 LArClusterCopier
 larClusterCopier
-CaloNoiseTool
 TopoClusterMapperGetter
 SRT
 sizep
 stl
-useNoiseTool
 AuditBeginRun
 theCaloCellContainerCheckerTool
 EMCellContainerName
@@ -952,7 +949,6 @@ AuditFinalize
 CopyCell
 leakRSS
 gapscin
-UseCaloNoiseTool
 MeVTool
 PreSampE
 TWOPI
@@ -1018,7 +1014,6 @@ jetrec
 makeinfo
 jobProperty
 CaloTopoSplitterClusterCell
-theCaloNoiseTool
 laplace
 ThresholdGeVCells
 thresholdGeVCells
@@ -1036,7 +1031,6 @@ EvtMax
 CaloTopoSplitterHashCluster
 CaloClusterStoreHelper
 DMCalib
-LArNoiseTool
 MinBias
 AuditRestart
 maxima
@@ -1048,7 +1042,6 @@ CellContainers
 maxEt
 LArCmbTwrBldr
 CaloTowerCmbGetter
-ICalorimeterNoiseTool
 LArCmbTwrBldrChronobuild
 ClusterContainer
 memexe
@@ -1087,7 +1080,6 @@ AthenaKernel
 CaloCellsName
 MyPackage
 TileCells
-CaloNoiseToolDefault
 lastEvForLeak
 LArRawChannels
 efcal
@@ -1232,7 +1224,6 @@ ntileb
 enums
 CellCorrections
 MonitorService
-ICaloNoiseTool
 iCalo
 EMTopo
 ntileg
@@ -1259,7 +1250,6 @@ eng
 Unal
 enum
 endif
-noiseTool
 MinLLongitudinal
 endcap
 EndCap

Calorimeter/CaloRec/python/CaloTopoTowerGetter.py

-# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+# Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 
 # specifies Tower/TopoCluster -> TopoTower
 from AthenaCommon.JobProperties import jobproperties as jp
@@ -111,8 +111,6 @@ class CaloTopoTowerGetter ( Configured )  :
                 theCaloTopoTowerAlgorithm.MinimumClusterEnergy   = -1000000000.0
                 theCaloTopoTowerAlgorithm.CellEnergySignificance = -1.
 
-                # noise tool
-                theCaloTopoTowerAlgorithm.UseCaloNoiseTool       = False
 
                 ########################
 

Calorimeter/CaloRec/python/MakeTowersFromClusters.py

-# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+# Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 
 #from AthenaCommon.AppMgr import ServiceMgr as svcMgr
 
@@ -132,11 +132,6 @@ def MakeTowersFromClusters(towerMakerName      = 'CaloTowerBuilderAlg',        #
     
     ''' External tools for moment calculation
     '''
-    ##from CaloTools.CaloNoiseToolDefault import CaloNoiseToolDefault
-    #from AthenaCommon.AppMgr import ToolSvc
-    #caloNoiseTool  = CaloNoiseToolDefault()
-    #ToolSvc       += caloNoiseTool
-        
     from CaloTools.CaloNoiseCondAlg import CaloNoiseCondAlg
     CaloNoiseCondAlg ()
     CaloNoiseCondAlg(noisetype="totalNoise") # "electronicNoise","pileupNoise","totalNoise"
@@ -148,8 +143,6 @@ def MakeTowersFromClusters(towerMakerName      = 'CaloTowerBuilderAlg',        #
     from CaloRec.CaloRecConf          import CaloClusterMomentsMaker
     clusterMoments                  = CaloClusterMomentsMaker (towerMakerName+'MomentMaker')
     clusterMoments.MaxAxisAngle     = 20*deg
-    #clusterMoments.CaloNoiseTool    = caloNoiseTool
-    #clusterMoments.UsePileUpNoise   = True
     clusterMoments.TwoGaussianNoise = jobproperties.CaloTopoClusterFlags.doTwoGaussianNoise()
     clusterMoments.MinBadLArQuality = 4000
     clusterMoments.MomentsNames     = [

Calorimeter/CaloRec/share/CaloRec_jobOptions.py

@@ -90,13 +90,9 @@ if rec.readESD and not rec.doESD:
 ## if rec.readESD() and DetFlags.detdescr.Calo_on():
 ##     from CaloRec.CaloRecConf import CaloCellMaker
 ##     theCaloCellMaker=CaloCellMaker(CaloCellsOutputName="AllCalo",CaloCellHack=True)    
-##     # calo modifying tool to be inserted there
-##     from CaloTools.CaloNoiseToolDefault import CaloNoiseToolDefault
-##     theCaloNoiseTool = CaloNoiseToolDefault()
-##     ToolSvc+=theCaloNoiseTool
     
 ##     from CaloCellCorrection.CaloCellCorrectionConf import CaloCellRandomizer     
-##     theCaloCellRandomizer=CaloCellRandomizer(noiseTool=theCaloNoiseTool,DoGaussRandomization=True)
+##     theCaloCellRandomizer=CaloCellRandomizer(DoGaussRandomization=True)
 ##     ToolSvc += theCaloCellRandomizer
 
 ##     from CaloRec.CaloRecConf import CaloCellContainerCorrectorTool

Calorimeter/CaloRec/src/CaloTopoTowerAlgorithm.cxx

 /*
-  Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 */
 
 #include "GaudiKernel/MsgStream.h"
@@ -18,7 +18,6 @@
 #include "CaloEvent/CaloCellContainer.h"
 #include "CaloEvent/CaloClusterContainer.h"
 #include "CaloEvent/CaloTopoTowerContainer.h"
-#include "CaloInterface/ICalorimeterNoiseTool.h"
 #include "CaloTopoTowerAlgorithm.h"
 
 #include <string>
@@ -59,8 +58,6 @@ CaloTopoTowerAlgorithm::CaloTopoTowerAlgorithm(const std::string& name,
 
   // Noise Tool stuff
   declareProperty("DefaultNoiseSigma",      m_noiseSigma           = 10.0);
-  declareProperty("UseCaloNoiseTool",       m_useNoiseTool         = false);
-  declareProperty("UsePileUpNoise",         m_usePileUpNoise       = true);
   declareProperty("CellEnergySignificance", m_cellESignificanceThreshold = -1);
 
   // Calo from which to use cells
@@ -97,13 +94,6 @@ StatusCode CaloTopoTowerAlgorithm::initialize()
   ATH_CHECK(m_towerContainerKey.initialize());
   ATH_CHECK(m_newTowerContainerKey.initialize());
 
-  // Report some information regarding the noise tool
-  if ( m_useNoiseTool && m_usePileUpNoise) {
-    ATH_MSG_DEBUG( "Pile-Up Noise from Noise Tool "
-         << " is selected! The noise sigma will be the"
-                   << " quadratic sum of the electronics noise and the pile up!"  );
-  }
-
   m_caloIndices.clear();
   for ( unsigned int iCalos=0; iCalos< m_includedCalos.size(); iCalos++ )
     {
@@ -268,8 +258,6 @@ StatusCode CaloTopoTowerAlgorithm::execute (const EventContext& ctx) const
   theTowers->SetUseCellWeights(m_useCellWeights);
 
   // Noise tool stuff
-  theTowers->SetUseNoiseTool(m_useNoiseTool);
-  theTowers->SetUsePileUpNoise(m_usePileUpNoise);
   theTowers->SetNoiseSigma(m_noiseSigma);
   theTowers->SetCellESignificanceThreshold(m_cellESignificanceThreshold);
 

Calorimeter/CaloRec/src/CaloTopoTowerAlgorithm.h

 /*
-  Copyright (C) 2002-2019 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 */
 
 #ifndef CALOREC_CALOTOPOTOWERALGORITHM_H
@@ -87,8 +87,6 @@ class CaloTopoTowerAlgorithm : public AthReentrantAlgorithm
   bool   m_useCellWeights;
 
   // Noise tool stuff
-  bool m_useNoiseTool;
-  bool m_usePileUpNoise;
   float m_noiseSigma;
   float m_cellESignificanceThreshold;
 

Control/AthenaBaseComps/AthenaBaseComps/DynamicDataHelper.h

@@ -66,7 +66,7 @@ namespace Ath {
       DataObjIDPtrColl m_inputHandles;
       DataObjIDPtrColl m_outputHandles;
 
-      std::vector< std::pair<IDynamicDataConsumer *,Gaudi::Algorithm *> > m_dynDataConsumer;
+      std::unordered_map< IDynamicDataConsumer *,std::vector<Gaudi::Algorithm *> > m_dynDataConsumer;
 
    };
 }

Control/AthenaBaseComps/src/DynamicDataHelper.cxx

@@ -68,7 +68,10 @@ namespace Ath {
             [this](Gaudi::Algorithm *an_alg) {
                IDynamicDataConsumer *a_dyn_data_consumer=dynamic_cast<IDynamicDataConsumer *>(an_alg);
                if (a_dyn_data_consumer) {
-                  m_dynDataConsumer.push_back(std::make_pair(a_dyn_data_consumer, an_alg));
+                  std::vector<Gaudi::Algorithm *> &alg_list = m_dynDataConsumer[a_dyn_data_consumer];
+                  if (std::find(alg_list.begin(),alg_list.end(),an_alg)==alg_list.end()) {
+                     alg_list.push_back(an_alg);
+                  }
                }
                // have to collect input and output handles because the sequence will only collect them  after initialize
                insertHandles(m_inputHandles, an_alg->inputHandles());
@@ -79,7 +82,10 @@ namespace Ath {
                auto tool_visitor = [this, an_alg](IAlgTool *alg_tool) {
                   IDynamicDataConsumer *a_dyn_data_consumer=dynamic_cast<IDynamicDataConsumer *>(alg_tool);
                   if (a_dyn_data_consumer) {
-                     m_dynDataConsumer.push_back(std::make_pair(a_dyn_data_consumer, an_alg));
+                     std::vector<Gaudi::Algorithm *> &alg_list = m_dynDataConsumer[a_dyn_data_consumer];
+                     if (std::find(alg_list.begin(),alg_list.end(),an_alg)==alg_list.end()) {
+                        alg_list.push_back(an_alg);
+                     }
                   }
                   if (auto* tool_impl = dynamic_cast<AlgTool*>( alg_tool ); tool_impl) {
                      insertHandles(m_inputHandles, tool_impl->inputHandles());
@@ -130,29 +136,32 @@ namespace Ath {
          }
          out << endmsg;
       }
-      for (std::pair<IDynamicDataConsumer *,Gaudi::Algorithm *> &dyn_data_consumer :  m_dynDataConsumer ) {
+      for (std::pair<IDynamicDataConsumer * const,std::vector<Gaudi::Algorithm *> > &dyn_data_consumer :  m_dynDataConsumer ) {
          std::vector<Gaudi::DataHandle *> tmp_new_input_handles;
          std::vector<Gaudi::DataHandle *> tmp_new_output_handles;
          bool ret = dyn_data_consumer.first->updateDataNeeds(input_data_in, output_data_in, tmp_new_input_handles, tmp_new_output_handles);
          if (msgLvl(out,MSG::VERBOSE)) {
-            out << MSG::VERBOSE << " new data from " << dyn_data_consumer.second->name()
-                << " i:" << tmp_new_input_handles.size()
+            out << MSG::VERBOSE << " new data for";
+            for (const Gaudi::Algorithm *parent_alg : dyn_data_consumer.second) {
+               out << " " << parent_alg->name();
+            }
+            out << " i:" << tmp_new_input_handles.size()
                 << " o:" << tmp_new_output_handles.size()
                 << endmsg;
          }
          updated |=  ret;
          if (ret) {
             if (msgLvl(out,MSG::DEBUG)) {
-               if (Gaudi::Algorithm *an_alg = dynamic_cast<Gaudi::Algorithm *>(dyn_data_consumer.first); an_alg) {
+               if (Gaudi::Algorithm * const an_alg = dynamic_cast<Gaudi::Algorithm * const>(dyn_data_consumer.first); an_alg) {
                   out << MSG::DEBUG << an_alg->name() << " updated its data needs." << endmsg;
                   // DEBUG:
                   out << MSG::DEBUG << an_alg->name() << " inputs:";
-                  for (const Gaudi::DataHandle *handle : an_alg->inputHandles()) {
-                     out << " " << handle->fullKey();
-                  }
-                  out << endmsg;
+                     for (const Gaudi::DataHandle *handle : an_alg->inputHandles()) {
+                        out << " " << handle->fullKey();
+                     }
+                     out << endmsg;
                }
-               else if (AlgTool *alg_tool = dynamic_cast<AlgTool *>(dyn_data_consumer.first); alg_tool) {
+               else if (AlgTool *alg_tool = dynamic_cast<AlgTool * const>(dyn_data_consumer.first); alg_tool) {
                   out << MSG::DEBUG << alg_tool->name() << " updated its data needs." << endmsg;
                   // DEBUG:
                   out << MSG::DEBUG << alg_tool->name() << " inputs:";
@@ -166,20 +175,21 @@ namespace Ath {
                }
             }
          }
-         Gaudi::Algorithm *parent_alg = dyn_data_consumer.second;
-         input_data_out.reserve(input_data_out.size()+tmp_new_input_handles.size());
-         for (Gaudi::DataHandle *a_handle : tmp_new_input_handles) {
-            parent_alg->addDependency(a_handle->fullKey(), Gaudi::DataHandle::Reader);
-            input_data_out.push_back(&(a_handle->fullKey()));
-            if (msgLvl(out,MSG::DEBUG)) out << MSG::DEBUG << " addDependency new input for " << parent_alg->name() << " "
-                                            << a_handle->fullKey().className() << " / " <<  a_handle->objKey() << endmsg;
-         }
-         output_data_out.reserve(output_data_out.size()+tmp_new_output_handles.size());
-         for (Gaudi::DataHandle *a_handle : tmp_new_output_handles) {
-            parent_alg->addDependency(a_handle->fullKey(), Gaudi::DataHandle::Writer);
-            output_data_out.push_back(&(a_handle->fullKey()));
-            if (msgLvl(out,MSG::DEBUG)) out << MSG::DEBUG << " addDependency new output for " << parent_alg->name() << " "
-                                            << a_handle->fullKey().className() << " / "  << a_handle->objKey() << endmsg;
+         for (Gaudi::Algorithm *parent_alg : dyn_data_consumer.second) {
+            input_data_out.reserve(input_data_out.size()+tmp_new_input_handles.size());
+            for (Gaudi::DataHandle *a_handle : tmp_new_input_handles) {
+               parent_alg->addDependency(a_handle->fullKey(), Gaudi::DataHandle::Reader);
+               input_data_out.push_back(&(a_handle->fullKey()));
+               if (msgLvl(out,MSG::DEBUG)) out << MSG::DEBUG << " addDependency new input for " << parent_alg->name() << " "
+                                               << a_handle->fullKey().className() << " / " <<  a_handle->objKey() << endmsg;
+            }
+            output_data_out.reserve(output_data_out.size()+tmp_new_output_handles.size());
+            for (Gaudi::DataHandle *a_handle : tmp_new_output_handles) {
+               parent_alg->addDependency(a_handle->fullKey(), Gaudi::DataHandle::Writer);
+               output_data_out.push_back(&(a_handle->fullKey()));
+               if (msgLvl(out,MSG::DEBUG)) out << MSG::DEBUG << " addDependency new output for " << parent_alg->name() << " "
+                                               << a_handle->fullKey().className() << " / "  << a_handle->objKey() << endmsg;
+            }
          }
       }
       return updated;

Control/AthenaCommon/python/AtlasThreadedJob.py

@@ -48,15 +48,16 @@ def _setupAtlasThreadedJob():
     topSequence += SGInputLoader (FailIfNoProxy = False)
     AlgScheduler.setDataLoaderAlg ('SGInputLoader' )
 
-    if theApp._opts.mtes :
-        # Multi-threaded Event Service
+    if (theApp._opts.mtes or jps.ConcurrencyFlags.NumProcs()>0):
+        # Either multi-threaded Event Service or hybrid MP+MT
         from AthenaServices.AthenaServicesConf import AthenaMtesEventLoopMgr
         
         svcMgr += AthenaMtesEventLoopMgr()
         svcMgr.AthenaMtesEventLoopMgr.WhiteboardSvc = "EventDataSvc"
         svcMgr.AthenaMtesEventLoopMgr.SchedulerSvc = AlgScheduler.getScheduler().getName()
+
+    if theApp._opts.mtes:
         svcMgr.AthenaMtesEventLoopMgr.EventRangeChannel = theApp._opts.mtes_channel
-        
         theApp.EventLoop = "AthenaMtesEventLoopMgr"
     else:
         from AthenaServices.AthenaServicesConf import AthenaHiveEventLoopMgr

Control/AthenaCommon/share/GenerateBootstrapTest.ref

 Willing to acquire file bootstrap.pkl from
 /afs/cern.ch/user/s/ssnyder/atlas-work3/build-x86_64-centos7-gcc8-dbg/x86_64-centos7-gcc8-dbg/share/bootstrap.pkl
 copy bootstrap.pkl
-... Read 2 items from python pickle file:  bootstrap.pkl
-... Read 2 items from python pickle file:  bootstrap_test.pkl
+... Read defaultdict from bootstrap.pkl
+... Read 27 items from python pickle file:  bootstrap.pkl
+... Read defaultdict from bootstrap_test.pkl
+... Read 27 items from python pickle file:  bootstrap_test.pkl
 Step 1: reference file #components: 27
 Step 2: file to check  #components: 27
 Willing to acquire file bootstrap_threaded.pkl from
 /afs/cern.ch/user/s/ssnyder/atlas-work3/build-x86_64-centos7-gcc8-dbg/x86_64-centos7-gcc8-dbg/share/bootstrap_threaded.pkl
 copy bootstrap_threaded.pkl
-... Read 2 items from python pickle file:  bootstrap_threaded.pkl
-... Read 2 items from python pickle file:  bootstrap_threaded_test.pkl
+... Read defaultdict from bootstrap_threaded.pkl
+... Read 34 items from python pickle file:  bootstrap_threaded.pkl
+... Read defaultdict from bootstrap_threaded_test.pkl
+... Read 34 items from python pickle file:  bootstrap_threaded_test.pkl
 Step 1: reference file #components: 34
 Step 2: file to check  #components: 34

Control/AthenaConfiguration/python/iconfTool/models/loaders.py

-# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+# Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 
 import pickle
 import logging
@@ -85,6 +85,8 @@ def loadConfigFile(fname, args) -> Dict:
                 cfg, (collections.defaultdict, dict)
             ):  # old configuration
                 conf.update(cfg)
+                conf.update(pickle.load(input_file)) # special services
+                # FIXME: there's a third pickle object with python components
             elif isinstance(cfg, (collections.Sequence)):
                 for c in cfg:
                     conf.update(c)

Control/AthenaKernel/AthenaKernel/IHybridProcessorHelper.h

+/*
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
+*/
+
+#ifndef ATHENAKERNEL_IHYBRIDPROCESSORHELPER_H
+#define ATHENAKERNEL_IHYBRIDPROCESSORHELPER_H
+
+/**
+ * @file  IHybridProcessorHelper.h
+ * @class IHybridProcessorHelper
+ * @brief Helper interface for implementing hybrid MP+MT.
+          Used by the Hybrid Shared Event Queue Consumer MP tool
+ **/
+
+class IHybridProcessorHelper {
+ public: 
+  virtual void resetAppReturnCode() = 0;
+  virtual void setCurrentEventNum(int num) = 0;
+  virtual bool terminateLoop() = 0;
+  virtual int drainScheduler(int& finishedEvts, bool report) = 0;
+};
+
+#endif

Control/AthenaMP/python/PyComps.py

-# Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+# Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 
 #-----Python imports---#
 import os, sys, shutil
@@ -118,12 +118,10 @@ class MpEvtLoopMgr(AthMpEvtLoopMgr):
                                                    ChunkSize=chunk_size) ]
 
             if (self.nThreads >= 1):
+                if(pileup):
+                    raise Exception('Running pileup digitization in mixed MP+MT currently not supported')
                 from AthenaMPTools.AthenaMPToolsConf import SharedHiveEvtQueueConsumer
-                self.Tools += [ SharedHiveEvtQueueConsumer(UseSharedReader=use_shared_reader,
-                                                           UseSharedWriter=use_shared_writer,
-                                                           IsPileup=pileup,
-                                                           IsRoundRobin=(strategy=='RoundRobin'),
-                                                           EventsBeforeFork=events_before_fork,
+                self.Tools += [ SharedHiveEvtQueueConsumer(EventsBeforeFork=events_before_fork,
                                                            Debug=debug_worker)   ]
             else:
                 from AthenaMPTools.AthenaMPToolsConf import SharedEvtQueueConsumer

Control/AthenaMPTools/src/SharedHiveEvtQueueConsumer.cxx

 /*
-  Copyright (C) 2002-2017 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 */
 
 #include "SharedHiveEvtQueueConsumer.h"
@@ -7,10 +7,8 @@
 #include "AthenaInterprocess/ProcessGroup.h"
 #include "AthenaInterprocess/Incidents.h"
 
-#include "AthenaKernel/IEventSeek.h"
 #include "AthenaKernel/IEvtSelectorSeek.h"
-#include "AthenaKernel/IEventShare.h"
-#include "AthenaKernel/IDataShare.h"
+#include "AthenaKernel/IHybridProcessorHelper.h"
 #include "GaudiKernel/IEvtSelector.h"
 #include "GaudiKernel/IIoComponentMgr.h"
 #include "GaudiKernel/IFileMgr.h"
@@ -40,31 +38,20 @@ namespace SharedHiveEvtQueueConsumer_d {
 }
 
 SharedHiveEvtQueueConsumer::SharedHiveEvtQueueConsumer(const std::string& type
-					       , const std::string& name
-					       , const IInterface* parent)
+						       , const std::string& name
+						       , const IInterface* parent)
   : AthenaMPToolBase(type,name,parent)
-  , m_useSharedReader(false)
-  , m_useSharedWriter(false)
-  , m_isPileup(false)
-  , m_isRoundRobin(false)
   , m_nEventsBeforeFork(0)
   , m_debug(false)
   , m_rankId(-1)
   , m_chronoStatSvc("ChronoStatSvc", name)
-  , m_evtSeek(0)
   , m_evtSelSeek(0)
   , m_evtContext(0)
-  , m_evtShare(0)
-  , m_dataShare(0)
   , m_sharedEventQueue(0)
   , m_sharedRankQueue(0)
 {
   declareInterface<IAthenaMPTool>(this);
 
-  declareProperty("UseSharedReader",m_useSharedReader);
-  declareProperty("UseSharedWriter",m_useSharedWriter);
-  declareProperty("IsPileup",m_isPileup);
-  declareProperty("IsRoundRobin",m_isRoundRobin);
   declareProperty("EventsBeforeFork",m_nEventsBeforeFork);
   declareProperty("Debug", m_debug);
 
@@ -82,60 +69,19 @@ SharedHiveEvtQueueConsumer::~SharedHiveEvtQueueConsumer()
 StatusCode SharedHiveEvtQueueConsumer::initialize()
 {
   ATH_MSG_DEBUG("In initialize");
-  if(m_isPileup) {
-    m_evtProcessor = ServiceHandle<IEventProcessor>("PileUpEventLoopMgr",name());
-    ATH_MSG_INFO("The job running in pileup mode");
-  }
-  else {
-    ATH_MSG_INFO("The job running in non-pileup mode");
-  }
 
   StatusCode sc = AthenaMPToolBase::initialize();
   if(!sc.isSuccess())
     return sc;
 
-  // For pile-up jobs use event loop manager for seeking
-  // otherwise use event selector
-  if(m_isPileup) {
-    m_evtSeek = dynamic_cast<IEventSeek*>(m_evtProcessor.operator->());
-    if(!m_evtSeek) {
-      ATH_MSG_ERROR("Unable to dyn-cast PileUpEventLoopMgr to IEventSeek");
-      return StatusCode::FAILURE;
-    }
-  }
-  else {
-    sc = serviceLocator()->service(m_evtSelName,m_evtSelSeek);
-    if(sc.isFailure() || m_evtSelSeek==0) {
-      ATH_MSG_ERROR("Error retrieving IEvtSelectorSeek");
-      return StatusCode::FAILURE;
-    }
-    ATH_CHECK( evtSelector()->createContext (m_evtContext) );
-  }
-
-  sc = serviceLocator()->service(m_evtSelName,m_evtShare);
-  if(sc.isFailure() || m_evtShare==0) {
-    if(m_useSharedReader) {
-      ATH_MSG_ERROR("Error retrieving IEventShare");
-      return StatusCode::FAILURE;
-    }
-    msg(MSG::INFO) << "Could not retrieve IEventShare" << endmsg;
-  }
-
-  IConversionSvc* cnvSvc = 0;
-  sc = serviceLocator()->service("AthenaPoolCnvSvc",cnvSvc);
-  m_dataShare = dynamic_cast<IDataShare*>(cnvSvc);
-  if(sc.isFailure() || m_dataShare==0) {
-    if(m_useSharedWriter) {
-      msg(MSG::ERROR) << "Error retrieving AthenaPoolCnvSvc " << cnvSvc << endmsg;
-      return StatusCode::FAILURE;
-    }
-  }
-
-  sc = m_chronoStatSvc.retrieve();
-  if (!sc.isSuccess()) {
-    ATH_MSG_ERROR("Cannot get ChronoStatSvc.");
+  sc = serviceLocator()->service(m_evtSelName,m_evtSelSeek);
+  if(sc.isFailure() || m_evtSelSeek==0) {
+    ATH_MSG_ERROR("Error retrieving IEvtSelectorSeek");
     return StatusCode::FAILURE;
   }
+  ATH_CHECK( evtSelector()->createContext (m_evtContext) );
+
+  ATH_CHECK(m_chronoStatSvc.retrieve());
   
   return StatusCode::SUCCESS;
 }
@@ -285,7 +231,6 @@ SharedHiveEvtQueueConsumer::subProcessLogs(std::vector<std::string>& filenames)
 std::unique_ptr<AthenaInterprocess::ScheduledWork>
 SharedHiveEvtQueueConsumer::bootstrap_func()
 {
-
   if (m_debug) {
     ATH_MSG_INFO("Bootstrap worker PID " << getpid() << " - waiting for SIGUSR1");
     sigset_t mask, oldmask;
@@ -369,28 +314,7 @@ SharedHiveEvtQueueConsumer::bootstrap_func()
     return outwork;
 
   ATH_MSG_INFO("File descriptors re-opened in the AthenaMP event worker PID=" << getpid());
-
   
-  // ________________________ Make Shared Reader/Writer Client ________________________
-  if(m_useSharedReader && m_evtShare) {
-    if(!m_evtShare->makeClient(m_rankId).isSuccess()) {
-      ATH_MSG_ERROR("Failed to make the event selector a share client");
-      return outwork;
-    } else {
-      ATH_MSG_DEBUG("Successfully made the event selector a share client");
-    }
-  }
-
-  if(m_useSharedWriter && m_dataShare) {
-    IProperty* propertyServer = dynamic_cast<IProperty*>(m_dataShare);
-    if (propertyServer==0 || propertyServer->setProperty("MakeStreamingToolClient", m_rankId + 1).isFailure()) {
-      ATH_MSG_ERROR("Could not change AthenaPoolCnvSvc MakeClient Property");
-      return outwork;
-    } else {
-      ATH_MSG_DEBUG("Successfully made the conversion service a share client");
-    }
-  }
-
   // ________________________ I/O reinit ________________________
   if(!m_ioMgr->io_reinitialize().isSuccess()) {
     ATH_MSG_ERROR("Failed to reinitialize I/O");
@@ -412,24 +336,6 @@ SharedHiveEvtQueueConsumer::bootstrap_func()
     ATH_MSG_DEBUG("Successfully restarted the event selector");
   }
 
-  // ________________________ Restart background event selectors in pileup jobs ________________________
-  if(m_isPileup) {
-    const std::list<IService*>& service_list = serviceLocator()->getServices();
-    std::list<IService*>::const_iterator itSvc = service_list.begin(),
-      itSvcLast = service_list.end();
-    for(;itSvc!=itSvcLast;++itSvc) {
-      IEvtSelector* evtsel = dynamic_cast<IEvtSelector*>(*itSvc);
-      if(evtsel && (evtsel != m_evtSelector)) {
-	if((*itSvc)->start().isSuccess())
-	  ATH_MSG_DEBUG("Restarted event selector " << (*itSvc)->name());
-	else {
-	  ATH_MSG_ERROR("Failed to restart event selector " << (*itSvc)->name());
-	  return outwork;
-	}
-      }
-    }
-  }
-
   // ________________________ Worker dir: chdir ________________________
   if(chdir(worker_rundir.string().c_str())==-1) {
     ATH_MSG_ERROR("Failed to chdir to " << worker_rundir.string());
@@ -446,14 +352,13 @@ SharedHiveEvtQueueConsumer::bootstrap_func()
 
 /* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * */
 
-std::unique_ptr<AthenaInterprocess::ScheduledWork> 
-SharedHiveEvtQueueConsumer::exec_func()
+std::unique_ptr<AthenaInterprocess::ScheduledWork> SharedHiveEvtQueueConsumer::exec_func()
 {
   ATH_MSG_INFO("Exec function in the AthenaMP worker PID=" << getpid());
 
   bool all_ok(true);
 
-  if (! initHive().isSuccess()) {
+  if (!initHive().isSuccess()) {
     ATH_MSG_FATAL("unable to initialize Hive");
     all_ok = false;
   }
@@ -476,141 +381,118 @@ SharedHiveEvtQueueConsumer::exec_func()
     }
   }
 
-
-  // ________________________ This is needed only for PileUp jobs __________________________________
-  // **
-  // If either EventsBeforeFork or SkipEvents is nonzero, first we need to advance the event selector
-  // by EventsBeforeFork+SkipEvents and only after that start seeking on the PileUpEventLoopMgr
-  // **
-  if(m_isPileup && all_ok) {
-    if (!m_evtSelSeek) {
-      StatusCode sc = serviceLocator()->service(m_evtSelName,m_evtSelSeek);
-      if(sc.isFailure() || m_evtSelSeek==0) {
-        ATH_MSG_ERROR("Error retrieving Event Selector with IEvtSelectorSeek interface for PileUp job");
-        all_ok = false;
-      }
-      if (evtSelector()->createContext (m_evtContext).isFailure()) {
-        ATH_MSG_ERROR("Error creating IEventSelector context.");
-        all_ok = false;
-      }
-    }
-    if (all_ok) {
-      if((m_nEventsBeforeFork+skipEvents)
-	 && m_evtSelSeek->seek(*m_evtContext, m_nEventsBeforeFork+skipEvents).isFailure()) {
-	ATH_MSG_ERROR("Unable to seek to " << m_nEventsBeforeFork+skipEvents);    
-	all_ok = false;
-      }
-    }
+  IHybridProcessorHelper* hybridHelper = dynamic_cast<IHybridProcessorHelper*>(m_evtProcessor.get());
+  if(!hybridHelper) {
+    ATH_MSG_FATAL("Failed to acquire IHybridProcessorHelper interface");
+    all_ok = false;
   }
-  // ________________________ This is needed only for PileUp jobs __________________________________
-
+  // Reset the application return code.
+  hybridHelper->resetAppReturnCode();
 
+  int finishedEvts =0;
+  int createdEvts =0;
   long intmask = pow(0x100,sizeof(int))-1; // Mask for decoding event number from the value posted to the queue
-  int nEvt(m_nEventsBeforeFork);
-  int nEventsProcessed(0);
   long evtnumAndChunk(0);
-
-  unsigned evtCounter(0);
+//  unsigned evtCounter(0);
   int evtnum(0), chunkSize(1);
 
-  // For the round robin we need to know the maximum number of events for this job
-  if(m_isRoundRobin) {
-    evtnumAndChunk = 1;
-    while(evtnumAndChunk>0) {
-      if(!m_sharedEventQueue->try_receive_basic<long>(evtnumAndChunk)) {
-        usleep(1000);
-      }
-    }
-    evtnumAndChunk *= -1;
+  ATH_MSG_INFO("Starting loop on events");
+
+  StatusCode sc(StatusCode::SUCCESS,true);
+
+  while(!m_sharedEventQueue->try_receive_basic<long>(evtnumAndChunk)) {
+    ATH_MSG_DEBUG("Event queue is empty");
+    usleep(1000);
   }
-  
-  if(all_ok) {
-    while(true) {
-      if(m_isRoundRobin) {
-	evtnum = skipEvents + m_nprocs*evtCounter + m_rankId; 
-	if(evtnum>=evtnumAndChunk+skipEvents) {
-          break;
-        }
-	evtCounter++;
+  bool loop_ended = (evtnumAndChunk<0);
+  if(!loop_ended) {
+    ATH_MSG_DEBUG("Received value from the queue 0x" << std::hex << evtnumAndChunk << std::dec);
+    chunkSize = evtnumAndChunk >> (sizeof(int)*8);
+    evtnum = evtnumAndChunk & intmask;
+    ATH_MSG_INFO("Received from the queue: event num=" << evtnum << " chunk size=" << chunkSize);
+    hybridHelper->setCurrentEventNum(++evtnum);
+  }
+
+  bool no_more_events = false;
+
+  while(!loop_ended) {
+    ATH_MSG_DEBUG(" -> createdEvts: " << createdEvts);
+
+    if(!hybridHelper->terminateLoop()         // No scheduled loop termination
+       && !no_more_events                     // We are not yet done getting events
+       && m_schedulerSvc->freeSlots()>0) {    // There are still free slots in the scheduler
+      ATH_MSG_DEBUG("createdEvts: " << createdEvts << ", freeslots: " << m_schedulerSvc->freeSlots());
+
+      auto ctx = m_evtProcessor->createEventContext();
+      if(!ctx.valid()) {
+        sc = StatusCode::FAILURE;
       }
       else {
-	if(!m_sharedEventQueue->try_receive_basic<long>(evtnumAndChunk)) {
-	  // The event queue is empty, but we should check whether there are more events to come or not
-	  ATH_MSG_DEBUG("Event queue is empty"); 
-	  usleep(1000);
-	  continue;
-	}
-	if(evtnumAndChunk<=0) {
-	  evtnumAndChunk *= -1;
-	  ATH_MSG_DEBUG("No more events are expected. The total number of events for this job = " << evtnumAndChunk);
-	  break;
-	}
-	while (m_schedulerSvc->freeSlots() < 1) {
-	  ATH_MSG_DEBUG("waiting for a free scheduler slot");
-	  usleep(1000000);
-	}
+        sc = m_evtProcessor->executeEvent(std::move(ctx));
+      }
 
-	ATH_MSG_DEBUG("Received value from the queue 0x" << std::hex << evtnumAndChunk << std::dec);
-	chunkSize = evtnumAndChunk >> (sizeof(int)*8);
-	evtnum = evtnumAndChunk & intmask;
-	ATH_MSG_INFO("Received from the queue: event num=" << evtnum << " chunk size=" << chunkSize);
+      if (sc.isFailure()) {
+        ATH_MSG_ERROR("Terminating event processing loop due to errors");
+        loop_ended = true;
       }
-      nEvt+=chunkSize;
-      StatusCode sc;
-      if(m_useSharedReader) {
-	sc = m_evtShare->share(evtnum);
-	if(sc.isFailure()){
-	  ATH_MSG_ERROR("Unable to share " << evtnum);
-	  all_ok=false;
-	  break;
-	} else {
-	  ATH_MSG_INFO("Share of " << evtnum << " succeeded");
+      else {
+	++createdEvts;
+	if(--chunkSize==0) {
+	  // Fetch next chunk
+	  while(!m_sharedEventQueue->try_receive_basic<long>(evtnumAndChunk)) {
+	    ATH_MSG_DEBUG("Event queue is empty");
+	    usleep(1000);
+	  }
+	  if(evtnumAndChunk<0) {
+	    no_more_events = true;
+	    evtnumAndChunk *= -1;
+	    ATH_MSG_DEBUG("No more events are expected. The total number of events for this job = " << evtnumAndChunk);
+	  }
+	  else {
+	    ATH_MSG_DEBUG("Received value from the queue 0x" << std::hex << evtnumAndChunk << std::dec);
+	    chunkSize = evtnumAndChunk >> (sizeof(int)*8);
+	    evtnum = evtnumAndChunk & intmask;
+	    ATH_MSG_INFO("Received from the queue: event num=" << evtnum << " chunk size=" << chunkSize);
+	  }
 	}
-      } else {
-	m_chronoStatSvc->chronoStart("AthenaMP_seek");
-        if (m_evtSeek) {
-          sc=m_evtSeek->seek(evtnum);
-        }
-        else {
-          sc=m_evtSelSeek->seek(*m_evtContext, evtnum);
-        }
-	if(sc.isFailure()){
-	  ATH_MSG_ERROR("Unable to seek to " << evtnum);
-	  all_ok=false;
-	  break;
-	} else {
-	  ATH_MSG_INFO("Seek to " << evtnum << " succeeded");
+	// Advance to the next event
+	if(!no_more_events) {
+	  hybridHelper->setCurrentEventNum(++evtnum);
 	}
-	m_chronoStatSvc->chronoStop("AthenaMP_seek");
       }
-      m_chronoStatSvc->chronoStart("AthenaMP_nextEvent");
-      sc = m_evtProcessor->nextEvent(chunkSize);
-      nEventsProcessed += chunkSize;
-      if(sc.isFailure()){
-	if(chunkSize==1)
-	  ATH_MSG_ERROR("Unable to process event " << evtnum);
-	else
-	  ATH_MSG_ERROR("Unable to process the chunk (" << evtnum << "," << evtnum+chunkSize-1 << ")");
-	all_ok=false;
-	break;
+    }
+    else {
+      // all the events were created but not all finished or the slots were
+      // all busy: the scheduler should finish its job
+      ATH_MSG_DEBUG("Draining the scheduler");
+
+      // Pull out of the scheduler the finished events
+      int ir = hybridHelper->drainScheduler(finishedEvts,true);
+      if(ir < 0) {
+        // some sort of error draining scheduler;
+        loop_ended = true;
+        sc = StatusCode::FAILURE;
+      }
+      else if(ir == 0) {
+	// no more events in scheduler
+        if(no_more_events) {
+          // We are done
+          loop_ended = true;
+          sc = StatusCode::SUCCESS;
+        }
+      }
+      else {
+        // keep going!
       }
-      m_chronoStatSvc->chronoStop("AthenaMP_nextEvent"); 
     }
-  }
+  } // end main loop on finished events
 
   if(all_ok) {
     if(m_evtProcessor->executeRun(0).isFailure()) {
       ATH_MSG_ERROR("Could not finalize the Run");
       all_ok=false;
     } else {
-      StatusCode sc;
-      if (m_evtSeek) {
-        sc = m_evtSeek->seek(evtnumAndChunk+skipEvents);
-      }
-      else {
-        sc = m_evtSelSeek->seek(*m_evtContext, evtnumAndChunk+skipEvents);
-      }
-      if(sc.isFailure()) { 
+      if(m_evtSelSeek->seek(*m_evtContext, evtnumAndChunk+skipEvents).isFailure()) { 
 	ATH_MSG_DEBUG("Seek past maxevt to " << evtnumAndChunk+skipEvents << " returned failure. As expected...");
       }
     }
@@ -624,7 +506,7 @@ SharedHiveEvtQueueConsumer::exec_func()
   *(int*)(outdata) = (all_ok?0:1); // Error code: for now use 0 success, 1 failure
   AthenaMPToolBase::Func_Flag func = AthenaMPToolBase::FUNC_EXEC;
   memcpy((char*)outdata+sizeof(int),&func,sizeof(func));
-  memcpy((char*)outdata+sizeof(int)+sizeof(func),&nEventsProcessed,sizeof(int));
+  memcpy((char*)outdata+sizeof(int)+sizeof(func),&createdEvts,sizeof(int));
 
   outwork->data = outdata;
   outwork->size = outsize;
@@ -753,7 +635,7 @@ SharedHiveEvtQueueConsumer::initHive() {
       }
     }      
 
-    m_evtProcessor = ServiceHandle<IEventProcessor>("AthenaHiveEventLoopMgr",name());
+    m_evtProcessor = ServiceHandle<IEventProcessor>("AthenaMtesEventLoopMgr",name());
 
     if (m_evtProcessor.retrieve().isFailure()) {
       ATH_MSG_ERROR("could not setup " << m_evtProcessor.typeAndName());

Control/AthenaMPTools/src/SharedHiveEvtQueueConsumer.h

 /*
-  Copyright (C) 2002-2019 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 */
 
 #ifndef ATHENAMPTOOLS_SHAREDHIVEEVTQUEUECONSUMER_H
@@ -12,10 +12,7 @@
 #include "GaudiKernel/IScheduler.h"
 #include "GaudiKernel/IEvtSelector.h"
 
-class IEventSeek;
 class IEvtSelectorSeek;
-class IEventShare;
-class IDataShare;
 class IChronoStatSvc;
 
 class SharedHiveEvtQueueConsumer final : public AthenaMPToolBase
@@ -56,21 +53,14 @@ class SharedHiveEvtQueueConsumer final : public AthenaMPToolBase
   int decodeProcessResult(const AthenaInterprocess::ProcessResult* presult, bool doFinalize);
 
   // Properties
-  bool m_useSharedReader; // Work in pair with a SharedReader
-  bool m_useSharedWriter; // Work in pair with a SharedWriter
-  bool m_isPileup;        // Are we doing pile-up digitization?
-  bool m_isRoundRobin;    // Are we running in the "reproducible mode"?
   int  m_nEventsBeforeFork;
   bool m_debug;
 
   int  m_rankId;          // Each worker has its own unique RankID from the range (0,...,m_nprocs-1) 
 
   ServiceHandle<IChronoStatSvc>  m_chronoStatSvc;
-  IEventSeek*                    m_evtSeek;
   IEvtSelectorSeek*              m_evtSelSeek;
   IEvtSelector::Context*         m_evtContext;
-  IEventShare*                   m_evtShare;
-  IDataShare*                    m_dataShare;
 
   AthenaInterprocess::SharedQueue*  m_sharedEventQueue;          
   AthenaInterprocess::SharedQueue*  m_sharedRankQueue;          

Control/AthenaServices/src/AthenaMtesEventLoopMgr.cxx

@@ -717,14 +717,16 @@ StatusCode AthenaMtesEventLoopMgr::stop()
 }
 
 
-StatusCode AthenaMtesEventLoopMgr::nextEvent(int /*maxevt*/)
+StatusCode AthenaMtesEventLoopMgr::nextEvent(int maxevt)
 {
+  if(maxevt==0) return StatusCode::SUCCESS;
+
   yampl::ISocketFactory* socketFactory = new yampl::SocketFactory();
   // Create a socket to communicate with the Pilot
-  yampl::ISocket* socket2Pilot = socketFactory->createClientSocket(yampl::Channel(m_eventRangeChannel.value(),yampl::LOCAL),yampl::MOVE_DATA);
+  m_socket = socketFactory->createClientSocket(yampl::Channel(m_eventRangeChannel.value(),yampl::LOCAL),yampl::MOVE_DATA);
 
   // Reset the application return code.
-  Gaudi::setAppReturnCode(m_appMgrProperty, Gaudi::ReturnCode::Success, true).ignore();  
+  resetAppReturnCode();
 
   int finishedEvts =0;
   int createdEvts =0;
@@ -740,7 +742,7 @@ StatusCode AthenaMtesEventLoopMgr::nextEvent(int /*maxevt*/)
 
   std::unique_ptr<RangeStruct> range;
   while(!range) {
-    range = getNextRange(socket2Pilot);
+    range = getNextRange(m_socket);
     usleep(1000);
   }
 
@@ -782,7 +784,7 @@ StatusCode AthenaMtesEventLoopMgr::nextEvent(int /*maxevt*/)
 	  // Fetch next event range
 	  range.reset();
 	  while(!range) {
-	    range = getNextRange(socket2Pilot);
+	    range = getNextRange(m_socket);
 	    usleep(1000);
 	  }
 	  if(range->eventRangeID.empty()) {
@@ -803,7 +805,7 @@ StatusCode AthenaMtesEventLoopMgr::nextEvent(int /*maxevt*/)
       debug() << "Draining the scheduler" << endmsg;
 
       // Pull out of the scheduler the finished events
-      int ir = drainScheduler(finishedEvts,socket2Pilot);
+      int ir = drainScheduler(finishedEvts,true);
       if(ir < 0) {
 	// some sort of error draining scheduler;
 	loop_ended = true;
@@ -825,7 +827,8 @@ StatusCode AthenaMtesEventLoopMgr::nextEvent(int /*maxevt*/)
 
   info() << "---> Loop Finished (seconds): " << secsFromStart() <<endmsg;
 
-  delete socket2Pilot;
+  delete m_socket;
+  m_socket=nullptr;
   delete socketFactory;
   return sc;
 }
@@ -1204,10 +1207,22 @@ EventContext AthenaMtesEventLoopMgr::createEventContext() {
   return ctx;
 }
 
+void AthenaMtesEventLoopMgr::resetAppReturnCode()
+{
+  Gaudi::setAppReturnCode(m_appMgrProperty, Gaudi::ReturnCode::Success, true).ignore();
+}
+
+void AthenaMtesEventLoopMgr::setCurrentEventNum(int num) {
+  m_currentEvntNum = num;
+}
+
+bool AthenaMtesEventLoopMgr::terminateLoop() {
+  return m_terminateLoop;
+}
 //---------------------------------------------------------------------------
 
 int 
-AthenaMtesEventLoopMgr::drainScheduler(int& finishedEvts,yampl::ISocket* socket){
+AthenaMtesEventLoopMgr::drainScheduler(int& finishedEvts,bool report){
 
   StatusCode sc(StatusCode::SUCCESS);
     
@@ -1273,20 +1288,23 @@ AthenaMtesEventLoopMgr::drainScheduler(int& finishedEvts,yampl::ISocket* socket)
     
     // Some code still needs global context in addition to that passed in the incident
     Gaudi::Hive::setCurrentContext( *thisFinishedEvtContext );
+    info() << "Firing EndProcessing" << endmsg;
     m_incidentSvc->fireIncident(Incident(name(), IncidentType::EndProcessing, *thisFinishedEvtContext ));
 
-    // If we completed an event range, then report it to the pilot
-    OutputStreamSequencerSvc::RangeReport_ptr rangeReport = m_outSeqSvc->getRangeReport();
-    if(rangeReport) {
-      std::string outputFileReport = rangeReport->second + std::string(",ID:")
-	+ rangeReport->first + std::string(",CPU:N/A,WALL:N/A");
-      if( not m_inTestMode ) {
-         // In standalone test mode there is no pilot to talk to
-         void* message2pilot = malloc(outputFileReport.size());
-         memcpy(message2pilot,outputFileReport.data(),outputFileReport.size());
-         socket->send(message2pilot,outputFileReport.size());
+    if(report) {
+      // If we completed an event range, then report it to the pilot
+      OutputStreamSequencerSvc::RangeReport_ptr rangeReport = m_outSeqSvc->getRangeReport();
+      if(rangeReport) {
+	std::string outputFileReport = rangeReport->second + std::string(",ID:")
+	  + rangeReport->first + std::string(",CPU:N/A,WALL:N/A");
+	if( not m_inTestMode ) {
+	  // In standalone test mode there is no pilot to talk to
+	  void* message2pilot = malloc(outputFileReport.size());
+	  memcpy(message2pilot,outputFileReport.data(),outputFileReport.size());
+	  m_socket->send(message2pilot,outputFileReport.size());
+	}
+	info() << "Reported the output " << outputFileReport << endmsg;
       }
-      info() << "Reported the output " << outputFileReport << endmsg;
     }
 
     debug() << "Clearing slot " << thisFinishedEvtContext->slot() 

Control/AthenaServices/src/AthenaMtesEventLoopMgr.h

 // -*- C++ -*-
 
 /*
-  Copyright (C) 2002-2020 CERN for the benefit of the ATLAS collaboration
+  Copyright (C) 2002-2021 CERN for the benefit of the ATLAS collaboration
 */
 
 #ifndef ATHENASERVICES_ATHENAMTESEVENTLOOPMGR_H
@@ -23,6 +23,7 @@
 #include "AthenaKernel/IEventSeek.h"
 #include "AthenaKernel/ICollectionSize.h"
 #include "AthenaKernel/IConditionsCleanerSvc.h"
+#include "AthenaKernel/IHybridProcessorHelper.h"
 #include "StoreGate/ActiveStoreSvc.h"
 
 #include <memory>
@@ -52,6 +53,7 @@ class AthenaMtesEventLoopMgr
   : virtual public IEventSeek,
     virtual public ICollectionSize,
     virtual public IIncidentListener,
+    virtual public IHybridProcessorHelper,
             public MinimalEventLoopMgr,
             public Athena::TimeoutMaster
 {
@@ -167,10 +169,6 @@ protected:
   StatusCode clearWBSlot(int evtSlot);
   /// Declare the root address of the event
   int declareEventRootAddress(EventContext&);
-  /// Create event context
-  virtual EventContext createEventContext() override;
-  /// Drain the scheduler from all actions that may be queued
-  int drainScheduler(int& finishedEvents,yampl::ISocket* socket);
   /// Instance of the incident listener waiting for AbortEvent. 
   SmartIF< IIncidentListener >  m_abortEventListener;
   /// Name of the scheduler to be used
@@ -201,6 +199,8 @@ public:
   virtual StatusCode finalize() override;
   /// implementation of IAppMgrUI::nextEvent. maxevt==0 returns immediately
   virtual StatusCode nextEvent(int maxevt) override;
+  /// implementation of IEventProcessor::createEventContext()
+  virtual EventContext createEventContext() override;
   /// implementation of IEventProcessor::executeEvent(void* par)
   virtual StatusCode executeEvent( EventContext&& ctx ) override;
   /// implementation of IEventProcessor::executeRun(int maxevt)
@@ -220,6 +220,15 @@ public:
   /// IIncidentListenet interfaces
   virtual void handle(const Incident& inc) override;
 
+  /// Reset the application return code
+  virtual void resetAppReturnCode() override;
+  
+  virtual void setCurrentEventNum(int num) override;
+  virtual bool terminateLoop() override;
+
+  /// Drain the scheduler from all actions that may be queued
+  virtual int drainScheduler(int& finishedEvents, bool report) override;
+
   /// interface dispatcher
   virtual StatusCode queryInterface( const InterfaceID& riid, 
                                      void** ppvInterface ) override;
@@ -294,6 +303,9 @@ private:
   // Hopefully a temporary measurement. For the time being we cannot
   // support event ranges from different input files.
   std::string m_pfn{""};
+
+  // For the event service running:
+  yampl::ISocket* m_socket{nullptr};
 };
 
 #endif // ATHENASERVICES_ATHENAHIVEEVENTLOOPMGR_H