From f01d91ef9041dad03899f17ee5ff3a5bbff5bee0 Mon Sep 17 00:00:00 2001 From: Martin Tournoij Date: Sat, 21 Oct 2023 21:33:32 +0200 Subject: [PATCH] Add NewBufferedWatcher() (#572) This PR slightly overhauls the NewWatcher() API so that a list of Options functions can be provided. Currently there is only one configured option but there are a few issues asking for more configuration options so this should allow for an extensible method to add configuration options to NewWatcher. The goals for this PR: 1. Do not change the API in any way that would require any refactor by users of the library. 2. Make all options optional and ensure default behavior is unchanged. 3. Provide the option to configure a userland buffer in the event channel. The motivation for this PR: We make extensive use of the fsnotify library but continually run into situations with older RedHat installations where the kernel fsnotify buffer is not large enough to accommodate high load scenarios like log file rotation and the end user cannot modify the kernel parameters. We have been running a fork that created a userland event channel buffer and found that we are able to deal with very large bursts of fsnotify events without being forced to catch errors and recover. The userland channel buffer will never be as fast as the kernel buffer, and in most cases straight won't be used, but this PR gives users of the library the option to add some padding without changing kernel parameters. --------- Co-authored-by: kris --- .circleci/config.yml | 9 +- .cirrus.yml | 3 +- .github/workflows/test.yml | 128 ++++++++++++---------- CHANGELOG.md | 14 ++- backend_fen.go | 21 +++- backend_inotify.go | 21 +++- backend_kqueue.go | 21 +++- backend_other.go | 17 ++- backend_windows.go | 21 +++- fsnotify.go | 7 +- fsnotify_test.go | 214 +++++++++++++++++++++++++------------ helpers_test.go | 29 ++++- mkdoc.zsh | 20 +++- 13 files changed, 360 insertions(+), 165 deletions(-) diff --git a/.circleci/config.yml b/.circleci/config.yml index ed77df92..beb2cd46 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -23,7 +23,8 @@ jobs: command: | uname -a go version - go test -parallel 1 -race ./... + FSNOTIFY_BUFFER=4096 go test -parallel 1 -race ./... + go test -parallel 1 -race ./... # iOS ios: @@ -48,7 +49,8 @@ jobs: export PATH=$PATH:/usr/local/Cellar/go/*/bin uname -a go version - go test -parallel 1 -race ./... + FSNOTIFY_BUFFER=4096 go test -parallel 1 -race ./... + go test -parallel 1 -race ./... # This is just Linux x86_64; also need to get a Go with GOOS=android, but # there aren't any pre-built versions of that on the Go site. Idk, disable for @@ -76,5 +78,6 @@ jobs: # uname -a # export PATH=/usr/local/go/bin:$PATH # go version - # go test -parallel 1 -race ./... + # FSNOTIFY_BUFFER=4096 go test -parallel 1 -race ./... + # go test -parallel 1 -race ./... # diff --git a/.cirrus.yml b/.cirrus.yml index d72f6e8b..ffc7b992 100644 --- a/.cirrus.yml +++ b/.cirrus.yml @@ -9,4 +9,5 @@ freebsd_task: # run tests as user "cirrus" instead of root - pw useradd cirrus -m - chown -R cirrus:cirrus . - - sudo -u cirrus go test -parallel 1 -race ./... + - FSNOTIFY_BUFFER=4096 sudo --preserve-env=FSNOTIFY_BUFFER -u cirrus go test -parallel 1 -race ./... + - sudo --preserve-env=FSNOTIFY_BUFFER -u cirrus go test -parallel 1 -race ./... diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml index fe299c33..0397e4e2 100644 --- a/.github/workflows/test.yml +++ b/.github/workflows/test.yml @@ -6,71 +6,74 @@ on: branches: ['main', 'aix'] jobs: - # Test Windows and Linux with the latest Go version and the oldest we support. - test: + linux: strategy: fail-fast: false matrix: - os: - - ubuntu-latest - - windows-latest - go: - - '1.17' - - '1.21' + os: ['ubuntu-latest'] + go: ['1.17', '1.21'] runs-on: ${{ matrix.os }} steps: - - name: checkout - uses: actions/checkout@v3 - - - name: setup Go - uses: actions/setup-go@v4 + - uses: 'actions/checkout@v3' + - uses: 'actions/setup-go@v4' with: go-version: ${{ matrix.go }} + - name: test + run: | + FSNOTIFY_BUFFER=4096 go test -parallel 1 -race ./... + go test -parallel 1 -race ./... + windows: + strategy: + fail-fast: false + matrix: + os: ['windows-latest'] + go: ['1.17', '1.21'] + runs-on: ${{ matrix.os }} + steps: + - uses: 'actions/checkout@v3' + - uses: 'actions/setup-go@v4' + with: + go-version: ${{ matrix.go }} - name: test run: | go test -parallel 1 -race ./... + set "FSNOTIFY_BUFFER=4096" + go test -parallel 1 -race ./... # Test gccgo - testgcc: - runs-on: ubuntu-22.04 - name: test (ubuntu-22.04, gccgo 12.1) + gcc: + runs-on: 'ubuntu-22.04' + name: 'test (ubuntu-22.04, gccgo 12.1)' steps: - - name: checkout - uses: actions/checkout@v3 - + - uses: 'actions/checkout@v3' - name: test run: | sudo apt-get -y install gccgo-12 - go-12 test -parallel 1 ./... + FSNOTIFY_BUFFER=4096 go-12 test -parallel 1 ./... + go-12 test -parallel 1 ./... # Test only the latest Go version on macOS; we use the macOS builders for BSD # and illumos, and GitHub doesn't allow many of them to run concurrently. If # it works on Windows and Linux with Go 1.17, then it probably does on macOS # too. - testMacOS: + macos: name: test strategy: fail-fast: false matrix: - os: - - macos-11 - - macos-13 - go: - - '1.21' + os: ['macos-11', 'macos-13'] + go: ['1.21'] runs-on: ${{ matrix.os }} steps: - - name: checkout - uses: actions/checkout@v3 - - - name: setup Go - uses: actions/setup-go@v4 + - uses: 'actions/checkout@v3' + - uses: 'actions/setup-go@v4' with: go-version: ${{ matrix.go }} - - name: test run: | - go test -parallel 1 -race ./... + FSNOTIFY_BUFFER=4096 go test -parallel 1 -race ./... + go test -parallel 1 -race ./... # OpenBSD; no -race as the VM doesn't include the comp set. # @@ -79,45 +82,50 @@ jobs: # so should probably look into that first. Go 1.19 is supposed to have a # much faster race detector, so maybe waiting until we have that is # enough. - testOpenBSD: - runs-on: macos-12 - name: test (openbsd, 1.17) + openbsd: + runs-on: 'macos-12' + timeout-minutes: 30 + name: 'test (openbsd, 1.17)' steps: - - uses: actions/checkout@v3 - - name: test (openbsd, 1.17) - id: test - uses: vmactions/openbsd-vm@v0 + - uses: 'actions/checkout@v3' + - name: 'test (openbsd, 1.17)' + id: 'openbsd' + uses: 'vmactions/openbsd-vm@v0' with: prepare: pkg_add go run: | useradd -mG wheel action - su action -c 'go test -parallel 1 ./...' + FSNOTIFY_BUFFER=4096 su action -c 'go test -parallel 1 ./...' + su action -c 'go test -parallel 1 ./...' # NetBSD - testNetBSD: + netbsd: runs-on: macos-12 + timeout-minutes: 30 name: test (netbsd, 1.20) steps: - - uses: actions/checkout@v3 - - name: test (netbsd, 1.20) - id: test - uses: vmactions/netbsd-vm@v0 + - uses: 'actions/checkout@v3' + - name: 'test (netbsd, 1.20)' + id: 'netbsd' + uses: 'vmactions/netbsd-vm@v0' with: prepare: pkg_add go # TODO: no -race for the same reason as OpenBSD (the timing; it does run). run: | useradd -mG wheel action - su action -c 'go120 test -parallel 1 ./...' + FSNOTIFY_BUFFER=4096 su action -c 'go120 test -parallel 1 ./...' + su action -c 'go120 test -parallel 1 ./...' # illumos - testillumos: + illumos: runs-on: macos-12 + timeout-minutes: 30 name: test (illumos, 1.19) steps: - - uses: actions/checkout@v3 - - name: test (illumos, 1.19) - id: test - uses: papertigers/illumos-vm@r38 + - uses: 'actions/checkout@v3' + - name: 'test (illumos, 1.19)' + id: 'illumos' + uses: 'papertigers/illumos-vm@r38' with: prepare: | pkg install go-119 @@ -125,11 +133,13 @@ jobs: useradd action export GOCACHE=/tmp/go-cache export GOPATH=/tmp/go-path - su action -c '/opt/ooce/go-1.19/bin/go test -parallel 1 ./...' + FSNOTIFY_BUFFER=4096 su action -c '/opt/ooce/go-1.19/bin/go test -parallel 1 ./...' + su action -c '/opt/ooce/go-1.19/bin/go test -parallel 1 ./...' # Older Debian 6, for old Linux kernels. - testDebian6: + debian6: runs-on: macos-12 + timeout-minutes: 30 name: test (debian6, 1.19) strategy: fail-fast: false @@ -149,16 +159,18 @@ jobs: with: go-version: '1.19' - - name: test (debian6, 1.19) - id: test + - name: 'test (debian6, 1.19)' + id: 'debian6' run: | cp -f .github/workflows/Vagrantfile.debian6 Vagrantfile export GOOS=linux export GOARCH=amd64 for p in $(go list ./...); do - go test -c -o ${p//\//-}.test $p + FSNOTIFY_BUFFER=4096 go test -c -o ${p//\//-}.test $p + go test -c -o ${p//\//-}.test $p done vagrant up for t in *.test; do - vagrant ssh -c "/vagrant/$t -test.parallel 1" + FSNOTIFY_BUFFER=4096 vagrant ssh -c "/vagrant/$t -test.parallel 1" + vagrant ssh -c "/vagrant/$t -test.parallel 1" done diff --git a/CHANGELOG.md b/CHANGELOG.md index e82581d7..1b67521d 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,12 +8,17 @@ This version of fsnotify needs Go 1.17. - illumos: add FEN backend to support illumos and Solaris. ([#371]) +- all: add `NewBufferedWatcher()` to use a buffered channel, which can be useful + in cases where you can't control the kernel buffer and receive a large number + of events in bursts. ([#550], [#572]) + - all: add `AddWith()`, which is identical to `Add()` but allows passing options. ([#521]) -- windows: allow setting the buffer size with `fsnotify.WithBufferSize()`; the - default of 64K is the highest value that works on all platforms and is enough - for most purposes, but in some cases a highest buffer is needed. ([#521]) +- windows: allow setting the ReadDirectoryChangesW() buffer size with + `fsnotify.WithBufferSize()`; the default of 64K is the highest value that + works on all platforms and is enough for most purposes, but in some cases a + highest buffer is needed. ([#521]) ### Changes and fixes @@ -57,7 +62,6 @@ This version of fsnotify needs Go 1.17. Google AppEngine forbids usage of the unsafe package so the inotify backend won't compile there. - [#371]: https://github.com/fsnotify/fsnotify/pull/371 [#516]: https://github.com/fsnotify/fsnotify/pull/516 [#518]: https://github.com/fsnotify/fsnotify/pull/518 @@ -67,6 +71,8 @@ This version of fsnotify needs Go 1.17. [#526]: https://github.com/fsnotify/fsnotify/pull/526 [#528]: https://github.com/fsnotify/fsnotify/pull/528 [#537]: https://github.com/fsnotify/fsnotify/pull/537 +[#550]: https://github.com/fsnotify/fsnotify/pull/550 +[#572]: https://github.com/fsnotify/fsnotify/pull/572 1.6.0 - 2022-10-13 ------------------- diff --git a/backend_fen.go b/backend_fen.go index cd07888a..d0daea3d 100644 --- a/backend_fen.go +++ b/backend_fen.go @@ -77,10 +77,10 @@ import ( // Sometimes it will send events for all times, sometimes it will send no // events, and often only for some files. // -// The default buffer size is 64K, which is the largest value that is guaranteed -// to work with SMB filesystems. If you have many events in quick succession -// this may not be enough, and you will have to use [WithBufferSize] to increase -// the value. +// The default ReadDirectoryChangesW() buffer size is 64K, which is the largest +// value that is guaranteed to work with SMB filesystems. If you have many +// events in quick succession this may not be enough, and you will have to use +// [WithBufferSize] to increase the value. type Watcher struct { // Events sends the filesystem change events. // @@ -139,8 +139,19 @@ type Watcher struct { // NewWatcher creates a new Watcher. func NewWatcher() (*Watcher, error) { + return NewBufferedWatcher(0) +} + +// NewBufferedWatcher creates a new Watcher with a buffered [Events] channel. +// +// The main use-case for this is situations with a very large number of events +// where the kernel buffer size can't be increased (e.g. due to lack of +// permissions). An unbuffered Watcher will perform better for almost all use +// cases, and whenever possible you will be better off increasing the kernel +// buffers instead of adding a large userspace buffer. +func NewBufferedWatcher(sz uint) (*Watcher, error) { w := &Watcher{ - Events: make(chan Event), + Events: make(chan Event, sz), Errors: make(chan error), dirs: make(map[string]struct{}), watches: make(map[string]struct{}), diff --git a/backend_inotify.go b/backend_inotify.go index c7c5be7d..3813e5f7 100644 --- a/backend_inotify.go +++ b/backend_inotify.go @@ -80,10 +80,10 @@ import ( // Sometimes it will send events for all times, sometimes it will send no // events, and often only for some files. // -// The default buffer size is 64K, which is the largest value that is guaranteed -// to work with SMB filesystems. If you have many events in quick succession -// this may not be enough, and you will have to use [WithBufferSize] to increase -// the value. +// The default ReadDirectoryChangesW() buffer size is 64K, which is the largest +// value that is guaranteed to work with SMB filesystems. If you have many +// events in quick succession this may not be enough, and you will have to use +// [WithBufferSize] to increase the value. type Watcher struct { // Events sends the filesystem change events. // @@ -238,6 +238,17 @@ func (w *watches) updatePath(path string, f func(*watch) (*watch, error)) error // NewWatcher creates a new Watcher. func NewWatcher() (*Watcher, error) { + return NewBufferedWatcher(0) +} + +// NewBufferedWatcher creates a new Watcher with a buffered [Events] channel. +// +// The main use-case for this is situations with a very large number of events +// where the kernel buffer size can't be increased (e.g. due to lack of +// permissions). An unbuffered Watcher will perform better for almost all use +// cases, and whenever possible you will be better off increasing the kernel +// buffers instead of adding a large userspace buffer. +func NewBufferedWatcher(sz uint) (*Watcher, error) { // Need to set nonblocking mode for SetDeadline to work, otherwise blocking // I/O operations won't terminate on close. fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK) @@ -249,7 +260,7 @@ func NewWatcher() (*Watcher, error) { fd: fd, inotifyFile: os.NewFile(uintptr(fd), ""), watches: newWatches(), - Events: make(chan Event), + Events: make(chan Event, sz), Errors: make(chan error), done: make(chan struct{}), doneResp: make(chan struct{}), diff --git a/backend_kqueue.go b/backend_kqueue.go index 3354d2d8..185626c4 100644 --- a/backend_kqueue.go +++ b/backend_kqueue.go @@ -77,10 +77,10 @@ import ( // Sometimes it will send events for all times, sometimes it will send no // events, and often only for some files. // -// The default buffer size is 64K, which is the largest value that is guaranteed -// to work with SMB filesystems. If you have many events in quick succession -// this may not be enough, and you will have to use [WithBufferSize] to increase -// the value. +// The default ReadDirectoryChangesW() buffer size is 64K, which is the largest +// value that is guaranteed to work with SMB filesystems. If you have many +// events in quick succession this may not be enough, and you will have to use +// [WithBufferSize] to increase the value. type Watcher struct { // Events sends the filesystem change events. // @@ -150,6 +150,17 @@ type pathInfo struct { // NewWatcher creates a new Watcher. func NewWatcher() (*Watcher, error) { + return NewBufferedWatcher(0) +} + +// NewBufferedWatcher creates a new Watcher with a buffered [Events] channel. +// +// The main use-case for this is situations with a very large number of events +// where the kernel buffer size can't be increased (e.g. due to lack of +// permissions). An unbuffered Watcher will perform better for almost all use +// cases, and whenever possible you will be better off increasing the kernel +// buffers instead of adding a large userspace buffer. +func NewBufferedWatcher(sz uint) (*Watcher, error) { kq, closepipe, err := newKqueue() if err != nil { return nil, err @@ -164,7 +175,7 @@ func NewWatcher() (*Watcher, error) { paths: make(map[int]pathInfo), fileExists: make(map[string]struct{}), userWatches: make(map[string]struct{}), - Events: make(chan Event), + Events: make(chan Event, sz), Errors: make(chan error), done: make(chan struct{}), } diff --git a/backend_other.go b/backend_other.go index 52056e1b..01d617eb 100644 --- a/backend_other.go +++ b/backend_other.go @@ -69,10 +69,10 @@ import "errors" // Sometimes it will send events for all times, sometimes it will send no // events, and often only for some files. // -// The default buffer size is 64K, which is the largest value that is guaranteed -// to work with SMB filesystems. If you have many events in quick succession -// this may not be enough, and you will have to use [WithBufferSize] to increase -// the value. +// The default ReadDirectoryChangesW() buffer size is 64K, which is the largest +// value that is guaranteed to work with SMB filesystems. If you have many +// events in quick succession this may not be enough, and you will have to use +// [WithBufferSize] to increase the value. type Watcher struct { // Events sends the filesystem change events. // @@ -128,6 +128,15 @@ func NewWatcher() (*Watcher, error) { return nil, errors.New("fsnotify not supported on the current platform") } +// NewBufferedWatcher creates a new Watcher with a buffered [Events] channel. +// +// The main use-case for this is situations with a very large number of events +// where the kernel buffer size can't be increased (e.g. due to lack of +// permissions). An unbuffered Watcher will perform better for almost all use +// cases, and whenever possible you will be better off increasing the kernel +// buffers instead of adding a large userspace buffer. +func NewBufferedWatcher(sz uint) (*Watcher, error) { return NewWatcher() } + // Close removes all watches and closes the events channel. func (w *Watcher) Close() error { return nil } diff --git a/backend_windows.go b/backend_windows.go index 0d09ee77..09740612 100644 --- a/backend_windows.go +++ b/backend_windows.go @@ -85,10 +85,10 @@ import ( // Sometimes it will send events for all times, sometimes it will send no // events, and often only for some files. // -// The default buffer size is 64K, which is the largest value that is guaranteed -// to work with SMB filesystems. If you have many events in quick succession -// this may not be enough, and you will have to use [WithBufferSize] to increase -// the value. +// The default ReadDirectoryChangesW() buffer size is 64K, which is the largest +// value that is guaranteed to work with SMB filesystems. If you have many +// events in quick succession this may not be enough, and you will have to use +// [WithBufferSize] to increase the value. type Watcher struct { // Events sends the filesystem change events. // @@ -149,6 +149,17 @@ type Watcher struct { // NewWatcher creates a new Watcher. func NewWatcher() (*Watcher, error) { + return NewBufferedWatcher(50) +} + +// NewBufferedWatcher creates a new Watcher with a buffered [Events] channel. +// +// The main use-case for this is situations with a very large number of events +// where the kernel buffer size can't be increased (e.g. due to lack of +// permissions). An unbuffered Watcher will perform better for almost all use +// cases, and whenever possible you will be better off increasing the kernel +// buffers instead of adding a large userspace buffer. +func NewBufferedWatcher(sz uint) (*Watcher, error) { port, err := windows.CreateIoCompletionPort(windows.InvalidHandle, 0, 0, 0) if err != nil { return nil, os.NewSyscallError("CreateIoCompletionPort", err) @@ -157,7 +168,7 @@ func NewWatcher() (*Watcher, error) { port: port, watches: make(watchMap), input: make(chan *input, 1), - Events: make(chan Event, 50), + Events: make(chan Event, sz), Errors: make(chan error), quit: make(chan chan<- error, 1), } diff --git a/fsnotify.go b/fsnotify.go index 31b9226a..24c99cc4 100644 --- a/fsnotify.go +++ b/fsnotify.go @@ -122,13 +122,16 @@ func getOptions(opts ...addOpt) withOpts { return with } -// WithBufferSize sets the buffer size for the Windows backend. This is a no-op -// for other backends. +// WithBufferSize sets the [ReadDirectoryChangesW] buffer size. +// +// This only has effect on Windows systems, and is a no-op for other backends. // // The default value is 64K (65536 bytes) which is the highest value that works // on all filesystems and should be enough for most applications, but if you // have a large burst of events it may not be enough. You can increase it if // you're hitting "queue or buffer overflow" errors ([ErrEventOverflow]). +// +// [ReadDirectoryChangesW]: https://learn.microsoft.com/en-gb/windows/win32/api/winbase/nf-winbase-readdirectorychangesw func WithBufferSize(bytes int) addOpt { return func(opt *withOpts) { opt.bufsize = bytes } } diff --git a/fsnotify_test.go b/fsnotify_test.go index 2c8c66f7..f6e33d88 100644 --- a/fsnotify_test.go +++ b/fsnotify_test.go @@ -948,20 +948,27 @@ func TestClose(t *testing.T) { time.Sleep(50 * time.Millisecond) } - select { - default: - t.Fatal("blocking on Events") - case _, ok := <-w.Events: - if ok { - t.Fatal("Events not closed") + tim := time.NewTimer(50 * time.Millisecond) + loop: + for { + select { + default: + t.Fatal("blocking on Events") + case <-tim.C: + t.Fatalf("Events not closed") + case _, ok := <-w.Events: + if !ok { + break loop + } } } + select { default: t.Fatal("blocking on Errors") - case _, ok := <-w.Errors: + case err, ok := <-w.Errors: if ok { - t.Fatal("Errors not closed") + t.Fatalf("Errors not closed; read:\n\t%s", err) } } } @@ -1001,6 +1008,7 @@ func TestClose(t *testing.T) { touch(t, tmp, "file") rm(t, tmp, "file") + eventSeparator() if err := w.Close(); err != nil { t.Fatal(err) } @@ -1062,21 +1070,40 @@ func TestClose(t *testing.T) { // a good reproducible test for this, but running it 150 times seems to // reproduce it in ~75% of cases and isn't too slow (~0.06s on my system). t.Run("double close", func(t *testing.T) { - t.Parallel() - - for i := 0; i < 150; i++ { - w, err := NewWatcher() - if err != nil { - if strings.Contains(err.Error(), "too many") { // syscall.EMFILE - time.Sleep(100 * time.Millisecond) - continue + t.Run("default", func(t *testing.T) { + t.Parallel() + + for i := 0; i < 150; i++ { + w, err := NewWatcher() + if err != nil { + if strings.Contains(err.Error(), "too many") { // syscall.EMFILE + time.Sleep(100 * time.Millisecond) + continue + } + t.Fatal(err) } - t.Fatal(err) + go w.Close() + go w.Close() + go w.Close() } - go w.Close() - go w.Close() - go w.Close() - } + }) + t.Run("buffered=4096", func(t *testing.T) { + t.Parallel() + + for i := 0; i < 150; i++ { + w, err := NewBufferedWatcher(4096) + if err != nil { + if strings.Contains(err.Error(), "too many") { // syscall.EMFILE + time.Sleep(100 * time.Millisecond) + continue + } + t.Fatal(err) + } + go w.Close() + go w.Close() + go w.Close() + } + }) }) t.Run("closes channels after read", func(t *testing.T) { @@ -1591,71 +1618,122 @@ func TestOpHas(t *testing.T) { } func BenchmarkWatch(b *testing.B) { - w, err := NewWatcher() - if err != nil { - b.Fatal(err) - } - - tmp := b.TempDir() - file := join(tmp, "file") - err = w.Add(tmp) - if err != nil { - b.Fatal(err) - } + do := func(b *testing.B, w *Watcher) { + tmp := b.TempDir() + file := join(tmp, "file") + err := w.Add(tmp) + if err != nil { + b.Fatal(err) + } - var wg sync.WaitGroup - wg.Add(1) - go func() { - for { - select { - case err, ok := <-w.Errors: - if !ok { - wg.Done() - return - } - b.Error(err) - case _, ok := <-w.Events: - if !ok { - wg.Done() - return + var wg sync.WaitGroup + wg.Add(1) + go func() { + for { + select { + case err, ok := <-w.Errors: + if !ok { + wg.Done() + return + } + b.Error(err) + case _, ok := <-w.Events: + if !ok { + wg.Done() + return + } } } + }() + + b.ResetTimer() + for n := 0; n < b.N; n++ { + fp, err := os.Create(file) + if err != nil { + b.Fatal(err) + } + err = fp.Close() + if err != nil { + b.Fatal(err) + } } - }() + err = w.Close() + if err != nil { + b.Fatal(err) + } + wg.Wait() + } - b.ResetTimer() - for n := 0; n < b.N; n++ { - fp, err := os.Create(file) + b.Run("default", func(b *testing.B) { + w, err := NewWatcher() if err != nil { b.Fatal(err) } - err = fp.Close() + do(b, w) + }) + b.Run("buffered=1", func(b *testing.B) { + w, err := NewBufferedWatcher(1) if err != nil { b.Fatal(err) } - } - err = w.Close() - if err != nil { - b.Fatal(err) - } - wg.Wait() + do(b, w) + }) + b.Run("buffered=1024", func(b *testing.B) { + w, err := NewBufferedWatcher(1024) + if err != nil { + b.Fatal(err) + } + do(b, w) + }) + b.Run("buffered=4096", func(b *testing.B) { + w, err := NewBufferedWatcher(4096) + if err != nil { + b.Fatal(err) + } + do(b, w) + }) } func BenchmarkAddRemove(b *testing.B) { - w, err := NewWatcher() - if err != nil { - b.Fatal(err) + do := func(b *testing.B, w *Watcher) { + tmp := b.TempDir() + b.ResetTimer() + for n := 0; n < b.N; n++ { + if err := w.Add(tmp); err != nil { + b.Fatal(err) + } + if err := w.Remove(tmp); err != nil { + b.Fatal(err) + } + } } - tmp := b.TempDir() - - b.ResetTimer() - for n := 0; n < b.N; n++ { - if err := w.Add(tmp); err != nil { + b.Run("default", func(b *testing.B) { + w, err := NewWatcher() + if err != nil { b.Fatal(err) } - if err := w.Remove(tmp); err != nil { + do(b, w) + }) + b.Run("buffered=1", func(b *testing.B) { + w, err := NewBufferedWatcher(1) + if err != nil { b.Fatal(err) } - } + do(b, w) + }) + b.Run("buffered=1024", func(b *testing.B) { + w, err := NewBufferedWatcher(1024) + if err != nil { + b.Fatal(err) + } + do(b, w) + }) + b.Run("buffered=4096", func(b *testing.B) { + w, err := NewBufferedWatcher(4096) + if err != nil { + b.Fatal(err) + } + do(b, w) + }) } diff --git a/helpers_test.go b/helpers_test.go index 8679e71c..c9a7dd78 100644 --- a/helpers_test.go +++ b/helpers_test.go @@ -7,6 +7,7 @@ import ( "path/filepath" "runtime" "sort" + "strconv" "strings" "sync" "testing" @@ -42,10 +43,36 @@ func (tt testCase) run(t *testing.T) { func eventSeparator() { time.Sleep(50 * time.Millisecond) } func waitForEvents() { time.Sleep(500 * time.Millisecond) } +// To test the buffered watcher we run the tests twice in the CI: once as "go +// test" and once with FSNOTIFY_BUFFER set. This is a bit hacky, but saves +// having to refactor a lot of this code. Besides, running the tests in the CI +// more than once isn't a bad thing, since it helps catch flaky tests (should +// probably run it even more). +var testBuffered = func() uint { + s, ok := os.LookupEnv("FSNOTIFY_BUFFER") + if ok { + i, err := strconv.ParseUint(s, 0, 0) + if err != nil { + panic(fmt.Sprintf("FSNOTIFY_BUFFER: %s", err)) + } + return uint(i) + } + return 0 +}() + // newWatcher initializes an fsnotify Watcher instance. func newWatcher(t *testing.T, add ...string) *Watcher { t.Helper() - w, err := NewWatcher() + + var ( + w *Watcher + err error + ) + if testBuffered > 0 { + w, err = NewBufferedWatcher(testBuffered) + } else { + w, err = NewWatcher() + } if err != nil { t.Fatalf("newWatcher: %s", err) } diff --git a/mkdoc.zsh b/mkdoc.zsh index 5e3e501d..ec01e6a5 100755 --- a/mkdoc.zsh +++ b/mkdoc.zsh @@ -67,10 +67,10 @@ watcher=$(<