| # Copyright 2018 The LUCI Authors. All rights reserved. |
| # Use of this source code is governed under the Apache License, Version 2.0 |
| # that can be found in the LICENSE file. |
| |
| """Define local cache policies.""" |
| |
| import contextlib |
| import errno |
| import io |
| import logging |
| import os |
| import random |
| import string |
| import sys |
| import time |
| |
| from utils import file_path |
| from utils import fs |
| from utils import lru |
| from utils import threading_utils |
| from utils import tools |
| |
| |
| # The file size to be used when we don't know the correct file size, |
| # generally used for .isolated files. |
| UNKNOWN_FILE_SIZE = None |
| |
| |
| def file_write(path, content_generator): |
| """Writes file content as generated by content_generator. |
| |
| Creates the intermediary directory as needed. |
| |
| Returns the number of bytes written. |
| |
| Meant to be mocked out in unit tests. |
| """ |
| file_path.ensure_tree(os.path.dirname(path)) |
| total = 0 |
| with fs.open(path, 'wb') as f: |
| for d in content_generator: |
| total += len(d) |
| f.write(d) |
| return total |
| |
| |
| def is_valid_file(path, size): |
| """Returns if the given files appears valid. |
| |
| Currently it just checks the file exists and its size matches the expectation. |
| """ |
| if size == UNKNOWN_FILE_SIZE: |
| return fs.isfile(path) |
| try: |
| actual_size = fs.stat(path).st_size |
| except OSError as e: |
| logging.warning( |
| 'Can\'t read item %s, assuming it\'s invalid: %s', |
| os.path.basename(path), e) |
| return False |
| if size != actual_size: |
| logging.warning( |
| 'Found invalid item %s; %d != %d', |
| os.path.basename(path), actual_size, size) |
| return False |
| return True |
| |
| |
| def trim_caches(caches, path, min_free_space, max_age_secs): |
| """Trims multiple caches. |
| |
| The goal here is to coherently trim all caches in a coherent LRU fashion, |
| deleting older items independent of which container they belong to. |
| |
| Two policies are enforced first: |
| - max_age_secs |
| - min_free_space |
| |
| Once that's done, then we enforce each cache's own policies. |
| |
| Returns: |
| Slice containing the size of all items evicted. |
| """ |
| min_ts = time.time() - max_age_secs if max_age_secs else 0 |
| free_disk = file_path.get_free_space(path) if min_free_space else 0 |
| total = [] |
| if min_ts or free_disk: |
| while True: |
| oldest = [(c, c.get_oldest()) for c in caches if len(c) > 0] |
| if not oldest: |
| break |
| oldest.sort(key=lambda (_, ts): ts) |
| c, ts = oldest[0] |
| if ts >= min_ts and free_disk >= min_free_space: |
| break |
| total.append(c.remove_oldest()) |
| if min_free_space: |
| free_disk = file_path.get_free_space(path) |
| # Evaluate each cache's own policies. |
| for c in caches: |
| total.extend(c.trim()) |
| return total |
| |
| |
| def _get_recursive_size(path): |
| """Returns the total data size for the specified path. |
| |
| This function can be surprisingly slow on OSX, so its output should be cached. |
| """ |
| try: |
| total = 0 |
| for root, _, files in fs.walk(path): |
| for f in files: |
| total += fs.lstat(os.path.join(root, f)).st_size |
| return total |
| except (IOError, OSError, UnicodeEncodeError) as exc: |
| logging.warning('Exception while getting the size of %s:\n%s', path, exc) |
| return None |
| |
| |
| class NamedCacheError(Exception): |
| """Named cache specific error.""" |
| |
| |
| class NoMoreSpace(Exception): |
| """Not enough space to map the whole directory.""" |
| pass |
| |
| |
| class CachePolicies(object): |
| def __init__(self, max_cache_size, min_free_space, max_items, max_age_secs): |
| """Common caching policies for the multiple caches (isolated, named, cipd). |
| |
| Arguments: |
| - max_cache_size: Trim if the cache gets larger than this value. If 0, the |
| cache is effectively a leak. |
| - min_free_space: Trim if disk free space becomes lower than this value. If |
| 0, it will unconditionally fill the disk. |
| - max_items: Maximum number of items to keep in the cache. If 0, do not |
| enforce a limit. |
| - max_age_secs: Maximum age an item is kept in the cache until it is |
| automatically evicted. Having a lot of dead luggage slows |
| everything down. |
| """ |
| self.max_cache_size = max_cache_size |
| self.min_free_space = min_free_space |
| self.max_items = max_items |
| self.max_age_secs = max_age_secs |
| |
| def __str__(self): |
| return ( |
| 'CachePolicies(max_cache_size=%s; max_items=%s; min_free_space=%s; ' |
| 'max_age_secs=%s)') % ( |
| self.max_cache_size, self.max_items, self.min_free_space, |
| self.max_age_secs) |
| |
| |
| class CacheMiss(Exception): |
| """Raised when an item is not in cache.""" |
| def __init__(self, digest): |
| self.digest = digest |
| super(CacheMiss, self).__init__( |
| 'Item with digest %r is not found in cache' % digest) |
| |
| |
| class Cache(object): |
| def __init__(self, cache_dir): |
| if cache_dir is not None: |
| assert isinstance(cache_dir, unicode), cache_dir |
| assert file_path.isabs(cache_dir), cache_dir |
| self.cache_dir = cache_dir |
| self._lock = threading_utils.LockWithAssert() |
| # Profiling values. |
| self._added = [] |
| self._used = [] |
| |
| def __nonzero__(self): |
| """A cache is always True. |
| |
| Otherwise it falls back to __len__, which is surprising. |
| """ |
| return True |
| |
| def __len__(self): |
| """Returns the number of entries in the cache.""" |
| raise NotImplementedError() |
| |
| def __iter__(self): |
| """Iterates over all the entries names.""" |
| raise NotImplementedError() |
| |
| def __contains__(self, name): |
| """Returns if an entry is in the cache.""" |
| raise NotImplementedError() |
| |
| @property |
| def total_size(self): |
| """Returns the total size of the cache in bytes.""" |
| raise NotImplementedError() |
| |
| @property |
| def added(self): |
| """Returns a list of the size for each entry added.""" |
| with self._lock: |
| return self._added[:] |
| |
| @property |
| def used(self): |
| """Returns a list of the size for each entry used.""" |
| with self._lock: |
| return self._used[:] |
| |
| def get_oldest(self): |
| """Returns timestamp of oldest cache entry or None. |
| |
| Returns: |
| Timestamp of the oldest item. |
| |
| Used for manual trimming. |
| """ |
| raise NotImplementedError() |
| |
| def remove_oldest(self): |
| """Removes the oldest item from the cache. |
| |
| Returns: |
| Size of the oldest item. |
| |
| Used for manual trimming. |
| """ |
| raise NotImplementedError() |
| |
| def save(self): |
| """Saves the current cache to disk.""" |
| raise NotImplementedError() |
| |
| def trim(self): |
| """Enforces cache policies, then calls save(). |
| |
| Returns: |
| Slice with the size of evicted items. |
| """ |
| raise NotImplementedError() |
| |
| def cleanup(self): |
| """Deletes any corrupted item from the cache, then calls trim(), then |
| save(). |
| |
| It is assumed to take significantly more time than trim(). |
| """ |
| raise NotImplementedError() |
| |
| |
| class ContentAddressedCache(Cache): |
| """Content addressed cache that stores objects temporarily. |
| |
| It can be accessed concurrently from multiple threads, so it should protect |
| its internal state with some lock. |
| """ |
| |
| def __enter__(self): |
| """Context manager interface.""" |
| # TODO(maruel): Remove. |
| return self |
| |
| def __exit__(self, _exc_type, _exec_value, _traceback): |
| """Context manager interface.""" |
| # TODO(maruel): Remove. |
| return False |
| |
| def touch(self, digest, size): |
| """Ensures item is not corrupted and updates its LRU position. |
| |
| Arguments: |
| digest: hash digest of item to check. |
| size: expected size of this item. |
| |
| Returns: |
| True if item is in cache and not corrupted. |
| """ |
| raise NotImplementedError() |
| |
| def getfileobj(self, digest): |
| """Returns a readable file like object. |
| |
| If file exists on the file system it will have a .name attribute with an |
| absolute path to the file. |
| """ |
| raise NotImplementedError() |
| |
| def write(self, digest, content): |
| """Reads data from |content| generator and stores it in cache. |
| |
| It is possible to write to an object that already exists. It may be |
| ignored (sent to /dev/null) but the timestamp is still updated. |
| |
| Returns digest to simplify chaining. |
| """ |
| raise NotImplementedError() |
| |
| |
| class MemoryContentAddressedCache(ContentAddressedCache): |
| """ContentAddressedCache implementation that stores everything in memory.""" |
| |
| def __init__(self, file_mode_mask=0500): |
| """Args: |
| file_mode_mask: bit mask to AND file mode with. Default value will make |
| all mapped files to be read only. |
| """ |
| super(MemoryContentAddressedCache, self).__init__(None) |
| self._file_mode_mask = file_mode_mask |
| # Items in a LRU lookup dict(digest: size). |
| self._lru = lru.LRUDict() |
| |
| # Cache interface implementation. |
| |
| def __len__(self): |
| with self._lock: |
| return len(self._lru) |
| |
| def __iter__(self): |
| # This is not thread-safe. |
| return self._lru.__iter__() |
| |
| def __contains__(self, digest): |
| with self._lock: |
| return digest in self._lru |
| |
| @property |
| def total_size(self): |
| with self._lock: |
| return sum(len(i) for i in self._lru.itervalues()) |
| |
| def get_oldest(self): |
| with self._lock: |
| try: |
| # (key, (value, ts)) |
| return self._lru.get_oldest()[1][1] |
| except KeyError: |
| return None |
| |
| def remove_oldest(self): |
| with self._lock: |
| # TODO(maruel): Update self._added. |
| # (key, (value, ts)) |
| return len(self._lru.pop_oldest()[1][0]) |
| |
| def save(self): |
| pass |
| |
| def trim(self): |
| """Trimming is not implemented for MemoryContentAddressedCache.""" |
| return [] |
| |
| def cleanup(self): |
| """Cleaning is irrelevant, as there's no stateful serialization.""" |
| pass |
| |
| # ContentAddressedCache interface implementation. |
| |
| def __contains__(self, digest): |
| with self._lock: |
| return digest in self._lru |
| |
| def touch(self, digest, size): |
| with self._lock: |
| try: |
| self._lru.touch(digest) |
| except KeyError: |
| return False |
| return True |
| |
| def getfileobj(self, digest): |
| with self._lock: |
| try: |
| d = self._lru[digest] |
| except KeyError: |
| raise CacheMiss(digest) |
| self._used.append(len(d)) |
| self._lru.touch(digest) |
| return io.BytesIO(d) |
| |
| def write(self, digest, content): |
| # Assemble whole stream before taking the lock. |
| data = ''.join(content) |
| with self._lock: |
| self._lru.add(digest, data) |
| self._added.append(len(data)) |
| return digest |
| |
| |
| class DiskContentAddressedCache(ContentAddressedCache): |
| """Stateful LRU cache in a flat hash table in a directory. |
| |
| Saves its state as json file. |
| """ |
| STATE_FILE = u'state.json' |
| |
| def __init__(self, cache_dir, policies, trim, time_fn=None): |
| """ |
| Arguments: |
| cache_dir: directory where to place the cache. |
| policies: CachePolicies instance, cache retention policies. |
| trim: if True to enforce |policies| right away. |
| It can be done later by calling trim() explicitly. |
| """ |
| # All protected methods (starting with '_') except _path should be called |
| # with self._lock held. |
| super(DiskContentAddressedCache, self).__init__(cache_dir) |
| self.policies = policies |
| self.state_file = os.path.join(cache_dir, self.STATE_FILE) |
| # Items in a LRU lookup dict(digest: size). |
| self._lru = lru.LRUDict() |
| # Current cached free disk space. It is updated by self._trim(). |
| file_path.ensure_tree(self.cache_dir) |
| self._free_disk = file_path.get_free_space(self.cache_dir) |
| # The first item in the LRU cache that must not be evicted during this run |
| # since it was referenced. All items more recent that _protected in the LRU |
| # cache are also inherently protected. It could be a set() of all items |
| # referenced but this increases memory usage without a use case. |
| self._protected = None |
| # Cleanup operations done by self._load(), if any. |
| self._operations = [] |
| with tools.Profiler('Setup'): |
| with self._lock: |
| self._load(trim, time_fn) |
| |
| # Cache interface implementation. |
| |
| def __len__(self): |
| with self._lock: |
| return len(self._lru) |
| |
| def __iter__(self): |
| # This is not thread-safe. |
| return self._lru.__iter__() |
| |
| def __contains__(self, digest): |
| with self._lock: |
| return digest in self._lru |
| |
| @property |
| def total_size(self): |
| with self._lock: |
| return sum(self._lru.itervalues()) |
| |
| def get_oldest(self): |
| with self._lock: |
| try: |
| # (key, (value, ts)) |
| return self._lru.get_oldest()[1][1] |
| except KeyError: |
| return None |
| |
| def remove_oldest(self): |
| with self._lock: |
| # TODO(maruel): Update self._added. |
| return self._remove_lru_file(True) |
| |
| def save(self): |
| with self._lock: |
| return self._save() |
| |
| def trim(self): |
| """Forces retention policies.""" |
| with self._lock: |
| return self._trim() |
| |
| def cleanup(self): |
| """Cleans up the cache directory. |
| |
| Ensures there is no unknown files in cache_dir. |
| Ensures the read-only bits are set correctly. |
| |
| At that point, the cache was already loaded, trimmed to respect cache |
| policies. |
| """ |
| with self._lock: |
| fs.chmod(self.cache_dir, 0700) |
| # Ensure that all files listed in the state still exist and add new ones. |
| previous = set(self._lru) |
| # It'd be faster if there were a readdir() function. |
| for filename in fs.listdir(self.cache_dir): |
| if filename == self.STATE_FILE: |
| fs.chmod(os.path.join(self.cache_dir, filename), 0600) |
| continue |
| if filename in previous: |
| fs.chmod(os.path.join(self.cache_dir, filename), 0400) |
| previous.remove(filename) |
| continue |
| |
| # An untracked file. Delete it. |
| logging.warning('Removing unknown file %s from cache', filename) |
| p = self._path(filename) |
| if fs.isdir(p): |
| try: |
| file_path.rmtree(p) |
| except OSError: |
| pass |
| else: |
| file_path.try_remove(p) |
| continue |
| |
| if previous: |
| # Filter out entries that were not found. |
| logging.warning('Removed %d lost files', len(previous)) |
| for filename in previous: |
| self._lru.pop(filename) |
| self._save() |
| |
| # What remains to be done is to hash every single item to |
| # detect corruption, then save to ensure state.json is up to date. |
| # Sadly, on a 50GiB cache with 100MiB/s I/O, this is still over 8 minutes. |
| # TODO(maruel): Let's revisit once directory metadata is stored in |
| # state.json so only the files that had been mapped since the last cleanup() |
| # call are manually verified. |
| # |
| #with self._lock: |
| # for digest in self._lru: |
| # if not isolated_format.is_valid_hash( |
| # self._path(digest), self.hash_algo): |
| # self.evict(digest) |
| # logging.info('Deleted corrupted item: %s', digest) |
| |
| # ContentAddressedCache interface implementation. |
| |
| def __contains__(self, digest): |
| with self._lock: |
| return digest in self._lru |
| |
| def touch(self, digest, size): |
| """Verifies an actual file is valid and bumps its LRU position. |
| |
| Returns False if the file is missing or invalid. |
| |
| Note that is doesn't compute the hash so it could still be corrupted if the |
| file size didn't change. |
| """ |
| # Do the check outside the lock. |
| looks_valid = is_valid_file(self._path(digest), size) |
| |
| # Update its LRU position. |
| with self._lock: |
| if digest not in self._lru: |
| if looks_valid: |
| # Exists but not in the LRU anymore. |
| self._delete_file(digest, size) |
| return False |
| if not looks_valid: |
| self._lru.pop(digest) |
| # Exists but not in the LRU anymore. |
| self._delete_file(digest, size) |
| return False |
| self._lru.touch(digest) |
| self._protected = self._protected or digest |
| return True |
| |
| def getfileobj(self, digest): |
| try: |
| f = fs.open(self._path(digest), 'rb') |
| except IOError: |
| raise CacheMiss(digest) |
| with self._lock: |
| try: |
| self._used.append(self._lru[digest]) |
| except KeyError: |
| # If the digest is not actually in _lru, assume it is a cache miss. |
| # Existing file will be overwritten by whoever uses the cache and added |
| # to _lru. |
| f.close() |
| raise CacheMiss(digest) |
| return f |
| |
| def write(self, digest, content): |
| assert content is not None |
| with self._lock: |
| self._protected = self._protected or digest |
| path = self._path(digest) |
| # A stale broken file may remain. It is possible for the file to have write |
| # access bit removed which would cause the file_write() call to fail to open |
| # in write mode. Take no chance here. |
| file_path.try_remove(path) |
| try: |
| size = file_write(path, content) |
| except: |
| # There are two possible places were an exception can occur: |
| # 1) Inside |content| generator in case of network or unzipping errors. |
| # 2) Inside file_write itself in case of disk IO errors. |
| # In any case delete an incomplete file and propagate the exception to |
| # caller, it will be logged there. |
| file_path.try_remove(path) |
| raise |
| # Make the file read-only in the cache. This has a few side-effects since |
| # the file node is modified, so every directory entries to this file becomes |
| # read-only. It's fine here because it is a new file. |
| file_path.set_read_only(path, True) |
| with self._lock: |
| self._add(digest, size) |
| return digest |
| |
| # Internal functions. |
| |
| def _load(self, trim, time_fn): |
| """Loads state of the cache from json file. |
| |
| If cache_dir does not exist on disk, it is created. |
| """ |
| self._lock.assert_locked() |
| |
| if not fs.isfile(self.state_file): |
| if not fs.isdir(self.cache_dir): |
| fs.makedirs(self.cache_dir) |
| else: |
| # Load state of the cache. |
| try: |
| self._lru = lru.LRUDict.load(self.state_file) |
| except ValueError as err: |
| logging.error('Failed to load cache state: %s' % (err,)) |
| # Don't want to keep broken state file. |
| file_path.try_remove(self.state_file) |
| if time_fn: |
| self._lru.time_fn = time_fn |
| if trim: |
| self._trim() |
| |
| def _save(self): |
| """Saves the LRU ordering.""" |
| self._lock.assert_locked() |
| if sys.platform != 'win32': |
| d = os.path.dirname(self.state_file) |
| if fs.isdir(d): |
| # Necessary otherwise the file can't be created. |
| file_path.set_read_only(d, False) |
| if fs.isfile(self.state_file): |
| file_path.set_read_only(self.state_file, False) |
| self._lru.save(self.state_file) |
| |
| def _trim(self): |
| """Trims anything we don't know, make sure enough free space exists.""" |
| self._lock.assert_locked() |
| evicted = [] |
| |
| # Trim old items. |
| if self.policies.max_age_secs: |
| cutoff = self._lru.time_fn() - self.policies.max_age_secs |
| while self._lru: |
| oldest = self._lru.get_oldest() |
| # (key, (data, ts) |
| if oldest[1][1] >= cutoff: |
| break |
| evicted.append(self._remove_lru_file(True)) |
| |
| # Ensure maximum cache size. |
| if self.policies.max_cache_size: |
| total_size = sum(self._lru.itervalues()) |
| while total_size > self.policies.max_cache_size: |
| e = self._remove_lru_file(True) |
| evicted.append(e) |
| total_size -= e |
| |
| # Ensure maximum number of items in the cache. |
| if self.policies.max_items and len(self._lru) > self.policies.max_items: |
| for _ in xrange(len(self._lru) - self.policies.max_items): |
| evicted.append(self._remove_lru_file(True)) |
| |
| # Ensure enough free space. |
| self._free_disk = file_path.get_free_space(self.cache_dir) |
| while ( |
| self.policies.min_free_space and |
| self._lru and |
| self._free_disk < self.policies.min_free_space): |
| # self._free_disk is updated by this call. |
| evicted.append(self._remove_lru_file(True)) |
| |
| if evicted: |
| total_usage = sum(self._lru.itervalues()) |
| usage_percent = 0. |
| if total_usage: |
| usage_percent = 100. * float(total_usage) / self.policies.max_cache_size |
| |
| logging.warning( |
| 'Trimmed %d file(s) (%.1fkb) due to not enough free disk space:' |
| ' %.1fkb free, %.1fkb cache (%.1f%% of its maximum capacity of ' |
| '%.1fkb)', |
| len(evicted), sum(evicted) / 1024., |
| self._free_disk / 1024., |
| total_usage / 1024., |
| usage_percent, |
| self.policies.max_cache_size / 1024.) |
| self._save() |
| return evicted |
| |
| def _path(self, digest): |
| """Returns the path to one item.""" |
| return os.path.join(self.cache_dir, digest) |
| |
| def _remove_lru_file(self, allow_protected): |
| """Removes the lastest recently used file and returns its size. |
| |
| Updates self._free_disk. |
| """ |
| self._lock.assert_locked() |
| try: |
| digest, (size, _ts) = self._lru.get_oldest() |
| if not allow_protected and digest == self._protected: |
| total_size = sum(self._lru.itervalues())+size |
| msg = ( |
| 'Not enough space to fetch the whole isolated tree.\n' |
| ' %s\n cache=%dbytes, %d items; %sb free_space') % ( |
| self.policies, total_size, len(self._lru)+1, self._free_disk) |
| raise NoMoreSpace(msg) |
| except KeyError: |
| # That means an internal error. |
| raise NoMoreSpace('Nothing to remove, can\'t happend') |
| digest, (size, _) = self._lru.pop_oldest() |
| logging.debug('Removing LRU file %s', digest) |
| self._delete_file(digest, size) |
| return size |
| |
| def _add(self, digest, size=UNKNOWN_FILE_SIZE): |
| """Adds an item into LRU cache marking it as a newest one.""" |
| self._lock.assert_locked() |
| if size == UNKNOWN_FILE_SIZE: |
| size = fs.stat(self._path(digest)).st_size |
| self._added.append(size) |
| self._lru.add(digest, size) |
| self._free_disk -= size |
| # Do a quicker version of self._trim(). It only enforces free disk space, |
| # not cache size limits. It doesn't actually look at real free disk space, |
| # only uses its cache values. self._trim() will be called later to enforce |
| # real trimming but doing this quick version here makes it possible to map |
| # an isolated that is larger than the current amount of free disk space when |
| # the cache size is already large. |
| while ( |
| self.policies.min_free_space and |
| self._lru and |
| self._free_disk < self.policies.min_free_space): |
| # self._free_disk is updated by this call. |
| if self._remove_lru_file(False) == -1: |
| break |
| |
| def _delete_file(self, digest, size=UNKNOWN_FILE_SIZE): |
| """Deletes cache file from the file system. |
| |
| Updates self._free_disk. |
| """ |
| self._lock.assert_locked() |
| try: |
| if size == UNKNOWN_FILE_SIZE: |
| try: |
| size = fs.stat(self._path(digest)).st_size |
| except OSError: |
| size = 0 |
| if file_path.try_remove(self._path(digest)): |
| self._free_disk += size |
| except OSError as e: |
| if e.errno != errno.ENOENT: |
| logging.error('Error attempting to delete a file %s:\n%s' % (digest, e)) |
| |
| |
| class NamedCache(Cache): |
| """Manages cache directories. |
| |
| A cache entry is a tuple (name, path), where |
| name is a short identifier that describes the contents of the cache, e.g. |
| "git_v8" could be all git repositories required by v8 builds, or |
| "build_chromium" could be build artefacts of the Chromium. |
| path is a directory path relative to the task run dir. Cache installation |
| puts the requested cache directory at the path. |
| """ |
| _DIR_ALPHABET = string.ascii_letters + string.digits |
| STATE_FILE = u'state.json' |
| NAMED_DIR = u'named' |
| |
| def __init__(self, cache_dir, policies, time_fn=None): |
| """Initializes NamedCaches. |
| |
| Arguments: |
| - cache_dir is a directory for persistent cache storage. |
| - policies is a CachePolicies instance. |
| - time_fn is a function that returns timestamp (float) and used to take |
| timestamps when new caches are requested. Used in unit tests. |
| """ |
| super(NamedCache, self).__init__(cache_dir) |
| self._policies = policies |
| # LRU {cache_name -> tuple(cache_location, size)} |
| self.state_file = os.path.join(cache_dir, self.STATE_FILE) |
| self._lru = lru.LRUDict() |
| if not fs.isdir(self.cache_dir): |
| fs.makedirs(self.cache_dir) |
| elif fs.isfile(self.state_file): |
| try: |
| self._lru = lru.LRUDict.load(self.state_file) |
| except ValueError: |
| logging.exception( |
| 'NamedCache: failed to load named cache state file; obliterating') |
| file_path.rmtree(self.cache_dir) |
| with self._lock: |
| self._try_upgrade() |
| if time_fn: |
| self._lru.time_fn = time_fn |
| |
| @property |
| def available(self): |
| """Returns a set of names of available caches.""" |
| with self._lock: |
| return set(self._lru) |
| |
| def install(self, dst, name): |
| """Creates the directory |dst| and moves a previous named cache |name| if it |
| was in the local named caches cache. |
| |
| dst must be absolute, unicode and must not exist. |
| |
| Returns the reused named cache size in bytes, or 0 if none was present. |
| |
| Raises NamedCacheError if cannot install the cache. |
| """ |
| logging.info('NamedCache.install(%r, %r)', dst, name) |
| with self._lock: |
| try: |
| if fs.isdir(dst): |
| raise NamedCacheError( |
| 'installation directory %r already exists' % dst) |
| |
| # Remove the named symlink if it exists. |
| link_name = self._get_named_path(name) |
| if fs.exists(link_name): |
| # Remove the symlink itself, not its destination. |
| fs.remove(link_name) |
| |
| if name in self._lru: |
| rel_cache, size = self._lru.get(name) |
| abs_cache = os.path.join(self.cache_dir, rel_cache) |
| if fs.isdir(abs_cache): |
| logging.info('- reusing %r; size was %d', rel_cache, size) |
| file_path.ensure_tree(os.path.dirname(dst)) |
| fs.rename(abs_cache, dst) |
| self._remove(name) |
| return size |
| |
| logging.warning('- expected directory %r, does not exist', rel_cache) |
| self._remove(name) |
| |
| # The named cache does not exist, create an empty directory. When |
| # uninstalling, we will move it back to the cache and create an an |
| # entry. |
| logging.info('- creating new directory') |
| file_path.ensure_tree(dst) |
| return 0 |
| except (IOError, OSError) as ex: |
| # Raise using the original traceback. |
| exc = NamedCacheError( |
| 'cannot install cache named %r at %r: %s' % (name, dst, ex)) |
| raise exc, None, sys.exc_info()[2] |
| finally: |
| self._save() |
| |
| def uninstall(self, src, name): |
| """Moves the cache directory back into the named cache hive for an eventual |
| reuse. |
| |
| The opposite of install(). |
| |
| src must be absolute and unicode. Its content is moved back into the local |
| named caches cache. |
| |
| Returns the named cache size in bytes. |
| |
| Raises NamedCacheError if cannot uninstall the cache. |
| """ |
| logging.info('NamedCache.uninstall(%r, %r)', src, name) |
| with self._lock: |
| try: |
| if not fs.isdir(src): |
| logging.warning( |
| 'NamedCache: Directory %r does not exist anymore. Cache lost.', |
| src) |
| return |
| |
| if name in self._lru: |
| # This shouldn't happen but just remove the preexisting one and move |
| # on. |
| logging.error('- overwriting existing cache!') |
| self._remove(name) |
| |
| # Calculate the size of the named cache to keep. It's important because |
| # if size is zero (it's empty), we do not want to add it back to the |
| # named caches cache. |
| size = _get_recursive_size(src) |
| logging.info('- Size is %d', size) |
| if not size: |
| # Do not save empty named cache. |
| return size |
| |
| # Move the dir and create an entry for the named cache. |
| rel_cache = self._allocate_dir() |
| abs_cache = os.path.join(self.cache_dir, rel_cache) |
| logging.info('- Moving to %r', rel_cache) |
| file_path.ensure_tree(os.path.dirname(abs_cache)) |
| fs.rename(src, abs_cache) |
| |
| self._lru.add(name, (rel_cache, size)) |
| self._added.append(size) |
| |
| # Create symlink <cache_dir>/<named>/<name> -> <cache_dir>/<short name> |
| # for user convenience. |
| named_path = self._get_named_path(name) |
| if fs.exists(named_path): |
| file_path.remove(named_path) |
| else: |
| file_path.ensure_tree(os.path.dirname(named_path)) |
| |
| try: |
| fs.symlink(os.path.join(u'..', rel_cache), named_path) |
| logging.info( |
| 'NamedCache: Created symlink %r to %r', named_path, abs_cache) |
| except OSError: |
| # Ignore on Windows. It happens when running as a normal user or when |
| # UAC is enabled and the user is a filtered administrator account. |
| if sys.platform != 'win32': |
| raise |
| return size |
| except (IOError, OSError) as ex: |
| # Raise using the original traceback. |
| exc = NamedCacheError( |
| 'cannot uninstall cache named %r at %r: %s' % (name, src, ex)) |
| raise exc, None, sys.exc_info()[2] |
| finally: |
| # Call save() at every uninstall. The assumptions are: |
| # - The total the number of named caches is low, so the state.json file |
| # is small, so the time it takes to write it to disk is short. |
| # - The number of mapped named caches per task is low, so the number of |
| # times save() is called on tear-down isn't high enough to be |
| # significant. |
| # - uninstall() sometimes throws due to file locking on Windows or |
| # access rights on Linux. We want to keep as many as possible. |
| self._save() |
| |
| # Cache interface implementation. |
| |
| def __len__(self): |
| with self._lock: |
| return len(self._lru) |
| |
| def __iter__(self): |
| # This is not thread-safe. |
| return self._lru.__iter__() |
| |
| def __contains__(self, digest): |
| with self._lock: |
| return digest in self._lru |
| |
| @property |
| def total_size(self): |
| with self._lock: |
| return sum(size for _rel_path, size in self._lru.itervalues()) |
| |
| def get_oldest(self): |
| with self._lock: |
| try: |
| # (key, (value, ts)) |
| return self._lru.get_oldest()[1][1] |
| except KeyError: |
| return None |
| |
| def remove_oldest(self): |
| with self._lock: |
| # TODO(maruel): Update self._added. |
| _name, size = self._remove_lru_item() |
| return size |
| |
| def save(self): |
| with self._lock: |
| return self._save() |
| |
| def trim(self): |
| evicted = [] |
| with self._lock: |
| if not fs.isdir(self.cache_dir): |
| return evicted |
| |
| # Trim according to maximum number of items. |
| if self._policies.max_items: |
| while len(self._lru) > self._policies.max_items: |
| name, size = self._remove_lru_item() |
| evicted.append(size) |
| logging.info( |
| 'NamedCache.trim(): Removed %r(%d) due to max_items(%d)', |
| name, size, self._policies.max_items) |
| |
| # Trim according to maximum age. |
| if self._policies.max_age_secs: |
| cutoff = self._lru.time_fn() - self._policies.max_age_secs |
| while self._lru: |
| _name, (_data, ts) = self._lru.get_oldest() |
| if ts >= cutoff: |
| break |
| name, size = self._remove_lru_item() |
| evicted.append(size) |
| logging.info( |
| 'NamedCache.trim(): Removed %r(%d) due to max_age_secs(%d)', |
| name, size, self._policies.max_age_secs) |
| |
| # Trim according to minimum free space. |
| if self._policies.min_free_space: |
| while self._lru: |
| free_space = file_path.get_free_space(self.cache_dir) |
| if free_space >= self._policies.min_free_space: |
| break |
| name, size = self._remove_lru_item() |
| evicted.append(size) |
| logging.info( |
| 'NamedCache.trim(): Removed %r(%d) due to min_free_space(%d)', |
| name, size, self._policies.min_free_space) |
| |
| # Trim according to maximum total size. |
| if self._policies.max_cache_size: |
| while self._lru: |
| total = sum(size for _rel_cache, size in self._lru.itervalues()) |
| if total <= self._policies.max_cache_size: |
| break |
| name, size = self._remove_lru_item() |
| evicted.append(size) |
| logging.info( |
| 'NamedCache.trim(): Removed %r(%d) due to max_cache_size(%d)', |
| name, size, self._policies.max_cache_size) |
| |
| self._save() |
| return evicted |
| |
| def cleanup(self): |
| """Removes unknown directories. |
| |
| Does not recalculate the cache size since it's surprisingly slow on some |
| OSes. |
| """ |
| success = True |
| with self._lock: |
| try: |
| actual = set(fs.listdir(self.cache_dir)) |
| actual.discard(self.NAMED_DIR) |
| actual.discard(self.STATE_FILE) |
| expected = {v[0]: k for k, v in self._lru.iteritems()} |
| # First, handle the actual cache content. |
| # Remove missing entries. |
| for missing in (set(expected) - actual): |
| name, size = self._lru.pop(expected[missing]) |
| logging.warning( |
| 'NamedCache.cleanup(): Missing on disk %r(%d)', name, size) |
| # Remove unexpected items. |
| for unexpected in (actual - set(expected)): |
| try: |
| p = os.path.join(self.cache_dir, unexpected) |
| logging.warning( |
| 'NamedCache.cleanup(): Unexpected %r', unexpected) |
| if fs.isdir(p) and not fs.islink(p): |
| file_path.rmtree(p) |
| else: |
| fs.remove(p) |
| except (IOError, OSError) as e: |
| logging.error('Failed to remove %s: %s', unexpected, e) |
| success = False |
| |
| # Second, fix named cache links. |
| named = os.path.join(self.cache_dir, self.NAMED_DIR) |
| if fs.isdir(named): |
| actual = set(fs.listdir(named)) |
| expected = set(self._lru) |
| # Confirm entries. Do not add missing ones for now. |
| for name in expected.intersection(actual): |
| p = os.path.join(self.cache_dir, self.NAMED_DIR, name) |
| expected_link = os.path.join(u'..', self._lru[name][0]) |
| if fs.islink(p): |
| link = fs.readlink(p) |
| if expected_link == link: |
| continue |
| logging.warning( |
| 'Unexpected symlink for cache %s: %s, expected %s', |
| name, link, expected_link) |
| else: |
| logging.warning('Unexpected non symlink for cache %s', name) |
| if fs.isdir(p) and not fs.islink(p): |
| file_path.rmtree(p) |
| else: |
| fs.remove(p) |
| # Remove unexpected items. |
| for unexpected in (actual - expected): |
| try: |
| p = os.path.join(self.cache_dir, self.NAMED_DIR, unexpected) |
| if fs.isdir(p): |
| file_path.rmtree(p) |
| else: |
| fs.remove(p) |
| except (IOError, OSError) as e: |
| logging.error('Failed to remove %s: %s', unexpected, e) |
| success = False |
| finally: |
| self._save() |
| return success |
| |
| # Internal functions. |
| |
| def _try_upgrade(self): |
| """Upgrades from the old format to the new one if necessary. |
| |
| This code can be removed so all bots are known to have the right new format. |
| """ |
| if not self._lru: |
| return |
| _name, (data, _ts) = self._lru.get_oldest() |
| if isinstance(data, (list, tuple)): |
| return |
| # Update to v2. |
| def upgrade(_name, rel_cache): |
| abs_cache = os.path.join(self.cache_dir, rel_cache) |
| return rel_cache, _get_recursive_size(abs_cache) |
| self._lru.transform(upgrade) |
| self._save() |
| |
| def _remove_lru_item(self): |
| """Removes the oldest LRU entry. LRU must not be empty.""" |
| name, ((_rel_path, size), _ts) = self._lru.get_oldest() |
| logging.info('Removing named cache %r', name) |
| self._remove(name) |
| return name, size |
| |
| def _allocate_dir(self): |
| """Creates and returns relative path of a new cache directory. |
| |
| In practice, it is a 2-letter string. |
| """ |
| # We randomly generate directory names that have two lower/upper case |
| # letters or digits. Total number of possibilities is (26*2 + 10)^2 = 3844. |
| abc_len = len(self._DIR_ALPHABET) |
| tried = set() |
| while len(tried) < 1000: |
| i = random.randint(0, abc_len * abc_len - 1) |
| rel_path = ( |
| self._DIR_ALPHABET[i / abc_len] + |
| self._DIR_ALPHABET[i % abc_len]) |
| if rel_path in tried: |
| continue |
| abs_path = os.path.join(self.cache_dir, rel_path) |
| if not fs.exists(abs_path): |
| return rel_path |
| tried.add(rel_path) |
| raise NamedCacheError( |
| 'could not allocate a new cache dir, too many cache dirs') |
| |
| def _remove(self, name): |
| """Removes a cache directory and entry. |
| |
| Returns: |
| Number of caches deleted. |
| """ |
| self._lock.assert_locked() |
| # First try to remove the alias if it exists. |
| named_dir = self._get_named_path(name) |
| if fs.islink(named_dir): |
| fs.unlink(named_dir) |
| |
| # Then remove the actual data. |
| if name not in self._lru: |
| return |
| rel_path, _size = self._lru.get(name) |
| abs_path = os.path.join(self.cache_dir, rel_path) |
| if fs.isdir(abs_path): |
| file_path.rmtree(abs_path) |
| self._lru.pop(name) |
| |
| def _save(self): |
| self._lock.assert_locked() |
| file_path.ensure_tree(self.cache_dir) |
| self._lru.save(self.state_file) |
| |
| def _get_named_path(self, name): |
| return os.path.join(self.cache_dir, self.NAMED_DIR, name) |
