Extend driver API and add subqueuettl driver

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
 

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'

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',

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/)

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,15 +798,16 @@ 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 = {
                 {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 = '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