I don't remember

1 files changed, 78 insertions, 47 deletions

diff --git a/src/proto.c b/src/proto.c
index b518f79..2191d33 100644
--- a/src/proto.c
+++ b/src/proto.c
@@ -20,6 +20,7 @@
 #include <netdb.h>
 #include <sys/types.h>
 #include <stdio.h>
+#include <arpa/inet.h>
 
 #define MAX(a, b)                 \
 	 ({                           \
@@ -35,6 +36,9 @@
 		 _a < _b ? _a : _b;       \
 	 })
 
+#define CLAMP(a, b, c)            \
+	MAX(MIN(a, c), b)
+
 
 void dbgl_id(struct nodeid* id) {
 	for(uint8_t i = 0; i < 5; i++) {
@@ -99,15 +103,31 @@ struct msgbuff {
 PROCESS_REPONSE(getclient_response);
 PROCESS_TIMEOUT(getclient_timeout);
 
+#if UINT8_MAX > RAND_MAX
+#error UINT8_MAX is larger than RAND_MAX
+#endif
 uint8_t rand_byte() {
-	int limit = RAND_MAX - (RAND_MAX % UINT8_MAX);
+	int limit = (RAND_MAX / UINT8_MAX)*UINT8_MAX;
+	int val;
+	while((val = rand()) >= limit);
+
+	return val % UINT8_MAX;
+}
+
+// Number of nodeid bits
+#define IDBITS 160
+#if IDBITS > RAND_MAX
+#error IDBITS is larger than RAND_MAX
+#endif
+uint8_t rand_bucket() {
+	int limit = (RAND_MAX / IDBITS)*IDBITS;
 	int val;
-	while((val = rand()) > limit);
+	while((val = rand()) >= limit);
 
-	return val;
+	return val % IDBITS;
 }
 
-int write_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* target, uint16_t tid) {
+int write_find_node(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid* target, uint16_t tid) {
 	char* buff_end = buff + *buff_len;
 
 	int rc = snprintf(buff, buff_end - buff, "d1:ad2:id20:");
@@ -120,9 +140,9 @@ int write_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid*
 	if(rc < 0)
 		fatal("Failed to write packet");
 	buff += rc;
-	memcpy(buff, &target, sizeof(struct nodeid));
+	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);
+	rc = snprintf(buff, buff_end - buff, "e1:q9:find_node1:t%d:%d1:y1:qe", tid == 0 ? 1 : (int)(log10(tid)+1), tid);
 	if(rc < 0)
 		fatal("Failed to write packet");
 	buff += rc;
@@ -131,6 +151,14 @@ int write_ping(char* buff, size_t* buff_len, struct nodeid* self, struct nodeid*
 	return 0;
 }
 
+struct nodeid random_node() {
+	struct nodeid target;
+	for(uint8_t *target_byte = (uint8_t*)&target; target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) {
+		*target_byte = rand_byte();
+	}
+	return target;
+}
+
 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;
@@ -140,7 +168,6 @@ int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is
 	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;
@@ -154,19 +181,14 @@ int send_ping(struct dht* dht, struct nodeid* expected, time_t now, bool node_is
 
 	dht->requestdata[reqId].fun = &getclient_response;
 	dht->requestdata[reqId].timeout = now + PROTO_TMOUT;
-	dht->requestdata[reqId].timeout_fun = getclient_timeout;
+	dht->requestdata[reqId].timeout_fun = &getclient_timeout;
 	memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len);
 	dht->requestdata[reqId].addr_len = dest_len;
 
-	// Generate a random target
-	struct nodeid target;
-	for(uint8_t *target_byte = (uint8_t*)&target; target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) {
-		*target_byte = rand_byte();
-	}
-	target = dht->self;
+	struct nodeid target = random_node();
 
-	message->payload_len = 128;
-	int rc = write_ping(message->payload, &message->payload_len, &dht->self, &target, reqId);
+	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;
 	}
@@ -184,8 +206,6 @@ PROCESS_TIMEOUT(getclient_timeout) {
 		if(cont->ping.is_new)
 			return 0;
 
-		dbg("Discarding node that didn't respond");
-
 		routing_remove(&cont->ping.remote_id);
 		return 0;
 	}
@@ -199,14 +219,10 @@ PROCESS_TIMEOUT(getclient_timeout) {
 	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; target_byte < ((uint8_t*)&target)+sizeof(target); target_byte++) {
-		*target_byte = rand_byte();
-	}
+	struct nodeid target = random_node();
 
-	message->payload_len = 128;
-	int rc = write_ping(message->payload, &message->payload_len, &dht->self, &target, reqId);
+	message->payload_len = sizeof(message->payload);
+	int rc = write_find_node(message->payload, &message->payload_len, &dht->self, &target, reqId);
 	if(rc != 0) {
 		fatal("Can't create ping");
 	}
@@ -267,7 +283,6 @@ PROCESS_REPONSE(getclient_response) {
 					}
 
 					nodes_len = MIN(bcursor.readhead->size/26, 8);
-					dbg("Remote gave us %d new candidates", nodes_len);
 					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);
@@ -320,14 +335,18 @@ PROCESS_REPONSE(getclient_response) {
 		}
 	} else {
 		struct entry* entry = routing_get(&id);
-		assert(entry != NULL);
-		entry->expire = now + PROTO_UNCTM;
+		// @CLEANUP: Figure out why this can be null. Is the node getting
+		// removed while we are waiting for a response?
+		if(entry != NULL) {
+			entry->expire = now + PROTO_UNCTM;
+		}
 	}
 
+	uint8_t accepted = 0;
 	// 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]));
+		// @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,
@@ -336,17 +355,18 @@ PROCESS_REPONSE(getclient_response) {
 		};
 
 		if(routing_interested(&nodes[i])) {
+			accepted++;
 			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");
 		}
 	}
 
+	dbg("Node provided %d nodes. %d of them were useful", nodes_len, accepted);
+
 	return 0;
 }
 
@@ -415,7 +435,6 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 
 		// Temporary null terminate the string to parse the number without a copy
 		char* end;
-		dbg("Transaction %s", transaction);
 		transaction_number = strtol(transaction, &end, 10);
 
 		if(end != transaction+transaction_len) {
@@ -428,7 +447,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 			err("DISCARD: unknown transaction id %d", transaction_number);
 			return 0;
 		}
-		dbg("Transaction id matches request %d", reqId);
+		dbg("Request %d gets a response", reqId);
 
 		if(sockaddr_cmp((struct sockaddr*)&dht->requestdata[reqId].addr, (struct sockaddr*)remote) != 0) {
 			err("DISCARD: Unexpected IP for valid transaction");
@@ -452,6 +471,7 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 			fatal("No query function in query request");
 		if(transaction == NULL)
 			fatal("No transaction in request");
+		assert(transaction_len <= 16);
 
 		assert(strlen(query) == query_len);
 
@@ -472,8 +492,10 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 			fatal("No space for response");
 		cursor += rc;
 
-		rc = handle_request(&dht->self, query, packet, packet_len, &cursor, end-cursor-1);
+		dbg("===== HANDLE %s ====", query);
+		rc = handle_request(&dht->self, query, (const struct sockaddr*)remote, remote_len, packet, packet_len, &cursor, end-cursor-1);
 		if(rc == QUERY_EUNK) {
+			dbg("Unknown method");
 			// @FRAGILE: @HACK: Static offsets to fiddle with already written
 			// out packet data. Acceptable because this is the uncommon error
 			// case.
@@ -503,6 +525,8 @@ int handle_packet(struct dht* dht, time_t now, enum commandType type, char* tran
 		memcpy(&message->dest, remote, remote_len);
 		message->dest_len = remote_len;
 		(*msgbuff->messages)++;
+	} else if(type == CT_ERROR) {
+		fatal("Unhandled error");
 	} else {
 		fatal("HOW");
 	}
@@ -517,6 +541,7 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 	};
 
 	if(recv_len == 0 && buff == NULL) {
+		uint8_t timedout = 0;
 		for(int i = 0; i < MAX_INFLIGHT; i++) {
 			if(!dht->reqalloc[i])
 				continue;
@@ -540,13 +565,15 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 				dht->pause = false;
 			}
 		}
+		if(timedout != 0) {
+			dbg("processed %d requests that timed out", timedout);
+		}
 
 		struct entry* oldest = NULL;
 		routing_oldest(&oldest);
 		while(oldest != NULL) {
 			if(difftime(now, oldest->expire) < 0)
 				break;
-			dbg("Node becomes uncertain");
 
 			struct sockaddr_in dest = {0};
 			dest.sin_family = AF_INET;
@@ -558,7 +585,6 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 				return 0;
 			} else if(rc != 0) {
 				fatal("NOPE %d", rc);
-				return 0;
 			}
 
 			oldest->expire = 0;
@@ -567,12 +593,10 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 
 		return 0;
 	}
-	printf("Received packet from %s:%d\n", inet_ntoa(remote->sin_addr), ntohs(remote->sin_port));
 	
 	struct bcursor bcursor;
 	struct benc_node stream[256];
 	bcur_open(&bcursor, buff, buff+recv_len, stream, 256);
-	benc_print(bcursor.readhead, bcursor.end - bcursor.readhead);
 
 	if(bcursor.readhead->type != BNT_DICT) {
 		fatal("First value is not a dict");
@@ -638,30 +662,37 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 	assert(rc == 0);
 
 	{
-		size_t allocated = 0;
+		printf("In flight |");
 		for(int i = 0; i < MAX_INFLIGHT; i++) {
 			if(dht->reqalloc[i]) {
-				allocated++;
+				printf("#");
+			} else {
+				printf(" ");
 			}
 		}
-
-		dbg("%ld/%ld requests pending", allocated, MAX_INFLIGHT);
+		printf("|\n");
 	}
 	{
 		int filled;
 		int total;
-#define LFACLEN 32
+#define LFACLEN 64
 		double load_factor[LFACLEN] = {0};
 		routing_status(&filled, &total, load_factor, LFACLEN);
 		dbg("%d/%d nodes in routing table", filled, total);
 
-#define GRAPHY 10
+#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++) {
-				if(load_factor[x] > (1.0/GRAPHY) * (GRAPHY-y)) {
+				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(" ");
+					printf("%d", (int)(cell_load*10));
 				}
 			}
 			printf("|\n");