@@ -50,6 +50,7 @@ class Config(object):

         'mtime',    # Modification timestamp

         'ctime',    # Creation timestamp

         'mode',     # File mode (e.g. rwxr-xr-x = 755)

+        'md5',      # File MD5 (if known)

         #'acl',     # Full ACL (not yet supported)

     ]

     delete_removed = False

@@ -16,7 +16,7 @@ import os

 import glob

 import copy

-__all__ = ["fetch_local_list", "fetch_remote_list", "compare_filelists", "filter_exclude_include"]

+__all__ = ["fetch_local_list", "fetch_remote_list", "compare_filelists", "filter_exclude_include", "parse_attrs_header"]

 def _fswalk_follow_symlinks(path):

         '''

@@ -173,8 +173,15 @@ def fetch_local_list(args, recursive = None):

                     'full_name' : full_name,

                     'size' : sr.st_size,

                     'mtime' : sr.st_mtime,

+		    'nlink' : sr.st_nlink, # record hardlink information

+		    'dev'   : sr.st_dev,

+		    'inode' : sr.st_ino,

+		    'uid' : sr.st_uid,

+		    'gid' : sr.st_gid,

+		    'sr': sr # save it all, may need it in preserve_attrs_list

                     ## TODO: Possibly more to save here...

                 }

+		loc_list.record_hardlink(relative_file, sr.st_dev, sr.st_ino)

         return loc_list, single_file

     cfg = Config()

@@ -257,7 +264,12 @@ def fetch_remote_list(args, require_attribs = False, recursive = None):

                 'object_key' : object['Key'],

                 'object_uri_str' : object_uri_str,

                 'base_uri' : remote_uri,

+		'nlink' : 1, # S3 doesn't support hardlinks itself

+		'dev' : None,

+		'inode' : None,

             }

+	    md5 = object['ETag'][1:-1]

+	    rem_list.record_md5(key, md5)

             if break_now:

                 break

         return rem_list

@@ -283,6 +295,7 @@ def fetch_remote_list(args, require_attribs = False, recursive = None):

             objectlist = _get_filelist_remote(uri)

             for key in objectlist:

                 remote_list[key] = objectlist[key]

+		remote_list.record_md5(key, objectlist.get_md5(key))

     else:

         for uri in remote_uris:

             uri_str = str(uri)

@@ -320,90 +333,135 @@ def fetch_remote_list(args, require_attribs = False, recursive = None):

                     'md5': response['headers']['etag'].strip('"\''),

                     'timestamp' : dateRFC822toUnix(response['headers']['date'])

                     })

+		    # get md5 from header if it's present.  We would have set that during upload

+		    if response['headers'].has_key('x-amz-meta-s3cmd-attrs'):

+                        attrs = parse_attrs_header(response['headers']['x-amz-meta-s3cmd-attrs'])

+			if attrs.has_key('md5'):

+                            remote_item.update({'md5': attrs['md5']})

+		    

                 remote_list[key] = remote_item

     return remote_list

+def parse_attrs_header(attrs_header):

+    attrs = {}

+    for attr in attrs_header.split("/"):

+        key, val = attr.split(":")

+	attrs[key] = val

+    return attrs

+

+

 def compare_filelists(src_list, dst_list, src_remote, dst_remote, delay_updates = False):

     def __direction_str(is_remote):

         return is_remote and "remote" or "local"

-    # We don't support local->local sync, use 'rsync' or something like that instead ;-)

+    def _compare(src_list, dst_lst, src_remote, dst_remote, file):

+        """Return True if src_list[file] matches dst_list[file], else False"""

+        attribs_match = True

+	if not (src_list.has_key(file) and dst_list.has_key(file)):

+            info(u"file does not exist in one side or the other: src_list=%s, dst_list=%s" % (src_list.has_key(file), dst_list.has_key(file)))

+            return False

+

+        ## check size first

+	if 'size' in cfg.sync_checks and dst_list[file]['size'] != src_list[file]['size']:

+            debug(u"xfer: %s (size mismatch: src=%s dst=%s)" % (file, src_list[file]['size'], dst_list[file]['size']))

+	    attribs_match = False

+

+        ## check md5

+        compare_md5 = 'md5' in cfg.sync_checks

+        # Multipart-uploaded files don't have a valid md5 sum - it ends with "...-nn"

+        if compare_md5:

+            if (src_remote == True and src_list[file]['md5'].find("-") >= 0) or (dst_remote == True and dst_list[file]['md5'].find("-") >= 0):

+                compare_md5 = False

+                info(u"disabled md5 check for %s" % file)

+        if attribs_match and compare_md5:

+            try:

+                src_md5 = src_list.get_md5(file)

+                dst_md5 = dst_list.get_md5(file)

+	    except (IOError,OSError), e:

+                # md5 sum verification failed - ignore that file altogether

+                debug(u"IGNR: %s (disappeared)" % (file))

+                warning(u"%s: file disappeared, ignoring." % (file))

+		raise

+

+            if src_md5 != dst_md5:

+                ## checksums are different.

+                attribs_match = False

+                debug(u"XFER: %s (md5 mismatch: src=%s dst=%s)" % (file, src_md5, dst_md5))

+

+        return attribs_match

+

+    # we don't support local->local sync, use 'rsync' or something like that instead ;-)

     assert(not(src_remote == False and dst_remote == False))

     info(u"Verifying attributes...")

     cfg = Config()

-    exists_list = SortedDict(ignore_case = False)

+    ## Items left on src_list will be transferred

+    ## Items left on update_list will be transferred after src_list

+    ## Items left on copy_pairs will be copied from dst1 to dst2

     update_list = SortedDict(ignore_case = False)

+    ## Items left on dst_list will be deleted

+    copy_pairs = []

+

     debug("Comparing filelists (direction: %s -> %s)" % (__direction_str(src_remote), __direction_str(dst_remote)))

-    debug("src_list.keys: %s" % src_list.keys())

-    debug("dst_list.keys: %s" % dst_list.keys())

-    for file in src_list.keys():

-        debug(u"CHECK: %s" % file)

-        if dst_list.has_key(file):

-            ## Was --skip-existing requested?

-            if cfg.skip_existing:

-                debug(u"IGNR: %s (used --skip-existing)" % (file))

-                exists_list[file] = src_list[file]

-                del(src_list[file])

-                ## Remove from destination-list, all that is left there will be deleted

-                del(dst_list[file])

-                continue

-

-            attribs_match = True

-            ## Check size first

-            if 'size' in cfg.sync_checks and dst_list[file]['size'] != src_list[file]['size']:

-                debug(u"XFER: %s (size mismatch: src=%s dst=%s)" % (file, src_list[file]['size'], dst_list[file]['size']))

-                attribs_match = False

+    for relative_file in src_list.keys():

+        debug(u"CHECK: %s: %s" % (relative_file, src_list.get_md5(relative_file)))

-            ## Check MD5

-            compare_md5 = 'md5' in cfg.sync_checks

-            # Multipart-uploaded files don't have a valid MD5 sum - it ends with "...-NN"

-            if compare_md5:

-                if (src_remote == True and src_list[file]['md5'].find("-") >= 0) or (dst_remote == True and dst_list[file]['md5'].find("-") >= 0):

-                    compare_md5 = False

-                    info(u"Disabled MD5 check for %s" % file)

-            if attribs_match and compare_md5:

-                try:

-                    if src_remote == False and dst_remote == True:

-                        src_md5 = hash_file_md5(src_list[file]['full_name'])

-                        dst_md5 = dst_list[file]['md5']

-                    elif src_remote == True and dst_remote == False:

-                        src_md5 = src_list[file]['md5']

-                        dst_md5 = hash_file_md5(dst_list[file]['full_name'])

-                    elif src_remote == True and dst_remote == True:

-                        src_md5 = src_list[file]['md5']

-                        dst_md5 = dst_list[file]['md5']

-                except (IOError,OSError), e:

-                    # MD5 sum verification failed - ignore that file altogether

-                    debug(u"IGNR: %s (disappeared)" % (file))

-                    warning(u"%s: file disappeared, ignoring." % (file))

-                    del(src_list[file])

-                    del(dst_list[file])

-                    continue

+        if dst_list.has_key(relative_file):

+            ## Was --skip-existing requested?

+	    if cfg.skip_existing:

+	        debug(u"IGNR: %s (used --skip-existing)" % (relative_file))

+	        del(src_list[relative_file])

+	        del(dst_list[relative_file])

+	        continue

-                if src_md5 != dst_md5:

-                    ## Checksums are different.

-                    attribs_match = False

-                    debug(u"XFER: %s (md5 mismatch: src=%s dst=%s)" % (file, src_md5, dst_md5))

+	    if _compare(src_list, dst_list, src_remote, dst_remote, relative_file):

+	        debug(u"IGNR: %s (transfer not needed)" % relative_file)

+	        del(src_list[relative_file])

+		del(dst_list[relative_file])

-            if attribs_match:

-                ## Remove from source-list, all that is left there will be transferred

-                debug(u"IGNR: %s (transfer not needed)" % file)

-                exists_list[file] = src_list[file]

-                del(src_list[file])

 	    else:

-	        if delay_updates:

-	            ## Remove from source-list, all that is left there will be transferred

-		    ## Add to update-list to transfer last

-		    debug(u"XFER UPDATE: %s" % file)

-		    update_list[file] = src_list[file]

-		    del(src_list[file])

-

-            ## Remove from destination-list, all that is left there will be deleted

-            del(dst_list[file])

-

-    return src_list, dst_list, exists_list, update_list

+                # look for matching file in src

+		md5 = src_list.get_md5(relative_file)

+		if md5 is not None and dst_list.by_md5.has_key(md5):

+                    # Found one, we want to copy

+                    dst1 = list(dst_list.by_md5[md5])[0]

+                    debug(u"REMOTE COPY src: %s -> %s" % (dst1, relative_file))

+		    copy_pairs.append((dst_list[dst1], relative_file))

+		    del(src_list[relative_file])

+		    del(dst_list[relative_file])

+                else:

+		    # record that we will get this file transferred to us (before all the copies), so if we come across it later again,

+		    # we can copy from _this_ copy (e.g. we only upload it once, and copy thereafter).

+                    debug(u"REMOTE COPY src before")

+		    dst_list.record_md5(relative_file, md5) 

+		    update_list[relative_file] = src_list[relative_file]

+		    del src_list[relative_file]

+		    del dst_list[relative_file]

+

+	else:

+            # dst doesn't have this file

+            # look for matching file elsewhere in dst

+	    md5 = src_list.get_md5(relative_file)

+	    dst1 = dst_list.find_md5_one(md5)

+	    if dst1 is not None:

+                # Found one, we want to copy

+                debug(u"REMOTE COPY dst: %s -> %s" % (dst1, relative_file))

+		copy_pairs.append((dst_list[dst1], relative_file))

+		del(src_list[relative_file])

+	    else:

+                # we don't have this file, and we don't have a copy of this file elsewhere.  Get it.

+	        # record that we will get this file transferred to us (before all the copies), so if we come across it later again,

+	        # we can copy from _this_ copy (e.g. we only upload it once, and copy thereafter).

+	        dst_list.record_md5(relative_file, md5) 

+		debug(u"REMOTE COPY dst before")

+

+    for f in dst_list.keys():

+        if not src_list.has_key(f) and not update_list.has_key(f):

+            # leave only those not on src_list + update_list

+            del dst_list[f]

+

+    return src_list, dst_list, update_list, copy_pairs

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

@@ -4,6 +4,7 @@

 ## License: GPL Version 2

 from BidirMap import BidirMap

+import Utils

 class SortedDictIterator(object):

     def __init__(self, sorted_dict, keys):

@@ -26,6 +27,8 @@ class SortedDict(dict):

         """

         dict.__init__(self, mapping, **kwargs)

         self.ignore_case = ignore_case

+        self.hardlinks = dict()

+        self.by_md5 = dict() # {md5: set(relative_files)}

     def keys(self):

         keys = dict.keys(self)

@@ -45,6 +48,61 @@ class SortedDict(dict):

     def __iter__(self):

         return SortedDictIterator(self, self.keys())

+

+    def record_md5(self, relative_file, md5):

+        if md5 not in self.by_md5:

+            self.by_md5[md5] = set()

+        self.by_md5[md5].add(relative_file)

+

+    def find_md5_one(self, md5):

+        try:

+            return list(self.by_md5.get(md5, set()))[0]

+        except:

+            return None

+            

+

+    def get_md5(self, relative_file):

+        md5 = None

+        if 'md5' in self[relative_file]:

+            return self[relative_file]['md5']

+	if self.is_hardlinked(relative_file): # speedup by getting it from one of the hardlinks already processed

+            md5 = self.get_hardlink_md5(relative_file)

+	    if md5 is None:

+                md5 = Utils.hash_file_md5(self[relative_file]['full_name'])

+		self.record_md5(relative_file, md5)

+	        self.set_hardlink_md5(relative_file, md5)

+	else:

+                md5 = Utils.hash_file_md5(self[relative_file]['full_name'])

+	        self[relative_file]['md5'] = md5

+		l.record_md5(relative_file, md5)

+	return md5

+

+    def record_hardlink(self, relative_file, dev, inode):

+        if dev not in self.hardlinks:

+            self.hardlinks[dev] = dict()

+        if inode not in self.hardlinks[dev]:

+            self.hardlinks[dev][inode] = dict(md5=None, relative_files=set())

+        self.hardlinks[dev][inode]['relative_files'].add(relative_file)

+

+    def set_hardlink_md5(self, relative_file, md5):

+        dev = self[relative_file]['dev']

+        inode = self[relative_file]['inode']

+        self.record_hardlink(relative_file, dev, inode)

+        self.hardlinks[dev][inode]['md5'] = md5

+

+    def is_hardlinked(self, relative_file):

+        return self[relative_file]['nlink'] > 1

+

+    def get_hardlink_md5(self, relative_file):

+        md5 = None

+        dev = self[relative_file]['dev']

+        inode = self[relative_file]['inode']

+        try:

+            md5 = self.hardlinks[dev][inode]['md5']

+        except:

+            pass

+        return md5

+

 if __name__ == "__main__":

     d = { 'AWS' : 1, 'Action' : 2, 'america' : 3, 'Auckland' : 4, 'America' : 5 }

     sd = SortedDict(d)

@@ -608,7 +608,7 @@ def cmd_sync_remote2remote(args):

     src_list, exclude_list = filter_exclude_include(src_list)

-    src_list, dst_list, existing_list, update_list = compare_filelists(src_list, dst_list, src_remote = True, dst_remote = True, delay_updates = cfg.delay_updates)

+    src_list, dst_list, update_list, copy_pairs = compare_filelists(src_list, dst_list, src_remote = True, dst_remote = True, delay_updates = cfg.delay_updates)

     src_count = len(src_list)

     update_count = len(update_list)

@@ -632,6 +632,11 @@ def cmd_sync_remote2remote(args):

         warning(u"Exitting now because of --dry-run")

         return

+    # if there are copy pairs, we can't do delete_before, on the chance

+    # we need one of the to-be-deleted files as a copy source.

+    if len(copy_pairs) > 0:

+        cfg.delete_after = True

+

     # Delete items in destination that are not in source

     if cfg.delete_removed and not cfg.delete_after:

         _do_deletes(s3, dst_list)

@@ -658,6 +663,7 @@ def cmd_sync_remote2remote(args):

     seq = 0

     seq = _upload(src_list, seq, src_count + update_count)

     seq = _upload(update_list, seq, src_count + update_count)

+    n_copied, bytes_saved = remote_copy(s3, dst_list, copy_pairs, destination_base)

     total_elapsed = time.time() - timestamp_start

     outstr = "Done. Copied %d files in %0.1f seconds, %0.2f files/s" % (seq, total_elapsed, seq/total_elapsed)

@@ -671,13 +677,6 @@ def cmd_sync_remote2remote(args):

         _do_deletes(s3, dst_list)

 def cmd_sync_remote2local(args):

-    def _parse_attrs_header(attrs_header):

-        attrs = {}

-        for attr in attrs_header.split("/"):

-            key, val = attr.split(":")

-            attrs[key] = val

-        return attrs

-

     def _do_deletes(local_list):

         for key in local_list:

             os.unlink(local_list[key]['full_name'])

@@ -696,7 +695,7 @@ def cmd_sync_remote2local(args):

     remote_list, exclude_list = filter_exclude_include(remote_list)

-    remote_list, local_list, existing_list, update_list = compare_filelists(remote_list, local_list, src_remote = True, dst_remote = False, delay_updates = cfg.delay_updates)

+    remote_list, local_list, update_list, copy_pairs = compare_filelists(remote_list, local_list, src_remote = True, dst_remote = False, delay_updates = cfg.delay_updates)

     local_count = len(local_list)

     remote_count = len(remote_list)

@@ -736,6 +735,11 @@ def cmd_sync_remote2local(args):

         warning(u"Exitting now because of --dry-run")

         return

+    # if there are copy pairs, we can't do delete_before, on the chance

+    # we need one of the to-be-deleted files as a copy source.

+    if len(copy_pairs) > 0:

+        cfg.delete_after = True

+

     if cfg.delete_removed and not cfg.delete_after:

         _do_deletes(local_list)

@@ -768,7 +772,7 @@ def cmd_sync_remote2local(args):

                     response = s3.object_get(uri, dst_stream, extra_label = seq_label)

                     dst_stream.close()

                     if response['headers'].has_key('x-amz-meta-s3cmd-attrs') and cfg.preserve_attrs:

-                        attrs = _parse_attrs_header(response['headers']['x-amz-meta-s3cmd-attrs'])

+                        attrs = parse_attrs_header(response['headers']['x-amz-meta-s3cmd-attrs'])

                         if attrs.has_key('mode'):

                             os.chmod(dst_file, int(attrs['mode']))

                         if attrs.has_key('mtime') or attrs.has_key('atime'):

@@ -837,33 +841,48 @@ def cmd_sync_remote2local(args):

     if cfg.delete_removed and cfg.delete_after:

         _do_deletes(local_list)

+def remote_copy(s3, remote_list, copy_pairs, destination_base):

+    saved_bytes = 0

+    for (dst1, dst2) in copy_pairs:

+        debug(u"Remote Copying from %s to %s" % (dst1, dst2))

+        dst_uri = S3Uri(destination_base + dst2)

+        extra_headers = copy(cfg.extra_headers)

+        try:

+            s3.object_copy(S3Uri(dst1['object_uri_str']), dst_uri, extra_headers)

+            saved_bytes = saved_bytes + dst1['size']

+        except:

+            raise

+    return (len(copy_pairs), saved_bytes)

+

+

 def cmd_sync_local2remote(args):

-    def _build_attr_header(src):

+    def _build_attr_header(local_list, src):

         import pwd, grp

         attrs = {}

-        src = deunicodise(src)

-        try:

-            st = os.stat_result(os.stat(src))

-        except OSError, e:

-            raise InvalidFileError(u"%s: %s" % (unicodise(src), e.strerror))

         for attr in cfg.preserve_attrs_list:

             if attr == 'uname':

                 try:

-                    val = pwd.getpwuid(st.st_uid).pw_name

+                    val = pwd.getpwuid(local_list[src]['uid']).pw_name

                 except KeyError:

                     attr = "uid"

-                    val = st.st_uid

-                    warning(u"%s: Owner username not known. Storing UID=%d instead." % (unicodise(src), val))

+                    val = local_list[src].get(uid)

+                    warning(u"%s: Owner username not known. Storing UID=%d instead." % (src, val))

             elif attr == 'gname':

                 try:

-                    val = grp.getgrgid(st.st_gid).gr_name

+                    val = grp.getgrgid(local_list[src].get('gid')).gr_name

                 except KeyError:

                     attr = "gid"

-                    val = st.st_gid

-                    warning(u"%s: Owner groupname not known. Storing GID=%d instead." % (unicodise(src), val))

+                    val = local_list[src].get('gid')

+                    warning(u"%s: Owner groupname not known. Storing GID=%d instead." % (src, val))

+            elif attr == 'md5':

+                val = local_list.get_md5(src)

             else:

-                val = getattr(st, 'st_' + attr)

+                val = getattr(local_list[src]['sr'], 'st_' + attr)

             attrs[attr] = val

+

+        if 'md5' in attrs and attrs['md5'] is None:

+            del attrs['md5']

+

         result = ""

         for k in attrs: result += "%s:%s/" % (k, attrs[k])

         return { 'x-amz-meta-s3cmd-attrs' : result[:-1] }

@@ -904,15 +923,15 @@ def cmd_sync_local2remote(args):

         # Flush remote_list, by the way

         remote_list = { local_list.keys()[0] : remote_list_entry }

-    local_list, remote_list, existing_list, update_list = compare_filelists(local_list, remote_list, src_remote = False, dst_remote = True, delay_updates = cfg.delay_updates)

+    local_list, remote_list, update_list, copy_pairs = compare_filelists(local_list, remote_list, src_remote = False, dst_remote = True, delay_updates = cfg.delay_updates)

     local_count = len(local_list)

     update_count = len(update_list)

+    copy_count = len(copy_pairs)

     remote_count = len(remote_list)

-    info(u"Summary: %d local files to upload, %d remote files to delete" % (local_count + update_count, remote_count))

-

+    info(u"Summary: %d local files to upload, %d files to remote copy, %d remote files to delete" % (local_count + update_count, copy_count, remote_count))

     def _set_remote_uri(local_list, destination_base, single_file_local):

         if len(local_list) > 0:

@@ -931,17 +950,24 @@ def cmd_sync_local2remote(args):

     if cfg.dry_run:

         for key in exclude_list:

             output(u"exclude: %s" % unicodise(key))

-        if cfg.delete_removed:

-            for key in remote_list:

-                output(u"delete: %s" % remote_list[key]['object_uri_str'])

         for key in local_list:

             output(u"upload: %s -> %s" % (local_list[key]['full_name_unicode'], local_list[key]['remote_uri']))

         for key in update_list:

             output(u"upload: %s -> %s" % (update_list[key]['full_name_unicode'], update_list[key]['remote_uri']))

+        for (dst1, dst2) in copy_pairs:

+            output(u"remote copy: %s -> %s" % (dst1['object_key'], remote_list[dst2]['object_key']))

+        if cfg.delete_removed:

+            for key in remote_list:

+                output(u"delete: %s" % remote_list[key]['object_uri_str'])

         warning(u"Exitting now because of --dry-run")

         return

+    # if there are copy pairs, we can't do delete_before, on the chance

+    # we need one of the to-be-deleted files as a copy source.

+    if len(copy_pairs) > 0:

+        cfg.delete_after = True

+

     if cfg.delete_removed and not cfg.delete_after:

         _do_deletes(s3, remote_list)

@@ -962,7 +988,7 @@ def cmd_sync_local2remote(args):

             extra_headers = copy(cfg.extra_headers)

             try:

                 if cfg.preserve_attrs:

-                    attr_header = _build_attr_header(src)

+                    attr_header = _build_attr_header(local_list, file)

                     debug(u"attr_header: %s" % attr_header)

                     extra_headers.update(attr_header)

                 response = s3.object_put(src, uri, extra_headers, extra_label = seq_label)

@@ -981,23 +1007,24 @@ def cmd_sync_local2remote(args):

             uploaded_objects_list.append(uri.object())

         return seq, total_size

+

     n, total_size = _upload(local_list, 0, local_count, total_size)

     n, total_size = _upload(update_list, n, local_count, total_size)

+    n_copies, saved_bytes = remote_copy(s3, remote_list, copy_pairs, destination_base)

+    if cfg.delete_removed and cfg.delete_after:

+        _do_deletes(s3, remote_list)

     total_elapsed = time.time() - timestamp_start

     total_speed = total_elapsed and total_size/total_elapsed or 0.0

     speed_fmt = formatSize(total_speed, human_readable = True, floating_point = True)

     # Only print out the result if any work has been done or

     # if the user asked for verbose output

-    outstr = "Done. Uploaded %d bytes in %0.1f seconds, %0.2f %sB/s" % (total_size, total_elapsed, speed_fmt[0], speed_fmt[1])

-    if total_size > 0:

+    outstr = "Done. Uploaded %d bytes in %0.1f seconds, %0.2f %sB/s.  Copied %d files saving %d bytes transfer." % (total_size, total_elapsed, speed_fmt[0], speed_fmt[1], n_copies, saved_bytes)

+    if total_size + saved_bytes > 0:

         output(outstr)

     else:

         info(outstr)

-    if cfg.delete_removed and cfg.delete_after:

-        _do_deletes(s3, remote_list)

-

     if cfg.invalidate_on_cf:

         if len(uploaded_objects_list) == 0:

             info("Nothing to invalidate in CloudFront")