This is an automated email from the ASF dual-hosted git repository.
tew pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 987b9c4b fix: background goroutines cannot exit gracefully (#996)
987b9c4b is described below
commit 987b9c4bb8cc6c3c5d31c06938d849d313e4b651
Author: WyRainBow <[email protected]>
AuthorDate: Tue Mar 10 17:35:53 2026 +0800
fix: background goroutines cannot exit gracefully (#996)
* fix: Background Goroutines Cannot Exit Gracefully
* trigger ci: rerun integration tests
* trigger ci: rerun integration tests
---
pkg/datasource/sql/async_worker.go | 10 ++-
pkg/datasource/sql/async_worker_test.go | 50 +++++++++++++
pkg/datasource/sql/at_resource_manager.go | 2 +-
pkg/datasource/sql/datasource/base/meta_cache.go | 87 ++++++++++++----------
.../sql/datasource/base/meta_cache_test.go | 47 +++++++++---
pkg/datasource/sql/datasource/mysql/meta_cache.go | 2 +-
6 files changed, 140 insertions(+), 58 deletions(-)
diff --git a/pkg/datasource/sql/async_worker.go
b/pkg/datasource/sql/async_worker.go
index bfdf736d..9f7f0afc 100644
--- a/pkg/datasource/sql/async_worker.go
+++ b/pkg/datasource/sql/async_worker.go
@@ -70,7 +70,7 @@ type AsyncWorker struct {
rePutBackToQueue prometheus.Counter
}
-func NewAsyncWorker(prom prometheus.Registerer, conf AsyncWorkerConfig,
sourceManager datasource.DataSourceManager) *AsyncWorker {
+func NewAsyncWorker(ctx context.Context, prom prometheus.Registerer, conf
AsyncWorkerConfig, sourceManager datasource.DataSourceManager) *AsyncWorker {
var asyncWorker AsyncWorker
asyncWorker.conf = conf
asyncWorker.commitQueue = make(chan phaseTwoContext,
asyncWorker.conf.ReceiveChanSize)
@@ -97,7 +97,7 @@ func NewAsyncWorker(prom prometheus.Registerer, conf
AsyncWorkerConfig, sourceMa
Help: "the counter of commit failure retry counter",
})
- go asyncWorker.run()
+ go asyncWorker.run(ctx)
return &asyncWorker
}
@@ -122,8 +122,9 @@ func (aw *AsyncWorker) BranchCommit(ctx context.Context,
req rm.BranchResource)
return branch.BranchStatusPhasetwoCommitted, nil
}
-func (aw *AsyncWorker) run() {
+func (aw *AsyncWorker) run(ctx context.Context) {
ticker := time.NewTicker(aw.conf.BufferCleanInterval)
+ defer ticker.Stop()
phaseCtxs := make([]phaseTwoContext, 0, aw.conf.BufferLimit)
for {
select {
@@ -134,6 +135,9 @@ func (aw *AsyncWorker) run() {
}
case <-ticker.C:
aw.doBranchCommit(&phaseCtxs)
+ case <-ctx.Done():
+ ticker.Stop()
+ return
}
}
}
diff --git a/pkg/datasource/sql/async_worker_test.go
b/pkg/datasource/sql/async_worker_test.go
new file mode 100644
index 00000000..e21006ef
--- /dev/null
+++ b/pkg/datasource/sql/async_worker_test.go
@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 sql
+
+import (
+ "context"
+ "testing"
+ "time"
+
+ "github.com/prometheus/client_golang/prometheus"
+)
+
+func TestAsyncWorker_Lifecycle(t *testing.T) {
+ cfg := AsyncWorkerConfig{
+ BufferLimit: 10,
+ BufferCleanInterval: time.Second,
+ ReceiveChanSize: 10,
+ CommitWorkerCount: 1,
+ CommitWorkerBufferSize: 10,
+ }
+
+ ctx, cancel := context.WithCancel(context.Background())
+
+ // Pass nil as sourceManager since we don't process tasks in this test
+ _ = NewAsyncWorker(ctx, prometheus.NewRegistry(), cfg, nil)
+
+ // Allow it to run for a bit
+ time.Sleep(100 * time.Millisecond)
+
+ // Cancel context to stop worker
+ cancel()
+
+ // Give it some time to exit (though we can't verify it easily without
exposing internal state)
+ time.Sleep(50 * time.Millisecond)
+}
diff --git a/pkg/datasource/sql/at_resource_manager.go
b/pkg/datasource/sql/at_resource_manager.go
index 49ce2f51..e6b59f39 100644
--- a/pkg/datasource/sql/at_resource_manager.go
+++ b/pkg/datasource/sql/at_resource_manager.go
@@ -41,7 +41,7 @@ func InitAT(cfg undo.Config, asyncCfg AsyncWorkerConfig) {
}
undo.InitUndoConfig(cfg)
- atSourceManager.worker = NewAsyncWorker(prometheus.DefaultRegisterer,
asyncCfg, atSourceManager)
+ atSourceManager.worker = NewAsyncWorker(context.Background(),
prometheus.DefaultRegisterer, asyncCfg, atSourceManager)
rm.GetRmCacheInstance().RegisterResourceManager(atSourceManager)
}
diff --git a/pkg/datasource/sql/datasource/base/meta_cache.go
b/pkg/datasource/sql/datasource/base/meta_cache.go
index 521ef4a9..7f2766aa 100644
--- a/pkg/datasource/sql/datasource/base/meta_cache.go
+++ b/pkg/datasource/sql/datasource/base/meta_cache.go
@@ -45,31 +45,29 @@ type (
// BaseTableMetaCache
type BaseTableMetaCache struct {
- lock sync.RWMutex
- expireDuration time.Duration
- capity int32
- size int32
- cache map[string]*entry
- cancel context.CancelFunc
- trigger trigger
- db *sql.DB
- cfg *mysql.Config
+ lock sync.RWMutex
+ expireDuration time.Duration
+ refreshInterval time.Duration
+ capity int32
+ size int32
+ cache map[string]*entry
+ trigger trigger
+ db *sql.DB
+ cfg *mysql.Config
}
// NewBaseCache
-func NewBaseCache(capity int32, expireDuration time.Duration, trigger trigger,
db *sql.DB, cfg *mysql.Config) *BaseTableMetaCache {
- ctx, cancel := context.WithCancel(context.Background())
-
+func NewBaseCache(ctx context.Context, capity int32, expireDuration
time.Duration, trigger trigger, db *sql.DB, cfg *mysql.Config)
*BaseTableMetaCache {
c := &BaseTableMetaCache{
- lock: sync.RWMutex{},
- capity: capity,
- size: 0,
- expireDuration: expireDuration,
- cache: map[string]*entry{},
- cancel: cancel,
- trigger: trigger,
- cfg: cfg,
- db: db,
+ lock: sync.RWMutex{},
+ capity: capity,
+ size: 0,
+ expireDuration: expireDuration,
+ refreshInterval: time.Minute,
+ cache: map[string]*entry{},
+ trigger: trigger,
+ cfg: cfg,
+ db: db,
}
c.Init(ctx)
@@ -77,11 +75,10 @@ func NewBaseCache(capity int32, expireDuration
time.Duration, trigger trigger, d
return c
}
-// init
+// Init
func (c *BaseTableMetaCache) Init(ctx context.Context) error {
go c.refresh(ctx)
go c.scanExpire(ctx)
-
return nil
}
@@ -128,10 +125,15 @@ func (c *BaseTableMetaCache) refresh(ctx context.Context)
{
f()
- ticker := time.NewTicker(time.Duration(1 * time.Minute))
+ ticker := time.NewTicker(c.refreshInterval)
defer ticker.Stop()
- for range ticker.C {
- f()
+ for {
+ select {
+ case <-ticker.C:
+ f()
+ case <-ctx.Done():
+ return
+ }
}
}
@@ -139,23 +141,27 @@ func (c *BaseTableMetaCache) refresh(ctx context.Context)
{
func (c *BaseTableMetaCache) scanExpire(ctx context.Context) {
ticker := time.NewTicker(c.expireDuration)
defer ticker.Stop()
- for range ticker.C {
-
- f := func() {
- c.lock.Lock()
- defer c.lock.Unlock()
-
- cur := time.Now()
- for k := range c.cache {
- entry := c.cache[k]
-
- if cur.Sub(entry.lastAccess) > c.expireDuration
{
- delete(c.cache, k)
+ for {
+ select {
+ case <-ticker.C:
+ f := func() {
+ c.lock.Lock()
+ defer c.lock.Unlock()
+
+ cur := time.Now()
+ for k := range c.cache {
+ entry := c.cache[k]
+
+ if cur.Sub(entry.lastAccess) >
c.expireDuration {
+ delete(c.cache, k)
+ }
}
}
- }
- f()
+ f()
+ case <-ctx.Done():
+ return
+ }
}
}
@@ -191,6 +197,5 @@ func (c *BaseTableMetaCache) GetTableMeta(ctx
context.Context, dbName, tableName
}
func (c *BaseTableMetaCache) Destroy() error {
- c.cancel()
return nil
}
diff --git a/pkg/datasource/sql/datasource/base/meta_cache_test.go
b/pkg/datasource/sql/datasource/base/meta_cache_test.go
index 0a5746cd..6f10303b 100644
--- a/pkg/datasource/sql/datasource/base/meta_cache_test.go
+++ b/pkg/datasource/sql/datasource/base/meta_cache_test.go
@@ -78,7 +78,6 @@ func TestBaseTableMetaCache_refresh(t *testing.T) {
capity int32
size int32
cache map[string]*entry
- cancel context.CancelFunc
trigger trigger
db *sql.DB
cfg *mysql.Config
@@ -87,6 +86,7 @@ func TestBaseTableMetaCache_refresh(t *testing.T) {
ctx context.Context
}
ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
tests := []struct {
name string
fields fields
@@ -105,7 +105,6 @@ func TestBaseTableMetaCache_refresh(t *testing.T) {
lastAccess: time.Now(),
},
},
- cancel: cancel,
trigger: &mockTrigger{},
cfg: &mysql.Config{},
},
@@ -124,7 +123,6 @@ func TestBaseTableMetaCache_refresh(t *testing.T) {
lastAccess: time.Now(),
},
},
- cancel: cancel,
trigger: &mockTrigger{},
cfg: &mysql.Config{},
},
@@ -149,14 +147,14 @@ func TestBaseTableMetaCache_refresh(t *testing.T) {
defer loadAllStub.Reset()
c := &BaseTableMetaCache{
- expireDuration: tt.fields.expireDuration,
- capity: tt.fields.capity,
- size: tt.fields.size,
- cache: tt.fields.cache,
- cancel: tt.fields.cancel,
- trigger: tt.fields.trigger,
- db: db,
- cfg: tt.fields.cfg,
+ expireDuration: tt.fields.expireDuration,
+ refreshInterval: time.Minute,
+ capity: tt.fields.capity,
+ size: tt.fields.size,
+ cache: tt.fields.cache,
+ trigger: tt.fields.trigger,
+ db: db,
+ cfg: tt.fields.cfg,
}
go c.refresh(tt.args.ctx)
time.Sleep(time.Second * 3)
@@ -220,7 +218,6 @@ func TestBaseTableMetaCache_refresh_EarlyReturn(t
*testing.T) {
capity: capacity,
size: 0,
cache: tt.cache,
- cancel: cancel,
trigger: &mockTrigger{},
db: tt.db,
cfg: tt.cfg,
@@ -364,3 +361,29 @@ func TestBaseTableMetaCache_GetTableMeta(t *testing.T) {
})
}
}
+
+func TestBaseTableMetaCache_GracefulShutdown(t *testing.T) {
+ // Create context manually as we are bypassing NewBaseCache
+ ctx, cancel := context.WithCancel(context.Background())
+
+ c := &BaseTableMetaCache{
+ expireDuration: 1 * time.Millisecond,
+ refreshInterval: 1 * time.Millisecond,
+ cache: make(map[string]*entry),
+ // db and cfg are nil, so refresh() logic will return early,
which is fine for coverage
+ }
+
+ // Init starts the goroutines
+ err := c.Init(ctx)
+ assert.Nil(t, err)
+
+ // Give enough time for tickers to trigger multiple times
+ time.Sleep(20 * time.Millisecond)
+
+ // Cancel context to stop goroutines
+ cancel()
+
+ // Destroy (now a no-op)
+ err = c.Destroy()
+ assert.Nil(t, err)
+}
diff --git a/pkg/datasource/sql/datasource/mysql/meta_cache.go
b/pkg/datasource/sql/datasource/mysql/meta_cache.go
index 4d7879f7..4f3440ea 100644
--- a/pkg/datasource/sql/datasource/mysql/meta_cache.go
+++ b/pkg/datasource/sql/datasource/mysql/meta_cache.go
@@ -43,7 +43,7 @@ type TableMetaCache struct {
func NewTableMetaInstance(db *sql.DB, cfg *mysql.Config) *TableMetaCache {
tableMetaInstance := &TableMetaCache{
- tableMetaCache: base.NewBaseCache(capacity, EexpireTime,
NewMysqlTrigger(), db, cfg),
+ tableMetaCache: base.NewBaseCache(context.Background(),
capacity, EexpireTime, NewMysqlTrigger(), db, cfg),
db: db,
}
return tableMetaInstance
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
