erasure: Fix block index matching.

This patch fixes an important issue of block reconstruction upon block corruption.
This commit is contained in:
Harshavardhana 2016-06-05 04:25:15 -07:00
parent 18b3871705
commit c6ac3fa6db
2 changed files with 12 additions and 10 deletions

View File

@ -153,35 +153,36 @@ func metaPartBlockChecksums(disks []StorageAPI, eInfos []erasureInfo, partName s
} }
// Takes block index and block distribution to get the disk index. // Takes block index and block distribution to get the disk index.
func toDiskIndex(blockIdx int, distribution []int) (diskIndex int) { func toDiskIndex(blockIdx int, distribution []int) int {
diskIndex = -1
// Find out the right disk index for the input block index. // Find out the right disk index for the input block index.
for index, blockIndex := range distribution { for index, blockIndex := range distribution {
if blockIndex == blockIdx { if blockIndex-1 == blockIdx {
diskIndex = index return index
} }
} }
return diskIndex return -1
} }
// isValidBlock - calculates the checksum hash for the block and // isValidBlock - calculates the checksum hash for the block and
// validates if its correct returns true for valid cases, false otherwise. // validates if its correct returns true for valid cases, false otherwise.
func isValidBlock(disks []StorageAPI, volume, path string, diskIndex int, blockCheckSums []checkSumInfo) bool { func isValidBlock(disks []StorageAPI, volume, path string, diskIndex int, blockCheckSums []checkSumInfo) (ok bool) {
ok = false
// Unknown block index requested, treat it as error. // Unknown block index requested, treat it as error.
if diskIndex == -1 { if diskIndex == -1 {
return false return ok
} }
// Disk is not present, treat entire block to be non existent. // Disk is not present, treat entire block to be non existent.
if disks[diskIndex] == nil { if disks[diskIndex] == nil {
return false return ok
} }
// Read everything for a given block and calculate hash. // Read everything for a given block and calculate hash.
hashWriter := newHash(blockCheckSums[diskIndex].Algorithm) hashWriter := newHash(blockCheckSums[diskIndex].Algorithm)
hashBytes, err := hashSum(disks[diskIndex], volume, path, hashWriter) hashBytes, err := hashSum(disks[diskIndex], volume, path, hashWriter)
if err != nil { if err != nil {
return false return ok
} }
return hex.EncodeToString(hashBytes) == blockCheckSums[diskIndex].Hash ok = hex.EncodeToString(hashBytes) == blockCheckSums[diskIndex].Hash
return ok
} }
// decodeData - decode encoded blocks. // decodeData - decode encoded blocks.

View File

@ -231,6 +231,7 @@ func checkDisksConsistency(formatConfigs []*formatConfigV1) error {
// Collect currently available disk uuids. // Collect currently available disk uuids.
for index, formatConfig := range formatConfigs { for index, formatConfig := range formatConfigs {
if formatConfig == nil { if formatConfig == nil {
disks[index] = ""
continue continue
} }
disks[index] = formatConfig.XL.Disk disks[index] = formatConfig.XL.Disk