[dpdk-dev] [RFC 2/4] vhost: add support for SocketPair Broker

[Ilya Maximets <i.maximets@ovn.org>]

New flag RTE_VHOST_USER_SOCKETPAIR_BROKER to say that provided socket
is a path to SocketPair Broker socket in a following format:

  '/path/to/socketpair/broker.sock,broker-key=<key>'

This format is chosen to avoid lots of code changes and refactoring
inside the vhost library, mainly because vhost library treats
socket path as a unique device identifier.

'<key>' is a broker key that will be used by a broker to identify
two clients that needs to be paired together, i.e. vhost device
will be connected with a client that provided the same key.

libspbroker needed for a build.

Signed-off-by: Ilya Maximets <i.maximets@ovn.org>
---
 doc/guides/prog_guide/vhost_lib.rst |  10 ++
 lib/librte_vhost/meson.build        |   7 +
 lib/librte_vhost/rte_vhost.h        |   1 +
 lib/librte_vhost/socket.c           | 245 +++++++++++++++++++++++++---
 4 files changed, 237 insertions(+), 26 deletions(-)

diff --git a/doc/guides/prog_guide/vhost_lib.rst b/doc/guides/prog_guide/vhost_lib.rst
index dc29229167..f0f0d3fde7 100644
--- a/doc/guides/prog_guide/vhost_lib.rst
+++ b/doc/guides/prog_guide/vhost_lib.rst
@@ -118,6 +118,16 @@ The following is an overview of some key Vhost API functions:
 
     It is disabled by default.
 
+  - ``RTE_VHOST_USER_SOCKETPAIR_BROKER``
+
+    Enabling of this flag makes vhost library to treat socket ``path`` as a
+    path to SocketPair Broker.  In this case ``path`` should include
+    ``,broker-key=<key>`` after the actual broker's socket path.  ``<key>``
+    will be used as a broker key, so it will be able to connect 2 processes
+    that provided the same key.
+
+    Incompatible with ``RTE_VHOST_USER_NO_RECONNECT``.
+
 * ``rte_vhost_driver_set_features(path, features)``
 
   This function sets the feature bits the vhost-user driver supports. The
diff --git a/lib/librte_vhost/meson.build b/lib/librte_vhost/meson.build
index 6185deab33..3292edcb52 100644
--- a/lib/librte_vhost/meson.build
+++ b/lib/librte_vhost/meson.build
@@ -15,6 +15,13 @@ elif (toolchain == 'clang' and cc.version().version_compare('>=3.7.0'))
 elif (toolchain == 'icc' and cc.version().version_compare('>=16.0.0'))
 	cflags += '-DVHOST_ICC_UNROLL_PRAGMA'
 endif
+
+spbroker_dep = dependency('spbroker', required: false)
+if spbroker_dep.found()
+	dpdk_conf.set('RTE_LIBRTE_VHOST_SOCKETPAIR_BROKER', 1)
+	ext_deps += spbroker_dep
+endif
+
 dpdk_conf.set('RTE_LIBRTE_VHOST_POSTCOPY',
 	      cc.has_header('linux/userfaultfd.h'))
 cflags += '-fno-strict-aliasing'
diff --git a/lib/librte_vhost/rte_vhost.h b/lib/librte_vhost/rte_vhost.h
index 010f160869..87662c9f7f 100644
--- a/lib/librte_vhost/rte_vhost.h
+++ b/lib/librte_vhost/rte_vhost.h
@@ -36,6 +36,7 @@ extern "C" {
 /* support only linear buffers (no chained mbufs) */
 #define RTE_VHOST_USER_LINEARBUF_SUPPORT	(1ULL << 6)
 #define RTE_VHOST_USER_ASYNC_COPY	(1ULL << 7)
+#define RTE_VHOST_USER_SOCKETPAIR_BROKER	(1ULL << 8)
 
 /* Features. */
 #ifndef VIRTIO_NET_F_GUEST_ANNOUNCE
diff --git a/lib/librte_vhost/socket.c b/lib/librte_vhost/socket.c
index 0169d36481..f0a1c9044c 100644
--- a/lib/librte_vhost/socket.c
+++ b/lib/librte_vhost/socket.c
@@ -16,6 +16,10 @@
 #include <fcntl.h>
 #include <pthread.h>
 
+#ifdef RTE_LIBRTE_VHOST_SOCKETPAIR_BROKER
+#include <socketpair-broker/helper.h>
+#endif
+
 #include <rte_log.h>
 
 #include "fd_man.h"
@@ -33,9 +37,11 @@ struct vhost_user_socket {
 	struct vhost_user_connection_list conn_list;
 	pthread_mutex_t conn_mutex;
 	char *path;
+	char *broker_key;
 	int socket_fd;
 	struct sockaddr_un un;
 	bool is_server;
+	bool is_broker;
 	bool reconnect;
 	bool iommu_support;
 	bool use_builtin_virtio_net;
@@ -81,7 +87,8 @@ struct vhost_user {
 static void vhost_user_server_new_connection(int fd, void *data, int *remove);
 static void vhost_user_read_cb(int fd, void *dat, int *remove);
 static int create_unix_socket(struct vhost_user_socket *vsocket);
-static int vhost_user_start_client(struct vhost_user_socket *vsocket);
+static int recreate_unix_socket(struct vhost_user_socket *vsocket);
+static int vhost_user_start(struct vhost_user_socket *vsocket);
 
 static struct vhost_user vhost_user = {
 	.fdset = {
@@ -283,6 +290,81 @@ vhost_user_add_connection(int fd, struct vhost_user_socket *vsocket)
 	close(fd);
 }
 
+#ifdef RTE_LIBRTE_VHOST_SOCKETPAIR_BROKER
+static int
+vhost_user_broker_msg_handler(int fd, struct vhost_user_socket *vsocket)
+{
+	int peer_fd;
+	char *err;
+
+	peer_fd = sp_broker_receive_set_pair(fd, &err);
+	if (peer_fd < 0) {
+		VHOST_LOG_CONFIG(ERR,
+			"failed to receive SP_BROKER_SET_PAIR on fd %d: %s\n",
+			fd, err);
+		free(err);
+		return -1;
+	}
+
+	VHOST_LOG_CONFIG(INFO, "new vhost user connection is %d\n", peer_fd);
+	vhost_user_add_connection(peer_fd, vsocket);
+	return 0;
+}
+
+static void
+vhost_user_broker_msg_cb(int connfd, void *dat, int *remove)
+{
+	struct vhost_user_socket *vsocket = dat;
+	int ret;
+
+	ret = vhost_user_broker_msg_handler(connfd, vsocket);
+
+	/* Don't need a broker connection anymore. */
+	*remove = 1;
+
+	if (ret < 0) {
+		recreate_unix_socket(vsocket);
+		vhost_user_start(vsocket);
+	}
+}
+
+static int
+vhost_user_start_broker_connection(
+	int fd __rte_unused,
+	struct vhost_user_socket *vsocket __rte_unused)
+{
+	char *err;
+	int ret;
+
+	ret = sp_broker_send_get_pair(fd, vsocket->broker_key,
+				      vsocket->is_server, &err);
+	if (ret) {
+		VHOST_LOG_CONFIG(ERR,
+			"failed to send SP_BROKER_GET_PAIR request on fd %d: %s\n",
+			fd, err);
+		free(err);
+		return -1;
+	}
+
+	ret = fdset_add(&vhost_user.fdset, fd, vhost_user_broker_msg_cb,
+			NULL, vsocket);
+	if (ret < 0) {
+		VHOST_LOG_CONFIG(ERR,
+			"failed to add broker fd %d to vhost fdset\n", fd);
+		return -1;
+	}
+	return 0;
+}
+#else
+static int
+vhost_user_start_broker_connection(
+	int fd __rte_unused,
+	struct vhost_user_socket *vsocket __rte_unused)
+{
+	return -1;
+}
+#endif
+
 /* call back when there is new vhost-user connection from client  */
 static void
 vhost_user_server_new_connection(int fd, void *dat, int *remove __rte_unused)
@@ -321,7 +403,7 @@ vhost_user_read_cb(int connfd, void *dat, int *remove)
 
 		if (vsocket->reconnect) {
 			create_unix_socket(vsocket);
-			vhost_user_start_client(vsocket);
+			vhost_user_start(vsocket);
 		}
 
 		pthread_mutex_lock(&vsocket->conn_mutex);
@@ -337,14 +419,17 @@ create_unix_socket(struct vhost_user_socket *vsocket)
 {
 	int fd;
 	struct sockaddr_un *un = &vsocket->un;
+	char *broker_key = strstr(vsocket->path, ",broker-key=");
 
 	fd = socket(AF_UNIX, SOCK_STREAM, 0);
 	if (fd < 0)
 		return -1;
-	VHOST_LOG_CONFIG(INFO, "vhost-user %s: socket created, fd: %d\n",
-		vsocket->is_server ? "server" : "client", fd);
+	VHOST_LOG_CONFIG(INFO, "vhost-user %s: %ssocket created, fd: %d\n",
+		vsocket->is_server ? "server" : "client",
+		vsocket->is_broker ? "broker " : "", fd);
 
-	if (!vsocket->is_server && fcntl(fd, F_SETFL, O_NONBLOCK)) {
+	if ((!vsocket->is_server || vsocket->is_broker)
+	    && fcntl(fd, F_SETFL, O_NONBLOCK)) {
 		VHOST_LOG_CONFIG(ERR,
 			"vhost-user: can't set nonblocking mode for socket, fd: "
 			"%d (%s)\n", fd, strerror(errno));
@@ -352,12 +437,21 @@ create_unix_socket(struct vhost_user_socket *vsocket)
 		return -1;
 	}
 
+	/* Temporarily limiting the path by the actual path. */
+	if (vsocket->is_broker && broker_key)
+		broker_key[0] = '\0';
+
 	memset(un, 0, sizeof(*un));
 	un->sun_family = AF_UNIX;
 	strncpy(un->sun_path, vsocket->path, sizeof(un->sun_path));
 	un->sun_path[sizeof(un->sun_path) - 1] = '\0';
 
 	vsocket->socket_fd = fd;
+
+	/* Restoring original path. */
+	if (vsocket->is_broker && broker_key)
+		broker_key[0] = ',';
+
 	return 0;
 }
 
@@ -425,7 +519,8 @@ static struct vhost_user_reconnect_list reconn_list;
 static pthread_t reconn_tid;
 
 static int
-vhost_user_connect_nonblock(int fd, struct sockaddr *un, size_t sz)
+vhost_user_connect_nonblock(int fd, struct sockaddr *un, size_t sz,
+			    bool disable_nonblock)
 {
 	int ret, flags;
 
@@ -433,6 +528,9 @@ vhost_user_connect_nonblock(int fd, struct sockaddr *un, size_t sz)
 	if (ret < 0 && errno != EISCONN)
 		return -1;
 
+	if (!disable_nonblock)
+		return 0;
+
 	flags = fcntl(fd, F_GETFL, 0);
 	if (flags < 0) {
 		VHOST_LOG_CONFIG(ERR,
@@ -447,8 +545,22 @@ vhost_user_connect_nonblock(int fd, struct sockaddr *un, size_t sz)
 	return 0;
 }
 
+static int
+recreate_unix_socket(struct vhost_user_socket *vsocket)
+{
+	close(vsocket->socket_fd);
+	if (create_unix_socket(vsocket) < 0) {
+		VHOST_LOG_CONFIG(ERR,
+			"Failed to re-create socket for %s\n",
+			vsocket->un.sun_path);
+		vsocket->socket_fd = -1;
+		return -2;
+	}
+	return 0;
+}
+
 static void *
-vhost_user_client_reconnect(void *arg __rte_unused)
+vhost_user_reconnect(void *arg __rte_unused)
 {
 	int ret;
 	struct vhost_user_reconnect *reconn, *next;
@@ -466,7 +578,8 @@ vhost_user_client_reconnect(void *arg __rte_unused)
 
 			ret = vhost_user_connect_nonblock(reconn->fd,
 						(struct sockaddr *)&reconn->un,
-						sizeof(reconn->un));
+						sizeof(reconn->un),
+						!reconn->vsocket->is_broker);
 			if (ret == -2) {
 				close(reconn->fd);
 				VHOST_LOG_CONFIG(ERR,
@@ -478,8 +591,26 @@ vhost_user_client_reconnect(void *arg __rte_unused)
 				continue;
 
 			VHOST_LOG_CONFIG(INFO,
-				"%s: connected\n", reconn->vsocket->path);
-			vhost_user_add_connection(reconn->fd, reconn->vsocket);
+				"%s: connected\n",
+				reconn->vsocket->un.sun_path);
+
+			if (reconn->vsocket->is_broker) {
+				struct vhost_user_socket *vsocket;
+
+				vsocket = reconn->vsocket;
+				if (vhost_user_start_broker_connection(
+					reconn->fd, vsocket)) {
+					if (recreate_unix_socket(vsocket)) {
+						goto remove_fd;
+					} else {
+						reconn->fd = vsocket->socket_fd;
+						continue;
+					}
+				}
+			} else {
+				vhost_user_add_connection(reconn->fd,
+							  reconn->vsocket);
+			}
 remove_fd:
 			TAILQ_REMOVE(&reconn_list.head, reconn, next);
 			free(reconn);
@@ -505,7 +636,7 @@ vhost_user_reconnect_init(void)
 	TAILQ_INIT(&reconn_list.head);
 
 	ret = rte_ctrl_thread_create(&reconn_tid, "vhost_reconn", NULL,
-			     vhost_user_client_reconnect, NULL);
+			     vhost_user_reconnect, NULL);
 	if (ret != 0) {
 		VHOST_LOG_CONFIG(ERR, "failed to create reconnect thread");
 		if (pthread_mutex_destroy(&reconn_list.mutex)) {
@@ -518,18 +649,25 @@ vhost_user_reconnect_init(void)
 }
 
 static int
-vhost_user_start_client(struct vhost_user_socket *vsocket)
+vhost_user_start(struct vhost_user_socket *vsocket)
 {
 	int ret;
 	int fd = vsocket->socket_fd;
-	const char *path = vsocket->path;
+	const char *path = vsocket->un.sun_path;
 	struct vhost_user_reconnect *reconn;
 
 	ret = vhost_user_connect_nonblock(fd, (struct sockaddr *)&vsocket->un,
-					  sizeof(vsocket->un));
+					  sizeof(vsocket->un),
+					  !vsocket->is_broker);
 	if (ret == 0) {
-		vhost_user_add_connection(fd, vsocket);
-		return 0;
+		if (!vsocket->is_broker) {
+			vhost_user_add_connection(fd, vsocket);
+			return 0;
+		} else if (vhost_user_start_broker_connection(fd, vsocket)) {
+			ret = recreate_unix_socket(vsocket);
+		} else {
+			return 0;
+		}
 	}
 
 	VHOST_LOG_CONFIG(WARNING,
@@ -822,6 +960,11 @@ rte_vhost_driver_get_queue_num(const char *path, uint32_t *queue_num)
 static void
 vhost_user_socket_mem_free(struct vhost_user_socket *vsocket)
 {
+	if (vsocket && vsocket->broker_key) {
+		free(vsocket->broker_key);
+		vsocket->broker_key = NULL;
+	}
+
 	if (vsocket && vsocket->path) {
 		free(vsocket->path);
 		vsocket->path = NULL;
@@ -946,15 +1089,50 @@ rte_vhost_driver_register(const char *path, uint64_t flags)
 #endif
 	}
 
-	if ((flags & RTE_VHOST_USER_CLIENT) != 0) {
+	if ((flags & RTE_VHOST_USER_SOCKETPAIR_BROKER) != 0) {
+#ifndef RTE_LIBRTE_VHOST_SOCKETPAIR_BROKER
+		VHOST_LOG_CONFIG(ERR,
+			"SocketPair Broker requested but not compiled\n");
+		ret = -1;
+		goto out_mutex;
+#endif
+		char *broker_key = strstr(vsocket->path, ",broker-key=");
+
+		if (!broker_key || !broker_key[12]) {
+			VHOST_LOG_CONFIG(ERR,
+				"Connection to SocketPair Broker requested but"
+				" key is not provided: %s\n", vsocket->path);
+			ret = -1;
+			goto out_mutex;
+		}
+		vsocket->is_broker = true;
+		vsocket->broker_key = strdup(broker_key + 12);
+		if (vsocket->broker_key == NULL) {
+			VHOST_LOG_CONFIG(ERR,
+				"error: failed to copy broker key\n");
+			ret = -1;
+			goto out_mutex;
+		}
+	}
+
+	if ((flags & RTE_VHOST_USER_CLIENT) == 0)
+		vsocket->is_server = true;
+
+	if (!vsocket->is_server || vsocket->is_broker) {
 		vsocket->reconnect = !(flags & RTE_VHOST_USER_NO_RECONNECT);
+		if (vsocket->is_broker && !vsocket->reconnect) {
+			VHOST_LOG_CONFIG(ERR,
+				"SocketPair Broker with NO_RECONNECT "
+				"is not supported\n");
+			ret = -1;
+			goto out_mutex;
+		}
 		if (vsocket->reconnect && reconn_tid == 0) {
 			if (vhost_user_reconnect_init() != 0)
 				goto out_mutex;
 		}
-	} else {
-		vsocket->is_server = true;
 	}
+
 	ret = create_unix_socket(vsocket);
 	if (ret < 0) {
 		goto out_mutex;
@@ -1052,7 +1230,23 @@ rte_vhost_driver_unregister(const char *path)
 			}
 			pthread_mutex_unlock(&vsocket->conn_mutex);
 
-			if (vsocket->is_server) {
+			if (vsocket->reconnect) {
+				if (vhost_user_remove_reconnect(vsocket)) {
+					/*
+					 * reconn->fd is a socket_fd for
+					 * client and broker connections and
+					 * it's closed now.
+					 */
+					vsocket->socket_fd = -1;
+				}
+			}
+
+			/*
+			 * socket_fd is still valid for server connection
+			 * or broker connection that is currently connected
+			 * to the broker.
+			 */
+			if (vsocket->socket_fd != -1) {
 				/*
 				 * If r/wcb is executing, release vhost_user's
 				 * mutex lock, and try again since the r/wcb
@@ -1063,13 +1257,12 @@ rte_vhost_driver_unregister(const char *path)
 					pthread_mutex_unlock(&vhost_user.mutex);
 					goto again;
 				}
-
 				close(vsocket->socket_fd);
-				unlink(path);
-			} else if (vsocket->reconnect) {
-				vhost_user_remove_reconnect(vsocket);
 			}
 
+			if (vsocket->is_server && !vsocket->is_broker)
+				unlink(path);
+
 			pthread_mutex_destroy(&vsocket->conn_mutex);
 			vhost_user_socket_mem_free(vsocket);
 
@@ -1152,8 +1345,8 @@ rte_vhost_driver_start(const char *path)
 		}
 	}
 
-	if (vsocket->is_server)
+	if (vsocket->is_server && !vsocket->is_broker)
 		return vhost_user_start_server(vsocket);
 	else
-		return vhost_user_start_client(vsocket);
+		return vhost_user_start(vsocket);
 }
-- 
2.26.2