Roll version of gsutil to 4.15.

third_party/gsutil/gslib/commands/perfdiag.py

 # -*- coding: utf-8 -*-
 # Copyright 2012 Google Inc. All Rights Reserved.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
 # You may obtain a copy of the License at
 #
 #     http://www.apache.org/licenses/LICENSE-2.0
 #
 # Unless required by applicable law or agreed to in writing, software
 # 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.
 """Contains the perfdiag gsutil command."""
 
 from __future__ import absolute_import
 
 import calendar
 from collections import defaultdict
+from collections import namedtuple
 import contextlib
 import cStringIO
 import datetime
 import httplib
 import json
 import logging
 import math
 import multiprocessing
 import os
 import random
 import re
 import socket
 import string
 import subprocess
 import tempfile
 import time
 
 import boto
 import boto.gs.connection
 
 import gslib
 from gslib.cloud_api import NotFoundException
 from gslib.cloud_api import ServiceException
-from gslib.cloud_api_helper import GetDownloadSerializationDict
+from gslib.cloud_api_helper import GetDownloadSerializationData
 from gslib.command import Command
 from gslib.command import DummyArgChecker
 from gslib.command_argument import CommandArgument
 from gslib.commands import config
 from gslib.cs_api_map import ApiSelector
 from gslib.exception import CommandException
+from gslib.file_part import FilePart
 from gslib.hashing_helper import CalculateB64EncodedMd5FromContents
 from gslib.storage_url import StorageUrlFromString
 from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
+from gslib.util import CheckFreeSpace
+from gslib.util import DivideAndCeil
 from gslib.util import GetCloudApiInstance
+from gslib.util import GetFileSize
 from gslib.util import GetMaxRetryDelay
 from gslib.util import HumanReadableToBytes
 from gslib.util import IS_LINUX
 from gslib.util import MakeBitsHumanReadable
 from gslib.util import MakeHumanReadable
 from gslib.util import Percentile
 from gslib.util import ResumableThreshold
 
-
 _SYNOPSIS = """
   gsutil perfdiag [-i in.json]
-  gsutil perfdiag [-o out.json] [-n iterations] [-c processes]
-      [-k threads] [-s size] [-t tests] url...
+  gsutil perfdiag [-o out.json] [-n objects] [-c processes]
+      [-k threads] [-p parallelism type] [-y slices] [-s size] [-d directory]
+      [-t tests] url...
 """
 
 _DETAILED_HELP_TEXT = ("""
 <B>SYNOPSIS</B>
 """ + _SYNOPSIS + """
 
 
 <B>DESCRIPTION</B>
   The perfdiag command runs a suite of diagnostic tests for a given Google
   Storage bucket.
@@ skipping 14 matching lines @@
 
 <B>PROVIDING DIAGNOSTIC OUTPUT TO GOOGLE CLOUD STORAGE TEAM</B>
   If the Google Cloud Storage Team asks you to run a performance diagnostic
   please use the following command, and email the output file (output.json)
   to gs-team@google.com:
 
     gsutil perfdiag -o output.json gs://your-bucket
 
 
 <B>OPTIONS</B>
-  -n          Sets the number of iterations performed when downloading and
-              uploading files during latency and throughput tests. Defaults to
-              5.
+  -n          Sets the number of objects to use when downloading and uploading
+              files during tests. Defaults to 5.
 
   -c          Sets the number of processes to use while running throughput
               experiments. The default value is 1.
 
   -k          Sets the number of threads per process to use while running
               throughput experiments. Each process will receive an equal number
               of threads. The default value is 1.
 
-  -s          Sets the size (in bytes) of the test file used to perform read
-              and write throughput tests. The default is 1 MiB. This can also
-              be specified using byte suffixes such as 500K or 1M. Note: these
-              values are interpreted as multiples of 1024 (K=1024, M=1024*1024,
-              etc.)
+              Note: All specified threads and processes will be created, but may
+              not by saturated with work if too few objects (specified with -n)
+              and too few components (specified with -y) are specified.
+
+  -p          Sets the type of parallelism to be used (only applicable when
+              threads or processes are specified and threads * processes > 1).
+              The default is to use fan. Must be one of the following:
+
+              fan
+                 Use one thread per object. This is akin to using gsutil -m cp,
+                 with sliced object download / parallel composite upload
+                 disabled.
+
+              slice
+                 Use Y (specified with -y) threads for each object, transferring
+                 one object at a time. This is akin to using parallel object
+                 download / parallel composite upload, without -m. Sliced
+                 uploads not supported for s3.
+
+              both
+                 Use Y (specified with -y) threads for each object, transferring
+                 multiple objects at a time. This is akin to simultaneously
+                 using sliced object download / parallel composite upload and
+                 gsutil -m cp. Sliced uploads not supported for s3.
+
+  -y          Sets the number of slices to divide each file/object into while
+              transferring data. Only applicable with the slice (or both)
+              parallelism type. The default is 4 slices.
+
+  -s          Sets the size (in bytes) for each of the N (set with -n) objects
+              used in the read and write throughput tests. The default is 1 MiB.
+              This can also be specified using byte suffixes such as 500K or 1M.
+              Note: these values are interpreted as multiples of 1024 (K=1024,
+              M=1024*1024, etc.)
+              Note: If rthru_file or wthru_file are performed, N (set with -n)
+              times as much disk space as specified will be required for the
+              operation.
+
+  -d          Sets the directory to store temporary local files in. If not
+              specified, a default temporary directory will be used.
 
   -t          Sets the list of diagnostic tests to perform. The default is to
-              run all diagnostic tests. Must be a comma-separated list
-              containing one or more of the following:
+              run the lat, rthru, and wthru diagnostic tests. Must be a
+              comma-separated list containing one or more of the following:
 
               lat
-                 Runs N iterations (set with -n) of writing the file,
-                 retrieving its metadata, reading the file, and deleting
-                 the file. Records the latency of each operation.
+                 For N (set with -n) objects, write the object, retrieve its
+                 metadata, read the object, and finally delete the object.
+                 Record the latency of each operation.
 
               list
                  Write N (set with -n) objects to the bucket, record how long
                  it takes for the eventually consistent listing call to return
                  the N objects in its result, delete the N objects, then record
                  how long it takes listing to stop returning the N objects.
                  This test is off by default.
 
               rthru
                  Runs N (set with -n) read operations, with at most C
                  (set with -c) reads outstanding at any given time.
 
+              rthru_file
+                 The same as rthru, but simultaneously writes data to the disk,
+                 to gauge the performance impact of the local disk on downloads.
+
               wthru
                  Runs N (set with -n) write operations, with at most C
                  (set with -c) writes outstanding at any given time.
 
+              wthru_file
+                 The same as wthru, but simultaneously reads data from the disk,
+                 to gauge the performance impact of the local disk on uploads.
+
   -m          Adds metadata to the result JSON file. Multiple -m values can be
               specified. Example:
 
-                  gsutil perfdiag -m "key1:value1" -m "key2:value2" \
-                                  gs://bucketname/
+                  gsutil perfdiag -m "key1:val1" -m "key2:val2" gs://bucketname
 
               Each metadata key will be added to the top-level "metadata"
               dictionary in the output JSON file.
 
   -o          Writes the results of the diagnostic to an output file. The output
               is a JSON file containing system information and performance
               diagnostic results. The file can be read and reported later using
               the -i option.
 
   -i          Reads the JSON output file created using the -o command and prints
@@ skipping 13 matching lines @@
 
 
 <B>NOTE</B>
   The perfdiag command collects system information. It collects your IP address,
   executes DNS queries to Google servers and collects the results, and collects
   network statistics information from the output of netstat -s. It will also
   attempt to connect to your proxy server if you have one configured. None of
   this information will be sent to Google unless you choose to send it.
 """)
 
+FileDataTuple = namedtuple(
+    'FileDataTuple',
+    'size md5 data')
+
+# Describes one object in a fanned download. If need_to_slice is specified as
+# True, the object should be downloaded with the slice strategy. Other field
+# names are the same as documented in PerfDiagCommand.Download.
+FanDownloadTuple = namedtuple(
+    'FanDownloadTuple',
+    'need_to_slice object_name file_name serialization_data')
+
+# Describes one slice in a sliced download.
+# Field names are the same as documented in PerfDiagCommand.Download.
+SliceDownloadTuple = namedtuple(
+    'SliceDownloadTuple',
+    'object_name file_name serialization_data start_byte end_byte')
+
+# Describes one file in a fanned upload. If need_to_slice is specified as
+# True, the file should be uploaded with the slice strategy. Other field
+# names are the same as documented in PerfDiagCommand.Upload.
+FanUploadTuple = namedtuple(
+    'FanUploadTuple',
+    'need_to_slice file_name object_name use_file')
+
+# Describes one slice in a sliced upload.
+# Field names are the same as documented in PerfDiagCommand.Upload.
+SliceUploadTuple = namedtuple(
+    'SliceUploadTuple',
+    'file_name object_name use_file file_start file_size')
+
+# Dict storing file_path:FileDataTuple for each temporary file used by
+# perfdiag. This data should be kept outside of the PerfDiagCommand class
+# since calls to Apply will make copies of all member data.
+temp_file_dict = {}
+
 
 class Error(Exception):
   """Base exception class for this module."""
   pass
 
 
 class InvalidArgument(Error):
   """Raised on invalid arguments to functions."""
   pass
 
 
-def _DownloadWrapper(cls, arg, thread_state=None):
-  cls.Download(arg, thread_state=thread_state)
+def _DownloadObject(cls, args, thread_state=None):
+  """Function argument to apply for performing fanned parallel downloads.
+
+  Args:
+    cls: The calling PerfDiagCommand class instance.
+    args: A FanDownloadTuple object describing this download.
+    thread_state: gsutil Cloud API instance to use for the operation.
+  """
+  cls.gsutil_api = GetCloudApiInstance(cls, thread_state)
+  if args.need_to_slice:
+    cls.PerformSlicedDownload(args.object_name, args.file_name,
+                              args.serialization_data)
+  else:
+    cls.Download(args.object_name, args.file_name, args.serialization_data)
 
 
-def _UploadWrapper(cls, arg, thread_state=None):
-  cls.Upload(arg, thread_state=thread_state)
+def _DownloadSlice(cls, args, thread_state=None):
+  """Function argument to apply for performing sliced downloads.
+
+  Args:
+    cls: The calling PerfDiagCommand class instance.
+    args: A SliceDownloadTuple object describing this download.
+    thread_state: gsutil Cloud API instance to use for the operation.
+  """
+  cls.gsutil_api = GetCloudApiInstance(cls, thread_state)
+  cls.Download(args.object_name, args.file_name, args.serialization_data,
+               args.start_byte, args.end_byte)
 
 
-def _DeleteWrapper(cls, arg, thread_state=None):
-  cls.Delete(arg, thread_state=thread_state)
+def _UploadObject(cls, args, thread_state=None):
+  """Function argument to apply for performing fanned parallel uploads.
+
+  Args:
+    cls: The calling PerfDiagCommand class instance.
+    args: A FanUploadTuple object describing this upload.
+    thread_state: gsutil Cloud API instance to use for the operation.
+  """
+  cls.gsutil_api = GetCloudApiInstance(cls, thread_state)
+  if args.need_to_slice:
+    cls.PerformSlicedUpload(args.file_name, args.object_name, args.use_file)
+  else:
+    cls.Upload(args.file_name, args.object_name, args.use_file)
+
+
+def _UploadSlice(cls, args, thread_state=None):
+  """Function argument to apply for performing sliced parallel uploads.
+
+  Args:
+    cls: The calling PerfDiagCommand class instance.
+    args: A SliceUploadTuple object describing this upload.
+    thread_state: gsutil Cloud API instance to use for the operation.
+  """
+  cls.gsutil_api = GetCloudApiInstance(cls, thread_state)
+  cls.Upload(args.file_name, args.object_name, args.use_file,
+             args.file_start, args.file_size)
+
+
+def _DeleteWrapper(cls, object_name, thread_state=None):
+  """Function argument to apply for performing parallel object deletions.
+
+  Args:
+    cls: The calling PerfDiagCommand class instance.
+    object_name: The object name to delete from the test bucket.
+    thread_state: gsutil Cloud API instance to use for the operation.
+  """
+  cls.gsutil_api = GetCloudApiInstance(cls, thread_state)
+  cls.Delete(object_name)
 
 
 def _PerfdiagExceptionHandler(cls, e):
   """Simple exception handler to allow post-completion status."""
   cls.logger.error(str(e))
 
 
 def _DummyTrackerCallback(_):
   pass
 
 
 class DummyFile(object):
   """A dummy, file-like object that throws away everything written to it."""
 
   def write(self, *args, **kwargs):  # pylint: disable=invalid-name
     pass
 
+  def close(self):  # pylint: disable=invalid-name
+    pass
+
 
 # Many functions in perfdiag re-define a temporary function based on a
 # variable from a loop, resulting in a false positive from the linter.
 # pylint: disable=cell-var-from-loop
 class PerfDiagCommand(Command):
   """Implementation of gsutil perfdiag command."""
 
   # Command specification. See base class for documentation.
   command_spec = Command.CreateCommandSpec(
       'perfdiag',
       command_name_aliases=['diag', 'diagnostic', 'perf', 'performance'],
       usage_synopsis=_SYNOPSIS,
       min_args=0,
       max_args=1,
-      supported_sub_args='n:c:k:s:t:m:i:o:',
+      supported_sub_args='n:c:k:p:y:s:d:t:m:i:o:',
       file_url_ok=False,
       provider_url_ok=False,
       urls_start_arg=0,
       gs_api_support=[ApiSelector.XML, ApiSelector.JSON],
       gs_default_api=ApiSelector.JSON,
       argparse_arguments=[
           CommandArgument.MakeNCloudBucketURLsArgument(1)
       ]
   )
   # Help specification. See help_provider.py for documentation.
   help_spec = Command.HelpSpec(
       help_name='perfdiag',
       help_name_aliases=[],
       help_type='command_help',
       help_one_line_summary='Run performance diagnostic',
       help_text=_DETAILED_HELP_TEXT,
       subcommand_help_text={},
   )
 
   # Byte sizes to use for latency testing files.
   # TODO: Consider letting the user specify these sizes with a configuration
   # parameter.
-  test_file_sizes = (
+  test_lat_file_sizes = (
       0,  # 0 bytes
       1024,  # 1 KiB
       102400,  # 100 KiB
       1048576,  # 1 MiB
   )
 
   # Test names.
   RTHRU = 'rthru'
+  RTHRU_FILE = 'rthru_file'
   WTHRU = 'wthru'
+  WTHRU_FILE = 'wthru_file'
   LAT = 'lat'
   LIST = 'list'
 
+  # Parallelism strategies.
+  FAN = 'fan'
+  SLICE = 'slice'
+  BOTH = 'both'
+
   # List of all diagnostic tests.
-  ALL_DIAG_TESTS = (RTHRU, WTHRU, LAT, LIST)
+  ALL_DIAG_TESTS = (RTHRU, RTHRU_FILE, WTHRU, WTHRU_FILE, LAT, LIST)
+
   # List of diagnostic tests to run by default.
   DEFAULT_DIAG_TESTS = (RTHRU, WTHRU, LAT)
 
+  # List of parallelism strategies.
+  PARALLEL_STRATEGIES = (FAN, SLICE, BOTH)
+
   # Google Cloud Storage XML API endpoint host.
   XML_API_HOST = boto.config.get(
       'Credentials', 'gs_host', boto.gs.connection.GSConnection.DefaultHost)
   # Google Cloud Storage XML API endpoint port.
   XML_API_PORT = boto.config.get('Credentials', 'gs_port', 80)
 
   # Maximum number of times to retry requests on 5xx errors.
   MAX_SERVER_ERROR_RETRIES = 5
   # Maximum number of times to retry requests on more serious errors like
   # the socket breaking.
@@ skipping 33 matching lines @@
     """
     self.logger.debug('Running command: %s', cmd)
     stderr = subprocess.PIPE if mute_stderr else None
     p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=stderr)
     (stdoutdata, _) = p.communicate()
     if raise_on_error and p.returncode:
       raise CommandException("Received non-zero return code (%d) from "
                              "subprocess '%s'." % (p.returncode, ' '.join(cmd)))
     return stdoutdata if return_output else p.returncode
 
+  def _WarnIfLargeData(self):
+    """Outputs a warning message if a large amount of data is being used."""
+    if self.num_objects * self.thru_filesize > HumanReadableToBytes('2GiB'):
+      self.logger.info('This is a large operation, and could take a while.')
+
+  def _MakeTempFile(self, file_size=0, mem_metadata=False,
+                    mem_data=False, prefix='gsutil_test_file'):
+    """Creates a temporary file of the given size and returns its path.
+
+    Args:
+      file_size: The size of the temporary file to create.
+      mem_metadata: If true, store md5 and file size in memory at
+                    temp_file_dict[fpath].md5, tempfile_data[fpath].file_size.
+      mem_data: If true, store the file data in memory at
+                temp_file_dict[fpath].data
+      prefix: The prefix to use for the temporary file. Defaults to
+              gsutil_test_file.
+
+    Returns:
+      The file path of the created temporary file.
+    """
+    fd, fpath = tempfile.mkstemp(suffix='.bin', prefix=prefix,
+                                 dir=self.directory, text=False)
+    with os.fdopen(fd, 'wb') as fp:
+      random_bytes = os.urandom(min(file_size,
+                                    self.MAX_UNIQUE_RANDOM_BYTES))
+      total_bytes_written = 0
+      while total_bytes_written < file_size:
+        num_bytes = min(self.MAX_UNIQUE_RANDOM_BYTES,
+                        file_size - total_bytes_written)
+        fp.write(random_bytes[:num_bytes])
+        total_bytes_written += num_bytes
+
+    if mem_metadata or mem_data:
+      with open(fpath, 'rb') as fp:
+        file_size = GetFileSize(fp) if mem_metadata else None
+        md5 = CalculateB64EncodedMd5FromContents(fp) if mem_metadata else None
+        data = fp.read() if mem_data else None
+        temp_file_dict[fpath] = FileDataTuple(file_size, md5, data)
+
+    self.temporary_files.add(fpath)
+    return fpath
+
   def _SetUp(self):
     """Performs setup operations needed before diagnostics can be run."""
 
     # Stores test result data.
     self.results = {}
-    # List of test files in a temporary location on disk for latency ops.
-    self.latency_files = []
-    # List of test objects to clean up in the test bucket.
-    self.test_object_names = set()
-    # Maps each test file path to its size in bytes.
-    self.file_sizes = {}
-    # Maps each test file to its contents as a string.
-    self.file_contents = {}
-    # Maps each test file to its MD5 hash.
-    self.file_md5s = {}
+    # Set of file paths for local temporary files.
+    self.temporary_files = set()
+    # Set of names for test objects that exist in the test bucket.
+    self.temporary_objects = set()
     # Total number of HTTP requests made.
     self.total_requests = 0
     # Total number of HTTP 5xx errors.
     self.request_errors = 0
     # Number of responses, keyed by response code.
     self.error_responses_by_code = defaultdict(int)
     # Total number of socket errors.
     self.connection_breaks = 0
+    # Boolean to prevent doing cleanup twice.
+    self.teardown_completed = False
 
-    def _MakeFile(file_size):
-      """Creates a temporary file of the given size and returns its path."""
-      fd, fpath = tempfile.mkstemp(suffix='.bin', prefix='gsutil_test_file',
-                                   text=False)
-      self.file_sizes[fpath] = file_size
-      random_bytes = os.urandom(min(file_size, self.MAX_UNIQUE_RANDOM_BYTES))
-      total_bytes = 0
-      file_contents = ''
-      while total_bytes < file_size:
-        num_bytes = min(self.MAX_UNIQUE_RANDOM_BYTES, file_size - total_bytes)
-        file_contents += random_bytes[:num_bytes]
-        total_bytes += num_bytes
-      self.file_contents[fpath] = file_contents
-      with os.fdopen(fd, 'wb') as f:
-        f.write(self.file_contents[fpath])
-      with open(fpath, 'rb') as f:
-        self.file_md5s[fpath] = CalculateB64EncodedMd5FromContents(f)
-      return fpath
+    # Create files for latency test.
+    if self.LAT in self.diag_tests:
+      self.latency_files = []
+      for file_size in self.test_lat_file_sizes:
+        fpath = self._MakeTempFile(file_size, mem_metadata=True, mem_data=True)
+        self.latency_files.append(fpath)
 
-    # Create files for latency tests.
-    for file_size in self.test_file_sizes:
-      fpath = _MakeFile(file_size)
-      self.latency_files.append(fpath)
+    # Create files for throughput tests.
+    if self.diag_tests.intersection(
+        (self.RTHRU, self.WTHRU, self.RTHRU_FILE, self.WTHRU_FILE)):
+      # Create a file for warming up the TCP connection.
+      self.tcp_warmup_file = self._MakeTempFile(
+          5 * 1024 * 1024, mem_metadata=True, mem_data=True)
 
-    # Creating a file for warming up the TCP connection.
-    self.tcp_warmup_file = _MakeFile(5 * 1024 * 1024)  # 5 Mebibytes.
-    # Remote file to use for TCP warmup.
-    self.tcp_warmup_remote_file = (str(self.bucket_url) +
-                                   os.path.basename(self.tcp_warmup_file))
+      # For in memory tests, throughput tests transfer the same object N times
+      # instead of creating N objects, in order to avoid excessive memory usage.
+      if self.diag_tests.intersection((self.RTHRU, self.WTHRU)):
+        self.mem_thru_file_name = self._MakeTempFile(
+            self.thru_filesize, mem_metadata=True, mem_data=True)
+        self.mem_thru_object_name = os.path.basename(self.mem_thru_file_name)
 
-    # Local file on disk for write throughput tests.
-    self.thru_local_file = _MakeFile(self.thru_filesize)
+      # For tests that use disk I/O, it is necessary to create N objects in
+      # in order to properly measure the performance impact of seeks.
+      if self.diag_tests.intersection((self.RTHRU_FILE, self.WTHRU_FILE)):
+        # List of file names and corresponding object names to use for file
+        # throughput tests.
+        self.thru_file_names = []
+        self.thru_object_names = []
+
+        free_disk_space = CheckFreeSpace(self.directory)
+        if free_disk_space >= self.thru_filesize * self.num_objects:
+          self.logger.info('\nCreating %d local files each of size %s.'
+                           % (self.num_objects,
+                              MakeHumanReadable(self.thru_filesize)))
+          self._WarnIfLargeData()
+          for _ in range(self.num_objects):
+            file_name = self._MakeTempFile(self.thru_filesize,
+                                           mem_metadata=True)
+            self.thru_file_names.append(file_name)
+            self.thru_object_names.append(os.path.basename(file_name))
+        else:
+          raise CommandException(
+              'Not enough free disk space for throughput files: '
+              '%s of disk space required, but only %s available.'
+              % (MakeHumanReadable(self.thru_filesize * self.num_objects),
+                 MakeHumanReadable(free_disk_space)))
 
     # Dummy file buffer to use for downloading that goes nowhere.
     self.discard_sink = DummyFile()
 
+    # Filter out misleading progress callback output and the incorrect
+    # suggestion to use gsutil -m perfdiag.
+    self.logger.addFilter(self._PerfdiagFilter())
+
   def _TearDown(self):
     """Performs operations to clean things up after performing diagnostics."""
-    for fpath in self.latency_files + [self.thru_local_file,
-                                       self.tcp_warmup_file]:
+    if not self.teardown_completed:
+      temp_file_dict.clear()
+
       try:
-        os.remove(fpath)
+        for fpath in self.temporary_files:
+          os.remove(fpath)
+        if self.delete_directory:
+          os.rmdir(self.directory)
       except OSError:
         pass
 
-    for object_name in self.test_object_names:
-
-      def _Delete():
-        try:
-          self.gsutil_api.DeleteObject(self.bucket_url.bucket_name,
-                                       object_name,
-                                       provider=self.provider)
-        except NotFoundException:
-          pass
-
-      self._RunOperation(_Delete)
+      if self.threads > 1 or self.processes > 1:
+        args = [obj for obj in self.temporary_objects]
+        self.Apply(_DeleteWrapper, args, _PerfdiagExceptionHandler,
+                   arg_checker=DummyArgChecker,
+                   parallel_operations_override=True,
+                   process_count=self.processes, thread_count=self.threads)
+      else:
+        for object_name in self.temporary_objects:
+          self.Delete(object_name)
+    self.teardown_completed = True
 
   @contextlib.contextmanager
   def _Time(self, key, bucket):
     """A context manager that measures time.
 
     A context manager that prints a status message before and after executing
     the inner command and times how long the inner command takes. Keeps track of
     the timing, aggregated by the given key.
 
     Args:
@@ skipping 54 matching lines @@
             break
         else:
           self.connection_breaks += 1
     return return_val
 
   def _RunLatencyTests(self):
     """Runs latency tests."""
     # Stores timing information for each category of operation.
     self.results['latency'] = defaultdict(list)
 
-    for i in range(self.num_iterations):
+    for i in range(self.num_objects):
       self.logger.info('\nRunning latency iteration %d...', i+1)
       for fpath in self.latency_files:
+        file_data = temp_file_dict[fpath]
         url = self.bucket_url.Clone()
         url.object_name = os.path.basename(fpath)
-        file_size = self.file_sizes[fpath]
+        file_size = file_data.size
         readable_file_size = MakeHumanReadable(file_size)
 
         self.logger.info(
             "\nFile of size %s located on disk at '%s' being diagnosed in the "
             "cloud at '%s'.", readable_file_size, fpath, url)
 
         upload_target = StorageUrlToUploadObjectMetadata(url)
 
         def _Upload():
-          io_fp = cStringIO.StringIO(self.file_contents[fpath])
+          io_fp = cStringIO.StringIO(file_data.data)
           with self._Time('UPLOAD_%d' % file_size, self.results['latency']):
             self.gsutil_api.UploadObject(
                 io_fp, upload_target, size=file_size, provider=self.provider,
                 fields=['name'])
         self._RunOperation(_Upload)
 
         def _Metadata():
           with self._Time('METADATA_%d' % file_size, self.results['latency']):
             return self.gsutil_api.GetObjectMetadata(
                 url.bucket_name, url.object_name,
                 provider=self.provider, fields=['name', 'contentType',
                                                 'mediaLink', 'size'])
         # Download will get the metadata first if we don't pass it in.
         download_metadata = self._RunOperation(_Metadata)
-        serialization_dict = GetDownloadSerializationDict(download_metadata)
-        serialization_data = json.dumps(serialization_dict)
+        serialization_data = GetDownloadSerializationData(download_metadata)
 
         def _Download():
           with self._Time('DOWNLOAD_%d' % file_size, self.results['latency']):
             self.gsutil_api.GetObjectMedia(
                 url.bucket_name, url.object_name, self.discard_sink,
                 provider=self.provider, serialization_data=serialization_data)
         self._RunOperation(_Download)
 
         def _Delete():
           with self._Time('DELETE_%d' % file_size, self.results['latency']):
             self.gsutil_api.DeleteObject(url.bucket_name, url.object_name,
                                          provider=self.provider)
         self._RunOperation(_Delete)
 
-  class _CpFilter(logging.Filter):
+  class _PerfdiagFilter(logging.Filter):
 
     def filter(self, record):
-      # Used to prevent cp._LogCopyOperation from spewing output from
-      # subprocesses about every iteration.
+      # Used to prevent unnecessary output when using multiprocessing.
       msg = record.getMessage()
       return not (('Copying file:///' in msg) or ('Copying gs://' in msg) or
-                  ('Computing CRC' in msg))
+                  ('Computing CRC' in msg) or ('gsutil -m perfdiag' in msg))
 
   def _PerfdiagExceptionHandler(self, e):
     """Simple exception handler to allow post-completion status."""
     self.logger.error(str(e))
 
-  def _RunReadThruTests(self):
+  def PerformFannedDownload(self, need_to_slice, object_names, file_names,
+                            serialization_data):
+    """Performs a parallel download of multiple objects using the fan strategy.
+
+    Args:
+      need_to_slice: If True, additionally apply the slice strategy to each
+                     object in object_names.
+      object_names: A list of object names to be downloaded. Each object must
+                    already exist in the test bucket.
+      file_names: A list, corresponding by index to object_names, of file names
+                  for downloaded data. If None, discard downloaded data.
+      serialization_data: A list, corresponding by index to object_names,
+                          of serialization data for each object.
+    """
+    args = []
+    for i in range(len(object_names)):
+      file_name = file_names[i] if file_names else None
+      args.append(FanDownloadTuple(
+          need_to_slice, object_names[i], file_name,
+          serialization_data[i]))
+    self.Apply(_DownloadObject, args, _PerfdiagExceptionHandler,
+               ('total_requests', 'request_errors'),
+               arg_checker=DummyArgChecker, parallel_operations_override=True,
+               process_count=self.processes, thread_count=self.threads)
+
+  def PerformSlicedDownload(self, object_name, file_name, serialization_data):
+    """Performs a download of an object using the slice strategy.
+
+    Args:
+      object_name: The name of the object to download.
+      file_name: The name of the file to download data to, or None if data
+                 should be discarded.
+      serialization_data: The serialization data for the object.
+    """
+    if file_name:
+      with open(file_name, 'ab') as fp:
+        fp.truncate(self.thru_filesize)
+    component_size = DivideAndCeil(self.thru_filesize, self.num_slices)
+    args = []
+    for i in range(self.num_slices):
+      start_byte = i * component_size
+      end_byte = min((i + 1) * (component_size) - 1, self.thru_filesize - 1)
+      args.append(SliceDownloadTuple(object_name, file_name, serialization_data,
+                                     start_byte, end_byte))
+    self.Apply(_DownloadSlice, args, _PerfdiagExceptionHandler,
+               ('total_requests', 'request_errors'),
+               arg_checker=DummyArgChecker, parallel_operations_override=True,
+               process_count=self.processes, thread_count=self.threads)
+
+  def PerformFannedUpload(self, need_to_slice, file_names, object_names,
+                          use_file):
+    """Performs a parallel upload of multiple files using the fan strategy.
+
+    The metadata for file_name should be present in temp_file_dict prior
+    to calling. Also, the data for file_name should be present in temp_file_dict
+    if use_file is specified as False.
+
+    Args:
+      need_to_slice: If True, additionally apply the slice strategy to each
+                     file in file_names.
+      file_names: A list of file names to be uploaded.
+      object_names: A list, corresponding by by index to file_names, of object
+                    names to upload data to.
+      use_file: If true, use disk I/O, otherwise read upload data from memory.
+    """
+    args = []
+    for i in range(len(file_names)):
+      args.append(FanUploadTuple(
+          need_to_slice, file_names[i], object_names[i], use_file))
+    self.Apply(_UploadObject, args, _PerfdiagExceptionHandler,
+               ('total_requests', 'request_errors'),
+               arg_checker=DummyArgChecker, parallel_operations_override=True,
+               process_count=self.processes, thread_count=self.threads)
+
+  def PerformSlicedUpload(self, file_name, object_name, use_file):
+    """Performs a parallel upload of a file using the slice strategy.
+
+    The metadata for file_name should be present in temp_file_dict prior
+    to calling. Also, the data from for file_name should be present in
+    temp_file_dict if use_file is specified as False.
+
+    Args:
+      file_name: The name of the file to upload.
+      object_name: The name of the object to upload to.
+      use_file: If true, use disk I/O, otherwise read upload data from memory.
+    """
+    # Divide the file into components.
+    component_size = DivideAndCeil(self.thru_filesize, self.num_slices)
+    component_object_names = (
+        [object_name + str(i) for i in range(self.num_slices)])
+
+    args = []
+    for i in range(self.num_slices):
+      component_start = i * component_size
+      component_size = min(component_size,
+                           temp_file_dict[file_name].size - component_start)
+      args.append(SliceUploadTuple(file_name, component_object_names[i],
+                                   use_file, component_start, component_size))
+
+    # Upload the components in parallel.
+    try:
+      self.Apply(_UploadSlice, args, _PerfdiagExceptionHandler,
+                 ('total_requests', 'request_errors'),
+                 arg_checker=DummyArgChecker, parallel_operations_override=True,
+                 process_count=self.processes, thread_count=self.threads)
+
+      # Compose the components into an object.
+      request_components = []
+      for i in range(self.num_slices):
+        src_obj_metadata = (
+            apitools_messages.ComposeRequest.SourceObjectsValueListEntry(
+                name=component_object_names[i]))
+        request_components.append(src_obj_metadata)
+
+      dst_obj_metadata = apitools_messages.Object()
+      dst_obj_metadata.name = object_name
+      dst_obj_metadata.bucket = self.bucket_url.bucket_name
+      def _Compose():
+        self.gsutil_api.ComposeObject(request_components, dst_obj_metadata,
+                                      provider=self.provider)
+      self._RunOperation(_Compose)
+    finally:
+      # Delete the temporary components.
+      self.Apply(_DeleteWrapper, component_object_names,
+                 _PerfdiagExceptionHandler,
+                 ('total_requests', 'request_errors'),
+                 arg_checker=DummyArgChecker, parallel_operations_override=True,
+                 process_count=self.processes, thread_count=self.threads)
+
+  def _RunReadThruTests(self, use_file=False):
     """Runs read throughput tests."""
+    test_name = 'read_throughput_file' if use_file else 'read_throughput'
+    file_io_string = 'with file I/O' if use_file else ''
     self.logger.info(
-        '\nRunning read throughput tests (%s iterations of size %s)' %
-        (self.num_iterations, MakeHumanReadable(self.thru_filesize)))
-
-    self.results['read_throughput'] = {'file_size': self.thru_filesize,
-                                       'num_times': self.num_iterations,
-                                       'processes': self.processes,
-                                       'threads': self.threads}
-
-    # Copy the TCP warmup file.
-    warmup_url = self.bucket_url.Clone()
-    warmup_url.object_name = os.path.basename(self.tcp_warmup_file)
-    warmup_target = StorageUrlToUploadObjectMetadata(warmup_url)
-    self.test_object_names.add(warmup_url.object_name)
-
-    def _Upload1():
-      self.gsutil_api.UploadObject(
-          cStringIO.StringIO(self.file_contents[self.tcp_warmup_file]),
-          warmup_target, provider=self.provider, fields=['name'])
-    self._RunOperation(_Upload1)
-
-    # Copy the file to remote location before reading.
-    thru_url = self.bucket_url.Clone()
-    thru_url.object_name = os.path.basename(self.thru_local_file)
-    thru_target = StorageUrlToUploadObjectMetadata(thru_url)
-    thru_target.md5Hash = self.file_md5s[self.thru_local_file]
-    self.test_object_names.add(thru_url.object_name)
-
-    # Get the mediaLink here so that we can pass it to download.
-    def _Upload2():
-      return self.gsutil_api.UploadObject(
-          cStringIO.StringIO(self.file_contents[self.thru_local_file]),
-          thru_target, provider=self.provider, size=self.thru_filesize,
-          fields=['name', 'mediaLink', 'size'])
-
-    # Get the metadata for the object so that we are just measuring performance
-    # on the actual bytes transfer.
-    download_metadata = self._RunOperation(_Upload2)
-    serialization_dict = GetDownloadSerializationDict(download_metadata)
-    serialization_data = json.dumps(serialization_dict)
-
+        '\nRunning read throughput tests %s (%s objects of size %s)' %
+        (file_io_string, self.num_objects,
+         MakeHumanReadable(self.thru_filesize)))
+    self._WarnIfLargeData()
+
+    self.results[test_name] = {'file_size': self.thru_filesize,
+                               'processes': self.processes,
+                               'threads': self.threads,
+                               'parallelism': self.parallel_strategy
+                              }
+
+    # Copy the file(s) to the test bucket, and also get the serialization data
+    # so that we can pass it to download.
+    if use_file:
+      # For test with file I/O use N files on disk to preserve seek performance.
+      file_names = self.thru_file_names
+      object_names = self.thru_object_names
+      serialization_data = []
+      for i in range(self.num_objects):
+        self.temporary_objects.add(self.thru_object_names[i])
+        if self.WTHRU_FILE in self.diag_tests:
+          # If we ran the WTHRU_FILE test, then the objects already exist.
+          obj_metadata = self.gsutil_api.GetObjectMetadata(
+              self.bucket_url.bucket_name, self.thru_object_names[i],
+              fields=['size', 'mediaLink'], provider=self.bucket_url.scheme)
+        else:
+          obj_metadata = self.Upload(self.thru_file_names[i],
+                                     self.thru_object_names[i], use_file)
+
+        # File overwrite causes performance issues with sliced downloads.
+        # Delete the file and reopen it for download. This matches what a real
+        # download would look like.
+        os.unlink(self.thru_file_names[i])
+        open(self.thru_file_names[i], 'ab').close()
+        serialization_data.append(GetDownloadSerializationData(obj_metadata))
+    else:
+      # For in-memory test only use one file but copy it num_objects times, to
+      # allow scalability in num_objects.
+      self.temporary_objects.add(self.mem_thru_object_name)
+      obj_metadata = self.Upload(self.mem_thru_file_name,
+                                 self.mem_thru_object_name, use_file)
+      file_names = None
+      object_names = [self.mem_thru_object_name] * self.num_objects
+      serialization_data = (
+          [GetDownloadSerializationData(obj_metadata)] * self.num_objects)
+
+    # Warmup the TCP connection.
+    warmup_obj_name = os.path.basename(self.tcp_warmup_file)
+    self.temporary_objects.add(warmup_obj_name)
+    self.Upload(self.tcp_warmup_file, warmup_obj_name)
+    self.Download(warmup_obj_name)
+
+    t0 = time.time()
     if self.processes == 1 and self.threads == 1:
-
-      # Warm up the TCP connection.
-      def _Warmup():
-        self.gsutil_api.GetObjectMedia(warmup_url.bucket_name,
-                                       warmup_url.object_name,
-                                       self.discard_sink,
-                                       provider=self.provider)
-      self._RunOperation(_Warmup)
-
-      times = []
-
-      def _Download():
-        t0 = time.time()
-        self.gsutil_api.GetObjectMedia(
-            thru_url.bucket_name, thru_url.object_name, self.discard_sink,
-            provider=self.provider, serialization_data=serialization_data)
-        t1 = time.time()
-        times.append(t1 - t0)
-      for _ in range(self.num_iterations):
-        self._RunOperation(_Download)
-      time_took = sum(times)
+      for i in range(self.num_objects):
+        file_name = file_names[i] if use_file else None
+        self.Download(object_names[i], file_name, serialization_data[i])
     else:
-      args = ([(thru_url.bucket_name, thru_url.object_name, serialization_data)]
-              * self.num_iterations)
-      self.logger.addFilter(self._CpFilter())
-
-      t0 = time.time()
-      self.Apply(_DownloadWrapper,
-                 args,
-                 _PerfdiagExceptionHandler,
-                 arg_checker=DummyArgChecker,
-                 parallel_operations_override=True,
-                 process_count=self.processes,
-                 thread_count=self.threads)
-      t1 = time.time()
-      time_took = t1 - t0
-
-    total_bytes_copied = self.thru_filesize * self.num_iterations
+      if self.parallel_strategy in (self.FAN, self.BOTH):
+        need_to_slice = (self.parallel_strategy == self.BOTH)
+        self.PerformFannedDownload(need_to_slice, object_names, file_names,
+                                   serialization_data)
+      elif self.parallel_strategy == self.SLICE:
+        for i in range(self.num_objects):
+          file_name = file_names[i] if use_file else None
+          self.PerformSlicedDownload(
+              object_names[i], file_name, serialization_data[i])
+    t1 = time.time()
+
+    time_took = t1 - t0
+    total_bytes_copied = self.thru_filesize * self.num_objects
     bytes_per_second = total_bytes_copied / time_took
 
-    self.results['read_throughput']['time_took'] = time_took
-    self.results['read_throughput']['total_bytes_copied'] = total_bytes_copied
-    self.results['read_throughput']['bytes_per_second'] = bytes_per_second
-
-  def _RunWriteThruTests(self):
+    self.results[test_name]['time_took'] = time_took
+    self.results[test_name]['total_bytes_copied'] = total_bytes_copied
+    self.results[test_name]['bytes_per_second'] = bytes_per_second
+
+  def _RunWriteThruTests(self, use_file=False):
     """Runs write throughput tests."""
+    test_name = 'write_throughput_file' if use_file else 'write_throughput'
+    file_io_string = 'with file I/O' if use_file else ''
     self.logger.info(
-        '\nRunning write throughput tests (%s iterations of size %s)' %
-        (self.num_iterations, MakeHumanReadable(self.thru_filesize)))
-
-    self.results['write_throughput'] = {'file_size': self.thru_filesize,
-                                        'num_copies': self.num_iterations,
-                                        'processes': self.processes,
-                                        'threads': self.threads}
-
-    warmup_url = self.bucket_url.Clone()
-    warmup_url.object_name = os.path.basename(self.tcp_warmup_file)
-    warmup_target = StorageUrlToUploadObjectMetadata(warmup_url)
-    self.test_object_names.add(warmup_url.object_name)
-
-    thru_url = self.bucket_url.Clone()
-    thru_url.object_name = os.path.basename(self.thru_local_file)
-    thru_target = StorageUrlToUploadObjectMetadata(thru_url)
-    thru_tuples = []
-    for i in xrange(self.num_iterations):
-      # Create a unique name for each uploaded object.  Otherwise,
-      # the XML API would fail when trying to non-atomically get metadata
-      # for the object that gets blown away by the overwrite.
-      remote_object_name = thru_target.name + str(i)
-      self.test_object_names.add(remote_object_name)
-      thru_tuples.append(UploadObjectTuple(thru_target.bucket,
-                                           remote_object_name,
-                                           filepath=self.thru_local_file))
-
+        '\nRunning write throughput tests %s (%s objects of size %s)' %
+        (file_io_string, self.num_objects,
+         MakeHumanReadable(self.thru_filesize)))
+    self._WarnIfLargeData()
+
+    self.results[test_name] = {'file_size': self.thru_filesize,
+                               'processes': self.processes,
+                               'threads': self.threads,
+                               'parallelism': self.parallel_strategy}
+
+    # Warmup the TCP connection.
+    warmup_obj_name = os.path.basename(self.tcp_warmup_file)
+    self.temporary_objects.add(warmup_obj_name)
+    self.Upload(self.tcp_warmup_file, warmup_obj_name)
+
+    if use_file:
+      # For test with file I/O use N files on disk to preserve seek performance.
+      file_names = self.thru_file_names
+      object_names = self.thru_object_names
+    else:
+      # For in-memory test only use one file but copy it num_objects times, to
+      # allow for scalability in num_objects.
+      file_names = [self.mem_thru_file_name] * self.num_objects
+      object_names = (
+          [self.mem_thru_object_name + str(i) for i in range(self.num_objects)])
+
+    for object_name in object_names:
+      self.temporary_objects.add(object_name)
+
+    t0 = time.time()
     if self.processes == 1 and self.threads == 1:
-      # Warm up the TCP connection.
-      def _Warmup():
-        self.gsutil_api.UploadObject(
-            cStringIO.StringIO(self.file_contents[self.tcp_warmup_file]),
-            warmup_target, provider=self.provider, size=self.thru_filesize,
-            fields=['name'])
-      self._RunOperation(_Warmup)
-
-      times = []
-
-      for i in xrange(self.num_iterations):
-        thru_tuple = thru_tuples[i]
-        def _Upload():
-          """Uploads the write throughput measurement object."""
-          upload_target = apitools_messages.Object(
-              bucket=thru_tuple.bucket_name, name=thru_tuple.object_name,
-              md5Hash=thru_tuple.md5)
-          io_fp = cStringIO.StringIO(self.file_contents[self.thru_local_file])
-          t0 = time.time()
-          if self.thru_filesize < ResumableThreshold():
-            self.gsutil_api.UploadObject(
-                io_fp, upload_target, provider=self.provider,
-                size=self.thru_filesize, fields=['name'])
-          else:
-            self.gsutil_api.UploadObjectResumable(
-                io_fp, upload_target, provider=self.provider,
-                size=self.thru_filesize, fields=['name'],
-                tracker_callback=_DummyTrackerCallback)
-
-          t1 = time.time()
-          times.append(t1 - t0)
-
-        self._RunOperation(_Upload)
-      time_took = sum(times)
-
+      for i in range(self.num_objects):
+        self.Upload(file_names[i], object_names[i], use_file)
     else:
-      args = thru_tuples
-      t0 = time.time()
-      self.Apply(_UploadWrapper,
-                 args,
-                 _PerfdiagExceptionHandler,
-                 arg_checker=DummyArgChecker,
-                 parallel_operations_override=True,
-                 process_count=self.processes,
-                 thread_count=self.threads)
-      t1 = time.time()
-      time_took = t1 - t0
-
-    total_bytes_copied = self.thru_filesize * self.num_iterations
+      if self.parallel_strategy in (self.FAN, self.BOTH):
+        need_to_slice = (self.parallel_strategy == self.BOTH)
+        self.PerformFannedUpload(need_to_slice, file_names, object_names,
+                                 use_file)
+      elif self.parallel_strategy == self.SLICE:
+        for i in range(self.num_objects):
+          self.PerformSlicedUpload(file_names[i], object_names[i], use_file)
+    t1 = time.time()
+
+    time_took = t1 - t0
+    total_bytes_copied = self.thru_filesize * self.num_objects
     bytes_per_second = total_bytes_copied / time_took
 
-    self.results['write_throughput']['time_took'] = time_took
-    self.results['write_throughput']['total_bytes_copied'] = total_bytes_copied
-    self.results['write_throughput']['bytes_per_second'] = bytes_per_second
+    self.results[test_name]['time_took'] = time_took
+    self.results[test_name]['total_bytes_copied'] = total_bytes_copied
+    self.results[test_name]['bytes_per_second'] = bytes_per_second
 
   def _RunListTests(self):
     """Runs eventual consistency listing latency tests."""
-    self.results['listing'] = {'num_files': self.num_iterations}
-
-    # Generate N random object names to put in the bucket.
+    self.results['listing'] = {'num_files': self.num_objects}
+
+    # Generate N random objects to put into the bucket.
     list_prefix = 'gsutil-perfdiag-list-'
+    list_fpaths = []
     list_objects = []
-    for _ in xrange(self.num_iterations):
-      list_object_name = u'%s%s' % (list_prefix, os.urandom(20).encode('hex'))
-      self.test_object_names.add(list_object_name)
-      list_objects.append(list_object_name)
+    args = []
+    for _ in xrange(self.num_objects):
+      fpath = self._MakeTempFile(0, mem_data=True, mem_metadata=True,
+                                 prefix=list_prefix)
+      list_fpaths.append(fpath)
+      object_name = os.path.basename(fpath)
+      list_objects.append(object_name)
+      args.append(FanUploadTuple(False, fpath, object_name, False))
+      self.temporary_objects.add(object_name)
 
     # Add the objects to the bucket.
     self.logger.info(
-        '\nWriting %s objects for listing test...', self.num_iterations)
-    empty_md5 = CalculateB64EncodedMd5FromContents(cStringIO.StringIO(''))
-    args = [
-        UploadObjectTuple(self.bucket_url.bucket_name, name, md5=empty_md5,
-                          contents='') for name in list_objects]
-    self.Apply(_UploadWrapper, args, _PerfdiagExceptionHandler,
+        '\nWriting %s objects for listing test...', self.num_objects)
+
+    self.Apply(_UploadObject, args, _PerfdiagExceptionHandler,
                arg_checker=DummyArgChecker)
 
     list_latencies = []
     files_seen = []
     total_start_time = time.time()
     expected_objects = set(list_objects)
     found_objects = set()
 
     def _List():
       """Lists and returns objects in the bucket. Also records latency."""
       t0 = time.time()
       objects = list(self.gsutil_api.ListObjects(
-          self.bucket_url.bucket_name, prefix=list_prefix, delimiter='/',
+          self.bucket_url.bucket_name, delimiter='/',
           provider=self.provider, fields=['items/name']))
       t1 = time.time()
       list_latencies.append(t1 - t0)
       return set([obj.data.name for obj in objects])
 
     self.logger.info(
         'Listing bucket %s waiting for %s objects to appear...',
-        self.bucket_url.bucket_name, self.num_iterations)
+        self.bucket_url.bucket_name, self.num_objects)
     while expected_objects - found_objects:
       def _ListAfterUpload():
         names = _List()
         found_objects.update(names & expected_objects)
         files_seen.append(len(found_objects))
       self._RunOperation(_ListAfterUpload)
       if expected_objects - found_objects:
         if time.time() - total_start_time > self.MAX_LISTING_WAIT_TIME:
           self.logger.warning('Maximum time reached waiting for listing.')
           break
     total_end_time = time.time()
 
     self.results['listing']['insert'] = {
         'num_listing_calls': len(list_latencies),
         'list_latencies': list_latencies,
         'files_seen_after_listing': files_seen,
         'time_took': total_end_time - total_start_time,
     }
 
+    args = [object_name for object_name in list_objects]
     self.logger.info(
-        'Deleting %s objects for listing test...', self.num_iterations)
+        'Deleting %s objects for listing test...', self.num_objects)
     self.Apply(_DeleteWrapper, args, _PerfdiagExceptionHandler,
                arg_checker=DummyArgChecker)
 
     self.logger.info(
         'Listing bucket %s waiting for %s objects to disappear...',
-        self.bucket_url.bucket_name, self.num_iterations)
+        self.bucket_url.bucket_name, self.num_objects)
     list_latencies = []
     files_seen = []
     total_start_time = time.time()
     found_objects = set(list_objects)
     while found_objects:
       def _ListAfterDelete():
         names = _List()
         found_objects.intersection_update(names)
         files_seen.append(len(found_objects))
       self._RunOperation(_ListAfterDelete)
       if found_objects:
         if time.time() - total_start_time > self.MAX_LISTING_WAIT_TIME:
           self.logger.warning('Maximum time reached waiting for listing.')
           break
     total_end_time = time.time()
 
     self.results['listing']['delete'] = {
         'num_listing_calls': len(list_latencies),
         'list_latencies': list_latencies,
         'files_seen_after_listing': files_seen,
         'time_took': total_end_time - total_start_time,
     }
 
-  def Upload(self, thru_tuple, thread_state=None):
-    gsutil_api = GetCloudApiInstance(self, thread_state)
+  def Upload(self, file_name, object_name, use_file=False, file_start=0,
+             file_size=None):
+    """Performs an upload to the test bucket.
 
-    md5hash = thru_tuple.md5
-    contents = thru_tuple.contents
-    if thru_tuple.filepath:
-      md5hash = self.file_md5s[thru_tuple.filepath]
-      contents = self.file_contents[thru_tuple.filepath]
-
-    upload_target = apitools_messages.Object(
-        bucket=thru_tuple.bucket_name, name=thru_tuple.object_name,
-        md5Hash=md5hash)
-    file_size = len(contents)
-    if file_size < ResumableThreshold():
-      gsutil_api.UploadObject(
-          cStringIO.StringIO(contents), upload_target,
-          provider=self.provider, size=file_size, fields=['name'])
-    else:
-      gsutil_api.UploadObjectResumable(
-          cStringIO.StringIO(contents), upload_target,
-          provider=self.provider, size=file_size, fields=['name'],
-          tracker_callback=_DummyTrackerCallback)
-
-  def Download(self, download_tuple, thread_state=None):
-    """Downloads a file.
+    The file is uploaded to the bucket referred to by self.bucket_url, and has
+    name object_name.
 
     Args:
-      download_tuple: (bucket name, object name, serialization data for object).
-      thread_state: gsutil Cloud API instance to use for the download.
+      file_name: The path to the local file, and the key to its entry in
+                 temp_file_dict.
+      object_name: The name of the remote object.
+      use_file: If true, use disk I/O, otherwise read everything from memory.
+      file_start: The first byte in the file to upload to the object.
+                  (only should be specified for sliced uploads)
+      file_size: The size of the file to upload.
+                 (only should be specified for sliced uploads)
+
+    Returns:
+      Uploaded Object Metadata.
     """
-    gsutil_api = GetCloudApiInstance(self, thread_state)
-    gsutil_api.GetObjectMedia(
-        download_tuple[0], download_tuple[1], self.discard_sink,
-        provider=self.provider, serialization_data=download_tuple[2])
+    fp = None
+    if file_size is None:
+      file_size = temp_file_dict[file_name].size
 
-  def Delete(self, thru_tuple, thread_state=None):
-    gsutil_api = thread_state or self.gsutil_api
-    gsutil_api.DeleteObject(
-        thru_tuple.bucket_name, thru_tuple.object_name, provider=self.provider)
+    upload_url = self.bucket_url.Clone()
+    upload_url.object_name = object_name
+    upload_target = StorageUrlToUploadObjectMetadata(upload_url)
+
+    try:
+      if use_file:
+        fp = FilePart(file_name, file_start, file_size)
+      else:
+        data = temp_file_dict[file_name].data[file_start:file_start+file_size]
+        fp = cStringIO.StringIO(data)
+
+      def _InnerUpload():
+        if file_size < ResumableThreshold():
+          return self.gsutil_api.UploadObject(
+              fp, upload_target, provider=self.provider, size=file_size,
+              fields=['name', 'mediaLink', 'size'])
+        else:
+          return self.gsutil_api.UploadObjectResumable(
+              fp, upload_target, provider=self.provider, size=file_size,
+              fields=['name', 'mediaLink', 'size'],
+              tracker_callback=_DummyTrackerCallback)
+      return self._RunOperation(_InnerUpload)
+    finally:
+      if fp:
+        fp.close()
+
+  def Download(self, object_name, file_name=None, serialization_data=None,
+               start_byte=0, end_byte=None):
+    """Downloads an object from the test bucket.
+
+    Args:
+      object_name: The name of the object (in the test bucket) to download.
+      file_name: Optional file name to write downloaded data to. If None,
+                 downloaded data is discarded immediately.
+      serialization_data: Optional serialization data, used so that we don't
+                          have to get the metadata before downloading.
+      start_byte: The first byte in the object to download.
+                  (only should be specified for sliced downloads)
+      end_byte: The last byte in the object to download.
+                (only should be specified for sliced downloads)
+    """
+    fp = None
+    try:
+      if file_name is not None:
+        fp = open(file_name, 'r+b')
+        fp.seek(start_byte)
+      else:
+        fp = self.discard_sink
+
+      def _InnerDownload():
+        self.gsutil_api.GetObjectMedia(
+            self.bucket_url.bucket_name, object_name, fp,
+            provider=self.provider, start_byte=start_byte, end_byte=end_byte,
+            serialization_data=serialization_data)
+      self._RunOperation(_InnerDownload)
+    finally:
+      if fp:
+        fp.close()
+
+  def Delete(self, object_name):
+    """Deletes an object from the test bucket.
+
+    Args:
+      object_name: The name of the object to delete.
+    """
+    try:
+      def _InnerDelete():
+        self.gsutil_api.DeleteObject(self.bucket_url.bucket_name, object_name,
+                                     provider=self.provider)
+      self._RunOperation(_InnerDelete)
+    except NotFoundException:
+      pass
 
   def _GetDiskCounters(self):
     """Retrieves disk I/O statistics for all disks.
 
     Adapted from the psutil module's psutil._pslinux.disk_io_counters:
       http://code.google.com/p/psutil/source/browse/trunk/psutil/_pslinux.py
 
     Originally distributed under under a BSD license.
     Original Copyright (c) 2009, Jay Loden, Dave Daeschler, Giampaolo Rodola.
 
@@ skipping 105 matching lines @@
       self.logger.info('DNS lookups are disallowed in this environment, so '
                        'some information is not included in this perfdiag run.')
 
     # Get the local IP address from socket lib.
     try:
       sysinfo['ip_address'] = socket.gethostbyname(socket.gethostname())
     except socket_errors:
       sysinfo['ip_address'] = ''
     # Record the temporary directory used since it can affect performance, e.g.
     # when on a networked filesystem.
-    sysinfo['tempdir'] = tempfile.gettempdir()
+    sysinfo['tempdir'] = self.directory
 
     # Produces an RFC 2822 compliant GMT timestamp.
     sysinfo['gmt_timestamp'] = time.strftime('%a, %d %b %Y %H:%M:%S +0000',
                                              time.gmtime())
 
     # Execute a CNAME lookup on Google DNS to find what Google server
     # it's routing to.
     cmd = ['nslookup', '-type=CNAME', self.XML_API_HOST]
     try:
       nslookup_cname_output = self._Exec(cmd, return_output=True)
@@ skipping 191 matching lines @@
           print 'Delete'.rjust(9), '',
           print MakeHumanReadable(numbytes).rjust(9), '',
           self._DisplayStats(trials)
 
     if 'write_throughput' in self.results:
       print
       print '-' * 78
       print 'Write Throughput'.center(78)
       print '-' * 78
       write_thru = self.results['write_throughput']
-      print 'Copied a %s file %d times for a total transfer size of %s.' % (
+      print 'Copied %s %s file(s) for a total transfer size of %s.' % (
+          self.num_objects,
           MakeHumanReadable(write_thru['file_size']),
-          write_thru['num_copies'],
           MakeHumanReadable(write_thru['total_bytes_copied']))
       print 'Write throughput: %s/s.' % (
           MakeBitsHumanReadable(write_thru['bytes_per_second'] * 8))
+      print 'Parallelism strategy: %s' % write_thru['parallelism']
+
+    if 'write_throughput_file' in self.results:
+      print
+      print '-' * 78
+      print 'Write Throughput With File I/O'.center(78)
+      print '-' * 78
+      write_thru_file = self.results['write_throughput_file']
+      print 'Copied %s %s file(s) for a total transfer size of %s.' % (
+          self.num_objects,
+          MakeHumanReadable(write_thru_file['file_size']),
+          MakeHumanReadable(write_thru_file['total_bytes_copied']))
+      print 'Write throughput: %s/s.' % (
+          MakeBitsHumanReadable(write_thru_file['bytes_per_second'] * 8))
+      print 'Parallelism strategy: %s' % write_thru_file['parallelism']
 
     if 'read_throughput' in self.results:
       print
       print '-' * 78
       print 'Read Throughput'.center(78)
       print '-' * 78
       read_thru = self.results['read_throughput']
-      print 'Copied a %s file %d times for a total transfer size of %s.' % (
+      print 'Copied %s %s file(s) for a total transfer size of %s.' % (
+          self.num_objects,
           MakeHumanReadable(read_thru['file_size']),
-          read_thru['num_times'],
           MakeHumanReadable(read_thru['total_bytes_copied']))
       print 'Read throughput: %s/s.' % (
           MakeBitsHumanReadable(read_thru['bytes_per_second'] * 8))
+      print 'Parallelism strategy: %s' % read_thru['parallelism']
+
+    if 'read_throughput_file' in self.results:
+      print
+      print '-' * 78
+      print 'Read Throughput With File I/O'.center(78)
+      print '-' * 78
+      read_thru_file = self.results['read_throughput_file']
+      print 'Copied %s %s file(s) for a total transfer size of %s.' % (
+          self.num_objects,
+          MakeHumanReadable(read_thru_file['file_size']),
+          MakeHumanReadable(read_thru_file['total_bytes_copied']))
+      print 'Read throughput: %s/s.' % (
+          MakeBitsHumanReadable(read_thru_file['bytes_per_second'] * 8))
+      print 'Parallelism strategy: %s' % read_thru_file['parallelism']
 
     if 'listing' in self.results:
       print
       print '-' * 78
       print 'Listing'.center(78)
       print '-' * 78
 
       listing = self.results['listing']
       insert = listing['insert']
       delete = listing['delete']
@@ skipping 172 matching lines @@
       val = int(val)
       if val < 1:
         raise CommandException(msg)
       return val
     except ValueError:
       raise CommandException(msg)
 
   def _ParseArgs(self):
     """Parses arguments for perfdiag command."""
     # From -n.
-    self.num_iterations = 5
+    self.num_objects = 5
     # From -c.
     self.processes = 1
     # From -k.
     self.threads = 1
+    # From -p
+    self.parallel_strategy = None
+    # From -y
+    self.num_slices = 4
     # From -s.
     self.thru_filesize = 1048576
+    # From -d.
+    self.directory = tempfile.gettempdir()
+    # Keep track of whether or not to delete the directory upon completion.
+    self.delete_directory = False
     # From -t.
-    self.diag_tests = self.DEFAULT_DIAG_TESTS
+    self.diag_tests = set(self.DEFAULT_DIAG_TESTS)
     # From -o.
     self.output_file = None
     # From -i.
     self.input_file = None
     # From -m.
     self.metadata_keys = {}
 
     if self.sub_opts:
       for o, a in self.sub_opts:
         if o == '-n':
-          self.num_iterations = self._ParsePositiveInteger(
+          self.num_objects = self._ParsePositiveInteger(
               a, 'The -n parameter must be a positive integer.')
         if o == '-c':
           self.processes = self._ParsePositiveInteger(
               a, 'The -c parameter must be a positive integer.')
         if o == '-k':
           self.threads = self._ParsePositiveInteger(
               a, 'The -k parameter must be a positive integer.')
+        if o == '-p':
+          if a.lower() in self.PARALLEL_STRATEGIES:
+            self.parallel_strategy = a.lower()
+          else:
+            raise CommandException(
+                "'%s' is not a valid parallelism strategy." % a)
+        if o == '-y':
+          self.num_slices = self._ParsePositiveInteger(
+              a, 'The -y parameter must be a positive integer.')
         if o == '-s':
           try:
             self.thru_filesize = HumanReadableToBytes(a)
           except ValueError:
             raise CommandException('Invalid -s parameter.')
-          if self.thru_filesize > (20 * 1024 ** 3):  # Max 20 GiB.
-            raise CommandException(
-                'Maximum throughput file size parameter (-s) is 20 GiB.')
+        if o == '-d':
+          self.directory = a
+          if not os.path.exists(self.directory):
+            self.delete_directory = True
+            os.makedirs(self.directory)
         if o == '-t':
-          self.diag_tests = []
+          self.diag_tests = set()
           for test_name in a.strip().split(','):
             if test_name.lower() not in self.ALL_DIAG_TESTS:
               raise CommandException("List of test names (-t) contains invalid "
                                      "test name '%s'." % test_name)
-            self.diag_tests.append(test_name)
+            self.diag_tests.add(test_name)
         if o == '-m':
           pieces = a.split(':')
           if len(pieces) != 2:
             raise CommandException(
                 "Invalid metadata key-value combination '%s'." % a)
           key, value = pieces
           self.metadata_keys[key] = value
         if o == '-o':
           self.output_file = os.path.abspath(a)
         if o == '-i':
           self.input_file = os.path.abspath(a)
           if not os.path.isfile(self.input_file):
             raise CommandException("Invalid input file (-i): '%s'." % a)
           try:
             with open(self.input_file, 'r') as f:
               self.results = json.load(f)
               self.logger.info("Read input file: '%s'.", self.input_file)
           except ValueError:
             raise CommandException("Could not decode input file (-i): '%s'." %
                                    a)
           return
+
+    # If parallelism is specified, default parallelism strategy to fan.
+    if (self.processes > 1 or self.threads > 1) and not self.parallel_strategy:
+      self.parallel_strategy = self.FAN
+    elif self.processes == 1 and self.threads == 1 and self.parallel_strategy:
+      raise CommandException(
+          'Cannot specify parallelism strategy (-p) without also specifying '
+          'multiple threads and/or processes (-c and/or -k).')
+
     if not self.args:
       self.RaiseWrongNumberOfArgumentsException()
 
     self.bucket_url = StorageUrlFromString(self.args[0])
     self.provider = self.bucket_url.scheme
-    if not (self.bucket_url.IsCloudUrl() and self.bucket_url.IsBucket()):
+    if not self.bucket_url.IsCloudUrl() and self.bucket_url.IsBucket():
       raise CommandException('The perfdiag command requires a URL that '
                              'specifies a bucket.\n"%s" is not '
                              'valid.' % self.args[0])
+
+    if (self.thru_filesize > HumanReadableToBytes('2GiB') and
+        (self.RTHRU in self.diag_tests or self.WTHRU in self.diag_tests)):
+      raise CommandException(
+          'For in-memory tests maximum file size is 2GiB. For larger file '
+          'sizes, specify rthru_file and/or wthru_file with the -t option.')
+
+    perform_slice = self.parallel_strategy in (self.SLICE, self.BOTH)
+    slice_not_available = (
+        self.provider == 's3' and self.diag_tests.intersection(self.WTHRU,
+                                                               self.WTHRU_FILE))
+    if perform_slice and slice_not_available:
+      raise CommandException('Sliced uploads are not available for s3. '
+                             'Use -p fan or sequential uploads for s3.')
+
     # Ensure the bucket exists.
     self.gsutil_api.GetBucket(self.bucket_url.bucket_name,
                               provider=self.bucket_url.scheme,
                               fields=['id'])
     self.exceptions = [httplib.HTTPException, socket.error, socket.gaierror,
                        socket.timeout, httplib.BadStatusLine,
                        ServiceException]
 
   # Command entry point.
   def RunCommand(self):
     """Called by gsutil when the command is being invoked."""
     self._ParseArgs()
 
     if self.input_file:
       self._DisplayResults()
       return 0
 
     # We turn off retries in the underlying boto library because the
     # _RunOperation function handles errors manually so it can count them.
     boto.config.set('Boto', 'num_retries', '0')
 
     self.logger.info(
         'Number of iterations to run: %d\n'
         'Base bucket URI: %s\n'
         'Number of processes: %d\n'
         'Number of threads: %d\n'
+        'Parallelism strategy: %s\n'
         'Throughput file size: %s\n'
         'Diagnostics to run: %s',
-        self.num_iterations,
+        self.num_objects,
         self.bucket_url,
         self.processes,
         self.threads,
+        self.parallel_strategy,
         MakeHumanReadable(self.thru_filesize),
         (', '.join(self.diag_tests)))
 
     try:
       self._SetUp()
 
       # Collect generic system info.
       self._CollectSysInfo()
       # Collect netstat info and disk counters before tests (and again later).
       netstat_output = self._GetTcpStats()
       if netstat_output:
         self.results['sysinfo']['netstat_start'] = netstat_output
       if IS_LINUX:
         self.results['sysinfo']['disk_counters_start'] = self._GetDiskCounters()
       # Record bucket URL.
       self.results['bucket_uri'] = str(self.bucket_url)
       self.results['json_format'] = 'perfdiag'
       self.results['metadata'] = self.metadata_keys
 
       if self.LAT in self.diag_tests:
         self._RunLatencyTests()
       if self.RTHRU in self.diag_tests:
         self._RunReadThruTests()
+      # Run WTHRU_FILE before RTHRU_FILE. If data is created in WTHRU_FILE it
+      # will be used in RTHRU_FILE to save time and bandwidth.
+      if self.WTHRU_FILE in self.diag_tests:
+        self._RunWriteThruTests(use_file=True)
+      if self.RTHRU_FILE in self.diag_tests:
+        self._RunReadThruTests(use_file=True)
       if self.WTHRU in self.diag_tests:
         self._RunWriteThruTests()
       if self.LIST in self.diag_tests:
         self._RunListTests()
 
       # Collect netstat info and disk counters after tests.
       netstat_output = self._GetTcpStats()
       if netstat_output:
         self.results['sysinfo']['netstat_end'] = netstat_output
       if IS_LINUX:
         self.results['sysinfo']['disk_counters_end'] = self._GetDiskCounters()
 
       self.results['total_requests'] = self.total_requests
       self.results['request_errors'] = self.request_errors
       self.results['error_responses_by_code'] = self.error_responses_by_code
       self.results['connection_breaks'] = self.connection_breaks
       self.results['gsutil_version'] = gslib.VERSION
       self.results['boto_version'] = boto.__version__
 
+      self._TearDown()
       self._DisplayResults()
     finally:
       # TODO: Install signal handlers so this is performed in response to a
       # terminating signal; consider multi-threaded object deletes during
       # cleanup so it happens quickly.
       self._TearDown()
 
     return 0
 
 
-class UploadObjectTuple(object):
-  """Picklable tuple with necessary metadata for an insert object call."""
-
-  def __init__(self, bucket_name, object_name, filepath=None, md5=None,
-               contents=None):
-    """Create an upload tuple.
-
-    Args:
-      bucket_name: Name of the bucket to upload to.
-      object_name: Name of the object to upload to.
-      filepath: A file path located in self.file_contents and self.file_md5s.
-      md5: The MD5 hash of the object being uploaded.
-      contents: The contents of the file to be uploaded.
-
-    Note: (contents + md5) and filepath are mutually exlusive. You may specify
-          one or the other, but not both.
-    Note: If one of contents or md5 are specified, they must both be specified.
-
-    Raises:
-      InvalidArgument: if the arguments are invalid.
-    """
-    self.bucket_name = bucket_name
-    self.object_name = object_name
-    self.filepath = filepath
-    self.md5 = md5
-    self.contents = contents
-    if filepath and (md5 or contents is not None):
-      raise InvalidArgument(
-          'Only one of filepath or (md5 + contents) may be specified.')
-    if not filepath and (not md5 or contents is None):
-      raise InvalidArgument(
-          'Both md5 and contents must be specified.')
-
-
 def StorageUrlToUploadObjectMetadata(storage_url):
   if storage_url.IsCloudUrl() and storage_url.IsObject():
     upload_target = apitools_messages.Object()
     upload_target.name = storage_url.object_name
     upload_target.bucket = storage_url.bucket_name
     return upload_target
   else:
     raise CommandException('Non-cloud URL upload target %s was created in '
                            'perfdiag implemenation.' % storage_url)