Files
seaweedfs/weed/filer/reader_at_shared_test.go
T

478 lines
17 KiB
Go

package filer
import (
"context"
"fmt"
"io"
"sync"
"sync/atomic"
"testing"
"github.com/seaweedfs/seaweedfs/weed/util/chunk_cache"
util_http "github.com/seaweedfs/seaweedfs/weed/util/http"
)
// Two streams (e.g. two S3 GETs) reading the same object through one shared
// ReaderCache, interleaved in small slices, must share each chunk download:
// one stream finishing a chunk, or moving on to the next one, must not drop a
// buffer the other stream is still reading.
func TestChunkReadAtConcurrentStreamsShareChunks(t *testing.T) {
const chunkSize = 64 << 10
const chunkCount = 3
const sliceSize = 16 << 10
var mu sync.Mutex
fetches := map[string]int{}
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil)
defer rc.destroy()
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
mu.Lock()
fetches[fileId]++
mu.Unlock()
for i := range buffer {
buffer[i] = fileId[len(fileId)-1]
}
return len(buffer), nil
}
newStream := func() *ChunkReadAt {
views := NewIntervalList[*ChunkView]()
for i := 0; i < chunkCount; i++ {
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: int64(i * chunkSize),
StopOffset: int64((i + 1) * chunkSize),
Value: &ChunkView{
FileId: fmt.Sprintf("chunk%d", i),
ViewSize: chunkSize,
ViewOffset: int64(i * chunkSize),
ChunkSize: chunkSize,
},
})
}
return NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize*chunkCount, 0)
}
streams := []*ChunkReadAt{newStream(), newStream()}
// The streams alternate slice by slice, the second one lagging one slice
// behind, so the leader always finishes a chunk while the follower is
// still inside it.
offsets := make([]int64, len(streams))
offsets[1] = -sliceSize
for offsets[1] < chunkSize*chunkCount {
for i, stream := range streams {
if offsets[i] < 0 || offsets[i] >= chunkSize*chunkCount {
offsets[i] += sliceSize
continue
}
buf := make([]byte, sliceSize)
n, err := stream.ReadAt(buf, offsets[i])
if (err != nil && err != io.EOF) || n != sliceSize {
t.Fatalf("stream %d at %d: n=%d err=%v", i, offsets[i], n, err)
}
if want := byte('0' + offsets[i]/chunkSize); buf[0] != want || buf[n-1] != want {
t.Fatalf("stream %d at %d: got %q, want %q", i, offsets[i], buf[0], want)
}
offsets[i] += sliceSize
}
}
mu.Lock()
defer mu.Unlock()
for i := 0; i < chunkCount; i++ {
fileId := fmt.Sprintf("chunk%d", i)
if fetches[fileId] != 1 {
t.Errorf("%s fetched %d times for two concurrent streams, want 1", fileId, fetches[fileId])
}
}
}
func newStreamTestReaderCache(chunkCache chunk_cache.ChunkCache) *ReaderCache {
rc := NewReaderCache(64, chunkCache, func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil)
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
return len(buffer), nil
}
return rc
}
func isRetained(rc *ReaderCache, fileId string) bool {
rc.Lock()
defer rc.Unlock()
_, found := rc.downloaders[fileId]
return found
}
// A stream moving on to a chunk served from the chunk cache must still
// release the chunk it was positioned in before.
func TestChunkStreamReleasesPreviousChunkOnCacheHit(t *testing.T) {
cache := newMockChunkCacheForReaderCache()
cache.SetChunk("chunk1", make([]byte, 4<<10))
rc := newStreamTestReaderCache(cache)
defer rc.destroy()
stream := &chunkStream{}
if _, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk0", nil, false, 0, 4<<10, false); err != nil {
t.Fatal(err)
}
if n, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk1", nil, false, 0, 4<<10, true); err != nil || n == 0 {
t.Fatalf("cache hit read: n=%d err=%v", n, err)
}
if isRetained(rc, "chunk0") {
t.Fatal("chunk0 still retained after the stream moved on to a cached chunk")
}
}
// A chunk the stream leaves while another read is in flight must be dropped
// once that read ends, even if it did not read the chunk to the end.
func TestChunkStreamDropsLeftChunkAfterInFlightRead(t *testing.T) {
rc := newStreamTestReaderCache(newMockChunkCacheForReaderCache())
defer rc.destroy()
stream := &chunkStream{}
if _, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk0", nil, false, 0, 4<<10, false); err != nil {
t.Fatal(err)
}
// Another reader is in the middle of a partial read of chunk0.
rc.Lock()
other := rc.downloaders["chunk0"]
other.wg.Add(1)
atomic.AddInt32(&other.readers, 1)
rc.Unlock()
if _, err := rc.readChunkAt(context.Background(), stream, make([]byte, 1<<10), "chunk1", nil, false, 0, 4<<10, false); err != nil {
t.Fatal(err)
}
if !isRetained(rc, "chunk0") {
t.Fatal("chunk0 dropped while a read was still in flight")
}
// The in-flight read ends without reaching the end of the chunk.
other.wg.Done()
atomic.AddInt32(&other.readers, -1)
rc.removeConsumed(other)
if isRetained(rc, "chunk0") {
t.Fatal("chunk0 retained after the stream left it and the last read ended")
}
}
// Concurrent ReadAt calls on one ChunkReadAt (as mount does) share its stream
// pin; they must neither race on it nor leak or double-release pins.
func TestChunkStreamConcurrentReadsOnOneReader(t *testing.T) {
const chunkSize = 16 << 10
const chunkCount = 4
const sliceSize = 4 << 10
rc := newStreamTestReaderCache((*chunk_cache.TieredChunkCache)(nil))
defer rc.destroy()
views := NewIntervalList[*ChunkView]()
for i := 0; i < chunkCount; i++ {
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: int64(i * chunkSize),
StopOffset: int64((i + 1) * chunkSize),
Value: &ChunkView{
FileId: fmt.Sprintf("chunk%d", i),
ViewSize: chunkSize,
ViewOffset: int64(i * chunkSize),
ChunkSize: chunkSize,
},
})
}
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize*chunkCount, 0)
var wg sync.WaitGroup
for g := 0; g < 8; g++ {
wg.Add(1)
go func() {
defer wg.Done()
for offset := int64(0); offset < chunkSize*chunkCount; offset += sliceSize {
if _, err := reader.ReadAt(make([]byte, sliceSize), offset); err != nil && err != io.EOF {
t.Error(err)
return
}
}
}()
}
wg.Wait()
// Finishing the last chunk releases the stream's final pin.
if _, err := reader.ReadAt(make([]byte, sliceSize), chunkSize*chunkCount-sliceSize); err != nil && err != io.EOF {
t.Fatal(err)
}
rc.Lock()
defer rc.Unlock()
for fileId, cacher := range rc.downloaders {
t.Errorf("%s retained after all reads finished: pins=%d readers=%d", fileId, atomic.LoadInt32(&cacher.pins), atomic.LoadInt32(&cacher.readers))
}
}
type recordedFetch struct {
fileId string
isFullChunk bool
offset int64
size int
}
// fetchRecorder stubs the volume fetch and records how each chunk was
// requested: isFullChunk=false is a range fetch of just the view's slice,
// isFullChunk=true is a whole-chunk download into the shared cache.
func fetchRecorder(rc *ReaderCache) (fetches *[]recordedFetch) {
var mu sync.Mutex
recorded := &[]recordedFetch{}
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, isFullChunk bool, offset int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
mu.Lock()
*recorded = append(*recorded, recordedFetch{fileId, isFullChunk, offset, len(buffer)})
mu.Unlock()
for i := range buffer {
buffer[i] = fileId[len(fileId)-1]
}
return len(buffer), nil
}
return recorded
}
// A reader whose views are clipped to a request window — how the S3 gateway
// builds a ranged GET — must fetch only the covered part of each chunk:
// clipped edge views take range fetches, a fully covered chunk keeps the
// shared whole-chunk path. This is what keeps a ranged GET larger than a
// buffer from multiplying volume-server reads (issue #11564), without giving
// up whole-chunk caching where the whole chunk is actually wanted.
func TestChunkReadAtClippedViewsFetchOnlyCoveredParts(t *testing.T) {
const chunkSize = 64 << 10
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil)
defer rc.destroy()
fetches := fetchRecorder(rc)
// Window [56KiB, 152KiB): tail of chunk0, all of chunk1, head of
// chunk2, head of ciphered chunk3, head of compressed chunk4 (file
// chunks need not be aligned).
views := NewIntervalList[*ChunkView]()
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: chunkSize - 8<<10,
StopOffset: chunkSize,
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: chunkSize - 8<<10, ViewSize: 8 << 10, ViewOffset: chunkSize - 8<<10, ChunkSize: chunkSize},
})
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: chunkSize,
StopOffset: 2 * chunkSize,
Value: &ChunkView{FileId: "chunk1", ViewSize: chunkSize, ViewOffset: chunkSize, ChunkSize: chunkSize},
})
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: 2 * chunkSize,
StopOffset: 2*chunkSize + 8<<10,
Value: &ChunkView{FileId: "chunk2", ViewSize: 8 << 10, ViewOffset: 2 * chunkSize, ChunkSize: chunkSize},
})
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: 2*chunkSize + 8<<10,
StopOffset: 2*chunkSize + 16<<10,
Value: &ChunkView{FileId: "chunk3", ViewSize: 8 << 10, ViewOffset: 2*chunkSize + 8<<10, ChunkSize: chunkSize, CipherKey: []byte("key")},
})
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: 2*chunkSize + 16<<10,
StopOffset: 2*chunkSize + 24<<10,
Value: &ChunkView{FileId: "chunk4", ViewSize: 8 << 10, ViewOffset: 2*chunkSize + 16<<10, ChunkSize: chunkSize, IsGzipped: true},
})
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, 4*chunkSize, 0)
buf := make([]byte, chunkSize+32<<10)
if n, err := reader.ReadAt(buf, chunkSize-8<<10); err != nil || n != len(buf) {
t.Fatalf("window read: n=%d err=%v", n, err)
}
// buf holds [56KiB, 152KiB): chunk0's tail, chunk1, and the heads of
// chunk2, chunk3 and chunk4.
for i, b := range buf {
want := byte('1')
if i < 8<<10 {
want = '0'
} else if i >= 24<<10+chunkSize {
want = '4'
} else if i >= 16<<10+chunkSize {
want = '3'
} else if i >= 8<<10+chunkSize {
want = '2'
}
if b != want {
t.Fatalf("buf[%d]=%q, want %q", i, b, want)
}
}
want := []recordedFetch{
{fileId: "chunk0", isFullChunk: false, offset: chunkSize - 8<<10, size: 8 << 10},
{fileId: "chunk1", isFullChunk: true, offset: 0, size: chunkSize},
{fileId: "chunk2", isFullChunk: false, offset: 0, size: 8 << 10},
// partial views, but ciphered and compressed chunks download whole
// either way and the shared path decrypts/decompresses once for
// every buffer
{fileId: "chunk3", isFullChunk: true, offset: 0, size: chunkSize},
{fileId: "chunk4", isFullChunk: true, offset: 0, size: chunkSize},
}
got := map[string]recordedFetch{}
for _, f := range *fetches {
if _, dup := got[f.fileId]; dup {
t.Fatalf("chunk %s fetched more than once: %+v", f.fileId, *fetches)
}
got[f.fileId] = f
}
for _, w := range want {
if g, ok := got[w.fileId]; !ok {
t.Fatalf("chunk %s never fetched: %+v", w.fileId, *fetches)
} else if g != w {
t.Fatalf("chunk %s fetched as %+v, want %+v", w.fileId, g, w)
}
}
}
// The regression from issue #11564: a ranged GET sitting inside one big
// chunk. Every buffer of the request must stay a range fetch — none may
// escalate into a whole-chunk download once the reads look sequential.
func TestChunkReadAtRangeInsideOneChunkStaysRangeFetch(t *testing.T) {
const chunkSize = 1 << 20
const sliceSize = 16 << 10
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil)
defer rc.destroy()
fetches := fetchRecorder(rc)
// Range [32KiB, 96KiB) inside one 1MiB chunk: a single clipped view.
views := NewIntervalList[*ChunkView]()
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: 32 << 10,
StopOffset: 96 << 10,
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: 32 << 10, ViewSize: 64 << 10, ViewOffset: 32 << 10, ChunkSize: chunkSize},
})
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize, 0)
for offset := int64(32 << 10); offset < 96<<10; offset += sliceSize {
buf := make([]byte, sliceSize)
if n, err := reader.ReadAt(buf, offset); err != nil || n != sliceSize {
t.Fatalf("read at %d: n=%d err=%v", offset, n, err)
}
}
if len(*fetches) != 4 {
t.Fatalf("got %d fetches, want 4 range fetches: %+v", len(*fetches), *fetches)
}
for i, f := range *fetches {
wantOffset := int64(32<<10) + int64(i)*sliceSize
if f.isFullChunk || f.offset != wantOffset || f.size != sliceSize {
t.Fatalf("fetch %d = %+v, want range fetch offset=%d size=%d", i, f, wantOffset, sliceSize)
}
}
}
// A compressed chunk larger than the reader cache budget can never be
// downloaded whole — the budget rejects the buffer — so its partial view
// must fall back to a range fetch even though each range costs a full
// decompress server-side. The alternative is a failed GET.
func TestChunkReadAtOversizedCompressedChunkFallsBackToRange(t *testing.T) {
const chunkSize = 1 << 20
budget := NewReaderCacheBudget(64 << 10) // smaller than the chunk
rc := NewReaderCache(64, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil, budget)
defer rc.destroy()
fetches := fetchRecorder(rc)
views := NewIntervalList[*ChunkView]()
views.AppendInterval(&Interval[*ChunkView]{
StartOffset: 32 << 10,
StopOffset: 64 << 10,
Value: &ChunkView{FileId: "chunk0", OffsetInChunk: 32 << 10, ViewSize: 32 << 10, ViewOffset: 32 << 10, ChunkSize: chunkSize, IsGzipped: true},
})
reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize, 0)
buf := make([]byte, 32<<10)
if n, err := reader.ReadAt(buf, 32<<10); err != nil || n != len(buf) {
t.Fatalf("read: n=%d err=%v", n, err)
}
if len(*fetches) != 1 {
t.Fatalf("got %d fetches, want 1 range fetch: %+v", len(*fetches), *fetches)
}
if f := (*fetches)[0]; f.isFullChunk || f.offset != 32<<10 || f.size != 32<<10 {
t.Fatalf("fetch = %+v, want range fetch offset=%d size=%d", f, 32<<10, 32<<10)
}
}
// A chunk a stream is positioned in must outlast downloader-limit eviction:
// otherwise a busy cache drops the buffer mid-stream and forces a refetch.
func TestChunkReadAtPinnedChunkSurvivesEviction(t *testing.T) {
const chunkSize = 64 << 10
var fetches int32
rc := NewReaderCache(2, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil)
defer rc.destroy()
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) {
if fileId == "chunk0" {
atomic.AddInt32(&fetches, 1)
}
return len(buffer), nil
}
stream := &chunkStream{}
buf := make([]byte, 16<<10)
// One slice in: the stream is positioned in chunk0 but has not left it.
if _, err := rc.readChunkAt(context.Background(), stream, buf, "chunk0", nil, false, 0, chunkSize, false); err != nil {
t.Fatal(err)
}
// Fill the downloader map past its limit with unpinned chunks.
for _, fileId := range []string{"chunk1", "chunk2", "chunk3"} {
if _, err := rc.ReadChunkAt(context.Background(), buf, fileId, nil, false, 0, chunkSize, true); err != nil {
t.Fatalf("read %s: %v", fileId, err)
}
}
// The stream's next slice must come from the still-pinned buffer.
if _, err := rc.readChunkAt(context.Background(), stream, buf, "chunk0", nil, false, 16<<10, chunkSize, false); err != nil {
t.Fatal(err)
}
if got := atomic.LoadInt32(&fetches); got != 1 {
t.Errorf("chunk0 fetched %d times, want 1", got)
}
}
// When every downloader is pinned the limit still applies: the oldest pinned
// buffer is evicted so abandoned streams cannot grow memory past the limit.
func TestChunkReadAtPinnedEvictionFallsBackWhenAllPinned(t *testing.T) {
const chunkSize = 64 << 10
rc := NewReaderCache(2, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) {
return []string{"unused"}, nil
}, nil)
defer rc.destroy()
rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, _ string, _ util_http.RefreshUrlsFunc) (int, error) {
return len(buffer), nil
}
buf := make([]byte, 16<<10)
streamA := &chunkStream{}
streamB := &chunkStream{}
if _, err := rc.readChunkAt(context.Background(), streamA, buf, "chunk0", nil, false, 0, chunkSize, false); err != nil {
t.Fatal(err)
}
if _, err := rc.readChunkAt(context.Background(), streamB, buf, "chunk1", nil, false, 0, chunkSize, false); err != nil {
t.Fatal(err)
}
// Every downloader is now pinned; the next chunk must still get in.
if _, err := rc.readChunkAt(context.Background(), &chunkStream{}, buf, "chunk2", nil, false, 0, chunkSize, false); err != nil {
t.Fatal(err)
}
if isRetained(rc, "chunk0") {
t.Fatal("oldest pinned downloader was not evicted past the limit")
}
if !isRetained(rc, "chunk2") {
t.Fatal("new downloader missing after pinned fallback eviction")
}
}