|
| 1 | +diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go |
| 2 | +--- a/weed/filer/reader_cache.go |
| 3 | ++++ b/weed/filer/reader_cache.go |
| 4 | +@@ -75,8 +75,10 @@ |
| 5 | + return |
| 6 | + } |
| 7 | + |
| 8 | +- cached := 0 |
| 9 | +- for x := chunkViews; x != nil && cached < count; x = x.Next { |
| 10 | ++ // The window is the next count chunks, whether or not a download for them |
| 11 | ++ // is already running: counting only newly started downloads would push the |
| 12 | ++ // window count chunks further on every call. |
| 13 | ++ for x, i := chunkViews, 0; x != nil && i < count; x, i = x.Next, i+1 { |
| 14 | + chunkView := x.Value |
| 15 | + if _, found := rc.downloaders[chunkView.FileId]; found { |
| 16 | + continue |
| 17 | +@@ -98,7 +100,6 @@ |
| 18 | + go cacher.startCaching() |
| 19 | + <-cacher.cacheStartedCh |
| 20 | + rc.downloaders[chunkView.FileId] = cacher |
| 21 | +- cached++ |
| 22 | + } |
| 23 | + |
| 24 | + return |
| 25 | +diff --git a/weed/filer/reader_at_prefetch_window_test.go b/weed/filer/reader_at_prefetch_window_test.go |
| 26 | +new file mode 100644 |
| 27 | +--- /dev/null |
| 28 | ++++ b/weed/filer/reader_at_prefetch_window_test.go |
| 29 | +@@ -0,0 +1,73 @@ |
| 30 | ++package filer |
| 31 | ++ |
| 32 | ++import ( |
| 33 | ++ "context" |
| 34 | ++ "fmt" |
| 35 | ++ "strconv" |
| 36 | ++ "strings" |
| 37 | ++ "sync/atomic" |
| 38 | ++ "testing" |
| 39 | ++ |
| 40 | ++ "github.com/seaweedfs/seaweedfs/weed/util/chunk_cache" |
| 41 | ++ util_http "github.com/seaweedfs/seaweedfs/weed/util/http" |
| 42 | ++) |
| 43 | ++ |
| 44 | ++// A sequential reader that consumes a chunk in several small ReadAt calls, as |
| 45 | ++// the S3 GET path does with its 256 KiB copy buffer over 4 MiB chunks, must |
| 46 | ++// keep its read-ahead at prefetchCount chunks. Each ReadAt triggers a |
| 47 | ++// prefetch; if chunks already being fetched are not counted toward the |
| 48 | ++// window, every call starts prefetchCount more downloads and the read-ahead |
| 49 | ++// runs to the ReaderCache downloader limit, buffering most of the object. |
| 50 | ++func TestChunkReadAtSequentialPrefetchStaysWithinWindow(t *testing.T) { |
| 51 | ++ const chunkSize = 64 << 10 |
| 52 | ++ const chunkCount = 64 |
| 53 | ++ const sliceSize = 16 << 10 |
| 54 | ++ |
| 55 | ++ var readerChunk atomic.Int64 |
| 56 | ++ var maxLead atomic.Int64 |
| 57 | ++ rc := NewReaderCache(256, (*chunk_cache.TieredChunkCache)(nil), func(context.Context, string) ([]string, error) { |
| 58 | ++ return []string{"unused"}, nil |
| 59 | ++ }, nil) |
| 60 | ++ defer rc.destroy() |
| 61 | ++ rc.fetchChunkDataFn = func(_ context.Context, buffer []byte, _ []string, _ []byte, _ bool, _ bool, _ int64, fileId string, _ util_http.RefreshUrlsFunc) (int, error) { |
| 62 | ++ idx, err := strconv.Atoi(strings.TrimPrefix(fileId, "chunk")) |
| 63 | ++ if err != nil { |
| 64 | ++ return 0, err |
| 65 | ++ } |
| 66 | ++ lead := int64(idx) - readerChunk.Load() |
| 67 | ++ for { |
| 68 | ++ cur := maxLead.Load() |
| 69 | ++ if lead <= cur || maxLead.CompareAndSwap(cur, lead) { |
| 70 | ++ break |
| 71 | ++ } |
| 72 | ++ } |
| 73 | ++ return len(buffer), nil |
| 74 | ++ } |
| 75 | ++ |
| 76 | ++ views := NewIntervalList[*ChunkView]() |
| 77 | ++ for i := 0; i < chunkCount; i++ { |
| 78 | ++ views.AppendInterval(&Interval[*ChunkView]{ |
| 79 | ++ StartOffset: int64(i * chunkSize), |
| 80 | ++ StopOffset: int64((i + 1) * chunkSize), |
| 81 | ++ Value: &ChunkView{ |
| 82 | ++ FileId: fmt.Sprintf("chunk%d", i), |
| 83 | ++ ViewSize: chunkSize, |
| 84 | ++ ViewOffset: int64(i * chunkSize), |
| 85 | ++ ChunkSize: chunkSize, |
| 86 | ++ }, |
| 87 | ++ }) |
| 88 | ++ } |
| 89 | ++ reader := NewChunkReaderAtFromClient(context.Background(), rc, views, chunkSize*chunkCount, DefaultPrefetchCount) |
| 90 | ++ |
| 91 | ++ buf := make([]byte, sliceSize) |
| 92 | ++ for off := int64(0); off < chunkSize*chunkCount; off += sliceSize { |
| 93 | ++ readerChunk.Store(off / chunkSize) |
| 94 | ++ if _, err := reader.ReadAt(buf, off); err != nil && off+sliceSize < chunkSize*chunkCount { |
| 95 | ++ t.Fatalf("ReadAt(%d): %v", off, err) |
| 96 | ++ } |
| 97 | ++ } |
| 98 | ++ |
| 99 | ++ if got := maxLead.Load(); got > DefaultPrefetchCount { |
| 100 | ++ t.Fatalf("read-ahead reached %d chunks past the reader, want at most %d", got, DefaultPrefetchCount) |
| 101 | ++ } |
| 102 | ++} |
0 commit comments