mirror of
https://github.com/pgsty/minio.git
synced 2026-08-09 07:43:29 +03:00
fix: restore safe erasure read buffer pooling
Wire the preallocated pooled shard slices into parallelReader instead of discarding them and allocating a buffer for each disk. Keep readerToBuf as a permutation while preferred readers are reordered. The former assignments could duplicate a buffer slot after multiple swaps, causing concurrent writes and a possible decode stall. Add pool aliasing and mapping regression coverage. Co-authored-by: ChatGPT <noreply@openai.com> Co-authored-by: Claude <noreply@anthropic.com>
This commit is contained in:
@@ -69,7 +69,7 @@ func newParallelReader(readers []io.ReaderAt, e Erasure, offset, totalLength int
|
|||||||
offset: (offset / e.blockSize) * e.ShardSize(),
|
offset: (offset / e.blockSize) * e.ShardSize(),
|
||||||
shardSize: e.ShardSize(),
|
shardSize: e.ShardSize(),
|
||||||
shardFileSize: e.ShardFileSize(totalLength),
|
shardFileSize: e.ShardFileSize(totalLength),
|
||||||
buf: make([][]byte, len(readers)),
|
buf: bufs,
|
||||||
readerToBuf: r2b,
|
readerToBuf: r2b,
|
||||||
stashBuffer: b,
|
stashBuffer: b,
|
||||||
}
|
}
|
||||||
@@ -106,8 +106,7 @@ func (p *parallelReader) preferReaders(prefer []bool) {
|
|||||||
// Move reader with index i to index next.
|
// Move reader with index i to index next.
|
||||||
// Do this by swapping next and i
|
// Do this by swapping next and i
|
||||||
p.readers[next], p.readers[i] = p.readers[i], p.readers[next]
|
p.readers[next], p.readers[i] = p.readers[i], p.readers[next]
|
||||||
p.readerToBuf[next] = i
|
p.readerToBuf[next], p.readerToBuf[i] = p.readerToBuf[i], p.readerToBuf[next]
|
||||||
p.readerToBuf[i] = next
|
|
||||||
next++
|
next++
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -26,8 +26,92 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/dustin/go-humanize"
|
"github.com/dustin/go-humanize"
|
||||||
|
"github.com/minio/minio/internal/bpool"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func TestNewParallelReaderUsesPooledStashBuffer(t *testing.T) {
|
||||||
|
erasure, err := NewErasure(t.Context(), 4, 4, blockSizeV2)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
previousPool := globalBytePoolCap.Load()
|
||||||
|
pool := bpool.NewBytePoolCap(1, blockSizeV2, blockSizeV2*2)
|
||||||
|
globalBytePoolCap.Store(pool)
|
||||||
|
t.Cleanup(func() {
|
||||||
|
globalBytePoolCap.Store(previousPool)
|
||||||
|
})
|
||||||
|
|
||||||
|
readers := make([]io.ReaderAt, 8)
|
||||||
|
reader := newParallelReader(readers, erasure, 0, blockSizeV2)
|
||||||
|
t.Cleanup(reader.Done)
|
||||||
|
if reader.stashBuffer == nil {
|
||||||
|
t.Fatal("expected a pooled stash buffer")
|
||||||
|
}
|
||||||
|
|
||||||
|
shardSize := int(erasure.ShardSize())
|
||||||
|
stash := reader.stashBuffer[:cap(reader.stashBuffer)]
|
||||||
|
for i, buf := range reader.buf {
|
||||||
|
if len(buf) != shardSize {
|
||||||
|
t.Fatalf("buffer %d has length %d, want %d", i, len(buf), shardSize)
|
||||||
|
}
|
||||||
|
buf[0] = byte(i + 1)
|
||||||
|
if got := stash[i*shardSize]; got != byte(i+1) {
|
||||||
|
t.Fatalf("buffer %d does not use the pooled stash buffer", i)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestParallelReaderPreferReadersMaintainsBufferMapping(t *testing.T) {
|
||||||
|
tests := []struct {
|
||||||
|
name string
|
||||||
|
prefer []bool
|
||||||
|
want []int
|
||||||
|
}{
|
||||||
|
{
|
||||||
|
name: "adjacent preferred readers",
|
||||||
|
prefer: []bool{false, true, true, false},
|
||||||
|
want: []int{1, 2, 0, 3},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "sparse preferred readers",
|
||||||
|
prefer: []bool{false, true, false, true},
|
||||||
|
want: []int{1, 3, 2, 0},
|
||||||
|
},
|
||||||
|
{
|
||||||
|
name: "preferred prefix",
|
||||||
|
prefer: []bool{true, true, false, false},
|
||||||
|
want: []int{0, 1, 2, 3},
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
for _, tt := range tests {
|
||||||
|
t.Run(tt.name, func(t *testing.T) {
|
||||||
|
original := make([]io.ReaderAt, len(tt.prefer))
|
||||||
|
readerToBuf := make([]int, len(tt.prefer))
|
||||||
|
for i := range original {
|
||||||
|
original[i] = bytes.NewReader([]byte{byte(i)})
|
||||||
|
readerToBuf[i] = i
|
||||||
|
}
|
||||||
|
|
||||||
|
reader := parallelReader{
|
||||||
|
orgReaders: original,
|
||||||
|
readerToBuf: readerToBuf,
|
||||||
|
}
|
||||||
|
reader.preferReaders(tt.prefer)
|
||||||
|
|
||||||
|
for i, originalIndex := range tt.want {
|
||||||
|
if reader.readers[i] != original[originalIndex] {
|
||||||
|
t.Errorf("reader %d maps to the wrong original reader", i)
|
||||||
|
}
|
||||||
|
if got := reader.readerToBuf[i]; got != originalIndex {
|
||||||
|
t.Errorf("reader %d maps to buffer %d, want %d", i, got, originalIndex)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func (a badDisk) ReadFile(ctx context.Context, volume string, path string, offset int64, buf []byte, verifier *BitrotVerifier) (n int64, err error) {
|
func (a badDisk) ReadFile(ctx context.Context, volume string, path string, offset int64, buf []byte, verifier *BitrotVerifier) (n int64, err error) {
|
||||||
return 0, errFaultyDisk
|
return 0, errFaultyDisk
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user