reworked capsul soma-workflow config

- added max_runnung_jobs and max_queued_jobs in config
- set them and use them to dynamically drive the workflow controller during execution
- added commandline parameters for them
- fix: I thing the SWF config was not properly handled in capsul runner and was basically unused.
  We still don't use it the way it was initially intended I think (config environment key should
  correspond to the computing resource if I remember, here we always use the "global" env),
  however at least it is consistent in all parts of the code.

Author: denisri

capsul/engine/module/somaworkflow.py

@@ -58,14 +58,26 @@ def init_settings(capsul_engine):
                     description="Soma-workflow paths translations mapping: "
                     "{local_path: (identifier, uuid)}",
                 ),
+                dict(
+                    name="max_running_jobs",
+                    type="int",
+                    description="max running jobs in queue (overrides "
+                    "soma-workflow config file settings)"
+                ),
+                dict(
+                    name="max_queued_jobs",
+                    type="int",
+                    description="max queued jobs (overrides "
+                    "soma-workflow config file settings)"
+                ),
             ],
         )
     initialize_callbacks(capsul_engine)
 
 
 def activate_configurations():
     """
-    Activate the SPM module (set env variables) from the global configurations,
+    Activate the Soma-Workflow module (set env variables) from the global configurations,
     in order to use them via :mod:`capsul.in_context.spm` functions
     """
     conf = engine.configurations.get("capsul.engine.module.somaworkflow", {})
@@ -84,7 +96,8 @@ def validate_config(widget):
         controller = widget.controller_widget.controller
         with widget.engine.settings as session:
             values = {}
-            for key in ("computing_resource", "config_file", "queue"):
+            for key in ("computing_resource", "config_file", "queue",
+                        "max_running_jobs", "max_queued_jobs"):
                 if getattr(controller, key) in (None, traits.Undefined, ""):
                     values[key] = None
                 else:
@@ -112,7 +125,9 @@ def validate_config(widget):
                     "keep_failed_workflows",
                     "keep_succeeded_workflows",
                     "transfer_paths",
-                    #'path_translations'
+                    "max_running_jobs",
+                    "max_queued_jobs",
+                    #'path_translations',
                 ):
                     setattr(conf, k, values[k])
             if id != widget.config_id:
@@ -156,15 +171,19 @@ def validate_config(widget):
                             desc=''))
     controller.add_trait('transfer_paths',
                          traits.List(traits.Directory,
-                            [], output=False,
-                            desc=''))
+                                     [], output=False,
+                                     desc=''))
     controller.add_trait('path_translations',
                          traits.Dict(
                             key_trait=traits.Directory,
                             value_trait=traits.List(str)(['', ''], minlen=2,
                                                          maxlen=2),
                             value={}, output=False,
                             desc=''))
+    controller.add_trait('max_running_jobs', traits.Int(0, output=False,
+                                                        desc=""))
+    controller.add_trait('max_queued_jobs', traits.Int(0, output=False,
+                                                       desc=""))
     conf = None
     if config_id == "any":
         conf = engine.settings.select_configurations(
@@ -205,6 +224,14 @@ def validate_config(widget):
         controller.path_translations = conf.get(
             "capsul.engine.module.somaworkflow", {}
         ).get("path_translations", {})
+        controller.max_running_jobs = conf.get(
+            "capsul.engine.module.somaworkflow", {}).get(
+            "max_running_jobs", Undefined
+        )
+        controller.max_queued_jobs = conf.get(
+            "capsul.engine.module.somaworkflow", {}).get(
+            "max_queued_jobs", Undefined
+        )
 
     # TODO handle several configs
 
@@ -244,6 +271,9 @@ def sync_from_sc(engine, param=None, value=None):
             "SomaWorkflowConfig" in sc.modules
             and "capsul.engine.module.somaworkflow" in engine._loaded_modules
         ):
+            resource_id = sc.somaworkflow_computing_resource
+            if resource_id is None:
+                resource_id = 'localhost'
             with engine.settings as session:
                 cif = engine.settings.config_id_field
                 config = session.config("somaworkflow", "global")

capsul/engine/run.py

@@ -170,15 +170,16 @@ def start(engine, process, workflow=None, history=True, get_pipeline=False, **kw
     from capsul.pipeline.pipeline_workflow import workflow_from_pipeline
     import soma_workflow.client as swclient
 
-    swf_config = engine.settings.select_configurations(
-        'global', {'somaworkflow': 'config_id=="somaworkflow"'})
     swm = engine.study_config.modules['SomaWorkflowConfig']
     swm.connect_resource(engine.connected_to())
     controller = swm.get_workflow_controller()
     resource_id = swm.get_resource_id()
 
+    query = 'config_id=="somaworkflow"'
+    if resource_id is not None:
+        query = f'computing_resource == "{resource_id}"'
     resource_config_d = engine.settings.select_configurations(
-        resource_id, {'somaworkflow': 'config_id=="somaworkflow"'})
+        'global', {'somaworkflow': query})
     resource_config = resource_config_d.get(
         'capsul.engine.module.somaworkflow', {})
     has_resource_config = ('capsul.engine.module.somaworkflow'
@@ -192,7 +193,9 @@ def start(engine, process, workflow=None, history=True, get_pipeline=False, **kw
     if workflow is None:
         workflow = workflow_from_pipeline(process, environment=environment)
 
-    queue = getattr(resource_config, 'queue', None)
+    queue = resource_config.get('queue', None)
+    max_running_jobs = resource_config.get('max_running_jobs', None)
+    max_queued_jobs = resource_config.get('max_queued_jobs', None)
     #if hasattr(engine.study_config.somaworkflow_computing_resources_config,
                #resource_id):
         #res_conf = getattr(
@@ -201,6 +204,21 @@ def start(engine, process, workflow=None, history=True, get_pipeline=False, **kw
         #queue = res_conf.queue
         #if queue is Undefined:
             #queue = None
+    if max_running_jobs is not None:
+        controller.config.change_running_jobs_limits(queue, max_running_jobs)
+        if controller.config.is_local_resource(
+                controller.config._config_parser, resource_id) \
+                and controller.config.get_scheduler_type() \
+                == 'local_basic':
+            controller.scheduler_config.set_max_proc_nb(max_running_jobs)
+            controller.scheduler_config.set_proc_nb(max_running_jobs)
+    if max_queued_jobs is not None:
+        controller.config.change_queue_limits(queue, max_queued_jobs)
+        if controller.config.is_local_resource(
+                controller.config._config_parser, resource_id) \
+                and controller.config.get_scheduler_type() \
+                == 'local_basic':
+            controller.scheduler_config.set_proc_nb(max_queued_jobs)
     workflow_name = process.name
     wf_id = controller.submit_workflow(workflow=workflow, name=workflow_name,
                                        queue=queue)

capsul/process/runprocess.py

@@ -256,7 +256,7 @@ def run_process_with_distribution(
         resource_id=None, password=None, config=None, rsa_key_pass=None,
         queue=None, input_file_processing=None, output_file_processing=None,
         keep_workflow=False, keep_failed_workflow=False,
-        write_workflow_only=None):
+        write_workflow_only=None, max_running_jobs=None, max_queued_jobs=None):
     ''' Run the given process, either sequentially or distributed through
     Soma-Workflow.
 
@@ -299,6 +299,10 @@ def run_process_with_distribution(
         if specified, this is an output filename where the workflow file will
         be written. The workflow will not be actually run, because int his
         situation the user probably wants to use the workflow on his own.
+    max_running_jobs: int
+        override the queue settings for OCFG_MAX_JOB_RUNNING in soma-workflow
+    max_queued_jobs: int
+        override the queue settings for OCFG_MAX_JOB_IN_QUEUE in soma-workflow
     '''
     if write_workflow_only:
         use_soma_workflow = True
@@ -332,6 +336,28 @@ def run_process_with_distribution(
                         resource_id, {})
             getattr(study_config.somaworkflow_computing_resources_config,
                     resource_id).queue = queue
+        if max_running_jobs is not None or max_queued_jobs is not None:
+            values = {}
+            values['computing_resource'] = resource_id
+            if queue is not None:
+                values['queue'] = queue
+            if max_running_jobs is not None:
+                values['max_running_jobs'] = max_running_jobs
+            if max_queued_jobs is not None:
+                values['max_queued_jobs'] = max_queued_jobs
+            engine = study_config.engine
+            engine.load_module('somaworkflow')
+            with engine.settings as session:
+                query = 'config_id == "somaworkflow"'
+                if resource_id is not None:
+                    query += f' AND computing_resource == "{resource_id}"'
+                conf = session.config('somaworkflow', 'global',
+                                      selection=query)
+                if conf is None:
+                    session.new_config('somaworkflow', 'global', values)
+                else:
+                    for k, v in values.items():
+                        setattr(conf, k, v)
 
     res = study_config.run(process)
     return res
@@ -508,6 +534,12 @@ def main():
     group2.add_option('--queue', dest='queue', default=None,
                       help='Queue to use on the computing resource. If not '
                       'specified, use the default queue.')
+    group2.add_option('--max_running', type=int, default=None,
+                      help='set maximum running jobs (override the queue '
+                      'settings in config file)')
+    group2.add_option('--max_queued', type=int, default=None,
+                      help='set maximum queued jobs (override the queue '
+                      'settings in config file)')
     #group2.add_option('--input-processing', dest='input_file_processing',
                       #default=None, help='Input files processing: local_path, '
                       #'transfer, translate, or translate_shared. The default is '
@@ -585,7 +617,7 @@ def main():
         args += new_args
 
     engine = capsul_engine()
-    engine.load_modules(['fom', 'axon'])
+    engine.load_modules(['fom', 'axon', 'somaworkflow'])
     study_config = engine.study_config
 
     if options.config:
@@ -755,6 +787,8 @@ def main():
     rsa_key_pass = options.rsa_key_pass
     queue = options.queue
     file_processing = []
+    max_running_jobs = options.max_running
+    max_queued_jobs = options.max_queued
 
     study_config.use_soma_workflow = options.soma_workflow
 
@@ -772,7 +806,8 @@ def main():
         password=password, rsa_key_pass=rsa_key_pass,
         queue=queue, input_file_processing=file_processing[0],
         output_file_processing=file_processing[1],
-        write_workflow_only=options.write_workflow)
+        write_workflow_only=options.write_workflow,
+        max_running_jobs=max_running_jobs, max_queued_jobs=max_queued_jobs)
 
     # if there was no exception, we assume the process has succeeded.
     # sys.exit(0)