add FSEvents-based watcher on macOS

commands/commands.go

@@ -208,9 +208,9 @@ type hugoBuilderCommon struct {
 	baseURL     string
 	environment string
 
-	buildWatch bool
-	poll       string
-	clock      string
+	buildWatch     bool
+	poll           string
+	clock          string
 
 	gc bool
 

commands/hugo.go

@@ -835,10 +835,6 @@ func (c *commandeer) fullRebuild(changeType string) {
 
 // newWatcher creates a new watcher to watch filesystem events.
 func (c *commandeer) newWatcher(pollIntervalStr string, dirList ...string) (*watcher.Batcher, error) {
-	if runtime.GOOS == "darwin" {
-		tweakLimit()
-	}
-
 	staticSyncer, err := newStaticSyncer(c)
 	if err != nil {
 		return nil, err

go.mod

@@ -94,6 +94,7 @@ require (
 	github.com/aws/smithy-go v1.8.0 // indirect
 	github.com/cpuguy83/go-md2man/v2 v2.0.2 // indirect
 	github.com/dlclark/regexp2 v1.7.0 // indirect
+	github.com/fsnotify/fsevents v0.1.1 // indirect
 	github.com/go-openapi/jsonpointer v0.19.5 // indirect
 	github.com/go-openapi/swag v0.19.5 // indirect
 	github.com/golang-jwt/jwt/v4 v4.0.0 // indirect

go.sum

@@ -249,6 +249,8 @@ github.com/frankban/quicktest v1.13.0/go.mod h1:qLE0fzW0VuyUAJgPU19zByoIr0HtCHN/
 github.com/frankban/quicktest v1.14.2/go.mod h1:mgiwOwqx65TmIk1wJ6Q7wvnVMocbUorkibMOrVTHZps=
 github.com/frankban/quicktest v1.14.4 h1:g2rn0vABPOOXmZUj+vbmUp0lPoXEMuhTpIluN0XL9UY=
 github.com/frankban/quicktest v1.14.4/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/fsnotify/fsevents v0.1.1 h1:/125uxJvvoSDDBPen6yUZbil8J9ydKZnnl3TWWmvnkw=
+github.com/fsnotify/fsevents v0.1.1/go.mod h1:+d+hS27T6k5J8CRaPLKFgwKYcpS7GwW3Ule9+SC2ZRc=
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
 github.com/fsnotify/fsnotify v1.5.4/go.mod h1:OVB6XrOHzAwXMpEM7uPOzcehqUV2UqJxmVXmkdnm1bU=
 github.com/fsnotify/fsnotify v1.6.0 h1:n+5WquG0fcWoWp6xPWfHdbskMCQaFnG6PfBrh1Ky4HY=

watcher/filenotify/eventwatcher_darwin.go

@@ -0,0 +1,306 @@
+// Copyright 2022 The Hugo Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+//go:build darwin && cgo
+
+package filenotify
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/fsnotify/fsevents"
+	"github.com/fsnotify/fsnotify"
+)
+
+var (
+	errFSEventsWatcherClosed              = errors.New("fsEventsWatcher is closed")
+	errFSEventsWatcherStreamNotRegistered = errors.New("stream not registered")
+)
+
+type eventStream struct {
+	*fsevents.EventStream
+	isDir    bool
+	watcher  *fsEventsWatcher
+	basePath string
+	removed  chan bool
+}
+
+type fsEventsWatcher struct {
+	streams map[string]*eventStream
+	events  chan fsnotify.Event
+	errors  chan error
+	mu      sync.Mutex
+	done    chan bool
+}
+
+func (w *fsEventsWatcher) Events() <-chan fsnotify.Event {
+	select {
+	case <-w.done:
+		return nil
+	default:
+		return w.events
+	}
+}
+
+func (w *fsEventsWatcher) Errors() <-chan error {
+	return w.errors
+}
+
+func (w *fsEventsWatcher) Add(path string) error {
+	select {
+	case <-w.done:
+		return errFSEventsWatcherClosed
+	default:
+	}
+
+	abs, err := filepath.Abs(path)
+	if err != nil {
+		return err
+	}
+
+	w.mu.Lock()
+	_, found := w.streams[abs]
+	w.mu.Unlock()
+
+	if found {
+		return fmt.Errorf("already registered: %s", abs)
+	}
+
+	if !w.hasParentEventStreamPath(abs) {
+		if err := w.add(abs); err != nil {
+			return err
+		}
+	}
+
+	if childPaths := w.getChildEventStreamPaths(abs); len(childPaths) > 0 {
+		if err := w.removePaths(childPaths); err != nil {
+			return err
+		}
+	}
+	// https://github.com/fsnotify/fsevents/issues/48
+	if len(w.streams) > 4096 {
+		return fmt.Errorf("too many fsevent streams: %d\n", len(w.streams))
+	}
+
+	return nil
+}
+
+func (w *fsEventsWatcher) add(path string) error {
+	dev, err := fsevents.DeviceForPath(path)
+	if err != nil {
+		return err
+	}
+	fi, err := os.Stat(path)
+	if err != nil {
+		return err
+	}
+	// Symlinked-path like "/temp" cannot be watched
+	evaled, err := filepath.EvalSymlinks(path)
+	if err != nil {
+		return err
+	}
+
+	isDir := fi.IsDir()
+	es := &fsevents.EventStream{
+		Paths:   []string{evaled},
+		Latency: 10 * time.Millisecond,
+		Device:  dev,
+		Flags:   fsevents.FileEvents | fsevents.WatchRoot,
+	}
+	stream := &eventStream{
+		es,
+		isDir,
+		w,
+		path,
+		make(chan bool),
+	}
+	w.mu.Lock()
+	w.streams[path] = stream
+	w.mu.Unlock()
+	go func(stream *eventStream) {
+		stream.Start()
+		stream.Flush(true)
+		for {
+			select {
+			case <-stream.watcher.done:
+			case <-stream.removed:
+				stream.Flush(true)
+				stream.Stop()
+				return
+			case evs := <-stream.Events:
+				for _, evt := range evs {
+					err := stream.sendEvent(evt)
+					if err != nil {
+						return
+					}
+				}
+			}
+		}
+	}(stream)
+	return nil
+}
+
+func matchEventFlag(t, m fsevents.EventFlags) bool {
+	return t&m == m
+}
+
+func (s *eventStream) convertEventPath(path string) (string, error) {
+	// Symlinks-evaled path
+	path = "/" + path
+
+	evaledBasePath, err := filepath.EvalSymlinks(s.basePath)
+	if err != nil {
+		return "", err
+	}
+
+	rel := path[len(evaledBasePath):]
+
+	return filepath.Join(s.basePath, rel), nil
+}
+
+func (s *eventStream) convertEvent(e fsevents.Event) (fsnotify.Event, error) {
+	name, err := s.convertEventPath(e.Path)
+
+	if err != nil {
+		return fsnotify.Event{}, err
+	}
+
+	ne := fsnotify.Event{
+		Name: name,
+		Op:   0,
+	}
+	if matchEventFlag(e.Flags, fsevents.ItemCreated) {
+		ne.Op = fsnotify.Create
+		return ne, nil
+	}
+	if matchEventFlag(e.Flags, fsevents.ItemRemoved) {
+		ne.Op = fsnotify.Remove
+		return ne, nil
+	}
+	if matchEventFlag(e.Flags, fsevents.ItemRenamed) {
+		ne.Op = fsnotify.Rename
+		return ne, nil
+	}
+	if matchEventFlag(e.Flags, fsevents.ItemModified) {
+		ne.Op = fsnotify.Write
+		return ne, nil
+	}
+
+	return ne, nil
+}
+
+func (s *eventStream) sendEvent(e fsevents.Event) error {
+	w := s.watcher
+	ne, err := s.convertEvent(e)
+	if err != nil {
+		return err
+	}
+	if ne.Op == 0 {
+		return nil
+	}
+	w.events <- ne
+	return nil
+}
+
+func (w *fsEventsWatcher) sendErr(e error) {
+    w.errors <- e
+}
+
+func (w *fsEventsWatcher) hasParentEventStreamPath(path string) bool {
+	for p, s := range w.streams {
+		if s.isDir && strings.HasPrefix(filepath.Dir(path), p) {
+			return true
+		}
+	}
+	return false
+}
+
+func (w *fsEventsWatcher) getChildEventStreamPaths(path string) (children []string) {
+	for p := range w.streams {
+		if strings.HasPrefix(filepath.Dir(p), path) {
+			children = append(children, p)
+		}
+	}
+
+	return
+}
+
+func (w *fsEventsWatcher) Remove(path string) error {
+	abs, err := filepath.Abs(path)
+	if err != nil {
+		return err
+	}
+	return w.remove(abs)
+}
+
+func (w *fsEventsWatcher) removePaths(paths []string) error {
+	for _, p := range paths {
+		if err := w.remove(p); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (w *fsEventsWatcher) remove(path string) error {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+
+	stream, exists := w.streams[path]
+	if !exists {
+		return errFSEventsWatcherStreamNotRegistered
+	}
+	close(stream.removed)
+	delete(w.streams, path)
+	return nil
+}
+
+func (w *fsEventsWatcher) Close() error {
+	select {
+	case <-w.done:
+		return nil
+	default:
+	}
+
+	close(w.done)
+	for path := range w.streams {
+		err := w.remove(path)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// NewFSEventsWatcher returns a fsevents file watcher
+func NewFSEventsWatcher() (FileWatcher, error) {
+	w := &fsEventsWatcher{
+		streams: make(map[string]*eventStream),
+		done:    make(chan bool),
+		events:  make(chan fsnotify.Event),
+		errors:  make(chan error),
+	}
+	return w, nil
+}
+
+// NewEventWatcher returns an FSEvents based file watcher on darwin
+func NewEventWatcher() (FileWatcher, error) {
+	return NewFSEventsWatcher()
+}

watcher/filenotify/eventwatcher_other.go

@@ -0,0 +1,22 @@
+// Copyright 2022 The Hugo Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+
+//go:build !darwin || !cgo
+
+package filenotify
+
+// NewEventWatcher returns an fsnotify based file watcher
+func NewEventWatcher() (FileWatcher, error) {
+	return NewFsNotifyWatcher()
+}

watcher/filenotify/filenotify.go

@@ -38,12 +38,3 @@ func NewPollingWatcher(interval time.Duration) FileWatcher {
 		errors:   make(chan error),
 	}
 }
-
-// NewEventWatcher returns an fs-event based file watcher
-func NewEventWatcher() (FileWatcher, error) {
-	watcher, err := fsnotify.NewWatcher()
-	if err != nil {
-		return nil, err
-	}
-	return &fsNotifyWatcher{watcher}, nil
-}