From 69528489cddd27b4d4b9411d00b60cccc175b0e8 Mon Sep 17 00:00:00 2001 From: Jesper Jensen Date: Fri, 18 Jul 2025 09:32:16 +0200 Subject: Add a simple API for lookups --- src/api.c | 311 +++++++++++++++++++++++++++++++++++++++++++++++++++++++++++ src/api.h | 6 ++ src/base64.c | 115 ++++++++++++++++++++++ src/base64.h | 12 +++ src/benc.c | 4 +- src/main.c | 77 +++------------ src/peers.c | 2 +- src/proto.c | 20 ++-- src/proto.h | 5 + 9 files changed, 473 insertions(+), 79 deletions(-) create mode 100644 src/api.c create mode 100644 src/api.h create mode 100644 src/base64.c create mode 100644 src/base64.h (limited to 'src') diff --git a/src/api.c b/src/api.c new file mode 100644 index 0000000..ffcd60f --- /dev/null +++ b/src/api.c @@ -0,0 +1,311 @@ +#include "api.h" +#include "log.h" +#include "base64.h" +#include "metrics.h" + +#include +#include +#include +#include + +#define BUILD_BUG_ON(condition) ((void)sizeof(char[1 - 2*!!(condition)])) + +const char *lookupStateStr[OP_LEN] = { + [OP_EMPTY] = "empty", + [OP_PENDING] = "pending", + [OP_ACTIVE] = "active", + [OP_COMPLETED] = "completed", +}; + +struct api { + struct dht *dht; +}; + +enum ReqType { + REQ_PUT_LOOKUP, +}; + +enum JsonState { + JSTATE_STR, +}; + +struct request { + enum ReqType type; + + bool target_set; + struct nodeid target; + enum Operation state; +}; + +static enum MHD_Result queue_lookup_response(struct api *api, struct MHD_Connection *connection, struct MHD_Response **response, bool locked) { + enum MHD_Result ret; + + // All states are named + BUILD_BUG_ON((sizeof(lookupStateStr) / sizeof(lookupStateStr[0])) != OP_LEN); + + char buf[1024]; + char *cursor = buf; + char *buf_end = buf + sizeof(buf); + size_t len; + + if(!locked) pthread_mutex_lock(&api->dht->mutex); + len = snprintf(cursor, buf_end-cursor, "{ \"state\": \"%s\"", lookupStateStr[api->dht->lookup.state]); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + if(api->dht->lookup.state == OP_ACTIVE || api->dht->lookup.state == OP_COMPLETED || api->dht->lookup.state == OP_PENDING) { + len = snprintf(cursor, buf_end-cursor, ", \"outstanding\": %ld, \"target\": \"", api->dht->lookup.outstanding); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + len = base64_encode_inplace((void*)&api->dht->lookup.target, sizeof(struct nodeid), cursor, buf_end - cursor); + if(len < 0) fatal("base64 failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + len = snprintf(cursor, buf_end-cursor, "\""); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + } + + if(api->dht->lookup.state == OP_COMPLETED || api->dht->lookup.state == OP_ACTIVE) { + len = snprintf(cursor, buf_end-cursor, ", \"result\": ["); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + char *lead = ""; + for(size_t i = 0; i < 8; i++) { + len = snprintf(cursor, buf_end-cursor, "%s{\"id\": \"", lead); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + len = base64_encode_inplace((void*)&api->dht->lookup.closest[i], sizeof(struct nodeid), cursor, buf_end - cursor); + if(len < 0) fatal("base64 failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + len = snprintf(cursor, buf_end-cursor, "\"}"); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + lead = ", "; + } + + len = snprintf(cursor, buf_end-cursor, "]"); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + } + + + len = snprintf(cursor, buf_end-cursor, " }"); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + pthread_mutex_unlock(&api->dht->mutex); + + *response = MHD_create_response_from_buffer_copy(cursor - buf, (void *)buf); + ret = MHD_add_response_header(*response, "Content-Type", "application/json"); + if(ret != MHD_YES) return ret; + + ret = MHD_queue_response(connection, MHD_HTTP_OK, *response); + return ret; +} + +static enum MHD_Result handler( + void *cls, + struct MHD_Connection *connection, + const char *url, + const char *method, + const char *version, + const char *upload_data, + size_t *upload_data_size, + void **con_cls +) { + struct api *api = (struct api*)cls; + struct request * conn = *con_cls; + enum MHD_Result ret = MHD_NO; + struct MHD_Response *response = NULL; + + if(conn != NULL) { + if(*upload_data_size > 0) { + const char* cursor = upload_data; + + if(memcmp(cursor, "{", 1) != 0) return MHD_NO; + cursor += 1; + + while(true) { + if(memcmp(cursor, "\"target\": \"", 11) == 0) { + cursor += 11; + + conn->target_set = true; + size_t read = base64_decode_incr((unsigned char*)cursor, *upload_data_size - (cursor - upload_data), (void*)&conn->target, sizeof(struct nodeid)); + if(read < 0) return MHD_NO; + // @CLEANUP This isn't necessarily correct since we also accept + // base64's with invalid final padding. Hopefully that will just + // lead to a bad error message to the user, but we have to look at + // that more carefully when this code is done. + cursor += 4 * ((read + 2) / 3); + + if(memcmp(cursor, "\"", 1) != 0) return MHD_NO; + cursor += 1; + } else if(memcmp(cursor, "\"state\": \"", 10) == 0) { + cursor += 10; + + for(size_t i = 0; i < OP_LEN; i++) { + size_t lookupLen = strlen(lookupStateStr[i]); + if(memcmp(cursor, lookupStateStr[i], lookupLen) == 0) { + cursor += lookupLen; + conn->state = i; + break; + } + } + + if(memcmp(cursor, "\"", 1) != 0) return MHD_NO; + cursor += 1; + } else { + return MHD_NO; + } + + if(memcmp(cursor, ", ", 2) != 0) break; + cursor += 2; + } + + if(memcmp(cursor, "}", 1) != 0) return MHD_NO; + cursor += 1; + + *upload_data_size = (*upload_data_size - (cursor - upload_data)); + + return MHD_YES; + } else { + + pthread_mutex_lock(&api->dht->mutex); + + if(conn->state == OP_PENDING) { + if(!conn->target_set) { + pthread_mutex_unlock(&api->dht->mutex); + return MHD_NO; + } + + if(api->dht->lookup.state != OP_EMPTY) { + pthread_mutex_unlock(&api->dht->mutex); + // @COMPL We should return some nice error message to the user + // here about how they lost a race + return MHD_NO; + } + + // Start a lookup + memcpy(&api->dht->lookup.target, &conn->target, sizeof(struct nodeid)); + api->dht->lookup.state = OP_PENDING; + prom_gauge_set(lookup_state, api->dht->lookup.state, NULL); + } else if(conn->state == OP_EMPTY) { + if(conn->target_set) { + pthread_mutex_unlock(&api->dht->mutex); + return MHD_NO; + } + + if(api->dht->lookup.state != OP_COMPLETED) { + pthread_mutex_unlock(&api->dht->mutex); + // @COMPL We should return some nice error message to the user + // here about how they lost a race + return MHD_NO; + } + + api->dht->lookup.state = OP_EMPTY; + prom_gauge_set(lookup_state, api->dht->lookup.state, NULL); + } else { + pthread_mutex_unlock(&api->dht->mutex); + return MHD_NO; + } + + ret = queue_lookup_response(api, connection, &response, true); + goto end; + } + } + + if(strcmp(url, "/") == 0) { + if(strcmp(method, "GET") != 0) { + response = MHD_create_response_from_buffer_static(0, NULL); + ret = MHD_queue_response(connection, MHD_HTTP_METHOD_NOT_ALLOWED, response); + goto end; + } + + char buf[1024]; + char *cursor = buf; + char *buf_end = buf + sizeof(buf); + size_t len; + + pthread_mutex_lock(&api->dht->mutex); + len = snprintf(cursor, buf_end-cursor, "{ \"id\": \""); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer too short"); + + len = base64_encode_inplace((void*)&api->dht->self, sizeof(struct nodeid), cursor, buf_end - cursor); + if(len < 0) fatal("base64 failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer to short"); + + len = snprintf(cursor, buf_end-cursor, "\" }"); + if(len < 0) fatal("printf failed\n"); + cursor += len; + if(cursor >= buf_end) fatal("buffer too short"); + pthread_mutex_unlock(&api->dht->mutex); + + response = MHD_create_response_from_buffer_copy(cursor - buf, (void *)buf); + ret = MHD_add_response_header(response, "Content-Type", "application/json"); + if(ret != MHD_YES) goto end; + ret = MHD_queue_response(connection, MHD_HTTP_OK, response); + goto end; + } + + if(strcmp(url, "/lookup") == 0) { + if(strcmp(method, "GET") == 0) { + ret = queue_lookup_response(api, connection, &response, false); + goto end; + } else if(strcmp(method, "PUT") == 0) { + conn = calloc(1, sizeof(struct request)); + + conn->type = REQ_PUT_LOOKUP; + + *con_cls = conn; + return MHD_YES; + } + + response = MHD_create_response_from_buffer_static(0, NULL); + ret = MHD_queue_response(connection, MHD_HTTP_METHOD_NOT_ALLOWED, response); + goto end; + } + + response = MHD_create_response_from_buffer_static(0, NULL); + ret = MHD_queue_response(connection, MHD_HTTP_BAD_REQUEST, response); +end: + assert(response == NULL || ret == MHD_YES); + if(response != NULL) MHD_destroy_response(response); + return ret; +} + +#define PORT 6982 +static struct MHD_Daemon *mDaemon; +static struct api api; + +void api_init(struct dht *dht) { + api.dht = dht; + + mDaemon = MHD_start_daemon(MHD_USE_THREAD_PER_CONNECTION, PORT, NULL, NULL, &handler, &api, MHD_OPTION_END); + if(mDaemon == NULL) { + fatal("Failed to start http server"); + } + dbg("API server started on port %d", PORT); +} + +void api_end() { + MHD_stop_daemon(mDaemon); + dbg("API server stopped"); +} diff --git a/src/api.h b/src/api.h new file mode 100644 index 0000000..953b7a2 --- /dev/null +++ b/src/api.h @@ -0,0 +1,6 @@ +#pragma once + +#include "proto.h" + +void api_init(struct dht *dht); +void api_end(); diff --git a/src/base64.c b/src/base64.c new file mode 100644 index 0000000..b8625c3 --- /dev/null +++ b/src/base64.c @@ -0,0 +1,115 @@ +#include "base64.h" +#include "log.h" + +#include + +static char encoding_table[] = { + 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', + 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', + 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', + 'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f', + 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', + 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', + 'w', 'x', 'y', 'z', '0', '1', '2', '3', + '4', '5', '6', '7', '8', '9', '+', '/' +}; +static int mod_table[] = {0, 2, 1}; +ssize_t base64_encode_inplace(const uint8_t *data, size_t data_len, char* buf, size_t buf_len) { + size_t output_len = 4 * ((data_len + 2) / 3); + + if(output_len > buf_len) return output_len; + + for (int i = 0, j = 0; i < data_len;) { + uint32_t octet_a = i < data_len ? (unsigned char)data[i++] : 0; + uint32_t octet_b = i < data_len ? (unsigned char)data[i++] : 0; + uint32_t octet_c = i < data_len ? (unsigned char)data[i++] : 0; + + uint32_t triple = (octet_a << 0x10) + (octet_b << 0x08) + octet_c; + + buf[j++] = encoding_table[(triple >> 3 * 6) & 0x3F]; + buf[j++] = encoding_table[(triple >> 2 * 6) & 0x3F]; + buf[j++] = encoding_table[(triple >> 1 * 6) & 0x3F]; + buf[j++] = encoding_table[(triple >> 0 * 6) & 0x3F]; + } + + for (int i = 0; i < mod_table[data_len % 3]; i++) { + buf[output_len - 1 - i] = '='; + } + + buf[output_len] = '\0'; + + return output_len; +} +// @CLEANUP This really should exist, but I've kept it around as an adapter for +// the code that uses it. Remove it at some point +char *base64_encode(const unsigned char *data, size_t input_length, size_t *output_length) { + *output_length = base64_encode_inplace(data, input_length, NULL, 0); + + char *encoded_data = malloc(*output_length + 1); + assert(encoded_data != NULL); + + base64_encode_inplace(data, input_length, encoded_data, *output_length+1); + return encoded_data; +} + +static int8_t decoding_table[] = { + 62, -1, -1, -1, 63, 52, 53, 54, + 55, 56, 57, 58, 59, 60, 61, -1, + -1, -1, -1, -1, -1, -1, 0, 1, + 2, 3, 4, 5, 6, 7, 8, 9, + 10, 11, 12, 13, 14, 15, 16, 17, + 18, 19, 20, 21, 22, 23, 24, 25, + -1, -1, -1, -1, -1, -1, 26, 27, + 28, 29, 30, 31, 32, 33, 34, 35, + 36, 37, 38, 39, 40, 41, 42, 43, + 44, 45, 46, 47, 48, 49, 50, 51 +}; + +ssize_t base64_decode_incr(const unsigned char *data, size_t data_len, uint8_t *buf, size_t buf_len) { + size_t i = 0; + size_t j = 0; + uint8_t runoff = 0; + while(runoff == 0 && i < data_len) { + uint32_t triple = 0; + for(uint8_t k = 0; k < 4; k++) { + int8_t sextent = 0; + char cp = data[i]; + if(cp < 43 || cp > 122) { + // NULL byte string terminator ends here + return k == 0 ? j : -1; + } else if(cp == '=' && k < 2) { + return -1; + } else if(cp == '=') { + sextent = 0; + runoff = 4-k; + break; + } else { + sextent = decoding_table[cp - 43]; + if(sextent < 0) return k == 0 ? j : -1; + i++; + } + + triple |= (sextent << (3-k) * 6); + } + + if (j+runoff >= buf_len) return -1; + if(runoff <= 2) buf[j++] = (triple >> 2 * 8) & 0xFF; + if(runoff <= 1) buf[j++] = (triple >> 1 * 8) & 0xFF; + if(runoff <= 0) buf[j++] = (triple >> 0 * 8) & 0xFF; + } + + return j; +} + +ssize_t base64_decode(const unsigned char *data, size_t data_len, uint8_t *buf, size_t buf_len) { + if (data_len % 4 != 0) return -1; + + size_t dest_len = data_len / 4 * 3; + if (data[data_len - 1] == '=') dest_len--; + if (data[data_len - 2] == '=') dest_len--; + + if(buf_len < dest_len) return -1; + + return base64_decode_incr(data, data_len, buf, buf_len); +} + diff --git a/src/base64.h b/src/base64.h new file mode 100644 index 0000000..6e52b87 --- /dev/null +++ b/src/base64.h @@ -0,0 +1,12 @@ +#pragma once + +#include +#include +#include +#include + +ssize_t base64_encode_inplace(const uint8_t *data, size_t data_len, char* buf, size_t buf_len); +char *base64_encode(const unsigned char *data, size_t input_length, size_t *output_length); + +ssize_t base64_decode_incr(const unsigned char *data, size_t data_len, uint8_t *buf, size_t buf_len); +ssize_t base64_decode(const unsigned char *data, size_t data_len, uint8_t *buf, size_t buf_len); diff --git a/src/benc.c b/src/benc.c index c2c2782..9842dd9 100644 --- a/src/benc.c +++ b/src/benc.c @@ -56,7 +56,7 @@ void benc_print(const struct benc_node* stream, size_t stream_len) { switch(cursor->type) { case BNT_INT: indent(depth); - printf("INT %.*s\n", cursor->size, cursor->loc); + printf("INT %.*s\n", (int)cursor->size, cursor->loc); break; case BNT_STRING: indent(depth); @@ -69,7 +69,7 @@ void benc_print(const struct benc_node* stream, size_t stream_len) { } printf("STR "); if(allprint) { - printf("%.*s", cursor->size, cursor->loc); + printf("%.*s", (int)cursor->size, cursor->loc); } else { for (const unsigned char* c = (const unsigned char*)cursor->loc; c < (unsigned char*)(cursor->loc + cursor->size); c++) { printf("\\x%02X", *c); diff --git a/src/main.c b/src/main.c index 02f39b1..bb1a67f 100644 --- a/src/main.c +++ b/src/main.c @@ -2,6 +2,8 @@ #include "peers.h" #include "log.h" #include "metrics.h" +#include "base64.h" +#include "api.h" #include #include @@ -12,47 +14,6 @@ #include #include -static char encoding_table[] = { - 'A', 'B', 'C', 'D', 'E', 'F', 'G', 'H', - 'I', 'J', 'K', 'L', 'M', 'N', 'O', 'P', - 'Q', 'R', 'S', 'T', 'U', 'V', 'W', 'X', - 'Y', 'Z', 'a', 'b', 'c', 'd', 'e', 'f', - 'g', 'h', 'i', 'j', 'k', 'l', 'm', 'n', - 'o', 'p', 'q', 'r', 's', 't', 'u', 'v', - 'w', 'x', 'y', 'z', '0', '1', '2', '3', - '4', '5', '6', '7', '8', '9', '+', '/' -}; -static int mod_table[] = {0, 2, 1}; - - -char *base64_encode(const unsigned char *data, size_t input_length, size_t *output_length) { - *output_length = 4 * ((input_length + 2) / 3); - - char *encoded_data = malloc(*output_length + 1); - assert(encoded_data != NULL); - - for (int i = 0, j = 0; i < input_length;) { - uint32_t octet_a = i < input_length ? (unsigned char)data[i++] : 0; - uint32_t octet_b = i < input_length ? (unsigned char)data[i++] : 0; - uint32_t octet_c = i < input_length ? (unsigned char)data[i++] : 0; - - uint32_t triple = (octet_a << 0x10) + (octet_b << 0x08) + octet_c; - - encoded_data[j++] = encoding_table[(triple >> 3 * 6) & 0x3F]; - encoded_data[j++] = encoding_table[(triple >> 2 * 6) & 0x3F]; - encoded_data[j++] = encoding_table[(triple >> 1 * 6) & 0x3F]; - encoded_data[j++] = encoding_table[(triple >> 0 * 6) & 0x3F]; - } - - for (int i = 0; i < mod_table[input_length % 3]; i++) { - encoded_data[*output_length - 1 - i] = '='; - } - - encoded_data[*output_length] = 0; - - return encoded_data; -} - static volatile bool killed = false; void sigint_handler(int sig) { killed = true; @@ -122,6 +83,7 @@ void flush_messages(int sfd, struct message* cursor, const struct message* const //now reply the client with the same data int rc = sendto(sfd, cursor->payload, cursor->payload_len, 0, (const struct sockaddr*)&cursor->dest, cursor->dest_len); if (rc < 0) { + dbg("dest_len %d", cursor->dest_len); fatal("Failed to send message %d %m: %d", errno, cursor->dest_len); } prom_counter_add(bytesSent, cursor->payload_len, NULL); @@ -145,6 +107,11 @@ int main(int argc, char** argv) { fatal("Couldn't set signal handler"); struct dht dht = {0}; + pthread_mutexattr_t mutexattr; + pthread_mutexattr_init(&mutexattr); + pthread_mutexattr_settype(&mutexattr, PTHREAD_MUTEX_ERRORCHECK); + pthread_mutex_init(&dht.mutex, &mutexattr); + pthread_mutex_lock(&dht.mutex); { routing_init(NULL); int rc = read_config(); @@ -152,7 +119,6 @@ int main(int argc, char** argv) { for(uint16_t i = 0; i < sizeof(myID.inner_b); i++) { myID.inner_b[i] = rand(); } - /* myID = (struct nodeid){.inner={0xebe9bbf1, 0x3cdba6b3, 0x993e0c87, 0x900d5e25, 0x00000000}}; */ routing_init(&myID); allocate_hashtable(); } else { @@ -164,6 +130,7 @@ int main(int argc, char** argv) { } metric_init(); + api_init(&dht); routing_update_metrics(); peer_update_metrics(); @@ -176,12 +143,6 @@ int main(int argc, char** argv) { proto_begin(&dht, time(NULL), &message_cursor, outbuff+32); flush_messages(dht.sfd, outbuff, message_cursor); - time_t lookup_refresh = 0; - // Init the lookup - dht.lookup.target = (struct nodeid){.inner={0x19b8a941, 0x38fa0191, 0x1403fac2, 0x581000ab, 0x19583cda}}; - dht.lookup.state = OP_PENDING; // We want this to run at some point. - prom_gauge_set(lookup_state, dht.lookup.state, NULL); - #define RECV_BUFF_SIZE 4096 char buff_storage[RECV_BUFF_SIZE+1]; int rc = 0; @@ -190,10 +151,6 @@ int main(int argc, char** argv) { bool timedout = false; time_t next = dht.wake; - if(lookup_refresh != 0 && difftime(lookup_refresh, next) < 0.0) { - next = lookup_refresh; - } - if(next != 0) { time_t sleepfor = next - time(NULL); struct timeval tv = { @@ -212,7 +169,9 @@ int main(int argc, char** argv) { socklen_t remote_len = sizeof(remote); ssize_t recv_len; if(!timedout) { + pthread_mutex_unlock(&dht.mutex); recv_len = recvfrom(dht.sfd, buff, RECV_BUFF_SIZE, 0, (struct sockaddr *)&remote, &remote_len); + pthread_mutex_lock(&dht.mutex); if(recv_len == -1) { // This is really strange. The man pages say we should be getting // an ETIMEDOUT here, but instead linux gives us this. @@ -243,19 +202,6 @@ int main(int argc, char** argv) { rc = proto_run(&dht, buff, recv_len, (struct sockaddr_in*)&remote, remote_len, now, &message_cursor, outbuff+OUTBOX_SIZE); flush_messages(dht.sfd, outbuff, message_cursor); - if(dht.lookup.state == OP_COMPLETED) { - if(lookup_refresh == 0) { - dbg("Lookup completed"); - lookup_refresh = now + 3600; - } else if (difftime(lookup_refresh, now) < 0.0) { - dbg("Restart LOOKUP"); - lookup_refresh = 0; - // Restart the lookup periodically - dht.lookup.state = OP_PENDING; - prom_gauge_set(lookup_state, dht.lookup.state, NULL); - } - } - if(dht.dirtyconf) { save_config(); dht.dirtyconf = false; @@ -263,6 +209,7 @@ int main(int argc, char** argv) { } proto_end(&dht); + api_end(); metric_end(); save_config(); diff --git a/src/peers.c b/src/peers.c index 3236a9a..7d393b0 100644 --- a/src/peers.c +++ b/src/peers.c @@ -219,7 +219,7 @@ void peer_update_metrics() { for(size_t i = 0; i < peer_table_size; i++) { if(!peer_table[i].set) continue; computed_load++; - if(now < peer_table[i].last_seen) dbg("%d was seen after now? (%ld < %ld)", i, now, peer_table[i].last_seen); + if(now < peer_table[i].last_seen) dbg("%ld was seen after now? (%ld < %ld)", i, now, peer_table[i].last_seen); time_t entry_expire = peer_table[i].last_seen + HASH_TIMEOUT; next_expire = (next_expire == -1 || entry_expire < next_expire) ? entry_expire : next_expire; diff --git a/src/proto.c b/src/proto.c index bd08c23..eb68566 100644 --- a/src/proto.c +++ b/src/proto.c @@ -398,7 +398,15 @@ PROCESS_REPONSE(lookup_response) { for(size_t i = 0; i < 8; i++) { if(cont->lookup->closest_addr[i].port == 0) { match_i = i; - match_score = UINT8_MAX; // Bogus value to signal that we found something + match_score = UINT8_MAX; // Bogus value to signal that we found an empty slot + break; + } + + if(memcmp(&cont->lookup->closest[i], &id, sizeof(struct nodeid)) == 0) { + // If the nodeid is already present in the lookup, we just ignore + // it completely + match_score = 0; + match_i = i; break; } @@ -419,16 +427,6 @@ PROCESS_REPONSE(lookup_response) { dbg("Discarding response from node behind the frontier"); } - uint8_t worst_match = UINT8_MAX; - for(size_t i = 0; i < 8; i++) { - if(cont->lookup->closest_addr[i].port == 0) { - worst_match = 0; - break; - } - - worst_match = MIN(worst_match, prefix(&cont->lookup->closest[i], &cont->lookup->target)); - } - // 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++) { uint8_t candidate_score = prefix(&nodes[i], &cont->lookup->target); diff --git a/src/proto.h b/src/proto.h index 95c7d7e..aaa3b3b 100644 --- a/src/proto.h +++ b/src/proto.h @@ -4,6 +4,7 @@ #include "sha256.h" #include #include +#include #define TOKEN_ITMO 20 #define TOKEN_VTMO 60 @@ -26,6 +27,8 @@ enum Operation { OP_PENDING, OP_ACTIVE, OP_COMPLETED, + + OP_LEN, }; struct lookup { @@ -74,6 +77,8 @@ typedef PROCESS_REPONSE(resp); typedef PROCESS_TIMEOUT(tmout); struct dht { + pthread_mutex_t mutex; + struct nodeid self; int sfd; -- cgit v1.2.3