Skip to content

Commit

Permalink
plugin: fix bug that watch loop will refresh frequently when channel …
Browse files Browse the repository at this point in the history
  • Loading branch information
ti-chi-bot authored Dec 8, 2023
1 parent a102bd1 commit 27ee7b8
Show file tree
Hide file tree
Showing 2 changed files with 67 additions and 3 deletions.
27 changes: 24 additions & 3 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 @@ -246,12 +247,32 @@ type flushWatcher struct {
}

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))
time.Sleep(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))
disabled, err := w.getPluginDisabledFlag()
if err != nil {
logutil.BgLogger().Error("get plugin disabled flag failure", zap.String("plugin", w.manifest.Name), zap.Error(err))
Expand Down
43 changes: 43 additions & 0 deletions plugin/plugin_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,13 @@ import (
"context"
"io"
"strconv"
"sync/atomic"
"testing"
"time"

"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 +245,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 27ee7b8

Please sign in to comment.