Line data Source code
1 : #define _GNU_SOURCE
2 : /* fd_snaprd_tile.c is a worker thread for reading accdb disk data.
3 : This tile is typically either sleeping or doing iowait. */
4 :
5 : #include "fd_backup.h"
6 : #include "../../disco/metrics/fd_metrics.h"
7 : #include "../../disco/topo/fd_topo.h"
8 : #include "../../flamenco/accdb/fd_accdb.h"
9 : #include "../../flamenco/accdb/fd_accdb_shmem.h"
10 : #include "../../tango/fseq/fd_fseq.h"
11 : #include <errno.h>
12 : #include <fcntl.h>
13 : #include <unistd.h>
14 : #include <stdatomic.h>
15 : #include <time.h>
16 : #include <linux/futex.h>
17 : #include "generated/fd_snaprd_tile_seccomp.h"
18 :
19 0 : #define SNAPRD_STATE_IDLE 0
20 126 : #define SNAPRD_STATE_READ 1
21 6 : #define SNAPRD_STATE_DONE 2
22 0 : #define SNAPRD_STATE_CAPTURE 3 /* sampling accdb partition bounds */
23 0 : #define SNAPRD_STATE_DRAIN 4 /* waiting for in-flight accdb writes to land */
24 :
25 30 : #define STEM_BURST 1UL
26 : #define SNAPRD_PART_MAX (1UL<<13)
27 :
28 : FD_STATIC_ASSERT( (FD_BACKUP_RD_MTU % 4096UL)==0UL, rd_mtu_direct_io );
29 :
30 : struct fd_snaprd {
31 : uint state;
32 :
33 : /* control signal from snapmk, which wakes up the tile */
34 : atomic_ulong * in_ctl;
35 : ulong in_ctl_seq;
36 :
37 : ulong idle_cnt;
38 :
39 : fd_accdb_shmem_t const * accdb;
40 :
41 : /* these are valid while reading a snapshot of accdb partition bounds */
42 : struct {
43 : ulong file_off;
44 : ulong sz;
45 : } part[ SNAPRD_PART_MAX ];
46 : ulong part_cnt;
47 : ulong part_idx;
48 : ulong part_cur; /* cursor in [0,part_sz] */
49 : ulong part_sz; /* byte size of partition */
50 : ulong part_file_off; /* accdb file offset of partition */
51 :
52 : fd_accdb_shmem_writer_barrier_t barrier[ 1 ]; /* writers whose pwritev2 has yet to land */
53 :
54 : struct {
55 : void * mem;
56 : ulong chunk0;
57 : ulong wmark;
58 : ulong chunk;
59 : ulong mtu;
60 : } out;
61 :
62 : struct {
63 : ulong bytes_read;
64 : ulong export_progress_bytes;
65 : ulong export_total_bytes;
66 : ulong io_blocked_ticks;
67 : } metrics;
68 :
69 : int accdb_direct_fd;
70 : };
71 :
72 : typedef struct fd_snaprd fd_snaprd_t;
73 :
74 : static ulong
75 : populate_allowed_fds( fd_topo_t const * topo,
76 : fd_topo_tile_t const * tile,
77 : ulong out_fds_cnt,
78 0 : int * out_fds ) {
79 0 : fd_snaprd_t const * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
80 0 : if( FD_UNLIKELY( out_fds_cnt<4UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
81 0 : ulong out_cnt = 0UL;
82 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
83 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
84 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
85 0 : out_fds[ out_cnt++ ] = FD_ACCDB_FD_RO; /* accounts db readonly fd */
86 0 : out_fds[ out_cnt++ ] = ctx->accdb_direct_fd; /* accounts db, O_DIRECT */
87 0 : return out_cnt;
88 0 : }
89 :
90 : static ulong
91 : populate_allowed_seccomp( fd_topo_t const * topo,
92 : fd_topo_tile_t const * tile,
93 : ulong out_cnt,
94 0 : struct sock_filter * out ) {
95 0 : fd_snaprd_t const * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
96 0 : populate_sock_filter_policy_fd_snaprd_tile( out_cnt, out, (uint)fd_log_private_logfile_fd(), (uint)ctx->accdb_direct_fd );
97 0 : return sock_filter_policy_fd_snaprd_tile_instr_cnt;
98 0 : }
99 :
100 : static void
101 : privileged_init( fd_topo_t const * topo,
102 0 : fd_topo_tile_t const * tile ) {
103 0 : fd_snaprd_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
104 :
105 0 : char path[ 64 ];
106 0 : FD_TEST( fd_cstr_printf_check( path, sizeof(path), NULL, "/proc/self/fd/%d", FD_ACCDB_FD_RO ) );
107 0 : ctx->accdb_direct_fd = open( path, O_RDONLY|O_DIRECT|O_NOATIME|O_CLOEXEC );
108 0 : if( FD_UNLIKELY( ctx->accdb_direct_fd<0 ) ) FD_LOG_ERR(( "open(accounts.db, O_DIRECT) failed (%i-%s); the filesystem holding [paths.accounts] must support direct IO", errno, fd_io_strerror( errno ) ));
109 0 : }
110 :
111 : FD_FN_CONST static inline ulong
112 0 : scratch_align( void ) {
113 0 : return alignof(fd_snaprd_t);
114 0 : }
115 :
116 : FD_FN_PURE static inline ulong
117 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
118 0 : (void)tile;
119 0 : return sizeof(fd_snaprd_t);
120 0 : }
121 :
122 : static void
123 : unprivileged_init( fd_topo_t const * topo,
124 0 : fd_topo_tile_t const * tile ) {
125 0 : fd_snaprd_t * ctx = fd_topo_obj_laddr( topo, tile->tile_obj_id );
126 :
127 0 : ctx->state = SNAPRD_STATE_IDLE;
128 0 : ctx->in_ctl = NULL;
129 0 : ctx->in_ctl_seq = 0UL;
130 0 : ctx->idle_cnt = 0UL;
131 0 : memset( &ctx->metrics, 0, sizeof(ctx->metrics) );
132 :
133 : /* snaprd queries accdb partition info to figure out where to read */
134 0 : void * _accdb_shmem = fd_topo_obj_laddr( topo, tile->snaprd.accdb_obj_id );
135 0 : ctx->accdb = fd_accdb_shmem_join( _accdb_shmem );
136 0 : FD_TEST( ctx->accdb );
137 :
138 0 : FD_CHECK_ERR( tile->out_cnt==1UL, "topology mismatch" );
139 0 : fd_topo_link_t const * out_link = &topo->links[ tile->out_link_id[ 0 ] ];
140 0 : FD_CHECK_ERR( !strcmp( out_link->name, "snaprd_out" ), "topology mismatch" );
141 0 : FD_CHECK_ERR( out_link->mtu && out_link->mtu<=UINT_MAX, "topology mismatch" );
142 :
143 :
144 0 : ulong snapmk_tile_id = fd_topo_find_tile( topo, "snapmk", 0UL );
145 0 : FD_CHECK_ERR( snapmk_tile_id!=ULONG_MAX, "missing snapmk tile" );
146 0 : fd_topo_tile_t const * snapmk_tile = &topo->tiles[ snapmk_tile_id ];
147 :
148 0 : ulong snapmk_in_idx = fd_topo_find_tile_in_link( topo, snapmk_tile, "snaprd_out", out_link->kind_id );
149 0 : FD_CHECK_ERR( snapmk_in_idx!=ULONG_MAX, "missing snapmk input link for snaprd_out" );
150 0 : FD_CHECK_ERR( snapmk_tile->in_link_reliable[ snapmk_in_idx ], "snaprd_out consumer is not reliable" );
151 :
152 0 : ulong * fseq = fd_fseq_join( fd_topo_obj_laddr( topo, snapmk_tile->in_link_fseq_obj_id[ snapmk_in_idx ] ) );
153 0 : FD_TEST( fseq );
154 0 : ctx->in_ctl = fd_fseq_app_laddr( fseq );
155 0 : FD_STATIC_ASSERT( sizeof(ulong)<=FD_FSEQ_APP_FOOTPRINT, fseq_app_space );
156 0 : ctx->in_ctl_seq = atomic_load_explicit( ctx->in_ctl, memory_order_acquire );
157 :
158 0 : ctx->out.mem = topo->workspaces[ topo->objs[ out_link->dcache_obj_id ].wksp_id ].wksp;
159 0 : ctx->out.chunk0 = fd_dcache_compact_chunk0( ctx->out.mem, out_link->dcache );
160 0 : ctx->out.wmark = fd_dcache_compact_wmark( ctx->out.mem, out_link->dcache, out_link->mtu );
161 0 : ctx->out.chunk = ctx->out.chunk0;
162 0 : ctx->out.mtu = out_link->mtu;
163 0 : }
164 :
165 : static int
166 18 : next_partition( fd_snaprd_t * ctx ) {
167 24 : while( ctx->part_idx<ctx->part_cnt ) {
168 15 : ulong part_idx = ctx->part_idx++;
169 15 : if( FD_UNLIKELY( !ctx->part[ part_idx ].sz ) ) continue;
170 9 : ctx->part_cur = 0UL;
171 9 : ctx->part_sz = ctx->part[ part_idx ].sz;
172 9 : ctx->part_file_off = ctx->part[ part_idx ].file_off;
173 9 : return 1;
174 15 : }
175 :
176 9 : ctx->part_cur = 0UL;
177 9 : ctx->part_sz = 0UL;
178 9 : ctx->part_file_off = 0UL;
179 9 : return 0;
180 18 : }
181 :
182 : /* backup_disk_capture determines which accdb partition ranges must be
183 : read to produce a snapshot. */
184 :
185 : static int
186 0 : backup_disk_capture( fd_snaprd_t * ctx ) {
187 0 : FD_CHECK_CRIT( fd_accdb_snapshot_sync_state( fd_accdb_shmem_snapshot_sync( ctx->accdb ) )==FD_ACCDB_SNAPSHOT_SYNC_RUNNING,
188 0 : "backup capture outside snapshot_sync RUNNING" );
189 :
190 : /* Accounts evicted during the cache pass may still be on their way
191 : to disk, so wait for those writers before sampling the bounds. */
192 0 : if( FD_UNLIKELY( fd_accdb_shmem_writer_barrier_poll( ctx->accdb, ctx->barrier ) ) ) return 0;
193 :
194 0 : ulong part_max = fd_accdb_shmem_partition_max( ctx->accdb );
195 0 : if( FD_UNLIKELY( part_max>SNAPRD_PART_MAX ) ) {
196 0 : FD_LOG_ERR(( "accdb partition count %lu exceeds snaprd capacity %lu", part_max, SNAPRD_PART_MAX ));
197 0 : }
198 :
199 0 : ulong part_sz = fd_accdb_shmem_partition_sz( ctx->accdb );
200 :
201 0 : ctx->part_cnt = 0UL;
202 0 : ulong export_total_bytes = 0UL;
203 0 : for( ulong i=0UL; i<part_max; i++ ) {
204 : /* the accdb partitions might grow after we save offsets into
205 : ctx->part, but we can safely ignore any future data (newly added
206 : rooted accounts will have been saved from cache, and non-rooted
207 : accounts are ignored regardless) */
208 :
209 0 : fd_accdb_shmem_partition_info_t info[ 1 ];
210 0 : fd_accdb_shmem_partition_info( ctx->accdb, i, info );
211 :
212 0 : ulong sz = info->write_offset_raw;
213 0 : if( FD_UNLIKELY( info->is_write_head && sz>part_sz ) ) return 0; /* mid-rotation, retry */
214 0 : FD_CHECK_CRIT( sz<=part_sz, "accdb partition write_offset exceeds partition size" );
215 :
216 0 : if( !sz ) continue;
217 0 : if( FD_UNLIKELY( info->compaction_offset>=sz ) ) continue; /* already compacted, partition effectively dead */
218 0 : ctx->part[ ctx->part_cnt ].file_off = i*part_sz;
219 0 : ctx->part[ ctx->part_cnt ].sz = sz;
220 0 : ctx->part_cnt++;
221 0 : export_total_bytes += sz;
222 0 : }
223 0 : ctx->metrics.export_progress_bytes = 0UL;
224 0 : ctx->metrics.export_total_bytes = export_total_bytes;
225 :
226 : /* need to wait for these writers to complete */
227 0 : fd_accdb_shmem_writer_barrier_capture( ctx->accdb, ctx->barrier );
228 0 : return 1;
229 0 : }
230 :
231 : /* backup_disk_drain waits out the writes that were in flight when
232 : backup_disk_capture sampled the partition bounds. */
233 :
234 : static void
235 0 : backup_disk_drain( fd_snaprd_t * ctx ) {
236 0 : if( FD_UNLIKELY( fd_accdb_shmem_writer_barrier_poll( ctx->accdb, ctx->barrier ) ) ) return;
237 :
238 : /* An accdb with no data on disk yields an empty stream, which snapmk
239 : still has to see terminated (a zero size frag carrying eom). */
240 0 : ctx->part_idx = 0UL;
241 0 : next_partition( ctx );
242 0 : ctx->state = SNAPRD_STATE_READ;
243 0 : }
244 :
245 : static void
246 : before_credit( fd_snaprd_t * ctx,
247 : fd_stem_context_t * stem,
248 0 : int * charge_busy ) {
249 0 : (void)charge_busy;
250 :
251 0 : ulong ctl_cur = atomic_load_explicit( ctx->in_ctl, memory_order_acquire );
252 0 : ulong ctl_prev = ctx->in_ctl_seq;
253 0 : if( FD_LIKELY( ctl_prev==ctl_cur ) ) {
254 0 : if( FD_UNLIKELY( ctx->idle_cnt++ > 16384UL && !stem->sleep ) ) fd_log_sleep( (long)1e6 );
255 0 : return;
256 0 : }
257 :
258 : /* new backup job */
259 0 : ctx->in_ctl_seq = ctl_cur;
260 0 : ctx->state = SNAPRD_STATE_CAPTURE;
261 0 : ctx->idle_cnt = 0UL;
262 0 : *charge_busy = 1;
263 :
264 0 : fd_accdb_shmem_writer_barrier_capture( ctx->accdb, ctx->barrier );
265 0 : }
266 :
267 : static void
268 : after_credit( fd_snaprd_t * ctx,
269 : fd_stem_context_t * stem,
270 : int * opt_poll_in,
271 30 : int * charge_busy ) {
272 30 : if( FD_UNLIKELY( ctx->state==SNAPRD_STATE_CAPTURE ) ) {
273 0 : *charge_busy = 1;
274 0 : if( FD_UNLIKELY( !backup_disk_capture( ctx ) ) ) return;
275 0 : ctx->state = SNAPRD_STATE_DRAIN;
276 0 : return;
277 0 : }
278 :
279 30 : if( FD_UNLIKELY( ctx->state==SNAPRD_STATE_DRAIN ) ) {
280 0 : *charge_busy = 1;
281 0 : backup_disk_drain( ctx );
282 0 : return;
283 0 : }
284 :
285 30 : if( FD_UNLIKELY( ctx->state!=SNAPRD_STATE_READ ) ) return;
286 :
287 30 : FD_CHECK_CRIT( *stem->cr_avail <= UINT_MAX, "cr_avail underflow" );
288 30 : FD_CHECK_CRIT( ctx->part_cur <= ctx->part_sz, "partition cursor overflow" );
289 30 : FD_CHECK_CRIT( fd_accdb_snapshot_sync_state( fd_accdb_shmem_snapshot_sync( ctx->accdb ) )==FD_ACCDB_SNAPSHOT_SYNC_RUNNING,
290 30 : "accdb resumed compaction while backup read in progress" );
291 :
292 30 : ulong burst_rem = STEM_BURST;
293 60 : while( ctx->state==SNAPRD_STATE_READ && stem->cr_avail[ 0 ] && burst_rem-- ) {
294 30 : ulong frag_sz = fd_ulong_min( ctx->out.mtu, ctx->part_sz-ctx->part_cur );
295 30 : ulong read_len = fd_ulong_align_up( frag_sz, 4096UL );
296 :
297 30 : ulong chunk = ctx->out.chunk;
298 30 : uchar * out = fd_chunk_to_laddr( ctx->out.mem, chunk );
299 30 : ulong src_off = ctx->part_file_off + ctx->part_cur;
300 :
301 30 : ulong read_sz = 0UL;
302 30 : long t0 = fd_tickcount();
303 120 : while( read_sz<read_len ) {
304 90 : long res = pread( ctx->accdb_direct_fd, out+read_sz, read_len-read_sz, (long)(src_off+read_sz) );
305 90 : if( FD_UNLIKELY( res<0L && (errno==EINTR || errno==EAGAIN || errno==EWOULDBLOCK) ) ) continue;
306 90 : if( FD_UNLIKELY( res<0L ) ) {
307 0 : FD_LOG_ERR(( "pread failed: %i-%s", errno, fd_io_strerror( errno ) ));
308 0 : }
309 90 : if( FD_UNLIKELY( !res ) ) {
310 0 : FD_LOG_ERR(( "pread returned EOF at offset %lu", src_off+read_sz ));
311 0 : }
312 90 : read_sz += (ulong)res;
313 90 : }
314 30 : long t1 = fd_tickcount();
315 30 : ctx->metrics.bytes_read += read_sz;
316 30 : ctx->metrics.export_progress_bytes += frag_sz;
317 30 : if( FD_LIKELY( frag_sz ) ) ctx->metrics.io_blocked_ticks += (ulong)( t1-t0 );
318 :
319 30 : ctx->part_cur += frag_sz;
320 :
321 30 : int eom = 0;
322 30 : if( FD_UNLIKELY( ctx->part_cur==ctx->part_sz && !next_partition( ctx ) ) ) {
323 6 : ctx->state = SNAPRD_STATE_DONE;
324 6 : eom = 1;
325 6 : }
326 :
327 30 : ulong sz = fd_ulong_min( frag_sz, (ulong)USHORT_MAX );
328 30 : ulong ctl = fd_frag_meta_ctl( FD_BACKUP_ORIG_DISK_FRAG, 0, eom, 0 );
329 30 : ulong tspub = frag_sz;
330 30 : fd_stem_publish( stem, 0UL, src_off, chunk, sz, ctl, 0UL, tspub );
331 30 : ctx->out.chunk = fd_dcache_compact_next( chunk, read_len, ctx->out.chunk0, ctx->out.wmark );
332 :
333 30 : ctx->idle_cnt = 0UL;
334 30 : *charge_busy = 1;
335 30 : *opt_poll_in = 0;
336 30 : }
337 30 : }
338 :
339 : static void
340 0 : metrics_write( fd_snaprd_t * ctx ) {
341 0 : FD_MCNT_SET ( SNAPRD, DISK_READ_BYTES, ctx->metrics.bytes_read );
342 0 : FD_MGAUGE_SET( SNAPRD, EXPORT_PROGRESS_BYTES, ctx->metrics.export_progress_bytes );
343 0 : FD_MGAUGE_SET( SNAPRD, EXPORT_EXPECTED_BYTES, ctx->metrics.export_total_bytes );
344 0 : FD_MCNT_SET ( SNAPRD, IO_BLOCKED_DURATION_NANOS, ctx->metrics.io_blocked_ticks );
345 0 : }
346 :
347 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_snaprd_t
348 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_snaprd_t)
349 0 : #define STEM_CALLBACK_BEFORE_CREDIT before_credit
350 0 : #define STEM_CALLBACK_AFTER_CREDIT after_credit
351 0 : #define STEM_CALLBACK_METRICS_WRITE metrics_write
352 : #include "../../disco/stem/fd_stem.c"
353 :
354 : fd_topo_run_tile_t fd_tile_snaprd = {
355 : .name = "snaprd",
356 : .populate_allowed_fds = populate_allowed_fds,
357 : .populate_allowed_seccomp = populate_allowed_seccomp,
358 : .scratch_align = scratch_align,
359 : .scratch_footprint = scratch_footprint,
360 : .privileged_init = privileged_init,
361 : .unprivileged_init = unprivileged_init,
362 : .run = stem_run,
363 : };
|
