From 1e1645c94304fa1657d49e3346b33405c4afb952 Mon Sep 17 00:00:00 2001 From: DerekBum Date: Wed, 8 May 2024 22:30:15 +0300 Subject: [PATCH] utubettl: fix slow take on busy utubes If some of the utubettl for tasks at the top of the queue were busy most of the time, `take` would slow down for every other task. This problem is fixed by creating a new space `space_ready`. It contains first task with `READY` status from each utube. This solution shows great results for the stated problem, with the cost of slowing the `put` method (it is ~3 times slower). Thus, this workaround is disabled by default. To enable it, user should set the `storage_mode = queue.driver.utubettl.STORAGE_MODE_UTUBE_READY_SPACE` as an option while creating the tube. As example: ```lua local test_queue = queue.create_tube('test_queue', 'utubettl', {temporary = true, storage_mode = queue.driver.utubettl.STORAGE_MODE_UTUBE_READY_SPACE}) ``` Closes #228 --- CHANGELOG.md | 10 +- README.md | 36 +++ queue/abstract/driver/utubettl.lua | 401 +++++++++++++++++++++++++++-- t/040-utubettl.t | 371 ++++++++++++++++---------- t/050-ttl.t | 32 ++- 5 files changed, 677 insertions(+), 173 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 69fe51db..e8bfe794 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 0dca442e..c1182076 100644 --- a/README.md +++ b/README.md @@ -244,6 +244,10 @@ in strict FIFO order. This queue type is effectively a combination of `fifottl` and `utube`. +It is advised not to use `utubettl` methods inside transactions with +`read-confirmed` isolation level. It can lead to errors when trying to make +parallel tube methods calls with mvcc enabled. + The following options can be specified when creating a `utubettl` queue: * `temporary` - boolean - if true, the contents of the queue do not persist on disk @@ -251,6 +255,38 @@ on disk already exists * `on_task_change` - function name - a callback to be executed on every operation + * `storage_mode` - string - one of + * `queue.driver.utubettl.STORAGE_MODE_DEFAULT` ("default") - default + implementation of `utubettl` + * `queue.driver.utubettl.STORAGE_MODE_UTUBE_READY_SPACE` + ("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. + WARNING: this is an experimental storage mode. + + Here is a benchmark comparison of these two modes: + * Benchmark for simple `put` and `take` methods. 30k utubes are created + with a single task each. Task creation time is calculated. After that + 30k consumers are calling `take` + `ack`, each in the separate fiber. + Time to ack all tasks is calculated. The results are as follows: + + | | put (30k) | take+ack | + |---------|-----------|----------| + | default | 200ms | 1.7s | + | ready | 320ms | 1.8s | + * Benchmark for the busy utubes. 10 tubes are created. + Each contains 1000 tasks. After that, 10 consumers are created (each works + on his tube only, one tube — one consumer). Each consumer will + `take`, then `yield` and then `ack` every task from their utube + (1000 tasks each). + After that, we can also run this benchmark with 10k tasks on each utube, + 100k tasks and 140k tasks. But all that with 10 utubes and 10 consumers. + The results are as follows: + + | | 1k | 10k | 50k | 140k | + |---------|-------|------|------|-------| + | default | 80s | 1.6h | 100h | 1000h | + | ready | 520ms | 5.4s | 28s | 83s | 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 51b0b612..19004c12 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/queue/abstract/driver/utubettl.lua @@ -11,6 +11,9 @@ local str_type = qc.str_type local tube = {} local method = {} +tube.STORAGE_MODE_DEFAULT = "default" +tube.STORAGE_MODE_UTUBE_READY_SPACE = "utube_ready_space" + local i_id = 1 local i_status = 2 local i_next_event = 3 @@ -38,6 +41,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 +109,65 @@ local delayed_state = { state.DELAYED } local ttl_states = { state.READY, state.BURIED } local ttr_state = { state.TAKEN } +local function put_next_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 put_ready(self, id, utube) + local taken = self.space.index.utube:min{state.TAKEN, utube} + if taken == nil or taken[i_status] ~= state.TAKEN then + -- 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, {id, utube}) + end +end + +local function delete_ready(self, id, utube) + self.space_ready:delete(id) + put_next_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_next_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. + -- https://github.com/tarantool/queue/issues/230. + 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 +176,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 +213,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 +224,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 +274,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 +287,50 @@ end function tube.new(space, on_task_change, opts) validate_space(space) + local space_ready_name = space.name .. "_utubettl_ready_buffer" + local space_ready = box.space[space_ready_name] + -- Feature implemented only for memtx engine for now. + -- https://github.com/tarantool/queue/issues/230. + if opts.storage_mode == tube.STORAGE_MODE_UTUBE_READY_SPACE and opts.engine == 'vinyl' then + error(string.format('"%s" storage mode cannot be used with vinyl engine', + tube.STORAGE_MODE_UTUBE_READY_SPACE)) + end + + local ready_space = (opts.storage_mode == tube.STORAGE_MODE_UTUBE_READY_SPACE or false) + if ready_space then + if not space_ready then + local space_opts = {} + space_opts.temporary = opts.temporary or false + space_opts.engine = opts.engine or 'memtx' + space_opts.format = { + {name = 'task_id', type = num_type()}, + {name = 'utube', type = str_type()} + } + + -- 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) + if space:len() == 0 then + space_ready:truncate() + end + 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 +339,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 +356,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. + -- https://github.com/tarantool/queue/issues/230. + 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 +398,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_id], task[i_utube]) + end + + if commit_requirements then + box.commit() + end + self:on_task_change(task, 'put') return task end @@ -293,8 +432,69 @@ 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 + + local commit_requirements = not box.is_in_txn() + + 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 +511,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. + -- https://github.com/tarantool/queue/issues/230. + 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 +550,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. + -- https://github.com/tarantool/queue/issues/230. + 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_next_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 +588,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. + -- https://github.com/tarantool/queue/issues/230. + 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 +622,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_next_ready(self, task[i_utube]) + end + + if commit_requirements then + box.commit() + end + return process_neighbour(self, task, 'release') end else @@ -381,6 +638,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 +659,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. + -- https://github.com/tarantool/queue/issues/230. + 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_next_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 +708,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. + -- https://github.com/tarantool/queue/issues/230. + 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_next_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 +769,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 +790,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 1d2c5f37..d64d0498 100755 --- a/t/040-utubettl.t +++ b/t/040-utubettl.t @@ -21,187 +21,280 @@ test:ok(queue, 'queue is loaded') local tube = queue.create_tube('test', 'utubettl', { engine = engine }) local tube2 = queue.create_tube('test_stat', 'utubettl', { engine = engine }) +local tube_ready, tube2_ready +if engine ~= 'vinyl' then + tube_ready = queue.create_tube('test_ready', 'utubettl', + { engine = engine, storage_mode = queue.driver.utubettl.STORAGE_MODE_UTUBE_READY_SPACE }) + tube2_ready = queue.create_tube('test_stat_ready', 'utubettl', + { engine = engine, storage_mode = queue.driver.utubettl.STORAGE_MODE_UTUBE_READY_SPACE }) +end 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') + if engine ~= 'vinyl' then + test:plan(26) + else + test:plan(13) + end + for _, tube_stat in ipairs({tube2, tube2_ready}) do + if tube_stat == nil then + break + end + + 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) + if engine ~= 'vinyl' then + test:plan(24) + else + test:plan(12) + end - 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') + for _, test_tube in ipairs({tube, tube_ready}) do + if test_tube == nil then + break + end + + 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) + if engine ~= 'vinyl' then + test:plan(6) + else + test:plan(3) + end + 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') + + if engine ~= 'vinyl' then + local my_queue_ready = queue.create_tube('trr_test_v2', 'utubettl', + { engine = engine, storage_mode = queue.driver.utubettl.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 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') + if engine ~= 'vinyl' then + test:plan(16) + else + test:plan(8) + end - 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) + for _, test_tube in ipairs({tube, tube_ready}) do + if test_tube == nil then + break + 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: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') + if engine ~= 'vinyl' then + test:plan(16) + else + test:plan(8) + end - 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) + for _, test_tube in ipairs({tube, tube_ready}) do + if test_tube == nil then + break + 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: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) + if engine ~= 'vinyl' then + test:plan(2) + else + test:plan(1) + end + tube:put(1, {ttr=60}) local taken = tube:take(.1) test:is(tube:bury(taken[1])[2], '!', 'task is buried') + + if tube_ready ~= nil then + tube_ready:put(1, {ttr=60}) + taken = tube_ready:take(.1) + test:is(tube_ready:bury(taken[1])[2], '!', 'task is buried') + end end) test:test('release in utube', function(test) - test:plan(8) + if engine ~= 'vinyl' then + test:plan(16) + else + test:plan(8) + end - test:ok(tube:put(678, {utube = 'def'}), 'task was put') - test:ok(tube:put(890, {utube = 'def'}), 'task was put') + for _, test_tube in ipairs({tube, tube_ready}) do + if test_tube == nil then + break + end - local state = 0 - fiber.create(function() - fiber.sleep(0.1) - local taken = tube:take() - test:ok(taken, 'first task was taken again') + 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') + if engine ~= 'vinyl' then + test:plan(16) + else + test:plan(8) + end - 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) + for _, test_tube in ipairs({tube, tube_ready}) do + if test_tube == nil then + break + 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: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) diff --git a/t/050-ttl.t b/t/050-ttl.t index 0695ffba..62231c17 100755 --- a/t/050-ttl.t +++ b/t/050-ttl.t @@ -2,7 +2,7 @@ local fiber = require('fiber') local test = require('tap').test() -test:plan(5) +test:plan(7) local queue = require('queue') @@ -55,6 +55,21 @@ test:test('one message per queue utttl', function (test) end end) +test:test('one message per queue utttl_ready', function (test) + if engine ~= 'vinyl' then + return + end + + test:plan(20) + local tube = queue.create_tube('ompq_utttl_ready', 'utubettl', + { engine = engine, storage_mode = queue.driver.utubettl.MODE_UTUBE_READY_SPACE }) + for i = 1, 20 do + tube:put('ompq_' .. i, {ttl=ttl}) + + test_take_after_ttl(test, tube, ttl) + end +end) + test:test('many messages, one queue ffttl', function (test) test:plan(20) for i = 1, 20 do @@ -75,6 +90,21 @@ test:test('many messages, one queue utttl', function (test) end end) +test:test('many messages, one queue utttl_ready', function (test) + if engine ~= 'vinyl' then + return + end + + test:plan(20) + for i = 1, 20 do + local tube = queue.create_tube('mmpq_utttl_ready_' .. i, 'utubettl', + { engine = engine, storage_mode = queue.driver.utubettl.MODE_UTUBE_READY_SPACE }) + tube:put('mmpq_' .. i, {ttl=ttl}) + + test_take_after_ttl(test, tube, ttl) + end +end) + tnt.finish() os.exit(test:check() and 0 or 1) -- vim: set ft=lua :