From 361fd6b263d86d7d7bae0f53210eb4af49bf0b60 Mon Sep 17 00:00:00 2001 From: ssshr-66 Date: Tue, 8 Sep 2026 09:44:47 +0800 Subject: [PATCH] [Filer] Parallelize Chunk Manifest Resolution to Reduce Large File Read Latency (#11215) * fix issue-11214 * fix(filer): cancel sibling manifest reads on failure * fix(filer): scope manifest cancellation to read batch and propagate context to encrypted reads Address PR review comments on #11215: - Scope cancellation to each parallel read batch instead of the resolver-wide context, so a later manifest failure does not cancel recursive work for an earlier successful manifest (CodeRabbit #3952446554). - Propagate the resolver context through GetAuthenticatedWithContext so encrypted sibling reads observe cancellation and stop promptly when another manifest fails (Greptile #3952422343). - Use net.ListenConfig.Listen with an explicit context in the test fixture to satisfy the noctx linter (CodeRabbit #3952111696). - Add regression tests for encrypted sibling cancellation and for preserving earlier manifest children on later failure. * fix(filer): return known manifest errors without blocking on earlier children Address Greptile review comment on #11215: After all parallel reads complete, pre-scan slots for the first real (non-internal-cancel) error before recursing into earlier manifests' children. If a later manifest already failed, return its error promptly with data chunks already in hand, instead of blocking on recursive network reads of earlier manifests' children. Updated the regression test to verify the error returns within 1 second when an earlier manifest's child has a 2-second delay, and that the child is never loaded. * fix(filer): filter partial child chunks by requested range on error path Address CodeRabbit review comment on #11215: The pre-scan error path appended non-manifest child chunks from earlier manifests without applying the [startOffset, stopOffset) overlap check used for top-level chunks. A child outside the requested range could be returned in dataChunks alongside the later manifest's error. Apply the same range predicate before appending. Add regression test with an out-of-range child chunk. * fix(filer): buffer job channel and abort submission on batch cancellation Address Greptile review comment on #11215: - Use a buffered job channel (capacity 128) so submission does not block when all workers are busy. This ensures a promptly-failing manifest is always queued and can cancel stalled sibling reads once a worker picks it up, instead of blocking the caller on the unbuffered channel send. - Add batchCtx.Done() to the submit select so submission aborts promptly when the batch is already cancelled by a sibling failure. - Add regression test with 5 manifests (4 stalled + 1 failing) verifying the failing job is queued and picked up after a stalled worker is freed. * fix(filer): avoid double WaitGroup decrement on batch cancellation in submit Address Devin review comment on #11215: When batchCtx.Done() fired in submit, it called job.done.Done() and returned false. The caller in resolve also called reads.Done() on the same WaitGroup, causing a double decrement that would panic with a negative counter. Fix: submit sets the result error but does not decrement the WaitGroup. The caller always owns the decrement and skips overwriting the result when submit already set it. * fix(filer): overflow execution when job queue buffer is full Address Greptile follow-up review comment on #11215: With a 128-entry buffer, if more than 132 in-range manifests (4 workers + 128 buffer) stall at one level, a promptly-failing manifest beyond the buffer cannot be submitted and cannot cancel the stalled reads. Fix: when the buffer is full, run the job directly in a goroutine instead of blocking on the channel send. This only triggers for >132 manifests at one level (exceedingly rare), so the bounded concurrency guarantee (4 workers) holds for all normal workloads. Extracted executeJob method shared by both workers and overflow goroutines. * fix(filer): bound overflow execution with a semaphore Address Devin review comment on #11215: The unbounded overflow goroutines could create thousands of concurrent reads for large files, defeating the four-worker resource bound. Fix: add a semaphore (capacity = maxChunkManifestResolveWorkers) that overflow goroutines must acquire before doing the read. While waiting for the semaphore, they also watch batchCtx and r.ctx so they exit promptly on cancellation. Total concurrency is now bounded to 2 * workers (4 workers + 4 overflow) in the degenerate case. * refactor(http): add ctx to GetAuthenticated signature instead of new function Reuse the existing GetAuthenticated name by adding ctx as the first parameter, matching the pattern of ReadUrl, ReadUrlAsStream, and RetriedFetchChunkData. Removes the GetAuthenticatedWithContext wrapper. --------- Co-authored-by: Chris Lu --- weed/filer/filechunk_manifest.go | 230 ++++++- weed/filer/filechunk_manifest_resolve_test.go | 595 ++++++++++++++++++ weed/util/http/http_global_client_util.go | 15 +- 3 files changed, 826 insertions(+), 14 deletions(-) create mode 100644 weed/filer/filechunk_manifest_resolve_test.go diff --git a/weed/filer/filechunk_manifest.go b/weed/filer/filechunk_manifest.go index 35fdc5063..597861a87 100644 --- a/weed/filer/filechunk_manifest.go +++ b/weed/filer/filechunk_manifest.go @@ -3,6 +3,7 @@ package filer import ( "bytes" "context" + "errors" "fmt" "io" "math" @@ -25,6 +26,14 @@ var bytesBufferPool = sync.Pool{ }, } +// Keep Manifest reads bounded across all recursion levels of one resolution. +const maxChunkManifestResolveWorkers = 4 + +// Size of the job queue buffer. Large enough that submission does not block +// under normal chunk counts, so a promptly-failing manifest is always queued +// and can cancel stalled sibling reads once a worker picks it up. +const chunkManifestResolveJobBufferSize = 128 + func HasChunkManifest(chunks []*filer_pb.FileChunk) bool { for _, chunk := range chunks { if chunk.IsChunkManifest { @@ -46,26 +55,229 @@ func SeparateManifestChunks(chunks []*filer_pb.FileChunk) (manifestChunks, nonMa } func ResolveChunkManifest(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, chunks []*filer_pb.FileChunk, startOffset, stopOffset int64, invalidator CacheInvalidator) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) { - // TODO maybe parallel this - for _, chunk := range chunks { + resolver := newChunkManifestResolver(ctx, lookupFileIdFn, invalidator) + defer resolver.close() + return resolver.resolve(chunks, startOffset, stopOffset) +} +type chunkManifestResolveJob struct { + chunk *filer_pb.FileChunk + result *chunkManifestResolveResult + done *sync.WaitGroup + batchCtx context.Context + batchCancel context.CancelFunc + batchOnce *sync.Once +} + +type chunkManifestResolveResult struct { + chunks []*filer_pb.FileChunk + err error + internalCancel bool +} + +type chunkManifestResolver struct { + ctx context.Context + parentCtx context.Context + cancel context.CancelFunc + lookupFileIdFn wdclient.LookupFileIdFunctionType + invalidator CacheInvalidator + jobs chan chunkManifestResolveJob + overflowSem chan struct{} + workers sync.WaitGroup + startOnce sync.Once + started bool +} + +func newChunkManifestResolver(ctx context.Context, lookupFileIdFn wdclient.LookupFileIdFunctionType, invalidator CacheInvalidator) *chunkManifestResolver { + workCtx, cancel := context.WithCancel(ctx) + resolver := &chunkManifestResolver{ + ctx: workCtx, + parentCtx: ctx, + cancel: cancel, + lookupFileIdFn: lookupFileIdFn, + invalidator: invalidator, + jobs: make(chan chunkManifestResolveJob, chunkManifestResolveJobBufferSize), + overflowSem: make(chan struct{}, maxChunkManifestResolveWorkers), + } + return resolver +} + +func (r *chunkManifestResolver) executeJob(job chunkManifestResolveJob) { + job.result.chunks, job.result.err = ResolveOneChunkManifest(job.batchCtx, r.lookupFileIdFn, job.chunk, r.invalidator) + if job.result.err != nil && r.parentCtx.Err() == nil { + if job.batchCtx.Err() != nil && errors.Is(job.result.err, context.Canceled) { + job.result.internalCancel = true + } else if job.batchCtx.Err() == nil { + job.batchOnce.Do(job.batchCancel) + } + } + job.done.Done() +} + +func (r *chunkManifestResolver) executeOverflowJob(job chunkManifestResolveJob) { + select { + case r.overflowSem <- struct{}{}: + defer func() { <-r.overflowSem }() + r.executeJob(job) + case <-job.batchCtx.Done(): + if job.result.err == nil { + job.result.err = job.batchCtx.Err() + job.result.internalCancel = r.parentCtx.Err() == nil && job.result.err != nil + } + job.done.Done() + case <-r.ctx.Done(): + if job.result.err == nil { + job.result.err = r.ctx.Err() + job.result.internalCancel = r.parentCtx.Err() == nil && job.result.err != nil + } + job.done.Done() + } +} + +func (r *chunkManifestResolver) worker() { + defer r.workers.Done() + for job := range r.jobs { + r.executeJob(job) + } +} + +func (r *chunkManifestResolver) close() { + r.cancel() + close(r.jobs) + if r.started { + r.workers.Wait() + } +} + +func (r *chunkManifestResolver) submit(job chunkManifestResolveJob) bool { + r.startOnce.Do(func() { + r.workers.Add(maxChunkManifestResolveWorkers) + for i := 0; i < maxChunkManifestResolveWorkers; i++ { + go r.worker() + } + r.started = true + }) + select { + case r.jobs <- job: + return true + case <-r.ctx.Done(): + return false + case <-job.batchCtx.Done(): + // Batch was cancelled by a sibling failure; don't queue this job. + // Set the result so the caller doesn't overwrite it; the caller + // owns the WaitGroup decrement. + job.result.err = job.batchCtx.Err() + job.result.internalCancel = r.parentCtx.Err() == nil && job.result.err != nil + return false + default: + // Buffer is full (exceedingly rare: more than chunkManifestResolveJobBufferSize + // in-range manifests at one level). Run the job in a bounded overflow goroutine + // so a promptly-failing manifest can still cancel the batch without waiting for + // a worker slot, while keeping total concurrency bounded. + go r.executeOverflowJob(job) + return true + } +} + +func (r *chunkManifestResolver) resolve(chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, manifestResolveErr error) { + type resolveSlot struct { + chunk *filer_pb.FileChunk + result chunkManifestResolveResult + } + + // Cancellation is scoped to this parallel read batch so a failure in a + // later manifest does not cancel recursive work for an earlier manifest + // that already completed. Recursion creates its own batch context derived + // from the still-alive resolver context. + batchCtx, batchCancel := context.WithCancel(r.ctx) + defer batchCancel() + var batchOnce sync.Once + + slots := make([]resolveSlot, len(chunks)) + var reads sync.WaitGroup + for i, chunk := range chunks { if max(chunk.Offset, startOffset) >= min(chunk.Offset+int64(chunk.Size), stopOffset) { continue } + slots[i].chunk = chunk if !chunk.IsChunkManifest { - dataChunks = append(dataChunks, chunk) continue } - resolvedChunks, err := ResolveOneChunkManifest(ctx, lookupFileIdFn, chunk, invalidator) - if err != nil { - return dataChunks, nil, err + reads.Add(1) + if !r.submit(chunkManifestResolveJob{ + chunk: chunk, + result: &slots[i].result, + done: &reads, + batchCtx: batchCtx, + batchCancel: batchCancel, + batchOnce: &batchOnce, + }) { + // submit may have already set the result (batch cancellation). + // Only set it here for the resolver-cancellation case. + if slots[i].result.err == nil { + slots[i].result.err = r.ctx.Err() + slots[i].result.internalCancel = r.parentCtx.Err() == nil && slots[i].result.err != nil + } + reads.Done() + } + } + reads.Wait() + if err := r.parentCtx.Err(); err != nil { + return dataChunks, nil, err + } + + // Recurse only after this level's reads release their worker slots. A + // worker must never wait for a child manifest while holding a slot. + // + // Pre-scan for the first real (non-internal-cancel) error before + // recursing. If a later manifest already failed, return its error + // promptly with data chunks that are already in hand, instead of + // blocking on recursive reads of earlier manifests' children. + for i, slot := range slots { + if slot.chunk == nil { + continue + } + if slot.result.err != nil { + if slot.result.internalCancel { + continue + } + for j := 0; j < i; j++ { + if slots[j].chunk == nil { + continue + } + if !slots[j].chunk.IsChunkManifest { + dataChunks = append(dataChunks, slots[j].chunk) + continue + } + for _, c := range slots[j].result.chunks { + if !c.IsChunkManifest && max(c.Offset, startOffset) < min(c.Offset+int64(c.Size), stopOffset) { + dataChunks = append(dataChunks, c) + } + } + } + return dataChunks, nil, slot.result.err + } + } + + for _, slot := range slots { + if slot.chunk == nil { + continue + } + if slot.result.err != nil { + if slot.result.internalCancel { + continue + } + return dataChunks, nil, slot.result.err + } + if !slot.chunk.IsChunkManifest { + dataChunks = append(dataChunks, slot.chunk) + continue } - manifestChunks = append(manifestChunks, chunk) - // recursive - subDataChunks, subManifestChunks, subErr := ResolveChunkManifest(ctx, lookupFileIdFn, resolvedChunks, startOffset, stopOffset, invalidator) + manifestChunks = append(manifestChunks, slot.chunk) + subDataChunks, subManifestChunks, subErr := r.resolve(slot.result.chunks, startOffset, stopOffset) if subErr != nil { return dataChunks, nil, subErr } diff --git a/weed/filer/filechunk_manifest_resolve_test.go b/weed/filer/filechunk_manifest_resolve_test.go new file mode 100644 index 000000000..0daf1f274 --- /dev/null +++ b/weed/filer/filechunk_manifest_resolve_test.go @@ -0,0 +1,595 @@ +package filer + +import ( + "context" + "errors" + "fmt" + "net" + "net/http" + "net/http/httptest" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" + "github.com/stretchr/testify/require" + "google.golang.org/protobuf/proto" +) + +type manifestReadFixture struct { + server *httptest.Server + mu sync.Mutex + started []string + completed []string + active atomic.Int32 + maxActive atomic.Int32 + loads atomic.Int32 + manifests map[string][]byte + delays map[string]time.Duration + stopReads chan struct{} +} + +func newManifestReadFixture(t testing.TB, manifests map[string][]*filer_pb.FileChunk, delays map[string]time.Duration) *manifestReadFixture { + t.Helper() + encoded := make(map[string][]byte, len(manifests)) + for id, chunks := range manifests { + data, err := proto.Marshal(&filer_pb.FileChunkManifest{Chunks: chunks}) + require.NoError(t, err) + encoded[id] = data + } + + fixture := &manifestReadFixture{manifests: encoded, delays: delays, stopReads: make(chan struct{})} + listener, err := (&net.ListenConfig{}).Listen(context.Background(), "tcp", "127.0.0.1:0") + require.NoError(t, err) + fixture.server = httptest.NewUnstartedServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + id := strings.TrimPrefix(r.URL.Path, "/") + fixture.loads.Add(1) + fixture.active.Add(1) + fixture.updateMaxActive() + fixture.mu.Lock() + fixture.started = append(fixture.started, id) + fixture.mu.Unlock() + finishRead := func() { + fixture.active.Add(-1) + fixture.mu.Lock() + fixture.completed = append(fixture.completed, id) + fixture.mu.Unlock() + } + + if delay := fixture.delays[id]; delay > 0 { + timer := time.NewTimer(delay) + select { + case <-r.Context().Done(): + timer.Stop() + finishRead() + return + case <-fixture.stopReads: + timer.Stop() + finishRead() + return + case <-timer.C: + } + } + data, ok := fixture.manifests[id] + if !ok { + http.NotFound(w, r) + finishRead() + return + } + w.Header().Set("Content-Length", fmt.Sprintf("%d", len(data))) + w.WriteHeader(http.StatusOK) + _, _ = w.Write(data) + finishRead() + })) + fixture.server.Listener = listener + fixture.server.Start() + t.Cleanup(fixture.server.Close) + return fixture +} + +func (f *manifestReadFixture) updateMaxActive() { + for { + current := f.active.Load() + maximum := f.maxActive.Load() + if current <= maximum || f.maxActive.CompareAndSwap(maximum, current) { + return + } + } +} + +func (f *manifestReadFixture) lookup(ctx context.Context, fileID string) ([]string, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return []string{f.server.URL + "/" + fileID}, nil +} + +func (f *manifestReadFixture) completionOrder() []string { + f.mu.Lock() + defer f.mu.Unlock() + return append([]string(nil), f.completed...) +} + +func resolveTestManifest(id string, offset int64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{FileId: id, IsChunkManifest: true, Offset: offset, Size: 100} +} + +func resolveTestData(id string, offset int64) *filer_pb.FileChunk { + return &filer_pb.FileChunk{FileId: id, Offset: offset, Size: 10} +} + +func TestResolveChunkManifestParallelReadsAreBounded(t *testing.T) { + manifests := make(map[string][]*filer_pb.FileChunk) + inputs := make([]*filer_pb.FileChunk, 0, 8) + delays := make(map[string]time.Duration) + for i := 0; i < 8; i++ { + id := fmt.Sprintf("m%d", i) + manifests[id] = []*filer_pb.FileChunk{resolveTestData(fmt.Sprintf("d%d", i), int64(i*10))} + inputs = append(inputs, resolveTestManifest(id, int64(i*100))) + delays[id] = 50 * time.Millisecond + } + fixture := newManifestReadFixture(t, manifests, delays) + + data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, inputs, 0, 1000, nil) + require.NoError(t, err) + require.Len(t, data, len(inputs)) + require.Len(t, meta, len(inputs)) + require.GreaterOrEqual(t, fixture.maxActive.Load(), int32(2), "independent manifests should overlap") + require.LessOrEqual(t, fixture.maxActive.Load(), int32(4), "manifest reads must be bounded") +} + +func TestResolveChunkManifestPreservesInputOrderWhenReadsCompleteOutOfOrder(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "slow": {resolveTestData("slow-data", 0)}, + "fast": {resolveTestData("fast-data", 100)}, + }, + map[string]time.Duration{"slow": 80 * time.Millisecond, "fast": 5 * time.Millisecond}, + ) + + data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{ + resolveTestManifest("slow", 0), + resolveTestManifest("fast", 100), + }, 0, 200, nil) + require.NoError(t, err) + require.Equal(t, []string{"slow-data", "fast-data"}, fileIDs(data)) + require.Equal(t, []string{"slow", "fast"}, fileIDs(meta)) + require.Equal(t, []string{"fast", "slow"}, fixture.completionOrder()) +} + +func TestResolveChunkManifestNestedManifestsKeepDepthFirstOrder(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "parent": {resolveTestData("parent-data", 0), resolveTestManifest("nested", 10)}, + "sibling": {resolveTestData("sibling-data", 20)}, + "nested": {resolveTestData("nested-data", 10)}, + }, + map[string]time.Duration{"parent": 60 * time.Millisecond, "sibling": 5 * time.Millisecond}, + ) + + data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{ + resolveTestManifest("parent", 0), + resolveTestManifest("sibling", 20), + }, 0, 100, nil) + require.NoError(t, err) + require.Equal(t, []string{"parent-data", "nested-data", "sibling-data"}, fileIDs(data)) + require.Equal(t, []string{"parent", "nested", "sibling"}, fileIDs(meta)) + require.Equal(t, []string{"sibling", "parent", "nested"}, fixture.completionOrder()) +} + +func TestResolveChunkManifestFiltersBeforeReadingAndRecursively(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "outside": {resolveTestData("never-read", 0)}, + "inside": {resolveTestData("before", 0), resolveTestData("selected", 100), resolveTestData("after", 200)}, + }, nil) + + data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{ + resolveTestManifest("outside", 0), + resolveTestManifest("inside", 90), + }, 100, 200, nil) + require.NoError(t, err) + require.Equal(t, []string{"selected"}, fileIDs(data)) + require.Equal(t, []string{"inside"}, fileIDs(meta)) + require.Equal(t, int32(1), fixture.loads.Load(), "the non-overlapping parent manifest must not be read") +} + +func TestResolveChunkManifestPropagatesReadAndFormatErrorsInInputOrder(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "valid": {resolveTestData("valid-data", 0)}, + "broken": nil, + }, map[string]time.Duration{"broken": 50 * time.Millisecond}) + fixture.manifests["broken"] = []byte("not a protobuf manifest") + + data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{ + resolveTestData("plain", 0), + resolveTestManifest("valid", 10), + resolveTestManifest("broken", 20), + }, 0, 100, nil) + require.Error(t, err) + require.Nil(t, meta) + require.Equal(t, []string{"plain", "valid-data"}, fileIDs(data)) + require.Contains(t, err.Error(), "fail to unmarshal manifest broken") + + lookupErr := errors.New("lookup failed") + lookup := func(context.Context, string) ([]string, error) { return nil, lookupErr } + _, _, err = ResolveChunkManifest(context.Background(), lookup, []*filer_pb.FileChunk{resolveTestManifest("lookup-error", 0)}, 0, 100, nil) + require.ErrorIs(t, err, lookupErr) +} + +func TestResolveChunkManifestCancellationStopsAllReads(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "one": {resolveTestData("one-data", 0)}, + "two": {resolveTestData("two-data", 10)}, + }, + map[string]time.Duration{"one": time.Second, "two": time.Second}, + ) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan error, 1) + go func() { + _, _, err := ResolveChunkManifest(ctx, fixture.lookup, []*filer_pb.FileChunk{ + resolveTestManifest("one", 0), + resolveTestManifest("two", 10), + }, 0, 100, nil) + result <- err + }() + + deadline := time.Now().Add(time.Second) + for fixture.loads.Load() < 1 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + close(fixture.stopReads) + cancel() + err := <-result + require.ErrorIs(t, err, context.Canceled) + require.Eventually(t, func() bool { + return fixture.active.Load() == 0 + }, time.Second, time.Millisecond, "all manifest readers must exit after ResolveChunkManifest returns") +} + +type manifestFailureReadFixture struct { + server *httptest.Server + fastRelease chan struct{} + slowStarted chan struct{} + fastStarted chan struct{} + slowCanceled chan struct{} +} + +func newManifestFailureReadFixture(t testing.TB) *manifestFailureReadFixture { + t.Helper() + fixture := &manifestFailureReadFixture{ + fastRelease: make(chan struct{}), + slowStarted: make(chan struct{}), + fastStarted: make(chan struct{}), + slowCanceled: make(chan struct{}), + } + fixture.server = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + switch strings.TrimPrefix(r.URL.Path, "/") { + case "fast": + close(fixture.fastStarted) + <-fixture.fastRelease + w.Header().Set("Content-Length", "7") + w.WriteHeader(http.StatusOK) + _, _ = w.Write([]byte("invalid")) + case "slow": + close(fixture.slowStarted) + <-r.Context().Done() + close(fixture.slowCanceled) + } + })) + t.Cleanup(fixture.server.Close) + return fixture +} + +func (f *manifestFailureReadFixture) lookup(ctx context.Context, fileID string) ([]string, error) { + if err := ctx.Err(); err != nil { + return nil, err + } + return []string{f.server.URL + "/" + fileID}, nil +} + +func waitForManifestFailureSignal(t *testing.T, signal <-chan struct{}) { + t.Helper() + select { + case <-signal: + case <-time.After(time.Second): + t.Fatal("timed out waiting for manifest request") + } +} + +func TestResolveChunkManifestFastFailureCancelsSlowSibling(t *testing.T) { + testCases := []struct { + name string + inputIDs []string + expectedID string + }{ + {name: "fast then slow returns quickly", inputIDs: []string{"fast", "slow"}, expectedID: "fast"}, + {name: "slow then fast keeps real error", inputIDs: []string{"slow", "fast"}, expectedID: "fast"}, + } + + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + fixture := newManifestFailureReadFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + chunks := make([]*filer_pb.FileChunk, 0, len(testCase.inputIDs)) + for i, id := range testCase.inputIDs { + chunks = append(chunks, resolveTestManifest(id, int64(i*100))) + } + result := make(chan error, 1) + go func() { + _, _, err := ResolveChunkManifest(ctx, fixture.lookup, chunks, 0, 200, nil) + result <- err + }() + + waitForManifestFailureSignal(t, fixture.fastStarted) + waitForManifestFailureSignal(t, fixture.slowStarted) + close(fixture.fastRelease) + + var err error + select { + case err = <-result: + case <-time.After(time.Second): + cancel() + select { + case <-result: + case <-time.After(time.Second): + t.Fatal("ResolveChunkManifest did not finish after caller cancellation") + } + t.Fatal("fast manifest failure waited for the slow sibling") + } + require.Error(t, err) + require.Contains(t, err.Error(), "fail to unmarshal manifest "+testCase.expectedID) + require.NotErrorIs(t, err, context.Canceled) + waitForManifestFailureSignal(t, fixture.slowCanceled) + }) + } +} + +func TestResolveChunkManifestFastFailureCancelsEncryptedSibling(t *testing.T) { + fixture := newManifestFailureReadFixture(t) + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + + slowChunk := resolveTestManifest("slow", 0) + slowChunk.CipherKey = []byte("0123456789abcdef") + chunks := []*filer_pb.FileChunk{ + slowChunk, + resolveTestManifest("fast", 100), + } + result := make(chan error, 1) + go func() { + _, _, err := ResolveChunkManifest(ctx, fixture.lookup, chunks, 0, 200, nil) + result <- err + }() + + waitForManifestFailureSignal(t, fixture.fastStarted) + waitForManifestFailureSignal(t, fixture.slowStarted) + close(fixture.fastRelease) + + var err error + select { + case err = <-result: + case <-time.After(time.Second): + cancel() + select { + case <-result: + case <-time.After(time.Second): + t.Fatal("ResolveChunkManifest did not finish after caller cancellation") + } + t.Fatal("fast manifest failure waited for the encrypted slow sibling") + } + require.Error(t, err) + require.Contains(t, err.Error(), "fail to unmarshal manifest fast") + require.NotErrorIs(t, err, context.Canceled) + waitForManifestFailureSignal(t, fixture.slowCanceled) +} + +func TestResolveChunkManifestKeepsRealErrorAfterInternalCancellation(t *testing.T) { + earlyStarted := make(chan struct{}) + lateStarted := make(chan struct{}) + earlyErr := errors.New("early lookup failed") + lateErr := errors.New("late lookup failed") + lookup := func(ctx context.Context, fileID string) ([]string, error) { + switch fileID { + case "early": + close(earlyStarted) + <-ctx.Done() + return nil, earlyErr + case "late": + close(lateStarted) + return nil, lateErr + default: + return nil, fmt.Errorf("unexpected manifest %s", fileID) + } + } + + result := make(chan error, 1) + go func() { + _, _, err := ResolveChunkManifest(context.Background(), lookup, []*filer_pb.FileChunk{ + resolveTestManifest("early", 0), + resolveTestManifest("late", 100), + }, 0, 200, nil) + result <- err + }() + waitForManifestFailureSignal(t, earlyStarted) + waitForManifestFailureSignal(t, lateStarted) + + select { + case err := <-result: + require.ErrorIs(t, err, earlyErr) + require.NotErrorIs(t, err, lateErr) + case <-time.After(time.Second): + t.Fatal("ResolveChunkManifest did not return the input-order error") + } +} + +func TestResolveChunkManifestReturnsLaterFailureWithoutRecursingEarlierChildren(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "a": {resolveTestManifest("a-child", 0)}, + "a-child": {resolveTestData("a-child-data", 0)}, + "b": nil, + }, + map[string]time.Duration{"a": 5 * time.Millisecond, "b": 50 * time.Millisecond, "a-child": 2 * time.Second}, + ) + fixture.manifests["b"] = []byte("not a protobuf manifest") + + start := time.Now() + data, meta, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{ + resolveTestManifest("a", 0), + resolveTestManifest("b", 100), + }, 0, 200, nil) + elapsed := time.Since(start) + + require.Error(t, err) + require.Contains(t, err.Error(), "fail to unmarshal manifest b") + require.Empty(t, data, "earlier manifest's children must not be recursed when a later manifest already failed") + require.Nil(t, meta) + require.Equal(t, int32(2), fixture.loads.Load(), "only the two top-level manifests must be read, not a-child") + require.Less(t, elapsed, time.Second, "must return promptly without waiting for a-child's slow read") +} + +func TestResolveChunkManifestExcludesOutOfRangeChildrenOnLaterFailure(t *testing.T) { + fixture := newManifestReadFixture(t, + map[string][]*filer_pb.FileChunk{ + "a": { + resolveTestData("a-in-range", 0), + resolveTestData("a-out-of-range", 500), + }, + "b": nil, + }, + map[string]time.Duration{"b": 50 * time.Millisecond}, + ) + fixture.manifests["b"] = []byte("not a protobuf manifest") + + data, _, err := ResolveChunkManifest(context.Background(), fixture.lookup, []*filer_pb.FileChunk{ + resolveTestManifest("a", 0), + resolveTestManifest("b", 100), + }, 0, 200, nil) + require.Error(t, err) + require.Contains(t, err.Error(), "fail to unmarshal manifest b") + require.Equal(t, []string{"a-in-range"}, fileIDs(data), "out-of-range child chunks must be excluded from partial results") +} + +func TestResolveChunkManifestQueuesJobsBeyondWorkerCount(t *testing.T) { + // More manifests than workers. The first four stall; the fifth fails + // promptly. With a buffered job channel, the fifth job is queued without + // blocking submission. When a stalled read is released, a worker picks up + // the failing job, which cancels the batch and unblocks the rest. + stallRelease := make(chan struct{}) + stallStarted := make(chan struct{}, 4) + lookup := func(ctx context.Context, fileID string) ([]string, error) { + switch fileID { + case "stall-0", "stall-1", "stall-2", "stall-3": + stallStarted <- struct{}{} + select { + case <-stallRelease: + return nil, nil + case <-ctx.Done(): + return nil, ctx.Err() + } + case "fail": + return nil, errors.New("prompt lookup failure") + default: + return nil, fmt.Errorf("unexpected manifest %s", fileID) + } + } + + result := make(chan error, 1) + go func() { + _, _, err := ResolveChunkManifest(context.Background(), lookup, []*filer_pb.FileChunk{ + resolveTestManifest("stall-0", 0), + resolveTestManifest("stall-1", 100), + resolveTestManifest("stall-2", 200), + resolveTestManifest("stall-3", 300), + resolveTestManifest("fail", 400), + }, 0, 500, nil) + result <- err + }() + + // Wait until all 4 workers are stalled. + for i := 0; i < 4; i++ { + waitForManifestFailureSignal(t, stallStarted) + } + + // Release the stalled reads so workers can pick up the queued failing job. + close(stallRelease) + + select { + case err := <-result: + require.Error(t, err) + require.Contains(t, err.Error(), "fail to read manifest fail") + case <-time.After(2 * time.Second): + t.Fatal("failing job was not picked up after releasing stalled workers") + } +} + +func fileIDs(chunks []*filer_pb.FileChunk) []string { + ids := make([]string, 0, len(chunks)) + for _, chunk := range chunks { + ids = append(ids, chunk.GetFileIdString()) + } + return ids +} + +func resolveChunkManifestSerialForBenchmark(ctx context.Context, lookupFileIdFn func(context.Context, string) ([]string, error), chunks []*filer_pb.FileChunk, startOffset, stopOffset int64) (dataChunks, manifestChunks []*filer_pb.FileChunk, err error) { + for _, chunk := range chunks { + if max(chunk.Offset, startOffset) >= min(chunk.Offset+int64(chunk.Size), stopOffset) { + continue + } + if !chunk.IsChunkManifest { + dataChunks = append(dataChunks, chunk) + continue + } + + resolvedChunks, resolveErr := ResolveOneChunkManifest(ctx, lookupFileIdFn, chunk, nil) + if resolveErr != nil { + return dataChunks, nil, resolveErr + } + manifestChunks = append(manifestChunks, chunk) + subDataChunks, subManifestChunks, subErr := resolveChunkManifestSerialForBenchmark(ctx, lookupFileIdFn, resolvedChunks, startOffset, stopOffset) + if subErr != nil { + return dataChunks, nil, subErr + } + dataChunks = append(dataChunks, subDataChunks...) + manifestChunks = append(manifestChunks, subManifestChunks...) + } + return dataChunks, manifestChunks, nil +} + +func BenchmarkResolveChunkManifestSerialVsParallel(b *testing.B) { + manifests := make(map[string][]*filer_pb.FileChunk) + inputs := make([]*filer_pb.FileChunk, 0, 8) + delays := make(map[string]time.Duration) + for i := 0; i < 8; i++ { + id := fmt.Sprintf("m%d", i) + manifests[id] = []*filer_pb.FileChunk{resolveTestData(fmt.Sprintf("d%d", i), int64(i*10))} + inputs = append(inputs, resolveTestManifest(id, int64(i*10))) + delays[id] = time.Millisecond + } + fixture := newManifestReadFixture(b, manifests, delays) + b.ReportAllocs() + + b.Run("serial", func(b *testing.B) { + for i := 0; i < b.N; i++ { + _, _, err := resolveChunkManifestSerialForBenchmark(context.Background(), fixture.lookup, inputs, 0, 1000) + if err != nil { + b.Fatal(err) + } + } + }) + b.Run("parallel", func(b *testing.B) { + for i := 0; i < b.N; i++ { + _, _, err := ResolveChunkManifest(context.Background(), fixture.lookup, inputs, 0, 1000, nil) + if err != nil { + b.Fatal(err) + } + } + }) +} diff --git a/weed/util/http/http_global_client_util.go b/weed/util/http/http_global_client_util.go index ab75c2ad9..fba922563 100644 --- a/weed/util/http/http_global_client_util.go +++ b/weed/util/http/http_global_client_util.go @@ -98,11 +98,11 @@ func Post(url string, values url.Values) ([]byte, error) { // github.com/seaweedfs/seaweedfs/unmaintained/repeated_vacuum/repeated_vacuum.go // may need increasing http.Client.Timeout func Get(url string) ([]byte, bool, error) { - return GetAuthenticated(url, "") + return GetAuthenticated(context.Background(), url, "") } -func GetAuthenticated(url, jwt string) ([]byte, bool, error) { - request, err := http.NewRequest(http.MethodGet, url, nil) +func GetAuthenticated(ctx context.Context, url, jwt string) ([]byte, bool, error) { + request, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) if err != nil { return nil, true, err } @@ -111,7 +111,9 @@ func GetAuthenticated(url, jwt string) ([]byte, bool, error) { response, err := GetGlobalHttpClient().Do(request) if err != nil { - recordUnreachable(request.URL.Host) + if ctx.Err() == nil { + recordUnreachable(request.URL.Host) + } return nil, true, err } recordReachable(request.URL.Host) @@ -135,6 +137,9 @@ func GetAuthenticated(url, jwt string) ([]byte, bool, error) { return nil, retryable, fmt.Errorf("%s: %s", url, response.Status) } if err != nil { + if ctxErr := ctx.Err(); ctxErr != nil { + return nil, false, ctxErr + } return nil, false, err } return b, false, nil @@ -454,7 +459,7 @@ func ReadUrlAsStream(ctx context.Context, fileUrl, jwt string, cipherKey []byte, } func readEncryptedUrl(ctx context.Context, fileUrl, jwt string, cipherKey []byte, isContentCompressed bool, isFullChunk bool, offset int64, size int, fn func(data []byte)) (bool, error) { - encryptedData, retryable, err := GetAuthenticated(fileUrl, jwt) + encryptedData, retryable, err := GetAuthenticated(ctx, fileUrl, jwt) if err != nil { return retryable, fmt.Errorf("fetch %s: %v", fileUrl, err) }