core: add master-replica switching support

This patch adds ability to use queue in a master-replica scheme.
The queue will monitor the operation mode of the tarantool and
perform the necessary actions accordingly.

This patch adds five states for queue: INIT, STARTUP, RUNNING, ENDING
and WAITING. When the tarantool is launched for the first time,
the state of the queue is always INIT until box.info.ro is false.

States switching scheme:

                       +-----------+
                       |  RUNNING  |
                       +-----------+
                        ^         | (rw -> ro)
                        |         v
+------+       +---------+       +--------+
| INIT | --->  | STARTUP |       | ENDING |
+------+       +---------+       +--------+
                        ^         |
             (ro -> rw) |         v
                       +-----------+
                       |  WAITING  |
                       +-----------+

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
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(). Each tube driver must implement start() and stop()
methods. The start() method should start the driver fibers, if any, in other
words, initialize all asynchronous work with the tubes space. The stop()
methods shuld stop the driver fibers, if any, respectively.

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
read_only flag.

All states except INIT is controlled by new fiber called 'queue_state_fiber'.

A new release_all() method has also been added, which forcibly returns all
taken tasks to a ready state. This method can be called per tube.

Closes #120

Author: 0x501D

queue-scm-1.rockspec

@@ -19,6 +19,7 @@ build = {
         ['queue.abstract']                   = 'queue/abstract.lua',
         ['queue.abstract.state']             = 'queue/abstract/state.lua',
         ['queue.abstract.queue_session']     = 'queue/abstract/queue_session.lua',
+        ['queue.abstract.queue_state']       = 'queue/abstract/queue_state.lua',
         ['queue.abstract.driver.fifottl']    = 'queue/abstract/driver/fifottl.lua',
         ['queue.abstract.driver.utubettl']   = 'queue/abstract/driver/utubettl.lua',
         ['queue.abstract.driver.fifo']       = 'queue/abstract/driver/fifo.lua',

queue/CMakeLists.txt

@@ -10,6 +10,8 @@ install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/state.lua
         DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract)
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/queue_session.lua
         DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract)
+install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/queue_state.lua
+        DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract)
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/fifo.lua
         DESTINATION ${TARANTOOL_INSTALL_LUADIR}/${PROJECT_NAME}/abstract/driver/)
 install(FILES ${CMAKE_CURRENT_SOURCE_DIR}/abstract/driver/utube.lua

queue/abstract.lua

@@ -4,6 +4,7 @@ local uuid     = require('uuid')
 
 local session  = require('queue.abstract.queue_session')
 local state    = require('queue.abstract.state')
+local queue_state = require('queue.abstract.queue_state')
 
 local util     = require('queue.util')
 local qc       = require('queue.compat')
@@ -36,7 +37,8 @@ local queue = {
     stat = {}
 }
 
-local function tube_release_all_tasks(tube)
+-- Release tasks that don't have session_uuid stored in inactive sessions.
+local function tube_release_all_orphaned_tasks(tube)
     local prefix = ('queue: [tube "%s"] '):format(tube.name)
 
     -- We lean on stable iterators in this function.
@@ -51,8 +53,17 @@ local function tube_release_all_tasks(tube)
     log.info(prefix .. 'releasing all taken task (may take a while)')
     local released = 0
     for _, task in tube.raw:tasks_by_state(state.TAKEN) do
-        tube.raw:release(task[1], {})
-        released = released + 1
+        local taken = box.space._queue_taken_2.index.task:get{
+            tube.tube_id, task[1]
+        }
+        if taken and session.exist_inactive(taken[4]) then
+            log.info(prefix ..
+                ('skipping task: %d, tube_id: %d'):format(task[1],
+                    tube.tube_id))
+        else
+            tube.raw:release(task[1], {})
+            released = released + 1
+        end
     end
     log.info(prefix .. ('released %d tasks'):format(released))
 end
@@ -72,7 +83,20 @@ end
 -- tube methods
 local tube = {}
 
+-- This check must be called from all public tube methods.
+local function check_state()
+    if queue_state.get() ~= queue_state.states.RUNNING then
+        log.error(('Queue is in %s state'):format(queue_state.show()))
+        return false
+    end
+
+    return true
+end
+
 function tube.put(self, data, opts)
+    if not check_state() then
+        return nil
+    end
     opts = opts or {}
     local task = self.raw:put(data, opts)
     return self.raw:normalize_task(task)
@@ -82,6 +106,9 @@ local conds = {}
 local releasing_connections = {}
 
 function tube.take(self, timeout)
+    if not check_state() then
+        return nil
+    end
     timeout = util.time(timeout or util.TIMEOUT_INFINITY)
     local task = self.raw:take()
     if task ~= nil then
@@ -120,6 +147,9 @@ function tube.take(self, timeout)
 end
 
 function tube.touch(self, id, delta)
+    if not check_state() then
+        return
+    end
     if delta == nil then
         return
     end
@@ -143,6 +173,9 @@ function tube.touch(self, id, delta)
 end
 
 function tube.ack(self, id)
+    if not check_state() then
+        return nil
+    end
     check_task_is_taken(self.tube_id, id)
     local tube = box.space._queue:get{self.name}
     local space_name = tube[3]
@@ -169,10 +202,32 @@ local function tube_release_internal(self, id, opts, session_uuid)
 end
 
 function tube.release(self, id, opts)
+    if not check_state() then
+        return nil
+    end
     return tube_release_internal(self, id, opts)
 end
 
+-- Release all tasks.
+function tube.release_all(self)
+    if not check_state() then
+        return
+    end
+    local prefix = ('queue: [tube "%s"] '):format(self.name)
+
+    log.info(prefix .. 'releasing all taken task (may take a while)')
+    local released = 0
+    for _, task in self.raw:tasks_by_state(state.TAKEN) do
+        self.raw:release(task[1], {})
+        released = released + 1
+    end
+    log.info(('%s released %d tasks'):format(prefix, released))
+end
+
 function tube.peek(self, id)
+    if not check_state() then
+        return nil
+    end
     local task = self.raw:peek(id)
     if task == nil then
         error(("Task %s not found"):format(tostring(id)))
@@ -181,6 +236,9 @@ function tube.peek(self, id)
 end
 
 function tube.bury(self, id)
+    if not check_state() then
+        return nil
+    end
     local task = self:peek(id)
     local is_taken, _ = pcall(check_task_is_taken, self.tube_id, id)
     if is_taken then
@@ -193,17 +251,26 @@ function tube.bury(self, id)
 end
 
 function tube.kick(self, count)
+    if not check_state() then
+        return nil
+    end
     count = count or 1
     return self.raw:kick(count)
 end
 
 function tube.delete(self, id)
+    if not check_state() then
+        return nil
+    end
     self:peek(id)
     return self.raw:normalize_task(self.raw:delete(id))
 end
 
 -- drop tube
 function tube.drop(self)
+    if not check_state() then
+        return nil
+    end
     local tube_name = self.name
 
     local tube = box.space._queue:get{tube_name}
@@ -238,6 +305,9 @@ end
 -- truncate tube
 -- (delete everything from tube)
 function tube.truncate(self)
+    if not check_state() then
+        return
+    end
     self.raw:truncate()
 end
 
@@ -248,6 +318,9 @@ function tube.on_task_change(self, cb)
 end
 
 function tube.grant(self, user, args)
+    if not check_state() then
+        return
+    end
     local function tube_grant_space(user, name, tp)
         box.schema.user.grant(user, tp or 'read,write', 'space', name, {
             if_not_exists = true,
@@ -414,6 +487,10 @@ local function release_session_tasks(session_uuid)
     end
 end
 
+function method.state()
+    return queue_state.show()
+end
+
 function method._on_consumer_disconnect()
     local conn_id = connection.id()
 
@@ -438,6 +515,37 @@ function method._on_consumer_disconnect()
     session.disconnect(conn_id)
 end
 
+-- Function takes new queue state.
+-- The "RUNNING" and "WAITING" states do not require additional actions.
+local function on_state_change(state)
+    if state == queue_state.states.STARTUP then
+        for name, tube in pairs(queue.tube) do
+            tube_release_all_orphaned_tasks(tube)
+            log.info('queue: [tube "%s"] start driver', name)
+            if not tube.raw.start then
+                log.warn('queue: [tube "%s"] method start is not implemented',
+                    tube.name)
+            else
+                tube.raw:start()
+            end
+        end
+        session.start()
+    elseif state == queue_state.states.ENDING then
+        for name, tube in pairs(queue.tube) do
+            log.info('queue: [tube "%s"] stop driver', name)
+            if not tube.raw.stop then
+                log.warn('queue: [tube "%s"] method stop is not implemented',
+                    tube.name)
+            else
+                tube.raw:stop()
+            end
+        end
+        session.stop()
+    else
+        error('on_state_change: unexpected queue state')
+    end
+end
+
 -- function takes tuples and recreates tube
 local function recreate_tube(tube_tuple)
     local name, id, space_name, tube_type, opts = tube_tuple:unpack()
@@ -457,6 +565,9 @@ end
 -------------------------------------------------------------------------------
 -- create tube
 function method.create_tube(tube_name, tube_type, opts)
+    if not check_state() then
+        return
+    end
     opts = opts or {}
     if opts.if_not_exists == nil then
         opts.if_not_exists = false
@@ -559,7 +670,7 @@ function method.start()
     if _taken == nil then
         -- tube_id, task_id, connection_id, session_uuid, time
         _taken = box.schema.create_space('_queue_taken_2', {
-            temporary = true,
+            temporary = false,
             format = {
                 {name = 'tube_id', type = num_type()},
                 {name = 'task_id', type = num_type()},
@@ -578,6 +689,12 @@ function method.start()
             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
     end
 
     for _, tube_tuple in _queue:pairs() do
@@ -586,14 +703,15 @@ function method.start()
         if queue.driver[tube_tuple[4]] ~= nil then
             local tube = recreate_tube(tube_tuple)
             -- gh-66: release all taken tasks on start
-            tube_release_all_tasks(tube)
+            tube_release_all_orphaned_tasks(tube)
         end
     end
 
     session.on_session_remove(release_session_tasks)
     session.start()
 
     connection.on_disconnect(queue._on_consumer_disconnect)
+    queue_state.init(on_state_change)
     return queue
 end
 
@@ -617,7 +735,7 @@ function method.register_driver(driver_name, tube_ctr)
         if tube_tuple[4] == driver_name then
             local tube = recreate_tube(tube_tuple)
             -- Release all task for tube on start.
-            tube_release_all_tasks(tube)
+            tube_release_all_orphaned_tasks(tube)
         end
     end
 end

queue/abstract/driver/fifo.lua

@@ -148,4 +148,14 @@ function method.truncate(self)
     self.space:truncate()
 end
 
+-- This driver has no background activity.
+-- Implement dummy methods for the API requirement.
+function method.start()
+    return
+end
+
+function method.stop()
+    return
+end
+
 return tube

queue/abstract/driver/fifottl.lua

@@ -159,6 +159,7 @@ local function fifottl_fiber(self)
     while true do
         if box.info.ro == false then
             local stat, err = pcall(fifottl_fiber_iteration, self, processed)
+
             if not stat and not (err.code == box.error.READONLY) then
                 log.error("error catched: %s", tostring(err))
                 log.error("exiting fiber '%s'", fiber.name())
@@ -167,7 +168,11 @@ local function fifottl_fiber(self)
                 processed = err
             end
         else
-            fiber.sleep(0.1)
+            -- 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")
+                break
+            end
         end
     end
 end
@@ -191,6 +196,7 @@ function tube.new(space, on_task_change, opts)
 
     self.cond  = qc.waiter()
     self.fiber = fiber.create(fifottl_fiber, self)
+    self.sync_chan = fiber.channel()
 
     return self
 end
@@ -365,4 +371,20 @@ function method.truncate(self)
     self.space:truncate()
 end
 
+function method.start(self)
+    if self.fiber then
+        return
+    end
+    self.fiber = fiber.create(fifottl_fiber, self)
+end
+
+function method.stop(self)
+    if not self.fiber then
+        return
+    end
+    self.cond:signal(self.fiber:id())
+    self.sync_chan:put(true)
+    self.fiber = nil
+end
+
 return tube

queue/abstract/driver/utube.lua

@@ -175,4 +175,14 @@ function method.truncate(self)
     self.space:truncate()
 end
 
+-- This driver has no background activity.
+-- Implement dummy methods for the API requirement.
+function method.start()
+    return
+end
+
+function method.stop()
+    return
+end
+
 return tube