erasure: simplify XL backend operations (#4649) (#4758)

This change provides new implementations of the XL backend operations:
 - create file
 - read   file
 - heal   file
Further this change adds table based tests for all three operations.

This affects also the bitrot algorithm integration. Algorithms are now
integrated in an idiomatic way (like crypto.Hash).
Fixes #4696
Fixes #4649
Fixes #4359

cmd/erasure-healfile.go

@@ -16,71 +16,75 @@
 
 package cmd
 
-import "encoding/hex"
+import "hash"
 
-// Heals the erasure coded file. reedsolomon.Reconstruct() is used to reconstruct the missing parts.
-func erasureHealFile(latestDisks []StorageAPI, outDatedDisks []StorageAPI, volume, path, healBucket, healPath string,
-	size, blockSize int64, dataBlocks, parityBlocks int, algo HashAlgo) (checkSums []string, err error) {
-
-	var offset int64
-	remainingSize := size
-
-	// Hash for bitrot protection.
-	hashWriters := newHashWriters(len(outDatedDisks), bitRotAlgo)
-
-	for remainingSize > 0 {
-		curBlockSize := blockSize
-		if remainingSize < curBlockSize {
-			curBlockSize = remainingSize
-		}
-
-		// Calculate the block size that needs to be read from each disk.
-		curEncBlockSize := getChunkSize(curBlockSize, dataBlocks)
-
-		// Memory for reading data from disks and reconstructing missing data using erasure coding.
-		enBlocks := make([][]byte, len(latestDisks))
-
-		// Read data from the latest disks.
-		// FIXME: no need to read from all the disks. dataBlocks+1 is enough.
-		for index, disk := range latestDisks {
-			if disk == nil {
-				continue
-			}
-			enBlocks[index] = make([]byte, curEncBlockSize)
-			_, err := disk.ReadFile(volume, path, offset, enBlocks[index])
-			if err != nil {
-				enBlocks[index] = nil
-			}
-		}
-
-		// Reconstruct any missing data and parity blocks.
-		err := decodeDataAndParity(enBlocks, dataBlocks, parityBlocks)
-		if err != nil {
-			return nil, err
-		}
-
-		// Write to the healPath file.
-		for index, disk := range outDatedDisks {
-			if disk == nil {
-				continue
-			}
-			err := disk.AppendFile(healBucket, healPath, enBlocks[index])
-			if err != nil {
-				return nil, traceError(err)
-			}
-			hashWriters[index].Write(enBlocks[index])
-		}
-		remainingSize -= curBlockSize
-		offset += curEncBlockSize
+// 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)
 	}
-
-	// Checksums for the bit rot.
-	checkSums = make([]string, len(outDatedDisks))
-	for index, disk := range outDatedDisks {
-		if disk == nil {
+	f.Checksums = make([][]byte, len(s.disks))
+	hashers, verifiers := make([]hash.Hash, len(s.disks)), make([]*BitrotVerifier, len(s.disks))
+	for i, disk := range s.disks {
+		if disk == OfflineDisk {
+			hashers[i] = algorithm.New()
+		} else {
+			verifiers[i] = NewBitrotVerifier(algorithm, checksums[i])
+			f.Checksums[i] = checksums[i]
+		}
+	}
+	blocks := make([][]byte, len(s.disks))
+	chunksize := getChunkSize(blocksize, s.dataBlocks)
+	for offset := int64(0); offset < size; offset += blocksize {
+		if size < blocksize {
+			blocksize = size
+			chunksize = getChunkSize(blocksize, s.dataBlocks)
+		}
+		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
+				}
+			}
+		}
+		if err = s.ErasureDecodeDataAndParityBlocks(blocks); err != nil {
+			return f, err
+		}
+		for i, disk := range s.disks {
+			if disk != OfflineDisk {
+				continue
+			}
+			if err = offlineDisks[i].AppendFile(healVolume, healPath, blocks[i]); err != nil {
+				return f, traceError(err)
+			}
+			hashers[i].Write(blocks[i])
+		}
+	}
+	f.Size = size
+	f.Algorithm = algorithm
+	for i, disk := range s.disks {
+		if disk != OfflineDisk {
 			continue
 		}
-		checkSums[index] = hex.EncodeToString(hashWriters[index].Sum(nil))
+		f.Checksums[i] = hashers[i].Sum(nil)
 	}
-	return checkSums, nil
+	return f, nil
 }