class Command(HelpProvider, GcloudStorageCommandMixin):
  """Base class for all gsutil commands."""

  # Each subclass must override this with an instance of CommandSpec.
  command_spec = None

  _commands_with_subcommands_and_subopts = ('acl', 'defacl', 'iam', 'kms',
                                            'label', 'logging', 'notification',
                                            'retention', 'web')

  # This keeps track of the recursive depth of the current call to Apply.
  recursive_apply_level = 0

  # If the multiprocessing module isn't available, we'll use this to keep track
  # of the caller_id.
  sequential_caller_id = -1

  @staticmethod
  def CreateCommandSpec(command_name,
                        usage_synopsis=None,
                        command_name_aliases=None,
                        min_args=0,
                        max_args=NO_MAX,
                        supported_sub_args='',
                        file_url_ok=False,
                        provider_url_ok=False,
                        urls_start_arg=0,
                        gs_api_support=None,
                        gs_default_api=None,
                        supported_private_args=None,
                        argparse_arguments=None):
    """Creates an instance of CommandSpec, with defaults."""
    return CommandSpec(command_name=command_name,
                       usage_synopsis=usage_synopsis,
                       command_name_aliases=command_name_aliases or [],
                       min_args=min_args,
                       max_args=max_args,
                       supported_sub_args=supported_sub_args,
                       file_url_ok=file_url_ok,
                       provider_url_ok=provider_url_ok,
                       urls_start_arg=urls_start_arg,
                       gs_api_support=gs_api_support or [ApiSelector.XML],
                       gs_default_api=gs_default_api or ApiSelector.XML,
                       supported_private_args=supported_private_args,
                       argparse_arguments=argparse_arguments or [])

  # Define a convenience property for command name, since it's used many places.
  def _GetDefaultCommandName(self):
    return self.command_spec.command_name

  command_name = property(_GetDefaultCommandName)

  def _CalculateUrlsStartArg(self):
    """Calculate the index in args of the first URL arg.

    Returns:
      Index of the first URL arg (according to the command spec).
    """
    return self.command_spec.urls_start_arg

  def _TranslateDeprecatedAliases(self, args):
    """Map deprecated aliases to the corresponding new command, and warn."""
    new_command_args = OLD_ALIAS_MAP.get(self.command_alias_used, None)
    if new_command_args:
      # Prepend any subcommands for the new command. The command name itself
      # is not part of the args, so leave it out.
      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,
               trace_token,
               parallel_operations,
               bucket_storage_uri_class,
               gsutil_api_class_map_factory,
               logging_filters=None,
               command_alias_used=None,
               perf_trace_token=None,
               user_project=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.
      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).
      perf_trace_token: Performance measurement trace token to use when making
          API calls.
      user_project: Project to be billed for this request.

    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.
    super().__init__()
    self.command_runner = command_runner
    self.unparsed_args = args
    self.headers = headers
    self.debug = debug
    self.trace_token = trace_token
    self.perf_trace_token = perf_trace_token
    self.parallel_operations = parallel_operations
    self.user_project = user_project
    self.bucket_storage_uri_class = bucket_storage_uri_class
    self.gsutil_api_class_map_factory = gsutil_api_class_map_factory
    self.exclude_symlinks = False
    self.recursion_requested = False
    self.all_versions = False
    self.command_alias_used = command_alias_used
    self.seek_ahead_gsutil_api = None
    # pylint: disable=global-variable-not-assigned
    # pylint: disable=global-variable-undefined
    global ui_controller
    # pylint: enable=global-variable-undefined
    # pylint: enable=global-variable-not-assigned
    # Global instance of a threaded logger object.
    self.logger = CreateOrGetGsutilLogger(self.command_name)
    if logging_filters:
      for log_filter in logging_filters:
        self.logger.addFilter(log_filter)

    if self.command_spec is None:
      raise CommandException('"%s" command implementation is missing a '
                             'command_spec definition.' % self.command_name)

    self.quiet_mode = not self.logger.isEnabledFor(logging.INFO)
    ui_controller = UIController(quiet_mode=self.quiet_mode,
                                 dump_status_messages_file=boto.config.get(
                                     'GSUtil', 'dump_status_messages_file',
                                     None))

    # Parse and validate args.
    self.args = self._TranslateDeprecatedAliases(args)
    self.ParseSubOpts()

    # Named tuple public functions start with _
    # pylint: disable=protected-access
    self.command_spec = self.command_spec._replace(
        urls_start_arg=self._CalculateUrlsStartArg())

    if (len(self.args) < self.command_spec.min_args or
        len(self.args) > self.command_spec.max_args):
      self.RaiseWrongNumberOfArgumentsException()

    if self.command_name not in self._commands_with_subcommands_and_subopts:
      self.CheckArguments()

    # Build the support and default maps from the command spec.
    support_map = {
        'gs': self.command_spec.gs_api_support,
        's3': [ApiSelector.XML]
    }
    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(self.bucket_storage_uri_class,
                                        self.gsutil_api_map,
                                        self.logger,
                                        MainThreadUIQueue(
                                            sys.stderr, ui_controller),
                                        debug=self.debug,
                                        trace_token=self.trace_token,
                                        perf_trace_token=self.perf_trace_token,
                                        user_project=self.user_project)
    # Cross-platform path to run gsutil binary.
    self.gsutil_cmd = ''
    # If running on Windows, invoke python interpreter explicitly.
    if 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 = (
        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))
    message += ' Usage:\n%s\nFor additional help run:\n  gsutil help %s' % (
        self.command_spec.usage_synopsis, self.command_name)
    raise CommandException(message)

  def RaiseInvalidArgumentException(self):
    """Raises exception for specifying an invalid argument to command."""
    message = ('Incorrect option(s) specified. Usage:\n%s\n'
               'For additional help run:\n  gsutil help %s' %
               (self.command_spec.usage_synopsis, self.command_name))
    raise CommandException(message)

  def ParseSubOpts(self,
                   check_args=False,
                   args=None,
                   should_update_sub_opts_and_args=True):
    """Parses sub-opt args.

    Args:
      check_args: True to have CheckArguments() called after parsing.
      args: List of args. If None, self.args will be used.
      should_update_sub_opts_and_args: True if self.sub_opts and self.args
        should be updated with the values returned after parsing. Else return a 
        tuple of sub_opts, args returned by getopt.getopt. This is done
        to allow this method to be called from get_gcloud_storage_args in which
        case we do not want to update self.sub_opts and self.args.

    Raises:
      RaiseInvalidArgumentException: Invalid args specified.
    """
    if args is None:
      unparsed_args = self.args
    else:
      unparsed_args = args
    try:
      parsed_sub_opts, parsed_args = getopt.getopt(
          unparsed_args, self.command_spec.supported_sub_args,
          self.command_spec.supported_private_args or [])
    except getopt.GetoptError:
      self.RaiseInvalidArgumentException()
    if should_update_sub_opts_and_args:
      self.sub_opts, self.args = parsed_sub_opts, parsed_args
      if check_args:
        self.CheckArguments()
    else:
      if check_args:
        # This is just for sanity check. Only get_gcloud_storage_args will
        # call this method with should_update_sub_opts_and_args=False, and it
        # does not set check_args to True.
        raise TypeError('Requested to check arguments'
                        ' but sub_opts and args have not been updated.')
      return parsed_sub_opts, parsed_args

  def CheckArguments(self):
    """Checks that command line arguments match the command_spec.

    Any commands in self._commands_with_subcommands_and_subopts are responsible
    for calling this method after handling initial parsing of their arguments.
    This prevents commands with sub-commands as well as options from breaking
    the parsing of getopt.

    TODO: Provide a function to parse commands and sub-commands more
    intelligently once we stop allowing the deprecated command versions.

    Raises:
      CommandException if the arguments don't match.
    """

    if (not self.command_spec.file_url_ok and
        HaveFileUrls(self.args[self.command_spec.urls_start_arg:])):
      raise CommandException('"%s" command does not support "file://" URLs. '
                             'Did you mean to use a gs:// URL?' %
                             self.command_name)
    if (not self.command_spec.provider_url_ok and
        HaveProviderUrls(self.args[self.command_spec.urls_start_arg:])):
      raise CommandException('"%s" command does not support provider-only '
                             'URLs.' % self.command_name)

  def WildcardIterator(self, url_string, all_versions=False):
    """Helper to instantiate gslib.WildcardIterator.

    Args are same as gslib.WildcardIterator interface, but this method fills in
    most of the values from instance state.

    Args:
      url_string: URL string naming wildcard objects to iterate.
      all_versions: If true, the iterator yields all versions of objects
                    matching the wildcard.  If false, yields just the live
                    object version.

    Returns:
      WildcardIterator for use by caller.
    """
    return CreateWildcardIterator(url_string,
                                  self.gsutil_api,
                                  all_versions=all_versions,
                                  project_id=self.project_id,
                                  logger=self.logger)

  def GetSeekAheadGsutilApi(self):
    """Helper to instantiate a Cloud API instance for a seek-ahead iterator.

    This must be separate from the core command.gsutil_api instance for
    thread-safety, since other iterators typically use that instance and the
    SeekAheadIterator operates in parallel.

    Returns:
      Cloud API instance for use by the seek-ahead iterator.
    """
    # This is initialized in Initialize(Multiprocessing|Threading)Variables
    # pylint: disable=global-variable-not-assigned
    # pylint: disable=global-variable-undefined
    global glob_status_queue
    # pylint: enable=global-variable-not-assigned
    # pylint: enable=global-variable-undefined
    if not self.seek_ahead_gsutil_api:
      self.seek_ahead_gsutil_api = CloudApiDelegator(
          self.bucket_storage_uri_class,
          self.gsutil_api_map,
          logging.getLogger('dummy'),
          glob_status_queue,
          debug=self.debug,
          trace_token=self.trace_token,
          perf_trace_token=self.perf_trace_token,
          user_project=self.user_project)
    return self.seek_ahead_gsutil_api

  def RunCommand(self):
    """Abstract function in base class. Subclasses must implement this.

    The return value of this function will be used as the exit status of the
    process, so subclass commands should return an integer exit code (0 for
    success, a value in [1,255] for failure).
    """
    raise CommandException('Command %s is missing its RunCommand() '
                           'implementation' % self.command_name)

  ############################################################
  # Shared helper functions that depend on base class state. #
  ############################################################

  # TODO: Refactor ACL functions to a different module and pass the
  # command object as state, as opposed to defining them as member functions
  # of the command class.
  def ApplyAclFunc(self,
                   acl_func,
                   acl_excep_handler,
                   url_strs,
                   object_fields=None):
    """Sets the standard or default object ACL depending on self.command_name.

    Args:
      acl_func: ACL function to be passed to Apply.
      acl_excep_handler: ACL exception handler to be passed to Apply.
      url_strs: URL strings on which to set ACL.
      object_fields: If present, list of object metadata fields to retrieve;
          if None, default name expansion iterator fields will be used.

    Raises:
      CommandException if an ACL could not be set.
    """
    multi_threaded_url_args = []
    # Handle bucket ACL setting operations single-threaded, because
    # our threading machinery currently assumes it's working with objects
    # (name_expansion_iterator), and normally we wouldn't expect users to need
    # to set ACLs on huge numbers of buckets at once anyway.
    for url_str in url_strs:
      url = StorageUrlFromString(url_str)
      if url.IsCloudUrl() and url.IsBucket():
        if self.recursion_requested:
          # If user specified -R option, convert any bucket args to bucket
          # wildcards (e.g., gs://bucket/*), to prevent the operation from
          # being applied to the buckets themselves.
          url.object_name = '*'
          multi_threaded_url_args.append(url.url_string)
        else:
          # Convert to a NameExpansionResult so we can re-use the threaded
          # function for the single-threaded implementation.  RefType is unused.
          for blr in self.WildcardIterator(
              url.url_string).IterBuckets(bucket_fields=['id']):
            name_expansion_for_url = NameExpansionResult(
                url, False, False, blr.storage_url, None)
            acl_func(self, name_expansion_for_url)
      else:
        multi_threaded_url_args.append(url_str)

    if len(multi_threaded_url_args) >= 1:
      name_expansion_iterator = NameExpansionIterator(
          self.command_name,
          self.debug,
          self.logger,
          self.gsutil_api,
          multi_threaded_url_args,
          self.recursion_requested,
          all_versions=self.all_versions,
          continue_on_error=self.continue_on_error or self.parallel_operations,
          bucket_listing_fields=object_fields)

      seek_ahead_iterator = SeekAheadNameExpansionIterator(
          self.command_name,
          self.debug,
          self.GetSeekAheadGsutilApi(),
          multi_threaded_url_args,
          self.recursion_requested,
          all_versions=self.all_versions)

      # Perform requests in parallel (-m) mode, if requested, using
      # configured number of parallel processes and threads. Otherwise,
      # perform requests with sequential function calls in current process.
      self.Apply(acl_func,
                 name_expansion_iterator,
                 acl_excep_handler,
                 fail_on_error=not self.continue_on_error,
                 seek_ahead_iterator=seek_ahead_iterator)

    if not self.everything_set_okay and not self.continue_on_error:
      raise CommandException('ACLs for some objects could not be set.')

  def SetAclFunc(self, name_expansion_result, thread_state=None):
    """Sets the object ACL for the name_expansion_result provided.

    Args:
      name_expansion_result: NameExpansionResult describing the target object.
      thread_state: If present, use this gsutil Cloud API instance for the set.
    """
    if thread_state:
      assert not self.def_acl
      gsutil_api = thread_state
    else:
      gsutil_api = self.gsutil_api
    op_string = 'default object ACL' if self.def_acl else 'ACL'
    url = name_expansion_result.expanded_storage_url
    self.logger.info('Setting %s on %s...', op_string, url)
    if (gsutil_api.GetApiSelector(url.scheme) == ApiSelector.XML and
        url.scheme != 'gs'):
      # If we are called with a non-google ACL model, we need to use the XML
      # passthrough. acl_arg should either be a canned ACL or an XML ACL.
      self._SetAclXmlPassthrough(url, gsutil_api)
    else:
      # Normal Cloud API path. acl_arg is a JSON ACL or a canned ACL.
      self._SetAclGsutilApi(url, gsutil_api)
    PutToQueueWithTimeout(gsutil_api.status_queue,
                          MetadataMessage(message_time=time.time()))

  def _SetAclXmlPassthrough(self, url, gsutil_api):
    """Sets the ACL for the URL provided using the XML passthrough functions.

    This function assumes that self.def_acl, self.canned,
    and self.continue_on_error are initialized, and that self.acl_arg is
    either an XML string or a canned ACL string.

    Args:
      url: CloudURL to set the ACL on.
      gsutil_api: gsutil Cloud API to use for the ACL set. Must support XML
          passthrough functions.
    """
    try:
      orig_prefer_api = gsutil_api.prefer_api
      gsutil_api.prefer_api = ApiSelector.XML
      gsutil_api.XmlPassThroughSetAcl(self.acl_arg,
                                      url,
                                      canned=self.canned,
                                      def_obj_acl=self.def_acl,
                                      provider=url.scheme)
    except ServiceException as e:
      if self.continue_on_error:
        self.everything_set_okay = False
        self.logger.error(e)
      else:
        raise
    finally:
      gsutil_api.prefer_api = orig_prefer_api

  def _SetAclGsutilApi(self, url, gsutil_api):
    """Sets the ACL for the URL provided using the gsutil Cloud API.

    This function assumes that self.def_acl, self.canned,
    and self.continue_on_error are initialized, and that self.acl_arg is
    either a JSON string or a canned ACL string.

    Args:
      url: CloudURL to set the ACL on.
      gsutil_api: gsutil Cloud API to use for the ACL set.
    """
    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:
            bucket_acl = AclTranslation.JsonToMessage(
                self.acl_arg, apitools_messages.BucketAccessControl)
            bucket_metadata = apitools_messages.Bucket(acl=bucket_acl)
            gsutil_api.PatchBucket(url.bucket_name,
                                   bucket_metadata,
                                   provider=url.scheme,
                                   fields=['id'])
      else:  # url.IsObject()
        if self.canned:
          gsutil_api.PatchObjectMetadata(url.bucket_name,
                                         url.object_name,
                                         apitools_messages.Object(),
                                         provider=url.scheme,
                                         generation=url.generation,
                                         canned_acl=self.acl_arg)
        else:
          object_acl = AclTranslation.JsonToMessage(
              self.acl_arg, apitools_messages.ObjectAccessControl)
          object_metadata = apitools_messages.Object(acl=object_acl)
          gsutil_api.PatchObjectMetadata(url.bucket_name,
                                         url.object_name,
                                         object_metadata,
                                         provider=url.scheme,
                                         generation=url.generation)
    except ArgumentException as e:
      raise
    except ServiceException as e:
      if self.continue_on_error:
        self.everything_set_okay = False
        self.logger.error(e)
      else:
        raise

  def SetAclCommandHelper(self, acl_func, acl_excep_handler):
    """Sets ACLs on the self.args using the passed-in acl function.

    Args:
      acl_func: ACL function to be passed to Apply.
      acl_excep_handler: ACL exception handler to be passed to Apply.
    """
    acl_arg = self.args[0]
    url_args = self.args[1:]
    # Disallow multi-provider setacl requests, because there are differences in
    # the ACL models.
    if not UrlsAreForSingleProvider(url_args):
      raise CommandException('"%s" command spanning providers not allowed.' %
                             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.
      # 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

    # Used to track if any ACLs failed to be set.
    self.everything_set_okay = True
    self.acl_arg = acl_arg

    self.ApplyAclFunc(acl_func, acl_excep_handler, url_args)
    if not self.everything_set_okay and not self.continue_on_error:
      raise CommandException('ACLs for some objects could not be set.')

  def _WarnServiceAccounts(self):
    """Warns service account users who have received an AccessDenied error.

    When one of the metadata-related commands fails due to AccessDenied, user
    must ensure that they are listed as an Owner in the API console.
    """
    # Import this here so that the value will be set first in
    # gcs_oauth2_boto_plugin.
    # pylint: disable=g-import-not-at-top
    from gcs_oauth2_boto_plugin.oauth2_plugin import IS_SERVICE_ACCOUNT

    if IS_SERVICE_ACCOUNT:
      # This method is only called when canned ACLs are used, so the warning
      # definitely applies.
      self.logger.warning('\n'.join(
          textwrap.wrap(
              'It appears that your service account has been denied access while '
              'attempting to perform a metadata operation. If you believe that you '
              'should have access to this metadata (i.e., if it is associated with '
              'your account), please make sure that your service account'
              's email '
              'address is listed as an Owner in the Permissions tab of the API '
              'console. See "gsutil help creds" for further information.\n')))

  def GetAndPrintAcl(self, url_str):
    """Prints the standard or default object ACL depending on self.command_name.

    Args:
      url_str: URL string to get ACL for.
    """
    blr = self.GetAclCommandBucketListingReference(url_str)
    url = StorageUrlFromString(url_str)
    if (self.gsutil_api.GetApiSelector(url.scheme) == ApiSelector.XML and
        url.scheme != 'gs'):
      # Need to use XML passthrough.
      try:
        acl = self.gsutil_api.XmlPassThroughGetAcl(url,
                                                   def_obj_acl=self.def_acl,
                                                   provider=url.scheme)
        print(acl.to_xml())
      except AccessDeniedException as _:
        self._WarnServiceAccounts()
        raise
    else:
      if self.command_name == 'defacl':
        acl = blr.root_object.defaultObjectAcl
        if not acl:
          self.logger.warn(
              'No default object ACL present for %s. This could occur if '
              'the default object ACL is private, in which case objects '
              'created in this bucket will be readable only by their '
              'creators. It could also mean you do not have OWNER permission '
              'on %s and therefore do not have permission to read the '
              'default object ACL. It could also mean that %s has Bucket '
              'Policy Only enabled and therefore object ACLs and default '
              'object ACLs are disabled (see '
              'https://cloud.google.com/storage/docs/bucket-policy-only).',
              url_str, url_str, url_str)
      else:
        acl = blr.root_object.acl
        # Use the access controls api to check if the acl is actually empty or
        # if the user has 403 access denied or 400 invalid argument.
        if not acl:
          self._ListAccessControlsAcl(url)

      print(AclTranslation.JsonFromMessage(acl))

  def _ListAccessControlsAcl(self, storage_url):
    """Returns either bucket or object access controls for a storage url.

    Args:
      storage_url: StorageUrl object representing the bucket or object.

    Returns:
      BucketAccessControls, ObjectAccessControls, or None if storage_url does
      not represent a cloud bucket or cloud object.

    Raises:
      ServiceException if there was an error in the request.
    """
    if storage_url.IsBucket():
      return self.gsutil_api.ListBucketAccessControls(
          storage_url.bucket_name, provider=storage_url.scheme)
    elif storage_url.IsObject():
      return self.gsutil_api.ListObjectAccessControls(
          storage_url.bucket_name,
          storage_url.object_name,
          provider=storage_url.scheme)
    else:
      return None

  def GetAclCommandBucketListingReference(self, url_str):
    """Gets a single bucket listing reference for an acl get command.

    Args:
      url_str: URL string to get the bucket listing reference for.

    Returns:
      BucketListingReference for the URL string.

    Raises:
      CommandException if string did not result in exactly one reference.
    """
    # We're guaranteed by caller that we have the appropriate type of url
    # string for the call (ex. we will never be called with an object string
    # by getdefacl)
    wildcard_url = StorageUrlFromString(url_str)
    if wildcard_url.IsObject():
      plurality_iter = PluralityCheckableIterator(
          self.WildcardIterator(url_str).IterObjects(
              bucket_listing_fields=['acl']))
    else:
      # Bucket or provider.  We call IterBuckets explicitly here to ensure that
      # the root object is populated with the acl.
      if self.command_name == 'defacl':
        bucket_fields = ['defaultObjectAcl']
      else:
        bucket_fields = ['acl']
      plurality_iter = PluralityCheckableIterator(
          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 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.

    Returns:
      (StorageUrl referring to a single bucket, Bucket metadata).

    Raises:
      CommandException if args did not match exactly one bucket.
    """
    plurality_checkable_iterator = self.GetBucketUrlIterFromArg(
        arg, bucket_fields=bucket_fields)
    if plurality_checkable_iterator.HasPlurality():
      raise CommandException('%s matched more than one URL, which is not\n'
                             'allowed by the %s command' %
                             (arg, self.command_name))
    blr = list(plurality_checkable_iterator)[0]
    return StorageUrlFromString(blr.url_string), blr.root_object

  def GetBucketUrlIterFromArg(self, arg, bucket_fields=None):
    """Gets a single bucket URL based on the command arguments.

    Args:
      arg: String argument to iterate over.
      bucket_fields: Fields to populate for the bucket.

    Returns:
      PluralityCheckableIterator over buckets.

    Raises:
      CommandException if iterator matched no buckets.
    """
    arg_url = StorageUrlFromString(arg)
    if not arg_url.IsCloudUrl() or arg_url.IsObject():
      raise CommandException('"%s" command must specify a bucket' %
                             self.command_name)

    plurality_checkable_iterator = PluralityCheckableIterator(
        self.WildcardIterator(arg).IterBuckets(bucket_fields=bucket_fields))
    if plurality_checkable_iterator.IsEmpty():
      raise CommandException('No URLs matched')
    return plurality_checkable_iterator

  ######################
  # Private functions. #
  ######################

  def _ResetConnectionPool(self):
    # Each OS process needs to establish its own set of connections to
    # the server to avoid writes from different OS processes interleaving
    # onto the same socket (and garbling the underlying SSL session).
    # We ensure each process gets its own set of connections here by
    # reinitializing state that tracks connections.
    connection_pool = StorageUri.provider_pool
    if connection_pool:
      for i in connection_pool:
        connection_pool[i].connection.close()

    StorageUri.provider_pool = {}
    StorageUri.connection = None

  def _GetProcessAndThreadCount(self, process_count, thread_count,
                                parallel_operations_override):
    """Determines the values of process_count and thread_count.

    These values are used for parallel operations.
    If we're not performing operations in parallel, then ignore
    existing values and use process_count = thread_count = 1.

    Args:
      process_count: A positive integer or None. In the latter case, we read
                     the value from the .boto config file.
      thread_count: A positive integer or None. In the latter case, we read
                    the value from the .boto config file.
      parallel_operations_override: Used to override self.parallel_operations.
                                    This allows the caller to safely override
                                    the top-level flag for a single call.

    Returns:
      (process_count, thread_count): The number of processes and threads to use,
                                     respectively.
    """
    # Set OS process and python thread count as a function of options
    # and config.
    if self.parallel_operations or parallel_operations_override:
      if not process_count:
        process_count = boto.config.getint(
            'GSUtil', 'parallel_process_count',
            gslib.commands.config.DEFAULT_PARALLEL_PROCESS_COUNT)
      if process_count < 1:
        raise CommandException('Invalid parallel_process_count "%d".' %
                               process_count)
      if not thread_count:
        thread_count = boto.config.getint(
            'GSUtil', 'parallel_thread_count',
            gslib.commands.config.DEFAULT_PARALLEL_THREAD_COUNT)
      if thread_count < 1:
        raise CommandException('Invalid parallel_thread_count "%d".' %
                               thread_count)
    else:
      # If -m not specified, then assume 1 OS process and 1 Python thread.
      process_count = 1
      thread_count = 1

    should_prohibit_multiprocessing, os_name = ShouldProhibitMultiprocessing()
    if should_prohibit_multiprocessing and process_count > 1:
      raise CommandException('\n'.join(
          textwrap.wrap(
              ('It is not possible to set process_count > 1 on %s. Please '
               'update your config file(s) (located at %s) and set '
               '"parallel_process_count = 1".') %
              (os_name, ', '.join(GetFriendlyConfigFilePaths())))))
    is_main_thread = self.recursive_apply_level == 0
    if os_name == 'macOS' and process_count > 1 and is_main_thread:
      self.logger.info(
          'If you experience problems with multiprocessing on MacOS, they '
          'might be related to https://bugs.python.org/issue33725. You can '
          'disable multiprocessing by editing your .boto config or by adding '
          'the following flag to your command: '
          '`-o "GSUtil:parallel_process_count=1"`. Note that multithreading is '
          'still available even if you disable multiprocessing.\n')

    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.Increment()
      caller_id = caller_id_counter.GetValue()

    # 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(s) can't be pickled, but are unused
    # anyway as each thread gets its own API delegator.
    cls.gsutil_api = None
    cls.seek_ahead_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[caller_id] = 0
    global_return_values_map[caller_id] = []
    return caller_id

  def _CreateNewConsumerPool(self, num_processes, num_threads, status_queue):
    """Create a new pool of processes that call _ApplyThreads."""
    processes = []
    task_queue = _NewMultiprocessingQueue()
    task_queues.append(task_queue)

    current_max_recursive_level.Increment()
    if current_max_recursive_level.GetValue() > MAX_RECURSIVE_DEPTH:
      raise CommandException('Recursion depth of Apply calls is too great.')
    for _ in range(num_processes):
      recursive_apply_level = len(consumer_pools)
      p = multiprocessing_context.Process(target=self._ApplyThreads,
                                          args=(num_threads, num_processes,
                                                recursive_apply_level,
                                                status_queue))
      p.daemon = True
      processes.append(p)
      _CryptoRandomAtFork()
      p.start()
    consumer_pool = _ConsumerPool(processes, task_queue)
    consumer_pools.append(consumer_pool)

  class ParallelOverrideReason(object):
    """Enum class to describe purpose of overriding parallel operations."""
    # For the case when we use slice parallelism.
    SLICE = 'slice'
    # For the case when we run a helper Apply call (such as in the _DiffIterator
    # of rsync) and override to make the command go faster.
    SPEED = 'speed'
    # For when we run Apply calls in perfdiag.
    PERFDIAG = 'perfdiag'

  def Apply(self,
            func,
            args_iterator,
            exception_handler,
            shared_attrs=None,
            arg_checker=_UrlArgChecker,
            parallel_operations_override=None,
            process_count=None,
            thread_count=None,
            should_return_results=False,
            fail_on_error=False,
            seek_ahead_iterator=None):
    """Calls _Parallel/SequentialApply based on multiprocessing availability.

    Args:
      func: Function to call to process each argument.
      args_iterator: Iterable collection of arguments to be put into the
                     work queue.
      exception_handler: Exception handler for WorkerThread class.
      shared_attrs: List of attributes to manage across sub-processes.
      arg_checker: Used to determine whether we should process the current
                   argument or simply skip it. Also handles any logging that
                   is specific to a particular type of argument.
      parallel_operations_override: A string (see ParallelOverrideReason)
                                    describing the reason to override
                                    self.parallel_operations. This allows the
                                    caller to safely override the top-level flag
                                    for a single call.
      process_count: The number of processes to use. If not specified, then
                     the configured default will be used.
      thread_count: The number of threads per process. If not specified, then
                    the configured default will be used..
      should_return_results: If true, then return the results of all successful
                             calls to func in a list.
      fail_on_error: If true, then raise any exceptions encountered when
                     executing func. This is only applicable in the case of
                     process_count == thread_count == 1.
      seek_ahead_iterator: If present, a seek-ahead iterator that will
          provide an approximation of the total number of tasks and bytes that
          will be iterated by the ProducerThread. Used only if multiple
          processes and/or threads are used.

    Returns:
      Results from spawned threads.
    """
    # This is initialized in Initialize(Multiprocessing|Threading)Variables
    # pylint: disable=global-variable-not-assigned
    # pylint: disable=global-variable-undefined
    global thread_stats
    # pylint: enable=global-variable-not-assigned
    # pylint: enable=global-variable-undefined
    if shared_attrs:
      original_shared_vars_values = {}  # We'll add these back in at the end.
      for name in shared_attrs:
        original_shared_vars_values[name] = getattr(self, name)
        # By setting this to 0, we simplify the logic for computing deltas.
        # We'll add it back after all of the tasks have been performed.
        setattr(self, name, 0)

    (process_count, thread_count) = self._GetProcessAndThreadCount(
        process_count, thread_count, parallel_operations_override)

    is_main_thread = (self.recursive_apply_level == 0 and
                      self.sequential_caller_id == -1)

    if is_main_thread:
      # This initializes the initial performance summary parameters.
      LogPerformanceSummaryParams(num_processes=process_count,
                                  num_threads=thread_count)

    # 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 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)

    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[(caller_id, name)] = 0

    # Make all of the requested function calls.
    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,
          usable_processes_count,
          thread_count,
          should_return_results,
          fail_on_error,
          seek_ahead_iterator=seek_ahead_iterator,
          parallel_operations_override=parallel_operations_override)
      if is_main_thread:
        _AggregateThreadStats()
    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)))
        setattr(self, name, final_value)

    if should_return_results:
      return global_return_values_map.get(caller_id)

  def _MaybeSuggestGsutilDashM(self):
    """Outputs a suggestion 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.' %
          self.command_spec.command_name) + '\n')

  # pylint: disable=g-doc-args
  def _SequentialApply(self, func, args_iterator, exception_handler, caller_id,
                       arg_checker, should_return_results, fail_on_error):
    """Performs all function calls sequentially in the current thread.

    No other threads or processes will be spawned. This degraded functionality
    is used when the multiprocessing module is not available or the user
    requests only one thread and one process.
    """
    # Create a WorkerThread to handle all of the logic needed to actually call
    # the function. Note that this thread will never be started, and all work
    # is done in the current thread.
    worker_thread = WorkerThread(None, False, user_project=self.user_project)
    args_iterator = iter(args_iterator)
    # Count of sequential calls that have been made. Used for producing
    # suggestion to use gsutil -m.
    sequential_call_count = 0
    while True:

      # Try to get the next argument, handling any exceptions that arise.
      try:
        args = next(args_iterator)
      except StopIteration as e:
        break
      except Exception as e:  # pylint: disable=broad-except
        _IncrementFailureCount()
        if fail_on_error:
          raise
        else:
          try:
            exception_handler(self, e)
          except Exception as _:  # pylint: disable=broad-except
            self.logger.debug(
                'Caught exception while handling exception for %s:\n%s', func,
                traceback.format_exc())
          continue

      sequential_call_count += 1
      if (sequential_call_count == OFFER_GSUTIL_M_SUGGESTION_THRESHOLD or
          sequential_call_count % OFFER_GSUTIL_M_SUGGESTION_FREQUENCY == 0):
        # Output suggestion near beginning of run, so user sees it early, and
        # every so often while the command is executing, so they can ^C and try
        # gsutil -m.
        self._MaybeSuggestGsutilDashM()
      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)

    lines_since_suggestion_last_printed = (sequential_call_count %
                                           OFFER_GSUTIL_M_SUGGESTION_FREQUENCY)
    if lines_since_suggestion_last_printed >= GetTermLines():
      # Output suggestion at end of long run, in case user missed it at the
      # start and it scrolled off-screen.
      self._MaybeSuggestGsutilDashM()

    PutToQueueWithTimeout(self.gsutil_api.status_queue,
                          FinalMessage(time.time()))

    # 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)

    # Now that all the work is done, log the types of source URLs encountered.
    self._ProcessSourceUrlTypes(args_iterator)

  # 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,
                     seek_ahead_iterator=None,
                     parallel_operations_override=None):
    r"""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
    threads in the current OS process.

    In the multi-process case, we will create one pool of worker processes for
    each level of the tree of recursive calls to Apply. E.g., if A calls
    Apply(B), and B ultimately calls Apply(C) followed by Apply(D), then we
    will only create two sets of worker processes - B will execute in the first,
    and C and D will execute in the second. If C is then changed to call
    Apply(E) and D is changed to call Apply(F), then we will automatically
    create a third set of processes (lazily, when needed) that will be used to
    execute calls to E and F. This might look something like:

    Pool1 Executes:                B
                                  / \
    Pool2 Executes:              C   D
                                /     \
    Pool3 Executes:            E       F

    Apply's parallelism is generally broken up into 4 cases:
    - If process_count == thread_count == 1, then all tasks will be executed
      by _SequentialApply.
    - If process_count > 1 and thread_count == 1, then the main thread will
      create a new pool of processes (if they don't already exist) and each of
      those processes will execute the tasks in a single thread.
    - If process_count == 1 and thread_count > 1, then this process will create
      a new pool of threads to execute the tasks.
    - If process_count > 1 and thread_count > 1, then the main thread will
      create a new pool of processes (if they don't already exist) and each of
      those processes will, upon creation, create a pool of threads to
      execute the tasks.

    Args:
      caller_id: The caller ID unique to this call to command.Apply.
      See command.Apply for description of other arguments.
    """
    # This is initialized in Initialize(Multiprocessing|Threading)Variables
    # pylint: disable=global-variable-not-assigned
    # pylint: disable=global-variable-undefined
    global glob_status_queue, ui_controller
    # pylint: enable=global-variable-not-assigned
    # pylint: enable=global-variable-undefined
    is_main_thread = self.recursive_apply_level == 0

    if (parallel_operations_override == self.ParallelOverrideReason.SLICE and
        self.recursive_apply_level <= 1):
      # The operation uses slice parallelism if the recursive apply level > 0 or
      # if we're executing _ParallelApply without the -m option.
      glob_status_queue.put(PerformanceSummaryMessage(time.time(), True))

    if not IS_WINDOWS and is_main_thread:
      # For multi-thread or multi-process scenarios, the main process must
      # kill itself on a terminating signal, because sys.exit(1) only exits
      # the currently executing thread, leaving orphaned processes. The main
      # thread is responsible for cleaning up multiprocessing variables such
      # as manager processes. Therefore, the main thread's signal handling
      # chain is:
      # 1: __main__._CleanupSignalHandler (clean up processes)
      # 2: MultithreadedSignalHandler (kill self)
      for signal_num in (signal.SIGINT, signal.SIGTERM):
        RegisterSignalHandler(signal_num,
                              MultithreadedMainSignalHandler,
                              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.
      if process_count > 1:
        task_queues.append(_NewMultiprocessingQueue())
      else:
        task_queue = _NewThreadsafeQueue()
        task_queues.append(task_queue)
        # Create a top-level worker pool since this is the first execution
        # of ParallelApply on the main thread.
        WorkerPool(thread_count,
                   self.logger,
                   task_queue=task_queue,
                   bucket_storage_uri_class=self.bucket_storage_uri_class,
                   gsutil_api_map=self.gsutil_api_map,
                   debug=self.debug,
                   status_queue=glob_status_queue,
                   user_project=self.user_project)

    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()
        if self.recursive_apply_level >= current_max_recursive_level.GetValue():
          with need_pool_or_done_cond:
            # Only the main thread is allowed to create new processes -
            # otherwise, we will run into some Python bugs.
            if is_main_thread:
              self._CreateNewConsumerPool(process_count, thread_count,
                                          glob_status_queue)
            else:
              # Notify the main thread that we need a new consumer pool.
              new_pool_needed.Reset(reset_value=1)
              need_pool_or_done_cond.notify_all()
              # The main thread will notify us when it finishes.
              need_pool_or_done_cond.wait()
      finally:
        if not is_main_thread:
          worker_checking_level_lock.release()
    else:  # Handle new worker thread pool creation.
      if not is_main_thread:
        try:
          worker_checking_level_lock.acquire()
          if self.recursive_apply_level > _GetCurrentMaxRecursiveLevel():
            # We don't have a thread pool for this level of recursive apply
            # calls, so create a pool and corresponding task queue.
            _IncrementCurrentMaxRecursiveLevel()
            task_queue = _NewThreadsafeQueue()
            task_queues.append(task_queue)
            WorkerPool(thread_count,
                       self.logger,
                       task_queue=task_queue,
                       bucket_storage_uri_class=self.bucket_storage_uri_class,
                       gsutil_api_map=self.gsutil_api_map,
                       debug=self.debug,
                       status_queue=glob_status_queue,
                       user_project=self.user_project)
        finally:
          worker_checking_level_lock.release()

    task_queue = task_queues[self.recursive_apply_level]

    # Only use the seek-ahead iterator in the main thread to provide an
    # overall estimate of operations.
    if seek_ahead_iterator and not is_main_thread:
      seek_ahead_iterator = None

    # 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,
        fail_on_error,
        seek_ahead_iterator=seek_ahead_iterator,
        status_queue=(glob_status_queue if is_main_thread else None))

    # Start the UI thread that is responsible for displaying operation status
    # (aggregated across processes and threads) to the user.
    ui_thread = None
    if is_main_thread:
      ui_thread = UIThread(glob_status_queue, sys.stderr, ui_controller)

    # Wait here until either:
    #   1. We're the main thread in the multi-process case, and someone needs
    #      a new consumer pool - in which case we create one and continue
    #      waiting.
    #   2. Someone notifies us that all of the work we requested is done, in
    #      which case we retrieve the results (if applicable) and stop
    #      waiting.
    # At most one of these can be true, because the main thread is blocked on
    # its call to Apply, and a thread will not ask for a new consumer pool
    # unless it had more work to do.
    while True:
      with need_pool_or_done_cond:
        if call_completed_map[caller_id]:
          break
        elif (process_count > 1 and is_main_thread and
              new_pool_needed.GetValue()):
          new_pool_needed.Reset()
          self._CreateNewConsumerPool(process_count, thread_count,
                                      glob_status_queue)
          need_pool_or_done_cond.notify_all()

        # Note that we must check the above conditions before the wait() call;
        # otherwise, the notification can happen before we start waiting, in
        # which case we'll block forever.
        need_pool_or_done_cond.wait()

    # We've completed all tasks (or excepted), so signal the UI thread to
    # terminate.
    if is_main_thread:
      PutToQueueWithTimeout(glob_status_queue, ZERO_TASKS_TO_DO_ARGUMENT)
      ui_thread.join(timeout=UI_THREAD_JOIN_TIMEOUT)
      # Now that all the work is done, log the types of source URLs encountered.
      self._ProcessSourceUrlTypes(producer_thread.args_iterator)

    # We encountered an exception from the producer thread before any arguments
    # were enqueued, but it wouldn't have been propagated, so we'll now
    # explicitly raise it here.
    if producer_thread.unknown_exception:
      # pylint: disable=raising-bad-type
      raise producer_thread.unknown_exception

    # We encountered an exception from the producer thread while iterating over
    # the arguments, so raise it here if we're meant to fail on error.
    if producer_thread.iterator_exception and fail_on_error:
      # pylint: disable=raising-bad-type
      raise producer_thread.iterator_exception
    if is_main_thread and not parallel_operations_override:
      PutToQueueWithTimeout(glob_status_queue, FinalMessage(time.time()))

  def _ProcessSourceUrlTypes(self, args_iterator):
    """Logs the URL type information to analytics collection."""
    if not isinstance(args_iterator, CopyObjectsIterator):
      return
    LogPerformanceSummaryParams(is_daisy_chain=args_iterator.is_daisy_chain,
                                has_file_src=args_iterator.has_file_src,
                                has_cloud_src=args_iterator.has_cloud_src,
                                provider_types=args_iterator.provider_types)

  def _ApplyThreads(self, thread_count, process_count, recursive_apply_level,
                    status_queue):
    """Assigns the work from the multi-process global task queue.

    Work is assigned to an individual process for later consumption either by
    the WorkerThreads or (if thread_count == 1) this thread.

    Args:
      thread_count: The number of threads used to perform the work. If 1, then
                    perform all work in this thread.
      process_count: The number of processes used to perform the work.
      recursive_apply_level: The depth in the tree of recursive calls to Apply
                             of this thread.
      status_queue: Multiprocessing/threading queue for progress reporting and
          performance aggregation.
    """
    assert process_count > 1, (
        'Invalid state, calling command._ApplyThreads with only one process.')

    _CryptoRandomAtFork()
    # Separate processes should exit on a terminating signal,
    # but to avoid race conditions only the main process should handle
    # multiprocessing cleanup. Override child processes to use a single signal
    # handler.
    for catch_signal in GetCaughtSignals():
      signal.signal(catch_signal, ChildProcessSignalHandler)

    self._ResetConnectionPool()
    self.recursive_apply_level = recursive_apply_level

    task_queue = 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)

    # 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,
        worker_semaphore=worker_semaphore,
        bucket_storage_uri_class=self.bucket_storage_uri_class,
        gsutil_api_map=self.gsutil_api_map,
        debug=self.debug,
        status_queue=status_queue,
        user_project=self.user_project)

    num_enqueued = 0
    while True:
      while not worker_semaphore.acquire(blocking=False):
        # Because Python signal handlers are only called in between atomic
        # instructions, if we block the main thread on an available worker
        # thread, we won't be able to respond to signals such as a
        # user-initiated CTRL-C until a worker thread completes a task.
        # We poll the semaphore periodically as a compromise between
        # efficiency and user responsiveness.
        time.sleep(0.01)
      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; since no work was dispatched to a thread, don't
        # block the semaphore on a WorkerThread completion.
        worker_semaphore.release() class CommandException(StandardError):
  """Exception raised when a problem is encountered running a gsutil command.

  This exception should be used to signal user errors or system failures
  (like timeouts), not bugs (like an incorrect param value). For the
  latter you should raise Exception so we can see where/how it happened
  via gsutil -D (which will include a stack trace for raised Exceptions).
  """

  def __init__(self, reason, informational=False):
    """Instantiate a CommandException.

    Args:
      reason: Text describing the problem.
      informational: Indicates reason should be printed as FYI, not a failure.
    """
    StandardError.__init__(self)
    self.reason = reason
    self.informational = informational

  def __repr__(self):
    return str(self)

  def __str__(self):
    return 'CommandException: %s' % self.reason def ContainsWildcard(url_string):
  """Checks whether url_string contains a wildcard.

  Args:
    url_string: URL string to check.

  Returns:
    bool indicator.
  """
  return bool(WILDCARD_REGEX.search(url_string)) def IsCloudSubdirPlaceholder(url, blr=None):
  """Determines if a StorageUrl is a cloud subdir placeholder.

  This function is needed because GUI tools (like the GCS cloud console) allow
  users to create empty "folders" by creating a placeholder object; and parts
  of gsutil need to treat those placeholder objects specially. For example,
  gsutil rsync needs to avoid downloading those objects because they can cause
  conflicts (see comments in rsync command for details).

  We currently detect two cases:
    - Cloud objects whose name ends with '_$folder$'
    - Cloud objects whose name ends with '/'

  Args:
    url: (gslib.storage_url.StorageUrl) The URL to be checked.
    blr: (gslib.BucketListingRef or None) The blr to check, or None if not
        available. If `blr` is None, size won't be checked.

  Returns:
    (bool) True if the URL is a cloud subdir placeholder, otherwise False.
  """
  if not url.IsCloudUrl():
    return False
  url_str = url.url_string
  if url_str.endswith('_$folder$'):
    return True
  if blr and blr.IsObject():
    size = blr.root_object.size
  else:
    size = 0
  return size == 0 and url_str.endswith('/') def StorageUrlFromString(url_str):
  """Static factory function for creating a StorageUrl from a string."""

  scheme = GetSchemeFromUrlString(url_str)

  if not IsKnownUrlScheme(scheme):
    raise InvalidUrlError('Unrecognized scheme "%s"' % scheme)
  if scheme == 'file':
    path = _GetPathFromUrlString(url_str)
    is_stream = (path == '-')
    is_fifo = False
    try:
      is_fifo = stat.S_ISFIFO(os.stat(path).st_mode)
    except OSError:
      pass
    return _FileUrl(url_str, is_stream=is_stream, is_fifo=is_fifo)
  return _CloudUrl(url_str) class Bucket(_messages.Message):
  class AutoclassValue(_messages.Message):
  class BillingValue(_messages.Message):
  class CorsValueListEntry(_messages.Message):
  class CustomPlacementConfigValue(_messages.Message):
  class EncryptionValue(_messages.Message):
  class IamConfigurationValue(_messages.Message):
    class BucketPolicyOnlyValue(_messages.Message):
  class LabelsValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
  class LifecycleValue(_messages.Message):
    class RuleValueListEntry(_messages.Message):
      class ActionValue(_messages.Message):
      class ConditionValue(_messages.Message):
  class LoggingValue(_messages.Message):
  class OwnerValue(_messages.Message):
  class RetentionPolicyValue(_messages.Message):
  class VersioningValue(_messages.Message):
  class WebsiteValue(_messages.Message):
class BucketAccessControl(_messages.Message):
  class ProjectTeamValue(_messages.Message):
class BucketAccessControls(_messages.Message):
class Buckets(_messages.Message):
class Channel(_messages.Message):
  class ParamsValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
class Channels(_messages.Message):
  class ItemsValueListEntry(_messages.Message):
class ComposeRequest(_messages.Message):
  class SourceObjectsValueListEntry(_messages.Message):
    class ObjectPreconditionsValue(_messages.Message):
class Expr(_messages.Message):
class Notification(_messages.Message):
  class CustomAttributesValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
class Notifications(_messages.Message):
class Object(_messages.Message):
  class CustomerEncryptionValue(_messages.Message):
  class MetadataValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
  class OwnerValue(_messages.Message):
class ObjectAccessControl(_messages.Message):
  class ProjectTeamValue(_messages.Message):
class ObjectAccessControls(_messages.Message):
class Objects(_messages.Message):
class Policy(_messages.Message):
  class BindingsValueListEntry(_messages.Message):
class RewriteResponse(_messages.Message):
class ServiceAccount(_messages.Message):
class StandardQueryParameters(_messages.Message):
  class AltValueValuesEnum(_messages.Enum):
class StorageBucketAccessControlsDeleteRequest(_messages.Message):
class StorageBucketAccessControlsDeleteResponse(_messages.Message):
class StorageBucketAccessControlsGetRequest(_messages.Message):
class StorageBucketAccessControlsInsertRequest(_messages.Message):
class StorageBucketAccessControlsListRequest(_messages.Message):
class StorageBucketAccessControlsPatchRequest(_messages.Message):
class StorageBucketAccessControlsUpdateRequest(_messages.Message):
class StorageBucketsDeleteRequest(_messages.Message):
class StorageBucketsDeleteResponse(_messages.Message):
class StorageBucketsGetIamPolicyRequest(_messages.Message):
class StorageBucketsGetRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsInsertRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class PredefinedDefaultObjectAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsListRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsListChannelsRequest(_messages.Message):
class StorageBucketsLockRetentionPolicyRequest(_messages.Message):
class StorageBucketsPatchRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class PredefinedDefaultObjectAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsSetIamPolicyRequest(_messages.Message):
class StorageBucketsTestIamPermissionsRequest(_messages.Message):
class StorageBucketsUpdateRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class PredefinedDefaultObjectAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageChannelsStopResponse(_messages.Message):
class StorageDefaultObjectAccessControlsDeleteRequest(_messages.Message):
class StorageDefaultObjectAccessControlsDeleteResponse(_messages.Message):
class StorageDefaultObjectAccessControlsGetRequest(_messages.Message):
class StorageDefaultObjectAccessControlsInsertRequest(_messages.Message):
class StorageDefaultObjectAccessControlsListRequest(_messages.Message):
class StorageDefaultObjectAccessControlsPatchRequest(_messages.Message):
class StorageDefaultObjectAccessControlsUpdateRequest(_messages.Message):
class StorageNotificationsDeleteRequest(_messages.Message):
class StorageNotificationsDeleteResponse(_messages.Message):
class StorageNotificationsGetRequest(_messages.Message):
class StorageNotificationsInsertRequest(_messages.Message):
class StorageNotificationsListRequest(_messages.Message):
class StorageObjectAccessControlsDeleteRequest(_messages.Message):
class StorageObjectAccessControlsDeleteResponse(_messages.Message):
class StorageObjectAccessControlsGetRequest(_messages.Message):
class StorageObjectAccessControlsInsertRequest(_messages.Message):
class StorageObjectAccessControlsListRequest(_messages.Message):
class StorageObjectAccessControlsPatchRequest(_messages.Message):
class StorageObjectAccessControlsUpdateRequest(_messages.Message):
class StorageObjectsComposeRequest(_messages.Message):
  class DestinationPredefinedAclValueValuesEnum(_messages.Enum):
class StorageObjectsCopyRequest(_messages.Message):
  class DestinationPredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsDeleteRequest(_messages.Message):
class StorageObjectsDeleteResponse(_messages.Message):
class StorageObjectsGetIamPolicyRequest(_messages.Message):
class StorageObjectsGetRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsInsertRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsListRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsPatchRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsRewriteRequest(_messages.Message):
  class DestinationPredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsSetIamPolicyRequest(_messages.Message):
class StorageObjectsTestIamPermissionsRequest(_messages.Message):
class StorageObjectsUpdateRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsWatchAllRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageProjectsServiceAccountGetRequest(_messages.Message):
class TestIamPermissionsResponse(_messages.Message):
class StorageProjectsHmacKeysCreateRequest(_messages.Message):
class StorageProjectsHmacKeysDeleteRequest(_messages.Message):
class StorageProjectsHmacKeysGetRequest(_messages.Message):
class StorageProjectsHmacKeysListRequest(_messages.Message):
class StorageProjectsHmacKeysUpdateRequest(_messages.Message):
class HmacKeyMetadata(_messages.Message):
class HmacKey(_messages.Message):
class HmacKeyMetadataList(_messages.Message):
class HmacKeysDeleteResponse(_messages.Message): _RENAME_ON_HASH_MISMATCH = False
_RENAME_ON_HASH_MISMATCH_SUFFIX = '_corrupt'
PARALLEL_UPLOAD_TEMP_NAMESPACE = (
    '/gsutil/tmp/parallel_composite_uploads/for_details_see/gsutil_help_cp/')
PARALLEL_UPLOAD_STATIC_SALT = u"""
PARALLEL_UPLOAD_SALT_TO_PREVENT_COLLISIONS.
The theory is that no user will have prepended this to the front of
one of their object names and then done an MD5 hash of the name, and
then prepended PARALLEL_UPLOAD_TEMP_NAMESPACE to the front of their object
name. Note that there will be no problems with object name length since we
hash the original name.
"""
UPLOAD_RETURN_FIELDS = [
    'crc32c',
    'customerEncryption',
    'etag',
    'generation',
    'md5Hash',
    'size',
]
GZIP_CHUNK_SIZE = 8192
GZIP_ALL_FILES = 'GZIP_ALL_FILES'
TRACKERFILE_UPDATE_THRESHOLD = TEN_MIB
PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD = 150 * 1024 * 1024
S3_MAX_UPLOAD_SIZE = 5 * 1024 * 1024 * 1024
COMMON_EXTENSION_RULES = {
    'md': 'text/markdown',
    'tgz': 'application/gzip',
}
EARLY_DELETION_MINIMUM_LIFETIME = {
    'nearline': 30 * SECONDS_PER_DAY,
    'coldline': 90 * SECONDS_PER_DAY,
    'archive': 365 * SECONDS_PER_DAY
}
class FileConcurrencySkipError(Exception):
class SlicedDownloadFileWrapper(object):
class Manifest(object):
class ItemExistsError(Exception):
class SkipUnsupportedObjectError(Exception):
class SkipGlacierError(SkipUnsupportedObjectError):
def _RmExceptionHandler(cls, e):
def _ParallelCopyExceptionHandler(cls, e):
def _PerformParallelUploadFileToObject(cls, args, thread_state=None):
def CreateCopyHelperOpts(perform_mv=False,
                         no_clobber=False,
                         daisy_chain=False,
                         read_args_from_stdin=False,
                         print_ver=False,
                         use_manifest=False,
                         preserve_acl=False,
                         canned_acl=None,
                         skip_unsupported_objects=False,
                         test_callback_file=None,
                         dest_storage_class=None):
def GetCopyHelperOpts():
def _SelectDownloadStrategy(dst_url):
def InsistDstUrlNamesContainer(exp_dst_url, have_existing_dst_container,
                               command_name):
def _ShouldTreatDstUrlAsBucketSubDir(have_multiple_srcs, dst_url,
                                     have_existing_dest_subdir,
                                     src_url_names_container,
                                     recursion_requested):
def _ShouldTreatDstUrlAsSingleton(src_url_names_container, have_multiple_srcs,
                                  have_existing_dest_subdir, dst_url,
                                  recursion_requested):
def ConstructDstUrl(src_url,
                    exp_src_url,
                    src_url_names_container,
                    have_multiple_srcs,
                    exp_dst_url,
                    have_existing_dest_subdir,
                    recursion_requested,
                    preserve_posix=False):
def _CreateDigestsFromDigesters(digesters):
def _CreateDigestsFromLocalFile(status_queue, algs, file_name, src_url,
                                src_obj_metadata):
def _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
                      dst_obj_metadata):
def _CheckHashes(logger,
                 obj_url,
                 obj_metadata,
                 file_name,
                 digests,
                 is_upload=False):
def IsNoClobberServerException(e):
def CheckForDirFileConflict(exp_src_url, dst_url):
def _PartitionFile(fp,
                   file_size,
                   src_url,
                   content_type,
                   canned_acl,
                   dst_bucket_url,
                   random_prefix,
                   tracker_file,
                   tracker_file_lock,
                   encryption_key_sha256=None,
                   gzip_encoded=False):
def _GetComponentNumber(component):
def _DoParallelCompositeUpload(fp,
                               src_url,
                               dst_url,
                               dst_obj_metadata,
                               canned_acl,
                               file_size,
                               preconditions,
                               gsutil_api,
                               command_obj,
                               copy_exception_handler,
                               logger,
                               gzip_encoded=False):
def _ShouldDoParallelCompositeUpload(logger,
                                     allow_splitting,
                                     src_url,
                                     dst_url,
                                     file_size,
                                     gsutil_api,
                                     canned_acl=None):
def ExpandUrlToSingleBlr(url_str,
                         gsutil_api,
                         project_id,
                         treat_nonexistent_object_as_subdir=False,
                         logger=None):
def FixWindowsNaming(src_url, dst_url):
def SrcDstSame(src_url, dst_url):
def _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata):
def _CopyObjToObjInTheCloud(src_url,
                            src_obj_metadata,
                            dst_url,
                            dst_obj_metadata,
                            preconditions,
                            gsutil_api,
                            decryption_key=None):
def _SetContentTypeFromFile(src_url, dst_obj_metadata):
def _UploadFileToObjectNonResumable(src_url,
                                    src_obj_filestream,
                                    src_obj_size,
                                    dst_url,
                                    dst_obj_metadata,
                                    preconditions,
                                    gsutil_api,
                                    gzip_encoded=False):
def _UploadFileToObjectResumable(src_url,
                                 src_obj_filestream,
                                 src_obj_size,
                                 dst_url,
                                 dst_obj_metadata,
                                 preconditions,
                                 gsutil_api,
                                 logger,
                                 is_component=False,
                                 gzip_encoded=False):
  def _UploadTrackerCallback(serialization_data):
def _SelectUploadCompressionStrategy(object_name,
                                     is_component=False,
                                     gzip_exts=False,
                                     gzip_encoded=False):
def _ApplyZippedUploadCompression(src_url, src_obj_filestream, src_obj_size,
                                  logger):
def _DelegateUploadFileToObject(upload_delegate, upload_url, upload_stream,
                                zipped_file, gzip_encoded_file,
                                parallel_composite_upload, logger):
def _UploadFileToObject(src_url,
                        src_obj_filestream,
                        src_obj_size,
                        dst_url,
                        dst_obj_metadata,
                        preconditions,
                        gsutil_api,
                        logger,
                        command_obj,
                        copy_exception_handler,
                        gzip_exts=None,
                        allow_splitting=True,
                        is_component=False,
                        gzip_encoded=False):
  def CallParallelCompositeUpload():
  def CallNonResumableUpload():
  def CallResumableUpload():
def _GetDownloadFile(dst_url, src_obj_metadata, logger):
def _ShouldDoSlicedDownload(download_strategy, src_obj_metadata,
                            allow_splitting, logger):
def _PerformSlicedDownloadObjectToFile(cls, args, thread_state=None):
def _MaintainSlicedDownloadTrackerFiles(src_obj_metadata, dst_url,
                                        download_file_name, logger,
                                        api_selector, num_components):
  def __init__(self, fp, tracker_file_name, src_obj_metadata, start_byte,
               end_byte):
  def mode(self):
  def write(self, data):  # pylint: disable=invalid-name
  def seek(self, offset, whence=os.SEEK_SET):  # pylint: disable=invalid-name
  def tell(self):  # pylint: disable=invalid-name
  def flush(self):  # pylint: disable=invalid-name
  def close(self):  # pylint: disable=invalid-name
def _PartitionObject(src_url,
                     src_obj_metadata,
                     dst_url,
                     download_file_name,
                     decryption_key=None):
def _DoSlicedDownload(src_url,
                      src_obj_metadata,
                      dst_url,
                      download_file_name,
                      command_obj,
                      logger,
                      copy_exception_handler,
                      api_selector,
                      decryption_key=None,
                      status_queue=None):
def _DownloadObjectToFileResumable(src_url,
                                   src_obj_metadata,
                                   dst_url,
                                   download_file_name,
                                   gsutil_api,
                                   logger,
                                   digesters,
                                   component_num=None,
                                   start_byte=0,
                                   end_byte=None,
                                   decryption_key=None):
def _DownloadObjectToFileNonResumable(src_url,
                                      src_obj_metadata,
                                      dst_url,
                                      download_file_name,
                                      gsutil_api,
                                      digesters,
                                      decryption_key=None):
def _DownloadObjectToFile(src_url,
                          src_obj_metadata,
                          dst_url,
                          gsutil_api,
                          logger,
                          command_obj,
                          copy_exception_handler,
                          allow_splitting=True,
                          decryption_key=None,
                          is_rsync=False,
                          preserve_posix=False,
                          use_stet=False):
def _ValidateAndCompleteDownload(logger,
                                 src_url,
                                 src_obj_metadata,
                                 dst_url,
                                 need_to_unzip,
                                 server_gzip,
                                 digesters,
                                 hash_algs,
                                 temporary_file_name,
                                 api_selector,
                                 bytes_transferred,
                                 gsutil_api,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 use_stet=False):
def _CopyFileToFile(src_url, dst_url, status_queue=None, src_obj_metadata=None):
def _DummyTrackerCallback(_):
def _CopyObjToObjDaisyChainMode(src_url,
                                src_obj_metadata,
                                dst_url,
                                dst_obj_metadata,
                                preconditions,
                                gsutil_api,
                                logger,
                                decryption_key=None):
def GetSourceFieldsNeededForCopy(dst_is_cloud,
                                 skip_unsupported_objects,
                                 preserve_acl,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 delete_source=False,
                                 file_size_will_change=False):
def WarnIfMvEarlyDeletionChargeApplies(src_url, src_obj_metadata, logger):
def MaybeSkipUnsupportedObject(src_url, src_obj_metadata):
def GetDecryptionCSEK(src_url, src_obj_metadata):
def PerformCopy(logger,
                src_url,
                dst_url,
                gsutil_api,
                command_obj,
                copy_exception_handler,
                src_obj_metadata=None,
                allow_splitting=True,
                headers=None,
                manifest=None,
                gzip_exts=None,
                is_rsync=False,
                preserve_posix=False,
                gzip_encoded=False,
                use_stet=False):
  def __init__(self, path):
  def _ParseManifest(self):
  def WasSuccessful(self, src):
  def _CreateManifestFile(self):
  def Set(self, url, key, value):
  def Initialize(self, source_url, destination_url):
  def SetResult(self, source_url, bytes_transferred, result, description=''):
  def _WriteRowToManifestFile(self, url):
  def _RemoveItemFromManifest(self, url):
  def __init__(self):
  def __init__(self):
def GetPathBeforeFinalDir(url, exp_src_url):
def ResolveWildcardsInPathBeforeFinalDir(src_url_path_sans_final_dir,
                                         exp_src_url):
def _GetPartitionInfo(file_size, max_components, default_component_size):
def _DeleteTempComponentObjectFn(cls, url_to_delete, thread_state=None):
def FilterExistingComponents(dst_args, existing_components, bucket_url,
                             gsutil_api): def CreateCopyHelperOpts(perform_mv=False,
                         no_clobber=False,
                         daisy_chain=False,
                         read_args_from_stdin=False,
                         print_ver=False,
                         use_manifest=False,
                         preserve_acl=False,
                         canned_acl=None,
                         skip_unsupported_objects=False,
                         test_callback_file=None,
                         dest_storage_class=None):
  """Creates CopyHelperOpts for passing options to CopyHelper."""
  # We create a tuple with union of options needed by CopyHelper and any
  # copy-related functionality in CpCommand, RsyncCommand, or Command class.
  global global_copy_helper_opts
  global_copy_helper_opts = CopyHelperOpts(
      perform_mv=perform_mv,
      no_clobber=no_clobber,
      daisy_chain=daisy_chain,
      read_args_from_stdin=read_args_from_stdin,
      print_ver=print_ver,
      use_manifest=use_manifest,
      preserve_acl=preserve_acl,
      canned_acl=canned_acl,
      skip_unsupported_objects=skip_unsupported_objects,
      test_callback_file=test_callback_file,
      dest_storage_class=dest_storage_class)
  return global_copy_helper_opts def GetSourceFieldsNeededForCopy(dst_is_cloud,
                                 skip_unsupported_objects,
                                 preserve_acl,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 delete_source=False,
                                 file_size_will_change=False):
  """Determines the metadata fields needed for a copy operation.

  This function returns the fields we will need to successfully copy any
  cloud objects that might be iterated. By determining this prior to iteration,
  the cp command can request this metadata directly from the iterator's
  get/list calls, avoiding the need for a separate get metadata HTTP call for
  each iterated result. As a trade-off, filtering objects at the leaf nodes of
  the iteration (based on a remaining wildcard) is more expensive. This is
  because more metadata will be requested when object name is all that is
  required for filtering.

  The rsync command favors fast listing and comparison, and makes the opposite
  trade-off, optimizing for the low-delta case by making per-object get
  metadata HTTP call so that listing can return minimal metadata. It uses
  this function to determine what is needed for get metadata HTTP calls.

  Args:
    dst_is_cloud: if true, destination is a Cloud URL.
    skip_unsupported_objects: if true, get metadata for skipping unsupported
        object types.
    preserve_acl: if true, get object ACL.
    is_rsync: if true, the calling function is rsync. Determines if metadata is
              needed to verify download.
    preserve_posix: if true, retrieves POSIX attributes into user metadata.
    delete_source: if true, source object will be deleted after the copy
                   (mv command).
    file_size_will_change: if true, do not try to record file size.

  Returns:
    List of necessary field metadata field names.

  """
  src_obj_fields_set = set()

  if dst_is_cloud:
    # For cloud or daisy chain copy, we need every copyable field.
    # If we're not modifying or overriding any of the fields, we can get
    # away without retrieving the object metadata because the copy
    # operation can succeed with just the destination bucket and object
    # name. But if we are sending any metadata, the JSON API will expect a
    # complete object resource. Since we want metadata like the object size for
    # our own tracking, we just get all of the metadata here.
    src_obj_fields_set.update([
        'cacheControl',
        'componentCount',
        'contentDisposition',
        'contentEncoding',
        'contentLanguage',
        'contentType',
        'crc32c',
        'customerEncryption',
        'etag',
        'generation',
        'md5Hash',
        'mediaLink',
        'metadata',
        'metageneration',
        'storageClass',
        'timeCreated',
    ])
    # We only need the ACL if we're going to preserve it.
    if preserve_acl:
      src_obj_fields_set.update(['acl'])
    if not file_size_will_change:
      src_obj_fields_set.update(['size'])

  else:
    # Just get the fields needed to perform and validate the download.
    src_obj_fields_set.update([
        'crc32c',
        'contentEncoding',
        'contentType',
        'customerEncryption',
        'etag',
        'mediaLink',
        'md5Hash',
        'size',
        'generation',
    ])
    if is_rsync:
      src_obj_fields_set.update(['metadata/%s' % MTIME_ATTR, 'timeCreated'])
    if preserve_posix:
      posix_fields = [
          'metadata/%s' % ATIME_ATTR,
          'metadata/%s' % MTIME_ATTR,
          'metadata/%s' % GID_ATTR,
          'metadata/%s' % MODE_ATTR,
          'metadata/%s' % UID_ATTR,
      ]
      src_obj_fields_set.update(posix_fields)

  if delete_source:
    src_obj_fields_set.update([
        'storageClass',
        'timeCreated',
    ])

  if skip_unsupported_objects:
    src_obj_fields_set.update(['storageClass'])

  return list(src_obj_fields_set) GZIP_ALL_FILES = 'GZIP_ALL_FILES' class ItemExistsError(Exception):
  """Exception class for objects that are skipped because they already exist."""
  pass class Manifest(object):
  """Stores the manifest items for the CpCommand class."""

  def __init__(self, path):
    # self.items contains a dictionary of rows
    self.items = {}
    self.manifest_filter = {}
    self.lock = parallelism_framework_util.CreateLock()

    self.manifest_path = os.path.expanduser(path)
    self._ParseManifest()
    self._CreateManifestFile()

  def _ParseManifest(self):
    """Load and parse a manifest file.

    This information will be used to skip any files that have a skip or OK
    status.
    """
    try:
      if os.path.exists(self.manifest_path):
        # Note: we can't use io.open here or CSV reader will become upset
        # https://stackoverflow.com/a/18449496
        with open(self.manifest_path, 'r') as f:
          first_row = True
          reader = csv.reader(f)
          for row in reader:
            if first_row:
              try:
                source_index = row.index('Source')
                result_index = row.index('Result')
              except ValueError:
                # No header and thus not a valid manifest file.
                raise CommandException('Missing headers in manifest file: %s' %
                                       self.manifest_path)
            first_row = False
            source = row[source_index]
            result = row[result_index]
            if result in ['OK', 'skip']:
              # We're always guaranteed to take the last result of a specific
              # source url.
              self.manifest_filter[source] = result
    except IOError:
      raise CommandException('Could not parse %s' % self.manifest_path)

  def WasSuccessful(self, src):
    """Returns whether the specified src url was marked as successful."""
    return src in self.manifest_filter

  def _CreateManifestFile(self):
    """Opens the manifest file and assigns it to the file pointer."""
    try:
      if ((not os.path.exists(self.manifest_path)) or
          (os.stat(self.manifest_path).st_size == 0)):
        # Add headers to the new file.
        if six.PY3:
          with open(self.manifest_path, 'w', newline='') as f:
            writer = csv.writer(f)
            writer.writerow([
                'Source',
                'Destination',
                'Start',
                'End',
                'Md5',
                'UploadId',
                'Source Size',
                'Bytes Transferred',
                'Result',
                'Description',
            ])
        else:
          with open(self.manifest_path, 'wb', 1) as f:
            writer = csv.writer(f)
            writer.writerow([
                'Source',
                'Destination',
                'Start',
                'End',
                'Md5',
                'UploadId',
                'Source Size',
                'Bytes Transferred',
                'Result',
                'Description',
            ])
    except IOError:
      raise CommandException('Could not create manifest file.')

  def Set(self, url, key, value):
    if value is None:
      # In case we don't have any information to set we bail out here.
      # This is so that we don't clobber existing information.
      # To zero information pass '' instead of None.
      return
    if url in self.items:
      self.items[url][key] = value
    else:
      self.items[url] = {key: value}

  def Initialize(self, source_url, destination_url):
    # Always use the source_url as the key for the item. This is unique.
    self.Set(source_url, 'source_uri', source_url)
    self.Set(source_url, 'destination_uri', destination_url)
    self.Set(source_url, 'start_time', datetime.datetime.utcnow())

  def SetResult(self, source_url, bytes_transferred, result, description=''):
    self.Set(source_url, 'bytes', bytes_transferred)
    self.Set(source_url, 'result', result)
    self.Set(source_url, 'description', description)
    self.Set(source_url, 'end_time', datetime.datetime.utcnow())
    self._WriteRowToManifestFile(source_url)
    self._RemoveItemFromManifest(source_url)

  def _WriteRowToManifestFile(self, url):
    """Writes a manifest entry to the manifest file for the url argument."""
    row_item = self.items[url]
    data = [
        row_item['source_uri'],
        row_item['destination_uri'],
        '%sZ' % row_item['start_time'].isoformat(),
        '%sZ' % row_item['end_time'].isoformat(),
        row_item['md5'] if 'md5' in row_item else '',
        row_item['upload_id'] if 'upload_id' in row_item else '',
        str(row_item['size']) if 'size' in row_item else '',
        str(row_item['bytes']) if 'bytes' in row_item else '',
        row_item['result'],
        row_item['description'],
    ]

    data = [six.ensure_str(value) for value in data]

    # Aquire a lock to prevent multiple threads writing to the same file at
    # the same time. This would cause a garbled mess in the manifest file.
    with self.lock:
      if IS_WINDOWS and six.PY3:
        f = open(self.manifest_path, 'a', 1, newline='')
      else:
        f = open(self.manifest_path, 'a', 1)  # 1 == line buffered
      writer = csv.writer(f)
      writer.writerow(data)
      f.close()

  def _RemoveItemFromManifest(self, url):
    # Remove the item from the dictionary since we're done with it and
    # we don't want the dictionary to grow too large in memory for no good
    # reason.
    del self.items[url] class SkipUnsupportedObjectError(Exception):
  """Exception for objects skipped because they are an unsupported type."""

  def __init__(self):
    super(SkipUnsupportedObjectError, self).__init__()
    self.unsupported_type = 'Unknown' def ConvertModeToBase8(mode):
  """Converts a base-10 mode integer from os.stat to base-8."""
  # Strip out unnecessary bits in the mode. Mode is given as a base-10
  # integer. It must be converted to base-8.
  return int(oct(mode)[-3:]) def DeserializeFileAttributesFromObjectMetadata(obj_metadata, url_str):
  """Parses the POSIX attributes from the supplied metadata.

  Args:
    obj_metadata: The metadata for an object.
    url_str: File/object path that provides context if a warning is thrown.

  Returns:
    A POSIXAttribute object with the retrieved values or a default value for
    any attribute that could not be found.
  """
  posix_attrs = POSIXAttributes()
  # Parse atime.
  found, atime = GetValueFromObjectCustomMetadata(obj_metadata, ATIME_ATTR,
                                                  NA_TIME)
  try:
    atime = long(atime)
    if found and atime <= NA_TIME:
      WarnNegativeAttribute('atime', url_str)
      atime = NA_TIME
    elif atime > long(time.time()) + SECONDS_PER_DAY:
      WarnFutureTimestamp('atime', url_str)
      atime = NA_TIME
  except ValueError:
    WarnInvalidValue('atime', url_str)
    atime = NA_TIME
  posix_attrs.atime = atime
  # Parse gid.
  DeserializeIDAttribute(obj_metadata, GID_ATTR, url_str, posix_attrs)
  # Parse uid.
  DeserializeIDAttribute(obj_metadata, UID_ATTR, url_str, posix_attrs)
  found, mode = GetValueFromObjectCustomMetadata(obj_metadata, MODE_ATTR,
                                                 NA_MODE)
  if found and MODE_REGEX.match(mode):
    try:
      # Parse mode into a 3-digit base-8 number.
      posix_attrs.mode = POSIXMode(int(mode))
    except ValueError:
      WarnInvalidValue('mode', url_str)
  return posix_attrs def InitializeUserGroups():
  """Initializes the set of groups that the user is in.

  Should only be called if the flag for preserving POSIX attributes is set.
  """
  global USER_GROUPS
  if IS_WINDOWS:
    return
  user_id = os.getuid()
  user_name = pwd.getpwuid(user_id).pw_name
  USER_GROUPS = set(
      # Primary group
      [pwd.getpwuid(user_id).pw_gid] +
      # Secondary groups
      [g.gr_gid for g in grp.getgrall() if user_name in g.gr_mem]) class POSIXAttributes(object):
  """Class to hold POSIX attributes for a file/object."""

  def __init__(self,
               atime=NA_TIME,
               mtime=NA_TIME,
               uid=NA_ID,
               gid=NA_ID,
               mode=None):
    """Constructor for POSIXAttributes class which holds relevant data.

    Args:
      atime: The access time of the file/object.
      mtime: The modification time of the file/object.
      uid: The user ID that owns the file.
      gid: The group ID that the user is in.
      mode: An instance of POSIXMode.
    """
    self.atime = atime
    self.mtime = mtime
    self.uid = uid
    self.gid = gid
    self.mode = POSIXMode(mode if mode else NA_MODE) def SerializeFileAttributesToObjectMetadata(posix_attrs,
                                            custom_metadata,
                                            preserve_posix=False):
  """Takes a POSIXAttributes object and serializes it into custom metadata.

  Args:
    posix_attrs: A POSIXAttributes object.
    custom_metadata: A custom metadata object to serialize values into.
    preserve_posix: Whether or not to preserve POSIX attributes other than
                    mtime.
  """
  # mtime will always be needed in the object metadata for rsync.
  if posix_attrs.mtime != NA_TIME:
    CreateCustomMetadata(entries={MTIME_ATTR: posix_attrs.mtime},
                         custom_metadata=custom_metadata)
  # Only add other POSIX attributes if the preserve_posix flag is set.
  if preserve_posix:
    if posix_attrs.atime != NA_TIME:
      CreateCustomMetadata(entries={ATIME_ATTR: posix_attrs.atime},
                           custom_metadata=custom_metadata)
    if posix_attrs.uid != NA_ID:
      CreateCustomMetadata(entries={UID_ATTR: posix_attrs.uid},
                           custom_metadata=custom_metadata)
    if posix_attrs.gid != NA_ID:
      CreateCustomMetadata(entries={GID_ATTR: posix_attrs.gid},
                           custom_metadata=custom_metadata)
    if posix_attrs.mode.permissions != NA_MODE:
      CreateCustomMetadata(entries={MODE_ATTR: posix_attrs.mode.permissions},
                           custom_metadata=custom_metadata) def ValidateFilePermissionAccess(url_str, uid=NA_ID, gid=NA_ID, mode=NA_MODE):
  """Validates that the user has file access if uid, gid, and mode are applied.

  Args:
    url_str: The path to the object for which this is validating.
    uid: A POSIX user ID.
    gid: A POSIX group ID.
    mode: A 3-digit, number representing POSIX permissions, must be in base-8.

  Returns:
    A (bool, str) tuple, True if and only if it's safe to copy the file, and a
    string containing details for the error.
  """
  # Windows doesn't use the POSIX system for file permissions, so all files will
  # validate.
  if IS_WINDOWS:
    return True, ''

  uid_present = uid > NA_ID
  gid_present = int(gid) > NA_ID
  mode_present = mode > NA_MODE
  # No need to perform validation if Posix attrs are not being preserved.
  if not (uid_present or gid_present or mode_present):
    return True, ''

  # The root user on non-Windows systems can access files regardless of their
  # permissions.
  if os.geteuid() == 0:
    return True, ''

  mode_valid = ValidatePOSIXMode(int(str(mode), 8))
  if mode_present:
    if not mode_valid:
      return False, 'Mode for %s won\'t allow read access.' % url_str
  else:
    # Calculate the default mode if the mode doesn't exist.
    # Convert mode to a 3-digit, base-8 integer.
    mode = int(GetDefaultMode())

  if uid_present:
    try:
      pwd.getpwuid(uid)
    except (KeyError, OverflowError):
      return (False, 'UID for %s doesn\'t exist on current system. uid: %d' %
              (url_str, uid))
  if gid_present:
    try:
      grp.getgrgid(gid)
    except (KeyError, OverflowError):
      return (False, 'GID for %s doesn\'t exist on current system. gid: %d' %
              (url_str, gid))

  # uid at this point must exist, but isn't necessarily the current user.
  # Likewise, gid must also exist at this point.
  uid_is_current_user = uid == os.getuid()

  # By this point uid and gid must exist on the system. However, the uid might
  # not match the current user's or the current user might not be a member of
  # the group identified by gid. In this case, the 'other' byte of the
  # permissions could provide sufficient access.
  mode = int(str(mode), 8)
  # Check that if the uid is not present and the gid and mode are, so that we
  # won't orphan the file. For example if the mode is set to 007, we can orphan
  # the file because the uid would default to the current user's ID and if the
  # current user wouldn't have read access or better, the file will be orphaned
  # even though they might otherwise have access through the gid or other bytes.
  if not uid_present and gid_present and mode_present and not bool(mode & U_R):
    return (False, 'Insufficient access with uid/gid/mode for %s, gid: %d, '
            'mode: %s' % (url_str, gid, oct(mode)[-3:]))
  if uid_is_current_user:
    valid = bool(mode & U_R)
    return (valid, '' if valid else
            'Insufficient access with uid/gid/mode for %s, uid: %d, '
            'mode: %s' % (url_str, uid, oct(mode)[-3:]))
  elif int(gid) in USER_GROUPS:
    valid = bool(mode & G_R)
    return (valid, '' if valid else
            'Insufficient access with uid/gid/mode for %s, gid: %d, '
            'mode: %s' % (url_str, gid, oct(mode)[-3:]))
  elif mode & O_R:
    return True, ''
  elif not uid_present and not gid_present and mode_valid:
    return True, ''
  return False, 'There was a problem validating %s.' % url_str class GcloudStorageFlag(object):

  def __init__(self, gcloud_flag, supports_output_translation=False):
    """Initializes GcloudStorageFlag.
    
    Args:
      gcloud_flag (str): The name of the gcloud flag.
      support_output_translation (bool): If True, this flag in gcloud storage
        supports printing gsutil formatted output.
    """
    self.gcloud_flag = gcloud_flag
    self.supports_output_translation = supports_output_translation class GcloudStorageMap(object):
  """Mapping to translate gsutil command to its gcloud storage equivalent."""

  def __init__(self,
               gcloud_command,
               flag_map,
               supports_output_translation=False):
    """Intalizes GcloudStorageMap.
    
    Args:
      gcloud_command (dict|str): The corresponding name of the command to be
        called in gcloud. If this command supports sub-commands, then this 
        field must be a dict of sub-command-name:GcloudStorageMap pairs.
      flag_map (dict): A dict of str to GcloudStorageFlag. Mapping of gsutil
        flags to their equivalent gcloud storage flag names.
      supports_output_translation (bool): Indicates if the corresponding
        gcloud storage command supports the printing gsutil formatted output.
    """
    self.gcloud_command = gcloud_command
    self.flag_map = flag_map
    self.supports_output_translation = supports_output_translation
import errno
import itertools
import logging
import os
import time
import traceback
from apitools.base.py import encoding
from gslib.command import Command
from gslib.command_argument import CommandArgument
from gslib.cs_api_map import ApiSelector
from gslib.exception import CommandException
from gslib.metrics import LogPerformanceSummaryParams
from gslib.name_expansion import CopyObjectsIterator
from gslib.name_expansion import DestinationInfo
from gslib.name_expansion import NameExpansionIterator
from gslib.name_expansion import NameExpansionIteratorDestinationTuple
from gslib.name_expansion import SeekAheadNameExpansionIterator
from gslib.storage_url import ContainsWildcard
from gslib.storage_url import IsCloudSubdirPlaceholder
from gslib.storage_url import StorageUrlFromString
from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
from gslib.utils import cat_helper
from gslib.utils import copy_helper
from gslib.utils import parallelism_framework_util
from gslib.utils.cloud_api_helper import GetCloudApiInstance
from gslib.utils.constants import DEBUGLEVEL_DUMP_REQUESTS
from gslib.utils.constants import NO_MAX
from gslib.utils.copy_helper import CreateCopyHelperOpts
from gslib.utils.copy_helper import GetSourceFieldsNeededForCopy
from gslib.utils.copy_helper import GZIP_ALL_FILES
from gslib.utils.copy_helper import ItemExistsError
from gslib.utils.copy_helper import Manifest
from gslib.utils.copy_helper import SkipUnsupportedObjectError
from gslib.utils.posix_util import ConvertModeToBase8
from gslib.utils.posix_util import DeserializeFileAttributesFromObjectMetadata
from gslib.utils.posix_util import InitializeUserGroups
from gslib.utils.posix_util import POSIXAttributes
from gslib.utils.posix_util import SerializeFileAttributesToObjectMetadata
from gslib.utils.posix_util import ValidateFilePermissionAccess
from gslib.utils.shim_util import GcloudStorageFlag
from gslib.utils.shim_util import GcloudStorageMap
from gslib.utils.system_util import GetStreamFromFileUrl
from gslib.utils.system_util import StdinIterator
from gslib.utils.system_util import StdinIteratorCls
from gslib.utils.text_util import NormalizeStorageClass
from gslib.utils.text_util import RemoveCRLFFromString
from gslib.utils.unit_util import CalculateThroughput
from gslib.utils.unit_util import MakeHumanReadable
CP_SUB_ARGS = 'a:AcDeIL:MNnpPrRs:tUvz:Zj:J'


def _CopyFuncWrapper(cls, args, thread_state=None):
  cls.CopyFunc(args,
               thread_state=thread_state,
               preserve_posix=cls.preserve_posix_attrs)


def _CopyExceptionHandler(cls, e):
  """Simple exception handler to allow post-completion status."""
  cls.logger.error(str(e))
  cls.op_failure_count += 1
  cls.logger.debug('\n\nEncountered exception while copying:\n%s\n',
                   traceback.format_exc())


def _RmExceptionHandler(cls, e):
  """Simple exception handler to allow post-completion status."""
  cls.logger.error(str(e))


class CpCommand(Command):
  """Implementation of gsutil cp command.

  Note that CpCommand is run for both gsutil cp and gsutil mv. The latter
  happens by MvCommand calling CpCommand and passing the hidden (undocumented)
  -M option. This allows the copy and remove needed for each mv to run
  together (rather than first running all the cp's and then all the rm's, as
  we originally had implemented), which in turn avoids the following problem
  with removing the wrong objects: starting with a bucket containing only
  the object gs://bucket/obj, say the user does:
    gsutil mv gs://bucket/* gs://bucket/d.txt
  If we ran all the cp's and then all the rm's and we didn't expand the wildcard
  first, the cp command would first copy gs://bucket/obj to gs://bucket/d.txt,
  and the rm command would then remove that object. In the implementation
  prior to gsutil release 3.12 we avoided this by building a list of objects
  to process and then running the copies and then the removes; but building
  the list up front limits scalability (compared with the current approach
  of processing the bucket listing iterator on the fly).
  """

  # Command specification. See base class for documentation.
  command_spec = Command.CreateCommandSpec(
      'cp',
      command_name_aliases=['copy'],
      usage_synopsis=_SYNOPSIS,
      min_args=1,
      max_args=NO_MAX,
      # -t is deprecated but leave intact for now to avoid breakage.
      supported_sub_args=CP_SUB_ARGS,
      file_url_ok=True,
      provider_url_ok=False,
      urls_start_arg=0,
      gs_api_support=[ApiSelector.XML, ApiSelector.JSON],
      gs_default_api=ApiSelector.JSON,
      # Unfortunately, "private" args are the only way to support non-single
      # character flags.
      supported_private_args=['stet', 'testcallbackfile='],
      argparse_arguments=[
          CommandArgument.MakeZeroOrMoreCloudOrFileURLsArgument(),
      ],
  )
  # Help specification. See help_provider.py for documentation.
  help_spec = Command.HelpSpec(
      help_name='cp',
      help_name_aliases=['copy'],
      help_type='command_help',
      help_one_line_summary='Copy files and objects',
      help_text=_DETAILED_HELP_TEXT,
      subcommand_help_text={},
  )

  # TODO(b/206151615) Add mappings for remaining flags.
  gcloud_storage_map = GcloudStorageMap(
      gcloud_command='alpha storage cp',
      flag_map={
          '-r': GcloudStorageFlag('-r'),
          '-R': GcloudStorageFlag('-r'),
          '-e': GcloudStorageFlag('--ignore-symlinks')
      },
  )

  # pylint: disable=too-many-statements
  def CopyFunc(self, copy_object_info, thread_state=None, preserve_posix=False):
    """Worker function for performing the actual copy (and rm, for mv)."""
    gsutil_api = GetCloudApiInstance(self, thread_state=thread_state)

    copy_helper_opts = copy_helper.GetCopyHelperOpts()
    if copy_helper_opts.perform_mv:
      cmd_name = 'mv'
    else:
      cmd_name = self.command_name
    src_url = copy_object_info.source_storage_url
    exp_src_url = copy_object_info.expanded_storage_url
    src_url_names_container = copy_object_info.names_container
    have_multiple_srcs = copy_object_info.is_multi_source_request

    if src_url.IsCloudUrl() and src_url.IsProvider():
      raise CommandException(
          'The %s command does not allow provider-only source URLs (%s)' %
          (cmd_name, src_url))
    if preserve_posix and src_url.IsFileUrl() and src_url.IsStream():
      raise CommandException('Cannot preserve POSIX attributes with a stream.')
    if self.parallel_operations and src_url.IsFileUrl() and src_url.IsStream():
      raise CommandException(
          'Cannot upload from a stream when using gsutil -m option.')
    if have_multiple_srcs:
      copy_helper.InsistDstUrlNamesContainer(
          copy_object_info.exp_dst_url,
          copy_object_info.have_existing_dst_container, cmd_name)

    # Various GUI tools (like the GCS web console) create placeholder objects
    # ending with '/' when the user creates an empty directory. Normally these
    # tools should delete those placeholders once objects have been written
    # "under" the directory, but sometimes the placeholders are left around. We
    # need to filter them out here, otherwise if the user tries to rsync from
    # GCS to a local directory it will result in a directory/file conflict
    # (e.g., trying to download an object called "mydata/" where the local
    # directory "mydata" exists).

    if IsCloudSubdirPlaceholder(exp_src_url):