diff --git a/CHANGELOG.md b/CHANGELOG.md index 6b82cfb1..75fc7e97 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` and `utubettl` drivers (and `take()` of + `utubettl` in the ready buffer mode) failed with `attempt to index a nil + value` when an expired task was deleted concurrently between its + selection and `delete()` (gh-264). - 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..cca05b80 100644 --- a/queue/abstract/driver/fifottl.lua +++ b/queue/abstract/driver/fifottl.lua @@ -113,8 +113,12 @@ local function fifottl_fiber_iteration(self, processed) task = self.space.index.watch:min{ task_state } if task ~= nil and task[i_status] == task_state then if now >= task[i_next_event] then - task = self:delete(task[i_id]):transform(2, 1, state.DONE) - self:on_task_change(task, 'ttl') + -- delete() returns the task with the DONE status or nil if + -- the task has been deleted concurrently (gh-264). + task = self:delete(task[i_id]) + if task ~= nil then + self:on_task_change(task, 'ttl') + end estimated = 0 processed = processed + 1 else diff --git a/queue/abstract/driver/utubettl.lua b/queue/abstract/driver/utubettl.lua index 3501dbc8..229b9cd9 100644 --- a/queue/abstract/driver/utubettl.lua +++ b/queue/abstract/driver/utubettl.lua @@ -240,8 +240,12 @@ local function utubettl_fiber_iteration(self, processed) task = self.space.index.watch:min{ task_state } if task ~= nil and task[i_status] == task_state then if now >= task[i_next_event] then - task = self:delete(task[i_id]):transform(2, 1, state.DONE) - self:on_task_change(task, 'ttl') + -- delete() returns the task with the DONE status or nil if + -- the task has been deleted concurrently (gh-264). + task = self:delete(task[i_id]) + if task ~= nil then + self:on_task_change(task, 'ttl') + end estimated = 0 processed = processed + 1 else @@ -518,7 +522,9 @@ local function take_ready(self) take_complete = true end else - task = self:delete(task[i_id]):transform(2, 1, state.DONE) + -- delete() returns the task with the DONE status or nil if + -- the task has been deleted concurrently (gh-264). + task = self:delete(task[i_id]) take_ttl = true end end @@ -528,7 +534,7 @@ local function take_ready(self) if take_complete then self:on_task_change(task, 'take') return task - elseif take_ttl then + elseif take_ttl and task ~= nil then self:on_task_change(task, 'ttl') end end diff --git a/t/280-ttl-delete-nil.t b/t/280-ttl-delete-nil.t new file mode 100755 index 00000000..d6fb1de4 --- /dev/null +++ b/t/280-ttl-delete-nil.t @@ -0,0 +1,52 @@ +#!/usr/bin/env tarantool +local fiber = require('fiber') + +local test = require('tap').test('ttl delete nil') +local queue = require('queue') +local tnt = require('t.tnt') +tnt.cfg{} + +-- gh-264: the ttl branch of the fiber iteration called :transform() on the +-- result of delete(), which is nil when the task has been deleted +-- concurrently between min() and delete(). That killed the ttl fiber. +local drivers = {'fifottl', 'utubettl'} +test:plan(#drivers) + +for _, driver in ipairs(drivers) do + test:test(driver, function(test) + test:plan(4) + local tube = queue.create_tube(driver .. '_nil', driver) + local raw = tube.raw + local ttl_fiber = raw.fiber + + -- Emulate a concurrent delete: the first delete() finds no task and + -- returns nil, exactly as the driver's delete() does in that case. + local calls = 0 + local delete = raw.delete + raw.delete = function(self, id) + calls = calls + 1 + if calls == 1 then + self.space:delete(id) + return nil + end + return delete(self, id) + end + + tube:put('data', {ttl = 0.1, utube = 'utube'}) + fiber.sleep(0.3) + test:is(calls, 1, 'delete() has returned nil to the ttl fiber') + test:isnt(ttl_fiber:status(), 'dead', 'ttl fiber survives a nil from delete()') + + local task = tube:put('data', {ttl = 0.1, utube = 'utube'}) + fiber.sleep(0.3) + test:is(calls, 2, 'delete() is called for the next expired task') + test:isnil(raw.space:get(task[1]), 'the next expired task is deleted') + + raw.delete = nil + tube:drop() + end) +end + +tnt.finish() +os.exit(test:check() and 0 or 1) +-- vim: set ft=lua :