summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat
-rw-r--r--src/main.c21
-rw-r--r--src/proto.c74
-rw-r--r--test/proto.c16
3 files changed, 47 insertions, 64 deletions
diff --git a/src/main.c b/src/main.c
index 267f324..7e5cf5b 100644
--- a/src/main.c
+++ b/src/main.c
@@ -174,16 +174,29 @@ int main(int argc, char** argv) {
// Init the lookup
struct lookup *lookup = &dht.lookup;
{
- lookup->timeout = 0;
+ 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);
- assert(found == 8);
for(size_t i = 0; i < found; i++) {
- lookup->closest[i] = entry[i]->id;
- lookup->closest_addr[i] = entry[i]->addr;
+ 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);
}
#define RECV_BUFF_SIZE 4096
diff --git a/src/proto.c b/src/proto.c
index d5d377c..2adc52b 100644
--- a/src/proto.c
+++ b/src/proto.c
@@ -218,7 +218,7 @@ int send_lookup(struct dht* dht, struct nodeid* target, time_t now, const struct
dht->requestdata[reqId].cont.lookup = &dht->lookup;
dht->requestdata[reqId].fun = &lookup_response;
- dht->requestdata[reqId].timeout = now + PROTO_TMOUT;
+ dht->requestdata[reqId].timeout = 0;
dht->requestdata[reqId].timeout_fun = NULL;
memcpy(&dht->requestdata[reqId].addr, dest_addr, dest_len);
dht->requestdata[reqId].addr_len = dest_len;
@@ -371,7 +371,6 @@ PROCESS_REPONSE(lookup_response) {
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
@@ -407,8 +406,21 @@ PROCESS_REPONSE(lookup_response) {
// 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;
+ uint8_t candidate_score = prefix(&nodes[i], &cont->lookup->target);
+
+ bool better_than_any = false;
+ bool already_matched = false;
+ for(uint8_t j = 0; j < 8; j++) {
+ if(candidate_score > prefix(&cont->lookup->closest[j], &cont->lookup->target))
+ better_than_any = true;
+
+ if(prefix(&nodes[i], &cont->lookup->closest[j]) == 160)
+ already_matched = true;
+ }
+
+ // Don't fan out to anything that is a worse match than or already
+ // included in our current frontier
+ if(!better_than_any || already_matched) continue;
// @ROBUST: Some nodes report a bunch of nodes in the same ip. Maybe we
// could check for that here
@@ -426,7 +438,7 @@ PROCESS_REPONSE(lookup_response) {
fatal("failed %d", rc);
}
- cont->lookup->timeout = now;
+ cont->lookup->timeout = now + 120;
}
return 0;
@@ -822,6 +834,10 @@ static void recalulate_waketime(struct dht *dht) {
if(dht->wake == 0 || (timeout != 0 && difftime(timeout, dht->wake) < 0))
dht->wake = timeout;
}
+
+ if(dht->wake == 0 || difftime(dht->lookup.timeout, dht->wake) < 0) {
+ dht->wake = dht->lookup.timeout;
+ }
}
int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in* remote, socklen_t remote_len, time_t now, struct message** output, const struct message* const output_end) {
@@ -958,56 +974,10 @@ int proto_run(struct dht* dht, char* buff, size_t recv_len, struct sockaddr_in*
}
}
- recalulate_waketime(dht);
-
int rc = handle_packet(dht, now, type, transaction_set ? transaction : NULL, transaction_len, query_set ? query : NULL, query_len, buff, recv_len, remote, remote_len, &msgbuff);
assert(rc == 0);
- // Debug output
-
- {
- printf("In flight |");
- uint16_t inflight = 0;
- for(int i = 0; i < MAX_INFLIGHT; i++) {
- if(dht->reqalloc[i]) {
- printf("#");
- inflight++;
- } else {
- printf(" ");
- }
- }
- printf("|\n");
- prom_gauge_set(requestsInFlight, inflight, NULL);
- }
- {
- int filled;
- int total;
-#define LFACLEN 64
- double load_factor[LFACLEN] = {0};
- routing_status(&filled, &total, load_factor, LFACLEN);
- prom_gauge_set(activeNodes, filled, NULL);
- dbg("%d/%d nodes in routing table", filled, total);
-
-#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++) {
- 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("%d", (int)(cell_load*10));
- }
- }
- printf("|\n");
- }
-#undef GRAPHY
-#undef LFACLEN
- }
+ recalulate_waketime(dht);
return 0;
}
diff --git a/test/proto.c b/test/proto.c
index 1c71d6e..f31d16a 100644
--- a/test/proto.c
+++ b/test/proto.c
@@ -757,7 +757,7 @@ void test_lookup_response() {
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:21: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));
}
@@ -775,7 +775,7 @@ void test_lookup_response() {
// The first two bytes match
dht.lookup.closest[i].inner_b[0] = 'a';
dht.lookup.closest[i].inner_b[1] = 'a';
- dht.lookup.closest[i].inner_b[2] += i;
+ dht.lookup.closest[i].inner_b[2] = 'a' + i;
dht.lookup.closest_addr[i].ip = 0;
dht.lookup.closest_addr[i].port = 1;
@@ -787,9 +787,9 @@ void test_lookup_response() {
// The node now finally responds, but woops only the first byte of its
// ID matches. That's worse than the frontier and shouldn't cause any
// addtional adjustment to the frontier.
- // It also includes a new node that's also behind the frontier, we
- // don't send anything to that one
- char buff[] = "d1:y1:r1:t1:21:rd2:id20:aBBBBBBBBBBBBBBBBBBB5:nodes26:aaBBBBBBBBBBBBBBBBBB\xFF\xFF\xFF\xFF\x00\x01""ee";
+ // It also includes a new node that's better than some nodes in the
+ // current frontier, but is also already included.
+ char buff[] = "d1:y1:r1:t1:21:rd2:id20:aBBBBBBBBBBBBBBBBBBB5:nodes52:aaaBBBBBBBBBBBBBBBBB\xFF\xFF\xFF\xFF\x00\x01""aaBBBBBBBBBBBBBBBBBB\xFF\xFF\xFF\xFE\x00\x01""ee";
struct message* message_cursor = outbuff;
int rc = proto_run(&dht, buff, sizeof(buff), (struct sockaddr_in*)&remote, sizeof(remote), now, &message_cursor, outbuff+2);
@@ -797,14 +797,14 @@ void test_lookup_response() {
// We only need to check this once since we always pick the first slot
// with a given score. It's a little implementation dependant, but it
// beats having 8 asserts.
- TEST_ASSERT_EQUAL_CHAR_ARRAY(&dht.lookup.closest[0], "aaBBBBBBBBBBBBBBBBBB", 20);
+ TEST_ASSERT_EQUAL_CHAR_ARRAY(&dht.lookup.closest[0], "aaaBBBBBBBBBBBBBBBBB", 20);
- // And we didn't fan out to the new node, even though it's actually a better match than any of the ones we have
+ // And we didn't fan out to the new node since it's already part of the current frontier
TEST_ASSERT_EQUAL_PTR(message_cursor, outbuff);
// Timeout isn't updated since we didn't send anything
// @FRAGILE this has to match the step size of now
- TEST_ASSERT_EQUAL(dht.lookup.timeout, now-1);
+ TEST_ASSERT_EQUAL(now+120-1, dht.lookup.timeout);
}
proto_end(&dht);
generated by cgit v1.2.3 (git 2.25.1) at 2026-08-03 20:09:49 +0000