Line data Source code
1 : #include "fd_gui_shred.h"
2 : #include "fd_gui.h"
3 :
4 0 : #define FD_GUI_SHRED_LITERAL (0U)
5 0 : #define FD_GUI_SHRED_DELTA (1U)
6 0 : #define FD_GUI_SHRED_TS_NEG (2U)
7 0 : #define FD_GUI_SHRED_IDX_NEG (4U)
8 0 : #define FD_GUI_SHRED_EVENT_SHIFT (3U)
9 :
10 : struct __attribute__((packed)) fd_gui_shred_delta {
11 : uint insert_delta;
12 : uint timestamp_delta : 24;
13 : uchar idx_delta;
14 : };
15 : typedef struct fd_gui_shred_delta fd_gui_shred_delta_t;
16 :
17 : FD_STATIC_ASSERT( sizeof(fd_gui_shred_delta_t)==8UL, shred_delta_sz );
18 : FD_STATIC_ASSERT( sizeof(fd_gui_shred_batch_t)==FD_GUI_SHRED_BATCH_SZ, shred_batch_sz );
19 :
20 : static ulong
21 0 : shred_window( long ns ) {
22 0 : return ns<=0L ? 0UL : (ulong)(ns/FD_GUI_HIST_RES_1S_NS);
23 0 : }
24 :
25 : static int
26 : batch_append( fd_gui_shred_builder_t * builder,
27 0 : fd_gui_shred_event_t const * event ) {
28 0 : fd_gui_shred_batch_t * batch = &builder->batch;
29 0 : fd_gui_shred_event_t const * prev = &builder->prev;
30 0 : ulong ts_delta = event->event_time_ns>=prev->event_time_ns
31 0 : ? (ulong)event->event_time_ns-(ulong)prev->event_time_ns
32 0 : : (ulong)prev->event_time_ns-(ulong)event->event_time_ns;
33 0 : ulong idx_delta = event->idx>=prev->idx ? (ulong)(event->idx-prev->idx) : (ulong)(prev->idx-event->idx);
34 0 : int delta = batch->event_cnt && event->slot==prev->slot && ts_delta<=FD_GUI_SHRED_EVENT_TS_MAX && idx_delta<=UCHAR_MAX &&
35 0 : event->event<=(UCHAR_MAX>>FD_GUI_SHRED_EVENT_SHIFT);
36 0 : ulong sz = 1UL + (delta ? sizeof(fd_gui_shred_delta_t) : sizeof(fd_gui_shred_event_t));
37 0 : if( FD_UNLIKELY( sz>sizeof(batch->data)-batch->data_sz ) ) return 0;
38 :
39 0 : uchar * dst = batch->data+batch->data_sz;
40 0 : if( FD_LIKELY( delta ) ) {
41 0 : *dst++ = (uchar)(FD_GUI_SHRED_DELTA |
42 0 : (event->event_time_ns<prev->event_time_ns ? FD_GUI_SHRED_TS_NEG : 0U) |
43 0 : (event->idx<prev->idx ? FD_GUI_SHRED_IDX_NEG : 0U) |
44 0 : ((uint)event->event<<FD_GUI_SHRED_EVENT_SHIFT));
45 0 : fd_gui_shred_delta_t d = {
46 0 : .insert_delta = (uint)(event->insert_time_ns-prev->insert_time_ns),
47 0 : .timestamp_delta = (uint)(ts_delta & FD_GUI_SHRED_EVENT_TS_MAX),
48 0 : .idx_delta = (uchar)idx_delta
49 0 : };
50 0 : fd_memcpy( dst, &d, sizeof(d) );
51 0 : } else {
52 0 : *dst++ = FD_GUI_SHRED_LITERAL;
53 0 : fd_memcpy( dst, event, sizeof(*event) );
54 0 : }
55 0 : batch->data_sz += (uint)sz;
56 0 : batch->event_cnt++;
57 0 : builder->prev = *event;
58 0 : return 1;
59 0 : }
60 :
61 : static void
62 0 : batch_next( fd_gui_shred_event_iter_t * iter ) {
63 0 : uchar const * src = iter->_batch->data+iter->_data_off;
64 0 : uchar tag = *src++;
65 0 : if( FD_LIKELY( tag & FD_GUI_SHRED_DELTA ) ) {
66 0 : fd_gui_shred_delta_t d;
67 0 : fd_memcpy( &d, src, sizeof(d) );
68 0 : iter->event.insert_time_ns += (long)d.insert_delta;
69 0 : iter->event.event_time_ns += (tag & FD_GUI_SHRED_TS_NEG) ? -(long)d.timestamp_delta : (long)d.timestamp_delta;
70 0 : iter->event.idx = (ushort)((tag & FD_GUI_SHRED_IDX_NEG) ? iter->event.idx-d.idx_delta : iter->event.idx+d.idx_delta);
71 0 : iter->event.event = (uchar)(tag>>FD_GUI_SHRED_EVENT_SHIFT);
72 0 : iter->_data_off += 1UL+sizeof(d);
73 0 : } else {
74 0 : fd_memcpy( &iter->event, src, sizeof(iter->event) );
75 0 : iter->_data_off += 1UL+sizeof(iter->event);
76 0 : }
77 0 : iter->_batch_idx++;
78 0 : }
79 :
80 : static fd_gui_shred_event_iter_t *
81 : shred_iter_begin( fd_gui_t * gui,
82 : fd_gui_shred_event_iter_t * iter,
83 : int event_time,
84 : long after_ns,
85 0 : long before_ns ) {
86 0 : fd_memset( iter, 0, sizeof(*iter) );
87 0 : iter->_gui = gui;
88 0 : iter->_after_ns = after_ns;
89 0 : iter->_before_ns = before_ns;
90 0 : iter->_event_time = event_time;
91 :
92 : /* Event-time queries widen the scan by +-1 second. The assumption is that
93 : event-ordering approximately equals insert-ordering, and in rare
94 : cases where there is more than a second of skew in the system we
95 : can tolerate missing data. */
96 0 : if( FD_LIKELY( gui->db && gui->hist && after_ns<=before_ns ) ) {
97 0 : long margin = event_time ? FD_GUI_HIST_RES_1S_NS : 0L;
98 0 : long lo_ns = fd_long_max( 0L, fd_long_sat_sub( after_ns, margin ) );
99 0 : long hi_ns = fd_long_min( LONG_MAX-1L, fd_long_sat_add( before_ns, margin ) );
100 0 : iter->_hist_active = !fd_gui_hist_range_begin( gui, &iter->_hist_iter, FD_GUI_HIST_SHRED_EVENTS, lo_ns, hi_ns, NULL, NULL );
101 0 : iter->_builder_pending = 1;
102 0 : }
103 0 : return iter;
104 0 : }
105 :
106 : fd_gui_shred_event_iter_t *
107 : fd_gui_shred_event_hist_iter_begin( fd_gui_t * gui,
108 : fd_gui_shred_event_iter_t * iter,
109 : long after_ns,
110 0 : long before_ns ) {
111 0 : return shred_iter_begin( gui, iter, 1, after_ns, before_ns );
112 0 : }
113 :
114 : fd_gui_shred_event_iter_t *
115 : fd_gui_shred_event_iter_begin( fd_gui_t * gui,
116 : fd_gui_shred_event_iter_t * iter,
117 : long after_ns,
118 0 : long before_ns ) {
119 0 : return shred_iter_begin( gui, iter, 0, fd_long_max( after_ns, 0L ), fd_long_min( before_ns, LONG_MAX-1L ) );
120 0 : }
121 :
122 : int
123 0 : fd_gui_shred_event_iter_next( fd_gui_shred_event_iter_t * iter ) {
124 0 : while( iter->_batch || iter->_hist_active || iter->_builder_pending ) {
125 0 : if( iter->_batch && iter->_batch_idx<iter->_batch->event_cnt ) {
126 0 : batch_next( iter );
127 0 : long ts = iter->_event_time ? iter->event.event_time_ns : iter->event.insert_time_ns;
128 0 : if( ts>=iter->_after_ns && (iter->_event_time ? ts<iter->_before_ns : ts<=iter->_before_ns) ) return 1;
129 0 : continue;
130 0 : }
131 0 : iter->_batch = NULL;
132 0 : iter->_batch_idx = 0UL;
133 0 : iter->_data_off = 0UL;
134 0 : if( iter->_hist_active && fd_gui_hist_range_next( &iter->_hist_iter ) ) {
135 0 : iter->_batch = iter->_hist_iter.rec;
136 0 : continue;
137 0 : }
138 0 : if( iter->_hist_active ) fd_gui_hist_range_end( &iter->_hist_iter );
139 0 : iter->_hist_active = 0;
140 0 : if( iter->_builder_pending ) {
141 0 : iter->_builder_pending = 0;
142 0 : iter->_batch = &iter->_gui->shreds.builder.batch;
143 0 : }
144 0 : }
145 0 : return 0;
146 0 : }
147 :
148 : void
149 0 : fd_gui_shred_event_iter_end( fd_gui_shred_event_iter_t * iter ) {
150 0 : if( iter->_hist_active ) fd_gui_hist_range_end( &iter->_hist_iter );
151 0 : iter->_hist_active = 0;
152 0 : iter->_builder_pending = 0;
153 0 : iter->_batch = NULL;
154 0 : }
155 :
156 : static void
157 0 : batch_flush( fd_gui_t * gui ) {
158 0 : fd_gui_shred_batch_t * batch = &gui->shreds.builder.batch;
159 0 : if( !batch->event_cnt ) return;
160 0 : if( FD_UNLIKELY( fd_gui_hist_ts_append( gui, FD_GUI_HIST_SHRED_EVENTS, batch ) ) ) {
161 0 : gui->shreds.dropped_event_cnt += batch->event_cnt;
162 0 : }
163 0 : batch->event_cnt = 0U;
164 0 : batch->data_sz = 0U;
165 0 : }
166 :
167 : void
168 : fd_gui_shred_event_append( fd_gui_t * gui,
169 : ulong slot,
170 : ulong idx,
171 : uchar event,
172 : long timestamp,
173 0 : long now ) {
174 0 : fd_gui_shred_builder_t * builder = &gui->shreds.builder;
175 0 : if( FD_UNLIKELY( !gui->db || !gui->hist || slot>UINT_MAX || idx>USHORT_MAX ||
176 0 : (idx==USHORT_MAX && event!=FD_GUI_SLOT_SHRED_SHRED_SLOT_COMPLETE) ||
177 0 : now<builder->prev.insert_time_ns ) ) {
178 0 : gui->shreds.dropped_event_cnt++;
179 0 : return;
180 0 : }
181 0 : fd_gui_shred_batch_t * batch = &builder->batch;
182 0 : long window_ns = (long)shred_window( now )*FD_GUI_HIST_RES_1S_NS;
183 0 : if( batch->event_cnt && batch->insert_time_ns!=window_ns ) batch_flush( gui );
184 0 : batch->insert_time_ns = window_ns;
185 0 : fd_gui_shred_event_t rec = { .insert_time_ns=now, .event_time_ns=timestamp, .slot=(uint)slot, .idx=(ushort)idx, .event=event };
186 0 : if( FD_UNLIKELY( !batch_append( builder, &rec ) ) ) {
187 0 : batch_flush( gui );
188 0 : FD_TEST( batch_append( builder, &rec ) );
189 0 : }
190 0 : }
191 :
192 : int
193 : fd_gui_shred_flush( fd_gui_t * gui,
194 0 : long now ) {
195 0 : if( FD_UNLIKELY( !gui->db || !gui->hist ) ) return 0;
196 0 : fd_gui_shred_batch_t const * batch = &gui->shreds.builder.batch;
197 0 : if( !batch->event_cnt || shred_window( batch->insert_time_ns )>=shred_window( now ) ) return 0;
198 0 : batch_flush( gui );
199 0 : return 1;
200 0 : }
201 :
202 : int
203 : fd_gui_event_bounds( fd_gui_t * gui,
204 : int dbi,
205 : long * lo,
206 0 : long * hi ) {
207 0 : long first;
208 0 : long last;
209 0 : int have = gui->db && fd_gui_store_ts_live_timestamp_bounds( gui->db, (ulong)dbi, &first, &last );
210 0 : if( dbi==FD_GUI_HIST_SHRED_EVENTS && gui->shreds.builder.batch.event_cnt ) {
211 0 : long active = gui->shreds.builder.batch.insert_time_ns;
212 0 : first = have ? fd_long_min( first, active ) : active;
213 0 : last = have ? fd_long_max( last, active ) : active;
214 0 : have = 1;
215 0 : }
216 :
217 0 : if( have && dbi==FD_GUI_HIST_SHRED_EVENTS ) last = fd_long_sat_add( last, FD_GUI_HIST_RES_1S_NS-1L );
218 0 : *lo = have ? fd_long_max( 0L, fd_long_sat_sub( first, FD_GUI_HIST_RES_1S_NS ) ) : LONG_MAX;
219 0 : *hi = have ? fd_long_min( LONG_MAX-1L, fd_long_sat_add( last, FD_GUI_HIST_RES_1S_NS+1L ) ) : LONG_MIN;
220 0 : return have;
221 0 : }
222 :
223 : int
224 : fd_gui_shred_window_is_empty( fd_gui_t * gui,
225 : long after_ns,
226 0 : long before_ns ) {
227 0 : fd_gui_shred_event_iter_t it[ 1 ];
228 0 : fd_gui_shred_event_iter_begin( gui, it, after_ns, before_ns );
229 0 : int empty = !fd_gui_shred_event_iter_next( it );
230 0 : fd_gui_shred_event_iter_end( it );
231 0 : return empty;
232 0 : }
|
