Repository: joblib/joblib
Pull request #1458 (2023-06-22 09:54:11)

`return_generator={True,False}` -> `return_as={'list','generator'}`
Change the boolean `return_generator` keyword, to `return_as` that is expected to take values in `{'list','submitted'}`, to anticipate a future `'completed'` keyword when implementing https://github.com/joblib/joblib/issues/1449 .

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/joblib/joblib/pull/1458?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) Report
Patch coverage: **`84.61`**% and project coverage change: **`-25.83`** :warning:
> Comparison is base [(`b469022`)](https://app.codecov.io/gh/joblib/joblib/commit/b46902217ebf071cddd5076826f8fcce06aecbec?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) 94.73% compared to head [(`e4ffb3f`)](https://app.codecov.io/gh/joblib/joblib/pull/1458?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) 68.90%.

<details><summary>Additional details and impacted files</summary>


```diff
@@             Coverage Diff             @@
##           master    #1458       +/-   ##
===========================================
- Coverage   94.73%   68.90%   -25.83%     
===========================================
  Files          44       44               
  Lines        7439     7235      -204     
===========================================
- Hits         7047     4985     -2062     
- Misses        392     2250     +1858     
```


| [Impacted Files](https://app.codecov.io/gh/joblib/joblib/pull/1458?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib) | Coverage Δ | |
|---|---|---|
| [joblib/parallel.py](https://app.codecov.io/gh/joblib/joblib/pull/1458?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3BhcmFsbGVsLnB5) | `54.92% <83.33%> (-41.18%)` | :arrow_down: |
| [joblib/test/test\_parallel.py](https://app.codecov.io/gh/joblib/joblib/pull/1458?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib#diff-am9ibGliL3Rlc3QvdGVzdF9wYXJhbGxlbC5weQ==) | `53.57% <85.71%> (-42.53%)` | :arrow_down: |

... and [33 files with indirect coverage changes](https://app.codecov.io/gh/joblib/joblib/pull/1458/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib)


</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/joblib/joblib/pull/1458?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=joblib).

betatim: Came here via the sprint/Franck at the sprint.

How about `Parallel(..., results=...)`, with possible values: `immediate` for "return results in any order as soon as they are ready", `ordered` for "return results in the order in which tasks were submitted" and `complete` for "return all results in order in which tasks were submitted when all tasks are ready". Could also use "unordered", "ordered" and "all" (or something?). I think main my comment is that I prefer `results=` over `return_as=`.
rth: As @adrinjalali mentioned it could be worth at least considering adding an extra method for different output type. Something like.
```py
joblib.Parallel(...).call_as_generator(delayed(sqrt)(i**2) for i in range(10)))
```

It also feels that something that returns a variable type depending on the input parameter is not great API wise, and would confuse type checkers. I mean between list and iterators of results it could still work. But if you want to add an iterator of future results later that's a completely different return type. 
ogrisel: - `return_type="list"/"generator"` (or `return_as`?)
- `return_order="as_submitted"/"as_completed"` (or `collection_order`?)

with `raise ValueError` when `return_as="list"` and `collection_order="as_completed"` because it's useless.

At least it's very explicit. It's a bit verbose but I think I prefer this side of the tradeoff.
ogrisel: >  But if you want to add an iterator of future results later that's a completely different return type.

If we ever go into returning "future" or "promise" objects I think we should introduce a completely new API (and probably mimic that of `concurrent.future` / `dask`).


fcharras: Some more online discussion later, we converge on `return_as=list/generator/unordered_generator`.

Let me update the PR in this direction now.

(About futures/promises, returning such objects is definitely not in the scope of what Parallel offers, the misunderstanding comes from bad wording from me early on, sorry about that.)

ogrisel: After discussion at the scikit-learn sprint, I prefer the following:

- `return_as="list"/"generator"/"unordered_generator"`

This is explicit enough, technically correct and concise enough. 


Patch:
diff --git a/CHANGES.rst b/CHANGES.rst
index 35bc65d..638592f 100644
--- a/CHANGES.rst
+++ b/CHANGES.rst
@@ -36,9 +36,10 @@ In development
   custom cache invalidation based on the metadata of the function call.
   https://github.com/joblib/joblib/pull/1149
 
-- Add a ``return_generator`` parameter for ``Parallel``, that allows
-  to consume results asynchronously.
+- Add a ``return_as`` parameter for ``Parallel``, that enables consuming
+  results asynchronously.
   https://github.com/joblib/joblib/pull/1393
+  https://github.com/joblib/joblib/pull/1458
 
 - Improve the behavior of ``joblib`` for ``n_jobs=1``, with simplified
   tracebacks and more efficient running time.
@@ -1343,4 +1344,4 @@ Gael Varoquaux
     BUG: Make sure that joblib still works with Python2.5. => release 0.4.2
 
 Release 0.4.1
-----------------
+----------------
\ No newline at end of file
diff --git a/doc/parallel.rst b/doc/parallel.rst
index 7da9fce..1fa238d 100644
--- a/doc/parallel.rst
+++ b/doc/parallel.rst
@@ -25,11 +25,11 @@ can be spread over 2 CPUs using the following::
 
 The output can be a generator that yields the results as soon as they're
 available, even if the subsequent tasks aren't completed yet. The order
-of the outputs always matches the order of the inputs::
+of the outputs always matches the order the inputs have been submitted with::
 
     >>> from math import sqrt
     >>> from joblib import Parallel, delayed
-    >>> parallel = Parallel(n_jobs=2, return_generator=True)
+    >>> parallel = Parallel(n_jobs=2, return_as="generator")
     >>> output_generator = parallel(delayed(sqrt)(i ** 2) for i in range(10))
     >>> print(type(output_generator))
     <class 'generator'>
@@ -40,9 +40,17 @@ of the outputs always matches the order of the inputs::
     >>> print(list(output_generator))
     [2.0, 3.0, 4.0, 5.0, 6.0, 7.0, 8.0, 9.0]
 
-This generator allows to reduce the memory footprint of :class:`joblib.Parallel`
-calls in case the results can benefit from on-the-fly aggregation, as illustrated
-in :ref:`sphx_glr_auto_examples_parallel_generator.py`.
+This generator enables reducing the memory footprint of
+:class:`joblib.Parallel` calls in case the results can benefit from on-the-fly
+aggregation, as illustrated in
+:ref:`sphx_glr_auto_examples_parallel_generator.py`.
+
+Future releases are planned to also support returning a generator that yields
+the results in the order of completion rather than the order of submission, by
+using ``return_as="unordered_generator"`` instead of ``return_as="generator"``.
+In this case the order of the outputs will depend on the concurrency of workers
+and will not be guaranteed to be deterministic, meaning the results can be
+yielded with a different order every time the code is executed.
 
 Thread-based parallelism vs process-based parallelism
 =====================================================
@@ -410,4 +418,4 @@ does not exist (but multiprocessing has more overhead).
 
 .. autoclass:: joblib.parallel.ParallelBackendBase
 
-.. autoclass:: joblib.parallel.AutoBatchingMixin
+.. autoclass:: joblib.parallel.AutoBatchingMixin
\ No newline at end of file
diff --git a/examples/parallel_generator.py b/examples/parallel_generator.py
index 43b84ea..286518f 100644
--- a/examples/parallel_generator.py
+++ b/examples/parallel_generator.py
@@ -10,8 +10,14 @@ If we call :class:`~joblib.Parallel` for several of these tasks directly, we
 observe a high memory usage, as all the results are held in RAM before being
 processed
 
-Using the ``return_generator=True`` option allows to progressively consumes
-the outputs as they arrive and keeps the memory at an acceptable level.
+Using ``return_as='generator'`` allows to progressively consume the outputs
+as they arrive and keeps the memory at an acceptable level.
+
+In this case, the output of the `Parallel` call is a generator that yields the
+results in the order the tasks have been submitted with. Future releases are
+also planned to support the ``return_as="unordered_generator"`` parameter to
+have the generator yield results as soon as available.
+
 """
 
 ##############################################################################
@@ -93,16 +99,16 @@ def accumulator_sum(generator):
 
 
 ##############################################################################
-# We process many of the tasks in parallel. If `return_generator=False`
-# (default), we should expect a usage of more than 2GB in RAM. Indeed, all the
-# results are computed and stored in ``res`` before being processed by
+# We process many of the tasks in parallel. If ``return_as="list"`` (default),
+# we should expect a usage of more than 2GB in RAM. Indeed, all the results
+# are computed and stored in ``res`` before being processed by
 # `accumulator_sum` and collected by the gc.
 
 from joblib import Parallel, delayed
 
 monitor = MemoryMonitor()
-print("Running tasks with return_generator=False...")
-res = Parallel(n_jobs=2, return_generator=False)(
+print("Running tasks with return_as='list'...")
+res = Parallel(n_jobs=2, return_as="list")(
     delayed(return_big_object)(i) for i in range(150)
 )
 print("Accumulate results:", end='')
@@ -117,14 +123,14 @@ print(f"Peak memory usage: {peak:.2f}GB")
 
 
 ##############################################################################
-# If we use ``return_generator=True``, ``res`` is simply a generator with the
+# If we use ``return_as="generator"``, ``res`` is simply a generator on the
 # results that are ready. Here we consume the results as soon as they arrive
 # with the ``accumulator_sum`` and once they have been used, they are collected
 # by the gc. The memory footprint is thus reduced, typically around 300MB.
 
 monitor_gen = MemoryMonitor()
-print("Create result generator with return_generator=True...")
-res = Parallel(n_jobs=2, return_generator=True)(
+print("Create result generator with return_as='generator'...")
+res = Parallel(n_jobs=2, return_as="generator")(
     delayed(return_big_object)(i) for i in range(150)
 )
 print("Accumulate results:", end='')
@@ -152,11 +158,11 @@ print(f"Peak memory usage: {peak:.2f}MB")
 import matplotlib.pyplot as plt
 plt.semilogy(
     np.maximum.accumulate(monitor.memory_buffer),
-    label='return_generator=False'
+    label='return_as="list"'
 )
 plt.semilogy(
     np.maximum.accumulate(monitor_gen.memory_buffer),
-    label='return_generator=True'
+    label='return_as="generator"'
 )
 plt.xlabel("Time")
 plt.xticks([], [])
@@ -166,7 +172,7 @@ plt.legend()
 plt.show()
 
 ##############################################################################
-# It is important to note that with ``return_generator``, the results are
+# It is important to note that with ``return_as="generator"``, the results are
 # still accumulated in RAM after computation. But as we asynchronously process
 # them, they can be freed sooner. However, if the generator is not consumed
 # the memory still grows linearly.
diff --git a/joblib/_parallel_backends.py b/joblib/_parallel_backends.py
index 22f1345..1852b0d 100644
--- a/joblib/_parallel_backends.py
+++ b/joblib/_parallel_backends.py
@@ -117,7 +117,7 @@ class ParallelBackendBase(metaclass=ABCMeta):
     def abort_everything(self, ensure_ready=True):
         """Abort any running tasks
 
-        This is called when an exception has been raised when executing a tasks
+        This is called when an exception has been raised when executing a task
         and all the remaining tasks will be ignored and can therefore be
         aborted to spare computation resources.
 
@@ -609,7 +609,7 @@ class LokyBackend(AutoBatchingMixin, ParallelBackendBase):
             raise RuntimeError(
                 "The executor underlying Parallel has been shutdown. "
                 "This is likely due to the garbage collection of a previous "
-                "generator from a call to Parallel with return_generator=True."
+                "generator from a call to Parallel with return_as='generator'."
                 " Make sure the generator is not garbage collected when "
                 "submitting a new job or that it is first properly exhausted."
             )
diff --git a/joblib/parallel.py b/joblib/parallel.py
index 4149f02..dd7d91e 100644
--- a/joblib/parallel.py
+++ b/joblib/parallel.py
@@ -584,7 +584,7 @@ class BatchedCalls(object):
     def __call__(self):
         # Set the default nested backend to self._backend but do not set the
         # change the default number of processes to -1
-        with parallel_config(self._backend, n_jobs=self._n_jobs):
+        with parallel_config(backend=self._backend, n_jobs=self._n_jobs):
             return [func(*args, **kwargs)
                     for func, args, kwargs in self.items]
 
@@ -975,11 +975,15 @@ class Parallel(Logger):
             soft hints (prefer) or hard constraints (require) so as to make it
             possible for library users to change the backend from the outside
             using the :func:`~parallel_config` context manager.
-        return_generator: bool
-            If True, calls to this instance will return a generator, yielding
-            the results as soon as they are available, in the original order.
-            Note that the intended usage is to run one call at a time. Multiple
-            calls to the same Parallel object will result in a ``RuntimeError``
+        return_as: str in {'list', 'generator'}, default: 'list'
+            If 'list', calls to this instance will return a list, only when
+            all results have been processed and retrieved.
+            If 'generator', it will return a generator that yields the results
+            as soon as they are available, in the order the tasks have been
+            submitted with.
+            Future releases are planned to also support 'generator_unordered',
+            in which case the generator immediately yields available results
+            independently of the submission order.
         prefer: str in {'processes', 'threads'} or None, default: None
             Soft hint to choose the default backend if no specific backend
             was selected with the :func:`~parallel_config` context manager.
@@ -1074,6 +1078,9 @@ class Parallel(Logger):
         * Ability to use shared memory efficiently with worker
           processes for large numpy-based datastructures.
 
+        Note that the intended usage is to run one call at a time. Multiple
+        calls to the same Parallel object will result in a ``RuntimeError``
+
         Examples
         --------
 
@@ -1169,7 +1176,7 @@ class Parallel(Logger):
         self,
         n_jobs=default_parallel_config["n_jobs"],
         backend=default_parallel_config['backend'],
-        return_generator=False,
+        return_as="list",
         verbose=default_parallel_config["verbose"],
         timeout=None,
         pre_dispatch='2 * n_jobs',
@@ -1189,7 +1196,14 @@ class Parallel(Logger):
         self.verbose = _get_config_param(verbose, context_config, "verbose")
         self.timeout = timeout
         self.pre_dispatch = pre_dispatch
-        self.return_generator = return_generator
+
+        if return_as not in {"list", "generator"}:
+            raise ValueError(
+                'Expected `return_as` parameter to be a string equal to "list"'
+                f' or "generator", but got {return_as} instead'
+            )
+        self.return_as = return_as
+        self.return_generator = return_as != "list"
 
         # Check if we are under a parallel_config or parallel_backend
         # context manager and use the config from the context manager
@@ -1272,10 +1286,10 @@ class Parallel(Logger):
                 % batch_size)
 
         if not isinstance(backend, SequentialBackend):
-            if return_generator and not backend.supports_return_generator:
+            if self.return_generator and not backend.supports_return_generator:
                 raise ValueError(
                     "Backend {} does not support "
-                    "return_generator=True".format(backend)
+                    "return_as={}".format(backend, return_as)
                 )
             # This lock is used to coordinate the main thread of this process
             # with the async callback thread of our the pool.
@@ -1391,7 +1405,7 @@ class Parallel(Logger):
         batch_size = self._get_batch_size()
 
         with self._lock:
-            # to ensure an even distribution of the workolad between workers,
+            # to ensure an even distribution of the workload between workers,
             # we look ahead in the original iterators more than batch_size
             # tasks - However, we keep consuming only one batch at each
             # dispatch_one_batch call. The extra tasks are stored in a local
@@ -1670,7 +1684,7 @@ class Parallel(Logger):
         while self._wait_retrieval():
 
             # If the callback thread of a worker has signaled that its task
-            # triggerd an exception, or if the retrieval loop has raised an
+            # triggered an exception, or if the retrieval loop has raised an
             # exception (e.g. `GeneratorExit`), exit the loop and surface the
             # worker traceback.
             if self._aborting:


Test patch:
diff --git a/joblib/test/test_parallel.py b/joblib/test/test_parallel.py
index c4770d2..9a7ad15 100644
--- a/joblib/test/test_parallel.py
+++ b/joblib/test/test_parallel.py
@@ -1213,12 +1213,12 @@ def set_list_value(input_list, index, value):
 
 
 @pytest.mark.parametrize('n_jobs', [1, 2, 4])
-def test_parallel_return_generator_order(n_jobs):
+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_generator=True,
+    result = Parallel(n_jobs=n_jobs, return_as="generator",
                       backend='threading')(
         delayed(set_list_value)(input_list, i, i) for i in range(5))
 
@@ -1253,7 +1253,7 @@ def test_deadlock_with_generator(backend, 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_generator=True) as parallel:
+                  return_as="generator") as parallel:
         result = parallel(delayed(get_large_object)(i) for i in range(10))
         next(result)
         next(result)
@@ -1271,7 +1271,7 @@ def test_multiple_generator_call(backend, n_jobs):
     # 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_generator=True)
+        parallel = Parallel(n_jobs, backend=backend, return_as="generator")
         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
@@ -1295,7 +1295,7 @@ def test_multiple_generator_call_managed(backend, n_jobs):
     # 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_generator=True) as parallel:
+                  return_as="generator") as parallel:
         g = parallel(delayed(sleep)(10) for _ in range(10))  # noqa: F841
         t_start = time.time()
         with raises(RuntimeError,
@@ -1318,10 +1318,10 @@ def test_multiple_generator_call_managed(backend, n_jobs):
 @parametrize('n_jobs', [1, 2, -2, -1])
 def test_multiple_generator_call_separated(backend, n_jobs):
     # Check that for separated Parallel, both tasks are correctly returned.
-    g = Parallel(n_jobs, backend=backend, return_generator=True)(
+    g = Parallel(n_jobs, backend=backend, return_as="generator")(
         delayed(sqrt)(i ** 2) for i in range(10)
     )
-    g2 = Parallel(n_jobs, backend=backend, return_generator=True)(
+    g2 = Parallel(n_jobs, backend=backend, return_as="generator")(
         delayed(sqrt)(i ** 2) for i in range(10, 20)
     )
 
@@ -1341,7 +1341,7 @@ def test_multiple_generator_call_separated_gc(backend, error):
 
     # Check that in loky, only one call can be run at a time with
     # a single executor.
-    parallel = Parallel(2, backend=backend, return_generator=True)
+    parallel = Parallel(2, backend=backend, return_as="generator")
     g = parallel(delayed(sleep)(10) for i in range(10))
     g_wr = weakref.finalize(g, lambda: print("Generator collected"))
     ctx = (
@@ -1354,7 +1354,7 @@ def test_multiple_generator_call_separated_gc(backend, error):
         # 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_generator=True)(
+        g = Parallel(2, backend=backend, return_as="generator")(
             delayed(sqrt)(i ** 2) for i in range(10, 20)
         )