Line data Source code
1 : #ifndef HEADER_fd_src_disco_quic_fd_tpu_h
2 : #define HEADER_fd_src_disco_quic_fd_tpu_h
3 :
4 : /* fd_tpu provides the server-side of the TPU/QUIC protocol.
5 :
6 : TPU/QUIC is the protocol used to submit transactions to a block
7 : producer. For each txn to be transferred, the client opens a
8 : unidirectional QUIC stream and sends its serialization (see
9 : fd_txn_parse). In the happy case, a txn only requires one packet.
10 :
11 : For txn exceeding MTU size, the txn is fragmented over multiple
12 : packets. For more information, see the specification:
13 : https://github.com/solana-foundation/specs/blob/main/p2p/tpu.md */
14 :
15 : #include "../fd_txn_m.h"
16 : #include "../../tango/fd_tango_base.h"
17 :
18 : /* fd_tpu_msg_t is the payload of a quic_verify frag. */
19 :
20 : struct __attribute__((aligned(FD_CHUNK_ALIGN))) fd_tpu_msg {
21 : fd_txn_m_t hdr;
22 : uchar payload[ FD_TPU_MTU ];
23 : /* No txn_t nor alut info follows */
24 : };
25 :
26 : typedef struct fd_tpu_msg fd_tpu_msg_t;
27 :
28 : /* FD_TPU_REASM_{SUCCESS,ERR_{...}} are error codes. These values are
29 : persisted to logs. Entries should not be renumbered and numeric
30 : values should never be reused. */
31 :
32 131682 : #define FD_TPU_REASM_SUCCESS (0)
33 0 : #define FD_TPU_REASM_ERR_SZ (1) /* oversz msg */
34 0 : #define FD_TPU_REASM_ERR_SKIP (2) /* out-of-order data within QUIC stream */
35 0 : #define FD_TPU_REASM_ERR_STATE (3) /* unexpected slot state */
36 :
37 : /* FD_TPU_REASM_STATE_{...} are reasm slot states */
38 :
39 19469967 : #define FD_TPU_REASM_STATE_FREE ((uchar)0) /* free */
40 19470351 : #define FD_TPU_REASM_STATE_BUSY ((uchar)1) /* active reassembly */
41 19535493 : #define FD_TPU_REASM_STATE_PUB ((uchar)2) /* published */
42 :
43 : /* fd_tpu_reasm_t handles incoming data fragments of TPU/QUIC streams.
44 : Frags are expected to be provided via fd_quic callback. Each
45 : tpu_reasm object may only serve a single fd_quic object. Dispatches
46 : reassembled messages to an mcache.) Should not be persisted.
47 :
48 : ### Flow Control
49 :
50 : fd_tpu_reasm is wired up as follows:
51 :
52 : ┌────────┐ ┌───────┐ ┌────────┐
53 : │ QUIC │ callbacks │ tpu_ │ tango │ sig_ │
54 : │ Server ├───────────► reasm ├───────► verify │
55 : └────────┘ └───────┘ └────────┘
56 :
57 : Neither of the pictured links backpressure. Packet loss occurs if
58 : (1) the QUIC server accepts more concurrent streams than available
59 : reassembly slots. Also if (2) the bank of sig verify tiles is too
60 : slow to keepup with incoming transactions.
61 :
62 : The application should thus adjust the QUIC server to throttle the
63 : concurrent stream count and transaction rate to appropriate levels.
64 : (Via QUIC connection quotas)
65 :
66 : The tpu_reasm MUST be the only writer to the mcache.
67 :
68 : ### Eviction Policy
69 :
70 : Aforementioned case 1 specifically happens whenever the QUIC server
71 : accepts a stream and tpu_reasm doesn't find a free slot. tpu_reasm
72 : hardcodes a FIFO eviction policy to handle this case by cancelling
73 : the least recently prepared reassembly. This also guarantees that
74 : unfragmented transaction never get dropped.
75 :
76 : ### Internals
77 :
78 : fd_tpu_reasm internally manages an array of message reassembly
79 : buffers. Each of these is called a "slot" (fd_tpu_reasm_slot_t).
80 :
81 : Slots are either owned by the reassembly fifo (FREE, BUSY states), or
82 : the mcache (PUB state). The ownership separation prevents in-flight
83 : reassemblies from thrashing data exposed to consumers via the mcache.
84 : (Data races transitioning between reassembly and fifo ownership are
85 : handled by the speculative receive pattern.)
86 :
87 : The lifecycle of a slot is:
88 :
89 : prepare() publish()
90 : ┌─► FREE ───► BUSY ───► PUB ─┐
91 : │ │ │
92 : ▲ ▼ cancel() ▼ implied by a later
93 : │ │ │ publish()/cancel()
94 : └──────◄───────┴──────◄──────┘
95 :
96 : prepare: The transition from FREE to BUSY occurs when a new QUIC
97 : stream is accepted.
98 : cancel: The transition from BUSY to FREE occurs when stream/txn
99 : reassembly is aborted. This can happen for whatever
100 : explicit reason (peer kicked, network error), or implicitly
101 : when prepare() is called but no free slot was found.
102 : publish: The transition from BUSY to PUB occurs when a slot holding
103 : a complete txn is made visible to downstream consumers.
104 : This moves a slot from the reassembly fifo to the mcache.
105 :
106 : The transition from PUB to FREE also occurs at the same time (for a
107 : different slot). This moves the least recently published slot from
108 : the mcache into the reassembly fifo. This keeps the number of slots
109 : owned by the mcache at _exactly_ depth at all times and exactly
110 : mirroring the set of packets exposed downstream (notwithstanding a
111 : startup transient of up to depth packets). This also guarantees that
112 : the number of slots in the FREE and BUSY states is kept at _exactly_
113 : reasm_max at all times.
114 :
115 : In order to support the above, the 'pub_slots' lookup table tracks
116 : which published mcache lines (indexed by `seq % depth`) correspond to
117 : which slot indexes. */
118 :
119 :
120 : /* fd_tpu_reasm_slot_t holds a message reassembly buffer.
121 : Carefully tuned to 32 byte size. */
122 :
123 : struct fd_tpu_reasm_key {
124 : ulong conn_uid; /* ULONG_MAX means invalid */
125 : ulong stream_id : 48;
126 : ulong sz : 14; /* size of the txn payload data. does not
127 : include the sizeof(fd_txn_m_t) bytes that
128 : precedes the payload in each slot. */
129 : ulong state : 2;
130 : };
131 :
132 19547799 : #define FD_TPU_REASM_SID_MASK (0xffffffffffffUL)
133 65841 : #define FD_TPU_REASM_SZ_MASK (0x3fffUL)
134 :
135 : typedef struct fd_tpu_reasm_key fd_tpu_reasm_key_t;
136 :
137 : struct __attribute__((aligned(16))) fd_tpu_reasm_slot {
138 : fd_tpu_reasm_key_t k; /* FIXME ugly: the compound key has to be a single struct member */
139 : uint lru_prev;
140 : uint lru_next;
141 : uint chain_next;
142 : };
143 :
144 : typedef struct fd_tpu_reasm_slot fd_tpu_reasm_slot_t;
145 :
146 : struct fd_tpu_reasm {
147 : ulong magic; /* ==FD_TPU_REASM_MAGIC */
148 :
149 : ulong slots_off; /* slots mem */
150 : ulong pub_slots_off; /* pub_slots mem */
151 : ulong map_off; /* map mem */
152 :
153 : fd_tpu_msg_t * dcache;
154 :
155 : uint depth; /* mcache depth */
156 : uint burst; /* max concurrent reassemblies */
157 :
158 : uint head; /* least recent reassembly */
159 : uint tail; /* most recent reassembly */
160 :
161 : uint slot_cnt;
162 : ushort orig; /* tango orig */
163 : };
164 :
165 : typedef struct fd_tpu_reasm fd_tpu_reasm_t;
166 :
167 : FD_PROTOTYPES_BEGIN
168 :
169 : /* Private accessors */
170 :
171 : static inline FD_FN_PURE fd_tpu_reasm_slot_t *
172 20227863 : fd_tpu_reasm_slots_laddr( fd_tpu_reasm_t * reasm ) {
173 20227863 : return (fd_tpu_reasm_slot_t *)( (ulong)reasm + reasm->slots_off );
174 20227863 : }
175 :
176 : static inline FD_FN_PURE fd_tpu_reasm_slot_t const *
177 435078 : fd_tpu_reasm_slots_laddr_const( fd_tpu_reasm_t const * reasm ) {
178 435078 : return (fd_tpu_reasm_slot_t const *)( (ulong)reasm + reasm->slots_off );
179 435078 : }
180 :
181 : static inline FD_FN_PURE uint *
182 216774 : fd_tpu_reasm_pub_slots_laddr( fd_tpu_reasm_t * reasm ) {
183 216774 : return (uint *)( (ulong)reasm + reasm->pub_slots_off );
184 216774 : }
185 :
186 : /* Construction API */
187 :
188 : /* fd_tpu_reasm_{align,footprint} return the required alignment and
189 : footprint of a memory region suitable for use as a tpu_reasm that
190 : can reassemble up to 'reasm_max' txns concurrently. 'depth' is the
191 : entry count of the target mcache. mtu is the max sz of a serialized
192 : txn (usually FD_TXN_MTU). */
193 :
194 : FD_FN_CONST ulong
195 : fd_tpu_reasm_align( void );
196 :
197 : FD_FN_CONST ulong
198 : fd_tpu_reasm_footprint( ulong depth, /* Assumed in {2^0,2^1,2^2,...,2^31} */
199 : ulong reasm_max ); /* Assumed in [1,2^31) */
200 :
201 : FD_FN_CONST static inline ulong
202 : fd_tpu_reasm_req_data_sz( ulong depth,
203 9 : ulong reasm_max ) { /* Assumed in [1,2^31) */
204 9 : return (depth+reasm_max) * sizeof(fd_tpu_msg_t);
205 9 : }
206 :
207 : /* fd_tpu_reasm_new formats an unused memory region for use as a
208 : tpu_reasm. shmem is a non-NULL pointer to this region in the local
209 : address space with the required footprint and alignment. {depth,
210 : reasm_max,mtu} as described above. orig is the Tango origin ID of
211 : this tpu_reasm. seed is the hash seed used by the reassembly map and
212 : should be randomly generated. dcache is a local join to an fd_dcache
213 : that tpu_reasm will write frags to. dcache should have at least
214 : fd_tpu_reasm_req_data_sz() bytes of data_sz. The dcache app region
215 : is ignored and not written to. */
216 :
217 : void *
218 : fd_tpu_reasm_new( void * shmem,
219 : ulong depth, /* Assumed in {2^0,2^1,2^2,...,2^32} */
220 : ulong reasm_max, /* Assumed in [1,2^32) */
221 : ulong seed,
222 : ulong orig, /* Assumed in [0,FD_FRAG_META_ORIG_MAX) */
223 : void * dcache );
224 :
225 : fd_tpu_reasm_t *
226 : fd_tpu_reasm_join( void * shreasm );
227 :
228 : void *
229 : fd_tpu_reasm_leave( fd_tpu_reasm_t * reasm );
230 :
231 : void *
232 : fd_tpu_reasm_delete( void * shreasm );
233 :
234 : /* Accessor API */
235 :
236 : fd_tpu_reasm_slot_t *
237 : fd_tpu_reasm_query( fd_tpu_reasm_t * reasm,
238 : ulong conn_uid,
239 : ulong stream_id );
240 :
241 : FD_FN_PURE static inline fd_tpu_reasm_slot_t *
242 0 : fd_tpu_reasm_peek_tail( fd_tpu_reasm_t * reasm ) {
243 0 : uint tail_idx = reasm->tail;
244 0 : fd_tpu_reasm_slot_t * tail = fd_tpu_reasm_slots_laddr( reasm ) + tail_idx;
245 0 : return tail;
246 0 : }
247 :
248 : fd_tpu_reasm_slot_t *
249 : fd_tpu_reasm_prepare( fd_tpu_reasm_t * reasm,
250 : ulong conn_uid,
251 : ulong stream_id,
252 : long first_seen_nanos );
253 :
254 : static inline fd_tpu_reasm_slot_t *
255 : fd_tpu_reasm_acquire( fd_tpu_reasm_t * reasm,
256 : ulong conn_uid,
257 : ulong stream_id,
258 76509 : long first_seen_nanos ) {
259 76509 : fd_tpu_reasm_slot_t * slot = fd_tpu_reasm_query( reasm, conn_uid, stream_id );
260 76509 : if( !slot ) {
261 76125 : slot = fd_tpu_reasm_prepare( reasm, conn_uid, stream_id, first_seen_nanos );
262 76125 : }
263 76509 : return slot;
264 76509 : }
265 :
266 : /* fd_tpu_reasm_frag appends a new stream frag to the reasm slot.
267 : [data,data+data_sz) is the memory region containing the stream data.
268 : data_off is the offset of this stream data. Slot reassembly buffer
269 : is appended with copy of [data,data+data_sz) on success. On failure,
270 : cancels the reassembly.
271 :
272 : Return values one of:
273 :
274 : FD_TPU_REASM_SUCCESS: success, fragment added to reassembly
275 : FD_TPU_REASM_EAGAIN: incomplete
276 : FD_TPU_REASM_ERR_SZ: fail, data_off + data_sz > mtu
277 : FD_TPU_REASM_ERR_SKIP: fail, data_off - slot->sz > 0
278 :
279 : Note on SKIP error: RFC 9000 Section 2.2 specifies "QUIC makes no
280 : specific allowances for delivery of stream data out of order." */
281 :
282 : int
283 : fd_tpu_reasm_frag( fd_tpu_reasm_t * reasm,
284 : fd_tpu_reasm_slot_t * slot,
285 : uchar const * data,
286 : ulong sz,
287 : ulong off );
288 :
289 : /* fd_tpu_reasm_publish completes a stream reassembly and publishes the
290 : message to an mcache for downstream consumption. base is the address
291 : of the chunk whose index is 0 (chunk0 param of fd_chunk_to_laddr).
292 : {seq,sig,tspub} are mcache frag params. If slot does not have active
293 : reassembly or txn parsing failed, returns NULL. If base is not valid
294 : for tpu_reasm, aborts. Final msg sz in [0,mtu+FD_CHUNK_SZ). */
295 :
296 : int
297 : fd_tpu_reasm_publish( fd_tpu_reasm_t * reasm,
298 : fd_tpu_reasm_slot_t * slot,
299 : fd_frag_meta_t * mcache,
300 : void * base, /* Assumed aligned FD_CHUNK_ALIGN */
301 : ulong seq,
302 : long tspub,
303 : uint source_ipv4,
304 : uchar source_tpu );
305 :
306 : /* fd_tpu_reasm_publish_fast is a streamlined version of acquire/frag/
307 : publish. */
308 :
309 : int
310 : fd_tpu_reasm_publish_fast( fd_tpu_reasm_t * reasm,
311 : uchar const * data,
312 : ulong sz,
313 : fd_frag_meta_t * mcache,
314 : void * base, /* Assumed aligned FD_CHUNK_ALIGN */
315 : ulong seq,
316 : long tspub,
317 : uint source_ipv4,
318 : uchar source_tpu );
319 :
320 : /* fd_tpu_reasm_cancel cancels the given stream reassembly. */
321 :
322 : void
323 : fd_tpu_reasm_cancel( fd_tpu_reasm_t * reasm,
324 : fd_tpu_reasm_slot_t * slot );
325 :
326 : /* fd_tpu_reasm_key_hash is an unrolled version of fd_hash (xxhash-r39) */
327 :
328 78228588 : #define C1 (11400714785074694791UL)
329 58671441 : #define C2 (14029467366897019727UL)
330 19557147 : #define C3 ( 1609587929392839161UL)
331 39114294 : #define C4 ( 9650029242287828579UL)
332 19557147 : #define C5 ( 2870177450012600261UL)
333 :
334 : FD_FN_PURE static inline ulong
335 : fd_tpu_reasm_key_hash( fd_tpu_reasm_key_t const * k,
336 19557147 : ulong seed ) {
337 :
338 19557147 : ulong h = seed + C5 + 16UL;
339 19557147 : ulong w0 = k->conn_uid;
340 19557147 : ulong w1 = k->stream_id;
341 :
342 19557147 : w0 *= C2; w0 = fd_ulong_rotate_left( w0, 31 ); w0 *= C1; h ^= w0; h = fd_ulong_rotate_left( h, 27 )*C1 + C4;
343 19557147 : w1 *= C2; w1 = fd_ulong_rotate_left( w1, 31 ); w1 *= C1; h ^= w1; h = fd_ulong_rotate_left( h, 27 )*C1 + C4;
344 :
345 : /* Final avalanche */
346 19557147 : h ^= h >> 33;
347 19557147 : h *= C2;
348 19557147 : h ^= h >> 29;
349 19557147 : h *= C3;
350 19557147 : h ^= h >> 32;
351 :
352 19557147 : return h;
353 19557147 : }
354 :
355 : #undef C1
356 : #undef C2
357 : #undef C3
358 : #undef C4
359 : #undef C5
360 :
361 : FD_PROTOTYPES_END
362 :
363 : #endif /* HEADER_fd_src_disco_quic_fd_tpu_h */
|
