From 25775ac8873bf1f62f0c41b0c5d7e067cad0c691 Mon Sep 17 00:00:00 2001 From: ThCompiler Date: Tue, 18 Aug 2026 22:49:23 +0300 Subject: [PATCH 1/3] take: support pass options via driver API Support driver-specific task selection through take() options. Some drivers partition tasks by driver-specific attributes and must select tasks only from the partition requested by a consumer. For example, a driver may use opts.subqueue to take tasks from a particular subqueue. Drivers can derive a consumer group from take() options and tasks. This lets the wake-up mechanism notify only consumers eligible to take a newly ready task, rather than waking an arbitrary consumer from the same tube that cannot process it. --- CHANGELOG.md | 5 +++- README.md | 2 +- queue/abstract.lua | 45 ++++++++++++++++++++++------ t/170-register-driver-after-reload.t | 15 +++++++++- 4 files changed, 55 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 50e89c16..b4e82c6b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,10 @@ 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. + ### Fixed diff --git a/README.md b/README.md index 550c6bba..dab09dc0 100644 --- a/README.md +++ b/README.md @@ -647,7 +647,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. diff --git a/queue/abstract.lua b/queue/abstract.lua index 45051cbd..0837d1f2 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 diff --git a/t/170-register-driver-after-reload.t b/t/170-register-driver-after-reload.t index 20ff97f8..c04350ff 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 : From ca5eba741f5f9f6b56eea48f3274fdc6c25b7c6e Mon Sep 17 00:00:00 2001 From: ThCompiler Date: Tue, 18 Aug 2026 22:59:49 +0300 Subject: [PATCH 2/3] statistics: support driver extra statistics Support driver-specific statistics. Some drivers maintain additional state or partition tasks using driver-specific attributes. They need to expose this data through the standard tube statistics API, alongside the common queue metrics. This allows each driver to report meaningful operational metrics without extending the generic statistics implementation for every driver-specific use case. --- CHANGELOG.md | 2 +- README.md | 6 +++++- queue/abstract.lua | 5 +++++ 3 files changed, 11 insertions(+), 2 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b4e82c6b..ac07a8e4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - 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. ### Fixed diff --git a/README.md b/README.md index dab09dc0..6f13c85c 100644 --- a/README.md +++ b/README.md @@ -825,7 +825,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: diff --git a/queue/abstract.lua b/queue/abstract.lua index 0837d1f2..73a6600d 100644 --- a/queue/abstract.lua +++ b/queue/abstract.lua @@ -941,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 From 1cfcd23db5fef671166184f309752fda2d2d6782 Mon Sep 17 00:00:00 2001 From: ThCompiler Date: Tue, 18 Aug 2026 23:03:30 +0300 Subject: [PATCH 3/3] driver: implement new driver `subqueuettl` driver for processing tasks from named subqueues in one tube space. Add subqueuettl driver with consumer groups Some workloads need to create queues dynamically with arbitrary names. Creating a separate tube for each queue requires converting those names to valid Tarantool space names and granting access whenever a new tube is created. This adds operational overhead and makes access management harder. The utube model is otherwise well suited for this use case: it stores multiple logical queues in one space and guarantees sequential processing inside each queue. However, it does not support taking a task from a specific utube, so it cannot be used when consumers must be assigned to particular dynamically named queues. Added subqueuettl as a TTL-aware alternative. It stores multiple dynamically named subqueues in one space, and both put() and take() require a subqueue name. This keeps task selection isolated while retaining TTL, TTR, priority, delay, and ready-buffer support. --- CHANGELOG.md | 1 + README.md | 51 ++- queue-scm-1.rockspec | 1 + queue/CMakeLists.txt | 2 + queue/abstract.lua | 2 +- queue/abstract/driver/subqueuettl.lua | 608 ++++++++++++++++++++++++++ queue/init.lua | 1 + queue/util.lua | 34 ++ t/250-subqueuettl.t | 408 +++++++++++++++++ 9 files changed, 1106 insertions(+), 2 deletions(-) create mode 100644 queue/abstract/driver/subqueuettl.lua create mode 100644 t/250-subqueuettl.t diff --git a/CHANGELOG.md b/CHANGELOG.md index ac07a8e4..323d4bf1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - 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 6f13c85c..17baa2c2 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. @@ -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. @@ -860,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 ... ``` @@ -1007,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 73122a96..a563fcbf 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 9234cb66..c374412a 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 73a6600d..4b8469b0 100644 --- a/queue/abstract.lua +++ b/queue/abstract.lua @@ -917,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 diff --git a/queue/abstract/driver/subqueuettl.lua b/queue/abstract/driver/subqueuettl.lua new file mode 100644 index 00000000..3d89ba20 --- /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 6e61faa3..c3d61835 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 84b350ac..42f6feb4 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/250-subqueuettl.t b/t/250-subqueuettl.t new file mode 100644 index 00000000..2535fff1 --- /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)