AWS SDK

AWS SDK

rev. 174400987dccd7e137fefa96b1143d21c7ddfb78 (ignoring whitespace)

Files changed:

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/stats.rs

@@ -0,1 +0,640 @@
           1  +
/*
           2  +
 * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
           3  +
 * SPDX-License-Identifier: Apache-2.0
           4  +
 */
           5  +
           6  +
//! Per-(partition, authority) connection counters and a pool-level inverted index.
           7  +
//!
           8  +
//! Counters are maintained with `Relaxed` atomics. A snapshot observes each counter
           9  +
//! independently; transient mutual inconsistency is expected (e.g. `active` may
          10  +
//! briefly exceed `established`). The index never blocks the connection hot path.
          11  +
          12  +
use std::collections::HashMap;
          13  +
use std::sync::atomic::{AtomicU8, AtomicUsize, Ordering};
          14  +
use std::sync::{Arc, Weak};
          15  +
          16  +
use super::connection::Authority;
          17  +
use super::partition::PartitionId;
          18  +
          19  +
// Protocol-tag encoding for the cell. A cell is per-(partition, authority);
          20  +
// protocol is per-connection, so a cell CAN observe both over its life
          21  +
// (multi-endpoint authority, server reconfig, proxy). The tag latches to
          22  +
// MIXED once it sees two different protocols and never leaves it, so
          23  +
// `capacity_hint` only answers `Some` for a uniformly-H1 cell and never
          24  +
// overstates reusable capacity.
          25  +
pub(crate) const PROTO_UNSET: u8 = 0;
          26  +
pub(crate) const PROTO_H1: u8 = 1;
          27  +
pub(crate) const PROTO_H2: u8 = 2;
          28  +
const PROTO_MIXED: u8 = 3;
          29  +
          30  +
/// Per-(partition, authority) connection counts maintained with `Relaxed` atomics.
          31  +
///
          32  +
/// Each counter is loaded independently; concurrent reads may observe transiently
          33  +
/// inconsistent combinations (e.g. `active` > `established`). Intended for heuristic
          34  +
/// reads that tolerate stale or momentarily inconsistent values.
          35  +
#[derive(Debug, Default)]
          36  +
pub(crate) struct ConnectionCounters {
          37  +
    /// Connections that have completed handshake and exist (idle + active).
          38  +
    pub(crate) established: AtomicUsize,
          39  +
    /// Handshakes in flight.
          40  +
    pub(crate) establishing: AtomicUsize,
          41  +
    /// Connections/streams currently checked out.
          42  +
    pub(crate) active: AtomicUsize,
          43  +
    /// Protocol tag for this cell (monotonic toward MIXED).
          44  +
    protocol: AtomicU8,
          45  +
}
          46  +
          47  +
/// Tracks a connection committed to handshaking (TCP + TLS + protocol).
          48  +
///
          49  +
/// Construction increments `establishing`. [`promote`](Self::promote) transitions to
          50  +
/// an [`EstablishedGuard`] (established++ then establishing--) on success; any other
          51  +
/// drop path (failure, cancel, panic) decrements `establishing`. Exactly-once by
          52  +
/// construction: `promote` consumes `self`.
          53  +
pub(crate) struct EstablishingGuard {
          54  +
    counters: Arc<ConnectionCounters>,
          55  +
    promoted: bool,
          56  +
}
          57  +
          58  +
impl EstablishingGuard {
          59  +
    pub(crate) fn new(counters: Arc<ConnectionCounters>) -> Self {
          60  +
        counters.establishing.fetch_add(1, Ordering::Relaxed);
          61  +
        Self {
          62  +
            counters,
          63  +
            promoted: false,
          64  +
        }
          65  +
    }
          66  +
          67  +
    /// Handshake succeeded: transition establishing → established.
          68  +
    ///
          69  +
    /// `established` is incremented BEFORE `establishing` is decremented so a
          70  +
    /// concurrent reader may observe a transient overcount. The overcount is in the
          71  +
    /// direction of over-reporting readiness, never under-reporting.
          72  +
    pub(crate) fn promote(mut self, proto: u8) -> EstablishedGuard {
          73  +
        self.counters.observe_protocol(proto);
          74  +
        self.counters.established.fetch_add(1, Ordering::Relaxed);
          75  +
        self.counters.establishing.fetch_sub(1, Ordering::Relaxed);
          76  +
        self.promoted = true;
          77  +
        EstablishedGuard {
          78  +
            counters: self.counters.clone(),
          79  +
        }
          80  +
    }
          81  +
}
          82  +
          83  +
impl Drop for EstablishingGuard {
          84  +
    fn drop(&mut self) {
          85  +
        if !self.promoted {
          86  +
            self.counters.establishing.fetch_sub(1, Ordering::Relaxed);
          87  +
        }
          88  +
    }
          89  +
}
          90  +
          91  +
/// Owns one connection's contribution to `established`. Held on the
          92  +
/// Arc-shared inner of a `ManagedConnection`, so for H2 (N clones share
          93  +
/// one connection) it fires `established--` exactly once — when the last
          94  +
/// clone drops. Non-`Clone` by design: single ownership is compiler-
          95  +
/// enforced.
          96  +
pub(crate) struct EstablishedGuard {
          97  +
    counters: Arc<ConnectionCounters>,
          98  +
}
          99  +
         100  +
impl Drop for EstablishedGuard {
         101  +
    fn drop(&mut self) {
         102  +
        self.counters.established.fetch_sub(1, Ordering::Relaxed);
         103  +
    }
         104  +
}
         105  +
         106  +
impl ConnectionCounters {
         107  +
    /// Increment the active (checked-out) count. Paired with exactly one
         108  +
    /// [`decr_active`](Self::decr_active) via an RAII checkout guard.
         109  +
    pub(crate) fn incr_active(&self) {
         110  +
        self.active.fetch_add(1, Ordering::Relaxed);
         111  +
    }
         112  +
         113  +
    /// Decrement the active count.
         114  +
    ///
         115  +
    /// Every `incr_active` is paired with exactly one `decr_active` via RAII checkout
         116  +
    /// guards, so `active` is non-negative by construction. A saturating sub here
         117  +
    /// would mask a broken-pairing bug; saturation belongs only on the cross-atomic
         118  +
    /// READ (`idle() = established.saturating_sub(active)`) where two independent
         119  +
    /// relaxed loads may transiently cross.
         120  +
    pub(crate) fn decr_active(&self) {
         121  +
        let prev = self.active.fetch_sub(1, Ordering::Relaxed);
         122  +
        debug_assert!(prev > 0, "active underflow: decr without matching incr");
         123  +
    }
         124  +
         125  +
    /// Record the negotiated protocol for a connection in this cell.
         126  +
    ///
         127  +
    /// Latches monotonically toward `MIXED`: `UNSET` → observed protocol; same
         128  +
    /// protocol → no-op; different protocol → `MIXED` (terminal). A cell reaches
         129  +
    /// `MIXED` when one authority negotiates differently across connections
         130  +
    /// (multi-endpoint DNS, server reconfiguration, an intermediary).
         131  +
    ///
         132  +
    /// The sole purpose of this tag is to keep `capacity_hint` from overstating:
         133  +
    /// `capacity_hint` returns `Some` only for a cell known to be uniformly one
         134  +
    /// protocol it can reason about (HTTP/1). Per-connection protocol and
         135  +
    /// stream-limit accounting would supersede this cell-level latch.
         136  +
    pub(crate) fn observe_protocol(&self, proto: u8) {
         137  +
        let mut cur = self.protocol.load(Ordering::Relaxed);
         138  +
        loop {
         139  +
            let next = match cur {
         140  +
                PROTO_UNSET => proto,
         141  +
                c if c == proto => return,
         142  +
                PROTO_MIXED => return,
         143  +
                _ => PROTO_MIXED,
         144  +
            };
         145  +
            match self.protocol.compare_exchange_weak(
         146  +
                cur,
         147  +
                next,
         148  +
                Ordering::Relaxed,
         149  +
                Ordering::Relaxed,
         150  +
            ) {
         151  +
                Ok(_) => return,
         152  +
                Err(actual) => cur = actual,
         153  +
            }
         154  +
        }
         155  +
    }
         156  +
         157  +
    /// Current protocol tag for this cell: `PROTO_UNSET` before the first
         158  +
    /// handshake, `PROTO_H1`/`PROTO_H2` for a uniform cell, or the internal
         159  +
    /// mixed marker once a cell has seen both.
         160  +
    pub(crate) fn protocol(&self) -> u8 {
         161  +
        self.protocol.load(Ordering::Relaxed)
         162  +
    }
         163  +
}
         164  +
         165  +
/// One authority's row in the inverted index: per-partition weak references to counters.
         166  +
#[derive(Debug, Default)]
         167  +
pub(crate) struct AuthorityCounters {
         168  +
    pub(crate) by_partition: HashMap<PartitionId, Weak<ConnectionCounters>>,
         169  +
}
         170  +
         171  +
/// Pool-level inverted index: authority → per-partition counters.
         172  +
///
         173  +
/// Non-owning projection: holds `Weak<ConnectionCounters>` references. Strong owners
         174  +
/// (pool entries, live checkouts) keep cells alive; once all strong references drop,
         175  +
/// the `Weak` goes dead and is pruned on the next `snapshot`. The index never extends
         176  +
/// connection lifetime and never blocks the per-request hot path.
         177  +
#[derive(Debug, Default)]
         178  +
pub(crate) struct StatsIndex {
         179  +
    inner: std::sync::Mutex<HashMap<Authority, AuthorityCounters>>,
         180  +
}
         181  +
         182  +
impl StatsIndex {
         183  +
    /// Register a cell's counters at first-touch of (authority, partition).
         184  +
    /// Stores a `Weak` reference; the caller retains the strong `Arc`.
         185  +
    /// Idempotent: re-registration overwrites the previous entry.
         186  +
    pub(crate) fn register(
         187  +
        &self,
         188  +
        authority: Authority,
         189  +
        partition: PartitionId,
         190  +
        counters: &Arc<ConnectionCounters>,
         191  +
    ) {
         192  +
        let mut idx = self.inner.lock().expect("stats index poisoned");
         193  +
        idx.entry(authority)
         194  +
            .or_default()
         195  +
            .by_partition
         196  +
            .insert(partition, Arc::downgrade(counters));
         197  +
    }
         198  +
         199  +
    /// Drop the `(authority, partition)` cell if its counters are no longer
         200  +
    /// referenced, removing the authority entry when its last partition is
         201  +
    /// pruned. A no-op if the cell is still strongly held (e.g. a checkout
         202  +
    /// is in flight when the host's idle connections are evicted), so it is
         203  +
    /// safe to call from the eviction path. Reconstructs nothing the caller
         204  +
    /// does not already hold.
         205  +
    pub(crate) fn prune_if_dead(&self, authority: &Authority, partition: PartitionId) {
         206  +
        let mut idx = self.inner.lock().expect("stats index poisoned");
         207  +
        if let Some(a) = idx.get_mut(authority) {
         208  +
            if a.by_partition
         209  +
                .get(&partition)
         210  +
                .is_some_and(|w| w.strong_count() == 0)
         211  +
            {
         212  +
                a.by_partition.remove(&partition);
         213  +
            }
         214  +
            if a.by_partition.is_empty() {
         215  +
                idx.remove(authority);
         216  +
            }
         217  +
        }
         218  +
    }
         219  +
         220  +
    #[cfg(test)]
         221  +
    pub(crate) fn len(&self) -> usize {
         222  +
        self.inner.lock().expect("stats index poisoned").len()
         223  +
    }
         224  +
         225  +
    #[cfg(test)]
         226  +
    pub(crate) fn established_for(&self, authority: &Authority, partition: PartitionId) -> usize {
         227  +
        let idx = self.inner.lock().expect("stats index poisoned");
         228  +
        idx.get(authority)
         229  +
            .and_then(|a| a.by_partition.get(&partition))
         230  +
            .and_then(|w| w.upgrade())
         231  +
            .map(|c| c.established.load(Ordering::Relaxed))
         232  +
            .unwrap_or(0)
         233  +
    }
         234  +
         235  +
    #[cfg(test)]
         236  +
    pub(crate) fn establishing_for(&self, authority: &Authority, partition: PartitionId) -> usize {
         237  +
        let idx = self.inner.lock().expect("stats index poisoned");
         238  +
        idx.get(authority)
         239  +
            .and_then(|a| a.by_partition.get(&partition))
         240  +
            .and_then(|w| w.upgrade())
         241  +
            .map(|c| c.establishing.load(Ordering::Relaxed))
         242  +
            .unwrap_or(0)
         243  +
    }
         244  +
         245  +
    /// Partitions with at least one idle connection to `authority`, as
         246  +
    /// `(partition, idle_count)`. Advisory: a relaxed snapshot that
         247  +
    /// *narrows* reclaim/borrow candidates — the cache pop is the
         248  +
    /// authoritative confirmation. Prunes dead `Weak`s under the lock,
         249  +
    /// same as [`Self::snapshot`].
         250  +
    pub(crate) fn idle_partitions_for(&self, authority: &Authority) -> Vec<(PartitionId, usize)> {
         251  +
        let handles: Vec<(PartitionId, Arc<ConnectionCounters>)> = {
         252  +
            let mut idx = self.inner.lock().expect("stats index poisoned");
         253  +
            match idx.get_mut(authority) {
         254  +
                Some(a) => {
         255  +
                    let mut live = Vec::new();
         256  +
                    a.by_partition.retain(|partition, weak| {
         257  +
                        if let Some(strong) = weak.upgrade() {
         258  +
                            live.push((*partition, strong));
         259  +
                            true
         260  +
                        } else {
         261  +
                            false
         262  +
                        }
         263  +
                    });
         264  +
                    if a.by_partition.is_empty() {
         265  +
                        idx.remove(authority);
         266  +
                    }
         267  +
                    live
         268  +
                }
         269  +
                None => Vec::new(),
         270  +
            }
         271  +
        }; // lock released before loading atomics
         272  +
        handles
         273  +
            .into_iter()
         274  +
            .filter_map(|(p, c)| {
         275  +
                let established = c.established.load(Ordering::Relaxed);
         276  +
                let active = c.active.load(Ordering::Relaxed);
         277  +
                let idle = established.saturating_sub(active);
         278  +
                (idle > 0).then_some((p, idle))
         279  +
            })
         280  +
            .collect()
         281  +
    }
         282  +
         283  +
    /// All `(authority, partition)` cells with at least one idle
         284  +
    /// connection. Drives `Global`-constraint reclaim, where a freed
         285  +
    /// permit is fungible across authorities. Advisory/narrowing, same
         286  +
    /// contract as [`Self::idle_partitions_for`].
         287  +
    pub(crate) fn idle_cells(&self) -> Vec<(Authority, PartitionId)> {
         288  +
        let handles: Vec<(Authority, PartitionId, Arc<ConnectionCounters>)> = {
         289  +
            let mut idx = self.inner.lock().expect("stats index poisoned");
         290  +
            let mut live = Vec::new();
         291  +
            idx.retain(|authority, a| {
         292  +
                a.by_partition.retain(|partition, weak| {
         293  +
                    if let Some(strong) = weak.upgrade() {
         294  +
                        live.push((authority.clone(), *partition, strong));
         295  +
                        true
         296  +
                    } else {
         297  +
                        false
         298  +
                    }
         299  +
                });
         300  +
                !a.by_partition.is_empty()
         301  +
            });
         302  +
            live
         303  +
        }; // lock released before loading atomics
         304  +
        handles
         305  +
            .into_iter()
         306  +
            .filter_map(|(authority, p, c)| {
         307  +
                let established = c.established.load(Ordering::Relaxed);
         308  +
                let active = c.active.load(Ordering::Relaxed);
         309  +
                (established.saturating_sub(active) > 0).then_some((authority, p))
         310  +
            })
         311  +
            .collect()
         312  +
    }
         313  +
         314  +
    /// Snapshot one authority's per-partition counters.
         315  +
    ///
         316  +
    /// Under the lock: upgrades each `Weak`, prunes dead entries (removing the
         317  +
    /// authority entirely if all its partitions are dead). Releases the lock before
         318  +
    /// loading atomics into the returned snapshot.
         319  +
    pub(crate) fn snapshot(&self, authority: &Authority) -> AuthorityStats {
         320  +
        let handles: Vec<(PartitionId, Arc<ConnectionCounters>)> = {
         321  +
            let mut idx = self.inner.lock().expect("stats index poisoned");
         322  +
            match idx.get_mut(authority) {
         323  +
                Some(a) => {
         324  +
                    let mut live = Vec::new();
         325  +
                    a.by_partition.retain(|partition, weak| {
         326  +
                        if let Some(strong) = weak.upgrade() {
         327  +
                            live.push((*partition, strong));
         328  +
                            true
         329  +
                        } else {
         330  +
                            false
         331  +
                        }
         332  +
                    });
         333  +
                    if a.by_partition.is_empty() {
         334  +
                        idx.remove(authority);
         335  +
                    }
         336  +
                    live
         337  +
                }
         338  +
                None => Vec::new(),
         339  +
            }
         340  +
        }; // lock released here
         341  +
        let by_partition = handles
         342  +
            .into_iter()
         343  +
            .map(|(p, c)| {
         344  +
                (
         345  +
                    p,
         346  +
                    PartitionStats {
         347  +
                        established: c.established.load(Ordering::Relaxed),
         348  +
                        establishing: c.establishing.load(Ordering::Relaxed),
         349  +
                        active: c.active.load(Ordering::Relaxed),
         350  +
                        protocol: c.protocol(),
         351  +
                    },
         352  +
                )
         353  +
            })
         354  +
            .collect();
         355  +
        AuthorityStats { by_partition }
         356  +
    }
         357  +
}
         358  +
         359  +
/// Point-in-time snapshot of one (partition, authority) cell's connection counts.
         360  +
///
         361  +
/// Plain `usize` values from relaxed loads; cheap and `Copy`. Each field is loaded
         362  +
/// independently and may be transiently inconsistent with the others.
         363  +
/// `#[non_exhaustive]` allows fields to be added without a breaking change.
         364  +
#[non_exhaustive]
         365  +
#[derive(Debug, Clone, Copy)]
         366  +
pub struct PartitionStats {
         367  +
    /// Connections that have completed handshake (idle + active).
         368  +
    pub established: usize,
         369  +
    /// Handshakes in flight.
         370  +
    pub establishing: usize,
         371  +
    /// Connections/streams currently checked out.
         372  +
    pub active: usize,
         373  +
    // Private: protocol tag for capacity_hint. Not public — implementation
         374  +
    // detail of the hint that would otherwise freeze an internal encoding
         375  +
    // into the API.
         376  +
    protocol: u8,
         377  +
}
         378  +
         379  +
impl PartitionStats {
         380  +
    /// Connections not currently checked out: `established.saturating_sub(active)`.
         381  +
    ///
         382  +
    /// Exact for HTTP/1 (one stream per connection). For HTTP/2 a positive value
         383  +
    /// means connections with no active streams; saturated multiplexed connections
         384  +
    /// contribute 0. The saturating subtraction handles transient inconsistency
         385  +
    /// between the two independently-loaded atomics.
         386  +
    pub fn idle(&self) -> usize {
         387  +
        self.established.saturating_sub(self.active)
         388  +
    }
         389  +
         390  +
    /// Spare stream capacity, if determinable from current state.
         391  +
    ///
         392  +
    /// - Uniformly HTTP/1 cell: `Some(idle)` (one stream per connection).
         393  +
    /// - HTTP/2, mixed-protocol, or not-yet-handshaken cell: `None` (per-connection
         394  +
    ///   stream limits are not indexed).
         395  +
    ///
         396  +
    /// When `Some`, still a relaxed snapshot — treat as a hint.
         397  +
    pub fn capacity_hint(&self) -> Option<usize> {
         398  +
        match self.protocol {
         399  +
            PROTO_H1 => Some(self.idle()),
         400  +
            _ => None,
         401  +
        }
         402  +
    }
         403  +
}
         404  +
         405  +
/// Point-in-time, per-partition snapshot of connection counts for one authority.
         406  +
///
         407  +
/// Sparse: only partitions that have opened a connection to the authority appear.
         408  +
/// Returned by [`super::SharedPool::stats`].
         409  +
pub struct AuthorityStats {
         410  +
    by_partition: Vec<(PartitionId, PartitionStats)>,
         411  +
}
         412  +
         413  +
impl AuthorityStats {
         414  +
    /// Stats for a specific partition, if it has opened a connection to this authority.
         415  +
    pub fn get(&self, partition: PartitionId) -> Option<PartitionStats> {
         416  +
        self.by_partition
         417  +
            .iter()
         418  +
            .find(|(p, _)| *p == partition)
         419  +
            .map(|(_, s)| *s)
         420  +
    }
         421  +
         422  +
    /// Iterate (partition, stats) pairs.
         423  +
    pub fn iter(&self) -> impl Iterator<Item = (PartitionId, PartitionStats)> + '_ {
         424  +
        self.by_partition.iter().copied()
         425  +
    }
         426  +
         427  +
    /// Number of partitions that have opened a connection to this authority.
         428  +
    pub fn len(&self) -> usize {
         429  +
        self.by_partition.len()
         430  +
    }
         431  +
         432  +
    /// Whether no partition has connected to this authority.
         433  +
    pub fn is_empty(&self) -> bool {
         434  +
        self.by_partition.is_empty()
         435  +
    }
         436  +
}
         437  +
         438  +
#[cfg(test)]
         439  +
mod tests {
         440  +
    use super::*;
         441  +
         442  +
    #[test]
         443  +
    fn active_incremented_on_checkout_decremented_on_drop() {
         444  +
        let counters = Arc::new(ConnectionCounters::default());
         445  +
        assert_eq!(counters.active.load(Ordering::Relaxed), 0);
         446  +
         447  +
        counters.incr_active();
         448  +
        assert_eq!(counters.active.load(Ordering::Relaxed), 1);
         449  +
         450  +
        counters.decr_active();
         451  +
        assert_eq!(counters.active.load(Ordering::Relaxed), 0);
         452  +
    }
         453  +
         454  +
    #[test]
         455  +
    fn establishing_guard_promote_increments_established_and_clears_establishing() {
         456  +
        let counters = Arc::new(ConnectionCounters::default());
         457  +
         458  +
        let guard = EstablishingGuard::new(counters.clone());
         459  +
        assert_eq!(counters.establishing.load(Ordering::Relaxed), 1);
         460  +
        assert_eq!(counters.established.load(Ordering::Relaxed), 0);
         461  +
         462  +
        let established = guard.promote(PROTO_H1);
         463  +
        assert_eq!(counters.establishing.load(Ordering::Relaxed), 0);
         464  +
        assert_eq!(counters.established.load(Ordering::Relaxed), 1);
         465  +
         466  +
        drop(established);
         467  +
        assert_eq!(counters.established.load(Ordering::Relaxed), 0);
         468  +
    }
         469  +
         470  +
    #[test]
         471  +
    fn establishing_guard_drop_without_promote_decrements() {
         472  +
        let counters = Arc::new(ConnectionCounters::default());
         473  +
         474  +
        let guard = EstablishingGuard::new(counters.clone());
         475  +
        assert_eq!(counters.establishing.load(Ordering::Relaxed), 1);
         476  +
         477  +
        drop(guard);
         478  +
        assert_eq!(counters.establishing.load(Ordering::Relaxed), 0);
         479  +
        assert_eq!(counters.established.load(Ordering::Relaxed), 0);
         480  +
    }
         481  +
         482  +
    #[test]
         483  +
    fn stats_index_registers_counters_arc() {
         484  +
        let index = StatsIndex::default();
         485  +
        assert_eq!(index.len(), 0);
         486  +
         487  +
        let counters = Arc::new(ConnectionCounters::default());
         488  +
        let authority_a = Authority::new("a.example.com:443");
         489  +
        let partition = PartitionId::from_index(0);
         490  +
        index.register(authority_a, partition, &counters);
         491  +
        assert_eq!(index.len(), 1);
         492  +
         493  +
        // Different authority → new entry
         494  +
        let authority_b = Authority::new("b.example.com:443");
         495  +
        let counters2 = Arc::new(ConnectionCounters::default());
         496  +
        index.register(authority_b, partition, &counters2);
         497  +
        assert_eq!(index.len(), 2);
         498  +
         499  +
        // Same authority, different partition → same entry (len unchanged)
         500  +
        let authority_a2 = Authority::new("a.example.com:443");
         501  +
        let counters3 = Arc::new(ConnectionCounters::default());
         502  +
        let partition2 = PartitionId::from_index(1);
         503  +
        index.register(authority_a2, partition2, &counters3);
         504  +
        assert_eq!(index.len(), 2);
         505  +
    }
         506  +
         507  +
    #[test]
         508  +
    fn stats_index_prunes_dead_cells() {
         509  +
        let index = StatsIndex::default();
         510  +
        let authority = Authority::new("ephemeral.example.com:443");
         511  +
        let partition = PartitionId::from_index(0);
         512  +
         513  +
        let counters = Arc::new(ConnectionCounters::default());
         514  +
        index.register(authority.clone(), partition, &counters);
         515  +
        assert_eq!(index.len(), 1);
         516  +
         517  +
        // Drop the only strong reference — the Weak in the index is now dead
         518  +
        drop(counters);
         519  +
         520  +
        // Snapshot triggers pruning; dead cell is removed
         521  +
        let snap = index.snapshot(&authority);
         522  +
        assert!(snap.is_empty());
         523  +
        assert_eq!(index.len(), 0);
         524  +
    }
         525  +
         526  +
    #[test]
         527  +
    fn prune_if_dead_keeps_live_cell_removes_dead_cell() {
         528  +
        let index = StatsIndex::default();
         529  +
        let authority = Authority::new("host.example.com:443");
         530  +
        let partition = PartitionId::from_index(0);
         531  +
        let counters = Arc::new(ConnectionCounters::default());
         532  +
        index.register(authority.clone(), partition, &counters);
         533  +
        assert_eq!(index.len(), 1);
         534  +
         535  +
        // Cell is still strongly held (mirrors a checkout in flight when the
         536  +
        // host's idle connections are evicted): prune is a no-op.
         537  +
        index.prune_if_dead(&authority, partition);
         538  +
        assert_eq!(index.len(), 1);
         539  +
         540  +
        // Strong ref gone (entry + checkouts dropped): prune removes the cell
         541  +
        // and, as its last partition, the authority entry.
         542  +
        drop(counters);
         543  +
        index.prune_if_dead(&authority, partition);
         544  +
        assert_eq!(index.len(), 0);
         545  +
    }
         546  +
         547  +
    #[test]
         548  +
    fn capacity_hint_h1_some_h2_none_mixed_none() {
         549  +
        // H1-only cell: capacity_hint == Some(idle)
         550  +
        let s = PartitionStats {
         551  +
            established: 5,
         552  +
            establishing: 0,
         553  +
            active: 2,
         554  +
            protocol: PROTO_H1,
         555  +
        };
         556  +
        assert_eq!(s.capacity_hint(), Some(3));
         557  +
         558  +
        // H2-only cell: None
         559  +
        let s = PartitionStats {
         560  +
            established: 5,
         561  +
            establishing: 0,
         562  +
            active: 2,
         563  +
            protocol: PROTO_H2,
         564  +
        };
         565  +
        assert_eq!(s.capacity_hint(), None);
         566  +
         567  +
        // UNSET cell: None
         568  +
        let s = PartitionStats {
         569  +
            established: 0,
         570  +
            establishing: 1,
         571  +
            active: 0,
         572  +
            protocol: PROTO_UNSET,
         573  +
        };
         574  +
        assert_eq!(s.capacity_hint(), None);
         575  +
         576  +
        // Mixed cell via observe_protocol transitions: None
         577  +
        let counters = Arc::new(ConnectionCounters::default());
         578  +
        counters.observe_protocol(PROTO_H1);
         579  +
        assert_eq!(counters.protocol(), PROTO_H1);
         580  +
        counters.observe_protocol(PROTO_H2);
         581  +
        assert_eq!(counters.protocol(), PROTO_MIXED);
         582  +
        let s = PartitionStats {
         583  +
            established: 5,
         584  +
            establishing: 0,
         585  +
            active: 2,
         586  +
            protocol: counters.protocol(),
         587  +
        };
         588  +
        assert_eq!(s.capacity_hint(), None);
         589  +
    }
         590  +
         591  +
    #[test]
         592  +
    fn observe_protocol_latches_monotonically_to_mixed() {
         593  +
        // First observation sets the protocol from UNSET.
         594  +
        let c = Arc::new(ConnectionCounters::default());
         595  +
        assert_eq!(c.protocol(), PROTO_UNSET);
         596  +
        c.observe_protocol(PROTO_H1);
         597  +
        assert_eq!(c.protocol(), PROTO_H1);
         598  +
         599  +
        // Same protocol again is a no-op (stays H1, does not advance to MIXED).
         600  +
        c.observe_protocol(PROTO_H1);
         601  +
        assert_eq!(c.protocol(), PROTO_H1);
         602  +
         603  +
        // A different protocol latches to MIXED.
         604  +
        c.observe_protocol(PROTO_H2);
         605  +
        assert_eq!(c.protocol(), PROTO_MIXED);
         606  +
         607  +
        // MIXED is terminal: observing either protocol again cannot un-mix it.
         608  +
        c.observe_protocol(PROTO_H1);
         609  +
        assert_eq!(c.protocol(), PROTO_MIXED);
         610  +
        c.observe_protocol(PROTO_H2);
         611  +
        assert_eq!(c.protocol(), PROTO_MIXED);
         612  +
         613  +
        // The symmetric first-observation path: H2 first, then H1 → MIXED.
         614  +
        let c = Arc::new(ConnectionCounters::default());
         615  +
        c.observe_protocol(PROTO_H2);
         616  +
        assert_eq!(c.protocol(), PROTO_H2);
         617  +
        c.observe_protocol(PROTO_H1);
         618  +
        assert_eq!(c.protocol(), PROTO_MIXED);
         619  +
    }
         620  +
         621  +
    #[test]
         622  +
    fn idle_is_established_minus_active_saturating() {
         623  +
        let s = PartitionStats {
         624  +
            established: 3,
         625  +
            establishing: 0,
         626  +
            active: 1,
         627  +
            protocol: PROTO_H1,
         628  +
        };
         629  +
        assert_eq!(s.idle(), 2);
         630  +
         631  +
        // H2 over-subscription: active > established (multiple streams per conn)
         632  +
        let s = PartitionStats {
         633  +
            established: 1,
         634  +
            establishing: 0,
         635  +
            active: 5,
         636  +
            protocol: PROTO_H2,
         637  +
        };
         638  +
        assert_eq!(s.idle(), 0);
         639  +
    }
         640  +
}

tmp-codegen-diff/aws-sdk/sdk/aws-smithy-http-client/src/client/pool/vendored_cache.rs

@@ -0,1 +0,548 @@
           1  +
/*
           2  +
 * Portions of this file are derived from hyper-util
           3  +
 * (https://github.com/hyperium/hyper-util), licensed under MIT:
           4  +
 *
           5  +
 *   Copyright (c) 2023-2025 Sean McArthur
           6  +
 *
           7  +
 *   Permission is hereby granted, free of charge, to any person obtaining
           8  +
 *   a copy of this software and associated documentation files (the
           9  +
 *   "Software"), to deal in the Software without restriction, including
          10  +
 *   without limitation the rights to use, copy, modify, merge, publish,
          11  +
 *   distribute, sublicense, and/or sell copies of the Software, and to
          12  +
 *   permit persons to whom the Software is furnished to do so, subject
          13  +
 *   to the following conditions:
          14  +
 *
          15  +
 *   The above copyright notice and this permission notice shall be
          16  +
 *   included in all copies or substantial portions of the Software.
          17  +
 *
          18  +
 *   THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
          19  +
 *   EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
          20  +
 *   MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.
          21  +
 *   IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY
          22  +
 *   CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT,
          23  +
 *   TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE
          24  +
 *   SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
          25  +
 *
          26  +
 * Modifications by Amazon.com, Inc. or its affiliates:
          27  +
 *   Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
          28  +
 *   SPDX-License-Identifier: Apache-2.0
          29  +
 *
          30  +
 * The derivative work as a whole is licensed under Apache-2.0 as part of the
          31  +
 * smithy-rs project. The MIT notice above applies to the original portions
          32  +
 * as required by that license.
          33  +
 *
          34  +
 * Source:   hyper-util src/client/pool/cache.rs
          35  +
 * Upstream: https://github.com/hyperium/hyper-util
          36  +
 * Commit:   e1c5a6c89bfaed11fb34bd483fe9ba616f403791
          37  +
 *
          38  +
 * Modifications from upstream:
          39  +
 *   1. Changed `pub use self::internal::builder;` to `pub(crate) use ...`
          40  +
 *      and dropped the three `#[cfg(docsrs)] pub use` lines. The composable
          41  +
 *      pool internals are `pub(crate)` in this crate — no types from this
          42  +
 *      file are exposed in the smithy-rs public API.
          43  +
 *   2. Added `#![allow(dead_code, unreachable_pub)]` so the file can be
          44  +
 *      kept close to upstream even when individual items aren't used yet.
          45  +
 *   3. Dropped the module- and struct-level rustdoc sections that reference
          46  +
 *      "Unnameable" (that rustdoc pattern is specific to hyper-util's public
          47  +
 *      API; these types aren't public here).
          48  +
 *   4. Added `Cached::discard(self)` — consumes self and prevents
          49  +
 *      reinsertion into the pool on drop. See `// SDK MODIFICATION` marker
          50  +
 *      below. Used to drop connections that a caller has learned are bad
          51  +
 *      (poisoned, GOAWAY, etc.) between checkout and `poll_ready`. The
          52  +
 *      upstream API requires the inner service's `poll_ready` to error
          53  +
 *      in order to skip reinsertion, which overloads `poll_ready` semantics.
          54  +
 */
          55  +
          56  +
//! A cache of services
          57  +
//!
          58  +
//! The cache is a single list of cached services, bundled with a `MakeService`.
          59  +
//! Calling the cache returns either an existing service, or makes a new one.
          60  +
//! The returned `impl Service` can be used to send requests, and when dropped,
          61  +
//! it will try to be returned back to the cache.
          62  +
          63  +
#![allow(dead_code, unreachable_pub)]
          64  +
          65  +
pub(crate) use self::internal::{builder, Cached};
          66  +
          67  +
mod internal {
          68  +
    use std::fmt;
          69  +
    use std::future::Future;
          70  +
    use std::pin::Pin;
          71  +
    use std::sync::{Arc, Mutex, Weak};
          72  +
    use std::task::{self, ready, Poll};
          73  +
          74  +
    use futures_util::future;
          75  +
    use tokio::sync::oneshot;
          76  +
    use tower_service::Service;
          77  +
          78  +
    use super::events;
          79  +
          80  +
    /// Start a builder to construct a `Cache` pool.
          81  +
    pub fn builder() -> Builder<events::Ignore> {
          82  +
        Builder {
          83  +
            events: events::Ignore,
          84  +
        }
          85  +
    }
          86  +
          87  +
    /// A cache pool of services from the inner make service.
          88  +
    #[derive(Debug)]
          89  +
    pub struct Cache<M, Dst, Ev>
          90  +
    where
          91  +
        M: Service<Dst>,
          92  +
    {
          93  +
        connector: M,
          94  +
        shared: Arc<Mutex<Shared<M::Response>>>,
          95  +
        events: Ev,
          96  +
    }
          97  +
          98  +
    /// A builder to configure a `Cache`.
          99  +
    #[derive(Debug)]
         100  +
    pub struct Builder<Ev> {
         101  +
        events: Ev,
         102  +
    }
         103  +
         104  +
    /// A cached service returned from a [`Cache`].
         105  +
    ///
         106  +
    /// Implements `Service` by delegating to the inner service. Once dropped,
         107  +
    /// tries to reinsert into the `Cache`.
         108  +
    pub struct Cached<S> {
         109  +
        is_closed: bool,
         110  +
        inner: Option<S>,
         111  +
        shared: Weak<Mutex<Shared<S>>>,
         112  +
        // todo: on_idle
         113  +
    }
         114  +
         115  +
    pub enum CacheFuture<M, Dst, Ev>
         116  +
    where
         117  +
        M: Service<Dst>,
         118  +
    {
         119  +
        Racing {
         120  +
            shared: Arc<Mutex<Shared<M::Response>>>,
         121  +
            select: future::Select<oneshot::Receiver<M::Response>, M::Future>,
         122  +
            events: Ev,
         123  +
        },
         124  +
        Connecting {
         125  +
            // TODO: could be Weak even here...
         126  +
            shared: Arc<Mutex<Shared<M::Response>>>,
         127  +
            future: M::Future,
         128  +
        },
         129  +
        Cached {
         130  +
            svc: Option<Cached<M::Response>>,
         131  +
        },
         132  +
    }
         133  +
         134  +
    // shouldn't be pub
         135  +
    #[derive(Debug)]
         136  +
    pub struct Shared<S> {
         137  +
        services: Vec<S>,
         138  +
        waiters: Vec<oneshot::Sender<S>>,
         139  +
    }
         140  +
         141  +
    // impl Builder
         142  +
         143  +
    impl<Ev> Builder<Ev> {
         144  +
        /// Provide a `Future` executor to be used by the `Cache`.
         145  +
        pub fn executor<E>(self, exec: E) -> Builder<events::WithExecutor<E>> {
         146  +
            Builder {
         147  +
                events: events::WithExecutor(exec),
         148  +
            }
         149  +
        }
         150  +
         151  +
        /// Build a `Cache` pool around the `connector`.
         152  +
        pub fn build<M, Dst>(self, connector: M) -> Cache<M, Dst, Ev>
         153  +
        where
         154  +
            M: Service<Dst>,
         155  +
        {
         156  +
            Cache {
         157  +
                connector,
         158  +
                events: self.events,
         159  +
                shared: Arc::new(Mutex::new(Shared {
         160  +
                    services: Vec::new(),
         161  +
                    waiters: Vec::new(),
         162  +
                })),
         163  +
            }
         164  +
        }
         165  +
    }
         166  +
         167  +
    // impl Cache
         168  +
         169  +
    impl<M, Dst, Ev> Cache<M, Dst, Ev>
         170  +
    where
         171  +
        M: Service<Dst>,
         172  +
    {
         173  +
        /// Retain all cached services indicated by the predicate.
         174  +
        pub fn retain<F>(&mut self, predicate: F)
         175  +
        where
         176  +
            F: FnMut(&mut M::Response) -> bool,
         177  +
        {
         178  +
            self.shared.lock().unwrap().services.retain_mut(predicate);
         179  +
        }
         180  +
         181  +
        /// Check whether this cache has no cached services.
         182  +
        pub fn is_empty(&self) -> bool {
         183  +
            self.shared.lock().unwrap().services.is_empty()
         184  +
        }
         185  +
         186  +
        // SDK MODIFICATION: added `try_pop_idle` so a caller can remove an
         187  +
        // idle cached service to free whatever resource it holds, instead
         188  +
        // of waiting for it to be re-handed-out or evicted.
         189  +
        /// Remove and return one idle cached service, if any.
         190  +
        ///
         191  +
        /// Unlike [`Service::call`], which wraps the taken service in a
         192  +
        /// [`Cached`] that returns to the pool on drop, this hands back the
         193  +
        /// raw service with no return-to-pool wrapper: dropping it drops the
         194  +
        /// service outright. Serialized against [`Self::retain`] by the
         195  +
        /// shared `Mutex`, so a popped service is removed before a retain
         196  +
        /// pass can observe it.
         197  +
        pub fn try_pop_idle(&self) -> Option<M::Response> {
         198  +
            self.shared.lock().unwrap().services.pop()
         199  +
        }
         200  +
         201  +
        // SDK MODIFICATION: added `try_checkout_idle` so a caller can take an
         202  +
        // idle cached service for one use and have it return to the pool on
         203  +
        // drop, without going through `Service::call` (which may also start a
         204  +
        // new connection when none is idle).
         205  +
        /// Take one idle cached service, if any, wrapped so it returns to the
         206  +
        /// pool on drop.
         207  +
        ///
         208  +
        /// Unlike [`Self::try_pop_idle`], which hands back the raw service
         209  +
        /// (dropping it drops the service), this returns the same
         210  +
        /// [`Cached`] wrapper [`Service::call`] produces: dropping it
         211  +
        /// re-inserts the service into the pool. Unlike [`Service::call`],
         212  +
        /// it never starts a new connection — it returns `None` when no
         213  +
        /// service is idle. Serialized against [`Self::retain`] and
         214  +
        /// [`Self::try_pop_idle`] by the shared `Mutex`.
         215  +
        pub fn try_checkout_idle(&self) -> Option<Cached<M::Response>> {
         216  +
            let inner = self.shared.lock().unwrap().services.pop()?;
         217  +
            Some(Cached::new(inner, Arc::downgrade(&self.shared)))
         218  +
        }
         219  +
    }
         220  +
         221  +
    impl<M, Dst, Ev> Service<Dst> for Cache<M, Dst, Ev>
         222  +
    where
         223  +
        M: Service<Dst>,
         224  +
        M::Future: Unpin,
         225  +
        M::Response: Unpin,
         226  +
        Ev: events::Events<BackgroundConnect<M::Future, M::Response>> + Clone + Unpin,
         227  +
    {
         228  +
        type Response = Cached<M::Response>;
         229  +
        type Error = M::Error;
         230  +
        type Future = CacheFuture<M, Dst, Ev>;
         231  +
         232  +
        fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
         233  +
            if !self.shared.lock().unwrap().services.is_empty() {
         234  +
                Poll::Ready(Ok(()))
         235  +
            } else {
         236  +
                self.connector.poll_ready(cx)
         237  +
            }
         238  +
        }
         239  +
         240  +
        fn call(&mut self, target: Dst) -> Self::Future {
         241  +
            // 1. If already cached, easy!
         242  +
            let waiter = {
         243  +
                let mut locked = self.shared.lock().unwrap();
         244  +
                if let Some(found) = locked.take() {
         245  +
                    return CacheFuture::Cached {
         246  +
                        svc: Some(Cached::new(found, Arc::downgrade(&self.shared))),
         247  +
                    };
         248  +
                }
         249  +
         250  +
                let (tx, rx) = oneshot::channel();
         251  +
                locked.waiters.push(tx);
         252  +
                rx
         253  +
            };
         254  +
         255  +
            // 2. Otherwise, we start a new connect, and also listen for
         256  +
            //    any newly idle.
         257  +
            CacheFuture::Racing {
         258  +
                shared: self.shared.clone(),
         259  +
                select: future::select(waiter, self.connector.call(target)),
         260  +
                events: self.events.clone(),
         261  +
            }
         262  +
        }
         263  +
    }
         264  +
         265  +
    impl<M, Dst, Ev> Clone for Cache<M, Dst, Ev>
         266  +
    where
         267  +
        M: Service<Dst> + Clone,
         268  +
        Ev: Clone,
         269  +
    {
         270  +
        fn clone(&self) -> Self {
         271  +
            Self {
         272  +
                connector: self.connector.clone(),
         273  +
                events: self.events.clone(),
         274  +
                shared: self.shared.clone(),
         275  +
            }
         276  +
        }
         277  +
    }
         278  +
         279  +
    impl<M, Dst, Ev> Future for CacheFuture<M, Dst, Ev>
         280  +
    where
         281  +
        M: Service<Dst>,
         282  +
        M::Future: Unpin,
         283  +
        M::Response: Unpin,
         284  +
        Ev: events::Events<BackgroundConnect<M::Future, M::Response>> + Unpin,
         285  +
    {
         286  +
        type Output = Result<Cached<M::Response>, M::Error>;
         287  +
         288  +
        fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
         289  +
            loop {
         290  +
                match &mut *self.as_mut() {
         291  +
                    CacheFuture::Racing {
         292  +
                        shared,
         293  +
                        select,
         294  +
                        events,
         295  +
                    } => {
         296  +
                        match ready!(Pin::new(select).poll(cx)) {
         297  +
                            future::Either::Left((Err(_pool_closed), connecting)) => {
         298  +
                                // pool was dropped, so we'll never get it from a waiter,
         299  +
                                // but if this future still exists, then the user still
         300  +
                                // wants a connection. just wait for the connecting
         301  +
                                *self = CacheFuture::Connecting {
         302  +
                                    shared: shared.clone(),
         303  +
                                    future: connecting,
         304  +
                                };
         305  +
                            }
         306  +
                            future::Either::Left((Ok(pool_got), connecting)) => {
         307  +
                                events.on_race_lost(BackgroundConnect {
         308  +
                                    future: connecting,
         309  +
                                    shared: Arc::downgrade(&shared),
         310  +
                                });
         311  +
                                return Poll::Ready(Ok(Cached::new(
         312  +
                                    pool_got,
         313  +
                                    Arc::downgrade(&shared),
         314  +
                                )));
         315  +
                            }
         316  +
                            future::Either::Right((connected, _waiter)) => {
         317  +
                                let inner = connected?;
         318  +
                                return Poll::Ready(Ok(Cached::new(
         319  +
                                    inner,
         320  +
                                    Arc::downgrade(&shared),
         321  +
                                )));
         322  +
                            }
         323  +
                        }
         324  +
                    }
         325  +
                    CacheFuture::Connecting { shared, future } => {
         326  +
                        let inner = ready!(Pin::new(future).poll(cx))?;
         327  +
                        return Poll::Ready(Ok(Cached::new(inner, Arc::downgrade(&shared))));
         328  +
                    }
         329  +
                    CacheFuture::Cached { svc } => {
         330  +
                        return Poll::Ready(Ok(svc.take().unwrap()));
         331  +
                    }
         332  +
                }
         333  +
            }
         334  +
        }
         335  +
    }
         336  +
         337  +
    // impl Cached
         338  +
         339  +
    impl<S> Cached<S> {
         340  +
        fn new(inner: S, shared: Weak<Mutex<Shared<S>>>) -> Self {
         341  +
            Cached {
         342  +
                is_closed: false,
         343  +
                inner: Some(inner),
         344  +
                shared,
         345  +
            }
         346  +
        }
         347  +
         348  +
        // TODO: inner()? looks like `tower` likes `get_ref()` and `get_mut()`.
         349  +
         350  +
        /// Get a reference to the inner service.
         351  +
        pub fn inner(&self) -> &S {
         352  +
            self.inner.as_ref().expect("inner only taken in drop")
         353  +
        }
         354  +
         355  +
        /// Get a mutable reference to the inner service.
         356  +
        pub fn inner_mut(&mut self) -> &mut S {
         357  +
            self.inner.as_mut().expect("inner only taken in drop")
         358  +
        }
         359  +
         360  +
        // SDK MODIFICATION: added `discard` so callers can prevent a bad
         361  +
        // connection from returning to the pool without having to cause a
         362  +
        // synthetic `poll_ready` error.
         363  +
        /// Prevent this cached service from being returned to the pool.
         364  +
        ///
         365  +
        /// Consumes `self`; the inner service is dropped without
         366  +
        /// reinsertion, regardless of whether it is still healthy.
         367  +
        pub fn discard(mut self) {
         368  +
            self.is_closed = true;
         369  +
        }
         370  +
    }
         371  +
         372  +
    impl<S, Req> Service<Req> for Cached<S>
         373  +
    where
         374  +
        S: Service<Req>,
         375  +
    {
         376  +
        type Response = S::Response;
         377  +
        type Error = S::Error;
         378  +
        type Future = S::Future;
         379  +
         380  +
        fn poll_ready(&mut self, cx: &mut task::Context<'_>) -> Poll<Result<(), Self::Error>> {
         381  +
            self.inner.as_mut().unwrap().poll_ready(cx).map_err(|err| {
         382  +
                self.is_closed = true;
         383  +
                err
         384  +
            })
         385  +
        }
         386  +
         387  +
        fn call(&mut self, req: Req) -> Self::Future {
         388  +
            self.inner.as_mut().unwrap().call(req)
         389  +
        }
         390  +
    }
         391  +
         392  +
    impl<S> Drop for Cached<S> {
         393  +
        fn drop(&mut self) {
         394  +
            if self.is_closed {
         395  +
                return;
         396  +
            }
         397  +
            if let Some(value) = self.inner.take() {
         398  +
                if let Some(shared) = self.shared.upgrade() {
         399  +
                    if let Ok(mut shared) = shared.lock() {
         400  +
                        shared.put(value);
         401  +
                    }
         402  +
                }
         403  +
            }
         404  +
        }
         405  +
    }
         406  +
         407  +
    impl<S: fmt::Debug> fmt::Debug for Cached<S> {
         408  +
        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         409  +
            f.debug_tuple("Cached")
         410  +
                .field(self.inner.as_ref().unwrap())
         411  +
                .finish()
         412  +
        }
         413  +
    }
         414  +
         415  +
    // impl Shared
         416  +
         417  +
    impl<V> Shared<V> {
         418  +
        fn put(&mut self, val: V) {
         419  +
            let mut val = Some(val);
         420  +
            while let Some(tx) = self.waiters.pop() {
         421  +
                if !tx.is_closed() {
         422  +
                    match tx.send(val.take().unwrap()) {
         423  +
                        Ok(()) => break,
         424  +
                        Err(v) => {
         425  +
                            val = Some(v);
         426  +
                        }
         427  +
                    }
         428  +
                }
         429  +
            }
         430  +
         431  +
            if let Some(val) = val {
         432  +
                self.services.push(val);
         433  +
            }
         434  +
        }
         435  +
         436  +
        fn take(&mut self) -> Option<V> {
         437  +
            // TODO: take in a loop
         438  +
            self.services.pop()
         439  +
        }
         440  +
    }
         441  +
         442  +
    pub struct BackgroundConnect<CF, S> {
         443  +
        future: CF,
         444  +
        shared: Weak<Mutex<Shared<S>>>,
         445  +
    }
         446  +
         447  +
    impl<CF, S, E> Future for BackgroundConnect<CF, S>
         448  +
    where
         449  +
        CF: Future<Output = Result<S, E>> + Unpin,
         450  +
    {
         451  +
        type Output = ();
         452  +
         453  +
        fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
         454  +
            match ready!(Pin::new(&mut self.future).poll(cx)) {
         455  +
                Ok(svc) => {
         456  +
                    if let Some(shared) = self.shared.upgrade() {
         457  +
                        if let Ok(mut locked) = shared.lock() {
         458  +
                            locked.put(svc);
         459  +
                        }
         460  +
                    }
         461  +
                    Poll::Ready(())
         462  +
                }
         463  +
                Err(_e) => Poll::Ready(()),
         464  +
            }
         465  +
        }
         466  +
    }
         467  +
}
         468  +
         469  +
mod events {
         470  +
    #[derive(Clone, Debug)]
         471  +
    #[non_exhaustive]
         472  +
    pub struct Ignore;
         473  +
         474  +
    #[derive(Clone, Debug)]
         475  +
    pub struct WithExecutor<E>(pub(super) E);
         476  +
         477  +
    pub trait Events<CF> {
         478  +
        fn on_race_lost(&self, fut: CF);
         479  +
    }
         480  +
         481  +
    impl<CF> Events<CF> for Ignore {
         482  +
        fn on_race_lost(&self, _fut: CF) {}
         483  +
    }
         484  +
         485  +
    impl<E, CF> Events<CF> for WithExecutor<E>
         486  +
    where
         487  +
        E: hyper::rt::Executor<CF>,
         488  +
    {
         489  +
        fn on_race_lost(&self, fut: CF) {
         490  +
            self.0.execute(fut);
         491  +
        }
         492  +
    }
         493  +
}
         494  +
         495  +
#[cfg(test)]
         496  +
mod tests {
         497  +
    use futures_util::future;
         498  +
    use tower_service::Service;
         499  +
    use tower_test::assert_request_eq;
         500  +
         501  +
    #[tokio::test]
         502  +
    async fn test_makes_svc_when_empty() {
         503  +
        let (mock, mut handle) = tower_test::mock::pair();
         504  +
        let mut cache = super::builder().build(mock);
         505  +
        handle.allow(1);
         506  +
         507  +
        std::future::poll_fn(|cx| cache.poll_ready(cx))
         508  +
            .await
         509  +
            .unwrap();
         510  +
         511  +
        let f = cache.call(1);
         512  +
         513  +
        future::join(f, async move {
         514  +
            assert_request_eq!(handle, 1).send_response("one");
         515  +
        })
         516  +
        .await
         517  +
        .0
         518  +
        .expect("call");
         519  +
    }
         520  +
         521  +
    #[tokio::test]
         522  +
    async fn test_reuses_after_idle() {
         523  +
        let (mock, mut handle) = tower_test::mock::pair();
         524  +
        let mut cache = super::builder().build(mock);
         525  +
         526  +
        // only 1 connection should ever be made
         527  +
        handle.allow(1);
         528  +
         529  +
        std::future::poll_fn(|cx| cache.poll_ready(cx))
         530  +
            .await
         531  +
            .unwrap();
         532  +
        let f = cache.call(1);
         533  +
        let cached = future::join(f, async {
         534  +
            assert_request_eq!(handle, 1).send_response("one");
         535  +
        })
         536  +
        .await
         537  +
        .0
         538  +
        .expect("call");
         539  +
        drop(cached);
         540  +
         541  +
        std::future::poll_fn(|cx| cache.poll_ready(cx))
         542  +
            .await
         543  +
            .unwrap();
         544  +
        let f = cache.call(1);
         545  +
        let cached = f.await.expect("call");
         546  +
        drop(cached);
         547  +
    }
         548  +
}