solana_core/
warm_quic_cache_service.rs

1// Connect to future leaders with some jitter so the quic connection is warm
2// by the time we need it.
3
4use {
5    rand::{Rng, rng},
6    solana_client::connection_cache::{ConnectionCache, Protocol},
7    solana_connection_cache::client_connection::ClientConnection as TpuConnection,
8    solana_gossip::{cluster_info::ClusterInfo, contact_info::ContactInfoQuery},
9    solana_poh::poh_recorder::PohRecorder,
10    solana_pubkey::Pubkey,
11    std::{
12        net::SocketAddr,
13        sync::{
14            Arc, RwLock,
15            atomic::{AtomicBool, Ordering},
16        },
17        thread::{self, Builder, JoinHandle, sleep},
18        time::Duration,
19    },
20};
21
22pub struct WarmQuicCacheService {
23    thread_hdl: JoinHandle<()>,
24}
25
26// ~50 seconds
27const CACHE_OFFSET_SLOT: i64 = 100;
28const CACHE_JITTER_SLOT: i64 = 20;
29
30impl WarmQuicCacheService {
31    fn warmup_connection(
32        cache: Option<&ConnectionCache>,
33        cluster_info: &ClusterInfo,
34        leader_pubkey: &Pubkey,
35        contact_info_selector: impl ContactInfoQuery<Option<SocketAddr>>,
36        log_context: &str,
37    ) {
38        if let Some(connection_cache) = cache
39            && let Some(Some(addr)) =
40                cluster_info.lookup_contact_info(leader_pubkey, contact_info_selector)
41        {
42            let conn = connection_cache.get_connection(&addr);
43            if let Err(err) = conn.send_data(&[]) {
44                warn!(
45                    "Failed to warmup QUIC connection to the leader {leader_pubkey:?} at \
46                     {addr:?}, Context: {log_context}, Error: {err:?}"
47                );
48            }
49        }
50    }
51
52    pub fn new(
53        tpu_connection_cache: Option<Arc<ConnectionCache>>,
54        vote_connection_cache: Option<Arc<ConnectionCache>>,
55        cluster_info: Arc<ClusterInfo>,
56        poh_recorder: Arc<RwLock<PohRecorder>>,
57        exit: Arc<AtomicBool>,
58    ) -> Self {
59        assert!(matches!(
60            tpu_connection_cache.as_deref(),
61            None | Some(ConnectionCache::Quic(_))
62        ));
63        assert!(matches!(
64            vote_connection_cache.as_deref(),
65            None | Some(ConnectionCache::Quic(_))
66        ));
67        let thread_hdl = Builder::new()
68            .name("solWarmQuicSvc".to_string())
69            .spawn(move || {
70                let slot_jitter = rng().random_range(-CACHE_JITTER_SLOT..CACHE_JITTER_SLOT);
71                let mut maybe_last_leader = None;
72                while !exit.load(Ordering::Relaxed) {
73                    let leader_pubkey = poh_recorder
74                        .read()
75                        .unwrap()
76                        .leader_after_n_slots((CACHE_OFFSET_SLOT + slot_jitter) as u64);
77                    if let Some(leader_pubkey) = leader_pubkey
78                        && maybe_last_leader != Some(leader_pubkey)
79                    {
80                        maybe_last_leader = Some(leader_pubkey);
81                        // Warm cache for regular transactions
82                        Self::warmup_connection(
83                            tpu_connection_cache.as_deref(),
84                            &cluster_info,
85                            &leader_pubkey,
86                            |node| node.tpu(Protocol::QUIC),
87                            "tpu",
88                        );
89                        // Warm cache for vote
90                        Self::warmup_connection(
91                            vote_connection_cache.as_deref(),
92                            &cluster_info,
93                            &leader_pubkey,
94                            |node| node.tpu_vote(Protocol::QUIC),
95                            "vote",
96                        );
97                    }
98                    sleep(Duration::from_millis(200));
99                }
100            })
101            .unwrap();
102        Self { thread_hdl }
103    }
104
105    pub fn join(self) -> thread::Result<()> {
106        self.thread_hdl.join()
107    }
108}
solana_core/warm_quic_cache_service.rs

solana_core/
warm_quic_cache_service.rs
