X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/650265966e83fca4ce8e9a416e9b5e358e82be98..d63d6ba645b8e34cf9d248bd0407fab0b9ef6534:/sdk/python/arvados/keep.py diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py index ce4c6f81f6..cbe96ffa2f 100644 --- a/sdk/python/arvados/keep.py +++ b/sdk/python/arvados/keep.py @@ -31,6 +31,7 @@ import resource from . import timer import urllib.parse import traceback +import weakref if sys.version_info >= (3, 0): from io import BytesIO @@ -185,6 +186,7 @@ class KeepBlockCache(object): self._max_slots = max_slots self._disk_cache = disk_cache self._disk_cache_dir = disk_cache_dir + self._cache_updating = threading.Condition(self._cache_lock) if self._disk_cache and self._disk_cache_dir is None: self._disk_cache_dir = os.path.join(os.path.expanduser("~"), ".cache", "arvados", "keep") @@ -192,10 +194,22 @@ class KeepBlockCache(object): if self._max_slots == 0: if self._disk_cache: - # default max slots to half of maximum file handles + # Each block uses two file descriptors, one used to + # open it initially and hold the flock(), and a second + # hidden one used by mmap(). + # + # Set max slots to 1/8 of maximum file handles. This + # means we'll use at most 1/4 of total file handles. + # # NOFILE typically defaults to 1024 on Linux so this - # will be 512 slots. - self._max_slots = resource.getrlimit(resource.RLIMIT_NOFILE)[0] / 2 + # is 128 slots (256 file handles), which means we can + # cache up to 8 GiB of 64 MiB blocks. This leaves + # 768 file handles for sockets and other stuff. + # + # When we want the ability to have more cache (e.g. in + # arv-mount) we'll increase rlimit before calling + # this. + self._max_slots = int(resource.getrlimit(resource.RLIMIT_NOFILE)[0] / 8) else: # RAM cache slots self._max_slots = 512 @@ -246,36 +260,50 @@ class KeepBlockCache(object): return len(self.content) def evict(self): - return True - - def cap_cache(self): - '''Cap the cache size to self.cache_max''' - with self._cache_lock: - # Select all slots except those where ready.is_set() and content is - # None (that means there was an error reading the block). - self._cache = [c for c in self._cache if not (c.ready.is_set() and c.content is None)] - sm = sum([slot.size() for slot in self._cache]) - while len(self._cache) > 0 and (sm > self.cache_max or len(self._cache) > self._max_slots): - for i in range(len(self._cache)-1, -1, -1): - # start from the back, find a slot that is a candidate to evict - if self._cache[i].ready.is_set(): - sz = self._cache[i].size() - - # If evict returns false it means the - # underlying disk cache couldn't lock the file - # for deletion because another process was using - # it. Don't count it as reducing the amount - # of data in the cache, find something else to - # throw out. - if self._cache[i].evict(): - sm -= sz - + self.content = None + return self.gone() + + def gone(self): + return (self.content is None) + + def _resize_cache(self, cache_max, max_slots): + # Try and make sure the contents of the cache do not exceed + # the supplied maximums. + + # Select all slots except those where ready.is_set() and content is + # None (that means there was an error reading the block). + self._cache = [c for c in self._cache if not (c.ready.is_set() and c.content is None)] + sm = sum([slot.size() for slot in self._cache]) + while len(self._cache) > 0 and (sm > cache_max or len(self._cache) > max_slots): + for i in range(len(self._cache)-1, -1, -1): + # start from the back, find a slot that is a candidate to evict + if self._cache[i].ready.is_set(): + sz = self._cache[i].size() + + # If evict returns false it means the + # underlying disk cache couldn't lock the file + # for deletion because another process was using + # it. Don't count it as reducing the amount + # of data in the cache, find something else to + # throw out. + if self._cache[i].evict(): + sm -= sz + + # check to make sure the underlying data is gone + if self._cache[i].gone(): # either way we forget about it. either the # other process will delete it, or if we need # it again and it is still there, we'll find # it on disk. del self._cache[i] - break + break + + + def cap_cache(self): + '''Cap the cache size to self.cache_max''' + with self._cache_updating: + self._resize_cache(self.cache_max, self._max_slots) + self._cache_updating.notify_all() def _get(self, locator): # Test if the locator is already in the cache @@ -302,12 +330,21 @@ class KeepBlockCache(object): def reserve_cache(self, locator): '''Reserve a cache slot for the specified locator, or return the existing slot.''' - with self._cache_lock: + with self._cache_updating: n = self._get(locator) if n: return n, False else: # Add a new cache slot for the locator + self._resize_cache(self.cache_max, self._max_slots-1) + while len(self._cache) >= self._max_slots: + # If there isn't a slot available, need to wait + # for something to happen that releases one of the + # cache slots. Idle for 200 ms or woken up by + # another thread + self._cache_updating.wait(timeout=0.2) + self._resize_cache(self.cache_max, self._max_slots-1) + if self._disk_cache: n = arvados.diskcache.DiskCacheSlot(locator, self._disk_cache_dir) else: @@ -316,12 +353,10 @@ class KeepBlockCache(object): return n, True def set(self, slot, blob): - tryagain = False - try: slot.set(blob) + return except OSError as e: - tryagain = True if e.errno == errno.ENOMEM: # Reduce max slots to current - 4, cap cache and retry with self._cache_lock: @@ -334,20 +369,24 @@ class KeepBlockCache(object): elif e.errno == errno.ENODEV: _logger.error("Unable to use disk cache: The underlying filesystem does not support memory mapping.") except Exception as e: - tryagain = True + pass + finally: + # Check if we should evict things from the cache. Either + # because we added a new thing or there was an error and + # we possibly adjusted the limits down, so we might need + # to push something out. + self.cap_cache() try: - if tryagain: - # There was an error. Evict some slots and try again. - self.cap_cache() - slot.set(blob) + # Only gets here if there was an error the first time. The + # exception handler adjusts limits downward in some cases + # to free up resources, which would make the operation + # succeed. + slot.set(blob) except Exception as e: # It failed again. Give up. + slot.set(None) raise arvados.errors.KeepCacheError("Unable to save block %s to disk cache: %s" % (slot.locator, e)) - finally: - # Set the notice that that we are done with the cache - # slot one way or another. - slot.ready.set() self.cap_cache()