Line data Source code
1 : #include "fd_tango.h"
2 :
3 : #if FD_HAS_X86
4 : #include <x86intrin.h>
5 : #endif
6 :
7 : #if FD_HAS_HOSTED && FD_HAS_AVX
8 :
9 : /* FIXME: UPDATE TO TIME BASED HOUSEKEEPING */
10 :
11 : #include <math.h> /* For expm1f */
12 :
13 : FD_STATIC_ASSERT( FD_CHUNK_SZ==64UL, unit_test );
14 :
15 0 : #define RX_MAX (128UL) /* Max _reliable_ (arb unreliable) */
16 :
17 : static uchar fctl_mem[ FD_FCTL_FOOTPRINT( RX_MAX ) ] __attribute__((aligned(FD_FCTL_ALIGN)));
18 : static char * _fseq[ RX_MAX ];
19 :
20 : #define FD_CNC_DIAG_IN_BACKP (0UL)
21 : #define FD_CNC_DIAG_BACKP_CNT (1UL)
22 :
23 : #define FD_FSEQ_DIAG_PUB_CNT (0UL)
24 : #define FD_FSEQ_DIAG_PUB_SZ (1UL)
25 : #define FD_FSEQ_DIAG_FILT_CNT (2UL)
26 : #define FD_FSEQ_DIAG_FILT_SZ (3UL)
27 : #define FD_FSEQ_DIAG_OVRNP_CNT (4UL)
28 : #define FD_FSEQ_DIAG_OVRNR_CNT (5UL)
29 : #define FD_FSEQ_DIAG_SLOW_CNT (6UL)
30 :
31 : int
32 : main( int argc,
33 0 : char ** argv ) {
34 0 : fd_boot( &argc, &argv );
35 :
36 0 : char const * _cnc = fd_env_strip_cmdline_cstr ( &argc, &argv, "--cnc", NULL, NULL ); /* (req) cnc */
37 0 : char const * _mcache = fd_env_strip_cmdline_cstr ( &argc, &argv, "--mcache", NULL, NULL ); /* (req) mcache */
38 0 : char const * _dcache = fd_env_strip_cmdline_cstr ( &argc, &argv, "--dcache", NULL, NULL ); /* (req) dcache */
39 0 : char const * _fseqs = fd_env_strip_cmdline_cstr ( &argc, &argv, "--fseqs", NULL, "" ); /* (opt) rx fseqs */
40 0 : char const * _init = fd_env_strip_cmdline_cstr ( &argc, &argv, "--init", NULL, NULL ); /* (opt) init seq */
41 0 : ulong tx_idx = fd_env_strip_cmdline_ulong( &argc, &argv, "--tx-idx", NULL, 0UL ); /* (opt) origin */
42 0 : uint seed = fd_env_strip_cmdline_uint ( &argc, &argv, "--seed", NULL, (uint)fd_tickcount() ); /* (opt) rng seed */
43 0 : int lazy = fd_env_strip_cmdline_int ( &argc, &argv, "--lazy", NULL, 7 ); /* (opt) lazyiness */
44 :
45 0 : if( FD_UNLIKELY( !_cnc ) ) FD_LOG_ERR(( "--cnc not specified" ));
46 0 : if( FD_UNLIKELY( !_mcache ) ) FD_LOG_ERR(( "--mcache not specified" ));
47 0 : if( FD_UNLIKELY( !_dcache ) ) FD_LOG_ERR(( "--dcache not specified" ));
48 0 : if( FD_UNLIKELY( tx_idx>=FD_FRAG_META_ORIG_MAX ) ) FD_LOG_ERR(( "--tx-idx too large" ));
49 :
50 0 : ulong rx_cnt = fd_cstr_tokenize( _fseq, RX_MAX, (char *)_fseqs, ',' ); /* Note: argv isn't const to okay to cast away const */
51 0 : if( FD_UNLIKELY( rx_cnt>RX_MAX ) ) FD_LOG_ERR(( "--rx-cnt too large for this unit-test" ));
52 :
53 : /* burst_avg is the average number of synthetic payload bytes in a
54 : packet burst (0 min, exponentially distributed). A burst's data
55 : bytes are sent as zero or more packets containing pkt_payload_max
56 : bytes followed by a one packet containing [1,pkt_payload_max]
57 : bytes. All the packets in a burst are sent in rapid succession.
58 : Packets have a constant number of bytes overhead for various
59 : headers and footers (pkt_framing).
60 :
61 : E.g. pkt_framing = 70, pkt_payload_max = 1472 would describe an
62 : Ethernet link that runs UDP / IP4 / VLAN tagged non-Jumbo Ethernet
63 : packets. pkt_framing = 70 comes from from 8 byte preamble + 14
64 : byte Ethernet header + 4 byte VLAN tag + 20 byte IP4 header
65 : (without options) + 8 byte UDP header + 4 byte frame-check-sequence
66 : + 12 byte interframe gap, 1472 comes from a 1500 byte MTU (maximum
67 : Ethernet payload size) - 20 byte IP4 header (without options) - 8
68 : byte UDP header. */
69 :
70 0 : # define RANDOMIZE_BURSTS 0
71 0 : float burst_avg = fd_env_strip_cmdline_float( &argc, &argv, "--burst-avg", NULL, 1472.f );
72 0 : ulong pkt_payload_max = fd_env_strip_cmdline_ulong( &argc, &argv, "--pkt-payload-max", NULL, 1472UL );
73 0 : ulong pkt_framing = fd_env_strip_cmdline_ulong( &argc, &argv, "--pkt-framing", NULL, 70UL );
74 :
75 0 : FD_LOG_NOTICE(( "Configuring synthetic load (--burst-avg %g B --pkt-framing %lu B --pkt-payload-max %lu B,",
76 0 : (double)burst_avg, pkt_framing, pkt_payload_max ));
77 :
78 : /* 2.1e17 ~ LONG_MAX / 43.668 is such that an exponentially
79 : distributed random from the util/rng API with an average in
80 : (0,2.1e17) will practically never overflow when rounded to a long. */
81 :
82 0 : if( FD_UNLIKELY( !((0.f<burst_avg) & (burst_avg<2.1e17f)) ) ) FD_LOG_ERR(( "--burst-avg out of range" ));
83 :
84 0 : if( FD_UNLIKELY( !pkt_payload_max ) ) FD_LOG_ERR(( "Zero --pkt-payload-max" ));
85 :
86 0 : ulong pkt_max = pkt_framing + pkt_payload_max;
87 0 : if( FD_UNLIKELY( (pkt_max<pkt_framing) | (pkt_max>(ulong)USHORT_MAX) ) )
88 0 : FD_LOG_ERR(( "Too large --pkt-framing + --pkt-payload-max" ));
89 :
90 0 : FD_LOG_NOTICE(( "Creating rng --seed %u", seed ));
91 :
92 0 : fd_rng_t _rng[1]; fd_rng_t * rng = fd_rng_join( fd_rng_new( _rng, seed, 0UL ) );
93 :
94 0 : FD_LOG_NOTICE(( "Joining to --cnc %s", _cnc ));
95 :
96 0 : fd_cnc_t * cnc = fd_cnc_join( fd_wksp_map( _cnc ) );
97 0 : if( FD_UNLIKELY( !cnc ) ) FD_LOG_ERR(( "join failed" ));
98 0 : if( FD_UNLIKELY( fd_cnc_app_sz( cnc )<16UL ) ) FD_LOG_ERR(( "cnc app sz must be at least 16" ));
99 :
100 0 : ulong * cnc_diag = (ulong *)fd_cnc_app_laddr( cnc );
101 0 : int in_backp = 1;
102 0 : FD_VOLATILE( cnc_diag[ FD_CNC_DIAG_IN_BACKP ] ) = 1UL;
103 0 : FD_VOLATILE( cnc_diag[ FD_CNC_DIAG_BACKP_CNT ] ) = 0UL;
104 :
105 0 : FD_LOG_NOTICE(( "Joining to --mcache %s", _mcache ));
106 :
107 0 : fd_frag_meta_t * mcache = fd_mcache_join( fd_wksp_map( _mcache ) );
108 0 : if( FD_UNLIKELY( !mcache ) ) FD_LOG_ERR(( "join failed" ));
109 :
110 0 : ulong depth = fd_mcache_depth ( mcache );
111 0 : ulong * sync = fd_mcache_seq_laddr( mcache );
112 :
113 0 : ulong seq = _init ? fd_cstr_to_ulong( _init ) : fd_mcache_seq_query( sync );
114 :
115 0 : FD_LOG_NOTICE(( "Joining to --dcache %s", _dcache ));
116 :
117 0 : uchar * dcache = fd_dcache_join( fd_wksp_map( _dcache ) );
118 0 : if( FD_UNLIKELY( !dcache ) ) FD_LOG_ERR(( "join failed" ));
119 :
120 0 : void * base = fd_wksp_containing( dcache );
121 0 : if( FD_UNLIKELY( !base ) ) FD_LOG_ERR(( "fd_wksp_containing failed" ));
122 :
123 0 : if( FD_UNLIKELY( !fd_dcache_compact_is_safe( base, dcache, pkt_max, depth ) ) )
124 0 : FD_LOG_ERR(( "--dcache not compatible with wksp base, --pkt-framing, --pkt-payload-max and --mcache depth" ));
125 :
126 0 : ulong chunk0 = fd_dcache_compact_chunk0( base, dcache );
127 0 : ulong wmark = fd_dcache_compact_wmark ( base, dcache, pkt_max );
128 0 : ulong chunk = chunk0; /* FIXME: command line / auto recover this from dcache app region for clean dcache recovery too */
129 :
130 0 : FD_LOG_NOTICE(( "Configuring flow control (rx_cnt %lu)", rx_cnt ));
131 :
132 0 : fd_fctl_t * fctl = fd_fctl_join( fd_fctl_new( fctl_mem, rx_cnt ) );
133 0 : if( FD_UNLIKELY( !fctl ) ) FD_LOG_ERR(( "join failed" ));
134 :
135 0 : for( ulong rx_idx=0UL; rx_idx<rx_cnt; rx_idx++ ) {
136 :
137 0 : FD_LOG_NOTICE(( "Joining to reliable rx %lu fseq %s", rx_idx, _fseq[ rx_idx ] ));
138 0 : ulong * fseq = fd_fseq_join( fd_wksp_map( _fseq[ rx_idx ] ) );
139 0 : if( FD_UNLIKELY( !fseq ) ) FD_LOG_ERR(( "join failed" ));
140 0 : ulong * fseq_diag = (ulong *)fd_fseq_app_laddr( fseq );
141 :
142 0 : if( FD_UNLIKELY( !fd_fctl_cfg_rx_add( fctl, depth, fseq, &fseq_diag[ FD_FSEQ_DIAG_SLOW_CNT ] ) ) )
143 0 : FD_LOG_ERR(( "fd_fctl_cfg_rx_add failed" ));
144 :
145 0 : FD_VOLATILE( fseq_diag[ FD_FSEQ_DIAG_SLOW_CNT ] ) = 0UL;
146 0 : }
147 :
148 : /* cr_burst is 1 because we only send at most 1 fragment metadata
149 : between checking cr_avail. We use defaults for cr_max, cr_resume
150 : and cr_refill. */
151 0 : if( FD_UNLIKELY( !fd_fctl_cfg_done( fctl, 1UL, 0UL, 0UL, 0UL ) ) ) FD_LOG_ERR(( "fd_fctl_cfg_done failed" ));
152 0 : FD_LOG_NOTICE(( "cr_burst %lu cr_max %lu cr_resume %lu cr_refill %lu",
153 0 : fd_fctl_cr_burst( fctl ), fd_fctl_cr_max( fctl ), fd_fctl_cr_resume( fctl ), fd_fctl_cr_refill( fctl ) ));
154 :
155 0 : ulong cr_avail = 0UL;
156 :
157 0 : FD_LOG_NOTICE(( "Running --tx-idx %lu --init %lu (%s) --lazy %i", tx_idx, seq, _init ? "manual" : "auto", lazy ));
158 :
159 0 : ulong async_min = 1UL << lazy;
160 0 : ulong async_rem = 1UL; /* Do housekeeping on the first iteration */
161 :
162 0 : long then = fd_log_wallclock();
163 0 : ulong iter = 0UL;
164 :
165 0 : int ctl_som = 1;
166 :
167 0 : ulong burst_rem;
168 : # if RANDOMIZE_BURSTS
169 : do { burst_rem = (ulong)(long)(0.5f + burst_avg*fd_rng_float_exp( rng )); } while( FD_UNLIKELY( !burst_rem ) );
170 : # else
171 0 : ulong _burst_avg = (ulong)(ulong)burst_avg;
172 0 : burst_rem = _burst_avg;
173 0 : # endif
174 :
175 0 : fd_cnc_signal( cnc, FD_CNC_SIGNAL_RUN );
176 0 : for(;;) {
177 :
178 : /* Do housekeeping in the background */
179 0 : if( FD_UNLIKELY( !async_rem ) ) {
180 :
181 : /* Send synchronization info */
182 0 : fd_mcache_seq_update( sync, seq );
183 :
184 : /* Send diagnostic info */
185 0 : long now = fd_log_wallclock();
186 0 : fd_cnc_heartbeat( cnc, now );
187 :
188 0 : long dt = now - then;
189 0 : if( FD_UNLIKELY( dt > (long)1e9 ) ) {
190 0 : float mfps = (1e3f*(float)iter) / (float)dt;
191 0 : FD_LOG_NOTICE(( "%7.3f Mfrag/s tx (in_backp %lu backp_cnt %lu)", (double)mfps,
192 0 : FD_VOLATILE_CONST( cnc_diag[ FD_CNC_DIAG_IN_BACKP ] ),
193 0 : FD_VOLATILE_CONST( cnc_diag[ FD_CNC_DIAG_BACKP_CNT ] ) ));
194 0 : for( ulong rx_idx=0UL; rx_idx<rx_cnt; rx_idx++ ) {
195 0 : ulong * slow = fd_fctl_rx_slow_laddr( fctl, rx_idx );
196 0 : FD_LOG_NOTICE(( "slow%lu %lu", rx_idx, FD_VOLATILE_CONST( *slow ) ));
197 0 : FD_VOLATILE( *slow ) = 0UL;
198 0 : }
199 0 : FD_VOLATILE( cnc_diag[ FD_CNC_DIAG_BACKP_CNT ] ) = 0UL;
200 0 : then = now;
201 0 : iter = 0UL;
202 0 : }
203 :
204 : /* Receive command-and-control signals */
205 0 : ulong s = fd_cnc_signal_query( cnc );
206 0 : if( FD_UNLIKELY( s!=FD_CNC_SIGNAL_RUN ) ) {
207 0 : if( FD_LIKELY( s==FD_CNC_SIGNAL_HALT ) ) break;
208 0 : char buf[ FD_CNC_SIGNAL_CSTR_BUF_MAX ];
209 0 : FD_LOG_WARNING(( "Unexpected signal %s (%lu) received; trying to resume", fd_cnc_signal_cstr( s, buf ), s ));
210 0 : fd_cnc_signal( cnc, FD_CNC_SIGNAL_RUN );
211 0 : }
212 :
213 : /* Receive flow control credits */
214 0 : cr_avail = fd_fctl_tx_cr_update( fctl, cr_avail, seq );
215 0 : if( FD_UNLIKELY( in_backp ) ) {
216 0 : if( FD_LIKELY( cr_avail ) ) {
217 0 : FD_VOLATILE( cnc_diag[ FD_CNC_DIAG_IN_BACKP ] ) = 0UL;
218 0 : in_backp = 0;
219 0 : }
220 0 : }
221 :
222 : /* Reload housekeeping timer */
223 0 : async_rem = fd_tempo_async_reload( rng, async_min );
224 0 : }
225 0 : async_rem--;
226 :
227 : /* Check if we are backpressured */
228 0 : if( FD_UNLIKELY( !cr_avail ) ) {
229 0 : if( FD_UNLIKELY( !in_backp ) ) {
230 0 : FD_VOLATILE( cnc_diag[ FD_CNC_DIAG_IN_BACKP ] ) = 0UL;
231 0 : FD_VOLATILE( cnc_diag[ FD_CNC_DIAG_BACKP_CNT ] ) = FD_VOLATILE_CONST( cnc_diag[ FD_CNC_DIAG_BACKP_CNT ] ) + 1UL;
232 0 : in_backp = 1;
233 0 : }
234 0 : FD_SPIN_PAUSE();
235 0 : continue;
236 0 : }
237 :
238 : /* We are in the process of "receiving" a fragment from the NIC.
239 : Compute the details of the synthetic fragment and fill the data
240 : region with a suitable test pattern as fast as we can. Note that
241 : the dcache region pointed by chunk through the double cache line
242 : region marked by sz is currently not marked as visible and has
243 : room for up to "MTU" worth of double chunks free. So we can
244 : write it at our hearts content. Since we align frag starts to
245 : double chunk boundaries, we are safe to tail clobber (which
246 : further speeds up test pattern generation as this amortizes loop
247 : overhead, flattens tail branches and only write to the dcache in
248 : complete double cache lines). We don't care if this is written
249 : atomic or not. */
250 :
251 0 : ulong frag_sz = fd_ulong_min( burst_rem, pkt_payload_max );
252 0 : burst_rem -= frag_sz;
253 0 : int ctl_eom = !burst_rem;
254 0 : int ctl_err = 0;
255 :
256 0 : ulong sig = seq; /* Test pattern */
257 : /*ulong chunk = ... already at location where next packet will be written ...; */
258 0 : ulong sz = pkt_framing + frag_sz;
259 0 : ulong ctl = fd_frag_meta_ctl( tx_idx, ctl_som, ctl_eom, ctl_err );
260 0 : ulong tsorig = 0UL;
261 0 : ulong tspub = 0UL;
262 :
263 0 : uchar * p = (uchar *)fd_chunk_to_laddr( base, chunk );
264 0 : __m256i avx = _mm256_set1_epi64x( (long)seq );
265 0 : for( ulong off=0UL; off<sz; off+=128UL ) {
266 0 : _mm256_store_si256( (__m256i *)(p ), avx );
267 0 : _mm256_store_si256( (__m256i *)(p+32UL), avx );
268 0 : _mm256_store_si256( (__m256i *)(p+64UL), avx );
269 0 : _mm256_store_si256( (__m256i *)(p+96UL), avx );
270 0 : p += 128UL;
271 0 : }
272 :
273 : /* We just "finished receiving" the next fragment of the burst from
274 : the "NIC". Publish to consumers as frag seq. This implicitly
275 : unpublishes frag seq-depth (cyclic) at the same time. */
276 :
277 0 : # define PUBLISH_STYLE 0
278 :
279 0 : # if PUBLISH_STYLE==0 /* Incompatible with WAIT_STYLE==2 */
280 :
281 0 : fd_mcache_publish( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
282 :
283 : # elif PUBLISH_STYLE==1 /* Incompatible with WAIT_STYLE==2 */
284 :
285 : fd_mcache_publish_sse( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
286 :
287 : # else /* Compatible with all wait styles, requires target with atomic
288 : aligned AVX load/store support */
289 :
290 : fd_mcache_publish_avx( mcache, depth, seq, sig, chunk, sz, ctl, tsorig, tspub );
291 :
292 : # endif
293 :
294 : /* Wind up for the next iteration */
295 :
296 0 : chunk = fd_dcache_compact_next( chunk, sz, chunk0, wmark );
297 0 : seq = fd_seq_inc( seq, 1UL );
298 0 : cr_avail--;
299 0 : iter++;
300 : # if RANDOMIZE_BURSTS
301 : if( FD_UNLIKELY( ctl_eom ) )
302 : do burst_rem = (ulong)(long)(0.5f + burst_avg*fd_rng_float_exp( rng )); while( FD_UNLIKELY( !burst_rem ) );
303 : # else
304 0 : burst_rem = fd_ulong_if( ctl_eom, _burst_avg, burst_rem );
305 0 : # endif
306 0 : ctl_som = ctl_eom;
307 0 : }
308 :
309 0 : FD_LOG_NOTICE(( "Cleaning up" ));
310 :
311 0 : while( rx_cnt ) fd_wksp_unmap( fd_fctl_rx_seq_laddr( fctl, --rx_cnt ) );
312 0 : fd_fctl_delete( fd_fctl_leave( fctl ) );
313 0 : fd_wksp_unmap( fd_dcache_leave( dcache ) );
314 0 : fd_wksp_unmap( fd_mcache_leave( mcache ) );
315 0 : fd_cnc_signal( cnc, FD_CNC_SIGNAL_BOOT );
316 0 : fd_wksp_unmap( fd_cnc_leave( cnc ) );
317 0 : fd_rng_delete( fd_rng_leave( rng ) );
318 :
319 0 : FD_LOG_NOTICE(( "pass" ));
320 0 : fd_halt();
321 0 : return 0;
322 0 : }
323 :
324 : #else
325 :
326 : int
327 : main( int argc,
328 : char ** argv ) {
329 : fd_boot( &argc, &argv );
330 : FD_LOG_WARNING(( "skip: unit test requires FD_HAS_HOSTED and FD_HAS_AVX capabilities" ));
331 : fd_halt();
332 : return 0;
333 : }
334 :
335 : #endif
|
