Change SharedDataManager to optimized way with synergine

synergine2/exceptions.py

@@ -13,3 +13,7 @@ class NotYetImplemented(SynergineException):
 
 class ConfigurationError(SynergineException):
     pass
+
+
+class UnknownSharedData(SynergineException):
+    pass

synergine2/share.py

@@ -5,6 +5,7 @@ import typing
 import redis
 
 from synergine2.exceptions import SynergineException
+from synergine2.exceptions import UnknownSharedData
 
 
 class SharedDataManager(object):
@@ -12,16 +13,36 @@ 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):
+    def __init__(self, clear: bool=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.
+
+        self._data = {}
+        self._modified_keys = set()
+
+        if clear:
+            self._r.flushdb()
 
     def set(self, key: str, value: typing.Any) -> None:
-        self._r.set(key, pickle.dumps(value))
+        self._data[key] = value
+        self._modified_keys.add(key)
 
     def get(self, key) -> typing.Any:
-        return pickle.loads(self._r.get(key))
+        if key not in self._data:
+            b_value = self._r.get(key)
+            if b_value is None:
+                # We not allow None value storage
+                raise UnknownSharedData('No shared data for key "{}"'.format(key))
+            self._data[key] = pickle.loads(b_value)
+
+        return self._data[key]
+
+    def commit(self) -> None:
+        for key in self._modified_keys:
+            self._r.set(key, pickle.dumps(self.get(key)))
+        self._modified_keys = set()
+
+    def refresh(self) -> None:
+        self._data = {}
 
     def create(
         self,

tests/test_processing.py

@@ -92,8 +92,10 @@ class TestProcessing(BaseTest):
     def test_shared_memory_with_shared_manager(self):
         shared = SharedDataManager()
         shared.set('counter', 42)
+        shared.commit()
 
         def job(*args, **kwargs):
+            shared.refresh()
             counter = shared.get('counter') or 0
             return counter + 1
 
@@ -116,6 +118,7 @@ class TestProcessing(BaseTest):
             counter = shared.create('counter', 0)
 
         def job(*args, **kwargs):
+            shared.refresh()
             counter = shared.get('counter') or 0
             return counter + 1
 
@@ -127,11 +130,13 @@ class TestProcessing(BaseTest):
 
         foo = Foo()
         foo.counter = 42
+        shared.commit()
 
         results = process_manager.make_them_work(None)
         assert results[0] == 43
 
         foo.counter = 45
+        shared.commit()
 
         results = process_manager.make_them_work(None)
         assert results[0] == 46
@@ -145,6 +150,7 @@ class TestProcessing(BaseTest):
         shared.set('foo_1', 0)
 
         def job(key):
+            shared.refresh()
             value = shared.get('foo_{}'.format(key)) or 0
             return value + 1
 
@@ -155,11 +161,13 @@ class TestProcessing(BaseTest):
         )
 
         shared.set('foo_1', 42)
+        shared.commit()
 
         results = process_manager.make_them_work('1')
         assert results[0] == 43
 
         shared.set('foo_2', 52)
+        shared.commit()
 
         results = process_manager.make_them_work('2')
         assert results[0] == 53

tests/test_share.py

@@ -1,4 +1,7 @@
 # coding: utf-8
+import pytest
+
+from synergine2.exceptions import UnknownSharedData
 from synergine2.share import SharedDataManager
 from tests import BaseTest
 
@@ -42,5 +45,63 @@ class TestShare(BaseTest):
 
         assert shared.get('{}_counter'.format(foo.id)) == 48
 
+    def test_update_dict_with_pointer(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            data = shared.create('data', {})
+
+        foo = Foo()
+        foo.data = {'foo': 'bar'}
+
+        assert shared.get('data') == {'foo': 'bar'}
+
+        foo.data['foo'] = 'buz'
+        assert shared.get('data') == {'foo': 'buz'}
+
+    def test_refresh_without_commit(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            counter = shared.create('counter', 0)
+
+        foo = Foo()
+        foo.counter = 42
+
+        assert shared.get('counter') == 42
+
+        shared.refresh()
+        with pytest.raises(UnknownSharedData):
+            shared.get('counter')
+
+    def test_commit(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            counter = shared.create('counter', 0)
+
+        foo = Foo()
+        foo.counter = 42
+
+        assert shared.get('counter') == 42
+
+        shared.commit()
+        assert shared.get('counter') == 42
+
+    def test_commit_then_refresh(self):
+        shared = SharedDataManager()
+
+        class Foo(object):
+            counter = shared.create('counter', 0)
+
+        foo = Foo()
+        foo.counter = 42
+
+        assert shared.get('counter') == 42
+
+        shared.commit()
+        shared.refresh()
+        assert shared.get('counter') == 42
+
     def test_indexes(self):
         pass