starting commit

1 files changed, 94 insertions, 0 deletions

diff --git a/firehose.c b/firehose.c
new file mode 100644
index 0000000..240dfd8
--- /dev/null
+++ b/firehose.c
@@ -0,0 +1,94 @@
+#include "common.c"
+#include <pthread.h>
+#include <stdbool.h>
+#include <stddef.h>
+#include <stdint.h>
+#include <stdio.h>
+#include <string.h>
+
+#define FIREHOSE_BUFFER_SIZE 1 << 14
+
+typedef struct {
+    hash     packet_hash;
+    uint64_t storage_index;
+} water_splash;
+
+typedef struct {
+    water_splash    entries[ FIREHOSE_BUFFER_SIZE ];
+    uint16_t        head;
+    uint16_t        tail;
+    uint16_t        count;
+    uint64_t        total_packets;
+    pthread_mutex_t mutex;
+} fire_buffer;
+
+static fire_buffer firehose_buffer;
+
+void     firehose_init();
+void     firehose_push(hash new_hash, uint64_t storage_index);
+uint16_t firehose_get_count();
+uint64_t firehose_get_total_packets();
+uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* output_entries);
+
+void firehose_init() {
+    memset(firehose_buffer.entries, 0, sizeof(firehose_buffer.entries));
+    firehose_buffer.head          = 0;
+    firehose_buffer.tail          = 0;
+    firehose_buffer.count         = 0;
+    firehose_buffer.total_packets = 0;
+    pthread_mutex_init(&firehose_buffer.mutex, NULL);
+}
+
+uint64_t current_base_offset = 0;
+
+void firehose_push(hash new_hash, uint64_t storage_index) {
+    pthread_mutex_lock(&firehose_buffer.mutex);
+
+    memcpy(firehose_buffer.entries[ firehose_buffer.tail ].packet_hash, new_hash, sizeof(hash));
+    firehose_buffer.entries[ firehose_buffer.tail ].storage_index = storage_index;
+    if (firehose_buffer.count == FIREHOSE_BUFFER_SIZE) {
+        firehose_buffer.head = (firehose_buffer.head + 1) % FIREHOSE_BUFFER_SIZE;
+        current_base_offset++;
+    } else {
+        firehose_buffer.count++;
+    }
+    firehose_buffer.tail = (firehose_buffer.tail + 1) % FIREHOSE_BUFFER_SIZE;
+
+    firehose_buffer.total_packets++;
+
+    pthread_mutex_unlock(&firehose_buffer.mutex);
+}
+
+uint16_t firehose_get_count() { return firehose_buffer.count; }
+uint64_t firehose_get_total_packets() { return firehose_buffer.total_packets; }
+
+uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* output_entries) {
+    pthread_mutex_lock(&firehose_buffer.mutex);
+
+    uint16_t copied_count = 0;
+    uint64_t start_offset = firehose_buffer.total_packets - firehose_buffer.count;
+
+    if (offset < start_offset || offset >= firehose_buffer.total_packets) {
+        pthread_mutex_unlock(&firehose_buffer.mutex);
+        return 0;
+    }
+
+    uint64_t relative_offset = offset - current_base_offset;
+    if (offset < current_base_offset || relative_offset >= firehose_buffer.count) {
+        pthread_mutex_unlock(&firehose_buffer.mutex);
+        return 0; // offset too old or beyond current entries
+    }
+    uint16_t start_index = (firehose_buffer.head + relative_offset) % FIREHOSE_BUFFER_SIZE;
+
+    for (int i = 0; i < count; i++) {
+        if (relative_offset + i >= firehose_buffer.count) { break; }
+
+        uint16_t current_index = (start_index + i) % FIREHOSE_BUFFER_SIZE;
+        memcpy(
+            &output_entries[ i ], &firehose_buffer.entries[ current_index ], sizeof(water_splash));
+        copied_count++;
+    }
+
+    pthread_mutex_unlock(&firehose_buffer.mutex);
+    return copied_count;
+}