Helpers and Tooling (merge, insert) for graph workflows; Analysis example

Author: benedikt-voelkel

MC/bin/README.md

@@ -0,0 +1,71 @@
+# Workflow editing
+
+The tool `$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py` provides some management of workflow files.
+
+## General help
+
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py [sub-command] --help
+```
+shows the available sub-commands and for each sub-command, a dedicated help message is provided.
+
+
+## Create an empty workflow file
+
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py create my_workflow
+```
+creates a new file `my_workflow.json` (the extension `.json` can be left out in the command and would be added automatically)
+
+## Add task skeletons to a workflow file
+
+New task skeletons can be added with its name by
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py create my_workflow --add-task task1 [task2 [...]]
+```
+
+Regarding the command line to be executet, the required `${O2_ROOT}/share/scripts/jobutils.sh; taskwrapper` is prepended automatically.
+
+## Update number of workers (in case of using relative number of workers)
+
+The number of workers can be updated by (in this case specifying 9 workers)
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py nworkers my_workflow 9
+```
+
+## Merge 2 workflow files
+
+Merging of 2 workflow files is done via
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py merge workflow1 workflow2 [-o workflow_merged]
+```
+
+If no output filename is provided, the default will be `workflow_merged`. Of course, after that the number of workers can be updated based on the merged one.
+
+## Inspect workflow
+
+This doesn't do much at the moment, but can be foreseen to be equipped with more functionality
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py inspect my_workflow --summary
+```
+yields a very brief summary of `my_workflow`, whereas
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py inspect my_workflow --check
+```
+conducts a quick sanity check, for instance checking whether any task names are duplicated or any dependencies are missing.
+
+## Modifying a single task
+A task can be updated via the command line and the is no need to do so inside the `JSON` file. To change the dependent tasks, for instance, do
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py modify --needs dep_task_1 [dep_task_2 [...]]
+```
+As usual, type
+```bash
+$O2DPG_ROOT/MC/bin/o2dpg-workflow-tools.py modify --help
+```
+to see all options.
+
+
+
+
+

MC/bin/o2dpg-workflow-tools.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python3
+
+import sys
+from os.path import join, dirname, exists
+import argparse
+
+sys.path.append(join(dirname(__file__), '.', 'o2dpg_workflow_utils'))
+
+from o2dpg_workflow_utils import createTask, read_workflow, dump_workflow, check_workflow, update_workflow_resource_requirements, make_workflow_filename
+
+def extend(args):
+    """extend a workflow by another one
+
+    The overall configuration from the original workflow
+    is kept
+    """
+    # load workflows
+    workflow_orig = read_workflow(args.orig_wf)
+    workflow_extend = read_workflow(args.extend_wf)
+
+    # extend
+    workflow_orig.extend(workflow_extend)
+
+    # dump in new file
+    filename = args.output if args.output else args.orig_wf
+    dump_workflow(workflow_orig, filename)
+
+
+def create(args):
+    """create an empty workflow skeleton or add task skeletons to existing workflow
+    """
+    filename = make_workflow_filename(args.file)
+    if not args.add_task and exists(filename):
+        print(f"Workflow file {filename} does already exist. Delete it and try again")
+        return
+    if not args.add_task or not exists(filename):
+        # just create an empty workflow
+        dump_workflow([], filename)
+    if args.add_task:
+        # add another task skeleton with name
+        workflow = read_workflow(filename)
+        for name in args.add_task:
+            workflow.append(createTask(name=name))
+        dump_workflow(workflow, filename)
+
+
+def find_task(workflow, task_name):
+    for s in workflow:
+        if s["name"] == task_name:
+            return s
+    return None
+
+
+def modify(args):
+
+    if args.task:
+        workflow = read_workflow(args.file)
+        # try to find the requested task
+        task = find_task(workflow, args.task)
+        if not task:
+            print(f"Task with name {args.task} does not exist")
+            exit(1)
+        for attr in ("name", "needs", "timeframe", "cwd", "labels", "cmd"):
+            if hasattr(args, attr) and getattr(args, attr) is not None:
+                task[attr] = getattr(args, attr)
+        for attr in ("cpu", "relative_cpu", "mem"):
+            if hasattr(args, attr) and getattr(args, attr) is not None:
+                task["resources"][attr] = getattr(args, attr)
+
+        dump_workflow(workflow, args.file)
+
+
+def nworkers(args):
+    workflow = read_workflow(args.file)
+    update_workflow_resource_requirements(workflow, args.jobs)
+    dump_workflow(workflow, args.file)
+
+
+def inspect(args):
+    """Inspecting a workflow
+
+    This is at the moment more show-casing what one could do
+    """
+    workflow = read_workflow(args.file)
+    if args.check:
+        check_workflow(workflow)
+    if args.summary:
+        summary_workflow(workflow)
+    if args.task:
+        task = find_task(workflow, args.task)
+        if not task:
+            print(f"Task with name {args.task}")
+            exit(1)
+        print("Here are the requested task information")
+        print(task)
+
+
+def main():
+
+    parser = argparse.ArgumentParser(description='Create an ALICE (Run3) MC simulation workflow')
+
+    sub_parsers = parser.add_subparsers(dest="command")
+
+    create_parser = sub_parsers.add_parser("create", help="manage a workflow")
+    create_parser.set_defaults(func=create)
+    create_parser.add_argument("file", help="workflow file to be created or modifed")
+    create_parser.add_argument("--add-task", dest="add_task", nargs="+", help="add named tasks to workflow file")
+
+    # Append to (sim) workflow
+    merge_parser = sub_parsers.add_parser("merge", help="append stages")
+    merge_parser.set_defaults(func=extend)
+    merge_parser.add_argument("orig_wf", help="original workflow")
+    merge_parser.add_argument("extend_wf", help="workflow JSON to be merged to orig")
+    merge_parser.add_argument("--output", "-o", help="extended workflow output file name", default="workflow_merged.json")
+
+    nworker_parser = sub_parsers.add_parser("nworkers", help="update number of workers")
+    nworker_parser.set_defaults(func=nworkers)
+    nworker_parser.add_argument("file", help="the workflow file to be modified")
+    nworker_parser.add_argument("jobs", type=int, help="number of workers to recompute relative cpu")
+
+    modify_parser = sub_parsers.add_parser("modify", help="modify a task")
+    modify_parser.set_defaults(func=modify)
+    modify_parser.add_argument("file", help="the workflow file to be modified")
+    modify_parser.add_argument("task", help="name of task to be modified")
+    # not allowing for changing the name at the moment as this also goes into the log-file name
+    #modify_parser.add_argument("--name", help="new name of this task")
+    modify_parser.add_argument("--needs", nargs="+", help="required tasks to be executed before this one")
+    modify_parser.add_argument("--timeframe", type=int, help="timeframe")
+    modify_parser.add_argument("--cwd", help="current working directory of this task")
+    modify_parser.add_argument("--labels", nargs="+", help="attached labels")
+    modify_parser.add_argument("--cpu", type=int, help="absolute number of workers to be used for this task")
+    modify_parser.add_argument("--relative-cpu", dest="relative_cpu", type=float, help="realtive fraction of maximum number of available workers")
+    modify_parser.add_argument("--mem", type=int, help="estimated memory")
+    modify_parser.add_argument("--cmd", help="command line to be executed")
+
+    inspect_parser = sub_parsers.add_parser("inspect", help="inspect a workflow")
+    inspect_parser.set_defaults(func=inspect)
+    inspect_parser.add_argument("file", help="Workflow file to inspect")
+    inspect_parser.add_argument("--summary", action="store_true", help="print summary of workflow")
+    inspect_parser.add_argument("--check", action="store_true", help="Check sanity of workflow") 
+
+    args = parser.parse_args()
+
+    if not hasattr(args, "func"):
+        parser.parse_args(["--help"])
+        exit(0)
+
+    args.func(args)
+
+
+if __name__ == "__main__":
+    # provide this also stand-alone if called directly from the interpreter
+    main()

MC/bin/o2dpg_sim_workflow.py

@@ -17,11 +17,17 @@
 #                             -col pp -eA 2.510 -proc "ccbar"  --embedding
 # 
 
+import sys
 import argparse
 from os import environ
+from os.path import join, dirname
 import json
 import array as arr
 
+sys.path.append(join(dirname(__file__), '.', 'o2dpg_workflow_utils'))
+
+from o2dpg_workflow_utils import createTask, dump_workflow
+
 parser = argparse.ArgumentParser(description='Create an ALICE (Run3) MC simulation workflow')
 
 parser.add_argument('-ns',help='number of signal events / timeframe', default=20)
@@ -96,20 +102,6 @@
 workflow={}
 workflow['stages'] = []
 
-def relativeCPU(n_rel, n_workers=NWORKERS):
-    # compute number of CPUs from a given number of workers
-    # n_workers and a fraction n_rel
-    # catch cases where n_rel > 1 or n_workers * n_rel
-    return round(min(n_workers, n_workers * n_rel), 2)
-
-taskcounter=0
-def createTask(name='', needs=[], tf=-1, cwd='./', lab=[], cpu=1, relative_cpu=None, mem=500):
-    if relative_cpu is not None:
-        # Re-compute, if relative number of CPUs requested
-        cpu = relativeCPU(relative_cpu)
-    global taskcounter
-    taskcounter = taskcounter + 1
-    return { 'name': name, 'cmd':'', 'needs': needs, 'resources': { 'cpu': cpu , 'mem': mem }, 'timeframe' : tf, 'labels' : lab, 'cwd' : cwd }
 
 def getDPL_global_options(bigshm=False,nosmallrate=False):
    common="-b --run --fairmq-ipc-prefix ${FAIRMQ_IPC_PREFIX:-./} --driver-client-backend ws:// " + ('--rate 1000','')[nosmallrate]
@@ -535,21 +527,6 @@ def createRestDigiTask(name, det='ALLSMALLER'):
      workflow['stages'].append(TFcleanup);
 
 
-def trimString(cmd):
-  return ' '.join(cmd.split())
-
-# insert taskwrapper stuff
-for s in workflow['stages']:
-  s['cmd']='. ${O2_ROOT}/share/scripts/jobutils.sh; taskwrapper ' + s['name']+'.log \'' + s['cmd'] + '\''
-
-# remove whitespaces etc
-for s in workflow['stages']:
-  s['cmd']=trimString(s['cmd'])
-
-
-# write workflow to json
-workflowfile=args.o
-with open(workflowfile, 'w') as outfile:
-    json.dump(workflow, outfile, indent=2)
+dump_workflow(workflow["stages"], args.o)
 
 exit (0)