Merge pull request #289 from mszhanyi/zhanyi/parallel

update parallel

Author: jph00

fastcore/_nbdev.py

@@ -182,6 +182,7 @@
          "threaded": "03a_parallel.ipynb",
          "startthread": "03a_parallel.ipynb",
          "set_num_threads": "03a_parallel.ipynb",
+         "check_parallel_num": "03a_parallel.ipynb",
          "ThreadPoolExecutor": "03a_parallel.ipynb",
          "ProcessPoolExecutor": "03a_parallel.ipynb",
          "parallel": "03a_parallel.ipynb",

fastcore/parallel.py

@@ -1,7 +1,7 @@
 # AUTOGENERATED! DO NOT EDIT! File to edit: nbs/03a_parallel.ipynb (unless otherwise specified).
 
-__all__ = ['threaded', 'startthread', 'set_num_threads', 'ThreadPoolExecutor', 'ProcessPoolExecutor', 'parallel',
-           'run_procs', 'parallel_gen']
+__all__ = ['threaded', 'startthread', 'set_num_threads', 'check_parallel_num', 'ThreadPoolExecutor',
+           'ProcessPoolExecutor', 'parallel', 'run_procs', 'parallel_gen']
 
 # Cell
 from .imports import *
@@ -57,6 +57,14 @@ def _call(lock, pause, n, g, item):
             if l: lock.release()
     return g(item)
 
+# Cell
+def check_parallel_num(param_name, num_workers):
+    if sys.platform == "win32" and IN_NOTEBOOK and num_workers > 0:
+        print("Due to IPython and Windows limitation, python multiprocessing isn't available now.")
+        print(f"So `{param_name}` is changed to 0 to avoid getting stuck")
+        num_workers = 0
+    return num_workers
+
 # Cell
 class ThreadPoolExecutor(concurrent.futures.ThreadPoolExecutor):
     "Same as Python's ThreadPoolExecutor, except can pass `max_workers==0` for serial execution"
@@ -80,6 +88,7 @@ class ProcessPoolExecutor(concurrent.futures.ProcessPoolExecutor):
     "Same as Python's ProcessPoolExecutor, 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
+        max_workers = check_parallel_num('max_workers', max_workers)
         store_attr()
         self.not_parallel = max_workers==0
         if self.not_parallel: max_workers=1
@@ -126,6 +135,7 @@ def _done_pg(queue, items): return (queue.get() for _ in items)
 # Cell
 def parallel_gen(cls, items, n_workers=defaults.cpus, **kwargs):
     "Instantiate `cls` in `n_workers` procs & call each on a subset of `items` in parallel."
+    n_workers = check_parallel_num('n_workers', n_workers)
     if n_workers==0:
         yield from enumerate(list(cls(**kwargs)(items)))
         return

nbs/03_xtras.ipynb

@@ -618,7 +618,7 @@
     {
      "data": {
       "text/plain": [
-       "['e', 'f', 'h', 'a', 'd', 'c', 'g', 'b']"
+       "['c', 'f', 'e', 'g', 'h', 'b', 'd', 'a']"
       ]
      },
      "execution_count": null,
@@ -710,7 +710,7 @@
     "    with maybe_open(fn) as f: return f.encoding\n",
     "\n",
     "fname = '00_test.ipynb'\n",
-    "sys_encoding = 'cp1252' if sys.platform == \"win32\" else 'UTF-8'\n",
+    "sys_encoding = 'cp1252' if sys.platform == 'win32' else 'UTF-8'\n",
     "test_eq(_f(fname), sys_encoding)\n",
     "with open(fname) as fh: test_eq(_f(fh), sys_encoding)"
    ]
@@ -1264,7 +1264,7 @@
     {
      "data": {
       "text/plain": [
-       "Path('08_script.ipynb')"
+       "Path('.gitattributes')"
       ]
      },
      "execution_count": null,
@@ -1298,7 +1298,7 @@
     {
      "data": {
       "text/plain": [
-       "(Path('../fastcore/xtras.py'), Path('08_script.ipynb'))"
+       "(Path('../fastcore/__init__.py'), Path('01_basics.ipynb'))"
       ]
      },
      "execution_count": null,
@@ -1575,8 +1575,8 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "Num Events: 1, Freq/sec: 340.3\n",
-      "Most recent:  ▅▁▇▁▅ 287.7 217.2 325.6 229.3 290.2\n"
+      "Num Events: 8, Freq/sec: 423.0\n",
+      "Most recent:  ▂▂▁▁▇ 318.5 319.0 266.9 275.6 427.7\n"
      ]
     }
    ],
@@ -1725,7 +1725,7 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "2000-01-01 12:00:00 UTC is 2000-01-01 04:00:00-08:00 local time\n"
+      "2000-01-01 12:00:00 UTC is 2000-01-01 12:00:00+00:00 local time\n"
      ]
     }
    ],
@@ -1755,7 +1755,7 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "2000-01-01 12:00:00 local is 2000-01-01 20:00:00+00:00 UTC time\n"
+      "2000-01-01 12:00:00 local is 2000-01-01 12:00:00+00:00 UTC time\n"
      ]
     }
    ],

nbs/03a_parallel.ipynb

@@ -186,6 +186,21 @@
     "    return g(item)"
    ]
   },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "#export\n",
+    "def check_parallel_num(param_name, num_workers):\n",
+    "    if sys.platform == \"win32\" and IN_NOTEBOOK and num_workers > 0:\n",
+    "        print(\"Due to IPython and Windows limitation, python multiprocessing isn't available now.\")\n",
+    "        print(f\"So `{param_name}` is changed to 0 to avoid getting stuck\")\n",
+    "        num_workers = 0\n",
+    "    return num_workers"
+   ]
+  },
   {
    "cell_type": "code",
    "execution_count": null,
@@ -221,7 +236,7 @@
       "text/markdown": [
        "<h4 id=\"ThreadPoolExecutor\" class=\"doc_header\"><code>class</code> <code>ThreadPoolExecutor</code><a href=\"\" class=\"source_link\" style=\"float:right\">[source]</a></h4>\n",
        "\n",
-       "> <code>ThreadPoolExecutor</code>(**`max_workers`**=*`16`*, **`on_exc`**=*`print`*, **`pause`**=*`0`*, **\\*\\*`kwargs`**) :: [`ThreadPoolExecutor`](/parallel.html#ThreadPoolExecutor)\n",
+       "> <code>ThreadPoolExecutor</code>(**`max_workers`**=*`20`*, **`on_exc`**=*`print`*, **`pause`**=*`0`*, **\\*\\*`kwargs`**) :: [`ThreadPoolExecutor`](/parallel.html#ThreadPoolExecutor)\n",
        "\n",
        "Same as Python's ThreadPoolExecutor, except can pass `max_workers==0` for serial execution"
       ],
@@ -248,6 +263,7 @@
     "    \"Same as Python's ProcessPoolExecutor, 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",
+    "        max_workers = check_parallel_num('max_workers', max_workers)\n",
     "        store_attr()\n",
     "        self.not_parallel = max_workers==0\n",
     "        if self.not_parallel: max_workers=1\n",
@@ -272,7 +288,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`**=*`16`*, **`on_exc`**=*`print`*, **`pause`**=*`0`*, **\\*\\*`kwargs`**) :: [`ProcessPoolExecutor`](/parallel.html#ProcessPoolExecutor)\n",
+       "> <code>ProcessPoolExecutor</code>(**`max_workers`**=*`20`*, **`on_exc`**=*`print`*, **`pause`**=*`0`*, **\\*\\*`kwargs`**) :: [`ProcessPoolExecutor`](/parallel.html#ProcessPoolExecutor)\n",
        "\n",
        "Same as Python's ProcessPoolExecutor, except can pass `max_workers==0` for serial execution"
       ],
@@ -305,7 +321,7 @@
    "metadata": {},
    "outputs": [],
    "source": [
-    "#export \n",
+    "#export\n",
     "def parallel(f, items, *args, n_workers=defaults.cpus, total=None, progress=None, pause=0,\n",
     "             threadpool=False, timeout=None, chunksize=1, **kwargs):\n",
     "    \"Applies `func` in parallel to `items`, using `n_workers`\"\n",
@@ -329,10 +345,10 @@
     "    return x+a\n",
     "\n",
     "inp,exp = range(50),range(1,51)\n",
-    "if sys.platform != \"win32\":\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=1, a=2), range(2,52))\n",
+    "\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=1, a=2), range(2,52))\n",
     "test_eq(parallel(add_one, inp, n_workers=0), exp)\n",
     "test_eq(parallel(add_one, inp, n_workers=0, a=2), range(2,52))"
    ]
@@ -362,11 +378,11 @@
      "name": "stdout",
      "output_type": "stream",
      "text": [
-      "0 2021-01-22 21:17:38.942321\n",
-      "1 2021-01-22 21:17:39.192929\n",
-      "2 2021-01-22 21:17:39.444098\n",
-      "3 2021-01-22 21:17:39.695087\n",
-      "4 2021-01-22 21:17:39.946463\n"
+      "0 2021-02-03 09:51:30.561681\n",
+      "1 2021-02-03 09:51:30.812066\n",
+      "2 2021-02-03 09:51:31.063662\n",
+      "3 2021-02-03 09:51:31.313478\n",
+      "4 2021-02-03 09:51:31.564776\n"
      ]
     }
    ],
@@ -375,8 +391,7 @@
     "    time.sleep(random.random()/1000)\n",
     "    print(i, datetime.now())\n",
     "\n",
-    "test_n_workers = 0 if sys.platform == \"win32\" else 2\n",
-    "parallel(print_time, range(5), n_workers=test_n_workers, pause=0.25);"
+    "parallel(print_time, range(5), n_workers=2, pause=0.25);"
    ]
   },
   {
@@ -423,6 +438,7 @@
     "#export \n",
     "def parallel_gen(cls, items, n_workers=defaults.cpus, **kwargs):\n",
     "    \"Instantiate `cls` in `n_workers` procs & call each on a subset of `items` in parallel.\"\n",
+    "    n_workers = check_parallel_num('n_workers', n_workers)\n",
     "    if n_workers==0:\n",
     "        yield from enumerate(list(cls(**kwargs)(items)))\n",
     "        return\n",
@@ -461,10 +477,9 @@
     "idxs,dat1 = zip(*res.sorted(itemgetter(0)))\n",
     "test_eq(dat1, range(1,6))\n",
     "\n",
-    "if sys.platform != \"win32\":\n",
-    "    res = L(parallel_gen(_C, items, n_workers=3))\n",
-    "    idxs,dat2 = zip(*res.sorted(itemgetter(0)))\n",
-    "    test_eq(dat2, dat1)"
+    "res = L(parallel_gen(_C, items, n_workers=3))\n",
+    "idxs,dat2 = zip(*res.sorted(itemgetter(0)))\n",
+    "test_eq(dat2, dat1)"
    ]
   },
   {
@@ -500,8 +515,8 @@
     "            yield k+self.a\n",
     "\n",
     "x = np.linspace(0,0.99,20)\n",
-    "test_n_workers = 0 if sys.platform == \"win32\" else 2\n",
-    "res = L(parallel_gen(TestSleepyBatchFunc, x, n_workers=test_n_workers))\n",
+    "\n",
+    "res = L(parallel_gen(TestSleepyBatchFunc, x, n_workers=2))\n",
     "test_eq(res.sorted().itemgot(1), x+1)"
    ]
   },
@@ -546,7 +561,14 @@
    "execution_count": null,
    "metadata": {},
    "outputs": [],
-   "source": []
+   "source": [
+    "from subprocess import Popen, PIPE\n",
+    "# test num_workers > 0 in scripts works when python process start method is spawn\n",
+    "process = Popen([\"python\", \"parallel_test.py\"], stdout=PIPE)\n",
+    "_, err = process.communicate(timeout=5)\n",
+    "exit_code = process.wait()\n",
+    "test_eq(exit_code, 0)"
+   ]
   }
  ],
  "metadata": {

nbs/parallel_test.py

@@ -0,0 +1,23 @@
+from fastcore.parallel import *
+from datetime import *
+import random
+import os
+
+# from contextlib import contextmanager,ExitStack
+from multiprocessing import Process, Queue
+import concurrent.futures,time
+from multiprocessing import Manager
+
+def print_time(i): 
+    time.sleep(random.random()/1000)
+    print(i, os.getpid(), datetime.now())
+
+class _C:
+    def __call__(self, o): return ((i+1) for i in o)
+
+items = range(5)
+
+if __name__ == "__main__":
+    parallel(print_time, range(5), n_workers=2, pause=0.25);
+    res = list(parallel_gen(_C, items, n_workers=2))
+    print(res)