diff --git a/CHANGELOG.md b/CHANGELOG.md index 69fe51d..e8bfe79 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,16 +8,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] ### Added -- `storage_mode` option for creating a `utube` tube (#228). It enables the - workaround for slow takes while working with busy tubes. +- `storage_mode` option for creating a `utube` and `utubettl` tube (#228). + It enables the workaround for slow takes while working with busy tubes. ### Fixed - Stuck in `INIT` state if an instance failed to enter the `running` mode in time (#226). This fix works only for Tarantool versions >= 2.10.0. -- Slow takes on busy `utube` tubes (#228). The workaround could be enabled by - passing the `storage_mode = "utube_ready_space"` option while creating - the tube. +- Slow takes on busy `utube` and `utubettl` tubes (#228). The workaround could + be enabled by passing the `storage_mode = "utube_ready_space"` option while + creating the tube. ## [1.3.3] - 2023-09-13 diff --git a/README.md b/README.md index 733721b..7aaf2d5 100644 --- a/README.md +++ b/README.md @@ -220,6 +220,11 @@ on disk already exists * `on_task_change` - function name - a callback to be executed on every operation + * `storage_mode` - string - one of + * `"default"` - default implementation of `utube` + * `"utube_ready_space"` - allows processing `take` requests faster, but + by the cost of `put` operations speed. Right now this option is enabled + only for `memtx` engine The following options can be specified when putting a task in a `utubettl` queue: diff --git a/queue/abstract/driver/utubettl.lua b/queue/abstract/driver/utubettl.lua index 51b0b61..0753641 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/queue/abstract/driver/utubettl.lua @@ -8,6 +8,11 @@ local qc = require('queue.compat') local num_type = qc.num_type local str_type = qc.str_type +local storage_modes = { + MODE_DEFAULT = "default", + MODE_UTUBE_READY_SPACE = "utube_ready_space", +} + local tube = {} local method = {} @@ -38,6 +43,18 @@ local function validate_space(space) end end +-- validate ready space of queue +local function validate_space_ready(space) + -- check indexes + local indexes = {'task_id', 'utube'} + for _, index in pairs(indexes) do + if space.index[index] == nil then + error(string.format('space "%s" does not have "%s" index', + space.name, index)) + end + end +end + -- create space function tube.create_space(space_name, opts) opts.ttl = opts.ttl or util.MAX_TIMEOUT @@ -94,11 +111,53 @@ local delayed_state = { state.DELAYED } local ttl_states = { state.READY, state.BURIED } local ttr_state = { state.TAKEN } +local function put_ready(self, utube) + local taken = self.space.index.utube:min{state.TAKEN, utube} + if taken == nil or taken[i_status] ~= state.TAKEN then + local next_task = self.space.index.utube:min{state.READY, utube} + if next_task == nil or next_task[i_status] ~= state.READY then + return + end + -- Ignoring ER_TUPLE_FOUND error, if a tuple with the same task_id + -- or utube name is already in the space. + -- Note that both task_id and utube indexes are unique, so there will be + -- no duplicates: each task_id can occur in the space not more than once, + -- there can be no more than one task from each utube in a space. + pcall(self.space_ready.insert, self.space_ready, {next_task[i_id], utube}) + end +end + +local function delete_ready(self, id, utube) + self.space_ready:delete(id) + put_ready(self, utube) +end + +local function update_ready(self, utube) + local prev_task = self.space_ready.index.utube:get{utube} + if prev_task ~= nil then + delete_ready(self, prev_task[1], utube) + else + put_ready(self, utube) + end +end + local function utubettl_fiber_iteration(self, processed) local now = util.time() local task = nil local estimated = util.MAX_TIMEOUT + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + if commit_requirements then + box.begin(transaction_opts) + end + -- delayed tasks task = self.space.index.watch:min(delayed_state) if task and task[i_status] == state.DELAYED then @@ -107,12 +166,25 @@ local function utubettl_fiber_iteration(self, processed) { '=', i_status, state.READY }, { '=', i_next_event, task[i_created] + task[i_ttl] } }) + + if self.ready_space then + update_ready(self, task[i_utube]) + end + if commit_requirements then + box.commit() + end + self:on_task_change(task, 'delayed') estimated = 0 processed = processed + 1 else + if commit_requirements then + box.commit() + end estimated = tonumber(task[i_next_event] - now) / 1000000 end + elseif commit_requirements then + box.commit() end -- ttl tasks @@ -131,6 +203,9 @@ local function utubettl_fiber_iteration(self, processed) end end + if commit_requirements then + box.begin(transaction_opts) + end -- ttr tasks task = self.space.index.watch:min(ttr_state) if task and task[i_status] == state.TAKEN then @@ -139,13 +214,26 @@ local function utubettl_fiber_iteration(self, processed) { '=', i_status, state.READY }, { '=', i_next_event, task[i_created] + task[i_ttl] } }) + + if self.ready_space then + update_ready(self, task[i_utube]) + end + if commit_requirements then + box.commit() + end + self:on_task_change(task, 'ttr') estimated = 0 processed = processed + 1 else + if commit_requirements then + box.commit() + end local et = tonumber(task[i_next_event] - now) / 1000000 estimated = et < estimated and et or estimated end + elseif commit_requirements then + box.commit() end if estimated > 0 or processed > 1000 then @@ -176,12 +264,11 @@ local function utubettl_fiber(self) elseif stat then processed = err end - else - -- When switching the master to the replica, the fiber will be stopped. - if self.sync_chan:get(0.1) ~= nil then - log.info("Queue utubettl fiber was stopped") - break - end + end + + if self.sync_chan:get(0.1) ~= nil then + log.info("Queue utubettl fiber was stopped") + break end end end @@ -190,9 +277,35 @@ end function tube.new(space, on_task_change, opts) validate_space(space) + local space_ready_name = space.name .. "_utubettl_ready" + local space_ready = box.space[space_ready_name] + -- Feature implemented only for memtx engine for now. + -- TODO . + local ready_space = (opts.storage_mode == storage_modes.MODE_UTUBE_READY_SPACE or false) + and opts.engine ~= 'vinyl' + if ready_space then + if not space_ready then + -- Create a space for first ready tasks from each utube. + space_ready = box.schema.create_space(space_ready_name, space_opts) + space_ready:create_index('task_id', { + type = 'tree', + parts = {1, num_type()}, + unique = true + }) + space_ready:create_index('utube', { + type = 'tree', + parts = {2, str_type()}, + unique = true + }) + else + validate_space_ready(space_ready) + end + end + on_task_change = on_task_change or (function() end) local self = setmetatable({ space = space, + space_ready = space_ready, on_task_change = function(self, task, stat_data) -- wakeup fiber if task ~= nil and self.fiber ~= nil then @@ -201,11 +314,12 @@ function tube.new(space, on_task_change, opts) on_task_change(task, stat_data) end, opts = opts, + ready_space = ready_space, }, { __index = method }) self.cond = qc.waiter() self.fiber = fiber.create(utubettl_fiber, self) - self.sync_chan = fiber.channel() + self.sync_chan = fiber.channel(1) return self end @@ -217,27 +331,19 @@ end -- put task in space function method.put(self, data, opts) - local max - - -- Taking the maximum of the index is an implicit transactions, so it is - -- always done with 'read-confirmed' mvcc isolation level. - -- It can lead to errors when trying to make parallel 'put' calls with mvcc enabled. - -- It is hapenning because 'max' for several puts in parallel will be the same since - -- read confirmed isolation level makes visible all transactions that finished the commit. - -- To fix it we wrap it with box.begin/commit and set right isolation level. - -- Current fix does not resolve that bug in situations when we already are in transaction - -- since it will open nested transactions. - -- See https://github.com/tarantool/queue/issues/207 - -- See https://www.tarantool.io/ru/doc/latest/concepts/atomic/txn_mode_mvcc/ - - if box.cfg.memtx_use_mvcc_engine and (not box.is_in_txn()) then - box.begin({txn_isolation = 'read-committed'}) - max = self.space.index.task_id:max() - box.commit() - else - max = self.space.index.task_id:max() + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} end + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + if commit_requirements then + box.begin(transaction_opts) + end + + local max = self.space.index.task_id:max() local id = max and max[i_id] + 1 or 0 local status @@ -267,6 +373,14 @@ function method.put(self, data, opts) tostring(opts.utube), data } + if self.ready_space and status == state.READY then + put_ready(self, task[i_utube]) + end + + if commit_requirements then + box.commit() + end + self:on_task_change(task, 'put') return task end @@ -293,8 +407,71 @@ function method.touch(self, id, delta) return task end +local function take_ready(self) + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + + while true do + if commit_requirements then + box.begin(transaction_opts) + end + + local task_ready = self.space_ready.index.task_id:min() + if task_ready == nil then + if commit_requirements then + box.commit() + end + return + end + + local id = task_ready[1] + local task = self.space:get(id) + + if task[i_status] == state.READY and not is_expired(task) then + local next_event = util.time() + task[i_ttr] + + local taken = self.space.index.utube:min{state.TAKEN, task[i_utube]} + + if taken == nil or taken[i_status] ~= state.TAKEN then + task = self.space:update(task[1], { + { '=', i_status, state.TAKEN }, + { '=', i_next_event, next_event } + }) + + self.space_ready:delete(id) + + if commit_requirements then + box.commit() + end + + self:on_task_change(task, 'take') + return task + end + end + + if commit_requirements then + box.commit() + end + end +end + -- take task function method.take(self) + if self.ready_space then + return take_ready(self) + end + + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + for s, t in self.space.index.status:pairs(state.READY, {iterator = 'GE'}) do if t[2] ~= state.READY then break @@ -311,8 +488,11 @@ function method.take(self) -- since it will open nested transactions. -- See https://github.com/tarantool/queue/issues/207 -- See https://www.tarantool.io/ru/doc/latest/concepts/atomic/txn_mode_mvcc/ - if box.cfg.memtx_use_mvcc_engine and (not box.is_in_txn()) then - box.begin({txn_isolation = 'read-committed'}) + -- + -- Implemented only for memtx engine for now. + -- TODO . + if not box.is_in_txn() and self.opts.engine ~= 'vinyl' then + box.begin(transaction_opts) taken = self.space.index.utube:min{state.TAKEN, t[i_utube]} box.commit() else @@ -347,11 +527,37 @@ end -- delete task function method.delete(self, id) + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + if commit_requirements then + box.begin(transaction_opts) + end + local task = self.space:get(id) if task ~= nil then local is_taken = task[i_status] == state.TAKEN self.space:delete(id) + + if self.ready_space then + if task[i_status] == state.TAKEN then + put_ready(self, task[i_utube]) + elseif task[i_status] == state.READY then + delete_ready(self, id, task[i_utube]) + end + end + task = task:transform(i_status, 1, state.DONE) + + if commit_requirements then + box.commit() + end + if is_taken then return process_neighbour(self, task, 'delete') else @@ -359,12 +565,31 @@ function method.delete(self, id) return task end end + + if commit_requirements then + box.commit() + end end -- release task function method.release(self, id, opts) + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + if commit_requirements then + box.begin(transaction_opts) + end + local task = self.space:get{id} if task == nil then + if commit_requirements then + box.commit() + end return end if opts.delay ~= nil and opts.delay > 0 then @@ -374,6 +599,15 @@ function method.release(self, id, opts) { '+', i_ttl, util.time(opts.delay) } }) if task ~= nil then + if self.ready_space then + -- We guarantee that release is called only on TAKEN tasks. + put_ready(self, task[i_utube]) + end + + if commit_requirements then + box.commit() + end + return process_neighbour(self, task, 'release') end else @@ -381,6 +615,20 @@ function method.release(self, id, opts) { '=', i_status, state.READY }, { '=', i_next_event, util.time(task[i_created] + task[i_ttl]) } }) + + if self.ready_space and task ~= nil then + -- We guarantee that release is called only on TAKEN tasks. + local inserted, err = + pcall(self.space_ready.insert, self.space_ready, {id, task[i_utube]}) + if not inserted then + require('log').warn( + 'queue: [tube "utubettl"] insert error: %s', err) + end + end + end + + if commit_requirements then + box.commit() end self:on_task_change(task, 'release') return task @@ -388,18 +636,47 @@ end -- bury task function method.bury(self, id) + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + if commit_requirements then + box.begin(transaction_opts) + end + -- The `i_next_event` should be updated because if the task has been -- "buried" after it was "taken" (and the task has "ttr") when the time in -- `i_next_event` will be interpreted as "ttl" in `utubettl_fiber_iteration` -- and the task will be deleted. local task = self.space:get{id} if task == nil then + if commit_requirements then + box.commit() + end return end + + local status = task[i_status] task = self.space:update(id, { { '=', i_status, state.BURIED }, { '=', i_next_event, task[i_created] + task[i_ttl] } }) + if self.ready_space then + local ready_task = self.space_ready:get{task[i_id]} + if ready_task ~= nil then + delete_ready(self, id, task[i_utube]) + elseif status == state.TAKEN then + put_ready(self, task[i_utube]) + end + end + + if commit_requirements then + box.commit() + end return process_neighbour( self, task:transform(i_status, 1, state.BURIED), 'bury' @@ -408,16 +685,50 @@ end -- unbury several tasks function method.kick(self, count) + local transaction_opts = {} + if box.cfg.memtx_use_mvcc_engine then + transaction_opts = {txn_isolation = 'read-committed'} + end + + -- Implemented only for memtx engine for now. + -- TODO . + local commit_requirements = not box.is_in_txn() and self.opts.engine ~= 'vinyl' + for i = 1, count do + if commit_requirements then + box.begin(transaction_opts) + end + local task = self.space.index.status:min{ state.BURIED } if task == nil then + if commit_requirements then + box.commit() + end + return i - 1 end if task[i_status] ~= state.BURIED then + if commit_requirements then + box.commit() + end + return i - 1 end task = self.space:update(task[i_id], {{ '=', i_status, state.READY }}) + if self.ready_space then + local prev_task = self.space_ready.index.utube:get{task[i_utube]} + if prev_task ~= nil then + delete_ready(self, prev_task[1], task[i_utube]) + else + put_ready(self, task[i_utube]) + end + end + + if commit_requirements then + box.commit() + end + self:on_task_change(task, 'kick') end return count @@ -435,6 +746,9 @@ end function method.truncate(self) self.space:truncate() + if self.ready_space then + self.space_ready:truncate() + end end function method.start(self) @@ -453,4 +767,12 @@ function method.stop(self) self.fiber = nil end +function method.drop(self) + self:stop() + box.space[self.space.name]:drop() + if self.ready_space then + box.space[self.space_ready.name]:drop() + end +end + return tube diff --git a/t/040-utubettl.t b/t/040-utubettl.t index 1d2c5f3..6173a9e 100755 --- a/t/040-utubettl.t +++ b/t/040-utubettl.t @@ -20,188 +20,216 @@ test:ok(rawget(box, 'space'), 'box started') test:ok(queue, 'queue is loaded') local tube = queue.create_tube('test', 'utubettl', { engine = engine }) +local tube_ready = queue.create_tube('test_ready', 'utubettl', + { engine = engine, storage_mode = "utube_ready_space" }) local tube2 = queue.create_tube('test_stat', 'utubettl', { engine = engine }) +local tube2_ready = queue.create_tube('test_stat_ready', 'utubettl', + { engine = engine, storage_mode = "utube_ready_space" }) test:ok(tube, 'test tube created') test:is(tube.name, 'test', 'tube.name') test:is(tube.type, 'utubettl', 'tube.type') test:test('Utubettl statistics', function(test) - test:plan(13) - tube2:put('stat_0') - tube2:put('stat_1') - tube2:put('stat_2') - tube2:put('stat_3') - tube2:put('stat_4') - tube2:put('stat_5', {delay=1000}) - tube2:delete(4) - tube2:take(.001) - tube2:release(0) - tube2:take(.001) - tube2:ack(0) - tube2:bury(1) - tube2:bury(2) - tube2:kick(1) - tube2:take(.001) - - local stats = queue.statistics('test_stat') - - -- check tasks statistics - test:is(stats.tasks.taken, 1, 'tasks.taken') - test:is(stats.tasks.buried, 1, 'tasks.buried') - test:is(stats.tasks.ready, 1, 'tasks.ready') - test:is(stats.tasks.done, 2, 'tasks.done') - test:is(stats.tasks.delayed, 1, 'tasks.delayed') - test:is(stats.tasks.total, 4, 'tasks.total') - - -- check function call statistics - test:is(stats.calls.delete, 1, 'calls.delete') - test:is(stats.calls.ack, 1, 'calls.ack') - test:is(stats.calls.take, 3, 'calls.take') - test:is(stats.calls.kick, 1, 'calls.kick') - test:is(stats.calls.bury, 2, 'calls.bury') - test:is(stats.calls.put, 6, 'calls.put') - test:is(stats.calls.release, 1, 'calls.release') + test:plan(26) + for _, tube_stat in ipairs({tube2, tube2_ready}) do + tube_stat:put('stat_0') + tube_stat:put('stat_1') + tube_stat:put('stat_2') + tube_stat:put('stat_3') + tube_stat:put('stat_4') + tube_stat:put('stat_5', {delay=1000}) + tube_stat:delete(4) + tube_stat:take(.001) + tube_stat:release(0) + tube_stat:take(.001) + tube_stat:ack(0) + tube_stat:bury(1) + tube_stat:bury(2) + tube_stat:kick(1) + tube_stat:take(.001) + + local stats = queue.statistics(tube_stat.name) + + -- check tasks statistics + test:is(stats.tasks.taken, 1, 'tasks.taken') + test:is(stats.tasks.buried, 1, 'tasks.buried') + test:is(stats.tasks.ready, 1, 'tasks.ready') + test:is(stats.tasks.done, 2, 'tasks.done') + test:is(stats.tasks.delayed, 1, 'tasks.delayed') + test:is(stats.tasks.total, 4, 'tasks.total') + + -- check function call statistics + test:is(stats.calls.delete, 1, 'calls.delete') + test:is(stats.calls.ack, 1, 'calls.ack') + test:is(stats.calls.take, 3, 'calls.take') + test:is(stats.calls.kick, 1, 'calls.kick') + test:is(stats.calls.bury, 2, 'calls.bury') + test:is(stats.calls.put, 6, 'calls.put') + test:is(stats.calls.release, 1, 'calls.release') + end end) test:test('Easy put/take/ack', function(test) - test:plan(12) - - test:ok(tube:put(123, {utube = 1}), 'task was put') - test:ok(tube:put(345, {utube = 1}), 'task was put') - local task = tube:take() - test:ok(task, 'task was taken') - test:is(task[2], state.TAKEN, 'task status') - test:is(task[3], 123, 'task.data') - test:ok(tube:take(.1) == nil, 'second task was not taken (the same tube)') - - task = tube:ack(task[1]) - test:ok(task, 'task was acked') - test:is(task[2], '-', 'task status') - test:is(task[3], 123, 'task.data') - - task = tube:take(.1) - test:ok(task, 'task2 was taken') - test:is(task[3], 345, 'task.data') - test:is(task[2], state.TAKEN, 'task.status') + test:plan(24) + + for _, test_tube in ipairs({tube, tube_ready}) do + test:ok(test_tube:put(123, {utube = 1}), 'task was put') + test:ok(test_tube:put(345, {utube = 1}), 'task was put') + local task = test_tube:take() + test:ok(task, 'task was taken') + test:is(task[2], state.TAKEN, 'task status') + test:is(task[3], 123, 'task.data') + test:ok(test_tube:take(.1) == nil, 'second task was not taken (the same tube)') + + task = test_tube:ack(task[1]) + test:ok(task, 'task was acked') + test:is(task[2], '-', 'task status') + test:is(task[3], 123, 'task.data') + + task = test_tube:take(.1) + test:ok(task, 'task2 was taken') + test:is(task[3], 345, 'task.data') + test:is(task[2], state.TAKEN, 'task.status') + end end) test:test('ttr put/take', function(test) - test:plan(3) + test:plan(6) local my_queue = queue.create_tube('trr_test', 'utubettl', { engine = engine }) test:ok(my_queue:put('ttr1', { ttr = 1 }), 'put ttr task') test:ok(my_queue:take(0.1) ~= nil, 'take this task') - fiber.sleep(1.1) + fiber.sleep(1.5) local task = my_queue:peek(0) test:is(task[2], state.READY, 'Ready state returned after one second') + + local my_queue_ready = queue.create_tube('trr_test_v2', 'utubettl', + { engine = engine, storage_mode = "utube_ready_space" }) + test:ok(my_queue_ready:put('ttr1', { ttr = 1 }), 'put ttr task') + test:ok(my_queue_ready:take(0.1) ~= nil, 'take this task') + fiber.sleep(1.5) + local task = my_queue_ready:peek(0) + test:is(task[2], state.READY, 'Ready state returned after one second') end) test:test('ack in utube', function(test) - test:plan(8) - - test:ok(tube:put(123, {utube = 'abc'}), 'task was put') - test:ok(tube:put(345, {utube = 'abc'}), 'task was put') - - local state = 0 - fiber.create(function() - fiber.sleep(0.1) - local taken = tube:take() - test:ok(taken, 'second task was taken') - test:is(taken[3], 345, 'task.data') - state = state + 1 - end) - - local taken = tube:take(.1) - state = 1 - test:ok(taken, 'task was taken') - test:is(taken[3], 123, 'task.data') - fiber.sleep(0.3) - test:is(state, 1, 'state was not changed') - tube:ack(taken[1]) - fiber.sleep(0.2) - test:is(state, 2, 'state was changed') + test:plan(16) + + for _, test_tube in ipairs({tube, tube_ready}) do + test:ok(test_tube:put(123, {utube = 'abc'}), 'task was put') + test:ok(test_tube:put(345, {utube = 'abc'}), 'task was put') + + local state = 0 + fiber.create(function() + fiber.sleep(0.1) + local taken = test_tube:take() + test:ok(taken, 'second task was taken') + test:is(taken[3], 345, 'task.data') + state = state + 1 + end) + + local taken = test_tube:take(.1) + state = 1 + test:ok(taken, 'task was taken') + test:is(taken[3], 123, 'task.data') + fiber.sleep(0.3) + test:is(state, 1, 'state was not changed') + test_tube:ack(taken[1]) + fiber.sleep(0.2) + test:is(state, 2, 'state was changed') + end end) test:test('bury in utube', function(test) - test:plan(8) - - test:ok(tube:put(567, {utube = 'cde'}), 'task was put') - test:ok(tube:put(789, {utube = 'cde'}), 'task was put') - - local state = 0 - fiber.create(function() - fiber.sleep(0.1) - local taken = tube:take() - test:ok(taken, 'second task was taken') - test:is(taken[3], 789, 'task.data') - state = state + 1 - end) - - local taken = tube:take(.1) - state = 1 - test:ok(taken, 'task was taken') - test:is(taken[3], 567, 'task.data') - fiber.sleep(0.3) - test:is(state, 1, 'state was not changed') - tube:bury(taken[1]) - fiber.sleep(0.2) - test:is(state, 2, 'state was changed') + test:plan(16) + + for _, test_tube in ipairs({tube, tube_ready}) do + test:ok(test_tube:put(567, {utube = 'cde'}), 'task was put') + test:ok(test_tube:put(789, {utube = 'cde'}), 'task was put') + + local state = 0 + fiber.create(function() + fiber.sleep(0.1) + local taken = test_tube:take() + test:ok(taken, 'second task was taken') + test:is(taken[3], 789, 'task.data') + state = state + 1 + end) + + local taken = test_tube:take(.1) + state = 1 + test:ok(taken, 'task was taken') + test:is(taken[3], 567, 'task.data') + fiber.sleep(0.3) + test:is(state, 1, 'state was not changed') + test_tube:bury(taken[1]) + fiber.sleep(0.2) + test:is(state, 2, 'state was changed') + end end) test:test('instant bury', function(test) - test:plan(1) + test:plan(2) tube:put(1, {ttr=60}) local taken = tube:take(.1) test:is(tube:bury(taken[1])[2], '!', 'task is buried') + + tube_ready:put(1, {ttr=60}) + taken = tube_ready:take(.1) + test:is(tube_ready:bury(taken[1])[2], '!', 'task is buried') end) test:test('release in utube', function(test) - test:plan(8) - - test:ok(tube:put(678, {utube = 'def'}), 'task was put') - test:ok(tube:put(890, {utube = 'def'}), 'task was put') - - local state = 0 - fiber.create(function() - fiber.sleep(0.1) - local taken = tube:take() - test:ok(taken, 'first task was taken again') + test:plan(16) + + for _, test_tube in ipairs({tube, tube_ready}) do + test:ok(test_tube:put(678, {utube = 'def'}), 'task was put') + test:ok(test_tube:put(890, {utube = 'def'}), 'task was put') + + local state = 0 + fiber.create(function() + fiber.sleep(0.1) + local taken = test_tube:take() + test:ok(taken, 'first task was taken again') + test:is(taken[3], 678, 'task.data') + state = state + 1 + end) + + local taken = test_tube:take(.1) + state = 1 + test:ok(taken, 'task was taken ' .. taken[1]) test:is(taken[3], 678, 'task.data') - state = state + 1 - end) - - local taken = tube:take(.1) - state = 1 - test:ok(taken, 'task was taken ' .. taken[1]) - test:is(taken[3], 678, 'task.data') - fiber.sleep(0.3) - test:is(state, 1, 'state was not changed') - tube:release(taken[1]) - fiber.sleep(0.2) - test:is(state, 2, 'state was changed') + fiber.sleep(0.3) + test:is(state, 1, 'state was not changed') + test_tube:release(taken[1]) + fiber.sleep(0.2) + test:is(state, 2, 'state was changed') + end end) test:test('release[delay] in utube', function(test) - test:plan(8) - - test:ok(tube:put(789, {utube = 'efg'}), 'task was put') - test:ok(tube:put(901, {utube = 'efg'}), 'task was put') - - local state = 0 - fiber.create(function() - fiber.sleep(0.1) - local taken = tube:take() - test:ok(taken, 'second task was taken') - test:is(taken[3], 901, 'task.data') - state = state + 1 - end) - - local taken = tube:take(.1) - state = 1 - test:ok(taken, 'task was taken ' .. taken[1]) - test:is(taken[3], 789, 'task.data') - fiber.sleep(0.3) - test:is(state, 1, 'state was not changed') - tube:release(taken[1], { delay = 10 }) -- - fiber.sleep(0.2) - test:is(state, 2, 'state was changed') + test:plan(16) + + for _, test_tube in ipairs({tube, tube_ready}) do + test:ok(test_tube:put(789, {utube = 'efg'}), 'task was put') + test:ok(test_tube:put(901, {utube = 'efg'}), 'task was put') + + local state = 0 + fiber.create(function() + fiber.sleep(0.1) + local taken = test_tube:take() + test:ok(taken, 'second task was taken') + test:is(taken[3], 901, 'task.data') + state = state + 1 + end) + + local taken = test_tube:take(.1) + state = 1 + test:ok(taken, 'task was taken ' .. taken[1]) + test:is(taken[3], 789, 'task.data') + fiber.sleep(0.3) + test:is(state, 1, 'state was not changed') + test_tube:release(taken[1], { delay = 10 }) + fiber.sleep(0.2) + test:is(state, 2, 'state was changed') + end end) test:test('if_not_exists test', function(test)