Shared manager with redis

requirements.txt

@@ -13,8 +13,10 @@ pymlconf==0.7.0
 pyparsing==2.1.10
 pytest==3.0.4
 pytest-cov==2.5.1
+pytest-timeout==1.2.0
 pytest-xdist==1.16.0
 PyYAML==3.12
+redis==2.10.6
 six==1.10.0
+tmx==1.9.1
 typing==3.6.1
-pylibmc==1.5.2

synergine2/share.py

@@ -1,7 +1,8 @@
 # coding: utf-8
+import pickle
 import typing
 
-import pylibmc
+import redis
 
 from synergine2.exceptions import SynergineException
 
@@ -12,13 +13,15 @@ class SharedDataManager(object):
     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})
+        self._r = redis.StrictRedis(host='localhost', port=6379, db=0)  # TODO: configs
+        # TODO: Il faut écrire dans REDIS que lorsque l'on veut passer à l'étape processes, genre de commit
+        # sinon on va ecrire dans redis a chaque fois qu'on modifie une shared data c'est pas optimal.
 
     def set(self, key: str, value: typing.Any) -> None:
-        self._mc.set(key, value)
+        self._r.set(key, pickle.dumps(value))
 
     def get(self, key) -> typing.Any:
-        return self._mc.get(key)
+        return pickle.loads(self._r.get(key))
 
     def create(
         self,

tests/test_processing.py

@@ -21,6 +21,7 @@ class MyFakeClass(object):
 
 
 class TestProcessing(BaseTest):
+    @pytest.mark.timeout(10)
     def make_job_with_scalar(
             self,
             data: list,
@@ -28,6 +29,7 @@ class TestProcessing(BaseTest):
         result = sum(data)
         return result
 
+    @pytest.mark.timeout(10)
     def make_job_with_object(
             self,
             data: list,
@@ -50,6 +52,7 @@ class TestProcessing(BaseTest):
 
         assert sum(results) == 39600
 
+    @pytest.mark.timeout(10)
     def test_non_parallel_jobs_with_scalar(self):
         # TODO: process manager utilise actuellement un cpu quand même, changer ca
         process_manager = ProcessManager(
@@ -66,6 +69,7 @@ class TestProcessing(BaseTest):
         assert len(results) == 1
         assert final_result == 4950
 
+    @pytest.mark.timeout(10)
     def test_parallel_jobs_with_objects(self):
         process_manager = ProcessManager(
             config=Config({}),
@@ -84,12 +88,14 @@ class TestProcessing(BaseTest):
 
         assert final_result == 39600
 
+    @pytest.mark.timeout(10)
     def test_shared_memory_with_shared_manager(self):
         shared = SharedDataManager()
         shared.set('counter', 42)
 
         def job(*args, **kwargs):
-            return shared.get('counter') + 1
+            counter = shared.get('counter') or 0
+            return counter + 1
 
         process_manager = ProcessManager(
             config=Config({}),
@@ -102,6 +108,7 @@ class TestProcessing(BaseTest):
 
         assert results[0] == 43
 
+    @pytest.mark.timeout(10)
     def test_share_data_with_function(self):
         shared = SharedDataManager()
 
@@ -109,7 +116,8 @@ class TestProcessing(BaseTest):
             counter = shared.create('counter', 0)
 
         def job(*args, **kwargs):
-            return shared.get('counter') + 1
+            counter = shared.get('counter') or 0
+            return counter + 1
 
         process_manager = ProcessManager(
             config=Config({}),
@@ -130,13 +138,15 @@ class TestProcessing(BaseTest):
 
         process_manager.terminate()
 
+    @pytest.mark.timeout(10)
     def test_after_created_shared_data(self):
         shared = SharedDataManager()
 
         shared.set('foo_1', 0)
 
         def job(key):
-            return shared.get('foo_{}'.format(key)) + 1
+            value = shared.get('foo_{}'.format(key)) or 0
+            return value + 1
 
         process_manager = ProcessManager(
             config=Config({}),