fifo: fix data race for put()/take() with vinyl

CHANGELOG.md

@@ -13,6 +13,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ### Fixed
 
+- Data race with fifo driver for put()/take() methods with vinyl
+  engine (#64).
+
 ## 0.1.1 - 2023-09-06
 
 The release fixes the loss of tasks in the `fifottl` driver.

sharded_queue/drivers/fifo.lua

@@ -74,21 +74,39 @@ end
 
 -- put task in space
 function method.put(args)
-    local idx = get_index(args)
-    local task_id = utils.pack_task_id(args.bucket_id, args.bucket_count, idx)
-    local task = get_space(args):insert { task_id, args.bucket_id, state.READY, args.data, idx }
+    local task = utils.atomic(function()
+        local idx = get_index(args)
+        local task_id = utils.pack_task_id(
+            args.bucket_id,
+            args.bucket_count,
+            idx)
+
+        return get_space(args):insert {
+            task_id,
+            args.bucket_id,
+            state.READY,
+            args.data,
+            idx
+        }
+    end)
+
     update_stat(args.tube_name, 'put')
     return normalize_task(task)
 end
 
 -- take task
 function method.take(args)
-    local task = get_space(args).index.status:min { state.READY }
-    if task ~= nil and task[3] == state.READY then
-        task = get_space(args):update(task.task_id, { { '=', 3, state.TAKEN } })
-        update_stat(args.tube_name, 'take')
-        return normalize_task(task)
-    end
+    local task = utils.atomic(function()
+        local task = get_space(args).index.status:min { state.READY }
+        if task == nil or task[3] ~= state.READY then
+            return
+        end
+        return get_space(args):update(task.task_id, { { '=', 3, state.TAKEN } })
+    end)
+    if task == nil then return end
+
+    update_stat(args.tube_name, 'take')
+    return normalize_task(task)
 end
 
 function method.ack(args)

sharded_queue/drivers/fifottl.lua

@@ -233,7 +233,7 @@ function method.put(args)
     local ttr = args.ttr or args.options.ttr or ttl
     local priority = args.priority or args.options.priority or 0
 
-    local task = box.atomic(function()
+    local task = utils.atomic(function()
         local idx = get_index(args.tube_name, args.bucket_id)
 
         local next_event
@@ -301,7 +301,7 @@ end
 
 function method.take(args)
 
-    local task = box.atomic(take, args)
+    local task = utils.atomic(take, args)
     if task == nil then return end
 
     if args.extra and args.extra.log_request then
@@ -314,10 +314,14 @@ function method.take(args)
 end
 
 function method.delete(args)
-    box.begin()
-    local task = box.space[args.tube_name]:get(args.task_id)
-    box.space[args.tube_name]:delete(args.task_id)
-    box.commit()
+    local task = utils.atomic(function()
+        local task = box.space[args.tube_name]:get(args.task_id)
+        if task ~= nil then
+            box.space[args.tube_name]:delete(args.task_id)
+        end
+        return task
+    end)
+
     if task ~= nil then
         task = task:tomap()
         task.status = state.DONE
@@ -355,10 +359,13 @@ function method.touch(args)
 end
 
 function method.ack(args)
-    box.begin()
-    local task = box.space[args.tube_name]:get(args.task_id)
-    box.space[args.tube_name]:delete(args.task_id)
-    box.commit()
+    local task = utils.atomic(function()
+        local task = box.space[args.tube_name]:get(args.task_id)
+        if task ~= nil then
+            box.space[args.tube_name]:delete(args.task_id)
+        end
+        return task
+    end)
     if task ~= nil then
         task = task:tomap()
         task.status = state.DONE
@@ -385,15 +392,16 @@ function method.peek(args)
 end
 
 function method.release(args)
-    box.begin()
-    local task = box.space[args.tube_name]:get(args.task_id)
-    if task ~= nil then
-        task = box.space[args.tube_name]:update(args.task_id, {
-            {'=', index.status, state.READY},
-            {'=', index.next_event, task[index.created] + task[index.ttl]},
-        })
-    end
-    box.commit()
+    local task = utils.atomic(function()
+        local task = box.space[args.tube_name]:get(args.task_id)
+        if task ~= nil then
+            task = box.space[args.tube_name]:update(args.task_id, {
+                {'=', index.status, state.READY},
+                {'=', index.next_event, task[index.created] + task[index.ttl]},
+            })
+        end
+        return task
+    end)
 
     if args.extra and args.extra.log_request then
         log_operation("release", task)
@@ -408,15 +416,16 @@ function method.bury(args)
     update_stat(args.tube_name, 'bury')
     wc_signal(args.tube_name)
 
-    box.begin()
-    local task = box.space[args.tube_name]:get(args.task_id)
-    if task ~= nil then
-        task = box.space[args.tube_name]:update(args.task_id, {
-            {'=', index.status, state.BURIED},
-            {'=', index.next_event, task[index.created] + task[index.ttl]},
-        })
-    end
-    box.commit()
+    local task = utils.atomic(function()
+        local task = box.space[args.tube_name]:get(args.task_id)
+        if task ~= nil then
+            task = box.space[args.tube_name]:update(args.task_id, {
+                {'=', index.status, state.BURIED},
+                {'=', index.next_event, task[index.created] + task[index.ttl]},
+            })
+        end
+        return task
+    end)
 
     if args.extra and args.extra.log_request then
         log_operation("bury", task)

sharded_queue/utils.lua

@@ -2,6 +2,32 @@ local fiber = require('fiber')
 
 local utils = {}
 
+local function atomic_tail(status, ...)
+    if not status then
+        box.rollback()
+        error((...), 2)
+     end
+     box.commit()
+     return ...
+end
+
+-- box.atomic(opts, fun, args) does not supported for all Tarantool's versions,
+-- so we an analog.
+function utils.atomic(fun, ...)
+    if box.cfg.memtx_use_mvcc_engine then
+        -- max() + insert() or min() + update() do not work as expected with
+        -- best-effort visibility: for write transactions it chooses
+        -- read-committed, for read transactions it chooses read-confirmed.
+        --
+        -- So max()/min() could return the same tuple even if a concurrent
+        -- insert()/update() has been committed, but has not confirmed yet.
+        box.begin({txn_isolation = 'read-committed'})
+    else
+        box.begin()
+    end
+    return atomic_tail(pcall(fun, ...))
+end
+
 function utils.array_shuffle(array)
     if not array then return nil end
     math.randomseed(tonumber(0ULL + fiber.time64()))