Line data Source code
1 : /* The snapsv tile is a HTTP file server that serves local snapshots to
2 : remote peers.
3 :
4 : This tile uses io_uring for network and file I/O.
5 :
6 : snapsv is fully cooperatively scheduled using io_uring waits.
7 : It wakes on network events, timeouts, file I/O completions, and
8 : futex wakes (new tango messages).
9 :
10 : Scheduling is event-driven (lazy). Apart from the initial accept op,
11 : every other op is triggered by a completion. */
12 :
13 : #define _GNU_SOURCE
14 : #include "fd_snapmk_tile.h"
15 : #include "../../disco/topo/fd_topo.h"
16 : #include "../../disco/metrics/fd_metrics.h"
17 : #include "../../util/net/fd_ip6.h"
18 : #include <errno.h>
19 : #include <fcntl.h>
20 : #include <netinet/in.h>
21 : #include <sys/socket.h>
22 : #include <linux/futex.h>
23 : #include <linux/io_uring.h>
24 : #include "../../util/io_uring/fd_io_uring.h"
25 : #include "../../util/io_uring/fd_io_uring_setup.h"
26 : #include "../../util/io_uring/fd_io_uring_register.h"
27 : #include "../../third_party/picohttpparser/picohttpparser.h"
28 : #include "generated/fd_snapsv_tile_seccomp.h"
29 :
30 : /* Open addressed hash map of snapshots */
31 :
32 : struct snap_key {
33 : ulong slot;
34 : ulong base_slot; /* ULONG_MAX if full snap */
35 : };
36 : typedef struct snap_key snap_key_t;
37 :
38 : struct snap_entry {
39 : snap_key_t key;
40 : uchar hash[ 32 ];
41 : ulong sz;
42 : int fd;
43 : uint locked:1;
44 : uint is_zstd:1;
45 : };
46 : typedef struct snap_entry snap_entry_t;
47 :
48 : static ulong
49 : snap_key_hash( snap_key_t const * key,
50 0 : ulong seed ) {
51 : /* xxHash3 adapted */
52 0 : ulong k0 = key->base_slot;
53 0 : ulong k1 = key->slot;
54 :
55 0 : ulong lo = k0 ^ (0x6782737bea4239b9UL + seed);
56 0 : ulong hi = k1 ^ (0xaf56bc3b0996523aUL - seed);
57 :
58 0 : uint128 product = (uint128)lo * (uint128)hi;
59 0 : ulong fold = (ulong)product ^ (ulong)( product>>64 );
60 :
61 0 : ulong acc = 16UL + fd_ulong_bswap( lo ) + hi + fold;
62 0 : acc ^= acc >> 37;
63 0 : acc *= 0x165667919E3779F9UL;
64 0 : acc ^= acc >> 32;
65 0 : return acc;
66 0 : }
67 :
68 : #define MAP_NAME snap_map
69 0 : #define MAP_ELE_T snap_entry_t
70 : #define MAP_KEY_T snap_key_t
71 0 : #define MAP_KEY_EQ(k0,k1) (((k0)->slot==(k1)->slot) & ((k0)->base_slot==(k1)->base_slot))
72 0 : #define MAP_KEY_HASH(k,s) snap_key_hash( (k), (s) )
73 0 : #define MAP_ELE_IS_FREE(e) (((e)->key.slot==ULONG_MAX) & ((e)->key.base_slot==ULONG_MAX))
74 0 : #define MAP_ELE_FREE(c,e) ((e)->key=(snap_key_t){ULONG_MAX,ULONG_MAX})
75 0 : #define MAP_ELE_MOVE(c,d,s) do { MAP_ELE_T * _src = (s); (*(d)) = *_src; _src->key = (snap_key_t){ULONG_MAX,ULONG_MAX}; } while(0)
76 : #include "../../util/tmpl/fd_map_slot.c"
77 :
78 : /* Tile state */
79 :
80 : #define IN_LINK_MAX 1 /* cannot trivially grow */
81 :
82 : /* 8 ought to be enough ... The SQE/CQE count per conn is limited due
83 : to fixed depth pipelining. */
84 0 : #define SQE_PER_CONN (8UL)
85 0 : #define CQE_PER_CONN (8UL)
86 :
87 : /* io_uring fixed file indices */
88 :
89 0 : #define FIXED_FD_LISTEN (0U) /* TCP listen socket */
90 0 : #define FIXED_FD_CNT (1U)
91 :
92 : /* HTTP request head limits. The request head is accumulated in the
93 : conn's iobuf, therefore it can never exceed the iobuf size. */
94 :
95 0 : #define REQ_HEADER_MAX (64UL) /* max number of request headers */
96 :
97 : /* conn state */
98 :
99 0 : #define CONN_STATE_FREE (0U) /* conn slot is unused */
100 0 : #define CONN_STATE_REQ_PEEK (1U) /* peeking request bytes */
101 0 : #define CONN_STATE_REQ_SKIP (2U) /* consume in-flight, head still incomplete */
102 0 : #define CONN_STATE_REQ_DONE (3U) /* consume in-flight, head complete */
103 0 : #define CONN_STATE_RES_WRITE_ERR (4U) /* writing response header */
104 0 : #define CONN_STATE_RES_WRITE_HDR (5U) /* writing response header */
105 0 : #define CONN_STATE_RES_REDIRECT (6U) /* writing snap redirect */
106 0 : #define CONN_STATE_RES_SHOVEL (7U) /* pipelinined snap data RX & TX */
107 :
108 : struct snapsv_conn {
109 : uint state; /* CONN_STATE_{...} */
110 : uint iobuf_idx; /* iobuf for networking, UINT_MAX if none */
111 : uint rdbuf_idx; /* iobuf for disk read, UINT_MAX if none */
112 : uint rdbuf_len; /* completed bytes in rdbuf */
113 :
114 : uint sick:1; /* force close after current request? */
115 : uint head:1; /* HEAD request? */
116 : uint range:1; /* range request? */
117 : uint incremental:1; /* incremental snap requested? */
118 : uint disk_inflight:1; /* file read pending completion */
119 : uint net_inflight:1; /* tcp send pending completion */
120 : uint closing:1; /* close once shovel operations drain? */
121 : uint http_err; /* http status code */
122 :
123 : struct {
124 : uint len;
125 : } req;
126 :
127 : struct {
128 : uint sent;
129 : uint len;
130 : } res;
131 :
132 : struct {
133 : snap_key_t key;
134 : snap_entry_t * slot;
135 : ulong range0; /* next file offset to read */
136 : ulong range1; /* exclusive response body end */
137 : } snap;
138 :
139 : long request_start_nanos;
140 : struct __kernel_timespec idle_timeout;
141 :
142 : fd_ip6_addr_t peer_ip;
143 : ushort peer_port;
144 : };
145 :
146 : typedef struct snapsv_conn snapsv_conn_t;
147 :
148 : struct fd_snapsv {
149 :
150 : /* io_uring */
151 : fd_io_uring_t ring[1];
152 : void * ring_shmem;
153 :
154 : /* I/O buffers */
155 : uchar * iobuf0;
156 : uint * iobuf_free;
157 : uint iobuf_free_cnt;
158 : uint iobuf_sz;
159 :
160 : /* TCP sockets */
161 : int listen_fd;
162 : uint conn0_fd_idx; /* io_uring file index */
163 : ulong conn_cnt;
164 : ulong conn_max;
165 : snapsv_conn_t * conn0;
166 : uint * conn_free;
167 : uint conn_free_cnt;
168 : long idle_timeout_nanos;
169 : struct __kernel_timespec send_timeout;
170 :
171 : /* Conn accept */
172 : uint accept_inflight:1;
173 : struct sockaddr_storage accept_addr;
174 : socklen_t accept_addr_len;
175 :
176 : /* snapshot hashmap */
177 : snap_map_t snap_map[1];
178 : ulong snap_cnt_full;
179 : ulong snap_cnt_incr;
180 : ulong snap_max;
181 :
182 : /* cache: newest snaps */
183 : snap_entry_t * newest_full;
184 : snap_entry_t * newest_incr;
185 :
186 : /* snapshot files */
187 : int dir_fd;
188 :
189 : /* tango input links */
190 : uchar in_kind[ IN_LINK_MAX ];
191 : struct {
192 : void * mem;
193 : ulong chunk0;
194 : ulong wmark;
195 : ulong mtu;
196 :
197 : fd_frag_meta_t const * mcache;
198 : ulong const * seq_prod; /* mcache->seq[0] */
199 : ulong seq_cons; /* last consumed seq */
200 : uint depth;
201 : uint futex_armed:1;
202 : } in[ IN_LINK_MAX ];
203 :
204 : };
205 :
206 : typedef struct fd_snapsv fd_snapsv_t;
207 :
208 : /* snapsv_udata_t represents io_uring user data. */
209 :
210 : union snapsv_udata {
211 : ulong user_data;
212 :
213 : struct __attribute__((packed)) {
214 : uint conn_idx;
215 : uchar op;
216 0 : #define UDATA_OP_ACCEPT ((uchar)0) /* accept new conn */
217 0 : #define UDATA_OP_PEEK ((uchar)1) /* read request bytes */
218 0 : #define UDATA_OP_CONSUME_FRAG ((uchar)2) /* consume request frag */
219 0 : #define UDATA_OP_CONSUME_TAIL ((uchar)3) /* consume complete request */
220 0 : #define UDATA_OP_WRITE_HDR ((uchar)4) /* write response header */
221 0 : #define UDATA_OP_SHOVEL_DISK ((uchar)5) /* read snapshot data */
222 0 : #define UDATA_OP_SHOVEL_NET ((uchar)6) /* send snapshot data */
223 0 : #define UDATA_OP_FUTEX ((uchar)7) /* wait for tango frag */
224 0 : #define UDATA_OP_TIMEOUT ((uchar)8) /* linked socket op timeout */
225 : };
226 : };
227 :
228 : typedef union snapsv_udata snapsv_udata_t;
229 :
230 : FD_STATIC_ASSERT( sizeof(snapsv_udata_t)==sizeof(ulong), layout );
231 :
232 0 : #define IN_KIND_SNAPMK 0
233 :
234 : static ulong
235 0 : scratch_align( void ) {
236 0 : return FD_SHMEM_NORMAL_PAGE_SZ;
237 0 : }
238 :
239 : FD_FN_PURE static ulong
240 0 : snapsv_fd_max( fd_topo_tile_t const * tile ) {
241 0 : return FIXED_FD_CNT + tile->snapsv.snap_max + tile->snapsv.conn_max;
242 0 : }
243 :
244 : FD_FN_PURE static ulong
245 0 : snapsv_map_ele_max( fd_topo_tile_t const * tile ) {
246 0 : return fd_ulong_pow2_up( tile->snapsv.snap_max );
247 0 : }
248 :
249 : FD_FN_PURE static ulong
250 0 : snapsv_sq_depth( fd_topo_tile_t const * tile ) {
251 0 : return fd_ulong_pow2_up( SQE_PER_CONN*tile->snapsv.conn_max );
252 0 : }
253 :
254 : FD_FN_PURE static ulong
255 0 : snapsv_cq_depth( fd_topo_tile_t const * tile ) {
256 0 : return fd_ulong_pow2_up( CQE_PER_CONN*tile->snapsv.conn_max );
257 0 : }
258 :
259 : static ulong
260 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
261 0 : ulong sq_depth = snapsv_sq_depth( tile );
262 0 : ulong cq_depth = snapsv_cq_depth( tile );
263 0 : ulong l = FD_LAYOUT_INIT;
264 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapsv_t), sizeof(fd_snapsv_t) );
265 0 : l = FD_LAYOUT_APPEND( l, fd_io_uring_shmem_align(), fd_io_uring_shmem_footprint( sq_depth, cq_depth ) );
266 0 : l = FD_LAYOUT_APPEND( l, snap_map_align(), snap_map_footprint( snapsv_map_ele_max( tile ) ) );
267 0 : l = FD_LAYOUT_APPEND( l, alignof(int), snapsv_fd_max( tile )*sizeof(int) );
268 0 : l = FD_LAYOUT_APPEND( l, alignof(snapsv_conn_t), tile->snapsv.conn_max*sizeof(snapsv_conn_t) );
269 0 : l = FD_LAYOUT_APPEND( l, alignof(uint), tile->snapsv.conn_max*sizeof(uint) );
270 0 : l = FD_LAYOUT_APPEND( l, alignof(uint), tile->snapsv.conn_max * 2 * sizeof(uint) );
271 0 : l = FD_LAYOUT_APPEND( l, FD_SHMEM_NORMAL_PAGE_SZ, tile->snapsv.conn_max * (tile->snapsv.send_buffer_size_kib<<11) );
272 0 : return FD_LAYOUT_FINI( l, scratch_align() );
273 0 : }
274 :
275 : /* snapsv_listen creates the TCP listen socket */
276 :
277 : static int
278 : snapsv_listen( fd_ip6_addr_t const * listen_addr,
279 : ushort listen_port,
280 0 : ulong conn_max ) {
281 0 : int ip4 = fd_ip6_addr_is_ip4_mapped( listen_addr->addr ) && !listen_addr->scope_id;
282 :
283 0 : int sock_fd = socket( ip4 ? AF_INET : AF_INET6, SOCK_STREAM|SOCK_NONBLOCK, 0 );
284 0 : if( FD_UNLIKELY( -1==sock_fd ) ) {
285 0 : FD_LOG_ERR(( "socket(AF_INET%s,SOCK_STREAM) failed (%i-%s)", ip4 ? "" : "6", errno, fd_io_strerror( errno ) ));
286 0 : }
287 :
288 : /* recover quickly from restart, if there are lingering TIME_WAIT conns */
289 0 : int optval = 1;
290 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(int) ) ) ) {
291 0 : FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_REUSEADDR) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
292 0 : }
293 : /* load balancing across multiple snapsv tiles */
294 0 : optval = 1;
295 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(int) ) ) ) {
296 0 : FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_REUSEPORT) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
297 0 : }
298 :
299 0 : union {
300 0 : struct sockaddr_in ip4;
301 0 : struct sockaddr_in6 ip6;
302 0 : } addr = {0};
303 0 : ulong addr_sz;
304 :
305 0 : if( ip4 ) {
306 0 : addr.ip4 = (struct sockaddr_in){
307 0 : .sin_family = AF_INET,
308 0 : .sin_port = fd_ushort_bswap( listen_port ),
309 0 : .sin_addr.s_addr = fd_ip6_addr_to_ip4( listen_addr->addr )
310 0 : };
311 0 : addr_sz = sizeof(struct sockaddr_in);
312 0 : } else {
313 0 : int v6only = 0;
314 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, IPPROTO_IPV6, IPV6_V6ONLY, &v6only, sizeof(int) ) ) ) {
315 0 : FD_LOG_ERR(( "setsockopt(IPPROTO_IPV6,IPV6_V6ONLY) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
316 0 : }
317 0 : addr.ip6 = (struct sockaddr_in6){
318 0 : .sin6_family = AF_INET6,
319 0 : .sin6_port = fd_ushort_bswap( listen_port ),
320 0 : .sin6_scope_id = listen_addr->scope_id
321 0 : };
322 0 : memcpy( addr.ip6.sin6_addr.s6_addr, listen_addr->addr, 16UL );
323 0 : addr_sz = sizeof(struct sockaddr_in6);
324 0 : }
325 :
326 0 : char addr_cstr[ FD_IP6_ADDR_CSTR_MAX ];
327 0 : if( FD_UNLIKELY( -1==bind( sock_fd, fd_type_pun( &addr ), (uint)addr_sz ) ) ) {
328 0 : FD_LOG_ERR(( "bind(%s:%u) failed (%i-%s)",
329 0 : fd_ip6_addr_cstr( addr_cstr, listen_addr ), listen_port,
330 0 : errno, fd_io_strerror( errno ) ));
331 0 : }
332 0 : if( FD_UNLIKELY( -1==listen( sock_fd, (int)conn_max ) ) ) {
333 0 : FD_LOG_ERR(( "listen() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
334 0 : }
335 :
336 0 : FD_LOG_NOTICE(( "snapshot server listening at %shttp://%s:%u%s",
337 0 : fd_log_style_bold(), fd_ip6_addr_cstr( addr_cstr, listen_addr ), listen_port,
338 0 : fd_log_style_normal() ));
339 :
340 0 : return sock_fd;
341 0 : }
342 :
343 : static void
344 : privileged_init( fd_topo_t const * topo,
345 0 : fd_topo_tile_t const * tile ) {
346 0 : ulong fd_max = snapsv_fd_max( tile );
347 0 : ulong sq_depth = snapsv_sq_depth( tile );
348 0 : ulong cq_depth = snapsv_cq_depth( tile );
349 0 : FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
350 0 : fd_snapsv_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapsv_t), sizeof(fd_snapsv_t) );
351 0 : void * ring_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_io_uring_shmem_align(), fd_io_uring_shmem_footprint( sq_depth, cq_depth ) );
352 0 : void * map_mem = FD_SCRATCH_ALLOC_APPEND( l, snap_map_align(), snap_map_footprint( snapsv_map_ele_max( tile ) ) );
353 0 : int * fd_table = FD_SCRATCH_ALLOC_APPEND( l, alignof(int), fd_max*sizeof(int) );
354 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(snapsv_conn_t), tile->snapsv.conn_max*sizeof(snapsv_conn_t) );
355 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max*sizeof(uint) );
356 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max*2UL*sizeof(uint) );
357 0 : uchar * iobuf0 = FD_SCRATCH_ALLOC_APPEND( l, FD_SHMEM_NORMAL_PAGE_SZ, tile->snapsv.conn_max*(tile->snapsv.send_buffer_size_kib<<11) );
358 :
359 0 : memset( ctx, 0, sizeof(fd_snapsv_t) );
360 0 : memset( fd_table, 0xff, fd_max*sizeof(int) );
361 :
362 : /* TCP listen socket */
363 :
364 0 : ctx->conn_max = tile->snapsv.conn_max;
365 0 : ctx->listen_fd = snapsv_listen( &tile->snapsv.listen_addr, tile->snapsv.listen_port, tile->snapsv.conn_max );
366 :
367 : /* io_uring setup */
368 :
369 0 : ctx->ring_shmem = ring_mem;
370 0 : fd_io_uring_params_t params[1];
371 0 : fd_io_uring_params_init( params, (uint)sq_depth );
372 0 : params->flags |= IORING_SETUP_COOP_TASKRUN | IORING_SETUP_DEFER_TASKRUN;
373 :
374 0 : if( FD_UNLIKELY( !fd_io_uring_init_shmem( ctx->ring, params, ctx->ring_shmem, sq_depth, cq_depth ) ) ) {
375 0 : FD_LOG_ERR(( "fd_io_uring_init_shmem failed (%i-%s)", errno, fd_io_strerror( errno ) ));
376 0 : }
377 :
378 : /* registered buffer to help with DMA */
379 :
380 0 : struct iovec iovec = {
381 0 : .iov_base = iobuf0,
382 0 : .iov_len = tile->snapsv.conn_max*(tile->snapsv.send_buffer_size_kib<<11)
383 0 : };
384 0 : if( FD_UNLIKELY( fd_io_uring_register_buffers( ctx->ring->ioring_fd, &iovec, 1UL )<0 ) ) {
385 0 : FD_LOG_ERR(( "io_uring_register_buffers() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
386 0 : }
387 :
388 : /* Build file descriptor table */
389 0 : FD_TEST( tile->snapsv.snap_max<=FD_SNAP_MAX );
390 0 : fd_table[ FIXED_FD_LISTEN ] = ctx->listen_fd;
391 0 : uint fixed_fd = FIXED_FD_CNT;
392 0 : for( ulong i=0UL; i<tile->snapsv.snap_max; i++ ) {
393 0 : fd_table[ fixed_fd++ ] = FD_SNAP_RO_FD( i );
394 0 : }
395 0 : ctx->conn0_fd_idx = fixed_fd;
396 0 : if( FD_UNLIKELY( fd_io_uring_register_files( ctx->ring->ioring_fd, fd_table, fd_max )<0 ) ) {
397 0 : FD_LOG_ERR(( "io_uring_register_files() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
398 0 : }
399 :
400 0 : uint max_workers[2] = { tile->snapsv.io_worker_cnt, tile->snapsv.io_worker_cnt };
401 0 : if( FD_UNLIKELY( fd_io_uring_register( ctx->ring->ioring_fd, IORING_REGISTER_IOWQ_MAX_WORKERS, max_workers, 2U )<0 ) ) {
402 0 : FD_LOG_ERR(( "io_uring_register(IORING_REGISTER_IOWQ_MAX_WORKERS) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
403 0 : }
404 :
405 0 : fd_io_uring_restriction_t restrictions[] = {
406 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_ACCEPT },
407 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_RECV },
408 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_SEND },
409 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_READ_FIXED },
410 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_FUTEX_WAIT },
411 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_LINK_TIMEOUT },
412 0 : { .opcode = FD_IORING_RESTRICTION_REGISTER_OP, /* deregister files */
413 0 : .register_op = FD_IORING_REGISTER_FILES_UPDATE },
414 0 : { .opcode = FD_IORING_RESTRICTION_SQE_FLAGS_ALLOWED,
415 0 : .sqe_flags = IOSQE_IO_LINK },
416 0 : { .opcode = FD_IORING_RESTRICTION_SQE_FLAGS_REQUIRED,
417 0 : .sqe_flags = IOSQE_FIXED_FILE },
418 0 : };
419 0 : if( FD_UNLIKELY( fd_io_uring_register_restrictions(
420 0 : ctx->ring->ioring_fd, restrictions,
421 0 : (uint)( sizeof(restrictions)/sizeof(restrictions[0]) ) )<0 ) ) {
422 0 : FD_LOG_ERR(( "io_uring_register_restrictions failed (%i-%s)", errno, fd_io_strerror( errno ) ));
423 0 : }
424 :
425 0 : if( FD_UNLIKELY( fd_io_uring_enable_rings( ctx->ring->ioring_fd )<0 ) ) {
426 0 : FD_LOG_ERR(( "io_uring_enable_rings failed (%i-%s)", errno, fd_io_strerror( errno ) ));
427 0 : }
428 :
429 : /* snapshot map setup */
430 :
431 0 : ulong map_ele_max = snapsv_map_ele_max( tile );
432 0 : snap_entry_t * slots = snap_map_new( map_mem, map_ele_max, 0 );
433 0 : FD_TEST( slots );
434 0 : for( ulong i=0UL; i<map_ele_max; i++ ) {
435 0 : slots[ i ].key = (snap_key_t){ ULONG_MAX,ULONG_MAX };
436 0 : }
437 0 : ulong map_seed; FD_TEST( fd_rng_secure( &map_seed, sizeof(map_seed) ) );
438 0 : FD_TEST( snap_map_join( ctx->snap_map, slots, map_ele_max, map_ele_max, map_seed ) );
439 0 : ctx->snap_max = tile->snapsv.snap_max;
440 0 : }
441 :
442 : static void
443 : prep_accept( fd_snapsv_t * ctx );
444 :
445 : static void
446 : unprivileged_init( fd_topo_t const * topo,
447 0 : fd_topo_tile_t const * tile ) {
448 0 : ulong sq_depth = snapsv_sq_depth( tile );
449 0 : ulong cq_depth = snapsv_cq_depth( tile );
450 0 : FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
451 0 : fd_snapsv_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapsv_t), sizeof(fd_snapsv_t) );
452 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, fd_io_uring_shmem_align(), fd_io_uring_shmem_footprint( sq_depth, cq_depth ) );
453 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, snap_map_align(), snap_map_footprint( snapsv_map_ele_max( tile ) ) );
454 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(int), snapsv_fd_max( tile )*sizeof(int) );
455 0 : snapsv_conn_t * conn0 = FD_SCRATCH_ALLOC_APPEND( l, alignof(snapsv_conn_t), tile->snapsv.conn_max*sizeof(snapsv_conn_t) );
456 0 : uint * conn_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max*sizeof(uint) );
457 0 : uint * iobuf_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max * 2 * sizeof(uint) );
458 0 : uchar * iobuf0 = FD_SCRATCH_ALLOC_APPEND( l, FD_SHMEM_NORMAL_PAGE_SZ, tile->snapsv.conn_max * (tile->snapsv.send_buffer_size_kib<<11) );
459 :
460 0 : FD_CHECK_ERR( tile->in_cnt<=IN_LINK_MAX, "too many input links" );
461 0 : FD_CHECK_ERR( tile->in_cnt==1, "snapsv is hardcoded to only support one input link" );
462 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
463 0 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
464 0 : FD_CHECK_ERR( !strcmp( link->name, "snapmk_out" ), "unexpected input link" );
465 0 : FD_CHECK_ERR( tile->in_link_poll[ i ], "expecting polled input link" );
466 0 : ctx->in_kind[ i ] = IN_KIND_SNAPMK;
467 0 : ctx->in[ i ].mem = fd_wksp_containing( link->dcache );
468 0 : ctx->in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->in[ i ].mem, link->dcache );
469 0 : ctx->in[ i ].wmark = fd_dcache_compact_wmark( ctx->in[ i ].mem, link->dcache, link->mtu );
470 0 : ctx->in[ i ].mtu = link->mtu;
471 :
472 0 : ulong depth = fd_mcache_depth( link->mcache );
473 0 : FD_CHECK_ERR( depth<=UINT_MAX, "input link mcache too deep" );
474 0 : ctx->in[ i ].mcache = link->mcache;
475 0 : ctx->in[ i ].depth = (uint)depth;
476 0 : ctx->in[ i ].seq_prod = fd_mcache_seq_laddr_const( link->mcache );
477 0 : ctx->in[ i ].seq_cons = ULONG_MAX;
478 0 : ctx->in[ i ].futex_armed = 0;
479 0 : }
480 :
481 0 : ctx->conn0 = conn0;
482 0 : ctx->conn_free = conn_free;
483 0 : ctx->conn_free_cnt = (uint)tile->snapsv.conn_max;
484 0 : ctx->idle_timeout_nanos = (long)( tile->snapsv.idle_timeout_millis * ((ulong)1e6) );
485 0 : long send_timeout_nanos = (long)( tile->snapsv.send_timeout_millis * ((ulong)1e6) );
486 0 : ctx->send_timeout = (struct __kernel_timespec) {
487 0 : .tv_sec = send_timeout_nanos / (long)1e9,
488 0 : .tv_nsec = send_timeout_nanos % (long)1e9
489 0 : };
490 0 : for( ulong i=0UL; i<tile->snapsv.conn_max; i++ ) {
491 0 : conn0[ i ] = (snapsv_conn_t){ .state = CONN_STATE_FREE, .iobuf_idx = UINT_MAX, .rdbuf_idx = UINT_MAX };
492 0 : conn_free[ i ] = (uint)tile->snapsv.conn_max-1U-(uint)i;
493 0 : }
494 :
495 0 : FD_CHECK_ERR( tile->snapsv.send_buffer_size_kib, "send_buffer_size_kib is zero" );
496 0 : ulong iobuf_cnt = tile->snapsv.conn_max * 2;
497 0 : ctx->iobuf_sz = (uint)( tile->snapsv.send_buffer_size_kib<<10 );
498 0 : ctx->iobuf0 = iobuf0;
499 0 : ctx->iobuf_free = iobuf_free;
500 0 : ctx->iobuf_free_cnt = (uint)iobuf_cnt;
501 0 : for( ulong i=0UL; i<iobuf_cnt; i++ ) {
502 0 : iobuf_free[ i ] = (uint)iobuf_cnt-1U-(uint)i;
503 0 : }
504 :
505 0 : prep_accept( ctx );
506 0 : }
507 :
508 : /* prep_accept enqueues an io_uring conn accept SQE. */
509 :
510 : static void
511 0 : prep_accept( fd_snapsv_t * ctx ) {
512 0 : FD_CHECK_ERR( !ctx->accept_inflight, "accept already inflight" );
513 0 : FD_CHECK_ERR( ctx->conn_cnt < ctx->conn_max, "conn table full" );
514 0 : FD_CHECK_ERR( ctx->conn_free_cnt, "conn free list empty" );
515 :
516 0 : uint conn_idx = ctx->conn_free[ --ctx->conn_free_cnt ];
517 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
518 0 : *conn = (snapsv_conn_t){ .state = CONN_STATE_FREE, .iobuf_idx = UINT_MAX, .rdbuf_idx = UINT_MAX };
519 0 : uint fd_idx = ctx->conn0_fd_idx + conn_idx;
520 0 : ctx->accept_addr_len = sizeof(ctx->accept_addr);
521 :
522 0 : fd_io_uring_t * ring = ctx->ring;
523 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ring->sq );
524 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
525 0 : *sqe = (struct io_uring_sqe) {
526 0 : .opcode = FD_IORING_OP_ACCEPT,
527 0 : .flags = IOSQE_FIXED_FILE,
528 0 : .fd = FIXED_FD_LISTEN,
529 0 : .off = (ulong)&ctx->accept_addr_len,
530 0 : .addr = (ulong)&ctx->accept_addr,
531 0 : .len = 0,
532 0 : .file_index = fd_idx + 1U /* registered files are 1-indexed */
533 0 : };
534 0 : snapsv_udata_t udata = { .op = UDATA_OP_ACCEPT, .conn_idx = conn_idx };
535 0 : sqe->user_data = udata.user_data;
536 0 : ctx->accept_inflight = 1;
537 0 : }
538 :
539 : /* conn_iobuf returns a pointer to the current I/O buffer of a conn. */
540 :
541 : FD_FN_PURE static uchar *
542 : conn_iobuf( fd_snapsv_t const * ctx,
543 0 : snapsv_conn_t const * conn ) {
544 0 : FD_CHECK_CRIT( conn->iobuf_idx!=UINT_MAX, "conn has no iobuf" );
545 0 : return ctx->iobuf0 + (ulong)conn->iobuf_idx*ctx->iobuf_sz;
546 0 : }
547 :
548 : /* iobuf_acquire lends an iobuf to a conn. */
549 :
550 : static void
551 : iobuf_acquire( fd_snapsv_t * ctx,
552 0 : uint * slot ) {
553 0 : if( FD_LIKELY( *slot!=UINT_MAX ) ) return;
554 : /* Every conn holds at most one buffer and the pool has two per conn,
555 : so a conn that needs one always finds one. */
556 0 : FD_CHECK_CRIT( ctx->iobuf_free_cnt, "iobuf pool exhausted" );
557 0 : *slot = ctx->iobuf_free[ --ctx->iobuf_free_cnt ];
558 0 : }
559 :
560 : /* iobuf_release frees a conn iobuf (if any). */
561 :
562 : static void
563 : iobuf_release( fd_snapsv_t * ctx,
564 0 : uint * slot ) {
565 0 : if( FD_UNLIKELY( *slot==UINT_MAX ) ) return;
566 0 : ctx->iobuf_free[ ctx->iobuf_free_cnt++ ] = *slot;
567 0 : *slot = UINT_MAX;
568 0 : }
569 :
570 : /* prep_peek enqueues a receive op to copy any newly arrived bytes into
571 : iobuf. Does not advance the socket's file pointer. */
572 :
573 : static void
574 : prep_peek( fd_snapsv_t * ctx,
575 0 : uint conn_idx ) {
576 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
577 0 : long now = fd_log_wallclock();
578 0 : if( FD_UNLIKELY( !conn->request_start_nanos ) ) {
579 0 : conn->request_start_nanos = now;
580 0 : }
581 0 : long idle_rem = fd_long_max( 1L, ctx->idle_timeout_nanos - (now - conn->request_start_nanos) );
582 0 : conn->idle_timeout = (struct __kernel_timespec) {
583 0 : .tv_sec = idle_rem / (long)1e9,
584 0 : .tv_nsec = idle_rem % (long)1e9
585 0 : };
586 0 : uint len = ctx->iobuf_sz - conn->req.len;
587 0 : FD_CHECK_CRIT( len, "request head buffer is full" );
588 0 : iobuf_acquire( ctx, &conn->iobuf_idx );
589 :
590 0 : fd_io_uring_t * ring = ctx->ring;
591 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ring->sq );
592 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
593 0 : *sqe = (struct io_uring_sqe) {
594 0 : .opcode = FD_IORING_OP_RECV,
595 0 : .flags = IOSQE_FIXED_FILE | IOSQE_IO_LINK,
596 : //.ioprio = IORING_RECVSEND_FIXED_BUF,
597 0 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
598 0 : .addr = (ulong)( conn_iobuf( ctx, conn ) + conn->req.len ),
599 0 : .len = len,
600 : //.buf_index = 0U,
601 0 : .msg_flags = MSG_PEEK
602 0 : };
603 0 : snapsv_udata_t udata = { .op = UDATA_OP_PEEK, .conn_idx = conn_idx };
604 0 : sqe->user_data = udata.user_data;
605 :
606 0 : sqe = fd_io_uring_get_sqe( ring->sq );
607 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
608 0 : *sqe = (struct io_uring_sqe) {
609 0 : .opcode = FD_IORING_OP_LINK_TIMEOUT,
610 0 : .flags = IOSQE_FIXED_FILE, /* ignored */
611 0 : .addr = (ulong)&conn->idle_timeout,
612 0 : .len = 1U
613 0 : };
614 0 : udata = (snapsv_udata_t){ .op = UDATA_OP_TIMEOUT };
615 0 : sqe->user_data = udata.user_data;
616 0 : }
617 :
618 : /* prep_consume consumes previously received (via peek) iobuf bytes. */
619 :
620 : static void
621 : prep_consume( fd_snapsv_t * ctx,
622 : uint conn_idx,
623 : uint len,
624 0 : uchar op ) {
625 0 : FD_CHECK_CRIT( len, "consuming zero bytes" );
626 0 : fd_io_uring_t * ring = ctx->ring;
627 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ring->sq );
628 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
629 0 : *sqe = (struct io_uring_sqe) {
630 0 : .opcode = FD_IORING_OP_RECV,
631 0 : .flags = IOSQE_FIXED_FILE,
632 0 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
633 0 : .addr = 0UL,
634 0 : .len = len,
635 0 : .msg_flags = MSG_TRUNC | MSG_WAITALL
636 0 : };
637 0 : snapsv_udata_t udata = { .op = op, .conn_idx = conn_idx };
638 0 : sqe->user_data = udata.user_data;
639 0 : }
640 :
641 : /* conn_close destroys a conn object. If no accept op is in-flight,
642 : enqueues one. */
643 :
644 : static void
645 : conn_close( fd_snapsv_t * ctx,
646 0 : uint conn_idx ) {
647 0 : FD_CHECK_CRIT( conn_idx < ctx->conn_max, "invalid conn_idx" );
648 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
649 0 : uint fd_idx = ctx->conn0_fd_idx + conn_idx;
650 :
651 : /* io_uring OP_CLOSE on a fixed file does not use IOSQE_FIXED_FILE,
652 : therefore would break the io_uring sandbox (restrictions). */
653 0 : int unreg = -1;
654 0 : if( FD_UNLIKELY( fd_io_uring_register_files_update( ctx->ring->ioring_fd, fd_idx, &unreg, 1U )<0 ) ) {
655 0 : FD_LOG_ERR(( "io_uring_register(IORING_REGISTER_FILES_UPDATE) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
656 0 : }
657 :
658 0 : iobuf_release( ctx, &conn->iobuf_idx );
659 0 : iobuf_release( ctx, &conn->rdbuf_idx );
660 0 : *conn = (snapsv_conn_t){ .state = CONN_STATE_FREE, .iobuf_idx = UINT_MAX, .rdbuf_idx = UINT_MAX };
661 :
662 0 : FD_CHECK_CRIT( ctx->conn_cnt, "conn count underflow" );
663 0 : ctx->conn_cnt--;
664 0 : ctx->conn_free[ ctx->conn_free_cnt++ ] = conn_idx;
665 :
666 0 : if( FD_UNLIKELY( !ctx->accept_inflight ) ) {
667 0 : prep_accept( ctx );
668 0 : }
669 0 : }
670 :
671 : static ulong
672 : populate_allowed_fds( fd_topo_t const * topo,
673 : fd_topo_tile_t const * tile,
674 : ulong out_fds_cnt,
675 0 : int * out_fds ) {
676 0 : fd_snapsv_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
677 0 : if( FD_UNLIKELY( out_fds_cnt<4+ctx->snap_max ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
678 0 : ulong out_cnt = 0UL;
679 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
680 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
681 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
682 0 : out_fds[ out_cnt++ ] = ctx->ring->ioring_fd;
683 0 : out_fds[ out_cnt++ ] = ctx->listen_fd;
684 0 : for( ulong i=0UL; i<ctx->snap_max; i++ )
685 0 : out_fds[ out_cnt++ ] = FD_SNAP_RO_FD( i ); /* snapshot pool */
686 0 : return out_cnt;
687 0 : }
688 :
689 : static ulong
690 : populate_allowed_seccomp( fd_topo_t const * topo,
691 : fd_topo_tile_t const * tile,
692 : ulong out_cnt,
693 0 : struct sock_filter * out ) {
694 0 : fd_snapsv_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
695 0 : populate_sock_filter_policy_fd_snapsv_tile(
696 0 : out_cnt, out,
697 0 : (uint)fd_log_private_logfile_fd(),
698 0 : (uint)ctx->ring->ioring_fd,
699 0 : (uint)FD_SNAP_RO_FD( 0 ),
700 0 : (uint)FD_SNAP_RO_FD( ctx->snap_max-1UL ) );
701 0 : return sock_filter_policy_fd_snapsv_tile_instr_cnt;
702 0 : }
703 :
704 : static ulong
705 : rlimit_file_cnt( fd_topo_t const * topo,
706 0 : fd_topo_tile_t const * tile ) {
707 0 : (void)topo;
708 : /* stderr, logfile, boot control pipe, io_uring, listen socket, one
709 : spare for accept, the snapshot pool, and active connections */
710 0 : return 6UL + tile->snapsv.snap_max + tile->snapsv.conn_max;
711 0 : }
712 :
713 : /* handle_accept handles the completion of an accept op. */
714 :
715 : static void
716 : handle_accept( fd_snapsv_t * ctx,
717 : uint conn_idx,
718 0 : int res ) {
719 0 : ctx->accept_inflight = 0;
720 :
721 0 : if( FD_UNLIKELY( res<0 ) ) {
722 0 : FD_LOG_WARNING(( "accept() failed (%i-%s)", -res, fd_io_strerror( -res ) ));
723 0 : ctx->conn_free[ ctx->conn_free_cnt++ ] = conn_idx;
724 0 : prep_accept( ctx );
725 0 : return;
726 0 : }
727 :
728 0 : ctx->conn_cnt++;
729 :
730 : /* set conn addr */
731 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
732 0 : switch( ctx->accept_addr.ss_family ) {
733 0 : case AF_INET: {
734 0 : struct sockaddr_in const * addr = fd_type_pun_const( &ctx->accept_addr );
735 0 : fd_ip6_addr_ip4_mapped( conn->peer_ip.addr, addr->sin_addr.s_addr );
736 0 : conn->peer_ip.scope_id = 0U;
737 0 : conn->peer_port = fd_ushort_bswap( addr->sin_port );
738 0 : break;
739 0 : }
740 0 : case AF_INET6: {
741 0 : struct sockaddr_in6 const * addr = fd_type_pun_const( &ctx->accept_addr );
742 0 : memcpy( conn->peer_ip.addr, addr->sin6_addr.s6_addr, 16UL );
743 0 : conn->peer_ip.scope_id = addr->sin6_scope_id;
744 0 : conn->peer_port = fd_ushort_bswap( addr->sin6_port );
745 0 : break;
746 0 : }
747 0 : default:
748 0 : FD_LOG_WARNING(( "accept() returned unknown address family %i", (int)ctx->accept_addr.ss_family ));
749 0 : memset( &conn->peer_ip, 0, sizeof(conn->peer_ip) );
750 0 : conn->peer_port = 0U;
751 0 : }
752 :
753 0 : prep_peek( ctx, conn_idx );
754 0 : if( FD_LIKELY( ctx->conn_cnt<ctx->conn_max ) ) {
755 0 : prep_accept( ctx );
756 0 : }
757 0 : }
758 :
759 : /* conn_snap_entry checks that the snapshot for the current operation is
760 : still valid. The return value of this function is invalidated once
761 : after_credit returns. */
762 :
763 : static snap_entry_t *
764 0 : conn_snap_entry( snapsv_conn_t * conn ) {
765 0 : snap_entry_t * slot = conn->snap.slot;
766 0 : FD_CHECK_ERR( slot, "snapshot slot is NULL" );
767 0 : if( FD_UNLIKELY( ( slot->key.base_slot != conn->snap.key.base_slot ) |
768 0 : ( slot->key.slot != conn->snap.key.slot ) ) ) {
769 : /* snapshot was deleted */
770 0 : return NULL;
771 0 : }
772 0 : return slot;
773 0 : }
774 :
775 : static int
776 : snap_lock( snap_entry_t * entry );
777 :
778 : /* shovel dispatches snapshot streaming I/O work.
779 : This forms a 2-job deep pipeline that attempts to concurrently do
780 : disk read and network write I/O. */
781 :
782 : static void
783 : shovel( fd_snapsv_t * ctx,
784 0 : uint conn_idx ) {
785 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
786 0 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_SHOVEL, "conn state confusion" );
787 :
788 0 : if( FD_UNLIKELY( conn->closing ) ) {
789 0 : if( !conn->disk_inflight && !conn->net_inflight ) {
790 0 : conn_close( ctx, conn_idx );
791 0 : }
792 0 : return;
793 0 : }
794 :
795 0 : if( conn->iobuf_idx!=UINT_MAX &&
796 0 : conn->res.sent==conn->res.len &&
797 0 : !conn->net_inflight ) {
798 0 : iobuf_release( ctx, &conn->iobuf_idx );
799 0 : conn->res.sent = 0U;
800 0 : conn->res.len = 0U;
801 0 : }
802 :
803 : /* disk read buffer complete, rename to net send buffer */
804 0 : if( conn->iobuf_idx==UINT_MAX && conn->rdbuf_len ) {
805 0 : FD_CHECK_CRIT( conn->rdbuf_idx!=UINT_MAX, "completed read has no buffer" );
806 0 : conn->iobuf_idx = conn->rdbuf_idx;
807 0 : conn->rdbuf_idx = UINT_MAX;
808 0 : conn->res.sent = 0U;
809 0 : conn->res.len = conn->rdbuf_len;
810 0 : conn->rdbuf_len = 0U;
811 0 : }
812 :
813 : /* generate net send SQE */
814 0 : if( conn->iobuf_idx!=UINT_MAX && conn->res.sent<conn->res.len && !conn->net_inflight ) {
815 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ctx->ring->sq );
816 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
817 0 : *sqe = (struct io_uring_sqe) {
818 0 : .opcode = FD_IORING_OP_SEND,
819 0 : .flags = IOSQE_FIXED_FILE | IOSQE_IO_LINK,
820 : //.ioprio = IORING_RECVSEND_FIXED_BUF,
821 0 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
822 0 : .addr = (ulong)conn_iobuf( ctx, conn ) + conn->res.sent,
823 0 : .len = conn->res.len - conn->res.sent,
824 : //.buf_index = 0U,
825 0 : .msg_flags = MSG_NOSIGNAL
826 0 : };
827 0 : snapsv_udata_t udata = { .op = UDATA_OP_SHOVEL_NET, .conn_idx = conn_idx };
828 0 : sqe->user_data = udata.user_data;
829 0 : conn->net_inflight = 1U;
830 :
831 0 : sqe = fd_io_uring_get_sqe( ctx->ring->sq );
832 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
833 0 : *sqe = (struct io_uring_sqe) {
834 0 : .opcode = FD_IORING_OP_LINK_TIMEOUT,
835 0 : .flags = IOSQE_FIXED_FILE, /* ignored */
836 0 : .addr = (ulong)&ctx->send_timeout,
837 0 : .len = 1U
838 0 : };
839 0 : udata = (snapsv_udata_t){ .op = UDATA_OP_TIMEOUT };
840 0 : sqe->user_data = udata.user_data;
841 0 : }
842 :
843 : /* generate disk read SQE */
844 0 : if( conn->rdbuf_idx==UINT_MAX && !conn->disk_inflight && conn->snap.range0<conn->snap.range1 ) {
845 0 : snap_entry_t const * snap = conn_snap_entry( conn );
846 0 : if( FD_UNLIKELY( !snap ) ) {
847 0 : conn->closing = 1U;
848 0 : shovel( ctx, conn_idx );
849 0 : return;
850 0 : }
851 :
852 0 : iobuf_acquire( ctx, &conn->rdbuf_idx );
853 0 : ulong read_sz = fd_ulong_min( ctx->iobuf_sz, conn->snap.range1-conn->snap.range0 );
854 0 : ulong pool_idx = (ulong)( snap->fd - FD_SNAP_RO_FD( 0 ) );
855 0 : FD_CHECK_CRIT( pool_idx<ctx->snap_max, "snapshot has invalid file descriptor" );
856 :
857 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ctx->ring->sq );
858 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
859 0 : *sqe = (struct io_uring_sqe) {
860 0 : .opcode = FD_IORING_OP_READ_FIXED,
861 0 : .flags = IOSQE_FIXED_FILE,
862 0 : .fd = (int)( FIXED_FD_CNT + pool_idx ),
863 0 : .off = conn->snap.range0,
864 0 : .addr = (ulong)( ctx->iobuf0 + (ulong)conn->rdbuf_idx*ctx->iobuf_sz ),
865 0 : .len = (uint)read_sz,
866 0 : .buf_index = 0U
867 0 : };
868 0 : snapsv_udata_t udata = { .op = UDATA_OP_SHOVEL_DISK, .conn_idx = conn_idx };
869 0 : sqe->user_data = udata.user_data;
870 0 : conn->disk_inflight = 1U;
871 0 : }
872 :
873 : /* everything done */
874 0 : if( FD_UNLIKELY( conn->snap.range0>=conn->snap.range1 &&
875 0 : conn->iobuf_idx==UINT_MAX &&
876 0 : conn->rdbuf_idx==UINT_MAX &&
877 0 : !conn->disk_inflight &&
878 0 : !conn->net_inflight ) ) {
879 0 : if( FD_UNLIKELY( conn->sick ) ) {
880 0 : conn_close( ctx, conn_idx );
881 0 : return;
882 0 : }
883 0 : conn->state = CONN_STATE_REQ_PEEK;
884 0 : prep_peek( ctx, conn_idx );
885 0 : }
886 0 : }
887 :
888 : /* shovel_comp_disk reacts to a snapshot streaming disk read completion. */
889 :
890 : static void
891 : shovel_comp_disk( fd_snapsv_t * ctx,
892 : uint conn_idx,
893 0 : int res ) {
894 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
895 0 : FD_CHECK_CRIT( conn->state==CONN_STATE_RES_SHOVEL && conn->disk_inflight,
896 0 : "unexpected shovel disk completion" );
897 0 : conn->disk_inflight = 0U;
898 :
899 0 : ulong rem = conn->snap.range1 - conn->snap.range0;
900 0 : if( FD_UNLIKELY( res<=0 || (ulong)res>fd_ulong_min( ctx->iobuf_sz, rem ) ) ) {
901 0 : if( res<0 ) FD_LOG_WARNING(( "snapshot read failed (%i-%s)", -res, fd_io_strerror( -res ) ));
902 0 : else FD_LOG_WARNING(( "snapshot file ended early" ));
903 0 : conn->closing = 1U;
904 0 : shovel( ctx, conn_idx );
905 0 : return;
906 0 : }
907 :
908 0 : conn->snap.range0 += (ulong)res;
909 0 : conn->rdbuf_len = (uint)res;
910 0 : shovel( ctx, conn_idx );
911 0 : }
912 :
913 : /* shovel_comp_net reacts to a snapshot streaming network write
914 : completion. */
915 :
916 : static void
917 : shovel_comp_net( fd_snapsv_t * ctx,
918 : uint conn_idx,
919 0 : int res ) {
920 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
921 0 : FD_CHECK_CRIT( conn->state==CONN_STATE_RES_SHOVEL, "unexpected shovel net completion" );
922 0 : FD_CHECK_CRIT( conn->net_inflight, "unexpected SEND completion" );
923 0 : conn->net_inflight = 0U;
924 :
925 0 : if( FD_UNLIKELY( res<=0 || (uint)res>conn->res.len-conn->res.sent ) ) {
926 0 : FD_IP6_ADDR_CSTR( addr_cstr, &conn->peer_ip );
927 0 : if( res<0 ) {
928 0 : FD_LOG_INFO(( "snapshot download peer %s:%u: snapshot data send error (%i-%s)",
929 0 : addr_cstr, conn->peer_port, -res, fd_io_strerror( -res ) ));
930 0 : }
931 0 : conn->closing = 1U;
932 0 : shovel( ctx, conn_idx );
933 0 : return;
934 0 : }
935 :
936 0 : conn->res.sent += (uint)res;
937 0 : FD_MCNT_INC( SNAPSV, BYTES_WRITTEN, (ulong)res );
938 0 : shovel( ctx, conn_idx );
939 0 : }
940 :
941 : /* prep_write_hdr enqueues a header write op. */
942 :
943 : static void
944 : prep_write_hdr( fd_snapsv_t * ctx,
945 0 : uint conn_idx ) {
946 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
947 0 : fd_io_uring_t * ring = ctx->ring;
948 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ring->sq );
949 0 : FD_CHECK_ERR( conn->res.sent < conn->res.len, "conn res state confusion" );
950 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
951 0 : *sqe = (struct io_uring_sqe) {
952 0 : .opcode = FD_IORING_OP_SEND,
953 0 : .flags = IOSQE_FIXED_FILE | IOSQE_IO_LINK,
954 : //.ioprio = IORING_RECVSEND_FIXED_BUF,
955 0 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
956 0 : .addr = (ulong)conn_iobuf( ctx, conn ) + conn->res.sent,
957 0 : .len = conn->res.len - conn->res.sent,
958 : //.buf_index = 0U,
959 0 : .msg_flags = MSG_NOSIGNAL
960 0 : };
961 0 : snapsv_udata_t udata = { .op = UDATA_OP_WRITE_HDR, .conn_idx = conn_idx };
962 0 : sqe->user_data = udata.user_data;
963 :
964 0 : sqe = fd_io_uring_get_sqe( ring->sq );
965 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
966 0 : *sqe = (struct io_uring_sqe) {
967 0 : .opcode = FD_IORING_OP_LINK_TIMEOUT,
968 0 : .flags = IOSQE_FIXED_FILE, /* ignored */
969 0 : .addr = (ulong)&ctx->send_timeout,
970 0 : .len = 1U
971 0 : };
972 0 : udata = (snapsv_udata_t){ .op = UDATA_OP_TIMEOUT };
973 0 : sqe->user_data = udata.user_data;
974 0 : }
975 :
976 : /* serve_http_err dispatches a RES_WRITE_HDR job for a generic HTTP
977 : error page. */
978 :
979 : static void
980 : serve_http_err( fd_snapsv_t * ctx,
981 0 : uint conn_idx ) {
982 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
983 0 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_WRITE_ERR, "state confusion" );
984 0 : iobuf_acquire( ctx, &conn->iobuf_idx );
985 0 : uchar * iobuf = conn_iobuf( ctx, conn );
986 0 : char * p = fd_cstr_init( (char *)iobuf );
987 0 : p = fd_cstr_append_cstr( p, "HTTP/1.1 ");
988 0 : switch( conn->http_err ) {
989 0 : case 400: p = fd_cstr_append_cstr( p, "400 Bad Request" ); break;
990 0 : case 404: p = fd_cstr_append_cstr( p, "404 Not Found" ); break;
991 0 : case 416: p = fd_cstr_append_cstr( p, "416 Range Not Satisfiable" ); break;
992 0 : default: p = fd_cstr_append_cstr( p, "500 Internal Server Error" ); break;
993 0 : }
994 0 : p = fd_cstr_append_cstr( p, "\r\n" );
995 0 : if( conn->http_err==416 && conn->snap.slot ) {
996 0 : p = fd_cstr_append_cstr( p, "Content-Range: bytes */" );
997 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, conn->snap.slot->sz, fd_ulong_base10_dig_cnt( conn->snap.slot->sz ) );
998 0 : p = fd_cstr_append_cstr( p, "\r\n" );
999 0 : }
1000 0 : p = fd_cstr_append_cstr( p, "Content-Length: 0\r\n" );
1001 0 : if( conn->sick ) {
1002 0 : p = fd_cstr_append_cstr( p, "Connection: close\r\n" );
1003 0 : }
1004 0 : p = fd_cstr_append_cstr( p, "\r\n" );
1005 0 : conn->res.sent = 0;
1006 0 : conn->res.len = (uint)( p - (char *)iobuf );
1007 0 : prep_write_hdr( ctx, conn_idx );
1008 0 : }
1009 :
1010 : /* serve_snap_res_hdr dispatches a RES_WRITE_HDR job for a snapshot GET
1011 : or HEAD request. */
1012 :
1013 : static void
1014 : serve_snap_res_hdr( fd_snapsv_t * ctx,
1015 0 : uint conn_idx ) {
1016 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1017 0 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_WRITE_HDR, "state confusion" );
1018 :
1019 0 : snap_entry_t * snap = conn_snap_entry( conn );
1020 0 : if( FD_UNLIKELY( !snap ) ) {
1021 : /* rare edge case: snap was deleted by the snapmk tile just after
1022 : the user requested it. Don't bother returning an error, just
1023 : abort the conn. */
1024 0 : conn_close( ctx, conn_idx );
1025 0 : return;
1026 0 : }
1027 0 : if( FD_UNLIKELY( !snap_lock( snap ) ) ) {
1028 0 : conn_close( ctx, conn_idx );
1029 0 : return;
1030 0 : }
1031 : /* return response header */
1032 0 : FD_CHECK_ERR( conn->iobuf_idx==UINT_MAX, "conn has stale iobuf" );
1033 0 : iobuf_acquire( ctx, &conn->iobuf_idx );
1034 0 : uchar * iobuf = conn_iobuf( ctx, conn );
1035 0 : char * p = fd_cstr_init( (char *)iobuf );
1036 0 : if( conn->range ) {
1037 0 : p = fd_cstr_append_cstr( p, "HTTP/1.1 206 Partial Content\r\n" );
1038 0 : } else {
1039 0 : p = fd_cstr_append_cstr( p, "HTTP/1.1 200 OK\r\n" );
1040 0 : }
1041 0 : if( snap->is_zstd ) {
1042 0 : p = fd_cstr_append_cstr( p, "Content-Type: application/zstd\r\n" );
1043 0 : } else {
1044 0 : p = fd_cstr_append_cstr( p, "Content-Type: application/x-tar\r\n" );
1045 0 : }
1046 0 : p = fd_cstr_append_cstr( p, "Accept-Ranges: bytes\r\n" );
1047 0 : if( conn->range ) {
1048 0 : p = fd_cstr_append_cstr( p, "Content-Range: bytes " );
1049 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, conn->snap.range0, fd_ulong_base10_dig_cnt( conn->snap.range0 ) );
1050 0 : p = fd_cstr_append_char( p, '-' );
1051 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, conn->snap.range1-1UL, fd_ulong_base10_dig_cnt( conn->snap.range1-1UL ) );
1052 0 : p = fd_cstr_append_char( p, '/' );
1053 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, snap->sz, fd_ulong_base10_dig_cnt( snap->sz ) );
1054 0 : p = fd_cstr_append_cstr( p, "\r\n" );
1055 0 : }
1056 0 : p = fd_cstr_append_cstr( p, "Content-Length: " );
1057 0 : ulong content_len = conn->snap.range1 - conn->snap.range0;
1058 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, content_len, fd_ulong_base10_dig_cnt( content_len ) );
1059 0 : p = fd_cstr_append_cstr( p, "\r\n" );
1060 0 : p = fd_cstr_append_cstr( p, "\r\n" );
1061 0 : conn->res.sent = 0;
1062 0 : conn->res.len = (uint)( p - (char *)iobuf );
1063 0 : prep_write_hdr( ctx, conn_idx );
1064 0 : }
1065 :
1066 : /* newest_snap returns the newest snapshot of the requested type. */
1067 :
1068 : static snap_entry_t *
1069 : newest_snap( fd_snapsv_t * ctx,
1070 0 : int incremental ) {
1071 :
1072 0 : if( FD_UNLIKELY( !( incremental ? ctx->snap_cnt_incr : ctx->snap_cnt_full ) ) ) {
1073 0 : return NULL; /* no such snapshot of this type */
1074 0 : }
1075 0 : snap_entry_t * entry = incremental ? ctx->newest_incr : ctx->newest_full;
1076 0 : if( FD_UNLIKELY( !entry ) ) {
1077 0 : snap_entry_t * entry0 = snap_map_ele0( ctx->snap_map );
1078 0 : ulong ele_max = snap_map_ele_max( ctx->snap_map );
1079 0 : for( ulong i=0UL; i<ele_max; i++ ) {
1080 0 : snap_entry_t * e = &entry0[ i ];
1081 0 : if( snap_map_ele_is_free( e ) ) continue;
1082 0 : if( (e->key.base_slot!=ULONG_MAX)!=!!incremental ) continue;
1083 0 : if( entry && e->key.slot<=entry->key.slot ) continue;
1084 0 : entry = e;
1085 0 : }
1086 0 : }
1087 0 : *( incremental ? &ctx->newest_incr : &ctx->newest_full ) = entry;
1088 0 : if( FD_UNLIKELY( !entry ) ) return NULL;
1089 0 : return entry;
1090 0 : }
1091 :
1092 : /* serve_redirect dispatches a RES_WRITE_HDR job for a /snapshot.tar.bz2
1093 : redirect. */
1094 :
1095 : static void
1096 : serve_redirect( fd_snapsv_t * ctx,
1097 0 : uint conn_idx ) {
1098 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1099 0 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_REDIRECT, "state confusion" );
1100 :
1101 0 : snap_entry_t * snap = newest_snap( ctx, conn->incremental );
1102 0 : if( FD_UNLIKELY( !snap ) ) {
1103 0 : conn->state = CONN_STATE_RES_WRITE_ERR;
1104 0 : conn->http_err = 404U;
1105 0 : serve_http_err( ctx, conn_idx );
1106 0 : return;
1107 0 : }
1108 :
1109 : /* return response */
1110 0 : FD_CHECK_ERR( conn->iobuf_idx==UINT_MAX, "conn has stale iobuf" );
1111 0 : iobuf_acquire( ctx, &conn->iobuf_idx );
1112 0 : uchar * iobuf = conn_iobuf( ctx, conn );
1113 0 : char * p = fd_cstr_init( (char *)iobuf );
1114 0 : p = fd_cstr_append_cstr( p, "HTTP/1.1 302 Found\r\n" );
1115 0 : p = fd_cstr_append_cstr( p, "Location: /" );
1116 0 : if( conn->incremental ) {
1117 0 : p = fd_cstr_append_cstr( p, "incremental-snapshot-" );
1118 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, snap->key.base_slot, fd_ulong_base10_dig_cnt( snap->key.base_slot ) );
1119 0 : p = fd_cstr_append_char( p, '-' );
1120 0 : } else {
1121 0 : p = fd_cstr_append_cstr( p, "snapshot-" );
1122 0 : }
1123 0 : p = fd_cstr_append_ulong_as_text( p, 0, 0, snap->key.slot, fd_ulong_base10_dig_cnt( snap->key.slot ) );
1124 0 : p = fd_cstr_append_char( p, '-' );
1125 0 : char hash_b58[ FD_BASE58_ENCODED_32_SZ ];
1126 0 : fd_base58_encode_32( snap->hash, NULL, hash_b58 );
1127 0 : p = fd_cstr_append_cstr( p, hash_b58 );
1128 0 : p = fd_cstr_append_cstr( p, snap->is_zstd ? ".tar.zst\r\n" : ".tar\r\n" );
1129 0 : p = fd_cstr_append_cstr( p, "Content-Length: 0\r\n" );
1130 0 : if( conn->sick ) {
1131 0 : p = fd_cstr_append_cstr( p, "Connection: close\r\n" );
1132 0 : }
1133 0 : p = fd_cstr_append_cstr( p, "\r\n" );
1134 0 : conn->res.sent = 0;
1135 0 : conn->res.len = (uint)( p - (char *)iobuf );
1136 0 : conn->state = CONN_STATE_RES_WRITE_HDR; /* shared completion path */
1137 0 : prep_write_hdr( ctx, conn_idx );
1138 0 : }
1139 :
1140 : static void
1141 : prep_peek( fd_snapsv_t * ctx,
1142 : uint conn_idx );
1143 :
1144 : /* handle_write_hdr_comp handles the completion of a header write op. */
1145 :
1146 : static void
1147 : handle_write_hdr_comp( fd_snapsv_t * ctx,
1148 : uint conn_idx,
1149 0 : int res ) {
1150 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1151 0 : if( FD_UNLIKELY( res<0 ) ) {
1152 0 : if( res<0 ) {
1153 0 : FD_IP6_ADDR_CSTR( addr_cstr, &conn->peer_ip );
1154 0 : FD_LOG_INFO(( "snapshot download peer %s:%u: response header send error (%i-%s)",
1155 0 : addr_cstr, conn->peer_port, -res, fd_io_strerror( -res ) ));
1156 0 : }
1157 0 : conn_close( ctx, conn_idx );
1158 0 : return;
1159 0 : }
1160 0 : conn->res.sent += (uint)res;
1161 0 : if( FD_LIKELY( conn->res.sent < conn->res.len ) ) {
1162 0 : prep_write_hdr( ctx, conn_idx );
1163 0 : return;
1164 0 : }
1165 : /* wrote response body */
1166 0 : iobuf_release( ctx, &conn->iobuf_idx );
1167 0 : if( FD_UNLIKELY( conn->sick ) ) {
1168 0 : conn_close( ctx, conn_idx );
1169 0 : return;
1170 0 : }
1171 0 : switch( conn->state ) {
1172 0 : case CONN_STATE_RES_WRITE_ERR:
1173 : /* returned an error, handle the next request */
1174 0 : conn->state = CONN_STATE_REQ_PEEK;
1175 0 : prep_peek( ctx, conn_idx );
1176 0 : return;
1177 0 : case CONN_STATE_RES_WRITE_HDR:
1178 0 : if( conn->head ) {
1179 0 : conn->state = CONN_STATE_REQ_PEEK;
1180 0 : prep_peek( ctx, conn_idx );
1181 0 : return;
1182 0 : }
1183 : /* now serve the snapshot body */
1184 0 : conn->state = CONN_STATE_RES_SHOVEL;
1185 0 : shovel( ctx, conn_idx );
1186 0 : return;
1187 0 : default:
1188 0 : FD_LOG_CRIT(( "conn %u: state confusion", conn_idx ));
1189 0 : }
1190 0 : }
1191 :
1192 : /* match_snapshot_path parses a snapshot HTTP request path. */
1193 :
1194 : static snap_key_t *
1195 : match_snapshot_path( char const * path,
1196 : ulong path_len,
1197 : snap_key_t * out,
1198 : uchar out_hash[ 32 ],
1199 0 : int * out_is_zstd ) {
1200 0 : if( path_len<9 ) return NULL;
1201 :
1202 : /* ugly: copy to cstr */
1203 0 : char cstr[ FD_SNAP_NAME_MAX ];
1204 0 : if( FD_UNLIKELY( path_len>=FD_SNAP_NAME_MAX-1 ) ) return NULL;
1205 0 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( cstr ), path, path_len ) );
1206 :
1207 0 : ulong full_slot, incremental_slot;
1208 0 : int res = fd_ssarchive_parse_filename( cstr, &full_slot, &incremental_slot, out_hash, out_is_zstd );
1209 0 : if( FD_UNLIKELY( res!=0 ) ) return NULL;
1210 0 : *out = (snap_key_t) {
1211 0 : .slot = incremental_slot!=ULONG_MAX ? incremental_slot : full_slot,
1212 0 : .base_slot = incremental_slot!=ULONG_MAX ? full_slot : ULONG_MAX
1213 0 : };
1214 0 : return out;
1215 0 : }
1216 :
1217 : /* parse_range_header parses a 'Range: bytes=start-end' header.
1218 : value points to the header value char[value_len].
1219 : object_sz is the size of the file, and [*out_range0, *out_range1)
1220 : is the requested range. */
1221 :
1222 : static int
1223 : parse_range_header( char const * value,
1224 : ulong value_len,
1225 : ulong object_sz,
1226 : ulong * out_range0,
1227 0 : ulong * out_range1 ) {
1228 0 : char const * p = value;
1229 0 : char const * end = value + value_len;
1230 :
1231 0 : while( p<end && (*p==' ' || *p=='\t') ) p++;
1232 0 : while( end>p && (end[-1]==' ' || end[-1]=='\t') ) end--;
1233 0 : if( FD_UNLIKELY( (ulong)(end-p)<7UL || strncasecmp( p, "bytes=", 6UL ) ) ) return -1;
1234 0 : p += 6;
1235 :
1236 0 : ulong first = 0UL;
1237 0 : int have_first = 0;
1238 0 : while( p<end && *p>='0' && *p<='9' ) {
1239 0 : uint digit = (uint)(*p-'0');
1240 0 : if( FD_UNLIKELY( first>(ULONG_MAX-digit)/10UL ) ) return -1;
1241 0 : first = first*10UL + digit;
1242 0 : have_first = 1;
1243 0 : p++;
1244 0 : }
1245 0 : if( FD_UNLIKELY( p==end || *p!='-' ) ) return -1;
1246 0 : p++;
1247 :
1248 0 : ulong last = 0UL;
1249 0 : int have_last = 0;
1250 0 : while( p<end && *p>='0' && *p<='9' ) {
1251 0 : uint digit = (uint)(*p-'0');
1252 0 : if( FD_UNLIKELY( last>(ULONG_MAX-digit)/10UL ) ) return -1;
1253 0 : last = last*10UL + digit;
1254 0 : have_last = 1;
1255 0 : p++;
1256 0 : }
1257 0 : if( FD_UNLIKELY( p!=end || (!have_first && !have_last) ) ) return -1;
1258 0 : if( FD_UNLIKELY( !object_sz ) ) return -2;
1259 :
1260 0 : if( !have_first ) {
1261 0 : if( FD_UNLIKELY( !last ) ) return -2;
1262 0 : *out_range0 = last<object_sz ? object_sz-last : 0UL;
1263 0 : *out_range1 = object_sz;
1264 0 : return 0;
1265 0 : }
1266 :
1267 0 : if( FD_UNLIKELY( first>=object_sz ) ) return -2;
1268 0 : if( FD_UNLIKELY( have_last && last<first ) ) return -2;
1269 0 : *out_range0 = first;
1270 0 : *out_range1 = have_last && last<object_sz-1UL ? last+1UL : object_sz;
1271 0 : return 0;
1272 0 : }
1273 :
1274 : /* handle_peek handles newly received TCP data. peek_len bytes of
1275 : unconsumed request head sit at conn->req_len in the iobuf. */
1276 :
1277 : static void
1278 : handle_peek( fd_snapsv_t * ctx,
1279 : uint conn_idx,
1280 0 : int res ) {
1281 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1282 0 : if( FD_UNLIKELY( res<=0 ) ) {
1283 0 : if( FD_UNLIKELY( res<0 ) ) {
1284 0 : FD_IP6_ADDR_CSTR( addr_cstr, &conn->peer_ip );
1285 0 : FD_LOG_INFO(( "snapshot download peer %s:%u: receive error (%i-%s)",
1286 0 : addr_cstr, conn->peer_port, -res, fd_io_strerror( -res ) ));
1287 0 : }
1288 0 : conn_close( ctx, conn_idx );
1289 0 : return;
1290 0 : }
1291 0 : uint peek_len = (uint)res;
1292 :
1293 0 : FD_MCNT_INC( SNAPSV, BYTES_READ, (ulong)res );
1294 :
1295 0 : char const * method; ulong method_len;
1296 0 : char const * path; ulong path_len;
1297 0 : int minor_version;
1298 0 : struct phr_header headers[ REQ_HEADER_MAX ];
1299 0 : ulong header_cnt = REQ_HEADER_MAX;
1300 0 : int parsed = phr_parse_request(
1301 0 : (char const *)conn_iobuf( ctx, conn ),
1302 0 : conn->req.len + peek_len,
1303 0 : &method, &method_len,
1304 0 : &path, &path_len,
1305 0 : &minor_version,
1306 0 : headers, &header_cnt,
1307 0 : conn->req.len /* slowloris defense */
1308 0 : );
1309 :
1310 0 : if( FD_UNLIKELY( parsed==-2 ) ) { /* request head incomplete */
1311 0 : if( FD_UNLIKELY( conn->req.len+peek_len >= ctx->iobuf_sz ) ) {
1312 0 : FD_LOG_DEBUG(( "conn %u: request head exceeds %u bytes", conn_idx, ctx->iobuf_sz ));
1313 0 : conn_close( ctx, conn_idx );
1314 0 : return;
1315 0 : }
1316 : /* avoid copying the same data again */
1317 0 : prep_consume( ctx, conn_idx, peek_len, UDATA_OP_CONSUME_FRAG );
1318 0 : conn->state = CONN_STATE_REQ_SKIP;
1319 0 : return;
1320 0 : }
1321 :
1322 0 : if( FD_UNLIKELY( parsed<0 ) ) { /* malformed request head */
1323 0 : FD_LOG_DEBUG(( "conn %u: malformed request", conn_idx ));
1324 0 : conn_close( ctx, conn_idx );
1325 0 : return;
1326 0 : }
1327 :
1328 : /* valid request at this point */
1329 0 : FD_CHECK_ERR( (uint)parsed > conn->req.len, "request head shrank" );
1330 0 : prep_consume( ctx, conn_idx, (uint)parsed - conn->req.len, UDATA_OP_CONSUME_TAIL );
1331 0 : conn->state = CONN_STATE_REQ_DONE;
1332 0 : conn->req.len = (uint)parsed - conn->req.len;
1333 :
1334 : /* must not do any work until CONSUME_TAIL completes,
1335 : so remember what to do for when that happens */
1336 0 : snap_key_t query;
1337 0 : uchar query_hash[ 32 ];
1338 0 : int query_is_zstd;
1339 0 : if( method_len==4UL && !memcmp( method, "HEAD", method_len ) ) {
1340 0 : conn->head = 1;
1341 0 : } else if( method_len==3UL && !memcmp( method, "GET", method_len ) ) {
1342 0 : conn->head = 0;
1343 0 : } else {
1344 0 : conn->state = CONN_STATE_RES_WRITE_ERR;
1345 0 : conn->http_err = 500;
1346 0 : conn->sick = 1;
1347 0 : return;
1348 0 : }
1349 :
1350 : /* strip leading slashes */
1351 0 : if( FD_UNLIKELY( path_len<10 ) ) goto not_found;
1352 0 : if( FD_UNLIKELY( path[ 0 ]!='/' ) ) goto not_found;
1353 0 : path++; path_len--;
1354 : /* skip superfluous leading slashes to help buggy clients */
1355 0 : for( ulong i=0UL; path_len && i<15; i++ ) {
1356 0 : if( FD_LIKELY( path[ 0 ]!='/' ) ) break;
1357 0 : path++; path_len--;
1358 0 : }
1359 :
1360 0 : if( path_len==16 ) {
1361 0 : if( !memcmp( path, "snapshot.tar.zst", 16UL ) ||
1362 0 : !memcmp( path, "snapshot.tar.bz2", 16UL ) ) {
1363 0 : conn->state = CONN_STATE_RES_REDIRECT;
1364 0 : conn->incremental = 0;
1365 0 : return;
1366 0 : }
1367 0 : } else if( path_len==28 ) {
1368 0 : if( !memcmp( path, "incremental-snapshot.tar.bz2", 28UL ) ||
1369 0 : !memcmp( path, "incremental-snapshot.tar.zst", 28UL ) ) {
1370 0 : conn->state = CONN_STATE_RES_REDIRECT;
1371 0 : conn->incremental = 1;
1372 0 : return;
1373 0 : }
1374 0 : }
1375 :
1376 : /* found a snapshot? */
1377 0 : if( match_snapshot_path( path, path_len, &query, query_hash, &query_is_zstd ) ) {
1378 0 : snap_entry_t * entry = snap_map_update( ctx->snap_map, &query );
1379 0 : if( FD_UNLIKELY( !entry ) ) goto not_found;
1380 0 : if( FD_UNLIKELY( 0!=memcmp( entry->hash, query_hash, 32UL ) ) ) goto not_found;
1381 0 : if( FD_UNLIKELY( entry->is_zstd!=query_is_zstd ) ) goto not_found;
1382 0 : conn->state = CONN_STATE_RES_WRITE_HDR;
1383 0 : conn->snap.key = query;
1384 0 : conn->snap.slot = entry;
1385 0 : conn->range = 0U;
1386 0 : conn->snap.range0 = 0UL;
1387 0 : conn->snap.range1 = entry->sz;
1388 0 : conn->incremental = entry->key.base_slot!=ULONG_MAX;
1389 :
1390 0 : char const * range = NULL;
1391 0 : ulong range_len = 0UL;
1392 0 : for( ulong i=0UL; i<header_cnt; i++ ) {
1393 0 : if( headers[ i ].name_len==5UL && !strncasecmp( headers[ i ].name, "range", 5UL ) ) {
1394 0 : if( !conn->head ) { /* range header only valid for GET request */
1395 0 : range = headers[ i ].value;
1396 0 : range_len = headers[ i ].value_len;
1397 0 : }
1398 0 : }
1399 0 : }
1400 0 : if( range ) {
1401 0 : int range_err = parse_range_header( range, range_len, entry->sz, &conn->snap.range0, &conn->snap.range1 );
1402 0 : if( FD_UNLIKELY( range_err ) ) {
1403 0 : conn->state = CONN_STATE_RES_WRITE_ERR;
1404 0 : conn->http_err = (uint)( range_err==-2 ? 416 : 400 );
1405 0 : } else {
1406 0 : conn->range = 1U;
1407 0 : }
1408 0 : }
1409 0 : return;
1410 0 : }
1411 :
1412 0 : not_found:
1413 0 : conn->state = CONN_STATE_RES_WRITE_ERR;
1414 0 : conn->http_err = 404;
1415 :
1416 : /* next step happens at handle_consume_tail_comp */
1417 0 : }
1418 :
1419 : /* handle_consume_frag_comp handles the completion of a CONSUME_FRAG op. */
1420 :
1421 : static void
1422 : handle_consume_frag_comp( fd_snapsv_t * ctx,
1423 : uint conn_idx,
1424 0 : int res ) {
1425 0 : if( FD_UNLIKELY( res<=0 ) ) {
1426 0 : if( FD_UNLIKELY( res<0 ) ) {
1427 0 : FD_LOG_DEBUG(( "conn %u: recv() failed (%i-%s)", conn_idx, -res, fd_io_strerror( -res ) ));
1428 0 : }
1429 0 : conn_close( ctx, conn_idx );
1430 0 : return;
1431 0 : }
1432 0 : uint consume_len = (uint)res;
1433 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1434 0 : FD_CHECK_ERR( conn->state==CONN_STATE_REQ_SKIP, "conn state confusion" );
1435 0 : conn->req.len += consume_len;
1436 0 : prep_peek( ctx, conn_idx );
1437 0 : }
1438 :
1439 : /* handle_consume_tail_comp handles the completion of a CONSUME_TAIL op. */
1440 :
1441 : static void
1442 : handle_consume_tail_comp( fd_snapsv_t * ctx,
1443 : uint conn_idx,
1444 0 : int res ) {
1445 0 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1446 0 : if( FD_UNLIKELY( res!=(int)conn->req.len ) ) {
1447 0 : conn_close( ctx, conn_idx );
1448 0 : return;
1449 0 : }
1450 0 : conn->req.len = 0;
1451 0 : conn->request_start_nanos = 0L;
1452 0 : iobuf_release( ctx, &conn->iobuf_idx );
1453 0 : switch( ctx->conn0[ conn_idx ].state ) {
1454 0 : case CONN_STATE_RES_WRITE_ERR:
1455 0 : serve_http_err( ctx, conn_idx );
1456 0 : break;
1457 0 : case CONN_STATE_RES_WRITE_HDR:
1458 0 : serve_snap_res_hdr( ctx, conn_idx );
1459 0 : break;
1460 0 : case CONN_STATE_RES_REDIRECT:
1461 0 : serve_redirect( ctx, conn_idx );
1462 0 : break;
1463 0 : default:
1464 0 : FD_LOG_CRIT(( "conn %u: state confusion", conn_idx ));
1465 0 : }
1466 0 : }
1467 :
1468 : /* futex_prep asks the kernel to send a CQE for when a new
1469 : snapmk_out frag becomes available. */
1470 :
1471 : static int
1472 : futex_prep( fd_snapsv_t * ctx,
1473 0 : ulong in_idx ) {
1474 0 : ulong seq_prod = fd_mcache_seq_query( ctx->in[ in_idx ].seq_prod );
1475 0 : ulong seq_next = fd_seq_inc( ctx->in[ in_idx ].seq_cons, 1UL );
1476 0 : fd_frag_meta_t const * mline = ctx->in[ in_idx ].mcache + fd_mcache_line_idx( seq_next, ctx->in[ in_idx ].depth );
1477 0 : if( FD_UNLIKELY( fd_frag_meta_seq_query( mline )==seq_next ) ) return 0;
1478 :
1479 0 : if( FD_LIKELY( ctx->in[ in_idx ].futex_armed ) ) return 1;
1480 0 : fd_io_uring_t * ring = ctx->ring;
1481 0 : struct io_uring_sqe * sqe = fd_io_uring_get_sqe( ring->sq );
1482 0 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1483 0 : *sqe = (struct io_uring_sqe) {
1484 0 : .opcode = FD_IORING_OP_FUTEX_WAIT,
1485 0 : .flags = IOSQE_FIXED_FILE, /* ignored, required for sandbox (hope no kernel ABI breakage) */
1486 0 : .fd = FUTEX2_SIZE_U32,
1487 0 : .addr = (ulong)ctx->in[ in_idx ].seq_prod,
1488 0 : .off = (ulong)(uint)seq_prod, /* value to compare against */
1489 0 : .addr3 = FUTEX_BITSET_MATCH_ANY
1490 0 : };
1491 0 : snapsv_udata_t udata = { .op = UDATA_OP_FUTEX, .conn_idx = (uint)in_idx };
1492 0 : sqe->user_data = udata.user_data;
1493 0 : ctx->in[ in_idx ].futex_armed = 1;
1494 0 : return 1;
1495 0 : }
1496 :
1497 : /* futex_comp is called when a snapmk_out frag was detected. Such
1498 : frags are also polled by the event loop, but this is required to wake
1499 : up fast from an io_uring_enter sleep. */
1500 :
1501 : static void
1502 : futex_comp( fd_snapsv_t * ctx,
1503 : uint in_idx,
1504 0 : int res ) {
1505 0 : FD_CHECK_CRIT( in_idx<IN_LINK_MAX, "io_uring completion has invalid in index" );
1506 0 : ctx->in[ in_idx ].futex_armed = 0;
1507 0 : if( FD_UNLIKELY( res<0 && res!=-EAGAIN && res!=-ECANCELED && res!=-EINTR ) ) {
1508 0 : FD_LOG_ERR(( "IORING_OP_FUTEX_WAIT failed (%i-%s)", -res, fd_io_strerror( -res ) ));
1509 0 : }
1510 0 : }
1511 :
1512 : /* handle_cqe handles a single io_uring completion. */
1513 :
1514 : static void
1515 : handle_cqe( fd_snapsv_t * ctx,
1516 0 : struct io_uring_cqe const * cqe ) {
1517 0 : snapsv_udata_t udata = { .user_data = cqe->user_data };
1518 0 : switch( udata.op ) {
1519 0 : case UDATA_OP_ACCEPT:
1520 0 : handle_accept( ctx, udata.conn_idx, cqe->res );
1521 0 : break;
1522 0 : case UDATA_OP_PEEK:
1523 0 : handle_peek( ctx, udata.conn_idx, cqe->res );
1524 0 : break;
1525 0 : case UDATA_OP_CONSUME_FRAG:
1526 0 : handle_consume_frag_comp( ctx, udata.conn_idx, cqe->res );
1527 0 : break;
1528 0 : case UDATA_OP_CONSUME_TAIL:
1529 0 : handle_consume_tail_comp( ctx, udata.conn_idx, cqe->res );
1530 0 : break;
1531 0 : case UDATA_OP_WRITE_HDR:
1532 0 : handle_write_hdr_comp( ctx, udata.conn_idx, cqe->res );
1533 0 : break;
1534 0 : case UDATA_OP_SHOVEL_DISK:
1535 0 : shovel_comp_disk( ctx, udata.conn_idx, cqe->res );
1536 0 : break;
1537 0 : case UDATA_OP_SHOVEL_NET:
1538 0 : shovel_comp_net( ctx, udata.conn_idx, cqe->res );
1539 0 : break;
1540 0 : case UDATA_OP_FUTEX:
1541 0 : futex_comp( ctx, udata.conn_idx, cqe->res );
1542 0 : break;
1543 0 : case UDATA_OP_TIMEOUT:
1544 0 : break;
1545 0 : default:
1546 0 : FD_LOG_CRIT(( "io_uring completion has invalid user_data 0x%016lx", (ulong)cqe->user_data ));
1547 0 : }
1548 0 : }
1549 :
1550 : /* stem_in_update grants flow control credits to an upstream producer. */
1551 :
1552 : static inline void
1553 : stem_in_update( fd_stem_tile_in_t * in );
1554 :
1555 : /* after_credit performs one io_uring event-loop iteration. */
1556 :
1557 : static void
1558 : after_credit( fd_snapsv_t * ctx,
1559 : fd_stem_context_t * stem,
1560 : int * opt_poll_in,
1561 0 : int * charge_busy ) {
1562 0 : (void)stem; (void)opt_poll_in;
1563 0 : fd_io_uring_t * ring = ctx->ring;
1564 :
1565 : /* Reap background completions */
1566 0 : uint ready = fd_io_uring_cq_ready( ring->cq );
1567 0 : for( uint i=0U; i<ready; i++ ) {
1568 0 : handle_cqe( ctx, fd_io_uring_cq_head( ring->cq ) );
1569 0 : fd_io_uring_cq_advance( ring->cq, 1U );
1570 0 : *charge_busy = 1;
1571 0 : }
1572 :
1573 : /* Prepare for sleep */
1574 0 : int waiting = !!futex_prep( ctx, 0UL );
1575 0 : stem_in_update( &stem->in[ 0 ] );
1576 :
1577 : /* Dispatch io_uring work, poll completions, and do a bounded sleep */
1578 0 : uint tail = ring->sq->sqe_tail;
1579 0 : atomic_store_explicit( ring->sq->ktail, tail, memory_order_release );
1580 0 : uint head = atomic_load_explicit( ring->sq->khead, memory_order_relaxed );
1581 0 : ring->sq->sqe_head = head;
1582 0 : uint to_submit = tail - head;
1583 0 : struct __kernel_timespec timeout = { .tv_nsec = (long)50e6L }; /* 50ms */
1584 0 : struct io_uring_getevents_arg enter_arg = { .ts = (ulong)&timeout };
1585 0 : int submitted = fd_io_uring_enter(
1586 0 : ring->ioring_fd, to_submit, !!waiting,
1587 0 : IORING_ENTER_GETEVENTS | IORING_ENTER_EXT_ARG,
1588 0 : &enter_arg, sizeof(enter_arg)
1589 0 : );
1590 0 : if( FD_UNLIKELY( submitted<0 ) ) {
1591 0 : if( FD_LIKELY( errno==ETIME ) ) {
1592 0 : submitted = 0; /* timeout, do housekeeping */
1593 0 : } else if( FD_LIKELY( errno==EINTR ) ) {
1594 0 : *charge_busy = !!submitted;
1595 0 : return;
1596 0 : } else {
1597 0 : FD_LOG_ERR(( "io_uring_enter failed (%i-%s)", errno, fd_io_strerror( errno ) ));
1598 0 : }
1599 0 : }
1600 0 : if( FD_UNLIKELY( fd_io_uring_sq_dropped( ring->sq ) ) ) {
1601 0 : FD_LOG_ERR(( "io_uring submission queue dropped entries" ));
1602 0 : }
1603 0 : if( FD_UNLIKELY( fd_io_uring_cq_overflow( ring->cq ) ) ) {
1604 0 : FD_LOG_ERR(( "io_uring completion queue overflowed" ));
1605 0 : }
1606 :
1607 : /* Reap deferred/immediate completions */
1608 0 : ready = fd_io_uring_cq_ready( ring->cq );
1609 0 : for( uint i=0U; i<ready; i++ ) {
1610 0 : handle_cqe( ctx, fd_io_uring_cq_head( ring->cq ) );
1611 0 : fd_io_uring_cq_advance( ring->cq, 1U );
1612 0 : *charge_busy = 1;
1613 0 : }
1614 0 : }
1615 :
1616 : /* snap_open opens the given snapshot by pool index. Silently ignores
1617 : ENOENT (rare race condition where the snapshot is immediately
1618 : deleted after being created.) */
1619 :
1620 : static void
1621 : snap_open( fd_snapsv_t * ctx,
1622 : ulong slot,
1623 : ulong base_slot,
1624 : ulong pool_idx,
1625 : ulong sz,
1626 0 : char const * name ) {
1627 0 : FD_CHECK_ERR( ctx->snap_cnt_full + ctx->snap_cnt_incr < ctx->snap_max, "too many snapshot files (snapmk_out desync?)" );
1628 0 : FD_CHECK_CRIT( pool_idx < ctx->snap_max, "invalid snapshot pool index" );
1629 :
1630 0 : ulong parsed_full_slot;
1631 0 : ulong parsed_incremental_slot;
1632 0 : uchar hash[ 32 ];
1633 0 : int is_zstd;
1634 0 : FD_CHECK_ERR( !fd_ssarchive_parse_filename( name, &parsed_full_slot, &parsed_incremental_slot, hash, &is_zstd ),
1635 0 : "invalid snapshot filename" );
1636 :
1637 0 : snap_entry_t * entry = snap_map_insert( ctx->snap_map, &(snap_key_t){slot, base_slot} );
1638 0 : FD_CHECK_ERR( !!entry, "snap_map_insert failed (snapmk_out desync?)" );
1639 :
1640 0 : memcpy( entry->hash, hash, sizeof(entry->hash) );
1641 0 : entry->sz = sz;
1642 0 : entry->fd = FD_SNAP_RO_FD( pool_idx );
1643 0 : entry->locked = 0;
1644 0 : entry->is_zstd = !!is_zstd;
1645 0 : if( base_slot==ULONG_MAX ) {
1646 0 : ctx->snap_cnt_full++;
1647 0 : ctx->newest_full = NULL;
1648 0 : } else {
1649 0 : ctx->snap_cnt_incr++;
1650 0 : ctx->newest_incr = NULL;
1651 0 : }
1652 0 : }
1653 :
1654 : /* snap_lock tries to acquire a read-lock to a snapshot file handle.
1655 : Returns 1 on success, 0 on failure (currently write locked).
1656 : Snapshots deliberately remain locked until the snapshot is about to
1657 : be deleted to prevent F_SETLK churn. */
1658 :
1659 : FD_FN_UNUSED static int
1660 0 : snap_lock( snap_entry_t * entry ) {
1661 0 : if( entry->locked ) return 1; /* already locked */
1662 0 : struct flock lock = {
1663 0 : .l_type = F_RDLCK,
1664 0 : .l_whence = SEEK_SET
1665 0 : };
1666 0 : if( FD_UNLIKELY( fcntl( entry->fd, F_SETLK, &lock ) ) ) {
1667 0 : if( errno==EAGAIN || errno==EACCES ) return 0; /* write locked */
1668 0 : FD_LOG_ERR(( "fcntl(F_RDLCK, %lu, %lu) failed (%i-%s)", entry->key.slot, entry->key.base_slot, errno, fd_io_strerror( errno ) ));
1669 0 : }
1670 0 : entry->locked = 1;
1671 0 : return 1;
1672 0 : }
1673 :
1674 : /* snap_close releases a snapshot file handle. */
1675 :
1676 : static void
1677 : snap_close( fd_snapsv_t * ctx,
1678 : ulong slot,
1679 0 : ulong base_slot ) {
1680 0 : snap_entry_t * entry = snap_map_update( ctx->snap_map, &(snap_key_t){slot, base_slot} );
1681 0 : if( FD_UNLIKELY( !entry ) ) return; /* ignore */
1682 :
1683 0 : if( entry->locked ) {
1684 0 : struct flock lock = {
1685 0 : .l_type = F_UNLCK,
1686 0 : .l_whence = SEEK_SET
1687 0 : };
1688 0 : if( FD_UNLIKELY( fcntl( entry->fd, F_SETLK, &lock ) ) ) {
1689 0 : FD_LOG_ERR(( "fcntl(F_UNLCK, %lu, %lu) failed (%i-%s)", slot, base_slot, errno, fd_io_strerror( errno ) ));
1690 0 : }
1691 0 : entry->locked = 0;
1692 0 : }
1693 0 : entry->fd = -1;
1694 :
1695 0 : snap_map_remove( ctx->snap_map, entry );
1696 0 : ctx->newest_full = ctx->newest_incr = NULL;
1697 0 : if( base_slot==ULONG_MAX ) {
1698 0 : FD_CHECK_CRIT( ctx->snap_cnt_full, "full snapshot count underflow" );
1699 0 : ctx->snap_cnt_full--;
1700 0 : } else {
1701 0 : FD_CHECK_CRIT( ctx->snap_cnt_incr, "incremental snapshot count underflow" );
1702 0 : ctx->snap_cnt_incr--;
1703 0 : }
1704 0 : }
1705 :
1706 : /* msg_snapmk is called for every snapmk_out frag. */
1707 :
1708 : static void
1709 : msg_snapmk( fd_snapsv_t * ctx,
1710 : ulong msg_type, /* sig */
1711 : fd_snapmk_msg_t const * msg,
1712 0 : ulong msg_sz ) {
1713 0 : switch( msg_type ) {
1714 0 : case FD_SNAPMK_MSG_FOUND: {
1715 0 : FD_CHECK_CRIT( msg_sz==sizeof(fd_snapmk_msg_found_t), "ABI mismatch" );
1716 0 : snap_open( ctx, msg->found.slot, msg->found.base_slot, msg->found.pool_idx, msg->found.sz, msg->found.name );
1717 0 : break;
1718 0 : }
1719 0 : case FD_SNAPMK_MSG_CREATED: {
1720 0 : FD_CHECK_CRIT( msg_sz==sizeof(fd_snapmk_msg_created_t), "ABI mismatch" );
1721 0 : snap_open( ctx, msg->created.slot, msg->created.base_slot, msg->created.pool_idx, msg->created.sz, msg->created.name );
1722 0 : break;
1723 0 : }
1724 0 : case FD_SNAPMK_MSG_DELETED: {
1725 0 : FD_CHECK_CRIT( msg_sz==sizeof(fd_snapmk_msg_deleted_t), "ABI mismatch" );
1726 0 : snap_close( ctx, msg->deleted.slot, msg->deleted.base_slot );
1727 0 : break;
1728 0 : }
1729 0 : default:
1730 0 : break;
1731 0 : }
1732 0 : }
1733 :
1734 : /* returnable_frag is called for every input frag */
1735 :
1736 : static int
1737 : returnable_frag( fd_snapsv_t * ctx,
1738 : ulong in_idx,
1739 : ulong seq,
1740 : ulong sig,
1741 : ulong chunk,
1742 : ulong sz,
1743 : ulong ctl,
1744 : ulong tsorig,
1745 : ulong tspub,
1746 0 : fd_stem_context_t * stem ) {
1747 0 : (void)ctl; (void)tsorig; (void)tspub; (void)stem;
1748 0 : ctx->in[ in_idx ].seq_cons = seq;
1749 0 : switch( ctx->in_kind[ in_idx ] ) {
1750 0 : case IN_KIND_SNAPMK:
1751 0 : FD_CHECK_CRIT( chunk >= ctx->in[ in_idx ].chunk0 &&
1752 0 : chunk <= ctx->in[ in_idx ].wmark &&
1753 0 : sz <= ctx->in[ in_idx ].mtu,
1754 0 : "input frag is out-of-bounds" );
1755 0 : fd_snapmk_msg_t const * msg = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
1756 0 : msg_snapmk( ctx, sig, msg, sz );
1757 0 : return 0; /* ok */
1758 0 : default:
1759 0 : FD_LOG_CRIT(( "unhandled frag from in_idx=%lu", in_idx ));
1760 0 : }
1761 0 : }
1762 :
1763 : static void
1764 0 : metrics_write( fd_snapsv_t * ctx ) {
1765 0 : FD_MGAUGE_SET( SNAPSV, SNAPSHOTS_AVAILABLE_FULL, ctx->snap_cnt_full );
1766 0 : FD_MGAUGE_SET( SNAPSV, SNAPSHOTS_AVAILABLE_INCREMENTAL, ctx->snap_cnt_incr );
1767 0 : FD_MGAUGE_SET( SNAPSV, CONN_ACTIVE, ctx->conn_cnt );
1768 0 : }
1769 :
1770 0 : #define STEM_BURST 1UL
1771 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapsv_t
1772 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapsv_t)
1773 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
1774 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
1775 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
1776 : #include "../../disco/stem/fd_stem.c"
1777 :
1778 : #ifndef FD_TILE_TEST
1779 : fd_topo_run_tile_t fd_tile_snapsv = {
1780 : .name = "snapsv",
1781 : .rlimit_file_cnt_fn = rlimit_file_cnt,
1782 : .populate_allowed_fds = populate_allowed_fds,
1783 : .populate_allowed_seccomp = populate_allowed_seccomp,
1784 : .scratch_align = scratch_align,
1785 : .scratch_footprint = scratch_footprint,
1786 : .privileged_init = privileged_init,
1787 : .unprivileged_init = unprivileged_init,
1788 : .run = stem_run,
1789 : .rlimit_nproc = 1024UL,
1790 : };
1791 : #endif
|
