Repository: joblib/joblib

Issue:
Test failures when number of CPU is 1
Hello. As reported in Debian:

https://bugs.debian.org/cgi-bin/bugreport.cgi?bug=1085692

Some tests fail when the machine has a single CPU:

```
=========================== short test summary info ============================
FAILED joblib/test/test_parallel.py::test_nested_parallelism_limit[parallel_config-loky] - AssertionError: assert [('LokyBacken...lBackend', 2)] == [('LokyBacken...lB...
FAILED joblib/test/test_parallel.py::test_nested_parallelism_limit[parallel_config-threading] - AssertionError: assert [('ThreadingB...lBackend', 2)] == [('ThreadingB...lB...
FAILED joblib/test/test_parallel.py::test_nested_parallelism_limit[parallel_backend-loky] - AssertionError: assert [('LokyBacken...lBackend', 2)] == [('LokyBacken...lB...
FAILED joblib/test/test_parallel.py::test_nested_parallelism_limit[parallel_backend-threading] - AssertionError: assert [('ThreadingB...lBackend', 2)] == [('ThreadingB...lB...
FAILED joblib/test/test_parallel.py::test_threadpool_limitation_in_child_override[parallel_config-OPENBLAS_NUM_THREADS--1] - AssertionError: assert ['4', '4'] == ['1', '1']
FAILED joblib/test/test_parallel.py::test_threadpool_limitation_in_child_override[parallel_config-MKL_NUM_THREADS--1] - AssertionError: assert ['4', '4'] == ['1', '1']
FAILED joblib/test/test_parallel.py::test_threadpool_limitation_in_child_override[parallel_config-OMP_NUM_THREADS--1] - AssertionError: assert ['4', '4'] == ['1', '1']
FAILED joblib/test/test_parallel.py::test_threadpool_limitation_in_child_override[parallel_backend-OPENBLAS_NUM_THREADS--1] - AssertionError: assert ['4', '4'] == ['1', '1']
FAILED joblib/test/test_parallel.py::test_threadpool_limitation_in_child_override[parallel_backend-MKL_NUM_THREADS--1] - AssertionError: assert ['4', '4'] == ['1', '1']
FAILED joblib/test/test_parallel.py::test_threadpool_limitation_in_child_override[parallel_backend-OMP_NUM_THREADS--1] - AssertionError: assert ['4', '4'] == ['1', '1']
= 10 failed, 1308 passed, 42 skipped, 2 deselected, 2 xfailed, 4 xpassed, 317 warnings in 84.09s (0:01:24) =
```

This used to work in version 0.13.0, and there are occurrences of `cpu_count() == 1` in the code, so I believe this was not supposed to happen.

Thanks.

Discussion:
@lesteve: I agree the tests with `cpu_count` looks like they should handle this [^1] so I would guess that `joblib.parallel.cpu_count()` is wrong in this case.

Is it doable to double-check what `joblib.parallel.cpu_count()` returns in the Debian CI?

Do you happen to know how the restriction to 1 CPU is done, for example `cgroup`, CPU affinity, something else? This would help trying to reproduce and maybe fix `cpu_count`.

[^1]: other tests would need to be adapted I think@sanvila: The error does not happen on Debian CI, it happens when trying to build the package on a virtual machine with a single-CPU (for example, AWS instances of type m7a.medium or  r7a.medium). Building the archive on single-CPU machines is still the most efficient way to build all Debian packages.

If you don't have a single-CPU virtual machine to test, `GRUB_CMDLINE_LINUX="nr_cpus=1"` usually yields the same outcome.

In the meantime, we fixed the Debian package in this commit:

https://salsa.debian.org/science-team/joblib/-/commit/89e386f951c43a7628d6cb19675540f4c588b699

but I agree that the portable way would have been to use a skipif using `os.cpu_count()`.

Thanks.@lesteve: Thanks for your answer. So I think the issue is that `cpu_count()` is wrong in some specific cases to be debugged. I would have thought that, with our cgroup handling, we were covering AWS instances but maybe not ...

I'll try to give a go to `GRUB_CMDLINE="nr_cpus=1"`.
@lesteve: Turns out, it is an issue with the tests themselves when `cpu_count() == 1` rather than `cpu_count` being wrong.

I can reproduce the failures by running the following command:
```
LOKY_MAX_CPU_COUNT=1 pytest joblib/test/test_parallel.py
```@lesteve: After having looked a bit more, this is not a problem in `joblib` but more than the tests needs to be updated for the `cpu_count() == 1` special case.

I guess it may also may make sense to have a CI build with `LOKY_MAX_CPU_COUNT=1` so that we avoid creating issues in Debian packaging ...@sanvila: > I guess it may also may make sense to have a CI build with LOKY_MAX_CPU_COUNT=1 so that we avoid creating issues in Debian packaging ...

If you can detect this problem in advance as part of your release process, that would be awesome, yes.

On the other hand, don't worry too much about the extra work on the Debian side. We already disabled the test in debian/rules, but naturally, we would prefer a solution which is valid for everybody.

Thanks.

(Note. This message has been edited because I was initially confused about what you meant),

Solution patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index ba966891c..debac7ddf 100644
--- a/.github/workflows/test.yml
+++ b/.github/workflows/test.yml
@@ -81,6 +81,10 @@ jobs:
             os: ubuntu-latest
             PYTHON_VERSION: "3.10"
             NO_LZ4: true
+          - name: one-cpu
+            os: ubuntu-latest
+            PYTHON_VERSION: "3.12"
+            LOKY_MAX_CPU_COUNT: 1
           - name: default-backend-threading
             os: ubuntu-latest
             PYTHON_VERSION: "3.9"
diff --git a/CHANGES.rst b/CHANGES.rst
index 74f272c08..430a4ad08 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -20,8 +20,8 @@ Memory:
   if it is in cache memory.
   https://github.com/joblib/joblib/pull/1584
 
-- The Memory object now automatically creates a .gitignore file in its cache directory,
-  instructing git to ignore the entire folder.
+- The Memory object now automatically creates a .gitignore file in its cache
+  directory, instructing git to ignore the entire folder.
   https://github.com/joblib/joblib/pull/1674
 
 Parallel:
@@ -76,12 +76,15 @@ Maintenance:
   an extra argument in ``Pickler._batch_setitems``.
   https://github.com/joblib/joblib/pull/1688
 
-- Bump vendored cloudpickle to 3.1.1 to support Python 3.14 (dev) and
+- Fix tests on platforms with only one CPU core.
+  https://github.com/joblib/joblib/pull/1682
+
+- Bump vendored cloudpickle to ``3.1.1`` to support Python 3.14 (dev) and
   various other fixes.
 
-- Bump vendored loky to 3.5.1 to support recent Python versions without raising
-  the warning on calls to `os.fork` and fix various sources of crashes and
-  deadlocks.
+- Bump vendored loky to ``3.5.3`` to support recent Python versions without
+  raising the warning on calls to `os.fork` and fix various sources of crashes
+  and deadlocks.
 
 - Use ``pickle`` protocol 5 for pickling ``numpy`` arrays with object type.
   https://github.com/joblib/joblib/pull/1682
diff --git a/joblib/externals/loky/__init__.py b/joblib/externals/loky/__init__.py
index 9a1a1faaa..cf08e3422 100644
--- a/joblib/externals/loky/__init__.py
+++ b/joblib/externals/loky/__init__.py
@@ -42,4 +42,4 @@
 ]
 
 
-__version__ = "3.5.2"
+__version__ = "3.5.3"
diff --git a/joblib/externals/loky/process_executor.py b/joblib/externals/loky/process_executor.py
index 3b20ad312..3f32994e1 100644
--- a/joblib/externals/loky/process_executor.py
+++ b/joblib/externals/loky/process_executor.py
@@ -1263,7 +1263,7 @@ def submit(self, fn, *args, **kwargs):
             # This check avoids spawning new processes at exit.
             if _global_shutdown:
                 raise RuntimeError(
-                    "cannot schedule new futures after " "interpreter shutdown"
+                    "cannot schedule new futures after interpreter shutdown"
                 )
 
             f = Future()
diff --git a/joblib/externals/loky/reusable_executor.py b/joblib/externals/loky/reusable_executor.py
index ad016fd38..faf604c21 100644
--- a/joblib/externals/loky/reusable_executor.py
+++ b/joblib/externals/loky/reusable_executor.py
@@ -192,11 +192,16 @@ def get_reusable_executor(
                     executor._flags.broken
                     or executor._flags.shutdown
                     or not reuse
+                    or executor.queue_size < max_workers
                 ):
                     if executor._flags.broken:
                         reason = "broken"
                     elif executor._flags.shutdown:
                         reason = "shutdown"
+                    elif executor.queue_size < max_workers:
+                        # Do not reuse the executor if the queue size is too
+                        # small as this would lead to limited parallelism.
+                        reason = "queue size is too small"
                     else:
                         reason = "arguments have changed"
                     mp.util.debug(
@@ -279,7 +284,11 @@ def _wait_job_completion(self):
     def _setup_queues(self, job_reducers, result_reducers):
         # As this executor can be resized, use a large queue size to avoid
         # underestimating capacity and introducing overhead
-        queue_size = 2 * cpu_count() + EXTRA_QUEUED_CALLS
+        # Also handle the case where the user set max_workers to a value larger
+        # than cpu_count(), to avoid limiting the number of parallel jobs.
+
+        min_queue_size = max(cpu_count(), self._max_workers)
+        self.queue_size = 2 * min_queue_size + EXTRA_QUEUED_CALLS
         super()._setup_queues(
-            job_reducers, result_reducers, queue_size=queue_size
+            job_reducers, result_reducers, queue_size=self.queue_size
         )


Test patch:
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index 6e4cda73f..db3fdc927 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -1825,19 +1825,12 @@ 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),
+        ("ThreadingBackend", 1),
+        ("SequentialBackend", 2),
+        ("SequentialBackend", 2),
     ]
     assert backend_types_and_levels == expected_types_and_levels
 
@@ -2051,7 +2044,9 @@ def test_threadpool_limitation_in_child_context(context, n_jobs, inner_max_num_t
         )
 
     n_jobs = effective_n_jobs(n_jobs)
-    if inner_max_num_threads is None:
+    if n_jobs == 1:
+        expected_child_num_threads = parent_info[0]["num_threads"]
+    elif 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
@@ -2069,6 +2064,10 @@ 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.
 
+    # Skip this test if the process is run sequetially
+    if effective_n_jobs(n_jobs) == 1:
+        pytest.skip("Skip test when n_jobs == 1")
+
     # 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()
@@ -2150,7 +2149,6 @@ def _check_status(status, n_jobs, wait_workers=False):
 @parametrize("backend", PROCESS_BACKENDS)
 @parametrize("context", [parallel_config, parallel_backend])
 def test_initializer_context(n_jobs, backend, context):
-    n_jobs = effective_n_jobs(n_jobs)
     manager = mp.Manager()
     status = manager.dict()
 
@@ -2169,7 +2167,6 @@ def test_initializer_context(n_jobs, backend, context):
 @parametrize("n_jobs", [2, 4])
 @parametrize("backend", PROCESS_BACKENDS)
 def test_initializer_parallel(n_jobs, backend):
-    n_jobs = effective_n_jobs(n_jobs)
     manager = mp.Manager()
     status = manager.dict()
 
@@ -2189,7 +2186,6 @@ def test_initializer_reused(n_jobs):
     # Check that it is possible to pass initializer config via the `Parallel`
     # call directly and the worker are reused when the arguments are the same.
     n_repetitions = 3
-    n_jobs = effective_n_jobs(n_jobs)
     manager = mp.Manager()
     status = manager.dict()
 
@@ -2217,7 +2213,6 @@ def test_initializer_not_reused(n_jobs):
     # own initializer args, independently of the previous calls, hence the loky workers
     # are not reused.
     n_repetitions = 3
-    n_jobs = effective_n_jobs(n_jobs)
     manager = mp.Manager()
 
     pids = set()