summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorGravatar moon <moon@coffin.ir> 2025-09-14 20:27:54 +0330
committerGravatar moon <moon@coffin.ir> 2025-09-14 20:27:54 +0330
commit067200b438e022cef840d0e772db96657e5cafa9 (patch)
tree4740cf9f453c56823b7ee8e09b032ef907bfc76f
parentbetter py client (diff)
stuff
-rw-r--r--src/common.c34
-rw-r--r--src/packet_storage.c309
-rw-r--r--src/server.c443
3 files changed, 546 insertions, 240 deletions
diff --git a/src/common.c b/src/common.c
index 0fef6da..74515c1 100644
--- a/src/common.c
+++ b/src/common.c
@@ -50,11 +50,23 @@ enum client_input {
// query packets for a specific author
// note: to prevent DoS, you must provide a signature by the requested author to be approved.
- // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64 offset`.
- // server must verify the signing key is of the author's pubkey and signing key by looking at the packet's metadata.
+ // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64
+ // offset`.
+ // server must verify the signing key is of the author's pubkey and signing key by looking at
+ // the packet's metadata.
// (input public_key, input pub_sign_key, signature, u16 count, u64 offset)
// (responds with queried_author_packets)
- query_packets = 0x16
+ query_packets = 0x16,
+
+ // query packets for a specific recipient
+ // note: to prevent DoS, you must provide a signature by the requested author to be approved.
+ // the signature needs to be from a struct shaped `public_key + pub_sign_key + u16 count + u64
+ // offset`.
+ // server must verify the signing key is of the recipient's pubkey and signing key by looking at
+ // the packet's metadata.
+ // (input public_key, input pub_sign_key, signature, u16 count, u64 offset)
+ // (responds with queried_author_packets)
+ query_by_recipient = 0x17,
};
enum server_reply {
@@ -62,19 +74,21 @@ enum server_reply {
error = 0x1,
// type `struct relay_info`
- audit_response = 0x17,
+ audit_response = 0x18,
// type `struct relay_neighbor[ neighbor_count ]`
- neighbors_info = 0x18,
+ neighbors_info = 0x19,
// type `uint64_t new_offset, struct data_packet[ count ]`
- firehose_reply = 0x19,
+ firehose_reply = 0x1a,
// type `bool result_code`
- push_packet_result = 0x1a,
+ push_packet_result = 0x1b,
// type `struct data_packet`
- get_packet_result = 0x1b,
+ get_packet_result = 0x1c,
// type `bool result_code`
- register_neighbor_result = 0x1c,
+ register_neighbor_result = 0x1d,
+ // type `u16 count, u64 new_offset, struct data_packet[ count ]`
+ queried_author_packets = 0x1e,
// type `u16 count, u64 new_offset, struct data_packet[ count ]`
- queried_author_packets = 0x1d
+ queried_recipient_packets = 0x1f,
};
typedef uint8_t get_neighbors_request;
diff --git a/src/packet_storage.c b/src/packet_storage.c
index 52a80d0..569eed4 100644
--- a/src/packet_storage.c
+++ b/src/packet_storage.c
@@ -8,8 +8,9 @@
#include <sys/stat.h>
#include <zlib.h>
-const uint64_t max_storage = 1 << 16;
-const uint64_t max_authors = 1 << 14;
+const uint64_t max_storage = 1 << 16;
+const uint64_t max_authors = 1 << 14;
+const uint64_t max_recipients = 1 << 14;
#define hash_fmt \
"%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x%02x" \
@@ -31,6 +32,8 @@ struct data_packet_entry {
uint64_t prev;
uint64_t author_next;
uint64_t author_prev;
+ uint64_t recipient_next;
+ uint64_t recipient_prev;
bool is_full;
} *packet_index;
byte *packet_arena;
@@ -44,7 +47,16 @@ struct author_entry {
bool is_full;
} *author_storage;
-pthread_mutex_t author_storage_mutex = PTHREAD_MUTEX_INITIALIZER;
+struct recipient_entry {
+ public_key recipient;
+ uint64_t packet_head;
+ uint64_t next;
+ uint64_t prev;
+ bool is_full;
+} *recipient_storage;
+
+pthread_mutex_t author_storage_mutex = PTHREAD_MUTEX_INITIALIZER;
+pthread_mutex_t recipient_storage_mutex = PTHREAD_MUTEX_INITIALIZER;
#define invalid_index ((uint64_t) 0)
@@ -142,6 +154,23 @@ void save_author_packet_hash(const public_key author, const hash packet_hash) {
fclose(f);
}
+void save_recipient_packet_hash(const public_key recipient, const hash packet_hash) {
+ const char *dir = "packets_by_recipient";
+ mkdir(dir, 0755);
+
+ char recipient_hex[ 65 ];
+ sprintf(recipient_hex, hash_fmt, hash_ref(recipient));
+
+ char filename[ 256 ];
+ snprintf(filename, sizeof(filename), "%s/%s.list", dir, recipient_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));
@@ -198,6 +227,62 @@ uint64_t get_author_entry(const public_key auth_key, bool create) {
return invalid_index;
}
+uint64_t get_recipient_entry(const public_key recip_key, bool create) {
+ uint64_t hash_val;
+ memcpy(&hash_val, recip_key, sizeof(uint64_t));
+ uint64_t start_index = hash_val % max_recipients;
+ uint64_t cur_index = start_index;
+
+ pthread_mutex_lock(&recipient_storage_mutex);
+
+ struct recipient_entry *current = &recipient_storage[ cur_index ];
+ while (true) {
+ if (current->is_full && memcmp(current->recipient, recip_key, 32) == 0) {
+ pthread_mutex_unlock(&recipient_storage_mutex);
+ return cur_index;
+ }
+ if (current->next == invalid_index) break;
+ cur_index = current->next;
+ current = &recipient_storage[ cur_index ];
+ }
+
+ if (!create) {
+ pthread_mutex_unlock(&recipient_storage_mutex);
+ return invalid_index;
+ }
+
+ if (!current->is_full) {
+ memcpy(current->recipient, recip_key, 32);
+ current->packet_head = invalid_index;
+ current->is_full = true;
+ current->next = invalid_index;
+ current->prev = invalid_index;
+ pthread_mutex_unlock(&recipient_storage_mutex);
+ return cur_index;
+ }
+
+ uint64_t probes = 0;
+ uint64_t lp_idx = (cur_index + 1) % max_recipients;
+ while (probes < max_recipients) {
+ struct recipient_entry *empty = &recipient_storage[ lp_idx ];
+ if (!empty->is_full) {
+ memcpy(empty->recipient, recip_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(&recipient_storage_mutex);
+ return lp_idx;
+ }
+ lp_idx = (lp_idx + 1) % max_recipients;
+ probes++;
+ }
+
+ pthread_mutex_unlock(&recipient_storage_mutex);
+ return invalid_index;
+}
+
struct data_packet *load_packet(FILE *file) {
if (!file) return NULL;
@@ -269,6 +354,9 @@ uint64_t set_data_packet(struct data_packet *new_packet) {
current_entry->compressed_size = compressed_len;
pthread_mutex_unlock(&packet_storage_mutex);
save_author_packet_hash(new_packet->contents.author, new_packet->hash);
+ for (uint8_t i = 0; i < new_packet->contents.recipient_count; i++) {
+ save_recipient_packet_hash(new_packet->contents.recipients[ i ], new_packet->hash);
+ }
return cur_index;
}
if (current_entry->next == invalid_index) break;
@@ -282,35 +370,65 @@ uint64_t set_data_packet(struct data_packet *new_packet) {
current_entry->is_full = true;
current_entry->next = invalid_index;
current_entry->prev = invalid_index;
+ current_entry->author_next = invalid_index;
+ current_entry->author_prev = invalid_index;
+ current_entry->recipient_next = invalid_index;
+ current_entry->recipient_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;
+ } else {
+ 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_index[ linear_probe_index ];
+ if (!empty_slot->is_full) {
+ 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;
+ empty_slot->author_next = invalid_index;
+ empty_slot->author_prev = invalid_index;
+ empty_slot->recipient_next = invalid_index;
+ empty_slot->recipient_prev = invalid_index;
+ current_entry->next = linear_probe_index;
+ packet_count++;
+ cur_index = linear_probe_index;
+ break;
+ }
+ linear_probe_index = (linear_probe_index + 1) % max_storage;
+ probes++;
+ }
+ if (probes >= max_storage) {
+ pthread_mutex_unlock(&packet_storage_mutex);
+ arena_free(compressed_payload);
+ return invalid_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_index[ linear_probe_index ];
- if (!empty_slot->is_full) {
- 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;
+ uint64_t auth_idx = get_author_entry(new_packet->contents.author, true);
+ if (auth_idx != invalid_index) {
+ current_entry->author_next = author_storage[ auth_idx ].packet_head;
+ if (current_entry->author_next != invalid_index) {
+ packet_index[ current_entry->author_next ].author_prev = cur_index;
+ }
+ author_storage[ auth_idx ].packet_head = cur_index;
+ }
+
+ for (uint8_t i = 0; i < new_packet->contents.recipient_count; i++) {
+ uint64_t recip_idx = get_recipient_entry(new_packet->contents.recipients[ i ], true);
+ if (recip_idx != invalid_index) {
+ current_entry->recipient_next = recipient_storage[ recip_idx ].packet_head;
+ if (current_entry->recipient_next != invalid_index) {
+ packet_index[ current_entry->recipient_next ].recipient_prev = cur_index;
+ }
+ recipient_storage[ recip_idx ].packet_head = cur_index;
+ save_recipient_packet_hash(new_packet->contents.recipients[ i ], new_packet->hash);
}
- linear_probe_index = (linear_probe_index + 1) % max_storage;
- probes++;
}
pthread_mutex_unlock(&packet_storage_mutex);
- arena_free(compressed_payload);
- return invalid_index;
+ save_author_packet_hash(new_packet->contents.author, new_packet->hash);
+ return cur_index;
}
struct data_packet_entry *get_data_packet(const hash _hash) {
@@ -384,6 +502,10 @@ struct data_packet_entry *get_data_packet(const hash _hash) {
temp_entry.is_full = true;
temp_entry.next = invalid_index;
temp_entry.prev = invalid_index;
+ temp_entry.author_next = invalid_index;
+ temp_entry.author_prev = invalid_index;
+ temp_entry.recipient_next = invalid_index;
+ temp_entry.recipient_prev = invalid_index;
free(packet);
return &temp_entry;
}
@@ -450,6 +572,26 @@ bool _delete_at_index(uint64_t index) {
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;
+ for (uint8_t i = 0;
+ i < ((struct data_packet *) current_entry->compressed_payload)->contents.recipient_count;
+ i++) {
+ public_key recip_key;
+ memcpy(
+ recip_key,
+ ((struct data_packet *) current_entry->compressed_payload)->contents.recipients[ i ],
+ 32);
+ uint64_t recip_idx = get_recipient_entry(recip_key, false);
+ if (recip_idx != invalid_index) {
+ if (recipient_storage[ recip_idx ].packet_head == index) {
+ recipient_storage[ recip_idx ].packet_head = current_entry->recipient_next;
+ }
+ }
+ uint64_t r_prev = current_entry->recipient_prev;
+ uint64_t r_next = current_entry->recipient_next;
+ if (r_prev != invalid_index) packet_index[ r_prev ].recipient_next = r_next;
+ if (r_next != invalid_index) packet_index[ r_next ].recipient_prev = r_prev;
+ }
+
if (current_entry->prev != invalid_index) {
packet_index[ current_entry->prev ].next = current_entry->next;
}
@@ -463,6 +605,10 @@ bool _delete_at_index(uint64_t index) {
current_entry->prev = invalid_index;
current_entry->compressed_payload = NULL;
current_entry->compressed_size = 0;
+ current_entry->author_next = invalid_index;
+ current_entry->author_prev = invalid_index;
+ current_entry->recipient_next = invalid_index;
+ current_entry->recipient_prev = invalid_index;
return true;
}
@@ -583,4 +729,117 @@ uint16_t get_packets_for_author(
free(all_packets);
return collected;
+}
+
+uint16_t get_packets_for_recipient(
+ const public_key recipient,
+ uint64_t offset,
+ uint16_t max_count,
+ struct data_packet *out_packets,
+ uint64_t *out_new_offset) {
+ uint64_t recip_idx = get_recipient_entry(recipient, false);
+ if (recip_idx == invalid_index) return 0;
+
+ uint64_t fetch_factor = ((offset + (uint64_t) max_count) * 5LLU) / 2;
+
+ pthread_mutex_lock(&packet_storage_mutex);
+ uint64_t curr = recipient_storage[ recip_idx ].packet_head;
+ uint16_t mem_count = 0;
+ while (curr != invalid_index) {
+ mem_count++;
+ curr = packet_index[ curr ].recipient_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 = recipient_storage[ recip_idx ].packet_head;
+ while (curr != invalid_index) {
+ 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_recipient: decompression failed\n");
+ continue;
+ }
+ memcpy(&all_packets[ all_count++ ], decompressed, sizeof(struct data_packet));
+ curr = packet_index[ curr ].recipient_next;
+ }
+ pthread_mutex_unlock(&packet_storage_mutex);
+
+ char recipient_hex[ 65 ];
+ sprintf(recipient_hex, hash_fmt, hash_ref(recipient));
+
+ char filename[ 256 ];
+ snprintf(filename, sizeof(filename), "packets_by_recipient/%s.list", recipient_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;
+
+ 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_recipient: decompression failed for disk packet\n");
+ continue;
+ }
+ memcpy(&all_packets[ all_count++ ], decompressed, sizeof(struct data_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
diff --git a/src/server.c b/src/server.c
index 0b7ea87..3f1a9ee 100644
--- a/src/server.c
+++ b/src/server.c
@@ -4,7 +4,9 @@
#include <sys/mman.h>
#include <sys/stat.h>
#include <unistd.h>
+// dependency: zlib
#include <zlib.h>
+//
#include "common.c"
#include "config.c"
#include "firehose.c"
@@ -18,26 +20,26 @@ uint16_t current_port = default_port;
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];
+ byte major_version;
+ byte minor_version;
+ byte motd[ 128 ];
uint16_t retention_hours;
- byte leading_zeros;
- bool accepts_plain;
+ byte leading_zeros;
+ bool accepts_plain;
} server_stats = {
- .major_version = 0,
- .minor_version = 1,
- .motd = "A moonpaste server.",
+ .major_version = 0,
+ .minor_version = 1,
+ .motd = "A moonpaste server.",
.retention_hours = 168,
- .leading_zeros = 12,
- .accepts_plain = true,
+ .leading_zeros = 12,
+ .accepts_plain = true,
};
bool add_neighbor(struct relay_neighbor new_neighbor) {
@@ -55,15 +57,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;
@@ -74,7 +76,7 @@ bool add_neighbor(struct relay_neighbor new_neighbor) {
}
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;
@@ -83,35 +85,36 @@ 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) ? ":" : "");
}
@@ -132,12 +135,15 @@ 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);
@@ -148,41 +154,41 @@ void paste_server(
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;
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);
@@ -199,28 +205,30 @@ 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++) {
- relay_neighbor2s(neighbor_storage[i], output_data + offset);
+ relay_neighbor2s(neighbor_storage[ i ], output_data + offset);
offset += RELAY_NEIGHBOR_WIRE_LEN;
}
@@ -232,7 +240,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;
@@ -244,12 +252,14 @@ 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;
@@ -259,84 +269,88 @@ 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;
}
- byte decompressed[sizeof(struct data_packet)];
+ 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) {
+ 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;
+ 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));
+ 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(*(struct data_packet*)decompressed, 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) {
- modify_packet_func(&packet);
- }
+ if (modify_packet_func) { 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(
@@ -348,7 +362,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;
}
@@ -356,7 +370,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;
}
@@ -364,14 +378,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;
}
@@ -379,7 +393,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,
@@ -397,7 +411,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;
}
@@ -406,11 +420,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;
@@ -420,7 +434,7 @@ void paste_server(
}
}
} else {
- output_data[1] = false;
+ output_data[ 1 ] = false;
}
reply(output_len, output_data);
@@ -430,18 +444,18 @@ 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;
@@ -451,38 +465,45 @@ void paste_server(
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_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) {
+ 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);
+ data_packet2s(*(struct data_packet*) decompressed, output_data + offset);
offset += PACKET_SIZE;
}
@@ -491,33 +512,34 @@ 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) ? ":" : "");
}
}
@@ -527,8 +549,8 @@ void paste_server(
&& audit_result.minor_version == 0) {
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;
@@ -536,21 +558,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);
@@ -561,60 +583,61 @@ 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);
+ 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;
@@ -631,9 +654,7 @@ void* alloc_map(uint64_t size) {
void* alloc;
if (try_hugepages) {
alloc = mmap(NULL, size, RW, PA | MAP_HUGE_2MB, -1, 0);
- if (alloc == MAP_FAILED) {
- goto NO_HP;
- }
+ if (alloc == MAP_FAILED) { goto NO_HP; }
} else {
NO_HP:
alloc = mmap(NULL, size, RW, PA, -1, 0);
@@ -647,13 +668,13 @@ 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;
+pthread_t neighbor_check_thread;
volatile bool neighbor_check_thread_running = true;
bool file_exists(const char* filename) { return access(filename, F_OK) == 0; }
@@ -667,19 +688,21 @@ 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));
- byte buffer[PACKET_SIZE];
+ byte buffer[ PACKET_SIZE ];
data_packet2s(packet, buffer);
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;
@@ -720,9 +743,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");
}
@@ -738,10 +761,11 @@ 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);
@@ -768,36 +792,38 @@ void* cleanup_packets(void* arg) {
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;
for (uint64_t i = 0; i < max_storage; i++) {
pthread_mutex_lock(&packet_storage_mutex);
- if (!packet_index[i].is_full) {
+ if (!packet_index[ i ].is_full) {
pthread_mutex_unlock(&packet_storage_mutex);
continue;
}
- byte decompressed[sizeof(struct data_packet)];
+ 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) {
+ 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;
+ struct data_packet* packet = (struct data_packet*) decompressed;
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->hash));
+ log(_debug, "cleaning up old packet hashed " hash_fmt, hash_ref(packet->hash));
- char filename[128];
+ 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);
@@ -815,17 +841,18 @@ void* cleanup_packets(void* arg) {
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)
- : (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);
@@ -850,32 +877,33 @@ void* check_neighbors(void* arg) {
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,
@@ -888,8 +916,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");
}
@@ -921,9 +949,12 @@ 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,
@@ -956,15 +987,17 @@ int main(int argc, char** argv) {
current_port);
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);
+ 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_index || !packet_arena || !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;
}