Merge pull request #827 from harshavardhana/krishnasrinivas-parallel-read2
Read from the disks in parallel during object read
This commit is contained in:
commit
87a20e3ba3
1 changed files with 22 additions and 4 deletions
|
@ -553,13 +553,31 @@ func (b bucket) decodeEncodedData(totalLeft, blockSize int64, readers map[int]io
|
|||
return nil, err.Trace()
|
||||
}
|
||||
encodedBytes := make([][]byte, encoder.k+encoder.m)
|
||||
errCh := make(chan error, len(readers))
|
||||
var errRet error
|
||||
var readCnt int
|
||||
|
||||
for i, reader := range readers {
|
||||
var bytesBuffer bytes.Buffer
|
||||
_, err := io.CopyN(&bytesBuffer, reader, int64(curChunkSize))
|
||||
go func(reader io.Reader, i int) {
|
||||
encodedBytes[i] = make([]byte, curChunkSize)
|
||||
_, err := io.ReadFull(reader, encodedBytes[i])
|
||||
if err != nil {
|
||||
encodedBytes[i] = nil
|
||||
errCh <- err
|
||||
return
|
||||
}
|
||||
errCh <- nil
|
||||
}(reader, i)
|
||||
// read through errCh for any errors
|
||||
err := <-errCh
|
||||
if err != nil {
|
||||
return nil, probe.NewError(err)
|
||||
errRet = err
|
||||
} else {
|
||||
readCnt++
|
||||
}
|
||||
encodedBytes[i] = bytesBuffer.Bytes()
|
||||
}
|
||||
if readCnt < int(encoder.k) {
|
||||
return nil, probe.NewError(errRet)
|
||||
}
|
||||
decodedData, err := encoder.Decode(encodedBytes, int(curBlockSize))
|
||||
if err != nil {
|
||||
|
|
Loading…
Reference in a new issue