Line data Source code
1 : #include "../../shared/fd_config.h"
2 : #include "../../shared/fd_bootinfo.h"
3 : #include "../../shared/fd_action.h"
4 : #include "../../../disco/metrics/fd_metrics.h"
5 : #include "../../../disco/fd_clock_tile.h"
6 : #include "../../../disco/net/fd_net_tile.h"
7 : #include "../../../disco/shred/fd_fec_set.h"
8 : #include "../../../util/clock/fd_clock.h"
9 : #include "../../../util/net/fd_pcap.h"
10 :
11 : #include <errno.h>
12 : #include <fcntl.h>
13 : #include <signal.h>
14 : #include <stdio.h>
15 : #include <unistd.h>
16 :
17 0 : #define FD_DUMP_FLAGS_NO_DATA 1
18 0 : #define FD_DUMP_FLAGS_NET_RX 2
19 0 : #define FD_DUMP_FLAGS_OVERRIDE_SZ 4 /* high 24 bits give size */
20 :
21 : struct dump_ctx {
22 : fd_io_buffered_ostream_t ostream;
23 : ulong pending_sz;
24 :
25 : struct link {
26 : fd_topo_link_t const * topo;
27 : union {
28 : void const * dcache_base;
29 : fd_net_rx_bounds_t net_rx_bounds[1];
30 : };
31 : uint link_hash;
32 : int flags; /* FD_DUMP_FLAGS_* */
33 : } links[ FD_TOPO_MAX_LINKS ];
34 : ulong link_cnt;
35 :
36 : ulong * metrics_base;
37 : long next_stat_log;
38 : ulong frags;
39 : ulong bytes;
40 : ulong last_overrun_frags;
41 :
42 : long warmup_until;
43 :
44 : fd_clock_tile_t clock[ 1 ];
45 : };
46 : typedef struct dump_ctx dump_ctx_t;
47 :
48 : void
49 : dump_cmd_args( int * argc,
50 : char * * * argv,
51 0 : args_t * args ) {
52 0 : char const * out_file = fd_env_strip_cmdline_cstr( argc, argv, "--out-file", NULL, "dump.pcap" );
53 0 : char const * link = fd_env_strip_cmdline_cstr( argc, argv, "--link", NULL, "" );
54 0 : args->dump.once = fd_env_strip_cmdline_int( argc, argv, "--once", NULL, 1 );
55 0 : fd_cstr_fini( fd_cstr_append_cstr_safe( fd_cstr_init( args->dump.pcap_path ), out_file, sizeof(args->dump.pcap_path)-1UL ) );
56 0 : fd_cstr_fini( fd_cstr_append_cstr_safe( fd_cstr_init( args->dump.link_name ), link, sizeof(args->dump.link_name)-1UL ) );
57 0 : }
58 :
59 : static FD_FN_CONST ulong
60 : frag_sz( int flags,
61 0 : ulong line_sz ) {
62 0 : return fd_ulong_if( flags & FD_DUMP_FLAGS_NO_DATA, 0UL,
63 0 : fd_ulong_if( flags & FD_DUMP_FLAGS_OVERRIDE_SZ, ((ulong)flags)>>8, line_sz ) );
64 0 : }
65 : static FD_FN_PURE void const *
66 : frag_ptr( int flags,
67 : struct link const * link,
68 : ulong chunk,
69 0 : ulong ctl ) {
70 0 : if( FD_UNLIKELY( flags & FD_DUMP_FLAGS_NET_RX ) ) return fd_net_rx_translate_frag( link->net_rx_bounds, chunk, ctl, 0UL );
71 0 : else return fd_chunk_to_laddr_const ( link->dcache_base, chunk );
72 0 : }
73 :
74 : static void
75 : dump_link_once( dump_ctx_t * ctx,
76 0 : struct link const * link ) {
77 0 : fd_frag_meta_t const * mcache = link->topo->mcache;
78 0 : ulong seq0 = fd_mcache_seq0( mcache );
79 0 : ulong seq_init = fd_mcache_seq_query( fd_mcache_seq_laddr_const( mcache ) );
80 0 : ulong depth = fd_mcache_depth( mcache );
81 :
82 0 : ulong frags = 0UL;
83 0 : ulong bytes = 0UL;
84 :
85 0 : int flags = link->flags;
86 :
87 : /* For links that are published reliably, we can trust the value of
88 : seq_init. For unreliable links, we'll just publish one whole
89 : depth. */
90 :
91 : /* We know at this point [seq0, seq_init) were published, but they may
92 : be long overwritten, and there may be more published than that. */
93 :
94 0 : ulong min_seq_seen = seq_init; /* What we actually dumped is [min_seq_seen, seq_init) */
95 0 : for( ulong seq=fd_seq_dec( seq_init, 1UL ); fd_seq_ge( seq, seq0 ); seq=fd_seq_dec( seq, 1UL ) ) {
96 : /* It's not necessary for this to be atomic, since this is a
97 : post-mortem tool. */
98 0 : fd_frag_meta_t const * line = mcache+fd_mcache_line_idx( seq, depth );
99 0 : ulong read_seq = fd_frag_meta_seq_query( line );
100 0 : if( FD_UNLIKELY( read_seq!=seq ) ) break;
101 :
102 0 : min_seq_seen = seq;
103 :
104 0 : ulong sz = frag_sz ( flags, line->sz );
105 0 : void const * buffer = frag_ptr( flags, link, line->chunk, line->ctl );
106 :
107 0 : fd_pcap_ostream_pkt( &ctx->ostream, (long)seq, line, sizeof(fd_frag_meta_t), buffer, sz, link->link_hash );
108 0 : bytes += sz;
109 0 : frags++;
110 0 : }
111 :
112 : /* Now check everything after seq_init. This could potentially loop
113 : forever if the producer is still going, so we cap it at one depth. */
114 0 : for( ulong off=0UL; off<depth; off++ ) {
115 0 : ulong seq = fd_seq_inc( seq_init, off );
116 :
117 0 : fd_frag_meta_t const * line = mcache+fd_mcache_line_idx( seq, depth );
118 0 : ulong read_seq = fd_frag_meta_seq_query( line );
119 :
120 : /* Skip anything we processed in the first loop, also skipping any
121 : with the err control bit set. When an mcache is initialized, all
122 : the frags have err set to true, so this will skip those in
123 : particular as well. */
124 0 : if( FD_UNLIKELY( (fd_seq_le( min_seq_seen, read_seq ) & fd_seq_lt( read_seq, seq_init )) | (line->ctl & (1<<2)) ) ) continue;
125 :
126 0 : ulong sz = frag_sz ( flags, line->sz );
127 0 : void const * buffer = frag_ptr( flags, link, line->chunk, line->ctl );
128 :
129 0 : fd_pcap_ostream_pkt( &ctx->ostream, (long)seq, line, sizeof(fd_frag_meta_t), buffer, sz, link->link_hash );
130 0 : bytes += sz;
131 0 : frags++;
132 0 : }
133 0 : FD_LOG_NOTICE(( "dumped %lu frags, %lu bytes in total from %s:%lu. Link hash: 0x%x",
134 0 : frags, bytes, link->topo->name, link->topo->kind_id, link->link_hash ));
135 0 : }
136 :
137 : static int running = 1;
138 :
139 : static void
140 0 : exit_signal( int sig FD_PARAM_UNUSED ) {
141 0 : running = 0;
142 0 : }
143 :
144 : static int
145 0 : should_shutdown( dump_ctx_t * ctx FD_PARAM_UNUSED ) {
146 0 : return !running;
147 0 : }
148 :
149 : static void
150 : during_frag( dump_ctx_t * ctx,
151 : ulong in_idx,
152 : ulong seq,
153 : ulong sig FD_PARAM_UNUSED ,
154 : ulong chunk,
155 : ulong sz,
156 0 : ulong ctl ) {
157 : /* We have a new candidate fragment to copy into the dump file.
158 : Because we attach read-only and do not backpressure any producers,
159 : this fragment could be overrun at any point during this function.
160 : So we copy the relevant data into local memory (in the buffered
161 : ostream peek space) and only commit it to the file later in
162 : after_frag. after_frag is only called if the fragment was not
163 : overrun while we were processing it. */
164 :
165 0 : ctx->pending_sz = 0UL;
166 :
167 0 : struct link const * llink = ctx->links+in_idx;
168 0 : int flags = llink->flags;
169 :
170 0 : sz = frag_sz( flags, sz );
171 :
172 :
173 0 : long pcap_sz = fd_pcap_pkt_sz( sizeof(fd_frag_meta_t), sz );
174 0 : if( FD_UNLIKELY( pcap_sz < 0L ) ) return;
175 :
176 0 : if( (ulong)pcap_sz > fd_io_buffered_ostream_peek_sz( &ctx->ostream ) ) {
177 0 : fd_io_buffered_ostream_flush( &ctx->ostream );
178 0 : if( FD_UNLIKELY( (ulong)pcap_sz > fd_io_buffered_ostream_peek_sz( &ctx->ostream ) ) ) {
179 0 : FD_LOG_ERR(( "packet size %ld too large for pcap, increase write buffer size (%lu)",
180 0 : pcap_sz, fd_io_buffered_ostream_peek_sz( &ctx->ostream ) ));
181 0 : }
182 0 : }
183 :
184 :
185 0 : fd_topo_link_t const * link = ctx->links[ in_idx ].topo;
186 0 : fd_frag_meta_t const * mline = link->mcache + fd_mcache_line_idx( seq, fd_mcache_depth( link->mcache ) );
187 0 : void const * frag_data = frag_ptr( flags, llink, chunk, ctl );
188 :
189 0 : fd_pcap_pkt( fd_io_buffered_ostream_peek( &ctx->ostream ),
190 0 : 0L,
191 0 : mline,
192 0 : sizeof(fd_frag_meta_t),
193 0 : frag_data,
194 0 : sz,
195 0 : ctx->links[ in_idx ].link_hash );
196 0 : ctx->pending_sz = (ulong)pcap_sz;
197 0 : }
198 :
199 : static void
200 : after_frag( dump_ctx_t * ctx,
201 : ulong in_idx FD_PARAM_UNUSED,
202 : ulong seq FD_PARAM_UNUSED,
203 : ulong sig FD_PARAM_UNUSED,
204 : ulong sz,
205 : ulong tsorig FD_PARAM_UNUSED,
206 : ulong tspub,
207 0 : fd_stem_context_t * stem FD_PARAM_UNUSED ) {
208 0 : if( FD_UNLIKELY( !ctx->pending_sz ) ) return;
209 :
210 0 : sz = frag_sz( ctx->links[ in_idx ].flags, sz );
211 0 : long pcap_sz = fd_pcap_pkt_sz( sizeof(fd_frag_meta_t), sz );
212 0 : FD_TEST( (ulong)pcap_sz == ctx->pending_sz );
213 :
214 0 : long tickcount = fd_tickcount();
215 0 : long now = fd_clock_epoch_y( ctx->clock->shmem->epoch, tickcount );
216 0 : if( FD_LIKELY( now>ctx->warmup_until ) ) {
217 0 : if( FD_LIKELY( tspub!=0UL ) ) {
218 0 : long tspub_decomp = fd_frag_meta_ts_decomp( tspub, tickcount );
219 0 : long tspub_wallclock = fd_clock_epoch_y( ctx->clock->shmem->epoch, tspub_decomp );
220 0 : uint * pcap_tss = fd_io_buffered_ostream_peek( &ctx->ostream );
221 0 : pcap_tss[ 0 ] = (uint)(((ulong)tspub_wallclock) / (ulong)1e9);
222 0 : pcap_tss[ 1 ] = (uint)(((ulong)tspub_wallclock) % (ulong)1e9); /* Actually nsec */
223 0 : }
224 0 : fd_io_buffered_ostream_seek( &ctx->ostream, ctx->pending_sz );
225 0 : ctx->frags += 1UL;
226 0 : ctx->bytes += sz;
227 0 : }
228 0 : ctx->pending_sz = 0UL;
229 0 : }
230 :
231 : static void
232 0 : log_metrics( dump_ctx_t * ctx ) {
233 0 : FD_LOG_NOTICE(( "dumped %lu frags, %lu bytes", ctx->frags, ctx->bytes ));
234 0 : ctx->frags = 0UL;
235 0 : ctx->bytes = 0UL;
236 :
237 0 : ulong overrun_frags = 0UL;
238 0 : for( ulong i=0UL; i<ctx->link_cnt; i++) {
239 0 : volatile ulong const * link_metrics = fd_metrics_link_in( ctx->metrics_base, i );
240 0 : overrun_frags += link_metrics[ FD_METRICS_COUNTER_LINK_FRAG_POLLING_OVERRUN_OFF ];
241 0 : overrun_frags += link_metrics[ FD_METRICS_COUNTER_LINK_FRAG_READING_OVERRUN_OFF ];
242 0 : }
243 0 : if( FD_UNLIKELY( overrun_frags != ctx->last_overrun_frags ) ) {
244 : /* Note: We expect overruns at startup because we have no way to
245 : know the current seq to start polling from, so we start at 0
246 : which has often been overrun. */
247 0 : if( FD_LIKELY( ctx->last_overrun_frags != 0 ) ) {
248 0 : long overrun_diff = fd_seq_diff( overrun_frags, ctx->last_overrun_frags );
249 0 : FD_LOG_WARNING(( "overrun detected, %ld frags dropped", overrun_diff ));
250 0 : }
251 0 : ctx->last_overrun_frags = overrun_frags;
252 0 : }
253 0 : }
254 :
255 : static void
256 0 : during_housekeeping( dump_ctx_t * ctx ) {
257 0 : long now = fd_clock_tile_now( ctx->clock );
258 0 : if( FD_UNLIKELY( now >= fd_clock_tile_recal_next( ctx->clock ) ) ) {
259 0 : fd_clock_tile_recal( ctx->clock );
260 0 : }
261 0 : if( FD_UNLIKELY( now > ctx->next_stat_log ) ) {
262 0 : ctx->next_stat_log = now + (long)1e9;
263 0 : log_metrics( ctx );
264 0 : }
265 0 : }
266 :
267 : #define STEM_BURST (0UL)
268 : #define STEM_CALLBACK_CONTEXT_TYPE dump_ctx_t
269 : #define STEM_CALLBACK_CONTEXT_ALIGN alignof(dump_ctx_t)
270 0 : #define STEM_CALLBACK_DURING_FRAG during_frag
271 0 : #define STEM_CALLBACK_AFTER_FRAG after_frag
272 : #define STEM_CALLBACK_SHOULD_SHUTDOWN should_shutdown
273 0 : #define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
274 : #include "../../../disco/stem/fd_stem.c"
275 :
276 : void
277 : dump_cmd_fn( args_t * args,
278 0 : fd_config_t * config ) {
279 0 : char * tokens[ 16 ];
280 0 : ulong token_count = fd_cstr_tokenize( tokens, 16UL, args->dump.link_name, ',' );
281 :
282 0 : dump_ctx_t ctx = { 0 };
283 :
284 0 : uchar write_buf[ 131072 ];
285 0 : int fd = open( args->dump.pcap_path, O_WRONLY | O_CREAT | O_TRUNC,
286 0 : S_IRUSR | S_IWUSR | S_IRGRP | S_IWGRP | S_IROTH | S_IWOTH );
287 0 : if( FD_UNLIKELY( fd < 0 ) )
288 0 : FD_LOG_ERR(( "open() failed (%i-%s)", errno, fd_io_strerror( errno ) ));
289 0 : if( FD_UNLIKELY( &ctx.ostream != fd_io_buffered_ostream_init( &ctx.ostream, fd, write_buf, sizeof(write_buf) ) ) )
290 0 : FD_LOG_ERR(( "fd_io_buffered_ostream_init() failed" ));
291 :
292 0 : FD_TEST( 0==fd_pcap_ostream_hdr( &ctx.ostream, FD_PCAP_LINK_LAYER_USER0 ) );
293 :
294 0 : fd_topo_t * topo = &config->topo;
295 0 : fd_bootinfo_adopt( config );
296 0 : fd_bootinfo_check_layout( config );
297 0 : fd_topo_join_workspaces( topo, FD_SHMEM_JOIN_MODE_READ_ONLY, FD_TOPO_CORE_DUMP_LEVEL_DISABLED );
298 0 : fd_topo_fill( topo );
299 :
300 0 : for( ulong i=0UL; i<topo->link_cnt; i++) {
301 0 : fd_topo_link_t * link = &topo->links[ i ];
302 0 : int found = (token_count==0UL);
303 0 : for( ulong j=0UL; (!found)&(j<token_count); j++ ) found |= !strcmp( tokens[ j ], topo->links[ i ].name );
304 0 : if( found ) {
305 0 : if( FD_UNLIKELY( NULL==link->mcache ) )
306 0 : FD_LOG_ERR(( "link %s:%lu mcache is null", link->name, link->kind_id ));
307 0 : ctx.links[ ctx.link_cnt ].topo = link;
308 0 : ctx.links[ ctx.link_cnt ].dcache_base = link->mtu ? fd_topo_obj_wksp_base( topo, link->dcache_obj_id ) : NULL;
309 0 : ctx.links[ ctx.link_cnt ].link_hash = (uint)((fd_hash( 17UL, link->name, strlen( link->name ) ) << 8) | link->kind_id);
310 :
311 0 : int flags = fd_int_if( !link->mtu, FD_DUMP_FLAGS_NO_DATA, 0 );
312 0 : if( FD_UNLIKELY( !strcmp( link->name, "shred_store" ) ) ) {
313 0 : flags |= FD_DUMP_FLAGS_OVERRIDE_SZ | (sizeof(fd_fec_set_t)<<8);
314 0 : }
315 0 : if( FD_UNLIKELY( strstr( link->name, "net_" ) ) ) {
316 0 : flags |= FD_DUMP_FLAGS_NET_RX;
317 : /* this overwrites dcache_base since it's part of the union, but
318 : that's okay. */
319 0 : uchar * dcache = fd_dcache_join( fd_topo_obj_laddr( topo, link->dcache_obj_id ) );
320 0 : fd_net_rx_bounds_init( ctx.links[ ctx.link_cnt ].net_rx_bounds, dcache );
321 0 : fd_dcache_leave( dcache );
322 0 : }
323 0 : ctx.links[ ctx.link_cnt ].flags = flags;
324 :
325 0 : ctx.link_cnt++;
326 0 : }
327 0 : }
328 :
329 0 : if( args->dump.once ) {
330 0 : for( ulong i=0UL; i<ctx.link_cnt; i++ ) {
331 0 : dump_link_once( &ctx, &ctx.links[ i ] );
332 0 : }
333 0 : } else {
334 0 : struct sigaction sa = {
335 0 : .sa_handler = exit_signal,
336 0 : .sa_flags = 0,
337 0 : };
338 0 : if( FD_UNLIKELY( sigaction( SIGTERM, &sa, NULL ) ) )
339 0 : FD_LOG_ERR(( "sigaction(SIGTERM) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
340 0 : if( FD_UNLIKELY( sigaction( SIGINT, &sa, NULL ) ) )
341 0 : FD_LOG_ERR(( "sigaction(SIGINT) failed (%i-%s)", errno, fd_io_strerror( errno ) ));
342 :
343 0 : fd_frag_meta_t const * mcaches[ FD_TOPO_MAX_LINKS ];
344 0 : for( ulong i=0UL; i<ctx.link_cnt; i++ ) {
345 0 : mcaches[ i ] = ctx.links[ i ].topo->mcache;
346 0 : }
347 :
348 0 : uchar fseq_mem[ FD_TOPO_MAX_LINKS ][ FD_FSEQ_FOOTPRINT ] __attribute__((aligned((FD_FSEQ_ALIGN))));
349 0 : ulong * fseqs[ FD_TOPO_MAX_LINKS ];
350 0 : for( ulong i=0UL; i<ctx.link_cnt; i++ ) {
351 0 : fseqs[ i ] = fd_fseq_join( fd_fseq_new( fseq_mem[ i ], 0UL ) );
352 0 : }
353 :
354 0 : fd_rng_t _rng[1];
355 0 : fd_rng_t * rng = fd_rng_join( fd_rng_new( _rng, (uint)fd_tickcount(), 0UL ) );
356 :
357 0 : uchar __attribute__((aligned(FD_STEM_SCRATCH_ALIGN))) scratch[ stem_scratch_footprint( ctx.link_cnt, 0UL, 0UL ) ];
358 0 : uchar __attribute__((aligned(FD_METRICS_ALIGN))) metrics_mem[ FD_METRICS_FOOTPRINT( ctx.link_cnt ) ];
359 :
360 0 : ctx.metrics_base = fd_metrics_join( fd_metrics_new( metrics_mem, ctx.link_cnt ) );
361 0 : fd_metrics_register( ctx.metrics_base );
362 :
363 : /* The purpose of the warmup period is to ensure that all frags in
364 : the pcap were from after the dump was started. fd_stem currently
365 : has no way to start from the most recently published frags when
366 : joining links, so at startup we get a full ~depth set of callbacks
367 : for old frags. This code assumes that we are caught up and only
368 : processing new frags by the time the warmup timer expires. */
369 0 : fd_clock_tile_init( ctx.clock );
370 0 : ctx.warmup_until = fd_clock_tile_now( ctx.clock ) + (long)1e9;
371 0 : ctx.next_stat_log = ctx.warmup_until + (long)1e9;
372 :
373 0 : stem_run1( ctx.link_cnt, /* in_cnt */
374 0 : mcaches, /* in_mcache */
375 0 : fseqs, /* in_fseq */
376 0 : 0UL, /* out_cnt */
377 0 : NULL, /* out_mcache */
378 0 : 0UL, /* cons_cnt */
379 0 : NULL, /* _cons_out */
380 0 : NULL, /* _cons_fseq */
381 0 : NULL, /* cons_slow */
382 0 : 0UL, /* burst */
383 0 : 0UL, /* lazy */
384 0 : rng, /* rng */
385 0 : scratch, /* scratch */
386 0 : &ctx ); /* ctx */
387 :
388 0 : for( ulong i=0UL; i<ctx.link_cnt; i++ ) {
389 0 : struct link const * link = &ctx.links[ i ];
390 0 : volatile ulong const * link_metrics = fd_metrics_link_in( ctx.metrics_base, i );
391 0 : ulong frags = link_metrics[ FD_METRICS_COUNTER_LINK_FRAG_CONSUMED_OFF ];
392 0 : ulong bytes = link_metrics[ FD_METRICS_COUNTER_LINK_FRAG_CONSUMED_BYTES_OFF ];
393 0 : FD_LOG_NOTICE(( "dumped %lu frags, %lu bytes in total from %s:%lu. Link hash: 0x%x",
394 0 : frags, bytes, link->topo->name, link->topo->kind_id, link->link_hash ));
395 0 : }
396 :
397 0 : fd_metrics_delete( fd_metrics_leave( ctx.metrics_base ) );
398 0 : ctx.metrics_base = NULL;
399 :
400 0 : fd_rng_delete( fd_rng_leave( rng ) );
401 :
402 0 : for( ulong i=0UL; i<ctx.link_cnt; i++ ) {
403 0 : fd_fseq_delete( fd_fseq_leave( fseqs[ i ] ) );
404 0 : }
405 0 : }
406 :
407 0 : fd_topo_leave_workspaces( topo );
408 :
409 0 : fd_io_buffered_ostream_flush( &ctx.ostream );
410 0 : fd_io_buffered_ostream_fini( &ctx.ostream );
411 0 : close( fd );
412 0 : }
413 :
414 : action_t fd_action_dump = {
415 : .name = "dump",
416 : .args = dump_cmd_args,
417 : .fn = dump_cmd_fn,
418 : .perm = NULL,
419 : .description = "Dump tango links to a packet capture file",
420 : .is_diagnostic = 1
421 : };
|
