Hojun-Cho
e3505855c1
Findings from a Rob-Pike-lens review (bugs/races/network), each verified against the code before fixing: - httpdl: name/out are now atomic.Pointer[string] -- the engine publishes a download to the reporter before Run resolves the name, so Stat raced the write (bt already did this) - httpdl: a malformed --proxy fails loudly instead of silently bypassing it - httpdl: a 206 must carry a matching Content-Range; a 200 in segmented mode is fatal so it fails over instead of burning the retry budget - httpdl: single-stream mirror failover validates the range before appending; ErrTooSlow only when the error is a real ctx cancellation - httpdl: idle guard tracks progress by timestamp (no Reset/Stop race, no sticky fired flag) - httpdl/control: reject a resume file whose segments don't tile [0,total) - bt: clamp the listen-port range; verify on-disk data before choosing pieces under --check-integrity - cli: reject size overflow; show --seed-time=MIN; clamp --select-file range - progress/engine/main: clamp ETA against int64 overflow; show queued downloads as waiting; join the reporter on exit instead of a 20ms sleep
112 lines
3.1 KiB
Go
112 lines
3.1 KiB
Go
package download
|
|
|
|
import (
|
|
"context"
|
|
"sync"
|
|
)
|
|
|
|
// Result pairs a finished download with its outcome and a final stat snapshot.
|
|
type Result struct {
|
|
Name string
|
|
Index int // position in the slice passed to Run, so callers can restore input order
|
|
Err error
|
|
Stat Stat
|
|
}
|
|
|
|
// Engine runs a set of downloads, at most maxConcurrent at a time, and exposes
|
|
// a live snapshot of the running ones for the progress renderer. A buffered
|
|
// channel of slots enforces the concurrency limit, and one goroutine per
|
|
// download replaces an explicit scheduler tick.
|
|
type Engine struct {
|
|
maxConcurrent int
|
|
|
|
mu sync.Mutex
|
|
active []Download
|
|
}
|
|
|
|
// NewEngine returns an engine that runs at most maxConcurrent downloads at once.
|
|
func NewEngine(maxConcurrent int) *Engine {
|
|
if maxConcurrent < 1 {
|
|
maxConcurrent = 1
|
|
}
|
|
return &Engine{maxConcurrent: maxConcurrent}
|
|
}
|
|
|
|
// Run starts every download, honouring the concurrency limit, and returns once
|
|
// all of them have finished or ctx is cancelled. There is one Result per
|
|
// download; results arrive in completion order, but each Result carries its
|
|
// input Index so callers can sort back to the original slice order.
|
|
func (e *Engine) Run(ctx context.Context, downloads []Download) []Result {
|
|
slots := make(chan struct{}, e.maxConcurrent)
|
|
results := make(chan Result, len(downloads))
|
|
|
|
var wg sync.WaitGroup
|
|
for i, d := range downloads {
|
|
wg.Add(1)
|
|
go func(i int, d Download) {
|
|
defer wg.Done()
|
|
|
|
// Track immediately so a download still queued for a concurrency slot
|
|
// is visible to the progress reporter as Waiting, rather than hidden
|
|
// until it starts running.
|
|
e.track(d)
|
|
defer e.untrack(d)
|
|
|
|
// Acquire a concurrency slot, or bail if we're shutting down
|
|
// before this download ever started.
|
|
select {
|
|
case slots <- struct{}{}:
|
|
case <-ctx.Done():
|
|
results <- Result{Name: d.Name(), Index: i, Err: ctx.Err(), Stat: d.Stat()}
|
|
return
|
|
}
|
|
defer func() { <-slots }()
|
|
|
|
err := d.Run(ctx)
|
|
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
|
|
}(i, d)
|
|
}
|
|
|
|
wg.Wait()
|
|
close(results)
|
|
|
|
out := make([]Result, 0, len(downloads))
|
|
for r := range results {
|
|
out = append(out, r)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// Snapshot returns a Stat for every tracked download — those running plus those
|
|
// still queued for a concurrency slot (reported as Waiting).
|
|
func (e *Engine) Snapshot() []Stat {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
out := make([]Stat, 0, len(e.active))
|
|
for _, d := range e.active {
|
|
out = append(out, d.Stat())
|
|
}
|
|
return out
|
|
}
|
|
|
|
func (e *Engine) track(d Download) {
|
|
e.mu.Lock()
|
|
e.active = append(e.active, d)
|
|
e.mu.Unlock()
|
|
}
|
|
|
|
// untrack removes d from the active set. Every Download implementation is a
|
|
// pointer type, so the == compares identities — never the fields of a value,
|
|
// which could panic on a non-comparable one. (Keying on Stat().ID is not an
|
|
// option: a torrent's ID changes from source to infohash once metadata loads.)
|
|
func (e *Engine) untrack(d Download) {
|
|
e.mu.Lock()
|
|
defer e.mu.Unlock()
|
|
for i, x := range e.active {
|
|
if x == d {
|
|
e.active = append(e.active[:i], e.active[i+1:]...)
|
|
return
|
|
}
|
|
}
|
|
}
|