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