129 lines
3.6 KiB
Go
129 lines
3.6 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
|
|
}
|
|
// Release the slot exactly once — whether the download finishes or, for
|
|
// a torrent, first detaches into seed-only. A seeding torrent is no
|
|
// longer downloading, so it stops counting against -j and lets a queued
|
|
// download start; this goroutine lives on to seed.
|
|
var once sync.Once
|
|
release := func() { once.Do(func() { <-slots }) }
|
|
defer release()
|
|
if s, ok := d.(Seeder); ok {
|
|
done := make(chan struct{})
|
|
defer close(done)
|
|
go func() {
|
|
select {
|
|
case <-s.SeedingStarted():
|
|
release()
|
|
case <-done:
|
|
}
|
|
}()
|
|
}
|
|
|
|
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
|
|
}
|
|
}
|
|
}
|