Skip to content

Commit d8dcd2f

Browse files
committed
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.
1 parent ca5eba7 commit d8dcd2f

9 files changed

Lines changed: 1128 additions & 2 deletions

File tree

‎CHANGELOG.md‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
1313
- Optional `consumer_group(opts, task)` driver API for routing waiting
1414
consumers to matching tasks.
1515
- Support extra statistics from driver API.
16+
- Add the `subqueuettl` driver for processing tasks from named subqueues in one tube space.
1617

1718
### Fixed
1819

‎README.md‎

Lines changed: 50 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -16,6 +16,7 @@ align="right">
1616
* [fifottl \- a simple priority queue with support for task time to live](#fifottl---a-simple-priority-queue-with-support-for-task-time-to-live)
1717
* [utube \- a queue with sub\-queues inside](#utube---a-queue-with-sub-queues-inside)
1818
* [utubettl \- extension of utube to support ttl](#utubettl---extension-of-utube-to-support-ttl)
19+
* [subqueuettl \- TTL subqueues in one space](#subqueuettl---ttl-subqueues-in-one-space)
1920
* [The underlying spaces](#the-underlying-spaces)
2021
* [Fields of the \_queue space](#fields-of-the-_queue-space)
2122
* [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`
302303
is changed to 'ready' so another worker may take it)
303304
* `delay` - time to wait before starting to execute the task, in seconds
304305

306+
## `subqueuettl` - TTL subqueues in one space
307+
308+
`subqueuettl` stores independent TTL subqueues in one Tarantool space. It
309+
does not provide an operation to take a task from all subqueues. The driver
310+
supports the `memtx` engine only; `vinyl` is not supported.
311+
312+
The following options can be specified when putting a task in a
313+
`subqueuettl` queue:
314+
315+
* `subqueue` - required name of the subqueue.
316+
* `pri` - task priority (`0` is the highest priority and is the default).
317+
* `ttl` - numeric time to live in seconds. If omitted, it is set to infinity.
318+
* `ttr` - numeric time allotted to process a task in seconds. If omitted, it
319+
is set to the same value as `ttl`.
320+
* `delay` - time in seconds to wait before a task becomes ready.
321+
322+
Both `put()` and `take()` require `opts.subqueue`:
323+
324+
```lua
325+
local tube = queue.create_tube('sites', 'subqueuettl')
326+
tube:put('some task', {subqueue = 'example_subqueue'})
327+
local task = tube:take(10, {subqueue = 'example_subqueue'})
328+
```
329+
330+
`subqueuettl` adds per-subqueue statistics under the common `extra` field:
331+
332+
```lua
333+
local stats = queue.statistics('sites')
334+
local example_stats = stats.extra.subqueues['example_subqueue']
335+
```
336+
337+
Each subqueue entry contains `ready`, `taken`, `buried`, `delayed`, and `total`
338+
task counts. The subqueue name registry is best-effort; a registry write error
339+
does not fail the queue operation, and a stale name can remain with zero counts.
340+
305341
# The underlying spaces
306342

307343
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'.
660696
If there is no such task, and timeout was specified, then
661697
the job waits until a task becomes ready or the timeout expires.
662698

699+
The options, if specified, must be one or more of the options described
700+
above
701+
(`subqueue`, required name of the subqueue to take a task from).
702+
663703
Effect: the value of `task_state` changes to 't' (taken).
664704
The `take` request tells the system that the task is being worked on.
665705
It should be followed by an `ack` request when the work is finished.
@@ -860,6 +900,14 @@ queue.statistics('list_of_sites')
860900
bury: 1
861901
put: 2
862902
delete: 1
903+
extra:
904+
subqueues:
905+
example.com:
906+
ready: 0
907+
taken: 0
908+
buried: 0
909+
delayed: 0
910+
total: 0
863911
...
864912
```
865913

@@ -1007,7 +1055,8 @@ API:
10071055
which is passed on to the user (removes the administrative fields)
10081056
* `tube:put(data[, opts])` - puts a task into the queue.
10091057
Returns a normalized task which represents a tuple in the space
1010-
* `tube:take()` - sets the task state to 'in progress' and returns the task.
1058+
* `tube:take([opts])` - sets the task state to 'in progress' and returns the
1059+
task. `subqueuettl` requires `opts.subqueue`.
10111060
If there are no 'ready' tasks in the queue, returns nil.
10121061
* `tube:delete(task_id)` - deletes a task from the queue.
10131062
Returns the original task with a state changed to 'done'

‎queue-scm-1.rockspec‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@ build = {
2222
['queue.abstract.queue_state'] = 'queue/abstract/queue_state.lua',
2323
['queue.abstract.driver.fifottl'] = 'queue/abstract/driver/fifottl.lua',
2424
['queue.abstract.driver.utubettl'] = 'queue/abstract/driver/utubettl.lua',
25+
['queue.abstract.driver.subqueuettl'] = 'queue/abstract/driver/subqueuettl.lua',
2526
['queue.abstract.driver.fifo'] = 'queue/abstract/driver/fifo.lua',
2627
['queue.abstract.driver.utube'] = 'queue/abstract/driver/utube.lua',
2728
['queue.abstract.driver.limfifottl'] = 'queue/abstract/driver/limfifottl.lua',

‎queue/CMakeLists.txt‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,5 +22,7 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/fifottl.lua
2222
DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/)
2323
install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/utubettl.lua
2424
DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/)
25+
install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/subqueuettl.lua
26+
DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/)
2527
install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/limfifottl.lua
2628
DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/)

‎queue/abstract.lua‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -917,7 +917,7 @@ local function build_stats(space)
917917
take = 0, touch = 0,
918918
-- for *ttl queues only
919919
ttl = 0, ttr = 0, delay = 0,
920-
}}
920+
}, extra = {}}
921921

922922
local st = rawget(queue.stat, space) or {}
923923
local idx_tube = 1

0 commit comments

Comments
 (0)