session: full support for shared sessions

In order for the user to be able to finish servicing the task after
the queue switched from replica to master, the `_queue_inactive_session`
space was renamed to `_queue_shared_sessions` and had an `active` field
in it. When the `take` method is called, the session uuid is automatically
added to the `_queue_shared_sessions` space with `exp_time` set to zero
and `active` field set to true. When the client closes the session, this
entry becomes inactive (`active` = false) and `exp_time` is set.

At the start of the queue, we make all active sessions inactive, and set
an expiration time for them. If the `ttr` setting is not set, then we delete
all sessions.

If `ttr` is not set shared sessions is disabled.

Follows up #120

Author: 0x501D

README.md

@@ -280,10 +280,11 @@ is set to false.
 1. `connection_id` - connection id (numeric)
 2. `session_uuid` - session UUID (string)
 
-## Fields of the `_queue_inactive_sessions` space
+## Fields of the `_queue_shared_sessions` space
 
 1. `uuid` - session UUID (string)
 2. `exp_time` - session expiration time (numeric)
+3. `active` - session state (boolean)
 
 This space is temporary if `in_replicaset` is set to false.
 
@@ -378,12 +379,10 @@ Current queue state can be shown by using `queue.state()` method.
 In the `STARTUP` state, the queue is waiting for possible data synchronization
 with other cluster members by the time of the largest upstream lag multiplied
 by two. After that, all taken tasks are released, except for tasks with
-session uuid matching inactive sessions uuids. This makes possible to take
+session uuid matching shared sessions uuids. This makes possible to take
 a task, switch roles on the cluster, and release the task within the timeout
-specified by the `queue.cfg({ttr = N})` parameter. Note: all clients that `take()`
-and do not `ack()/release()` tasks must be disconnected before changing the role.
-And the last step in the `STARTUP` state is starting tube driver using new
-method called `start()`.
+specified by the `queue.cfg({ttr = N})` parameter. And the last step in the
+`STARTUP` state is starting tube driver using new method called `start()`.
 
 In the `RUNNING` state, the queue is working as usually. The `ENDING` state calls
 `stop()` method. in the `WAITING` state, the queue listens for a change in the

queue/abstract.lua

@@ -56,7 +56,7 @@ local function tube_release_all_orphaned_tasks(tube)
         local taken = box.space._queue_taken_2.index.task:get{
             tube.tube_id, task[1]
         }
-        if taken and session.exist_inactive(taken[4]) then
+        if taken and session.exist_shared(taken[4]) then
             log.info(prefix ..
                 ('skipping task: %d, tube_id: %d'):format(task[1],
                     tube.tube_id))
@@ -619,26 +619,26 @@ end
 -- 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
+local function switch_in_replicaset(replicaset_mode)
+    if replicaset_mode == nil then
         log.warn('queue: queue required after box.cfg{}')
-        mode = false
+        replicaset_mode = false
     end
 
     if not box.space._queue_taken_2 then
         return
     end
 
-    if box.space._queue_taken_2.temporary and mode == false then
+    if box.space._queue_taken_2.temporary and replicaset_mode == false then
         return
     end
 
-    if not box.space._queue_taken_2.temporary and mode == true then
+    if not box.space._queue_taken_2.temporary and replicaset_mode == true then
         return
     end
 
     box.schema.create_space('_queue_taken_2_mgr', {
-        temporary = not mode,
+        temporary = not replicaset_mode,
         format = {
             {name = 'tube_id', type = num_type()},
             {name = 'task_id', type = num_type()},
@@ -728,11 +728,11 @@ function method.start()
         box.space._queue_taken:drop()
     end
 
-    local _mode = queue.cfg['in_replicaset'] or false
+    local replicaset_mode = queue.cfg['in_replicaset'] or false
     if box.space._queue_taken_2 == nil then
         -- tube_id, task_id, connection_id, session_uuid, time
         box.schema.create_space('_queue_taken_2', {
-            temporary = not _mode,
+            temporary = not replicaset_mode,
             format = {
                 {name = 'tube_id', type = num_type()},
                 {name = 'task_id', type = num_type()},

queue/abstract/driver/fifottl.lua

@@ -170,7 +170,7 @@ local function fifottl_fiber(self)
         else
             -- When switching the master to the replica, the fiber will be stopped.
             if self.sync_chan:get(0.1) ~= nil then
-                print("Queue fifottl was stopped")
+                log.info("Queue fifottl fiber was stopped")
                 break
             end
         end

queue/abstract/driver/utubettl.lua

@@ -178,7 +178,7 @@ local function utubettl_fiber(self)
         else
             -- When switching the master to the replica, the fiber will be stopped.
             if self.sync_chan:get(0.1) ~= nil then
-                print("Queue utubettl was stopped")
+                log.info("Queue utubettl fiber was stopped")
                 break
             end
         end

queue/abstract/queue_session.lua

@@ -14,52 +14,60 @@ local queue_session = {}
 -- Replicaset mode switch.
 --
 -- Running a queue in master-replica mode requires that
--- `_queue_inactive_sessions` space was not temporary.
+-- `_queue_shared_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
+local function switch_in_replicaset(replicaset_mode)
+    if replicaset_mode == nil then
+        replicaset_mode = false
     end
 
-    if not box.space._queue_inactive_sessions then
+    if not box.space._queue_shared_sessions then
         return
     end
 
-    if box.space._queue_inactive_sessions.temporary and mode == false then
+    if box.space._queue_shared_sessions.temporary
+        and replicaset_mode == false then
         return
     end
 
-    if not box.space._queue_inactive_sessions.temporary and mode == true then
+    if not box.space._queue_shared_sessions.temporary
+        and replicaset_mode == true then
         return
     end
 
-    box.schema.create_space('_queue_inactive_sessions_mgr', {
-        temporary = not mode,
+    box.schema.create_space('_queue_shared_sessions_mgr', {
+        temporary = not replicaset_mode,
         format = {
             { name = 'uuid', type = str_type() },
-            { name = 'exp_time', type = num_type() }
+            { name = 'exp_time', type = num_type() },
+            { name = 'active', type = 'boolean' },
         }
     })
 
-    box.space._queue_inactive_sessions_mgr:create_index('uuid', {
+    box.space._queue_shared_sessions_mgr:create_index('uuid', {
         type = 'tree',
         parts = { 1, str_type() },
         unique = true
     })
+    box.space._queue_shared_sessions_mgr:create_index('active', {
+        type = 'tree',
+        parts = { 3, 'boolean', 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)
+    for _, tuple in box.space._queue_shared_sessions:pairs() do
+        box.space._queue_shared_sessions_mgr:insert(tuple)
     end
 
-    box.space._queue_inactive_sessions:drop()
-    box.space._queue_inactive_sessions_mgr:rename('_queue_inactive_sessions')
+    box.space._queue_shared_sessions:drop()
+    box.space._queue_shared_sessions_mgr:rename('_queue_shared_sessions')
 
     local status, err = pcall(box.commit)
     if not status then
-        error(('Error migrate _queue_inactive_sessions: %s'):format(tostring(err)))
+        error(('Error migrate _queue_shared_sessions: %s'):format(tostring(err)))
     end
 end
 
@@ -87,37 +95,63 @@ local function identification_init()
         })
     end
 
-    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,
+    local replicaset_mode = queue_session.cfg['in_replicaset'] or false
+    if box.space._queue_shared_sessions == nil then
+        box.schema.create_space('_queue_shared_sessions', {
+            temporary = not replicaset_mode,
             format = {
                 { name = 'uuid', type = str_type() },
-                { name = 'exp_time', type = num_type() }
+                { name = 'exp_time', type = num_type() },
+                { name = 'active', type = 'boolean' },
             }
         })
 
-        box.space._queue_inactive_sessions:create_index('uuid', {
+        box.space._queue_shared_sessions:create_index('uuid', {
             type = 'tree',
             parts = { 1, str_type() },
             unique = true
         })
+        box.space._queue_shared_sessions:create_index('active', {
+            type = 'tree',
+            parts = { 3, 'boolean', 1, str_type() },
+            unique = true
+        })
     else
         switch_in_replicaset(queue_session.cfg['in_replicaset'])
+
+        -- At the start of the queue, we make all active sessions inactive,
+        -- and set an expiration time for them. If the ttr setting is not set,
+        -- then we delete all sessions.
+        local ttr = queue_session.cfg['ttr'] or 0
+        if ttr > 0 then
+            for _, tuple in box.space._queue_shared_sessions.index.active:pairs{true} do
+                    box.space._queue_shared_sessions:update(tuple[1], {
+                        {'=', 2, util.event_time(ttr)},
+                        {'=', 3, false},
+                    })
+            end
+        else
+            if queue_session._on_session_remove ~= nil then
+                for _, tuple in box.space._queue_shared_sessions.index.uuid:pairs() do
+                    queue_session._on_session_remove(tuple[1])
+                end
+            end
+            box.space._queue_shared_sessions:truncate()
+        end
     end
 end
 
 local function cleanup_inactive_sessions()
     local cur_time = util.time()
 
-    for _, val in box.space._queue_inactive_sessions:pairs() do
+    for _, val in box.space._queue_shared_sessions.index.active:pairs{false} do
         local session_uuid = val[1]
         local exp_time = val[2]
         if cur_time >= exp_time then
             if queue_session._on_session_remove ~= nil then
                 queue_session._on_session_remove(session_uuid)
             end
-            box.space._queue_inactive_sessions:delete{session_uuid}
+            box.space._queue_shared_sessions:delete{session_uuid}
         end
     end
 end
@@ -160,6 +194,7 @@ local function identify(conn_id, session_uuid)
         -- Generate new UUID for the session.
         cur_uuid = uuid.bin()
         queue_session_ids:insert{conn_id, cur_uuid}
+        box.space._queue_shared_sessions:insert{cur_uuid, 0, true}
     elseif session_uuid ~= nil then
         -- Validate UUID.
         if not pcall(uuid.frombin, session_uuid) then
@@ -170,8 +205,8 @@ 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 = box.space._queue_inactive_sessions:get(session_uuid)
-        if ids_by_uuid == nil and inactive_session == nil then
+        local shared_session = box.space._queue_shared_sessions:get(session_uuid)
+        if ids_by_uuid == nil and shared_session == nil then
             error('The UUID ' .. uuid.frombin(session_uuid):str() ..
                 ' is unknown.')
         end
@@ -183,10 +218,15 @@ local function identify(conn_id, session_uuid)
             queue_session_ids:insert({conn_id, session_uuid})
             cur_uuid = session_uuid
         end
-    end
 
-    -- Exclude the session from inactive.
-    box.space._queue_inactive_sessions:delete{cur_uuid}
+        -- Make session active.
+        if shared_session then
+            box.space._queue_shared_sessions:update(shared_session[1], {
+                {'=', 2, 0},
+                {'=', 3, true},
+            })
+        end
+    end
 
     return cur_uuid
 end
@@ -209,21 +249,33 @@ local function disconnect(conn_id)
         { limit = 1 })[1]
 
     -- If a queue session doesn't have any active connections it should be
-    -- removed (if ttr is absent) or moved to the "inactive sessions" list.
+    -- removed (if ttr is absent) or change the `active` flag to make the
+    -- session inactive.
     if session_ids == nil then
         local ttr = queue_session.cfg['ttr'] or 0
         if ttr > 0 then
-            box.space._queue_inactive_sessions:insert{session_uuid, util.event_time(ttr)}
+            local tuple = box.space._queue_shared_sessions:get{session_uuid}
+            if tuple == nil then
+                box.space._queue_shared_sessions:insert{
+                    session_uuid, util.event_time(ttr), false
+                }
+            else
+                box.space._queue_shared_sessions:update(tuple[1], {
+                    {'=', 2, util.event_time(ttr)},
+                    {'=', 3, false},
+                })
+            end
         elseif queue_session._on_session_remove ~= nil then
             queue_session._on_session_remove(session_uuid)
+            box.space._queue_shared_sessions.index.uuid:delete{session_uuid}
         end
     end
 end
 
 local function grant(user)
     box.schema.user.grant(user, 'read, write', 'space', '_queue_session_ids',
         { if_not_exists = true })
-    box.schema.user.grant(user, 'read, write', 'space', '_queue_inactive_sessions',
+    box.schema.user.grant(user, 'read, write', 'space', '_queue_shared_sessions',
         { if_not_exists = true })
 end
 
@@ -246,7 +298,7 @@ local function validate_opts(opts)
                 error('Invalid value of ttr: ' .. tostring(val))
             end
         elseif key == 'in_replicaset' then
-            -- do nothing
+            -- Do nothing.
         else
             error('Unknown option ' .. tostring(key))
         end
@@ -269,8 +321,8 @@ local function cfg(self, opts)
     switch_in_replicaset(self['in_replicaset'])
 end
 
-local function exist_inactive(session_uuid)
-    if box.space._queue_inactive_sessions:get{session_uuid} then
+local function exist_shared(session_uuid)
+    if box.space._queue_shared_sessions:get{session_uuid} then
         return true
     end
 
@@ -287,7 +339,7 @@ local method = {
     on_session_remove = on_session_remove,
     start = start,
     stop = stop,
-    exist_inactive = exist_inactive
+    exist_shared = exist_shared
 }
 
 return setmetatable(queue_session, { __index = method })