client: remove load lock in tpc_finish

neo/client/app.py

@@ -725,10 +725,7 @@ class Application(ThreadedApplication):
         txn_container = self._txn_container
         if not txn_container.get(transaction).voted:
             self.tpc_vote(transaction)
-        checked_list = []
-        self._load_lock_acquire()
-        try:
-            # Call finish on master
+        if 1:
             txn_context = txn_container.pop(transaction)
             cache_dict = txn_context.cache_dict
             checked_list = [oid for oid, data  in cache_dict.iteritems()
@@ -745,8 +742,6 @@ class Application(ThreadedApplication):
                 if not tid:
                     raise
             return tid
-        finally:
-            self._load_lock_release()
 
     def _getFinalTID(self, ttid):
         try:

neo/client/handlers/master.py

@@ -82,9 +82,13 @@ class PrimaryNotificationsHandler(MTEventHandler):
         cache = app._cache
         app._cache_lock_acquire()
         try:
+            invalidate = app._cache.invalidate
+            loading = app._loading_oid
             for oid, data in cache_dict.iteritems():
                 # Update ex-latest value in cache
-                cache.invalidate(oid, tid)
+                invalidate(oid, tid)
+                if oid == loading:
+                    app._loading_invalidated.append(tid)
                 if data is not None:
                     # Store in cache with no next_tid
                     cache.store(oid, data, tid, None)

neo/tests/threaded/test.py

@@ -862,6 +862,27 @@ class Test(NEOThreadedTest):
         self.assertEqual(c.root()['1'].value, 1)
         self.assertNotIn('2', c.root())
 
+    @with_cluster()
+    def testLoadVsFinish(self, cluster):
+        t1, c1 = cluster.getTransaction()
+        c1.root()['x'] = x1 = PCounter()
+        t1.commit()
+        t1.begin()
+        x1.value = 1
+        t2, c2 = cluster.getTransaction()
+        x2 = c2.root()['x']
+        cluster.client._cache.clear()
+        def _loadFromStorage(orig, *args):
+            r = orig(*args)
+            ll()
+            return r
+        with LockLock() as ll, Patch(cluster.client,
+                _loadFromStorage=_loadFromStorage):
+            t = self.newThread(x2._p_activate)
+            ll()
+            t1.commit()
+        t.join()
+
     @with_cluster()
     def testInternalInvalidation(self, cluster):
         def _handlePacket(orig, conn, packet, kw={}, handler=None):