add "shared" sessions to queue

LeonidVas · LeonidVas · commit 2747a8e206a7 · 2020-11-08T19:49:34.000+03:00
Before the patch a connection to server was synonym of the queue session. Now the session has a unique UUID (returned by the "queue.identificate()" method), and one session can have many connections. The session will be deleted (all session tasks will be released) after the last connection is disconnected. To connect to an existing session, call "queue.identificate(uuid)" with the previously obtained UUID. Part of #85
diff --git a/README.md b/README.md
@@ -19,6 +19,7 @@ align="right">
   * [Fields of the \_queue space](#fields-of-the-_queue-space)
   * [Fields of the \_queue\_consumers space](#fields-of-the-_queue_consumers-space)
   * [Fields of the \_queue\_taken space](#fields-of-the-_queue_taken-space)
+  * [Fields of the \_queue\_session\_ids space](#fields-of-the-_queue_session_ids-space)
   * [Fields of the space associated with each queue](#fields-of-the-space-associated-with-each-queue)
 * [Installing](#installing)
 * [Using the queue module](#using-the-queue-module)
@@ -245,7 +246,7 @@ Consumers may be simply waiting for tasks to be put in the queues.
 
 ## Fields of the `_queue_consumers` space
 
-1. `session` - session (connection) ID of the client
+1. `session` - connection ID of the client
 1. `fid` - client fiber ID
 1. `tube_id` - queue ID, referring to the `tube_id` field in the `_queue`
 space; the client waits for tasks in this queue
@@ -257,8 +258,7 @@ processing a task in the queue.
 
 ## Fields of the `_queue_taken` space
 
-1. `session` - session (connection) ID of the client, referring to the
-`session_id` field of the `_queue_consumers` space
+1. `session_uuid` - session UUID (string)
 1. `tube_id` - queue ID, to which the task belongs
 1. `task_id` - task ID (of the task being taken)
 1. `time` - the time when the client began to execute the task
@@ -267,6 +267,14 @@ 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.
 
+The `_queue_session_ids` temporary space contains a map: box session id to the
+session UUID.
+
+## Fields of the `_queue_session_ids` space
+
+1. box_session_id - connection id (numeric)
+2. queue_session_uuid - session UUID (string)
+
 ## Fields of the space associated with each queue
 
 1. task_id - numeric - see below
diff --git a/queue/abstract.lua b/queue/abstract.lua
@@ -1,5 +1,6 @@
 local log      = require('log')
 local fiber    = require('fiber')
+local uuid     = require('uuid')
 
 local state    = require('queue.abstract.state')
 
@@ -75,6 +76,19 @@ local function tube_release_all_tasks(tube)
     log.info(prefix .. ('released %d tasks'):format(released))
 end
 
+--- Get session UUID by box session ID.
+local function get_quuid_by_sid(box_sid)
+    local session_ids = box.space._queue_session_ids.index.box_sid:get(box_sid)
+    local quuid
+    if session_ids ~= nil then
+        quuid = session_ids[2]
+    else
+        quuid = queue.identificate()
+    end
+
+    return quuid
+end
+
 -- tube methods
 local tube = {}
 
@@ -140,7 +154,8 @@ function tube.touch(self, id, delta)
         return
     end
 
-    local _taken = box.space._queue_taken:get{session.id(), self.tube_id, id}
+    local _taken = box.space._queue_taken:get{get_quuid_by_sid(session.id()),
+        self.tube_id, id}
     if _taken == nil then
         error("Task was not taken in the session")
     end
@@ -152,7 +167,8 @@ function tube.touch(self, id, delta)
 end
 
 function tube.ack(self, id)
-    local _taken = box.space._queue_taken:get{session.id(), self.tube_id, id}
+    local quuid = get_quuid_by_sid(session.id())
+    local _taken = box.space._queue_taken:get{quuid, self.tube_id, id}
     if _taken == nil then
         error("Task was not taken in the session")
     end
@@ -161,7 +177,7 @@ function tube.ack(self, id)
 
     self:peek(id)
     -- delete task
-    box.space._queue_taken:delete{session.id(), self.tube_id, id}
+    box.space._queue_taken:delete{quuid, self.tube_id, id}
     local result = self.raw:normalize_task(
         self.raw:delete(id):transform(2, 1, state.DONE)
     )
@@ -171,20 +187,20 @@ function tube.ack(self, id)
     return result
 end
 
-local function tube_release_internal(self, id, opts, session_id)
+local function tube_release_internal(self, id, opts, queue_uuid)
     opts = opts or {}
-    local _taken = box.space._queue_taken:get{session_id, self.tube_id, id}
+    local _taken = box.space._queue_taken:get{queue_uuid, self.tube_id, id}
     if _taken == nil then
         error("Task was not taken in the session")
     end
 
-    box.space._queue_taken:delete{session_id, self.tube_id, id}
+    box.space._queue_taken:delete{queue_uuid, self.tube_id, id}
     self:peek(id)
     return self.raw:normalize_task(self.raw:release(id, opts))
 end
 
 function tube.release(self, id, opts)
-    return tube_release_internal(self, id, opts, session.id())
+    return tube_release_internal(self, id, opts, get_quuid_by_sid(session.id()))
 end
 
 function tube.peek(self, id)
@@ -197,9 +213,10 @@ end
 
 function tube.bury(self, id)
     local task = self:peek(id)
-    local _taken = box.space._queue_taken:get{session.id(), self.tube_id, id}
+    local quuid = get_quuid_by_sid(session.id())
+    local _taken = box.space._queue_taken:get{quuid, self.tube_id, id}
     if _taken ~= nil then
-        box.space._queue_taken:delete{session.id(), self.tube_id, id}
+        box.space._queue_taken:delete{quuid, self.tube_id, id}
     end
     if task[2] == state.BURIED then
         return task
@@ -281,9 +298,11 @@ function tube.grant(self, user, args)
     tube_grant_space(user, '_queue', 'read')
     tube_grant_space(user, '_queue_consumers')
     tube_grant_space(user, '_queue_taken')
+    tube_grant_space(user, '_queue_session_ids')
     tube_grant_space(user, self.name)
 
     if args.call then
+        tube_grant_func(user, 'queue.identificate')
         local prefix = (args.prefix or 'queue.tube') .. ('.%s:'):format(self.name)
         tube_grant_func(user, prefix .. 'take')
         tube_grant_func(user, prefix .. 'touch')
@@ -362,7 +381,8 @@ local function make_self(driver, space, tube_name, tube_type, tube_id, opts)
             end
         -- task switched to taken - register in taken space
         elseif task[2] == state.TAKEN then
-            queue_taken:insert{session.id(), self.tube_id, task[1], fiber.time64()}
+            queue_taken:insert{get_quuid_by_sid(session.id()), self.tube_id,
+                task[1], fiber.time64()}
         end
         if stats_data ~= nil then
             queue.stat[space.name]:inc(stats_data)
@@ -393,17 +413,47 @@ local function make_self(driver, space, tube_name, tube_type, tube_id, opts)
     return self
 end
 
+--- Release all session tasks if the session has no actual connections.
+local function cleanup_session_tasks(queue_uuid)
+    local sesion_ids = box.space._queue_session_ids.index.queue_uuid:select(
+        queue_uuid,
+        {limit = 1})[1]
+    if sesion_ids ~= nil then
+        return
+    end
+
+    -- Release all session tasks.
+    while true do
+        local task = box.space._queue_taken.index.pk:min{queue_uuid}
+        if task == nil or task[1] ~= queue_uuid then
+            break
+        end
+
+        local tube_row = box.space._queue.index.tube_id:get{task[2]}
+        if tube_row == nil then
+            log.error("Inconsistent queue state: tube %d not found", task[2])
+            box.space._queue_taken:delete{queue_uuid, task[2], task[3]}
+        else
+            log.warn("Session %s closed, release task %s",
+                uuid.frombin(queue_uuid):str(), task[3])
+
+            tube_release_internal(queue.tube[tube_row[1]], task[3], nil,
+                queue_uuid)
+        end
+    end
+end
+
 function method._on_consumer_disconnect()
-    local waiter, fb, task, tube, id
-    id = session.id()
+    local sid = session.id()
+
     -- wakeup all waiters
     while true do
-        waiter = box.space._queue_consumers.index.pk:min{id}
+        local waiter = box.space._queue_consumers.index.pk:min{sid}
         if waiter == nil then
             break
         end
         -- Don't touch the other consumers
-        if waiter[1] ~= id then
+        if waiter[1] ~= sid then
             break
         end
         box.space._queue_consumers:delete{ waiter[1], waiter[2] }
@@ -414,24 +464,11 @@ function method._on_consumer_disconnect()
         end
     end
 
-    -- release all session tasks
-    while true do
-        task = box.space._queue_taken.index.pk:min{id}
-        if task == nil or task[1] ~= id then
-            break
-        end
-
-        tube = box.space._queue.index.tube_id:get{task[2]}
-        if tube == nil then
-            log.error("Inconsistent queue state: tube %d not found", task[2])
-            box.space._queue_taken:delete{task[1], task[2], task[3] }
-        else
-            log.warn("Consumer %s disconnected, release task %s(%s)",
-                id, task[3], tube[1])
-
-            tube_release_internal(queue.tube[tube[1]], task[3], nil, id)
-        end
-    end
+    -- Remove connection from the list of active connections and
+    -- release tasks if necessary.
+    local quuid = get_quuid_by_sid(sid)
+    box.space._queue_session_ids:delete{sid}
+    cleanup_session_tasks(quuid)
 end
 
 -- function takes tuples and recreates tube
@@ -495,6 +532,31 @@ function method.create_tube(tube_name, tube_type, opts)
     return self
 end
 
+--- Create everything that's needed to work with "shared" sessions.
+local function identification_init()
+    local queue_session_ids = box.space._queue_session_ids
+    if queue_session_ids == nil then
+        queue_session_ids = box.schema.create_space('_queue_session_ids', {
+            temporary = true,
+            format = {
+                {name = 'box_session_id', type = num_type()},
+                {name = 'queue_session_uuid', type = str_type()}
+            }
+        })
+
+        queue_session_ids:create_index('box_sid', {
+            type = 'tree',
+            parts = {1, num_type()},
+            unique = true
+        })
+        queue_session_ids:create_index('queue_uuid', {
+            type = 'tree',
+            parts = {2, str_type()},
+            unique = false
+        })
+    end
+end
+
 -- create or join infrastructure
 function method.start()
     -- tube_name, tube_id, space_name, tube_type, opts
@@ -527,7 +589,7 @@ function method.start()
         _cons = box.schema.create_space('_queue_consumers', {
             temporary = true,
             format = {
-                {name = 'session_id', type = num_type()},
+                {name = 'box_session_id', type = num_type()},
                 {name = 'fiber_id', type = num_type()},
                 {name = 'tube_id', type = num_type()},
                 {name = 'event_time', type = num_type()},
@@ -547,19 +609,27 @@ function method.start()
     end
 
     local _taken = box.space._queue_taken
+
+    -- Format of the temporary space has been changed.
+    -- So recreate it if needed.
+    if _taken ~= nil and _taken:format()[1]['type'] == 'unsigned' then
+        _taken:drop()
+        _taken = nil
+    end
+
     if _taken == nil then
         -- session_id, tube_id, task_id, time
         _taken = box.schema.create_space('_queue_taken', {
             temporary = true,
             format = {
-                {name = 'session_id', type = num_type()},
+                {name = 'queue_session_uuid', type = str_type()},
                 {name = 'tube_id', type = num_type()},
                 {name = 'task_id', type = num_type()},
                 {name = 'taken_time', type = num_type()}
             }})
         _taken:create_index('pk', {
             type = 'tree',
-            parts = {1, num_type(), 2, num_type(), 3, num_type()},
+            parts = {1, str_type(), 2, num_type(), 3, num_type()},
             unique = true})
 
         _taken:create_index('task',{
@@ -575,6 +645,8 @@ function method.start()
         tube_release_all_tasks(tube)
     end
 
+    identification_init()
+
     session.on_disconnect(queue._on_consumer_disconnect)
     return queue
 end
@@ -613,6 +685,38 @@ local function build_stats(space)
     return stats
 end
 
+--- Identificate the session.
+-- If "queue_uuid" == nil - create a new UUID for the session
+-- and return it, otherwise connect to the existing one.
+function method.identificate(queue_uuid)
+    local queue_session_ids = box.space._queue_session_ids
+    local sid = session.id()
+    local session_ids = queue_session_ids.index.box_sid:get(sid)
+    local quuid = session_ids and session_ids[2]
+
+    if queue_uuid == nil and quuid == nil then
+        -- Generate new UUID for the session.
+        quuid = uuid.bin()
+        queue_session_ids:insert{sid, quuid}
+    elseif queue_uuid ~= nil then
+        -- Identificate using a previously created session.
+
+        --Validate UUID.
+        uuid.frombin(queue_uuid)
+
+        if quuid ~= queue_uuid then
+            queue_session_ids:upsert({sid, queue_uuid}, {{'=', 2, queue_uuid}})
+            if quuid ~= nil then
+                -- Cleanup old session tasks.
+                cleanup_session_tasks(quuid)
+            end
+            quuid = queue_uuid
+        end
+    end
+
+    return quuid
+end
+
 queue.statistics = function(space)
     if space ~= nil then
         return build_stats(space)
diff --git a/t/170-work-with-quuid.t b/t/170-work-with-quuid.t
