diff --git a/CHANGELOG.md b/CHANGELOG.md index 50e89c1..323d4bf 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,11 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added -### Changed +- Allow pass options to `take()` via driver API. +- Optional `consumer_group(opts, task)` driver API for routing waiting + consumers to matching tasks. +- Support extra statistics from driver API. +- Add the `subqueuettl` driver for processing tasks from named subqueues in one tube space. ### Fixed diff --git a/README.md b/README.md index 550c6bb..17baa2c 100644 --- a/README.md +++ b/README.md @@ -16,6 +16,7 @@ align="right"> * [fifottl \- a simple priority queue with support for task time to live](#fifottl---a-simple-priority-queue-with-support-for-task-time-to-live) * [utube \- a queue with sub\-queues inside](#utube---a-queue-with-sub-queues-inside) * [utubettl \- extension of utube to support ttl](#utubettl---extension-of-utube-to-support-ttl) + * [subqueuettl \- TTL subqueues in one space](#subqueuettl---ttl-subqueues-in-one-space) * [The underlying spaces](#the-underlying-spaces) * [Fields of the \_queue space](#fields-of-the-_queue-space) * [Fields of the \_queue\_consumers space](#fields-of-the-_queue_consumers-space) @@ -302,6 +303,41 @@ seconds; if `ttr` is not specified, it is set to the same as `ttl` is changed to 'ready' so another worker may take it) * `delay` - time to wait before starting to execute the task, in seconds +## `subqueuettl` - TTL subqueues in one space + +`subqueuettl` stores independent TTL subqueues in one Tarantool space. It +does not provide an operation to take a task from all subqueues. The driver +supports the `memtx` engine only; `vinyl` is not supported. + +The following options can be specified when putting a task in a +`subqueuettl` queue: + + * `subqueue` - required name of the subqueue. + * `pri` - task priority (`0` is the highest priority and is the default). + * `ttl` - numeric time to live in seconds. If omitted, it is set to infinity. + * `ttr` - numeric time allotted to process a task in seconds. If omitted, it + is set to the same value as `ttl`. + * `delay` - time in seconds to wait before a task becomes ready. + +Both `put()` and `take()` require `opts.subqueue`: + +```lua +local tube = queue.create_tube('sites', 'subqueuettl') +tube:put('https://example.com', {subqueue = 'example.com'}) +local task = tube:take(10, {subqueue = 'example.com'}) +``` + +`subqueuettl` adds per-subqueue statistics under the common `driver` field: + +```lua +local stats = queue.statistics('sites') +local example_stats = stats.driver.subqueues['example.com'] +``` + +Each subqueue entry contains `ready`, `taken`, `buried`, `delayed`, and `total` +task counts. The subqueue name registry is best-effort; a registry write error +does not fail the queue operation, and a stale name can remain with zero counts. + # The underlying spaces The queue system consists of fibers, IPC channels, functions, and spaces. @@ -647,7 +683,7 @@ or it may be acted on by a worker (usually with a `take` request). ## Taking a task from the queue ("consuming") ```lua -queue.tube.tube_name:take([timeout]) +queue.tube.tube_name:take([timeout [, {options} ]]) ``` Take a queue task. @@ -660,6 +696,10 @@ than any other tuple which also has `task_state` = 'r'. If there is no such task, and timeout was specified, then the job waits until a task becomes ready or the timeout expires. +The options, if specified, must be one or more of the options described +above +(`subqueue`, required name of the subqueue to take a task from). + Effect: the value of `task_state` changes to 't' (taken). The `take` request tells the system that the task is being worked on. It should be followed by an `ack` request when the work is finished. @@ -825,7 +865,11 @@ queue.statistics( [queue name] ) Show the number of tasks in a queue broken down by `task_state`, and the number of requests broken down by the type of request. If the queue name is not -specified, show these numbers for all queues. +specified, show these numbers for all queues. + +In addition, any driver can add extra information for statistics by implementing their own `statistics` method. +The result of this method will be added as additional key of the returned value `queue.statistics`. + Statistics are temporary, they are reset whenever the Tarantool server restarts. Example: @@ -856,6 +900,14 @@ queue.statistics('list_of_sites') bury: 1 put: 2 delete: 1 + extra: + subqueues: + example.com: + ready: 0 + taken: 0 + buried: 0 + delayed: 0 + total: 0 ... ``` @@ -1003,7 +1055,8 @@ API: which is passed on to the user (removes the administrative fields) * `tube:put(data[, opts])` - puts a task into the queue. Returns a normalized task which represents a tuple in the space -* `tube:take()` - sets the task state to 'in progress' and returns the task. +* `tube:take([opts])` - sets the task state to 'in progress' and returns the + task. `subqueuettl` requires `opts.subqueue`. If there are no 'ready' tasks in the queue, returns nil. * `tube:delete(task_id)` - deletes a task from the queue. Returns the original task with a state changed to 'done' diff --git a/queue-scm-1.rockspec b/queue-scm-1.rockspec index 73122a9..a563fcb 100644 --- a/queue-scm-1.rockspec +++ b/queue-scm-1.rockspec @@ -22,6 +22,7 @@ build = { ['queue.abstract.queue_state'] = 'queue/abstract/queue_state.lua', ['queue.abstract.driver.fifottl'] = 'queue/abstract/driver/fifottl.lua', ['queue.abstract.driver.utubettl'] = 'queue/abstract/driver/utubettl.lua', + ['queue.abstract.driver.subqueuettl'] = 'queue/abstract/driver/subqueuettl.lua', ['queue.abstract.driver.fifo'] = 'queue/abstract/driver/fifo.lua', ['queue.abstract.driver.utube'] = 'queue/abstract/driver/utube.lua', ['queue.abstract.driver.limfifottl'] = 'queue/abstract/driver/limfifottl.lua', diff --git a/queue/CMakeLists.txt b/queue/CMakeLists.txt index 9234cb6..c374412 100644 --- a/queue/CMakeLists.txt +++ b/queue/CMakeLists.txt @@ -22,5 +22,7 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/fifottl.lua DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/) install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/utubettl.lua DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/) +install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/subqueuettl.lua + DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/) install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/limfifottl.lua DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/) diff --git a/queue/abstract.lua b/queue/abstract.lua index 45051cb..4b8469b 100644 --- a/queue/abstract.lua +++ b/queue/abstract.lua @@ -107,14 +107,15 @@ function tube.put(self, data, opts) end local conds = {} +local CONSUMER_GROUP_ANY = '' local releasing_connections = {} -function tube.take(self, timeout) +function tube.take(self, timeout, opts) if not check_state("take") then return nil end timeout = util.time(timeout or util.TIMEOUT_INFINITY) - local task = self.raw:take() + local task = self.raw:take(opts) if task ~= nil then return self.raw:normalize_task(task) end @@ -125,8 +126,14 @@ function tube.take(self, timeout) local tid = self.tube_id local fid = fiber.id() local conn_id = connection.id() + local consumer_group = CONSUMER_GROUP_ANY + if self.raw.consumer_group ~= nil then + consumer_group = self.raw:consumer_group(opts) + end - box.space._queue_consumers:insert{conn_id, fid, tid, time, started} + box.space._queue_consumers:insert{ + conn_id, fid, tid, time, started, consumer_group + } conds[fid] = qc.waiter() conds[fid]:wait(tonumber(timeout) / 1000000) conds[fid]:free() @@ -139,7 +146,7 @@ function tube.take(self, timeout) return nil end - task = self.raw:take() + task = self.raw:take(opts) if task ~= nil then return self.raw:normalize_task(task) @@ -453,10 +460,16 @@ local function make_self(driver, space, tube_name, tube_type, tube_id, opts) -- task switched to ready (or new task) if task[2] == state.READY then local tube_id = self.tube_id - local consumer = queue_consumers.index.consumer:min{tube_id} + local consumer_group = CONSUMER_GROUP_ANY + + if self.raw.consumer_group ~= nil then + consumer_group = self.raw:consumer_group(nil, task) + end + + local consumer = queue_consumers.index.consumer:min{tube_id, consumer_group} if consumer ~= nil then - if consumer[3] == tube_id then + if consumer[3] == tube_id and consumer[6] == consumer_group then queue_consumers:delete{consumer[1], consumer[2]} local cond = conds[consumer[2]] if cond then @@ -785,7 +798,7 @@ function method.start() local _cons = box.space._queue_consumers if _cons == nil then - -- connection, fid, tube, time + -- connection, fid, tube, time, consumer group _cons = box.schema.create_space('_queue_consumers', { temporary = true, format = { @@ -793,7 +806,8 @@ function method.start() {name = 'fiber_id', type = num_type()}, {name = 'tube_id', type = num_type()}, {name = 'event_time', type = num_type()}, - {name = 'fiber_time', type = num_type()} + {name = 'fiber_time', type = num_type()}, + {name = 'consumer_group', type = str_type()} } }) _cons:create_index('pk', { @@ -803,9 +817,22 @@ function method.start() }) _cons:create_index('consumer', { type = 'tree', - parts = {3, num_type(), 4, num_type()}, + parts = {3, num_type(), 6, str_type(), 4, num_type()}, unique = false }) + elseif _cons:format()[6] == nil then + _cons:truncate() + _cons:format({ + {name = 'connection_id', type = num_type()}, + {name = 'fiber_id', type = num_type()}, + {name = 'tube_id', type = num_type()}, + {name = 'event_time', type = num_type()}, + {name = 'fiber_time', type = num_type()}, + {name = 'consumer_group', type = str_type()} + }) + _cons.index.consumer:alter({ + parts = {3, num_type(), 6, str_type(), 4, num_type()} + }) end -- Remove deprecated space @@ -890,7 +917,7 @@ local function build_stats(space) take = 0, touch = 0, -- for *ttl queues only ttl = 0, ttr = 0, delay = 0, - }} + }, extra = {}} local st = rawget(queue.stat, space) or {} local idx_tube = 1 @@ -914,6 +941,11 @@ local function build_stats(space) stats['tasks']['total'] = total stats['tasks']['done'] = st.done or 0 + local tube = queue.tube[space] + if tube ~= nil and tube.raw.statistics ~= nil then + stats['extra'] = tube.raw:statistics() + end + return stats end diff --git a/queue/abstract/driver/subqueuettl.lua b/queue/abstract/driver/subqueuettl.lua new file mode 100644 index 0000000..3d89ba2 --- /dev/null +++ b/queue/abstract/driver/subqueuettl.lua @@ -0,0 +1,608 @@ +local log = require('log') +local fiber = require('fiber') + +local state = require('queue.abstract.state') + +local util = require('queue.util') +local qc = require('queue.compat') +local num_type = qc.num_type +local str_type = qc.str_type + +local tube = {} +local method = {} + +local i_id = 1 +local i_status = 2 +local i_next_event = 3 +local i_ttl = 4 +local i_ttr = 5 +local i_pri = 6 +local i_created = 7 +local i_subqueue = 8 +local i_data = 9 + +-- subqueue registry space: stores only unique subqueue names (single field). +-- The registry is best-effort: a write failure must never fail or roll back a +-- queue operation, and entries are never updated/deleted on physical deletes. +-- Stale names simply report zero counts in statistics(). +local i_reg_subqueue = 1 + +local function is_expired(task) + local dead_event = task[i_created] + task[i_ttl] + return (dead_event <= fiber.time64()) +end + +-- validate space of queue +local function validate_space(space) + -- check indexes + local indexes = {'task_id', 'status', 'watch', 'subqueue_pri'} + 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 + +-- validate subqueue registry space +local function validate_registry(space) + if space.index['subqueue'] == nil then + error(string.format('space "%s" does not have "subqueue" index',space.name)) + end +end + +local function register_subqueue(registry, subqueue) + local ok, err = pcall(function() + registry:upsert({subqueue}, {}) + end) + if not ok then + log.error('failed to register subqueue %s: %s', subqueue, tostring(err)) + end +end + +-- create space +function tube.create_space(space_name, opts) + if opts.engine == 'vinyl' then + -- vinyl can not work properly in a competitive take. + -- The transaction may be aborted during taking. + -- Without a transaction, the driver cannot guarantee that all + -- consumers will receive unique tasks. + error('subqueuettl queue does not support vinyl engine') + end + + opts.ttl = opts.ttl or util.MAX_TIMEOUT + opts.ttr = opts.ttr or opts.ttl + opts.pri = opts.pri or 0 + + local space_opts = {} + local if_not_exists = opts.if_not_exists or false + space_opts.temporary = opts.temporary or false + space_opts.engine = opts.engine or 'memtx' + space_opts.format = { + {name = 'task_id', type = num_type()}, + {name = 'status', type = str_type()}, + {name = 'next_event', type = num_type()}, + {name = 'ttl', type = num_type()}, + {name = 'ttr', type = num_type()}, + {name = 'pri', type = num_type()}, + {name = 'created', type = num_type()}, + {name = 'subqueue', type = str_type()}, + {name = 'data', type = '*'} + } + + -- 1 2 3 4 5 6 7, 8 9 + -- task_id, status, next_event, ttl, ttr, pri, created, subqueue, data + local space = box.space[space_name] + if if_not_exists and space then + -- Validate the existing space. + validate_space(box.space[space_name]) + return space + end + + space = box.schema.create_space(space_name, space_opts) + space:create_index('task_id', { + type = 'tree', + parts = {i_id, num_type()} + }) + space:create_index('status', { + type = 'tree', + parts = {i_status, str_type(), i_pri, num_type(), i_id, num_type()} + }) + space:create_index('watch', { + type = 'tree', + parts = {i_status, str_type(), i_next_event, num_type()}, + unique = false + }) + space:create_index('subqueue_pri', { + type = 'tree', + parts = {i_status, str_type(), i_subqueue, str_type(), i_pri, num_type(), i_id, num_type()} + }) + return space +end + +local delayed_state = { state.DELAYED } +local ttl_states = { state.READY, state.BURIED } +local ttr_state = { state.TAKEN } + + +local function subqueuettl_fiber_iteration(self, processed) + local now = util.time() + local task = nil + local estimated = util.MAX_TIMEOUT + + -- delayed tasks + task = util.atomic(function() + local delayed_task = self.space.index.watch:min(delayed_state) + if delayed_task == nil or delayed_task[i_status] ~= state.DELAYED then + return nil + end + + if now < delayed_task[i_next_event] then + estimated = tonumber(delayed_task[i_next_event] - now) / 1000000 + + return nil + end + + estimated = 0 + processed = processed + 1 + + return self.space:update(delayed_task[i_id], { + { '=', i_status, state.READY }, + { '=', i_next_event, delayed_task[i_created] + delayed_task[i_ttl] } + }) + end) + + if task ~= nil then + self:on_task_change(task, 'delayed') + end + + -- ttl tasks + for _, ttl_state in pairs(ttl_states) do + task = self.space.index.watch:min{ ttl_state } + if task ~= nil and task[i_status] == ttl_state then + if now >= task[i_next_event] then + task = self:delete(task[i_id]):transform(2, 1, state.DONE) + self:on_task_change(task, 'ttl') + estimated = 0 + processed = processed + 1 + else + local et = tonumber(task[i_next_event] - now) / 1000000 + estimated = et < estimated and et or estimated + end + end + end + + -- ttr tasks + task = util.atomic(function() + local ttr_task = self.space.index.watch:min(ttr_state) + if ttr_task == nil or ttr_task[i_status] ~= state.TAKEN then + return nil + end + + if now < ttr_task[i_next_event] then + local et = tonumber(ttr_task[i_next_event] - now) / 1000000 + estimated = et < estimated and et or estimated + + return nil + end + + estimated = 0 + processed = processed + 1 + + return self.space:update(ttr_task[i_id], { + { '=', i_status, state.READY }, + { '=', i_next_event, ttr_task[i_created] + ttr_task[i_ttl] } + }) + end) + + if task ~= nil then + self:on_task_change(task, 'ttr') + end + + if estimated > 0 or processed > 1000 then + -- free refcounter + estimated = processed > 1000 and 0 or estimated + estimated = estimated > 0 and estimated or 0 + processed = 0 + self.cond:wait(estimated) + end + + return processed +end + +-- watch fiber +local function subqueuettl_fiber(self) + fiber.name('subqueuettl') + log.info("Started queue subqueuettl fiber") + local processed = 0 + + while true do + if box.info.ro == false then + local stat, err = pcall(subqueuettl_fiber_iteration, self, processed) + + if not stat and not (err.code == box.error.READONLY) then + log.error("error catched: %s", tostring(err)) + log.error("exiting fiber '%s'", fiber.name()) + return 1 + 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 subqueuettl fiber was stopped") + break + end + end + end +end + +-- start tube on space +function tube.new(space, on_task_change, opts) + validate_space(space) + + -- Create or restore the best-effort subqueue registry. The registry + -- stores ONLY unique subqueue names (no counters); statistics() iterates + -- it and computes per-state counts on the fly via subqueue_pri:count(). + -- Registry writes are best-effort: failures never fail or roll back a + -- queue operation, and entries are never removed on physical deletes, so + -- stale names may remain and simply report zero counts in statistics(). + local registry_name = space.name .. '_subqueues' + local registry = box.space[registry_name] + if registry == nil then + local registry_opts = { + temporary = opts.temporary or false, + engine = opts.engine or 'memtx', + format = { + {name = 'subqueue', type = str_type()}, + }, + } + registry = box.schema.create_space(registry_name, registry_opts) + registry:create_index('subqueue', { + type = 'tree', + parts = {i_reg_subqueue, str_type()}, + unique = true, + }) + -- Backfill names from existing tasks (e.g. after an upgrade or an + -- unclean restart of a temporary registry). Best-effort: individual + -- insert errors (duplicates, read-only, etc.) never block queue + -- initialization. + for _, task in space.index.subqueue_pri:pairs() do + register_subqueue(registry, task[i_subqueue]) + end + else + validate_registry(registry) + end + + on_task_change = on_task_change or (function() end) + local self = setmetatable({ + space = space, + registry = registry, + on_task_change = function(self, task, stat_data) + -- wakeup fiber + if task ~= nil and self.fiber ~= nil then + self.cond:signal(self.fiber:id()) + end + on_task_change(task, stat_data) + end, + opts = opts, + }, { __index = method }) + + self.cond = qc.waiter() + self.fiber = fiber.create(subqueuettl_fiber, self) + self.sync_chan = fiber.channel(1) + + return self +end + +-- method.grant grants provided user to all spaces of driver. +function method.grant(self, user, opts) + box.schema.user.grant(user, 'read,write', 'space', self.space.name, opts) + if self.registry ~= nil then + box.schema.user.grant(user, 'read,write', 'space', + self.registry.name, opts) + end +end + +function method.grant_role(self, role, opts) + box.schema.role.grant(role, 'read,write', 'space', self.space.name, opts) + if self.registry ~= nil then + box.schema.role.grant(role, 'read,write', 'space', + self.registry.name, opts) + end +end + +-- cleanup internal fields in task +function method.normalize_task(self, task) + return task and task:transform(i_next_event, i_data - i_next_event) +end + +-- put task in space +function method.put(self, data, opts) + if opts.subqueue == nil then + error('subqueue is required') + end + + local status + local ttl = opts.ttl or self.opts.ttl + local ttr = opts.ttr or self.opts.ttr + local pri = opts.pri or self.opts.pri or 0 + + local next_event + + if opts.delay ~= nil and opts.delay > 0 then + status = state.DELAYED + ttl = ttl + opts.delay + next_event = util.event_time(opts.delay) + else + status = state.READY + next_event = util.event_time(ttl) + end + + local subqueue = tostring(opts.subqueue) + + local task = util.atomic(function() + local max = self.space.index.task_id:max() + local id = max and max[i_id] + 1 or 0 + + return self.space:insert{ + id, + status, + next_event, + util.time(ttl), + util.time(ttr), + pri, + util.time(), + subqueue, + data + } + end) + + -- Best-effort: record the subqueue name in the registry. Done outside + -- the task transaction so a registry write failure (duplicate key, + -- read-only, etc.) never fails or rolls back the put. + register_subqueue(self.registry, subqueue) + + self:on_task_change(task, 'put') + return task +end + +local TIMEOUT_INFINITY_TIME = util.time(util.MAX_TIMEOUT) + +-- touch task +function method.touch(self, id, delta) + local ops = { + {'+', i_next_event, delta}, + {'+', i_ttl, delta}, + {'+', i_ttr, delta} + } + if delta == util.MAX_TIMEOUT then + ops = { + {'=', i_next_event, delta}, + {'=', i_ttl, delta}, + {'=', i_ttr, delta} + } + end + local task = self.space:update(id, ops) + + self:on_task_change(task, 'touch') + + return task +end + + +local function take(self, subqueue) + for _, task in self.space.index.subqueue_pri:pairs( + {state.READY, subqueue}, {iterator = 'GE'}) do + + if task == nil + or task[i_status] ~= state.READY + or task[i_subqueue] ~= subqueue then + break + end + if not is_expired(task) then + task = self.space:update(task[i_id], { + { '=', i_status, state.TAKEN }, + { '=', i_next_event, util.time() + task[i_ttr] } + }) + + if task ~= nil then + return task + end + end + end +end + +-- take task +function method.take(self, opts) + if opts == nil or opts.subqueue == nil then + error('subqueue is required') + end + + local task = util.atomic(function() + return take(self, tostring(opts.subqueue)) + end) + + if task ~= nil then + self:on_task_change(task, 'take') + end + + return task +end + +function method.consumer_group(self, opts, task) + local subqueue = task and task[i_subqueue] or opts and opts.subqueue + + return subqueue and 'subqueue:' .. tostring(subqueue) or nil +end + +-- delete task +function method.delete(self, id) + local task = util.atomic(function() + local task = self.space:get(id) + if task ~= nil then + self.space:delete(id) + end + + return task + end) + + if task == nil then + return nil + end + + task = task:transform(i_status, 1, state.DONE) + self:on_task_change(task, 'delete') + + return task +end + +-- release task +function method.release(self, id, opts) + local task = util.atomic(function() + local task = self.space:get{id} + if task == nil then + return nil + end + + if opts.delay ~= nil and opts.delay > 0 then + return self.space:update(id, { + { '=', i_status, state.DELAYED }, + { '=', i_next_event, util.event_time(opts.delay) }, + { '+', i_ttl, util.time(opts.delay) } + }) + end + + return self.space:update(id, { + { '=', i_status, state.READY }, + { '=', i_next_event, util.time(task[i_created] + task[i_ttl]) } + }) + end) + + if task == nil then + return + end + + self:on_task_change(task, 'release') + + return task +end + +-- bury task +function method.bury(self, id) + local task = util.atomic(function() + -- 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 `subqueuettl_fiber_iteration` + -- and the task will be deleted. + local task = self.space:get{id} + if task == nil then + return nil + end + + return self.space:update(id, { + { '=', i_status, state.BURIED }, + { '=', i_next_event, task[i_created] + task[i_ttl] } + }) + end) + + if task == nil then + return + end + + task = task:transform(i_status, 1, state.BURIED) + self:on_task_change(task, 'bury') + + return task +end + +-- unbury several tasks +function method.kick(self, count) + for i = 1, count do + local task = util.atomic(function() + local task = self.space.index.status:min{ state.BURIED } + if task == nil or task[i_status] ~= state.BURIED then + return nil + end + + return self.space:update(task[i_id], {{ '=', i_status, state.READY }}) + end) + if task == nil then + return i - 1 + end + + self:on_task_change(task, 'kick') + end + + return count +end + +-- peek task +function method.peek(self, id) + return self.space:get{id} +end + +-- get iterator to tasks in a certain state +function method.tasks_by_state(self, task_state) + return self.space.index.status:pairs(task_state) +end + +function method.statistics(self) + local statistics = {} + -- Iterate the registry of unique subqueue names. Per-state counts are + -- computed on the fly via subqueue_pri:count(); stale names (whose last + -- task was deleted/acked) simply report zero counts. + for _, r in self.registry:pairs() do + local subqueue = r[i_reg_subqueue] + local stats = {total = 0} + for name, task_state in pairs(state) do + if task_state ~= state.DONE then + local count = self.space.index.subqueue_pri:count{ + task_state, subqueue, + } + stats[name:lower()] = count + stats.total = stats.total + count + end + end + + statistics[subqueue] = stats + end + + return {subqueues = statistics} +end + +function method.truncate(self) + self.space:truncate() + if self.registry ~= nil then + self.registry:truncate() + end +end + +function method.start(self) + if self.fiber then + return + end + + self.fiber = fiber.create(subqueuettl_fiber, self) +end + +function method.stop(self) + if not self.fiber then + return + end + + self.cond:signal(self.fiber:id()) + self.sync_chan:put(true) + self.fiber = nil +end + +function method.drop(self) + self:stop() + + box.space[self.space.name]:drop() + if self.registry ~= nil and box.space[self.registry.name] ~= nil then + box.space[self.registry.name]:drop() + end +end + +return tube diff --git a/queue/init.lua b/queue/init.lua index 6e61faa..c3d6183 100644 --- a/queue/init.lua +++ b/queue/init.lua @@ -11,6 +11,7 @@ local core_drivers = { fifottl = require('queue.abstract.driver.fifottl'), utube = require('queue.abstract.driver.utube'), utubettl = require('queue.abstract.driver.utubettl'), + subqueuettl = require('queue.abstract.driver.subqueuettl'), limfifottl = require('queue.abstract.driver.limfifottl') } diff --git a/queue/util.lua b/queue/util.lua index 84b350a..42f6feb 100644 --- a/queue/util.lua +++ b/queue/util.lua @@ -39,8 +39,42 @@ local util = { TIMEOUT_INFINITY = TIMEOUT_INFINITY } +local function atomic_tail(status, ...) + if not status then + box.rollback() + error((...), 2) + end + + box.commit() + + return ... +end + +-- Analog of box.atomic() with checks for queue operations. +-- It is used to wrap queue operations in a transaction. +local function atomic(fun, ...) + if box.is_in_txn() then + return fun(...) + end + + if box.cfg.memtx_use_mvcc_engine then + -- max() + insert() or min() + update() do not work as expected with + -- best-effort visibility: for write transactions it chooses + -- read-committed, for read transactions it chooses read-confirmed. + -- + -- So max()/min() could return the same tuple even if a concurrent + -- insert()/update() has been committed, but has not confirmed yet. + box.begin({txn_isolation = 'read-committed'}) + else + box.begin() + end + + return atomic_tail(pcall(fun, ...)) +end + -- methods local method = { + atomic = atomic, time = time, event_time = event_time } diff --git a/t/170-register-driver-after-reload.t b/t/170-register-driver-after-reload.t index 20ff97f..c04350f 100755 --- a/t/170-register-driver-after-reload.t +++ b/t/170-register-driver-after-reload.t @@ -6,7 +6,7 @@ local tap = require('tap') local tnt = require('t.tnt') local test = tap.test('custom driver registration after reload') -test:plan(1) +test:plan(2) tnt.cfg() @@ -38,6 +38,19 @@ end check_driver_registration_after_reload() +local consumers = box.space._queue_consumers +consumers:format({ + {name = 'connection_id', type = 'unsigned'}, + {name = 'fiber_id', type = 'unsigned'}, + {name = 'tube_id', type = 'unsigned'}, + {name = 'event_time', type = 'unsigned'}, + {name = 'fiber_time', type = 'unsigned'} +}) +queue.start() +test:ok(consumers:format()[6].name == 'consumer_group' and + consumers.index.consumer.parts[2].fieldno == 6, + 'upgrade consumer space format') + tnt.finish() os.exit(test:check() and 0 or 1) -- vim: set ft=lua : diff --git a/t/250-subqueuettl.t b/t/250-subqueuettl.t new file mode 100644 index 0000000..2535fff --- /dev/null +++ b/t/250-subqueuettl.t @@ -0,0 +1,408 @@ +#!/usr/bin/env tarantool + +local os = require('os') +local fiber = require('fiber') +local log = require('log') +local tnt = require('t.tnt') +local test = require('tap').test('subqueuettl') + +local queue = require('queue') +local state = require('queue.abstract.state') +local queue_state = require('queue.abstract.queue_state') +local qc = require('queue.compat') + +test:plan(20) +tnt.cfg{} + +local engine = os.getenv('ENGINE') or 'memtx' +if engine == 'vinyl' then + print('1..0 # SKIP subqueuettl does not support vinyl engine') + os.exit(0) +end +local tube = queue.create_tube('subqueue', 'subqueuettl', {engine = engine}) +local tube_stat = queue.create_tube('subqueue_stat', 'subqueuettl', {engine = engine}) + +-- Counts the number of subqueues reported in stats.driver.subqueues. +local function count_subqueues(stats) + local n = 0 + for _ in pairs(stats.extra.subqueues) do + n = n + 1 + end + return n +end + +test:ok(rawget(box, 'space'), 'box started') +test:ok(queue, 'queue is loaded') +test:ok(tube, 'test tube created') +test:is(tube.name, 'subqueue', 'tube.name') +test:is(tube.type, 'subqueuettl', 'driver is registered') + +test:test('statistics', function(test) + test:plan(23) + for i = 0, 4 do + tube_stat:put('stat_' .. i, {subqueue = 'stat_' .. i}) + end + + tube_stat:put('stat_5', {subqueue = 'stat_5', delay = 1000}) + tube_stat:delete(4) + tube_stat:take(.001, {subqueue = 'stat_0'}) + tube_stat:release(0) + tube_stat:take(.001, {subqueue = 'stat_0'}) + tube_stat:ack(0) + tube_stat:bury(1) + tube_stat:bury(2) + tube_stat:kick(1) + tube_stat:take(.001, {subqueue = 'stat_1'}) + + local stats = queue.statistics(tube_stat.name) + + 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') + + 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:is(stats.extra.subqueues.stat_1.taken, 1, 'subqueue taken task count') + test:is(stats.extra.subqueues.stat_2.buried, 1, 'subqueue buried task count') + test:is(stats.extra.subqueues.stat_3.ready, 1, 'subqueue ready task count') + test:is(stats.extra.subqueues.stat_5.delayed, 1, 'subqueue delayed task count') + test:is(stats.extra.subqueues.stat_5.total, 1, 'subqueue total task count') + + -- All subqueue names are recorded in the best-effort registry, even + -- those whose tasks were later removed. + test:is(count_subqueues(stats), 6, 'all six subqueue names recorded') + + -- Stale subqueues (all tasks removed via ack/delete) remain in the + -- best-effort registry and are reported with zero counts. + test:is(stats.extra.subqueues.stat_0.total, 0, + 'emptied stat_0 reports zero total') + test:is(stats.extra.subqueues.stat_0.ready, 0, + 'emptied stat_0 reports zero ready') + test:is(stats.extra.subqueues.stat_4.total, 0, + 'deleted stat_4 reports zero total') + test:is(stats.extra.subqueues.stat_4.ready, 0, + 'deleted stat_4 reports zero ready') +end) + +test:test('basic put, take, and ack', function(test) + test:plan(11) + test:ok(tube:put(123, {subqueue = 'basic'}), 'task was put') + test:ok(tube:put(345, {subqueue = 'basic'}), 'task was put') + + local task = tube:take(.1, {subqueue = 'basic'}) + test:ok(task, 'task was taken') + test:is(task[2], state.TAKEN, 'task status') + test:is(task[3], 123, 'task.data') + + 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, {subqueue = 'basic'}) + test:ok(task, 'second task was taken') + test:is(task[3], 345, 'task.data') + test:is(task[2], state.TAKEN, 'task status') + tube:ack(task[1]) +end) + +test:test('TTR', function(test) + test:plan(3) + local subqueue = 'ttr' + + test:ok(tube:put('ttr', {subqueue = subqueue, ttr = 1}), 'put TTR task') + test:ok(tube:take(.1, {subqueue = subqueue}), 'take TTR task') + fiber.sleep(1.1) + + local task = tube:peek(tube.raw.space.index.task_id:max()[1]) + test:is(task[2], state.READY, 'task becomes ready after TTR') + + task = tube:take(.1, {subqueue = subqueue}) + tube:ack(task[1]) +end) + +test:test('parallel take from one subqueue', function(test) + test:plan(7) + local subqueue = 'parallel' + test:ok(tube:put(678, {subqueue = subqueue}), 'first task was put') + test:ok(tube:put(890, {subqueue = subqueue}), 'second task was put') + + local session_uuid = queue.identify() + local result = fiber.channel(2) + + for i = 1, 2 do + fiber.create(function() + queue.identify(session_uuid) + + result:put(tube:take(.1, {subqueue = subqueue})) + end) + end + + local first = result:get(.2) + local second = result:get(.2) + test:ok(first, 'first task was taken') + test:ok(second, 'second task was taken concurrently') + if first and second then + test:isnt(first[1], second[1], 'different tasks were taken') + else + test:fail('different tasks were taken') + end + + test:ok(tube:ack(first[1]), 'first task was acked') + test:ok(tube:ack(second[1]), 'second task was acked') +end) + +test:test('release with delay', function(test) + test:plan(4) + local subqueue = 'delay' + test:ok(tube:put(789, {subqueue = subqueue}), 'task was put') + test:ok(tube:put(901, {subqueue = subqueue}), 'task was put') + + local task = tube:take(.1, {subqueue = subqueue}) + test:is(task[3], 789, 'first task was taken') + + tube:release(task[1], {delay = .2}) + task = tube:take(.1, {subqueue = subqueue}) + test:is(task[3], 901, 'second task was taken while first is delayed') + + tube:ack(task[1]) + fiber.sleep(.25) + task = tube:take(.1, {subqueue = subqueue}) + tube:ack(task[1]) +end) + +test:test('priority', function(test) + test:plan(4) + local subqueue = 'priority' + test:ok(tube:put(670, {subqueue = subqueue, pri = 1}), 'task was put') + test:ok(tube:put(671, {subqueue = subqueue, pri = 0}), 'task was put') + + local task = tube:take(.1, {subqueue = subqueue}) + test:is(task[3], 671, 'higher priority task was taken first') + + tube:release(task[1]) + task = tube:take(.1, {subqueue = subqueue}) + test:is(task[3], 671, 'released higher priority task remains first') + + tube:ack(task[1]) + task = tube:take(.1, {subqueue = subqueue}) + tube:ack(task[1]) +end) + +test:test('if_not_exists', function(test) + test:plan(2) + local existing = queue.create_tube('subqueue_ine', 'subqueuettl', { + if_not_exists = true, engine = engine, + }) + + local same = queue.create_tube('subqueue_ine', 'subqueuettl', { + if_not_exists = true, engine = engine, + }) + test:is(existing, same, 'existing tube is reused') + + queue.tube.subqueue_ine = nil + local reloaded = queue.create_tube('subqueue_ine', 'subqueuettl', { + if_not_exists = true, engine = engine, + }) + test:isnt(existing, reloaded, 'tube is loaded from existing space') +end) + +test:test('read-only mode', function(test) + test:plan(7) + tube:put('read_only', {subqueue = 'read_only', delay = .1}) + + local ttl_fiber = tube.raw.fiber + box.cfg{read_only = true} + + test:ok(queue_state.poll(queue_state.states.WAITING, 10), + 'queue state changed to waiting') + fiber.sleep(.11) + test:is(ttl_fiber:status(), 'dead', 'background fiber is canceled') + test:isnil(tube.raw.fiber, 'background fiber object is cleaned') + if qc.check_version({1, 7}) then + test:isnil(tube:take(.2, {subqueue = 'read_only'}), + 'delayed task is not moved to ready') + else + local ok = pcall(tube.take, tube, .2, {subqueue = 'read_only'}) + test:is(ok, false, 'task cannot be taken while read-only') + end + + box.cfg{read_only = false} + test:ok(queue_state.poll(queue_state.states.RUNNING, 10), + 'queue state changed to running') + test:is(tube.raw.fiber:status(), 'suspended', 'background fiber restarted') + + local task = tube:take(.2, {subqueue = 'read_only'}) + test:ok(task, 'task can be taken after read-write restore') + tube:ack(task[1]) +end) + +test:test('TTL after delayed release', function(test) + test:plan(2) + local ttl, ttr, delay = 10, 20, 5 + local ttl_tube = queue.create_tube('subqueue_ttl_release', 'subqueuettl', { + if_not_exists = true, engine = engine, + }) + ttl_tube:put('task', {subqueue = 'ttl', ttl = ttl, ttr = ttr}) + local task = ttl_tube:take(.1, {subqueue = 'ttl'}) + ttl_tube:release(task[1], {delay = delay}) + + task = box.space.subqueue_ttl_release:get(task[1]) + test:is(task.ttl, (ttl + delay) * 1000000, 'TTL includes release delay') + test:is(task.ttr, ttr * 1000000, 'TTR is unchanged') +end) + +test:test('tasks by state', function(test) + test:plan(2) + local state_tube = queue.create_tube('subqueue_by_state', 'subqueuettl', { + engine = engine, + }) + for i = 1, 10 do + state_tube:put('task_' .. i, {subqueue = tostring(i)}) + end + for i = 1, 4 do + state_tube:take(.001, {subqueue = tostring(i)}) + end + + local ready, taken = 0, 0 + for _, task in state_tube.raw:tasks_by_state(state.READY) do + if task[2] == state.READY then ready = ready + 1 end + end + for _, task in state_tube.raw:tasks_by_state(state.TAKEN) do + if task[2] == state.TAKEN then taken = taken + 1 end + end + + test:is(ready, 6, 'ready task count') + test:is(taken, 4, 'taken task count') +end) + +test:test('consumer group wakeup', function(test) + test:plan(10) + test:ok(tube:put('first', {subqueue = 'first'}), 'put first subqueue task') + test:ok(tube:put('second', {subqueue = 'second'}), 'put second subqueue task') + + local task = tube:take(.1, {subqueue = 'second'}) + test:ok(task, 'take requested subqueue task') + test:is(task and task[3], 'second', 'take only requested subqueue task') + test:ok(not pcall(tube.take, tube, .1), 'take without subqueue is rejected') + test:ok(tube:ack(task[1]), 'ack subqueue task') + + local result = fiber.channel(1) + fiber.create(function() + result:put(tube:take(.2, {subqueue = 'target'})) + end) + fiber.sleep(.01) + + test:ok(tube:put('other', {subqueue = 'other'}), 'put other subqueue task') + test:isnil(result:get(.05), 'other subqueue does not wake target consumer') + + test:ok(tube:put('target', {subqueue = 'target'}), 'put target subqueue task') + task = result:get(.1) + test:is(task and task[3], 'target', 'target subqueue wakes matching consumer') +end) + +test:test('registry names persist after ack', function(test) + test:plan(5) + local t = queue.create_tube('subqueue_reg_del', 'subqueuettl', + {engine = engine}) + + t:put('a', {subqueue = 'x'}) + t:put('b', {subqueue = 'x'}) + t:put('c', {subqueue = 'y'}) + + local stats = queue.statistics(t.name) + test:is(stats.extra.subqueues.x.total, 2, 'two tasks in subqueue x') + test:is(stats.extra.subqueues.y.total, 1, 'one task in subqueue y') + test:is(count_subqueues(stats), 2, 'two subqueues registered') + + -- Ack all tasks in x; the subqueue name must remain in the registry + -- (best-effort, no cleanup on delete) and report zero counts. + local task = t:take(.001, {subqueue = 'x'}) + t:ack(task[1]) + task = t:take(.001, {subqueue = 'x'}) + t:ack(task[1]) + + stats = queue.statistics(t.name) + test:is(stats.extra.subqueues.x.total, 0, + 'subqueue x reports zero total after all tasks acked') + test:is(count_subqueues(stats), 2, + 'subqueue names still registered after acks') + + t:drop() +end) + +test:test('registry cleared on truncate', function(test) + test:plan(4) + local t = queue.create_tube('subqueue_reg_trunc', 'subqueuettl', + {engine = engine}) + + t:put('a', {subqueue = 'x'}) + t:put('b', {subqueue = 'y'}) + + local stats = queue.statistics(t.name) + test:is(count_subqueues(stats), 2, 'two subqueues before truncate') + test:is(stats.tasks.total, 2, 'two tasks before truncate') + + t:truncate() + + stats = queue.statistics(t.name) + test:is(count_subqueues(stats), 0, 'no subqueues after truncate') + test:is(stats.tasks.total, 0, 'no tasks after truncate') + + t:drop() +end) + +test:test('registry backfilled on reload', function(test) + test:plan(4) + local t = queue.create_tube('subqueue_reg_reload', 'subqueuettl', + {if_not_exists = true, engine = engine}) + + t:put('a', {subqueue = 'r1'}) + t:put('b', {subqueue = 'r2'}) + + -- Simulate reload: drop the in-memory tube reference and the registry + -- space, then recreate from the persisted main space. The driver must + -- backfill the registry from existing tasks. + queue.tube.subqueue_reg_reload = nil + box.space.subqueue_reg_reload_subqueues:drop() + local reloaded = queue.create_tube('subqueue_reg_reload', 'subqueuettl', + {if_not_exists = true, engine = engine}) + + local stats = queue.statistics(reloaded.name) + test:is(stats.extra.subqueues.r1.total, 1, + 'subqueue r1 backfilled after reload') + test:is(stats.extra.subqueues.r2.total, 1, + 'subqueue r2 backfilled after reload') + test:is(count_subqueues(stats), 2, 'two subqueues backfilled after reload') + test:is(stats.tasks.total, 2, 'all tasks preserved after reload') + + reloaded:drop() +end) + +test:test('registry space is dropped with tube', function(test) + test:plan(3) + local t = queue.create_tube('subqueue_reg_drop', 'subqueuettl', + {engine = engine}) + + t:put('a', {subqueue = 'd1'}) + test:ok(box.space.subqueue_reg_drop_subqueues ~= nil, + 'registry space exists while tube is alive') + + t:drop() + test:isnil(box.space.subqueue_reg_drop, + 'main space dropped with tube') + test:isnil(box.space.subqueue_reg_drop_subqueues, + 'registry space dropped with tube') +end) + +tnt.finish() +os.exit(test:check() and 0 or 1)