✨ Introduce maxtasksperchild to increase stability (#29)

Author: ddelange

README.md

@@ -11,7 +11,7 @@
 
 ### mapply vs. pandarallel vs. swifter
 
-Where [`pandarallel`](https://github.com/nalepae/pandarallel) only requires [`dill`](https://github.com/uqfoundation/dill) (and therefore has to rely on in-house multiprocessing and progressbars), [`swifter`](https://github.com/jmcarpenter2/swifter) relies on the heavy [`dask`](https://github.com/dask/dask) framework, converting to Dask DataFrames and back. In an attempt to find the golden mean, `mapply` is highly customizable and remains lightweight, leveraging the powerful [`pathos`](https://github.com/uqfoundation/pathos) framework, which shadows Python's built-in multiprocessing module using `dill` for universal pickling.
+Where [`pandarallel`](https://github.com/nalepae/pandarallel) only requires [`dill`](https://github.com/uqfoundation/dill) (and therefore has to rely on in-house multiprocessing and progressbars), [`swifter`](https://github.com/jmcarpenter2/swifter) relies on the heavy [`dask`](https://github.com/dask/dask) framework, converting to Dask DataFrames and back. In an attempt to find the golden mean, `mapply` is highly customizable and remains lightweight, leveraging [`tqdm`](https://github.com/tqdm/tqdm) and [`multiprocess`](https://github.com/uqfoundation/multiprocess), which shadows Python's built-in multiprocessing module using [`dill`](https://github.com/uqfoundation/dill) for universal pickling.
 
 
 ## Installation

docs/source/conf.py

@@ -181,6 +181,6 @@ def linkcode_resolve(  # noqa:CCR001
 intersphinx_mapping = {
     "python": ("https://docs.python.org/3", None),
     "pandas": ("https://pandas.pydata.org/pandas-docs/stable", None),
-    "pathos": ("https://pathos.readthedocs.io/en/latest", None),
+    "multiprocess": ("https://multiprocess.readthedocs.io/en/latest/", None),
     # "tqdm": ("https://tqdm.github.io/docs/tqdm", None),  # mkdocs not working
 }

requirements/prod.txt

@@ -1,3 +1,3 @@
-pathos>=0.2.0
+multiprocess
 psutil
 tqdm>=4.27  # from tqdm.auto import tqdm

src/mapply/parallel.py

@@ -1,4 +1,4 @@
-"""Submodule containing code to distribute computation over multiple processes using :class:`pathos.multiprocessing.ProcessPool`.
+"""Submodule containing code to distribute computation over multiple processes using :class:`multiprocess.pool.Pool`.
 
 Standalone usage:
 ::
@@ -19,11 +19,12 @@ def some_heavy_computation(x, power):
     )
 """
 import logging
+import os
 from functools import partial
 from typing import Any, Callable, Iterable, Iterator, Optional
 
 import psutil
-from pathos.multiprocessing import ProcessPool
+from multiprocess.pool import Pool
 from tqdm.auto import tqdm as _tqdm
 
 logger = logging.getLogger(__name__)
@@ -37,6 +38,7 @@ def sensible_cpu_count() -> int:
 
 
 N_CORES = sensible_cpu_count()
+MAX_TASKS_PER_CHILD = os.environ.get("MAPPLY_MAX_TASKS_PER_CHILD", 4)
 
 
 def _choose_n_workers(n_chunks: Optional[int], n_workers: int) -> int:
@@ -78,10 +80,6 @@ def multiprocessing_imap(
 
     Yields:
         Results in same order as input iterable.
-
-    Raises:
-        Exception: Any error occurred during computation (will terminate the pool early).
-        KeyboardInterrupt: Any KeyboardInterrupt sent by the user (will terminate the pool early).
     """
     n_chunks: Optional[int] = tqdm(iterable, disable=True).__len__()  # doesn't exhaust
     func = partial(func, *args, **kwargs)
@@ -94,7 +92,7 @@ def multiprocessing_imap(
         stage = map(func, iterable)
     else:
         logger.debug("Starting ProcessPool with %d workers", n_workers)
-        pool = ProcessPool(n_workers)
+        pool = Pool(n_workers, maxtasksperchild=MAX_TASKS_PER_CHILD)
 
         stage = pool.imap(func, iterable)
 
@@ -103,12 +101,7 @@ def multiprocessing_imap(
 
     try:
         yield from stage
-    except (Exception, KeyboardInterrupt):
+    finally:
         if pool:
             logger.debug("Terminating ProcessPool")
             pool.terminate()
-        raise
-    finally:
-        if pool:
-            logger.debug("Closing ProcessPool")
-            pool.clear()