murdock: drop fasttrack_queue, use PriorityQueue

Author: kaspar030

murdock/murdock.py

@@ -1,10 +1,11 @@
+from dataclasses import dataclass, field
 import asyncio
 import json
 import os
 import re
 from datetime import datetime, timezone
 
-from typing import List, Optional, Union
+from typing import Any, List, Optional, Union
 
 import websockets
 from fastapi import WebSocket
@@ -46,6 +47,14 @@
     "reopened",
 ]
 
+JOB_FASTTRACK_BONUS = 100
+
+
+@dataclass(order=True)
+class PrioritizedItem:
+    priority: int
+    item: Any = field(compare=False)
+
 
 class Murdock:
     def __init__(
@@ -70,8 +79,7 @@ def __init__(
         self.num_workers: int = num_workers
         self.queued: MurdockJobList = MurdockJobList()
         self.running: MurdockJobPool = MurdockJobPool(num_workers)
-        self.queue: asyncio.Queue = asyncio.Queue()
-        self.fasttrack_queue: asyncio.Queue = asyncio.Queue()
+        self.queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
         self.db = Database()
         self.notifier = Notifier()
         self.instrumentator = Instrumentator()
@@ -128,18 +136,14 @@ async def _process_job(self, job: MurdockJob):
     async def job_processing_task(self):
         current_task = asyncio.current_task().get_name()
         while True:
-            if self.fasttrack_queue.qsize():
-                job = self.fasttrack_queue.get_nowait()
+            try:
+                prioritized_job = await self.queue.get()
+                job = prioritized_job.item
                 await self._process_job(job)
-                self.fasttrack_queue.task_done()
-            else:
-                try:
-                    job = await self.queue.get()
-                    await self._process_job(job)
-                    self.queue.task_done()
-                except RuntimeError as exc:
-                    LOGGER.info(f"Exiting worker {current_task}: {exc}")
-                    break
+                self.queue.task_done()
+            except RuntimeError as exc:
+                LOGGER.info(f"Exiting worker {current_task}: {exc}")
+                break
 
     async def job_prepare(self, job: MurdockJob):
         self.queued.remove(job)
@@ -210,13 +214,17 @@ async def add_job_to_queue(self, job: MurdockJob):
                 "target_url": self.base_url,
             },
         )
-        all_busy = all(running is not None for running in self.running.jobs)
         self.queued.add(job)
         job.state = "queued"
-        if all_busy and job.fasttracked:
-            self.fasttrack_queue.put_nowait(job)
+        if job.fasttracked:
+            prio = JOB_FASTTRACK_BONUS
         else:
-            self.queue.put_nowait(job)
+            prio = 0
+
+        # PriorityQueue is a min_queue, but "higher priority equals higher value"
+        # makes more sense. So, invert the priority by multiplying with `-1`
+        self.queue.put_nowait(PrioritizedItem(prio * -1, job))
+
         LOGGER.info(f"{job} added to queued jobs")
         await self.reload_jobs()