Aluode/PerceptionLabPortable
0
1"""2Test the parallel module.3"""4 5# Author: Gael Varoquaux <gael dot varoquaux at normalesup dot org>6# Copyright (c) 2010-2011 Gael Varoquaux7# License: BSD Style, 3 clauses.8 9import mmap10import os11import re12import sys13import threading14import time15import warnings16import weakref17from contextlib import nullcontext18from math import sqrt19from multiprocessing import TimeoutError20from pickle import PicklingError21from time import sleep22from traceback import format_exception23 24import pytest25 26import joblib27from joblib import dump, load, parallel28from joblib._multiprocessing_helpers import mp29from joblib.test.common import (30 IS_GIL_DISABLED,31 np,32 with_multiprocessing,33 with_numpy,34)35from joblib.testing import check_subprocess_call, parametrize, raises, skipif, warns36 37if mp is not None:38 # Loky is not available if multiprocessing is not39 from joblib.externals.loky import get_reusable_executor40 41from queue import Queue42 43try:44 import posix45except ImportError:46 posix = None47 48try:49 from ._openmp_test_helper.parallel_sum import parallel_sum50except ImportError:51 parallel_sum = None52 53try:54 import distributed55except ImportError:56 distributed = None57 58from joblib._parallel_backends import (59 LokyBackend,60 MultiprocessingBackend,61 ParallelBackendBase,62 SequentialBackend,63 ThreadingBackend,64)65from joblib.parallel import (66 BACKENDS,67 Parallel,68 cpu_count,69 delayed,70 effective_n_jobs,71 mp,72 parallel_backend,73 parallel_config,74 register_parallel_backend,75)76 77RETURN_GENERATOR_BACKENDS = BACKENDS.copy()78RETURN_GENERATOR_BACKENDS.pop("multiprocessing", None)79 80ALL_VALID_BACKENDS = [None] + sorted(BACKENDS.keys())81# Add instances of backend classes deriving from ParallelBackendBase82ALL_VALID_BACKENDS += [BACKENDS[backend_str]() for backend_str in BACKENDS]83if mp is None:84 PROCESS_BACKENDS = []85else:86 PROCESS_BACKENDS = ["multiprocessing", "loky"]87PARALLEL_BACKENDS = PROCESS_BACKENDS + ["threading"]88 89if hasattr(mp, "get_context"):90 # Custom multiprocessing context in Python 3.4+91 ALL_VALID_BACKENDS.append(mp.get_context("spawn"))92 93 94def get_default_backend_instance():95 # The default backend can be changed before running the tests through96 # JOBLIB_DEFAULT_PARALLEL_BACKEND environment variable so we need to use97 # parallel.DEFAULT_BACKEND here and not98 # from joblib.parallel import DEFAULT_BACKEND99 return BACKENDS[parallel.DEFAULT_BACKEND]100 101 102def get_workers(backend):103 return getattr(backend, "_pool", getattr(backend, "_workers", None))104 105 106def division(x, y):107 return x / y108 109 110def square(x):111 return x**2112 113 114class MyExceptionWithFinickyInit(Exception):115 """An exception class with non trivial __init__"""116 117 def __init__(self, a, b, c, d):118 pass119 120 121def exception_raiser(x, custom_exception=False):122 if x == 7:123 raise (124 MyExceptionWithFinickyInit("a", "b", "c", "d")125 if custom_exception126 else ValueError127 )128 return x129 130 131def interrupt_raiser(x):132 time.sleep(0.05)133 raise KeyboardInterrupt134 135 136def f(x, y=0, z=0):137 """A module-level function so that it can be spawn with138 multiprocessing.139 """140 return x**2 + y + z141 142 143def _active_backend_type():144 return type(parallel.get_active_backend()[0])145 146 147def parallel_func(inner_n_jobs, backend):148 return Parallel(n_jobs=inner_n_jobs, backend=backend)(149 delayed(square)(i) for i in range(3)150 )151 152 153###############################################################################154def test_cpu_count():155 assert cpu_count() > 0156 157 158def test_effective_n_jobs():159 assert effective_n_jobs() > 0160 161 162@parametrize("context", [parallel_config, parallel_backend])163@pytest.mark.parametrize(164 "backend_n_jobs, expected_n_jobs",165 [(3, 3), (-1, effective_n_jobs(n_jobs=-1)), (None, 1)],166 ids=["positive-int", "negative-int", "None"],167)168@with_multiprocessing169def test_effective_n_jobs_None(context, backend_n_jobs, expected_n_jobs):170 # check the number of effective jobs when `n_jobs=None`171 # non-regression test for https://github.com/joblib/joblib/issues/984172 with context("threading", n_jobs=backend_n_jobs):173 # when using a backend, the default of number jobs will be the one set174 # in the backend175 assert effective_n_jobs(n_jobs=None) == expected_n_jobs176 # without any backend, None will default to a single job177 assert effective_n_jobs(n_jobs=None) == 1178 179 180###############################################################################181# Test parallel182 183 184@parametrize("backend", ALL_VALID_BACKENDS)185@parametrize("n_jobs", [1, 2, -1, -2])186@parametrize("verbose", [2, 11, 100])187def test_simple_parallel(backend, n_jobs, verbose):188 assert [square(x) for x in range(5)] == Parallel(189 n_jobs=n_jobs, backend=backend, verbose=verbose190 )(delayed(square)(x) for x in range(5))191 192 193@parametrize("backend", ALL_VALID_BACKENDS)194@parametrize("n_jobs", [1, 2])195def test_parallel_pretty_print(backend, n_jobs):196 n_tasks = 100197 pattern = re.compile(r"(Done\s+\d+ out of \d+ \|)")198 199 class ParallelLog(Parallel):200 messages = []201 202 def _print(self, msg):203 self.messages.append(msg)204 205 executor = ParallelLog(n_jobs=n_jobs, backend=backend, verbose=10000)206 executor([delayed(f)(i) for i in range(n_tasks)])207 lens = set()208 for message in executor.messages:209 if s := pattern.search(message):210 a, b = s.span()211 lens.add(b - a)212 assert len(lens) == 1213 214 215@parametrize("backend", ALL_VALID_BACKENDS)216def test_main_thread_renamed_no_warning(backend, monkeypatch):217 # Check that no default backend relies on the name of the main thread:218 # https://github.com/joblib/joblib/issues/180#issuecomment-253266247219 # Some programs use a different name for the main thread. This is the case220 # for uWSGI apps for instance.221 monkeypatch.setattr(222 target=threading.current_thread(),223 name="name",224 value="some_new_name_for_the_main_thread",225 )226 227 with warnings.catch_warnings(record=True) as warninfo:228 results = Parallel(n_jobs=2, backend=backend)(229 delayed(square)(x) for x in range(3)230 )231 assert results == [0, 1, 4]232 233 # Due to the default parameters of LokyBackend, there is a chance that234 # warninfo catches Warnings from worker timeouts. We remove it if it exists235 # We also remove DeprecationWarnings which could lead to false negatives.236 warninfo = [237 w238 for w in warninfo239 if "worker timeout" not in str(w.message)240 and not isinstance(w.message, DeprecationWarning)241 ]242 243 # Under Python 3.13 if backend='multiprocessing', you will get a244 # warning saying that forking a multi-threaded process is not a good idea,245 # we ignore them in this test246 if backend in [None, "multiprocessing"] or isinstance(247 backend, MultiprocessingBackend248 ):249 message_part = "multi-threaded, use of fork() may lead to deadlocks"250 warninfo = [w for w in warninfo if message_part not in str(w.message)]251 252 # The multiprocessing backend will raise a warning when detecting that is253 # started from the non-main thread. Let's check that there is no false254 # positive because of the name change.255 assert len(warninfo) == 0256 257 258def _assert_warning_nested(backend, inner_n_jobs, expected):259 with warnings.catch_warnings(record=True) as warninfo:260 warnings.simplefilter("always")261 parallel_func(backend=backend, inner_n_jobs=inner_n_jobs)262 263 warninfo = [w.message for w in warninfo]264 if expected:265 if warninfo:266 warnings_are_correct = all(267 "backed parallel loops cannot" in each.args[0] for each in warninfo268 )269 # With free-threaded Python, when the outer backend is threading,270 # we might see more that one warning271 warnings_have_the_right_length = (272 len(warninfo) >= 1 if IS_GIL_DISABLED else len(warninfo) == 1273 )274 return warnings_are_correct and warnings_have_the_right_length275 276 return False277 else:278 assert not warninfo279 return True280 281 282@with_multiprocessing283@parametrize(284 "parent_backend,child_backend,expected",285 [286 ("loky", "multiprocessing", True),287 ("loky", "loky", False),288 ("multiprocessing", "multiprocessing", True),289 ("multiprocessing", "loky", True),290 ("threading", "multiprocessing", True),291 ("threading", "loky", True),292 ],293)294def test_nested_parallel_warnings(parent_backend, child_backend, expected):295 # no warnings if inner_n_jobs=1296 Parallel(n_jobs=2, backend=parent_backend)(297 delayed(_assert_warning_nested)(298 backend=child_backend, inner_n_jobs=1, expected=False299 )300 for _ in range(5)301 )302 303 # warnings if inner_n_jobs != 1 and expected304 res = Parallel(n_jobs=2, backend=parent_backend)(305 delayed(_assert_warning_nested)(306 backend=child_backend, inner_n_jobs=2, expected=expected307 )308 for _ in range(5)309 )310 311 # warning handling is not thread safe. One thread might see multiple312 # warning or no warning at all.313 if parent_backend == "threading":314 assert any(res)315 else:316 assert all(res)317 318 319@with_multiprocessing320@parametrize("backend", ["loky", "multiprocessing", "threading"])321def test_background_thread_parallelism(backend):322 is_run_parallel = [False]323 324 def background_thread(is_run_parallel):325 with warnings.catch_warnings(record=True) as warninfo:326 Parallel(n_jobs=2)(delayed(sleep)(0.1) for _ in range(4))327 print(len(warninfo))328 is_run_parallel[0] = len(warninfo) == 0329 330 t = threading.Thread(target=background_thread, args=(is_run_parallel,))331 t.start()332 t.join()333 assert is_run_parallel[0]334 335 336def nested_loop(backend):337 Parallel(n_jobs=2, backend=backend)(delayed(square)(0.01) for _ in range(2))338 339 340@parametrize("child_backend", BACKENDS)341@parametrize("parent_backend", BACKENDS)342def test_nested_loop(parent_backend, child_backend):343 Parallel(n_jobs=2, backend=parent_backend)(344 delayed(nested_loop)(child_backend) for _ in range(2)345 )346 347 348def raise_exception(backend):349 raise ValueError350 351 352@with_multiprocessing353def test_nested_loop_with_exception_with_loky():354 with raises(ValueError):355 with Parallel(n_jobs=2, backend="loky") as parallel:356 parallel([delayed(nested_loop)("loky"), delayed(raise_exception)("loky")])357 358 359def test_mutate_input_with_threads():360 """Input is mutable when using the threading backend"""361 q = Queue(maxsize=5)362 Parallel(n_jobs=2, backend="threading")(delayed(q.put)(1) for _ in range(5))363 assert q.full()364 365 366@parametrize("n_jobs", [1, 2, 3])367def test_parallel_kwargs(n_jobs):368 """Check the keyword argument processing of pmap."""369 lst = range(10)370 assert [f(x, y=1) for x in lst] == Parallel(n_jobs=n_jobs)(371 delayed(f)(x, y=1) for x in lst372 )373 374 375@parametrize("backend", PARALLEL_BACKENDS)376def test_parallel_as_context_manager(backend):377 lst = range(10)378 expected = [f(x, y=1) for x in lst]379 380 with Parallel(n_jobs=4, backend=backend) as p:381 # Internally a pool instance has been eagerly created and is managed382 # via the context manager protocol383 managed_backend = p._backend384 385 # We make call with the managed parallel object several times inside386 # the managed block:387 assert expected == p(delayed(f)(x, y=1) for x in lst)388 assert expected == p(delayed(f)(x, y=1) for x in lst)389 390 # Those calls have all used the same pool instance:391 if mp is not None:392 assert get_workers(managed_backend) is get_workers(p._backend)393 394 # As soon as we exit the context manager block, the pool is terminated and395 # no longer referenced from the parallel object:396 if mp is not None:397 assert get_workers(p._backend) is None398 399 # It's still possible to use the parallel instance in non-managed mode:400 assert expected == p(delayed(f)(x, y=1) for x in lst)401 if mp is not None:402 assert get_workers(p._backend) is None403 404 405@with_multiprocessing406def test_parallel_pickling():407 """Check that pmap captures the errors when it is passed an object408 that cannot be pickled.409 """410 411 class UnpicklableObject(object):412 def __reduce__(self):413 raise RuntimeError("123")414 415 with raises(PicklingError, match=r"the task to send"):416 Parallel(n_jobs=2, backend="loky")(417 delayed(id)(UnpicklableObject()) for _ in range(10)418 )419 420 421@with_numpy422@with_multiprocessing423@parametrize("byteorder", ["<", ">", "="])424@parametrize("max_nbytes", [1, "1M"])425def test_parallel_byteorder_corruption(byteorder, max_nbytes):426 def inspect_byteorder(x):427 return x, x.dtype.byteorder428 429 x = np.arange(6).reshape((2, 3)).view(f"{byteorder}i4")430 431 initial_np_byteorder = x.dtype.byteorder432 433 result = Parallel(n_jobs=2, backend="loky", max_nbytes=max_nbytes)(434 delayed(inspect_byteorder)(x) for _ in range(3)435 )436 437 for x_returned, byteorder_in_worker in result:438 assert byteorder_in_worker == initial_np_byteorder439 assert byteorder_in_worker == x_returned.dtype.byteorder440 np.testing.assert_array_equal(x, x_returned)441 442 443@parametrize("backend", PARALLEL_BACKENDS)444def test_parallel_timeout_success(backend):445 # Check that timeout isn't thrown when function is fast enough446 assert (447 len(448 Parallel(n_jobs=2, backend=backend, timeout=30)(449 delayed(sleep)(0.001) for x in range(10)450 )451 )452 == 10453 )454 455 456@with_multiprocessing457@parametrize("backend", PARALLEL_BACKENDS)458def test_parallel_timeout_fail(backend):459 # Check that timeout properly fails when function is too slow460 with raises(TimeoutError):461 Parallel(n_jobs=2, backend=backend, timeout=0.01)(462 delayed(sleep)(10) for x in range(10)463 )464 465 466@with_multiprocessing467@parametrize("backend", set(RETURN_GENERATOR_BACKENDS) - {"sequential"})468@parametrize("return_as", ["generator", "generator_unordered"])469def test_parallel_timeout_fail_with_generator(backend, return_as):470 # Check that timeout properly fails when function is too slow with471 # return_as=generator472 with raises(TimeoutError):473 list(474 Parallel(n_jobs=2, backend=backend, return_as=return_as, timeout=0.1)(475 delayed(sleep)(10) for x in range(10)476 )477 )478 479 # Fast tasks and high timeout should not raise480 list(481 Parallel(n_jobs=2, backend=backend, return_as=return_as, timeout=10)(482 delayed(sleep)(0.01) for x in range(10)483 )484 )485 486 487@with_multiprocessing488@parametrize("backend", PROCESS_BACKENDS)489def test_error_capture(backend):490 # Check that error are captured, and that correct exceptions491 # are raised.492 if mp is not None:493 with raises(ZeroDivisionError):494 Parallel(n_jobs=2, backend=backend)(495 [delayed(division)(x, y) for x, y in zip((0, 1), (1, 0))]496 )497 498 with raises(KeyboardInterrupt):499 Parallel(n_jobs=2, backend=backend)(500 [delayed(interrupt_raiser)(x) for x in (1, 0)]501 )502 503 # Try again with the context manager API504 with Parallel(n_jobs=2, backend=backend) as parallel:505 assert get_workers(parallel._backend) is not None506 original_workers = get_workers(parallel._backend)507 508 with raises(ZeroDivisionError):509 parallel([delayed(division)(x, y) for x, y in zip((0, 1), (1, 0))])510 511 # The managed pool should still be available and be in a working512 # state despite the previously raised (and caught) exception513 assert get_workers(parallel._backend) is not None514 515 # The pool should have been interrupted and restarted:516 assert get_workers(parallel._backend) is not original_workers517 518 assert [f(x, y=1) for x in range(10)] == parallel(519 delayed(f)(x, y=1) for x in range(10)520 )521 522 original_workers = get_workers(parallel._backend)523 with raises(KeyboardInterrupt):524 parallel([delayed(interrupt_raiser)(x) for x in (1, 0)])525 526 # The pool should still be available despite the exception527 assert get_workers(parallel._backend) is not None528 529 # The pool should have been interrupted and restarted:530 assert get_workers(parallel._backend) is not original_workers531 532 assert [f(x, y=1) for x in range(10)] == parallel(533 delayed(f)(x, y=1) for x in range(10)534 ), (535 parallel._iterating,536 parallel.n_completed_tasks,537 parallel.n_dispatched_tasks,538 parallel._aborting,539 )540 541 # Check that the inner pool has been terminated when exiting the542 # context manager543 assert get_workers(parallel._backend) is None544 else:545 with raises(KeyboardInterrupt):546 Parallel(n_jobs=2)([delayed(interrupt_raiser)(x) for x in (1, 0)])547 548 # wrapped exceptions should inherit from the class of the original549 # exception to make it easy to catch them550 with raises(ZeroDivisionError):551 Parallel(n_jobs=2)([delayed(division)(x, y) for x, y in zip((0, 1), (1, 0))])552 553 with raises(MyExceptionWithFinickyInit):554 Parallel(n_jobs=2, verbose=0)(555 (delayed(exception_raiser)(i, custom_exception=True) for i in range(30))556 )557 558 559@with_multiprocessing560@parametrize("backend", BACKENDS)561def test_error_in_task_iterator(backend):562 def my_generator(raise_at=0):563 for i in range(20):564 if i == raise_at:565 raise ValueError("Iterator Raising Error")566 yield i567 568 with Parallel(n_jobs=2, backend=backend) as p:569 # The error is raised in the pre-dispatch phase570 with raises(ValueError, match="Iterator Raising Error"):571 p(delayed(square)(i) for i in my_generator(raise_at=0))572 573 # The error is raised when dispatching a new task after the574 # pre-dispatch (likely to happen in a different thread)575 with raises(ValueError, match="Iterator Raising Error"):576 p(delayed(square)(i) for i in my_generator(raise_at=5))577 578 # Same, but raises long after the pre-dispatch phase579 with raises(ValueError, match="Iterator Raising Error"):580 p(delayed(square)(i) for i in my_generator(raise_at=19))581 582 583def consumer(queue, item):584 queue.append("Consumed %s" % item)585 586 587@parametrize("backend", BACKENDS)588@parametrize(589 "batch_size, expected_queue",590 [591 (592 1,593 [594 "Produced 0",595 "Consumed 0",596 "Produced 1",597 "Consumed 1",598 "Produced 2",599 "Consumed 2",600 "Produced 3",601 "Consumed 3",602 "Produced 4",603 "Consumed 4",604 "Produced 5",605 "Consumed 5",606 ],607 ),608 (609 4,610 [ # First Batch611 "Produced 0",612 "Produced 1",613 "Produced 2",614 "Produced 3",615 "Consumed 0",616 "Consumed 1",617 "Consumed 2",618 "Consumed 3",619 # Second batch620 "Produced 4",621 "Produced 5",622 "Consumed 4",623 "Consumed 5",624 ],625 ),626 ],627)628def test_dispatch_one_job(backend, batch_size, expected_queue):629 """Test that with only one job, Parallel does act as a iterator."""630 queue = list()631 632 def producer():633 for i in range(6):634 queue.append("Produced %i" % i)635 yield i636 637 Parallel(n_jobs=1, batch_size=batch_size, backend=backend)(638 delayed(consumer)(queue, x) for x in producer()639 )640 assert queue == expected_queue641 assert len(queue) == 12642 643 644@with_multiprocessing645@parametrize("backend", PARALLEL_BACKENDS)646def test_dispatch_multiprocessing(backend):647 """Check that using pre_dispatch Parallel does indeed dispatch items648 lazily.649 """650 manager = mp.Manager()651 queue = manager.list()652 653 def producer():654 for i in range(6):655 queue.append("Produced %i" % i)656 yield i657 658 Parallel(n_jobs=2, batch_size=1, pre_dispatch=3, backend=backend)(659 delayed(consumer)(queue, "any") for _ in producer()660 )661 662 queue_contents = list(queue)663 assert queue_contents[0] == "Produced 0"664 665 # Only 3 tasks are pre-dispatched out of 6. The 4th task is dispatched only666 # after any of the first 3 jobs have completed.667 first_consumption_index = queue_contents[:4].index("Consumed any")668 assert first_consumption_index > -1669 670 produced_3_index = queue_contents.index("Produced 3") # 4th task produced671 assert produced_3_index > first_consumption_index672 673 assert len(queue) == 12674 675 676def test_batching_auto_threading():677 # batching='auto' with the threading backend leaves the effective batch678 # size to 1 (no batching) as it has been found to never be beneficial with679 # this low-overhead backend.680 681 with Parallel(n_jobs=2, batch_size="auto", backend="threading") as p:682 p(delayed(id)(i) for i in range(5000)) # many very fast tasks683 assert p._backend.compute_batch_size() == 1684 685 686@with_multiprocessing687@parametrize("backend", PROCESS_BACKENDS)688def test_batching_auto_subprocesses(backend):689 with Parallel(n_jobs=2, batch_size="auto", backend=backend) as p:690 p(delayed(id)(i) for i in range(5000)) # many very fast tasks691 692 # It should be strictly larger than 1 but as we don't want heisen693 # failures on clogged CI worker environment be safe and only check that694 # it's a strictly positive number.695 assert p._backend.compute_batch_size() > 0696 697 698def test_exception_dispatch():699 """Make sure that exception raised during dispatch are indeed captured"""700 with raises(ValueError):701 Parallel(n_jobs=2, pre_dispatch=16, verbose=0)(702 delayed(exception_raiser)(i) for i in range(30)703 )704 705 706def nested_function_inner(i):707 Parallel(n_jobs=2)(delayed(exception_raiser)(j) for j in range(30))708 709 710def nested_function_outer(i):711 Parallel(n_jobs=2)(delayed(nested_function_inner)(j) for j in range(30))712 713 714@with_multiprocessing715@parametrize("backend", PARALLEL_BACKENDS)716@pytest.mark.xfail(reason="https://github.com/joblib/loky/pull/255")717def test_nested_exception_dispatch(backend):718 """Ensure errors for nested joblib cases gets propagated719 720 We rely on the Python 3 built-in __cause__ system that already721 report this kind of information to the user.722 """723 with raises(ValueError) as excinfo:724 Parallel(n_jobs=2, backend=backend)(725 delayed(nested_function_outer)(i) for i in range(30)726 )727 728 # Check that important information such as function names are visible729 # in the final error message reported to the user730 report_lines = format_exception(excinfo.type, excinfo.value, excinfo.tb)731 report = "".join(report_lines)732 assert "nested_function_outer" in report733 assert "nested_function_inner" in report734 assert "exception_raiser" in report735 736 assert type(excinfo.value) is ValueError737 738 739class FakeParallelBackend(SequentialBackend):740 """Pretends to run concurrently while running sequentially."""741 742 def configure(self, n_jobs=1, parallel=None, **backend_args):743 self.n_jobs = self.effective_n_jobs(n_jobs)744 self.parallel = parallel745 return n_jobs746 747 def effective_n_jobs(self, n_jobs=1):748 if n_jobs < 0:749 n_jobs = max(mp.cpu_count() + 1 + n_jobs, 1)750 return n_jobs751 752 753def test_invalid_backend():754 with raises(ValueError, match="Invalid backend:"):755 Parallel(backend="unit-testing")756 757 with raises(ValueError, match="Invalid backend:"):758 with parallel_config(backend="unit-testing"):759 pass760 761 with raises(ValueError, match="Invalid backend:"):762 with parallel_config(backend="unit-testing"):763 pass764 765 766@parametrize("backend", ALL_VALID_BACKENDS)767def test_invalid_njobs(backend):768 with raises(ValueError) as excinfo:769 Parallel(n_jobs=0, backend=backend)._initialize_backend()770 assert "n_jobs == 0 in Parallel has no meaning" in str(excinfo.value)771 772 with raises(ValueError) as excinfo:773 Parallel(n_jobs=0.5, backend=backend)._initialize_backend()774 assert "n_jobs == 0 in Parallel has no meaning" in str(excinfo.value)775 776 with raises(ValueError) as excinfo:777 Parallel(n_jobs="2.3", backend=backend)._initialize_backend()778 assert "n_jobs could not be converted to int" in str(excinfo.value)779 780 with raises(ValueError) as excinfo:781 Parallel(n_jobs="invalid_str", backend=backend)._initialize_backend()782 assert "n_jobs could not be converted to int" in str(excinfo.value)783 784 785@with_multiprocessing786@parametrize("backend", PARALLEL_BACKENDS)787@parametrize("n_jobs", ["2", 2.3, 2])788def test_njobs_converted_to_int(backend, n_jobs):789 p = Parallel(n_jobs=n_jobs, backend=backend)790 assert p._effective_n_jobs() == 2791 792 res = p(delayed(square)(i) for i in range(10))793 assert all(r == square(i) for i, r in enumerate(res))794 795 796def test_register_parallel_backend():797 try:798 register_parallel_backend("test_backend", FakeParallelBackend)799 assert "test_backend" in BACKENDS800 assert BACKENDS["test_backend"] == FakeParallelBackend801 finally:802 del BACKENDS["test_backend"]803 804 805def test_overwrite_default_backend():806 default_backend_orig = parallel.DEFAULT_BACKEND807 assert _active_backend_type() == get_default_backend_instance()808 try:809 register_parallel_backend("threading", BACKENDS["threading"], make_default=True)810 assert _active_backend_type() == ThreadingBackend811 finally:812 # Restore the global default manually813 parallel.DEFAULT_BACKEND = default_backend_orig814 assert _active_backend_type() == get_default_backend_instance()815 816 817@skipif(mp is not None, reason="Only without multiprocessing")818def test_backend_no_multiprocessing():819 with warns(UserWarning, match="joblib backend '.*' is not available on.*"):820 Parallel(backend="loky")(delayed(square)(i) for i in range(3))821 822 # The below should now work without problems823 with parallel_config(backend="loky"):824 Parallel()(delayed(square)(i) for i in range(3))825 826 827def check_backend_context_manager(context, backend_name):828 with context(backend_name, n_jobs=3):829 active_backend, active_n_jobs = parallel.get_active_backend()830 assert active_n_jobs == 3831 assert effective_n_jobs(3) == 3832 p = Parallel()833 assert p.n_jobs == 3834 if backend_name == "multiprocessing":835 assert type(active_backend) is MultiprocessingBackend836 assert type(p._backend) is MultiprocessingBackend837 elif backend_name == "loky":838 assert type(active_backend) is LokyBackend839 assert type(p._backend) is LokyBackend840 elif backend_name == "threading":841 assert type(active_backend) is ThreadingBackend842 assert type(p._backend) is ThreadingBackend843 elif backend_name.startswith("test_"):844 assert type(active_backend) is FakeParallelBackend845 assert type(p._backend) is FakeParallelBackend846 847 848all_backends_for_context_manager = PARALLEL_BACKENDS[:]849all_backends_for_context_manager.extend(["test_backend_%d" % i for i in range(3)])850 851 852@with_multiprocessing853@parametrize("backend", all_backends_for_context_manager)854@parametrize("context", [parallel_backend, parallel_config])855def test_backend_context_manager(monkeypatch, backend, context):856 if backend not in BACKENDS:857 monkeypatch.setitem(BACKENDS, backend, FakeParallelBackend)858 859 assert _active_backend_type() == get_default_backend_instance()860 # check that this possible to switch parallel backends sequentially861 check_backend_context_manager(context, backend)862 863 # The default backend is restored864 assert _active_backend_type() == get_default_backend_instance()865 866 # Check that context manager switching is thread safe:867 Parallel(n_jobs=2, backend="threading")(868 delayed(check_backend_context_manager)(context, b)869 for b in all_backends_for_context_manager870 if not b871 )872 873 # The default backend is again restored874 assert _active_backend_type() == get_default_backend_instance()875 876 877class ParameterizedParallelBackend(SequentialBackend):878 """Pretends to run conncurrently while running sequentially."""879 880 def __init__(self, param=None):881 if param is None:882 raise ValueError("param should not be None")883 self.param = param884 885 886@parametrize("context", [parallel_config, parallel_backend])887def test_parameterized_backend_context_manager(monkeypatch, context):888 monkeypatch.setitem(BACKENDS, "param_backend", ParameterizedParallelBackend)889 assert _active_backend_type() == get_default_backend_instance()890 891 with context("param_backend", param=42, n_jobs=3):892 active_backend, active_n_jobs = parallel.get_active_backend()893 assert type(active_backend) is ParameterizedParallelBackend894 assert active_backend.param == 42895 assert active_n_jobs == 3896 p = Parallel()897 assert p.n_jobs == 3898 assert p._backend is active_backend899 results = p(delayed(sqrt)(i) for i in range(5))900 assert results == [sqrt(i) for i in range(5)]901 902 # The default backend is again restored903 assert _active_backend_type() == get_default_backend_instance()904 905 906@parametrize("context", [parallel_config, parallel_backend])907def test_directly_parameterized_backend_context_manager(context):908 assert _active_backend_type() == get_default_backend_instance()909 910 # Check that it's possible to pass a backend instance directly,911 # without registration912 with context(ParameterizedParallelBackend(param=43), n_jobs=5):913 active_backend, active_n_jobs = parallel.get_active_backend()914 assert type(active_backend) is ParameterizedParallelBackend915 assert active_backend.param == 43916 assert active_n_jobs == 5917 p = Parallel()918 assert p.n_jobs == 5919 assert p._backend is active_backend920 results = p(delayed(sqrt)(i) for i in range(5))921 assert results == [sqrt(i) for i in range(5)]922 923 # The default backend is again restored924 assert _active_backend_type() == get_default_backend_instance()925 926 927def sleep_and_return_pid():928 sleep(0.1)929 return os.getpid()930 931 932def get_nested_pids():933 assert _active_backend_type() == ThreadingBackend934 # Assert that the nested backend does not change the default number of935 # jobs used in Parallel936 assert Parallel()._effective_n_jobs() == 1937 938 # Assert that the tasks are running only on one process939 return Parallel(n_jobs=2)(delayed(sleep_and_return_pid)() for _ in range(2))940 941 942class MyBackend(joblib._parallel_backends.LokyBackend):943 """Backend to test backward compatibility with older backends"""944 945 def get_nested_backend(946 self,947 ):948 # Older backends only return a backend, without n_jobs indications.949 return super(MyBackend, self).get_nested_backend()[0]950 951 952register_parallel_backend("back_compat_backend", MyBackend)953 954 955@with_multiprocessing956@parametrize("backend", ["threading", "loky", "multiprocessing", "back_compat_backend"])957@parametrize("context", [parallel_config, parallel_backend])958def test_nested_backend_context_manager(context, backend):959 # Check that by default, nested parallel calls will always use the960 # ThreadingBackend961 962 with context(backend):963 pid_groups = Parallel(n_jobs=2)(delayed(get_nested_pids)() for _ in range(10))964 for pid_group in pid_groups:965 assert len(set(pid_group)) == 1966 967 968@with_multiprocessing969@parametrize("n_jobs", [2, -1, None])970@parametrize("backend", PARALLEL_BACKENDS)971@parametrize("context", [parallel_config, parallel_backend])972def test_nested_backend_in_sequential(backend, n_jobs, context):973 # Check that by default, nested parallel calls will always use the974 # ThreadingBackend975 976 def check_nested_backend(expected_backend_type, expected_n_job):977 # Assert that the sequential backend at top level, does not change the978 # backend for nested calls.979 assert _active_backend_type() == BACKENDS[expected_backend_type]980 981 # Assert that the nested backend in SequentialBackend does not change982 # the default number of jobs used in Parallel983 expected_n_job = effective_n_jobs(expected_n_job)984 assert Parallel()._effective_n_jobs() == expected_n_job985 986 Parallel(n_jobs=1)(987 delayed(check_nested_backend)(parallel.DEFAULT_BACKEND, 1) for _ in range(10)988 )989 990 with context(backend, n_jobs=n_jobs):991 Parallel(n_jobs=1)(992 delayed(check_nested_backend)(backend, n_jobs) for _ in range(10)993 )994 995 996def check_nesting_level(context, inner_backend, expected_level):997 with context(inner_backend) as ctx:998 if context is parallel_config:999 backend = ctx["backend"]1000 if context is parallel_backend:1001 backend = ctx[0]1002 assert backend.nesting_level == expected_level1003 1004 1005@with_multiprocessing1006@parametrize("outer_backend", PARALLEL_BACKENDS)1007@parametrize("inner_backend", PARALLEL_BACKENDS)1008@parametrize("context", [parallel_config, parallel_backend])1009def test_backend_nesting_level(context, outer_backend, inner_backend):1010 # Check that the nesting level for the backend is correctly set1011 check_nesting_level(context, outer_backend, 0)1012 1013 Parallel(n_jobs=2, backend=outer_backend)(1014 delayed(check_nesting_level)(context, inner_backend, 1) for _ in range(10)1015 )1016 1017 with context(inner_backend, n_jobs=2):1018 Parallel()(1019 delayed(check_nesting_level)(context, inner_backend, 1) for _ in range(10)1020 )1021 1022 1023@with_multiprocessing1024@parametrize("context", [parallel_config, parallel_backend])1025@parametrize("with_retrieve_callback", [True, False])1026def test_retrieval_context(context, with_retrieve_callback):1027 import contextlib1028 1029 class MyBackend(ThreadingBackend):1030 i = 01031 supports_retrieve_callback = with_retrieve_callback1032 1033 @contextlib.contextmanager1034 def retrieval_context(self):1035 self.i += 11036 yield1037 1038 register_parallel_backend("retrieval", MyBackend)1039 1040 def nested_call(n):1041 return Parallel(n_jobs=2)(delayed(id)(i) for i in range(n))1042 1043 with context("retrieval") as ctx:1044 Parallel(n_jobs=2)(delayed(nested_call)(i) for i in range(5))1045 if context is parallel_config:1046 assert ctx["backend"].i == 11047 if context is parallel_backend:1048 assert ctx[0].i == 11049 1050 1051###############################################################################1052# Test helpers1053 1054 1055@parametrize("batch_size", [0, -1, 1.42])1056def test_invalid_batch_size(batch_size):1057 with raises(ValueError):1058 Parallel(batch_size=batch_size)1059 1060 1061@parametrize(1062 "n_tasks, n_jobs, pre_dispatch, batch_size",1063 [1064 (2, 2, "all", "auto"),1065 (2, 2, "n_jobs", "auto"),1066 (10, 2, "n_jobs", "auto"),1067 (517, 2, "n_jobs", "auto"),1068 (10, 2, "n_jobs", "auto"),1069 (10, 4, "n_jobs", "auto"),1070 (200, 12, "n_jobs", "auto"),1071 (25, 12, "2 * n_jobs", 1),1072 (250, 12, "all", 1),1073 (250, 12, "2 * n_jobs", 7),1074 (200, 12, "2 * n_jobs", "auto"),1075 ],1076)1077def test_dispatch_race_condition(n_tasks, n_jobs, pre_dispatch, batch_size):1078 # Check that using (async-)dispatch does not yield a race condition on the1079 # iterable generator that is not thread-safe natively.1080 # This is a non-regression test for the "Pool seems closed" class of error1081 params = {"n_jobs": n_jobs, "pre_dispatch": pre_dispatch, "batch_size": batch_size}1082 expected = [square(i) for i in range(n_tasks)]1083 results = Parallel(**params)(delayed(square)(i) for i in range(n_tasks))1084 assert results == expected1085 1086 1087@with_multiprocessing1088def test_default_mp_context():1089 mp_start_method = mp.get_start_method()1090 p = Parallel(n_jobs=2, backend="multiprocessing")1091 context = p._backend_kwargs.get("context")1092 start_method = context.get_start_method()1093 assert start_method == mp_start_method1094 1095 1096@with_numpy1097@with_multiprocessing1098@parametrize("backend", PROCESS_BACKENDS)1099def test_no_blas_crash_or_freeze_with_subprocesses(backend):1100 if backend == "multiprocessing":1101 # Use the spawn backend that is both robust and available on all1102 # platforms1103 backend = mp.get_context("spawn")1104 1105 # Check that on recent Python version, the 'spawn' start method can make1106 # it possible to use multiprocessing in conjunction of any BLAS1107 # implementation that happens to be used by numpy with causing a freeze or1108 # a crash1109 rng = np.random.RandomState(42)1110 1111 # call BLAS DGEMM to force the initialization of the internal thread-pool1112 # in the main process1113 a = rng.randn(1000, 1000)1114 np.dot(a, a.T)1115 1116 # check that the internal BLAS thread-pool is not in an inconsistent state1117 # in the worker processes managed by multiprocessing1118 Parallel(n_jobs=2, backend=backend)(delayed(np.dot)(a, a.T) for i in range(2))1119 1120 1121UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_NO_MAIN = """\1122from joblib import Parallel, delayed1123 1124def square(x):1125 return x ** 21126 1127backend = "{}"1128if backend == "spawn":1129 from multiprocessing import get_context1130 backend = get_context(backend)1131 1132print(Parallel(n_jobs=2, backend=backend)(1133 delayed(square)(i) for i in range(5)))1134"""1135 1136 1137@with_multiprocessing1138@parametrize("backend", PROCESS_BACKENDS)1139def test_parallel_with_interactively_defined_functions(backend):1140 # When using the "-c" flag, interactive functions defined in __main__1141 # should work with any backend.1142 if backend == "multiprocessing" and mp.get_start_method() != "fork":1143 pytest.skip(1144 "Require fork start method to use interactively defined "1145 "functions with multiprocessing."1146 )1147 code = UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_NO_MAIN.format(backend)1148 check_subprocess_call(1149 [sys.executable, "-c", code], timeout=10, stdout_regex=r"\[0, 1, 4, 9, 16\]"1150 )1151 1152 1153UNPICKLABLE_CALLABLE_SCRIPT_TEMPLATE_MAIN = """\1154import sys1155# Make sure that joblib is importable in the subprocess launching this1156# script. This is needed in case we run the tests from the joblib root1157# folder without having installed joblib1158sys.path.insert(0, {joblib_root_folder!r})1159 1160from joblib import Parallel, delayed1161 1162def run(f, x):1163 return f(x)1164 1165{define_func}1166 1167if __name__ == "__main__":1168 backend = "{backend}"1169 if backend == "spawn":1170 from multiprocessing import get_context1171 backend = get_context(backend)1172 1173 callable_position = "{callable_position}"1174 if callable_position == "delayed":1175 print(Parallel(n_jobs=2, backend=backend)(1176 delayed(square)(i) for i in range(5)))1177 elif callable_position == "args":1178 print(Parallel(n_jobs=2, backend=backend)(1179 delayed(run)(square, i) for i in range(5)))1180 else:1181 print(Parallel(n_jobs=2, backend=backend)(1182 delayed(run)(f=square, x=i) for i in range(5)))1183"""1184 1185SQUARE_MAIN = """\1186def square(x):1187 return x ** 21188"""1189SQUARE_LOCAL = """\1190def gen_square():1191 def square(x):1192 return x ** 21193 return square1194square = gen_square()1195"""1196SQUARE_LAMBDA = """\1197square = lambda x: x ** 21198"""1199 1200