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
This commit is contained in:
parent
656ed3c847
commit
5f3c132523
|
@ -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)
|
||||
|
|
|
@ -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()
|
||||
|
|
Loading…
Reference in New Issue