cfg: add in_replicaset option

Running a queue in master-replica mode requires that
_queue_inactive_sessions and _queue_taken_2 were not
temporary spases. This results in a 20 percent performance loss.
The `in_replicaset` option allows you to get the same performance
in single mode.

Follows up #120

Author: 0x501D

README.md

@@ -271,8 +271,9 @@ tuples for each job which is processing a task in the queue.
 1. `session_uuid` - session UUID (string)
 1. `time` - the time when the client began to execute the task
 
-The `_queue_session_ids` temporary space contains a map: connection id (box
-session id) to the session UUID.
+The `_queue_session_ids` space contains a map: connection id (box
+session id) to the session UUID. This space is temporary if `in_replicaset`
+is set to false.
 
 ## Fields of the `_queue_session_ids` space
 
@@ -284,6 +285,8 @@ session id) to the session UUID.
 1. `uuid` - session UUID (string)
 2. `exp_time` - session expiration time (numeric)
 
+This space is temporary if `in_replicaset` is set to false.
+
 Also, there is a space which is associated with each queue,
 which is named in the `space` field of the `_queue` space.
 The associated space contains one tuple for each task.
@@ -456,6 +459,9 @@ If an invalid value or an unknown option is used, an error will be thrown.
 Available `options`:
 * `ttr` - time to release in seconds. The time after which, if there is no active
 connection in the session, it will be released with all its tasks.
+* `in_replicaset` - enable replication mode. Must be true if the queue is used
+in master and replica mode. With replication mode enabled, the potential loss of
+performance can be ~20% compared to single mode. Default value is false.
 
 ## Session identify
 
@@ -746,6 +752,11 @@ Usage example:
 
 ```lua
 -- Instance file for the master.
+queue = require("queue")
+-- Queue is in replicaset.
+-- Clean up session after 5 minutes after disconnect.
+queue.cfg({ttr = 300, in_replicaset = true})
+
 box.cfg{
   listen = 3301,
   replication = {'replicator:password@127.0.0.1:3301',  -- Master URI.
@@ -758,23 +769,23 @@ box.once("schema", function()
    box.schema.user.grant('replicator', 'replication') -- grant replication role
 end)
 
-queue = require("queue")
-queue.cfg({ttr = 300}) -- Clean up session after 5 minutes after disconnect.
 require('console').start()
 os.exit()
 ```
 
 ```lua
 -- Instance file for the replica.
+queue = require("queue")
+-- Queue is in replicaset.
+-- Clean up session after 5 minutes after disconnect.
+queue.cfg({ttr = 300, in_replicaset = true})
 box.cfg{
   listen = 3302,
   replication = {'replicator:password@127.0.0.1:3301',  -- Master URI.
                  'replicator:password@127.0.0.1:3302'}, -- Replica URI.
   read_only = true
 }
 
-queue = require("queue")
-queue.cfg({ttr = 300}) -- Clean up session after 5 minutes after disconnect.
 require('console').start()
 os.exit()
 ```

queue/abstract.lua

@@ -406,13 +406,14 @@ local function make_self(driver, space, tube_name, tube_type, tube_id, opts)
         -- task was removed
         if task == nil then return end
 
+        -- We cannot use a local variable to access the space `_queue_taken_2`
+        -- because it can be recreated in `switch_in_replicaset()`.
         local queue_consumers = box.space._queue_consumers
-        local queue_taken     = box.space._queue_taken_2
 
         -- if task was taken and become other state
-        local taken = queue_taken.index.task:get{tube_id, task[1]}
+        local taken = box.space._queue_taken_2.index.task:get{tube_id, task[1]}
         if taken ~= nil then
-            queue_taken:delete{taken[1], taken[2]}
+            box.space._queue_taken_2:delete{taken[1], taken[2]}
         end
         -- task switched to ready (or new task)
         if task[2] == state.READY then
@@ -432,7 +433,7 @@ local function make_self(driver, space, tube_name, tube_type, tube_id, opts)
         elseif task[2] == state.TAKEN then
             local conn_id = connection.id()
             local session_uuid = session.identify(conn_id)
-            queue_taken:insert{
+            box.space._queue_taken_2:insert{
                 self.tube_id,
                 task[1],
                 conn_id,
@@ -610,6 +611,67 @@ function method.create_tube(tube_name, tube_type, opts)
     return self
 end
 
+--
+-- Replicaset mode switch.
+--
+-- Running a queue in master-replica mode requires that
+-- `_queue_taken_2` space was not temporary.
+-- When the queue is running in single mode,
+-- the space is converted to temporary mode to increase performance.
+--
+local function switch_in_replicaset(mode)
+    if mode == nil then
+        log.warn('queue: queue required after box.cfg{}')
+        mode = false
+    end
+
+    if not box.space._queue_taken_2 then
+        return
+    end
+
+    if box.space._queue_taken_2.temporary and mode == false then
+        return
+    end
+
+    if not box.space._queue_taken_2.temporary and mode == true then
+        return
+    end
+
+    box.schema.create_space('_queue_taken_2_mgr', {
+        temporary = not mode,
+        format = {
+            {name = 'tube_id', type = num_type()},
+            {name = 'task_id', type = num_type()},
+            {name = 'connection_id', type = num_type()},
+            {name = 'session_uuid', type = str_type()},
+            {name = 'taken_time', type = num_type()}
+        }})
+
+    box.space._queue_taken_2_mgr:create_index('task', {
+        type = 'tree',
+        parts = {1, num_type(), 2, num_type()},
+        unique = true
+    })
+    box.space._queue_taken_2_mgr:create_index('uuid', {
+        type = 'tree',
+        parts = {4, str_type()},
+        unique = false
+    })
+
+    box.begin() -- Disable implicit yields until the transaction ends.
+    for _, tuple in box.space._queue_taken_2:pairs() do
+        box.space._queue_taken_2_mgr:insert(tuple)
+    end
+
+    box.space._queue_taken_2:drop()
+    box.space._queue_taken_2_mgr:rename('_queue_taken_2')
+
+    local status, err = pcall(box.commit)
+    if not status then
+        error(('Error migrate _queue_taken_2: %s'):format(tostring(err)))
+    end
+end
+
 -- create or join infrastructure
 function method.start()
     -- tube_name, tube_id, space_name, tube_type, opts
@@ -666,11 +728,11 @@ function method.start()
         box.space._queue_taken:drop()
     end
 
-    local _taken = box.space._queue_taken_2
-    if _taken == nil then
+    local _mode = queue.cfg['in_replicaset'] or false
+    if box.space._queue_taken_2 == nil then
         -- tube_id, task_id, connection_id, session_uuid, time
-        _taken = box.schema.create_space('_queue_taken_2', {
-            temporary = false,
+        box.schema.create_space('_queue_taken_2', {
+            temporary = not _mode,
             format = {
                 {name = 'tube_id', type = num_type()},
                 {name = 'task_id', type = num_type()},
@@ -679,22 +741,18 @@ function method.start()
                 {name = 'taken_time', type = num_type()}
             }})
 
-        _taken:create_index('task', {
+        box.space._queue_taken_2:create_index('task', {
             type = 'tree',
             parts = {1, num_type(), 2, num_type()},
             unique = true
         })
-        _taken:create_index('uuid', {
+        box.space._queue_taken_2:create_index('uuid', {
             type = 'tree',
             parts = {4, str_type()},
             unique = false
         })
     else
-        -- Upgrade space, queue states require that this space
-        -- was not temporary.
-        if _taken.temporary then
-            _taken:alter{temporary = false}
-        end
+        switch_in_replicaset(queue.cfg['in_replicaset'])
     end
 
     for _, tube_tuple in _queue:pairs() do
@@ -787,16 +845,28 @@ local function cfg(self, opts)
     opts = opts or {}
     local session_opts = {}
 
+    -- Set default in_replicaset value.
+    if opts['in_replicaset'] == nil then
+        opts['in_replicaset'] = false
+    end
+
     -- Check all options before configuring so that
     -- the configuration is done transactionally.
     for key, val in pairs(opts) do
         if key == 'ttr' then
             session_opts[key] = val
+        elseif key == 'in_replicaset' then
+            if type(val) ~= 'boolean' then
+                error('Invalid value of in_replicaset: ' .. tostring(val))
+            end
+            session_opts[key] = val
         else
             error('Unknown option ' .. tostring(key))
         end
     end
 
+    switch_in_replicaset(opts['in_replicaset'])
+
     -- Configuring the queue_session module.
     session.cfg(session_opts)
 

queue/abstract/queue_session.lua

@@ -10,6 +10,59 @@ local str_type = qc.str_type
 
 local queue_session = {}
 
+--
+-- Replicaset mode switch.
+--
+-- Running a queue in master-replica mode requires that
+-- `_queue_inactive_sessions` space was not temporary.
+-- When the queue is running in single mode,
+-- the space is converted to temporary mode to increase performance.
+--
+local function switch_in_replicaset(mode)
+    if mode == nil then
+        mode = false
+    end
+
+    if not box.space._queue_inactive_sessions then
+        return
+    end
+
+    if box.space._queue_inactive_sessions.temporary and mode == false then
+        return
+    end
+
+    if not box.space._queue_inactive_sessions.temporary and mode == true then
+        return
+    end
+
+    box.schema.create_space('_queue_inactive_sessions_mgr', {
+        temporary = not mode,
+        format = {
+            { name = 'uuid', type = str_type() },
+            { name = 'exp_time', type = num_type() }
+        }
+    })
+
+    box.space._queue_inactive_sessions_mgr:create_index('uuid', {
+        type = 'tree',
+        parts = { 1, str_type() },
+        unique = true
+    })
+
+    box.begin() -- Disable implicit yields until the transaction ends.
+    for _, tuple in box.space._queue_inactive_sessions:pairs() do
+        box.space._queue_inactive_sessions_mgr:insert(tuple)
+    end
+
+    box.space._queue_inactive_sessions:drop()
+    box.space._queue_inactive_sessions_mgr:rename('_queue_inactive_sessions')
+
+    local status, err = pcall(box.commit)
+    if not status then
+        error(('Error migrate _queue_inactive_sessions: %s'):format(tostring(err)))
+    end
+end
+
 --- Create everything that's needed to work with "shared" sessions.
 local function identification_init()
     local queue_session_ids = box.space._queue_session_ids
@@ -34,21 +87,23 @@ local function identification_init()
         })
     end
 
-    local queue_inactive_sessions = box.space._queue_inactive_sessions
-    if queue_inactive_sessions == nil then
-        queue_inactive_sessions = box.schema.create_space('_queue_inactive_sessions', {
-            temporary = false,
+    local _mode = queue_session.cfg['in_replicaset'] or false
+    if box.space._queue_inactive_sessions == nil then
+        box.schema.create_space('_queue_inactive_sessions', {
+            temporary = not _mode,
             format = {
                 { name = 'uuid', type = str_type() },
                 { name = 'exp_time', type = num_type() }
             }
         })
 
-        queue_inactive_sessions:create_index('uuid', {
+        box.space._queue_inactive_sessions:create_index('uuid', {
             type = 'tree',
             parts = { 1, str_type() },
             unique = true
         })
+    else
+        switch_in_replicaset(queue_session.cfg['in_replicaset'])
     end
 end
 
@@ -95,7 +150,6 @@ end
 -- an error will be thrown.
 local function identify(conn_id, session_uuid)
     local queue_session_ids = box.space._queue_session_ids
-    local queue_inactive_sessions = box.space._queue_inactive_sessions
     local session_ids = queue_session_ids:get(conn_id)
     local cur_uuid = session_ids and session_ids[2]
 
@@ -116,7 +170,7 @@ local function identify(conn_id, session_uuid)
         -- Check that a session with this uuid exists.
         local ids_by_uuid = queue_session_ids.index.uuid:select(
             session_uuid, { limit = 1 })[1]
-        local inactive_session = queue_inactive_sessions:get(session_uuid)
+        local inactive_session = box.space._queue_inactive_sessions:get(session_uuid)
         if ids_by_uuid == nil and inactive_session == nil then
             error('The UUID ' .. uuid.frombin(session_uuid):str() ..
                 ' is unknown.')
@@ -132,7 +186,7 @@ local function identify(conn_id, session_uuid)
     end
 
     -- Exclude the session from inactive.
-    queue_inactive_sessions:delete{cur_uuid}
+    box.space._queue_inactive_sessions:delete{cur_uuid}
 
     return cur_uuid
 end
@@ -148,7 +202,6 @@ end
 -- release its tasks if necessary.
 local function disconnect(conn_id)
     local queue_session_ids = box.space._queue_session_ids
-    local queue_inactive_sessions = box.space._queue_inactive_sessions
     local session_uuid = queue_session.identify(conn_id)
 
     queue_session_ids:delete{conn_id}
@@ -160,7 +213,7 @@ local function disconnect(conn_id)
     if session_ids == nil then
         local ttr = queue_session.cfg['ttr'] or 0
         if ttr > 0 then
-            queue_inactive_sessions:insert{session_uuid, util.event_time(ttr)}
+            box.space._queue_inactive_sessions:insert{session_uuid, util.event_time(ttr)}
         elseif queue_session._on_session_remove ~= nil then
             queue_session._on_session_remove(session_uuid)
         end
@@ -192,6 +245,8 @@ local function validate_opts(opts)
             if type(val) ~= 'number' or val < 0 then
                 error('Invalid value of ttr: ' .. tostring(val))
             end
+        elseif key == 'in_replicaset' then
+            -- do nothing
         else
             error('Unknown option ' .. tostring(key))
         end
@@ -210,6 +265,8 @@ local function cfg(self, opts)
     for key, val in pairs(opts) do
         self[key] = val
     end
+
+    switch_in_replicaset(self['in_replicaset'])
 end
 
 local function exist_inactive(session_uuid)

queue/abstract/queue_state.lua

@@ -44,8 +44,9 @@ end
 
 local function max_lag()
     local max_lag = 0
+    local n_replica = table.maxn(box.info.replication)
 
-    for i = 1, #box.info.replication do
+    for i = 1, n_replica do
         if box.info.replication[i].upstream then
             local lag = box.info.replication[i].upstream.lag
             if lag > max_lag then