fn handle_ready_connection(&mut self, evt: OpenedConnectionEvent) {
        match evt.result {
            Err(err) => {
                if evt.requested_shard.is_some() {
                    // If we failed to connect to a shard-aware port,
                    // fall back to the non-shard-aware port.
                    // Don't set `had_error_since_last_refill` here;
                    // the shard-aware port might be unreachable, but
                    // the regular port might be reachable. If we set
                    // `had_error_since_last_refill` here, it would cause
                    // the backoff to increase on each refill. With
                    // the non-shard aware port, multiple refills are sometimes
                    // necessary, so increasing the backoff would delay
                    // filling the pool even if the non-shard-aware port works
                    // and does not cause any errors.
                    debug!(
                        "[{}] Failed to open connection to the shard-aware port: {:?}, will retry with regular port",
                        self.address,
                        err,
                    );
                    self.start_opening_connection(None);
                } else {
                    // Encountered an error while connecting to the non-shard-aware
                    // port. Set the `had_error_since_last_refill` flag so that
                    // the next refill will be delayed more than this one.
                    self.had_error_since_last_refill = true;
                    debug!(
                        "[{}] Failed to open connection to the non-shard-aware port: {:?}",
                        self.address, err,
                    );
                }
            }
            Ok((connection, error_receiver)) => {
                // Update sharding and optionally reshard
                let shard_info = connection.get_shard_info().as_ref();
                let sharder = shard_info.map(|s| s.get_sharder());
                let shard_id = shard_info.map_or(0, |s| s.shard as usize);
                self.maybe_reshard(sharder);

                // Update the shard-aware port
                if self.shard_aware_port != connection.get_shard_aware_port() {
                    debug!(
                        "[{}] Updating shard aware port: {:?}",
                        self.address,
                        connection.get_shard_aware_port(),
                    );
                    self.shard_aware_port = connection.get_shard_aware_port();
                }

                // Before the connection can be put to the pool, we need
                // to make sure that it uses appropriate keyspace
                if let Some(keyspace) = &self.current_keyspace {
                    if evt.keyspace_name.as_ref() != Some(keyspace) {
                        // Asynchronously start setting keyspace for this
                        // connection. It will be received on the ready
                        // connections channel and will travel through
                        // this logic again, to be finally put into
                        // the conns.
                        self.start_setting_keyspace_for_connection(
                            connection,
                            error_receiver,
                            evt.requested_shard,
                        );
                        return;
                    }
                }

                // Decide if the connection can be accepted, according to
                // the pool filling strategy
                let can_be_accepted = match self.pool_config.pool_size {
                    PoolSize::PerHost(target) => self.active_connection_count() < target.get(),
                    PoolSize::PerShard(target) => self.conns[shard_id].len() < target.get(),
                };

                if can_be_accepted {
                    // Don't complain and just put the connection to the pool.
                    // If this was a shard-aware port connection which missed
                    // the right shard, we still want to accept it
                    // because it fills our pool.
                    let conn = Arc::new(connection);
                    trace!(
                        "[{}] Adding connection {:p} to shard {} pool, now there are {} for the shard, total {}",
                        self.address,
                        Arc::as_ptr(&conn),
                        shard_id,
                        self.conns[shard_id].len() + 1,
                        self.active_connection_count() + 1,
                    );

                    self.connection_errors
                        .push(wait_for_error(Arc::downgrade(&conn), error_receiver).boxed());
                    self.conns[shard_id].push(conn);

                    self.update_shared_conns();
                } else if evt.requested_shard.is_some() {
                    // This indicates that some shard-aware connections
                    // missed the target shard (probably due to NAT).
                    // Because we don't know how address translation
                    // works here, it's better to leave the task
                    // of choosing the shard to Scylla. We will retry
                    // immediately with a non-shard-aware port here.
                    debug!(
                        "[{}] Excess shard-aware port connection for shard {}; will retry with non-shard-aware port",
                        self.address,
                        shard_id,
                    );

                    self.start_opening_connection(None);
                } else {
                    // We got unlucky and Scylla didn't distribute
                    // shards across connections evenly.
                    // We will retry in the next iteration,
                    // for now put it into the excess connection
                    // pool.
                    let conn = Arc::new(connection);
                    trace!(
                        "[{}] Storing excess connection {:p} for shard {}",
                        self.address,
                        Arc::as_ptr(&conn),
                        shard_id,
                    );

                    self.connection_errors
                        .push(wait_for_error(Arc::downgrade(&conn), error_receiver).boxed());
                    self.excess_connections.push(conn);

                    let excess_connection_limit = self.excess_connection_limit();
                    if self.excess_connections.len() > excess_connection_limit {
                        debug!(
                            "[{}] Excess connection pool exceeded limit of {} connections - clearing",
                            self.address,
                            excess_connection_limit,
                        );
                        self.excess_connections.clear();
                    }
                }
            }
        }
    }
// Handles a newly opened connection and decides what to do with it.
Handles a newly opened connection and decides what to do with it.
Handles a newly opened connection and decides what to do with it .
fn handle_ready_connection(&mut self, evt: OpenedConnectionEvent) {
        match evt.result {
            Err(err) => {
                if evt.requested_shard.is_some() {
                    // If we failed to connect to a shard-aware port,
                    // fall back to the non-shard-aware port.
                    // Don't set `had_error_since_last_refill` here;
                    // the shard-aware port might be unreachable, but
                    // the regular port might be reachable. If we set
                    // `had_error_since_last_refill` here, it would cause
                    // the backoff to increase on each refill. With
                    // the non-shard aware port, multiple refills are sometimes
                    // necessary, so increasing the backoff would delay
                    // filling the pool even if the non-shard-aware port works
                    // and does not cause any errors.
                    debug!(
                        "[{}] Failed to open connection to the shard-aware port: {:?}, will retry with regular port",
                        self.address,
                        err,
                    );
                    self.start_opening_connection(None);
                } else {
                    // Encountered an error while connecting to the non-shard-aware
                    // port. Set the `had_error_since_last_refill` flag so that
                    // the next refill will be delayed more than this one.
                    self.had_error_since_last_refill = true;
                    debug!(
                        "[{}] Failed to open connection to the non-shard-aware port: {:?}",
                        self.address, err,
                    );
                }
            }
            Ok((connection, error_receiver)) => {
                // Update sharding and optionally reshard
                let shard_info = connection.get_shard_info().as_ref();
                let sharder = shard_info.map(|s| s.get_sharder());
                let shard_id = shard_info.map_or(0, |s| s.shard as usize);
                self.maybe_reshard(sharder);
                // Update the shard-aware port
                if self.shard_aware_port != connection.get_shard_aware_port() {
                    debug!(
                        "[{}] Updating shard aware port: {:?}",
                        self.address,
                        connection.get_shard_aware_port(),
                    );
                    self.shard_aware_port = connection.get_shard_aware_port();
                }
                // Before the connection can be put to the pool, we need
                // to make sure that it uses appropriate keyspace
                if let Some(keyspace) = &self.current_keyspace {
                    if evt.keyspace_name.as_ref() != Some(keyspace) {
                        // Asynchronously start setting keyspace for this
                        // connection. It will be received on the ready
                        // connections channel and will travel through
                        // this logic again, to be finally put into
                        // the conns.
                        self.start_setting_keyspace_for_connection(
                            connection,
                            error_receiver,
                            evt.requested_shard,
                        );
                        return;
                    }
                }
                // Decide if the connection can be accepted, according to
                // the pool filling strategy
                let can_be_accepted = match self.pool_config.pool_size {
                    PoolSize::PerHost(target) => self.active_connection_count() < target.get(),
                    PoolSize::PerShard(target) => self.conns[shard_id].len() < target.get(),
                };
                if can_be_accepted {
                    // Don't complain and just put the connection to the pool.
                    // If this was a shard-aware port connection which missed
                    // the right shard, we still want to accept it
                    // because it fills our pool.
                    let conn = Arc::new(connection);
                    trace!(
                        "[{}] Adding connection {:p} to shard {} pool, now there are {} for the shard, total {}",
                        self.address,
                        Arc::as_ptr(&conn),
                        shard_id,
                        self.conns[shard_id].len() + 1,
                        self.active_connection_count() + 1,
                    );
                    self.connection_errors
                        .push(wait_for_error(Arc::downgrade(&conn), error_receiver).boxed());
                    self.conns[shard_id].push(conn);
                    self.update_shared_conns();
                } else if evt.requested_shard.is_some() {
                    // This indicates that some shard-aware connections
                    // missed the target shard (probably due to NAT).
                    // Because we don't know how address translation
                    // works here, it's better to leave the task
                    // of choosing the shard to Scylla. We will retry
                    // immediately with a non-shard-aware port here.
                    debug!(
                        "[{}] Excess shard-aware port connection for shard {}; will retry with non-shard-aware port",
                        self.address,
                        shard_id,
                    );
                    self.start_opening_connection(None);
                } else {
                    // We got unlucky and Scylla didn't distribute
                    // shards across connections evenly.
                    // We will retry in the next iteration,
                    // for now put it into the excess connection
                    // pool.
                    let conn = Arc::new(connection);
                    trace!(
                        "[{}] Storing excess connection {:p} for shard {}",
                        self.address,
                        Arc::as_ptr(&conn),
                        shard_id,
                    );
                    self.connection_errors
                        .push(wait_for_error(Arc::downgrade(&conn), error_receiver).boxed());
                    self.excess_connections.push(conn);
                    let excess_connection_limit = self.excess_connection_limit();
                    if self.excess_connections.len() > excess_connection_limit {
                        debug!(
                            "[{}] Excess connection pool exceeded limit of {} connections - clearing",
                            self.address,
                            excess_connection_limit,
                        );
                        self.excess_connections.clear();
                    }
                }
            }
        }
    }
fn handle_ready_connection ( & mut self , evt : OpenedConnectionEvent ) { match evt . result { Err ( err ) => { if evt . requested_shard . is_some ( ) { // If we failed to connect to a shard-aware port, // fall back to the non-shard-aware port. // Don't set `had_error_since_last_refill` here; // the shard-aware port might be unreachable, but // the regular port might be reachable. If we set // `had_error_since_last_refill` here, it would cause // the backoff to increase on each refill. With // the non-shard aware port, multiple refills are sometimes // necessary, so increasing the backoff would delay // filling the pool even if the non-shard-aware port works // and does not cause any errors. debug ! ( " " self address err ) ; self . start_opening_connection ( None ) ; } else { // Encountered an error while connecting to the non-shard-aware // port. Set the `had_error_since_last_refill` flag so that // the next refill will be delayed more than this one. self . had_error_since_last_refill = true ; debug ! ( " " self address err ) ; } } Ok ( ( connection , error_receiver ) ) => { // Update sharding and optionally reshard let shard_info = connection . get_shard_info ( ) . as_ref ( ) ; let sharder = shard_info . map ( | s | s . get_sharder ( ) ) ; let shard_id = shard_info . map_or ( 0 , | s | s . shard as usize ) ; self . maybe_reshard ( sharder ) ; // Update the shard-aware port if self . shard_aware_port != connection . get_shard_aware_port ( ) { debug ! ( " " self address connection get_shard_aware_port ( ) ) ; self . shard_aware_port = connection . get_shard_aware_port ( ) ; } // Before the connection can be put to the pool, we need // to make sure that it uses appropriate keyspace if let Some ( keyspace ) = & self . current_keyspace { if evt . keyspace_name . as_ref ( ) != Some ( keyspace ) { // Asynchronously start setting keyspace for this // connection. It will be received on the ready // connections channel and will travel through // this logic again, to be finally put into // the conns. self . start_setting_keyspace_for_connection ( connection , error_receiver , evt . requested_shard , ) ; return ; } } // Decide if the connection can be accepted, according to // the pool filling strategy let can_be_accepted = match self . pool_config . pool_size { PoolSize :: PerHost ( target ) => self . active_connection_count ( ) < target . get ( ) , PoolSize :: PerShard ( target ) => self . conns [ shard_id ] . len ( ) < target . get ( ) , } ; if can_be_accepted { // Don't complain and just put the connection to the pool. // If this was a shard-aware port connection which missed // the right shard, we still want to accept it // because it fills our pool. let conn = Arc :: new ( connection ) ; trace ! ( " " self address Arc as_ptr ( conn ) shard_id self conns [ shard_id ] len ( ) 1 self active_connection_count ( ) 1 ) ; self . connection_errors . push ( wait_for_error ( Arc :: downgrade ( & conn ) , error_receiver ) . boxed ( ) ) ; self . conns [ shard_id ] . push ( conn ) ; self . update_shared_conns ( ) ; } else if evt . requested_shard . is_some ( ) { // This indicates that some shard-aware connections // missed the target shard (probably due to NAT). // Because we don't know how address translation // works here, it's better to leave the task // of choosing the shard to Scylla. We will retry // immediately with a non-shard-aware port here. debug ! ( " " self address shard_id ) ; self . start_opening_connection ( None ) ; } else { // We got unlucky and Scylla didn't distribute // shards across connections evenly. // We will retry in the next iteration, // for now put it into the excess connection // pool. let conn = Arc :: new ( connection ) ; trace ! ( " " self address Arc as_ptr ( conn ) shard_id ) ; self . connection_errors . push ( wait_for_error ( Arc :: downgrade ( & conn ) , error_receiver ) . boxed ( ) ) ; self . excess_connections . push ( conn ) ; let excess_connection_limit = self . excess_connection_limit ( ) ; if self . excess_connections . len ( ) > excess_connection_limit { debug ! ( " " self address excess_connection_limit ) ; self . excess_connections . clear ( ) ; } } } } }
Handles a newly opened connection and decides what to do with it.
Handles a newly opened connection and decides what to do with it .
&mut self
evt
OpenedConnectionEvent
&mut self
evt
OpenedConnectionEvent