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)