From 3d12c1d36e01d7fe4613521cc87d7edcf527df9f Mon Sep 17 00:00:00 2001 From: Ilan Date: Thu, 13 Aug 2015 15:54:57 +0200 Subject: [PATCH] adds queue_size() returns the size of the current redis queue --- django_q/__init__.py | 2 +- django_q/tasks.py | 13 +++++++++++++ django_q/tests/test_cluster.py | 10 +++++----- django_q/tests/test_scheduler.py | 4 ++-- docs/tasks.rst | 8 ++++++++ 5 files changed, 29 insertions(+), 8 deletions(-) diff --git a/django_q/__init__.py b/django_q/__init__.py index 5a59427..a355c83 100644 --- a/django_q/__init__.py +++ b/django_q/__init__.py @@ -4,7 +4,7 @@ import sys myPath = os.path.dirname(os.path.abspath(__file__)) sys.path.insert(0, myPath) -from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group +from .tasks import async, schedule, result, result_group, fetch, fetch_group, count_group, delete_group, queue_size from .models import Task, Schedule, Success, Failure from .cluster import Cluster from .monitor import Stat diff --git a/django_q/tasks.py b/django_q/tasks.py index ea03ffa..ebbc318 100644 --- a/django_q/tasks.py +++ b/django_q/tasks.py @@ -152,6 +152,19 @@ def delete_group(group_id, tasks=False): return Task.delete_group(group_id, tasks) +def queue_size(list_key=Conf.Q_LIST, r=redis_client): + """ + Returns the current queue size. + Note that this doesn't count any tasks currently being processed by workers. + + :param list_key: optional redis key + :param r: optional redis connection + :return: current queue size + :rtype: int + """ + return r.llen(list_key) + + def _sync(pack): """Simulate a package travelling through the cluster.""" task_queue = Queue() diff --git a/django_q/tests/test_cluster.py b/django_q/tests/test_cluster.py index 684bb86..a5c1154 100644 --- a/django_q/tests/test_cluster.py +++ b/django_q/tests/test_cluster.py @@ -11,7 +11,7 @@ sys.path.insert(0, myPath + '/../') from django_q.cluster import Cluster, Sentinel, pusher, worker, monitor from django_q.humanhash import DEFAULT_WORDLIST -from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group +from django_q.tasks import fetch, fetch_group, async, result, result_group, count_group, delete_group, queue_size from django_q.models import Task, Success from django_q.conf import Conf, redis_client from django_q.monitor import Stat @@ -77,7 +77,7 @@ def test_cluster(r): list_key = 'cluster_test:q' r.delete(list_key) task = async('django_q.tests.tasks.count_letters', DEFAULT_WORDLIST, list_key=list_key) - assert r.llen(list_key) == 1 + assert queue_size(list_key=list_key, r=r) == 1 task_queue = Queue() assert task_queue.qsize() == 0 result_queue = Queue() @@ -87,7 +87,7 @@ def test_cluster(r): # Test push pusher(task_queue, event, list_key=list_key, r=r) assert task_queue.qsize() == 1 - assert r.llen(list_key) == 0 + assert queue_size(list_key=list_key, r=r) == 0 # Test work task_queue.put('STOP') worker(task_queue, result_queue, Value('f', -1)) @@ -142,14 +142,14 @@ def test_async(r, admin_user): assert isinstance(k, str) # run the cluster to execute the tasks task_count = 10 - assert r.llen(list_key) == task_count + assert queue_size(list_key=list_key, r=r) == task_count task_queue = Queue() stop_event = Event() stop_event.set() # push the tasks for i in range(task_count): pusher(task_queue, stop_event, list_key=list_key, r=r) - assert r.llen(list_key) == 0 + assert queue_size(list_key=list_key, r=r) == 0 assert task_queue.qsize() == task_count task_queue.put('STOP') # let a worker handle them diff --git a/django_q/tests/test_scheduler.py b/django_q/tests/test_scheduler.py index 074ed84..6a0a2f9 100644 --- a/django_q/tests/test_scheduler.py +++ b/django_q/tests/test_scheduler.py @@ -6,7 +6,7 @@ from django.utils import timezone from django_q.conf import redis_client from django_q.cluster import pusher, worker, monitor, scheduler -from django_q.tasks import Schedule, fetch, schedule as create_schedule +from django_q.tasks import Schedule, fetch, schedule as create_schedule, queue_size @pytest.fixture @@ -34,7 +34,7 @@ def test_scheduler(r): # push it pusher(task_queue, stop_event, list_key=list_key, r=r) assert task_queue.qsize() == 1 - assert r.llen(list_key) == 0 + assert queue_size(list_key=list_key,r=r) == 0 task_queue.put('STOP') # let a worker handle them result_queue = Queue() diff --git a/docs/tasks.rst b/docs/tasks.rst index 1125e95..8e7c969 100644 --- a/docs/tasks.rst +++ b/docs/tasks.rst @@ -213,6 +213,14 @@ Reference Renamed from get_task +.. py:function:: queue_size() + + Returns the size of the broker queue. + Note that this does not count tasks currently being processed. + + :returns: The amount of task packages in the broker + :rtype: int + .. py:function:: result_group(group_id, failures=False) Returns the results of a task group