From a4a4e05dfeb0dee6f1e00844be3c83efdd9e5b7a Mon Sep 17 00:00:00 2001 From: GDay <1939656+GDay@users.noreply.github.com> Date: Thu, 23 Feb 2023 02:42:13 +0100 Subject: [PATCH] wip --- django_q/cluster.py | 675 +++++------------------------------- django_q/conf.py | 7 +- django_q/exceptions.py | 23 ++ django_q/models.py | 19 +- django_q/monitor.py | 567 ++++-------------------------- django_q/process_manager.py | 75 ++++ django_q/puller.py | 68 ++++ django_q/queue_task.py | 203 +++++++++++ django_q/queues.py | 82 ----- django_q/scheduler.py | 128 +++++++ django_q/status.py | 10 +- django_q/tasks.py | 135 ++++---- django_q/utils.py | 29 +- django_q/worker.py | 203 +++++++++++ 14 files changed, 956 insertions(+), 1268 deletions(-) create mode 100644 django_q/exceptions.py create mode 100644 django_q/process_manager.py create mode 100644 django_q/puller.py create mode 100644 django_q/queue_task.py delete mode 100644 django_q/queues.py create mode 100644 django_q/scheduler.py create mode 100644 django_q/worker.py diff --git a/django_q/cluster.py b/django_q/cluster.py index 13d7a2f..ef5d05a 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -1,7 +1,10 @@ # Standard import ast +from queue import Empty +from django_q.worker import Pool, Worker import pydoc import signal +from django_q.monitor import Monitor import socket import traceback import uuid @@ -37,15 +40,8 @@ from django_q.conf import ( resource, ) from django_q.humanhash import humanize -from django_q.models import Schedule, Success, Task -from django_q.queues import Queue -from django_q.signals import post_execute, pre_execute -from django_q.signing import BadSignature, SignedPackage from django_q.status import Stat, Status -from .utils import get_func_repr, localtime - - class Cluster: def __init__(self, broker: Broker = None): self.broker = broker or get_broker() @@ -156,16 +152,16 @@ class Sentinel: self.tob = timezone.now() self.stop_event = stop_event self.start_event = start_event - self.pool_size = Conf.WORKERS self.pool = [] self.timeout = timeout - self.task_queue = ( - Queue(maxsize=Conf.QUEUE_LIMIT) if Conf.QUEUE_LIMIT else Queue() - ) - self.result_queue = Queue() self.event_out = Event() - self.monitor = None - self.pusher = None + 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() @@ -182,105 +178,21 @@ class Sentinel: 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.pusher.is_alive() or len(self.pool) > 0: + if self.monitor.is_alive() or self.puller.is_alive() or len(self.pool) > 0: return Conf.STOPPING return Conf.STOPPED - def spawn_process(self, target, *args) -> Process: - """ - :type target: function or class - """ - p = Process(target=target, args=args) - p.daemon = True - if target == worker: - p.daemon = Conf.DAEMONIZE_WORKERS - p.timer = args[2] - self.pool.append(p) - p.start() - return p - - def spawn_pusher(self) -> Process: - return self.spawn_process(pusher, self.task_queue, self.event_out, self.broker) - - def spawn_worker(self): - self.spawn_process( - worker, self.task_queue, self.result_queue, Value("f", -1), self.timeout - ) - - def spawn_monitor(self) -> Process: - return self.spawn_process(monitor, self.result_queue, self.broker) - - def reincarnate(self, process): - """ - :param process: the process to reincarnate - :type process: Process or None - """ - # close connections before spawning new process - if not Conf.SYNC: - db.connections.close_all() - if process == self.monitor: - self.monitor = self.spawn_monitor() - logger.critical( - _("reincarnated monitor %(name)s after sudden death") - % {"name": process.name} - ) - elif process == self.pusher: - self.pusher = self.spawn_pusher() - logger.critical( - _("reincarnated pusher %(name)s after sudden death") - % {"name": process.name} - ) - else: - self.pool.remove(process) - self.spawn_worker() - if process.timer.value == 0: - # only need to terminate on timeout, otherwise we risk destabilizing - # the queues - task_name = "" - if psutil: - try: - process_name = psutil.Process(process.pid).name() - name_splits = process_name.split(" ") - task_name = name_splits[3] if len(name_splits) >= 4 and name_splits[2] == "processing" else "" - except psutil.NoSuchProcess: - pass - process.terminate() - if task_name: - msg = ( - _("reincarnated worker %(name)s after timeout while processing task %(task_name)s") - % {"name": process.name, "task_name": task_name} - ) - else: - msg = ( - _("reincarnated worker %(name)s after timeout") - % {"name": process.name} - ) - logger.critical(msg) - elif int(process.timer.value) == -2: - logger.info(_("recycled worker %(name)s") % {"name": process.name}) - else: - logger.critical( - _("reincarnated worker %(name)s after death") - % {"name": process.name} - ) - - self.reincarnations += 1 - def spawn_cluster(self): - self.pool = [] - Stat(self).save() + # Stat(self).save() # close connections before spawning new process if not Conf.SYNC: db.connections.close_all() # spawn worker pool - for __ in range(self.pool_size): - self.spawn_worker() - # spawn auxiliary - self.monitor = self.spawn_monitor() - self.pusher = self.spawn_pusher() + self.pool = Pool() # set worker cpu affinity if needed if psutil and Conf.CPU_AFFINITY: - set_cpu_affinity(Conf.CPU_AFFINITY, [w.pid for w in self.pool]) + set_cpu_affinity(Conf.CPU_AFFINITY, [w.process.process_id for w in self.pool.workers]) + def guard(self): logger.info( @@ -291,508 +203,94 @@ class Sentinel: } ) self.start_event.set() - Stat(self).save() logger.info( _("Q Cluster %(cluster_name)s running.") % {"cluster_name": humanize(self.cluster_id.hex)} ) counter = 0 - cycle = Conf.GUARD_CYCLE # guard loop sleep in seconds # Guard loop. Runs at least once while not self.stop_event.is_set() or not counter: - # Check Workers - for p in self.pool: - with p.timer.get_lock(): - # Are you alive? - if not p.is_alive() or p.timer.value == 0: - self.reincarnate(p) - continue - # Decrement timer if work is being done - if p.timer.value > 0: - p.timer.value -= cycle - # Check Monitor - if not self.monitor.is_alive(): - self.reincarnate(self.monitor) - # Check Pusher - if not self.pusher.is_alive(): - self.reincarnate(self.pusher) - # Call scheduler once a minute (or so) - counter += cycle - if counter >= 30 and Conf.SCHEDULER: - counter = 0 - scheduler(broker=self.broker) - # Save current status - Stat(self).save() - sleep(cycle) + 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: + # reincarnate workers that died + print("reincarnate workers") + self.pool.reincarnate_stopped_workers() + + print("Check if puller is healthy") + if not self.puller.is_alive: + self.puller.reincarnate_process() + + print("Check if monitor is healthy") + if not self.monitor.is_alive: + self.monitor.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 + self.monitor.add_task(worker.get_result()) + # mark task back to idle or reincarnate to be picked up for a new task + if worker.is_recycle: + worker.reincarnate_process() + else: + worker.mark_idle() + + # check if monitor has items to process + print("run monitor item") + self.monitor.run_item() + + print("Add task to worker pool") + if self.puller.has_results: + self.pool.add_task(self.puller.get_result()) + + # delegate tasks to workers that are now available + print("delegate tasks") + self.pool.delegate_tasks() + + logger.info("sleep") + sleep(Conf.GUARD_CYCLE) + counter += 1 self.stop() def stop(self): - Stat(self).save() name = current_process().name logger.info(_("%(name)s stopping cluster processes") % {"name": name}) - # Stopping pusher - self.event_out.set() - # Wait for it to stop - while self.pusher.is_alive(): - sleep(0.1) - Stat(self).save() - # Put poison pills in the queue - for __ in range(len(self.pool)): - self.task_queue.put("STOP") - self.task_queue.close() - # wait for the task queue to empty - self.task_queue.join_thread() - # Wait for all the workers to exit - while len(self.pool): - for p in self.pool: - if not p.is_alive(): - self.pool.remove(p) - sleep(0.1) - Stat(self).save() - # Finally stop the monitor - self.result_queue.put("STOP") - self.result_queue.close() - # Wait for the result queue to empty - self.result_queue.join_thread() - logger.info(_("%(name)s waiting for the monitor.") % {"name": name}) - # Wait for everything to close or time out - count = 0 - if not self.timeout: - self.timeout = 30 - while self.status() == Conf.STOPPING and count < self.timeout * 10: - sleep(0.1) - Stat(self).save() - count += 1 - # Final status - Stat(self).save() + # Stopping guard + self.stop_event.set() + logger.info(_("Guard has stopped")) + # End all workers gracefully + for __ in range(Conf.WORKERS): + self.pool.add_task("STOP") -def pusher(task_queue: Queue, event: Event, broker: Broker = None): - """ - Pulls tasks of the broker and puts them in the task queue - :type broker: - :type task_queue: multiprocessing.Queue - :type event: multiprocessing.Event - """ - if not broker: - broker = get_broker() - proc_name = current_process().name - if setproctitle: - setproctitle.setproctitle(f"qcluster {proc_name} pusher") - logger.info( - _("%(name)s pushing tasks at %(id)s") - % {"name": proc_name, "id": current_process().pid} - ) - while True: - try: - task_set = broker.dequeue() - except Exception: - logger.exception("Failed to pull task from broker") - # broker probably crashed. Let the sentinel handle it. - sleep(10) - break - if task_set: - for task in task_set: - ack_id = task[0] - # unpack the task - try: - task = SignedPackage.loads(task[1]) - except (TypeError, BadSignature): - logger.exception("Failed to push task to queue") - broker.fail(ack_id) - continue - task["ack_id"] = ack_id - task_queue.put(task) - logger.debug( - _("queueing from %(list_key)s") % {"list_key": broker.list_key} - ) - if event.is_set(): - break - logger.info(_("%(name)s stopped pushing tasks") % {"name": current_process().name}) + # manually loop through the tasks + while not self.pool.task_queue.empty(): + self.monitor.run_item() + self.pool.delegate_tasks() + sleep(0.5) + logger.info(_("All tasks were processed and workers where stopped")) -def monitor(result_queue: Queue, broker: Broker = None): - """ - Gets finished tasks from the result queue and saves them to Django - :type broker: brokers.Broker - :type result_queue: multiprocessing.Queue - """ - if not broker: - broker = get_broker() - proc_name = current_process().name - if setproctitle: - setproctitle.setproctitle(f"qcluster {proc_name} monitor") - logger.info( - _("%(name)s monitoring at %(id)s") % {"name": proc_name, "id": current_process().pid} - ) - for task in iter(result_queue.get, "STOP"): - # save the result - if task.get("cached", False): - save_cached(task, broker) - else: - save_task(task, broker) - # acknowledge result - ack_id = task.pop("ack_id", False) - if ack_id and (task["success"] or task.get("ack_failure", False)): - broker.acknowledge(ack_id) - # signal execution done - post_execute.send(sender="django_q", task=task) - # log the result - info_name = get_func_repr(task["func"]) - if task["success"]: - # log success - logger.info( - _("Processed '%(info_name)s' (%(task_name)s)") - % {"info_name": info_name, "task_name": task["name"]} - ) - else: - # log failure - logger.error( - _("Failed '%(info_name)s' (%(task_name)s) - %(task_result)s") - % { - "info_name": info_name, - "task_name": task["name"], - "task_result": task["result"], - } - ) - logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name}) + self.monitor.add_task("STOP") + while not self.monitor.task_queue.empty(): + # 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")) -def worker( - task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT -): - """ - Takes a task from the task queue, tries to execute it and puts the result back in - the result queue - :param timeout: number of seconds wait for a worker to finish. - :type task_queue: multiprocessing.Queue - :type result_queue: multiprocessing.Queue - :type timer: multiprocessing.Value - """ - proc_name = current_process().name - logger.info( - _("%(proc_name)s ready for work at %(id)s") - % {"proc_name": proc_name, "id": current_process().pid} - ) - if setproctitle: - setproctitle.setproctitle(f"qcluster {proc_name} idle") - task_count = 0 - if timeout is None: - timeout = -1 - # Start reading the task queue - for task in iter(task_queue.get, "STOP"): - result = None - timer.value = -1 # Idle - task_count += 1 - f = task["func"] + self.puller.stop_puller() - # Log task creation and set process name - # Get the function from the task - func_name = get_func_repr(f) - task_name = task["name"] - task_desc = ( - _("%(proc_name)s processing %(task_name)s '%(func_name)s'") - % { - "proc_name": proc_name, - "func_name": func_name, - "task_name": task_name, - } - ) - if "group" in task: - task_desc += f" [{task['group']}]" - logger.info(task_desc) + # make sure all processes are terminated + for worker in self.pool.workers: + worker.process.terminate() - if setproctitle: - proc_title = f"qcluster {proc_name} processing {task_name} '{func_name}'" - if "group" in task: - proc_title += f" [{task['group']}]" - setproctitle.setproctitle(proc_title) + self.monitor.process.terminate() + self.puller.process.terminate() - # if it's not an instance try to get it from the string - if not callable(f): - # locate() returns None if f cannot be loaded - f = pydoc.locate(f) - close_old_django_connections() - timer_value = task.pop("timeout", timeout) - # signal execution - pre_execute.send(sender="django_q", func=f, task=task) - # execute the payload - timer.value = timer_value # Busy - try: - if f is None: - # raise a meaningfull error if task["func"] is not a valid function - raise ValueError(f"Function {task['func']} is not defined") - res = f(*task["args"], **task["kwargs"]) - result = (res, True) - except Exception as e: - result = (f"{e} : {traceback.format_exc()}", False) - if error_reporter: - error_reporter.report() - if task.get("sync", False): - raise - with timer.get_lock(): - # Process result - task["result"] = result[0] - task["success"] = result[1] - task["stopped"] = timezone.now() - result_queue.put(task) - timer.value = -1 # Idle - if setproctitle: - setproctitle.setproctitle(f"qcluster {proc_name} idle") - # Recycle - if task_count == Conf.RECYCLE or rss_check(): - timer.value = -2 # Recycled - break - logger.info(_("%(proc_name)s stopped doing work") % {"proc_name": proc_name}) - - -def save_task(task, broker: Broker): - """ - Saves the task package to Django or the cache - :param task: the task package - :type broker: brokers.Broker - """ - # SAVE LIMIT < 0 : Don't save success - if not task.get("save", Conf.SAVE_LIMIT >= 0) and task["success"]: - return - # enqueues next in a chain - if task.get("chain", None): - django_q.tasks.async_chain( - task["chain"], - group=task["group"], - cached=task["cached"], - sync=task["sync"], - broker=broker, - ) - # SAVE LIMIT > 0: Prune database, SAVE_LIMIT 0: No pruning - close_old_django_connections() - - try: - filters = {} - if ( - Conf.SAVE_LIMIT_PER - and Conf.SAVE_LIMIT_PER in {"group", "name", "func"} - and Conf.SAVE_LIMIT_PER in task - ): - value = task[Conf.SAVE_LIMIT_PER] - if Conf.SAVE_LIMIT_PER == "func": - value = get_func_repr(value) - filters[Conf.SAVE_LIMIT_PER] = value - - with db.transaction.atomic(using=db.router.db_for_write(Success)): - last = Success.objects.filter(**filters).select_for_update().last() - if ( - task["success"] - and 0 < Conf.SAVE_LIMIT <= Success.objects.filter(**filters).count() - ): - last.delete() - - # check if this task has previous results - try: - existing_task = Task.objects.get(id=task["id"], name=task["name"]) - # only update the result if it hasn't succeeded yet - if not existing_task.success: - existing_task.stopped = task["stopped"] - existing_task.result = task["result"] - existing_task.success = task["success"] - existing_task.attempt_count = existing_task.attempt_count + 1 - existing_task.save() - - if ( - Conf.MAX_ATTEMPTS > 0 - and existing_task.attempt_count >= Conf.MAX_ATTEMPTS - ): - broker.acknowledge(task["ack_id"]) - - except Task.DoesNotExist: - # convert func to string - func = get_func_repr(task["func"]) - Task.objects.create( - id=task["id"], - name=task["name"], - func=func, - hook=task.get("hook"), - args=task["args"], - kwargs=task["kwargs"], - started=task["started"], - stopped=task["stopped"], - result=task["result"], - group=task.get("group"), - success=task["success"], - attempt_count=1, - ) - except Exception: - logger.exception("Could not save task result") - - -def save_cached(task, broker: Broker): - task_key = f'{broker.list_key}:{task["id"]}' - timeout = task["cached"] - if timeout is True: - timeout = None - try: - group = task.get("group", None) - iter_count = task.get("iter_count", 0) - # if it's a group append to the group list - if group: - group_key = f"{broker.list_key}:{group}:keys" - group_list = broker.cache.get(group_key) or [] - # if it's an iter group, check if we are ready - if iter_count and len(group_list) == iter_count - 1: - group_args = f"{broker.list_key}:{group}:args" - # collate the results into a Task result - results = [ - SignedPackage.loads(broker.cache.get(k))["result"] - for k in group_list - ] - results.append(task["result"]) - task["result"] = results - task["id"] = group - task["args"] = SignedPackage.loads(broker.cache.get(group_args)) - task.pop("iter_count", None) - task.pop("group", None) - if task.get("iter_cached", None): - task["cached"] = task.pop("iter_cached", None) - save_cached(task, broker=broker) - else: - save_task(task, broker) - broker.cache.delete_many(group_list) - broker.cache.delete_many([group_key, group_args]) - return - # save the group list - group_list.append(task_key) - broker.cache.set(group_key, group_list, timeout) - # async_task next in a chain - if task.get("chain", None): - django_q.tasks.async_chain( - task["chain"], - group=group, - cached=task["cached"], - sync=task["sync"], - broker=broker, - ) - # save the task - broker.cache.set(task_key, SignedPackage.dumps(task), timeout) - except Exception: - logger.exception("Could not save task result") - - -def scheduler(broker: Broker = None): - """ - Creates a task from a schedule at the scheduled time and schedules next run - """ - if not broker: - broker = get_broker() - close_old_django_connections() - try: - with db.transaction.atomic(using=db.router.db_for_write(Schedule)): - for s in ( - Schedule.objects.select_for_update() - .exclude(repeats=0) - .filter(next_run__lt=timezone.now()) - .filter( - db.models.Q(cluster__isnull=True) | db.models.Q(cluster=Conf.PREFIX) - ) - ): - args = () - kwargs = {} - # get args, kwargs and hook - if s.kwargs: - try: - # first try the dict syntax - kwargs = ast.literal_eval(s.kwargs) - except (SyntaxError, ValueError): - # else use the kwargs syntax - try: - parsed_kwargs = ( - ast.parse(f"f({s.kwargs})").body[0].value.keywords - ) - kwargs = { - kwarg.arg: ast.literal_eval(kwarg.value) - for kwarg in parsed_kwargs - } - except (SyntaxError, ValueError): - kwargs = {} - if s.args: - args = ast.literal_eval(s.args) - # single value won't eval to tuple, so: - if type(args) != tuple: - args = (args,) - q_options = kwargs.get("q_options", {}) - if s.intended_date_kwarg: - kwargs[s.intended_date_kwarg] = s.next_run.isoformat() - if s.hook: - q_options["hook"] = s.hook - # set up the next run time - if s.schedule_type != s.ONCE: - next_run = s.next_run - while True: - next_run = s.calculate_next_run(next_run) - if Conf.CATCH_UP or next_run > localtime(): - break - - s.next_run = next_run - s.repeats += -1 - # send it to the cluster - scheduled_broker = broker - try: - scheduled_broker = get_broker(q_options["broker_name"]) - except: # noqa: E722 - # invalid broker_name or non existing broker with broker_name - pass - q_options["broker"] = scheduled_broker - q_options["group"] = q_options.get("group", s.name or s.id) - kwargs["q_options"] = q_options - s.task = django_q.tasks.async_task(s.func, *args, **kwargs) - # log it - if not s.task: - logger.error( - _( - "%(process_name)s failed to create a task from schedule " - "[%(schedule)s]" - ) - % { - "process_name": current_process().name, - "schedule": s.name or s.id, - } - ) - else: - logger.info( - _( - "%(process_name)s created task %(task_name)s from schedule " - "[%(schedule)s]" - ) - % { - "process_name": current_process().name, - "task_name": humanize(s.task), - "schedule": s.name or s.id, - } - ) - # default behavior is to delete a ONCE schedule - if s.schedule_type == s.ONCE: - if s.repeats < 0: - s.delete() - continue - # but not if it has a positive repeats - s.repeats = 0 - # save the schedule - s.save() - except Exception: - logger.exception("Could not create task from schedule") - - -def close_old_django_connections(): - """ - Close django connections unless running with sync=True. - """ - if Conf.SYNC: - logger.warning( - "Preserving django database connections because sync=True. Beware " - "that tasks are now injected in the calling context/transactions " - "which may result in unexpected behaviour." - ) - else: - db.close_old_connections() + logger.info(_("All processes were terminated")) def set_cpu_affinity(n: int, process_ids: list, actual: bool = not Conf.TESTING): @@ -835,12 +333,3 @@ def set_cpu_affinity(n: int, process_ids: list, actual: bool = not Conf.TESTING) _("%(pid)s will use cpu %(affinity)s") % {"pid": pid, "affinity": affinity} ) - - -def rss_check(): - if Conf.MAX_RSS: - if resource: - return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss >= Conf.MAX_RSS - elif psutil: - return psutil.Process().memory_info().rss >= Conf.MAX_RSS * 1024 - return False diff --git a/django_q/conf.py b/django_q/conf.py index fb556ba..0cf1220 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -9,7 +9,7 @@ import pkg_resources from django.conf import settings from django.utils.translation import gettext_lazy as _ -from django_q.queues import Queue +from queue import Queue # optional try: @@ -73,7 +73,7 @@ class Conf: PREFIX = conf.get("name", "default") # Log output level - LOG_LEVEL = conf.get("log_level", "INFO") + LOG_LEVEL = conf.get("log_level", "DEBUG") # Maximum number of successful tasks kept in the database. 0 saves everything. # -1 saves none @@ -94,7 +94,7 @@ class Conf: ) # Guard loop sleep in seconds. Should be between 0 and 60 seconds. - GUARD_CYCLE = conf.get("guard_cycle", 0.5) + GUARD_CYCLE = conf.get("guard_cycle", 1) # Disable the scheduler SCHEDULER = conf.get("scheduler", True) @@ -222,6 +222,7 @@ class Conf: # logger logger = logging.getLogger("django-q") + # Set up standard logging handler in case there is none if not logger.hasHandlers(): logger.setLevel(level=getattr(logging, Conf.LOG_LEVEL)) diff --git a/django_q/exceptions.py b/django_q/exceptions.py new file mode 100644 index 0000000..bbdca98 --- /dev/null +++ b/django_q/exceptions.py @@ -0,0 +1,23 @@ +import signal + +class TimeoutException(SystemExit): + """Exception for when a worker takes too long to complete a task""" + pass + + +class TimeoutHandler: + def __init__(self, timeout: int): + self._timeout = timeout + + def raise_timeout_exception(self, signum, frame): + raise TimeoutException('Task exceeded maximum timeout value ' + '({0} seconds)'.format(self._timeout)) + + def __enter__(self): + signal.signal(signal.SIGALRM, self.raise_timeout_exception) + signal.alarm(self._timeout) + + def __exit__(self, exc_type, exc_value, traceback): + """When getting out of the timeout, reset the alarm, so it won't trigger""" + signal.alarm(0) + signal.signal(signal.SIGALRM, signal.SIG_DFL) diff --git a/django_q/models.py b/django_q/models.py index 8bf7f5c..b2491b2 100644 --- a/django_q/models.py +++ b/django_q/models.py @@ -21,8 +21,6 @@ from django_q.conf import croniter, Conf from django_q.signing import SignedPackage from django_q.utils import localtime, add_months, add_years -from .utils import get_func_repr - class Task(models.Model): id = models.CharField(max_length=32, primary_key=True, editable=False) @@ -306,20 +304,29 @@ class OrmQ(models.Model): payload = models.TextField() lock = models.DateTimeField(null=True) + @property def task(self): return SignedPackage.loads(self.payload) def func(self): - return get_func_repr(self.task()["func"]) + if isinstance(self.task, dict): + return self.task.get("func_name", "") + return self.task.func_name def task_id(self): - return self.task()["id"] + if isinstance(self.task, dict): + return self.task.get("id", "") + return self.task.id def name(self): - return self.task()["name"] + if isinstance(self.task, dict): + return self.task["name"] + return self.task.name def group(self): - return self.task().get("group") + if isinstance(self.task, dict): + return self.task.get("group", "") + return self.task.group class Meta: app_label = "django_q" diff --git a/django_q/monitor.py b/django_q/monitor.py index 1c89f6e..f5c96ab 100644 --- a/django_q/monitor.py +++ b/django_q/monitor.py @@ -1,510 +1,83 @@ -from datetime import timedelta - -# django -from django.db import connection -from django.db.models import F, Sum -from django.utils import timezone -from django.utils.translation import gettext as _ - -from django_q import VERSION, models +import multiprocessing +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.utils.translation import gettext_lazy as _ +from multiprocessing import Event, Process, Value, current_process -# local -from django_q.conf import Conf -from django_q.status import Stat - -# optional try: - import psutil -except ImportError: - psutil = None + import setproctitle +except ModuleNotFoundError: + setproctitle = None -def get_process_mb(pid): - try: - process = psutil.Process(pid) - mb_used = round(process.memory_info().rss / 1024**2, 2) - except psutil.NoSuchProcess: - mb_used = "NO_PROCESS_FOUND" - return mb_used +class Monitor(ProcessManager): + def __init__(self): + super().__init__() + self.task_queue = Queue(ctx=multiprocessing.get_context()) + self.status.value = self.Status.IDLE.value + def get_target(self): + return self.run_monitor -BLESSED_INSTALL_MESSAGE = ( - "Blessed is not installed. Please install blessed to use this: " - "https://pypi.org/project/blessed/" -) + def run_item(self): + if self.is_idle: + try: + task = self.task_queue.get_nowait() + except Empty: + # if the queue is empty, then just stop + return + self.manager_pipe.send(task) + def add_task(self, task): + self.task_queue.put(task) -def monitor(run_once=False, broker=None): - if not broker: + def run_monitor(self, status, pipe) -> None: broker = get_broker() - try: - from blessed import Terminal - - term = Terminal() - except ImportError: - print(BLESSED_INSTALL_MESSAGE) - return - - broker.ping() - with term.fullscreen(), term.hidden_cursor(), term.cbreak(): - val = None - start_width = int(term.width / 8) - while val not in ( - "q", - "Q", - ): - col_width = int(term.width / 8) - # In case of resize - if col_width != start_width: - print(term.clear()) - start_width = col_width - print( - term.move(0, 0) - + term.black_on_green(term.center(_("Host"), width=col_width - 1)) - ) - print( - term.move(0, 1 * col_width) - + term.black_on_green(term.center(_("Id"), width=col_width - 1)) - ) - print( - term.move(0, 2 * col_width) - + term.black_on_green(term.center(_("State"), width=col_width - 1)) - ) - print( - term.move(0, 3 * col_width) - + term.black_on_green(term.center(_("Pool"), width=col_width - 1)) - ) - print( - term.move(0, 4 * col_width) - + term.black_on_green(term.center(_("TQ"), width=col_width - 1)) - ) - print( - term.move(0, 5 * col_width) - + term.black_on_green(term.center(_("RQ"), width=col_width - 1)) - ) - print( - term.move(0, 6 * col_width) - + term.black_on_green(term.center(_("RC"), width=col_width - 1)) - ) - print( - term.move(0, 7 * col_width) - + term.black_on_green(term.center(_("Up"), width=col_width - 1)) - ) - i = 2 - stats = Stat.get_all(broker=broker) - print(term.clear_eos()) - for stat in stats: - status = stat.status - # color status - if stat.status == Conf.WORKING: - status = term.green(str(Conf.WORKING)) - elif stat.status == Conf.STOPPING: - status = term.yellow(str(Conf.STOPPING)) - elif stat.status == Conf.STOPPED: - status = term.red(str(Conf.STOPPED)) - elif stat.status == Conf.IDLE: - status = str(Conf.IDLE) - # color q's - tasks = str(stat.task_q_size) - if stat.task_q_size > 0: - tasks = term.cyan(str(stat.task_q_size)) - if Conf.QUEUE_LIMIT and stat.task_q_size == Conf.QUEUE_LIMIT: - tasks = term.green(str(stat.task_q_size)) - results = stat.done_q_size - if results > 0: - results = term.cyan(str(results)) - # color workers - workers = len(stat.workers) - if workers < Conf.WORKERS: - workers = term.yellow(str(workers)) - # format uptime - uptime = (timezone.now() - stat.tob).total_seconds() - hours, remainder = divmod(uptime, 3600) - minutes, seconds = divmod(remainder, 60) - uptime = "%d:%02d:%02d" % (hours, minutes, seconds) - # print to the terminal - print( - term.move(i, 0) - + term.center(stat.host[: col_width - 1], width=col_width - 1) - ) - print( - term.move(i, 1 * col_width) - + term.center(str(stat.cluster_id)[-8:], width=col_width - 1) - ) - print( - term.move(i, 2 * col_width) - + term.center(status, width=col_width - 1) - ) - print( - term.move(i, 3 * col_width) - + term.center(workers, width=col_width - 1) - ) - print( - term.move(i, 4 * col_width) - + term.center(tasks, width=col_width - 1) - ) - print( - term.move(i, 5 * col_width) - + term.center(results, width=col_width - 1) - ) - print( - term.move(i, 6 * col_width) - + term.center(stat.reincarnations, width=col_width - 1) - ) - print( - term.move(i, 7 * col_width) - + term.center(uptime, width=col_width - 1) - ) - i += 1 - # bottom bar - i += 1 - queue_size = broker.queue_size() - lock_size = broker.lock_size() - if lock_size: - queue_size = f"{queue_size}({lock_size})" - print( - term.move(i, 0) - + term.white_on_cyan(term.center(broker.info(), width=col_width * 2)) - ) - print( - term.move(i, 2 * col_width) - + term.black_on_cyan(term.center(_("Queued"), width=col_width)) - ) - print( - term.move(i, 3 * col_width) - + term.white_on_cyan(term.center(queue_size, width=col_width)) - ) - print( - term.move(i, 4 * col_width) - + term.black_on_cyan(term.center(_("Success"), width=col_width)) - ) - print( - term.move(i, 5 * col_width) - + term.white_on_cyan( - term.center(models.Success.objects.count(), width=col_width) - ) - ) - print( - term.move(i, 6 * col_width) - + term.black_on_cyan(term.center(_("Failures"), width=col_width)) - ) - print( - term.move(i, 7 * col_width) - + term.white_on_cyan( - term.center(models.Failure.objects.count(), width=col_width) - ) - ) - # for testing - if run_once: - return Stat.get_all(broker=broker) - print(term.move(i + 2, 0) + term.center(_("[Press q to quit]"))) - val = term.inkey(timeout=1) - - -def info(broker=None): - if not broker: - broker = get_broker() - try: - from blessed import Terminal - - term = Terminal() - except ImportError: - print(BLESSED_INSTALL_MESSAGE) - return - - broker.ping() - stat = Stat.get_all(broker=broker) - # general stats - clusters = len(stat) - workers = 0 - reincarnations = 0 - for cluster in stat: - workers += len(cluster.workers) - reincarnations += cluster.reincarnations - # calculate tasks pm and avg exec time - tasks_per = 0 - per = _("day") - exec_time = 0 - last_tasks = models.Success.objects.filter( - stopped__gte=timezone.now() - timedelta(hours=24) - ) - tasks_per_day = last_tasks.count() - if tasks_per_day > 0: - # average execution time over the last 24 hours - if connection.vendor != "sqlite": - exec_time = last_tasks.aggregate( - time_taken=Sum(F("stopped") - F("started")) - ) - exec_time = exec_time["time_taken"].total_seconds() / tasks_per_day - else: - # can't sum timedeltas on sqlite - for t in last_tasks: - exec_time += t.time_taken() - exec_time = exec_time / tasks_per_day - # tasks per second/minute/hour/day in the last 24 hours - if tasks_per_day > 24 * 60 * 60: - tasks_per = tasks_per_day / (24 * 60 * 60) - per = _("second") - elif tasks_per_day > 24 * 60: - tasks_per = tasks_per_day / (24 * 60) - per = _("minute") - elif tasks_per_day > 24: - tasks_per = tasks_per_day / 24 - per = _("hour") - else: - tasks_per = tasks_per_day - # print to terminal - print(term.clear_eos()) - col_width = int(term.width / 6) - print( - term.black_on_green( - term.center( - _("-- %(prefix)s %(version)s on %(info)s --") - % { - "prefix": Conf.PREFIX.capitalize(), - "version": ".".join(str(v) for v in VERSION), - "info": broker.info(), - } - ) + proc_name = current_process().name + if setproctitle: + setproctitle.setproctitle(f"qcluster {proc_name} monitor") + logger.info( + _("%(name)s monitoring at %(id)s") % {"name": proc_name, "id": current_process().pid} ) - ) - print( - term.cyan(_("Clusters")) - + term.move_x(1 * col_width) - + term.white(str(clusters)) - + term.move_x(2 * col_width) - + term.cyan(_("Workers")) - + term.move_x(3 * col_width) - + term.white(str(workers)) - + term.move_x(4 * col_width) - + term.cyan(_("Restarts")) - + term.move_x(5 * col_width) - + term.white(str(reincarnations)) - ) - print( - term.cyan(_("Queued")) - + term.move_x(1 * col_width) - + term.white(str(broker.queue_size())) - + term.move_x(2 * col_width) - + term.cyan(_("Successes")) - + term.move_x(3 * col_width) - + term.white(str(models.Success.objects.count())) - + term.move_x(4 * col_width) - + term.cyan(_("Failures")) - + term.move_x(5 * col_width) - + term.white(str(models.Failure.objects.count())) - ) - print( - term.cyan(_("Schedules")) - + term.move_x(1 * col_width) - + term.white(str(models.Schedule.objects.count())) - + term.move_x(2 * col_width) - + term.cyan(_("Tasks/%(per)s") % {"per": per}) - + term.move_x(3 * col_width) - + term.white(f"{tasks_per:.2f}") - + term.move_x(4 * col_width) - + term.cyan(_("Avg time")) - + term.move_x(5 * col_width) - + term.white(f"{exec_time:.4f}") - ) - return True + while True: + task = pipe.recv() + status.value = self.Status.BUSY + if task == "STOP": + logger.info(f"Monitor {proc_name} shut down") + break + # save the result + if task.cached: + task.save_cached(broker) + else: + task.save_to_db(broker) + # acknowledge result + if task.ack_id and (not task.has_succeeded or task.ack_failure): + broker.acknowledge(task.ack_id) + # signal execution done + post_execute.send(sender="django_q", task=task) + # log the result + if task.has_succeeded: + # log success + logger.info( + _("Processed '%(info_name)s' (%(task_name)s)") + % {"info_name": task.func_name, "task_name": task.name} + ) + else: + # log failure + logger.error( + _("Failed '%(info_name)s' (%(task_name)s) - %(task_result)s") + % { + "info_name": task.func_name, + "task_name": task.name, + "task_result": task.result_payload, + } + ) + status.value = self.Status.IDLE + logger.info(_("%(name)s stopped monitoring results") % {"name": proc_name}) -def memory(run_once=False, workers=False, broker=None): - if not broker: - broker = get_broker() - try: - from blessed import Terminal - - term = Terminal() - except ImportError: - print(BLESSED_INSTALL_MESSAGE) - return - broker.ping() - if not psutil: - print(term.clear_eos()) - print( - term.white_on_red( - 'Cannot start "qmemory" command. Missing "psutil" library.' - ) - ) - return - with term.fullscreen(), term.hidden_cursor(), term.cbreak(): - MEMORY_AVAILABLE_LOWEST_PERCENTAGE = 100.0 - MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT = timezone.now() - cols = 8 - val = None - start_width = int(term.width / cols) - while val not in ["q", "Q"]: - col_width = int(term.width / cols) - # In case of resize - if col_width != start_width: - print(term.clear()) - start_width = col_width - # sentinel, monitor and workers memory usage - print( - term.move(0, 0 * col_width) - + term.black_on_green(term.center(_("Host"), width=col_width - 1)) - ) - print( - term.move(0, 1 * col_width) - + term.black_on_green(term.center(_("Id"), width=col_width - 1)) - ) - print( - term.move(0, 2 * col_width) - + term.black_on_green( - term.center(_("Available (%)"), width=col_width - 1) - ) - ) - print( - term.move(0, 3 * col_width) - + term.black_on_green( - term.center(_("Available (MB)"), width=col_width - 1) - ) - ) - print( - term.move(0, 4 * col_width) - + term.black_on_green(term.center(_("Total (MB)"), width=col_width - 1)) - ) - print( - term.move(0, 5 * col_width) - + term.black_on_green( - term.center(_("Sentinel (MB)"), width=col_width - 1) - ) - ) - print( - term.move(0, 6 * col_width) - + term.black_on_green( - term.center(_("Monitor (MB)"), width=col_width - 1) - ) - ) - print( - term.move(0, 7 * col_width) - + term.black_on_green( - term.center(_("Workers (MB)"), width=col_width - 1) - ) - ) - row = 2 - stats = Stat.get_all(broker=broker) - print(term.clear_eos()) - for stat in stats: - # memory available (%) - memory_available_percentage = round( - psutil.virtual_memory().available - * 100 - / psutil.virtual_memory().total, - 2, - ) - # memory available (MB) - memory_available = round( - psutil.virtual_memory().available / 1024**2, 2 - ) - if memory_available_percentage < MEMORY_AVAILABLE_LOWEST_PERCENTAGE: - MEMORY_AVAILABLE_LOWEST_PERCENTAGE = memory_available_percentage - MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT = timezone.now() - print( - term.move(row, 0 * col_width) - + term.center(stat.host[: col_width - 1], width=col_width - 1) - ) - print( - term.move(row, 1 * col_width) - + term.center(str(stat.cluster_id)[-8:], width=col_width - 1) - ) - print( - term.move(row, 2 * col_width) - + term.center(memory_available_percentage, width=col_width - 1) - ) - print( - term.move(row, 3 * col_width) - + term.center(memory_available, width=col_width - 1) - ) - print( - term.move(row, 4 * col_width) - + term.center( - round(psutil.virtual_memory().total / 1024**2, 2), - width=col_width - 1, - ) - ) - print( - term.move(row, 5 * col_width) - + term.center(get_process_mb(stat.sentinel), width=col_width - 1) - ) - print( - term.move(row, 6 * col_width) - + term.center( - get_process_mb(getattr(stat, "monitor", None)), - width=col_width - 1, - ) - ) - workers_mb = 0 - for worker_pid in stat.workers: - result = get_process_mb(worker_pid) - if isinstance(result, str): - result = 0 - workers_mb += result - print( - term.move(row, 7 * col_width) - + term.center( - workers_mb or "NO_PROCESSES_FOUND", width=col_width - 1 - ) - ) - row += 1 - # each worker's memory usage - if workers: - row += 2 - col_width = int(term.width / (1 + Conf.WORKERS)) - print( - term.move(row, 0 * col_width) - + term.black_on_cyan(term.center(_("Id"), width=col_width - 1)) - ) - for worker_num in range(Conf.WORKERS): - print( - term.move(row, (worker_num + 1) * col_width) - + term.black_on_cyan( - term.center( - "Worker #{} (MB)".format(worker_num + 1), - width=col_width - 1, - ) - ) - ) - row += 2 - for stat in stats: - print( - term.move(row, 0 * col_width) - + term.center(str(stat.cluster_id)[-8:], width=col_width - 1) - ) - for idx, worker_pid in enumerate(stat.workers): - mb_used = get_process_mb(worker_pid) - print( - term.move(row, (idx + 1) * col_width) - + term.center(mb_used, width=col_width - 1) - ) - row += 1 - row += 1 - print( - term.move(row, 0) - + _("Available lowest (): %(memory_percent)s ((at)s)") - % { - "memory_percent": str(MEMORY_AVAILABLE_LOWEST_PERCENTAGE), - "at": MEMORY_AVAILABLE_LOWEST_PERCENTAGE_AT.strftime( - "%Y-%m-%d %H:%M:%S+00:00" - ), - } - ) - # for testing - if run_once: - return Stat.get_all(broker=broker) - print(term.move(row + 2, 0) + term.center(_("[Press q to quit]"))) - val = term.inkey(timeout=1) - - -def get_ids(): - # prints id (PID) of running clusters - stat = Stat.get_all() - if stat: - for s in stat: - print(s.cluster_id) - else: - print(_("No clusters appear to be running.")) - return True diff --git a/django_q/process_manager.py b/django_q/process_manager.py new file mode 100644 index 0000000..80e49b5 --- /dev/null +++ b/django_q/process_manager.py @@ -0,0 +1,75 @@ +from abc import ABC +from django_q.conf import Conf, logger +from django_q.humanhash import humanize +import uuid +from django_q.conf import Conf +import enum +from django import db +from typing import Callable +from django.utils.translation import gettext_lazy as _ +import multiprocessing +from multiprocessing import Process, Value + +class ProcessManager(ABC): + class Status(enum.IntEnum): + IDLE = 1 + BUSY = 2 + DONE = 3 + RECYCLE = 4 + + target = None + status = Value("i", Status.IDLE.value) + + def get_target(self) -> Callable: + if self.target is None: + raise ValueError("Process must have target specified") + return self.target + + def __init__(self): + self.process = self.spawn_process() + self.name = humanize(uuid.uuid4().hex) + + def spawn_process(self) -> Process: + self.manager_pipe, process_pipe = multiprocessing.Pipe(duplex=True) + p = Process(target=self.get_target(), args=(self.status, process_pipe)) + p.start() + return p + + def reincarnate_process(self): + # kill connections before killing the process + logger.critical(_("reincarnated worker %(name)s after death") % {"name": self.process.name}) + if not Conf.SYNC: + db.connections.close_all() + self.process.kill() + self.process = self.spawn_process() + self.mark_idle() + + @property + def has_results(self): + # poll puller pipe for new tasks + return self.manager_pipe.poll() + + def get_result(self): + # get the puller pipe task object back from the worker + return self.manager_pipe.recv() + + @property + def is_alive(self): + # get the puller status + return self.process.is_alive() + + @property + def is_done(self): + return self.status.value == self.Status.DONE.value + + @property + def is_idle(self): + return self.status.value == self.Status.IDLE.value + + @property + def is_recycle(self): + # worker needs to be recycled/reincarnated + return self.status.value == self.Status.RECYCLE.value + + def mark_idle(self): + self.status.value = self.Status.IDLE.value diff --git a/django_q/puller.py b/django_q/puller.py new file mode 100644 index 0000000..cf3fafe --- /dev/null +++ b/django_q/puller.py @@ -0,0 +1,68 @@ +from django_q.worker import Worker +from django_q.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 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.process_manager import ProcessManager + + +class Puller(ProcessManager): + """The Puller is responsible for pulling the tasks from the broker, then return them to be picked up by the + guard""" + + def get_target(self): + return self.run_puller + + def stop_puller(self): + self.status.value = self.Status.DONE.value + + def run_puller(self, status, pipe) -> None: + broker = get_broker() + proc_name = current_process().name + if setproctitle: + setproctitle.setproctitle(f"qcluster {proc_name} puller") + logger.info( + _("%(name)s pulling tasks from broker %(id)s") + % {"name": proc_name, "id": current_process().pid} + ) + while True: + if status.value == Worker.Status.DONE.value: + logger.info("Stopping Puller") + break + try: + task_set = broker.dequeue() + except Exception: + logger.exception("Failed to pull task from broker") + # broker probably crashed. Let the sentinel handle it. + sleep(10) + break + if task_set: + logger.info( + _("Found %(amount_tasks)s tasks") % {"amount_tasks": len(task_set)} + ) + for task in task_set: + ack_id = task[0] + # unpack the task + try: + queue_task = SignedPackage.loads(task[1]) + except (TypeError, BadSignature): + logger.exception("Failed to pull task from broker - bad task") + broker.fail(ack_id) + continue + queue_task.ack_id = ack_id + # send back to main process + pipe.send(queue_task) + logger.debug( + _("queueing from %(list_key)s") % {"list_key": broker.list_key} + ) + logger.info(_("%(name)s stopped pushing tasks") % {"name": current_process().name}) diff --git a/django_q/queue_task.py b/django_q/queue_task.py new file mode 100644 index 0000000..387b373 --- /dev/null +++ b/django_q/queue_task.py @@ -0,0 +1,203 @@ +from __future__ import annotations +from datetime import datetime +from django.utils import timezone +from django_q.signing import SignedPackage +from django_q.models import Success, Task +from django_q.utils import close_old_django_connections +import enum +from django_q import tasks +import inspect +import pydoc +from django import db + +from dataclasses import dataclass, field +from typing import Any, Callable, Optional, Union + +from django_q.conf import Conf, logger + + +@dataclass +class QueueTask: + class Result(enum.IntEnum): + SUCCESS = 1 + FAILED = 2 + TIMEOUT = 3 + + func: Union[Callable, str] + name: str + group: Optional[None] = None + queued_at: Optional[datetime] = timezone.now() + finished_at: Optional[datetime] = None + ack_id: Optional[str] = None + started_at: Optional[datetime] = None + id: str = "-1" + timeout: Union[int, None] = Conf.TIMEOUT + result: Union[Result, None] = None + result_payload: Any = None + save: bool = Conf.SAVE_LIMIT >= 0 + chain: Union[str, QueueTask] = "" + cached: bool = False + sync: bool = False + hook: Union[str, None] = None + args: tuple = field(default_factory=tuple) + kwargs: dict = field(default_factory=dict) + ack_failure: bool = Conf.ACK_FAILURES + iter_count: Union[int, None] = None + iter_cached: Union[int, None] = None + + def callable_func(self): + func = self.func + if not callable(func): + func = pydoc.locate(func) + return func + + @property + def has_succeeded(self): + return self.result == self.Result.SUCCESS + + @property + def has_timed_out(self): + return self.result == self.Result.TIMEOUT + + @property + def is_callable(self): + return self.callable_func is not None + + @property + def func_name(self): + if inspect.isfunction(self.func): + return f"{self.func.__module__}.{self.func.__name__}" + elif inspect.ismethod(self.func) and hasattr(self.func.__self__, "__name__"): + return ( + f"{self.func.__self__.__module__}." f"{self.func.__self__.__name__}.{self.func.__name__}" + ) + else: + return str(self.func) + + def save_to_db(self, broker: Broker): + """ + Saves the task package to Django or the cache + :param task: the task package + :type broker: brokers.Broker + """ + # SAVE LIMIT < 0 : Don't save success + if not self.save and self.has_succeeded: + return + # enqueues next in a chain + if self.chain: + tasks.async_chain( + self.chain, + group=self.group, + cached=self.cached, + sync=self.sync, + broker=broker, + ) + close_old_django_connections() + + try: + filters = {} + if ( + Conf.SAVE_LIMIT_PER + and Conf.SAVE_LIMIT_PER in {"group", "name", "func"} + and Conf.SAVE_LIMIT_PER in self + ): + value = getattr(self, Conf.SAVE_LIMIT_PER) + if Conf.SAVE_LIMIT_PER == "func": + value = self.func_name + filters[Conf.SAVE_LIMIT_PER] = value + + with db.transaction.atomic(using=db.router.db_for_write(Success)): + last = Success.objects.filter(**filters).select_for_update().last() + if ( + self.has_succeeded + and 0 < Conf.SAVE_LIMIT <= Success.objects.filter(**filters).count() + ): + # delete the last entry if we are hitting the limit + last.delete() + + # check if this task has previous results + existing_task, created = Task.objects.get_or_create( + id=self.id, + name=self.name, + defaults={ + 'func': self.func_name, + 'stopped': self.finished_at, + 'hook': self.hook, + 'args': self.args, + 'kwargs': self.kwargs, + 'started': self.started_at, + 'result': self.result_payload, + 'group': self.group, + 'success': self.has_succeeded, + 'attempt_count': 1 + } + ) + + # only update the result if it hasn't succeeded yet + if not created and not existing_task.success: + existing_task.stopped = self.finished_at + existing_task.result = self.result_payload + existing_task.success = self.has_succeeded + existing_task.save() + + if ( + Conf.MAX_ATTEMPTS > 0 + and existing_task.attempt_count >= Conf.MAX_ATTEMPTS + ): + broker.acknowledge(self.ack_id) + + except Exception: + logger.exception("Could not save task result") + + + def save_cached(self, broker: Broker): + task_key = f'{broker.list_key}:{self.id}' + timeout = self.cached + if timeout is True: + timeout = None + try: + group = self.group + iter_count = self.iter_count + # if it's a group append to the group list + if group: + group_key = f"{broker.list_key}:{group}:keys" + group_list = broker.cache.get(group_key) or [] + # if it's an iter group, check if we are ready + if iter_count and len(group_list) == iter_count - 1: + group_args = f"{broker.list_key}:{group}:args" + # collate the results into a Task result + results = [ + SignedPackage.get_queue_task(broker.cache.get(k)).result_payload + for k in group_list + ] + results.append(self.result_payload) + self.result_payload = results + self.id = group + self.args = SignedPackage.loads(broker.cache.get(group_args)) + self.iter_count = None + self.group = None + if self.iter_cached: + self.cached = self.iter_cached + self.save_cached(broker=broker) + else: + self.save_to_db(broker) + broker.cache.delete_many(group_list) + broker.cache.delete_many([group_key, group_args]) + return + # save the group list + group_list.append(task_key) + broker.cache.set(group_key, group_list, timeout) + # async_task next in a chain + if self.chain: + tasks.async_chain( + self.chain, + group=group, + cached=self.cached, + sync=self.sync, + broker=broker, + ) + # save the task + broker.cache.set(task_key, SignedPackage.dumps(self), timeout) + except Exception: + logger.exception("Could not save task result") + diff --git a/django_q/queues.py b/django_q/queues.py deleted file mode 100644 index 5af9a5a..0000000 --- a/django_q/queues.py +++ /dev/null @@ -1,82 +0,0 @@ -""" -The code is derived from -https://github.com/althonos/pronto/commit/3384010dfb4fc7c66a219f59276adef3288a886b -""" -import multiprocessing -import multiprocessing.queues -import sys - - -class SharedCounter: - """A synchronized shared counter. - - The locking done by multiprocessing.Value ensures that only a single - process or thread may read or write the in-memory ctypes object. However, - in order to do n += 1, Python performs a read followed by a write, so a - second process may read the old value before the new one is written by - the first process. The solution is to use a multiprocessing.Lock to - guarantee the atomicity of the modifications to Value. - - This class comes almost entirely from Eli Bendersky's blog: - http://eli.thegreenplace.net/2012/01/04/shared-counter-with-pythons-multiprocessing/ - """ - - def __init__(self, n=0): - self.count = multiprocessing.Value("i", n) - - def increment(self, n=1): - """Increment the counter by n (default = 1)""" - with self.count.get_lock(): - self.count.value += n - - @property - def value(self): - """Return the value of the counter""" - return self.count.value - - -class Queue(multiprocessing.queues.Queue): - """A portable implementation of multiprocessing.Queue. - - Because of multithreading / multiprocessing semantics, Queue.qsize() may - raise the NotImplementedError exception on Unix platforms like Mac OS X - where sem_getvalue() is not implemented. This subclass addresses this - problem by using a synchronized shared counter (initialized to zero) and - increasing / decreasing its value every time the put() and get() methods - are called, respectively. This not only prevents NotImplementedError from - being raised, but also allows us to implement a reliable version of both - qsize() and empty(). - """ - - def __init__(self, *args, **kwargs): - if sys.version_info < (3, 0): - super(Queue, self).__init__(*args, **kwargs) - else: - super(Queue, self).__init__( - *args, ctx=multiprocessing.get_context(), **kwargs - ) - self.size = SharedCounter(0) - - def __getstate__(self): - return super(Queue, self).__getstate__() + (self.size,) - - def __setstate__(self, state): - super(Queue, self).__setstate__(state[:-1]) - self.size = state[-1] - - def put(self, *args, **kwargs): - super(Queue, self).put(*args, **kwargs) - self.size.increment(1) - - def get(self, *args, **kwargs): - x = super(Queue, self).get(*args, **kwargs) - self.size.increment(-1) - return x - - def qsize(self) -> int: - """Reliable implementation of multiprocessing.Queue.qsize()""" - return self.size.value - - def empty(self) -> bool: - """Reliable implementation of multiprocessing.Queue.empty()""" - return not self.qsize() > 0 diff --git a/django_q/scheduler.py b/django_q/scheduler.py new file mode 100644 index 0000000..6fffbf8 --- /dev/null +++ b/django_q/scheduler.py @@ -0,0 +1,128 @@ +from utils import localtime +from models import Schedule +from djang_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 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.process_manager import ProcessManager + + +class Scheduler(ProcessManager): + """The Scheduler is responsible for scheduling new tasks""" + + def get_target(self): + return self.run_scheduler + + def run_scheduler(self, status, pipe) -> None: + while True: + broker = get_broker() + close_old_django_connections() + try: + with db.transaction.atomic(using=db.router.db_for_write(Schedule)): + for s in ( + Schedule.objects.select_for_update() + .exclude(repeats=0) + .filter(db.models.Q(next_run__lt=timezone.now()), db.models.Q(cluster__isnull=True) | db.models.Q(cluster=Conf.PREFIX)) + ): + args = () + kwargs = {} + # get args, kwargs and hook + if s.kwargs: + try: + # first try the dict syntax + kwargs = ast.literal_eval(s.kwargs) + except (SyntaxError, ValueError): + # else use the kwargs syntax + try: + parsed_kwargs = ( + ast.parse(f"f({s.kwargs})").body[0].value.keywords + ) + kwargs = { + kwarg.arg: ast.literal_eval(kwarg.value) + for kwarg in parsed_kwargs + } + except (SyntaxError, ValueError): + kwargs = {} + if s.args: + args = ast.literal_eval(s.args) + # single value won't eval to tuple, so: + if type(args) != tuple: + args = (args,) + q_options = kwargs.get("q_options", {}) + if s.intended_date_kwarg: + kwargs[s.intended_date_kwarg] = s.next_run.isoformat() + if s.hook: + q_options["hook"] = s.hook + # set up the next run time + if s.schedule_type != s.ONCE: + next_run = s.next_run + while True: + next_run = s.calculate_next_run(next_run) + if Conf.CATCH_UP or next_run > localtime(): + break + + s.next_run = next_run + s.repeats += -1 + # send it to the cluster + scheduled_broker = broker + try: + scheduled_broker = get_broker(q_options["broker_name"]) + except: # noqa: E722 + # invalid broker_name or non existing broker with broker_name + pass + q_options["broker"] = scheduled_broker + q_options["group"] = q_options.get("group", s.name or s.id) + kwargs["q_options"] = q_options + s.task = tasks.async_task(s.func, *args, **kwargs) + # log it + if not s.task: + logger.error( + _( + "%(process_name)s failed to create a task from schedule " + "[%(schedule)s]" + ) + % { + "process_name": current_process().name, + "schedule": s.name or s.id, + } + ) + else: + logger.info( + _( + "%(process_name)s created task %(task_name)s from schedule " + "[%(schedule)s]" + ) + % { + "process_name": current_process().name, + "task_name": humanize(s.task), + "schedule": s.name or s.id, + } + ) + # default behavior is to delete a ONCE schedule + if s.schedule_type == s.ONCE: + if s.repeats < 0: + s.delete() + continue + # but not if it has a positive repeats + s.repeats = 0 + # save the schedule + s.save() + except Exception: + logger.exception("Could not create task from schedule") + # sleep 60 seconds for next schedule + sleep(60) + diff --git a/django_q/status.py b/django_q/status.py index 7ac01ef..0272288 100644 --- a/django_q/status.py +++ b/django_q/status.py @@ -44,11 +44,11 @@ class Stat(Status): if Conf.QSIZE: self.done_q_size = sentinel.result_queue.qsize() self.task_q_size = sentinel.task_queue.qsize() - if sentinel.monitor: - self.monitor = sentinel.monitor.pid - if sentinel.pusher: - self.pusher = sentinel.pusher.pid - self.workers = [w.pid for w in sentinel.pool] + # if sentinel.monitor: + # self.monitor = sentinel.monitor.pid + # if sentinel.pusher: + # self.pusher = sentinel.pusher.pid + # self.workers = [w.pid for w in sentinel.pool] def uptime(self) -> float: return (timezone.now() - self.tob).total_seconds() diff --git a/django_q/tasks.py b/django_q/tasks.py index 5f91ba5..cb91ee1 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -1,5 +1,6 @@ """Provides task functionality.""" # Standard +from django_q.queue_task import QueueTask from multiprocessing import Value from time import sleep, time @@ -12,13 +13,16 @@ from django_q.brokers import get_broker from django_q.conf import Conf, logger from django_q.humanhash import uuid from django_q.models import Schedule, Task -from django_q.queues import Queue from django_q.signals import pre_enqueue from django_q.signing import SignedPackage def async_task(func, *args, **kwargs): """Queue a task for the cluster.""" + logger.info("Adding task") + logger.info(func) + logger.info(args) + logger.info(kwargs) keywords = kwargs.copy() opt_keys = ( "hook", @@ -37,43 +41,41 @@ def async_task(func, *args, **kwargs): # get an id tag = uuid() # build the task package - task = { - "id": tag[1], - "name": keywords.pop("task_name", None) - or q_options.pop("task_name", None) - or tag[0], - "func": func, - "args": args, - } + task = QueueTask( + id=tag[1], + name=keywords.pop("task_name", None) or q_options.pop("task_name", None) or tag[0], + func=func, + args=args + ) + + # push optionals - for key in opt_keys: - if q_options and key in q_options: - task[key] = q_options[key] - elif key in keywords: - task[key] = keywords.pop(key) - # don't serialize the broker - broker = task.pop("broker", get_broker()) - # overrides - if "cached" not in task and Conf.CACHED: - task["cached"] = Conf.CACHED - if "sync" not in task and Conf.SYNC: - task["sync"] = Conf.SYNC - if "ack_failure" not in task and Conf.ACK_FAILURES: - task["ack_failure"] = Conf.ACK_FAILURES - # finalize - task["kwargs"] = keywords - task["started"] = timezone.now() + # for key in opt_keys: + # if q_options and key in q_options: + # task[key] = q_options[key] + # elif key in keywords: + # task[key] = keywords.pop(key) + # # don't serialize the broker + #broker = task.pop("broker", get_broker()) + broker = get_broker() + # # overrides + # if "cached" not in task and Conf.CACHED: + # task["cached"] = Conf.CACHED + # if "sync" not in task and Conf.SYNC: + # task["sync"] = Conf.SYNC + # # finalize + task.kwargs = keywords # signal it pre_enqueue.send(sender="django_q", task=task) # sign it pack = SignedPackage.dumps(task) - if task.get("sync", False): - return _sync(pack) + # if task.get("sync", False): + # return _sync(pack) # push it enqueue_id = broker.enqueue(pack) logger.info(f"Enqueued {enqueue_id}") logger.debug(f"Pushed {tag}") - return task["id"] + return task.id def schedule(func, *args, **kwargs): @@ -110,7 +112,7 @@ def schedule(func, *args, **kwargs): raise IntegrityError("A schedule with the same name already exists.") # create and return the schedule - s = Schedule( + return Schedule.objects.create( name=name, func=func, hook=hook, @@ -124,11 +126,6 @@ def schedule(func, *args, **kwargs): cluster=cluster, intended_date_kwarg=intended_date_kwarg, ) - # make sure we trigger validation - s.full_clean() - s.save() - return s - def result(task_id, wait=0, cached=Conf.CACHED): """ @@ -267,16 +264,16 @@ def fetch_cached(task_id, wait=0, broker=None): if r: task = SignedPackage.loads(r) return Task( - id=task["id"], - name=task["name"], - func=task["func"], - hook=task.get("hook"), - args=task["args"], - kwargs=task["kwargs"], - started=task["started"], - stopped=task["stopped"], - result=task["result"], - success=task["success"], + id=task.id, + name=task.name, + func=task.func, + hook=task.hook, + args=task.args, + kwargs=task.kwargs, + started=task.started_at, + stopped=task.finished_at, + result=task.result, + success=task.result_payload, ) if (time() - start) * 1000 >= wait >= 0: break @@ -337,17 +334,17 @@ def fetch_group_cached(group_id, failures=True, wait=0, count=None, broker=None) task = SignedPackage.loads(broker.cache.get(task_key)) if task["success"] or failures: t = Task( - id=task["id"], - name=task["name"], - func=task["func"], - hook=task.get("hook"), - args=task["args"], - kwargs=task["kwargs"], - started=task["started"], - stopped=task["stopped"], - result=task["result"], - group=task.get("group"), - success=task["success"], + id=task.id, + name=task.name, + func=task.func, + hook=task.hook, + args=task.args, + kwargs=task.kwargs, + started=task.started_at, + stopped=task.finished_at, + result=task.result_payload, + group=task.group, + success=task.result, ) task_list.append(t) return task_list @@ -384,7 +381,7 @@ def count_group_cached(group_id, failures=False, broker=None): failure_count = 0 for task_key in group_list: task = SignedPackage.loads(broker.cache.get(task_key)) - if not task["success"]: + if not task.has_succeeded: failure_count += 1 return failure_count @@ -762,16 +759,14 @@ def _sync(pack): """Simulate a package travelling through the cluster.""" from django_q.cluster import monitor, worker - task_queue = Queue() - result_queue = Queue() - task = SignedPackage.loads(pack) - task_queue.put(task) - task_queue.put("STOP") - worker(task_queue, result_queue, Value("f", -1)) - result_queue.put("STOP") - monitor(result_queue) - task_queue.close() - task_queue.join_thread() - result_queue.close() - result_queue.join_thread() - return task["id"] + # task = SignedPackage.loads(pack) + # task_queue.put(task) + # task_queue.put("STOP") + # worker(task_queue, result_queue, Value("f", -1)) + # result_queue.put("STOP") + # monitor(result_queue) + # task_queue.close() + # task_queue.join_thread() + # result_queue.close() + # result_queue.join_thread() + # return task["id"] diff --git a/django_q/utils.py b/django_q/utils.py index 936214d..d982683 100644 --- a/django_q/utils.py +++ b/django_q/utils.py @@ -1,11 +1,11 @@ from datetime import datetime +from django import db import calendar -import inspect from datetime import date import django from django.utils import timezone -from django.conf import settings +from django_q.conf import settings, logger from django_q.conf import Conf @@ -45,16 +45,6 @@ def add_years(d, years): return d.replace(year=new_date.year, month=new_date.month, day=new_date.day) -def get_func_repr(func): - # convert func to string - if inspect.isfunction(func): - return f"{func.__module__}.{func.__name__}" - elif inspect.ismethod(func) and hasattr(func.__self__, "__name__"): - return ( - f"{func.__self__.__module__}." f"{func.__self__.__name__}.{func.__name__}" - ) - else: - return str(func) def localtime(value=None) -> datetime: @@ -72,3 +62,18 @@ def localtime(value=None) -> datetime: return datetime.now() else: return value + + +def close_old_django_connections(): + """ + Close django connections unless running with sync=True. + """ + if Conf.SYNC: + logger.warning( + "Preserving django database connections because sync=True. Beware " + "that tasks are now injected in the calling context/transactions " + "which may result in unexpected behaviour." + ) + else: + db.close_old_connections() + diff --git a/django_q/worker.py b/django_q/worker.py new file mode 100644 index 0000000..caaaaea --- /dev/null +++ b/django_q/worker.py @@ -0,0 +1,203 @@ +import multiprocessing +from multiprocessing.queues import Queue +from queue import Empty +from typing import Optional +from django_q.queue_task import QueueTask +from django.utils import timezone +import traceback + +from multiprocessing import Process, Value, 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.signals import pre_execute +from django_q.exceptions import TimeoutException, TimeoutHandler +from django_q.process_manager import ProcessManager + + + +class Worker(ProcessManager): + + def spawn_process(self) -> Process: + """ + :type target: function or class + """ + self.status = Value("i", Worker.Status.IDLE.value) + self.manager_pipe, worker_process_pipe = multiprocessing.Pipe(duplex=True) + p = WorkerProcess(args=(self.status, worker_process_pipe)) + p.daemon = Conf.DAEMONIZE_WORKERS + p.start() + return p + + def start_task(self, task) -> None: + # send task to worker + self.manager_pipe.send(task) + + +class Pool: + """This will manager the individual workers""" + def __init__(self, workers=Conf.WORKERS): + self.amount_workers = workers + self.workers = [] + self.task_queue = Queue(ctx=multiprocessing.get_context()) + self.start_workers() + + def start_workers(self): + for __ in range(self.amount_workers): + self.workers.append(Worker()) + + def get_worker(self, id) -> Optional[Worker]: + worker = next((worker for worker in self.workers if worker.id == id), None) + if worker is None: + logger.error("Couldn't find worker") + return + return worker + + @property + def is_healthy(self): + """Checks if all workers are still operating""" + return all(worker.is_alive for worker in self.workers) + + 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] + for worker in stopped_workers: + worker.reincarnate_process() + + def add_task(self, task): + self.task_queue.put(task) + + def get_done_workers(self): + """Worker tasks that have been completed, but need to be saved to cache/db - to be processed by monitor worker""" + return [worker for worker in self.workers if worker.is_done] + + def mark_workers_idle(self, worker_ids): + """Mark workers idle when they are ready to be used again""" + for worker_id in worker_ids: + # We are going to process the result, mark them idle, so they can be used for a different task + worker = self.get_worker(id=worker_id) + if worker is not None: + worker.mark_idle() + + def delegate_tasks(self): + available_workers = [worker for worker in self.workers if worker.is_idle] + for worker in available_workers: + try: + task = self.task_queue.get_nowait() + except Empty: + # if the queue is empty, then just stop + break + worker.start_task(task) + + +class WorkerProcess(Process): + + def __init__(self, group=None, name=None, args=(), kwargs={}, daemon=None): + target = self.processing_tasks + super().__init__(group=group, target=target, name=name, args=args, kwargs=kwargs, daemon=daemon) + + def mark_ready(self): + self.process_name = current_process().name + self.process_id = current_process().pid + self.task_count = 0 + logger.info( + _("%(proc_name)s ready for work at %(id)s") + % {"proc_name": self.process_name, "id": self.process_id} + ) + + + def mark_start_task(self, task): + # Log task creation and set process name + task_desc = ( + _("%(proc_name)s processing %(task_name)s '%(func_name)s'") + % { + "proc_name": self.process_name, + "func_name": task.func_name, + "task_name": task.name, + } + ) + if task.group is not None: + task_desc += f" [{task.group}]" + logger.info(task_desc) + + if setproctitle: + proc_title = f"qcluster {self.process_name} processing {task.name} '{task.func_name}'" + if task.group is not None: + proc_title += f" [{task.group}]" + setproctitle.setproctitle(proc_title) + + def processing_tasks(self, status: Value, pipe): + self.mark_ready() + while True: + task = pipe.recv() + if task == "STOP": + logger.info(f"Worker {self.process_name} stopped processing") + break + # got a new task, let's mark it starting + self.mark_start_task(task) + + # make sure the function actually exists, before we try to run it + try: + if not task.is_callable: + raise ValueError(f"Function {task.func_name} is not defined") + except Exception as e: + result = (f"{e} : {traceback.format_exc()}", False) + if error_reporter: + error_reporter.report() + if task.sync: + raise + # stop here, move on to the next one + continue + + close_old_django_connections() + # signal execution + pre_execute.send(sender="django_q", func=task.func, task=task) + + status.value = Worker.Status.BUSY.value + task.started_at = timezone.now() + try: + with TimeoutHandler(timeout=task.timeout): + func = task.callable_func() + res = func(*task.args, **task.kwargs) + result = res + except (TimeoutException, Exception) as e: + if isinstance(e, TimeoutException): + task.result = QueueTask.Result.TIMEOUT + else: + task.result = QueueTask.Result.FAILED + result = f"{e} : {traceback.format_exc()}" + logger.info(result) + + if error_reporter: + error_reporter.report() + if task.sync: + raise + else: + # succeeded + task.result = QueueTask.Result.SUCCESS + finally: + task.result_payload = result + task.finished_at = timezone.now() + + # Add task towards total + self.task_count += 1 + + # Set to DONE so main process can pick it up + status.value = Worker.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 + break + + pipe.send(task) + + def rss_check(self): + if Conf.MAX_RSS: + if resource: + return resource.getrusage(resource.RUSAGE_SELF).ru_maxrss >= Conf.MAX_RSS + elif psutil: + return psutil.Process().memory_info().rss >= Conf.MAX_RSS * 1024 + return False