From 48f4feb83019598c26ac9cf11be10d2d841c8015 Mon Sep 17 00:00:00 2001 From: Maxim Uymin Date: Tue, 22 Sep 2026 15:18:52 +0300 Subject: [PATCH] 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 :