Line data Source code
1 : #define _GNU_SOURCE
2 : #include "../../disco/tiles.h"
3 : #include "generated/fd_writer_tile_seccomp.h"
4 :
5 : #include "../../util/pod/fd_pod_format.h"
6 :
7 : #include "../../flamenco/runtime/fd_runtime.h"
8 : #include "../../flamenco/runtime/fd_runtime_public.h"
9 : #include "../../flamenco/runtime/fd_executor.h"
10 :
11 : #include "../../funk/fd_funk.h"
12 :
13 : struct fd_writer_tile_in_ctx {
14 : fd_wksp_t * mem;
15 : ulong chunk0;
16 : ulong wmark;
17 : };
18 : typedef struct fd_writer_tile_in_ctx fd_writer_tile_in_ctx_t;
19 :
20 : struct fd_writer_tile_ctx {
21 : fd_wksp_t * wksp;
22 : fd_spad_t * spad;
23 : ulong tile_cnt;
24 : ulong tile_idx;
25 : ulong exec_tile_cnt;
26 :
27 : /* R/W by this tile and the replay tile. */
28 : ulong * fseq;
29 :
30 : /* Local join of Funk. R/W. */
31 : fd_funk_t funk[1];
32 : fd_funk_txn_t * funk_txn;
33 :
34 : /* Link management. */
35 : fd_writer_tile_in_ctx_t exec_writer_in[ FD_PACK_MAX_BANK_TILES ];
36 :
37 : /* Runtime public and local joins of its members. */
38 : fd_wksp_t const * runtime_public_wksp;
39 : fd_runtime_public_t const * runtime_public;
40 :
41 : /* Local joins of exec spads. Read-only. */
42 : fd_spad_t * exec_spad[ FD_PACK_MAX_BANK_TILES ];
43 : fd_wksp_t * exec_spad_wksp[ FD_PACK_MAX_BANK_TILES ];
44 :
45 : /* Local joins of exec tile txn ctx. Read-only. */
46 : fd_exec_txn_ctx_t * txn_ctx[ FD_PACK_MAX_BANK_TILES ];
47 :
48 : /* Local join of bank manager. R/W */
49 : fd_banks_t * banks;
50 : fd_bank_t * bank;
51 : };
52 : typedef struct fd_writer_tile_ctx fd_writer_tile_ctx_t;
53 :
54 : FD_FN_CONST static inline ulong
55 0 : scratch_align( void ) {
56 0 : return 128UL;
57 0 : }
58 :
59 : FD_FN_PURE static inline ulong
60 0 : scratch_footprint( fd_topo_tile_t const * tile ) {
61 0 : (void)tile;
62 0 : ulong l = FD_LAYOUT_INIT;
63 0 : l = FD_LAYOUT_APPEND( l, alignof(fd_writer_tile_ctx_t), sizeof(fd_writer_tile_ctx_t) );
64 0 : l = FD_LAYOUT_APPEND( l, fd_spad_align(), fd_spad_footprint( FD_RUNTIME_TRANSACTION_FINALIZATION_FOOTPRINT ) );
65 0 : return FD_LAYOUT_FINI( l, scratch_align() );
66 0 : }
67 :
68 : static void
69 : join_txn_ctx( fd_writer_tile_ctx_t * ctx,
70 : ulong exec_tile_idx,
71 0 : uint txn_ctx_offset ) {
72 :
73 0 : ulong exec_spad_gaddr = fd_wksp_gaddr( ctx->exec_spad_wksp[ exec_tile_idx ], ctx->exec_spad[ exec_tile_idx ] );
74 0 : if( FD_UNLIKELY( !exec_spad_gaddr ) ) {
75 0 : FD_LOG_CRIT(( "Unable to get gaddr of exec_spad %lu", exec_tile_idx ));
76 0 : }
77 :
78 0 : ulong txn_ctx_gaddr = exec_spad_gaddr + txn_ctx_offset;
79 0 : uchar * txn_ctx_laddr = fd_wksp_laddr( ctx->exec_spad_wksp[ exec_tile_idx ], txn_ctx_gaddr );
80 0 : if( FD_UNLIKELY( !txn_ctx_laddr ) ) {
81 0 : FD_LOG_CRIT(( "Unable to get laddr of the txn ctx at gaddr 0x%lx from exec_spad %lu", txn_ctx_gaddr, exec_tile_idx ));
82 0 : }
83 :
84 0 : ctx->txn_ctx[ exec_tile_idx ] = fd_exec_txn_ctx_join( txn_ctx_laddr,
85 0 : ctx->exec_spad[ exec_tile_idx ],
86 0 : ctx->exec_spad_wksp[ exec_tile_idx ] );
87 0 : if( FD_UNLIKELY( !ctx->txn_ctx[ exec_tile_idx ] ) ) {
88 0 : FD_LOG_CRIT(( "Unable to join txn ctx at gaddr 0x%lx laddr 0x%lx from exec_spad %lu", txn_ctx_gaddr, (ulong)txn_ctx_laddr, exec_tile_idx ));
89 0 : }
90 0 : }
91 :
92 : static int
93 : before_frag( fd_writer_tile_ctx_t * ctx,
94 : ulong in_idx,
95 : ulong seq,
96 0 : ulong sig ) {
97 :
98 : /* Round-robin.
99 :
100 : The usual round-robin strategy of returning
101 : (seq % ctx->tile_cnt) != ctx->tile_idx
102 : here suffers somewhat from a sort of convoy effect.
103 : This is because exec tiles do not proceed to the next transaction
104 : until transaction finalization has been done. In other words, exec
105 : tiles block on writer tiles, rather than truly pipelining. As a
106 : result, when all the exec tiles publish to seq 0, the 0th writer
107 : tile becomes busy, and all exec tiles block on it. Then writer tile
108 : 1 becomes busy, while all other writer tiles sit idle. So on and so
109 : forth.
110 :
111 : So we offset by in_idx to try to mitigate this.
112 : */
113 0 : return ((seq+in_idx) % ctx->tile_cnt) != ctx->tile_idx && sig != FD_WRITER_BOOT_SIG; /* The boot message should go through to all writer tiles. */
114 0 : }
115 :
116 : static void
117 : during_frag( fd_writer_tile_ctx_t * ctx,
118 : ulong in_idx,
119 : ulong seq,
120 : ulong sig,
121 : ulong chunk,
122 : ulong sz,
123 0 : ulong ctl ) {
124 :
125 0 : (void)seq;
126 0 : (void)ctl;
127 :
128 : /* exec_writer is a reliable flow controlled link so we are not gonna
129 : bother with copying the incoming frag. */
130 :
131 0 : fd_writer_tile_in_ctx_t * in_ctx = &(ctx->exec_writer_in[ in_idx ]);
132 :
133 0 : if( FD_UNLIKELY( chunk < in_ctx->chunk0 || chunk > in_ctx->wmark ) ) {
134 0 : FD_LOG_CRIT(( "chunk %lu %lu corrupt, not in range [%lu,%lu]",
135 0 : chunk,
136 0 : sz,
137 0 : in_ctx->chunk0,
138 0 : in_ctx->wmark ));
139 0 : }
140 :
141 : /* Process messages from exec tiles. */
142 :
143 0 : if( FD_UNLIKELY( sig == FD_WRITER_BOOT_SIG ) ) {
144 0 : fd_runtime_public_exec_writer_boot_msg_t * msg = fd_type_pun( fd_chunk_to_laddr( in_ctx->mem, chunk ) );
145 0 : join_txn_ctx( ctx, in_idx, msg->txn_ctx_offset );
146 0 : ulong txn_ctx_cnt = 0UL;
147 0 : for( ulong i=0UL; i<ctx->exec_tile_cnt; i++ ) {
148 0 : txn_ctx_cnt += fd_ulong_if( ctx->txn_ctx[ i ]!=NULL, 1UL, 0UL );
149 0 : }
150 0 : if( txn_ctx_cnt==ctx->exec_tile_cnt ) {
151 0 : fd_fseq_update( ctx->fseq, FD_WRITER_STATE_READY );
152 0 : FD_LOG_INFO(( "writer tile %lu fully booted", ctx->tile_idx ));
153 0 : }
154 0 : return;
155 0 : }
156 :
157 0 : if( FD_LIKELY( sig == FD_WRITER_TXN_SIG ) ) {
158 0 : fd_runtime_public_exec_writer_txn_msg_t * msg = fd_type_pun( fd_chunk_to_laddr( in_ctx->mem, chunk ) );
159 0 : if( FD_UNLIKELY( msg->exec_tile_id!=in_idx ) ) {
160 0 : FD_LOG_CRIT(( "exec_tile_id %u should be == in_idx %lu", msg->exec_tile_id, in_idx ));
161 0 : }
162 0 : fd_execute_txn_task_info_t info = {0};
163 0 : info.txn_ctx = ctx->txn_ctx[ in_idx ];
164 0 : info.exec_res = info.txn_ctx->exec_err;
165 :
166 0 : fd_banks_lock( ctx->banks );
167 0 : ctx->bank = fd_banks_get_bank( ctx->banks, info.txn_ctx->slot );
168 0 : if( FD_UNLIKELY( !ctx->bank ) ) {
169 0 : FD_LOG_CRIT(( "Could not find bank for slot %lu", info.txn_ctx->slot ));
170 0 : }
171 :
172 0 : if( !ctx->funk_txn || info.txn_ctx->slot != ctx->funk_txn->xid.ul[0] ) {
173 0 : fd_funk_txn_map_t * txn_map = fd_funk_txn_map( ctx->funk );
174 0 : if( FD_UNLIKELY( !txn_map->map ) ) {
175 0 : FD_LOG_CRIT(( "Could not find valid funk transaction map" ));
176 0 : }
177 0 : fd_funk_txn_xid_t xid = { .ul = { fd_bank_slot_get( ctx->bank ), fd_bank_slot_get( ctx->bank ) } };
178 0 : fd_funk_txn_start_read( ctx->funk );
179 0 : ctx->funk_txn = fd_funk_txn_query( &xid, txn_map );
180 0 : if( FD_UNLIKELY( !ctx->funk_txn ) ) {
181 0 : FD_LOG_CRIT(( "Could not find valid funk transaction" ));
182 0 : }
183 0 : fd_funk_txn_end_read( ctx->funk );
184 0 : }
185 :
186 0 : if( FD_LIKELY( info.txn_ctx->flags & FD_TXN_P_FLAGS_EXECUTE_SUCCESS ) ) {
187 0 : FD_SPAD_FRAME_BEGIN( ctx->spad ) {
188 0 : if( FD_UNLIKELY( !ctx->bank ) ) {
189 0 : FD_LOG_CRIT(( "No bank for slot %lu", info.txn_ctx->slot ));
190 0 : }
191 :
192 0 : fd_runtime_finalize_txn( ctx->funk, ctx->funk_txn, &info, ctx->spad, ctx->bank );
193 0 : } FD_SPAD_FRAME_END;
194 0 : fd_banks_unlock( ctx->banks );
195 0 : while( fd_writer_fseq_get_state( fd_fseq_query( ctx->fseq ) )!=FD_WRITER_STATE_READY ) {
196 : /* Spin to wait for the replay tile to ack the previous txn
197 : done. */
198 0 : FD_SPIN_PAUSE();
199 0 : }
200 0 : }
201 : /* Notify the replay tile. */
202 0 : fd_fseq_update( ctx->fseq, fd_writer_fseq_set_txn_done( msg->txn_id, msg->exec_tile_id ) );
203 0 : return;
204 0 : }
205 :
206 0 : FD_LOG_CRIT(( "Unknown sig %lu", sig ));
207 0 : }
208 :
209 : static void
210 : privileged_init( fd_topo_t * topo,
211 0 : fd_topo_tile_t * tile ) {
212 0 : (void)topo;
213 0 : (void)tile;
214 0 : }
215 :
216 : static void
217 : unprivileged_init( fd_topo_t * topo,
218 0 : fd_topo_tile_t * tile ) {
219 :
220 : /********************************************************************/
221 : /* Validate allocations */
222 : /********************************************************************/
223 :
224 0 : void * scratch = fd_topo_obj_laddr( topo, tile->tile_obj_id );
225 :
226 0 : FD_SCRATCH_ALLOC_INIT( l, scratch );
227 0 : fd_writer_tile_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(fd_writer_tile_ctx_t), sizeof(fd_writer_tile_ctx_t) );
228 0 : void * spad_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_spad_align(), fd_spad_footprint( FD_RUNTIME_TRANSACTION_FINALIZATION_FOOTPRINT ) );
229 0 : ulong scratch_alloc_mem = FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
230 0 : if( FD_UNLIKELY( scratch_alloc_mem - (ulong)scratch - scratch_footprint( tile ) ) ) {
231 0 : FD_LOG_CRIT( ( "scratch_alloc_mem did not match scratch_footprint diff: %lu alloc: %lu footprint: %lu",
232 0 : scratch_alloc_mem - (ulong)scratch - scratch_footprint( tile ),
233 0 : scratch_alloc_mem,
234 0 : (ulong)scratch + scratch_footprint( tile ) ) );
235 0 : }
236 0 : fd_memset( ctx, 0, sizeof(*ctx) );
237 0 : ctx->wksp = topo->workspaces[ topo->objs[ tile->tile_obj_id ].wksp_id ].wksp;
238 0 : ctx->spad = fd_spad_join( fd_spad_new( spad_mem, FD_RUNTIME_TRANSACTION_FINALIZATION_FOOTPRINT ) );
239 :
240 : /********************************************************************/
241 : /* Links */
242 : /********************************************************************/
243 :
244 0 : ctx->tile_cnt = fd_topo_tile_name_cnt( topo, tile->name );
245 0 : ctx->tile_idx = tile->kind_id;
246 :
247 0 : ulong exec_tile_cnt = fd_topo_tile_name_cnt( topo, "exec" );
248 0 : ctx->exec_tile_cnt = exec_tile_cnt;
249 :
250 : /* Find and setup all the exec_writer links. */
251 0 : if( FD_UNLIKELY( exec_tile_cnt!=tile->in_cnt ) ) {
252 0 : FD_LOG_CRIT(( "Expecting one exec_writer link per exec tile but found %lu links and %lu tiles", tile->in_cnt, exec_tile_cnt ));
253 0 : }
254 0 : for( ulong i=0UL; i<tile->in_cnt; i++ ) {
255 0 : ulong exec_writer_idx = fd_topo_find_tile_in_link( topo, tile, "exec_writer", i );
256 0 : if( FD_UNLIKELY( exec_writer_idx==ULONG_MAX ) ) {
257 0 : FD_LOG_CRIT(( "Could not find exec_writer in-link %lu", i ));
258 0 : }
259 0 : fd_topo_link_t * exec_writer_in_link = &topo->links[ tile->in_link_id[ i ] ];
260 0 : if( FD_UNLIKELY( !exec_writer_in_link ) ) {
261 0 : FD_LOG_CRIT(( "Invalid exec_writer in-link %lu", i ));
262 0 : }
263 0 : ctx->exec_writer_in[ i ].mem = topo->workspaces[ topo->objs[ exec_writer_in_link->dcache_obj_id ].wksp_id ].wksp;
264 0 : ctx->exec_writer_in[ i ].chunk0 = fd_dcache_compact_chunk0( ctx->exec_writer_in[ i ].mem, exec_writer_in_link->dcache );
265 0 : ctx->exec_writer_in[ i ].wmark = fd_dcache_compact_wmark( ctx->exec_writer_in[ i ].mem,
266 0 : exec_writer_in_link->dcache,
267 0 : exec_writer_in_link->mtu );
268 0 : }
269 :
270 : /********************************************************************/
271 : /* Setup runtime public */
272 : /********************************************************************/
273 :
274 0 : ulong runtime_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "runtime_pub" );
275 0 : if( FD_UNLIKELY( runtime_obj_id==ULONG_MAX ) ) {
276 0 : FD_LOG_ERR(( "Could not find topology object for runtime public" ));
277 0 : }
278 :
279 0 : ctx->runtime_public_wksp = topo->workspaces[ topo->objs[ runtime_obj_id ].wksp_id ].wksp;
280 0 : if( FD_UNLIKELY( !ctx->runtime_public_wksp ) ) {
281 0 : FD_LOG_ERR(( "No runtime_public workspace" ));
282 0 : }
283 :
284 0 : ctx->runtime_public = fd_runtime_public_join( fd_topo_obj_laddr( topo, runtime_obj_id ) );
285 0 : if( FD_UNLIKELY( !ctx->runtime_public ) ) {
286 0 : FD_LOG_ERR(( "Failed to join runtime public" ));
287 0 : }
288 :
289 : /********************************************************************/
290 : /* Spad */
291 : /********************************************************************/
292 :
293 : /* Join all of the exec spads. */
294 0 : for( ulong i=0UL; i<exec_tile_cnt; i++ ) {
295 0 : ulong exec_spad_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "exec_spad.%lu", i );
296 0 : if( FD_UNLIKELY( exec_spad_obj_id==ULONG_MAX ) ) {
297 0 : FD_LOG_CRIT(( "Could not find topology object for exec_spad.%lu", i ));
298 0 : }
299 :
300 0 : ctx->exec_spad[ i ] = fd_spad_join( fd_topo_obj_laddr( topo, exec_spad_obj_id ) );
301 0 : if( FD_UNLIKELY( !ctx->exec_spad[ i ] ) ) {
302 0 : FD_LOG_CRIT(( "Failed to join exec_spad.%lu", i ));
303 0 : }
304 0 : ctx->exec_spad_wksp[ i ] = fd_wksp_containing( ctx->exec_spad[ i ] );
305 0 : if( FD_UNLIKELY( !ctx->exec_spad_wksp[ i ] ) ) {
306 0 : FD_LOG_CRIT(( "Failed to find wksp for exec_spad.%lu", i ));
307 0 : }
308 0 : }
309 :
310 : /********************************************************************/
311 : /* Funk */
312 : /********************************************************************/
313 :
314 0 : if( FD_UNLIKELY( !fd_funk_join( ctx->funk, fd_topo_obj_laddr( topo, tile->writer.funk_obj_id ) ) ) ) {
315 0 : FD_LOG_ERR(( "Failed to join database cache" ));
316 0 : }
317 :
318 : /********************************************************************/
319 : /* Setup fseq */
320 : /********************************************************************/
321 :
322 0 : ulong writer_fseq_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "writer_fseq.%lu", ctx->tile_idx );
323 0 : ctx->fseq = fd_fseq_join( fd_topo_obj_laddr( topo, writer_fseq_id ) );
324 0 : if( FD_UNLIKELY( !ctx->fseq ) ) {
325 0 : FD_LOG_CRIT(( "writer tile %lu fseq setup failed", ctx->tile_idx ));
326 0 : }
327 0 : fd_fseq_update( ctx->fseq, FD_WRITER_STATE_NOT_BOOTED );
328 :
329 : /********************************************************************/
330 : /* Bank */
331 : /********************************************************************/
332 :
333 0 : ulong banks_obj_id = fd_pod_queryf_ulong( topo->props, ULONG_MAX, "banks" );
334 0 : if( FD_UNLIKELY( banks_obj_id==ULONG_MAX ) ) {
335 0 : FD_LOG_ERR(( "Could not find topology object for banks" ));
336 0 : }
337 :
338 0 : ctx->banks = fd_banks_join( fd_topo_obj_laddr( topo, banks_obj_id ) );
339 0 : if( FD_UNLIKELY( !ctx->banks ) ) {
340 0 : FD_LOG_ERR(( "Failed to join banks" ));
341 0 : }
342 0 : }
343 :
344 : static ulong
345 : populate_allowed_seccomp( fd_topo_t const * topo,
346 : fd_topo_tile_t const * tile,
347 : ulong out_cnt,
348 0 : struct sock_filter * out ) {
349 0 : (void)topo;
350 0 : (void)tile;
351 :
352 0 : populate_sock_filter_policy_fd_writer_tile( out_cnt, out, (uint)fd_log_private_logfile_fd() );
353 0 : return sock_filter_policy_fd_writer_tile_instr_cnt;
354 0 : }
355 :
356 : static ulong
357 : populate_allowed_fds( fd_topo_t const * topo,
358 : fd_topo_tile_t const * tile,
359 : ulong out_fds_cnt,
360 0 : int * out_fds ) {
361 0 : (void)topo;
362 0 : (void)tile;
363 :
364 0 : if( FD_UNLIKELY( out_fds_cnt<2UL ) ) FD_LOG_ERR(( "out_fds_cnt %lu", out_fds_cnt ));
365 :
366 0 : ulong out_cnt = 0UL;
367 0 : out_fds[ out_cnt++ ] = 2; /* stderr */
368 0 : if( FD_LIKELY( -1!=fd_log_private_logfile_fd() ) )
369 0 : out_fds[ out_cnt++ ] = fd_log_private_logfile_fd(); /* logfile */
370 0 : return out_cnt;
371 0 : }
372 :
373 0 : #define STEM_BURST (1UL)
374 :
375 0 : #define STEM_CALLBACK_CONTEXT_TYPE fd_writer_tile_ctx_t
376 0 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(fd_writer_tile_ctx_t)
377 :
378 0 : #define STEM_CALLBACK_BEFORE_FRAG before_frag
379 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
380 :
381 : #include "../../disco/stem/fd_stem.c"
382 :
383 : fd_topo_run_tile_t fd_tile_writer = {
384 : .name = "writer",
385 : .loose_footprint = 0UL,
386 : .populate_allowed_seccomp = populate_allowed_seccomp,
387 : .populate_allowed_fds = populate_allowed_fds,
388 : .scratch_align = scratch_align,
389 : .scratch_footprint = scratch_footprint,
390 : .privileged_init = privileged_init,
391 : .unprivileged_init = unprivileged_init,
392 : .run = stem_run,
393 : };
|
