XL: RenameFile should rename and cleanup on writeQuorum. (#1702)

Fixes #1683

xl-erasure-v1.go

@@ -511,15 +511,8 @@ func (xl XL) StatFile(volume, path string) (FileInfo, error) {
 	}, nil
 }
 
-// DeleteFile - delete a file
+// deleteXLFiles - delete all XL backend files.
-func (xl XL) DeleteFile(volume, path string) error {
+func (xl XL) deleteXLFiles(volume, path string) error {
-	if !isValidVolname(volume) {
-		return errInvalidArgument
-	}
-	if !isValidPath(path) {
-		return errInvalidArgument
-	}
-
 	errCount := 0
 	// Update meta data file and remove part file
 	for index, disk := range xl.storageDisks {
@@ -551,11 +544,23 @@ func (xl XL) DeleteFile(volume, path string) error {
 			return err
 		}
 	}
-
 	// Return success.
 	return nil
 }
 
+// DeleteFile - delete a file
+func (xl XL) DeleteFile(volume, path string) error {
+	if !isValidVolname(volume) {
+		return errInvalidArgument
+	}
+	if !isValidPath(path) {
+		return errInvalidArgument
+	}
+
+	// Delete all XL files.
+	return xl.deleteXLFiles(volume, path)
+}
+
 // RenameFile - rename file.
 func (xl XL) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) error {
 	// Validate inputs.
@@ -572,21 +577,49 @@ func (xl XL) RenameFile(srcVolume, srcPath, dstVolume, dstPath string) error {
 		return errInvalidArgument
 	}
 
-	errCount := 0
+	// Initialize sync waitgroup.
-	for _, disk := range xl.storageDisks {
+	var wg = &sync.WaitGroup{}
+
+	// Initialize list of errors.
+	var errs = make([]error, len(xl.storageDisks))
+
+	// Rename file on all underlying storage disks.
+	for index, disk := range xl.storageDisks {
 		// Append "/" as srcPath and dstPath are either leaf-dirs or non-leaf-dris.
 		// If srcPath is an object instead of prefix we just rename the leaf-dir and
 		// not rename the part and metadata files separately.
-		err := disk.RenameFile(srcVolume, retainSlash(srcPath), dstVolume, retainSlash(dstPath))
+		wg.Add(1)
-		if err != nil {
+		go func(index int, disk StorageAPI) {
-			errCount++
+			defer wg.Done()
-			// We can safely allow RenameFile errors up to len(xl.storageDisks) - xl.writeQuorum
+			err := disk.RenameFile(srcVolume, retainSlash(srcPath), dstVolume, retainSlash(dstPath))
-			// otherwise return failure.
+			if err != nil {
-			if errCount <= len(xl.storageDisks)-xl.writeQuorum {
+				errs[index] = err
-				continue
 			}
-			return err
+			errs[index] = nil
+		}(index, disk)
+	}
+
+	// Wait for all RenameFile to finish.
+	wg.Wait()
+
+	// Gather err count.
+	var errCount = 0
+	for _, err := range errs {
+		if err == nil {
+			continue
 		}
+		errCount++
+	}
+	// We can safely allow RenameFile errors up to len(xl.storageDisks) - xl.writeQuorum
+	// otherwise return failure. Cleanup successful renames.
+	if errCount > len(xl.storageDisks)-xl.writeQuorum {
+		// Special condition if readQuorum exists, then return success.
+		if errCount <= len(xl.storageDisks)-xl.readQuorum {
+			return nil
+		}
+		// Ignore errors here, delete all successfully written files.
+		xl.deleteXLFiles(dstVolume, dstPath)
+		return errWriteQuorum
 	}
 	return nil
 }