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

[Laurent Vivier <lvivier@redhat.com>]

On 9/4/26 23:28, aerosouund wrote:
> From: Ammar Yasser <aerosound161@gmail.com>
> 
> 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 <eperezma@redhat.com>
> Signed-off-by: Ammar Yasser <aerosound161@gmail.com>
> ---
>   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) {