Line data Source code
1 : #include "fd_tpu.h"
2 : #include "fd_tpu_reasm_private.h"
3 : #include "../../tango/dcache/fd_dcache.h"
4 : #include "../../tango/mcache/fd_mcache.h"
5 :
6 : FD_FN_CONST ulong
7 81 : fd_tpu_reasm_align( void ) {
8 81 : return alignof(fd_tpu_reasm_t);
9 81 : }
10 :
11 : FD_FN_CONST ulong
12 : fd_tpu_reasm_footprint( ulong depth,
13 30 : ulong burst ) {
14 :
15 30 : if( FD_UNLIKELY(
16 30 : ( fd_ulong_popcnt( depth )!=1 ) |
17 30 : ( depth>0x7fffffffUL ) |
18 30 : ( burst<2 ) |
19 30 : ( burst>0x7fffffffUL ) ) )
20 15 : return 0UL;
21 :
22 15 : ulong slot_cnt = depth+burst;
23 15 : ulong chain_cnt = fd_tpu_reasm_map_chain_cnt_est( slot_cnt );
24 15 : return FD_LAYOUT_FINI( FD_LAYOUT_APPEND( FD_LAYOUT_APPEND( FD_LAYOUT_APPEND( FD_LAYOUT_APPEND( FD_LAYOUT_INIT,
25 30 : fd_tpu_reasm_align(), sizeof(fd_tpu_reasm_t) ), /* hdr */
26 30 : alignof(uint), depth *sizeof(uint) ), /* pub_slots */
27 30 : alignof(fd_tpu_reasm_slot_t), slot_cnt*sizeof(fd_tpu_reasm_slot_t) ), /* slots */
28 30 : fd_tpu_reasm_map_align(), fd_tpu_reasm_map_footprint( chain_cnt ) ), /* map */
29 30 : fd_tpu_reasm_align() );
30 :
31 30 : }
32 :
33 : void *
34 : fd_tpu_reasm_new( void * shmem,
35 : ulong depth,
36 : ulong burst,
37 : ulong seed,
38 : ulong orig,
39 6 : void * dcache ) {
40 :
41 6 : if( FD_UNLIKELY( !shmem ) ) return NULL;
42 6 : if( FD_UNLIKELY( !fd_ulong_is_aligned( (ulong)shmem, alignof(fd_tpu_reasm_t) ) ) ) return NULL;
43 6 : if( FD_UNLIKELY( !fd_tpu_reasm_footprint( depth, burst ) ) ) return NULL;
44 6 : if( FD_UNLIKELY( orig > FD_FRAG_META_ORIG_MAX ) ) return NULL;
45 :
46 6 : ulong req_data_sz = fd_tpu_reasm_req_data_sz( depth, burst );
47 6 : if( FD_UNLIKELY( fd_dcache_data_sz( dcache )<req_data_sz ) ) {
48 0 : FD_LOG_WARNING(( "dcache data_sz is too small (need %lu, have %lu)", req_data_sz, fd_dcache_data_sz( dcache ) ));
49 0 : return NULL;
50 0 : }
51 :
52 : /* Memory layout */
53 :
54 6 : ulong slot_cnt = depth+burst;
55 6 : if( FD_UNLIKELY( !slot_cnt ) ) return NULL;
56 6 : ulong chain_cnt = fd_tpu_reasm_map_chain_cnt_est( slot_cnt );
57 :
58 6 : FD_SCRATCH_ALLOC_INIT( l, shmem );
59 6 : fd_tpu_reasm_t * reasm = FD_SCRATCH_ALLOC_APPEND( l, fd_tpu_reasm_align(), sizeof(fd_tpu_reasm_t) );
60 6 : uint * pub_slots = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), depth*sizeof(uint) );
61 6 : fd_tpu_reasm_slot_t * slots = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_tpu_reasm_slot_t), slot_cnt*sizeof(fd_tpu_reasm_slot_t) );
62 6 : void * map_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_tpu_reasm_map_align(), fd_tpu_reasm_map_footprint( chain_cnt ) );
63 6 : FD_SCRATCH_ALLOC_FINI( l, fd_tpu_reasm_align() );
64 :
65 6 : fd_memset( reasm, 0, sizeof(fd_tpu_reasm_t) );
66 6 : fd_memset( slots, 0, burst*sizeof(fd_tpu_reasm_slot_t) );
67 :
68 6 : fd_tpu_reasm_map_t * map = fd_tpu_reasm_map_join( fd_tpu_reasm_map_new( map_mem, chain_cnt, seed ) );
69 6 : if( FD_UNLIKELY( !map ) ) {
70 0 : FD_LOG_WARNING(( "fd_tpu_reasm_map_new failed" ));
71 0 : return NULL;
72 0 : }
73 :
74 : /* Initialize reasm object */
75 :
76 6 : reasm->slots_off = (ulong)( (uchar *)slots - (uchar *)reasm );
77 6 : reasm->pub_slots_off = (ulong)( (uchar *)pub_slots - (uchar *)reasm );
78 6 : reasm->map_off = (ulong)( (uchar *)map - (uchar *)reasm );
79 6 : reasm->dcache = dcache;
80 :
81 6 : reasm->depth = (uint)depth;
82 6 : reasm->burst = (uint)burst;
83 6 : reasm->head = (uint)slot_cnt-1U;
84 6 : reasm->tail = (uint)depth;
85 6 : reasm->slot_cnt = (uint)slot_cnt;
86 6 : reasm->orig = (ushort)orig;
87 :
88 : /* The initial state moves the first 'depth' slots to the mcache (PUB)
89 : and leaves the rest as FREE. */
90 :
91 774 : for( uint j=0U; j<depth; j++ ) {
92 768 : fd_tpu_reasm_slot_t * slot = slots + j;
93 768 : slot->k.state = FD_TPU_REASM_STATE_PUB;
94 768 : slot->k.conn_uid = ULONG_MAX;
95 768 : slot->k.stream_id = 0xffffffffffff;
96 768 : slot->k.sz = 0;
97 768 : slot->chain_next = UINT_MAX;
98 768 : pub_slots[ j ] = j;
99 768 : }
100 774 : for( uint j=(uint)depth; j<slot_cnt; j++ ) {
101 768 : fd_tpu_reasm_slot_t * slot = slots + j;
102 768 : slot->k.state = FD_TPU_REASM_STATE_FREE;
103 768 : slot->k.conn_uid = ULONG_MAX;
104 768 : slot->k.stream_id = 0xffffffffffff;
105 768 : slot->k.sz = 0;
106 768 : slot->lru_prev = fd_uint_if( j<slot_cnt-1U, j+1U, UINT_MAX );
107 768 : slot->lru_next = fd_uint_if( j>depth, j-1U, UINT_MAX );
108 768 : slot->chain_next = UINT_MAX;
109 768 : }
110 :
111 : /* Clear the entire hash map */
112 :
113 6 : uint * chains = fd_tpu_reasm_map_private_chain( map );
114 774 : for( uint j=0U; j<chain_cnt; j++ ) {
115 768 : chains[ j ] = UINT_MAX;
116 768 : }
117 :
118 6 : FD_COMPILER_MFENCE();
119 6 : reasm->magic = FD_TPU_REASM_MAGIC;
120 6 : FD_COMPILER_MFENCE();
121 :
122 6 : return reasm;
123 6 : }
124 :
125 : fd_tpu_reasm_t *
126 6 : fd_tpu_reasm_join( void * shreasm ) {
127 6 : fd_tpu_reasm_t * reasm = shreasm;
128 6 : if( FD_UNLIKELY( reasm->magic != FD_TPU_REASM_MAGIC ) ) {
129 0 : FD_LOG_WARNING(( "bad magic" ));
130 0 : return NULL;
131 0 : }
132 6 : return reasm;
133 6 : }
134 :
135 : void *
136 3 : fd_tpu_reasm_leave( fd_tpu_reasm_t * reasm ) {
137 3 : return reasm;
138 3 : }
139 :
140 : void *
141 3 : fd_tpu_reasm_delete( void * shreasm ) {
142 3 : fd_tpu_reasm_t * reasm = shreasm;
143 3 : if( FD_UNLIKELY( !reasm ) ) return NULL;
144 3 : reasm->magic = 0UL;
145 3 : return shreasm;
146 3 : }
147 :
148 : fd_tpu_reasm_slot_t *
149 : fd_tpu_reasm_query( fd_tpu_reasm_t * reasm,
150 : ulong conn_uid,
151 76509 : ulong stream_id ) {
152 76509 : return smap_query( reasm, conn_uid, stream_id );
153 76509 : }
154 :
155 : fd_tpu_reasm_slot_t *
156 : fd_tpu_reasm_prepare( fd_tpu_reasm_t * reasm,
157 : ulong conn_uid,
158 : ulong stream_id,
159 76125 : long first_seen_nanos ) {
160 76125 : fd_tpu_reasm_slot_t * slot = slotq_pop_tail( reasm );
161 76125 : smap_remove( reasm, slot );
162 76125 : slot_begin( slot );
163 76125 : slotq_push_head( reasm, slot );
164 76125 : slot->k.conn_uid = conn_uid;
165 76125 : slot->k.stream_id = stream_id & FD_TPU_REASM_SID_MASK;
166 76125 : smap_insert( reasm, slot );
167 76125 : reasm->dcache[ slot_get_idx( reasm, slot ) ].hdr.first_seen_nanos = first_seen_nanos;
168 76125 : return slot;
169 76125 : }
170 :
171 : int
172 : fd_tpu_reasm_frag( fd_tpu_reasm_t * reasm,
173 : fd_tpu_reasm_slot_t * slot,
174 : uchar const * data,
175 : ulong data_sz,
176 65841 : ulong data_off ) {
177 :
178 65841 : if( FD_UNLIKELY( slot->k.state != FD_TPU_REASM_STATE_BUSY ) )
179 0 : return FD_TPU_REASM_ERR_STATE;
180 :
181 65841 : ulong slot_idx = slot_get_idx( reasm, slot );
182 65841 : ulong sz0 = slot->k.sz;
183 :
184 65841 : if( FD_UNLIKELY( data_off>sz0 ) ) {
185 0 : return FD_TPU_REASM_ERR_SKIP;
186 0 : }
187 :
188 65841 : if( FD_UNLIKELY( data_off<sz0 ) ) {
189 : /* Fragment partially known ... should not happen */
190 0 : ulong skip = sz0 - data_off;
191 0 : if( skip>data_sz ) return FD_TPU_REASM_SUCCESS;
192 0 : data_off += skip;
193 0 : data_sz -= skip;
194 0 : data += skip;
195 0 : }
196 :
197 65841 : ulong sz1 = sz0 + data_sz;
198 65841 : if( FD_UNLIKELY( (sz1<sz0)|(sz1>FD_TPU_MTU) ) ) {
199 0 : fd_tpu_reasm_cancel( reasm, slot );
200 0 : return FD_TPU_REASM_ERR_SZ;
201 0 : }
202 :
203 65841 : fd_memcpy( reasm->dcache[ slot_idx ].payload + sz0, data, data_sz );
204 :
205 65841 : slot->k.sz = (ushort)( sz1 & FD_TPU_REASM_SZ_MASK );
206 65841 : return FD_TPU_REASM_SUCCESS;
207 65841 : }
208 :
209 : int
210 : fd_tpu_reasm_publish( fd_tpu_reasm_t * reasm,
211 : fd_tpu_reasm_slot_t * slot,
212 : fd_frag_meta_t * mcache,
213 : void * base, /* Assumed aligned FD_CHUNK_ALIGN */
214 : ulong seq,
215 : long tspub,
216 : uint source_ipv4,
217 65841 : uchar source_tpu ) {
218 :
219 65841 : ulong depth = reasm->depth;
220 :
221 65841 : if( FD_UNLIKELY( slot->k.state != FD_TPU_REASM_STATE_BUSY ) )
222 0 : return FD_TPU_REASM_ERR_STATE;
223 :
224 : /* Derive chunk index */
225 65841 : uint slot_idx = slot_get_idx( reasm, slot );
226 65841 : fd_tpu_msg_t * buf = &reasm->dcache[ slot_idx ];
227 65841 : ulong chunk = fd_laddr_to_chunk( base, buf );
228 65841 : if( FD_UNLIKELY( ( (ulong)buf<(ulong)base ) |
229 65841 : ( chunk>UINT_MAX ) ) ) {
230 0 : FD_LOG_CRIT(( "invalid base %p for slot %p in tpu_reasm %p",
231 0 : base, (void *)slot, (void *)reasm ));
232 0 : }
233 :
234 : /* Find least recently published slot. This is our "freed slot".
235 : (Every time a new slot is published, another slot is simultaneously
236 : freed) */
237 65841 : uint * pub_slot = fd_tpu_reasm_pub_slots_laddr( reasm ) + fd_mcache_line_idx( seq, depth );
238 65841 : uint freed_slot_idx = *pub_slot;
239 65841 : FD_CHECK_CRIT( freed_slot_idx < reasm->slot_cnt, "corruption detected" );
240 :
241 : /* Publish to mcache */
242 65841 : ulong sz = slot->k.sz;
243 65841 : ulong ctl = fd_frag_meta_ctl( reasm->orig, 1, 1, 0 );
244 65841 : ulong tspub_comp = fd_frag_meta_ts_comp( tspub );
245 :
246 65841 : fd_txn_m_t * txnm = &buf->hdr;
247 65841 : long first_seen_nanos = txnm->first_seen_nanos;
248 65841 : *txnm = (fd_txn_m_t){0};
249 65841 : txnm->payload_sz = (ushort)sz;
250 65841 : txnm->first_seen_nanos = first_seen_nanos;
251 65841 : txnm->source_ipv4 = source_ipv4;
252 65841 : txnm->source_tpu = source_tpu;
253 :
254 65841 : # if FD_HAS_AVX
255 65841 : fd_mcache_publish_avx( mcache, depth, seq, 0UL, chunk, fd_txn_m_realized_footprint( txnm, 0, 0 ), ctl, 0UL, tspub_comp );
256 : # elif FD_HAS_SSE
257 : fd_mcache_publish_sse( mcache, depth, seq, 0UL, chunk, fd_txn_m_realized_footprint( txnm, 0, 0 ), ctl, 0UL, tspub_comp );
258 : # else
259 : fd_mcache_publish ( mcache, depth, seq, 0UL, chunk, fd_txn_m_realized_footprint( txnm, 0, 0 ), ctl, 0UL, tspub_comp );
260 : # endif
261 :
262 : /* Mark new slot as published */
263 65841 : slotq_remove( reasm, slot );
264 65841 : slot->k.state = FD_TPU_REASM_STATE_PUB;
265 65841 : *pub_slot = slot_idx;
266 :
267 : /* Free oldest published slot */
268 65841 : fd_tpu_reasm_slot_t * free_slot = fd_tpu_reasm_slots_laddr( reasm ) + freed_slot_idx;
269 65841 : uint free_slot_state = free_slot->k.state;
270 65841 : FD_CHECK_CRIT( free_slot_state == FD_TPU_REASM_STATE_PUB, "corruption detected" );
271 65841 : free_slot->k.state = FD_TPU_REASM_STATE_FREE;
272 65841 : slotq_push_tail( reasm, free_slot );
273 :
274 65841 : return FD_TPU_REASM_SUCCESS;
275 65841 : }
276 :
277 : void
278 : fd_tpu_reasm_cancel( fd_tpu_reasm_t * reasm,
279 9348 : fd_tpu_reasm_slot_t * slot ) {
280 9348 : if( FD_UNLIKELY( slot->k.state != FD_TPU_REASM_STATE_BUSY ) ) return;
281 9348 : slotq_remove( reasm, slot );
282 9348 : smap_remove( reasm, slot );
283 9348 : slot->k.state = FD_TPU_REASM_STATE_FREE;
284 9348 : slot->k.conn_uid = ULONG_MAX;
285 9348 : slot->k.stream_id = 0UL;
286 9348 : slotq_push_tail( reasm, slot );
287 9348 : }
288 :
289 : int
290 : fd_tpu_reasm_publish_fast( fd_tpu_reasm_t * reasm,
291 : uchar const * data,
292 : ulong sz,
293 : fd_frag_meta_t * mcache,
294 : void * base, /* Assumed aligned FD_CHUNK_ALIGN */
295 : ulong seq,
296 : long tspub,
297 : uint source_ipv4,
298 0 : uchar source_tpu ) {
299 :
300 0 : ulong depth = reasm->depth;
301 0 : if( FD_UNLIKELY( sz>FD_TPU_MTU ) ) return FD_TPU_REASM_ERR_SZ;
302 :
303 : /* Acquire least recent slot. This is our "new slot" */
304 0 : fd_tpu_reasm_slot_t * slot = slotq_pop_tail( reasm );
305 0 : smap_remove( reasm, slot );
306 0 : slot_begin( slot );
307 :
308 : /* Derive buffer address of new slot */
309 0 : uint slot_idx = slot_get_idx( reasm, slot );
310 0 : fd_tpu_msg_t * buf = &reasm->dcache[ slot_idx ];
311 0 : ulong chunk = fd_laddr_to_chunk( base, buf );
312 0 : if( FD_UNLIKELY( ( (ulong)buf<(ulong)base ) |
313 0 : ( chunk>UINT_MAX ) ) ) {
314 0 : FD_LOG_ERR(( "Computed invalid chunk index (base=%p buf=%p chunk=%lx)",
315 0 : base, (void *)buf, chunk ));
316 0 : }
317 :
318 : /* Find least recently published slot. This is our "freed slot".
319 : (Every time a new slot is published, another slot is simultaneously
320 : freed) */
321 0 : uint * pub_slot = fd_tpu_reasm_pub_slots_laddr( reasm ) + fd_mcache_line_idx( seq, depth );
322 0 : uint freed_slot_idx = *pub_slot;
323 0 : FD_CHECK_CRIT( freed_slot_idx < reasm->slot_cnt, "corruption detected" );
324 :
325 : /* Copy data into new slot */
326 0 : FD_COMPILER_MFENCE();
327 0 : slot->k.sz = sz & FD_TPU_REASM_SZ_MASK;
328 0 : fd_txn_m_t * txnm = &buf->hdr;
329 0 : *txnm = (fd_txn_m_t){0};
330 0 : txnm->payload_sz = (ushort)slot->k.sz;
331 0 : txnm->first_seen_nanos = tspub; /* single packet: arrival==publish */
332 0 : txnm->source_ipv4 = source_ipv4;
333 0 : txnm->source_tpu = source_tpu;
334 0 : fd_memcpy( buf->payload, data, sz );
335 0 : FD_COMPILER_MFENCE();
336 0 : slot->k.state = FD_TPU_REASM_STATE_PUB;
337 0 : FD_COMPILER_MFENCE();
338 :
339 : /* Publish new slot, while simultaneously removing all references to
340 : the old slot */
341 0 : *pub_slot = slot_idx;
342 0 : ulong ctl = fd_frag_meta_ctl( reasm->orig, 1, 1, 0 );
343 0 : uint tspub_comp = (uint)fd_frag_meta_ts_comp( tspub );
344 0 : # if FD_HAS_AVX
345 0 : fd_mcache_publish_avx( mcache, depth, seq, 0UL, chunk, fd_txn_m_realized_footprint( txnm, 0, 0 ), ctl, 0UL, tspub_comp );
346 : # elif FD_HAS_SSE
347 : fd_mcache_publish_sse( mcache, depth, seq, 0UL, chunk, fd_txn_m_realized_footprint( txnm, 0, 0 ), ctl, 0UL, tspub_comp );
348 : # else
349 : fd_mcache_publish ( mcache, depth, seq, 0UL, chunk, fd_txn_m_realized_footprint( txnm, 0, 0 ), ctl, 0UL, tspub_comp );
350 : # endif
351 :
352 : /* Free old slot */
353 0 : fd_tpu_reasm_slot_t * free_slot = fd_tpu_reasm_slots_laddr( reasm ) + freed_slot_idx;
354 0 : uint free_slot_state = free_slot->k.state;
355 0 : FD_CHECK_CRIT( free_slot_state == FD_TPU_REASM_STATE_PUB, "corruption detected" );
356 0 : free_slot->k.state = FD_TPU_REASM_STATE_FREE;
357 0 : slotq_push_tail( reasm, free_slot );
358 0 : return FD_TPU_REASM_SUCCESS;
359 0 : }
|
