Source code for wholecell.utils.test_parallelization

"""
Test the parallelization utility.
"""

import multiprocessing
import platform
import time
import unittest

import pytest

from wholecell.utils import parallelization

# Silence Sphinx autodoc warning
unittest.TestCase.__module__ = "unittest"


[docs] def _square(i): # Runs in inner_pool square = i**2 time.sleep(i / 100) return square
[docs] def _sum_squares(i, j): # Runs in outer_pool with parallelization.pool(num_processes=2, nestable=False) as inner_pool: squares = inner_pool.map(_square, (i, j)) sum_squares = sum(squares) time.sleep(0.01) return sum_squares
[docs] class Test_parallelization(unittest.TestCase):
[docs] def _check_multi_sum(self, processes, nestable): i = 2 j = 3 expected_sum = i * i + j * j with parallelization.pool( num_processes=processes, nestable=nestable ) as outer_pool: actual_sum = outer_pool.apply_async(_sum_squares, (i, j)).get() assert actual_sum == expected_sum
def test_cpus(self): assert 1 <= parallelization.cpus() <= multiprocessing.cpu_count() assert 1 <= parallelization.cpus(4) <= 4 def test_is_macos(self): assert parallelization.is_macos() == (platform.system() == "Darwin") def test_nested_pools(self): """Test nested process pools. A nested multiprocessing.Pool will raise an AssertionError that daemonic processes are not allowed to have children. """ self._check_multi_sum(1, nestable=False) # InlinePool (always nestable) self._check_multi_sum(2, nestable=True) # NoDaemonPool if parallelization.cpus(2) > 1: # else an InlinePool with pytest.raises(AssertionError, match="daemonic"): self._check_multi_sum(2, nestable=False) # mp.Pool