feat: add logfile and errfile parameters to ensemble run for optional targeted stdout/stderr filenames for ensembles

Author: markcoletti

ipsframework/services.py

@@ -2279,15 +2279,40 @@ def submit_tasks(
         submitted.
 
         Optionally, dask can be used to schedule and run the task pool.
+
+        :param task_pool_name: name of task pool to submit
+        :param block: if True, return when all tasks have been launched.
+          If False, return when all tasks that can be launched immediately have
+          been launched.
+        :param use_dask: if True, use dask to schedule and run the task pool
+        :param dask_nodes: if using dask, number of nodes to use
+        :param dask_ppw: if using dask, number of processes per worker to use;
+          if None, use number of cores per node
+        :param launch_interval: number of seconds to wait between launching tasks
+        :param use_shifter: if True, use shifter to run tasks in a container
+        :param shifter_args: args for running under shifter
+        :param dask_worker_plugin: plugin class for Dask workers
+        :param dask_worker_per_gpu: how many Dask workers per GPU?
+        :param oversubscribe: if True, oversubscribe available resources
+        :param hwthreads: if True, use hardware threads as the basis for
+          resource allocation; if False, use physical cores as the basis for
+          resource allocation
+        :returns: task return value
         """
         start_time = time.time()
-        self._send_monitor_event('IPS_TASK_POOL_BEGIN', 'task_pool = %s ' % task_pool_name)
+        self._send_monitor_event('IPS_TASK_POOL_BEGIN',
+                                 'task_pool = %s ' % task_pool_name)
         task_pool: TaskPool = self.task_pools[task_pool_name]
         retval = task_pool.submit_tasks(
-            block, use_dask, dask_nodes, dask_ppw, launch_interval, use_shifter, shifter_args, dask_worker_plugin, dask_worker_per_gpu, oversubscribe, hwthreads
+                block, use_dask, dask_nodes, dask_ppw, launch_interval,
+                use_shifter, shifter_args, dask_worker_plugin,
+                dask_worker_per_gpu, oversubscribe, hwthreads
         )
         elapsed_time = time.time() - start_time
-        self._send_monitor_event('IPS_TASK_POOL_END', 'task_pool = %s  elapsed time = %.2f S' % (task_pool_name, elapsed_time), elapsed_time=elapsed_time)
+        self._send_monitor_event('IPS_TASK_POOL_END',
+                                 'task_pool = %s  elapsed time = %.2f S' % (
+                                         task_pool_name, elapsed_time),
+                                 elapsed_time=elapsed_time)
         return retval
 
     def get_finished_tasks(self, task_pool_name: str):
@@ -2429,6 +2454,8 @@ def run_ensemble(
         cores_per_instance: Optional[int] = None,
         oversubscribe: bool = False,
         hwthreads: bool = False,
+        logfile: Union[str, os.PathLike] = None,
+        errfile: Union[str, os.PathLike] = None,
     ):
         """Run ensemble of simulations given the template and variables.
 
@@ -2456,6 +2483,14 @@ def run_ensemble(
         config file created from `template` with `?` variables replaced
         with the values from `variables`.
 
+        .. seealso::
+          :py:func:`ipsutil.params_from_csv()` for a convenient way to generate
+          the `variables` dict from a csv file.)
+
+        .. note::
+          `logfile` and `errfile` will write to the working directory local
+          to the ensemble instance.
+
         TODO be able to specify the number of cores per instance
 
         :param template: configuration template file
@@ -2470,6 +2505,8 @@ def run_ensemble(
         :param oversubscribe: Whether to allow oversubscription of nodes
             when launching the ensemble runs. Default is False.
         :param hwthreads: Whether to use hardware threads
+        :param logfile: Optional file name in which to write stdout
+        :param errfile: Optional file name in which to write stderr
         :returns: a list of dicts mapping created subdirs to simulation names
             and their parameters
         """
@@ -2737,13 +2774,20 @@ def send_ensemble_instance_to_portal(ensemble_name: str, data_path: Path) -> Non
             # IPS run pointed to that config file.
             args = [f'--simulation={simulation_filename}', f'--log={log_file}', f'--platform={platform_filename!s}']
 
+            kwargs = {} # optionally add logfile and errfile
+            if logfile:
+                kwargs['logfile'] = logfile
+            if errfile:
+                kwargs['errfile'] = errfile
+
+
             if self.fwk.logger.getEffectiveLevel() == logging.DEBUG:
                 # If we're in debug mode, then also pass the debug flag.
                 # May as well pass in the --verbose, too.
                 args.insert(1, '--debug')
                 args.insert(1, '--verbose')
 
-            self.add_task(task_pool_name, instance[0], 1, working_dir, 'ips.py', *args)
+            self.add_task(task_pool_name, instance[0], 1, working_dir, 'ips.py', *args, keywords=kwargs)
 
         try:
             # Note that we *always* use Dask to run the ensemble tasks
@@ -3014,7 +3058,7 @@ def submit_dask_tasks(
         dask_worker_plugin=None,
         dask_worker_per_gpu=False,
         oversubscribe=False,
-        hwthreads=False,
+        hwthreads=False
     ):
         """Launch tasks in *queued_tasks* using dask.
 
@@ -3310,7 +3354,7 @@ def submit_tasks(
         dask_worker_plugin=None,
         dask_worker_per_gpu=False,
         oversubscribe=False,
-        hwthreads=False,
+        hwthreads=False
     ):
         """Launch tasks in *queued_tasks*.  Finished tasks are handled before
         launching new ones.  If *block* is ``True``, the number of
@@ -3349,22 +3393,29 @@ def submit_tasks(
         :param hwthreads: If True then use hardware threads when launching
             tasks. Default is False.
         :type hwthreads: bool
+        :returns:
         """
-
         if use_dask:
             if TaskPool.dask and TaskPool.distributed and self.serial_pool:
                 self.dask_pool = True
                 if use_shifter and not self.shifter:
-                    self.services.error('Requested to run dask within shifter but shifter not available')
+                    self.services.error('Requested to run dask within shifter '
+                                        'but shifter not available')
                     raise RuntimeError('shifter not found')
                 else:
                     return self.submit_dask_tasks(
-                        block, dask_nodes, dask_ppw, use_shifter, shifter_args, dask_worker_plugin, dask_worker_per_gpu, oversubscribe, hwthreads
+                            block, dask_nodes, dask_ppw, use_shifter,
+                            shifter_args, dask_worker_plugin,
+                            dask_worker_per_gpu, oversubscribe, hwthreads
                     )
             elif not TaskPool.dask or not TaskPool.distributed:
-                raise RuntimeError('Requested use_dask but cannot because import dask or distributed failed')
+                raise RuntimeError(
+                    'Requested use_dask but cannot because import dask or '
+                    'distributed failed')
             elif not self.serial_pool:
-                self.services.warning('Requested use_dask but cannot because multiple processors requested')
+                self.services.warning(
+                    'Requested use_dask but cannot because multiple '
+                    'processors requested')
 
         submit_count = 0
         # Make sure any finished tasks are handled before attempting to submit
@@ -3373,7 +3424,8 @@ def submit_tasks(
         while True:
             if len(self.queued_tasks) == 0:
                 break
-            active_tasks = self.services.launch_task_pool(self.name, launch_interval)
+            active_tasks = self.services.launch_task_pool(self.name,
+                                                          launch_interval)
             for task_name, task_id in active_tasks.items():
                 self.active_tasks[task_id] = self.queued_tasks.pop(task_name)
                 submit_count += 1