2057 lines
72 KiB
Python
2057 lines
72 KiB
Python
"""
|
|
Test the parallel module.
|
|
"""
|
|
|
|
# Author: Gael Varoquaux <gael dot varoquaux at normalesup dot org>
|
|
# Copyright (c) 2010-2011 Gael Varoquaux
|
|
# License: BSD Style, 3 clauses.
|
|
|
|
import os
|
|
import sys
|
|
import time
|
|
import mmap
|
|
import weakref
|
|
import warnings
|
|
import threading
|
|
from traceback import format_exception
|
|
from math import sqrt
|
|
from time import sleep
|
|
from pickle import PicklingError
|
|
from contextlib import nullcontext
|
|
from multiprocessing import TimeoutError
|
|
import pytest
|
|
|
|
import joblib
|
|
from joblib import parallel
|
|
from joblib import dump, load
|
|
|
|
from joblib._multiprocessing_helpers import mp
|
|
|
|
from joblib.test.common import np, with_numpy
|
|
from joblib.test.common import with_multiprocessing
|
|
from joblib.test.common import IS_PYPY, force_gc_pypy
|
|
from joblib.testing import (parametrize, raises, check_subprocess_call,
|
|
skipif, warns)
|
|
|
|
if mp is not None:
|
|
# Loky is not available if multiprocessing is not
|
|
from joblib.externals.loky import get_reusable_executor
|
|
|
|
from queue import Queue
|
|
|
|
try:
|
|
import posix
|
|
except ImportError:
|
|
posix = None
|
|
|
|
try:
|
|
from ._openmp_test_helper.parallel_sum import parallel_sum
|
|
except ImportError:
|
|
parallel_sum = None
|
|
|
|
try:
|
|
import distributed
|
|
except ImportError:
|
|
distributed = None
|
|
|
|
from joblib._parallel_backends import SequentialBackend
|
|
from joblib._parallel_backends import ThreadingBackend
|
|
from joblib._parallel_backends import MultiprocessingBackend
|
|
from joblib._parallel_backends import ParallelBackendBase
|
|
from joblib._parallel_backends import LokyBackend
|
|
|
|
from joblib.parallel import Parallel, delayed
|
|
from joblib.parallel import parallel_config
|
|
from joblib.parallel import parallel_backend
|
|
from joblib.parallel import register_parallel_backend
|
|
from joblib.parallel import effective_n_jobs, cpu_count
|
|
|
|
from joblib.parallel import mp, BACKENDS, DEFAULT_BACKEND
|
|
|
|
|
|
RETURN_GENERATOR_BACKENDS = BACKENDS.copy()
|
|
RETURN_GENERATOR_BACKENDS.pop("multiprocessing", None)
|
|
|
|
ALL_VALID_BACKENDS = [None] + sorted(BACKENDS.keys())
|
|
# Add instances of backend classes deriving from ParallelBackendBase
|
|
ALL_VALID_BACKENDS += [BACKENDS[backend_str]() for backend_str in BACKENDS]
|
|
if mp is None:
|
|
PROCESS_BACKENDS = []
|
|
else:
|
|
PROCESS_BACKENDS = ['multiprocessing', 'loky']
|
|
PARALLEL_BACKENDS = PROCESS_BACKENDS + ['threading']
|
|
|
|
if hasattr(mp, 'get_context'):
|
|
# Custom multiprocessing context in Python 3.4+
|
|
ALL_VALID_BACKENDS.append(mp.get_context('spawn'))
|
|
|
|
DefaultBackend = BACKENDS[DEFAULT_BACKEND]
|
|
|
|
|
|
def get_workers(backend):
|
|
return getattr(backend, '_pool', getattr(backend, '_workers', None))
|
|
|
|
|
|
def division(x, y):
|
|
return x / y
|
|
|
|
|
|
def square(x):
|
|
return x ** 2
|
|
|
|
|
|
class MyExceptionWithFinickyInit(Exception):
|
|
"""An exception class with non trivial __init__
|
|
"""
|
|
def __init__(self, a, b, c, d):
|
|
pass
|
|
|
|
|
|
def exception_raiser(x, custom_exception=False):
|
|
if x == 7:
|
|
raise (MyExceptionWithFinickyInit('a', 'b', 'c', 'd')
|
|
if custom_exception else ValueError)
|
|
return x
|
|
|
|
|
|
def interrupt_raiser(x):
|
|
time.sleep(.05)
|
|
raise KeyboardInterrupt
|
|
|
|
|
|
def f(x, y=0, z=0):
|
|
""" A module-level function so that it can be spawn with
|
|
multiprocessing.
|
|
"""
|
|
return x ** 2 + y + z
|
|
|
|
|
|
def _active_backend_type():
|
|
return type(parallel.get_active_backend()[0])
|
|
|
|
|
|
def parallel_func(inner_n_jobs, backend):
|
|
return Parallel(n_jobs=inner_n_jobs, backend=backend)(
|
|
delayed(square)(i) for i in range(3))
|
|
|
|
|
|
###############################################################################
|
|
def test_cpu_count():
|
|
assert cpu_count() > 0
|
|
|
|
|
|
def test_effective_n_jobs():
|
|
assert effective_n_jobs() > 0
|
|
|
|
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@pytest.mark.parametrize(
|
|
"backend_n_jobs, expected_n_jobs",
|
|
[(3, 3), (-1, effective_n_jobs(n_jobs=-1)), (None, 1)],
|
|
ids=["positive-int", "negative-int", "None"]
|
|
)
|
|
@with_multiprocessing
|
|
def test_effective_n_jobs_None(context, backend_n_jobs, expected_n_jobs):
|
|
# 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
|
|
assert effective_n_jobs(n_jobs=None) == 1
|
|
|
|
|
|
###############################################################################
|
|
# Test parallel
|
|
|
|
@parametrize('backend', ALL_VALID_BACKENDS)
|
|
@parametrize('n_jobs', [1, 2, -1, -2])
|
|
@parametrize('verbose', [2, 11, 100])
|
|
def test_simple_parallel(backend, n_jobs, verbose):
|
|
assert ([square(x) for x in range(5)] ==
|
|
Parallel(n_jobs=n_jobs, backend=backend,
|
|
verbose=verbose)(
|
|
delayed(square)(x) for x in range(5)))
|
|
|
|
|
|
@parametrize('backend', ALL_VALID_BACKENDS)
|
|
def test_main_thread_renamed_no_warning(backend, monkeypatch):
|
|
# 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
|
|
warninfo = [w for w in warninfo if "worker timeout" 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.
|
|
assert len(warninfo) == 0
|
|
|
|
|
|
def _assert_warning_nested(backend, inner_n_jobs, expected):
|
|
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 Python nogil, when the outer backend is threading, we might
|
|
# see more that one warning
|
|
warnings_have_the_right_length = (
|
|
len(warninfo) >= 1 if getattr(sys.flags, 'nogil', False)
|
|
else len(warninfo) == 1)
|
|
return warnings_are_correct and warnings_have_the_right_length
|
|
|
|
return False
|
|
else:
|
|
assert not warninfo
|
|
return True
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('parent_backend,child_backend,expected', [
|
|
('loky', 'multiprocessing', True),
|
|
('loky', 'loky', False),
|
|
('multiprocessing', 'multiprocessing', True),
|
|
('multiprocessing', 'loky', True),
|
|
('threading', 'multiprocessing', True),
|
|
('threading', 'loky', True),
|
|
])
|
|
def test_nested_parallel_warnings(parent_backend, child_backend, expected):
|
|
|
|
# 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":
|
|
if IS_PYPY and not any(res):
|
|
# Related to joblib#1426, should be removed once it is solved.
|
|
pytest.xfail(reason="This test often fails in PyPy.")
|
|
assert any(res)
|
|
else:
|
|
assert all(res)
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', ['loky', 'multiprocessing', 'threading'])
|
|
def test_background_thread_parallelism(backend):
|
|
is_run_parallel = [False]
|
|
|
|
def background_thread(is_run_parallel):
|
|
with warnings.catch_warnings(record=True) as warninfo:
|
|
Parallel(n_jobs=2)(
|
|
delayed(sleep)(.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()
|
|
assert is_run_parallel[0]
|
|
|
|
|
|
def nested_loop(backend):
|
|
Parallel(n_jobs=2, backend=backend)(
|
|
delayed(square)(.01) for _ in range(2))
|
|
|
|
|
|
@parametrize('child_backend', BACKENDS)
|
|
@parametrize('parent_backend', BACKENDS)
|
|
def test_nested_loop(parent_backend, child_backend):
|
|
Parallel(n_jobs=2, backend=parent_backend)(
|
|
delayed(nested_loop)(child_backend) for _ in range(2))
|
|
|
|
|
|
def raise_exception(backend):
|
|
raise ValueError
|
|
|
|
|
|
@with_multiprocessing
|
|
def test_nested_loop_with_exception_with_loky():
|
|
with raises(ValueError):
|
|
with Parallel(n_jobs=2, backend="loky") as parallel:
|
|
parallel([delayed(nested_loop)("loky"),
|
|
delayed(raise_exception)("loky")])
|
|
|
|
|
|
def test_mutate_input_with_threads():
|
|
"""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))
|
|
assert q.full()
|
|
|
|
|
|
@parametrize('n_jobs', [1, 2, 3])
|
|
def test_parallel_kwargs(n_jobs):
|
|
"""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))
|
|
|
|
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
def test_parallel_as_context_manager(backend):
|
|
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:
|
|
assert get_workers(p._backend) is None
|
|
|
|
|
|
@with_multiprocessing
|
|
def test_parallel_pickling():
|
|
""" Check that pmap captures the errors when it is passed an object
|
|
that cannot be pickled.
|
|
"""
|
|
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))
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
@parametrize('byteorder', ['<', '>', '='])
|
|
def test_parallel_byteorder_corruption(byteorder):
|
|
|
|
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')(
|
|
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
|
|
np.testing.assert_array_equal(x, x_returned)
|
|
|
|
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
def test_parallel_timeout_success(backend):
|
|
# 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
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
def test_parallel_timeout_fail(backend):
|
|
# 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))
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_error_capture(backend):
|
|
# 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)))
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', BACKENDS)
|
|
def test_error_in_task_iterator(backend):
|
|
|
|
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"):
|
|
p(delayed(square)(i) for i in my_generator(raise_at=19))
|
|
|
|
|
|
def consumer(queue, item):
|
|
queue.append('Consumed %s' % item)
|
|
|
|
|
|
@parametrize('backend', BACKENDS)
|
|
@parametrize('batch_size, expected_queue',
|
|
[(1, ['Produced 0', 'Consumed 0',
|
|
'Produced 1', 'Consumed 1',
|
|
'Produced 2', 'Consumed 2',
|
|
'Produced 3', 'Consumed 3',
|
|
'Produced 4', 'Consumed 4',
|
|
'Produced 5', 'Consumed 5']),
|
|
(4, [ # First Batch
|
|
'Produced 0', 'Produced 1', 'Produced 2', 'Produced 3',
|
|
'Consumed 0', 'Consumed 1', 'Consumed 2', 'Consumed 3',
|
|
# Second batch
|
|
'Produced 4', 'Produced 5', 'Consumed 4', 'Consumed 5'])])
|
|
def test_dispatch_one_job(backend, batch_size, expected_queue):
|
|
""" 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
|
|
assert len(queue) == 12
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
def test_dispatch_multiprocessing(backend):
|
|
""" Check that using pre_dispatch Parallel does indeed dispatch items
|
|
lazily.
|
|
"""
|
|
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
|
|
|
|
assert len(queue) == 12
|
|
|
|
|
|
def test_batching_auto_threading():
|
|
# 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
|
|
assert p._backend.compute_batch_size() == 1
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_batching_auto_subprocesses(backend):
|
|
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.
|
|
assert p._backend.compute_batch_size() > 0
|
|
|
|
|
|
def test_exception_dispatch():
|
|
"""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))
|
|
|
|
|
|
def nested_function_inner(i):
|
|
Parallel(n_jobs=2)(
|
|
delayed(exception_raiser)(j) for j in range(30))
|
|
|
|
|
|
def nested_function_outer(i):
|
|
Parallel(n_jobs=2)(
|
|
delayed(nested_function_inner)(j) for j in range(30))
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
@pytest.mark.xfail(reason="https://github.com/joblib/loky/pull/255")
|
|
def test_nested_exception_dispatch(backend):
|
|
"""Ensure errors for nested joblib cases gets propagated
|
|
|
|
We rely on the Python 3 built-in __cause__ system that already
|
|
report this kind of information to the user.
|
|
"""
|
|
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
|
|
|
|
assert type(excinfo.value) is ValueError
|
|
|
|
|
|
class FakeParallelBackend(SequentialBackend):
|
|
"""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)
|
|
return n_jobs
|
|
|
|
|
|
def test_invalid_backend():
|
|
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'):
|
|
pass
|
|
|
|
|
|
@parametrize('backend', ALL_VALID_BACKENDS)
|
|
def test_invalid_njobs(backend):
|
|
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()
|
|
assert "n_jobs could not be converted to int" in str(excinfo.value)
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
@parametrize('n_jobs', ['2', 2.3, 2])
|
|
def test_njobs_converted_to_int(backend, n_jobs):
|
|
p = Parallel(n_jobs=n_jobs, backend=backend)
|
|
assert p._effective_n_jobs() == 2
|
|
|
|
res = p(delayed(square)(i) for i in range(10))
|
|
assert all(r == square(i) for i, r in enumerate(res))
|
|
|
|
|
|
def test_register_parallel_backend():
|
|
try:
|
|
register_parallel_backend("test_backend", FakeParallelBackend)
|
|
assert "test_backend" in BACKENDS
|
|
assert BACKENDS["test_backend"] == FakeParallelBackend
|
|
finally:
|
|
del BACKENDS["test_backend"]
|
|
|
|
|
|
def test_overwrite_default_backend():
|
|
assert _active_backend_type() == DefaultBackend
|
|
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
|
|
assert _active_backend_type() == DefaultBackend
|
|
|
|
|
|
@skipif(mp is not None, reason="Only without multiprocessing")
|
|
def test_backend_no_multiprocessing():
|
|
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'):
|
|
Parallel()(delayed(square)(i) for i in range(3))
|
|
|
|
|
|
def check_backend_context_manager(context, backend_name):
|
|
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
|
|
assert type(p._backend) is FakeParallelBackend
|
|
|
|
|
|
all_backends_for_context_manager = PARALLEL_BACKENDS[:]
|
|
all_backends_for_context_manager.extend(
|
|
['test_backend_%d' % i for i in range(3)]
|
|
)
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', all_backends_for_context_manager)
|
|
@parametrize('context', [parallel_backend, parallel_config])
|
|
def test_backend_context_manager(monkeypatch, backend, context):
|
|
if backend not in BACKENDS:
|
|
monkeypatch.setitem(BACKENDS, backend, FakeParallelBackend)
|
|
|
|
assert _active_backend_type() == DefaultBackend
|
|
# check that this possible to switch parallel backends sequentially
|
|
check_backend_context_manager(context, backend)
|
|
|
|
# The default backend is restored
|
|
assert _active_backend_type() == DefaultBackend
|
|
|
|
# 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
|
|
assert _active_backend_type() == DefaultBackend
|
|
|
|
|
|
class ParameterizedParallelBackend(SequentialBackend):
|
|
"""Pretends to run conncurrently while running sequentially."""
|
|
|
|
def __init__(self, param=None):
|
|
if param is None:
|
|
raise ValueError('param should not be None')
|
|
self.param = param
|
|
|
|
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_parameterized_backend_context_manager(monkeypatch, context):
|
|
monkeypatch.setitem(BACKENDS, 'param_backend',
|
|
ParameterizedParallelBackend)
|
|
assert _active_backend_type() == DefaultBackend
|
|
|
|
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() == DefaultBackend
|
|
|
|
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_directly_parameterized_backend_context_manager(context):
|
|
assert _active_backend_type() == DefaultBackend
|
|
|
|
# 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() == DefaultBackend
|
|
|
|
|
|
def sleep_and_return_pid():
|
|
sleep(.1)
|
|
return os.getpid()
|
|
|
|
|
|
def get_nested_pids():
|
|
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
|
|
return Parallel(n_jobs=2)(delayed(sleep_and_return_pid)()
|
|
for _ in range(2))
|
|
|
|
|
|
class MyBackend(joblib._parallel_backends.LokyBackend):
|
|
"""Backend to test backward compatibility with older backends"""
|
|
def get_nested_backend(self, ):
|
|
# Older backends only return a backend, without n_jobs indications.
|
|
return super(MyBackend, self).get_nested_backend()[0]
|
|
|
|
|
|
register_parallel_backend('back_compat_backend', MyBackend)
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', ['threading', 'loky', 'multiprocessing',
|
|
'back_compat_backend'])
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_nested_backend_context_manager(context, backend):
|
|
# 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:
|
|
assert len(set(pid_group)) == 1
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('n_jobs', [2, -1, None])
|
|
@parametrize('backend', PARALLEL_BACKENDS)
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_nested_backend_in_sequential(backend, n_jobs, context):
|
|
# 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)(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)
|
|
)
|
|
|
|
|
|
def check_nesting_level(context, inner_backend, expected_level):
|
|
with context(inner_backend) as ctx:
|
|
if context is parallel_config:
|
|
backend = ctx["backend"]
|
|
if context is parallel_backend:
|
|
backend = ctx[0]
|
|
assert backend.nesting_level == expected_level
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('outer_backend', PARALLEL_BACKENDS)
|
|
@parametrize('inner_backend', PARALLEL_BACKENDS)
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_backend_nesting_level(context, outer_backend, inner_backend):
|
|
# Check that the nesting level for the backend is correctly set
|
|
check_nesting_level(context, outer_backend, 0)
|
|
|
|
Parallel(n_jobs=2, backend=outer_backend)(
|
|
delayed(check_nesting_level)(context, inner_backend, 1)
|
|
for _ in range(10)
|
|
)
|
|
|
|
with context(inner_backend, n_jobs=2):
|
|
Parallel()(delayed(check_nesting_level)(context, inner_backend, 1)
|
|
for _ in range(10))
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@parametrize('with_retrieve_callback', [True, False])
|
|
def test_retrieval_context(context, with_retrieve_callback):
|
|
import contextlib
|
|
|
|
class MyBackend(ThreadingBackend):
|
|
i = 0
|
|
supports_retrieve_callback = with_retrieve_callback
|
|
|
|
@contextlib.contextmanager
|
|
def retrieval_context(self):
|
|
self.i += 1
|
|
yield
|
|
|
|
register_parallel_backend("retrieval", MyBackend)
|
|
|
|
def nested_call(n):
|
|
return Parallel(n_jobs=2)(delayed(id)(i) for i in range(n))
|
|
|
|
with context("retrieval") as ctx:
|
|
Parallel(n_jobs=2)(
|
|
delayed(nested_call)(i)
|
|
for i in range(5)
|
|
)
|
|
if context is parallel_config:
|
|
assert ctx["backend"].i == 1
|
|
if context is parallel_backend:
|
|
assert ctx[0].i == 1
|
|
|
|
|
|
###############################################################################
|
|
# Test helpers
|
|
|
|
@parametrize('batch_size', [0, -1, 1.42])
|
|
def test_invalid_batch_size(batch_size):
|
|
with raises(ValueError):
|
|
Parallel(batch_size=batch_size)
|
|
|
|
|
|
@parametrize('n_tasks, n_jobs, pre_dispatch, batch_size',
|
|
[(2, 2, 'all', 'auto'),
|
|
(2, 2, 'n_jobs', 'auto'),
|
|
(10, 2, 'n_jobs', 'auto'),
|
|
(517, 2, 'n_jobs', 'auto'),
|
|
(10, 2, 'n_jobs', 'auto'),
|
|
(10, 4, 'n_jobs', 'auto'),
|
|
(200, 12, 'n_jobs', 'auto'),
|
|
(25, 12, '2 * n_jobs', 1),
|
|
(250, 12, 'all', 1),
|
|
(250, 12, '2 * n_jobs', 7),
|
|
(200, 12, '2 * n_jobs', 'auto')])
|
|
def test_dispatch_race_condition(n_tasks, n_jobs, pre_dispatch, batch_size):
|
|
# Check that using (async-)dispatch does not yield a race condition on the
|
|
# iterable generator that is not thread-safe natively.
|
|
# This is a non-regression test for the "Pool seems closed" class of error
|
|
params = {'n_jobs': n_jobs, 'pre_dispatch': pre_dispatch,
|
|
'batch_size': batch_size}
|
|
expected = [square(i) for i in range(n_tasks)]
|
|
results = Parallel(**params)(delayed(square)(i) for i in range(n_tasks))
|
|
assert results == expected
|
|
|
|
|
|
@with_multiprocessing
|
|
def test_default_mp_context():
|
|
mp_start_method = mp.get_start_method()
|
|
p = Parallel(n_jobs=2, backend='multiprocessing')
|
|
context = p._backend_args.get('context')
|
|
start_method = context.get_start_method()
|
|
assert start_method == mp_start_method
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_no_blas_crash_or_freeze_with_subprocesses(backend):
|
|
if backend == 'multiprocessing':
|
|
# Use the spawn backend that is both robust and available on all
|
|
# platforms
|
|
backend = mp.get_context('spawn')
|
|
|
|
# Check that on recent Python version, the 'spawn' start method can make
|
|
# it possible to use multiprocessing in conjunction of any BLAS
|
|
# implementation that happens to be used by numpy with causing a freeze or
|
|
# a crash
|
|
rng = np.random.RandomState(42)
|
|
|
|
# call BLAS DGEMM to force the initialization of the internal thread-pool
|
|
# in the main process
|
|
a = rng.randn(1000, 1000)
|
|
np.dot(a, a.T)
|
|
|
|
# check that the internal BLAS thread-pool is not in an inconsistent state
|
|
# in the worker processes managed by multiprocessing
|
|
Parallel(n_jobs=2, backend=backend)(
|
|
delayed(np.dot)(a, a.T) for i in range(2))
|
|
|
|
|
|
UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_NO_MAIN = """\
|
|
from joblib import Parallel, delayed
|
|
|
|
def square(x):
|
|
return x ** 2
|
|
|
|
backend = "{}"
|
|
if backend == "spawn":
|
|
from multiprocessing import get_context
|
|
backend = get_context(backend)
|
|
|
|
print(Parallel(n_jobs=2, backend=backend)(
|
|
delayed(square)(i) for i in range(5)))
|
|
"""
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_parallel_with_interactively_defined_functions(backend):
|
|
# When using the "-c" flag, interactive functions defined in __main__
|
|
# should work with any backend.
|
|
if backend == "multiprocessing" and mp.get_start_method() != "fork":
|
|
pytest.skip("Require fork start method to use interactively defined "
|
|
"functions with multiprocessing.")
|
|
code = UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_NO_MAIN.format(backend)
|
|
check_subprocess_call(
|
|
[sys.executable, '-c', code], timeout=10,
|
|
stdout_regex=r'\[0, 1, 4, 9, 16\]')
|
|
|
|
|
|
UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_MAIN = """\
|
|
import sys
|
|
# Make sure that joblib is importable in the subprocess launching this
|
|
# script. This is needed in case we run the tests from the joblib root
|
|
# folder without having installed joblib
|
|
sys.path.insert(0, {joblib_root_folder!r})
|
|
|
|
from joblib import Parallel, delayed
|
|
|
|
def run(f, x):
|
|
return f(x)
|
|
|
|
{define_func}
|
|
|
|
if __name__ == "__main__":
|
|
backend = "{backend}"
|
|
if backend == "spawn":
|
|
from multiprocessing import get_context
|
|
backend = get_context(backend)
|
|
|
|
callable_position = "{callable_position}"
|
|
if callable_position == "delayed":
|
|
print(Parallel(n_jobs=2, backend=backend)(
|
|
delayed(square)(i) for i in range(5)))
|
|
elif callable_position == "args":
|
|
print(Parallel(n_jobs=2, backend=backend)(
|
|
delayed(run)(square, i) for i in range(5)))
|
|
else:
|
|
print(Parallel(n_jobs=2, backend=backend)(
|
|
delayed(run)(f=square, x=i) for i in range(5)))
|
|
"""
|
|
|
|
SQUARE_MAIN = """\
|
|
def square(x):
|
|
return x ** 2
|
|
"""
|
|
SQUARE_LOCAL = """\
|
|
def gen_square():
|
|
def square(x):
|
|
return x ** 2
|
|
return square
|
|
square = gen_square()
|
|
"""
|
|
SQUARE_LAMBDA = """\
|
|
square = lambda x: x ** 2
|
|
"""
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS + ([] if mp is None else ['spawn']))
|
|
@parametrize('define_func', [SQUARE_MAIN, SQUARE_LOCAL, SQUARE_LAMBDA])
|
|
@parametrize('callable_position', ['delayed', 'args', 'kwargs'])
|
|
def test_parallel_with_unpicklable_functions_in_args(
|
|
backend, define_func, callable_position, tmpdir):
|
|
if backend in ['multiprocessing', 'spawn'] and (
|
|
define_func != SQUARE_MAIN or sys.platform == "win32"):
|
|
pytest.skip("Not picklable with pickle")
|
|
code = UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_MAIN.format(
|
|
define_func=define_func, backend=backend,
|
|
callable_position=callable_position,
|
|
joblib_root_folder=os.path.dirname(os.path.dirname(joblib.__file__)))
|
|
code_file = tmpdir.join("unpicklable_func_script.py")
|
|
code_file.write(code)
|
|
check_subprocess_call(
|
|
[sys.executable, code_file.strpath], timeout=10,
|
|
stdout_regex=r'\[0, 1, 4, 9, 16\]')
|
|
|
|
|
|
INTERACTIVE_DEFINED_FUNCTION_AND_CLASS_SCRIPT_CONTENT = """\
|
|
import sys
|
|
import faulthandler
|
|
# Make sure that joblib is importable in the subprocess launching this
|
|
# script. This is needed in case we run the tests from the joblib root
|
|
# folder without having installed joblib
|
|
sys.path.insert(0, {joblib_root_folder!r})
|
|
|
|
from joblib import Parallel, delayed
|
|
from functools import partial
|
|
|
|
class MyClass:
|
|
'''Class defined in the __main__ namespace'''
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
|
|
def square(x, ignored=None, ignored2=None):
|
|
'''Function defined in the __main__ namespace'''
|
|
return x.value ** 2
|
|
|
|
|
|
square2 = partial(square, ignored2='something')
|
|
|
|
# Here, we do not need the `if __name__ == "__main__":` safeguard when
|
|
# using the default `loky` backend (even on Windows).
|
|
|
|
# To make debugging easier
|
|
faulthandler.dump_traceback_later(30, exit=True)
|
|
|
|
# The following baroque function call is meant to check that joblib
|
|
# introspection rightfully uses cloudpickle instead of the (faster) pickle
|
|
# module of the standard library when necessary. In particular cloudpickle is
|
|
# necessary for functions and instances of classes interactively defined in the
|
|
# __main__ module.
|
|
|
|
print(Parallel(backend="loky", n_jobs=2)(
|
|
delayed(square2)(MyClass(i), ignored=[dict(a=MyClass(1))])
|
|
for i in range(5)
|
|
))
|
|
""".format(joblib_root_folder=os.path.dirname(
|
|
os.path.dirname(joblib.__file__)))
|
|
|
|
|
|
@with_multiprocessing
|
|
def test_parallel_with_interactively_defined_functions_loky(tmpdir):
|
|
# loky accepts interactive functions defined in __main__ and does not
|
|
# require if __name__ == '__main__' even when the __main__ module is
|
|
# defined by the result of the execution of a filesystem script.
|
|
script = tmpdir.join('joblib_interactively_defined_function.py')
|
|
script.write(INTERACTIVE_DEFINED_FUNCTION_AND_CLASS_SCRIPT_CONTENT)
|
|
check_subprocess_call(
|
|
[sys.executable, script.strpath],
|
|
stdout_regex=r'\[0, 1, 4, 9, 16\]',
|
|
timeout=None, # rely on faulthandler to kill the process
|
|
)
|
|
|
|
|
|
INTERACTIVELY_DEFINED_SUBCLASS_WITH_METHOD_SCRIPT_CONTENT = """\
|
|
import sys
|
|
# Make sure that joblib is importable in the subprocess launching this
|
|
# script. This is needed in case we run the tests from the joblib root
|
|
# folder without having installed joblib
|
|
sys.path.insert(0, {joblib_root_folder!r})
|
|
|
|
from joblib import Parallel, delayed, hash
|
|
import multiprocessing as mp
|
|
mp.util.log_to_stderr(5)
|
|
|
|
class MyList(list):
|
|
'''MyList is interactively defined by MyList.append is a built-in'''
|
|
def __hash__(self):
|
|
# XXX: workaround limitation in cloudpickle
|
|
return hash(self).__hash__()
|
|
|
|
l = MyList()
|
|
|
|
print(Parallel(backend="loky", n_jobs=2)(
|
|
delayed(l.append)(i) for i in range(3)
|
|
))
|
|
""".format(joblib_root_folder=os.path.dirname(
|
|
os.path.dirname(joblib.__file__)))
|
|
|
|
|
|
@with_multiprocessing
|
|
def test_parallel_with_interactively_defined_bound_method_loky(tmpdir):
|
|
script = tmpdir.join('joblib_interactive_bound_method_script.py')
|
|
script.write(INTERACTIVELY_DEFINED_SUBCLASS_WITH_METHOD_SCRIPT_CONTENT)
|
|
check_subprocess_call([sys.executable, script.strpath],
|
|
stdout_regex=r'\[None, None, None\]',
|
|
stderr_regex=r'LokyProcess',
|
|
timeout=15)
|
|
|
|
|
|
def test_parallel_with_exhausted_iterator():
|
|
exhausted_iterator = iter([])
|
|
assert Parallel(n_jobs=2)(exhausted_iterator) == []
|
|
|
|
|
|
def _cleanup_worker():
|
|
"""Helper function to force gc in each worker."""
|
|
force_gc_pypy()
|
|
time.sleep(.1)
|
|
|
|
|
|
def check_memmap(a):
|
|
if not isinstance(a, np.memmap):
|
|
raise TypeError('Expected np.memmap instance, got %r',
|
|
type(a))
|
|
return a.copy() # return a regular array instead of a memmap
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_auto_memmap_on_arrays_from_generator(backend):
|
|
# Non-regression test for a problem with a bad interaction between the
|
|
# GC collecting arrays recently created during iteration inside the
|
|
# parallel dispatch loop and the auto-memmap feature of Parallel.
|
|
# See: https://github.com/joblib/joblib/pull/294
|
|
def generate_arrays(n):
|
|
for i in range(n):
|
|
yield np.ones(10, dtype=np.float32) * i
|
|
# Use max_nbytes=1 to force the use of memory-mapping even for small
|
|
# arrays
|
|
results = Parallel(n_jobs=2, max_nbytes=1, backend=backend)(
|
|
delayed(check_memmap)(a) for a in generate_arrays(100))
|
|
for result, expected in zip(results, generate_arrays(len(results))):
|
|
np.testing.assert_array_equal(expected, result)
|
|
|
|
# Second call to force loky to adapt the executor by growing the number
|
|
# of worker processes. This is a non-regression test for:
|
|
# https://github.com/joblib/joblib/issues/629.
|
|
results = Parallel(n_jobs=4, max_nbytes=1, backend=backend)(
|
|
delayed(check_memmap)(a) for a in generate_arrays(100))
|
|
for result, expected in zip(results, generate_arrays(len(results))):
|
|
np.testing.assert_array_equal(expected, result)
|
|
|
|
|
|
def identity(arg):
|
|
return arg
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
def test_memmap_with_big_offset(tmpdir):
|
|
fname = tmpdir.join('test.mmap').strpath
|
|
size = mmap.ALLOCATIONGRANULARITY
|
|
obj = [np.zeros(size, dtype='uint8'), np.ones(size, dtype='uint8')]
|
|
dump(obj, fname)
|
|
memmap = load(fname, mmap_mode='r')
|
|
result, = Parallel(n_jobs=2)(delayed(identity)(memmap) for _ in [0])
|
|
assert isinstance(memmap[1], np.memmap)
|
|
assert memmap[1].offset > size
|
|
np.testing.assert_array_equal(obj, result)
|
|
|
|
|
|
def test_warning_about_timeout_not_supported_by_backend():
|
|
with warnings.catch_warnings(record=True) as warninfo:
|
|
Parallel(n_jobs=1, timeout=1)(delayed(square)(i) for i in range(50))
|
|
assert len(warninfo) == 1
|
|
w = warninfo[0]
|
|
assert isinstance(w.message, UserWarning)
|
|
assert str(w.message) == (
|
|
"The backend class 'SequentialBackend' does not support timeout. "
|
|
"You have set 'timeout=1' in Parallel but the 'timeout' parameter "
|
|
"will not be used.")
|
|
|
|
|
|
def set_list_value(input_list, index, value):
|
|
input_list[index] = value
|
|
return value
|
|
|
|
|
|
@pytest.mark.parametrize('n_jobs', [1, 2, 4])
|
|
def test_parallel_return_order_with_return_as_generator_parameter(n_jobs):
|
|
# This test inserts values in a list in some expected order
|
|
# in sequential computing, and then checks that this order has been
|
|
# respected by Parallel output generator.
|
|
input_list = [0] * 5
|
|
result = Parallel(n_jobs=n_jobs, return_as="generator",
|
|
backend='threading')(
|
|
delayed(set_list_value)(input_list, i, i) for i in range(5))
|
|
|
|
# Ensure that all the tasks are completed before checking the result
|
|
result = list(result)
|
|
|
|
assert all(v == r for v, r in zip(input_list, result))
|
|
|
|
|
|
def _sqrt_with_delay(e, delay):
|
|
if delay:
|
|
sleep(30)
|
|
return sqrt(e)
|
|
|
|
|
|
def _test_parallel_unordered_generator_returns_fastest_first(backend, n_jobs):
|
|
# This test submits 10 tasks, but the second task is super slow. This test
|
|
# checks that the 9 other tasks return before the slow task is done, when
|
|
# `return_as` parameter is set to `'generator_unordered'`
|
|
result = Parallel(n_jobs=n_jobs, return_as="generator_unordered",
|
|
backend=backend)(
|
|
delayed(_sqrt_with_delay)(i**2, (i == 1)) for i in range(10))
|
|
|
|
quickly_returned = sorted(next(result) for _ in range(9))
|
|
|
|
expected_quickly_returned = [0] + list(range(2, 10))
|
|
|
|
assert all(
|
|
v == r for v, r in zip(expected_quickly_returned, quickly_returned)
|
|
)
|
|
|
|
del result
|
|
force_gc_pypy()
|
|
|
|
|
|
@pytest.mark.parametrize('n_jobs', [2, 4])
|
|
# NB: for this test to work, the backend must be allowed to process tasks
|
|
# concurrently, so at least two jobs with a non-sequential backend are
|
|
# mandatory.
|
|
@with_multiprocessing
|
|
@parametrize('backend', set(RETURN_GENERATOR_BACKENDS) - {"sequential"})
|
|
def test_parallel_unordered_generator_returns_fastest_first(backend, n_jobs):
|
|
_test_parallel_unordered_generator_returns_fastest_first(backend, n_jobs)
|
|
|
|
|
|
@pytest.mark.parametrize('n_jobs', [2, -1])
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@skipif(distributed is None, reason='This test requires dask')
|
|
def test_parallel_unordered_generator_returns_fastest_first_with_dask(
|
|
n_jobs, context
|
|
):
|
|
with distributed.Client(
|
|
n_workers=2, threads_per_worker=2
|
|
), context("dask"):
|
|
_test_parallel_unordered_generator_returns_fastest_first(None, n_jobs)
|
|
|
|
|
|
@parametrize('backend', ALL_VALID_BACKENDS)
|
|
@parametrize('n_jobs', [1, 2, -2, -1])
|
|
def test_abort_backend(n_jobs, backend):
|
|
delays = ["a"] + [10] * 100
|
|
with raises(TypeError):
|
|
t_start = time.time()
|
|
Parallel(n_jobs=n_jobs, backend=backend)(
|
|
delayed(time.sleep)(i) for i in delays)
|
|
dt = time.time() - t_start
|
|
assert dt < 20
|
|
|
|
|
|
def get_large_object(arg):
|
|
result = np.ones(int(5 * 1e5), dtype=bool)
|
|
result[0] = False
|
|
return result
|
|
|
|
|
|
def _test_deadlock_with_generator(backend, return_as, n_jobs):
|
|
# Non-regression test for a race condition in the backends when the pickler
|
|
# is delayed by a large object.
|
|
with Parallel(n_jobs=n_jobs, backend=backend,
|
|
return_as=return_as) as parallel:
|
|
result = parallel(delayed(get_large_object)(i) for i in range(10))
|
|
next(result)
|
|
next(result)
|
|
del result
|
|
# The gc in pypy can be delayed. Force it to make sure this test does
|
|
# not cause timeout on the CI.
|
|
force_gc_pypy()
|
|
|
|
|
|
@with_numpy
|
|
@parametrize('backend', RETURN_GENERATOR_BACKENDS)
|
|
@parametrize('return_as', ["generator", "generator_unordered"])
|
|
@parametrize('n_jobs', [1, 2, -2, -1])
|
|
def test_deadlock_with_generator(backend, return_as, n_jobs):
|
|
_test_deadlock_with_generator(backend, return_as, n_jobs)
|
|
|
|
|
|
@with_numpy
|
|
@pytest.mark.parametrize('n_jobs', [2, -1])
|
|
@parametrize('return_as', ["generator", "generator_unordered"])
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@skipif(distributed is None, reason='This test requires dask')
|
|
def test_deadlock_with_generator_and_dask(context, return_as, n_jobs):
|
|
with distributed.Client(
|
|
n_workers=2, threads_per_worker=2
|
|
), context("dask"):
|
|
_test_deadlock_with_generator(None, return_as, n_jobs)
|
|
|
|
|
|
@parametrize('backend', RETURN_GENERATOR_BACKENDS)
|
|
@parametrize('return_as', ["generator", "generator_unordered"])
|
|
@parametrize('n_jobs', [1, 2, -2, -1])
|
|
def test_multiple_generator_call(backend, return_as, n_jobs):
|
|
# Non-regression test that ensures the dispatch of the tasks starts
|
|
# immediately when Parallel.__call__ is called. This test relies on the
|
|
# assumption that only one generator can be submitted at a time.
|
|
with raises(RuntimeError,
|
|
match="This Parallel instance is already running"):
|
|
parallel = Parallel(n_jobs, backend=backend, return_as=return_as)
|
|
g = parallel(delayed(sleep)(1) for _ in range(10)) # noqa: F841
|
|
t_start = time.time()
|
|
gen2 = parallel(delayed(id)(i) for i in range(100)) # noqa: F841
|
|
|
|
# Make sure that the error is raised quickly
|
|
assert time.time() - t_start < 2, (
|
|
"The error should be raised immediatly when submitting a new task "
|
|
"but it took more than 2s."
|
|
)
|
|
|
|
del g
|
|
# The gc in pypy can be delayed. Force it to make sure this test does not
|
|
# cause timeout on the CI.
|
|
force_gc_pypy()
|
|
|
|
|
|
@parametrize('backend', RETURN_GENERATOR_BACKENDS)
|
|
@parametrize('return_as', ["generator", "generator_unordered"])
|
|
@parametrize('n_jobs', [1, 2, -2, -1])
|
|
def test_multiple_generator_call_managed(backend, return_as, n_jobs):
|
|
# Non-regression test that ensures the dispatch of the tasks starts
|
|
# immediately when Parallel.__call__ is called. This test relies on the
|
|
# assumption that only one generator can be submitted at a time.
|
|
with Parallel(n_jobs, backend=backend,
|
|
return_as=return_as) as parallel:
|
|
g = parallel(delayed(sleep)(10) for _ in range(10)) # noqa: F841
|
|
t_start = time.time()
|
|
with raises(RuntimeError,
|
|
match="This Parallel instance is already running"):
|
|
g2 = parallel(delayed(id)(i) for i in range(100)) # noqa: F841
|
|
|
|
# Make sure that the error is raised quickly
|
|
assert time.time() - t_start < 2, (
|
|
"The error should be raised immediatly when submitting a new task "
|
|
"but it took more than 2s."
|
|
)
|
|
|
|
# The gc in pypy can be delayed. Force it to make sure this test does not
|
|
# cause timeout on the CI.
|
|
del g
|
|
force_gc_pypy()
|
|
|
|
|
|
@parametrize('backend', RETURN_GENERATOR_BACKENDS)
|
|
@parametrize('return_as_1', ["generator", "generator_unordered"])
|
|
@parametrize('return_as_2', ["generator", "generator_unordered"])
|
|
@parametrize('n_jobs', [1, 2, -2, -1])
|
|
def test_multiple_generator_call_separated(
|
|
backend, return_as_1, return_as_2, n_jobs
|
|
):
|
|
# Check that for separated Parallel, both tasks are correctly returned.
|
|
g = Parallel(n_jobs, backend=backend, return_as=return_as_1)(
|
|
delayed(sqrt)(i ** 2) for i in range(10)
|
|
)
|
|
g2 = Parallel(n_jobs, backend=backend, return_as=return_as_2)(
|
|
delayed(sqrt)(i ** 2) for i in range(10, 20)
|
|
)
|
|
|
|
if return_as_1 == "generator_unordered":
|
|
g = sorted(g)
|
|
|
|
if return_as_2 == "generator_unordered":
|
|
g2 = sorted(g2)
|
|
|
|
assert all(res == i for res, i in zip(g, range(10)))
|
|
assert all(res == i for res, i in zip(g2, range(10, 20)))
|
|
|
|
|
|
@parametrize('backend, error', [
|
|
('loky', True),
|
|
('threading', False),
|
|
('sequential', False),
|
|
])
|
|
@parametrize('return_as_1', ["generator", "generator_unordered"])
|
|
@parametrize('return_as_2', ["generator", "generator_unordered"])
|
|
def test_multiple_generator_call_separated_gc(
|
|
backend, return_as_1, return_as_2, error
|
|
):
|
|
|
|
if (backend == 'loky') and (mp is None):
|
|
pytest.skip("Requires multiprocessing")
|
|
|
|
# Check that in loky, only one call can be run at a time with
|
|
# a single executor.
|
|
parallel = Parallel(2, backend=backend, return_as=return_as_1)
|
|
g = parallel(delayed(sleep)(10) for i in range(10))
|
|
g_wr = weakref.finalize(g, lambda: print("Generator collected"))
|
|
ctx = (
|
|
raises(RuntimeError, match="The executor underlying Parallel")
|
|
if error else nullcontext()
|
|
)
|
|
with ctx:
|
|
# For loky, this call will raise an error as the gc of the previous
|
|
# generator will shutdown the shared executor.
|
|
# For the other backends, as the worker pools are not shared between
|
|
# the two calls, this should proceed correctly.
|
|
t_start = time.time()
|
|
g = Parallel(2, backend=backend, return_as=return_as_2)(
|
|
delayed(sqrt)(i ** 2) for i in range(10, 20)
|
|
)
|
|
|
|
# The gc in pypy can be delayed. Force it to test the behavior when it
|
|
# will eventually be collected.
|
|
force_gc_pypy()
|
|
|
|
if return_as_2 == "generator_unordered":
|
|
g = sorted(g)
|
|
|
|
assert all(res == i for res, i in zip(g, range(10, 20)))
|
|
|
|
assert time.time() - t_start < 5
|
|
|
|
# Make sure that the computation are stopped for the gc'ed generator
|
|
retry = 0
|
|
while g_wr.alive and retry < 3:
|
|
retry += 1
|
|
time.sleep(.5)
|
|
assert time.time() - t_start < 5
|
|
|
|
if parallel._effective_n_jobs() != 1:
|
|
# check that the first parallel object is aborting (the final _aborted
|
|
# state might be delayed).
|
|
assert parallel._aborting
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_memmapping_leaks(backend, tmpdir):
|
|
# Non-regression test for memmapping backends. Ensure that the data
|
|
# does not stay too long in memory
|
|
tmpdir = tmpdir.strpath
|
|
|
|
# Use max_nbytes=1 to force the use of memory-mapping even for small
|
|
# arrays
|
|
with Parallel(n_jobs=2, max_nbytes=1, backend=backend,
|
|
temp_folder=tmpdir) as p:
|
|
p(delayed(check_memmap)(a) for a in [np.random.random(10)] * 2)
|
|
|
|
# The memmap folder should not be clean in the context scope
|
|
assert len(os.listdir(tmpdir)) > 0
|
|
|
|
# Cleaning of the memmap folder is triggered by the garbage
|
|
# collection. With pypy the garbage collection has been observed to be
|
|
# delayed, sometimes up until the shutdown of the interpreter. This
|
|
# cleanup job executed in the worker ensures that it's triggered
|
|
# immediately.
|
|
p(delayed(_cleanup_worker)() for _ in range(2))
|
|
|
|
# Make sure that the shared memory is cleaned at the end when we exit
|
|
# the context
|
|
for _ in range(100):
|
|
if not os.listdir(tmpdir):
|
|
break
|
|
sleep(.1)
|
|
else:
|
|
raise AssertionError('temporary directory of Parallel was not removed')
|
|
|
|
# Make sure that the shared memory is cleaned at the end of a call
|
|
p = Parallel(n_jobs=2, max_nbytes=1, backend=backend)
|
|
p(delayed(check_memmap)(a) for a in [np.random.random(10)] * 2)
|
|
p(delayed(_cleanup_worker)() for _ in range(2))
|
|
|
|
for _ in range(100):
|
|
if not os.listdir(tmpdir):
|
|
break
|
|
sleep(.1)
|
|
else:
|
|
raise AssertionError('temporary directory of Parallel was not removed')
|
|
|
|
|
|
@parametrize('backend',
|
|
([None, 'threading'] if mp is None
|
|
else [None, 'loky', 'threading'])
|
|
)
|
|
def test_lambda_expression(backend):
|
|
# cloudpickle is used to pickle delayed callables
|
|
results = Parallel(n_jobs=2, backend=backend)(
|
|
delayed(lambda x: x ** 2)(i) for i in range(10))
|
|
assert results == [i ** 2 for i in range(10)]
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', PROCESS_BACKENDS)
|
|
def test_backend_batch_statistics_reset(backend):
|
|
"""Test that a parallel backend correctly resets its batch statistics."""
|
|
n_jobs = 2
|
|
n_inputs = 500
|
|
task_time = 2. / n_inputs
|
|
|
|
p = Parallel(verbose=10, n_jobs=n_jobs, backend=backend)
|
|
p(delayed(time.sleep)(task_time) for i in range(n_inputs))
|
|
assert (p._backend._effective_batch_size ==
|
|
p._backend._DEFAULT_EFFECTIVE_BATCH_SIZE)
|
|
assert (p._backend._smoothed_batch_duration ==
|
|
p._backend._DEFAULT_SMOOTHED_BATCH_DURATION)
|
|
|
|
p(delayed(time.sleep)(task_time) for i in range(n_inputs))
|
|
assert (p._backend._effective_batch_size ==
|
|
p._backend._DEFAULT_EFFECTIVE_BATCH_SIZE)
|
|
assert (p._backend._smoothed_batch_duration ==
|
|
p._backend._DEFAULT_SMOOTHED_BATCH_DURATION)
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_backend_hinting_and_constraints(context):
|
|
for n_jobs in [1, 2, -1]:
|
|
assert type(Parallel(n_jobs=n_jobs)._backend) == DefaultBackend
|
|
|
|
p = Parallel(n_jobs=n_jobs, prefer='threads')
|
|
assert type(p._backend) is ThreadingBackend
|
|
|
|
p = Parallel(n_jobs=n_jobs, prefer='processes')
|
|
assert type(p._backend) is DefaultBackend
|
|
|
|
p = Parallel(n_jobs=n_jobs, require='sharedmem')
|
|
assert type(p._backend) is ThreadingBackend
|
|
|
|
# Explicit backend selection can override backend hinting although it
|
|
# is useless to pass a hint when selecting a backend.
|
|
p = Parallel(n_jobs=2, backend='loky', prefer='threads')
|
|
assert type(p._backend) is LokyBackend
|
|
|
|
with context('loky', n_jobs=2):
|
|
# Explicit backend selection by the user with the context manager
|
|
# should be respected when combined with backend hints only.
|
|
p = Parallel(prefer='threads')
|
|
assert type(p._backend) is LokyBackend
|
|
assert p.n_jobs == 2
|
|
|
|
with context('loky', n_jobs=2):
|
|
# Locally hard-coded n_jobs value is respected.
|
|
p = Parallel(n_jobs=3, prefer='threads')
|
|
assert type(p._backend) is LokyBackend
|
|
assert p.n_jobs == 3
|
|
|
|
with context('loky', n_jobs=2):
|
|
# Explicit backend selection by the user with the context manager
|
|
# should be ignored when the Parallel call has hard constraints.
|
|
# In this case, the default backend that supports shared mem is
|
|
# used an the default number of processes is used.
|
|
p = Parallel(require='sharedmem')
|
|
assert type(p._backend) is ThreadingBackend
|
|
assert p.n_jobs == 1
|
|
|
|
with context('loky', n_jobs=2):
|
|
p = Parallel(n_jobs=3, require='sharedmem')
|
|
assert type(p._backend) is ThreadingBackend
|
|
assert p.n_jobs == 3
|
|
|
|
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_backend_hinting_and_constraints_with_custom_backends(
|
|
capsys, context
|
|
):
|
|
# Custom backends can declare that they use threads and have shared memory
|
|
# semantics:
|
|
class MyCustomThreadingBackend(ParallelBackendBase):
|
|
supports_sharedmem = True
|
|
use_threads = True
|
|
|
|
def apply_async(self):
|
|
pass
|
|
|
|
def effective_n_jobs(self, n_jobs):
|
|
return n_jobs
|
|
|
|
with context(MyCustomThreadingBackend()):
|
|
p = Parallel(n_jobs=2, prefer='processes') # ignored
|
|
assert type(p._backend) is MyCustomThreadingBackend
|
|
|
|
p = Parallel(n_jobs=2, require='sharedmem')
|
|
assert type(p._backend) is MyCustomThreadingBackend
|
|
|
|
class MyCustomProcessingBackend(ParallelBackendBase):
|
|
supports_sharedmem = False
|
|
use_threads = False
|
|
|
|
def apply_async(self):
|
|
pass
|
|
|
|
def effective_n_jobs(self, n_jobs):
|
|
return n_jobs
|
|
|
|
with context(MyCustomProcessingBackend()):
|
|
p = Parallel(n_jobs=2, prefer='processes')
|
|
assert type(p._backend) is MyCustomProcessingBackend
|
|
|
|
out, err = capsys.readouterr()
|
|
assert out == ""
|
|
assert err == ""
|
|
|
|
p = Parallel(n_jobs=2, require='sharedmem', verbose=10)
|
|
assert type(p._backend) is ThreadingBackend
|
|
|
|
out, err = capsys.readouterr()
|
|
expected = ("Using ThreadingBackend as joblib backend "
|
|
"instead of MyCustomProcessingBackend as the latter "
|
|
"does not provide shared memory semantics.")
|
|
assert out.strip() == expected
|
|
assert err == ""
|
|
|
|
with raises(ValueError):
|
|
Parallel(backend=MyCustomProcessingBackend(), require='sharedmem')
|
|
|
|
|
|
def test_invalid_backend_hinting_and_constraints():
|
|
with raises(ValueError):
|
|
Parallel(prefer='invalid')
|
|
|
|
with raises(ValueError):
|
|
Parallel(require='invalid')
|
|
|
|
with raises(ValueError):
|
|
# It is inconsistent to prefer process-based parallelism while
|
|
# requiring shared memory semantics.
|
|
Parallel(prefer='processes', require='sharedmem')
|
|
|
|
if mp is not None:
|
|
# It is inconsistent to ask explicitly for a process-based
|
|
# parallelism while requiring shared memory semantics.
|
|
with raises(ValueError):
|
|
Parallel(backend='loky', require='sharedmem')
|
|
with raises(ValueError):
|
|
Parallel(backend='multiprocessing', require='sharedmem')
|
|
|
|
|
|
def _recursive_backend_info(limit=3, **kwargs):
|
|
"""Perform nested parallel calls and introspect the backend on the way"""
|
|
|
|
with Parallel(n_jobs=2) as p:
|
|
this_level = [(type(p._backend).__name__, p._backend.nesting_level)]
|
|
if limit == 0:
|
|
return this_level
|
|
results = p(delayed(_recursive_backend_info)(limit=limit - 1, **kwargs)
|
|
for i in range(1))
|
|
return this_level + results[0]
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('backend', ['loky', 'threading'])
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_nested_parallelism_limit(context, backend):
|
|
with context(backend, n_jobs=2):
|
|
backend_types_and_levels = _recursive_backend_info()
|
|
|
|
if cpu_count() == 1:
|
|
second_level_backend_type = 'SequentialBackend'
|
|
max_level = 1
|
|
else:
|
|
second_level_backend_type = 'ThreadingBackend'
|
|
max_level = 2
|
|
|
|
top_level_backend_type = backend.title() + 'Backend'
|
|
expected_types_and_levels = [
|
|
(top_level_backend_type, 0),
|
|
(second_level_backend_type, 1),
|
|
('SequentialBackend', max_level),
|
|
('SequentialBackend', max_level)
|
|
]
|
|
assert backend_types_and_levels == expected_types_and_levels
|
|
|
|
|
|
@with_numpy
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@skipif(distributed is None, reason='This test requires dask')
|
|
def test_nested_parallelism_with_dask(context):
|
|
with distributed.Client(n_workers=2, threads_per_worker=2):
|
|
# 10 MB of data as argument to trigger implicit scattering
|
|
data = np.ones(int(1e7), dtype=np.uint8)
|
|
for i in range(2):
|
|
with context('dask'):
|
|
backend_types_and_levels = _recursive_backend_info(data=data)
|
|
assert len(backend_types_and_levels) == 4
|
|
assert all(name == 'DaskDistributedBackend'
|
|
for name, _ in backend_types_and_levels)
|
|
|
|
# No argument
|
|
with context('dask'):
|
|
backend_types_and_levels = _recursive_backend_info()
|
|
assert len(backend_types_and_levels) == 4
|
|
assert all(name == 'DaskDistributedBackend'
|
|
for name, _ in backend_types_and_levels)
|
|
|
|
|
|
def _recursive_parallel(nesting_limit=None):
|
|
"""A horrible function that does recursive parallel calls"""
|
|
return Parallel()(delayed(_recursive_parallel)() for i in range(2))
|
|
|
|
|
|
@pytest.mark.no_cover
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@parametrize(
|
|
'backend', (['threading'] if mp is None else ['loky', 'threading'])
|
|
)
|
|
def test_thread_bomb_mitigation(context, backend):
|
|
# Test that recursive parallelism raises a recursion rather than
|
|
# saturating the operating system resources by creating a unbounded number
|
|
# of threads.
|
|
with context(backend, n_jobs=2):
|
|
with raises(BaseException) as excinfo:
|
|
_recursive_parallel()
|
|
exc = excinfo.value
|
|
if backend == "loky":
|
|
# Local import because loky may not be importable for lack of
|
|
# multiprocessing
|
|
from joblib.externals.loky.process_executor import TerminatedWorkerError # noqa
|
|
if isinstance(exc, (TerminatedWorkerError, PicklingError)):
|
|
# The recursion exception can itself cause an error when
|
|
# pickling it to be send back to the parent process. In this
|
|
# case the worker crashes but the original traceback is still
|
|
# printed on stderr. This could be improved but does not seem
|
|
# simple to do and this is not critical for users (as long
|
|
# as there is no process or thread bomb happening).
|
|
pytest.xfail("Loky worker crash when serializing RecursionError")
|
|
|
|
assert isinstance(exc, RecursionError)
|
|
|
|
|
|
def _run_parallel_sum():
|
|
env_vars = {}
|
|
for var in ['OMP_NUM_THREADS', 'OPENBLAS_NUM_THREADS', 'MKL_NUM_THREADS',
|
|
'VECLIB_MAXIMUM_THREADS', 'NUMEXPR_NUM_THREADS',
|
|
'NUMBA_NUM_THREADS', 'ENABLE_IPC']:
|
|
env_vars[var] = os.environ.get(var)
|
|
return env_vars, parallel_sum(100)
|
|
|
|
|
|
@parametrize("backend", ([None, 'loky'] if mp is not None else [None]))
|
|
@skipif(parallel_sum is None, reason="Need OpenMP helper compiled")
|
|
def test_parallel_thread_limit(backend):
|
|
results = Parallel(n_jobs=2, backend=backend)(
|
|
delayed(_run_parallel_sum)() for _ in range(2)
|
|
)
|
|
expected_num_threads = max(cpu_count() // 2, 1)
|
|
for worker_env_vars, omp_num_threads in results:
|
|
assert omp_num_threads == expected_num_threads
|
|
for name, value in worker_env_vars.items():
|
|
if name.endswith("_THREADS"):
|
|
assert value == str(expected_num_threads)
|
|
else:
|
|
assert name == "ENABLE_IPC"
|
|
assert value == "1"
|
|
|
|
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
@skipif(distributed is not None, reason='This test requires dask')
|
|
def test_dask_backend_when_dask_not_installed(context):
|
|
with raises(ValueError, match='Please install dask'):
|
|
context('dask')
|
|
|
|
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_zero_worker_backend(context):
|
|
# joblib.Parallel should reject with an explicit error message parallel
|
|
# backends that have no worker.
|
|
class ZeroWorkerBackend(ThreadingBackend):
|
|
def configure(self, *args, **kwargs):
|
|
return 0
|
|
|
|
def apply_async(self, func, callback=None): # pragma: no cover
|
|
raise TimeoutError("No worker available")
|
|
|
|
def effective_n_jobs(self, n_jobs): # pragma: no cover
|
|
return 0
|
|
|
|
expected_msg = "ZeroWorkerBackend has no active worker"
|
|
with context(ZeroWorkerBackend()):
|
|
with pytest.raises(RuntimeError, match=expected_msg):
|
|
Parallel(n_jobs=2)(delayed(id)(i) for i in range(2))
|
|
|
|
|
|
def test_globals_update_at_each_parallel_call():
|
|
# This is a non-regression test related to joblib issues #836 and #833.
|
|
# Cloudpickle versions between 0.5.4 and 0.7 introduced a bug where global
|
|
# variables changes in a parent process between two calls to
|
|
# joblib.Parallel would not be propagated into the workers.
|
|
global MY_GLOBAL_VARIABLE
|
|
MY_GLOBAL_VARIABLE = "original value"
|
|
|
|
def check_globals():
|
|
global MY_GLOBAL_VARIABLE
|
|
return MY_GLOBAL_VARIABLE
|
|
|
|
assert check_globals() == "original value"
|
|
|
|
workers_global_variable = Parallel(n_jobs=2)(
|
|
delayed(check_globals)() for i in range(2))
|
|
assert set(workers_global_variable) == {"original value"}
|
|
|
|
# Change the value of MY_GLOBAL_VARIABLE, and make sure this change gets
|
|
# propagated into the workers environment
|
|
MY_GLOBAL_VARIABLE = "changed value"
|
|
assert check_globals() == "changed value"
|
|
|
|
workers_global_variable = Parallel(n_jobs=2)(
|
|
delayed(check_globals)() for i in range(2))
|
|
assert set(workers_global_variable) == {"changed value"}
|
|
|
|
|
|
##############################################################################
|
|
# Test environment variable in child env, in particular for limiting
|
|
# the maximal number of threads in C-library threadpools.
|
|
#
|
|
|
|
def _check_numpy_threadpool_limits():
|
|
import numpy as np
|
|
# Let's call BLAS on a Matrix Matrix multiplication with dimensions large
|
|
# enough to ensure that the threadpool managed by the underlying BLAS
|
|
# implementation is actually used so as to force its initialization.
|
|
a = np.random.randn(100, 100)
|
|
np.dot(a, a)
|
|
from threadpoolctl import threadpool_info
|
|
return threadpool_info()
|
|
|
|
|
|
def _parent_max_num_threads_for(child_module, parent_info):
|
|
for parent_module in parent_info:
|
|
if parent_module['filepath'] == child_module['filepath']:
|
|
return parent_module['num_threads']
|
|
raise ValueError("An unexpected module was loaded in child:\n{}"
|
|
.format(child_module))
|
|
|
|
|
|
def check_child_num_threads(workers_info, parent_info, num_threads):
|
|
# Check that the number of threads reported in workers_info is consistent
|
|
# with the expectation. We need to be careful to handle the cases where
|
|
# the requested number of threads is below max_num_thread for the library.
|
|
for child_threadpool_info in workers_info:
|
|
for child_module in child_threadpool_info:
|
|
parent_max_num_threads = _parent_max_num_threads_for(
|
|
child_module, parent_info)
|
|
expected = {min(num_threads, parent_max_num_threads), num_threads}
|
|
assert child_module['num_threads'] in expected
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
@parametrize('n_jobs', [2, 4, -2, -1])
|
|
def test_threadpool_limitation_in_child_loky(n_jobs):
|
|
# Check that the protection against oversubscription in workers is working
|
|
# using threadpoolctl functionalities.
|
|
|
|
# Skip this test if numpy is not linked to a BLAS library
|
|
parent_info = _check_numpy_threadpool_limits()
|
|
if len(parent_info) == 0:
|
|
pytest.skip(reason="Need a version of numpy linked to BLAS")
|
|
|
|
workers_threadpool_infos = Parallel(backend="loky", n_jobs=n_jobs)(
|
|
delayed(_check_numpy_threadpool_limits)() for i in range(2))
|
|
|
|
n_jobs = effective_n_jobs(n_jobs)
|
|
expected_child_num_threads = max(cpu_count() // n_jobs, 1)
|
|
|
|
check_child_num_threads(workers_threadpool_infos, parent_info,
|
|
expected_child_num_threads)
|
|
|
|
|
|
@with_numpy
|
|
@with_multiprocessing
|
|
@parametrize('inner_max_num_threads', [1, 2, 4, None])
|
|
@parametrize('n_jobs', [2, -1])
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_threadpool_limitation_in_child_context(
|
|
context, n_jobs, inner_max_num_threads
|
|
):
|
|
# Check that the protection against oversubscription in workers is working
|
|
# using threadpoolctl functionalities.
|
|
|
|
# Skip this test if numpy is not linked to a BLAS library
|
|
parent_info = _check_numpy_threadpool_limits()
|
|
if len(parent_info) == 0:
|
|
pytest.skip(reason="Need a version of numpy linked to BLAS")
|
|
|
|
with context('loky', inner_max_num_threads=inner_max_num_threads):
|
|
workers_threadpool_infos = Parallel(n_jobs=n_jobs)(
|
|
delayed(_check_numpy_threadpool_limits)() for i in range(2))
|
|
|
|
n_jobs = effective_n_jobs(n_jobs)
|
|
if inner_max_num_threads is None:
|
|
expected_child_num_threads = max(cpu_count() // n_jobs, 1)
|
|
else:
|
|
expected_child_num_threads = inner_max_num_threads
|
|
|
|
check_child_num_threads(workers_threadpool_infos, parent_info,
|
|
expected_child_num_threads)
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('n_jobs', [2, -1])
|
|
@parametrize('var_name', ["OPENBLAS_NUM_THREADS",
|
|
"MKL_NUM_THREADS",
|
|
"OMP_NUM_THREADS"])
|
|
@parametrize("context", [parallel_config, parallel_backend])
|
|
def test_threadpool_limitation_in_child_override(context, n_jobs, var_name):
|
|
# Check that environment variables set by the user on the main process
|
|
# always have the priority.
|
|
|
|
# Clean up the existing executor because we change the environment of the
|
|
# parent at runtime and it is not detected in loky intentionally.
|
|
get_reusable_executor(reuse=True).shutdown()
|
|
|
|
def _get_env(var_name):
|
|
return os.environ.get(var_name)
|
|
|
|
original_var_value = os.environ.get(var_name)
|
|
try:
|
|
os.environ[var_name] = "4"
|
|
# Skip this test if numpy is not linked to a BLAS library
|
|
results = Parallel(n_jobs=n_jobs)(
|
|
delayed(_get_env)(var_name) for i in range(2))
|
|
assert results == ["4", "4"]
|
|
|
|
with context('loky', inner_max_num_threads=1):
|
|
results = Parallel(n_jobs=n_jobs)(
|
|
delayed(_get_env)(var_name) for i in range(2))
|
|
assert results == ["1", "1"]
|
|
|
|
finally:
|
|
if original_var_value is None:
|
|
del os.environ[var_name]
|
|
else:
|
|
os.environ[var_name] = original_var_value
|
|
|
|
|
|
@with_multiprocessing
|
|
@parametrize('n_jobs', [2, 4, -1])
|
|
def test_loky_reuse_workers(n_jobs):
|
|
# Non-regression test for issue #967 where the workers are not reused when
|
|
# calling multiple Parallel loops.
|
|
|
|
def parallel_call(n_jobs):
|
|
x = range(10)
|
|
Parallel(n_jobs=n_jobs)(delayed(sum)(x) for i in range(10))
|
|
|
|
# Run a parallel loop and get the workers used for computations
|
|
parallel_call(n_jobs)
|
|
first_executor = get_reusable_executor(reuse=True)
|
|
|
|
# Ensure that the workers are reused for the next calls, as the executor is
|
|
# not restarted.
|
|
for _ in range(10):
|
|
parallel_call(n_jobs)
|
|
executor = get_reusable_executor(reuse=True)
|
|
assert executor == first_executor
|