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);
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)