rpms
/
kronosnet
7
0
Fork 0
Code Issues Pull Requests Releases Activity
kronosnet/SOURCES/bz1763714-fix-data-deliver-...

416 lines
15 KiB
Diff
Raw Blame History

commit db21da87bba6017c8343f9c6f255b21813ffd5d0
Author: Fabio M. Di Nitto <fdinitto@redhat.com>
Date: Tue Oct 15 06:46:36 2019 +0200
[host] rename variables to make it easier to read the code
Signed-off-by: Fabio M. Di Nitto <fdinitto@redhat.com>
diff --git a/libknet/host.c b/libknet/host.c
index abb1f89..ac26b89 100644
--- a/libknet/host.c
+++ b/libknet/host.c
@@ -569,7 +569,7 @@ static void _clear_cbuffers(struct knet_host *host, seq_num_t rx_seq_num)
int _seq_num_lookup(struct knet_host *host, seq_num_t seq_num, int defrag_buf, int clear_buf)
{
- size_t i, j; /* circular buffer indexes */
+ size_t head, tail; /* circular buffer indexes */
seq_num_t seq_dist;
char *dst_cbuf = host->circular_buffer;
char *dst_cbuf_defrag = host->circular_buffer_defrag;
@@ -585,13 +585,13 @@ int _seq_num_lookup(struct knet_host *host, seq_num_t seq_num, int defrag_buf, i
seq_dist = *dst_seq_num - seq_num;
}
- j = seq_num % KNET_CBUFFER_SIZE;
+ head = seq_num % KNET_CBUFFER_SIZE;
if (seq_dist < KNET_CBUFFER_SIZE) { /* seq num is in ring buffer */
if (!defrag_buf) {
- return (dst_cbuf[j] == 0) ? 1 : 0;
+ return (dst_cbuf[head] == 0) ? 1 : 0;
} else {
- return (dst_cbuf_defrag[j] == 0) ? 1 : 0;
+ return (dst_cbuf_defrag[head] == 0) ? 1 : 0;
}
} else if (seq_dist <= SEQ_MAX - KNET_CBUFFER_SIZE) {
memset(dst_cbuf, 0, KNET_CBUFFER_SIZE);
@@ -600,16 +600,16 @@ int _seq_num_lookup(struct knet_host *host, seq_num_t seq_num, int defrag_buf, i
}
/* cleaning up circular buffer */
- i = (*dst_seq_num + 1) % KNET_CBUFFER_SIZE;
+ tail = (*dst_seq_num + 1) % KNET_CBUFFER_SIZE;
- if (i > j) {
- memset(dst_cbuf + i, 0, KNET_CBUFFER_SIZE - i);
- memset(dst_cbuf, 0, j + 1);
- memset(dst_cbuf_defrag + i, 0, KNET_CBUFFER_SIZE - i);
- memset(dst_cbuf_defrag, 0, j + 1);
+ if (tail > head) {
+ memset(dst_cbuf + tail, 0, KNET_CBUFFER_SIZE - tail);
+ memset(dst_cbuf, 0, head + 1);
+ memset(dst_cbuf_defrag + tail, 0, KNET_CBUFFER_SIZE - tail);
+ memset(dst_cbuf_defrag, 0, head + 1);
} else {
- memset(dst_cbuf + i, 0, j - i + 1);
- memset(dst_cbuf_defrag + i, 0, j - i + 1);
+ memset(dst_cbuf + tail, 0, head - tail + 1);
+ memset(dst_cbuf_defrag + tail, 0, head - tail + 1);
}
*dst_seq_num = seq_num;
commit 1e473cf26d55c2b6ff8d5bfaa5aa689554de803c
Author: Fabio M. Di Nitto <fdinitto@redhat.com>
Date: Tue Oct 15 06:53:24 2019 +0200
[host] fix defrag buffers reclaim logic
The problem:
- let's assume a 2 nodes (A and B) cluster setup
- node A sends fragmented packets to node B and there is
packet loss on the network.
- node B receives all those fragments and attempts to
reassemble them.
- node A sends packet seq_num X in Y fragments.
- node B receives only part of the fragments and stores
them in a defrag buf.
- packet loss stops.
- node A continues to send packets and a seq_num
roll-over takes place.
- node A sends a new packet seq_num X in Y fragments.
- node B gets confused here because the parts of the old
packet seq_num X are still stored and the buffer
has not been reclaimed.
- node B continues to rebuild packet seq_num X with
old stale data and new data from after the roll-over.
- node B completes reassembling the packet and delivers
junk to the application.
The solution:
Add a much stronger buffer reclaim logic that will apply
on each received packet and not only when defrag buffers
are needed, as there might be a mix of fragmented and not
fragmented packets in-flight.
The new logic creates a window of N packets that can be
handled at the same time (based on the number of buffers)
and clear everything else.
Fixes https://github.com/kronosnet/kronosnet/issues/261
Signed-off-by: Fabio M. Di Nitto <fdinitto@redhat.com>
diff --git a/libknet/host.c b/libknet/host.c
index ac26b89..85d4626 100644
--- a/libknet/host.c
+++ b/libknet/host.c
@@ -562,6 +562,35 @@ static void _clear_cbuffers(struct knet_host *host, seq_num_t rx_seq_num)
}
}
+static void _reclaim_old_defrag_bufs(struct knet_host *host, seq_num_t seq_num)
+{
+ seq_num_t head, tail; /* seq_num boundaries */
+ int i;
+
+ head = seq_num + 1;
+ tail = seq_num - (KNET_MAX_LINK + 1);
+
+ /*
+ * expire old defrag buffers
+ */
+ for (i = 0; i < KNET_MAX_LINK; i++) {
+ if (host->defrag_buf[i].in_use) {
+ /*
+ * head has done a rollover to 0+
+ */
+ if (tail > head) {
+ if ((host->defrag_buf[i].pckt_seq >= head) && (host->defrag_buf[i].pckt_seq <= tail)) {
+ host->defrag_buf[i].in_use = 0;
+ }
+ } else {
+ if ((host->defrag_buf[i].pckt_seq >= head) || (host->defrag_buf[i].pckt_seq <= tail)){
+ host->defrag_buf[i].in_use = 0;
+ }
+ }
+ }
+ }
+}
+
/*
* check if a given packet seq num is in the circular buffers
* defrag_buf = 0 -> use normal cbuf 1 -> use the defrag buffer lookup
@@ -579,6 +608,8 @@ int _seq_num_lookup(struct knet_host *host, seq_num_t seq_num, int defrag_buf, i
_clear_cbuffers(host, seq_num);
}
+ _reclaim_old_defrag_bufs(host, seq_num);
+
if (seq_num < *dst_seq_num) {
seq_dist = (SEQ_MAX - seq_num) + *dst_seq_num;
} else {
commit 5bd88ebd63af20577095c2c98975f0f1781ba46a
Author: Fabio M. Di Nitto <fdinitto@redhat.com>
Date: Tue Oct 15 07:02:05 2019 +0200
[rx] copy data into the defrag buffer only if we know the size of the frame
Signed-off-by: Fabio M. Di Nitto <fdinitto@redhat.com>
diff --git a/libknet/threads_rx.c b/libknet/threads_rx.c
index b2a5dad..6c26cdc 100644
--- a/libknet/threads_rx.c
+++ b/libknet/threads_rx.c
@@ -186,8 +186,10 @@ static int pckt_defrag(knet_handle_t knet_h, struct knet_header *inbuf, ssize_t
defrag_buf->frag_size = *len;
}
- memmove(defrag_buf->buf + ((inbuf->khp_data_frag_seq - 1) * defrag_buf->frag_size),
- inbuf->khp_data_userdata, *len);
+ if (defrag_buf->frag_size) {
+ memmove(defrag_buf->buf + ((inbuf->khp_data_frag_seq - 1) * defrag_buf->frag_size),
+ inbuf->khp_data_userdata, *len);
+ }
defrag_buf->frag_recv++;
defrag_buf->frag_map[inbuf->khp_data_frag_seq] = 1;
commit cd59986900510119d8e7b63d33ad35466d480858
Author: Fabio M. Di Nitto <fdinitto@redhat.com>
Date: Tue Oct 15 07:16:22 2019 +0200
[test] add ability to knet_bench to specify a fixed packet size for perf test
Signed-off-by: Fabio M. Di Nitto <fdinitto@redhat.com>
diff --git a/libknet/tests/knet_bench.c b/libknet/tests/knet_bench.c
index dc04239..54b5303 100644
--- a/libknet/tests/knet_bench.c
+++ b/libknet/tests/knet_bench.c
@@ -67,6 +67,8 @@ static int test_type = TEST_PING;
static uint64_t perf_by_size_size = 1 * ONE_GIGABYTE;
static uint64_t perf_by_time_secs = 10;
+static uint32_t force_packet_size = 0;
+
struct node {
int nodeid;
int links;
@@ -109,6 +111,7 @@ static void print_help(void)
printf(" -s nodeid that will generate traffic for benchmarks\n");
printf(" -S [size|seconds] when used in combination with -T perf-by-size it indicates how many GB of traffic to generate for the test. (default: 1GB)\n");
printf(" when used in combination with -T perf-by-time it indicates how many Seconds of traffic to generate for the test. (default: 10 seconds)\n");
+ printf(" -x force packet size for perf-by-time or perf-by-size\n");
printf(" -C repeat the test continously (default: off)\n");
printf(" -X[XX] show stats at the end of the run (default: 1)\n");
printf(" 1: show handle stats, 2: show summary link stats\n");
@@ -250,7 +253,7 @@ static void setup_knet(int argc, char *argv[])
memset(nodes, 0, sizeof(nodes));
- while ((rv = getopt(argc, argv, "aCT:S:s:ldfom:wb:t:n:c:p:X::P:z:h")) != EOF) {
+ while ((rv = getopt(argc, argv, "aCT:S:s:ldfom:wb:t:n:c:p:x:X::P:z:h")) != EOF) {
switch(rv) {
case 'h':
print_help();
@@ -406,6 +409,13 @@ static void setup_knet(int argc, char *argv[])
perf_by_size_size = (uint64_t)atoi(optarg) * ONE_GIGABYTE;
perf_by_time_secs = (uint64_t)atoi(optarg);
break;
+ case 'x':
+ force_packet_size = (uint32_t)atoi(optarg);
+ if ((force_packet_size < 1) || (force_packet_size > 65536)) {
+ printf("Unsupported packet size %u (accepted 1 - 65536)\n", force_packet_size);
+ exit(FAIL);
+ }
+ break;
case 'C':
continous = 1;
break;
@@ -874,7 +884,7 @@ static int setup_send_buffers_common(struct knet_mmsghdr *msg, struct iovec *iov
printf("TXT: Unable to malloc!\n");
return -1;
}
- memset(tx_buf[i], 0, KNET_MAX_PACKET_SIZE);
+ memset(tx_buf[i], i, KNET_MAX_PACKET_SIZE);
iov_out[i].iov_base = (void *)tx_buf[i];
memset(&msg[i].msg_hdr, 0, sizeof(struct msghdr));
msg[i].msg_hdr.msg_iov = &iov_out[i];
@@ -898,6 +908,9 @@ static void send_perf_data_by_size(void)
setup_send_buffers_common(msg, iov_out, tx_buf);
while (packetsize <= KNET_MAX_PACKET_SIZE) {
+ if (force_packet_size) {
+ packetsize = force_packet_size;
+ }
for (i = 0; i < PCKT_FRAG_MAX; i++) {
iov_out[i].iov_len = packetsize;
}
@@ -926,7 +939,7 @@ static void send_perf_data_by_size(void)
knet_send(knet_h, ctrl_message, TEST_STOP, channel);
- if (packetsize == KNET_MAX_PACKET_SIZE) {
+ if ((packetsize == KNET_MAX_PACKET_SIZE) || (force_packet_size)) {
break;
}
@@ -1175,6 +1188,9 @@ static void send_perf_data_by_time(void)
memset(&clock_end, 0, sizeof(clock_start));
while (packetsize <= KNET_MAX_PACKET_SIZE) {
+ if (force_packet_size) {
+ packetsize = force_packet_size;
+ }
for (i = 0; i < PCKT_FRAG_MAX; i++) {
iov_out[i].iov_len = packetsize;
}
@@ -1205,7 +1221,7 @@ static void send_perf_data_by_time(void)
knet_send(knet_h, ctrl_message, TEST_STOP, channel);
- if (packetsize == KNET_MAX_PACKET_SIZE) {
+ if ((packetsize == KNET_MAX_PACKET_SIZE) || (force_packet_size)) {
break;
}
commit e28e2ea7c7e8139a6792ec1508215d4560b53e65
Author: Fabio M. Di Nitto <fdinitto@redhat.com>
Date: Wed Oct 16 08:10:23 2019 +0200
[test] add packet verification option to knet_bench
Signed-off-by: Fabio M. Di Nitto <fdinitto@redhat.com>
diff --git a/libknet/tests/knet_bench.c b/libknet/tests/knet_bench.c
index 54b5303..c9e1c06 100644
--- a/libknet/tests/knet_bench.c
+++ b/libknet/tests/knet_bench.c
@@ -47,6 +47,7 @@ static char *compresscfg = NULL;
static char *cryptocfg = NULL;
static int machine_output = 0;
static int use_access_lists = 0;
+static int use_pckt_verification = 0;
static int bench_shutdown_in_progress = 0;
static pthread_mutex_t shutdown_mutex = PTHREAD_MUTEX_INITIALIZER;
@@ -76,6 +77,11 @@ struct node {
struct sockaddr_storage address[KNET_MAX_LINK];
};
+struct pckt_ver {
+ uint32_t len;
+ uint32_t chksum;
+};
+
static void print_help(void)
{
printf("knet_bench usage:\n");
@@ -117,6 +123,7 @@ static void print_help(void)
printf(" 1: show handle stats, 2: show summary link stats\n");
printf(" 3: show detailed link stats\n");
printf(" -a enable machine parsable output (default: off).\n");
+ printf(" -v enable packet verification for performance tests (default: off).\n");
}
static void parse_nodes(char *nodesinfo[MAX_NODES], int onidx, int port, struct node nodes[MAX_NODES], int *thisidx)
@@ -253,7 +260,7 @@ static void setup_knet(int argc, char *argv[])
memset(nodes, 0, sizeof(nodes));
- while ((rv = getopt(argc, argv, "aCT:S:s:ldfom:wb:t:n:c:p:x:X::P:z:h")) != EOF) {
+ while ((rv = getopt(argc, argv, "aCT:S:s:lvdfom:wb:t:n:c:p:x:X::P:z:h")) != EOF) {
switch(rv) {
case 'h':
print_help();
@@ -411,11 +418,14 @@ static void setup_knet(int argc, char *argv[])
break;
case 'x':
force_packet_size = (uint32_t)atoi(optarg);
- if ((force_packet_size < 1) || (force_packet_size > 65536)) {
- printf("Unsupported packet size %u (accepted 1 - 65536)\n", force_packet_size);
+ if ((force_packet_size < 64) || (force_packet_size > 65536)) {
+ printf("Unsupported packet size %u (accepted 64 - 65536)\n", force_packet_size);
exit(FAIL);
}
break;
+ case 'v':
+ use_pckt_verification = 1;
+ break;
case 'C':
continous = 1;
break;
@@ -654,6 +664,24 @@ static void setup_knet(int argc, char *argv[])
}
}
+/*
+ * calculate weak chksum (stole from corosync for debugging purposes)
+ */
+static uint32_t compute_chsum(const unsigned char *data, uint32_t data_len)
+{
+ unsigned int i;
+ unsigned int checksum = 0;
+
+ for (i = 0; i < data_len; i++) {
+ if (checksum & 1) {
+ checksum |= 0x10000;
+ }
+
+ checksum = ((checksum >> 1) + (unsigned char)data[i]) & 0xffff;
+ }
+ return (checksum);
+}
+
static void *_rx_thread(void *args)
{
int rx_epoll;
@@ -766,6 +794,20 @@ static void *_rx_thread(void *args)
}
continue;
}
+ if (use_pckt_verification) {
+ struct pckt_ver *recv_pckt = (struct pckt_ver *)msg[i].msg_hdr.msg_iov->iov_base;
+ uint32_t chksum;
+
+ if (msg[i].msg_len != recv_pckt->len) {
+ printf("Wrong packet len received: %u expected: %u!\n", msg[i].msg_len, recv_pckt->len);
+ exit(FAIL);
+ }
+ chksum = compute_chsum((const unsigned char *)msg[i].msg_hdr.msg_iov->iov_base + sizeof(struct pckt_ver), msg[i].msg_len - sizeof(struct pckt_ver));
+ if (recv_pckt->chksum != chksum){
+ printf("Wrong packet checksum received: %u expected: %u!\n", recv_pckt->chksum, chksum);
+ exit(FAIL);
+ }
+ }
rx_pkts++;
rx_bytes = rx_bytes + msg[i].msg_len;
current_pckt_size = msg[i].msg_len;
@@ -913,6 +955,11 @@ static void send_perf_data_by_size(void)
}
for (i = 0; i < PCKT_FRAG_MAX; i++) {
iov_out[i].iov_len = packetsize;
+ if (use_pckt_verification) {
+ struct pckt_ver *tx_pckt = (struct pckt_ver *)&iov_out[i].iov_base;
+ tx_pckt->len = iov_out[i].iov_len;
+ tx_pckt->chksum = compute_chsum((const unsigned char *)iov_out[i].iov_base + sizeof(struct pckt_ver), iov_out[i].iov_len - sizeof(struct pckt_ver));
+ }
}
total_pkts_to_tx = perf_by_size_size / packetsize;
@@ -1193,6 +1240,11 @@ static void send_perf_data_by_time(void)
}
for (i = 0; i < PCKT_FRAG_MAX; i++) {
iov_out[i].iov_len = packetsize;
+ if (use_pckt_verification) {
+ struct pckt_ver *tx_pckt = (struct pckt_ver *)iov_out[i].iov_base;
+ tx_pckt->len = iov_out[i].iov_len;
+ tx_pckt->chksum = compute_chsum((const unsigned char *)iov_out[i].iov_base + sizeof(struct pckt_ver), iov_out[i].iov_len - sizeof(struct pckt_ver));
+ }
}
printf("[info]: testing with %u bytes packet size for %" PRIu64 " seconds.\n", packetsize, perf_by_time_secs);
Reference in New Issue View Git Blame Copy Permalink