Skip to content

Commit

Permalink
This is an automated cherry-pick of pingcap#49275
Browse files Browse the repository at this point in the history
Signed-off-by: ti-chi-bot <[email protected]>
  • Loading branch information
lcwangchao authored and ti-chi-bot committed Dec 8, 2023
1 parent 66f9635 commit e500fbb
Show file tree
Hide file tree
Showing 3 changed files with 75 additions and 4 deletions.
5 changes: 5 additions & 0 deletions plugin/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -38,6 +38,10 @@ go_test(
],
embed = [":plugin"],
flaky = True,
<<<<<<< HEAD:plugin/BUILD.bazel
=======
shard_count = 12,
>>>>>>> 169617d22dc (plugin: fix bug that watch loop will refresh frequently when channel closed (#49275)):pkg/plugin/BUILD.bazel
deps = [
"//kv",
"//parser/mysql",
Expand All @@ -47,6 +51,7 @@ go_test(
"//testkit",
"//testkit/testsetup",
"@com_github_stretchr_testify//require",
"@io_etcd_go_etcd_client_v3//:client",
"@org_uber_go_goleak//:goleak",
],
)
27 changes: 23 additions & 4 deletions plugin/plugin.go
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ import (
gplugin "plugin"
"strconv"
"sync/atomic"
"time"
"unsafe"

"github.com/pingcap/errors"
Expand Down Expand Up @@ -272,14 +273,32 @@ func (w *flushWatcher) refreshPluginState() error {
}
return nil
}

func (w *flushWatcher) watchLoop() {
watchChan := w.etcd.Watch(w.ctx, w.path)
const reWatchInterval = time.Second * 5
logutil.BgLogger().Info("plugin flushWatcher loop started", zap.String("plugin", w.manifest.Name))
for w.ctx.Err() == nil {
ch := w.etcd.Watch(w.ctx, w.path)
if exit := w.watchLoopWithChan(ch); exit {
break
}

logutil.BgLogger().Info(
"plugin flushWatcher old chan closed, restart loop later",
zap.String("plugin", w.manifest.Name),
zap.Duration("after", reWatchInterval))
}
}

func (w *flushWatcher) watchLoopWithChan(ch clientv3.WatchChan) (exit bool) {
for {
select {
case <-w.ctx.Done():
return
case <-watchChan:
return true
case _, ok := <-ch:
if !ok {
return false
}
logutil.BgLogger().Info("plugin flushWatcher detected event to reload plugin config", zap.String("plugin", w.manifest.Name))
_ = w.refreshPluginState()
}
}
Expand Down
47 changes: 47 additions & 0 deletions plugin/plugin_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,17 @@ import (
"context"
"io"
"strconv"
"sync/atomic"
"testing"
<<<<<<< HEAD:plugin/plugin_test.go
=======
"time"
"unsafe"
>>>>>>> 169617d22dc (plugin: fix bug that watch loop will refresh frequently when channel closed (#49275)):pkg/plugin/plugin_test.go

"github.com/pingcap/tidb/sessionctx/variable"
"github.com/stretchr/testify/require"
clientv3 "go.etcd.io/etcd/client/v3"
)

func TestLoadPluginSuccess(t *testing.T) {
Expand Down Expand Up @@ -242,3 +249,43 @@ func TestPluginsClone(t *testing.T) {
require.Equal(t, uint16(1), cps.versions["whitelist"])
require.Len(t, cps.dyingPlugins, 1)
}

func TestPluginWatcherLoop(t *testing.T) {
// exit when ctx done
ctx, cancel := context.WithCancel(context.Background())
watcher := &flushWatcher{
ctx: ctx,
manifest: &Manifest{
Name: "test",
},
}
ch := make(chan clientv3.WatchResponse)
var cancelled atomic.Bool
go func() {
time.Sleep(10 * time.Millisecond)
cancelled.Store(true)
cancel()
}()
exit := watcher.watchLoopWithChan(ch)
require.True(t, exit)
require.True(t, cancelled.Load())

// exit when ch closed
watcher = &flushWatcher{
ctx: context.Background(),
manifest: &Manifest{
Name: "test",
},
}

var closed atomic.Bool
ch = make(chan clientv3.WatchResponse)
go func() {
time.Sleep(10 * time.Millisecond)
closed.Store(true)
close(ch)
}()
exit = watcher.watchLoopWithChan(ch)
require.False(t, exit)
require.True(t, cancelled.Load())
}

0 comments on commit e500fbb

Please sign in to comment.