Refactor lookup init and add routing table metric

I think it would be nice to see the detailed routing table occupancy.
I've added a metric that exposes the current occupancy of each bucket.
That should provide me some insight into the current distribution of
nodes and if we need to find a way to keep the closer buckets alive.

7 files changed, 87 insertions, 26 deletions

diff --git a/src/main.c b/src/main.c
index 7e5cf5b..e17638f 100644
--- a/src/main.c
+++ b/src/main.c
@@ -161,6 +161,7 @@ int main(int argc, char** argv) {
 		dht.self = myID;
 	}
 
+	routing_update_metrics();
 	metric_init();
 
 	size_t outLen;
@@ -172,32 +173,8 @@ int main(int argc, char** argv) {
 	flush_messages(dht.sfd, outbuff, message_cursor);
 
 	// Init the lookup
-	struct lookup *lookup = &dht.lookup;
-	{
-		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);
-		for(size_t i = 0; i < found; i++) {
-			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);
-	}
+	dht.lookup.target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}};
+	dht.lookup.state = OP_EMPTY;
 
 #define RECV_BUFF_SIZE 4096
 	char buff_storage[RECV_BUFF_SIZE+1];
diff --git a/src/metrics.c b/src/metrics.c
index 115b782..cb5fe1c 100644
--- a/src/metrics.c
+++ b/src/metrics.c
@@ -18,6 +18,8 @@ prom_gauge_t *activeNodes = NULL;
 prom_gauge_t *requestsInFlight = NULL;
 prom_counter_t *retries = NULL;
 prom_counter_t *keepalive_count = NULL;
+prom_counter_t *lookup_count = NULL;
+prom_gauge_t *routing_table_occupied = NULL;
 
 enum MHD_Result promhttp_handler(
 	void *cls,
@@ -202,6 +204,24 @@ void metric_init() {
 			NULL
 		)
 	);
+
+	lookup_count = prom_collector_registry_must_register_metric(
+		prom_counter_new(
+			"dht_lookup_count",
+			"Number of lookup commands issued",
+			0,
+			NULL
+		)
+	);
+
+	routing_table_occupied = prom_collector_registry_must_register_metric(
+		prom_gauge_new(
+			"dht_routing_occupancy",
+			"Number of nodes in each routing bucket",
+			1,
+			(const char *[]) {"dist"}
+		)
+	);
 }
 
 void metric_end() {
diff --git a/src/metrics.h b/src/metrics.h
index f129ce7..098c20b 100644
--- a/src/metrics.h
+++ b/src/metrics.h
@@ -22,6 +22,9 @@ extern prom_counter_t *retries;
 extern prom_counter_t *keepalive_count;
 
 extern prom_counter_t *requestsProcessed;
+extern prom_counter_t *lookup_count;
+
+extern prom_gauge_t *routing_table_occupied;
 
 void metric_init();
 void metric_end();
diff --git a/src/proto.c b/src/proto.c
index 2adc52b..a0f6e94 100644
--- a/src/proto.c
+++ b/src/proto.c
@@ -282,6 +282,8 @@ PROCESS_REPONSE(lookup_response) {
 		fatal("Response too short");
 	}
 
+	assert(cont->lookup->state == OP_ACTIVE);
+
 	struct nodeid id;
 	uint8_t nodes_len;
 	struct nodeid nodes[8];
@@ -586,6 +588,7 @@ PROCESS_REPONSE(getclient_response) {
 			entry->addr.ip = ipv4->sin_addr.s_addr;
 			entry->addr.port = ipv4->sin_port;
 			entry->expire = now + PROTO_UNCTM;
+			routing_update_metrics();
 		} else {
 			dbg("We are no longer interested");
 		}
@@ -901,6 +904,33 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
 			routing_oldest(&oldest);
 		}
 
+		if(dht->lookup.state == OP_PENDING) {
+			dht->lookup.timeout = now + 120;
+
+			for(size_t i = 0; i < 8; i++) {
+				dht->lookup.closest_addr[i].port = 0;
+			}
+
+			// Issue the first round of requests
+			struct entry* entry[8];
+			int found = routing_closest(&dht->lookup.target, sizeof(entry)/sizeof(entry[0]), entry);
+			for(size_t i = 0; i < found; i++) {
+				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, &dht->lookup.target, time(NULL), (struct sockaddr*)&dest, sizeof(struct sockaddr_in), &msgbuff);
+				assert(rc == 0);
+			}
+
+			dht->lookup.state = OP_ACTIVE;
+			prom_counter_inc(lookup_count, NULL);
+		} else if(dht->lookup.state == OP_ACTIVE && dht->lookup.timeout <= now) {
+			dht->lookup.state = OP_COMPLETED;
+		}
+
 		// @HACK 0 means unitialized, only happens in tests
 		if(peer_table_size != 0) {
 			expire_hashes(now);
diff --git a/src/proto.h b/src/proto.h
index 0e1bcbc..328c9ca 100644
--- a/src/proto.h
+++ b/src/proto.h
@@ -21,7 +21,15 @@ struct tokens {
 	size_t head;
 };
 
+enum Operation {
+	OP_EMPTY,
+	OP_PENDING,
+	OP_ACTIVE,
+	OP_COMPLETED,
+};
+
 struct lookup {
+	enum Operation state;
 	struct nodeid target;
 
 	struct nodeid closest[8];
diff --git a/src/routing.c b/src/routing.c
index edffffc..7c10e16 100644
--- a/src/routing.c
+++ b/src/routing.c
@@ -8,7 +8,9 @@
 #include <string.h>
 #include <stdbool.h>
 #include <stdlib.h>
+#include <pthread.h>
 #include <arpa/inet.h>
+#include <prom_metric_sample_histogram.h>
 
 // The DHT routing table has a keyspace of 0 -- 2^160 split into buckets of 8.
 // When a bucket becomes full, we split it in half. As we further expand the
@@ -56,6 +58,24 @@ void routing_init(struct nodeid* myid) {
 	routing_flush();
 }
 
+void routing_update_metrics() {
+	char buf[4]; // Largest value is 157 + \0
+
+	struct entry *entry = pTable->table;
+	for(uint8_t i = 0; i < RT_IDBITS-RT_BBITS; i++) {
+		sprintf(buf, "%d", i);
+
+		uint8_t filled = 0;
+		for(uint8_t j = 0; j < RT_BSIZE; j++) {
+			filled += entry->set;
+			entry++;
+		}
+
+		prom_gauge_set(routing_table_occupied, filled, (const char*[]){buf});
+	}
+	dbg("Update metrics!!");
+}
+
 void routing_flush() {
 	memset(pTable->table, 0, sizeof(pTable->table));
 }
@@ -159,6 +179,7 @@ void routing_remove(struct nodeid* id) {
 
 	entry->set = false;
 	entry->expire = 0;
+    routing_update_metrics();
 }
 
 bool routing_interested(struct nodeid* id) {
@@ -295,3 +316,4 @@ void routing_status(int* filled, int* size, double* load_factor, size_t load_fac
 		load_factor[i] /= per_bucket + is_overflow;
 	}
 }
+
diff --git a/src/routing.h b/src/routing.h
index 63e7fa2..be1ae8d 100644
--- a/src/routing.h
+++ b/src/routing.h
@@ -37,6 +37,7 @@ extern struct entry *table;
 extern int table_size;
 
 void routing_init(struct nodeid* myid);
+void routing_update_metrics();
 void routing_flush();
 bool routing_interested(struct nodeid* id);
 bool routing_offer(struct nodeid* id, struct entry **dest);