MM-42412 - guard watcher stop/stopped channels with mutex to prevent race condition (#19759)
* MM-42412 - Guard watcher stop/stopped channels with mutex to prevent race condition * reinitialize channels on Stop fn call Co-authored-by: Pablo Velez Vidal <pablo.velez@mattermost.com> Co-authored-by: Mattermod <mattermod@users.noreply.github.com>
Этот коммит содержится в:
коммит произвёл
GitHub
родитель
9d8c030b8b
Коммит
f6d9f2c184
@@ -26,6 +26,8 @@ type Watcher struct {
|
|||||||
|
|
||||||
func (srv *JobServer) MakeWatcher(workers *Workers, pollingInterval int) *Watcher {
|
func (srv *JobServer) MakeWatcher(workers *Workers, pollingInterval int) *Watcher {
|
||||||
return &Watcher{
|
return &Watcher{
|
||||||
|
stop: make(chan struct{}),
|
||||||
|
stopped: make(chan struct{}),
|
||||||
pollingInterval: pollingInterval,
|
pollingInterval: pollingInterval,
|
||||||
workers: workers,
|
workers: workers,
|
||||||
srv: srv,
|
srv: srv,
|
||||||
@@ -34,8 +36,6 @@ func (srv *JobServer) MakeWatcher(workers *Workers, pollingInterval int) *Watche
|
|||||||
|
|
||||||
func (watcher *Watcher) Start() {
|
func (watcher *Watcher) Start() {
|
||||||
mlog.Debug("Watcher Started")
|
mlog.Debug("Watcher Started")
|
||||||
watcher.stop = make(chan struct{})
|
|
||||||
watcher.stopped = make(chan struct{})
|
|
||||||
// Delay for some random number of milliseconds before starting to ensure that multiple
|
// Delay for some random number of milliseconds before starting to ensure that multiple
|
||||||
// instances of the jobserver don't poll at a time too close to each other.
|
// instances of the jobserver don't poll at a time too close to each other.
|
||||||
rand.Seed(time.Now().UTC().UnixNano())
|
rand.Seed(time.Now().UTC().UnixNano())
|
||||||
@@ -61,6 +61,9 @@ func (watcher *Watcher) Stop() {
|
|||||||
mlog.Debug("Watcher Stopping")
|
mlog.Debug("Watcher Stopping")
|
||||||
close(watcher.stop)
|
close(watcher.stop)
|
||||||
<-watcher.stopped
|
<-watcher.stopped
|
||||||
|
|
||||||
|
watcher.stop = make(chan struct{})
|
||||||
|
watcher.stopped = make(chan struct{})
|
||||||
}
|
}
|
||||||
|
|
||||||
func (watcher *Watcher) PollAndNotify() {
|
func (watcher *Watcher) PollAndNotify() {
|
||||||
|
|||||||
@@ -11,7 +11,6 @@ import (
|
|||||||
)
|
)
|
||||||
|
|
||||||
func TestStartWorkers(t *testing.T) {
|
func TestStartWorkers(t *testing.T) {
|
||||||
t.Skip("MM-42412")
|
|
||||||
t.Run("uninitialized", func(t *testing.T) {
|
t.Run("uninitialized", func(t *testing.T) {
|
||||||
jobServer, _, _ := makeJobServer(t)
|
jobServer, _, _ := makeJobServer(t)
|
||||||
err := jobServer.StartWorkers()
|
err := jobServer.StartWorkers()
|
||||||
@@ -44,7 +43,6 @@ func TestStartWorkers(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func TestStopWorkers(t *testing.T) {
|
func TestStopWorkers(t *testing.T) {
|
||||||
t.Skip("MM-42412")
|
|
||||||
t.Run("uninitialized", func(t *testing.T) {
|
t.Run("uninitialized", func(t *testing.T) {
|
||||||
jobServer, _, _ := makeJobServer(t)
|
jobServer, _, _ := makeJobServer(t)
|
||||||
err := jobServer.StopWorkers()
|
err := jobServer.StopWorkers()
|
||||||
|
|||||||
Ссылка в новой задаче
Block a user