diff --git a/django_q/cluster.py b/django_q/cluster.py index ef5d05a..89e884e 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -1,15 +1,12 @@ # Standard -import ast -from queue import Empty -from django_q.worker import Pool, Worker -import pydoc +from django_q.scheduler import Scheduler +from django_q.puller import Puller +from django_q.worker import Pool import signal from django_q.monitor import Monitor import socket -import traceback import uuid -from datetime import datetime, timedelta -from multiprocessing import Event, Process, Value, current_process +from multiprocessing import Event, Process, 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 - while not self.pool.task_queue.empty(): + # make sure the tasks queue in the pool is empty and workers are idle max timeout 20 sec + 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): diff --git a/django_q/monitor.py b/django_q/monitor.py index f5c96ab..45ae3f2 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -1,13 +1,11 @@ -import multiprocessing -from multiprocessing.queues import Queue +from queue 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.status.value = self.Status.IDLE.value + self.task_queue = Queue() + + @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}) diff --git a/django_q/process_manager.py b/django_q/process_manager.py index 80e49b5..8ca047c 100644 --- a/django_q/process_manager.py +++ b/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) diff --git a/django_q/scheduler.py b/django_q/scheduler.py index 6fffbf8..67e852e 100644 --- a/django_q/scheduler.py +++ b/django_q/scheduler.py @@ -1,23 +1,17 @@ -from utils import localtime -from models import Schedule -from djang_q import tasks +from django_q.utils import localtime +from django_q.models import Schedule +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.exceptions import TimeoutException, TimeoutHandler +from django_q.conf import Conf, logger 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) - diff --git a/django_q/worker.py b/django_q/worker.py index caaaaea..634f09c 100644 --- a/django_q/worker.py +++ b/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)