gc: make it compatible with memtx mvcc

Bucket garbage collector wakes up on _bucket generation change. It
used to happen on each update in _bucket space. When GC fiber
wakes up, it deletes the garbage and goes to sleep until next
generation change.

It didn't work with mvcc because the generation was bumped during
replace before commit, GC fiber woke up, saw no changes, and went
to sleep until next bump. Then _bucket update would commit and
there would be a SENT/GARBAGE bucket visible, but GC fiber
wouldn't even try to find it because the generation didn't change
since last wakeup.

The patch makes so the generation is bumped on commit to _bucket,
not on replace.

Closes #314

Author: Gerold103

test/instances/storage.lua

@@ -44,6 +44,11 @@ local function get_uuid()
     return instance_uuid
 end
 
+local function get_first_bucket()
+    local res = box.space._bucket.index.status:min(vconst.BUCKET.ACTIVE)
+    return res ~= nil and res.id or nil
+end
+
 local function session_set(key, value)
     box.session.storage[key] = value
     return true
@@ -69,6 +74,7 @@ end
 _G.box_error = box_error
 _G.echo = echo
 _G.get_uuid = get_uuid
+_G.get_first_bucket = get_first_bucket
 _G.session_set = session_set
 _G.session_get = session_get
 _G.wait_bucket_gc = wait_bucket_gc

test/luatest_helpers/vtest.lua

@@ -2,7 +2,6 @@ local t = require('luatest')
 local helpers = require('test.luatest_helpers')
 local cluster = require('test.luatest_helpers.cluster')
 local fiber = require('fiber')
-local vconst = require('vshard.consts')
 local vrepset = require('vshard.replicaset')
 
 local uuid_idx = 1
@@ -115,6 +114,7 @@ local function storage_new(g, cfg)
             box_cfg.listen = helpers.instance_uri(replica.name)
             -- Need to specify read-only explicitly to know how is master.
             box_cfg.read_only = not replica.master
+            box_cfg.memtx_use_mvcc_engine = cfg.memtx_use_mvcc_engine
             local server = g.cluster:build_server({
                 alias = name,
                 box_cfg = box_cfg,
@@ -343,10 +343,9 @@ end
 -- where the buckets are located by hardcoded numbers and uuids.
 --
 local function storage_first_bucket(storage)
-    return storage:exec(function(status)
-        local res = box.space._bucket.index.status:min(status)
-        return res ~= nil and res.id or nil
-    end, {vconst.BUCKET.ACTIVE})
+    return storage:exec(function()
+        return _G.get_first_bucket()
+    end)
 end
 
 --

test/storage-luatest/garbage_collector_test.lua

@@ -1,10 +1,21 @@
 local t = require('luatest')
 local vtest = require('test.luatest_helpers.vtest')
+local vutil = require('vshard.util')
 local wait_timeout = 120
 
-local test_group = t.group('bucket_gc', {
-    {engine = 'memtx'}, {engine = 'vinyl'}
-})
+local group_config = {{engine = 'memtx'}, {engine = 'vinyl'}}
+
+if vutil.feature.memtx_mvcc then
+    table.insert(group_config, {
+        engine = 'memtx', memtx_use_mvcc_engine = true
+    })
+    table.insert(group_config, {
+        engine = 'vinyl', memtx_use_mvcc_engine = true
+    })
+end
+
+local test_group = t.group('bucket_gc', group_config)
+
 local cfg_template = {
     sharding = {
         {
@@ -18,9 +29,12 @@ local cfg_template = {
     },
     bucket_count = 20
 }
-local cluster_cfg = vtest.config_new(cfg_template)
+local cluster_cfg
 
 test_group.before_all(function(g)
+    cfg_template.memtx_use_mvcc_engine = g.params.memtx_use_mvcc_engine
+    cluster_cfg = vtest.config_new(cfg_template)
+
     vtest.storage_new(g, cluster_cfg)
     vtest.storage_bootstrap(g, cluster_cfg)
     vtest.storage_exec_each_master(g, function(engine)
@@ -137,3 +151,51 @@ test_group.test_basic = function(g)
             ivshard.storage.internal.collect_bucket_garbage_fiber, nil)
     end)
 end
+
+test_group.test_yield_before_send_commit = function(g)
+    t.run_only_if(cluster_cfg.memtx_use_mvcc_engine)
+
+    g.replica_1_a:exec(function(timeout)
+        local s = box.space.test
+        local bucket_space = box.space._bucket
+        local bid = _G.get_first_bucket()
+        ilt.assert_not_equals(bid, nil, 'get any bucket')
+
+        -- Fill a bucket with some data.
+        local tuple_count = 10
+        box.begin()
+        for i = 1, tuple_count do
+            s:replace{i, bid}
+        end
+        box.commit()
+
+        -- Start its "sending" but yield before commit.
+        local is_send_blocked = true
+        local f_send = ifiber.new(function()
+            box.begin()
+            bucket_space:update({bid}, {{'=', 2, ivconst.BUCKET.SENT}})
+            while is_send_blocked do
+                ifiber.sleep(0.01)
+            end
+            box.commit()
+        end)
+        f_send:set_joinable(true)
+        ifiber.sleep(0.01)
+        ilt.assert_equals(bucket_space:get{bid}.status, ivconst.BUCKET.ACTIVE,
+                        'sending is not visible yet')
+        is_send_blocked = false
+        ilt.assert(f_send:join(), 'long send succeeded')
+
+        -- Bucket GC should react on commit. Not wakeup on replace, notice no
+        -- changes, and go to sleep.
+        _G.wait_bucket_gc(timeout)
+        ilt.assert_equals(s:count(), 0, 'no garbage data')
+
+        -- Restore the bucket for next tests.
+        ivshard.storage.bucket_force_create(bid)
+        s:truncate()
+    end, {wait_timeout})
+
+    -- Ensure the replica received all that and didn't break.
+    g.replica_1_b:wait_vclock_of(g.replica_1_a)
+end

vshard/storage/init.lua

@@ -291,6 +291,22 @@ local function bucket_generation_increment()
     M.bucket_generation_cond:broadcast()
 end
 
+local function bucket_on_replace_f()
+    local f = bucket_generation_increment
+    local f_del = f
+    -- One transaction can affect many buckets. Do not create a trigger for each
+    -- bucket. Instead create it only first time. For that the trigger replaces
+    -- itself when installed not first time.
+    -- Although in case of hot reload during changing _bucket the function
+    -- pointer will change and more than one trigger could be installed.
+    -- Shouldn't matter anyway.
+    if not pcall(box.on_commit, f, f) then
+        box.on_commit(f)
+        f_del = nil
+    end
+    box.on_rollback(f, f_del)
+end
+
 local function bucket_generation_wait(timeout)
     return fiber_cond_wait(M.bucket_generation_cond, timeout)
 end
@@ -430,8 +446,8 @@ local function schema_install_triggers()
                      '_bucket: %s', err)
         end
     end
-    _bucket:on_replace(bucket_generation_increment)
-    M.bucket_on_replace = bucket_generation_increment
+    _bucket:on_replace(bucket_on_replace_f)
+    M.bucket_on_replace = bucket_on_replace_f
 end
 
 local function schema_install_on_replace(_, new)

vshard/util.lua

@@ -284,6 +284,10 @@ local feature = {
     msgpack_object = lmsgpack.object ~= nil,
     netbox_return_raw = version_is_at_least(2, 10, 0, 'beta', 2, 86),
     multilisten = luri.parse_many ~= nil,
+    -- It appeared earlier but at 2.9.0 there is a strange bug about an immortal
+    -- tuple - after :delete(pk) it still stays in the space. That makes the
+    -- feature barely working, can be treated like it didn't exist.
+    memtx_mvcc = version_is_at_least(2, 10, 0, nil, 0, 0)
 }
 
 return {