Clean possibly leaked CCDB semaphores at workflow start

Use new feature of O2 to scan for leaked CCDB semaphores
related to CCDB caches and clean them up before workflow execution.

To this end, expand the __global_init__ mechanism with a "cmd" (not just
environment variables) field. The pipeline runner will execute such init command
before workflows start.

Solves a problem, where second run/pass of workflow running hangs due
to previously leaked semaphores.

(cherry picked from commit ec4acee)

Author: sawenzel

MC/bin/o2_dpg_workflow_runner.py

@@ -864,11 +864,12 @@ def __init__(self, workflowfile, args, jmax=100):
       self.is_productionmode = args.production_mode == True # os.getenv("ALIEN_PROC_ID") != None
       self.workflowfile = workflowfile
       self.workflowspec = load_json(workflowfile)
-      self.globalenv = self.extract_global_environment(self.workflowspec) # initialize global environment settings
-      for e in self.globalenv:
+      self.globalinit = self.extract_global_environment(self.workflowspec) # initialize global environment settings
+      for e in self.globalinit['env']:
         if os.environ.get(e, None) == None:
-           actionlogger.info("Applying global environment from init section " + str(e) + " : " + str(self.globalenv[e]))
-           os.environ[e] = str(self.globalenv[e])
+           value = self.globalinit['env'][e]
+           actionlogger.info("Applying global environment from init section " + str(e) + " : " + str(value))
+           os.environ[e] = str(value)
 
       # only keep those tasks that are necessary to be executed based on user's filters
       self.workflowspec = filter_workflow(self.workflowspec, args.target_tasks, args.target_labels)
@@ -968,13 +969,33 @@ def extract_global_environment(self, workflowspec):
         """
         init_index = 0 # this has to be the first task in the workflow
         globalenv = {}
+        initcmd = None
         if workflowspec['stages'][init_index]['name'] == '__global_init_task__':
           env = workflowspec['stages'][init_index].get('env', None)
           if env != None:
             globalenv = { e : env[e] for e in env }
+          cmd = workflowspec['stages'][init_index].get('cmd', None)
+          if cmd != 'NO-COMMAND':
+            initcmd = cmd
+
           del workflowspec['stages'][init_index]
 
-        return globalenv
+        return {"env" : globalenv, "cmd" : initcmd }
+
+    def execute_globalinit_cmd(self, cmd):
+        actionlogger.info("Executing global setup cmd " + str(cmd))
+        # perform the global init command (think of cleanup/setup things to be done in any case)
+        p = subprocess.Popen(['/bin/bash','-c', cmd], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
+        stdout, stderr = p.communicate()
+
+        # Check if the command was successful (return code 0)
+        if p.returncode == 0:
+          actionlogger.info(stdout.decode())
+        else:
+          # this should be an error
+          actionlogger.error("Error executing global init function")
+          return False
+        return True
 
     def get_global_task_name(self, name):
         """
@@ -1611,6 +1632,12 @@ def speedup_ROOT_Init():
           self.produce_script(args.produce_script)
           exit (0)
 
+        # execute the user-given global init cmd for this workflow
+        globalinitcmd = self.globalinit.get("cmd", None)
+        if globalinitcmd != None:
+           if not self.execute_globalinit_cmd(globalinitcmd):
+              exit (1)
+
         if args.rerun_from:
           reruntaskfound=False
           for task in self.workflowspec['stages']:

MC/bin/o2dpg_sim_workflow.py

@@ -342,7 +342,9 @@ def extractVertexArgs(configKeyValuesStr, finalDiamondDict):
        globalenv['ALICEO2_CCDB_LOCALCACHE'] = environ.get('ALICEO2_CCDB_LOCALCACHE')
    globalenv['IGNORE_VALIDITYCHECK_OF_CCDB_LOCALCACHE'] = '${ALICEO2_CCDB_LOCALCACHE:+"ON"}'
 
-workflow['stages'].append(createGlobalInitTask(globalenv))
+globalinittask = createGlobalInitTask(globalenv)
+globalinittask['cmd'] = 'o2-ccdb-cleansemaphores -p ${ALICEO2_CCDB_LOCALCACHE}'
+workflow['stages'].append(globalinittask)
 ####
 
 def getDPL_global_options(bigshm=False, ccdbbackend=True):

MC/bin/o2dpg_workflow_utils.py

@@ -140,7 +140,7 @@ def dump_workflow(workflow, filename, meta=None):
     to_dump = deepcopy(workflow)
 
     for s in to_dump:
-        if s["cmd"] and taskwrapper_string not in s["cmd"]:
+        if s["cmd"] and s["name"] != '__global_init_task__' and taskwrapper_string not in s["cmd"]:
             # insert taskwrapper stuff if not there already, only do it if cmd string is not empty
             s['cmd'] = '. ' + taskwrapper_string + ' ' + s['name']+'.log \'' + s['cmd'] + '\''
         # remove unnecessary whitespaces for better readibility