Execution

The scheduling abstraction dispatching one job per recording — sequentially, or in parallel via Dask.

execution

Execution backends for dispatching one pipeline job per recording.

MEEGFlow's job manager (MEEGFlowPipeline.run_pipeline) discovers the list of recordings to process, then dispatches one self-contained job per recording through the backend selected here:

• sequential (default): today's single-process, in-order loop. Requires no new dependency.
• local: an in-process Dask cluster (distributed.LocalCluster), comparable to concurrent.futures.ProcessPoolExecutor or joblib.
• slurm / pbs / sge / lsf / htcondor: a dask-jobqueue cluster, submitting one Dask worker job per HPC scheduler job.

dask and dask-jobqueue are optional dependencies (see extras_require in setup.py) and are only imported lazily, inside the functions that need them, so importing this module (and therefore meegflow.pipeline) never requires Dask to be installed unless a non-sequential backend is actually requested.

See docs/dask_parallel_execution.md for the full design rationale.

ExecutionConfig dataclass

Execution backend selection, parsed from the execution config block.

Parameters

backend : str One of 'sequential' (default), 'local', or a dask-jobqueue cluster type ('slurm', 'pbs', 'sge', 'lsf', 'htcondor'). n_workers : int Number of parallel workers (local processes, or cluster jobs). Ignored for the sequential backend. cluster_kwargs : dict Extra keyword arguments forwarded verbatim to the underlying Dask cluster constructor (e.g. queue, cores, memory, walltime for dask-jobqueue backends).

Source code in src/meegflow/execution.py

@dataclass
class ExecutionConfig:
    """Execution backend selection, parsed from the ``execution`` config block.

    Parameters
    ----------
    backend : str
        One of ``'sequential'`` (default), ``'local'``, or a
        ``dask-jobqueue`` cluster type (``'slurm'``, ``'pbs'``, ``'sge'``,
        ``'lsf'``, ``'htcondor'``).
    n_workers : int
        Number of parallel workers (local processes, or cluster jobs).
        Ignored for the ``sequential`` backend.
    cluster_kwargs : dict
        Extra keyword arguments forwarded verbatim to the underlying Dask
        cluster constructor (e.g. ``queue``, ``cores``, ``memory``,
        ``walltime`` for ``dask-jobqueue`` backends).
    """

    backend: str = SEQUENTIAL_BACKEND
    n_workers: int = 1
    cluster_kwargs: Dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_config(cls, config: Optional[Dict[str, Any]]) -> "ExecutionConfig":
        """Parse the ``execution`` block of a pipeline config, if present.

        Parameters
        ----------
        config : dict or None
            The full pipeline configuration dictionary. Recognizes a top-level
            ``execution`` mapping with keys ``backend``, ``n_workers``, and
            ``cluster_kwargs``. Missing or absent -> sequential execution.

        Returns
        -------
        ExecutionConfig
        """
        exec_cfg = (config or {}).get("execution") or {}
        backend = exec_cfg.get("backend", SEQUENTIAL_BACKEND)
        if backend not in KNOWN_BACKENDS:
            raise ValueError(
                f"Unknown execution backend '{backend}'. Choose from: {KNOWN_BACKENDS}."
            )
        return cls(
            backend=backend,
            n_workers=int(exec_cfg.get("n_workers", 1)),
            cluster_kwargs=dict(exec_cfg.get("cluster_kwargs", {}) or {}),
        )

from_config classmethod

from_config(config)

Parse the execution block of a pipeline config, if present.

Parameters

config : dict or None The full pipeline configuration dictionary. Recognizes a top-level execution mapping with keys backend, n_workers, and cluster_kwargs. Missing or absent -> sequential execution.

Returns

ExecutionConfig

Source code in src/meegflow/execution.py

@classmethod
def from_config(cls, config: Optional[Dict[str, Any]]) -> "ExecutionConfig":
    """Parse the ``execution`` block of a pipeline config, if present.

    Parameters
    ----------
    config : dict or None
        The full pipeline configuration dictionary. Recognizes a top-level
        ``execution`` mapping with keys ``backend``, ``n_workers``, and
        ``cluster_kwargs``. Missing or absent -> sequential execution.

    Returns
    -------
    ExecutionConfig
    """
    exec_cfg = (config or {}).get("execution") or {}
    backend = exec_cfg.get("backend", SEQUENTIAL_BACKEND)
    if backend not in KNOWN_BACKENDS:
        raise ValueError(
            f"Unknown execution backend '{backend}'. Choose from: {KNOWN_BACKENDS}."
        )
    return cls(
        backend=backend,
        n_workers=int(exec_cfg.get("n_workers", 1)),
        cluster_kwargs=dict(exec_cfg.get("cluster_kwargs", {}) or {}),
    )

run_sequential

run_sequential(recordings, reader, output_root, config, step_functions, io_backend)

Process recordings one at a time, in this process (today's default behavior).

Parameters

recordings : list of dict Output of reader.find_recordings(...). reader : DatasetReader Reader used to load each recording's files. output_root : str or Path, optional Derivatives root override. config : dict Full pipeline configuration. step_functions : dict Mapping of step name -> callable (built-in + custom). io_backend : str MNE IO backend used to read raw files.

Returns

all_results : dict Mapping {subject: [result_or_error, ...]}, matching the shape returned by run_dask.

Source code in src/meegflow/execution.py

def run_sequential(
    recordings: List[Dict[str, Any]],
    reader: "DatasetReader",
    output_root: Optional[Union[str, Path]],
    config: Dict[str, Any],
    step_functions: Dict[str, Callable],
    io_backend: str,
) -> Dict[str, List[Dict[str, Any]]]:
    """Process recordings one at a time, in this process (today's default behavior).

    Parameters
    ----------
    recordings : list of dict
        Output of ``reader.find_recordings(...)``.
    reader : DatasetReader
        Reader used to load each recording's files.
    output_root : str or Path, optional
        Derivatives root override.
    config : dict
        Full pipeline configuration.
    step_functions : dict
        Mapping of step name -> callable (built-in + custom).
    io_backend : str
        MNE IO backend used to read raw files.

    Returns
    -------
    all_results : dict
        Mapping ``{subject: [result_or_error, ...]}``, matching the shape
        returned by ``run_dask``.
    """
    from rich.progress import (
        BarColumn,
        Progress,
        SpinnerColumn,
        TaskProgressColumn,
        TextColumn,
        TimeRemainingColumn,
    )

    from .pipeline import process_recording

    all_results: Dict[str, List[Dict[str, Any]]] = {}

    with Progress(
        SpinnerColumn(),
        TextColumn("[progress.description]{task.description}"),
        BarColumn(),
        TaskProgressColumn(),
        TimeRemainingColumn(),
    ) as progress:
        overall_task = progress.add_task("[green]Processing recordings", total=len(recordings))

        for i, recording in enumerate(recordings):
            paths = recording["paths"]
            metadata = recording["metadata"]
            recording_name = recording["recording_name"]
            subject_key = _subject_key(metadata)

            progress.update(overall_task, description=f"[cyan]{recording_name}")

            try:
                result = process_recording(
                    reader=reader,
                    output_root=output_root,
                    config=config,
                    step_functions=step_functions,
                    paths=paths,
                    metadata=metadata,
                    io_backend=io_backend,
                )
                all_results.setdefault(subject_key, []).append(result)
                logger.info(f"Successfully completed {recording_name}")
            except Exception as exc:
                # Do not stop the whole batch if one recording fails.
                logger.error(f"Error processing {recording_name}: {exc}")
                all_results.setdefault(subject_key, []).append({"error": str(exc)})
            finally:
                progress.update(overall_task, completed=i + 1)

    return all_results

run_dask

run_dask(recordings, reader, output_root, config, io_backend, exec_config)

Dispatch one job per recording through a Dask cluster.

Used for the local backend and every dask-jobqueue backend (slurm, pbs, sge, lsf, htcondor). Preserves the sequential backend's contract: one recording's failure is captured as an {'error': ...} entry rather than aborting the batch, and results are returned in the same {subject: [result, ...]} shape as :func:run_sequential.

Parameters

recordings : list of dict Output of reader.find_recordings(...). reader : DatasetReader Reader used to load each recording's files. Must be picklable. output_root : str or Path, optional Derivatives root override. config : dict Full pipeline configuration (plain, picklable dict). io_backend : str MNE IO backend used to read raw files. exec_config : ExecutionConfig Selects and configures the Dask cluster (backend, worker count, cluster-specific kwargs).

Returns

all_results : dict Mapping {subject: [result_or_error, ...]}.

Source code in src/meegflow/execution.py

def run_dask(
    recordings: List[Dict[str, Any]],
    reader: "DatasetReader",
    output_root: Optional[Union[str, Path]],
    config: Dict[str, Any],
    io_backend: str,
    exec_config: ExecutionConfig,
) -> Dict[str, List[Dict[str, Any]]]:
    """Dispatch one job per recording through a Dask cluster.

    Used for the ``local`` backend and every ``dask-jobqueue`` backend
    (``slurm``, ``pbs``, ``sge``, ``lsf``, ``htcondor``). Preserves the
    sequential backend's contract: one recording's failure is captured as an
    ``{'error': ...}`` entry rather than aborting the batch, and results are
    returned in the same ``{subject: [result, ...]}`` shape as
    :func:`run_sequential`.

    Parameters
    ----------
    recordings : list of dict
        Output of ``reader.find_recordings(...)``.
    reader : DatasetReader
        Reader used to load each recording's files. Must be picklable.
    output_root : str or Path, optional
        Derivatives root override.
    config : dict
        Full pipeline configuration (plain, picklable dict).
    io_backend : str
        MNE IO backend used to read raw files.
    exec_config : ExecutionConfig
        Selects and configures the Dask cluster (backend, worker count,
        cluster-specific kwargs).

    Returns
    -------
    all_results : dict
        Mapping ``{subject: [result_or_error, ...]}``.
    """
    from distributed import Client, as_completed

    cluster = _build_cluster(exec_config)
    client = None
    all_results: Dict[str, List[Dict[str, Any]]] = {}
    try:
        client = Client(cluster)
        logger.info(f"Dask dashboard: {client.dashboard_link}")

        futures = {}
        for recording in recordings:
            future = client.submit(
                _run_recording_job,
                reader,
                output_root,
                config,
                recording["paths"],
                recording["metadata"],
                io_backend,
                pure=False,
            )
            futures[future] = recording

        n_total = len(recordings)
        n_done = 0
        for future in as_completed(futures):
            recording = futures[future]
            recording_name = recording["recording_name"]
            subject_key = _subject_key(recording["metadata"])
            n_done += 1
            try:
                result = future.result()
                all_results.setdefault(subject_key, []).append(result)
                logger.info(f"[{n_done}/{n_total}] Completed {recording_name}")
            except Exception as exc:
                # Do not stop the whole batch if one recording fails.
                logger.error(f"[{n_done}/{n_total}] Error processing {recording_name}: {exc}")
                all_results.setdefault(subject_key, []).append({"error": str(exc)})
    finally:
        if client is not None:
            client.close()
        cluster.close()

    return all_results

dispatch

dispatch(recordings, reader, output_root, config, step_functions, io_backend, exec_config)

Dispatch one job per recording, routing to the backend named in exec_config.

Source code in src/meegflow/execution.py

def dispatch(
    recordings: List[Dict[str, Any]],
    reader: "DatasetReader",
    output_root: Optional[Union[str, Path]],
    config: Dict[str, Any],
    step_functions: Dict[str, Callable],
    io_backend: str,
    exec_config: ExecutionConfig,
) -> Dict[str, List[Dict[str, Any]]]:
    """Dispatch one job per recording, routing to the backend named in ``exec_config``."""
    if exec_config.backend == SEQUENTIAL_BACKEND:
        return run_sequential(
            recordings,
            reader=reader,
            output_root=output_root,
            config=config,
            step_functions=step_functions,
            io_backend=io_backend,
        )
    return run_dask(
        recordings,
        reader=reader,
        output_root=output_root,
        config=config,
        io_backend=io_backend,
        exec_config=exec_config,
    )