Add a new class to take care of out-of-order execution.

gclient.py

 #!/usr/bin/python
 # Copyright (c) 2010 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.
 
 """Meta checkout manager supporting both Subversion and GIT.
 
 Files
   .gclient      : Current client configuration, written by 'config' command.
                   Format is a Python script defining 'solutions', a list whose
@@ skipping 31 matching lines @@
                it will be removed from the list and the list will be extended
                by the list of matching files.
 
   Example:
     hooks = [
       { "pattern": "\\.(gif|jpe?g|pr0n|png)$",
         "action":  ["python", "image_indexer.py", "--all"]},
     ]
 """
 
-__version__ = "0.5"
+__version__ = "0.5.1"
 
 import logging
 import optparse
 import os
 import posixpath
 import pprint
 import re
 import subprocess
 import sys
+import threading
 import urlparse
 import urllib
 
 import breakpad
 
 import gclient_scm
 import gclient_utils
 from third_party.repo.progress import Progress
 
 
 def attr(attr, data):
   """Sets an attribute on a function."""
   def hook(fn):
     setattr(fn, attr, data)
     return fn
   return hook
 
 
 ## GClient implementation.
 
+class WorkItem(object):
+  """One work item."""
+  requirements = []
+  name = None
+
+  def run(self):
+    pass
+
+
+class ExecutionQueue(object):
+  """Dependencies sometime needs to be run out of order due to From() keyword.
+
+  This class manages that all the required dependencies are run before running
+  each one.
+
+  Methods of this class are multithread safe.
+  """
+  def __init__(self, progress):
+    self.lock = threading.Lock()
+    # List of Dependency.
+    self.queued = []
+    # List of strings representing each Dependency.name that was run.
+    self.ran = []
+    # List of items currently running.
+    self.running = []
+    self.progress = progress
+    if self.progress:
+      self.progress.update()
+
+  def enqueue(self, d):
+    """Enqueue one Dependency to be executed later once its requirements are
+    satisfied.
+    """
+    assert isinstance(d, WorkItem)
+    try:
+      self.lock.acquire()
+      self.queued.append(d)
+      total = len(self.queued) + len(self.ran) + len(self.running)
+    finally:
+      self.lock.release()
+    if self.progress:
+      self.progress._total = total + 1
+      self.progress.update(0)
+
+  def flush(self, *args, **kwargs):
+    """Runs all enqueued items until all are executed."""
+    while self._run_one_item(*args, **kwargs):
+      pass
+    queued = []
+    running = []
+    try:
+      self.lock.acquire()
+      if self.queued:
+        queued = self.queued
+        self.queued = []
+      if self.running:
+        running = self.running
+        self.running = []
+    finally:
+      self.lock.release()
+    if self.progress:
+      self.progress.end()
+    if queued:
+      raise gclient_utils.Error('Entries still queued: %s' % str(queued))
+    if running:
+      raise gclient_utils.Error('Entries still queued: %s' % str(running))
+
+  def _run_one_item(self, *args, **kwargs):
+    """Removes one item from the queue that has all its requirements completed
+    and execute it.
+
+    Returns False if no item could be run.
+    """
+    i = 0
+    d = None
+    try:
+      self.lock.acquire()
+      while i != len(self.queued) and not d:
+        d = self.queued.pop(i)
+        for r in d.requirements:
+          if not r in self.ran:
+            self.queued.insert(i, d)
+            d = None
+            break
+        i += 1
+      if not d:
+        return False
+      self.running.append(d)
+    finally:
+      self.lock.release()
+    d.run(*args, **kwargs)
+    try:
+      self.lock.acquire()
+      # TODO(maruel): http://crbug.com/51711
+      #assert not d.name in self.ran
+      if not d.name in self.ran:
+        self.ran.append(d.name)
+      self.running.remove(d)
+      if self.progress:
+        self.progress.update(1)
+    finally:
+      self.lock.release()
+    return True
+
 
 class GClientKeywords(object):
   class FromImpl(object):
     """Used to implement the From() syntax."""
 
     def __init__(self, module_name, sub_target_name=None):
       """module_name is the dep module we want to include from.  It can also be
       the name of a subdirectory to include from.
 
       sub_target_name is an optional parameter if the module name in the other
@@ skipping 35 matching lines @@
 
     def Lookup(self, var_name):
       """Implements the Var syntax."""
       if var_name in self._custom_vars:
         return self._custom_vars[var_name]
       elif var_name in self._local_scope.get("vars", {}):
         return self._local_scope["vars"][var_name]
       raise gclient_utils.Error("Var is not defined: %s" % var_name)
 
 
-class Dependency(GClientKeywords):
+class Dependency(GClientKeywords, WorkItem):
   """Object that represents a dependency checkout."""
   DEPS_FILE = 'DEPS'
 
   def __init__(self, parent, name, url, safesync_url, custom_deps,
                custom_vars, deps_file):
     GClientKeywords.__init__(self)
     self.parent = parent
     self.name = name
     self.url = url
     self.parsed_url = None
@@ skipping 165 matching lines @@
       deps = rel_deps
 
     # Convert the deps into real Dependency.
     for name, url in deps.iteritems():
       if name in [s.name for s in self.dependencies]:
         raise gclient_utils.Error(
             'The same name "%s" appears multiple times in the deps section' %
                 name)
       self.dependencies.append(Dependency(self, name, url, None, None, None,
           None))
-    # Sorting by name would in theory make the whole thing coherent, since
-    # subdirectories will be sorted after the parent directory, but that doens't
-    # work with From() that fetch from a dependency with a name being sorted
-    # later. But if this would be removed right now, many projects wouldn't be
-    # able to sync anymore.
-    self.dependencies.sort(key=lambda x: x.name)
     logging.debug('Loaded: %s' % str(self))
 
-  def RunCommandRecursively(self, options, revision_overrides,
-                            command, args, pm):
+  def run(self, options, revision_overrides, command, args, work_queue):
     """Runs 'command' before parsing the DEPS in case it's a initial checkout
     or a revert."""
     assert self._file_list == []
     # When running runhooks, there's no need to consult the SCM.
     # All known hooks are expected to run unconditionally regardless of working
     # copy state, so skip the SCM status check.
     run_scm = command not in ('runhooks', None)
     self.parsed_url = self.LateOverride(self.url)
     if run_scm and self.parsed_url:
       if isinstance(self.parsed_url, self.FileImpl):
         # Special support for single-file checkout.
         if not command in (None, 'cleanup', 'diff', 'pack', 'status'):
           options.revision = self.parsed_url.GetRevision()
           scm = gclient_scm.SVNWrapper(self.parsed_url.GetPath(),
                                        self.root_dir(),
                                        self.name)
           scm.RunCommand('updatesingle', options,
                          args + [self.parsed_url.GetFilename()],
                          self._file_list)
       else:
         options.revision = revision_overrides.get(self.name)
         scm = gclient_scm.CreateSCM(self.parsed_url, self.root_dir(), self.name)
         scm.RunCommand(command, options, args, self._file_list)
         self._file_list = [os.path.join(self.name, f.strip())
                            for f in self._file_list]
       options.revision = None
     self.processed = True
-    if pm:
-      # The + 1 comes from the fact that .gclient is considered a step in
-      # itself, .i.e. this code is called one time for the .gclient. This is not
-      # conceptually correct but it simplifies code.
-      pm._total = len(self.tree(False)) + 1
-      pm.update()
     if self.recursion_limit():
       # Then we can parse the DEPS file.
       self.ParseDepsFile(True)
-      if pm:
-        pm._total = len(self.tree(False)) + 1
-        pm.update(0)
       # Adjust the implicit dependency requirement; e.g. if a DEPS file contains
       # both src/foo and src/foo/bar, src/foo/bar is implicitly dependent of
-      # src/foo. Yes, it's O(n^2)...
+      # src/foo. Yes, it's O(n^2)... It's important to do that before
+      # enqueueing them.
       for s in self.dependencies:
         for s2 in self.dependencies:
           if s is s2:
             continue
           if s.name.startswith(posixpath.join(s2.name, '')):
             s.requirements.append(s2.name)
 
       # Parse the dependencies of this dependency.
       for s in self.dependencies:
-        # TODO(maruel): All these can run concurrently! No need for threads,
-        # just buffer stdout&stderr on pipes and flush as they complete.
-        # Watch out for stdin.
-        s.RunCommandRecursively(options, revision_overrides, command, args, pm)
+        work_queue.enqueue(s)
 
   def RunHooksRecursively(self, options):
-    """Evaluates all hooks, running actions as needed. RunCommandRecursively()
+    """Evaluates all hooks, running actions as needed. run()
     must have been called before to load the DEPS."""
     # If "--force" was specified, run all hooks regardless of what files have
     # changed.
     if self.deps_hooks and self.direct_reference:
       # TODO(maruel): If the user is using git or git-svn, then we don't know
       # what files have changed so we always run all hooks. It'd be nice to fix
       # that.
       if (options.force or
           isinstance(self.parsed_url, self.FileImpl) or
           gclient_scm.GetScmName(self.parsed_url) in ('git', None) or
@@ skipping 85 matching lines @@
   def file_list(self):
     result = self._file_list[:]
     for d in self.dependencies:
       result.extend(d.file_list())
     return result
 
   def __str__(self):
     out = []
     for i in ('name', 'url', 'parsed_url', 'safesync_url', 'custom_deps',
               'custom_vars', 'deps_hooks', '_file_list', 'processed',
-              'hooks_ran', 'deps_parsed', 'requirements'):
+              'hooks_ran', 'deps_parsed', 'requirements', 'direct_reference'):
       # 'deps_file'
       if self.__dict__[i]:
         out.append('%s: %s' % (i, self.__dict__[i]))
 
     for d in self.dependencies:
       out.extend(['  ' + x for x in str(d).splitlines()])
       out.append('')
     return '\n'.join(out)
 
   def __repr__(self):
@@ skipping 87 matching lines @@
             self, s['name'], s['url'],
             s.get('safesync_url', None),
             s.get('custom_deps', {}),
             s.get('custom_vars', {}),
             None))
       except KeyError:
         raise gclient_utils.Error('Invalid .gclient file. Solution is '
                                   'incomplete: %s' % s)
     # .gclient can have hooks.
     self.deps_hooks = config_dict.get('hooks', [])
+    self.direct_reference = True
+    self.deps_parsed = True
 
   def SaveConfig(self):
     gclient_utils.FileWrite(os.path.join(self.root_dir(),
                                          self._options.config_filename),
                             self.config_content)
 
   @staticmethod
   def LoadCurrentConfig(options):
     """Searches for and loads a .gclient file relative to the current working
     dir. Returns a GClient object."""
@@ skipping 84 matching lines @@
 
     Args:
       command: The command to use (e.g., 'status' or 'diff')
       args: list of str - extra arguments to add to the command line.
     """
     if not self.dependencies:
       raise gclient_utils.Error('No solution specified')
     revision_overrides = self._EnforceRevisions()
     pm = None
     if command == 'update' and not self._options.verbose:
-      pm = Progress('Syncing projects', len(self.tree(False)) + 1)
-    self.RunCommandRecursively(self._options, revision_overrides,
-                               command, args, pm)
-    if pm:
-      pm.end()
+      pm = Progress('Syncing projects', 1)
+    work_queue = ExecutionQueue(pm)
+    for s in self.dependencies:
+      work_queue.enqueue(s)
+    work_queue.flush(self._options, revision_overrides, command, args,
+        work_queue)
 
     # Once all the dependencies have been processed, it's now safe to run the
     # hooks.
     if not self._options.nohooks:
       self.RunHooksRecursively(self._options)
 
     if command == 'update':
       # Notify the user if there is an orphaned entry in their working copy.
       # Only delete the directory if there are no changes in it, and
       # delete_unversioned_trees is set to true.
@@ skipping 20 matching lines @@
                 entry_fixed, self.root_dir()))
             gclient_utils.RemoveDirectory(e_dir)
       # record the current list of entries for next time
       self._SaveEntries()
     return 0
 
   def PrintRevInfo(self):
     if not self.dependencies:
       raise gclient_utils.Error('No solution specified')
     # Load all the settings.
-    self.RunCommandRecursively(self._options, {}, None, [], None)
+    work_queue = ExecutionQueue(None)
+    for s in self.dependencies:
+      work_queue.enqueue(s)
+    work_queue.flush(self._options, {}, None, [], work_queue)
 
     def GetURLAndRev(dep):
       """Returns the revision-qualified SCM url for a Dependency."""
       if dep.parsed_url is None:
         return None
       if isinstance(dep.parsed_url, self.FileImpl):
         original_url = dep.parsed_url.file_location
       else:
         original_url = dep.parsed_url
       url, _ = gclient_utils.SplitUrlRevision(original_url)
@@ skipping 38 matching lines @@
       keys = sorted(entries.keys())
       for x in keys:
         line = '%s: %s' % (x, entries[x])
         if x is not keys[-1]:
           line += ';'
         print line
     logging.info(str(self))
 
   def ParseDepsFile(self, direct_reference):
     """No DEPS to parse for a .gclient file."""
-    self.direct_reference = True
-    self.deps_parsed = True
+    raise gclient_utils.Error('Internal error')
 
   def root_dir(self):
     """Root directory of gclient checkout."""
     return self._root_dir
 
   def enforced_os(self):
     """What deps_os entries that are to be parsed."""
     return self._enforced_os
 
   def recursion_limit(self):
@@ skipping 388 matching lines @@
     return CMDhelp(parser, argv)
   except gclient_utils.Error, e:
     print >> sys.stderr, 'Error: %s' % str(e)
     return 1
 
 
 if '__main__' == __name__:
   sys.exit(Main(sys.argv[1:]))
 
 # vim: ts=2:sw=2:tw=80:et: