wip

Author: cunla

scheduler/worker/commands/kill_worker.py

@@ -19,6 +19,7 @@ def __init__(self, *args: Any, **kwargs: Any) -> None:
 
     def process_command(self, connection: ConnectionType) -> None:
         from scheduler.worker import Worker
+
         logger.info(f"Received kill-worker command for {self.worker_name}")
         worker_model = WorkerModel.get(self.worker_name, connection)
         if worker_model is None or worker_model.pid is None:

scheduler/worker/scheduler.py

@@ -32,11 +32,11 @@ def _reschedule_tasks() -> None:
 
 class WorkerScheduler:
     def __init__(
-            self,
-            queues: Sequence[Queue],
-            connection: ConnectionType,
-            worker_name: str,
-            interval: Optional[int] = None,
+        self,
+        queues: Sequence[Queue],
+        connection: ConnectionType,
+        worker_name: str,
+        interval: Optional[int] = None,
     ) -> None:
         interval = interval or SCHEDULER_CONFIG.SCHEDULER_INTERVAL
         self._queues = queues

scheduler/worker/worker.py

@@ -99,18 +99,18 @@ def from_model(cls, model: WorkerModel) -> Self:
         return res
 
     def __init__(
-            self,
-            queues: Iterable[Union[str, Queue]],
-            name: str,
-            connection: ConnectionType,
-            maintenance_interval: int = SCHEDULER_CONFIG.DEFAULT_MAINTENANCE_TASK_INTERVAL,
-            job_monitoring_interval: int = SCHEDULER_CONFIG.DEFAULT_JOB_MONITORING_INTERVAL,
-            dequeue_strategy: DequeueStrategy = DequeueStrategy.DEFAULT,
-            disable_default_exception_handler: bool = False,
-            fork_job_execution: bool = True,
-            with_scheduler: bool = True,
-            burst: bool = False,
-            model: Optional[WorkerModel] = None,
+        self,
+        queues: Iterable[Union[str, Queue]],
+        name: str,
+        connection: ConnectionType,
+        maintenance_interval: int = SCHEDULER_CONFIG.DEFAULT_MAINTENANCE_TASK_INTERVAL,
+        job_monitoring_interval: int = SCHEDULER_CONFIG.DEFAULT_JOB_MONITORING_INTERVAL,
+        dequeue_strategy: DequeueStrategy = DequeueStrategy.DEFAULT,
+        disable_default_exception_handler: bool = False,
+        fork_job_execution: bool = True,
+        with_scheduler: bool = True,
+        burst: bool = False,
+        model: Optional[WorkerModel] = None,
     ) -> None:
         self.fork_job_execution = fork_job_execution
         self.job_monitoring_interval: int = job_monitoring_interval
@@ -376,7 +376,7 @@ def run_maintenance_tasks(self) -> None:
             self._model.save(connection=self.connection)
 
     def dequeue_job_and_maintain_ttl(
-            self, timeout: Optional[int], max_idle_time: Optional[int] = None
+        self, timeout: Optional[int], max_idle_time: Optional[int] = None
     ) -> Tuple[Optional[JobModel], Optional[Queue]]:
         """Dequeues a job while maintaining the TTL.
         :param timeout: The timeout for the dequeue operation.
@@ -551,7 +551,7 @@ def reorder_queues(self, reference_queue: Queue) -> None:
             return
         if self._dequeue_strategy == DequeueStrategy.ROUND_ROBIN:
             pos = self._ordered_queues.index(reference_queue)
-            self._ordered_queues = self._ordered_queues[pos + 1:] + self._ordered_queues[: pos + 1]
+            self._ordered_queues = self._ordered_queues[pos + 1 :] + self._ordered_queues[: pos + 1]
             return
         if self._dequeue_strategy == DequeueStrategy.RANDOM:
             shuffle(self._ordered_queues)
@@ -635,7 +635,7 @@ def monitor_job_execution_process(self, job: JobModel, queue: Queue) -> None:
         while True:
             try:
                 with SCHEDULER_CONFIG.DEATH_PENALTY_CLASS(
-                        self.job_monitoring_interval, JobExecutionMonitorTimeoutException
+                    self.job_monitoring_interval, JobExecutionMonitorTimeoutException
                 ):
                     retpid, ret_val = self.wait_for_job_execution_process()
                 break
@@ -878,7 +878,7 @@ class RoundRobinWorker(Worker):
 
     def reorder_queues(self, reference_queue: Queue) -> None:
         pos = self._ordered_queues.index(reference_queue)
-        self._ordered_queues = self._ordered_queues[pos + 1:] + self._ordered_queues[: pos + 1]
+        self._ordered_queues = self._ordered_queues[pos + 1 :] + self._ordered_queues[: pos + 1]
 
 
 class RandomWorker(Worker):