From e3eb0d2eda0bb8c4decf9bc56b025157e70387c2 Mon Sep 17 00:00:00 2001 From: Jesper Jensen Date: Sat, 2 Oct 2021 00:25:37 +0200 Subject: Add timeout handling --- src/main.c | 24 +++++++- src/proto.c | 185 +++++++++++++++++++++++++++++++++++++++------------------ src/proto.h | 23 ++++--- src/routing.c | 8 +++ src/routing.h | 1 + test/proto.c | 138 +++++++++++++++++++++++++++++++++++++++--- test/routing.c | 58 ++++++++++++++++-- 7 files changed, 351 insertions(+), 86 deletions(-) diff --git a/src/main.c b/src/main.c index 7534e76..5094c8c 100644 --- a/src/main.c +++ b/src/main.c @@ -35,12 +35,27 @@ int main(int argc, char** argv) { bool timedout = false; if(!dht.pause){ + time_t next = 0; + struct entry* oldest; routing_oldest(&oldest); - if(oldest != NULL) { - dbg("Set timeout to %ld", oldest->expire - time(NULL)); + if(oldest != NULL) + next = oldest->expire; + + for(int i = 0; i < MAX_INFLIGHT; i++) { + if(!dht.reqalloc[i]) + continue; + + time_t timeout = dht.requestdata[i].timeout; + if(next == 0 || (timeout != 0 && difftime(timeout, next) < 0)) + next = timeout; + } + + if(next != 0) { + time_t sleepfor = next - time(NULL); + dbg("Set timeout to %ld", sleepfor); struct timeval tv = { - .tv_sec = oldest->expire - time(NULL), + .tv_sec = sleepfor, .tv_usec = 0, }; if(tv.tv_sec <= 0) { @@ -48,7 +63,10 @@ int main(int argc, char** argv) { } else { setsockopt(dht.sfd, SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv)); } + } else { + fatal("We have to wait some amount of time right now"); } + } else { printf("DHT timeout is paused"); } diff --git a/src/proto.c b/src/proto.c index d8d7580..520f766 100644 --- a/src/proto.c +++ b/src/proto.c @@ -97,6 +97,7 @@ struct msgbuff { #define PROTO_ENOREQ 2 PROCESS_REPONSE(getclient_response); +PROCESS_TIMEOUT(getclient_timeout); uint8_t rand_byte() { int limit = RAND_MAX - (RAND_MAX % UINT8_MAX); @@ -106,56 +107,112 @@ uint8_t rand_byte() { return val; } -int send_ping(struct dht* dht, struct nodeid* self, time_t now, const int sfd, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff) { - // Generate a random target - struct nodeid target; - for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { - *target_byte = rand_byte(); - } +int create_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* target, uint16_t tid) { + char* buff_end = buff + *buff_len; - uint16_t reqId; - if(!alloc_req(dht, &reqId)) { - return PROTO_ENOREQ; - } + int rc = snprintf(buff, buff_end - buff, "d1:ad2:id20:"); + if(rc < 0) + fatal("Failed to write packet"); + buff += rc; + memcpy(buff, self, sizeof(struct nodeid)); + buff += sizeof(struct nodeid); + rc = snprintf(buff, buff_end - buff, "6:target20:"); + if(rc < 0) + fatal("Failed to write packet"); + buff += rc; + memcpy(buff, &target, sizeof(struct nodeid)); + buff += sizeof(struct nodeid); + rc = snprintf(buff, buff_end - buff, "e1:q9:find_node1:t%d:%d1:y1:qe", (int)(log10(tid+1)+1), tid); + if(rc < 0) + fatal("Failed to write packet"); + buff += rc; + + *buff_len = buff - (buff_end - *buff_len); + return 0; +} + +int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is_new, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff) { if(*msgbuff->messages >= msgbuff->messages_end) return PROTO_ENOREQ; struct message* message = *msgbuff->messages; + uint16_t reqId; + if(!alloc_req(dht, &reqId)) { + return PROTO_ENOREQ; + } + dbg("Allocating request %d", reqId); + memcpy(&message->dest, dest_addr, dest_len); message->dest_len = dest_len; - char* buff = message->payload; - size_t i = 0; - dbg("Allocating request %d", reqId); + struct ping* data = &dht->requestdata[reqId].cont.ping; + if(!node_is_new) + data->remote_id = *expected; + data->is_new = node_is_new; + data->attempt = 0; + dht->requestdata[reqId].fun = &getclient_response; dht->requestdata[reqId].timeout = now + PROTO_TMOUT; - dht->requestdata[reqId].timeout_fun = NULL; + dht->requestdata[reqId].timeout_fun = getclient_timeout; memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len); + dht->requestdata[reqId].addr_len = dest_len; - int rc = snprintf(buff+i, 128-i, "d1:ad2:id20:"); - if(rc < 0) - fatal("Failed to write packet"); - i += rc; - memcpy(buff+i, self, sizeof(struct nodeid)); - i += sizeof(struct nodeid); - rc = snprintf(buff+i, 128-i, "6:target20:"); - if(rc < 0) - fatal("Failed to write packet"); - i += rc; - memcpy(buff+i, &target, sizeof(struct nodeid)); - i += sizeof(struct nodeid); - rc = snprintf(buff+i, 128-i, "e1:q9:find_node1:t%d:%d1:y1:qe", (int)(log10(reqId+1)+1), reqId); - if(rc < 0) - fatal("Failed to write packet"); - i += rc; + // Generate a random target + struct nodeid target; + for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { + *target_byte = rand_byte(); + } - message->payload_len = i; + message->payload_len = 128; + int rc = create_ping(message->payload, &message->payload_len, &dht->self, &target, reqId); + if(rc != 0) { + return rc; + } (*msgbuff->messages)++; return 0; } +PROCESS_TIMEOUT(getclient_timeout) { + // @HACK: This really sucks. maybe we should just pass in the request id + size_t reqId = (typeof(dht->requestdata[0])*)((void*)cont - offsetof(typeof(dht->requestdata[0]), cont)) - dht->requestdata; + + if(cont->ping.attempt >= 2) { + if(cont->ping.is_new) + return 0; + + routing_remove(&cont->ping.remote_id); + return 0; + } + + dbg("Retrying request %d", reqId); + + if(*msgbuff->messages >= msgbuff->messages_end) + return PROTO_ENOREQ; + struct message* message = *msgbuff->messages; + + memcpy(&message->dest, &dht->requestdata[reqId].addr, dht->requestdata[reqId].addr_len); + message->dest_len = dht->requestdata[reqId].addr_len; + + // Generate a random target + struct nodeid target; + for(uint8_t *target_byte = (uint8_t*)⌖ target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) { + *target_byte = rand_byte(); + } + + message->payload_len = 128; + int rc = create_ping(message->payload, &message->payload_len, &dht->self, &target, reqId); + if(rc != 0) { + fatal("Can't create ping"); + } + (*msgbuff->messages)++; + + dht->requestdata[reqId].timeout = now + PROTO_TMOUT; + cont->ping.attempt++; + return PROTO_EDISC; +} + PROCESS_REPONSE(getclient_response) { struct benc_node stream[256]; struct bcursor bcursor; @@ -246,7 +303,24 @@ PROCESS_REPONSE(getclient_response) { } } - // Print the candidates + // The response was good, so save the node + if(cont->ping.is_new) { + struct entry* entry; + if(!routing_offer(&id, &entry)) { + dbg("We are no longer interested"); + return 0; + } + + struct sockaddr_in* ipv4 = (struct sockaddr_in*)remote; + entry->addr.ip = ipv4->sin_addr.s_addr; + entry->addr.port = ipv4->sin_port; + entry->expire = now + PROTO_UNCTM; + } else { + fatal("Ping has not been implemented"); + // Touch the routing entry + } + + // Fan out the search if the results were interesting for(uint8_t i = 0; i < nodes_len; i++) { dbgl_id(&nodes[i]); printf("Candidate %s:%d\n", inet_ntoa(ips[i]), ntohs(ports[i])); @@ -258,26 +332,17 @@ PROCESS_REPONSE(getclient_response) { }; if(routing_interested(&nodes[i])) { - int rc = send_ping(dht, &dht->self, now, socket, (struct sockaddr*)&dest, sizeof(struct sockaddr_in), msgbuff); - if(rc != 0) { - err("send_ping failed %d", rc); + int rc = send_ping(dht, &nodes[i], now, true, (struct sockaddr*)&dest, sizeof(struct sockaddr_in), msgbuff); + if(rc == PROTO_ENOREQ) { + return rc; + } else if(rc != 0) { + fatal("send_ping failed %d", rc); } } else { dbg("Not interested in node"); } } - struct entry* entry; - if(!routing_offer(&id, &entry)) { - dbg("We are no longer interested"); - return 0; - } - - struct sockaddr_in* ipv4 = (struct sockaddr_in*)remote; - entry->addr.ip = ipv4->sin_addr.s_addr; - entry->addr.port = ipv4->sin_port; - entry->expire = now + UNCERTAIN_TIME; - return 0; } @@ -339,7 +404,7 @@ void proto_begin(struct dht* dht, time_t now, struct message** output, const str /* rc = snprintf(buff+i, 128-i, "e1:q4:ping1:t2:ab1:y1:qe"); */ /* i += rc; */ - send_ping(dht, &dht->self, now, dht->sfd, cur->ai_addr, cur->ai_addrlen, &msgbuff); + send_ping(dht, NULL, now, true, cur->ai_addr, cur->ai_addrlen, &msgbuff); } freeaddrinfo(res); @@ -359,18 +424,21 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* for(int i = 0; i < MAX_INFLIGHT; i++) { if(!dht->reqalloc[i]) continue; + if(dht->requestdata[i].timeout == 0) + continue; if(difftime(now, dht->requestdata[i].timeout) < 0) continue; - if(dht->requestdata[i].timeout_fun != NULL) - dht->requestdata[i].timeout_fun(dht, &dht->self, now, &msgbuff); + int rc = dht->requestdata[i].timeout_fun(dht, &dht->self, now, &dht->requestdata[i].cont, &msgbuff); - dht->requestdata[i].fun = NULL; - dht->requestdata[i].timeout_fun = NULL; - dht->requestdata[i].timeout = 0; - dht->reqalloc[i] = false; + if(rc != PROTO_EDISC) { + dht->requestdata[i].fun = NULL; + dht->requestdata[i].timeout_fun = NULL; + dht->requestdata[i].timeout = 0; + dht->reqalloc[i] = false; - dht->pause = false; + dht->pause = false; + } } struct entry* oldest = NULL; @@ -384,7 +452,7 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* dest.sin_family = AF_INET; dest.sin_addr.s_addr = oldest->addr.ip; dest.sin_port = oldest->addr.port; - int rc = send_ping(dht, &dht->self, now, dht->sfd, (const struct sockaddr*)&dest, sizeof(dest), &msgbuff); + int rc = send_ping(dht, &oldest->id, now, false, (const struct sockaddr*)&dest, sizeof(dest), &msgbuff); if(rc == PROTO_ENOREQ) { dht->pause = true; return 0; @@ -496,9 +564,11 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* return 0; } - int rc = dht->requestdata[reqId].fun(dht, now, buff, recv_len, dht->sfd, (struct sockaddr*)remote, remote_len, &msgbuff); - if(rc == PROTO_EDISC) { + dht->pause = false; + int rc = dht->requestdata[reqId].fun(dht, now, &dht->requestdata[reqId].cont, buff, recv_len, dht->sfd, (struct sockaddr*)remote, remote_len, &msgbuff); + if(rc == PROTO_ENOREQ) { dht->pause = true; + } else if(rc == PROTO_EDISC) { return 0; } @@ -506,7 +576,6 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* dht->requestdata[reqId].timeout_fun = NULL; dht->requestdata[reqId].timeout = 0; dht->reqalloc[reqId] = false; - dht->pause = false; } else if(type == CT_QUERY) { // Must be a query if(!query_set) fatal("No query function in query request"); diff --git a/src/proto.h b/src/proto.h index 0e85918..3ca7fc7 100644 --- a/src/proto.h +++ b/src/proto.h @@ -9,42 +9,41 @@ #define MAX_DISC 32 #define MAX_INFLIGHT 32 -#define UNCERTAIN_TIME 900 +#define PROTO_UNCTM 900 #define PROTO_TMOUT 60 -struct discovery { - struct in_addr addr; - uint16_t port; - struct nodeid expected_id; +struct ping { + struct nodeid remote_id; + int attempt; + bool is_new; }; union message_cont { - struct discovery discovery; + struct ping ping; }; struct dht; struct msgbuff; -#define PROCESS_REPONSE(NAME) int NAME(struct dht* dht, time_t now, union message_cont* cont, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff) +#define PROCESS_REPONSE(NAME) int (NAME)(struct dht* dht, time_t now, union message_cont* cont, char* packet, size_t packet_len, int socket, struct sockaddr* remote, socklen_t remote_len, struct msgbuff* msgbuff) typedef PROCESS_REPONSE(cont); -#define PROCESS_TIMEOUT(NAME) int NAME(struct dht* dht, struct nodeid* self, time_t now, union message_cont* cont, struct msgbuff* msgbuff) +#define PROCESS_TIMEOUT(NAME) int (NAME)(struct dht* dht, struct nodeid* self, time_t now, union message_cont* cont, struct msgbuff* msgbuff) typedef PROCESS_TIMEOUT(tmout); struct dht { struct nodeid self; int sfd; - struct in_addr addrs[MAX_DISC]; - struct discovery pending_discover[MAX_DISC]; bool pause; bool reqalloc[MAX_INFLIGHT]; struct { struct sockaddr_storage addr; - cont fun; + socklen_t addr_len; + cont* fun; time_t timeout; - tmout timeout_fun; + tmout* timeout_fun; union message_cont cont; } requestdata[MAX_INFLIGHT]; }; diff --git a/src/routing.c b/src/routing.c index f26f1ba..1448f55 100644 --- a/src/routing.c +++ b/src/routing.c @@ -99,6 +99,8 @@ static uint16_t base_bucket(struct nodeid* id) { struct entry* routing_get(struct nodeid* id) { uint16_t baseIndex = base_bucket(id); for(size_t i = baseIndex; i < baseIndex + BUCKETSIZE; i++) { + if(!table[i].set) continue; + if(memcmp(&table[i].id, id, sizeof(struct nodeid)) == 0) { return &table[i]; } @@ -107,6 +109,12 @@ struct entry* routing_get(struct nodeid* id) { return NULL; } +void routing_remove(struct nodeid* id) { + struct entry* entry = routing_get(id); + + entry->set = false; +} + bool routing_interested(struct nodeid* id) { uint16_t bucketIndex = prefix(&myID, id); // The nodeid is the same as our own diff --git a/src/routing.h b/src/routing.h index e42eb71..3972afa 100644 --- a/src/routing.h +++ b/src/routing.h @@ -33,3 +33,4 @@ void routing_oldest(struct entry** dest); size_t routing_closest(struct nodeid* needle, size_t n, struct entry** res); struct entry* routing_get(struct nodeid* id); +void routing_remove(struct nodeid* self); diff --git a/test/proto.c b/test/proto.c index 3becc98..03a4a45 100644 --- a/test/proto.c +++ b/test/proto.c @@ -12,7 +12,7 @@ void test_begin_pings_bootstrap_node() { dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; struct message* message_cursor = outbuff; - proto_begin(&dht, &message_cursor, outbuff+10); + proto_begin(&dht, time(NULL), &message_cursor, outbuff+10); TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); @@ -30,14 +30,14 @@ void test_response_from_initial_probe() { dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; { struct message* message_cursor = outbuff; - proto_begin(&dht, &message_cursor, outbuff+2); + proto_begin(&dht, time(NULL), &message_cursor, outbuff+2); remote_len = outbuff[0].dest_len; memcpy(&remote, &outbuff[0].dest, remote_len); } char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes26:bbbbbbbbbbbbbbbbbbbb\xFF\xFF\xFF\xFF\x00\x00""ee"; struct message* message_cursor = outbuff; - proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, &message_cursor, outbuff+2); + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, 10, &message_cursor, outbuff+2); TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff+1, "Sends one packet"); TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); @@ -50,6 +50,7 @@ void test_response_from_initial_probe() { TEST_ASSERT_NOT_NULL(entry); TEST_ASSERT_EQUAL(((struct sockaddr_in*)&remote)->sin_addr.s_addr, entry->addr.ip); TEST_ASSERT_EQUAL(((struct sockaddr_in*)&remote)->sin_port, entry->addr.port); + TEST_ASSERT_EQUAL(PROTO_UNCTM+10, entry->expire); } void test_reponse_from_wrong_ip() { @@ -59,7 +60,7 @@ void test_reponse_from_wrong_ip() { dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; { struct message* message_cursor = outbuff; - proto_begin(&dht, &message_cursor, outbuff+2); + proto_begin(&dht, time(NULL), &message_cursor, outbuff+2); } // This is fragile, since the ip of the bootstrap node could change, and we @@ -72,7 +73,7 @@ void test_reponse_from_wrong_ip() { char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes26:bbbbbbbbbbbbbbbbbbbb\xFF\xFF\xFF\xFF\x00\x00""ee"; struct message* message_cursor = outbuff; - proto_run(&dht, buff, sizeof(buff), &remote, sizeof(remote), &message_cursor, outbuff+2); + proto_run(&dht, buff, sizeof(buff), &remote, sizeof(remote), 0, &message_cursor, outbuff+2); // We shouldn't send any packets, since the response is rejected TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff); @@ -91,7 +92,7 @@ void test_ping() { dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; { struct message* message_cursor = outbuff; - proto_begin(&dht, &message_cursor, outbuff+2); + proto_begin(&dht, time(NULL), &message_cursor, outbuff+2); } struct sockaddr_in remote; @@ -101,7 +102,7 @@ void test_ping() { char buff[] = "d1:ad2:id20:abcdefghij0123456789e1:q4:ping1:t2:aa1:y1:qe"; struct message* message_cursor = outbuff; - proto_run(&dht, buff, sizeof(buff), &remote, sizeof(remote), &message_cursor, outbuff+2); + proto_run(&dht, buff, sizeof(buff), &remote, sizeof(remote), 0, &message_cursor, outbuff+2); // We should have sent a response TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); @@ -109,3 +110,126 @@ void test_ping() { TEST_ASSERT_EQUAL(48, outbuff[0].payload_len); TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:t2:aa1:y1:r1:rd2:id20:BBBBBBBBBBBBBBBBBBBBee", outbuff[0].payload, 47); } + +void test_note_times_out() { + routing_flush(); + struct message outbuff[2] = {0}; + + struct sockaddr_storage remote; + socklen_t remote_len; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + { + struct message* message_cursor = outbuff; + proto_begin(&dht, 0, &message_cursor, outbuff+2); + remote_len = outbuff[0].dest_len; + memcpy(&remote, &outbuff[0].dest, remote_len); + } + + // The node responds to the ping at t=5 + { + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes26:bbbbbbbbbbbbbbbbbbbb\xFF\xFF\xFF\xFF\x00\x00""ee"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, 5, &message_cursor, outbuff+2); + } + + // 15 minutes later a timeout is fired + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)&remote, remote_len, 905, &message_cursor, outbuff+2); + + // Which should create a retry ping and a ping for the (now) uncertain node + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+2); + TEST_ASSERT_EQUAL(91, outbuff[1].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:ad2:id20:BBBBBBBBBBBBBBBBBBBB6:target20:", outbuff[1].payload, 43); + TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:01:y1:qe", outbuff[1].payload+63, 28); +} + +void test_response_after_retry() { + struct message outbuff[10] = {0}; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + + { + struct message* message_cursor = outbuff; + proto_begin(&dht, 0, &message_cursor, outbuff+2); + } + + // After TMOUT seconds we retry the ping + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, PROTO_TMOUT, &message_cursor, outbuff+2); + + TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); + TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); + TEST_ASSERT_EQUAL_CHAR_ARRAY("d1:ad2:id20:BBBBBBBBBBBBBBBBBBBB6:target20:", outbuff[0].payload, 43); + TEST_ASSERT_EQUAL_CHAR_ARRAY("e1:q9:find_node1:t1:01:y1:qe", outbuff[0].payload+63, 28); +} + +void test_remove_from_routing_after_3_retries() { + struct message outbuff[10] = {0}; + time_t now = 0; + + struct sockaddr_storage remote; + socklen_t remote_len; + + struct dht dht; + dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + struct nodeid other = (struct nodeid){.inner={0x61616161, 0x61616161, 0x61616161, 0x61616161, 0x61616161}}; + + { + struct message* message_cursor = outbuff; + proto_begin(&dht, 0, &message_cursor, outbuff+2); + remote_len = outbuff[0].dest_len; + memcpy(&remote, &outbuff[0].dest, remote_len); + } + now += 10; + + { + // The node responds + // We return no new nodes to stop any new pings from going out + char buff[] = "d1:y1:r1:t1:01:rd2:id20:aaaaaaaaaaaaaaaaaaaa5:nodes0:""ee"; + struct message* message_cursor = outbuff; + proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+2); + } + struct entry* entry = routing_get(&other); + TEST_ASSERT_NOT_NULL(entry); + + now += PROTO_UNCTM; + { + // The node becomes uncertain + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, now, &message_cursor, outbuff+2); + TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff+1, "Ping was not sent"); + } + + now += PROTO_TMOUT; + { + // 1st retry + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, now, &message_cursor, outbuff+2); + } + + now += PROTO_TMOUT; + { + // 2nd retry + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, now, &message_cursor, outbuff+2); + } + + now += PROTO_TMOUT; + { + // 3rd retry + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, now, &message_cursor, outbuff+2); + } + + now += PROTO_TMOUT; + // Drop the node + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)NULL, 0, now, &message_cursor, outbuff+2); + + TEST_ASSERT_EQUAL_PTR_MESSAGE(message_cursor, outbuff, "The timeout should send a message"); + entry = routing_get(&other); + TEST_ASSERT_NULL(entry); +} diff --git a/test/routing.c b/test/routing.c index 7905bc0..eb899ba 100644 --- a/test/routing.c +++ b/test/routing.c @@ -36,7 +36,7 @@ void test_can_find_added() { // Set the entries entry->addr = addr; - entry->last = time(NULL); + entry->expire = time(NULL); // Ask for 3 nodes struct entry *out[3] = {0}; @@ -65,7 +65,7 @@ void test_discard_offer_when_bucket_full() { // Set the entries entry->addr = addr; - entry->last = time(NULL); + entry->expire = time(NULL); new.inner[4] += 1; } @@ -86,7 +86,7 @@ void test_discard_offer_when_nodeid_added_twice() { struct entry* entry; TEST_ASSERT_TRUE(routing_offer(&new, &entry)); entry->addr = addr; - entry->last = time(NULL); + entry->expire = time(NULL); bool accept = routing_offer(&new, &entry); TEST_ASSERT_FALSE_MESSAGE(accept, "Accepted entry when bucket was full"); @@ -112,7 +112,7 @@ void test_not_interested_when_nodeid_in_table() { struct entry* entry; TEST_ASSERT_TRUE(routing_offer(&new, &entry)); entry->addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; - entry->last = time(NULL); + entry->expire = time(NULL); bool interest = routing_interested(&new); TEST_ASSERT_FALSE_MESSAGE(interest, "Was interested in node"); @@ -135,7 +135,7 @@ void test_not_interested_when_bucket_is_full() { // Set the entries entry->addr = addr; - entry->last = time(NULL); + entry->expire = time(NULL); new.inner[4] += 1; } @@ -162,7 +162,7 @@ void test_lowest_ts_is_oldest() { entry->id = new; entry->addr = addr; // Invert the timestamps to make the last one have lowest timestamp - entry->last = 2-i; + entry->expire = 2-i; new.inner_b[19] += 1; } @@ -173,3 +173,49 @@ void test_lowest_ts_is_oldest() { TEST_ASSERT_NOT_NULL(dest); TEST_ASSERT_EQUAL(1, dest->id.inner_b[19]); } + +void test_get_after_offer() { + routing_flush(); + + // The address we are going to store + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + + // Make a nodeid that is one bit different + struct nodeid new = self; + + new.inner[4] += 1; + struct entry* entry; + TEST_ASSERT_TRUE_MESSAGE(routing_offer(&new, &entry), "Did not accept new entry"); + + // Set the entry + entry->addr = addr; + entry->expire = 10; + + struct entry* e = routing_get(&new); + + TEST_ASSERT_NOT_NULL(e); + TEST_ASSERT_EQUAL_MEMORY(&addr, &e->addr, sizeof(struct addr)); +} + +void test_get_after_offer_and_remove() { + routing_flush(); + + // The address we are going to store + struct addr addr = (struct addr){.ip = IP(128,0,0,1), .port = 0}; + + // Make a nodeid that is one bit different + struct nodeid new = self; + + new.inner[4] += 1; + struct entry* entry; + TEST_ASSERT_TRUE_MESSAGE(routing_offer(&new, &entry), "Did not accept new entry"); + + // Set the entry + entry->addr = addr; + entry->expire = 10; + + routing_remove(&new); + struct entry* e = routing_get(&new); + + TEST_ASSERT_NULL(e); +} -- cgit v1.2.3