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