X-Git-Url: https://git.arvados.org/arvados.git/blobdiff_plain/67e56f190b9a78e3c45cc7d90510fc631e0d04b6..650265966e83fca4ce8e9a416e9b5e358e82be98:/sdk/python/arvados/keep.py

diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py
index 7f316c153c..ce4c6f81f6 100644
--- a/sdk/python/arvados/keep.py
+++ b/sdk/python/arvados/keep.py
@@ -15,6 +15,7 @@ from builtins import object
 import collections
 import datetime
 import hashlib
+import errno
 import io
 import logging
 import math
@@ -29,6 +30,7 @@ import threading
 import resource
 from . import timer
 import urllib.parse
+import traceback
 
 if sys.version_info >= (3, 0):
     from io import BytesIO
@@ -190,23 +192,37 @@ class KeepBlockCache(object):
 
         if self._max_slots == 0:
             if self._disk_cache:
-                # default set max slots to half of maximum file handles
+                # default max slots to half of maximum 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
             else:
-                self._max_slots = 1024
+                # RAM cache slots
+                self._max_slots = 512
 
         if self.cache_max == 0:
             if self._disk_cache:
                 fs = os.statvfs(self._disk_cache_dir)
-                avail = (fs.f_bavail * fs.f_bsize) / 2
-                # Half the available space or max_slots * 64 MiB
-                self.cache_max = min(avail, (self._max_slots * 64 * 1024 * 1024))
+                # Calculation of available space incorporates existing cache usage
+                existing_usage = arvados.diskcache.DiskCacheSlot.cache_usage(self._disk_cache_dir)
+                avail = (fs.f_bavail * fs.f_bsize + existing_usage) / 4
+                maxdisk = int((fs.f_blocks * fs.f_bsize) * 0.10)
+                # pick smallest of:
+                # 10% of total disk size
+                # 25% of available space
+                # max_slots * 64 MiB
+                self.cache_max = min(min(maxdisk, avail), (self._max_slots * 64 * 1024 * 1024))
             else:
-                # 256 GiB in RAM
+                # 256 MiB in RAM
                 self.cache_max = (256 * 1024 * 1024)
 
         self.cache_max = max(self.cache_max, 64 * 1024 * 1024)
 
+        if self._disk_cache:
+            self._cache = arvados.diskcache.DiskCacheSlot.init_cache(self._disk_cache_dir, self._max_slots)
+            self.cap_cache()
+
+
     class CacheSlot(object):
         __slots__ = ("locator", "ready", "content")
 
@@ -230,7 +246,7 @@ class KeepBlockCache(object):
                 return len(self.content)
 
         def evict(self):
-            pass
+            return True
 
     def cap_cache(self):
         '''Cap the cache size to self.cache_max'''
@@ -241,11 +257,25 @@ class KeepBlockCache(object):
             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():
-                        self._cache[i].evict()
+                        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
+
+                        # 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
-                sm = sum([slot.size() for slot in self._cache])
 
     def _get(self, locator):
         # Test if the locator is already in the cache
@@ -285,6 +315,42 @@ class KeepBlockCache(object):
                 self._cache.insert(0, n)
                 return n, True
 
+    def set(self, slot, blob):
+        tryagain = False
+
+        try:
+            slot.set(blob)
+        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:
+                    self._max_slots = max(4, len(self._cache) - 4)
+            elif e.errno == errno.ENOSPC:
+                # Reduce disk max space to current - 256 MiB, cap cache and retry
+                with self._cache_lock:
+                    sm = sum([st.size() for st in self._cache])
+                    self.cache_max = max((256 * 1024 * 1024), sm - (256 * 1024 * 1024))
+            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
+
+        try:
+            if tryagain:
+                # There was an error.  Evict some slots and try again.
+                self.cap_cache()
+                slot.set(blob)
+        except Exception as e:
+            # It failed again.  Give up.
+            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()
+
 class Counter(object):
     def __init__(self, v=0):
         self._lk = threading.Lock()
@@ -1210,8 +1276,7 @@ class KeepClient(object):
                 return blob
         finally:
             if slot is not None:
-                slot.set(blob)
-                self.block_cache.cap_cache()
+                self.block_cache.set(slot, blob)
 
         # Q: Including 403 is necessary for the Keep tests to continue
         # passing, but maybe they should expect KeepReadError instead?