Skip to content

fix(table): bound AllManifests concurrency - #1668

Merged
zeroshade merged 4 commits into
apache:mainfrom
fallintoplace:fix/all-manifests-concurrency
Aug 7, 2026
Merged

fix(table): bound AllManifests concurrency#1668
zeroshade merged 4 commits into
apache:mainfrom
fallintoplace:fix/all-manifests-concurrency

Conversation

@fallintoplace

@fallintoplace fallintoplace commented Aug 5, 2026

Copy link
Copy Markdown
Contributor

Summary

Bound the remote work started by AllManifests.

Why

The old implementation started one goroutine and one manifest-list read per snapshot. Long table histories could create a large burst of remote I/O even when the caller stopped early.

What changed

AllManifests now uses a bounded worker pool, stops scheduling after cancellation, preserves snapshot order, and keeps the existing error behavior.

Tests

  • go test ./table -count=1

@zeroshade zeroshade left a comment

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The change is correct. I verified the bounding and the error propagation:

  • The bound is real and ordering is unaffected. MakeSequencedChan (table/internal/utils.go:83-99) still re-sequences results by Index through a heap and is untouched by this PR, so capping worker count does not change the order manifests are yielded in.
  • Error semantics are preserved: errgroup.WithContext retains the first error and cancels the remaining workers, and the feeder goroutine unblocks on groupCtx.Done() rather than blocking forever on a full jobs channel.
  • No leak on early exit. out is still allocated with capacity n (table/internal/utils.go:86), so workers can always drain even if the consumer stops iterating, and the existing drain goroutine plus the new defer cancel() (table/table.go:403) close that loop.

Everything below is optional — nothing here blocks.

Comment thread table/table.go Outdated
n := len(snapshots)
workCtx, cancel := context.WithCancel(ctx)
jobs := make(chan int)
ch := make(chan list, max(1, min(n, 16)))

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Non-blocking: because out inside MakeSequencedChan is allocated with capacity n, this cap 16 provides essentially no backpressure — peak memory is still all n manifest lists held at once. Worth a short comment so a future reader doesn't mistake this expression for a memory bound.

Comment thread table/table.go Outdated
workCtx, cancel := context.WithCancel(ctx)
jobs := make(chan int)
ch := make(chan list, max(1, min(n, 16)))
workers := max(1, min(n, min(runtime.GOMAXPROCS(0), 16)))

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Two small things here, both optional.

  1. 16 appears on this line and the one above as an unnamed literal. A named constant would make the intent legible and keep the two in sync.
  2. GOMAXPROCS(0) derives a CPU bound for what is purely remote I/O. On a two-core CI runner, a 1000-snapshot history now fetches two-wide, which is likely slower than intended. Consider a bound independent of CPU count — and possibly an option, since Delete/Overwrite already expose a concurrency knob (WithOverwriteConcurrency), which leaves AllManifests as the odd one out.

Comment thread table/table.go
workers := max(1, min(n, min(runtime.GOMAXPROCS(0), 16)))
g, groupCtx := errgroup.WithContext(workCtx)

for range workers {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Worth a follow-up: errgroup.Group.SetLimit(workers) plus a single scheduling goroutine yields the same bound in roughly five lines and removes the hand-rolled jobs channel entirely. The current implementation is correct — this is purely a simplification, so it's fine to defer or skip.

Comment thread table/all_manifests_internal_test.go Outdated
trackingFS.mu.Lock()
maxOpen := trackingFS.maxOpen
trackingFS.mu.Unlock()
require.Greater(t, maxOpen, 1)

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Non-blocking, but this is the one assertion I'd soften. maxOpen > 1 requires two goroutines to overlap within the 5 ms sleep, so it can flake on a loaded or effectively single-CPU runner. The upper-bound assertion on the next line is the property this PR actually adds; consider keeping only that, or having the tracking IO rendezvous through a semaphore or WaitGroup so the overlap is deterministic rather than timing-derived. Mutating global GOMAXPROCS is also worth a second look if tests ever run in parallel here.

}
}

func TestAllManifestsLimitsConcurrentReads(t *testing.T) {

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Suggestions for hardening, none blocking — the bound itself is well covered by this test. Cases that would round it out: a worker error cancels the remaining workers and the first error is what surfaces; an early break out of the range leaks no goroutines (goleak would catch this cheaply); context cancellation propagates to in-flight workers; and the n == 0 / n == 1 edges where max(1, min(n, ...)) collapses.

@zeroshade
zeroshade merged commit 93b6986 into apache:main Aug 7, 2026
15 checks passed
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

2 participants