Add -j option to gclient to run parallel updates

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 33 matching lines @@
 
   Example:
     hooks = [
       { "pattern": "\\.(gif|jpe?g|pr0n|png)$",
         "action":  ["python", "image_indexer.py", "--all"]},
     ]
 """
 
 __version__ = "0.4.1"
 
+import copy
 import errno
 import logging
 import optparse
 import os
 import pprint
 import re
 import subprocess
 import sys
 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
 
 
@@ skipping 481 matching lines @@
       sol, rev = revision.split('@', 1)
       if not sol in solutions_names:
         #raise gclient_utils.Error('%s is not a valid solution.' % sol)
         print >> sys.stderr, ('Please fix your script, having invalid '
                               '--revision flags will soon considered an error.')
       else:
         revision_overrides[sol] = rev
       index += 1
     return revision_overrides
 
+  def GetSCMCommandClosure(self, path, url, revision, command, args, file_list,
+                           continuation=None):
+    """Gets a closure that runs a SCM command on a particular dependency."""
+    def _Closure():
+      logging.debug("Running %s in %s to %s %s" % (command, path, url,
+                                                   revision))
+      options = copy.copy(self._options)
+      options.revision = revision
+      scm = gclient_scm.CreateSCM(url, self.root_dir(), path)
+      scm.RunCommand(command, options, args, file_list)
+      if continuation:
+        continuation()
+    return _Closure
+
   def RunOnDeps(self, command, args):
     """Runs a command on each dependency in a client and its dependencies.
 
     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()
 
     # When running runhooks --force, 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 = not (command == 'runhooks' and self._options.force)
 
     entries = {}
-    file_list = []
-    # Run on the base solutions first.
-    for solution in self.dependencies:
-      name = solution.name
-      if name in entries:
-        raise gclient_utils.Error("solution %s specified more than once" % name)
-      url = solution.url
-      entries[name] = url
-      if run_scm and url:
-        self._options.revision = revision_overrides.get(name)
-        scm = gclient_scm.CreateSCM(url, self.root_dir(), name)
-        scm.RunCommand(command, self._options, args, file_list)
-        file_list = [os.path.join(name, f.strip()) for f in file_list]
-        self._options.revision = None
 
-    # Process the dependencies next (sort alphanumerically to ensure that
-    # containing directories get populated first and for readability)
-    deps = self._ParseAllDeps(entries)
-    deps_to_process = deps.keys()
-    deps_to_process.sort()
+    # To avoid threading issues, all file lists get constructed separately then
+    # gathered in a flattened list at the end.
+    file_list_list = []
+    file_list_dict = {}
 
-    # First pass for direct dependencies.
-    if command == 'update' and not self._options.verbose:
-      pm = Progress('Syncing projects', len(deps_to_process))
-    for d in deps_to_process:
-      if command == 'update' and not self._options.verbose:
-        pm.update()
-      if type(deps[d]) == str:
-        url = deps[d]
-        entries[d] = url
-        if run_scm:
-          self._options.revision = revision_overrides.get(d)
-          scm = gclient_scm.CreateSCM(url, self.root_dir(), d)
-          scm.RunCommand(command, self._options, args, file_list)
-          self._options.revision = None
-      elif isinstance(deps[d], self.FileImpl):
-        if command in (None, 'cleanup', 'diff', 'pack', 'status'):
-          continue
-        file_dep = deps[d]
-        self._options.revision = file_dep.GetRevision()
-        if run_scm:
-          scm = gclient_scm.SVNWrapper(file_dep.GetPath(), self.root_dir(), d)
-          scm.RunCommand('updatesingle', self._options,
-                         args + [file_dep.GetFilename()], file_list)
+    thread_pool = gclient_utils.ThreadPool(self._options.jobs)
+    thread_pool.Start()
 
-    if command == 'update' and not self._options.verbose:
-      pm.end()
+    show_progress = command == 'update' and not self._options.verbose
+    pm_update = None
 
-    # Second pass for inherited deps (via the From keyword)
-    for d in deps_to_process:
-      if isinstance(deps[d], self.FromImpl):
-        # Getting the URL from the sub_deps file can involve having to resolve
-        # a File() or having to resolve a relative URL.  To resolve relative
-        # URLs, we need to pass in the orignal sub deps URL.
-        sub_deps_base_url = deps[deps[d].module_name]
-        sub_deps = Dependency(self, deps[d].module_name, sub_deps_base_url
-            ).ParseDepsFile(False)
-        url = deps[d].GetUrl(d, sub_deps_base_url, self.root_dir(), sub_deps)
-        entries[d] = url
-        if run_scm:
-          self._options.revision = revision_overrides.get(d)
-          scm = gclient_scm.CreateSCM(url, self.root_dir(), d)
-          scm.RunCommand(command, self._options, args, file_list)
-          self._options.revision = None
+    try:
+      # Run on the base solutions first.
+      if show_progress:
+        pm = gclient_utils.ThreadSafeProgress('Syncing base solutions',
+                                              len(self.dependencies))
+        pm_update = lambda: pm.update()
+
+      for solution in self.dependencies:
+        name = solution.name
+        if name in entries:
+          raise gclient_utils.Error("solution %s specified more than once"
+                                    % name)
+        url = solution.url
+        entries[name] = url
+        if run_scm and url:
+          revision = revision_overrides.get(name)
+          file_list = []
+          file_list_dict[name] = file_list
+          thread_pool.AddJob(self.GetSCMCommandClosure(
+              name, url, revision, command, args, file_list, pm_update))
+
+      thread_pool.WaitJobs()
+
+      if show_progress:
+        pm.end()
+
+      for solution in self.dependencies:
+        name = solution.name
+        try:
+          file_list_list.append([os.path.join(name, f.strip())
+                                 for f in file_list_dict[name]])
+        except KeyError:
+          # We may not have added the file list to the dict, see tests above.
+          # Instead of duplicating the tests, it's less fragile to just ignore
+          # the exception.
+          pass
+        try:
+          deps_content = gclient_utils.FileRead(
+              os.path.join(self.root_dir(), name, solution.deps_file))
+        except IOError, e:
+          if e.errno != errno.ENOENT:
+            raise
+          deps_content = ""
+
+      # Process the dependencies next (sort alphanumerically to ensure that
+      # containing directories get populated first and for readability)
+      # TODO(piman): when using multiple threads, the ordering is not ensured.
+      # In many cases (e.g. updates to an existing checkout where DEPS don't
+      # move between directories), it'll still be correct but for completeness
+      # this should be fixed.
+      deps = self._ParseAllDeps(entries)
+      deps_to_process = deps.keys()
+      deps_to_process.sort()
+
+      # First pass for direct dependencies.
+      if show_progress:
+        pm = gclient_utils.ThreadSafeProgress('Syncing projects',
+                                              len(deps_to_process))
+        pm_update = lambda: pm.update()
+      for d in deps_to_process:
+        file_list = []
+        file_list_list.append(file_list)
+
+        if type(deps[d]) == str:
+          url = deps[d]
+          entries[d] = url
+          if run_scm:
+            revision = revision_overrides.get(d)
+            thread_pool.AddJob(self.GetSCMCommandClosure(d, url, revision,
+                                                         command, args,
+                                                         file_list, pm_update))
+        elif isinstance(deps[d], self.FileImpl):
+          if command in (None, 'cleanup', 'diff', 'pack', 'status'):
+            continue
+          file_dep = deps[d]
+          revision = file_dep.GetRevision()
+          if run_scm:
+            thread_pool.AddJob(self.GetSCMCommandClosure(
+                d, file_dep.GetPath(), revision, "updatesingle",
+                args + [file_dep.GetFilename()], file_list))
+
+      thread_pool.WaitJobs()
+
+      if show_progress:
+        pm.end()
+
+      # Second pass for inherited deps (via the From keyword)
+      for d in deps_to_process:
+        if isinstance(deps[d], self.FromImpl):
+          # Getting the URL from the sub_deps file can involve having to resolve
+          # a File() or having to resolve a relative URL.  To resolve relative
+          # URLs, we need to pass in the orignal sub deps URL.
+          sub_deps_base_url = deps[deps[d].module_name]
+          sub_deps = Dependency(self, deps[d].module_name,
+                                sub_deps_base_url).ParseDepsFile(False)
+          url = deps[d].GetUrl(d, sub_deps_base_url, self.root_dir(), sub_deps)
+          entries[d] = url
+          if run_scm:
+            revision = revision_overrides.get(d)
+            file_list = []
+            file_list_list.append(file_list)
+            thread_pool.AddJob(self.GetSCMCommandClosure(d, url, revision,
+                                                         command, args,
+                                                         file_list))
+
+      thread_pool.WaitJobs()
+
+    finally:
+      thread_pool.Stop()
+
+    file_list = sum(file_list_list, [])
 
     # Convert all absolute paths to relative.
     for i in range(len(file_list)):
       # TODO(phajdan.jr): We should know exactly when the paths are absolute.
       # It depends on the command being executed (like runhooks vs sync).
       if not os.path.isabs(file_list[i]):
         continue
 
       prefix = os.path.commonprefix([self.root_dir().lower(),
                                      file_list[i].lower()])
@@ skipping 490 matching lines @@
     CMDhelp.usage = ('\n\nCommands are:\n' + '\n'.join([
         '  %-10s %s' % (fn[3:], Command(fn[3:]).__doc__.split('\n')[0].strip())
         for fn in dir(sys.modules[__name__]) if fn.startswith('CMD')]))
     parser = optparse.OptionParser(version='%prog ' + __version__)
     parser.add_option('-v', '--verbose', action='count', default=0,
                       help='Produces additional output for diagnostics. Can be '
                            'used up to three times for more logging info.')
     parser.add_option('--gclientfile', dest='config_filename',
                       default=os.environ.get('GCLIENT_FILE', '.gclient'),
                       help='Specify an alternate %default file')
+    parser.add_option("-j", "--jobs", default=0, type="int",
+                      help=("specify how many SCM commands can run in "
+                            "parallel"))
+
     # Integrate standard options processing.
     old_parser = parser.parse_args
     def Parse(args):
       (options, args) = old_parser(args)
       level = None
       if options.verbose == 2:
         level = logging.INFO
       elif options.verbose > 2:
         level = logging.DEBUG
       logging.basicConfig(level=level,
@@ skipping 29 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: