3 # © Copyright 2021-2022, Scott Gasch
5 """A decorator to help with dead simple parallelization."""
15 """How should we parallelize; by threads, processes or remote workers?"""
23 _funct: typing.Optional[typing.Callable] = None, *, method: Method = Method.THREAD
27 @parallelize # defaults to thread-mode
28 def my_function(a, b, c) -> int:
29 ...do some slow / expensive work, e.g., an http request
31 @parallelize(method=Method.PROCESS)
32 def my_other_function(d, e, f) -> str:
33 ...do more really expensice work, e.g., a network read
35 @parallelize(method=Method.REMOTE)
36 def my_other_other_function(g, h) -> int:
37 ...this work will be distributed to a remote machine pool
39 This decorator will invoke the wrapped function on:
41 Method.THREAD (default): a background thread
42 Method.PROCESS: a background process
43 Method.REMOTE: a process on a remote host
45 The wrapped function returns immediately with a value that is
46 wrapped in a SmartFuture. This value will block if it is either
47 read directly (via a call to result._resolve) or indirectly (by
48 using the result in an expression, printing it, hashing it,
49 passing it a function argument, etc...). See comments on the
50 SmartFuture class for details.
52 Note: you may stack @parallelized methods and it will "work".
53 That said, having multiple layers of Method.PROCESS or
54 Method.REMOTE may prove to be problematic because each process in
55 the stack will use its own independent pool which may overload
56 your machine with processes or your network with remote processes
57 beyond the control mechanisms built into one instance of the pool.
60 Also note: there is a non trivial overhead of pickling code and
61 scp'ing it over the network when you use Method.REMOTE. There's
62 a smaller but still considerable cost of creating a new process
63 and passing code to/from it when you use Method.PROCESS.
66 def wrapper(funct: typing.Callable):
67 @functools.wraps(funct)
68 def inner_wrapper(*args, **kwargs):
72 # Look for as of yet unresolved arguments in _funct's
73 # argument list and resolve them now.
76 newargs.append(smart_future.SmartFuture.resolve(arg))
79 newkwargs[kw] = smart_future.SmartFuture.resolve(kwargs[kw])
82 if method == Method.PROCESS:
83 executor = executors.DefaultExecutors().process_pool()
84 elif method == Method.THREAD:
85 executor = executors.DefaultExecutors().thread_pool()
86 elif method == Method.REMOTE:
87 executor = executors.DefaultExecutors().remote_pool()
88 assert executor is not None
89 atexit.register(executors.DefaultExecutors().shutdown)
91 future = executor.submit(funct, *newargs, **newkwargs)
93 # Wrap the future that's returned in a SmartFuture object
94 # so that callers do not need to call .result(), they can
95 # just use is as normal.
96 return smart_future.SmartFuture(future)
103 return wrapper(_funct)