From b632f37e3d072aae886644cf36c490a80ff37723 Mon Sep 17 00:00:00 2001 From: Hojun-Cho Date: Mon, 22 Jun 2026 01:57:03 +0900 Subject: [PATCH] download: free the -j slot once a torrent starts seeding --- bt/bt.go | 14 +++++- download/download.go | 11 +++++ download/engine.go | 19 ++++++- download/engine_test.go | 107 ++++++++++++++++++++++++++++++++++++++++ 4 files changed, 149 insertions(+), 2 deletions(-) create mode 100644 download/engine_test.go diff --git a/bt/bt.go b/bt/bt.go index 604aaf9..d5b4fda 100644 --- a/bt/bt.go +++ b/bt/bt.go @@ -203,12 +203,17 @@ type Download struct { // t is published once Run adds the torrent to the client. It is atomic so // Stat stays lock-free, exactly like httpdl.Stat and bt's own name field. t atomic.Pointer[torrent.Torrent] + + // seeding is closed once Run finishes downloading and transitions to + // seed-only, so the engine can drop this torrent from the -j concurrency + // count. Created in New, closed exactly once in Run. + seeding chan struct{} } // New builds a torrent download on the shared client. source is a .torrent path // when isFile is true, otherwise a magnet URI. func New(client *torrent.Client, source string, isFile bool, opts Options) *Download { - d := &Download{client: client, source: source, isFile: isFile, opts: opts} + d := &Download{client: client, source: source, isFile: isFile, opts: opts, seeding: make(chan struct{})} if isFile { d.setName(strings.TrimSuffix(filepath.Base(source), ".torrent")) } else { @@ -233,6 +238,12 @@ func (d *Download) setStatus(s download.Status) { atomic.StoreInt32(&d.status, int32(s)) } +// SeedingStarted is closed when this torrent finishes downloading and switches +// to seed-only. The engine watches it to release the -j concurrency slot at that +// moment: seeding is not downloading, so a seed-only torrent must not hold a slot +// that a queued download could use. Implements download.Seeder. +func (d *Download) SeedingStarted() <-chan struct{} { return d.seeding } + // id is the process-unique identity for this download: the infohash once known, // otherwise the source string (a pre-metadata magnet has no infohash yet, and // two such magnets would otherwise collide on Name). @@ -329,6 +340,7 @@ func (d *Download) Run(ctx context.Context) (err error) { } d.setStatus(download.Seeding) + close(d.seeding) // free the -j slot: a seed-only torrent no longer downloads d.seed(ctx, t) d.setStatus(download.Complete) return nil diff --git a/download/download.go b/download/download.go index 8a0e6c1..8844302 100644 --- a/download/download.go +++ b/download/download.go @@ -64,3 +64,14 @@ type Download interface { Run(ctx context.Context) error Stat() Stat } + +// Seeder is an optional capability of a Download whose work continues after its +// content is fully downloaded — a torrent that keeps running to seed. +// SeedingStarted is closed at that download->seed transition. The engine uses it +// to free the concurrency slot the moment a download stops downloading: a +// seed-only torrent is no longer a download, so it must not count against +// --max-concurrent-downloads (-j) and block a queued download. HTTP downloads do +// not implement this, so they hold their slot until they finish. +type Seeder interface { + SeedingStarted() <-chan struct{} +} diff --git a/download/engine.go b/download/engine.go index 5c8c262..b580c46 100644 --- a/download/engine.go +++ b/download/engine.go @@ -60,7 +60,24 @@ func (e *Engine) Run(ctx context.Context, downloads []Download) []Result { results <- Result{Name: d.Name(), Index: i, Err: ctx.Err(), Stat: d.Stat()} return } - defer func() { <-slots }() + // 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()} diff --git a/download/engine_test.go b/download/engine_test.go new file mode 100644 index 0000000..2fff027 --- /dev/null +++ b/download/engine_test.go @@ -0,0 +1,107 @@ +package download + +import ( + "context" + "testing" + "time" +) + +// fakeDownload is a Download that blocks in Run until released, and (optionally) +// announces a download->seed transition so the engine can free its -j slot. +type fakeDownload struct { + name string + started chan struct{} // closed when Run begins + seeding chan struct{} // SeedingStarted signal + release chan struct{} // Run returns once this is closed + detach bool // if set, Run signals seeding right after starting +} + +func newFake(name string, detach bool) *fakeDownload { + return &fakeDownload{ + name: name, + started: make(chan struct{}), + seeding: make(chan struct{}), + release: make(chan struct{}), + detach: detach, + } +} + +func (f *fakeDownload) Name() string { return f.name } +func (f *fakeDownload) Stat() Stat { return Stat{Name: f.name} } +func (f *fakeDownload) SeedingStarted() <-chan struct{} { return f.seeding } + +func (f *fakeDownload) Run(ctx context.Context) error { + close(f.started) + if f.detach { + close(f.seeding) + } + select { + case <-f.release: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +func started(f *fakeDownload, d time.Duration) bool { + select { + case <-f.started: + return true + case <-time.After(d): + return false + } +} + +// A torrent that has finished downloading and is now seeding must not keep +// occupying a -j slot: with -j 1, a seeding download has to let a queued one run. +func TestEngineSeedingFreesSlot(t *testing.T) { + eng := NewEngine(1) + a := newFake("a", true) + b := newFake("b", true) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan []Result, 1) + go func() { done <- eng.Run(ctx, []Download{a, b}) }() + + // Neither is ever released, yet both must start: each frees the single slot + // the instant it detaches into seeding. + if !started(a, time.Second) || !started(b, time.Second) { + t.Fatal("both downloads should start once seeding frees the -j slot") + } + close(a.release) + close(b.release) + <-done +} + +// Control: without a seeding transition, -j 1 still serializes — the second +// download waits for the first to finish before it starts. +func TestEngineLimitsConcurrentDownloads(t *testing.T) { + eng := NewEngine(1) + a := newFake("a", false) + b := newFake("b", false) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + done := make(chan []Result, 1) + go func() { done <- eng.Run(ctx, []Download{a, b}) }() + + var first, second *fakeDownload + select { + case <-a.started: + first, second = a, b + case <-b.started: + first, second = b, a + case <-time.After(time.Second): + t.Fatal("one download should start") + } + if started(second, 100*time.Millisecond) { + t.Fatal("second download must not start while the only slot is held") + } + close(first.release) // first finishes, freeing the slot + if !started(second, time.Second) { + t.Fatal("second download should start after the first frees the slot") + } + close(second.release) + <-done +}