rpc-sts: add config options for stake-weighted qos (#197)
* rpc-sts: plumb options for swqos config * rpc-sts: send to specific tpu peers when configured
This commit is contained in:
parent
0f8f8cd970
commit
f41fb84e15
|
@ -115,6 +115,7 @@ pub struct Config {
|
|||
pub batch_send_rate_ms: u64,
|
||||
/// When the retry pool exceeds this max size, new transactions are dropped after their first broadcast attempt
|
||||
pub retry_pool_max_size: usize,
|
||||
pub tpu_peers: Option<Vec<SocketAddr>>,
|
||||
}
|
||||
|
||||
impl Default for Config {
|
||||
|
@ -127,6 +128,7 @@ impl Default for Config {
|
|||
batch_size: DEFAULT_TRANSACTION_BATCH_SIZE,
|
||||
batch_send_rate_ms: DEFAULT_BATCH_SEND_RATE_MS,
|
||||
retry_pool_max_size: MAX_TRANSACTION_RETRY_POOL_SIZE,
|
||||
tpu_peers: None,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
@ -566,12 +568,18 @@ impl SendTransactionService {
|
|||
stats: &SendTransactionServiceStats,
|
||||
) {
|
||||
// Processing the transactions in batch
|
||||
let addresses = Self::get_tpu_addresses_with_slots(
|
||||
let mut addresses = config
|
||||
.tpu_peers
|
||||
.as_ref()
|
||||
.map(|addrs| addrs.iter().map(|a| (a, 0)).collect::<Vec<_>>())
|
||||
.unwrap_or_default();
|
||||
let leader_addresses = Self::get_tpu_addresses_with_slots(
|
||||
tpu_address,
|
||||
leader_info,
|
||||
config,
|
||||
connection_cache.protocol(),
|
||||
);
|
||||
addresses.extend(leader_addresses);
|
||||
|
||||
let wire_transactions = transactions
|
||||
.iter()
|
||||
|
@ -584,8 +592,8 @@ impl SendTransactionService {
|
|||
})
|
||||
.collect::<Vec<&[u8]>>();
|
||||
|
||||
for address in &addresses {
|
||||
Self::send_transactions(address.0, &wire_transactions, connection_cache, stats);
|
||||
for (address, _) in &addresses {
|
||||
Self::send_transactions(address, &wire_transactions, connection_cache, stats);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -702,14 +710,20 @@ impl SendTransactionService {
|
|||
|
||||
let iter = wire_transactions.chunks(config.batch_size);
|
||||
for chunk in iter {
|
||||
let mut addresses = config
|
||||
.tpu_peers
|
||||
.as_ref()
|
||||
.map(|addrs| addrs.iter().collect::<Vec<_>>())
|
||||
.unwrap_or_default();
|
||||
let mut leader_info_provider = leader_info_provider.lock().unwrap();
|
||||
let leader_info = leader_info_provider.get_leader_info();
|
||||
let addresses = Self::get_tpu_addresses(
|
||||
let leader_addresses = Self::get_tpu_addresses(
|
||||
tpu_address,
|
||||
leader_info,
|
||||
config,
|
||||
connection_cache.protocol(),
|
||||
);
|
||||
addresses.extend(leader_addresses);
|
||||
|
||||
for address in &addresses {
|
||||
Self::send_transactions(address, chunk, connection_cache, stats);
|
||||
|
|
|
@ -1163,6 +1163,22 @@ pub fn app<'a>(version: &'a str, default_args: &'a DefaultArgs) -> App<'a, 'a> {
|
|||
.default_value(&default_args.rpc_send_transaction_retry_pool_max_size)
|
||||
.help("The maximum size of transactions retry pool."),
|
||||
)
|
||||
.arg(
|
||||
Arg::with_name("rpc_send_transaction_tpu_peer")
|
||||
.long("rpc-send-transaction-tpu-peer")
|
||||
.takes_value(true)
|
||||
.number_of_values(1)
|
||||
.multiple(true)
|
||||
.value_name("HOST:PORT")
|
||||
.validator(solana_net_utils::is_host_port)
|
||||
.help("Peer(s) to broadcast transactions to instead of the current leader")
|
||||
)
|
||||
.arg(
|
||||
Arg::with_name("rpc_send_transaction_also_leader")
|
||||
.long("rpc-send-transaction-also-leader")
|
||||
.requires("rpc_send_transaction_tpu_peer")
|
||||
.help("With `--rpc-send-transaction-tpu-peer HOST:PORT`, also send to the current leader")
|
||||
)
|
||||
.arg(
|
||||
Arg::with_name("rpc_scan_and_fix_roots")
|
||||
.long("rpc-scan-and-fix-roots")
|
||||
|
|
|
@ -1308,6 +1308,27 @@ pub fn main() {
|
|||
);
|
||||
exit(1);
|
||||
}
|
||||
let rpc_send_transaction_tpu_peers = matches
|
||||
.values_of("rpc_send_transaction_tpu_peer")
|
||||
.map(|values| {
|
||||
values
|
||||
.map(solana_net_utils::parse_host_port)
|
||||
.collect::<Result<Vec<SocketAddr>, String>>()
|
||||
})
|
||||
.transpose()
|
||||
.unwrap_or_else(|e| {
|
||||
eprintln!("failed to parse rpc send-transaction-service tpu peer address: {e}");
|
||||
exit(1);
|
||||
});
|
||||
let rpc_send_transaction_also_leader = matches.is_present("rpc_send_transaction_also_leader");
|
||||
let leader_forward_count =
|
||||
if rpc_send_transaction_tpu_peers.is_some() && !rpc_send_transaction_also_leader {
|
||||
// rpc-sts is configured to send only to specific tpu peers. disable leader forwards
|
||||
0
|
||||
} else {
|
||||
value_t_or_exit!(matches, "rpc_send_transaction_leader_forward_count", u64)
|
||||
};
|
||||
|
||||
let full_api = matches.is_present("full_rpc_api");
|
||||
|
||||
let mut validator_config = ValidatorConfig {
|
||||
|
@ -1399,11 +1420,7 @@ pub fn main() {
|
|||
contact_debug_interval,
|
||||
send_transaction_service_config: send_transaction_service::Config {
|
||||
retry_rate_ms: rpc_send_retry_rate_ms,
|
||||
leader_forward_count: value_t_or_exit!(
|
||||
matches,
|
||||
"rpc_send_transaction_leader_forward_count",
|
||||
u64
|
||||
),
|
||||
leader_forward_count,
|
||||
default_max_retries: value_t!(
|
||||
matches,
|
||||
"rpc_send_transaction_default_max_retries",
|
||||
|
@ -1422,6 +1439,7 @@ pub fn main() {
|
|||
"rpc_send_transaction_retry_pool_max_size",
|
||||
usize
|
||||
),
|
||||
tpu_peers: rpc_send_transaction_tpu_peers,
|
||||
},
|
||||
no_poh_speed_test: matches.is_present("no_poh_speed_test"),
|
||||
no_os_memory_stats_reporting: matches.is_present("no_os_memory_stats_reporting"),
|
||||
|
|
Loading…
Reference in New Issue