download: free the -j slot once a torrent starts seeding
This commit is contained in:
14
bt/bt.go
14
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
|
// 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.
|
// Stat stays lock-free, exactly like httpdl.Stat and bt's own name field.
|
||||||
t atomic.Pointer[torrent.Torrent]
|
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
|
// New builds a torrent download on the shared client. source is a .torrent path
|
||||||
// when isFile is true, otherwise a magnet URI.
|
// when isFile is true, otherwise a magnet URI.
|
||||||
func New(client *torrent.Client, source string, isFile bool, opts Options) *Download {
|
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 {
|
if isFile {
|
||||||
d.setName(strings.TrimSuffix(filepath.Base(source), ".torrent"))
|
d.setName(strings.TrimSuffix(filepath.Base(source), ".torrent"))
|
||||||
} else {
|
} else {
|
||||||
@@ -233,6 +238,12 @@ func (d *Download) setStatus(s download.Status) {
|
|||||||
atomic.StoreInt32(&d.status, int32(s))
|
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,
|
// 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
|
// otherwise the source string (a pre-metadata magnet has no infohash yet, and
|
||||||
// two such magnets would otherwise collide on Name).
|
// 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)
|
d.setStatus(download.Seeding)
|
||||||
|
close(d.seeding) // free the -j slot: a seed-only torrent no longer downloads
|
||||||
d.seed(ctx, t)
|
d.seed(ctx, t)
|
||||||
d.setStatus(download.Complete)
|
d.setStatus(download.Complete)
|
||||||
return nil
|
return nil
|
||||||
|
|||||||
@@ -64,3 +64,14 @@ type Download interface {
|
|||||||
Run(ctx context.Context) error
|
Run(ctx context.Context) error
|
||||||
Stat() Stat
|
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{}
|
||||||
|
}
|
||||||
|
|||||||
@@ -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()}
|
results <- Result{Name: d.Name(), Index: i, Err: ctx.Err(), Stat: d.Stat()}
|
||||||
return
|
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)
|
err := d.Run(ctx)
|
||||||
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
|
results <- Result{Name: d.Name(), Index: i, Err: err, Stat: d.Stat()}
|
||||||
|
|||||||
107
download/engine_test.go
Normal file
107
download/engine_test.go
Normal file
@@ -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
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user