[PATCH v3 0/5] vhost: FD manager improvements

[PATCH v3 0/5] vhost: FD manager improvements

[Maxime Coquelin]

This series aims at improving the Vhost FD manager.
The fdset internals is hidden from its callers, the
notification/synchronization is made systematic for
every FD add/del operations and finally the code is
reworked to make use of epoll insteal of poll which
reduces the complexity (less locks involved,
synchrnoization no more necessary).

The series has been tested with both Vhost-user/Virtio-user
and with VDUSE.

Only change in this revision is removing mistakenly
added temporary file in patch 3.

David Marchand (1):
  vhost: manage FD with epoll

Maxime Coquelin (4):
  vhost: rename polling mutex
  vhost: make use of FD manager init function
  vhost: hide synchronization within FD manager
  vhost: improve fdset initialization

 lib/vhost/fd_man.c | 431 +++++++++++++++++++++------------------------
 lib/vhost/fd_man.h |  48 +----
 lib/vhost/socket.c |  38 +---
 lib/vhost/vduse.c  |  52 ++----
 4 files changed, 224 insertions(+), 345 deletions(-)

-- 
2.44.0

[PATCH v3 1/5] vhost: rename polling mutex

[Maxime Coquelin]

This trivial patch fixes a typo in fd's manager polling
mutex name.

Reviewed-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
---
 lib/vhost/fd_man.c | 8 ++++----
 lib/vhost/fd_man.h | 2 +-
 2 files changed, 5 insertions(+), 5 deletions(-)

diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
index 481e6b900a..67ee1589e1 100644
--- a/lib/vhost/fd_man.c
+++ b/lib/vhost/fd_man.c
@@ -125,9 +125,9 @@ fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
 	pthread_mutex_lock(&pfdset->fd_mutex);
 	i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
 	if (i == -1) {
-		pthread_mutex_lock(&pfdset->fd_pooling_mutex);
+		pthread_mutex_lock(&pfdset->fd_polling_mutex);
 		fdset_shrink_nolock(pfdset);
-		pthread_mutex_unlock(&pfdset->fd_pooling_mutex);
+		pthread_mutex_unlock(&pfdset->fd_polling_mutex);
 		i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
 		if (i == -1) {
 			pthread_mutex_unlock(&pfdset->fd_mutex);
@@ -244,9 +244,9 @@ fdset_event_dispatch(void *arg)
 		numfds = pfdset->num;
 		pthread_mutex_unlock(&pfdset->fd_mutex);
 
-		pthread_mutex_lock(&pfdset->fd_pooling_mutex);
+		pthread_mutex_lock(&pfdset->fd_polling_mutex);
 		val = poll(pfdset->rwfds, numfds, 1000 /* millisecs */);
-		pthread_mutex_unlock(&pfdset->fd_pooling_mutex);
+		pthread_mutex_unlock(&pfdset->fd_polling_mutex);
 		if (val < 0)
 			continue;
 
diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
index 7816fb11ac..4e00f94758 100644
--- a/lib/vhost/fd_man.h
+++ b/lib/vhost/fd_man.h
@@ -24,7 +24,7 @@ struct fdset {
 	struct pollfd rwfds[MAX_FDS];
 	struct fdentry fd[MAX_FDS];
 	pthread_mutex_t fd_mutex;
-	pthread_mutex_t fd_pooling_mutex;
+	pthread_mutex_t fd_polling_mutex;
 	int num;	/* current fd number of this fdset */
 
 	union pipefds {
-- 
2.44.0

[PATCH v3 2/5] vhost: make use of FD manager init function

[Maxime Coquelin]

Instead of statically initialize the fdset, this patch
converts VDUSE and Vhost-user to use fdset_init() function,
which now also initialize the mutexes.

This is preliminary rework to hide FDs manager pipe from
its users.

Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
---
 lib/vhost/fd_man.c | 11 +++++++++--
 lib/vhost/fd_man.h |  2 +-
 lib/vhost/socket.c | 12 +++++-------
 lib/vhost/vduse.c  | 15 ++++++---------
 4 files changed, 21 insertions(+), 19 deletions(-)

diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
index 67ee1589e1..e42c491fdb 100644
--- a/lib/vhost/fd_man.c
+++ b/lib/vhost/fd_man.c
@@ -96,19 +96,26 @@ fdset_add_fd(struct fdset *pfdset, int idx, int fd,
 	pfd->revents = 0;
 }
 
-void
+int
 fdset_init(struct fdset *pfdset)
 {
 	int i;
 
 	if (pfdset == NULL)
-		return;
+		return -1;
+
+	pthread_mutex_init(&pfdset->fd_mutex, NULL);
+	pthread_mutex_init(&pfdset->fd_polling_mutex, NULL);
+	pthread_mutex_init(&pfdset->sync_mutex, NULL);
+	pthread_cond_init(&pfdset->sync_cond, NULL);
 
 	for (i = 0; i < MAX_FDS; i++) {
 		pfdset->fd[i].fd = -1;
 		pfdset->fd[i].dat = NULL;
 	}
 	pfdset->num = 0;
+
+	return 0;
 }
 
 /**
diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
index 4e00f94758..0f4cddfe56 100644
--- a/lib/vhost/fd_man.h
+++ b/lib/vhost/fd_man.h
@@ -43,7 +43,7 @@ struct fdset {
 };
 
 
-void fdset_init(struct fdset *pfdset);
+int fdset_init(struct fdset *pfdset);
 
 int fdset_add(struct fdset *pfdset, int fd,
 	fd_cb rcb, fd_cb wcb, void *dat);
diff --git a/lib/vhost/socket.c b/lib/vhost/socket.c
index 96b3ab5595..8155749972 100644
--- a/lib/vhost/socket.c
+++ b/lib/vhost/socket.c
@@ -89,13 +89,6 @@ static int create_unix_socket(struct vhost_user_socket *vsocket);
 static int vhost_user_start_client(struct vhost_user_socket *vsocket);
 
 static struct vhost_user vhost_user = {
-	.fdset = {
-		.fd = { [0 ... MAX_FDS - 1] = {-1, NULL, NULL, NULL, 0} },
-		.fd_mutex = PTHREAD_MUTEX_INITIALIZER,
-		.fd_pooling_mutex = PTHREAD_MUTEX_INITIALIZER,
-		.sync_mutex = PTHREAD_MUTEX_INITIALIZER,
-		.num = 0
-	},
 	.vsocket_cnt = 0,
 	.mutex = PTHREAD_MUTEX_INITIALIZER,
 };
@@ -1188,6 +1181,11 @@ rte_vhost_driver_start(const char *path)
 		return vduse_device_create(path, vsocket->net_compliant_ol_flags);
 
 	if (fdset_tid.opaque_id == 0) {
+		if (fdset_init(&vhost_user.fdset) < 0) {
+			VHOST_CONFIG_LOG(path, ERR, "failed to init Vhost-user fdset");
+			return -1;
+		}
+
 		/**
 		 * create a pipe which will be waited by poll and notified to
 		 * rebuild the wait list of poll.
diff --git a/lib/vhost/vduse.c b/lib/vhost/vduse.c
index e0c6991b69..530c148399 100644
--- a/lib/vhost/vduse.c
+++ b/lib/vhost/vduse.c
@@ -31,15 +31,7 @@ struct vduse {
 	struct fdset fdset;
 };
 
-static struct vduse vduse = {
-	.fdset = {
-		.fd = { [0 ... MAX_FDS - 1] = {-1, NULL, NULL, NULL, 0} },
-		.fd_mutex = PTHREAD_MUTEX_INITIALIZER,
-		.fd_pooling_mutex = PTHREAD_MUTEX_INITIALIZER,
-		.sync_mutex = PTHREAD_MUTEX_INITIALIZER,
-		.num = 0
-	},
-};
+static struct vduse vduse;
 
 static bool vduse_events_thread;
 
@@ -435,6 +427,11 @@ vduse_device_create(const char *path, bool compliant_ol_flags)
 
 	/* If first device, create events dispatcher thread */
 	if (vduse_events_thread == false) {
+		if (fdset_init(&vduse.fdset) < 0) {
+			VHOST_CONFIG_LOG(path, ERR, "failed to init VDUSE fdset");
+			return -1;
+		}
+
 		/**
 		 * create a pipe which will be waited by poll and notified to
 		 * rebuild the wait list of poll.
-- 
2.44.0

Re: [PATCH v3 2/5] vhost: make use of FD manager init function

[Stephen Hemminger]

On Tue,  9 Apr 2024 13:48:42 +0200
Maxime Coquelin <maxime.coquelin@redhat.com> wrote:

> -void
> +int
>  fdset_init(struct fdset *pfdset)
>  {
>  	int i;
>  
>  	if (pfdset == NULL)
> -		return;
> +		return -1;

This test is unnecessary. The function is not exported, and therefore
can only be called from vhost library. And all call sites are passing
a pointer to local data.

Adding extra parameter checks to internal functions bloats code
and creates lots of untestable paths. Coverity might even be
smart enough to detect these.

Re: [PATCH v3 2/5] vhost: make use of FD manager init function

[Maxime Coquelin]

On 4/9/24 18:38, Stephen Hemminger wrote:
> On Tue,  9 Apr 2024 13:48:42 +0200
> Maxime Coquelin <maxime.coquelin@redhat.com> wrote:
> 
>> -void
>> +int
>>   fdset_init(struct fdset *pfdset)
>>   {
>>   	int i;
>>   
>>   	if (pfdset == NULL)
>> -		return;
>> +		return -1;
> 
> This test is unnecessary. The function is not exported, and therefore
> can only be called from vhost library. And all call sites are passing
> a pointer to local data.
> 
> Adding extra parameter checks to internal functions bloats code
> and creates lots of untestable paths. Coverity might even be
> smart enough to detect these.
> 

Agree. I will remove the test in next revision.

Thanks,
Maxime

[PATCH v3 3/5] vhost: hide synchronization within FD manager

[Maxime Coquelin]

This patch forces synchronization for all FDs additions
or deletions in the FD set. With that, it is no more
necessary for the user to know about the FD set pipe, so
hide its initialization in the FD manager.

Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
---
 lib/vhost/fd_man.c | 180 ++++++++++++++++++++++++---------------------
 lib/vhost/fd_man.h |   8 +-
 lib/vhost/socket.c |  12 +--
 lib/vhost/vduse.c  |  18 +----
 4 files changed, 101 insertions(+), 117 deletions(-)

diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
index e42c491fdb..0ae481b785 100644
--- a/lib/vhost/fd_man.c
+++ b/lib/vhost/fd_man.c
@@ -2,7 +2,9 @@
  * Copyright(c) 2010-2014 Intel Corporation
  */
 
+#include <errno.h>
 #include <stdio.h>
+#include <string.h>
 #include <unistd.h>
 
 #include <rte_common.h>
@@ -17,6 +19,87 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
 
 #define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
 
+static void
+fdset_pipe_read_cb(int readfd, void *dat,
+		   int *remove __rte_unused)
+{
+	char charbuf[16];
+	struct fdset *fdset = dat;
+	int r = read(readfd, charbuf, sizeof(charbuf));
+	/*
+	 * Just an optimization, we don't care if read() failed
+	 * so ignore explicitly its return value to make the
+	 * compiler happy
+	 */
+	RTE_SET_USED(r);
+
+	pthread_mutex_lock(&fdset->sync_mutex);
+	fdset->sync = true;
+	pthread_cond_broadcast(&fdset->sync_cond);
+	pthread_mutex_unlock(&fdset->sync_mutex);
+}
+
+static void
+fdset_pipe_uninit(struct fdset *fdset)
+{
+	fdset_del(fdset, fdset->u.readfd);
+	close(fdset->u.readfd);
+	fdset->u.readfd = -1;
+	close(fdset->u.writefd);
+	fdset->u.writefd = -1;
+}
+
+static int
+fdset_pipe_init(struct fdset *fdset)
+{
+	int ret;
+
+	pthread_mutex_init(&fdset->sync_mutex, NULL);
+	pthread_cond_init(&fdset->sync_cond, NULL);
+
+	if (pipe(fdset->u.pipefd) < 0) {
+		VHOST_FDMAN_LOG(ERR,
+			"failed to create pipe for vhost fdset");
+		return -1;
+	}
+
+	ret = fdset_add(fdset, fdset->u.readfd,
+			fdset_pipe_read_cb, NULL, fdset);
+	if (ret < 0) {
+		VHOST_FDMAN_LOG(ERR,
+			"failed to add pipe readfd %d into vhost server fdset",
+			fdset->u.readfd);
+
+		fdset_pipe_uninit(fdset);
+		return -1;
+	}
+
+	return 0;
+}
+
+static void
+fdset_sync(struct fdset *fdset)
+{
+	int ret;
+
+	pthread_mutex_lock(&fdset->sync_mutex);
+
+	fdset->sync = false;
+	ret = write(fdset->u.writefd, "1", 1);
+	if (ret < 0) {
+		VHOST_FDMAN_LOG(ERR,
+			"Failed to write to notification pipe: %s",
+			strerror(errno));
+		goto out_unlock;
+	}
+
+	while (!fdset->sync)
+		pthread_cond_wait(&fdset->sync_cond, &fdset->sync_mutex);
+
+out_unlock:
+	pthread_mutex_unlock(&fdset->sync_mutex);
+}
+
 static int
 get_last_valid_idx(struct fdset *pfdset, int last_valid_idx)
 {
@@ -96,6 +179,12 @@ fdset_add_fd(struct fdset *pfdset, int idx, int fd,
 	pfd->revents = 0;
 }
 
+void
+fdset_uninit(struct fdset *pfdset)
+{
+	fdset_pipe_uninit(pfdset);
+}
+
 int
 fdset_init(struct fdset *pfdset)
 {
@@ -106,8 +195,6 @@ fdset_init(struct fdset *pfdset)
 
 	pthread_mutex_init(&pfdset->fd_mutex, NULL);
 	pthread_mutex_init(&pfdset->fd_polling_mutex, NULL);
-	pthread_mutex_init(&pfdset->sync_mutex, NULL);
-	pthread_cond_init(&pfdset->sync_cond, NULL);
 
 	for (i = 0; i < MAX_FDS; i++) {
 		pfdset->fd[i].fd = -1;
@@ -115,7 +202,7 @@ fdset_init(struct fdset *pfdset)
 	}
 	pfdset->num = 0;
 
-	return 0;
+	return fdset_pipe_init(pfdset);
 }
 
 /**
@@ -145,6 +232,8 @@ fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
 	fdset_add_fd(pfdset, i, fd, rcb, wcb, dat);
 	pthread_mutex_unlock(&pfdset->fd_mutex);
 
+	fdset_sync(pfdset);
+
 	return 0;
 }
 
@@ -176,6 +265,8 @@ fdset_del(struct fdset *pfdset, int fd)
 		pthread_mutex_unlock(&pfdset->fd_mutex);
 	} while (i != -1);
 
+	fdset_sync(pfdset);
+
 	return dat;
 }
 
@@ -209,6 +300,9 @@ fdset_try_del(struct fdset *pfdset, int fd)
 	}
 
 	pthread_mutex_unlock(&pfdset->fd_mutex);
+
+	fdset_sync(pfdset);
+
 	return 0;
 }
 
@@ -314,83 +408,3 @@ fdset_event_dispatch(void *arg)
 
 	return 0;
 }
-
-static void
-fdset_pipe_read_cb(int readfd, void *dat,
-		   int *remove __rte_unused)
-{
-	char charbuf[16];
-	struct fdset *fdset = dat;
-	int r = read(readfd, charbuf, sizeof(charbuf));
-	/*
-	 * Just an optimization, we don't care if read() failed
-	 * so ignore explicitly its return value to make the
-	 * compiler happy
-	 */
-	RTE_SET_USED(r);
-
-	pthread_mutex_lock(&fdset->sync_mutex);
-	fdset->sync = true;
-	pthread_cond_broadcast(&fdset->sync_cond);
-	pthread_mutex_unlock(&fdset->sync_mutex);
-}
-
-void
-fdset_pipe_uninit(struct fdset *fdset)
-{
-	fdset_del(fdset, fdset->u.readfd);
-	close(fdset->u.readfd);
-	close(fdset->u.writefd);
-}
-
-int
-fdset_pipe_init(struct fdset *fdset)
-{
-	int ret;
-
-	if (pipe(fdset->u.pipefd) < 0) {
-		VHOST_FDMAN_LOG(ERR,
-			"failed to create pipe for vhost fdset");
-		return -1;
-	}
-
-	ret = fdset_add(fdset, fdset->u.readfd,
-			fdset_pipe_read_cb, NULL, fdset);
-
-	if (ret < 0) {
-		VHOST_FDMAN_LOG(ERR,
-			"failed to add pipe readfd %d into vhost server fdset",
-			fdset->u.readfd);
-
-		fdset_pipe_uninit(fdset);
-		return -1;
-	}
-
-	return 0;
-}
-
-void
-fdset_pipe_notify(struct fdset *fdset)
-{
-	int r = write(fdset->u.writefd, "1", 1);
-	/*
-	 * Just an optimization, we don't care if write() failed
-	 * so ignore explicitly its return value to make the
-	 * compiler happy
-	 */
-	RTE_SET_USED(r);
-}
-
-void
-fdset_pipe_notify_sync(struct fdset *fdset)
-{
-	pthread_mutex_lock(&fdset->sync_mutex);
-
-	fdset->sync = false;
-	fdset_pipe_notify(fdset);
-
-	while (!fdset->sync)
-		pthread_cond_wait(&fdset->sync_cond, &fdset->sync_mutex);
-
-	pthread_mutex_unlock(&fdset->sync_mutex);
-}
diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
index 0f4cddfe56..c18e3a435c 100644
--- a/lib/vhost/fd_man.h
+++ b/lib/vhost/fd_man.h
@@ -42,6 +42,7 @@ struct fdset {
 	bool sync;
 };
 
+void fdset_uninit(struct fdset *pfdset);
 
 int fdset_init(struct fdset *pfdset);
 
@@ -53,11 +54,4 @@ int fdset_try_del(struct fdset *pfdset, int fd);
 
 uint32_t fdset_event_dispatch(void *arg);
 
-int fdset_pipe_init(struct fdset *fdset);
-
-void fdset_pipe_uninit(struct fdset *fdset);
-
-void fdset_pipe_notify(struct fdset *fdset);
-void fdset_pipe_notify_sync(struct fdset *fdset);
-
 #endif
diff --git a/lib/vhost/socket.c b/lib/vhost/socket.c
index 8155749972..5afb952a21 100644
--- a/lib/vhost/socket.c
+++ b/lib/vhost/socket.c
@@ -278,7 +278,6 @@ vhost_user_add_connection(int fd, struct vhost_user_socket *vsocket)
 	TAILQ_INSERT_TAIL(&vsocket->conn_list, conn, next);
 	pthread_mutex_unlock(&vsocket->conn_mutex);
 
-	fdset_pipe_notify(&vhost_user.fdset);
 	return;
 
 err_cleanup:
@@ -1186,20 +1185,11 @@ rte_vhost_driver_start(const char *path)
 			return -1;
 		}
 
-		/**
-		 * create a pipe which will be waited by poll and notified to
-		 * rebuild the wait list of poll.
-		 */
-		if (fdset_pipe_init(&vhost_user.fdset) < 0) {
-			VHOST_CONFIG_LOG(path, ERR, "failed to create pipe for vhost fdset");
-			return -1;
-		}
-
 		int ret = rte_thread_create_internal_control(&fdset_tid,
 				"vhost-evt", fdset_event_dispatch, &vhost_user.fdset);
 		if (ret != 0) {
 			VHOST_CONFIG_LOG(path, ERR, "failed to create fdset handling thread");
-			fdset_pipe_uninit(&vhost_user.fdset);
+			fdset_uninit(&vhost_user.fdset);
 			return -1;
 		}
 	}
diff --git a/lib/vhost/vduse.c b/lib/vhost/vduse.c
index 530c148399..d87fc500d4 100644
--- a/lib/vhost/vduse.c
+++ b/lib/vhost/vduse.c
@@ -225,7 +225,6 @@ vduse_vring_setup(struct virtio_net *dev, unsigned int index)
 			close(vq->kickfd);
 			vq->kickfd = VIRTIO_UNINITIALIZED_EVENTFD;
 		}
-		fdset_pipe_notify(&vduse.fdset);
 		vhost_enable_guest_notification(dev, vq, 1);
 		VHOST_CONFIG_LOG(dev->ifname, INFO, "Ctrl queue event handler installed");
 	}
@@ -238,10 +237,8 @@ vduse_vring_cleanup(struct virtio_net *dev, unsigned int index)
 	struct vduse_vq_eventfd vq_efd;
 	int ret;
 
-	if (vq == dev->cvq && vq->kickfd >= 0) {
+	if (vq == dev->cvq && vq->kickfd >= 0)
 		fdset_del(&vduse.fdset, vq->kickfd);
-		fdset_pipe_notify(&vduse.fdset);
-	}
 
 	vq_efd.index = index;
 	vq_efd.fd = VDUSE_EVENTFD_DEASSIGN;
@@ -432,20 +429,11 @@ vduse_device_create(const char *path, bool compliant_ol_flags)
 			return -1;
 		}
 
-		/**
-		 * create a pipe which will be waited by poll and notified to
-		 * rebuild the wait list of poll.
-		 */
-		if (fdset_pipe_init(&vduse.fdset) < 0) {
-			VHOST_CONFIG_LOG(path, ERR, "failed to create pipe for vduse fdset");
-			return -1;
-		}
-
 		ret = rte_thread_create_internal_control(&fdset_tid, "vduse-evt",
 				fdset_event_dispatch, &vduse.fdset);
 		if (ret != 0) {
 			VHOST_CONFIG_LOG(path, ERR, "failed to create vduse fdset handling thread");
-			fdset_pipe_uninit(&vduse.fdset);
+			fdset_uninit(&vduse.fdset);
 			return -1;
 		}
 
@@ -573,7 +561,6 @@ vduse_device_create(const char *path, bool compliant_ol_flags)
 				dev->vduse_dev_fd);
 		goto out_dev_destroy;
 	}
-	fdset_pipe_notify(&vduse.fdset);
 
 	free(dev_config);
 
@@ -616,7 +603,6 @@ vduse_device_destroy(const char *path)
 	vduse_device_stop(dev);
 
 	fdset_del(&vduse.fdset, dev->vduse_dev_fd);
-	fdset_pipe_notify_sync(&vduse.fdset);
 
 	if (dev->vduse_dev_fd >= 0) {
 		close(dev->vduse_dev_fd);
-- 
2.44.0

[PATCH v3 4/5] vhost: improve fdset initialization

[Maxime Coquelin]

This patch heavily reworks fdset initialization:
 - fdsets are now dynamically allocated by the FD manager
 - the event dispatcher is now created by the FD manager
 - struct fdset is now opaque to VDUSE and Vhost

Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
---
 lib/vhost/fd_man.c | 177 ++++++++++++++++++++++++++++++++++++++++-----
 lib/vhost/fd_man.h |  39 +---------
 lib/vhost/socket.c |  24 ++----
 lib/vhost/vduse.c  |  29 ++------
 4 files changed, 177 insertions(+), 92 deletions(-)

diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
index 0ae481b785..8b47c97d45 100644
--- a/lib/vhost/fd_man.c
+++ b/lib/vhost/fd_man.c
@@ -3,12 +3,16 @@
  */
 
 #include <errno.h>
+#include <pthread.h>
 #include <stdio.h>
 #include <string.h>
 #include <unistd.h>
 
 #include <rte_common.h>
 #include <rte_log.h>
+#include <rte_malloc.h>
+#include <rte_string_fns.h>
+#include <rte_thread.h>
 
 #include "fd_man.h"
 
@@ -19,6 +23,79 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
 
 #define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
 
+struct fdentry {
+	int fd;		/* -1 indicates this entry is empty */
+	fd_cb rcb;	/* callback when this fd is readable. */
+	fd_cb wcb;	/* callback when this fd is writeable.*/
+	void *dat;	/* fd context */
+	int busy;	/* whether this entry is being used in cb. */
+};
+
+struct fdset {
+	char name[RTE_THREAD_NAME_SIZE];
+	struct pollfd rwfds[MAX_FDS];
+	struct fdentry fd[MAX_FDS];
+	rte_thread_t tid;
+	pthread_mutex_t fd_mutex;
+	pthread_mutex_t fd_polling_mutex;
+	int num;	/* current fd number of this fdset */
+
+	union pipefds {
+		struct {
+			int pipefd[2];
+		};
+		struct {
+			int readfd;
+			int writefd;
+		};
+	} u;
+
+	pthread_mutex_t sync_mutex;
+	pthread_cond_t sync_cond;
+	bool sync;
+	bool destroy;
+};
+
+static int fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat);
+static uint32_t fdset_event_dispatch(void *arg);
+
+#define MAX_FDSETS 8
+
+static struct fdset *fdsets[MAX_FDSETS];
+pthread_mutex_t fdsets_mutex = PTHREAD_MUTEX_INITIALIZER;
+
+static struct fdset *
+fdset_lookup(const char *name)
+{
+	int i;
+
+	for (i = 0; i < MAX_FDSETS; i++) {
+		struct fdset *fdset = fdsets[i];
+		if (fdset == NULL)
+			continue;
+
+		if (!strncmp(fdset->name, name, RTE_THREAD_NAME_SIZE))
+			return fdset;
+	}
+
+	return NULL;
+}
+
+static int
+fdset_insert(struct fdset *fdset)
+{
+	int i;
+
+	for (i = 0; i < MAX_FDSETS; i++) {
+		if (fdsets[i] == NULL) {
+			fdsets[i] = fdset;
+			return 0;
+		}
+	}
+
+	return -1;
+}
+
 static void
 fdset_pipe_read_cb(int readfd, void *dat,
 		   int *remove __rte_unused)
@@ -63,7 +140,7 @@ fdset_pipe_init(struct fdset *fdset)
 		return -1;
 	}
 
-	ret = fdset_add(fdset, fdset->u.readfd,
+	ret = fdset_add_no_sync(fdset, fdset->u.readfd,
 			fdset_pipe_read_cb, NULL, fdset);
 	if (ret < 0) {
 		VHOST_FDMAN_LOG(ERR,
@@ -179,37 +256,82 @@ fdset_add_fd(struct fdset *pfdset, int idx, int fd,
 	pfd->revents = 0;
 }
 
-void
-fdset_uninit(struct fdset *pfdset)
-{
-	fdset_pipe_uninit(pfdset);
-}
-
-int
-fdset_init(struct fdset *pfdset)
+struct fdset *
+fdset_init(const char *name)
 {
+	struct fdset *fdset;
+	uint32_t val;
 	int i;
 
-	if (pfdset == NULL)
-		return -1;
+	if (name == NULL) {
+		VHOST_FDMAN_LOG(ERR, "Invalid name");
+		goto err;
+	}
 
-	pthread_mutex_init(&pfdset->fd_mutex, NULL);
-	pthread_mutex_init(&pfdset->fd_polling_mutex, NULL);
+	pthread_mutex_lock(&fdsets_mutex);
+	fdset = fdset_lookup(name);
+	if (fdset) {
+		pthread_mutex_unlock(&fdsets_mutex);
+		return fdset;
+	}
+
+	fdset = rte_zmalloc(NULL, sizeof(*fdset), 0);
+	if (!fdset) {
+		VHOST_FDMAN_LOG(ERR, "Failed to alloc fdset %s", name);
+		goto err_unlock;
+	}
+
+	rte_strscpy(fdset->name, name, RTE_THREAD_NAME_SIZE);
+
+	pthread_mutex_init(&fdset->fd_mutex, NULL);
+	pthread_mutex_init(&fdset->fd_polling_mutex, NULL);
 
 	for (i = 0; i < MAX_FDS; i++) {
-		pfdset->fd[i].fd = -1;
-		pfdset->fd[i].dat = NULL;
+		fdset->fd[i].fd = -1;
+		fdset->fd[i].dat = NULL;
 	}
-	pfdset->num = 0;
+	fdset->num = 0;
 
-	return fdset_pipe_init(pfdset);
+	if (fdset_pipe_init(fdset)) {
+		VHOST_FDMAN_LOG(ERR, "Failed to init pipe for %s", name);
+		goto err_free;
+	}
+
+	if (rte_thread_create_internal_control(&fdset->tid, fdset->name,
+					fdset_event_dispatch, fdset)) {
+		VHOST_FDMAN_LOG(ERR, "Failed to create %s event dispatch thread",
+				fdset->name);
+		goto err_pipe;
+	}
+
+	if (fdset_insert(fdset)) {
+		VHOST_FDMAN_LOG(ERR, "Failed to insert fdset %s", name);
+		goto err_thread;
+	}
+
+	pthread_mutex_unlock(&fdsets_mutex);
+
+	return fdset;
+
+err_thread:
+	fdset->destroy = true;
+	fdset_sync(fdset);
+	rte_thread_join(fdset->tid, &val);
+err_pipe:
+	fdset_pipe_uninit(fdset);
+err_free:
+	rte_free(fdset);
+err_unlock:
+	pthread_mutex_unlock(&fdsets_mutex);
+err:
+	return NULL;
 }
 
 /**
  * Register the fd in the fdset with read/write handler and context.
  */
-int
-fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
+static int
+fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
 {
 	int i;
 
@@ -232,6 +354,18 @@ fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
 	fdset_add_fd(pfdset, i, fd, rcb, wcb, dat);
 	pthread_mutex_unlock(&pfdset->fd_mutex);
 
+	return 0;
+}
+
+int
+fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
+{
+	int ret;
+
+	ret = fdset_add_no_sync(pfdset, fd, rcb, wcb, dat);
+	if (ret < 0)
+		return ret;
+
 	fdset_sync(pfdset);
 
 	return 0;
@@ -315,7 +449,7 @@ fdset_try_del(struct fdset *pfdset, int fd)
  * will wait until the flag is reset to zero(which indicates the callback is
  * finished), then it could free the context after fdset_del.
  */
-uint32_t
+static uint32_t
 fdset_event_dispatch(void *arg)
 {
 	int i;
@@ -404,6 +538,9 @@ fdset_event_dispatch(void *arg)
 
 		if (need_shrink)
 			fdset_shrink(pfdset);
+
+		if (pfdset->destroy)
+			break;
 	}
 
 	return 0;
diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
index c18e3a435c..079fa0155f 100644
--- a/lib/vhost/fd_man.h
+++ b/lib/vhost/fd_man.h
@@ -8,50 +8,19 @@
 #include <poll.h>
 #include <stdbool.h>
 
+struct fdset;
+
 #define MAX_FDS 1024
 
 typedef void (*fd_cb)(int fd, void *dat, int *remove);
 
-struct fdentry {
-	int fd;		/* -1 indicates this entry is empty */
-	fd_cb rcb;	/* callback when this fd is readable. */
-	fd_cb wcb;	/* callback when this fd is writeable.*/
-	void *dat;	/* fd context */
-	int busy;	/* whether this entry is being used in cb. */
-};
-
-struct fdset {
-	struct pollfd rwfds[MAX_FDS];
-	struct fdentry fd[MAX_FDS];
-	pthread_mutex_t fd_mutex;
-	pthread_mutex_t fd_polling_mutex;
-	int num;	/* current fd number of this fdset */
-
-	union pipefds {
-		struct {
-			int pipefd[2];
-		};
-		struct {
-			int readfd;
-			int writefd;
-		};
-	} u;
-
-	pthread_mutex_t sync_mutex;
-	pthread_cond_t sync_cond;
-	bool sync;
-};
-
-void fdset_uninit(struct fdset *pfdset);
-
-int fdset_init(struct fdset *pfdset);
+struct fdset *fdset_init(const char *name);
 
 int fdset_add(struct fdset *pfdset, int fd,
 	fd_cb rcb, fd_cb wcb, void *dat);
 
 void *fdset_del(struct fdset *pfdset, int fd);
-int fdset_try_del(struct fdset *pfdset, int fd);
 
-uint32_t fdset_event_dispatch(void *arg);
+int fdset_try_del(struct fdset *pfdset, int fd);
 
 #endif
diff --git a/lib/vhost/socket.c b/lib/vhost/socket.c
index 5afb952a21..c68e9bd5a8 100644
--- a/lib/vhost/socket.c
+++ b/lib/vhost/socket.c
@@ -76,7 +76,7 @@ struct vhost_user_connection {
 #define MAX_VHOST_SOCKET 1024
 struct vhost_user {
 	struct vhost_user_socket *vsockets[MAX_VHOST_SOCKET];
-	struct fdset fdset;
+	struct fdset *fdset;
 	int vsocket_cnt;
 	pthread_mutex_t mutex;
 };
@@ -261,7 +261,7 @@ vhost_user_add_connection(int fd, struct vhost_user_socket *vsocket)
 	conn->connfd = fd;
 	conn->vsocket = vsocket;
 	conn->vid = vid;
-	ret = fdset_add(&vhost_user.fdset, fd, vhost_user_read_cb,
+	ret = fdset_add(vhost_user.fdset, fd, vhost_user_read_cb,
 			NULL, conn);
 	if (ret < 0) {
 		VHOST_CONFIG_LOG(vsocket->path, ERR,
@@ -394,7 +394,7 @@ vhost_user_start_server(struct vhost_user_socket *vsocket)
 	if (ret < 0)
 		goto err;
 
-	ret = fdset_add(&vhost_user.fdset, fd, vhost_user_server_new_connection,
+	ret = fdset_add(vhost_user.fdset, fd, vhost_user_server_new_connection,
 		  NULL, vsocket);
 	if (ret < 0) {
 		VHOST_CONFIG_LOG(path, ERR, "failed to add listen fd %d to vhost server fdset",
@@ -1079,7 +1079,7 @@ rte_vhost_driver_unregister(const char *path)
 			 * mutex lock, and try again since the r/wcb
 			 * may use the mutex lock.
 			 */
-			if (fdset_try_del(&vhost_user.fdset, vsocket->socket_fd) == -1) {
+			if (fdset_try_del(vhost_user.fdset, vsocket->socket_fd) == -1) {
 				pthread_mutex_unlock(&vhost_user.mutex);
 				goto again;
 			}
@@ -1099,7 +1099,7 @@ rte_vhost_driver_unregister(const char *path)
 			 * try again since the r/wcb may use the
 			 * conn_mutex and mutex locks.
 			 */
-			if (fdset_try_del(&vhost_user.fdset,
+			if (fdset_try_del(vhost_user.fdset,
 					  conn->connfd) == -1) {
 				pthread_mutex_unlock(&vsocket->conn_mutex);
 				pthread_mutex_unlock(&vhost_user.mutex);
@@ -1167,7 +1167,6 @@ int
 rte_vhost_driver_start(const char *path)
 {
 	struct vhost_user_socket *vsocket;
-	static rte_thread_t fdset_tid;
 
 	pthread_mutex_lock(&vhost_user.mutex);
 	vsocket = find_vhost_user_socket(path);
@@ -1179,19 +1178,12 @@ rte_vhost_driver_start(const char *path)
 	if (vsocket->is_vduse)
 		return vduse_device_create(path, vsocket->net_compliant_ol_flags);
 
-	if (fdset_tid.opaque_id == 0) {
-		if (fdset_init(&vhost_user.fdset) < 0) {
+	if (vhost_user.fdset == NULL) {
+		vhost_user.fdset = fdset_init("vhost-evt");
+		if (vhost_user.fdset == NULL) {
 			VHOST_CONFIG_LOG(path, ERR, "failed to init Vhost-user fdset");
 			return -1;
 		}
-
-		int ret = rte_thread_create_internal_control(&fdset_tid,
-				"vhost-evt", fdset_event_dispatch, &vhost_user.fdset);
-		if (ret != 0) {
-			VHOST_CONFIG_LOG(path, ERR, "failed to create fdset handling thread");
-			fdset_uninit(&vhost_user.fdset);
-			return -1;
-		}
 	}
 
 	if (vsocket->is_server)
diff --git a/lib/vhost/vduse.c b/lib/vhost/vduse.c
index d87fc500d4..c66602905c 100644
--- a/lib/vhost/vduse.c
+++ b/lib/vhost/vduse.c
@@ -28,13 +28,11 @@
 #define VDUSE_CTRL_PATH "/dev/vduse/control"
 
 struct vduse {
-	struct fdset fdset;
+	struct fdset *fdset;
 };
 
 static struct vduse vduse;
 
-static bool vduse_events_thread;
-
 static const char * const vduse_reqs_str[] = {
 	"VDUSE_GET_VQ_STATE",
 	"VDUSE_SET_STATUS",
@@ -215,7 +213,7 @@ vduse_vring_setup(struct virtio_net *dev, unsigned int index)
 	}
 
 	if (vq == dev->cvq) {
-		ret = fdset_add(&vduse.fdset, vq->kickfd, vduse_control_queue_event, NULL, dev);
+		ret = fdset_add(vduse.fdset, vq->kickfd, vduse_control_queue_event, NULL, dev);
 		if (ret) {
 			VHOST_CONFIG_LOG(dev->ifname, ERR,
 					"Failed to setup kickfd handler for VQ %u: %s",
@@ -238,7 +236,7 @@ vduse_vring_cleanup(struct virtio_net *dev, unsigned int index)
 	int ret;
 
 	if (vq == dev->cvq && vq->kickfd >= 0)
-		fdset_del(&vduse.fdset, vq->kickfd);
+		fdset_del(vduse.fdset, vq->kickfd);
 
 	vq_efd.index = index;
 	vq_efd.fd = VDUSE_EVENTFD_DEASSIGN;
@@ -413,7 +411,6 @@ int
 vduse_device_create(const char *path, bool compliant_ol_flags)
 {
 	int control_fd, dev_fd, vid, ret;
-	rte_thread_t fdset_tid;
 	uint32_t i, max_queue_pairs, total_queues;
 	struct virtio_net *dev;
 	struct virtio_net_config vnet_config = {{ 0 }};
@@ -422,22 +419,12 @@ vduse_device_create(const char *path, bool compliant_ol_flags)
 	struct vduse_dev_config *dev_config = NULL;
 	const char *name = path + strlen("/dev/vduse/");
 
-	/* If first device, create events dispatcher thread */
-	if (vduse_events_thread == false) {
-		if (fdset_init(&vduse.fdset) < 0) {
+	if (vduse.fdset == NULL) {
+		vduse.fdset = fdset_init("vduse-evt");
+		if (vduse.fdset == NULL) {
 			VHOST_CONFIG_LOG(path, ERR, "failed to init VDUSE fdset");
 			return -1;
 		}
-
-		ret = rte_thread_create_internal_control(&fdset_tid, "vduse-evt",
-				fdset_event_dispatch, &vduse.fdset);
-		if (ret != 0) {
-			VHOST_CONFIG_LOG(path, ERR, "failed to create vduse fdset handling thread");
-			fdset_uninit(&vduse.fdset);
-			return -1;
-		}
-
-		vduse_events_thread = true;
 	}
 
 	control_fd = open(VDUSE_CTRL_PATH, O_RDWR);
@@ -555,7 +542,7 @@ vduse_device_create(const char *path, bool compliant_ol_flags)
 
 	dev->cvq = dev->virtqueue[max_queue_pairs * 2];
 
-	ret = fdset_add(&vduse.fdset, dev->vduse_dev_fd, vduse_events_handler, NULL, dev);
+	ret = fdset_add(vduse.fdset, dev->vduse_dev_fd, vduse_events_handler, NULL, dev);
 	if (ret) {
 		VHOST_CONFIG_LOG(name, ERR, "Failed to add fd %d to vduse fdset",
 				dev->vduse_dev_fd);
@@ -602,7 +589,7 @@ vduse_device_destroy(const char *path)
 
 	vduse_device_stop(dev);
 
-	fdset_del(&vduse.fdset, dev->vduse_dev_fd);
+	fdset_del(vduse.fdset, dev->vduse_dev_fd);
 
 	if (dev->vduse_dev_fd >= 0) {
 		close(dev->vduse_dev_fd);
-- 
2.44.0

Re: [PATCH v3 4/5] vhost: improve fdset initialization

[Chenbo Xia]

Hi Maxime,

> On Apr 9, 2024, at 19:48, Maxime Coquelin <maxime.coquelin@redhat.com> wrote:
> 
> External email: Use caution opening links or attachments
> 
> 
> This patch heavily reworks fdset initialization:
> - fdsets are now dynamically allocated by the FD manager
> - the event dispatcher is now created by the FD manager
> - struct fdset is now opaque to VDUSE and Vhost
> 
> Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
> ---
> lib/vhost/fd_man.c | 177 ++++++++++++++++++++++++++++++++++++++++-----
> lib/vhost/fd_man.h |  39 +---------
> lib/vhost/socket.c |  24 ++----
> lib/vhost/vduse.c  |  29 ++------
> 4 files changed, 177 insertions(+), 92 deletions(-)
> 
> diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
> index 0ae481b785..8b47c97d45 100644
> --- a/lib/vhost/fd_man.c
> +++ b/lib/vhost/fd_man.c
> @@ -3,12 +3,16 @@
>  */
> 
> #include <errno.h>
> +#include <pthread.h>
> #include <stdio.h>
> #include <string.h>
> #include <unistd.h>
> 
> #include <rte_common.h>
> #include <rte_log.h>
> +#include <rte_malloc.h>
> +#include <rte_string_fns.h>
> +#include <rte_thread.h>
> 
> #include "fd_man.h"
> 
> @@ -19,6 +23,79 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
> 
> #define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
> 
> +struct fdentry {
> +       int fd;         /* -1 indicates this entry is empty */
> +       fd_cb rcb;      /* callback when this fd is readable. */
> +       fd_cb wcb;      /* callback when this fd is writeable.*/
> +       void *dat;      /* fd context */
> +       int busy;       /* whether this entry is being used in cb. */
> +};
> +
> +struct fdset {
> +       char name[RTE_THREAD_NAME_SIZE];
> +       struct pollfd rwfds[MAX_FDS];
> +       struct fdentry fd[MAX_FDS];
> +       rte_thread_t tid;
> +       pthread_mutex_t fd_mutex;
> +       pthread_mutex_t fd_polling_mutex;
> +       int num;        /* current fd number of this fdset */
> +
> +       union pipefds {
> +               struct {
> +                       int pipefd[2];
> +               };
> +               struct {
> +                       int readfd;
> +                       int writefd;
> +               };
> +       } u;
> +
> +       pthread_mutex_t sync_mutex;
> +       pthread_cond_t sync_cond;
> +       bool sync;
> +       bool destroy;
> +};
> +
> +static int fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat);
> +static uint32_t fdset_event_dispatch(void *arg);
> +
> +#define MAX_FDSETS 8
> +
> +static struct fdset *fdsets[MAX_FDSETS];
> +pthread_mutex_t fdsets_mutex = PTHREAD_MUTEX_INITIALIZER;

Static pthread_mutex_t ?

Thanks,
Chenbo

Re: [PATCH v3 4/5] vhost: improve fdset initialization

[Maxime Coquelin]

Hi Chenbo,

On 4/26/24 09:40, Chenbo Xia wrote:
> Hi Maxime,
> 
>> On Apr 9, 2024, at 19:48, Maxime Coquelin <maxime.coquelin@redhat.com> wrote:
>>
>> External email: Use caution opening links or attachments
>>
>>
>> This patch heavily reworks fdset initialization:
>> - fdsets are now dynamically allocated by the FD manager
>> - the event dispatcher is now created by the FD manager
>> - struct fdset is now opaque to VDUSE and Vhost
>>
>> Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
>> ---
>> lib/vhost/fd_man.c | 177 ++++++++++++++++++++++++++++++++++++++++-----
>> lib/vhost/fd_man.h |  39 +---------
>> lib/vhost/socket.c |  24 ++----
>> lib/vhost/vduse.c  |  29 ++------
>> 4 files changed, 177 insertions(+), 92 deletions(-)
>>
>> diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
>> index 0ae481b785..8b47c97d45 100644
>> --- a/lib/vhost/fd_man.c
>> +++ b/lib/vhost/fd_man.c
>> @@ -3,12 +3,16 @@
>>   */
>>
>> #include <errno.h>
>> +#include <pthread.h>
>> #include <stdio.h>
>> #include <string.h>
>> #include <unistd.h>
>>
>> #include <rte_common.h>
>> #include <rte_log.h>
>> +#include <rte_malloc.h>
>> +#include <rte_string_fns.h>
>> +#include <rte_thread.h>
>>
>> #include "fd_man.h"
>>
>> @@ -19,6 +23,79 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
>>
>> #define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
>>
>> +struct fdentry {
>> +       int fd;         /* -1 indicates this entry is empty */
>> +       fd_cb rcb;      /* callback when this fd is readable. */
>> +       fd_cb wcb;      /* callback when this fd is writeable.*/
>> +       void *dat;      /* fd context */
>> +       int busy;       /* whether this entry is being used in cb. */
>> +};
>> +
>> +struct fdset {
>> +       char name[RTE_THREAD_NAME_SIZE];
>> +       struct pollfd rwfds[MAX_FDS];
>> +       struct fdentry fd[MAX_FDS];
>> +       rte_thread_t tid;
>> +       pthread_mutex_t fd_mutex;
>> +       pthread_mutex_t fd_polling_mutex;
>> +       int num;        /* current fd number of this fdset */
>> +
>> +       union pipefds {
>> +               struct {
>> +                       int pipefd[2];
>> +               };
>> +               struct {
>> +                       int readfd;
>> +                       int writefd;
>> +               };
>> +       } u;
>> +
>> +       pthread_mutex_t sync_mutex;
>> +       pthread_cond_t sync_cond;
>> +       bool sync;
>> +       bool destroy;
>> +};
>> +
>> +static int fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat);
>> +static uint32_t fdset_event_dispatch(void *arg);
>> +
>> +#define MAX_FDSETS 8
>> +
>> +static struct fdset *fdsets[MAX_FDSETS];
>> +pthread_mutex_t fdsets_mutex = PTHREAD_MUTEX_INITIALIZER;
> 
> Static pthread_mutex_t ?

I think so, thanks for spotting it!
Will fix in next revision.

Maxime

> Thanks,
> Chenbo
> 

[PATCH v3 5/5] vhost: manage FD with epoll

[Maxime Coquelin]

From: David Marchand <david.marchand@redhat.com>

Switch to epoll so that the concern over the poll() fd array
is removed.
Add a simple list of used entries and track the next free entry.

epoll() is thread safe, we no more need a synchronization
mechanism and so can remove the notification pipe.

Signed-off-by: David Marchand <david.marchand@redhat.com>
Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
---
 lib/vhost/fd_man.c | 399 ++++++++++++---------------------------------
 lib/vhost/fd_man.h |   5 +-
 2 files changed, 106 insertions(+), 298 deletions(-)

diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
index 8b47c97d45..a4a2965da1 100644
--- a/lib/vhost/fd_man.c
+++ b/lib/vhost/fd_man.c
@@ -3,9 +3,9 @@
  */
 
 #include <errno.h>
-#include <pthread.h>
 #include <stdio.h>
 #include <string.h>
+#include <sys/epoll.h>
 #include <unistd.h>
 
 #include <rte_common.h>
@@ -21,49 +21,34 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
 #define VHOST_FDMAN_LOG(level, ...) \
 	RTE_LOG_LINE(level, VHOST_FDMAN, "" __VA_ARGS__)
 
-#define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
-
 struct fdentry {
 	int fd;		/* -1 indicates this entry is empty */
 	fd_cb rcb;	/* callback when this fd is readable. */
 	fd_cb wcb;	/* callback when this fd is writeable.*/
 	void *dat;	/* fd context */
 	int busy;	/* whether this entry is being used in cb. */
+	LIST_ENTRY(fdentry) next;
 };
 
 struct fdset {
 	char name[RTE_THREAD_NAME_SIZE];
-	struct pollfd rwfds[MAX_FDS];
+	int epfd;
 	struct fdentry fd[MAX_FDS];
+	LIST_HEAD(, fdentry) fdlist;
+	int next_free_idx;
 	rte_thread_t tid;
 	pthread_mutex_t fd_mutex;
-	pthread_mutex_t fd_polling_mutex;
-	int num;	/* current fd number of this fdset */
-
-	union pipefds {
-		struct {
-			int pipefd[2];
-		};
-		struct {
-			int readfd;
-			int writefd;
-		};
-	} u;
-
-	pthread_mutex_t sync_mutex;
-	pthread_cond_t sync_cond;
-	bool sync;
+
 	bool destroy;
 };
 
-static int fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat);
-static uint32_t fdset_event_dispatch(void *arg);
-
 #define MAX_FDSETS 8
 
 static struct fdset *fdsets[MAX_FDSETS];
 pthread_mutex_t fdsets_mutex = PTHREAD_MUTEX_INITIALIZER;
 
+static uint32_t fdset_event_dispatch(void *arg);
+
 static struct fdset *
 fdset_lookup(const char *name)
 {
@@ -96,166 +81,6 @@ fdset_insert(struct fdset *fdset)
 	return -1;
 }
 
-static void
-fdset_pipe_read_cb(int readfd, void *dat,
-		   int *remove __rte_unused)
-{
-	char charbuf[16];
-	struct fdset *fdset = dat;
-	int r = read(readfd, charbuf, sizeof(charbuf));
-	/*
-	 * Just an optimization, we don't care if read() failed
-	 * so ignore explicitly its return value to make the
-	 * compiler happy
-	 */
-	RTE_SET_USED(r);
-
-	pthread_mutex_lock(&fdset->sync_mutex);
-	fdset->sync = true;
-	pthread_cond_broadcast(&fdset->sync_cond);
-	pthread_mutex_unlock(&fdset->sync_mutex);
-}
-
-static void
-fdset_pipe_uninit(struct fdset *fdset)
-{
-	fdset_del(fdset, fdset->u.readfd);
-	close(fdset->u.readfd);
-	fdset->u.readfd = -1;
-	close(fdset->u.writefd);
-	fdset->u.writefd = -1;
-}
-
-static int
-fdset_pipe_init(struct fdset *fdset)
-{
-	int ret;
-
-	pthread_mutex_init(&fdset->sync_mutex, NULL);
-	pthread_cond_init(&fdset->sync_cond, NULL);
-
-	if (pipe(fdset->u.pipefd) < 0) {
-		VHOST_FDMAN_LOG(ERR,
-			"failed to create pipe for vhost fdset");
-		return -1;
-	}
-
-	ret = fdset_add_no_sync(fdset, fdset->u.readfd,
-			fdset_pipe_read_cb, NULL, fdset);
-	if (ret < 0) {
-		VHOST_FDMAN_LOG(ERR,
-			"failed to add pipe readfd %d into vhost server fdset",
-			fdset->u.readfd);
-
-		fdset_pipe_uninit(fdset);
-		return -1;
-	}
-
-	return 0;
-}
-
-static void
-fdset_sync(struct fdset *fdset)
-{
-	int ret;
-
-	pthread_mutex_lock(&fdset->sync_mutex);
-
-	fdset->sync = false;
-	ret = write(fdset->u.writefd, "1", 1);
-	if (ret < 0) {
-		VHOST_FDMAN_LOG(ERR,
-			"Failed to write to notification pipe: %s",
-			strerror(errno));
-		goto out_unlock;
-	}
-
-	while (!fdset->sync)
-		pthread_cond_wait(&fdset->sync_cond, &fdset->sync_mutex);
-
-out_unlock:
-	pthread_mutex_unlock(&fdset->sync_mutex);
-}
-
-static int
-get_last_valid_idx(struct fdset *pfdset, int last_valid_idx)
-{
-	int i;
-
-	for (i = last_valid_idx; i >= 0 && pfdset->fd[i].fd == -1; i--)
-		;
-
-	return i;
-}
-
-static void
-fdset_move(struct fdset *pfdset, int dst, int src)
-{
-	pfdset->fd[dst]    = pfdset->fd[src];
-	pfdset->rwfds[dst] = pfdset->rwfds[src];
-}
-
-static void
-fdset_shrink_nolock(struct fdset *pfdset)
-{
-	int i;
-	int last_valid_idx = get_last_valid_idx(pfdset, pfdset->num - 1);
-
-	for (i = 0; i < last_valid_idx; i++) {
-		if (pfdset->fd[i].fd != -1)
-			continue;
-
-		fdset_move(pfdset, i, last_valid_idx);
-		last_valid_idx = get_last_valid_idx(pfdset, last_valid_idx - 1);
-	}
-	pfdset->num = last_valid_idx + 1;
-}
-
-/*
- * Find deleted fd entries and remove them
- */
-static void
-fdset_shrink(struct fdset *pfdset)
-{
-	pthread_mutex_lock(&pfdset->fd_mutex);
-	fdset_shrink_nolock(pfdset);
-	pthread_mutex_unlock(&pfdset->fd_mutex);
-}
-
-/**
- * Returns the index in the fdset for a given fd.
- * @return
- *   index for the fd, or -1 if fd isn't in the fdset.
- */
-static int
-fdset_find_fd(struct fdset *pfdset, int fd)
-{
-	int i;
-
-	for (i = 0; i < pfdset->num && pfdset->fd[i].fd != fd; i++)
-		;
-
-	return i == pfdset->num ? -1 : i;
-}
-
-static void
-fdset_add_fd(struct fdset *pfdset, int idx, int fd,
-	fd_cb rcb, fd_cb wcb, void *dat)
-{
-	struct fdentry *pfdentry = &pfdset->fd[idx];
-	struct pollfd *pfd = &pfdset->rwfds[idx];
-
-	pfdentry->fd  = fd;
-	pfdentry->rcb = rcb;
-	pfdentry->wcb = wcb;
-	pfdentry->dat = dat;
-
-	pfd->fd = fd;
-	pfd->events  = rcb ? POLLIN : 0;
-	pfd->events |= wcb ? POLLOUT : 0;
-	pfd->revents = 0;
-}
-
 struct fdset *
 fdset_init(const char *name)
 {
@@ -284,16 +109,20 @@ fdset_init(const char *name)
 	rte_strscpy(fdset->name, name, RTE_THREAD_NAME_SIZE);
 
 	pthread_mutex_init(&fdset->fd_mutex, NULL);
-	pthread_mutex_init(&fdset->fd_polling_mutex, NULL);
 
-	for (i = 0; i < MAX_FDS; i++) {
+	for (i = 0; i < (int)RTE_DIM(fdset->fd); i++) {
 		fdset->fd[i].fd = -1;
 		fdset->fd[i].dat = NULL;
 	}
-	fdset->num = 0;
+	LIST_INIT(&fdset->fdlist);
 
-	if (fdset_pipe_init(fdset)) {
-		VHOST_FDMAN_LOG(ERR, "Failed to init pipe for %s", name);
+	/*
+	 * Any non-zero value would work (see man epoll_create),
+	 * but pass MAX_FDS for consistency.
+	 */
+	fdset->epfd = epoll_create(MAX_FDS);
+	if (fdset->epfd < 0) {
+		VHOST_FDMAN_LOG(ERR, "failed to create epoll for %s fdset", name);
 		goto err_free;
 	}
 
@@ -301,7 +130,7 @@ fdset_init(const char *name)
 					fdset_event_dispatch, fdset)) {
 		VHOST_FDMAN_LOG(ERR, "Failed to create %s event dispatch thread",
 				fdset->name);
-		goto err_pipe;
+		goto err_epoll;
 	}
 
 	if (fdset_insert(fdset)) {
@@ -315,10 +144,9 @@ fdset_init(const char *name)
 
 err_thread:
 	fdset->destroy = true;
-	fdset_sync(fdset);
 	rte_thread_join(fdset->tid, &val);
-err_pipe:
-	fdset_pipe_uninit(fdset);
+err_epoll:
+	close(fdset->epfd);
 err_free:
 	rte_free(fdset);
 err_unlock:
@@ -330,78 +158,99 @@ fdset_init(const char *name)
 /**
  * Register the fd in the fdset with read/write handler and context.
  */
-static int
-fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
+int
+fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
 {
-	int i;
+	struct fdentry *pfdentry;
+	struct epoll_event ev;
 
 	if (pfdset == NULL || fd == -1)
 		return -1;
 
 	pthread_mutex_lock(&pfdset->fd_mutex);
-	i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
-	if (i == -1) {
-		pthread_mutex_lock(&pfdset->fd_polling_mutex);
-		fdset_shrink_nolock(pfdset);
-		pthread_mutex_unlock(&pfdset->fd_polling_mutex);
-		i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
-		if (i == -1) {
-			pthread_mutex_unlock(&pfdset->fd_mutex);
-			return -2;
-		}
+	if (pfdset->next_free_idx >= (int)RTE_DIM(pfdset->fd)) {
+		pthread_mutex_unlock(&pfdset->fd_mutex);
+		return -2;
 	}
 
-	fdset_add_fd(pfdset, i, fd, rcb, wcb, dat);
+	pfdentry = &pfdset->fd[pfdset->next_free_idx];
+	pfdentry->fd  = fd;
+	pfdentry->rcb = rcb;
+	pfdentry->wcb = wcb;
+	pfdentry->dat = dat;
+
+	LIST_INSERT_HEAD(&pfdset->fdlist, pfdentry, next);
+
+	/* Find next free slot */
+	pfdset->next_free_idx++;
+	for (; pfdset->next_free_idx < (int)RTE_DIM(pfdset->fd); pfdset->next_free_idx++) {
+		if (pfdset->fd[pfdset->next_free_idx].fd != -1)
+			continue;
+		break;
+	}
 	pthread_mutex_unlock(&pfdset->fd_mutex);
 
+	ev.events = EPOLLERR;
+	ev.events |= rcb ? EPOLLIN : 0;
+	ev.events |= wcb ? EPOLLOUT : 0;
+	ev.data.fd = fd;
+
+	if (epoll_ctl(pfdset->epfd, EPOLL_CTL_ADD, fd, &ev) == -1)
+		VHOST_FDMAN_LOG(ERR, "could not add %d fd to %d epfd: %s",
+			fd, pfdset->epfd, strerror(errno));
+
 	return 0;
 }
 
-int
-fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
+static struct fdentry *
+fdset_find_entry_locked(struct fdset *pfdset, int fd)
 {
-	int ret;
+	struct fdentry *pfdentry;
 
-	ret = fdset_add_no_sync(pfdset, fd, rcb, wcb, dat);
-	if (ret < 0)
-		return ret;
+	LIST_FOREACH(pfdentry, &pfdset->fdlist, next) {
+		if (pfdentry->fd != fd)
+			continue;
+		return pfdentry;
+	}
 
-	fdset_sync(pfdset);
+	return NULL;
+}
 
-	return 0;
+static void
+fdset_del_locked(struct fdset *pfdset, struct fdentry *pfdentry)
+{
+	int entry_idx;
+
+	if (epoll_ctl(pfdset->epfd, EPOLL_CTL_DEL, pfdentry->fd, NULL) == -1)
+		VHOST_FDMAN_LOG(ERR, "could not remove %d fd from %d epfd: %s",
+			pfdentry->fd, pfdset->epfd, strerror(errno));
+
+	pfdentry->fd = -1;
+	pfdentry->rcb = pfdentry->wcb = NULL;
+	pfdentry->dat = NULL;
+	entry_idx = pfdentry - pfdset->fd;
+	if (entry_idx < pfdset->next_free_idx)
+		pfdset->next_free_idx = entry_idx;
+	LIST_REMOVE(pfdentry, next);
 }
 
-/**
- *  Unregister the fd from the fdset.
- *  Returns context of a given fd or NULL.
- */
-void *
+void
 fdset_del(struct fdset *pfdset, int fd)
 {
-	int i;
-	void *dat = NULL;
+	struct fdentry *pfdentry;
 
 	if (pfdset == NULL || fd == -1)
-		return NULL;
+		return;
 
 	do {
 		pthread_mutex_lock(&pfdset->fd_mutex);
-
-		i = fdset_find_fd(pfdset, fd);
-		if (i != -1 && pfdset->fd[i].busy == 0) {
-			/* busy indicates r/wcb is executing! */
-			dat = pfdset->fd[i].dat;
-			pfdset->fd[i].fd = -1;
-			pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
-			pfdset->fd[i].dat = NULL;
-			i = -1;
+		pfdentry = fdset_find_entry_locked(pfdset, fd);
+		if (pfdentry != NULL && pfdentry->busy == 0) {
+			fdset_del_locked(pfdset, pfdentry);
+			pfdentry = NULL;
 		}
 		pthread_mutex_unlock(&pfdset->fd_mutex);
-	} while (i != -1);
-
-	fdset_sync(pfdset);
-
-	return dat;
+	} while (pfdentry != NULL);
 }
 
 /**
@@ -415,28 +264,22 @@ fdset_del(struct fdset *pfdset, int fd)
 int
 fdset_try_del(struct fdset *pfdset, int fd)
 {
-	int i;
+	struct fdentry *pfdentry;
 
 	if (pfdset == NULL || fd == -1)
 		return -2;
 
 	pthread_mutex_lock(&pfdset->fd_mutex);
-	i = fdset_find_fd(pfdset, fd);
-	if (i != -1 && pfdset->fd[i].busy) {
+	pfdentry = fdset_find_entry_locked(pfdset, fd);
+	if (pfdentry != NULL && pfdentry->busy != 0) {
 		pthread_mutex_unlock(&pfdset->fd_mutex);
 		return -1;
 	}
 
-	if (i != -1) {
-		pfdset->fd[i].fd = -1;
-		pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
-		pfdset->fd[i].dat = NULL;
-	}
+	if (pfdentry != NULL)
+		fdset_del_locked(pfdset, pfdentry);
 
 	pthread_mutex_unlock(&pfdset->fd_mutex);
-
-	fdset_sync(pfdset);
-
 	return 0;
 }
 
@@ -453,53 +296,29 @@ static uint32_t
 fdset_event_dispatch(void *arg)
 {
 	int i;
-	struct pollfd *pfd;
-	struct fdentry *pfdentry;
 	fd_cb rcb, wcb;
 	void *dat;
 	int fd, numfds;
 	int remove1, remove2;
-	int need_shrink;
 	struct fdset *pfdset = arg;
-	int val;
 
 	if (pfdset == NULL)
 		return 0;
 
 	while (1) {
+		struct epoll_event events[MAX_FDS];
+		struct fdentry *pfdentry;
 
-		/*
-		 * When poll is blocked, other threads might unregister
-		 * listenfds from and register new listenfds into fdset.
-		 * When poll returns, the entries for listenfds in the fdset
-		 * might have been updated. It is ok if there is unwanted call
-		 * for new listenfds.
-		 */
-		pthread_mutex_lock(&pfdset->fd_mutex);
-		numfds = pfdset->num;
-		pthread_mutex_unlock(&pfdset->fd_mutex);
-
-		pthread_mutex_lock(&pfdset->fd_polling_mutex);
-		val = poll(pfdset->rwfds, numfds, 1000 /* millisecs */);
-		pthread_mutex_unlock(&pfdset->fd_polling_mutex);
-		if (val < 0)
+		numfds = epoll_wait(pfdset->epfd, events, RTE_DIM(events), 1000);
+		if (numfds < 0)
 			continue;
 
-		need_shrink = 0;
 		for (i = 0; i < numfds; i++) {
 			pthread_mutex_lock(&pfdset->fd_mutex);
 
-			pfdentry = &pfdset->fd[i];
-			fd = pfdentry->fd;
-			pfd = &pfdset->rwfds[i];
-
-			if (fd < 0) {
-				need_shrink = 1;
-				pthread_mutex_unlock(&pfdset->fd_mutex);
-				continue;
-			}
-
-			if (!pfd->revents) {
+			fd = events[i].data.fd;
+			pfdentry = fdset_find_entry_locked(pfdset, fd);
+			if (pfdentry == NULL) {
 				pthread_mutex_unlock(&pfdset->fd_mutex);
 				continue;
 			}
@@ -513,9 +332,9 @@ fdset_event_dispatch(void *arg)
 
 			pthread_mutex_unlock(&pfdset->fd_mutex);
 
-			if (rcb && pfd->revents & (POLLIN | FDPOLLERR))
+			if (rcb && events[i].events & (EPOLLIN | EPOLLERR | EPOLLHUP))
 				rcb(fd, dat, &remove1);
-			if (wcb && pfd->revents & (POLLOUT | FDPOLLERR))
+			if (wcb && events[i].events & (EPOLLOUT | EPOLLERR | EPOLLHUP))
 				wcb(fd, dat, &remove2);
 			pfdentry->busy = 0;
 			/*
@@ -524,23 +343,13 @@ fdset_event_dispatch(void *arg)
 			 * directly.
 			 */
 			/*
-			 * When we are to clean up the fd from fdset,
-			 * because the fd is closed in the cb,
-			 * the old fd val could be reused by when creates new
-			 * listen fd in another thread, we couldn't call
-			 * fdset_del.
+			 * A concurrent fdset_del may have been waiting for the
+			 * fdentry not to be busy, so we can't call
+			 * fdset_del_locked().
 			 */
-			if (remove1 || remove2) {
-				pfdentry->fd = -1;
-				need_shrink = 1;
-			}
+			if (remove1 || remove2)
+				fdset_del(pfdset, fd);
 		}
-
-		if (need_shrink)
-			fdset_shrink(pfdset);
-
-		if (pfdset->destroy)
-			break;
 	}
 
 	return 0;
diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
index 079fa0155f..6398343a6a 100644
--- a/lib/vhost/fd_man.h
+++ b/lib/vhost/fd_man.h
@@ -6,7 +6,7 @@
 #define _FD_MAN_H_
 #include <pthread.h>
 #include <poll.h>
-#include <stdbool.h>
+#include <sys/queue.h>
 
 struct fdset;
 
@@ -19,8 +19,7 @@ struct fdset *fdset_init(const char *name);
 int fdset_add(struct fdset *pfdset, int fd,
 	fd_cb rcb, fd_cb wcb, void *dat);
 
-void *fdset_del(struct fdset *pfdset, int fd);
-
+void fdset_del(struct fdset *pfdset, int fd);
 int fdset_try_del(struct fdset *pfdset, int fd);
 
 #endif
-- 
2.44.0

Re: [PATCH v3 5/5] vhost: manage FD with epoll

[Chenbo Xia]

Hi Maxime,

> On Apr 9, 2024, at 19:48, Maxime Coquelin <maxime.coquelin@redhat.com> wrote:
> 
> External email: Use caution opening links or attachments
> 
> 
> From: David Marchand <david.marchand@redhat.com>
> 
> Switch to epoll so that the concern over the poll() fd array
> is removed.
> Add a simple list of used entries and track the next free entry.
> 
> epoll() is thread safe, we no more need a synchronization
> mechanism and so can remove the notification pipe.
> 
> Signed-off-by: David Marchand <david.marchand@redhat.com>
> Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
> ---
> lib/vhost/fd_man.c | 399 ++++++++++++---------------------------------
> lib/vhost/fd_man.h |   5 +-
> 2 files changed, 106 insertions(+), 298 deletions(-)
> 
> diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
> index 8b47c97d45..a4a2965da1 100644
> --- a/lib/vhost/fd_man.c
> +++ b/lib/vhost/fd_man.c
> @@ -3,9 +3,9 @@
>  */
> 
> #include <errno.h>
> -#include <pthread.h>
> #include <stdio.h>
> #include <string.h>
> +#include <sys/epoll.h>
> #include <unistd.h>
> 
> #include <rte_common.h>
> @@ -21,49 +21,34 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
> #define VHOST_FDMAN_LOG(level, ...) \
>        RTE_LOG_LINE(level, VHOST_FDMAN, "" __VA_ARGS__)
> 
> -#define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
> -
> struct fdentry {
>        int fd;         /* -1 indicates this entry is empty */
>        fd_cb rcb;      /* callback when this fd is readable. */
>        fd_cb wcb;      /* callback when this fd is writeable.*/
>        void *dat;      /* fd context */
>        int busy;       /* whether this entry is being used in cb. */
> +       LIST_ENTRY(fdentry) next;
> };
> 
> struct fdset {
>        char name[RTE_THREAD_NAME_SIZE];
> -       struct pollfd rwfds[MAX_FDS];
> +       int epfd;
>        struct fdentry fd[MAX_FDS];
> +       LIST_HEAD(, fdentry) fdlist;
> +       int next_free_idx;
>        rte_thread_t tid;
>        pthread_mutex_t fd_mutex;
> -       pthread_mutex_t fd_polling_mutex;
> -       int num;        /* current fd number of this fdset */
> -
> -       union pipefds {
> -               struct {
> -                       int pipefd[2];
> -               };
> -               struct {
> -                       int readfd;
> -                       int writefd;
> -               };
> -       } u;
> -
> -       pthread_mutex_t sync_mutex;
> -       pthread_cond_t sync_cond;
> -       bool sync;
> +

Not sure this blank line is intended or not :)

>        bool destroy;
> };
> 
> -static int fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat);
> -static uint32_t fdset_event_dispatch(void *arg);
> -
> #define MAX_FDSETS 8
> 
> static struct fdset *fdsets[MAX_FDSETS];
> pthread_mutex_t fdsets_mutex = PTHREAD_MUTEX_INITIALIZER;
> 
> +static uint32_t fdset_event_dispatch(void *arg);
> +
> static struct fdset *
> fdset_lookup(const char *name)
> {
> @@ -96,166 +81,6 @@ fdset_insert(struct fdset *fdset)
>        return -1;
> }
> 
> -static void
> -fdset_pipe_read_cb(int readfd, void *dat,
> -                  int *remove __rte_unused)
> -{
> -       char charbuf[16];
> -       struct fdset *fdset = dat;
> -       int r = read(readfd, charbuf, sizeof(charbuf));
> -       /*
> -        * Just an optimization, we don't care if read() failed
> -        * so ignore explicitly its return value to make the
> -        * compiler happy
> -        */
> -       RTE_SET_USED(r);
> -
> -       pthread_mutex_lock(&fdset->sync_mutex);
> -       fdset->sync = true;
> -       pthread_cond_broadcast(&fdset->sync_cond);
> -       pthread_mutex_unlock(&fdset->sync_mutex);
> -}
> -
> -static void
> -fdset_pipe_uninit(struct fdset *fdset)
> -{
> -       fdset_del(fdset, fdset->u.readfd);
> -       close(fdset->u.readfd);
> -       fdset->u.readfd = -1;
> -       close(fdset->u.writefd);
> -       fdset->u.writefd = -1;
> -}
> -
> -static int
> -fdset_pipe_init(struct fdset *fdset)
> -{
> -       int ret;
> -
> -       pthread_mutex_init(&fdset->sync_mutex, NULL);
> -       pthread_cond_init(&fdset->sync_cond, NULL);
> -
> -       if (pipe(fdset->u.pipefd) < 0) {
> -               VHOST_FDMAN_LOG(ERR,
> -                       "failed to create pipe for vhost fdset");
> -               return -1;
> -       }
> -
> -       ret = fdset_add_no_sync(fdset, fdset->u.readfd,
> -                       fdset_pipe_read_cb, NULL, fdset);
> -       if (ret < 0) {
> -               VHOST_FDMAN_LOG(ERR,
> -                       "failed to add pipe readfd %d into vhost server fdset",
> -                       fdset->u.readfd);
> -
> -               fdset_pipe_uninit(fdset);
> -               return -1;
> -       }
> -
> -       return 0;
> -}
> -
> -static void
> -fdset_sync(struct fdset *fdset)
> -{
> -       int ret;
> -
> -       pthread_mutex_lock(&fdset->sync_mutex);
> -
> -       fdset->sync = false;
> -       ret = write(fdset->u.writefd, "1", 1);
> -       if (ret < 0) {
> -               VHOST_FDMAN_LOG(ERR,
> -                       "Failed to write to notification pipe: %s",
> -                       strerror(errno));
> -               goto out_unlock;
> -       }
> -
> -       while (!fdset->sync)
> -               pthread_cond_wait(&fdset->sync_cond, &fdset->sync_mutex);
> -
> -out_unlock:
> -       pthread_mutex_unlock(&fdset->sync_mutex);
> -}
> -
> -static int
> -get_last_valid_idx(struct fdset *pfdset, int last_valid_idx)
> -{
> -       int i;
> -
> -       for (i = last_valid_idx; i >= 0 && pfdset->fd[i].fd == -1; i--)
> -               ;
> -
> -       return i;
> -}
> -
> -static void
> -fdset_move(struct fdset *pfdset, int dst, int src)
> -{
> -       pfdset->fd[dst]    = pfdset->fd[src];
> -       pfdset->rwfds[dst] = pfdset->rwfds[src];
> -}
> -
> -static void
> -fdset_shrink_nolock(struct fdset *pfdset)
> -{
> -       int i;
> -       int last_valid_idx = get_last_valid_idx(pfdset, pfdset->num - 1);
> -
> -       for (i = 0; i < last_valid_idx; i++) {
> -               if (pfdset->fd[i].fd != -1)
> -                       continue;
> -
> -               fdset_move(pfdset, i, last_valid_idx);
> -               last_valid_idx = get_last_valid_idx(pfdset, last_valid_idx - 1);
> -       }
> -       pfdset->num = last_valid_idx + 1;
> -}
> -
> -/*
> - * Find deleted fd entries and remove them
> - */
> -static void
> -fdset_shrink(struct fdset *pfdset)
> -{
> -       pthread_mutex_lock(&pfdset->fd_mutex);
> -       fdset_shrink_nolock(pfdset);
> -       pthread_mutex_unlock(&pfdset->fd_mutex);
> -}
> -
> -/**
> - * Returns the index in the fdset for a given fd.
> - * @return
> - *   index for the fd, or -1 if fd isn't in the fdset.
> - */
> -static int
> -fdset_find_fd(struct fdset *pfdset, int fd)
> -{
> -       int i;
> -
> -       for (i = 0; i < pfdset->num && pfdset->fd[i].fd != fd; i++)
> -               ;
> -
> -       return i == pfdset->num ? -1 : i;
> -}
> -
> -static void
> -fdset_add_fd(struct fdset *pfdset, int idx, int fd,
> -       fd_cb rcb, fd_cb wcb, void *dat)
> -{
> -       struct fdentry *pfdentry = &pfdset->fd[idx];
> -       struct pollfd *pfd = &pfdset->rwfds[idx];
> -
> -       pfdentry->fd  = fd;
> -       pfdentry->rcb = rcb;
> -       pfdentry->wcb = wcb;
> -       pfdentry->dat = dat;
> -
> -       pfd->fd = fd;
> -       pfd->events  = rcb ? POLLIN : 0;
> -       pfd->events |= wcb ? POLLOUT : 0;
> -       pfd->revents = 0;
> -}
> -
> struct fdset *
> fdset_init(const char *name)
> {
> @@ -284,16 +109,20 @@ fdset_init(const char *name)
>        rte_strscpy(fdset->name, name, RTE_THREAD_NAME_SIZE);
> 
>        pthread_mutex_init(&fdset->fd_mutex, NULL);
> -       pthread_mutex_init(&fdset->fd_polling_mutex, NULL);
> 
> -       for (i = 0; i < MAX_FDS; i++) {
> +       for (i = 0; i < (int)RTE_DIM(fdset->fd); i++) {
>                fdset->fd[i].fd = -1;
>                fdset->fd[i].dat = NULL;
>        }
> -       fdset->num = 0;
> +       LIST_INIT(&fdset->fdlist);
> 
> -       if (fdset_pipe_init(fdset)) {
> -               VHOST_FDMAN_LOG(ERR, "Failed to init pipe for %s", name);
> +       /*
> +        * Any non-zero value would work (see man epoll_create),
> +        * but pass MAX_FDS for consistency.
> +        */
> +       fdset->epfd = epoll_create(MAX_FDS);
> +       if (fdset->epfd < 0) {
> +               VHOST_FDMAN_LOG(ERR, "failed to create epoll for %s fdset", name);

failed -> Failed like other logs

>                goto err_free;
>        }
> 
> @@ -301,7 +130,7 @@ fdset_init(const char *name)
>                                        fdset_event_dispatch, fdset)) {
>                VHOST_FDMAN_LOG(ERR, "Failed to create %s event dispatch thread",
>                                fdset->name);
> -               goto err_pipe;
> +               goto err_epoll;
>        }
> 
>        if (fdset_insert(fdset)) {
> @@ -315,10 +144,9 @@ fdset_init(const char *name)
> 
> err_thread:
>        fdset->destroy = true;
> -       fdset_sync(fdset);
>        rte_thread_join(fdset->tid, &val);
> -err_pipe:
> -       fdset_pipe_uninit(fdset);
> +err_epoll:
> +       close(fdset->epfd);
> err_free:
>        rte_free(fdset);
> err_unlock:
> @@ -330,78 +158,99 @@ fdset_init(const char *name)
> /**
>  * Register the fd in the fdset with read/write handler and context.
>  */
> -static int
> -fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
> +int
> +fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
> {
> -       int i;
> +       struct fdentry *pfdentry;
> +       struct epoll_event ev;
> 
>        if (pfdset == NULL || fd == -1)
>                return -1;
> 
>        pthread_mutex_lock(&pfdset->fd_mutex);
> -       i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
> -       if (i == -1) {
> -               pthread_mutex_lock(&pfdset->fd_polling_mutex);
> -               fdset_shrink_nolock(pfdset);
> -               pthread_mutex_unlock(&pfdset->fd_polling_mutex);
> -               i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
> -               if (i == -1) {
> -                       pthread_mutex_unlock(&pfdset->fd_mutex);
> -                       return -2;
> -               }
> +       if (pfdset->next_free_idx >= (int)RTE_DIM(pfdset->fd)) {
> +               pthread_mutex_unlock(&pfdset->fd_mutex);
> +               return -2;
>        }
> 
> -       fdset_add_fd(pfdset, i, fd, rcb, wcb, dat);
> +       pfdentry = &pfdset->fd[pfdset->next_free_idx];
> +       pfdentry->fd  = fd;
> +       pfdentry->rcb = rcb;
> +       pfdentry->wcb = wcb;
> +       pfdentry->dat = dat;
> +
> +       LIST_INSERT_HEAD(&pfdset->fdlist, pfdentry, next);
> +
> +       /* Find next free slot */
> +       pfdset->next_free_idx++;
> +       for (; pfdset->next_free_idx < (int)RTE_DIM(pfdset->fd); pfdset->next_free_idx++) {
> +               if (pfdset->fd[pfdset->next_free_idx].fd != -1)
> +                       continue;
> +               break;
> +       }
>        pthread_mutex_unlock(&pfdset->fd_mutex);
> 
> +       ev.events = EPOLLERR;
> +       ev.events |= rcb ? EPOLLIN : 0;
> +       ev.events |= wcb ? EPOLLOUT : 0;
> +       ev.data.fd = fd;
> +
> +       if (epoll_ctl(pfdset->epfd, EPOLL_CTL_ADD, fd, &ev) == -1)
> +               VHOST_FDMAN_LOG(ERR, "could not add %d fd to %d epfd: %s",
> +                       fd, pfdset->epfd, strerror(errno));

Should not return 0 if this fails ?

> +
>        return 0;
> }
> 
> -int
> -fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
> +static struct fdentry *
> +fdset_find_entry_locked(struct fdset *pfdset, int fd)
> {
> -       int ret;
> +       struct fdentry *pfdentry;
> 
> -       ret = fdset_add_no_sync(pfdset, fd, rcb, wcb, dat);
> -       if (ret < 0)
> -               return ret;
> +       LIST_FOREACH(pfdentry, &pfdset->fdlist, next) {
> +               if (pfdentry->fd != fd)
> +                       continue;
> +               return pfdentry;
> +       }
> 
> -       fdset_sync(pfdset);
> +       return NULL;
> +}
> 
> -       return 0;
> +static void
> +fdset_del_locked(struct fdset *pfdset, struct fdentry *pfdentry)
> +{
> +       int entry_idx;
> +
> +       if (epoll_ctl(pfdset->epfd, EPOLL_CTL_DEL, pfdentry->fd, NULL) == -1)
> +               VHOST_FDMAN_LOG(ERR, "could not remove %d fd from %d epfd: %s",
> +                       pfdentry->fd, pfdset->epfd, strerror(errno));
> +
> +       pfdentry->fd = -1;
> +       pfdentry->rcb = pfdentry->wcb = NULL;
> +       pfdentry->dat = NULL;
> +       entry_idx = pfdentry - pfdset->fd;
> +       if (entry_idx < pfdset->next_free_idx)
> +               pfdset->next_free_idx = entry_idx;
> +       LIST_REMOVE(pfdentry, next);
> }
> 
> -/**
> - *  Unregister the fd from the fdset.
> - *  Returns context of a given fd or NULL.
> - */
> -void *
> +void
> fdset_del(struct fdset *pfdset, int fd)
> {
> -       int i;
> -       void *dat = NULL;
> +       struct fdentry *pfdentry;
> 
>        if (pfdset == NULL || fd == -1)
> -               return NULL;
> +               return;
> 
>        do {
>                pthread_mutex_lock(&pfdset->fd_mutex);
> -
> -               i = fdset_find_fd(pfdset, fd);
> -               if (i != -1 && pfdset->fd[i].busy == 0) {
> -                       /* busy indicates r/wcb is executing! */
> -                       dat = pfdset->fd[i].dat;
> -                       pfdset->fd[i].fd = -1;
> -                       pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
> -                       pfdset->fd[i].dat = NULL;
> -                       i = -1;
> +               pfdentry = fdset_find_entry_locked(pfdset, fd);
> +               if (pfdentry != NULL && pfdentry->busy == 0) {
> +                       fdset_del_locked(pfdset, pfdentry);
> +                       pfdentry = NULL;
>                }
>                pthread_mutex_unlock(&pfdset->fd_mutex);
> -       } while (i != -1);
> -
> -       fdset_sync(pfdset);
> -
> -       return dat;
> +       } while (pfdentry != NULL);
> }
> 
> /**
> @@ -415,28 +264,22 @@ fdset_del(struct fdset *pfdset, int fd)
> int
> fdset_try_del(struct fdset *pfdset, int fd)
> {
> -       int i;
> +       struct fdentry *pfdentry;
> 
>        if (pfdset == NULL || fd == -1)
>                return -2;
> 
>        pthread_mutex_lock(&pfdset->fd_mutex);
> -       i = fdset_find_fd(pfdset, fd);
> -       if (i != -1 && pfdset->fd[i].busy) {
> +       pfdentry = fdset_find_entry_locked(pfdset, fd);
> +       if (pfdentry != NULL && pfdentry->busy != 0) {
>                pthread_mutex_unlock(&pfdset->fd_mutex);
>                return -1;
>        }
> 
> -       if (i != -1) {
> -               pfdset->fd[i].fd = -1;
> -               pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
> -               pfdset->fd[i].dat = NULL;
> -       }
> +       if (pfdentry != NULL)
> +               fdset_del_locked(pfdset, pfdentry);
> 
>        pthread_mutex_unlock(&pfdset->fd_mutex);
> -
> -       fdset_sync(pfdset);
> -
>        return 0;
> }
> 
> @@ -453,53 +296,29 @@ static uint32_t
> fdset_event_dispatch(void *arg)
> {
>        int i;
> -       struct pollfd *pfd;
> -       struct fdentry *pfdentry;
>        fd_cb rcb, wcb;
>        void *dat;
>        int fd, numfds;
>        int remove1, remove2;
> -       int need_shrink;
>        struct fdset *pfdset = arg;
> -       int val;
> 
>        if (pfdset == NULL)
>                return 0;
> 
>        while (1) {
> +               struct epoll_event events[MAX_FDS];
> +               struct fdentry *pfdentry;
> 
> -               /*
> -                * When poll is blocked, other threads might unregister
> -                * listenfds from and register new listenfds into fdset.
> -                * When poll returns, the entries for listenfds in the fdset
> -                * might have been updated. It is ok if there is unwanted call
> -                * for new listenfds.
> -                */
> -               pthread_mutex_lock(&pfdset->fd_mutex);
> -               numfds = pfdset->num;
> -               pthread_mutex_unlock(&pfdset->fd_mutex);
> -
> -               pthread_mutex_lock(&pfdset->fd_polling_mutex);
> -               val = poll(pfdset->rwfds, numfds, 1000 /* millisecs */);
> -               pthread_mutex_unlock(&pfdset->fd_polling_mutex);
> -               if (val < 0)
> +               numfds = epoll_wait(pfdset->epfd, events, RTE_DIM(events), 1000);
> +               if (numfds < 0)
>                        continue;
> 
> -               need_shrink = 0;
>                for (i = 0; i < numfds; i++) {
>                        pthread_mutex_lock(&pfdset->fd_mutex);
> 
> -                       pfdentry = &pfdset->fd[i];
> -                       fd = pfdentry->fd;
> -                       pfd = &pfdset->rwfds[i];
> -
> -                       if (fd < 0) {
> -                               need_shrink = 1;
> -                               pthread_mutex_unlock(&pfdset->fd_mutex);
> -                               continue;
> -                       }
> -
> -                       if (!pfd->revents) {
> +                       fd = events[i].data.fd;
> +                       pfdentry = fdset_find_entry_locked(pfdset, fd);
> +                       if (pfdentry == NULL) {
>                                pthread_mutex_unlock(&pfdset->fd_mutex);
>                                continue;
>                        }
> @@ -513,9 +332,9 @@ fdset_event_dispatch(void *arg)
> 
>                        pthread_mutex_unlock(&pfdset->fd_mutex);
> 
> -                       if (rcb && pfd->revents & (POLLIN | FDPOLLERR))
> +                       if (rcb && events[i].events & (EPOLLIN | EPOLLERR | EPOLLHUP))
>                                rcb(fd, dat, &remove1);
> -                       if (wcb && pfd->revents & (POLLOUT | FDPOLLERR))
> +                       if (wcb && events[i].events & (EPOLLOUT | EPOLLERR | EPOLLHUP))
>                                wcb(fd, dat, &remove2);
>                        pfdentry->busy = 0;
>                        /*
> @@ -524,23 +343,13 @@ fdset_event_dispatch(void *arg)
>                         * directly.
>                         */
>                        /*
> -                        * When we are to clean up the fd from fdset,
> -                        * because the fd is closed in the cb,
> -                        * the old fd val could be reused by when creates new
> -                        * listen fd in another thread, we couldn't call
> -                        * fdset_del.
> +                        * A concurrent fdset_del may have been waiting for the
> +                        * fdentry not to be busy, so we can't call
> +                        * fdset_del_locked().
>                         */
> -                       if (remove1 || remove2) {
> -                               pfdentry->fd = -1;
> -                               need_shrink = 1;
> -                       }
> +                       if (remove1 || remove2)
> +                               fdset_del(pfdset, fd);
>                }
> -
> -               if (need_shrink)
> -                       fdset_shrink(pfdset);
> -
> -               if (pfdset->destroy)
> -                       break;

I guess we want to keep the destroy logic

Thanks,
Chenbo

>        }
> 
>        return 0;
> diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
> index 079fa0155f..6398343a6a 100644
> --- a/lib/vhost/fd_man.h
> +++ b/lib/vhost/fd_man.h
> @@ -6,7 +6,7 @@
> #define _FD_MAN_H_
> #include <pthread.h>
> #include <poll.h>
> -#include <stdbool.h>
> +#include <sys/queue.h>
> 
> struct fdset;
> 
> @@ -19,8 +19,7 @@ struct fdset *fdset_init(const char *name);
> int fdset_add(struct fdset *pfdset, int fd,
>        fd_cb rcb, fd_cb wcb, void *dat);
> 
> -void *fdset_del(struct fdset *pfdset, int fd);
> -
> +void fdset_del(struct fdset *pfdset, int fd);
> int fdset_try_del(struct fdset *pfdset, int fd);
> 
> #endif
> --
> 2.44.0
> 

Re: [PATCH v3 5/5] vhost: manage FD with epoll

[Maxime Coquelin]

Hi Chenbo,

On 4/28/24 05:22, Chenbo Xia wrote:
> Hi Maxime,
> 
>> On Apr 9, 2024, at 19:48, Maxime Coquelin <maxime.coquelin@redhat.com> wrote:
>>
>> External email: Use caution opening links or attachments
>>
>>
>> From: David Marchand <david.marchand@redhat.com>
>>
>> Switch to epoll so that the concern over the poll() fd array
>> is removed.
>> Add a simple list of used entries and track the next free entry.
>>
>> epoll() is thread safe, we no more need a synchronization
>> mechanism and so can remove the notification pipe.
>>
>> Signed-off-by: David Marchand <david.marchand@redhat.com>
>> Signed-off-by: Maxime Coquelin <maxime.coquelin@redhat.com>
>> ---
>> lib/vhost/fd_man.c | 399 ++++++++++++---------------------------------
>> lib/vhost/fd_man.h |   5 +-
>> 2 files changed, 106 insertions(+), 298 deletions(-)
>>
>> diff --git a/lib/vhost/fd_man.c b/lib/vhost/fd_man.c
>> index 8b47c97d45..a4a2965da1 100644
>> --- a/lib/vhost/fd_man.c
>> +++ b/lib/vhost/fd_man.c
>> @@ -3,9 +3,9 @@
>>   */
>>
>> #include <errno.h>
>> -#include <pthread.h>
>> #include <stdio.h>
>> #include <string.h>
>> +#include <sys/epoll.h>
>> #include <unistd.h>
>>
>> #include <rte_common.h>
>> @@ -21,49 +21,34 @@ RTE_LOG_REGISTER_SUFFIX(vhost_fdset_logtype, fdset, INFO);
>> #define VHOST_FDMAN_LOG(level, ...) \
>>         RTE_LOG_LINE(level, VHOST_FDMAN, "" __VA_ARGS__)
>>
>> -#define FDPOLLERR (POLLERR | POLLHUP | POLLNVAL)
>> -
>> struct fdentry {
>>         int fd;         /* -1 indicates this entry is empty */
>>         fd_cb rcb;      /* callback when this fd is readable. */
>>         fd_cb wcb;      /* callback when this fd is writeable.*/
>>         void *dat;      /* fd context */
>>         int busy;       /* whether this entry is being used in cb. */
>> +       LIST_ENTRY(fdentry) next;
>> };
>>
>> struct fdset {
>>         char name[RTE_THREAD_NAME_SIZE];
>> -       struct pollfd rwfds[MAX_FDS];
>> +       int epfd;
>>         struct fdentry fd[MAX_FDS];
>> +       LIST_HEAD(, fdentry) fdlist;
>> +       int next_free_idx;
>>         rte_thread_t tid;
>>         pthread_mutex_t fd_mutex;
>> -       pthread_mutex_t fd_polling_mutex;
>> -       int num;        /* current fd number of this fdset */
>> -
>> -       union pipefds {
>> -               struct {
>> -                       int pipefd[2];
>> -               };
>> -               struct {
>> -                       int readfd;
>> -                       int writefd;
>> -               };
>> -       } u;
>> -
>> -       pthread_mutex_t sync_mutex;
>> -       pthread_cond_t sync_cond;
>> -       bool sync;
>> +
> 
> Not sure this blank line is intended or not :)

It isn't, removing it in next revision.

> 
>>         bool destroy;
>> };
>>
>> -static int fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat);
>> -static uint32_t fdset_event_dispatch(void *arg);
>> -
>> #define MAX_FDSETS 8
>>
>> static struct fdset *fdsets[MAX_FDSETS];
>> pthread_mutex_t fdsets_mutex = PTHREAD_MUTEX_INITIALIZER;
>>
>> +static uint32_t fdset_event_dispatch(void *arg);
>> +
>> static struct fdset *
>> fdset_lookup(const char *name)
>> {
>> @@ -96,166 +81,6 @@ fdset_insert(struct fdset *fdset)
>>         return -1;
>> }
>>
>> -static void
>> -fdset_pipe_read_cb(int readfd, void *dat,
>> -                  int *remove __rte_unused)
>> -{
>> -       char charbuf[16];
>> -       struct fdset *fdset = dat;
>> -       int r = read(readfd, charbuf, sizeof(charbuf));
>> -       /*
>> -        * Just an optimization, we don't care if read() failed
>> -        * so ignore explicitly its return value to make the
>> -        * compiler happy
>> -        */
>> -       RTE_SET_USED(r);
>> -
>> -       pthread_mutex_lock(&fdset->sync_mutex);
>> -       fdset->sync = true;
>> -       pthread_cond_broadcast(&fdset->sync_cond);
>> -       pthread_mutex_unlock(&fdset->sync_mutex);
>> -}
>> -
>> -static void
>> -fdset_pipe_uninit(struct fdset *fdset)
>> -{
>> -       fdset_del(fdset, fdset->u.readfd);
>> -       close(fdset->u.readfd);
>> -       fdset->u.readfd = -1;
>> -       close(fdset->u.writefd);
>> -       fdset->u.writefd = -1;
>> -}
>> -
>> -static int
>> -fdset_pipe_init(struct fdset *fdset)
>> -{
>> -       int ret;
>> -
>> -       pthread_mutex_init(&fdset->sync_mutex, NULL);
>> -       pthread_cond_init(&fdset->sync_cond, NULL);
>> -
>> -       if (pipe(fdset->u.pipefd) < 0) {
>> -               VHOST_FDMAN_LOG(ERR,
>> -                       "failed to create pipe for vhost fdset");
>> -               return -1;
>> -       }
>> -
>> -       ret = fdset_add_no_sync(fdset, fdset->u.readfd,
>> -                       fdset_pipe_read_cb, NULL, fdset);
>> -       if (ret < 0) {
>> -               VHOST_FDMAN_LOG(ERR,
>> -                       "failed to add pipe readfd %d into vhost server fdset",
>> -                       fdset->u.readfd);
>> -
>> -               fdset_pipe_uninit(fdset);
>> -               return -1;
>> -       }
>> -
>> -       return 0;
>> -}
>> -
>> -static void
>> -fdset_sync(struct fdset *fdset)
>> -{
>> -       int ret;
>> -
>> -       pthread_mutex_lock(&fdset->sync_mutex);
>> -
>> -       fdset->sync = false;
>> -       ret = write(fdset->u.writefd, "1", 1);
>> -       if (ret < 0) {
>> -               VHOST_FDMAN_LOG(ERR,
>> -                       "Failed to write to notification pipe: %s",
>> -                       strerror(errno));
>> -               goto out_unlock;
>> -       }
>> -
>> -       while (!fdset->sync)
>> -               pthread_cond_wait(&fdset->sync_cond, &fdset->sync_mutex);
>> -
>> -out_unlock:
>> -       pthread_mutex_unlock(&fdset->sync_mutex);
>> -}
>> -
>> -static int
>> -get_last_valid_idx(struct fdset *pfdset, int last_valid_idx)
>> -{
>> -       int i;
>> -
>> -       for (i = last_valid_idx; i >= 0 && pfdset->fd[i].fd == -1; i--)
>> -               ;
>> -
>> -       return i;
>> -}
>> -
>> -static void
>> -fdset_move(struct fdset *pfdset, int dst, int src)
>> -{
>> -       pfdset->fd[dst]    = pfdset->fd[src];
>> -       pfdset->rwfds[dst] = pfdset->rwfds[src];
>> -}
>> -
>> -static void
>> -fdset_shrink_nolock(struct fdset *pfdset)
>> -{
>> -       int i;
>> -       int last_valid_idx = get_last_valid_idx(pfdset, pfdset->num - 1);
>> -
>> -       for (i = 0; i < last_valid_idx; i++) {
>> -               if (pfdset->fd[i].fd != -1)
>> -                       continue;
>> -
>> -               fdset_move(pfdset, i, last_valid_idx);
>> -               last_valid_idx = get_last_valid_idx(pfdset, last_valid_idx - 1);
>> -       }
>> -       pfdset->num = last_valid_idx + 1;
>> -}
>> -
>> -/*
>> - * Find deleted fd entries and remove them
>> - */
>> -static void
>> -fdset_shrink(struct fdset *pfdset)
>> -{
>> -       pthread_mutex_lock(&pfdset->fd_mutex);
>> -       fdset_shrink_nolock(pfdset);
>> -       pthread_mutex_unlock(&pfdset->fd_mutex);
>> -}
>> -
>> -/**
>> - * Returns the index in the fdset for a given fd.
>> - * @return
>> - *   index for the fd, or -1 if fd isn't in the fdset.
>> - */
>> -static int
>> -fdset_find_fd(struct fdset *pfdset, int fd)
>> -{
>> -       int i;
>> -
>> -       for (i = 0; i < pfdset->num && pfdset->fd[i].fd != fd; i++)
>> -               ;
>> -
>> -       return i == pfdset->num ? -1 : i;
>> -}
>> -
>> -static void
>> -fdset_add_fd(struct fdset *pfdset, int idx, int fd,
>> -       fd_cb rcb, fd_cb wcb, void *dat)
>> -{
>> -       struct fdentry *pfdentry = &pfdset->fd[idx];
>> -       struct pollfd *pfd = &pfdset->rwfds[idx];
>> -
>> -       pfdentry->fd  = fd;
>> -       pfdentry->rcb = rcb;
>> -       pfdentry->wcb = wcb;
>> -       pfdentry->dat = dat;
>> -
>> -       pfd->fd = fd;
>> -       pfd->events  = rcb ? POLLIN : 0;
>> -       pfd->events |= wcb ? POLLOUT : 0;
>> -       pfd->revents = 0;
>> -}
>> -
>> struct fdset *
>> fdset_init(const char *name)
>> {
>> @@ -284,16 +109,20 @@ fdset_init(const char *name)
>>         rte_strscpy(fdset->name, name, RTE_THREAD_NAME_SIZE);
>>
>>         pthread_mutex_init(&fdset->fd_mutex, NULL);
>> -       pthread_mutex_init(&fdset->fd_polling_mutex, NULL);
>>
>> -       for (i = 0; i < MAX_FDS; i++) {
>> +       for (i = 0; i < (int)RTE_DIM(fdset->fd); i++) {
>>                 fdset->fd[i].fd = -1;
>>                 fdset->fd[i].dat = NULL;
>>         }
>> -       fdset->num = 0;
>> +       LIST_INIT(&fdset->fdlist);
>>
>> -       if (fdset_pipe_init(fdset)) {
>> -               VHOST_FDMAN_LOG(ERR, "Failed to init pipe for %s", name);
>> +       /*
>> +        * Any non-zero value would work (see man epoll_create),
>> +        * but pass MAX_FDS for consistency.
>> +        */
>> +       fdset->epfd = epoll_create(MAX_FDS);
>> +       if (fdset->epfd < 0) {
>> +               VHOST_FDMAN_LOG(ERR, "failed to create epoll for %s fdset", name);
> 
> failed -> Failed like other logs

Ack.

> 
>>                 goto err_free;
>>         }
>>
>> @@ -301,7 +130,7 @@ fdset_init(const char *name)
>>                                         fdset_event_dispatch, fdset)) {
>>                 VHOST_FDMAN_LOG(ERR, "Failed to create %s event dispatch thread",
>>                                 fdset->name);
>> -               goto err_pipe;
>> +               goto err_epoll;
>>         }
>>
>>         if (fdset_insert(fdset)) {
>> @@ -315,10 +144,9 @@ fdset_init(const char *name)
>>
>> err_thread:
>>         fdset->destroy = true;
>> -       fdset_sync(fdset);
>>         rte_thread_join(fdset->tid, &val);
>> -err_pipe:
>> -       fdset_pipe_uninit(fdset);
>> +err_epoll:
>> +       close(fdset->epfd);
>> err_free:
>>         rte_free(fdset);
>> err_unlock:
>> @@ -330,78 +158,99 @@ fdset_init(const char *name)
>> /**
>>   * Register the fd in the fdset with read/write handler and context.
>>   */
>> -static int
>> -fdset_add_no_sync(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
>> +int
>> +fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
>> {
>> -       int i;
>> +       struct fdentry *pfdentry;
>> +       struct epoll_event ev;
>>
>>         if (pfdset == NULL || fd == -1)
>>                 return -1;
>>
>>         pthread_mutex_lock(&pfdset->fd_mutex);
>> -       i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
>> -       if (i == -1) {
>> -               pthread_mutex_lock(&pfdset->fd_polling_mutex);
>> -               fdset_shrink_nolock(pfdset);
>> -               pthread_mutex_unlock(&pfdset->fd_polling_mutex);
>> -               i = pfdset->num < MAX_FDS ? pfdset->num++ : -1;
>> -               if (i == -1) {
>> -                       pthread_mutex_unlock(&pfdset->fd_mutex);
>> -                       return -2;
>> -               }
>> +       if (pfdset->next_free_idx >= (int)RTE_DIM(pfdset->fd)) {
>> +               pthread_mutex_unlock(&pfdset->fd_mutex);
>> +               return -2;
>>         }
>>
>> -       fdset_add_fd(pfdset, i, fd, rcb, wcb, dat);
>> +       pfdentry = &pfdset->fd[pfdset->next_free_idx];
>> +       pfdentry->fd  = fd;
>> +       pfdentry->rcb = rcb;
>> +       pfdentry->wcb = wcb;
>> +       pfdentry->dat = dat;
>> +
>> +       LIST_INSERT_HEAD(&pfdset->fdlist, pfdentry, next);
>> +
>> +       /* Find next free slot */
>> +       pfdset->next_free_idx++;
>> +       for (; pfdset->next_free_idx < (int)RTE_DIM(pfdset->fd); pfdset->next_free_idx++) {
>> +               if (pfdset->fd[pfdset->next_free_idx].fd != -1)
>> +                       continue;
>> +               break;
>> +       }
>>         pthread_mutex_unlock(&pfdset->fd_mutex);
>>
>> +       ev.events = EPOLLERR;
>> +       ev.events |= rcb ? EPOLLIN : 0;
>> +       ev.events |= wcb ? EPOLLOUT : 0;
>> +       ev.data.fd = fd;
>> +
>> +       if (epoll_ctl(pfdset->epfd, EPOLL_CTL_ADD, fd, &ev) == -1)
>> +               VHOST_FDMAN_LOG(ERR, "could not add %d fd to %d epfd: %s",
>> +                       fd, pfdset->epfd, strerror(errno));
> 
> Should not return 0 if this fails ?

Right, in upcoming revision I'm improving the error handling with doing
some refactoring for the pdset management.

> 
>> +
>>         return 0;
>> }
>>
>> -int
>> -fdset_add(struct fdset *pfdset, int fd, fd_cb rcb, fd_cb wcb, void *dat)
>> +static struct fdentry *
>> +fdset_find_entry_locked(struct fdset *pfdset, int fd)
>> {
>> -       int ret;
>> +       struct fdentry *pfdentry;
>>
>> -       ret = fdset_add_no_sync(pfdset, fd, rcb, wcb, dat);
>> -       if (ret < 0)
>> -               return ret;
>> +       LIST_FOREACH(pfdentry, &pfdset->fdlist, next) {
>> +               if (pfdentry->fd != fd)
>> +                       continue;
>> +               return pfdentry;
>> +       }
>>
>> -       fdset_sync(pfdset);
>> +       return NULL;
>> +}
>>
>> -       return 0;
>> +static void
>> +fdset_del_locked(struct fdset *pfdset, struct fdentry *pfdentry)
>> +{
>> +       int entry_idx;
>> +
>> +       if (epoll_ctl(pfdset->epfd, EPOLL_CTL_DEL, pfdentry->fd, NULL) == -1)
>> +               VHOST_FDMAN_LOG(ERR, "could not remove %d fd from %d epfd: %s",
>> +                       pfdentry->fd, pfdset->epfd, strerror(errno));
>> +
>> +       pfdentry->fd = -1;
>> +       pfdentry->rcb = pfdentry->wcb = NULL;
>> +       pfdentry->dat = NULL;
>> +       entry_idx = pfdentry - pfdset->fd;
>> +       if (entry_idx < pfdset->next_free_idx)
>> +               pfdset->next_free_idx = entry_idx;
>> +       LIST_REMOVE(pfdentry, next);
>> }
>>
>> -/**
>> - *  Unregister the fd from the fdset.
>> - *  Returns context of a given fd or NULL.
>> - */
>> -void *
>> +void
>> fdset_del(struct fdset *pfdset, int fd)
>> {
>> -       int i;
>> -       void *dat = NULL;
>> +       struct fdentry *pfdentry;
>>
>>         if (pfdset == NULL || fd == -1)
>> -               return NULL;
>> +               return;
>>
>>         do {
>>                 pthread_mutex_lock(&pfdset->fd_mutex);
>> -
>> -               i = fdset_find_fd(pfdset, fd);
>> -               if (i != -1 && pfdset->fd[i].busy == 0) {
>> -                       /* busy indicates r/wcb is executing! */
>> -                       dat = pfdset->fd[i].dat;
>> -                       pfdset->fd[i].fd = -1;
>> -                       pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
>> -                       pfdset->fd[i].dat = NULL;
>> -                       i = -1;
>> +               pfdentry = fdset_find_entry_locked(pfdset, fd);
>> +               if (pfdentry != NULL && pfdentry->busy == 0) {
>> +                       fdset_del_locked(pfdset, pfdentry);
>> +                       pfdentry = NULL;
>>                 }
>>                 pthread_mutex_unlock(&pfdset->fd_mutex);
>> -       } while (i != -1);
>> -
>> -       fdset_sync(pfdset);
>> -
>> -       return dat;
>> +       } while (pfdentry != NULL);
>> }
>>
>> /**
>> @@ -415,28 +264,22 @@ fdset_del(struct fdset *pfdset, int fd)
>> int
>> fdset_try_del(struct fdset *pfdset, int fd)
>> {
>> -       int i;
>> +       struct fdentry *pfdentry;
>>
>>         if (pfdset == NULL || fd == -1)
>>                 return -2;
>>
>>         pthread_mutex_lock(&pfdset->fd_mutex);
>> -       i = fdset_find_fd(pfdset, fd);
>> -       if (i != -1 && pfdset->fd[i].busy) {
>> +       pfdentry = fdset_find_entry_locked(pfdset, fd);
>> +       if (pfdentry != NULL && pfdentry->busy != 0) {
>>                 pthread_mutex_unlock(&pfdset->fd_mutex);
>>                 return -1;
>>         }
>>
>> -       if (i != -1) {
>> -               pfdset->fd[i].fd = -1;
>> -               pfdset->fd[i].rcb = pfdset->fd[i].wcb = NULL;
>> -               pfdset->fd[i].dat = NULL;
>> -       }
>> +       if (pfdentry != NULL)
>> +               fdset_del_locked(pfdset, pfdentry);
>>
>>         pthread_mutex_unlock(&pfdset->fd_mutex);
>> -
>> -       fdset_sync(pfdset);
>> -
>>         return 0;
>> }
>>
>> @@ -453,53 +296,29 @@ static uint32_t
>> fdset_event_dispatch(void *arg)
>> {
>>         int i;
>> -       struct pollfd *pfd;
>> -       struct fdentry *pfdentry;
>>         fd_cb rcb, wcb;
>>         void *dat;
>>         int fd, numfds;
>>         int remove1, remove2;
>> -       int need_shrink;
>>         struct fdset *pfdset = arg;
>> -       int val;
>>
>>         if (pfdset == NULL)
>>                 return 0;
>>
>>         while (1) {
>> +               struct epoll_event events[MAX_FDS];
>> +               struct fdentry *pfdentry;
>>
>> -               /*
>> -                * When poll is blocked, other threads might unregister
>> -                * listenfds from and register new listenfds into fdset.
>> -                * When poll returns, the entries for listenfds in the fdset
>> -                * might have been updated. It is ok if there is unwanted call
>> -                * for new listenfds.
>> -                */
>> -               pthread_mutex_lock(&pfdset->fd_mutex);
>> -               numfds = pfdset->num;
>> -               pthread_mutex_unlock(&pfdset->fd_mutex);
>> -
>> -               pthread_mutex_lock(&pfdset->fd_polling_mutex);
>> -               val = poll(pfdset->rwfds, numfds, 1000 /* millisecs */);
>> -               pthread_mutex_unlock(&pfdset->fd_polling_mutex);
>> -               if (val < 0)
>> +               numfds = epoll_wait(pfdset->epfd, events, RTE_DIM(events), 1000);
>> +               if (numfds < 0)
>>                         continue;
>>
>> -               need_shrink = 0;
>>                 for (i = 0; i < numfds; i++) {
>>                         pthread_mutex_lock(&pfdset->fd_mutex);
>>
>> -                       pfdentry = &pfdset->fd[i];
>> -                       fd = pfdentry->fd;
>> -                       pfd = &pfdset->rwfds[i];
>> -
>> -                       if (fd < 0) {
>> -                               need_shrink = 1;
>> -                               pthread_mutex_unlock(&pfdset->fd_mutex);
>> -                               continue;
>> -                       }
>> -
>> -                       if (!pfd->revents) {
>> +                       fd = events[i].data.fd;
>> +                       pfdentry = fdset_find_entry_locked(pfdset, fd);
>> +                       if (pfdentry == NULL) {
>>                                 pthread_mutex_unlock(&pfdset->fd_mutex);
>>                                 continue;
>>                         }
>> @@ -513,9 +332,9 @@ fdset_event_dispatch(void *arg)
>>
>>                         pthread_mutex_unlock(&pfdset->fd_mutex);
>>
>> -                       if (rcb && pfd->revents & (POLLIN | FDPOLLERR))
>> +                       if (rcb && events[i].events & (EPOLLIN | EPOLLERR | EPOLLHUP))
>>                                 rcb(fd, dat, &remove1);
>> -                       if (wcb && pfd->revents & (POLLOUT | FDPOLLERR))
>> +                       if (wcb && events[i].events & (EPOLLOUT | EPOLLERR | EPOLLHUP))
>>                                 wcb(fd, dat, &remove2);
>>                         pfdentry->busy = 0;
>>                         /*
>> @@ -524,23 +343,13 @@ fdset_event_dispatch(void *arg)
>>                          * directly.
>>                          */
>>                         /*
>> -                        * When we are to clean up the fd from fdset,
>> -                        * because the fd is closed in the cb,
>> -                        * the old fd val could be reused by when creates new
>> -                        * listen fd in another thread, we couldn't call
>> -                        * fdset_del.
>> +                        * A concurrent fdset_del may have been waiting for the
>> +                        * fdentry not to be busy, so we can't call
>> +                        * fdset_del_locked().
>>                          */
>> -                       if (remove1 || remove2) {
>> -                               pfdentry->fd = -1;
>> -                               need_shrink = 1;
>> -                       }
>> +                       if (remove1 || remove2)
>> +                               fdset_del(pfdset, fd);
>>                 }
>> -
>> -               if (need_shrink)
>> -                       fdset_shrink(pfdset);
>> -
>> -               if (pfdset->destroy)
>> -                       break;
> 
> I guess we want to keep the destroy logic

Right, good catch!
Adding it back.

Thanks for the review,
Maxime
> Thanks,
> Chenbo
> 
>>         }
>>
>>         return 0;
>> diff --git a/lib/vhost/fd_man.h b/lib/vhost/fd_man.h
>> index 079fa0155f..6398343a6a 100644
>> --- a/lib/vhost/fd_man.h
>> +++ b/lib/vhost/fd_man.h
>> @@ -6,7 +6,7 @@
>> #define _FD_MAN_H_
>> #include <pthread.h>
>> #include <poll.h>
>> -#include <stdbool.h>
>> +#include <sys/queue.h>
>>
>> struct fdset;
>>
>> @@ -19,8 +19,7 @@ struct fdset *fdset_init(const char *name);
>> int fdset_add(struct fdset *pfdset, int fd,
>>         fd_cb rcb, fd_cb wcb, void *dat);
>>
>> -void *fdset_del(struct fdset *pfdset, int fd);
>> -
>> +void fdset_del(struct fdset *pfdset, int fd);
>> int fdset_try_del(struct fdset *pfdset, int fd);
>>
>> #endif
>> --
>> 2.44.0
>>
>