[Devel] [PATCH hci-8.1 2/3] fs/fuse kio: refactor transmit path in pcs_sockio_send()
Liu Kui
kui.liu at virtuozzo.com
Wed Sep 30 10:34:32 MSK 2026
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 at 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)
More information about the Devel
mailing list