Repository: fsnotify/fsnotify
Pull request #434 (2022-02-28 03:08:00)

Replace Use of Kthread-blocking Epoll with Poller Read, Remove Per-Event LStats on Linux #433
NOTE: Please do not open a pull request that adds or changes the public API without giving consideration to all supported operating systems.

### What does this pull request do?
This PR:
1. replaces the use of a raw kthread-blocking epoll call for monitoring inotify events with a blocking Read() call using the Go runtime's poller system. This will allow the individual goroutine to block while waiting for I/O, rather than needing to block an entire kernel thread (see: https://github.com/fsnotify/fsnotify/issues/240)
2. Removes Linux per-event uses of LStat to check that the file still exists before passing the event to the user (see: https://github.com/fsnotify/fsnotify/issues/404) 

### Change #1
#### Where should the reviewer start?
The changes themselves are fairly trivial, but the concepts are not, so I'd recommend familiarizing yourself with the conceptual background. I've provided a few sources with notes on takeaways, and my explanation and rundown of the Go source code as it pertains to this change.

##### References
1. https://github.com/golang/go/issues/26439
- calling Fd() will prevent Read() from returning after close
- one supported workflow for performing I/O on file descirptors is: setting the FD to non-blocking mode, then passing the FD to os.NewFile (this is what I ended up doing)

2. https://github.com/golang/go/issues/22939
- if a file passed to NewFile is in non-blocking mode, then it will be added to the runtime I/O poller

3. https://morsmachine.dk/netpoller
- Socket i/O in Golang uses the netpoller mechanism, which handles I/O events in a separate thread, and enables I/O to block the goroutine rather than the entire kernel thread
- uses platform-specific async I/O mechanisms (epoll, kqueue, etc)
- this is socket-specific, but normal FS I/O is probably handled the same way

##### Golang I/O Rundown
In Golang, I/O is blocking, but we can't directly use blocking syscalls, as this would block whichever kernel thread the Goroutine was running in. As a solution, Go I/O is performed via the internal poller package (https://pkg.go.dev/internal/poll), which parks goroutines while waiting for I/O, and uses the platform-specific asynchronous I/O mechanism (e.g. epoll) to detect FDs for which I/O can be performed.

I went through the code to make sure that this is the case, here's a small rundown:
Here's the implementation of read in *os.FIle:
```go
// Read reads up to len(b) bytes from the File and stores them in b.
// It returns the number of bytes read and any error encountered.
// At end of file, Read returns 0, io.EOF.
func (f *File) Read(b []byte) (n int, err error) {
	if err := f.checkValid("read"); err != nil {
		return 0, err
	}
	n, e := f.read(b)
	return n, f.wrapErr("read", e)
}
```

we can see this method calls .read(), which is defined as follows for posix:
```go
// read reads up to len(b) bytes from the File.
// It returns the number of bytes read and an error, if any.
func (f *File) read(b []byte) (n int, err error) {
	n, err = f.pfd.Read(b)
	runtime.KeepAlive(f)
	return n, err
}
```
It calls pfd.Read(), which we can see refers to poll.FD:

```go
// File represents an open file descriptor.
type File struct {
	*file // os specific
}
```

```go
// file is the real representation of *File.
// The extra level of indirection ensures that no clients of os
// can overwrite this data, which could cause the finalizer
// to close the wrong file descriptor.
type file struct {
	pfd         poll.FD
	name        string
	dirinfo     *dirInfo // nil unless directory being read
	nonblock    bool     // whether we set nonblocking mode
	stdoutOrErr bool     // whether this is stdout or stderr
	appendMode  bool     // whether file is opened for appending
}
```

The implementation of poll.FD's read for unix is as follows:
```go
// Read implements io.Reader.
func (fd *FD) Read(p []byte) (int, error) {
	if err := fd.readLock(); err != nil {
		return 0, err
	}
	defer fd.readUnlock()
	if len(p) == 0 {
		// If the caller wanted a zero byte read, return immediately
		// without trying (but after acquiring the readLock).
		// Otherwise syscall.Read returns 0, nil which looks like
		// io.EOF.
		// TODO(bradfitz): make it wait for readability? (Issue 15735)
		return 0, nil
	}
	if err := fd.pd.prepareRead(fd.isFile); err != nil {
		return 0, err
	}
	if fd.IsStream && len(p) > maxRW {
		p = p[:maxRW]
	}
	for {
		n, err := ignoringEINTRIO(syscall.Read, fd.Sysfd, p)
		if err != nil {
			n = 0
			if err == syscall.EAGAIN && fd.pd.pollable() {
				if err = fd.pd.waitRead(fd.isFile); err == nil {
					continue
				}
			}
		}
		err = fd.eofError(n, err)
		return n, err
	}
}
```
We first execute the read syscall. If the file descriptor is non-blocking, then it will return quickly with EAGAIN and wait for waitRead to return. If it's blocking, then it'll cause the current kernel thread to sleep, and will not call waitRead(), as the fd isn't pollable and doesn't return EAGAIN.

waitRead is as follows:
```go
func (pd *pollDesc) waitRead(isFile bool) error {
	return pd.wait('r', isFile)
}
```
which calls pd.Wait:
```go
func (pd *pollDesc) wait(mode int, isFile bool) error {
	if pd.runtimeCtx == 0 {
		return errors.New("waiting for unsupported file type")
	}
	res := runtime_pollWait(pd.runtimeCtx, mode)
	return convertErr(res, isFile)
}

``` 
which calls runtime_pollWait() (note that this is the internal/poll package)

```go
// poll_runtime_pollWait, which is internal/poll.runtime_pollWait,
// waits for a descriptor to be ready for reading or writing,
// according to mode, which is 'r' or 'w'.
// This returns an error code; the codes are defined above.
//go:linkname poll_runtime_pollWait internal/poll.runtime_pollWait
func poll_runtime_pollWait(pd *pollDesc, mode int) int {
	errcode := netpollcheckerr(pd, int32(mode))
	if errcode != pollNoError {
		return errcode
	}
	// As for now only Solaris, illumos, and AIX use level-triggered IO.
	if GOOS == "solaris" || GOOS == "illumos" || GOOS == "aix" {
		netpollarm(pd, mode)
	}
	for !netpollblock(pd, int32(mode), false) {
		errcode = netpollcheckerr(pd, int32(mode))
		if errcode != pollNoError {
			return errcode
		}
		// Can happen if timeout has fired and unblocked us,
		// but before we had a chance to run, timeout has been reset.
		// Pretend it has not happened and retry.
	}
	return pollNoError
}
```

which calls netpollblock:
```go
// returns true if IO is ready, or false if timedout or closed
// waitio - wait only for completed IO, ignore errors
// Concurrent calls to netpollblock in the same mode are forbidden, as pollDesc
// can hold only a single waiting goroutine for each mode.
func netpollblock(pd *pollDesc, mode int32, waitio bool) bool {
	gpp := &pd.rg
	if mode == 'w' {
		gpp = &pd.wg
	}

	// set the gpp semaphore to pdWait
	for {
		// Consume notification if already ready.
		if gpp.CompareAndSwap(pdReady, 0) {
			return true
		}
		if gpp.CompareAndSwap(0, pdWait) {
			break
		}

		// Double check that this isn't corrupt; otherwise we'd loop
		// forever.
		if v := gpp.Load(); v != pdReady && v != 0 {
			throw("runtime: double wait")
		}
	}

	// need to recheck error states after setting gpp to pdWait
	// this is necessary because runtime_pollUnblock/runtime_pollSetDeadline/deadlineimpl
	// do the opposite: store to closing/rd/wd, publishInfo, load of rg/wg
	if waitio || netpollcheckerr(pd, mode) == pollNoError {
		gopark(netpollblockcommit, unsafe.Pointer(gpp), waitReasonIOWait, traceEvGoBlockNet, 5)
	}
	// be careful to not lose concurrent pdReady notification
	old := gpp.Swap(0)
	if old > pdWait {
		throw("runtime: corrupted polldesc")
	}
	return old == pdReady
}
```
I'll assume that the comment, " returns true if IO is ready, or false if timedout or closed" is true. The current goroutine is parked until I/O is ready.

gpp is set to ready within the function netpollunblock, which is called by netpollready :
```go
// netpollready is called by the platform-specific netpoll function.
// It declares that the fd associated with pd is ready for I/O.
// The toRun argument is used to build a list of goroutines to return
// from netpoll. The mode argument is 'r', 'w', or 'r'+'w' to indicate
// whether the fd is ready for reading or writing or both.
//
// This may run while the world is stopped, so write barriers are not allowed.
//go:nowritebarrier
func netpollready(toRun *gList, pd *pollDesc, mode int32) {
	var rg, wg *g
	if mode == 'r' || mode == 'r'+'w' {
		rg = netpollunblock(pd, 'r', true)
	}
	if mode == 'w' || mode == 'r'+'w' {
		wg = netpollunblock(pd, 'w', true)
	}
	if rg != nil {
		toRun.push(rg)
	}
	if wg != nil {
		toRun.push(wg)
	}
}
```

netpollready is called by the platform-specific implementation of netpoll(); here's the implementation for Linux, which uses epoll to determine which FDs are ready for I/O, and notifies the corresponding poll descriptor.
```go
// netpoll checks for ready network connections.
// Returns list of goroutines that become runnable.
// delay < 0: blocks indefinitely
// delay == 0: does not block, just polls
// delay > 0: block for up to that many nanoseconds
func netpoll(delay int64) gList {
	if epfd == -1 {
		return gList{}
	}
	var waitms int32
	if delay < 0 {
		waitms = -1
	} else if delay == 0 {
		waitms = 0
	} else if delay < 1e6 {
		waitms = 1
	} else if delay < 1e15 {
		waitms = int32(delay / 1e6)
	} else {
		// An arbitrary cap on how long to wait for a timer.
		// 1e9 ms == ~11.5 days.
		waitms = 1e9
	}
	var events [128]epollevent
retry:
	n := epollwait(epfd, &events[0], int32(len(events)), waitms)
	if n < 0 {
		if n != -_EINTR {
			println("runtime: epollwait on fd", epfd, "failed with", -n)
			throw("runtime: netpoll failed")
		}
		// If a timed sleep was interrupted, just return to
		// recalculate how long we should sleep now.
		if waitms > 0 {
			return gList{}
		}
		goto retry
	}
	var toRun gList
	for i := int32(0); i < n; i++ {
		ev := &events[i]
		if ev.events == 0 {
			continue
		}

		if *(**uintptr)(unsafe.Pointer(&ev.data)) == &netpollBreakRd {
			if ev.events != _EPOLLIN {
				println("runtime: netpoll: break fd ready for", ev.events)
				throw("runtime: netpoll: break fd ready for something unexpected")
			}
			if delay != 0 {
				// netpollBreak could be picked up by a
				// nonblocking poll. Only read the byte
				// if blocking.
				var tmp [16]byte
				read(int32(netpollBreakRd), noescape(unsafe.Pointer(&tmp[0])), int32(len(tmp)))
				atomic.Store(&netpollWakeSig, 0)
			}
			continue
		}

		var mode int32
		if ev.events&(_EPOLLIN|_EPOLLRDHUP|_EPOLLHUP|_EPOLLERR) != 0 {
			mode += 'r'
		}
		if ev.events&(_EPOLLOUT|_EPOLLHUP|_EPOLLERR) != 0 {
			mode += 'w'
		}
		if mode != 0 {
			pd := *(**pollDesc)(unsafe.Pointer(&ev.data))
			pd.setEventErr(ev.events == _EPOLLERR)
			netpollready(&toRun, pd, mode)
		}
	}
	return toRun
}
```

#### How should this be manually tested?
I think that the changes I've made are fairly easy to reason about, and automated tests should validate that the behavior is correct. I tested this change on Linux, which is the affected platform. I did not test on other platforms.

### Change #2 
#### Where should the reviewer start?
I would read https://github.com/fsnotify/fsnotify/issues/404. My argument is that since inotify guarantees that events will be sent in order, and by the time the client receives the event for which we're calling lstat the file may have been removed anyway, there's no need to perform LStat for so many file events.

#### How should this be manually tested?
I think the automated tests should cover it. I tested this change on Linux, which is the affected platform. I did not test on other platforms.




nshalman: Thank you!
I'm not an expert on inotify and haven't done a close read of your changes yet, but I have a couple of questions that I didn't see addressed by your otherwise very thorough writeup.

Does this change imply a newer minimum version of Linux than the previous implementation? If so, do you know what version that would be?

Is there any way to add a test (ideally an integration test that runs on all platforms) that fails before this change is applied and that passes after applying it? Perhaps rather than the usual test pattern of spinning up a goroutine to consume events, explicitly do all the work to queue up the events in the kernel before then consuming all of them afterwards in a single thread?

Thank you for digging into this issue! I will try to find time to review these changes if some other linux expert doesn't get to it first.
horahoradev: I don't see any reason why this change would increase the minimum Linux version. My changes as it relates to the kernel's API would be:
1. Removed use of epoll (this is not an issue)
2. I added a call to change a FD to non-blocking... I'm 99.9% this isn't an issue. I'll need to dig up the man page for fcntl on 2.6.37 (the minimum Linux version we support) and see if it includes non-blocking IO

The changes I've made would only be relevant for Linux; other platforms aren't affected, so I wouldn't be able to add a test for them, but I can definitely add one for Linux. My initial idea is to monitor whether the kernel thread blocks too often on epoll... but I'll need to think about it some more.
horahoradev: Here's my new test:
```go
func TestINotifyNoBlockingSyscalls(t *testing.T) {
	startingThreads := pprof.Lookup("threadcreate").Count()

	for i := 0; i <= runtime.NumCPU()*3; i++ {
		testDir := tempMkdir(t)
		defer os.RemoveAll(testDir)

		w, err := NewWatcher()
		if err != nil {
			t.Fatalf("Failed to create watcher: %v", err)
		}

		w.Add(testDir)

		testSubdir := fmt.Sprintf("%s/1", testDir)

		err = os.Mkdir(testSubdir, 0777)
		if err != nil {
			t.Fatalf("Cannot create subdir: %v", err)
		}

		<-w.Events
	}

	// Bad synchronization mechanism
	time.Sleep(time.Second * 2)

	endingThreads := pprof.Lookup("threadcreate").Count()

	// Did we spawn any new threads?
	if startingThreads != endingThreads {
		t.Fatalf("Expected %v threads but got %v", startingThreads, endingThreads)
	}

	// Is the Go runtime spawning MANY more threads than necessary?
	if runtime.NumCPU()+10 <= endingThreads {
		t.Fatalf("Expected %v threads (num cores) but got %v", runtime.NumCPU(), endingThreads)
	}
}
```
The gist is that we try to spawn 3x the number of cores as watchers, and see if:
1. the number of kernel threads increases over the course of the test
2. the number of kernel threads is significantly higher than the number of cores on the machine

From the runtime package docs:
```
The GOMAXPROCS variable limits the number of operating system threads that can execute user-level Go code simultaneously. There is no limit to the number of threads that can be blocked in system calls on behalf of Go code; those do not count against the GOMAXPROCS limit. This package's GOMAXPROCS function queries and changes the limit.
```
GOMAXPROCS defaults to the number of cores, so if we have 16 logical cores and 200 goroutines, we can expect that the Go runtime will spawn at most 16 threads for those goroutines to run on, unless those goroutines have blocking system calls, which may create additional threads, so my test checks to ensure that the number of threads is no more than 10 above the number of cores.

This is currently working on my machine (I have 16 cores and end up getting 22-23 threads), and failing on my machine for the old branch (I end up getting ~53 kernel threads here), but both branches yield an excessive number of threads on the git workflow tests. I'll need to keep working on this tomorrow.

horahoradev: I've added a new Github action which runs the tests on Debian 6, whose kernel version is 2.6.32. Let me know what you think; this isn't exactly the minimum version we support, but it's reasonably close. I think we can be confident that this change doesn't break platform support.
horahoradev: I made my test better, so now the improvement is pretty easy to see. In summary, I create one watcher, then spawn 60 watcher.readEvents() goroutines, and see if that yields any (>0) new kthreads. This passes with my changes, but on main:

```
--- FAIL: TestINotifyNoBlockingSyscalls (2.01s)
    inotify_test.go:493: Got a nonzero diff 41. starting: 27. ending: 68
FAIL
FAIL    github.com/fsnotify/fsnotify    12.084s
FAIL
```
It spawns 41 new kernel threads, whereas my changes spawn 0. This works because previously readEvents had the blocking read syscall, so spawning 60 readEvents() goroutinse will result in ~60ish new kernel threads.

Sorry for the wall of text, I think I'm done now. Ready for review!
arp242: NB: I rebased this branch on the latest main and removed the Debian 6/Vagrant tests for now; I added those in #469.

I think this mostly looks good, except for my comment about the Lstat. What I don't understand is why the epoll approach was done in the first place, since using IN_NONBLOCK is so much simpler. Basically, it looks "too good to be true", so I'm wondering if I'm missing something 🤔

Maybe IN_NONBLOCK was added later? Or maybe the original author simply didn't realize it could be done like this (I've written "useless" heaps of code like this myself that could be replaced with something much simpler)?
arp242: The epoll method was added in #66, which has next to no information on *why* this approach was chosen. In #1 and #5 IN_NONBLOCK is mentioned as  "If we use IN_NONBLOCK fsnotify would require Linux 2.6.27." and "I noticed that inotify_init1() has a IN_NONBLOCK option in Linux 2.6.27+ (Go 1.3 requires Linux 2.6.23)". The old repo [has a reference to it as well](https://github.com/howeyc/fsnotify/blob/f0c08ee9c60704c1879025f2ae0ff3e000082c13/fsnotify_linux.go#L22).

That's all the context I could find.
horahoradev: The epoll code was written in ~2015, but Golang didn't add the ability to register file descriptors (as returned by raw syscalls, like inotify_init) with Go's runtime poller until years later. See the issues referenced in https://github.com/fsnotify/fsnotify/issues/240 . I think that's the main reason: we're taking advantage of functionality that didn't exist when the code was originally written.
horahoradev: PR amended as suggested

Patch:
diff --git a/inotify.go b/inotify.go
index b01124a..a42da58 100644
--- a/inotify.go
+++ b/inotify.go
@@ -22,39 +22,36 @@ import (
 
 // Watcher watches a set of files, delivering events to a channel.
 type Watcher struct {
-	Events   chan Event
-	Errors   chan error
-	mu       sync.Mutex // Map access
-	fd       int
-	poller   *fdPoller
-	watches  map[string]*watch // Map of inotify watches (key: path)
-	paths    map[int]string    // Map of watched paths (key: watch descriptor)
-	done     chan struct{}     // Channel for sending a "quit message" to the reader goroutine
-	doneResp chan struct{}     // Channel to respond to Close
+	fd          int // https://github.com/golang/go/issues/26439 can't call .Fd() on os.FIle or Read will no longer return on Close()
+	Events      chan Event
+	Errors      chan error
+	mu          sync.Mutex // Map access
+	inotifyFile *os.File
+	watches     map[string]*watch // Map of inotify watches (key: path)
+	paths       map[int]string    // Map of watched paths (key: watch descriptor)
+	done        chan struct{}     // Channel for sending a "quit message" to the reader goroutine
+	doneResp    chan struct{}     // Channel to respond to Close
 }
 
 // NewWatcher establishes a new watcher with the underlying OS and begins waiting for events.
 func NewWatcher() (*Watcher, error) {
 	// Create inotify fd
-	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC)
+	// Need to set the FD to nonblocking mode in order for SetDeadline methods to work
+	// Otherwise, blocking i/o operations won't terminate on close
+	fd, errno := unix.InotifyInit1(unix.IN_CLOEXEC | unix.IN_NONBLOCK)
 	if fd == -1 {
 		return nil, errno
 	}
-	// Create epoll
-	poller, err := newFdPoller(fd)
-	if err != nil {
-		unix.Close(fd)
-		return nil, err
-	}
+
 	w := &Watcher{
-		fd:       fd,
-		poller:   poller,
-		watches:  make(map[string]*watch),
-		paths:    make(map[int]string),
-		Events:   make(chan Event),
-		Errors:   make(chan error),
-		done:     make(chan struct{}),
-		doneResp: make(chan struct{}),
+		fd:          fd,
+		inotifyFile: os.NewFile(uintptr(fd), ""),
+		watches:     make(map[string]*watch),
+		paths:       make(map[int]string),
+		Events:      make(chan Event),
+		Errors:      make(chan error),
+		done:        make(chan struct{}),
+		doneResp:    make(chan struct{}),
 	}
 
 	go w.readEvents()
@@ -82,8 +79,11 @@ func (w *Watcher) Close() error {
 	close(w.done)
 	w.mu.Unlock()
 
-	// Wake up goroutine
-	w.poller.wake()
+	// Causes any blocking reads to return with an error, provided the file still supports deadline operations
+	err := w.inotifyFile.Close()
+	if err != nil {
+		return err
+	}
 
 	// Wait for goroutine to close
 	<-w.doneResp
@@ -189,16 +189,12 @@ type watch struct {
 func (w *Watcher) readEvents() {
 	var (
 		buf   [unix.SizeofInotifyEvent * 4096]byte // Buffer for a maximum of 4096 raw events
-		n     int                                  // Number of bytes read with read()
 		errno error                                // Syscall errno
-		ok    bool                                 // For poller.wait
 	)
 
 	defer close(w.doneResp)
 	defer close(w.Errors)
 	defer close(w.Events)
-	defer unix.Close(w.fd)
-	defer w.poller.close()
 
 	for {
 		// See if we have been closed.
@@ -206,33 +202,19 @@ func (w *Watcher) readEvents() {
 			return
 		}
 
-		ok, errno = w.poller.wait()
-		if errno != nil {
+		n, err := w.inotifyFile.Read(buf[:])
+		switch {
+		case errors.Unwrap(err) == os.ErrClosed:
+			return
+		case err != nil:
 			select {
-			case w.Errors <- errno:
+			case w.Errors <- err:
 			case <-w.done:
 				return
 			}
 			continue
 		}
 
-		if !ok {
-			continue
-		}
-
-		n, errno = unix.Read(w.fd, buf[:])
-		// If a signal interrupted execution, see if we've been asked to close, and try again.
-		// http://man7.org/linux/man-pages/man7/signal.7.html :
-		// "Before Linux 3.8, reads from an inotify(7) file descriptor were not restartable"
-		if errno == unix.EINTR {
-			continue
-		}
-
-		// unix.Read might have been woken up by Close. If so, we're done.
-		if w.isClosed() {
-			return
-		}
-
 		if n < unix.SizeofInotifyEvent {
 			var err error
 			if n == 0 {
diff --git a/inotify_poller.go b/inotify_poller.go
deleted file mode 100644
index b572a37..0000000
--- a/inotify_poller.go
+++ /dev/null
@@ -1,187 +0,0 @@
-// Copyright 2015 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-//go:build linux
-// +build linux
-
-package fsnotify
-
-import (
-	"errors"
-
-	"golang.org/x/sys/unix"
-)
-
-type fdPoller struct {
-	fd   int    // File descriptor (as returned by the inotify_init() syscall)
-	epfd int    // Epoll file descriptor
-	pipe [2]int // Pipe for waking up
-}
-
-func emptyPoller(fd int) *fdPoller {
-	poller := new(fdPoller)
-	poller.fd = fd
-	poller.epfd = -1
-	poller.pipe[0] = -1
-	poller.pipe[1] = -1
-	return poller
-}
-
-// Create a new inotify poller.
-// This creates an inotify handler, and an epoll handler.
-func newFdPoller(fd int) (*fdPoller, error) {
-	var errno error
-	poller := emptyPoller(fd)
-	defer func() {
-		if errno != nil {
-			poller.close()
-		}
-	}()
-
-	// Create epoll fd
-	poller.epfd, errno = unix.EpollCreate1(unix.EPOLL_CLOEXEC)
-	if poller.epfd == -1 {
-		return nil, errno
-	}
-	// Create pipe; pipe[0] is the read end, pipe[1] the write end.
-	errno = unix.Pipe2(poller.pipe[:], unix.O_NONBLOCK|unix.O_CLOEXEC)
-	if errno != nil {
-		return nil, errno
-	}
-
-	// Register inotify fd with epoll
-	event := unix.EpollEvent{
-		Fd:     int32(poller.fd),
-		Events: unix.EPOLLIN,
-	}
-	errno = unix.EpollCtl(poller.epfd, unix.EPOLL_CTL_ADD, poller.fd, &event)
-	if errno != nil {
-		return nil, errno
-	}
-
-	// Register pipe fd with epoll
-	event = unix.EpollEvent{
-		Fd:     int32(poller.pipe[0]),
-		Events: unix.EPOLLIN,
-	}
-	errno = unix.EpollCtl(poller.epfd, unix.EPOLL_CTL_ADD, poller.pipe[0], &event)
-	if errno != nil {
-		return nil, errno
-	}
-
-	return poller, nil
-}
-
-// Wait using epoll.
-// Returns true if something is ready to be read,
-// false if there is not.
-func (poller *fdPoller) wait() (bool, error) {
-	// 3 possible events per fd, and 2 fds, makes a maximum of 6 events.
-	// I don't know whether epoll_wait returns the number of events returned,
-	// or the total number of events ready.
-	// I decided to catch both by making the buffer one larger than the maximum.
-	events := make([]unix.EpollEvent, 7)
-	for {
-		n, errno := unix.EpollWait(poller.epfd, events, -1)
-		if n == -1 {
-			if errno == unix.EINTR {
-				continue
-			}
-			return false, errno
-		}
-		if n == 0 {
-			// If there are no events, try again.
-			continue
-		}
-		if n > 6 {
-			// This should never happen. More events were returned than should be possible.
-			return false, errors.New("epoll_wait returned more events than I know what to do with")
-		}
-		ready := events[:n]
-		epollhup := false
-		epollerr := false
-		epollin := false
-		for _, event := range ready {
-			if event.Fd == int32(poller.fd) {
-				if event.Events&unix.EPOLLHUP != 0 {
-					// This should not happen, but if it does, treat it as a wakeup.
-					epollhup = true
-				}
-				if event.Events&unix.EPOLLERR != 0 {
-					// If an error is waiting on the file descriptor, we should pretend
-					// something is ready to read, and let unix.Read pick up the error.
-					epollerr = true
-				}
-				if event.Events&unix.EPOLLIN != 0 {
-					// There is data to read.
-					epollin = true
-				}
-			}
-			if event.Fd == int32(poller.pipe[0]) {
-				if event.Events&unix.EPOLLHUP != 0 {
-					// Write pipe descriptor was closed, by us. This means we're closing down the
-					// watcher, and we should wake up.
-				}
-				if event.Events&unix.EPOLLERR != 0 {
-					// If an error is waiting on the pipe file descriptor.
-					// This is an absolute mystery, and should never ever happen.
-					return false, errors.New("Error on the pipe descriptor.")
-				}
-				if event.Events&unix.EPOLLIN != 0 {
-					// This is a regular wakeup, so we have to clear the buffer.
-					err := poller.clearWake()
-					if err != nil {
-						return false, err
-					}
-				}
-			}
-		}
-
-		if epollhup || epollerr || epollin {
-			return true, nil
-		}
-		return false, nil
-	}
-}
-
-// Close the write end of the poller.
-func (poller *fdPoller) wake() error {
-	buf := make([]byte, 1)
-	n, errno := unix.Write(poller.pipe[1], buf)
-	if n == -1 {
-		if errno == unix.EAGAIN {
-			// Buffer is full, poller will wake.
-			return nil
-		}
-		return errno
-	}
-	return nil
-}
-
-func (poller *fdPoller) clearWake() error {
-	// You have to be woken up a LOT in order to get to 100!
-	buf := make([]byte, 100)
-	n, errno := unix.Read(poller.pipe[0], buf)
-	if n == -1 {
-		if errno == unix.EAGAIN {
-			// Buffer is empty, someone else cleared our wake.
-			return nil
-		}
-		return errno
-	}
-	return nil
-}
-
-// Close all poller file descriptors, but not the one passed to it.
-func (poller *fdPoller) close() {
-	if poller.pipe[1] != -1 {
-		unix.Close(poller.pipe[1])
-	}
-	if poller.pipe[0] != -1 {
-		unix.Close(poller.pipe[0])
-	}
-	if poller.epfd != -1 {
-		unix.Close(poller.epfd)
-	}
-}


Test patch:
diff --git a/inotify_poller_test.go b/inotify_poller_test.go
deleted file mode 100644
index 110e00d..0000000
--- a/inotify_poller_test.go
+++ /dev/null
@@ -1,234 +0,0 @@
-// Copyright 2015 The Go Authors. All rights reserved.
-// Use of this source code is governed by a BSD-style
-// license that can be found in the LICENSE file.
-
-//go:build linux
-// +build linux
-
-package fsnotify
-
-import (
-	"testing"
-	"time"
-
-	"golang.org/x/sys/unix"
-)
-
-type testFd [2]int
-
-func makeTestFd(t *testing.T) testFd {
-	var tfd testFd
-	errno := unix.Pipe(tfd[:])
-	if errno != nil {
-		t.Fatalf("Failed to create pipe: %v", errno)
-	}
-	return tfd
-}
-
-func (tfd testFd) fd() int {
-	return tfd[0]
-}
-
-func (tfd testFd) closeWrite(t *testing.T) {
-	errno := unix.Close(tfd[1])
-	if errno != nil {
-		t.Fatalf("Failed to close write end of pipe: %v", errno)
-	}
-}
-
-func (tfd testFd) put(t *testing.T) {
-	buf := make([]byte, 10)
-	_, errno := unix.Write(tfd[1], buf)
-	if errno != nil {
-		t.Fatalf("Failed to write to pipe: %v", errno)
-	}
-}
-
-func (tfd testFd) get(t *testing.T) {
-	buf := make([]byte, 10)
-	_, errno := unix.Read(tfd[0], buf)
-	if errno != nil {
-		t.Fatalf("Failed to read from pipe: %v", errno)
-	}
-}
-
-func (tfd testFd) close() {
-	unix.Close(tfd[1])
-	unix.Close(tfd[0])
-}
-
-func makePoller(t *testing.T) (testFd, *fdPoller) {
-	tfd := makeTestFd(t)
-	poller, err := newFdPoller(tfd.fd())
-	if err != nil {
-		t.Fatalf("Failed to create poller: %v", err)
-	}
-	return tfd, poller
-}
-
-func TestPollerWithBadFd(t *testing.T) {
-	_, err := newFdPoller(-1)
-	if err != unix.EBADF {
-		t.Fatalf("Expected EBADF, got: %v", err)
-	}
-}
-
-func TestPollerWithData(t *testing.T) {
-	tfd, poller := makePoller(t)
-	defer tfd.close()
-	defer poller.close()
-
-	tfd.put(t)
-	ok, err := poller.wait()
-	if err != nil {
-		t.Fatalf("poller failed: %v", err)
-	}
-	if !ok {
-		t.Fatalf("expected poller to return true")
-	}
-	tfd.get(t)
-}
-
-func TestPollerWithWakeup(t *testing.T) {
-	tfd, poller := makePoller(t)
-	defer tfd.close()
-	defer poller.close()
-
-	err := poller.wake()
-	if err != nil {
-		t.Fatalf("wake failed: %v", err)
-	}
-	ok, err := poller.wait()
-	if err != nil {
-		t.Fatalf("poller failed: %v", err)
-	}
-	if ok {
-		t.Fatalf("expected poller to return false")
-	}
-}
-
-func TestPollerWithClose(t *testing.T) {
-	tfd, poller := makePoller(t)
-	defer tfd.close()
-	defer poller.close()
-
-	tfd.closeWrite(t)
-	ok, err := poller.wait()
-	if err != nil {
-		t.Fatalf("poller failed: %v", err)
-	}
-	if !ok {
-		t.Fatalf("expected poller to return true")
-	}
-}
-
-func TestPollerWithWakeupAndData(t *testing.T) {
-	tfd, poller := makePoller(t)
-	defer tfd.close()
-	defer poller.close()
-
-	tfd.put(t)
-	err := poller.wake()
-	if err != nil {
-		t.Fatalf("wake failed: %v", err)
-	}
-
-	// both data and wakeup
-	ok, err := poller.wait()
-	if err != nil {
-		t.Fatalf("poller failed: %v", err)
-	}
-	if !ok {
-		t.Fatalf("expected poller to return true")
-	}
-
-	// data is still in the buffer, wakeup is cleared
-	ok, err = poller.wait()
-	if err != nil {
-		t.Fatalf("poller failed: %v", err)
-	}
-	if !ok {
-		t.Fatalf("expected poller to return true")
-	}
-
-	tfd.get(t)
-	// data is gone, only wakeup now
-	err = poller.wake()
-	if err != nil {
-		t.Fatalf("wake failed: %v", err)
-	}
-	ok, err = poller.wait()
-	if err != nil {
-		t.Fatalf("poller failed: %v", err)
-	}
-	if ok {
-		t.Fatalf("expected poller to return false")
-	}
-}
-
-func TestPollerConcurrent(t *testing.T) {
-	tfd, poller := makePoller(t)
-	defer tfd.close()
-	defer poller.close()
-
-	oks := make(chan bool)
-	live := make(chan bool)
-	defer close(live)
-	go func() {
-		defer close(oks)
-		for {
-			ok, err := poller.wait()
-			if err != nil {
-				t.Errorf("poller failed: %v", err)
-			}
-			oks <- ok
-			if !<-live {
-				return
-			}
-		}
-	}()
-
-	// Try a write
-	select {
-	case <-time.After(50 * time.Millisecond):
-	case <-oks:
-		t.Fatalf("poller did not wait")
-	}
-	tfd.put(t)
-	if !<-oks {
-		t.Fatalf("expected true")
-	}
-	tfd.get(t)
-	live <- true
-
-	// Try a wakeup
-	select {
-	case <-time.After(50 * time.Millisecond):
-	case <-oks:
-		t.Fatalf("poller did not wait")
-	}
-	err := poller.wake()
-	if err != nil {
-		t.Fatalf("wake failed: %v", err)
-	}
-	if <-oks {
-		t.Fatalf("expected false")
-	}
-	live <- true
-
-	// Try a close
-	select {
-	case <-time.After(50 * time.Millisecond):
-	case <-oks:
-		t.Fatalf("poller did not wait")
-	}
-	tfd.closeWrite(t)
-	if !<-oks {
-		t.Fatalf("expected true")
-	}
-	tfd.get(t)
-
-	// wait for all goroutines for finish.
-	live <- false
-	<-oks
-}
diff --git a/inotify_test.go b/inotify_test.go
index 269c3ff..d9ee53f 100644
--- a/inotify_test.go
+++ b/inotify_test.go
@@ -550,3 +550,35 @@ func TestInotifyDeleteOpenedFile(t *testing.T) {
 	fd.Close()
 	checkEvent(Remove)
 }
+
+func TestINotifyNoBlockingSyscalls(t *testing.T) {
+	getThreads := func() int {
+		d := fmt.Sprintf("/proc/%d/task", os.Getpid())
+		ls, err := os.ReadDir(d)
+		if err != nil {
+			t.Fatalf("reading %q: %s", d, err)
+		}
+		return len(ls)
+	}
+
+	w, err := NewWatcher()
+	if err != nil {
+		t.Fatalf("Failed to create watcher: %v", err)
+	}
+
+	startingThreads := getThreads()
+	// Call readEvents a bunch of times; if this function has a blocking raw syscall, it'll create many new kthreads
+	for i := 0; i <= 60; i++ {
+		go w.readEvents()
+	}
+
+	// Bad synchronization mechanism
+	time.Sleep(time.Second * 2)
+
+	endingThreads := getThreads()
+
+	// Did we spawn any new threads?
+	if diff := endingThreads - startingThreads; diff > 0 {
+		t.Fatalf("Got a nonzero diff %v. starting: %v. ending: %v", diff, startingThreads, endingThreads)
+	}
+}