From 293aae0ba8b23bc3792a2751fd7ba4627bc6ef62 Mon Sep 17 00:00:00 2001 From: DerekBum Date: Sun, 28 Apr 2024 22:25:29 +0300 Subject: [PATCH] utube: fix slow take on busy utubes If some of the utube 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.utube.STORAGE_MODE_UTUBE_READY_SPACE` as an option while creating the tube. As example: ```lua local test_queue = queue.create_tube('test_queue', 'utube', {temporary = true, storage_mode = queue.driver.utube.STORAGE_MODE_UTUBE_READY_SPACE}) ``` Part of #228 --- CHANGELOG.md | 7 + README.md | 36 ++++ queue/abstract.lua | 6 +- queue/abstract/driver/utube.lua | 312 +++++++++++++++++++++++++++++--- t/030-utube.t | 249 +++++++++++++++---------- t/benchmark/busy_utubes.lua | 90 +++++++++ t/benchmark/many_utubes.lua | 86 +++++++++ 7 files changed, 667 insertions(+), 119 deletions(-) create mode 100644 t/benchmark/busy_utubes.lua create mode 100644 t/benchmark/many_utubes.lua diff --git a/CHANGELOG.md b/CHANGELOG.md index 0c39331a..69fe51db 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,10 +7,17 @@ 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. + ### 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. ## [1.3.3] - 2023-09-13 diff --git a/README.md b/README.md index b156944f..0dca442e 100644 --- a/README.md +++ b/README.md @@ -165,6 +165,10 @@ The main idea of this queue backend is the same as in a `fifo` queue: the tasks are executed in FIFO order. However, tasks may be grouped into sub-queues. +It is advised not to use `utube` 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 `utube` queue: * `temporary` - boolean - if true, the contents of the queue do not persist on disk @@ -172,6 +176,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.utube.STORAGE_MODE_DEFAULT` ("default") - default + implementation of `utube` + * `queue.driver.utube.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 | 180ms | 1.6s | + | ready | 270ms | 1.7s | + * 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 150k tasks. But all that with 10 utubes and 10 consumers. + The results are as follows: + + | | 1k | 10k | 50k | 150k | + |---------|-------|------|------|-------| + | default | 53s | 1.5h | 100h | 1000h | + | ready | 450ms | 4.7s | 26s | 72s | The following options can be specified when putting a task in a `utube` queue: diff --git a/queue/abstract.lua b/queue/abstract.lua index d8dcff88..01010174 100644 --- a/queue/abstract.lua +++ b/queue/abstract.lua @@ -297,7 +297,11 @@ function tube.drop(self) local space_name = tube[3] - box.space[space_name]:drop() + if self.raw.drop ~= nil then + self.raw:drop() + else + box.space[space_name]:drop() + end box.space._queue:delete{tube_name} -- drop queue queue.tube[tube_name] = nil diff --git a/queue/abstract/driver/utube.lua b/queue/abstract/driver/utube.lua index d3b7e006..c1dd912d 100644 --- a/queue/abstract/driver/utube.lua +++ b/queue/abstract/driver/utube.lua @@ -5,6 +5,9 @@ local str_type = require('queue.compat').str_type local tube = {} local method = {} +tube.STORAGE_MODE_DEFAULT = "default" +tube.STORAGE_MODE_UTUBE_READY_SPACE = "utube_ready_space" + local i_status = 2 -- validate space of queue @@ -19,6 +22,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) local space_opts = {} @@ -57,13 +72,56 @@ function tube.create_space(space_name, opts) end -- start tube on space -function tube.new(space, on_task_change) +function tube.new(space, on_task_change, opts) validate_space(space) + 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()} + } + + local space_ready_name = space.name .. "_utube_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) + 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) + 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 = on_task_change, + ready_space = ready_space, + opts = opts, }, { __index = method }) return self end @@ -73,37 +131,120 @@ function method.normalize_task(self, task) return task and task:transform(3, 1) end +local function put_next_ready(self, utube) + local taken = self.space.index.utube:min{state.TAKEN, utube} + if taken == nil or taken[2] ~= state.TAKEN then + local next_task = self.space.index.utube:min{state.READY, utube} + if next_task == nil or next_task[2] ~= 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[1], 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[2] ~= 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 + -- 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 + + local commit_requirements = not box.is_in_txn() + if commit_requirements then + box.begin(transaction_opts) end + local max = self.space.index.task_id:max() + local id = max and max[1] + 1 or 0 local task = self.space:insert{id, state.READY, tostring(opts.utube), data} + if self.ready_space then + put_ready(self, task[1], task[3]) + end + + if commit_requirements then + box.commit() + end + self.on_task_change(task, 'put') 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[2] == state.READY then + local taken = self.space.index.utube:min{state.TAKEN, task[3]} + + if taken == nil or taken[2] ~= state.TAKEN then + task = self.space:update(id, { { '=', 2, state.TAKEN } }) + + 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, task in self.space.index.status:pairs(state.READY, { iterator = 'GE' }) do if task[2] ~= state.READY then @@ -120,8 +261,8 @@ 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'}) + if not box.is_in_txn() then + box.begin(transaction_opts) taken = self.space.index.utube:min{state.TAKEN, task[3]} box.commit() else @@ -141,41 +282,133 @@ function method.touch(self, id, ttr) error('utube queue does not support touch') end +local function delete_ready(self, id, utube) + self.space_ready:delete(id) + put_next_ready(self, utube) +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 + + local commit_requirements = not box.is_in_txn() + if commit_requirements then + box.begin(transaction_opts) + end + local task = self.space:get(id) self.space:delete(id) if task ~= nil then + if self.ready_space then + if task[2] == state.TAKEN then + put_next_ready(self, task[3]) + elseif task[2] == state.READY then + delete_ready(self, id, task[3]) + end + end + task = task:transform(2, 1, state.DONE) local neighbour = self.space.index.utube:min{state.READY, task[3]} + + if commit_requirements then + box.commit() + end + self.on_task_change(task, 'delete') if neighbour then self.on_task_change(neighbour) end + return task + end + + if commit_requirements then + box.commit() end return task 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 + + local commit_requirements = not box.is_in_txn() + if commit_requirements then + box.begin(transaction_opts) + end + local task = self.space:update(id, {{ '=', 2, state.READY }}) if task ~= nil then + if self.ready_space then + -- We guarantee that release is called only on TAKEN tasks. + local inserted, err = + pcall(self.space_ready.insert, self.space_ready, {id, task[3]}) + if not inserted then + require('log').warn( + 'queue: [tube "utube"] insert error: %s', err) + end + end + + if commit_requirements then + box.commit() + end + self.on_task_change(task, 'release') + return task + end + + if commit_requirements then + box.commit() end return task 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 + + local commit_requirements = not box.is_in_txn() + if commit_requirements then + box.begin(transaction_opts) + end + + local current_task = self.space:get{id} local task = self.space:update(id, {{ '=', 2, state.BURIED }}) if task ~= nil then + if self.ready_space then + local status = current_task[2] + local ready_task = self.space_ready:get{task[1]} + if ready_task ~= nil then + delete_ready(self, id, task[3]) + elseif status == state.TAKEN then + put_next_ready(self, task[3]) + end + end + local neighbour = self.space.index.utube:min{state.READY, task[3]} + + if commit_requirements then + box.commit() + end + self.on_task_change(task, 'bury') if neighbour and neighbour[i_status] == state.READY then self.on_task_change(neighbour) end else + if commit_requirements then + box.commit() + end + self.on_task_change(task, 'bury') end return task @@ -183,7 +416,18 @@ 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 + + local commit_requirements = not box.is_in_txn() + 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 return i - 1 @@ -193,6 +437,19 @@ function method.kick(self, count) end task = self.space:update(task[1], {{ '=', 2, state.READY }}) + if self.ready_space then + local prev_task = self.space_ready.index.utube:get{task[3]} + if prev_task ~= nil then + delete_ready(self, prev_task[1], task[3]) + else + put_ready(self, task[3]) + end + end + + if commit_requirements then + box.commit() + end + self.on_task_change(task, 'kick') end return count @@ -210,6 +467,9 @@ end function method.truncate(self) self.space:truncate() + if self.ready_space then + self.space_ready:truncate() + end end -- This driver has no background activity. @@ -222,4 +482,12 @@ function method.stop() return 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/030-utube.t b/t/030-utube.t index ffd92136..5f59407d 100755 --- a/t/030-utube.t +++ b/t/030-utube.t @@ -18,125 +18,182 @@ test:ok(queue, 'queue is loaded') local tube = queue.create_tube('test', 'utube', { engine = engine }) local tube2 = queue.create_tube('test_stat', 'utube', { engine = engine }) +local tube_ready, tube2_ready +if engine ~= 'vinyl' then + tube_ready = queue.create_tube('test_ready', 'utube', + { engine = engine, storage_mode = queue.driver.utube.STORAGE_MODE_UTUBE_READY_SPACE }) + tube2_ready = queue.create_tube('test_stat_ready', 'utube', + { engine = engine, storage_mode = queue.driver.utube.STORAGE_MODE_UTUBE_READY_SPACE }) +end test:ok(tube, 'test tube created') test:is(tube.name, 'test', 'tube.name') test:is(tube.type, 'utube', 'tube.type') test:test('Utube 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: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, 0, 'tasks.delayed') - test:is(stats.tasks.total, 3, '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, 5, '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: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, 0, 'tasks.delayed') + test:is(stats.tasks.total, 3, '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, 5, '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') + if engine ~= 'vinyl' then + test:plan(24) + else + test:plan(12) + end + + 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('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}) + local taken = tube_ready:take(.1) + test:is(tube_ready:bury(taken[1])[2], '!', 'task is buried') + end end) test:test('if_not_exists test', function(test) diff --git a/t/benchmark/busy_utubes.lua b/t/benchmark/busy_utubes.lua new file mode 100644 index 00000000..b089289c --- /dev/null +++ b/t/benchmark/busy_utubes.lua @@ -0,0 +1,90 @@ +#!/usr/bin/env tarantool + +local clock = require('clock') +local os = require('os') +local fiber = require('fiber') +local queue = require('queue') + +-- Set the number of consumers. +local consumers_count = 10 +-- Set the number of tasks processed by one consumer per iteration. +local batch_size = 150000 + +local barrier = fiber.cond() +local wait_count = 0 + +box.cfg() + +local test_queue = queue.create_tube('test_queue', 'utube', + {temporary = true, storage_mode = queue.driver.utube.STORAGE_MODE_UTUBE_READY_SPACE}) + +local function prepare_tasks() + local test_data = 'test data' + + for i = 1, consumers_count do + for _ = 1, batch_size do + test_queue:put(test_data, {utube = tostring(i)}) + end + end +end + +local function prepare_consumers() + local consumers = {} + + for i = 1, consumers_count do + consumers[i] = fiber.create(function() + wait_count = wait_count + 1 + -- Wait for all consumers to start. + barrier:wait() + + -- Ack the tasks. + for _ = 1, batch_size do + local task = test_queue:take() + fiber.yield() + test_queue:ack(task[1]) + end + + wait_count = wait_count + 1 + end) + end + + return consumers +end + +local function multi_consumer_bench() + --- Wait for all consumer fibers. + local wait_all = function() + while (wait_count ~= consumers_count) do + fiber.yield() + end + wait_count = 0 + end + + fiber.set_max_slice(100) + + prepare_tasks() + + -- Wait for all consumers to start. + local consumers = prepare_consumers() + wait_all() + + -- Start timing of task confirmation. + local start_ack_time = clock.proc64() + barrier:broadcast() + -- Wait for all tasks to be acked. + wait_all() + -- Complete the timing of task confirmation. + local complete_time = clock.proc64() + + -- Print the result in milliseconds. + print(string.format("Time it takes to confirm the tasks: %i", + tonumber((complete_time - start_ack_time) / 10^6))) +end + +-- Start benchmark. +multi_consumer_bench() + +-- Cleanup. +test_queue:drop() + +os.exit(0) diff --git a/t/benchmark/many_utubes.lua b/t/benchmark/many_utubes.lua new file mode 100644 index 00000000..20c39780 --- /dev/null +++ b/t/benchmark/many_utubes.lua @@ -0,0 +1,86 @@ +#!/usr/bin/env tarantool + +local clock = require('clock') +local os = require('os') +local fiber = require('fiber') +local queue = require('queue') + +-- Set the number of consumers. +local consumers_count = 30000 + +local barrier = fiber.cond() +local wait_count = 0 + +box.cfg() + +local test_queue = queue.create_tube('test_queue', 'utube', + {temporary = true, storage_mode = queue.driver.utube.STORAGE_MODE_UTUBE_READY_SPACE}) + +local function prepare_tasks() + local test_data = 'test data' + + for i = 1, consumers_count do + test_queue:put(test_data, {utube = tostring(i)}) + end +end + +local function prepare_consumers() + local consumers = {} + + for i = 1, consumers_count do + consumers[i] = fiber.create(function() + wait_count = wait_count + 1 + -- Wait for all consumers to start. + barrier:wait() + + -- Ack the task. + local task = test_queue:take() + test_queue:ack(task[1]) + + wait_count = wait_count + 1 + end) + end + + return consumers +end + +local function multi_consumer_bench() + --- Wait for all consumer fibers. + local wait_all = function() + while (wait_count ~= consumers_count) do + fiber.yield() + end + wait_count = 0 + end + + fiber.set_max_slice(100) + + -- Wait for all consumers to start. + local consumers = prepare_consumers() + wait_all() + + -- Start timing creation of tasks. + local start_put_time = clock.proc64() + prepare_tasks() + -- Start timing of task confirmation. + local start_ack_time = clock.proc64() + barrier:broadcast() + -- Wait for all tasks to be acked. + wait_all() + -- Complete the timing of task confirmation. + local complete_time = clock.proc64() + + -- Print results in milliseconds. + print(string.format("Time it takes to fill the queue: %i", + tonumber((start_ack_time - start_put_time) / 10^6))) + print(string.format("Time it takes to confirm the tasks: %i", + tonumber((complete_time - start_ack_time) / 10^6))) +end + +-- Start benchmark. +multi_consumer_bench() + +-- Cleanup. +test_queue:drop() + +os.exit(0)