From 2d33662041eefabd83b656fd8d285d5864d0c12d Mon Sep 17 00:00:00 2001 From: Samuel Cui Date: Sat, 19 Sep 2026 01:26:59 +0800 Subject: [PATCH] feat: preserve request order for linear targets Concurrent source preparation can no longer reorder writes to a linear target. Preparation reports one outcome per indexed request; the ordering stage buffers later outcomes until the request's turn and only then transfers the reader, so a source waits for its consumer after the ordering stage has taken it. Skipped and failed requests advance the order without a write, so one unresolvable source no longer stalls the requests behind it. Random targets keep completion-order concurrency. --- README.md | 1 + TESTING.md | 8 +++ index.go | 6 +- job.go | 1 + prepare.go | 157 +++++++++++++++++++++++++++++++++++++++++++------ stream_test.go | 63 +++++++++++++++++--- 6 files changed, 210 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index 2c4e49d..10183c4 100644 --- a/README.md +++ b/README.md @@ -2,6 +2,7 @@ An Advanced Copy Tools, with following extra features: - Process bar - Sorted copy order, to improve tape device read performance +- Concurrent source preparation with request-ordered writes for linear targets - Multi target path, read once write many - Read file with mmap, with small file prefetch hint - JSON format job report diff --git a/TESTING.md b/TESTING.md index 6738941..e8c0930 100644 --- a/TESTING.md +++ b/TESTING.md @@ -79,6 +79,14 @@ Run path and mountpoint tests: go test -run '^(TestComparePath|TestSourceRoot|TestFindMountpoint)$' . ``` +Run linear stream-order tests: + +```sh +go test -run '^(TestRunStreamCopiesRequestsToLinearTarget|TestForwardPreparedOrdersOnlyLinearTargets|TestRunStreamAppliesBoundedBackpressure|TestRunStreamCancellationDrainsPrefetchedJobs|TestLinearTargetStopsWhenDiskUsageEstimateIsInsufficient)$' . +``` + +On Linux, include `TestRunStreamMapsDeviceFullToTargetNoSpace` to exercise an actual `/dev/full` write failure. + Run `acp-rewrite` tests: ```sh diff --git a/index.go b/index.go index d96a939..8e8bd6b 100644 --- a/index.go +++ b/index.go @@ -34,7 +34,8 @@ func (c *Copyer) index(ctx context.Context) (<-chan *baseJob, error) { go wrap(ctx, func() { defer close(ch) - for _, job := range jobs { + for order, job := range jobs { + job.order = uint64(order) select { case <-ctx.Done(): return @@ -52,6 +53,7 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob { defer close(ch) var bytes, files int64 + var order uint64 defer func() { c.submit(&EventUpdateCount{Bytes: bytes, Files: files, Finished: true}) }() @@ -94,6 +96,7 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob { stat: stat, targets: append([]string(nil), request.Targets...), streamID: request.ID, + order: order, } c.submit(&EventUpdateJob{job.report()}) bytes += stat.size @@ -104,6 +107,7 @@ func (c *Copyer) indexStream(ctx context.Context) <-chan *baseJob { c.setError(ctx.Err()) return case ch <- job: + order++ } } }) diff --git a/job.go b/job.go index 0953a25..8c4360b 100644 --- a/job.go +++ b/job.go @@ -42,6 +42,7 @@ type baseJob struct { path string stat *stat streamID int64 + order uint64 lock sync.Mutex writeTime time.Time diff --git a/prepare.go b/prepare.go index fb7a037..f3abad0 100644 --- a/prepare.go +++ b/prepare.go @@ -11,6 +11,12 @@ import ( "github.com/samuelncui/acp/mmap" ) +type prepareResult struct { + order uint64 + job *writeJob + release chan struct{} +} + func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *writeJob { // Disable read-ahead when a linear source must wait for each consumer. chanLen := 32 @@ -18,21 +24,16 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w chanLen = 0 } - // Close the prepared stream only after every source worker exits. - var wg sync.WaitGroup - ch := make(chan *writeJob, chanLen) - defer func() { - go wrap(ctx, func() { - defer close(ch) - wg.Wait() - }) - }() + // Collect every preparation outcome so skipped jobs can still advance linear ordering. + completed := make(chan prepareResult, chanLen) + prepared := make(chan *writeJob, chanLen) + var workers sync.WaitGroup // Prepare source readers with the configured source-device concurrency. for idx := 0; idx < c.fromDevice.threads; idx++ { - wg.Add(1) + workers.Add(1) go wrap(ctx, func() { - defer wg.Done() + defer workers.Done() // Consume indexed jobs until cancellation or source exhaustion. for { @@ -43,7 +44,14 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w if !ok { return } + result := prepareResult{order: job.order} + if c.toDevice.linear { + result.release = make(chan struct{}) + } if c.linearTargetStopped() { + if !sendPrepareResult(ctx, completed, result) { + return + } continue } @@ -96,16 +104,17 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w c.reportError(job.path, "", err) job.fail("", err) job.setStatus(jobStatusFinished) + if !sendPrepareResult(ctx, completed, result) { + return + } continue } } - // Publish the prepared reader and preserve linear-source consumption order. + // Transfer the prepared reader to the ordering stage before waiting on a linear source. wj := newWriteJob(job, file, size, c.fromDevice.linear) - select { - case ch <- wj: - case <-ctx.Done(): - wj.finishSource() + result.job = wj + if !sendPrepareResult(ctx, completed, result) { return } if !wj.waitConsumed(ctx) { @@ -116,5 +125,119 @@ func (c *Copyer) prepare(ctx context.Context, indexed <-chan *baseJob) <-chan *w }) } - return ch + // Close preparation outcomes only after every source worker relinquishes ownership. + go wrap(ctx, func() { + workers.Wait() + close(completed) + }) + + // Preserve request order only where one linear writer requires it. + go wrap(ctx, func() { + defer close(prepared) + c.forwardPrepared(ctx, completed, prepared) + }) + return prepared +} + +func sendPrepareResult(ctx context.Context, completed chan<- prepareResult, result prepareResult) bool { + select { + case completed <- result: + case <-ctx.Done(): + result.finish() + return false + } + if result.release == nil { + return true + } + select { + case <-result.release: + return true + case <-ctx.Done(): + return false + } +} + +func (result prepareResult) releaseWorker() { + if result.release != nil { + close(result.release) + } +} + +func (result prepareResult) finish() { + if result.job != nil { + result.job.finishSource() + } + result.releaseWorker() +} + +func (c *Copyer) forwardPrepared( + ctx context.Context, + completed <-chan prepareResult, + prepared chan<- *writeJob, +) { + // Close readers that remain owned by this stage when cancellation stops forwarding. + defer func() { + for result := range completed { + result.finish() + } + }() + + // Random targets retain completion-order concurrency. + if !c.toDevice.linear { + for result := range completed { + if result.job == nil { + continue + } + select { + case prepared <- result.job: + case <-ctx.Done(): + result.finish() + return + } + } + return + } + + // A bounded reorder buffer feeds the linear writer in original request order. + next := uint64(0) + pending := make(map[uint64]prepareResult, c.fromDevice.threads) + defer func() { + for _, result := range pending { + result.finish() + } + }() + for { + select { + case <-ctx.Done(): + return + case result, ok := <-completed: + if !ok { + if len(pending) != 0 { + c.setError(fmt.Errorf("ordered preparation ended before request %d", next)) + } + return + } + pending[result.order] = result + } + + for { + result, ok := pending[next] + if !ok { + break + } + delete(pending, next) + next++ + if result.job == nil { + result.releaseWorker() + continue + } + select { + case prepared <- result.job: + result.releaseWorker() + case <-ctx.Done(): + result.finish() + return + } + } + } } diff --git a/stream_test.go b/stream_test.go index a25cdee..23ad240 100644 --- a/stream_test.go +++ b/stream_test.go @@ -9,6 +9,7 @@ import ( "io" "os" "path/filepath" + "strings" "sync/atomic" "testing" "time" @@ -78,15 +79,10 @@ func TestRunStreamCopiesRequestsToLinearTarget(t *testing.T) { if len(sink.results) != len(source.requests) { t.Fatalf("received %d results, want %d", len(sink.results), len(source.requests)) } - seen := make(map[int64]struct{}, len(sink.results)) - for _, result := range sink.results { - if result.ID < 1 || result.ID > int64(len(source.requests)) { - t.Fatalf("unexpected result ID: %d", result.ID) + for index, result := range sink.results { + if result.ID != int64(index+1) { + t.Fatalf("result %d has ID %d, want %d", index, result.ID, index+1) } - if _, exists := seen[result.ID]; exists { - t.Fatalf("duplicate result ID: %d", result.ID) - } - seen[result.ID] = struct{}{} if result.Job.Status != JobStatusFinished || len(result.Job.SuccessTargets) != 1 || result.Job.SHA256 == "" { t.Fatalf("unexpected result: %#v", result.Job) } @@ -96,6 +92,57 @@ func TestRunStreamCopiesRequestsToLinearTarget(t *testing.T) { } } +func TestForwardPreparedOrdersOnlyLinearTargets(t *testing.T) { + newJob := func(order uint64, id int64) *writeJob { + return newWriteJob( + &baseJob{order: order, streamID: id}, + io.NopCloser(strings.NewReader("fixture")), + int64(len("fixture")), + false, + ) + } + tests := []struct { + name string + linear bool + wantIDs []int64 + }{ + {name: "linear request order", linear: true, wantIDs: []int64{1, 3}}, + {name: "random completion order", wantIDs: []int64{3, 1}}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + // Deliver later preparation results first; a failed or skipped middle request has no write Job. + copyer := &Copyer{option: &option{ + fromDevice: &deviceOption{threads: 3}, + toDevice: &deviceOption{linear: test.linear}, + }} + completed := make(chan prepareResult, 3) + completed <- prepareResult{order: 2, job: newJob(2, 3)} + completed <- prepareResult{order: 0, job: newJob(0, 1)} + completed <- prepareResult{order: 1} + close(completed) + prepared := make(chan *writeJob, 3) + + // Forwarding must reorder only the serialized target path. + copyer.forwardPrepared(context.Background(), completed, prepared) + close(prepared) + var ids []int64 + for job := range prepared { + ids = append(ids, job.streamID) + job.finishSource() + } + if len(ids) != len(test.wantIDs) { + t.Fatalf("prepared IDs = %v, want %v", ids, test.wantIDs) + } + for index := range ids { + if ids[index] != test.wantIDs[index] { + t.Fatalf("prepared IDs = %v, want %v", ids, test.wantIDs) + } + } + }) + } +} + func TestRunStreamHashesWithoutTargets(t *testing.T) { // Submit one targetless request through the same bounded stream interface. content := []byte("hash-only fixture")