storage: Add bucket GC pacing options

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`:<br>
 Suppose it to be equal to 100, total bucket count is 1000 and there are

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})

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

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'

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;

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,15 +2354,30 @@ 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
     end
     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)