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

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
-        self._thread = None
-        self._worker = None
+        # job_id -> (QThread, JobWorker) for jobs currently running in parallel.
+        self._running = {}
         # 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:
-            self._worker.cancel()
+        if job.status == "running":
+            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:
-            self._worker.cancel()
+        for _t, worker in self._running.values():
+            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:
-            return
+        while len(self._running) < self.max_concurrent and self._pending:
             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)
-        self._worker = JobWorker(job.source, job.dest_dir, job)
-        self._worker.moveToThread(self._thread)
-        self._thread.started.connect(self._worker.run)
-        self._worker.progress.connect(self._on_progress)
-        self._worker.finished.connect(self._on_finished)
-        self._worker.skipped.connect(self._on_skipped)
-        self._worker.cancelled.connect(self._on_cancelled)
-        self._worker.failed.connect(self._on_failed)
-        for sig in (self._worker.finished, self._worker.skipped,
-                    self._worker.cancelled, self._worker.failed):
-            sig.connect(self._thread.quit)
-        self._thread.finished.connect(self._on_thread_finished)
-        self._threads.append(self._thread)
-        self._workers.append(self._worker)
-        self._thread.start()
+            thread = QThread(self)
+            worker = JobWorker(job.source, job.dest_dir, job)
+            worker.moveToThread(thread)
+            thread.started.connect(worker.run)
+            worker.progress.connect(self._on_progress)
+            worker.finished.connect(self._on_finished)
+            worker.skipped.connect(self._on_skipped)
+            worker.cancelled.connect(self._on_cancelled)
+            worker.failed.connect(self._on_failed)
+            for sig in (worker.finished, worker.skipped,
+                        worker.cancelled, worker.failed):
+                sig.connect(thread.quit)
+            thread.finished.connect(self._on_thread_finished)
+            self._running[job.id] = (thread, worker)
+            self._threads.append(thread)
+            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,
-        # so it is now safe to drop the "current" pointers and advance.
-        self._current = None
-        self._thread = None
-        self._worker = None
+        # Queued to the main thread once a worker thread has fully stopped, so
+        # it is now safe to drop its pointers and advance the queue.
+        thread = self.sender()
+        for jid, pair in list(self._running.items()):
+            if pair[0] is thread:
+                del self._running[jid]
+                break
         self._start_next()