Merge pull request #129 from blockworks-foundation/changes_till_20_04_2023
Revert "increase default tpu size"
This commit is contained in:
commit
972800f57d
|
@ -1,5 +1,5 @@
|
|||
use std::{
|
||||
collections::{HashMap, VecDeque},
|
||||
collections::HashMap,
|
||||
net::{IpAddr, Ipv4Addr, SocketAddr},
|
||||
sync::{
|
||||
atomic::{AtomicBool, AtomicU64, Ordering},
|
||||
|
@ -15,7 +15,10 @@ use quinn::{
|
|||
ClientConfig, Connection, ConnectionError, Endpoint, EndpointConfig, IdleTimeout, SendStream,
|
||||
TokioRuntime, TransportConfig,
|
||||
};
|
||||
use solana_sdk::pubkey::Pubkey;
|
||||
use solana_sdk::{
|
||||
pubkey::Pubkey,
|
||||
quic::{QUIC_MAX_STAKED_RECEIVE_WINDOW_RATIO, QUIC_MIN_STAKED_RECEIVE_WINDOW_RATIO},
|
||||
};
|
||||
use solana_streamer::nonblocking::quic::compute_max_allowed_uni_streams;
|
||||
use tokio::{
|
||||
sync::{broadcast::Receiver, broadcast::Sender, RwLock},
|
||||
|
@ -27,7 +30,6 @@ use super::{rotating_queue::RotatingQueue, tpu_service::IdentityStakes};
|
|||
pub const ALPN_TPU_PROTOCOL_ID: &[u8] = b"solana-tpu";
|
||||
const QUIC_CONNECTION_TIMEOUT_DURATION_IN_SEC: Duration = Duration::from_secs(10);
|
||||
const CONNECTION_RETRY_COUNT: usize = 10;
|
||||
const TRANSACTIONS_SENT_PER_TASK: usize = 5;
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
static ref NB_QUIC_CONNECTIONS: GenericGauge<prometheus::core::AtomicI64> =
|
||||
|
@ -195,14 +197,14 @@ impl ActiveConnection {
|
|||
|
||||
async fn send_transaction_batch(
|
||||
connection: Arc<RwLock<Connection>>,
|
||||
mut txs: VecDeque<Vec<u8>>,
|
||||
txs: Vec<Vec<u8>>,
|
||||
identity: Pubkey,
|
||||
endpoint: Endpoint,
|
||||
socket_addr: SocketAddr,
|
||||
exit_signal: Arc<AtomicBool>,
|
||||
last_stable_id: Arc<AtomicU64>,
|
||||
) {
|
||||
for _ in 0..CONNECTION_RETRY_COUNT {
|
||||
for _ in 0..3 {
|
||||
if exit_signal.load(Ordering::Relaxed) {
|
||||
// return
|
||||
return;
|
||||
|
@ -242,34 +244,24 @@ impl ActiveConnection {
|
|||
}
|
||||
};
|
||||
let mut retry = false;
|
||||
while !txs.is_empty() {
|
||||
let tx = txs.pop_front().unwrap();
|
||||
for tx in &txs {
|
||||
let (stream, retry_conn) =
|
||||
Self::open_unistream(conn.clone(), last_stable_id.clone()).await;
|
||||
if retry_conn {
|
||||
txs.push_back(tx);
|
||||
retry = true;
|
||||
break;
|
||||
}
|
||||
|
||||
if let Some(send_stream) = stream {
|
||||
if exit_signal.load(Ordering::Relaxed) {
|
||||
return;
|
||||
}
|
||||
|
||||
if Self::write_all(
|
||||
retry = Self::write_all(
|
||||
send_stream,
|
||||
&tx,
|
||||
tx,
|
||||
identity,
|
||||
last_stable_id.clone(),
|
||||
conn.stable_id() as u64,
|
||||
)
|
||||
.await
|
||||
{
|
||||
txs.push_back(tx);
|
||||
retry = true;
|
||||
break;
|
||||
}
|
||||
.await;
|
||||
} else {
|
||||
retry = retry_conn;
|
||||
}
|
||||
}
|
||||
if !retry {
|
||||
|
@ -278,6 +270,28 @@ impl ActiveConnection {
|
|||
}
|
||||
}
|
||||
|
||||
// copied from solana code base
|
||||
fn compute_receive_window_ratio_for_staked_node(
|
||||
max_stake: u64,
|
||||
min_stake: u64,
|
||||
stake: u64,
|
||||
) -> u64 {
|
||||
if stake > max_stake {
|
||||
return QUIC_MAX_STAKED_RECEIVE_WINDOW_RATIO;
|
||||
}
|
||||
|
||||
let max_ratio = QUIC_MAX_STAKED_RECEIVE_WINDOW_RATIO;
|
||||
let min_ratio = QUIC_MIN_STAKED_RECEIVE_WINDOW_RATIO;
|
||||
if max_stake > min_stake {
|
||||
let a = (max_ratio - min_ratio) as f64 / (max_stake - min_stake) as f64;
|
||||
let b = max_ratio as f64 - ((max_stake as f64) * a);
|
||||
let ratio = (a * stake as f64) + b;
|
||||
ratio.round() as u64
|
||||
} else {
|
||||
QUIC_MAX_STAKED_RECEIVE_WINDOW_RATIO
|
||||
}
|
||||
}
|
||||
|
||||
async fn listen(
|
||||
transaction_reciever: Receiver<Vec<u8>>,
|
||||
exit_oneshot_channel: tokio::sync::mpsc::Receiver<()>,
|
||||
|
@ -296,7 +310,16 @@ impl ActiveConnection {
|
|||
identity_stakes.stakes,
|
||||
identity_stakes.total_stakes,
|
||||
) as u64;
|
||||
let number_of_transactions_per_unistream = TRANSACTIONS_SENT_PER_TASK;
|
||||
let number_of_transactions_per_unistream = match identity_stakes.peer_type {
|
||||
solana_streamer::nonblocking::quic::ConnectionPeerType::Staked => {
|
||||
Self::compute_receive_window_ratio_for_staked_node(
|
||||
identity_stakes.max_stakes,
|
||||
identity_stakes.min_stakes,
|
||||
identity_stakes.stakes,
|
||||
)
|
||||
}
|
||||
solana_streamer::nonblocking::quic::ConnectionPeerType::Unstaked => 1,
|
||||
};
|
||||
|
||||
let task_counter: Arc<AtomicU64> = Arc::new(AtomicU64::new(0));
|
||||
let mut connection: Option<Arc<RwLock<Connection>>> = None;
|
||||
|
@ -308,7 +331,7 @@ impl ActiveConnection {
|
|||
break;
|
||||
}
|
||||
|
||||
if task_counter.load(Ordering::Relaxed) >= max_uni_stream_connections {
|
||||
if task_counter.load(Ordering::Relaxed) > max_uni_stream_connections {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
|
||||
continue;
|
||||
}
|
||||
|
@ -333,11 +356,10 @@ impl ActiveConnection {
|
|||
}
|
||||
};
|
||||
|
||||
let mut txs = VecDeque::new();
|
||||
txs.push_back(first_tx);
|
||||
let mut txs = vec![first_tx];
|
||||
for _ in 1..number_of_transactions_per_unistream {
|
||||
if let Ok(tx) = transaction_reciever.try_recv() {
|
||||
txs.push_back(tx);
|
||||
txs.push(tx);
|
||||
}
|
||||
}
|
||||
|
||||
|
@ -381,9 +403,7 @@ impl ActiveConnection {
|
|||
};
|
||||
}
|
||||
drop(transaction_reciever);
|
||||
if connection.is_some() {
|
||||
NB_QUIC_CONNECTIONS.dec();
|
||||
}
|
||||
NB_QUIC_CONNECTIONS.dec();
|
||||
NB_QUIC_ACTIVE_CONNECTIONS.dec();
|
||||
}
|
||||
|
||||
|
|
|
@ -35,7 +35,7 @@ const CACHE_NEXT_SLOT_LEADERS_PUBKEY_SIZE: usize = 1024; // Save pubkey and cont
|
|||
const CLUSTERINFO_REFRESH_TIME: u64 = 60; // refresh cluster every minute
|
||||
const LEADER_SCHEDULE_UPDATE_INTERVAL: u64 = 10; // update leader schedule every 10s
|
||||
const AVERAGE_SLOT_CHANGE_TIME_IN_MILLIS: u64 = 400;
|
||||
const MAXIMUM_TRANSACTIONS_IN_QUEUE: usize = 200_000;
|
||||
const MAXIMUM_TRANSACTIONS_IN_QUEUE: usize = 16384;
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
static ref NB_CLUSTER_NODES: GenericGauge<prometheus::core::AtomicI64> =
|
||||
|
|
Loading…
Reference in New Issue