Make Storage.load generate a network barrier to master node.

This makes sure all invalidations about loaded object up to its actual load
have been received by client, and forwarded to ZODB.
This fixes a potential problem of a transaction being able to see objects
committed by another transaction after it started itself.

git-svn-id: https://svn.erp5.org/repos/neo/trunk@2286 71dcc9de-d417-0410-9af5-da40c76e7ee4

neo/client/app.py

@@ -550,7 +550,15 @@ class Application(object):
             finally:
                 self._cache_lock_release()
             # Otherwise get it from storage node
-            return self._load(oid, cache=1)[:2]
+            result = self._load(oid, cache=1)[:2]
+            # Start a network barrier, so we get all invalidations *after* we
+            # received data. This ensures we get any invalidation message that
+            # would have been about the version we loaded.
+            # Those invalidations are checked at ZODB level, so it decides if
+            # loaded data can be handed to current transaction or if a separate
+            # loadBefore call is required.
+            self._askPrimary(Packets.AskBarrier())
+            return result
         finally:
             self._load_lock_release()
 

neo/handler.py

@@ -347,6 +347,12 @@ class EventHandler(object):
     def answerHasLock(self, conn, oid, status):
         raise UnexpectedPacketError
 
+    def askBarrier(self, conn):
+        conn.answer(Packets.AnswerBarrier())
+
+    def answerBarrier(self, conn):
+        pass
+
     # Error packet handlers.
 
     def error(self, conn, code, message):
@@ -460,6 +466,8 @@ class EventHandler(object):
         d[Packets.AnswerObjectUndoSerial] = self.answerObjectUndoSerial
         d[Packets.AskHasLock] = self.askHasLock
         d[Packets.AnswerHasLock] = self.answerHasLock
+        d[Packets.AskBarrier] = self.askBarrier
+        d[Packets.AnswerBarrier] = self.answerBarrier
 
         return d
 

neo/protocol.py

@@ -1623,6 +1623,17 @@ class AnswerHasLock(Packet):
         oid, state = unpack(self._header_format, body)
         return (oid, _decodeLockState(state))
 
+class AskBarrier(Packet):
+    """
+    Initates a "network barrier", allowing the node sending this packet to know
+    when all packets sent previously on the same connection have been handled
+    by its peer.
+    """
+    pass
+
+class AnswerBarrier(Packet):
+    pass
+
 class Error(Packet):
     """
     Error is a special type of message, because this can be sent against
@@ -1858,6 +1869,10 @@ class PacketRegistry(dict):
             0x0036,
             AskObjectHistoryFrom,
             AnswerObjectHistoryFrom)
+    AskBarrier, AnswerBarrier = register(
+            0x037,
+            AskBarrier,
+            AnswerBarrier)
 
 # build a "singleton"
 Packets = PacketRegistry()

neo/tests/client/testClientApp.py

@@ -261,6 +261,12 @@ class ClientApplicationTests(NeoTestBase):
         })
         app.cp = Mock({ 'getConnForCell' : conn})
         app.local_var.asked_object = an_object[:-1]
+        answer_barrier = Packets.AnswerBarrier()
+        answer_barrier.setId(1)
+        app.master_conn = Mock({
+            'getNextId': 1,
+            'fakeReceived': answer_barrier,
+        })
         result = app.load(oid)
         self.assertEquals(result, ('OBJ', tid1))
         self.checkAskObject(conn)