1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
|
#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);
void firehose_push(hash new_hash, uint64_t storage_index);
uint16_t firehose_get_count(void);
uint64_t firehose_get_total_packets(void);
uint16_t firehose_get_packets(uint64_t offset, uint16_t count, water_splash* output_entries);
void firehose_init(void) {
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(void) { return firehose_buffer.count; }
uint64_t firehose_get_total_packets(void) { 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;
pthread_mutex_lock(&packet_storage_mutex);
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(&packet_storage_mutex);
pthread_mutex_unlock(&firehose_buffer.mutex);
return copied_count;
}
|
