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 : uint tsorig_comp;
143 : };
144 :
145 : typedef struct fd_tpu_reasm_slot fd_tpu_reasm_slot_t;
146 :
147 : struct fd_tpu_reasm {
148 : ulong magic; /* ==FD_TPU_REASM_MAGIC */
149 :
150 : ulong slots_off; /* slots mem */
151 : ulong pub_slots_off; /* pub_slots mem */
152 : ulong map_off; /* map mem */
153 :
154 : fd_tpu_msg_t * dcache;
155 :
156 : uint depth; /* mcache depth */
157 : uint burst; /* max concurrent reassemblies */
158 :
159 : uint head; /* least recent reassembly */
160 : uint tail; /* most recent reassembly */
161 :
162 : uint slot_cnt;
163 : ushort orig; /* tango orig */
164 : };
165 :
166 : typedef struct fd_tpu_reasm fd_tpu_reasm_t;
167 :
168 : FD_PROTOTYPES_BEGIN
169 :
170 : /* Private accessors */
171 :
172 : static inline FD_FN_PURE fd_tpu_reasm_slot_t *
173 20227863 : fd_tpu_reasm_slots_laddr( fd_tpu_reasm_t * reasm ) {
174 20227863 : return (fd_tpu_reasm_slot_t *)( (ulong)reasm + reasm->slots_off );
175 20227863 : }
176 :
177 : static inline FD_FN_PURE fd_tpu_reasm_slot_t const *
178 358953 : fd_tpu_reasm_slots_laddr_const( fd_tpu_reasm_t const * reasm ) {
179 358953 : return (fd_tpu_reasm_slot_t const *)( (ulong)reasm + reasm->slots_off );
180 358953 : }
181 :
182 : static inline FD_FN_PURE uint *
183 216774 : fd_tpu_reasm_pub_slots_laddr( fd_tpu_reasm_t * reasm ) {
184 216774 : return (uint *)( (ulong)reasm + reasm->pub_slots_off );
185 216774 : }
186 :
187 : /* Construction API */
188 :
189 : /* fd_tpu_reasm_{align,footprint} return the required alignment and
190 : footprint of a memory region suitable for use as a tpu_reasm that
191 : can reassemble up to 'reasm_max' txns concurrently. 'depth' is the
192 : entry count of the target mcache. mtu is the max sz of a serialized
193 : txn (usually FD_TXN_MTU). */
194 :
195 : FD_FN_CONST ulong
196 : fd_tpu_reasm_align( void );
197 :
198 : FD_FN_CONST ulong
199 : fd_tpu_reasm_footprint( ulong depth, /* Assumed in {2^0,2^1,2^2,...,2^31} */
200 : ulong reasm_max ); /* Assumed in [1,2^31) */
201 :
202 : FD_FN_CONST static inline ulong
203 : fd_tpu_reasm_req_data_sz( ulong depth,
204 9 : ulong reasm_max ) { /* Assumed in [1,2^31) */
205 9 : return (depth+reasm_max) * sizeof(fd_tpu_msg_t);
206 9 : }
207 :
208 : /* fd_tpu_reasm_new formats an unused memory region for use as a
209 : tpu_reasm. shmem is a non-NULL pointer to this region in the local
210 : address space with the required footprint and alignment. {depth,
211 : reasm_max,mtu} as described above. orig is the Tango origin ID of
212 : this tpu_reasm. dcache is a local join to an fd_dcache that
213 : 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 orig, /* Assumed in [0,FD_FRAG_META_ORIG_MAX) */
222 : void * dcache );
223 :
224 : fd_tpu_reasm_t *
225 : fd_tpu_reasm_join( void * shreasm );
226 :
227 : void *
228 : fd_tpu_reasm_leave( fd_tpu_reasm_t * reasm );
229 :
230 : void *
231 : fd_tpu_reasm_delete( void * shreasm );
232 :
233 : /* Accessor API */
234 :
235 : fd_tpu_reasm_slot_t *
236 : fd_tpu_reasm_query( fd_tpu_reasm_t * reasm,
237 : ulong conn_uid,
238 : ulong stream_id );
239 :
240 : FD_FN_PURE static inline fd_tpu_reasm_slot_t *
241 0 : fd_tpu_reasm_peek_tail( fd_tpu_reasm_t * reasm ) {
242 0 : uint tail_idx = reasm->tail;
243 0 : fd_tpu_reasm_slot_t * tail = fd_tpu_reasm_slots_laddr( reasm ) + tail_idx;
244 0 : return tail;
245 0 : }
246 :
247 : fd_tpu_reasm_slot_t *
248 : fd_tpu_reasm_prepare( fd_tpu_reasm_t * reasm,
249 : ulong conn_uid,
250 : ulong stream_id,
251 : long tspub );
252 :
253 : static inline fd_tpu_reasm_slot_t *
254 : fd_tpu_reasm_acquire( fd_tpu_reasm_t * reasm,
255 : ulong conn_uid,
256 : ulong stream_id,
257 76509 : long tspub ) {
258 76509 : fd_tpu_reasm_slot_t * slot = fd_tpu_reasm_query( reasm, conn_uid, stream_id );
259 76509 : if( !slot ) {
260 76125 : slot = fd_tpu_reasm_prepare( reasm, conn_uid, stream_id, tspub );
261 76125 : }
262 76509 : return slot;
263 76509 : }
264 :
265 : /* fd_tpu_reasm_frag appends a new stream frag to the reasm slot.
266 : [data,data+data_sz) is the memory region containing the stream data.
267 : data_off is the offset of this stream data. Slot reassembly buffer
268 : is appended with copy of [data,data+data_sz) on success. On failure,
269 : cancels the reassembly.
270 :
271 : Return values one of:
272 :
273 : FD_TPU_REASM_SUCCESS: success, fragment added to reassembly
274 : FD_TPU_REASM_EAGAIN: incomplete
275 : FD_TPU_REASM_ERR_SZ: fail, data_off + data_sz > mtu
276 : FD_TPU_REASM_ERR_SKIP: fail, data_off - slot->sz > 0
277 :
278 : Note on SKIP error: RFC 9000 Section 2.2 specifies "QUIC makes no
279 : specific allowances for delivery of stream data out of order." */
280 :
281 : int
282 : fd_tpu_reasm_frag( fd_tpu_reasm_t * reasm,
283 : fd_tpu_reasm_slot_t * slot,
284 : uchar const * data,
285 : ulong sz,
286 : ulong off );
287 :
288 : /* fd_tpu_reasm_publish completes a stream reassembly and publishes the
289 : message to an mcache for downstream consumption. base is the address
290 : of the chunk whose index is 0 (chunk0 param of fd_chunk_to_laddr).
291 : {seq,sig,tspub} are mcache frag params. If slot does not have active
292 : reassembly or txn parsing failed, returns NULL. If base is not valid
293 : for tpu_reasm, aborts. Final msg sz in [0,mtu+FD_CHUNK_SZ). */
294 :
295 : int
296 : fd_tpu_reasm_publish( fd_tpu_reasm_t * reasm,
297 : fd_tpu_reasm_slot_t * slot,
298 : fd_frag_meta_t * mcache,
299 : void * base, /* Assumed aligned FD_CHUNK_ALIGN */
300 : ulong seq,
301 : long tspub,
302 : uint source_ipv4,
303 : uchar source_tpu );
304 :
305 : /* fd_tpu_reasm_publish_fast is a streamlined version of acquire/frag/
306 : publish. */
307 :
308 : int
309 : fd_tpu_reasm_publish_fast( fd_tpu_reasm_t * reasm,
310 : uchar const * data,
311 : ulong sz,
312 : fd_frag_meta_t * mcache,
313 : void * base, /* Assumed aligned FD_CHUNK_ALIGN */
314 : ulong seq,
315 : long tspub,
316 : uint source_ipv4,
317 : uchar source_tpu );
318 :
319 : /* fd_tpu_reasm_cancel cancels the given stream reassembly. */
320 :
321 : void
322 : fd_tpu_reasm_cancel( fd_tpu_reasm_t * reasm,
323 : fd_tpu_reasm_slot_t * slot );
324 :
325 : /* fd_tpu_reasm_key_hash is an unrolled version of fd_hash (xxhash-r39) */
326 :
327 78228588 : #define C1 (11400714785074694791UL)
328 58671441 : #define C2 (14029467366897019727UL)
329 19557147 : #define C3 ( 1609587929392839161UL)
330 39114294 : #define C4 ( 9650029242287828579UL)
331 19557147 : #define C5 ( 2870177450012600261UL)
332 :
333 : FD_FN_PURE static inline ulong
334 : fd_tpu_reasm_key_hash( fd_tpu_reasm_key_t const * k,
335 19557147 : ulong seed ) {
336 :
337 19557147 : ulong h = seed + C5 + 16UL;
338 19557147 : ulong w0 = k->conn_uid;
339 19557147 : ulong w1 = k->stream_id;
340 :
341 19557147 : w0 *= C2; w0 = fd_ulong_rotate_left( w0, 31 ); w0 *= C1; h ^= w0; h = fd_ulong_rotate_left( h, 27 )*C1 + C4;
342 19557147 : w1 *= C2; w1 = fd_ulong_rotate_left( w1, 31 ); w1 *= C1; h ^= w1; h = fd_ulong_rotate_left( h, 27 )*C1 + C4;
343 :
344 : /* Final avalanche */
345 19557147 : h ^= h >> 33;
346 19557147 : h *= C2;
347 19557147 : h ^= h >> 29;
348 19557147 : h *= C3;
349 19557147 : h ^= h >> 32;
350 :
351 19557147 : return h;
352 19557147 : }
353 :
354 : #undef C1
355 : #undef C2
356 : #undef C3
357 : #undef C4
358 : #undef C5
359 :
360 : FD_PROTOTYPES_END
361 :
362 : #endif /* HEADER_fd_src_disco_quic_fd_tpu_h */
|
