From 48f4feb83019598c26ac9cf11be10d2d841c8015 Mon Sep 17 00:00:00 2001 From: Maxim Uymin Date: Tue, 22 Sep 2026 15:18:52 +0300 Subject: [PATCH 1/2] drivers: stop the ttl fiber in rw mode as well fifottl and limfifottl used an unbuffered channel to stop the ttl fiber, but the fiber read it only in ro mode. So stop() blocked forever in rw mode and drop() did not stop the fiber at all, leaking one fiber per dropped tube. utubettl used a buffered channel: stop() returned, but the fiber kept running and crashed on the dropped space. Now the fiber works while it is the fiber registered in the tube: stop() deregisters it, wakes it up and joins it, and fifottl got a drop() which stops the fiber before dropping the space. limfifottl forwards start()/stop()/drop() to the parent driver so that the registration lives in a single object. Closes #262 Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 3 ++ queue/abstract/driver/fifottl.lua | 49 ++++++++++++++------ queue/abstract/driver/limfifottl.lua | 20 ++++++++- queue/abstract/driver/utubettl.lua | 44 ++++++++++++------ t/260-ttl-fiber-stop.t | 67 ++++++++++++++++++++++++++++ 5 files changed, 156 insertions(+), 27 deletions(-) create mode 100755 t/260-ttl-fiber-stop.t diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b82cfb1..3b64e871 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Fixed +- `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 f9ba1db3..af3eab55 100644 --- a/queue/abstract/driver/fifottl.lua +++ b/queue/abstract/driver/fifottl.lua @@ -87,6 +87,13 @@ local delayed_state = { state.DELAYED } local ttl_states = { state.READY, state.BURIED } local ttr_state = { state.TAKEN } +-- 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 +152,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,7 +167,7 @@ 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) @@ -169,13 +179,13 @@ local function fifottl_fiber(self) 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 +206,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 +425,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 123a9f86..23d8e1cf 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 3501dbc8..7f4a163c 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/queue/abstract/driver/utubettl.lua @@ -201,6 +201,13 @@ local function begin_if_not_in_txn(self) end end +-- 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 +292,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,7 +307,7 @@ 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) @@ -309,13 +319,13 @@ local function utubettl_fiber(self) 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 +393,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 +758,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 00000000..b984ae64 --- /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 : From 531b062ec7ef400264d3f1b1aeaec0721f5fb16d Mon Sep 17 00:00:00 2001 From: Maxim Uymin Date: Tue, 22 Sep 2026 15:21:24 +0300 Subject: [PATCH 2/2] drivers: do not exit the ttl fiber on an error Any error inside a ttl iteration (a user on_task_change callback raising, a transaction conflict, a failed WAL write) terminated the ttl fiber for good: start() saw the dead fiber in self.fiber and did nothing, so ttl/ttr/delay processing stopped until the next ro -> rw switch, and with the unbuffered channel of fifottl stop() blocked the queue state machine in ENDING forever. Now the fiber logs the error, backs off for a second and retries the iteration. It still exits when it is cancelled or stopped. Closes #263 Part of #238 Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 4 +++ queue/abstract/driver/fifottl.lua | 12 +++++-- queue/abstract/driver/utubettl.lua | 12 +++++-- t/270-ttl-fiber-errors.t | 54 ++++++++++++++++++++++++++++++ 4 files changed, 78 insertions(+), 4 deletions(-) create mode 100755 t/270-ttl-fiber-errors.t diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b64e871..ffc3bf4c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,10 @@ 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). diff --git a/queue/abstract/driver/fifottl.lua b/queue/abstract/driver/fifottl.lua index af3eab55..d54fec46 100644 --- a/queue/abstract/driver/fifottl.lua +++ b/queue/abstract/driver/fifottl.lua @@ -87,6 +87,9 @@ 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. @@ -172,9 +175,14 @@ local function fifottl_fiber(self) 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 diff --git a/queue/abstract/driver/utubettl.lua b/queue/abstract/driver/utubettl.lua index 7f4a163c..424ea362 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/queue/abstract/driver/utubettl.lua @@ -201,6 +201,9 @@ 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. @@ -312,9 +315,14 @@ local function utubettl_fiber(self) 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 diff --git a/t/270-ttl-fiber-errors.t b/t/270-ttl-fiber-errors.t new file mode 100755 index 00000000..28083bf3 --- /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 :