Re: [PATCH 7/7] tap/tcp: Replace tcp_payload_used with a ring buffer style index

On 9/4/26 23:28, aerosouund wrote:

From: Ammar Yasser

Because pasta resets tcp_payload_used to 0 after every tcp flush this means that pasta will very quickly override data in the tcp buffers after sending. This is not a problem when using writev because the data will have been copied elsewhere by the kernel, but this is a problem for vhost where the kernel thread may not have finished processing this data. This is why a synchronous send was required in vhost, which obviously affects performance.

Replace tcp_payload_used with a struct tcp_payload_idx holding the last index we wrote to and a counter that gets updated when we flush packets, from where we last flushed up to that index. Both are grow only and access gets modded by TCP_FRAMES_MEM.

Rewrite functions in tcp_buf.c to use that index and handle the wraparound case (sending a batch that started near the end of the buffers and continued to be at the start) in tcp_payload_flush().

With the buffers no longer reused from under the kernel, drop the synchronous wait from tap_send_frames_vhost() and reclaim descriptors until there are enough for the send instead.

Signed-off-by: Eugenio Pérez Signed-off-by: Ammar Yasser --- tap.c | 9 ++--- tcp_buf.c | 115 ++++++++++++++++++++++++++++++++++++++++-------------- 2 files changed, 88 insertions(+), 36 deletions(-)

diff --git a/tap.c b/tap.c index 0f51e62..670605a 100644 --- a/tap.c +++ b/tap.c @@ -418,7 +418,8 @@ static size_t tap_send_frames_vhost(const struct ctx *c, /* reclaim descriptors if we don't have enough available buffers to * perform this send */ - if (vhost_vq_state[1].num_free < bufs_per_frame * nframes) + while (vhost_vq_state[1].num_free < bufs_per_frame * nframes && + vhost_vq_state[1].last_used_idx != le16toh(avail->idx)) tx_reap();

for (i = 0; i < nframes; i++) { @@ -485,10 +486,6 @@ static size_t tap_send_frames_vhost(const struct ctx *c,

vhost_kick(&vring_used_all[1].used, c->vhost.vq[1].kick_fd);

- /* wait until the kernel finishes processing this send */ - while (avail->idx != vring_used_all[1].used.idx) - ; - return processed_frames; }

@@ -516,7 +513,7 @@ static size_t tap_send_frames_pasta(const struct ctx *c, size_t i;

if (vhost) - return tap_send_frames_vhost(c, iov, bufs_per_frame, nframes); + return tap_send_frames_vhost(c, iov, bufs_per_frame, nframes);

unrelated change, whitespace added in the middle of the tabs.

for (i = 0; i < nbufs; i += bufs_per_frame) { ssize_t rc = writev(c->fd_tap, iov + i, bufs_per_frame); diff --git a/tcp_buf.c b/tcp_buf.c index 9cc541a..1b57ebd 100644 --- a/tcp_buf.c +++ b/tcp_buf.c @@ -57,7 +57,22 @@ static_assert(MSS6 <= sizeof(tcp_payload[0].data), "MSS6 is greater than 65516")

/* References tracking the owner connection of frames in the tap outqueue */ static struct tcp_tap_conn *tcp_frame_conns[TCP_FRAMES_MEM]; -static unsigned int tcp_payload_used; + +/** + * struct tcp_payload_idx - Grow-only cursors into the TCP frame buffers + * @tcp_buf_idx: The last index we wrote data to in the tcp buffers + * @processed: Last index we pushed to the underlying connection + */ +static struct tcp_payload_idx { + unsigned int tcp_buf_idx; + unsigned int processed; +} tcp_idx; + +/* latest index we wrote to into the tcp buffers */ +#define TCP_CURR_IDX (tcp_idx.tcp_buf_idx % TCP_FRAMES_MEM) + +/* current amount of frames queued in tcp buffers but not sent */ +#define TCP_FRAME_COUNT (tcp_idx.tcp_buf_idx - tcp_idx.processed)

Generally macro in upper case are constant. I don't know if it's the rule in passt, but these macros make this hard to read for me.

/* recvmsg()/sendmsg() data for tap */ static struct iovec iov_sock [TCP_FRAMES_MEM + DISCARD_IOV_NUM]; @@ -152,15 +167,46 @@ static void tcp_revert_seq(const struct ctx *c, struct tcp_tap_conn **conns, */ void tcp_payload_flush(const struct ctx *c, const struct timespec *now) { - size_t m; + unsigned int total = TCP_FRAME_COUNT; + unsigned int start = tcp_idx.processed % TCP_FRAMES_MEM; + unsigned int first_batch_size, sent;

- m = tap_send_frames(c, &tcp_l2_iov[0][0], TCP_NUM_IOVS, - tcp_payload_used); - if (m != tcp_payload_used) { - tcp_revert_seq(c, &tcp_frame_conns[m], &tcp_l2_iov[m], - tcp_payload_used - m, now); + if (!total) + return; + + /* What is smaller ? all we what we want to send ? or 128 - the index + * indicating the start of where we wrote this batch ? + */ + first_batch_size = MIN(total, TCP_FRAMES_MEM - start); + + sent = tap_send_frames(c, &tcp_l2_iov[start][0], TCP_NUM_IOVS, + first_batch_size); + if (sent < first_batch_size) { + tcp_revert_seq(c, &tcp_frame_conns[start + sent], + &tcp_l2_iov[start + sent], + first_batch_size - sent, now); + goto out; } - tcp_payload_used = 0; + + /* There was more data to send than from tcp_idx.processed up to 128. + * the rest of the batch is going to be at index 0 up total - + * first_batch_size. + */ + if (total > first_batch_size) { + unsigned int second_batch_size = total - first_batch_size; + size_t m2; + + m2 = tap_send_frames(c, &tcp_l2_iov[0][0], TCP_NUM_IOVS, + second_batch_size); + sent += m2; + + if (m2 < second_batch_size) + tcp_revert_seq(c, &tcp_frame_conns[m2], &tcp_l2_iov[m2], + second_batch_size - m2, now); + } + +out: + tcp_idx.processed += sent; }

/** @@ -238,13 +284,13 @@ int tcp_buf_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, uint32_t seq; int ret;

- iov = tcp_l2_iov[tcp_payload_used]; + iov = tcp_l2_iov[TCP_CURR_IDX]; if (CONN_V4(conn)) - iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp4_payload_ip[tcp_payload_used]); + iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp4_payload_ip[TCP_CURR_IDX]); else - iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp6_payload_ip[tcp_payload_used]); + iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp6_payload_ip[TCP_CURR_IDX]);

- iov[TCP_IOV_ETH] = IOV_OF_LVALUE(tcp_eth_hdr[tcp_payload_used]); + iov[TCP_IOV_ETH] = IOV_OF_LVALUE(tcp_eth_hdr[TCP_CURR_IDX]); payload = iov[TCP_IOV_PAYLOAD].iov_base; seq = conn->seq_to_tap; ret = tcp_prepare_flags(c, conn, flags, &payload->th, @@ -253,7 +299,8 @@ int tcp_buf_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, if (ret <= 0) return ret;

- tcp_frame_conns[tcp_payload_used++] = conn; + tcp_idx.tcp_buf_idx++; + tcp_frame_conns[TCP_CURR_IDX] = conn;

There is a change here: tcp_idx.tcp_buf_idx is incremented before the assignment of tcp_frame_conns[], in previous code it was incremented after.

l4len = optlen + sizeof(struct tcphdr); iov[TCP_IOV_PAYLOAD].iov_len = l4len;

@@ -265,8 +312,9 @@ int tcp_buf_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, tcp_l2_buf_pad(iov);

if (flags & DUP_ACK) { - struct iovec *dup_iov = tcp_l2_iov[tcp_payload_used]; - tcp_frame_conns[tcp_payload_used++] = conn; + struct iovec *dup_iov = tcp_l2_iov[TCP_CURR_IDX]; + tcp_frame_conns[TCP_CURR_IDX] = conn; + tcp_idx.tcp_buf_idx++;

memcpy(dup_iov[TCP_IOV_TAP].iov_base, iov[TCP_IOV_TAP].iov_base, iov[TCP_IOV_TAP].iov_len); @@ -278,7 +326,7 @@ int tcp_buf_send_flag(const struct ctx *c, struct tcp_tap_conn *conn, int flags, dup_iov[TCP_IOV_ETH_PAD].iov_len = iov[TCP_IOV_ETH_PAD].iov_len; }

- if (tcp_payload_used > TCP_FRAMES_MEM - 2) + if (TCP_FRAME_COUNT > TCP_FRAMES_MEM - 2) tcp_payload_flush(c, now);

return 0; @@ -303,21 +351,29 @@ static void tcp_data_to_tap(const struct ctx *c, struct tcp_tap_conn *conn, struct iovec *iov;

conn->seq_to_tap = seq + dlen; - tcp_frame_conns[tcp_payload_used] = conn; - iov = tcp_l2_iov[tcp_payload_used]; + tcp_frame_conns[TCP_CURR_IDX] = conn; + iov = tcp_l2_iov[TCP_CURR_IDX]; if (CONN_V4(conn)) { if (no_csum) { - struct iovec *iov_prev = tcp_l2_iov[tcp_payload_used - 1]; + /* TCP_CURR_IDX may be zero if the underlying + * tcp_idx.tcp_buf_idx is a multiple of 128, minus one + * will yield an invalid index. The previous index to 0 + * is 127. + */ + unsigned int prev_idx = (TCP_CURR_IDX + + TCP_FRAMES_MEM - 1) % + TCP_FRAMES_MEM; + struct iovec *iov_prev = tcp_l2_iov[prev_idx]; const struct iphdr *iph = iov_prev[TCP_IOV_IP].iov_base;

/* overwrite IP4_CSUM flag as we set the checksum */ check = iph->check; } - iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp4_payload_ip[tcp_payload_used]); + iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp4_payload_ip[TCP_CURR_IDX]); } else if (CONN_V6(conn)) { - iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp6_payload_ip[tcp_payload_used]); + iov[TCP_IOV_IP] = IOV_OF_LVALUE(tcp6_payload_ip[TCP_CURR_IDX]); } - iov[TCP_IOV_ETH].iov_base = &tcp_eth_hdr[tcp_payload_used]; + iov[TCP_IOV_ETH].iov_base = &tcp_eth_hdr[TCP_CURR_IDX]; payload = iov[TCP_IOV_PAYLOAD].iov_base; payload->th.th_off = sizeof(struct tcphdr) / 4; payload->th.th_x2 = 0; @@ -329,7 +385,8 @@ static void tcp_data_to_tap(const struct ctx *c, struct tcp_tap_conn *conn,

tcp_l2_buf_pad(iov);

- if (++tcp_payload_used > TCP_FRAMES_MEM - 1) + tcp_idx.tcp_buf_idx++; + if (TCP_FRAME_COUNT > TCP_FRAMES_MEM - 1) tcp_payload_flush(c, now); }

@@ -369,15 +426,13 @@ int tcp_buf_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, return -1; }

- if (tcp_payload_used + fill_bufs > TCP_FRAMES_MEM) { + if (TCP_FRAME_COUNT + (unsigned int)fill_bufs > TCP_FRAMES_MEM) tcp_payload_flush(c, now);

- /* Silence Coverity CWE-125 false positive */ - tcp_payload_used = 0; - } - for (i = 0, iov = iov_sock + DISCARD_IOV_NUM; i < fill_bufs; i++, iov++) { - iov->iov_base = &tcp_payload[tcp_payload_used + i].data; + unsigned int idx = (TCP_CURR_IDX + i) % TCP_FRAMES_MEM; + + iov->iov_base = &tcp_payload[idx].data; iov->iov_len = mss; } if (iov_rem) @@ -451,7 +506,7 @@ int tcp_buf_data_from_sock(const struct ctx *c, struct tcp_tap_conn *conn, dlen = mss; seq = conn->seq_to_tap; for (i = 0; i < send_bufs; i++) { - int no_csum = i && i != send_bufs - 1 && tcp_payload_used; + int no_csum = i && i != send_bufs - 1 && TCP_CURR_IDX;

no_csum is here to not recompute the csum if nothing change between the packet, so we compute only on the first packet. With the TCP_CURR_IDX we recompute it each time TCP_CURR_IDX is not zero.

bool push = false;

if (i == send_bufs - 1) {