diff options
Diffstat (limited to 'src/packet_storage.c')
| -rw-r--r-- | src/packet_storage.c | 255 |
1 files changed, 238 insertions, 17 deletions
diff --git a/src/packet_storage.c b/src/packet_storage.c index 1094afd..067c742 100644 --- a/src/packet_storage.c +++ b/src/packet_storage.c @@ -1,12 +1,15 @@ +#include <dirent.h> #include <pthread.h> #include <stdint.h> #include <stdlib.h> +#include <sys/stat.h> // dependency: zlib #include <zlib.h> // #include "common.c" const uint64_t max_storage = 1 << 16; +const uint64_t max_authors = 1 << 14; #define hash_fmt \ "%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x" \ @@ -25,11 +28,96 @@ 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; -// TODO: yes, i know it's weird, but this has to be zero -#define minus_one ((uint64_t) 0) +struct author_entry { + public_key author; + uint64_t packet_head; + uint64_t next; + uint64_t prev; + bool is_full; +}* 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) + +void save_author_packet_hash(const public_key author, const hash packet_hash) { + const char* dir = "packets_by_author"; + mkdir(dir, 0755); + + char author_hex[ 65 ]; + sprintf(author_hex, hash_fmt, hash_ref(author)); + + char filename[ 256 ]; + snprintf(filename, sizeof(filename), "%s/%s.list", dir, author_hex); + + FILE* f = fopen(filename, "ab"); + if (!f) return; + fwrite(packet_hash, 32, 1, f); + fputc('\n', f); + fclose(f); +} + +uint64_t get_author_entry(const public_key auth_key, bool create) { + uint64_t hash_val; + memcpy(&hash_val, auth_key, sizeof(uint64_t)); + uint64_t start_index = hash_val % max_authors; + uint64_t cur_index = start_index; + + pthread_mutex_lock(&author_storage_mutex); + + 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); + return cur_index; + } + if (current->next == invalid_index) break; + cur_index = current->next; + current = &author_storage[ cur_index ]; + } + + if (!create) { + pthread_mutex_unlock(&author_storage_mutex); + return invalid_index; + } + + if (!current->is_full) { + memcpy(current->author, auth_key, 32); + current->packet_head = invalid_index; + current->is_full = true; + current->next = invalid_index; + current->prev = invalid_index; + pthread_mutex_unlock(&author_storage_mutex); + return cur_index; + } + + 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 ]; + if (!empty->is_full) { + memcpy(empty->author, auth_key, 32); + empty->packet_head = invalid_index; + empty->is_full = true; + empty->next = invalid_index; + empty->prev = cur_index; + current->next = lp_idx; + pthread_mutex_unlock(&author_storage_mutex); + return lp_idx; + } + lp_idx = (lp_idx + 1) % max_authors; + probes++; + } + + pthread_mutex_unlock(&author_storage_mutex); + return invalid_index; +} struct data_packet* load_packet(FILE* file) { if (!file) { return NULL; } @@ -79,7 +167,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { pthread_mutex_lock(&packet_storage_mutex); struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; - while (cur_index != minus_one) { + 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); @@ -87,7 +175,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { } cur_index = current_entry->next; - if (cur_index != minus_one) { current_entry = &packet_storage[ cur_index ]; } + if (cur_index != invalid_index) { current_entry = &packet_storage[ cur_index ]; } } pthread_mutex_unlock(&packet_storage_mutex); @@ -99,7 +187,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { if (packet) { uint64_t add_idx = set_data_packet(packet); - if (add_idx != minus_one) { + if (add_idx != invalid_index) { free(packet); return &packet_storage[ add_idx ]; } else { @@ -113,7 +201,7 @@ struct data_packet_entry* get_data_packet(const hash _hash) { return NULL; } -uint64_t set_data_packet(struct data_packet* new_packet) { +uint64_t __set_data_packet(struct data_packet* new_packet) { pthread_mutex_lock(&packet_storage_mutex); uint64_t hash_val; @@ -131,7 +219,7 @@ uint64_t set_data_packet(struct data_packet* new_packet) { return cur_index; } - if (current_entry->next == minus_one) break; + if (current_entry->next == invalid_index) break; cur_index = current_entry->next; current_entry = &packet_storage[ cur_index ]; } @@ -139,8 +227,8 @@ uint64_t set_data_packet(struct data_packet* new_packet) { if (!current_entry->is_full) { memcpy(¤t_entry->packet, new_packet, sizeof(struct data_packet)); current_entry->is_full = true; - current_entry->next = minus_one; - current_entry->prev = minus_one; + current_entry->next = invalid_index; + current_entry->prev = invalid_index; packet_count++; pthread_mutex_unlock(&packet_storage_mutex); return cur_index; @@ -155,7 +243,7 @@ uint64_t set_data_packet(struct data_packet* new_packet) { if (!empty_slot->is_full) { memcpy(&empty_slot->packet, new_packet, sizeof(struct data_packet)); empty_slot->is_full = true; - empty_slot->next = minus_one; + empty_slot->next = invalid_index; empty_slot->prev = cur_index; current_entry->next = linear_probe_index; packet_count++; @@ -169,7 +257,26 @@ uint64_t set_data_packet(struct data_packet* new_packet) { // table full pthread_mutex_unlock(&packet_storage_mutex); - return minus_one; + 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; + + save_author_packet_hash(new_packet->contents.author, new_packet->hash); + + 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); + } + return result; } #define overload __attribute__((overloadable)) @@ -184,7 +291,7 @@ overload void delete_data_packet(hash _hash) { uint64_t start_index = hash_val % max_storage; uint64_t cur_index = start_index; - while (cur_index != minus_one) { + while (cur_index != invalid_index) { struct data_packet_entry* current_entry = &packet_storage[ cur_index ]; if (current_entry->is_full && memcmp(current_entry->packet.hash, _hash, sizeof(hash)) == 0) { @@ -193,7 +300,7 @@ overload void delete_data_packet(hash _hash) { return; } - if (current_entry->next == minus_one) { + if (current_entry->next == invalid_index) { pthread_mutex_unlock(&packet_storage_mutex); return; } @@ -215,19 +322,133 @@ bool _delete_at_index(uint64_t index) { struct data_packet_entry* current_entry = &packet_storage[ index ]; if (!current_entry->is_full) return false; - if (current_entry->prev != minus_one) { + public_key auth_key; + memcpy(auth_key, packet_storage[ index ].packet.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; + } + } + 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; + + if (current_entry->prev != invalid_index) { struct data_packet_entry* prev_entry = &packet_storage[ current_entry->prev ]; prev_entry->next = current_entry->next; } - if (current_entry->next != minus_one) { + if (current_entry->next != invalid_index) { struct data_packet_entry* next_entry = &packet_storage[ current_entry->next ]; next_entry->prev = current_entry->prev; } current_entry->is_full = false; - current_entry->next = minus_one; - current_entry->prev = minus_one; + current_entry->next = invalid_index; + current_entry->prev = invalid_index; 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; + return (ta > tb) ? -1 : (ta < tb) ? 1 : 0; +} + +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) { + 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); + uint64_t curr = author_storage[ auth_idx ].packet_head; + uint16_t mem_count = 0; + while (curr != invalid_index) { + mem_count++; + curr = packet_storage[ curr ].author_next; + } + + struct data_packet* all_packets + = malloc((mem_count + fetch_factor) * sizeof(struct data_packet)); + if (!all_packets) { + pthread_mutex_unlock(&packet_storage_mutex); + return 0; + } + size_t all_count = 0; + + 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; + } + pthread_mutex_unlock(&packet_storage_mutex); + + char author_hex[ 65 ]; + sprintf(author_hex, hash_fmt, hash_ref(author)); + + char filename[ 256 ]; + snprintf(filename, sizeof(filename), "packets_by_author/%s.list", author_hex); + + FILE* f = fopen(filename, "rb"); + size_t disk_added = 0; + if (f) { + fseek(f, 0, SEEK_END); + long fsize = ftell(f); + fseek(f, 0, SEEK_SET); + uint64_t total = fsize / 33; + + uint64_t fetch_count = fetch_factor; + if (fetch_count > total) fetch_count = total; + + for (uint64_t i = 0; i < fetch_count; i++) { + long pos = (total - 1 - i) * 33; + fseek(f, pos, SEEK_SET); + + hash phash; + if (fread(phash, 32, 1, f) != 1) break; + if (fgetc(f) != '\n') break; + + bool dupe = false; + for (size_t j = 0; j < all_count; j++) { + if (memcmp(all_packets[ j ].hash, phash, 32) == 0) { + dupe = true; + break; + } + } + if (dupe) continue; + + struct data_packet_entry* ent = get_data_packet(phash); + if (!ent) continue; + + all_packets[ all_count++ ] = ent->packet; + disk_added++; + } + fclose(f); + } + + qsort(all_packets, all_count, sizeof(struct data_packet), get_packets_for_author__cmp); + + if (offset >= all_count) { + free(all_packets); + *out_new_offset = offset; + return 0; + } + + size_t avail = all_count - offset; + uint16_t collected = (avail > max_count) ? max_count : avail; + memcpy(out_packets, all_packets + offset, collected * sizeof(struct data_packet)); + *out_new_offset = offset + collected; + + free(all_packets); + return collected; +}
\ No newline at end of file |