Workflow parser, parallelize subtopology generation

Author: davidrohr

DATA/production/qc-workflow.sh

@@ -5,6 +5,11 @@ if [[ -z "$WORKFLOW" ]] || [[ -z "$MYDIR" ]]; then
   exit 1
 fi
 
+if [[ ! -z $GEN_TOPO_QC_JSON_FILE ]]; then
+  exec 101>$GEN_TOPO_QC_JSON_FILE.lock || exit 1
+  flock 101 || exit 1
+fi
+
 if [[ -z $QC_JSON_FROM_OUTSIDE && ! -z $GEN_TOPO_QC_JSON_FILE && -f $GEN_TOPO_QC_JSON_FILE ]]; then
   QC_JSON_FROM_OUTSIDE=$GEN_TOPO_QC_JSON_FILE
 elif [[ -z $QC_JSON_FROM_OUTSIDE ]]; then
@@ -197,4 +202,8 @@ if [[ ! -z "$QC_JSON_FROM_OUTSIDE" ]]; then
   add_W o2-qc "--config json://$QC_JSON_FROM_OUTSIDE ${QC_CONFIG_PARAM:---local --host ${QC_HOST:-localhost}} ${QC_CONFIG}"
 fi
 
+if [[ ! -z $GEN_TOPO_QC_JSON_FILE ]]; then
+  flock -u 101 || exit 1
+fi
+
 true # everything OK up to this point, so the script should return 0 (it is !=0 if the last check failed)

DATA/tools/parse

@@ -1,5 +1,7 @@
 #!/usr/bin/env python3
 import os
+import subprocess
+import threading
 import sys
 import shlex
 import tempfile
@@ -103,7 +105,8 @@ for line in f:
             if NO_PROCESSING_MODE and len(args) > 2:
                 print('Cannot use DPL workflow together with DD mode', os.environ['DDMODE'])
                 raise
-            for i in range(2, len(args)):
+                
+            def processSubtopology(i, args, tmpdir, reconodes, reconodesmin, recoworkflows, calibworkflows, calibworkflowsdds):
                 filename = tmpdir + '/wf' + str(i) + '.dds'
                 calibcores = '1'
                 if args[i].startswith('reco'):
@@ -131,8 +134,8 @@ for line in f:
                 if os.environ['WORKFLOWMODE'] == 'print':
                   command += ' > ' + filename
                 print('Running DPL command', command)
-                retVal = os.system(command)
-                if retVal != 0:
+                retVal = subprocess.run(command, shell=True)
+                if retVal.returncode != 0:
                     print('Error (' + str(retVal) + ') running command', command)
                     ftmp = open(filename, 'r')
                     rg = re.compile('^<topology')
@@ -153,9 +156,17 @@ for line in f:
                                 isempty = 0
                                 break
                         if isempty:
-                            continue
+                            return
                     calibworkflows.append(filename)
                     calibworkflowsdds.append(filename + ':' + calibcores)
+            
+            subtopologyThreads = []
+            for i in range(2, len(args)):
+                t = threading.Thread(target = processSubtopology, args = (i, args, tmpdir, reconodes, reconodesmin, recoworkflows, calibworkflows, calibworkflowsdds))
+                t.start()
+                subtopologyThreads.append(t)
+            for t in subtopologyThreads:
+                t.join()
             if reco_num_nodes_override > 0:
                 reconodes = reco_num_nodes_override
                 reconodesmin = min(reconodes, reconodesmin)