drivers: fix concurrency with mvcc

By default idx:max() or idx:min() have read confirmed isolation level.
It could lead to a task duplication or double task take when we
already insert or update a task, commited, but it is not yet
confirmed.

See also:

1. tarantool/queue#207
2. tarantool/queue#211

sharded_queue/drivers/fifo.lua

@@ -74,7 +74,7 @@ end
 
 -- put task in space
 function method.put(args)
-    local task = box.atomic(function()
+    local task = utils.atomic(function()
         local idx = get_index(args)
         local task_id = utils.pack_task_id(
             args.bucket_id,
@@ -96,7 +96,7 @@ end
 
 -- take task
 function method.take(args)
-    local task = box.atomic(function()
+    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

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

sharded_queue/utils.lua

@@ -2,6 +2,24 @@ local fiber = require('fiber')
 
 local utils = {}
 
+local function atomic_tail(status, ...)
+    if not status then
+        box.rollback()
+        error((...), 2)
+     end
+     box.commit()
+     return ...
+end
+
+function utils.atomic(fun, ...)
+    --if box.cfg.memtx_use_mvcc_engine then
+    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()))