Aluode/PerceptionLabPortable
0
1import faulthandler2import gc3import itertools4import mmap5import os6import pickle7import platform8import subprocess9import sys10import threading11from time import sleep12 13import pytest14 15import joblib._memmapping_reducer as jmr16from joblib._memmapping_reducer import (17 ArrayMemmapForwardReducer,18 _get_backing_memmap,19 _get_temp_dir,20 _strided_from_memmap,21 _WeakArrayKeyMap,22 has_shareable_memory,23)24from joblib.backports import make_memmap25from joblib.executor import _TestingMemmappingExecutor as TestExecutor26from joblib.parallel import Parallel, delayed27from joblib.pool import MemmappingPool28from joblib.test.common import (29 IS_GIL_DISABLED,30 np,31 with_dev_shm,32 with_multiprocessing,33 with_numpy,34)35from joblib.testing import parametrize, raises, skipif36 37 38def setup_module():39 faulthandler.dump_traceback_later(timeout=300, exit=True)40 41 42def teardown_module():43 faulthandler.cancel_dump_traceback_later()44 45 46def check_memmap_and_send_back(array):47 assert _get_backing_memmap(array) is not None48 return array49 50 51def check_array(args):52 """Dummy helper function to be executed in subprocesses53 54 Check that the provided array has the expected values in the provided55 range.56 57 """58 data, position, expected = args59 np.testing.assert_array_equal(data[position], expected)60 61 62def inplace_double(args):63 """Dummy helper function to be executed in subprocesses64 65 66 Check that the input array has the right values in the provided range67 and perform an inplace modification to double the values in the range by68 two.69 70 """71 data, position, expected = args72 assert data[position] == expected73 data[position] *= 274 np.testing.assert_array_equal(data[position], 2 * expected)75 76 77@with_numpy78@with_multiprocessing79def test_memmap_based_array_reducing(tmpdir):80 """Check that it is possible to reduce a memmap backed array"""81 assert_array_equal = np.testing.assert_array_equal82 filename = tmpdir.join("test.mmap").strpath83 84 # Create a file larger than what will be used by a85 buffer = np.memmap(filename, dtype=np.float64, shape=500, mode="w+")86 87 # Fill the original buffer with negative markers to detect over of88 # underflow in case of test failures89 buffer[:] = -1.0 * np.arange(buffer.shape[0], dtype=buffer.dtype)90 buffer.flush()91 92 # Memmap a 2D fortran array on a offsetted subsection of the previous93 # buffer94 a = np.memmap(95 filename, dtype=np.float64, shape=(3, 5, 4), mode="r+", order="F", offset=496 )97 a[:] = np.arange(60).reshape(a.shape)98 99 # Build various views that share the buffer with the original memmap100 101 # b is an memmap sliced view on an memmap instance102 b = a[1:-1, 2:-1, 2:4]103 104 # b2 is a memmap 2d with memmap 1d as base105 # non-regression test for https://github.com/joblib/joblib/issues/1703106 b2 = buffer.reshape(10, 50)107 108 # c and d are array views109 c = np.asarray(b)110 d = c.T111 112 # Array reducer with auto dumping disabled113 reducer = ArrayMemmapForwardReducer(None, tmpdir.strpath, "c", True)114 115 def reconstruct_array_or_memmap(x):116 cons, args = reducer(x)117 return cons(*args)118 119 # Reconstruct original memmap120 a_reconstructed = reconstruct_array_or_memmap(a)121 assert has_shareable_memory(a_reconstructed)122 assert isinstance(a_reconstructed, np.memmap)123 assert_array_equal(a_reconstructed, a)124 125 # Reconstruct strided memmap view126 b_reconstructed = reconstruct_array_or_memmap(b)127 assert has_shareable_memory(b_reconstructed)128 assert_array_equal(b_reconstructed, b)129 130 # Reconstruct memmap 2d with memmap 1d as base131 b2_reconstructed = reconstruct_array_or_memmap(b2)132 assert has_shareable_memory(b2_reconstructed)133 assert_array_equal(b2_reconstructed, b2)134 135 # Reconstruct arrays views on memmap base136 c_reconstructed = reconstruct_array_or_memmap(c)137 assert not isinstance(c_reconstructed, np.memmap)138 assert has_shareable_memory(c_reconstructed)139 assert_array_equal(c_reconstructed, c)140 141 d_reconstructed = reconstruct_array_or_memmap(d)142 assert not isinstance(d_reconstructed, np.memmap)143 assert has_shareable_memory(d_reconstructed)144 assert_array_equal(d_reconstructed, d)145 146 # Test graceful degradation on fake memmap instances with in-memory147 # buffers148 a3 = a * 3149 assert not has_shareable_memory(a3)150 a3_reconstructed = reconstruct_array_or_memmap(a3)151 assert not has_shareable_memory(a3_reconstructed)152 assert not isinstance(a3_reconstructed, np.memmap)153 assert_array_equal(a3_reconstructed, a * 3)154 155 # Test graceful degradation on arrays derived from fake memmap instances156 b3 = np.asarray(a3)157 assert not has_shareable_memory(b3)158 159 b3_reconstructed = reconstruct_array_or_memmap(b3)160 assert isinstance(b3_reconstructed, np.ndarray)161 assert not has_shareable_memory(b3_reconstructed)162 assert_array_equal(b3_reconstructed, b3)163 164 165@with_numpy166@with_multiprocessing167@skipif(168 sys.platform != "win32", reason="PermissionError only easily triggerable on Windows"169)170def test_resource_tracker_retries_when_permissionerror(tmpdir):171 # Test resource_tracker retry mechanism when unlinking memmaps. See more172 # thorough information in the ``unlink_file`` documentation of joblib.173 filename = tmpdir.join("test.mmap").strpath174 cmd = """if 1:175 import os176 import numpy as np177 import time178 from joblib.externals.loky.backend import resource_tracker179 resource_tracker.VERBOSE = 1180 181 # Start the resource tracker182 resource_tracker.ensure_running()183 time.sleep(1)184 185 # Create a file containing numpy data186 memmap = np.memmap(r"{filename}", dtype=np.float64, shape=10, mode='w+')187 memmap[:] = np.arange(10).astype(np.int8).data188 memmap.flush()189 assert os.path.exists(r"{filename}")190 del memmap191 192 # Create a np.memmap backed by this file193 memmap = np.memmap(r"{filename}", dtype=np.float64, shape=10, mode='w+')194 resource_tracker.register(r"{filename}", "file")195 196 # Ask the resource_tracker to delete the file backing the np.memmap , this197 # should raise PermissionError that the resource_tracker will log.198 resource_tracker.maybe_unlink(r"{filename}", "file")199 200 # Wait for the resource_tracker to process the maybe_unlink before cleaning201 # up the memmap202 time.sleep(2)203 """.format(filename=filename)204 p = subprocess.Popen(205 [sys.executable, "-c", cmd], stderr=subprocess.PIPE, stdout=subprocess.PIPE206 )207 p.wait()208 out, err = p.communicate()209 assert p.returncode == 0, err.decode()210 assert out == b""211 msg = "tried to unlink {}, got PermissionError".format(filename)212 assert msg in err.decode()213 214 215@with_numpy216@with_multiprocessing217def test_high_dimension_memmap_array_reducing(tmpdir):218 assert_array_equal = np.testing.assert_array_equal219 220 filename = tmpdir.join("test.mmap").strpath221 222 # Create a high dimensional memmap223 a = np.memmap(filename, dtype=np.float64, shape=(100, 15, 15, 3), mode="w+")224 a[:] = np.arange(100 * 15 * 15 * 3).reshape(a.shape)225 226 # Create some slices/indices at various dimensions227 b = a[0:10]228 c = a[:, 5:10]229 d = a[:, :, :, 0]230 e = a[1:3:4]231 232 # Array reducer with auto dumping disabled233 reducer = ArrayMemmapForwardReducer(None, tmpdir.strpath, "c", True)234 235 def reconstruct_array_or_memmap(x):236 cons, args = reducer(x)237 return cons(*args)238 239 a_reconstructed = reconstruct_array_or_memmap(a)240 assert has_shareable_memory(a_reconstructed)241 assert isinstance(a_reconstructed, np.memmap)242 assert_array_equal(a_reconstructed, a)243 244 b_reconstructed = reconstruct_array_or_memmap(b)245 assert has_shareable_memory(b_reconstructed)246 assert_array_equal(b_reconstructed, b)247 248 c_reconstructed = reconstruct_array_or_memmap(c)249 assert has_shareable_memory(c_reconstructed)250 assert_array_equal(c_reconstructed, c)251 252 d_reconstructed = reconstruct_array_or_memmap(d)253 assert has_shareable_memory(d_reconstructed)254 assert_array_equal(d_reconstructed, d)255 256 e_reconstructed = reconstruct_array_or_memmap(e)257 assert has_shareable_memory(e_reconstructed)258 assert_array_equal(e_reconstructed, e)259 260 261@with_numpy262def test__strided_from_memmap(tmpdir):263 fname = tmpdir.join("test.mmap").strpath264 size = 5 * mmap.ALLOCATIONGRANULARITY265 offset = mmap.ALLOCATIONGRANULARITY + 1266 # This line creates the mmap file that is reused later267 memmap_obj = np.memmap(fname, mode="w+", shape=size + offset)268 # filename, dtype, mode, offset, order, shape, strides, total_buffer_len269 memmap_obj = _strided_from_memmap(270 fname,271 dtype="uint8",272 mode="r",273 offset=offset,274 order="C",275 shape=size,276 strides=None,277 total_buffer_len=None,278 unlink_on_gc_collect=False,279 )280 assert isinstance(memmap_obj, np.memmap)281 assert memmap_obj.offset == offset282 memmap_backed_obj = _strided_from_memmap(283 fname,284 dtype="uint8",285 mode="r",286 offset=offset,287 order="C",288 shape=(size // 2,),289 strides=(2,),290 total_buffer_len=size,291 unlink_on_gc_collect=False,292 )293 assert _get_backing_memmap(memmap_backed_obj).offset == offset294 295 296@with_numpy297@with_multiprocessing298@parametrize(299 "factory",300 [MemmappingPool, TestExecutor.get_memmapping_executor],301 ids=["multiprocessing", "loky"],302)303def test_pool_with_memmap(factory, tmpdir):304 """Check that subprocess can access and update shared memory memmap"""305 assert_array_equal = np.testing.assert_array_equal306 307 # Fork the subprocess before allocating the objects to be passed308 pool_temp_folder = tmpdir.mkdir("pool").strpath309 p = factory(10, max_nbytes=2, temp_folder=pool_temp_folder)310 try:311 filename = tmpdir.join("test.mmap").strpath312 a = np.memmap(filename, dtype=np.float32, shape=(3, 5), mode="w+")313 a.fill(1.0)314 315 p.map(316 inplace_double,317 [(a, (i, j), 1.0) for i in range(a.shape[0]) for j in range(a.shape[1])],318 )319 320 assert_array_equal(a, 2 * np.ones(a.shape))321 322 # Open a copy-on-write view on the previous data323 b = np.memmap(filename, dtype=np.float32, shape=(5, 3), mode="c")324 325 p.map(326 inplace_double,327 [(b, (i, j), 2.0) for i in range(b.shape[0]) for j in range(b.shape[1])],328 )329 330 # Passing memmap instances to the pool should not trigger the creation331 # of new files on the FS332 assert os.listdir(pool_temp_folder) == []333 334 # the original data is untouched335 assert_array_equal(a, 2 * np.ones(a.shape))336 assert_array_equal(b, 2 * np.ones(b.shape))337 338 # readonly maps can be read but not updated339 c = np.memmap(filename, dtype=np.float32, shape=(10,), mode="r", offset=5 * 4)340 341 with raises(AssertionError):342 p.map(check_array, [(c, i, 3.0) for i in range(c.shape[0])])343 344 # depending on the version of numpy one can either get a RuntimeError345 # or a ValueError346 with raises((RuntimeError, ValueError)):347 p.map(inplace_double, [(c, i, 2.0) for i in range(c.shape[0])])348 finally:349 # Clean all filehandlers held by the pool350 p.terminate()351 del p352 353 354@with_numpy355@with_multiprocessing356@parametrize(357 "factory",358 [MemmappingPool, TestExecutor.get_memmapping_executor],359 ids=["multiprocessing", "loky"],360)361def test_pool_with_memmap_array_view(factory, tmpdir):362 """Check that subprocess can access and update shared memory array"""363 assert_array_equal = np.testing.assert_array_equal364 365 # Fork the subprocess before allocating the objects to be passed366 pool_temp_folder = tmpdir.mkdir("pool").strpath367 p = factory(10, max_nbytes=2, temp_folder=pool_temp_folder)368 try:369 filename = tmpdir.join("test.mmap").strpath370 a = np.memmap(filename, dtype=np.float32, shape=(3, 5), mode="w+")371 a.fill(1.0)372 373 # Create an ndarray view on the memmap instance374 a_view = np.asarray(a)375 assert not isinstance(a_view, np.memmap)376 assert has_shareable_memory(a_view)377 378 p.map(379 inplace_double,380 [381 (a_view, (i, j), 1.0)382 for i in range(a.shape[0])383 for j in range(a.shape[1])384 ],385 )386 387 # Both a and the a_view have been updated388 assert_array_equal(a, 2 * np.ones(a.shape))389 assert_array_equal(a_view, 2 * np.ones(a.shape))390 391 # Passing memmap array view to the pool should not trigger the392 # creation of new files on the FS393 assert os.listdir(pool_temp_folder) == []394 395 finally:396 p.terminate()397 del p398 399 400@with_numpy401@with_multiprocessing402@parametrize("backend", ["multiprocessing", "loky"])403def test_permission_error_windows_reference_cycle(backend):404 # Non regression test for:405 # https://github.com/joblib/joblib/issues/806406 #407 # The issue happens when trying to delete a memory mapped file that has408 # not yet been closed by one of the worker processes.409 cmd = """if 1:410 import numpy as np411 from joblib import Parallel, delayed412 413 414 data = np.random.rand(int(2e6)).reshape((int(1e6), 2))415 416 # Build a complex cyclic reference that is likely to delay garbage417 # collection of the memmapped array in the worker processes.418 first_list = current_list = [data]419 for i in range(10):420 current_list = [current_list]421 first_list.append(current_list)422 423 if __name__ == "__main__":424 results = Parallel(n_jobs=2, backend="{b}")(425 delayed(len)(current_list) for i in range(10))426 assert results == [1] * 10427 """.format(b=backend)428 p = subprocess.Popen(429 [sys.executable, "-c", cmd], stderr=subprocess.PIPE, stdout=subprocess.PIPE430 )431 p.wait()432 out, err = p.communicate()433 assert p.returncode == 0, out.decode() + "\n\n" + err.decode()434 435 436@with_numpy437@with_multiprocessing438@parametrize("backend", ["multiprocessing", "loky"])439def test_permission_error_windows_memmap_sent_to_parent(backend):440 # Second non-regression test for:441 # https://github.com/joblib/joblib/issues/806442 # previously, child process would not convert temporary memmaps to numpy443 # arrays when sending the data back to the parent process. This would lead444 # to permission errors on windows when deleting joblib's temporary folder,445 # as the memmaped files handles would still opened in the parent process.446 cmd = """if 1:447 import os448 import time449 450 import numpy as np451 452 from joblib import Parallel, delayed453 from testutils import return_slice_of_data454 455 data = np.ones(int(2e6))456 457 if __name__ == '__main__':458 # warm-up call to launch the workers and start the resource_tracker459 _ = Parallel(n_jobs=2, verbose=5, backend='{b}')(460 delayed(id)(i) for i in range(20))461 462 time.sleep(0.5)463 464 slice_of_data = Parallel(n_jobs=2, verbose=5, backend='{b}')(465 delayed(return_slice_of_data)(data, 0, 20) for _ in range(10))466 """.format(b=backend)467 468 for _ in range(3):469 env = os.environ.copy()470 env["PYTHONPATH"] = os.path.dirname(__file__)471 p = subprocess.Popen(472 [sys.executable, "-c", cmd],473 stderr=subprocess.PIPE,474 stdout=subprocess.PIPE,475 env=env,476 )477 p.wait()478 out, err = p.communicate()479 assert p.returncode == 0, err480 assert out == b""481 assert b"resource_tracker" not in err482 483 484@with_numpy485@with_multiprocessing486@parametrize("backend", ["multiprocessing", "loky"])487def test_parallel_isolated_temp_folders(backend):488 # Test that consecutive Parallel call use isolated subfolders, even489 # for the loky backend that reuses its executor instance across calls.490 array = np.arange(int(1e2))491 [filename_1] = Parallel(n_jobs=2, backend=backend, max_nbytes=10)(492 delayed(getattr)(array, "filename") for _ in range(1)493 )494 [filename_2] = Parallel(n_jobs=2, backend=backend, max_nbytes=10)(495 delayed(getattr)(array, "filename") for _ in range(1)496 )497 assert os.path.dirname(filename_2) != os.path.dirname(filename_1)498 499 500@with_numpy501@with_multiprocessing502@parametrize("backend", ["multiprocessing", "loky"])503def test_managed_backend_reuse_temp_folder(backend):504 # Test that calls to a managed parallel object reuse the same memmaps.505 array = np.arange(int(1e2))506 with Parallel(n_jobs=2, backend=backend, max_nbytes=10) as p:507 [filename_1] = p(delayed(getattr)(array, "filename") for _ in range(1))508 [filename_2] = p(delayed(getattr)(array, "filename") for _ in range(1))509 assert os.path.dirname(filename_2) == os.path.dirname(filename_1)510 511 512@with_numpy513@with_multiprocessing514def test_memmapping_temp_folder_thread_safety():515 # Concurrent calls to Parallel with the loky backend will use the same516 # executor, and thus the same reducers. Make sure that those reducers use517 # different temporary folders depending on which Parallel objects called518 # them, which is necessary to limit potential race conditions during the519 # garbage collection of temporary memmaps.520 array = np.arange(int(1e2))521 522 temp_dirs_thread_1 = set()523 temp_dirs_thread_2 = set()524 525 def concurrent_get_filename(array, temp_dirs):526 with Parallel(backend="loky", n_jobs=2, max_nbytes=10) as p:527 for i in range(10):528 [filename] = p(delayed(getattr)(array, "filename") for _ in range(1))529 temp_dirs.add(os.path.dirname(filename))530 531 t1 = threading.Thread(532 target=concurrent_get_filename, args=(array, temp_dirs_thread_1)533 )534 t2 = threading.Thread(535 target=concurrent_get_filename, args=(array, temp_dirs_thread_2)536 )537 538 t1.start()539 t2.start()540 541 t1.join()542 t2.join()543 544 assert len(temp_dirs_thread_1) == 1545 assert len(temp_dirs_thread_2) == 1546 547 assert temp_dirs_thread_1 != temp_dirs_thread_2548 549 550@with_numpy551@with_multiprocessing552def test_multithreaded_parallel_termination_resource_tracker_silent():553 # test that concurrent termination attempts of a same executor does not554 # emit any spurious error from the resource_tracker. We test various555 # situations making 0, 1 or both parallel call sending a task that will556 # make the worker (and thus the whole Parallel call) error out.557 cmd = """if 1:558 import os559 import numpy as np560 from joblib import Parallel, delayed561 from joblib.externals.loky.backend import resource_tracker562 from concurrent.futures import ThreadPoolExecutor, wait563 564 resource_tracker.VERBOSE = 0565 566 array = np.arange(int(1e2))567 568 temp_dirs_thread_1 = set()569 temp_dirs_thread_2 = set()570 571 572 def raise_error(array):573 raise ValueError574 575 576 def parallel_get_filename(array, temp_dirs):577 with Parallel(backend="loky", n_jobs=2, max_nbytes=10) as p:578 for i in range(10):579 [filename] = p(580 delayed(getattr)(array, "filename") for _ in range(1)581 )582 temp_dirs.add(os.path.dirname(filename))583 584 585 def parallel_raise(array, temp_dirs):586 with Parallel(backend="loky", n_jobs=2, max_nbytes=10) as p:587 for i in range(10):588 [filename] = p(589 delayed(raise_error)(array) for _ in range(1)590 )591 temp_dirs.add(os.path.dirname(filename))592 593 594 executor = ThreadPoolExecutor(max_workers=2)595 596 # both function calls will use the same loky executor, but with a597 # different Parallel object.598 future_1 = executor.submit({f1}, array, temp_dirs_thread_1)599 future_2 = executor.submit({f2}, array, temp_dirs_thread_2)600 601 # Wait for both threads to terminate their backend602 wait([future_1, future_2])603 604 future_1.result()605 future_2.result()606 """607 functions_and_returncodes = [608 ("parallel_get_filename", "parallel_get_filename", 0),609 ("parallel_get_filename", "parallel_raise", 1),610 ("parallel_raise", "parallel_raise", 1),611 ]612 613 for f1, f2, returncode in functions_and_returncodes:614 p = subprocess.Popen(615 [sys.executable, "-c", cmd.format(f1=f1, f2=f2)],616 stderr=subprocess.PIPE,617 stdout=subprocess.PIPE,618 )619 p.wait()620 _, err = p.communicate()621 assert p.returncode == returncode, err.decode()622 assert b"resource_tracker" not in err, err.decode()623 624 625@with_numpy626@with_multiprocessing627@parametrize("backend", ["multiprocessing", "loky"])628def test_many_parallel_calls_on_same_object(backend):629 # After #966 got merged, consecutive Parallel objects were sharing temp630 # folder, which would lead to race conditions happening during the631 # temporary resources management with the resource_tracker. This is a632 # non-regression test that makes sure that consecutive Parallel operations633 # on the same object do not error out.634 cmd = """if 1:635 import os636 import time637 638 import numpy as np639 640 from joblib import Parallel, delayed641 from testutils import return_slice_of_data642 643 data = np.ones(100)644 645 if __name__ == '__main__':646 for i in range(5):647 slice_of_data = Parallel(648 n_jobs=2, max_nbytes=1, backend='{b}')(649 delayed(return_slice_of_data)(data, 0, 20)650 for _ in range(10)651 )652 """.format(b=backend)653 env = os.environ.copy()654 env["PYTHONPATH"] = os.path.dirname(__file__)655 p = subprocess.Popen(656 [sys.executable, "-c", cmd],657 stderr=subprocess.PIPE,658 stdout=subprocess.PIPE,659 env=env,660 )661 p.wait()662 out, err = p.communicate()663 assert p.returncode == 0, err.decode()664 assert out == b"", out.decode()665 assert b"resource_tracker" not in err666 667 668@with_numpy669@with_multiprocessing670@parametrize("backend", ["multiprocessing", "loky"])671def test_memmap_returned_as_regular_array(backend):672 data = np.ones(int(1e3))673 # Check that child processes send temporary memmaps back as numpy arrays.674 [result] = Parallel(n_jobs=2, backend=backend, max_nbytes=100)(675 delayed(check_memmap_and_send_back)(data) for _ in range(1)676 )677 assert _get_backing_memmap(result) is None678 679 680@with_numpy681@with_multiprocessing682@parametrize("backend", ["multiprocessing", "loky"])683def test_resource_tracker_silent_when_reference_cycles(backend):684 # There is a variety of reasons that can make joblib with loky backend685 # output noisy warnings when a reference cycle is preventing a memmap from686 # being garbage collected. Especially, joblib's main process finalizer687 # deletes the temporary folder if it was not done before, which can688 # interact badly with the resource_tracker. We don't risk leaking any689 # resources, but this will likely make joblib output a lot of low-level690 # confusing messages.691 #692 # This test makes sure that the resource_tracker is silent when a reference693 # has been collected concurrently on non-Windows platforms.694 #695 # Note that the script in ``cmd`` is the exact same script as in696 # test_permission_error_windows_reference_cycle.697 if backend == "loky" and sys.platform.startswith("win"):698 # XXX: on Windows, reference cycles can delay timely garbage collection699 # and make it impossible to properly delete the temporary folder in the700 # main process because of permission errors.701 pytest.xfail(702 "The temporary folder cannot be deleted on Windows in the "703 "presence of a reference cycle"704 )705 706 cmd = """if 1:707 import numpy as np708 from joblib import Parallel, delayed709 710 711 data = np.random.rand(int(2e6)).reshape((int(1e6), 2))712 713 # Build a complex cyclic reference that is likely to delay garbage714 # collection of the memmapped array in the worker processes.715 first_list = current_list = [data]716 for i in range(10):717 current_list = [current_list]718 first_list.append(current_list)719 720 if __name__ == "__main__":721 results = Parallel(n_jobs=2, backend="{b}")(722 delayed(len)(current_list) for i in range(10))723 assert results == [1] * 10724 """.format(b=backend)725 p = subprocess.Popen(726 [sys.executable, "-c", cmd], stderr=subprocess.PIPE, stdout=subprocess.PIPE727 )728 p.wait()729 out, err = p.communicate()730 out = out.decode()731 err = err.decode()732 assert p.returncode == 0, out + "\n\n" + err733 assert "resource_tracker" not in err, err734 735 736@with_numpy737@with_multiprocessing738@parametrize(739 "factory",740 [MemmappingPool, TestExecutor.get_memmapping_executor],741 ids=["multiprocessing", "loky"],742)743def test_memmapping_pool_for_large_arrays(factory, tmpdir):744 """Check that large arrays are not copied in memory"""745 746 # Check that the tempfolder is empty747 assert os.listdir(tmpdir.strpath) == []748 749 # Build an array reducers that automatically dump large array content750 # to filesystem backed memmap instances to avoid memory explosion751 p = factory(3, max_nbytes=40, temp_folder=tmpdir.strpath, verbose=2)752 try:753 # The temporary folder for the pool is not provisioned in advance754 assert os.listdir(tmpdir.strpath) == []755 assert not os.path.exists(p._temp_folder)756 757 small = np.ones(5, dtype=np.float32)758 assert small.nbytes == 20759 p.map(check_array, [(small, i, 1.0) for i in range(small.shape[0])])760 761 # Memory has been copied, the pool filesystem folder is unused762 assert os.listdir(tmpdir.strpath) == []763 764 # Try with a file larger than the memmap threshold of 40 bytes765 large = np.ones(100, dtype=np.float64)766 assert large.nbytes == 800767 p.map(check_array, [(large, i, 1.0) for i in range(large.shape[0])])768 769 # The data has been dumped in a temp folder for subprocess to share it770 # without per-child memory copies771 assert os.path.isdir(p._temp_folder)772 dumped_filenames = os.listdir(p._temp_folder)773 assert len(dumped_filenames) == 1774 775 # Check that memory mapping is not triggered for arrays with776 # dtype='object'777 objects = np.array(["abc"] * 100, dtype="object")778 results = p.map(has_shareable_memory, [objects])779 assert not results[0]780 781 finally:782 # check FS garbage upon pool termination783 p.terminate()784 for i in range(10):785 sleep(0.1)786 if not os.path.exists(p._temp_folder):787 break788 else: # pragma: no cover789 raise AssertionError(790 "temporary folder {} was not deleted".format(p._temp_folder)791 )792 del p793 794 795@with_numpy796@with_multiprocessing797@parametrize(798 "backend",799 [800 pytest.param(801 "multiprocessing",802 marks=pytest.mark.xfail(803 reason="https://github.com/joblib/joblib/issues/1086"804 ),805 ),806 "loky",807 ],808)809def test_child_raises_parent_exits_cleanly(backend):810 # When a task executed by a child process raises an error, the parent811 # process's backend is notified, and calls abort_everything.812 # In loky, abort_everything itself calls shutdown(kill_workers=True) which813 # sends SIGKILL to the worker, preventing it from running the finalizers814 # supposed to signal the resource_tracker when the worker is done using815 # objects relying on a shared resource (e.g np.memmaps). Because this816 # behavior is prone to :817 # - cause a resource leak818 # - make the resource tracker emit noisy resource warnings819 # we explicitly test that, when the said situation occurs:820 # - no resources are actually leaked821 # - the temporary resources are deleted as soon as possible (typically, at822 # the end of the failing Parallel call)823 # - the resource_tracker does not emit any warnings.824 cmd = """if 1:825 import os826 from pathlib import Path827 from time import sleep828 829 import numpy as np830 from joblib import Parallel, delayed831 from testutils import print_filename_and_raise832 833 data = np.random.rand(1000)834 835 def get_temp_folder(parallel_obj, backend):836 if "{b}" == "loky":837 return Path(parallel_obj._backend._workers._temp_folder)838 else:839 return Path(parallel_obj._backend._pool._temp_folder)840 841 842 if __name__ == "__main__":843 try:844 with Parallel(n_jobs=2, backend="{b}", max_nbytes=100) as p:845 temp_folder = get_temp_folder(p, "{b}")846 p(delayed(print_filename_and_raise)(data)847 for i in range(1))848 except ValueError as e:849 # the temporary folder should be deleted by the end of this850 # call but apparently on some file systems, this takes851 # some time to be visible.852 #853 # We attempt to write into the temporary folder to test for854 # its existence and we wait for a maximum of 10 seconds.855 for i in range(100):856 try:857 with open(temp_folder / "some_file.txt", "w") as f:858 f.write("some content")859 except FileNotFoundError:860 # temp_folder has been deleted, all is fine861 break862 863 # ... else, wait a bit and try again864 sleep(.1)865 else:866 raise AssertionError(867 str(temp_folder) + " was not deleted"868 ) from e869 """.format(b=backend)870 env = os.environ.copy()871 env["PYTHONPATH"] = os.path.dirname(__file__)872 p = subprocess.Popen(873 [sys.executable, "-c", cmd],874 stderr=subprocess.PIPE,875 stdout=subprocess.PIPE,876 env=env,877 )878 p.wait()879 out, err = p.communicate()880 out, err = out.decode(), err.decode()881 filename = out.split("\n")[0]882 assert p.returncode == 0, err or out883 assert err == "" # no resource_tracker warnings.884 assert not os.path.exists(filename)885 886 887@with_numpy888@with_multiprocessing889@parametrize(890 "factory",891 [MemmappingPool, TestExecutor.get_memmapping_executor],892 ids=["multiprocessing", "loky"],893)894def test_memmapping_pool_for_large_arrays_disabled(factory, tmpdir):895 """Check that large arrays memmapping can be disabled"""896 # Set max_nbytes to None to disable the auto memmapping feature897 p = factory(3, max_nbytes=None, temp_folder=tmpdir.strpath)898 try:899 # Check that the tempfolder is empty900 assert os.listdir(tmpdir.strpath) == []901 902 # Try with a file largish than the memmap threshold of 40 bytes903 large = np.ones(100, dtype=np.float64)904 assert large.nbytes == 800905 p.map(check_array, [(large, i, 1.0) for i in range(large.shape[0])])906 907 # Check that the tempfolder is still empty908 assert os.listdir(tmpdir.strpath) == []909 910 finally:911 # Cleanup open file descriptors912 p.terminate()913 del p914 915 916@with_numpy917@with_multiprocessing918@with_dev_shm919@parametrize(920 "factory",921 [MemmappingPool, TestExecutor.get_memmapping_executor],922 ids=["multiprocessing", "loky"],923)924def test_memmapping_on_large_enough_dev_shm(factory):925 """Check that memmapping uses /dev/shm when possible"""926 orig_size = jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE927 try:928 # Make joblib believe that it can use /dev/shm even when running on a929 # CI container where the size of the /dev/shm is not very large (that930 # is at least 32 MB instead of 2 GB by default).931 jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE = int(32e6)932 p = factory(3, max_nbytes=10)933 try:934 # Check that the pool has correctly detected the presence of the935 # shared memory filesystem.936 pool_temp_folder = p._temp_folder937 folder_prefix = "/dev/shm/joblib_memmapping_folder_"938 assert pool_temp_folder.startswith(folder_prefix)939 assert os.path.exists(pool_temp_folder)940 941 # Try with a file larger than the memmap threshold of 10 bytes942 a = np.ones(100, dtype=np.float64)943 assert a.nbytes == 800944 p.map(id, [a] * 10)945 # a should have been memmapped to the pool temp folder: the joblib946 # pickling procedure generate one .pkl file:947 assert len(os.listdir(pool_temp_folder)) == 1948 949 # create a new array with content that is different from 'a' so950 # that it is mapped to a different file in the temporary folder of951 # the pool.952 b = np.ones(100, dtype=np.float64) * 2953 assert b.nbytes == 800954 p.map(id, [b] * 10)955 # A copy of both a and b are now stored in the shared memory folder956 assert len(os.listdir(pool_temp_folder)) == 2957 finally:958 # Cleanup open file descriptors959 p.terminate()960 del p961 962 for i in range(100):963 # The temp folder is cleaned up upon pool termination964 if not os.path.exists(pool_temp_folder):965 break966 sleep(0.1)967 else: # pragma: no cover968 raise AssertionError("temporary folder of pool was not deleted")969 finally:970 jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE = orig_size971 972 973@with_numpy974@with_multiprocessing975@with_dev_shm976@parametrize(977 "factory",978 [MemmappingPool, TestExecutor.get_memmapping_executor],979 ids=["multiprocessing", "loky"],980)981def test_memmapping_on_too_small_dev_shm(factory):982 orig_size = jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE983 try:984 # Make joblib believe that it cannot use /dev/shm unless there is985 # 42 exabytes of available shared memory in /dev/shm986 jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE = int(42e18)987 988 p = factory(3, max_nbytes=10)989 try:990 # Check that the pool has correctly detected the presence of the991 # shared memory filesystem.992 pool_temp_folder = p._temp_folder993 assert not pool_temp_folder.startswith("/dev/shm")994 finally:995 # Cleanup open file descriptors996 p.terminate()997 del p998 999 # The temp folder is cleaned up upon pool termination1000 assert not os.path.exists(pool_temp_folder)1001 finally:1002 jmr.SYSTEM_SHARED_MEM_FS_MIN_SIZE = orig_size1003 1004 1005@with_numpy1006@with_multiprocessing1007@parametrize(1008 "factory",1009 [MemmappingPool, TestExecutor.get_memmapping_executor],1010 ids=["multiprocessing", "loky"],1011)1012def test_memmapping_pool_for_large_arrays_in_return(factory, tmpdir):1013 """Check that large arrays are not copied in memory in return"""1014 assert_array_equal = np.testing.assert_array_equal1015 1016 # Build an array reducers that automatically dump large array content1017 # but check that the returned datastructure are regular arrays to avoid1018 # passing a memmap array pointing to a pool controlled temp folder that1019 # might be confusing to the user1020 1021 # The MemmappingPool user can always return numpy.memmap object explicitly1022 # to avoid memory copy1023 p = factory(3, max_nbytes=10, temp_folder=tmpdir.strpath)1024 try:1025 res = p.apply_async(np.ones, args=(1000,))1026 large = res.get()1027 assert not has_shareable_memory(large)1028 assert_array_equal(large, np.ones(1000))1029 finally:1030 p.terminate()1031 del p1032 1033 1034def _worker_multiply(a, n_times):1035 """Multiplication function to be executed by subprocess"""1036 assert has_shareable_memory(a)1037 return a * n_times1038 1039 1040@with_numpy1041@with_multiprocessing1042@parametrize(1043 "factory",1044 [MemmappingPool, TestExecutor.get_memmapping_executor],1045 ids=["multiprocessing", "loky"],1046)1047def test_workaround_against_bad_memmap_with_copied_buffers(factory, tmpdir):1048 """Check that memmaps with a bad buffer are returned as regular arrays1049 1050 Unary operations and ufuncs on memmap instances return a new memmap1051 instance with an in-memory buffer (probably a numpy bug).1052 """1053 assert_array_equal = np.testing.assert_array_equal1054 1055 p = factory(3, max_nbytes=10, temp_folder=tmpdir.strpath)1056 try:1057 # Send a complex, large-ish view on a array that will be converted to1058 # a memmap in the worker process1059 a = np.asarray(np.arange(6000).reshape((1000, 2, 3)), order="F")[:, :1, :]1060 1061 # Call a non-inplace multiply operation on the worker and memmap and1062 # send it back to the parent.1063 b = p.apply_async(_worker_multiply, args=(a, 3)).get()1064 assert not has_shareable_memory(b)1065 assert_array_equal(b, 3 * a)1066 finally:1067 p.terminate()1068 del p1069 1070 1071def identity(arg):1072 return arg1073 1074 1075@with_numpy1076@with_multiprocessing1077@parametrize(1078 "factory,retry_no",1079 list(1080 itertools.product(1081 [MemmappingPool, TestExecutor.get_memmapping_executor], range(3)1082 )1083 ),1084 ids=[1085 "{}, {}".format(x, y)1086 for x, y in itertools.product(["multiprocessing", "loky"], map(str, range(3)))1087 ],1088)1089def test_pool_memmap_with_big_offset(factory, retry_no, tmpdir):1090 # Test that numpy memmap offset is set correctly if greater than1091 # mmap.ALLOCATIONGRANULARITY, see1092 # https://github.com/joblib/joblib/issues/451 and1093 # https://github.com/numpy/numpy/pull/8443 for more details.1094 fname = tmpdir.join("test.mmap").strpath1095 size = 5 * mmap.ALLOCATIONGRANULARITY1096 offset = mmap.ALLOCATIONGRANULARITY + 11097 obj = make_memmap(fname, mode="w+", shape=size, dtype="uint8", offset=offset)1098 1099 p = factory(2, temp_folder=tmpdir.strpath)1100 result = p.apply_async(identity, args=(obj,)).get()1101 assert isinstance(result, np.memmap)1102 assert result.offset == offset1103 np.testing.assert_array_equal(obj, result)1104 p.terminate()1105 1106 1107def test_pool_get_temp_dir(tmpdir):1108 pool_folder_name = "test.tmpdir"1109 pool_folder, shared_mem = _get_temp_dir(pool_folder_name, tmpdir.strpath)1110 assert shared_mem is False1111 assert pool_folder == tmpdir.join("test.tmpdir").strpath1112 1113 pool_folder, shared_mem = _get_temp_dir(pool_folder_name, temp_folder=None)1114 if sys.platform.startswith("win"):1115 assert shared_mem is False1116 assert pool_folder.endswith(pool_folder_name)1117 1118 1119def test_pool_get_temp_dir_no_statvfs(tmpdir, monkeypatch):1120 """Check that _get_temp_dir works when os.statvfs is not defined1121 1122 Regression test for #9021123 """1124 pool_folder_name = "test.tmpdir"1125 import joblib._memmapping_reducer1126 1127 if hasattr(joblib._memmapping_reducer.os, "statvfs"):1128 # We are on Unix, since Windows doesn't have this function1129 monkeypatch.delattr(joblib._memmapping_reducer.os, "statvfs")1130 1131 pool_folder, shared_mem = _get_temp_dir(pool_folder_name, temp_folder=None)1132 if sys.platform.startswith("win"):1133 assert shared_mem is False1134 assert pool_folder.endswith(pool_folder_name)1135 1136 1137@with_numpy1138@skipif(1139 sys.platform == "win32", reason="This test fails with a PermissionError on Windows"1140)1141@parametrize("mmap_mode", ["r+", "w+"])1142def test_numpy_arrays_use_different_memory(mmap_mode):1143 def func(arr, value):1144 arr[:] = value1145 return arr1146 1147 arrays = [np.zeros((10, 10), dtype="float64") for i in range(10)]1148 1149 results = Parallel(mmap_mode=mmap_mode, max_nbytes=0, n_jobs=2)(1150 delayed(func)(arr, i) for i, arr in enumerate(arrays)1151 )1152 1153 for i, arr in enumerate(results):1154 np.testing.assert_array_equal(arr, i)1155 1156 1157@with_numpy1158def test_weak_array_key_map():1159 def assert_empty_after_gc_collect(container, retries=100):1160 for i in range(retries):1161 if len(container) == 0:1162 return1163 gc.collect()1164 sleep(0.1)1165 assert len(container) == 01166 1167 a = np.ones(42)1168 m = _WeakArrayKeyMap()1169 m.set(a, "a")1170 assert m.get(a) == "a"1171 1172 b = a1173 assert m.get(b) == "a"1174 m.set(b, "b")1175 assert m.get(a) == "b"1176 1177 del a1178 gc.collect()1179 assert len(m._data) == 11180 assert m.get(b) == "b"1181 1182 del b1183 assert_empty_after_gc_collect(m._data)1184 1185 c = np.ones(42)1186 m.set(c, "c")1187 assert len(m._data) == 11188 assert m.get(c) == "c"1189 1190 with raises(KeyError):1191 m.get(np.ones(42))1192 1193 del c1194 assert_empty_after_gc_collect(m._data)1195 1196 # Check that creating and dropping numpy arrays with potentially the same1197 # object id will not cause the map to get confused.1198 def get_set_get_collect(m, i):1199 a = np.ones(42)1200 with raises(KeyError):