Adding upstream version 16.2.11+ds.upstream/16.2.11+dsupstream

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 1328 insertions, 0 deletions

diff --git a/src/spdk/module/sock/uring/uring.c b/src/spdk/module/sock/uring/uring.c
new file mode 100644
index 000000000..3066f2d16
--- /dev/null
+++ b/src/spdk/module/sock/uring/uring.c
@@ -0,0 +1,1328 @@
+/*-
+ *   BSD LICENSE
+ *
+ *   Copyright (c) Intel Corporation.
+ *   All rights reserved.
+ *
+ *   Redistribution and use in source and binary forms, with or without
+ *   modification, are permitted provided that the following conditions
+ *   are met:
+ *
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in
+ *       the documentation and/or other materials provided with the
+ *       distribution.
+ *     * Neither the name of Intel Corporation nor the names of its
+ *       contributors may be used to endorse or promote products derived
+ *       from this software without specific prior written permission.
+ *
+ *   THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+ *   "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+ *   LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+ *   A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+ *   OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+ *   SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+ *   LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+ *   DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+ *   THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ *   (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+ *   OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+#include "spdk/stdinc.h"
+#include "spdk/config.h"
+
+#include <sys/epoll.h>
+#include <liburing.h>
+
+#include "spdk/barrier.h"
+#include "spdk/likely.h"
+#include "spdk/log.h"
+#include "spdk/pipe.h"
+#include "spdk/sock.h"
+#include "spdk/string.h"
+#include "spdk/util.h"
+
+#include "spdk_internal/sock.h"
+#include "spdk_internal/assert.h"
+
+#define MAX_TMPBUF 1024
+#define PORTNUMLEN 32
+#define SO_RCVBUF_SIZE (2 * 1024 * 1024)
+#define SO_SNDBUF_SIZE (2 * 1024 * 1024)
+#define SPDK_SOCK_GROUP_QUEUE_DEPTH 4096
+#define IOV_BATCH_SIZE 64
+
+enum spdk_sock_task_type {
+	SPDK_SOCK_TASK_POLLIN = 0,
+	SPDK_SOCK_TASK_WRITE,
+	SPDK_SOCK_TASK_CANCEL,
+};
+
+enum spdk_uring_sock_task_status {
+	SPDK_URING_SOCK_TASK_NOT_IN_USE = 0,
+	SPDK_URING_SOCK_TASK_IN_PROCESS,
+};
+
+struct spdk_uring_task {
+	enum spdk_uring_sock_task_status	status;
+	enum spdk_sock_task_type		type;
+	struct spdk_uring_sock			*sock;
+	struct msghdr				msg;
+	struct iovec				iovs[IOV_BATCH_SIZE];
+	int					iov_cnt;
+	struct spdk_sock_request		*last_req;
+	STAILQ_ENTRY(spdk_uring_task)		link;
+};
+
+struct spdk_uring_sock {
+	struct spdk_sock			base;
+	int					fd;
+	struct spdk_uring_sock_group_impl	*group;
+	struct spdk_uring_task			write_task;
+	struct spdk_uring_task			pollin_task;
+	struct spdk_uring_task			cancel_task;
+	struct spdk_pipe			*recv_pipe;
+	void					*recv_buf;
+	int					recv_buf_sz;
+	bool					pending_recv;
+	int					connection_status;
+	TAILQ_ENTRY(spdk_uring_sock)		link;
+};
+
+struct spdk_uring_sock_group_impl {
+	struct spdk_sock_group_impl		base;
+	struct io_uring				uring;
+	uint32_t				io_inflight;
+	uint32_t				io_queued;
+	uint32_t				io_avail;
+	TAILQ_HEAD(, spdk_uring_sock)		pending_recv;
+};
+
+#define SPDK_URING_SOCK_REQUEST_IOV(req) ((struct iovec *)((uint8_t *)req + sizeof(struct spdk_sock_request)))
+
+static int
+get_addr_str(struct sockaddr *sa, char *host, size_t hlen)
+{
+	const char *result = NULL;
+
+	if (sa == NULL || host == NULL) {
+		return -1;
+	}
+
+	switch (sa->sa_family) {
+	case AF_INET:
+		result = inet_ntop(AF_INET, &(((struct sockaddr_in *)sa)->sin_addr),
+				   host, hlen);
+		break;
+	case AF_INET6:
+		result = inet_ntop(AF_INET6, &(((struct sockaddr_in6 *)sa)->sin6_addr),
+				   host, hlen);
+		break;
+	default:
+		break;
+	}
+
+	if (result != NULL) {
+		return 0;
+	} else {
+		return -1;
+	}
+}
+
+#define __uring_sock(sock) (struct spdk_uring_sock *)sock
+#define __uring_group_impl(group) (struct spdk_uring_sock_group_impl *)group
+
+static int
+uring_sock_getaddr(struct spdk_sock *_sock, char *saddr, int slen, uint16_t *sport,
+		   char *caddr, int clen, uint16_t *cport)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct sockaddr_storage sa;
+	socklen_t salen;
+	int rc;
+
+	assert(sock != NULL);
+
+	memset(&sa, 0, sizeof sa);
+	salen = sizeof sa;
+	rc = getsockname(sock->fd, (struct sockaddr *) &sa, &salen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getsockname() failed (errno=%d)\n", errno);
+		return -1;
+	}
+
+	switch (sa.ss_family) {
+	case AF_UNIX:
+		/* Acceptable connection types that don't have IPs */
+		return 0;
+	case AF_INET:
+	case AF_INET6:
+		/* Code below will get IP addresses */
+		break;
+	default:
+		/* Unsupported socket family */
+		return -1;
+	}
+
+	rc = get_addr_str((struct sockaddr *)&sa, saddr, slen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getnameinfo() failed (errno=%d)\n", errno);
+		return -1;
+	}
+
+	if (sport) {
+		if (sa.ss_family == AF_INET) {
+			*sport = ntohs(((struct sockaddr_in *) &sa)->sin_port);
+		} else if (sa.ss_family == AF_INET6) {
+			*sport = ntohs(((struct sockaddr_in6 *) &sa)->sin6_port);
+		}
+	}
+
+	memset(&sa, 0, sizeof sa);
+	salen = sizeof sa;
+	rc = getpeername(sock->fd, (struct sockaddr *) &sa, &salen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getpeername() failed (errno=%d)\n", errno);
+		return -1;
+	}
+
+	rc = get_addr_str((struct sockaddr *)&sa, caddr, clen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getnameinfo() failed (errno=%d)\n", errno);
+		return -1;
+	}
+
+	if (cport) {
+		if (sa.ss_family == AF_INET) {
+			*cport = ntohs(((struct sockaddr_in *) &sa)->sin_port);
+		} else if (sa.ss_family == AF_INET6) {
+			*cport = ntohs(((struct sockaddr_in6 *) &sa)->sin6_port);
+		}
+	}
+
+	return 0;
+}
+
+enum uring_sock_create_type {
+	SPDK_SOCK_CREATE_LISTEN,
+	SPDK_SOCK_CREATE_CONNECT,
+};
+
+static int
+uring_sock_alloc_pipe(struct spdk_uring_sock *sock, int sz)
+{
+	uint8_t *new_buf;
+	struct spdk_pipe *new_pipe;
+	struct iovec siov[2];
+	struct iovec diov[2];
+	int sbytes;
+	ssize_t bytes;
+
+	if (sock->recv_buf_sz == sz) {
+		return 0;
+	}
+
+	/* If the new size is 0, just free the pipe */
+	if (sz == 0) {
+		spdk_pipe_destroy(sock->recv_pipe);
+		free(sock->recv_buf);
+		sock->recv_pipe = NULL;
+		sock->recv_buf = NULL;
+		return 0;
+	} else if (sz < MIN_SOCK_PIPE_SIZE) {
+		SPDK_ERRLOG("The size of the pipe must be larger than %d\n", MIN_SOCK_PIPE_SIZE);
+		return -1;
+	}
+
+	/* Round up to next 64 byte multiple */
+	new_buf = calloc(SPDK_ALIGN_CEIL(sz + 1, 64), sizeof(uint8_t));
+	if (!new_buf) {
+		SPDK_ERRLOG("socket recv buf allocation failed\n");
+		return -ENOMEM;
+	}
+
+	new_pipe = spdk_pipe_create(new_buf, sz + 1);
+	if (new_pipe == NULL) {
+		SPDK_ERRLOG("socket pipe allocation failed\n");
+		free(new_buf);
+		return -ENOMEM;
+	}
+
+	if (sock->recv_pipe != NULL) {
+		/* Pull all of the data out of the old pipe */
+		sbytes = spdk_pipe_reader_get_buffer(sock->recv_pipe, sock->recv_buf_sz, siov);
+		if (sbytes > sz) {
+			/* Too much data to fit into the new pipe size */
+			spdk_pipe_destroy(new_pipe);
+			free(new_buf);
+			return -EINVAL;
+		}
+
+		sbytes = spdk_pipe_writer_get_buffer(new_pipe, sz, diov);
+		assert(sbytes == sz);
+
+		bytes = spdk_iovcpy(siov, 2, diov, 2);
+		spdk_pipe_writer_advance(new_pipe, bytes);
+
+		spdk_pipe_destroy(sock->recv_pipe);
+		free(sock->recv_buf);
+	}
+
+	sock->recv_buf_sz = sz;
+	sock->recv_buf = new_buf;
+	sock->recv_pipe = new_pipe;
+
+	return 0;
+}
+
+static int
+uring_sock_set_recvbuf(struct spdk_sock *_sock, int sz)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	int rc;
+
+	assert(sock != NULL);
+
+#ifndef __aarch64__
+	/* On ARM systems, this buffering does not help. Skip it. */
+	/* The size of the pipe is purely derived from benchmarks. It seems to work well. */
+	rc = uring_sock_alloc_pipe(sock, sz);
+	if (rc) {
+		SPDK_ERRLOG("unable to allocate sufficient recvbuf with sz=%d on sock=%p\n", sz, _sock);
+		return rc;
+	}
+#endif
+
+	if (sz < SO_RCVBUF_SIZE) {
+		sz = SO_RCVBUF_SIZE;
+	}
+
+	rc = setsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &sz, sizeof(sz));
+	if (rc < 0) {
+		return rc;
+	}
+
+	return 0;
+}
+
+static int
+uring_sock_set_sendbuf(struct spdk_sock *_sock, int sz)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	int rc;
+
+	assert(sock != NULL);
+
+	if (sz < SO_SNDBUF_SIZE) {
+		sz = SO_SNDBUF_SIZE;
+	}
+
+	rc = setsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &sz, sizeof(sz));
+	if (rc < 0) {
+		return rc;
+	}
+
+	return 0;
+}
+
+static struct spdk_uring_sock *
+uring_sock_alloc(int fd)
+{
+	struct spdk_uring_sock *sock;
+
+	sock = calloc(1, sizeof(*sock));
+	if (sock == NULL) {
+		SPDK_ERRLOG("sock allocation failed\n");
+		return NULL;
+	}
+
+	sock->fd = fd;
+	return sock;
+}
+
+static struct spdk_sock *
+uring_sock_create(const char *ip, int port,
+		  enum uring_sock_create_type type,
+		  struct spdk_sock_opts *opts)
+{
+	struct spdk_uring_sock *sock;
+	char buf[MAX_TMPBUF];
+	char portnum[PORTNUMLEN];
+	char *p;
+	struct addrinfo hints, *res, *res0;
+	int fd, flag;
+	int val = 1;
+	int rc;
+
+	if (ip == NULL) {
+		return NULL;
+	}
+	if (ip[0] == '[') {
+		snprintf(buf, sizeof(buf), "%s", ip + 1);
+		p = strchr(buf, ']');
+		if (p != NULL) {
+			*p = '\0';
+		}
+		ip = (const char *) &buf[0];
+	}
+
+	snprintf(portnum, sizeof portnum, "%d", port);
+	memset(&hints, 0, sizeof hints);
+	hints.ai_family = PF_UNSPEC;
+	hints.ai_socktype = SOCK_STREAM;
+	hints.ai_flags = AI_NUMERICSERV;
+	hints.ai_flags |= AI_PASSIVE;
+	hints.ai_flags |= AI_NUMERICHOST;
+	rc = getaddrinfo(ip, portnum, &hints, &res0);
+	if (rc != 0) {
+		SPDK_ERRLOG("getaddrinfo() failed (errno=%d)\n", errno);
+		return NULL;
+	}
+
+	/* try listen */
+	fd = -1;
+	for (res = res0; res != NULL; res = res->ai_next) {
+retry:
+		fd = socket(res->ai_family, res->ai_socktype, res->ai_protocol);
+		if (fd < 0) {
+			/* error */
+			continue;
+		}
+
+		val = SO_RCVBUF_SIZE;
+		rc = setsockopt(fd, SOL_SOCKET, SO_RCVBUF, &val, sizeof val);
+		if (rc) {
+			/* Not fatal */
+		}
+
+		val = SO_SNDBUF_SIZE;
+		rc = setsockopt(fd, SOL_SOCKET, SO_SNDBUF, &val, sizeof val);
+		if (rc) {
+			/* Not fatal */
+		}
+
+		rc = setsockopt(fd, SOL_SOCKET, SO_REUSEADDR, &val, sizeof val);
+		if (rc != 0) {
+			close(fd);
+			/* error */
+			continue;
+		}
+		rc = setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &val, sizeof val);
+		if (rc != 0) {
+			close(fd);
+			/* error */
+			continue;
+		}
+
+#if defined(SO_PRIORITY)
+		if (opts != NULL && opts->priority) {
+			rc = setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &opts->priority, sizeof val);
+			if (rc != 0) {
+				close(fd);
+				/* error */
+				continue;
+			}
+		}
+#endif
+		if (res->ai_family == AF_INET6) {
+			rc = setsockopt(fd, IPPROTO_IPV6, IPV6_V6ONLY, &val, sizeof val);
+			if (rc != 0) {
+				close(fd);
+				/* error */
+				continue;
+			}
+		}
+
+		if (type == SPDK_SOCK_CREATE_LISTEN) {
+			rc = bind(fd, res->ai_addr, res->ai_addrlen);
+			if (rc != 0) {
+				SPDK_ERRLOG("bind() failed at port %d, errno = %d\n", port, errno);
+				switch (errno) {
+				case EINTR:
+					/* interrupted? */
+					close(fd);
+					goto retry;
+				case EADDRNOTAVAIL:
+					SPDK_ERRLOG("IP address %s not available. "
+						    "Verify IP address in config file "
+						    "and make sure setup script is "
+						    "run before starting spdk app.\n", ip);
+				/* FALLTHROUGH */
+				default:
+					/* try next family */
+					close(fd);
+					fd = -1;
+					continue;
+				}
+			}
+			/* bind OK */
+			rc = listen(fd, 512);
+			if (rc != 0) {
+				SPDK_ERRLOG("listen() failed, errno = %d\n", errno);
+				close(fd);
+				fd = -1;
+				break;
+			}
+		} else if (type == SPDK_SOCK_CREATE_CONNECT) {
+			rc = connect(fd, res->ai_addr, res->ai_addrlen);
+			if (rc != 0) {
+				SPDK_ERRLOG("connect() failed, errno = %d\n", errno);
+				/* try next family */
+				close(fd);
+				fd = -1;
+				continue;
+			}
+		}
+
+		flag = fcntl(fd, F_GETFL);
+		if (fcntl(fd, F_SETFL, flag | O_NONBLOCK) < 0) {
+			SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%d)\n", fd, errno);
+			close(fd);
+			fd = -1;
+			break;
+		}
+		break;
+	}
+	freeaddrinfo(res0);
+
+	if (fd < 0) {
+		return NULL;
+	}
+
+	sock = uring_sock_alloc(fd);
+	if (sock == NULL) {
+		SPDK_ERRLOG("sock allocation failed\n");
+		close(fd);
+		return NULL;
+	}
+
+	return &sock->base;
+}
+
+static struct spdk_sock *
+uring_sock_listen(const char *ip, int port, struct spdk_sock_opts *opts)
+{
+	return uring_sock_create(ip, port, SPDK_SOCK_CREATE_LISTEN, opts);
+}
+
+static struct spdk_sock *
+uring_sock_connect(const char *ip, int port, struct spdk_sock_opts *opts)
+{
+	return uring_sock_create(ip, port, SPDK_SOCK_CREATE_CONNECT, opts);
+}
+
+static struct spdk_sock *
+uring_sock_accept(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock		*sock = __uring_sock(_sock);
+	struct sockaddr_storage		sa;
+	socklen_t			salen;
+	int				rc, fd;
+	struct spdk_uring_sock		*new_sock;
+	int				flag;
+
+	memset(&sa, 0, sizeof(sa));
+	salen = sizeof(sa);
+
+	assert(sock != NULL);
+
+	rc = accept(sock->fd, (struct sockaddr *)&sa, &salen);
+
+	if (rc == -1) {
+		return NULL;
+	}
+
+	fd = rc;
+
+	flag = fcntl(fd, F_GETFL);
+	if ((!(flag & O_NONBLOCK)) && (fcntl(fd, F_SETFL, flag | O_NONBLOCK) < 0)) {
+		SPDK_ERRLOG("fcntl can't set nonblocking mode for socket, fd: %d (%d)\n", fd, errno);
+		close(fd);
+		return NULL;
+	}
+
+#if defined(SO_PRIORITY)
+	/* The priority is not inherited, so call this function again */
+	if (sock->base.opts.priority) {
+		rc = setsockopt(fd, SOL_SOCKET, SO_PRIORITY, &sock->base.opts.priority, sizeof(int));
+		if (rc != 0) {
+			close(fd);
+			return NULL;
+		}
+	}
+#endif
+
+	new_sock = uring_sock_alloc(fd);
+	if (new_sock == NULL) {
+		close(fd);
+		return NULL;
+	}
+
+	return &new_sock->base;
+}
+
+static int
+uring_sock_close(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	int rc;
+
+	assert(TAILQ_EMPTY(&_sock->pending_reqs));
+	assert(sock->group == NULL);
+
+	spdk_pipe_destroy(sock->recv_pipe);
+	free(sock->recv_buf);
+	rc = close(sock->fd);
+	if (rc == 0) {
+		free(sock);
+	}
+
+	return rc;
+}
+
+static ssize_t
+uring_sock_recv_from_pipe(struct spdk_uring_sock *sock, struct iovec *diov, int diovcnt)
+{
+	struct iovec siov[2];
+	int sbytes;
+	ssize_t bytes;
+	struct spdk_uring_sock_group_impl *group;
+
+	sbytes = spdk_pipe_reader_get_buffer(sock->recv_pipe, sock->recv_buf_sz, siov);
+	if (sbytes < 0) {
+		errno = EINVAL;
+		return -1;
+	} else if (sbytes == 0) {
+		errno = EAGAIN;
+		return -1;
+	}
+
+	bytes = spdk_iovcpy(siov, 2, diov, diovcnt);
+
+	if (bytes == 0) {
+		/* The only way this happens is if diov is 0 length */
+		errno = EINVAL;
+		return -1;
+	}
+
+	spdk_pipe_reader_advance(sock->recv_pipe, bytes);
+
+	/* If we drained the pipe, take it off the level-triggered list */
+	if (sock->base.group_impl && spdk_pipe_reader_bytes_available(sock->recv_pipe) == 0) {
+		group = __uring_group_impl(sock->base.group_impl);
+		TAILQ_REMOVE(&group->pending_recv, sock, link);
+		sock->pending_recv = false;
+	}
+
+	return bytes;
+}
+
+static inline ssize_t
+uring_sock_read(struct spdk_uring_sock *sock)
+{
+	struct iovec iov[2];
+	int bytes;
+	struct spdk_uring_sock_group_impl *group;
+
+	bytes = spdk_pipe_writer_get_buffer(sock->recv_pipe, sock->recv_buf_sz, iov);
+
+	if (bytes > 0) {
+		bytes = readv(sock->fd, iov, 2);
+		if (bytes > 0) {
+			spdk_pipe_writer_advance(sock->recv_pipe, bytes);
+			if (sock->base.group_impl) {
+				group = __uring_group_impl(sock->base.group_impl);
+				TAILQ_INSERT_TAIL(&group->pending_recv, sock, link);
+				sock->pending_recv = true;
+			}
+		}
+	}
+
+	return bytes;
+}
+
+static ssize_t
+uring_sock_readv(struct spdk_sock *_sock, struct iovec *iov, int iovcnt)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	int rc, i;
+	size_t len;
+
+	if (sock->recv_pipe == NULL) {
+		return readv(sock->fd, iov, iovcnt);
+	}
+
+	len = 0;
+	for (i = 0; i < iovcnt; i++) {
+		len += iov[i].iov_len;
+	}
+
+	if (spdk_pipe_reader_bytes_available(sock->recv_pipe) == 0) {
+		/* If the user is receiving a sufficiently large amount of data,
+		 * receive directly to their buffers. */
+		if (len >= MIN_SOCK_PIPE_SIZE) {
+			return readv(sock->fd, iov, iovcnt);
+		}
+
+		/* Otherwise, do a big read into our pipe */
+		rc = uring_sock_read(sock);
+		if (rc <= 0) {
+			return rc;
+		}
+	}
+
+	return uring_sock_recv_from_pipe(sock, iov, iovcnt);
+}
+
+static ssize_t
+uring_sock_recv(struct spdk_sock *sock, void *buf, size_t len)
+{
+	struct iovec iov[1];
+
+	iov[0].iov_base = buf;
+	iov[0].iov_len = len;
+
+	return uring_sock_readv(sock, iov, 1);
+}
+
+static ssize_t
+uring_sock_writev(struct spdk_sock *_sock, struct iovec *iov, int iovcnt)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+
+	if (sock->write_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE) {
+		errno = EAGAIN;
+		return -1;
+	}
+
+	return writev(sock->fd, iov, iovcnt);
+}
+
+static int
+sock_prep_reqs(struct spdk_sock *_sock, struct iovec *iovs, int index,
+	       struct spdk_sock_request **last_req)
+{
+	int iovcnt, i;
+	struct spdk_sock_request *req;
+	unsigned int offset;
+
+	/* Gather an iov */
+	iovcnt = index;
+	if (spdk_unlikely(iovcnt >= IOV_BATCH_SIZE)) {
+		goto end;
+	}
+
+	if (last_req != NULL && *last_req != NULL) {
+		req = TAILQ_NEXT(*last_req, internal.link);
+	} else {
+		req = TAILQ_FIRST(&_sock->queued_reqs);
+	}
+
+	while (req) {
+		offset = req->internal.offset;
+
+		for (i = 0; i < req->iovcnt; i++) {
+			/* Consume any offset first */
+			if (offset >= SPDK_SOCK_REQUEST_IOV(req, i)->iov_len) {
+				offset -= SPDK_SOCK_REQUEST_IOV(req, i)->iov_len;
+				continue;
+			}
+
+			iovs[iovcnt].iov_base = SPDK_SOCK_REQUEST_IOV(req, i)->iov_base + offset;
+			iovs[iovcnt].iov_len = SPDK_SOCK_REQUEST_IOV(req, i)->iov_len - offset;
+			iovcnt++;
+
+			offset = 0;
+
+			if (iovcnt >= IOV_BATCH_SIZE) {
+				break;
+			}
+		}
+		if (iovcnt >= IOV_BATCH_SIZE) {
+			break;
+		}
+
+		if (last_req != NULL) {
+			*last_req = req;
+		}
+		req = TAILQ_NEXT(req, internal.link);
+	}
+
+end:
+	return iovcnt;
+}
+
+static int
+sock_complete_reqs(struct spdk_sock *_sock, ssize_t rc)
+{
+	struct spdk_sock_request *req;
+	int i, retval;
+	unsigned int offset;
+	size_t len;
+
+	/* Consume the requests that were actually written */
+	req = TAILQ_FIRST(&_sock->queued_reqs);
+	while (req) {
+		offset = req->internal.offset;
+
+		for (i = 0; i < req->iovcnt; i++) {
+			/* Advance by the offset first */
+			if (offset >= SPDK_SOCK_REQUEST_IOV(req, i)->iov_len) {
+				offset -= SPDK_SOCK_REQUEST_IOV(req, i)->iov_len;
+				continue;
+			}
+
+			/* Calculate the remaining length of this element */
+			len = SPDK_SOCK_REQUEST_IOV(req, i)->iov_len - offset;
+
+			if (len > (size_t)rc) {
+				/* This element was partially sent. */
+				req->internal.offset += rc;
+				return 0;
+			}
+
+			offset = 0;
+			req->internal.offset += len;
+			rc -= len;
+		}
+
+		/* Handled a full request. */
+		spdk_sock_request_pend(_sock, req);
+
+		retval = spdk_sock_request_put(_sock, req, 0);
+		if (retval) {
+			return retval;
+		}
+
+		if (rc == 0) {
+			break;
+		}
+
+		req = TAILQ_FIRST(&_sock->queued_reqs);
+	}
+
+	return 0;
+}
+
+static void
+_sock_flush(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct spdk_uring_task *task = &sock->write_task;
+	uint32_t iovcnt;
+	struct io_uring_sqe *sqe;
+
+	if (task->status == SPDK_URING_SOCK_TASK_IN_PROCESS) {
+		return;
+	}
+
+	iovcnt = sock_prep_reqs(&sock->base, task->iovs, task->iov_cnt, &task->last_req);
+	if (!iovcnt) {
+		return;
+	}
+
+	task->iov_cnt = iovcnt;
+	assert(sock->group != NULL);
+	task->msg.msg_iov = task->iovs;
+	task->msg.msg_iovlen = task->iov_cnt;
+
+	sock->group->io_queued++;
+
+	sqe = io_uring_get_sqe(&sock->group->uring);
+	io_uring_prep_sendmsg(sqe, sock->fd, &sock->write_task.msg, 0);
+	io_uring_sqe_set_data(sqe, task);
+	task->status = SPDK_URING_SOCK_TASK_IN_PROCESS;
+}
+
+static void
+_sock_prep_pollin(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct spdk_uring_task *task = &sock->pollin_task;
+	struct io_uring_sqe *sqe;
+
+	/* Do not prepare pollin event */
+	if (task->status == SPDK_URING_SOCK_TASK_IN_PROCESS || sock->pending_recv) {
+		return;
+	}
+
+	assert(sock->group != NULL);
+	sock->group->io_queued++;
+
+	sqe = io_uring_get_sqe(&sock->group->uring);
+	io_uring_prep_poll_add(sqe, sock->fd, POLLIN);
+	io_uring_sqe_set_data(sqe, task);
+	task->status = SPDK_URING_SOCK_TASK_IN_PROCESS;
+}
+
+static void
+_sock_prep_cancel_task(struct spdk_sock *_sock, void *user_data)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct spdk_uring_task *task = &sock->cancel_task;
+	struct io_uring_sqe *sqe;
+
+	if (task->status == SPDK_URING_SOCK_TASK_IN_PROCESS) {
+		return;
+	}
+
+	assert(sock->group != NULL);
+	sock->group->io_queued++;
+
+	sqe = io_uring_get_sqe(&sock->group->uring);
+	io_uring_prep_cancel(sqe, user_data, 0);
+	io_uring_sqe_set_data(sqe, task);
+	task->status = SPDK_URING_SOCK_TASK_IN_PROCESS;
+}
+
+static int
+sock_uring_group_reap(struct spdk_uring_sock_group_impl *group, int max, int max_read_events,
+		      struct spdk_sock **socks)
+{
+	int i, count, ret;
+	struct io_uring_cqe *cqe;
+	struct spdk_uring_sock *sock, *tmp;
+	struct spdk_uring_task *task;
+	int status;
+
+	for (i = 0; i < max; i++) {
+		ret = io_uring_peek_cqe(&group->uring, &cqe);
+		if (ret != 0) {
+			break;
+		}
+
+		if (cqe == NULL) {
+			break;
+		}
+
+		task = (struct spdk_uring_task *)cqe->user_data;
+		assert(task != NULL);
+		sock = task->sock;
+		assert(sock != NULL);
+		assert(sock->group != NULL);
+		assert(sock->group == group);
+		sock->group->io_inflight--;
+		sock->group->io_avail++;
+		status = cqe->res;
+		io_uring_cqe_seen(&group->uring, cqe);
+
+		task->status = SPDK_URING_SOCK_TASK_NOT_IN_USE;
+
+		if (spdk_unlikely(status <= 0)) {
+			if (status == -EAGAIN || status == -EWOULDBLOCK) {
+				continue;
+			}
+		}
+
+		switch (task->type) {
+		case SPDK_SOCK_TASK_POLLIN:
+			if ((status & POLLIN) == POLLIN) {
+				if (sock->base.cb_fn != NULL) {
+					assert(sock->pending_recv == false);
+					sock->pending_recv = true;
+					TAILQ_INSERT_TAIL(&group->pending_recv, sock, link);
+				}
+			}
+			break;
+		case SPDK_SOCK_TASK_WRITE:
+			assert(TAILQ_EMPTY(&sock->base.pending_reqs));
+			task->last_req = NULL;
+			task->iov_cnt = 0;
+			if (spdk_unlikely(status) < 0) {
+				sock->connection_status = status;
+				spdk_sock_abort_requests(&sock->base);
+			} else {
+				sock_complete_reqs(&sock->base, status);
+			}
+
+			break;
+		case SPDK_SOCK_TASK_CANCEL:
+			/* Do nothing */
+			break;
+		default:
+			SPDK_UNREACHABLE();
+		}
+	}
+
+	if (!socks) {
+		return 0;
+	}
+	count = 0;
+	TAILQ_FOREACH_SAFE(sock, &group->pending_recv, link, tmp) {
+		if (count == max_read_events) {
+			break;
+		}
+
+		socks[count++] = &sock->base;
+	}
+
+	/* Cycle the pending_recv list so that each time we poll things aren't
+	 * in the same order. */
+	for (i = 0; i < count; i++) {
+		sock = __uring_sock(socks[i]);
+
+		TAILQ_REMOVE(&group->pending_recv, sock, link);
+
+		if (sock->recv_pipe == NULL || spdk_pipe_reader_bytes_available(sock->recv_pipe) == 0) {
+			sock->pending_recv = false;
+		} else {
+			TAILQ_INSERT_TAIL(&group->pending_recv, sock, link);
+		}
+	}
+
+	return count;
+}
+
+static int
+_sock_flush_client(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct msghdr msg = {};
+	struct iovec iovs[IOV_BATCH_SIZE];
+	int iovcnt;
+	ssize_t rc;
+
+	/* Can't flush from within a callback or we end up with recursive calls */
+	if (_sock->cb_cnt > 0) {
+		return 0;
+	}
+
+	/* Gather an iov */
+	iovcnt = sock_prep_reqs(_sock, iovs, 0, NULL);
+	if (iovcnt == 0) {
+		return 0;
+	}
+
+	/* Perform the vectored write */
+	msg.msg_iov = iovs;
+	msg.msg_iovlen = iovcnt;
+	rc = sendmsg(sock->fd, &msg, 0);
+	if (rc <= 0) {
+		if (errno == EAGAIN || errno == EWOULDBLOCK) {
+			return 0;
+		}
+		return rc;
+	}
+
+	sock_complete_reqs(_sock, rc);
+
+	return 0;
+}
+
+static void
+uring_sock_writev_async(struct spdk_sock *_sock, struct spdk_sock_request *req)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	int rc;
+
+	if (spdk_unlikely(sock->connection_status)) {
+		req->cb_fn(req->cb_arg, sock->connection_status);
+		return;
+	}
+
+	spdk_sock_request_queue(_sock, req);
+
+	if (!sock->group) {
+		if (_sock->queued_iovcnt >= IOV_BATCH_SIZE) {
+			rc = _sock_flush_client(_sock);
+			if (rc) {
+				spdk_sock_abort_requests(_sock);
+			}
+		}
+	}
+}
+
+static int
+uring_sock_set_recvlowat(struct spdk_sock *_sock, int nbytes)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	int val;
+	int rc;
+
+	assert(sock != NULL);
+
+	val = nbytes;
+	rc = setsockopt(sock->fd, SOL_SOCKET, SO_RCVLOWAT, &val, sizeof val);
+	if (rc != 0) {
+		return -1;
+	}
+	return 0;
+}
+
+static bool
+uring_sock_is_ipv6(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct sockaddr_storage sa;
+	socklen_t salen;
+	int rc;
+
+	assert(sock != NULL);
+
+	memset(&sa, 0, sizeof sa);
+	salen = sizeof sa;
+	rc = getsockname(sock->fd, (struct sockaddr *) &sa, &salen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getsockname() failed (errno=%d)\n", errno);
+		return false;
+	}
+
+	return (sa.ss_family == AF_INET6);
+}
+
+static bool
+uring_sock_is_ipv4(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct sockaddr_storage sa;
+	socklen_t salen;
+	int rc;
+
+	assert(sock != NULL);
+
+	memset(&sa, 0, sizeof sa);
+	salen = sizeof sa;
+	rc = getsockname(sock->fd, (struct sockaddr *) &sa, &salen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getsockname() failed (errno=%d)\n", errno);
+		return false;
+	}
+
+	return (sa.ss_family == AF_INET);
+}
+
+static bool
+uring_sock_is_connected(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	uint8_t byte;
+	int rc;
+
+	rc = recv(sock->fd, &byte, 1, MSG_PEEK);
+	if (rc == 0) {
+		return false;
+	}
+
+	if (rc < 0) {
+		if (errno == EAGAIN || errno == EWOULDBLOCK) {
+			return true;
+		}
+
+		return false;
+	}
+
+	return true;
+}
+
+static int
+uring_sock_get_placement_id(struct spdk_sock *_sock, int *placement_id)
+{
+	int rc = -1;
+
+#if defined(SO_INCOMING_NAPI_ID)
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	socklen_t salen = sizeof(int);
+
+	rc = getsockopt(sock->fd, SOL_SOCKET, SO_INCOMING_NAPI_ID, placement_id, &salen);
+	if (rc != 0) {
+		SPDK_ERRLOG("getsockopt() failed (errno=%d)\n", errno);
+	}
+
+#endif
+	return rc;
+}
+
+static struct spdk_sock_group_impl *
+uring_sock_group_impl_create(void)
+{
+	struct spdk_uring_sock_group_impl *group_impl;
+
+	group_impl = calloc(1, sizeof(*group_impl));
+	if (group_impl == NULL) {
+		SPDK_ERRLOG("group_impl allocation failed\n");
+		return NULL;
+	}
+
+	group_impl->io_avail = SPDK_SOCK_GROUP_QUEUE_DEPTH;
+
+	if (io_uring_queue_init(SPDK_SOCK_GROUP_QUEUE_DEPTH, &group_impl->uring, 0) < 0) {
+		SPDK_ERRLOG("uring I/O context setup failure\n");
+		free(group_impl);
+		return NULL;
+	}
+
+	TAILQ_INIT(&group_impl->pending_recv);
+
+	return &group_impl->base;
+}
+
+static int
+uring_sock_group_impl_add_sock(struct spdk_sock_group_impl *_group,
+			       struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct spdk_uring_sock_group_impl *group = __uring_group_impl(_group);
+
+	sock->group = group;
+	sock->write_task.sock = sock;
+	sock->write_task.type = SPDK_SOCK_TASK_WRITE;
+
+	sock->pollin_task.sock = sock;
+	sock->pollin_task.type = SPDK_SOCK_TASK_POLLIN;
+
+	sock->cancel_task.sock = sock;
+	sock->cancel_task.type = SPDK_SOCK_TASK_CANCEL;
+
+	/* switched from another polling group due to scheduling */
+	if (spdk_unlikely(sock->recv_pipe != NULL &&
+			  (spdk_pipe_reader_bytes_available(sock->recv_pipe) > 0))) {
+		assert(sock->pending_recv == false);
+		sock->pending_recv = true;
+		TAILQ_INSERT_TAIL(&group->pending_recv, sock, link);
+	}
+
+	return 0;
+}
+
+static int
+uring_sock_group_impl_poll(struct spdk_sock_group_impl *_group, int max_events,
+			   struct spdk_sock **socks)
+{
+	struct spdk_uring_sock_group_impl *group = __uring_group_impl(_group);
+	int count, ret;
+	int to_complete, to_submit;
+	struct spdk_sock *_sock, *tmp;
+	struct spdk_uring_sock *sock;
+
+	if (spdk_likely(socks)) {
+		TAILQ_FOREACH_SAFE(_sock, &group->base.socks, link, tmp) {
+			sock = __uring_sock(_sock);
+			if (spdk_unlikely(sock->connection_status)) {
+				continue;
+			}
+			_sock_flush(_sock);
+			_sock_prep_pollin(_sock);
+		}
+	}
+
+	to_submit = group->io_queued;
+
+	/* For network I/O, it cannot be set with O_DIRECT, so we do not need to call spdk_io_uring_enter */
+	if (to_submit > 0) {
+		/* If there are I/O to submit, use io_uring_submit here.
+		 * It will automatically call io_uring_enter appropriately. */
+		ret = io_uring_submit(&group->uring);
+		if (ret < 0) {
+			return 1;
+		}
+		group->io_queued = 0;
+		group->io_inflight += to_submit;
+		group->io_avail -= to_submit;
+	}
+
+	count = 0;
+	to_complete = group->io_inflight;
+	if (to_complete > 0) {
+		count = sock_uring_group_reap(group, to_complete, max_events, socks);
+	}
+
+	return count;
+}
+
+static int
+uring_sock_group_impl_remove_sock(struct spdk_sock_group_impl *_group,
+				  struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+	struct spdk_uring_sock_group_impl *group = __uring_group_impl(_group);
+
+	if (sock->write_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE) {
+		_sock_prep_cancel_task(_sock, &sock->write_task);
+		/* Since spdk_sock_group_remove_sock is not asynchronous interface, so
+		 * currently can use a while loop here. */
+		while ((sock->write_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE) ||
+		       (sock->cancel_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE)) {
+			uring_sock_group_impl_poll(_group, 32, NULL);
+		}
+	}
+
+	if (sock->pollin_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE) {
+		_sock_prep_cancel_task(_sock, &sock->pollin_task);
+		/* Since spdk_sock_group_remove_sock is not asynchronous interface, so
+		 * currently can use a while loop here. */
+		while ((sock->pollin_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE) ||
+		       (sock->cancel_task.status != SPDK_URING_SOCK_TASK_NOT_IN_USE)) {
+			uring_sock_group_impl_poll(_group, 32, NULL);
+		}
+	}
+
+	if (sock->pending_recv) {
+		TAILQ_REMOVE(&group->pending_recv, sock, link);
+		sock->pending_recv = false;
+	}
+	assert(sock->pending_recv == false);
+
+	sock->group = NULL;
+	return 0;
+}
+
+static int
+uring_sock_group_impl_close(struct spdk_sock_group_impl *_group)
+{
+	struct spdk_uring_sock_group_impl *group = __uring_group_impl(_group);
+
+	/* try to reap all the active I/O */
+	while (group->io_inflight) {
+		uring_sock_group_impl_poll(_group, 32, NULL);
+	}
+	assert(group->io_inflight == 0);
+	assert(group->io_avail == SPDK_SOCK_GROUP_QUEUE_DEPTH);
+
+	io_uring_queue_exit(&group->uring);
+
+	free(group);
+	return 0;
+}
+
+static int
+uring_sock_flush(struct spdk_sock *_sock)
+{
+	struct spdk_uring_sock *sock = __uring_sock(_sock);
+
+	if (!sock->group) {
+		return _sock_flush_client(_sock);
+	}
+
+	return 0;
+}
+
+static struct spdk_net_impl g_uring_net_impl = {
+	.name		= "uring",
+	.getaddr	= uring_sock_getaddr,
+	.connect	= uring_sock_connect,
+	.listen		= uring_sock_listen,
+	.accept		= uring_sock_accept,
+	.close		= uring_sock_close,
+	.recv		= uring_sock_recv,
+	.readv		= uring_sock_readv,
+	.writev		= uring_sock_writev,
+	.writev_async	= uring_sock_writev_async,
+	.flush          = uring_sock_flush,
+	.set_recvlowat	= uring_sock_set_recvlowat,
+	.set_recvbuf	= uring_sock_set_recvbuf,
+	.set_sendbuf	= uring_sock_set_sendbuf,
+	.is_ipv6	= uring_sock_is_ipv6,
+	.is_ipv4	= uring_sock_is_ipv4,
+	.is_connected   = uring_sock_is_connected,
+	.get_placement_id	= uring_sock_get_placement_id,
+	.group_impl_create	= uring_sock_group_impl_create,
+	.group_impl_add_sock	= uring_sock_group_impl_add_sock,
+	.group_impl_remove_sock = uring_sock_group_impl_remove_sock,
+	.group_impl_poll	= uring_sock_group_impl_poll,
+	.group_impl_close	= uring_sock_group_impl_close,
+};
+
+SPDK_NET_IMPL_REGISTER(uring, &g_uring_net_impl, DEFAULT_SOCK_PRIORITY + 1);