From 5ec66c5b0e100467b533b264d6ccdead6858a8ba Mon Sep 17 00:00:00 2001 From: bmccandless Date: Tue, 24 Mar 2020 12:23:47 -0700 Subject: [PATCH] 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 | 7 +++- server/common/default_config.py | 6 +++ server/data_common/matrix_loader.py | 65 ++++++++++++++++++----------- server/data_common/rwlock.py | 27 ++++++++++++ 4 files changed, 79 insertions(+), 26 deletions(-) diff --git a/server/common/app_config.py b/server/common/app_config.py index f7b57ccf..a3b745ef 100644 --- a/server/common/app_config.py +++ b/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) diff --git a/server/common/default_config.py b/server/common/default_config.py index 4b6618b3..4f8ee4d5 100644 --- a/server/common/default_config.py +++ b/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 + """ diff --git a/server/data_common/matrix_loader.py b/server/data_common/matrix_loader.py index 65145b40..bd80b0a9 100644 --- a/server/data_common/matrix_loader.py +++ b/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): - """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""" - + def acquire_existing(self): + """If the data_adaptor exists, take a read lock and return it, else return None""" self.data_lock.r_acquire() if self.data_adaptor: return self.data_adaptor self.data_lock.r_release() - try: - 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 + return None + 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()) - sorted(items, key=lambda x: x[1][1]) - if len(items) < self.MAX_CACHED: + if len(self.datasets) < 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() diff --git a/server/data_common/rwlock.py b/server/data_common/rwlock.py index 0d8ac283..562b8a49 100644 --- a/server/data_common/rwlock.py +++ b/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):