Release v0.0.4 and add loky as parallel backend. (#4)

Author: tobiasraabe

Diff for: .conda/meta.yaml

@@ -26,6 +26,7 @@ requirements:
 test:
   requires:
     - pytest
+    - loky
   source_files:
     - tox.ini
     - tests
@@ -41,4 +42,4 @@ about:
   home: https://github.com/pytask-dev/pytask-parallel
   license: none
   summary: Parallelize the execution of tasks.
-  dev_url: https://github.com/pytask-dev/pytask-parallel/
+  dev_url: https://github.com/pytask-dev/pytask-parallel

Diff for: CHANGES.rst

@@ -3,7 +3,14 @@ Changes
 
 This is a record of all past pytask-parallel releases and what went into them in reverse
 chronological order. Releases follow `semantic versioning <https://semver.org/>`_ and
-all releases are available on `Anaconda.org <https://anaconda.org/pytask/pytask-parallel>`_.
+all releases are available on `Anaconda.org
+<https://anaconda.org/pytask/pytask-parallel>`_.
+
+
+0.0.4 - 2020-10-30
+------------------
+
+- :gh:`4` implement an executor with ``loky``.
 
 
 0.0.3 - 2020-09-12

Diff for: README.rst

@@ -21,21 +21,38 @@ pytask-parallel
 Parallelize the execution of tasks with `pytask-parallel` which is a plugin for `pytask
 <https://github.com/pytask-dev/pytask>`_.
 
+
+Installation
+------------
+
 Install the plugin via ``conda`` with
 
-.. code-block:: bash
+.. code-block:: console
 
     $ conda config --add channels conda-forge --add channels pytask
     $ conda install pytask-parallel
 
 The plugin uses the ``ProcessPoolExecutor`` or ``ThreadPoolExecutor`` in the
 `concurrent.futures <https://docs.python.org/3/library/concurrent.futures.html>`_ module
-to execute tasks asynchronously.
+to execute tasks asynchronously. By default, processes are used for parallelization.
+
+It is also possible to install ``loky`` with
+
+.. code-block:: console
+
+    $ conda install -c conda-forge loky
+
+which is a more robust implementation of the ``ProcessPoolExecutor`` and the default
+backend if installed.
+
+
+Usage
+-----
 
 To parallelize your tasks across many workers, pass an integer greater than 1 or
 ``'auto'`` to the command-line interface.
 
-.. code-block:: bash
+.. code-block:: console
 
     $ pytask -n 2
     $ pytask --n-workers 2
@@ -44,15 +61,14 @@ To parallelize your tasks across many workers, pass an integer greater than 1 or
     $ pytask -n auto
 
 
-By default, processes are used to parallelize the execution of tasks. This is useful for
-CPU bound tasks such as numerical computations. (`Here
-<https://stackoverflow.com/a/868577/7523785>`_ is an explanation on what CPU or IO bound
-means.)
+Using processes to parallelize the execution of tasks is useful for CPU bound tasks such
+as numerical computations. (`Here <https://stackoverflow.com/a/868577/7523785>`_ is an
+explanation on what CPU or IO bound means.)
 
 For IO bound tasks, tasks where the limiting factor are network responses, accesses to
 files, you can parallelize via threads.
 
-.. code-block:: bash
+.. code-block:: console
 
     $ pytask --parallel-backend threads
 
@@ -65,7 +81,7 @@ You can also set the options in one of the configuration files (``pytask.ini``,
 
     [pytask]
     n_processes = 1
-    parallel_backend = processes
+    parallel_backend = processes  # or loky if installed.
 
 
 Changes

Diff for: environment.yml

@@ -14,8 +14,10 @@ dependencies:
   # Package dependencies
   - pytask >= 0.0.6
   - cloudpickle
+  - loky
 
   # Misc
+  - black
   - bump2version
   - jupyterlab
   - matplotlib

Diff for: setup.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.0.3
+current_version = 0.0.4
 parse = (?P<major>\d+)\.(?P<minor>\d+)(\.(?P<patch>\d+))(\-?((dev)?(?P<dev>\d+))?)
 serialize = 
 	{major}.{minor}.{patch}dev{dev}

Diff for: setup.py

@@ -3,7 +3,7 @@
 
 setup(
     name="pytask-parallel",
-    version="0.0.3",
+    version="0.0.4",
     packages=find_packages(where="src"),
     package_dir={"": "src"},
     entry_points={"pytask": ["pytask_parallel = pytask_parallel.plugin"]},

Diff for: src/pytask_parallel/__init__.py

@@ -1 +1 @@
-__version__ = "0.0.3"
+__version__ = "0.0.4"

Diff for: src/pytask_parallel/backends.py

@@ -0,0 +1,19 @@
+"""This module configures the available backends."""
+from concurrent.futures import ProcessPoolExecutor
+from concurrent.futures import ThreadPoolExecutor
+
+
+PARALLEL_BACKENDS = {
+    "processes": ProcessPoolExecutor,
+    "threads": ThreadPoolExecutor,
+}
+
+PARALLEL_BACKENDS_DEFAULT = "processes"
+
+try:
+    from loky import get_reusable_executor
+
+    PARALLEL_BACKENDS["loky"] = get_reusable_executor
+    PARALLEL_BACKENDS_DEFAULT = "loky"
+except ImportError:
+    pass

Diff for: src/pytask_parallel/build.py

@@ -1,6 +1,8 @@
 """Extend the build command."""
 import click
 from _pytask.config import hookimpl
+from pytask_parallel.backends import PARALLEL_BACKENDS
+from pytask_parallel.backends import PARALLEL_BACKENDS_DEFAULT
 
 
 @hookimpl
@@ -18,8 +20,11 @@ def pytask_extend_command_line_interface(cli):
         ),
         click.Option(
             ["--parallel-backend"],
-            type=click.Choice(["processes", "threads"]),
-            help="Backend for the parallelization.  [default: processes]",
+            type=click.Choice(PARALLEL_BACKENDS),
+            help=(
+                "Backend for the parallelization.  "
+                f"[default: {PARALLEL_BACKENDS_DEFAULT}]"
+            ),
             default=None,
         ),
         click.Option(

Diff for: src/pytask_parallel/callbacks.py

@@ -1,11 +1,12 @@
 """Validate command line inputs and configuration values."""
+from pytask_parallel.backends import PARALLEL_BACKENDS
 
 
 def n_workers_callback(value):
     """Validate the n-workers option."""
     if value == "auto":
         pass
-    elif value is None or value == "None":
+    elif value in [None, "None", "none"]:
         value = None
     elif isinstance(value, int) and 1 <= value:
         pass
@@ -19,16 +20,20 @@ def n_workers_callback(value):
 
 def parallel_backend_callback(value):
     """Validate the input for the parallel backend."""
-    if value == "None":
+    if value in [None, "None", "none"]:
         value = None
-    if value not in ["processes", "threads", None]:
-        raise ValueError("parallel_backend has to be 'processes' or 'threads'.")
+    elif value in PARALLEL_BACKENDS:
+        pass
+    else:
+        raise ValueError(
+            f"parallel_backend has to be one of {list(PARALLEL_BACKENDS)}."
+        )
     return value
 
 
 def delay_callback(value):
     """Validate the delay option."""
-    if value is None or value == "None":
+    if value in [None, "None", "none"]:
         value = None
     else:
         try:

Diff for: src/pytask_parallel/config.py

@@ -3,6 +3,7 @@
 
 from _pytask.config import hookimpl
 from _pytask.shared import get_first_non_none_value
+from pytask_parallel.backends import PARALLEL_BACKENDS_DEFAULT
 from pytask_parallel.callbacks import delay_callback
 from pytask_parallel.callbacks import n_workers_callback
 from pytask_parallel.callbacks import parallel_backend_callback
@@ -33,7 +34,7 @@ def pytask_parse_config(config, config_from_cli, config_from_file):
         config_from_cli,
         config_from_file,
         key="parallel_backend",
-        default="processes",
+        default=PARALLEL_BACKENDS_DEFAULT,
         callback=parallel_backend_callback,
     )
 

Diff for: src/pytask_parallel/execute.py

@@ -1,27 +1,22 @@
+"""Contains code relevant to the execution."""
 import sys
 import time
-from concurrent.futures import ProcessPoolExecutor
-from concurrent.futures import ThreadPoolExecutor
 
 import cloudpickle
 import networkx as nx
 from _pytask.config import hookimpl
 from _pytask.report import ExecutionReport
+from pytask_parallel.backends import PARALLEL_BACKENDS
 from pytask_parallel.scheduler import TopologicalSorter
 
-PARALLEL_BACKEND = {
-    "processes": ProcessPoolExecutor,
-    "threads": ThreadPoolExecutor,
-}
-
 
 @hookimpl
 def pytask_post_parse(config):
     """Register the parallel backend."""
     if config["parallel_backend"] == "processes":
         config["pm"].register(ProcessesNameSpace)
-    elif config["parallel_backend"] == "threads":
-        config["pm"].register(ThreadsNameSpace)
+    elif config["parallel_backend"] in ["threads", "loky"]:
+        config["pm"].register(DefaultBackendNameSpace)
 
 
 @hookimpl(tryfirst=True)
@@ -57,7 +52,7 @@ def pytask_execute_build(session):
         reports = []
         running_tasks = {}
 
-        parallel_backend = PARALLEL_BACKEND[session.config["parallel_backend"]]
+        parallel_backend = PARALLEL_BACKENDS[session.config["parallel_backend"]]
 
         with parallel_backend(max_workers=session.config["n_workers"]) as executor:
 
@@ -149,20 +144,20 @@ def pytask_execute_task(session, task):  # noqa: N805
 def unserialize_and_execute_task(bytes_):
     """Unserialize and execute task.
 
-    This function receives bytes and unpickles them to a task which is them execute in a
-    spawned process or thread.
+    This function receives bytes and unpickles them to a task which is them execute
+    in a spawned process or thread.
 
     """
     task = cloudpickle.loads(bytes_)
     task.execute()
 
 
-class ThreadsNameSpace:
+class DefaultBackendNameSpace:
     @hookimpl(tryfirst=True)
     def pytask_execute_task(session, task):  # noqa: N805
         """Execute a task.
 
-        Since threads share their memory, it is not necessary to pickle and unpickle the
+        Since threads have shared memory, it is not necessary to pickle and unpickle the
         task.
 
         """

Diff for: src/pytask_parallel/logging.py

@@ -0,0 +1,11 @@
+"""Contains code relevant to logging."""
+import click
+from _pytask.config import hookimpl
+
+
+@hookimpl(trylast=True)
+def pytask_log_session_header(session):
+    """Add a note for how many workers are spawned."""
+    n_workers = session.config["n_workers"]
+    if n_workers > 1:
+        click.echo(f"Started {n_workers} workers.")

Diff for: src/pytask_parallel/plugin.py

@@ -3,6 +3,7 @@
 from pytask_parallel import build
 from pytask_parallel import config
 from pytask_parallel import execute
+from pytask_parallel import logging
 
 
 @hookimpl
@@ -11,3 +12,4 @@ def pytask_add_hooks(pm):
     pm.register(build)
     pm.register(config)
     pm.register(execute)
+    pm.register(logging)

Diff for: tests/test_callbacks.py

@@ -1,6 +1,7 @@
 from contextlib import ExitStack as does_not_raise  # noqa: N813
 
 import pytest
+from pytask_parallel.backends import PARALLEL_BACKENDS
 from pytask_parallel.callbacks import delay_callback
 from pytask_parallel.callbacks import n_workers_callback
 from pytask_parallel.callbacks import parallel_backend_callback
@@ -17,6 +18,7 @@
         ("asdad", pytest.raises(ValueError)),
         (None, does_not_raise()),
         ("None", does_not_raise()),
+        ("none", does_not_raise()),
         ("1", does_not_raise()),
         ("1.1", pytest.raises(ValueError)),
     ],
@@ -30,13 +32,13 @@ def test_n_workers_callback(value, expectation):
 @pytest.mark.parametrize(
     "value, expectation",
     [
-        ("threads", does_not_raise()),
-        ("processes", does_not_raise()),
         (1, pytest.raises(ValueError)),
         ("asdad", pytest.raises(ValueError)),
         (None, does_not_raise()),
         ("None", does_not_raise()),
-    ],
+        ("none", does_not_raise()),
+    ]
+    + [(i, does_not_raise()) for i in PARALLEL_BACKENDS],
 )
 def test_parallel_backend_callback(value, expectation):
     with expectation:
@@ -53,6 +55,7 @@ def test_parallel_backend_callback(value, expectation):
         ("asdad", pytest.raises(ValueError)),
         (None, does_not_raise()),
         ("None", does_not_raise()),
+        ("none", does_not_raise()),
     ],
 )
 def test_delay_callback(value, expectation):