sandwich: Use cachetool's batch mode to speed-up cache processing.

tools/android/loading/chrome_cache.py

 # Copyright 2016 The Chromium Authors. All rights reserved.
 # Use of this source code is governed by a BSD-style license that can be
 # found in the LICENSE file.
 
 """Takes care of manipulating the chrome's HTTP cache.
 """
 
 from datetime import datetime
 import json
 import os
 import re
 import shutil
+import struct
 import subprocess
 import sys
 import tempfile
 import zipfile
 
 _SRC_DIR = os.path.abspath(os.path.join(
     os.path.dirname(__file__), '..', '..', '..'))
 
 sys.path.append(os.path.join(_SRC_DIR, 'build', 'android'))
 from pylib import constants
 
 import device_setup
 import options
 
 
 OPTIONS = options.OPTIONS
 
 
+class CacheBackendType(object):
+  Simple = 'simple'

[mattcary, 2016/07/01 12:09:29]

Constants should be all caps: SIMPLE, BLOCKFILE

[gabadie, 2016/07/01 14:10:25]

Done.

+  Blockfile = 'blockfile'
+
 # Cache back-end types supported by cachetool.
-BACKEND_TYPES = {'simple', 'blockfile'}
+BACKEND_TYPES = {CacheBackendType.Simple, CacheBackendType.Blockfile}
 
 # Regex used to parse HTTP headers line by line.
 HEADER_PARSING_REGEX = re.compile(r'^(?P<header>\S+):(?P<value>.*)$')
 
 
 def _EnsureCleanCacheDirectory(directory_dest_path):
   """Ensure that a cache directory is created and clean.
 
   Args:
     directory_dest_path: Path of the cache directory to ensure cleanliness.
@@ skipping 182 matching lines @@
 
   Args:
     directory_src_path: Path of the cache directory source.
     directory_dest_path: Path of the cache directory destination.
   """
   assert os.path.isdir(directory_src_path)
   _EnsureCleanCacheDirectory(directory_dest_path)
   shutil.copytree(directory_src_path, directory_dest_path)
 
 
+class CacheBackendError(Exception):
+  def __init__(self, errors):
+    Exception.__init__(self, repr(errors))
+    self.errors = errors
+
+
 class CacheBackend(object):
   """Takes care of reading and deleting cached keys.

[mattcary, 2016/07/01 12:09:29]

Add that this can be used as a context manager now, and what it does (ie,
suppress exceptions).

[gabadie, 2016/07/01 14:10:25]

Done.

   """
 
   def __init__(self, cache_directory_path, cache_backend_type):
     """Chrome cache back-end constructor.
 
     Args:
       cache_directory_path: The directory path where the cache is locally
         stored.
       cache_backend_type: A cache back-end type in BACKEND_TYPES.
     """
     assert os.path.isdir(cache_directory_path)
     assert cache_backend_type in BACKEND_TYPES
     self._cache_directory_path = cache_directory_path
     self._cache_backend_type = cache_backend_type
     # Make sure cache_directory_path is a valid cache.
-    self._CachetoolCmd('validate')
+    self._CachetoolCmd('stop')
 
   def GetSize(self):
     """Gets total size of cache entries in bytes."""
     size = self._CachetoolCmd('get_size')
     return int(size.strip())
 
   def ListKeys(self):
     """Lists cache's keys.
 
     Returns:
       A list of all keys stored in the cache.
     """
-    return [k.strip() for k in self._CachetoolCmd('list_keys').split('\n')[:-1]]
+    return [k.strip() for k in self._CachetoolCmd('list_keys').split('\n')[:-2]]

[mattcary, 2016/07/01 12:09:29]

This -2 is a magic constant that's a little worrisome. Should make it a
descriptive constant instead, and possibly assert that the output of cachetool
is expected.

[gabadie, 2016/07/01 14:10:26]

Oh this change is because of a impl change in cachetool. Added  comment.

 
   def GetStreamForKey(self, key, index):
     """Gets a key's stream.
 
     Args:
       key: The key to access the stream.
       index: The stream index:
           index=0 is the HTTP response header;
           index=1 is the transport encoded content;
           index=2 is the compiled content.
@@ skipping 24 matching lines @@
     """Runs the cache editor tool and return the stdout.
 
     Args:
       operation: Cachetool operation.
       args: Additional operation argument to append to the command line.
       stdin: String to pipe to the Cachetool's stdin.
 
     Returns:
       Cachetool's stdout string.
     """
+    args = args or []
     editor_tool_cmd = [
         OPTIONS.LocalBinary('cachetool'),
         self._cache_directory_path,
         self._cache_backend_type,
-        operation]
-    editor_tool_cmd.extend(args or [])
-    process = subprocess.Popen(
-        editor_tool_cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE)
-    stdout_data, _ = process.communicate(input=stdin)
-    assert process.returncode == 0
+        operation] + args
+    process = subprocess.Popen(editor_tool_cmd, stdout=subprocess.PIPE,
+                               stderr=subprocess.PIPE, stdin=subprocess.PIPE)
+    stdout_data, stderr_data = process.communicate(input=stdin)
+    if process.returncode != 0:
+      raise CacheBackendError([([operation] + args, stderr_data.strip())])
     return stdout_data
 
   def UpdateRawResponseHeaders(self, key, raw_headers):
     """Updates a key's raw response headers.
 
     Args:
       key: The key to modify.
       raw_headers: Raw response headers to set.
     """
     self._CachetoolCmd('update_raw_headers', [key], stdin=raw_headers)
@@ skipping 27 matching lines @@
 
     cmd = [OPTIONS.LocalBinary('content_decoder_tool')]
     cmd.extend([s.strip() for s in content_encoding.split(',')])
     process = subprocess.Popen(cmd,
                                stdin=subprocess.PIPE,
                                stdout=subprocess.PIPE)
     decoded_content, _ = process.communicate(input=encoded_content)
     assert process.returncode == 0
     return decoded_content
 
+  def __enter__(self):
+    return self
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    del exc_type, exc_val, exc_tb

[mattcary, 2016/07/01 12:09:29]

Shouldn't you return True if you're going to suppress the exception?

[gabadie, 2016/07/01 14:10:26]

return False (the implicit return None was already enough)

+
+
+class OnlineCacheBackend(object):

[mattcary, 2016/07/01 12:09:29]

Class comment, including context manager semantics.

[gabadie, 2016/07/01 14:10:26]

Done.

+  _INST_IDS = {
+    'stop': 0,
+    'get_size': 1,
+    'list_keys': 2,
+    'get_stream_for_key': 3,
+    'delete_stream': 4,
+    'delete_key': 5,
+    'update_raw_headers': 6
+  }
+
+  def __init__(self, cache_directory_path, cache_backend_type, auto_sync=False):
+    assert os.path.isdir(cache_directory_path)
+    assert cache_backend_type in BACKEND_TYPES
+    self._cache_directory_path = cache_directory_path
+    self._cache_backend_type = cache_backend_type
+    self._in_flight_insts = []
+    self._cachetool_process = None
+    self._cachetool_stdin = None
+    self._cachetool_stdout = None
+    self._auto_sync = auto_sync
+
+  def __enter__(self):
+    self.Start()
+    return self
+
+  def __exit__(self, exc_type, exc_val, exc_tb):
+    del exc_val, exc_tb
+    self.Stop(force_stop=exc_type == CacheBackendError)

[mattcary, 2016/07/01 12:09:29]

return True?

[gabadie, 2016/07/01 14:10:25]

Not really. The point here is to just stop cachetool in any cases, not to catch
and process errors. The point of del is to just mark the argument as unused so
that pylint is happy.

[mattcary, 2016/07/01 14:30:21]

I think prefixing the variables with an underscore is enough to make pylint
happy. That's a little more clear than the del's, I think.

[gabadie, 2016/07/01 14:35:39]

The del ... # unused. is a common pattern in chromium:
https://cs.chromium.org/search/?q=file:.py$+del.*unused&sq=package:chromium&t...

[mattcary, 2016/07/01 20:24:24]

Fair enough, thanks.

+
+  def GetSize(self):
+    self._PushInsts('get_size')
+    self.Sync()
+    return self._UnpackResult('i')[0]
+
+  def ListKeys(self):
+    self._PushInsts('list_keys')
+    self.Sync()
+    keys = []
+    while True:
+      key_size = self._UnpackResult('i')[0]
+      if key_size == 0:
+        break
+      keys.append(self._UnpackResult('{}s'.format(key_size))[0])
+    return keys
+
+  def GetStreamForKey(self, key, index):
+    self._PushInsts('update_raw_headers', str(key), index)
+    self.Sync()
+    stream_size = self._UnpackResult('i')[0]
+    return self._UnpackResult('{}s'.format(stream_size))[0]
+
+  def DeleteStreamForKey(self, key, index):
+    self._PushInsts('delete_stream', str(key), index)
+
+  def DeleteKey(self, key):
+    self._PushInsts('delete_key', str(key))
+
+  def UpdateRawResponseHeaders(self, key, raw_headers):
+    self._PushInsts('update_raw_headers', str(key), raw_headers)
+
+  def Start(self):
+    assert self._cachetool_process == None
+    stdin = os.pipe()
+    stdout = os.pipe()
+    cache_tool_cmd = [
+        OPTIONS.LocalBinary('cachetool'),
+        self._cache_directory_path,
+        self._cache_backend_type,
+        'online']
+    self._cachetool_process = subprocess.Popen(
+        cache_tool_cmd, stdout=subprocess.PIPE, stdin=subprocess.PIPE)
+    os.close(stdin[0])
+    self._cachetool_stdin = stdin[1]
+    self._cachetool_stdout = stdout[0]
+    os.close(stdout[1])
+    assert not self._in_flight_insts
+
+  def Stop(self, force_stop=False):
+    assert self._cachetool_process != None
+    if force_stop:
+      self._cachetool_process.kill()
+      self._cachetool_process.wait()
+      del self._in_flight_insts[:]
+    else:
+      self._PushInsts('stop')
+      self.Sync()
+      self._cachetool_process.wait()
+      assert not self._in_flight_insts
+      assert self._cachetool_process.returncode == 0
+    os.close(self._cachetool_stdin)
+    os.close(self._cachetool_stdout)
+    assert len(self._in_flight_insts) == 0
+    self._cachetool_process = None
+
+  def Sync(self):
+    self._PullInstsResults(len(self._in_flight_insts))
+
+  def _PushInsts(self, inst_name, *args):
+    assert self._cachetool_process != None
+    inst_id = self._INST_IDS[inst_name]
+    inst_code = struct.pack('b', inst_id)
+    for param in args:
+      if type(param) == int:
+        inst_code += struct.pack('i', param)
+      elif type(param) == str:
+        inst_code += struct.pack('i{}s'.format(len(param)), len(param), param)
+      else:
+        assert False, 'Couldn\'t passdown parameter: {}'.format(repr(param))
+    self._cachetool_process.stdin.write(inst_code)

[mattcary, 2016/07/01 12:09:29]

This is dangerous. If the subprocess stdout buffer fills up, stdin.write will
block waiting for that buffer to fill, and you'll deadlock.

The easy solution is to use Popen.communicate, but that won't work in your case
as you want to keep the process alive for multiple stdin/stdout roundtrips.

The way I've gotten around this is to have a read thread that buffers data. Then
_UnpackResult will read that buffer instead. You need to have locks etc.

A pain, certainly, but deadlocks are very difficult to identify and so we should
avoid known deadlock patterns even if it seems unlikely that we'll fill up the
pipe buffers.

[gabadie, 2016/07/01 14:10:25]

Good catch, I always forgot that python buffer stuf in subprocess piped. Managed
it without a thread by using the old fashion pipes.

[mattcary, 2016/07/01 14:30:21]

Don't system pipes have a finite buffer as well?

[gabadie, 2016/07/01 14:35:39]

Of course they do, but using os.write() allows me to know how much bytes were
written in the pipe, and we doing a _UnpackResult() that is the only place I do
a os.read() that is non blocking, I continue pushing stuf I was not able to push
before in the mean time I got all the data I want to read.

[pasko, 2016/07/04 16:53:52]

just noticed this discussion. Can you explain more why this would not work:
1. fill up all requests to cachetool
2. Popen.communicate() it and get a response
3. repeat if necessary

will that launch cachetool too many times?

+    self._cachetool_process.stdin.flush()
+    self._in_flight_insts.append([inst_name] + list(args))
+    if self._auto_sync:
+      assert len(self._in_flight_insts) == 1
+      self.Sync()
+
+  def _UnpackResult(self, fmt):
+    buf_size = struct.calcsize(fmt)
+    return struct.unpack(fmt, self._cachetool_process.stdout.read(buf_size))
+
+  def _PullInstsResults(self, count):
+    assert self._cachetool_process != None
+    if count == 0:
+      return
+    assert count <= len(self._in_flight_insts)
+    errors = []
+    for inst_position in xrange(count):
+      status_len = self._UnpackResult('i')[0]
+      if status_len == 0:
+        # print repr(self._in_flight_insts[inst_position]) + ' OK'
+        continue
+      status = self._UnpackResult('{}s'.format(status_len))[0]
+      errors.append((self._in_flight_insts[inst_position], status))
+    del self._in_flight_insts[:count]
+    if errors:
+      raise CacheBackendError(errors)
+
 
 def ApplyUrlWhitelistToCacheArchive(cache_archive_path,
                                     whitelisted_urls,
                                     output_cache_archive_path):
   """Generate a new cache archive containing only whitelisted urls.
 
   Args:
     cache_archive_path: Path of the cache archive to apply the white listing.
     whitelisted_urls: Set of url to keep in cache.
     output_cache_archive_path: Destination path of cache archive containing only
       white-listed urls.
   """
   cache_temp_directory = tempfile.mkdtemp(suffix='.cache')
   try:
     UnzipDirectoryContent(cache_archive_path, cache_temp_directory)
-    backend = CacheBackend(cache_temp_directory, 'simple')
-    cached_urls = backend.ListKeys()
-    for cached_url in cached_urls:
-      if cached_url not in whitelisted_urls:
-        backend.DeleteKey(cached_url)
-    for cached_url in backend.ListKeys():
-      assert cached_url in whitelisted_urls
+    with OnlineCacheBackend(
+        cache_temp_directory, CacheBackendType.Simple) as backend:
+      cached_urls = backend.ListKeys()
+      for cached_url in cached_urls:
+        if cached_url not in whitelisted_urls:
+          backend.DeleteKey(cached_url)
+      for cached_url in backend.ListKeys():
+        assert cached_url in whitelisted_urls
     ZipDirectoryContent(cache_temp_directory, output_cache_archive_path)
   finally:
     shutil.rmtree(cache_temp_directory)
 
 
 def ManualTestMain():
   import argparse
   parser = argparse.ArgumentParser(description='Tests cache back-end.')
   parser.add_argument('cache_archive_path', type=str)
   parser.add_argument('backend_type', type=str, choices=BACKEND_TYPES)
@@ skipping 15 matching lines @@
   print '{}\'s HTTP response header:'.format(selected_key)
   print cache_backend.GetStreamForKey(selected_key, 0)
   print cache_backend.GetDecodedContentForKey(selected_key)
   cache_backend.DeleteKey(keys[1])
   assert keys[1] not in cache_backend.ListKeys()
   shutil.rmtree(cache_path)
 
 
 if __name__ == '__main__':
   ManualTestMain()