diff --git a/django_q/cluster.py b/django_q/cluster.py index 4dad340..59e57b1 100644 --- a/django_q/cluster.py +++ b/django_q/cluster.py @@ -132,7 +132,7 @@ class Sentinel(object): self.pool_size = Conf.WORKERS self.pool = [] self.timeout = timeout - self.task_queue = Queue() + self.task_queue = Queue(maxsize=Conf.QUEUE_LIMIT) if Conf.QUEUE_LIMIT else Queue() self.result_queue = Queue() self.event_out = Event() self.monitor = Process() @@ -314,7 +314,7 @@ def pusher(task_queue, event, list_key=Conf.Q_LIST, r=redis_client): sleep(10) break if task: - task_queue.put(task[1]) + task_queue.put(task[1], block=True) logger.debug(_('queueing from {}').format(list_key)) if event.is_set(): break diff --git a/django_q/conf.py b/django_q/conf.py index 407d17a..a2b3517 100644 --- a/django_q/conf.py +++ b/django_q/conf.py @@ -33,6 +33,9 @@ class Conf(object): # Failures are always saved SAVE_LIMIT = conf.get('save_limit', 250) + # Maximum number of tasks that each cluster can work on + QUEUE_LIMIT = conf.get('queue_limit', None) + # Number of workers in the pool. Default is cpu count. +2 for monitor and pusher WORKERS = conf.get('workers', cpu_count())