20596439889

Committed 30 Dec 2025 12:21PM UTC coverage: 88.326% (-0.08%) from 88.405%

Build # 20596439889

Build Type

Pull #475

github

Committed by

Satbek

Commit Message

safe/fast mode: remove fiber kill

In fast mode every fiber were named as "fast" in order
to kill it on safe enabling.

But there are no yields in crud's storage methods.
And rebalancing trigger is insert/replace in `_bucket` space
which is yield. So when fiber start in fast mode it will
be in fast mode until first `box.space` operation, "write" for
memtx, "read/write" for vinyl.

Therefor there is no need to mark and kill iproto fibers, which handle
fast requests, because there is no situation, when fiber is in fast mode
and rebalancing is in progress.

Pull Request Pull Request #475: safe/fast mode: remove fiber kill

Run Details

79 of 96 new or added lines in 11 files covered. (82.29%)

17 existing lines in 9 files now uncovered.

5266 of 5962 relevant lines covered (88.33%)

13541.92 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

90.99

/crud/delete.lua

local checks = require('checks')
local errors = require('errors')
local yield_checks = require('crud.common.yield_checks')

local call = require('crud.common.call')
local const = require('crud.common.const')
local utils = require('crud.common.utils')
local sharding = require('crud.common.sharding')
local sharding_key_module = require('crud.common.sharding.sharding_key')
local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')
local dev_checks = require('crud.common.dev_checks')
local schema = require('crud.common.schema')
local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')

local DeleteError = errors.new_class('DeleteError', {capture_stack = false})

local delete = {}

local DELETE_FUNC_NAME = 'delete_on_storage'
local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME)

local function delete_on_storage(space_name, key, field_names, opts)
    dev_checks('string', '?', '?table', {
        bucket_id = 'number|cdata',
        sharding_key_hash = '?number',
        sharding_func_hash = '?number',
        skip_sharding_hash_check = '?boolean',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })
    yield_checks.register_fiber()

    opts = opts or {}

    local space = box.space[space_name]
    if space == nil then
        yield_checks.unregister_fiber()
        return nil, DeleteError:new("Space %q doesn't exist", space_name)
    end

    local _, err = sharding.check_sharding_hash(space_name,
                                                opts.sharding_func_hash,
                                                opts.sharding_key_hash,
                                                opts.skip_sharding_hash_check)

    if err ~= nil then
        yield_checks.unregister_fiber()
        return nil, err
    end

    local ref_ok, bucket_ref_err, unref = bucket_ref_unref.bucket_refrw(opts.bucket_id)
    if not ref_ok then
        yield_checks.unregister_fiber()
        return nil, bucket_ref_err
    end

    yield_checks.check_no_yields()
    -- add_space_schema_hash is false because
    -- reloading space format on router can't avoid delete error on storage
    local result =  schema.wrap_func_result(space, space.delete, {
        add_space_schema_hash = false,
        field_names = field_names,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    }, space, key)

    yield_checks.unregister_fiber()

    local unref_ok, err_unref = unref(opts.bucket_id)
    if not unref_ok then
        return nil, err_unref
    end

    return result
end

delete.storage_api = {[DELETE_FUNC_NAME] = delete_on_storage}

-- returns result, err, need_reload
-- need_reload indicates if reloading schema could help
-- see crud.common.schema.wrap_func_reload()
local function call_delete_on_router(vshard_router, space_name, key, opts)
    dev_checks('table', 'string', '?', {
        timeout = '?number',
        bucket_id = '?',
        fields = '?table',
        vshard_router = '?string|table',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)
    if err ~= nil then
        return nil, DeleteError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD
    end
    if space == nil then
        return nil, DeleteError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD
    end

    if box.tuple.is(key) then
        key = key:totable()
    end

    local sharding_key_hash = nil
    local skip_sharding_hash_check = nil

    local sharding_key = key
    if opts.bucket_id == nil then
        if space.index[0] == nil then
            return nil, DeleteError:new("Cannot fetch primary index parts"), const.NEED_SCHEMA_RELOAD
        end
        local primary_index_parts = space.index[0].parts

        local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)
        if err ~= nil then
            return nil, err
        end

        sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,
                                                                space_name,
                                                                sharding_key_data.value,
                                                                primary_index_parts, key)
        if err ~= nil then
            return nil, err
        end

        sharding_key_hash = sharding_key_data.hash
    else
        skip_sharding_hash_check = true
    end

    local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)
    if err ~= nil then
        return nil, err
    end

    -- When the sharding index (bucket_id) is the primary index, bucket_id can be passed as box.NULL.
    sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)

    local delete_on_storage_opts = {
        bucket_id = bucket_id_data.bucket_id,
        sharding_func_hash = bucket_id_data.sharding_func_hash,
        sharding_key_hash = sharding_key_hash,
        skip_sharding_hash_check = skip_sharding_hash_check,
        noreturn = opts.noreturn,
        fetch_latest_metadata = opts.fetch_latest_metadata,
    }

    local call_opts = {
        mode = 'write',
        timeout = opts.timeout,
    }

    local storage_result, err = call.single(vshard_router,
        bucket_id_data.bucket_id, CRUD_DELETE_FUNC_NAME,
        {space_name, key, opts.fields, delete_on_storage_opts},
        call_opts
    )

    if err ~= nil then
        local err_wrapped = DeleteError:new("Failed to call delete on storage-side: %s", err)

        if sharding.result_needs_sharding_reload(err) then
            return nil, err_wrapped, const.NEED_SHARDING_RELOAD
        end

        return nil, err_wrapped
    end

    if storage_result.err ~= nil then
        return nil, DeleteError:new("Failed to delete: %s", storage_result.err)
    end

    if opts.noreturn == true then
        return nil
    end

    local tuple = storage_result.res

    if opts.fetch_latest_metadata == true then
        -- This option is temporary and is related to [1], [2].
        -- [1] https://github.com/tarantool/crud/issues/236
        -- [2] https://github.com/tarantool/crud/issues/361
        space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,
                                                                vshard_router, opts, storage_result.storage_info)
    end

    return utils.format_result({tuple}, space, opts.fields)
end

--- Deletes tuple from the specified space by key
--
-- @function call
--
-- @param string space_name
--  A space name
--
-- @param key
--  Primary key value
--
-- @tparam ?number opts.timeout
--  Function call timeout
--
-- @tparam ?number opts.bucket_id
--  Bucket ID
--  (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
--
-- @tparam ?string|table opts.vshard_router
--  Cartridge vshard group name or vshard router instance.
--  Set this parameter if your space is not a part of the
--  default vshard cluster.
--
-- @tparam ?boolean opts.noreturn
--  Suppress returning successfully processed tuple.
--
-- @return[1] object
-- @treturn[2] nil
-- @treturn[2] table Error description
--
function delete.call(space_name, key, opts)
    checks('string', '?', {
        timeout = '?number',
        bucket_id = '?',
        fields = '?table',
        vshard_router = '?string|table',
        noreturn = '?boolean',
        fetch_latest_metadata = '?boolean',
    })

    opts = opts or {}

    local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)
    if err ~= nil then
        return nil, DeleteError:new(err)
    end

    return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_delete_on_router,
                                   space_name, key, opts)
end

return delete

1	local checks = require('checks')	593✔
2	local errors = require('errors')	593✔
3	local yield_checks = require('crud.common.yield_checks')	593✔
4
5	local call = require('crud.common.call')	593✔
6	local const = require('crud.common.const')	593✔
7	local utils = require('crud.common.utils')	593✔
8	local sharding = require('crud.common.sharding')	593✔
9	local sharding_key_module = require('crud.common.sharding.sharding_key')	593✔
10	local sharding_metadata_module = require('crud.common.sharding.sharding_metadata')	593✔
11	local dev_checks = require('crud.common.dev_checks')	593✔
12	local schema = require('crud.common.schema')	593✔
13	local bucket_ref_unref = require('crud.common.sharding.bucket_ref_unref')	593✔
14
15	local DeleteError = errors.new_class('DeleteError', {capture_stack = false})	593✔
16
17	local delete = {}	593✔
18
19	local DELETE_FUNC_NAME = 'delete_on_storage'	593✔
20	local CRUD_DELETE_FUNC_NAME = utils.get_storage_call(DELETE_FUNC_NAME)	593✔
21
22	local function delete_on_storage(space_name, key, field_names, opts)
23	dev_checks('string', '?', '?table', {	150✔
24	bucket_id = 'number\|cdata',
25	sharding_key_hash = '?number',
26	sharding_func_hash = '?number',
27	skip_sharding_hash_check = '?boolean',
28	noreturn = '?boolean',
29	fetch_latest_metadata = '?boolean',
30	})
31	yield_checks.register_fiber()	150✔
32
33	opts = opts or {}	150✔
34
35	local space = box.space[space_name]	150✔
36	if space == nil then	150✔
NEW 37	yield_checks.unregister_fiber()	×
UNCOV 38	return nil, DeleteError:new("Space %q doesn't exist", space_name)	×
39	end
40
41	local _, err = sharding.check_sharding_hash(space_name,	300✔
42	opts.sharding_func_hash,	150✔
43	opts.sharding_key_hash,	150✔
44	opts.skip_sharding_hash_check)	150✔
45
46	if err ~= nil then	150✔
47	yield_checks.unregister_fiber()	4✔
48	return nil, err	4✔
49	end
50
51	local ref_ok, bucket_ref_err, unref = bucket_ref_unref.bucket_refrw(opts.bucket_id)	146✔
52	if not ref_ok then	146✔
NEW 53	yield_checks.unregister_fiber()	×
UNCOV 54	return nil, bucket_ref_err	×
55	end
56
57	yield_checks.check_no_yields()	146✔
58	-- add_space_schema_hash is false because
59	-- reloading space format on router can't avoid delete error on storage
60	local result = schema.wrap_func_result(space, space.delete, {	292✔
61	add_space_schema_hash = false,
62	field_names = field_names,	146✔
63	noreturn = opts.noreturn,	146✔
64	fetch_latest_metadata = opts.fetch_latest_metadata,	146✔
65	}, space, key)	146✔
66
67	yield_checks.unregister_fiber()	146✔
68
69	local unref_ok, err_unref = unref(opts.bucket_id)	146✔
70	if not unref_ok then	146✔
71	return nil, err_unref	×
72	end
73
74	return result	146✔
75	end
76
77	delete.storage_api = {[DELETE_FUNC_NAME] = delete_on_storage}	593✔
78
79	-- returns result, err, need_reload
80	-- need_reload indicates if reloading schema could help
81	-- see crud.common.schema.wrap_func_reload()
82	local function call_delete_on_router(vshard_router, space_name, key, opts)
83	dev_checks('table', 'string', '?', {	171✔
84	timeout = '?number',
85	bucket_id = '?',
86	fields = '?table',
87	vshard_router = '?string\|table',
88	noreturn = '?boolean',
89	fetch_latest_metadata = '?boolean',
90	})
91
92	local space, err, netbox_schema_version = utils.get_space(space_name, vshard_router, opts.timeout)	171✔
93	if err ~= nil then	171✔
94	return nil, DeleteError:new("An error occurred during the operation: %s", err), const.NEED_SCHEMA_RELOAD	×
95	end
96	if space == nil then	171✔
97	return nil, DeleteError:new("Space %q doesn't exist", space_name), const.NEED_SCHEMA_RELOAD	34✔
98	end
99
100	if box.tuple.is(key) then	308✔
101	key = key:totable()	×
102	end
103
104	local sharding_key_hash = nil	154✔
105	local skip_sharding_hash_check = nil
106
107	local sharding_key = key	154✔
108	if opts.bucket_id == nil then	154✔
109	if space.index[0] == nil then	142✔
110	return nil, DeleteError:new("Cannot fetch primary index parts"), const.NEED_SCHEMA_RELOAD	×
111	end
112	local primary_index_parts = space.index[0].parts	142✔
113
114	local sharding_key_data, err = sharding_metadata_module.fetch_sharding_key_on_router(vshard_router, space_name)	142✔
115	if err ~= nil then	142✔
116	return nil, err	×
117	end
118
119	sharding_key, err = sharding_key_module.extract_from_pk(vshard_router,	284✔
120	space_name,	142✔
121	sharding_key_data.value,	142✔
122	primary_index_parts, key)	284✔
123	if err ~= nil then	142✔
124	return nil, err	4✔
125	end
126
127	sharding_key_hash = sharding_key_data.hash	138✔
128	else
129	skip_sharding_hash_check = true	12✔
130	end
131
132	local bucket_id_data, err = sharding.key_get_bucket_id(vshard_router, space_name, sharding_key, opts.bucket_id)	150✔
133	if err ~= nil then	150✔
134	return nil, err	2✔
135	end
136
137	-- When the sharding index (bucket_id) is the primary index, bucket_id can be passed as box.NULL.
138	sharding.fill_bucket_id_pk(space, key, bucket_id_data.bucket_id)	148✔
139
140	local delete_on_storage_opts = {	148✔
141	bucket_id = bucket_id_data.bucket_id,	148✔
142	sharding_func_hash = bucket_id_data.sharding_func_hash,	148✔
143	sharding_key_hash = sharding_key_hash,	148✔
144	skip_sharding_hash_check = skip_sharding_hash_check,	148✔
145	noreturn = opts.noreturn,	148✔
146	fetch_latest_metadata = opts.fetch_latest_metadata,	148✔
147	}
148
149	local call_opts = {	148✔
150	mode = 'write',
151	timeout = opts.timeout,	148✔
152	}
153
154	local storage_result, err = call.single(vshard_router,	296✔
155	bucket_id_data.bucket_id, CRUD_DELETE_FUNC_NAME,	148✔
156	{space_name, key, opts.fields, delete_on_storage_opts},	148✔
157	call_opts
158	)	148✔
159
160	if err ~= nil then	148✔
161	local err_wrapped = DeleteError:new("Failed to call delete on storage-side: %s", err)	2✔
162
163	if sharding.result_needs_sharding_reload(err) then	4✔
164	return nil, err_wrapped, const.NEED_SHARDING_RELOAD	2✔
165	end
166
167	return nil, err_wrapped	×
168	end
169
170	if storage_result.err ~= nil then	146✔
171	return nil, DeleteError:new("Failed to delete: %s", storage_result.err)	68✔
172	end
173
174	if opts.noreturn == true then	112✔
175	return nil	2✔
176	end
177
178	local tuple = storage_result.res	110✔
179
180	if opts.fetch_latest_metadata == true then	110✔
181	-- This option is temporary and is related to [1], [2].
182	-- [1] https://github.com/tarantool/crud/issues/236
183	-- [2] https://github.com/tarantool/crud/issues/361
184	space = utils.fetch_latest_metadata_when_single_storage(space, space_name, netbox_schema_version,	4✔
185	vshard_router, opts, storage_result.storage_info)	4✔
186	end
187
188	return utils.format_result({tuple}, space, opts.fields)	110✔
189	end
190
191	--- Deletes tuple from the specified space by key
192	--
193	-- @function call
194	--
195	-- @param string space_name
196	-- A space name
197	--
198	-- @param key
199	-- Primary key value
200	--
201	-- @tparam ?number opts.timeout
202	-- Function call timeout
203	--
204	-- @tparam ?number opts.bucket_id
205	-- Bucket ID
206	-- (by default, it's vshard.router.bucket_id_strcrc32 of primary key)
207	--
208	-- @tparam ?string\|table opts.vshard_router
209	-- Cartridge vshard group name or vshard router instance.
210	-- Set this parameter if your space is not a part of the
211	-- default vshard cluster.
212	--
213	-- @tparam ?boolean opts.noreturn
214	-- Suppress returning successfully processed tuple.
215	--
216	-- @return[1] object
217	-- @treturn[2] nil
218	-- @treturn[2] table Error description
219	--
220	function delete.call(space_name, key, opts)	593✔
221	checks('string', '?', {	162✔
222	timeout = '?number',
223	bucket_id = '?',
224	fields = '?table',
225	vshard_router = '?string\|table',
226	noreturn = '?boolean',
227	fetch_latest_metadata = '?boolean',
228	})
229
230	opts = opts or {}	162✔
231
232	local vshard_router, err = utils.get_vshard_router_instance(opts.vshard_router)	162✔
233	if err ~= nil then	162✔
234	return nil, DeleteError:new(err)	8✔
235	end
236
237	return schema.wrap_func_reload(vshard_router, sharding.wrap_method, call_delete_on_router,	158✔
238	space_name, key, opts)	158✔
239	end
240
241	return delete	593✔

tarantool / crud / 20596439889

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous