client: fix invalidation issues when reconnecting to the master

neo/client/app.py

@@ -332,10 +332,7 @@ class Application(ThreadedApplication):
 
         acquire = self._cache_lock_acquire
         release = self._cache_lock_release
-        # XXX: Is it possible this giant lock ?
-        #      See commit b77c946d67c9d7cc1e9ee9b15437568dee144aa4
-        #      for a way to invalidate cache properly when several loads
-        #      are done simultaneously.
+        # XXX: Consider using a more fine-grained lock.
         self._load_lock_acquire()
         try:
             acquire()
@@ -356,14 +353,20 @@ class Application(ThreadedApplication):
             data, tid, next_tid, _ = self._loadFromStorage(oid, tid, before_tid)
             acquire()
             try:
-                result = data, tid, (next_tid if self._loading_oid or next_tid
-                                              else self._loading_invalidated)
-                self._cache.store(oid, *result)
-                return result
+                if self._loading_oid:
+                    # Common case (no race condition).
+                    self._cache.store(oid, data, tid, next_tid)
+                elif self._loading_invalidated:
+                    # oid has just been invalidated.
+                    if not next_tid:
+                        next_tid = self._loading_invalidated
+                    self._cache.store(oid, data, tid, next_tid)
+                # Else, we just reconnected to the master.
             finally:
                 release()
         finally:
             self._load_lock_release()
+        return data, tid, next_tid
 
     def _loadFromStorage(self, oid, at_tid, before_tid):
         packet = Packets.AskObject(oid, at_tid, before_tid)

neo/client/cache.py

@@ -264,7 +264,6 @@ class ClientCache(object):
                 assert item.next_tid <= tid, (item, oid, tid)
 
     def clear_current(self):
-        oid_list = []
         for oid, item_list in self._oid_dict.items():
             item = item_list[-1]
             if item.next_tid is None:
@@ -276,8 +275,6 @@ class ClientCache(object):
                 # probably not worth it.
                 if not item_list:
                     del self._oid_dict[oid]
-                oid_list.append(oid)
-        return oid_list
 
 
 def test(self):
@@ -295,11 +292,11 @@ def test(self):
     data = '15', 15, None
     cache.store(1, *data)
     self.assertEqual(cache.load(1, None), data)
-    self.assertEqual(cache.clear_current(), [1])
+    cache.clear_current()
     self.assertEqual(cache.load(1, None), None)
     cache.store(1, *data)
     cache.invalidate(1, 20)
-    self.assertEqual(cache.clear_current(), [])
+    cache.clear_current()
     self.assertEqual(cache.load(1, 20), ('15', 15, 20))
     cache.store(1, '10', 10, 15)
     cache.store(1, '20', 20, 21)

neo/client/handlers/master.py

@@ -102,19 +102,24 @@ class PrimaryNotificationsHandler(MTEventHandler):
                 if app.master_conn is None:
                     app._cache_lock_acquire()
                     try:
-                        db = app.getDB()
                         if app.last_tid < ltid:
-                            oid_list = app._cache.clear_current()
-                            db is None or db.invalidate(
-                                app.last_tid and add64(app.last_tid, 1),
-                                oid_list)
+                            app._cache.clear_current()
+                            # In the past, we tried not to invalidate the
+                            # Connection caches entirely, using the list of
+                            # oids that are invalidated by clear_current.
+                            # This was wrong because these caches may have
+                            # entries that are not in the NEO cache anymore.
                         else:
                             # The DB was truncated. It happens so
                             # rarely that we don't need to optimize.
                             app._cache.clear()
-                            db is None or db.invalidateCache()
+                        # Make sure a parallel load won't refill the cache
+                        # with garbage.
+                        app._loading_oid = app._loading_invalidated = None
                     finally:
                         app._cache_lock_release()
+                    db = app.getDB()
+                    db is None or db.invalidateCache()
                 app.last_tid = ltid
         elif type(packet) is Packets.AnswerTransactionFinished:
             app = self.app