share data with memcached

requirements.txt

@@ -17,3 +17,4 @@ pytest-xdist==1.16.0
 PyYAML==3.12
 six==1.10.0
 typing==3.6.1
+pylibmc==1.5.2

synergine2/cycle.py

@@ -24,6 +24,7 @@ class CycleManager(BaseObject):
     ):
         if process_manager is None:
             process_manager = ProcessManager(
+                config=config,
                 # TODO: Changer de config de merde (core.use_x_cores)
                 process_count=config.get('core', {}).get('use_x_cores', multiprocessing.cpu_count()),
                 chunk_manager=ChunkManager(multiprocessing.cpu_count()),

synergine2/processing.py

@@ -1,47 +1,120 @@
 # coding: utf-8
-import types
-from multiprocessing import Pool
+import typing
+from multiprocessing import Process
+from multiprocessing.connection import Connection
+from multiprocessing.connection import Pipe
 
 from synergine2.base import BaseObject
-from synergine2.utils import ChunkManager
+from synergine2.config import Config
+from synergine2.share import SharedDataManager
+
+STOP = '__STOP__'
+
+# global shared manager
+shared_data = SharedDataManager()
+
+
+# TODO: se jobs
+class Job(object):
+    pass
+
+
+class Worker(object):
+    def __init__(
+        self,
+        config: Config,
+        real_job: typing.Callable[..., typing.Any],
+    ) -> None:
+        self.config = config
+
+        local_read_pipe, local_write_pipe = Pipe(duplex=False)
+        process_read_pipe, process_write_pipe = Pipe(duplex=False)
+
+        self.local_read_pipe = local_read_pipe  # type: Connection
+        self.local_write_pipe = local_write_pipe  # type: Connection
+        self.process_read_pipe = process_read_pipe  # type: Connection
+        self.process_write_pipe = process_write_pipe  # type: Connection
+
+        self.real_job = real_job
+        self.process = Process(
+            target=self.work,
+            args=(
+                self.local_write_pipe,
+                self.process_read_pipe,
+            )
+        )
+        self.db = None  # type: RedisDatabase
+        self.process.start()
+
+    def work(self, *args, **kwargs):
+        while True:
+            message = self.process_read_pipe.recv()
+            if message == STOP:
+                return
+
+            result = self.real_job(message)
+            self.local_write_pipe.send(result)
 
 
 class ProcessManager(BaseObject):
     def __init__(
             self,
+            config: Config,
             process_count: int,
-            chunk_manager: ChunkManager,
-    ):
+            job: typing.Callable[..., typing.Any],
+    ) -> None:
+        self.config = config
         self._process_count = process_count
-        self._chunk_manager = chunk_manager
-        self.pool = Pool(processes=self._process_count)
-
-    def __getstate__(self):
-        self_dict = self.__dict__.copy()
-        self_dict['pool'] = None
-        return self_dict
-
-    def chunk_and_execute_jobs(self, data: list, job_maker: types.FunctionType) -> list:
-        chunks = self._chunk_manager.make_chunks(data)
-
-        if self._process_count > 1:
-            print('USE POOL')
-            results = self.pool.starmap(job_maker, [(chunk, i, self._process_count) for i, chunk in enumerate(chunks)])
-        else:
-            print('USE MONO')
-            results = [job_maker(data, 0, 1)]
-
-        return results
-
-    def execute_jobs(self, data: object, job_maker: types.FunctionType) -> list:
-        # TODO: Is there a reason to make multiprocessing here ? data is not chunked ...
-        if self._process_count > 1:
-            results = self.pool.starmap(job_maker, [(data, i, self._process_count) for i in range(self._process_count)])
-        else:
-            results = [job_maker(data, 0, 1)]
-
-        return results
-
-    def __del__(self):
-        if self.pool:
-            self.pool.terminate()
+        self.workers = []
+        self.start_workers(process_count, job)
+
+    def start_workers(self, worker_count: int, job: typing.Callable[..., typing.Any]) -> None:
+        assert not self.workers
+        for i in range(worker_count):
+            self.workers.append(Worker(self.config, job))
+
+    def make_them_work(self, message: typing.Any) -> 'TODO':
+        responses = []
+
+        for worker in self.workers:
+            worker.process_write_pipe.send(message)
+
+        for worker in self.workers:
+            responses.append(worker.local_read_pipe.recv())
+
+        return responses
+
+    def terminate(self) -> None:
+        for worker in self.workers:
+            worker.process_write_pipe.send(STOP)
+
+        for worker in self.workers:
+            worker.process.join()
+
+    #
+    # def chunk_and_execute_jobs(self, data: list, job_maker: types.FunctionType) -> list:
+    #     chunks = self._chunk_manager.make_chunks(data)
+    #
+    #     if self._process_count > 1:
+    #         print('USE POOL')
+    #         results = self.pool.starmap(job_maker, [(chunk, i, self._process_count) for i, chunk in enumerate(chunks)])
+    #     else:
+    #         print('USE MONO')
+    #         results = [job_maker(data, 0, 1)]
+    #
+    #     return results
+    #
+    # def execute_jobs(self, data: object, job_maker: types.FunctionType) -> list:
+    #     # TODO: Is there a reason to make multiprocessing here ? data is not chunked ...
+    #     if self._process_count > 1:
+    #         results = self.pool.starmap(job_maker, [(data, i, self._process_count) for i in range(self._process_count)])
+    #     else:
+    #         results = [job_maker(data, 0, 1)]
+    #
+    #     return results
+    #
+    # def __del__(self):
+    #     # TODO: DEV
+    #     return
+    #     if self.pool:
+    #         self.pool.terminate()

synergine2/share.py

@@ -0,0 +1,57 @@
+# coding: utf-8
+import typing
+
+import pylibmc
+
+from synergine2.exceptions import SynergineException
+
+
+class SharedDataManager(object):
+    """
+    This object is designed to own shared memory between processes. It must be feed (with set method) before
+    start of processes. Processes will only be able to access shared memory filled here before start.
+    """
+    def __init__(self):
+        self._mc = pylibmc.Client(['127.0.0.1'], binary=True, behaviors={"tcp_nodelay": True, "ketama": True})
+
+    def set(self, key: str, value: typing.Any) -> None:
+        self._mc.set(key, value)
+
+    def get(self, key) -> typing.Any:
+        return self._mc.get(key)
+
+    def create(
+        self,
+        key: str,
+        value,
+        indexes=None,
+    ):
+        def get_key(obj):
+            return key
+
+        def get_key_with_id(obj):
+            return key.format(id=obj.id)
+
+        if '{id}' in key:
+            key_formatter = get_key_with_id
+        else:
+            self.set(key, value)
+            key_formatter = get_key
+
+        def fget(self_):
+            return self.get(key)
+
+        def fset(self_, value_):
+            self.set(key_formatter(self_), value_)
+
+        def fdel(self_):
+            raise SynergineException('You cannot delete a shared data')
+
+        shared_property = property(
+            fget=fget,
+            fset=fset,
+            fdel=fdel,
+        )
+
+        return shared_property
+

tests/test_processing.py

@@ -1,9 +1,15 @@
 # coding: utf-8
-import os
+import ctypes
+import multiprocessing
 import psutil
+from multiprocessing import Manager
+from multiprocessing import Array
 
+import pytest
+
+from synergine2.config import Config
 from synergine2.processing import ProcessManager
-from synergine2.utils import ChunkManager
+from synergine2.share import SharedDataManager
 from tests import BaseTest
 
 available_cores = len(psutil.Process().cpu_affinity())
@@ -17,85 +23,135 @@ class MyFakeClass(object):
 class TestProcessing(BaseTest):
     def make_job_with_scalar(
             self,
-            data_chunk: list,
-            process_number: int,
-            process_count: int,
-    ) -> tuple:
-        current_pid = os.getpid()
-        result = sum(data_chunk)
-        return current_pid, result
+            data: list,
+    ):
+        result = sum(data)
+        return result
 
     def make_job_with_object(
             self,
-            data_chunk: list,
-            process_number: int,
-            process_count: int,
-    ) -> tuple:
-        current_pid = os.getpid()
-        data = [o.value for o in data_chunk]
+            data: list,
+    ):
+        data = [o.value for o in data]
         result = sum(data)
-        return current_pid, MyFakeClass(result)
+        return MyFakeClass(result)
 
     def test_parallel_jobs_with_scalar(self):
-        chunk_manager = ChunkManager(available_cores)
         process_manager = ProcessManager(
+            config=Config({}),
             process_count=available_cores,
-            chunk_manager=chunk_manager,
+            job=self.make_job_with_scalar,
         )
 
         data = list(range(100))
-        process_id_list = []
-        final_result = 0
 
-        results = process_manager.chunk_and_execute_jobs(
-            data,
-            job_maker=self.make_job_with_scalar,
-        )
-
-        for process_id, result in results:
-            final_result += result
-            process_id_list.append(process_id)
+        results = process_manager.make_them_work(data)
+        process_manager.terminate()
 
-        # Goal is 4950
-        assert final_result == 4950
+        assert sum(results) == 39600
 
     def test_non_parallel_jobs_with_scalar(self):
-        chunk_manager = ChunkManager(1)
+        # TODO: process manager utilise actuellement un cpu quand même, changer ca
         process_manager = ProcessManager(
+            config=Config({}),
             process_count=1,
-            chunk_manager=chunk_manager,
+            job=self.make_job_with_scalar,
         )
 
         data = list(range(100))
-        results = process_manager.chunk_and_execute_jobs(
-            data,
-            job_maker=self.make_job_with_scalar,
-        )
-        process_id, final_result = results[0]
+        results = process_manager.make_them_work(data)
+        process_manager.terminate()
+        final_result = results[0]
 
         assert len(results) == 1
-        assert process_id == os.getpid()
         assert final_result == 4950
 
     def test_parallel_jobs_with_objects(self):
-        chunk_manager = ChunkManager(available_cores)
         process_manager = ProcessManager(
+            config=Config({}),
             process_count=available_cores,
-            chunk_manager=chunk_manager,
+            job=self.make_job_with_object,
         )
 
         data = [MyFakeClass(v) for v in range(100)]
-        process_id_list = []
         final_result = 0
 
-        results = process_manager.chunk_and_execute_jobs(
-            data,
-            job_maker=self.make_job_with_object,
-        )
+        results = process_manager.make_them_work(data)
+        process_manager.terminate()
 
-        for process_id, result_object in results:
+        for result_object in results:
             final_result += result_object.value
-            process_id_list.append(process_id)
 
-        # Goal is 4950
-        assert final_result == 4950
+        assert final_result == 39600
+
+    def test_shared_memory_with_shared_manager(self):
+        shared = SharedDataManager()
+        shared.set('counter', 42)
+
+        def job(*args, **kwargs):
+            return shared.get('counter') + 1
+
+        process_manager = ProcessManager(
+            config=Config({}),
+            process_count=available_cores,
+            job=job,
+        )
+
+        results = process_manager.make_them_work(None)
+        process_manager.terminate()
+
+        assert results[0] == 43
+
+    def test_share_data_with_function(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            counter = shared.create('counter', 0)
+
+        def job(*args, **kwargs):
+            return shared.get('counter') + 1
+
+        process_manager = ProcessManager(
+            config=Config({}),
+            process_count=available_cores,
+            job=job,
+        )
+
+        foo = Foo()
+        foo.counter = 42
+
+        results = process_manager.make_them_work(None)
+        assert results[0] == 43
+
+        foo.counter = 45
+
+        results = process_manager.make_them_work(None)
+        assert results[0] == 46
+
+        process_manager.terminate()
+
+    def test_after_created_shared_data(self):
+        shared = SharedDataManager()
+
+        shared.set('foo_1', 0)
+
+        def job(key):
+            return shared.get('foo_{}'.format(key)) + 1
+
+        process_manager = ProcessManager(
+            config=Config({}),
+            process_count=available_cores,
+            job=job,
+        )
+
+        shared.set('foo_1', 42)
+
+        results = process_manager.make_them_work('1')
+        assert results[0] == 43
+
+        shared.set('foo_2', 52)
+
+        results = process_manager.make_them_work('2')
+        assert results[0] == 53
+
+        process_manager.terminate()

tests/test_share.py

@@ -0,0 +1,46 @@
+# coding: utf-8
+from synergine2.share import SharedDataManager
+from tests import BaseTest
+
+
+class TestShare(BaseTest):
+    def test_simple_share_with_class(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            counter = shared.create('counter', 0)
+
+        foo = Foo()
+        foo.counter = 42
+
+        assert shared.get('counter') == 42
+
+        foo.counter = 48
+
+        assert shared.get('counter') == 48
+
+    def test_dynamic_key(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            counter = shared.create(
+                '{id}_counter',
+                (0, 0, 0),
+                indexes=[],
+            )
+
+            @property
+            def id(self):
+                return id(self)
+
+        foo = Foo()
+        foo.counter = 42
+
+        assert shared.get('{}_counter'.format(foo.id)) == 42
+
+        foo.counter = 48
+
+        assert shared.get('{}_counter'.format(foo.id)) == 48
+
+    def test_indexes(self):
+        pass