test_parallel.py

# Authors: The MNE-Python contributors.
# License: BSD-3-Clause
# Copyright the MNE-Python contributors.

import multiprocessing
import os
from contextlib import nullcontext

import pytest

from mne.parallel import parallel_func


@pytest.mark.parametrize(
    "n_jobs",
    [
        None,
        1,
        -1,
        "loky 2",
        "threading 3",
        "multiprocessing 4",
    ],
)
def test_parallel_func(n_jobs):
    """Test Parallel wrapping."""
    joblib = pytest.importorskip("joblib")
    if os.getenv("MNE_FORCE_SERIAL", "").lower() in ("true", "1"):
        pytest.skip("MNE_FORCE_SERIAL is set")

    def fun(x):
        return x * 2

    if isinstance(n_jobs, str):
        backend, n_jobs = n_jobs.split()
        n_jobs = want_jobs = int(n_jobs)
        try:
            func = joblib.parallel_config
        except AttributeError:
            # joblib < 1.3
            func = joblib.parallel_backend
        ctx = func(backend, n_jobs=n_jobs)
        n_jobs = None
    else:
        ctx = nullcontext()
        if n_jobs is not None and n_jobs < 0:
            want_jobs = multiprocessing.cpu_count() + 1 + n_jobs
        else:
            want_jobs = 1
    with ctx:
        parallel, p_fun, got_jobs = parallel_func(fun, n_jobs, verbose="debug")
    assert got_jobs == want_jobs


def test_parallel_func_n_jobs_none():
    """Test n_jobs=None is same as n_jobs=1."""
    joblib = pytest.importorskip("joblib")

    def fun(x):
        return x * 2

    # test that n_jobs=None (outside context) behaves identically to n_jobs=1.
    parallel_none, p_fun_none, n_jobs_none = parallel_func(fun, n_jobs=None)
    parallel_one, p_fun_one, n_jobs_one = parallel_func(fun, n_jobs=1)

    assert parallel_none is parallel_one is list
    assert n_jobs_none == n_jobs_one == 1
    assert p_fun_none is p_fun_one is fun, "fun should not be wrapped but is"

    # Test that n_jobs=None inside a joblib context uses Parallel.
    with joblib.parallel_config(backend="loky", n_jobs=2):
        parallel, p_fun, n_jobs = parallel_func(fun, n_jobs=None)
    assert parallel is not list
    assert n_jobs == 2
    assert fun is not p_fun, "fun should be wrapped but is not"