sock: asynchronous readv interface

This patch defines a new function, spdk_sock_readv_async(), which allows
the user to send a readv request and receive a callback once the
supplied buffer is filled with data from the socket.  It works simiarly
to asynchronous writes, but there can only be a single outstanding read
request at a time.

For now, the interface isn't implemented and any calls will return
-ENOTSUP.  Subsequent patches will add support for it in the uring
module and as well as emulation in the posix module.

Signed-off-by: Konrad Sztyber <konrad.sztyber@intel.com>
Change-Id: I924e2cdade49ffa18be6390109dc7e65c2728087
Reviewed-on: https://review.spdk.io/gerrit/c/spdk/spdk/+/12170
Tested-by: SPDK CI Jenkins <sys_sgci@intel.com>
Community-CI: Broadcom CI <spdk-ci.pdl@broadcom.com>
Community-CI: Mellanox Build Bot
Reviewed-by: Shuhei Matsumoto <smatsumoto@nvidia.com>
Reviewed-by: Ben Walker <benjamin.walker@intel.com>

include/spdk/sock.h

@@ -34,7 +34,11 @@ struct spdk_sock_group;
  */
 struct spdk_sock_request {
 	/* When the request is completed, this callback will be called.
-	 * err will be 0 on success or a negated errno value on failure. */
+	 * On success, err will be:
+	 *   - for writes: 0,
+	 *   - for reads: number of bytes read.
+	 * On failure: negative errno value.
+	 */
 	void	(*cb_fn)(void *cb_arg, int err);
 	void				*cb_arg;
 
@@ -332,6 +336,16 @@ void spdk_sock_writev_async(struct spdk_sock *sock, struct spdk_sock_request *re
  */
 ssize_t spdk_sock_readv(struct spdk_sock *sock, struct iovec *iov, int iovcnt);
 
+/**
+ * Read message from the given socket asynchronously, calling the provided callback when the whole
+ * buffer is filled or an error is encountered.  Only a single read request can be active at a time
+ * (including synchronous reads).
+ *
+ * \param sock Socket to receive message.
+ * \param req The read request to submit.
+ */
+void spdk_sock_readv_async(struct spdk_sock *sock, struct spdk_sock_request *req);
+
 /**
  * Set the value used to specify the low water mark (in bytes) for this socket.
  *

include/spdk_internal/sock.h

@@ -34,6 +34,7 @@ struct spdk_sock {
 
 	TAILQ_HEAD(, spdk_sock_request)	queued_reqs;
 	TAILQ_HEAD(, spdk_sock_request)	pending_reqs;
+	struct spdk_sock_request	*read_req;
 	int				queued_iovcnt;
 	int				cb_cnt;
 	spdk_sock_cb			cb_fn;
@@ -77,6 +78,7 @@ struct spdk_net_impl {
 	ssize_t (*writev)(struct spdk_sock *sock, struct iovec *iov, int iovcnt);
 
 	void (*writev_async)(struct spdk_sock *sock, struct spdk_sock_request *req);
+	void (*readv_async)(struct spdk_sock *sock, struct spdk_sock_request *req);
 	int (*flush)(struct spdk_sock *sock);
 
 	int (*set_recvlowat)(struct spdk_sock *sock, int nbytes);
@@ -202,6 +204,12 @@ spdk_sock_abort_requests(struct spdk_sock *sock)
 
 		req = TAILQ_FIRST(&sock->queued_reqs);
 	}
+
+	req = sock->read_req;
+	if (req != NULL) {
+		sock->read_req = NULL;
+		req->cb_fn(req->cb_arg, -ECANCELED);
+	}
 	assert(sock->cb_cnt > 0);
 	sock->cb_cnt--;
 

lib/sock/sock.c

@@ -476,6 +476,25 @@ spdk_sock_readv(struct spdk_sock *sock, struct iovec *iov, int iovcnt)
 	return sock->net_impl->readv(sock, iov, iovcnt);
 }
 
+void
+spdk_sock_readv_async(struct spdk_sock *sock, struct spdk_sock_request *req)
+{
+	assert(req->cb_fn != NULL);
+
+	if (spdk_unlikely(sock == NULL || sock->flags.closed)) {
+		req->cb_fn(req->cb_arg, -EBADF);
+		return;
+	}
+
+	/* The socket needs to be part of a poll group */
+	if (spdk_unlikely(sock->group_impl == NULL)) {
+		req->cb_fn(req->cb_arg, -EPERM);
+		return;
+	}
+
+	sock->net_impl->readv_async(sock, req);
+}
+
 ssize_t
 spdk_sock_writev(struct spdk_sock *sock, struct iovec *iov, int iovcnt)
 {

lib/sock/spdk_sock.map

@@ -15,6 +15,7 @@
 	spdk_sock_writev;
 	spdk_sock_writev_async;
 	spdk_sock_readv;
+	spdk_sock_readv_async;
 	spdk_sock_set_recvlowat;
 	spdk_sock_set_recvbuf;
 	spdk_sock_set_sendbuf;

module/sock/posix/posix.c

@@ -1368,6 +1368,12 @@ posix_sock_recv(struct spdk_sock *sock, void *buf, size_t len)
 	return posix_sock_readv(sock, iov, 1);
 }
 
+static void
+posix_sock_readv_async(struct spdk_sock *sock, struct spdk_sock_request *req)
+{
+	req->cb_fn(req->cb_arg, -ENOTSUP);
+}
+
 static ssize_t
 posix_sock_writev(struct spdk_sock *_sock, struct iovec *iov, int iovcnt)
 {
@@ -1957,6 +1963,7 @@ static struct spdk_net_impl g_posix_net_impl = {
 	.close		= posix_sock_close,
 	.recv		= posix_sock_recv,
 	.readv		= posix_sock_readv,
+	.readv_async	= posix_sock_readv_async,
 	.writev		= posix_sock_writev,
 	.writev_async	= posix_sock_writev_async,
 	.flush		= posix_sock_flush,

module/sock/uring/uring.c

@@ -1215,6 +1215,12 @@ uring_sock_writev_async(struct spdk_sock *_sock, struct spdk_sock_request *req)
 	}
 }
 
+static void
+uring_sock_readv_async(struct spdk_sock *sock, struct spdk_sock_request *req)
+{
+	req->cb_fn(req->cb_arg, -ENOTSUP);
+}
+
 static int
 uring_sock_set_recvlowat(struct spdk_sock *_sock, int nbytes)
 {
@@ -1589,6 +1595,7 @@ static struct spdk_net_impl g_uring_net_impl = {
 	.close		= uring_sock_close,
 	.recv		= uring_sock_recv,
 	.readv		= uring_sock_readv,
+	.readv_async	= uring_sock_readv_async,
 	.writev		= uring_sock_writev,
 	.writev_async	= uring_sock_writev_async,
 	.flush          = uring_sock_flush,