From 92471792691826b5225e9b17890047ef7c83080b Mon Sep 17 00:00:00 2001 From: Feng Ruohang Date: Sat, 1 Aug 2026 11:06:24 +0800 Subject: [PATCH] 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 Co-authored-by: Claude --- cmd/erasure-decode.go | 5 +-- cmd/erasure-decode_test.go | 84 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 86 insertions(+), 3 deletions(-) diff --git a/cmd/erasure-decode.go b/cmd/erasure-decode.go index f0cc90ab0..8b6aeb72a 100644 --- a/cmd/erasure-decode.go +++ b/cmd/erasure-decode.go @@ -69,7 +69,7 @@ func newParallelReader(readers []io.ReaderAt, e Erasure, offset, totalLength int offset: (offset / e.blockSize) * e.ShardSize(), shardSize: e.ShardSize(), shardFileSize: e.ShardFileSize(totalLength), - buf: make([][]byte, len(readers)), + buf: bufs, readerToBuf: r2b, stashBuffer: b, } @@ -106,8 +106,7 @@ func (p *parallelReader) preferReaders(prefer []bool) { // Move reader with index i to index next. // Do this by swapping next and i p.readers[next], p.readers[i] = p.readers[i], p.readers[next] - p.readerToBuf[next] = i - p.readerToBuf[i] = next + p.readerToBuf[next], p.readerToBuf[i] = p.readerToBuf[i], p.readerToBuf[next] next++ } } diff --git a/cmd/erasure-decode_test.go b/cmd/erasure-decode_test.go index 229047e16..a4d832a44 100644 --- a/cmd/erasure-decode_test.go +++ b/cmd/erasure-decode_test.go @@ -26,8 +26,92 @@ import ( "testing" "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) { return 0, errFaultyDisk }