AWS SDK

rev. 214b1cfffb5029555908aba78574468110a9ad23 (ignoring whitespace)

Files changed:

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

@@ -1,1 +179,187 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Cell-local acquisition order and delivered-result ownership.
 //!
 //! [`AcquisitionQueue`] is the sole owner of every live acquisition attempt
 //! in one origin cell. It combines the bounded FIFO, unlocked delivery
 //! crossings, establishment launch, and terminal result handoff in one state
 //! machine.
 //!
 //! Normal transitions are:
 //!
 //! ```text
 //! bounded register ----------------------------------> Waiting
 //! unbounded register --------------------------------> ReadyToEstablish

 //! Waiting --reserve admission or reuse delivery-----> DeliveryPending
 //! Waiting --local H1 return--------------------------> Ready
 //! DeliveryPending --capacity-------------------------> ReadyToEstablish
 //! DeliveryPending --borrowed H1----------------------> Ready
 //! DeliveryPending --local H1--> DeliveryPending(pending H1)
 //! DeliveryPending(pending H1) --delivery--> Ready + return delivered value
 //! ReadyToEstablish --local H1 return-----------------> Ready + return permit
 //! ReadyToEstablish --poll----------------------------> Launching(Submitted)
 //! Launching(Submitted) --first owner-runtime poll----> Launching(Started)
 //! Launching --local H1 or establishment result------> Ready
 //! Waiting --local or peer H2 activation--------------> Ready
 //! DeliveryPending --H2 activation--------------------> DeliveryPending(pending H2)
 //! ReadyToEstablish / Launching --H2 activation-------> Ready
 //! Ready --poll---------------------------------------> removed; result transferred
 //! ```
 //!
 //! Cancellation has one residence used only while a committed delivery is
 //! crossing the cell lock boundary:
 //!
 //! ```text
 //! Waiting ----------------------> removed; retire or advance demand
 //! DeliveryPending --------------> DeliveryCancelled
 //! DeliveryCancelled --delivery-> removed; return payloads
 //! ReadyToEstablish / Ready -----> removed; return the held event
 //! Launching --------------------> removed; task settles separately
 //! ```
 //!
 //! Admission updates, returned values, and task wakes are detached for the
 //! caller to process after releasing the cell lock.
-47
 use super::super::admission::{DemandId, DemandSnapshot, ProtocolRequirement, SnapshotVersion};
 use super::super::partition::EligibilityGroup;
 use super::{AcquisitionEvent, AcquisitionResult, EstablishmentPermit};
 use std::collections::{BTreeSet, HashMap};
 use std::num::NonZeroUsize;
 use std::task::{Context, Poll, Waker};
-54
 /// Local waiter identity within one cell.
 #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
 pub(in crate::client::pool) struct WaiterId(pub(in crate::client::pool) u64);
-58
 /// Complete acquisition state protected by one origin cell lock.
 ///
 /// At every completed transition:
 ///
 /// - `records` owns every retained acquisition attempt.
 /// - [`WaitingQueueState::Active`] contains exactly the linked
 ///   [`WaiterState::Waiting`] records.
 /// - The active demand identifies and describes the FIFO head.
 /// - `h1_candidates` contains exactly each H1-compatible
 ///   [`WaiterState::DeliveryPending`] without a pending H1,
 ///   [`WaiterState::ReadyToEstablish`], or [`WaiterState::Launching`] attempt.
 /// - `h2_activation_candidates` contains every delivery-pending, ready-to-establish, or
 ///   launching attempt that has not received another result. The waiting FIFO
 ///   head is compared with this set without duplicating its queue residence.
 /// - Delivery-pending, ready-to-establish, launching, delivery-cancelled, and
 ///   ready attempts remain in `records` but never in the bounded FIFO.
 ///
 /// Transitions return admission publications, values requiring fallback, and
 /// wakers to the caller; none runs while this state is locked.
 #[derive(Debug, Default)]
 pub(super) struct AcquisitionQueue {
     /// Every attempt still waiting, launching, or holding a terminal result.
     records: HashMap<WaiterId, WaiterRecord>,
     /// FIFO endpoints and demand for the records currently waiting.
     waiting: WaitingQueueState,
     /// H1-compatible attempts still competing with a returned sender.
     h1_candidates: BTreeSet<WaiterId>,
     /// Attempts outside the FIFO that may accept an H2 activation.
     h2_activation_candidates: BTreeSet<WaiterId>,
     /// Next cell-local waiter identity.
     next_waiter: u64,
     /// Next identity for a head-waiter demand generation.
     next_demand_id: u64,
-92
+}
-93
 /// Owner-runtime progress of one submitted establishment task.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 enum EstablishmentPhase {
     /// The task was submitted but has not begun polling the connector.
     Submitted,
     /// The owner runtime polled the task and the pool owns completion.
     Started,
-101
+}
-102
 /// Whether the cell has an active FIFO and aggregate head demand.
 #[derive(Debug, Default)]
 enum WaitingQueueState {
     /// No waiter is linked and no demand is active.
     #[default]
     Empty,
     /// A nonempty FIFO represented by its endpoints, length, and head demand.
     Active {
         /// Oldest waiting record.
         head: WaiterId,
         /// Youngest waiting record.
         tail: WaiterId,
         /// Number of records linked from `head` through `tail`.
         len: NonZeroUsize,
         /// Demand generation represented by `head`.
         demand: DemandTicket,
     },
-120
+}
-121
 /// One request retained while waiting for or owning an acquisition result.
 #[derive(Debug)]
 struct WaiterRecord {
     /// Protocol requirement published while this waiter is the head.
     requirement: ProtocolRequirement,
     /// Queue residence, delivery state, and any owned result.
     state: WaiterState,
-129
+}
-130
 /// Authoritative cell-local ownership state for one waiter.
 #[derive(Debug)]
 enum WaiterState {
     /// The waiter is linked in the cell-local FIFO.
     Waiting {
         /// Older waiting record, or `None` at the head.
         previous: Option<WaiterId>,
         /// Newer waiting record, or `None` at the tail.
         next: Option<WaiterId>,
         /// Latest task waiting for an acquisition result.
         waker: Option<Waker>,
     },
     /// A committed capacity or reuse delivery reserved this waiter but has
     /// not installed its payload.
     DeliveryPending {
         /// Latest task waiting for the crossing delivery.
         waker: Option<Waker>,
         /// H1 return that won while capacity was crossing.
         pending_h1: Option<AcquisitionResult>,
         /// Local protocol result that won while delivery was crossing.
         pending_result: Option<AcquisitionResult>,
     },
     /// Cancellation won after reservation while the committed delivery was
     /// outside the lock.
     DeliveryCancelled {
         /// Task detached for wake after the crossing payload is returned.
         waker: Option<Waker>,
         /// H1 result returned after the crossing closes, if one arrived first.
         pending_h1: Option<AcquisitionResult>,
         /// Protocol result returned after the crossing closes, if one arrived first.
         pending_result: Option<AcquisitionResult>,
     },
     /// Capacity is ready to start an establishment attempt.
     ReadyToEstablish {
         /// Optional bounded-origin capacity owned until connection install.
         permit: EstablishmentPermit,
     },
     /// Establishment was submitted to the owner runtime and may be unpolled.
     Launching {
         /// Ownership phase used to distinguish submission from first poll.
         phase: EstablishmentPhase,

↧ Expand context

         /// Latest task waiting for the attempt or returned H1 to complete.
         waker: Option<Waker>,
     },
     /// The waiter owns its terminal acquisition result.
     Ready(AcquisitionResult),
-173
+}
-174
 /// Cell-local ownership of the aggregate head demand.
 #[derive(Debug)]
 struct DemandTicket {
     /// Identity of this head-waiter generation.
     id: DemandId,
     /// Version assigned to the next complete publication for this demand
     /// identity.
     version: SnapshotVersion,
     /// Protocol capability required by this head waiter.
     requirement: ProtocolRequirement,
-185
+}
-186
 impl DemandTicket {

@@ -183,191 +245,251 @@

↥ Expand context

             self.id,
             self.version,
             self.requirement,
             eligibility_group.clone(),
-195
+        )
-196
+    }
-197
+}
-198
 impl AcquisitionQueue {
     /// Registers one acquisition attempt.
     ///
     /// An unbounded attempt starts ready to establish. A bounded attempt joins
     /// the capacity FIFO and publishes demand only when it becomes the head.
     pub(super) fn register_waiter(
         &mut self,
         requirement: ProtocolRequirement,
         eligibility_group: &EligibilityGroup,
         bounded: bool,
     ) -> (WaiterId, Option<DemandSnapshot>) {
         let waiter = self.take_waiter_id();

         if !bounded {
             self.records.insert(
                 waiter,
                 WaiterRecord {
                     requirement,
                     state: WaiterState::ReadyToEstablish {
                         permit: EstablishmentPermit::unbounded(),
                     },
                 },
             );
             if requirement == ProtocolRequirement::H1Compatible {
                 self.h1_candidates.insert(waiter);
--
+            }
             self.add_protocol_candidates(waiter, requirement);
             self.assert_consistent();
             return (waiter, None);
-224
+        }
-225
         let previous = match &self.waiting {
             WaitingQueueState::Empty => None,
             WaitingQueueState::Active { tail, .. } => Some(*tail),
         };
         let initial_demand = previous.is_none().then(|| self.new_demand(requirement));
-231

↧ Expand context

         if let Some(previous) = previous {
             let previous = self
                 .records
                 .get_mut(&previous)
                 .expect("waiting tail disappeared");
             let WaiterState::Waiting { next, .. } = &mut previous.state else {
                 unreachable!("waiting tail left the waiting state");
             };
             debug_assert!(next.is_none());
             *next = Some(waiter);
-242
+        }
-243
         let replaced = self.records.insert(
             waiter,
             WaiterRecord {
                 requirement,
                 state: WaiterState::Waiting {
                     previous,
                     next: None,
                     waker: None,

@@ -252,258 +488,616 @@

↥ Expand context

             (waiting @ WaitingQueueState::Empty, Some(demand)) => {
                 let snapshot = demand.snapshot(eligibility_group);
                 *waiting = WaitingQueueState::Active {
                     head: waiter,
                     tail: waiter,
                     len: NonZeroUsize::MIN,
                     demand,
                 };
                 Some(snapshot)
-267
+            }
             (WaitingQueueState::Active { tail, len, .. }, None) => {
                 *tail = waiter;
                 *len = len.checked_add(1).expect("waiter queue length exhausted");
                 None
-272
+            }
             _ => unreachable!("waiter queue occupancy changed during registration"),
         };
         self.assert_consistent();
         (waiter, snapshot)
-277
+    }

-278
     /// Returns the oldest acquisition attempt that can accept HTTP/1.
     ///
     /// A bounded waiter remains a candidate after receiving capacity so a
     /// returned sender can still beat its lazy or pool-owned establishment
     /// attempt. Comparing waiter identities preserves arrival order between
     /// that set and the current capacity-waiting head.
     fn oldest_h1_candidate(&self) -> Option<WaiterId> {
         let waiting = match &self.waiting {
             WaitingQueueState::Active { head, .. }
                 if self.records[head].requirement == ProtocolRequirement::H1Compatible =>
                 if self.records[head].requirement.accepts_h1() =>
-289
+            {
                 Some(*head)
-291
+            }
             WaitingQueueState::Empty | WaitingQueueState::Active { .. } => None,
         };
         match (waiting, self.h1_candidates.first().copied()) {
             (Some(waiting), Some(launching)) => Some(waiting.min(launching)),
             (Some(waiting), None) => Some(waiting),
             (None, launching) => launching,
-298
+        }
-299
+    }
-300
     /// Returns whether a returned H1 can satisfy a live acquisition attempt.
     pub(super) fn can_accept_h1(&self) -> bool {
         self.oldest_h1_candidate().is_some()
-304
+    }
-305
     /// Returns whether a previously admitted local attempt still accepts H1.
     ///
     /// These candidates left the capacity FIFO before its current head and
     /// therefore precede cross-cell reuse for that newer aggregate demand.
     pub(super) fn has_prior_h1_candidate(&self) -> bool {
         !self.h1_candidates.is_empty()
-312
+    }
-313
     /// Returns the oldest attempt that may accept an HTTP/2 activation.
     pub(super) fn oldest_h2_candidate(&self) -> Option<WaiterId> {
         let waiting = match &self.waiting {
             WaitingQueueState::Active { head, .. }
                 if self.records[head].requirement.accepts_h2() =>
-+
+            {
                 Some(*head)
-+
+            }
             WaitingQueueState::Active { .. } => None,
             WaitingQueueState::Empty => None,
         };
         match (waiting, self.h2_activation_candidates.first().copied()) {
             (Some(waiting), Some(active)) => Some(waiting.min(active)),
             (Some(waiting), None) => Some(waiting),
             (None, active) => active,
-+
+        }
-+
+    }
-+
     /// Advances the active demand beyond an accepted route publication.
     ///
     /// Admission retires the published version when it acknowledges route
     /// visibility. The waiter remains queued locally while the route gate
     /// offers activations. Reserving the following version here ensures that
     /// route invalidation can republish the same demand generation without an
     /// older inactive snapshot winning.
     pub(super) fn suppress_published_demand(&mut self, demand: DemandId) -> bool {
         let WaitingQueueState::Active {
             demand: current, ..
         } = &mut self.waiting
         else {
             return false;
         };
         if current.id != demand {
             return false;
-+
+        }
         let acknowledged = current.version.next();
         current.version = acknowledged.next();
         self.assert_consistent();
         true
-+
+    }
-+
     /// Returns the current head demand for publication after route loss.
     pub(super) fn current_demand_snapshot(
         &self,
         eligibility_group: &EligibilityGroup,
     ) -> Option<DemandSnapshot> {
         match &self.waiting {
             WaitingQueueState::Empty => None,
             WaitingQueueState::Active { demand, .. } => Some(demand.snapshot(eligibility_group)),
-+
+        }
-+
+    }
-+
     /// Returns whether `waiter` is still the oldest H2 activation candidate.
     pub(super) fn is_oldest_h2_candidate(&self, waiter: WaiterId) -> bool {
         self.oldest_h2_candidate() == Some(waiter)
-+
+    }
-+
     /// Returns the newest waiter identity committed before a publication.
     pub(super) fn publication_cutoff(&self) -> Option<WaiterId> {
         (!self.records.is_empty()).then(|| {
             WaiterId(
                 self.next_waiter
                     .checked_sub(1)
                     .expect("nonempty waiter set had no allocated identity"),
-+
+            )
         })
-+
+    }
-+
     /// Returns whether an acquisition at or before `cutoff` still needs H2.
     pub(super) fn has_h2_candidate_through(&self, cutoff: WaiterId) -> bool {
         self.oldest_h2_candidate()
             .is_some_and(|waiter| waiter <= cutoff)
-+
+    }
-+
     /// Returns whether any acquisition must precede a new direct H2 arrival.
     pub(super) fn has_h2_candidate(&self) -> bool {
         self.oldest_h2_candidate().is_some()
-+
+    }
-+
     /// Returns whether one transferred generation waiter remains launchable.
     pub(super) fn is_launching_h2_candidate(&self, waiter: WaiterId) -> bool {
         self.h2_activation_candidates.contains(&waiter)
             && self.records.get(&waiter).is_some_and(|record| {
                 record.requirement.accepts_h2()
                     && matches!(record.state, WaiterState::Launching { .. })
             })
-+
+    }
-+
     /// Adds one waiter to each protocol index its requirement accepts.
     fn add_protocol_candidates(&mut self, waiter: WaiterId, requirement: ProtocolRequirement) {
         if requirement.accepts_h1() {
             self.h1_candidates.insert(waiter);
-+
+        }
         if requirement.accepts_h2() {
             self.h2_activation_candidates.insert(waiter);
-+
+        }
-+
+    }
-+
     /// Removes one waiter from both protocol candidate indexes.
     fn remove_protocol_candidates(&mut self, waiter: WaiterId) {
         self.h1_candidates.remove(&waiter);
         self.h2_activation_candidates.remove(&waiter);
-+
+    }
-+
     /// Commits a returned HTTP/1 sender to the oldest compatible attempt.
     ///
     /// Payload construction is delayed until all panic-capable requesting-cell checks
     /// have completed. Once constructed, the result is either state-owned or
     /// returned to the caller for cleanup after the cell lock is released.
     pub(super) fn install_returned_h1(
         &mut self,
         result: impl FnOnce() -> AcquisitionResult,
         eligibility_group: &EligibilityGroup,
     ) -> H1Install {
     ) -> WaiterInstall {
         let Some(waiter) = self.oldest_h1_candidate() else {
             return H1Install::rejected(result());
             return WaiterInstall::rejected(result());
         };
         self.install_protocol_result(waiter, result, eligibility_group)
-+
+    }
-433
     /// Installs a selected protocol result through the shared waiter transition.
     fn install_protocol_result(
         &mut self,
         waiter: WaiterId,
         result: impl FnOnce() -> AcquisitionResult,
         eligibility_group: &EligibilityGroup,
     ) -> WaiterInstall {
         if matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::Waiting { .. })
         ) {
             let removed = self.pop_head(eligibility_group);
             debug_assert_eq!(removed.waiter, waiter);
             let record = self
                 .records
                 .get_mut(&waiter)
                 .expect("HTTP/1 requesting waiter disappeared");
                 .expect("selected requesting waiter disappeared");
             let WaiterState::Waiting { waker, .. } = &mut record.state else {
                 unreachable!("HTTP/1 requesting waiter left the waiting state");
                 unreachable!("selected requesting waiter left the waiting state");
             };
             let waker = waker.take();
             record.state = WaiterState::Ready(result());
             let retired =
                 DemandSnapshot::inactive(removed.demand.id, removed.demand.version.next());
             self.assert_consistent();
             return H1Install {
             return WaiterInstall {
                 waiter: Some(waiter),
                 demand_updates: [Some(retired), removed.successor],
                 returned_event: None,
                 waker,
             };
-465
+        }
-466
         let record = self
             .records
             .get_mut(&waiter)
             .expect("HTTP/1 candidate waiter disappeared");
             .expect("selected candidate waiter disappeared");
         let (returned_event, waker) = match &mut record.state {
             WaiterState::DeliveryPending {
                 waker: _,
                 pending_h1,
                 pending_result,
             } => {
                 debug_assert!(pending_h1.is_none());
                 *pending_h1 = Some(result());
                 debug_assert!(pending_result.is_none());
                 *pending_result = Some(result());
                 (None, None)
-479
+            }
             WaiterState::ReadyToEstablish { .. } | WaiterState::Launching { .. } => {
                 let previous = std::mem::replace(&mut record.state, WaiterState::Ready(result()));
                 match previous {
                     WaiterState::ReadyToEstablish { permit } => {
                         (Some(AcquisitionEvent::Establish(permit)), None)
-485
+                    }
                     WaiterState::Launching { waker, .. } => (None, waker),
                     _ => unreachable!("HTTP/1 candidate changed state under the cell lock"),
                     _ => unreachable!("selected candidate changed state under the cell lock"),
-488
+                }
-489
+            }
             WaiterState::Waiting { .. }
             | WaiterState::DeliveryCancelled { .. }
             | WaiterState::Ready(_) => {
                 unreachable!("HTTP/1 candidate was not eligible for a returned sender")
                 unreachable!("selected waiter was not eligible for a protocol result")
-494
+            }
         };
         self.h1_candidates.remove(&waiter);
         self.remove_protocol_candidates(waiter);
         if returned_event.is_none() {
             // The incoming H1 is state-owned and no permit is detached, so an
             // The result is state-owned and no permit is detached, so an
             // invariant failure cannot drop a pool value under this lock.
             self.assert_consistent();
-501
+        }
         H1Install {
         WaiterInstall {
             waiter: Some(waiter),
             demand_updates: [None, None],
             returned_event,
             waker,
-507
+        }
-508
+    }
-509
     /// Cancels one waiter and detaches all cross-lock cleanup work.
     ///
     /// Removing the head retires its demand and starts a successor generation.
     /// Removing another waiting record leaves the active demand unchanged.
     /// Cancellation during delivery leaves a marker for result installation to
     /// observe. A ready result is returned to the caller for cleanup after unlock.
     pub(super) fn cancel_waiter(
         &mut self,
         waiter: WaiterId,
         eligibility_group: &EligibilityGroup,
     ) -> Option<WaiterCancellation> {
         let state = &self.records.get(&waiter)?.state;
         let cancellation = if matches!(state, WaiterState::Waiting { .. }) {
             let is_head = matches!(
                 self.waiting,
                 WaitingQueueState::Active { head, .. } if head == waiter
             );
             let demand_updates = if is_head {
                 let removed = self.pop_head(eligibility_group);
                 debug_assert_eq!(waiter, removed.waiter);
                 let record = self
                     .records
                     .remove(&waiter)
                     .expect("cancelled head waiter disappeared");
                 debug_assert!(matches!(record.state, WaiterState::Waiting { .. }));
                 let retired =
                     DemandSnapshot::inactive(removed.demand.id, removed.demand.version.next());
                 [Some(retired), removed.successor]
             } else {
                 self.remove_non_head(waiter);
                 [None, None]
             };
             WaiterCancellation {
                 demand_updates,
                 returned_events: [None, None],
-545
+            }
         } else if matches!(state, WaiterState::DeliveryPending { .. }) {
             self.h1_candidates.remove(&waiter);
             self.remove_protocol_candidates(waiter);
             let record = self.records.get_mut(&waiter)?;
             let WaiterState::DeliveryPending { waker, pending_h1 } = &mut record.state else {
             let WaiterState::DeliveryPending {
                 waker,
                 pending_result,
             } = &mut record.state
             else {
                 unreachable!("delivery-pending waiter changed state under the cell lock");
             };
             let waker = waker.take();
             let pending_h1 = pending_h1.take();
             record.state = WaiterState::DeliveryCancelled { waker, pending_h1 };
             let pending_result = pending_result.take();
             record.state = WaiterState::DeliveryCancelled {
                 waker,
                 pending_result,
             };
             WaiterCancellation {
                 demand_updates: [None, None],
                 returned_events: [None, None],
-565
+            }
         } else if matches!(
             state,
             WaiterState::ReadyToEstablish { .. } | WaiterState::Ready(_)
         ) {
             // Validate before detaching the event. Once it is local,
             // this method must return without another panic-capable check.
             self.assert_consistent();
             let record = self.records.remove(&waiter)?;
             self.h1_candidates.remove(&waiter);
             self.remove_protocol_candidates(waiter);
             let event = match record.state {
                 WaiterState::ReadyToEstablish { permit } => AcquisitionEvent::Establish(permit),
                 WaiterState::Ready(result) => AcquisitionEvent::Complete(result),
                 _ => unreachable!("ready waiter changed state under the cell lock"),
             };
             return Some(WaiterCancellation {
                 demand_updates: [None, None],
                 returned_events: [Some(event), None],
             });
         } else if matches!(state, WaiterState::Launching { .. }) {
             self.assert_consistent();
             self.h1_candidates.remove(&waiter);
             self.remove_protocol_candidates(waiter);
             self.records.remove(&waiter)?;
             return Some(WaiterCancellation {
                 demand_updates: [None, None],
                 returned_events: [None, None],
             });
         } else {
             debug_assert!(matches!(state, WaiterState::DeliveryCancelled { .. }));
             return None;
         };
-596

↧ Expand context

         self.assert_consistent();
         Some(cancellation)
-599
+    }
-600
     /// Returns the next event or records the latest waker for a pending waiter.
     ///
     /// # Panics
     ///
     /// Panics if `waiter` is unknown, was cancelled, or was already consumed
     /// by an earlier ready poll.
     pub(super) fn poll_waiter(
         &mut self,
         waiter: WaiterId,
         cx: &mut Context<'_>,
     ) -> Poll<AcquisitionEvent> {
         if matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::ReadyToEstablish { .. })
         ) {
             self.assert_consistent();

@@ -525,653 +831,993 @@

↥ Expand context

             WaiterState::DeliveryCancelled { .. } => {
                 panic!("polled a cancelled acquisition waiter")
-655
+            }
             WaiterState::ReadyToEstablish { .. } | WaiterState::Ready(_) => {
                 unreachable!("ready waiter changed state under the cell lock")
-658
+            }
         };
         if waker
             .as_ref()
             .is_none_or(|waker| !waker.will_wake(cx.waker()))
-663
+        {
             *waker = Some(cx.waker().clone());
-665
+        }
         Poll::Pending
-667
+    }
-668
     /// Transfers completion ownership to the pool immediately before the first poll.
     ///
     /// A returned H1 or cancellation that wins first removes the submitted
     /// attempt's authority, allowing its unpolled future and permit to drop.

     pub(super) fn start_establishment(&mut self, waiter: WaiterId) -> bool {
         let Some(record) = self.records.get_mut(&waiter) else {
             return false;
         };
         match &mut record.state {
             WaiterState::Launching { phase, .. } if *phase == EstablishmentPhase::Submitted => {
                 *phase = EstablishmentPhase::Started;
                 self.assert_consistent();
                 true
-682
+            }
             WaiterState::Ready(AcquisitionResult::H1(_)) => false,
             WaiterState::Ready(_) => false,
             WaiterState::Launching {
                 phase: EstablishmentPhase::Started,
                 ..
             } => {
                 panic!("establishment attempt was started more than once")
-689
+            }
             _ => {
                 panic!("establishment start did not name a launching waiter")
-692
+            }
-693
+        }
-694
+    }
-695
     /// Reserves the current head when a delivery still names its demand
     /// generation.
     ///
     /// The reserved waiter leaves the FIFO and changes in place to
     /// [`WaiterState::DeliveryPending`]. Any remaining head receives a new demand
     /// generation returned for admission acknowledgement.
     pub(super) fn reserve_delivery_waiter(
         &mut self,
         demand: DemandId,
         eligibility_group: &EligibilityGroup,
     ) -> DeliveryReservation {
         let current = matches!(
             &self.waiting,
             WaitingQueueState::Active {
                 demand: current,
                 ..
             } if current.id == demand
         );
         if !current {
             return DeliveryReservation::Rejected;
-716
+        }
-717
         let removed = self.pop_head(eligibility_group);
         debug_assert_eq!(removed.demand.id, demand);
         let record = self
             .records
             .get_mut(&removed.waiter)
             .expect("reserved queue head disappeared");
         let WaiterState::Waiting { waker, .. } = &mut record.state else {
             unreachable!("reserved queue head left the waiting state");
         };
         let waker = waker.take();
         let accepts_h1 = record.requirement == ProtocolRequirement::H1Compatible;
         let requirement = record.requirement;
         record.state = WaiterState::DeliveryPending {
             waker,
             pending_h1: None,
             pending_result: None,
         };
         if accepts_h1 {
             self.h1_candidates.insert(removed.waiter);
--
+        }
         self.add_protocol_candidates(removed.waiter, requirement);
-734
         self.assert_consistent();
         DeliveryReservation::Reserved {
             waiter: removed.waiter,
             successor: removed.successor,
-739
+        }
-740
+    }
-741
     /// Completes the unlocked capacity crossing without panicking while the
     /// incoming permit remains a droppable local.
     ///
     /// A live receiver takes ownership of the permit before another
     /// panic-capable operation. If an H1 return already won, both resources
     /// leave the lock as separate events and the permit is refunnelled.
     pub(super) fn install_capacity(
         &mut self,
         waiter: WaiterId,
         permit: EstablishmentPermit,
     ) -> ResultInstall {
         let Some(record) = self.records.get_mut(&waiter) else {
             return ResultInstall::invalid(
                 AcquisitionEvent::Establish(permit),
                 ResultInstallError::MissingWaiter,
             );
         };
-759
         match &mut record.state {
             WaiterState::DeliveryPending { waker, pending_h1 } => {
             WaiterState::DeliveryPending {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 if let Some(h1) = pending_h1.take() {
                     record.state = WaiterState::Ready(h1);
                     self.h1_candidates.remove(&waiter);
                 if let Some(result) = pending_result.take() {
                     record.state = WaiterState::Ready(result);
                     self.remove_protocol_candidates(waiter);
                     // The rejected permit must cross the lock boundary even
                     // if other state is already inconsistent.
                     return ResultInstall {
                         returned_events: [Some(AcquisitionEvent::Establish(permit)), None],
                         waker,
                         error: None,
                         accepted: false,
                     };
-777
+                }
                 record.state = WaiterState::ReadyToEstablish { permit };
                 // The permit is state-owned before this check can panic.
                 self.assert_consistent();
                 ResultInstall {
                     returned_events: [None, None],
                     waker,
                     error: None,
                     accepted: true,
-786
+                }
-787
+            }
             WaiterState::DeliveryCancelled { waker, pending_h1 } => {
             WaiterState::DeliveryCancelled {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 let pending_h1 = pending_h1.take();
                 let pending_result = pending_result.take();
                 self.records.remove(&waiter);
                 ResultInstall {
                     returned_events: [
                         pending_h1.map(AcquisitionEvent::Complete),
                         pending_result.map(AcquisitionEvent::Complete),
                         Some(AcquisitionEvent::Establish(permit)),
                     ],
                     waker,
                     error: None,
                     accepted: false,
-803
+                }
-804
+            }
             WaiterState::Waiting { .. }
             | WaiterState::ReadyToEstablish { .. }
             | WaiterState::Launching { .. }
             | WaiterState::Ready(_) => ResultInstall::invalid(
                 AcquisitionEvent::Establish(permit),
                 ResultInstallError::UnexpectedState,
             ),
-812
+        }
-813
+    }
-814
     /// Installs a borrowed HTTP/1 result after its requesting waiter was reserved.
     ///
     /// A local return may have won while the borrowed sender crossed its
     /// owning-cell and requesting-cell locks. In that case the local result stays ready and
     /// the borrowed result leaves the lock for ordinary owning-cell return.
     pub(super) fn install_borrowed_h1(
         &mut self,
         waiter: WaiterId,
         result: AcquisitionResult,
     ) -> ResultInstall {
         let Some(record) = self.records.get_mut(&waiter) else {
             return ResultInstall::invalid(
                 AcquisitionEvent::Complete(result),
                 ResultInstallError::MissingWaiter,
             );
         };
-831
         match &mut record.state {
             WaiterState::DeliveryPending { waker, pending_h1 } => {
             WaiterState::DeliveryPending {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 if let Some(local_h1) = pending_h1.take() {
                     record.state = WaiterState::Ready(local_h1);
                     self.h1_candidates.remove(&waiter);
                 if let Some(local_result) = pending_result.take() {
                     record.state = WaiterState::Ready(local_result);
                     self.remove_protocol_candidates(waiter);
                     // The rejected borrowed sender must cross the lock
                     // boundary before any panic-capable consistency check.
                     return ResultInstall {
                         returned_events: [Some(AcquisitionEvent::Complete(result)), None],
                         waker,
                         error: None,
                         accepted: false,
                     };
-849
+                }
                 record.state = WaiterState::Ready(result);
                 self.h1_candidates.remove(&waiter);
                 self.remove_protocol_candidates(waiter);
                 self.assert_consistent();
                 ResultInstall {
                     returned_events: [None, None],
                     waker,
                     error: None,
                     accepted: true,
-858
+                }
-859
+            }
             WaiterState::DeliveryCancelled { waker, pending_h1 } => {
             WaiterState::DeliveryCancelled {
                 waker,
                 pending_result,
             } => {
                 let waker = waker.take();
                 let pending_h1 = pending_h1.take();
                 let pending_result = pending_result.take();
                 self.records.remove(&waiter);
                 ResultInstall {
                     returned_events: [
                         pending_h1.map(AcquisitionEvent::Complete),
                         pending_result.map(AcquisitionEvent::Complete),
                         Some(AcquisitionEvent::Complete(result)),
                     ],
                     waker,
                     error: None,
                     accepted: false,
-875
+                }
-876
+            }
             WaiterState::Waiting { .. }
             | WaiterState::ReadyToEstablish { .. }
             | WaiterState::Launching { .. }
             | WaiterState::Ready(_) => ResultInstall::invalid(
                 AcquisitionEvent::Complete(result),
                 ResultInstallError::UnexpectedState,
             ),
-884
+        }
-885
+    }
-886
     /// Commits one terminal establishment result when its waiter is still live.
     ///
     /// Completion may occur from either launch phase: normal execution starts
     /// the connector first, while dropping an unpolled submitted task installs
     /// a terminal runtime error.
     pub(super) fn install_establishment_result(
         &mut self,
         waiter: WaiterId,
         result: AcquisitionResult,
     ) -> ResultInstall {
         let Some(record) = self.records.get_mut(&waiter) else {
             return ResultInstall::rejected(AcquisitionEvent::Complete(result));
         };
         if matches!(record.state, WaiterState::Ready(AcquisitionResult::H1(_))) {
         if matches!(record.state, WaiterState::Ready(_)) {
             return ResultInstall::rejected(AcquisitionEvent::Complete(result));
-902
+        }
         let WaiterState::Launching { waker, .. } = &mut record.state else {
         if !matches!(record.state, WaiterState::Launching { .. }) {
             return ResultInstall::invalid(
                 AcquisitionEvent::Complete(result),
                 ResultInstallError::UnexpectedState,
             );
-+
+        }
         let WaiterState::Launching { waker, .. } = &mut record.state else {
             unreachable!("validated launching waiter changed state")
         };
         let waker = waker.take();
         record.state = WaiterState::Ready(result);
         self.h1_candidates.remove(&waiter);
         self.remove_protocol_candidates(waiter);
         self.assert_consistent();
         ResultInstall {
             returned_events: [None, None],
             waker,
             error: None,
             accepted: true,
-921
+        }
-922
+    }
-923
     /// Removes a ready waiter and transfers ownership of its result.
     pub(super) fn take_ready_result(&mut self, waiter: WaiterId) -> Option<AcquisitionResult> {
         if !matches!(
             self.records.get(&waiter).map(|record| &record.state),
             Some(WaiterState::Ready(_))
         ) {
             return None;
-931
+        }
-932
         // Validate before detaching the result. Returning it must be the last
         // operation performed while the cell lock is held.
         self.assert_consistent();
         let record = self.records.remove(&waiter)?;
         let WaiterState::Ready(result) = record.state else {
             unreachable!("ready waiter changed state under the cell lock");
         };
         Some(result)
-941
+    }
-942
     /// Offers an H2 activation to the oldest compatible acquisition.
     ///
     /// `cutoff` retains publication priority. A later waiter is left pending
     /// until every live acquisition at or before the cutoff has received an
     /// activation opportunity.
     pub(super) fn install_h2(
         &mut self,
         cutoff: Option<WaiterId>,
         result: impl FnOnce(WaiterId) -> AcquisitionResult,
         eligibility_group: &EligibilityGroup,
     ) -> WaiterInstall {
         let Some(waiter) = self.oldest_h2_candidate() else {
             return WaiterInstall::empty();
         };
         if cutoff.is_some_and(|cutoff| waiter > cutoff) {
             return WaiterInstall::empty();
-+
+        }
-+
         self.install_protocol_result(waiter, || result(waiter), eligibility_group)
-+
+    }
-+
     /// Unlinks the FIFO head and installs any successor demand.
     ///
     /// The head record remains in `records`; the caller either changes it to
     /// `DeliveryPending` or removes it for cancellation.
     fn pop_head(&mut self, eligibility_group: &EligibilityGroup) -> RemovedHead {
         let waiting = std::mem::take(&mut self.waiting);
         let WaitingQueueState::Active {
             head,
             tail,
             len,

↧ Expand context

             demand,
         } = waiting
         else {
             unreachable!("removed a head from an empty waiter queue");
         };
         let next = match &self
             .records
             .get(&head)
             .expect("waiting head disappeared")
             .state
-984
+        {
             WaiterState::Waiting { previous, next, .. } => {
                 debug_assert!(previous.is_none());
                 *next
-988
+            }
             _ => unreachable!("waiting head left the waiting state"),
         };
-991
         let successor = match next {
             Some(next) => {

@@ -928,1090 +989,1156 @@

↥ Expand context

         *len = NonZeroUsize::new(
             len.get()
                 .checked_sub(1)
                 .expect("waiter queue length underflowed"),
-1094
+        )
         .expect("removing a non-head waiter emptied the queue");
         record
-1097
+    }
-1098
     /// Allocates an identity that is never reused within this cell.
     fn take_waiter_id(&mut self) -> WaiterId {
         let value = self.next_waiter;
         self.next_waiter = value.checked_add(1).expect("waiter identity exhausted");
         WaiterId(value)
-1104
+    }
-1105
     /// Starts a demand generation for a waiter that became the FIFO head.
     fn new_demand(&mut self, requirement: ProtocolRequirement) -> DemandTicket {
         let value = self.next_demand_id;
         self.next_demand_id = value.checked_add(1).expect("demand identity exhausted");

         DemandTicket {
             id: DemandId::from_u64(value),
             version: SnapshotVersion::INITIAL,
             requirement,
-1114
+        }
-1115
+    }
-1116
     /// Checks map, FIFO, and demand relationships in debug and test builds.
     pub(super) fn assert_consistent(&self) {
         #[cfg(debug_assertions)]
-+
+        {
             if std::thread::panicking() {
                 return;
-+
+            }
             self.assert_consistent_debug();
-1125
+        }
-+
+    }
-1127
     #[cfg(debug_assertions)]
     fn assert_consistent_debug(&self) {
         let waiting_records = self
             .records
             .values()
             .filter(|record| matches!(record.state, WaiterState::Waiting { .. }))
             .count();
         match &self.waiting {
             WaitingQueueState::Empty => {

↧ Expand context

                 assert_eq!(0, waiting_records, "empty queue retained waiting records");
-1138
+            }
             WaitingQueueState::Active {
                 head,
                 tail,
                 len,
                 demand,
             } => {
                 let head_record = self.records.get(head).expect("waiting head disappeared");
                 assert_eq!(
                     demand.requirement, head_record.requirement,
                     "aggregate demand did not describe the waiting head"
                 );
-1150
                 let mut current = Some(*head);
                 let mut previous = None;
                 let mut traversed = 0;
                 while let Some(waiter) = current {
                     assert!(
                         traversed < self.records.len(),

@@ -1000,1167 +1129,1330 @@

↥ Expand context

                     } = &record.state
                     else {
                         panic!("linked waiter left the waiting state");
                     };
                     assert_eq!(
                         previous, *linked_previous,
                         "waiter queue contains inconsistent backward links"
                     );
                     traversed += 1;
                     previous = Some(waiter);
                     current = *next;
-1178
+                }
-1179
                 assert_eq!(Some(*tail), previous, "waiting tail was not reachable");
                 assert_eq!(
                     len.get(),
                     traversed,
                     "waiter queue length did not match its links"
                 );
                 assert_eq!(

                     waiting_records, traversed,
                     "waiting record was not reachable from the queue head"
                 );
-1190
+            }
-1191
+        }
-1192
         let expected_h1_candidates = self
             .records
             .iter()
             .filter_map(|(waiter, record)| {
                 (record.requirement == ProtocolRequirement::H1Compatible
                 (record.requirement.accepts_h1()
                     && matches!(
                         record.state,
                         WaiterState::DeliveryPending {
                             pending_h1: None,
                             pending_result: None,
                             ..
                         } | WaiterState::ReadyToEstablish { .. }
                             | WaiterState::Launching { .. }
                     ))
                 .then_some(*waiter)
             })
             .collect::<BTreeSet<_>>();
         assert_eq!(
             expected_h1_candidates, self.h1_candidates,
             "HTTP/1 acquisition candidates did not match waiter state"
         );
-+
         let expected_h2_activation_candidates = self
             .records
             .iter()
             .filter_map(|(waiter, record)| {
                 (record.requirement.accepts_h2()
                     && matches!(
                         record.state,
                         WaiterState::DeliveryPending {
                             pending_result: None,
                             ..
                         } | WaiterState::ReadyToEstablish { .. }
                             | WaiterState::Launching { .. }
                     ))
                 .then_some(*waiter)
             })
             .collect::<BTreeSet<_>>();
         assert_eq!(
             expected_h2_activation_candidates, self.h2_activation_candidates,
             "HTTP/2 acquisition candidates did not match waiter state"
         );
-1234
+    }
-1235
     #[cfg(test)]
     pub(super) fn snapshot(&self) -> CellSnapshot {
         let (waiting, demand) = match &self.waiting {
             WaitingQueueState::Empty => (0, None),
             WaitingQueueState::Active { len, demand, .. } => (len.get(), Some(demand.id)),
         };
         CellSnapshot {
             waiting,
             retained: self.records.len(),
             demand,
-1246
+        }
-1247
+    }
-1248
+}
-1249
 /// Result of reserving a requesting waiter for one unlocked delivery.
 pub(super) enum DeliveryReservation {
     /// The current head was reserved and may have a successor demand.
     Reserved {
         waiter: WaiterId,
         successor: Option<DemandSnapshot>,
     },
     /// The delivery no longer matches live cell demand.
     Rejected,
-1259
+}
-1260
 /// Values detached from cell state by cancellation.
 pub(super) struct WaiterCancellation {
     /// Demand retirement and optional successor published after unlocking.
     pub(super) demand_updates: [Option<DemandSnapshot>; 2],
     /// Intermediate and terminal events returned for cleanup after unlocking.
     pub(super) returned_events: [Option<AcquisitionEvent>; 2],
-1267
+}
-1268
 /// Values detached after attempting cell-local HTTP/1 service.
 pub(super) struct H1Install {
 /// Values detached after installing a selected protocol result.
 pub(super) struct WaiterInstall {
     /// Waiter that accepted the result, when one remained eligible.
     pub(super) waiter: Option<WaiterId>,
     /// Demand retirement and optional successor published after unlocking.
     pub(super) demand_updates: [Option<DemandSnapshot>; 2],
     /// Establishment permit or rejected H1 returned after unlocking.
     /// Establishment permit or rejected result returned after unlocking.
     pub(super) returned_event: Option<AcquisitionEvent>,
     /// Waiting task woken after demand publication.
     pub(super) waker: Option<Waker>,
-1279
+}
-1280
 impl H1Install {
 impl WaiterInstall {
     /// Preserves a result when no live attempt can accept HTTP/1.
     fn rejected(result: AcquisitionResult) -> Self {
         Self {
             waiter: None,
             demand_updates: [None, None],
             returned_event: Some(AcquisitionEvent::Complete(result)),
             waker: None,
-1289
+        }
-1290
+    }
-+
     /// Reports that no live attempt can accept an HTTP/2 activation.
     fn empty() -> Self {
         Self {
             waiter: None,
             demand_updates: [None, None],
             returned_event: None,
             waker: None,
-+
+        }
-+
+    }
-1301
+}
-1302
 /// Invalid state observed while a committed event was crossing to the cell.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) enum ResultInstallError {
     /// The waiter reserved for the delivery no longer exists.
     MissingWaiter,
     /// The waiter exists but cannot receive the reserved delivery.
     UnexpectedState,
-1310
+}

↧ Expand context

-1311
 /// Values produced after installing or rejecting a committed event.
 pub(super) struct ResultInstall {
     /// Events rejected by cancellation and returned after unlocking.
     ///
     /// When both are present, the H1 fallback precedes the capacity fallback
     /// so successor demand can reuse the connection before starting another.
     pub(super) returned_events: [Option<AcquisitionEvent>; 2],
     /// Waiting task woken after the delivery fence closes.
     pub(super) waker: Option<Waker>,
     /// Invalid state reported only after the returned result runs its fallback.
     pub(super) error: Option<ResultInstallError>,
     /// Whether cell state became authoritative for the incoming event.
     pub(super) accepted: bool,
-1325
+}
-1326
 impl ResultInstall {
     /// Preserves the incoming result for unlocked cleanup after invalid state.
     fn invalid(event: AcquisitionEvent, error: ResultInstallError) -> Self {
         Self {

@@ -1140,1341 +1199,1414 @@

↥ Expand context

             returned_events: [Some(event), None],
             waker: None,
             error: None,
             accepted: false,
-1345
+        }
-1346
+    }
-1347
+}
-1348
 /// FIFO state detached while advancing the active head.
 struct RemovedHead {
     /// Identity of the unlinked waiter.
     waiter: WaiterId,
     /// Retired demand generation that represented this head.
     demand: DemandTicket,
     /// New demand published if another waiter became the head.
     successor: Option<DemandSnapshot>,
-1357
+}
-1358
 #[cfg(test)]
 #[derive(Debug)]

 pub(super) struct CellSnapshot {
     pub(super) waiting: usize,
     pub(super) retained: usize,
     pub(super) demand: Option<DemandId>,
-1365
+}
-1366
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
-1370
     #[test]
     fn h1_required_waiter_is_not_an_h2_candidate() {
         let mut queue = AcquisitionQueue::default();
         let (waiter, snapshot) = queue.register_waiter(
             ProtocolRequirement::H1Required,
             &EligibilityGroup::Pool,
             false,
         );
-+
         assert!(snapshot.is_none());
         assert_eq!(Some(waiter), queue.oldest_h1_candidate());
         assert_eq!(None, queue.oldest_h2_candidate());
-+
+    }
-+
     #[test]
     fn head_cancellation_uses_the_successors_protocol_requirement() {
         let mut queue = AcquisitionQueue::default();
         let (head, initial) = queue.register_waiter(
             ProtocolRequirement::H2Required,
             &EligibilityGroup::Pool,
             true,
         );
         let (_successor, no_new_demand) = queue.register_waiter(
             ProtocolRequirement::H1Compatible,

↧ Expand context

             &EligibilityGroup::Pool,
             true,
         );
         assert!(initial.is_some());
         assert!(no_new_demand.is_none());
-1400
         let cancelled = queue
             .cancel_waiter(head, &EligibilityGroup::Pool)
             .expect("head waiter was not cancelled");
         assert_eq!(
             Some(DemandSnapshot::active(
                 DemandId::from_u64(1),
                 SnapshotVersion::INITIAL,
                 ProtocolRequirement::H1Compatible,
                 EligibilityGroup::Pool,
             )),
             cancelled.demand_updates[1]
         );
-1413
+    }
-1414

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

@@ -30,30 +448,495 @@

↥ Expand context

 use std::task::{Context, Poll};
-31
 /// Why a connection stopped accepting new work.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 #[non_exhaustive]
 pub enum CloseReason {
     /// The connection exceeded its configured idle timeout.
     IdleTimeout,
     /// The connection was explicitly marked unsafe for reuse.
     Poisoned,
     /// The protocol driver or peer closed the connection.
     ProtocolClosed,
     /// HTTP/1 did not prove a complete reusable message boundary.
     IncompleteH1Exchange,
     /// The transport left HTTP/1 pool ownership through an upgrade.
     Upgraded,
     /// The connection closed to move bounded capacity to another cell.
     Reclaimed,
     /// The connection pool was dropped.
     PoolDropped,

     /// The runtime driving the connection shut down.
     OwnerRuntimeShutdown,
-52
+}
-53
 /// Protocol selected for one installed physical connection.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) enum NegotiatedProtocol {
     /// HTTP/1.1 with one exclusive request sender.
     Http1,
     /// HTTP/2 with a multiplexed request sender.
     #[allow(
         dead_code,
         reason = "constructed once HTTP/2 generations are implemented"
     )]
     Http2,
-61
+}
-62
 /// Immutable identity and transport facts shared by a connection's owners.
 ///
 /// The protocol record, request metadata, tracing, and future lifecycle events
 /// The protocol record, request metadata, tracing, and lifecycle events
 /// all retain this one allocation rather than reconstructing origin or address
 /// data at each transition.
 #[derive(Debug)]
 pub(super) struct ConnectionInfo {
     /// Stable identity assigned by the owning pool.
     id: ConnectionId,
     /// Canonical origin this connection may serve.
     origin: OriginKey,
     /// Partition that retains the transport and protocol driver.
     owner_partition: PartitionId,
     /// Protocol selected by configuration or ALPN.
     #[allow(
         dead_code,
         reason = "read when HTTP/2 generation selection is implemented"
     )]
     protocol: NegotiatedProtocol,
     /// Local socket address reported by the connector, when available.
     local_addr: Option<SocketAddr>,
     /// Remote socket address reported by the connector, when available.
     remote_addr: Option<SocketAddr>,
     /// Whether the transport reaches the origin through a proxy.
     proxied: bool,
     /// Connector metadata copied into every response on this connection.
     connected: Connected,
-86
+}
-87
 impl ConnectionInfo {
     /// Captures immutable facts when protocol establishment succeeds.
     /// Captures immutable facts after transport negotiation selects a protocol.
     pub(super) fn new(
         id: ConnectionId,
         origin: OriginKey,
         owner_partition: PartitionId,
         protocol: NegotiatedProtocol,
         connected: Connected,
     ) -> Arc<Self> {
         let mut extras = Extensions::new();
         connected.get_extras(&mut extras);
         let http_info = extras.get::<HttpInfo>();
         Arc::new(Self {
             id,
             origin,
             owner_partition,
             protocol,
             local_addr: http_info.map(HttpInfo::local_addr),
             remote_addr: http_info.map(HttpInfo::remote_addr),
             proxied: connected.is_proxied(),
             connected,
         })
-110
+    }
-111
     /// Returns the pool-assigned physical connection identity.
     pub(super) fn id(&self) -> ConnectionId {
         self.id
-115
+    }
-116
     /// Returns the canonical origin this connection may serve.
     pub(super) fn origin(&self) -> &OriginKey {
         &self.origin
-120
+    }
-121
     /// Returns the partition that owns this connection's I/O and driver.
     pub(super) fn owner_partition(&self) -> PartitionId {
         self.owner_partition
-125
+    }
-126
     /// Returns the established HTTP protocol.
     #[cfg(test)]
     pub(super) fn protocol(&self) -> NegotiatedProtocol {
         self.protocol
-130
+    }
-131
     /// Returns the connector-reported local socket address.
     #[cfg(test)]
     pub(super) fn local_addr(&self) -> Option<SocketAddr> {
         self.local_addr
-136
+    }
-137
     /// Returns the connector-reported remote socket address.
     #[cfg(test)]
     pub(super) fn remote_addr(&self) -> Option<SocketAddr> {
         self.remote_addr
-142
+    }
-143
     /// Returns whether the origin is reached through a proxy.
     pub(super) fn is_proxied(&self) -> bool {
         self.proxied
-147
+    }
-148
     /// Copies connector-provided values into a response extension map.
     pub(super) fn apply_connector_extras(&self, extensions: &mut Extensions) {
         self.connected.get_extras(extensions);
-152
+    }
-153
     /// Builds Smithy metadata with close authority for this H1 generation.
     /// Builds Smithy metadata with close authority for this H1 record.
     pub(super) fn metadata(&self, close: super::cell::h1::H1CloseHandle) -> ConnectionMetadata {
         let mut builder = ConnectionMetadata::builder()
             .proxied(self.proxied)
             .connection_id(self.id)
             .poison_fn(move || {
                 close.close(CloseReason::Poisoned);
             });
         builder
             .set_local_addr(self.local_addr)
             .set_remote_addr(self.remote_addr);
         builder.build()
-166
+    }
-167
     /// Builds Smithy metadata with close authority for this H2 generation.
     pub(super) fn h2_metadata(&self, close: super::cell::h2::H2CloseHandle) -> ConnectionMetadata {
         let mut builder = ConnectionMetadata::builder()
             .proxied(self.proxied)
             .connection_id(self.id)
             .poison_fn(move || {
                 close.close(CloseReason::Poisoned);
             });
         builder
             .set_local_addr(self.local_addr)
             .set_remote_addr(self.remote_addr);
         builder.build()
-+
+    }
-+
     /// Creates synthetic HTTP/1 information for state-machine tests.
     #[cfg(test)]
     pub(super) fn for_test(id: ConnectionId, owner_partition: PartitionId) -> Arc<Self> {
         Self::new(
             id,
             OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None)
                 .expect("synthetic test origin is valid"),
             owner_partition,
             NegotiatedProtocol::Http1,
             Connected::new(),
-192
+        )
-193
+    }
-194
+}
-195
 /// Shared protocol-independent ownership for one installed connection.
 /// Shared connection ownership from negotiated transport through close.
 pub(super) struct ConnectionState {
     /// Identity and transport facts shared with metadata and lifecycle events.
     info: Arc<ConnectionInfo>,
     /// Dispatch, logical-close, and root-I/O completion state.
     lifecycle: Mutex<LifecycleState>,
-202
+}
-203
 /// Connection lifetime state serialized with dispatch commitment and close.
 #[derive(Debug)]
 struct LifecycleState {
     /// Dispatch eligibility and ownership of bounded capacity.
     logical: LogicalState,
     /// Requests that committed before logical close.
     in_flight: usize,
     /// Whether ownership of root transport I/O has ended.
     physical_complete: bool,
-213
+}
-214
 /// Whether a connection may accept dispatch and still owns bounded capacity.
 #[derive(Debug)]
 enum LogicalState {
     /// The connector returned connected I/O and selected the HTTP protocol,
     /// but Hyper has not produced the request handle required for dispatch.
     PendingOpen,
     /// Dispatch may commit.
     Open {
         /// Bounded-origin slot released by logical close, when configured.
         lease: Option<CapacityLease>,
     },
     /// New dispatch is rejected while accepted work may still drain.
     Closed {
         /// Reason recorded by the first logical-close transition.
         reason: CloseReason,
     },
-231
+}
-232
 impl ConnectionState {
     /// Creates state after transport establishment and protocol selection.
     ///
     /// For TLS transports, the connector has completed TLS and ALPN before this
     /// call. Hyper protocol setup and cell installation have not occurred.
     ///
     /// The returned guard is the unique physical-lifetime owner. After Hyper
     /// returns a request handle, [`Self::open`] attaches optional bounded
     /// capacity before cell installation makes the connection discoverable.
     pub(super) fn pending_open(info: Arc<ConnectionInfo>) -> (Arc<Self>, PhysicalConnectionGuard) {
         let connection = Arc::new(Self {
             info,
             lifecycle: Mutex::new(LifecycleState {
                 logical: LogicalState::PendingOpen,
                 in_flight: 0,
                 physical_complete: false,
             }),
         });
         let physical = PhysicalConnectionGuard {
             connection: connection.clone(),
             active: true,
         };
         (connection, physical)
-+
+    }
-+
     /// Opens dispatch commitment and transfers optional bounded capacity.
     ///
     /// The caller performs this transition after Hyper protocol setup and
     /// before publishing the connection through its cell. Opening first
     /// ensures that a newly visible request handle can commit dispatch.
     ///
     /// Returns `lease` when logical close won before opening.
     pub(super) fn open(&self, lease: Option<CapacityLease>) -> Result<(), Option<CapacityLease>> {
         let mut lifecycle = self.lifecycle.lock();
         if !matches!(lifecycle.logical, LogicalState::PendingOpen) {
             return Err(lease);
-+
+        }
         lifecycle.logical = LogicalState::Open { lease };
         Ok(())
-+
+    }
-+
     /// Creates a connection whose origin has no admission bound.
     ///
     /// The returned guard is the unique physical-lifetime owner and must move
     /// with the root I/O task.
     #[cfg(test)]
     pub(super) fn unbounded(info: Arc<ConnectionInfo>) -> (Arc<Self>, PhysicalConnectionGuard) {
         Self::new(info, None)
         let (connection, physical) = Self::pending_open(info);
         connection
             .open(None)
             .expect("new unbounded connection could not open");
         (connection, physical)
-285
+    }
-286
     /// Creates a connection that takes ownership of one bounded-origin slot.
     ///
     /// The returned guard is the unique physical-lifetime owner and must move
     /// with the root I/O task. Logical close returns `lease` independently of
     /// that guard.
     #[cfg(test)]
     pub(super) fn bounded(
         info: Arc<ConnectionInfo>,
         lease: CapacityLease,
     ) -> (Arc<Self>, PhysicalConnectionGuard) {
         Self::new(info, Some(lease))
--
+    }
--
     /// Builds shared connection state and its unique physical-lifetime guard.
     fn new(
         info: Arc<ConnectionInfo>,
         lease: Option<CapacityLease>,
     ) -> (Arc<Self>, PhysicalConnectionGuard) {
         let connection = Arc::new(Self {
             info,
             lifecycle: Mutex::new(LifecycleState {
                 logical: LogicalState::Open { lease },
                 in_flight: 0,
                 physical_complete: false,
             }),
         });
         let physical = PhysicalConnectionGuard {
             connection: connection.clone(),
             active: true,
         };
         let (connection, physical) = Self::pending_open(info);
         connection
             .open(Some(lease))
             .expect("new bounded connection could not open");
         (connection, physical)
-302
+    }
-303
     /// Returns this connection's stable identity.
     pub(super) fn id(&self) -> ConnectionId {
         self.info.id()
-307
+    }
-308
     /// Returns the partition that owns this connection's I/O and driver.
     pub(super) fn owner_partition(&self) -> PartitionId {
         self.info.owner_partition()
-312
+    }
-313
     /// Returns immutable identity and transport facts for this connection.
     pub(super) fn info(&self) -> &Arc<ConnectionInfo> {
         &self.info
-317
+    }
-318
     /// Attempts to commit one request against logical close.
     ///
     /// Returns a guard and increments the in-flight count while the connection
     /// is open. Returns `None` without changing state after logical close.
     pub(super) fn try_commit_dispatch(connection: &Arc<Self>) -> Option<DispatchGuard> {
         let mut lifecycle = connection.lifecycle.lock();
         if !matches!(lifecycle.logical, LogicalState::Open { .. }) {
             return None;
-327
+        }
         lifecycle.in_flight = lifecycle
             .in_flight
             .checked_add(1)
             .expect("in-flight dispatch count exhausted");
         drop(lifecycle);
-333
         Some(DispatchGuard {
             connection: connection.clone(),
             active: true,
         })
-338
+    }
-339
     /// Performs the first logical-close transition.
     ///
     /// Returns `true` when this call closes the connection and records
     /// `reason`. Returns `false` when another close already won; the original
     /// reason remains unchanged.
     ///
     /// The detached lease is dropped only after the connection lock is
     /// released, so admission and connection locks are never nested.
     pub(super) fn logical_close(&self, reason: CloseReason) -> bool {
         let lease = {
             let mut lifecycle = self.lifecycle.lock();
             let LogicalState::Open { lease } = &mut lifecycle.logical else {
                 return false;
             };
             match &mut lifecycle.logical {
                 LogicalState::PendingOpen => {
                     lifecycle.logical = LogicalState::Closed { reason };
                     None
-+
+                }
                 LogicalState::Open { lease } => {
                     let lease = lease.take();
                     lifecycle.logical = LogicalState::Closed { reason };
                     lease
-+
+                }
                 LogicalState::Closed { .. } => return false,
-+
+            }
         };
         drop(lease);
         tracing::debug!(
             connection_id = %self.id(),
             connection_partition = ?self.owner_partition(),
             protocol = ?self.info.protocol(),
             origin_scheme = %self.info.origin().scheme(),
             origin_host = self.info.origin().host(),
             origin_port = ?self.info.origin().port(),
             close_reason = ?reason,
             "connection logically closed"
         );
         true
-376
+    }
-377
     /// Refines a driver-observed close after Hyper confirms an upgrade.
     ///
     /// Hyper may transfer upgraded I/O and complete its HTTP/1 driver in one
     /// poll. The driver can therefore record [`CloseReason::ProtocolClosed`]
     /// before the request task observes the upgrading response. This method
     /// changes that stored reason to [`CloseReason::Upgraded`].
     ///
     /// Returns `true` only when that refinement was applied. It does not close
     /// the connection again, change I/O ownership, or release capacity a second
     /// time.
     pub(super) fn refine_protocol_close_as_upgrade(&self) -> bool {
         let refined = {
             let mut lifecycle = self.lifecycle.lock();
             match &mut lifecycle.logical {
                 LogicalState::Closed { reason } if *reason == CloseReason::ProtocolClosed => {
                     *reason = CloseReason::Upgraded;
                     true
-395
+                }
                 _ => false,
-397
+            }
         };
         if refined {
             tracing::debug!(
                 connection_id = %self.id(),
                 connection_partition = ?self.owner_partition(),
                 origin_scheme = %self.info.origin().scheme(),
                 origin_host = self.info.origin().host(),
                 origin_port = ?self.info.origin().port(),
                 previous_close_reason = ?CloseReason::ProtocolClosed,
                 close_reason = ?CloseReason::Upgraded,
                 "refined connection close reason after HTTP/1 upgrade"
             );
-410
+        }
         refined
-412
+    }
-413
     /// Verifies a close reason at a protocol handoff boundary in debug builds.
     #[cfg(debug_assertions)]
     pub(super) fn debug_assert_close_reason(&self, expected: CloseReason) {
         let lifecycle = self.lifecycle.lock();
         let actual = match lifecycle.logical {
             LogicalState::PendingOpen => None,
             LogicalState::Open { .. } => None,
             LogicalState::Closed { reason } => Some(reason),
         };
         debug_assert_eq!(Some(expected), actual);
-424
+    }
-425
     /// Removes one dispatch previously committed by [`Self::try_commit_dispatch`].
     fn finish_dispatch(&self) {
         let mut lifecycle = self.lifecycle.lock();
         lifecycle.in_flight = lifecycle
             .in_flight
             .checked_sub(1)
             .expect("completed a dispatch that was not in flight");
-433
+    }
-434
     /// Records that the connection's root I/O is no longer live.
     ///
     /// # Panics
     ///
     /// Panics if physical ownership completes more than once.
     fn finish_physical(&self) {
-441
+        {
             let mut lifecycle = self.lifecycle.lock();
             assert!(
                 !lifecycle.physical_complete,
                 "physical connection ownership completed more than once"
             );
             lifecycle.physical_complete = true;
-448
+        }
         tracing::debug!(
             connection_id = %self.id(),
             connection_partition = ?self.owner_partition(),
             origin_scheme = %self.info.origin().scheme(),
             origin_host = self.info.origin().host(),
             origin_port = ?self.info.origin().port(),
             "connection root I/O ownership ended"
         );
-457
+    }
-458
     /// Returns a consistent lifecycle snapshot.
     #[cfg(test)]
     pub(super) fn snapshot(&self) -> ConnectionSnapshot {
         let lifecycle = self.lifecycle.lock();
         ConnectionSnapshot {
             close_reason: match lifecycle.logical {
                 LogicalState::PendingOpen => None,
                 LogicalState::Open { .. } => None,
                 LogicalState::Closed { reason } => Some(reason),
             },
             in_flight: lifecycle.in_flight,
             physical_complete: lifecycle.physical_complete,
-471
+        }
-472
+    }
-473
+}
-474
 impl fmt::Debug for ConnectionState {

↧ Expand context

     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("ConnectionState")
             .field("info", &self.info)
             .field("lifecycle", &self.lifecycle)
             .finish()
-481
+    }
-482
+}
-483
 #[cfg(test)]
 /// Observable lifecycle state used by protocol coordination and focused tests.
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub(super) struct ConnectionSnapshot {
     /// Reason logical close won, or `None` while dispatch is accepted.
     pub(super) close_reason: Option<CloseReason>,
     /// Number of dispatches that have not completed.
     pub(super) in_flight: usize,
     /// Whether root-I/O ownership has completed.
     pub(super) physical_complete: bool,
-494
+}
-495

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

@@ -1,1 +0,530 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! Request routing and protocol dispatch.
 //!
 //! This module is the request entry point after a [`super::Client`] has
 //! resolved its partition. It rejects unsupported request forms, resolves the
 //! origin cell and partition runtime, and starts partition maintenance before
 //! handing the request to a protocol-specific dispatcher. Connection
 //! acquisition, wire-form preparation, and response ownership live in child
 //! modules.
 //! origin cell and partition runtime, and owns acquisition across protocol
 //! selection. Child modules prepare protocol wire forms and retain response
 //! ownership.
-13
 mod h1;
 mod h2;
-16
 use super::cell::OriginCell;
 use super::establish::TransportTimeout;
 use self::h1::H1DispatchResult;
 use self::h2::H2DispatchResult;
 use super::admission::ProtocolRequirement;
 use super::cell::{AcquisitionEvent, AcquisitionResult, OriginCell, WaiterId};
 use super::establish::{self, TransportTimeout};
 use super::partition::DriverSpawner;
 use super::registry::PartitionState;
 use super::{ConnectionPool, PoolInner};
 use crate::sync::Arc;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use http_1x::{Method, Request, Response, Uri, Version};
 use http_1x::{header, Method, Request, Response, Uri, Version};
 use std::future::poll_fn;
 use std::sync::Arc as StdArc;
-31
 /// Replacement selections allowed after the initial HTTP/2 dispatch attempt.
 const MAX_H2_REACQUISITIONS: usize = 2;
-+
 /// Stable state shared by acquisition and protocol dispatch for one request.
 ///
 /// This context owns ordinary shared configuration and identities. Linear
 /// acquisition values, such as an establishment permit or an exclusive
 /// protocol handle, remain separate arguments so their transfer is explicit.
 #[derive(Clone)]
 pub(super) struct AcquisitionContext {
     /// Shared pool policy and transport construction.
     pub(super) pool: Arc<PoolInner>,
     /// Partition from which the request was issued.
     pub(super) partition: Arc<PartitionState>,
     /// Partition-local state for the request's canonical origin.
     pub(super) cell: Arc<OriginCell>,
     /// Absolute request URI retained across protocol wire-form changes.
     pub(super) absolute_uri: Uri,
     /// Runtime that owns establishment, drivers, and pending return work.
     pub(super) owner_spawner: StdArc<dyn DriverSpawner>,
     /// Transport connection timeout configured for this operation.
     pub(super) connect_timeout: Option<TransportTimeout>,
-54
+}
-55
 impl ConnectionPool {
     /// Resolves common acquisition state and dispatches one request.
     pub(super) async fn send_request(
         &self,
         partition: Arc<PartitionState>,
         request: Request<SdkBody>,
         connect_timeout: Option<TransportTimeout>,
     ) -> Result<Response<SdkBody>, ConnectorError> {
         validate_request_before_acquisition(&request)
             .map_err(|error| ConnectorError::user(error.into()))?;
-66
         let absolute_uri = request.uri().clone();
         let cell = self
             .inner
             .registry
             .resolve_cell(&partition, &absolute_uri)
             .map_err(|error| ConnectorError::user(error.into()))?;
         tracing::trace!(
             request_partition = ?partition.id(),
             origin_scheme = %cell.id().origin().scheme(),
             origin_host = cell.id().origin().host(),
             origin_port = ?cell.id().origin().port(),
             "request resolved to connection-pool cell"
         );
-80
         let owner_spawner = partition
             .owner_spawner()
             .map_err(|error| ConnectorError::user(error.into()))?;
         partition.ensure_maintenance_started(owner_spawner.as_ref());
-85
         let context = AcquisitionContext {
             pool: self.inner.clone(),
             partition,
             cell,
             absolute_uri,
             owner_spawner,
             connect_timeout,
         };
-94
         self.send_h1_request(context, request).await
         self.dispatch_acquired(context, request).await
-+
+    }
-+
     /// Acquires a compatible protocol value and dispatches the request.
     ///
     /// Reacquisition is allowed only while a protocol dispatcher returns the
     /// original request envelope.
     async fn dispatch_acquired(
         &self,
         context: AcquisitionContext,
         mut request: Request<SdkBody>,
     ) -> Result<Response<SdkBody>, ConnectorError> {
         let requirement = protocol_requirement(&request);
         let mut h2_reacquisitions = H2ReacquisitionBudget::default();
-+
         loop {
             match Self::acquire(&context, requirement).await? {
                 AcquisitionResult::H1(selection) => {
                     match self.dispatch_h1(&context, request, selection).await? {
                         H1DispatchResult::Response(response) => return Ok(response),
                         H1DispatchResult::Reacquire(returned) => request = returned,
-+
+                    }
-+
+                }
                 AcquisitionResult::H2(activation) => {
                     match self.dispatch_h2(&context, request, activation).await? {
                         H2DispatchResult::Response(response) => return Ok(response),
                         H2DispatchResult::Reacquire(reacquisition) => {
                             let (returned, error) = reacquisition.into_parts();
                             if !h2_reacquisitions.admit_replacement() {
                                 return Err(error);
-+
+                            }
                             request = returned;
-+
+                        }
-+
+                    }
-+
+                }
                 AcquisitionResult::Failed(_) => {
                     unreachable!("failed acquisition returned as a successful result")
-+
+                }
                 AcquisitionResult::Reacquire => {
                     unreachable!("internal reacquisition escaped the acquisition loop")
-+
+                }
-+
+            }
-+
+        }
-+
+    }
-+
     /// Acquires one protocol value compatible with the request.
     ///
     /// Local reusable values complete immediately. Otherwise one waiter
     /// remains registered while it receives a reusable protocol value, an
     /// establishment failure, or authority to start establishment.
     async fn acquire(
         context: &AcquisitionContext,
         requirement: ProtocolRequirement,
     ) -> Result<AcquisitionResult, ConnectorError> {
         'acquire: loop {
             if requirement.accepts_h2() {
                 if let Some(activation) = OriginCell::select_h2(&context.cell) {
                     tracing::trace!(
                         connection_id = %activation.connection().id(),
                         request_partition = ?context.partition.id(),
                         connection_partition = ?activation.connection().owner_partition(),
                         origin_scheme = %context.cell.id().origin().scheme(),
                         origin_host = context.cell.id().origin().host(),
                         origin_port = ?context.cell.id().origin().port(),
                         "HTTP/2 pool hit; activating local generation"
                     );
                     return Ok(AcquisitionResult::H2(activation));
-+
+                }
-+
+            }
             if requirement.accepts_h1() {
                 if let Some(selection) = OriginCell::select_h1(&context.cell) {
                     tracing::trace!(
                         connection_id = %selection.connection_id(),
                         request_partition = ?context.partition.id(),
                         connection_partition = ?selection.connection().owner_partition(),
                         origin_scheme = %selection.connection().info().origin().scheme(),
                         origin_host = selection.connection().info().origin().host(),
                         origin_port = ?selection.connection().info().origin().port(),
                         "HTTP/1 pool hit; reusing idle connection"
                     );
                     return Ok(AcquisitionResult::H1(selection));
-+
+                }
-+
+            }
-+
             let waiter = OriginCell::register_waiter(&context.cell, requirement);
             tracing::trace!(
                 request_partition = ?context.partition.id(),
                 origin_scheme = %context.cell.id().origin().scheme(),
                 origin_host = context.cell.id().origin().host(),
                 origin_port = ?context.cell.id().origin().port(),
                 protocol_requirement = ?requirement,
                 "connection acquisition queued"
             );
             let mut waiter_guard = WaiterGuard::new(context.cell.clone(), waiter);
             loop {
                 match poll_fn(|cx| context.cell.poll_waiter(waiter, cx)).await {
                     AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => {
                         waiter_guard.disarm();
                         return Ok(AcquisitionResult::H1(selection));
-+
+                    }
                     AcquisitionEvent::Complete(AcquisitionResult::H2(activation)) => {
                         waiter_guard.disarm();
                         OriginCell::service_h2_waiters(&context.cell);
                         OriginCell::service_peer_h2_waiters(&context.cell);
                         tracing::trace!(
                             connection_id = %activation.connection().id(),
                             request_partition = ?context.partition.id(),
                             connection_partition = ?activation.connection().owner_partition(),
                             origin_scheme = %context.cell.id().origin().scheme(),
                             origin_host = context.cell.id().origin().host(),
                             origin_port = ?context.cell.id().origin().port(),
                             "HTTP/2 acquisition completed"
                         );
                         return Ok(AcquisitionResult::H2(activation));
-+
+                    }
                     AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
                         waiter_guard.disarm();
                         return Err(error);
-+
+                    }
                     AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
                         waiter_guard.disarm();
                         continue 'acquire;
-+
+                    }
                     AcquisitionEvent::Establish(permit) => {
                         tracing::trace!(
                             request_partition = ?context.partition.id(),
                             origin_scheme = %context.cell.id().origin().scheme(),
                             origin_host = context.cell.id().origin().host(),
                             origin_port = ?context.cell.id().origin().port(),
                             "connection establishment starting"
                         );
                         let attempt =
                             establish::establish(context.clone(), waiter, permit, requirement);
                         let completion =
                             EstablishmentCompletionGuard::new(context.cell.clone(), waiter);
                         context.owner_spawner.spawn(Box::pin(async move {
                             let mut completion = completion;
                             if !completion.start() {
                                 drop(attempt);
                                 completion.disarm();
                                 return;
-+
+                            }
                             match attempt.await {
                                 establish::EstablishmentOutcome::Complete(result) => {
                                     completion.complete(result);
-+
+                                }
                                 establish::EstablishmentOutcome::Transferred => completion.disarm(),
-+
+                            }
                         }));
-+
+                    }
-+
+                }
-+
+            }
-+
+        }
-+
+    }
-+
+}
-+
 /// Per-request bound on replacement HTTP/2 selections.
 #[derive(Default)]
 struct H2ReacquisitionBudget {
     /// Replacement selections admitted after the initial selection.
     completed: usize,
-+
+}
-+
 impl H2ReacquisitionBudget {
     /// Returns whether one more replacement selection may proceed.
     fn admit_replacement(&mut self) -> bool {
         if self.completed >= MAX_H2_REACQUISITIONS {
             return false;
-+
+        }
         self.completed += 1;
         true
-+
+    }
-+
+}
-+
 /// Guarantees a terminal result for one submitted establishment attempt.
 ///
 /// The guard enters the owner-runtime future before submission. Runtime task
 /// drop reports failure to the waiter. A result that wins before the submitted
 /// future starts disarms the guard without polling the connector.
 struct EstablishmentCompletionGuard {
     cell: Arc<OriginCell>,
     waiter: WaiterId,
     active: bool,
-+
+}
-+
 impl EstablishmentCompletionGuard {
     fn new(cell: Arc<OriginCell>, waiter: WaiterId) -> Self {
         Self {
             cell,
             waiter,
             active: true,
-+
+        }
-+
+    }
-+
     fn start(&self) -> bool {
         self.cell.start_establishment(self.waiter)
-+
+    }
-+
     fn complete(mut self, result: AcquisitionResult) {
         self.active = false;
         self.cell.complete_establishment(self.waiter, result);
-+
+    }
-+
     fn disarm(&mut self) {
         self.active = false;
-+
+    }
-+
+}
-+
 impl Drop for EstablishmentCompletionGuard {
     fn drop(&mut self) {
         if self.active {
             let error = ConnectorError::io(EstablishmentTaskDropped.into());
             self.cell
                 .complete_establishment(self.waiter, AcquisitionResult::Failed(error));
             tracing::debug!(
                 request_partition = ?self.cell.id().partition(),
                 connection_partition = ?self.cell.id().partition(),
                 origin_scheme = %self.cell.id().origin().scheme(),
                 origin_host = self.cell.id().origin().host(),
                 origin_port = ?self.cell.id().origin().port(),
                 "connection establishment task dropped"
             );
-+
+        }
-+
+    }
-+
+}
-+
 /// The owner runtime discarded an establishment future before completion.
 #[derive(Debug)]
 struct EstablishmentTaskDropped;
-+
 impl std::fmt::Display for EstablishmentTaskDropped {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("the owner-runtime connection establishment task was dropped")
-+
+    }
-+
+}
-+
 impl std::error::Error for EstablishmentTaskDropped {}
-+
 /// Cancels a request's waiter until it consumes a terminal acquisition result.
 struct WaiterGuard {
     cell: Arc<OriginCell>,
     waiter: WaiterId,
     active: bool,
-+
+}
-+
 impl WaiterGuard {
     fn new(cell: Arc<OriginCell>, waiter: WaiterId) -> Self {
         Self {
             cell,
             waiter,
             active: true,
-+
+        }
-+
+    }
-+
     fn disarm(&mut self) {
         self.active = false;
-+
+    }
-+
+}
-+
 impl Drop for WaiterGuard {
     fn drop(&mut self) {
         if self.active {
             OriginCell::cancel_waiter(&self.cell, self.waiter);
-+
+        }
-359
+    }
-360
+}
-361
 /// Rejects request forms the pool cannot route before acquiring connection state.
 fn validate_request_before_acquisition(
     request: &Request<SdkBody>,
 ) -> Result<(), RequestDispatchError> {
     if request.version() == Version::HTTP_2 && request.method() == Method::CONNECT {
         return Err(RequestDispatchError::ExtendedConnectUnsupported);
-+
+    }
     if request.version() == Version::HTTP_2 && has_upgrade_semantics(request) {
         return Err(RequestDispatchError::UpgradeOverHttp2);
-+
+    }
     match request.version() {
         Version::HTTP_11 | Version::HTTP_2 => Ok(()),
         Version::HTTP_10 if request.method() == Method::CONNECT => {
             Err(RequestDispatchError::ConnectOverHttp10)
-376
+        }
         Version::HTTP_10 => Ok(()),
         version => Err(RequestDispatchError::UnsupportedVersion(version)),
-379
+    }
-380
+}
-381
 /// Returns the protocol capability needed to preserve the request's wire semantics.
 fn protocol_requirement(request: &Request<SdkBody>) -> ProtocolRequirement {
     if request.version() == Version::HTTP_2 {
         ProtocolRequirement::H2Required
     } else if request.version() == Version::HTTP_10
         || request.method() == Method::CONNECT
         || has_upgrade_semantics(request)
-+
+    {
         ProtocolRequirement::H1Required
     } else {
         ProtocolRequirement::H1Compatible
-+
+    }
-+
+}
-+
 /// Returns whether the request asks HTTP/1 to switch protocols.
 fn has_upgrade_semantics(request: &Request<SdkBody>) -> bool {
     request.headers().contains_key(header::UPGRADE)
         || request
             .headers()
             .get_all(header::CONNECTION)
             .iter()
             .any(|value| {
                 value.to_str().is_ok_and(|value| {
                     value
                         .split(',')
                         .any(|token| token.trim().eq_ignore_ascii_case("upgrade"))
                 })
             })
-+
+}
-+
 /// Marker for a `Host` header synthesized during an HTTP/1 dispatch attempt.
 #[derive(Clone, Copy, Debug)]
 struct H1HostHeaderInserted;
-+
 /// A request rejected before Hyper accepts it for dispatch.
 #[derive(Debug)]
 enum RequestDispatchError {
     /// The absolute URI did not contain the authority required for `Host`.
     MissingAuthority,
     /// The derived `Host` value was not a valid HTTP header value.
     InvalidHostHeader(http_1x::header::InvalidHeaderValue),
     /// HTTP/1.0 cannot represent a CONNECT request accepted by this client.
     ConnectOverHttp10,
     /// The request selected an HTTP version this client cannot dispatch.
     UnsupportedVersion(Version),
     /// Extended CONNECT requires HTTP/2 stream-lifecycle support.
     ExtendedConnectUnsupported,
     /// HTTP/1 upgrade semantics cannot be represented by this HTTP/2 path.
     UpgradeOverHttp2,
-431
+}
-432
 impl std::fmt::Display for RequestDispatchError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
             Self::MissingAuthority => f.write_str("request URI must contain an authority"),
             Self::InvalidHostHeader(_) => {
                 f.write_str("request URI authority is not a valid Host header")
-439
+            }
             Self::ConnectOverHttp10 => f.write_str("CONNECT is not supported over HTTP/1.0"),
             Self::UnsupportedVersion(version) => {
                 write!(f, "unsupported HTTP version: {version:?}")
-443
+            }
             Self::ExtendedConnectUnsupported => {
                 f.write_str("extended CONNECT over HTTP/2 is not supported")
-+
+            }
             Self::UpgradeOverHttp2 => {
                 f.write_str("HTTP/1 protocol upgrade semantics cannot use HTTP/2")
-+
+            }
-450
+        }
-451
+    }
-452
+}
-453
 impl std::error::Error for RequestDispatchError {
     fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
         match self {
             Self::InvalidHostHeader(error) => Some(error),
             _ => None,
-459
+        }
-460
+    }
-461
+}
-+
 #[cfg(all(test, not(smithy_http_client_loom)))]
 mod tests {
     use super::*;
-+
     fn request(method: Method, version: Version) -> Request<SdkBody> {
         Request::builder()
             .method(method)
             .uri("https://example.com/resource")
             .version(version)
             .body(SdkBody::empty())
             .unwrap()
-+
+    }
-+
     #[test]
     fn request_semantics_select_protocol_capability() {
         assert_eq!(
             ProtocolRequirement::H1Compatible,
             protocol_requirement(&request(Method::GET, Version::HTTP_11))
         );
         assert_eq!(
             ProtocolRequirement::H2Required,
             protocol_requirement(&request(Method::GET, Version::HTTP_2))
         );
         assert_eq!(
             ProtocolRequirement::H1Required,
             protocol_requirement(&request(Method::GET, Version::HTTP_10))
         );
         assert_eq!(
             ProtocolRequirement::H1Required,
             protocol_requirement(&request(Method::CONNECT, Version::HTTP_11))
         );
-+
         let mut upgrade = request(Method::GET, Version::HTTP_11);
         upgrade
             .headers_mut()
             .insert(header::CONNECTION, "keep-alive, Upgrade".parse().unwrap());
         assert_eq!(
             ProtocolRequirement::H1Required,
             protocol_requirement(&upgrade)
         );
-+
+    }
-+
     #[test]
     fn unsupported_h2_tunnel_and_upgrade_forms_are_rejected() {
         assert!(matches!(
             validate_request_before_acquisition(&request(Method::CONNECT, Version::HTTP_2)),
             Err(RequestDispatchError::ExtendedConnectUnsupported)
         ));
-+
         let mut upgrade = request(Method::GET, Version::HTTP_2);
         upgrade
             .headers_mut()
             .insert(header::UPGRADE, "websocket".parse().unwrap());
         assert!(matches!(
             validate_request_before_acquisition(&upgrade),
             Err(RequestDispatchError::UpgradeOverHttp2)
         ));
-+
+    }
-+
     #[test]
     fn h2_reacquisition_is_bounded_after_two_replacements() {
         let mut budget = H2ReacquisitionBudget::default();
         for _ in 0..MAX_H2_REACQUISITIONS {
             assert!(budget.admit_replacement());
-+
+        }
         assert!(!budget.admit_replacement());
-+
+    }
-+
+}

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

@@ -1,1 +374,177 @@

 /*
  * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
  * SPDX-License-Identifier: Apache-2.0
  */
-5
 //! HTTP/1 acquisition, exclusive dispatch, and response completion.
 //! HTTP/1 wire preparation, exclusive dispatch, and response completion.
 //!
 //! The parent dispatcher has already validated the request, resolved its
 //! origin cell, and selected the partition runtime. This module prepares the
 //! HTTP/1 wire form, acquires Hyper's exclusive request sender, and owns that
 //! origin cell, acquired an HTTP/1 selection, and selected the partition
 //! runtime. This module prepares the HTTP/1 wire form and owns the exclusive
 //! sender until the response reaches a reusable message boundary. Every
 //! cancellation path either returns the sender to its connection-owning cell
 //! or retires the connection.
 //! cancellation path returns the sender to its connection cell or retires the
 //! connection.
-14
 use super::super::admission::ProtocolRequirement;
 use super::super::cell::h1::{H1Exchange, H1Selection};
 use super::super::cell::{AcquisitionEvent, AcquisitionResult, OriginCell, WaiterId};
 use super::super::connection::{CloseReason, ConnectionState, DispatchGuard};
 use super::super::establish;
 use super::super::partition::DriverSpawner;
 use super::super::ConnectionPool;
 use super::{AcquisitionContext, RequestDispatchError};
 use crate::sync::Arc;
 use super::{AcquisitionContext, H1HostHeaderInserted, RequestDispatchError};
 use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
 use http_1x::{Method, Request, Response, Uri, Version};
 use hyper::body::Body;
 use std::future::poll_fn;
 use std::pin::Pin;
 use std::sync::Arc as StdArc;
 use std::task::{Context, Poll};
-29
 /// Result of one HTTP/1 dispatch attempt while the pool owns the request.
 pub(super) enum H1DispatchResult {
     /// Hyper accepted the request and produced a guarded response.
     Response(Response<SdkBody>),
     /// The selection became stale or Hyper returned the request unsent.
     Reacquire(Request<SdkBody>),
-+
+}
-+
 impl ConnectionPool {
     /// Acquires an HTTP/1 connection and sends one request.
     /// Attempts one dispatch through an acquired HTTP/1 selection.
     ///
     /// The reacquisition loop runs only while the pool still owns the request. A
     /// selection that closes before dispatch can be replaced immediately.
     /// Hyper may also return an unconsumed request when a reused connection
     /// rejects it before writing. Once Hyper accepts a request on a fresh
     /// connection, or begins an exchange on any connection, this function does
     /// not reacquire.
     pub(super) async fn send_h1_request(
     /// The parent may reacquire only when this function returns the original
     /// request. A fresh connection's send error is terminal even if Hyper
     /// returns the request.
     pub(super) async fn dispatch_h1(
         &self,
         context: AcquisitionContext,
         context: &AcquisitionContext,
         mut request: Request<SdkBody>,
     ) -> Result<Response<SdkBody>, ConnectorError> {
         mut selection: H1Selection,
     ) -> Result<H1DispatchResult, ConnectorError> {
         let request_method = request.method().clone();
         // Add the HTTP/1 header that depends on the absolute request URI.
         add_host_header(&mut request, &context.absolute_uri)
             .map_err(|error| ConnectorError::user(error.into()))?;
--
         // Redispatch is valid only when the original request is recovered intact.
         loop {
             let mut selection = Self::acquire_h1(&context).await?;
         let reused = selection.is_reused();
         let connection = selection.connection().clone();
         let close_handle = selection.close_handle();
         let captured_metadata = request
             .extensions()
             .get::<CaptureSmithyConnection>()
             .cloned()
             .map(|capture| {
                 let metadata = connection.info().metadata(close_handle.clone());
                 let captured = metadata.clone();
                 capture.set_connection_retriever(move || Some(captured.clone()));
                 metadata
             });
-64
         if request.version() == Version::HTTP_2 {
             let metadata =
                 captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
             return Err(ConnectorError::user(
                 "an HTTP/2 request cannot use an HTTP/1 connection".into(),
-70
+            )
             .with_connection(metadata));
-72
+        }
-73
         add_host_header(&mut request, &context.absolute_uri)
             .map_err(|error| ConnectorError::user(error.into()))?;
         rewrite_h1_request_target(&mut request, connection.info().is_proxied());
-77
         // Commit against logical close before transferring the request to
         // Hyper. A close that wins this race leaves the request untouched.
         let Some(dispatch) = ConnectionState::try_commit_dispatch(&connection) else {
             tracing::trace!(
                 connection_id = %connection.id(),
                 request_partition = ?context.partition.id(),
                 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 selection became stale before dispatch; reacquiring"
                 "HTTP/1 selection became stale before dispatch"
             );
             *request.uri_mut() = context.absolute_uri.clone();
             selection.retire_connection(CloseReason::ProtocolClosed);
                 continue;
             return Ok(H1DispatchResult::Reacquire(request));
         };
         let send = selection.sender_mut().hyper_mut().try_send_request(request);
-95
         let exchange = selection.into_exchange();
         match send.await {
             Ok(mut response) => {
                 // Response-body ownership keeps both the accepted dispatch
                 // and exclusive request handle out of the pool until Hyper
                 // proves a complete message boundary.
                 connection
                     .info()
                     .apply_connector_extras(response.extensions_mut());
                     return Ok(guard_response(
                 Ok(H1DispatchResult::Response(guard_response(
                     response,
                     request_method,
                     exchange,
                     dispatch,
                     context.owner_spawner.clone(),
                     ));
                 )))
-112
+            }
             Err(mut error) => {
                 if let Some(mut returned) = error.take_message() {
                     exchange.retire_connection(CloseReason::ProtocolClosed);
                     drop(dispatch);
                     if reused {
                         tracing::trace!(
                             connection_id = %connection.id(),
                             request_partition = ?context.partition.id(),
                             connection_partition = ?connection.owner_partition(),
                             origin_scheme = %connection.info().origin().scheme(),
                             origin_host = connection.info().origin().host(),
                             origin_port = ?connection.info().origin().port(),
                                 "reused HTTP/1 connection rejected request; reacquiring"
                             "reused HTTP/1 connection returned request unsent"
                         );
                         *returned.uri_mut() = context.absolute_uri.clone();
                             // Hyper returned the original request without
                             // consuming its body, so reacquisition is authoritative.
                             request = returned;
                             continue;
                         return Ok(H1DispatchResult::Reacquire(returned));
-129
+                    }
                         let metadata = captured_metadata
                             .unwrap_or_else(|| connection.info().metadata(close_handle));
                         return Err(super::super::super::downcast_error(Box::new(
                             error.into_error(),
                         ))
                         .with_connection(metadata));
--
+                    }
                     exchange.retire_connection(CloseReason::IncompleteH1Exchange);
                     drop(dispatch);
                     let metadata = captured_metadata
                         .unwrap_or_else(|| connection.info().metadata(close_handle));
                     return Err(
                         super::super::super::downcast_error(Box::new(error.into_error()))
                             .with_connection(metadata),
                     );
-136
+                }
                 exchange.retire_connection(CloseReason::IncompleteH1Exchange);
                 drop(dispatch);
                 let metadata =
                     captured_metadata.unwrap_or_else(|| connection.info().metadata(close_handle));
                 Err(
                     super::super::super::downcast_error(Box::new(error.into_error()))
                         .with_connection(metadata),
-+
+                )
-145
+            }
-146
+        }
-147
+    }
--
     /// Returns one exclusive HTTP/1 request handle for this request.
     ///
     /// A local idle handle completes immediately. Otherwise one cell-local
     /// waiter remains registered while it receives either a reusable handle,
     /// an establishment failure, or capacity to start establishment. Starting
     /// establishment does not complete the waiter: a returning connection may
     /// still satisfy it first.
     async fn acquire_h1(context: &AcquisitionContext) -> Result<H1Selection, ConnectorError> {
         if let Some(selection) = OriginCell::select_h1(&context.cell) {
             tracing::trace!(
                 connection_id = %selection.connection_id(),
                 request_partition = ?context.partition.id(),
                 connection_partition = ?selection.connection().owner_partition(),
                 origin_scheme = %selection.connection().info().origin().scheme(),
                 origin_host = selection.connection().info().origin().host(),
                 origin_port = ?selection.connection().info().origin().port(),
                 "HTTP/1 pool hit; reusing idle connection"
             );
             return Ok(selection);
--
+        }
--
         let waiter = context
             .cell
             .register_waiter(ProtocolRequirement::H1Compatible);
         tracing::trace!(
             request_partition = ?context.partition.id(),
             origin_scheme = %context.cell.id().origin().scheme(),
             origin_host = context.cell.id().origin().host(),
             origin_port = ?context.cell.id().origin().port(),
             "HTTP/1 pool miss; acquisition queued"
         );
         let mut waiter_guard = WaiterGuard::new(context.cell.clone(), waiter);
         loop {
             match poll_fn(|cx| context.cell.poll_waiter(waiter, cx)).await {
                 AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => {
                     tracing::trace!(
                         connection_id = %selection.connection_id(),
                         request_partition = ?context.partition.id(),
                         connection_partition = ?selection.connection().owner_partition(),
                         origin_scheme = %selection.connection().info().origin().scheme(),
                         origin_host = selection.connection().info().origin().host(),
                         origin_port = ?selection.connection().info().origin().port(),
                         "HTTP/1 waiter acquired a reusable connection"
                     );
                     waiter_guard.disarm();
                     return Ok(selection);
--
+                }
                 AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
                     waiter_guard.disarm();
                     return Err(error);
--
+                }
                 AcquisitionEvent::Establish(permit) => {
                     tracing::trace!(
                         request_partition = ?context.partition.id(),
                         origin_scheme = %context.cell.id().origin().scheme(),
                         origin_host = context.cell.id().origin().host(),
                         origin_port = ?context.cell.id().origin().port(),
                         "HTTP/1 waiter starting connection establishment"
                     );
                     let attempt = establish::establish_h1(context.clone(), permit);
                     let completion = EstablishmentCompletion::new(context.cell.clone(), waiter);
                     context.owner_spawner.spawn(Box::pin(async move {
                         let mut completion = completion;
                         if !completion.start() {
                             drop(attempt);
                             completion.disarm();
                             return;
--
+                        }
                         let result = attempt
                             .await
                             .map(AcquisitionResult::H1)
                             .unwrap_or_else(AcquisitionResult::Failed);
                         completion.complete(result);
                     }));
--
+                }
--
+            }
--
+        }
--
+    }
--
+}
--
 /// Guarantees a terminal result for one submitted establishment attempt.
 ///
 /// This guard moves into the owner-runtime future before it is submitted. If
 /// the runtime discards that future, [`Drop`] reports a failure to the waiter.
 /// If another connection satisfies the waiter before the submitted future
 /// starts, [`Self::start`] rejects the attempt and the guard is disarmed.
 struct EstablishmentCompletion {
     /// Cell retaining the waiter and establishment state.
     cell: Arc<OriginCell>,
     /// Waiter that receives the attempt's terminal result.
     waiter: WaiterId,
     /// Whether task drop still owes a failure result.
     active: bool,
--
+}
--
 impl EstablishmentCompletion {
     /// Arms task-drop completion before the future is submitted.
     fn new(cell: Arc<OriginCell>, waiter: WaiterId) -> Self {
         Self {
             cell,
             waiter,
             active: true,
--
+        }
--
+    }
--
     /// Marks the waiter started immediately before the connector is first polled.
     fn start(&self) -> bool {
         self.cell.start_establishment(self.waiter)
--
+    }
--
     /// Delivers the attempt result and disarms task-drop completion.
     fn complete(mut self, result: AcquisitionResult) {
         self.active = false;
         self.cell.complete_establishment(self.waiter, result);
--
+    }
--
     /// Disarms after another connection has already completed the waiter.
     fn disarm(&mut self) {
         self.active = false;
--
+    }
--
+}
--
 impl Drop for EstablishmentCompletion {
     fn drop(&mut self) {
         if self.active {
             let error = ConnectorError::other(EstablishmentTaskDropped.into(), None);
             self.cell
                 .complete_establishment(self.waiter, AcquisitionResult::Failed(error));
             tracing::debug!(
                 request_partition = ?self.cell.id().partition(),
                 connection_partition = ?self.cell.id().partition(),
                 origin_scheme = %self.cell.id().origin().scheme(),
                 origin_host = self.cell.id().origin().host(),
                 origin_port = ?self.cell.id().origin().port(),
                 "HTTP/1 connection establishment task dropped"
             );
--
+        }
--
+    }
--
+}
--
 /// The owner runtime discarded an establishment future before completion.
 #[derive(Debug)]
 struct EstablishmentTaskDropped;
--
 impl std::fmt::Display for EstablishmentTaskDropped {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         f.write_str("the owner-runtime connection establishment task was dropped")
--
+    }
--
+}
--
 impl std::error::Error for EstablishmentTaskDropped {}
--
 /// Cancels a request's retained waiter if acquisition does not complete.
 ///
 /// The request future owns this guard from waiter registration until it
 /// consumes a terminal result. Dropping the future cancels the waiter; cell
 /// cancellation returns any crossing capacity or HTTP/1 handle after releasing
 /// the cell lock.
 struct WaiterGuard {
     /// Cell retaining the waiter.
     cell: Arc<OriginCell>,
     /// Waiter cancelled if the request future does not consume a result.
     waiter: WaiterId,
     /// Whether request cancellation is still armed.
     active: bool,
--
+}
--
 impl WaiterGuard {
     /// Arms cancellation for a newly registered waiter.
     fn new(cell: Arc<OriginCell>, waiter: WaiterId) -> Self {
         Self {
             cell,
             waiter,
             active: true,
--
+        }
--
+    }
--
     /// Disarms after the request consumes its terminal result.
     fn disarm(&mut self) {
         self.active = false;
--
+    }
--
+}
--
 impl Drop for WaiterGuard {
     fn drop(&mut self) {
         if self.active {
             OriginCell::cancel_waiter(&self.cell, self.waiter);
--
+        }
--
+    }
-148
+}
-149
 /// Attaches connection return and dispatch completion to an accepted response.
 ///
 /// An ordinary response retains both guards until Hyper proves a complete
 /// message boundary. Successful CONNECT and `101` responses retire the pool
 /// record before the response exposes Hyper's upgraded root I/O.
 fn guard_response(
     response: Response<hyper::body::Incoming>,
     method: Method,

↧ Expand context

     exchange: H1Exchange,
     dispatch: DispatchGuard,
     spawner: StdArc<dyn DriverSpawner>,
 ) -> Response<SdkBody> {
     let upgrade = response.status() == http_1x::StatusCode::SWITCHING_PROTOCOLS
         || (method == Method::CONNECT && response.status().is_success());
     let (parts, body) = response.into_parts();
     if upgrade {
         exchange.retire_connection(CloseReason::Upgraded);
         dispatch.complete();
         return Response::from_parts(parts, SdkBody::from_body_1_x(body));
-169
+    }
     let body = H1ResponseBody::new(body, exchange, dispatch, spawner);
     Response::from_parts(parts, SdkBody::from_body_1_x(body))
-172
+}
-173
 /// Response body that retains the HTTP/1 exchange until its message boundary.
 ///
 /// End-of-stream returns a ready request handle to pool policy. A body error,
 /// or dropping the body before Hyper reports end-of-stream, retires the

@@ -575,378 +668,474 @@

↥ Expand context

             exchange.retire_connection(CloseReason::OwnerRuntimeShutdown);
-379
+        }
-380
+    }
-381
+}
-382
 /// Adds the mandatory HTTP/1 Host header from the absolute request URI.
 fn add_host_header(
     request: &mut Request<SdkBody>,
     absolute_uri: &Uri,
 ) -> Result<(), RequestDispatchError> {
     if request.headers().contains_key(http_1x::header::HOST) {
         return Ok(());
-390
+    }
     let authority = absolute_uri
         .authority()
         .ok_or(RequestDispatchError::MissingAuthority)?;
     let default_port = match absolute_uri.scheme_str() {
         Some("http") => Some(80),
         Some("https") => Some(443),
         _ => None,

     };
     let host = match absolute_uri.port_u16() {
         Some(port) if Some(port) != default_port => {
             format!("{}:{port}", authority.host())
-402
+        }
         _ => authority.host().to_string(),
     };
     let value =
         http_1x::HeaderValue::from_str(&host).map_err(RequestDispatchError::InvalidHostHeader)?;
     request.headers_mut().insert(http_1x::header::HOST, value);
     request.extensions_mut().insert(H1HostHeaderInserted);
     Ok(())
-410
+}
-411
 /// Rewrites an absolute URI into the request target required by HTTP/1.
 fn rewrite_h1_request_target(request: &mut Request<SdkBody>, is_proxied: bool) {
     use http_1x::uri::Parts;
-415
     if request.method() == Method::CONNECT {
         if let Some(authority) = request.uri().authority().cloned() {
             let mut parts = Parts::default();
             parts.authority = Some(authority);
             *request.uri_mut() = Uri::from_parts(parts).expect("request authority is a valid URI");
-421
+        }
         return;
-423
+    }
     if is_proxied {
         return;
-426
+    }
     let mut parts = Parts::default();
     parts.path_and_query = request.uri().path_and_query().cloned();
     *request.uri_mut() = Uri::from_parts(parts).expect("request path and query are a valid URI");
-430
+}
-431
 #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
 mod tests {
     use super::*;
     use crate::client::pool::cell::h1::H1Sender;
     use crate::client::pool::cell::OriginCell;
     use crate::client::pool::connection::ConnectionInfo;
     use crate::client::pool::registry::PartitionState;
     use crate::client::pool::{
         Client, ConnectionId, ConnectionPool, ConnectionReuseScope, Partition, PartitionId,
         TokioDriverSpawner,
     };
     use crate::client::timeout::test::{NeverConnects, NeverReplies};
     use crate::sync::Arc;
     use aws_smithy_async::rt::sleep::TokioSleep;
     use aws_smithy_runtime_api::client::connection::CaptureSmithyConnection;
     use aws_smithy_runtime_api::client::http::{HttpClient, HttpConnector, HttpConnectorSettings};
     use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
     use aws_smithy_runtime_api::client::runtime_components::RuntimeComponentsBuilder;
     use http_body_util::BodyExt;
     use hyper_util::client::legacy::connect::{Connected, Connection};
     use hyper_util::rt::TokioIo;
     use std::future::Future;
     use std::io::{self, IoSlice};

↧ Expand context

     use std::sync::atomic::{AtomicUsize, Ordering};
     use std::task::{Context, Poll};
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio::net::{TcpListener, TcpStream};
     use tokio::sync::Notify;
     use tower::Service;
-461
     #[derive(Debug)]
     struct DroppingSpawner;
-464
     impl DriverSpawner for DroppingSpawner {
         fn spawn(&self, driver: Pin<Box<dyn std::future::Future<Output = ()> + Send + 'static>>) {
             drop(driver);
-468
+        }
-469
+    }
-470
     #[derive(Clone, Debug)]
     struct TrackingSpawner {
         active: StdArc<AtomicUsize>,
-474
+    }

@@ -829,635 +889,695 @@

↥ Expand context

             OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
-636
         drop(H1ReadinessTask::new(selection.into_exchange()));
-638
         assert_eq!(
             Some(CloseReason::OwnerRuntimeShutdown),
             connection.snapshot().close_reason
         );
-643
+    }
-644
     #[tokio::test]
     async fn dropped_owner_task_completes_the_establishment_waiter() {
         let pool = ConnectionPool::builder()
             .idle_timeout(None)
             .partitions([Partition::new(PartitionId::from_index(7), DroppingSpawner)])
             .build_http()
             .unwrap();
         let partition = pool
             .inner
             .registry

             .partition(PartitionId::from_index(7))
             .unwrap();
         let request = Request::get("http://127.0.0.1:9/")
             .body(SdkBody::empty())
             .unwrap();
-660
         let error = pool
             .send_request(partition, request, None)
             .await
             .expect_err("dropped establishment task did not fail the request");
         assert!(error.is_other());
         assert!(error.is_io());
         assert_eq!(
             "the owner-runtime connection establishment task was dropped",
             std::error::Error::source(&error).unwrap().to_string()
         );
-670
+    }
-671
     struct TestServer {
         endpoint: String,
         accepted: StdArc<AtomicUsize>,
         task: tokio::task::JoinHandle<()>,

↧ Expand context

-676
+    }
-677
     impl TestServer {
         async fn start() -> Self {
             let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
             let endpoint = format!("http://{}", listener.local_addr().unwrap());
             let accepted = StdArc::new(AtomicUsize::new(0));
             let accepted_for_task = accepted.clone();
             let task = tokio::spawn(async move {
                 while let Ok((stream, _)) = listener.accept().await {
                     accepted_for_task.fetch_add(1, Ordering::SeqCst);
                     tokio::spawn(serve_connection(stream));
-688
+                }
             });
             Self {
                 endpoint,
                 accepted,
                 task,
-694
+            }
-695
+        }

@@ -1322,1128 +1405,1227 @@

↥ Expand context

         assert!(second.await.unwrap_err().is_cancelled());
         for _ in 0..10 {
             if cell.retained_waiters_for_test() == 0 {
                 break;
-1132
+            }
             tokio::task::yield_now().await;
-1134
+        }
         assert_eq!(0, cell.retained_waiters_for_test());
         assert_eq!(0, cell.admission().unwrap().ordered_demand_count_for_test());
-1137
         consume(first).await;
         consume(request(&pool, partition, server.uri("/third")).await).await;
         assert_eq!(1, server.accepted());
-1141
+    }
-1142
     #[tokio::test]
     async fn dropping_incomplete_body_retires_connection() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         let partition = anonymous_partition(&pool);

-1148
         drop(request(&pool, partition.clone(), server.uri("/first")).await);
         assert_eq!(
             hyper::body::Bytes::from_static(b"ok"),
             consume(request(&pool, partition, server.uri("/second")).await).await
         );
         assert_eq!(2, server.accepted());
-1155
+    }
-1156
     #[tokio::test]
     async fn incompatible_h1_selection_reports_connection_identity() {
     async fn incompatible_h1_result_remains_reusable_for_compatible_demand() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         let pool = ConnectionPool::builder()
             .max_connections_per_host(1)
             .build_http()
             .unwrap();
         let partition = anonymous_partition(&pool);
         let capture = CaptureSmithyConnection::new();
         let mut request = Request::get(server.uri("/h2"))
         let mut h2_request = Request::get(server.uri("/h2"))
             .version(Version::HTTP_2)
             .body(SdkBody::empty())
             .unwrap();
         request.extensions_mut().insert(capture.clone());
         h2_request.extensions_mut().insert(capture.clone());
-1171
         let error = pool
             .send_request(partition, request, None)
             .send_request(partition.clone(), h2_request, None)
             .await
             .expect_err("an HTTP/2 request unexpectedly used HTTP/1");
         assert!(error.is_user());
         assert!(error
         let error_connection = error
             .connection_metadata()
             .and_then(|metadata| metadata.connection_id())
             .is_some());
         assert!(capture
             .expect("the mismatch error omitted connection metadata");
         let captured_connection = capture
             .get()
             .and_then(|metadata| metadata.connection_id())
             .is_some());
             .expect("the mismatch request did not capture its connection");
         assert_eq!(
             error_connection.connection_id(),
             captured_connection.connection_id()
         );
-+
         let response = request(&pool, partition, server.uri("/h1")).await;
         assert_eq!(
             hyper::body::Bytes::from_static(b"ok"),
             consume(response).await
         );
         assert_eq!(
 ,
             server.accepted(),
             "the rejected HTTP/2 request discarded a reusable HTTP/1 connection"
         );
-1198
+    }
-1199
     #[tokio::test]
     async fn captured_metadata_poisons_only_selected_generation() {
         let server = TestServer::start().await;
         let pool = ConnectionPool::builder().build_http().unwrap();
         let partition = anonymous_partition(&pool);
         let capture = CaptureSmithyConnection::new();
         let mut first = Request::get(server.uri("/first"))
             .body(SdkBody::empty())

↧ Expand context

             .unwrap();
         first.extensions_mut().insert(capture.clone());
         let response = pool
             .send_request(partition.clone(), first, None)
             .await
             .unwrap();
         consume(response).await;
-1215
         let metadata = capture.get().expect("connection metadata was captured");
         assert!(metadata.connection_id().is_some());
         metadata.poison();
-1219
         consume(request(&pool, partition.clone(), server.uri("/second")).await).await;
         assert_eq!(2, server.accepted());
-1222
         metadata.poison();
         consume(request(&pool, partition, server.uri("/third")).await).await;
         assert_eq!(
 ,
             server.accepted(),

Pages: 1 2 3 4 5 6 7