solana_connection_cache/
connection_cache.rs

1use {
2    crate::{
3        client_connection::ClientConnection as BlockingClientConnection,
4        connection_cache_stats::{CONNECTION_STAT_SUBMISSION_INTERVAL, ConnectionCacheStats},
5        nonblocking::client_connection::ClientConnection as NonblockingClientConnection,
6    },
7    crossbeam_channel::{Receiver, RecvError, Sender},
8    indexmap::map::IndexMap,
9    log::*,
10    rand::{Rng, rng},
11    solana_keypair::Keypair,
12    solana_measure::measure::Measure,
13    solana_time_utils::AtomicInterval,
14    std::{
15        net::SocketAddr,
16        sync::{Arc, RwLock, atomic::Ordering},
17        thread::{Builder, JoinHandle},
18    },
19    thiserror::Error,
20};
21
22/// Default maximum number of connections to keep
23pub const DEFAULT_MAX_CONNECTIONS: usize = 1024;
24
25/// Default connection pool size per remote address
26pub const DEFAULT_CONNECTION_POOL_SIZE: usize = 2;
27
28pub use solana_net_utils::Protocol;
29
30pub trait ConnectionManager: Send + Sync + 'static {
31    type ConnectionPool: ConnectionPool;
32    type NewConnectionConfig: NewConnectionConfig;
33
34    const PROTOCOL: Protocol;
35
36    fn new_connection_pool(&self) -> Self::ConnectionPool;
37    fn new_connection_config(&self) -> Self::NewConnectionConfig;
38    fn update_key(&self, _key: &Keypair) -> Result<(), Box<dyn std::error::Error>>;
39}
40
41pub struct ConnectionCache<
42    R, // ConnectionPool
43    S, // ConnectionManager
44    T, // NewConnectionConfig
45> {
46    name: &'static str,
47    map: Arc<RwLock<IndexMap<SocketAddr, /*ConnectionPool:*/ R>>>,
48    connection_manager: Arc<S>,
49    stats: Arc<ConnectionCacheStats>,
50    last_stats: AtomicInterval,
51    connection_pool_size: usize,
52    max_connections: usize,
53    connection_config: Arc<T>,
54    sender: Sender<(usize, SocketAddr)>,
55}
56
57impl<P, M, C> ConnectionCache<P, M, C>
58where
59    P: ConnectionPool<NewConnectionConfig = C>,
60    M: ConnectionManager<ConnectionPool = P, NewConnectionConfig = C>,
61    C: NewConnectionConfig,
62{
63    pub fn new(
64        name: &'static str,
65        connection_manager: M,
66        connection_pool_size: usize,
67    ) -> Result<Self, ClientError> {
68        Self::new_with_max_connections(
69            name,
70            connection_manager,
71            connection_pool_size,
72            DEFAULT_MAX_CONNECTIONS,
73        )
74    }
75
76    pub fn new_with_max_connections(
77        name: &'static str,
78        connection_manager: M,
79        connection_pool_size: usize,
80        max_connections: usize,
81    ) -> Result<Self, ClientError> {
82        let config = Arc::new(connection_manager.new_connection_config());
83        let max_connections = 1.max(max_connections); // The minimum is 1.
84        info!(
85            "Creating ConnectionCache {name}, pool size: {connection_pool_size}, max connections: \
86             {max_connections}"
87        );
88        let (sender, receiver) = crossbeam_channel::unbounded();
89
90        let map = Arc::new(RwLock::new(IndexMap::with_capacity(max_connections)));
91        let connection_manager = Arc::new(connection_manager);
92        let connection_pool_size = 1.max(connection_pool_size); // The minimum pool size is 1.
93
94        let stats = Arc::new(ConnectionCacheStats::default());
95
96        let _async_connection_thread =
97            Self::create_connection_async_thread(map.clone(), receiver, stats.clone());
98        Ok(Self {
99            name,
100            map,
101            stats,
102            connection_manager,
103            last_stats: AtomicInterval::default(),
104            connection_pool_size,
105            max_connections,
106            connection_config: config,
107            sender,
108        })
109    }
110
111    /// This actually triggers the connection creation by sending empty data
112    fn create_connection_async_thread(
113        map: Arc<RwLock<IndexMap<SocketAddr, /*ConnectionPool:*/ P>>>,
114        receiver: Receiver<(usize, SocketAddr)>,
115        stats: Arc<ConnectionCacheStats>,
116    ) -> JoinHandle<()> {
117        Builder::new()
118            .name("solQAsynCon".to_string())
119            .spawn(move || {
120                loop {
121                    let recv_result = receiver.recv();
122                    match recv_result {
123                        Err(RecvError) => {
124                            break;
125                        }
126                        Ok((idx, addr)) => {
127                            let map = map.read().unwrap();
128                            let pool = map.get(&addr);
129                            if let Some(pool) = pool {
130                                let conn = pool.get(idx);
131                                if let Ok(conn) = conn {
132                                    drop(map);
133                                    let conn = conn.new_blocking_connection(addr, stats.clone());
134                                    let result = conn.send_data(&[]);
135                                    debug!("Create async connection result {result:?} for {addr}");
136                                }
137                            }
138                        }
139                    }
140                }
141            })
142            .unwrap()
143    }
144
145    pub fn update_key(&self, key: &Keypair) -> Result<(), Box<dyn std::error::Error>> {
146        let mut map = self.map.write().unwrap();
147        map.clear();
148        self.connection_manager.update_key(key)
149    }
150    /// Create a lazy connection object under the exclusive lock of the cache map if there is not
151    /// enough used connections in the connection pool for the specified address.
152    /// Returns CreateConnectionResult.
153    fn create_connection(
154        &self,
155        lock_timing_ms: &mut u64,
156        addr: &SocketAddr,
157    ) -> CreateConnectionResult<<P as ConnectionPool>::BaseClientConnection> {
158        let mut get_connection_map_lock_measure = Measure::start("get_connection_map_lock_measure");
159        let mut map = self.map.write().unwrap();
160        get_connection_map_lock_measure.stop();
161        *lock_timing_ms = lock_timing_ms.saturating_add(get_connection_map_lock_measure.as_ms());
162        // Read again, as it is possible that between read lock dropped and the write lock acquired
163        // another thread could have setup the connection.
164
165        let pool_status = map
166            .get(addr)
167            .map(|pool| pool.check_pool_status(self.connection_pool_size))
168            .unwrap_or(PoolStatus::Empty);
169
170        let (cache_hit, num_evictions, eviction_timing_ms) =
171            if matches!(pool_status, PoolStatus::Empty) {
172                Self::create_connection_internal(
173                    &self.connection_config,
174                    &self.connection_manager,
175                    &mut map,
176                    addr,
177                    self.connection_pool_size,
178                    self.max_connections,
179                    None,
180                )
181            } else {
182                (true, 0, 0)
183            };
184
185        if matches!(pool_status, PoolStatus::PartiallyFull) {
186            // trigger an async connection create
187            debug!("Triggering async connection for {addr:?}");
188            Self::create_connection_internal(
189                &self.connection_config,
190                &self.connection_manager,
191                &mut map,
192                addr,
193                self.connection_pool_size,
194                self.max_connections,
195                Some(&self.sender),
196            );
197        }
198
199        let pool = map.get(addr).unwrap();
200        let connection = pool.borrow_connection();
201
202        CreateConnectionResult {
203            connection,
204            cache_hit,
205            connection_cache_stats: self.stats.clone(),
206            num_evictions,
207            eviction_timing_ms,
208        }
209    }
210
211    fn create_connection_internal(
212        config: &C,
213        connection_manager: &M,
214        map: &mut std::sync::RwLockWriteGuard<'_, IndexMap<SocketAddr, P>>,
215        addr: &SocketAddr,
216        connection_pool_size: usize,
217        max_connections: usize,
218        async_connection_sender: Option<&Sender<(usize, SocketAddr)>>,
219    ) -> (bool, u64, u64) {
220        // evict a connection if the cache is reaching upper bounds
221        let mut num_evictions = 0;
222        let mut get_connection_cache_eviction_measure =
223            Measure::start("get_connection_cache_eviction_measure");
224        let existing_index = map.get_index_of(addr);
225        while map.len() >= max_connections {
226            let mut rng = rng();
227            let n = rng.random_range(0..max_connections);
228            if let Some(index) = existing_index
229                && n == index
230            {
231                continue;
232            }
233            map.swap_remove_index(n);
234            num_evictions += 1;
235        }
236        get_connection_cache_eviction_measure.stop();
237
238        let mut hit_cache = false;
239        map.entry(*addr)
240            .and_modify(|pool| {
241                if matches!(
242                    pool.check_pool_status(connection_pool_size),
243                    PoolStatus::PartiallyFull
244                ) {
245                    let idx = pool.add_connection(config, addr);
246                    if let Some(sender) = async_connection_sender {
247                        debug!(
248                            "Sending async connection creation {} for {addr}",
249                            pool.num_connections() - 1
250                        );
251                        sender.send((idx, *addr)).unwrap();
252                    };
253                } else {
254                    hit_cache = true;
255                }
256            })
257            .or_insert_with(|| {
258                let mut pool = connection_manager.new_connection_pool();
259                pool.add_connection(config, addr);
260                pool
261            });
262        (
263            hit_cache,
264            num_evictions,
265            get_connection_cache_eviction_measure.as_ms(),
266        )
267    }
268
269    fn get_or_add_connection(
270        &self,
271        addr: &SocketAddr,
272    ) -> GetConnectionResult<<P as ConnectionPool>::BaseClientConnection> {
273        let mut get_connection_map_lock_measure = Measure::start("get_connection_map_lock_measure");
274        let map = self.map.read().unwrap();
275        get_connection_map_lock_measure.stop();
276
277        let mut lock_timing_ms = get_connection_map_lock_measure.as_ms();
278
279        let report_stats = self
280            .last_stats
281            .should_update(CONNECTION_STAT_SUBMISSION_INTERVAL);
282
283        let mut get_connection_map_measure = Measure::start("get_connection_hit_measure");
284        let CreateConnectionResult {
285            connection,
286            cache_hit,
287            connection_cache_stats,
288            num_evictions,
289            eviction_timing_ms,
290        } = match map.get(addr) {
291            Some(pool) => {
292                let pool_status = pool.check_pool_status(self.connection_pool_size);
293                match pool_status {
294                    PoolStatus::Empty => {
295                        // create more connection and put it in the pool
296                        drop(map);
297                        self.create_connection(&mut lock_timing_ms, addr)
298                    }
299                    PoolStatus::PartiallyFull | PoolStatus::Full => {
300                        let connection = pool.borrow_connection();
301                        if matches!(pool_status, PoolStatus::PartiallyFull) {
302                            debug!("Creating connection async for {addr}");
303                            drop(map);
304                            let mut map = self.map.write().unwrap();
305                            Self::create_connection_internal(
306                                &self.connection_config,
307                                &self.connection_manager,
308                                &mut map,
309                                addr,
310                                self.connection_pool_size,
311                                self.max_connections,
312                                Some(&self.sender),
313                            );
314                        }
315                        CreateConnectionResult {
316                            connection,
317                            cache_hit: true,
318                            connection_cache_stats: self.stats.clone(),
319                            num_evictions: 0,
320                            eviction_timing_ms: 0,
321                        }
322                    }
323                }
324            }
325            None => {
326                // Upgrade to write access by dropping read lock and acquire write lock
327                drop(map);
328                self.create_connection(&mut lock_timing_ms, addr)
329            }
330        };
331        get_connection_map_measure.stop();
332
333        GetConnectionResult {
334            connection,
335            cache_hit,
336            report_stats,
337            map_timing_ms: get_connection_map_measure.as_ms(),
338            lock_timing_ms,
339            connection_cache_stats,
340            num_evictions,
341            eviction_timing_ms,
342        }
343    }
344
345    fn get_connection_and_log_stats(
346        &self,
347        addr: &SocketAddr,
348    ) -> (
349        Arc<<P as ConnectionPool>::BaseClientConnection>,
350        Arc<ConnectionCacheStats>,
351    ) {
352        let mut get_connection_measure = Measure::start("get_connection_measure");
353        let GetConnectionResult {
354            connection,
355            cache_hit,
356            report_stats,
357            map_timing_ms,
358            lock_timing_ms,
359            connection_cache_stats,
360            num_evictions,
361            eviction_timing_ms,
362        } = self.get_or_add_connection(addr);
363
364        if report_stats {
365            connection_cache_stats.report(self.name);
366        }
367
368        if cache_hit {
369            connection_cache_stats
370                .cache_hits
371                .fetch_add(1, Ordering::Relaxed);
372            connection_cache_stats
373                .get_connection_hit_ms
374                .fetch_add(map_timing_ms, Ordering::Relaxed);
375        } else {
376            connection_cache_stats
377                .cache_misses
378                .fetch_add(1, Ordering::Relaxed);
379            connection_cache_stats
380                .get_connection_miss_ms
381                .fetch_add(map_timing_ms, Ordering::Relaxed);
382            connection_cache_stats
383                .cache_evictions
384                .fetch_add(num_evictions, Ordering::Relaxed);
385            connection_cache_stats
386                .eviction_time_ms
387                .fetch_add(eviction_timing_ms, Ordering::Relaxed);
388        }
389
390        get_connection_measure.stop();
391        connection_cache_stats
392            .get_connection_lock_ms
393            .fetch_add(lock_timing_ms, Ordering::Relaxed);
394        connection_cache_stats
395            .get_connection_ms
396            .fetch_add(get_connection_measure.as_ms(), Ordering::Relaxed);
397
398        (connection, connection_cache_stats)
399    }
400
401    pub fn get_connection(&self, addr: &SocketAddr) -> Arc<<<P as ConnectionPool>::BaseClientConnection as BaseClientConnection>::BlockingClientConnection>{
402        let (connection, connection_cache_stats) = self.get_connection_and_log_stats(addr);
403        connection.new_blocking_connection(*addr, connection_cache_stats)
404    }
405
406    pub fn get_nonblocking_connection(
407        &self,
408        addr: &SocketAddr,
409    ) -> Arc<<<P as ConnectionPool>::BaseClientConnection as BaseClientConnection>::NonblockingClientConnection>{
410        let (connection, connection_cache_stats) = self.get_connection_and_log_stats(addr);
411        connection.new_nonblocking_connection(*addr, connection_cache_stats)
412    }
413}
414
415#[derive(Error, Debug)]
416pub enum ConnectionPoolError {
417    #[error("connection index is out of range of the pool")]
418    IndexOutOfRange,
419}
420
421#[derive(Error, Debug)]
422pub enum ClientError {
423    #[error("IO error: {0:?}")]
424    IoError(#[from] std::io::Error),
425}
426
427pub trait NewConnectionConfig: Sized + Send + Sync + 'static {
428    fn new() -> Result<Self, ClientError>;
429}
430
431pub enum PoolStatus {
432    Empty,
433    PartiallyFull,
434    Full,
435}
436
437pub trait ConnectionPool: Send + Sync + 'static {
438    type NewConnectionConfig: NewConnectionConfig;
439    type BaseClientConnection: BaseClientConnection;
440
441    /// Add a connection to the pool and return its index
442    fn add_connection(&mut self, config: &Self::NewConnectionConfig, addr: &SocketAddr) -> usize;
443
444    /// Get the number of current connections in the pool
445    fn num_connections(&self) -> usize;
446
447    /// Get a connection based on its index in the pool, without checking if the
448    fn get(&self, index: usize) -> Result<Arc<Self::BaseClientConnection>, ConnectionPoolError>;
449
450    /// Get a connection from the pool. It must have at least one connection in the pool.
451    /// This randomly picks a connection in the pool.
452    fn borrow_connection(&self) -> Arc<Self::BaseClientConnection> {
453        let mut rng = rng();
454        let n = rng.random_range(0..self.num_connections());
455        self.get(n).expect("index is within num_connections")
456    }
457
458    /// Check if we need to create a new connection. If the count of the connections
459    /// is smaller than the pool size and if there is no connection at all.
460    fn check_pool_status(&self, required_pool_size: usize) -> PoolStatus {
461        if self.num_connections() == 0 {
462            PoolStatus::Empty
463        } else if self.num_connections() < required_pool_size {
464            PoolStatus::PartiallyFull
465        } else {
466            PoolStatus::Full
467        }
468    }
469
470    fn create_pool_entry(
471        &self,
472        config: &Self::NewConnectionConfig,
473        addr: &SocketAddr,
474    ) -> Arc<Self::BaseClientConnection>;
475}
476
477pub trait BaseClientConnection {
478    type BlockingClientConnection: BlockingClientConnection;
479    type NonblockingClientConnection: NonblockingClientConnection;
480
481    fn new_blocking_connection(
482        &self,
483        addr: SocketAddr,
484        stats: Arc<ConnectionCacheStats>,
485    ) -> Arc<Self::BlockingClientConnection>;
486
487    fn new_nonblocking_connection(
488        &self,
489        addr: SocketAddr,
490        stats: Arc<ConnectionCacheStats>,
491    ) -> Arc<Self::NonblockingClientConnection>;
492}
493
494struct GetConnectionResult<T> {
495    connection: Arc</*BaseClientConnection:*/ T>,
496    cache_hit: bool,
497    report_stats: bool,
498    map_timing_ms: u64,
499    lock_timing_ms: u64,
500    connection_cache_stats: Arc<ConnectionCacheStats>,
501    num_evictions: u64,
502    eviction_timing_ms: u64,
503}
504
505struct CreateConnectionResult<T> {
506    connection: Arc</*BaseClientConnection:*/ T>,
507    cache_hit: bool,
508    connection_cache_stats: Arc<ConnectionCacheStats>,
509    num_evictions: u64,
510    eviction_timing_ms: u64,
511}
512
513#[cfg(test)]
514mod tests {
515    use {
516        super::*,
517        crate::{
518            client_connection::ClientConnection as BlockingClientConnection,
519            nonblocking::client_connection::ClientConnection as NonblockingClientConnection,
520        },
521        async_trait::async_trait,
522        rand::{Rng, SeedableRng},
523        rand_chacha::ChaChaRng,
524        solana_net_utils::sockets::bind_to_localhost_unique,
525        solana_transaction_error::TransportResult,
526        std::{
527            net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket},
528            sync::Arc,
529        },
530    };
531
532    struct MockUdpPool {
533        connections: Vec<Arc<MockUdp>>,
534    }
535    impl ConnectionPool for MockUdpPool {
536        type NewConnectionConfig = MockUdpConfig;
537        type BaseClientConnection = MockUdp;
538
539        /// Add a connection into the pool and return its index in the pool.
540        fn add_connection(
541            &mut self,
542            config: &Self::NewConnectionConfig,
543            addr: &SocketAddr,
544        ) -> usize {
545            let connection = self.create_pool_entry(config, addr);
546            let idx = self.connections.len();
547            self.connections.push(connection);
548            idx
549        }
550
551        fn num_connections(&self) -> usize {
552            self.connections.len()
553        }
554
555        fn get(
556            &self,
557            index: usize,
558        ) -> Result<Arc<Self::BaseClientConnection>, ConnectionPoolError> {
559            self.connections
560                .get(index)
561                .cloned()
562                .ok_or(ConnectionPoolError::IndexOutOfRange)
563        }
564
565        fn create_pool_entry(
566            &self,
567            config: &Self::NewConnectionConfig,
568            _addr: &SocketAddr,
569        ) -> Arc<Self::BaseClientConnection> {
570            Arc::new(MockUdp(config.udp_socket.clone()))
571        }
572    }
573
574    struct MockUdpConfig {
575        udp_socket: Arc<UdpSocket>,
576    }
577
578    impl Default for MockUdpConfig {
579        fn default() -> Self {
580            Self {
581                udp_socket: Arc::new(bind_to_localhost_unique().unwrap()),
582            }
583        }
584    }
585
586    impl NewConnectionConfig for MockUdpConfig {
587        fn new() -> Result<Self, ClientError> {
588            Ok(Self {
589                udp_socket: Arc::new(
590                    bind_to_localhost_unique().map_err(Into::<ClientError>::into)?,
591                ),
592            })
593        }
594    }
595
596    struct MockUdp(Arc<UdpSocket>);
597    impl BaseClientConnection for MockUdp {
598        type BlockingClientConnection = MockUdpConnection;
599        type NonblockingClientConnection = MockUdpConnection;
600
601        fn new_blocking_connection(
602            &self,
603            addr: SocketAddr,
604            _stats: Arc<ConnectionCacheStats>,
605        ) -> Arc<Self::BlockingClientConnection> {
606            Arc::new(MockUdpConnection {
607                _socket: self.0.clone(),
608                addr,
609            })
610        }
611
612        fn new_nonblocking_connection(
613            &self,
614            addr: SocketAddr,
615            _stats: Arc<ConnectionCacheStats>,
616        ) -> Arc<Self::NonblockingClientConnection> {
617            Arc::new(MockUdpConnection {
618                _socket: self.0.clone(),
619                addr,
620            })
621        }
622    }
623
624    struct MockUdpConnection {
625        _socket: Arc<UdpSocket>,
626        addr: SocketAddr,
627    }
628
629    #[derive(Default)]
630    struct MockConnectionManager {}
631
632    impl ConnectionManager for MockConnectionManager {
633        type ConnectionPool = MockUdpPool;
634        type NewConnectionConfig = MockUdpConfig;
635
636        const PROTOCOL: Protocol = Protocol::QUIC;
637
638        fn new_connection_pool(&self) -> Self::ConnectionPool {
639            MockUdpPool {
640                connections: Vec::default(),
641            }
642        }
643
644        fn new_connection_config(&self) -> Self::NewConnectionConfig {
645            MockUdpConfig::new().unwrap()
646        }
647
648        fn update_key(&self, _key: &Keypair) -> Result<(), Box<dyn std::error::Error>> {
649            Ok(())
650        }
651    }
652
653    impl BlockingClientConnection for MockUdpConnection {
654        fn server_addr(&self) -> &SocketAddr {
655            &self.addr
656        }
657        fn send_data(&self, _buffer: &[u8]) -> TransportResult<()> {
658            unimplemented!()
659        }
660        fn send_data_async(&self, _data: Arc<Vec<u8>>) -> TransportResult<()> {
661            unimplemented!()
662        }
663        fn send_data_batch(&self, _buffers: &[Vec<u8>]) -> TransportResult<()> {
664            unimplemented!()
665        }
666        fn send_data_batch_async(&self, _buffers: Vec<Vec<u8>>) -> TransportResult<()> {
667            unimplemented!()
668        }
669    }
670
671    #[async_trait]
672    impl NonblockingClientConnection for MockUdpConnection {
673        fn server_addr(&self) -> &SocketAddr {
674            &self.addr
675        }
676        async fn send_data(&self, _data: &[u8]) -> TransportResult<()> {
677            unimplemented!()
678        }
679        async fn send_data_batch(&self, _buffers: &[Vec<u8>]) -> TransportResult<()> {
680            unimplemented!()
681        }
682    }
683
684    fn get_addr(rng: &mut ChaChaRng) -> SocketAddr {
685        let a = rng.random_range(1..255);
686        let b = rng.random_range(1..255);
687        let c = rng.random_range(1..255);
688        let d = rng.random_range(1..255);
689
690        let addr_str = format!("{a}.{b}.{c}.{d}:80");
691
692        addr_str.parse().expect("Invalid address")
693    }
694
695    #[test]
696    fn test_connection_cache() {
697        agave_logger::setup();
698        // Allow the test to run deterministically
699        // with the same pseudorandom sequence between runs
700        // and on different platforms - the cryptographic security
701        // property isn't important here but ChaChaRng provides a way
702        // to get the same pseudorandom sequence on different platforms
703        let mut rng = ChaChaRng::seed_from_u64(42);
704
705        // Generate a bunch of random addresses and create connections to them
706        // Since ClientConnection::new is infallible, it should't matter whether or not
707        // we can actually connect to those addresses - ClientConnection implementations should either
708        // be lazy and not connect until first use or handle connection errors somehow
709        // (without crashing, as would be required in a real practical validator)
710        let connection_manager = MockConnectionManager::default();
711        let connection_cache = ConnectionCache::new(
712            "connection_cache_test",
713            connection_manager,
714            DEFAULT_CONNECTION_POOL_SIZE,
715        )
716        .unwrap();
717        let addrs = (0..DEFAULT_MAX_CONNECTIONS)
718            .map(|_| {
719                let addr = get_addr(&mut rng);
720                connection_cache.get_connection(&addr);
721                addr
722            })
723            .collect::<Vec<_>>();
724        {
725            let map = connection_cache.map.read().unwrap();
726            assert!(map.len() == DEFAULT_MAX_CONNECTIONS);
727            addrs.iter().for_each(|addr| {
728                let conn = &map.get(addr).expect("Address not found").get(0).unwrap();
729                let conn = conn.new_blocking_connection(*addr, connection_cache.stats.clone());
730                assert_eq!(
731                    BlockingClientConnection::server_addr(&*conn).ip(),
732                    addr.ip(),
733                );
734                assert_eq!(
735                    NonblockingClientConnection::server_addr(&*conn).ip(),
736                    addr.ip(),
737                );
738            });
739        }
740
741        let addr = &get_addr(&mut rng);
742        connection_cache.get_connection(addr);
743
744        let port = addr.port();
745        let addr_with_quic_port = SocketAddr::new(addr.ip(), port);
746        let map = connection_cache.map.read().unwrap();
747        assert!(map.len() == DEFAULT_MAX_CONNECTIONS);
748        let _conn = map.get(&addr_with_quic_port).expect("Address not found");
749    }
750
751    // Test that we can get_connection with a connection cache configured
752    // on an address with a port that would overflow to
753    // an invalid port.
754    #[test]
755    fn test_overflow_address() {
756        let port = u16::MAX;
757        let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port);
758        let connection_manager = MockConnectionManager::default();
759        let connection_cache =
760            ConnectionCache::new("connection_cache_test", connection_manager, 1).unwrap();
761
762        let conn = connection_cache.get_connection(&addr);
763        // We (intentionally) don't have an interface that allows us to distinguish between
764        // UDP and Quic connections, so check instead that the port is valid (non-zero)
765        // and is the same as the input port (falling back on UDP)
766        assert_ne!(port, 0u16);
767        assert_eq!(BlockingClientConnection::server_addr(&*conn).port(), port);
768        assert_eq!(
769            NonblockingClientConnection::server_addr(&*conn).port(),
770            port
771        );
772    }
773}
solana_connection_cache/connection_cache.rs

solana_connection_cache/
connection_cache.rs
