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):