diff options
author | Rick van de Loo <rickvandeloo@gmail.com> | 2017-05-13 16:55:50 +0200 |
---|---|---|
committer | Rick van de Loo <rickvandeloo@gmail.com> | 2017-05-13 16:55:50 +0200 |
commit | 5f3c1325239bace46c6adfe775f544467af44b03 (patch) | |
tree | 0add2f6c96837a06591ff4e727733a0cfc96b552 | |
parent | 656ed3c8479f4bb263d86d6bba6eadad77073dd5 (diff) | |
download | taskflow-5f3c1325239bace46c6adfe775f544467af44b03.tar.gz |
do not allow redis job reclaim by same owner
Running a nonblocking conductor or two conductors on the same host will re-execute the same job multiple times with the current implementation of 'claim' for the redis jobboard backend. This is different from the ZooKeeper jobboard backend, there the same owner of a job is not allowed to reclaim the job again (https://github.com/openstack/taskflow/blob/master/taskflow/jobs/backends/impl_zookeeper.py#L554). If the same owner is allowed to reclaim the job again there can be no concurrent execution on the same owner because all jobs will be re-claimed and re-executed by the same owner every pass as long as it's on the jobboard.
To reproduce this behavior:
- Use the redis jobboard backend
- Create a flow with a task that sleeps 10 seconds in the execute method
- Post that flow as a job
- Run a nonblocking conductor
It will claim and execute the same job multiple times in a loop until the first worker is finished and consumes the job. After this change it will not re-execute the same job multiple times.
Change-Id: I4f6c364211500e510fc496f23b03ce056771417d
-rw-r--r-- | taskflow/jobs/backends/impl_redis.py | 8 | ||||
-rw-r--r-- | taskflow/tests/unit/jobs/test_redis_job.py | 23 |
2 files changed, 26 insertions, 5 deletions
diff --git a/taskflow/jobs/backends/impl_redis.py b/taskflow/jobs/backends/impl_redis.py index a0ebe00..fca2699 100644 --- a/taskflow/jobs/backends/impl_redis.py +++ b/taskflow/jobs/backends/impl_redis.py @@ -410,12 +410,10 @@ if redis.call("hexists", listings_key, job_key) == 1 then -- Owner is the same, leave it alone... redis.call("set", last_modified_key, last_modified_blob) apply_ttl(owner_key, ms_expiry) - result["status"] = "${ok}" - else - result["status"] = "${error}" - result["reason"] = "${already_claimed}" - result["owner"] = owner end + result["status"] = "${error}" + result["reason"] = "${already_claimed}" + result["owner"] = owner else redis.call("set", owner_key, expected_owner) redis.call("set", last_modified_key, last_modified_blob) diff --git a/taskflow/tests/unit/jobs/test_redis_job.py b/taskflow/tests/unit/jobs/test_redis_job.py index f988788..2bde77f 100644 --- a/taskflow/tests/unit/jobs/test_redis_job.py +++ b/taskflow/tests/unit/jobs/test_redis_job.py @@ -20,6 +20,7 @@ from oslo_utils import uuidutils import six import testtools +from taskflow import exceptions as excp from taskflow.jobs.backends import impl_redis from taskflow import states from taskflow import test @@ -76,6 +77,28 @@ class RedisJobboardTest(test.TestCase, base.BoardTestMixin): possible_jobs = list(self.board.iterjobs(only_unclaimed=True)) self.assertEqual(1, len(possible_jobs)) + def test_posting_claim_same_owner(self): + with base.connect_close(self.board): + with self.flush(self.client): + self.board.post('test', p_utils.temporary_log_book()) + + self.assertEqual(1, self.board.job_count) + possible_jobs = list(self.board.iterjobs(only_unclaimed=True)) + self.assertEqual(1, len(possible_jobs)) + j = possible_jobs[0] + self.assertEqual(states.UNCLAIMED, j.state) + + with self.flush(self.client): + self.board.claim(j, self.board.name) + + possible_jobs = list(self.board.iterjobs()) + self.assertEqual(1, len(possible_jobs)) + with self.flush(self.client): + self.assertRaises(excp.UnclaimableJob, self.board.claim, + possible_jobs[0], self.board.name) + possible_jobs = list(self.board.iterjobs(only_unclaimed=True)) + self.assertEqual(0, len(possible_jobs)) + def setUp(self): super(RedisJobboardTest, self).setUp() self.client, self.board = self.create_board() |