fixes #102

Author: jph00

fastcore/__init__.py

@@ -1 +1 @@
-__version__ = "1.0.15"
+__version__ = "1.0.16"

fastcore/_nbdev.py

@@ -120,6 +120,7 @@
          "ContextManagers": "02_utils.ipynb",
          "set_num_threads": "02_utils.ipynb",
          "ProcessPoolExecutor": "02_utils.ipynb",
+         "ThreadPoolExecutor": "02_utils.ipynb",
          "parallel": "02_utils.ipynb",
          "run_procs": "02_utils.ipynb",
          "parallel_gen": "02_utils.ipynb",

fastcore/all.py

@@ -7,3 +7,4 @@
 from .meta import *
 from .imports import *
 from .logargs import *
+from .script import *

fastcore/utils.py

@@ -9,7 +9,7 @@
            'partialler', 'mapped', 'instantiate', 'using_attr', 'Self', 'Self', 'remove_patches_path', 'bunzip',
            'join_path_file', 'urlread', 'urljson', 'run_proc', 'do_request', 'sort_by_run', 'PrettyString',
            'round_multiple', 'even_mults', 'num_cpus', 'add_props', 'ContextManagers', 'set_num_threads',
-           'ProcessPoolExecutor', 'parallel', 'run_procs', 'parallel_gen']
+           'ProcessPoolExecutor', 'ThreadPoolExecutor', 'parallel', 'run_procs', 'parallel_gen']
 
 # Cell
 from .imports import *
@@ -694,16 +694,35 @@ def map(self, f, items, timeout=None, chunksize=1, *args, **kwargs):
         try: return super().map(_g, items, timeout=timeout, chunksize=chunksize)
         except Exception as e: self.on_exc(e)
 
+# Cell
+class ThreadPoolExecutor(concurrent.futures.ThreadPoolExecutor):
+    "Same as Python's ThreadPoolExecutor, except can pass `max_workers==0` for serial execution"
+    def __init__(self, max_workers=defaults.cpus, on_exc=print, pause=0, **kwargs):
+        if max_workers is None: max_workers=defaults.cpus
+        store_attr()
+        self.not_parallel = max_workers==0
+        if self.not_parallel: max_workers=1
+        super().__init__(max_workers, **kwargs)
+
+    def map(self, f, items, timeout=None, chunksize=1, *args, **kwargs):
+        self.lock = Manager().Lock()
+        g = partial(f, *args, **kwargs)
+        if self.not_parallel: return map(g, items)
+        _g = partial(_call, self.lock, self.pause, self.max_workers, g)
+        try: return super().map(_g, items, timeout=timeout, chunksize=chunksize)
+        except Exception as e: self.on_exc(e)
+
 # Cell
 try: from fastprogress import progress_bar
 except: progress_bar = None
 
 # Cell
 def parallel(f, items, *args, n_workers=defaults.cpus, total=None, progress=None, pause=0,
-             timeout=None, chunksize=1, **kwargs):
+             threadpool=False, timeout=None, chunksize=1, **kwargs):
     "Applies `func` in parallel to `items`, using `n_workers`"
     if progress is None: progress = progress_bar is not None
-    with ProcessPoolExecutor(n_workers, pause=pause) as ex:
+    pool = ThreadPoolExecutor if threadpool else ProcessPoolExecutor
+    with pool(n_workers, pause=pause) as ex:
         r = ex.map(f,items, *args, timeout=timeout, chunksize=chunksize, **kwargs)
         if progress:
             if total is None: total = len(items)

nbs/02_utils.ipynb

@@ -294,7 +294,7 @@
     {
      "data": {
       "text/plain": [
-       "<__main__._t at 0x7efe2eb5ef90>"
+       "<__main__._t at 0x7f756accd110>"
       ]
      },
      "execution_count": null,
@@ -1668,7 +1668,7 @@
     {
      "data": {
       "text/plain": [
-       "['a', 'e', 'b', 'h', 'c', 'd', 'g', 'f']"
+       "['c', 'a', 'f', 'e', 'g', 'h', 'd', 'b']"
       ]
      },
      "execution_count": null,
@@ -3570,6 +3570,31 @@
     "        except Exception as e: self.on_exc(e)"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#export\n",
+    "class ThreadPoolExecutor(concurrent.futures.ThreadPoolExecutor):\n",
+    "    \"Same as Python's ThreadPoolExecutor, except can pass `max_workers==0` for serial execution\"\n",
+    "    def __init__(self, max_workers=defaults.cpus, on_exc=print, pause=0, **kwargs):\n",
+    "        if max_workers is None: max_workers=defaults.cpus\n",
+    "        store_attr()\n",
+    "        self.not_parallel = max_workers==0\n",
+    "        if self.not_parallel: max_workers=1\n",
+    "        super().__init__(max_workers, **kwargs)\n",
+    "\n",
+    "    def map(self, f, items, timeout=None, chunksize=1, *args, **kwargs):\n",
+    "        self.lock = Manager().Lock()\n",
+    "        g = partial(f, *args, **kwargs)\n",
+    "        if self.not_parallel: return map(g, items)\n",
+    "        _g = partial(_call, self.lock, self.pause, self.max_workers, g)\n",
+    "        try: return super().map(_g, items, timeout=timeout, chunksize=chunksize)\n",
+    "        except Exception as e: self.on_exc(e)"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -3580,7 +3605,7 @@
       "text/markdown": [
        "<h4 id=\"ProcessPoolExecutor\" class=\"doc_header\"><code>class</code> <code>ProcessPoolExecutor</code><a href=\"\" class=\"source_link\" style=\"float:right\">[source]</a></h4>\n",
        "\n",
-       "> <code>ProcessPoolExecutor</code>(**`max_workers`**=*`64`*, **`on_exc`**=*`print`*, **`pause`**=*`0`*, **`mp_context`**=*`None`*, **`initializer`**=*`None`*, **`initargs`**=*`()`*) :: [`ProcessPoolExecutor`](/utils.html#ProcessPoolExecutor)\n",
+       "> <code>ProcessPoolExecutor</code>(**`max_workers`**=*`64`*, **`on_exc`**=*`print`*, **`pause`**=*`0`*, **\\*\\*`kwargs`**) :: [`ProcessPoolExecutor`](/utils.html#ProcessPoolExecutor)\n",
        "\n",
        "Same as Python's ProcessPoolExecutor, except can pass `max_workers==0` for serial execution"
       ],
@@ -3615,10 +3640,11 @@
    "source": [
     "#export \n",
     "def parallel(f, items, *args, n_workers=defaults.cpus, total=None, progress=None, pause=0,\n",
-    "             timeout=None, chunksize=1, **kwargs):\n",
+    "             threadpool=False, timeout=None, chunksize=1, **kwargs):\n",
     "    \"Applies `func` in parallel to `items`, using `n_workers`\"\n",
     "    if progress is None: progress = progress_bar is not None\n",
-    "    with ProcessPoolExecutor(n_workers, pause=pause) as ex:\n",
+    "    pool = ThreadPoolExecutor if threadpool else ProcessPoolExecutor\n",
+    "    with pool(n_workers, pause=pause) as ex:\n",
     "        r = ex.map(f,items, *args, timeout=timeout, chunksize=chunksize, **kwargs)\n",
     "        if progress:\n",
     "            if total is None: total = len(items)\n",
@@ -3651,16 +3677,6 @@
      "metadata": {},
      "output_type": "display_data"
     },
-    {
-     "data": {
-      "text/html": [],
-      "text/plain": [
-       "<IPython.core.display.HTML object>"
-      ]
-     },
-     "metadata": {},
-     "output_type": "display_data"
-    },
     {
      "data": {
       "text/html": [],
@@ -3678,7 +3694,8 @@
     "    return x+a\n",
     "\n",
     "inp,exp = range(50),range(1,51)\n",
-    "test_eq(parallel(add_one, inp, n_workers=2), exp)\n",
+    "test_eq(parallel(add_one, inp, n_workers=2, progress=False), exp)\n",
+    "test_eq(parallel(add_one, inp, threadpool=True, n_workers=2, progress=False), exp)\n",
     "test_eq(parallel(add_one, inp, n_workers=0), exp)\n",
     "test_eq(parallel(add_one, inp, n_workers=1, a=2), range(2,52))\n",
     "test_eq(parallel(add_one, inp, n_workers=0, a=2), range(2,52))"
@@ -3688,7 +3705,7 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "Use the `pause` parameter to ensure a pause of `pause` seconds between processes starting. This is in case there are race conditions in starting some process, or to stagger the time each process starts, for example when making many requests to a webserver."
+    "Use the `pause` parameter to ensure a pause of `pause` seconds between processes starting. This is in case there are race conditions in starting some process, or to stagger the time each process starts, for example when making many requests to a webserver. Set `threadpool=True` to use `ThreadPoolExecutor` instead of `ProcessPoolExecutor`."
    ]
   },
   {

settings.ini

@@ -7,7 +7,7 @@ author = Jeremy Howard and Sylvain Gugger
 author_email = [email protected]
 copyright = fast.ai
 branch = master
-version = 1.0.15
+version = 1.0.16
 min_python = 3.6
 audience = Developers
 language = English