Fixing scheduler

django_q/cluster.py

@@ -1,15 +1,12 @@
 # Standard
-import ast
+from django_q.scheduler import Scheduler
-from queue import Empty
+from django_q.puller import Puller
-from django_q.worker import Pool, Worker
+from django_q.worker import Pool
-import pydoc
 import signal
 from django_q.monitor import Monitor
 import socket
-import traceback
 import uuid
-from datetime import datetime, timedelta
+from multiprocessing import Event, Process, current_process
-from multiprocessing import Event, Process, Value, current_process
 from time import sleep
 
 # Django
@@ -31,13 +28,10 @@ import django_q.tasks
 from django_q.brokers import Broker, get_broker
 from django_q.conf import (
     Conf,
-    croniter,
-    error_reporter,
     get_ppid,
     logger,
     psutil,
     setproctitle,
-    resource,
 )
 from django_q.humanhash import humanize
 from django_q.status import Stat, Status
@@ -152,16 +146,11 @@ class Sentinel:
         self.tob = timezone.now()
         self.stop_event = stop_event
         self.start_event = start_event
-        self.pool = []
         self.timeout = timeout
         self.event_out = Event()
         logger.info(
             _("%(name)s main at %(id)s") % {"name": self.name, "id": current_process().pid}
         )
-        from django_q.puller import Puller
-        self.puller = Puller()
-
-        self.monitor = Monitor()
         if start:
             self.start()
 
@@ -174,11 +163,11 @@ class Sentinel:
         if not self.start_event.is_set() and not self.stop_event.is_set():
             return Conf.STARTING
         elif self.start_event.is_set() and not self.stop_event.is_set():
-            if self.result_queue.empty() and self.task_queue.empty():
+            if self.monitor.is_idle and self.pool.is_done:
                 return Conf.IDLE
             return Conf.WORKING
         elif self.stop_event.is_set() and self.start_event.is_set():
-            if self.monitor.is_alive() or self.puller.is_alive() or len(self.pool) > 0:
+            if self.monitor.is_alive or self.puller.is_alive or len(self.pool.workers) > 0:
                 return Conf.STOPPING
             return Conf.STOPPED
 
@@ -189,6 +178,9 @@ class Sentinel:
             db.connections.close_all()
         # spawn worker pool
         self.pool = Pool()
+        self.puller = Puller()
+        self.monitor = Monitor()
+        self.scheduler = Scheduler()
         # set worker cpu affinity if needed
         if psutil and Conf.CPU_AFFINITY:
             set_cpu_affinity(Conf.CPU_AFFINITY, [w.process.process_id for w in self.pool.workers])
@@ -210,8 +202,6 @@ class Sentinel:
         counter = 0
         # Guard loop. Runs at least once
         while not self.stop_event.is_set() or not counter:
-            logger.info("is set")
-            logger.info(self.stop_event.is_set())
             # Check if the pool of workers is healthy
             logger.info("Check if pool is healthy")
             if not self.pool.is_healthy:
@@ -227,6 +217,11 @@ class Sentinel:
             if not self.monitor.is_alive:
                 self.monitor.reincarnate_process()
 
+            print("Check if scheduler is healthy")
+            if not self.scheduler.is_alive:
+                self.scheduler.reincarnate_process()
+
+
             print("add tasks and mark workers idle")
             for worker in self.pool.get_done_workers():
                 # put result in task_queue to be picked up by monitor for processing
@@ -250,8 +245,8 @@ class Sentinel:
             self.pool.delegate_tasks()
 
             logger.info("sleep")
-            sleep(Conf.GUARD_CYCLE)
             counter += 1
+            sleep(Conf.GUARD_CYCLE)
         self.stop()
 
     def stop(self):
@@ -259,27 +254,36 @@ class Sentinel:
         logger.info(_("%(name)s stopping cluster processes") % {"name": name})
         # Stopping guard
         self.stop_event.set()
-        logger.info(_("Guard has stopped"))
+        logger.debug(_("Guard has stopped"))
+        # Stop scheduler
+        self.scheduler.stop_scheduler()
 
         # End all workers gracefully
         for __ in range(Conf.WORKERS):
             self.pool.add_task("STOP")
 
-        # manually loop through the tasks
+        # make sure the tasks queue in the pool is empty and workers are idle max timeout 20 sec
-        while not self.pool.task_queue.empty():
+        time_passed = 0
+        while not self.pool.is_done and time_passed <= 20:
             self.monitor.run_item()
             self.pool.delegate_tasks()
+            time_passed += 0.5
             sleep(0.5)
+        if time_passed >= 20:
+            logger.error(_("Couldn't terminate tasks within 20 seconds, killing processes now"))
+            for worker in self.pool.workers:
+                worker.process.kill()
 
-        logger.info(_("All tasks were processed and workers where stopped"))
+
+        logger.debug(_("All tasks were processed and workers where stopped"))
 
         self.monitor.add_task("STOP")
-        while not self.monitor.task_queue.empty():
+        while not self.monitor.is_done:
             # in the case the monitor was behind, let's run through all
             self.monitor.run_item()
             sleep(0.5)
 
-        logger.info(_("All tasks were saved"))
+        logger.debug(_("All tasks were saved"))
 
         self.puller.stop_puller()
 
@@ -289,8 +293,9 @@ class Sentinel:
 
         self.monitor.process.terminate()
         self.puller.process.terminate()
+        self.scheduler.process.terminate()
 
-        logger.info(_("All processes were terminated"))
+        logger.debug(_("All processes were terminated"))
 
 
 def set_cpu_affinity(n: int, process_ids: list, actual: bool = not Conf.TESTING):

django_q/monitor.py

@@ -1,13 +1,11 @@
-import multiprocessing
+from queue import Queue
-from multiprocessing.queues import Queue
 from queue import Empty
-from django_q.queue_task import QueueTask
 from django_q.brokers import get_broker
 from django_q.process_manager import ProcessManager
 from django_q.signals import post_execute
-from django_q.conf import Conf, logger, error_reporter, resource
+from django_q.conf import logger
 from django.utils.translation import gettext_lazy as _
-from multiprocessing import Event, Process, Value, current_process
+from multiprocessing import current_process
 
 try:
     import setproctitle
@@ -18,8 +16,11 @@ except ModuleNotFoundError:
 class Monitor(ProcessManager):
     def __init__(self):
         super().__init__()
-        self.task_queue = Queue(ctx=multiprocessing.get_context())
+        self.task_queue = Queue()
-        self.status.value = self.Status.IDLE.value
+
+    @property
+    def is_done(self):
+        return self.status.value == self.Status.IDLE.value and self.task_queue.empty()
 
     def get_target(self):
         return self.run_monitor
@@ -44,13 +45,14 @@ class Monitor(ProcessManager):
         logger.info(
             _("%(name)s monitoring at %(id)s") % {"name": proc_name, "id": current_process().pid}
         )
+        status.value = self.Status.IDLE.value
 
         while True:
             task = pipe.recv()
-            status.value = self.Status.BUSY
             if task == "STOP":
                 logger.info(f"Monitor {proc_name} shut down")
                 break
+            status.value = self.Status.BUSY.value
             # save the result
             if task.cached:
                 task.save_cached(broker)
@@ -78,6 +80,6 @@ class Monitor(ProcessManager):
                         "task_result": task.result_payload,
                     }
                 )
-            status.value = self.Status.IDLE
+            status.value = self.Status.IDLE.value
         logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name})
 

django_q/process_manager.py

@@ -18,7 +18,6 @@ class ProcessManager(ABC):
         RECYCLE = 4
 
     target = None
-    status = Value("i", Status.IDLE.value)
 
     def get_target(self) -> Callable:
         if self.target is None:
@@ -26,6 +25,7 @@ class ProcessManager(ABC):
         return self.target
 
     def __init__(self):
+        self.status = Value("i", self.Status.IDLE.value)
         self.process = self.spawn_process()
         self.name = humanize(uuid.uuid4().hex)
 

django_q/scheduler.py

@@ -1,23 +1,17 @@
-from utils import localtime
+from django_q.utils import localtime
-from models import Schedule
+from django_q.models import Schedule
-from djang_q import tasks
+from django_q import tasks
 import ast
 from django_q.humanhash import humanize
 from django import db
-from signing import BadSignature, SignedPackage
 from time import sleep
 from django_q.brokers import get_broker
-import multiprocessing
-from django_q.queue_task import QueueTask
 from django.utils import timezone
-import enum
-import traceback
 
-from multiprocessing import Event, Process, Value, current_process
+from multiprocessing import current_process
 from django_q.utils import close_old_django_connections
 from django.utils.translation import gettext_lazy as _
-from django_q.conf import Conf, logger, setproctitle, error_reporter, resource, psutil
+from django_q.conf import Conf, logger
-from django_q.exceptions import TimeoutException, TimeoutHandler
 from django_q.process_manager import ProcessManager
 
 
@@ -27,8 +21,22 @@ class Scheduler(ProcessManager):
     def get_target(self):
         return self.run_scheduler
 
+    def stop_scheduler(self) -> None:
+        # send task to worker
+        self.manager_pipe.send("STOP")
+
     def run_scheduler(self, status, pipe) -> None:
+        self.process_name = current_process().name
+        self.process_id = current_process().pid
+        status.value = self.Status.BUSY.value
+        logger.info(
+            _("%(proc_name)s scheduling at %(id)s")
+            % {"proc_name": self.process_name, "id": self.process_id}
+        )
         while True:
+            if pipe.poll() and pipe.recv() == "STOP":
+                status.value = self.Status.DONE.value
+                break
             broker = get_broker()
             close_old_django_connections()
             try:
@@ -125,4 +133,3 @@ class Scheduler(ProcessManager):
                 logger.exception("Could not create task from schedule")
             # sleep 60 seconds for next schedule
             sleep(60)
-

django_q/worker.py

@@ -1,5 +1,5 @@
 import multiprocessing
-from multiprocessing.queues import Queue
+from queue import Queue
 from queue import Empty
 from typing import Optional
 from django_q.queue_task import QueueTask
@@ -39,7 +39,7 @@ class Pool:
     def __init__(self, workers=Conf.WORKERS):
         self.amount_workers = workers
         self.workers = []
-        self.task_queue = Queue(ctx=multiprocessing.get_context())
+        self.task_queue = Queue()
         self.start_workers()
 
     def start_workers(self):
@@ -58,6 +58,16 @@ class Pool:
         """Checks if all workers are still operating"""
         return all(worker.is_alive for worker in self.workers)
 
+    @property
+    def is_idle(self):
+        """Checks if all workers are idle"""
+        return all(worker.is_idle for worker in self.workers)
+
+    @property
+    def is_done(self):
+        """Checks if all workers are idle and task queue is empty"""
+        return self.is_idle and self.task_queue.empty()
+
     def reincarnate_stopped_workers(self):
         """Reincarnates workers that are not alive anymore"""
         stopped_workers = [worker for worker in self.workers if not worker.is_alive]
@@ -153,7 +163,7 @@ class WorkerProcess(Process):
             # signal execution
             pre_execute.send(sender="django_q", func=task.func, task=task)
 
-            status.value = Worker.Status.BUSY.value
+            status.value = ProcessManager.Status.BUSY.value
             task.started_at = timezone.now()
             try:
                 with TimeoutHandler(timeout=task.timeout):
@@ -183,13 +193,13 @@ class WorkerProcess(Process):
             self.task_count += 1
 
             # Set to DONE so main process can pick it up
-            status.value = Worker.Status.DONE.value
+            status.value = ProcessManager.Status.DONE.value
             if setproctitle:
                 setproctitle.setproctitle(f"qcluster {self.process_name} completed with task")
 
             # Recreate a new process if this task has had the max amount of runs or exceeded resources
             if self.task_count == Conf.RECYCLE or self.rss_check():
-                status.value = Worker.Status.RECYCLE
+                status.value = ProcessManager.Status.RECYCLE
                 break
 
             pipe.send(task)