pub async fn run(
        mut self,
        mut use_keyspace_request_receiver: mpsc::Receiver<UseKeyspaceRequest>,
    ) {
        debug!("[{}] Started asynchronous pool worker", self.address);

        let mut next_refill_time = tokio::time::Instant::now();
        let mut refill_scheduled = true;

        loop {
            tokio::select! {
                _ = tokio::time::sleep_until(next_refill_time), if refill_scheduled => {
                    self.had_error_since_last_refill = false;
                    self.start_filling();
                    refill_scheduled = false;
                }

                evt = self.ready_connections.select_next_some(), if !self.ready_connections.is_empty() => {
                    self.handle_ready_connection(evt);

                    if self.is_full() {
                        debug!(
                            "[{}] Pool is full, clearing {} excess connections",
                            self.address,
                            self.excess_connections.len()
                        );
                        self.excess_connections.clear();
                    }
                }

                evt = self.connection_errors.select_next_some(), if !self.connection_errors.is_empty() => {
                    if let Some(conn) = evt.connection.upgrade() {
                        debug!("[{}] Got error for connection {:p}: {:?}", self.address, Arc::as_ptr(&conn), evt.error);
                        self.remove_connection(conn);
                    }
                }

                req = use_keyspace_request_receiver.recv() => {
                    if let Some(req) = req {
                        debug!("[{}] Requested keyspace change: {}", self.address, req.keyspace_name.as_str());
                        self.use_keyspace(&req.keyspace_name, req.response_sender);
                    } else {
                        // The keyspace request channel is dropped.
                        // This means that the corresponding pool is dropped.
                        // We can stop here.
                        trace!("[{}] Keyspace request channel dropped, stopping asynchronous pool worker", self.address);
                        return;
                    }
                }
            }

            // Schedule refilling here
            if !refill_scheduled && self.need_filling() {
                // Update shared_conns here even if there are no connections.
                // This will signal the waiters in `wait_until_initialized`.
                self.update_shared_conns();
                if self.had_error_since_last_refill {
                    self.refill_delay_strategy.on_fill_error();
                } else {
                    self.refill_delay_strategy.on_successful_fill();
                }
                let delay = self.refill_delay_strategy.get_delay();
                debug!(
                    "[{}] Scheduling next refill in {} ms",
                    self.address,
                    delay.as_millis(),
                );

                next_refill_time = tokio::time::Instant::now() + delay;
                refill_scheduled = true;
            }
        }
    }
// The main loop of the pool refiller
The main loop of the pool refiller
The main loop of the pool refiller
pub async fn run(
        mut self,
        mut use_keyspace_request_receiver: mpsc::Receiver<UseKeyspaceRequest>,
    ) {
        debug!("[{}] Started asynchronous pool worker", self.address);
        let mut next_refill_time = tokio::time::Instant::now();
        let mut refill_scheduled = true;
        loop {
            tokio::select! {
                _ = tokio::time::sleep_until(next_refill_time), if refill_scheduled => {
                    self.had_error_since_last_refill = false;
                    self.start_filling();
                    refill_scheduled = false;
                }
                evt = self.ready_connections.select_next_some(), if !self.ready_connections.is_empty() => {
                    self.handle_ready_connection(evt);
                    if self.is_full() {
                        debug!(
                            "[{}] Pool is full, clearing {} excess connections",
                            self.address,
                            self.excess_connections.len()
                        );
                        self.excess_connections.clear();
                    }
                }
                evt = self.connection_errors.select_next_some(), if !self.connection_errors.is_empty() => {
                    if let Some(conn) = evt.connection.upgrade() {
                        debug!("[{}] Got error for connection {:p}: {:?}", self.address, Arc::as_ptr(&conn), evt.error);
                        self.remove_connection(conn);
                    }
                }
                req = use_keyspace_request_receiver.recv() => {
                    if let Some(req) = req {
                        debug!("[{}] Requested keyspace change: {}", self.address, req.keyspace_name.as_str());
                        self.use_keyspace(&req.keyspace_name, req.response_sender);
                    } else {
                        // The keyspace request channel is dropped.
                        // This means that the corresponding pool is dropped.
                        // We can stop here.
                        trace!("[{}] Keyspace request channel dropped, stopping asynchronous pool worker", self.address);
                        return;
                    }
                }
            }
            // Schedule refilling here
            if !refill_scheduled && self.need_filling() {
                // Update shared_conns here even if there are no connections.
                // This will signal the waiters in `wait_until_initialized`.
                self.update_shared_conns();
                if self.had_error_since_last_refill {
                    self.refill_delay_strategy.on_fill_error();
                } else {
                    self.refill_delay_strategy.on_successful_fill();
                }
                let delay = self.refill_delay_strategy.get_delay();
                debug!(
                    "[{}] Scheduling next refill in {} ms",
                    self.address,
                    delay.as_millis(),
                );
                next_refill_time = tokio::time::Instant::now() + delay;
                refill_scheduled = true;
            }
        }
    }
pub async fn run ( mut self , mut use_keyspace_request_receiver : mpsc :: Receiver < UseKeyspaceRequest > , ) { debug ! ( " " self address ) ; let mut next_refill_time = tokio :: time :: Instant :: now ( ) ; let mut refill_scheduled = true ; loop { tokio :: select ! { tokio time sleep_until ( next_refill_time ) if refill_scheduled { self had_error_since_last_refill false self start_filling ( ) refill_scheduled false } evt self ready_connections select_next_some ( ) if self ready_connections is_empty ( ) { self handle_ready_connection ( evt ) if self is_full ( ) { debug ( " " self address self excess_connections len ( ) ) self excess_connections clear ( ) } } evt self connection_errors select_next_some ( ) if self connection_errors is_empty ( ) { if let Some ( conn ) evt connection upgrade ( ) { debug ( " " self address Arc as_ptr ( conn ) evt error ) self remove_connection ( conn ) } } req use_keyspace_request_receiver recv ( ) { if let Some ( req ) req { debug ( " " self address req keyspace_name as_str ( ) ) self use_keyspace ( req keyspace_name req response_sender ) } else { // The keyspace request channel is dropped. // This means that the corresponding pool is dropped. // We can stop here. trace ( " " self address ) return } } } // Schedule refilling here if ! refill_scheduled && self . need_filling ( ) { // Update shared_conns here even if there are no connections. // This will signal the waiters in `wait_until_initialized`. self . update_shared_conns ( ) ; if self . had_error_since_last_refill { self . refill_delay_strategy . on_fill_error ( ) ; } else { self . refill_delay_strategy . on_successful_fill ( ) ; } let delay = self . refill_delay_strategy . get_delay ( ) ; debug ! ( " " self address delay as_millis ( ) ) ; next_refill_time = tokio :: time :: Instant :: now ( ) + delay ; refill_scheduled = true ; } } }
The main loop of the pool refiller
The main loop of the pool refiller
mut self
use_keyspace_request_receiver
mpsc::Receiver<UseKeyspaceRequest>
mut self
use_keyspace_request_receiver
mpsc::Receiver<UseKeyspaceRequest>