diff --git a/weed/command/filer.go b/weed/command/filer.go index 42099c19a..97cd923aa 100644 --- a/weed/command/filer.go +++ b/weed/command/filer.go @@ -168,6 +168,7 @@ func init() { filerS3Options.externalUrl = cmdFiler.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.") filerS3Options.defaultFileMode = cmdFiler.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666") filerS3Options.cacheSizeMB = cmdFiler.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)") + filerS3Options.readerCacheSizeMB = cmdFiler.Flag.Int64("s3.readerCacheSizeMB", 0, "memory budget in MiB for downloaded and in-flight reader buffers across all S3 GETs (0 means unlimited)") filerS3Options.allowUntrustedRemoteEndpoints = cmdFiler.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage) // start webdav on filer diff --git a/weed/command/mini.go b/weed/command/mini.go index 8ce45436a..5d8a54915 100644 --- a/weed/command/mini.go +++ b/weed/command/mini.go @@ -531,6 +531,7 @@ func initMiniS3Flags() { miniS3Options.externalUrl = cmdMini.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.") miniS3Options.defaultFileMode = cmdMini.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666") miniS3Options.cacheSizeMB = cmdMini.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)") + miniS3Options.readerCacheSizeMB = cmdMini.Flag.Int64("s3.readerCacheSizeMB", 0, "memory budget in MiB for downloaded and in-flight reader buffers across all S3 GETs (0 means unlimited)") miniS3Options.allowUntrustedRemoteEndpoints = cmdMini.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage) // In mini mode, S3 uses the shared debug server started at line 681, not its own separate debug server miniS3Options.debug = new(bool) // explicitly false diff --git a/weed/command/s3.go b/weed/command/s3.go index 6ca45f057..d054aac62 100644 --- a/weed/command/s3.go +++ b/weed/command/s3.go @@ -80,6 +80,7 @@ type S3Options struct { externalUrl *string defaultFileMode *string cacheSizeMB *int64 + readerCacheSizeMB *int64 allowUntrustedRemoteEndpoints *bool // shutdownCtx, when non-nil, tells startS3Server/startIcebergServer to @@ -129,6 +130,7 @@ func init() { s3StandaloneOptions.externalUrl = cmdS3.Flag.String("externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.") s3StandaloneOptions.defaultFileMode = cmdS3.Flag.String("defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666") s3StandaloneOptions.cacheSizeMB = cmdS3.Flag.Int64("cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)") + s3StandaloneOptions.readerCacheSizeMB = cmdS3.Flag.Int64("readerCacheSizeMB", 0, "memory budget in MiB for downloaded and in-flight reader buffers across all S3 GETs (0 means unlimited)") s3StandaloneOptions.allowUntrustedRemoteEndpoints = cmdS3.Flag.Bool("allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage) } @@ -354,6 +356,11 @@ func (s3opt *S3Options) startS3Server() bool { glog.Fatalf("S3 API Server startup error: %v", fileModeErr) } + var readerCacheSizeMB int64 + if s3opt.readerCacheSizeMB != nil { + readerCacheSizeMB = *s3opt.readerCacheSizeMB + } + s3ApiServer, s3ApiServer_err = s3api.NewS3ApiServer(router, &s3api.S3ApiServerOption{ Filers: filerAddresses, Masters: masterAddresses, @@ -380,6 +387,7 @@ func (s3opt *S3Options) startS3Server() bool { ExternalUrl: s3opt.resolveExternalUrl(), DefaultFileMode: defaultFileMode, CacheSizeMB: *s3opt.cacheSizeMB, + ReaderCacheSizeMB: readerCacheSizeMB, MaxMB: filerMaxMB, AllowUntrustedRemoteEndpoints: *s3opt.allowUntrustedRemoteEndpoints, diff --git a/weed/command/server.go b/weed/command/server.go index 2942dd035..7ff8b9449 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -192,6 +192,7 @@ func init() { s3Options.externalUrl = cmdServer.Flag.String("s3.externalUrl", "", "the external URL clients use to connect (e.g. https://api.example.com:9000). Advertised to Iceberg and Lance clients, and tried first when verifying S3 signatures behind a reverse proxy. Falls back to S3_EXTERNAL_URL env var.") s3Options.defaultFileMode = cmdServer.Flag.String("s3.defaultFileMode", "", "default file mode for S3 uploaded objects, e.g. 0660, 0644, 0666") s3Options.cacheSizeMB = cmdServer.Flag.Int64("s3.cacheCapacityMB", 0, "in-memory chunk cache capacity in MB for S3 GETs shared across requests (0 disables)") + s3Options.readerCacheSizeMB = cmdServer.Flag.Int64("s3.readerCacheSizeMB", 0, "memory budget in MiB for downloaded and in-flight reader buffers across all S3 GETs (0 means unlimited)") s3Options.allowUntrustedRemoteEndpoints = cmdServer.Flag.Bool("s3.allowUntrustedRemoteEndpoints", false, allowUntrustedRemoteEndpointsUsage) sftpOptions.port = cmdServer.Flag.Int("sftp.port", 2022, "SFTP server listen port") diff --git a/weed/filer/reader_cache.go b/weed/filer/reader_cache.go index 2177510bc..b54ff6ecf 100644 --- a/weed/filer/reader_cache.go +++ b/weed/filer/reader_cache.go @@ -33,6 +33,8 @@ type ReaderCache struct { type SingleChunkCacher struct { completedTimeNew int64 + readers int32 + consumed int32 sync.Mutex parent *ReaderCache chunkFileId string @@ -52,9 +54,6 @@ func NewReaderCache(limit int, chunkCache chunk_cache.ChunkCache, lookupFileIdFn if len(budgets) > 0 { budget = budgets[0] } - if budget == nil { - budget = NewReaderCacheBudget(DefaultReaderCacheMemoryLimit) - } return &ReaderCache{ limit: limit, budget: budget, @@ -130,6 +129,7 @@ retry: // concurrent destroy() (error eviction here, LRU, or UnCache) cannot // start wg.Wait() on a zero counter while this read is about to register. cacher.wg.Add(1) + atomic.AddInt32(&cacher.readers, 1) rc.Unlock() n, err := cacher.readChunkAt(ctx, buffer, offset) if n > 0 || err != nil { @@ -174,6 +174,7 @@ retry: <-cacher.cacheStartedCh rc.downloaders[fileId] = cacher cacher.wg.Add(1) + atomic.AddInt32(&cacher.readers, 1) rc.Unlock() return cacher.readChunkAt(ctx, buffer, offset) @@ -191,11 +192,32 @@ func (rc *ReaderCache) UnCache(fileId string) { func (rc *ReaderCache) remove(downloader *SingleChunkCacher) { rc.Lock() - if rc.downloaders[downloader.chunkFileId] == downloader { + removed := rc.downloaders[downloader.chunkFileId] == downloader + if removed { delete(rc.downloaders, downloader.chunkFileId) } rc.Unlock() - downloader.destroy() + if removed { + downloader.destroy() + } +} + +// removeConsumed drops a cacher once its buffer was fully read and no +// readers remain attached. The checks run under the ReaderCache lock so a +// reader attaching at the same time either wins (the cacher stays and that +// reader's detach retries the removal) or misses the map and refetches. +func (rc *ReaderCache) removeConsumed(downloader *SingleChunkCacher) { + rc.Lock() + removed := rc.downloaders[downloader.chunkFileId] == downloader && + atomic.LoadInt32(&downloader.readers) == 0 && + atomic.LoadInt32(&downloader.consumed) != 0 + if removed { + delete(rc.downloaders, downloader.chunkFileId) + } + rc.Unlock() + if removed { + downloader.destroy() + } } func (rc *ReaderCache) destroy() { @@ -333,8 +355,12 @@ func (s *SingleChunkCacher) destroy() { // The ctx parameter allows the reader to cancel its wait (but the download continues // for other readers - see comment in startCaching about shared resource semantics). // The caller must s.wg.Add(1) under the ReaderCache lock before calling; this only releases it. -func (s *SingleChunkCacher) readChunkAt(ctx context.Context, buf []byte, offset int64) (int, error) { - defer s.wg.Done() +func (s *SingleChunkCacher) readChunkAt(ctx context.Context, buf []byte, offset int64) (n int, err error) { + defer func() { + s.wg.Done() + atomic.AddInt32(&s.readers, -1) + s.parent.removeConsumed(s) + }() // Wait for download to complete, but allow reader cancellation. // Prioritize checking done first - if data is already available, @@ -364,5 +390,9 @@ func (s *SingleChunkCacher) readChunkAt(ctx context.Context, buf []byte, offset return 0, nil } - return copy(buf, s.data[offset:]), nil + n = copy(buf, s.data[offset:]) + if offset+int64(n) == int64(len(s.data)) { + atomic.StoreInt32(&s.consumed, 1) + } + return n, nil } diff --git a/weed/filer/reader_cache_budget.go b/weed/filer/reader_cache_budget.go index fc08869f0..c932235b0 100644 --- a/weed/filer/reader_cache_budget.go +++ b/weed/filer/reader_cache_budget.go @@ -36,9 +36,12 @@ func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error { if s.chunkSize < 0 { return fmt.Errorf("invalid chunk size %d", s.chunkSize) } + if b == nil { + return nil + } size := int64(mem.AllocationSize(s.chunkSize)) if size > b.limit { - return fmt.Errorf("chunk buffer needs %d bytes, exceeding reader cache budget %d; increase -readerCacheSizeMB", size, b.limit) + return fmt.Errorf("chunk buffer needs %d bytes, exceeding reader cache budget %d; increase the readerCacheSizeMB budget", size, b.limit) } for { b.Lock() @@ -63,6 +66,9 @@ func (b *ReaderCacheBudget) reserve(s *SingleChunkCacher) error { } func (b *ReaderCacheBudget) complete(s *SingleChunkCacher) { + if b == nil { + return + } b.Lock() defer b.Unlock() if _, found := b.reservations[s]; found && b.idleEntries[s] == nil { @@ -73,6 +79,9 @@ func (b *ReaderCacheBudget) complete(s *SingleChunkCacher) { } func (b *ReaderCacheBudget) release(s *SingleChunkCacher) { + if b == nil { + return + } b.Lock() defer b.Unlock() if size, found := b.reservations[s]; found { diff --git a/weed/filer/reader_cache_memory_test.go b/weed/filer/reader_cache_memory_test.go index e1a647163..a723ba3e2 100644 --- a/weed/filer/reader_cache_memory_test.go +++ b/weed/filer/reader_cache_memory_test.go @@ -201,6 +201,180 @@ func TestReaderCacheFailedPrefetchReleasesBudget(t *testing.T) { } } +func TestReaderCacheUnboundedWithoutBudget(t *testing.T) { + const readers = 96 // more than 256MiB / 4MiB = 64 buffers + started := make(chan struct{}, readers) + gate := make(chan struct{}) + rc := NewReaderCache(256, 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) { + started <- struct{}{} + <-gate + buffer[0] = 42 + return len(buffer), nil + } + var readersWg sync.WaitGroup + for i := 0; i < readers; i++ { + readersWg.Add(1) + go func(i int) { + defer readersWg.Done() + rc.ReadChunkAt(context.Background(), make([]byte, 1), fmt.Sprint(i), nil, false, 0, 4<<20, false) + }(i) + } + for i := 0; i < readers; i++ { + select { + case <-started: + case <-time.After(5 * time.Second): + close(gate) + readersWg.Wait() + t.Fatalf("only %d of %d downloads started; an implicit memory budget throttled the reader cache", i, readers) + } + } + close(gate) + readersWg.Wait() +} + +func TestReaderCacheDropsConsumedChunks(t *testing.T) { + var fetchCount int32 + rc := NewReaderCache(10, 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) { + atomic.AddInt32(&fetchCount, 1) + return len(buffer), nil + } + buf := make([]byte, 4<<10) + if _, err := rc.ReadChunkAt(context.Background(), buf, "chunk", nil, false, 0, 4<<10, false); err != nil { + t.Fatal(err) + } + rc.Lock() + _, retained := rc.downloaders["chunk"] + rc.Unlock() + if retained { + t.Fatal("fully consumed chunk buffer still retained") + } + if _, err := rc.ReadChunkAt(context.Background(), buf, "chunk", nil, false, 0, 4<<10, false); err != nil { + t.Fatal(err) + } + if got := atomic.LoadInt32(&fetchCount); got != 2 { + t.Fatalf("consumed chunk was not refetched, fetchCount=%d", got) + } +} + +// A prefetched chunk must survive until its reader arrives, then be dropped +// once consumed: the read hits the prefetched buffer (no second fetch) and a +// later read fetches again. +func TestReaderCachePrefetchBufferDroppedAfterRead(t *testing.T) { + var fetchCount int32 + rc := NewReaderCache(10, 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) { + atomic.AddInt32(&fetchCount, 1) + buffer[0] = 42 + return len(buffer), nil + } + rc.MaybeCache(&Interval[*ChunkView]{Value: &ChunkView{FileId: "chunk", ChunkSize: 4 << 10}}, 1) + + buf := make([]byte, 4<<10) + n, err := rc.ReadChunkAt(context.Background(), buf, "chunk", nil, false, 0, 4<<10, false) + if err != nil || n != len(buf) || buf[0] != 42 { + t.Fatalf("read of prefetched chunk: n=%d data=%d err=%v", n, buf[0], err) + } + if got := atomic.LoadInt32(&fetchCount); got != 1 { + t.Fatalf("read did not hit the prefetched buffer, fetchCount=%d", got) + } + rc.Lock() + _, retained := rc.downloaders["chunk"] + rc.Unlock() + if retained { + t.Fatal("consumed prefetch buffer still retained") + } + if _, err := rc.ReadChunkAt(context.Background(), buf, "chunk", nil, false, 0, 4<<10, false); err != nil { + t.Fatal(err) + } + if got := atomic.LoadInt32(&fetchCount); got != 2 { + t.Fatalf("consumed chunk was not refetched, fetchCount=%d", got) + } +} + +// A consumed chunk survives while another reader is still attached; it is +// dropped once no readers remain, regardless of which reader reached the end. +func TestReaderCacheConsumedBufferSurvivesAttachedReader(t *testing.T) { + var fetchCount int32 + rc := NewReaderCache(10, 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) { + atomic.AddInt32(&fetchCount, 1) + buffer[0] = 42 + return len(buffer), nil + } + + // A partial read primes the cacher, then a second reader attaches. + rc.ReadChunkAt(context.Background(), make([]byte, 1), "chunk", nil, false, 0, 4<<10, false) + rc.Lock() + downloader := rc.downloaders["chunk"] + if downloader == nil { + rc.Unlock() + t.Fatal("cacher missing before attach") + } + downloader.wg.Add(1) + atomic.AddInt32(&downloader.readers, 1) + rc.Unlock() + + buf := make([]byte, 4<<10) + if n, err := rc.ReadChunkAt(context.Background(), buf, "chunk", nil, false, 0, 4<<10, false); err != nil || n != len(buf) { + t.Fatalf("full read: n=%d err=%v", n, err) + } + rc.Lock() + _, retained := rc.downloaders["chunk"] + rc.Unlock() + if !retained { + t.Fatal("consumed chunk dropped while a reader was still attached") + } + + // The attached reader detaches without reading: since the buffer was + // already consumed, the last detach drops it. + downloader.wg.Done() + atomic.AddInt32(&downloader.readers, -1) + rc.removeConsumed(downloader) + rc.Lock() + _, retained = rc.downloaders["chunk"] + rc.Unlock() + if retained { + t.Fatal("consumed chunk retained after last reader detached") + } + if got := atomic.LoadInt32(&fetchCount); got != 1 { + t.Fatalf("attached readers fetched more than once, fetchCount=%d", got) + } +} + +func TestReaderCacheConsumedChunkReleasesBudget(t *testing.T) { + budget := NewReaderCacheBudget(4 << 10) + rc := NewReaderCache(10, newMockChunkCacheForReaderCache(), func(context.Context, string) ([]string, error) { + return []string{"unused"}, nil + }, nil, budget) + 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 + } + if _, err := rc.ReadChunkAt(context.Background(), make([]byte, 4<<10), "chunk", nil, false, 0, 4<<10, false); err != nil { + t.Fatal(err) + } + budget.Lock() + used := budget.used + budget.Unlock() + if used != 0 { + t.Fatalf("consumed chunk still reserves %d bytes", used) + } +} + // TestReaderCacheReReadAfterEviction verifies that a chunk evicted by budget // pressure is transparently re-downloaded on the next read and returns the // correct data. This is the core correctness property of eviction: a reader diff --git a/weed/filer/reader_cache_test.go b/weed/filer/reader_cache_test.go index 585f65bbb..672f3d4df 100644 --- a/weed/filer/reader_cache_test.go +++ b/weed/filer/reader_cache_test.go @@ -637,7 +637,7 @@ func TestReaderCacheDownloaderDedup(t *testing.T) { for i := 0; i < numReaders; i++ { go func() { defer wg.Done() - buffer := make([]byte, 100) + buffer := make([]byte, 50) rc.ReadChunkAt(context.Background(), buffer, "dedup-file", nil, false, 0, 100, false) }() } diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index 8e415eee6..6f060636b 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -6,6 +6,7 @@ import ( "encoding/json" "fmt" "io" + "math" "net" "net/http" "os" @@ -68,6 +69,7 @@ type S3ApiServerOption struct { ExternalUrl string // external URL clients use, tried first during signature verification behind a reverse proxy DefaultFileMode uint32 // default file permission mode for S3 uploads (e.g. 0660, 0644) CacheSizeMB int64 // in-memory chunk cache capacity in MB for the shared ReaderCache; 0 disables + ReaderCacheSizeMB int64 // memory budget in MiB for downloaded and in-flight reader buffers across all S3 GETs; 0 means unlimited MaxMB int32 // filer's -maxMB, read from the filer configuration at startup // AllowUntrustedRemoteEndpoints lets a read of a remote-only object dial a // mounted endpoint that resolves to a loopback / private / metadata host. @@ -282,7 +284,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl // assumed chunk size (s3ChunkCacheChunkSizeMB), clamped to a small // floor so tiny caches still function. // - // Downloader slots: each slot holds one in-flight / recently-completed + // Downloader slots: each slot holds one in-flight or not-yet-consumed // chunk buffer (~4 MiB by default), so this caps both peak memory for // in-flight chunks (s3ReaderCacheDownloaderLimit × chunkSize) and the // global fetch concurrency across all S3 GET requests. WebDAV uses 32 @@ -312,7 +314,14 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl } else { chunkCache = (*chunk_cache.TieredChunkCache)(nil) } - readerCache := filer.NewReaderCache(s3ReaderCacheDownloaderLimit, chunkCache, filerClient.GetLookupFileIdFunction(), filerClient) + if option.ReaderCacheSizeMB < 0 || option.ReaderCacheSizeMB > math.MaxInt64>>20 { + return nil, fmt.Errorf("invalid readerCacheSizeMB %d: must be non-negative and fit in an int64 byte budget", option.ReaderCacheSizeMB) + } + var readerCacheBudget *filer.ReaderCacheBudget + if option.ReaderCacheSizeMB > 0 { + readerCacheBudget = filer.NewReaderCacheBudget(option.ReaderCacheSizeMB << 20) + } + readerCache := filer.NewReaderCache(s3ReaderCacheDownloaderLimit, chunkCache, filerClient.GetLookupFileIdFunction(), filerClient, readerCacheBudget) s3ApiServer = &S3ApiServer{ option: option,