From 228ed25a01e46fdd30c107608b3e5133c351ab0e Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 14 Apr 2026 07:46:05 -0700 Subject: [PATCH] perf(s3): route GET through ChunkReadAt + per-request ReaderCache (#9068) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit perf(s3): route GET through ChunkReadAt + shared ReaderCache The S3 GET path previously used filer.PrepareStreamContentWithPrefetch, which hands chunk bytes from the volume-server fetch goroutine to the consumer through an io.Pipe. io.Pipe is a synchronous rendezvous, so the prefetch=4 window only overlapped HTTP connection setup — the actual data bytes still flowed one pipe at a time. Switch to the same path WebDAV uses (server/webdav_server.go): build a filer.ChunkReadAt backed by a server-wide filer.ReaderCache. ReaderCache prefetches whole chunks into []byte buffers, so the prefetch window translates into real in-flight bytes and the consumer copies them out as memcpys. The ReaderCache is server-wide (not per-request) for two reasons: 1. ChunkReadAt.Close() destroys the ReaderCache's downloader map. With a per-request cache, the defer on the handler would wait for background chunk downloads that run on context.Background() — so a client disconnect would block handler cleanup on downloads that the client no longer wants, tying up goroutines and memory. 2. Concurrent requests for the same object can share in-flight downloads through the shared downloader map. No persistent ChunkCache is added in this commit — the ReaderCache is constructed with a nil *chunk_cache.TieredChunkCache (all its methods are nil-receiver safe). A follow-up PR wires in an in-memory chunk cache for cross-request warm hits. JWT for volume-server requests is generated internally by util_http.RetriedFetchChunkData from jwtSigningReadKey, so the new path remains compatible with JWT-protected clusters — this is the same mechanism the WebDAV and mount read paths have been using. Measured on weed mini + 1 GiB random object over loopback, cold cache, single-stream curl on a presigned URL: before (io.Pipe): 2100-2200 MB/s after (ChunkReadAt): 2900-3800 MB/s --- weed/s3api/s3api_object_handlers.go | 67 +++++++++++++++-------------- weed/s3api/s3api_server.go | 24 +++++++++++ 2 files changed, 58 insertions(+), 33 deletions(-) diff --git a/weed/s3api/s3api_object_handlers.go b/weed/s3api/s3api_object_handlers.go index e0f05f16b..f01ff85ad 100644 --- a/weed/s3api/s3api_object_handlers.go +++ b/weed/s3api/s3api_object_handlers.go @@ -1026,14 +1026,15 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R } } - // CRITICAL: Resolve chunks and prepare stream BEFORE WriteHeader - // This ensures we can write proper error responses if these operations fail + // CRITICAL: Resolve chunks and prepare reader BEFORE WriteHeader so failures + // can still return a proper S3 error response. ctx := r.Context() lookupFileIdFn := s3a.createLookupFileIdFunction() - // Resolve chunk manifests with the requested range + // Resolve chunk manifests into visible intervals for the requested range. + // NonOverlappingVisibleIntervals internally calls ResolveChunkManifest. tChunkResolve := time.Now() - resolvedChunks, _, err := filer.ResolveChunkManifest(ctx, lookupFileIdFn, chunks, offset, offset+size) + visibleIntervals, err := filer.NonOverlappingVisibleIntervals(ctx, lookupFileIdFn, chunks, offset, offset+size) chunkResolveTime = time.Since(tChunkResolve) if err != nil { if isCanceledStreamingError(err) { @@ -1050,34 +1051,23 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R return newStreamErrorWithResponse(fmt.Errorf("failed to resolve chunks: %v", err)) } - // Prepare streaming function with simple master client wrapper + // Build a ChunkReadAt backed by the server-wide ReaderCache. This mirrors + // the WebDAV read path (server/webdav_server.go) and outperforms the + // io.Pipe-based streamChunksPrefetched path: ReaderCache prefetches whole + // chunks into memory buffers that the consumer can memcpy out of, so + // prefetchCount translates into actual in-flight bytes rather than just + // parallel TCP handshakes. + // + // We do NOT call reader.Close() here — ChunkReadAt.Close() would destroy + // the ReaderCache's downloaders map, which is shared across concurrent + // requests. Eviction is handled by the ReaderCache's own downloader + // limit. JWT for volume-server requests is generated internally by + // util_http.RetriedFetchChunkData from jwtSigningReadKey, matching the + // WebDAV and mount read paths. tStreamPrep := time.Now() - // Use filerClient directly (not wrapped) so it can support cache invalidation - streamFn, err := filer.PrepareStreamContentWithPrefetch( - ctx, - s3a.filerClient, - filer.JwtForVolumeServer, // Use filer's JWT function (loads config once, generates JWT locally) - resolvedChunks, - offset, - size, - 0, // no throttling - 4, // prefetch 4 chunks ahead for overlapped fetching - ) + chunkViews := filer.ViewFromVisibleIntervals(visibleIntervals, offset, size) + reader := filer.NewChunkReaderAtFromClient(ctx, s3a.readerCache, chunkViews, totalSize, filer.DefaultPrefetchCount) streamPrepTime = time.Since(tStreamPrep) - if err != nil { - if isCanceledStreamingError(err) { - glog.V(3).Infof("streamFromVolumeServers: request canceled while preparing stream: %v", err) - return err - } - if errors.Is(err, context.DeadlineExceeded) { - glog.Warningf("streamFromVolumeServers: request deadline exceeded while preparing stream: %v", err) - } else { - glog.Errorf("streamFromVolumeServers: failed to prepare stream: %v", err) - } - // Write S3-compliant XML error response - s3err.WriteErrorResponse(w, r, s3err.ErrInternalError) - return newStreamErrorWithResponse(fmt.Errorf("failed to prepare stream: %v", err)) - } // All validation and preparation successful - NOW set headers and write status tHeaderSet := time.Now() @@ -1102,11 +1092,22 @@ func (s3a *S3ApiServer) streamFromVolumeServers(w http.ResponseWriter, r *http.R // Track time to first byte metric TimeToFirstByte(r.Method, t0, r) - // Stream directly to response with counting wrapper + // Stream directly to response with counting wrapper. + // ChunkReadAt's ReadAt is backed by in-memory prefetched chunk buffers, so + // io.CopyBuffer drains them as fast memcpys. tStreamExec := time.Now() - glog.V(4).Infof("streamFromVolumeServers: starting streamFn, offset=%d, size=%d", offset, size) + glog.V(4).Infof("streamFromVolumeServers: starting chunk reader, offset=%d, size=%d", offset, size) cw := &countingWriter{w: w} - err = streamFn(cw) + // Cap the copy buffer to the response size so small-object GETs (common + // for thumbnails, config files, etc.) don't allocate a 256 KiB scratch + // buffer per request. + const maxCopyBuf = 256 * 1024 + copyBufSize := int64(maxCopyBuf) + if size > 0 && size < copyBufSize { + copyBufSize = size + } + copyBuf := make([]byte, copyBufSize) + _, err = io.CopyBuffer(cw, io.NewSectionReader(reader, offset, size), copyBuf) streamExecTime = time.Since(tStreamExec) // Track traffic even on partial writes for accurate egress accounting if cw.written > 0 { diff --git a/weed/s3api/s3api_server.go b/weed/s3api/s3api_server.go index c48aa3072..f3a606b92 100644 --- a/weed/s3api/s3api_server.go +++ b/weed/s3api/s3api_server.go @@ -31,6 +31,7 @@ import ( "github.com/seaweedfs/seaweedfs/weed/s3api/s3err" "github.com/seaweedfs/seaweedfs/weed/security" "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/chunk_cache" "github.com/seaweedfs/seaweedfs/weed/util/grace" util_http "github.com/seaweedfs/seaweedfs/weed/util/http" util_http_client "github.com/seaweedfs/seaweedfs/weed/util/http/client" @@ -84,6 +85,12 @@ type S3ApiServer struct { stsHandlers *STSHandlers // STS HTTP handlers for AssumeRoleWithWebIdentity cipher bool // encrypt data on volume servers newObjectWriteLock func(bucket, object string) objectWriteLock + // Shared ReaderCache used by the S3 GET streaming path. It lives for the + // lifetime of the server so that concurrent and repeat reads share a + // single in-flight download per chunk, and so that no per-request + // teardown waits on context.Background() fetches. The chunkCache field + // is nil in this commit; a follow-up wires in an in-memory chunk cache. + readerCache *filer.ReaderCache } type objectWriteLock interface { @@ -178,6 +185,22 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl } }() + // Shared ReaderCache for the S3 GET streaming path. The chunk cache is + // nil for now — all TieredChunkCache receiver methods are nil-safe. A + // follow-up adds an in-memory chunk cache on top. Keeping this shared + // (rather than per-request) avoids the per-request Close(), which would + // otherwise wait for background chunk downloads that run on + // context.Background() even after the client disconnects. + // + // Downloader slots: each slot holds one in-flight / recently-completed + // 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 + // because it typically has a handful of clients; S3 serves many + // concurrent readers, so we pick a more generous default here. + const s3ReaderCacheDownloaderLimit = 256 + readerCache := filer.NewReaderCache(s3ReaderCacheDownloaderLimit, (*chunk_cache.TieredChunkCache)(nil), filerClient.GetLookupFileIdFunction()) + s3ApiServer = &S3ApiServer{ option: option, iam: iam, @@ -190,6 +213,7 @@ func NewS3ApiServerWithStore(router *mux.Router, option *S3ApiServerOption, expl policyEngine: policyEngine, // Initialize bucket policy engine inFlightDataLimitCond: sync.NewCond(new(sync.Mutex)), cipher: option.Cipher, + readerCache: readerCache, } if len(option.Filers) > 0 {