summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/main.c29
-rw-r--r--src/metrics.c20
-rw-r--r--src/metrics.h3
-rw-r--r--src/proto.c30
-rw-r--r--src/proto.h8
-rw-r--r--src/routing.c22
-rw-r--r--src/routing.h1
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);
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-03 22:07:59 +0000