xQueue

Add power of the Queue into Tarantool Space

Latest Release: 5.0.0

Backward compatibility rockspec: rockspecs/xqueue-scm-5.rockspec

Always latest rockspec: rockspecs/xqueue-dev-1.rockspec

Status

• R - ready - task is ready to be taken

  • created by put without delay
  • turned on by release/kick without delay
  • turned on from W when delay passed
  • deleted after ttl if ttl is enabled

• T - taken - task is taken by consumer. may not be taken by other.

  • turned into R after ttr if ttr is enabled

• W - waiting - task is not ready to be taken. waiting for its delay

  • requires runat
  • delay may be set during put, release, kick
  • turned into R after delay

• B - buried - task was temporary discarded from queue by consumer

  • may be revived using kick by administrator
  • use it in unpredicted conditions, when man intervention is required
  • Without mandatory monitoring (stats.buried) usage of buried is useless and awry

• Z - zombie - task was processed and ack'ed and temporary kept for delay

• D - done - task was processed and ack'ed and permanently left in database

  • enabled when keep feature is set

Interface

Creator methods

• .upgrade(space, options)

Imbue space with power of queue

M.upgrade(space, {
    format = {
        -- space format. applied to space.format() if passed
    },
    fields = {
        -- id is always taken from pk
        status   = 'status_field_name'    | status_field_no,
        runat    = 'runat_field_name'     | runat_field_no,
        priority = 'priority_field_name'  | priority_field_no,
    },
    features = {
        id = 'auto_increment' | 'time64' | 'uuid' | 'required' | function() ... return task_id end
        -- auto_increment - if pk is number, then use it for auto_increment
        -- uuid - if pk is string, then use uuid for id
        -- required - primary key MUST be present in tuple during put
        -- function - funciton will be called to aquire id for task
        retval = 'table' | 'tuple'
        -- table requires space format. default if applicable. a bit slower

        buried = true,           -- if true, support bury/kick
        delayed = true,          -- if true, support delayed tasks, requires `runat`

        keep = true,             -- if true, keep ack'ed tasks in [D]one state, instead of deleting
        -- mutually exclusive with zombie

        zombie = true|number,    -- requires `runat` field
        -- if number, then with default zombie delay, otherwise only if set delay during ack
        -- mutually exclusive with keep

        ttl    = true|number,    -- requires `runat` field
        -- Time To Live. Task is expired unless taken within time
        -- if number, then with default ttl, otherwise only if set during put/release
        ttr    = true|number,    -- requires `runat` field
        -- Time To Release. Task is returned into [R]eady unless processed (turned to ack|release from taken) within time
        -- if number, then with default ttl, otherwise only if set during take
    },
    -- Set tubes for which statistics collector will be enabled
    tube_stats = { 'tube-1', 'tube-2' },
})

Producer methods

• space:put
  • task - table or array or tuple
    • table
      • requires space format
      • suitable for id generation
    • array
      • ignores space format
      • for id generation use NULL (not nil)
    • tuple
      • ignores space format
      • can't be used with id generation
  • attr - table of attributes
    • delay - number of seconds
      • if set, task will become W instead of R for delay seconds
    • ttl - number of seconds
      • if set, task will be discarded after ttl seconds unless was taken

box.space.myqueue:put{ name="xxx"; data="yyy"; }
box.space.myqueue:put{ "xxx","yyy" }
box.space.myqueue:put(box.tuple.new{ 1,"xxx","yyy" })

box.space.myqueue:put({ name="xxx"; data="yyy"; }, { delay = 1.5 })
box.space.myqueue:put({ name="xxx"; data="yyy"; }, { delay = 1.5; ttl = 100 })

Consumer methods

• space:take(timeout)

  • timeout - number of seconds to wait for new task
    • choose reasonable time
    • beware of readahead size (see tarantool docs)
  • returns task tuple or table (see retval) or nothing on timeout
  • TODO: ttr must be there

• space:ack(id, [attr])

  • id:
    • string | number - primary key
    • tuple - key will be extracted using index
    • TODO: composite pk
  • attr:
    • update - table for update, like in space:update
    • delay - number of seconds
      • applicable only when zombie enabled
      • if zombie is enabled, delay deletion for delay time
  • returns task tuple or table (see retval)

• space:release(id, [attr])

  • id - as in :ack
  • attr
    • update - table for update, like in space:update
    • ttl - timeout for time to live
    • delay - number of seconds
      • if set, task will become W instead of R for delay seconds

• space:bury(id, [attr])

  • id - as in :ack
  • attr
    • update - table for update, like in space:update
    • ttl - timeout for time to live (?)

• space:wait(id, [timeout])

  • id - as in :ack
  • timeout - number of seconds to wait for the task processing
  • returns task tuple or table (see retval) and boolean was_processed flag

Admin methods

• space:dig(id, [attr]) - dig out task from buried state

  • id - as in :ack
  • attr
    • update - table for update, like in space:update
    • ttl - timeout for time to live
    • delay - number of seconds
      • if set, task will become W instead of R for delay seconds
  • returns task tuple or table (see retval)

• space:kill(id) - kill the task when it was taken

  • id - as in :ack
  • returns task tuple or table (see retval)

• space:dig(N, [attr]) - dig out N oldest tasks from buried state

  • N - number. To dig out all buried tasks set infinity or nil
  • attr - attrs for every task, like in dig

• space:stats()

  • returns table

  tarantool> box.space.queue:stats()
  ---
  - tube:
      B:
        counts: {'T': 0, 'W': 0, 'R': 0, 'D': 0, 'Z': 0, 'B': 0}
        transition:
          X-R: 6
          R-T: 7
          T-S: 3
          T-X: 4
      A:
        counts: {'B': 0, 'W': 0, 'D': 0, 'T': 0, 'Z': 0, 'R': 0}
        transition: []
      C:
        counts: {'T': 0, 'W': 0, 'R': 0, 'D': 0, 'Z': 0, 'B': 0}
        transition:
          T-R: 16
          R-T: 19
          S-R: 3
          T-X: 3
    transition:
      X-R: 6
      R-T: 26
      T-R: 19
      T-X: 7
    counts: {'T': 0, 'W': 0, 'R': 0, 'D': 0, 'Z': 0, 'B': 0}
  ...

Explanation of Statuses in :stats()

Status	Description
X	Special status, means task went to void, or appeared from void (makes sense in transitions)
R	(Ready) Task is Ready to be Taken
W	(Waiting) Task processing was delayed by producer or consumer
T	(Taken) Task has been taken by consumer
B	(Buried) Fatal error happened during task processing (and consumer buried it)
Z	(Zombie) Task was processed successfully and left in the Queue for zombie_delay
D	(Done) Task was processed successfully and left in the Queue
S	Extra status for tube transition

Status Transitions

Transition	Description
X -> R	Task was put into the Queue into Ready state (method :put)
X -> W	Task was put into the Queue into Waiting state (method :put with specified delay)
W -> R	Task was scheduled by runat mechanism of the Queue and became ready to be Taken by consumer
R -> T	Task was taken by consumer
T -> W	Consumer returned task into the Queue and delayed its execution
T -> R	Consumer returned task into the Queue for immediate execution
T -> X	Consumer acknowledged successful execution of the Task, and task was removed from the Queue
T -> Z	Consumer acknowledged successful execution of the Task, and task was moved to Zombie state for zombie_delay timeout
T -> D	Consumer acknowledged successful execution of the Task, and task was moved to Done status and will be left in the Queue forever
T -> S	Task was moved into another tube
S -> R	Task was moved from another tube to be Ready
S -> W	Task was moved from another tube to be Waiting
S -> D	Task was moved from another tube to be Done
S -> B	Task was moved from another tube to be Buried

Benchmarks

❯ .rocks/bin/luabench -d 1000000x
Tarantool version: Tarantool 2.10.7-0-g60f7e18
Tarantool build: Darwin-arm64-RelWithDebInfo (static)
Tarantool build flags:  -Wno-unknown-pragmas -fexceptions -funwind-tables -fno-common -Wformat -Wformat-security -Werror=format-security -fstack-protector-strong -fPIC -fmacro-prefix-map=/var/folders/8x/1m5v3n6d4mn62g9w_65vvt_r0000gn/T/tarantool_install1565297302=. -std=c11 -Wall -Wextra -Wno-strict-aliasing -Wno-char-subscripts -Wno-gnu-alignof-expression -Wno-cast-function-type
CPU: Apple M1 @ 8
JIT: Enabled
JIT: fold cse dce fwd dse narrow loop abc sink fuse
Duration: 1000000 iters

--- BENCH: 001_put_take_bench::bench_producer:producer-1
 1000000             44046 ns/op             22703 op/s     4137 B/op   +3946.04MB

--- BENCH: 001_put_take_bench::bench_producer:producer-2
 1000000             27519 ns/op             36339 op/s     4282 B/op   +4084.35MB

--- BENCH: 001_put_take_bench::bench_producer:producer-4
 1000000             21510 ns/op             46491 op/s     4500 B/op   +4291.69MB

--- BENCH: 001_put_take_bench::bench_producer:producer-8
 1000000             20804 ns/op             48068 op/s     4379 B/op   +4176.26MB

--- BENCH: 001_put_take_bench::bench_producer:producer-12
 1000000             20383 ns/op             49062 op/s     4372 B/op   +4169.59MB

--- BENCH: 001_put_take_bench::bench_producer:producer-16
 1000000             21495 ns/op             46523 op/s     4215 B/op   +4019.98MB

--- BENCH: 001_put_take_bench::bench_producer:producer-20
 1000000             23676 ns/op             42238 op/s     4258 B/op   +4061.49MB

--- BENCH: 001_put_take_bench::bench_producer:producer-24
 1000000             24456 ns/op             40891 op/s     4274 B/op   +4076.65MB

TODO

• reload/upgrade and feature switch
• composite primary keys
• switch - turn non-taken task into any state
• snatch - take away Taken task from its owner and turn it into R/W
• restart (renew, requeue?) - take task from W/D/Z states and put it into R/W
• it seems, that ttl/ttr/bury requires attrs field