Merge pull request #154 from jt70471/jt70471-patch-1

Skip chunks to copy if already on destination for issue #134

src/duplicacy_backupmanager.go

@@ -1484,14 +1484,27 @@ func (manager *BackupManager) CopySnapshots(otherManager *BackupManager, snapsho
         return false
     }
 
-    revisionMap := make(map[int]bool)
+    if snapshotID == "" && len(revisionsToBeCopied) > 0 {
+        LOG_ERROR("SNAPSHOT_ERROR", "You must specify the snapshot id when one or more revisions are specified.")
+        return false
+    }
+
+    revisionMap := make(map[string]map[int]bool)
+
+    _, found := revisionMap[snapshotID]
+    if !found {
+        revisionMap[snapshotID] = make(map[int]bool)
+    }
+
     for _, revision := range revisionsToBeCopied {
-        revisionMap[revision] = true
+        revisionMap[snapshotID][revision] = true
     }
 
     var snapshots [] *Snapshot
+    var otherSnapshots [] *Snapshot
     var snapshotIDs [] string
     var err error
+
     if snapshotID == "" {
         snapshotIDs, err = manager.SnapshotManager.ListSnapshotIDs()
         if err != nil {
@@ -1503,6 +1516,10 @@ func (manager *BackupManager) CopySnapshots(otherManager *BackupManager, snapsho
     }
 
     for _, id := range snapshotIDs {
+        _, found := revisionMap[id]
+        if !found {
+            revisionMap[id] = make(map[int]bool)
+        }
         revisions, err := manager.SnapshotManager.ListSnapshotRevisions(id)
         if err != nil {
             LOG_ERROR("SNAPSHOT_LIST", "Failed to list all revisions for snapshot %s: %v", id, err)
@@ -1511,9 +1528,14 @@ func (manager *BackupManager) CopySnapshots(otherManager *BackupManager, snapsho
 
         for _, revision := range revisions {
             if len(revisionsToBeCopied) > 0 {
-                if _, found := revisionMap[revision]; !found {
+                if _, found := revisionMap[id][revision]; found {
+                    revisionMap[id][revision] = true
+                } else {
+                    revisionMap[id][revision] = false
                     continue
                 }
+            } else {
+                revisionMap[id][revision] = true
             }
 
             snapshotPath := fmt.Sprintf("snapshots/%s/%d", id, revision)
@@ -1525,21 +1547,44 @@ func (manager *BackupManager) CopySnapshots(otherManager *BackupManager, snapsho
             }
 
             if exist {
-                LOG_INFO("SNAPSHOT_EXIST", "Snapshot %s at revision %d already exists in the destination storage",
+                LOG_INFO("SNAPSHOT_EXIST", "Snapshot %s at revision %d already exists at the destination storage",
                           id, revision)
+                revisionMap[id][revision] = false
                 continue
             }
 
             snapshot := manager.SnapshotManager.DownloadSnapshot(id, revision)
             snapshots = append(snapshots, snapshot)
         }
+
+        otherRevisions, err := otherManager.SnapshotManager.ListSnapshotRevisions(id)
+        if err != nil {
+            LOG_ERROR("SNAPSHOT_LIST", "Failed to list all revisions at the destination for snapshot %s: %v", id, err)
+            return false
+        }
+
+        for _, otherRevision := range otherRevisions {
+            otherSnapshot := otherManager.SnapshotManager.DownloadSnapshot(id, otherRevision)
+            otherSnapshots = append(otherSnapshots, otherSnapshot)
+        }
+
+    }
+
+    if len(snapshots) == 0 {
+        LOG_INFO("SNAPSHOT_COPY", "Nothing to copy, all snapshot revisions exist at the destination.")
+        return true
     }
 
     chunks := make(map[string]bool)
 
     for _, snapshot := range snapshots {
 
+        if revisionMap[snapshot.ID][snapshot.Revision] == false {
+            continue
+        }
+
         LOG_TRACE("SNAPSHOT_COPY", "Copying snapshot %s at revision %d", snapshot.ID, snapshot.Revision)
+
         for _, chunkHash := range snapshot.FileSequence {
             chunks[chunkHash] = true
         }
@@ -1565,42 +1610,90 @@ func (manager *BackupManager) CopySnapshots(otherManager *BackupManager, snapsho
         }
     }
 
+    for _, otherSnapshot := range otherSnapshots {
+
+        for _, chunkHash := range otherSnapshot.FileSequence {
+            if  _, found := chunks[chunkHash]; found {
+                chunks[chunkHash] = false
+            }
+        }
+
+        for _, chunkHash := range otherSnapshot.ChunkSequence {
+            if _, found := chunks[chunkHash]; found {
+                chunks[chunkHash] = false
+            }
+        }
+
+        for _, chunkHash := range otherSnapshot.LengthSequence {
+            if _, found := chunks[chunkHash]; found {
+                chunks[chunkHash] = false
+            }
+        }
+
+        description := otherManager.SnapshotManager.DownloadSequence(otherSnapshot.ChunkSequence)
+        err := otherSnapshot.LoadChunks(description)
+        if err != nil {
+            LOG_ERROR("SNAPSHOT_CHUNK", "Failed to load chunks for destination snapshot %s at revision %d: %v",
+                      otherSnapshot.ID, otherSnapshot.Revision, err)
+            return false
+        }
+
+        for _, chunkHash := range otherSnapshot.ChunkHashes {
+            if _, found := chunks[chunkHash]; found {
+                chunks[chunkHash] = false
+            }
+        }
+    }
+
     chunkDownloader := CreateChunkDownloader(manager.config, manager.storage, nil, false, threads)
 
     chunkUploader := CreateChunkUploader(otherManager.config, otherManager.storage, nil, threads,
         func(chunk *Chunk, chunkIndex int, skipped bool, chunkSize int, uploadSize int) {
             if skipped {
-                LOG_INFO("SNAPSHOT_COPY", "Chunk %s (%d/%d) exists in the destination", chunk.GetID(), chunkIndex, len(chunks))
+                LOG_INFO("SNAPSHOT_COPY", "Chunk %s (%d/%d) exists at the destination", chunk.GetID(), chunkIndex, len(chunks))
             } else {
-                LOG_INFO("SNAPSHOT_COPY", "Copied chunk %s (%d/%d)", chunk.GetID(), chunkIndex, len(chunks))
+                LOG_INFO("SNAPSHOT_COPY", "Chunk %s (%d/%d) copied to the destination", chunk.GetID(), chunkIndex, len(chunks))
             }
             otherManager.config.PutChunk(chunk)
         })
+
     chunkUploader.Start()
 
+    totalCopied := 0
+    totalSkipped := 0
     chunkIndex := 0
-    for chunkHash, _ := range chunks {
+
+    for chunkHash, needsCopy := range chunks {
         chunkIndex++
         chunkID := manager.config.GetChunkIDFromHash(chunkHash)
+        if needsCopy {
             newChunkID := otherManager.config.GetChunkIDFromHash(chunkHash)
-
             LOG_DEBUG("SNAPSHOT_COPY", "Copying chunk %s to %s", chunkID, newChunkID)
-
             i := chunkDownloader.AddChunk(chunkHash)
             chunk := chunkDownloader.WaitForChunk(i)
             newChunk := otherManager.config.GetChunk()
             newChunk.Reset(true)
             newChunk.Write(chunk.GetBytes())
             chunkUploader.StartChunk(newChunk, chunkIndex)
+            totalCopied++
+        } else {
+            LOG_INFO("SNAPSHOT_COPY", "Chunk %s (%d/%d) exists at the destination.", chunkID, chunkIndex, len(chunks))
+            totalSkipped++
+        }
     }
 
     chunkDownloader.Stop()
     chunkUploader.Stop()
 
+    LOG_INFO("SNAPSHOT_COPY", "Total chunks copied = %d, skipped = %d.", totalCopied, totalSkipped)
+
     for _, snapshot := range snapshots {
-        otherManager.storage.CreateDirectory(0, fmt.Sprintf("snapshots/%s", manager.snapshotID))
+        if revisionMap[snapshot.ID][snapshot.Revision] == false {
+            continue
+        }
+        otherManager.storage.CreateDirectory(0, fmt.Sprintf("snapshots/%s", snapshot.ID))
         description, _ := snapshot.MarshalJSON()
-        path := fmt.Sprintf("snapshots/%s/%d", manager.snapshotID, snapshot.Revision)
+        path := fmt.Sprintf("snapshots/%s/%d", snapshot.ID, snapshot.Revision)
         otherManager.SnapshotManager.UploadFile(path, path, description)
         LOG_INFO("SNAPSHOT_COPY", "Copied snapshot %s at revision %d", snapshot.ID, snapshot.Revision)
     }