diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b82cfb..ffc3bf4 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/queue/abstract/driver/fifottl.lua b/queue/abstract/driver/fifottl.lua index f9ba1db..d54fec4 100644 --- a/queue/abstract/driver/fifottl.lua +++ b/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 diff --git a/queue/abstract/driver/limfifottl.lua b/queue/abstract/driver/limfifottl.lua index 123a9f8..23d8e1c 100644 --- a/queue/abstract/driver/limfifottl.lua +++ b/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 diff --git a/queue/abstract/driver/utubettl.lua b/queue/abstract/driver/utubettl.lua index 3501dbc..424ea36 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/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) diff --git a/t/260-ttl-fiber-stop.t b/t/260-ttl-fiber-stop.t new file mode 100755 index 0000000..b984ae6 --- /dev/null +++ b/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 : diff --git a/t/270-ttl-fiber-errors.t b/t/270-ttl-fiber-errors.t new file mode 100755 index 0000000..28083bf --- /dev/null +++ b/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 :