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