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

tools/android/loading/chrome_cache.py

Index: tools/android/loading/chrome_cache.py
diff --git a/tools/android/loading/chrome_cache.py b/tools/android/loading/chrome_cache.py
index 24d3f75638d33f864831dab97c329a84b61519aa..e4ca956b9d3db1e472827f536036e357da00632b 100644
--- a/tools/android/loading/chrome_cache.py
+++ b/tools/android/loading/chrome_cache.py
@@ -10,6 +10,7 @@ import json
 import os
 import re
 import shutil
+import struct
 import subprocess
 import sys
 import tempfile
@@ -28,8 +29,12 @@ 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>.*)$')
@@ -232,6 +237,12 @@ def CopyCacheDirectory(directory_src_path, 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.

   """
@@ -249,7 +260,7 @@ class CacheBackend(object):
     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."""
@@ -262,7 +273,7 @@ class CacheBackend(object):
     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.
@@ -307,16 +318,17 @@ class CacheBackend(object):
     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):
@@ -364,6 +376,150 @@ class CacheBackend(object):
     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,
@@ -379,13 +535,14 @@ def ApplyUrlWhitelistToCacheArchive(cache_archive_path,
   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)