fix duplicate + better propagation

2 files changed, 117 insertions, 51 deletions

diff --git a/src/packet_storage.c b/src/packet_storage.c
index 5781661..922cc45 100644
--- a/src/packet_storage.c
+++ b/src/packet_storage.c
@@ -141,17 +141,31 @@ 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 ];
+    char author_hex[65];
     sprintf(author_hex, hash_fmt, hash_ref(author));
 
-    char filename[ 256 ];
+    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);
+    FILE *f = fopen(filename, "rb");
+    if (f) {
+        unsigned char buf[64];
+        while (fread(buf, 1, 64, f) >= 32) {
+            int c = fgetc(f);
+            (void)c;
+            if (memcmp(buf, packet_hash, 32) == 0) {
+                fclose(f);
+                return;
+            }
+        }
+        fclose(f);
+    }
+
+    FILE *fa = fopen(filename, "ab");
+    if (!fa) return;
+    fwrite(packet_hash, 32, 1, fa);
+    fputc('\n', fa);
+    fclose(fa);
 }
 
 void save_recipient_packet_hash(const recipient recipient, const hash packet_hash) {
diff --git a/src/server.c b/src/server.c
index 630df87..7dc7382 100644
--- a/src/server.c
+++ b/src/server.c
@@ -75,6 +75,61 @@ bool add_neighbor(struct relay_neighbor new_neighbor) {
     return false;
 }
 
+struct propagation_stream_data {
+    size_t                send_len;
+    byte*                 send_data;
+    struct relay_neighbor neighbor;
+};
+
+void* call_neighbor_for_propagation(void* arg) {
+    struct propagation_stream_data* data = arg;
+
+    byte*  response_data = NULL;
+    size_t response_len  = 0;
+    int    result        = -1;
+
+    if (data->neighbor.has_ipv4) {
+        char ip_string[ 16 ];
+        sprintf(
+            ip_string,
+            "%d.%d.%d.%d",
+            data->neighbor.v4_addr[ 0 ],
+            data->neighbor.v4_addr[ 1 ],
+            data->neighbor.v4_addr[ 2 ],
+            data->neighbor.v4_addr[ 3 ]);
+
+        result = study(
+            ip_string,
+            data->neighbor.port,
+            data->send_data,
+            data->send_len,
+            &response_data,
+            &response_len);
+    } else if (data->neighbor.has_ipv6) {
+        char ip_string[ 46 ];
+        int  offset = 0;
+        for (int j = 0; j < 8; j++) {
+            uint16_t word
+                = (data->neighbor.v6_addr[ j * 2 ] << 8) | data->neighbor.v6_addr[ j * 2 + 1 ];
+            offset += sprintf(ip_string + offset, "%x%s", word, (j < 7) ? ":" : "");
+        }
+
+        result = study(
+            ip_string,
+            data->neighbor.port,
+            data->send_data,
+            data->send_len,
+            &response_data,
+            &response_len);
+    }
+
+    if (result != 0) { log(_error, "failed to send packet to neighbor"); }
+    if (response_data) { free(response_data); }
+
+    free(arg);
+    return NULL;
+}
+
 void* propagate_packet(void* _packet) {
     struct data_packet packet = *(struct data_packet*) _packet;
 
@@ -82,53 +137,50 @@ void* propagate_packet(void* _packet) {
         if (!decide_packet_propogation_func(packet)) return NULL;
     }
 
-    log(_debug, "propagating packet " hash_fmt, hash_ref(packet.hash));
+    log(_debug, "propagating packet");
 
-    size_t send_len = 1 + sizeof(struct data_packet);
-    byte   send_data[ send_len * sizeof(byte) ];
-    send_data[ 0 ] = push_packet;
+    size_t send_len  = 1 + sizeof(struct data_packet);
+    byte*  send_data = malloc(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 ];
-        if (neighbor.active) {
-            byte*  response_data = NULL;
-            size_t response_len  = 0;
-            int    result        = -1;
-
-            if (neighbor.has_ipv4) {
-                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 ]);
 
-                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;
-                for (int j = 0; j < 8; j++) {
-                    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) ? ":" : "");
-                }
+    uint64_t active_count = 0;
+    for (uint64_t i = 0; i < neighbor_count; i++) {
+        if (neighbor_storage[ i ].active) active_count++;
+    }
 
-                result = study(
-                    ip_string, neighbor.port, send_data, send_len, &response_data, &response_len);
-            }
+    if (active_count == 0) {
+        pthread_mutex_unlock(&neighbor_storage_mutex);
+        free(send_data);
+        free(_packet);
+        return NULL;
+    }
 
-            if (result != 0) { log(_error, "failed to send packet to neighbor"); }
+    pthread_t neighbors_threads[ active_count ];
+    int       sent_counts = 0;
 
-            if (response_data) { free(response_data); }
+    for (uint64_t i = 0; i < neighbor_count; i++) {
+        struct relay_neighbor neighbor = neighbor_storage[ i ];
+        if (neighbor.active) {
+            struct propagation_stream_data* data = malloc(sizeof(struct propagation_stream_data));
+            *data = (struct propagation_stream_data) { .send_data = send_data,
+                                                       .send_len  = send_len,
+                                                       .neighbor  = neighbor };
+            pthread_create(
+                &neighbors_threads[ sent_counts ], NULL, call_neighbor_for_propagation, data);
+            pthread_detach(neighbors_threads[ sent_counts ]);
         }
     }
+
     pthread_mutex_unlock(&neighbor_storage_mutex);
 
+    for (uint64_t i = 0; i < active_count; i++) { pthread_join(neighbors_threads[ i ], NULL); }
+    
+    log(_debug, "done");
+
+    free(send_data);
     free(_packet);
     return NULL;
 }
@@ -424,15 +476,15 @@ void paste_server(
 
                 save_packet(packet);
 
-                if (((float) rand() / (float) RAND_MAX) < 0.75) {
-                    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.75) {
+                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;
             }
@@ -866,7 +918,7 @@ void save_packet(const struct data_packet packet) {
         if (!decide_packet_save_to_disk_func(packet)) return;
     }
 
-    log(_debug, "saving packet " hash_fmt " to disk", hash_ref(packet.hash));
+    log(_debug, "saving packet to disk");
 
     mkdir("packets", 0755);