Process Manager and Chun Manager + tests

synergine2/processing.py

@@ -0,0 +1,51 @@
+import types
+from multiprocessing import Process
+from multiprocessing import Manager
+
+from synergine2.utils import ChunkManager
+
+
+class ProcessManager(object):
+    def __init__(
+            self,
+            process_count: int,
+            chunk_manager: ChunkManager,
+            job_maker: types.FunctionType,
+    ):
+        self._process_count = process_count
+        self._chunk_manager = chunk_manager
+        self._job_maker = job_maker
+
+    def execute_jobs(self, data: list) -> tuple:
+        with Manager() as manager:
+            processes = list()
+            chunks = self._chunk_manager.make_chunks(data)
+            results = manager.dict()
+
+            # TODO: retrouver tests pour savoir si
+            # les keeped alive sont mieux
+            for process_number in range(self._process_count):
+                processes.append(Process(
+                    target=self._job_maker_wrapper,
+                    args=(
+                        process_number,
+                        chunks[process_number],
+                        results,
+                    )
+                ))
+
+            for process in processes:
+                process.start()
+
+            for process in processes:
+                process.join()
+
+            return results.values()
+
+    def _job_maker_wrapper(
+            self,
+            process_number: int,
+            chunk: list,
+            results: dict,
+    ):
+        results[process_number] = self._job_maker(chunk)

synergine2/utils.py

@@ -0,0 +1,12 @@
+
+
+class ChunkManager(object):
+    def __init__(self, chunks_numbers: int):
+        self._chunks_numbers = chunks_numbers
+
+    def make_chunks(self, data: list) -> list:
+        i, j, x = len(data), 0, []
+        for k in range(self._chunks_numbers):
+            a, j = j, j + (i + k) // self._chunks_numbers
+            x.append(data[a:j])
+        return x

tests/__init__.py

@@ -0,0 +1,4 @@
+
+
+class BaseTest(object):
+    pass

tests/test_processing.py

@@ -0,0 +1,38 @@
+import os
+
+from synergine2.processing import ProcessManager
+from synergine2.utils import ChunkManager
+from tests import BaseTest
+
+
+class TestProcessing(BaseTest):
+    @staticmethod
+    def _make_job(data_chunk: list) -> tuple:
+        current_pid = os.getpid()
+        result = sum(data_chunk)
+        return current_pid, result
+
+    def test_parallel_jobs(self):
+        chunk_manager = ChunkManager(4)
+        process_manager = ProcessManager(
+            process_count=4,
+            chunk_manager=chunk_manager,
+            job_maker=self._make_job,
+        )
+
+        data = list(range(100))
+        process_id_list = []
+        final_result = 0
+
+        results = process_manager.execute_jobs(data)
+
+        for process_id, result in results:
+            final_result += result
+            process_id_list.append(process_id)
+
+        # Test each process ids are differents
+        assert sorted(process_id_list) == \
+            sorted(list(set(process_id_list)))
+
+        # Goal is 4950
+        assert final_result == 4950

tests/test_utils.py

@@ -0,0 +1,27 @@
+from synergine2.utils import ChunkManager
+from tests import BaseTest
+
+
+class TestUtils(BaseTest):
+    def test_chunk_manager_round(self):
+        chunk_manager = ChunkManager(4)
+        data = list(range(100))
+
+        chunks = chunk_manager.make_chunks(data)
+
+        assert len(chunks) == 4
+        for chunk in chunks:
+            assert len(chunk) == 25
+
+    def test_chunk_manager_not_round(self):
+        chunk_manager = ChunkManager(4)
+        data = list(range(101))
+
+        chunks = chunk_manager.make_chunks(data)
+
+        assert len(chunks) == 4
+        for chunk_number, chunk in enumerate(chunks):
+            if chunk_number == 3:
+                assert len(chunk) == 26
+            else:
+                assert len(chunk) == 25