Adds max_rss memory limit for recycle

django_q/cluster.py

@@ -1,6 +1,7 @@
 # Standard
 import ast
 import importlib
+import resource
 import signal
 import socket
 import traceback
@@ -10,6 +11,7 @@ from time import sleep
 
 # External
 import arrow
+
 # Django
 from django import db
 from django.conf import settings
@@ -102,10 +104,10 @@ class Cluster:
     @property
     def is_stopping(self) -> bool:
         return (
-                self.stop_event
+            self.stop_event
-                and self.start_event
+            and self.start_event
-                and self.start_event.is_set()
+            and self.start_event.is_set()
-                and self.stop_event.is_set()
+            and self.stop_event.is_set()
         )
 
     @property
@@ -115,13 +117,13 @@ class Cluster:
 
 class Sentinel:
     def __init__(
-            self,
+        self,
-            stop_event,
+        stop_event,
-            start_event,
+        start_event,
-            cluster_id,
+        cluster_id,
-            broker=None,
+        broker=None,
-            timeout=Conf.TIMEOUT,
+        timeout=Conf.TIMEOUT,
-            start=True,
+        start=True,
     ):
         # Make sure we catch signals for the pool
         signal.signal(signal.SIGINT, signal.SIG_IGN)
@@ -376,7 +378,7 @@ def monitor(result_queue: Queue, broker: Broker = None):
 
 
 def worker(
-        task_queue: Queue, result_queue: Queue, timer: Value, timeout: int = Conf.TIMEOUT
+    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
@@ -433,7 +435,7 @@ def worker(
             result_queue.put(task)
             timer.value = -1  # Idle
             # Recycle
-            if task_count == Conf.RECYCLE:
+            if task_count == Conf.RECYCLE or (Conf.MAX_RSS and resource.getrusage(resource.RUSAGE_SELF).ru_maxrss >= Conf.MAX_RSS):
                 timer.value = -2  # Recycled
                 break
     logger.info(_(f"{name} stopped doing work"))
@@ -551,9 +553,9 @@ def scheduler(broker: Broker = None):
     try:
         with db.transaction.atomic(using=Schedule.objects.db):
             for s in (
-                    Schedule.objects.select_for_update()
+                Schedule.objects.select_for_update()
-                            .exclude(repeats=0)
+                .exclude(repeats=0)
-                            .filter(next_run__lt=timezone.now())
+                .filter(next_run__lt=timezone.now())
             ):
                 args = ()
                 kwargs = {}

django_q/conf.py

@@ -104,6 +104,9 @@ class Conf:
     # Number of tasks each worker can handle before it gets recycled. Useful for releasing memory
     RECYCLE = conf.get("recycle", 500)
 
+    # The maximum resident set size in kilobytes before a worker will recycle. Useful for limiting memory usage.
+    MAX_RSS = conf.get("max_rss", None)
+
     # Number of seconds to wait for a worker to finish.
     TIMEOUT = conf.get("timeout", None)