Fix bug in ErasureStorage.HealFile (#4913)

cmd/erasure-healfile.go

@@ -16,72 +16,140 @@
 
 package cmd
 
-import "hash"
+import (
+	"hash"
+)
+
+// HealFile tries to reconstruct an erasure-coded file spread over all
+// available disks. HealFile will read the valid parts of the file,
+// reconstruct the missing data and write the reconstructed parts back
+// to `staleDisks`.
+//
+// `staleDisks` is a slice of disks where each non-nil entry has stale
+// or no data, and so will be healed.
+//
+// It is required that `s.disks` have a (read-quorum) majority of
+// disks with valid data for healing to work.
+//
+// In addition, `staleDisks` and `s.disks` must have the same ordering
+// of disks w.r.t. erasure coding of the object.
+//
+// The function will try to read the valid parts from the file under
+// the given volume and path and tries to reconstruct the file under
+// the given healVolume and healPath (on staleDisks). The given
+// algorithm will be used to verify the valid parts and to protect the
+// reconstructed file.
+//
+// It returns bitrot checksums for the non-nil staleDisks.
+func (s ErasureStorage) HealFile(staleDisks []StorageAPI, volume, path string,
+	blocksize int64, healVolume, healPath string, size int64,
+	algorithm BitrotAlgorithm, checksums [][]byte) (f ErasureFileInfo,
+	err error) {
 
-// HealFile tries to reconstruct a bitrot encoded file spread over all available disks. HealFile will read the valid parts of the file,
-// reconstruct the missing data and write the reconstructed parts back to the disks.
-// It will try to read the valid parts from the file under the given volume and path and tries to reconstruct the file under the given
-// healVolume and healPath. The given algorithm will be used to verify the valid parts and to protect the reconstructed file.
-func (s ErasureStorage) HealFile(offlineDisks []StorageAPI, volume, path string, blocksize int64, healVolume, healPath string, size int64, algorithm BitrotAlgorithm, checksums [][]byte) (f ErasureFileInfo, err error) {
 	if !algorithm.Available() {
 		return f, traceError(errBitrotHashAlgoInvalid)
 	}
+
+	// Initialization
 	f.Checksums = make([][]byte, len(s.disks))
-	hashers, verifiers := make([]hash.Hash, len(s.disks)), make([]*BitrotVerifier, len(s.disks))
+	hashers := make([]hash.Hash, len(s.disks))
+	verifiers := make([]*BitrotVerifier, len(s.disks))
 	for i, disk := range s.disks {
-		if disk == OfflineDisk {
+		switch {
+		case staleDisks[i] != nil:
 			hashers[i] = algorithm.New()
-		} else {
+		case disk == nil:
+			// disregard unavailable disk
+			continue
+		default:
 			verifiers[i] = NewBitrotVerifier(algorithm, checksums[i])
 			f.Checksums[i] = checksums[i]
 		}
 	}
-	blocks := make([][]byte, len(s.disks))
+
+	// Scan part files on disk, block-by-block reconstruct it and
+	// write to stale disks.
 	chunksize := getChunkSize(blocksize, s.dataBlocks)
-	for offset := int64(0); offset < size; offset += blocksize {
-		if size < blocksize {
-			blocksize = size
+	var chunkOffset, blockOffset int64
+	for ; blockOffset < size; blockOffset += blocksize {
+		// last iteration may have less than blocksize data
+		// left, so chunksize needs to be recomputed.
+		if size < blockOffset+blocksize {
+			blocksize = size - blockOffset
 			chunksize = getChunkSize(blocksize, s.dataBlocks)
 		}
+
+		// read a chunk from each disk, until we have
+		// `s.dataBlocks` number of chunks set to non-nil in
+		// `blocks`
+		blocks := make([][]byte, len(s.disks))
+		var buffer []byte
 		numReads := 0
 		for i, disk := range s.disks {
-			if disk != OfflineDisk {
-				if blocks[i] == nil {
-					blocks[i] = make([]byte, chunksize)
-				}
-				blocks[i] = blocks[i][:chunksize]
-				if !verifiers[i].IsVerified() {
-					_, err = disk.ReadFileWithVerify(volume, path, offset, blocks[i], verifiers[i])
-				} else {
-					_, err = disk.ReadFile(volume, path, offset, blocks[i])
-				}
-				if err != nil {
-					blocks[i] = nil
-				} else {
-					numReads++
-				}
-				if numReads == s.dataBlocks { // we have enough data to reconstruct
-					break
-				}
+			// skip reading from unavailable or stale disks
+			if disk == nil || staleDisks[i] != nil {
+				continue
+			}
+			// allocate buffer only when needed - when
+			// reads fail, the buffer can be reused
+			if int64(len(buffer)) != chunksize {
+				buffer = make([]byte, chunksize)
+			}
+			if !verifiers[i].IsVerified() {
+				_, err = disk.ReadFileWithVerify(volume, path,
+					chunkOffset, buffer, verifiers[i])
+			} else {
+				_, err = disk.ReadFile(volume, path,
+					chunkOffset, buffer)
+			}
+			if err != nil {
+				// LOG FIXME: add a conditional log
+				// for read failures, once per-disk
+				// per-function-invocation.
+				continue
+			}
+
+			// read was successful, so set the buffer as
+			// blocks[i], and reset buffer to nil to force
+			// allocation on next iteration
+			blocks[i], buffer = buffer, nil
+
+			numReads++
+			if numReads == s.dataBlocks {
+				// we have enough data to reconstruct
+				break
 			}
 		}
+
+		// advance the chunk offset to prepare for next loop
+		// iteration
+		chunkOffset += chunksize
+
+		// reconstruct data - this computes all data and parity shards
 		if err = s.ErasureDecodeDataAndParityBlocks(blocks); err != nil {
 			return f, err
 		}
-		for i, disk := range s.disks {
-			if disk != OfflineDisk {
+
+		// write computed shards as chunks on file in each
+		// stale disk
+		for i, disk := range staleDisks {
+			if disk == nil {
 				continue
 			}
-			if err = offlineDisks[i].AppendFile(healVolume, healPath, blocks[i]); err != nil {
+
+			err = disk.AppendFile(healVolume, healPath, blocks[i])
+			if err != nil {
 				return f, traceError(err)
 			}
 			hashers[i].Write(blocks[i])
 		}
 	}
+
+	// copy computed file hashes into output variable
 	f.Size = size
 	f.Algorithm = algorithm
-	for i, disk := range s.disks {
-		if disk != OfflineDisk {
+	for i, disk := range staleDisks {
+		if disk == nil {
 			continue
 		}
 		f.Checksums[i] = hashers[i].Sum(nil)