Roll version of gsutil to 4.15.

third_party/gsutil/gslib/command.py

 # -*- coding: utf-8 -*-
 # Copyright 2010 Google Inc. All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
@@ skipping 32 matching lines @@
 from gslib.cloud_api import AccessDeniedException
 from gslib.cloud_api import ArgumentException
 from gslib.cloud_api import ServiceException
 from gslib.cloud_api_delegator import CloudApiDelegator
 from gslib.cs_api_map import ApiSelector
 from gslib.cs_api_map import GsutilApiMapFactory
 from gslib.exception import CommandException
 from gslib.help_provider import HelpProvider
 from gslib.name_expansion import NameExpansionIterator
 from gslib.name_expansion import NameExpansionResult
-from gslib.parallelism_framework_util import AtomicIncrementDict
-from gslib.parallelism_framework_util import BasicIncrementDict
-from gslib.parallelism_framework_util import ThreadAndProcessSafeDict
+from gslib.parallelism_framework_util import AtomicDict
 from gslib.plurality_checkable_iterator import PluralityCheckableIterator
 from gslib.sig_handling import RegisterSignalHandler
 from gslib.storage_url import StorageUrlFromString
 from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
 from gslib.translation_helper import AclTranslation
+from gslib.translation_helper import PRIVATE_DEFAULT_OBJ_ACL
+from gslib.util import CheckMultiprocessingAvailableAndInit
 from gslib.util import GetConfigFilePath
 from gslib.util import GsutilStreamHandler
 from gslib.util import HaveFileUrls
 from gslib.util import HaveProviderUrls
 from gslib.util import IS_WINDOWS
-from gslib.util import MultiprocessingIsAvailable
 from gslib.util import NO_MAX
 from gslib.util import UrlsAreForSingleProvider
 from gslib.util import UTF8
 from gslib.wildcard_iterator import CreateWildcardIterator
 
 OFFER_GSUTIL_M_SUGGESTION_THRESHOLD = 5
 
 if IS_WINDOWS:
   import ctypes  # pylint: disable=g-import-not-at-top
 
@@ skipping 137 matching lines @@
 
   # List of all existing task queues - used by all pools to find the queue
   # that's appropriate for the given recursive_apply_level.
   task_queues = []
 
   # Used to assign a globally unique caller ID to each Apply call.
   caller_id_lock = manager.Lock()
   caller_id_counter = multiprocessing.Value('i', 0)
 
   # Map from caller_id to total number of tasks to be completed for that ID.
-  total_tasks = ThreadAndProcessSafeDict(manager)
+  total_tasks = AtomicDict(manager=manager)
 
   # Map from caller_id to a boolean which is True iff all its tasks are
   # finished.
-  call_completed_map = ThreadAndProcessSafeDict(manager)
+  call_completed_map = AtomicDict(manager=manager)
 
   # Used to keep track of the set of return values for each caller ID.
-  global_return_values_map = AtomicIncrementDict(manager)
+  global_return_values_map = AtomicDict(manager=manager)
 
   # Condition used to notify any waiting threads that a task has finished or
   # that a call to Apply needs a new set of consumer processes.
   need_pool_or_done_cond = manager.Condition()
 
   # Lock used to prevent multiple worker processes from asking the main thread
   # to create a new consumer pool for the same level.
   worker_checking_level_lock = manager.Lock()
 
   # Map from caller_id to the current number of completed tasks for that ID.
-  caller_id_finished_count = AtomicIncrementDict(manager)
+  caller_id_finished_count = AtomicDict(manager=manager)
 
   # Used as a way for the main thread to distinguish between being woken up
   # by another call finishing and being woken up by a call that needs a new set
   # of consumer processes.
   new_pool_needed = multiprocessing.Value('i', 0)
 
   current_max_recursive_level = multiprocessing.Value('i', 0)
 
   # Map from (caller_id, name) to the value of that shared variable.
-  shared_vars_map = AtomicIncrementDict(manager)
-  shared_vars_list_map = ThreadAndProcessSafeDict(manager)
+  shared_vars_map = AtomicDict(manager=manager)
+  shared_vars_list_map = AtomicDict(manager=manager)
 
   # Map from caller_id to calling class.
   class_map = manager.dict()
 
   # Number of tasks that resulted in an exception in calls to Apply().
   failure_count = multiprocessing.Value('i', 0)
 
 
+def InitializeThreadingVariables():
+  """Initializes module-level variables used when running multi-threaded.
+
+  When multiprocessing is not available (or on Windows where only 1 process
+  is used), thread-safe analogs to the multiprocessing global variables
+  must be initialized. This function is the thread-safe analog to
+  InitializeMultiprocessingVariables.
+  """
+  # pylint: disable=global-variable-undefined
+  global global_return_values_map, shared_vars_map, failure_count
+  global caller_id_finished_count, shared_vars_list_map, total_tasks
+  global need_pool_or_done_cond, call_completed_map, class_map
+  global task_queues, caller_id_lock, caller_id_counter
+  caller_id_counter = 0
+  caller_id_finished_count = AtomicDict()
+  caller_id_lock = threading.Lock()
+  call_completed_map = AtomicDict()
+  class_map = AtomicDict()
+  failure_count = 0
+  global_return_values_map = AtomicDict()
+  need_pool_or_done_cond = threading.Condition()
+  shared_vars_list_map = AtomicDict()
+  shared_vars_map = AtomicDict()
+  task_queues = []
+  total_tasks = AtomicDict()
+
+
 # Each subclass of Command must define a property named 'command_spec' that is
 # an instance of the following class.
 CommandSpec = namedtuple('CommandSpec', [
     # Name of command.
     'command_name',
     # Usage synopsis.
     'usage_synopsis',
     # List of command name aliases.
     'command_name_aliases',
     # Min number of args required by this command.
@@ skipping 80 matching lines @@
       args = new_command_args[1:] + args
       self.logger.warn('\n'.join(textwrap.wrap(
           ('You are using a deprecated alias, "%(used_alias)s", for the '
            '"%(command_name)s" command. This will stop working on 9/9/2014. '
            'Please use "%(command_name)s" with the appropriate sub-command in '
            'the future. See "gsutil help %(command_name)s" for details.') %
           {'used_alias': self.command_alias_used,
            'command_name': self.command_name})))
     return args
 
-  def __init__(self, command_runner, args, headers, debug, parallel_operations,
-               bucket_storage_uri_class, gsutil_api_class_map_factory,
-               test_method=None, logging_filters=None,
+  def __init__(self, command_runner, args, headers, debug, trace_token,
+               parallel_operations, bucket_storage_uri_class,
+               gsutil_api_class_map_factory, logging_filters=None,
                command_alias_used=None):
     """Instantiates a Command.
 
     Args:
       command_runner: CommandRunner (for commands built atop other commands).
       args: Command-line args (arg0 = actual arg, not command name ala bash).
       headers: Dictionary containing optional HTTP headers to pass to boto.
       debug: Debug level to pass in to boto connection (range 0..3).
+      trace_token: Trace token to pass to the API implementation.
       parallel_operations: Should command operations be executed in parallel?
       bucket_storage_uri_class: Class to instantiate for cloud StorageUris.
                                 Settable for testing/mocking.
       gsutil_api_class_map_factory: Creates map of cloud storage interfaces.
                                     Settable for testing/mocking.
-      test_method: Optional general purpose method for testing purposes.
-                   Application and semantics of this method will vary by
-                   command and test type.
-      logging_filters: Optional list of logging.Filters to apply to this
+      logging_filters: Optional list of logging. Filters to apply to this
                        command's logger.
       command_alias_used: The alias that was actually used when running this
                           command (as opposed to the "official" command name,
                           which will always correspond to the file name).
 
     Implementation note: subclasses shouldn't need to define an __init__
     method, and instead depend on the shared initialization that happens
     here. If you do define an __init__ method in a subclass you'll need to
     explicitly call super().__init__(). But you're encouraged not to do this,
     because it will make changing the __init__ interface more painful.
     """
     # Save class values from constructor params.
     self.command_runner = command_runner
     self.unparsed_args = args
     self.headers = headers
     self.debug = debug
+    self.trace_token = trace_token
     self.parallel_operations = parallel_operations
     self.bucket_storage_uri_class = bucket_storage_uri_class
     self.gsutil_api_class_map_factory = gsutil_api_class_map_factory
-    self.test_method = test_method
     self.exclude_symlinks = False
     self.recursion_requested = False
     self.all_versions = False
     self.command_alias_used = command_alias_used
 
     # Global instance of a threaded logger object.
     self.logger = CreateGsutilLogger(self.command_name)
     if logging_filters:
       for log_filter in logging_filters:
         self.logger.addFilter(log_filter)
@@ skipping 26 matching lines @@
     default_map = {
         'gs': self.command_spec.gs_default_api,
         's3': ApiSelector.XML
     }
     self.gsutil_api_map = GsutilApiMapFactory.GetApiMap(
         self.gsutil_api_class_map_factory, support_map, default_map)
 
     self.project_id = None
     self.gsutil_api = CloudApiDelegator(
         bucket_storage_uri_class, self.gsutil_api_map,
-        self.logger, debug=self.debug)
+        self.logger, debug=self.debug, trace_token=self.trace_token)
 
     # Cross-platform path to run gsutil binary.
     self.gsutil_cmd = ''
     # If running on Windows, invoke python interpreter explicitly.
     if gslib.util.IS_WINDOWS:
       self.gsutil_cmd += 'python '
     # Add full path to gsutil to make sure we test the correct version.
     self.gsutil_path = gslib.GSUTIL_PATH
     self.gsutil_cmd += self.gsutil_path
 
     # We're treating recursion_requested like it's used by all commands, but
     # only some of the commands accept the -R option.
     if self.sub_opts:
       for o, unused_a in self.sub_opts:
         if o == '-r' or o == '-R':
           self.recursion_requested = True
           break
 
-    self.multiprocessing_is_available = MultiprocessingIsAvailable()[0]
+    self.multiprocessing_is_available = (
+        CheckMultiprocessingAvailableAndInit().is_available)
 
   def RaiseWrongNumberOfArgumentsException(self):
     """Raises exception for wrong number of arguments supplied to command."""
     if len(self.args) < self.command_spec.min_args:
       tail_str = 's' if self.command_spec.min_args > 1 else ''
       message = ('The %s command requires at least %d argument%s.' %
                  (self.command_name, self.command_spec.min_args, tail_str))
     else:
       message = ('The %s command accepts at most %d arguments.' %
                  (self.command_name, self.command_spec.max_args))
@@ skipping 204 matching lines @@
     try:
       if url.IsBucket():
         if self.def_acl:
           if self.canned:
             gsutil_api.PatchBucket(
                 url.bucket_name, apitools_messages.Bucket(),
                 canned_def_acl=self.acl_arg, provider=url.scheme, fields=['id'])
           else:
             def_obj_acl = AclTranslation.JsonToMessage(
                 self.acl_arg, apitools_messages.ObjectAccessControl)
+            if not def_obj_acl:
+              # Use a sentinel value to indicate a private (no entries) default
+              # object ACL.
+              def_obj_acl.append(PRIVATE_DEFAULT_OBJ_ACL)
             bucket_metadata = apitools_messages.Bucket(
                 defaultObjectAcl=def_obj_acl)
             gsutil_api.PatchBucket(url.bucket_name, bucket_metadata,
                                    provider=url.scheme, fields=['id'])
         else:
           if self.canned:
             gsutil_api.PatchBucket(
                 url.bucket_name, apitools_messages.Bucket(),
                 canned_acl=self.acl_arg, provider=url.scheme, fields=['id'])
           else:
@@ skipping 40 matching lines @@
                              self.command_name)
 
     # Determine whether acl_arg names a file containing XML ACL text vs. the
     # string name of a canned ACL.
     if os.path.isfile(acl_arg):
       with codecs.open(acl_arg, 'r', UTF8) as f:
         acl_arg = f.read()
       self.canned = False
     else:
       # No file exists, so expect a canned ACL string.
-      # Canned ACLs are not supported in JSON and we need to use the XML API
-      # to set them.
       # validate=False because we allow wildcard urls.
       storage_uri = boto.storage_uri(
           url_args[0], debug=self.debug, validate=False,
           bucket_storage_uri_class=self.bucket_storage_uri_class)
 
       canned_acls = storage_uri.canned_acls()
       if acl_arg not in canned_acls:
         raise CommandException('Invalid canned ACL "%s".' % acl_arg)
       self.canned = True
 
@@ skipping 95 matching lines @@
           self.WildcardIterator(url_str).IterBuckets(
               bucket_fields=bucket_fields))
     if plurality_iter.IsEmpty():
       raise CommandException('No URLs matched')
     if plurality_iter.HasPlurality():
       raise CommandException(
           '%s matched more than one URL, which is not allowed by the %s '
           'command' % (url_str, self.command_name))
     return list(plurality_iter)[0]
 
-  def _HandleMultiProcessingSigs(self, unused_signal_num,
-                                 unused_cur_stack_frame):
+  def _HandleMultiProcessingSigs(self, signal_num, unused_cur_stack_frame):
     """Handles signals INT AND TERM during a multi-process/multi-thread request.
 
     Kills subprocesses.
 
     Args:
       unused_signal_num: signal generated by ^C.
       unused_cur_stack_frame: Current stack frame.
     """
     # Note: This only works under Linux/MacOS. See
     # https://github.com/GoogleCloudPlatform/gsutil/issues/99 for details
     # about why making it work correctly across OS's is harder and still open.
     ShutDownGsutil()
-    sys.stderr.write('Caught ^C - exiting\n')
+    if signal_num == signal.SIGINT:
+      sys.stderr.write('Caught ^C - exiting\n')
     # Simply calling sys.exit(1) doesn't work - see above bug for details.
     KillProcess(os.getpid())
 
   def GetSingleBucketUrlFromArg(self, arg, bucket_fields=None):
     """Gets a single bucket URL based on the command arguments.
 
     Args:
       arg: String argument to get bucket URL for.
       bucket_fields: Fields to populate for the bucket.
 
@@ skipping 101 matching lines @@
            'update your config file (located at %s) and set '
            '"parallel_process_count = 1".') %
           GetConfigFilePath())))
     self.logger.debug('process count: %d', process_count)
     self.logger.debug('thread count: %d', thread_count)
 
     return (process_count, thread_count)
 
   def _SetUpPerCallerState(self):
     """Set up the state for a caller id, corresponding to one Apply call."""
+    # pylint: disable=global-variable-undefined,global-variable-not-assigned
+    # These variables are initialized in InitializeMultiprocessingVariables or
+    # InitializeThreadingVariables
+    global global_return_values_map, shared_vars_map, failure_count
+    global caller_id_finished_count, shared_vars_list_map, total_tasks
+    global need_pool_or_done_cond, call_completed_map, class_map
+    global task_queues, caller_id_lock, caller_id_counter
     # Get a new caller ID.
     with caller_id_lock:
-      caller_id_counter.value += 1
-      caller_id = caller_id_counter.value
+      if isinstance(caller_id_counter, int):
+        caller_id_counter += 1
+        caller_id = caller_id_counter
+      else:
+        caller_id_counter.value += 1
+        caller_id = caller_id_counter.value
 
     # Create a copy of self with an incremented recursive level. This allows
     # the class to report its level correctly if the function called from it
     # also needs to call Apply.
     cls = copy.copy(self)
     cls.recursive_apply_level += 1
 
     # Thread-safe loggers can't be pickled, so we will remove it here and
     # recreate it later in the WorkerThread. This is not a problem since any
     # logger with the same name will be treated as a singleton.
     cls.logger = None
 
     # Likewise, the default API connection can't be pickled, but it is unused
     # anyway as each thread gets its own API delegator.
     cls.gsutil_api = None
 
     class_map[caller_id] = cls
     total_tasks[caller_id] = -1  # -1 => the producer hasn't finished yet.
     call_completed_map[caller_id] = False
-    caller_id_finished_count.Put(caller_id, 0)
-    global_return_values_map.Put(caller_id, [])
+    caller_id_finished_count[caller_id] = 0
+    global_return_values_map[caller_id] = []
     return caller_id
 
   def _CreateNewConsumerPool(self, num_processes, num_threads):
     """Create a new pool of processes that call _ApplyThreads."""
     processes = []
     task_queue = _NewMultiprocessingQueue()
     task_queues.append(task_queue)
 
     current_max_recursive_level.value += 1
     if current_max_recursive_level.value > MAX_RECURSIVE_DEPTH:
@@ skipping 56 matching lines @@
                       and self.sequential_caller_id == -1)
 
     # We don't honor the fail_on_error flag in the case of multiple threads
     # or processes.
     fail_on_error = fail_on_error and (process_count * thread_count == 1)
 
     # Only check this from the first call in the main thread. Apart from the
     # fact that it's  wasteful to try this multiple times in general, it also
     # will never work when called from a subprocess since we use daemon
     # processes, and daemons can't create other processes.
-    if is_main_thread:
-      if ((not self.multiprocessing_is_available)
-          and thread_count * process_count > 1):
-        # Run the check again and log the appropriate warnings. This was run
-        # before, when the Command object was created, in order to calculate
-        # self.multiprocessing_is_available, but we don't want to print the
-        # warning until we're sure the user actually tried to use multiple
-        # threads or processes.
-        MultiprocessingIsAvailable(logger=self.logger)
+    if (is_main_thread and not self.multiprocessing_is_available and
+        process_count > 1):
+      # Run the check again and log the appropriate warnings. This was run
+      # before, when the Command object was created, in order to calculate
+      # self.multiprocessing_is_available, but we don't want to print the
+      # warning until we're sure the user actually tried to use multiple
+      # threads or processes.
+      CheckMultiprocessingAvailableAndInit(logger=self.logger)
 
-    if self.multiprocessing_is_available:
-      caller_id = self._SetUpPerCallerState()
-    else:
-      self.sequential_caller_id += 1
-      caller_id = self.sequential_caller_id
-
-      if is_main_thread:
-        # pylint: disable=global-variable-undefined
-        global global_return_values_map, shared_vars_map, failure_count
-        global caller_id_finished_count, shared_vars_list_map
-        global_return_values_map = BasicIncrementDict()
-        global_return_values_map.Put(caller_id, [])
-        shared_vars_map = BasicIncrementDict()
-        caller_id_finished_count = BasicIncrementDict()
-        shared_vars_list_map = {}
-        failure_count = 0
+    caller_id = self._SetUpPerCallerState()
 
     # If any shared attributes passed by caller, create a dictionary of
     # shared memory variables for every element in the list of shared
     # attributes.
     if shared_attrs:
       shared_vars_list_map[caller_id] = shared_attrs
       for name in shared_attrs:
-        shared_vars_map.Put((caller_id, name), 0)
+        shared_vars_map[(caller_id, name)] = 0
 
     # Make all of the requested function calls.
-    if self.multiprocessing_is_available and thread_count * process_count > 1:
+    usable_processes_count = (process_count if self.multiprocessing_is_available
+                              else 1)
+    if thread_count * usable_processes_count > 1:
       self._ParallelApply(func, args_iterator, exception_handler, caller_id,
-                          arg_checker, process_count, thread_count,
+                          arg_checker, usable_processes_count, thread_count,
                           should_return_results, fail_on_error)
     else:
       self._SequentialApply(func, args_iterator, exception_handler, caller_id,
                             arg_checker, should_return_results, fail_on_error)
 
     if shared_attrs:
       for name in shared_attrs:
         # This allows us to retain the original value of the shared variable,
         # and simply apply the delta after what was done during the call to
         # apply.
         final_value = (original_shared_vars_values[name] +
-                       shared_vars_map.Get((caller_id, name)))
+                       shared_vars_map.get((caller_id, name)))
         setattr(self, name, final_value)
 
     if should_return_results:
-      return global_return_values_map.Get(caller_id)
+      return global_return_values_map.get(caller_id)
 
   def _MaybeSuggestGsutilDashM(self):
     """Outputs a sugestion to the user to use gsutil -m."""
     if not (boto.config.getint('GSUtil', 'parallel_process_count', 0) == 1 and
             boto.config.getint('GSUtil', 'parallel_thread_count', 0) == 1):
       self.logger.info('\n' + textwrap.fill(
           '==> NOTE: You are performing a sequence of gsutil operations that '
           'may run significantly faster if you instead use gsutil -m %s ...\n'
           'Please see the -m section under "gsutil help options" for further '
           'information about when gsutil -m can be advantageous.'
@@ skipping 44 matching lines @@
       if arg_checker(self, args):
         # Now that we actually have the next argument, perform the task.
         task = Task(func, args, caller_id, exception_handler,
                     should_return_results, arg_checker, fail_on_error)
         worker_thread.PerformTask(task, self)
     if sequential_call_count >= gslib.util.GetTermLines():
       # Output suggestion at end of long run, in case user missed it at the
       # start and it scrolled off-screen.
       self._MaybeSuggestGsutilDashM()
 
+    # If the final iterated argument results in an exception, and that
+    # exception modifies shared_attrs, we need to publish the results.
+    worker_thread.shared_vars_updater.Update(caller_id, self)
+
   # pylint: disable=g-doc-args
   def _ParallelApply(self, func, args_iterator, exception_handler, caller_id,
                      arg_checker, process_count, thread_count,
                      should_return_results, fail_on_error):
     """Dispatches input arguments across a thread/process pool.
 
     Pools are composed of parallel OS processes and/or Python threads,
     based on options (-m or not) and settings in the user's config file.
 
     If only one OS process is requested/available, dispatch requests across
@@ skipping 41 matching lines @@
       RegisterSignalHandler(signal.SIGINT, self._HandleMultiProcessingSigs,
                             is_final_handler=True)
       RegisterSignalHandler(signal.SIGTERM, self._HandleMultiProcessingSigs,
                             is_final_handler=True)
 
     if not task_queues:
       # The process we create will need to access the next recursive level
       # of task queues if it makes a call to Apply, so we always keep around
       # one more queue than we know we need. OTOH, if we don't create a new
       # process, the existing process still needs a task queue to use.
-      task_queues.append(_NewMultiprocessingQueue())
+      if process_count > 1:
+        task_queues.append(_NewMultiprocessingQueue())
+      else:
+        task_queues.append(_NewThreadsafeQueue())
 
     if process_count > 1:  # Handle process pool creation.
       # Check whether this call will need a new set of workers.
 
       # Each worker must acquire a shared lock before notifying the main thread
       # that it needs a new worker pool, so that at most one worker asks for
       # a new worker pool at once.
       try:
         if not is_main_thread:
           worker_checking_level_lock.acquire()
@@ skipping 11 matching lines @@
               need_pool_or_done_cond.wait()
       finally:
         if not is_main_thread:
           worker_checking_level_lock.release()
 
     # If we're running in this process, create a separate task queue. Otherwise,
     # if Apply has already been called with process_count > 1, then there will
     # be consumer pools trying to use our processes.
     if process_count > 1:
       task_queue = task_queues[self.recursive_apply_level]
+    elif self.multiprocessing_is_available:
+      task_queue = _NewMultiprocessingQueue()
     else:
-      task_queue = _NewMultiprocessingQueue()
+      task_queue = _NewThreadsafeQueue()
 
     # Kick off a producer thread to throw tasks in the global task queue. We
     # do this asynchronously so that the main thread can be free to create new
     # consumer pools when needed (otherwise, any thread with a task that needs
     # a new consumer pool must block until we're completely done producing; in
     # the worst case, every worker blocks on such a call and the producer fills
     # up the task queue before it finishes, so we block forever).
     producer_thread = ProducerThread(copy.copy(self), args_iterator, caller_id,
                                      func, task_queue, should_return_results,
                                      exception_handler, arg_checker,
@@ skipping 57 matching lines @@
                              of this thread.
       is_blocking_call: True iff the call to Apply is blocked on this call
                         (which is true iff process_count == 1), implying that
                         _ApplyThreads must behave as a blocking call.
     """
     self._ResetConnectionPool()
     self.recursive_apply_level = recursive_apply_level
 
     task_queue = task_queue or task_queues[recursive_apply_level]
 
+    # Ensure fairness across processes by filling our WorkerPool only with
+    # as many tasks as it has WorkerThreads. This semaphore is acquired each
+    # time that a task is retrieved from the queue and released each time
+    # a task is completed by a WorkerThread.
+    worker_semaphore = threading.BoundedSemaphore(thread_count)
+
     assert thread_count * process_count > 1, (
         'Invalid state, calling command._ApplyThreads with only one thread '
         'and process.')
+    # TODO: Presently, this pool gets recreated with each call to Apply. We
+    # should be able to do it just once, at process creation time.
     worker_pool = WorkerPool(
-        thread_count, self.logger,
+        thread_count, self.logger, worker_semaphore,
         bucket_storage_uri_class=self.bucket_storage_uri_class,
         gsutil_api_map=self.gsutil_api_map, debug=self.debug)
 
     num_enqueued = 0
     while True:
+      worker_semaphore.acquire()
       task = task_queue.get()
       if task.args != ZERO_TASKS_TO_DO_ARGUMENT:
         # If we have no tasks to do and we're performing a blocking call, we
         # need a special signal to tell us to stop - otherwise, we block on
         # the call to task_queue.get() forever.
         worker_pool.AddTask(task)
         num_enqueued += 1
+      else:
+        # No tasks remain; don't block the semaphore on WorkerThread completion.
+        worker_semaphore.release()
 
       if is_blocking_call:
         num_to_do = total_tasks[task.caller_id]
         # The producer thread won't enqueue the last task until after it has
         # updated total_tasks[caller_id], so we know that num_to_do < 0 implies
         # we will do this check again.
         if num_to_do >= 0 and num_enqueued == num_to_do:
           if thread_count == 1:
             return
           else:
@@ skipping 157 matching lines @@
       total_tasks[self.caller_id] = num_tasks
       if not cur_task:
         # This happens if there were zero arguments to be put in the queue.
         cur_task = Task(None, ZERO_TASKS_TO_DO_ARGUMENT, self.caller_id,
                         None, None, None, None)
       self.task_queue.put(cur_task)
 
       # It's possible that the workers finished before we updated total_tasks,
       # so we need to check here as well.
       _NotifyIfDone(self.caller_id,
-                    caller_id_finished_count.Get(self.caller_id))
+                    caller_id_finished_count.get(self.caller_id))
 
 
 class WorkerPool(object):
   """Pool of worker threads to which tasks can be added."""
 
-  def __init__(self, thread_count, logger, bucket_storage_uri_class=None,
-               gsutil_api_map=None, debug=0):
+  def __init__(self, thread_count, logger, worker_semaphore,
+               bucket_storage_uri_class=None, gsutil_api_map=None, debug=0):
     self.task_queue = _NewThreadsafeQueue()
     self.threads = []
     for _ in range(thread_count):
       worker_thread = WorkerThread(
-          self.task_queue, logger,
+          self.task_queue, logger, worker_semaphore=worker_semaphore,
           bucket_storage_uri_class=bucket_storage_uri_class,
           gsutil_api_map=gsutil_api_map, debug=debug)
       self.threads.append(worker_thread)
       worker_thread.start()
 
   def AddTask(self, task):
     self.task_queue.put(task)
 
 
 class WorkerThread(threading.Thread):
   """Thread where all the work will be performed.
 
   This makes the function calls for Apply and takes care of all error handling,
   return value propagation, and shared_vars.
 
   Note that this thread is NOT started upon instantiation because the function-
   calling logic is also used in the single-threaded case.
   """
 
-  def __init__(self, task_queue, logger, bucket_storage_uri_class=None,
-               gsutil_api_map=None, debug=0):
+  def __init__(self, task_queue, logger, worker_semaphore=None,
+               bucket_storage_uri_class=None, gsutil_api_map=None, debug=0):
     """Initializes the worker thread.
 
     Args:
       task_queue: The thread-safe queue from which this thread should obtain
                   its work.
       logger: Logger to use for this thread.
+      worker_semaphore: threading.BoundedSemaphore to be released each time a
+          task is completed, or None for single-threaded execution.
       bucket_storage_uri_class: Class to instantiate for cloud StorageUris.
                                 Settable for testing/mocking.
       gsutil_api_map: Map of providers and API selector tuples to api classes
                       which can be used to communicate with those providers.
                       Used for the instantiating CloudApiDelegator class.
       debug: debug level for the CloudApiDelegator class.
     """
     super(WorkerThread, self).__init__()
     self.task_queue = task_queue
+    self.worker_semaphore = worker_semaphore
     self.daemon = True
     self.cached_classes = {}
     self.shared_vars_updater = _SharedVariablesUpdater()
 
     self.thread_gsutil_api = None
     if bucket_storage_uri_class and gsutil_api_map:
       self.thread_gsutil_api = CloudApiDelegator(
           bucket_storage_uri_class, gsutil_api_map, logger, debug=debug)
 
   def PerformTask(self, task, cls):
     """Makes the function call for a task.
 
     Args:
       task: The Task to perform.
       cls: The instance of a class which gives context to the functions called
            by the Task's function. E.g., see SetAclFuncWrapper.
     """
     caller_id = task.caller_id
     try:
       results = task.func(cls, task.args, thread_state=self.thread_gsutil_api)
       if task.should_return_results:
-        global_return_values_map.Update(caller_id, [results], default_value=[])
+        global_return_values_map.Increment(caller_id, [results],
+                                           default_value=[])
     except Exception, e:  # pylint: disable=broad-except
       _IncrementFailureCount()
       if task.fail_on_error:
         raise  # Only happens for single thread and process case.
       else:
         try:
           task.exception_handler(cls, e)
         except Exception, _:  # pylint: disable=broad-except
           # Don't allow callers to raise exceptions here and kill the worker
           # threads.
           cls.logger.debug(
               'Caught exception while handling exception for %s:\n%s',
               task, traceback.format_exc())
     finally:
+      if self.worker_semaphore:
+        self.worker_semaphore.release()
       self.shared_vars_updater.Update(caller_id, cls)
 
       # Even if we encounter an exception, we still need to claim that that
       # the function finished executing. Otherwise, we won't know when to
       # stop waiting and return results.
-      num_done = caller_id_finished_count.Update(caller_id, 1)
-
-      if cls.multiprocessing_is_available:
-        _NotifyIfDone(caller_id, num_done)
+      num_done = caller_id_finished_count.Increment(caller_id, 1)
+      _NotifyIfDone(caller_id, num_done)
 
   def run(self):
     while True:
       task = self.task_queue.get()
       caller_id = task.caller_id
 
       # Get the instance of the command with the appropriate context.
       cls = self.cached_classes.get(caller_id, None)
       if not cls:
         cls = copy.copy(class_map[caller_id])
@@ skipping 39 matching lines @@
         key = (caller_id, name)
         last_value = self.last_shared_var_values.get(key, 0)
         # Compute the change made since the last time we updated here. This is
         # calculated by simply subtracting the last known value from the current
         # value in the class instance.
         delta = getattr(cls, name) - last_value
         self.last_shared_var_values[key] = delta + last_value
 
         # Update the globally-consistent value by simply increasing it by the
         # computed delta.
-        shared_vars_map.Update(key, delta)
+        shared_vars_map.Increment(key, delta)
 
 
 def _NotifyIfDone(caller_id, num_done):
   """Notify any threads waiting for results that something has finished.
 
   Each waiting thread will then need to check the call_completed_map to see if
   its work is done.
 
   Note that num_done could be calculated here, but it is passed in as an
   optimization so that we have one less call to a globally-locked data
@@ skipping 46 matching lines @@
 def ResetFailureCount():
   """Resets the failure_count variable to 0 - useful if error is expected."""
   try:
     global failure_count
     if isinstance(failure_count, int):
       failure_count = 0
     else:  # It's a multiprocessing.Value() of type 'i'.
       failure_count = multiprocessing.Value('i', 0)
   except NameError:  # If it wasn't initialized, Apply() wasn't called.
     pass