erasure-readfile: write to given Writer than returning buffer. (#1910)

Fixes #1889
master
Bala FA 9 years ago committed by Harshavardhana
parent c41bf26712
commit 1ea1dba528
  1. 105
      erasure-readfile.go
  2. 25
      erasure-utils.go
  3. 34
      xl-v1-object.go

@ -17,26 +17,22 @@
package main
import (
"bytes"
"encoding/hex"
"errors"
"io"
"github.com/klauspost/reedsolomon"
)
// erasureReadFile - read an entire erasure coded file at into a byte
// array. Erasure coded parts are often few mega bytes in size and it
// is convenient to return them as byte slice. This function also
// supports bit-rot detection by verifying checksum of individual
// block's checksum.
func erasureReadFile(disks []StorageAPI, volume string, path string, partName string, size int64, eInfos []erasureInfo) ([]byte, error) {
// Return data buffer.
var buffer []byte
// Total size left
totalSizeLeft := size
// Starting offset for reading.
startOffset := int64(0)
// erasureReadFile - read bytes from erasure coded files and writes to given writer.
// Erasure coded files are read block by block as per given erasureInfo and data chunks
// are decoded into a data block. Data block is trimmed for given offset and length,
// then written to given writer. This function also supports bit-rot detection by
// verifying checksum of individual block's checksum.
func erasureReadFile(writer io.Writer, disks []StorageAPI, volume string, path string, partName string, eInfos []erasureInfo, offset int64, length int64) (int64, error) {
// Total bytes written to writer
bytesWritten := int64(0)
// Gather previously calculated block checksums.
blockCheckSums := metaPartBlockChecksums(disks, eInfos, partName)
@ -44,20 +40,10 @@ func erasureReadFile(disks []StorageAPI, volume string, path string, partName st
// Pick one erasure info.
eInfo := pickValidErasureInfo(eInfos)
// Write until each parts are read and exhausted.
for totalSizeLeft > 0 {
// Calculate the proper block size.
var curBlockSize int64
if eInfo.BlockSize < totalSizeLeft {
curBlockSize = eInfo.BlockSize
} else {
curBlockSize = totalSizeLeft
}
// Calculate the current encoded block size.
curEncBlockSize := getEncodedBlockLen(curBlockSize, eInfo.DataBlocks)
offsetEncOffset := getEncodedBlockLen(startOffset, eInfo.DataBlocks)
// Get block info for given offset, length and block size.
startBlock, bytesToSkip, endBlock := getBlockInfo(offset, length, eInfo.BlockSize)
for block := startBlock; block <= endBlock; block++ {
// Allocate encoded blocks up to storage disks.
enBlocks := make([][]byte, len(disks))
@ -65,6 +51,10 @@ func erasureReadFile(disks []StorageAPI, volume string, path string, partName st
var successDataBlocksCount = 0
var noReconstruct bool // Set for no reconstruction.
// Keep how many bytes are read for this block.
// In most cases, last block in the file is shorter than eInfo.BlockSize.
lastReadSize := int64(0)
// Read from all the disks.
for index, disk := range disks {
blockIndex := eInfo.Distribution[index] - 1
@ -74,13 +64,29 @@ func erasureReadFile(disks []StorageAPI, volume string, path string, partName st
if disk == nil {
continue
}
// Initialize shard slice and fill the data from each parts.
enBlocks[blockIndex] = make([]byte, curEncBlockSize)
// Initialize chunk slice and fill the data from each parts.
enBlocks[blockIndex] = make([]byte, eInfo.BlockSize)
// Read the necessary blocks.
_, err := disk.ReadFile(volume, path, offsetEncOffset, enBlocks[blockIndex])
n, err := disk.ReadFile(volume, path, block*eInfo.BlockSize, enBlocks[blockIndex])
if err != nil {
enBlocks[blockIndex] = nil
} else if n < eInfo.BlockSize {
// As the data we got is smaller than eInfo.BlockSize, keep only required chunk slice
enBlocks[blockIndex] = append([]byte{}, enBlocks[blockIndex][:n]...)
}
// Remember bytes read at first time.
if lastReadSize == 0 {
lastReadSize = n
}
// If bytes read is not equal to bytes read lastly, treat it as corrupted chunk.
if n != lastReadSize {
return bytesWritten, errXLDataCorrupt
}
// Verify if we have successfully read all the data blocks.
if blockIndex < eInfo.DataBlocks && enBlocks[blockIndex] != nil {
successDataBlocksCount++
@ -95,38 +101,43 @@ func erasureReadFile(disks []StorageAPI, volume string, path string, partName st
}
}
// Check blocks if they are all zero in length, we have corruption return error.
if checkBlockSize(enBlocks) == 0 {
return nil, errXLDataCorrupt
}
// Verify if reconstruction is needed, proceed with reconstruction.
if !noReconstruct {
err := decodeData(enBlocks, eInfo.DataBlocks, eInfo.ParityBlocks)
if err != nil {
return nil, err
return bytesWritten, err
}
}
// Get data blocks from encoded blocks.
dataBlocks, err := getDataBlocks(enBlocks, eInfo.DataBlocks, int(curBlockSize))
dataBlocks, err := getDataBlocks(enBlocks, eInfo.DataBlocks, int(lastReadSize)*eInfo.DataBlocks)
if err != nil {
return nil, err
return bytesWritten, err
}
// Copy data blocks.
buffer = append(buffer, dataBlocks...)
// Keep required bytes into buf.
buf := dataBlocks
// Negate the 'n' size written to client.
totalSizeLeft -= int64(len(dataBlocks))
// If this is start block, skip unwanted bytes.
if block == startBlock {
buf = append([]byte{}, dataBlocks[bytesToSkip:]...)
}
// Increase the offset to move forward.
startOffset += int64(len(dataBlocks))
// If this is end block, retain only required bytes.
if block == endBlock {
buf = append([]byte{}, buf[:length-bytesWritten]...)
}
// Relenquish memory.
dataBlocks = nil
// Copy data blocks.
var n int64
n, err = io.Copy(writer, bytes.NewReader(buf))
bytesWritten += int64(n)
if err != nil {
return bytesWritten, err
}
}
return buffer, nil
return bytesWritten, nil
}
// PartObjectChecksum - returns the checksum for the part name from the checksum slice.

@ -88,21 +88,14 @@ func getDataBlocks(enBlocks [][]byte, dataBlocks int, curBlockSize int) (data []
return data, nil
}
// checkBlockSize return the size of a single block.
// The first non-zero size is returned,
// or 0 if all blocks are size 0.
func checkBlockSize(blocks [][]byte) int {
for _, block := range blocks {
if len(block) != 0 {
return len(block)
}
}
return 0
}
// getBlockInfo - find start/end block and bytes to skip for given offset, length and block size.
func getBlockInfo(offset, length, blockSize int64) (startBlock, bytesToSkip, endBlock int64) {
// Calculate start block for given offset and how many bytes to skip to get the offset.
startBlock = offset / blockSize
bytesToSkip = offset % blockSize
// Calculate end block for given size to read
endBlock = (offset + length) / blockSize
// calculate the blockSize based on input length and total number of
// data blocks.
func getEncodedBlockLen(inputLen int64, dataBlocks int) (curEncBlockSize int64) {
curEncBlockSize = (inputLen + int64(dataBlocks) - 1) / int64(dataBlocks)
return curEncBlockSize
return
}

@ -17,7 +17,6 @@
package main
import (
"bytes"
"crypto/md5"
"encoding/hex"
"io"
@ -71,46 +70,41 @@ func (xl xlObjects) GetObject(bucket, object string, startOffset int64, length i
}
}
// Get part index offset.
// Get start part index and offset.
partIndex, partOffset, err := xlMeta.ObjectToPartOffset(startOffset)
if err != nil {
return toObjectErr(err, bucket, object)
}
// Get last part index to read given length.
lastPartIndex, _, err := xlMeta.ObjectToPartOffset(startOffset + length - 1)
if err != nil {
return toObjectErr(err, bucket, object)
}
// Collect all the previous erasure infos across the disk.
var eInfos []erasureInfo
for index := range onlineDisks {
eInfos = append(eInfos, metaArr[index].Erasure)
}
totalBytesRead := int64(0)
// Read from all parts.
for ; partIndex < len(xlMeta.Parts); partIndex++ {
for ; partIndex <= lastPartIndex; partIndex++ {
// Save the current part name and size.
partName := xlMeta.Parts[partIndex].Name
partSize := xlMeta.Parts[partIndex].Size
if partSize > (length - totalBytesRead) {
partSize = length - totalBytesRead
}
// Start reading the part name.
var buffer []byte
buffer, err = erasureReadFile(onlineDisks, bucket, pathJoin(object, partName), partName, partSize, eInfos)
n, err := erasureReadFile(writer, onlineDisks, bucket, pathJoin(object, partName), partName, eInfos, partOffset, partSize)
if err != nil {
return err
}
// Copy to client until length requested.
if length > int64(len(buffer)) {
var m int64
m, err = io.Copy(writer, bytes.NewReader(buffer[partOffset:]))
if err != nil {
return err
}
length -= m
} else {
_, err = io.CopyN(writer, bytes.NewReader(buffer[partOffset:]), length)
if err != nil {
return err
}
return nil
}
totalBytesRead += n
// Reset part offset to 0 to read rest of the part from the beginning.
partOffset = 0

Loading…
Cancel
Save