Add a first approximation of a lookup algorithm

This isn't really "correct" but it's probably going to work. It's
a rough implementation of the Kademlia lookup algorithm, with the
modification that we don't keep track of anything behind the frontier.
We are therefore counting on peers being ok with us potentially sending
them the "same" request multiple times. If they respond in a reasonable
time, the double request rate _should_ be minimal, but that of course
carries the risk that an already overloaded peer would get even more
queries.

I don't know. It's a first try at something. I need to see it work
before I write it off completely.

3 files changed, 225 insertions, 32 deletions

diff --git a/src/main.c b/src/main.c
index a18cb78..267f324 100644
--- a/src/main.c
+++ b/src/main.c
@@ -129,18 +129,6 @@ void flush_messages(int sfd, struct message* cursor, const struct message* const
 	}
 }
 
-struct lookup {
-	struct nodeid target;
-
-	struct nodeid closest[8];
-	struct addr closest_addr[8];
-	bool closest_valid[8]; // @SLOP: This could be a single word
-
-	uint64_t outstanding;
-
-	time_t wake;
-};
-
 #define OUTBOX_SIZE 32
 int main(int argc, char** argv) {
 	srand(time(NULL));
@@ -183,20 +171,20 @@ int main(int argc, char** argv) {
 	proto_begin(&dht, time(NULL), &message_cursor, outbuff+32);
 	flush_messages(dht.sfd, outbuff, message_cursor);
 
-	/* struct lookup lookup; */
-	/* // Init the lookup */
-	/* { */
-	/* 	lookup.wake = 0; */
-	/* 	lookup.target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}}; */
-
-	/* 	struct entry* entry[8]; */
-	/* 	int found = routing_closest(&lookup.target, 8, entry); */
-	/* 	for(size_t i = 0; i < found; i++) { */
-	/* 		lookup.closest[i] = entry[i]->id; */
-	/* 		lookup.closest_addr[i] = entry[i]->addr; */
-	/* 		lookup.closest_valid[i] = true; */
-	/* 	} */
-	/* } */
+	// Init the lookup
+	struct lookup *lookup = &dht.lookup;
+	{
+		lookup->timeout = 0;
+		lookup->target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}};
+
+		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;
+		}
+	}
 
 #define RECV_BUFF_SIZE 4096
 	char buff_storage[RECV_BUFF_SIZE+1];
diff --git a/src/proto.c b/src/proto.c
index c1cc392..d5d377c 100644
--- a/src/proto.c
+++ b/src/proto.c
@@ -157,17 +157,14 @@ bool find_req(struct dht* dht, uint32_t transId, uint16_t* reqId) {
 	return dht->reqalloc[transId];
 }
 
-struct msgbuff {
-	struct message** messages;
-	const struct message* const messages_end;
-};
-
 #define PROTO_EDISC 1
 #define PROTO_ENOREQ 2
 
 PROCESS_REPONSE(getclient_response);
 PROCESS_TIMEOUT(getclient_timeout);
 
+PROCESS_REPONSE(lookup_response);
+
 // Number of nodeid bits
 #define IDBITS 160
 #if IDBITS > RAND_MAX
@@ -205,6 +202,37 @@ int write_find_node(char* buff, size_t* buff_len, struct nodeid* self, struct no
 	return 0;
 }
 
+int send_lookup(struct dht* dht, struct nodeid* target, time_t now, 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;
+	}
+
+	memcpy(&message->dest, dest_addr, dest_len);
+	message->dest_len = dest_len;
+
+	dht->requestdata[reqId].cont.lookup = &dht->lookup;
+
+	dht->requestdata[reqId].fun = &lookup_response;
+	dht->requestdata[reqId].timeout = now + PROTO_TMOUT;
+	dht->requestdata[reqId].timeout_fun = NULL;
+	memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len);
+	dht->requestdata[reqId].addr_len = dest_len;
+
+	message->payload_len = sizeof(message->payload);
+	int rc = write_find_node(message->payload, &message->payload_len, &dht->self, target, reqId);
+	if(rc != 0) {
+		return rc;
+	}
+	(*msgbuff->messages)++;
+
+	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;
@@ -245,6 +273,165 @@ int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is
 	return 0;
 }
 
+PROCESS_REPONSE(lookup_response) {
+	struct benc_node stream[256];
+	struct bcursor bcursor;
+	bcur_open(&bcursor, packet, packet+packet_len, stream, 256);
+
+	if(bcursor.end - bcursor.readhead <= 0) {
+		fatal("Response too short");
+	}
+
+	struct nodeid id;
+	uint8_t nodes_len;
+	struct nodeid nodes[8];
+	struct in_addr ips[8];
+	uint16_t ports[8];
+
+	// Read the payload
+	{
+		// Check that we have a dict
+		if(bcursor.readhead->type != BNT_DICT) {
+			fatal("Response is not a dict");
+		}
+		bcur_next(&bcursor, 1);
+
+		bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING}, (const char*[]){"r"}, (const size_t[]){1}, 1);
+		// Skip the key
+		bcur_next(&bcursor, 1);
+
+		if(bcursor.readhead->type != BNT_DICT) {
+			fatal("Wrong value type for response");
+		}
+
+		// Skip the dict element
+		bcur_next(&bcursor, 1);
+
+		uint8_t parts = 0;
+		while(bcursor.readhead->type != BNT_END) {
+			switch(bcur_find_key(&bcursor, (const enum benc_nodetype[]){BNT_STRING, BNT_STRING}, (const char*[]){"nodes", "id"}, (const size_t[]){5, 2}, 2)) {
+				case 0:
+					// Skip the key
+					bcur_next(&bcursor, 1);
+
+					if(bcursor.readhead->type != BNT_STRING) {
+						fatal("Nodes must be a string");
+					}
+
+					if((bcursor.readhead->size % 26) != 0) {
+						fatal("Nodes string value must be a multiple of 26");
+					}
+
+					nodes_len = MIN(bcursor.readhead->size/26, 8);
+					for(int i = 0; i < nodes_len; i++) {
+						memcpy(nodes+i, bcursor.readhead->loc+(26*i), 20);
+						memcpy(ips+i, bcursor.readhead->loc+(26*i)+20, 4);
+						memcpy(ports+i, bcursor.readhead->loc+(26*i)+24, 2);
+					}
+
+					parts++;
+
+					// Skip the value
+					bcur_next(&bcursor, 1);
+					break;
+				case 1:
+					// Skip the key
+					bcur_next(&bcursor, 1);
+
+					if(bcursor.readhead->type != BNT_STRING) {
+						fatal("Wrong value type for response");
+					}
+
+					if(bcursor.readhead->size != 20) {
+						fatal("remote node id was not 20 bytes long");
+					}
+
+					memcpy(&id, bcursor.readhead->loc, 20);
+
+					parts++;
+
+					// Skip the value
+					bcur_next(&bcursor, 1);
+					break;
+				case -BENC_EBADP:
+					err("Bad Dictionary, Discard packet");
+					return PROTO_EDISC;
+			}
+		}
+
+		if(parts < 2) {
+			err("Response didn't contain nodes and id");
+			return PROTO_EDISC;
+		}
+	}
+
+	uint8_t my_score = prefix(&id, &cont->lookup->target);
+
+	// We need to find the best match that we are still better than
+	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
+			break;
+		}
+
+		uint8_t their_score = prefix(&cont->lookup->closest[i], &cont->lookup->target);
+		if(their_score > match_score && my_score > their_score) {
+			match_score = their_score;
+			match_i = i;
+		}
+	}
+
+	if(match_score != 0) {
+		cont->lookup->closest[match_i] = id;
+
+		struct sockaddr_in* ipv4 = (struct sockaddr_in*)remote;
+		cont->lookup->closest_addr[match_i].ip = ipv4->sin_addr.s_addr;
+		cont->lookup->closest_addr[match_i].port = ipv4->sin_port;
+	} else {
+		dbg("Discarding response from node behind the frontier");
+	}
+
+	uint8_t worst_match = UINT8_MAX;
+	for(size_t i = 0; i < 8; i++) {
+		if(cont->lookup->closest_addr[i].port == 0) {
+			worst_match = 0;
+			break;
+		}
+
+		worst_match = MIN(worst_match, prefix(&cont->lookup->closest[i], &cont->lookup->target));
+	}
+
+	// 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;
+
+		// @ROBUST: Some nodes report a bunch of nodes in the same ip. Maybe we
+		// could check for that here
+
+		struct sockaddr_in dest = {
+			.sin_family = AF_INET,
+			.sin_addr = ips[i],
+			.sin_port = ports[i],
+		};
+
+		int rc = send_lookup(dht, &cont->lookup->target, now, (struct sockaddr*)&dest, sizeof(struct sockaddr_in), msgbuff);
+		if(rc == PROTO_ENOREQ) {
+			return rc;
+		} else if(rc != 0) {
+			fatal("failed %d", rc);
+		}
+
+		cont->lookup->timeout = now;
+	}
+
+	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;
diff --git a/src/proto.h b/src/proto.h
index 0336c0c..0e1bcbc 100644
--- a/src/proto.h
+++ b/src/proto.h
@@ -21,6 +21,15 @@ struct tokens {
 	size_t head;
 };
 
+struct lookup {
+	struct nodeid target;
+
+	struct nodeid closest[8];
+	struct addr closest_addr[8];
+
+	time_t timeout;
+};
+
 void token_create(struct tokens* tokens, time_t now, struct addr* remote, char* token);
 int token_validate(struct tokens* tokens, time_t now, struct addr* remote, char* token);
 
@@ -40,10 +49,15 @@ struct ping {
 
 union message_cont {
 	struct ping ping;
+	struct lookup *lookup;
 };
 
 struct dht;
-struct msgbuff;
+struct msgbuff {
+	struct message** messages;
+	const struct message* const messages_end;
+};
+
 
 #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(resp);
@@ -67,6 +81,8 @@ struct dht {
 		union message_cont cont;
 	} requestdata[MAX_INFLIGHT];
 
+	struct lookup lookup;
+
 	time_t wake;
 	struct tokens tokens;
 };
@@ -78,6 +94,8 @@ struct message {
 	socklen_t dest_len;
 };
 
+int send_lookup(struct dht* dht, struct nodeid* target, time_t now, const struct sockaddr* dest_addr, socklen_t dest_len, struct msgbuff* msgbuff);
+
 void proto_begin(struct dht* dht, time_t now, struct message** output, const struct message* const output_end);
 int proto_run(struct dht* dht, char* buffer, size_t buffer_len, struct sockaddr_in* remote, socklen_t remote_len, time_t now, struct message** output, const struct message* const output_end);
 void proto_end(struct dht* dht);