3 # © Copyright 2021-2022, Scott Gasch
5 """parallelize unittest."""
10 from pyutils import bootstrap, decorator_utils
11 from pyutils.parallelize import executors
12 from pyutils.parallelize import parallelize as p
13 from pyutils.parallelize import smart_future
15 logger = logging.getLogger(__name__)
18 @p.parallelize(method=p.Method.THREAD)
19 def compute_factorial_thread(n):
26 @p.parallelize(method=p.Method.PROCESS)
27 def compute_factorial_process(n):
34 @p.parallelize(method=p.Method.REMOTE)
35 def compute_factorial_remote(n):
42 @decorator_utils.timed
43 def test_thread_parallelization() -> None:
46 f = compute_factorial_thread(_)
48 smart_future.wait_all(results)
49 for future in results:
50 print(f'Thread: {future}')
51 texecutor = executors.DefaultExecutors().thread_pool()
55 @decorator_utils.timed
56 def test_process_parallelization() -> None:
59 results.append(compute_factorial_process(_))
60 for future in smart_future.wait_any(results):
61 print(f'Process: {future}')
62 pexecutor = executors.DefaultExecutors().process_pool()
66 @decorator_utils.timed
67 def test_remote_parallelization() -> None:
70 results.append(compute_factorial_remote(_))
71 for result in smart_future.wait_any(results):
73 rexecutor = executors.DefaultExecutors().remote_pool()
79 test_thread_parallelization()
80 test_process_parallelization()
81 test_remote_parallelization()
85 if __name__ == '__main__':
88 except Exception as e: