diff --git a/src/app/firedancer/config/default.toml b/src/app/firedancer/config/default.toml index 32c2bac2711..1dcac330185 100644 --- a/src/app/firedancer/config/default.toml +++ b/src/app/firedancer/config/default.toml @@ -2267,6 +2267,15 @@ telemetry = true # development purposes. report_runtime_diffs = false + # Maximum upload speed for telemetry data in megabits per second (Mbps). + # The steady state of collection sits well under the default here, and + # it is set higher to allow draining a backlog of data in the event of + # server outage. + # + # It is not recommended to change this value except in test or + # development environments. + max_upload_rate_mbps = 32 + # The following options relate to the 'firedance-dev pktgen' tool # only. pktgen tests net tile transmit throughput by generating a # flood of non-routable Ethernet frames. Only use 'firedancer-dev diff --git a/src/app/firedancer/topology.c b/src/app/firedancer/topology.c index 1ddd15f83ec..06532f8e284 100644 --- a/src/app/firedancer/topology.c +++ b/src/app/firedancer/topology.c @@ -1378,6 +1378,7 @@ fd_topo_configure_tile( fd_topo_tile_t * tile, fd_cstr_ncpy( tile->event.identity_key_path, config->paths.identity_key, sizeof(tile->event.identity_key_path) ); fd_cstr_ncpy( tile->event.url, config->tiles.event.url, sizeof(tile->event.url) ); + tile->event.tx_rate_mbps = config->development.event.max_upload_rate_mbps; fd_cstr_ncpy( tile->event.action, config->action, sizeof(tile->event.action) ); FD_TEST( fd_cstr_printf_check( tile->event.accounts_path, sizeof(tile->event.accounts_path), NULL, "%s", config->paths.accounts ) ); FD_TEST( fd_cstr_printf_check( tile->event.snapshots_path, sizeof(tile->event.snapshots_path), NULL, "%s", config->paths.snapshots ) ); diff --git a/src/app/shared/fd_config.c b/src/app/shared/fd_config.c index 12d3eaad4c6..faff3608520 100644 --- a/src/app/shared/fd_config.c +++ b/src/app/shared/fd_config.c @@ -549,6 +549,8 @@ fd_config_validate( fd_config_t const * config ) { FD_LOG_ERR(( "invalid [development.bench.max_cost_per_block]: must be 0 or in [%lu,%u)", FD_PACK_MAX_COST_PER_BLOCK_UPPER_BOUND, UINT_MAX )); if( FD_UNLIKELY( bench_shreds && (bench_shreds%FD_SHRED_BLK_MAX || bench_shreds>FD_SHRED_BLK_MAX_RAISED) ) ) FD_LOG_ERR(( "invalid [development.bench.max_shreds_per_block]: must be 0 or a multiple of %lu up to %lu", (ulong)FD_SHRED_BLK_MAX, FD_SHRED_BLK_MAX_RAISED )); + if( FD_UNLIKELY( config->development.event.max_upload_rate_mbps>(ulong)LONG_MAX/125000UL ) ) + FD_LOG_ERR(( "invalid [development.event.max_upload_rate_mbps]: must be at most %lu", (ulong)LONG_MAX/125000UL )); if( 0==strcmp( config->net.provider, "xdp" ) ) { if( 0!=strcmp( config->net.xdp.xdp_mode, "skb" ) && diff --git a/src/app/shared/fd_config.h b/src/app/shared/fd_config.h index 86a4126188f..7ae79868f13 100644 --- a/src/app/shared/fd_config.h +++ b/src/app/shared/fd_config.h @@ -322,9 +322,10 @@ struct fd_config { } bundle; struct { - int report_shreds; - int report_transactions; - int report_runtime_diffs; + int report_shreds; + int report_transactions; + int report_runtime_diffs; + ulong max_upload_rate_mbps; } event; struct { diff --git a/src/app/shared/fd_config_json.c b/src/app/shared/fd_config_json.c index 56129762d1a..22662be5b77 100644 --- a/src/app/shared/fd_config_json.c +++ b/src/app/shared/fd_config_json.c @@ -12,7 +12,7 @@ or knowingly skipped) before the constant is bumped. String keys of the user's own file are separately forced through the classification lists below. */ -FD_STATIC_ASSERT( sizeof(fd_config_t)==24458368UL, update_fd_config_to_json_for_the_layout_change ); +FD_STATIC_ASSERT( sizeof(fd_config_t)==24458384UL, update_fd_config_to_json_for_the_layout_change ); #define REDACTED "[redacted]" @@ -505,7 +505,8 @@ fd_config_to_json( fd_config_t const * config, jw_obj_open( &w, "event" ); jw_bool( &w, "report_shreds", config->development.event.report_shreds ); jw_bool( &w, "report_transactions", config->development.event.report_transactions ); - jw_bool( &w, "report_runtime_diffs", config->development.event.report_runtime_diffs ); + jw_bool ( &w, "report_runtime_diffs", config->development.event.report_runtime_diffs ); + jw_ulong( &w, "max_upload_rate_mbps", config->development.event.max_upload_rate_mbps ); jw_obj_close( &w ); jw_obj_open( &w, "pktgen" ); jw_str( &w, "affinity", config->development.pktgen.affinity ); diff --git a/src/app/shared/fd_config_parse.c b/src/app/shared/fd_config_parse.c index ad342d3494a..c9a0447f941 100644 --- a/src/app/shared/fd_config_parse.c +++ b/src/app/shared/fd_config_parse.c @@ -278,6 +278,7 @@ fd_config_extract_pod( uchar * pod, CFG_POP ( bool, development.event.report_shreds ); CFG_POP ( bool, development.event.report_transactions ); CFG_POP ( bool, development.event.report_runtime_diffs ); + CFG_POP ( ulong, development.event.max_upload_rate_mbps ); CFG_POP ( cstr, development.pktgen.affinity ); CFG_POP ( cstr, development.pktgen.fake_dst_ip ); diff --git a/src/disco/events/fd_event_client.c b/src/disco/events/fd_event_client.c index b483d03fed7..1e719ba0acb 100644 --- a/src/disco/events/fd_event_client.c +++ b/src/disco/events/fd_event_client.c @@ -41,9 +41,9 @@ #define FD_EVENT_CLIENT_HEARTBEAT_NANOS (15L*(long)1e9) #define FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS (60L*(long)1e9) -#define FD_EVENT_CLIENT_TX_RATE_BPS (4L*1000L*1000L) #define FD_EVENT_CLIENT_TX_BURST (256L<<10) #define FD_EVENT_CLIENT_CREDIT_STALL_NANOS (20L*(long)1e9) +#define FD_EVENT_CLIENT_TX_DRAIN_MAX (1024UL) /* The edge proxy fails a request after 100 MB of body. Rotate to a new StreamEvents request on the same connection before that. The server @@ -102,6 +102,7 @@ struct fd_event_client { long tx_tokens; long tx_tokens_ns; + long tx_byte_rate; long stall_since; ulong stall_rem; ulong stall_events_sent; @@ -240,7 +241,8 @@ fd_event_client_new( void * shmem, ulong machine_id, ulong buf_max, int use_tls, - fd_x509_ca_store_t const * ca_store ) { + fd_x509_ca_store_t const * ca_store, + ulong tx_rate_mbps ) { if( FD_UNLIKELY( !shmem ) ) { FD_LOG_WARNING(( "NULL shmem" )); return NULL; @@ -319,6 +321,7 @@ fd_event_client_new( void * shmem, client->now = 0L; client->last_stream_send_ns = 0L; client->last_response_ns = 0L; + client->tx_byte_rate = (long)( ( tx_rate_mbps ? tx_rate_mbps : FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT )*125000UL ); client->tx_tokens = FD_EVENT_CLIENT_TX_BURST; client->tx_tokens_ns = 0L; client->stall_since = 0L; @@ -881,11 +884,13 @@ fd_event_client_grpc_ping_ack( void * app_ctx ) { static long pace_refill( fd_event_client_t * client, long now ) { - long ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS; - long dt = fd_long_max( now-client->tx_tokens_ns, 0L ); - long grant = fd_long_min( dt/ns_per_byte, FD_EVENT_CLIENT_TX_BURST-client->tx_tokens ); + ulong rate = (ulong)client->tx_byte_rate; + long dt = fd_long_max( now-client->tx_tokens_ns, 0L ); + uint128 avail = (uint128)(ulong)dt*rate/1000000000UL; + ulong room = (ulong)( FD_EVENT_CLIENT_TX_BURST-client->tx_tokens ); + long grant = (long)( availtx_tokens += grant; - client->tx_tokens_ns = client->tx_tokens==FD_EVENT_CLIENT_TX_BURST ? now : client->tx_tokens_ns+grant*ns_per_byte; + client->tx_tokens_ns = client->tx_tokens==FD_EVENT_CLIENT_TX_BURST ? now : client->tx_tokens_ns+(long)( ( (uint128)(ulong)grant*1000000000UL+rate-1UL )/rate ); return client->tx_tokens; } @@ -1076,11 +1081,12 @@ fd_event_client_next_deadline( fd_event_client_t const * client, if( FD_UNLIKELY( client->state==FD_EVENT_CLIENT_STATE_AUTHENTICATING ) ) deadline = fd_long_min( deadline, client->auth_deadline ); if( FD_UNLIKELY( client->auth_send_pending && !fd_grpc_client_request_is_blocked( client->grpc_client ) ) ) deadline = now; if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_CONNECTED && client->event_stream ) ) { - deadline = fd_long_min( deadline, fd_long_min( client->last_response_ns +FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS, - client->last_stream_send_ns+FD_EVENT_CLIENT_HEARTBEAT_NANOS ) ); - if( FD_UNLIKELY( client->tx_tokens<=0L && fd_circq_unsent_cnt( client->circq ) ) ) { - deadline = fd_long_min( deadline, client->tx_tokens_ns + (1L-client->tx_tokens)*(long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS ); + long send_deadline = client->last_stream_send_ns+FD_EVENT_CLIENT_HEARTBEAT_NANOS; + if( FD_UNLIKELY( client->tx_tokens<=0L && fd_circq_unsent_cnt( client->circq ) ) ) { /* paced: no heartbeat until tokens return */ + ulong rate = (ulong)client->tx_byte_rate; + send_deadline = client->tx_tokens_ns + (long)( ( (uint128)(ulong)(1L-client->tx_tokens)*1000000000UL+rate-1UL )/rate ); } + deadline = fd_long_min( deadline, fd_long_min( client->last_response_ns+FD_EVENT_CLIENT_RESPONSE_TIMEOUT_NANOS, send_deadline ) ); if( FD_UNLIKELY( client->stall_since ) ) deadline = fd_long_min( deadline, client->stall_since+FD_EVENT_CLIENT_CREDIT_STALL_NANOS ); if( FD_UNLIKELY( client->zbatch_cnt ) ) deadline = fd_long_min( deadline, client->zbatch_ns+FD_EVENT_CLIENT_ZSTD_FLUSH_NANOS ); } @@ -1164,6 +1170,19 @@ poll1( fd_event_client_t * client, } } +static int +tx_flush( fd_event_client_t * client, + long now ) { + if( FD_LIKELY( client->state==FD_EVENT_CLIENT_STATE_DISCONNECTED || !fd_grpc_client_tx_pending( client->grpc_client ) ) ) return 0; + int flush_err = client->use_tls ? fd_grpc_client_tls_flush ( client->grpc_client, client->sockfd ) + : fd_grpc_client_tx_flush_socket( client->grpc_client, client->sockfd ); + if( FD_UNLIKELY( -1==flush_err ) ) { + disconnect( client, now, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 ); + return -1; + } + return 0; +} + void fd_event_client_poll( fd_event_client_t * client, long now, @@ -1171,18 +1190,22 @@ fd_event_client_poll( fd_event_client_t * client, client->now = now; poll1( client, now, charge_busy ); - /* poll1 flushes before tx() enqueues: flush once more so bytes queued - this poll go out now instead of arming EPOLLOUT on a writable - socket (a spurious waker roundtrip per message). */ - if( FD_UNLIKELY( client->state!=FD_EVENT_CLIENT_STATE_DISCONNECTED && fd_grpc_client_tx_pending( client->grpc_client ) ) ) { - int flush_err = client->use_tls ? fd_grpc_client_tls_flush ( client->grpc_client, client->sockfd ) - : fd_grpc_client_tx_flush_socket( client->grpc_client, client->sockfd ); - if( FD_UNLIKELY( -1==flush_err ) ) { - disconnect( client, now, DISCONNECT_REASON_TRANSPORT_FAILED, errno, 1 ); - return; + for( ulong i=0UL; istate!=FD_EVENT_CLIENT_STATE_CONNECTED ) ) break; + if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) { + if( FD_UNLIKELY( tx_flush( client, now ) ) ) return; + if( FD_UNLIKELY( fd_grpc_client_stream_send_is_blocked( client->grpc_client ) ) ) break; } + if( FD_LIKELY( !fd_circq_unsent_cnt( client->circq ) ) ) break; + if( FD_UNLIKELY( client->tx_tokens<=0L ) ) break; + int progress = 0; + tx( client, now, &progress ); + if( FD_UNLIKELY( !progress ) ) break; + *charge_busy = 1; } + if( FD_UNLIKELY( tx_flush( client, now ) ) ) return; + /* Arm EPOLLOUT only on write demand: unsent HTTP/2 bytes, or a TLS handshake blocked on write (which is also what an in-progress TCP connect looks like: the handshake write EAGAINed, and connect diff --git a/src/disco/events/fd_event_client.h b/src/disco/events/fd_event_client.h index e78076b0b37..687bbf6faca 100644 --- a/src/disco/events/fd_event_client.h +++ b/src/disco/events/fd_event_client.h @@ -71,7 +71,10 @@ fd_event_client_new( void * shmem, ulong machine_id, ulong buf_max, int use_tls, - fd_x509_ca_store_t const * ca_store ); + fd_x509_ca_store_t const * ca_store, + ulong tx_rate_mbps ); + +#define FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT (32UL) fd_event_client_t * fd_event_client_join( void * shec ); diff --git a/src/disco/events/fd_event_tile.c b/src/disco/events/fd_event_tile.c index 9e991e9c70e..b85064fd19a 100644 --- a/src/disco/events/fd_event_tile.c +++ b/src/disco/events/fd_event_tile.c @@ -648,7 +648,8 @@ unprivileged_init( fd_topo_t const * topo, ctx->machine_id, GRPC_BUF_MAX, ctx->use_tls, - ctx->ca_store ) ); + ctx->ca_store, + tile->event.tx_rate_mbps ) ); FD_TEST( ctx->client ); ctx->next_poll_deadline = 0L; diff --git a/src/disco/events/test_event_client.c b/src/disco/events/test_event_client.c index afa89243281..3a400cb0fb4 100644 --- a/src/disco/events/test_event_client.c +++ b/src/disco/events/test_event_client.c @@ -100,7 +100,7 @@ FD_UNIT_TEST( conn_tls_lifecycle ) { static fd_x509_ca_store_t ca_store[1]; /* empty: the pinned mock cert bypasses chain verification */ fd_event_client_t * client = fd_event_client_join( fd_event_client_new( client_mem, NULL, rng, circq, g_epoll_fd, 1<<20, "https://localhost:1", identity_pubkey, "0.0.0", - "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, 4096UL, 1, ca_store ) ); + "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, 4096UL, 1, ca_store, 0UL ) ); FD_TEST( client ); static fd_tlsrec_conn_t server_conn[1]; @@ -207,7 +207,8 @@ FD_UNIT_TEST( stream_heartbeat ) { 3UL, 4096UL, 0, - NULL ) ); + NULL, + 0UL ) ); FD_TEST( client ); fd_grpc_client_t * grpc = client->grpc_client; @@ -256,15 +257,16 @@ FD_UNIT_TEST( stream_heartbeat ) { } static fd_event_client_t * -test_connected_client( fd_circq_t * circq, - fd_rng_t * rng, - ulong buf_max ) { +test_connected_client_rate( fd_circq_t * circq, + fd_rng_t * rng, + ulong buf_max, + ulong tx_rate_mbps ) { void * client_mem = aligned_alloc( fd_event_client_align(), fd_event_client_footprint( buf_max ) ); FD_TEST( client_mem ); uchar identity_pubkey[32] = {0}; fd_event_client_t * client = fd_event_client_join( fd_event_client_new( client_mem, NULL, rng, circq, g_epoll_fd, 1<<20, "http://localhost:1", identity_pubkey, "0.0.0", - "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, buf_max, 0, NULL ) ); + "0000000000000000000000000000000000000000", "test", 1UL, 2UL, 3UL, buf_max, 0, NULL, tx_rate_mbps ) ); FD_TEST( client ); fd_grpc_client_t * grpc = client->grpc_client; client->state = FD_EVENT_CLIENT_STATE_CONNECTED; @@ -282,6 +284,13 @@ test_connected_client( fd_circq_t * circq, return client; } +static fd_event_client_t * +test_connected_client( fd_circq_t * circq, + fd_rng_t * rng, + ulong buf_max ) { + return test_connected_client_rate( circq, rng, buf_max, 0UL ); +} + /* Queue an event of sz bytes, incompressible unless zero. */ static void test_push( fd_circq_t * circq, @@ -360,7 +369,7 @@ FD_UNIT_TEST( tx_pacing ) { FD_TEST( dl>now && dl<=now+(long)1e9 ); /* Polls shorter than one token's worth of time keep accruing. */ - long const ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS; + long const ns_per_byte = (long)1e9/(long)( FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT*125000UL ); long tokens0 = client->tx_tokens, tokens_ns0 = client->tx_tokens_ns; for( long i=1L; imetrics.events_sent==1UL && !test_drain( grpc ) ); client->last_response_ns = now; - long const ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS; + long const ns_per_byte = (long)1e9/(long)( FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT*125000UL ); long dl = fd_event_client_next_deadline( client, now ); FD_TEST( dl==now+(1L-client->tx_tokens)*ns_per_byte ); @@ -429,6 +438,61 @@ FD_UNIT_TEST( tx_pacing_large_msg ) { fd_rng_delete( fd_rng_leave( rng ) ); } +/* An explicit rate is honored (0 keeps the default), and a paced event + goes out at its deadline even when a byte is not a whole number of + nanoseconds. */ +FD_UNIT_TEST( tx_pacing_rates ) { + static uchar circq_mem[ 4096UL+(1UL<<20) ] __attribute__((aligned(FD_CIRCQ_ALIGN))); + ulong const mbps[ 3 ] = { 0UL, 3UL, 64UL }; + for( ulong i=0UL; i<3UL; i++ ) { + fd_circq_t * circq = fd_circq_join( fd_circq_new( circq_mem, 1UL<<20 ) ); + FD_TEST( circq ); + fd_rng_t rng_mem[1]; + fd_rng_t * rng = fd_rng_join( fd_rng_new( rng_mem, 0U, 1UL ) ); + fd_event_client_t * client = test_connected_client_rate( circq, rng, 1UL<<20, mbps[ i ] ); + long const rate = (long)( ( mbps[ i ] ? mbps[ i ] : FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT )*125000UL ); + FD_TEST( client->tx_byte_rate==rate ); + + test_push( circq, rng, 16UL, 0 ); + long now = fd_log_wallclock(); + client->tx_tokens = -1000L; client->tx_tokens_ns = now; + client->last_response_ns = now; client->last_stream_send_ns = now; + long dl = fd_event_client_next_deadline( client, now ); + FD_TEST( dl==now+( 1001L*(long)1e9+rate-1L )/rate ); + FD_TEST( (ulong)( (uint128)(ulong)( dl-1L-now )*(ulong)rate/1000000000UL )<=1000UL ); + test_poll_tx( client, dl ); + FD_TEST( client->zbatch_cnt==1UL ); + + free( client ); + fd_rng_delete( fd_rng_leave( rng ) ); + } +} + +/* At a low rate a large batch leaves more than a heartbeat interval of + debt. No heartbeat can go out while events are queued, so the + deadline is the refill, not an already expired heartbeat. */ +FD_UNIT_TEST( tx_pacing_debt_past_heartbeat ) { + static uchar circq_mem[ 4096UL+(1UL<<20) ] __attribute__((aligned(FD_CIRCQ_ALIGN))); + fd_circq_t * circq = fd_circq_join( fd_circq_new( circq_mem, 1UL<<20 ) ); + FD_TEST( circq ); + fd_rng_t rng_mem[1]; + fd_rng_t * rng = fd_rng_join( fd_rng_new( rng_mem, 0U, 1UL ) ); + fd_event_client_t * client = test_connected_client_rate( circq, rng, 1UL<<20, 1UL ); + long const rate = 125000L; + + test_push( circq, rng, 16UL, 0 ); + long now = fd_log_wallclock(); + client->tx_tokens = -3000000L; client->tx_tokens_ns = now; /* 24 s of debt */ + client->last_response_ns = now; + client->last_stream_send_ns = now-FD_EVENT_CLIENT_HEARTBEAT_NANOS-1L; + long dl = fd_event_client_next_deadline( client, now ); + FD_TEST( dl==now+( 3000001L*(long)1e9+rate-1L )/rate ); + FD_TEST( dl>now+FD_EVENT_CLIENT_HEARTBEAT_NANOS ); + + free( client ); + fd_rng_delete( fd_rng_leave( rng ) ); +} + /* Zero peer credit with no progress for the stall window reconnects; genuine progress resets the timer. */ FD_UNIT_TEST( credit_stall ) { @@ -466,7 +530,7 @@ FD_UNIT_TEST( credit_stall ) { /* The bucket keeps refilling while parked, so an expired pacing deadline does not pin the tile awake. */ - long const ns_per_byte = (long)1e9/FD_EVENT_CLIENT_TX_RATE_BPS; + long const ns_per_byte = (long)1e9/(long)( FD_EVENT_CLIENT_TX_RATE_MBPS_DEFAULT*125000UL ); test_push( circq, rng, 16UL, 1 ); client->tx_tokens = -1000L; client->tx_tokens_ns = now; long t0 = now+2000L*ns_per_byte; diff --git a/src/disco/topo/fd_topo.h b/src/disco/topo/fd_topo.h index 889b8df1021..4a59be38c35 100644 --- a/src/disco/topo/fd_topo.h +++ b/src/disco/topo/fd_topo.h @@ -347,6 +347,7 @@ struct fd_topo_tile { long boot_timestamp_nanos; ulong launch_cpus[ 1024UL/64UL ]; char launch_cgroup[ PATH_MAX ]; + ulong tx_rate_mbps; } event; struct {