tests/parallelize_itest.py

   1 #!/usr/bin/env python3
   2
   3 # © Copyright 2021-2022, Scott Gasch
   4
   5 """parallelize unittest."""
   6
   7 import logging
   8 import sys
   9
  10 import bootstrap
  11 import decorator_utils
  12 import executors
  13 import parallelize as p
  14 import smart_future
  15
  16 logger = logging.getLogger(__name__)
  17
  18
  19 @p.parallelize(method=p.Method.THREAD)
  20 def compute_factorial_thread(n):
  21     total = 1
  22     for x in range(2, n):
  23         total *= x
  24     return total
  25
  26
  27 @p.parallelize(method=p.Method.PROCESS)
  28 def compute_factorial_process(n):
  29     total = 1
  30     for x in range(2, n):
  31         total *= x
  32     return total
  33
  34
  35 @p.parallelize(method=p.Method.REMOTE)
  36 def compute_factorial_remote(n):
  37     total = 1
  38     for x in range(2, n):
  39         total *= x
  40     return total
  41
  42
  43 @decorator_utils.timed
  44 def test_thread_parallelization() -> None:
  45     results = []
  46     for _ in range(50):
  47         f = compute_factorial_thread(_)
  48         results.append(f)
  49     smart_future.wait_all(results)
  50     for future in results:
  51         print(f'Thread: {future}')
  52     texecutor = executors.DefaultExecutors().thread_pool()
  53     texecutor.shutdown()
  54
  55
  56 @decorator_utils.timed
  57 def test_process_parallelization() -> None:
  58     results = []
  59     for _ in range(50):
  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()
  64     pexecutor.shutdown()
  65
  66
  67 @decorator_utils.timed
  68 def test_remote_parallelization() -> None:
  69     results = []
  70     for _ in range(10):
  71         results.append(compute_factorial_remote(_))
  72     for result in smart_future.wait_any(results):
  73         print(result)
  74     rexecutor = executors.DefaultExecutors().remote_pool()
  75     rexecutor.shutdown()
  76
  77
  78 @bootstrap.initialize
  79 def main() -> None:
  80     test_thread_parallelization()
  81     test_process_parallelization()
  82     test_remote_parallelization()
  83     sys.exit(0)
  84
  85
  86 if __name__ == '__main__':
  87     try:
  88         main()
  89     except Exception as e:
  90         logger.exception(e)
  91         sys.exit(1)