3 # © Copyright 2021-2022, Scott Gasch
5 """parallelize unittest."""
11 import decorator_utils
13 import parallelize as p
16 logger = logging.getLogger(__name__)
19 @p.parallelize(method=p.Method.THREAD)
20 def compute_factorial_thread(n):
27 @p.parallelize(method=p.Method.PROCESS)
28 def compute_factorial_process(n):
35 @p.parallelize(method=p.Method.REMOTE)
36 def compute_factorial_remote(n):
43 @decorator_utils.timed
44 def test_thread_parallelization() -> None:
47 f = compute_factorial_thread(_)
49 smart_future.wait_all(results)
50 for future in results:
51 print(f'Thread: {future}')
52 texecutor = executors.DefaultExecutors().thread_pool()
56 @decorator_utils.timed
57 def test_process_parallelization() -> None:
60 results.append(compute_factorial_process(_))
61 for future in smart_future.wait_any(results):
62 print(f'Process: {future}')
63 pexecutor = executors.DefaultExecutors().process_pool()
67 @decorator_utils.timed
68 def test_remote_parallelization() -> None:
71 results.append(compute_factorial_remote(_))
72 for result in smart_future.wait_any(results):
74 rexecutor = executors.DefaultExecutors().remote_pool()
80 test_thread_parallelization()
81 test_process_parallelization()
82 test_remote_parallelization()
86 if __name__ == '__main__':
89 except Exception as e: