# The default backend can be changed before running the tests through

# JOBLIB_DEFAULT_PARALLEL_BACKEND environment variable so we need to use

# parallel.DEFAULT_BACKEND here and not

# from joblib.parallel import DEFAULT_BACKEND

"""An exception class with non trivial __init__"""

def __init__(self, a, b, c, d):

if x == 7:

raise (

MyExceptionWithFinickyInit("a", "b", "c", "d")

if custom_exception

else ValueError

)

"""A module-level function so that it can be spawn with

# check the number of effective jobs when `n_jobs=None`

# non-regression test for https://github.com/joblib/joblib/issues/984

with context("threading", n_jobs=backend_n_jobs):

# when using a backend, the default of number jobs will be the one set

# in the backend

assert effective_n_jobs(n_jobs=None) == expected_n_jobs

# without any backend, None will default to a single job

n_tasks = 100

pattern = re.compile(r"(Done\s+\d+ out of \d+ \|)")

class ParallelLog(Parallel):

messages = []

def _print(self, msg):

self.messages.append(msg)

executor = ParallelLog(n_jobs=n_jobs, backend=backend, verbose=10000)

executor([delayed(f)(i) for i in range(n_tasks)])

lens = set()

for message in executor.messages:

if s := pattern.search(message):

a, b = s.span()

lens.add(b - a)

# Check that no default backend relies on the name of the main thread:

# https://github.com/joblib/joblib/issues/180#issuecomment-253266247

# Some programs use a different name for the main thread. This is the case

# for uWSGI apps for instance.

monkeypatch.setattr(

target=threading.current_thread(),

name="name",

value="some_new_name_for_the_main_thread",

)

with warnings.catch_warnings(record=True) as warninfo:

results = Parallel(n_jobs=2, backend=backend)(

delayed(square)(x) for x in range(3)

)

assert results == [0, 1, 4]

# Due to the default parameters of LokyBackend, there is a chance that

# warninfo catches Warnings from worker timeouts. We remove it if it exists

# We also remove DeprecationWarnings which could lead to false negatives.

warninfo = [

w

for w in warninfo

if "worker timeout" not in str(w.message)

and not isinstance(w.message, DeprecationWarning)

]

# Under Python 3.13 if backend='multiprocessing', you will get a

# warning saying that forking a multi-threaded process is not a good idea,

# we ignore them in this test

if backend in [None, "multiprocessing"] or isinstance(

backend, MultiprocessingBackend

):

message_part = "multi-threaded, use of fork() may lead to deadlocks"

warninfo = [w for w in warninfo if message_part not in str(w.message)]

# The multiprocessing backend will raise a warning when detecting that is

# started from the non-main thread. Let's check that there is no false

# positive because of the name change.

with warnings.catch_warnings(record=True) as warninfo:

warnings.simplefilter("always")

parallel_func(backend=backend, inner_n_jobs=inner_n_jobs)

warninfo = [w.message for w in warninfo]

if expected:

if warninfo:

warnings_are_correct = all(

"backed parallel loops cannot" in each.args[0] for each in warninfo

)

# With free-threaded Python, when the outer backend is threading,

# we might see more that one warning

warnings_have_the_right_length = (

len(warninfo) >= 1 if IS_GIL_DISABLED else len(warninfo) == 1

)

return warnings_are_correct and warnings_have_the_right_length

return False

else:

assert not warninfo

# no warnings if inner_n_jobs=1

Parallel(n_jobs=2, backend=parent_backend)(

delayed(_assert_warning_nested)(

backend=child_backend, inner_n_jobs=1, expected=False

)

for _ in range(5)

)

# warnings if inner_n_jobs != 1 and expected

res = Parallel(n_jobs=2, backend=parent_backend)(

delayed(_assert_warning_nested)(

backend=child_backend, inner_n_jobs=2, expected=expected

)

for _ in range(5)

)

# warning handling is not thread safe. One thread might see multiple

# warning or no warning at all.

if parent_backend == "threading":

assert any(res)

else:

is_run_parallel = [False]

def background_thread(is_run_parallel):

with warnings.catch_warnings(record=True) as warninfo:

Parallel(n_jobs=2)(delayed(sleep)(0.1) for _ in range(4))

print(len(warninfo))

is_run_parallel[0] = len(warninfo) == 0

t = threading.Thread(target=background_thread, args=(is_run_parallel,))

t.start()

t.join()

"""Input is mutable when using the threading backend"""

q = Queue(maxsize=5)

Parallel(n_jobs=2, backend="threading")(delayed(q.put)(1) for _ in range(5))

"""Check the keyword argument processing of pmap."""

lst = range(10)

assert [f(x, y=1) for x in lst] == Parallel(n_jobs=n_jobs)(

delayed(f)(x, y=1) for x in lst

lst = range(10)

expected = [f(x, y=1) for x in lst]

with Parallel(n_jobs=4, backend=backend) as p:

# Internally a pool instance has been eagerly created and is managed

# via the context manager protocol

managed_backend = p._backend

# We make call with the managed parallel object several times inside

# the managed block:

assert expected == p(delayed(f)(x, y=1) for x in lst)

assert expected == p(delayed(f)(x, y=1) for x in lst)

# Those calls have all used the same pool instance:

if mp is not None:

assert get_workers(managed_backend) is get_workers(p._backend)

# As soon as we exit the context manager block, the pool is terminated and

# no longer referenced from the parallel object:

if mp is not None:

assert get_workers(p._backend) is None

# It's still possible to use the parallel instance in non-managed mode:

assert expected == p(delayed(f)(x, y=1) for x in lst)

if mp is not None:

"""Check that pmap captures the errors when it is passed an object

class UnpicklableObject(object):

def __reduce__(self):

raise RuntimeError("123")

with raises(PicklingError, match=r"the task to send"):

Parallel(n_jobs=2, backend="loky")(

delayed(id)(UnpicklableObject()) for _ in range(10)

def inspect_byteorder(x):

return x, x.dtype.byteorder

x = np.arange(6).reshape((2, 3)).view(f"{byteorder}i4")

initial_np_byteorder = x.dtype.byteorder

result = Parallel(n_jobs=2, backend="loky", max_nbytes=max_nbytes)(

delayed(inspect_byteorder)(x) for _ in range(3)

)

for x_returned, byteorder_in_worker in result:

assert byteorder_in_worker == initial_np_byteorder

assert byteorder_in_worker == x_returned.dtype.byteorder

# Check that timeout isn't thrown when function is fast enough

assert (

len(

Parallel(n_jobs=2, backend=backend, timeout=30)(

delayed(sleep)(0.001) for x in range(10)

)

)

== 10

# Check that timeout properly fails when function is too slow

with raises(TimeoutError):

Parallel(n_jobs=2, backend=backend, timeout=0.01)(

delayed(sleep)(10) for x in range(10)

# Check that timeout properly fails when function is too slow with

# return_as=generator

with raises(TimeoutError):

list(

Parallel(n_jobs=2, backend=backend, return_as=return_as, timeout=0.1)(

delayed(sleep)(10) for x in range(10)

)

)

# Fast tasks and high timeout should not raise

list(

Parallel(n_jobs=2, backend=backend, return_as=return_as, timeout=10)(

delayed(sleep)(0.01) for x in range(10)

)

# Check that error are captured, and that correct exceptions

# are raised.

if mp is not None:

with raises(ZeroDivisionError):

Parallel(n_jobs=2, backend=backend)(

[delayed(division)(x, y) for x, y in zip((0, 1), (1, 0))]

)

with raises(KeyboardInterrupt):

Parallel(n_jobs=2, backend=backend)(

[delayed(interrupt_raiser)(x) for x in (1, 0)]

)

# Try again with the context manager API

with Parallel(n_jobs=2, backend=backend) as parallel:

assert get_workers(parallel._backend) is not None

original_workers = get_workers(parallel._backend)

with raises(ZeroDivisionError):

parallel([delayed(division)(x, y) for x, y in zip((0, 1), (1, 0))])

# The managed pool should still be available and be in a working

# state despite the previously raised (and caught) exception

assert get_workers(parallel._backend) is not None

# The pool should have been interrupted and restarted:

assert get_workers(parallel._backend) is not original_workers

assert [f(x, y=1) for x in range(10)] == parallel(

delayed(f)(x, y=1) for x in range(10)

)

original_workers = get_workers(parallel._backend)

with raises(KeyboardInterrupt):

parallel([delayed(interrupt_raiser)(x) for x in (1, 0)])

# The pool should still be available despite the exception

assert get_workers(parallel._backend) is not None

# The pool should have been interrupted and restarted:

assert get_workers(parallel._backend) is not original_workers

assert [f(x, y=1) for x in range(10)] == parallel(

delayed(f)(x, y=1) for x in range(10)

), (

parallel._iterating,

parallel.n_completed_tasks,

parallel.n_dispatched_tasks,

parallel._aborting,

)

# Check that the inner pool has been terminated when exiting the

# context manager

assert get_workers(parallel._backend) is None

else:

with raises(KeyboardInterrupt):

Parallel(n_jobs=2)([delayed(interrupt_raiser)(x) for x in (1, 0)])

# wrapped exceptions should inherit from the class of the original

# exception to make it easy to catch them

with raises(ZeroDivisionError):

Parallel(n_jobs=2)([delayed(division)(x, y) for x, y in zip((0, 1), (1, 0))])

with raises(MyExceptionWithFinickyInit):

Parallel(n_jobs=2, verbose=0)(

(delayed(exception_raiser)(i, custom_exception=True) for i in range(30))

def my_generator(raise_at=0):

for i in range(20):

if i == raise_at:

raise ValueError("Iterator Raising Error")

yield i

with Parallel(n_jobs=2, backend=backend) as p:

# The error is raised in the pre-dispatch phase

with raises(ValueError, match="Iterator Raising Error"):

p(delayed(square)(i) for i in my_generator(raise_at=0))

# The error is raised when dispatching a new task after the

# pre-dispatch (likely to happen in a different thread)

with raises(ValueError, match="Iterator Raising Error"):

p(delayed(square)(i) for i in my_generator(raise_at=5))

# Same, but raises long after the pre-dispatch phase

with raises(ValueError, match="Iterator Raising Error"):

"""Test that with only one job, Parallel does act as a iterator."""

queue = list()

def producer():

for i in range(6):

queue.append("Produced %i" % i)

yield i

Parallel(n_jobs=1, batch_size=batch_size, backend=backend)(

delayed(consumer)(queue, x) for x in producer()

)

assert queue == expected_queue

"""Check that using pre_dispatch Parallel does indeed dispatch items

manager = mp.Manager()

queue = manager.list()

def producer():

for i in range(6):

queue.append("Produced %i" % i)

yield i

Parallel(n_jobs=2, batch_size=1, pre_dispatch=3, backend=backend)(

delayed(consumer)(queue, "any") for _ in producer()

)

queue_contents = list(queue)

assert queue_contents[0] == "Produced 0"

# Only 3 tasks are pre-dispatched out of 6. The 4th task is dispatched only

# after any of the first 3 jobs have completed.

first_consumption_index = queue_contents[:4].index("Consumed any")

assert first_consumption_index > -1

produced_3_index = queue_contents.index("Produced 3") # 4th task produced

assert produced_3_index > first_consumption_index

# batching='auto' with the threading backend leaves the effective batch

# size to 1 (no batching) as it has been found to never be beneficial with

# this low-overhead backend.

with Parallel(n_jobs=2, batch_size="auto", backend="threading") as p:

p(delayed(id)(i) for i in range(5000)) # many very fast tasks

with Parallel(n_jobs=2, batch_size="auto", backend=backend) as p:

p(delayed(id)(i) for i in range(5000)) # many very fast tasks

# It should be strictly larger than 1 but as we don't want heisen

# failures on clogged CI worker environment be safe and only check that

# it's a strictly positive number.

"""Make sure that exception raised during dispatch are indeed captured"""

with raises(ValueError):

Parallel(n_jobs=2, pre_dispatch=16, verbose=0)(

delayed(exception_raiser)(i) for i in range(30)

"""Ensure errors for nested joblib cases gets propagated

with raises(ValueError) as excinfo:

Parallel(n_jobs=2, backend=backend)(

delayed(nested_function_outer)(i) for i in range(30)

)

# Check that important information such as function names are visible

# in the final error message reported to the user

report_lines = format_exception(excinfo.type, excinfo.value, excinfo.tb)

report = "".join(report_lines)

assert "nested_function_outer" in report

assert "nested_function_inner" in report

assert "exception_raiser" in report

"""Pretends to run concurrently while running sequentially."""

def configure(self, n_jobs=1, parallel=None, **backend_args):

self.n_jobs = self.effective_n_jobs(n_jobs)

self.parallel = parallel

return n_jobs

def effective_n_jobs(self, n_jobs=1):

if n_jobs < 0:

n_jobs = max(mp.cpu_count() + 1 + n_jobs, 1)

with raises(ValueError, match="Invalid backend:"):

Parallel(backend="unit-testing")

with raises(ValueError, match="Invalid backend:"):

with parallel_config(backend="unit-testing"):

pass

with raises(ValueError, match="Invalid backend:"):

with parallel_config(backend="unit-testing"):

with raises(ValueError) as excinfo:

Parallel(n_jobs=0, backend=backend)._initialize_backend()

assert "n_jobs == 0 in Parallel has no meaning" in str(excinfo.value)

with raises(ValueError) as excinfo:

Parallel(n_jobs=0.5, backend=backend)._initialize_backend()

assert "n_jobs == 0 in Parallel has no meaning" in str(excinfo.value)

with raises(ValueError) as excinfo:

Parallel(n_jobs="2.3", backend=backend)._initialize_backend()

assert "n_jobs could not be converted to int" in str(excinfo.value)

with raises(ValueError) as excinfo:

Parallel(n_jobs="invalid_str", backend=backend)._initialize_backend()

p = Parallel(n_jobs=n_jobs, backend=backend)

assert p._effective_n_jobs() == 2

res = p(delayed(square)(i) for i in range(10))

try:

register_parallel_backend("test_backend", FakeParallelBackend)

assert "test_backend" in BACKENDS

assert BACKENDS["test_backend"] == FakeParallelBackend

finally:

default_backend_orig = parallel.DEFAULT_BACKEND

assert _active_backend_type() == get_default_backend_instance()

try:

register_parallel_backend("threading", BACKENDS["threading"], make_default=True)

assert _active_backend_type() == ThreadingBackend

finally:

# Restore the global default manually

parallel.DEFAULT_BACKEND = default_backend_orig

with warns(UserWarning, match="joblib backend '.*' is not available on.*"):

Parallel(backend="loky")(delayed(square)(i) for i in range(3))

# The below should now work without problems

with parallel_config(backend="loky"):

with context(backend_name, n_jobs=3):

active_backend, active_n_jobs = parallel.get_active_backend()

assert active_n_jobs == 3

assert effective_n_jobs(3) == 3

p = Parallel()

assert p.n_jobs == 3

if backend_name == "multiprocessing":

assert type(active_backend) is MultiprocessingBackend

assert type(p._backend) is MultiprocessingBackend

elif backend_name == "loky":

assert type(active_backend) is LokyBackend

assert type(p._backend) is LokyBackend

elif backend_name == "threading":

assert type(active_backend) is ThreadingBackend

assert type(p._backend) is ThreadingBackend

elif backend_name.startswith("test_"):

assert type(active_backend) is FakeParallelBackend

if backend not in BACKENDS:

monkeypatch.setitem(BACKENDS, backend, FakeParallelBackend)

assert _active_backend_type() == get_default_backend_instance()

# check that this possible to switch parallel backends sequentially

check_backend_context_manager(context, backend)

# The default backend is restored

assert _active_backend_type() == get_default_backend_instance()

# Check that context manager switching is thread safe:

Parallel(n_jobs=2, backend="threading")(

delayed(check_backend_context_manager)(context, b)

for b in all_backends_for_context_manager

if not b

)

# The default backend is again restored

"""Pretends to run conncurrently while running sequentially."""

def __init__(self, param=None):

if param is None:

raise ValueError("param should not be None")

monkeypatch.setitem(BACKENDS, "param_backend", ParameterizedParallelBackend)

assert _active_backend_type() == get_default_backend_instance()

with context("param_backend", param=42, n_jobs=3):

active_backend, active_n_jobs = parallel.get_active_backend()

assert type(active_backend) is ParameterizedParallelBackend

assert active_backend.param == 42

assert active_n_jobs == 3

p = Parallel()

assert p.n_jobs == 3

assert p._backend is active_backend

results = p(delayed(sqrt)(i) for i in range(5))

assert results == [sqrt(i) for i in range(5)]

# The default backend is again restored

assert _active_backend_type() == get_default_backend_instance()

# Check that it's possible to pass a backend instance directly,

# without registration

with context(ParameterizedParallelBackend(param=43), n_jobs=5):

active_backend, active_n_jobs = parallel.get_active_backend()

assert type(active_backend) is ParameterizedParallelBackend

assert active_backend.param == 43

assert active_n_jobs == 5

p = Parallel()

assert p.n_jobs == 5

assert p._backend is active_backend

results = p(delayed(sqrt)(i) for i in range(5))

assert results == [sqrt(i) for i in range(5)]

# The default backend is again restored

assert _active_backend_type() == ThreadingBackend

# Assert that the nested backend does not change the default number of

# jobs used in Parallel

assert Parallel()._effective_n_jobs() == 1

# Assert that the tasks are running only on one process

"""Backend to test backward compatibility with older backends"""

def get_nested_backend(

self,

):

# Older backends only return a backend, without n_jobs indications.

# Check that by default, nested parallel calls will always use the

# ThreadingBackend

with context(backend):

pid_groups = Parallel(n_jobs=2)(delayed(get_nested_pids)() for _ in range(10))

for pid_group in pid_groups:

# Check that by default, nested parallel calls will always use the

# ThreadingBackend

def check_nested_backend(expected_backend_type, expected_n_job):

# Assert that the sequential backend at top level, does not change the

# backend for nested calls.

assert _active_backend_type() == BACKENDS[expected_backend_type]

# Assert that the nested backend in SequentialBackend does not change

# the default number of jobs used in Parallel

expected_n_job = effective_n_jobs(expected_n_job)

assert Parallel()._effective_n_jobs() == expected_n_job

Parallel(n_jobs=1)(

delayed(check_nested_backend)(parallel.DEFAULT_BACKEND, 1) for _ in range(10)

)

with context(backend, n_jobs=n_jobs):

Parallel(n_jobs=1)(

delayed(check_nested_backend)(backend, n_jobs) for _ in range(10)

with context(inner_backend) as ctx:

if context is parallel_config:

backend = ctx["backend"]

if context is parallel_backend: