Re: [RFC PATCH v3 02/13] af_vsock: prepare 'vsock_connectible_recvmsg()'

[Stefano Garzarella <sgarzare@redhat.com>]

On Mon, Jan 25, 2021 at 02:11:57PM +0300, Arseny Krasnov wrote:
>This prepares 'vsock_connectible_recvmg()' to call SEQPACKET receive
>loop:
>1) Some shared check left in this function, then socket type
>   specific receive loop is called.
>2) Stream receive loop is moved to separate function.
>
>Signed-off-by: Arseny Krasnov <arseny.krasnov@kaspersky.com>
>---
> net/vmw_vsock/af_vsock.c | 242 ++++++++++++++++++++++-----------------
> 1 file changed, 138 insertions(+), 104 deletions(-)
>
>diff --git a/net/vmw_vsock/af_vsock.c b/net/vmw_vsock/af_vsock.c
>index c9ce57db9554..524df8fc84cd 100644
>--- a/net/vmw_vsock/af_vsock.c
>+++ b/net/vmw_vsock/af_vsock.c
>@@ -1858,65 +1858,69 @@ static int vsock_stream_sendmsg(struct socket *sock, struct msghdr *msg,
> 	return vsock_connectible_sendmsg(sock, msg, len);
> }
>
>-
>-static int
>-vsock_connectible_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
>-			  int flags)
>+static int vsock_wait_data(struct sock *sk, struct wait_queue_entry *wait,
>+			   long timeout,
>+			   struct vsock_transport_recv_notify_data *recv_data,
>+			   size_t target)
> {
>-	struct sock *sk;
>+	int err = 0;
> 	struct vsock_sock *vsk;
> 	const struct vsock_transport *transport;
>-	int err;
>-	size_t target;
>-	ssize_t copied;
>-	long timeout;
>-	struct vsock_transport_recv_notify_data recv_data;
>-
>-	DEFINE_WAIT(wait);
>
>-	sk = sock->sk;
> 	vsk = vsock_sk(sk);
> 	transport = vsk->transport;
>-	err = 0;
>-
>-	lock_sock(sk);
>-
>-	if (!transport || sk->sk_state != TCP_ESTABLISHED) {
>-		/* Recvmsg is supposed to return 0 if a peer performs an
>-		 * orderly shutdown. Differentiate between that case and when a
>-		 * peer has not connected or a local shutdown occured with the
>-		 * SOCK_DONE flag.
>-		 */
>-		if (sock_flag(sk, SOCK_DONE))
>-			err = 0;
>-		else
>-			err = -ENOTCONN;
>
>+	if (sk->sk_err != 0 ||
>+	    (sk->sk_shutdown & RCV_SHUTDOWN) ||
>+	    (vsk->peer_shutdown & SEND_SHUTDOWN)) {
>+		err = -1;
> 		goto out;
> 	}
>-
>-	if (flags & MSG_OOB) {
>-		err = -EOPNOTSUPP;
>+	/* Don't wait for non-blocking sockets. */
>+	if (timeout == 0) {
>+		err = -EAGAIN;
> 		goto out;
> 	}
>
>-	/* We don't check peer_shutdown flag here since peer may actually shut
>-	 * down, but there can be data in the queue that a local socket can
>-	 * receive.
>-	 */
>-	if (sk->sk_shutdown & RCV_SHUTDOWN) {
>-		err = 0;
>-		goto out;
>+	if (recv_data) {
>+		err = transport->notify_recv_pre_block(vsk, target, recv_data);
>+		if (err < 0)
>+			goto out;
> 	}
>
>-	/* It is valid on Linux to pass in a zero-length receive buffer.  This
>-	 * is not an error.  We may as well bail out now.
>-	 */
>-	if (!len) {
>-		err = 0;
>+	release_sock(sk);
>+	timeout = schedule_timeout(timeout);
>+	lock_sock(sk);
>+
>+	if (signal_pending(current)) {
>+		err = sock_intr_errno(timeout);
>+		goto out;
>+	} else if (timeout == 0) {
>+		err = -EAGAIN;
> 		goto out;
> 	}
>
>+out:
>+	finish_wait(sk_sleep(sk), wait);
>+	return err;
>+}
>+
>+static int __vsock_stream_recvmsg(struct sock *sk, struct msghdr *msg,
>+				  size_t len, int flags)
>+{
>+	struct vsock_transport_recv_notify_data recv_data;
>+	const struct vsock_transport *transport;
>+	struct vsock_sock *vsk;
>+	ssize_t copied;
>+	size_t target;
>+	long timeout;
>+	int err;
>+
>+	DEFINE_WAIT(wait);
>+
>+	vsk = vsock_sk(sk);
>+	transport = vsk->transport;
>+
> 	/* We must not copy less than target bytes into the user's buffer
> 	 * before returning successfully, so we wait for the consume queue to
> 	 * have that much data to consume before dequeueing.  Note that this
>@@ -1937,85 +1941,53 @@ vsock_connectible_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
>
>
> 	while (1) {
>+		ssize_t read;
> 		s64 ready;
>
> 		prepare_to_wait(sk_sleep(sk), &wait, 
> 		TASK_INTERRUPTIBLE);
> 		ready = vsock_stream_has_data(vsk);

Maybe we can move also these lines in vsock_wait_data() that can return 
'ready' or an error.

>
> 		if (ready == 0) {
>-			if (sk->sk_err != 0 ||
>-			    (sk->sk_shutdown & RCV_SHUTDOWN) ||
>-			    (vsk->peer_shutdown & SEND_SHUTDOWN)) {
>-				finish_wait(sk_sleep(sk), &wait);
>-				break;
>-			}
>-			/* Don't wait for non-blocking sockets. */
>-			if (timeout == 0) {
>-				err = -EAGAIN;
>-				finish_wait(sk_sleep(sk), &wait);
>-				break;
>-			}
>-
>-			err = transport->notify_recv_pre_block(
>-					vsk, target, &recv_data);
>-			if (err < 0) {
>-				finish_wait(sk_sleep(sk), &wait);
>-				break;
>-			}
>-			release_sock(sk);
>-			timeout = schedule_timeout(timeout);
>-			lock_sock(sk);
>-
>-			if (signal_pending(current)) {
>-				err = sock_intr_errno(timeout);
>-				finish_wait(sk_sleep(sk), &wait);
>-				break;
>-			} else if (timeout == 0) {
>-				err = -EAGAIN;
>-				finish_wait(sk_sleep(sk), &wait);
>+			if (vsock_wait_data(sk, &wait, timeout, &recv_data, target))
> 				break;
>-			}
>-		} else {
>-			ssize_t read;
>+			continue;
>+		}
>
>-			finish_wait(sk_sleep(sk), &wait);
>+		finish_wait(sk_sleep(sk), &wait);

And also this one can be moved in vsock_wait_data().
>
>-			if (ready < 0) {
>-				/* Invalid queue pair content. XXX This should
>-				* be changed to a connection reset in a later
>-				* change.
>-				*/
>+		if (ready < 0) {
>+			/* Invalid queue pair content. XXX This should
>+			 * be changed to a connection reset in a later
>+			 * change.
>+			 */
>
>-				err = -ENOMEM;
>-				goto out;
>-			}
>+			err = -ENOMEM;
>+			goto out;
>+		}
>
>-			err = transport->notify_recv_pre_dequeue(
>-					vsk, target, &recv_data);
>-			if (err < 0)
>-				break;
>+		err = transport->notify_recv_pre_dequeue(vsk,
>+					target, &recv_data);
>+		if (err < 0)
>+			break;
>+		read = transport->stream_dequeue(vsk, msg, len - copied, flags);
>
>-			read = transport->stream_dequeue(
>-					vsk, msg,
>-					len - copied, flags);
>-			if (read < 0) {
>-				err = -ENOMEM;
>-				break;
>-			}
>+		if (read < 0) {
>+			err = -ENOMEM;
>+			break;
>+		}
>
>-			copied += read;
>+		copied += read;
>
>-			err = transport->notify_recv_post_dequeue(
>-					vsk, target, read,
>+		err = transport->notify_recv_post_dequeue(vsk,
>+					target, read,
> 					!(flags & MSG_PEEK), &recv_data);
>-			if (err < 0)
>-				goto out;
>+		if (err < 0)
>+			goto out;
>
>-			if (read >= target || flags & MSG_PEEK)
>-				break;
>+		if (read >= target || flags & MSG_PEEK)
>+			break;
>
>-			target -= read;
>-		}
>+		target -= read;
> 	}
>
> 	if (sk->sk_err)
>@@ -2031,6 +2003,68 @@ vsock_connectible_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
> 	return err;
> }
>
>+static int __vsock_seqpacket_recvmsg(struct sock *sk, struct msghdr *msg,
>+				     size_t len, int flags)
>+{
>+	return -1;
>+}
>+

You can add this function later, when you implement it...

>+static int
>+vsock_connectible_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
>+			  int flags)
>+{
>+	const struct vsock_transport *transport;
>+	struct vsock_sock *vsk;
>+	struct sock *sk;
>+	int err = 0;
>+
>+	sk = sock->sk;
>+
>+	lock_sock(sk);
>+
>+	vsk = vsock_sk(sk);
>+	transport = vsk->transport;
>+
>+	if (!transport || sk->sk_state != TCP_ESTABLISHED) {
>+		/* Recvmsg is supposed to return 0 if a peer performs an
>+		 * orderly shutdown. Differentiate between that case and when a
>+		 * peer has not connected or a local shutdown occurred 
>with the
>+		 * SOCK_DONE flag.
>+		 */
>+		if (!sock_flag(sk, SOCK_DONE))
>+			err = -ENOTCONN;
>+
>+		goto out;
>+	}
>+
>+	if (flags & MSG_OOB) {
>+		err = -EOPNOTSUPP;
>+		goto out;
>+	}
>+
>+	/* We don't check peer_shutdown flag here since peer may actually shut
>+	 * down, but there can be data in the queue that a local socket can
>+	 * receive.
>+	 */
>+	if (sk->sk_shutdown & RCV_SHUTDOWN)
>+		goto out;
>+
>+	/* It is valid on Linux to pass in a zero-length receive buffer.  This
>+	 * is not an error.  We may as well bail out now.
>+	 */
>+	if (!len)
>+		goto out;
>+
>+	if (sk->sk_type == SOCK_STREAM)
>+		err = __vsock_stream_recvmsg(sk, msg, len, flags);
>+	else
>+		err = __vsock_seqpacket_recvmsg(sk, msg, len, flags);

...and also this 'else' branch.

>+
>+out:
>+	release_sock(sk);
>+	return err;
>+}
>+
> static int
> vsock_stream_recvmsg(struct socket *sock, struct msghdr *msg, size_t len,
> 		     int flags)
>-- 
>2.25.1
>