Run transfer queue jobs concurrently (max 3) so downloads don't block each other

app/gui.py

@@ -1831,6 +1831,8 @@ class StorePage(QWidget):
         self._lib_cover = {}  # norm -> (release_id, system)
         self._psp_norm = set()   # norm of library items installed on the PSP
         self._cover_attempted = set()
+        self._cover_busy = False
+        self._date_busy = False
         self._generation = 0
         self._scan_gen = 0
         self._meta_gen = 0
@@ -2017,9 +2019,27 @@ class StorePage(QWidget):
         if self._items:
             self._apply_filter()
 
+    def _begin_scan(self):
+        """Clear the stale view and invalidate in-flight cover/date batches."""
+        self._generation += 1
+        self._cover_busy = False
+        self._date_busy = False
+        self._cover_queue = []
+        self._cover_next = 0
+        self._date_queue = []
+        self._date_next = 0
+        self._cover_attempted.clear()
+        for loader in self._loaders:
+            loader.cancel()
+        self._loaders = []
+        self.grid.clear()
+        self.table.setRowCount(0)
+        self.status.setText("Loading…")
+
     def refresh(self, force=False):
         if self._closing:
             return
+        self._begin_scan()
         self._scan_gen += 1
         gen = self._scan_gen
         if self.source_combo.currentIndex() == 0:  # Server
@@ -2080,6 +2100,8 @@ class StorePage(QWidget):
         self._items = []
         self._filtered = []
         self._dirty = False
+        self.grid.clear()
+        self.table.setRowCount(0)
         log.warning("store unavailable: %s", message)
         self.status.setText(f"Store unavailable: {message}")
 
@@ -2180,6 +2202,8 @@ class StorePage(QWidget):
         self._cover_next = 0
         self._date_queue = []
         self._date_next = 0
+        self._cover_busy = False
+        self._date_busy = False
         self._cover_attempted.clear()
         self._placeholder = QIcon(_placeholder_pixmap(self.ICON_W, self.ICON_H))
 
@@ -2189,6 +2213,7 @@ class StorePage(QWidget):
         self._loaders = []
 
         self.grid.clear()
+        self.grid.setUpdatesEnabled(False)
         self.table.setUpdatesEnabled(False)
         self.table.setSortingEnabled(False)
         self.table.setRowCount(show_count)
@@ -2293,6 +2318,7 @@ class StorePage(QWidget):
     def _finish_fill(self):
         if self._fill_timer is not None:
             self._fill_timer.stop()
+        self.grid.setUpdatesEnabled(True)
         self.table.setUpdatesEnabled(True)
         self.table.setSortingEnabled(True)
         total = len(self._filtered)
@@ -2305,13 +2331,14 @@ class StorePage(QWidget):
         self._load_date_batch(self.DATE_BATCH)
 
     def _load_cover_batch(self, n):
-        if self._closing or not self._cover_queue:
+        if self._closing or not self._cover_queue or self._cover_busy:
             return
         end = min(self._cover_next + n, len(self._cover_queue))
         if self._cover_next >= end:
             return
         tasks = self._cover_queue[self._cover_next:end]
         self._cover_next = end
+        self._cover_busy = True
         self._start_cover_loader(self._generation, tasks)
 
     def _active_scrollbar(self):
@@ -2342,13 +2369,14 @@ class StorePage(QWidget):
             self._fill_timer.start()
 
     def _load_date_batch(self, n):
-        if self._closing or not self._date_queue:
+        if self._closing or not self._date_queue or self._date_busy:
             return
         end = min(self._date_next + n, len(self._date_queue))
         if self._date_next >= end:
             return
         tasks = self._date_queue[self._date_next:end]
         self._date_next = end
+        self._date_busy = True
         self._start_date_loader(self._generation, tasks)
 
     def _start_date_loader(self, gen, tasks):
@@ -2359,12 +2387,22 @@ class StorePage(QWidget):
         loader.moveToThread(thread)
         thread.started.connect(loader.run)
         loader.loaded.connect(self._on_date_loaded)
+        loader.finished.connect(self._on_date_batch_done)
         loader.finished.connect(thread.quit)
         thread.finished.connect(self._on_thread_finished)
         self._threads.append(thread)
         self._loaders.append(loader)
         thread.start()
 
+    def _on_date_batch_done(self):
+        if self._closing:
+            return
+        loader = self.sender()
+        if getattr(loader, "gen", None) != self._generation:
+            return
+        self._date_busy = False
+        self._load_date_batch(self.DATE_BATCH)
+
     def _on_date_loaded(self, gen, row, date):
         if self._closing or gen != self._generation or not date:
             return
@@ -2382,12 +2420,22 @@ class StorePage(QWidget):
         loader.moveToThread(thread)
         thread.started.connect(loader.run)
         loader.loaded.connect(self._on_cover_loaded)
+        loader.finished.connect(self._on_cover_batch_done)
         loader.finished.connect(thread.quit)
         thread.finished.connect(self._on_thread_finished)
         self._threads.append(thread)
         self._loaders.append(loader)
         thread.start()
 
+    def _on_cover_batch_done(self):
+        if self._closing:
+            return
+        loader = self.sender()
+        if getattr(loader, "gen", None) != self._generation:
+            return
+        self._cover_busy = False
+        self._load_cover_batch(self.COVER_SCROLL_BATCH)
+
     def _on_cover_loaded(self, gen, row, image):
         if self._closing or gen != self._generation:
             return
@@ -2622,6 +2670,7 @@ class QueuePage(QWidget):
         "umd-dump": "UMD dump",
         "save-sync": "Save sync",
         "nps-download": "NoPayStation",
+        "covers": "Covers bundle",
     }
 
     def __init__(self, parent=None):
@@ -3761,35 +3810,6 @@ class LibraryWorker(QObject):
         self.finished.emit()
 
 
-class CoversBundleWorker(QObject):
-    """Download + extract the offline cover bundle off the GUI thread."""
-    progress = Signal(int, int)   # done bytes, total bytes (0 = unknown)
-    succeeded = Signal()
-    failed = Signal(str)
-    finished = Signal()
-
-    def __init__(self):
-        super().__init__()
-        self._cancel = threading.Event()
-
-    def cancel(self):
-        self._cancel.set()
-
-    def run(self):
-        try:
-            covers.download_covers_bundle(
-                progress_cb=lambda d, t: self.progress.emit(d, t),
-                cancel_event=self._cancel)
-        except InterruptedError:
-            pass
-        except Exception as exc:  # noqa: BLE001
-            self.failed.emit(str(exc))
-        else:
-            self.succeeded.emit()
-        finally:
-            self.finished.emit()
-
-
 class SaveSyncPlanWorker(QObject):
     """Compute the save-sync plan off the GUI thread (content hashing is I/O-heavy)."""
     plan = Signal(list, bool)  # plan, switch_to_queue
@@ -5386,6 +5406,9 @@ class MainWindow(QMainWindow):
             QTimer.singleShot(0, self._refresh_psp)
         if kind == "save-sync":
             QTimer.singleShot(0, self._refresh_saves)
+        if kind == "covers":
+            covers.clear_miss_cache()
+            QTimer.singleShot(0, self.load_library)
         if kind == "ps1-convert":
             QTimer.singleShot(0, self.ps1_page.refresh)
             QTimer.singleShot(0, self.load_library)
@@ -5748,51 +5771,20 @@ class MainWindow(QMainWindow):
             self._download_covers_bundle()
 
     def _download_covers_bundle(self):
-        """Download + extract the offline cover bundle with a progress dialog."""
+        """Queue the offline cover bundle download in the transfer queue."""
-        if self._closing or getattr(self, "_bundle_worker", None) is not None:
+        if self._closing:
             return
-        prog = QProgressDialog("Downloading cover bundle…", "Cancel", 0, 0, self)
+        for j in self.queue.jobs:
-        prog.setWindowTitle("Offline covers")
+            if j.kind == "covers" and j.status in ("queued", "running", "cancelling"):
-        prog.setWindowModality(Qt.WindowModal)
+                return  # already queued or running
-        prog.setMinimumDuration(0)
+        self.queue.enqueue(
-        prog.setAutoClose(False)
+            transfers.CoversBundleSource(),
-        prog.setAutoReset(False)
+            str(config.COVERS_DIR),
-
+            None,
-        result = {"ok": False, "err": ""}
+            "Offline cover bundle",
-        thread = QThread(self)
+            kind="covers",
-        worker = CoversBundleWorker()
+        )
-        worker.moveToThread(thread)
+        self.tabs.setCurrentWidget(self.queue_page)
-        thread.started.connect(worker.run)
-        worker.progress.connect(lambda d, t: self._on_bundle_progress(prog, d, t))
-        worker.succeeded.connect(lambda: result.update(ok=True))
-        worker.failed.connect(lambda e: result.update(err=str(e)))
-        prog.canceled.connect(worker.cancel)
-        worker.finished.connect(thread.quit)
-        thread.finished.connect(lambda: self._on_bundle_done(prog, result, worker, thread))
-        self._bundle_worker = worker
-        self._bundle_thread = thread
-        thread.start()
-
-    @staticmethod
-    def _on_bundle_progress(prog, done, total):
-        if total > 0:
-            prog.setRange(0, total)
-            prog.setValue(done)
-            prog.setLabelText(
-                f"Downloading cover bundle… {done / (1 << 20):.0f} / {total / (1 << 20):.0f} MB")
-        else:
-            prog.setRange(0, 0)
-
-    def _on_bundle_done(self, prog, result, worker, thread):
-        prog.close()
-        self._bundle_worker = None
-        self._bundle_thread = None
-        if result["ok"]:
-            QMessageBox.information(self, "Offline covers", "Cover bundle installed.")
-            covers.clear_miss_cache()
-            self.load_library()
-        elif result["err"]:
-            QMessageBox.warning(self, "Offline covers", f"Download failed:\n{result['err']}")
 
     def closeEvent(self, event):
         self._closing = True
@@ -5802,11 +5794,6 @@ class MainWindow(QMainWindow):
                 self.worker.choose(None)
             self.worker.cancel()
         self._thread_cleanup()
-        if getattr(self, "_bundle_worker", None) is not None:
-            self._bundle_worker.cancel()
-            if getattr(self, "_bundle_thread", None) is not None and _thread_running(self._bundle_thread):
-                self._bundle_thread.quit()
-                self._bundle_thread.wait()
         self.cover_grid.shutdown()
         for t in self._meta_threads:
             if _thread_running(t):

app/queue.py

@@ -79,13 +79,13 @@ class TransferQueue(QObject):
     job_finished = Signal(dict)
     job_removed = Signal(int)
 
-    def __init__(self, parent=None):
+    def __init__(self, parent=None, max_concurrent=3):
         super().__init__(parent)
+        self.max_concurrent = max(1, int(max_concurrent))
         self.jobs = []
         self._pending = []
-        self._current = None
+        # job_id -> (QThread, JobWorker) for jobs currently running in parallel.
-        self._thread = None
+        self._running = {}
-        self._worker = None
         # Keep strong references to every worker/thread for their full lifetime.
         # Dropping a Python ref to a QObject that was moved to another thread
         # deletes it directly and crashes (bus error).
@@ -115,8 +115,10 @@ class TransferQueue(QObject):
             self.job_updated.emit(job.to_dict())
             self.job_finished.emit(job.to_dict())
             return
-        if job.status == "running" and self._worker is not None:
+        if job.status == "running":
-            self._worker.cancel()
+            pair = self._running.get(job_id)
+            if pair is not None:
+                pair[1].cancel()
             job.status = "cancelling"
             self.job_updated.emit(job.to_dict())
 
@@ -140,8 +142,8 @@ class TransferQueue(QObject):
             self.remove(jid)
 
     def shutdown(self):
-        if self._worker is not None:
+        for _t, worker in self._running.values():
-            self._worker.cancel()
+            worker.cancel()
         for t in self._threads:
             try:
                 if t.isRunning():
@@ -157,30 +159,29 @@ class TransferQueue(QObject):
         return None
 
     def _start_next(self):
-        if self._current is not None or not self._pending:
+        while len(self._running) < self.max_concurrent and self._pending:
-            return
             job = self._pending.pop(0)
-        self._current = job
             job.status = "running"
             log.info("starting job #%d: %s (%s)", job.id, job.title, job.kind)
             self.job_updated.emit(job.to_dict())
 
-        self._thread = QThread(self)
+            thread = QThread(self)
-        self._worker = JobWorker(job.source, job.dest_dir, job)
+            worker = JobWorker(job.source, job.dest_dir, job)
-        self._worker.moveToThread(self._thread)
+            worker.moveToThread(thread)
-        self._thread.started.connect(self._worker.run)
+            thread.started.connect(worker.run)
-        self._worker.progress.connect(self._on_progress)
+            worker.progress.connect(self._on_progress)
-        self._worker.finished.connect(self._on_finished)
+            worker.finished.connect(self._on_finished)
-        self._worker.skipped.connect(self._on_skipped)
+            worker.skipped.connect(self._on_skipped)
-        self._worker.cancelled.connect(self._on_cancelled)
+            worker.cancelled.connect(self._on_cancelled)
-        self._worker.failed.connect(self._on_failed)
+            worker.failed.connect(self._on_failed)
-        for sig in (self._worker.finished, self._worker.skipped,
+            for sig in (worker.finished, worker.skipped,
-                    self._worker.cancelled, self._worker.failed):
+                        worker.cancelled, worker.failed):
-            sig.connect(self._thread.quit)
+                sig.connect(thread.quit)
-        self._thread.finished.connect(self._on_thread_finished)
+            thread.finished.connect(self._on_thread_finished)
-        self._threads.append(self._thread)
+            self._running[job.id] = (thread, worker)
-        self._workers.append(self._worker)
+            self._threads.append(thread)
-        self._thread.start()
+            self._workers.append(worker)
+            thread.start()
 
     def _on_progress(self, job_id, stage, fraction, detail):
         job = self._find(job_id)
@@ -218,10 +219,12 @@ class TransferQueue(QObject):
             self.job_finished.emit(job.to_dict())
 
     def _on_thread_finished(self):
-        # Queued to the main thread once the worker thread has fully stopped,
+        # Queued to the main thread once a worker thread has fully stopped, so
-        # so it is now safe to drop the "current" pointers and advance.
+        # it is now safe to drop its pointers and advance the queue.
-        self._current = None
+        thread = self.sender()
-        self._thread = None
+        for jid, pair in list(self._running.items()):
-        self._worker = None
+            if pair[0] is thread:
+                del self._running[jid]
+                break
         self._start_next()
 

app/transfers.py

@@ -175,6 +175,33 @@ class Source:
         raise NotImplementedError
 
 
+class CoversBundleSource:
+    """Download the combined offline cover bundle (runs as a queue job)."""
+    label = "psppt-covers.zip"
+
+    def __init__(self, url=""):
+        self.url = url
+
+    def download(self, item, dest_dir, progress_cb=None, cancel_event=None):
+        from . import covers
+
+        def on_progress(done, total):
+            if progress_cb is None:
+                return
+            frac = (done / total) if total else 0.0
+            detail = (f"{done / (1 << 20):.0f} / {total / (1 << 20):.0f} MB"
+                      if total else "")
+            progress_cb("Downloading", frac, detail)
+
+        try:
+            result = covers.download_covers_bundle(self.url,
+                                                   progress_cb=on_progress,
+                                                   cancel_event=cancel_event)
+        except InterruptedError:
+            raise TransferCancelled() from None
+        return str(result)
+
+
 class FileCopySource(Source):
     """Copy a single local file (library -> PSP or PSP -> library).