Skip to content

Commit

Permalink
driver: fix duplicate id error with mvvc on on take
Browse files Browse the repository at this point in the history
Taking the minimum of the index is an implicit transactions, so it is
always done with 'read-confirmed' mvcc isolation level.
It can lead to errors when trying to make parallel 'take' calls with mvcc enabled.
It is hapenning because 'min' for several takes in parallel will be the same since
read confirmed isolation level makes visible all transactions that finished the commit.
To fix it we wrap it with box.begin/commit and set right isolation level.
Current fix does not resolve that bug in situations when we already are in transaction
since it will open nested transactions.

Part of #207
  • Loading branch information
better0fdead committed Aug 10, 2023
1 parent 12b0495 commit 8db176b
Show file tree
Hide file tree
Showing 4 changed files with 71 additions and 7 deletions.
20 changes: 19 additions & 1 deletion queue/abstract/driver/fifo.lua
Original file line number Diff line number Diff line change
Expand Up @@ -97,7 +97,25 @@ end

-- take task
function method.take(self)
local task = self.space.index.status:min{state.READY}
local task
-- Taking the minimum is an implicit transactions, so it is
-- always done with 'read-confirmed' mvcc isolation level.
-- It can lead to errors when trying to make parallel 'take' calls with mvcc enabled.
-- It is hapenning because 'min' for several takes in parallel will be the same since
-- read confirmed isolation level makes visible all transactions that finished the commit.
-- To fix it we wrap it with box.begin/commit and set right isolation level.
-- Current fix does not resolve that bug in situations when we already are in transaction
-- since it will open nested transactions.
-- See https://github.com/tarantool/queue/issues/207
-- See https://www.tarantool.io/ru/doc/latest/concepts/atomic/txn_mode_mvcc/
if box.cfg.memtx_use_mvcc_engine and (not box.is_in_txn()) then
box.begin({txn_isolation = 'read-committed'})
task = self.space.index.status:min{state.READY}
box.commit()
else
task = self.space.index.status:min{state.READY}
end

if task ~= nil and task[2] == state.READY then
task = self.space:update(task[1], { { '=', 2, state.TAKEN } })
self.on_task_change(task, 'take')
Expand Down
19 changes: 15 additions & 4 deletions queue/abstract/driver/fifottl.lua
Original file line number Diff line number Diff line change
Expand Up @@ -285,10 +285,21 @@ end
-- take task
function method.take(self)
local task = nil
for _, t in self.space.index.status:pairs({state.READY}) do
if not is_expired(t) then
task = t
break
if box.cfg.memtx_use_mvcc_engine and (not box.is_in_txn()) then
box.begin({txn_isolation = 'read-committed'})
for _, t in self.space.index.status:pairs({state.READY}) do
if not is_expired(t) then
task = t
break
end
end
box.commit()
else
for _, t in self.space.index.status:pairs({state.READY}) do
if not is_expired(t) then
task = t
break
end
end
end

Expand Down
19 changes: 18 additions & 1 deletion queue/abstract/driver/utube.lua
Original file line number Diff line number Diff line change
Expand Up @@ -109,8 +109,25 @@ function method.take(self)
if task[2] ~= state.READY then
break
end
local taken
-- Taking the minimum is an implicit transactions, so it is
-- always done with 'read-confirmed' mvcc isolation level.
-- It can lead to errors when trying to make parallel 'take' calls with mvcc enabled.
-- It is hapenning because 'min' for several takes in parallel will be the same since
-- read confirmed isolation level makes visible all transactions that finished the commit.
-- To fix it we wrap it with box.begin/commit and set right isolation level.
-- Current fix does not resolve that bug in situations when we already are in transaction
-- since it will open nested transactions.
-- See https://github.com/tarantool/queue/issues/207
-- See https://www.tarantool.io/ru/doc/latest/concepts/atomic/txn_mode_mvcc/
if box.cfg.memtx_use_mvcc_engine and (not box.is_in_txn()) then
box.begin({txn_isolation = 'read-committed'})
taken = self.space.index.utube:min{state.TAKEN, task[3]}
box.commit()
else
taken = self.space.index.utube:min{state.TAKEN, task[3]}
end

local taken = self.space.index.utube:min{state.TAKEN, task[3]}
if taken == nil or taken[2] ~= state.TAKEN then
task = self.space:update(task[1], { { '=', 2, state.TAKEN } })
self.on_task_change(task, 'take')
Expand Down
20 changes: 19 additions & 1 deletion queue/abstract/driver/utubettl.lua
Original file line number Diff line number Diff line change
Expand Up @@ -300,7 +300,25 @@ function method.take(self)
break
elseif not is_expired(t) then
local next_event = util.time() + t[i_ttr]
local taken = self.space.index.utube:min{state.TAKEN, t[i_utube]}
local taken
-- Taking the minimum is an implicit transactions, so it is
-- always done with 'read-confirmed' mvcc isolation level.
-- It can lead to errors when trying to make parallel 'take' calls with mvcc enabled.
-- It is hapenning because 'min' for several takes in parallel will be the same since
-- read confirmed isolation level makes visible all transactions that finished the commit.
-- To fix it we wrap it with box.begin/commit and set right isolation level.
-- Current fix does not resolve that bug in situations when we already are in transaction
-- since it will open nested transactions.
-- See https://github.com/tarantool/queue/issues/207
-- See https://www.tarantool.io/ru/doc/latest/concepts/atomic/txn_mode_mvcc/
if box.cfg.memtx_use_mvcc_engine and (not box.is_in_txn()) then
box.begin({txn_isolation = 'read-committed'})
taken = self.space.index.utube:min{state.TAKEN, t[i_utube]}
box.commit()
else
taken = self.space.index.utube:min{state.TAKEN, t[i_utube]}
end

if taken == nil or taken[i_status] ~= state.TAKEN then
t = self.space:update(t[1], {
{ '=', i_status, state.TAKEN },
Expand Down

0 comments on commit 8db176b

Please sign in to comment.