Speed improvements for MC workflow (#397)

* combine DPL devices in O2DPG

* Setup env variables to speedup ROOT init

Author: sawenzel

MC/bin/o2_dpg_workflow_runner.py

@@ -51,6 +51,7 @@
 parser.add_argument('--checkpoint-on-failure', help=argparse.SUPPRESS) # debug option making a debug-tarball and sending to specified address
                                                                        # argument is alien-path
 parser.add_argument('--retry-on-failure', help=argparse.SUPPRESS, default=2) # number of times a failing task is retried
+parser.add_argument('--rootinit-speedup', help=argparse.SUPPRESS, action='store_true') # enable init of ROOT environment vars to speedup init/startup
 parser.add_argument('--action-logfile', help='Logfilename for action logs. If none given, pipeline_action_#PID.log will be used')
 parser.add_argument('--metric-logfile', help='Logfilename for metric logs. If none given, pipeline_metric_#PID.log will be used')
 args = parser.parse_args()
@@ -1001,6 +1002,31 @@ def execute(self):
         psutil.cpu_percent(interval=None)
         os.environ['JOBUTILS_SKIPDONE'] = "ON"
 
+        def speedup_ROOT_Init():
+               """initialize some env variables that speed up ROOT init
+               and prevent ROOT from spawning many short-lived child
+               processes"""
+
+               # a) the PATH for system libraries
+               # search taken from ROOT TUnixSystem
+               cmd='LD_DEBUG=libs LD_PRELOAD=DOESNOTEXIST ls /tmp/DOESNOTEXIST 2>&1 | grep -m 1 "system search path" | sed \'s/.*=//g\' | awk \'//{print $1}\''
+               proc = subprocess.Popen([cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
+               libpath, err = proc.communicate()
+               if (args.rootinit_speedup):
+                  print ("setting up ROOT system")
+                  os.environ['ROOT_LDSYSPATH'] = libpath.decode()
+
+               # b) the PATH for compiler includes needed by Cling
+               cmd='LC_ALL=C c++ -xc++ -E -v /dev/null 2>&1 | sed -n \'/^.include/,${/^ \/.*++/{p}}\'' # | sed \'s/ //\''
+               proc = subprocess.Popen([cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE, shell=True)
+               incpath, err = proc.communicate()
+               incpaths = [ line.lstrip() for line in incpath.decode().splitlines() ]
+               joined = ':'.join(incpaths)
+               if (args.rootinit_speedup):
+                  os.environ['ROOT_CPPSYSINCL'] = joined
+
+        speedup_ROOT_Init()
+
         # we make our own "tmp" folder
         # where we can put stuff such as tmp socket files etc (for instance DPL FAIR-MQ sockets)
         # (In case of running within docker/singularity, this may not be so important)

MC/bin/o2dpg_sim_workflow.py

@@ -94,6 +94,7 @@
 # power feature (for playing) --> does not appear in help message
 #  help='Treat smaller sensors in a single digitization')
 parser.add_argument('--combine-smaller-digi', action='store_true', help=argparse.SUPPRESS)
+parser.add_argument('--combine-dpl-devices', action='store_true', help=argparse.SUPPRESS)
 parser.add_argument('--combine-tpc-clusterization', action='store_true', help=argparse.SUPPRESS) #<--- useful for small productions (pp, low interaction rate, small number of events)
 parser.add_argument('--first-orbit', default=0, type=int, help=argparse.SUPPRESS)  # to set the first orbit number of the run for HBFUtils (only used when anchoring)
                                                             # (consider doing this rather in O2 digitization code directly)
@@ -208,7 +209,7 @@ def addWhenActive(detID, needslist, appendstring):
 
 
 def getDPL_global_options(bigshm=False):
-   common=" -b --run --driver-client-backend ws:// "
+   common=" -b --run " # --driver-client-backend ws:// "
    if args.noIPC!=None:
       return common + " --no-IPC "
    if bigshm:
@@ -652,10 +653,12 @@ def createRestDigiTask(name, det='ALLSMALLER'):
             for d in itertools.chain(smallsensorlist, ctp_trigger_inputlist):
                tneeds += [ BKG_HITDOWNLOADER_TASKS[d]['name'] ]
          t = createTask(name=name, needs=tneeds,
-                     tf=tf, cwd=timeframeworkdir, lab=["DIGI","SMALLDIGI"], cpu=NWORKERS)
+                        tf=tf, cwd=timeframeworkdir, lab=["DIGI","SMALLDIGI"], cpu='1')
          t['cmd'] = ('','ln -nfs ../bkg_Hits*.root . ;')[doembedding]
-         t['cmd'] += commondigicmd + ' --skipDet TPC,TRD '
-         t['cmd'] += ' --ccdb-tof-sa'
+         # t['cmd'] += commondigicmd + ' --skipDet TPC,TRD,FT0,FV0,CTP '
+         t['cmd'] += commondigicmd + ' --onlyDet TOF,CPV,EMC,HMP,PHS,ITS,MFT,MID,MCH,FDD'
+         t['cmd'] += ' --ccdb-tof-sa '
+         t['cmd'] += ('',' --combine-devices ')[args.combine_dpl_devices]
          workflow['stages'].append(t)
          return t
 
@@ -682,14 +685,14 @@ def createRestDigiTask(name, det='ALLSMALLER'):
       t = det_to_digitask['ALLSMALLER'] if args.combine_smaller_digi==True else createRestDigiTask(name, det)
       det_to_digitask[det]=t
 
-   if args.combine_smaller_digi==False:
+   if True or args.combine_smaller_digi==False:
       # detectors serving CTP need to be treated somewhat special since CTP needs
       # these inputs at the same time --> still need to be made better
       tneeds = [ContextTask['name']]
       t = createTask(name="ft0fv0ctp_digi_" + str(tf), needs=tneeds,
                      tf=tf, cwd=timeframeworkdir, lab=["DIGI","SMALLDIGI"], cpu='1')
       t['cmd'] = ('','ln -nfs ../bkg_HitsFT0.root . ; ln -nfs ../bkg_HitsFV0.root . ;')[doembedding]
-      t['cmd'] += '${O2_ROOT}/bin/o2-sim-digitizer-workflow ' + getDPL_global_options() + ' -n ' + str(args.ns) + simsoption + ' --onlyDet FT0,FV0,CTP  --interactionRate ' + str(INTRATE) + '  --incontext ' + str(CONTEXTFILE) + ' --disable-write-ini' + putConfigValuesNew()
+      t['cmd'] += '${O2_ROOT}/bin/o2-sim-digitizer-workflow ' + getDPL_global_options() + ' -n ' + str(args.ns) + simsoption + ' --onlyDet FT0,FV0,CTP  --interactionRate ' + str(INTRATE) + '  --incontext ' + str(CONTEXTFILE) + ' --disable-write-ini' + putConfigValuesNew() + ('',' --combine-devices')[args.combine_dpl_devices]
       workflow['stages'].append(t)
       det_to_digitask["FT0"]=t
       det_to_digitask["FV0"]=t
@@ -775,7 +778,7 @@ def getDigiTaskName(det):
                                                     'TPCGasParam',
                                                     'ITSCATrackerParam',
                                                     'MFTClustererParam'])                         \
-                              + " --track-sources " + anchorConfig.get("o2-tof-matcher-workflow-options",{}).get("track-sources",toftracksrcdefault)
+                              + " --track-sources " + anchorConfig.get("o2-tof-matcher-workflow-options",{}).get("track-sources",toftracksrcdefault) + ('',' --combine-devices')[args.combine_dpl_devices]
    workflow['stages'].append(TOFTPCMATCHERtask)
 
    MFTRECOtask = createTask(name='mftreco_'+str(tf), needs=[getDigiTaskName("MFT")], tf=tf, cwd=timeframeworkdir, lab=["RECO"], mem='1500')
@@ -875,7 +878,7 @@ def getDigiTaskName(det):
    PVFINDERtask = createTask(name='pvfinder_'+str(tf), needs=pvfinderneeds, tf=tf, cwd=timeframeworkdir, lab=["RECO"], cpu=NWORKERS, mem='4000')
    PVFINDERtask['cmd'] = '${O2_ROOT}/bin/o2-primary-vertexing-workflow ' \
                          + getDPL_global_options() + putConfigValuesNew(['ITSAlpideParam','MFTAlpideParam', 'pvertexer', 'TPCGasParam'], {"NameConf.mDirMatLUT" : ".."})
-   PVFINDERtask['cmd'] += ' --vertexing-sources ' + pvfinder_sources + ' --vertex-track-matching-sources ' + pvfinder_matching_sources
+   PVFINDERtask['cmd'] += ' --vertexing-sources ' + pvfinder_sources + ' --vertex-track-matching-sources ' + pvfinder_matching_sources + ('',' --combine-source-devices')[args.combine_dpl_devices]
    workflow['stages'].append(PVFINDERtask)
 
    if includeFullQC or includeLocalQC:
@@ -995,7 +998,7 @@ def addQCPerTF(taskName, needs, readerCommand, configFilePath, objectsFile=''):
        svfinder_sources = "ITS,ITS-TPC,TPC-TRD,TPC-TOF,ITS-TPC-TRD,ITS-TPC-TOF"
        if isActive("MID"):
            svfinder_sources += ",MID"
-   SVFINDERtask['cmd'] += ' --vertexing-sources ' + svfinder_sources
+   SVFINDERtask['cmd'] += ' --vertexing-sources ' + svfinder_sources + ('',' --combine-source-devices')[args.combine_dpl_devices]
    workflow['stages'].append(SVFINDERtask)
 
   # -----------
@@ -1043,6 +1046,7 @@ def addQCPerTF(taskName, needs, readerCommand, configFilePath, objectsFile=''):
    AODtask['cmd'] += ' --lpmp-prod-tag ${ALIEN_JDL_LPMPRODUCTIONTAG:-unknown}'
    AODtask['cmd'] += ' --anchor-pass ${ALIEN_JDL_LPMANCHORPASSNAME:-unknown}'
    AODtask['cmd'] += ' --anchor-prod ${ALIEN_JDL_MCANCHOR:-unknown}'
+   AODtask['cmd'] += ('',' --combine-source-devices ')[args.combine_dpl_devices]
    if environ.get('O2DPG_AOD_NOTRUNCATE') != None or environ.get('ALIEN_JDL_O2DPG_AOD_NOTRUNCATE') != None:
       AODtask['cmd'] += ' --enable-truncation 0'  # developer option to suppress precision truncation