#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; #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 ] 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; 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; } fseek(file, 0, SEEK_END); size_t file_size = ftell(file); fseek(file, 0, SEEK_SET); if (file_size == 0) { fclose(file); return NULL; } byte comp_buf[ file_size ]; size_t r = fread(comp_buf, 1, file_size, file); fclose(file); 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; } bool success; struct data_packet* packet = malloc(sizeof(struct data_packet)); if (!packet) { return NULL; } *packet = s2data_packet(PACKET_SIZE, buffer, &success); if (!success) { free(packet); return NULL; } 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; } } return NULL; } 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 ]; 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)); pthread_mutex_unlock(&packet_storage_mutex); return cur_index; } if (current_entry->next == invalid_index) break; cur_index = current_entry->next; current_entry = &packet_storage[ 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; packet_count++; pthread_mutex_unlock(&packet_storage_mutex); 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 ]; 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; packet_count++; pthread_mutex_unlock(&packet_storage_mutex); return linear_probe_index; } linear_probe_index = (linear_probe_index + 1) % max_storage; probes++; } // table full pthread_mutex_unlock(&packet_storage_mutex); 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)) bool _delete_at_index(uint64_t index); overload void delete_data_packet(hash _hash) { pthread_mutex_lock(&packet_storage_mutex); uint64_t hash_val; memcpy(&hash_val, _hash, sizeof(uint64_t)); uint64_t start_index = hash_val % max_storage; uint64_t cur_index = start_index; 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) { _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; } cur_index = current_entry->next; } pthread_mutex_unlock(&packet_storage_mutex); } overload void delete_data_packet(uint64_t index) { pthread_mutex_lock(&packet_storage_mutex); _delete_at_index(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 ]; if (!current_entry->is_full) return false; 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 != 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 = 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; }