Line data Source code
1 : /* The snapsv tile is a HTTP file server that serves local snapshots to
2 : remote peers.
3 :
4 : snapsv is entirely single-threaded and uses tango queues to communicate
5 : with other parts of the system (e.g. snapmk, the snapshot producer, and
6 : event feeds for the GUI server and telemetry client). snapsv scales by
7 : launching multiple independent tile instances using SO_REUSEPORT /
8 : receive-side scaling.
9 :
10 : This tile uses io_uring for network and file I/O.
11 :
12 : snapsv is fully cooperatively scheduled using io_uring waits.
13 : It wakes on network events, timeouts, file I/O completions, and
14 : futex wakes (new tango messages).
15 :
16 : Scheduling is event-driven (lazy). Apart from the initial accept op,
17 : every other op is triggered by a completion. */
18 :
19 : #define _GNU_SOURCE
20 : #include "fd_snapsv_tile.h"
21 : #include "fd_snapmk_tile.h"
22 : #include "../../disco/fd_clock_tile.h"
23 : #include "../../disco/stem/fd_stem.h"
24 : #include "../../disco/topo/fd_topo.h"
25 : #include "../../disco/metrics/fd_metrics.h"
26 : #include "../../util/net/fd_ip6.h"
27 : #include <errno.h>
28 : #include <fcntl.h>
29 : #include <netinet/in.h>
30 : #include <sys/socket.h>
31 : #include <sys/types.h>
32 : #include <linux/futex.h>
33 : #include "../../util/io_uring/fd_io_uring.h"
34 : #include "../../util/io_uring/fd_io_uring_setup.h"
35 : #include "../../util/io_uring/fd_io_uring_register.h"
36 : #include "../../third_party/picohttpparser/picohttpparser.h"
37 : #include "generated/fd_snapsv_tile_seccomp.h"
38 :
39 : /* Open addressed hash map of snapshots */
40 :
41 : struct snap_key {
42 : ulong slot;
43 : ulong base_slot; /* ULONG_MAX if full snap */
44 : };
45 : typedef struct snap_key snap_key_t;
46 :
47 : struct snap_entry {
48 : snap_key_t key;
49 : uchar hash[ 32 ];
50 : ulong sz;
51 : int fd;
52 : uint locked:1;
53 : uint is_zstd:1;
54 : };
55 : typedef struct snap_entry snap_entry_t;
56 :
57 : static ulong
58 : snap_key_hash( snap_key_t const * key,
59 3170823 : ulong seed ) {
60 : /* xxHash3 adapted */
61 3170823 : ulong k0 = key->base_slot;
62 3170823 : ulong k1 = key->slot;
63 :
64 3170823 : ulong lo = k0 ^ (0x6782737bea4239b9UL + seed);
65 3170823 : ulong hi = k1 ^ (0xaf56bc3b0996523aUL - seed);
66 :
67 3170823 : uint128 product = (uint128)lo * (uint128)hi;
68 3170823 : ulong fold = (ulong)product ^ (ulong)( product>>64 );
69 :
70 3170823 : ulong acc = 16UL + fd_ulong_bswap( lo ) + hi + fold;
71 3170823 : acc ^= acc >> 37;
72 3170823 : acc *= 0x165667919E3779F9UL;
73 3170823 : acc ^= acc >> 32;
74 3170823 : return acc;
75 3170823 : }
76 :
77 : #define MAP_NAME snap_map
78 7767507 : #define MAP_ELE_T snap_entry_t
79 : #define MAP_KEY_T snap_key_t
80 3453489 : #define MAP_KEY_EQ(k0,k1) (((k0)->slot==(k1)->slot) & ((k0)->base_slot==(k1)->base_slot))
81 3170760 : #define MAP_KEY_HASH(k,s) snap_key_hash( (k), (s) )
82 6760605 : #define MAP_ELE_IS_FREE(e) (((e)->key.slot==ULONG_MAX) & ((e)->key.base_slot==ULONG_MAX))
83 29823 : #define MAP_ELE_FREE(c,e) ((e)->key=(snap_key_t){ULONG_MAX,ULONG_MAX})
84 9975 : #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)
85 : #include "../../util/tmpl/fd_map_slot.c"
86 :
87 : /* Tile state */
88 :
89 : #define IN_LINK_MAX 1 /* cannot trivially grow */
90 :
91 : /* 8 ought to be enough ... The SQE/CQE count per conn is limited due
92 : to fixed depth pipelining. */
93 486 : #define SQE_PER_CONN (8UL)
94 486 : #define CQE_PER_CONN (8UL)
95 :
96 : /* io_uring fixed file indices */
97 :
98 324 : #define FIXED_FD_LISTEN (0U) /* TCP listen socket */
99 732 : #define FIXED_FD_CNT (1U)
100 :
101 : /* HTTP request head limits. The request head is accumulated in the
102 : conn's iobuf, therefore it can never exceed the iobuf size. */
103 :
104 183 : #define REQ_HEADER_MAX (64UL) /* max number of request headers */
105 :
106 : /* RES_HDR_MAX bounds the number of bytes that any of the
107 : build_{err,snap,redirect}_hdr functions write into the response
108 : buffer, including the NUL terminator. */
109 :
110 : #define RES_HDR_MAX (256UL)
111 :
112 : /* conn state */
113 :
114 5259 : #define CONN_STATE_FREE (0U) /* conn slot is unused */
115 324 : #define CONN_STATE_ALLOCATED (1U)
116 6660 : #define CONN_STATE_ACCEPT (2U) /* waiting for HTTP conn to be accepted */
117 936798 : #define CONN_STATE_REQ_PEEK (3U) /* peeking request bytes */
118 936216 : #define CONN_STATE_REQ_SKIP (4U) /* consume in-flight, head still incomplete */
119 468057 : #define CONN_STATE_REQ_DONE (5U) /* consume in-flight, head complete */
120 180 : #define CONN_STATE_RES_WRITE_ERR (6U) /* writing response header */
121 156 : #define CONN_STATE_RES_WRITE_HDR (7U) /* writing response header */
122 117 : #define CONN_STATE_RES_REDIRECT (8U) /* writing snap redirect */
123 36 : #define CONN_STATE_RES_SHOVEL (9U) /* pipelinined snap data RX & TX */
124 :
125 : struct snapsv_conn {
126 : uint state; /* CONN_STATE_{...} */
127 : uint iobuf_idx; /* iobuf for networking, UINT_MAX if none */
128 : uint rdbuf_idx; /* iobuf for disk read, UINT_MAX if none */
129 : uint rdbuf_len; /* completed bytes in rdbuf */
130 :
131 : uint sick:1; /* force close after current request? */
132 : uint disk_inflight:1; /* file read pending completion */
133 : uint net_inflight:1; /* tcp send pending completion */
134 : uint closing:1; /* close once shovel operations drain? */
135 : uint shutdown:1; /* idle sweep issued a socket shutdown? */
136 :
137 : struct {
138 : ulong req_seq; /* seq no of this HTTP request */
139 : uint len;
140 : uint skip_len; /* bytes requested by the in-flight CONSUME_FRAG */
141 : uint head:1; /* HEAD request? */
142 : uint get_snap:1; /* GET request for a snapshot? (streaming) */
143 : uint incremental:1; /* incremental snap requested? */
144 : } req;
145 :
146 : struct {
147 : uint sent;
148 : uint len;
149 : uint status; /* http status code */
150 : long hdr_ts; /* unix nanos timestamp when we started sending resp headers */
151 : uchar close_kind; /* FD_SNAPSV_CLOSE_* */
152 : } res;
153 :
154 : struct {
155 : snap_key_t key;
156 : ulong file_sz; /* snapshot file size */
157 : ulong req_off0; /* file offset of request */
158 : ulong req_off1; /* [req_off,req_off1) give the requested file range */
159 : ulong req_cur; /* next file offset to read */
160 : ulong req_sent; /* payload bytes handed to the socket */
161 : uint range:1; /* range request? */
162 : } snap;
163 :
164 : fd_kernel_timespec_t idle_timeout;
165 :
166 : /* active_ts: last time this conn did work worth keeping it alive for:
167 : accept, a completed request, or snapshot payload delivery. In unix
168 : nanoseconds. */
169 : long active_ts;
170 :
171 : fd_ip6_addr_t peer_ip;
172 : ushort peer_port;
173 : };
174 :
175 : typedef struct snapsv_conn snapsv_conn_t;
176 :
177 : struct fd_snapsv {
178 :
179 : uint kind_id;
180 :
181 : fd_clock_tile_t clock[1];
182 :
183 : /* io_uring */
184 : fd_io_uring_t ring[1];
185 : void * ring_shmem;
186 :
187 : /* I/O buffers */
188 : uchar * iobuf0;
189 : uint * iobuf_free;
190 : uint iobuf_free_cnt;
191 : uint iobuf_sz;
192 :
193 : /* TCP sockets */
194 : int listen_fd;
195 : uint conn0_fd_idx; /* io_uring file index */
196 : ulong conn_cnt;
197 : ulong conn_max;
198 : snapsv_conn_t * conn0;
199 : uint * conn_free;
200 : uint conn_free_cnt;
201 : long idle_timeout_nanos;
202 : fd_kernel_timespec_t send_timeout;
203 :
204 : /* Conn accept */
205 : uint accept_inflight:1;
206 : struct sockaddr_storage accept_addr;
207 : socklen_t accept_addr_len;
208 :
209 : /* snapshot hashmap */
210 : snap_map_t snap_map[1];
211 : ulong snap_cnt_full;
212 : ulong snap_cnt_incr;
213 : ulong snap_max;
214 :
215 : /* cache: newest snaps */
216 : snap_entry_t * newest_full;
217 : snap_entry_t * newest_incr;
218 :
219 : /* snapshot files */
220 : int dir_fd;
221 :
222 : /* tango input links */
223 : uchar in_kind[ IN_LINK_MAX ];
224 : struct {
225 : void * mem;
226 : ulong chunk0;
227 : ulong wmark;
228 : ulong mtu;
229 :
230 : fd_frag_meta_t const * mcache;
231 : ulong const * seq_prod; /* mcache->seq[0] */
232 : ulong seq_cons; /* last consumed seq */
233 : uint depth;
234 : uint futex_armed:1;
235 : } in[ IN_LINK_MAX ];
236 :
237 : /* streaming access logs */
238 : ulong req_seq;
239 : struct {
240 : long out_idx; /* -1 if the link is absent */
241 : fd_wksp_t * mem;
242 : ulong chunk0;
243 : ulong wmark;
244 : ulong chunk;
245 : long next_heartbeat; /* nanos */
246 : } out;
247 :
248 : };
249 :
250 : typedef struct fd_snapsv fd_snapsv_t;
251 :
252 : /* snapsv_udata_t represents io_uring user data. */
253 :
254 : union snapsv_udata {
255 : ulong user_data;
256 :
257 : struct __attribute__((packed)) {
258 : uint conn_idx;
259 : uchar op;
260 486 : #define UDATA_OP_ACCEPT ((uchar)0) /* accept new conn */
261 444 : #define UDATA_OP_PEEK ((uchar)1) /* read request bytes */
262 36 : #define UDATA_OP_CONSUME_FRAG ((uchar)2) /* consume request frag */
263 330 : #define UDATA_OP_CONSUME_TAIL ((uchar)3) /* consume complete request */
264 243 : #define UDATA_OP_WRITE_HDR ((uchar)4) /* write response header */
265 492 : #define UDATA_OP_SHOVEL_DISK ((uchar)5) /* read snapshot data */
266 474 : #define UDATA_OP_SHOVEL_NET ((uchar)6) /* send snapshot data */
267 29241 : #define UDATA_OP_FUTEX ((uchar)7) /* wait for tango frag */
268 1161 : #define UDATA_OP_TIMEOUT ((uchar)8) /* linked socket op timeout */
269 555 : #define UDATA_OP_SHUTDOWN ((uchar)9) /* idle sweep socket shutdown */
270 : };
271 : };
272 :
273 : typedef union snapsv_udata snapsv_udata_t;
274 :
275 : FD_STATIC_ASSERT( sizeof(snapsv_udata_t)==sizeof(ulong), layout );
276 :
277 288 : #define IN_KIND_SNAPMK 0
278 :
279 : static ulong
280 486 : scratch_align( void ) {
281 486 : return FD_SHMEM_NORMAL_PAGE_SZ;
282 486 : }
283 :
284 : FD_FN_PURE static ulong
285 324 : snapsv_fd_max( fd_topo_tile_t const * tile ) {
286 324 : return FIXED_FD_CNT + tile->snapsv.snap_max + tile->snapsv.conn_max;
287 324 : }
288 :
289 : FD_FN_PURE static ulong
290 486 : snapsv_map_ele_max( fd_topo_tile_t const * tile ) {
291 486 : return fd_ulong_pow2_up( tile->snapsv.snap_max );
292 486 : }
293 :
294 : FD_FN_PURE static ulong
295 486 : snapsv_sq_depth( fd_topo_tile_t const * tile ) {
296 486 : return fd_ulong_pow2_up( SQE_PER_CONN*tile->snapsv.conn_max );
297 486 : }
298 :
299 : FD_FN_PURE static ulong
300 486 : snapsv_cq_depth( fd_topo_tile_t const * tile ) {
301 486 : return fd_ulong_pow2_up( CQE_PER_CONN*tile->snapsv.conn_max );
302 486 : }
303 :
304 : static ulong
305 162 : scratch_footprint( fd_topo_tile_t const * tile ) {
306 162 : ulong sq_depth = snapsv_sq_depth( tile );
307 162 : ulong cq_depth = snapsv_cq_depth( tile );
308 162 : ulong l = FD_LAYOUT_INIT;
309 162 : l = FD_LAYOUT_APPEND( l, alignof(fd_snapsv_t), sizeof(fd_snapsv_t) );
310 162 : l = FD_LAYOUT_APPEND( l, fd_io_uring_shmem_align(), fd_io_uring_shmem_footprint( sq_depth, cq_depth ) );
311 162 : l = FD_LAYOUT_APPEND( l, snap_map_align(), snap_map_footprint( snapsv_map_ele_max( tile ) ) );
312 162 : l = FD_LAYOUT_APPEND( l, alignof(int), snapsv_fd_max( tile )*sizeof(int) );
313 162 : l = FD_LAYOUT_APPEND( l, alignof(snapsv_conn_t), tile->snapsv.conn_max*sizeof(snapsv_conn_t) );
314 162 : l = FD_LAYOUT_APPEND( l, alignof(uint), tile->snapsv.conn_max*sizeof(uint) );
315 162 : l = FD_LAYOUT_APPEND( l, alignof(uint), tile->snapsv.conn_max * 2 * sizeof(uint) );
316 162 : l = FD_LAYOUT_APPEND( l, FD_SHMEM_NORMAL_PAGE_SZ, tile->snapsv.conn_max * (tile->snapsv.send_buffer_size_kib<<11) );
317 162 : return FD_LAYOUT_FINI( l, scratch_align() );
318 162 : }
319 :
320 : /* snapsv_listen creates the TCP listen socket */
321 :
322 : static int
323 : snapsv_listen( fd_ip6_addr_t const * listen_addr,
324 : ushort listen_port,
325 0 : ulong conn_max ) {
326 0 : int ip4 = fd_ip6_addr_is_ip4_mapped( listen_addr->addr ) && !listen_addr->scope_id;
327 :
328 0 : int sock_fd = socket( ip4 ? AF_INET : AF_INET6, SOCK_STREAM|SOCK_NONBLOCK, 0 );
329 0 : if( FD_UNLIKELY( -1==sock_fd ) ) {
330 0 : FD_LOG_ERR(( "socket(AF_INET%s,SOCK_STREAM) failed (%i-%s)", ip4 ? "" : "6", errno, fd_io_strerror( errno ) ));
331 0 : }
332 :
333 : /* recover quickly from restart, if there are lingering TIME_WAIT conns */
334 0 : int optval = 1;
335 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(int) ) ) ) {
336 0 : FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_REUSEADDR) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
337 0 : }
338 : /* load balancing across multiple snapsv tiles */
339 0 : optval = 1;
340 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(int) ) ) ) {
341 0 : FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_REUSEPORT) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
342 0 : }
343 : /* TCP urgent meaningless for snapshot protocol */
344 0 : optval = 1;
345 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, SOL_SOCKET, SO_OOBINLINE, &optval, sizeof(int) ) ) ) {
346 0 : FD_LOG_ERR(( "setsockopt(SOL_SOCKET,SO_OOBINLINE) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
347 0 : }
348 :
349 0 : union {
350 0 : struct sockaddr_in ip4;
351 0 : struct sockaddr_in6 ip6;
352 0 : } addr = {0};
353 0 : ulong addr_sz;
354 :
355 0 : if( ip4 ) {
356 0 : addr.ip4 = (struct sockaddr_in){
357 0 : .sin_family = AF_INET,
358 0 : .sin_port = fd_ushort_bswap( listen_port ),
359 0 : .sin_addr.s_addr = fd_ip6_addr_to_ip4( listen_addr->addr )
360 0 : };
361 0 : addr_sz = sizeof(struct sockaddr_in);
362 0 : } else {
363 0 : int v6only = 0;
364 0 : if( FD_UNLIKELY( -1==setsockopt( sock_fd, IPPROTO_IPV6, IPV6_V6ONLY, &v6only, sizeof(int) ) ) ) {
365 0 : FD_LOG_ERR(( "setsockopt(IPPROTO_IPV6,IPV6_V6ONLY) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
366 0 : }
367 0 : addr.ip6 = (struct sockaddr_in6){
368 0 : .sin6_family = AF_INET6,
369 0 : .sin6_port = fd_ushort_bswap( listen_port ),
370 0 : .sin6_scope_id = listen_addr->scope_id
371 0 : };
372 0 : memcpy( addr.ip6.sin6_addr.s6_addr, listen_addr->addr, 16UL );
373 0 : addr_sz = sizeof(struct sockaddr_in6);
374 0 : }
375 :
376 0 : char addr_cstr[ FD_IP6_ADDR_CSTR_MAX ];
377 0 : if( FD_UNLIKELY( -1==bind( sock_fd, fd_type_pun( &addr ), (uint)addr_sz ) ) ) {
378 0 : FD_LOG_ERR(( "bind(%s:%u) failed (%i-%s)",
379 0 : fd_ip6_addr_cstr( addr_cstr, listen_addr ), listen_port,
380 0 : errno, fd_io_strerror( errno ) ));
381 0 : }
382 0 : if( FD_UNLIKELY( -1==listen( sock_fd, (int)conn_max ) ) ) {
383 0 : FD_LOG_ERR(( "listen() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
384 0 : }
385 :
386 0 : FD_LOG_NOTICE(( "snapshot server listening at %shttp://%s:%u%s",
387 0 : fd_log_style_bold(), fd_ip6_addr_cstr( addr_cstr, listen_addr ), listen_port,
388 0 : fd_log_style_normal() ));
389 :
390 0 : return sock_fd;
391 0 : }
392 :
393 : static void
394 : privileged_init( fd_topo_t const * topo,
395 0 : fd_topo_tile_t const * tile ) {
396 0 : ulong fd_max = snapsv_fd_max( tile );
397 0 : ulong sq_depth = snapsv_sq_depth( tile );
398 0 : ulong cq_depth = snapsv_cq_depth( tile );
399 0 : FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
400 0 : fd_snapsv_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapsv_t), sizeof(fd_snapsv_t) );
401 0 : void * ring_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_io_uring_shmem_align(), fd_io_uring_shmem_footprint( sq_depth, cq_depth ) );
402 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, snap_map_align(), snap_map_footprint( snapsv_map_ele_max( tile ) ) );
403 0 : int * fd_table = FD_SCRATCH_ALLOC_APPEND( l, alignof(int), fd_max*sizeof(int) );
404 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(snapsv_conn_t), tile->snapsv.conn_max*sizeof(snapsv_conn_t) );
405 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max*sizeof(uint) );
406 0 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max*2UL*sizeof(uint) );
407 0 : uchar * iobuf0 = FD_SCRATCH_ALLOC_APPEND( l, FD_SHMEM_NORMAL_PAGE_SZ, tile->snapsv.conn_max*(tile->snapsv.send_buffer_size_kib<<11) );
408 :
409 0 : memset( ctx, 0, sizeof(fd_snapsv_t) );
410 0 : memset( fd_table, 0xff, fd_max*sizeof(int) );
411 :
412 0 : FD_CHECK_ERR( tile->snapsv.conn_max, "snapsv conn_max is zero" );
413 0 : FD_CHECK_ERR( tile->snapsv.snap_max, "snapsv snap_max is zero" );
414 0 : ctx->snap_max = tile->snapsv.snap_max;
415 :
416 : /* TCP listen socket */
417 :
418 0 : ctx->conn_max = tile->snapsv.conn_max;
419 0 : ctx->listen_fd = snapsv_listen( &tile->snapsv.listen_addr, tile->snapsv.listen_port, tile->snapsv.conn_max );
420 :
421 : /* io_uring setup */
422 :
423 0 : ctx->ring_shmem = ring_mem;
424 0 : fd_io_uring_params_t params[1];
425 0 : fd_io_uring_params_init( params, (uint)sq_depth );
426 0 : params->flags |= FD_IORING_SETUP_COOP_TASKRUN;
427 :
428 0 : if( FD_UNLIKELY( !fd_io_uring_init_shmem( ctx->ring, params, ctx->ring_shmem, sq_depth, cq_depth ) ) ) {
429 0 : FD_LOG_ERR(( "fd_io_uring_init_shmem failed (%i-%s)", errno, fd_io_strerror( errno ) ));
430 0 : }
431 :
432 : /* registered buffer to help with DMA */
433 :
434 0 : struct iovec iovec = {
435 0 : .iov_base = iobuf0,
436 0 : .iov_len = tile->snapsv.conn_max*(tile->snapsv.send_buffer_size_kib<<11)
437 0 : };
438 0 : if( FD_UNLIKELY( fd_io_uring_register_buffers( ctx->ring->ioring_fd, &iovec, 1UL )<0 ) ) {
439 0 : FD_LOG_ERR(( "io_uring_register_buffers() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
440 0 : }
441 :
442 : /* Build file descriptor table */
443 0 : FD_TEST( tile->snapsv.snap_max<=FD_SNAP_MAX );
444 0 : fd_table[ FIXED_FD_LISTEN ] = ctx->listen_fd;
445 0 : uint fixed_fd = FIXED_FD_CNT;
446 0 : for( ulong i=0UL; i<tile->snapsv.snap_max; i++ ) {
447 0 : fd_table[ fixed_fd++ ] = FD_SNAP_RO_FD( i );
448 0 : }
449 0 : ctx->conn0_fd_idx = fixed_fd;
450 0 : if( FD_UNLIKELY( fd_io_uring_register_files( ctx->ring->ioring_fd, fd_table, fd_max )<0 ) ) {
451 0 : FD_LOG_ERR(( "io_uring_register_files() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
452 0 : }
453 :
454 0 : uint max_workers[2] = { tile->snapsv.io_worker_cnt, tile->snapsv.io_worker_cnt };
455 0 : if( FD_UNLIKELY( fd_io_uring_register( ctx->ring->ioring_fd, FD_IORING_REGISTER_IOWQ_MAX_WORKERS, max_workers, 2U )<0 ) ) {
456 0 : FD_LOG_ERR(( "io_uring_register(IORING_REGISTER_IOWQ_MAX_WORKERS) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
457 0 : }
458 :
459 0 : fd_io_uring_restriction_t restrictions[] = {
460 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_ACCEPT },
461 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_RECV },
462 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_SEND },
463 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_READ_FIXED },
464 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_FUTEX_WAIT },
465 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_LINK_TIMEOUT },
466 0 : { .opcode = FD_IORING_RESTRICTION_SQE_OP, .sqe_op = FD_IORING_OP_SHUTDOWN },
467 0 : { .opcode = FD_IORING_RESTRICTION_REGISTER_OP, /* deregister files */
468 0 : .register_op = FD_IORING_REGISTER_FILES_UPDATE },
469 0 : { .opcode = FD_IORING_RESTRICTION_SQE_FLAGS_ALLOWED,
470 0 : .sqe_flags = FD_IOSQE_IO_LINK },
471 0 : { .opcode = FD_IORING_RESTRICTION_SQE_FLAGS_REQUIRED,
472 0 : .sqe_flags = FD_IOSQE_FIXED_FILE },
473 0 : };
474 0 : if( FD_UNLIKELY( fd_io_uring_register_restrictions(
475 0 : ctx->ring->ioring_fd, restrictions,
476 0 : (uint)( sizeof(restrictions)/sizeof(restrictions[0]) ) )<0 ) ) {
477 0 : FD_LOG_ERR(( "io_uring_register_restrictions failed (%i-%s)", errno, fd_io_strerror( errno ) ));
478 0 : }
479 :
480 0 : if( FD_UNLIKELY( fd_io_uring_enable_rings( ctx->ring->ioring_fd )<0 ) ) {
481 0 : FD_LOG_ERR(( "io_uring_enable_rings failed (%i-%s)", errno, fd_io_strerror( errno ) ));
482 0 : }
483 :
484 0 : FD_TEST( fd_rng_secure( &ctx->snap_map->seed, sizeof(ulong) ) );
485 0 : }
486 :
487 : static void
488 : prep_accept( fd_snapsv_t * ctx );
489 :
490 : static void
491 : unprivileged_init( fd_topo_t const * topo,
492 162 : fd_topo_tile_t const * tile ) {
493 162 : ulong sq_depth = snapsv_sq_depth( tile );
494 162 : ulong cq_depth = snapsv_cq_depth( tile );
495 162 : FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
496 162 : fd_snapsv_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_snapsv_t), sizeof(fd_snapsv_t) );
497 162 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, fd_io_uring_shmem_align(), fd_io_uring_shmem_footprint( sq_depth, cq_depth ) );
498 162 : void * map_mem = FD_SCRATCH_ALLOC_APPEND( l, snap_map_align(), snap_map_footprint( snapsv_map_ele_max( tile ) ) );
499 162 : /* */(void)FD_SCRATCH_ALLOC_APPEND( l, alignof(int), snapsv_fd_max( tile )*sizeof(int) );
500 162 : snapsv_conn_t * conn0 = FD_SCRATCH_ALLOC_APPEND( l, alignof(snapsv_conn_t), tile->snapsv.conn_max*sizeof(snapsv_conn_t) );
501 162 : uint * conn_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max*sizeof(uint) );
502 162 : uint * iobuf_free = FD_SCRATCH_ALLOC_APPEND( l, alignof(uint), tile->snapsv.conn_max * 2 * sizeof(uint) );
503 162 : uchar * iobuf0 = FD_SCRATCH_ALLOC_APPEND( l, FD_SHMEM_NORMAL_PAGE_SZ, tile->snapsv.conn_max * (tile->snapsv.send_buffer_size_kib<<11) );
504 :
505 162 : ctx->kind_id = (uint)tile->kind_id;
506 :
507 162 : fd_clock_tile_init( ctx->clock );
508 :
509 : /* snapshot map setup */
510 :
511 162 : ulong map_ele_max = snapsv_map_ele_max( tile );
512 162 : snap_entry_t * slots = snap_map_new( map_mem, map_ele_max, 0 );
513 162 : FD_TEST( slots );
514 5346 : for( ulong i=0UL; i<map_ele_max; i++ ) {
515 5184 : slots[ i ].key = (snap_key_t){ ULONG_MAX,ULONG_MAX };
516 5184 : }
517 162 : ulong map_seed = ctx->snap_map->seed;
518 162 : FD_TEST( snap_map_join( ctx->snap_map, slots, map_ele_max, map_ele_max, map_seed ) );
519 162 : ctx->snap_max = tile->snapsv.snap_max;
520 :
521 : /* link setup */
522 :
523 162 : FD_CHECK_ERR( tile->in_cnt<=IN_LINK_MAX, "too many input links" );
524 162 : FD_CHECK_ERR( tile->in_cnt==1, "snapsv is hardcoded to only support one input link" );
525 324 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
526 162 : fd_topo_link_t const * link = &topo->links[ tile->in_link_id[ i ] ];
527 162 : FD_CHECK_ERR( !strcmp( link->name, "snapmk_out" ), "unexpected input link" );
528 162 : FD_CHECK_ERR( tile->in_link_poll[ i ], "expecting polled input link" );
529 162 : ctx->in_kind[ i ] = IN_KIND_SNAPMK;
530 162 : ctx->in[ i ].mem = fd_wksp_containing( link->dcache );
531 162 : ctx->in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->in[ i ].mem, link->dcache );
532 162 : ctx->in[ i ].wmark = fd_dcache_compact_wmark( ctx->in[ i ].mem, link->dcache, link->mtu );
533 162 : ctx->in[ i ].mtu = link->mtu;
534 :
535 162 : ulong depth = fd_mcache_depth( link->mcache );
536 162 : FD_CHECK_ERR( depth<=UINT_MAX, "input link mcache too deep" );
537 162 : ctx->in[ i ].mcache = link->mcache;
538 162 : ctx->in[ i ].depth = (uint)depth;
539 162 : ctx->in[ i ].seq_prod = fd_mcache_seq_laddr_const( link->mcache );
540 162 : ctx->in[ i ].seq_cons = ULONG_MAX;
541 162 : ctx->in[ i ].futex_armed = 0;
542 162 : }
543 :
544 162 : FD_CHECK_ERR( tile->out_cnt<=1UL, "too many output links" );
545 162 : memset( &ctx->out, 0, sizeof(ctx->out) );
546 162 : ctx->out.out_idx = -1L;
547 324 : for( ulong i=0UL; i<tile->out_cnt; i++ ) {
548 162 : fd_topo_link_t const * link = &topo->links[ tile->out_link_id[ i ] ];
549 162 : FD_CHECK_ERR( !strcmp( link->name, "snapsv_out" ), "unexpected output link" );
550 162 : FD_CHECK_ERR( link->mtu>=sizeof(fd_snapsv_msg_t), "snapsv_out link mtu too small" );
551 162 : ctx->out.out_idx = (long)i;
552 162 : ctx->out.mem = fd_wksp_containing( link->dcache );
553 162 : ctx->out.chunk0 = fd_dcache_compact_chunk0( ctx->out.mem, link->dcache );
554 162 : ctx->out.wmark = fd_dcache_compact_wmark ( ctx->out.mem, link->dcache, link->mtu );
555 162 : ctx->out.chunk = ctx->out.chunk0;
556 162 : }
557 162 : ctx->out.next_heartbeat = 0L;
558 :
559 : /* conn table */
560 :
561 162 : ctx->conn0 = conn0;
562 162 : ctx->conn_free = conn_free;
563 162 : ctx->conn_free_cnt = (uint)tile->snapsv.conn_max;
564 162 : ctx->idle_timeout_nanos = (long)( tile->snapsv.idle_timeout_millis * ((ulong)1e6) );
565 162 : long send_timeout_nanos = (long)( tile->snapsv.send_timeout_millis * ((ulong)1e6) );
566 162 : ctx->send_timeout = (fd_kernel_timespec_t) {
567 162 : .tv_sec = send_timeout_nanos / (long)1e9,
568 162 : .tv_nsec = send_timeout_nanos % (long)1e9
569 162 : };
570 5346 : for( ulong i=0UL; i<tile->snapsv.conn_max; i++ ) {
571 5184 : conn0[ i ] = (snapsv_conn_t){ .state = CONN_STATE_FREE, .iobuf_idx = UINT_MAX, .rdbuf_idx = UINT_MAX };
572 5184 : conn_free[ i ] = (uint)tile->snapsv.conn_max-1U-(uint)i;
573 5184 : }
574 :
575 : /* io bufs */
576 :
577 162 : ulong iobuf_sz = tile->snapsv.send_buffer_size_kib<<10;
578 162 : FD_CHECK_ERR( iobuf_sz, "send_buffer_size_kib is zero" );
579 162 : FD_CHECK_ERR( iobuf_sz>=RES_HDR_MAX, "send_buffer_size_kib is too small" );
580 162 : FD_CHECK_ERR( iobuf_sz<=UINT_MAX, "[snapshots.server.send_buffer_size_kib] is too large, must be at most 4194303 KiB" );
581 162 : ulong iobuf_cnt = tile->snapsv.conn_max * 2;
582 162 : ctx->iobuf_sz = (uint)iobuf_sz;
583 162 : ctx->iobuf0 = iobuf0;
584 162 : ctx->iobuf_free = iobuf_free;
585 162 : ctx->iobuf_free_cnt = (uint)iobuf_cnt;
586 10530 : for( ulong i=0UL; i<iobuf_cnt; i++ ) {
587 10368 : iobuf_free[ i ] = (uint)iobuf_cnt-1U-(uint)i;
588 10368 : }
589 :
590 : /* kick off main async op */
591 :
592 162 : prep_accept( ctx );
593 162 : }
594 :
595 : /* iobuf_alloc lends an iobuf to a conn. */
596 :
597 : static void
598 : iobuf_alloc( fd_snapsv_t * ctx,
599 633 : uint * slot ) {
600 633 : if( FD_LIKELY( *slot!=UINT_MAX ) ) return;
601 : /* Every conn holds at most two buffers (iobuf and rdbuf). This
602 : slot is empty, so this conn holds at most one; a pool sized for
603 : two buffers per conn therefore has a free buffer for this slot. */
604 630 : FD_CHECK_CRIT( ctx->iobuf_free_cnt, "iobuf pool exhausted" );
605 630 : *slot = ctx->iobuf_free[ --ctx->iobuf_free_cnt ];
606 630 : }
607 :
608 : /* iobuf_free frees a conn iobuf (if any). */
609 :
610 : static void
611 : iobuf_free( fd_snapsv_t * ctx,
612 630 : uint * slot ) {
613 630 : if( FD_UNLIKELY( *slot==UINT_MAX ) ) return;
614 543 : FD_CHECK_CRIT( ctx->iobuf_free_cnt < 2U*(uint)ctx->conn_max, "iobuf double free detected" );
615 543 : ctx->iobuf_free[ ctx->iobuf_free_cnt++ ] = *slot;
616 543 : *slot = UINT_MAX;
617 543 : }
618 :
619 : /* conn_alloc pops a new conn from the free stack / object pool. Returns
620 : the index of the allocated conn. */
621 :
622 : static uint
623 324 : conn_alloc( fd_snapsv_t * ctx ) {
624 324 : if( FD_UNLIKELY( !ctx->conn_free_cnt ) ) return UINT_MAX;
625 324 : uint conn_idx = ctx->conn_free[ --ctx->conn_free_cnt ];
626 324 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
627 324 : conn->state = CONN_STATE_ALLOCATED;
628 324 : FD_CHECK_CRIT( conn->iobuf_idx==UINT_MAX, "memory corruption detected" );
629 324 : FD_CHECK_CRIT( conn->rdbuf_idx==UINT_MAX, "memory corruption detected" );
630 324 : return conn_idx;
631 324 : }
632 :
633 : /* conn_free releases all conn resources and returns the conn to the free
634 : pool. */
635 :
636 : static void
637 : conn_free( fd_snapsv_t * ctx,
638 75 : uint conn_idx ) {
639 75 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
640 75 : FD_CHECK_CRIT( conn->state!=CONN_STATE_FREE, "double free detected" );
641 75 : FD_CHECK_CRIT( ctx->conn_free_cnt < ctx->conn_max, "double free detected" );
642 75 : iobuf_free( ctx, &conn->iobuf_idx );
643 75 : iobuf_free( ctx, &conn->rdbuf_idx );
644 75 : *conn = (snapsv_conn_t){ .state = CONN_STATE_FREE, .iobuf_idx = UINT_MAX, .rdbuf_idx = UINT_MAX };
645 75 : ctx->conn_free[ ctx->conn_free_cnt++ ] = conn_idx;
646 75 : }
647 :
648 : /* event_prepare prepares a snapsv_out frag (HTTP access log), which the
649 : caller can fill in with more detail. Returns a pointer to an
650 : unpublished frag in shm dcache. There can only be one in-preparation
651 : frag at a time.
652 :
653 : Currently, snapsv_out frags are only generated for "GET /snapshot-*"
654 : requests. This may change in the future. */
655 :
656 : static fd_snapsv_msg_snap_t *
657 : event_prepare( fd_snapsv_t const * ctx,
658 336 : uint conn_idx ) {
659 336 : snapsv_conn_t const * conn = &ctx->conn0[ conn_idx ];
660 336 : fd_snapsv_msg_t * msg = fd_chunk_to_laddr( ctx->out.mem, ctx->out.chunk );
661 336 : msg->snap = (fd_snapsv_msg_snap_t) {
662 336 : .key = (fd_snapsv_conn_key_t) {
663 336 : .kind_id = ctx->kind_id,
664 336 : .req_seq = conn->req.req_seq,
665 336 : .slot_idx = conn_idx
666 336 : },
667 336 : .slot = conn->snap.key.slot,
668 336 : .base_slot = conn->snap.key.base_slot,
669 336 : .snap_sz = conn->snap.file_sz,
670 336 : .req_off = conn->snap.req_off0,
671 336 : .req_sz = conn->snap.req_off1 - conn->snap.req_off0,
672 336 : .req_cur = conn->snap.req_sent,
673 336 : .req_ip = conn->peer_ip,
674 336 : .req_port = conn->peer_port,
675 336 : .resp_ts = conn->res.hdr_ts,
676 336 : .close_kind = conn->res.close_kind
677 336 : };
678 336 : return &msg->snap;
679 336 : }
680 :
681 : /* event_publish publishes the in-preparation snapsv_out frag (earlier
682 : alloc). */
683 :
684 : static void
685 : event_publish( fd_snapsv_t * ctx,
686 : fd_stem_context_t * stem,
687 : long now, /* unix nanos */
688 : int som,
689 336 : int eom ) {
690 336 : ulong sig = FD_SNAPSV_MSG_SNAP;
691 336 : ulong chunk = ctx->out.chunk;
692 336 : ulong sz = sizeof(fd_snapsv_msg_snap_t);
693 336 : ulong ctl = fd_frag_meta_ctl( 0UL, som, eom, 0 );
694 336 : ulong tspub = fd_frag_meta_ts_comp( now );
695 336 : ctx->out.chunk = fd_dcache_compact_next( chunk, sz, ctx->out.chunk0, ctx->out.wmark );
696 336 : fd_stem_publish( stem, (ulong)ctx->out.out_idx, sig, chunk, sz, ctl, 0UL, tspub );
697 336 : }
698 :
699 : /* event_snap emits an access log event for a snapshot request. Silently
700 : does nothing if the tile was configured without a snapsv_out link. */
701 :
702 : static void
703 : event_snap( fd_snapsv_t * ctx,
704 : fd_stem_context_t * stem,
705 : uint conn_idx,
706 : long now, /* unix nanos */
707 : int som,
708 336 : int eom ) {
709 336 : if( FD_UNLIKELY( ctx->out.out_idx < 0L ) ) return;
710 336 : event_prepare( ctx, conn_idx );
711 336 : event_publish( ctx, stem, now, som, eom );
712 336 : }
713 :
714 : /* prep_accept enqueues an io_uring conn accept SQE. */
715 :
716 : static void
717 324 : prep_accept( fd_snapsv_t * ctx ) {
718 324 : FD_CHECK_ERR( !ctx->accept_inflight, "accept already inflight" );
719 324 : FD_CHECK_ERR( ctx->conn_cnt < ctx->conn_max, "conn table full" );
720 :
721 324 : uint conn_idx = conn_alloc( ctx );
722 324 : FD_CHECK_CRIT( conn_idx!=UINT_MAX, "conn free list empty" );
723 324 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
724 324 : conn->state = CONN_STATE_ACCEPT;
725 :
726 324 : uint fd_idx = ctx->conn0_fd_idx + conn_idx;
727 324 : ctx->accept_addr_len = sizeof(ctx->accept_addr);
728 :
729 324 : fd_io_uring_t * ring = ctx->ring;
730 324 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ring->sq );
731 324 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
732 324 : *sqe = (fd_io_uring_sqe_t) {
733 324 : .opcode = FD_IORING_OP_ACCEPT,
734 324 : .flags = FD_IOSQE_FIXED_FILE,
735 324 : .fd = FIXED_FD_LISTEN,
736 324 : .off = (ulong)&ctx->accept_addr_len,
737 324 : .addr = (ulong)&ctx->accept_addr,
738 324 : .len = 0,
739 324 : .file_index = fd_idx + 1U /* registered files are 1-indexed */
740 324 : };
741 324 : snapsv_udata_t udata = { .op = UDATA_OP_ACCEPT, .conn_idx = conn_idx };
742 324 : sqe->user_data = udata.user_data;
743 324 : ctx->accept_inflight = 1;
744 324 : }
745 :
746 : /* conn_iobuf returns a pointer to the current I/O buffer of a conn. */
747 :
748 : FD_FN_PURE static uchar *
749 : conn_iobuf( fd_snapsv_t const * ctx,
750 972 : snapsv_conn_t const * conn ) {
751 972 : FD_CHECK_CRIT( conn->iobuf_idx!=UINT_MAX, "conn has no iobuf" );
752 972 : return ctx->iobuf0 + (ulong)conn->iobuf_idx*ctx->iobuf_sz;
753 972 : }
754 :
755 : /* prep_peek enqueues a receive op to copy any newly arrived bytes into
756 : iobuf. Does not advance the socket's file pointer. */
757 :
758 : static void
759 : prep_peek( fd_snapsv_t * ctx,
760 222 : uint conn_idx ) {
761 222 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
762 222 : long now = fd_log_wallclock();
763 222 : if( FD_UNLIKELY( !conn->active_ts ) ) {
764 162 : conn->active_ts = now;
765 162 : }
766 222 : long idle_rem = fd_long_max( 1L, ctx->idle_timeout_nanos - (now - conn->active_ts) );
767 222 : conn->idle_timeout = (fd_kernel_timespec_t) {
768 222 : .tv_sec = idle_rem / (long)1e9,
769 222 : .tv_nsec = idle_rem % (long)1e9
770 222 : };
771 222 : uint len = ctx->iobuf_sz - conn->req.len;
772 222 : FD_CHECK_CRIT( len, "request head buffer is full" );
773 222 : iobuf_alloc( ctx, &conn->iobuf_idx );
774 :
775 222 : fd_io_uring_t * ring = ctx->ring;
776 222 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ring->sq );
777 222 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
778 222 : *sqe = (fd_io_uring_sqe_t) {
779 222 : .opcode = FD_IORING_OP_RECV,
780 222 : .flags = FD_IOSQE_FIXED_FILE | FD_IOSQE_IO_LINK,
781 : //.ioprio = FD_IORING_RECVSEND_FIXED_BUF,
782 222 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
783 222 : .addr = (ulong)( conn_iobuf( ctx, conn ) + conn->req.len ),
784 222 : .len = len,
785 : //.buf_index = 0U,
786 222 : .msg_flags = MSG_PEEK
787 222 : };
788 222 : snapsv_udata_t udata = { .op = UDATA_OP_PEEK, .conn_idx = conn_idx };
789 222 : sqe->user_data = udata.user_data;
790 :
791 222 : sqe = fd_io_uring_get_sqe( ring->sq );
792 222 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
793 222 : *sqe = (fd_io_uring_sqe_t) {
794 222 : .opcode = FD_IORING_OP_LINK_TIMEOUT,
795 222 : .flags = FD_IOSQE_FIXED_FILE, /* ignored */
796 222 : .addr = (ulong)&conn->idle_timeout,
797 222 : .len = 1U
798 222 : };
799 222 : udata = (snapsv_udata_t){ .op = UDATA_OP_TIMEOUT };
800 222 : sqe->user_data = udata.user_data;
801 222 : }
802 :
803 : /* conn_req_next discards request state at the end of a request. Prepares
804 : for receipt of the next request (HTTP/1.1 keep-alive). */
805 :
806 : static void
807 : conn_req_next( fd_snapsv_t * ctx,
808 57 : uint conn_idx ) {
809 57 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
810 57 : memset( &conn->req, 0, sizeof(conn->req ) );
811 57 : memset( &conn->res, 0, sizeof(conn->res ) );
812 57 : memset( &conn->snap, 0, sizeof(conn->snap) );
813 57 : conn->snap.key = (snap_key_t){ ULONG_MAX, ULONG_MAX };
814 57 : conn->state = CONN_STATE_REQ_PEEK;
815 57 : prep_peek( ctx, conn_idx );
816 57 : }
817 :
818 : /* prep_consume consumes previously received (via peek) iobuf bytes. */
819 :
820 : static void
821 : prep_consume( fd_snapsv_t * ctx,
822 : uint conn_idx,
823 : uint len,
824 183 : uchar op ) {
825 183 : FD_CHECK_CRIT( len, "consuming zero bytes" );
826 183 : fd_io_uring_t * ring = ctx->ring;
827 183 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ring->sq );
828 183 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
829 183 : *sqe = (fd_io_uring_sqe_t) {
830 183 : .opcode = FD_IORING_OP_RECV,
831 183 : .flags = FD_IOSQE_FIXED_FILE,
832 183 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
833 183 : .addr = 0UL,
834 183 : .len = len,
835 183 : .msg_flags = MSG_TRUNC | MSG_WAITALL
836 183 : };
837 183 : snapsv_udata_t udata = { .op = op, .conn_idx = conn_idx };
838 183 : sqe->user_data = udata.user_data;
839 183 : }
840 :
841 : /* conn_close destroys a conn object. If no accept op is in-flight,
842 : enqueues one. */
843 :
844 : static void
845 : conn_close( fd_snapsv_t * ctx,
846 : fd_stem_context_t * stem,
847 : uint conn_idx,
848 75 : long now ) {
849 75 : FD_CHECK_CRIT( conn_idx < ctx->conn_max, "invalid conn_idx" );
850 75 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
851 :
852 75 : if( conn->req.get_snap ) {
853 : /* the transfer is ending before the whole range was sent, so this is
854 : not a clean completion unless a more specific reason was recorded */
855 9 : if( conn->res.close_kind==FD_SNAPSV_CLOSE_DONE &&
856 9 : conn->snap.req_cur < conn->snap.req_off1 ) {
857 0 : conn->res.close_kind = FD_SNAPSV_CLOSE_NET;
858 0 : }
859 9 : event_snap( ctx, stem, conn_idx, now, 0, /* eom */ 1 );
860 9 : conn->req.get_snap = 0;
861 9 : }
862 :
863 : #ifndef FD_TILE_TEST
864 : /* io_uring OP_CLOSE on a fixed file does not use IOSQE_FIXED_FILE,
865 : therefore would break the io_uring sandbox (restrictions). */
866 : uint fd_idx = ctx->conn0_fd_idx + conn_idx;
867 : int unreg = -1;
868 0 : if( FD_UNLIKELY( fd_io_uring_register_files_update( ctx->ring->ioring_fd, fd_idx, &unreg, 1U )<0 ) ) {
869 0 : FD_LOG_ERR(( "io_uring_register(IORING_REGISTER_FILES_UPDATE) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
870 0 : }
871 0 : #endif
872 :
873 : /* conn no longer considered active */
874 75 : FD_CHECK_CRIT( ctx->conn_cnt, "conn count underflow" );
875 75 : ctx->conn_cnt--;
876 :
877 : /* return dead conn object */
878 75 : conn_free( ctx, conn_idx );
879 :
880 : /* allocate a new conn object, and schedule an accept() for it */
881 75 : if( FD_UNLIKELY( !ctx->accept_inflight ) ) {
882 0 : prep_accept( ctx );
883 0 : }
884 75 : }
885 :
886 : /* conn_req_done finishes a response that has no more bytes to send.
887 : Closes the conn if the client asked for a close, otherwise waits for
888 : the next request. */
889 :
890 : static void
891 : conn_req_done( fd_snapsv_t * ctx,
892 : fd_stem_context_t * stem,
893 : uint conn_idx,
894 69 : long now ) {
895 69 : if( FD_UNLIKELY( ctx->conn0[ conn_idx ].sick ) ) {
896 12 : conn_close( ctx, stem, conn_idx, now );
897 12 : return;
898 12 : }
899 57 : ctx->conn0[ conn_idx ].active_ts = now;
900 57 : conn_req_next( ctx, conn_idx );
901 57 : }
902 :
903 : static ulong
904 : populate_allowed_fds( fd_topo_t const * topo,
905 : fd_topo_tile_t const * tile,
906 : ulong out_fds_cnt,
907 0 : int * out_fds ) {
908 0 : fd_snapsv_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
909 0 : if( FD_UNLIKELY( out_fds_cnt<4+ctx->snap_max ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
910 0 : ulong out_cnt = 0UL;
911 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
912 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
913 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
914 0 : out_fds[ out_cnt++ ] = ctx->ring->ioring_fd;
915 0 : out_fds[ out_cnt++ ] = ctx->listen_fd;
916 0 : for( ulong i=0UL; i<ctx->snap_max; i++ )
917 0 : out_fds[ out_cnt++ ] = FD_SNAP_RO_FD( i ); /* snapshot pool */
918 0 : return out_cnt;
919 0 : }
920 :
921 : static ulong
922 : populate_allowed_seccomp( fd_topo_t const * topo,
923 : fd_topo_tile_t const * tile,
924 : ulong out_cnt,
925 0 : struct sock_filter * out ) {
926 0 : fd_snapsv_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
927 0 : FD_CHECK_ERR( ctx->snap_max, "snapsv snap_max is zero" );
928 0 : populate_sock_filter_policy_fd_snapsv_tile(
929 0 : out_cnt, out,
930 0 : (uint)fd_log_private_logfile_fd(),
931 0 : (uint)ctx->ring->ioring_fd,
932 0 : (uint)FD_SNAP_RO_FD( 0 ),
933 0 : (uint)FD_SNAP_RO_FD( ctx->snap_max-1UL ) );
934 0 : return sock_filter_policy_fd_snapsv_tile_instr_cnt;
935 0 : }
936 :
937 : static ulong
938 : rlimit_file_cnt( fd_topo_t const * topo,
939 0 : fd_topo_tile_t const * tile ) {
940 0 : (void)topo;
941 : /* stderr, logfile, boot control pipe, io_uring, listen socket, one
942 : spare for accept, the snapshot pool, and active connections */
943 0 : return 6UL + tile->snapsv.snap_max + tile->snapsv.conn_max;
944 0 : }
945 :
946 : /* handle_accept handles the completion of an accept op. */
947 :
948 : static void
949 : handle_accept( fd_snapsv_t * ctx,
950 : uint conn_idx,
951 : int res,
952 162 : long now ) {
953 162 : ctx->accept_inflight = 0;
954 :
955 162 : if( FD_UNLIKELY( res<0 ) ) {
956 0 : FD_LOG_WARNING(( "accept() failed (%i-%s)", -res, fd_io_strerror( -res ) ));
957 0 : conn_free( ctx, conn_idx );
958 0 : prep_accept( ctx );
959 0 : return;
960 0 : }
961 :
962 162 : ctx->conn_cnt++;
963 :
964 : /* set conn addr */
965 162 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
966 162 : switch( ctx->accept_addr.ss_family ) {
967 162 : case AF_INET: {
968 162 : struct sockaddr_in const * addr = fd_type_pun_const( &ctx->accept_addr );
969 162 : fd_ip6_addr_ip4_mapped( conn->peer_ip.addr, addr->sin_addr.s_addr );
970 162 : conn->peer_ip.scope_id = 0U;
971 162 : conn->peer_port = fd_ushort_bswap( addr->sin_port );
972 162 : break;
973 0 : }
974 0 : case AF_INET6: {
975 0 : struct sockaddr_in6 const * addr = fd_type_pun_const( &ctx->accept_addr );
976 0 : memcpy( conn->peer_ip.addr, addr->sin6_addr.s6_addr, 16UL );
977 0 : conn->peer_ip.scope_id = addr->sin6_scope_id;
978 0 : conn->peer_port = fd_ushort_bswap( addr->sin6_port );
979 0 : break;
980 0 : }
981 0 : default:
982 0 : FD_LOG_WARNING(( "accept() returned unknown address family %i", (int)ctx->accept_addr.ss_family ));
983 0 : memset( &conn->peer_ip, 0, sizeof(conn->peer_ip) );
984 0 : conn->peer_port = 0U;
985 162 : }
986 :
987 162 : conn->active_ts = now;
988 162 : conn->state = CONN_STATE_REQ_PEEK;
989 :
990 162 : prep_peek( ctx, conn_idx );
991 162 : if( FD_LIKELY( ctx->conn_cnt<ctx->conn_max ) ) {
992 162 : prep_accept( ctx );
993 162 : }
994 162 : }
995 :
996 : /* conn_snap_entry checks that the snapshot for the current operation is
997 : still valid. The return value of this function is invalidated once
998 : after_credit returns. */
999 :
1000 : static snap_entry_t *
1001 : conn_snap_entry( fd_snapsv_t * ctx,
1002 621 : snapsv_conn_t * conn ) {
1003 621 : FD_CHECK_ERR( conn->snap.key.slot!=ULONG_MAX, "no snapshot selected" );
1004 621 : snap_entry_t * entry = snap_map_update( ctx->snap_map, &conn->snap.key );
1005 621 : if( FD_UNLIKELY( !entry || entry->sz!=conn->snap.file_sz ) ) return NULL;
1006 546 : return entry;
1007 621 : }
1008 :
1009 : static int
1010 : snap_lock( snap_entry_t * entry );
1011 :
1012 : /* shovel dispatches snapshot streaming I/O work.
1013 : This forms a 2-job deep pipeline that attempts to concurrently do
1014 : disk read and network write I/O. */
1015 :
1016 : static void
1017 : shovel( fd_snapsv_t * ctx,
1018 : fd_stem_context_t * stem,
1019 : uint conn_idx,
1020 516 : long now ) {
1021 516 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1022 516 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_SHOVEL, "conn state confusion" );
1023 :
1024 516 : if( FD_UNLIKELY( conn->closing ) ) {
1025 9 : if( !conn->disk_inflight && !conn->net_inflight ) {
1026 9 : conn_close( ctx, stem, conn_idx, now );
1027 9 : }
1028 9 : return;
1029 9 : }
1030 :
1031 507 : if( conn->iobuf_idx!=UINT_MAX &&
1032 507 : conn->res.sent==conn->res.len &&
1033 507 : !conn->net_inflight ) {
1034 237 : iobuf_free( ctx, &conn->iobuf_idx );
1035 237 : conn->res.sent = 0U;
1036 237 : conn->res.len = 0U;
1037 237 : }
1038 :
1039 : /* disk read buffer complete, rename to net send buffer */
1040 507 : if( conn->iobuf_idx==UINT_MAX && conn->rdbuf_len ) {
1041 237 : FD_CHECK_CRIT( conn->rdbuf_idx!=UINT_MAX, "completed read has no buffer" );
1042 237 : conn->iobuf_idx = conn->rdbuf_idx;
1043 237 : conn->rdbuf_idx = UINT_MAX;
1044 237 : conn->res.sent = 0U;
1045 237 : conn->res.len = conn->rdbuf_len;
1046 237 : conn->rdbuf_len = 0U;
1047 237 : }
1048 :
1049 : /* generate net send SQE */
1050 507 : if( conn->iobuf_idx!=UINT_MAX && conn->res.sent<conn->res.len && !conn->net_inflight ) {
1051 237 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ctx->ring->sq );
1052 237 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1053 237 : *sqe = (fd_io_uring_sqe_t) {
1054 237 : .opcode = FD_IORING_OP_SEND,
1055 237 : .flags = FD_IOSQE_FIXED_FILE | FD_IOSQE_IO_LINK,
1056 : //.ioprio = FD_IORING_RECVSEND_FIXED_BUF,
1057 237 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
1058 237 : .addr = (ulong)conn_iobuf( ctx, conn ) + conn->res.sent,
1059 237 : .len = conn->res.len - conn->res.sent,
1060 : //.buf_index = 0U,
1061 237 : .msg_flags = MSG_NOSIGNAL
1062 237 : };
1063 237 : snapsv_udata_t udata = { .op = UDATA_OP_SHOVEL_NET, .conn_idx = conn_idx };
1064 237 : sqe->user_data = udata.user_data;
1065 237 : conn->net_inflight = 1U;
1066 :
1067 237 : sqe = fd_io_uring_get_sqe( ctx->ring->sq );
1068 237 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1069 237 : *sqe = (fd_io_uring_sqe_t) {
1070 237 : .opcode = FD_IORING_OP_LINK_TIMEOUT,
1071 237 : .flags = FD_IOSQE_FIXED_FILE, /* ignored */
1072 237 : .addr = (ulong)&ctx->send_timeout,
1073 237 : .len = 1U
1074 237 : };
1075 237 : udata = (snapsv_udata_t){ .op = UDATA_OP_TIMEOUT };
1076 237 : sqe->user_data = udata.user_data;
1077 237 : }
1078 :
1079 : /* generate disk read SQE */
1080 507 : if( conn->rdbuf_idx==UINT_MAX &&
1081 507 : !conn->disk_inflight &&
1082 507 : conn->snap.req_cur < conn->snap.req_off1 ) {
1083 246 : snap_entry_t const * snap = conn_snap_entry( ctx, conn );
1084 246 : if( FD_UNLIKELY( !snap ) ) {
1085 0 : conn->res.close_kind = FD_SNAPSV_CLOSE_ABORT;
1086 0 : conn->closing = 1U;
1087 0 : shovel( ctx, stem, conn_idx, now );
1088 0 : return;
1089 0 : }
1090 :
1091 246 : iobuf_alloc( ctx, &conn->rdbuf_idx );
1092 246 : ulong read_sz = fd_ulong_min( ctx->iobuf_sz, conn->snap.req_off1-conn->snap.req_cur );
1093 246 : ulong pool_idx = (ulong)( snap->fd - FD_SNAP_RO_FD( 0 ) );
1094 246 : FD_CHECK_CRIT( pool_idx<ctx->snap_max, "snapshot has invalid file descriptor" );
1095 :
1096 246 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ctx->ring->sq );
1097 246 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1098 246 : *sqe = (fd_io_uring_sqe_t) {
1099 246 : .opcode = FD_IORING_OP_READ_FIXED,
1100 246 : .flags = FD_IOSQE_FIXED_FILE,
1101 246 : .fd = (int)( FIXED_FD_CNT + pool_idx ),
1102 246 : .off = conn->snap.req_cur,
1103 246 : .addr = (ulong)( ctx->iobuf0 + (ulong)conn->rdbuf_idx*ctx->iobuf_sz ),
1104 246 : .len = (uint)read_sz,
1105 246 : .buf_index = 0U
1106 246 : };
1107 246 : snapsv_udata_t udata = { .op = UDATA_OP_SHOVEL_DISK, .conn_idx = conn_idx };
1108 246 : sqe->user_data = udata.user_data;
1109 246 : conn->disk_inflight = 1U;
1110 246 : }
1111 :
1112 : /* everything done */
1113 507 : if( FD_UNLIKELY( conn->snap.req_cur>=conn->snap.req_off1 &&
1114 507 : conn->iobuf_idx==UINT_MAX &&
1115 507 : conn->rdbuf_idx==UINT_MAX &&
1116 507 : !conn->disk_inflight &&
1117 507 : !conn->net_inflight ) ) {
1118 24 : conn->res.close_kind = FD_SNAPSV_CLOSE_DONE;
1119 24 : event_snap( ctx, stem, conn_idx, now, 0, 1 ); /* eom */
1120 24 : conn->req.get_snap = 0;
1121 24 : conn_req_done( ctx, stem, conn_idx, now );
1122 24 : }
1123 507 : }
1124 :
1125 : /* shovel_comp_disk reacts to a snapshot streaming disk read completion. */
1126 :
1127 : static void
1128 : shovel_comp_disk( fd_snapsv_t * ctx,
1129 : fd_stem_context_t * stem,
1130 : uint conn_idx,
1131 : int res,
1132 246 : long now ) {
1133 246 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1134 246 : FD_CHECK_CRIT( conn->state==CONN_STATE_RES_SHOVEL && conn->disk_inflight,
1135 246 : "unexpected shovel disk completion" );
1136 246 : conn->disk_inflight = 0U;
1137 :
1138 : /* snapshot file recycled between submission and completion? */
1139 246 : if( FD_UNLIKELY( !conn_snap_entry( ctx, conn ) ) ) {
1140 9 : conn->res.close_kind = FD_SNAPSV_CLOSE_ABORT;
1141 9 : conn->closing = 1U;
1142 9 : shovel( ctx, stem, conn_idx, now );
1143 9 : return;
1144 9 : }
1145 :
1146 237 : ulong rem = conn->snap.req_off1 - conn->snap.req_cur;
1147 237 : ulong read_sz = fd_ulong_min( ctx->iobuf_sz, rem );
1148 237 : if( FD_UNLIKELY( res<=0 || (ulong)res>read_sz ) ) {
1149 0 : if( res<0 ) {
1150 0 : FD_LOG_WARNING(( "snapshot read failed (%i-%s)", -res, fd_io_strerror( -res ) ));
1151 0 : } else if( !res ) {
1152 0 : FD_LOG_WARNING(( "snapshot file ended early" ));
1153 0 : } else {
1154 0 : FD_LOG_WARNING(( "snapshot read returned too many bytes (%i>%lu)", res, read_sz ));
1155 0 : }
1156 0 : conn->res.close_kind = FD_SNAPSV_CLOSE_ABORT;
1157 0 : conn->closing = 1U;
1158 0 : shovel( ctx, stem, conn_idx, now );
1159 0 : return;
1160 0 : }
1161 :
1162 237 : conn->snap.req_cur += (ulong)res;
1163 237 : conn->rdbuf_len = (uint)res;
1164 237 : shovel( ctx, stem, conn_idx, now );
1165 237 : }
1166 :
1167 : /* shovel_comp_net reacts to a snapshot streaming network write
1168 : completion. */
1169 :
1170 : static void
1171 : shovel_comp_net( fd_snapsv_t * ctx,
1172 : fd_stem_context_t * stem,
1173 : uint conn_idx,
1174 : int res,
1175 237 : long now ) {
1176 237 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1177 237 : FD_CHECK_CRIT( conn->state==CONN_STATE_RES_SHOVEL, "unexpected shovel net completion" );
1178 237 : FD_CHECK_CRIT( conn->net_inflight, "unexpected SEND completion" );
1179 237 : conn->net_inflight = 0U;
1180 :
1181 237 : if( FD_UNLIKELY( res<=0 || (uint)res>conn->res.len-conn->res.sent ) ) {
1182 0 : FD_IP6_ADDR_CSTR( addr_cstr, &conn->peer_ip );
1183 0 : if( res<0 ) {
1184 0 : FD_LOG_INFO(( "snapshot download peer %s:%u: snapshot data send error (%i-%s)",
1185 0 : addr_cstr, conn->peer_port, -res, fd_io_strerror( -res ) ));
1186 0 : }
1187 0 : conn->res.close_kind = FD_SNAPSV_CLOSE_NET;
1188 0 : conn->closing = 1U;
1189 0 : shovel( ctx, stem, conn_idx, now );
1190 0 : return;
1191 0 : }
1192 :
1193 237 : conn->res.sent += (uint)res;
1194 237 : conn->snap.req_sent += (ulong)res;
1195 237 : conn->active_ts = now;
1196 237 : FD_MCNT_INC( SNAPSV, HTTP_TX_BYTES, (ulong)res );
1197 237 : shovel( ctx, stem, conn_idx, now );
1198 237 : }
1199 :
1200 : /* prep_write_hdr enqueues a header write op. */
1201 :
1202 : static void
1203 : prep_write_hdr( fd_snapsv_t * ctx,
1204 165 : uint conn_idx ) {
1205 165 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1206 165 : fd_io_uring_t * ring = ctx->ring;
1207 165 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ring->sq );
1208 165 : FD_CHECK_ERR( conn->res.sent < conn->res.len, "conn res state confusion" );
1209 165 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1210 165 : *sqe = (fd_io_uring_sqe_t) {
1211 165 : .opcode = FD_IORING_OP_SEND,
1212 165 : .flags = FD_IOSQE_FIXED_FILE | FD_IOSQE_IO_LINK,
1213 : //.ioprio = FD_IORING_RECVSEND_FIXED_BUF,
1214 165 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
1215 165 : .addr = (ulong)conn_iobuf( ctx, conn ) + conn->res.sent,
1216 165 : .len = conn->res.len - conn->res.sent,
1217 : //.buf_index = 0U,
1218 165 : .msg_flags = MSG_NOSIGNAL
1219 165 : };
1220 165 : snapsv_udata_t udata = { .op = UDATA_OP_WRITE_HDR, .conn_idx = conn_idx };
1221 165 : sqe->user_data = udata.user_data;
1222 :
1223 165 : sqe = fd_io_uring_get_sqe( ring->sq );
1224 165 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1225 165 : *sqe = (fd_io_uring_sqe_t) {
1226 165 : .opcode = FD_IORING_OP_LINK_TIMEOUT,
1227 165 : .flags = FD_IOSQE_FIXED_FILE, /* ignored */
1228 165 : .addr = (ulong)&ctx->send_timeout,
1229 165 : .len = 1U
1230 165 : };
1231 165 : udata = (snapsv_udata_t){ .op = UDATA_OP_TIMEOUT };
1232 165 : sqe->user_data = udata.user_data;
1233 165 : }
1234 :
1235 : static ulong
1236 : build_err_hdr( char out[ static RES_HDR_MAX ], /* not a cstr */
1237 : uint http_err,
1238 : ulong object_sz, /* 'Content-Range' hint */
1239 114 : int sick ) { /* set 'Connection: close'? */
1240 114 : char * p = fd_cstr_init( out );
1241 114 : p = fd_cstr_append_cstr( p, "HTTP/1.1 ");
1242 114 : switch( http_err ) {
1243 30 : case 400: p = fd_cstr_append_cstr( p, "400 Bad Request" ); break;
1244 60 : case 404: p = fd_cstr_append_cstr( p, "404 Not Found" ); break;
1245 18 : case 416: p = fd_cstr_append_cstr( p, "416 Range Not Satisfiable" ); break;
1246 6 : default: p = fd_cstr_append_cstr( p, "500 Internal Server Error" ); break;
1247 114 : }
1248 114 : p = fd_cstr_append_cstr( p, "\r\n" );
1249 114 : if( http_err==416 && object_sz!=ULONG_MAX ) {
1250 15 : p = fd_cstr_append_cstr( p, "Content-Range: bytes */" );
1251 15 : p = fd_cstr_append_ulong_as_text( p, 0, 0, object_sz, fd_ulong_base10_dig_cnt( object_sz ) );
1252 15 : p = fd_cstr_append_cstr( p, "\r\n" );
1253 15 : }
1254 114 : p = fd_cstr_append_cstr( p, "Content-Length: 0\r\n" );
1255 114 : if( sick ) {
1256 39 : p = fd_cstr_append_cstr( p, "Connection: close\r\n" );
1257 39 : }
1258 114 : p = fd_cstr_append_cstr( p, "\r\n" );
1259 114 : return (ulong)( p - out );
1260 114 : }
1261 :
1262 : /* serve_http_err dispatches a RES_WRITE_HDR job for a generic HTTP
1263 : error page. */
1264 :
1265 : static void
1266 : serve_http_err( fd_snapsv_t * ctx,
1267 84 : uint conn_idx ) {
1268 84 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1269 84 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_WRITE_ERR, "state confusion" );
1270 84 : iobuf_alloc( ctx, &conn->iobuf_idx );
1271 84 : uchar * iobuf = conn_iobuf( ctx, conn );
1272 84 : snap_entry_t const * snap = conn->snap.key.slot!=ULONG_MAX ? conn_snap_entry( ctx, conn ) : NULL;
1273 84 : conn->res.sent = 0;
1274 84 : conn->res.len = (uint)build_err_hdr(
1275 84 : (char *)iobuf,
1276 84 : conn->res.status,
1277 84 : snap ? snap->sz : ULONG_MAX,
1278 84 : conn->sick );
1279 84 : prep_write_hdr( ctx, conn_idx );
1280 84 : }
1281 :
1282 : static ulong
1283 : build_snap_res_hdr( char out[ static RES_HDR_MAX ],
1284 : int range,
1285 : int is_zstd,
1286 : ulong range0,
1287 : ulong range1,
1288 : ulong object_sz,
1289 81 : int sick ) { /* set 'Connection: close'? */
1290 81 : char * p = fd_cstr_init( out );
1291 81 : if( range ) {
1292 33 : p = fd_cstr_append_cstr( p, "HTTP/1.1 206 Partial Content\r\n" );
1293 48 : } else {
1294 48 : p = fd_cstr_append_cstr( p, "HTTP/1.1 200 OK\r\n" );
1295 48 : }
1296 81 : if( is_zstd ) {
1297 72 : p = fd_cstr_append_cstr( p, "Content-Type: application/zstd\r\n" );
1298 72 : } else {
1299 9 : p = fd_cstr_append_cstr( p, "Content-Type: application/x-tar\r\n" );
1300 9 : }
1301 81 : p = fd_cstr_append_cstr( p, "Accept-Ranges: bytes\r\n" );
1302 81 : if( range ) {
1303 33 : p = fd_cstr_append_cstr( p, "Content-Range: bytes " );
1304 33 : p = fd_cstr_append_ulong_as_text( p, 0, 0, range0, fd_ulong_base10_dig_cnt( range0 ) );
1305 33 : p = fd_cstr_append_char( p, '-' );
1306 33 : p = fd_cstr_append_ulong_as_text( p, 0, 0, range1-1UL, fd_ulong_base10_dig_cnt( range1-1UL ) );
1307 33 : p = fd_cstr_append_char( p, '/' );
1308 33 : p = fd_cstr_append_ulong_as_text( p, 0, 0, object_sz, fd_ulong_base10_dig_cnt( object_sz ) );
1309 33 : p = fd_cstr_append_cstr( p, "\r\n" );
1310 33 : }
1311 81 : p = fd_cstr_append_cstr( p, "Content-Length: " );
1312 81 : ulong content_len = range1 - range0;
1313 81 : p = fd_cstr_append_ulong_as_text( p, 0, 0, content_len, fd_ulong_base10_dig_cnt( content_len ) );
1314 81 : p = fd_cstr_append_cstr( p, "\r\n" );
1315 81 : if( sick ) {
1316 12 : p = fd_cstr_append_cstr( p, "Connection: close\r\n" );
1317 12 : }
1318 81 : p = fd_cstr_append_cstr( p, "\r\n" );
1319 81 : fd_cstr_fini( p );
1320 81 : return (ulong)( p - out );
1321 81 : }
1322 :
1323 : /* serve_snap_res_hdr dispatches a RES_WRITE_HDR job for a snapshot GET
1324 : or HEAD request. */
1325 :
1326 : static void
1327 : serve_snap_res_hdr( fd_snapsv_t * ctx,
1328 : fd_stem_context_t * stem,
1329 : uint conn_idx,
1330 54 : long now ) {
1331 54 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1332 54 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_WRITE_HDR, "state confusion" );
1333 :
1334 54 : snap_entry_t * snap = conn_snap_entry( ctx, conn );
1335 54 : if( FD_UNLIKELY( !snap ) ) {
1336 : /* rare edge case: snap was deleted by the snapmk tile just after
1337 : the user requested it. Don't bother returning an error, just
1338 : abort the conn. */
1339 0 : conn_close( ctx, stem, conn_idx, now );
1340 0 : return;
1341 0 : }
1342 54 : if( FD_UNLIKELY( !snap_lock( snap ) ) ) {
1343 0 : conn_close( ctx, stem, conn_idx, now );
1344 0 : return;
1345 0 : }
1346 : /* return response header */
1347 54 : FD_CHECK_ERR( conn->iobuf_idx==UINT_MAX, "conn has stale iobuf" );
1348 54 : iobuf_alloc( ctx, &conn->iobuf_idx );
1349 54 : uchar * iobuf = conn_iobuf( ctx, conn );
1350 54 : conn->res.sent = 0;
1351 54 : conn->res.len = (uint)build_snap_res_hdr(
1352 54 : (char *)iobuf,
1353 54 : conn->snap.range,
1354 54 : snap->is_zstd,
1355 54 : conn->snap.req_off0, conn->snap.req_off1,
1356 54 : snap->sz,
1357 54 : conn->sick );
1358 54 : conn->res.hdr_ts = now;
1359 54 : prep_write_hdr( ctx, conn_idx );
1360 54 : }
1361 :
1362 : /* newest_snap returns the newest snapshot of the requested type. */
1363 :
1364 : static snap_entry_t *
1365 : newest_snap( fd_snapsv_t * ctx,
1366 141 : int incremental ) {
1367 :
1368 141 : if( FD_UNLIKELY( !( incremental ? ctx->snap_cnt_incr : ctx->snap_cnt_full ) ) ) {
1369 30 : return NULL; /* no such snapshot of this type */
1370 30 : }
1371 111 : snap_entry_t * entry = incremental ? ctx->newest_incr : ctx->newest_full;
1372 111 : if( FD_UNLIKELY( !entry ) ) {
1373 87 : snap_entry_t * full = incremental ? newest_snap( ctx, 0 ) : NULL;
1374 87 : if( FD_UNLIKELY( incremental && !full ) ) return NULL;
1375 84 : ulong base_slot = full ? full->key.slot : ULONG_MAX;
1376 84 : snap_entry_t * entry0 = snap_map_ele0( ctx->snap_map );
1377 84 : ulong ele_max = snap_map_ele_max( ctx->snap_map );
1378 2100 : for( ulong i=0UL; i<ele_max; i++ ) {
1379 2016 : snap_entry_t * e = &entry0[ i ];
1380 2016 : if( snap_map_ele_is_free( e ) ) continue;
1381 258 : if( e->key.base_slot!=base_slot ) continue;
1382 126 : if( entry && e->key.slot<=entry->key.slot ) continue;
1383 114 : entry = e;
1384 114 : }
1385 84 : }
1386 108 : *( incremental ? &ctx->newest_incr : &ctx->newest_full ) = entry;
1387 108 : if( FD_UNLIKELY( !entry ) ) return NULL;
1388 102 : return entry;
1389 108 : }
1390 :
1391 : static ulong
1392 : build_redirect_hdr( char out[ static RES_HDR_MAX ], /* not a cstr */
1393 : int incremental,
1394 : ulong slot,
1395 : ulong base_slot,
1396 : uchar const hash[ static 32 ],
1397 : int is_zstd,
1398 45 : int sick ) {
1399 45 : char * p = fd_cstr_init( out );
1400 45 : p = fd_cstr_append_cstr( p, "HTTP/1.1 302 Found\r\n" );
1401 45 : p = fd_cstr_append_cstr( p, "Location: /" );
1402 45 : if( incremental ) {
1403 21 : p = fd_cstr_append_cstr( p, "incremental-snapshot-" );
1404 21 : p = fd_cstr_append_ulong_as_text( p, 0, 0, base_slot, fd_ulong_base10_dig_cnt( base_slot ) );
1405 21 : p = fd_cstr_append_char( p, '-' );
1406 24 : } else {
1407 24 : p = fd_cstr_append_cstr( p, "snapshot-" );
1408 24 : }
1409 45 : p = fd_cstr_append_ulong_as_text( p, 0, 0, slot, fd_ulong_base10_dig_cnt( slot ) );
1410 45 : p = fd_cstr_append_char( p, '-' );
1411 45 : char hash_b58[ FD_BASE58_ENCODED_32_SZ ];
1412 45 : fd_base58_encode_32( hash, NULL, hash_b58 );
1413 45 : p = fd_cstr_append_cstr( p, hash_b58 );
1414 45 : p = fd_cstr_append_cstr( p, is_zstd ? ".tar.zst\r\n" : ".tar\r\n" );
1415 45 : p = fd_cstr_append_cstr( p, "Content-Length: 0\r\n" );
1416 45 : if( sick ) {
1417 6 : p = fd_cstr_append_cstr( p, "Connection: close\r\n" );
1418 6 : }
1419 45 : p = fd_cstr_append_cstr( p, "\r\n" );
1420 45 : return (ulong)( p - out );
1421 45 : }
1422 :
1423 : /* serve_redirect dispatches a RES_WRITE_HDR job for a /snapshot.tar.bz2
1424 : redirect. */
1425 :
1426 : static void
1427 : serve_redirect( fd_snapsv_t * ctx,
1428 39 : uint conn_idx ) {
1429 39 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1430 39 : FD_CHECK_ERR( conn->state==CONN_STATE_RES_REDIRECT, "state confusion" );
1431 :
1432 39 : snap_entry_t * snap = newest_snap( ctx, conn->req.incremental );
1433 39 : if( FD_UNLIKELY( !snap ) ) {
1434 12 : conn->state = CONN_STATE_RES_WRITE_ERR;
1435 12 : conn->res.status = 404U;
1436 12 : serve_http_err( ctx, conn_idx );
1437 12 : return;
1438 12 : }
1439 :
1440 : /* return response */
1441 27 : FD_CHECK_ERR( conn->iobuf_idx==UINT_MAX, "conn has stale iobuf" );
1442 27 : iobuf_alloc( ctx, &conn->iobuf_idx );
1443 27 : uchar * iobuf = conn_iobuf( ctx, conn );
1444 27 : conn->res.sent = 0;
1445 27 : conn->res.len = (uint)build_redirect_hdr(
1446 27 : (char *)iobuf,
1447 27 : conn->req.incremental,
1448 27 : snap->key.slot, snap->key.base_slot,
1449 27 : snap->hash,
1450 27 : snap->is_zstd,
1451 27 : conn->sick );
1452 27 : prep_write_hdr( ctx, conn_idx );
1453 27 : }
1454 :
1455 : static void
1456 : prep_peek( fd_snapsv_t * ctx,
1457 : uint conn_idx );
1458 :
1459 : /* handle_write_hdr_comp handles the completion of a header write op. */
1460 :
1461 : static void
1462 : handle_write_hdr_comp( fd_snapsv_t * ctx,
1463 : fd_stem_context_t * stem,
1464 : uint conn_idx,
1465 : int res,
1466 78 : long now ) {
1467 78 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1468 78 : if( FD_UNLIKELY( res<0 ) ) {
1469 0 : if( res<0 ) {
1470 0 : FD_IP6_ADDR_CSTR( addr_cstr, &conn->peer_ip );
1471 0 : FD_LOG_INFO(( "snapshot download peer %s:%u: response header send error (%i-%s)",
1472 0 : addr_cstr, conn->peer_port, -res, fd_io_strerror( -res ) ));
1473 0 : }
1474 0 : conn_close( ctx, stem, conn_idx, now );
1475 0 : return;
1476 0 : }
1477 78 : conn->res.sent += (uint)res;
1478 78 : if( FD_LIKELY( conn->res.sent < conn->res.len ) ) {
1479 0 : prep_write_hdr( ctx, conn_idx );
1480 0 : return;
1481 0 : }
1482 : /* wrote response header */
1483 78 : iobuf_free( ctx, &conn->iobuf_idx );
1484 78 : switch( conn->state ) {
1485 24 : case CONN_STATE_RES_WRITE_ERR:
1486 39 : case CONN_STATE_RES_REDIRECT:
1487 : /* returned header-only response, handle the next request */
1488 39 : conn_req_done( ctx, stem, conn_idx, now );
1489 39 : return;
1490 39 : case CONN_STATE_RES_WRITE_HDR:
1491 39 : if( conn->req.head || conn->snap.key.slot==ULONG_MAX ) {
1492 6 : conn_req_done( ctx, stem, conn_idx, now );
1493 6 : return;
1494 6 : }
1495 : /* now serve the snapshot body */
1496 33 : conn->req.get_snap = 1; /* streaming begin */
1497 33 : event_snap( ctx, stem, conn_idx, now, 1, 0 ); /* som */
1498 33 : conn->state = CONN_STATE_RES_SHOVEL;
1499 33 : shovel( ctx, stem, conn_idx, now );
1500 33 : return;
1501 0 : default:
1502 0 : FD_LOG_CRIT(( "conn %u: state confusion", conn_idx ));
1503 78 : }
1504 78 : }
1505 :
1506 : /* match_snapshot_path parses a snapshot HTTP request path. */
1507 :
1508 : static snap_key_t *
1509 : match_snapshot_path( char const * path,
1510 : ulong path_len,
1511 : snap_key_t * out,
1512 : uchar out_hash[ 32 ],
1513 150 : int * out_is_zstd ) {
1514 150 : if( path_len<9 ) return NULL;
1515 :
1516 : /* ugly: copy to cstr */
1517 144 : char cstr[ FD_SNAP_NAME_MAX ];
1518 144 : if( FD_UNLIKELY( path_len>=FD_SNAP_NAME_MAX-1 ) ) return NULL;
1519 132 : fd_cstr_fini( fd_cstr_append_text( fd_cstr_init( cstr ), path, path_len ) );
1520 :
1521 132 : ulong full_slot, incremental_slot;
1522 132 : int res = fd_ssarchive_parse_filename( cstr, &full_slot, &incremental_slot, out_hash, out_is_zstd );
1523 132 : if( FD_UNLIKELY( res!=0 ) ) return NULL;
1524 99 : *out = (snap_key_t) {
1525 99 : .slot = incremental_slot!=ULONG_MAX ? incremental_slot : full_slot,
1526 99 : .base_slot = incremental_slot!=ULONG_MAX ? full_slot : ULONG_MAX
1527 99 : };
1528 99 : return out;
1529 132 : }
1530 :
1531 : /* parse_range_header parses a 'Range: bytes=start-end' header.
1532 : value points to the header value char[value_len].
1533 : object_sz is the size of the file, and [*out_range0, *out_range1)
1534 : is the requested range. */
1535 :
1536 : static int
1537 : parse_range_header( char const * value,
1538 : ulong value_len,
1539 : ulong object_sz,
1540 : ulong * out_range0,
1541 309 : ulong * out_range1 ) {
1542 309 : char const * p = value;
1543 309 : char const * end = value + value_len;
1544 :
1545 342 : while( p<end && (*p==' ' || *p=='\t') ) p++;
1546 330 : while( end>p && (end[-1]==' ' || end[-1]=='\t') ) end--;
1547 309 : if( FD_UNLIKELY( (ulong)(end-p)<7UL || strncasecmp( p, "bytes=", 6UL ) ) ) return -1;
1548 273 : p += 6;
1549 :
1550 273 : ulong first = 0UL;
1551 273 : int have_first = 0;
1552 1035 : while( p<end && *p>='0' && *p<='9' ) {
1553 774 : uint digit = (uint)(*p-'0');
1554 774 : if( FD_UNLIKELY( first>(ULONG_MAX-digit)/10UL ) ) return -1;
1555 762 : first = first*10UL + digit;
1556 762 : have_first = 1;
1557 762 : p++;
1558 762 : }
1559 261 : if( FD_UNLIKELY( p==end || *p!='-' ) ) return -1;
1560 234 : p++;
1561 :
1562 234 : ulong last = 0UL;
1563 234 : int have_last = 0;
1564 900 : while( p<end && *p>='0' && *p<='9' ) {
1565 678 : uint digit = (uint)(*p-'0');
1566 678 : if( FD_UNLIKELY( last>(ULONG_MAX-digit)/10UL ) ) return -1;
1567 666 : last = last*10UL + digit;
1568 666 : have_last = 1;
1569 666 : p++;
1570 666 : }
1571 222 : if( FD_UNLIKELY( p!=end || (!have_first && !have_last) ) ) return -1;
1572 198 : if( FD_UNLIKELY( !object_sz ) ) return -2;
1573 :
1574 183 : if( !have_first ) {
1575 39 : if( FD_UNLIKELY( !last ) ) return -2;
1576 33 : *out_range0 = last<object_sz ? object_sz-last : 0UL;
1577 33 : *out_range1 = object_sz;
1578 33 : return 0;
1579 39 : }
1580 :
1581 144 : if( FD_UNLIKELY( first>=object_sz ) ) return -2;
1582 117 : if( FD_UNLIKELY( have_last && last<first ) ) return -2;
1583 108 : *out_range0 = first;
1584 108 : *out_range1 = have_last && last<object_sz-1UL ? last+1UL : object_sz;
1585 108 : return 0;
1586 117 : }
1587 :
1588 : /* hdr_has_token returns 1 if the given comma separated header value
1589 : contains tok (case insensitive). */
1590 :
1591 : static int
1592 : hdr_has_token( char const * value,
1593 : ulong value_len,
1594 : char const * tok,
1595 24 : ulong tok_len ) {
1596 24 : char const * p = value;
1597 24 : char const * end = value + value_len;
1598 33 : while( p<end ) {
1599 33 : while( p<end && ( *p==' ' || *p=='\t' || *p==',' ) ) p++;
1600 27 : char const * t0 = p;
1601 237 : while( p<end && *p!=',' ) p++;
1602 27 : char const * t1 = p;
1603 27 : while( t1>t0 && ( t1[-1]==' ' || t1[-1]=='\t' ) ) t1--;
1604 27 : if( (ulong)( t1-t0 )==tok_len && !strncasecmp( t0, tok, tok_len ) ) return 1;
1605 27 : }
1606 6 : return 0;
1607 24 : }
1608 :
1609 : /* handle_peek handles newly received TCP data. peek_len bytes of
1610 : unconsumed request head sit at conn->req_len in the iobuf. */
1611 :
1612 : static void
1613 : handle_peek( fd_snapsv_t * ctx,
1614 : fd_stem_context_t * stem,
1615 : uint conn_idx,
1616 : int res,
1617 222 : long now ) {
1618 222 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1619 222 : if( FD_UNLIKELY( res<=0 ) ) {
1620 39 : if( FD_UNLIKELY( res<0 ) ) {
1621 30 : FD_IP6_ADDR_CSTR( addr_cstr, &conn->peer_ip );
1622 30 : FD_LOG_INFO(( "snapshot download peer %s:%u: receive error (%i-%s)",
1623 30 : addr_cstr, conn->peer_port, -res, fd_io_strerror( -res ) ));
1624 30 : }
1625 39 : conn_close( ctx, stem, conn_idx, now );
1626 39 : return;
1627 39 : }
1628 183 : uint peek_len = (uint)res;
1629 :
1630 183 : char const * method; ulong method_len;
1631 183 : char const * path; ulong path_len;
1632 183 : int minor_version;
1633 183 : struct phr_header headers[ REQ_HEADER_MAX ];
1634 183 : ulong header_cnt = REQ_HEADER_MAX;
1635 183 : int parsed = phr_parse_request(
1636 183 : (char const *)conn_iobuf( ctx, conn ),
1637 183 : conn->req.len + peek_len,
1638 183 : &method, &method_len,
1639 183 : &path, &path_len,
1640 183 : &minor_version,
1641 183 : headers, &header_cnt,
1642 183 : conn->req.len /* slowloris defense */
1643 183 : );
1644 :
1645 183 : if( FD_UNLIKELY( parsed==-2 ) ) { /* request head incomplete */
1646 18 : if( FD_UNLIKELY( conn->req.len+peek_len >= ctx->iobuf_sz ) ) {
1647 0 : FD_LOG_DEBUG(( "conn %u: request head exceeds %u bytes", conn_idx, ctx->iobuf_sz ));
1648 0 : conn_close( ctx, stem, conn_idx, now );
1649 0 : return;
1650 0 : }
1651 : /* avoid copying the same data again */
1652 18 : FD_MCNT_INC( SNAPSV, HTTP_RX_BYTES, peek_len );
1653 18 : prep_consume( ctx, conn_idx, peek_len, UDATA_OP_CONSUME_FRAG );
1654 18 : conn->state = CONN_STATE_REQ_SKIP;
1655 18 : conn->req.skip_len = peek_len;
1656 18 : return;
1657 18 : }
1658 :
1659 165 : if( FD_UNLIKELY( parsed<0 ) ) { /* malformed request head */
1660 0 : FD_LOG_DEBUG(( "conn %u: malformed request", conn_idx ));
1661 0 : FD_MCNT_INC( SNAPSV, HTTP_RX_BYTES, peek_len );
1662 0 : conn_close( ctx, stem, conn_idx, now );
1663 0 : return;
1664 0 : }
1665 :
1666 : /* valid request at this point */
1667 :
1668 165 : FD_CHECK_ERR( (uint)parsed > conn->req.len, "request head shrank" );
1669 165 : FD_MCNT_INC( SNAPSV, HTTP_RX_BYTES, (uint)parsed - conn->req.len );
1670 165 : prep_consume( ctx, conn_idx, (uint)parsed - conn->req.len, UDATA_OP_CONSUME_TAIL );
1671 165 : conn->state = CONN_STATE_REQ_DONE;
1672 165 : conn->req.len = (uint)parsed - conn->req.len;
1673 165 : conn->req.req_seq = ctx->req_seq++;
1674 :
1675 : /* must not do any work until CONSUME_TAIL completes,
1676 : so remember what to do for when that happens */
1677 165 : snap_key_t query;
1678 165 : uchar query_hash[ 32 ];
1679 165 : int query_is_zstd;
1680 165 : if( method_len==4UL && !memcmp( method, "HEAD", method_len ) ) {
1681 12 : conn->req.head = 1;
1682 153 : } else if( method_len==3UL && !memcmp( method, "GET", method_len ) ) {
1683 153 : conn->req.head = 0;
1684 153 : } else {
1685 0 : conn->state = CONN_STATE_RES_WRITE_ERR;
1686 0 : conn->res.status = 500;
1687 0 : conn->sick = 1;
1688 0 : return;
1689 0 : }
1690 :
1691 : /* parse request headers */
1692 :
1693 165 : #define HDR_RANGE (1U<<0)
1694 165 : #define HDR_CLEN (1U<<1)
1695 165 : #define HDR_CONN (1U<<2)
1696 :
1697 165 : char const * range = NULL;
1698 165 : ulong range_len = 0UL;
1699 165 : int keepalive = minor_version>=1; /* HTTP/1.0 defaults to close */
1700 165 : int bad_req = 0; /* reject request with a 400? */
1701 165 : uint seen = 0U; /* set of HDR_* seen, to detect duplicates */
1702 246 : for( ulong i=0UL; i<header_cnt; i++ ) {
1703 81 : char const * name = headers[ i ].name;
1704 81 : ulong name_len = headers[ i ].name_len;
1705 81 : char const * value = headers[ i ].value;
1706 81 : ulong value_len = headers[ i ].value_len;
1707 81 : if( FD_UNLIKELY( !name_len ) ) { /* obs-fold continuation line */
1708 3 : bad_req = 1;
1709 3 : continue;
1710 3 : }
1711 78 : if( name_len==5UL && !strncasecmp( name, "range", 5UL ) ) {
1712 39 : bad_req |= !!( seen & HDR_RANGE );
1713 39 : seen |= HDR_RANGE;
1714 39 : range = value;
1715 39 : range_len = value_len;
1716 39 : } else if( name_len==14UL && !strncasecmp( name, "content-length", 14UL ) ) {
1717 15 : bad_req |= !!( seen & HDR_CLEN );
1718 15 : seen |= HDR_CLEN;
1719 15 : bad_req |= !( value_len==1UL && value[ 0 ]=='0' ); /* request bodies are not supported */
1720 24 : } else if( name_len==17UL && !strncasecmp( name, "transfer-encoding", 17UL ) ) {
1721 3 : bad_req = 1; /* chunked bodies are not supported */
1722 21 : } else if( name_len==10UL && !strncasecmp( name, "connection", 10UL ) ) {
1723 18 : bad_req |= !!( seen & HDR_CONN );
1724 18 : seen |= HDR_CONN;
1725 18 : if( hdr_has_token( value, value_len, "close", 5UL ) ) keepalive = 0;
1726 6 : else if( hdr_has_token( value, value_len, "keep-alive", 10UL ) ) keepalive = 1;
1727 18 : }
1728 78 : }
1729 :
1730 165 : #undef HDR_RANGE
1731 165 : #undef HDR_CLEN
1732 165 : #undef HDR_CONN
1733 :
1734 165 : if( FD_UNLIKELY( !keepalive ) ) conn->sick = 1;
1735 :
1736 165 : if( FD_UNLIKELY( bad_req ) ) {
1737 21 : conn->state = CONN_STATE_RES_WRITE_ERR;
1738 21 : conn->res.status = 400;
1739 21 : conn->sick = 1;
1740 21 : return;
1741 21 : }
1742 :
1743 144 : conn->snap.range = 0U;
1744 144 : conn->snap.req_off0 = 0UL;
1745 144 : conn->snap.req_off1 = 0UL;
1746 :
1747 : /* strip leading slashes */
1748 144 : if( FD_UNLIKELY( path_len<10 ) ) goto not_found;
1749 114 : if( FD_UNLIKELY( path[ 0 ]!='/' ) ) goto not_found;
1750 114 : path++; path_len--;
1751 : /* skip superfluous leading slashes to help buggy clients */
1752 114 : for( ulong i=0UL; path_len && i<15; i++ ) {
1753 114 : if( FD_LIKELY( path[ 0 ]!='/' ) ) break;
1754 0 : path++; path_len--;
1755 0 : }
1756 :
1757 114 : if( path_len==16 ) {
1758 21 : if( !memcmp( path, "snapshot.tar.zst", 16UL ) ||
1759 21 : !memcmp( path, "snapshot.tar.bz2", 16UL ) ) {
1760 21 : conn->state = CONN_STATE_RES_REDIRECT;
1761 21 : conn->req.incremental = 0;
1762 21 : return;
1763 21 : }
1764 93 : } else if( path_len==28 ) {
1765 18 : if( !memcmp( path, "incremental-snapshot.tar.bz2", 28UL ) ||
1766 18 : !memcmp( path, "incremental-snapshot.tar.zst", 28UL ) ) {
1767 18 : conn->state = CONN_STATE_RES_REDIRECT;
1768 18 : conn->req.incremental = 1;
1769 18 : return;
1770 18 : }
1771 18 : }
1772 :
1773 : /* found a snapshot? */
1774 75 : if( match_snapshot_path( path, path_len, &query, query_hash, &query_is_zstd ) ) {
1775 72 : snap_entry_t * entry = snap_map_update( ctx->snap_map, &query );
1776 72 : if( FD_UNLIKELY( !entry ) ) goto not_found;
1777 69 : if( FD_UNLIKELY( 0!=memcmp( entry->hash, query_hash, 32UL ) ) ) goto not_found;
1778 66 : if( FD_UNLIKELY( entry->is_zstd!=query_is_zstd ) ) goto not_found;
1779 63 : conn->state = CONN_STATE_RES_WRITE_HDR;
1780 63 : conn->snap.key = query;
1781 63 : conn->snap.file_sz = entry->sz;
1782 63 : conn->snap.range = 0U;
1783 63 : conn->snap.req_off0 = 0UL;
1784 63 : conn->snap.req_off1 = entry->sz;
1785 63 : conn->snap.req_cur = 0UL;
1786 63 : conn->req.incremental = entry->key.base_slot!=ULONG_MAX;
1787 :
1788 63 : if( range && !conn->req.head ) {
1789 27 : int range_err = parse_range_header( range, range_len, entry->sz, &conn->snap.req_off0, &conn->snap.req_off1 );
1790 27 : if( FD_UNLIKELY( range_err ) ) {
1791 9 : conn->state = CONN_STATE_RES_WRITE_ERR;
1792 9 : conn->res.status = (uint)( range_err==-2 ? 416 : 400 );
1793 18 : } else {
1794 18 : conn->snap.req_cur = conn->snap.req_off0;
1795 18 : conn->snap.range = 1U;
1796 18 : }
1797 27 : }
1798 63 : return;
1799 66 : }
1800 :
1801 42 : not_found:
1802 42 : conn->state = CONN_STATE_RES_WRITE_ERR;
1803 42 : conn->res.status = 404;
1804 :
1805 : /* next step happens at handle_consume_tail_comp */
1806 42 : }
1807 :
1808 : /* handle_consume_frag_comp handles the completion of a CONSUME_FRAG op. */
1809 :
1810 : static void
1811 : handle_consume_frag_comp( fd_snapsv_t * ctx,
1812 : fd_stem_context_t * stem,
1813 : uint conn_idx,
1814 : int res,
1815 18 : long now ) {
1816 18 : if( FD_UNLIKELY( res<=0 ) ) {
1817 0 : if( FD_UNLIKELY( res<0 ) ) {
1818 0 : FD_LOG_DEBUG(( "conn %u: recv() failed (%i-%s)", conn_idx, -res, fd_io_strerror( -res ) ));
1819 0 : }
1820 0 : conn_close( ctx, stem, conn_idx, now );
1821 0 : return;
1822 0 : }
1823 18 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1824 18 : FD_CHECK_ERR( conn->state==CONN_STATE_REQ_SKIP, "conn state confusion" );
1825 18 : if( FD_UNLIKELY( (uint)res!=conn->req.skip_len ) ) {
1826 15 : FD_LOG_DEBUG(( "conn %u: consumed %i bytes, requested %u", conn_idx, res, conn->req.skip_len ));
1827 15 : conn_close( ctx, stem, conn_idx, now );
1828 15 : return;
1829 15 : }
1830 3 : conn->req.len += conn->req.skip_len;
1831 3 : conn->state = CONN_STATE_REQ_PEEK;
1832 3 : prep_peek( ctx, conn_idx );
1833 3 : }
1834 :
1835 : /* handle_consume_tail_comp handles the completion of a CONSUME_TAIL op. */
1836 :
1837 : static void
1838 : handle_consume_tail_comp( fd_snapsv_t * ctx,
1839 : fd_stem_context_t * stem,
1840 : uint conn_idx,
1841 : int res,
1842 165 : long now ) {
1843 165 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1844 165 : if( FD_UNLIKELY( res!=(int)conn->req.len ) ) {
1845 0 : conn_close( ctx, stem, conn_idx, now );
1846 0 : return;
1847 0 : }
1848 165 : conn->req.len = 0;
1849 165 : iobuf_free( ctx, &conn->iobuf_idx );
1850 165 : switch( ctx->conn0[ conn_idx ].state ) {
1851 72 : case CONN_STATE_RES_WRITE_ERR:
1852 72 : serve_http_err( ctx, conn_idx );
1853 72 : break;
1854 54 : case CONN_STATE_RES_WRITE_HDR:
1855 54 : serve_snap_res_hdr( ctx, stem, conn_idx, now );
1856 54 : break;
1857 39 : case CONN_STATE_RES_REDIRECT:
1858 39 : serve_redirect( ctx, conn_idx );
1859 39 : break;
1860 0 : default:
1861 0 : FD_LOG_CRIT(( "conn %u: state confusion", conn_idx ));
1862 165 : }
1863 165 : FD_MCNT_INC( SNAPSV, HTTP_REQUEST_SERVED, 1UL );
1864 165 : }
1865 :
1866 : /* futex_prep asks the kernel to send a CQE for when a new
1867 : snapmk_out frag becomes available. */
1868 :
1869 : static int
1870 : futex_prep( fd_snapsv_t * ctx,
1871 14664 : ulong in_idx ) {
1872 14664 : ulong seq_prod = fd_mcache_seq_query( ctx->in[ in_idx ].seq_prod );
1873 14664 : ulong seq_next = fd_seq_inc( ctx->in[ in_idx ].seq_cons, 1UL );
1874 14664 : fd_frag_meta_t const * mline = ctx->in[ in_idx ].mcache + fd_mcache_line_idx( seq_next, ctx->in[ in_idx ].depth );
1875 14664 : if( FD_UNLIKELY( fd_frag_meta_seq_query( mline )==seq_next ) ) return 0;
1876 :
1877 14664 : if( FD_LIKELY( ctx->in[ in_idx ].futex_armed ) ) return 1;
1878 14664 : fd_io_uring_t * ring = ctx->ring;
1879 14664 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ring->sq );
1880 14664 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1881 14664 : *sqe = (fd_io_uring_sqe_t) {
1882 14664 : .opcode = FD_IORING_OP_FUTEX_WAIT,
1883 14664 : .flags = FD_IOSQE_FIXED_FILE, /* ignored, required for sandbox (hope no kernel ABI breakage) */
1884 14664 : .fd = FUTEX2_SIZE_U32,
1885 14664 : .addr = (ulong)ctx->in[ in_idx ].seq_prod,
1886 14664 : .off = (ulong)(uint)seq_prod, /* value to compare against */
1887 14664 : .addr3 = FUTEX_BITSET_MATCH_ANY
1888 14664 : };
1889 14664 : snapsv_udata_t udata = { .op = UDATA_OP_FUTEX, .conn_idx = (uint)in_idx };
1890 14664 : sqe->user_data = udata.user_data;
1891 14664 : ctx->in[ in_idx ].futex_armed = 1;
1892 14664 : return 1;
1893 14664 : }
1894 :
1895 : /* futex_comp is called when a snapmk_out frag was detected. Such
1896 : frags are also polled by the event loop, but this is required to wake
1897 : up fast from an io_uring_enter sleep. */
1898 :
1899 : static void
1900 : futex_comp( fd_snapsv_t * ctx,
1901 : uint in_idx,
1902 14577 : int res ) {
1903 14577 : FD_CHECK_CRIT( in_idx<IN_LINK_MAX, "io_uring completion has invalid in index" );
1904 14577 : ctx->in[ in_idx ].futex_armed = 0;
1905 14577 : if( FD_UNLIKELY( res<0 && res!=-EAGAIN && res!=-ECANCELED && res!=-EINTR ) ) {
1906 0 : FD_LOG_ERR(( "IORING_OP_FUTEX_WAIT failed (%i-%s)", -res, fd_io_strerror( -res ) ));
1907 0 : }
1908 14577 : }
1909 :
1910 : /* handle_cqe handles a single io_uring completion. */
1911 :
1912 : static void
1913 : handle_cqe( fd_snapsv_t * ctx,
1914 : fd_stem_context_t * stem,
1915 : fd_io_uring_cqe_t const * cqe,
1916 16251 : long now ) {
1917 16251 : snapsv_udata_t udata = { .user_data = cqe->user_data };
1918 16251 : uint conn_idx = udata.conn_idx;
1919 16251 : switch( udata.op ) {
1920 162 : case UDATA_OP_ACCEPT:
1921 162 : handle_accept( ctx, conn_idx, cqe->res, now );
1922 162 : break;
1923 222 : case UDATA_OP_PEEK:
1924 222 : handle_peek( ctx, stem, conn_idx, cqe->res, now );
1925 222 : break;
1926 18 : case UDATA_OP_CONSUME_FRAG:
1927 18 : handle_consume_frag_comp( ctx, stem, conn_idx, cqe->res, now );
1928 18 : break;
1929 165 : case UDATA_OP_CONSUME_TAIL:
1930 165 : handle_consume_tail_comp( ctx, stem, conn_idx, cqe->res, now );
1931 165 : break;
1932 78 : case UDATA_OP_WRITE_HDR:
1933 78 : handle_write_hdr_comp( ctx, stem, conn_idx, cqe->res, now );
1934 78 : break;
1935 246 : case UDATA_OP_SHOVEL_DISK:
1936 246 : shovel_comp_disk( ctx, stem, conn_idx, cqe->res, now );
1937 246 : break;
1938 237 : case UDATA_OP_SHOVEL_NET:
1939 237 : shovel_comp_net( ctx, stem, conn_idx, cqe->res, now );
1940 237 : break;
1941 14577 : case UDATA_OP_FUTEX:
1942 14577 : futex_comp( ctx, conn_idx, cqe->res );
1943 14577 : break;
1944 537 : case UDATA_OP_TIMEOUT:
1945 546 : case UDATA_OP_SHUTDOWN:
1946 546 : break;
1947 0 : default:
1948 0 : FD_LOG_CRIT(( "io_uring completion has invalid user_data 0x%016lx", (ulong)cqe->user_data ));
1949 16251 : }
1950 16251 : }
1951 :
1952 : /* prep_shutdown enqueues a shutdown of a conn's socket, which fails
1953 : any pending request receive op. */
1954 :
1955 : static void
1956 : prep_shutdown( fd_snapsv_t * ctx,
1957 9 : uint conn_idx ) {
1958 9 : fd_io_uring_t * ring = ctx->ring;
1959 9 : fd_io_uring_sqe_t * sqe = fd_io_uring_get_sqe( ring->sq );
1960 9 : FD_CHECK_ERR( sqe, "io_uring submission queue full" );
1961 9 : *sqe = (fd_io_uring_sqe_t) {
1962 9 : .opcode = FD_IORING_OP_SHUTDOWN,
1963 9 : .flags = FD_IOSQE_FIXED_FILE,
1964 9 : .fd = (int)( ctx->conn0_fd_idx + conn_idx ),
1965 9 : .len = SHUT_RDWR
1966 9 : };
1967 9 : snapsv_udata_t udata = { .op = UDATA_OP_SHUTDOWN, .conn_idx = conn_idx };
1968 9 : sqe->user_data = udata.user_data;
1969 9 : ctx->conn0[ conn_idx ].shutdown = 1;
1970 9 : }
1971 :
1972 : /* idle_sweep shuts down conns that are waiting for a request past the
1973 : idle deadline. The linked timeout on the request peek is not enough
1974 : on its own: the kernel can lose a timeout firing while it retries the
1975 : receive op. */
1976 :
1977 : static void
1978 : idle_sweep( fd_snapsv_t * ctx,
1979 : uint conn_idx,
1980 468288 : long now ) {
1981 468288 : snapsv_conn_t * conn = &ctx->conn0[ conn_idx ];
1982 468288 : if( conn->shutdown ) return;
1983 468288 : if( conn->state!=CONN_STATE_REQ_PEEK &&
1984 468288 : conn->state!=CONN_STATE_REQ_SKIP &&
1985 468288 : conn->state!=CONN_STATE_REQ_DONE ) return;
1986 396 : if( now - conn->active_ts < ctx->idle_timeout_nanos ) return;
1987 9 : prep_shutdown( ctx, conn_idx );
1988 9 : }
1989 :
1990 : /* stem_in_update grants flow control credits to an upstream producer. */
1991 :
1992 : static inline void
1993 : stem_in_update( fd_stem_tile_in_t * in );
1994 :
1995 : /* after_credit performs one io_uring event loop iteration. */
1996 :
1997 : static int
1998 : after_credit_pre( fd_snapsv_t * ctx,
1999 : fd_stem_context_t * stem,
2000 : int * charge_busy,
2001 14664 : long now ) {
2002 14664 : fd_io_uring_t * ring = ctx->ring;
2003 :
2004 : /* Reap background completions */
2005 14664 : uint ready = fd_io_uring_cq_ready( ring->cq );
2006 14691 : for( uint i=0U; i<ready; i++ ) {
2007 27 : handle_cqe( ctx, stem, fd_io_uring_cq_head( ring->cq ), now );
2008 27 : fd_io_uring_cq_advance( ring->cq, 1U );
2009 27 : *charge_busy = 1;
2010 27 : }
2011 :
2012 : /* Periodically send a burst update over all GUI tiles.
2013 : (This is a deep burst, so the link depth should be about 2x larger
2014 : than the conn table size.) */
2015 14664 : if( FD_UNLIKELY( now >= ctx->out.next_heartbeat ) ) {
2016 14634 : ulong conn_max = ctx->conn_max;
2017 482922 : for( uint i=0U; i<conn_max; i++ ) {
2018 468288 : snapsv_conn_t * conn = &ctx->conn0[ i ];
2019 468288 : if( conn->req.get_snap ) {
2020 270 : event_snap( ctx, stem, i, now, 0, 0 );
2021 270 : }
2022 468288 : idle_sweep( ctx, i, now );
2023 468288 : }
2024 14634 : ctx->out.next_heartbeat += FD_SNAPSV_SYNC_PERIOD;
2025 14634 : if( FD_UNLIKELY( ctx->out.next_heartbeat < now ) ) {
2026 : /* more than one period missed, skip ahead timer */
2027 0 : ctx->out.next_heartbeat = now + FD_SNAPSV_SYNC_PERIOD;
2028 0 : }
2029 14634 : }
2030 :
2031 : /* Prepare for sleep */
2032 14664 : int waiting = !!futex_prep( ctx, 0UL );
2033 14664 : stem_in_update( &stem->in[ 0 ] );
2034 14664 : return waiting;
2035 14664 : }
2036 :
2037 : static void
2038 : after_credit_post( fd_snapsv_t * ctx,
2039 : fd_stem_context_t * stem,
2040 : int * charge_busy,
2041 14568 : long now ) {
2042 14568 : fd_io_uring_t * ring = ctx->ring;
2043 :
2044 14568 : if( FD_UNLIKELY( fd_io_uring_sq_dropped( ring->sq ) ) ) {
2045 0 : FD_LOG_ERR(( "io_uring submission queue dropped entries" ));
2046 0 : }
2047 14568 : if( FD_UNLIKELY( fd_io_uring_cq_overflow( ring->cq ) ) ) {
2048 0 : FD_LOG_ERR(( "io_uring completion queue overflowed" ));
2049 0 : }
2050 :
2051 : /* Reap deferred/immediate completions */
2052 14568 : uint ready = fd_io_uring_cq_ready( ring->cq );
2053 30792 : for( uint i=0U; i<ready; i++ ) {
2054 16224 : handle_cqe( ctx, stem, fd_io_uring_cq_head( ring->cq ), now );
2055 16224 : fd_io_uring_cq_advance( ring->cq, 1U );
2056 16224 : *charge_busy = 1;
2057 16224 : }
2058 14568 : }
2059 :
2060 : static void
2061 : after_credit( fd_snapsv_t * ctx,
2062 : fd_stem_context_t * stem,
2063 : int * opt_poll_in,
2064 0 : int * charge_busy ) {
2065 0 : (void)opt_poll_in;
2066 :
2067 0 : long now = fd_clock_tile_now( ctx->clock );
2068 0 : int waiting = after_credit_pre( ctx, stem, charge_busy, now );
2069 :
2070 : /* Dispatch io_uring work, poll completions, and do a bounded sleep */
2071 0 : fd_io_uring_t * ring = ctx->ring;
2072 0 : uint tail = ring->sq->sqe_tail;
2073 0 : atomic_store_explicit( ring->sq->ktail, tail, memory_order_release );
2074 0 : uint head = atomic_load_explicit( ring->sq->khead, memory_order_relaxed );
2075 0 : ring->sq->sqe_head = head;
2076 0 : uint to_submit = tail - head;
2077 0 : FD_STATIC_ASSERT( FD_SNAPSV_WAKE_PERIOD<(long)1e9, bounds );
2078 0 : fd_kernel_timespec_t timeout = { .tv_nsec = FD_SNAPSV_WAKE_PERIOD };
2079 0 : fd_io_uring_getevents_arg_t enter_arg = { .ts = (ulong)&timeout };
2080 0 : int submitted = fd_io_uring_enter(
2081 0 : ring->ioring_fd, to_submit, !!waiting,
2082 0 : FD_IORING_ENTER_GETEVENTS | FD_IORING_ENTER_EXT_ARG,
2083 0 : &enter_arg, sizeof(enter_arg)
2084 0 : );
2085 0 : now = fd_clock_tile_now( ctx->clock );
2086 0 : if( FD_UNLIKELY( submitted<0 ) ) {
2087 0 : if( FD_LIKELY( errno==ETIME ) ) {
2088 0 : submitted = 0; /* timeout, do housekeeping */
2089 0 : } else if( FD_LIKELY( errno==EINTR ) ) {
2090 0 : *charge_busy = !!submitted;
2091 0 : return;
2092 0 : } else {
2093 0 : FD_LOG_ERR(( "io_uring_enter failed (%i-%s)", errno, fd_io_strerror( errno ) ));
2094 0 : }
2095 0 : }
2096 :
2097 0 : after_credit_post( ctx, stem, charge_busy, now );
2098 0 : if( waiting ) *charge_busy = 0; /* do not count sleep time as busy */
2099 0 : }
2100 :
2101 : /* snap_open registers metadata from snapmk for the given pre-opened
2102 : snapshot pool file. */
2103 :
2104 : static void
2105 : snap_open( fd_snapsv_t * ctx,
2106 : ulong slot,
2107 : ulong base_slot,
2108 : ulong pool_idx,
2109 : ulong sz,
2110 147 : char const * name ) {
2111 147 : FD_CHECK_ERR( ctx->snap_cnt_full + ctx->snap_cnt_incr < ctx->snap_max, "too many snapshot files (snapmk_out desync?)" );
2112 147 : FD_CHECK_CRIT( pool_idx < ctx->snap_max, "invalid snapshot pool index" );
2113 :
2114 147 : ulong parsed_full_slot;
2115 147 : ulong parsed_incremental_slot;
2116 147 : uchar hash[ 32 ];
2117 147 : int is_zstd;
2118 147 : FD_CHECK_ERR( !fd_ssarchive_parse_filename( name, &parsed_full_slot, &parsed_incremental_slot, hash, &is_zstd ),
2119 147 : "invalid snapshot filename" );
2120 :
2121 147 : snap_entry_t * entry = snap_map_insert( ctx->snap_map, &(snap_key_t){slot, base_slot} );
2122 147 : FD_CHECK_ERR( !!entry, "snap_map_insert failed (snapmk_out desync?)" );
2123 :
2124 147 : memcpy( entry->hash, hash, sizeof(entry->hash) );
2125 147 : entry->sz = sz;
2126 147 : entry->fd = FD_SNAP_RO_FD( pool_idx );
2127 147 : entry->locked = 0;
2128 147 : entry->is_zstd = !!is_zstd;
2129 147 : ctx->newest_incr = NULL;
2130 147 : if( base_slot==ULONG_MAX ) {
2131 120 : ctx->snap_cnt_full++;
2132 120 : ctx->newest_full = NULL;
2133 120 : } else {
2134 27 : ctx->snap_cnt_incr++;
2135 27 : }
2136 147 : }
2137 :
2138 : /* snap_lock tries to acquire a read-lock to a snapshot file handle.
2139 : Returns 1 on success, 0 on failure (currently write locked).
2140 : Snapshots deliberately remain locked until the snapshot is about to
2141 : be deleted to prevent F_SETLK churn. */
2142 :
2143 : FD_FN_UNUSED static int
2144 54 : snap_lock( snap_entry_t * entry ) {
2145 54 : if( entry->locked ) return 1; /* already locked */
2146 : #ifndef FD_TILE_TEST
2147 0 : struct flock lock = {
2148 0 : .l_type = F_RDLCK,
2149 0 : .l_whence = SEEK_SET
2150 0 : };
2151 0 : if( FD_UNLIKELY( fcntl( entry->fd, F_SETLK, &lock ) ) ) {
2152 0 : if( errno==EAGAIN || errno==EACCES ) return 0; /* write locked */
2153 0 : FD_LOG_ERR(( "fcntl(F_RDLCK, %lu, %lu) failed (%i-%s)", entry->key.slot, entry->key.base_slot, errno, fd_io_strerror( errno ) ));
2154 0 : }
2155 0 : #endif
2156 54 : entry->locked = 1;
2157 54 : return 1;
2158 0 : }
2159 :
2160 : /* snap_close releases a snapshot file handle. */
2161 :
2162 : static void
2163 : snap_close( fd_snapsv_t * ctx,
2164 : ulong slot,
2165 60 : ulong base_slot ) {
2166 60 : snap_entry_t * entry = snap_map_update( ctx->snap_map, &(snap_key_t){slot, base_slot} );
2167 60 : if( FD_UNLIKELY( !entry ) ) return; /* ignore */
2168 :
2169 51 : if( entry->locked ) {
2170 : #ifndef FD_TILE_TEST
2171 : struct flock lock = {
2172 : .l_type = F_UNLCK,
2173 : .l_whence = SEEK_SET
2174 : };
2175 0 : if( FD_UNLIKELY( fcntl( entry->fd, F_SETLK, &lock ) ) ) {
2176 0 : FD_LOG_ERR(( "fcntl(F_UNLCK, %lu, %lu) failed (%i-%s)", slot, base_slot, errno, fd_io_strerror( errno ) ));
2177 0 : }
2178 0 : #endif
2179 0 : entry->locked = 0;
2180 0 : }
2181 0 : entry->fd = -1;
2182 :
2183 0 : snap_map_remove( ctx->snap_map, entry );
2184 0 : ctx->newest_full = ctx->newest_incr = NULL;
2185 51 : if( base_slot==ULONG_MAX ) {
2186 36 : FD_CHECK_CRIT( ctx->snap_cnt_full, "full snapshot count underflow" );
2187 36 : ctx->snap_cnt_full--;
2188 36 : } else {
2189 15 : FD_CHECK_CRIT( ctx->snap_cnt_incr, "incremental snapshot count underflow" );
2190 15 : ctx->snap_cnt_incr--;
2191 15 : }
2192 0 : }
2193 :
2194 : /* msg_snapmk is called for every snapmk_out frag. */
2195 :
2196 : static void
2197 : msg_snapmk( fd_snapsv_t * ctx,
2198 : ulong msg_type, /* sig */
2199 : fd_snapmk_msg_t const * msg,
2200 126 : ulong msg_sz ) {
2201 126 : switch( msg_type ) {
2202 108 : case FD_SNAPMK_MSG_FOUND: {
2203 108 : FD_CHECK_CRIT( msg_sz==sizeof(fd_snapmk_msg_found_t), "ABI mismatch" );
2204 108 : snap_open( ctx, msg->found.slot, msg->found.base_slot, msg->found.pool_idx, msg->found.sz, msg->found.name );
2205 108 : break;
2206 108 : }
2207 0 : case FD_SNAPMK_MSG_CREATED: {
2208 0 : FD_CHECK_CRIT( msg_sz==sizeof(fd_snapmk_msg_created_t), "ABI mismatch" );
2209 0 : snap_open( ctx, msg->created.slot, msg->created.base_slot, msg->created.pool_idx, msg->created.sz, msg->created.name );
2210 0 : break;
2211 0 : }
2212 18 : case FD_SNAPMK_MSG_DELETED: {
2213 18 : FD_CHECK_CRIT( msg_sz==sizeof(fd_snapmk_msg_deleted_t), "ABI mismatch" );
2214 18 : snap_close( ctx, msg->deleted.slot, msg->deleted.base_slot );
2215 18 : break;
2216 18 : }
2217 0 : default:
2218 0 : break;
2219 126 : }
2220 126 : }
2221 :
2222 : /* returnable_frag is called for every input frag */
2223 :
2224 : static int
2225 : returnable_frag( fd_snapsv_t * ctx,
2226 : ulong in_idx,
2227 : ulong seq,
2228 : ulong sig,
2229 : ulong chunk,
2230 : ulong sz,
2231 : ulong ctl,
2232 : ulong tsorig,
2233 : ulong tspub,
2234 126 : fd_stem_context_t * stem ) {
2235 126 : (void)ctl; (void)tsorig; (void)tspub; (void)stem;
2236 126 : ctx->in[ in_idx ].seq_cons = seq;
2237 126 : switch( ctx->in_kind[ in_idx ] ) {
2238 126 : case IN_KIND_SNAPMK:
2239 126 : FD_CHECK_CRIT( chunk >= ctx->in[ in_idx ].chunk0 &&
2240 126 : chunk <= ctx->in[ in_idx ].wmark &&
2241 126 : sz <= ctx->in[ in_idx ].mtu,
2242 126 : "input frag is out-of-bounds" );
2243 126 : fd_snapmk_msg_t const * msg = fd_chunk_to_laddr_const( ctx->in[ in_idx ].mem, chunk );
2244 126 : msg_snapmk( ctx, sig, msg, sz );
2245 126 : return 0; /* ok */
2246 0 : default:
2247 0 : FD_LOG_CRIT(( "unhandled frag from in_idx=%lu", in_idx ));
2248 126 : }
2249 126 : }
2250 :
2251 : static void
2252 0 : during_housekeeping( fd_snapsv_t * ctx ) {
2253 0 : if( FD_UNLIKELY( fd_clock_tile_recal_due( ctx->clock ) ) ) {
2254 0 : fd_clock_tile_recal( ctx->clock );
2255 0 : }
2256 0 : }
2257 :
2258 : static void
2259 0 : metrics_write( fd_snapsv_t * ctx ) {
2260 0 : FD_MGAUGE_SET( SNAPSV, SNAPSHOT_AVAILABLE_FULL, ctx->snap_cnt_full );
2261 0 : FD_MGAUGE_SET( SNAPSV, SNAPSHOT_AVAILABLE_INCREMENTAL, ctx->snap_cnt_incr );
2262 0 : FD_MGAUGE_SET( SNAPSV, CONN_OPEN, ctx->conn_cnt );
2263 0 : }
2264 :
2265 : /* snapsv uses a custom sleep with io_uring_enter and doesn't currently
2266 : support parking. TODO: Fix and align to fd_stem in future. */
2267 : #define STEM_NEVER_PARK 1
2268 0 : #define STEM_BURST 1UL
2269 0 : #define STEM_LAZY ((long)1e6) /* 1 ms */
2270 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snapsv_t
2271 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snapsv_t)
2272 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
2273 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
2274 0 : #define STEM_CALLBACK_RETURNABLE_FRAG returnable_frag
2275 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
2276 : #include "../../disco/stem/fd_stem.c"
2277 :
2278 : #ifndef FD_TILE_TEST
2279 : fd_topo_run_tile_t fd_tile_snapsv = {
2280 : .name = "snapsv",
2281 : .rlimit_file_cnt_fn = rlimit_file_cnt,
2282 : .populate_allowed_fds = populate_allowed_fds,
2283 : .populate_allowed_seccomp = populate_allowed_seccomp,
2284 : .scratch_align = scratch_align,
2285 : .scratch_footprint = scratch_footprint,
2286 : .privileged_init = privileged_init,
2287 : .unprivileged_init = unprivileged_init,
2288 : .run = stem_run,
2289 : .rlimit_nproc = 1024UL,
2290 : };
2291 : #endif
|
