[RFC PATCH 5/8] rxrpc: Implement sendfile() support

[David Howells]

Implement the sendpage protocol operation so that sendfile() will work
directly with AF_RXRPC calls.  To use sendfile() to communicate with a call
requires the call to be specified beforehand with setsockopt():

	setsockopt(client, SOL_RXRPC, RXRPC_SELECT_CALL_FOR_SEND,
		   &call_id, sizeof(call_id));
	sendfile(client, source, &pos, st.st_size);

The specified call ID can be cleared:

	call_id = 0;
	setsockopt(client, SOL_RXRPC, RXRPC_SELECT_CALL_FOR_SEND,
		   &call_id, sizeof(call_id));

or changed.

Signed-off-by: David Howells <dhowells@xxxxxxxxxx>
---

 net/rxrpc/af_rxrpc.c    |   40 +++++++++++++++++++++++++++++++++++++++-
 net/rxrpc/ar-internal.h |    2 ++
 net/rxrpc/sendmsg.c     |   39 +++++++++++++++++++++++++++++++++++++++
 3 files changed, 80 insertions(+), 1 deletion(-)

diff --git a/net/rxrpc/af_rxrpc.c b/net/rxrpc/af_rxrpc.c
index 8ac014aff7a2..41420c456e77 100644
--- a/net/rxrpc/af_rxrpc.c
+++ b/net/rxrpc/af_rxrpc.c
@@ -798,6 +798,44 @@ static int rxrpc_bind_channel(struct rxrpc_sock *rx2, int fd)
 	return ret;
 }
 
+/*
+ * Splice into a call.  The call to send as part of must have been set with
+ * setsockopt(RXRPC_SELECT_CALL_FOR_SEND).
+ */
+static ssize_t rxrpc_sendpage(struct socket *sock, struct page *page, int offset,
+			      size_t size, int flags)
+{
+	struct rxrpc_sock *rx = rxrpc_sk(sock->sk);
+	struct rxrpc_call *call;
+	ssize_t ret;
+
+	_enter("{%d},,%u,%zu,%x", rx->sk.sk_state, offset, size, flags);
+
+	lock_sock(&rx->sk);
+
+	read_lock_bh(&rx->recvmsg_lock);
+	call = rx->selected_send_call;
+	if (!call) {
+		read_unlock_bh(&rx->recvmsg_lock);
+		release_sock(&rx->sk);
+		return -EBADSLT;
+	}
+
+	rxrpc_get_call(call, rxrpc_call_got);
+	read_unlock_bh(&rx->recvmsg_lock);
+
+	ret = mutex_lock_interruptible(&call->user_mutex);
+	release_sock(&rx->sk);
+	if (ret == 0) {
+		ret = rxrpc_do_sendpage(rx, call, page, offset, size, flags);
+		mutex_unlock(&call->user_mutex);
+	}
+
+	rxrpc_put_call(call, rxrpc_call_put);
+	_leave(" = %zd", ret);
+	return ret;
+}
+
 /*
  * Set the default call for 'targetless' operations such as splice(), SIOCINQ
  * and SIOCOUTQ and also as a filter for recvmsg().  Calling this function
@@ -1279,9 +1317,9 @@ static const struct proto_ops rxrpc_rpc_ops = {
 	.setsockopt	= rxrpc_setsockopt,
 	.getsockopt	= rxrpc_getsockopt,
 	.sendmsg	= rxrpc_sendmsg,
+	.sendpage	= rxrpc_sendpage,
 	.recvmsg	= rxrpc_recvmsg,
 	.mmap		= sock_no_mmap,
-	.sendpage	= sock_no_sendpage,
 };
 
 static struct proto rxrpc_proto = {
diff --git a/net/rxrpc/ar-internal.h b/net/rxrpc/ar-internal.h
index ec2c614082b9..bec398c66341 100644
--- a/net/rxrpc/ar-internal.h
+++ b/net/rxrpc/ar-internal.h
@@ -1107,6 +1107,8 @@ struct key *rxrpc_look_up_server_security(struct rxrpc_connection *,
  * sendmsg.c
  */
 int rxrpc_do_sendmsg(struct rxrpc_sock *, struct msghdr *, size_t);
+ssize_t rxrpc_do_sendpage(struct rxrpc_sock *, struct rxrpc_call *,
+			  struct page *, int, size_t, int);
 
 /*
  * server_key.c
diff --git a/net/rxrpc/sendmsg.c b/net/rxrpc/sendmsg.c
index 1d38e279e2ef..77699008c428 100644
--- a/net/rxrpc/sendmsg.c
+++ b/net/rxrpc/sendmsg.c
@@ -762,6 +762,45 @@ int rxrpc_do_sendmsg(struct rxrpc_sock *rx, struct msghdr *msg, size_t len)
 	return ret;
 }
 
+/*
+ * Handle data input from a splice.  The call to send as part of must have been
+ * set with setsockopt(RXRPC_SELECT_CALL_FOR_SEND).
+ */
+ssize_t rxrpc_do_sendpage(struct rxrpc_sock *rx, struct rxrpc_call *call,
+			  struct page *page, int offset, size_t size, int flags)
+{
+	struct bio_vec bv = {
+		.bv_page	= page,
+		.bv_offset	= offset,
+		.bv_len		= size,
+	};
+	struct msghdr msg = {
+		.msg_flags	= flags,
+	};
+
+	_enter(",,%lx,%u,%zu,%x", page->index, offset, size, flags);
+
+	switch (READ_ONCE(call->state)) {
+	case RXRPC_CALL_COMPLETE:
+		return -ESHUTDOWN;
+	default:
+		return -EPROTO;
+	case RXRPC_CALL_CLIENT_SEND_REQUEST:
+	case RXRPC_CALL_SERVER_ACK_REQUEST:
+	case RXRPC_CALL_SERVER_SEND_REPLY:
+		break;
+	}
+
+	/* Ideally, we'd allow sendfile() to end the Tx phase - but there's no
+	 * way for userspace to communicate this option through that syscall.
+	 */
+	//if (flags & MSG_SENDPAGE_NOTLAST)
+	msg.msg_flags |= MSG_MORE;
+
+	iov_iter_bvec(&msg.msg_iter, WRITE, &bv, 1, size);
+	return rxrpc_send_data(rx, call, &msg, size, NULL);
+}
+
 /**
  * rxrpc_kernel_send_data - Allow a kernel service to send data on a call
  * @sock: The socket the call is on