Add end timeout to lookup function

3 files changed, 47 insertions, 64 deletions

diff --git a/src/main.c b/src/main.c
index 267f324..7e5cf5b 100644
--- a/src/main.c
+++ b/src/main.c
@@ -174,16 +174,29 @@ int main(int argc, char** argv) {
 	// Init the lookup
 	struct lookup *lookup = &dht.lookup;
 	{
-		lookup->timeout = 0;
+		lookup->timeout = time(NULL) + 120;
 		lookup->target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}};
 
+		for(size_t i = 0; i < 8; i++) {
+			lookup->closest_addr[i].port = 0;
+		}
+
+		struct message* message_cursor = outbuff;
+		// Issue the first round of requests manually
 		struct entry* entry[8];
 		int found = routing_closest(&lookup->target, sizeof(entry)/sizeof(entry[0]), entry);
-		assert(found == 8);
 		for(size_t i = 0; i < found; i++) {
-			lookup->closest[i] = entry[i]->id;
-			lookup->closest_addr[i] = entry[i]->addr;
+			struct sockaddr_in dest = {
+				.sin_family = AF_INET,
+				.sin_addr = {entry[i]->addr.ip},
+				.sin_port = entry[i]->addr.port,
+			};
+
+			int rc = send_lookup(&dht, &lookup->target, time(NULL), (struct sockaddr*)&dest, sizeof(struct sockaddr_in), &(struct msgbuff){&message_cursor, outbuff + OUTBOX_SIZE});
+			assert(rc == 0);
 		}
+
+		flush_messages(dht.sfd, outbuff, message_cursor);
 	}
 
 #define RECV_BUFF_SIZE 4096
diff --git a/src/proto.c b/src/proto.c
index d5d377c..2adc52b 100644
--- a/src/proto.c
+++ b/src/proto.c
@@ -218,7 +218,7 @@ int send_lookup(struct dht* dht, struct nodeid* target, time_t now, const struct
 	dht->requestdata[reqId].cont.lookup = &dht->lookup;
 
 	dht->requestdata[reqId].fun = &lookup_response;
-	dht->requestdata[reqId].timeout = now + PROTO_TMOUT;
+	dht->requestdata[reqId].timeout = 0;
 	dht->requestdata[reqId].timeout_fun = NULL;
 	memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len);
 	dht->requestdata[reqId].addr_len = dest_len;
@@ -371,7 +371,6 @@ PROCESS_REPONSE(lookup_response) {
 	uint32_t match_i = 0;
 	uint8_t match_score = 0;
 	for(size_t i = 0; i < 8; i++) {
-		dbg("PORT %d", cont->lookup->closest_addr[i].port);
 		if(cont->lookup->closest_addr[i].port == 0) {
 			match_i = i;
 			match_score = UINT8_MAX; // Bogus value to signal that we found something
@@ -407,8 +406,21 @@ PROCESS_REPONSE(lookup_response) {
 
 	// Fan out the search if the nodes are better than the worst one in the frontier
 	for(uint8_t i = 0; i < nodes_len; i++) {
-		// Don't fan out to anything that is a worse match than our current frontier
-		if(prefix(&nodes[i], &cont->lookup->target) <= worst_match) continue;
+		uint8_t candidate_score = prefix(&nodes[i], &cont->lookup->target);
+
+		bool better_than_any = false;
+		bool already_matched = false;
+		for(uint8_t j = 0; j < 8; j++) {
+			if(candidate_score > prefix(&cont->lookup->closest[j], &cont->lookup->target))
+				better_than_any = true;
+
+			if(prefix(&nodes[i], &cont->lookup->closest[j]) == 160)
+				already_matched = true;
+		}
+
+		// Don't fan out to anything that is a worse match than or already
+		// included in our current frontier
+		if(!better_than_any || already_matched) continue;
 
 		// @ROBUST: Some nodes report a bunch of nodes in the same ip. Maybe we
 		// could check for that here
@@ -426,7 +438,7 @@ PROCESS_REPONSE(lookup_response) {
 			fatal("failed %d", rc);
 		}
 
-		cont->lookup->timeout = now;
+		cont->lookup->timeout = now + 120;
 	}
 
 	return 0;
@@ -822,6 +834,10 @@ static void recalulate_waketime(struct dht *dht) {
 		if(dht->wake == 0 || (timeout != 0 && difftime(timeout, dht->wake) < 0))
 			dht->wake = timeout;
 	}
+
+	if(dht->wake == 0 || difftime(dht->lookup.timeout, dht->wake) < 0) {
+		dht->wake = dht->lookup.timeout;
+	}
 }
 
 int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* remote, socklen_t remote_len, time_t now, struct message** output, const struct message* const output_end) {
@@ -958,56 +974,10 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 		}
 	}
 
-	recalulate_waketime(dht);
-
 	int rc = handle_packet(dht, now, type, transaction_set ? transaction : NULL, transaction_len, query_set ? query : NULL, query_len, buff, recv_len, remote, remote_len, &msgbuff);
 	assert(rc == 0);
 
-	// Debug output
-
-	{
-		printf("In flight |");
-		uint16_t inflight = 0;
-		for(int i = 0; i < MAX_INFLIGHT; i++) {
-			if(dht->reqalloc[i]) {
-				printf("#");
-				inflight++;
-			} else {
-				printf(" ");
-			}
-		}
-		printf("|\n");
-		prom_gauge_set(requestsInFlight, inflight, NULL);
-	}
-	{
-		int filled;
-		int total;
-#define LFACLEN 64
-		double load_factor[LFACLEN] = {0};
-		routing_status(&filled, &total, load_factor, LFACLEN);
-		prom_gauge_set(activeNodes, filled, NULL);
-		dbg("%d/%d nodes in routing table", filled, total);
-
-#define GRAPHY 5
-		// @HACK @CLEANUP: I'm pretty zooted right now. I have zero confidence
-		// that this is correct. It looks allright though.
-		for(int y = 0; y < GRAPHY; y++) {
-			printf("|");
-			for(int x = 0; x < LFACLEN; x++) {
-				double cell_load = CLAMP((load_factor[x] - ((1.0/GRAPHY) * (GRAPHY-y-1))) * GRAPHY, 0, 1);
-				if(cell_load == 0.0) {
-					printf(" ");
-				} else if (cell_load > 1.0 - 1.0/GRAPHY) {
-					printf("#");
-				} else {
-					printf("%d", (int)(cell_load*10));
-				}
-			}
-			printf("|\n");
-		}
-#undef GRAPHY
-#undef LFACLEN
-	}
+	recalulate_waketime(dht);
 
 	return 0;
 }
diff --git a/test/proto.c b/test/proto.c
index 1c71d6e..f31d16a 100644
--- a/test/proto.c
+++ b/test/proto.c
@@ -757,7 +757,7 @@ void test_lookup_response() {
 		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:aaaaaaaaaaaaaaaaaaaae1:q9:find_node1:t1:21:y1:qe", outbuff[0].payload, 91);
-		TEST_ASSERT_EQUAL(dht.lookup.timeout, now);
+		TEST_ASSERT_EQUAL(now+120, dht.lookup.timeout);
 
 		memcpy(&remote, &outbuff[0].dest, sizeof(remote));
 	}
@@ -775,7 +775,7 @@ void test_lookup_response() {
 			// The first two bytes match
 			dht.lookup.closest[i].inner_b[0] = 'a';
 			dht.lookup.closest[i].inner_b[1] = 'a';
-			dht.lookup.closest[i].inner_b[2] += i;
+			dht.lookup.closest[i].inner_b[2] = 'a' + i;
 
 			dht.lookup.closest_addr[i].ip = 0;
 			dht.lookup.closest_addr[i].port = 1;
@@ -787,9 +787,9 @@ void test_lookup_response() {
 		// The node now finally responds, but woops only the first byte of its
 		// ID matches. That's worse than the frontier and shouldn't cause any
 		// addtional adjustment to the frontier.
-		// It also includes a new node that's also behind the frontier, we
-		// don't send anything to that one
-		char buff[] = "d1:y1:r1:t1:21:rd2:id20:aBBBBBBBBBBBBBBBBBBB5:nodes26:aaBBBBBBBBBBBBBBBBBB\xFF\xFF\xFF\xFF\x00\x01""ee";
+		// It also includes a new node that's better than some nodes in the
+		// current frontier, but is also already included.
+		char buff[] = "d1:y1:r1:t1:21:rd2:id20:aBBBBBBBBBBBBBBBBBBB5:nodes52:aaaBBBBBBBBBBBBBBBBB\xFF\xFF\xFF\xFF\x00\x01""aaBBBBBBBBBBBBBBBBBB\xFF\xFF\xFF\xFE\x00\x01""ee";
 		struct message* message_cursor = outbuff;
 		int rc = proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, sizeof(remote), now, &message_cursor, outbuff+2);
 
@@ -797,14 +797,14 @@ void test_lookup_response() {
 		// We only need to check this once since we always pick the first slot
 		// with a given score. It's a little implementation dependant, but it
 		// beats having 8 asserts.
-		TEST_ASSERT_EQUAL_CHAR_ARRAY(&dht.lookup.closest[0], "aaBBBBBBBBBBBBBBBBBB", 20);
+		TEST_ASSERT_EQUAL_CHAR_ARRAY(&dht.lookup.closest[0], "aaaBBBBBBBBBBBBBBBBB", 20);
 
-		// And we didn't fan out to the new node, even though it's actually a better match than any of the ones we have
+		// And we didn't fan out to the new node since it's already part of the current frontier
 		TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff);
 
 		// Timeout isn't updated since we didn't send anything
 		// @FRAGILE this has to match the step size of now
-		TEST_ASSERT_EQUAL(dht.lookup.timeout, now-1);
+		TEST_ASSERT_EQUAL(now+120-1, dht.lookup.timeout);
 	}
 
 	proto_end(&dht);