Repository: fsnotify/fsnotify
Pull request #572 (2023-07-12 18:33:48)

Add NewBufferedWatcher()
Follow-up to #550


Patch:
diff --git a/.circleci/config.yml b/.circleci/config.yml
index ed77df9..beb2cd4 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 d72f6e8..ffc7b99 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/CHANGELOG.md b/CHANGELOG.md
index e82581d..1b67521 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 cd07888..d0daea3 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 c7c5be7..3813e5f 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 3354d2d..185626c 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 52056e1..01d617e 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 0d09ee7..0974061 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 31b9226..24c99cc 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/mkdoc.zsh b/mkdoc.zsh
index 5e3e501..ec01e6a 100755
--- a/mkdoc.zsh
+++ b/mkdoc.zsh
@@ -67,10 +67,10 @@ watcher=$(<<EOF
 // 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.
 EOF
 )
 
@@ -79,6 +79,17 @@ new=$(<<EOF
 EOF
 )
 
+newbuffered=$(<<EOF
+// 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.
+EOF
+)
+
 add=$(<<EOF
 // Add starts monitoring the path for changes.
 //
@@ -236,6 +247,7 @@ set-cmt() {
 
 set-cmt '^type Watcher struct '             $watcher
 set-cmt '^func NewWatcher('                 $new
+set-cmt '^func NewBufferedWatcher('         $newbuffered
 set-cmt '^func (w \*Watcher) Add('          $add
 set-cmt '^func (w \*Watcher) AddWith('      $addwith
 set-cmt '^func (w \*Watcher) Remove('       $remove


Test patch:
diff --git a/.github/workflows/test.yml b/.github/workflows/test.yml
index fe299c3..0397e4e 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/fsnotify_test.go b/fsnotify_test.go
index 2c8c66f..f6e33d8 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 8679e71..c9a7dd7 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)
 	}