.

lib/zodb.py

@@ -145,11 +145,16 @@ def _zversion():
     dzodb  = pkg_resources.working_set.find(pkg_resources.Requirement.parse('ZODB'))
     v311   = pkg_resources.parse_version('3.11dev')
 
-    vzodb3 = dzodb3.parsed_version
-    if dzodb3 is not None and vzodb3 >= v311:
-        vzodb = dzodb.parsed_version # ZODB 3.11 just requires latest ZODB & ZEO
+    if dzodb3 is None and dzodb is None:
+        raise RuntimeError('ZODB is not installed')
+
+    if dzodb3 is not None:
+        if dzodb3.parsed_version >= v311:
+            vzodb = dzodb.parsed_version # ZODB 3.11 just requires latest ZODB & ZEO
+        else:
+            vzodb = dzodb3.parsed_version
     else:
-        vzodb = vzodb3
+        vzodb = dzodb.parsed_version
 
     assert vzodb is not None
     return vzodb

zopenrace4.py

@@ -8,6 +8,7 @@ from ZODB.MappingStorage import MappingStorage
 import transaction
 from persistent import Persistent
 
+from golang import func, defer, select, default
 from golang import sync, context
 
 
@@ -16,14 +17,41 @@ class PInt(Persistent):
     def __init__(self, i):
         self.i = i
 
+from wendelin.lib.testing import TestDB_ZEO
 
+
+@func
 def main():
-    zstor = MappingStorage()
-    db = DB(zstor)
+    """
+    if 0:
+        zstor = MappingStorage()
+        db = DB(zstor)
+    else:
+        tdb = TestDB_ZEO('<zeo>')
+        tdb.setup()
+        defer(tdb.teardown)
+
+        zstor = tdb.getZODBStorage()
+        db = DB(zstor)
+        #zstor.app.poll_thread.name = 'C.poll'
+    """
+
+    tdb = TestDB_ZEO('<zeo>')
+    tdb.setup()
+    defer(tdb.teardown)
+
+    def dbopen():
+        zstor = tdb.getZODBStorage()
+        db = DB(zstor)
+        return db
 
 
     # init initializes the database with two integer objects - obj1/obj2 that are set to 0.
+    @func
     def init():
+        db = dbopen()
+        defer(db.close)
+
         transaction.begin()
         zconn = db.open()
 
@@ -39,7 +67,11 @@ def main():
     #
     # access to obj1 is organized to always trigger loading from zstor.
     # access to obj2 goes through zconn cache and so verifies whether the cache is not stale.
-    def T1(ctx, N):
+    @func
+    def T1(ctx, name, N):
+        db = dbopen()
+        defer(db.close)
+
         def t1():
             transaction.begin()
             zconn = db.open()
@@ -62,14 +94,20 @@ def main():
             zconn.close()
 
         for i in range(N):
-            #print('T1.%d' % i)
+            #print('T1%s.%d' % (name, i))
             t1()
 
+        raise RuntimeError("T1: done")
+
 
     # T2 changes obj1/obj2 in a loop by doing `objX.i += 1`.
     #
     # Since both objects start from 0, the invariant that `obj1.i == obj2.i` is always preserved.
+    @func
     def T2(ctx, N):
+        db = dbopen()
+        defer(db.close)
+
         def t2():
             transaction.begin()
             zconn = db.open()
@@ -85,6 +123,8 @@ def main():
             zconn.close()
 
         for i in range(N):
+            if ready(ctx.done()):
+                break
             #print('T2.%d' % i)
             t2()
 
@@ -93,12 +133,25 @@ def main():
     # Connection.open, it triggers the bug where T1 sees stale obj2 with obj1.i != obj2.i
     init()
 
-    N = 1000
+    N = 100000
     wg = sync.WorkGroup(context.background())
-    wg.go(T1, N)
+    for x in 'abcdefgh':
+        wg.go(T1, x, N)
     wg.go(T2, N)
     wg.wait()
 
+    print('OK')
+
+
+# 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()