blockworks-foundation
/
lite-rpc
mirror of https://github.com/blockworks-foundation/lite-rpc.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

cargo fix

This commit is contained in:
GroovieGermanikus 2023-07-29 11:26:21 +02:00
parent a27b146ec9
commit 20c2710202
18 changed files with 158 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
core/src/proxy_request_format.rs
View File

@ -1,6 +1,6 @@
use std::fmt;
use std::fmt::Display;
use std::net::{SocketAddr, SocketAddrV4};
use std::net::{SocketAddr};
use anyhow::Context;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;

2
core/src/solana_utils.rs
View File

@ -14,7 +14,7 @@ use std::{
};
use serde::Serialize;
use solana_sdk::hash::Hash;
use solana_sdk::message::{LegacyMessage, v0};
use solana_sdk::signature::Signature;
use solana_sdk::transaction::{Transaction, uses_durable_nonce, VersionedTransaction};
use tokio::sync::mpsc::UnboundedReceiver;

2
lite-rpc/src/bridge.rs
View File

@ -38,7 +38,7 @@ use solana_sdk::{
};
use solana_transaction_status::TransactionStatus;
use std::{ops::Deref, str::FromStr, sync::Arc, time::Duration};
use std::convert::identity;
use tokio::{
net::ToSocketAddrs,
sync::mpsc::{self, Sender},

40
quic-forward-proxy-integration-test/tests/quic_proxy_tpu_integrationtest.rs
View File

@ -1,9 +1,9 @@
use anyhow::bail;
use countmap::CountMap;
use crossbeam_channel::{Receiver, RecvError, RecvTimeoutError, Sender};
use futures::future::join_all;
use log::{debug, error, info, trace, warn};
use quinn::TokioRuntime;
use crossbeam_channel::{Sender};
use log::{debug, info, trace, warn};
use solana_lite_rpc_core::quic_connection_utils::QuicConnectionParameters;
use solana_lite_rpc_core::structures::identity_stakes::IdentityStakes;
use solana_lite_rpc_core::tx_store::empty_tx_store;
@ -14,7 +14,7 @@ use solana_sdk::instruction::Instruction;
use solana_sdk::message::Message;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, Signature, Signer};
use solana_sdk::signer::keypair;
use solana_sdk::transaction::{Transaction, VersionedTransaction};
use solana_streamer::nonblocking::quic::ConnectionPeerType;
use solana_streamer::packet::PacketBatch;
@ -23,21 +23,21 @@ use solana_streamer::streamer::StakedNodes;
use solana_streamer::tls_certificates::new_self_signed_tls_certificate;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::ops::Deref;
use std::option::Option;
use std::path::Path;
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::{Duration, Instant};
use tokio::runtime::{Builder, Runtime};
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::SendError;
use tokio::runtime::{Builder};
use tokio::task::{JoinHandle, yield_now};
use tokio::time::{interval, sleep};
use tokio::time::{sleep};
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{filter::LevelFilter, fmt};
use tracing_subscriber::fmt::format::FmtSpan;
use solana_lite_rpc_quic_forward_proxy::proxy::QuicForwardProxy;
use solana_lite_rpc_quic_forward_proxy::tls_config_provicer::SelfSignedTlsConfigProvider;
@ -480,7 +480,7 @@ async fn solana_quic_streamer_start() {
let addr = sock.local_addr().unwrap().ip();
let port = sock.local_addr().unwrap().port();
let tpu_addr = SocketAddr::new(addr, port);
let _tpu_addr = SocketAddr::new(addr, port);
// sleep(Duration::from_millis(500)).await;
@ -587,7 +587,7 @@ async fn start_literpc_client_proxy_mode(
) -> anyhow::Result<()> {
info!("Start lite-rpc test client using quic proxy at {} ...", forward_proxy_address);
let fanout_slots = 4;
let _fanout_slots = 4;
// (String, Vec<u8>) (signature, transaction)
let (sender, _) = tokio::sync::broadcast::channel(MAXIMUM_TRANSACTIONS_IN_QUEUE);
@ -629,7 +629,7 @@ async fn start_literpc_client_proxy_mode(
// get information about the optional validator identity stake
// populated from get_stakes_for_identity()
let identity_stakes = IdentityStakes {
let _identity_stakes = IdentityStakes {
peer_type: ConnectionPeerType::Staked,
stakes: if test_case_params.stake_connection { 30 } else { 0 }, // stake of lite-rpc
min_stakes: 0,
@ -677,7 +677,7 @@ async fn start_literpc_client_proxy_mode(
async fn start_quic_proxy(proxy_listen_addr: SocketAddr) -> anyhow::Result<()> {
let tls_configuration = SelfSignedTlsConfigProvider::new_singleton_self_signed_localhost();
let _tls_configuration = SelfSignedTlsConfigProvider::new_singleton_self_signed_localhost();
let random_unstaked_validator_identity = ValidatorIdentity::new(None);
let tls_config = SelfSignedTlsConfigProvider::new_singleton_self_signed_localhost();
@ -744,7 +744,7 @@ impl SolanaQuicStreamer {
let addr = udp_socket.local_addr().unwrap().ip();
let port = udp_socket.local_addr().unwrap().port();
let tpu_addr = SocketAddr::new(addr, port);
let _tpu_addr = SocketAddr::new(addr, port);
Self {
sock: udp_socket,

8
quic-forward-proxy/src/main.rs
View File

@ -1,12 +1,12 @@
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use anyhow::bail;
use clap::Parser;
use dotenv::dotenv;
use log::info;
use crate::cli::{Args, get_identity_keypair};
use crate::proxy::QuicForwardProxy;
use crate::test_client::quic_test_client::QuicTestClient;
pub use tls_config_provicer::SelfSignedTlsConfigProvider;
use crate::validator_identity::ValidatorIdentity;
@ -39,7 +39,7 @@ pub async fn main() -> anyhow::Result<()> {
// TODO build args struct dedicyted to proxy
let proxy_listener_addr = proxy_rpc_addr.parse().unwrap();
let tls_configuration = SelfSignedTlsConfigProvider::new_singleton_self_signed_localhost();
let _tls_configuration = SelfSignedTlsConfigProvider::new_singleton_self_signed_localhost();
let validator_identity =
ValidatorIdentity::new(get_identity_keypair(&identity_keypair).await);

72
quic-forward-proxy/src/proxy.rs
View File

@ -1,42 +1,42 @@
use std::collections::HashMap;
use std::future::Future;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
use std::path::Path;
use std::net::{SocketAddr};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicU64};
use std::thread;
use std::thread::sleep;
use std::sync::atomic::{AtomicBool};
use std::time::Duration;
use tracing::{debug_span, instrument, Instrument, span};
use anyhow::{anyhow, bail, Context, Error};
use dashmap::DashMap;
use anyhow::{anyhow, bail, Context};
use fan::tokio::mpsc::FanOut;
use futures::sink::Fanout;
use itertools::{any, Itertools};
use log::{debug, error, info, trace, warn};
use quinn::{Connecting, Connection, ConnectionError, Endpoint, SendStream, ServerConfig, TransportConfig, VarInt};
use rcgen::generate_simple_self_signed;
use rustls::{Certificate, PrivateKey};
use rustls::server::ResolvesServerCert;
use serde::{Deserialize, Serialize};
use quinn::{Connection, Endpoint, ServerConfig, VarInt};
use solana_sdk::packet::PACKET_DATA_SIZE;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::quic::QUIC_MAX_UNSTAKED_CONCURRENT_STREAMS;
use solana_sdk::signature::Keypair;
use solana_sdk::signer::Signer;
use solana_sdk::transaction::VersionedTransaction;
use tokio::net::ToSocketAddrs;
use solana_streamer::tls_certificates::new_self_signed_tls_certificate;
use tokio::sync::mpsc::{channel, Receiver, Sender};
use tokio::sync::{RwLock, RwLockReadGuard};
use tokio::task::JoinHandle;
use tokio::time::error::Elapsed;
use tokio::time::timeout;
use tracing::field::debug;
use crate::proxy_request_format::TpuForwardingRequest;
use crate::quic_connection_utils::{connection_stats, QuicConnectionUtils};
use crate::quinn_auto_reconnect::AutoReconnect;
use crate::tpu_quic_client::{send_txs_to_tpu_static, SingleTPUConnectionManager, TpuQuicClient};
use crate::tpu_quic_client::{send_txs_to_tpu_static, TpuQuicClient};
use crate::tls_config_provicer::{ProxyTlsConfigProvider, SelfSignedTlsConfigProvider};
use crate::util::AnyhowJoinHandle;
use crate::validator_identity::ValidatorIdentity;
@ -89,7 +89,7 @@ impl QuicForwardProxy {
}
pub async fn start_services(
mut self,
self,
) -> anyhow::Result<()> {
let exit_signal = Arc::new(AtomicBool::new(false));
@ -159,8 +159,8 @@ async fn accept_client_connection(client_connection: Connection, forwarder_chann
return Err(anyhow::Error::msg("error accepting stream"));
}
Ok(recv_stream) => {
let exit_signal_copy = exit_signal.clone();
let tpu_quic_client_copy = tpu_quic_client.clone();
let _exit_signal_copy = exit_signal.clone();
let _tpu_quic_client_copy = tpu_quic_client.clone();
let forwarder_channel_copy = forwarder_channel.clone();
tokio::spawn(async move {
@ -172,7 +172,7 @@ async fn accept_client_connection(client_connection: Connection, forwarder_chann
let proxy_request = TpuForwardingRequest::deserialize_from_raw_request(&raw_request);
trace!("proxy request details: {}", proxy_request);
let tpu_identity = proxy_request.get_identity_tpunode();
let _tpu_identity = proxy_request.get_identity_tpunode();
let tpu_address = proxy_request.get_tpu_socket_addr();
let txs = proxy_request.get_transactions();
@ -212,7 +212,7 @@ async fn tx_forwarder(tpu_quic_client: TpuQuicClient, mut transaction_channel: R
// TODO cleanup agent after a while of iactivity
let mut senders = Vec::new();
for i in 0..4 {
for _i in 0..4 {
let (sender, mut receiver) = channel::<ForwardPacket>(100000);
senders.push(sender);
let exit_signal = exit_signal.clone();
@ -228,7 +228,7 @@ async fn tx_forwarder(tpu_quic_client: TpuQuicClient, mut transaction_channel: R
// let mut connection = tpu_quic_client_copy.create_connection(tpu_address).await.expect("handshake");
loop {
let exit_signal = exit_signal.clone();
let _exit_signal = exit_signal.clone();
loop {
let packet = receiver.recv().await.unwrap();
assert_eq!(packet.tpu_address, tpu_address, "routing error");
@ -295,9 +295,9 @@ async fn tx_forwarder(tpu_quic_client: TpuQuicClient, mut transaction_channel: R
// tpu_quic_client_copy.send_txs_to_tpu(tpu_address, &forward_packet.transactions, exit_signal_copy)).await;
// tpu_quic_client_copy.send_txs_to_tpu(forward_packet.tpu_address, &forward_packet.transactions, exit_signal_copy).await;
}
} // -- loop over transactions from ustream channels
panic!("not reachable");
// not reachable
}

2
quic-forward-proxy/src/proxy_request_format.rs
View File

@ -1,6 +1,6 @@
use std::fmt;
use std::fmt::Display;
use std::net::{SocketAddr, SocketAddrV4};
use std::net::{SocketAddr};
use anyhow::Context;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;

21
quic-forward-proxy/src/quic_connection_utils.rs
View File

@ -1,22 +1,21 @@
use log::{debug, error, info, trace, warn};
use quinn::{ClientConfig, Connection, ConnectionError, Endpoint, EndpointConfig, IdleTimeout, SendStream, TokioRuntime, TransportConfig, VarInt, WriteError};
use log::{debug, error, trace, warn};
use quinn::{ClientConfig, Connection, ConnectionError, Endpoint, EndpointConfig, IdleTimeout, SendStream, TokioRuntime, TransportConfig, VarInt};
use solana_sdk::pubkey::Pubkey;
use std::{
collections::VecDeque,
net::{IpAddr, Ipv4Addr, SocketAddr},
sync::{
atomic::{AtomicBool, AtomicU64, Ordering},
atomic::{AtomicBool, Ordering},
Arc,
},
time::Duration,
};
use anyhow::bail;
use futures::future::join_all;
use itertools::Itertools;
use solana_sdk::quic::QUIC_MAX_TIMEOUT_MS;
use tokio::{sync::RwLock, time::timeout};
use tokio::time::error::Elapsed;
use tracing::instrument;
use tokio::{time::timeout};
const ALPN_TPU_PROTOCOL_ID: &[u8] = b"solana-tpu";
@ -297,7 +296,7 @@ impl QuicConnectionUtils {
}
}
}
Err(elapsed) => {
Err(_elapsed) => {
warn!("timeout sending transactions");
}
}
@ -320,7 +319,7 @@ impl QuicConnectionUtils {
pub async fn send_transaction_batch_parallel(
connection: Connection,
txs: Vec<Vec<u8>>,
exit_signal: Arc<AtomicBool>,
_exit_signal: Arc<AtomicBool>,
connection_timeout: Duration,
) {
assert_ne!(txs.len(), 0, "no transactions to send");
@ -350,7 +349,7 @@ impl QuicConnectionUtils {
}
}
}
Err(elapsed) => {
Err(_elapsed) => {
warn!("timeout sending transactions");
}
}

4
quic-forward-proxy/src/quinn_auto_reconnect.rs
View File

@ -1,10 +1,10 @@
use std::cell::RefCell;
use std::fmt;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU32, Ordering};
use tracing::{debug, info};
use quinn::{Connection, Endpoint};
use tokio::sync::{RwLock, RwLockWriteGuard};
use tokio::sync::{RwLock};
pub struct AutoReconnect {
endpoint: Endpoint,

10
quic-forward-proxy/src/test_client/quic_test_client.rs
View File

@ -1,9 +1,9 @@
use std::net::{SocketAddr, SocketAddrV4};
use std::net::{SocketAddr};
use std::str::FromStr;
use std::sync::Arc;
use std::time::Duration;
use anyhow::bail;
use bytes::BufMut;
use log::{info, trace};
use quinn::{Endpoint, VarInt};
use rustls::ClientConfig;
@ -14,7 +14,7 @@ use crate::proxy_request_format::TpuForwardingRequest;
use crate::quic_connection_utils::SkipServerVerification;
use crate::quic_util::ALPN_TPU_FORWARDPROXY_PROTOCOL_ID;
use crate::tls_config_provicer::ProxyTlsConfigProvider;
use crate::test_client::sample_data_factory::build_raw_sample_tx;
use crate::util::AnyhowJoinHandle;
pub struct QuicTestClient {
@ -36,7 +36,7 @@ impl QuicTestClient {
// connect to a server
pub async fn start_services(
mut self,
self,
) -> anyhow::Result<()> {
let endpoint_copy = self.endpoint.clone();
let test_client_service: AnyhowJoinHandle = tokio::spawn(async move {
@ -50,7 +50,7 @@ impl QuicTestClient {
let connecting = endpoint_copy.connect(self.proxy_addr, "localhost").unwrap();
let connection = tokio::time::timeout(connection_timeout, connecting).await??;
for si in 0..5 {
for _si in 0..5 {
let mut send = connection.open_uni().await?;
let raw = build_memo_tx_raw();

2
quic-forward-proxy/src/test_client/sample_data_factory.rs
View File

@ -4,7 +4,7 @@ use solana_sdk::hash::Hash;
use solana_sdk::instruction::Instruction;
use solana_sdk::message::Message;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, keypair, Signature, Signer};
use solana_sdk::signature::{Keypair, keypair, Signer};
use solana_sdk::transaction::{Transaction, VersionedTransaction};
const MEMO_PROGRAM_ID: &str = "MemoSq4gqABAXKb96qnH8TysNcWxMyWCqXgDLGmfcHr";

36
quic-forward-proxy/src/tpu_quic_client.rs
View File

@ -1,32 +1,32 @@
use std::collections::{HashMap, VecDeque};
use std::io::Write;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, SocketAddrV4};
use std::path::Path;
use std::str::FromStr;
use std::sync::{Arc, Mutex};
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::{Arc};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::time::Duration;
use anyhow::{anyhow, bail, Error};
use anyhow::{anyhow};
use async_trait::async_trait;
use dashmap::DashMap;
use futures::future::join_all;
use itertools::{any, Itertools};
use log::{debug, error, info, trace, warn};
use quinn::{Connecting, Connection, ConnectionError, Endpoint, SendStream, ServerConfig, VarInt};
use rcgen::{generate_simple_self_signed, KeyPair};
use rustls::{Certificate, PrivateKey};
use rustls::server::ResolvesServerCert;
use serde::{Deserialize, Serialize};
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::Keypair;
use solana_sdk::signer::Signer;
use itertools::{Itertools};
use log::{debug, info, warn};
use quinn::{Connection, Endpoint};
use solana_sdk::transaction::VersionedTransaction;
use tokio::net::ToSocketAddrs;
use solana_streamer::tls_certificates::new_self_signed_tls_certificate;
use tokio::sync::RwLock;
use crate::quic_connection_utils::{connection_stats, QuicConnectionError, QuicConnectionParameters, QuicConnectionUtils};
use crate::quinn_auto_reconnect::AutoReconnect;
use crate::tls_config_provicer::{ProxyTlsConfigProvider, SelfSignedTlsConfigProvider};
use crate::validator_identity::ValidatorIdentity;
const QUIC_CONNECTION_TIMEOUT: Duration = Duration::from_secs(5);

16
quic-forward-proxy/tests/proxy_request_format.rs
View File

@ -1,14 +1,14 @@
use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
use std::str::FromStr;
use anyhow::Context;
use bincode::DefaultOptions;
use log::info;
use serde::Serialize;
use solana_sdk::hash::{Hash, Hasher};
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, Signer};
use solana_sdk::transaction::{Transaction, VersionedTransaction};
use spl_memo::solana_program::message::VersionedMessage;
use solana_sdk::transaction::{Transaction};
use solana_lite_rpc_quic_forward_proxy::proxy_request_format::TpuForwardingRequest;
#[test]

38
services/src/tpu_utils/quic_proxy_connection_manager.rs
View File

@ -1,29 +1,29 @@
use std::cell::Cell;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, AtomicPtr, Ordering};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::atomic::Ordering::Relaxed;
use std::thread;
use std::time::Duration;
use anyhow::{bail, Context};
use async_trait::async_trait;
use anyhow::{bail};
use futures::FutureExt;
use itertools::Itertools;
use log::{debug, error, info, trace, warn};
use quinn::{ClientConfig, Connection, Endpoint, EndpointConfig, IdleTimeout, TokioRuntime, TransportConfig, VarInt};
use log::{debug, error, info, trace};
use quinn::{ClientConfig, Endpoint, EndpointConfig, IdleTimeout, TokioRuntime, TransportConfig, VarInt};
use solana_sdk::packet::PACKET_DATA_SIZE;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signer::Signer;
use solana_sdk::signature::Keypair;
use solana_sdk::transaction::VersionedTransaction;
use tokio::sync::{broadcast::Receiver, broadcast::Sender, RwLock};
use tokio::time::timeout;
use tracing::field::debug;
use solana_lite_rpc_core::proxy_request_format::TpuForwardingRequest;
use solana_lite_rpc_core::quic_connection_utils::{connection_stats, QuicConnectionParameters, QuicConnectionUtils, SkipServerVerification};
use solana_lite_rpc_core::structures::identity_stakes::IdentityStakes;
use solana_lite_rpc_core::tx_store::TxStore;
use solana_lite_rpc_core::quic_connection_utils::{connection_stats, SkipServerVerification};
use crate::tpu_utils::quinn_auto_reconnect::AutoReconnect;
#[derive(Clone, Copy, Debug)]
@ -93,7 +93,7 @@ impl QuicProxyConnectionManager {
info!("Starting very simple proxy thread");
let mut transaction_receiver = transaction_sender.subscribe();
let transaction_receiver = transaction_sender.subscribe();
// TODO use it
let exit_signal = Arc::new(AtomicBool::new(false));
@ -135,7 +135,7 @@ impl QuicProxyConnectionManager {
// note: this config must be aligned with quic-proxy's server config
let mut transport_config = TransportConfig::default();
let timeout = IdleTimeout::try_from(Duration::from_secs(1)).unwrap();
let _timeout = IdleTimeout::try_from(Duration::from_secs(1)).unwrap();
// no remotely-initiated streams required
transport_config.max_concurrent_uni_streams(VarInt::from_u32(0));
transport_config.max_concurrent_bidi_streams(VarInt::from_u32(0));
@ -175,7 +175,7 @@ impl QuicProxyConnectionManager {
let first_tx: Vec<u8> = match tx {
Ok((sig, tx)) => {
Ok((_sig, tx)) => {
// if Self::check_for_confirmation(&txs_sent_store, sig) {
// // transaction is already confirmed/ no need to send
// continue;
@ -193,9 +193,9 @@ impl QuicProxyConnectionManager {
let mut txs = vec![first_tx];
// TODO comment in
let foo = PACKET_DATA_SIZE;
let _foo = PACKET_DATA_SIZE;
for _ in 1..number_of_transactions_per_unistream {
if let Ok((signature, tx)) = transaction_receiver.try_recv() {
if let Ok((_signature, tx)) = transaction_receiver.try_recv() {
// if Self::check_for_confirmation(&txs_sent_store, signature) {
// continue;
// }
@ -223,7 +223,7 @@ impl QuicProxyConnectionManager {
}
async fn send_copy_of_txs_to_quicproxy(raw_tx_batch: &Vec<Vec<u8>>, auto_connection: &AutoReconnect,
proxy_address: SocketAddr, tpu_target_address: SocketAddr,
_proxy_address: SocketAddr, tpu_target_address: SocketAddr,
target_tpu_identity: Pubkey) -> anyhow::Result<()> {
// TODO add timeout
@ -276,7 +276,7 @@ impl QuicProxyConnectionManager {
async fn send_proxy_request(endpoint: Endpoint, proxy_address: SocketAddr, proxy_request_raw: &Vec<u8>) -> anyhow::Result<()> {
info!("sending {} bytes to proxy", proxy_request_raw.len());
let mut connecting = endpoint.connect(proxy_address, "localhost")?;
let connecting = endpoint.connect(proxy_address, "localhost")?;
let connection = timeout(Duration::from_millis(500), connecting).await??;
let mut send = connection.open_uni().await?;

6
services/src/tpu_utils/quinn_auto_reconnect.rs
View File

@ -1,11 +1,11 @@
use std::cell::RefCell;
use std::fmt;
use std::net::SocketAddr;
use std::sync::atomic::{AtomicU32, Ordering};
use log::{trace, warn};
use tracing::{debug, info};
use tracing::{debug};
use quinn::{Connection, Endpoint};
use tokio::sync::{RwLock, RwLockWriteGuard};
use tokio::sync::{RwLock};
pub struct AutoReconnect {
endpoint: Endpoint,

6
services/src/tpu_utils/tpu_connection_manager.rs
View File

@ -1,5 +1,5 @@
use dashmap::DashMap;
use log::{error, info, trace};
use log::{error, trace};
use prometheus::{core::GenericGauge, opts, register_int_gauge};
use quinn::Endpoint;
use solana_lite_rpc_core::{
@ -20,8 +20,8 @@ use std::{
},
};
use tokio::sync::{broadcast::Receiver, broadcast::Sender};
use crate::tpu_utils::tpu_connection_path::TpuConnectionPath;
use crate::tpu_utils::tpu_connection_path::TpuConnectionPath::QuicForwardProxyPath;
lazy_static::lazy_static! {
static ref NB_QUIC_CONNECTIONS: GenericGauge<prometheus::core::AtomicI64> =

16
services/src/tpu_utils/tpu_connection_path.rs
View File

@ -1,13 +1,13 @@
use std::collections::HashMap;
use std::fmt::Display;
use std::net::SocketAddr;
use std::sync::Arc;
use async_trait::async_trait;
use solana_sdk::pubkey::Pubkey;
use tokio::sync::{broadcast::Receiver, broadcast::Sender};
use solana_lite_rpc_core::quic_connection_utils::QuicConnectionParameters;
use solana_lite_rpc_core::structures::identity_stakes::IdentityStakes;
use solana_lite_rpc_core::tx_store::TxStore;
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub enum TpuConnectionPath {

34
services/tests/literpc_tpu_quic_server_integrationtest.rs
View File

@ -1,9 +1,9 @@
use anyhow::bail;
use countmap::CountMap;
use crossbeam_channel::{Receiver, RecvError, RecvTimeoutError, Sender};
use futures::future::join_all;
use log::{debug, error, info, trace, warn};
use quinn::TokioRuntime;
use crossbeam_channel::{Sender};
use log::{debug, info, trace, warn};
use solana_lite_rpc_core::quic_connection_utils::QuicConnectionParameters;
use solana_lite_rpc_core::structures::identity_stakes::IdentityStakes;
use solana_lite_rpc_core::tx_store::empty_tx_store;
@ -14,7 +14,7 @@ use solana_sdk::instruction::Instruction;
use solana_sdk::message::Message;
use solana_sdk::pubkey::Pubkey;
use solana_sdk::signature::{Keypair, Signature, Signer};
use solana_sdk::signer::keypair;
use solana_sdk::transaction::{Transaction, VersionedTransaction};
use solana_streamer::nonblocking::quic::ConnectionPeerType;
use solana_streamer::packet::PacketBatch;
@ -23,21 +23,21 @@ use solana_streamer::streamer::StakedNodes;
use solana_streamer::tls_certificates::new_self_signed_tls_certificate;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::ops::Deref;
use std::option::Option;
use std::path::Path;
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::{Duration, Instant};
use tokio::runtime::{Builder, Runtime};
use tokio::sync::broadcast;
use tokio::sync::broadcast::error::SendError;
use tokio::runtime::{Builder};
use tokio::task::JoinHandle;
use tokio::time::{interval, sleep};
use tokio::time::{sleep};
use tracing_subscriber::util::SubscriberInitExt;
use tracing_subscriber::{filter::LevelFilter, fmt};
#[derive(Copy, Clone, Debug)]
struct TestCaseParams {
@ -378,7 +378,7 @@ async fn solana_quic_streamer_start() {
let addr = sock.local_addr().unwrap().ip();
let port = sock.local_addr().unwrap().port();
let tpu_addr = SocketAddr::new(addr, port);
let _tpu_addr = SocketAddr::new(addr, port);
// sleep(Duration::from_millis(500)).await;
@ -439,7 +439,7 @@ impl SolanaQuicStreamer {
let addr = udp_socket.local_addr().unwrap().ip();
let port = udp_socket.local_addr().unwrap().port();
let tpu_addr = SocketAddr::new(addr, port);
let _tpu_addr = SocketAddr::new(addr, port);
Self {
sock: udp_socket,