fusefs: support kqueue for /dev/fuse

/dev/fuse was already pollable with poll and select.  Add support for
kqueue, too.  And add tests for polling with poll, select, and kqueue.

Sponsored by:	The FreeBSD Foundation

sys/fs/fuse/fuse_device.c

@@ -93,12 +93,14 @@ SDT_PROBE_DEFINE2(fusefs, , device, trace, "int", "char*");
 
 static struct cdev *fuse_dev;
 
+static d_kqfilter_t fuse_device_filter;
 static d_open_t fuse_device_open;
 static d_poll_t fuse_device_poll;
 static d_read_t fuse_device_read;
 static d_write_t fuse_device_write;
 
 static struct cdevsw fuse_device_cdevsw = {
+	.d_kqfilter = fuse_device_filter,
 	.d_open = fuse_device_open,
 	.d_name = "fuse",
 	.d_poll = fuse_device_poll,
@@ -107,6 +109,15 @@ static struct cdevsw fuse_device_cdevsw = {
 	.d_version = D_VERSION,
 };
 
+static int fuse_device_filt_read(struct knote *kn, long hint);
+static void fuse_device_filt_detach(struct knote *kn);
+
+struct filterops fuse_device_rfiltops = {
+	.f_isfd = 1,
+	.f_detach = fuse_device_filt_detach,
+	.f_event = fuse_device_filt_read,
+};
+
 /****************************
  *
  * >>> Fuse device op defs
@@ -145,6 +156,68 @@ fdata_dtor(void *arg)
 	fdata_trydestroy(fdata);
 }
 
+static int
+fuse_device_filter(struct cdev *dev, struct knote *kn)
+{
+	struct fuse_data *data;
+	int error;
+
+	error = devfs_get_cdevpriv((void **)&data);
+
+	/* EVFILT_WRITE is not supported; the device is always ready to write */
+	if (error == 0 && kn->kn_filter == EVFILT_READ) {
+		kn->kn_fop = &fuse_device_rfiltops;
+		kn->kn_hook = data;
+		knlist_add(&data->ks_rsel.si_note, kn, 0);
+		error = 0;
+	} else if (error == 0) {
+		error = EINVAL;
+		kn->kn_data = error;
+	}
+
+	return (error);
+}
+
+static void
+fuse_device_filt_detach(struct knote *kn)
+{
+	struct fuse_data *data;
+
+	data = (struct fuse_data*)kn->kn_hook;
+	MPASS(data != NULL);
+	knlist_remove(&data->ks_rsel.si_note, kn, 0);
+	kn->kn_hook = NULL;
+}
+
+static int
+fuse_device_filt_read(struct knote *kn, long hint)
+{
+	struct fuse_data *data;
+	int ready;
+
+	data = (struct fuse_data*)kn->kn_hook;
+	MPASS(data != NULL);
+
+	mtx_assert(&data->ms_mtx, MA_OWNED);
+	if (fdata_get_dead(data)) {
+		kn->kn_flags |= EV_EOF;
+		kn->kn_fflags = ENODEV;
+		kn->kn_data = 1;
+		ready = 1;
+	} else if (STAILQ_FIRST(&data->ms_head)) {
+		/* 
+		 * There is at least one event to read.
+		 * TODO: keep a counter of the number of events to read
+		 */
+		kn->kn_data = 1;
+		ready = 1;
+	} else {
+		ready = 0;
+	}
+
+	return (ready);
+}
+
 /*
  * Resources are set up on a per-open basis
  */

sys/fs/fuse/fuse_ipc.c

@@ -586,6 +586,7 @@ fdata_alloc(struct cdev *fdev, struct ucred *cred)
 	data->fdev = fdev;
 	mtx_init(&data->ms_mtx, "fuse message list mutex", NULL, MTX_DEF);
 	STAILQ_INIT(&data->ms_head);
+	knlist_init_mtx(&data->ks_rsel.si_note, &data->ms_mtx);
 	mtx_init(&data->aw_mtx, "fuse answer list mutex", NULL, MTX_DEF);
 	TAILQ_INIT(&data->aw_head);
 	data->daemoncred = crhold(cred);
@@ -605,11 +606,12 @@ fdata_trydestroy(struct fuse_data *data)
 		return;
 
 	/* Driving off stage all that stuff thrown at device... */
-	mtx_destroy(&data->ms_mtx);
-	mtx_destroy(&data->aw_mtx);
 	sx_destroy(&data->rename_lock);
-
 	crfree(data->daemoncred);
+	mtx_destroy(&data->aw_mtx);
+	knlist_delete(&data->ks_rsel.si_note, curthread, 0);
+	knlist_destroy(&data->ks_rsel.si_note);
+	mtx_destroy(&data->ms_mtx);
 
 	free(data, M_FUSEMSG);
 }
@@ -702,6 +704,7 @@ fuse_insert_message(struct fuse_ticket *ftick, bool urgent)
 		fuse_ms_push(ftick);
 	wakeup_one(ftick->tk_data);
 	selwakeuppri(&ftick->tk_data->ks_rsel, PZERO + 1);
+	KNOTE_LOCKED(&ftick->tk_data->ks_rsel.si_note, 0);
 	fuse_lck_mtx_unlock(ftick->tk_data->ms_mtx);
 }
 

tests/sys/fs/fusefs/Makefile

@@ -13,6 +13,7 @@ GTESTS+=	create
 GTESTS+=	default_permissions
 GTESTS+=	default_permissions_privileged
 GTESTS+=	destroy
+GTESTS+=	dev_fuse_poll
 GTESTS+=	fifo
 GTESTS+=	flush
 GTESTS+=	fsync

tests/sys/fs/fusefs/destroy.cc

@@ -33,23 +33,7 @@
 
 using namespace testing;
 
-class Destroy: public FuseTest {
-public:
-void expect_destroy(int error)
-{
-	EXPECT_CALL(*m_mock, process(
-		ResultOf([=](auto in) {
-			return (in->header.opcode == FUSE_DESTROY);
-		}, Eq(true)),
-		_)
-	).WillOnce(Invoke( ReturnImmediate([&](auto in, auto out) {
-		m_mock->m_quit = true;
-		out->header.len = sizeof(out->header);
-		out->header.unique = in->header.unique;
-		out->header.error = -error;
-	})));}
-
-};
+class Destroy: public FuseTest {};
 
 /*
  * On unmount the kernel should send a FUSE_DESTROY operation.  It should also

tests/sys/fs/fusefs/dev_fuse_poll.cc

@@ -0,0 +1,93 @@
+/*-
+ * SPDX-License-Identifier: BSD-2-Clause-FreeBSD
+ *
+ * Copyright (c) 2019 The FreeBSD Foundation
+ *
+ * This software was developed by BFF Storage Systems, LLC under sponsorship
+ * from the FreeBSD Foundation.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions
+ * are met:
+ * 1. Redistributions of source code must retain the above copyright
+ *    notice, this list of conditions and the following disclaimer.
+ * 2. Redistributions in binary form must reproduce the above copyright
+ *    notice, this list of conditions and the following disclaimer in the
+ *    documentation and/or other materials provided with the distribution.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+ * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+ * ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
+ * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
+ * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
+ * OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
+ * HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
+ * LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
+ * OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
+ * SUCH DAMAGE.
+ */
+
+/*
+ * This file tests different polling methods for the /dev/fuse device
+ */
+
+extern "C" {
+#include <fcntl.h>
+#include <unistd.h>
+}
+
+#include "mockfs.hh"
+#include "utils.hh"
+
+using namespace testing;
+
+const char FULLPATH[] = "mountpoint/some_file.txt";
+const char RELPATH[] = "some_file.txt";
+const uint64_t ino = 42;
+const mode_t access_mode = R_OK;
+
+/*
+ * Translate a poll method's string representation to the enum value.
+ * Using strings with ::testing::Values gives better output with
+ * --gtest_list_tests
+ */
+enum poll_method poll_method_from_string(const char *s)
+{
+	if (0 == strcmp("BLOCKING", s))
+		return BLOCKING;
+	else if (0 == strcmp("KQ", s))
+		return KQ;
+	else if (0 == strcmp("POLL", s))
+		return POLL;
+	else
+		return SELECT;
+}
+
+class DevFusePoll: public FuseTest, public WithParamInterface<const char *> {
+	virtual void SetUp() {
+		m_pm = poll_method_from_string(GetParam());
+		FuseTest::SetUp();
+	}
+};
+
+TEST_P(DevFusePoll, access)
+{
+	expect_access(1, X_OK, 0);
+	expect_lookup(RELPATH, ino, S_IFREG | 0644, 0, 1);
+	expect_access(ino, access_mode, 0);
+
+	ASSERT_EQ(0, access(FULLPATH, access_mode)) << strerror(errno);
+}
+
+/* Ensure that we wake up pollers during unmount */
+TEST_P(DevFusePoll, destroy)
+{
+	expect_forget(1, 1);
+	expect_destroy(0);
+
+	m_mock->unmount();
+}
+
+INSTANTIATE_TEST_CASE_P(PM, DevFusePoll,
+		::testing::Values("BLOCKING", "KQ", "POLL", "SELECT"));

tests/sys/fs/fusefs/mockfs.cc

@@ -32,12 +32,14 @@ extern "C" {
 #include <sys/param.h>
 
 #include <sys/mount.h>
+#include <sys/select.h>
 #include <sys/stat.h>
 #include <sys/uio.h>
 #include <sys/user.h>
 
 #include <fcntl.h>
 #include <libutil.h>
+#include <poll.h>
 #include <pthread.h>
 #include <signal.h>
 #include <stdlib.h>
@@ -282,7 +284,7 @@ void debug_fuseop(const mockfs_buf_in *in)
 }
 
 MockFS::MockFS(int max_readahead, bool allow_other, bool default_permissions,
-	bool push_symlinks_in, bool ro, uint32_t flags)
+	bool push_symlinks_in, bool ro, enum poll_method pm, uint32_t flags)
 {
 	struct sigaction sa;
 	struct iovec *iov = NULL;
@@ -292,7 +294,12 @@ MockFS::MockFS(int max_readahead, bool allow_other, bool default_permissions,
 
 	m_daemon_id = NULL;
 	m_maxreadahead = max_readahead;
+	m_pm = pm;
 	m_quit = false;
+	if (m_pm == KQ)
+		m_kq = kqueue();
+	else
+		m_kq = -1;
 
 	/*
 	 * Kyua sets pwd to a testcase-unique tempdir; no need to use
@@ -306,11 +313,17 @@ MockFS::MockFS(int max_readahead, bool allow_other, bool default_permissions,
 		throw(std::system_error(errno, std::system_category(),
 			"Couldn't make mountpoint directory"));
 
-	m_fuse_fd = open("/dev/fuse", O_CLOEXEC | O_RDWR);
+	switch (m_pm) {
+	case BLOCKING:
+		m_fuse_fd = open("/dev/fuse", O_CLOEXEC | O_RDWR);
+		break;
+	default:
+		m_fuse_fd = open("/dev/fuse", O_CLOEXEC | O_RDWR | O_NONBLOCK);
+		break;
+	}
 	if (m_fuse_fd < 0)
 		throw(std::system_error(errno, std::system_category(),
 			"Couldn't open /dev/fuse"));
-	sprintf(fdstr, "%d", m_fuse_fd);
 
 	m_pid = getpid();
 	m_child_pid = -1;
@@ -319,6 +332,7 @@ MockFS::MockFS(int max_readahead, bool allow_other, bool default_permissions,
 	build_iovec(&iov, &iovlen, "fspath",
 		    __DECONST(void *, "mountpoint"), -1);
 	build_iovec(&iov, &iovlen, "from", __DECONST(void *, "/dev/fuse"), -1);
+	sprintf(fdstr, "%d", m_fuse_fd);
 	build_iovec(&iov, &iovlen, "fd", fdstr, -1);
 	if (allow_other) {
 		build_iovec(&iov, &iovlen, "allow_other",
@@ -364,6 +378,8 @@ MockFS::~MockFS() {
 	}
 	::unmount("mountpoint", MNT_FORCE);
 	rmdir("mountpoint");
+	if (m_kq >= 0)
+		close(m_kq);
 }
 
 void MockFS::init(uint32_t flags) {
@@ -440,9 +456,7 @@ void MockFS::loop() {
 			process_default(in, out);
 		}
 		for (auto &it: out) {
-			ASSERT_TRUE(write(m_fuse_fd, it, it->header.len) > 0 ||
-				    errno == EAGAIN)
-				<< strerror(errno);
+			write_response(it);
 			delete it;
 		}
 		out.clear();
@@ -485,13 +499,94 @@ void MockFS::process_default(const mockfs_buf_in *in,
 
 void MockFS::read_request(mockfs_buf_in *in) {
 	ssize_t res;
+	int nready;
+	fd_set readfds;
+	pollfd fds[1];
+	struct kevent changes[1];
+	struct kevent events[1];
+	int nfds;
 
+	switch (m_pm) {
+	case BLOCKING:
+		break;
+	case KQ:
+		EV_SET(&changes[0], m_fuse_fd, EVFILT_READ, EV_ADD, 0, 0, 0);
+		nready = kevent(m_kq, &changes[0], 1, &events[0], 1, NULL);
+		if (m_quit)
+			return;
+		ASSERT_LE(0, nready) << strerror(errno);
+		ASSERT_EQ(1, nready) << "NULL timeout expired?";
+		ASSERT_EQ(events[0].ident, (uintptr_t)m_fuse_fd);
+		if (events[0].flags & EV_ERROR)
+			FAIL() << strerror(events[0].data);
+		else if (events[0].flags & EV_EOF)
+			FAIL() << strerror(events[0].fflags);
+		break;
+	case POLL:
+		fds[0].fd = m_fuse_fd;
+		fds[0].events = POLLIN;
+		nready = poll(fds, 1, INFTIM);
+		if (m_quit)
+			return;
+		ASSERT_LE(0, nready) << strerror(errno);
+		ASSERT_EQ(1, nready) << "NULL timeout expired?";
+		ASSERT_TRUE(fds[0].revents & POLLIN);
+		break;
+	case SELECT:
+		FD_ZERO(&readfds);
+		FD_SET(m_fuse_fd, &readfds);
+		nfds = m_fuse_fd + 1;
+		nready = select(nfds, &readfds, NULL, NULL, NULL);
+		if (m_quit)
+			return;
+		ASSERT_LE(0, nready) << strerror(errno);
+		ASSERT_EQ(1, nready) << "NULL timeout expired?";
+		ASSERT_TRUE(FD_ISSET(m_fuse_fd, &readfds));
+		break;
+	default:
+		FAIL() << "not yet implemented";
+	}
 	res = read(m_fuse_fd, in, sizeof(*in));
+
 	if (res < 0 && !m_quit)
 		perror("read");
 	ASSERT_TRUE(res >= (ssize_t)sizeof(in->header) || m_quit);
 }
 
+void MockFS::write_response(mockfs_buf_out *out) {
+	fd_set writefds;
+	pollfd fds[1];
+	int nready, nfds;
+	ssize_t r;
+
+	switch (m_pm) {
+	case BLOCKING:
+	case KQ:	/* EVFILT_WRITE is not supported */
+		break;
+	case POLL:
+		fds[0].fd = m_fuse_fd;
+		fds[0].events = POLLOUT;
+		nready = poll(fds, 1, INFTIM);
+		ASSERT_LE(0, nready) << strerror(errno);
+		ASSERT_EQ(1, nready) << "NULL timeout expired?";
+		ASSERT_TRUE(fds[0].revents & POLLOUT);
+		break;
+	case SELECT:
+		FD_ZERO(&writefds);
+		FD_SET(m_fuse_fd, &writefds);
+		nfds = m_fuse_fd + 1;
+		nready = select(nfds, NULL, &writefds, NULL, NULL);
+		ASSERT_LE(0, nready) << strerror(errno);
+		ASSERT_EQ(1, nready) << "NULL timeout expired?";
+		ASSERT_TRUE(FD_ISSET(m_fuse_fd, &writefds));
+		break;
+	default:
+		FAIL() << "not yet implemented";
+	}
+	r = write(m_fuse_fd, out, out->header.len);
+	ASSERT_TRUE(r > 0 || errno == EAGAIN) << strerror(errno);
+}
+
 void* MockFS::service(void *pthr_data) {
 	MockFS *mock_fs = (MockFS*)pthr_data;
 

tests/sys/fs/fusefs/utils.cc

@@ -96,7 +96,7 @@ void FuseTest::SetUp() {
 	try {
 		m_mock = new MockFS(m_maxreadahead, m_allow_other,
 			m_default_permissions, m_push_symlinks_in, m_ro,
-			m_init_flags);
+			m_pm, m_init_flags);
 		/* 
 		 * FUSE_ACCESS is called almost universally.  Expecting it in
 		 * each test case would be super-annoying.  Instead, set a
@@ -130,6 +130,22 @@ FuseTest::expect_access(uint64_t ino, mode_t access_mode, int error)
 	).WillOnce(Invoke(ReturnErrno(error)));
 }
 
+void
+FuseTest::expect_destroy(int error)
+{
+	EXPECT_CALL(*m_mock, process(
+		ResultOf([=](auto in) {
+			return (in->header.opcode == FUSE_DESTROY);
+		}, Eq(true)),
+		_)
+	).WillOnce(Invoke( ReturnImmediate([&](auto in, auto out) {
+		m_mock->m_quit = true;
+		out->header.len = sizeof(out->header);
+		out->header.unique = in->header.unique;
+		out->header.error = -error;
+	})));
+}
+
 void
 FuseTest::expect_flush(uint64_t ino, int times, ProcessMockerT r)
 {

tests/sys/fs/fusefs/utils.hh

@@ -52,6 +52,7 @@ class FuseTest : public ::testing::Test {
 	uint32_t m_init_flags;
 	bool m_allow_other;
 	bool m_default_permissions;
+	enum poll_method m_pm;
 	bool m_push_symlinks_in;
 	bool m_ro;
 	MockFS *m_mock = NULL;
@@ -69,6 +70,7 @@ class FuseTest : public ::testing::Test {
 		m_init_flags(0),
 		m_allow_other(false),
 		m_default_permissions(false),
+		m_pm(BLOCKING),
 		m_push_symlinks_in(false),
 		m_ro(false)
 	{}
@@ -86,6 +88,9 @@ class FuseTest : public ::testing::Test {
 	 */
 	void expect_access(uint64_t ino, mode_t access_mode, int error);
 
+	/* Expect FUSE_DESTROY and shutdown the daemon */
+	void expect_destroy(int error);
+
 	/*
 	 * Create an expectation that FUSE_FLUSH will be called times times for
 	 * the given inode