bug fixes

4 files changed, 42 insertions, 11 deletions

diff --git a/client.c b/client.c
index 6dd9f00..f8f3c93 100644
--- a/client.c
+++ b/client.c
@@ -335,7 +335,7 @@ void get_firehose_stream(const char* ip, int port, uint64_t offset, uint16_t cou
 
         if (ret != 0 || resp_len < 9 || response[ 0 ] != firehose_reply) {
             free(response);
-            sleep(100);
+            sleep(5);
             continue;
         }
 
@@ -345,7 +345,7 @@ void get_firehose_stream(const char* ip, int port, uint64_t offset, uint16_t cou
 
         if (available_packets == 0) {
             free(response);
-            sleep(100);
+            sleep(5);
             continue;
         }
 
@@ -361,7 +361,7 @@ void get_firehose_stream(const char* ip, int port, uint64_t offset, uint16_t cou
         printf("offset: %li\n", last_offset);
         last_offset += available_packets;
         free(response);
-        sleep(100);
+        // sleep(5);
     }
 }
 
diff --git a/firehose.c b/firehose.c
index 240dfd8..67d62c8 100644
--- a/firehose.c
+++ b/firehose.c
@@ -62,6 +62,8 @@ void firehose_push(hash new_hash, uint64_t storage_index) {
 uint16_t firehose_get_count() { return firehose_buffer.count; }
 uint64_t firehose_get_total_packets() { return firehose_buffer.total_packets; }
 
+pthread_mutex_t packet_storage_mutex;
+
 uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* output_entries) {
     pthread_mutex_lock(&firehose_buffer.mutex);
 
@@ -80,6 +82,7 @@ uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* out
     }
     uint16_t start_index = (firehose_buffer.head + relative_offset) % FIREHOSE_BUFFER_SIZE;
 
+    pthread_mutex_lock(&packet_storage_mutex);
     for (int i = 0; i < count; i++) {
         if (relative_offset + i >= firehose_buffer.count) { break; }
 
@@ -88,6 +91,7 @@ uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* out
             &output_entries[ i ], &firehose_buffer.entries[ current_index ], sizeof(water_splash));
         copied_count++;
     }
+    pthread_mutex_unlock(&packet_storage_mutex);
 
     pthread_mutex_unlock(&firehose_buffer.mutex);
     return copied_count;
diff --git a/packet_storage.c b/packet_storage.c
index e618490..8dd0b16 100644
--- a/packet_storage.c
+++ b/packet_storage.c
@@ -103,12 +103,13 @@ struct data_packet_entry* get_data_packet(hash _hash) {
 
     if (packet) {
         uint64_t add_idx = set_data_packet(packet);
-
         if (add_idx != minus_one) {
+            free(packet);
             return &packet_storage[ add_idx ];
         } else {
-            static struct data_packet_entry temp_entry;
+            __thread static struct data_packet_entry temp_entry;
             temp_entry.packet = *packet;
+            free(packet);
             return &temp_entry;
         }
     }
diff --git a/server.c b/server.c
index 9c1a710..9715467 100644
--- a/server.c
+++ b/server.c
@@ -1,3 +1,4 @@
+#include <dirent.h>
 #include <fcntl.h>
 #include <sys/mman.h>
 #include <sys/stat.h>
@@ -403,12 +404,14 @@ void paste_server(
 
                 save_packet(packet);
 
-                struct data_packet* thread_arg = malloc(sizeof(struct data_packet));
-                if (thread_arg) {
-                    *thread_arg = packet;
-                    pthread_t prop_thread;
-                    pthread_create(&prop_thread, NULL, propagate_packet, thread_arg);
-                    pthread_detach(prop_thread);
+                if (((float) rand() / (float) RAND_MAX) > 0.5) {
+                    struct data_packet* thread_arg = malloc(sizeof(struct data_packet));
+                    if (thread_arg) {
+                        *thread_arg = packet;
+                        pthread_t prop_thread;
+                        pthread_create(&prop_thread, NULL, propagate_packet, thread_arg);
+                        pthread_detach(prop_thread);
+                    }
                 }
             } else {
                 output_data[ 1 ] = false;
@@ -477,6 +480,11 @@ void paste_server(
             return;
         }
 
+        case register_neighbor: {
+            // TODO
+        }
+        // break;
+
         default: {
             log(_info, "<%.*s> request type: invalid method\n", (int) ip_size, ip_addr);
 
@@ -611,6 +619,7 @@ void* cleanup_packets(void* arg) {
         uint64_t current_time_ms = get_current_time_ms();
         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
@@ -633,6 +642,23 @@ void* cleanup_packets(void* arg) {
             }
             pthread_mutex_unlock(&packet_storage_mutex);
         }
+
+        DIR* dir = opendir("packets");
+        if (dir) {
+            struct dirent* entry;
+            while ((entry = readdir(dir)) != NULL) {
+                if (entry->d_type == DT_REG) {
+                    char filepath[ 256 ];
+                    snprintf(filepath, sizeof(filepath), "packets/%s", entry->d_name);
+                    struct stat st;
+                    if (stat(filepath, &st) == 0
+                        && current_time_ms - (uint64_t) st.st_mtime * 1000 > retention_ms) {
+                        if (remove(filepath) != 0) log(_error, "failed to delete %s\n", filepath);
+                    }
+                }
+            }
+            closedir(dir);
+        }
     }
 
     return NULL;