1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
|
package cli
import (
"context"
"errors"
"runtime"
"sync"
"sync/atomic"
)
// pool is a bounded worker pool for concurrent, independent tasks (§14). A
// single caller feeds work via submit; up to concurrency tasks run in flight at
// once, bounded by the sem token held from submit until the task returns.
//
// The pool is deliberately upload-agnostic: a task is any func(ctx) error, so
// the same machinery serves push uploads today and prune deletes later. Task
// wording, error identity, and any success side-effect live in the closure.
//
// submit is the only place that logs a per-action line, and it is only ever
// called from one goroutine (the caller feeding work), so output needs no
// locking; only the shared counters are guarded. Tasks run concurrently, so any
// state a task closure touches beyond its own arguments is the caller's
// responsibility to synchronize.
type workerPool struct {
ctx context.Context
cancel context.CancelCauseFunc
sem chan struct{}
wg sync.WaitGroup
// failFast cancels the pool on the first failure so in-flight tasks are
// abandoned and queued ones never start (the push-asset policy, §8.3). When
// false, every submitted task runs and all failures are collected (the
// commit/prune-delete policy).
failFast bool
mtx sync.Mutex
errs []error
succeeded int64
}
// newWorkerPool creates a new worker pool to process tasks concurrently. If `concurrency`
// is zero, the number of parallel jobs is aut-detected based on the number of CPUs.
func newWorkerPool(ctx context.Context, concurrency uint, failFast bool) *workerPool {
if concurrency == 0 {
concurrency = 2 * uint(runtime.NumCPU())
}
ctx, cancel := context.WithCancelCause(ctx)
return &workerPool{
ctx: ctx,
cancel: cancel,
sem: make(chan struct{}, concurrency),
failFast: failFast,
}
}
// submit blocks until a worker slot is free, then runs task in the background.
// It reports false when the pool has been cancelled (fail-fast or the parent
// context), signalling the caller to stop feeding work. action is the text that
// will be logged before the task is executed.
func (p *workerPool) submit(action string, task func(ctx context.Context) error) bool {
select {
case p.sem <- struct{}{}:
case <-p.ctx.Done():
return false
}
logAction(action)
p.wg.Add(1)
go func() {
defer p.wg.Done()
defer func() { <-p.sem }()
if err := task(p.ctx); err != nil {
if !errors.Is(err, context.Canceled) {
p.mtx.Lock()
p.errs = append(p.errs, err)
p.mtx.Unlock()
}
if p.failFast {
p.cancel(CancellationError("operation aborted"))
}
return
}
atomic.AddInt64(&p.succeeded, 1)
}()
return true
}
// cancelled reports whether the pool's context is already done (fail-fast
// triggered or the parent context cancelled). A feeder loop can poll this to
// stop producing/buffering work promptly, instead of only discovering the
// cancellation on its next submit.
func (p *workerPool) cancelled() bool {
select {
case <-p.ctx.Done():
return true
default:
return false
}
}
// wait blocks until all in-flight tasks finish, then releases the pool's
// context. It returns the number of succeeded tasks and the errors the
// workers collected.
func (p *workerPool) wait() (int, []error) {
p.wg.Wait()
p.cancel(nil)
return int(p.succeeded), p.errs
}
|
