[TNTP-2109] Remove atomic from single calls

Author: ita-sammann

crud/common/call.lua

@@ -137,28 +137,23 @@ local function call_with_retry_and_recovery(vshard_router,
     -- This is a partial copy of error handling from vshard.router.router_call_impl()
     -- It is much simpler mostly because bucket_set() can't be accessed from outside vshard.
     if err.class_name == bucket_ref_unref.BucketRefError.name then
-        local redirect_replicaset
         if is_single_call and #err.bucket_ref_errs == 1 then
             local single_err = err.bucket_ref_errs[1]
             local destination = single_err.vshard_err.destination
             if destination and vshard_router.replicasets[destination] then
-                redirect_replicaset = vshard_router.replicasets[destination]
+                replicaset = vshard_router.replicasets[destination]
             end
         end
 
         for _, bucket_ref_err in pairs(err.bucket_ref_errs) do
             local bucket_id = bucket_ref_err.bucket_id
             local vshard_err = bucket_ref_err.vshard_err
             if vshard_err.name == 'WRONG_BUCKET' or
-                vshard_err.name == 'BUCKET_IS_LOCKED' or
-                vshard_err.name == 'TRANSFER_IS_IN_PROGRESS' then
+               vshard_err.name == 'BUCKET_IS_LOCKED' or
+               vshard_err.name == 'TRANSFER_IS_IN_PROGRESS' then
                 vshard_router:_bucket_reset(bucket_id)
             end
         end
-
-        if redirect_replicaset ~= nil then
-            replicaset = redirect_replicaset
-        end
     elseif err.name == 'MISSING_MASTER' and replicaset.locate_master ~= nil then
         replicaset:locate_master()
     end

crud/common/sharding/bucket_ref_unref.lua

@@ -1,9 +1,9 @@
---- module to call vshard.storage.bucket_ref / vshard.storage.bucket_unref
+--- Module to call vshard.storage.bucket_ref / vshard.storage.bucket_unref
 --- on write requests
 --- there are two modes: safe and fast. on safe mode module
 --- calls vshard.storage.bucket_ref / vshard.storage.bucket_unref
 --- on fast mode it does nothing.
---- default is fast mode.
+--- Default is fast mode.
 
 --- bucket_ref/bucket_unref must be called in one transaction in order to prevent
 --- safe_mode change during execution.
@@ -15,12 +15,12 @@ local rebalance = require('crud.common.rebalance')
 local safe_methods
 local fast_methods
 
-local M = {
+local bucket_ref_unref = {
     BucketRefError = errors.new_class('bucket_ref_error', {capture_stack = false})
 }
 
 local function make_bucket_ref_err(bucket_id, vshard_ref_err)
-    local err = M.BucketRefError:new(
+    local err = bucket_ref_unref.BucketRefError:new(
         "failed bucket_ref: %s, bucket_id: %s",
         vshard_ref_err.name,
         bucket_id
@@ -34,83 +34,74 @@ local function make_bucket_ref_err(bucket_id, vshard_ref_err)
     return err
 end
 
---- on module initialization safe_mode_status func must be set
---- it's rebalance.safe_mode_status
-function M.safe_mode_status()
-    error('safe_mode_status not set')
-end
-
---- Slow bucket_refrw implementation that calls vshard.storage.bucket_refrw.
+--- Safe bucket_refrw implementation that calls vshard.storage.bucket_refrw.
 --- must be called with bucket_unrefrw in transaction
-function M._bucket_refrw(bucket_id)
+function bucket_ref_unref._bucket_refrw(bucket_id)
     local ref_ok, vshard_ref_err = vshard.storage.bucket_refrw(bucket_id)
     if not ref_ok then
-        return false, make_bucket_ref_err(bucket_id, vshard_ref_err)
+        return nil, make_bucket_ref_err(bucket_id, vshard_ref_err)
     end
 
     return true
 end
 
---- Slow bucket_unrefrw implementation that calls vshard.storage.bucket_unrefrw.
+--- Safe bucket_unrefrw implementation that calls vshard.storage.bucket_unrefrw.
 --- must be called with bucket_refrw in transaction
-function M._bucket_unrefrw(bucket_id)
+function bucket_ref_unref._bucket_unrefrw(bucket_id)
     return vshard.storage.bucket_unrefrw(bucket_id)
 end
 
---- Slow bucket_refro implementation that calls vshard.storage.bucket_refro.
-function M._bucket_refro(bucket_id)
+--- Safe bucket_refro implementation that calls vshard.storage.bucket_refro.
+function bucket_ref_unref._bucket_refro(bucket_id)
     local ref_ok, vshard_ref_err = vshard.storage.bucket_refro(bucket_id)
     if not ref_ok then
-        return false, make_bucket_ref_err(bucket_id, vshard_ref_err)
+        return nil, make_bucket_ref_err(bucket_id, vshard_ref_err)
     end
 
     return true
 end
 
---- Slow bucket_unrefro implementation that calls vshard.storage.bucket_unrefro.
+--- Safe bucket_unrefro implementation that calls vshard.storage.bucket_unrefro.
 --- must be called in one transaction with bucket_refrw_many
-function M._bucket_unrefro(bucket_id)
+function bucket_ref_unref._bucket_unrefro(bucket_id)
     return vshard.storage.bucket_unrefro(bucket_id)
 end
 
---- Slow bucket_refrw_many that calls bucket_refrw for every bucket and aggregates errors
+--- Safe bucket_refrw_many that calls bucket_refrw for every bucket and aggregates errors
 --- @param bucket_ids table<number, boolean>
-function M._bucket_refrw_many(bucket_ids)
+function bucket_ref_unref._bucket_refrw_many(bucket_ids)
     local bucket_ref_errs = {}
     local reffed_bucket_ids = {}
     for bucket_id in pairs(bucket_ids) do
         local ref_ok, bucket_refrw_err = safe_methods.bucket_refrw(bucket_id)
         if not ref_ok then
-
             table.insert(bucket_ref_errs, bucket_refrw_err.bucket_ref_errs[1])
-            goto continue
+            break
         end
-
         reffed_bucket_ids[bucket_id] = true
-        ::continue::
     end
 
     if next(bucket_ref_errs) ~= nil then
-        local err = M.BucketRefError:new(M.BucketRefError:new("failed bucket_ref"))
+        local err = bucket_ref_unref.BucketRefError:new(bucket_ref_unref.BucketRefError:new("failed bucket_ref"))
         err.bucket_ref_errs = bucket_ref_errs
-        safe_methods.bucket_unrefrw_many(reffed_bucket_ids)
+        bucket_ref_unref._bucket_unrefrw_many(reffed_bucket_ids)
         return nil, err
     end
 
     return true
 end
 
---- Slow bucket_unrefrw_many that calls vshard.storage.bucket_unrefrw for every bucket.
+--- Safe bucket_unrefrw_many that calls vshard.storage.bucket_unrefrw for every bucket.
 --- must be called in one transaction with bucket_refrw_many
 --- will never happen in called in one transaction with bucket_refrw_many
---- @param bucket_ids table<number, true>
-function M._bucket_unrefrw_many(bucket_ids)
+--- @param bucket_ids table<number, boolean>
+function bucket_ref_unref._bucket_unrefrw_many(bucket_ids)
     local unref_all_ok = true
     local unref_last_err
     for reffed_bucket_id in pairs(bucket_ids) do
         local unref_ok, unref_err = safe_methods.bucket_unrefrw(reffed_bucket_id)
         if not unref_ok then
-            unref_all_ok = nil
+            unref_all_ok = false
             unref_last_err = unref_err
         end
     end
@@ -122,31 +113,31 @@ function M._bucket_unrefrw_many(bucket_ids)
 end
 
 --- _fast implements module logic for fast mode
-function M._fast()
+function bucket_ref_unref._fast()
     return true
 end
 
 safe_methods = {
-    bucket_refrw = M._bucket_refrw,
-    bucket_unrefrw = M._bucket_unrefrw,
-    bucket_refro = M._bucket_refro,
-    bucket_unrefro = M._bucket_unrefro,
-    bucket_refrw_many = M._bucket_refrw_many,
-    bucket_unrefrw_many = M._bucket_unrefrw_many,
+    bucket_refrw = bucket_ref_unref._bucket_refrw,
+    bucket_unrefrw = bucket_ref_unref._bucket_unrefrw,
+    bucket_refro = bucket_ref_unref._bucket_refro,
+    bucket_unrefro = bucket_ref_unref._bucket_unrefro,
+    bucket_refrw_many = bucket_ref_unref._bucket_refrw_many,
+    bucket_unrefrw_many = bucket_ref_unref._bucket_unrefrw_many,
 }
 
 fast_methods = {
-    bucket_refrw = M._fast,
-    bucket_unrefrw = M._fast,
-    bucket_refro = M._fast,
-    bucket_unrefro = M._fast,
-    bucket_refrw_many = M._fast,
-    bucket_unrefrw_many = M._fast,
+    bucket_refrw = bucket_ref_unref._fast,
+    bucket_unrefrw = bucket_ref_unref._fast,
+    bucket_refro = bucket_ref_unref._fast,
+    bucket_unrefro = bucket_ref_unref._fast,
+    bucket_refrw_many = bucket_ref_unref._fast,
+    bucket_unrefrw_many = bucket_ref_unref._fast,
 }
 
 local function set_methods(methods)
     for method_name, func in pairs(methods) do
-        M[method_name] = func
+        bucket_ref_unref[method_name] = func
     end
 end
 
@@ -158,32 +149,18 @@ local function set_fast_mode()
     set_methods(fast_methods)
 end
 
-local hooks_registered = false
-
---- set safe mode func
---- from rebalance.safe_mode_status
-function M.set_safe_mode_status(safe_mode_status)
-    M.safe_mode_status = safe_mode_status
+if rebalance.safe_mode then
+    set_safe_mode()
+else
+    set_fast_mode()
+end
 
-    if safe_mode_status() then
+rebalance.on_safe_mode_toggle(function(is_enabled)
+    if is_enabled then
         set_safe_mode()
     else
         set_fast_mode()
     end
+end)
 
-    if not hooks_registered then
-        rebalance.on_safe_mode_toggle(function(is_enabled)
-            if is_enabled then
-                set_safe_mode()
-            else
-                set_fast_mode()
-            end
-        end)
-
-        hooks_registered = true
-    end
-end
-
-set_fast_mode()
-
-return M
+return bucket_ref_unref

crud/delete.lua

@@ -44,29 +44,25 @@ local function delete_on_storage(space_name, key, field_names, opts)
         return nil, err
     end
 
-    local function make_delete()
-        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(opts.bucket_id)
-        if not ref_ok then
-            return nil, bucket_ref_err
-        end
-
-        -- add_space_schema_hash is false because
-        -- reloading space format on router can't avoid delete error on storage
-        local result = schema.wrap_box_space_func_result(space, 'delete', {key}, {
-            add_space_schema_hash = false,
-            field_names = field_names,
-            noreturn = opts.noreturn,
-            fetch_latest_metadata = opts.fetch_latest_metadata,
-        })
-        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(opts.bucket_id)
-        if not unref_ok then
-            return nil, err_unref
-        end
+    local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(opts.bucket_id)
+    if not ref_ok then
+        return nil, bucket_ref_err
+    end
 
-        return result
+    -- add_space_schema_hash is false because
+    -- reloading space format on router can't avoid delete error on storage
+    local result = schema.wrap_box_space_func_result(space, 'delete', {key}, {
+        add_space_schema_hash = false,
+        field_names = field_names,
+        noreturn = opts.noreturn,
+        fetch_latest_metadata = opts.fetch_latest_metadata,
+    })
+    local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(opts.bucket_id)
+    if not unref_ok then
+        return nil, err_unref
     end
 
-    return box.atomic(make_delete)
+    return result
 end
 
 delete.storage_api = {[DELETE_FUNC_NAME] = delete_on_storage}

crud/insert.lua

@@ -43,33 +43,29 @@ local function insert_on_storage(space_name, tuple, opts)
         return nil, err
     end
 
-    local function make_insert()
-        local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
-        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
+    local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
+    local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
 
-        if not ref_ok then
-            return nil, bucket_ref_err
-        end
+    if not ref_ok then
+        return nil, bucket_ref_err
+    end
 
-        -- add_space_schema_hash is true only in case of insert_object
-        -- the only one case when reloading schema can avoid insert error
-        -- is flattening object on router
-        local result = schema.wrap_box_space_func_result(space, 'insert', {tuple}, {
-            add_space_schema_hash = opts.add_space_schema_hash,
-            field_names = opts.fields,
-            noreturn = opts.noreturn,
-            fetch_latest_metadata = opts.fetch_latest_metadata,
-        })
-
-        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
-        if not unref_ok then
-            return nil, err_unref
-        end
+    -- add_space_schema_hash is true only in case of insert_object
+    -- the only one case when reloading schema can avoid insert error
+    -- is flattening object on router
+    local result = schema.wrap_box_space_func_result(space, 'insert', {tuple}, {
+        add_space_schema_hash = opts.add_space_schema_hash,
+        field_names = opts.fields,
+        noreturn = opts.noreturn,
+        fetch_latest_metadata = opts.fetch_latest_metadata,
+    })
 
-        return result
+    local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
+    if not unref_ok then
+        return nil, err_unref
     end
 
-    return box.atomic(make_insert)
+    return result
 end
 
 insert.storage_api = {[INSERT_FUNC_NAME] = insert_on_storage}

crud/replace.lua

@@ -43,29 +43,25 @@ local function replace_on_storage(space_name, tuple, opts)
         return nil, err
     end
 
-    local function make_replace()
-        local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
-        local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
-        if not ref_ok then
-            return nil, bucket_ref_err
-        end
-        -- add_space_schema_hash is true only in case of replace_object
-        -- the only one case when reloading schema can avoid insert error
-        -- is flattening object on router
-        local result = schema.wrap_box_space_func_result(space, 'replace', {tuple}, {
-            add_space_schema_hash = opts.add_space_schema_hash,
-            field_names = opts.fields,
-            noreturn = opts.noreturn,
-            fetch_latest_metadata = opts.fetch_latest_metadata,
-        })
-        local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
-        if not unref_ok then
-            return nil, err_unref
-        end
-        return result
+    local bucket_id = tuple[utils.get_bucket_id_fieldno(space)]
+    local ref_ok, bucket_ref_err = bucket_ref_unref.bucket_refrw(bucket_id)
+    if not ref_ok then
+        return nil, bucket_ref_err
     end
-
-    return box.atomic(make_replace)
+    -- add_space_schema_hash is true only in case of replace_object
+    -- the only one case when reloading schema can avoid insert error
+    -- is flattening object on router
+    local result = schema.wrap_box_space_func_result(space, 'replace', {tuple}, {
+        add_space_schema_hash = opts.add_space_schema_hash,
+        field_names = opts.fields,
+        noreturn = opts.noreturn,
+        fetch_latest_metadata = opts.fetch_latest_metadata,
+    })
+    local unref_ok, err_unref = bucket_ref_unref.bucket_unrefrw(bucket_id)
+    if not unref_ok then
+        return nil, err_unref
+    end
+    return result
 end
 
 replace.storage_api = {[REPLACE_FUNC_NAME] = replace_on_storage}