parallel in win notebook

Author: mszhanyi

fastcore/_nbdev.py

@@ -182,10 +182,11 @@
          "threaded": "03a_parallel.ipynb",
          "startthread": "03a_parallel.ipynb",
          "set_num_threads": "03a_parallel.ipynb",
-         "check_parallel_num": "03a_parallel.ipynb",
+         "parallelable": "03a_parallel.ipynb",
          "ThreadPoolExecutor": "03a_parallel.ipynb",
          "ProcessPoolExecutor": "03a_parallel.ipynb",
          "parallel": "03a_parallel.ipynb",
+         "add_one": "03a_parallel.ipynb",
          "run_procs": "03a_parallel.ipynb",
          "parallel_gen": "03a_parallel.ipynb",
          "url_default_headers": "03b_net.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', 'check_parallel_num', 'ThreadPoolExecutor',
-           'ProcessPoolExecutor', 'parallel', 'run_procs', 'parallel_gen']
+__all__ = ['threaded', 'startthread', 'set_num_threads', 'parallelable', 'ThreadPoolExecutor', 'ProcessPoolExecutor',
+           'parallel', 'add_one', 'run_procs', 'parallel_gen']
 
 # Cell
 from .imports import *
@@ -58,12 +58,13 @@ def _call(lock, pause, n, g, item):
     return g(item)
 
 # Cell
-def check_parallel_num(param_name, num_workers):
-    if sys.platform == "win32" and IN_NOTEBOOK and num_workers > 0:
+def parallelable(param_name, num_workers, f=None):
+    f_in_main = f == None or sys.modules[f.__module__].__name__ == "__main__"
+    if sys.platform == "win32" and IN_NOTEBOOK and num_workers > 0 and f_in_main:
         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
+        print(f"So `{param_name}` has to be changed to 0 to avoid getting stuck")
+        return False
+    return True
 
 # Cell
 class ThreadPoolExecutor(concurrent.futures.ThreadPoolExecutor):
@@ -88,13 +89,16 @@ 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
         super().__init__(max_workers, **kwargs)
 
     def map(self, f, items, *args, timeout=None, chunksize=1, **kwargs):
+        if not parallelable('max_workers', self.max_workers, f): self.max_workers = 0
+        self.not_parallel = self.max_workers==0
+        if self.not_parallel: self.max_workers=1
+
         if self.not_parallel == False: self.lock = Manager().Lock()
         g = partial(f, *args, **kwargs)
         if self.not_parallel: return map(g, items)
@@ -118,6 +122,13 @@ def parallel(f, items, *args, n_workers=defaults.cpus, total=None, progress=None
             r = progress_bar(r, total=total, leave=False)
         return L(r)
 
+# Cell
+def add_one(x, a=1):
+    # this import is necessary for multiprocessing in notebook on windows
+    import random
+    time.sleep(random.random()/80)
+    return x+a
+
 # Cell
 def run_procs(f, f_done, args):
     "Call `f` for each item in `args` in parallel, yielding `f_done`"
@@ -135,7 +146,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 not parallelable('n_workers', n_workers): n_workers = 0
     if n_workers==0:
         yield from enumerate(list(cls(**kwargs)(items)))
         return