Skip to content

Commit 2349400

Browse files
riptlriptl
committed
Revive test_dedup
1 parent 6b2baba commit 2349400

File tree

12 files changed

+460
-751
lines changed

12 files changed

+460
-751
lines changed

‎src/app/firedancer-dev/commands/gossip.c

Lines changed: 0 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -88,7 +88,6 @@ gossip_topo( config_t * config ) {
8888

8989
fd_topos_net_tile_finish( topo, 0UL );
9090
fd_topob_auto_layout( topo, 0 );
91-
topo->agave_affinity_cnt = 0;
9291
fd_topob_finish( topo, CALLBACKS );
9392
fd_topo_print_log( /* stdout */ 1, topo );
9493
}

‎src/app/shared/fd_action.h

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -98,6 +98,10 @@ union fdctl_args {
9898
ushort listen_port;
9999
} udpecho;
100100

101+
struct {
102+
char affinity[ AFFINITY_SZ ];
103+
} test_dedup;
104+
101105
};
102106

103107
typedef union fdctl_args args_t;

‎src/app/shared_dev/commands/pktgen/pktgen.c

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -43,7 +43,7 @@ pktgen_topo( config_t * config ) {
4343
}
4444
if( FD_LIKELY( !is_auto_affinity ) ) {
4545
if( FD_UNLIKELY( affinity_tile_cnt!=4UL ) )
46-
FD_LOG_ERR(( "Invalid [development.pktgen.affinity]: must include exactly three CPUs" ));
46+
FD_LOG_ERR(( "Invalid [development.pktgen.affinity]: must include exactly 4 CPUs" ));
4747
}
4848

4949
/* Reset topology from scratch */
@@ -71,7 +71,6 @@ pktgen_topo( config_t * config ) {
7171

7272
fd_topos_net_tile_finish( topo, 0UL );
7373
if( FD_UNLIKELY( is_auto_affinity ) ) fd_topob_auto_layout( topo, 0 );
74-
topo->agave_affinity_cnt = 0;
7574
fd_topob_finish( topo, CALLBACKS );
7675
fd_topo_print_log( /* stdout */ 1, topo );
7776
}

‎src/app/shared_dev/commands/test_dedup/Local.mk

Lines changed: 11 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,11 @@
1+
ifdef FD_HAS_HOSTED
2+
ifdef FD_HAS_LINUX # FIXME why is this needed
3+
ifdef FD_HAS_AVX
4+
5+
$(call add-objs,test_dedup,fddev_shared)
6+
$(call add-objs,test_dedup_rx_tile,fddev_shared)
7+
$(call add-objs,test_dedup_tx_tile,fddev_shared)
8+
9+
endif
10+
endif
11+
endif

‎src/app/shared_dev/commands/test_dedup/test_dedup.c

Lines changed: 88 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,88 @@
1+
#include "../../../shared/commands/configure/configure.h" /* CONFIGURE_CMD_INIT */
2+
#include "../../../shared/commands/run/run.h" /* fdctl_check_configure */
3+
#include "../../../../disco/net/fd_net_tile.h"
4+
#include "../../../../disco/topo/fd_topob.h"
5+
#include "../../../../disco/topo/fd_cpu_topo.h"
6+
#include "../../../../util/tile/fd_tile_private.h" /* fd_tile_private_cpus_parse */
7+
8+
#include <unistd.h> /* pause */
9+
10+
extern fd_topo_obj_callbacks_t * CALLBACKS[];
11+
12+
fd_topo_run_tile_t
13+
fdctl_tile_run( fd_topo_tile_t const * tile );
14+
15+
static void
16+
test_dedup_topo( config_t * config,
17+
char const * affinity ) {
18+
int is_auto_affinity = !strcmp( affinity, "auto" );
19+
ushort parsed_tile_to_cpu[ FD_TILE_MAX ];
20+
for( ulong i=0UL; i<FD_TILE_MAX; i++ ) parsed_tile_to_cpu[ i ] = USHORT_MAX;
21+
22+
fd_topo_cpus_t cpus[1];
23+
fd_topo_cpus_init( cpus );
24+
25+
ulong affinity_tile_cnt = 0UL;
26+
if( FD_LIKELY( !is_auto_affinity ) ) affinity_tile_cnt = fd_tile_private_cpus_parse( affinity, parsed_tile_to_cpu );
27+
28+
ulong tile_to_cpu[ FD_TILE_MAX ] = {0};
29+
for( ulong i=0UL; i<affinity_tile_cnt; i++ ) {
30+
if( FD_UNLIKELY( parsed_tile_to_cpu[ i ]!=USHORT_MAX && parsed_tile_to_cpu[ i ]>=cpus->cpu_cnt ) )
31+
FD_LOG_ERR(( "The --affinity flag specifies a CPU index of %hu, but the system only has %lu CPUs. You should either change the CPU allocations in the affinity string, or increase the number of CPUs in the system.",
32+
parsed_tile_to_cpu[ i ], cpus->cpu_cnt ));
33+
tile_to_cpu[ i ] = fd_ulong_if( parsed_tile_to_cpu[ i ]==USHORT_MAX, ULONG_MAX, (ulong)parsed_tile_to_cpu[ i ] );
34+
}
35+
if( FD_LIKELY( !is_auto_affinity ) ) {
36+
if( FD_UNLIKELY( affinity_tile_cnt!=3UL ) )
37+
FD_LOG_ERR(( "Invalid --affinity: must include exactly 3 CPUs" ));
38+
}
39+
40+
/* Reset topology from scratch */
41+
fd_topo_t * topo = &config->topo;
42+
fd_topob_new( &config->topo, config->name );
43+
topo->max_page_size = fd_cstr_to_shmem_page_sz( config->hugetlbfs.max_page_size );
44+
45+
fd_topob_wksp( topo, "test_dedup" );
46+
fd_topob_wksp( topo, "metric_in" );
47+
fd_topob_tile( topo, "dedup", "test_dedup", "metric_in", tile_to_cpu[ 0 ], 0, 0 );
48+
fd_topob_tile( topo, "TDupTx", "test_dedup", "metric_in", tile_to_cpu[ 1 ], 0, 0 );
49+
fd_topob_tile( topo, "TDupRx", "test_dedup", "metric_in", tile_to_cpu[ 2 ], 0, 0 );
50+
51+
if( FD_UNLIKELY( is_auto_affinity ) ) fd_topob_auto_layout( topo, 0 );
52+
fd_topob_finish( topo, CALLBACKS );
53+
fd_topo_print_log( /* stdout */ 1, topo );
54+
}
55+
56+
void
57+
test_dedup_cmd_args( int * pargc,
58+
char *** pargv,
59+
args_t * args ) {
60+
char const * affinity = fd_env_strip_cmdline_cstr( pargc, pargv, "--affinity", "1-3", 0 );
61+
FD_TEST( strlen( affinity )<sizeof(args->test_dedup.affinity) );
62+
fd_cstr_fini( fd_cstr_append_cstr( fd_cstr_init( args->test_dedup.affinity ), args->test_dedup.affinity ) );
63+
}
64+
65+
void
66+
test_dedup_cmd_fn( args_t * args,
67+
config_t * config ) {
68+
test_dedup_topo( config, args->test_dedup.affinity );
69+
fd_topo_t * topo = &config->topo;
70+
71+
configure_stage( &fd_cfg_stage_hugetlbfs, CONFIGURE_CMD_INIT, config );
72+
73+
fdctl_check_configure( config );
74+
initialize_workspaces( config );
75+
initialize_stacks( config );
76+
fd_topo_join_workspaces( topo, FD_SHMEM_JOIN_MODE_READ_WRITE );
77+
78+
/* FIXME allow running sandboxed/multiprocess */
79+
fd_topo_run_single_process( topo, 2, config->uid, config->gid, fdctl_tile_run );
80+
for(;;) pause();
81+
}
82+
83+
action_t fd_action_test_dedup = {
84+
.name = "test-dedup",
85+
.args = test_dedup_cmd_args,
86+
.fn = test_dedup_cmd_fn,
87+
.description = "Test the dedup tile"
88+
};

‎src/app/shared_dev/commands/test_dedup/test_dedup_rx_tile.c

Lines changed: 145 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,145 @@
1+
#include "../../../../disco/topo/fd_topo.h"
2+
3+
struct test_dedup_rx_ctx {
4+
fd_rng_t rng[1];
5+
6+
void * rx_base;
7+
8+
ulong tcache_depth;
9+
ulong tcache_map_cnt;
10+
ulong * _tcache_sync;
11+
ulong tcache_sync;
12+
ulong * _tcache_ring;
13+
ulong * _tcache_map;
14+
15+
ulong diag_iter;
16+
long diag_last_ts;
17+
long diag_interval;
18+
19+
uint corrupt : 1;
20+
};
21+
22+
typedef struct test_dedup_rx_ctx test_dedup_rx_ctx_t;
23+
24+
static void
25+
during_housekeeping( test_dedup_rx_ctx_t * ctx ) {
26+
/* Update synchronization info */
27+
FD_COMPILER_MFENCE();
28+
FD_VOLATILE( *ctx->_tcache_sync ) = ctx->tcache_sync;
29+
FD_COMPILER_MFENCE();
30+
31+
/* Send diagnostic info */
32+
long now = fd_log_wallclock();
33+
long dt = now - ctx->diag_last_ts;
34+
if( FD_UNLIKELY( dt > (long)1e9 ) ) {
35+
float mfps = (1e3f*(float)ctx->diag_iter) / (float)dt;
36+
FD_LOG_NOTICE(( "%7.3f Mfrag/s rx", (double)mfps ));
37+
ctx->diag_last_ts = now;
38+
ctx->diag_iter = 0UL;
39+
}
40+
}
41+
42+
static inline void
43+
during_frag( test_dedup_rx_ctx_t * ctx,
44+
ulong in_idx,
45+
ulong seq,
46+
ulong sig,
47+
ulong chunk,
48+
ulong sz,
49+
ulong ctl ) {
50+
(void)in_idx; (void)seq; (void)ctl;
51+
52+
/* Process the received fragment (FIXME: also validate continuity of
53+
the individual tx streams via sig too, validate control bits, add
54+
latency and bandwidth stats). */
55+
56+
int is_dup;
57+
FD_TCACHE_INSERT( is_dup, ctx->tcache_sync, ctx->_tcache_ring, ctx->tcache_depth, ctx->_tcache_map, ctx->tcache_map_cnt, sig );
58+
if( FD_UNLIKELY( is_dup ) ) FD_LOG_ERR(( "Received a duplicate" ));
59+
60+
uchar const * p = (uchar const *)fd_chunk_to_laddr_const( ctx->rx_base, chunk );
61+
__m256i avx = _mm256_set1_epi64x( (long)sig );
62+
int mask0 = -1;
63+
int mask1 = -1;
64+
int mask2 = -1;
65+
int mask3 = -1;
66+
for( ulong off=0UL; off<sz; off+=128UL ) {
67+
mask0 &= _mm256_movemask_epi8( _mm256_cmpeq_epi8( _mm256_load_si256( (__m256i *) p ), avx ) );
68+
mask1 &= _mm256_movemask_epi8( _mm256_cmpeq_epi8( _mm256_load_si256( (__m256i *)(p+32UL) ), avx ) );
69+
mask2 &= _mm256_movemask_epi8( _mm256_cmpeq_epi8( _mm256_load_si256( (__m256i *)(p+64UL) ), avx ) );
70+
mask3 &= _mm256_movemask_epi8( _mm256_cmpeq_epi8( _mm256_load_si256( (__m256i *)(p+96UL) ), avx ) );
71+
p += 128UL;
72+
}
73+
74+
/* Validate that the frag payload was as expected */
75+
ctx->corrupt = ((mask0 & mask1 & mask2 & mask3)!=-1);
76+
}
77+
78+
static inline void
79+
after_frag( test_dedup_rx_ctx_t * ctx,
80+
ulong in_idx,
81+
ulong in_seq,
82+
ulong in_sig,
83+
ulong in_sz,
84+
ulong in_tsorig,
85+
ulong in_tspub,
86+
fd_stem_context_t * stem ) {
87+
(void)in_idx; (void)in_seq; (void)in_sig; (void)in_sz; (void)in_tsorig; (void)in_tspub; (void)stem;
88+
if( FD_UNLIKELY( ctx->corrupt ) ) FD_LOG_ERR(( "Corrupt payload received" ));
89+
ctx->diag_iter++;
90+
}
91+
92+
#define STEM_BURST 1
93+
#define STEM_LAZY ((long)2e6)
94+
#define STEM_CALLBACK_CONTEXT_ALIGN alignof(test_dedup_rx_ctx_t)
95+
#define STEM_CALLBACK_CONTEXT_TYPE test_dedup_rx_ctx_t
96+
#define STEM_CALLBACK_DURING_HOUSEKEEPING during_housekeeping
97+
#define STEM_CALLBACK_DURING_FRAG during_frag
98+
#define STEM_CALLBACK_AFTER_FRAG after_frag
99+
#include "../../../../disco/stem/fd_stem.c"
100+
101+
static ulong
102+
scratch_align( void ) {
103+
return fd_ulong_max( alignof(test_dedup_rx_ctx_t), fd_tcache_align() );
104+
}
105+
106+
static ulong
107+
scratch_footprint( fd_topo_tile_t const * tile ) {
108+
(void)tile;
109+
return FD_LAYOUT_FINI( FD_LAYOUT_APPEND( FD_LAYOUT_APPEND( FD_LAYOUT_INIT,
110+
alignof(test_dedup_rx_ctx_t), sizeof(test_dedup_rx_ctx_t) ),
111+
fd_tcache_align(), fd_tcache_footprint( tile->test_dedup_rx.tcache_depth, tile->test_dedup_rx.tcache_map_cnt ) ),
112+
scratch_align() );
113+
}
114+
115+
static void
116+
unprivileged_init( fd_topo_t * topo,
117+
fd_topo_tile_t * tile ) {
118+
FD_TEST( tile->in_cnt>0 );
119+
120+
FD_SCRATCH_ALLOC_INIT( l, fd_topo_obj_laddr( topo, tile->tile_obj_id ) );
121+
test_dedup_rx_ctx_t * ctx = FD_SCRATCH_ALLOC_APPEND( l, alignof(test_dedup_rx_ctx_t), sizeof(test_dedup_rx_ctx_t) );
122+
void * tcache_mem = FD_SCRATCH_ALLOC_APPEND( l, fd_tcache_align(), fd_tcache_footprint( tile->test_dedup_rx.tcache_depth, tile->test_dedup_rx.tcache_map_cnt ) );
123+
FD_SCRATCH_ALLOC_FINI( l, scratch_align() );
124+
125+
memset( ctx, 0, sizeof(test_dedup_rx_ctx_t) );
126+
fd_tcache_t * tcache = fd_tcache_join( fd_tcache_new( tcache_mem, tile->test_dedup_rx.tcache_depth, tile->test_dedup_rx.tcache_map_cnt ) );
127+
FD_TEST( tcache );
128+
129+
ctx->tcache_depth = fd_tcache_depth ( tcache );
130+
ctx->tcache_map_cnt = fd_tcache_map_cnt ( tcache );
131+
ctx->_tcache_sync = fd_tcache_oldest_laddr( tcache );
132+
ctx->_tcache_ring = fd_tcache_ring_laddr ( tcache );
133+
ctx->_tcache_map = fd_tcache_map_laddr ( tcache );
134+
ctx->tcache_sync = *ctx->_tcache_sync;
135+
136+
FD_TEST( fd_rng_join( fd_rng_new( ctx->rng, tile->test_dedup_rx.rng_seq, 0UL ) ) );
137+
}
138+
139+
fd_topo_run_tile_t fd_tile_TDupRx = {
140+
.name = "TDupRx",
141+
.scratch_align = scratch_align,
142+
.scratch_footprint = scratch_footprint,
143+
.unprivileged_init = unprivileged_init,
144+
.run = stem_run
145+
};

0 commit comments

Comments
 (0)