getblobs parallelization - #502
Conversation
There was a problem hiding this comment.
Pull request overview
Adds caller-controlled parallelization to Repository.GetBlobs, allowing packfile range reads and blob decoding to be performed concurrently while attempting to preserve the existing “set semantics” (deduped requests) and error-per-blob delivery model.
Changes:
- Introduces
GetBlobsOptswith aConcurrencysetting (defaulting to 1) and threads it intoGetBlobs. - Refactors
GetBlobsinto a feeder/worker pipeline using channels and cancellation to avoid goroutine leaks on early iterator exit. - Adds a dedicated concurrency test suite covering roundtrip correctness, range read errors, early break/cancellation, and worker/read count mismatches.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| repository/getblobs.go | Adds GetBlobsOpts and implements concurrent range-read execution with cancellation-aware shutdown. |
| repository/getblobs_concurrency_test.go | Adds tests validating correctness under concurrency and checking for goroutine leaks on early exit/cancel. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| returnChannel := make(chan iterItem, opts.Concurrency) | ||
| var wg sync.WaitGroup | ||
| for range opts.Concurrency { | ||
| wg.Go(func() { | ||
| for rp := range readsChannel { | ||
| data, err := r.GetPackfileRange(rp.loc) | ||
|
|
||
| for _, b := range rp.blobs { | ||
| it := iterItem{b: BlobResp{b.BlobReq, nil}} | ||
| if err != nil { | ||
| it.err = err | ||
| } else { | ||
| blobStart := b.Offset - rp.loc.Offset | ||
| blobBytes := data[blobStart : blobStart+uint64(b.Length)] | ||
| it.b.Data, it.err = r.decodeBuffer(blobBytes) | ||
| } | ||
|
|
||
| select { | ||
| case returnChannel <- it: | ||
| case <-innerCtx.Done(): | ||
| return | ||
| } | ||
| } | ||
| } | ||
| }) |
There was a problem hiding this comment.
You might want to refresh your memories on the WaitGroup API :).
* This adds a new know that let the caller specify how much concurrency they want for the actual Gets.
25b049d to
7fad0c1
Compare
| if opts == nil { | ||
| opts = makeDefaultGetBlobsOpts() | ||
| } | ||
|
|
||
| if opts.Concurrency == 0 { | ||
| opts.Concurrency = 1 | ||
| } |
There was a problem hiding this comment.
It's a pointer.... caller should think it'll be mutated. Thoughts from humans?
| // Derived context to handle cancellation from the iterator loop, | ||
| // otherwise an early exit will leak all goroutines. | ||
| innerCtx, cancel := context.WithCancel(ctx) | ||
| defer cancel() | ||
|
|
||
| for _, b := range g.blobs { | ||
| if err != nil { | ||
| if !yield(BlobResp{b.BlobReq, nil}, err) { | ||
| return | ||
| } | ||
| // Feed the read plans to the workers. | ||
| readsChannel := make(chan *rangeRead) | ||
| go func() { | ||
| defer close(readsChannel) |
There was a problem hiding this comment.
This is all dominated by i/o so no I don't think it's worth the added complexity... Thoughts from humans?
This PR adds support for GetBlobs parallelization (chosen by the caller) of the reads.