聊聊dapr的fswatcher

序

本文主要研究一下dapr的fswatcher

fswatcher

dapr/pkg/fswatcher/fswatcher.go

import (
    "context"
    "strings"
    "time"

    "github.com/fsnotify/fsnotify"
    "github.com/pkg/errors"
)

func Watch(ctx context.Context, dir string, eventCh chan<- struct{}) error {
    watcher, err := fsnotify.NewWatcher()
    if err != nil {
        return errors.Wrap(err, "failed to create watcher")
    }
    defer watcher.Close()

    if err := watcher.Add(dir); err != nil {
        return errors.Wrap(err, "watcher error")
    }

LOOP:
    for {
        select {
        // watch for events
        case event := <-watcher.Events:
            if event.Op == fsnotify.Create || event.Op == fsnotify.Write {
                if strings.Contains(event.Name, dir) {
                    // give time for other updates to occur
                    time.Sleep(time.Second * 1)
                    eventCh <- struct{}{}
                }
            }
        case <-watcher.Errors:
            break LOOP
        case <-ctx.Done():
            break LOOP
        }
    }
    return nil
}

Watch方法使用fsnotify的watcher来监听文件，之后通过for循环进行select，如果监听到fsnotify.Create或者fsnotify.Write的时候判断event.Name是否包含dir，如果包含则sleep一秒然后通知eventCh

Add

github.com/fsnotify/fsnotify@v1.4.9/kqueue.go

// Add starts watching the named file or directory (non-recursively).
func (w *Watcher) Add(name string) error {
    w.mu.Lock()
    w.externalWatches[name] = true
    w.mu.Unlock()
    _, err := w.addWatch(name, noteAllEvents)
    return err
}

Add方法设置externalWatches[name]为true，然后执行addWatch(name, noteAllEvents)

addWatch

github.com/fsnotify/fsnotify@v1.4.9/kqueue.go

// addWatch adds name to the watched file set.
// The flags are interpreted as described in kevent(2).
// Returns the real path to the file which was added, if any, which may be different from the one passed in the case of symlinks.
func (w *Watcher) addWatch(name string, flags uint32) (string, error) {
    var isDir bool
    // Make ./name and name equivalent
    name = filepath.Clean(name)

    w.mu.Lock()
    if w.isClosed {
        w.mu.Unlock()
        return "", errors.New("kevent instance already closed")
    }
    watchfd, alreadyWatching := w.watches[name]
    // We already have a watch, but we can still override flags.
    if alreadyWatching {
        isDir = w.paths[watchfd].isDir
    }
    w.mu.Unlock()

    if !alreadyWatching {
        fi, err := os.Lstat(name)
        if err != nil {
            return "", err
        }

        // Don't watch sockets.
        if fi.Mode()&os.ModeSocket == os.ModeSocket {
            return "", nil
        }

        // Don't watch named pipes.
        if fi.Mode()&os.ModeNamedPipe == os.ModeNamedPipe {
            return "", nil
        }

        // Follow Symlinks
        // Unfortunately, Linux can add bogus symlinks to watch list without
        // issue, and Windows can't do symlinks period (AFAIK). To  maintain
        // consistency, we will act like everything is fine. There will simply
        // be no file events for broken symlinks.
        // Hence the returns of nil on errors.
        if fi.Mode()&os.ModeSymlink == os.ModeSymlink {
            name, err = filepath.EvalSymlinks(name)
            if err != nil {
                return "", nil
            }

            w.mu.Lock()
            _, alreadyWatching = w.watches[name]
            w.mu.Unlock()

            if alreadyWatching {
                return name, nil
            }

            fi, err = os.Lstat(name)
            if err != nil {
                return "", nil
            }
        }

        watchfd, err = unix.Open(name, openMode, 0700)
        if watchfd == -1 {
            return "", err
        }

        isDir = fi.IsDir()
    }

    const registerAdd = unix.EV_ADD | unix.EV_CLEAR | unix.EV_ENABLE
    if err := register(w.kq, []int{watchfd}, registerAdd, flags); err != nil {
        unix.Close(watchfd)
        return "", err
    }

    if !alreadyWatching {
        w.mu.Lock()
        w.watches[name] = watchfd
        w.paths[watchfd] = pathInfo{name: name, isDir: isDir}
        w.mu.Unlock()
    }

    if isDir {
        // Watch the directory if it has not been watched before,
        // or if it was watched before, but perhaps only a NOTE_DELETE (watchDirectoryFiles)
        w.mu.Lock()

        watchDir := (flags&unix.NOTE_WRITE) == unix.NOTE_WRITE &&
            (!alreadyWatching || (w.dirFlags[name]&unix.NOTE_WRITE) != unix.NOTE_WRITE)
        // Store flags so this watch can be updated later
        w.dirFlags[name] = flags
        w.mu.Unlock()

        if watchDir {
            if err := w.watchDirectoryFiles(name); err != nil {
                return "", err
            }
        }
    }
    return name, nil
}

addWatch方法针对尚未watch的执行os.Lstat(name)及unix.Open(name, openMode, 0700)；之后注册registerAdd；另外针对isDir的情况执行watchDirectoryFiles

watchDirectoryFiles

github.com/fsnotify/fsnotify@v1.4.9/kqueue.go

// watchDirectoryFiles to mimic inotify when adding a watch on a directory
func (w *Watcher) watchDirectoryFiles(dirPath string) error {
    // Get all files
    files, err := ioutil.ReadDir(dirPath)
    if err != nil {
        return err
    }

    for _, fileInfo := range files {
        filePath := filepath.Join(dirPath, fileInfo.Name())
        filePath, err = w.internalWatch(filePath, fileInfo)
        if err != nil {
            return err
        }

        w.mu.Lock()
        w.fileExists[filePath] = true
        w.mu.Unlock()
    }

    return nil
}

watchDirectoryFiles遍历files，挨个执行internalWatch

internalWatch

github.com/fsnotify/fsnotify@v1.4.9/kqueue.go

func (w *Watcher) internalWatch(name string, fileInfo os.FileInfo) (string, error) {
    if fileInfo.IsDir() {
        // mimic Linux providing delete events for subdirectories
        // but preserve the flags used if currently watching subdirectory
        w.mu.Lock()
        flags := w.dirFlags[name]
        w.mu.Unlock()

        flags |= unix.NOTE_DELETE | unix.NOTE_RENAME
        return w.addWatch(name, flags)
    }

    // watch file to mimic Linux inotify
    return w.addWatch(name, noteAllEvents)
}

internalWatch针对dir设置的flag为NOTE_DELETE、NOTE_RENAME

小结

dapr的fswatcher使用fsnotify的watcher来监听文件，之后通过for循环进行select，如果监听到fsnotify.Create或者fsnotify.Write的时候判断event.Name是否包含dir，如果包含则sleep一秒然后通知eventCh。

doc

• dapr