Avoid race condition in the matrix cache handling. (#1280)

* Improved fix for matrix cache handling.

During the MatrixDataCacheItem acquire function there was a
time when the write lock was released and the read lock was taken.
During that time, the dataset could have been deleted, later
result in the MatrixDataCacheManageri data adaptor returning None.

The solution is to demote the writer lock to a reader lock instead
of unlocking and relocking.

Also, when a the cache needs to delete an entry, the delete
is done outside the MatrixDataCacheManager lock.   This operation
only requires the write lock for the MatrixDataCacheItem.

Fixes #1255

server/common/app_config.py

@@ -56,6 +56,7 @@ class AppConfig(object):
             self.multi_dataset__dataroot = dc["multi_dataset"]["dataroot"]
             self.multi_dataset__index = dc["multi_dataset"]["index"]
             self.multi_dataset__allowed_matrix_types = dc["multi_dataset"]["allowed_matrix_types"]
+            self.multi_dataset__matrix_cache__max_datasets = dc["multi_dataset"]["matrix_cache"]["max_datasets"]
             self.single_dataset__datapath = dc["single_dataset"]["datapath"]
             self.single_dataset__obs_names = dc["single_dataset"]["obs_names"]
             self.single_dataset__var_names = dc["single_dataset"]["var_names"]
@@ -244,7 +245,8 @@ class AppConfig(object):
     def handle_multi_dataset(self, context):
         self.__check_attr("multi_dataset__dataroot", (type(None), str))
         self.__check_attr("multi_dataset__index", (type(None), bool, str))
-        self.__check_attr("multi_dataset__allowed_matrix_types", (list))
+        self.__check_attr("multi_dataset__allowed_matrix_types", (tuple, list))
+        self.__check_attr("multi_dataset__matrix_cache__max_datasets", int)
 
         if self.multi_dataset__dataroot is None:
             return
@@ -256,6 +258,9 @@ class AppConfig(object):
             except ValueError:
                 raise ConfigurationError(f'Invalid matrix type in "allowed_matrix_types": {mtype}')
 
+        # matrix cache
+        MatrixDataCacheManager.set_max_datasets(self.multi_dataset__matrix_cache__max_datasets)
+
     def handle_user_annotations(self, context):
         self.__check_attr("user_annotations__enable", bool)
         self.__check_attr("user_annotations__type", str)

server/common/default_config.py

@@ -29,6 +29,11 @@ multi_dataset:
   # A list of allowed matrix types.  If an empty list, then all matrix types are allowed
   allowed_matrix_types: []
 
+  matrix_cache:
+    # The maximum number of datasets that may be opened at one time.  The least recently used dataset
+    # is evicted from the cache first.
+    max_datasets: 5
+
 single_dataset:
   datapath: null
   obs_names: null
@@ -63,6 +68,7 @@ adaptor:
 
   anndata_adaptor:
       backed: false
+
 """
 
 

server/data_common/matrix_loader.py

@@ -19,32 +19,33 @@ class MatrixDataCacheItem(object):
         self.data_adaptor = None
         self.data_lock = RWLock()
 
-    def acquire(self, app_config):
+    def acquire_existing(self):
-        """returns the data_adaptor if cached.  opens the data_adaptor if not.
+        """If the data_adaptor exists, take a read lock and return it, else return None"""
-        In either case, the a reader lock is taken.  Must call release when
-        the data_adaptor is no longer needed"""
-
         self.data_lock.r_acquire()
         if self.data_adaptor:
             return self.data_adaptor
 
         self.data_lock.r_release()
-        try:
+        return None
-            with self.data_lock.w_locked():
-                # the data may have been loaded while waiting on the lock
-                if not self.data_adaptor:
-                    self.loader.pre_load_validation()
-                    self.data_adaptor = self.loader.open(app_config)
-
-        except Exception:
-            # necessary to acquire after an exception, since the release will occur when
-            # the context exits
-            self.data_lock.r_acquire()
-            raise
 
+    def acquire_and_open(self, app_config):
+        """returns the data_adaptor if cached.  opens the data_adaptor if not.
+        In either case, the a reader lock is taken.  Must call release when
+        the data_adaptor is no longer needed"""
         self.data_lock.r_acquire()
         if self.data_adaptor:
             return self.data_adaptor
+        self.data_lock.r_release()
+
+        self.data_lock.w_acquire()
+        # the data may have been loaded while waiting on the lock
+        if not self.data_adaptor:
+            self.loader.pre_load_validation()
+            self.data_adaptor = self.loader.open(app_config)
+
+        # demote the write lock to a read lock.
+        self.data_lock.w_demote()
+        return self.data_adaptor
 
     def release(self):
         """Release the reader lock"""
@@ -65,7 +66,7 @@ class MatrixDataCacheManager(object):
     when the context ends.  This class currently implements a simple least recently used cache,
     which can delete a dataset from the cache to make room for a new oneo
 
-    This is the indended usage pattern:
+    This is the intended usage pattern:
 
            m = MatrixDataCacheManager()
            with m.data_adaptor(location, app_config) as data_adaptor:
@@ -77,7 +78,11 @@ class MatrixDataCacheManager(object):
     #  TODO:  This is very simple.  This can be improved by taking into account how much space is actually
     #         taken by each dataset, instead of arbitrarily picking a max datasets to cache.
     #         Also, this should be controlled by a configuration parameter.
-    MAX_CACHED = 3
+    MAX_CACHED = 5
+
+    @staticmethod
+    def set_max_datasets(max_cached):
+        MatrixDataCacheManager.MAX_CACHED = max_cached
 
     # FIXME:   If the number of active datasets exceeds the MAX_CACHED, then each request could
     # lead to a dataset being deleted and a new only being opened: the cache will get thrashed.
@@ -97,33 +102,43 @@ class MatrixDataCacheManager(object):
     @contextmanager
     def data_adaptor(self, location, app_config):
         # create a loader for to this location if it does not already exist
+
+        delete_adaptor = None
+        data_adaptor = None
+
         with self.lock:
             value = self.datasets.get(location)
             if value is not None:
                 cache_item = value[0]
                 last_accessed = time.time()
                 self.datasets[location] = (cache_item, last_accessed)
-            else:
+                data_adaptor = cache_item.acquire_existing()
+
+            if data_adaptor is None:
                 while True:
                     # find the last access times for each loader
-                    items = list(self.datasets.items())
+                    if len(self.datasets) < self.MAX_CACHED:
-                    sorted(items, key=lambda x: x[1][1])
-                    if len(items) < self.MAX_CACHED:
                         break
 
+                    items = list(self.datasets.items())
+                    sorted(items, key=lambda x: x[1][1])
                     # close the least recently used loader
                     oldest = items[0]
                     oldest_cache = oldest[1][0]
                     oldest_key = oldest[0]
-                    oldest_cache.delete()
                     del self.datasets[oldest_key]
+                    delete_adaptor = oldest_cache
 
                 last_accessed = time.time()
                 loader = MatrixDataLoader(location, app_config=app_config)
                 cache_item = MatrixDataCacheItem(loader)
                 self.datasets[location] = (cache_item, last_accessed)
+
         try:
-            data_adaptor = cache_item.acquire(app_config)
+            if delete_adaptor:
+                delete_adaptor.delete()
+            if data_adaptor is None:
+                data_adaptor = cache_item.acquire_and_open(app_config)
             yield data_adaptor
         finally:
             cache_item.release()

server/data_common/rwlock.py

@@ -12,6 +12,8 @@
     Code written by Tyler Neylon at Unbox Research.
 
     This file is public domain.
+
+    Modified to add a w_demote function to convert a writer lock to a reader lock
 """
 
 
@@ -52,15 +54,26 @@ class RWLock(object):
         self.num_r_lock = Lock()
         self.num_r = 0
 
+        # The d_lock is needed to handle the demotion case,
+        # so that the writer can become a reader without releasing the w_lock.
+        # the d_lock is held by the writer, and prevents any other thread from taking the
+        # num_r_lock during that time, which means the writer thread is able to take the
+        # num_r_lock to update the num_r.
+        self.d_lock = Lock()
+
     # ___________________________________________________________________
     # Reading methods.
 
     def r_acquire(self):
+        self.d_lock.acquire()
         self.num_r_lock.acquire()
         self.num_r += 1
+
         if self.num_r == 1:
             self.w_lock.acquire()
+
         self.num_r_lock.release()
+        self.d_lock.release()
 
     def r_release(self):
         assert self.num_r > 0
@@ -68,6 +81,7 @@ class RWLock(object):
         self.num_r -= 1
         if self.num_r == 0:
             self.w_lock.release()
+
         self.num_r_lock.release()
 
     @contextmanager
@@ -83,10 +97,23 @@ class RWLock(object):
     # Writing methods.
 
     def w_acquire(self):
+        self.d_lock.acquire()
         self.w_lock.acquire()
 
     def w_release(self):
         self.w_lock.release()
+        self.d_lock.release()
+
+    def w_demote(self):
+        """demote a writer lock to a reader lock"""
+
+        # the d_lock is already held from w_acquire.
+        # releasing the d_lock at the end of this function allows multiple readers.
+        # incrementing num_r makes this thread one of those readers.
+        self.num_r_lock.acquire()
+        self.num_r += 1
+        self.num_r_lock.release()
+        self.d_lock.release()
 
     @contextmanager
     def w_locked(self):