[Devel] [PATCH hci-8.1 2/3] fs/fuse kio: refactor transmit path in pcs_sockio_send()

[Liu Kui <kui.liu@virtuozzo.com>]

The transmit path sends one iov_iter segment per sock_sendmsg() call,
and the bvec builder produces one bio_vec per 4k page, so a 1M data
chunk costs up to 256 sendmsg calls where the socket could take the
whole iov_iter at once.

Replace do_send_one_seg() with do_sockio_sendmsg(), which submits the
entire iov_iter in a single sock_sendmsg() call (with MSG_SPLICE_PAGES
when all pages qualify), reducing pcs_sockio_send() to pure queue
management.

Also clear PCS_MSG_BUSY in the transport whenever IO on the msg ends,
like the rdma transport does, instead of relying on krpc to mop it up.

https://virtuozzo.atlassian.net/browse/VSTOR-142167

Signed-off-by: Liu Kui <kui.liu@virtuozzo.com>
---
 fs/fuse/kio/pcs/pcs_sock_io.c | 207 ++++++++++++++++++++++--------------------
 1 file changed, 109 insertions(+), 98 deletions(-)

diff --git a/fs/fuse/kio/pcs/pcs_sock_io.c b/fs/fuse/kio/pcs/pcs_sock_io.c
index fa26105b2185..828c8a98cabf 100644
--- a/fs/fuse/kio/pcs/pcs_sock_io.c
+++ b/fs/fuse/kio/pcs/pcs_sock_io.c
@@ -139,54 +139,6 @@ static bool pcs_should_fail_sock_io(void)
 }
 #endif
 
-static int do_send_one_seg(struct socket *sock, struct iov_iter *it, size_t left, int has_more)
-{
-	int ret = -EIO;
-	size_t size = iov_iter_single_seg_count(it);
-	bool more = (size < left);
-	int flags = (MSG_DONTWAIT | MSG_NOSIGNAL) | ((more || has_more) ? MSG_MORE : 0);
-	struct msghdr msg = { .msg_flags = flags };
-
-	if (unlikely(pcs_use_eor) && !more)
-		flags |= MSG_EOR;
-
-	DTRACE("sock(%px)  len:%ld, more:%d\n", sock, iov_iter_count(it), more);
-
-	if (unlikely(!size))
-		return 0;
-
-	if (pcs_should_fail_sock_io())
-		goto out;
-
-	if (iov_iter_is_kvec(it)) {
-		struct kvec kv;
-
-		iov_iter_get_kvec(it, &kv);
-		BUG_ON(kv.iov_len != size);
-
-		iov_iter_kvec(&msg.msg_iter, ITER_SOURCE, &kv, 1, size);
-	} else if (iov_iter_is_bvec(it)) {
-		struct bio_vec bv;
-
-		/* iov_iter_iovec() for ITER_BVEC */
-		bvec_set_page(&bv, it->bvec->bv_page,
-			      min(size, it->bvec->bv_len - it->iov_offset),
-			      it->bvec->bv_offset + it->iov_offset);
-
-		if (sendpage_ok(bv.bv_page))
-			msg.msg_flags |= MSG_SPLICE_PAGES;
-		else
-			msg.msg_flags &= ~MSG_SPLICE_PAGES;
-
-		iov_iter_bvec(&msg.msg_iter, ITER_SOURCE, &bv, 1, bv.bv_len);
-	}
-
-	ret = sock_sendmsg(sock, &msg);
-out:
-	DTRACE("sock(%px) len:%ld, more:%d ret:%d\n", sock, iov_iter_count(it), more, ret);
-	return ret;
-}
-
 static int do_sock_recv(struct socket *sock, void *buf, size_t len)
 {
 
@@ -352,14 +304,110 @@ static void pcs_sockio_recv(struct pcs_sockio *sio)
 
 }
 
+static bool iov_iter_sendpages_ok(struct iov_iter *it)
+{
+	const struct bio_vec *bvec = it->bvec;
+
+	if (!iov_iter_is_bvec(it))
+		return false;
+
+	while (bvec < (it->bvec + it->nr_segs)) {
+		if (!sendpages_ok(bvec->bv_page, bvec->bv_len, bvec->bv_offset))
+			return false;
+		bvec++;
+	}
+
+	return true;
+}
+
+static int do_sockio_sendmsg(struct pcs_sockio *sio, struct pcs_msg *msg)
+{
+	struct iov_iter *it = &sio->write_iter;
+	unsigned int flags = MSG_DONTWAIT | MSG_NOSIGNAL;
+
+	if (unlikely(pcs_should_fail_sock_io()))
+		return -PCS_ERR_NET_ABORT;
+
+	set_bit(PCS_MSG_BUSY, &msg->flags);
+
+	/* pair with the smp_mb() between set PCS_MSG_ABORTED and test PCS_MSG_BUSY */
+	smp_mb();
+	/* Shouldn't abort a half sent message  */
+	if (unlikely(test_bit(PCS_MSG_ABORTED, &msg->flags)) && !sio->write_offset) {
+		pcs_set_local_error(&msg->error, PCS_ERR_CANCEL_REQUEST);
+		goto out;
+	}
+
+	while (sio->write_offset < msg->size) {
+		size_t left = msg->size - sio->write_offset;
+		struct msghdr sockmsg = {.msg_flags = flags};
+
+		TRACE("sio(%px) offset:%d msg:%px left:%zu, it->len:%zu\n", sio, sio->write_offset, msg, left, iov_iter_count(it));
+
+		if (!iov_iter_count(it)) {
+			/* Current iter is exhausted, init new one */
+			msg->get_iter(msg, sio->write_offset, it, WRITE);
+		}
+
+		BUG_ON(iov_iter_count(it) > left);
+
+		if (iov_iter_count(it) < left ||
+		    msg->list.next != &sio->write_queue)
+			sockmsg.msg_flags |= MSG_MORE;
+		if (iov_iter_count(it) == left && unlikely(pcs_use_eor))
+			sockmsg.msg_flags |= MSG_EOR;
+
+		if (iov_iter_sendpages_ok(it))
+			sockmsg.msg_flags |= MSG_SPLICE_PAGES;
+
+		sockmsg.msg_iter = *it;
+
+		while (iov_iter_count(it)) {
+			int ret;
+
+			DTRACE("sock(%px) len:%zu, flags:%08x\n", sio->socket, iov_iter_count(it), sockmsg.msg_flags);
+			ret = sock_sendmsg(sio->socket, &sockmsg);
+			DTRACE("sock(%px) len:%zu, flags:%08x, ret=%d\n", sio->socket, iov_iter_count(it), sockmsg.msg_flags, ret);
+			if (ret <= 0) {
+				if (ret == -EAGAIN) {
+					if (!time_is_before_jiffies((unsigned long)(msg->start_time + sio->send_timeout)))
+						/* Keep PCS_MSG_BUSY: the msg is half
+						 * sent and the transfer will resume,
+						 * it must not be aborted meanwhile. */
+						return 0;
+
+					ret = -PCS_ERR_WRITE_TIMEOUT;
+				} else {
+					WARN_ON(ret == 0);
+					ret = -PCS_ERR_NET_ABORT;
+				}
+
+				/* The caller aborts the connection: no more IO on
+				 * this msg, release it for pcs_msg_abort() before
+				 * sio_abort() completes it. */
+				clear_and_wake_up_bit(PCS_MSG_BUSY, &msg->flags);
+				return ret;
+			}
+
+			iov_iter_advance(it, ret);
+			sio->write_offset += ret;
+		}
+		WARN_ON(msg_data_left(&sockmsg));
+	}
+
+out:
+	clear_and_wake_up_bit(PCS_MSG_BUSY, &msg->flags);
+	return 1;
+}
+
 static void pcs_sockio_send(struct pcs_sockio *sio)
 {
 	struct pcs_rpc *ep __maybe_unused = sio->netio.parent;
-	struct iov_iter *it = &sio->write_iter;
 	unsigned long loop_timeout = jiffies + PCS_SIO_SLICE;
 	struct pcs_msg * msg;
-	int done = 0;
 	int count = 0;
+	int ret = 0;
+	bool to_push = false;
 
 	while (!list_empty(&sio->write_queue)) {
 		msg = list_first_entry(&sio->write_queue, struct pcs_msg, list);
@@ -380,54 +428,15 @@ static void pcs_sockio_send(struct pcs_sockio *sio)
 			return;
 		}
 
-		set_bit(PCS_MSG_BUSY, &msg->flags);
-
-		/* pair with the smp_mb() between set PCS_MSG_ABORTED and test PCS_MSG_BUSY */
-		smp_mb();
-
-		/* Shouldn't abort a half sent message  */
-		if (unlikely(test_bit(PCS_MSG_ABORTED, &msg->flags)) && !sio->write_offset) {
-			pcs_set_local_error(&msg->error, PCS_ERR_CANCEL_REQUEST);
-			goto skip_send;
-		}
-
-		/* TODO: cond resched here? */
-		while (sio->write_offset < msg->size) {
-			size_t left = msg->size - sio->write_offset;
-			int has_more = (msg->list.next != &sio->write_queue);
-			int n;
-
-			TRACE(PEER_FMT "offset:%d msg:%px left:%ld, it->len:%ld\n", PEER_ARGS(ep), sio->write_offset, msg,
-			      left, iov_iter_count(it));
-
-			if (!iov_iter_count(it)) {
-				/* Current iter is exhausted, init new one */
-				msg->get_iter(msg, sio->write_offset, it, WRITE);
-			}
-			BUG_ON(iov_iter_count(it) > left);
-			n = do_send_one_seg(sio->socket, it, left, has_more);
-			if (n > 0) {
-				sio->write_offset += n;
-				iov_iter_advance(it, n);
-				done = 1;
-			} else {
-				if (n == 0)
-					WARN_ON(1);
-
-				if (n == -EAGAIN) {
-					unsigned long timeout = msg->start_time + sio->send_timeout;
-					if (time_is_before_jiffies(timeout))
-						sio_abort(sio, PCS_ERR_WRITE_TIMEOUT);
-					return;
-				}
-				sio_abort(sio, PCS_ERR_NET_ABORT);
-				return;
-			}
+		/* Send out this msg */
+		ret = do_sockio_sendmsg(sio, msg);
+		if (ret <= 0) {
+			if (ret < 0)
+				sio_abort(sio, -ret);
+			return;
 		}
 
-skip_send:
-		clear_and_wake_up_bit(PCS_MSG_BUSY, &msg->flags);
-
+		to_push = true;
 		list_del_init(&msg->list);
 		sio->write_queue_len -= msg->size;
 
@@ -435,8 +444,9 @@ static void pcs_sockio_send(struct pcs_sockio *sio)
 			if (sio->write_wakeup)
 				sio->write_wakeup(sio);
 		}
+
 		sio->write_offset = 0;
-		iov_iter_kvec(it, WRITE, NULL, 0, 0);
+		iov_iter_kvec(&sio->write_iter, WRITE, NULL, 0, 0);
 		pcs_msg_sent(msg);
 		msg->done(msg);
 		if (++count >= PCS_SIO_PREEMPT_LIMIT ||
@@ -445,7 +455,8 @@ static void pcs_sockio_send(struct pcs_sockio *sio)
 			break;
 		}
 	}
-	if (done)
+
+	if (to_push)
 		sio_push(sio);
 }
 

-- 
Git-155)

_______________________________________________
Devel mailing list
Devel@openvz.org
https://lists.openvz.org/mailman/listinfo/devel