@@ -161,11 +161,13 @@ class Config(object):

     use_mime_magic = True

     mime_type = u""

     enable_multipart = True

-    multipart_chunk_size_mb = 15    # MB

-    multipart_max_chunks = 10000    # Maximum chunks on AWS S3, could be different on other S3-compatible APIs

-    #- minimum size to use multipart remote s3-to-s3 copy with byte range is 5gb

-    #multipart_copy_size = (5 * 1024 * 1024 * 1024) - 1

-    multipart_copy_size = 5 * 1024 * 1024 * 1024

+    # Chunk size is at the same time the chunk size and the threshold

+    multipart_chunk_size_mb = 15    # MiB

+    # Maximum chunk size for s3-to-s3 copy is 5 GiB.

+    # But, use a lot lower value by default

+    multipart_copy_chunk_size_mb = 2 * 1024

+    # Maximum chunks on AWS S3, could be different on other S3-compatible APIs

+    multipart_max_chunks = 10000

     # List of checks to be performed for 'sync'

     sync_checks = ['size', 'md5']   # 'weak-timestamp'

     # List of compiled REGEXPs

@@ -6,24 +6,43 @@

 from __future__ import absolute_import

-import os

 import sys

-from stat import ST_SIZE

 from logging import debug, info, warning, error

-from .Utils import getTextFromXml, getTreeFromXml, formatSize, unicodise, deunicodise, calculateChecksum, parseNodes, encode_to_s3

+from .Exceptions import ParameterError

+from .S3Uri import S3UriS3

+from .Utils import (getTextFromXml, getTreeFromXml, formatSize,

+                    calculateChecksum, parseNodes)

-class MultiPartUpload(object):

-    MIN_CHUNK_SIZE_MB = 5       # 5MB

-    MAX_CHUNK_SIZE_MB = 5120    # 5GB

-    MAX_FILE_SIZE = 42949672960 # 5TB

+class MultiPartUpload(object):

+    """Supports MultiPartUpload and MultiPartUpload(Copy) operation"""

+    MIN_CHUNK_SIZE_MB = 5        # 5MB

+    MAX_CHUNK_SIZE_MB = 5120     # 5GB

+    MAX_FILE_SIZE = 42949672960  # 5TB

-    def __init__(self, s3, file_stream, uri, headers_baseline=None):

+    def __init__(self, s3, src, dst_uri, headers_baseline=None,

+                 src_size=None):

         self.s3 = s3

-        self.file_stream = file_stream

-        self.uri = uri

+        self.file_stream = None

+        self.src_uri = None

+        self.src_size = src_size

+        self.dst_uri = dst_uri

         self.parts = {}

         self.headers_baseline = headers_baseline or {}

+

+        if isinstance(src, S3UriS3):

+            # Source is the uri of an object to s3-to-s3 copy with multipart.

+            self.src_uri = src

+            if not src_size:

+                raise ParameterError("Source size is missing for "

+                                     "MultipartUploadCopy operation")

+            c_size = self.s3.config.multipart_copy_chunk_size_mb * 1024 * 1024

+        else:

+            # Source is a file_stream to upload

+            self.file_stream = src

+            c_size = self.s3.config.multipart_chunk_size_mb * 1024 * 1024

+

+        self.chunk_size = c_size

         self.upload_id = self.initiate_multipart_upload()

     def get_parts_information(self, uri, upload_id):

@@ -33,7 +52,10 @@ class MultiPartUpload(object):

         parts = dict()

         for elem in parseNodes(tree):

             try:

-                parts[int(elem['PartNumber'])] = {'checksum': elem['ETag'], 'size': elem['Size']}

+                parts[int(elem['PartNumber'])] = {

+                    'checksum': elem['ETag'],

+                    'size': elem['Size']

+                }

             except KeyError:

                 pass

@@ -50,7 +72,11 @@ class MultiPartUpload(object):

                 info("mp_path: %s, object: %s" % (mp_path, uri.object()))

                 if mp_path == uri.object():

                     if upload_id:

-                        raise ValueError("More than one UploadId for URI %s.  Disable multipart upload, or use\n %s multipart %s\nto list the Ids, then pass a unique --upload-id into the put command." % (uri, sys.argv[0], uri))

+                        raise ValueError(

+                            "More than one UploadId for URI %s.  Disable "

+                            "multipart upload, or use\n %s multipart %s\n"

+                            "to list the Ids, then pass a unique --upload-id "

+                            "into the put command." % (uri, sys.argv[0], uri))

                     upload_id = mp_upload_id

             except KeyError:

                 pass

@@ -65,14 +91,14 @@ class MultiPartUpload(object):

         if self.s3.config.upload_id:

             self.upload_id = self.s3.config.upload_id

         elif self.s3.config.put_continue:

-            self.upload_id = self.get_unique_upload_id(self.uri)

+            self.upload_id = self.get_unique_upload_id(self.dst_uri)

         else:

             self.upload_id = ""

         if not self.upload_id:

-            request = self.s3.create_request("OBJECT_POST", uri = self.uri,

-                                             headers = self.headers_baseline,

-                                             uri_params = {'uploads': None})

+            request = self.s3.create_request("OBJECT_POST", uri=self.dst_uri,

+                                             headers=self.headers_baseline,

+                                             uri_params={'uploads': None})

             response = self.s3.send_request(request)

             data = response["data"]

             self.upload_id = getTextFromXml(data, "UploadId")

@@ -86,97 +112,178 @@ class MultiPartUpload(object):

         TODO use num_processes to thread it

         """

         if not self.upload_id:

-            raise RuntimeError("Attempting to use a multipart upload that has not been initiated.")

+            raise ParameterError("Attempting to use a multipart upload that "

+                                 "has not been initiated.")

-        self.chunk_size = self.s3.config.multipart_chunk_size_mb * 1024 * 1024

-        filename = self.file_stream.stream_name

+        remote_statuses = {}

-        if filename != u"<stdin>":

-                size_left = file_size = os.stat(deunicodise(filename))[ST_SIZE]

-                nr_parts = file_size // self.chunk_size + (file_size % self.chunk_size and 1)

-                debug("MultiPart: Uploading %s in %d parts" % (filename, nr_parts))

+        if self.src_uri:

+            filename = self.src_uri.uri()

+            # Continue is not possible with multipart copy

         else:

-            debug("MultiPart: Uploading from %s" % filename)

+            filename = self.file_stream.stream_name

-        remote_statuses = dict()

         if self.s3.config.put_continue:

-            remote_statuses = self.get_parts_information(self.uri, self.upload_id)

+            remote_statuses = self.get_parts_information(self.dst_uri,

+                                                         self.upload_id)

         if extra_label:

             extra_label = u' ' + extra_label

+        labels = {

+            'source' : filename,

+            'destination' : self.dst_uri.uri(),

+        }

+

         seq = 1

-        if filename != u"<stdin>":

+

+        if self.src_size:

+            size_left = self.src_size

+            nr_parts = self.src_size // self.chunk_size \

+                + (self.src_size % self.chunk_size and 1)

+            debug("MultiPart: Uploading %s in %d parts" % (filename, nr_parts))

+

             while size_left > 0:

                 offset = self.chunk_size * (seq - 1)

-                current_chunk_size = min(file_size - offset, self.chunk_size)

+                current_chunk_size = min(self.src_size - offset,

+                                         self.chunk_size)

                 size_left -= current_chunk_size

-                labels = {

-                    'source' : filename,

-                    'destination' : self.uri.uri(),

-                    'extra' : "[part %d of %d, %s]%s" % (seq, nr_parts, "%d%sB" % formatSize(current_chunk_size, human_readable = True), extra_label)

-                }

-                try:

-                    self.upload_part(seq, offset, current_chunk_size, labels, remote_status = remote_statuses.get(seq))

-                except:

-                    error(u"\nUpload of '%s' part %d failed. Use\n  %s abortmp %s %s\nto abort the upload, or\n  %s --upload-id %s put ...\nto continue the upload."

-                          % (filename, seq, sys.argv[0], self.uri, self.upload_id, sys.argv[0], self.upload_id))

-                    raise

-                seq += 1

-        else:

-            while True:

-                buffer = self.file_stream.read(self.chunk_size)

-                offset = 0 # send from start of the buffer

-                current_chunk_size = len(buffer)

-                labels = {

-                    'source' : filename,

-                    'destination' : self.uri.uri(),

-                    'extra' : "[part %d, %s]" % (seq, "%d%sB" % formatSize(current_chunk_size, human_readable = True))

-                }

-                if len(buffer) == 0: # EOF

-                    break

+                labels['extra'] = "[part %d of %d, %s]%s" % (

+                    seq, nr_parts, "%d%sB" % formatSize(current_chunk_size,

+                                                        human_readable=True),

+                    extra_label)

                 try:

-                    self.upload_part(seq, offset, current_chunk_size, labels, buffer, remote_status = remote_statuses.get(seq))

+                    if self.file_stream:

+                        self.upload_part(

+                            seq, offset, current_chunk_size, labels,

+                            remote_status=remote_statuses.get(seq))

+                    else:

+                        self.copy_part(

+                            seq, offset, current_chunk_size, labels,

+                            remote_status=remote_statuses.get(seq))

                 except:

-                    error(u"\nUpload of '%s' part %d failed. Use\n  %s abortmp %s %s\nto abort, or\n  %s --upload-id %s put ...\nto continue the upload."

-                          % (filename, seq, sys.argv[0], self.uri, self.upload_id, sys.argv[0], self.upload_id))

+                    error(u"\nUpload of '%s' part %d failed. Use\n  "

+                          "%s abortmp %s %s\nto abort the upload, or\n  "

+                          "%s --upload-id %s put ...\nto continue the upload."

+                          % (filename, seq, sys.argv[0], self.dst_uri,

+                             self.upload_id, sys.argv[0], self.upload_id))

                     raise

                 seq += 1

+            debug("MultiPart: Upload finished: %d parts", seq - 1)

+            return

+

+

+        # Else -> Case of u"<stdin>" source

+        debug("MultiPart: Uploading from %s" % filename)

+        while True:

+            buffer = self.file_stream.read(self.chunk_size)

+            offset = 0 # send from start of the buffer

+            current_chunk_size = len(buffer)

+            labels['extra'] = "[part %d of %d, %s]%s" % (

+                seq, nr_parts,

+                "%d%sB" % formatSize(current_chunk_size,

+                                     human_readable=True),

+                extra_label)

+            if not buffer:

+                # EOF

+                break

+            try:

+                self.upload_part(seq, offset, current_chunk_size, labels,

+                                 buffer,

+                                 remote_status=remote_statuses.get(seq))

+            except:

+                error(u"\nUpload of '%s' part %d failed. Use\n  "

+                      "%s abortmp %s %s\nto abort, or\n  "

+                      "%s --upload-id %s put ...\nto continue the upload."

+                      % (filename, seq, sys.argv[0], self.dst_uri,

+                         self.upload_id, sys.argv[0], self.upload_id))

+                raise

+            seq += 1

+

         debug("MultiPart: Upload finished: %d parts", seq - 1)

-    def upload_part(self, seq, offset, chunk_size, labels, buffer = '', remote_status = None):

+    def upload_part(self, seq, offset, chunk_size, labels, buffer='',

+                    remote_status=None):

         """

         Upload a file chunk

         http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadUploadPart.html

         """

         # TODO implement Content-MD5

-        debug("Uploading part %i of %r (%s bytes)" % (seq, self.upload_id, chunk_size))

+        debug("Uploading part %i of %r (%s bytes)" % (seq, self.upload_id,

+                                                      chunk_size))

         if remote_status is not None:

             if int(remote_status['size']) == chunk_size:

-                checksum = calculateChecksum(buffer, self.file_stream, offset, chunk_size, self.s3.config.send_chunk)

+                checksum = calculateChecksum(buffer, self.file_stream, offset,

+                                             chunk_size,

+                                             self.s3.config.send_chunk)

                 remote_checksum = remote_status['checksum'].strip('"\'')

                 if remote_checksum == checksum:

-                    warning("MultiPart: size and md5sum match for %s part %d, skipping." % (self.uri, seq))

+                    warning("MultiPart: size and md5sum match for %s part %d, "

+                            "skipping." % (self.dst_uri, seq))

                     self.parts[seq] = remote_status['checksum']

-                    return

+                    return None

                 else:

-                    warning("MultiPart: checksum (%s vs %s) does not match for %s part %d, reuploading."

-                            % (remote_checksum, checksum, self.uri, seq))

+                    warning("MultiPart: checksum (%s vs %s) does not match for"

+                            " %s part %d, reuploading."

+                            % (remote_checksum, checksum, self.dst_uri, seq))

             else:

-                warning("MultiPart: size (%d vs %d) does not match for %s part %d, reuploading."

-                        % (int(remote_status['size']), chunk_size, self.uri, seq))

+                warning("MultiPart: size (%d vs %d) does not match for %s part"

+                        " %d, reuploading." % (int(remote_status['size']),

+                                               chunk_size, self.dst_uri, seq))

-        headers = { "content-length": str(chunk_size) }

-        query_string_params = {'partNumber':'%s' % seq,

+        headers = {"content-length": str(chunk_size)}

+        query_string_params = {'partNumber': '%s' % seq,

                                'uploadId': self.upload_id}

-        request = self.s3.create_request("OBJECT_PUT", uri = self.uri,

-                                         headers = headers,

-                                         uri_params = query_string_params)

-        response = self.s3.send_file(request, self.file_stream, labels, buffer, offset = offset, chunk_size = chunk_size)

+        request = self.s3.create_request("OBJECT_PUT", uri=self.dst_uri,

+                                         headers=headers,

+                                         uri_params=query_string_params)

+        response = self.s3.send_file(request, self.file_stream, labels, buffer,

+                                     offset=offset, chunk_size=chunk_size)

         self.parts[seq] = response["headers"].get('etag', '').strip('"\'')

         return response

+    def copy_part(self, seq, offset, chunk_size, labels, remote_status=None):

+        """

+        Copy a remote file chunk

+        http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadUploadPart.html

+        http://docs.amazonwebservices.com/AmazonS3/latest/API/mpUploadUploadPartCopy.html

+        """

+        debug("Copying part %i of %r (%s bytes)" % (seq, self.upload_id,

+                                                    chunk_size))

+

+        # set up headers with copy-params.

+        # Examples:

+        #    x-amz-copy-source: /source_bucket/sourceObject

+        #    x-amz-copy-source-range:bytes=first-last

+        #    x-amz-copy-source-if-match: etag

+        #    x-amz-copy-source-if-none-match: etag

+        #    x-amz-copy-source-if-unmodified-since: time_stamp

+        #    x-amz-copy-source-if-modified-since: time_stamp

+        headers = {

+            "x-amz-copy-source": "/%s/%s" % (self.src_uri.bucket(),

+                                             self.src_uri.object()),

+        }

+

+        # byte range, with end byte included. A 10 byte file has bytes=0-9

+        headers["x-amz-copy-source-range"] = \

+            "bytes=%d-%d" % (offset, (offset + chunk_size - 1))

+

+        query_string_params = {'partNumber': '%s' % seq,

+                               'uploadId': self.upload_id}

+        request = self.s3.create_request("OBJECT_PUT", uri=self.dst_uri,

+                                         headers=headers,

+                                         uri_params=query_string_params)

+        response = self.s3.send_request(request)

+

+        # NOTE: Amazon sends whitespace while upload progresses, which

+        # accumulates in response body and seems to confuse XML parser.

+        # Strip newlines to find ETag in XML response data

+        #data = response["data"].replace("\n", '')

+        self.parts[seq] = getTextFromXml(response['data'], "ETag") or ''

+

+        return response

+

     def complete_multipart_upload(self):

         """

         Finish a multipart upload

@@ -188,12 +295,13 @@ class MultiPartUpload(object):

         part_xml = "<Part><PartNumber>%i</PartNumber><ETag>%s</ETag></Part>"

         for seq, etag in self.parts.items():

             parts_xml.append(part_xml % (seq, etag))

-        body = "<CompleteMultipartUpload>%s</CompleteMultipartUpload>" % ("".join(parts_xml))

+        body = "<CompleteMultipartUpload>%s</CompleteMultipartUpload>" % (

+            "".join(parts_xml))

-        headers = { "content-length": str(len(body)) }

-        request = self.s3.create_request("OBJECT_POST", uri = self.uri,

-                                         headers = headers, body = body,

-                                         uri_params = {'uploadId': self.upload_id})

+        headers = {"content-length": str(len(body))}

+        request = self.s3.create_request(

+            "OBJECT_POST", uri=self.dst_uri, headers=headers, body=body,

+            uri_params={'uploadId': self.upload_id})

         response = self.s3.send_request(request)

         return response

@@ -211,99 +319,4 @@ class MultiPartUpload(object):

         return response

-class MultiPartCopy(MultiPartUpload):

-

-    def __init__(self, s3, src_uri, dst_uri, src_size, headers_baseline = {}):

-        self.s3 = s3

-        self.file = self.src_uri = src_uri

-        self.uri  = self.dst_uri = dst_uri

-        # ...

-        self.src_size = src_size

-        self.parts = {}

-        self.headers_baseline = headers_baseline

-        self.upload_id = self.initiate_multipart_copy()

-

-    def initiate_multipart_copy(self):

-        return self.initiate_multipart_upload()

-

-    def complete_multipart_copy(self):

-        return self.complete_multipart_upload()

-

-    def abort_copy(self):

-        return self.abort_upload()

-

-

-    def copy_all_parts(self):

-        """

-        Execute a full multipart upload copy on a remote file

-        Returns the seq/etag dict

-        """

-        if not self.upload_id:

-            raise RuntimeError("Attempting to use a multipart copy that has not been initiated.")

-

-        size_left = file_size = self.src_size

-        self.chunk_size = self.s3.config.multipart_copy_size # - 1

-        nr_parts = file_size / self.chunk_size + (file_size % self.chunk_size and 1)

-        debug("MultiPart: Copying %s in %d parts" % (self.src_uri, nr_parts))

-

-        seq = 1

-        while size_left > 0:

-            offset = self.chunk_size * (seq - 1)

-            current_chunk_size = min(file_size - offset, self.chunk_size)

-            size_left -= current_chunk_size

-            labels = {

-                'source' : unicodise(self.src_uri.uri()),

-                'destination' : unicodise(self.uri.uri()),

-                'extra' : "[part %d of %d, %s]" % (seq, nr_parts, "%d%sB" % formatSize(current_chunk_size, human_readable = True))

-            }

-            try:

-                self.copy_part(seq, offset, current_chunk_size, labels)

-            except:

-                # TODO: recover from some "retriable" errors?

-                error(u"Upload copy of '%s' part %d failed. Aborting multipart upload copy." % (self.src_uri, seq))

-                self.abort_copy()

-                raise

-            seq += 1

-

-        debug("MultiPart: Copy finished: %d parts", seq - 1)

-

-    def copy_part(self, seq, offset, chunk_size, labels):

-        """

-        Copy a remote file chunk

-        http://docs.amazonwebservices.com/AmazonS3/latest/API/index.html?mpUploadUploadPart.html

-        http://docs.amazonwebservices.com/AmazonS3/latest/API/mpUploadUploadPartCopy.html

-        """

-        debug("Copying part %i of %r (%s bytes)" % (seq, self.upload_id, chunk_size))

-

-        # set up headers with copy-params.

-        # Examples:

-        #    x-amz-copy-source: /source_bucket/sourceObject

-        #    x-amz-copy-source-range:bytes=first-last

-        #    x-amz-copy-source-if-match: etag

-        #    x-amz-copy-source-if-none-match: etag

-        #    x-amz-copy-source-if-unmodified-since: time_stamp

-        #    x-amz-copy-source-if-modified-since: time_stamp

-        headers = { "x-amz-copy-source": "/%s/%s" % (self.src_uri.bucket(), self.src_uri.object()) }

-

-        # include byte range header if already on next sequence or original file is > 5gb

-        if (seq > 1) or (chunk_size >= self.s3.config.multipart_copy_size):

-            # a 10 byte file has bytes=0-9

-            headers["x-amz-copy-source-range"] = "bytes=%d-%d" % (offset, (offset + chunk_size - 1))

-

-        query_string = "?partNumber=%i&uploadId=%s" % (seq, self.upload_id)

-

-        request = self.s3.create_request("OBJECT_PUT", uri = self.uri, headers = headers, extra = query_string)

-        response = self.s3.send_request(request)

-

-        # NOTE: Amazon sends whitespace while upload progresses, which

-        # accumulates in response body and seems to confuse XML parser.

-        # Strip newlines to find ETag in XML response data

-        data = response["data"].replace("\n", '')

-        self.parts[seq] = getTextFromXml(data, "ETag")

-

-        # TODO: how to fail if no ETag found ... raise Exception?

-        #debug("Uploaded copy part %i of %r (%s bytes): etag=%s" % (seq, self.upload_id, chunk_size, self.parts[seq]))

-

-        return response

-

 # vim:et:ts=4:sts=4:ai

@@ -40,7 +40,7 @@ from .ACL import ACL, GranteeLogDelivery

 from .BidirMap import BidirMap

 from .Config import Config

 from .Exceptions import *

-from .MultiPart import MultiPartUpload, MultiPartCopy

+from .MultiPart import MultiPartUpload

 from .S3Uri import S3Uri

 from .ConnMan import ConnMan

 from .Crypto import (sign_request_v2, sign_request_v4, checksum_sha256_file,

@@ -845,19 +845,15 @@ class S3(object):

             headers.update(extra_headers)

         ## Multipart decision - only do multipart copy for remote s3 files > 5gb

-        multipart = False

-        # TODO: does it need new config option for: enable_multipart_copy ?

         if self.config.enable_multipart:

             # get size of remote src only if multipart is enabled

             src_info = self.object_info(src_uri)

             size = int(src_info["headers"]["content-length"])

-            if size > self.config.multipart_copy_size:

-                multipart = True

-

-        if multipart:

-            # Multipart requests are quite different... drop here

-            return self.copy_file_multipart(src_uri, dst_uri, size, headers)

+            if size > self.config.multipart_copy_chunk_size_mb * 1024 * 1024:

+                # Multipart requests are quite different... drop here

+                return self.copy_file_multipart(src_uri, dst_uri, size,

+                                                headers)

         ## Not multipart...

         headers['x-amz-copy-source'] = "/%s/%s" % (

@@ -866,7 +862,8 @@ class S3(object):

         )

         headers['x-amz-metadata-directive'] = "COPY"

-        request = self.create_request("OBJECT_PUT", uri = dst_uri, headers = headers)

+        request = self.create_request("OBJECT_PUT", uri=dst_uri,

+                                      headers=headers)

         response = self.send_request(request)

         if response["data"] and getRootTagName(response["data"]) == "Error":

             #http://doc.s3.amazonaws.com/proposals/copy.html

@@ -1627,9 +1624,9 @@ class S3(object):

         return response

-    def send_file_multipart(self, stream, headers, uri, size, extra_label = ""):

+    def send_file_multipart(self, stream, headers, uri, size, extra_label=""):

         timestamp_start = time.time()

-        upload = MultiPartUpload(self, stream, uri, headers)

+        upload = MultiPartUpload(self, stream, uri, headers, size)

         upload.upload_all_parts(extra_label)

         response = upload.complete_multipart_upload()

         timestamp_end = time.time()

@@ -1643,17 +1640,10 @@ class S3(object):

             raise S3UploadError(getTextFromXml(response["data"], 'Message'))

         return response

-    def copy_file_multipart(self, src_uri, dst_uri, size, headers):

-        debug("copying multi-part ..." )

-        timestamp_start = time.time()

-        multicopy = MultiPartCopy(self, src_uri, dst_uri, size, headers)

-        multicopy.copy_all_parts()

-        response = multicopy.complete_multipart_copy()

-        timestamp_end = time.time()

-        response["elapsed"] = timestamp_end - timestamp_start

-        response["size"] = size

-        response["speed"] = response["elapsed"] and float(response["size"]) / response["elapsed"] or float(-1)

-        return response

+    def copy_file_multipart(self, src_uri, dst_uri, size, headers,

+                            extra_label=""):

+        return self.send_file_multipart(src_uri, headers, dst_uri, size,

+                                        extra_label)

     def recv_file(self, request, stream, labels, start_position = 0, retries = _max_retries):

         self.update_region_inner_request(request)