[arvados] updated: 2.1.0-3093-g650265966
git repository hosting
git at public.arvados.org
Mon Nov 28 21:43:20 UTC 2022
Summary of changes:
sdk/python/arvados/diskcache.py | 113 +++++++++++++++++++----------------
sdk/python/arvados/errors.py | 2 +
sdk/python/arvados/keep.py | 47 +++++++++++++--
sdk/python/tests/test_keep_client.py | 30 +++++++++-
4 files changed, 135 insertions(+), 57 deletions(-)
via 650265966e83fca4ce8e9a416e9b5e358e82be98 (commit)
from 2cddd3c4e531e37c6d960c08da91a552bf75a1cc (commit)
Those revisions listed above that are new to this repository have
not appeared on any other notification email; so we list those
revisions in full, below.
commit 650265966e83fca4ce8e9a416e9b5e358e82be98
Author: Peter Amstutz <peter.amstutz at curii.com>
Date: Mon Nov 28 16:39:03 2022 -0500
18842: Turn disk write errors into KeepCacheError
* DiskCacheSlot properly keeps track of the file handle now
* Failure to write disk block to cache will attempt to shrink cache
and try again, if it still fails, turns into KeepCacheError
* init_cache writes a test block to check for working cache
Arvados-DCO-1.1-Signed-off-by: Peter Amstutz <peter.amstutz at curii.com>
diff --git a/sdk/python/arvados/diskcache.py b/sdk/python/arvados/diskcache.py
index 939ca3bb3..74b2a77b2 100644
--- a/sdk/python/arvados/diskcache.py
+++ b/sdk/python/arvados/diskcache.py
@@ -18,13 +18,14 @@ _logger = logging.getLogger('arvados.keep')
cacheblock_suffix = ".keepcacheblock"
class DiskCacheSlot(object):
- __slots__ = ("locator", "ready", "content", "cachedir")
+ __slots__ = ("locator", "ready", "content", "cachedir", "filehandle")
def __init__(self, locator, cachedir):
self.locator = locator
self.ready = threading.Event()
self.content = None
self.cachedir = cachedir
+ self.filehandle = None
def get(self):
self.ready.wait()
@@ -51,31 +52,20 @@ class DiskCacheSlot(object):
final = os.path.join(blockdir, self.locator) + cacheblock_suffix
- f = tempfile.NamedTemporaryFile(dir=blockdir, delete=False, prefix="tmp", suffix=cacheblock_suffix)
- tmpfile = f.name
+ self.filehandle = tempfile.NamedTemporaryFile(dir=blockdir, delete=False, prefix="tmp", suffix=cacheblock_suffix)
+ tmpfile = self.filehandle.name
os.chmod(tmpfile, stat.S_IRUSR | stat.S_IWUSR)
# aquire a shared lock, this tells other processes that
# we're using this block and to please not delete it.
- fcntl.flock(f, fcntl.LOCK_SH)
+ fcntl.flock(self.filehandle, fcntl.LOCK_SH)
- f.write(value)
- f.flush()
+ self.filehandle.write(value)
+ self.filehandle.flush()
os.rename(tmpfile, final)
tmpfile = None
- self.content = mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ)
- except OSError as e:
- if e.errno == errno.ENODEV:
- _logger.error("Unable to use disk cache: The underlying filesystem does not support memory mapping.")
- elif e.errno == errno.ENOMEM:
- _logger.error("Unable to use disk cache: The process's maximum number of mappings would have been exceeded.")
- elif e.errno == errno.ENOSPC:
- _logger.error("Unable to use disk cache: Out of disk space.")
- else:
- traceback.print_exc()
- except Exception as e:
- traceback.print_exc()
+ self.content = mmap.mmap(self.filehandle.fileno(), 0, access=mmap.ACCESS_READ)
finally:
if tmpfile is not None:
# If the tempfile hasn't been renamed on disk yet, try to delete it.
@@ -83,12 +73,6 @@ class DiskCacheSlot(object):
os.remove(tmpfile)
except:
pass
- if self.content is None:
- # Something went wrong with the disk cache, fall back
- # to RAM cache behavior (the alternative is to cache
- # nothing and return a read error).
- self.content = value
- self.ready.set()
def size(self):
if self.content is None:
@@ -120,36 +104,36 @@ class DiskCacheSlot(object):
blockdir = os.path.join(self.cachedir, self.locator[0:3])
final = os.path.join(blockdir, self.locator) + cacheblock_suffix
try:
- with open(final, "rb") as f:
- # unlock
- fcntl.flock(f, fcntl.LOCK_UN)
- self.content = None
-
- # try to get an exclusive lock, this ensures other
- # processes are not using the block. It is
- # nonblocking and will throw an exception if we
- # can't get it, which is fine because that means
- # we just won't try to delete it.
- #
- # I should note here, the file locking is not
- # strictly necessary, we could just remove it and
- # the kernel would ensure that the underlying
- # inode remains available as long as other
- # processes still have the file open. However, if
- # you have multiple processes sharing the cache
- # and deleting each other's files, you'll end up
- # with a bunch of ghost files that don't show up
- # in the file system but are still taking up
- # space, which isn't particularly user friendly.
- # The locking strategy ensures that cache blocks
- # in use remain visible.
- #
- fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
-
- os.remove(final)
- return True
+ fcntl.flock(self.filehandle, fcntl.LOCK_UN)
+
+ # try to get an exclusive lock, this ensures other
+ # processes are not using the block. It is
+ # nonblocking and will throw an exception if we
+ # can't get it, which is fine because that means
+ # we just won't try to delete it.
+ #
+ # I should note here, the file locking is not
+ # strictly necessary, we could just remove it and
+ # the kernel would ensure that the underlying
+ # inode remains available as long as other
+ # processes still have the file open. However, if
+ # you have multiple processes sharing the cache
+ # and deleting each other's files, you'll end up
+ # with a bunch of ghost files that don't show up
+ # in the file system but are still taking up
+ # space, which isn't particularly user friendly.
+ # The locking strategy ensures that cache blocks
+ # in use remain visible.
+ #
+ fcntl.flock(self.filehandle, fcntl.LOCK_EX | fcntl.LOCK_NB)
+
+ os.remove(final)
+ return True
except OSError:
pass
+ finally:
+ self.filehandle = None
+ self.content = None
return False
@staticmethod
@@ -166,6 +150,7 @@ class DiskCacheSlot(object):
content = mmap.mmap(filehandle.fileno(), 0, access=mmap.ACCESS_READ)
dc = DiskCacheSlot(locator, cachedir)
+ dc.filehandle = filehandle
dc.content = content
dc.ready.set()
return dc
@@ -176,8 +161,32 @@ class DiskCacheSlot(object):
return None
+ @staticmethod
+ def cache_usage(cachedir):
+ usage = 0
+ for root, dirs, files in os.walk(cachedir):
+ for name in files:
+ if not name.endswith(cacheblock_suffix):
+ continue
+
+ blockpath = os.path.join(root, name)
+ res = os.stat(blockpath)
+ usage += res.st_size
+ return usage
+
+
@staticmethod
def init_cache(cachedir, maxslots):
+ #
+ # First check the disk cache works at all by creating a 1 byte cache entry
+ #
+ checkexists = DiskCacheSlot.get_from_disk('0cc175b9c0f1b6a831c399e269772661', cachedir)
+ ds = DiskCacheSlot('0cc175b9c0f1b6a831c399e269772661', cachedir)
+ ds.set(b'a')
+ if checkexists is None:
+ # Don't keep the test entry around unless it existed beforehand.
+ ds.evict()
+
# map in all the files in the cache directory, up to max slots.
# after max slots, try to delete the excess blocks.
#
diff --git a/sdk/python/arvados/errors.py b/sdk/python/arvados/errors.py
index 4fe1f7654..15b1f6d4b 100644
--- a/sdk/python/arvados/errors.py
+++ b/sdk/python/arvados/errors.py
@@ -82,6 +82,8 @@ class KeepReadError(KeepRequestError):
pass
class KeepWriteError(KeepRequestError):
pass
+class KeepCacheError(KeepRequestError):
+ pass
class NotFoundError(KeepReadError):
pass
class NotImplementedError(Exception):
diff --git a/sdk/python/arvados/keep.py b/sdk/python/arvados/keep.py
index dd99e8b92..ce4c6f81f 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
@@ -201,7 +203,9 @@ class KeepBlockCache(object):
if self.cache_max == 0:
if self._disk_cache:
fs = os.statvfs(self._disk_cache_dir)
- avail = (fs.f_bavail * fs.f_bsize) / 4
+ # 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
@@ -209,7 +213,7 @@ class KeepBlockCache(object):
# 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)
@@ -311,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()
@@ -1236,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?
diff --git a/sdk/python/tests/test_keep_client.py b/sdk/python/tests/test_keep_client.py
index 570710a58..5a7cf2c2a 100644
--- a/sdk/python/tests/test_keep_client.py
+++ b/sdk/python/tests/test_keep_client.py
@@ -18,6 +18,7 @@ import re
import shutil
import socket
import sys
+import stat
import tempfile
import time
import unittest
@@ -1584,6 +1585,8 @@ class KeepDiskCacheTestCase(unittest.TestCase, tutil.ApiClientMock):
with tutil.mock_keep_responses(self.data, 200) as mock:
self.assertTrue(tutil.binary_compare(keep_client.get(self.locator), self.data))
+ self.assertIsNotNone(keep_client.get_from_cache(self.locator))
+
with open(os.path.join(self.disk_cache_dir, self.locator[0:3], self.locator+".keepcacheblock"), "rb") as f:
self.assertTrue(tutil.binary_compare(f.read(), self.data))
@@ -1672,10 +1675,35 @@ class KeepDiskCacheTestCase(unittest.TestCase, tutil.ApiClientMock):
self.assertTrue(os.path.exists(os.path.join(self.disk_cache_dir, self.locator[0:3], self.locator+".keepcacheblock")))
self.assertTrue(os.path.exists(os.path.join(self.disk_cache_dir, "acb", "acbd18db4cc2f85cedef654fccc4a4d8.keepcacheblock")))
-
block_cache2 = arvados.keep.KeepBlockCache(disk_cache=True,
disk_cache_dir=self.disk_cache_dir,
max_slots=1)
self.assertTrue(os.path.exists(os.path.join(self.disk_cache_dir, self.locator[0:3], self.locator+".keepcacheblock")))
self.assertTrue(os.path.exists(os.path.join(self.disk_cache_dir, "acb", "acbd18db4cc2f85cedef654fccc4a4d8.keepcacheblock")))
+
+
+
+ def test_disk_cache_error(self):
+ os.chmod(self.disk_cache_dir, stat.S_IRUSR)
+
+ # Fail during cache initialization.
+ with self.assertRaises(OSError):
+ block_cache = arvados.keep.KeepBlockCache(disk_cache=True,
+ disk_cache_dir=self.disk_cache_dir)
+
+
+ def test_disk_cache_write_error(self):
+ block_cache = arvados.keep.KeepBlockCache(disk_cache=True,
+ disk_cache_dir=self.disk_cache_dir)
+
+ keep_client = arvados.KeepClient(api_client=self.api_client, block_cache=block_cache)
+
+ # Make the cache dir read-only
+ os.makedirs(os.path.join(self.disk_cache_dir, self.locator[0:3]))
+ os.chmod(os.path.join(self.disk_cache_dir, self.locator[0:3]), stat.S_IRUSR)
+
+ # Cache fails
+ with self.assertRaises(arvados.errors.KeepCacheError):
+ with tutil.mock_keep_responses(self.data, 200) as mock:
+ keep_client.get(self.locator)
-----------------------------------------------------------------------
hooks/post-receive
--
More information about the arvados-commits
mailing list