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