drivers: do not exit the ttl fiber on an error

CHANGELOG.md

@@ -13,6 +13,13 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 
+- The ttl fiber of the `fifottl`, `limfifottl` and `utubettl` drivers exited
+  on any unexpected error and could not be restarted, so ttl/ttr/delay
+  processing stopped until the next ro -> rw switch. Now it logs the error
+  and retries after a delay (gh-263).
+- `stop()` of the `fifottl`, `limfifottl` and `utubettl` drivers blocked
+  forever or did not stop the ttl fiber while the instance was in rw mode,
+  and `drop()` of `fifottl`/`limfifottl` leaked the fiber (gh-262).
 - The `on_task_change` callback got an expired task with a `NULL` status
   instead of `DONE` (`fifottl`, `utubettl`) (#255).
 - `kick()` corrupted the ready buffer of the `utube` driver in the

queue/abstract/driver/fifottl.lua

@@ -87,6 +87,16 @@ local delayed_state = { state.DELAYED }
 local ttl_states    = { state.READY, state.BURIED }
 local ttr_state     = { state.TAKEN }
 
+-- Delay before the next ttl iteration after an unexpected error.
+local ERROR_RETRY_DELAY = 1
+
+-- The ttl fiber works while it is the fiber registered in the tube.
+-- start() and stop() change the registration, so a stop request is
+-- noticed at the next check both in rw and in ro mode.
+local function is_registered_fiber(self)
+    return self.fiber ~= nil and self.fiber:id() == fiber.id()
+end
+
 local function fifottl_fiber_iteration(self, processed)
     local now       = util.time()
     local task      = nil
@@ -145,7 +155,10 @@ local function fifottl_fiber_iteration(self, processed)
         -- free refcounter
         estimated = estimated > 0 and estimated or 0
         processed = 0
-        self.cond:wait(estimated)
+        -- Do not fall asleep if stop() has been called meanwhile.
+        if is_registered_fiber(self) then
+            self.cond:wait(estimated)
+        end
     end
 
     return processed
@@ -157,25 +170,30 @@ local function fifottl_fiber(self)
     log.info("Started queue fifottl fiber")
     local processed = 0
 
-    while true do
+    while is_registered_fiber(self) 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
+                -- Do not exit: a dead fiber can not be restarted until the
+                -- next ro -> rw switch (gh-263). The error may well be
+                -- transient (a failed WAL write, a transaction conflict, a
+                -- user on_task_change callback), so back off and retry.
                 log.error("error catched: %s", tostring(err))
-                log.error("exiting fiber '%s'", fiber.name())
-                return 1
+                log.error("ttl fiber '%s' retries in %d sec", fiber.name(),
+                    ERROR_RETRY_DELAY)
+                fiber.sleep(ERROR_RETRY_DELAY)
             elseif stat then
                 processed = err
             end
         else
-            -- When switching the master to the replica, the fiber will be stopped.
-            if self.sync_chan:get(0.1) ~= nil then
-                log.info("Queue fifottl fiber was stopped")
-                break
-            end
+            -- When switching the master to the replica, the fiber will be
+            -- stopped by the queue state machine.
+            self.cond:wait(0.1)
         end
     end
+
+    log.info("Queue fifottl fiber was stopped")
 end
 
 -- start tube on space
@@ -196,8 +214,7 @@ function tube.new(space, on_task_change, opts)
     }, { __index = method })
 
     self.cond  = qc.waiter()
-    self.fiber = fiber.create(fifottl_fiber, self)
-    self.sync_chan = fiber.channel()
+    self:start()
 
     return self
 end
@@ -416,16 +433,30 @@ function method.start(self)
     if self.fiber then
         return
     end
-    self.fiber = fiber.create(fifottl_fiber, self)
+    -- fiber.new(): the fiber must not run before it is registered in
+    -- self.fiber, otherwise it would see itself as already stopped.
+    self.fiber = fiber.new(fifottl_fiber, self)
+    self.fiber:name('fifottl')
+    self.fiber:set_joinable(true)
 end
 
 function method.stop(self)
-    if not self.fiber then
+    local ttl_fiber = self.fiber
+    if ttl_fiber == nil then
         return
     end
-    self.cond:signal(self.fiber:id())
-    self.sync_chan:put(true)
+    -- Deregister the fiber, wake it up and wait until it exits, so that
+    -- the caller (e.g. drop()) can not race with an in-flight iteration.
     self.fiber = nil
+    self.cond:signal(ttl_fiber:id())
+    if ttl_fiber:id() ~= fiber.id() then
+        ttl_fiber:join()
+    end
+end
+
+function method.drop(self)
+    self:stop()
+    box.space[self.space.name]:drop()
 end
 
 return tube

queue/abstract/driver/limfifottl.lua

@@ -39,9 +39,27 @@ function tube.new(space, on_task_change, opts)
         return self.space:len()
     end
 
+    -- The ttl fiber is registered in the parent object: forward the
+    -- lifecycle methods there so that stop() does not shadow the
+    -- registration in this wrapper.
+    local start = function (self)
+        return state.parent:start()
+    end
+
+    local stop = function (self)
+        return state.parent:stop()
+    end
+
+    local drop = function (self)
+        return state.parent:drop()
+    end
+
     return setmetatable({
         put = put,
-        len = len
+        len = len,
+        start = start,
+        stop = stop,
+        drop = drop,
     }, {__index = state.parent})
 end
 

queue/abstract/driver/utubettl.lua

@@ -201,6 +201,16 @@ local function begin_if_not_in_txn(self)
     end
 end
 
+-- Delay before the next ttl iteration after an unexpected error.
+local ERROR_RETRY_DELAY = 1
+
+-- The ttl fiber works while it is the fiber registered in the tube.
+-- start() and stop() change the registration, so a stop request is
+-- noticed at the next check both in rw and in ro mode.
+local function is_registered_fiber(self)
+    return self.fiber ~= nil and self.fiber:id() == fiber.id()
+end
+
 local function utubettl_fiber_iteration(self, processed)
     local now       = util.time()
     local task      = nil
@@ -285,7 +295,10 @@ local function utubettl_fiber_iteration(self, processed)
         estimated = processed > 1000 and 0 or estimated
         estimated = estimated > 0 and estimated or 0
         processed = 0
-        self.cond:wait(estimated)
+        -- Do not fall asleep if stop() has been called meanwhile.
+        if is_registered_fiber(self) then
+            self.cond:wait(estimated)
+        end
     end
 
     return processed
@@ -297,25 +310,30 @@ local function utubettl_fiber(self)
     log.info("Started queue utubettl fiber")
     local processed = 0
 
-    while true do
+    while is_registered_fiber(self) do
         if box.info.ro == false then
             local stat, err = pcall(utubettl_fiber_iteration, self, processed)
 
             if not stat and not (err.code == box.error.READONLY) then
+                -- Do not exit: a dead fiber can not be restarted until the
+                -- next ro -> rw switch (gh-263). The error may well be
+                -- transient (a failed WAL write, a transaction conflict, a
+                -- user on_task_change callback), so back off and retry.
                 log.error("error catched: %s", tostring(err))
-                log.error("exiting fiber '%s'", fiber.name())
-                return 1
+                log.error("ttl fiber '%s' retries in %d sec", fiber.name(),
+                    ERROR_RETRY_DELAY)
+                fiber.sleep(ERROR_RETRY_DELAY)
             elseif stat then
                 processed = err
             end
         else
-            -- When switching the master to the replica, the fiber will be stopped.
-            if self.sync_chan:get(0.1) ~= nil then
-                log.info("Queue utubettl fiber was stopped")
-                break
-            end
+            -- When switching the master to the replica, the fiber will be
+            -- stopped by the queue state machine.
+            self.cond:wait(0.1)
         end
     end
+
+    log.info("Queue utubettl fiber was stopped")
 end
 
 -- start tube on space
@@ -383,8 +401,7 @@ function tube.new(space, on_task_change, opts)
     }, { __index = method })
 
     self.cond  = qc.waiter()
-    self.fiber = fiber.create(utubettl_fiber, self)
-    self.sync_chan = fiber.channel(1)
+    self:start()
 
     return self
 end
@@ -749,16 +766,25 @@ function method.start(self)
     if self.fiber then
         return
     end
-    self.fiber = fiber.create(utubettl_fiber, self)
+    -- fiber.new(): the fiber must not run before it is registered in
+    -- self.fiber, otherwise it would see itself as already stopped.
+    self.fiber = fiber.new(utubettl_fiber, self)
+    self.fiber:name('utubettl')
+    self.fiber:set_joinable(true)
 end
 
 function method.stop(self)
-    if not self.fiber then
+    local ttl_fiber = self.fiber
+    if ttl_fiber == nil then
         return
     end
-    self.cond:signal(self.fiber:id())
-    self.sync_chan:put(true)
+    -- Deregister the fiber, wake it up and wait until it exits, so that
+    -- the caller (e.g. drop()) can not race with an in-flight iteration.
     self.fiber = nil
+    self.cond:signal(ttl_fiber:id())
+    if ttl_fiber:id() ~= fiber.id() then
+        ttl_fiber:join()
+    end
 end
 
 function method.drop(self)

t/260-ttl-fiber-stop.t

@@ -0,0 +1,67 @@
+#!/usr/bin/env tarantool
+local fiber = require('fiber')
+
+local test = require('tap').test('ttl fiber stop')
+local queue = require('queue')
+local tnt = require('t.tnt')
+tnt.cfg{}
+
+-- gh-262: stop() blocked forever (fifottl) or left the ttl fiber running
+-- (utubettl) while the instance was rw, and drop() leaked the fiber.
+local drivers = {'fifottl', 'utubettl', 'limfifottl'}
+test:plan(#drivers)
+
+local function fibers_named(name)
+    local count = 0
+    for _, info in pairs(fiber.info()) do
+        if info.name == name then
+            count = count + 1
+        end
+    end
+    return count
+end
+
+-- Runs fn in a separate fiber and waits for it up to timeout seconds.
+local function finishes(fn, timeout)
+    local done = false
+    fiber.create(function()
+        fn()
+        done = true
+    end)
+    local deadline = fiber.time() + timeout
+    while not done and fiber.time() < deadline do
+        fiber.sleep(0.01)
+    end
+    return done
+end
+
+for _, driver in ipairs(drivers) do
+    test:test(driver, function(test)
+        test:plan(9)
+        local fiber_name = driver == 'limfifottl' and 'fifottl' or driver
+        local before = fibers_named(fiber_name)
+        local tube = queue.create_tube(driver .. '_stop', driver)
+        test:is(fibers_named(fiber_name), before + 1, 'ttl fiber is started')
+
+        local ttl_fiber = tube.raw.fiber
+        test:ok(finishes(function() tube.raw:stop() end, 1),
+            'stop() returns in rw mode')
+        test:is(ttl_fiber:status(), 'dead', 'ttl fiber is terminated by stop()')
+        test:isnil(tube.raw.fiber, 'ttl fiber is unregistered by stop()')
+        test:is(fibers_named(fiber_name), before, 'no ttl fibers left after stop()')
+
+        tube.raw:start()
+        test:is(fibers_named(fiber_name), before + 1, 'start() creates one ttl fiber')
+        tube:put('data', {ttl = 0.1, utube = 'utube'})
+        fiber.sleep(0.3)
+        test:is(tube.raw.space:len(), 0, 'ttl is processed after start()')
+
+        ttl_fiber = tube.raw.fiber
+        test:ok(finishes(function() tube:drop() end, 1), 'drop() returns in rw mode')
+        test:is(fibers_named(fiber_name), before, 'no ttl fibers leaked by drop()')
+    end)
+end
+
+tnt.finish()
+os.exit(test:check() and 0 or 1)
+-- vim: set ft=lua :

t/270-ttl-fiber-errors.t

@@ -0,0 +1,54 @@
+#!/usr/bin/env tarantool
+local fiber = require('fiber')
+
+local test = require('tap').test('ttl fiber errors')
+local queue = require('queue')
+local queue_state = require('queue.abstract.queue_state')
+local tnt = require('t.tnt')
+tnt.cfg{}
+
+-- gh-263: an error inside a ttl iteration killed the ttl fiber for good,
+-- and the next ro switch got the queue stuck in the ENDING state.
+local drivers = {'fifottl', 'utubettl'}
+test:plan(#drivers)
+
+for _, driver in ipairs(drivers) do
+    test:test(driver, function(test)
+        test:plan(7)
+        local fired = false
+        local tube = queue.create_tube(driver .. '_errors', driver, {
+            on_task_change = function(task, stat)
+                if stat == 'ttl' and not fired then
+                    fired = true
+                    error('user callback failed once')
+                end
+            end,
+        })
+        local ttl_fiber = tube.raw.fiber
+
+        tube:put('data', {ttl = 0.1, utube = 'utube'})
+        fiber.sleep(0.3)
+        test:ok(fired, 'the callback has raised an error on ttl')
+        test:isnt(ttl_fiber:status(), 'dead', 'ttl fiber survives the error')
+
+        local task = tube:put('data', {ttl = 0.1, utube = 'utube'})
+        fiber.sleep(1.5) -- ttl plus the retry delay
+        test:isnil(tube.raw.space:get(task[1]),
+            'ttl processing continues after the error')
+
+        box.cfg{read_only = true}
+        test:ok(queue_state.poll(queue_state.states.WAITING, 10),
+            'queue state changed to waiting')
+        box.cfg{read_only = false}
+        test:ok(queue_state.poll(queue_state.states.RUNNING, 10),
+            'queue state changed to running')
+        test:isnt(tube.raw.fiber, nil, 'ttl fiber is registered')
+        test:isnt(tube.raw.fiber:status(), 'dead', 'ttl fiber is running')
+
+        tube:drop()
+    end)
+end
+
+tnt.finish()
+os.exit(test:check() and 0 or 1)
+-- vim: set ft=lua :