X-Git-Url: https://wannabe.guru.org/gitweb/?a=blobdiff_plain;f=parallelize.py;h=77d7649fe956a1542c1a60ce99ca96365ef59ce0;hb=532df2c5b57c7517dfb3dddd8c1358fbadf8baf3;hp=334dc4e992c56aebcad796d2f6696bd63b14653e;hpb=497fb9e21f45ec08e1486abaee6dfa7b20b8a691;p=python_utils.git

diff --git a/parallelize.py b/parallelize.py
index 334dc4e..77d7649 100644
--- a/parallelize.py
+++ b/parallelize.py
@@ -1,43 +1,40 @@
 #!/usr/bin/env python3
 
+# © Copyright 2021-2022, Scott Gasch
+
 """A decorator to help with dead simple parallelization."""
 
-from enum import Enum
+
+import atexit
 import functools
 import typing
-
-import executors
-import smart_future
-
-ps_count = 0
-thread_count = 0
-remote_count = 0
+from enum import Enum
 
 
 class Method(Enum):
+    """How should we parallelize; by threads, processes or remote workers?"""
+
     THREAD = 1
     PROCESS = 2
     REMOTE = 3
 
 
 def parallelize(
-        _funct: typing.Optional[typing.Callable] = None,
-        *,
-        method: Method = Method.THREAD
+    _funct: typing.Optional[typing.Callable] = None, *, method: Method = Method.THREAD
 ) -> typing.Callable:
     """Usage:
 
-    @parallelize    # defaults to thread-mode
-    def my_function(a, b, c) -> int:
-        ...do some slow / expensive work, e.g., an http request
+        @parallelize    # defaults to thread-mode
+        def my_function(a, b, c) -> int:
+            ...do some slow / expensive work, e.g., an http request
 
-    @parallelize(method=Method.PROCESS)
-    def my_other_function(d, e, f) -> str:
-        ...do more really expensice work, e.g., a network read
+        @parallelize(method=Method.PROCESS)
+        def my_other_function(d, e, f) -> str:
+            ...do more really expensice work, e.g., a network read
 
-    @parallelize(method=Method.REMOTE)
-    def my_other_other_function(g, h) -> int:
-        ...this work will be distributed to a remote machine pool
+        @parallelize(method=Method.REMOTE)
+        def my_other_other_function(g, h) -> int:
+            ...this work will be distributed to a remote machine pool
 
     This decorator will invoke the wrapped function on:
 
@@ -59,11 +56,19 @@ def parallelize(
     your machine with processes or your network with remote processes
     beyond the control mechanisms built into one instance of the pool.
     Be careful.
+
+    Also note: there is a non trivial overhead of pickling code and
+    scp'ing it over the network when you use Method.REMOTE.  There's
+    a smaller but still considerable cost of creating a new process
+    and passing code to/from it when you use Method.PROCESS.
     """
-    def wrapper(funct: typing.Callable):
 
+    def wrapper(funct: typing.Callable):
         @functools.wraps(funct)
         def inner_wrapper(*args, **kwargs):
+            import executors
+            import smart_future
+
             # Look for as of yet unresolved arguments in _funct's
             # argument list and resolve them now.
             newargs = []
@@ -71,9 +76,7 @@ def parallelize(
                 newargs.append(smart_future.SmartFuture.resolve(arg))
             newkwargs = {}
             for kw in kwargs:
-                newkwargs[kw] = smart_future.SmartFuture.resolve(
-                    kwargs[kw]
-                )
+                newkwargs[kw] = smart_future.SmartFuture.resolve(kwargs[kw])
 
             executor = None
             if method == Method.PROCESS:
@@ -83,6 +86,7 @@ def parallelize(
             elif method == Method.REMOTE:
                 executor = executors.DefaultExecutors().remote_pool()
             assert executor is not None
+            atexit.register(executors.DefaultExecutors().shutdown)
 
             future = executor.submit(funct, *newargs, **newkwargs)