.

zloadrace.py

@@ -2,20 +2,41 @@
 """Program zloadrace.py demonstrates concurrency bug in ZODB Connection.setstate()
 that leads to XXX
 
-XXX
+XXX no, there is no load vs invalidation race on ZODB4 (ZODB3 is probably the same):
+
+ZEO
+---
+
+ZEO server explicitly guarantees that it does not mix processing load
+requests inside tpc_finish + send invalidations. This way if load is processed
+after new commit, load reply is guranteed to come to client after invalidation
+message. This was explicitly fixed by
+
+    https://github.com/zopefoundation/ZEO/commit/71eb1456
+    (search for callAsyncNoPoll there)
+
+and later again by https://github.com/zopefoundation/ZEO/commit/94f275c3 .
+
+NEO
+---
+
 
 XXX load vs invalidation race is there on ZODB4 and ZODB3, but on ZODB5 there is
 another open vs invalidation race.
 """
 
+from __future__ import print_function
 from ZODB import DB
 import transaction
 from persistent import Persistent
 
 from wendelin.lib.testing import TestDB_ZEO, TestDB_NEO
 
-from golang import func, defer
+from golang import func, defer, chan, select, default
 from golang import sync, context
+import threading
+
+from ZODB.utils import u64
 
 
 # PInt is persistent integer.
@@ -25,8 +46,8 @@ class PInt(Persistent):
 
 @func
 def main():
-    tdb = TestDB_ZEO('<zeo>')
-    #tdb = TestDB_NEO('<neo>')
+    #tdb = TestDB_ZEO('<zeo>')
+    tdb = TestDB_NEO('<neo>')
 
     tdb.setup()
     # XXX defer(tdb.teardown)
@@ -38,6 +59,9 @@ def main():
     db1 = DB(zstor1)
     db2 = DB(zstor2)
 
+    zstor1.app.poll_thread.name = 'C1.poll'
+    zstor2.app.poll_thread.name = 'C2.poll'
+
 
     # XXX doc
     def init():
@@ -52,35 +76,64 @@ def main():
         zconn.close()
 
 
+    c2ready = chan()        # c1 <- c2  "I'm ready to commit"
+    c2start = chan()        # c1 -> c2  "go on to commit"
+
     def C1(ctx, N):
+        threading.current_thread().name = "C1"
         def c1():
             transaction.begin()
             zconn = db1.open()
 
+            print('C1: (1) neo.app.last_tid = @%d' % u64(zstor1.app.last_tid))
+
             root = zconn.root()
             obj1 = root['obj1']
             obj2 = root['obj2']
 
+            I = obj1.i
+
+            print('C1: (2) neo.app.last_tid = @%d' % u64(zstor1.app.last_tid))
+            print('C1: (2) obj1.serial = @%d' % u64(obj1._p_serial))
+
+            c2ready.recv()
+            c2start.send(1)
+            import time
+            time.sleep(0.5)
+
             # obj1 - reload it from zstor
             # obj2 - get it from zconn cache
-            for i in range(10*N):
+            #for i in range(N):
+            for i in range(15):
                 obj1._p_invalidate()
 
+                print('C1: (X) neo.app.last_tid = @%d' % u64(zstor1.app.last_tid))
+
                 # both objects must have the same values
                 i1 = obj1.i
                 i2 = obj2.i
+                print('C1: (X) obj1.serial = @%d' % u64(obj1._p_serial))
+                print('C1: (X) obj2.serial = @%d' % u64(obj2._p_serial))
                 if i1 != i2:
                     raise AssertionError("C1: obj1.i (%d)  !=  obj2.i (%d)" % (i1, i2))
+                if i1 != I:
+                    raise AssertionError(
+                        "C1: obj1.i (%d) mutated inside transaction (started with %d)" % (i1, I))
 
             transaction.abort()
             zconn.close()
 
         for i in range(N):
+            if ready(ctx.done()):
+                break
             print('C1.%d' % i)
             c1()
 
+        print('C1.fin')
+
 
     def C2(ctx, N):
+        threading.current_thread().name = "C2"
         def c2():
             transaction.begin()
             zconn = db2.open()
@@ -92,24 +145,44 @@ def main():
             obj2.i += 1
             assert obj1.i == obj2.i
 
+            c2ready.send(1)
+            c2start.recv()
+
             transaction.commit()
             zconn.close()
 
         for i in range(N):
+            if ready(ctx.done()):
+                break
             print('C2.%d' % i)
             c2()
 
+        print('C2.fin')
+
 
 
 
     init()
+    import time
+    time.sleep(2)
+    print()
 
-    N = 100
+    N = 1000
     wg = sync.WorkGroup(context.background())
     wg.go(C1, N)
     wg.go(C2, N)
     wg.wait()
 
 
+# ready returns whether channel ch is ready.
+def ready(ch):
+    _, _rx = select(
+        default,        # 0
+        ch.recv,        # 1
+    )
+    if _ == 0:
+        return False
+    return True
+
 if __name__ == '__main__':
     main()

zopenrace.py

@@ -87,6 +87,7 @@ XXX but they do have load vs invalidation race.
 [2] https://github.com/zopefoundation/ZODB/blob/5.5.1-29-g0b3db5aee/src/ZODB/mvccadapter.py#L130-L139
 """
 
+from __future__ import print_function
 from ZODB import DB
 from ZODB.MappingStorage import MappingStorage
 import transaction