simplify pluginwatcher closing

k3s-v1.15.3
Mike Danese 2019-05-14 18:21:48 -07:00
parent 4990c5e41c
commit 531a50c776
1 changed files with 10 additions and 22 deletions

View File

@ -39,10 +39,10 @@ import (
type Watcher struct { type Watcher struct {
path string path string
deprecatedPath string deprecatedPath string
stopCh chan interface{} stopCh chan struct{}
stopped chan struct{}
fs utilfs.Filesystem fs utilfs.Filesystem
fsWatcher *fsnotify.Watcher fsWatcher *fsnotify.Watcher
wg sync.WaitGroup
mutex sync.Mutex mutex sync.Mutex
handlers map[string]PluginHandler handlers map[string]PluginHandler
@ -88,7 +88,8 @@ func (w *Watcher) getHandler(pluginType string) (PluginHandler, bool) {
// Start watches for the creation of plugin sockets at the path // Start watches for the creation of plugin sockets at the path
func (w *Watcher) Start() error { func (w *Watcher) Start() error {
klog.V(2).Infof("Plugin Watcher Start at %s", w.path) klog.V(2).Infof("Plugin Watcher Start at %s", w.path)
w.stopCh = make(chan interface{}) w.stopCh = make(chan struct{})
w.stopped = make(chan struct{})
// Creating the directory to be watched if it doesn't exist yet, // Creating the directory to be watched if it doesn't exist yet,
// and walks through the directory to discover the existing plugins. // and walks through the directory to discover the existing plugins.
@ -104,22 +105,20 @@ func (w *Watcher) Start() error {
// Traverse plugin dir and add filesystem watchers before starting the plugin processing goroutine. // Traverse plugin dir and add filesystem watchers before starting the plugin processing goroutine.
if err := w.traversePluginDir(w.path); err != nil { if err := w.traversePluginDir(w.path); err != nil {
w.Stop() w.fsWatcher.Close()
return fmt.Errorf("failed to traverse plugin socket path %q, err: %v", w.path, err) return fmt.Errorf("failed to traverse plugin socket path %q, err: %v", w.path, err)
} }
// Traverse deprecated plugin dir, if specified. // Traverse deprecated plugin dir, if specified.
if len(w.deprecatedPath) != 0 { if len(w.deprecatedPath) != 0 {
if err := w.traversePluginDir(w.deprecatedPath); err != nil { if err := w.traversePluginDir(w.deprecatedPath); err != nil {
w.Stop() w.fsWatcher.Close()
return fmt.Errorf("failed to traverse deprecated plugin socket path %q, err: %v", w.deprecatedPath, err) return fmt.Errorf("failed to traverse deprecated plugin socket path %q, err: %v", w.deprecatedPath, err)
} }
} }
w.wg.Add(1) go func() {
go func(fsWatcher *fsnotify.Watcher) { defer close(w.stopped)
defer w.wg.Done()
for { for {
select { select {
case event := <-fsWatcher.Events: case event := <-fsWatcher.Events:
@ -135,17 +134,15 @@ func (w *Watcher) Start() error {
klog.Errorf("error %v when handling delete event: %s", err, event) klog.Errorf("error %v when handling delete event: %s", err, event)
} }
} }
continue
case err := <-fsWatcher.Errors: case err := <-fsWatcher.Errors:
if err != nil { if err != nil {
klog.Errorf("fsWatcher received error: %v", err) klog.Errorf("fsWatcher received error: %v", err)
} }
continue
case <-w.stopCh: case <-w.stopCh:
return return
} }
} }
}(fsWatcher) }()
return nil return nil
} }
@ -154,18 +151,9 @@ func (w *Watcher) Start() error {
func (w *Watcher) Stop() error { func (w *Watcher) Stop() error {
close(w.stopCh) close(w.stopCh)
c := make(chan struct{})
var once sync.Once
closeFunc := func() { close(c) }
go func() {
defer once.Do(closeFunc)
w.wg.Wait()
}()
select { select {
case <-c: case <-w.stopped:
case <-time.After(11 * time.Second): case <-time.After(11 * time.Second):
once.Do(closeFunc)
return fmt.Errorf("timeout on stopping watcher") return fmt.Errorf("timeout on stopping watcher")
} }