def _DoParallelCompositeUpload(fp, src_url, dst_url, dst_obj_metadata,
                               canned_acl, file_size, preconditions, gsutil_api,
                               command_obj, copy_exception_handler, logger):
  start_time = time.time()
  dst_bucket_url = StorageUrlFromString(dst_url.bucket_url_string)
  api_selector = gsutil_api.GetApiSelector(provider=dst_url.scheme)

  encryption_tuple, encryption_key_sha256 = GetEncryptionTupleAndSha256Hash()

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

  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 = CreateLock()

  # Get the set of all components that should be uploaded.
  dst_args = _PartitionFile(
      fp, file_size, src_url, dst_obj_metadata.contentType, canned_acl,
      dst_bucket_url, random_prefix, tracker_file_name, tracker_file_lock,
      encryption_key_sha256=encryption_key_sha256)

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

  # 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=True, should_return_results=True)
  uploaded_components = []
  for cp_result in cp_results:
    uploaded_components.append(cp_result[2])
  components = uploaded_components + existing_components

  if len(components) == len(dst_args):
    # Only try to compose if all of the components were uploaded successfully.

    def _GetComponentNumber(component):
      return int(component.object_name[component.object_name.rfind('_')+1:])
    # 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_tuple)

    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=True)
    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
Uploads a local file to a cloud object using parallel composite upload.
dataset/ETHPy150Open GoogleCloudPlatform/gsutil/gslib/copy_helper.py/_DoParallelCompositeUpload