Add Strace parallel support.

tools/isolate/trace_inputs.py

 #!/usr/bin/env python
 # coding=utf-8
 # Copyright (c) 2012 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.
 
 """Traces an executable and its child processes and extract the files accessed
 by them.
 
 The implementation uses OS-specific API. The native Kernel logger and the ETL
@@ skipping 11 matching lines @@
 import csv
 import getpass
 import glob
 import json
 import logging
 import optparse
 import os
 import re
 import subprocess
 import sys
+import threading
 import weakref
 
 ## OS-specific imports
 
 if sys.platform == 'win32':
   from ctypes.wintypes import byref, create_unicode_buffer, c_int, c_wchar_p
   from ctypes.wintypes import windll, FormatError  # pylint: disable=E0611
   from ctypes.wintypes import GetLastError  # pylint: disable=E0611
 elif sys.platform == 'darwin':
   import Carbon.File  #  pylint: disable=F0401
@@ skipping 567 matching lines @@
         self.files.add(filepath)
 
     def __init__(self, blacklist):
       self.blacklist = blacklist
       self.processes = {}
 
   class Tracer(object):
     """During it's lifetime, the tracing subsystem is enabled."""
     def __init__(self, logname):
       self._logname = logname
+      self._lock = threading.Lock()
+      self._traces = []
       self._initialized = True
 
     def trace(self, cmd, cwd, tracename, output):
       """Runs the OS-specific trace program on an executable.
 
       Arguments:
       - cmd: The command (a list) to run.
       - cwd: Current directory to start the child process in.
       - tracename: Name of the trace in the logname file.
       - output: If False, redirects output to PIPEs.
 
       Returns a tuple (resultcode, output) and updates the internal trace
       entries.
       """
+      # The implementation adds an item to self._traces.
       raise NotImplementedError(self.__class__.__name__)
 
-    def close(self):
-      self._initialized = False
+    def close(self, _timeout=None):
+      """Saves the meta-data in the logname file.
+
+      For kernel-based tracing, stops the tracing subsystem.
+
+      Must not be used manually when using 'with' construct.
+      """
+      with self._lock:
+        assert self._initialized
+        try:
+          data = {
+            'traces': self._traces,
+          }
+          write_json(self._logname, data, False)
+        finally:
+          self._initialized = False
 
     def post_process_log(self):
       """Post-processes the log so it becomes faster to load afterward.
 
       Must not be used manually when using 'with' construct.
       """
       assert not self._initialized, 'Must stop tracing first.'
 
     def __enter__(self):
       """Enables 'with' statement."""
@@ skipping 12 matching lines @@
     Initializes the tracing subsystem, which is a requirement for kernel-based
     tracers. Only one tracer instance should be live at a time!
 
     logname is the filepath to the json file that will contain the meta-data
     about the logs.
     """
     return self.Tracer(logname)
 
   @staticmethod
   def clean_trace(logname):
-    """Deletes the old log."""
+    """Deletes an old log."""
     raise NotImplementedError()
 
   @classmethod
   def parse_log(cls, logname, blacklist):
-    """Processes a trace log and returns the files opened and the files that do
+    """Processes trace logs and returns the files opened and the files that do
     not exist.
 
     It does not track directories.
 
     Most of the time, files that do not exist are temporary test files that
     should be put in /tmp instead. See http://crbug.com/116251.
 
-    Returns a tuple (existing files, non existing files, nb_processes_created)
+    Returns a list of dict with keys:
+      - results: A Results instance.
+      - trace: The corresponding tracename parameter provided to
+               get_tracer().trace().
+      - output: Output gathered during execution, if get_tracer().trace(...,
+                output=False) was used.
     """
     raise NotImplementedError(cls.__class__.__name__)
 
 
 class Strace(ApiBase):
   """strace implies linux."""
   IGNORED = (
     '/bin',
     '/dev',
     '/etc',
@@ skipping 301 matching lines @@
       """Returns the list of all handled traces to pass this as an argument to
       strace.
       """
       prefix = 'handle_'
       return [i[len(prefix):] for i in dir(cls.Process) if i.startswith(prefix)]
 
   class Tracer(ApiBase.Tracer):
     def trace(self, cmd, cwd, tracename, output):
       """Runs strace on an executable."""
       logging.info('trace(%s, %s, %s, %s)' % (cmd, cwd, tracename, output))
+      with self._lock:
+        if not self._initialized:
+          raise TracingFailure(
+              'Called Tracer.trace() on an unitialized object',
+              None, None, None, tracename)
+        assert tracename not in (i['trace'] for i in self._traces)
       stdout = stderr = None
       if output:
         stdout = subprocess.PIPE
         stderr = subprocess.STDOUT
       traces = ','.join(Strace.Context.traces())
       trace_cmd = [
         'strace',
         '-ff',
         '-s', '256',
         '-e', 'trace=%s' % traces,
-        '-o', self._logname,
+        '-o', self._logname + '.' + tracename,
       ]
       child = subprocess.Popen(
           trace_cmd + cmd,
           cwd=cwd,
           stdin=subprocess.PIPE,
           stdout=stdout,
           stderr=stderr)
       out = child.communicate()[0]
-      # Once it's done, write metadata into the log file to be able to follow
-      # the pid files.
-      value = {
-        'cwd': cwd,
-        # The pid of strace process, not very useful.
-        'pid': child.pid,
-      }
-      write_json(self._logname, value, False)
+      # TODO(maruel): Walk the logs and figure out the root process would
+      # simplify parsing the logs a *lot*.
+      with self._lock:
+        assert tracename not in (i['trace'] for i in self._traces)
+        self._traces.append(
+          {
+            'cmd': cmd,
+            'cwd': cwd,
+            # The pid of strace process, not very useful.
+            'pid': child.pid,
+            'trace': tracename,
+            'output': out,
+          })
       return child.returncode, out
 
   @staticmethod
   def clean_trace(logname):
     if os.path.isfile(logname):
       os.remove(logname)
     # Also delete any pid specific file from previous traces.
     for i in glob.iglob(logname + '.*'):
       if i.rsplit('.', 1)[1].isdigit():
         os.remove(i)
 
   @classmethod
   def parse_log(cls, logname, blacklist):
     logging.info('parse_log(%s, %s)' % (logname, blacklist))
     data = read_json(logname)
-    context = cls.Context(blacklist, data['cwd'])
-    for pidfile in glob.iglob(logname + '.*'):
-      pid = pidfile.rsplit('.', 1)[1]
-      if pid.isdigit():
-        pid = int(pid)
-        # TODO(maruel): Load as utf-8
-        for line in open(pidfile, 'rb'):
-          context.on_line(pid, line)
-
-    return context.to_results()
+    out = []
+    for item in data['traces']:
+      result = {
+        'trace': item['trace'],
+        'output': item['output'],
+      }
+      try:
+        context = cls.Context(blacklist, item['cwd'])
+        for pidfile in glob.iglob('%s.%s.*' % (logname, item['trace'])):
+          pid = pidfile.rsplit('.', 1)[1]
+          if pid.isdigit():
+            pid = int(pid)
+            # TODO(maruel): Load as utf-8
+            for line in open(pidfile, 'rb'):
+              context.on_line(pid, line)
+        result['results'] = context.to_results()
+      except TracingFailure, e:
+        result['exception'] = e
+      out.append(result)
+    return out
 
 
 class Dtrace(ApiBase):
   """Uses DTrace framework through dtrace. Requires root access.
 
   Implies Mac OSX.
 
   dtruss can't be used because it has compatibility issues with python.
 
   Also, the pid->cwd handling needs to be done manually since OSX has no way to
@@ skipping 542 matching lines @@
 
       This dtruss is broken when it starts the process itself or when tracing
       child processes, this code starts a wrapper process
       trace_child_process.py, which waits for dtrace to start, then
       trace_child_process.py starts the executable to trace.
       """
       logging.info('trace(%s, %s, %s, %s)' % (cmd, cwd, tracename, output))
       logging.info('Running: %s' % cmd)
       signal = 'Go!'
       logging.debug('Our pid: %d' % os.getpid())
+      with self._lock:
+        if not self._initialized:
+          raise TracingFailure(
+              'Called Tracer.trace() on an unitialized object',
+              None, None, None, tracename)
+        assert tracename not in (i['trace'] for i in self._traces)
 
       # Part 1: start the child process.
       stdout = stderr = None
       if output:
         stdout = subprocess.PIPE
         stderr = subprocess.STDOUT
       child_cmd = [
         sys.executable,
         os.path.join(BASE_DIR, 'trace_child_process.py'),
         '--wait',
@@ skipping 38 matching lines @@
       out = child.communicate(signal)[0]
 
       dtrace.wait()
       if dtrace.returncode != 0:
         print 'dtrace failure: %d' % dtrace.returncode
         with open(logname) as logfile:
           print ''.join(logfile.readlines()[-100:])
         # Find a better way.
         os.remove(logname)
 
+      with self._lock:
+        assert tracename not in (i['trace'] for i in self._traces)
+        self._traces.append(
+          {
+            'cmd': cmd,
+            'cwd': cwd,
+            # The pid of strace process, not very useful.
+            'pid': child.pid,
+            'trace': tracename,
+            'output': out,
+          })
       return dtrace.returncode or child.returncode, out
 
     def post_process_log(self):
       """Sorts the log back in order when each call occured.
 
       dtrace doesn't save the buffer in strict order since it keeps one buffer
       per CPU.
       """
       super(Dtrace.Tracer, self).post_process_log()
       logname = self._logname + '.log'
@@ skipping 16 matching lines @@
 
   @staticmethod
   def clean_trace(logname):
     for ext in ('', '.log'):
       if os.path.isfile(logname + ext):
         os.remove(logname + ext)
 
   @classmethod
   def parse_log(cls, logname, blacklist):
     logging.info('parse_log(%s, %s)' % (logname, blacklist))
-    context = cls.Context(blacklist)
-    for line in open(logname + '.log', 'rb'):
-      context.on_line(line)
-    return context.to_results()
+    data = read_json(logname)
+    out = []
+    for item in data['traces']:
+      context = cls.Context(blacklist)
+      for line in open(logname + '.log', 'rb'):
+        context.on_line(line)
+      out.append(
+          {
+            'results': context.to_results(),
+            'trace': item['trace'],
+            'output': item['output'],
+          })
+    return out
 
 
 class LogmanTrace(ApiBase):
   """Uses the native Windows ETW based tracing functionality to trace a child
   process.
 
   Caveat: this implementations doesn't track cwd or initial_cwd. It is because
   the Windows Kernel doesn't have a concept of 'current working directory' at
   all. A Win32 process has a map of current directories, one per drive letter
   and it is managed by the user mode kernel32.dll. In kernel, a file is always
@@ skipping 320 matching lines @@
     PROCESSOR_ID = 11
     TIMESTAMP = 16
     NULL_GUID = '{00000000-0000-0000-0000-000000000000}'
     USER_DATA = 19
 
     def trace(self, cmd, cwd, tracename, output):
       """Uses logman.exe to start and stop the NT Kernel Logger while the
       executable to be traced is run.
       """
       logging.info('trace(%s, %s, %s, %s)' % (cmd, cwd, tracename, output))
+      with self._lock:
+        if not self._initialized:
+          raise TracingFailure(
+              'Called Tracer.trace() on an unitialized object',
+              None, None, None, tracename)
+        assert tracename not in (i['trace'] for i in self._traces)
+
       # Use "logman -?" for help.
 
       stdout = stderr = None
       if output:
         stdout = subprocess.PIPE
         stderr = subprocess.STDOUT
 
       # 1. Start the log collection.
       self._start_log(self._logname + '.etl')
 
@@ skipping 14 matching lines @@
             cwd=cwd,
             stdin=subprocess.PIPE,
             stdout=stdout,
             stderr=stderr)
         logging.debug('Started child pid: %d' % child.pid)
         out = child.communicate()[0]
       finally:
         # 3. Stop the log collection.
         self._stop_log()
 
-      # 5. Save metadata.
-      value = {
-        'pid': child.pid,
-        'format': 'csv',
-      }
-      write_json(self._logname, value, False)
+      with self._lock:
+        assert tracename not in (i['trace'] for i in self._traces)
+        self._traces.append({
+          'command': cmd,
+          'cwd': cwd,
+          'pid': child.pid,
+          'trace': tracename,
+          'output': out,
+        })
+
       return child.returncode, out
 
     @classmethod
     def _start_log(cls, etl):
       """Starts the log collection.
 
       One can get the list of potentially interesting providers with:
       "logman query providers | findstr /i file"
       """
       cmd_start = [
@@ skipping 212 matching lines @@
   @classmethod
   def parse_log(cls, logname, blacklist):
     logging.info('parse_log(%s, %s)' % (logname, blacklist))
 
     def blacklist_more(filepath):
       # All the NTFS metadata is in the form x:\$EXTEND or stuff like that.
       return blacklist(filepath) or re.match(r'[A-Z]\:\\\$EXTEND', filepath)
 
     data = read_json(logname)
     lines = read_json(logname + '.json')
-    context = cls.Context(blacklist_more, data['pid'])
-    for line in lines:
-      context.on_line(line)
-    return context.to_results()
+    out = []
+    for index, item in enumerate(data['traces']):
+      print >> sys.stderr, ('%d out of %d' % (index, len(data['traces'])))
+      context = cls.Context(blacklist_more, item['pid'])
+      for line in lines:
+        context.on_line(line)
+      out.append(
+          {
+            'results': context.to_results(),
+            'trace': item['trace'],
+            'output': item['output'],
+          })
+    return out
 
 
 def get_api():
   """Returns the correct implementation for the current OS."""
   if sys.platform == 'cygwin':
     raise NotImplementedError(
         'Not implemented for cygwin, start the script from Win32 python')
   flavors = {
     'win32': LogmanTrace,
     'darwin': Dtrace,
@@ skipping 97 matching lines @@
 
 def load_trace(logfile, root_dir, api):
   """Loads a trace file and returns the Results instance.
 
   Arguments:
   - logfile: File to load.
   - root_dir: Root directory to use to determine if a file is relevant to the
               trace or not.
   - api: A tracing api instance.
   """
-  results = api.parse_log(logfile, get_blacklist(api))
+  data = api.parse_log(logfile, get_blacklist(api))
+  assert len(data) == 1, 'More than one trace was detected!'
+  results = data[0]['results']
   if root_dir:
     results = results.strip_root(root_dir)
   return results
 
 
 def CMDclean(parser, args):
   """Cleans up traces."""
   options, args = parser.parse_args(args)
   api = get_api()
   api.clean_trace(options.log)
@@ skipping 144 matching lines @@
       for fn in dir(sys.modules[__name__])
       if fn.startswith('CMD')))
 
   command = get_command_handler(argv[0] if argv else None)
   parser = gen_parser(command)
   return command(parser, argv[1:])
 
 
 if __name__ == '__main__':
   sys.exit(main(sys.argv[1:]))