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 Transitions

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