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

[aerosouund <aerosound161@gmail.com>]

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);
 
 	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)
 
 /* 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;
 	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;
 		bool push = false;
 
 		if (i == send_bufs - 1) {
-- 
2.39.5 (Apple Git-154)