From 9f3c9e58aa0528b637b62dd29d8a9a25061b46ff Mon Sep 17 00:00:00 2001 From: Andre Branchizio Date: Thu, 9 Jul 2026 14:15:47 -0600 Subject: [PATCH] storage: add bucket GC pacing options VShard limits concurrent bucket transfers, but bucket garbage collection keeps deleting data until all garbage buckets are empty. Transaction chunking bounds a single TX-thread burst, but it does not let operators reserve TX capacity for foreground requests. Add bucket_gc_batch_size to control the number of tuples deleted in one GC transaction and bucket_gc_batch_delay to pause between non-empty GC delete transactions. Keep the defaults at 1000 tuples and zero delay to preserve the existing behavior. The settings are dynamic and apply to both the automatic collector and vshard.storage.bucket_delete_garbage(). Delays happen after commit and bucket state is checked again after each pacing yield. @TarantoolBot document Title: Document bucket GC pacing options Two new root configuration options control bucket garbage collection: * `bucket_gc_batch_size` is the maximum number of tuples deleted in one bucket GC transaction. It defaults to 1000. Smaller values reduce the longest uninterrupted TX-thread burst, but create more transactions. * `bucket_gc_batch_delay` is the delay in seconds between non-empty bucket GC delete transactions. It defaults to 0, which disables the delay. A positive value reduces sustained GC pressure on the TX thread, but makes removal of transferred bucket data take longer. Both options can be changed dynamically and affect automatic bucket GC as well as `vshard.storage.bucket_delete_garbage()`. --- README.md | 4 + .../garbage_collector_test.lua | 171 ++++++++++++++++++ test/unit-luatest/config_test.lua | 40 ++++ vshard/cfg.lua | 8 + vshard/consts.lua | 4 + vshard/storage/init.lua | 44 ++++- 6 files changed, 263 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 3ed3629b..cd1d81c7 100644 --- a/README.md +++ b/README.md @@ -92,6 +92,8 @@ local cfg = { bucket_count = 10000, rebalancer_disbalance_threshold = 10, rebalancer_max_receiving = 100, + bucket_gc_batch_size = 1000, + bucket_gc_batch_delay = 0, sharding = { ['cbf06940-0790-498b-948d-042b62cf3d29'] = { -- replicaset #1 replicas = { @@ -128,6 +130,8 @@ local cfg = { * `bucket_count` total bucket count in a cluster. **It can not be changed after bootstrap!**; * `rebalancer_disbalance_threshold` maximal bucket disbalance percents. Disbalance for each replicaset is calculated by formula: `|etalon_bucket_count - real_bucket_count| / etalon_bucket_count * 100`. * `rebalancer_max_receiving` maximal bucket count that can be received in parallel by single replicaset. This count must be limited, because else, when a new replicaset is added to a cluster, the rebalancer would send to it very big amount of buckets from existing replicasets - it produces heavy load on a new replicaset to apply all these buckets. +* `bucket_gc_batch_size` maximal number of tuples that bucket garbage collection deletes in one transaction. The default is 1000. The `BUCKET_EVENT.GC` trigger runs once per transaction, so smaller values increase its invocation rate. +* `bucket_gc_batch_delay` delay in seconds between bucket garbage collection transactions that delete data. The default is 0, which disables the delay between batches. A positive value limits garbage collection pressure on the transaction thread, but makes removal of transferred bucket data take longer. The setting affects both automatic garbage collection and `vshard.storage.bucket_delete_garbage()`. Example of usage `rebalancer_max_receiving`:
Suppose it to be equal to 100, total bucket count is 1000 and there are diff --git a/test/storage-luatest/garbage_collector_test.lua b/test/storage-luatest/garbage_collector_test.lua index c69dd593..1fecf210 100644 --- a/test/storage-luatest/garbage_collector_test.lua +++ b/test/storage-luatest/garbage_collector_test.lua @@ -160,6 +160,177 @@ test_group.test_basic = function(g) end) end +-- +-- Bucket GC can limit transaction size and pause between delete batches. +-- +test_group.test_pacing = function(g) + local batch_size = 10 + local batch_delay = 0.5 + local cfg = table.deepcopy(global_cfg) + cfg.bucket_gc_batch_size = batch_size + cfg.bucket_gc_batch_delay = batch_delay + vtest.cluster_cfg(g, cfg) + + g.replica_1_a:exec(function(batch_size, batch_delay) + local s = box.space.test + local bid = _G.get_first_bucket() + ilt.assert_not_equals(bid, nil, 'get any bucket') + + local tuple_count = batch_size * 2 + box.begin() + for i = 1, tuple_count do + s:replace{i, bid} + end + box.commit() + + local gc = ifiber.new(function() + ivshard.storage.bucket_delete_garbage(bid, {force = true}) + end) + gc:set_joinable(true) + + local bucket_index = s.index.bucket_id + local expected_count = tuple_count - batch_size + local deadline = ifiber.clock() + batch_delay * 3 + 1 + while bucket_index:count({bid}) == tuple_count and + ifiber.clock() < deadline do + ifiber.sleep(0.01) + end + ilt.assert_equals(bucket_index:count({bid}), expected_count, + 'first GC batch is committed') + ifiber.sleep(batch_delay / 2) + ilt.assert_equals(bucket_index:count({bid}), expected_count, + 'GC waits before the next batch') + + local ok, err = gc:join() + ilt.assert(ok, err) + ilt.assert_equals(bucket_index:count({bid}), 0, + 'all bucket data is deleted') + end, {batch_size, batch_delay}) + + vtest.cluster_wait_vclock_all(g) + vtest.cluster_cfg(g, global_cfg) +end + +-- +-- Automatic bucket GC uses one pacing state across buckets. +-- +test_group.test_pacing_automatic_gc = function(g) + local batch_size = 10 + local batch_delay = 0.5 + local cfg = table.deepcopy(global_cfg) + cfg.bucket_gc_batch_size = batch_size + cfg.bucket_gc_batch_delay = batch_delay + vtest.cluster_cfg(g, cfg) + g.replica_1_b:exec(bucket_set_protection, {false}) + + g.replica_1_a:exec(function(batch_size, batch_delay) + _G.bucket_gc_pause() + local s = box.space.test + local _bucket = box.space._bucket + local active = _bucket.index.status:select({ivconst.BUCKET.ACTIVE}, + {limit = 2}) + ilt.assert_equals(#active, 2, 'get two active buckets') + local bids = {active[1].id, active[2].id} + local tuple_counts = {batch_size + 1, 1} + local id = 100000 + box.begin() + for i, bid in ipairs(bids) do + for _ = 1, tuple_counts[i] do + s:replace{id, bid} + id = id + 1 + end + end + box.commit() + + ivshard.storage.internal.is_bucket_protected = false + for _, bid in ipairs(bids) do + _bucket:replace{bid, ivconst.BUCKET.GARBAGE} + end + ivshard.storage.internal.is_bucket_protected = true + _G.bucket_gc_continue() + + local bucket_index = s.index.bucket_id + local function tuple_count() + return bucket_index:count({bids[1]}) + + bucket_index:count({bids[2]}) + end + local total_count = tuple_counts[1] + tuple_counts[2] + local expected_count = total_count - batch_size + local deadline = ifiber.clock() + batch_delay * 3 + 1 + while tuple_count() == total_count and ifiber.clock() < deadline do + ifiber.sleep(0.01) + end + ilt.assert_equals(tuple_count(), expected_count, + 'first automatic GC batch is committed') + ifiber.sleep(batch_delay / 2) + ilt.assert_equals(tuple_count(), expected_count, + 'automatic GC waits before the next batch') + + _G.bucket_gc_wait() + ilt.assert_equals(tuple_count(), 0, 'all bucket data is deleted') + for _, bid in ipairs(bids) do + ilt.assert_equals(_bucket:get({bid}), nil, 'bucket is deleted') + ivshard.storage.bucket_force_create(bid) + end + end, {batch_size, batch_delay}) + + vtest.cluster_wait_vclock_all(g) + g.replica_1_b:exec(bucket_set_protection, {true}) + vtest.cluster_cfg(g, global_cfg) +end + +-- +-- Bucket state is checked again after the pacing delay. +-- +test_group.test_pacing_checks_bucket_state = function(g) + local batch_size = 10 + local batch_delay = 0.5 + local cfg = table.deepcopy(global_cfg) + cfg.bucket_gc_batch_size = batch_size + cfg.bucket_gc_batch_delay = batch_delay + vtest.cluster_cfg(g, cfg) + + g.replica_1_a:exec(function(batch_size, batch_delay) + local s = box.space.test + local bid = _G.get_first_bucket() + ilt.assert_not_equals(bid, nil, 'get any bucket') + local tuple_count = batch_size + 1 + box.begin() + for i = 1, tuple_count do + s:replace{i, bid} + end + box.commit() + + local gc = ifiber.new(function() + return pcall(ivshard.storage.bucket_delete_garbage, bid, + {force = true}) + end) + gc:set_joinable(true) + + local bucket_index = s.index.bucket_id + local deadline = ifiber.clock() + batch_delay * 3 + 1 + while bucket_index:count({bid}) == tuple_count and + ifiber.clock() < deadline do + ifiber.sleep(0.01) + end + ilt.assert_equals(bucket_index:count({bid}), 1, + 'first GC batch is committed') + ilt.assert(ivshard.storage.bucket_pin(bid)) + + local joined, ok, err = gc:join() + ilt.assert(joined, err) + ilt.assert_not(ok, 'GC must fail after the bucket changes') + ilt.assert_str_contains(err.message, 'bucket status is changed') + ilt.assert_equals(bucket_index:count({bid}), 1, + 'GC does not delete after the bucket changes') + ilt.assert(ivshard.storage.bucket_unpin(bid)) + s:truncate() + end, {batch_size, batch_delay}) + + vtest.cluster_wait_vclock_all(g) + vtest.cluster_cfg(g, global_cfg) +end + test_group.test_yield_before_send_commit = function(g) t.run_only_if(global_cfg.memtx_use_mvcc_engine) g.replica_1_b:exec(bucket_set_protection, {false}) diff --git a/test/unit-luatest/config_test.lua b/test/unit-luatest/config_test.lua index 5c280f7c..d723580f 100644 --- a/test/unit-luatest/config_test.lua +++ b/test/unit-luatest/config_test.lua @@ -1,5 +1,6 @@ local t = require('luatest') local vcfg = require('vshard.cfg') +local vconsts = require('vshard.consts') local vutil = require('vshard.util') local luuid = require('uuid') @@ -392,3 +393,42 @@ g.test_identification_mode_uuid_as_key = function() 'uuid option can be specified only when ' .. 'identification_mode = "name_as_key"', vcfg.check, config) end + +g.test_bucket_gc_pacing = function() + local config = { + sharding = { + storage_1_uuid = { + replicas = {}, + }, + }, + } + local checked = vcfg.check(config) + t.assert_equals(checked.bucket_gc_batch_size, + vconsts.DEFAULT_BUCKET_GC_BATCH_SIZE) + t.assert_equals(checked.bucket_gc_batch_delay, + vconsts.DEFAULT_BUCKET_GC_BATCH_DELAY) + + config.bucket_gc_batch_size = 100 + config.bucket_gc_batch_delay = 0.01 + checked = vcfg.check(config) + t.assert_equals(checked.bucket_gc_batch_size, 100) + t.assert_equals(checked.bucket_gc_batch_delay, 0.01) + + config.bucket_gc_batch_size = 0 + t.assert_error_msg_content_equals( + 'Bucket GC batch size must be positive integer', vcfg.check, config) + config.bucket_gc_batch_size = 1.5 + t.assert_error_msg_content_equals( + 'Bucket GC batch size must be positive integer', vcfg.check, config) + config.bucket_gc_batch_size = '100' + t.assert_error_msg_content_equals( + 'Bucket GC batch size must be positive integer', vcfg.check, config) + config.bucket_gc_batch_size = 100 + + config.bucket_gc_batch_delay = -0.01 + t.assert_error_msg_content_equals( + 'Bucket GC batch delay must be non-negative number', vcfg.check, config) + config.bucket_gc_batch_delay = '0.01' + t.assert_error_msg_content_equals( + 'Bucket GC batch delay must be non-negative number', vcfg.check, config) +end diff --git a/vshard/cfg.lua b/vshard/cfg.lua index ab909004..36004425 100644 --- a/vshard/cfg.lua +++ b/vshard/cfg.lua @@ -427,6 +427,14 @@ local cfg_template = { default = 'auto', enum = {'auto', 'manual', 'off'}, }, + bucket_gc_batch_size = { + type = 'positive integer', name = 'Bucket GC batch size', + is_optional = true, default = consts.DEFAULT_BUCKET_GC_BATCH_SIZE, + }, + bucket_gc_batch_delay = { + type = 'non-negative number', name = 'Bucket GC batch delay', + is_optional = true, default = consts.DEFAULT_BUCKET_GC_BATCH_DELAY, + }, collect_bucket_garbage_interval = { name = 'Garbage bucket collect interval', is_deprecated = true, reason = 'Has no effect anymore' diff --git a/vshard/consts.lua b/vshard/consts.lua index 80c49f5b..f9bc1fa3 100644 --- a/vshard/consts.lua +++ b/vshard/consts.lua @@ -36,6 +36,10 @@ return { DEFAULT_BUCKET_COUNT = 3000; BUCKET_SENT_GARBAGE_DELAY = 60; BUCKET_CHUNK_SIZE = 1000; + -- Tuple count per GC transaction. Independent from BUCKET_CHUNK_SIZE, + -- which is also used by bucket sending and receiving. + DEFAULT_BUCKET_GC_BATCH_SIZE = 1000; + DEFAULT_BUCKET_GC_BATCH_DELAY = 0; LUA_CHUNK_SIZE = 100000, DEFAULT_REBALANCER_DISBALANCE_THRESHOLD = 1; REBALANCER_IDLE_INTERVAL = 60 * 60; diff --git a/vshard/storage/init.lua b/vshard/storage/init.lua index 1ad2d44b..4d27af20 100644 --- a/vshard/storage/init.lua +++ b/vshard/storage/init.lua @@ -204,6 +204,10 @@ if not M then gc_service = nil, -- How many times the GC fiber performed the garbage collection. bucket_gc_count = 0, + -- Maximum number of tuples deleted in a single bucket GC transaction. + bucket_gc_batch_size = consts.DEFAULT_BUCKET_GC_BATCH_SIZE, + -- Delay between bucket GC transactions which delete data. + bucket_gc_batch_delay = consts.DEFAULT_BUCKET_GC_BATCH_DELAY, -------------------- Bucket recovery --------------------- recovery_fiber = nil, @@ -2350,7 +2354,7 @@ end -- @param bucket_id Id of the bucket to cleanup. -- @param status Bucket status for guard checks. -- -local function gc_bucket_in_space_xc(space, bucket_id, status) +local function gc_bucket_in_space_xc(space, bucket_id, status, pacing) local bucket_index = space.index[lschema.shard_index] if #bucket_index:select({bucket_id}, {limit = 1}) == 0 then return @@ -2358,7 +2362,22 @@ local function gc_bucket_in_space_xc(space, bucket_id, status) local bucket_generation = M.bucket_generation local pk_parts = space.index[0].parts ::restart:: - local limit = consts.BUCKET_CHUNK_SIZE + if pacing.needs_delay then + local delay = M.bucket_gc_batch_delay + if delay > 0 then + -- Avoid a trailing delay when the preceding batch deleted the last + -- tuple. Keep the old fast path unchanged when pacing is disabled. + if #bucket_index:select({bucket_id}, {limit = 1}) == 0 then + return + end + lfiber.sleep(delay) + end + pacing.needs_delay = false + bucket_generation = + bucket_guard_xc(bucket_generation, bucket_id, status) + end + local limit = M.bucket_gc_batch_size + local initial_limit = limit box.begin() M._on_bucket_event:run(consts.BUCKET_EVENT.GC, bucket_id, {spaces = {space.name}}) @@ -2367,19 +2386,20 @@ local function gc_bucket_in_space_xc(space, bucket_id, status) limit = limit - 1 if limit == 0 then box.commit() - bucket_generation = - bucket_guard_xc(bucket_generation, bucket_id, status) + pacing.needs_delay = true goto restart end end box.commit() + pacing.needs_delay = limit < initial_limit end -- -- Exception safe version of gc_bucket_in_space_xc. -- -local function gc_bucket_in_space(space, bucket_id, status) - local ok, err = pcall(gc_bucket_in_space_xc, space, bucket_id, status) +local function gc_bucket_in_space(space, bucket_id, status, pacing) + local ok, err = pcall(gc_bucket_in_space_xc, space, bucket_id, status, + pacing) if not ok then box.rollback() end @@ -2395,6 +2415,10 @@ local function gc_bucket_drop_xc(status, route_map) local limit = consts.BUCKET_CHUNK_SIZE local _bucket = box.space._bucket local sharded_spaces = lschema.find_sharded_spaces() + -- Share the pacing state so the delay applies between any two non-empty + -- data deletion transactions, including transactions for different spaces + -- or buckets. + local pacing = {needs_delay = false} for _, b in _bucket.index.status:pairs(status) do local id = b.id local ref = M.bucket_refs[id] @@ -2406,7 +2430,7 @@ local function gc_bucket_drop_xc(status, route_map) end end for _, space in pairs(sharded_spaces) do - gc_bucket_in_space_xc(space, id, status) + gc_bucket_in_space_xc(space, id, status, pacing) limit = limit - 1 if limit == 0 then lfiber.sleep(0) @@ -2725,8 +2749,10 @@ local function bucket_delete_garbage(bucket_id, opts) 'ignore this attention') end local sharded_spaces = lschema.find_sharded_spaces() + local pacing = {needs_delay = false} for _, space in pairs(sharded_spaces) do - local status, err = gc_bucket_in_space(space, bucket_id, bucket_status) + local status, err = gc_bucket_in_space(space, bucket_id, bucket_status, + pacing) if not status then error(err) end @@ -4195,6 +4221,8 @@ local function storage_cfg_xc(cfgctx) M.rebalancer_receiving_quota = new_cfg.rebalancer_max_receiving M.rebalancer_worker_count = new_cfg.rebalancer_max_sending M.rebalancer_bucket_send_timeout = new_cfg.rebalancer_bucket_send_timeout + M.bucket_gc_batch_size = new_cfg.bucket_gc_batch_size + M.bucket_gc_batch_delay = new_cfg.bucket_gc_batch_delay M.sync_timeout = new_cfg.sync_timeout M.current_cfg = new_cfg storage_cfg_master_commit(cfgctx)