From e78d6c693154f205eeadeff1a40f2f183db6ef7f Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Thu, 1 Dec 2016 18:41:28 +0100 Subject: [PATCH 01/13] allow to lock the job for a specific task path, new expired status related --- mrq/basetasks/cleaning.py | 15 +++++++++++++++ mrq/exceptions.py | 4 ++++ mrq/job.py | 21 +++++++++++++++++++-- mrq/task.py | 1 + mrq/worker.py | 6 +++++- 5 files changed, 44 insertions(+), 3 deletions(-) diff --git a/mrq/basetasks/cleaning.py b/mrq/basetasks/cleaning.py index 532a98a1..8c1a76f3 100644 --- a/mrq/basetasks/cleaning.py +++ b/mrq/basetasks/cleaning.py @@ -10,6 +10,8 @@ class RequeueInterruptedJobs(Task): """ Requeue jobs that were marked as status=interrupt when a worker got a SIGTERM. """ + locked_job = True + def run(self, params): return run_task("mrq.basetasks.utils.JobAction", { "status": "interrupt", @@ -21,6 +23,8 @@ class RequeueRetryJobs(Task): """ Requeue jobs that were marked as retry. """ + locked_job = True + def run(self, params): return run_task("mrq.basetasks.utils.JobAction", { "status": "retry", @@ -37,6 +41,8 @@ class RequeueStartedJobs(Task): The timeout parameter of this task is in addition to the task's own timeout. """ + locked_job = True + def run(self, params): additional_timeout = params.get("timeout", 300) @@ -76,6 +82,8 @@ class RequeueRedisStartedJobs(Task): redis.lpop and mongodb.update """ + locked_job = True + def run(self, params): redis_key_started = Queue.redis_key_started() @@ -124,6 +132,8 @@ class RequeueLostJobs(Task): They could have been lost by a Redis flush or another severe issue """ + locked_job = True + def run(self, params): # If there are more than this much items on the queue, we don't try to check if our mongodb @@ -186,6 +196,9 @@ class MigrateKnownQueues(Task): """ Migrate known_queues from old set format to new zset """ + + locked_job = True + def run(self, params): key = "%s:known_queues" % get_current_config()["redis_prefix"] for queue in connections.redis.smembers(key): @@ -202,6 +215,8 @@ class CleanKnownQueues(Task): - be empty """ + locked_job = True + def run(self, params): max_age = int(params.get("max_age") or (7 * 86400)) diff --git a/mrq/exceptions.py b/mrq/exceptions.py index d4217778..dde254b3 100644 --- a/mrq/exceptions.py +++ b/mrq/exceptions.py @@ -50,3 +50,7 @@ class StopRequested(GreenletExit): class JobInterrupt(GreenletExit): """ Interrupts that stop a job in its execution, e.g. when responding to a SIGTERM. """ pass + + +class LockExpiredInterrupt(_MrqInterrupt): + pass diff --git a/mrq/job.py b/mrq/job.py index d00492f4..9922962d 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -1,7 +1,7 @@ import datetime from bson import ObjectId import time -from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt +from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt, LockExpiredInterrupt from .utils import load_class_by_path, group_iter import gevent import objgraph @@ -18,6 +18,8 @@ import copy_reg from . import context +import redis.lock + class Job(object): @@ -274,7 +276,22 @@ def perform(self): self.task.is_main_task = True - result = self.task.run_wrapped(self.data["params"]) + lock = None + + if self.task.locked_job: + lock_key = "mrq:l:%s" % self.data["path"] + context.log.debug("Trying to acquire lock '%s'" % lock_key) + lock = context.connections.redis.lock(lock_key, timeout=self.timeout) + if not lock.acquire(blocking=True, blocking_timeout=1): + raise LockExpiredInterrupt() + + context.log.debug("Lock '%s' acquired." % lock_key) + + try: + result = self.task.run_wrapped(self.data["params"]) + finally: + if lock: + lock.release() self.save_success(result) diff --git a/mrq/task.py b/mrq/task.py index 22dd0202..83ab25ce 100644 --- a/mrq/task.py +++ b/mrq/task.py @@ -3,6 +3,7 @@ class Task(object): # Are we the first task that a Job called? is_main_task = False + locked_job = False # Default write concern values when setting status=success # http://docs.mongodb.org/manual/reference/write-concern/ diff --git a/mrq/worker.py b/mrq/worker.py index 51bcaeae..32b319f3 100644 --- a/mrq/worker.py +++ b/mrq/worker.py @@ -16,7 +16,7 @@ from .job import Job from .exceptions import (TimeoutInterrupt, StopRequested, JobInterrupt, AbortInterrupt, - RetryInterrupt, MaxRetriesInterrupt) + RetryInterrupt, MaxRetriesInterrupt, LockExpiredInterrupt) from .context import (set_current_worker, set_current_job, get_current_job, get_current_config, connections, enable_greenlet_tracing) from .queue import Queue @@ -617,6 +617,10 @@ def perform_job(self, job): try: job.perform() + except LockExpiredInterrupt: + self.log.error("Lock expired") + job._save_status("expired") + except RetryInterrupt: self.log.error("Caught retry") job.save_retry(sys.exc_info()[1]) From 079ccb52cf570b4f8d8cda6d84c6cc6cd1368d14 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Thu, 1 Dec 2016 18:42:55 +0100 Subject: [PATCH 02/13] added test using concurrency --- mrq/job.py | 12 ++++++++---- tests/tasks/lock.py | 11 +++++++++++ tests/test_locked_job.py | 20 ++++++++++++++++++++ 3 files changed, 39 insertions(+), 4 deletions(-) create mode 100644 tests/tasks/lock.py create mode 100644 tests/test_locked_job.py diff --git a/mrq/job.py b/mrq/job.py index 9922962d..fe51575a 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -72,6 +72,11 @@ def __init__(self, job_id, queue=None, start=False, fetch=False): elif fetch: self.fetch(start=False, full_data=False) + @property + def redis_key_lock(self): + """ Returns the global redis key used to store started job ids """ + return "%s:l:%s" % (context.get_current_config()["redis_prefix"], self.data["path"]) + def exists(self): """ Returns True if a job with the current _id exists in MongoDB. """ return bool(self.collection.find_one({"_id": self.id}, projection={"_id": 1})) @@ -279,13 +284,12 @@ def perform(self): lock = None if self.task.locked_job: - lock_key = "mrq:l:%s" % self.data["path"] - context.log.debug("Trying to acquire lock '%s'" % lock_key) - lock = context.connections.redis.lock(lock_key, timeout=self.timeout) + context.log.debug("Trying to acquire lock '%s'" % self.redis_key_lock) + lock = context.connections.redis.lock(self.redis_key_lock, timeout=self.timeout) if not lock.acquire(blocking=True, blocking_timeout=1): raise LockExpiredInterrupt() - context.log.debug("Lock '%s' acquired." % lock_key) + context.log.debug("Lock '%s' acquired." % self.redis_key_lock) try: result = self.task.run_wrapped(self.data["params"]) diff --git a/tests/tasks/lock.py b/tests/tasks/lock.py new file mode 100644 index 00000000..5ccd8be1 --- /dev/null +++ b/tests/tasks/lock.py @@ -0,0 +1,11 @@ +import time +from mrq.task import Task +from mrq.context import log + + +class Locked(Task): + + locked_job = True + + def run(self, params): + time.sleep(1) diff --git a/tests/test_locked_job.py b/tests/test_locked_job.py new file mode 100644 index 00000000..81539673 --- /dev/null +++ b/tests/test_locked_job.py @@ -0,0 +1,20 @@ +from mrq.job import Job + + +def test_locked_job(worker): + + worker.start(flags="--greenlets=2") + + job_ids = [] + + for i in range(4): + job_ids.append( + worker.send_task("tests.tasks.lock.Locked", {"queue": "default"}, block=False)) + + worker.wait_for_tasks_results(job_ids, accept_statuses=["success", "failed", "expired"]) + worker.stop() + + statuses = [Job(job_id).fetch().data["status"] for job_id in job_ids] + assert "success" in statuses + assert "expired" in statuses + assert "failed" not in statuses From 8af73f4aed25192ba2ed540b39953034f885a7b0 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Thu, 1 Dec 2016 19:05:48 +0100 Subject: [PATCH 03/13] added lock_timeout parameter --- mrq/config.py | 7 +++++++ mrq/job.py | 4 +++- 2 files changed, 10 insertions(+), 1 deletion(-) diff --git a/mrq/config.py b/mrq/config.py index 044ce7cf..64d42c4a 100644 --- a/mrq/config.py +++ b/mrq/config.py @@ -391,6 +391,13 @@ def add_parser_args(parser, config_type): help='Strategy for dequeuing multiple queues. Default is \'sequential\',' + 'to dequeue them in command-line order.') + parser.add_argument( + '--lock_timeout', + default=1., + type=float, + action='store', + help='Max seconds to wait for a lock before changing the job status to expired.') + def get_config( sources=( diff --git a/mrq/job.py b/mrq/job.py index fe51575a..7ce816c9 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -286,7 +286,9 @@ def perform(self): if self.task.locked_job: context.log.debug("Trying to acquire lock '%s'" % self.redis_key_lock) lock = context.connections.redis.lock(self.redis_key_lock, timeout=self.timeout) - if not lock.acquire(blocking=True, blocking_timeout=1): + acquire_timeout = context.get_current_config()["lock_timeout"] + + if not lock.acquire(blocking=True, blocking_timeout=acquire_timeout): raise LockExpiredInterrupt() context.log.debug("Lock '%s' acquired." % self.redis_key_lock) From dd0d02759f0f04905845e88e0774282f54a0706c Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Fri, 2 Dec 2016 16:35:34 +0100 Subject: [PATCH 04/13] worker.stop will stop mongodb too --- tests/test_locked_job.py | 1 - 1 file changed, 1 deletion(-) diff --git a/tests/test_locked_job.py b/tests/test_locked_job.py index 81539673..fa836b60 100644 --- a/tests/test_locked_job.py +++ b/tests/test_locked_job.py @@ -12,7 +12,6 @@ def test_locked_job(worker): worker.send_task("tests.tasks.lock.Locked", {"queue": "default"}, block=False)) worker.wait_for_tasks_results(job_ids, accept_statuses=["success", "failed", "expired"]) - worker.stop() statuses = [Job(job_id).fetch().data["status"] for job_id in job_ids] assert "success" in statuses From ff8db380352da6bd8882bdf6c813b34e170dccf2 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 16:13:54 +0100 Subject: [PATCH 05/13] from lock to job max concurrency --- mrq/basetasks/cleaning.py | 14 +++++++------- mrq/exceptions.py | 2 +- mrq/task.py | 2 +- mrq/worker.py | 8 ++++---- tests/tasks/lock.py | 11 ----------- tests/test_locked_job.py | 19 ------------------- 6 files changed, 13 insertions(+), 43 deletions(-) delete mode 100644 tests/tasks/lock.py delete mode 100644 tests/test_locked_job.py diff --git a/mrq/basetasks/cleaning.py b/mrq/basetasks/cleaning.py index 8c1a76f3..6130ac70 100644 --- a/mrq/basetasks/cleaning.py +++ b/mrq/basetasks/cleaning.py @@ -10,7 +10,7 @@ class RequeueInterruptedJobs(Task): """ Requeue jobs that were marked as status=interrupt when a worker got a SIGTERM. """ - locked_job = True + max_concurrency = 1 def run(self, params): return run_task("mrq.basetasks.utils.JobAction", { @@ -23,7 +23,7 @@ class RequeueRetryJobs(Task): """ Requeue jobs that were marked as retry. """ - locked_job = True + max_concurrency = 1 def run(self, params): return run_task("mrq.basetasks.utils.JobAction", { @@ -41,7 +41,7 @@ class RequeueStartedJobs(Task): The timeout parameter of this task is in addition to the task's own timeout. """ - locked_job = True + max_concurrency = 1 def run(self, params): @@ -82,7 +82,7 @@ class RequeueRedisStartedJobs(Task): redis.lpop and mongodb.update """ - locked_job = True + max_concurrency = 1 def run(self, params): @@ -132,7 +132,7 @@ class RequeueLostJobs(Task): They could have been lost by a Redis flush or another severe issue """ - locked_job = True + max_concurrency = 1 def run(self, params): @@ -197,7 +197,7 @@ class MigrateKnownQueues(Task): Migrate known_queues from old set format to new zset """ - locked_job = True + max_concurrency = 1 def run(self, params): key = "%s:known_queues" % get_current_config()["redis_prefix"] @@ -215,7 +215,7 @@ class CleanKnownQueues(Task): - be empty """ - locked_job = True + max_concurrency = 1 def run(self, params): diff --git a/mrq/exceptions.py b/mrq/exceptions.py index dde254b3..87c3537c 100644 --- a/mrq/exceptions.py +++ b/mrq/exceptions.py @@ -52,5 +52,5 @@ class JobInterrupt(GreenletExit): pass -class LockExpiredInterrupt(_MrqInterrupt): +class MaxConcurrencyInterrupt(_MrqInterrupt): pass diff --git a/mrq/task.py b/mrq/task.py index 83ab25ce..2aea0711 100644 --- a/mrq/task.py +++ b/mrq/task.py @@ -3,7 +3,7 @@ class Task(object): # Are we the first task that a Job called? is_main_task = False - locked_job = False + max_concurrency = 0 # Default write concern values when setting status=success # http://docs.mongodb.org/manual/reference/write-concern/ diff --git a/mrq/worker.py b/mrq/worker.py index 32b319f3..9bfc6c8b 100644 --- a/mrq/worker.py +++ b/mrq/worker.py @@ -16,7 +16,7 @@ from .job import Job from .exceptions import (TimeoutInterrupt, StopRequested, JobInterrupt, AbortInterrupt, - RetryInterrupt, MaxRetriesInterrupt, LockExpiredInterrupt) + RetryInterrupt, MaxRetriesInterrupt, MaxConcurrencyInterrupt) from .context import (set_current_worker, set_current_job, get_current_job, get_current_config, connections, enable_greenlet_tracing) from .queue import Queue @@ -617,9 +617,9 @@ def perform_job(self, job): try: job.perform() - except LockExpiredInterrupt: - self.log.error("Lock expired") - job._save_status("expired") + except MaxConcurrencyInterrupt: + self.log.error("Max concurrency reached") + job._save_status("maxconcurrency") except RetryInterrupt: self.log.error("Caught retry") diff --git a/tests/tasks/lock.py b/tests/tasks/lock.py deleted file mode 100644 index 5ccd8be1..00000000 --- a/tests/tasks/lock.py +++ /dev/null @@ -1,11 +0,0 @@ -import time -from mrq.task import Task -from mrq.context import log - - -class Locked(Task): - - locked_job = True - - def run(self, params): - time.sleep(1) diff --git a/tests/test_locked_job.py b/tests/test_locked_job.py deleted file mode 100644 index fa836b60..00000000 --- a/tests/test_locked_job.py +++ /dev/null @@ -1,19 +0,0 @@ -from mrq.job import Job - - -def test_locked_job(worker): - - worker.start(flags="--greenlets=2") - - job_ids = [] - - for i in range(4): - job_ids.append( - worker.send_task("tests.tasks.lock.Locked", {"queue": "default"}, block=False)) - - worker.wait_for_tasks_results(job_ids, accept_statuses=["success", "failed", "expired"]) - - statuses = [Job(job_id).fetch().data["status"] for job_id in job_ids] - assert "success" in statuses - assert "expired" in statuses - assert "failed" not in statuses From 90bc82c688be67a9ed45ea46f87ca32b1256fee0 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 16:15:22 +0100 Subject: [PATCH 06/13] used redis pipeline, safer job run --- mrq/job.py | 33 +++++++++++++++------------------ 1 file changed, 15 insertions(+), 18 deletions(-) diff --git a/mrq/job.py b/mrq/job.py index 7ce816c9..da41314d 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -1,7 +1,7 @@ import datetime from bson import ObjectId import time -from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt, LockExpiredInterrupt +from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt, MaxConcurrencyInterrupt from .utils import load_class_by_path, group_iter import gevent import objgraph @@ -18,8 +18,6 @@ import copy_reg from . import context -import redis.lock - class Job(object): @@ -73,9 +71,9 @@ def __init__(self, job_id, queue=None, start=False, fetch=False): self.fetch(start=False, full_data=False) @property - def redis_key_lock(self): + def redis_concurrency_key(self): """ Returns the global redis key used to store started job ids """ - return "%s:l:%s" % (context.get_current_config()["redis_prefix"], self.data["path"]) + return "%s:c:%s" % (context.get_current_config()["redis_prefix"], self.data["path"]) def exists(self): """ Returns True if a job with the current _id exists in MongoDB. """ @@ -281,23 +279,22 @@ def perform(self): self.task.is_main_task = True - lock = None - - if self.task.locked_job: - context.log.debug("Trying to acquire lock '%s'" % self.redis_key_lock) - lock = context.connections.redis.lock(self.redis_key_lock, timeout=self.timeout) - acquire_timeout = context.get_current_config()["lock_timeout"] - - if not lock.acquire(blocking=True, blocking_timeout=acquire_timeout): - raise LockExpiredInterrupt() + try: + if self.task.max_concurrency: + pipeline = context.connections.redis.pipeline() + pipeline.incr(self.redis_concurrency_key) + pipeline.expireat(self.redis_concurrency_key, int(time.time()) + self.timeout) + current = pipeline.execute()[0] - context.log.debug("Lock '%s' acquired." % self.redis_key_lock) + if current > self.task.max_concurrency: + raise MaxConcurrencyInterrupt() - try: result = self.task.run_wrapped(self.data["params"]) + finally: - if lock: - lock.release() + if self.task.max_concurrency: + pipeline.decr(self.redis_concurrency_key) + pipeline.execute() self.save_success(result) From aa2532c61c6878429b245475e299948311620ce2 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 16:16:30 +0100 Subject: [PATCH 07/13] added test on max concurrency interrupt --- tests/tasks/concurrency.py | 18 ++++++++++++++++++ tests/test_interrupts.py | 22 ++++++++++++++++++++++ 2 files changed, 40 insertions(+) create mode 100644 tests/tasks/concurrency.py diff --git a/tests/tasks/concurrency.py b/tests/tasks/concurrency.py new file mode 100644 index 00000000..4675b5a8 --- /dev/null +++ b/tests/tasks/concurrency.py @@ -0,0 +1,18 @@ +import time +from mrq.task import Task +from mrq.context import log, get_current_job, get_current_worker, get_current_config +from .general import Add + +class LockedAdd(Add): + + max_concurrency = 1 + + def run(self, params): + log.info("adding", params) + res = params.get("a", 0) + params.get("b", 0) + + if params.get("sleep", 0): + log.info("sleeping", params.get("sleep", 0)) + time.sleep(params.get("sleep", 0)) + + return res diff --git a/tests/test_interrupts.py b/tests/test_interrupts.py index 5404658a..135924ca 100644 --- a/tests/test_interrupts.py +++ b/tests/test_interrupts.py @@ -319,3 +319,25 @@ def test_interrupt_maxjobs(worker): time.sleep(2) assert Queue("default").size() == 7 + + +def test_interrupt_maxconcurrency(worker): + + # The worker will raise a maxconcurrency on the second job + worker.start(flags="--greenlets=2") + + job_ids = worker.send_tasks("tests.tasks.concurrency.LockedAdd", [ + {"a": i, "b": 1, "sleep": 2} + for i in range(2) + ], block=False) + + worker.wait_for_tasks_results(job_ids, accept_statuses=["success", "failed", "maxconcurrency"]) + job_statuses = [Job(job_id).fetch().data["status"] for job_id in job_ids] + assert job_statuses == ["success", "maxconcurrency"] + + # the job concurrency key must be equal to 0 + last_job_id = worker.send_task("tests.tasks.concurrency.LockedAdd", + {"a": 1, "b": 1, "sleep": 2}, block=False + ) + last_job = Job(last_job_id).wait(poll_interval=0.01) + assert last_job.get("status") == "success" From 91946d4d70431525e4be327fab396995db93d3b4 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 19:56:12 +0100 Subject: [PATCH 08/13] added expire on concurrency key --- mrq/job.py | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/mrq/job.py b/mrq/job.py index da41314d..b4e716c5 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -70,10 +70,10 @@ def __init__(self, job_id, queue=None, start=False, fetch=False): elif fetch: self.fetch(start=False, full_data=False) - @property - def redis_concurrency_key(self): + def redis_max_concurrency_key(self, time): """ Returns the global redis key used to store started job ids """ - return "%s:c:%s" % (context.get_current_config()["redis_prefix"], self.data["path"]) + return "%s:c:%s:%s" % (context.get_current_config()["redis_prefix"], + self.data["path"], time // self.timeout) def exists(self): """ Returns True if a job with the current _id exists in MongoDB. """ @@ -281,9 +281,12 @@ def perform(self): try: if self.task.max_concurrency: + now = int(time.time()) + key = self.redis_max_concurrency_key(now) + pipeline = context.connections.redis.pipeline() - pipeline.incr(self.redis_concurrency_key) - pipeline.expireat(self.redis_concurrency_key, int(time.time()) + self.timeout) + pipeline.incr(key) + pipeline.expireat(key, now + self.timeout) current = pipeline.execute()[0] if current > self.task.max_concurrency: @@ -293,7 +296,7 @@ def perform(self): finally: if self.task.max_concurrency: - pipeline.decr(self.redis_concurrency_key) + pipeline.decr(key) pipeline.execute() self.save_success(result) From c2b99dc4ed8f82b9d9636e97f32c474830f48431 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 19:58:22 +0100 Subject: [PATCH 09/13] removed unused imports --- tests/tasks/concurrency.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/tasks/concurrency.py b/tests/tasks/concurrency.py index 4675b5a8..5e4ca174 100644 --- a/tests/tasks/concurrency.py +++ b/tests/tasks/concurrency.py @@ -1,6 +1,6 @@ import time from mrq.task import Task -from mrq.context import log, get_current_job, get_current_worker, get_current_config +from mrq.context import log from .general import Add class LockedAdd(Add): From f9b990efbd364252f091371624689d9237fc0145 Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 20:10:45 +0100 Subject: [PATCH 10/13] no longer necessary --- mrq/config.py | 7 ------- 1 file changed, 7 deletions(-) diff --git a/mrq/config.py b/mrq/config.py index 64d42c4a..044ce7cf 100644 --- a/mrq/config.py +++ b/mrq/config.py @@ -391,13 +391,6 @@ def add_parser_args(parser, config_type): help='Strategy for dequeuing multiple queues. Default is \'sequential\',' + 'to dequeue them in command-line order.') - parser.add_argument( - '--lock_timeout', - default=1., - type=float, - action='store', - help='Max seconds to wait for a lock before changing the job status to expired.') - def get_config( sources=( From 5777e7cb0f9f935f2c700993638e53c3016e974e Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Wed, 7 Dec 2016 20:19:49 +0100 Subject: [PATCH 11/13] exception on interrupt --- mrq/worker.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/mrq/worker.py b/mrq/worker.py index 9bfc6c8b..5f4e01d9 100644 --- a/mrq/worker.py +++ b/mrq/worker.py @@ -619,7 +619,7 @@ def perform_job(self, job): except MaxConcurrencyInterrupt: self.log.error("Max concurrency reached") - job._save_status("maxconcurrency") + job._save_status("maxconcurrency", exception=True) except RetryInterrupt: self.log.error("Caught retry") From 7cc825a52613592b7f5289d7de33afc9eec5ebba Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Thu, 8 Dec 2016 22:55:50 +0100 Subject: [PATCH 12/13] use of a simple lock for the current need --- mrq/job.py | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/mrq/job.py b/mrq/job.py index b4e716c5..b7c76021 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -70,10 +70,11 @@ def __init__(self, job_id, queue=None, start=False, fetch=False): elif fetch: self.fetch(start=False, full_data=False) - def redis_max_concurrency_key(self, time): + @property + def redis_max_concurrency_key(self): """ Returns the global redis key used to store started job ids """ - return "%s:c:%s:%s" % (context.get_current_config()["redis_prefix"], - self.data["path"], time // self.timeout) + return "%s:c:%s" % (context.get_current_config()["redis_prefix"], + self.data["path"]) def exists(self): """ Returns True if a job with the current _id exists in MongoDB. """ @@ -281,23 +282,26 @@ def perform(self): try: if self.task.max_concurrency: - now = int(time.time()) - key = self.redis_max_concurrency_key(now) - pipeline = context.connections.redis.pipeline() - pipeline.incr(key) - pipeline.expireat(key, now + self.timeout) - current = pipeline.execute()[0] + if self.task.max_concurrency > 1: + raise NotImplementedError() - if current > self.task.max_concurrency: + lock_acquired = False + + # TODO: implement a semaphore + lock = context.connections.redis.lock(self.redis_max_concurrency_key, timeout=self.timeout + 5) + if not lock.acquire(blocking=True, blocking_timeout=0): raise MaxConcurrencyInterrupt() + lock_acquired = True result = self.task.run_wrapped(self.data["params"]) finally: - if self.task.max_concurrency: - pipeline.decr(key) - pipeline.execute() + if lock_acquired: + try: + lock.release() + except Exception: + pass self.save_success(result) From 50f79141975423fa8dd3b95d6afca4ed18518dae Mon Sep 17 00:00:00 2001 From: Geoffrey Gueret Date: Fri, 9 Dec 2016 10:47:29 +0100 Subject: [PATCH 13/13] catch proper exception --- mrq/job.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/mrq/job.py b/mrq/job.py index b7c76021..5bcf3b8a 100644 --- a/mrq/job.py +++ b/mrq/job.py @@ -1,5 +1,6 @@ import datetime from bson import ObjectId +from redis.exceptions import LockError import time from .exceptions import RetryInterrupt, MaxRetriesInterrupt, AbortInterrupt, MaxConcurrencyInterrupt from .utils import load_class_by_path, group_iter @@ -281,26 +282,25 @@ def perform(self): self.task.is_main_task = True try: + lock = None + if self.task.max_concurrency: if self.task.max_concurrency > 1: raise NotImplementedError() - lock_acquired = False - # TODO: implement a semaphore lock = context.connections.redis.lock(self.redis_max_concurrency_key, timeout=self.timeout + 5) if not lock.acquire(blocking=True, blocking_timeout=0): raise MaxConcurrencyInterrupt() - lock_acquired = True result = self.task.run_wrapped(self.data["params"]) finally: - if lock_acquired: + if lock: try: lock.release() - except Exception: + except LockError: pass self.save_success(result)