Add unit test to clean_caches()

client/isolateserver.py

 #!/usr/bin/env python
 # Copyright 2013 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.
 
 """Archives a set of files or directories to an Isolate Server."""
 
 __version__ = '0.8.0'
 
 import errno
@@ skipping 962 matching lines @@
     """
     raise NotImplementedError()
 
   def write(self, digest, content):
     """Reads data from |content| generator and stores it in cache.
 
     Returns digest to simplify chaining.
     """
     raise NotImplementedError()
 
+  def trim(self):
+    """Enforces cache policies.
+
+    Returns:
+      Number of items evicted.
+    """
+    raise NotImplementedError()
+
 
 class MemoryCache(LocalCache):
   """LocalCache 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(MemoryCache, self).__init__()
@@ skipping 31 matching lines @@
     return io.BytesIO(d)
 
   def write(self, digest, content):
     # Assemble whole stream before taking the lock.
     data = ''.join(content)
     with self._lock:
       self._contents[digest] = data
       self._added.append(len(data))
     return digest
 
+  def trim(self):
+    """Trimming is not implemented for MemoryCache."""
+    return 0
+
 
 class CachePolicies(object):
   def __init__(self, max_cache_size, min_free_space, max_items):
     """
     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 unconditionally fill the disk.
     - max_items: Maximum number of items to keep in the cache. If 0, do not
                  enforce a limit.
     """
     self.max_cache_size = max_cache_size
     self.min_free_space = min_free_space
     self.max_items = max_items
 
 
 class DiskCache(LocalCache):
   """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, hash_algo, trim=True):
+  def __init__(self, cache_dir, policies, hash_algo, trim, time_fn=None):
     """
     Arguments:
       cache_dir: directory where to place the cache.
       policies: cache retention policies.
       algo: hashing algorithm used.
       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(DiskCache, self).__init__()
     self.cache_dir = cache_dir
     self.policies = policies
     self.hash_algo = hash_algo
     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=trim)
+        self._load(trim, time_fn)
 
   def __contains__(self, digest):
     with self._lock:
       return digest in self._lru
 
   def __enter__(self):
     return self
 
   def __exit__(self, _exc_type, _exec_value, _traceback):
     with tools.Profiler('CleanupTrimming'):
@@ skipping 51 matching lines @@
           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 50Gb 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:
@@ skipping 78 matching lines @@
   def get_timestamp(self, digest):
     """Returns timestamp of last use of an item.
 
     Raises KeyError if item is not found.
     """
     return self._lru.get_timestamp(digest)
 
   def trim(self):
     """Forces retention policies."""
     with self._lock:
-      self._trim()
+      return self._trim()
 
-  def _load(self, trim):
+  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 os.path.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()
     # We want the initial cache size after trimming, i.e. what is readily
     # avaiable.
     self._initial_number_items = len(self._lru)
     self._initial_size = sum(self._lru.itervalues())
     if self._evicted:
       logging.info(
           'Trimming evicted items with the following sizes: %s',
           sorted(self._evicted))
@@ skipping 43 matching lines @@
 
       logging.warning(
           'Trimmed %s file(s) due to not enough free disk space: %.1fkb free,'
           ' %.1fkb cache (%.1f%% of its maximum capacity of %.1fkb)',
           trimmed_due_to_space,
           self._free_disk / 1024.,
           total_usage / 1024.,
           usage_percent,
           self.policies.max_cache_size / 1024.)
     self._save()
+    return trimmed_due_to_space
 
   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."""
     self._lock.assert_locked()
     try:
       digest, (size, _) = self._lru.get_oldest()
       if not allow_protected and digest == self._protected:
-        raise Error('Not enough space to fetch the whole isolated tree')
+        raise Error(
+            'Not enough space to fetch the whole isolated tree; %sb free, min '
+            'is %sb' % (self._free_disk, self.policies.min_free_space))
     except KeyError:
       raise Error('Nothing to remove')
     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()
@@ skipping 622 matching lines @@
   cache_group.add_option(
       '--max-items',
       type='int',
       metavar='NNN',
       default=100000,
       help='Trim if more than this number of items are in the cache '
            'default=%default')
   parser.add_option_group(cache_group)
 
 
-def process_cache_options(options, trim=True):
+def process_cache_options(options, **kwargs):
   if options.cache:
     policies = CachePolicies(
         options.max_cache_size, options.min_free_space, options.max_items)
 
     # |options.cache| path may not exist until DiskCache() instance is created.
     return DiskCache(
         unicode(os.path.abspath(options.cache)),
         policies,
         isolated_format.get_hash_algo(options.namespace),
-        trim=trim)
+        **kwargs)
   else:
     return MemoryCache()
 
 
 class OptionParserIsolateServer(logging_utils.OptionParserWithLogging):
   def __init__(self, **kwargs):
     logging_utils.OptionParserWithLogging.__init__(
         self,
         version=__version__,
         prog=os.path.basename(sys.modules[__name__].__file__),
@@ skipping 12 matching lines @@
   return dispatcher.execute(OptionParserIsolateServer(), args)
 
 
 if __name__ == '__main__':
   subprocess42.inhibit_os_error_reporting()
   fix_encoding.fix_encoding()
   tools.disable_buffering()
   colorama.init()
   file_path.enable_symlink()
   sys.exit(main(sys.argv[1:]))