diff --git a/backend/download/concurrency_test.go b/backend/download/concurrency_test.go index 729c07f..2181582 100644 --- a/backend/download/concurrency_test.go +++ b/backend/download/concurrency_test.go @@ -79,9 +79,9 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) { want int }{ { - name: "slskd defaults to one", + name: "slskd defaults to a few peers", cfg: Config{Kind: KindSlskd}, - want: 1, + want: 3, }, { name: "usenet defaults higher", @@ -92,9 +92,9 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) { name: "explicit override wins", cfg: Config{ Kind: KindSlskd, - Settings: map[string]string{concurrencyKey: "3"}, + Settings: map[string]string{concurrencyKey: "1"}, }, - want: 3, + want: 1, }, { name: "nonsense override falls back", @@ -102,7 +102,7 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) { Kind: KindSlskd, Settings: map[string]string{concurrencyKey: "not a number"}, }, - want: 1, + want: 3, }, { name: "zero override falls back", @@ -110,7 +110,7 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) { Kind: KindSlskd, Settings: map[string]string{concurrencyKey: "0"}, }, - want: 1, + want: 3, }, { name: "unknown kind falls back to the global default", @@ -126,9 +126,9 @@ func TestConcurrencyForPrefersOverrideThenKind(t *testing.T) { } } -// The reason the per-provider cap exists: a Soulseek daemon capped at -// one transfer must serialize, even when the global cap would allow -// more and the user has queued several albums at once. +// The reason the per-provider cap exists: a daemon capped at one +// transfer must serialize, even when the global cap would allow more and +// the user has queued several albums at once. func TestPerProviderCapSerializesTransfers(t *testing.T) { t.Parallel() @@ -142,6 +142,7 @@ func TestPerProviderCapSerializesTransfers(t *testing.T) { ID: 1, Kind: KindSlskd, Priority: 50, + Settings: map[string]string{concurrencyKey: "1"}, }, slow) // Three requests against the same one-at-a-time provider. @@ -210,8 +211,8 @@ func TestSyncSemaphoresReplacesChangedLimits(t *testing.T) { f.manager.installProvider(Config{ID: 1, Kind: KindSlskd}, nil) first := f.manager.semaphoreFor(1) - if cap(first) != 1 { - t.Fatalf("slskd semaphore cap = %d, want 1", cap(first)) + if want := kindConcurrency[KindSlskd]; cap(first) != want { + t.Fatalf("slskd semaphore cap = %d, want %d", cap(first), want) } // Same limit: the semaphore is kept, so in-flight accounting is not diff --git a/backend/download/keyedlock.go b/backend/download/keyedlock.go new file mode 100644 index 0000000..95f45f8 --- /dev/null +++ b/backend/download/keyedlock.go @@ -0,0 +1,77 @@ +package download + +import ( + "context" + "sync" +) + +// keyedLock is a set of mutexes created on demand, one per key, that +// honour a context while waiting. An entry lives only while someone +// holds or waits on it, so a key per Soulseek peer or per folder name +// does not accumulate for the life of the process. +type keyedLock[K comparable] struct { + mu sync.Mutex + held map[K]*keyedEntry +} + +type keyedEntry struct { + ch chan struct{} + + // refs counts holders and waiters; the entry is dropped at zero. + refs int +} + +// acquire blocks until k is free or ctx ends, and returns the function +// that frees it. +func (l *keyedLock[K]) acquire(ctx context.Context, k K) (func(), error) { + l.mu.Lock() + + if l.held == nil { + l.held = map[K]*keyedEntry{} + } + + e, ok := l.held[k] + if !ok { + e = &keyedEntry{ch: make(chan struct{}, 1)} + l.held[k] = e + } + + e.refs++ + + l.mu.Unlock() + + select { + case e.ch <- struct{}{}: + case <-ctx.Done(): + l.drop(k, e) + + return nil, ctx.Err() + } + + var once sync.Once + + return func() { + once.Do(func() { + <-e.ch + l.drop(k, e) + }) + }, nil +} + +func (l *keyedLock[K]) drop(k K, e *keyedEntry) { + l.mu.Lock() + defer l.mu.Unlock() + + e.refs-- + if e.refs == 0 { + delete(l.held, k) + } +} + +// size reports how many keys are held or awaited, for tests. +func (l *keyedLock[K]) size() int { + l.mu.Lock() + defer l.mu.Unlock() + + return len(l.held) +} diff --git a/backend/download/manager.go b/backend/download/manager.go index 2af26c4..3b429de 100644 --- a/backend/download/manager.go +++ b/backend/download/manager.go @@ -59,13 +59,15 @@ const concurrencyKey = "maxConcurrent" // A single global cap is the wrong shape here: usenet and torrent // clients are built to run many transfers at once and are throttled by // bandwidth, while Soulseek transfers come from one person's home -// upload slot. Hitting the same peer with parallel requests gets you -// queued behind everyone else at best and banned at worst, so slskd is -// capped at one — the polite number, and the one that actually -// completes fastest, because a Soulseek peer serves one file at a time -// regardless of how many you ask for. +// upload slot. Politeness there is per *peer* — asking one user for two +// folders at once gets you queued behind everyone else at best and +// banned at worst — and the manager holds that line separately, one +// grab per peer (peerLocks). Two different users do not compete for +// anyone's slot, so the daemon-wide number only bounds how many peers +// are asked at once, and one slow peer no longer serialises every other +// Soulseek download behind it. var kindConcurrency = map[Kind]int{ - KindSlskd: 1, + KindSlskd: 3, KindYtDlp: 2, KindQBittorrent: 4, KindSABnzbd: 4, @@ -155,6 +157,11 @@ type Manager struct { semMu sync.Mutex provSem map[int64]chan struct{} + // peerLocks holds one grab per Soulseek peer, taken before any + // slot: a grab waiting for a busy peer must not sit on a provider + // slot another peer could be using. + peerLocks keyedLock[peerKey] + // delegatePoll is how often delegating managers are asked for // status. A field rather than the constant so tests can drive the // full delegate flow without sleeping through it. @@ -790,6 +797,23 @@ func (m *Manager) grab( // whole list for six hours. const maxGrabAttempts = 3 +// peerKey names one Soulseek user on one daemon. The same username on +// two daemons is two logins and two queues. +type peerKey struct { + provider int64 + peer string +} + +// peerKeyFor returns the peer a candidate is fetched from, when the +// source is one where asking a peer for two things at once is rude. +func peerKeyFor(c Candidate) (peerKey, bool) { + if c.Kind != KindSlskd || c.Origin == "" { + return peerKey{}, false + } + + return peerKey{provider: c.ProviderID, peer: c.Origin}, true +} + // grabOutcome is how one candidate's attempt ended. type grabOutcome struct { item DownloadItem @@ -825,6 +849,15 @@ func (m *Manager) attemptGrab( } if !plan.delegated() { + if key, ok := peerKeyFor(c); ok { + release, err := m.peerLocks.acquire(ctx, key) + if err != nil { + return grabOutcome{err: err} + } + + defer release() + } + provSem := m.semaphoreFor(plan.transportID) select { diff --git a/backend/download/peer_concurrency_test.go b/backend/download/peer_concurrency_test.go new file mode 100644 index 0000000..113604f --- /dev/null +++ b/backend/download/peer_concurrency_test.go @@ -0,0 +1,224 @@ +package download + +import ( + "context" + "errors" + "path/filepath" + "sync" + "testing" + "time" +) + +// Soulseek politeness is per peer, not per daemon (#272). + +func TestKeyedLockSerialisesOneKeyOnly(t *testing.T) { + t.Parallel() + + var l keyedLock[string] + + ctx := context.Background() + + releaseA, err := l.acquire(ctx, "a") + if err != nil { + t.Fatalf("acquire a: %v", err) + } + + // Another key is free while "a" is held. + releaseB, err := l.acquire(ctx, "b") + if err != nil { + t.Fatalf("acquire b: %v", err) + } + + releaseB() + + // The same key waits, and gives up with its context. + short, cancel := context.WithTimeout(ctx, 20*time.Millisecond) + defer cancel() + + if _, err := l.acquire(short, "a"); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("second acquire of a held key = %v, want the deadline", err) + } + + releaseA() + releaseA() // Idempotent: a second call must not free someone else's hold. + + if n := l.size(); n != 0 { + t.Errorf("%d keys left behind, want none once nobody holds or waits", n) + } +} + +// grabEach runs one grab per candidate and returns a function that waits +// for all of them; grabAll's reasons for waiting apply. +func grabEach(t *testing.T, f managerFixture, cands []Candidate) func() { + t.Helper() + + ctx := context.Background() + + var wg sync.WaitGroup + + for i, c := range cands { + dl := fourTrackDownload() + dl.ID = "dl-" + string(rune('a'+i)) + + if err := f.store.CreateDownload(ctx, dl); err != nil { + t.Fatalf("CreateDownload: %v", err) + } + + wg.Add(1) + + go func() { + defer wg.Done() + + f.manager.grab(ctx, dl, c, nil, false) + }() + } + + return func() { + done := make(chan struct{}) + + go func() { + wg.Wait() + close(done) + }() + + select { + case <-done: + case <-time.After(5 * time.Second): + t.Error("transfers did not finish") + } + } +} + +func slskdCandidates(p *FakeProvider, peers ...string) []Candidate { + out := make([]Candidate, 0, len(peers)) + + for i, peer := range peers { + c := p.Candidates[0] + c.ID = c.ID + "-" + itoa(i) + c.Kind = KindSlskd + c.ProviderID = 1 + c.Origin = peer + out = append(out, c) + } + + return out +} + +// Three albums from one user are asked for one at a time, even though +// the daemon would allow three transfers. +func TestOnePeerIsAskedForOneThingAtATime(t *testing.T) { + t.Parallel() + + f := newManagerFixture(t) + f.manager.SetMaxConcurrent(4) + + p := fakeWithAlbum(1, "slskd", ".flac") + p.GrabGate = make(chan struct{}) + + f.manager.installProvider(Config{ID: 1, Kind: KindSlskd, Priority: 50}, p) + + wait := grabEach(t, f, slskdCandidates(p, "alice", "alice", "alice")) + + waitFor(t, func() bool { return p.GrabCallCount() >= 1 }, "no grab started") + time.Sleep(150 * time.Millisecond) + + if got := p.MaxParallelGrabs(); got != 1 { + t.Errorf("%d simultaneous grabs from one peer, want 1", got) + } + + close(p.GrabGate) + + waitFor(t, func() bool { return p.GrabCallCount() == 3 }, "queued grabs never ran") + wait() + + if n := f.manager.peerLocks.size(); n != 0 { + t.Errorf("%d peer locks left behind", n) + } +} + +// Different users run at once, up to the daemon's cap — the point of +// the change: one slow peer no longer holds up every other. +func TestDifferentPeersRunTogether(t *testing.T) { + t.Parallel() + + f := newManagerFixture(t) + f.manager.SetMaxConcurrent(8) + + p := fakeWithAlbum(1, "slskd", ".flac") + p.GrabGate = make(chan struct{}) + + f.manager.installProvider(Config{ID: 1, Kind: KindSlskd, Priority: 50}, p) + + wait := grabEach(t, f, slskdCandidates(p, "alice", "bob", "carol", "dave")) + + waitFor( + t, + func() bool { return p.MaxParallelGrabs() >= kindConcurrency[KindSlskd] }, + "different peers were serialised", + ) + time.Sleep(100 * time.Millisecond) + + if got := p.MaxParallelGrabs(); got != kindConcurrency[KindSlskd] { + t.Errorf("%d simultaneous grabs, want the daemon cap %d", got, kindConcurrency[KindSlskd]) + } + + close(p.GrabGate) + wait() +} + +func TestSlskdLocalFolders(t *testing.T) { + t.Parallel() + + s := &slskd{downloadsPath: "/dl"} + + got := s.localFolders(Candidate{Files: []CandidateFile{ + {Path: `\m\The Wall\CD2\01 Hey You.flac`}, + {Path: `\m\The Wall\CD1\01 In The Flesh.flac`}, + {Path: `\m\The Wall\CD1\02 The Thin Ice.flac`}, + }}) + + want := []string{filepath.Join("/dl", "CD1"), filepath.Join("/dl", "CD2")} + if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { + t.Errorf("localFolders = %q, want %q", got, want) + } +} + +// Two peers' "Greatest Hits" land in one slskd directory, so the second +// grab does not enqueue until the first has collected its files. +func TestSlskdSameFolderNameWaits(t *testing.T) { + t.Parallel() + + stub := newSlskdStub(t) + s, downloads := newStubSlskd(t, stub) + + c := Candidate{ + Payload: map[string]string{"username": "bob"}, + Files: []CandidateFile{ + {Path: `\music\Greatest Hits\01 Intro.flac`, Size: 1, IsAudio: true}, + }, + } + + release, err := lockSlskdFolders( + context.Background(), []string{filepath.Join(downloads, "Greatest Hits")}, + ) + if err != nil { + t.Fatalf("lock: %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond) + defer cancel() + + if _, err := s.Grab(ctx, c, t.TempDir(), nil); !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("Grab = %v, want it to wait on the held folder", err) + } + + release() + + stub.mu.Lock() + posted := stub.posted + stub.mu.Unlock() + + if posted { + t.Error("enqueued transfers into a folder another grab held") + } +} diff --git a/backend/download/provider.go b/backend/download/provider.go index f748978..b2d9965 100644 --- a/backend/download/provider.go +++ b/backend/download/provider.go @@ -236,17 +236,18 @@ func Register(d Descriptor, c Constructor) { } // concurrencyField describes the per-provider transfer limit, with help -// text explaining why the default is what it is — a user who raises -// slskd from 1 to 8 and gets themselves queued behind every other -// Soulseek user deserves to have been warned. +// text explaining what the number means where it means something +// unusual: on slskd it counts peers, since each peer is only ever asked +// for one folder at a time whatever it is set to. func concurrencyField(k Kind) Field { help := "Maximum simultaneous transfers from this client." if k == KindSlskd { - help = "Maximum simultaneous transfers. Soulseek peers serve " + - "one file at a time and queue or ban clients that ask for " + - "more, so 1 is both the polite setting and usually the " + - "fastest." + help = "How many Soulseek users to download from at once. " + + "Each user is only ever asked for one album at a time, " + + "since peers queue or ban clients that ask for more; " + + "this bounds how many different users are asked in " + + "parallel." } return Field{ diff --git a/backend/download/provider_slskd.go b/backend/download/provider_slskd.go index 1286c72..51a6064 100644 --- a/backend/download/provider_slskd.go +++ b/backend/download/provider_slskd.go @@ -10,6 +10,7 @@ import ( "path" "path/filepath" "regexp" + "slices" "strings" "time" @@ -742,6 +743,12 @@ func (s *slskd) Grab( // first poll came back — an old failure failing a transfer that has // not started. So what is already terminal is noted before enqueueing // and ignored after. + release, err := lockSlskdFolders(ctx, s.localFolders(c)) + if err != nil { + return Result{}, err + } + defer release() + stale := s.terminalTransferIDs(ctx, username) wanted := make([]map[string]any, 0, len(c.Files)) @@ -767,6 +774,59 @@ func (s *slskd) Grab( return s.collect(c, dst) } +// slskdFolders serialises grabs that land in the same local folder. +// +// slskd names a download's directory after the remote *leaf* folder, so +// two different albums both shared as "Greatest Hits" — or any two +// multi-disc rips, whose leaves are "CD1" and "CD2" — are written into +// one directory, and collect finds files by name there. Run at once, +// a file one peer never sent is filled by the other peer's file of the +// same name. One grab per peer made that impossible; several peers at +// once makes it likely. It is package-level and keyed on the full +// path because two configured clients can share one daemon. +var slskdFolders keyedLock[string] + +// localFolders returns the directories under downloadsPath a candidate's +// files will be written to, sorted so every grab takes them in the same +// order and two cannot each hold what the other waits for. +func (s *slskd) localFolders(c Candidate) []string { + var out []string + + for _, f := range c.Files { + norm := strings.ReplaceAll(f.Path, `\`, "/") + out = append(out, filepath.Join(s.downloadsPath, path.Base(path.Dir(norm)))) + } + + slices.Sort(out) + + return slices.Compact(out) +} + +// lockSlskdFolders takes every folder in order, releasing what it holds +// if the context ends part way. +func lockSlskdFolders(ctx context.Context, folders []string) (func(), error) { + releases := make([]func(), 0, len(folders)) + + releaseAll := func() { + for _, r := range slices.Backward(releases) { + r() + } + } + + for _, f := range folders { + r, err := slskdFolders.acquire(ctx, f) + if err != nil { + releaseAll() + + return nil, err + } + + releases = append(releases, r) + } + + return releaseAll, nil +} + // slskdDownloadsPath is the transfers endpoint for one peer. Soulseek // usernames may contain spaces and punctuation, so the name is escaped // rather than spliced into the path.