From 4a35c6bf09afc527d3a68d9ff3e853ff4da47cf6 Mon Sep 17 00:00:00 2001 From: moon Date: Sun, 14 Sep 2025 16:00:17 +0330 Subject: in-memory compression --- src/packet_storage.c | 400 +++++++++++++++++++++++++++-------------- src/server.c | 496 ++++++++++++++++++++++++--------------------------- 2 files changed, 496 insertions(+), 400 deletions(-) diff --git a/src/packet_storage.c b/src/packet_storage.c index 067c742..52a80d0 100644 --- a/src/packet_storage.c +++ b/src/packet_storage.c @@ -1,12 +1,12 @@ +#include "common.c" #include #include #include +#include #include +#include #include -// dependency: zlib #include -// -#include "common.c" const uint64_t max_storage = 1 << 16; const uint64_t max_authors = 1 << 14; @@ -25,13 +25,16 @@ const uint64_t max_authors = 1 << 14; uint64_t current_packet_ptr = 0; uint64_t packet_count = 0; struct data_packet_entry { - struct data_packet packet; - uint64_t next; - uint64_t prev; - uint64_t author_next; - uint64_t author_prev; - bool is_full; -}* packet_storage; + byte *compressed_payload; + uint32_t compressed_size; + uint64_t next; + uint64_t prev; + uint64_t author_next; + uint64_t author_prev; + bool is_full; +} *packet_index; +byte *packet_arena; +uint64_t packet_arena_size; struct author_entry { public_key author; @@ -39,15 +42,91 @@ struct author_entry { uint64_t next; uint64_t prev; bool is_full; -}* author_storage; +} *author_storage; pthread_mutex_t author_storage_mutex = PTHREAD_MUTEX_INITIALIZER; -// TODO: set to zero so we can avoid allocating all the requested pages (init to zero) #define invalid_index ((uint64_t) 0) +#define BLOCK_HEADER_SIZE (sizeof(uint8_t) + sizeof(uint32_t)) + +static void set_block_header(byte *block, uint8_t full, uint32_t size) { + block[ 0 ] = full; + *(uint32_t *) (block + 1) = size; +} + +static uint8_t get_block_full(byte *block) { return block[ 0 ]; } + +static uint32_t get_block_size(byte *block) { return *(uint32_t *) (block + 1); } + +byte *arena_alloc(uint32_t size) { + uint32_t payload_size = (size + 7) & ~7; + uint32_t total_size = BLOCK_HEADER_SIZE + payload_size; + + byte *curr = packet_arena; + while (curr < packet_arena + current_packet_ptr) { + if (!get_block_full(curr) && get_block_size(curr) >= payload_size) { + uint32_t curr_size = get_block_size(curr); + if (curr_size >= payload_size + BLOCK_HEADER_SIZE + 8) { + byte *new_block = curr + total_size; + set_block_header(new_block, 0, curr_size - total_size); + set_block_header(curr, 1, payload_size); + } else { + set_block_header(curr, 1, payload_size); + } + return curr + BLOCK_HEADER_SIZE; + } + curr += BLOCK_HEADER_SIZE + get_block_size(curr); + } + + if (current_packet_ptr + total_size > packet_arena_size) { + printf("arena_alloc: out of space for %u bytes\n", total_size); + return NULL; + } + byte *block = packet_arena + current_packet_ptr; + set_block_header(block, 1, payload_size); + current_packet_ptr += total_size; + return block + BLOCK_HEADER_SIZE; +} + +void arena_free(byte *ptr) { + if (!ptr || ptr < packet_arena || ptr >= packet_arena + packet_arena_size) { + printf("arena_free: invalid pointer\n"); + return; + } + + byte *block = ptr - BLOCK_HEADER_SIZE; + if (!get_block_full(block)) { + printf("arena_free: double free\n"); + return; + } + + set_block_header(block, 0, get_block_size(block)); + + byte *next = block + BLOCK_HEADER_SIZE + get_block_size(block); + if (next < packet_arena + current_packet_ptr && !get_block_full(next)) { + uint32_t new_size = get_block_size(block) + BLOCK_HEADER_SIZE + get_block_size(next); + set_block_header(block, 0, new_size); + } + + byte *curr = packet_arena; + byte *prev = NULL; + while (curr < block) { + prev = curr; + curr += BLOCK_HEADER_SIZE + get_block_size(curr); + } + if (prev && !get_block_full(prev)) { + uint32_t new_size = get_block_size(prev) + BLOCK_HEADER_SIZE + get_block_size(block); + set_block_header(prev, 0, new_size); + } + + if (block + BLOCK_HEADER_SIZE + get_block_size(block) == packet_arena + current_packet_ptr) { + current_packet_ptr = block - packet_arena; + } +} + void save_author_packet_hash(const public_key author, const hash packet_hash) { - const char* dir = "packets_by_author"; + const char *dir = "packets_by_author"; mkdir(dir, 0755); char author_hex[ 65 ]; @@ -56,7 +135,7 @@ void save_author_packet_hash(const public_key author, const hash packet_hash) { char filename[ 256 ]; snprintf(filename, sizeof(filename), "%s/%s.list", dir, author_hex); - FILE* f = fopen(filename, "ab"); + FILE *f = fopen(filename, "ab"); if (!f) return; fwrite(packet_hash, 32, 1, f); fputc('\n', f); @@ -71,7 +150,7 @@ uint64_t get_author_entry(const public_key auth_key, bool create) { pthread_mutex_lock(&author_storage_mutex); - struct author_entry* current = &author_storage[ cur_index ]; + struct author_entry *current = &author_storage[ cur_index ]; while (true) { if (current->is_full && memcmp(current->author, auth_key, 32) == 0) { pthread_mutex_unlock(&author_storage_mutex); @@ -100,7 +179,7 @@ uint64_t get_author_entry(const public_key auth_key, bool create) { uint64_t probes = 0; uint64_t lp_idx = (cur_index + 1) % max_authors; while (probes < max_authors) { - struct author_entry* empty = &author_storage[ lp_idx ]; + struct author_entry *empty = &author_storage[ lp_idx ]; if (!empty->is_full) { memcpy(empty->author, auth_key, 32); empty->packet_head = invalid_index; @@ -119,8 +198,8 @@ uint64_t get_author_entry(const public_key auth_key, bool create) { return invalid_index; } -struct data_packet* load_packet(FILE* file) { - if (!file) { return NULL; } +struct data_packet *load_packet(FILE *file) { + if (!file) return NULL; fseek(file, 0, SEEK_END); size_t file_size = ftell(file); @@ -135,18 +214,18 @@ struct data_packet* load_packet(FILE* file) { size_t r = fread(comp_buf, 1, file_size, file); fclose(file); - if (r != file_size) { return NULL; } + if (r != file_size) return NULL; byte buffer[ PACKET_SIZE ]; uLongf dest_len = (uLongf) PACKET_SIZE; int unres = uncompress(buffer, &dest_len, comp_buf, (uLongf) file_size); - if (unres != Z_OK || dest_len != (uLongf) PACKET_SIZE) { return NULL; } + if (unres != Z_OK || dest_len != (uLongf) PACKET_SIZE) return NULL; bool success; - struct data_packet* packet = malloc(sizeof(struct data_packet)); - if (!packet) { return NULL; } + struct data_packet *packet = malloc(sizeof(struct data_packet)); + if (!packet) return NULL; *packet = s2data_packet(PACKET_SIZE, buffer, &success); if (!success) { @@ -157,126 +236,159 @@ struct data_packet* load_packet(FILE* file) { return packet; } -uint64_t set_data_packet(struct data_packet* new_packet); - -struct data_packet_entry* get_data_packet(const hash _hash) { - uint64_t hash_val; - memcpy(&hash_val, _hash, sizeof(uint64_t)); - uint64_t cur_index = hash_val % max_storage; - - pthread_mutex_lock(&packet_storage_mutex); - - struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; - while (cur_index != invalid_index) { - if (current_entry->is_full - && memcmp(current_entry->packet.hash, _hash, sizeof(hash)) == 0) { - pthread_mutex_unlock(&packet_storage_mutex); - return current_entry; - } - - cur_index = current_entry->next; - if (cur_index != invalid_index) { current_entry = &packet_storage[ cur_index ]; } - } - - pthread_mutex_unlock(&packet_storage_mutex); - char filename[ 128 ]; - sprintf(filename, "packets/" hash_fmt ".bin", hash_ref(_hash)); - FILE* file = fopen(filename, "rb"); - - struct data_packet* packet = load_packet(file); - - if (packet) { - uint64_t add_idx = set_data_packet(packet); - if (add_idx != invalid_index) { - free(packet); - return &packet_storage[ add_idx ]; - } else { - static __thread struct data_packet_entry temp_entry; - temp_entry.packet = *packet; - free(packet); - return &temp_entry; - } +uint64_t set_data_packet(struct data_packet *new_packet) { + byte temp_buffer[ sizeof(struct data_packet) ]; + memcpy(temp_buffer, new_packet, sizeof(struct data_packet)); + byte compressed[ sizeof(struct data_packet) ]; + uLongf compressed_len = sizeof(struct data_packet); + if (compress(compressed, &compressed_len, temp_buffer, sizeof(struct data_packet)) != Z_OK) { + printf("set_data_packet: compression failed\n"); + return invalid_index; } - return NULL; -} + byte *compressed_payload = arena_alloc(compressed_len); + if (!compressed_payload) { return invalid_index; } + memcpy(compressed_payload, compressed, compressed_len); -uint64_t __set_data_packet(struct data_packet* new_packet) { pthread_mutex_lock(&packet_storage_mutex); - uint64_t hash_val; memcpy(&hash_val, new_packet->hash, sizeof(uint64_t)); - uint64_t start_index = hash_val % max_storage; - uint64_t cur_index = start_index; - - struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; + uint64_t start_index = hash_val % max_storage; + uint64_t cur_index = start_index; + struct data_packet_entry *current_entry = &packet_index[ cur_index ]; while (true) { if (current_entry->is_full - && memcmp(current_entry->packet.hash, new_packet->hash, sizeof(hash)) == 0) { - memcpy(¤t_entry->packet, new_packet, sizeof(struct data_packet)); + && memcmp( + ((struct data_packet *) current_entry->compressed_payload)->hash, + new_packet->hash, + sizeof(hash)) + == 0) { + arena_free(current_entry->compressed_payload); + current_entry->compressed_payload = compressed_payload; + current_entry->compressed_size = compressed_len; pthread_mutex_unlock(&packet_storage_mutex); + save_author_packet_hash(new_packet->contents.author, new_packet->hash); return cur_index; } - if (current_entry->next == invalid_index) break; cur_index = current_entry->next; - current_entry = &packet_storage[ cur_index ]; + current_entry = &packet_index[ cur_index ]; } if (!current_entry->is_full) { - memcpy(¤t_entry->packet, new_packet, sizeof(struct data_packet)); - current_entry->is_full = true; - current_entry->next = invalid_index; - current_entry->prev = invalid_index; + current_entry->compressed_payload = compressed_payload; + current_entry->compressed_size = compressed_len; + current_entry->is_full = true; + current_entry->next = invalid_index; + current_entry->prev = invalid_index; packet_count++; pthread_mutex_unlock(&packet_storage_mutex); + save_author_packet_hash(new_packet->contents.author, new_packet->hash); return cur_index; } uint64_t probes = 0; uint64_t linear_probe_index = (cur_index + 1) % max_storage; - while (probes < max_storage) { - struct data_packet_entry* empty_slot = &packet_storage[ linear_probe_index ]; - + struct data_packet_entry *empty_slot = &packet_index[ linear_probe_index ]; if (!empty_slot->is_full) { - memcpy(&empty_slot->packet, new_packet, sizeof(struct data_packet)); - empty_slot->is_full = true; - empty_slot->next = invalid_index; - empty_slot->prev = cur_index; - current_entry->next = linear_probe_index; + empty_slot->compressed_payload = compressed_payload; + empty_slot->compressed_size = compressed_len; + empty_slot->is_full = true; + empty_slot->next = invalid_index; + empty_slot->prev = cur_index; + current_entry->next = linear_probe_index; packet_count++; pthread_mutex_unlock(&packet_storage_mutex); + save_author_packet_hash(new_packet->contents.author, new_packet->hash); return linear_probe_index; } - linear_probe_index = (linear_probe_index + 1) % max_storage; probes++; } - // table full pthread_mutex_unlock(&packet_storage_mutex); + arena_free(compressed_payload); return invalid_index; } -uint64_t set_data_packet(struct data_packet* new_packet) { - uint64_t result = __set_data_packet(new_packet); - if (result == invalid_index) return invalid_index; +struct data_packet_entry *get_data_packet(const hash _hash) { + uint64_t hash_val; + memcpy(&hash_val, _hash, sizeof(uint64_t)); + uint64_t cur_index = hash_val % max_storage; - save_author_packet_hash(new_packet->contents.author, new_packet->hash); + pthread_mutex_lock(&packet_storage_mutex); + struct data_packet_entry *current_entry = &packet_index[ cur_index ]; + while (cur_index != invalid_index) { + if (current_entry->is_full + && memcmp( + ((struct data_packet *) current_entry->compressed_payload)->hash, + _hash, + sizeof(hash)) + == 0) { + byte decompressed[ sizeof(struct data_packet) ]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress( + decompressed, + &decompressed_len, + current_entry->compressed_payload, + current_entry->compressed_size) + != Z_OK) { + printf("get_data_packet: decompression failed\n"); + pthread_mutex_unlock(&packet_storage_mutex); + return NULL; + } + static __thread struct data_packet_entry temp_entry; + temp_entry = *current_entry; + temp_entry.compressed_payload = arena_alloc(decompressed_len); + if (!temp_entry.compressed_payload) { + pthread_mutex_unlock(&packet_storage_mutex); + return NULL; + } + memcpy(temp_entry.compressed_payload, decompressed, decompressed_len); + temp_entry.compressed_size = decompressed_len; + pthread_mutex_unlock(&packet_storage_mutex); + return &temp_entry; + } + cur_index = current_entry->next; + if (cur_index != invalid_index) current_entry = &packet_index[ cur_index ]; + } + pthread_mutex_unlock(&packet_storage_mutex); - uint64_t auth_idx = get_author_entry(new_packet->contents.author, true); - if (auth_idx != invalid_index) { - pthread_mutex_lock(&packet_storage_mutex); - uint64_t old_head = author_storage[ auth_idx ].packet_head; - packet_storage[ result ].author_next = old_head; - packet_storage[ result ].author_prev = invalid_index; - if (old_head != invalid_index) packet_storage[ old_head ].author_prev = result; - author_storage[ auth_idx ].packet_head = result; - pthread_mutex_unlock(&packet_storage_mutex); + char filename[ 128 ]; + sprintf(filename, "packets/" hash_fmt ".bin", hash_ref(_hash)); + FILE *file = fopen(filename, "rb"); + struct data_packet *packet = load_packet(file); + if (packet) { + uint64_t add_idx = set_data_packet(packet); + if (add_idx != invalid_index) { + free(packet); + return &packet_index[ add_idx ]; + } else { + static __thread struct data_packet_entry temp_entry; + byte compressed[ sizeof(struct data_packet) ]; + uLongf compressed_len = sizeof(struct data_packet); + if (compress(compressed, &compressed_len, (byte *) packet, sizeof(struct data_packet)) + != Z_OK) { + free(packet); + return NULL; + } + temp_entry.compressed_payload = arena_alloc(compressed_len); + if (!temp_entry.compressed_payload) { + free(packet); + return NULL; + } + memcpy(temp_entry.compressed_payload, compressed, compressed_len); + temp_entry.compressed_size = compressed_len; + temp_entry.is_full = true; + temp_entry.next = invalid_index; + temp_entry.prev = invalid_index; + free(packet); + return &temp_entry; + } } - return result; + return NULL; } #define overload __attribute__((overloadable)) @@ -292,14 +404,17 @@ overload void delete_data_packet(hash _hash) { uint64_t cur_index = start_index; while (cur_index != invalid_index) { - struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; + struct data_packet_entry *current_entry = &packet_index[ cur_index ]; if (current_entry->is_full - && memcmp(current_entry->packet.hash, _hash, sizeof(hash)) == 0) { + && memcmp( + ((struct data_packet *) current_entry->compressed_payload)->hash, + _hash, + sizeof(hash)) + == 0) { _delete_at_index(cur_index); pthread_mutex_unlock(&packet_storage_mutex); return; } - if (current_entry->next == invalid_index) { pthread_mutex_unlock(&packet_storage_mutex); return; @@ -316,45 +431,44 @@ overload void delete_data_packet(uint64_t index) { pthread_mutex_unlock(&packet_storage_mutex); } -// caller must already have a lock! bool _delete_at_index(uint64_t index) { if (index >= max_storage) return false; - struct data_packet_entry* current_entry = &packet_storage[ index ]; + struct data_packet_entry *current_entry = &packet_index[ index ]; if (!current_entry->is_full) return false; public_key auth_key; - memcpy(auth_key, packet_storage[ index ].packet.contents.author, 32); + memcpy( + auth_key, ((struct data_packet *) current_entry->compressed_payload)->contents.author, 32); uint64_t auth_idx = get_author_entry(auth_key, false); if (auth_idx != invalid_index) { if (author_storage[ auth_idx ].packet_head == index) { - author_storage[ auth_idx ].packet_head = packet_storage[ index ].author_next; + author_storage[ auth_idx ].packet_head = current_entry->author_next; } } - uint64_t a_prev = packet_storage[ index ].author_prev; - uint64_t a_next = packet_storage[ index ].author_next; - if (a_prev != invalid_index) packet_storage[ a_prev ].author_next = a_next; - if (a_next != invalid_index) packet_storage[ a_next ].author_prev = a_prev; + uint64_t a_prev = current_entry->author_prev; + uint64_t a_next = current_entry->author_next; + if (a_prev != invalid_index) packet_index[ a_prev ].author_next = a_next; + if (a_next != invalid_index) packet_index[ a_next ].author_prev = a_prev; if (current_entry->prev != invalid_index) { - struct data_packet_entry* prev_entry = &packet_storage[ current_entry->prev ]; - prev_entry->next = current_entry->next; + packet_index[ current_entry->prev ].next = current_entry->next; } - if (current_entry->next != invalid_index) { - struct data_packet_entry* next_entry = &packet_storage[ current_entry->next ]; - next_entry->prev = current_entry->prev; + packet_index[ current_entry->next ].prev = current_entry->prev; } - current_entry->is_full = false; - current_entry->next = invalid_index; - current_entry->prev = invalid_index; - + arena_free(current_entry->compressed_payload); + current_entry->is_full = false; + current_entry->next = invalid_index; + current_entry->prev = invalid_index; + current_entry->compressed_payload = NULL; + current_entry->compressed_size = 0; return true; } -int get_packets_for_author__cmp(const void* a, const void* b) { - uint64_t ta = ((struct data_packet*) a)->contents.created_time_ms; - uint64_t tb = ((struct data_packet*) b)->contents.created_time_ms; +int get_packets_for_author__cmp(const void *a, const void *b) { + uint64_t ta = ((struct data_packet *) a)->contents.created_time_ms; + uint64_t tb = ((struct data_packet *) b)->contents.created_time_ms; return (ta > tb) ? -1 : (ta < tb) ? 1 : 0; } @@ -362,12 +476,11 @@ uint16_t get_packets_for_author( const public_key author, uint64_t offset, uint16_t max_count, - struct data_packet* out_packets, - uint64_t* out_new_offset) { + struct data_packet *out_packets, + uint64_t *out_new_offset) { uint64_t auth_idx = get_author_entry(author, false); if (auth_idx == invalid_index) return 0; - // overfetch uint64_t fetch_factor = ((offset + (uint64_t) max_count) * 5LLU) / 2; pthread_mutex_lock(&packet_storage_mutex); @@ -375,10 +488,10 @@ uint16_t get_packets_for_author( uint16_t mem_count = 0; while (curr != invalid_index) { mem_count++; - curr = packet_storage[ curr ].author_next; + curr = packet_index[ curr ].author_next; } - struct data_packet* all_packets + struct data_packet *all_packets = malloc((mem_count + fetch_factor) * sizeof(struct data_packet)); if (!all_packets) { pthread_mutex_unlock(&packet_storage_mutex); @@ -388,8 +501,19 @@ uint16_t get_packets_for_author( curr = author_storage[ auth_idx ].packet_head; while (curr != invalid_index) { - all_packets[ all_count++ ] = packet_storage[ curr ].packet; - curr = packet_storage[ curr ].author_next; + byte decompressed[ sizeof(struct data_packet) ]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress( + decompressed, + &decompressed_len, + packet_index[ curr ].compressed_payload, + packet_index[ curr ].compressed_size) + != Z_OK) { + printf("get_packets_for_author: decompression failed\n"); + continue; + } + memcpy(&all_packets[ all_count++ ], decompressed, sizeof(struct data_packet)); + curr = packet_index[ curr ].author_next; } pthread_mutex_unlock(&packet_storage_mutex); @@ -399,7 +523,7 @@ uint16_t get_packets_for_author( char filename[ 256 ]; snprintf(filename, sizeof(filename), "packets_by_author/%s.list", author_hex); - FILE* f = fopen(filename, "rb"); + FILE *f = fopen(filename, "rb"); size_t disk_added = 0; if (f) { fseek(f, 0, SEEK_END); @@ -427,10 +551,18 @@ uint16_t get_packets_for_author( } if (dupe) continue; - struct data_packet_entry* ent = get_data_packet(phash); + struct data_packet_entry *ent = get_data_packet(phash); if (!ent) continue; - all_packets[ all_count++ ] = ent->packet; + byte decompressed[ sizeof(struct data_packet) ]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress( + decompressed, &decompressed_len, ent->compressed_payload, ent->compressed_size) + != Z_OK) { + printf("get_packets_for_author: decompression failed for disk packet\n"); + continue; + } + memcpy(&all_packets[ all_count++ ], decompressed, sizeof(struct data_packet)); disk_added++; } fclose(f); diff --git a/src/server.c b/src/server.c index 71722ba..0b7ea87 100644 --- a/src/server.c +++ b/src/server.c @@ -4,9 +4,7 @@ #include #include #include -// dependency: zlib #include -// #include "common.c" #include "config.c" #include "firehose.c" @@ -15,33 +13,31 @@ #include "plugin.c" #include "security.c" #include "sha256.c" -// uint16_t current_port = default_port; -const uint64_t max_neighbors = 1 << 10; // probably more than enough for a lot of folks +const uint64_t max_neighbors = 1 << 10; -uint64_t current_neighbor_ptr = 0; -uint64_t neighbor_count = 0; +uint64_t current_neighbor_ptr = 0; +uint64_t neighbor_count = 0; struct relay_neighbor* neighbor_storage; -bool* empty_neighbor_slots; -pthread_mutex_t neighbor_storage_mutex = PTHREAD_MUTEX_INITIALIZER; +bool* empty_neighbor_slots; +pthread_mutex_t neighbor_storage_mutex = PTHREAD_MUTEX_INITIALIZER; struct server_stats { byte major_version; byte minor_version; - byte motd[ 128 ]; - // - uint16_t retention_hours; // should default to 1w (168h) - byte leading_zeros; // i recommend 10 - bool accepts_plain; // for legal reasons + byte motd[128]; + uint16_t retention_hours; + byte leading_zeros; + bool accepts_plain; } server_stats = { - .major_version = 0, - .minor_version = 1, - .motd = "A moonpaste server.", // minecraft much? - .retention_hours = 168, // 1 week - .leading_zeros = 12, // the moon guarantee - .accepts_plain = true, // probably legal + .major_version = 0, + .minor_version = 1, + .motd = "A moonpaste server.", + .retention_hours = 168, + .leading_zeros = 12, + .accepts_plain = true, }; bool add_neighbor(struct relay_neighbor new_neighbor) { @@ -59,15 +55,15 @@ bool add_neighbor(struct relay_neighbor new_neighbor) { uint64_t add_idx = -1; for (uint64_t i = 0; i < max_neighbors; i++) { - if (empty_neighbor_slots[ i ]) { + if (empty_neighbor_slots[i]) { add_idx = i; break; } } if (add_idx != -1) { - neighbor_storage[ add_idx ] = new_neighbor; - empty_neighbor_slots[ add_idx ] = false; + neighbor_storage[add_idx] = new_neighbor; + empty_neighbor_slots[add_idx] = false; neighbor_count++; pthread_mutex_unlock(&neighbor_storage_mutex); return true; @@ -77,19 +73,8 @@ bool add_neighbor(struct relay_neighbor new_neighbor) { return false; } -#define hash_fmt \ - "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x" \ - "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x" - -#define hash_ref(_hash) \ - _hash[ 0 ], _hash[ 1 ], _hash[ 2 ], _hash[ 3 ], _hash[ 4 ], _hash[ 5 ], _hash[ 6 ], \ - _hash[ 7 ], _hash[ 8 ], _hash[ 9 ], _hash[ 10 ], _hash[ 11 ], _hash[ 12 ], _hash[ 13 ], \ - _hash[ 14 ], _hash[ 15 ], _hash[ 16 ], _hash[ 17 ], _hash[ 18 ], _hash[ 19 ], _hash[ 20 ], \ - _hash[ 21 ], _hash[ 22 ], _hash[ 23 ], _hash[ 24 ], _hash[ 25 ], _hash[ 26 ], _hash[ 27 ], \ - _hash[ 28 ], _hash[ 29 ], _hash[ 30 ], _hash[ 31 ] - void* propagate_packet(void* _packet) { - struct data_packet packet = *(struct data_packet*) _packet; + struct data_packet packet = *(struct data_packet*)_packet; if (decide_packet_propogation_func) { if (!decide_packet_propogation_func(packet)) return NULL; @@ -98,36 +83,35 @@ void* propagate_packet(void* _packet) { log(_debug, "propagating packet " hash_fmt, hash_ref(packet.hash)); size_t send_len = 1 + sizeof(struct data_packet); - byte send_data[ send_len * sizeof(byte) ]; - send_data[ 0 ] = push_packet; + byte send_data[send_len * sizeof(byte)]; + send_data[0] = push_packet; data_packet2s(packet, send_data + 1); pthread_mutex_lock(&neighbor_storage_mutex); for (uint64_t i = 0; i < neighbor_count; i++) { - struct relay_neighbor neighbor = neighbor_storage[ i ]; + struct relay_neighbor neighbor = neighbor_storage[i]; if (neighbor.active) { - byte* response_data = NULL; - size_t response_len = 0; - int result = -1; + byte* response_data = NULL; + size_t response_len = 0; + int result = -1; if (neighbor.has_ipv4) { - char ip_string[ 16 ]; + char ip_string[16]; sprintf( ip_string, "%d.%d.%d.%d", - neighbor.v4_addr[ 0 ], - neighbor.v4_addr[ 1 ], - neighbor.v4_addr[ 2 ], - neighbor.v4_addr[ 3 ]); + neighbor.v4_addr[0], + neighbor.v4_addr[1], + neighbor.v4_addr[2], + neighbor.v4_addr[3]); result = study( ip_string, neighbor.port, send_data, send_len, &response_data, &response_len); } else if (neighbor.has_ipv6) { - char ip_string[ 46 ]; - int offset = 0; + char ip_string[46]; + int offset = 0; for (int j = 0; j < 8; j++) { - uint16_t word - = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ]; + uint16_t word = (neighbor.v6_addr[j * 2] << 8) | neighbor.v6_addr[j * 2 + 1]; offset += sprintf(ip_string + offset, "%x%s", word, (j < 7) ? ":" : ""); } @@ -148,64 +132,57 @@ void* propagate_packet(void* _packet) { void save_packet(const struct data_packet packet); -#define reply(__len, ...) \ - ({ \ - *out = (byte*) __VA_ARGS__; \ - *out_len = __len; \ - }) +#define reply(__len, ...) \ + ({ *out = (byte*)__VA_ARGS__; *out_len = __len; }) void paste_server( char* ip_addr, size_t ip_size, byte* in, size_t in_len, byte** out, size_t* out_len) { - log(_info, "<%.*s> incoming request with size %li", (int) ip_size, ip_addr, in_len); + log(_info, "<%.*s> incoming request with size %li", (int)ip_size, ip_addr, in_len); if (paste_server_start_func) { paste_server_start_func(&ip_addr, &ip_size, &in, &in_len, &out, &out_len); } - // tip: if you want the server to not mandate the 8 bytes minimum, modify in_len using the - // plugin. - - if (in_len < 8) { // server mandates 8 bytes minimum. pad it if you don't need it! + if (in_len < 8) { reply(17, strdup("\x01input too small")); return; } - enum client_input input_type = in[ 0 ]; + enum client_input input_type = in[0]; switch (input_type) { case audit_server: { - log(_info, "<%.*s> request type: audit", (int) ip_size, ip_addr); + log(_info, "<%.*s> request type: audit", (int)ip_size, ip_addr); byte* output_data = calloc(1 + 237, sizeof(byte)); if (!output_data) { byte* err = calloc(1 + 14, sizeof(byte)); if (err) { - err[ 0 ] = error; + err[0] = error; memcpy(err + 1, "server OOM", 10); reply(11, err); } return; } - output_data[ 0 ] = audit_response; + output_data[0] = audit_response; - struct relay_info info = { 0 }; + struct relay_info info = {0}; pthread_mutex_lock(&neighbor_storage_mutex); pthread_mutex_lock(&packet_storage_mutex); - info.major_version = server_stats.major_version; - info.minor_version = server_stats.minor_version; + info.major_version = server_stats.major_version; + info.minor_version = server_stats.minor_version; info.capabilities.retention_hours = server_stats.retention_hours; - info.capabilities.can_store = packet_count < max_storage - 1; - info.capabilities.leading_zeros = server_stats.leading_zeros; - info.capabilities.accepts_plain = server_stats.accepts_plain; + info.capabilities.can_store = packet_count < max_storage - 1; + info.capabilities.leading_zeros = server_stats.leading_zeros; + info.capabilities.accepts_plain = server_stats.accepts_plain; - info.statistics.neighbors = neighbor_count; + info.statistics.neighbors = neighbor_count; info.statistics.held_packets = packet_count; - // TODO: for now we return the first 4 for (int i = 0; i < 4; ++i) { - info.statistics.quick_connect[ i ] = neighbor_storage[ i ]; + info.statistics.quick_connect[i] = neighbor_storage[i]; } memcpy(info.motd, server_stats.motd, 128); @@ -222,31 +199,28 @@ void paste_server( } case get_neighbors: { - byte count = in[ 1 ]; + byte count = in[1]; log(_info, "<%.*s> request type: get neighbors, count %i", - (int) ip_size, - ip_addr, - count); + (int)ip_size, ip_addr, count); - size_t output_len = 1 + (RELAY_NEIGHBOR_WIRE_LEN * count); - byte* output_data = calloc(output_len, sizeof(byte)); + size_t output_len = 1 + (RELAY_NEIGHBOR_WIRE_LEN * count); + byte* output_data = calloc(output_len, sizeof(byte)); if (!output_data) { byte* err = calloc(1 + 14, sizeof(byte)); if (err) { - err[ 0 ] = error; + err[0] = error; memcpy(err + 1, "server OOM", 10); reply(11, err); } return; } - output_data[ 0 ] = neighbors_info; + output_data[0] = neighbors_info; size_t offset = 1; for (byte i = 0; i < count; i++) { - // TODO: maybe mutex? - relay_neighbor2s(neighbor_storage[ i ], output_data + offset); + relay_neighbor2s(neighbor_storage[i], output_data + offset); offset += RELAY_NEIGHBOR_WIRE_LEN; } @@ -258,7 +232,7 @@ void paste_server( if (in_len < 1 + 32) { byte* output_data = calloc(1 + 17, sizeof(byte)); if (!output_data) return; - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "input too small", 17); reply(18, output_data); return; @@ -270,14 +244,12 @@ void paste_server( log(_info, "<%.*s> request type: get packet with hash " hash_fmt, - (int) ip_size, - ip_addr, - hash_ref(packet_hash)); + (int)ip_size, ip_addr, hash_ref(packet_hash)); if (!success) { byte* output_data = calloc(1 + 12, sizeof(byte)); if (!output_data) return; - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "invalid hash", 12); reply(13, output_data); return; @@ -287,79 +259,84 @@ void paste_server( if (entry == NULL) { byte* output_data = calloc(1 + 14, sizeof(byte)); if (!output_data) return; - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "packet not found", 14); reply(15, output_data); return; } - size_t output_len = 1 + PACKET_SIZE; - byte* output_data = calloc(output_len, sizeof(byte)); + byte decompressed[sizeof(struct data_packet)]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress(decompressed, &decompressed_len, entry->compressed_payload, + entry->compressed_size) != Z_OK) { + byte* output_data = calloc(1 + 22, sizeof(byte)); + if (!output_data) return; + output_data[0] = error; + memcpy(output_data + 1, "decompression failed", 22); + reply(23, output_data); + return; + } + + size_t output_len = 1 + PACKET_SIZE; + byte* output_data = calloc(output_len, sizeof(byte)); if (!output_data) { byte* err = calloc(1 + 14, sizeof(byte)); if (err) { - err[ 0 ] = error; + err[0] = error; memcpy(err + 1, "server OOM", 10); reply(11, err); } return; } - output_data[ 0 ] = get_packet_result; - data_packet2s(entry->packet, output_data + 1); + output_data[0] = get_packet_result; + data_packet2s(*(struct data_packet*)decompressed, output_data + 1); reply(output_len, output_data); return; } case push_packet: { - bool success; + bool success; struct data_packet packet = s2data_packet(in_len - 1, in + 1, &success); log(_info, "<%.*s> request type: push packet with hash " hash_fmt, - (int) ip_size, - ip_addr, - hash_ref(packet.hash)); + (int)ip_size, ip_addr, hash_ref(packet.hash)); - size_t output_len = 1 + 1; - byte* output_data = calloc(output_len, sizeof(byte)); + size_t output_len = 1 + 1; + byte* output_data = calloc(output_len, sizeof(byte)); if (!output_data) { return; } - output_data[ 0 ] = push_packet_result; + output_data[0] = push_packet_result; if (!success) { - log(_warn, "<%.*s> malformed data", (int) ip_size, ip_addr); - output_data[ 1 ] = false; + log(_warn, "<%.*s> malformed data", (int)ip_size, ip_addr); + output_data[1] = false; reply(output_len, output_data); return; } if (modify_packet_func) { - // things you can do here: - // - prefer certain packet authors - // - run PoW - // - increase leading zeroes - // - recompute the hash modify_packet_func(&packet); } if (!check_leading_zeros(packet.hash, server_stats.leading_zeros)) { log(_warn, "bad leads"); - output_data[ 1 ] = false; + output_data[1] = false; reply(output_len, output_data); return; } const size_t hash_input_len = 4 + CONTENTS_SIZE; - byte hash_input_buffer[ hash_input_len ]; + byte hash_input_buffer[hash_input_len]; write_u32_le(hash_input_buffer, packet.nonce); byte* contents_buffer = hash_input_buffer + 4; - contents_buffer[ 0 ] = packet.contents.encrypted ? 1 : 0; + contents_buffer[0] = packet.contents.encrypted ? 1 : 0; write_u64_le(contents_buffer + 1, packet.contents.created_time_ms); memcpy(contents_buffer + 9, packet.contents.author, 32); memcpy(contents_buffer + 41, packet.contents.author_signing_pub, 32); - contents_buffer[ 73 ] = packet.contents.recipient_count; + contents_buffer[73] = packet.contents.recipient_count; memcpy(contents_buffer + 74, packet.contents.recipients, 32 * MAX_RECIPIENTS); memcpy(contents_buffer + 74 + 32 * MAX_RECIPIENTS, packet.contents.data, 65536); memcpy( @@ -371,7 +348,7 @@ void paste_server( "bad hash " hash_fmt " vs " hash_fmt, hash_ref(calculated_hash_struct.hash), hash_ref(packet.hash)); - output_data[ 1 ] = false; + output_data[1] = false; reply(output_len, output_data); return; } @@ -379,7 +356,7 @@ void paste_server( uint64_t current_time = get_current_time_ms(); if (packet.contents.created_time_ms > current_time) { log(_warn, "packet timestamp is in the future"); - output_data[ 1 ] = false; + output_data[1] = false; reply(output_len, output_data); return; } @@ -387,14 +364,14 @@ void paste_server( if (current_time - packet.contents.created_time_ms > server_stats.retention_hours * (60 * 60 * 1000)) { log(_warn, "packet expired"); - output_data[ 1 ] = false; + output_data[1] = false; reply(output_len, output_data); return; } if (!packet.contents.encrypted && !server_stats.accepts_plain) { log(_warn, "plaintext not accepted by server"); - output_data[ 1 ] = false; + output_data[1] = false; reply(output_len, output_data); return; } @@ -402,7 +379,7 @@ void paste_server( bool sig_ok = false; if (!packet.contents.encrypted) { size_t msg_len = CONTENTS_SIZE - sizeof(signature); - sig_ok = verify_raw_signature( + sig_ok = verify_raw_signature( contents_buffer, msg_len, packet.contents.signature, @@ -420,7 +397,7 @@ void paste_server( if (!sig_ok) { log(_warn, "bad signature"); - output_data[ 1 ] = false; + output_data[1] = false; reply(output_len, output_data); return; } @@ -429,11 +406,11 @@ void paste_server( firehose_push(packet.hash, storage_index); if (storage_index != invalid_index) { - output_data[ 1 ] = true; + output_data[1] = true; save_packet(packet); - if (((float) rand() / (float) RAND_MAX) > 0.5) { + if (((float)rand() / (float)RAND_MAX) > 0.5) { struct data_packet* thread_arg = malloc(sizeof(struct data_packet)); if (thread_arg) { *thread_arg = packet; @@ -443,7 +420,7 @@ void paste_server( } } } else { - output_data[ 1 ] = false; + output_data[1] = false; } reply(output_len, output_data); @@ -453,55 +430,59 @@ void paste_server( case firehose: { if (in_len < 1 + 9) { byte* output_data = calloc(1 + 17, sizeof(byte)); - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "input too small", 17); reply(18, output_data); return; } struct firehose_request req; - bool success; + bool success; req = s2firehose_request(in_len - 1, in + 1, &success); if (!success) { byte* output_data = calloc(1 + 14, sizeof(byte)); - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "invalid request", 14); reply(15, output_data); return; } - if (unlikely(req.count > 64)) req.count = 64; // clamp + if (unlikely(req.count > 64)) req.count = 64; log(_info, "<%.*s> request type: firehose count %i offset %li", - (int) ip_size, - ip_addr, - req.count, - req.offset); + (int)ip_size, ip_addr, req.count, req.offset); - water_splash entries[ req.count ]; - uint16_t copied_count = firehose_get_packets(req.offset, req.count, entries); + water_splash entries[req.count]; + uint16_t copied_count = firehose_get_packets(req.offset, req.count, entries); size_t total_packets_size = PACKET_SIZE * copied_count; - size_t output_len = 1 + 8 + total_packets_size; - byte* output_data = calloc(output_len, sizeof(byte)); - output_data[ 0 ] = firehose_reply; - size_t offset = 1; - - uint64_t new_offset = req.offset + copied_count; - output_data[ offset++ ] = new_offset & 0xFF; - output_data[ offset++ ] = (new_offset >> 8) & 0xFF; - output_data[ offset++ ] = (new_offset >> 16) & 0xFF; - output_data[ offset++ ] = (new_offset >> 24) & 0xFF; - output_data[ offset++ ] = (new_offset >> 32) & 0xFF; - output_data[ offset++ ] = (new_offset >> 40) & 0xFF; - output_data[ offset++ ] = (new_offset >> 48) & 0xFF; - output_data[ offset++ ] = (new_offset >> 56) & 0xFF; + size_t output_len = 1 + 8 + total_packets_size; + byte* output_data = calloc(output_len, sizeof(byte)); + output_data[0] = firehose_reply; + size_t offset = 1; + + uint64_t new_offset = req.offset + copied_count; + output_data[offset++] = new_offset & 0xFF; + output_data[offset++] = (new_offset >> 8) & 0xFF; + output_data[offset++] = (new_offset >> 16) & 0xFF; + output_data[offset++] = (new_offset >> 24) & 0xFF; + output_data[offset++] = (new_offset >> 32) & 0xFF; + output_data[offset++] = (new_offset >> 40) & 0xFF; + output_data[offset++] = (new_offset >> 48) & 0xFF; + output_data[offset++] = (new_offset >> 56) & 0xFF; for (int i = 0; i < copied_count; i++) { - struct data_packet_entry* entry = &packet_storage[ entries[ i ].storage_index ]; - data_packet2s(entry->packet, output_data + offset); + struct data_packet_entry* entry = &packet_index[entries[i].storage_index]; + byte decompressed[sizeof(struct data_packet)]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress(decompressed, &decompressed_len, entry->compressed_payload, + entry->compressed_size) != Z_OK) { + log(_error, "firehose: decompression failed"); + continue; + } + data_packet2s(*(struct data_packet*)decompressed, output_data + offset); offset += PACKET_SIZE; } @@ -510,34 +491,33 @@ void paste_server( } case register_neighbor: { - bool success; + bool success; struct relay_neighbor neighbor = s2relay_neighbor(in_len - 1, in + 1, &success); if (!success) { byte* output_data = calloc(1 + 14, sizeof(byte)); - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "invalid request", 14); reply(15, output_data); return; } - log(_info, "<%.*s> request type: neighbor registration", (int) ip_size, ip_addr); + log(_info, "<%.*s> request type: neighbor registration", (int)ip_size, ip_addr); - char ip_string[ INET6_ADDRSTRLEN ]; + char ip_string[INET6_ADDRSTRLEN]; if (neighbor.has_ipv4) { sprintf( ip_string, "%d.%d.%d.%d", - neighbor.v4_addr[ 0 ], - neighbor.v4_addr[ 1 ], - neighbor.v4_addr[ 2 ], - neighbor.v4_addr[ 3 ]); + neighbor.v4_addr[0], + neighbor.v4_addr[1], + neighbor.v4_addr[2], + neighbor.v4_addr[3]); } else if (neighbor.has_ipv6) { int offset = 0; for (int j = 0; j < 8; j++) { - uint16_t word - = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ]; + uint16_t word = (neighbor.v6_addr[j * 2] << 8) | neighbor.v6_addr[j * 2 + 1]; offset += sprintf(ip_string + offset, "%x%s", word, (j < 7) ? ":" : ""); } } @@ -545,11 +525,10 @@ void paste_server( struct relay_info audit_result = run_server_audit(ip_string, neighbor.port); if (!audit_result.capabilities.can_store && audit_result.major_version == 0 && audit_result.minor_version == 0) { - // invalid byte* err_reply = malloc(2); if (err_reply) { - err_reply[ 0 ] = error; - err_reply[ 1 ] = 0; + err_reply[0] = error; + err_reply[1] = 0; } reply(2, err_reply); break; @@ -557,21 +536,21 @@ void paste_server( byte* __reply = malloc(2); if (__reply) { - __reply[ 0 ] = register_neighbor_result; - __reply[ 1 ] = add_neighbor(neighbor); + __reply[0] = register_neighbor_result; + __reply[1] = add_neighbor(neighbor); } reply(2, __reply); } break; case query_packets: { if (in_len != 1 + 32 + 32 + 64 + 2 + 8) { - *out = NULL; + *out = NULL; *out_len = 0; return; } public_key author_enc, author_sign; - signature sig; - size_t off = 1; + signature sig; + size_t off = 1; memcpy(author_enc, in + off, 32); off += 32; memcpy(author_sign, in + off, 32); @@ -582,61 +561,60 @@ void paste_server( off += 2; uint64_t req_offset = read_u64_le(in + off); off += 8; - byte msg[ 32 + 32 + 2 + 8 ]; + byte msg[32 + 32 + 2 + 8]; memcpy(msg, author_enc, 32); memcpy(msg + 32, author_sign, 32); write_u16_le(msg + 64, req_count); write_u64_le(msg + 66, req_offset); if (!verify_raw_signature(msg, sizeof(msg), sig, author_sign)) { - *out = NULL; + *out = NULL; *out_len = 0; return; } if (req_count > 128) req_count = 128; - struct data_packet* packets = malloc((size_t) req_count * sizeof(struct data_packet)); + struct data_packet* packets = malloc((size_t)req_count * sizeof(struct data_packet)); if (!packets) { - *out = NULL; + *out = NULL; *out_len = 0; return; } uint64_t new_offset; - uint16_t actual_count - = get_packets_for_author(author_enc, req_offset, req_count, packets, &new_offset); + uint16_t actual_count = get_packets_for_author(author_enc, req_offset, req_count, packets, &new_offset); uint16_t verified_count = 0; for (uint16_t i = 0; i < actual_count; i++) { - if (memcmp(packets[ i ].contents.author_signing_pub, author_sign, 32) == 0) { - packets[ verified_count++ ] = packets[ i ]; + if (memcmp(packets[i].contents.author_signing_pub, author_sign, 32) == 0) { + packets[verified_count++] = packets[i]; } } - size_t _out_len = 1 + 2 + 8 + (size_t) verified_count * PACKET_SIZE; - byte* _out = malloc(_out_len); - if (!out) { + size_t _out_len = 1 + 2 + 8 + (size_t)verified_count * PACKET_SIZE; + byte* _out = malloc(_out_len); + if (!_out) { free(packets); - *out = NULL; + *out = NULL; *out_len = 0; return; } - off = 0; - _out[ off++ ] = queried_author_packets; + off = 0; + _out[off++] = queried_author_packets; write_u16_le(_out + off, verified_count); off += 2; write_u64_le(_out + off, new_offset); off += 8; for (uint16_t i = 0; i < verified_count; i++) { - data_packet2s(packets[ i ], _out + off); + data_packet2s(packets[i], _out + off); off += PACKET_SIZE; } free(packets); - *out = _out; + *out = _out; *out_len = _out_len; return; } break; default: { - log(_info, "<%.*s> request type: invalid method", (int) ip_size, ip_addr); + log(_info, "<%.*s> request type: invalid method", (int)ip_size, ip_addr); byte* output_data = calloc(1 + 14, sizeof(byte)); - output_data[ 0 ] = error; + output_data[0] = error; memcpy(output_data + 1, "invalid input", 14); reply(15, output_data); return; @@ -654,14 +632,12 @@ void* alloc_map(uint64_t size) { if (try_hugepages) { alloc = mmap(NULL, size, RW, PA | MAP_HUGE_2MB, -1, 0); if (alloc == MAP_FAILED) { - // alloc failed. maybe the fault of hugepages? goto NO_HP; } } else { NO_HP: alloc = mmap(NULL, size, RW, PA, -1, 0); if (alloc != MAP_FAILED) { - // hugepages failed, remove flag printf("[note] no hugepages support present"); try_hugepages = false; } @@ -671,12 +647,15 @@ void* alloc_map(uint64_t size) { return alloc; } -pthread_t packet_cleanup_thread; +pthread_t packet_cleanup_thread; volatile bool packet_cleanup_thread_running = true; -pthread_t disk_persistence_thread; +pthread_t disk_persistence_thread; volatile bool disk_persistence_thread_running = true; +pthread_t neighbor_check_thread; +volatile bool neighbor_check_thread_running = true; + bool file_exists(const char* filename) { return access(filename, F_OK) == 0; } void save_packet(const struct data_packet packet) { @@ -688,23 +667,19 @@ void save_packet(const struct data_packet packet) { mkdir("packets", 0755); - char filename_tmp[ 256 ]; - char filename_final[ 256 ]; - snprintf( - filename_tmp, sizeof(filename_tmp), "packets/" hash_fmt ".bin.tmp", hash_ref(packet.hash)); - snprintf( - filename_final, sizeof(filename_final), "packets/" hash_fmt ".bin", hash_ref(packet.hash)); + char filename_tmp[256]; + char filename_final[256]; + snprintf(filename_tmp, sizeof(filename_tmp), "packets/" hash_fmt ".bin.tmp", hash_ref(packet.hash)); + snprintf(filename_final, sizeof(filename_final), "packets/" hash_fmt ".bin", hash_ref(packet.hash)); - /* serialize packet into buffer */ - byte buffer[ PACKET_SIZE ]; + byte buffer[PACKET_SIZE]; data_packet2s(packet, buffer); - /* compress */ uLongf comp_bound = compressBound(PACKET_SIZE); - byte comp_buf[ comp_bound ]; + byte comp_buf[comp_bound]; uLongf comp_len = comp_bound; - int zres = compress2(comp_buf, &comp_len, buffer, PACKET_SIZE, Z_BEST_COMPRESSION); + int zres = compress2(comp_buf, &comp_len, buffer, PACKET_SIZE, Z_BEST_COMPRESSION); if (zres != Z_OK) { log(_error, "save_packet: compress2 failed (%d)", zres); return; @@ -745,9 +720,9 @@ void save_neighbors() { pthread_mutex_lock(&neighbor_storage_mutex); for (uint64_t i = 0; i < max_neighbors; i++) { - if (!empty_neighbor_slots[ i ]) { - byte buffer[ RELAY_NEIGHBOR_WIRE_LEN ]; - relay_neighbor2s(neighbor_storage[ i ], buffer); + if (!empty_neighbor_slots[i]) { + byte buffer[RELAY_NEIGHBOR_WIRE_LEN]; + relay_neighbor2s(neighbor_storage[i], buffer); if (fwrite(buffer, 1, RELAY_NEIGHBOR_WIRE_LEN, file) != RELAY_NEIGHBOR_WIRE_LEN) { perror("failed to write neighbor entry"); } @@ -763,11 +738,10 @@ void load_neighbors() { size_t count = 0; - byte buffer[ RELAY_NEIGHBOR_WIRE_LEN ]; + byte buffer[RELAY_NEIGHBOR_WIRE_LEN]; while (fread(buffer, 1, RELAY_NEIGHBOR_WIRE_LEN, file) == RELAY_NEIGHBOR_WIRE_LEN) { - bool success; - struct relay_neighbor neighbor - = s2relay_neighbor(RELAY_NEIGHBOR_WIRE_LEN, buffer, &success); + bool success; + struct relay_neighbor neighbor = s2relay_neighbor(RELAY_NEIGHBOR_WIRE_LEN, buffer, &success); if (success) { count++; add_neighbor(neighbor); @@ -780,7 +754,7 @@ void load_neighbors() { void* fs_persistence(void* arg) { while (disk_persistence_thread_running) { - sleep(2 * 60); // 2m + sleep(2 * 60); log(_info, "saving neighbors..."); save_neighbors(); } @@ -790,70 +764,68 @@ void* fs_persistence(void* arg) { void* cleanup_packets(void* arg) { while (packet_cleanup_thread_running) { - sleep(30 * 60); // 30m + sleep(30 * 60); log(_warn, "running cleanup"); uint64_t current_time_ms = get_current_time_ms(); - uint64_t retention_ms = (uint64_t) server_stats.retention_hours * 60 * 60 * 1000; + uint64_t retention_ms = (uint64_t)server_stats.retention_hours * 60 * 60 * 1000; - // memory for (uint64_t i = 0; i < max_storage; i++) { pthread_mutex_lock(&packet_storage_mutex); - if (!packet_storage[ i ].is_full) { + if (!packet_index[i].is_full) { + pthread_mutex_unlock(&packet_storage_mutex); + continue; + } + + byte decompressed[sizeof(struct data_packet)]; + uLongf decompressed_len = sizeof(struct data_packet); + if (uncompress(decompressed, &decompressed_len, packet_index[i].compressed_payload, + packet_index[i].compressed_size) != Z_OK) { + log(_error, "cleanup_packets: decompression failed"); pthread_mutex_unlock(&packet_storage_mutex); continue; } + struct data_packet* packet = (struct data_packet*)decompressed; - bool result - = decide_packet_cleanup_func - ? decide_packet_cleanup_func(packet_storage[ i ].packet) // delegate to user - : ( // do it natively - current_time_ms - packet_storage[ i ].packet.contents.created_time_ms - > retention_ms); + bool result = decide_packet_cleanup_func + ? decide_packet_cleanup_func(*packet) + : (current_time_ms - packet->contents.created_time_ms > retention_ms); if (result) { log(_debug, "cleaning up old packet hashed " hash_fmt, - hash_ref(packet_storage[ i ].packet.hash)); + hash_ref(packet->hash)); - char filename[ 128 ]; - sprintf( - filename, - "packets/" hash_fmt ".bin", - hash_ref(packet_storage[ i ].packet.hash)); + char filename[128]; + sprintf(filename, "packets/" hash_fmt ".bin", hash_ref(packet->hash)); if (remove(filename) != 0) log(_error, "failed to delete packet %s", filename); - packet_storage[ i ].is_full = false; - packet_count--; + _delete_at_index(i); } pthread_mutex_unlock(&packet_storage_mutex); } - // external cleanup system("find packets/ -mtime +7 -delete"); - // disk cleanup, tho not sure this is needed DIR* dir = opendir("packets"); if (dir) { struct dirent* entry; while ((entry = readdir(dir)) != NULL) { if (entry->d_type == DT_REG) { - char filepath[ 256 ]; + char filepath[256]; snprintf(filepath, sizeof(filepath), "packets/%s", entry->d_name); FILE* fd = decide_packet_cleanup_func ? fopen(filepath, "rb") : NULL; struct data_packet* new = decide_packet_cleanup_func ? load_packet(fd) : NULL; struct stat st; - bool result = decide_packet_cleanup_func - ? decide_packet_cleanup_func(*new) // delegate to user - : ( // do it natively - stat(filepath, &st) == 0 - && current_time_ms - (uint64_t) st.st_mtime * 1000 - > retention_ms); + bool result = decide_packet_cleanup_func + ? decide_packet_cleanup_func(*new) + : (stat(filepath, &st) == 0 + && current_time_ms - (uint64_t)st.st_mtime * 1000 > retention_ms); if (decide_packet_cleanup_func) { fclose(fd); @@ -872,43 +844,38 @@ void* cleanup_packets(void* arg) { return NULL; } -pthread_t neighbor_check_thread; -volatile bool neighbor_check_thread_running = true; - void* check_neighbors(void* arg) { sleep(5 * 30); while (neighbor_check_thread_running) { pthread_mutex_lock(&neighbor_storage_mutex); for (uint64_t i = 0; i < max_neighbors; i++) { - if (!empty_neighbor_slots[ i ]) { - struct relay_neighbor neighbor = neighbor_storage[ i ]; + if (!empty_neighbor_slots[i]) { + struct relay_neighbor neighbor = neighbor_storage[i]; - byte* response_data = NULL; - size_t response_len = 0; - int result = -1; + byte* response_data = NULL; + size_t response_len = 0; + int result = -1; - byte audit_request[ 1 ] = { audit_server }; + byte audit_request[1] = {audit_server}; if (neighbor.has_ipv4) { - char ip_string[ 16 ]; + char ip_string[16]; sprintf( ip_string, "%d.%d.%d.%d", - neighbor.v4_addr[ 0 ], - neighbor.v4_addr[ 1 ], - neighbor.v4_addr[ 2 ], - neighbor.v4_addr[ 3 ]); + neighbor.v4_addr[0], + neighbor.v4_addr[1], + neighbor.v4_addr[2], + neighbor.v4_addr[3]); result = study( ip_string, neighbor.port, audit_request, 1, &response_data, &response_len); } else if (neighbor.has_ipv6) { - char ip_string[ 46 ]; - int offset = 0; + char ip_string[46]; + int offset = 0; for (int j = 0; j < 8; j++) { - uint16_t word - = (neighbor.v6_addr[ j * 2 ] << 8) | neighbor.v6_addr[ j * 2 + 1 ]; - + uint16_t word = (neighbor.v6_addr[j * 2] << 8) | neighbor.v6_addr[j * 2 + 1]; offset += snprintf( ip_string + offset, sizeof(ip_string) - offset, @@ -921,8 +888,8 @@ void* check_neighbors(void* arg) { ip_string, neighbor.port, audit_request, 1, &response_data, &response_len); } - if (result != 0 || response_data[ 0 ] != audit_response) { - empty_neighbor_slots[ i ] = true; + if (result != 0 || response_data[0] != audit_response) { + empty_neighbor_slots[i] = true; neighbor_count--; log(_warn, "removed dead neighbor"); } @@ -932,7 +899,7 @@ void* check_neighbors(void* arg) { } pthread_mutex_unlock(&neighbor_storage_mutex); - sleep(60 * 5); // 5m + sleep(60 * 5); } return NULL; } @@ -954,12 +921,9 @@ int main(int argc, char** argv) { log_level = _log_level; } else { if (strcmp(__log_level, "debug") == 0) log_level = 1; - else if (strcmp(__log_level, "info") == 0) - log_level = 2; - else if (strcmp(__log_level, "warn") == 0) - log_level = 3; - else if (strcmp(__log_level, "error") == 0) - log_level = 4; + else if (strcmp(__log_level, "info") == 0) log_level = 2; + else if (strcmp(__log_level, "warn") == 0) log_level = 3; + else if (strcmp(__log_level, "error") == 0) log_level = 4; else { fprintf( stderr, @@ -991,15 +955,16 @@ int main(int argc, char** argv) { server_stats.minor_version, current_port); - // init - packet_storage = alloc_map(max_storage * sizeof(struct data_packet_entry)); + packet_index = alloc_map(max_storage * sizeof(struct data_packet_entry)); + packet_arena_size = max_storage * (sizeof(uint8_t) + sizeof(uint32_t) + sizeof(struct data_packet)); + packet_arena = alloc_map(packet_arena_size); author_storage = alloc_map(max_authors * sizeof(struct author_entry)); - neighbor_storage = alloc_map(max_neighbors * sizeof(struct relay_neighbor)); + neighbor_storage = alloc_map(max_neighbors * sizeof(struct relay_neighbor)); empty_neighbor_slots = alloc_map(max_neighbors * sizeof(bool)); memset(empty_neighbor_slots, 1, max_neighbors * sizeof(bool)); - if (!packet_storage || !author_storage || !neighbor_storage || !empty_neighbor_slots) { + if (!packet_index || !packet_arena || !author_storage || !neighbor_storage || !empty_neighbor_slots) { perror("init alloc failed"); return 1; } @@ -1023,7 +988,6 @@ int main(int argc, char** argv) { firehose_init(); - // load the plugins! char* plugin_path = get_config("plugin", false); if (plugin_path) { load_plugins(plugin_path); @@ -1036,4 +1000,4 @@ int main(int argc, char** argv) { pthread_join(packet_cleanup_thread, NULL); return 0; -} +} \ No newline at end of file -- cgit v1.2.3