fork/django-q2
1
0
Fork 0
mirror of https://github.com/django-q2/django-q2.git synced 2026-10-05 13:58:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #336 from jannero/timeout

Browse Source 
Timeout handling fix and improvements to related tests
This commit is contained in:
Ilan Steemers
2019-01-27 18:26:51 +01:00
committed by GitHub
parent 7ffb3e4da2 0e2df88d92
commit 9611423732
2 changed files with 27 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
django_q/cluster.py
+6 -4
View File
@@ -172,7 +172,7 @@ class Sentinel(object):
else: else:
self.pool.remove(process) self.pool.remove(process)
self.spawn_worker() self.spawn_worker()
if self.timeout and int(process.timer.value) == 0: if process.timer.value == 0:
# only need to terminate on timeout, otherwise we risk destabilizing the queues # only need to terminate on timeout, otherwise we risk destabilizing the queues
process.terminate() process.terminate()
logger.warn(_("reincarnated worker {} after timeout").format(process.name)) logger.warn(_("reincarnated worker {} after timeout").format(process.name))
@@ -210,11 +210,11 @@ class Sentinel(object):
# Check Workers # Check Workers
for p in self.pool: for p in self.pool:
# Are you alive? # Are you alive?
if not p.is_alive() or (self.timeout and p.timer.value == 0): if not p.is_alive() or p.timer.value == 0:
self.reincarnate(p) self.reincarnate(p)
continue continue
# Decrement timer if work is being done # Decrement timer if work is being done
if self.timeout and p.timer.value > 0: if p.timer.value > 0:
p.timer.value -= cycle p.timer.value -= cycle
# Check Monitor # Check Monitor
if not self.monitor.is_alive(): if not self.monitor.is_alive():
@@ -347,6 +347,8 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
name = current_process().name name = current_process().name
logger.info(_('{} ready for work at {}').format(name, current_process().pid)) logger.info(_('{} ready for work at {}').format(name, current_process().pid))
task_count = 0 task_count = 0
if timeout is None:
timeout = -1
# Start reading the task queue # Start reading the task queue
for task in iter(task_queue.get, 'STOP'): for task in iter(task_queue.get, 'STOP'):
result = None result = None
@@ -368,7 +370,7 @@ def worker(task_queue, result_queue, timer, timeout=Conf.TIMEOUT):
# We're still going # We're still going
if not result: if not result:
db.close_old_connections() db.close_old_connections()
timer_value = task['kwargs'].pop('timeout', timeout or 0) timer_value = task['kwargs'].pop('timeout', timeout)
# signal execution # signal execution
pre_execute.send(sender="django_q", func=f, task=task) pre_execute.send(sender="django_q", func=f, task=task)
# execute the payload # execute the payload
django_q/tests/test_cluster.py
+21 -9
View File
@@ -244,16 +244,21 @@ def test_enqueue(broker, admin_user):
@pytest.mark.django_db @pytest.mark.django_db
def test_timeout(broker): @pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', (
(1, {}),
(10, {'timeout': 1}),
(None, {'timeout': 1}),
))
def test_timeout(broker, cluster_config_timeout, async_task_kwargs):
# set up the Sentinel # set up the Sentinel
broker.list_key = 'timeout_test:q' broker.list_key = 'timeout_test:q'
broker.purge_queue() broker.purge_queue()
async_task('django_q.tests.tasks.count_forever', broker=broker) async_task('time.sleep', 5, broker=broker, **async_task_kwargs)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
# Set a timer to stop the Sentinel # Set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start() threading.Timer(3, stop_event.set).start()
s = Sentinel(stop_event, start_event, broker=broker, timeout=1) s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout)
assert start_event.is_set() assert start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
assert s.reincarnations == 1 assert s.reincarnations == 1
@@ -261,18 +266,25 @@ def test_timeout(broker):
@pytest.mark.django_db @pytest.mark.django_db
def test_timeout_override(broker): @pytest.mark.parametrize('cluster_config_timeout, async_task_kwargs', (
(5, {}),
(10, {'timeout': 5}),
(1, {'timeout': 5}),
(None, {'timeout': 5}),
))
def test_timeout_task_finishes(broker, cluster_config_timeout, async_task_kwargs):
# set up the Sentinel # set up the Sentinel
broker.list_key = 'timeout_override_test:q' broker.list_key = 'timeout_test:q'
async_task('django_q.tests.tasks.count_forever', broker=broker, timeout=1) broker.purge_queue()
async_task('time.sleep', 3, broker=broker, **async_task_kwargs)
start_event = Event() start_event = Event()
stop_event = Event() stop_event = Event()
# Set a timer to stop the Sentinel # Set a timer to stop the Sentinel
threading.Timer(3, stop_event.set).start() threading.Timer(6, stop_event.set).start()
s = Sentinel(stop_event, start_event, broker=broker, timeout=10) s = Sentinel(stop_event, start_event, broker=broker, timeout=cluster_config_timeout)
assert start_event.is_set() assert start_event.is_set()
assert s.status() == Conf.STOPPED assert s.status() == Conf.STOPPED
assert s.reincarnations == 1 assert s.reincarnations == 0
broker.delete_queue() broker.delete_queue()