RootZanli & NinjaCr3

: / opt / gsutil / gslib / utils /
Filename : copy_helper.py
back
# -*- coding: utf-8 -*-
# Copyright 2011 Google Inc. All Rights Reserved.
# Copyright 2011, Nexenta Systems Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Helper functions for copy functionality."""

from __future__ import absolute_import
from __future__ import print_function
from __future__ import division
from __future__ import unicode_literals

import base64
from collections import namedtuple
import csv
import datetime
import errno
import gzip
import json
import logging
import mimetypes
from operator import attrgetter
import os
import pickle
import pyu2f
import random
import re
import shutil
import six
import stat
import subprocess
import tempfile
import textwrap
import time
import traceback

import six
from six.moves import range
from six.moves import range

from apitools.base.protorpclite import protojson

from boto import config

import crcmod

import gslib
from gslib.cloud_api import AccessDeniedException
from gslib.cloud_api import ArgumentException
from gslib.cloud_api import CloudApi
from gslib.cloud_api import EncryptionException
from gslib.cloud_api import NotFoundException
from gslib.cloud_api import PreconditionException
from gslib.cloud_api import Preconditions
from gslib.cloud_api import ResumableDownloadException
from gslib.cloud_api import ResumableUploadAbortException
from gslib.cloud_api import ResumableUploadException
from gslib.cloud_api import ResumableUploadStartOverException
from gslib.cloud_api import ServiceException
from gslib.commands.compose import MAX_COMPOSE_ARITY
from gslib.commands.config import DEFAULT_PARALLEL_COMPOSITE_UPLOAD_COMPONENT_SIZE
from gslib.commands.config import DEFAULT_PARALLEL_COMPOSITE_UPLOAD_THRESHOLD
from gslib.commands.config import DEFAULT_SLICED_OBJECT_DOWNLOAD_COMPONENT_SIZE
from gslib.commands.config import DEFAULT_SLICED_OBJECT_DOWNLOAD_MAX_COMPONENTS
from gslib.commands.config import DEFAULT_SLICED_OBJECT_DOWNLOAD_THRESHOLD
from gslib.commands.config import DEFAULT_GZIP_COMPRESSION_LEVEL
from gslib.cs_api_map import ApiSelector
from gslib.daisy_chain_wrapper import DaisyChainWrapper
from gslib.exception import CommandException
from gslib.exception import HashMismatchException
from gslib.exception import InvalidUrlError
from gslib.file_part import FilePart
from gslib.parallel_tracker_file import GenerateComponentObjectPrefix
from gslib.parallel_tracker_file import ReadParallelUploadTrackerFile
from gslib.parallel_tracker_file import ValidateParallelCompositeTrackerData
from gslib.parallel_tracker_file import WriteComponentToParallelUploadTrackerFile
from gslib.parallel_tracker_file import WriteParallelUploadTrackerFile
from gslib.progress_callback import FileProgressCallbackHandler
from gslib.progress_callback import ProgressCallbackWithTimeout
from gslib.resumable_streaming_upload import ResumableStreamingJsonUploadWrapper
from gslib import storage_url
from gslib.storage_url import ContainsWildcard
from gslib.storage_url import GenerationFromUrlAndString
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.thread_message import FileMessage
from gslib.thread_message import RetryableErrorMessage
from gslib.tracker_file import DeleteDownloadTrackerFiles
from gslib.tracker_file import DeleteTrackerFile
from gslib.tracker_file import ENCRYPTION_UPLOAD_TRACKER_ENTRY
from gslib.tracker_file import GetDownloadStartByte
from gslib.tracker_file import GetTrackerFilePath
from gslib.tracker_file import GetUploadTrackerData
from gslib.tracker_file import RaiseUnwritableTrackerFileException
from gslib.tracker_file import ReadOrCreateDownloadTrackerFile
from gslib.tracker_file import SERIALIZATION_UPLOAD_TRACKER_ENTRY
from gslib.tracker_file import TrackerFileType
from gslib.tracker_file import WriteDownloadComponentTrackerFile
from gslib.tracker_file import WriteJsonDataToTrackerFile
from gslib.utils import parallelism_framework_util
from gslib.utils import stet_util
from gslib.utils import temporary_file_util
from gslib.utils import text_util
from gslib.utils.boto_util import GetJsonResumableChunkSize
from gslib.utils.boto_util import GetMaxRetryDelay
from gslib.utils.boto_util import GetNumRetries
from gslib.utils.boto_util import ResumableThreshold
from gslib.utils.boto_util import UsingCrcmodExtension
from gslib.utils.cloud_api_helper import GetCloudApiInstance
from gslib.utils.cloud_api_helper import GetDownloadSerializationData
from gslib.utils.constants import DEFAULT_FILE_BUFFER_SIZE
from gslib.utils.constants import MIN_SIZE_COMPUTE_LOGGING
from gslib.utils.constants import UTF8
from gslib.utils.encryption_helper import CryptoKeyType
from gslib.utils.encryption_helper import CryptoKeyWrapperFromKey
from gslib.utils.encryption_helper import FindMatchingCSEKInBotoConfig
from gslib.utils.encryption_helper import GetEncryptionKeyWrapper
from gslib.utils.hashing_helper import Base64EncodeHash
from gslib.utils.hashing_helper import CalculateB64EncodedMd5FromContents
from gslib.utils.hashing_helper import CalculateHashesFromContents
from gslib.utils.hashing_helper import CHECK_HASH_IF_FAST_ELSE_FAIL
from gslib.utils.hashing_helper import CHECK_HASH_NEVER
from gslib.utils.hashing_helper import ConcatCrc32c
from gslib.utils.hashing_helper import GetDownloadHashAlgs
from gslib.utils.hashing_helper import GetMd5
from gslib.utils.hashing_helper import GetUploadHashAlgs
from gslib.utils.hashing_helper import HashingFileUploadWrapper
from gslib.utils.metadata_util import ObjectIsGzipEncoded
from gslib.utils.parallelism_framework_util import AtomicDict
from gslib.utils.parallelism_framework_util import CheckMultiprocessingAvailableAndInit
from gslib.utils.parallelism_framework_util import PutToQueueWithTimeout
from gslib.utils.posix_util import ATIME_ATTR
from gslib.utils.posix_util import ConvertDatetimeToPOSIX
from gslib.utils.posix_util import GID_ATTR
from gslib.utils.posix_util import MODE_ATTR
from gslib.utils.posix_util import MTIME_ATTR
from gslib.utils.posix_util import ParseAndSetPOSIXAttributes
from gslib.utils.posix_util import UID_ATTR
from gslib.utils.system_util import CheckFreeSpace
from gslib.utils.system_util import GetFileSize
from gslib.utils.system_util import GetStreamFromFileUrl
from gslib.utils.system_util import IS_WINDOWS
from gslib.utils.translation_helper import AddS3MarkerAclToObjectMetadata
from gslib.utils.translation_helper import CopyObjectMetadata
from gslib.utils.translation_helper import DEFAULT_CONTENT_TYPE
from gslib.utils.translation_helper import ObjectMetadataFromHeaders
from gslib.utils.translation_helper import PreconditionsFromHeaders
from gslib.utils.translation_helper import S3MarkerAclFromObjectMetadata
from gslib.utils.unit_util import DivideAndCeil
from gslib.utils.unit_util import HumanReadableToBytes
from gslib.utils.unit_util import MakeHumanReadable
from gslib.utils.unit_util import SECONDS_PER_DAY
from gslib.utils.unit_util import TEN_MIB
from gslib.wildcard_iterator import CreateWildcardIterator

if six.PY3:
  long = int

# pylint: disable=g-import-not-at-top
if IS_WINDOWS:
  import msvcrt

# Declare copy_helper_opts as a global because namedtuple isn't aware of
# assigning to a class member (which breaks pickling done by multiprocessing).
# For details see
# http://stackoverflow.com/questions/16377215/how-to-pickle-a-namedtuple-instance-correctly
# pylint: disable=global-at-module-level
global global_copy_helper_opts

# In-memory map of local files that are currently opened for write. Used to
# ensure that if we write to the same file twice (say, for example, because the
# user specified two identical source URLs), the writes occur serially.
global open_files_map, open_files_lock
open_files_map = AtomicDict(
    manager=(parallelism_framework_util.top_level_manager
             if CheckMultiprocessingAvailableAndInit().is_available else None))

# We don't allow multiple processes on Windows, so using a process-safe lock
# would be unnecessary.
open_files_lock = parallelism_framework_util.CreateLock()

# For debugging purposes; if True, files and objects that fail hash validation
# will be saved with the below suffix appended.
_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.
"""

# When uploading a file, get the following fields in the response for
# filling in command output and manifests.
UPLOAD_RETURN_FIELDS = [
    'crc32c',
    'customerEncryption',
    'etag',
    'generation',
    'md5Hash',
    'size',
]

# This tuple is used only to encapsulate the arguments needed for
# command.Apply() in the parallel composite upload case.
# Note that content_type is used instead of a full apitools Object() because
# apitools objects are not picklable.
# filename: String name of file.
# file_start: start byte of file (may be in the middle of a file for partitioned
#             files).
# file_length: length of upload (may not be the entire length of a file for
#              partitioned files).
# src_url: FileUrl describing the source file.
# dst_url: CloudUrl describing the destination component file.
# canned_acl: canned_acl to apply to the uploaded file/component.
# content_type: content-type for final object, used for setting content-type
#               of components and final object.
# tracker_file: tracker file for this component.
# tracker_file_lock: tracker file lock for tracker file(s).
# gzip_encoded: Whether to use gzip transport encoding for the upload.
PerformParallelUploadFileToObjectArgs = namedtuple(
    'PerformParallelUploadFileToObjectArgs',
    'filename file_start file_length src_url dst_url canned_acl '
    'content_type storage_class tracker_file tracker_file_lock encryption_key_sha256 '
    'gzip_encoded')

PerformSlicedDownloadObjectToFileArgs = namedtuple(
    'PerformSlicedDownloadObjectToFileArgs',
    'component_num src_url src_obj_metadata_json dst_url download_file_name '
    'start_byte end_byte decryption_key')

# This tuple is used only to encapsulate the arguments returned by
#   _PerformSlicedDownloadObjectToFile.
# component_num: Component number.
# crc32c: CRC32C hash value (integer) of the downloaded bytes
# bytes_transferred: The number of bytes transferred, potentially less
#   than the component size if the download was resumed.
# component_total_size: The number of bytes corresponding to the whole
#    component size, potentially more than bytes_transferred
#    if the download was resumed.
# server_encoding: Content-encoding string if it was detected that the server
#    sent encoded bytes during transfer, None otherwise.
PerformSlicedDownloadReturnValues = namedtuple(
    'PerformSlicedDownloadReturnValues',
    'component_num crc32c bytes_transferred component_total_size '
    'server_encoding')

# TODO: Refactor this file to be less cumbersome. In particular, some of the
# different paths (e.g., uploading a file to an object vs. downloading an
# object to a file) could be split into separate files.

# Chunk size to use while zipping/unzipping gzip files.
GZIP_CHUNK_SIZE = 8192

# Indicates that all files should be gzipped, in _UploadFileToObject
GZIP_ALL_FILES = 'GZIP_ALL_FILES'

# Number of bytes to wait before updating a sliced download component tracker
# file.
TRACKERFILE_UPDATE_THRESHOLD = TEN_MIB

PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD = 150 * 1024 * 1024

# S3 requires special Multipart upload logic (that we currently don't implement)
# for files > 5GiB in size.
S3_MAX_UPLOAD_SIZE = 5 * 1024 * 1024 * 1024

# TODO: Create a message class that serializes posting this message once
# through the UI's global status queue.
global suggested_sliced_transfers, suggested_sliced_transfers_lock
suggested_sliced_transfers = AtomicDict(
    manager=(parallelism_framework_util.top_level_manager
             if CheckMultiprocessingAvailableAndInit().is_available else None))
suggested_sliced_transfers_lock = parallelism_framework_util.CreateLock()

COMMON_EXTENSION_RULES = {
    'md': 'text/markdown',
    'tgz': 'application/gzip',
}


class FileConcurrencySkipError(Exception):
  """Raised when skipping a file due to a concurrent, duplicate copy."""


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


def _ParallelCopyExceptionHandler(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 _PerformParallelUploadFileToObject(cls, args, thread_state=None):
  """Function argument to Apply for performing parallel composite uploads.

  Args:
    cls: Calling Command class.
    args: PerformParallelUploadFileToObjectArgs tuple describing the target.
    thread_state: gsutil Cloud API instance to use for the operation.

  Returns:
    StorageUrl representing a successfully uploaded component.
  """
  fp = FilePart(args.filename, args.file_start, args.file_length)
  gsutil_api = GetCloudApiInstance(cls, thread_state=thread_state)
  with fp:
    # We take many precautions with the component names that make collisions
    # effectively impossible. Specifying preconditions will just allow us to
    # reach a state in which uploads will always fail on retries.
    preconditions = None

    # Fill in content type and storage class if one was provided.
    dst_object_metadata = apitools_messages.Object(
        name=args.dst_url.object_name,
        bucket=args.dst_url.bucket_name,
        contentType=args.content_type,
        storageClass=args.storage_class)

    orig_prefer_api = gsutil_api.prefer_api
    try:
      if global_copy_helper_opts.canned_acl:
        # No canned ACL support in JSON, force XML API to be used for
        # upload/copy operations.
        gsutil_api.prefer_api = ApiSelector.XML
      ret = _UploadFileToObject(args.src_url,
                                fp,
                                args.file_length,
                                args.dst_url,
                                dst_object_metadata,
                                preconditions,
                                gsutil_api,
                                cls.logger,
                                cls,
                                _ParallelCopyExceptionHandler,
                                gzip_exts=None,
                                allow_splitting=False,
                                is_component=True,
                                gzip_encoded=args.gzip_encoded)
    finally:
      if global_copy_helper_opts.canned_acl:
        gsutil_api.prefer_api = orig_prefer_api

  component = ret[2]
  WriteComponentToParallelUploadTrackerFile(
      args.tracker_file,
      args.tracker_file_lock,
      component,
      cls.logger,
      encryption_key_sha256=args.encryption_key_sha256)
  return ret


CopyHelperOpts = namedtuple('CopyHelperOpts', [
    'perform_mv',
    'no_clobber',
    'daisy_chain',
    'read_args_from_stdin',
    'print_ver',
    'use_manifest',
    'preserve_acl',
    'canned_acl',
    'skip_unsupported_objects',
    'test_callback_file',
    'dest_storage_class',
])


# pylint: disable=global-variable-undefined
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


# pylint: disable=global-variable-undefined
# pylint: disable=global-variable-not-assigned
def GetCopyHelperOpts():
  """Returns namedtuple holding CopyHelper options."""
  global global_copy_helper_opts
  return global_copy_helper_opts


def _SelectDownloadStrategy(dst_url):
  """Get download strategy based on the destination object.

  Args:
    dst_url: Destination StorageUrl.

  Returns:
    gsutil Cloud API DownloadStrategy.
  """
  dst_is_special = False
  if dst_url.IsFileUrl():
    # Check explicitly first because os.stat doesn't work on 'nul' in Windows.
    if dst_url.object_name == os.devnull:
      dst_is_special = True
    try:
      mode = os.stat(dst_url.object_name).st_mode
      if stat.S_ISCHR(mode):
        dst_is_special = True
    except OSError:
      pass

  if dst_is_special:
    return CloudApi.DownloadStrategy.ONE_SHOT
  else:
    return CloudApi.DownloadStrategy.RESUMABLE


def InsistDstUrlNamesContainer(exp_dst_url, have_existing_dst_container,
                               command_name):
  """Ensures the destination URL names a container.

  Acceptable containers include directory, bucket, bucket
  subdir, and non-existent bucket subdir.

  Args:
    exp_dst_url: Wildcard-expanded destination StorageUrl.
    have_existing_dst_container: bool indicator of whether exp_dst_url
      names a container (directory, bucket, or existing bucket subdir).
    command_name: Name of command making call. May not be the same as the
        calling class's self.command_name in the case of commands implemented
        atop other commands (like mv command).

  Raises:
    CommandException: if the URL being checked does not name a container.
  """
  if ((exp_dst_url.IsFileUrl() and not exp_dst_url.IsDirectory()) or
      (exp_dst_url.IsCloudUrl() and exp_dst_url.IsBucket() and
       not have_existing_dst_container)):
    raise CommandException('Destination URL must name a directory, bucket, '
                           'or bucket\nsubdirectory for the multiple '
                           'source form of the %s command.' % command_name)


def _ShouldTreatDstUrlAsBucketSubDir(have_multiple_srcs, dst_url,
                                     have_existing_dest_subdir,
                                     src_url_names_container,
                                     recursion_requested):
  """Checks whether dst_url should be treated as a bucket "sub-directory".

  The decision about whether something constitutes a bucket "sub-directory"
  depends on whether there are multiple sources in this request and whether
  there is an existing bucket subdirectory. For example, when running the
  command:
    gsutil cp file gs://bucket/abc
  if there's no existing gs://bucket/abc bucket subdirectory we should copy
  file to the object gs://bucket/abc. In contrast, if
  there's an existing gs://bucket/abc bucket subdirectory we should copy
  file to gs://bucket/abc/file. And regardless of whether gs://bucket/abc
  exists, when running the command:
    gsutil cp file1 file2 gs://bucket/abc
  we should copy file1 to gs://bucket/abc/file1 (and similarly for file2).
  Finally, for recursive copies, if the source is a container then we should
  copy to a container as the target.  For example, when running the command:
    gsutil cp -r dir1 gs://bucket/dir2
  we should copy the subtree of dir1 to gs://bucket/dir2.

  Note that we don't disallow naming a bucket "sub-directory" where there's
  already an object at that URL. For example it's legitimate (albeit
  confusing) to have an object called gs://bucket/dir and
  then run the command
  gsutil cp file1 file2 gs://bucket/dir
  Doing so will end up with objects gs://bucket/dir, gs://bucket/dir/file1,
  and gs://bucket/dir/file2.

  Args:
    have_multiple_srcs: Bool indicator of whether this is a multi-source
        operation.
    dst_url: StorageUrl to check.
    have_existing_dest_subdir: bool indicator whether dest is an existing
      subdirectory.
    src_url_names_container: bool indicator of whether the source URL
      is a container.
    recursion_requested: True if a recursive operation has been requested.

  Returns:
    bool indicator.
  """
  if have_existing_dest_subdir:
    return True
  if dst_url.IsCloudUrl():
    return (have_multiple_srcs or
            (src_url_names_container and recursion_requested))


def _ShouldTreatDstUrlAsSingleton(src_url_names_container, have_multiple_srcs,
                                  have_existing_dest_subdir, dst_url,
                                  recursion_requested):
  """Checks that dst_url names a single file/object after wildcard expansion.

  It is possible that an object path might name a bucket sub-directory.

  Args:
    src_url_names_container: Bool indicator of whether the source for the
        operation is a container (bucket, bucket subdir, or directory).
    have_multiple_srcs: Bool indicator of whether this is a multi-source
        operation.
    have_existing_dest_subdir: bool indicator whether dest is an existing
      subdirectory.
    dst_url: StorageUrl to check.
    recursion_requested: True if a recursive operation has been requested.

  Returns:
    bool indicator.
  """
  if recursion_requested and src_url_names_container:
    return False
  if dst_url.IsFileUrl():
    return not dst_url.IsDirectory()
  else:  # dst_url.IsCloudUrl()
    return (not have_multiple_srcs and not have_existing_dest_subdir and
            dst_url.IsObject())


def _IsUrlValidParentDir(url):
  """Returns True if not FileUrl ending in  relative path symbols.

  A URL is invalid if it is a FileUrl and the parent directory of the file is a
  relative path symbol. Unix will not allow a file itself to be named with a
  relative path symbol, but one can be the parent. Notably, "../obj" can lead
  to unexpected behavior at the copy destination. We examine the pre-recursion
  "url", which might point to "..", to see if the parent is valid.

  If the user does a recursive copy from a URL, it may not end up
  the final parent of the copied object. For example, see: "dir/nested_dir/obj".

  If you ran "cp -r dir gs://bucket" from the parent of "dir", then the "url"
  arg would be "dir", but "nested_dir" would be the parent of "obj".
  This actually doesn't matter since recursion won't add relative path symbols
  to the path. However, we still return if "url" is valid because
  there are cases where we need to copy every parent directory up to
  "dir" to prevent file name conflicts.

  Args:
    url: StorageUrl before recursion.

  Returns:
    Boolean indicating if the "url" is valid as a parent directory.
  """
  if not url.IsFileUrl():
    return True

  url_string_minus_trailing_delimiter = url.versionless_url_string.rstrip(
      url.delim)
  _, _, after_last_delimiter = (url_string_minus_trailing_delimiter.rpartition(
      url.delim))

  return after_last_delimiter not in storage_url.RELATIVE_PATH_SYMBOLS and (
      after_last_delimiter not in [
          url.scheme + '://' + symbol
          for symbol in storage_url.RELATIVE_PATH_SYMBOLS
      ])


def ConstructDstUrl(src_url,
                    exp_src_url,
                    src_url_names_container,
                    have_multiple_srcs,
                    has_multiple_top_level_srcs,
                    exp_dst_url,
                    have_existing_dest_subdir,
                    recursion_requested,
                    preserve_posix=False):
  """Constructs the destination URL for a given exp_src_url/exp_dst_url pair.

  Uses context-dependent naming rules that mimic Linux cp and mv behavior.

  Args:
    src_url: Source StorageUrl to be copied.
    exp_src_url: Single StorageUrl from wildcard expansion of src_url.
    src_url_names_container: True if src_url names a container (including the
        case of a wildcard-named bucket subdir (like gs://bucket/abc,
        where gs://bucket/abc/* matched some objects).
    have_multiple_srcs: True if this is a multi-source request. This can be
        true if src_url wildcard-expanded to multiple URLs or if there were
        multiple source URLs in the request.
    has_multiple_top_level_srcs: Same as have_multiple_srcs but measured
        before recursion.
    exp_dst_url: the expanded StorageUrl requested for the cp destination.
        Final written path is constructed from this plus a context-dependent
        variant of src_url.
    have_existing_dest_subdir: bool indicator whether dest is an existing
      subdirectory.
    recursion_requested: True if a recursive operation has been requested.
    preserve_posix: True if preservation of posix attributes has been requested.

  Returns:
    StorageUrl to use for copy.

  Raises:
    CommandException if destination object name not specified for
    source and source is a stream.
  """
  if (exp_dst_url.IsFileUrl() and exp_dst_url.IsStream() and preserve_posix):
    raise CommandException('Cannot preserve POSIX attributes with a stream.')

  if _ShouldTreatDstUrlAsSingleton(src_url_names_container, have_multiple_srcs,
                                   have_existing_dest_subdir, exp_dst_url,
                                   recursion_requested):
    # We're copying one file or object to one file or object.
    return exp_dst_url

  if exp_src_url.IsFileUrl() and (exp_src_url.IsStream() or
                                  exp_src_url.IsFifo()):
    if have_existing_dest_subdir:
      type_text = 'stream' if exp_src_url.IsStream() else 'named pipe'
      raise CommandException('Destination object name needed when '
                             'source is a %s' % type_text)
    return exp_dst_url

  if not recursion_requested and not have_multiple_srcs:
    # We're copying one file or object to a subdirectory. Append final comp
    # of exp_src_url to exp_dst_url.
    src_final_comp = exp_src_url.object_name.rpartition(src_url.delim)[-1]
    return StorageUrlFromString('%s%s%s' % (exp_dst_url.url_string.rstrip(
        exp_dst_url.delim), exp_dst_url.delim, src_final_comp))

  # Else we're copying multiple sources to a directory, bucket, or a bucket
  # "sub-directory".

  # Ensure exp_dst_url ends in delim char if we're doing a multi-src copy or
  # a copy to a directory. (The check for copying to a directory needs
  # special-case handling so that the command:
  #   gsutil cp gs://bucket/obj dir
  # will turn into file://dir/ instead of file://dir -- the latter would cause
  # the file "dirobj" to be created.)
  # Note: need to check have_multiple_srcs or src_url.names_container()
  # because src_url could be a bucket containing a single object, named
  # as gs://bucket.
  if ((have_multiple_srcs or src_url_names_container or
       (exp_dst_url.IsFileUrl() and exp_dst_url.IsDirectory())) and
      not exp_dst_url.url_string.endswith(exp_dst_url.delim)):
    exp_dst_url = StorageUrlFromString(
        '%s%s' % (exp_dst_url.url_string, exp_dst_url.delim))

  src_url_is_valid_parent = _IsUrlValidParentDir(src_url)
  if not src_url_is_valid_parent and has_multiple_top_level_srcs:
    # To avoid top-level name conflicts, we need to copy the parent dir.
    # However, that cannot be done because the parent dir has an invalid name.
    raise InvalidUrlError(
        'Presence of multiple top-level sources and invalid expanded URL'
        ' make file name conflicts possible for URL: {}'.format(src_url))

  # Making naming behavior match how things work with local Linux cp and mv
  # operations depends on many factors, including whether the destination is a
  # container, and the plurality of the source(s).
  # 1. Recursively copying from directories, buckets, or bucket subdirs should
  #    result in objects/files mirroring the source hierarchy. For example:
  #      gsutil cp -r dir1/dir2 gs://bucket
  #    should create the object gs://bucket/dir2/file2, assuming dir1/dir2
  #    contains file2).
  #
  #    To be consistent with Linux cp behavior, there's one more wrinkle when
  #    working with subdirs: The resulting object names depend on whether the
  #    destination subdirectory exists. For example, if gs://bucket/subdir
  #    exists, the command:
  #      gsutil cp -r dir1/dir2 gs://bucket/subdir
  #    should create objects named like gs://bucket/subdir/dir2/a/b/c. In
  #    contrast, if gs://bucket/subdir does not exist, this same command
  #    should create objects named like gs://bucket/subdir/a/b/c.
  #
  #    If there are multiple top-level source items, preserve source parent
  #    dirs. This is similar to when the destination dir already exists and
  #    avoids conflicts such as "dir1/f.txt" and "dir2/f.txt" both getting
  #    copied to "gs://bucket/f.txt". Linux normally errors on these conflicts,
  #    but we cannot do that because we need to give users the ability to create
  #    dirs as they copy to the cloud.
  #
  #    Note: "mv" is similar to running "cp -r" followed by source deletion.
  #
  # 2. Copying individual files or objects to dirs, buckets or bucket subdirs
  #    should result in objects/files named by the final source file name
  #    component. Example:
  #      gsutil cp dir1/*.txt gs://bucket
  #    should create the objects gs://bucket/f1.txt and gs://bucket/f2.txt,
  #    assuming dir1 contains f1.txt and f2.txt.

  # Ignore the "multiple top-level sources" rule if using double wildcard **
  # because that treats all files as top-level, in which case the user doesn't
  # want to preserve directories.
  preserve_src_top_level_dirs = ('**' not in src_url.versionless_url_string and
                                 src_url_is_valid_parent and
                                 (has_multiple_top_level_srcs or
                                  have_existing_dest_subdir))
  if preserve_src_top_level_dirs or (src_url_names_container and
                                     (exp_dst_url.IsCloudUrl() or
                                      exp_dst_url.IsDirectory())):
    # Case 1.  Container copy to a destination other than a file.
    # Build dst_key_name from subpath of exp_src_url past
    # where src_url ends. For example, for src_url=gs://bucket/ and
    # exp_src_url=gs://bucket/src_subdir/obj, dst_key_name should be
    # src_subdir/obj.
    src_url_path_sans_final_dir = GetPathBeforeFinalDir(src_url, exp_src_url)
    dst_key_name = exp_src_url.versionless_url_string[
        len(src_url_path_sans_final_dir):].lstrip(src_url.delim)

    if not preserve_src_top_level_dirs:
      # Only copy file name, not parent dir.
      dst_key_name = dst_key_name.partition(src_url.delim)[-1]

  else:
    # Case 2.
    dst_key_name = exp_src_url.object_name.rpartition(src_url.delim)[-1]

  if (exp_dst_url.IsFileUrl() or _ShouldTreatDstUrlAsBucketSubDir(
      have_multiple_srcs, exp_dst_url, have_existing_dest_subdir,
      src_url_names_container, recursion_requested)):
    if exp_dst_url.object_name and exp_dst_url.object_name.endswith(
        exp_dst_url.delim):
      dst_key_name = '%s%s%s' % (exp_dst_url.object_name.rstrip(
          exp_dst_url.delim), exp_dst_url.delim, dst_key_name)
    else:
      delim = exp_dst_url.delim if exp_dst_url.object_name else ''
      dst_key_name = '%s%s%s' % (exp_dst_url.object_name or
                                 '', delim, dst_key_name)

  new_exp_dst_url = exp_dst_url.Clone()
  new_exp_dst_url.object_name = dst_key_name.replace(src_url.delim,
                                                     exp_dst_url.delim)
  return new_exp_dst_url


def _CreateDigestsFromDigesters(digesters):
  digests = {}
  if digesters:
    for alg in digesters:
      digests[alg] = base64.b64encode(
          digesters[alg].digest()).rstrip(b'\n').decode('ascii')
  return digests


def _CreateDigestsFromLocalFile(status_queue, algs, file_name, src_url,
                                src_obj_metadata):
  """Creates a base64 CRC32C and/or MD5 digest from file_name.

  Args:
    status_queue: Queue for posting progress messages for UI/Analytics.
    algs: List of algorithms to compute.
    file_name: File to digest.
    src_url: StorageUrl for local object. Used to track progress.
    src_obj_metadata: Metadata of source object.

  Returns:
    Dict of algorithm name : base 64 encoded digest
  """
  hash_dict = {}
  if 'md5' in algs:
    hash_dict['md5'] = GetMd5()
  if 'crc32c' in algs:
    hash_dict['crc32c'] = crcmod.predefined.Crc('crc-32c')
  with open(file_name, 'rb') as fp:
    CalculateHashesFromContents(fp,
                                hash_dict,
                                callback_processor=ProgressCallbackWithTimeout(
                                    src_obj_metadata.size,
                                    FileProgressCallbackHandler(
                                        status_queue,
                                        src_url=src_url,
                                        operation_name='Hashing').call))
  digests = {}
  for alg_name, digest in six.iteritems(hash_dict):
    digests[alg_name] = Base64EncodeHash(digest.hexdigest())
  return digests


def _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
                      dst_obj_metadata):
  """Validates integrity of two cloud objects copied via daisy-chain.

  Args:
    logger: for outputting log messages.
    src_url: CloudUrl for source cloud object.
    dst_url: CloudUrl for destination cloud object.
    src_obj_metadata: Cloud Object metadata for object being downloaded from.
    dst_obj_metadata: Cloud Object metadata for object being uploaded to.

  Raises:
    CommandException: if cloud digests don't match local digests.
  """
  # See hack comment in _CheckHashes.

  # Sometimes (e.g. when kms is enabled for s3) the values we check below are
  # not actually content hashes. The early exit here provides users a workaround
  # for this case and any others we've missed.
  check_hashes_config = config.get('GSUtil', 'check_hashes',
                                   CHECK_HASH_IF_FAST_ELSE_FAIL)
  if check_hashes_config == CHECK_HASH_NEVER:
    return

  checked_one = False
  download_hashes = {}
  upload_hashes = {}
  if src_obj_metadata.md5Hash:
    src_md5hash = six.ensure_binary(src_obj_metadata.md5Hash)
    download_hashes['md5'] = src_md5hash
  if src_obj_metadata.crc32c:
    src_crc32c_hash = six.ensure_binary(src_obj_metadata.crc32c)
    download_hashes['crc32c'] = src_crc32c_hash
  if dst_obj_metadata.md5Hash:
    dst_md5hash = six.ensure_binary(dst_obj_metadata.md5Hash)
    upload_hashes['md5'] = dst_md5hash
  if dst_obj_metadata.crc32c:
    dst_crc32c_hash = six.ensure_binary(dst_obj_metadata.crc32c)
    upload_hashes['crc32c'] = dst_crc32c_hash

  for alg, upload_b64_digest in six.iteritems(upload_hashes):
    if alg not in download_hashes:
      continue

    download_b64_digest = download_hashes[alg]
    if six.PY3 and isinstance(download_b64_digest, str):
      download_b64_digest = download_b64_digest.encode('ascii')
    logger.debug('Comparing source vs destination %s-checksum for %s. (%s/%s)',
                 alg, dst_url, download_b64_digest, upload_b64_digest)
    if download_b64_digest != upload_b64_digest:
      raise HashMismatchException(
          '%s signature for source object (%s) doesn\'t match '
          'destination object digest (%s). Object (%s) will be deleted.' %
          (alg, download_b64_digest, upload_b64_digest, dst_url))
    checked_one = True
  if not checked_one:
    # One known way this can currently happen is when downloading objects larger
    # than 5 GiB from S3 (for which the etag is not an MD5).
    logger.warn(
        'WARNING: Found no hashes to validate object downloaded from %s and '
        'uploaded to %s. Integrity cannot be assured without hashes.', src_url,
        dst_url)


def _CheckHashes(logger,
                 obj_url,
                 obj_metadata,
                 file_name,
                 digests,
                 is_upload=False):
  """Validates integrity by comparing cloud digest to local digest.

  Args:
    logger: for outputting log messages.
    obj_url: CloudUrl for cloud object.
    obj_metadata: Cloud Object being downloaded from or uploaded to.
    file_name: Local file name on disk being downloaded to or uploaded from
               (used only for logging).
    digests: Computed Digests for the object.
    is_upload: If true, comparing for an uploaded object (controls logging).

  Raises:
    CommandException: if cloud digests don't match local digests.
  """
  # Hack below.
  # I cannot track down all of the code paths that get here, so I finally
  # gave up and opted to convert all of the hashes to str. I know that they
  # *should* be bytes, but the path of least resistance led to str.
  # Not a nice thing, but for now it makes tests pass...
  # Update: Since the _CheckCloudHashes function above needs to be changed
  # as well, I am going to make the executive decision that hashes are
  # bytes - here as well. It's what the hash and base64 PY3 libs return,
  # and should be the native format for these things.
  local_hashes = digests
  cloud_hashes = {}
  if obj_metadata.md5Hash:
    md5_b64_digest = six.ensure_binary(obj_metadata.md5Hash)
    cloud_hashes['md5'] = md5_b64_digest.rstrip(b'\n')
  if obj_metadata.crc32c:
    crc32c_b64_hash = six.ensure_binary(obj_metadata.crc32c)
    cloud_hashes['crc32c'] = crc32c_b64_hash.rstrip(b'\n')

  checked_one = False
  for alg in local_hashes:
    if alg not in cloud_hashes:
      continue

    local_b64_digest = six.ensure_binary(local_hashes[alg])
    cloud_b64_digest = cloud_hashes[alg]
    logger.debug('Comparing local vs cloud %s-checksum for %s. (%s/%s)', alg,
                 file_name, local_b64_digest, cloud_b64_digest)
    if local_b64_digest != cloud_b64_digest:
      raise HashMismatchException(
          '%s signature computed for local file (%s) doesn\'t match '
          'cloud-supplied digest (%s). %s (%s) will be deleted.' %
          (alg, local_b64_digest, cloud_b64_digest, 'Cloud object'
           if is_upload else 'Local file', obj_url if is_upload else file_name))
    checked_one = True
  if not checked_one:
    if is_upload:
      logger.warn(
          'WARNING: Found no hashes to validate object uploaded to %s. '
          'Integrity cannot be assured without hashes.', obj_url)
    else:
      # One known way this can currently happen is when downloading objects larger
      # than 5 GB from S3 (for which the etag is not an MD5).
      logger.warn(
          'WARNING: Found no hashes to validate object downloaded to %s. '
          'Integrity cannot be assured without hashes.', file_name)


def IsNoClobberServerException(e):
  """Checks to see if the server attempted to clobber a file.

  In this case we specified via a precondition that we didn't want the file
  clobbered.

  Args:
    e: The Exception that was generated by a failed copy operation

  Returns:
    bool indicator - True indicates that the server did attempt to clobber
        an existing file.
  """
  return ((isinstance(e, PreconditionException)) or
          (isinstance(e, ResumableUploadException) and '412' in e.message))


def CheckForDirFileConflict(exp_src_url, dst_url):
  """Checks whether copying exp_src_url into dst_url is not possible.

     This happens if a directory exists in local file system where a file
     needs to go or vice versa. In that case we print an error message and
     exits. Example: if the file "./x" exists and you try to do:
       gsutil cp gs://mybucket/x/y .
     the request can't succeed because it requires a directory where
     the file x exists.

     Note that we don't enforce any corresponding restrictions for buckets,
     because the flat namespace semantics for buckets doesn't prohibit such
     cases the way hierarchical file systems do. For example, if a bucket
     contains an object called gs://bucket/dir and then you run the command:
       gsutil cp file1 file2 gs://bucket/dir
     you'll end up with objects gs://bucket/dir, gs://bucket/dir/file1, and
     gs://bucket/dir/file2.

  Args:
    exp_src_url: Expanded source StorageUrl.
    dst_url: Destination StorageUrl.

  Raises:
    CommandException: if errors encountered.
  """
  if dst_url.IsCloudUrl():
    # The problem can only happen for file destination URLs.
    return
  dst_path = dst_url.object_name
  final_dir = os.path.dirname(dst_path)
  if os.path.isfile(final_dir):
    raise CommandException('Cannot retrieve %s because a file exists '
                           'where a directory needs to be created (%s).' %
                           (exp_src_url.url_string, final_dir))
  if os.path.isdir(dst_path):
    raise CommandException('Cannot retrieve %s because a directory exists '
                           '(%s) where the file needs to be created.' %
                           (exp_src_url.url_string, dst_path))


def _PartitionFile(canned_acl,
                   content_type,
                   dst_bucket_url,
                   file_size,
                   fp,
                   random_prefix,
                   src_url,
                   storage_class,
                   tracker_file,
                   tracker_file_lock,
                   encryption_key_sha256=None,
                   gzip_encoded=False):
  """Partitions a file into FilePart objects to be uploaded and later composed.

  These objects, when composed, will match the original file. This entails
  splitting the file into parts, naming and forming a destination URL for each
  part, and also providing the PerformParallelUploadFileToObjectArgs
  corresponding to each part.

  Args:
    canned_acl: The user-provided canned_acl, if applicable.
    content_type: content type for the component and final objects.
    dst_bucket_url: CloudUrl for the destination bucket.
    file_size: The size of fp, in bytes.
    fp: The file object to be partitioned.
    random_prefix: The randomly-generated prefix used to prevent collisions
                   among the temporary component names.
    src_url: Source FileUrl from the original command.
    storage_class: storage class for the component and final objects.
    tracker_file: The path to the parallel composite upload tracker file.
    tracker_file_lock: The lock protecting access to the tracker file.
    encryption_key_sha256: Encryption key SHA256 for use in this upload, if any.
    gzip_encoded: Whether to use gzip transport encoding for the upload.

  Returns:
    dst_args: The destination URIs for the temporary component objects.
  """
  parallel_composite_upload_component_size = HumanReadableToBytes(
      config.get('GSUtil', 'parallel_composite_upload_component_size',
                 DEFAULT_PARALLEL_COMPOSITE_UPLOAD_COMPONENT_SIZE))
  (num_components, component_size) = _GetPartitionInfo(
      file_size, MAX_COMPOSE_ARITY, parallel_composite_upload_component_size)

  dst_args = {}  # Arguments to create commands and pass to subprocesses.
  file_names = []  # Used for the 2-step process of forming dst_args.
  for i in range(num_components):
    # "Salt" the object name with something a user is very unlikely to have
    # used in an object name, then hash the extended name to make sure
    # we don't run into problems with name length. Using a deterministic
    # naming scheme for the temporary components allows users to take
    # advantage of resumable uploads for each component.
    encoded_name = six.ensure_binary(PARALLEL_UPLOAD_STATIC_SALT + fp.name)
    content_md5 = GetMd5()
    content_md5.update(encoded_name)
    digest = content_md5.hexdigest()
    temp_file_name = (random_prefix + PARALLEL_UPLOAD_TEMP_NAMESPACE + digest +
                      '_' + str(i))
    tmp_dst_url = dst_bucket_url.Clone()
    tmp_dst_url.object_name = temp_file_name

    if i < (num_components - 1):
      # Every component except possibly the last is the same size.
      file_part_length = component_size
    else:
      # The last component just gets all of the remaining bytes.
      file_part_length = (file_size - ((num_components - 1) * component_size))
    offset = i * component_size
    func_args = PerformParallelUploadFileToObjectArgs(
        fp.name, offset, file_part_length, src_url, tmp_dst_url, canned_acl,
        content_type, storage_class, tracker_file, tracker_file_lock,
        encryption_key_sha256, gzip_encoded)
    file_names.append(temp_file_name)
    dst_args[temp_file_name] = func_args

  return dst_args


def _GetComponentNumber(component):
  """Gets component number from component CloudUrl.

  Used during parallel composite upload.

  Args:
    component: CloudUrl representing component.

  Returns:
    component number
  """
  return int(component.object_name[component.object_name.rfind('_') + 1:])


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):
  """Uploads a local file to a cloud object using parallel composite upload.

  The file is partitioned into parts, and then the parts are uploaded in
  parallel, composed to form the original destination object, and deleted.

  Args:
    fp: The file object to be uploaded.
    src_url: FileUrl representing the local file.
    dst_url: CloudUrl representing the destination file.
    dst_obj_metadata: apitools Object describing the destination object.
    canned_acl: The canned acl to apply to the object, if any.
    file_size: The size of the source file in bytes.
    preconditions: Cloud API Preconditions for the final object.
    gsutil_api: gsutil Cloud API instance to use.
    command_obj: Command object (for calling Apply).
    copy_exception_handler: Copy exception handler (for use in Apply).
    logger: logging.Logger for outputting log messages.
    gzip_encoded: Whether to use gzip transport encoding for the upload.

  Returns:
    Elapsed upload time, uploaded Object with generation, crc32c, and size
    fields populated.
  """
  start_time = time.time()
  dst_bucket_url = StorageUrlFromString(dst_url.bucket_url_string)
  api_selector = gsutil_api.GetApiSelector(provider=dst_url.scheme)

  encryption_keywrapper = GetEncryptionKeyWrapper(config)
  encryption_key_sha256 = (encryption_keywrapper.crypto_key_sha256
                           if encryption_keywrapper else None)

  # Determine which components, if any, have already been successfully
  # uploaded.
  tracker_file_name = GetTrackerFilePath(dst_url,
                                         TrackerFileType.PARALLEL_UPLOAD,
                                         api_selector, src_url)

  (existing_enc_key_sha256, existing_prefix,
   existing_components) = (ReadParallelUploadTrackerFile(
       tracker_file_name, logger))

  # Ensure that the tracker data is still valid (encryption keys match) and
  # perform any necessary cleanup.
  (existing_prefix, existing_components) = ValidateParallelCompositeTrackerData(
      tracker_file_name, existing_enc_key_sha256, existing_prefix,
      existing_components, encryption_key_sha256, dst_bucket_url, command_obj,
      logger, _DeleteTempComponentObjectFn, _RmExceptionHandler)

  encryption_key_sha256 = (encryption_key_sha256.decode('ascii')
                           if encryption_key_sha256 is not None else None)
  random_prefix = (existing_prefix if existing_prefix is not None else
                   GenerateComponentObjectPrefix(
                       encryption_key_sha256=encryption_key_sha256))

  # Create (or overwrite) the tracker file for the upload.
  WriteParallelUploadTrackerFile(tracker_file_name,
                                 random_prefix,
                                 existing_components,
                                 encryption_key_sha256=encryption_key_sha256)

  # Protect the tracker file within calls to Apply.
  tracker_file_lock = parallelism_framework_util.CreateLock()
  # Dict to track component info so we may align FileMessage values
  # before and after the operation.
  components_info = {}
  # Get the set of all components that should be uploaded.
  dst_args = _PartitionFile(canned_acl,
                            dst_obj_metadata.contentType,
                            dst_bucket_url,
                            file_size,
                            fp,
                            random_prefix,
                            src_url,
                            dst_obj_metadata.storageClass,
                            tracker_file_name,
                            tracker_file_lock,
                            encryption_key_sha256=encryption_key_sha256,
                            gzip_encoded=gzip_encoded)

  (components_to_upload, existing_components,
   existing_objects_to_delete) = (FilterExistingComponents(
       dst_args, existing_components, dst_bucket_url, gsutil_api))

  # Assign a start message to each different component type
  for component in components_to_upload:
    components_info[component.dst_url.url_string] = (
        FileMessage.COMPONENT_TO_UPLOAD, component.file_length)
    PutToQueueWithTimeout(
        gsutil_api.status_queue,
        FileMessage(component.src_url,
                    component.dst_url,
                    time.time(),
                    size=component.file_length,
                    finished=False,
                    component_num=_GetComponentNumber(component.dst_url),
                    message_type=FileMessage.COMPONENT_TO_UPLOAD))

  for component in existing_components:
    component_str = component[0].versionless_url_string
    components_info[component_str] = (FileMessage.EXISTING_COMPONENT,
                                      component[1])
    PutToQueueWithTimeout(
        gsutil_api.status_queue,
        FileMessage(src_url,
                    component[0],
                    time.time(),
                    finished=False,
                    size=component[1],
                    component_num=_GetComponentNumber(component[0]),
                    message_type=FileMessage.EXISTING_COMPONENT))

  for component in existing_objects_to_delete:
    PutToQueueWithTimeout(
        gsutil_api.status_queue,
        FileMessage(src_url,
                    component,
                    time.time(),
                    finished=False,
                    message_type=FileMessage.EXISTING_OBJECT_TO_DELETE))
  # In parallel, copy all of the file parts that haven't already been
  # uploaded to temporary objects.
  cp_results = command_obj.Apply(
      _PerformParallelUploadFileToObject,
      components_to_upload,
      copy_exception_handler, ('op_failure_count', 'total_bytes_transferred'),
      arg_checker=gslib.command.DummyArgChecker,
      parallel_operations_override=command_obj.ParallelOverrideReason.SLICE,
      should_return_results=True)
  uploaded_components = []
  for cp_result in cp_results:
    uploaded_components.append(cp_result[2])
  components = uploaded_components + [i[0] for i in existing_components]

  if len(components) == len(dst_args):
    # Only try to compose if all of the components were uploaded successfully.
    # Sort the components so that they will be composed in the correct order.
    components = sorted(components, key=_GetComponentNumber)

    request_components = []
    for component_url in components:
      src_obj_metadata = (
          apitools_messages.ComposeRequest.SourceObjectsValueListEntry(
              name=component_url.object_name))
      if component_url.HasGeneration():
        src_obj_metadata.generation = long(component_url.generation)
      request_components.append(src_obj_metadata)

    composed_object = gsutil_api.ComposeObject(
        request_components,
        dst_obj_metadata,
        preconditions=preconditions,
        provider=dst_url.scheme,
        fields=['crc32c', 'generation', 'size'],
        encryption_tuple=encryption_keywrapper)

    try:
      # Make sure only to delete things that we know were successfully
      # uploaded (as opposed to all of the objects that we attempted to
      # create) so that we don't delete any preexisting objects, except for
      # those that were uploaded by a previous, failed run and have since
      # changed (but still have an old generation lying around).
      objects_to_delete = components + existing_objects_to_delete
      command_obj.Apply(
          _DeleteTempComponentObjectFn,
          objects_to_delete,
          _RmExceptionHandler,
          arg_checker=gslib.command.DummyArgChecker,
          parallel_operations_override=command_obj.ParallelOverrideReason.SLICE)
      # Assign an end message to each different component type
      for component in components:
        component_str = component.versionless_url_string
        try:
          PutToQueueWithTimeout(
              gsutil_api.status_queue,
              FileMessage(src_url,
                          component,
                          time.time(),
                          finished=True,
                          component_num=_GetComponentNumber(component),
                          size=components_info[component_str][1],
                          message_type=components_info[component_str][0]))
        except:  # pylint: disable=bare-except
          PutToQueueWithTimeout(
              gsutil_api.status_queue,
              FileMessage(src_url, component, time.time(), finished=True))

      for component in existing_objects_to_delete:
        PutToQueueWithTimeout(
            gsutil_api.status_queue,
            FileMessage(src_url,
                        component,
                        time.time(),
                        finished=True,
                        message_type=FileMessage.EXISTING_OBJECT_TO_DELETE))
    except Exception:  # pylint: disable=broad-except
      # If some of the delete calls fail, don't cause the whole command to
      # fail. The copy was successful iff the compose call succeeded, so
      # reduce this to a warning.
      logger.warn(
          'Failed to delete some of the following temporary objects:\n' +
          '\n'.join(dst_args.keys()))
    finally:
      with tracker_file_lock:
        DeleteTrackerFile(tracker_file_name)
  else:
    # Some of the components failed to upload. In this case, we want to exit
    # without deleting the objects.
    raise CommandException(
        'Some temporary components were not uploaded successfully. '
        'Please retry this upload.')

  elapsed_time = time.time() - start_time
  return elapsed_time, composed_object


def _ShouldDoParallelCompositeUpload(logger,
                                     allow_splitting,
                                     src_url,
                                     dst_url,
                                     file_size,
                                     gsutil_api,
                                     canned_acl=None):
  """Determines whether parallel composite upload strategy should be used.

  Args:
    logger: for outputting log messages.
    allow_splitting: If false, then this function returns false.
    src_url: FileUrl corresponding to a local file.
    dst_url: CloudUrl corresponding to destination cloud object.
    file_size: The size of the source file, in bytes.
    gsutil_api: CloudApi that may be used to check if the destination bucket
        has any metadata attributes set that would discourage us from using
        parallel composite uploads.
    canned_acl: Canned ACL to apply to destination object, if any.

  Returns:
    True iff a parallel upload should be performed on the source file.
  """

  global suggested_sliced_transfers, suggested_sliced_transfers_lock
  parallel_composite_upload_threshold = HumanReadableToBytes(
      config.get('GSUtil', 'parallel_composite_upload_threshold',
                 DEFAULT_PARALLEL_COMPOSITE_UPLOAD_THRESHOLD))

  all_factors_but_size = (
      allow_splitting  # Don't split the pieces multiple times.
      and not src_url.IsStream()  # We can't partition streams.
      and not src_url.IsFifo()  # We can't partition fifos.
      and dst_url.scheme == 'gs'  # Compose is only for gs.
      and not canned_acl)  # TODO: Implement canned ACL support for compose.

  # Since parallel composite uploads are disabled by default, make user aware of
  # them.
  # TODO: Once compiled crcmod is being distributed by major Linux distributions
  # remove this check.
  if (all_factors_but_size and parallel_composite_upload_threshold == 0 and
      file_size >= PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD):
    with suggested_sliced_transfers_lock:
      if not suggested_sliced_transfers.get('suggested'):
        logger.info('\n'.join(
            textwrap.wrap(
                '==> NOTE: You are uploading one or more large file(s), which '
                'would run significantly faster if you enable parallel composite '
                'uploads. This feature can be enabled by editing the '
                '"parallel_composite_upload_threshold" value in your .boto '
                'configuration file. However, note that if you do this large files '
                'will be uploaded as '
                '`composite objects <https://cloud.google.com/storage/docs/composite-objects>`_,'  # pylint: disable=line-too-long
                'which means that any user who downloads such objects will need to '
                'have a compiled crcmod installed (see "gsutil help crcmod"). This '
                'is because without a compiled crcmod, computing checksums on '
                'composite objects is so slow that gsutil disables downloads of '
                'composite objects.')) + '\n')
        suggested_sliced_transfers['suggested'] = True

  return (all_factors_but_size and parallel_composite_upload_threshold > 0 and
          file_size >= parallel_composite_upload_threshold)


def ExpandUrlToSingleBlr(url_str,
                         gsutil_api,
                         project_id,
                         treat_nonexistent_object_as_subdir=False,
                         logger=None):
  """Expands wildcard if present in url_str.

  Args:
    url_str: String representation of requested url.
    gsutil_api: gsutil Cloud API instance to use.
    project_id: project ID to use (for iterators).
    treat_nonexistent_object_as_subdir: indicates if should treat a non-existent
        object as a subdir.
    logger: logging.Logger instance to use for output. If None, the root Logger
        will be used.

  Returns:
      (exp_url, have_existing_dst_container)
      where exp_url is a StorageUrl
      and have_existing_dst_container is a bool indicating whether
      exp_url names an existing directory, bucket, or bucket subdirectory.
      In the case where we match a subdirectory AND an object, the
      object is returned.

  Raises:
    CommandException: if url_str matched more than 1 URL.
  """
  logger = logger or logging.Logger()
  # Handle wildcarded url case.
  if ContainsWildcard(url_str):
    blr_expansion = list(
        CreateWildcardIterator(url_str,
                               gsutil_api,
                               project_id=project_id,
                               logger=logger))
    if len(blr_expansion) != 1:
      raise CommandException('Destination (%s) must match exactly 1 URL' %
                             url_str)
    blr = blr_expansion[0]
    # BLR is either an OBJECT, PREFIX, or BUCKET; the latter two represent
    # directories.
    return (StorageUrlFromString(blr.url_string), not blr.IsObject())

  storage_url = StorageUrlFromString(url_str)

  # Handle non-wildcarded URL.
  if storage_url.IsFileUrl():
    return (storage_url, storage_url.IsDirectory())

  # At this point we have a cloud URL.
  if storage_url.IsBucket():
    return (storage_url, True)

  # For object/prefix URLs, there are four cases that indicate the destination
  # is a cloud subdirectory; these are always considered to be an existing
  # container. Checking each case allows gsutil to provide Unix-like
  # destination folder semantics, but requires up to three HTTP calls, noted
  # below.

  # Case 1: If a placeholder object ending with '/' exists.
  if IsCloudSubdirPlaceholder(storage_url):
    return (storage_url, True)

  # Get version of object name without trailing slash for matching prefixes
  prefix = storage_url.object_name.rstrip('/')

  # HTTP call to make an eventually consistent check for a matching prefix,
  # _$folder$, or empty listing.
  list_iterator = gsutil_api.ListObjects(storage_url.bucket_name,
                                         prefix=prefix,
                                         delimiter='/',
                                         provider=storage_url.scheme,
                                         fields=['prefixes', 'items/name'])
  for obj_or_prefix in list_iterator:
    # To conserve HTTP calls for the common case, we make a single listing
    # that covers prefixes and object names. Listing object names covers the
    # _$folder$ case and the nonexistent-object-as-subdir case. However, if
    # there are many existing objects for which the target URL is an exact
    # prefix, this listing could be paginated and span multiple HTTP calls.
    # If this case becomes common, we could heurestically abort the
    # listing operation after the first page of results and just query for the
    # _$folder$ object directly using GetObjectMetadata.
    # TODO: currently the ListObjects iterator yields objects before prefixes,
    # because ls depends on this iteration order for proper display.  We could
    # save up to 1ms in determining that a destination is a prefix if we had a
    # way to yield prefixes first, but this would require poking a major hole
    # through the abstraction to control this iteration order.
    if (obj_or_prefix.datatype == CloudApi.CsObjectOrPrefixType.PREFIX and
        obj_or_prefix.data == prefix + '/'):
      # Case 2: If there is a matching prefix when listing the destination URL.
      return (storage_url, True)
    elif (obj_or_prefix.datatype == CloudApi.CsObjectOrPrefixType.OBJECT and
          obj_or_prefix.data.name == storage_url.object_name + '_$folder$'):
      # Case 3: If a placeholder object matching destination + _$folder$
      # exists.
      return (storage_url, True)
    elif (obj_or_prefix.datatype == CloudApi.CsObjectOrPrefixType.OBJECT and
          obj_or_prefix.data.name == storage_url.object_name):
      # The object exists but it is not a container
      return (storage_url, False)

  # Case 4: If no objects/prefixes matched, and nonexistent objects should be
  # treated as subdirectories.
  return (storage_url, treat_nonexistent_object_as_subdir)


def TriggerReauthForDestinationProviderIfNecessary(destination_url, gsutil_api,
                                                   worker_count):
  """Makes a request to the destination API provider to trigger reauth.

  Addresses https://github.com/GoogleCloudPlatform/gsutil/issues/1639.

  If an API call occurs in a child process, the library that handles
  reauth will fail. We need to make at least one API call in the main
  process to allow a user to reauthorize.

  For cloud source URLs this already happens because the plurality of
  the source name expansion iterator is checked in the main thread. For
  cloud destination URLs, only some situations result in a similar API
  call. In these situations, this function exits without performing an
  API call. In others, this function performs an API call to trigger
  reauth.

  Args:
    destination_url (StorageUrl): The destination of the transfer.
    gsutil_api (CloudApiDelegator): API to use for the GetBucket call.
    worker_count (int): If greater than 1, assume that parallel execution
      is used. Technically, reauth challenges can be answered in the main
      process, but they may be triggered multiple times if multithreading
      is used.

  Returns:
    None, but performs an API call if necessary.
  """
  # Only perform this check if the user has opted in.
  if not config.getbool(
      'GSUtil', 'trigger_reauth_challenge_for_parallel_operations', False):
    return

  # Reauth is not necessary for non-cloud destinations.
  if not destination_url.IsCloudUrl():
    return

  # Destination wildcards are expanded by an API call in the main process.
  if ContainsWildcard(destination_url.url_string):
    return

  # If gsutil executes sequentially, all calls will occur in the main process.
  if worker_count == 1:
    return

  try:
    # The specific API call is not important, but one must occur.
    gsutil_api.GetBucket(
        destination_url.bucket_name,
        fields=['location'],  # Single field to limit XML API calls.
        provider=destination_url.scheme)
  except Exception as e:
    if isinstance(e, pyu2f.errors.PluginError):
      raise

    # Other exceptions can be ignored. The purpose was just to trigger
    # a reauth challenge.


def FixWindowsNaming(src_url, dst_url):
  """Translates Windows pathnames to cloud pathnames.

  Rewrites the destination URL built by ConstructDstUrl().

  Args:
    src_url: Source StorageUrl to be copied.
    dst_url: The destination StorageUrl built by ConstructDstUrl().

  Returns:
    StorageUrl to use for copy.
  """
  if (src_url.IsFileUrl() and src_url.delim == '\\' and dst_url.IsCloudUrl()):
    trans_url_str = re.sub(r'\\', '/', dst_url.url_string)
    dst_url = StorageUrlFromString(trans_url_str)
  return dst_url


def SrcDstSame(src_url, dst_url):
  """Checks if src_url and dst_url represent the same object or file.

  We don't handle anything about hard or symbolic links.

  Args:
    src_url: Source StorageUrl.
    dst_url: Destination StorageUrl.

  Returns:
    Bool indicator.
  """
  if src_url.IsFileUrl() and dst_url.IsFileUrl():
    # Translate a/b/./c to a/b/c, so src=dst comparison below works.
    new_src_path = os.path.normpath(src_url.object_name)
    new_dst_path = os.path.normpath(dst_url.object_name)
    return new_src_path == new_dst_path
  else:
    return (src_url.url_string == dst_url.url_string and
            src_url.generation == dst_url.generation)


def _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata):
  """Logs copy operation, including Content-Type if appropriate.

  Args:
    logger: logger instance to use for output.
    src_url: Source StorageUrl.
    dst_url: Destination StorageUrl.
    dst_obj_metadata: Object-specific metadata that should be overidden during
                      the copy.
  """
  if (dst_url.IsCloudUrl() and dst_obj_metadata and
      dst_obj_metadata.contentType):
    content_type_msg = ' [Content-Type=%s]' % dst_obj_metadata.contentType
  else:
    content_type_msg = ''
  if src_url.IsFileUrl() and (src_url.IsStream() or src_url.IsFifo()):
    src_text = '<STDIN>' if src_url.IsStream() else 'named pipe'
    logger.info('Copying from %s%s...', src_text, content_type_msg)
  else:
    logger.info('Copying %s%s...', src_url.url_string, content_type_msg)


# pylint: disable=undefined-variable
def _CopyObjToObjInTheCloud(src_url,
                            src_obj_metadata,
                            dst_url,
                            dst_obj_metadata,
                            preconditions,
                            gsutil_api,
                            decryption_key=None):
  """Performs copy-in-the cloud from specified src to dest object.

  Args:
    src_url: Source CloudUrl.
    src_obj_metadata: Metadata for source object; must include etag and size.
    dst_url: Destination CloudUrl.
    dst_obj_metadata: Object-specific metadata that should be overidden during
                      the copy.
    preconditions: Preconditions to use for the copy.
    gsutil_api: gsutil Cloud API instance to use for the copy.
    decryption_key: Base64-encoded decryption key for the source object, if any.

  Returns:
    (elapsed_time, bytes_transferred, dst_url with generation,
    md5 hash of destination) excluding overhead like initial GET.

  Raises:
    CommandException: if errors encountered.
  """
  decryption_keywrapper = CryptoKeyWrapperFromKey(decryption_key)
  encryption_keywrapper = GetEncryptionKeyWrapper(config)

  start_time = time.time()

  progress_callback = FileProgressCallbackHandler(gsutil_api.status_queue,
                                                  src_url=src_url,
                                                  dst_url=dst_url,
                                                  operation_name='Copying').call
  if global_copy_helper_opts.test_callback_file:
    with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
      progress_callback = pickle.loads(test_fp.read()).call
  dst_obj = gsutil_api.CopyObject(src_obj_metadata,
                                  dst_obj_metadata,
                                  src_generation=src_url.generation,
                                  canned_acl=global_copy_helper_opts.canned_acl,
                                  preconditions=preconditions,
                                  progress_callback=progress_callback,
                                  provider=dst_url.scheme,
                                  fields=UPLOAD_RETURN_FIELDS,
                                  decryption_tuple=decryption_keywrapper,
                                  encryption_tuple=encryption_keywrapper)

  end_time = time.time()

  result_url = dst_url.Clone()
  result_url.generation = GenerationFromUrlAndString(result_url,
                                                     dst_obj.generation)

  PutToQueueWithTimeout(
      gsutil_api.status_queue,
      FileMessage(src_url,
                  dst_url,
                  end_time,
                  message_type=FileMessage.FILE_CLOUD_COPY,
                  size=src_obj_metadata.size,
                  finished=True))

  return (end_time - start_time, src_obj_metadata.size, result_url,
          dst_obj.md5Hash)


def _SetContentTypeFromFile(src_url, dst_obj_metadata):
  """Detects and sets Content-Type if src_url names a local file.

  Args:
    src_url: Source StorageUrl.
    dst_obj_metadata: Object-specific metadata that should be overidden during
                     the copy.
  """
  # contentType == '' if user requested default type.
  if (dst_obj_metadata.contentType is None and src_url.IsFileUrl() and
      not src_url.IsStream() and not src_url.IsFifo()):
    # Only do content type recognition if src_url is a file. Object-to-object
    # copies with no -h Content-Type specified re-use the content type of the
    # source object.
    object_name = src_url.object_name
    content_type = None
    # Streams (denoted by '-') are expected to be 'application/octet-stream'
    # and 'file' would partially consume them.
    if object_name != '-':
      real_file_path = os.path.realpath(object_name)
      if config.getbool('GSUtil', 'use_magicfile', False) and not IS_WINDOWS:
        try:
          p = subprocess.Popen(['file', '-b', '--mime', real_file_path],
                               stdout=subprocess.PIPE,
                               stderr=subprocess.PIPE)
          output, error = p.communicate()
          p.stdout.close()
          p.stderr.close()
          if p.returncode != 0 or error:
            raise CommandException(
                'Encountered error running "file -b --mime %s" '
                '(returncode=%d).\n%s' % (real_file_path, p.returncode, error))
          # Parse output by removing line delimiter
          content_type = output.rstrip()
          content_type = six.ensure_str(content_type)
        except OSError as e:  # 'file' executable may not always be present.
          raise CommandException(
              'Encountered OSError running "file -b --mime %s"\n%s' %
              (real_file_path, e))
      else:
        _, _, extension = real_file_path.rpartition('.')
        if extension in COMMON_EXTENSION_RULES:
          content_type = COMMON_EXTENSION_RULES[extension]
        else:
          content_type, _ = mimetypes.guess_type(real_file_path)
    if not content_type:
      content_type = DEFAULT_CONTENT_TYPE
    dst_obj_metadata.contentType = content_type


# pylint: disable=undefined-variable
def _UploadFileToObjectNonResumable(src_url,
                                    src_obj_filestream,
                                    src_obj_size,
                                    dst_url,
                                    dst_obj_metadata,
                                    preconditions,
                                    gsutil_api,
                                    gzip_encoded=False):
  """Uploads the file using a non-resumable strategy.

  This function does not support component transfers.

  Args:
    src_url: Source StorageUrl to upload.
    src_obj_filestream: File pointer to uploadable bytes.
    src_obj_size (int or None): Size of the source object.
    dst_url: Destination StorageUrl for the upload.
    dst_obj_metadata: Metadata for the target object.
    preconditions: Preconditions for the upload, if any.
    gsutil_api: gsutil Cloud API instance to use for the upload.
    gzip_encoded: Whether to use gzip transport encoding for the upload.

  Returns:
    Elapsed upload time, uploaded Object with generation, md5, and size fields
    populated.
  """
  progress_callback = FileProgressCallbackHandler(
      gsutil_api.status_queue,
      src_url=src_url,
      dst_url=dst_url,
      operation_name='Uploading').call

  if global_copy_helper_opts.test_callback_file:
    with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
      progress_callback = pickle.loads(test_fp.read()).call
  start_time = time.time()

  encryption_keywrapper = GetEncryptionKeyWrapper(config)

  if src_url.IsStream() or src_url.IsFifo():
    # TODO: gsutil-beta: Provide progress callbacks for streaming uploads.
    uploaded_object = gsutil_api.UploadObjectStreaming(
        src_obj_filestream,
        object_metadata=dst_obj_metadata,
        canned_acl=global_copy_helper_opts.canned_acl,
        preconditions=preconditions,
        progress_callback=progress_callback,
        encryption_tuple=encryption_keywrapper,
        provider=dst_url.scheme,
        fields=UPLOAD_RETURN_FIELDS,
        gzip_encoded=gzip_encoded)
  else:
    uploaded_object = gsutil_api.UploadObject(
        src_obj_filestream,
        object_metadata=dst_obj_metadata,
        canned_acl=global_copy_helper_opts.canned_acl,
        size=src_obj_size,
        preconditions=preconditions,
        progress_callback=progress_callback,
        encryption_tuple=encryption_keywrapper,
        provider=dst_url.scheme,
        fields=UPLOAD_RETURN_FIELDS,
        gzip_encoded=gzip_encoded)
  end_time = time.time()
  elapsed_time = end_time - start_time

  return elapsed_time, uploaded_object


# pylint: disable=undefined-variable
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):
  """Uploads the file using a resumable strategy.

  Args:
    src_url: Source FileUrl to upload.  Must not be a stream.
    src_obj_filestream: File pointer to uploadable bytes.
    src_obj_size (int or None): Size of the source object.
    dst_url: Destination StorageUrl for the upload.
    dst_obj_metadata: Metadata for the target object.
    preconditions: Preconditions for the upload, if any.
    gsutil_api: gsutil Cloud API instance to use for the upload.
    logger: for outputting log messages.
    is_component: indicates whether this is a single component or whole file.
    gzip_encoded: Whether to use gzip transport encoding for the upload.

  Returns:
    Elapsed upload time, uploaded Object with generation, md5, and size fields
    populated.
  """
  tracker_file_name = GetTrackerFilePath(
      dst_url, TrackerFileType.UPLOAD,
      gsutil_api.GetApiSelector(provider=dst_url.scheme))

  encryption_keywrapper = GetEncryptionKeyWrapper(config)
  encryption_key_sha256 = (
      encryption_keywrapper.crypto_key_sha256.decode('ascii')
      if encryption_keywrapper and encryption_keywrapper.crypto_key_sha256 else
      None)

  def _UploadTrackerCallback(serialization_data):
    """Creates a new tracker file for starting an upload from scratch.

    This function is called by the gsutil Cloud API implementation and the
    the serialization data is implementation-specific.

    Args:
      serialization_data: Serialization data used in resuming the upload.
    """
    data = {
        ENCRYPTION_UPLOAD_TRACKER_ENTRY: encryption_key_sha256,
        SERIALIZATION_UPLOAD_TRACKER_ENTRY: str(serialization_data)
    }
    WriteJsonDataToTrackerFile(tracker_file_name, data)

  # This contains the upload URL, which will uniquely identify the
  # destination object.
  tracker_data = GetUploadTrackerData(
      tracker_file_name, logger, encryption_key_sha256=encryption_key_sha256)
  if tracker_data:
    logger.info('Resuming upload for %s', src_url.url_string)

  retryable = True

  component_num = _GetComponentNumber(dst_url) if is_component else None
  progress_callback = FileProgressCallbackHandler(
      gsutil_api.status_queue,
      src_url=src_url,
      component_num=component_num,
      dst_url=dst_url,
      operation_name='Uploading').call

  if global_copy_helper_opts.test_callback_file:
    with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
      progress_callback = pickle.loads(test_fp.read()).call

  start_time = time.time()
  num_startover_attempts = 0
  # This loop causes us to retry when the resumable upload failed in a way that
  # requires starting over with a new upload ID. Retries within a single upload
  # ID within the current process are handled in
  # gsutil_api.UploadObjectResumable, and retries within a single upload ID
  # spanning processes happens if an exception occurs not caught below (which
  # will leave the tracker file in place, and cause the upload ID to be reused
  # the next time the user runs gsutil and attempts the same upload).
  while retryable:
    try:
      uploaded_object = gsutil_api.UploadObjectResumable(
          src_obj_filestream,
          object_metadata=dst_obj_metadata,
          canned_acl=global_copy_helper_opts.canned_acl,
          preconditions=preconditions,
          provider=dst_url.scheme,
          size=src_obj_size,
          serialization_data=tracker_data,
          encryption_tuple=encryption_keywrapper,
          fields=UPLOAD_RETURN_FIELDS,
          tracker_callback=_UploadTrackerCallback,
          progress_callback=progress_callback,
          gzip_encoded=gzip_encoded)
      retryable = False
    except ResumableUploadStartOverException as e:
      logger.info('Caught ResumableUploadStartOverException for upload of %s.' %
                  src_url.url_string)
      # This can happen, for example, if the server sends a 410 response code.
      # In that case the current resumable upload ID can't be reused, so delete
      # the tracker file and try again up to max retries.
      num_startover_attempts += 1
      retryable = (num_startover_attempts < GetNumRetries())
      if not retryable:
        raise

      # If the server sends a 404 response code, then the upload should only
      # be restarted if it was the object (and not the bucket) that was missing.
      try:
        logger.info('Checking that bucket %s exists before retrying upload...' %
                    dst_obj_metadata.bucket)
        gsutil_api.GetBucket(dst_obj_metadata.bucket, provider=dst_url.scheme)
      except AccessDeniedException:
        # Proceed with deleting the tracker file in the event that the bucket
        # exists, but the user does not have permission to view its metadata.
        pass
      except NotFoundException:
        raise
      finally:
        DeleteTrackerFile(tracker_file_name)
        logger.info('Deleted tracker file %s for resumable upload of %s before '
                    'retrying.' % (tracker_file_name, src_url.url_string))

      logger.info(
          'Restarting upload of %s from scratch (retry #%d) after exception '
          'indicating we need to start over with a new resumable upload ID: %s'
          % (src_url.url_string, num_startover_attempts, e))
      tracker_data = None
      src_obj_filestream.seek(0)
      # Reset the progress callback handler.
      component_num = _GetComponentNumber(dst_url) if is_component else None
      progress_callback = FileProgressCallbackHandler(
          gsutil_api.status_queue,
          src_url=src_url,
          component_num=component_num,
          dst_url=dst_url,
          operation_name='Uploading').call

      # Report the retryable error to the global status queue.
      PutToQueueWithTimeout(
          gsutil_api.status_queue,
          RetryableErrorMessage(e,
                                time.time(),
                                num_retries=num_startover_attempts))
      time.sleep(
          min(random.random() * (2**num_startover_attempts),
              GetMaxRetryDelay()))
    except ResumableUploadAbortException:
      retryable = False
      raise
    finally:
      if not retryable:
        DeleteTrackerFile(tracker_file_name)

  end_time = time.time()
  elapsed_time = end_time - start_time

  return (elapsed_time, uploaded_object)


def _SelectUploadCompressionStrategy(object_name,
                                     is_component=False,
                                     gzip_exts=False,
                                     gzip_encoded=False):
  """Selects how an upload should be compressed.

  This is a helper function for _UploadFileToObject.

  Args:
    object_name: The object name of the source FileUrl.
    is_component: indicates whether this is a single component or whole file.
    gzip_exts: List of file extensions to gzip prior to upload, if any.
               If gzip_exts is GZIP_ALL_FILES, gzip all files.
    gzip_encoded: Whether to use gzip transport encoding for the upload. Used
        in conjunction with gzip_exts for selecting which files will be
        encoded. Streaming files compressed is only supported on the JSON GCS
        API.

  Returns:
    A tuple: (If the file should be gzipped locally, if the file should be gzip
    transport encoded).
  """
  zipped_file = False
  gzip_encoded_file = False
  fname_parts = object_name.split('.')

  # If gzip_encoded and is_component are marked as true, the file was already
  # filtered through the original gzip_exts filter and we must compress the
  # component via gzip transport encoding.
  if gzip_encoded and is_component:
    gzip_encoded_file = True
  elif (gzip_exts == GZIP_ALL_FILES or
        (gzip_exts and len(fname_parts) > 1 and fname_parts[-1] in gzip_exts)):
    zipped_file = not gzip_encoded
    gzip_encoded_file = gzip_encoded

  return zipped_file, gzip_encoded_file


def _ApplyZippedUploadCompression(src_url, src_obj_filestream, src_obj_size,
                                  logger):
  """Compresses a to-be-uploaded local file to save bandwidth.

  This is a helper function for _UploadFileToObject.

  Args:
    src_url: Source FileUrl.
    src_obj_filestream: Read stream of the source file - will be consumed
      and closed.
    src_obj_size (int or None): Size of the source file.
    logger: for outputting log messages.

  Returns:
    StorageUrl path to compressed file, read stream of the compressed file,
    compressed file size.
  """
  # TODO: Compress using a streaming model as opposed to all at once here.
  if src_obj_size is not None and src_obj_size >= MIN_SIZE_COMPUTE_LOGGING:
    logger.info('Compressing %s (to tmp)...', src_url)
  (gzip_fh, gzip_path) = tempfile.mkstemp()
  gzip_fp = None
  try:
    # Check for temp space. Assume the compressed object is at most 2x
    # the size of the object (normally should compress to smaller than
    # the object)
    if src_url.IsStream() or src_url.IsFifo():
      # TODO: Support streaming gzip uploads.
      # https://github.com/GoogleCloudPlatform/gsutil/issues/364
      raise CommandException(
          'gzip compression is not currently supported on streaming uploads. '
          'Remove the compression flag or save the streamed output '
          'temporarily to a file before uploading.')
    if src_obj_size is not None and (CheckFreeSpace(gzip_path)
                                     < 2 * int(src_obj_size)):
      raise CommandException('Inadequate temp space available to compress '
                             '%s. See the CHANGING TEMP DIRECTORIES section '
                             'of "gsutil help cp" for more info.' % src_url)
    compression_level = config.getint('GSUtil', 'gzip_compression_level',
                                      DEFAULT_GZIP_COMPRESSION_LEVEL)
    gzip_fp = gzip.open(gzip_path, 'wb', compresslevel=compression_level)
    data = src_obj_filestream.read(GZIP_CHUNK_SIZE)
    while data:
      gzip_fp.write(data)
      data = src_obj_filestream.read(GZIP_CHUNK_SIZE)
  finally:
    if gzip_fp:
      gzip_fp.close()
    os.close(gzip_fh)
    src_obj_filestream.close()
  gzip_size = os.path.getsize(gzip_path)
  compressed_filestream = open(gzip_path, 'rb')
  return StorageUrlFromString(gzip_path), compressed_filestream, gzip_size


def _DelegateUploadFileToObject(upload_delegate, upload_url, upload_stream,
                                zipped_file, gzip_encoded_file,
                                parallel_composite_upload, logger):
  """Handles setup and tear down logic for uploads.

  This is a helper function for _UploadFileToObject.

  Args:
    upload_delegate: Function that handles uploading the file.
    upload_url: StorageURL path to the file.
    upload_stream: Read stream of the file being uploaded. This will be closed
      after the upload.
    zipped_file: Flag for if the file is locally compressed prior to calling
      this function. If true, the local temporary file is deleted after the
      upload.
    gzip_encoded_file: Flag for if the file will be uploaded with the gzip
      transport encoding. If true, a lock is used to limit resource usage.
    parallel_composite_upload: Set to true if this upload represents a
      top-level parallel composite upload (not an upload of a component). If
      true, resource locking is skipped.
    logger: For outputting log messages.

  Returns:
    The elapsed upload time, the uploaded object.
  """
  elapsed_time = None
  uploaded_object = None
  try:
    # Parallel transport compressed uploads use a signifcant amount of memory.
    # The number of threads that may run concurrently are restricted as a
    # result. Parallel composite upload's don't actually upload data, but
    # instead fork for each component and calling _UploadFileToObject
    # individually. The parallel_composite_upload flag is false for the actual
    # upload invocation.
    if gzip_encoded_file and not parallel_composite_upload:
      with gslib.command.concurrent_compressed_upload_lock:
        elapsed_time, uploaded_object = upload_delegate()
    else:
      elapsed_time, uploaded_object = upload_delegate()

  finally:
    # In the zipped_file case, this is the gzip stream. When the gzip stream is
    # created, the original source stream is closed in
    # _ApplyZippedUploadCompression. This means that we do not have to
    # explicitly close the source stream here in the zipped_file case.
    upload_stream.close()
    if zipped_file:
      try:
        os.unlink(upload_url.object_name)
      # Windows sometimes complains the temp file is locked when you try to
      # delete it.
      except Exception:  # pylint: disable=broad-except
        logger.warning(
            'Could not delete %s. This can occur in Windows because the '
            'temporary file is still locked.', upload_url.object_name)
  return elapsed_time, uploaded_object


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):
  """Uploads a local file to an object.

  Args:
    src_url: Source FileUrl.
    src_obj_filestream: Read stream of the source file to be read and closed.
    src_obj_size (int or None): Size of the source file.
    dst_url: Destination CloudUrl.
    dst_obj_metadata: Metadata to be applied to the destination object.
    preconditions: Preconditions to use for the copy.
    gsutil_api: gsutil Cloud API to use for the copy.
    logger: for outputting log messages.
    command_obj: command object for use in Apply in parallel composite uploads.
    copy_exception_handler: For handling copy exceptions during Apply.
    gzip_exts: List of file extensions to gzip prior to upload, if any.
               If gzip_exts is GZIP_ALL_FILES, gzip all files.
    allow_splitting: Whether to allow the file to be split into component
                     pieces for an parallel composite upload.
    is_component: indicates whether this is a single component or whole file.
    gzip_encoded: Whether to use gzip transport encoding for the upload. Used
        in conjunction with gzip_exts for selecting which files will be
        encoded. Streaming files compressed is only supported on the JSON GCS
        API.

  Returns:
    (elapsed_time, bytes_transferred, dst_url with generation,
    md5 hash of destination) excluding overhead like initial GET.

  Raises:
    CommandException: if errors encountered.
  """
  if not dst_obj_metadata or not dst_obj_metadata.contentLanguage:
    content_language = config.get_value('GSUtil', 'content_language')
    if content_language:
      dst_obj_metadata.contentLanguage = content_language

  upload_url = src_url
  upload_stream = src_obj_filestream
  upload_size = src_obj_size

  zipped_file, gzip_encoded_file = _SelectUploadCompressionStrategy(
      src_url.object_name, is_component, gzip_exts, gzip_encoded)
  # The component's parent already printed this debug message.
  if gzip_encoded_file and not is_component:
    logger.debug('Using compressed transport encoding for %s.', src_url)
  elif zipped_file:
    upload_url, upload_stream, upload_size = _ApplyZippedUploadCompression(
        src_url, src_obj_filestream, src_obj_size, logger)
    dst_obj_metadata.contentEncoding = 'gzip'
    # If we're sending an object with gzip encoding, it's possible it also
    # has an incompressible content type. Google Cloud Storage will remove
    # the top layer of compression when serving the object, which would cause
    # the served content not to match the CRC32C/MD5 hashes stored and make
    # integrity checking impossible. Therefore we set cache control to
    # no-transform to ensure it is served in its original form. The caveat is
    # that to read this object, other clients must then support
    # accept-encoding:gzip.
    if not dst_obj_metadata.cacheControl:
      dst_obj_metadata.cacheControl = 'no-transform'
    elif 'no-transform' not in dst_obj_metadata.cacheControl.lower():
      dst_obj_metadata.cacheControl += ',no-transform'

  if not is_component:
    PutToQueueWithTimeout(
        gsutil_api.status_queue,
        FileMessage(upload_url,
                    dst_url,
                    time.time(),
                    message_type=FileMessage.FILE_UPLOAD,
                    size=upload_size,
                    finished=False))
  elapsed_time = None
  uploaded_object = None
  hash_algs = GetUploadHashAlgs()
  digesters = dict((alg, hash_algs[alg]()) for alg in hash_algs or {})

  parallel_composite_upload = _ShouldDoParallelCompositeUpload(
      logger,
      allow_splitting,
      upload_url,
      dst_url,
      src_obj_size,
      gsutil_api,
      canned_acl=global_copy_helper_opts.canned_acl)
  non_resumable_upload = ((0 if upload_size is None else upload_size)
                          < ResumableThreshold() or src_url.IsStream() or
                          src_url.IsFifo())

  if ((src_url.IsStream() or src_url.IsFifo()) and
      gsutil_api.GetApiSelector(provider=dst_url.scheme) == ApiSelector.JSON):
    orig_stream = upload_stream
    # Add limited seekable properties to the stream via buffering.
    upload_stream = ResumableStreamingJsonUploadWrapper(
        orig_stream, GetJsonResumableChunkSize())

  if not parallel_composite_upload and len(hash_algs):
    # Parallel composite uploads calculate hashes per-component in subsequent
    # calls to this function, but the composition of the final object is a
    # cloud-only operation.
    wrapped_filestream = HashingFileUploadWrapper(upload_stream, digesters,
                                                  hash_algs, upload_url, logger)
  else:
    wrapped_filestream = upload_stream

  def CallParallelCompositeUpload():
    return _DoParallelCompositeUpload(upload_stream,
                                      upload_url,
                                      dst_url,
                                      dst_obj_metadata,
                                      global_copy_helper_opts.canned_acl,
                                      upload_size,
                                      preconditions,
                                      gsutil_api,
                                      command_obj,
                                      copy_exception_handler,
                                      logger,
                                      gzip_encoded=gzip_encoded_file)

  def CallNonResumableUpload():
    return _UploadFileToObjectNonResumable(upload_url,
                                           wrapped_filestream,
                                           upload_size,
                                           dst_url,
                                           dst_obj_metadata,
                                           preconditions,
                                           gsutil_api,
                                           gzip_encoded=gzip_encoded_file)

  def CallResumableUpload():
    return _UploadFileToObjectResumable(upload_url,
                                        wrapped_filestream,
                                        upload_size,
                                        dst_url,
                                        dst_obj_metadata,
                                        preconditions,
                                        gsutil_api,
                                        logger,
                                        is_component=is_component,
                                        gzip_encoded=gzip_encoded_file)

  if parallel_composite_upload:
    delegate = CallParallelCompositeUpload
  elif non_resumable_upload:
    delegate = CallNonResumableUpload
  else:
    delegate = CallResumableUpload

  elapsed_time, uploaded_object = _DelegateUploadFileToObject(
      delegate, upload_url, upload_stream, zipped_file, gzip_encoded_file,
      parallel_composite_upload, logger)

  if not parallel_composite_upload:
    try:
      digests = _CreateDigestsFromDigesters(digesters)
      _CheckHashes(logger,
                   dst_url,
                   uploaded_object,
                   src_url.object_name,
                   digests,
                   is_upload=True)
    except HashMismatchException:
      if _RENAME_ON_HASH_MISMATCH:
        corrupted_obj_metadata = apitools_messages.Object(
            name=dst_obj_metadata.name,
            bucket=dst_obj_metadata.bucket,
            etag=uploaded_object.etag)
        dst_obj_metadata.name = (dst_url.object_name +
                                 _RENAME_ON_HASH_MISMATCH_SUFFIX)
        gsutil_api.CopyObject(corrupted_obj_metadata,
                              dst_obj_metadata,
                              provider=dst_url.scheme)
      # If the digest doesn't match, delete the object.
      gsutil_api.DeleteObject(dst_url.bucket_name,
                              dst_url.object_name,
                              generation=uploaded_object.generation,
                              provider=dst_url.scheme)
      raise

  result_url = dst_url.Clone()

  result_url.generation = uploaded_object.generation
  result_url.generation = GenerationFromUrlAndString(result_url,
                                                     uploaded_object.generation)

  if not is_component:
    PutToQueueWithTimeout(
        gsutil_api.status_queue,
        FileMessage(upload_url,
                    dst_url,
                    time.time(),
                    message_type=FileMessage.FILE_UPLOAD,
                    size=upload_size,
                    finished=True))

  return (elapsed_time, uploaded_object.size, result_url,
          uploaded_object.md5Hash)


def _GetDownloadFile(dst_url, src_obj_metadata, logger):
  """Creates a new download file, and deletes the file that will be replaced.

  Names and creates a temporary file for this download. Also, if there is an
  existing file at the path where this file will be placed after the download
  is completed, that file will be deleted.

  Args:
    dst_url: Destination FileUrl.
    src_obj_metadata: Metadata from the source object.
    logger: for outputting log messages.

  Returns:
    (download_file_name, need_to_unzip)
    download_file_name: The name of the temporary file to which the object will
                        be downloaded.
    need_to_unzip: If true, a temporary zip file was used and must be
                   uncompressed as part of validation.
  """
  dir_name = os.path.dirname(dst_url.object_name)
  if dir_name and not os.path.exists(dir_name):
    # Do dir creation in try block so can ignore case where dir already
    # exists. This is needed to avoid a race condition when running gsutil
    # -m cp.
    try:
      os.makedirs(dir_name)
    except OSError as e:
      if e.errno != errno.EEXIST:
        raise

  need_to_unzip = False
  # For gzipped objects download to a temp file and unzip. For the XML API,
  # this represents the result of a HEAD request. For the JSON API, this is
  # the stored encoding which the service may not respect. However, if the
  # server sends decompressed bytes for a file that is stored compressed
  # (double compressed case), there is no way we can validate the hash and
  # we will fail our hash check for the object.
  if ObjectIsGzipEncoded(src_obj_metadata):
    need_to_unzip = True
    download_file_name = temporary_file_util.GetTempZipFileName(dst_url)
    logger.info('Downloading to temp gzip filename %s', download_file_name)
  else:
    download_file_name = temporary_file_util.GetTempFileName(dst_url)

  # If a file exists at the permanent destination (where the file will be moved
  # after the download is completed), delete it here to reduce disk space
  # requirements.
  if os.path.exists(dst_url.object_name):
    os.unlink(dst_url.object_name)

  # Downloads open the temporary download file in r+b mode, which requires it
  # to already exist, so we create it here if it doesn't exist already.
  if not os.path.exists(download_file_name):
    fp = open(download_file_name, 'w')
    fp.close()
  return download_file_name, need_to_unzip


def _ShouldDoSlicedDownload(download_strategy, src_obj_metadata,
                            allow_splitting, logger):
  """Determines whether the sliced download strategy should be used.

  Args:
    download_strategy: CloudApi download strategy.
    src_obj_metadata: Metadata from the source object.
    allow_splitting: If false, then this function returns false.
    logger: logging.Logger for log message output.

  Returns:
    True iff a sliced download should be performed on the source file.
  """
  sliced_object_download_threshold = HumanReadableToBytes(
      config.get('GSUtil', 'sliced_object_download_threshold',
                 DEFAULT_SLICED_OBJECT_DOWNLOAD_THRESHOLD))

  max_components = config.getint('GSUtil',
                                 'sliced_object_download_max_components',
                                 DEFAULT_SLICED_OBJECT_DOWNLOAD_MAX_COMPONENTS)

  # Don't use sliced download if it will prevent us from performing an
  # integrity check.
  check_hashes_config = config.get('GSUtil', 'check_hashes',
                                   CHECK_HASH_IF_FAST_ELSE_FAIL)
  parallel_hashing = src_obj_metadata.crc32c and UsingCrcmodExtension()
  hashing_okay = parallel_hashing or check_hashes_config == CHECK_HASH_NEVER

  use_slice = (allow_splitting and
               download_strategy is not CloudApi.DownloadStrategy.ONE_SHOT and
               max_components > 1 and hashing_okay and
               sliced_object_download_threshold > 0 and
               src_obj_metadata.size >= sliced_object_download_threshold)

  if (not use_slice and
      src_obj_metadata.size >= PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD and
      not UsingCrcmodExtension() and check_hashes_config != CHECK_HASH_NEVER):
    with suggested_sliced_transfers_lock:
      if not suggested_sliced_transfers.get('suggested'):
        logger.info('\n'.join(
            textwrap.wrap(
                '==> NOTE: You are downloading one or more large file(s), which '
                'would run significantly faster if you enabled sliced object '
                'downloads. This feature is enabled by default but requires that '
                'compiled crcmod be installed (see "gsutil help crcmod").')) +
                    '\n')
        suggested_sliced_transfers['suggested'] = True

  return use_slice


def _PerformSlicedDownloadObjectToFile(cls, args, thread_state=None):
  """Function argument to Apply for performing sliced downloads.

  Args:
    cls: Calling Command class.
    args: PerformSlicedDownloadObjectToFileArgs tuple describing the target.
    thread_state: gsutil Cloud API instance to use for the operation.

  Returns:
    PerformSlicedDownloadReturnValues named-tuple filled with:
    component_num: The component number for this download.
    crc32c: CRC32C hash value (integer) of the downloaded bytes.
    bytes_transferred: The number of bytes transferred, potentially less
                       than the component size if the download was resumed.
    component_total_size: The number of bytes corresponding to the whole
                       component size, potentially more than bytes_transferred
                       if the download was resumed.
  """
  gsutil_api = GetCloudApiInstance(cls, thread_state=thread_state)
  # Deserialize the picklable object metadata.
  src_obj_metadata = protojson.decode_message(apitools_messages.Object,
                                              args.src_obj_metadata_json)
  hash_algs = GetDownloadHashAlgs(cls.logger,
                                  consider_crc32c=src_obj_metadata.crc32c)
  digesters = dict((alg, hash_algs[alg]()) for alg in hash_algs or {})

  (bytes_transferred, server_encoding) = (_DownloadObjectToFileResumable(
      args.src_url,
      src_obj_metadata,
      args.dst_url,
      args.download_file_name,
      gsutil_api,
      cls.logger,
      digesters,
      component_num=args.component_num,
      start_byte=args.start_byte,
      end_byte=args.end_byte,
      decryption_key=args.decryption_key))

  crc32c_val = None
  if 'crc32c' in digesters:
    crc32c_val = digesters['crc32c'].crcValue
  return PerformSlicedDownloadReturnValues(args.component_num, crc32c_val,
                                           bytes_transferred,
                                           args.end_byte - args.start_byte + 1,
                                           server_encoding)


def _MaintainSlicedDownloadTrackerFiles(src_obj_metadata, dst_url,
                                        download_file_name, logger,
                                        api_selector, num_components):
  """Maintains sliced download tracker files in order to permit resumability.

  Reads or creates a sliced download tracker file representing this object
  download. Upon an attempt at cross-process resumption, the contents of the
  sliced download tracker file are verified to make sure a resumption is
  possible and appropriate. In the case that a resumption should not be
  attempted, existing component tracker files are deleted (to prevent child
  processes from attempting resumption), and a new sliced download tracker
  file is created.

  Args:
    src_obj_metadata: Metadata from the source object. Must include etag and
                      generation.
    dst_url: Destination FileUrl.
    download_file_name: Temporary file name to be used for the download.
    logger: for outputting log messages.
    api_selector: The Cloud API implementation used.
    num_components: The number of components to perform this download with.
  """
  assert src_obj_metadata.etag
  tracker_file = None

  # Only can happen if the resumable threshold is set higher than the
  # parallel transfer threshold.
  if src_obj_metadata.size < ResumableThreshold():
    return

  tracker_file_name = GetTrackerFilePath(dst_url,
                                         TrackerFileType.SLICED_DOWNLOAD,
                                         api_selector)

  fp = None
  # Check to see if we should attempt resuming the download.
  try:
    fp = open(download_file_name, 'rb')
    existing_file_size = GetFileSize(fp)
    # A parallel resumption should be attempted only if the destination file
    # size is exactly the same as the source size and the tracker file matches.
    if existing_file_size == src_obj_metadata.size:
      tracker_file = open(tracker_file_name, 'r')
      tracker_file_data = json.load(tracker_file)
      if (tracker_file_data['etag'] == src_obj_metadata.etag and
          tracker_file_data['generation'] == src_obj_metadata.generation and
          tracker_file_data['num_components'] == num_components):
        return
      else:
        tracker_file.close()
        logger.warn('Sliced download tracker file doesn\'t match for '
                    'download of %s. Restarting download from scratch.' %
                    dst_url.object_name)

  except (IOError, ValueError) as e:
    # Ignore non-existent file (happens first time a download
    # is attempted on an object), but warn user for other errors.
    if isinstance(e, ValueError) or e.errno != errno.ENOENT:
      logger.warn('Couldn\'t read sliced download tracker file (%s): %s. '
                  'Restarting download from scratch.' %
                  (tracker_file_name, str(e)))
  finally:
    if fp:
      fp.close()
    if tracker_file:
      tracker_file.close()

  # Delete component tracker files to guarantee download starts from scratch.
  DeleteDownloadTrackerFiles(dst_url, api_selector)

  # Create a new sliced download tracker file to represent this download.
  try:
    with open(tracker_file_name, 'w') as tracker_file:
      tracker_file_data = {
          'etag': src_obj_metadata.etag,
          'generation': src_obj_metadata.generation,
          'num_components': num_components,
      }
      tracker_file.write(json.dumps(tracker_file_data))
  except IOError as e:
    RaiseUnwritableTrackerFileException(tracker_file_name, e.strerror)


class SlicedDownloadFileWrapper(object):
  """Wraps a file object to be used in GetObjectMedia for sliced downloads.

  In order to allow resumability, the file object used by each thread in a
  sliced object download should be wrapped using SlicedDownloadFileWrapper.
  Passing a SlicedDownloadFileWrapper object to GetObjectMedia will allow the
  download component tracker file for this component to be updated periodically,
  while the downloaded bytes are normally written to file.
  """

  def __init__(self, fp, tracker_file_name, src_obj_metadata, start_byte,
               end_byte):
    """Initializes the SlicedDownloadFileWrapper.

    Args:
      fp: The already-open file object to be used for writing in
          GetObjectMedia. Data will be written to file starting at the current
          seek position.
      tracker_file_name: The name of the tracker file for this component.
      src_obj_metadata: Metadata from the source object. Must include etag and
                        generation.
      start_byte: The first byte to be downloaded for this parallel component.
      end_byte: The last byte to be downloaded for this parallel component.
    """
    self._orig_fp = fp
    self._tracker_file_name = tracker_file_name
    self._src_obj_metadata = src_obj_metadata
    self._last_tracker_file_byte = None
    self._start_byte = start_byte
    self._end_byte = end_byte

  @property
  def mode(self):
    """Returns the mode of the underlying file descriptor, or None."""
    return getattr(self._orig_fp, 'mode', None)

  def write(self, data):  # pylint: disable=invalid-name
    current_file_pos = self._orig_fp.tell()
    assert (self._start_byte <= current_file_pos and
            current_file_pos + len(data) <= self._end_byte + 1)

    text_util.write_to_fd(self._orig_fp, data)
    current_file_pos = self._orig_fp.tell()

    threshold = TRACKERFILE_UPDATE_THRESHOLD
    if (self._last_tracker_file_byte is None or
        current_file_pos - self._last_tracker_file_byte > threshold or
        current_file_pos == self._end_byte + 1):
      WriteDownloadComponentTrackerFile(self._tracker_file_name,
                                        self._src_obj_metadata,
                                        current_file_pos)
      self._last_tracker_file_byte = current_file_pos

  def seek(self, offset, whence=os.SEEK_SET):  # pylint: disable=invalid-name
    if whence == os.SEEK_END:
      self._orig_fp.seek(offset + self._end_byte + 1)
    else:
      self._orig_fp.seek(offset, whence)
    assert self._start_byte <= self._orig_fp.tell() <= self._end_byte + 1

  def tell(self):  # pylint: disable=invalid-name
    return self._orig_fp.tell()

  def flush(self):  # pylint: disable=invalid-name
    self._orig_fp.flush()

  def close(self):  # pylint: disable=invalid-name
    if self._orig_fp:
      self._orig_fp.close()


def _PartitionObject(src_url,
                     src_obj_metadata,
                     dst_url,
                     download_file_name,
                     decryption_key=None):
  """Partitions an object into components to be downloaded.

  Each component is a byte range of the object. The byte ranges
  of the returned components are mutually exclusive and collectively
  exhaustive. The byte ranges are inclusive at both end points.

  Args:
    src_url: Source CloudUrl.
    src_obj_metadata: Metadata from the source object with non-pickleable
        fields removed.
    dst_url: Destination FileUrl.
    download_file_name: Temporary file name to be used for the download.
    decryption_key: Base64-encoded decryption key for the source object, if any.

  Returns:
    components_to_download: A list of PerformSlicedDownloadObjectToFileArgs
                            to be used in Apply for the sliced download.
  """
  sliced_download_component_size = HumanReadableToBytes(
      config.get('GSUtil', 'sliced_object_download_component_size',
                 DEFAULT_SLICED_OBJECT_DOWNLOAD_COMPONENT_SIZE))

  max_components = config.getint('GSUtil',
                                 'sliced_object_download_max_components',
                                 DEFAULT_SLICED_OBJECT_DOWNLOAD_MAX_COMPONENTS)

  num_components, component_size = _GetPartitionInfo(
      src_obj_metadata.size, max_components, sliced_download_component_size)

  components_to_download = []
  component_lengths = []
  for i in range(num_components):
    start_byte = i * component_size
    end_byte = min((i + 1) * (component_size) - 1, src_obj_metadata.size - 1)
    component_lengths.append(end_byte - start_byte + 1)

    # We need to serialize src_obj_metadata for pickling since it can
    # contain nested classes such as custom metadata.
    src_obj_metadata_json = protojson.encode_message(src_obj_metadata)

    components_to_download.append(
        PerformSlicedDownloadObjectToFileArgs(i, src_url, src_obj_metadata_json,
                                              dst_url, download_file_name,
                                              start_byte, end_byte,
                                              decryption_key))
  return components_to_download, component_lengths


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):
  """Downloads a cloud object to a local file using sliced download.

  Byte ranges are decided for each thread/process, and then the parts are
  downloaded in parallel.

  Args:
    src_url: Source CloudUrl.
    src_obj_metadata: Metadata from the source object.
    dst_url: Destination FileUrl.
    download_file_name: Temporary file name to be used for download.
    command_obj: command object for use in Apply in parallel composite uploads.
    logger: for outputting log messages.
    copy_exception_handler: For handling copy exceptions during Apply.
    api_selector: The Cloud API implementation used.
    decryption_key: Base64-encoded decryption key for the source object, if any.
    status_queue: Queue for posting file messages for UI/Analytics.

  Returns:
    (bytes_transferred, crc32c)
    bytes_transferred: Number of bytes transferred from server this call.
    crc32c: a crc32c hash value (integer) for the downloaded bytes, or None if
            crc32c hashing wasn't performed.
  """
  # CustomerEncryptionValue is a subclass and thus not pickleable for
  # multiprocessing, but at this point we already have the matching key,
  # so just discard the metadata.
  src_obj_metadata.customerEncryption = None

  components_to_download, component_lengths = _PartitionObject(
      src_url, src_obj_metadata, dst_url, download_file_name, decryption_key)

  num_components = len(components_to_download)
  _MaintainSlicedDownloadTrackerFiles(src_obj_metadata, dst_url,
                                      download_file_name, logger, api_selector,
                                      num_components)

  # Resize the download file so each child process can seek to its start byte.
  with open(download_file_name, 'r+b') as fp:
    fp.truncate(src_obj_metadata.size)
  # Assign a start FileMessage to each component
  for (i, component) in enumerate(components_to_download):
    size = component.end_byte - component.start_byte + 1

    download_start_byte = GetDownloadStartByte(src_obj_metadata, dst_url,
                                               api_selector,
                                               component.start_byte, size, i)
    bytes_already_downloaded = download_start_byte - component.start_byte
    PutToQueueWithTimeout(
        status_queue,
        FileMessage(src_url,
                    dst_url,
                    time.time(),
                    size=size,
                    finished=False,
                    component_num=i,
                    message_type=FileMessage.COMPONENT_TO_DOWNLOAD,
                    bytes_already_downloaded=bytes_already_downloaded))

  cp_results = command_obj.Apply(
      _PerformSlicedDownloadObjectToFile,
      components_to_download,
      copy_exception_handler,
      arg_checker=gslib.command.DummyArgChecker,
      parallel_operations_override=command_obj.ParallelOverrideReason.SLICE,
      should_return_results=True)

  if len(cp_results) < num_components:
    raise CommandException(
        'Some components of %s were not downloaded successfully. '
        'Please retry this download.' % dst_url.object_name)

  # Crc32c hashes have to be concatenated in the correct order.
  cp_results = sorted(cp_results, key=attrgetter('component_num'))
  crc32c = cp_results[0].crc32c
  if crc32c is not None:
    for i in range(1, num_components):
      crc32c = ConcatCrc32c(crc32c, cp_results[i].crc32c, component_lengths[i])

  bytes_transferred = 0
  expect_gzip = ObjectIsGzipEncoded(src_obj_metadata)
  # Assign an end FileMessage to each component
  for cp_result in cp_results:
    PutToQueueWithTimeout(
        status_queue,
        FileMessage(src_url,
                    dst_url,
                    time.time(),
                    size=cp_result.component_total_size,
                    finished=True,
                    component_num=cp_result.component_num,
                    message_type=FileMessage.COMPONENT_TO_DOWNLOAD))

    bytes_transferred += cp_result.bytes_transferred
    server_gzip = (cp_result.server_encoding and
                   cp_result.server_encoding.lower().endswith('gzip'))
    # If the server gzipped any components on the fly, we will have no chance of
    # properly reconstructing the file.
    if server_gzip and not expect_gzip:
      raise CommandException(
          'Download of %s failed because the server sent back data with an '
          'unexpected encoding.' % dst_url.object_name)

  return bytes_transferred, crc32c


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):
  """Downloads an object to a local file using the resumable strategy.

  Args:
    src_url: Source CloudUrl.
    src_obj_metadata: Metadata from the source object.
    dst_url: Destination FileUrl.
    download_file_name: Temporary file name to be used for download.
    gsutil_api: gsutil Cloud API instance to use for the download.
    logger: for outputting log messages.
    digesters: Digesters corresponding to the hash algorithms that will be used
               for validation.
    component_num: Which component of a sliced download this call is for, or
                   None if this is not a sliced download.
    start_byte: The first byte of a byte range for a sliced download.
    end_byte: The last byte of a byte range for a sliced download.
    decryption_key: Base64-encoded decryption key for the source object, if any.

  Returns:
    (bytes_transferred, server_encoding)
    bytes_transferred: Number of bytes transferred from server this call.
    server_encoding: Content-encoding string if it was detected that the server
                     sent encoded bytes during transfer, None otherwise.
  """
  if end_byte is None:
    end_byte = src_obj_metadata.size - 1
  download_size = end_byte - start_byte + 1

  is_sliced = component_num is not None
  api_selector = gsutil_api.GetApiSelector(provider=src_url.scheme)
  server_encoding = None

  # Used for logging
  download_name = dst_url.object_name
  if is_sliced:
    download_name += ' component %d' % component_num

  fp = None
  try:
    fp = open(download_file_name, 'r+b')
    fp.seek(start_byte)
    api_selector = gsutil_api.GetApiSelector(provider=src_url.scheme)
    existing_file_size = GetFileSize(fp)

    tracker_file_name, download_start_byte = ReadOrCreateDownloadTrackerFile(
        src_obj_metadata,
        dst_url,
        logger,
        api_selector,
        start_byte,
        existing_file_size,
        component_num,
    )

    if download_start_byte < start_byte or download_start_byte > end_byte + 1:
      DeleteTrackerFile(tracker_file_name)
      raise CommandException(
          'Resumable download start point for %s is not in the correct byte '
          'range. Deleting tracker file, so if you re-try this download it '
          'will start from scratch' % download_name)

    download_complete = (download_start_byte == start_byte + download_size)
    resuming = (download_start_byte != start_byte) and not download_complete
    if resuming:
      logger.info('Resuming download for %s', download_name)
    elif download_complete:
      logger.info(
          'Download already complete for %s, skipping download but '
          'will run integrity checks.', download_name)

    # This is used for resuming downloads, but also for passing the mediaLink
    # and size into the download for new downloads so that we can avoid
    # making an extra HTTP call.
    serialization_data = GetDownloadSerializationData(
        src_obj_metadata,
        progress=download_start_byte,
        user_project=gsutil_api.user_project)

    if resuming or download_complete:
      # Catch up our digester with the hash data.
      bytes_digested = 0
      total_bytes_to_digest = download_start_byte - start_byte
      hash_callback = ProgressCallbackWithTimeout(
          total_bytes_to_digest,
          FileProgressCallbackHandler(gsutil_api.status_queue,
                                      component_num=component_num,
                                      src_url=src_url,
                                      dst_url=dst_url,
                                      operation_name='Hashing').call)

      while bytes_digested < total_bytes_to_digest:
        bytes_to_read = min(DEFAULT_FILE_BUFFER_SIZE,
                            total_bytes_to_digest - bytes_digested)
        data = fp.read(bytes_to_read)
        bytes_digested += bytes_to_read
        for alg_name in digesters:
          digesters[alg_name].update(six.ensure_binary(data))
        hash_callback.Progress(len(data))

    elif not is_sliced:
      # Delete file contents and start entire object download from scratch.
      fp.truncate(0)
      existing_file_size = 0

    progress_callback = FileProgressCallbackHandler(
        gsutil_api.status_queue,
        start_byte=start_byte,
        override_total_size=download_size,
        src_url=src_url,
        dst_url=dst_url,
        component_num=component_num,
        operation_name='Downloading').call

    if global_copy_helper_opts.test_callback_file:
      with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
        progress_callback = pickle.loads(test_fp.read()).call

    if is_sliced and src_obj_metadata.size >= ResumableThreshold():
      fp = SlicedDownloadFileWrapper(fp, tracker_file_name, src_obj_metadata,
                                     start_byte, end_byte)

    compressed_encoding = ObjectIsGzipEncoded(src_obj_metadata)

    # TODO: With gzip encoding (which may occur on-the-fly and not be part of
    # the object's metadata), when we request a range to resume, it's possible
    # that the server will just resend the entire object, which means our
    # caught-up hash will be incorrect.  We recalculate the hash on
    # the local file in the case of a failed gzip hash anyway, but it would
    # be better if we actively detected this case.
    if not download_complete:
      fp.seek(download_start_byte)
      server_encoding = gsutil_api.GetObjectMedia(
          src_url.bucket_name,
          src_url.object_name,
          fp,
          start_byte=download_start_byte,
          end_byte=end_byte,
          compressed_encoding=compressed_encoding,
          generation=src_url.generation,
          object_size=src_obj_metadata.size,
          download_strategy=CloudApi.DownloadStrategy.RESUMABLE,
          provider=src_url.scheme,
          serialization_data=serialization_data,
          digesters=digesters,
          progress_callback=progress_callback,
          decryption_tuple=CryptoKeyWrapperFromKey(decryption_key))

  except ResumableDownloadException as e:
    logger.warning('Caught ResumableDownloadException (%s) for download of %s.',
                   e.reason, download_name)
    raise
  finally:
    if fp:
      fp.close()

  bytes_transferred = end_byte - download_start_byte + 1
  return bytes_transferred, server_encoding


def _DownloadObjectToFileNonResumable(src_url,
                                      src_obj_metadata,
                                      dst_url,
                                      download_file_name,
                                      gsutil_api,
                                      digesters,
                                      decryption_key=None):
  """Downloads an object to a local file using the non-resumable strategy.

  This function does not support component transfers.
  Args:
    src_url: Source CloudUrl.
    src_obj_metadata: Metadata from the source object.
    dst_url: Destination FileUrl.
    download_file_name: Temporary file name to be used for download.
    gsutil_api: gsutil Cloud API instance to use for the download.
    digesters: Digesters corresponding to the hash algorithms that will be used
               for validation.
    decryption_key: Base64-encoded decryption key for the source object, if any.

  Returns:
    (bytes_transferred, server_encoding)
    bytes_transferred: Number of bytes transferred from server this call.
    server_encoding: Content-encoding string if it was detected that the server
                     sent encoded bytes during transfer, None otherwise.
  """
  fp = None
  try:
    fp = open(download_file_name, 'w')

    # This is used to pass the mediaLink and the size into the download so that
    # we can avoid making an extra HTTP call.
    serialization_data = GetDownloadSerializationData(
        src_obj_metadata, 0, user_project=gsutil_api.user_project)

    progress_callback = FileProgressCallbackHandler(
        gsutil_api.status_queue,
        src_url=src_url,
        dst_url=dst_url,
        operation_name='Downloading').call

    if global_copy_helper_opts.test_callback_file:
      with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
        progress_callback = pickle.loads(test_fp.read()).call

    server_encoding = gsutil_api.GetObjectMedia(
        src_url.bucket_name,
        src_url.object_name,
        fp,
        generation=src_url.generation,
        object_size=src_obj_metadata.size,
        download_strategy=CloudApi.DownloadStrategy.ONE_SHOT,
        provider=src_url.scheme,
        serialization_data=serialization_data,
        digesters=digesters,
        progress_callback=progress_callback,
        decryption_tuple=CryptoKeyWrapperFromKey(decryption_key))
  finally:
    if fp:
      fp.close()

  return src_obj_metadata.size, server_encoding


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):
  """Downloads an object to a local file.

  Args:
    src_url: Source CloudUrl.
    src_obj_metadata: Metadata from the source object.
    dst_url: Destination FileUrl.
    gsutil_api: gsutil Cloud API instance to use for the download.
    logger: for outputting log messages.
    command_obj: command object for use in Apply in sliced downloads.
    copy_exception_handler: For handling copy exceptions during Apply.
    allow_splitting: Whether or not to allow sliced download.
    decryption_key: Base64-encoded decryption key for the source object, if any.
    is_rsync: Whether or not the caller is the rsync command.
    preserve_posix: Whether or not to preserve POSIX attributes.
    use_stet: Decrypt downloaded file with STET binary if available on system.

  Returns:
    (elapsed_time, bytes_transferred, dst_url, md5), where time elapsed
    excludes initial GET.

  Raises:
    FileConcurrencySkipError: if this download is already in progress.
    CommandException: if other errors encountered.
  """
  global open_files_map, open_files_lock
  if dst_url.object_name.endswith(dst_url.delim):
    logger.warn('\n'.join(
        textwrap.wrap(
            'Skipping attempt to download to filename ending with slash (%s). This '
            'typically happens when using gsutil to download from a subdirectory '
            'created by the Cloud Console (https://cloud.google.com/console)' %
            dst_url.object_name)))
    # The warning above is needed because errors might get ignored
    # for parallel processing.
    raise InvalidUrlError('Invalid destination path: %s' % dst_url.object_name)

  api_selector = gsutil_api.GetApiSelector(provider=src_url.scheme)
  download_strategy = _SelectDownloadStrategy(dst_url)
  sliced_download = _ShouldDoSlicedDownload(download_strategy, src_obj_metadata,
                                            allow_splitting, logger)

  download_file_name, need_to_unzip = _GetDownloadFile(dst_url,
                                                       src_obj_metadata, logger)

  # Ensure another process/thread is not already writing to this file.
  with open_files_lock:
    if open_files_map.get(download_file_name, False):
      raise FileConcurrencySkipError
    open_files_map[download_file_name] = True

  # Set up hash digesters.
  consider_md5 = src_obj_metadata.md5Hash and not sliced_download
  hash_algs = GetDownloadHashAlgs(logger,
                                  consider_md5=consider_md5,
                                  consider_crc32c=src_obj_metadata.crc32c)
  digesters = dict((alg, hash_algs[alg]()) for alg in hash_algs or {})

  # Tracks whether the server used a gzip encoding.
  server_encoding = None
  download_complete = (src_obj_metadata.size == 0)
  bytes_transferred = 0

  start_time = time.time()
  if not download_complete:
    if sliced_download:
      (bytes_transferred,
       crc32c) = (_DoSlicedDownload(src_url,
                                    src_obj_metadata,
                                    dst_url,
                                    download_file_name,
                                    command_obj,
                                    logger,
                                    copy_exception_handler,
                                    api_selector,
                                    decryption_key=decryption_key,
                                    status_queue=gsutil_api.status_queue))
      if 'crc32c' in digesters:
        digesters['crc32c'].crcValue = crc32c
    elif download_strategy is CloudApi.DownloadStrategy.ONE_SHOT:
      bytes_transferred, server_encoding = _DownloadObjectToFileNonResumable(
          src_url,
          src_obj_metadata,
          dst_url,
          download_file_name,
          gsutil_api,
          digesters,
          decryption_key=decryption_key,
      )
    elif download_strategy is CloudApi.DownloadStrategy.RESUMABLE:
      bytes_transferred, server_encoding = _DownloadObjectToFileResumable(
          src_url,
          src_obj_metadata,
          dst_url,
          download_file_name,
          gsutil_api,
          logger,
          digesters,
          decryption_key=decryption_key,
      )
    else:
      raise CommandException('Invalid download strategy %s chosen for'
                             'file %s' %
                             (download_strategy, download_file_name))
  end_time = time.time()

  server_gzip = server_encoding and server_encoding.lower().endswith('gzip')
  local_md5 = _ValidateAndCompleteDownload(logger,
                                           src_url,
                                           src_obj_metadata,
                                           dst_url,
                                           need_to_unzip,
                                           server_gzip,
                                           digesters,
                                           hash_algs,
                                           download_file_name,
                                           api_selector,
                                           bytes_transferred,
                                           gsutil_api,
                                           is_rsync=is_rsync,
                                           preserve_posix=preserve_posix,
                                           use_stet=use_stet)

  with open_files_lock:
    open_files_map.delete(download_file_name)

  PutToQueueWithTimeout(
      gsutil_api.status_queue,
      FileMessage(src_url,
                  dst_url,
                  message_time=end_time,
                  message_type=FileMessage.FILE_DOWNLOAD,
                  size=src_obj_metadata.size,
                  finished=True))

  return (end_time - start_time, bytes_transferred, dst_url, local_md5)


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):
  """Validates and performs necessary operations on a downloaded file.

  Validates the integrity of the downloaded file using hash_algs. If the file
  was compressed (temporarily), the file will be decompressed. Then, if the
  integrity of the file was successfully validated, the file will be moved
  from its temporary download location to its permanent location on disk.

  Args:
    logger: For outputting log messages.
    src_url: StorageUrl for the source object.
    src_obj_metadata: Metadata for the source object, potentially containing
                      hash values.
    dst_url: StorageUrl describing the destination file.
    need_to_unzip: If true, a temporary zip file was used and must be
                   uncompressed as part of validation.
    server_gzip: If true, the server gzipped the bytes (regardless of whether
                 the object metadata claimed it was gzipped).
    digesters: dict of {string, hash digester} that contains up-to-date digests
               computed during the download. If a digester for a particular
               algorithm is None, an up-to-date digest is not available and the
               hash must be recomputed from the local file.
    hash_algs: dict of {string, hash algorithm} that can be used if digesters
               don't have up-to-date digests.
    temporary_file_name: Temporary file name that was used for download.
    api_selector: The Cloud API implementation used (used tracker file naming).
    bytes_transferred: Number of bytes downloaded (used for logging).
    gsutil_api: Cloud API to use for service and status.
    is_rsync: Whether or not the caller is the rsync function. Used to determine
              if timeCreated should be used.
    preserve_posix: Whether or not to preserve the posix attributes.
    use_stet: If True, attempt to decrypt downloaded files with the STET
              binary if it's present on the system.

  Returns:
    An MD5 of the local file, if one was calculated as part of the integrity
    check.
  """
  final_file_name = dst_url.object_name
  digesters_succeeded = True

  for alg in digesters:
    # If we get a digester with a None algorithm, the underlying
    # implementation failed to calculate a digest, so we will need to
    # calculate one from scratch.
    if not digesters[alg]:
      digesters_succeeded = False
      break

  if digesters_succeeded:
    local_hashes = _CreateDigestsFromDigesters(digesters)
  else:
    local_hashes = _CreateDigestsFromLocalFile(gsutil_api.status_queue,
                                               hash_algs, temporary_file_name,
                                               src_url, src_obj_metadata)

  digest_verified = True
  hash_invalid_exception = None
  try:
    _CheckHashes(logger, src_url, src_obj_metadata, final_file_name,
                 local_hashes)
    DeleteDownloadTrackerFiles(dst_url, api_selector)
  except HashMismatchException as e:
    # If an non-gzipped object gets sent with gzip content encoding, the hash
    # we calculate will match the gzipped bytes, not the original object. Thus,
    # we'll need to calculate and check it after unzipping.
    if server_gzip:
      logger.debug('Hash did not match but server gzipped the content, will '
                   'recalculate.')
      digest_verified = False
    elif api_selector == ApiSelector.XML:
      logger.debug(
          'Hash did not match but server may have gzipped the content, will '
          'recalculate.')
      # Save off the exception in case this isn't a gzipped file.
      hash_invalid_exception = e
      digest_verified = False
    else:
      DeleteDownloadTrackerFiles(dst_url, api_selector)
      if _RENAME_ON_HASH_MISMATCH:
        os.rename(temporary_file_name,
                  final_file_name + _RENAME_ON_HASH_MISMATCH_SUFFIX)
      else:
        os.unlink(temporary_file_name)
      raise

  if not (need_to_unzip or server_gzip):
    unzipped_temporary_file_name = temporary_file_name
  else:
    # This will not result in the same string as temporary_file_name b/c
    # GetTempFileName returns ".gstmp" and gzipped temp files have ".gztmp".
    unzipped_temporary_file_name = temporary_file_util.GetTempFileName(dst_url)
    # Log that we're uncompressing if the file is big enough that
    # decompressing would make it look like the transfer "stalled" at the end.
    if bytes_transferred > TEN_MIB:
      logger.info('Uncompressing temporarily gzipped file to %s...',
                  final_file_name)

    gzip_fp = None
    try:
      # Downloaded temporarily gzipped file, unzip to file without '_.gztmp'
      # suffix.
      gzip_fp = gzip.open(temporary_file_name, 'rb')
      with open(unzipped_temporary_file_name, 'wb') as f_out:
        data = gzip_fp.read(GZIP_CHUNK_SIZE)
        while data:
          f_out.write(data)
          data = gzip_fp.read(GZIP_CHUNK_SIZE)
    except IOError as e:
      # In the XML case where we don't know if the file was gzipped, raise
      # the original hash exception if we find that it wasn't.
      if 'Not a gzipped file' in str(e) and hash_invalid_exception:
        # Linter improperly thinks we're raising None despite the above check.
        # pylint: disable=raising-bad-type
        raise hash_invalid_exception
    finally:
      if gzip_fp:
        gzip_fp.close()

    os.unlink(temporary_file_name)

  if not digest_verified:
    try:
      # Recalculate hashes on the unzipped local file.
      local_hashes = _CreateDigestsFromLocalFile(gsutil_api.status_queue,
                                                 hash_algs,
                                                 unzipped_temporary_file_name,
                                                 src_url, src_obj_metadata)
      _CheckHashes(logger, src_url, src_obj_metadata, final_file_name,
                   local_hashes)
      DeleteDownloadTrackerFiles(dst_url, api_selector)
    except HashMismatchException:
      DeleteDownloadTrackerFiles(dst_url, api_selector)
      if _RENAME_ON_HASH_MISMATCH:
        os.rename(
            unzipped_temporary_file_name,
            unzipped_temporary_file_name + _RENAME_ON_HASH_MISMATCH_SUFFIX)
      else:
        os.unlink(unzipped_temporary_file_name)
      raise

  if use_stet:
    # Decrypt data using STET binary.
    stet_util.decrypt_download(src_url, dst_url, unzipped_temporary_file_name,
                               logger)

  os.rename(unzipped_temporary_file_name, final_file_name)
  ParseAndSetPOSIXAttributes(final_file_name,
                             src_obj_metadata,
                             is_rsync=is_rsync,
                             preserve_posix=preserve_posix)

  if 'md5' in local_hashes:
    return local_hashes['md5']


def _CopyFileToFile(src_url, dst_url, status_queue=None, src_obj_metadata=None):
  """Copies a local file to a local file.

  Args:
    src_url: Source FileUrl.
    dst_url: Destination FileUrl.
    status_queue: Queue for posting file messages for UI/Analytics.
    src_obj_metadata: An apitools Object that may contain file size, or None.

  Returns:
    (elapsed_time, bytes_transferred, dst_url, md5=None).

  Raises:
    CommandException: if errors encountered.
  """
  src_fp = GetStreamFromFileUrl(src_url)
  dir_name = os.path.dirname(dst_url.object_name)

  if dir_name:
    try:
      os.makedirs(dir_name)
    except OSError as e:
      if e.errno != errno.EEXIST:
        raise

  with open(dst_url.object_name, 'wb') as dst_fp:
    start_time = time.time()
    shutil.copyfileobj(src_fp, dst_fp)
  if not src_url.IsStream():
    src_fp.close()  # Explicitly close the src fp - necessary if it is a fifo.
  end_time = time.time()
  PutToQueueWithTimeout(
      status_queue,
      FileMessage(src_url,
                  dst_url,
                  end_time,
                  message_type=FileMessage.FILE_LOCAL_COPY,
                  size=src_obj_metadata.size if src_obj_metadata else None,
                  finished=True))

  return (end_time - start_time, os.path.getsize(dst_url.object_name), dst_url,
          None)


def _DummyTrackerCallback(_):
  pass


# pylint: disable=undefined-variable
def _CopyObjToObjDaisyChainMode(src_url,
                                src_obj_metadata,
                                dst_url,
                                dst_obj_metadata,
                                preconditions,
                                gsutil_api,
                                logger,
                                decryption_key=None):
  """Copies from src_url to dst_url in "daisy chain" mode.

  See -D OPTION documentation about what daisy chain mode is.

  Args:
    src_url: Source CloudUrl
    src_obj_metadata: Metadata from source object
    dst_url: Destination CloudUrl
    dst_obj_metadata: Object-specific metadata that should be overidden during
                      the copy.
    preconditions: Preconditions to use for the copy.
    gsutil_api: gsutil Cloud API to use for the copy.
    logger: For outputting log messages.
    decryption_key: Base64-encoded decryption key for the source object, if any.

  Returns:
    (elapsed_time, bytes_transferred, dst_url with generation,
    md5 hash of destination) excluding overhead like initial GET.

  Raises:
    CommandException: if errors encountered.
  """
  # We don't attempt to preserve ACLs across providers because
  # GCS and S3 support different ACLs and disjoint principals.
  if (global_copy_helper_opts.preserve_acl and
      src_url.scheme != dst_url.scheme):
    raise NotImplementedError('Cross-provider cp -p not supported')
  if not global_copy_helper_opts.preserve_acl:
    dst_obj_metadata.acl = []

  # Don't use callbacks for downloads on the daisy chain wrapper because
  # upload callbacks will output progress, but respect test hooks if present.
  progress_callback = None
  if global_copy_helper_opts.test_callback_file:
    with open(global_copy_helper_opts.test_callback_file, 'rb') as test_fp:
      progress_callback = pickle.loads(test_fp.read()).call

  compressed_encoding = ObjectIsGzipEncoded(src_obj_metadata)
  encryption_keywrapper = GetEncryptionKeyWrapper(config)

  start_time = time.time()
  upload_fp = DaisyChainWrapper(src_url,
                                src_obj_metadata.size,
                                gsutil_api,
                                compressed_encoding=compressed_encoding,
                                progress_callback=progress_callback,
                                decryption_key=decryption_key)
  uploaded_object = None
  if src_obj_metadata.size == 0:
    # Resumable uploads of size 0 are not supported.
    uploaded_object = gsutil_api.UploadObject(
        upload_fp,
        object_metadata=dst_obj_metadata,
        canned_acl=global_copy_helper_opts.canned_acl,
        preconditions=preconditions,
        provider=dst_url.scheme,
        fields=UPLOAD_RETURN_FIELDS,
        size=src_obj_metadata.size,
        encryption_tuple=encryption_keywrapper)
  else:
    # TODO: Support process-break resumes. This will resume across connection
    # breaks and server errors, but the tracker callback is a no-op so this
    # won't resume across gsutil runs.
    # TODO: Test retries via test_callback_file.
    uploaded_object = gsutil_api.UploadObjectResumable(
        upload_fp,
        object_metadata=dst_obj_metadata,
        canned_acl=global_copy_helper_opts.canned_acl,
        preconditions=preconditions,
        provider=dst_url.scheme,
        fields=UPLOAD_RETURN_FIELDS,
        size=src_obj_metadata.size,
        progress_callback=FileProgressCallbackHandler(
            gsutil_api.status_queue,
            src_url=src_url,
            dst_url=dst_url,
            operation_name='Uploading').call,
        tracker_callback=_DummyTrackerCallback,
        encryption_tuple=encryption_keywrapper)
  end_time = time.time()

  try:
    _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
                      uploaded_object)
  except HashMismatchException:
    if _RENAME_ON_HASH_MISMATCH:
      corrupted_obj_metadata = apitools_messages.Object(
          name=dst_obj_metadata.name,
          bucket=dst_obj_metadata.bucket,
          etag=uploaded_object.etag)
      dst_obj_metadata.name = (dst_url.object_name +
                               _RENAME_ON_HASH_MISMATCH_SUFFIX)
      decryption_keywrapper = CryptoKeyWrapperFromKey(decryption_key)
      gsutil_api.CopyObject(corrupted_obj_metadata,
                            dst_obj_metadata,
                            provider=dst_url.scheme,
                            decryption_tuple=decryption_keywrapper,
                            encryption_tuple=encryption_keywrapper)
    # If the digest doesn't match, delete the object.
    gsutil_api.DeleteObject(dst_url.bucket_name,
                            dst_url.object_name,
                            generation=uploaded_object.generation,
                            provider=dst_url.scheme)
    raise

  result_url = dst_url.Clone()
  result_url.generation = GenerationFromUrlAndString(result_url,
                                                     uploaded_object.generation)

  PutToQueueWithTimeout(
      gsutil_api.status_queue,
      FileMessage(src_url,
                  dst_url,
                  end_time,
                  message_type=FileMessage.FILE_DAISY_COPY,
                  size=src_obj_metadata.size,
                  finished=True))

  return (end_time - start_time, src_obj_metadata.size, result_url,
          uploaded_object.md5Hash)


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)


# Map of (lowercase) storage classes with early deletion charges to their
# minimum lifetime in seconds.
EARLY_DELETION_MINIMUM_LIFETIME = {
    'nearline': 30 * SECONDS_PER_DAY,
    'coldline': 90 * SECONDS_PER_DAY,
    'archive': 365 * SECONDS_PER_DAY
}


def WarnIfMvEarlyDeletionChargeApplies(src_url, src_obj_metadata, logger):
  """Warns when deleting a gs:// object could incur an early deletion charge.

  This function inspects metadata for Google Cloud Storage objects that are
  subject to early deletion charges (such as Nearline), and warns when
  performing operations like mv that would delete them.

  Args:
    src_url: CloudUrl for the source object.
    src_obj_metadata: source object metadata with necessary fields
        (per GetSourceFieldsNeededForCopy).
    logger: logging.Logger for outputting warning.
  """
  if (src_url.scheme == 'gs' and src_obj_metadata and
      src_obj_metadata.timeCreated and src_obj_metadata.storageClass):
    object_storage_class = src_obj_metadata.storageClass.lower()
    early_deletion_cutoff_seconds = EARLY_DELETION_MINIMUM_LIFETIME.get(
        object_storage_class, None)
    if early_deletion_cutoff_seconds:
      minimum_delete_age = (
          early_deletion_cutoff_seconds +
          ConvertDatetimeToPOSIX(src_obj_metadata.timeCreated))
      if time.time() < minimum_delete_age:
        logger.warn(
            'Warning: moving %s object %s may incur an early deletion '
            'charge, because the original object is less than %s '
            'days old according to the local system time.',
            object_storage_class, src_url.url_string,
            early_deletion_cutoff_seconds // SECONDS_PER_DAY)


def MaybeSkipUnsupportedObject(src_url, src_obj_metadata):
  """Skips unsupported object types if requested.

  Args:
    src_url: CloudUrl for the source object.
    src_obj_metadata: source object metadata with storageClass field
        (per GetSourceFieldsNeededForCopy).

  Raises:
    SkipGlacierError: if skipping a s3 Glacier object.
  """
  if (src_url.scheme == 's3' and
      global_copy_helper_opts.skip_unsupported_objects and
      src_obj_metadata.storageClass == 'GLACIER'):
    raise SkipGlacierError()


def GetDecryptionCSEK(src_url, src_obj_metadata):
  """Ensures a matching decryption key is available for the source object.

  Args:
    src_url: CloudUrl for the source object.
    src_obj_metadata: source object metadata with optional customerEncryption
        field.

  Raises:
    EncryptionException if the object is encrypted and no matching key is found.

  Returns:
    Base64-encoded decryption key string if the object is encrypted and a
    matching key is found, or None if object is not encrypted.
  """
  if src_obj_metadata.customerEncryption:
    decryption_key = FindMatchingCSEKInBotoConfig(
        src_obj_metadata.customerEncryption.keySha256, config)
    if not decryption_key:
      raise EncryptionException(
          'Missing decryption key with SHA256 hash %s. No decryption key '
          'matches object %s' %
          (src_obj_metadata.customerEncryption.keySha256, src_url))
    return decryption_key


# pylint: disable=undefined-variable
# pylint: disable=too-many-statements
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):
  """Performs copy from src_url to dst_url, handling various special cases.

  Args:
    logger: for outputting log messages.
    src_url: Source StorageUrl.
    dst_url: Destination StorageUrl.
    gsutil_api: gsutil Cloud API instance to use for the copy.
    command_obj: command object for use in Apply in parallel composite uploads
        and sliced object downloads.
    copy_exception_handler: for handling copy exceptions during Apply.
    src_obj_metadata: If source URL is a cloud object, source object metadata
        with all necessary fields (per GetSourceFieldsNeededForCopy).
        Required for cloud source URLs. If source URL is a file, an
        apitools Object that may contain file size, or None.
    allow_splitting: Whether to allow the file to be split into component
                     pieces for an parallel composite upload or download.
    headers: optional headers to use for the copy operation.
    manifest: optional manifest for tracking copy operations.
    gzip_exts: List of file extensions to gzip, if any.
               If gzip_exts is GZIP_ALL_FILES, gzip all files.
    is_rsync: Whether or not the caller is the rsync command.
    preserve_posix: Whether or not to preserve posix attributes.
    gzip_encoded: Whether to use gzip transport encoding for the upload. Used
        in conjunction with gzip_exts. Streaming files compressed is only
        supported on the JSON GCS API.
    use_stet: If True, will perform STET encryption or decryption using
        the binary specified in the boto config or PATH.

  Returns:
    (elapsed_time, bytes_transferred, version-specific dst_url) excluding
    overhead like initial GET.

  Raises:
    ItemExistsError: if no clobber flag is specified and the destination
        object already exists.
    SkipUnsupportedObjectError: if skip_unsupported_objects flag is specified
        and the source is an unsupported type.
    CommandException: if other errors encountered.
  """
  # TODO: Remove elapsed_time as it is currently unused by all callers.
  if headers:
    dst_obj_headers = headers.copy()
  else:
    dst_obj_headers = {}

  # Create a metadata instance for each destination object so metadata
  # such as content-type can be applied per-object.
  # Initialize metadata from any headers passed in via -h.
  dst_obj_metadata = ObjectMetadataFromHeaders(dst_obj_headers)

  if dst_url.IsCloudUrl() and dst_url.scheme == 'gs':
    preconditions = PreconditionsFromHeaders(dst_obj_headers)
  else:
    preconditions = Preconditions()

  src_obj_filestream = None
  decryption_key = None
  copy_in_the_cloud = False
  if src_url.IsCloudUrl():
    if (dst_url.IsCloudUrl() and src_url.scheme == dst_url.scheme and
        not global_copy_helper_opts.daisy_chain):
      copy_in_the_cloud = True

    if global_copy_helper_opts.perform_mv:
      WarnIfMvEarlyDeletionChargeApplies(src_url, src_obj_metadata, logger)
    MaybeSkipUnsupportedObject(src_url, src_obj_metadata)
    decryption_key = GetDecryptionCSEK(src_url, src_obj_metadata)

    src_obj_size = src_obj_metadata.size
    dst_obj_metadata.contentType = src_obj_metadata.contentType
    if global_copy_helper_opts.preserve_acl and dst_url.IsCloudUrl():
      if src_url.scheme == 'gs' and not src_obj_metadata.acl:
        raise CommandException(
            'No OWNER permission found for object %s. OWNER permission is '
            'required for preserving ACLs.' % src_url)
      dst_obj_metadata.acl = src_obj_metadata.acl
      # Special case for S3-to-S3 copy URLs using
      # global_copy_helper_opts.preserve_acl.
      # dst_url will be verified in _CopyObjToObjDaisyChainMode if it
      # is not s3 (and thus differs from src_url).
      if src_url.scheme == 's3':
        acl_text = S3MarkerAclFromObjectMetadata(src_obj_metadata)
        if acl_text:
          AddS3MarkerAclToObjectMetadata(dst_obj_metadata, acl_text)
  else:  # src_url.IsFileUrl()
    if use_stet:
      source_stream_url = stet_util.encrypt_upload(src_url, dst_url, logger)
    else:
      source_stream_url = src_url
    try:
      src_obj_filestream = GetStreamFromFileUrl(source_stream_url)
    except Exception as e:  # pylint: disable=broad-except
      message = 'Error opening file "%s": %s.' % (src_url, str(e))
      if command_obj.continue_on_error:
        command_obj.op_failure_count += 1
        logger.error(message)
        return
      else:
        raise CommandException(message)
    if src_url.IsStream() or src_url.IsFifo():
      src_obj_size = None
    elif src_obj_metadata and src_obj_metadata.size and not use_stet:
      # Iterator retrieved the file's size, no need to stat it again.
      # Unless STET changed the file size.
      src_obj_size = src_obj_metadata.size
    else:
      src_obj_size = os.path.getsize(source_stream_url.object_name)

  if global_copy_helper_opts.use_manifest:
    # Set the source size in the manifest.
    manifest.Set(src_url.url_string, 'size', src_obj_size)

  if (dst_url.scheme == 's3' and src_url != 's3' and
      src_obj_size is not None and  # Can't compare int to None in py3
      src_obj_size > S3_MAX_UPLOAD_SIZE):
    raise CommandException(
        '"%s" exceeds the maximum gsutil-supported size for an S3 upload. S3 '
        'objects greater than %s in size require multipart uploads, which '
        'gsutil does not support.' %
        (src_url, MakeHumanReadable(S3_MAX_UPLOAD_SIZE)))

  # On Windows, stdin is opened as text mode instead of binary which causes
  # problems when piping a binary file, so this switches it to binary mode.
  if IS_WINDOWS and src_url.IsFileUrl() and src_url.IsStream():
    msvcrt.setmode(GetStreamFromFileUrl(src_url).fileno(), os.O_BINARY)

  if global_copy_helper_opts.no_clobber:
    # There are two checks to prevent clobbering:
    # 1) The first check is to see if the URL
    #    already exists at the destination and prevent the upload/download
    #    from happening. This is done by the exists() call.
    # 2) The second check is only relevant if we are writing to gs. We can
    #    enforce that the server only writes the object if it doesn't exist
    #    by specifying the header below. This check only happens at the
    #    server after the complete file has been uploaded. We specify this
    #    header to prevent a race condition where a destination file may
    #    be created after the first check and before the file is fully
    #    uploaded.
    # In order to save on unnecessary uploads/downloads we perform both
    # checks. However, this may come at the cost of additional HTTP calls.
    if preconditions.gen_match:
      raise ArgumentException('Specifying x-goog-if-generation-match is '
                              'not supported with cp -n')
    else:
      preconditions.gen_match = 0
    if dst_url.IsFileUrl() and os.path.exists(dst_url.object_name):
      raise ItemExistsError()
    elif dst_url.IsCloudUrl():
      try:
        dst_object = gsutil_api.GetObjectMetadata(dst_url.bucket_name,
                                                  dst_url.object_name,
                                                  provider=dst_url.scheme)
      except NotFoundException:
        dst_object = None
      if dst_object:
        raise ItemExistsError()

  if dst_url.IsCloudUrl():
    # Cloud storage API gets object and bucket name from metadata.
    dst_obj_metadata.name = dst_url.object_name
    dst_obj_metadata.bucket = dst_url.bucket_name
    if src_url.IsCloudUrl():
      # Preserve relevant metadata from the source object if it's not already
      # provided from the headers.
      src_obj_metadata.name = src_url.object_name
      src_obj_metadata.bucket = src_url.bucket_name
    else:
      _SetContentTypeFromFile(src_url, dst_obj_metadata)
    # Only set KMS key name if destination provider is 'gs'.
    encryption_keywrapper = GetEncryptionKeyWrapper(config)
    if (encryption_keywrapper and
        encryption_keywrapper.crypto_type == CryptoKeyType.CMEK and
        dst_url.scheme == 'gs'):
      dst_obj_metadata.kmsKeyName = encryption_keywrapper.crypto_key

  if src_obj_metadata:
    # Note that CopyObjectMetadata only copies specific fields. We intentionally
    # do not copy storageClass, as the bucket's default storage class should be
    # used (when copying to a gs:// bucket) unless explicitly overridden.
    CopyObjectMetadata(src_obj_metadata, dst_obj_metadata, override=False)

  if global_copy_helper_opts.dest_storage_class:
    dst_obj_metadata.storageClass = global_copy_helper_opts.dest_storage_class

  if config.get('GSUtil', 'check_hashes') == CHECK_HASH_NEVER:
    # GCS server will perform MD5 validation if the md5 hash is present.
    # Remove md5_hash if check_hashes=never.
    dst_obj_metadata.md5Hash = None

  _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata)

  if src_url.IsCloudUrl():
    if dst_url.IsFileUrl():
      PutToQueueWithTimeout(
          gsutil_api.status_queue,
          FileMessage(src_url,
                      dst_url,
                      time.time(),
                      message_type=FileMessage.FILE_DOWNLOAD,
                      size=src_obj_size,
                      finished=False))
      return _DownloadObjectToFile(src_url,
                                   src_obj_metadata,
                                   dst_url,
                                   gsutil_api,
                                   logger,
                                   command_obj,
                                   copy_exception_handler,
                                   allow_splitting=allow_splitting,
                                   decryption_key=decryption_key,
                                   is_rsync=is_rsync,
                                   preserve_posix=preserve_posix,
                                   use_stet=use_stet)
    elif copy_in_the_cloud:
      PutToQueueWithTimeout(
          gsutil_api.status_queue,
          FileMessage(src_url,
                      dst_url,
                      time.time(),
                      message_type=FileMessage.FILE_CLOUD_COPY,
                      size=src_obj_size,
                      finished=False))
      return _CopyObjToObjInTheCloud(src_url,
                                     src_obj_metadata,
                                     dst_url,
                                     dst_obj_metadata,
                                     preconditions,
                                     gsutil_api,
                                     decryption_key=decryption_key)
    else:
      PutToQueueWithTimeout(
          gsutil_api.status_queue,
          FileMessage(src_url,
                      dst_url,
                      time.time(),
                      message_type=FileMessage.FILE_DAISY_COPY,
                      size=src_obj_size,
                      finished=False))
      return _CopyObjToObjDaisyChainMode(src_url,
                                         src_obj_metadata,
                                         dst_url,
                                         dst_obj_metadata,
                                         preconditions,
                                         gsutil_api,
                                         logger,
                                         decryption_key=decryption_key)
  else:  # src_url.IsFileUrl()
    if dst_url.IsCloudUrl():
      # The FileMessage for this upload object is inside _UploadFileToObject().
      # This is such because the function may alter src_url, which would prevent
      # us from correctly tracking the new url.
      uploaded_metadata = _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=gzip_exts,
                                              allow_splitting=allow_splitting,
                                              gzip_encoded=gzip_encoded)
      if use_stet:
        # Delete temporary file.
        os.unlink(src_obj_filestream.name)
      return uploaded_metadata
    else:  # dst_url.IsFileUrl()
      PutToQueueWithTimeout(
          gsutil_api.status_queue,
          FileMessage(src_url,
                      dst_url,
                      time.time(),
                      message_type=FileMessage.FILE_LOCAL_COPY,
                      size=src_obj_size,
                      finished=False))
      result = _CopyFileToFile(src_url,
                               dst_url,
                               status_queue=gsutil_api.status_queue,
                               src_obj_metadata=src_obj_metadata)
      # Need to let _CopyFileToFile return before setting the POSIX attributes.
      if not src_url.IsStream() and not dst_url.IsStream():
        ParseAndSetPOSIXAttributes(dst_url.object_name,
                                   src_obj_metadata,
                                   is_rsync=is_rsync,
                                   preserve_posix=preserve_posix)
      return result


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.now(tz=datetime.timezone.utc).replace(tzinfo=None))

  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.now(tz=datetime.timezone.utc).replace(tzinfo=None))
    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 ItemExistsError(Exception):
  """Exception class for objects that are skipped because they already exist."""
  pass


class SkipUnsupportedObjectError(Exception):
  """Exception for objects skipped because they are an unsupported type."""

  def __init__(self):
    super(SkipUnsupportedObjectError, self).__init__()
    self.unsupported_type = 'Unknown'


class SkipGlacierError(SkipUnsupportedObjectError):
  """Exception for objects skipped because they are an unsupported type."""

  def __init__(self):
    super(SkipGlacierError, self).__init__()
    self.unsupported_type = 'GLACIER'


def GetPathBeforeFinalDir(url, exp_src_url):
  """Returns the path section before the final directory component of the URL.

  This handles cases for file system directories, bucket, and bucket
  subdirectories. Example: for gs://bucket/dir/ we'll return 'gs://bucket',
  and for file://dir we'll return file://

  Args:
    url: StorageUrl representing a filesystem directory, cloud bucket or
         bucket subdir.
    exp_src_url: StorageUrl representing the fully expanded object
        to-be-copied; used for resolving cloud wildcards.

  Returns:
    String name of above-described path, sans final path separator.
  """
  sep = url.delim
  if url.IsFileUrl():
    past_scheme = url.url_string[len('file://'):]
    if past_scheme.find(sep) == -1:
      return 'file://'
    else:
      return 'file://%s' % past_scheme.rstrip(sep).rpartition(sep)[0]
  if url.IsBucket():
    return '%s://' % url.scheme
  # Else it names a bucket subdir.
  path_sans_final_dir = url.url_string.rstrip(sep).rpartition(sep)[0]
  return ResolveWildcardsInPathBeforeFinalDir(path_sans_final_dir, exp_src_url)


def ResolveWildcardsInPathBeforeFinalDir(src_url_path_sans_final_dir,
                                         exp_src_url):
  """Returns the path section for a bucket subdir with wildcards resolved.

  This handles cases for bucket subdirectories where the initial source URL
  contains a wildcard. In this case, src_url must be wildcard-expanded
  before calculating the final directory.

  Example:
    A bucket containing:
      gs://bucket/dir1/subdir/foo
      gs://bucket/dir2/subdir/foo

    and source URL gs://bucket/*/subdir
    and src_url_path_sans_final dir gs://bucket/*

    should yield final path gs://bucket/dir1 or gs://bucket/dir2 according to
    the expanded source URL.

  Args:
    src_url_path_sans_final_dir: URL string with wildcards representing a
        bucket subdir as computed from GetPathBeforeFinalDir.
    exp_src_url: CloudUrl representing the fully expanded object to-be-copied.

  Returns:
    String name of above-described path, sans final path separator.
  """
  if not ContainsWildcard(src_url_path_sans_final_dir):
    return src_url_path_sans_final_dir

  # Parse the expanded source URL, replacing wildcarded
  # portions of the path with what they actually expanded to.
  wildcarded_src_obj_path = StorageUrlFromString(
      src_url_path_sans_final_dir).object_name.split('/')
  expanded_src_obj_path = exp_src_url.object_name.split('/')
  for path_segment_index in range(len(wildcarded_src_obj_path)):
    if ContainsWildcard(wildcarded_src_obj_path[path_segment_index]):
      # The expanded path is guaranteed to be have at least as many path
      # segments as the wildcarded path.
      wildcarded_src_obj_path[path_segment_index] = (
          expanded_src_obj_path[path_segment_index])
  resolved_src_path = '/'.join(wildcarded_src_obj_path)
  final_path_url = exp_src_url.Clone()
  final_path_url.object_name = resolved_src_path
  return final_path_url.url_string


def _GetPartitionInfo(file_size, max_components, default_component_size):
  """Gets info about a file partition for parallel file/object transfers.

  Args:
    file_size: The number of bytes in the file to be partitioned.
    max_components: The maximum number of components that can be composed.
    default_component_size: The size of a component, assuming that
                            max_components is infinite.
  Returns:
    The number of components in the partitioned file, and the size of each
    component (except the last, which will have a different size iff
    file_size != 0 (mod num_components)).
  """
  # num_components = ceil(file_size / default_component_size)
  num_components = DivideAndCeil(file_size, default_component_size)

  # num_components must be in the range [2, max_components]
  num_components = max(min(num_components, max_components), 2)

  # component_size = ceil(file_size / num_components)
  component_size = DivideAndCeil(file_size, num_components)
  return (num_components, component_size)


def _DeleteTempComponentObjectFn(cls, url_to_delete, thread_state=None):
  """Wrapper func to be used with command.Apply to delete temporary objects."""
  gsutil_api = GetCloudApiInstance(cls, thread_state)
  try:
    gsutil_api.DeleteObject(url_to_delete.bucket_name,
                            url_to_delete.object_name,
                            generation=url_to_delete.generation,
                            provider=url_to_delete.scheme)
  except NotFoundException:
    # The temporary object could already be gone if a retry was
    # issued at a lower layer but the original request succeeded.
    # Barring other errors, the top-level command should still report success,
    # so don't raise here.
    pass


def FilterExistingComponents(dst_args, existing_components, bucket_url,
                             gsutil_api):
  """Determines course of action for component objects.

  Given the list of all target objects based on partitioning the file and
  the list of objects that have already been uploaded successfully,
  this function determines which objects should be uploaded, which
  existing components are still valid, and which existing components should
  be deleted.

  Args:
    dst_args: The map of file_name -> PerformParallelUploadFileToObjectArgs
              calculated by partitioning the file.
    existing_components: A list of ObjectFromTracker objects that have been
                         uploaded in the past.
    bucket_url: CloudUrl of the bucket in which the components exist.
    gsutil_api: gsutil Cloud API instance to use for retrieving object metadata.

  Returns:
    components_to_upload: List of components that need to be uploaded.
    uploaded_components: List of components that have already been
                         uploaded and are still valid. Each element of the list
                         contains the dst_url for the uploaded component and
                         its size.
    existing_objects_to_delete: List of components that have already
                                been uploaded, but are no longer valid
                                and are in a versioned bucket, and
                                therefore should be deleted.
  """
  components_to_upload = []
  existing_component_names = [
      component.object_name for component in existing_components
  ]
  for component_name in dst_args:
    if component_name not in existing_component_names:
      components_to_upload.append(dst_args[component_name])

  objects_already_chosen = []

  # Don't reuse any temporary components whose MD5 doesn't match the current
  # MD5 of the corresponding part of the file. If the bucket is versioned,
  # also make sure that we delete the existing temporary version.
  existing_objects_to_delete = []
  uploaded_components = []
  for tracker_object in existing_components:
    if (tracker_object.object_name not in dst_args.keys() or
        tracker_object.object_name in objects_already_chosen):
      # This could happen if the component size has changed. This also serves
      # to handle object names that get duplicated in the tracker file due
      # to people doing things they shouldn't (e.g., overwriting an existing
      # temporary component in a versioned bucket).

      url = bucket_url.Clone()
      url.object_name = tracker_object.object_name
      url.generation = tracker_object.generation
      existing_objects_to_delete.append(url)
      continue

    dst_arg = dst_args[tracker_object.object_name]
    file_part = FilePart(dst_arg.filename, dst_arg.file_start,
                         dst_arg.file_length)
    # TODO: calculate MD5's in parallel when possible.
    content_md5 = CalculateB64EncodedMd5FromContents(file_part)

    try:
      # Get the MD5 of the currently-existing component.
      dst_url = dst_arg.dst_url
      dst_metadata = gsutil_api.GetObjectMetadata(
          dst_url.bucket_name,
          dst_url.object_name,
          generation=dst_url.generation,
          provider=dst_url.scheme,
          fields=['customerEncryption', 'etag', 'md5Hash'])
      cloud_md5 = dst_metadata.md5Hash
    except Exception:  # pylint: disable=broad-except
      # We don't actually care what went wrong - we couldn't retrieve the
      # object to check the MD5, so just upload it again.
      cloud_md5 = None

    if cloud_md5 != content_md5:
      components_to_upload.append(dst_arg)
      objects_already_chosen.append(tracker_object.object_name)
      if tracker_object.generation:
        # If the old object doesn't have a generation (i.e., it isn't in a
        # versioned bucket), then we will just overwrite it anyway.
        invalid_component_with_generation = dst_arg.dst_url.Clone()
        invalid_component_with_generation.generation = tracker_object.generation
        existing_objects_to_delete.append(invalid_component_with_generation)
    else:
      url = dst_arg.dst_url.Clone()
      url.generation = tracker_object.generation
      uploaded_components.append((url, dst_arg.file_length))
      objects_already_chosen.append(tracker_object.object_name)

  if uploaded_components:
    logging.info('Found %d existing temporary components to reuse.',
                 len(uploaded_components))

  return (components_to_upload, uploaded_components, existing_objects_to_delete)