mirror of
https://github.com/minio/minio.git
synced 2025-03-31 09:43:43 -04:00
XL: CreateFile/ReadFile should write and read from all disks in parallel. (#1612)
* XL: CreateFile should write to all disks in parallel. * XL: ReadFile should read from all disks in parallel.
This commit is contained in:
parent
7264cd2ab3
commit
025054fb36
@ -20,6 +20,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
slashpath "path"
|
slashpath "path"
|
||||||
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/Sirupsen/logrus"
|
"github.com/Sirupsen/logrus"
|
||||||
@ -199,24 +200,34 @@ func (xl XL) writeErasure(volume, path string, reader *io.PipeReader, wcloser *w
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
|
var wg = &sync.WaitGroup{}
|
||||||
|
var wErrs = make([]error, len(writers))
|
||||||
// Loop through and write encoded data to quorum disks.
|
// Loop through and write encoded data to quorum disks.
|
||||||
for index, writer := range writers {
|
for index, writer := range writers {
|
||||||
if writer == nil {
|
if writer == nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
encodedData := dataBlocks[index]
|
wg.Add(1)
|
||||||
_, err = writers[index].Write(encodedData)
|
go func(index int, writer io.Writer) {
|
||||||
if err != nil {
|
defer wg.Done()
|
||||||
log.WithFields(logrus.Fields{
|
encodedData := dataBlocks[index]
|
||||||
"volume": volume,
|
_, wErr := writers[index].Write(encodedData)
|
||||||
"path": path,
|
wErrs[index] = wErr
|
||||||
"diskIndex": index,
|
}(index, writer)
|
||||||
}).Errorf("Writing encoded blocks failed with %s", err)
|
}
|
||||||
// Remove all temp writers upon error.
|
wg.Wait()
|
||||||
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
|
for _, wErr := range wErrs {
|
||||||
reader.CloseWithError(err)
|
if wErr == nil {
|
||||||
return
|
continue
|
||||||
}
|
}
|
||||||
|
log.WithFields(logrus.Fields{
|
||||||
|
"volume": volume,
|
||||||
|
"path": path,
|
||||||
|
}).Errorf("Writing encoded blocks failed with %s", wErr)
|
||||||
|
// Remove all temp writers upon error.
|
||||||
|
xl.cleanupCreateFileOps(volume, path, append(writers, metadataWriters...)...)
|
||||||
|
reader.CloseWithError(wErr)
|
||||||
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Update total written.
|
// Update total written.
|
||||||
|
@ -21,6 +21,7 @@ import (
|
|||||||
"fmt"
|
"fmt"
|
||||||
"io"
|
"io"
|
||||||
slashpath "path"
|
slashpath "path"
|
||||||
|
"sync"
|
||||||
|
|
||||||
"github.com/Sirupsen/logrus"
|
"github.com/Sirupsen/logrus"
|
||||||
)
|
)
|
||||||
@ -95,6 +96,7 @@ func (xl XL) ReadFile(volume, path string, startOffset int64) (io.ReadCloser, er
|
|||||||
// Calculate the current encoded block size.
|
// Calculate the current encoded block size.
|
||||||
curEncBlockSize := getEncodedBlockLen(curBlockSize, metadata.Erasure.DataBlocks)
|
curEncBlockSize := getEncodedBlockLen(curBlockSize, metadata.Erasure.DataBlocks)
|
||||||
enBlocks := make([][]byte, len(xl.storageDisks))
|
enBlocks := make([][]byte, len(xl.storageDisks))
|
||||||
|
var wg = &sync.WaitGroup{}
|
||||||
// Loop through all readers and read.
|
// Loop through all readers and read.
|
||||||
for index, reader := range readers {
|
for index, reader := range readers {
|
||||||
// Initialize shard slice and fill the data from each parts.
|
// Initialize shard slice and fill the data from each parts.
|
||||||
@ -102,11 +104,19 @@ func (xl XL) ReadFile(volume, path string, startOffset int64) (io.ReadCloser, er
|
|||||||
if reader == nil {
|
if reader == nil {
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
_, err = io.ReadFull(reader, enBlocks[index])
|
// Parallelize reading.
|
||||||
if err != nil && err != io.ErrUnexpectedEOF {
|
wg.Add(1)
|
||||||
readers[index] = nil
|
go func(index int, reader io.Reader) {
|
||||||
}
|
defer wg.Done()
|
||||||
|
// Read the necessary blocks.
|
||||||
|
_, rErr := io.ReadFull(reader, enBlocks[index])
|
||||||
|
if rErr != nil && rErr != io.ErrUnexpectedEOF {
|
||||||
|
readers[index] = nil
|
||||||
|
}
|
||||||
|
}(index, reader)
|
||||||
}
|
}
|
||||||
|
// Wait for the read routines to finish.
|
||||||
|
wg.Wait()
|
||||||
|
|
||||||
// Check blocks if they are all zero in length.
|
// Check blocks if they are all zero in length.
|
||||||
if checkBlockSize(enBlocks) == 0 {
|
if checkBlockSize(enBlocks) == 0 {
|
||||||
|
@ -109,8 +109,7 @@ func newXLObjects(exportPaths ...string) (ObjectLayer, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
// Validate if format exists and input arguments are validated
|
// Validate if format exists and input arguments are validated with backend format.
|
||||||
// with backend format.
|
|
||||||
if !isValidFormat(storage, exportPaths...) {
|
if !isValidFormat(storage, exportPaths...) {
|
||||||
return nil, fmt.Errorf("Command-line arguments %s is not valid.", exportPaths)
|
return nil, fmt.Errorf("Command-line arguments %s is not valid.", exportPaths)
|
||||||
}
|
}
|
||||||
|
Loading…
x
Reference in New Issue
Block a user