backend_kqueue.go

  1//go:build freebsd || openbsd || netbsd || dragonfly || darwin
  2
  3package fsnotify
  4
  5import (
  6	"errors"
  7	"fmt"
  8	"os"
  9	"path/filepath"
 10	"runtime"
 11	"sync"
 12	"time"
 13
 14	"github.com/fsnotify/fsnotify/internal"
 15	"golang.org/x/sys/unix"
 16)
 17
 18type kqueue struct {
 19	Events chan Event
 20	Errors chan error
 21
 22	kq        int    // File descriptor (as returned by the kqueue() syscall).
 23	closepipe [2]int // Pipe used for closing kq.
 24	watches   *watches
 25	done      chan struct{}
 26	doneMu    sync.Mutex
 27}
 28
 29type (
 30	watches struct {
 31		mu     sync.RWMutex
 32		wd     map[int]watch               // wd → watch
 33		path   map[string]int              // pathname → wd
 34		byDir  map[string]map[int]struct{} // dirname(path) → wd
 35		seen   map[string]struct{}         // Keep track of if we know this file exists.
 36		byUser map[string]struct{}         // Watches added with Watcher.Add()
 37	}
 38	watch struct {
 39		wd       int
 40		name     string
 41		linkName string // In case of links; name is the target, and this is the link.
 42		isDir    bool
 43		dirFlags uint32
 44	}
 45)
 46
 47func newWatches() *watches {
 48	return &watches{
 49		wd:     make(map[int]watch),
 50		path:   make(map[string]int),
 51		byDir:  make(map[string]map[int]struct{}),
 52		seen:   make(map[string]struct{}),
 53		byUser: make(map[string]struct{}),
 54	}
 55}
 56
 57func (w *watches) listPaths(userOnly bool) []string {
 58	w.mu.RLock()
 59	defer w.mu.RUnlock()
 60
 61	if userOnly {
 62		l := make([]string, 0, len(w.byUser))
 63		for p := range w.byUser {
 64			l = append(l, p)
 65		}
 66		return l
 67	}
 68
 69	l := make([]string, 0, len(w.path))
 70	for p := range w.path {
 71		l = append(l, p)
 72	}
 73	return l
 74}
 75
 76func (w *watches) watchesInDir(path string) []string {
 77	w.mu.RLock()
 78	defer w.mu.RUnlock()
 79
 80	l := make([]string, 0, 4)
 81	for fd := range w.byDir[path] {
 82		info := w.wd[fd]
 83		if _, ok := w.byUser[info.name]; !ok {
 84			l = append(l, info.name)
 85		}
 86	}
 87	return l
 88}
 89
 90// Mark path as added by the user.
 91func (w *watches) addUserWatch(path string) {
 92	w.mu.Lock()
 93	defer w.mu.Unlock()
 94	w.byUser[path] = struct{}{}
 95}
 96
 97func (w *watches) addLink(path string, fd int) {
 98	w.mu.Lock()
 99	defer w.mu.Unlock()
100
101	w.path[path] = fd
102	w.seen[path] = struct{}{}
103}
104
105func (w *watches) add(path, linkPath string, fd int, isDir bool) {
106	w.mu.Lock()
107	defer w.mu.Unlock()
108
109	w.path[path] = fd
110	w.wd[fd] = watch{wd: fd, name: path, linkName: linkPath, isDir: isDir}
111
112	parent := filepath.Dir(path)
113	byDir, ok := w.byDir[parent]
114	if !ok {
115		byDir = make(map[int]struct{}, 1)
116		w.byDir[parent] = byDir
117	}
118	byDir[fd] = struct{}{}
119}
120
121func (w *watches) byWd(fd int) (watch, bool) {
122	w.mu.RLock()
123	defer w.mu.RUnlock()
124	info, ok := w.wd[fd]
125	return info, ok
126}
127
128func (w *watches) byPath(path string) (watch, bool) {
129	w.mu.RLock()
130	defer w.mu.RUnlock()
131	info, ok := w.wd[w.path[path]]
132	return info, ok
133}
134
135func (w *watches) updateDirFlags(path string, flags uint32) {
136	w.mu.Lock()
137	defer w.mu.Unlock()
138
139	fd := w.path[path]
140	info := w.wd[fd]
141	info.dirFlags = flags
142	w.wd[fd] = info
143}
144
145func (w *watches) remove(fd int, path string) bool {
146	w.mu.Lock()
147	defer w.mu.Unlock()
148
149	isDir := w.wd[fd].isDir
150	delete(w.path, path)
151	delete(w.byUser, path)
152
153	parent := filepath.Dir(path)
154	delete(w.byDir[parent], fd)
155
156	if len(w.byDir[parent]) == 0 {
157		delete(w.byDir, parent)
158	}
159
160	delete(w.wd, fd)
161	delete(w.seen, path)
162	return isDir
163}
164
165func (w *watches) markSeen(path string, exists bool) {
166	w.mu.Lock()
167	defer w.mu.Unlock()
168	if exists {
169		w.seen[path] = struct{}{}
170	} else {
171		delete(w.seen, path)
172	}
173}
174
175func (w *watches) seenBefore(path string) bool {
176	w.mu.RLock()
177	defer w.mu.RUnlock()
178	_, ok := w.seen[path]
179	return ok
180}
181
182func newBackend(ev chan Event, errs chan error) (backend, error) {
183	return newBufferedBackend(0, ev, errs)
184}
185
186func newBufferedBackend(sz uint, ev chan Event, errs chan error) (backend, error) {
187	kq, closepipe, err := newKqueue()
188	if err != nil {
189		return nil, err
190	}
191
192	w := &kqueue{
193		Events:    ev,
194		Errors:    errs,
195		kq:        kq,
196		closepipe: closepipe,
197		done:      make(chan struct{}),
198		watches:   newWatches(),
199	}
200
201	go w.readEvents()
202	return w, nil
203}
204
205// newKqueue creates a new kernel event queue and returns a descriptor.
206//
207// This registers a new event on closepipe, which will trigger an event when
208// it's closed. This way we can use kevent() without timeout/polling; without
209// the closepipe, it would block forever and we wouldn't be able to stop it at
210// all.
211func newKqueue() (kq int, closepipe [2]int, err error) {
212	kq, err = unix.Kqueue()
213	if kq == -1 {
214		return kq, closepipe, err
215	}
216
217	// Register the close pipe.
218	err = unix.Pipe(closepipe[:])
219	if err != nil {
220		unix.Close(kq)
221		return kq, closepipe, err
222	}
223	unix.CloseOnExec(closepipe[0])
224	unix.CloseOnExec(closepipe[1])
225
226	// Register changes to listen on the closepipe.
227	changes := make([]unix.Kevent_t, 1)
228	// SetKevent converts int to the platform-specific types.
229	unix.SetKevent(&changes[0], closepipe[0], unix.EVFILT_READ,
230		unix.EV_ADD|unix.EV_ENABLE|unix.EV_ONESHOT)
231
232	ok, err := unix.Kevent(kq, changes, nil, nil)
233	if ok == -1 {
234		unix.Close(kq)
235		unix.Close(closepipe[0])
236		unix.Close(closepipe[1])
237		return kq, closepipe, err
238	}
239	return kq, closepipe, nil
240}
241
242// Returns true if the event was sent, or false if watcher is closed.
243func (w *kqueue) sendEvent(e Event) bool {
244	select {
245	case <-w.done:
246		return false
247	case w.Events <- e:
248		return true
249	}
250}
251
252// Returns true if the error was sent, or false if watcher is closed.
253func (w *kqueue) sendError(err error) bool {
254	if err == nil {
255		return true
256	}
257	select {
258	case <-w.done:
259		return false
260	case w.Errors <- err:
261		return true
262	}
263}
264
265func (w *kqueue) isClosed() bool {
266	select {
267	case <-w.done:
268		return true
269	default:
270		return false
271	}
272}
273
274func (w *kqueue) Close() error {
275	w.doneMu.Lock()
276	if w.isClosed() {
277		w.doneMu.Unlock()
278		return nil
279	}
280	close(w.done)
281	w.doneMu.Unlock()
282
283	pathsToRemove := w.watches.listPaths(false)
284	for _, name := range pathsToRemove {
285		w.Remove(name)
286	}
287
288	// Send "quit" message to the reader goroutine.
289	unix.Close(w.closepipe[1])
290	return nil
291}
292
293func (w *kqueue) Add(name string) error { return w.AddWith(name) }
294
295func (w *kqueue) AddWith(name string, opts ...addOpt) error {
296	if debug {
297		fmt.Fprintf(os.Stderr, "FSNOTIFY_DEBUG: %s  AddWith(%q)\n",
298			time.Now().Format("15:04:05.000000000"), name)
299	}
300
301	with := getOptions(opts...)
302	if !w.xSupports(with.op) {
303		return fmt.Errorf("%w: %s", xErrUnsupported, with.op)
304	}
305
306	_, err := w.addWatch(name, noteAllEvents)
307	if err != nil {
308		return err
309	}
310	w.watches.addUserWatch(name)
311	return nil
312}
313
314func (w *kqueue) Remove(name string) error {
315	if debug {
316		fmt.Fprintf(os.Stderr, "FSNOTIFY_DEBUG: %s  Remove(%q)\n",
317			time.Now().Format("15:04:05.000000000"), name)
318	}
319	return w.remove(name, true)
320}
321
322func (w *kqueue) remove(name string, unwatchFiles bool) error {
323	if w.isClosed() {
324		return nil
325	}
326
327	name = filepath.Clean(name)
328	info, ok := w.watches.byPath(name)
329	if !ok {
330		return fmt.Errorf("%w: %s", ErrNonExistentWatch, name)
331	}
332
333	err := w.register([]int{info.wd}, unix.EV_DELETE, 0)
334	if err != nil {
335		return err
336	}
337
338	unix.Close(info.wd)
339
340	isDir := w.watches.remove(info.wd, name)
341
342	// Find all watched paths that are in this directory that are not external.
343	if unwatchFiles && isDir {
344		pathsToRemove := w.watches.watchesInDir(name)
345		for _, name := range pathsToRemove {
346			// Since these are internal, not much sense in propagating error to
347			// the user, as that will just confuse them with an error about a
348			// path they did not explicitly watch themselves.
349			w.Remove(name)
350		}
351	}
352	return nil
353}
354
355func (w *kqueue) WatchList() []string {
356	if w.isClosed() {
357		return nil
358	}
359	return w.watches.listPaths(true)
360}
361
362// Watch all events (except NOTE_EXTEND, NOTE_LINK, NOTE_REVOKE)
363const noteAllEvents = unix.NOTE_DELETE | unix.NOTE_WRITE | unix.NOTE_ATTRIB | unix.NOTE_RENAME
364
365// addWatch adds name to the watched file set; the flags are interpreted as
366// described in kevent(2).
367//
368// Returns the real path to the file which was added, with symlinks resolved.
369func (w *kqueue) addWatch(name string, flags uint32) (string, error) {
370	if w.isClosed() {
371		return "", ErrClosed
372	}
373
374	name = filepath.Clean(name)
375
376	info, alreadyWatching := w.watches.byPath(name)
377	if !alreadyWatching {
378		fi, err := os.Lstat(name)
379		if err != nil {
380			return "", err
381		}
382
383		// Don't watch sockets or named pipes.
384		if (fi.Mode()&os.ModeSocket == os.ModeSocket) || (fi.Mode()&os.ModeNamedPipe == os.ModeNamedPipe) {
385			return "", nil
386		}
387
388		// Follow symlinks.
389		if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
390			link, err := os.Readlink(name)
391			if err != nil {
392				// Return nil because Linux can add unresolvable symlinks to the
393				// watch list without problems, so maintain consistency with
394				// that. There will be no file events for broken symlinks.
395				// TODO: more specific check; returns os.PathError; ENOENT?
396				return "", nil
397			}
398
399			_, alreadyWatching = w.watches.byPath(link)
400			if alreadyWatching {
401				// Add to watches so we don't get spurious Create events later
402				// on when we diff the directories.
403				w.watches.addLink(name, 0)
404				return link, nil
405			}
406
407			info.linkName = name
408			name = link
409			fi, err = os.Lstat(name)
410			if err != nil {
411				return "", nil
412			}
413		}
414
415		// Retry on EINTR; open() can return EINTR in practice on macOS.
416		// See #354, and Go issues 11180 and 39237.
417		for {
418			info.wd, err = unix.Open(name, openMode, 0)
419			if err == nil {
420				break
421			}
422			if errors.Is(err, unix.EINTR) {
423				continue
424			}
425
426			return "", err
427		}
428
429		info.isDir = fi.IsDir()
430	}
431
432	err := w.register([]int{info.wd}, unix.EV_ADD|unix.EV_CLEAR|unix.EV_ENABLE, flags)
433	if err != nil {
434		unix.Close(info.wd)
435		return "", err
436	}
437
438	if !alreadyWatching {
439		w.watches.add(name, info.linkName, info.wd, info.isDir)
440	}
441
442	// Watch the directory if it has not been watched before, or if it was
443	// watched before, but perhaps only a NOTE_DELETE (watchDirectoryFiles)
444	if info.isDir {
445		watchDir := (flags&unix.NOTE_WRITE) == unix.NOTE_WRITE &&
446			(!alreadyWatching || (info.dirFlags&unix.NOTE_WRITE) != unix.NOTE_WRITE)
447		w.watches.updateDirFlags(name, flags)
448
449		if watchDir {
450			if err := w.watchDirectoryFiles(name); err != nil {
451				return "", err
452			}
453		}
454	}
455	return name, nil
456}
457
458// readEvents reads from kqueue and converts the received kevents into
459// Event values that it sends down the Events channel.
460func (w *kqueue) readEvents() {
461	defer func() {
462		close(w.Events)
463		close(w.Errors)
464		_ = unix.Close(w.kq)
465		unix.Close(w.closepipe[0])
466	}()
467
468	eventBuffer := make([]unix.Kevent_t, 10)
469	for {
470		kevents, err := w.read(eventBuffer)
471		// EINTR is okay, the syscall was interrupted before timeout expired.
472		if err != nil && err != unix.EINTR {
473			if !w.sendError(fmt.Errorf("fsnotify.readEvents: %w", err)) {
474				return
475			}
476		}
477
478		for _, kevent := range kevents {
479			var (
480				wd   = int(kevent.Ident)
481				mask = uint32(kevent.Fflags)
482			)
483
484			// Shut down the loop when the pipe is closed, but only after all
485			// other events have been processed.
486			if wd == w.closepipe[0] {
487				return
488			}
489
490			path, ok := w.watches.byWd(wd)
491			if debug {
492				internal.Debug(path.name, &kevent)
493			}
494
495			// On macOS it seems that sometimes an event with Ident=0 is
496			// delivered, and no other flags/information beyond that, even
497			// though we never saw such a file descriptor. For example in
498			// TestWatchSymlink/277 (usually at the end, but sometimes sooner):
499			//
500			// fmt.Printf("READ: %2d  %#v\n", kevent.Ident, kevent)
501			// unix.Kevent_t{Ident:0x2a, Filter:-4, Flags:0x25, Fflags:0x2, Data:0, Udata:(*uint8)(nil)}
502			// unix.Kevent_t{Ident:0x0,  Filter:-4, Flags:0x25, Fflags:0x2, Data:0, Udata:(*uint8)(nil)}
503			//
504			// The first is a normal event, the second with Ident 0. No error
505			// flag, no data, no ... nothing.
506			//
507			// I read a bit through bsd/kern_event.c from the xnu source, but I
508			// don't really see an obvious location where this is triggered –
509			// this doesn't seem intentional, but idk...
510			//
511			// Technically fd 0 is a valid descriptor, so only skip it if
512			// there's no path, and if we're on macOS.
513			if !ok && kevent.Ident == 0 && runtime.GOOS == "darwin" {
514				continue
515			}
516
517			event := w.newEvent(path.name, path.linkName, mask)
518
519			if event.Has(Rename) || event.Has(Remove) {
520				w.remove(event.Name, false)
521				w.watches.markSeen(event.Name, false)
522			}
523
524			if path.isDir && event.Has(Write) && !event.Has(Remove) {
525				w.dirChange(event.Name)
526			} else if !w.sendEvent(event) {
527				return
528			}
529
530			if event.Has(Remove) {
531				// Look for a file that may have overwritten this; for example,
532				// mv f1 f2 will delete f2, then create f2.
533				if path.isDir {
534					fileDir := filepath.Clean(event.Name)
535					_, found := w.watches.byPath(fileDir)
536					if found {
537						// TODO: this branch is never triggered in any test.
538						// Added in d6220df (2012).
539						// isDir check added in 8611c35 (2016): https://github.com/fsnotify/fsnotify/pull/111
540						//
541						// I don't really get how this can be triggered either.
542						// And it wasn't triggered in the patch that added it,
543						// either.
544						//
545						// Original also had a comment:
546						//   make sure the directory exists before we watch for
547						//   changes. When we do a recursive watch and perform
548						//   rm -rf, the parent directory might have gone
549						//   missing, ignore the missing directory and let the
550						//   upcoming delete event remove the watch from the
551						//   parent directory.
552						err := w.dirChange(fileDir)
553						if !w.sendError(err) {
554							return
555						}
556					}
557				} else {
558					path := filepath.Clean(event.Name)
559					if fi, err := os.Lstat(path); err == nil {
560						err := w.sendCreateIfNew(path, fi)
561						if !w.sendError(err) {
562							return
563						}
564					}
565				}
566			}
567		}
568	}
569}
570
571// newEvent returns an platform-independent Event based on kqueue Fflags.
572func (w *kqueue) newEvent(name, linkName string, mask uint32) Event {
573	e := Event{Name: name}
574	if linkName != "" {
575		// If the user watched "/path/link" then emit events as "/path/link"
576		// rather than "/path/target".
577		e.Name = linkName
578	}
579
580	if mask&unix.NOTE_DELETE == unix.NOTE_DELETE {
581		e.Op |= Remove
582	}
583	if mask&unix.NOTE_WRITE == unix.NOTE_WRITE {
584		e.Op |= Write
585	}
586	if mask&unix.NOTE_RENAME == unix.NOTE_RENAME {
587		e.Op |= Rename
588	}
589	if mask&unix.NOTE_ATTRIB == unix.NOTE_ATTRIB {
590		e.Op |= Chmod
591	}
592	// No point sending a write and delete event at the same time: if it's gone,
593	// then it's gone.
594	if e.Op.Has(Write) && e.Op.Has(Remove) {
595		e.Op &^= Write
596	}
597	return e
598}
599
600// watchDirectoryFiles to mimic inotify when adding a watch on a directory
601func (w *kqueue) watchDirectoryFiles(dirPath string) error {
602	files, err := os.ReadDir(dirPath)
603	if err != nil {
604		return err
605	}
606
607	for _, f := range files {
608		path := filepath.Join(dirPath, f.Name())
609
610		fi, err := f.Info()
611		if err != nil {
612			return fmt.Errorf("%q: %w", path, err)
613		}
614
615		cleanPath, err := w.internalWatch(path, fi)
616		if err != nil {
617			// No permission to read the file; that's not a problem: just skip.
618			// But do add it to w.fileExists to prevent it from being picked up
619			// as a "new" file later (it still shows up in the directory
620			// listing).
621			switch {
622			case errors.Is(err, unix.EACCES) || errors.Is(err, unix.EPERM):
623				cleanPath = filepath.Clean(path)
624			default:
625				return fmt.Errorf("%q: %w", path, err)
626			}
627		}
628
629		w.watches.markSeen(cleanPath, true)
630	}
631
632	return nil
633}
634
635// Search the directory for new files and send an event for them.
636//
637// This functionality is to have the BSD watcher match the inotify, which sends
638// a create event for files created in a watched directory.
639func (w *kqueue) dirChange(dir string) error {
640	files, err := os.ReadDir(dir)
641	if err != nil {
642		// Directory no longer exists: we can ignore this safely. kqueue will
643		// still give us the correct events.
644		if errors.Is(err, os.ErrNotExist) {
645			return nil
646		}
647		return fmt.Errorf("fsnotify.dirChange: %w", err)
648	}
649
650	for _, f := range files {
651		fi, err := f.Info()
652		if err != nil {
653			return fmt.Errorf("fsnotify.dirChange: %w", err)
654		}
655
656		err = w.sendCreateIfNew(filepath.Join(dir, fi.Name()), fi)
657		if err != nil {
658			// Don't need to send an error if this file isn't readable.
659			if errors.Is(err, unix.EACCES) || errors.Is(err, unix.EPERM) {
660				return nil
661			}
662			return fmt.Errorf("fsnotify.dirChange: %w", err)
663		}
664	}
665	return nil
666}
667
668// Send a create event if the file isn't already being tracked, and start
669// watching this file.
670func (w *kqueue) sendCreateIfNew(path string, fi os.FileInfo) error {
671	if !w.watches.seenBefore(path) {
672		if !w.sendEvent(Event{Name: path, Op: Create}) {
673			return nil
674		}
675	}
676
677	// Like watchDirectoryFiles, but without doing another ReadDir.
678	path, err := w.internalWatch(path, fi)
679	if err != nil {
680		return err
681	}
682	w.watches.markSeen(path, true)
683	return nil
684}
685
686func (w *kqueue) internalWatch(name string, fi os.FileInfo) (string, error) {
687	if fi.IsDir() {
688		// mimic Linux providing delete events for subdirectories, but preserve
689		// the flags used if currently watching subdirectory
690		info, _ := w.watches.byPath(name)
691		return w.addWatch(name, info.dirFlags|unix.NOTE_DELETE|unix.NOTE_RENAME)
692	}
693
694	// watch file to mimic Linux inotify
695	return w.addWatch(name, noteAllEvents)
696}
697
698// Register events with the queue.
699func (w *kqueue) register(fds []int, flags int, fflags uint32) error {
700	changes := make([]unix.Kevent_t, len(fds))
701	for i, fd := range fds {
702		// SetKevent converts int to the platform-specific types.
703		unix.SetKevent(&changes[i], fd, unix.EVFILT_VNODE, flags)
704		changes[i].Fflags = fflags
705	}
706
707	// Register the events.
708	success, err := unix.Kevent(w.kq, changes, nil, nil)
709	if success == -1 {
710		return err
711	}
712	return nil
713}
714
715// read retrieves pending events, or waits until an event occurs.
716func (w *kqueue) read(events []unix.Kevent_t) ([]unix.Kevent_t, error) {
717	n, err := unix.Kevent(w.kq, nil, events, nil)
718	if err != nil {
719		return nil, err
720	}
721	return events[0:n], nil
722}
723
724func (w *kqueue) xSupports(op Op) bool {
725	if runtime.GOOS == "freebsd" {
726		//return true // Supports everything.
727	}
728	if op.Has(xUnportableOpen) || op.Has(xUnportableRead) ||
729		op.Has(xUnportableCloseWrite) || op.Has(xUnportableCloseRead) {
730		return false
731	}
732	return true
733}