potential deadlock fix

notify.go

@@ -9,7 +9,8 @@ var g *Tree
 func tree() *Tree {
 	once.Do(func() {
 		if g == nil {
-			g = NewTree(NewWatcher(nil))
+			c := make(chan EventInfo, 128)
+			g = NewTree(NewWatcher(c), c)
 		}
 	})
 	return g

tree.go

@@ -106,15 +106,13 @@ func (t *Tree) loopdispatch(c <-chan EventInfo) {
 }
 
 // NewTree TODO
-func NewTree(w Watcher) *Tree {
-	c := make(chan EventInfo, 128)
+func NewTree(w Watcher, c <-chan EventInfo) *Tree {
 	t := &Tree{
 		Root: Node{Child: make(map[string]Node), Watch: make(Watchpoint)},
 		cnd:  make(ChanNodesMap),
 		stop: make(chan struct{}),
 	}
 	t.setimpl(w)
-	t.impl.Dispatch(c, t.stop)
 	go t.loopdispatch(c)
 	return t
 }

treemock_test.go

@@ -71,7 +71,6 @@ type FuncType string
 const (
 	FuncWatch            = FuncType("Watch")
 	FuncUnwatch          = FuncType("Unwatch")
-	FuncDispatch         = FuncType("Dispatch")
 	FuncRewatch          = FuncType("Rewatch")
 	FuncRecursiveWatch   = FuncType("RecursiveWatch")
 	FuncRecursiveUnwatch = FuncType("RecursiveUnwatch")
@@ -184,11 +183,6 @@ type MockedTree struct {
 	C    chan<- EventInfo // event dispatch channel
 }
 
-// Dispatch implements Watcher interface.
-func (mt *MockedTree) Dispatch(c chan<- EventInfo, _ <-chan struct{}) {
-	mt.C = c
-}
-
 // Invoke TODO
 func (mt *MockedTree) Invoke(call Call) error {
 	switch call.F {
@@ -212,9 +206,10 @@ func NewTreeFixture() (tf TreeFixture) {
 	for _, typ := range TreeTypes {
 		// TODO(rjeczalik): Copy FS to allow for modying tree via Create and
 		// Delete events.
-		mt := &MockedTree{}
+		c := make(chan EventInfo, 128)
+		mt := &MockedTree{C: c}
 		tf[typ] = mt
-		mt.Tree = NewTree(SpyWatcher(typ, mt))
+		mt.Tree = NewTree(SpyWatcher(typ, mt), c)
 		mt.Tree.FS = MFS
 	}
 	return

watcher.go

@@ -39,15 +39,6 @@ type Watcher interface {
 	// the existing event set currently registered for the given path, and the
 	// new, requested event set.
 	Rewatch(path string, old, new Event) error
-
-	// Dispatch requests to fan in all events from all the created watchers into c.
-	// It is guaranteed the c is non-nil. All unexpected events are ignored.
-	//
-	// The Dispatch method is called once on package init by the notify runtime.
-	//
-	// The stop channel is closed when the notify runtime is stopped and is no
-	// longer receiving events sent to c.
-	Dispatch(c chan<- EventInfo, stop <-chan struct{})
 }
 
 // RecursiveWatcher is an interface for a Watcher for those OS, which do support

watcher_fsevents.go

@@ -11,13 +11,25 @@ import (
 	"sync/atomic"
 )
 
+const (
+	failure = uint32(FSEventsMustScanSubDirs | FSEventsUserDropped | FSEventsKernelDropped)
+	filter  = uint32(FSEventsCreated | FSEventsRemoved | FSEventsRenamed |
+		FSEventsModified | FSEventsInodeMetaMod)
+)
+
 var (
 	errAlreadyWatched  = errors.New("path is already watched")
 	errNotWatched      = errors.New("path is not being watched")
 	errInvalidEventSet = errors.New("invalid event set provided")
+	errDepth           = errors.New("exceeded allowed iteration count (circular symlink?)")
 )
 
-var errDepth = errors.New("exceeded allowed iteration count (circular symlink?)")
+// FSEvent represents single file event.
+type FSEvent struct {
+	Path  string
+	ID    uint64
+	Flags uint32
+}
 
 // canonical resolves any symlink in the given path and returns it in a clean form.
 // It expects the path to be absolute. It fails to resolve circular symlinks by
@@ -50,12 +62,6 @@ func canonical(p string) (string, error) {
 	return filepath.Clean(p), nil
 }
 
-const (
-	failure = uint32(FSEventsMustScanSubDirs | FSEventsUserDropped | FSEventsKernelDropped)
-	filter  = uint32(FSEventsCreated | FSEventsRemoved | FSEventsRenamed |
-		FSEventsModified | FSEventsInodeMetaMod)
-)
-
 // splitflags separates event flags from single set into slice of flags.
 func splitflags(set uint32) (e []uint32) {
 	for i := uint32(1); set != 0; i, set = i<<1, set>>1 {
@@ -237,15 +243,6 @@ func (fse *fsevents) Rewatch(path string, oldevent, newevent Event) error {
 	return nil
 }
 
-// TODO(rjeczalik): remove
-func (fse *fsevents) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
-	fse.c = c
-	go func() {
-		<-stop
-		fse.Close()
-	}()
-}
-
 // RecursiveWatch implements RecursiveWatcher interface. It fails with non-nil
 // error when setting the watch-point by FSEvents fails or with errAlreadyWatched
 // error when the given path is already watched.

watcher_fsevents_cgo.go

@@ -83,13 +83,6 @@ func gostream(_, ctx unsafe.Pointer, n C.size_t, paths, flags, ids uintptr) {
 	(*(*StreamFunc)(ctx))(ev)
 }
 
-// FSEvent represents single file event.
-type FSEvent struct {
-	Path  string
-	ID    uint64
-	Flags uint32
-}
-
 // StreamFunc is a callback called when stream receives file events.
 type StreamFunc func([]FSEvent)
 

watcher_inotify.go

@@ -308,13 +308,6 @@ func (w *watcher) Unwatch(pathname string) (err error) {
 	return nil
 }
 
-// Dispatch implements notify.Watcher interface.
-//
-// TODO(rjeczalik): remove
-func (w *watcher) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
-	w.c = c
-}
-
 func (w *watcher) Close() (err error) {
 	w.Lock()
 	if fd := atomic.LoadInt32(&w.fd); fd == invalidDescriptor {

watcher_kqueue.go

@@ -379,9 +379,3 @@ func isdir(p string) (bool, error) {
 	}
 	return fi.IsDir(), nil
 }
-
-// Dispatch implements `Watcher` interface.
-//
-// TODO(rjeczalik): remove
-func (k *kqueue) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
-}

watcher_windows.go

@@ -497,13 +497,6 @@ func (w *watcher) unwatch(path string) (err error) {
 	return
 }
 
-// Dispatch implements notify.Watcher interface.
-//
-// TODO(rjeczalik): remove
-func (w *watcher) Dispatch(c chan<- EventInfo, stop <-chan struct{}) {
-	w.c = c
-}
-
 // Close resets the whole watcher object, closes all existing file descriptors,
 // and sends stateCPClose state as completion key to the main watcher's loop.
 func (w *watcher) Close() (err error) {