AWS SDK

rev. 01140188efe3be26ec6653bc30094c80ed1a1b99

Files changed:

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

@@ -1,1 +380,162 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Transport construction below HTTP protocol establishment.
 //! Transport connection followed by HTTP protocol establishment.
 //!
 //! [`transport`] creates connected I/O and reports the negotiated protocol.
 //! This module then routes that transport through the matching Hyper handshake,
 //! creates the protocol-neutral connection identity, and returns or transfers
 //! completion ownership for the launching waiter.
-12
 mod h1;
 mod h2;
 mod transport;
-16
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 pub(super) use transport::from_cached_interface_connector;
 #[cfg(any(
     all(feature = "test-util", aws_sdk_unstable),
     all(test, feature = "rt-tokio")
 ))]
 pub(super) use transport::from_connector;
 pub(super) use transport::{from_interface_connector, TransportFactory, TransportTimeout};
-+
 use self::transport::TransportConnectContext;
 use super::admission::ProtocolRequirement;
 use super::cell::{AcquisitionResult, EstablishmentPermit, WaiterId};
 use super::cell::{AcquisitionOutcome, EstablishmentPermit, WaiterId};
 use super::dispatch::AcquisitionContext;
 use super::registry::PartitionState;
 use super::PoolInner;
 use crate::client::connect::{AsyncConn, BoxConn};
 use crate::client::timeout::{self, TimeoutKind};
 use aws_smithy_async::rt::sleep::SharedAsyncSleep;
 use aws_smithy_runtime_api::box_error::BoxError;
 use crate::client::downcast_error;
 use aws_smithy_runtime_api::client::connection::ConnectionId;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use http_1x::Uri;
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
 use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::atomic::Ordering;
 use std::sync::Arc as StdArc;
 use std::time::Duration;
 use tower::Service;
--
 /// Timeout policy for one transport connection operation.
 ///
 /// Connector readiness completes before this timeout starts. Keeping the
 /// duration and timer together lets request acquisition pass one complete
 /// timeout value into transport construction.
 #[derive(Clone, Debug)]
 pub(super) struct TransportTimeout {
     /// Maximum duration of the connector's connection future.
     duration: Duration,
     /// Runtime timer used to enforce the deadline.
     sleep: SharedAsyncSleep,
--
+}
--
 impl TransportTimeout {
     /// Creates a timeout enforced by the supplied runtime timer.
     pub(super) fn new(duration: Duration, sleep: SharedAsyncSleep) -> Self {
         Self { duration, sleep }
--
+    }
--
+}
--
 /// Future returned by a type-erased transport factory.
 type TransportFuture = Pin<Box<dyn Future<Output = Result<BoxConn, BoxError>> + Send + 'static>>;
--
 /// Type-erased transport construction below HTTP protocol establishment.
 ///
 /// A pool retains one factory that applies partition interface placement,
 /// connector readiness, and the operation's connection timeout. The concrete
 /// connector remains responsible for DNS, proxy, TLS, socket construction,
 /// and transport metadata. Protocol establishment consumes the resulting
 /// [`BoxConn`].
 pub(super) trait TransportFactory: Send + Sync + 'static {
     /// Returns whether this factory can guarantee HTTP/1 for a required attempt.
     fn guarantees_http1(&self) -> bool;
--
     /// Creates one partition-bound transport for `uri`.
     ///
     /// Connector readiness completes before `timeout` starts.
     fn connect(
         &self,
         partition: &PartitionState,
         uri: Uri,
         timeout: Option<TransportTimeout>,
         alpn_protocols: AlpnProtocols,
     ) -> TransportFuture;
--
+}
--
 /// Generic connector factory erased at pool construction.
 struct ServiceTransportFactory<F> {
     /// Builds a connector with the selected network-interface binding.
     connector_for_interface: F,
     /// Whether every H1-required connection is guaranteed to negotiate H1.
     guarantees_http1: bool,
--
+}
--
 impl<F, C, IO> TransportFactory for ServiceTransportFactory<F>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     fn guarantees_http1(&self) -> bool {
         self.guarantees_http1
--
+    }
--
     fn connect(
         &self,
         partition: &PartitionState,
         uri: Uri,
         timeout: Option<TransportTimeout>,
         _alpn_protocols: AlpnProtocols,
     ) -> TransportFuture {
         let interface = partition.interface().map(|interface| interface.as_ref());
         let mut connector = (self.connector_for_interface)(interface);
         Box::pin(async move {
             // Readiness follows the existing connector contract and is not timed.
             poll_fn(|cx| connector.poll_ready(cx))
                 .await
                 .map_err(Into::into)?;
             let connect = connector.call(uri);
             let io = timeout::maybe_timeout_future(
                 connect,
                 timeout.as_ref().map(|timeout| timeout.duration),
                 timeout.as_ref().map(|timeout| &timeout.sleep),
                 TimeoutKind::Connect,
--
+            )
             .await?;
             Ok(Box::new(io) as BoxConn)
         })
--
+    }
--
+}
--
 #[cfg(any(
     all(feature = "test-util", aws_sdk_unstable),
     all(test, feature = "rt-tokio")
 ))]
 /// Erases a clonable connector for pool storage.
 pub(super) fn transport_factory<C, IO>(connector: C) -> StdArc<dyn TransportFactory>
 where
     C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     service_transport_factory_for_interface(move |_| connector.clone(), false)
--
+}
--
 /// Erases a connector constructor that applies an optional interface binding.
 ///
 /// The default HTTP connector consumes this value. Custom connectors use
 /// `transport_factory` and ignore pool interface placement.
 pub(super) fn transport_factory_for_interface<F, C, IO>(
     connector_for_interface: F,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     service_transport_factory_for_interface(connector_for_interface, true)
--
+}
--
 /// Erases a service connector and its HTTP/1 negotiation guarantee.
 fn service_transport_factory_for_interface<F, C, IO>(
     connector_for_interface: F,
     guarantees_http1: bool,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     StdArc::new(ServiceTransportFactory {
         connector_for_interface,
         guarantees_http1,
     })
--
+}
--
 /// Erases and caches connectors whose construction carries TLS or other setup.
 ///
 /// Interface placement is immutable for a partition. The first request for
 /// each `(interface, ALPN offer)` constructs its connector while holding the
 /// cache lock; later requests clone that retained connector. Provider
 /// configuration and certificate loading therefore occur at most once for
 /// each placement and offer rather than once per connection.
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 pub(super) fn cached_transport_factory_for_interface<F, C, IO>(
     connector_for_interface: F,
     guarantees_http1: bool,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>, AlpnProtocols) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
--
+{
     struct Cached<F, C> {
         factory: F,
         guarantees_http1: bool,
         connectors: crate::sync::Mutex<HashMap<(Option<String>, AlpnProtocols), C>>,
--
+    }
--
     impl<F, C, IO> TransportFactory for Cached<F, C>
     where
         F: Fn(Option<&str>, AlpnProtocols) -> C + Send + Sync + 'static,
         C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
         C::Error: Into<BoxError>,
         C::Future: Send + 'static,
         IO: AsyncConn,
--
+    {
         fn guarantees_http1(&self) -> bool {
             self.guarantees_http1
--
+        }
--
         fn connect(
             &self,
             partition: &PartitionState,
             uri: Uri,
             timeout: Option<TransportTimeout>,
             alpn_protocols: AlpnProtocols,
         ) -> TransportFuture {
             let interface = partition.interface().map(|value| value.to_string());
             let mut connector = self
                 .connectors
                 .lock()
                 .entry((interface.clone(), alpn_protocols))
                 .or_insert_with(|| (self.factory)(interface.as_deref(), alpn_protocols))
                 .clone();
             Box::pin(async move {
                 poll_fn(|cx| connector.poll_ready(cx))
                     .await
                     .map_err(Into::into)?;
                 let connect = connector.call(uri);
                 let io = timeout::maybe_timeout_future(
                     connect,
                     timeout.as_ref().map(|timeout| timeout.duration),
                     timeout.as_ref().map(|timeout| &timeout.sleep),
                     TimeoutKind::Connect,
--
+                )
                 .await?;
                 Ok(Box::new(io) as BoxConn)
             })
--
+        }
--
+    }
--
     StdArc::new(Cached {
         factory: connector_for_interface,
         guarantees_http1,
         connectors: crate::sync::Mutex::new(HashMap::new()),
     })
--
+}
-37
 /// Result of one owner-runtime establishment task.
 pub(super) enum EstablishmentOutcome {
     /// The launching waiter receives this terminal result.
     Complete(AcquisitionResult),
     /// Waiter completion transferred to an H2 flight or generation.
     Transferred,
     Complete(AcquisitionOutcome),
     /// An H2 flight or generation now owns the launching waiter's completion.
     WaiterCompletionTransferred,
-44
+}
-45
 /// Connects one transport and dispatches protocol establishment after ALPN.
 pub(super) async fn establish(
     context: AcquisitionContext,
     waiter: WaiterId,
     permit: EstablishmentPermit,
     requirement: ProtocolRequirement,
 ) -> EstablishmentOutcome {
     let io = match context
         .pool
         .transport
         .connect(
             &context.partition,
             context.absolute_uri.clone(),
             context.connect_timeout.clone(),
             alpn_protocols(requirement),
--
+        )
         .await
--
+    {
     let connect = TransportConnectContext::new(
         &context.partition,
         context.absolute_uri.clone(),
         context.connect_timeout.clone(),
         requirement,
     );
     let io = match context.pool.transport.connect(connect).await {
         Ok(io) => io,
         Err(error) => {
             tracing::debug!(
                 request_partition = ?context.partition.id(),
                 connection_partition = ?context.cell.id().partition(),
                 origin_scheme = %context.cell.id().origin().scheme(),
                 origin_host = context.cell.id().origin().host(),
                 origin_port = ?context.cell.id().origin().port(),
                 error = ?error,
                 "transport establishment failed"
             );
             return EstablishmentOutcome::Complete(AcquisitionResult::Failed(
                 super::super::downcast_error(error),
             ));
             return EstablishmentOutcome::Complete(AcquisitionOutcome::Failed(downcast_error(
                 error,
             )));
-74
+        }
     };
     let connected = io.connected();
     let negotiated_h2 = connected.is_negotiated_h2();
     tracing::debug!(
         request_partition = ?context.partition.id(),
         connection_partition = ?context.cell.id().partition(),
         origin_scheme = %context.cell.id().origin().scheme(),
         origin_host = context.cell.id().origin().host(),
         origin_port = ?context.cell.id().origin().port(),
         negotiated_protocol = if negotiated_h2 { "HTTP/2" } else { "HTTP/1.1" },
         "transport protocol negotiated"
     );
     if negotiated_h2 && !requirement.accepts_h2() {
         drop(io);
         drop(permit);
         return EstablishmentOutcome::Complete(AcquisitionResult::Failed(ConnectorError::user(
             NegotiatedProtocolMismatch { requirement }.into(),
         )));
         return EstablishmentOutcome::Complete(AcquisitionOutcome::Failed(
             negotiated_protocol_mismatch(requirement),
         ));
-93
+    }
-94
     if negotiated_h2 {
         h2::establish_h2(context, waiter, permit, io, connected).await
     } else {
         EstablishmentOutcome::Complete(
             h1::establish_h1(context, permit, io, connected)
                 .await
                 .map(AcquisitionResult::H1)
                 .unwrap_or_else(AcquisitionResult::Failed),
                 .map(AcquisitionOutcome::H1)
                 .unwrap_or_else(AcquisitionOutcome::Failed),
-103
+        )
-104
+    }
-105
+}
--
 /// An established transport selected a protocol incompatible with the request.
 #[derive(Debug)]
 struct NegotiatedProtocolMismatch {
     requirement: ProtocolRequirement,
-110
+}
-111
 impl fmt::Display for NegotiatedProtocolMismatch {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         write!(
             f,
             "transport negotiated HTTP/2, which does not satisfy {:?} request semantics",
             self.requirement
-118
+        )
-119
+    }
-120
+}
-121
 impl Error for NegotiatedProtocolMismatch {}
-123
 fn negotiated_protocol_mismatch(requirement: ProtocolRequirement) -> ConnectorError {
     ConnectorError::other(NegotiatedProtocolMismatch { requirement }.into(), None)
-+
+}
-+
 /// Mints one non-wrapping physical-connection identity.
 fn next_connection_id(pool: &PoolInner) -> Result<ConnectionId, ConnectionIdExhausted> {
     let value = pool
         .next_connection_id
         .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
             current.checked_add(1)
         })
         .map_err(|_| ConnectionIdExhausted)?;
     Ok(ConnectionId::new(value))
-137
+}
-138
 /// The pool exhausted its monotonic physical-connection identity space.
 #[derive(Debug)]
 struct ConnectionIdExhausted;
-142
 impl fmt::Display for ConnectionIdExhausted {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str("connection identifier space exhausted")
-146
+    }
-147
+}
-148
 impl Error for ConnectionIdExhausted {}
-150
 /// Static ALPN protocol offer used for one transport connection.
 pub(super) type AlpnProtocols = &'static [&'static [u8]];
--
 /// Default offer for requests that may use HTTP/2.
 pub(super) const HTTP_ALPN_PROTOCOLS: AlpnProtocols = &[b"h2", b"http/1.1"];
 #[cfg(test)]
 mod tests {
     use super::*;
-154
 /// Narrowed offer for requests that require HTTP/1 wire semantics.
 pub(super) const HTTP1_ALPN_PROTOCOLS: AlpnProtocols = &[b"http/1.1"];
     #[test]
     fn negotiated_protocol_mismatch_is_not_a_user_error() {
         let error = negotiated_protocol_mismatch(ProtocolRequirement::H1Required);
-158
 fn alpn_protocols(requirement: ProtocolRequirement) -> AlpnProtocols {
     match requirement {
         ProtocolRequirement::H1Required => HTTP1_ALPN_PROTOCOLS,
         ProtocolRequirement::H1Compatible | ProtocolRequirement::H2Required => HTTP_ALPN_PROTOCOLS,
         assert!(error.is_other());
         assert!(!error.is_user());
-161
+    }
-162
+}

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

@@ -1,1 +142,143 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! HTTP/1 connection establishment.
 //!
 //! Establishment receives an already connected transport, performs Hyper's
 //! client handshake, installs the resulting exclusive sender in the origin

 //! cell, and submits the protocol driver to the connection-owning partition.
-11
 use super::super::cell::h1::{H1CloseHandle, H1DriverGuard, H1Selection, H1Sender};
 use super::super::cell::{EstablishmentPermit, OriginCell};
 use super::super::connection::{
     CloseReason, ConnectionInfo, ConnectionIo, ConnectionState, NegotiatedProtocol,
 };
 use super::super::dispatch::AcquisitionContext;
 use super::next_connection_id;
 use crate::client::connect::BoxConn;
 use crate::client::downcast_error;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use hyper_util::client::legacy::connect::Connected;
-24
 /// Handshakes, installs, and starts the owner-partition driver.
 ///
 /// `permit` remains the sole owner of optional bounded capacity until the
 /// connection state is created. An earlier failure returns that capacity
 /// through the permit's fallback. A handshake failure logically closes the
 /// new connection state, while success transfers driver ownership to the
 /// partition runtime and returns the installed sender selection.
 pub(in crate::client::pool) async fn establish_h1(
     context: AcquisitionContext,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) -> Result<H1Selection, ConnectorError> {
     let request_partition = context.partition.id();
     let connection_cell = context.cell.clone();
     tracing::debug!(
         request_partition = ?request_partition,
         connection_partition = ?connection_cell.id().partition(),
         origin_scheme = %connection_cell.id().origin().scheme(),
         origin_host = connection_cell.id().origin().host(),
         origin_port = ?connection_cell.id().origin().port(),
         "HTTP/1 connection establishment started"
     );
-48
     let result = handshake_and_install_h1(context, permit, io, connected).await;
     let result = run_h1_handshake(context, permit, io, connected).await;
     match &result {
         Ok(selection) => {
             let connection = selection.connection();
             tracing::debug!(
                 connection_id = %connection.id(),
                 request_partition = ?request_partition,
                 connection_partition = ?connection.owner_partition(),
                 origin_scheme = %connection.info().origin().scheme(),
                 origin_host = connection.info().origin().host(),
                 origin_port = ?connection.info().origin().port(),
                 "HTTP/1 connection established"
             );
-62
+        }
         Err(error) => tracing::debug!(
             request_partition = ?request_partition,
             connection_partition = ?connection_cell.id().partition(),
             origin_scheme = %connection_cell.id().origin().scheme(),
             origin_host = connection_cell.id().origin().host(),
             origin_port = ?connection_cell.id().origin().port(),
             error = ?error,
             "HTTP/1 connection establishment failed"
         ),
-72
+    }
     result
-74
+}
-75
 /// Handshakes and installs one already connected HTTP/1 transport.
 async fn handshake_and_install_h1(
 async fn run_h1_handshake(
     context: AcquisitionContext,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) -> Result<H1Selection, ConnectorError> {
     let AcquisitionContext {
         pool,
         partition,
         cell,
         absolute_uri: _,
         owner_spawner,
         connect_timeout: _,
     } = context;
-91
     let id =
         next_connection_id(&pool).map_err(|error| ConnectorError::other(error.into(), None))?;
     let info = ConnectionInfo::new(
         id,
         cell.id().origin().clone(),
         partition.id(),
         NegotiatedProtocol::Http1,
         connected,
     );
     let (connection, physical) = ConnectionState::pending_open(info);
     let io = ConnectionIo::new(io, physical);
-103
     let (sender, driver) = match hyper::client::conn::http1::Builder::new()
         .handshake::<_, SdkBody>(io)
         .await
-107
+    {
         Ok(established) => established,
         Err(error) => {
             connection.logical_close(CloseReason::ProtocolClosed);
             return Err(super::super::super::downcast_error(Box::new(error)));
             return Err(downcast_error(Box::new(error)));
-112
+        }
     };
-114
     if let Err(lease) = connection.open(permit.into_lease()) {
         drop(lease);
         connection.logical_close(CloseReason::ProtocolClosed);
         return Err(ConnectorError::io(
             "HTTP/1 connection closed before installation".into(),
         ));
-121
+    }
-122
     let selection =
         OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::from_hyper(sender));
         OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::from_hyper(sender));
     let driver_guard = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));
     let driver_info = connection.info().clone();
     owner_spawner.spawn(Box::pin(async move {
         let result = driver.with_upgrades().await;
         if let Err(error) = result {
             tracing::debug!(
                 connection_id = %driver_info.id(),
                 connection_partition = ?driver_info.owner_partition(),
                 origin_scheme = %driver_info.origin().scheme(),
                 origin_host = driver_info.origin().host(),

↧ Expand context

                 origin_port = ?driver_info.origin().port(),
                 error = ?error,
                 "HTTP/1 connection driver failed"
             );
-139
+        }
         driver_guard.protocol_closed();
     }));
     Ok(selection)
-143
+}

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

@@ -1,1 +58,59 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! HTTP/2 flight convergence, handshake, and driver installation.
 //!
 //! ALPN-selected attempts first enter one cell-local transition: use an

 //! accepting generation, join the current flight, or become the flight's
 //! owner task. Only the owner performs Hyper's HTTP/2 handshake. It opens the
 //! connection, installs the generation, submits the driver on the partition
 //! spawner, and then disarms flight cleanup. Participants own only their
 //! waiter entries; losing attempts close their unused transport and return
 //! their capacity before this function reports transfer. The connector timeout
 //! ends before this phase; no separate pool timeout covers the HTTP/2
 //! handshake or driver submission.
-17
 use super::super::cell::h2::{
     H2CloseHandle, H2DriverGuard, H2FlightId, H2FlightInstall, H2GenerationJoin, H2Sender,
     H2CloseHandle, H2DriverGuard, H2FlightDecision, H2FlightId, H2GenerationJoinOutcome, H2Sender,
 };
 use super::super::cell::{AcquisitionResult, EstablishmentPermit, OriginCell, WaiterId};
 use super::super::cell::{AcquisitionOutcome, EstablishmentPermit, OriginCell, WaiterId};
 use super::super::connection::{
     CloseReason, ConnectionInfo, ConnectionIo, ConnectionState, NegotiatedProtocol,
 };
 use super::super::dispatch::AcquisitionContext;
 use super::super::partition::DriverSpawner;
 use super::{next_connection_id, EstablishmentOutcome};
 use crate::client::connect::BoxConn;
 use crate::client::downcast_error;
 use aws_smithy_runtime_api::box_error::BoxError;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use aws_smithy_types::retry::ErrorKind;
 use hyper::rt::Executor;
 use hyper_util::client::legacy::connect::Connected;
 use std::future::Future;
 use std::sync::Arc as StdArc;
-38
 /// Hyper executor that keeps spawned HTTP/2 work on the connection partition.

↧ Expand context

 #[derive(Clone, Debug)]
 struct PartitionExecutor {
     /// Runtime placement inherited from the connection-owning partition.
     spawner: StdArc<dyn DriverSpawner>,
-44
+}
-45
 impl<F> Executor<F> for PartitionExecutor
 where
     F: Future<Output = ()> + Send + 'static,
-49
+{
     fn execute(&self, future: F) {
         self.spawner.spawn(Box::pin(future));
-52
+    }
-53
+}
-54
 /// Settles a flight if its owner-runtime task is discarded.
 struct FlightCompletionGuard {
     /// Cell that owns the exact flight.
     cell: crate::sync::Arc<OriginCell>,
     /// Flight identity protected from stale task completion.

@@ -76,77 +221,222 @@

↥ Expand context

-77
     fn fail(&mut self, error: ConnectorError) {
         self.active = false;
         fail_participants(&self.cell, self.flight, error);
-81
+    }
-82
+}
-83
 impl Drop for FlightCompletionGuard {
     fn drop(&mut self) {
         if self.active {
             fail_participants(
                 &self.cell,
                 self.flight,
                 ConnectorError::io(Box::new(std::io::Error::other(
                     "HTTP/2 establishment task was dropped",
                 ))),
             );
-94
+        }
-95
+    }
-96
+}

-97
 /// Converges after ALPN and performs at most one HTTP/2 handshake per cell.
 pub(super) async fn establish_h2(
     context: AcquisitionContext,
     waiter: WaiterId,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) -> EstablishmentOutcome {
     loop {
         match context.cell.install_or_join_h2_flight(waiter) {
             H2FlightInstall::Accepting(generation) => {
         match context.cell.converge_h2_flight(waiter) {
             H2FlightDecision::UseGeneration(generation) => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     h2_generation = ?generation,
                     "HTTP/2 establishment found an accepting generation"
                 );
                 match OriginCell::join_h2_generation(&context.cell, waiter, generation) {
                     H2GenerationJoin::GenerationChanged => continue,
                     H2GenerationJoin::Joined | H2GenerationJoin::WaiterCompleted => {
                     H2GenerationJoinOutcome::GenerationChanged => continue,
                     H2GenerationJoinOutcome::Joined | H2GenerationJoinOutcome::WaiterResolved => {
                         drop(io);
                         drop(permit);
                         return EstablishmentOutcome::Transferred;
                         return EstablishmentOutcome::WaiterCompletionTransferred;
-124
+                    }
-125
+                }
-126
+            }
             H2FlightInstall::Joined => {
             H2FlightDecision::JoinedFlight => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     "HTTP/2 establishment joined the active flight"
                 );
                 drop(io);
                 drop(permit);
                 return EstablishmentOutcome::Transferred;
                 return EstablishmentOutcome::WaiterCompletionTransferred;
-139
+            }
             H2FlightInstall::Driver(flight) => {
             H2FlightDecision::RunFlight(flight) => {
                 tracing::trace!(
                     request_partition = ?context.partition.id(),
                     connection_partition = ?context.cell.id().partition(),
                     origin_scheme = %context.cell.id().origin().scheme(),
                     origin_host = context.cell.id().origin().host(),
                     origin_port = ?context.cell.id().origin().port(),
                     h2_flight = ?flight,
                     "HTTP/2 establishment started a flight"
                 );
                 drive_flight(context, flight, permit, io, connected).await;
                 return EstablishmentOutcome::Transferred;
                 return EstablishmentOutcome::WaiterCompletionTransferred;
-152
+            }
-153
+        }
-154
+    }
-155
+}
-156
 /// Handshakes, installs, and publishes one winning flight.
 async fn drive_flight(
     context: AcquisitionContext,
     flight: H2FlightId,
     permit: EstablishmentPermit,
     io: BoxConn,
     connected: Connected,
 ) {
     let mut completion = FlightCompletionGuard::new(context.cell.clone(), flight);
     let id = match next_connection_id(&context.pool) {
         Ok(id) => id,
         Err(error) => {
             completion.fail(ConnectorError::other(Box::new(error), None));
             return;
-171
+        }
     };
     let info = ConnectionInfo::new(
         id,
         context.cell.id().origin().clone(),
         context.partition.id(),
         NegotiatedProtocol::Http2,
         connected,
     );
     let (connection, physical) = ConnectionState::pending_open(info);
     let io = ConnectionIo::new(io, physical);
     let executor = PartitionExecutor {
         spawner: context.owner_spawner.clone(),
     };
     let (sender, driver) = match hyper::client::conn::http2::Builder::new(executor)
         .handshake::<_, SdkBody>(io)
         .await
-188
+    {
         Ok(established) => established,
         Err(error) => {
             connection.logical_close(CloseReason::ProtocolClosed);
             completion.fail(super::super::super::downcast_error(Box::new(error)));
             completion.fail(downcast_error(Box::new(error)));
             return;
-194
+        }
     };
-196
     if let Err(lease) = connection.open(permit.into_lease()) {
         drop(lease);
         connection.logical_close(CloseReason::ProtocolClosed);
         completion.fail(ConnectorError::io(Box::new(std::io::Error::other(
             "HTTP/2 connection closed before installation",
         ))));

↧ Expand context

         return;
-204
+    }
-205
     let installed = OriginCell::complete_h2_flight(
         &context.cell,
         flight,
         connection.clone(),
         H2Sender::from_hyper(sender),
         context.cell.idle_deadline(),
     );
     let installed = match installed {
         Ok(installed) => installed,
         Err((_connection, _sender)) => {
             connection.logical_close(CloseReason::ProtocolClosed);
             completion.fail(ConnectorError::io(Box::new(std::io::Error::other(
                 "HTTP/2 flight became stale before installation",
             ))));
             return;
-221
+        }
     };

@@ -296,297 +447,448 @@

↥ Expand context

         match self.kind {
             SharedFailureKind::Timeout => ConnectorError::timeout(source),
             SharedFailureKind::User => ConnectorError::user(source),
             SharedFailureKind::Io => ConnectorError::io(source),
             SharedFailureKind::Other(kind) => ConnectorError::other(source, kind),
-302
+        }
-303
+    }
-304
+}
-305
 impl std::fmt::Display for SharedFlightFailure {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         std::fmt::Display::fmt(&self.source, f)
-309
+    }
-310
+}
-311
 impl std::error::Error for SharedFlightFailure {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         Some(self.source.as_ref().as_ref())
-315
+    }
-316
+}

-317
 /// Fails every participant still retained by one exact flight.
 fn fail_participants(cell: &OriginCell, flight: H2FlightId, error: ConnectorError) {
     let Some(participants) = cell.fail_h2_flight(flight) else {
         return;
     };
     let error = SharedFlightFailure::new(error);
     for participant in participants {
         cell.complete_establishment(
             participant,
             AcquisitionResult::Failed(error.connector_error()),
             AcquisitionOutcome::Failed(error.connector_error()),
         );
-329
+    }
-330
+}
-331
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use crate::client::pool::admission::ProtocolRequirement;
     use crate::client::pool::cell::AcquisitionEvent;
     use crate::client::pool::cell::{AcquisitionOutcome, AcquisitionStep};
     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::EligibilityGroup;
     use http_1x::uri::Scheme;
     use std::error::Error as _;
     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::task::{Context, Poll, Waker};
-343
     fn cell() -> crate::sync::Arc<OriginCell> {
         crate::sync::Arc::new(OriginCell::new(
             super::super::super::partition::PartitionId::from_index(1),
             OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
             EligibilityGroup::Pool,
             None,
             None,
         ))
-352
+    }
-353
     fn launching_waiter(cell: &crate::sync::Arc<OriginCell>) -> WaiterId {
         let waiter = OriginCell::register_waiter(cell, ProtocolRequirement::H2Required);
         let event = cell.poll_waiter(waiter, &mut Context::from_waker(Waker::noop()));
         let Poll::Ready(AcquisitionEvent::Establish(permit)) = event else {
         let Poll::Ready(AcquisitionStep::StartEstablishment(permit)) = event else {
             panic!("new H2 waiter did not receive establishment authority");
         };
         assert!(cell.start_establishment(waiter));
         drop(permit);
         waiter
-363
+    }
-364
     fn failed_event(cell: &OriginCell, waiter: WaiterId) -> ConnectorError {
         let event = cell.poll_waiter(waiter, &mut Context::from_waker(Waker::noop()));
         let Poll::Ready(AcquisitionEvent::Complete(AcquisitionResult::Failed(error))) = event
         let Poll::Ready(AcquisitionStep::Resolved(AcquisitionOutcome::Failed(error))) = event
         else {
             panic!("flight participant did not receive a failure");
         };
         error
-372
+    }
-373
     #[test]
     fn flight_failure_reaches_every_live_participant() {
         let cell = cell();
         let first = launching_waiter(&cell);
         let second = launching_waiter(&cell);
         let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
         let H2FlightDecision::RunFlight(flight) = cell.converge_h2_flight(first) else {
             panic!("first participant did not become the flight driver");
         };
         assert!(matches!(
             cell.install_or_join_h2_flight(second),
             H2FlightInstall::Joined
             cell.converge_h2_flight(second),
             H2FlightDecision::JoinedFlight
         ));
-386
         let mut completion = FlightCompletionGuard::new(cell.clone(), flight);
         completion.fail(ConnectorError::io(Box::new(std::io::Error::other(
             "synthetic HTTP/2 handshake failure",
         ))));
-391
         for waiter in [first, second] {
             let error = failed_event(&cell, waiter);
             assert!(error.is_io());
             let source = error.source().expect("flight failure lost its source");
             assert_eq!("synthetic HTTP/2 handshake failure", source.to_string());
             assert!(
                 source
                     .source()
                     .expect("flight failure lost its original source")
                     .downcast_ref::<std::io::Error>()
                     .is_some(),
                 "flight failure did not preserve the original error"
             );
-405
+        }
-406
+    }
-407
     #[test]
     fn task_drop_fails_only_participants_still_owned_by_the_flight() {
         let cell = cell();
         let live = launching_waiter(&cell);
         let cancelled = launching_waiter(&cell);
         let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(live) else {
         let H2FlightDecision::RunFlight(flight) = cell.converge_h2_flight(live) else {
             panic!("first participant did not become the flight driver");
         };
         assert!(matches!(
             cell.install_or_join_h2_flight(cancelled),
             H2FlightInstall::Joined
             cell.converge_h2_flight(cancelled),
             H2FlightDecision::JoinedFlight
         ));
         assert!(OriginCell::cancel_waiter(&cell, cancelled));
-421
         drop(FlightCompletionGuard::new(cell.clone(), flight));
-423
         let error = failed_event(&cell, live);
         assert!(error.is_io());
         assert_eq!(
             "HTTP/2 establishment task was dropped",
             error

↧ Expand context

                 .source()
                 .expect("task-drop failure lost its source")
                 .to_string()
         );
         let successor = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
         assert_ne!(cancelled, successor);
         assert!(OriginCell::cancel_waiter(&cell, successor));
-436
+    }
-437
     #[derive(Debug)]
     struct CountingSpawner {
         submissions: StdArc<AtomicUsize>,
-441
+    }
-442
     impl DriverSpawner for CountingSpawner {
         fn spawn(&self, future: std::pin::Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submissions.fetch_add(1, Ordering::Relaxed);
             drop(future);
-447
+        }
-448
+    }

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

@@ -0,1 +0,292 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-+
 //! Transport connection policy below HTTP protocol establishment.
 //!
 //! This module owns the boundary from a pool connection attempt to connected
 //! I/O with a negotiated protocol. [`TransportConnectContext`] keeps placement,
 //! timeout, and ALPN inputs together. A factory may delegate DNS, proxy, TLS,
 //! and ALPN to an injected connector, but it must report whether an HTTP/1-only
 //! attempt is enforceable before admission may reclaim an idle H2 connection.
 //!
 //! Cached factories retain configured connector services, not connections or
 //! DNS results. Network I/O begins only when [`TransportFactory::connect`] is
 //! called.
-+
 use super::super::admission::ProtocolRequirement;
 use super::super::registry::PartitionState;
 use crate::client::connect::{AsyncConn, BoxConn};
 use crate::client::timeout::{self, TimeoutKind};
 use aws_smithy_async::rt::sleep::SharedAsyncSleep;
 use aws_smithy_runtime_api::box_error::BoxError;
 use http_1x::Uri;
 use std::future::{poll_fn, Future};
 use std::pin::Pin;
 use std::sync::Arc as StdArc;
 use std::time::Duration;
 use tower::Service;
-+
 /// Timeout policy for one transport connection operation.
 ///
 /// Connector readiness completes before this timeout starts. Keeping the
 /// duration and timer together lets request acquisition pass one complete
 /// timeout value into transport construction.
 #[derive(Clone, Debug)]
 pub(in crate::client::pool) struct TransportTimeout {
     /// Maximum duration of the connector's connection future.
     duration: Duration,
     /// Runtime timer used to enforce the deadline.
     sleep: SharedAsyncSleep,
-+
+}
-+
 impl TransportTimeout {
     /// Creates a timeout enforced by the supplied runtime timer.
     pub(in crate::client::pool) fn new(duration: Duration, sleep: SharedAsyncSleep) -> Self {
         Self { duration, sleep }
-+
+    }
-+
+}
-+
 /// Static ALPN protocol offer used for one transport connection.
 pub(in crate::client::pool) type AlpnProtocols = &'static [&'static [u8]];
-+
 /// Default offer for requests that may use HTTP/2.
 const HTTP_ALPN_PROTOCOLS: AlpnProtocols = &[b"h2", b"http/1.1"];
-+
 /// Narrowed offer for requests that require HTTP/1 wire semantics.
 const HTTP1_ALPN_PROTOCOLS: AlpnProtocols = &[b"http/1.1"];
-+
 /// Inputs needed to create one transport connection.
 ///
 /// This context is the transport-attempt boundary for DNS, socket, proxy, TLS,
 /// and ALPN lifecycle observation. It carries no installed connection state
 /// because an attempt may fail before a connection exists.
 pub(in crate::client::pool) struct TransportConnectContext<'a> {
     /// Partition whose placement policy owns the connection attempt.
     partition: &'a PartitionState,
     /// Absolute URI passed to the connector contract.
     uri: Uri,
     /// Optional deadline for the connector's connection future.
     timeout: Option<TransportTimeout>,
     /// Protocols the default TLS connector may advertise for this attempt.
     alpn_protocols: AlpnProtocols,
-+
+}
-+
 impl<'a> TransportConnectContext<'a> {
     /// Creates transport inputs after request protocol classification.
     pub(super) fn new(
         partition: &'a PartitionState,
         uri: Uri,
         timeout: Option<TransportTimeout>,
         requirement: ProtocolRequirement,
     ) -> Self {
         Self {
             partition,
             uri,
             timeout,
             alpn_protocols: alpn_protocols(requirement),
-+
+        }
-+
+    }
-+
+}
-+
 /// Future returned by a type-erased transport factory.
 type TransportFuture = Pin<Box<dyn Future<Output = Result<BoxConn, BoxError>> + Send + 'static>>;
-+
 /// Type-erased transport construction below HTTP protocol establishment.
 ///
 /// A pool retains one factory. The concrete connector remains responsible for
 /// the transport stages it implements and for returning connector metadata.
 /// Protocol establishment consumes the resulting [`BoxConn`].
 pub(in crate::client::pool) trait TransportFactory:
     Send + Sync + 'static
-+
+{
     /// Returns whether an H1-required attempt is guaranteed to negotiate H1.
     fn can_guarantee_http1(&self) -> bool;
-+
     /// Creates one partition-bound transport.
     ///
     /// Connector readiness completes before the connection timeout starts.
     fn connect(&self, context: TransportConnectContext<'_>) -> TransportFuture;
-+
+}
-+
 /// Connector factory whose concrete service is selected by interface.
 struct ServiceTransportFactory<F> {
     /// Builds a connector with the selected network-interface binding.
     connector_for_interface: F,
     /// Whether every H1-required connection is guaranteed to negotiate H1.
     can_guarantee_http1: bool,
-+
+}
-+
 impl<F, C, IO> TransportFactory for ServiceTransportFactory<F>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
-+
+{
     fn can_guarantee_http1(&self) -> bool {
         self.can_guarantee_http1
-+
+    }
-+
     fn connect(&self, context: TransportConnectContext<'_>) -> TransportFuture {
         let TransportConnectContext {
             partition,
             uri,
             timeout,
             alpn_protocols: _alpn_protocols,
         } = context;
         let interface = partition.interface().map(|interface| interface.as_ref());
         let mut connector = (self.connector_for_interface)(interface);
         Box::pin(async move {
             // Readiness follows the existing connector contract and is not timed.
             poll_fn(|cx| connector.poll_ready(cx))
                 .await
                 .map_err(Into::into)?;
             let connect = connector.call(uri);
             let io = timeout::maybe_timeout_future(
                 connect,
                 timeout.as_ref().map(|timeout| timeout.duration),
                 timeout.as_ref().map(|timeout| &timeout.sleep),
                 TimeoutKind::Connect,
-+
+            )
             .await?;
             Ok(Box::new(io) as BoxConn)
         })
-+
+    }
-+
+}
-+
 /// Erases an injected connector that owns placement and protocol negotiation.
 ///
 /// The pool cannot apply its interface or ALPN inputs to this connector. It is
 /// therefore conservative about H2 reclaim for H1-required demand.
 #[cfg(any(
     all(feature = "test-util", aws_sdk_unstable),
     all(test, feature = "rt-tokio")
 ))]
 pub(in crate::client::pool) fn from_connector<C, IO>(connector: C) -> StdArc<dyn TransportFactory>
 where
     C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
-+
+{
     service_factory(move |_| connector.clone(), false)
-+
+}
-+
 /// Erases a cleartext connector constructor that applies interface placement.
 pub(in crate::client::pool) fn from_interface_connector<F, C, IO>(
     connector_for_interface: F,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
-+
+{
     service_factory(connector_for_interface, true)
-+
+}
-+
 /// Erases a service connector and its HTTP/1 negotiation guarantee.
 fn service_factory<F, C, IO>(
     connector_for_interface: F,
     can_guarantee_http1: bool,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Send + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
-+
+{
     StdArc::new(ServiceTransportFactory {
         connector_for_interface,
         can_guarantee_http1,
     })
-+
+}
-+
 /// Erases and caches configured TLS connector services.
 ///
 /// Interface placement is immutable for a partition. The first request for
 /// each `(interface, ALPN offer)` constructs its connector while holding the
 /// cache lock; later requests clone that retained connector. Provider
 /// configuration and certificate loading therefore occur at most once for
 /// each placement and offer rather than once per connection.
 #[cfg(any(feature = "__rustls", feature = "s2n-tls"))]
 pub(in crate::client::pool) fn from_cached_interface_connector<F, C, IO>(
     connector_for_interface: F,
     can_guarantee_http1: bool,
 ) -> StdArc<dyn TransportFactory>
 where
     F: Fn(Option<&str>, AlpnProtocols) -> C + Send + Sync + 'static,
     C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
     C::Error: Into<BoxError>,
     C::Future: Send + 'static,
     IO: AsyncConn,
-+
+{
     struct CachedTransportFactory<F, C> {
         factory: F,
         can_guarantee_http1: bool,
         connectors:
             crate::sync::Mutex<std::collections::HashMap<(Option<String>, AlpnProtocols), C>>,
-+
+    }
-+
     impl<F, C, IO> TransportFactory for CachedTransportFactory<F, C>
     where
         F: Fn(Option<&str>, AlpnProtocols) -> C + Send + Sync + 'static,
         C: Service<Uri, Response = IO> + Clone + Send + Sync + 'static,
         C::Error: Into<BoxError>,
         C::Future: Send + 'static,
         IO: AsyncConn,
-+
+    {
         fn can_guarantee_http1(&self) -> bool {
             self.can_guarantee_http1
-+
+        }
-+
         fn connect(&self, context: TransportConnectContext<'_>) -> TransportFuture {
             let TransportConnectContext {
                 partition,
                 uri,
                 timeout,
                 alpn_protocols,
             } = context;
             let interface = partition.interface().map(|value| value.to_string());
             let mut connector = self
                 .connectors
                 .lock()
                 .entry((interface.clone(), alpn_protocols))
                 .or_insert_with(|| (self.factory)(interface.as_deref(), alpn_protocols))
                 .clone();
             Box::pin(async move {
                 poll_fn(|cx| connector.poll_ready(cx))
                     .await
                     .map_err(Into::into)?;
                 let connect = connector.call(uri);
                 let io = timeout::maybe_timeout_future(
                     connect,
                     timeout.as_ref().map(|timeout| timeout.duration),
                     timeout.as_ref().map(|timeout| &timeout.sleep),
                     TimeoutKind::Connect,
-+
+                )
                 .await?;
                 Ok(Box::new(io) as BoxConn)
             })
-+
+        }
-+
+    }
-+
     StdArc::new(CachedTransportFactory {
         factory: connector_for_interface,
         can_guarantee_http1,
         connectors: crate::sync::Mutex::new(std::collections::HashMap::new()),
     })
-+
+}
-+
 /// Selects the ALPN offer that preserves the request's wire semantics.
 fn alpn_protocols(requirement: ProtocolRequirement) -> AlpnProtocols {
     match requirement {
         ProtocolRequirement::H1Required => HTTP1_ALPN_PROTOCOLS,
         ProtocolRequirement::H1Compatible | ProtocolRequirement::H2Required => HTTP_ALPN_PROTOCOLS,
-+
+    }
-+
+}

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

@@ -165,165 +227,227 @@

↥ Expand context

         if state.shutdown {
             return ScheduleResult::Shutdown;
-167
+        }
         if state.revision != observed {
             return ScheduleResult::Retry;
-170
+        }
         state.scheduled_deadline = deadline;
         ScheduleResult::Wait {
             revision: state.revision,
             deadline,
-175
+        }
-176
+    }
-177
     /// Polls until a deadline publication or shutdown changes `observed`.
     fn poll_revision(&self, observed: u64, cx: &Context<'_>) -> Poll<()> {
         let mut state = self.state.lock();
         if state.shutdown || state.revision != observed {
             return Poll::Ready(());
-183
+        }
         if state

             .waker
             .as_ref()
             .is_none_or(|registered| !registered.will_wake(cx.waker()))
-188
+        {
             state.waker = Some(cx.waker().clone());
-190
+        }
         Poll::Pending
-192
+    }
-193
     #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
     fn snapshot(&self) -> MaintenanceSnapshot {
     fn probe(&self) -> MaintenanceProbe {
         let state = self.state.lock();
         MaintenanceSnapshot {
         MaintenanceProbe {
             started: self.started.load(Ordering::Acquire),
             shutdown: state.shutdown,
             scheduled_deadline: state.scheduled_deadline,
-201
+        }
-202
+    }
-203
+}
-204
 /// Mutable scheduler state protected independently from every origin cell.
 #[derive(Debug, Default)]
 struct MaintenanceState {

↧ Expand context

     /// Retained partition cells indexed for replacement on duplicate publish.
     cells: HashMap<CellId, Weak<OriginCell>>,
     /// Monotonic change counter observed by the maintenance future.
     revision: u64,
     /// Task waiting for a new revision.
     waker: Option<Waker>,
     /// Deadline the current sleep represents, if any.
     scheduled_deadline: Option<SystemTime>,
     /// Whether pool drop requested task termination.
     shutdown: bool,
-218
+}
-219
 impl MaintenanceState {
     /// Advances the wake revision and detaches the current task waker.
     fn signal(&mut self) -> Option<Waker> {
         self.revision = self
             .revision
             .checked_add(1)
             .expect("partition maintenance revision exhausted");
         self.waker.take()

@@ -351,351 +411,411 @@

↥ Expand context

     /// The sleep for the nearest observed deadline completed.
     DeadlineElapsed,
     /// Idle residence changed or the scheduler shut down.
     Revision,
-355
+}
-356
 /// Races one pool-provided sleep with a scheduler publication.
 async fn wait_for_sleep_or_revision(
     scheduler: &Weak<PartitionMaintenance>,
     observed: u64,
     sleep: SharedAsyncSleep,
     duration: Duration,
 ) -> MaintenanceWake {
     let mut sleeping = Box::pin(sleep.sleep(duration));
     poll_fn(|cx| {
         if sleeping.as_mut().poll(cx).is_ready() {
             return Poll::Ready(MaintenanceWake::DeadlineElapsed);
-368
+        }
         scheduler
             .upgrade()

             .map_or(Poll::Ready(MaintenanceWake::Revision), |current| {
                 current
                     .poll_revision(observed, cx)
                     .map(|()| MaintenanceWake::Revision)
             })
     })
     .await
-378
+}
-379
 #[cfg(all(test, feature = "rt-tokio", not(smithy_http_client_loom)))]
 struct MaintenanceSnapshot {
 struct MaintenanceProbe {
     started: bool,
     shutdown: bool,
     scheduled_deadline: Option<SystemTime>,
-385
+}
-386
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use std::sync::atomic::{AtomicUsize, Ordering as StdOrdering};
     use std::sync::Arc as StdArc;

↧ Expand context

     use std::task::{Context, Wake, Waker};
     use std::time::UNIX_EPOCH;
-394
     struct WakeCounter(AtomicUsize);
-396
     impl Wake for WakeCounter {
         fn wake(self: StdArc<Self>) {
             self.0.fetch_add(1, StdOrdering::SeqCst);
-400
+        }
-401
         fn wake_by_ref(self: &StdArc<Self>) {
             self.0.fetch_add(1, StdOrdering::SeqCst);
-404
+        }
-405
+    }
-406
     #[test]
     fn deadline_publication_wakes_a_registered_task() {
         loom::model(|| {
             let maintenance = PartitionMaintenance::new(MaintenanceConfig::default());
             let observed = maintenance

@@ -489,489 +569,569 @@

↥ Expand context

     use crate::client::pool::origin::OriginKey;
     use crate::client::pool::partition::{EligibilityGroup, PartitionId, TokioDriverSpawner};
     use aws_smithy_async::test_util::controlled_time_and_sleep;
     use aws_smithy_async::time::TimeSource;
     use aws_smithy_runtime_api::client::connection::ConnectionId;
     use http_1x::uri::Scheme;
     use std::sync::atomic::{AtomicU64, Ordering};
     use std::time::{Duration, UNIX_EPOCH};
-497
     #[derive(Debug)]
     struct DroppingSpawner {
         submitted: Arc<AtomicU64>,
-501
+    }
-502
     impl DriverSpawner for DroppingSpawner {
         fn spawn(&self, driver: std::pin::Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             self.submitted.fetch_add(1, Ordering::SeqCst);
             drop(driver);
-507
+        }
-508
+    }

-509
     #[test]
     fn discarded_task_releases_the_start_latch() {
         let (maintenance, _cell, _gate) = managed_cell(Duration::from_secs(10));
         let submitted = Arc::new(AtomicU64::new(0));
         let spawner = DroppingSpawner {
             submitted: submitted.clone(),
         };
-517
         PartitionMaintenance::start(&maintenance, &spawner);
         assert!(!maintenance.snapshot().started);
         assert!(!maintenance.probe().started);
         PartitionMaintenance::start(&maintenance, &spawner);
         assert_eq!(2, submitted.load(Ordering::SeqCst));
         assert!(!maintenance.snapshot().started);
         assert!(!maintenance.probe().started);
-523
+    }
-524
     #[test]
     fn disabled_idle_timeout_submits_no_maintenance_task() {
         let maintenance = PartitionMaintenance::new(MaintenanceConfig {
             idle_timeout: None,
             ..MaintenanceConfig::default()
         });
         let submitted = Arc::new(AtomicU64::new(0));
         let spawner = DroppingSpawner {
             submitted: submitted.clone(),
         };
-535
         PartitionMaintenance::start(&maintenance, &spawner);
-537
         assert_eq!(0, submitted.load(Ordering::SeqCst));
         assert!(!maintenance.snapshot().started);
         assert!(!maintenance.probe().started);
-540
+    }
-541
     #[derive(Clone, Debug)]
     struct RewindableTimeSource(Arc<AtomicU64>);
-544
     impl RewindableTimeSource {
         fn new(seconds: u64) -> Self {
             Self(Arc::new(AtomicU64::new(seconds)))
-548
+        }
-549

↧ Expand context

         fn set(&self, seconds: u64) {
             self.0.store(seconds, Ordering::SeqCst);
-552
+        }
-553
+    }
-554
     impl TimeSource for RewindableTimeSource {
         fn now(&self) -> SystemTime {
             UNIX_EPOCH + Duration::from_secs(self.0.load(Ordering::SeqCst))
-558
+        }
-559
+    }
-560
     fn cell_with_maintenance(
         timeout: Duration,
         time_source: SharedTimeSource,
         sleep: SharedAsyncSleep,
     ) -> (Arc<PartitionMaintenance>, Arc<OriginCell>) {
         let maintenance = PartitionMaintenance::new(MaintenanceConfig {
             idle_timeout: Some(timeout),
             time_source,
             sleep: Some(sleep),

@@ -642,642 +848,848 @@

↥ Expand context

             active.fetch_add(1, Ordering::SeqCst);
             tokio::spawn(async move {
                 struct ActiveGuard(Arc<AtomicU64>);
                 impl Drop for ActiveGuard {
                     fn drop(&mut self) {
                         self.0.fetch_sub(1, Ordering::SeqCst);
-648
+                    }
-649
+                }
                 let _guard = ActiveGuard(active);
                 driver.await;
             });
-653
+        }
-654
+    }
-655
     #[tokio::test]
     async fn start_is_once_and_shutdown_terminates_the_task() {
         let (maintenance, _cell, _gate) = managed_cell(Duration::from_secs(10));
         let spawner = TrackingSpawner::new();
-660
         for _ in 0..10 {

             PartitionMaintenance::start(&maintenance, &spawner);
-663
+        }
         for _ in 0..10 {
             if spawner.active.load(Ordering::SeqCst) == 1 {
                 break;
-667
+            }
             tokio::task::yield_now().await;
-669
+        }
         assert_eq!(1, spawner.submitted.load(Ordering::SeqCst));
         assert_eq!(1, spawner.active.load(Ordering::SeqCst));
         assert!(maintenance.snapshot().started);
         assert!(maintenance.probe().started);
-673
         maintenance.shutdown();
         for _ in 0..10 {
             if spawner.active.load(Ordering::SeqCst) == 0 {
                 break;
-678
+            }
             tokio::task::yield_now().await;
-680
+        }
         assert!(maintenance.snapshot().shutdown);
         assert!(maintenance.probe().shutdown);
         assert_eq!(0, spawner.active.load(Ordering::SeqCst));
-683
+    }
-684
     #[tokio::test]
     async fn idle_insertion_wakes_a_task_with_no_scheduled_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
         tokio::task::yield_now().await;
         assert_eq!(None, maintenance.snapshot().scheduled_deadline);
         assert_eq!(None, maintenance.probe().scheduled_deadline);
-692
         OriginCell::install_idle_h1(&cell, connection(1), H1Sender::test(1));
         OriginCell::insert_idle_h1(&cell, connection(1), H1Sender::test(1));
         let sleep = gate.expect_sleep().await;
         assert_eq!(timeout, sleep.duration());
         assert!(maintenance.snapshot().scheduled_deadline.is_some());
         assert!(maintenance.probe().scheduled_deadline.is_some());
         maintenance.shutdown();
-698
+    }
-699
     #[test]
     fn only_an_earlier_deadline_advances_the_revision() {
         let (time, sleep, _gate) = controlled_time_and_sleep(UNIX_EPOCH);
         let maintenance = PartitionMaintenance::new(MaintenanceConfig {
             idle_timeout: Some(Duration::from_secs(10)),
             time_source: SharedTimeSource::new(time),
             sleep: Some(SharedAsyncSleep::new(sleep)),
         });
         let later = UNIX_EPOCH + Duration::from_secs(20);
         let latest = UNIX_EPOCH + Duration::from_secs(30);
         let earlier = UNIX_EPOCH + Duration::from_secs(10);
-711
         maintenance.notify_deadline(Some(later));
         let revision = maintenance.state.lock().revision;
         maintenance.notify_deadline(Some(latest));
         assert_eq!(revision, maintenance.state.lock().revision);
         maintenance.notify_deadline(Some(earlier));
         assert!(maintenance.state.lock().revision > revision);
         assert_eq!(Some(earlier), maintenance.snapshot().scheduled_deadline);
         assert_eq!(Some(earlier), maintenance.probe().scheduled_deadline);
-719
+    }
-720
     #[tokio::test]
     async fn idle_record_closes_at_its_fake_time_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         let connection = connection(1);
         OriginCell::install_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         OriginCell::insert_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-728
         let sleep = gate.expect_sleep().await;
         assert_eq!(timeout, sleep.duration());
         assert_eq!(None, connection.snapshot().close_reason);
         assert_eq!(None, connection.probe().close_reason);
         sleep.allow_progress();
-733
         for _ in 0..10 {
             if connection.snapshot().close_reason == Some(super::super::CloseReason::IdleTimeout) {
             if connection.probe().close_reason == Some(super::super::CloseReason::IdleTimeout) {
                 break;
-737
+            }
             tokio::task::yield_now().await;
-739
+        }
         assert_eq!(
             Some(super::super::CloseReason::IdleTimeout),
             connection.snapshot().close_reason
             connection.probe().close_reason
         );
-744
+    }
     #[tokio::test]
     async fn h2_generation_closes_at_its_fake_time_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         let connection = h2_connection(1);
         OriginCell::install_h2_for_test(&cell, connection.clone(), 1, maintenance.idle_deadline());
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-752
         let sleep = gate.expect_sleep().await;
         assert_eq!(timeout, sleep.duration());
         assert_eq!(None, connection.snapshot().close_reason);
         assert_eq!(None, connection.probe().close_reason);
         sleep.allow_progress();
-757
         for _ in 0..10 {
             if connection.snapshot().close_reason == Some(super::super::CloseReason::IdleTimeout) {
             if connection.probe().close_reason == Some(super::super::CloseReason::IdleTimeout) {
                 break;
-761
+            }
             tokio::task::yield_now().await;
-763
+        }
         assert_eq!(
             Some(super::super::CloseReason::IdleTimeout),
             connection.snapshot().close_reason
             connection.probe().close_reason
         );
         assert_eq!(None, cell.accepting_h2_generation());
-769
+    }
-770
     #[test]
     fn accepted_h2_dispatch_resets_the_idle_deadline() {
         let timeout = Duration::from_secs(10);
         let time = RewindableTimeSource::new(0);
         let (_unused_time, sleep, _gate) = controlled_time_and_sleep(UNIX_EPOCH);
         let (maintenance, cell) = cell_with_maintenance(
             timeout,
             SharedTimeSource::new(time.clone()),
             SharedAsyncSleep::new(sleep),
         );
         let connection = h2_connection(1);
         OriginCell::install_h2_for_test(&cell, connection.clone(), 1, maintenance.idle_deadline());
         assert_eq!(Some(UNIX_EPOCH + timeout), cell.nearest_idle_deadline());
-784
         time.set(20);
         let mut activation =
             OriginCell::select_h2(&cell).expect("HTTP/2 generation was not selectable");
         let _dispatch_parts = activation.take_dispatch_parts();
         let dispatch = ConnectionState::try_commit_dispatch(&connection)
             .expect("HTTP/2 connection rejected dispatch");
         activation.accept(dispatch);
-792
         assert_eq!(
             Some(UNIX_EPOCH + Duration::from_secs(20) + timeout),
             cell.nearest_idle_deadline()
         );
-797
+    }
-798
     #[tokio::test]
     async fn selected_record_has_no_idle_deadline() {
         let timeout = Duration::from_secs(10);
         let (maintenance, cell, mut gate) = managed_cell(timeout);
         let connection = connection(1);
         OriginCell::install_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         OriginCell::insert_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-806
         let sleep = gate.expect_sleep().await;
         let selection = OriginCell::select_h1(&cell).expect("idle H1 was not selected");
         sleep.allow_progress();
         for _ in 0..10 {
             tokio::task::yield_now().await;
-812
+        }
-813
         assert_eq!(None, connection.snapshot().close_reason);
         assert_eq!(None, connection.probe().close_reason);
         drop(selection);
-816
+    }
-817
     #[tokio::test]
     async fn completed_sleep_is_an_expiration_floor_after_clock_moves_backward() {
         let timeout = Duration::from_secs(10);
         let clock = RewindableTimeSource::new(100);
         let (_unused_time, sleep, mut gate) = controlled_time_and_sleep(UNIX_EPOCH);
         let (maintenance, cell) = cell_with_maintenance(
             timeout,
             SharedTimeSource::new(clock.clone()),
             SharedAsyncSleep::new(sleep),
         );
         let connection = connection(1);
         OriginCell::install_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         OriginCell::insert_idle_h1(&cell, connection.clone(), H1Sender::test(1));
         PartitionMaintenance::start(&maintenance, &TokioDriverSpawner::current());
-831
         let sleep = gate.expect_sleep().await;
         assert_eq!(timeout, sleep.duration());
         clock.set(90);
         sleep.allow_progress();
-836
         for _ in 0..10 {
             if connection.snapshot().close_reason == Some(super::super::CloseReason::IdleTimeout) {
             if connection.probe().close_reason == Some(super::super::CloseReason::IdleTimeout) {
                 break;
-840
+            }
             tokio::task::yield_now().await;
-842
+        }
         assert_eq!(
             Some(super::super::CloseReason::IdleTimeout),
             connection.snapshot().close_reason
             connection.probe().close_reason
         );
-847
+    }
-848
+}

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

@@ -162,162 +222,222 @@

↥ Expand context

-162
+    }
-163
+}
-164
 #[cfg(feature = "rt-tokio")]
 impl DriverSpawner for TokioDriverSpawner {
     fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
         drop(self.handle.spawn(driver));
-169
+    }
-170
+}
-171
 /// Controls which partitions may dispatch through each other's connections.
 #[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum ConnectionReuseScope {
     /// A connection serves only requests from its owning partition.
     Partition,
     /// Partitions with the same interface binding may share connections.
     #[default]
     NetworkInterface,
     /// Every partition in the pool may share connections.

     Pool,
-183
+}
-184
 /// Identifies the exact set of partitions allowed to share a connection.
 ///
 /// The configured [`ConnectionReuseScope`] maps each partition to one group:
 /// partition-local scope creates a group for one partition, network-interface
 /// scope groups equal interface bindings, and pool scope creates one group for
 /// the whole pool.
 #[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub(crate) enum EligibilityGroup {
 pub(in crate::client::pool) enum EligibilityGroup {
     /// Only the partition with this identity is eligible.
     Partition(PartitionId),
     /// Partitions with this exact interface binding are eligible.
     NetworkInterface(Option<Arc<str>>),
     /// Every partition in the pool is eligible.
     Pool,
-199
+}
-200
 #[cfg(test)]
 mod tests {

↧ Expand context

     use super::*;
     #[cfg(feature = "rt-tokio")]
     use std::sync::atomic::{AtomicBool, Ordering};
-206
     #[derive(Debug)]
     struct TestSpawner;
-209
     impl DriverSpawner for TestSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-213
+        }
-214
+    }
-215
     #[test]
     fn anonymous_identity_is_reserved() {
         assert!(PartitionId::ANONYMOUS.is_anonymous());
         assert!(PartitionId::from_index(usize::MAX).is_anonymous());
         assert!(!PartitionId::from_index(0).is_anonymous());
-221
+    }
-222

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

@@ -1,1 +307,329 @@

↥ Expand context

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Pool topology and stable origin-cell publication.
 //!
 //! [`PartitionRegistry`] retains the fixed partition set and one shared
 //! [`OriginAdmission`] for each bounded origin. Each [`PartitionState`] owns
 //! runtime placement and lazily retains one [`OriginCell`] per canonical
 //! origin.
-12
 use super::admission::OriginAdmission;
 use super::cell::OriginCell;
 use super::connection::CloseReason;
 use super::maintenance::{MaintenanceConfig, PartitionMaintenance};
 use super::origin::{InvalidOrigin, OriginKey, OriginLookup, SchemeKey};
 #[cfg(feature = "rt-tokio")]
 use super::partition::TokioDriverSpawner;
 use super::partition::{

     ConnectionReuseScope, DriverSpawner, EligibilityGroup, Partition, PartitionId,
 };
 use crate::sync::{Arc, Mutex, RwLock};
 use http_1x::Uri;
 use std::collections::HashMap;
 use std::error::Error;
 use std::fmt;
 use std::num::{NonZeroU16, NonZeroUsize};
 use std::sync::{Arc as StdArc, OnceLock};
-30
 /// Bounded-origin policy shared by every partition cell for that origin.
 #[derive(Clone, Copy, Debug)]
 pub(super) struct AdmissionPolicy {
     /// Maximum number of logically open connections for one canonical origin.
     connection_limit: NonZeroUsize,
     /// Whether replacement establishment can be forced to HTTP/1.
     can_reclaim_h2_for_h1: bool,
-+
+}
-+
 impl AdmissionPolicy {
     pub(super) fn new(connection_limit: NonZeroUsize, can_reclaim_h2_for_h1: bool) -> Self {
         Self {
             connection_limit,
             can_reclaim_h2_for_h1,
-+
+        }
-+
+    }
-+
     pub(super) fn connection_limit(self) -> NonZeroUsize {
         self.connection_limit
-+
+    }
-+
     pub(super) fn can_reclaim_h2_for_h1(self) -> bool {
         self.can_reclaim_h2_for_h1
-+
+    }
-+
+}
-+
 /// Fixed partition set and the origin-wide admission states it shares.
 #[derive(Debug)]
 pub(crate) struct PartitionRegistry {
 pub(in crate::client::pool) struct PartitionRegistry {
     /// Immutable partition identities resolved when a client is built.
     partitions: HashMap<PartitionId, Arc<PartitionState>>,
     /// Policy used to derive a new cell's eligibility group.
     reuse_scope: ConnectionReuseScope,
     /// Optional origin-wide bound used when admission is first created.
     max_connections_per_host: Option<NonZeroUsize>,
     /// Whether admission may close idle H2 capacity for H1-required demand.
     allow_h2_reclaim_for_h1: bool,
     /// Bounded-origin policy, absent when connection count is unbounded.
     admission_policy: Option<AdmissionPolicy>,
     /// Admission authorities retained by canonical origin.
     bounded_origins: Mutex<HashMap<OriginKey, Arc<OriginAdmission>>>,
     admissions: Mutex<HashMap<OriginKey, Arc<OriginAdmission>>>,
-68
+}
-69
 impl PartitionRegistry {
     /// Creates the anonymous partition or validates and retains explicit partitions.
     pub(crate) fn new(
     pub(in crate::client::pool) fn new(
         partitions: Option<Vec<Partition>>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
         allow_h2_reclaim_for_h1: bool,
         admission_policy: Option<AdmissionPolicy>,
         maintenance: MaintenanceConfig,
     ) -> Result<Self, PartitionRegistryError> {
         let Some(partitions) = partitions else {
             let partition = Arc::new(PartitionState::anonymous(maintenance));
             let mut partitions = HashMap::new();
             partitions.insert(PartitionId::ANONYMOUS, partition);
             return Ok(Self {
                 partitions,
                 reuse_scope,
                 max_connections_per_host,
                 allow_h2_reclaim_for_h1,
                 bounded_origins: Mutex::new(HashMap::new()),
                 admission_policy,
                 admissions: Mutex::new(HashMap::new()),
             });
         };
-89
         let mut by_id = HashMap::new();
         for partition in partitions {
             let id = partition.id();
             if id.is_anonymous() {
                 return Err(PartitionRegistryError::ReservedAnonymousPartition);
-95
+            }
             if by_id
                 .insert(
                     id,
                     Arc::new(PartitionState::explicit(partition, maintenance.clone())),
-100
+                )
                 .is_some()
-102
+            {
                 return Err(PartitionRegistryError::DuplicatePartition(id));
-104
+            }
-105
+        }
-106
         if by_id.is_empty() {
             return Err(PartitionRegistryError::EmptyExplicitPartitionSet);
-109
+        }
-110
         Ok(Self {
             partitions: by_id,
             reuse_scope,
             max_connections_per_host,
             allow_h2_reclaim_for_h1,
             bounded_origins: Mutex::new(HashMap::new()),
             admission_policy,
             admissions: Mutex::new(HashMap::new()),
         })
-117
+    }
-118
     /// Resolves a retained partition once during client construction.
     pub(crate) fn partition(&self, id: PartitionId) -> Option<Arc<PartitionState>> {
     pub(in crate::client::pool) fn partition(
         &self,
         id: PartitionId,
     ) -> Option<Arc<PartitionState>> {
         self.partitions.get(&id).cloned()
-125
+    }
-126
     /// Resolves the stable cell for an already-resolved partition and URI.
     pub(crate) fn resolve_cell(
     pub(in crate::client::pool) fn resolve_cell(
         &self,
         partition: &PartitionState,
         uri: &Uri,
     ) -> Result<Arc<OriginCell>, InvalidOrigin> {
         let lookup = OriginLookup::from_uri(uri)?;
         if let Some(cell) = partition.find_cell(&lookup) {
             return Ok(cell);
-136
+        }
-137
         let eligibility_group = self.eligibility_group(partition);
         let origin = lookup.into_owned();
         let admission = self.origin_admission(&origin);
         let candidate = Arc::new(OriginCell::new(
             partition.id(),
             origin,
             eligibility_group,
             admission.clone(),
             Some(partition.maintenance.clone()),
         ));
         let cell = match admission {
             Some(admission) => OriginAdmission::register_cell(&admission, candidate),
             None => candidate,
         };
         let cell = partition.publish_cell(cell);
         partition.maintenance.register(&cell);
         Ok(cell)
-155
+    }
-156
     /// Returns the shared admission authority for a bounded origin.
     ///
     /// Unbounded registries return `None`. Concurrent first use of a bounded
     /// origin converges on one retained authority.
     fn origin_admission(&self, origin: &OriginKey) -> Option<Arc<OriginAdmission>> {
         let limit = self.max_connections_per_host?;
         let policy = self.admission_policy?;
         let admission = {
             let mut origins = self.bounded_origins.lock();
             let mut origins = self.admissions.lock();
             origins
                 .entry(origin.clone())
                 .or_insert_with(|| {
                     OriginAdmission::new(origin.clone(), limit, self.allow_h2_reclaim_for_h1)
                 })
                 .or_insert_with(|| OriginAdmission::new(origin.clone(), policy))
                 .clone()
         };
         Some(admission)
-171
+    }
-172
     /// Derives which partitions may relieve a new cell's bounded demand.
     fn eligibility_group(&self, partition: &PartitionState) -> EligibilityGroup {
         match self.reuse_scope {
             ConnectionReuseScope::Partition => EligibilityGroup::Partition(partition.id()),
             ConnectionReuseScope::NetworkInterface => {
                 EligibilityGroup::NetworkInterface(partition.interface().cloned())
-179
+            }
             ConnectionReuseScope::Pool => EligibilityGroup::Pool,
-181
+        }
-182
+    }
-183
     /// Stops partition maintenance and logically closes every connection.
     pub(crate) fn close_all(&self, reason: CloseReason) {
     pub(in crate::client::pool) fn close_all(&self, reason: CloseReason) {
         for partition in self.partitions.values() {
             partition.shutdown_maintenance();
-188
+        }
         let cells = self
             .partitions
             .values()
             .flat_map(|partition| partition.cells())
             .collect::<Vec<_>>();
         for cell in cells {
             OriginCell::close_all(&cell, reason);
-196
+        }
-197
+    }
-198
+}
-199
 /// Runtime placement and retained origin cells for one partition.
 #[derive(Debug)]
 pub(crate) struct PartitionState {
 pub(in crate::client::pool) struct PartitionState {
     /// Stable identity copied into every cell and owned connection.
     id: PartitionId,
     /// Configured spawner, or the first spawner published for the anonymous
     /// partition.
     spawner: OnceLock<StdArc<dyn DriverSpawner>>,
     /// Network interface used for placement and reuse eligibility.
     interface: Option<StdArc<str>>,
     /// Cells retained for the lifetime of this partition.
     origins: RwLock<OriginMap>,
     /// Owner-runtime idle maintenance for this partition.
     maintenance: Arc<PartitionMaintenance>,
-214
+}
-215
 impl PartitionState {
     /// Creates the implicit partition whose spawner is published on first use.
     fn anonymous(maintenance: MaintenanceConfig) -> Self {
         Self {
             id: PartitionId::ANONYMOUS,
             spawner: OnceLock::new(),
             interface: None,
             origins: RwLock::new(OriginMap::default()),
             maintenance: PartitionMaintenance::new(maintenance),
-225
+        }
-226
+    }
-227
     /// Moves one validated explicit declaration into retained partition state.
     fn explicit(partition: Partition, maintenance: MaintenanceConfig) -> Self {
         let (id, spawner, interface) = partition.into_parts();
         Self {
             id,
             spawner: OnceLock::from(spawner),
             interface,
             origins: RwLock::new(OriginMap::default()),
             maintenance: PartitionMaintenance::new(maintenance),
-237
+        }
-238
+    }
-239
     /// Returns this partition's stable identity.
     pub(crate) fn id(&self) -> PartitionId {
     pub(in crate::client::pool) fn id(&self) -> PartitionId {
         self.id
-243
+    }
-244
     /// Returns this partition's optional network-interface binding.
     pub(crate) fn interface(&self) -> Option<&StdArc<str>> {
     pub(in crate::client::pool) fn interface(&self) -> Option<&StdArc<str>> {
         self.interface.as_ref()
-248
+    }
-249
     /// Returns the runtime that owns connection drivers for this partition.
     ///
     /// Explicit partitions retain their declared spawner. The anonymous
     /// partition captures the first Tokio runtime on which it is used, and
     /// all later requests use that same runtime.
     pub(crate) fn owner_spawner(
     pub(in crate::client::pool) fn owner_spawner(
         &self,
     ) -> Result<StdArc<dyn DriverSpawner>, MissingAnonymousRuntime> {
         if let Some(spawner) = self.spawner.get() {
             return Ok(spawner.clone());
-260
+        }
-261
         if !self.id.is_anonymous() {
             unreachable!("an explicit partition always has a driver spawner");
-264
+        }
-265
         #[cfg(feature = "rt-tokio")]
-267
+        {
             let handle =
                 tokio::runtime::Handle::try_current().map_err(|_| MissingAnonymousRuntime)?;
             Ok(self
                 .spawner
                 .get_or_init(|| StdArc::new(TokioDriverSpawner::from_handle(handle)))
                 .clone())
-274
+        }
-275
         #[cfg(not(feature = "rt-tokio"))]
-277
+        {
             Err(MissingAnonymousRuntime)
-279
+        }
-280
+    }
-281
     /// Looks up a cell without materializing an owned origin key.
     fn find_cell(&self, lookup: &OriginLookup<'_>) -> Option<Arc<OriginCell>> {
         self.origins.read().get(lookup)
-285
+    }
-286
     /// Publishes `cell` or returns the value that won concurrent publication.
     fn publish_cell(&self, cell: Arc<OriginCell>) -> Arc<OriginCell> {
         self.origins.write().get_or_insert(cell)
-290
+    }
-291
     /// Returns the number of cells retained by this partition.
     #[cfg(test)]
     pub(crate) fn cell_count(&self) -> usize {
     pub(in crate::client::pool) fn cell_count(&self) -> usize {
         self.origins.read().len()
-296
+    }
-297
     /// Ensures idle maintenance is running on this partition's owner runtime.
     pub(crate) fn ensure_maintenance_started(&self, spawner: &dyn DriverSpawner) {
     pub(in crate::client::pool) fn ensure_maintenance_started(&self, spawner: &dyn DriverSpawner) {
         PartitionMaintenance::start(&self.maintenance, spawner);
-301
+    }
-302
     /// Stops this partition's maintenance task during pool teardown.
     fn shutdown_maintenance(&self) {
         self.maintenance.shutdown();
-306
+    }
-307
     /// Snapshots retained cells before invoking any cell transition.
     fn cells(&self) -> Vec<Arc<OriginCell>> {

↧ Expand context

         self.origins.read().cells()
-311
+    }
-312
+}
-313
 /// Retained origin cells owned by one partition.
 #[derive(Default, Debug)]
 struct OriginMap {
     /// Host maps grouped by scheme and non-default port for borrowed lookup.
     indexes: HashMap<SchemePortKey, HashMap<StdArc<str>, Arc<OriginCell>>>,
     /// Cached cell total, avoiding a scan across every first-level bucket.
     cell_count: usize,
-321
+}
-322
 impl OriginMap {
     /// Resolves a cell through borrowed scheme, port, and host components.
     fn get(&self, lookup: &OriginLookup<'_>) -> Option<Arc<OriginCell>> {
         self.indexes
             .get(&SchemePortKey::from_lookup(lookup))?
             .get(lookup.host())
             .cloned()

@@ -336,358 +480,504 @@

↥ Expand context

     /// Clones every retained cell without holding a lock during callbacks.
     fn cells(&self) -> Vec<Arc<OriginCell>> {
         self.indexes
             .values()
             .flat_map(|hosts| hosts.values().cloned())
             .collect()
-364
+    }
-365
+}
-366
 /// Scheme and non-default port used for the first level of origin lookup.
 #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
 struct SchemePortKey {
     /// Canonical HTTP or HTTPS scheme.
     scheme: SchemeKey,
     /// Canonical non-default port.
     port: Option<NonZeroU16>,
-374
+}
-375
 impl SchemePortKey {
     /// Extracts the first-level index from a canonical borrowed lookup.

     fn from_lookup(lookup: &OriginLookup<'_>) -> Self {
         Self {
             scheme: lookup.scheme(),
             port: lookup.port(),
-382
+        }
-383
+    }
-384
+}
-385
 /// Error returned for an invalid explicit partition set.
 #[derive(Clone, Debug, Eq, PartialEq)]
 pub(crate) enum PartitionRegistryError {
 pub(in crate::client::pool) enum PartitionRegistryError {
     /// No partitions were declared.
     EmptyExplicitPartitionSet,
     /// More than one partition used the same identity.
     DuplicatePartition(PartitionId),
     /// An explicit partition used the identity reserved for the default.
     ReservedAnonymousPartition,
-395
+}
-396
 impl fmt::Display for PartitionRegistryError {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             Self::EmptyExplicitPartitionSet => {
                 f.write_str("explicit partition set must not be empty")
-402
+            }
             Self::DuplicatePartition(id) => write!(f, "duplicate partition identifier: {id:?}"),
             Self::ReservedAnonymousPartition => {
                 f.write_str("the anonymous partition identifier is reserved")
-406
+            }
-407
+        }
-408
+    }
-409
+}
-410
 impl Error for PartitionRegistryError {}
-412
 /// Anonymous partition use requires a runtime that can own connection tasks.
 #[derive(Debug)]
 pub(super) struct MissingAnonymousRuntime;
-416
 impl fmt::Display for MissingAnonymousRuntime {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.write_str(
             "the anonymous connection-pool partition requires an active Tokio runtime on first use",
-421
+        )
-422
+    }
-423
+}
-424
 impl Error for MissingAnonymousRuntime {}
-426
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
     use crate::client::pool::admission::ProtocolRequirement;
     use std::future::Future;
     use std::pin::Pin;
     use std::sync::Barrier;
-434
     #[derive(Debug)]
     struct TestSpawner;
-437
     impl DriverSpawner for TestSpawner {
         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-441
+        }
-442
+    }
-443
     fn partition(index: usize) -> Partition {
         Partition::new(PartitionId::from_index(index), TestSpawner)
-446
+    }
-447
     fn admission_policy(limit: Option<NonZeroUsize>) -> Option<AdmissionPolicy> {
         limit.map(|limit| AdmissionPolicy::new(limit, true))
-+
+    }
-+
     #[cfg(feature = "rt-tokio")]
     fn anonymous_registry(
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> PartitionRegistry {
         PartitionRegistry::new(
             None,
             reuse_scope,
             max_connections_per_host,
             true,
             admission_policy(max_connections_per_host),
             MaintenanceConfig::default(),
-462
+        )
         .unwrap()
-464
+    }
-465
     fn explicit_registry(
         partitions: impl IntoIterator<Item = Partition>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> Result<PartitionRegistry, PartitionRegistryError> {
         PartitionRegistry::new(
             Some(partitions.into_iter().collect()),
             reuse_scope,
             max_connections_per_host,
             true,
             admission_policy(max_connections_per_host),
             MaintenanceConfig::default(),
-476
+        )
-477
+    }
-478
     fn explicit_partition(registry: &PartitionRegistry, index: usize) -> Arc<PartitionState> {
         registry
             .partition(PartitionId::from_index(index))
             .expect("test partition was not registered")
-483
+    }
-484

↧ Expand context

     #[test]
     #[cfg(feature = "rt-tokio")]
     fn anonymous_partition_publishes_one_spawner() {
         const THREADS: usize = 8;
-489
         let runtime = tokio::runtime::Builder::new_current_thread()
             .build()
             .unwrap();
         let handle = runtime.handle().clone();
         let registry = Arc::new(anonymous_registry(
             ConnectionReuseScope::NetworkInterface,
             None,
         ));
         let partition = registry.partition(PartitionId::ANONYMOUS).unwrap();
         let barrier = Arc::new(std::sync::Barrier::new(THREADS));
         let mut threads = Vec::new();
-501
         for _ in 0..THREADS {
             let handle = handle.clone();
             let partition = partition.clone();

@@ -665,689 +725,749 @@

↥ Expand context

     fn bounded_origins_share_one_admission_across_partitions() {
         let registry = explicit_registry(
             [partition(1), partition(2)],
             ConnectionReuseScope::Pool,
             NonZeroUsize::new(1),
-694
+        )
         .unwrap();
         let partition_1 = explicit_partition(&registry, 1);
         let partition_2 = explicit_partition(&registry, 2);
         let uri: Uri = "https://example.com/".parse().unwrap();
         let other: Uri = "https://other.example.com/".parse().unwrap();
-700
         let first = registry.resolve_cell(&partition_1, &uri).unwrap();
         let second = registry.resolve_cell(&partition_2, &uri).unwrap();
         let other = registry.resolve_cell(&partition_1, &other).unwrap();
-704
         let admission = first.admission().unwrap();
         assert!(Arc::ptr_eq(admission, second.admission().unwrap()));
         assert!(!Arc::ptr_eq(admission, other.admission().unwrap()));
-708
+    }

-709
     #[test]
     fn unbounded_origins_construct_no_admission_state() {
         let registry = explicit_registry([partition(1)], ConnectionReuseScope::Pool, None).unwrap();
         let partition = explicit_partition(&registry, 1);
         let cell = registry
             .resolve_cell(&partition, &"https://example.com/".parse().unwrap())
             .unwrap();
-717
         assert!(cell.admission().is_none());
         assert!(registry.bounded_origins.lock().is_empty());
         assert!(registry.admissions.lock().is_empty());
-720
+    }
-721
     #[test]
     fn first_cell_publication_is_stable_under_contention() {
         const THREADS: usize = 8;
         let registry = Arc::new(
             explicit_registry(
                 [partition(1)],
                 ConnectionReuseScope::default(),
                 NonZeroUsize::new(1),

↧ Expand context

-730
+            )
             .unwrap(),
         );
         let partition = explicit_partition(&registry, 1);
         let barrier = Arc::new(Barrier::new(THREADS));
         let mut threads = Vec::new();
-736
         for _ in 0..THREADS {
             let registry = registry.clone();
             let partition = partition.clone();
             let barrier = barrier.clone();
             threads.push(std::thread::spawn(move || {
                 let uri: Uri = "https://example.com/".parse().unwrap();
                 barrier.wait();
                 registry.resolve_cell(&partition, &uri).unwrap()
             }));
-746
+        }
-747
         let cells: Vec<_> = threads
             .into_iter()

@@ -731,755 +795,820 @@

↥ Expand context

             .all(|cell| Arc::ptr_eq(&cells[0], cell)));
         assert_eq!(1, partition.cell_count());
-757
         let waiter = OriginCell::register_waiter(&cells[0], ProtocolRequirement::H1Compatible);
         let lease = OriginCell::take_ready_lease(&cells[0], waiter)
             .expect("admission targeted a different cell than the registry retained");
         drop(lease);
-762
+    }
-763
+}
-764
 #[cfg(all(test, smithy_http_client_loom))]
 mod loom_tests {
     use super::*;
     use std::future::Future;
     use std::pin::Pin;
-770
     #[derive(Debug)]
     struct TestSpawner;
-773
     impl DriverSpawner for TestSpawner {

         fn spawn(&self, driver: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-777
+        }
-778
+    }
-779
     fn explicit_registry(
         partitions: impl IntoIterator<Item = Partition>,
         reuse_scope: ConnectionReuseScope,
         max_connections_per_host: Option<NonZeroUsize>,
     ) -> Result<PartitionRegistry, PartitionRegistryError> {
         let admission_policy =
             max_connections_per_host.map(|limit| AdmissionPolicy::new(limit, true));
         PartitionRegistry::new(
             Some(partitions.into_iter().collect()),
             reuse_scope,
             max_connections_per_host,
             true,
             admission_policy,
             MaintenanceConfig::default(),
-792
+        )
-793
+    }
-794
     #[test]
     fn first_cell_publication_is_stable_under_contention() {
         loom::model(|| {
             let registry = Arc::new(
                 explicit_registry(
                     [Partition::new(PartitionId::from_index(1), TestSpawner)],

↧ Expand context

                     ConnectionReuseScope::default(),
                     None,
-803
+                )
                 .unwrap(),
             );
             let partition = registry.partition(PartitionId::from_index(1)).unwrap();
-807
             let first_registry = registry.clone();
             let first_partition = partition.clone();
             let first = loom::thread::spawn(move || {
                 let uri: Uri = "https://example.com/".parse().unwrap();
                 first_registry.resolve_cell(&first_partition, &uri).unwrap()
             });
             let second_registry = registry.clone();
             let second_partition = partition.clone();
             let second = loom::thread::spawn(move || {
                 let uri: Uri = "https://example.com/".parse().unwrap();
                 second_registry
                     .resolve_cell(&second_partition, &uri)
                     .unwrap()

Pages: 1 2 3 4 5 6 7 8 9 10 11