diff options
| -rw-r--r-- | src/main.c | 29 | ||||
| -rw-r--r-- | src/metrics.c | 20 | ||||
| -rw-r--r-- | src/metrics.h | 3 | ||||
| -rw-r--r-- | src/proto.c | 30 | ||||
| -rw-r--r-- | src/proto.h | 8 | ||||
| -rw-r--r-- | src/routing.c | 22 | ||||
| -rw-r--r-- | src/routing.h | 1 | ||||
| -rw-r--r-- | test/proto.c | 53 |
8 files changed, 118 insertions, 48 deletions
@@ -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); diff --git a/test/proto.c b/test/proto.c index f31d16a..d876762 100644 --- a/test/proto.c +++ b/test/proto.c @@ -701,9 +701,19 @@ void test_lookup_response() { dht.self = (struct nodeid){.inner={0x42424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; routing_init(&dht.self); - - // There's no requirement that nodes we issue lookups against are in our - // routing table. We therefore don't insert anything in there. + { + // The initial round of requests for the lookup looks in the routing table. + // We could do a whole song and dance of getting stuff in there, or just + // assume that that part works. We just assume. + struct nodeid other = {.inner={0x43424242, 0x42424242, 0x42424242, 0x42424242, 0x42424242}}; + struct entry *entry; + TEST_ASSERT_TRUE(routing_offer(&other, &entry)); + entry->addr = (struct addr){ + .ip = 0xFFFFFFFF, + .port = 1, + }; + entry->expire = now + PROTO_UNCTM; + } { struct message* message_cursor = outbuff; @@ -712,34 +722,22 @@ void test_lookup_response() { } now += 10; - // There's no fixed api for issuing a lookup command. We just fill in the - // structure and issue the requests { + // We start the lookup by filling in the struct and setting the state struct lookup *lookup = &dht.lookup; - lookup->timeout = now; lookup->target = (struct nodeid){.inner={0x61616161, 0x61616161, 0x61616161, 0x61616161, 0x61616161}}; + lookup->state = OP_PENDING; - for(size_t i = 0; i < 8; i++) { - // Setting the port to 0 indicates that this slot is unallocated - lookup->closest_addr[i].port = 0; - } - } - - { - // After setting up the command, we issue a request to a node. - remote = (struct sockaddr_in){ - .sin_family = AF_INET, - .sin_addr = {0xFFFFFFFF}, - .sin_port = 1, - }; struct message* message_cursor = outbuff; - int rc = send_lookup(&dht, &dht.lookup.target, now, (struct sockaddr*)&remote, sizeof(remote), &(struct msgbuff){&message_cursor, outbuff+2}); + int rc = proto_run(&dht, NULL, 0, (struct sockaddr_in*)&remote, sizeof(remote), now, &message_cursor, outbuff+2); TEST_ASSERT_EQUAL(rc, 0); 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:11: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)); } now += 1; @@ -752,7 +750,7 @@ void test_lookup_response() { int rc = proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, sizeof(remote), now, &message_cursor, outbuff+2); TEST_ASSERT_EQUAL(rc, 0); - TEST_ASSERT_EQUAL_CHAR_ARRAY(&dht.lookup.closest[0], "CBBBBBBBBBBBBBBBBBBB", 20); + TEST_ASSERT_EQUAL_CHAR_ARRAY("CBBBBBBBBBBBBBBBBBBB", &dht.lookup.closest[0], 20); TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff+1); TEST_ASSERT_EQUAL(91, outbuff[0].payload_len); @@ -807,5 +805,16 @@ void test_lookup_response() { TEST_ASSERT_EQUAL(now+120-1, dht.lookup.timeout); } + // No other packets arrive for this lookup command and it should time out + now += 119; + + { + // 15 minutes later a timeout is fired + struct message* message_cursor = outbuff; + proto_run(&dht, NULL, 0, (struct sockaddr_in*)&remote, sizeof(remote), now, &message_cursor, outbuff+2); + + TEST_ASSERT_EQUAL(OP_COMPLETED, dht.lookup.state); + } + proto_end(&dht); } |