fix(worker): parallelize per-source searches in the match phase
Soulseek's slskd search polls up to ~2 min; searching adapters sequentially added that on top of Qobuz's instant search. Run the per-adapter searches concurrently via a ThreadPoolExecutor so total match time is the slowest single source, not their sum. Only Qobuz drives streamrip's shared asyncio loop and there's one Qobuz adapter, so no two threads use that loop at once. Results collected in adapter order (map preserves order) for deterministic candidate ordering; per-adapter failures still swallowed (down source = no candidates). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -2,6 +2,7 @@ import os
|
||||
import shutil
|
||||
import threading
|
||||
import time
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import replace
|
||||
from typing import Callable, Sequence
|
||||
|
||||
@@ -94,6 +95,16 @@ def _poll_download_progress(job_id: str, staging: str, expected: int, stop: "thr
|
||||
conn.close()
|
||||
|
||||
|
||||
def _search_adapter(adapter: SourceAdapter, target: MBTarget) -> list[Candidate]:
|
||||
"""Search one adapter, swallowing failures — a down source contributes no candidates rather
|
||||
than crashing the job. Runs on a worker thread (searches are parallelized across sources)."""
|
||||
try:
|
||||
return list(adapter.search(target))
|
||||
except Exception as e:
|
||||
print(f"pipeline: adapter {adapter.name} search failed: {e}", flush=True)
|
||||
return []
|
||||
|
||||
|
||||
def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) -> None:
|
||||
with conn.cursor() as cur:
|
||||
cur.execute(
|
||||
@@ -306,15 +317,16 @@ def run_pipeline(
|
||||
conn.commit()
|
||||
return
|
||||
|
||||
# 2. match — search all adapters; score + persist EVERY candidate found
|
||||
# 2. match — search all adapters CONCURRENTLY (Soulseek's search polls up to ~2 min; run it
|
||||
# alongside Qobuz's instant search so total match time = the slowest source, not their sum).
|
||||
# Only Qobuz touches streamrip's shared asyncio loop and there's a single Qobuz adapter, so no
|
||||
# two threads drive that loop at once. Results are collected in adapter order (map preserves
|
||||
# input order) so candidate ordering stays deterministic.
|
||||
_set_state(conn, job_id, "matching", "match")
|
||||
found: list[Candidate] = []
|
||||
for adapter in adapters:
|
||||
try:
|
||||
results = adapter.search(target)
|
||||
except Exception as e: # a down source contributes no candidates, never crashes the job
|
||||
print(f"pipeline: adapter {adapter.name} search failed: {e}", flush=True)
|
||||
continue
|
||||
with ThreadPoolExecutor(max_workers=max(1, len(adapters))) as pool:
|
||||
per_adapter = list(pool.map(lambda a: (a, _search_adapter(a, target)), adapters))
|
||||
for adapter, results in per_adapter:
|
||||
for c in results:
|
||||
found.append(replace(c, source_tier=adapter.tier, confidence=score_confidence(target, c)))
|
||||
_persist_candidates(conn, job_id, found)
|
||||
|
||||
Reference in New Issue
Block a user