From b393e021616a32968c9c53db47f9a1bf253f141d Mon Sep 17 00:00:00 2001 From: ljluestc <63439129+ljluestc@users.noreply.github.com> Date: Sun, 13 Sep 2026 14:44:46 -0700 Subject: [PATCH] tikv, txnkv: unify the way background goroutines are spawned Background jobs used to be started in three different ways: a builtin `go` statement, `KVStore.Go`, and `KVTxn.spawnWithStorePool`. Each of them tracked (or failed to track) the goroutine differently, which is the kind of mismatch that caused pingcap/tidb#57213. Make `KVStore.Go` the single entry point. It now registers the job with the store WaitGroup, runs it in the store pool, and rejects new jobs once the store starts closing, so `Close` waits for background jobs instead of racing with them. `KVTxn.spawn` and `KVTxn.spawnWithStorePool` collapse into one `spawn` that goes through it. Along the way: - `Spool.Run` returns `ErrPoolClosed` instead of letting the underlying pool silently drop the function, so a rejected job is no longer indistinguishable from an accepted one. `Spool.Close` is now idempotent. - `KVTxn.spawn` calls the `Post` lifecycle hook when the goroutine is rejected, which keeps it balanced with the `Pre` hook that already ran. - Callers that prepared state for a background job (`cleanWg`, the async pessimistic rollback WaitGroup, the broadcast rate limit) release it when the job is rejected, instead of leaving a waiter blocked forever. - `WaitGroup()` is dropped from the `kvstore` interface used by the transaction package and deprecated on `KVStore`. Foreground fan-out goroutines that the caller joins before returning, such as `batchExecutor.startWorker` and `KVStore.updateMinSafeTS`, are left as plain `go` statements: they are not background jobs and do not outlive their caller. Closes #1510 Signed-off-by: ljluestc <63439129+ljluestc@users.noreply.github.com> --- tikv/kv.go | 73 +++++++++++++++++++---- tikv/pool.go | 40 ++++++++++++- tikv/pool_test.go | 87 ++++++++++++++++++++++++++++ txnkv/transaction/2pc.go | 28 ++++++--- txnkv/transaction/pipelined_flush.go | 11 +++- txnkv/transaction/spawn_test.go | 78 +++++++++++++++++++++++++ txnkv/transaction/test_util.go | 6 -- txnkv/transaction/txn.go | 56 +++++++++--------- txnkv/transaction/txn_file.go | 2 +- txnkv/transaction/txn_file_test.go | 1 - 10 files changed, 325 insertions(+), 57 deletions(-) create mode 100644 tikv/pool_test.go create mode 100644 txnkv/transaction/spawn_test.go diff --git a/tikv/kv.go b/tikv/kv.go index 984093443a..f0ec2fec22 100644 --- a/tikv/kv.go +++ b/tikv/kv.go @@ -120,6 +120,10 @@ func createEtcdKV(addrs []string, tlsConfig *tls.Config) (*clientv3.Client, erro // or at runtime by calling SetLowResolutionTimestampUpdateInterval on the oracle var defaultOracleUpdateInterval = 2 * time.Second +// ErrStoreClosed is returned by KVStore.Go when the store is closing, which +// means the submitted function is rejected and will never be executed. +var ErrStoreClosed = errors.New("the store is closed") + // KVStore contains methods to interact with a TiKV cluster. type KVStore struct { clusterID uint64 @@ -160,16 +164,57 @@ type KVStore struct { ctx context.Context cancel context.CancelFunc - wg sync.WaitGroup close atomicutil.Bool gP Pool + + // backgroundJobs guards the spawning of background goroutines. Every + // background goroutine owned by the store is started by Go and tracked by + // wg, and no new one is admitted once the store starts closing. The lock is + // what makes `wg.Add` in Go and `wg.Wait` in stopBackgroundJobs mutually + // exclusive, which a plain atomic flag cannot guarantee. + backgroundJobs struct { + sync.RWMutex + closed bool + } + wg sync.WaitGroup } var _ Storage = (*KVStore)(nil) -// Go run the function in a separate goroutine. +// Go runs f in a background goroutine owned by the store. +// +// This is the only supported way to spawn a background goroutine on behalf of a +// store: f runs in the store's goroutine pool and is tracked by the store, so +// that Close waits for it instead of racing with it. +// +// A non-nil error means f is rejected and will never run, which happens when +// the store is closing. The caller then owns the cleanup of whatever it has +// prepared for f, for example releasing a token or marking a WaitGroup done. func (s *KVStore) Go(f func()) error { - return s.gP.Run(f) + s.backgroundJobs.RLock() + if s.backgroundJobs.closed { + s.backgroundJobs.RUnlock() + return ErrStoreClosed + } + s.wg.Add(1) + err := s.gP.Run(func() { + defer s.wg.Done() + f() + }) + s.backgroundJobs.RUnlock() + if err != nil { + s.wg.Done() + } + return err +} + +// stopBackgroundJobs rejects any new background job and waits for the running +// ones to finish. +func (s *KVStore) stopBackgroundJobs() { + s.backgroundJobs.Lock() + s.backgroundJobs.closed = true + s.backgroundJobs.Unlock() + s.wg.Wait() } // UpdateTxnSafePointCache updates the cached txn safe point, which is used for safety check of data access @@ -359,6 +404,9 @@ func NewKVStore( if retErr != nil { regionCache.Close() cancel() + // Background jobs may already have been started below, stop them + // before reporting the failure. + store.stopBackgroundJobs() } }() @@ -378,9 +426,11 @@ func NewKVStore( store.lockResolver = txnlock.NewLockResolver(store) loadOption(store, opt...) - store.wg.Add(2) - go store.runTxnSafePointUpdater() - go store.safeTSUpdater() + for _, updater := range []func(){store.runTxnSafePointUpdater, store.safeTSUpdater} { + if err := store.Go(updater); err != nil { + return nil, errors.WithStack(err) + } + } return store, nil } @@ -425,7 +475,6 @@ func (s *KVStore) IsLatchEnabled() bool { } func (s *KVStore) runTxnSafePointUpdater() { - defer s.wg.Done() if _, e := util.EvalFailpoint("noBuiltInTxnSafePointUpdater"); e == nil { return } @@ -502,7 +551,7 @@ func (s *KVStore) Close() error { defer s.gP.Close() s.close.Store(true) s.cancel() - s.wg.Wait() + s.stopBackgroundJobs() s.oracle.Close() if s.txnLatches != nil { @@ -715,7 +764,12 @@ func (s *KVStore) IsClose() bool { return s.close.Load() } -// WaitGroup returns wg +// WaitGroup returns the WaitGroup that tracks the store's background +// goroutines. +// +// Deprecated: use Go to spawn a background goroutine, which registers it with +// this WaitGroup for you. Adding to the WaitGroup directly races with the +// Wait done by Close. func (s *KVStore) WaitGroup() *sync.WaitGroup { return &s.wg } @@ -776,7 +830,6 @@ func (s *KVStore) updateMinSafeTS(txnScope string, storeIDs []uint64) { } func (s *KVStore) safeTSUpdater() { - defer s.wg.Done() t := time.NewTicker(safeTSUpdateInterval) if _, e := util.EvalFailpoint("mockFastSafeTSUpdater"); e == nil { t.Reset(time.Millisecond * 100) diff --git a/tikv/pool.go b/tikv/pool.go index ca2799b53d..82ee50338c 100644 --- a/tikv/pool.go +++ b/tikv/pool.go @@ -15,31 +15,65 @@ package tikv import ( + "sync" "time" + "github.com/pingcap/errors" "github.com/tiancaiamao/gp" ) +// ErrPoolClosed is returned by Pool.Run when the pool has been closed. The +// submitted function is rejected and will never be executed. +var ErrPoolClosed = errors.New("the goroutine pool is closed") + // Pool is a simple interface for goroutine pool. type Pool interface { + // Run executes the function in a separate goroutine. A non-nil error must + // be returned if and only if the function is rejected, which means it will + // never be executed and the caller owns the cleanup of anything it has + // prepared for the function. Run(func()) error + // Close releases the goroutines of the pool. In-flight functions are not + // interrupted. It must be safe to call Close more than once. Close() } // Spool is a simple implementation of Pool. type Spool struct { gp.Pool + mu struct { + sync.RWMutex + closed bool + } } // NewSpool creates a new Spool. func NewSpool(n int, dur time.Duration) *Spool { - return &Spool{ - *gp.New(n, dur), - } + return &Spool{Pool: *gp.New(n, dur)} } // Run implements Pool.Run. func (p *Spool) Run(fn func()) error { + // The underlying pool silently drops the functions submitted after it is + // closed. Submitting under the read lock makes the outcome unambiguous: the + // function is either accepted by a pool that is still open, or rejected + // with ErrPoolClosed here. + p.mu.RLock() + defer p.mu.RUnlock() + if p.mu.closed { + return ErrPoolClosed + } p.Go(fn) return nil } + +// Close implements Pool.Close. +func (p *Spool) Close() { + p.mu.Lock() + defer p.mu.Unlock() + if p.mu.closed { + return + } + p.mu.closed = true + p.Pool.Close() +} diff --git a/tikv/pool_test.go b/tikv/pool_test.go new file mode 100644 index 0000000000..5374542ce6 --- /dev/null +++ b/tikv/pool_test.go @@ -0,0 +1,87 @@ +// Copyright 2025 TiKV Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package tikv + +import ( + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/require" + "github.com/tikv/client-go/v2/internal/mockstore/mocktikv" + "github.com/tikv/client-go/v2/testutils" +) + +func TestSpoolRejectsTasksAfterClose(t *testing.T) { + p := NewSpool(2, 10*time.Millisecond) + + var wg sync.WaitGroup + wg.Add(1) + require.NoError(t, p.Run(wg.Done)) + wg.Wait() + + p.Close() + + var executed atomic.Bool + require.ErrorIs(t, p.Run(func() { executed.Store(true) }), ErrPoolClosed) + require.False(t, executed.Load()) + + // Close must be idempotent: closing the underlying pool twice panics. + p.Close() +} + +func TestKVStoreGoTracksBackgroundJobs(t *testing.T) { + client, cluster, pdClient, err := testutils.NewMockTiKV("", nil) + require.NoError(t, err) + mocktikv.BootstrapWithMultiStores(cluster, 1) + store, err := NewTestTiKVStore(client, pdClient, nil, nil, 0) + require.NoError(t, err) + + started := make(chan struct{}) + release := make(chan struct{}) + finished := make(chan struct{}) + require.NoError(t, store.Go(func() { + close(started) + <-release + close(finished) + })) + <-started + + var closeErr error + returned := make(chan struct{}) + go func() { + defer close(returned) + closeErr = store.Close() + }() + + // Close waits for the background job instead of racing with it. + select { + case <-returned: + require.FailNow(t, "Close returned before the background job finished") + case <-time.After(100 * time.Millisecond): + } + + close(release) + <-finished + <-returned + require.NoError(t, closeErr) + + // A closing store admits no new background job, and reports the rejection + // so that the caller can clean up after it. + var executed atomic.Bool + require.ErrorIs(t, store.Go(func() { executed.Store(true) }), ErrStoreClosed) + require.False(t, executed.Load()) +} diff --git a/txnkv/transaction/2pc.go b/txnkv/transaction/2pc.go index 50b7619c9a..c97128132f 100644 --- a/txnkv/transaction/2pc.go +++ b/txnkv/transaction/2pc.go @@ -116,13 +116,14 @@ type kvstore interface { GetTiKVClient() (client client.Client) GetLockResolver() *txnlock.LockResolver Ctx() context.Context - WaitGroup() *sync.WaitGroup // TxnLatches returns txnLatches. TxnLatches() *latch.LatchesScheduler GetClusterID() uint64 // IsClose checks whether the store is closed. IsClose() bool - // Go run the function in a separate goroutine. + // Go runs the function in a background goroutine tracked by the store. It + // returns a non-nil error if and only if the function is rejected and will + // never be executed, which happens when the store is closing. Go(f func()) error } @@ -1096,7 +1097,7 @@ func (c *twoPhaseCommitter) doActionOnGroupMutations(bo *retry.Backoffer, action zap.Uint64("sessionID", c.sessionID)) return nil } - err = c.txn.spawnWithStorePool(func() { + err = c.txn.spawn(func() { if c.sessionID > 0 { if v, err := util.EvalFailpoint("beforeCommitSecondaries"); err == nil { if s, ok := v.(string); !ok { @@ -1478,7 +1479,7 @@ func broadcastToAllStores( wg.Add(1) target := s - err := txn.spawnWithStorePool(func() { + err := txn.spawn(func() { defer wg.Done() defer func() { <-rateLimit }() @@ -1522,7 +1523,7 @@ func broadcastToAllStores( wg.Wait() } - if err := txn.spawnWithStorePool(broadcastFunc); err != nil { + if err := txn.spawn(broadcastFunc); err != nil { logutil.Logger(store.Ctx()).Error("failed to spawn goroutine for broadcasting txn status", zap.Error(err)) } @@ -1675,7 +1676,7 @@ func (c *twoPhaseCommitter) cleanup(ctx context.Context) { return } c.cleanWg.Add(1) - c.txn.spawn(func() { + err := c.txn.spawn(func() { defer c.cleanWg.Done() if _, err := util.EvalFailpoint("commitFailedSkipCleanup"); err == nil { @@ -1735,6 +1736,14 @@ func (c *twoPhaseCommitter) cleanup(ctx context.Context) { } } }) + if err != nil { + // The cleanup will never run, release the waiter so that it does not + // block forever. + c.cleanWg.Done() + logutil.Logger(ctx).Warn("twoPhaseCommitter fail to spawn the cleanup goroutine", + zap.Uint64("txnStartTS", c.startTS), zap.Bool("isPessimistic", c.isPessimistic), + zap.Bool("isOnePC", c.isOnePC()), zap.Error(err)) + } } // execute executes the two-phase commit protocol. @@ -2056,7 +2065,7 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) { zap.Uint64("sessionID", c.sessionID)) return nil } - c.txn.spawn(func() { + err := c.txn.spawn(func() { if _, err := util.EvalFailpoint("asyncCommitDoNothing"); err == nil { return } @@ -2067,6 +2076,11 @@ func (c *twoPhaseCommitter) execute(ctx context.Context) (err error) { zap.Uint64("startTS", c.startTS), zap.Uint64("commitTS", c.commitTS), zap.Error(err)) } }) + if err != nil { + logutil.Logger(ctx).Warn("2PC fail to spawn the async commit goroutine", + zap.Uint64("startTS", c.startTS), zap.Uint64("commitTS", c.commitTS), + zap.Uint64("sessionID", c.sessionID), zap.Error(err)) + } return nil } return c.commitTxn(ctx, commitDetail) diff --git a/txnkv/transaction/pipelined_flush.go b/txnkv/transaction/pipelined_flush.go index c5649e350d..e2fa314a47 100644 --- a/txnkv/transaction/pipelined_flush.go +++ b/txnkv/transaction/pipelined_flush.go @@ -490,7 +490,7 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end runner.SetStatLogInterval(30 * time.Second) runner.SetRegionsPerTask(1) - c.txn.spawnWithStorePool(func() { + errSpawn := c.txn.spawn(func() { if err = runner.RunOnRange(bo.GetCtx(), start, end); err != nil { logutil.Logger(bo.GetCtx()).Error("[pipelined dml] resolve flushed locks failed", zap.String("txn-status", status), @@ -534,4 +534,13 @@ func (c *twoPhaseCommitter) resolveFlushedLocks(bo *retry.Backoffer, start, end ) } }) + if errSpawn != nil { + logutil.Logger(bo.GetCtx()).Warn("[pipelined dml] failed to spawn the goroutine resolving flushed locks", + zap.String("txn-status", status), + zap.Uint64("startTS", c.startTS), + zap.Uint64("commitTS", commitTs), + zap.Uint64("session", c.sessionID), + zap.Error(errSpawn), + ) + } } diff --git a/txnkv/transaction/spawn_test.go b/txnkv/transaction/spawn_test.go new file mode 100644 index 0000000000..1634f68ea2 --- /dev/null +++ b/txnkv/transaction/spawn_test.go @@ -0,0 +1,78 @@ +// Copyright 2025 TiKV Authors +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package transaction + +import ( + "sync" + "sync/atomic" + "testing" + + "github.com/pkg/errors" + "github.com/stretchr/testify/require" +) + +// spawnKVStore records the functions submitted through Go, and can be told to +// reject them the way a closing store does. +type spawnKVStore struct { + unimplementedKVStore + rejectErr error + wg sync.WaitGroup +} + +func (s *spawnKVStore) Go(f func()) error { + if s.rejectErr != nil { + return s.rejectErr + } + s.wg.Add(1) + go func() { + defer s.wg.Done() + f() + }() + return nil +} + +func TestSpawnKeepsLifecycleHooksBalanced(t *testing.T) { + rejectErr := errors.New("the store is closed") + + for _, rejected := range []bool{false, true} { + store := &spawnKVStore{} + if rejected { + store.rejectErr = rejectErr + } + + var pre, post atomic.Int32 + txn := &KVTxn{store: store} + txn.SetBackgroundGoroutineLifecycleHooks(LifecycleHooks{ + Pre: func() { pre.Add(1) }, + Post: func() { post.Add(1) }, + }) + + var executed atomic.Bool + err := txn.spawn(func() { executed.Store(true) }) + store.wg.Wait() + + if rejected { + require.ErrorIs(t, err, rejectErr) + require.False(t, executed.Load(), "a rejected function must not run") + } else { + require.NoError(t, err) + require.True(t, executed.Load()) + } + // Post must run even when the goroutine never starts, otherwise the + // caller of Pre waits for a goroutine that does not exist. + require.Equal(t, int32(1), pre.Load()) + require.Equal(t, int32(1), post.Load()) + } +} diff --git a/txnkv/transaction/test_util.go b/txnkv/transaction/test_util.go index f350eedd48..90af60187b 100644 --- a/txnkv/transaction/test_util.go +++ b/txnkv/transaction/test_util.go @@ -16,7 +16,6 @@ package transaction import ( "context" - "sync" "time" "github.com/pingcap/kvproto/pkg/keyspacepb" @@ -473,11 +472,6 @@ func (u *unimplementedKVStore) TxnLatches() *latch.LatchesScheduler { panic("unimplemented") } -// WaitGroup implements kvstore. -func (u *unimplementedKVStore) WaitGroup() *sync.WaitGroup { - panic("unimplemented") -} - // WaitScatterRegionFinish implements kvstore. func (u *unimplementedKVStore) WaitScatterRegionFinish(ctx context.Context, regionID uint64, backOff int) error { panic("unimplemented") diff --git a/txnkv/transaction/txn.go b/txnkv/transaction/txn.go index ceeccf5324..80c432a9e8 100644 --- a/txnkv/transaction/txn.go +++ b/txnkv/transaction/txn.go @@ -464,38 +464,30 @@ func (txn *KVTxn) SetBackgroundGoroutineLifecycleHooks(hooks LifecycleHooks) { txn.backgroundGoroutineLifecycleHooks = hooks } -// spawn starts a goroutine to run the given function. -func (txn *KVTxn) spawn(f func()) { - if txn.backgroundGoroutineLifecycleHooks.Pre != nil { - txn.backgroundGoroutineLifecycleHooks.Pre() - } - txn.store.WaitGroup().Add(1) - go func() { - if txn.backgroundGoroutineLifecycleHooks.Post != nil { - defer txn.backgroundGoroutineLifecycleHooks.Post() - } - defer txn.store.WaitGroup().Done() - - f() - }() -} - -// spawnWithStorePool starts a goroutine to run the given function with the store's goroutine pool. -func (txn *KVTxn) spawnWithStorePool(f func()) error { +// spawn starts a background goroutine to run the given function. +// +// It is the only way a transaction spawns a background goroutine: the function +// runs in the store's goroutine pool and is tracked by the store, so that +// closing the store waits for it instead of racing with it. +// +// A non-nil error means the function is rejected and will never run, which +// happens when the store is closing. The caller then owns the cleanup of +// whatever it has prepared for the function. +func (txn *KVTxn) spawn(f func()) error { if txn.backgroundGoroutineLifecycleHooks.Pre != nil { txn.backgroundGoroutineLifecycleHooks.Pre() } - txn.store.WaitGroup().Add(1) err := txn.store.Go(func() { if txn.backgroundGoroutineLifecycleHooks.Post != nil { defer txn.backgroundGoroutineLifecycleHooks.Post() } - defer txn.store.WaitGroup().Done() f() }) - if err != nil { - txn.store.WaitGroup().Done() + if err != nil && txn.backgroundGoroutineLifecycleHooks.Post != nil { + // Pre has been called but the goroutine will never run, so Post must be + // called here to keep the hooks balanced. + txn.backgroundGoroutineLifecycleHooks.Post() } return err } @@ -1020,7 +1012,7 @@ func (txn *KVTxn) Rollback() error { // no need to clean up locks when no flush triggered. pipelinedStart, pipelinedEnd := txn.committer.pipelinedCommitInfo.pipelinedStart, txn.committer.pipelinedCommitInfo.pipelinedEnd needCleanUpLocks := len(pipelinedStart) != 0 && len(pipelinedEnd) != 0 - txn.spawnWithStorePool( + err := txn.spawn( func() { broadcastToAllStores( txn, @@ -1042,6 +1034,10 @@ func (txn *KVTxn) Rollback() error { ) }, ) + if err != nil { + logutil.BgLogger().Warn("[kv] failed to spawn the goroutine broadcasting the rolled back txn status", + zap.Uint64("txnStartTS", txn.startTS), zap.Error(err)) + } if needCleanUpLocks { rollbackBo := retry.NewBackofferWithVars(txn.store.Ctx(), CommitSecondaryMaxBackoff, txn.vars) txn.committer.resolveFlushedLocks(rollbackBo, pipelinedStart, pipelinedEnd, false) @@ -2141,16 +2137,14 @@ func (txn *KVTxn) asyncPessimisticRollback(ctx context.Context, keys [][]byte, s } wg := new(sync.WaitGroup) wg.Add(1) - txn.store.WaitGroup().Add(1) - go func() { - defer txn.store.WaitGroup().Done() + err := txn.spawn(func() { + defer wg.Done() if val, err := util.EvalFailpoint("beforeAsyncPessimisticRollback"); err == nil { if s, ok := val.(string); ok { switch s { case "skip": logutil.Logger(ctx).Info("[failpoint] injected skip async pessimistic rollback", zap.Uint64("txnStartTS", txn.startTS)) - wg.Done() return case "delay": duration := time.Duration(rand.Int63n(int64(time.Second) * 2)) @@ -2165,8 +2159,14 @@ func (txn *KVTxn) asyncPessimisticRollback(ctx context.Context, keys [][]byte, s if err != nil { logutil.Logger(ctx).Warn("[kv] pessimisticRollback failed.", zap.Error(err)) } + }) + if err != nil { + // The rollback will never run, release the waiter so that it does not + // block forever. wg.Done() - }() + logutil.Logger(ctx).Warn("[kv] failed to spawn the async pessimistic rollback", + zap.Uint64("txnStartTS", txn.startTS), zap.Error(err)) + } return wg } diff --git a/txnkv/transaction/txn_file.go b/txnkv/transaction/txn_file.go index 227d93c0dd..a69bd3e0e8 100644 --- a/txnkv/transaction/txn_file.go +++ b/txnkv/transaction/txn_file.go @@ -1059,7 +1059,7 @@ func (c *twoPhaseCommitter) executeTxnFileAction(bo *retry.Backoffer, chunkSlice zap.Stringer("action", action)) return nil } - errGo := c.txn.spawnWithStorePool(func() { + errGo := c.txn.spawn(func() { err := c.executeTxnFileSliceWithRetry(secondaryBo, chunkSlice, secondaries, action) logutil.Logger(bo.GetCtx()).Debug("txn file: async execute secondaries finished", zap.Uint64("startTS", c.startTS), diff --git a/txnkv/transaction/txn_file_test.go b/txnkv/transaction/txn_file_test.go index 055d5cbd58..85e7667423 100644 --- a/txnkv/transaction/txn_file_test.go +++ b/txnkv/transaction/txn_file_test.go @@ -1310,7 +1310,6 @@ func (s *stubKVStore) SendReq(_ *retry.Backoffer, _ *tikvrpc.Request, _ locate.R func (s *stubKVStore) GetTiKVClient() client.Client { panic("not implemented") } func (s *stubKVStore) GetLockResolver() *txnlock.LockResolver { panic("not implemented") } func (s *stubKVStore) Ctx() context.Context { panic("not implemented") } -func (s *stubKVStore) WaitGroup() *sync.WaitGroup { panic("not implemented") } func (s *stubKVStore) TxnLatches() *latch.LatchesScheduler { panic("not implemented") } func (s *stubKVStore) GetClusterID() uint64 { return 0 } func (s *stubKVStore) IsClose() bool { return false }