1 + | /*
|
2 + | * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
3 + | * SPDX-License-Identifier: Apache-2.0
|
4 + | */
|
5 + |
|
6 + | //! HTTP/2 flight, generation, route, and request-lease ownership.
|
7 + | //!
|
8 + | //! One connection-owning cell stores the authoritative HTTP/2 generation.
|
9 + | //! Other cells store only [`H2Route`] identities and revalidate them at the
|
10 + | //! connection-owning cell before dispatch. A generation may stop accepting
|
11 + | //! new requests while prospective and accepted requests still drain.
|
12 + | //!
|
13 + | //! [`H2Records`] is the invariant owner under the cell lock. It stores one
|
14 + | //! post-ALPN flight, the accepting and draining generations, a local generation
|
15 + | //! gate, and at most one peer route. A route carries only connection partition
|
16 + | //! and generation identity. [`H2Activation`] is the unlocked value that carries
|
17 + | //! a prospective lease and transient sender from selection to Hyper acceptance.
|
18 + | //!
|
19 + | //! ```text
|
20 + | //! requesting cell connection-owning cell
|
21 + | //! peer route + local gate --identity---> exact accepting generation
|
22 + | //! |-- increment prospective count
|
23 + | //! `-- clone transient sender
|
24 + | //! |
|
25 + | //! `--> H2Activation
|
26 + | //! ```
|
27 + | //!
|
28 + | //! ```text
|
29 + | //! no flight -- post-ALPN owner task ----------------------> Flight
|
30 + | //! Flight -- successful handshake and install ------------> Accepting
|
31 + | //! Flight -- failure, stale completion, or task drop ------> no flight
|
32 + | //! Accepting -- close with retained request leases --------> Draining
|
33 + | //! Accepting -- close without retained request leases -----> removed
|
34 + | //! Draining -- last prospective or accepted lease --------> removed
|
35 + | //! ```
|
36 + | //!
|
37 + | //! ```text
|
38 + | //! no peer route -- publication ---------------------------> PeerRoute(generation)
|
39 + | //! PeerRoute(A) -- replacement publication ---------------> PeerRoute(B)
|
40 + | //! PeerRoute -- stale activation or local generation -----> no peer route
|
41 + | //! ```
|
42 + | //!
|
43 + | //! Generation installation makes the sender visible before the owner task
|
44 + | //! submits the Hyper driver. An activation in that interval is accepted by
|
45 + | //! Hyper's dispatch channel and remains pending until the driver is polled.
|
46 + | //!
|
47 + | //! Activation reserves a prospective lease before a sender clone leaves the
|
48 + | //! cell lock. Hyper acceptance converts that reservation to one accepted
|
49 + | //! lease. The accepted lease is released only after both request-send and
|
50 + | //! response-receive endpoints terminate.
|
51 + |
|
52 + | use super::super::connection::{ConnectionInfo, ConnectionState, DispatchGuard};
|
53 + | use super::super::partition::PartitionId;
|
54 + | use super::waiters::{AcquisitionQueue, WaiterInstall};
|
55 + | use super::{AcquisitionEvent, AcquisitionResult, CellState, OriginCell, WaiterId};
|
56 + | use crate::sync::{Arc, Mutex, Weak};
|
57 + | use aws_smithy_runtime_api::client::connection::ConnectionId;
|
58 + | use aws_smithy_types::body::SdkBody;
|
59 + | use std::collections::{BTreeSet, HashMap};
|
60 + | use std::time::SystemTime;
|
61 + |
|
62 + | /// Identity of one post-ALPN HTTP/2 establishment flight.
|
63 + | #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
|
64 + | pub(in crate::client::pool) struct H2FlightId(u64);
|
65 + |
|
66 + | /// Identity of one installed HTTP/2 generation within a cell.
|
67 + | #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
|
68 + | pub(in crate::client::pool) struct H2GenerationId(u64);
|
69 + |
|
70 + | #[cfg(test)]
|
71 + | impl H2GenerationId {
|
72 + | /// Creates a generation identity for admission-index tests.
|
73 + | pub(in crate::client::pool) fn for_test(value: u64) -> Self {
|
74 + | Self(value)
|
75 + | }
|
76 + | }
|
77 + |
|
78 + | /// Cloneable Hyper request sender retained only by its owning generation.
|
79 + | ///
|
80 + | /// The test-only variant exercises generation and lease transitions without a
|
81 + | /// parallel fake generation implementation.
|
82 + | #[derive(Clone)]
|
83 + | pub(in crate::client::pool) enum H2Sender {
|
84 + | /// Sender returned by a successful Hyper HTTP/2 handshake.
|
85 + | Hyper(hyper::client::conn::http2::SendRequest<SdkBody>),
|
86 + | /// Synthetic sender identity used by state-machine tests.
|
87 + | #[cfg(test)]
|
88 + | Test(u64),
|
89 + | }
|
90 + |
|
91 + | impl H2Sender {
|
92 + | /// Wraps a sender produced by a successful Hyper HTTP/2 handshake.
|
93 + | pub(in crate::client::pool) fn from_hyper(
|
94 + | inner: hyper::client::conn::http2::SendRequest<SdkBody>,
|
95 + | ) -> Self {
|
96 + | Self::Hyper(inner)
|
97 + | }
|
98 + |
|
99 + | /// Returns whether Hyper has observed connection closure.
|
100 + | pub(in crate::client::pool) fn is_closed(&self) -> bool {
|
101 + | match self {
|
102 + | Self::Hyper(sender) => sender.is_closed(),
|
103 + | #[cfg(test)]
|
104 + | Self::Test(_) => false,
|
105 + | }
|
106 + | }
|
107 + |
|
108 + | /// Returns mutable access to a transient sender clone for dispatch.
|
109 + | ///
|
110 + | /// # Panics
|
111 + | ///
|
112 + | /// Panics when a test-only sender reaches real dispatch.
|
113 + | pub(in crate::client::pool) fn hyper_mut(
|
114 + | &mut self,
|
115 + | ) -> &mut hyper::client::conn::http2::SendRequest<SdkBody> {
|
116 + | match self {
|
117 + | Self::Hyper(sender) => sender,
|
118 + | #[cfg(test)]
|
119 + | Self::Test(_) => panic!("test HTTP/2 sender reached Hyper dispatch"),
|
120 + | }
|
121 + | }
|
122 + |
|
123 + | /// Creates a synthetic sender for state-machine tests.
|
124 + | #[cfg(test)]
|
125 + | fn test(id: u64) -> Self {
|
126 + | Self::Test(id)
|
127 + | }
|
128 + | }
|
129 + |
|
130 + | impl std::fmt::Debug for H2Sender {
|
131 + | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
132 + | match self {
|
133 + | Self::Hyper(_) => f.write_str("H2Sender::Hyper"),
|
134 + | #[cfg(test)]
|
135 + | Self::Test(id) => f.debug_tuple("H2Sender::Test").field(id).finish(),
|
136 + | }
|
137 + | }
|
138 + | }
|
139 + |
|
140 + | /// Identity-only reference to one accepting generation.
|
141 + | ///
|
142 + | /// The route owns no sender, connection capacity, driver, or socket. Every
|
143 + | /// activation upgrades the cell reference and checks the generation identity.
|
144 + | #[derive(Clone, Debug)]
|
145 + | pub(in crate::client::pool) struct H2Route {
|
146 + | id: H2RouteId,
|
147 + | connection_cell: Weak<OriginCell>,
|
148 + | }
|
149 + |
|
150 + | /// Exact connection-owning generation named by one peer route.
|
151 + | #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
|
152 + | pub(super) struct H2RouteId {
|
153 + | connection_partition: PartitionId,
|
154 + | generation: H2GenerationId,
|
155 + | }
|
156 + |
|
157 + | impl H2Route {
|
158 + | /// Creates a route after its generation has been installed.
|
159 + | pub(in crate::client::pool) fn new(
|
160 + | connection_cell: &Arc<OriginCell>,
|
161 + | generation: H2GenerationId,
|
162 + | ) -> Self {
|
163 + | Self {
|
164 + | connection_cell: Weak::from_arc(connection_cell),
|
165 + | id: H2RouteId {
|
166 + | connection_partition: connection_cell.id().partition(),
|
167 + | generation,
|
168 + | },
|
169 + | }
|
170 + | }
|
171 + |
|
172 + | /// Returns the exact connection-owning generation named by this route.
|
173 + | fn id(&self) -> H2RouteId {
|
174 + | self.id
|
175 + | }
|
176 + |
|
177 + | /// Returns the partition that owns the advertised generation.
|
178 + | pub(super) fn connection_partition(&self) -> PartitionId {
|
179 + | self.id.connection_partition
|
180 + | }
|
181 + |
|
182 + | /// Returns the advertised generation identity.
|
183 + | #[cfg(any(debug_assertions, test))]
|
184 + | pub(in crate::client::pool) fn generation(&self) -> H2GenerationId {
|
185 + | self.id.generation
|
186 + | }
|
187 + |
|
188 + | /// Attempts to reserve a prospective request lease for one requesting partition.
|
189 + | pub(super) fn activate(&self, request_partition: PartitionId) -> Option<H2Activation> {
|
190 + | let cell = self.connection_cell.upgrade()?;
|
191 + | let mut activation = OriginCell::activate_h2(&cell, self.id.generation)?;
|
192 + | activation.request_partition = request_partition;
|
193 + | Some(activation)
|
194 + | }
|
195 + | }
|
196 + |
|
197 + | /// Cell-local ownership of one HTTP/2 flight and installed generations.
|
198 + | ///
|
199 + | /// At every completed transition:
|
200 + | ///
|
201 + | /// - at most one flight and one accepting generation exist;
|
202 + | /// - the accepting identity names an `Accepting` record;
|
203 + | /// - every other generation is `Draining`;
|
204 + | /// - prospective and accepted request counts are checked and non-wrapping;
|
205 + | /// - a draining record remains until both counts reach zero; and
|
206 + | /// - each flight participant identity appears at most once.
|
207 + | #[derive(Debug, Default)]
|
208 + | pub(super) struct H2Records {
|
209 + | /// Post-ALPN convergence in progress for this cell.
|
210 + | flight: Option<H2Flight>,
|
211 + | /// Installed accepting and draining generations by exact identity.
|
212 + | generations: HashMap<H2GenerationId, H2Generation>,
|
213 + | /// Sole generation permitted to issue new activations.
|
214 + | accepting: Option<H2GenerationId>,
|
215 + | /// Identity-only route to one peer cell's accepting generation.
|
216 + | peer_route: Option<PeerH2Route>,
|
217 + | /// Local activation order for the accepting generation.
|
218 + | gate: GenerationGate,
|
219 + | /// Next cell-local flight identity.
|
220 + | next_flight: u64,
|
221 + | /// Next cell-local generation identity.
|
222 + | next_generation: u64,
|
223 + | }
|
224 + |
|
225 + | /// One post-ALPN flight and the waiters awaiting its result.
|
226 + | #[derive(Debug)]
|
227 + | struct H2Flight {
|
228 + | /// Identity checked by the owner task at completion.
|
229 + | id: H2FlightId,
|
230 + | /// Waiters that receive the shared handshake result.
|
231 + | participants: BTreeSet<WaiterId>,
|
232 + | }
|
233 + |
|
234 + | /// One installed multiplexed connection generation.
|
235 + | #[derive(Debug)]
|
236 + | struct H2Generation {
|
237 + | /// Establishment waiters transferred to this generation but not yet served.
|
238 + | pending_waiters: BTreeSet<WaiterId>,
|
239 + | /// Protocol-neutral connection and capacity owner.
|
240 + | connection: Arc<ConnectionState>,
|
241 + | /// Authoritative Hyper sender cloned only after generation validation.
|
242 + | sender: H2Sender,
|
243 + | /// Whether new activations may be issued.
|
244 + | residence: H2Residence,
|
245 + | /// Activations selected but not yet accepted by Hyper.
|
246 + | prospective: usize,
|
247 + | /// Whether Hyper has accepted a request on this generation.
|
248 + | has_dispatched: bool,
|
249 + | /// Accepted requests whose two lease endpoints have not both terminated.
|
250 + | accepted: usize,
|
251 + | /// Expiration deadline while the generation remains accepting.
|
252 + | idle_deadline: Option<SystemTime>,
|
253 + | }
|
254 + |
|
255 + | /// Whether an installed generation may accept new requests.
|
256 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
257 + | enum H2Residence {
|
258 + | /// New requests may reserve prospective leases.
|
259 + | Accepting,
|
260 + | /// No new requests are admitted; retained leases may finish.
|
261 + | Draining,
|
262 + | }
|
263 + |
|
264 + | /// Peer route visibility and requesting-cell activation order.
|
265 + | #[derive(Debug)]
|
266 + | struct PeerH2Route {
|
267 + | /// Exact connection-cell generation visible to this requesting cell.
|
268 + | route: H2Route,
|
269 + | /// Local waiter priority for uses of this route.
|
270 + | gate: GenerationGate,
|
271 + | /// Waiter whose route activation is crossing the connection-cell lock.
|
272 + | crossing: Option<WaiterId>,
|
273 + | }
|
274 + |
|
275 + | /// Local priority state for one accepting generation.
|
276 + | #[derive(Debug, Default)]
|
277 + | enum GenerationGate {
|
278 + | /// No accepting generation is visible.
|
279 + | #[default]
|
280 + | Closed,
|
281 + | /// Waiters committed through `cutoff` precede later arrivals.
|
282 + | Prioritizing {
|
283 + | /// Exact generation governed by this gate.
|
284 + | generation: H2GenerationId,
|
285 + | /// Newest waiter committed before generation visibility.
|
286 + | cutoff: WaiterId,
|
287 + | /// Prioritized waiter whose activation has not accepted or cancelled.
|
288 + | activating: Option<WaiterId>,
|
289 + | },
|
290 + | /// The publication cutoff drained; queued work still precedes direct arrivals.
|
291 + | Open { generation: H2GenerationId },
|
292 + | }
|
293 + |
|
294 + | /// One activation opportunity returned by a generation gate.
|
295 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
296 + | enum GateTurn {
|
297 + | /// The gate cannot issue another activation.
|
298 + | Unavailable,
|
299 + | /// Any oldest compatible acquisition may activate.
|
300 + | Open,
|
301 + | /// Only an acquisition at or before this waiter may activate.
|
302 + | Through(WaiterId),
|
303 + | }
|
304 + |
|
305 + | impl GenerationGate {
|
306 + | /// Creates a gate over every waiter committed through `cutoff`.
|
307 + | fn for_generation(generation: H2GenerationId, cutoff: Option<WaiterId>) -> Self {
|
308 + | match cutoff {
|
309 + | Some(cutoff) => Self::Prioritizing {
|
310 + | generation,
|
311 + | cutoff,
|
312 + | activating: None,
|
313 + | },
|
314 + | None => Self::Open { generation },
|
315 + | }
|
316 + | }
|
317 + |
|
318 + | fn generation(&self) -> Option<H2GenerationId> {
|
319 + | match self {
|
320 + | Self::Closed => None,
|
321 + | Self::Prioritizing { generation, .. } | Self::Open { generation } => Some(*generation),
|
322 + | }
|
323 + | }
|
324 + |
|
325 + | /// Returns the next activation opportunity and opens a drained priority gate.
|
326 + | fn next_turn(&mut self, has_prioritized: bool) -> GateTurn {
|
327 + | match self {
|
328 + | Self::Closed => GateTurn::Unavailable,
|
329 + | Self::Prioritizing {
|
330 + | cutoff,
|
331 + | activating,
|
332 + | generation,
|
333 + | } => {
|
334 + | if activating.is_some() {
|
335 + | return GateTurn::Unavailable;
|
336 + | }
|
337 + | if !has_prioritized {
|
338 + | let generation = *generation;
|
339 + | *self = Self::Open { generation };
|
340 + | GateTurn::Open
|
341 + | } else {
|
342 + | GateTurn::Through(*cutoff)
|
343 + | }
|
344 + | }
|
345 + | Self::Open { .. } => GateTurn::Open,
|
346 + | }
|
347 + | }
|
348 + |
|
349 + | /// Records a prioritized activation until it accepts or cancels.
|
350 + | fn begin_gate_activation(&mut self, waiter: WaiterId) -> bool {
|
351 + | let activating = match self {
|
352 + | Self::Closed => {
|
353 + | unreachable!("started an HTTP/2 activation while its gate was closed")
|
354 + | }
|
355 + | Self::Prioritizing { activating, .. } => activating,
|
356 + | Self::Open { .. } => return false,
|
357 + | };
|
358 + | assert!(
|
359 + | activating.replace(waiter).is_none(),
|
360 + | "HTTP/2 generation gate admitted two activation opportunities"
|
361 + | );
|
362 + | true
|
363 + | }
|
364 + |
|
365 + | /// Discharges one exact activation opportunity.
|
366 + | fn finish_gate_activation(&mut self, generation: H2GenerationId, waiter: WaiterId) -> bool {
|
367 + | if self.generation() != Some(generation) {
|
368 + | return false;
|
369 + | }
|
370 + | let activating = match self {
|
371 + | Self::Closed => return false,
|
372 + | Self::Prioritizing { activating, .. } => activating,
|
373 + | Self::Open { .. } => return false,
|
374 + | };
|
375 + | if *activating != Some(waiter) {
|
376 + | return false;
|
377 + | }
|
378 + | *activating = None;
|
379 + | true
|
380 + | }
|
381 + |
|
382 + | fn priority_cutoff(&self) -> Option<WaiterId> {
|
383 + | match self {
|
384 + | Self::Prioritizing { cutoff, .. } => Some(*cutoff),
|
385 + | Self::Closed | Self::Open { .. } => None,
|
386 + | }
|
387 + | }
|
388 + |
|
389 + | fn activating(&self) -> Option<WaiterId> {
|
390 + | match self {
|
391 + | Self::Prioritizing { activating, .. } => *activating,
|
392 + | Self::Closed | Self::Open { .. } => None,
|
393 + | }
|
394 + | }
|
395 + |
|
396 + | fn is_open_for(&self, generation: H2GenerationId) -> bool {
|
397 + | matches!(self, Self::Open { generation: current } if *current == generation)
|
398 + | }
|
399 + | }
|
400 + |
|
401 + | /// Result of atomically converging one post-ALPN attempt.
|
402 + | #[derive(Debug)]
|
403 + | pub(in crate::client::pool) enum H2FlightInstall {
|
404 + | /// An installed generation can serve this waiter.
|
405 + | Accepting(H2GenerationId),
|
406 + | /// The waiter joined the current flight as a result participant.
|
407 + | Joined,
|
408 + | /// The caller owns the task that must drive this new flight.
|
409 + | Driver(H2FlightId),
|
410 + | }
|
411 + |
|
412 + | /// Result of joining an accepting generation after ALPN selected HTTP/2.
|
413 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
414 + | pub(in crate::client::pool) enum H2GenerationJoin {
|
415 + | /// The waiter was transferred to the named generation.
|
416 + | Joined,
|
417 + | /// The named generation stopped accepting before the transfer.
|
418 + | GenerationChanged,
|
419 + | /// Another transition already installed the waiter's result.
|
420 + | WaiterCompleted,
|
421 + | }
|
422 + |
|
423 + | impl H2Records {
|
424 + | /// Installs or joins the one current flight after ALPN selected HTTP/2.
|
425 + | pub(super) fn install_or_join_flight(&mut self, waiter: WaiterId) -> H2FlightInstall {
|
426 + | if let Some(generation) = self.accepting {
|
427 + | return H2FlightInstall::Accepting(generation);
|
428 + | }
|
429 + | if let Some(flight) = &mut self.flight {
|
430 + | assert!(
|
431 + | flight.participants.insert(waiter),
|
432 + | "HTTP/2 waiter joined one flight more than once"
|
433 + | );
|
434 + | self.assert_consistent();
|
435 + | return H2FlightInstall::Joined;
|
436 + | }
|
437 + |
|
438 + | let id = self.take_flight_id();
|
439 + | self.flight = Some(H2Flight {
|
440 + | id,
|
441 + | participants: BTreeSet::from([waiter]),
|
442 + | });
|
443 + | self.assert_consistent();
|
444 + | H2FlightInstall::Driver(id)
|
445 + | }
|
446 + |
|
447 + | /// Removes one cancelled waiter from its current flight, when present.
|
448 + | pub(super) fn cancel_flight_participant(&mut self, waiter: WaiterId) {
|
449 + | if let Some(flight) = &mut self.flight {
|
450 + | flight.participants.remove(&waiter);
|
451 + | }
|
452 + | self.assert_consistent();
|
453 + | }
|
454 + |
|
455 + | /// Removes a waiter that no longer awaits an accepting generation.
|
456 + | pub(super) fn cancel_pending_waiter(&mut self, waiter: WaiterId) {
|
457 + | if let Some(generation) = self.accepting {
|
458 + | self.generations
|
459 + | .get_mut(&generation)
|
460 + | .expect("accepting HTTP/2 generation disappeared")
|
461 + | .pending_waiters
|
462 + | .remove(&waiter);
|
463 + | }
|
464 + | self.assert_consistent();
|
465 + | }
|
466 + |
|
467 + | /// Installs the generation produced by the named flight.
|
468 + | pub(in crate::client::pool) fn complete_flight(
|
469 + | &mut self,
|
470 + | flight: H2FlightId,
|
471 + | connection: Arc<ConnectionState>,
|
472 + | sender: H2Sender,
|
473 + | idle_deadline: Option<SystemTime>,
|
474 + | ) -> Result<H2GenerationId, (Arc<ConnectionState>, H2Sender)> {
|
475 + | if self.flight.as_ref().map(|current| current.id) != Some(flight)
|
476 + | || self.accepting.is_some()
|
477 + | {
|
478 + | return Err((connection, sender));
|
479 + | }
|
480 + | let flight = self
|
481 + | .flight
|
482 + | .take()
|
483 + | .expect("validated HTTP/2 flight disappeared");
|
484 + | let generation =
|
485 + | self.install_generation(flight.participants, connection, sender, idle_deadline);
|
486 + | self.assert_consistent();
|
487 + | Ok(generation)
|
488 + | }
|
489 + |
|
490 + | /// Installs one accepting generation with its transferred waiters.
|
491 + | fn install_generation(
|
492 + | &mut self,
|
493 + | pending_waiters: BTreeSet<WaiterId>,
|
494 + | connection: Arc<ConnectionState>,
|
495 + | sender: H2Sender,
|
496 + | idle_deadline: Option<SystemTime>,
|
497 + | ) -> H2GenerationId {
|
498 + | let generation = self.take_generation_id();
|
499 + | let replaced = self.generations.insert(
|
500 + | generation,
|
501 + | H2Generation {
|
502 + | pending_waiters,
|
503 + | connection,
|
504 + | sender,
|
505 + | residence: H2Residence::Accepting,
|
506 + | prospective: 0,
|
507 + | has_dispatched: false,
|
508 + | accepted: 0,
|
509 + | idle_deadline,
|
510 + | },
|
511 + | );
|
512 + | assert!(replaced.is_none(), "HTTP/2 generation identity was reused");
|
513 + | self.accepting = Some(generation);
|
514 + | self.peer_route = None;
|
515 + | debug_assert!(matches!(self.gate, GenerationGate::Closed));
|
516 + | self.gate = GenerationGate::Open { generation };
|
517 + | generation
|
518 + | }
|
519 + |
|
520 + | /// Retires a failed or dropped flight and returns its participants.
|
521 + | pub(super) fn fail_flight(&mut self, flight: H2FlightId) -> Option<Vec<WaiterId>> {
|
522 + | if self.flight.as_ref().map(|current| current.id) != Some(flight) {
|
523 + | return None;
|
524 + | }
|
525 + | let participants = self
|
526 + | .flight
|
527 + | .take()
|
528 + | .expect("validated HTTP/2 flight disappeared")
|
529 + | .participants
|
530 + | .into_iter()
|
531 + | .collect();
|
532 + | self.assert_consistent();
|
533 + | Some(participants)
|
534 + | }
|
535 + |
|
536 + | /// Closes the gate over waiters committed when a generation became visible.
|
537 + | pub(super) fn prioritize_through(&mut self, cutoff: Option<WaiterId>) {
|
538 + | let generation = self
|
539 + | .accepting
|
540 + | .expect("HTTP/2 generation gate opened without an accepting generation");
|
541 + | self.gate = GenerationGate::for_generation(generation, cutoff);
|
542 + | self.assert_consistent();
|
543 + | }
|
544 + |
|
545 + | /// Extends priority to a post-ALPN waiter that found this generation.
|
546 + | pub(super) fn prioritize_waiter(
|
547 + | &mut self,
|
548 + | generation: H2GenerationId,
|
549 + | waiter: WaiterId,
|
550 + | ) -> bool {
|
551 + | if self.accepting != Some(generation) {
|
552 + | return false;
|
553 + | }
|
554 + | match &mut self.gate {
|
555 + | GenerationGate::Closed => {
|
556 + | unreachable!("accepting HTTP/2 generation had no generation gate")
|
557 + | }
|
558 + | GenerationGate::Prioritizing { cutoff, .. } => {
|
559 + | *cutoff = (*cutoff).max(waiter);
|
560 + | }
|
561 + | GenerationGate::Open { .. } => {
|
562 + | self.gate = GenerationGate::Prioritizing {
|
563 + | generation,
|
564 + | cutoff: waiter,
|
565 + | activating: None,
|
566 + | };
|
567 + | }
|
568 + | }
|
569 + | self.generations
|
570 + | .get_mut(&generation)
|
571 + | .expect("accepting HTTP/2 generation disappeared")
|
572 + | .pending_waiters
|
573 + | .insert(waiter);
|
574 + | self.assert_consistent();
|
575 + | true
|
576 + | }
|
577 + |
|
578 + | /// Returns the next queued activation opportunity.
|
579 + | fn next_gate_turn(&mut self, has_prioritized: bool) -> GateTurn {
|
580 + | self.gate.next_turn(has_prioritized)
|
581 + | }
|
582 + |
|
583 + | /// Records a prioritized activation and releases its transferred waiter.
|
584 + | pub(super) fn begin_gate_activation(&mut self, waiter: WaiterId) -> bool {
|
585 + | let gated = self.gate.begin_gate_activation(waiter);
|
586 + | if let Some(generation) = self.accepting {
|
587 + | self.generations
|
588 + | .get_mut(&generation)
|
589 + | .expect("accepting HTTP/2 generation disappeared")
|
590 + | .pending_waiters
|
591 + | .remove(&waiter);
|
592 + | }
|
593 + | self.assert_consistent();
|
594 + | gated
|
595 + | }
|
596 + |
|
597 + | /// Discharges one accepted or cancelled activation opportunity.
|
598 + | pub(super) fn finish_gate_activation(
|
599 + | &mut self,
|
600 + | generation: H2GenerationId,
|
601 + | waiter: WaiterId,
|
602 + | ) -> bool {
|
603 + | let finished = self.gate.finish_gate_activation(generation, waiter);
|
604 + | self.assert_consistent();
|
605 + | finished
|
606 + | }
|
607 + |
|
608 + | /// Returns whether a direct arrival may use the accepting generation.
|
609 + | pub(super) fn direct_is_allowed(&self, queued: bool) -> bool {
|
610 + | !queued && matches!(self.gate, GenerationGate::Open { .. })
|
611 + | }
|
612 + |
|
613 + | /// Returns the publication cutoff while older waiters remain prioritized.
|
614 + | fn priority_cutoff(&self) -> Option<WaiterId> {
|
615 + | self.gate.priority_cutoff()
|
616 + | }
|
617 + |
|
618 + | /// Reserves one prospective lease against an accepting generation.
|
619 + | fn activate(&mut self, generation: H2GenerationId) -> Option<H2ActivationParts> {
|
620 + | if self.accepting != Some(generation) {
|
621 + | return None;
|
622 + | }
|
623 + | let record = self.generations.get_mut(&generation)?;
|
624 + | if record.residence != H2Residence::Accepting {
|
625 + | return None;
|
626 + | }
|
627 + | record.prospective = record
|
628 + | .prospective
|
629 + | .checked_add(1)
|
630 + | .expect("HTTP/2 prospective request count exhausted");
|
631 + | let parts = H2ActivationParts {
|
632 + | sender: record.sender.clone(),
|
633 + | reused: record.has_dispatched,
|
634 + | connection: record.connection.clone(),
|
635 + | };
|
636 + | self.assert_consistent();
|
637 + | Some(parts)
|
638 + | }
|
639 + |
|
640 + | /// Converts one prospective reservation to an accepted request lease.
|
641 + | fn accept(&mut self, generation: H2GenerationId) -> bool {
|
642 + | let Some(record) = self.generations.get_mut(&generation) else {
|
643 + | return false;
|
644 + | };
|
645 + | if record.prospective == 0 {
|
646 + | return false;
|
647 + | }
|
648 + | record.prospective -= 1;
|
649 + | record.accepted = record
|
650 + | .accepted
|
651 + | .checked_add(1)
|
652 + | .expect("HTTP/2 accepted request count exhausted");
|
653 + | record.has_dispatched = true;
|
654 + | self.assert_consistent();
|
655 + | true
|
656 + | }
|
657 + |
|
658 + | /// Cancels one prospective reservation and detaches an empty drain record.
|
659 + | ///
|
660 + | /// # Panics
|
661 + | ///
|
662 + | /// Panics if the activation's exact generation or prospective count is
|
663 + | /// missing.
|
664 + | #[must_use]
|
665 + | fn cancel(&mut self, generation: H2GenerationId) -> Option<H2Generation> {
|
666 + | let record = self
|
667 + | .generations
|
668 + | .get_mut(&generation)
|
669 + | .expect("HTTP/2 activation generation disappeared before cancellation");
|
670 + | assert!(
|
671 + | record.prospective > 0,
|
672 + | "HTTP/2 prospective request count underflowed"
|
673 + | );
|
674 + | record.prospective -= 1;
|
675 + | self.remove_finished_drain(generation)
|
676 + | }
|
677 + |
|
678 + | /// Releases one accepted request and detaches an empty drain record.
|
679 + | ///
|
680 + | /// # Panics
|
681 + | ///
|
682 + | /// Panics if the lease's exact generation or accepted count is missing.
|
683 + | #[must_use]
|
684 + | fn complete_request(&mut self, generation: H2GenerationId) -> Option<H2Generation> {
|
685 + | let record = self
|
686 + | .generations
|
687 + | .get_mut(&generation)
|
688 + | .expect("HTTP/2 request generation disappeared before lease completion");
|
689 + | assert!(
|
690 + | record.accepted > 0,
|
691 + | "HTTP/2 accepted request count underflowed"
|
692 + | );
|
693 + | record.accepted -= 1;
|
694 + | self.remove_finished_drain(generation)
|
695 + | }
|
696 + |
|
697 + | /// Moves one exact accepting generation to draining.
|
698 + | #[must_use]
|
699 + | fn begin_close(&mut self, generation: H2GenerationId) -> Option<H2CloseTransition> {
|
700 + | if self.accepting != Some(generation) {
|
701 + | return None;
|
702 + | }
|
703 + | let record = self.generations.get_mut(&generation)?;
|
704 + | if record.residence != H2Residence::Accepting {
|
705 + | return None;
|
706 + | }
|
707 + | record.residence = H2Residence::Draining;
|
708 + | self.accepting = None;
|
709 + | self.gate = GenerationGate::Closed;
|
710 + | let pending_waiters = std::mem::take(&mut record.pending_waiters);
|
711 + | let remove_record = record.prospective == 0 && record.accepted == 0;
|
712 + | let connection = record.connection.clone();
|
713 + | let removed_generation = remove_record
|
714 + | .then(|| self.generations.remove(&generation))
|
715 + | .flatten();
|
716 + | Some(H2CloseTransition {
|
717 + | connection,
|
718 + | pending_waiters,
|
719 + | removed_generation,
|
720 + | })
|
721 + | }
|
722 + |
|
723 + | /// Detaches an exact accepting generation only while it has no request work.
|
724 + | #[must_use]
|
725 + | fn begin_idle_reclaim(&mut self, generation: H2GenerationId) -> Option<H2CloseTransition> {
|
726 + | self.is_idle(generation)
|
727 + | .then(|| self.begin_close(generation))
|
728 + | .flatten()
|
729 + | }
|
730 + |
|
731 + | /// Returns the accepting generation when one is locally reusable.
|
732 + | pub(in crate::client::pool) fn accepting(&self) -> Option<H2GenerationId> {
|
733 + | self.accepting
|
734 + | }
|
735 + |
|
736 + | /// Returns whether an exact generation remains accepting.
|
737 + | pub(super) fn is_accepting(&self, generation: H2GenerationId) -> bool {
|
738 + | self.accepting == Some(generation)
|
739 + | }
|
740 + |
|
741 + | /// Returns the generation peers may discover after the local cutoff drains.
|
742 + | pub(super) fn publishable_generation(&self) -> Option<H2GenerationId> {
|
743 + | let generation = self.accepting?;
|
744 + | matches!(
|
745 + | self.gate,
|
746 + | GenerationGate::Open { generation: gate_generation } if gate_generation == generation
|
747 + | )
|
748 + | .then_some(generation)
|
749 + | }
|
750 + |
|
751 + | /// Returns whether an exact publishable generation has no request work.
|
752 + | pub(super) fn is_idle(&self, generation: H2GenerationId) -> bool {
|
753 + | if self.publishable_generation() != Some(generation) {
|
754 + | return false;
|
755 + | }
|
756 + | self.generations.get(&generation).is_some_and(|record| {
|
757 + | record.pending_waiters.is_empty() && record.prospective == 0 && record.accepted == 0
|
758 + | })
|
759 + | }
|
760 + |
|
761 + | /// Returns whether a local generation or peer route suppresses admission demand.
|
762 + | pub(super) fn has_visible_h2(&self) -> bool {
|
763 + | self.accepting.is_some() || self.peer_route.is_some()
|
764 + | }
|
765 + |
|
766 + | /// Installs or refreshes one identity-only peer route.
|
767 + | pub(super) fn install_peer_route(&mut self, route: H2Route, cutoff: Option<WaiterId>) {
|
768 + | if self.accepting.is_some() {
|
769 + | self.peer_route = None;
|
770 + | self.assert_consistent();
|
771 + | return;
|
772 + | }
|
773 + | let id = route.id();
|
774 + | match &mut self.peer_route {
|
775 + | Some(current) if current.route.id() == id => {
|
776 + | if let Some(cutoff) = cutoff {
|
777 + | match &mut current.gate {
|
778 + | GenerationGate::Closed => {
|
779 + | unreachable!("visible peer HTTP/2 route had a closed gate")
|
780 + | }
|
781 + | GenerationGate::Prioritizing {
|
782 + | cutoff: current, ..
|
783 + | } => *current = (*current).max(cutoff),
|
784 + | GenerationGate::Open { generation } => {
|
785 + | current.gate = GenerationGate::Prioritizing {
|
786 + | generation: *generation,
|
787 + | cutoff,
|
788 + | activating: None,
|
789 + | };
|
790 + | }
|
791 + | }
|
792 + | }
|
793 + | }
|
794 + | _ => {
|
795 + | self.peer_route = Some(PeerH2Route {
|
796 + | gate: GenerationGate::for_generation(id.generation, cutoff),
|
797 + | route,
|
798 + | crossing: None,
|
799 + | });
|
800 + | }
|
801 + | }
|
802 + | self.assert_consistent();
|
803 + | }
|
804 + |
|
805 + | /// Marks one requesting waiter as the route's current activation opportunity.
|
806 + | fn prepare_peer_activation(
|
807 + | &mut self,
|
808 + | waiters: &AcquisitionQueue,
|
809 + | ) -> Option<PreparedPeerActivation> {
|
810 + | let peer = self.peer_route.as_mut()?;
|
811 + | if peer.crossing.is_some() {
|
812 + | return None;
|
813 + | }
|
814 + | let has_prioritized = peer
|
815 + | .gate
|
816 + | .priority_cutoff()
|
817 + | .is_some_and(|cutoff| waiters.has_h2_candidate_through(cutoff));
|
818 + | let cutoff = match peer.gate.next_turn(has_prioritized) {
|
819 + | GateTurn::Unavailable => return None,
|
820 + | GateTurn::Open => None,
|
821 + | GateTurn::Through(cutoff) => Some(cutoff),
|
822 + | };
|
823 + | let waiter = waiters.oldest_h2_candidate()?;
|
824 + | if cutoff.is_some_and(|cutoff| waiter > cutoff) {
|
825 + | return None;
|
826 + | }
|
827 + | let gated = peer.gate.begin_gate_activation(waiter);
|
828 + | peer.crossing = Some(waiter);
|
829 + | Some(PreparedPeerActivation {
|
830 + | route: peer.route.clone(),
|
831 + | waiter,
|
832 + | cutoff,
|
833 + | gated,
|
834 + | })
|
835 + | }
|
836 + |
|
837 + | /// Revalidates one peer activation after the connection-cell crossing.
|
838 + | fn peer_activation_is_current(
|
839 + | &self,
|
840 + | prepared: &PreparedPeerActivation,
|
841 + | waiters: &AcquisitionQueue,
|
842 + | ) -> bool {
|
843 + | self.peer_route.as_ref().is_some_and(|peer| {
|
844 + | peer.route.id() == prepared.route.id()
|
845 + | && peer.crossing == Some(prepared.waiter)
|
846 + | && (!prepared.gated || peer.gate.activating() == Some(prepared.waiter))
|
847 + | && waiters.is_oldest_h2_candidate(prepared.waiter)
|
848 + | })
|
849 + | }
|
850 + |
|
851 + | /// Ends one route crossing after its result is installed or rejected.
|
852 + | fn finish_peer_crossing(&mut self, route: H2RouteId, waiter: WaiterId) -> bool {
|
853 + | let Some(peer) = &mut self.peer_route else {
|
854 + | return false;
|
855 + | };
|
856 + | if peer.route.id() != route || peer.crossing != Some(waiter) {
|
857 + | return false;
|
858 + | }
|
859 + | peer.crossing = None;
|
860 + | self.assert_consistent();
|
861 + | true
|
862 + | }
|
863 + |
|
864 + | /// Discharges one prioritized peer-route activation opportunity.
|
865 + | fn finish_peer_gate(&mut self, route: H2RouteId, waiter: WaiterId) -> bool {
|
866 + | let Some(peer) = &mut self.peer_route else {
|
867 + | return false;
|
868 + | };
|
869 + | if peer.route.id() != route {
|
870 + | return false;
|
871 + | }
|
872 + | let finished = peer.gate.finish_gate_activation(route.generation, waiter);
|
873 + | self.assert_consistent();
|
874 + | finished
|
875 + | }
|
876 + |
|
877 + | /// Clears a peer activation marker when its requesting waiter is cancelled.
|
878 + | pub(super) fn cancel_peer_activation(&mut self, waiter: WaiterId) -> bool {
|
879 + | let Some(route) = self.peer_route.as_ref().map(|peer| peer.route.id()) else {
|
880 + | return false;
|
881 + | };
|
882 + | if self
|
883 + | .peer_route
|
884 + | .as_ref()
|
885 + | .and_then(|peer| peer.gate.activating())
|
886 + | != Some(waiter)
|
887 + | {
|
888 + | return false;
|
889 + | }
|
890 + | self.finish_peer_gate(route, waiter)
|
891 + | }
|
892 + |
|
893 + | /// Returns an open peer route when no queued acquisition precedes it.
|
894 + | fn open_peer_route(&self, queued: bool) -> Option<H2Route> {
|
895 + | let peer = self.peer_route.as_ref()?;
|
896 + | (!queued && peer.gate.is_open_for(peer.route.id().generation)).then(|| peer.route.clone())
|
897 + | }
|
898 + |
|
899 + | /// Revalidates a direct peer route after its connection-cell crossing.
|
900 + | fn direct_peer_route_is_current(&self, route: H2RouteId, queued: bool) -> bool {
|
901 + | !queued
|
902 + | && self.peer_route.as_ref().is_some_and(|peer| {
|
903 + | peer.route.id() == route && peer.gate.is_open_for(route.generation)
|
904 + | })
|
905 + | }
|
906 + |
|
907 + | /// Removes one stale exact peer route.
|
908 + | fn clear_peer_route(&mut self, route: H2RouteId) -> bool {
|
909 + | if self.peer_route.as_ref().map(|peer| peer.route.id()) != Some(route) {
|
910 + | return false;
|
911 + | }
|
912 + | self.peer_route = None;
|
913 + | self.assert_consistent();
|
914 + | true
|
915 + | }
|
916 + |
|
917 + | /// Returns all installed generation identities for shutdown.
|
918 + | pub(super) fn generation_ids(&self) -> Vec<H2GenerationId> {
|
919 + | self.generations.keys().copied().collect()
|
920 + | }
|
921 + |
|
922 + | /// Returns the accepting generation's idle deadline.
|
923 + | pub(super) fn nearest_idle_deadline(&self) -> Option<SystemTime> {
|
924 + | self.accepting
|
925 + | .and_then(|generation| self.generations.get(&generation))
|
926 + | .and_then(|record| record.idle_deadline)
|
927 + | }
|
928 + |
|
929 + | /// Returns an accepting generation whose idle deadline elapsed.
|
930 + | pub(super) fn expired(&self, now: SystemTime) -> Option<H2GenerationId> {
|
931 + | let generation = self.accepting?;
|
932 + | let deadline = self.generations.get(&generation)?.idle_deadline?;
|
933 + | (deadline <= now).then_some(generation)
|
934 + | }
|
935 + |
|
936 + | /// Replaces the accepting generation's idle deadline after dispatch.
|
937 + | pub(super) fn reset_idle_deadline(
|
938 + | &mut self,
|
939 + | generation: H2GenerationId,
|
940 + | deadline: Option<SystemTime>,
|
941 + | ) -> bool {
|
942 + | if self.accepting != Some(generation) {
|
943 + | return false;
|
944 + | }
|
945 + | let record = self
|
946 + | .generations
|
947 + | .get_mut(&generation)
|
948 + | .expect("accepting HTTP/2 generation disappeared");
|
949 + | record.idle_deadline = deadline;
|
950 + | self.assert_consistent();
|
951 + | true
|
952 + | }
|
953 + |
|
954 + | /// Detaches a draining generation after its last retained request ends.
|
955 + | #[must_use]
|
956 + | fn remove_finished_drain(&mut self, generation: H2GenerationId) -> Option<H2Generation> {
|
957 + | let remove = self.generations.get(&generation).is_some_and(|record| {
|
958 + | record.residence == H2Residence::Draining
|
959 + | && record.prospective == 0
|
960 + | && record.accepted == 0
|
961 + | });
|
962 + | remove
|
963 + | .then(|| self.generations.remove(&generation))
|
964 + | .flatten()
|
965 + | }
|
966 + |
|
967 + | fn take_flight_id(&mut self) -> H2FlightId {
|
968 + | let value = self.next_flight;
|
969 + | self.next_flight = value
|
970 + | .checked_add(1)
|
971 + | .expect("HTTP/2 flight identity exhausted");
|
972 + | H2FlightId(value)
|
973 + | }
|
974 + |
|
975 + | fn take_generation_id(&mut self) -> H2GenerationId {
|
976 + | let value = self.next_generation;
|
977 + | self.next_generation = value
|
978 + | .checked_add(1)
|
979 + | .expect("HTTP/2 generation identity exhausted");
|
980 + | H2GenerationId(value)
|
981 + | }
|
982 + |
|
983 + | /// Checks flight, generation, residence, and request-count relationships.
|
984 + | pub(super) fn assert_consistent(&self) {
|
985 + | #[cfg(any(debug_assertions, test))]
|
986 + | {
|
987 + | if std::thread::panicking() {
|
988 + | return;
|
989 + | }
|
990 + | let accepting_records = self
|
991 + | .generations
|
992 + | .iter()
|
993 + | .filter(|(_, record)| record.residence == H2Residence::Accepting)
|
994 + | .map(|(generation, _)| *generation)
|
995 + | .collect::<Vec<_>>();
|
996 + | match self.accepting {
|
997 + | Some(generation) => assert_eq!(
|
998 + | vec![generation],
|
999 + | accepting_records,
|
1000 + | "HTTP/2 accepting identity did not match generation residence"
|
1001 + | ),
|
1002 + | None => assert!(
|
1003 + | accepting_records.is_empty(),
|
1004 + | "HTTP/2 accepting generation lacked an accepting identity"
|
1005 + | ),
|
1006 + | }
|
1007 + | assert_eq!(
|
1008 + | self.accepting,
|
1009 + | self.gate.generation(),
|
1010 + | "HTTP/2 generation gate did not name the accepting generation"
|
1011 + | );
|
1012 + | assert!(
|
1013 + | self.flight.is_none() || self.accepting.is_none(),
|
1014 + | "HTTP/2 flight coexisted with an accepting generation"
|
1015 + | );
|
1016 + | if let Some(peer) = &self.peer_route {
|
1017 + | assert!(
|
1018 + | self.accepting.is_none(),
|
1019 + | "local accepting generation retained a peer route"
|
1020 + | );
|
1021 + | assert_eq!(
|
1022 + | Some(peer.route.generation()),
|
1023 + | peer.gate.generation(),
|
1024 + | "peer route gate did not name the advertised generation"
|
1025 + | );
|
1026 + | }
|
1027 + |
|
1028 + | for record in self.generations.values() {
|
1029 + | if record.residence == H2Residence::Draining {
|
1030 + | assert!(
|
1031 + | record.prospective > 0 || record.accepted > 0,
|
1032 + | "empty HTTP/2 draining generation was retained"
|
1033 + | );
|
1034 + | assert!(
|
1035 + | record.pending_waiters.is_empty(),
|
1036 + | "draining HTTP/2 generation retained unserved waiters"
|
1037 + | );
|
1038 + | }
|
1039 + | }
|
1040 + | }
|
1041 + | }
|
1042 + |
|
1043 + | /// Checks transferred waiter identities against the cell's acquisition state.
|
1044 + | #[cfg(any(debug_assertions, test))]
|
1045 + | pub(super) fn assert_pending_waiters(&self, waiters: &AcquisitionQueue) {
|
1046 + | for record in self.generations.values() {
|
1047 + | for waiter in &record.pending_waiters {
|
1048 + | assert!(
|
1049 + | waiters.is_launching_h2_candidate(*waiter),
|
1050 + | "HTTP/2 generation retained a waiter that was no longer launchable"
|
1051 + | );
|
1052 + | }
|
1053 + | }
|
1054 + | }
|
1055 + | }
|
1056 + |
|
1057 + | /// One requesting-cell activation prepared before crossing to its connection cell.
|
1058 + | struct PreparedPeerActivation {
|
1059 + | /// Route validated in the requesting cell before the lock crossing.
|
1060 + | route: H2Route,
|
1061 + | /// Oldest waiter selected for this activation.
|
1062 + | waiter: WaiterId,
|
1063 + | /// Publication cutoff that this waiter must satisfy.
|
1064 + | cutoff: Option<WaiterId>,
|
1065 + | /// Whether acceptance or cancellation must discharge a priority turn.
|
1066 + | gated: bool,
|
1067 + | }
|
1068 + |
|
1069 + | /// State detached when an accepting generation begins draining.
|
1070 + | struct H2CloseTransition {
|
1071 + | /// Connection whose logical close follows cell unlock.
|
1072 + | connection: Arc<ConnectionState>,
|
1073 + | /// Unserved transferred waiters that must re-enter acquisition.
|
1074 + | pending_waiters: BTreeSet<WaiterId>,
|
1075 + | /// Removed record retained until the cell lock is released.
|
1076 + | removed_generation: Option<H2Generation>,
|
1077 + | }
|
1078 + |
|
1079 + | /// Sender and connection cloned while a prospective lease is state-owned.
|
1080 + | struct H2ActivationParts {
|
1081 + | /// Transient sender clone for one activation.
|
1082 + | sender: H2Sender,
|
1083 + | /// Whether the generation has accepted an earlier request.
|
1084 + | reused: bool,
|
1085 + | /// Connection retained through prospective dispatch.
|
1086 + | connection: Arc<ConnectionState>,
|
1087 + | }
|
1088 + |
|
1089 + | /// Values transferred together when an activation begins Hyper dispatch.
|
1090 + | pub(in crate::client::pool) struct H2DispatchParts {
|
1091 + | /// Transient sender clone for one dispatch attempt.
|
1092 + | pub(in crate::client::pool) sender: H2Sender,
|
1093 + | /// Request-send endpoint retained by the request body.
|
1094 + | pub(in crate::client::pool) send_endpoint: H2LeaseEndpoint,
|
1095 + | /// Response-receive endpoint retained by the response body.
|
1096 + | pub(in crate::client::pool) receive_endpoint: H2LeaseEndpoint,
|
1097 + | }
|
1098 + |
|
1099 + | /// Prospective request reservation against one exact generation.
|
1100 + | ///
|
1101 + | /// Dropping the activation before Hyper accepts the request cancels the
|
1102 + | /// prospective reservation. Acceptance creates endpoint guards and transfers
|
1103 + | /// dispatch accounting to the response lifecycle.
|
1104 + | pub(in crate::client::pool) struct H2Activation {
|
1105 + | /// Cell that owns the selected generation.
|
1106 + | cell: Arc<OriginCell>,
|
1107 + | /// Partition issuing this request.
|
1108 + | request_partition: PartitionId,
|
1109 + | /// Exact generation that owns the prospective count.
|
1110 + | generation: H2GenerationId,
|
1111 + | /// Transient sender transferred at most once into dispatch.
|
1112 + | sender: Option<H2Sender>,
|
1113 + | /// Whether the generation has accepted an earlier request.
|
1114 + | reused: bool,
|
1115 + | /// Connection retained until acceptance or cancellation.
|
1116 + | connection: Arc<ConnectionState>,
|
1117 + | /// Shared state for the request's send and receive endpoints.
|
1118 + | lease: Arc<H2LeaseCore>,
|
1119 + | /// Requesting-cell priority discharged only at acceptance or cancellation.
|
1120 + | gate: Option<H2GateToken>,
|
1121 + | /// Whether drop must cancel the prospective generation count.
|
1122 + | active: bool,
|
1123 + | }
|
1124 + |
|
1125 + | impl std::fmt::Debug for H2Activation {
|
1126 + | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
1127 + | f.debug_struct("H2Activation")
|
1128 + | .field("generation", &self.generation)
|
1129 + | .field("connection_id", &self.connection.id())
|
1130 + | .field("active", &self.active)
|
1131 + | .finish()
|
1132 + | }
|
1133 + | }
|
1134 + |
|
1135 + | impl H2Activation {
|
1136 + | /// Builds an activation after the generation records its prospective lease.
|
1137 + | fn new(
|
1138 + | cell: Arc<OriginCell>,
|
1139 + | generation: H2GenerationId,
|
1140 + | parts: H2ActivationParts,
|
1141 + | gate: Option<H2GateToken>,
|
1142 + | ) -> Self {
|
1143 + | let request_partition = cell.id().partition();
|
1144 + | let lease = Arc::new(H2LeaseCore {
|
1145 + | cell: Weak::from_arc(&cell),
|
1146 + | generation,
|
1147 + | state: Mutex::new(H2LeaseState::Prospective {
|
1148 + | send_complete: false,
|
1149 + | receive_complete: false,
|
1150 + | }),
|
1151 + | });
|
1152 + | Self {
|
1153 + | cell,
|
1154 + | request_partition,
|
1155 + | generation,
|
1156 + | sender: Some(parts.sender),
|
1157 + | reused: parts.reused,
|
1158 + | connection: parts.connection,
|
1159 + | lease,
|
1160 + | gate,
|
1161 + | active: true,
|
1162 + | }
|
1163 + | }
|
1164 + |
|
1165 + | /// Returns the exact selected generation.
|
1166 + | #[cfg(test)]
|
1167 + | pub(in crate::client::pool) fn generation(&self) -> H2GenerationId {
|
1168 + | self.generation
|
1169 + | }
|
1170 + |
|
1171 + | /// Returns whether Hyper previously accepted a request on this generation.
|
1172 + | pub(in crate::client::pool) fn is_reused(&self) -> bool {
|
1173 + | self.reused
|
1174 + | }
|
1175 + |
|
1176 + | /// Returns the selected protocol-neutral connection.
|
1177 + | pub(in crate::client::pool) fn connection(&self) -> &Arc<ConnectionState> {
|
1178 + | &self.connection
|
1179 + | }
|
1180 + |
|
1181 + | /// Transfers the sender and both lease endpoints into one dispatch attempt.
|
1182 + | ///
|
1183 + | /// # Panics
|
1184 + | ///
|
1185 + | /// Panics if dispatch parts were already taken.
|
1186 + | pub(in crate::client::pool) fn take_dispatch_parts(&mut self) -> H2DispatchParts {
|
1187 + | let sender = self
|
1188 + | .sender
|
1189 + | .take()
|
1190 + | .expect("HTTP/2 activation dispatch parts already taken");
|
1191 + | let trace = H2EndpointTrace::new(self);
|
1192 + | H2DispatchParts {
|
1193 + | sender,
|
1194 + | send_endpoint: H2LeaseEndpoint::new(
|
1195 + | self.lease.clone(),
|
1196 + | H2Endpoint::Send,
|
1197 + | Some(trace.clone()),
|
1198 + | ),
|
1199 + | receive_endpoint: H2LeaseEndpoint::new(
|
1200 + | self.lease.clone(),
|
1201 + | H2Endpoint::Receive,
|
1202 + | Some(trace),
|
1203 + | ),
|
1204 + | }
|
1205 + | }
|
1206 + |
|
1207 + | /// Retains a requesting-cell route opportunity until acceptance or cancellation.
|
1208 + | fn attach_peer_gate(
|
1209 + | &mut self,
|
1210 + | requesting_cell: &Arc<OriginCell>,
|
1211 + | route: H2RouteId,
|
1212 + | waiter: WaiterId,
|
1213 + | ) {
|
1214 + | debug_assert_eq!(
|
1215 + | self.request_partition,
|
1216 + | requesting_cell.id().partition(),
|
1217 + | "peer HTTP/2 activation changed requesting partition"
|
1218 + | );
|
1219 + | assert!(
|
1220 + | self.gate.is_none(),
|
1221 + | "HTTP/2 activation acquired two requesting-cell gates"
|
1222 + | );
|
1223 + | self.gate = Some(H2GateToken::peer(requesting_cell, route, waiter));
|
1224 + | }
|
1225 + |
|
1226 + | /// Returns close authority for the connection-owning generation.
|
1227 + | pub(in crate::client::pool) fn close_handle(&self) -> H2CloseHandle {
|
1228 + | H2CloseHandle::new(&self.cell, self.generation)
|
1229 + | }
|
1230 + |
|
1231 + | /// Converts the prospective reservation after Hyper accepts the request.
|
1232 + | ///
|
1233 + | /// # Panics
|
1234 + | ///
|
1235 + | /// Panics if dispatch parts were not transferred or the prospective
|
1236 + | /// reservation disappeared before acceptance.
|
1237 + | pub(in crate::client::pool) fn accept(mut self, dispatch: DispatchGuard) {
|
1238 + | assert!(
|
1239 + | self.sender.is_none(),
|
1240 + | "HTTP/2 activation accepted before dispatch parts were taken"
|
1241 + | );
|
1242 + | assert!(
|
1243 + | OriginCell::accept_h2_activation(&self.cell, self.generation),
|
1244 + | "prospective HTTP/2 activation disappeared before acceptance"
|
1245 + | );
|
1246 + | self.active = false;
|
1247 + | self.lease.accept(dispatch);
|
1248 + | if let Some(gate) = self.gate.take() {
|
1249 + | gate.finish();
|
1250 + | }
|
1251 + | tracing::trace!(
|
1252 + | connection_id = %self.connection.id(),
|
1253 + | request_partition = ?self.request_partition,
|
1254 + | connection_partition = ?self.connection.owner_partition(),
|
1255 + | origin_scheme = %self.connection.info().origin().scheme(),
|
1256 + | origin_host = self.connection.info().origin().host(),
|
1257 + | origin_port = ?self.connection.info().origin().port(),
|
1258 + | h2_generation = ?self.generation,
|
1259 + | "HTTP/2 request accepted"
|
1260 + | );
|
1261 + | }
|
1262 + | }
|
1263 + |
|
1264 + | impl Drop for H2Activation {
|
1265 + | fn drop(&mut self) {
|
1266 + | if self.active {
|
1267 + | OriginCell::cancel_h2_activation(&self.cell, self.generation);
|
1268 + | self.lease.cancel();
|
1269 + | if let Some(gate) = self.gate.take() {
|
1270 + | gate.finish();
|
1271 + | }
|
1272 + | tracing::trace!(
|
1273 + | connection_id = %self.connection.id(),
|
1274 + | request_partition = ?self.request_partition,
|
1275 + | connection_partition = ?self.connection.owner_partition(),
|
1276 + | origin_scheme = %self.connection.info().origin().scheme(),
|
1277 + | origin_host = self.connection.info().origin().host(),
|
1278 + | origin_port = ?self.connection.info().origin().port(),
|
1279 + | h2_generation = ?self.generation,
|
1280 + | "HTTP/2 activation cancelled before request acceptance"
|
1281 + | );
|
1282 + | }
|
1283 + | }
|
1284 + | }
|
1285 + |
|
1286 + | /// One activation opportunity retained until Hyper acceptance or cancellation.
|
1287 + | struct H2GateToken {
|
1288 + | /// Requesting cell whose gate owns this turn.
|
1289 + | cell: Weak<OriginCell>,
|
1290 + | /// Waiter that received the activation opportunity.
|
1291 + | waiter: WaiterId,
|
1292 + | /// Local-generation or peer-route gate identity.
|
1293 + | kind: H2GateKind,
|
1294 + | }
|
1295 + |
|
1296 + | /// Cell transition run when an activation accepts or cancels.
|
1297 + | enum H2GateKind {
|
1298 + | /// Gate attached to the connection cell's local generation.
|
1299 + | Local { generation: H2GenerationId },
|
1300 + | /// Gate attached to a requesting cell's peer route.
|
1301 + | Peer { route: H2RouteId },
|
1302 + | }
|
1303 + |
|
1304 + | impl H2GateToken {
|
1305 + | fn local(cell: &Arc<OriginCell>, generation: H2GenerationId, waiter: WaiterId) -> Self {
|
1306 + | Self {
|
1307 + | cell: Weak::from_arc(cell),
|
1308 + | waiter,
|
1309 + | kind: H2GateKind::Local { generation },
|
1310 + | }
|
1311 + | }
|
1312 + |
|
1313 + | fn peer(cell: &Arc<OriginCell>, route: H2RouteId, waiter: WaiterId) -> Self {
|
1314 + | Self {
|
1315 + | cell: Weak::from_arc(cell),
|
1316 + | waiter,
|
1317 + | kind: H2GateKind::Peer { route },
|
1318 + | }
|
1319 + | }
|
1320 + |
|
1321 + | fn finish(self) {
|
1322 + | if let Some(cell) = self.cell.upgrade() {
|
1323 + | match self.kind {
|
1324 + | H2GateKind::Local { generation } => {
|
1325 + | OriginCell::finish_h2_gate(&cell, generation, self.waiter)
|
1326 + | }
|
1327 + | H2GateKind::Peer { route } => {
|
1328 + | OriginCell::finish_peer_h2_gate(&cell, route, self.waiter)
|
1329 + | }
|
1330 + | }
|
1331 + | }
|
1332 + | }
|
1333 + | }
|
1334 + |
|
1335 + | /// One terminal side of an accepted HTTP/2 request lease.
|
1336 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
1337 + | enum H2Endpoint {
|
1338 + | /// Request-body upload lifetime.
|
1339 + | Send,
|
1340 + | /// Response future and response-body lifetime.
|
1341 + | Receive,
|
1342 + | }
|
1343 + |
|
1344 + | /// Shared request-lease phase and endpoint bits.
|
1345 + | enum H2LeaseState {
|
1346 + | /// Generation count reserved before Hyper accepts the request.
|
1347 + | Prospective {
|
1348 + | /// Whether upload ownership already terminated.
|
1349 + | send_complete: bool,
|
1350 + | /// Whether response ownership already terminated.
|
1351 + | receive_complete: bool,
|
1352 + | },
|
1353 + | /// Hyper accepted the request and connection dispatch is retained.
|
1354 + | Accepted {
|
1355 + | /// Whether upload ownership already terminated.
|
1356 + | send_complete: bool,
|
1357 + | /// Whether response ownership already terminated.
|
1358 + | receive_complete: bool,
|
1359 + | /// Accepted-dispatch accounting released with the second endpoint.
|
1360 + | dispatch: DispatchGuard,
|
1361 + | },
|
1362 + | /// Both endpoints terminated or prospective dispatch was cancelled.
|
1363 + | Complete,
|
1364 + | }
|
1365 + |
|
1366 + | /// Shared endpoint state for one prospective or accepted request.
|
1367 + | struct H2LeaseCore {
|
1368 + | /// Connection cell updated after both accepted endpoints terminate.
|
1369 + | cell: Weak<OriginCell>,
|
1370 + | /// Exact generation whose request count this lease owns.
|
1371 + | generation: H2GenerationId,
|
1372 + | /// Endpoint phase; never held while locking the cell or connection state.
|
1373 + | state: Mutex<H2LeaseState>,
|
1374 + | }
|
1375 + |
|
1376 + | impl H2LeaseCore {
|
1377 + | /// Converts prospective endpoint state to accepted state.
|
1378 + | fn accept(&self, dispatch: DispatchGuard) {
|
1379 + | let mut dispatch = Some(dispatch);
|
1380 + | let complete = {
|
1381 + | let mut state = self.state.lock();
|
1382 + | let (send_complete, receive_complete) = match &*state {
|
1383 + | H2LeaseState::Prospective {
|
1384 + | send_complete,
|
1385 + | receive_complete,
|
1386 + | } => (*send_complete, *receive_complete),
|
1387 + | H2LeaseState::Accepted { .. } | H2LeaseState::Complete => {
|
1388 + | drop(state);
|
1389 + | panic!("HTTP/2 request lease accepted outside prospective state");
|
1390 + | }
|
1391 + | };
|
1392 + | if send_complete && receive_complete {
|
1393 + | *state = H2LeaseState::Complete;
|
1394 + | true
|
1395 + | } else {
|
1396 + | *state = H2LeaseState::Accepted {
|
1397 + | send_complete,
|
1398 + | receive_complete,
|
1399 + | dispatch: dispatch
|
1400 + | .take()
|
1401 + | .expect("HTTP/2 dispatch guard disappeared before acceptance"),
|
1402 + | };
|
1403 + | false
|
1404 + | }
|
1405 + | };
|
1406 + | drop(dispatch);
|
1407 + | if complete {
|
1408 + | self.complete_generation();
|
1409 + | }
|
1410 + | }
|
1411 + |
|
1412 + | /// Cancels endpoint state after the prospective reservation ends.
|
1413 + | fn cancel(&self) {
|
1414 + | let mut state = self.state.lock();
|
1415 + | if matches!(*state, H2LeaseState::Prospective { .. }) {
|
1416 + | *state = H2LeaseState::Complete;
|
1417 + | }
|
1418 + | }
|
1419 + |
|
1420 + | /// Marks one endpoint terminal and releases the accepted lease on the second.
|
1421 + | fn complete_endpoint(&self, endpoint: H2Endpoint) -> bool {
|
1422 + | let dispatch = {
|
1423 + | let mut state = self.state.lock();
|
1424 + | match &mut *state {
|
1425 + | H2LeaseState::Prospective {
|
1426 + | send_complete,
|
1427 + | receive_complete,
|
1428 + | } => {
|
1429 + | match endpoint {
|
1430 + | H2Endpoint::Send => *send_complete = true,
|
1431 + | H2Endpoint::Receive => *receive_complete = true,
|
1432 + | }
|
1433 + | None
|
1434 + | }
|
1435 + | H2LeaseState::Accepted {
|
1436 + | send_complete,
|
1437 + | receive_complete,
|
1438 + | ..
|
1439 + | } => {
|
1440 + | match endpoint {
|
1441 + | H2Endpoint::Send => *send_complete = true,
|
1442 + | H2Endpoint::Receive => *receive_complete = true,
|
1443 + | }
|
1444 + | if *send_complete && *receive_complete {
|
1445 + | let previous = std::mem::replace(&mut *state, H2LeaseState::Complete);
|
1446 + | let H2LeaseState::Accepted { dispatch, .. } = previous else {
|
1447 + | unreachable!("completed HTTP/2 lease changed state under its lock");
|
1448 + | };
|
1449 + | Some(dispatch)
|
1450 + | } else {
|
1451 + | None
|
1452 + | }
|
1453 + | }
|
1454 + | H2LeaseState::Complete => None,
|
1455 + | }
|
1456 + | };
|
1457 + | let request_complete = dispatch.is_some();
|
1458 + | if let Some(dispatch) = dispatch {
|
1459 + | // Dispatch completion takes the connection lifecycle lock. Keep it
|
1460 + | // outside the request-lease lock so endpoint completion cannot
|
1461 + | // nest pool synchronization.
|
1462 + | drop(dispatch);
|
1463 + | self.complete_generation();
|
1464 + | }
|
1465 + | request_complete
|
1466 + | }
|
1467 + |
|
1468 + | /// Releases the generation count after the lease lock is released.
|
1469 + | fn complete_generation(&self) {
|
1470 + | if let Some(cell) = self.cell.upgrade() {
|
1471 + | OriginCell::complete_h2_request(&cell, self.generation);
|
1472 + | }
|
1473 + | }
|
1474 + | }
|
1475 + |
|
1476 + | /// Structured identity retained by one request-lease endpoint.
|
1477 + | #[derive(Clone)]
|
1478 + | struct H2EndpointTrace {
|
1479 + | /// Partition that issued the request.
|
1480 + | request_partition: PartitionId,
|
1481 + | /// Stable connection identity and origin metadata.
|
1482 + | connection: Arc<ConnectionInfo>,
|
1483 + | /// Generation that owns the endpoint.
|
1484 + | generation: H2GenerationId,
|
1485 + | }
|
1486 + |
|
1487 + | impl H2EndpointTrace {
|
1488 + | fn new(activation: &H2Activation) -> Self {
|
1489 + | Self {
|
1490 + | request_partition: activation.request_partition,
|
1491 + | connection: activation.connection.info().clone(),
|
1492 + | generation: activation.generation,
|
1493 + | }
|
1494 + | }
|
1495 + | }
|
1496 + |
|
1497 + | /// Linear guard for one request-lease endpoint.
|
1498 + | pub(in crate::client::pool) struct H2LeaseEndpoint {
|
1499 + | /// Shared request-lease state.
|
1500 + | core: Arc<H2LeaseCore>,
|
1501 + | /// Send or receive side owned by this guard.
|
1502 + | endpoint: H2Endpoint,
|
1503 + | /// Structured fields emitted on terminal completion.
|
1504 + | trace: Option<H2EndpointTrace>,
|
1505 + | /// Whether drop must finish this endpoint.
|
1506 + | active: bool,
|
1507 + | }
|
1508 + |
|
1509 + | impl H2LeaseEndpoint {
|
1510 + | fn new(core: Arc<H2LeaseCore>, endpoint: H2Endpoint, trace: Option<H2EndpointTrace>) -> Self {
|
1511 + | Self {
|
1512 + | core,
|
1513 + | endpoint,
|
1514 + | trace,
|
1515 + | active: true,
|
1516 + | }
|
1517 + | }
|
1518 + |
|
1519 + | /// Creates a detached endpoint and observation handle for body tests.
|
1520 + | #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
|
1521 + | fn for_test(cell: &Arc<OriginCell>, endpoint: H2Endpoint) -> (Self, H2LeaseProbe) {
|
1522 + | let core = Arc::new(H2LeaseCore {
|
1523 + | cell: Weak::from_arc(cell),
|
1524 + | generation: H2GenerationId(0),
|
1525 + | state: Mutex::new(H2LeaseState::Prospective {
|
1526 + | send_complete: false,
|
1527 + | receive_complete: false,
|
1528 + | }),
|
1529 + | });
|
1530 + | (
|
1531 + | Self::new(core.clone(), endpoint, None),
|
1532 + | H2LeaseProbe { core },
|
1533 + | )
|
1534 + | }
|
1535 + |
|
1536 + | /// Creates a detached send endpoint and observation handle for body tests.
|
1537 + | #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
|
1538 + | pub(in crate::client::pool) fn send_for_test(cell: &Arc<OriginCell>) -> (Self, H2LeaseProbe) {
|
1539 + | Self::for_test(cell, H2Endpoint::Send)
|
1540 + | }
|
1541 + |
|
1542 + | /// Creates a detached receive endpoint and observation handle for body tests.
|
1543 + | #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
|
1544 + | pub(in crate::client::pool) fn receive_for_test(
|
1545 + | cell: &Arc<OriginCell>,
|
1546 + | ) -> (Self, H2LeaseProbe) {
|
1547 + | Self::for_test(cell, H2Endpoint::Receive)
|
1548 + | }
|
1549 + |
|
1550 + | /// Completes this endpoint before dropping the guard.
|
1551 + | pub(in crate::client::pool) fn complete(mut self) {
|
1552 + | self.finish();
|
1553 + | }
|
1554 + |
|
1555 + | fn finish(&mut self) {
|
1556 + | if self.active {
|
1557 + | self.active = false;
|
1558 + | let request_complete = self.core.complete_endpoint(self.endpoint);
|
1559 + | if let Some(trace) = self.trace.take() {
|
1560 + | tracing::trace!(
|
1561 + | connection_id = %trace.connection.id(),
|
1562 + | request_partition = ?trace.request_partition,
|
1563 + | connection_partition = ?trace.connection.owner_partition(),
|
1564 + | origin_scheme = %trace.connection.origin().scheme(),
|
1565 + | origin_host = trace.connection.origin().host(),
|
1566 + | origin_port = ?trace.connection.origin().port(),
|
1567 + | h2_generation = ?trace.generation,
|
1568 + | endpoint = ?self.endpoint,
|
1569 + | request_complete,
|
1570 + | "HTTP/2 request endpoint completed"
|
1571 + | );
|
1572 + | }
|
1573 + | }
|
1574 + | }
|
1575 + | }
|
1576 + |
|
1577 + | impl Drop for H2LeaseEndpoint {
|
1578 + | fn drop(&mut self) {
|
1579 + | self.finish();
|
1580 + | }
|
1581 + | }
|
1582 + |
|
1583 + | /// Test observation of a request-lease endpoint's production state.
|
1584 + | #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
|
1585 + | pub(in crate::client::pool) struct H2LeaseProbe {
|
1586 + | /// Production lease state observed by the test.
|
1587 + | core: Arc<H2LeaseCore>,
|
1588 + | }
|
1589 + |
|
1590 + | #[cfg(all(test, not(smithy_http_client_loom), feature = "rt-tokio"))]
|
1591 + | impl H2LeaseProbe {
|
1592 + | /// Returns whether the send endpoint reached its terminal transition.
|
1593 + | pub(in crate::client::pool) fn send_complete(&self) -> bool {
|
1594 + | matches!(
|
1595 + | &*self.core.state.lock(),
|
1596 + | H2LeaseState::Prospective {
|
1597 + | send_complete: true,
|
1598 + | ..
|
1599 + | } | H2LeaseState::Accepted {
|
1600 + | send_complete: true,
|
1601 + | ..
|
1602 + | } | H2LeaseState::Complete
|
1603 + | )
|
1604 + | }
|
1605 + |
|
1606 + | /// Returns whether the receive endpoint reached its terminal transition.
|
1607 + | pub(in crate::client::pool) fn receive_complete(&self) -> bool {
|
1608 + | matches!(
|
1609 + | &*self.core.state.lock(),
|
1610 + | H2LeaseState::Prospective {
|
1611 + | receive_complete: true,
|
1612 + | ..
|
1613 + | } | H2LeaseState::Accepted {
|
1614 + | receive_complete: true,
|
1615 + | ..
|
1616 + | } | H2LeaseState::Complete
|
1617 + | )
|
1618 + | }
|
1619 + | }
|
1620 + |
|
1621 + | impl OriginCell {
|
1622 + | /// Reserves a prospective request lease from one exact local generation.
|
1623 + | pub(in crate::client::pool) fn activate_h2(
|
1624 + | cell: &Arc<Self>,
|
1625 + | generation: H2GenerationId,
|
1626 + | ) -> Option<H2Activation> {
|
1627 + | let (parts, advertisement) = {
|
1628 + | let mut state = cell.state.lock();
|
1629 + | let parts = state.h2.activate(generation)?;
|
1630 + | state.assert_consistent();
|
1631 + | let advertisement = state.take_h2_advertisement_update();
|
1632 + | (parts, advertisement)
|
1633 + | };
|
1634 + | Self::publish_h2_advertisement(cell, advertisement);
|
1635 + | Some(H2Activation::new(cell.clone(), generation, parts, None))
|
1636 + | }
|
1637 + |
|
1638 + | /// Converts a prospective activation to an accepted generation lease.
|
1639 + | fn accept_h2_activation(cell: &Arc<Self>, generation: H2GenerationId) -> bool {
|
1640 + | let deadline = cell.idle_deadline();
|
1641 + | let (accepted, reset_deadline) = {
|
1642 + | let mut state = cell.state.lock();
|
1643 + | let accepted = state.h2.accept(generation);
|
1644 + | let reset_deadline = accepted && state.h2.reset_idle_deadline(generation, deadline);
|
1645 + | state.assert_consistent();
|
1646 + | (accepted, reset_deadline)
|
1647 + | };
|
1648 + | if reset_deadline {
|
1649 + | cell.notify_maintenance(deadline);
|
1650 + | }
|
1651 + | accepted
|
1652 + | }
|
1653 + |
|
1654 + | /// Cancels one prospective activation after sender rejection or task drop.
|
1655 + | fn cancel_h2_activation(cell: &Arc<Self>, generation: H2GenerationId) {
|
1656 + | // Keep the detached generation outside the lock's unwind scope. Its
|
1657 + | // sender drop may wake Hyper's driver.
|
1658 + | let (removed, advertisement) = {
|
1659 + | let mut state = cell.state.lock();
|
1660 + | let removed = state.h2.cancel(generation);
|
1661 + | state.assert_consistent();
|
1662 + | let advertisement = state.take_h2_advertisement_update();
|
1663 + | (removed, advertisement)
|
1664 + | };
|
1665 + | drop(removed);
|
1666 + | Self::publish_h2_advertisement(cell, advertisement);
|
1667 + | Self::service_h2_waiters(cell);
|
1668 + | }
|
1669 + |
|
1670 + | /// Releases one accepted lease after its second endpoint terminates.
|
1671 + | fn complete_h2_request(cell: &Arc<Self>, generation: H2GenerationId) {
|
1672 + | // Keep the detached generation outside the lock's unwind scope. Its
|
1673 + | // sender drop may wake Hyper's driver.
|
1674 + | let (removed, advertisement) = {
|
1675 + | let mut state = cell.state.lock();
|
1676 + | let removed = state.h2.complete_request(generation);
|
1677 + | state.assert_consistent();
|
1678 + | let advertisement = state.take_h2_advertisement_update();
|
1679 + | (removed, advertisement)
|
1680 + | };
|
1681 + | drop(removed);
|
1682 + | Self::publish_h2_advertisement(cell, advertisement);
|
1683 + | }
|
1684 + |
|
1685 + | /// Offers one local activation while preserving the generation cutoff.
|
1686 + | fn service_h2_gate_locked(
|
1687 + | cell: &Arc<Self>,
|
1688 + | state: &mut CellState,
|
1689 + | returned_event: &mut Option<AcquisitionEvent>,
|
1690 + | ) -> Option<WaiterInstall> {
|
1691 + | let generation = state.h2.accepting()?;
|
1692 + | let has_prioritized = state
|
1693 + | .h2
|
1694 + | .priority_cutoff()
|
1695 + | .is_some_and(|cutoff| state.waiters.has_h2_candidate_through(cutoff));
|
1696 + | let cutoff = match state.h2.next_gate_turn(has_prioritized) {
|
1697 + | GateTurn::Unavailable => return None,
|
1698 + | GateTurn::Open => None,
|
1699 + | GateTurn::Through(cutoff) => Some(cutoff),
|
1700 + | };
|
1701 + | state.waiters.oldest_h2_candidate()?;
|
1702 + |
|
1703 + | let h2 = &mut state.h2;
|
1704 + | let waiters = &mut state.waiters;
|
1705 + | let mut install = waiters.install_h2(
|
1706 + | cutoff,
|
1707 + | |waiter| {
|
1708 + | let parts = h2
|
1709 + | .activate(generation)
|
1710 + | .expect("accepting HTTP/2 generation could not create a gated activation");
|
1711 + | let gated = h2.begin_gate_activation(waiter);
|
1712 + | AcquisitionResult::H2(H2Activation::new(
|
1713 + | cell.clone(),
|
1714 + | generation,
|
1715 + | parts,
|
1716 + | gated.then(|| H2GateToken::local(cell, generation, waiter)),
|
1717 + | ))
|
1718 + | },
|
1719 + | &cell.eligibility_group,
|
1720 + | );
|
1721 + | *returned_event = install.returned_event.take();
|
1722 + | install.demand_updates = state.publishable_demand_updates(install.demand_updates);
|
1723 + | state.assert_consistent();
|
1724 + | install.waiter.is_some().then_some(install)
|
1725 + | }
|
1726 + |
|
1727 + | /// Runs publication, fallback, and wake work after the cell lock is released.
|
1728 + | fn finish_h2_install(cell: &Arc<Self>, install: Option<WaiterInstall>) {
|
1729 + | let Some(install) = install else {
|
1730 + | return;
|
1731 + | };
|
1732 + | if let Some(admission) = &cell.admission {
|
1733 + | for snapshot in install.demand_updates.into_iter().flatten() {
|
1734 + | super::super::admission::OriginAdmission::publish_demand(
|
1735 + | admission,
|
1736 + | cell.id.partition(),
|
1737 + | snapshot,
|
1738 + | );
|
1739 + | }
|
1740 + | }
|
1741 + | drop(install.returned_event);
|
1742 + | if let Some(waker) = install.waker {
|
1743 + | waker.wake();
|
1744 + | }
|
1745 + | }
|
1746 + |
|
1747 + | /// Publishes one complete local-generation report after cell unlock.
|
1748 + | fn publish_h2_advertisement(
|
1749 + | cell: &Arc<Self>,
|
1750 + | snapshot: Option<super::super::admission::H2AdvertisementSnapshot>,
|
1751 + | ) {
|
1752 + | if let (Some(admission), Some(snapshot)) = (&cell.admission, snapshot) {
|
1753 + | super::super::admission::OriginAdmission::update_h2_advertisement(
|
1754 + | admission,
|
1755 + | cell.id.partition(),
|
1756 + | cell.eligibility_group.clone(),
|
1757 + | snapshot,
|
1758 + | );
|
1759 + | }
|
1760 + | }
|
1761 + |
|
1762 + | /// Publishes the current local demand after its last H2 route disappears.
|
1763 + | fn publish_current_demand(
|
1764 + | cell: &Arc<Self>,
|
1765 + | snapshot: Option<super::super::admission::DemandSnapshot>,
|
1766 + | ) {
|
1767 + | if let (Some(admission), Some(snapshot)) = (&cell.admission, snapshot) {
|
1768 + | super::super::admission::OriginAdmission::publish_demand(
|
1769 + | admission,
|
1770 + | cell.id.partition(),
|
1771 + | snapshot,
|
1772 + | );
|
1773 + | }
|
1774 + | }
|
1775 + |
|
1776 + | /// Advances a generation gate after one activation accepts or cancels.
|
1777 + | fn finish_h2_gate(cell: &Arc<Self>, generation: H2GenerationId, waiter: WaiterId) {
|
1778 + | let mut returned_event = None;
|
1779 + | let (install, advertisement) = {
|
1780 + | let mut state = cell.state.lock();
|
1781 + | if !state.h2.finish_gate_activation(generation, waiter) {
|
1782 + | return;
|
1783 + | }
|
1784 + | let install = Self::service_h2_gate_locked(cell, &mut state, &mut returned_event);
|
1785 + | state.assert_consistent();
|
1786 + | let advertisement = state.take_h2_advertisement_update();
|
1787 + | (install, advertisement)
|
1788 + | };
|
1789 + | drop(returned_event);
|
1790 + | Self::publish_h2_advertisement(cell, advertisement);
|
1791 + | Self::finish_h2_install(cell, install);
|
1792 + | }
|
1793 + |
|
1794 + | /// Offers the accepting local generation to one queued acquisition.
|
1795 + | pub(in crate::client::pool) fn service_h2_waiters(cell: &Arc<Self>) {
|
1796 + | let mut returned_event = None;
|
1797 + | let (install, advertisement) = {
|
1798 + | let mut state = cell.state.lock();
|
1799 + | let install = Self::service_h2_gate_locked(cell, &mut state, &mut returned_event);
|
1800 + | state.assert_consistent();
|
1801 + | let advertisement = state.take_h2_advertisement_update();
|
1802 + | (install, advertisement)
|
1803 + | };
|
1804 + | drop(returned_event);
|
1805 + | Self::publish_h2_advertisement(cell, advertisement);
|
1806 + | Self::finish_h2_install(cell, install);
|
1807 + | }
|
1808 + |
|
1809 + | /// Returns whether one exact connection generation still accepts activations.
|
1810 + | pub(in crate::client::pool) fn h2_generation_is_accepting(
|
1811 + | cell: &Arc<Self>,
|
1812 + | generation: H2GenerationId,
|
1813 + | ) -> bool {
|
1814 + | cell.state.lock().h2.is_accepting(generation)
|
1815 + | }
|
1816 + |
|
1817 + | /// Installs requesting-cell visibility for one peer generation publication.
|
1818 + | ///
|
1819 + | /// The named local demand remains queued behind the route gate. Advancing
|
1820 + | /// its snapshot version makes admission acknowledgement authoritative
|
1821 + | /// without losing the demand if this exact route later becomes stale.
|
1822 + | pub(in crate::client::pool) fn install_h2_route(
|
1823 + | cell: &Arc<Self>,
|
1824 + | route: H2Route,
|
1825 + | group: &super::super::partition::EligibilityGroup,
|
1826 + | demand: super::super::admission::DemandId,
|
1827 + | ) -> bool {
|
1828 + | if &cell.eligibility_group != group || route.connection_partition() == cell.id.partition() {
|
1829 + | return false;
|
1830 + | }
|
1831 + | let mut state = cell.state.lock();
|
1832 + | if !state.waiters.suppress_published_demand(demand) {
|
1833 + | return false;
|
1834 + | }
|
1835 + | let cutoff = state.waiters.publication_cutoff();
|
1836 + | state.h2.install_peer_route(route, cutoff);
|
1837 + | state.assert_consistent();
|
1838 + | true
|
1839 + | }
|
1840 + |
|
1841 + | /// Activates queued requests through the currently visible peer route.
|
1842 + | ///
|
1843 + | /// Preparing the requesting-cell opportunity, activating the connection
|
1844 + | /// generation, and committing the result each use a separate lock scope.
|
1845 + | pub(in crate::client::pool) fn service_peer_h2_waiters(cell: &Arc<Self>) {
|
1846 + | loop {
|
1847 + | let prepared = {
|
1848 + | let mut state = cell.state.lock();
|
1849 + | let CellState { h2, waiters, .. } = &mut *state;
|
1850 + | let prepared = h2.prepare_peer_activation(waiters);
|
1851 + | state.assert_consistent();
|
1852 + | prepared
|
1853 + | };
|
1854 + | let Some(prepared) = prepared else {
|
1855 + | return;
|
1856 + | };
|
1857 + | let route_id = prepared.route.id();
|
1858 + | let Some(activation) = prepared.route.activate(cell.id.partition()) else {
|
1859 + | let snapshot = {
|
1860 + | let mut state = cell.state.lock();
|
1861 + | if !state.h2.clear_peer_route(route_id) {
|
1862 + | return;
|
1863 + | }
|
1864 + | let snapshot = state
|
1865 + | .waiters
|
1866 + | .current_demand_snapshot(&cell.eligibility_group);
|
1867 + | state.assert_consistent();
|
1868 + | snapshot
|
1869 + | };
|
1870 + | if let (Some(admission), Some(snapshot)) = (&cell.admission, snapshot) {
|
1871 + | super::super::admission::OriginAdmission::publish_demand(
|
1872 + | admission,
|
1873 + | cell.id.partition(),
|
1874 + | snapshot,
|
1875 + | );
|
1876 + | }
|
1877 + | return;
|
1878 + | };
|
1879 + |
|
1880 + | let mut activation = Some(activation);
|
1881 + | let mut returned_event = None;
|
1882 + | let install = {
|
1883 + | let mut state = cell.state.lock();
|
1884 + | let current = state
|
1885 + | .h2
|
1886 + | .peer_activation_is_current(&prepared, &state.waiters);
|
1887 + | if !current {
|
1888 + | state.h2.finish_peer_crossing(route_id, prepared.waiter);
|
1889 + | if prepared.gated {
|
1890 + | state.h2.finish_peer_gate(route_id, prepared.waiter);
|
1891 + | }
|
1892 + | state.assert_consistent();
|
1893 + | None
|
1894 + | } else {
|
1895 + | if prepared.gated {
|
1896 + | activation
|
1897 + | .as_mut()
|
1898 + | .expect("peer HTTP/2 activation disappeared")
|
1899 + | .attach_peer_gate(cell, route_id, prepared.waiter);
|
1900 + | }
|
1901 + | let mut install = state.waiters.install_h2(
|
1902 + | prepared.cutoff,
|
1903 + | |_| {
|
1904 + | AcquisitionResult::H2(
|
1905 + | activation
|
1906 + | .take()
|
1907 + | .expect("peer HTTP/2 activation was installed twice"),
|
1908 + | )
|
1909 + | },
|
1910 + | &cell.eligibility_group,
|
1911 + | );
|
1912 + | returned_event = install.returned_event.take();
|
1913 + | install.demand_updates =
|
1914 + | state.publishable_demand_updates(install.demand_updates);
|
1915 + | state.h2.finish_peer_crossing(route_id, prepared.waiter);
|
1916 + | if install.waiter.is_none() && prepared.gated {
|
1917 + | state.h2.finish_peer_gate(route_id, prepared.waiter);
|
1918 + | }
|
1919 + | state.assert_consistent();
|
1920 + | install.waiter.is_some().then_some(install)
|
1921 + | }
|
1922 + | };
|
1923 + | drop(activation);
|
1924 + | drop(returned_event);
|
1925 + | if install.is_some() {
|
1926 + | Self::finish_h2_install(cell, install);
|
1927 + | return;
|
1928 + | }
|
1929 + | }
|
1930 + | }
|
1931 + |
|
1932 + | /// Advances an exact peer-route gate after acceptance or cancellation.
|
1933 + | fn finish_peer_h2_gate(cell: &Arc<Self>, route: H2RouteId, waiter: WaiterId) {
|
1934 + | let finished = {
|
1935 + | let mut state = cell.state.lock();
|
1936 + | let finished = state.h2.finish_peer_gate(route, waiter);
|
1937 + | state.assert_consistent();
|
1938 + | finished
|
1939 + | };
|
1940 + | if finished {
|
1941 + | Self::service_peer_h2_waiters(cell);
|
1942 + | }
|
1943 + | }
|
1944 + | }
|
1945 + |
|
1946 + | /// Generation-specific close authority that does not retain the cell.
|
1947 + | #[derive(Clone, Debug)]
|
1948 + | pub(in crate::client::pool) struct H2CloseHandle {
|
1949 + | /// Cell weak reference so driver lifetime does not retain the pool.
|
1950 + | cell: Weak<OriginCell>,
|
1951 + | /// Exact generation this handle may close.
|
1952 + | generation: H2GenerationId,
|
1953 + | }
|
1954 + |
|
1955 + | impl H2CloseHandle {
|
1956 + | /// Creates close authority for one installed generation.
|
1957 + | pub(in crate::client::pool) fn new(cell: &Arc<OriginCell>, generation: H2GenerationId) -> Self {
|
1958 + | Self {
|
1959 + | cell: Weak::from_arc(cell),
|
1960 + | generation,
|
1961 + | }
|
1962 + | }
|
1963 + |
|
1964 + | /// Begins drain when the cell still contains this generation.
|
1965 + | pub(in crate::client::pool) fn close(
|
1966 + | &self,
|
1967 + | reason: super::super::connection::CloseReason,
|
1968 + | ) -> bool {
|
1969 + | self.cell
|
1970 + | .upgrade()
|
1971 + | .is_some_and(|cell| OriginCell::close_h2(&cell, self.generation, reason))
|
1972 + | }
|
1973 + | }
|
1974 + |
|
1975 + | /// Closes an H2 generation if its owner-runtime driver ends or is dropped.
|
1976 + | pub(in crate::client::pool) struct H2DriverGuard {
|
1977 + | /// Exact-generation close authority.
|
1978 + | close: H2CloseHandle,
|
1979 + | /// Whether drop must report owner-runtime shutdown.
|
1980 + | active: bool,
|
1981 + | }
|
1982 + |
|
1983 + | impl H2DriverGuard {
|
1984 + | /// Arms generation cleanup before driver submission.
|
1985 + | pub(in crate::client::pool) fn new(close: H2CloseHandle) -> Self {
|
1986 + | Self {
|
1987 + | close,
|
1988 + | active: true,
|
1989 + | }
|
1990 + | }
|
1991 + |
|
1992 + | /// Records ordinary driver completion.
|
1993 + | pub(in crate::client::pool) fn protocol_closed(mut self) {
|
1994 + | self.active = false;
|
1995 + | self.close
|
1996 + | .close(super::super::connection::CloseReason::ProtocolClosed);
|
1997 + | }
|
1998 + | }
|
1999 + |
|
2000 + | impl Drop for H2DriverGuard {
|
2001 + | fn drop(&mut self) {
|
2002 + | if self.active {
|
2003 + | self.close
|
2004 + | .close(super::super::connection::CloseReason::OwnerRuntimeShutdown);
|
2005 + | }
|
2006 + | }
|
2007 + | }
|
2008 + |
|
2009 + | impl OriginCell {
|
2010 + | /// Selects an accepting local generation before consulting a peer route.
|
2011 + | pub(in crate::client::pool) fn select_h2(cell: &Arc<Self>) -> Option<H2Activation> {
|
2012 + | let local = {
|
2013 + | let mut state = cell.state.lock();
|
2014 + | let queued = state.waiters.has_h2_candidate();
|
2015 + | if state.h2.direct_is_allowed(queued) {
|
2016 + | let generation = state.h2.accepting()?;
|
2017 + | let parts = state.h2.activate(generation)?;
|
2018 + | state.assert_consistent();
|
2019 + | let advertisement = state.take_h2_advertisement_update();
|
2020 + | Some((generation, parts, advertisement))
|
2021 + | } else {
|
2022 + | None
|
2023 + | }
|
2024 + | };
|
2025 + | if let Some((generation, parts, advertisement)) = local {
|
2026 + | Self::publish_h2_advertisement(cell, advertisement);
|
2027 + | return Some(H2Activation::new(cell.clone(), generation, parts, None));
|
2028 + | }
|
2029 + |
|
2030 + | let route = {
|
2031 + | let state = cell.state.lock();
|
2032 + | state.h2.open_peer_route(state.waiters.has_h2_candidate())?
|
2033 + | };
|
2034 + | let route_id = route.id();
|
2035 + | let Some(activation) = route.activate(cell.id.partition()) else {
|
2036 + | Self::clear_stale_peer_route(cell, route_id);
|
2037 + | return None;
|
2038 + | };
|
2039 + | let current = {
|
2040 + | let state = cell.state.lock();
|
2041 + | state
|
2042 + | .h2
|
2043 + | .direct_peer_route_is_current(route_id, state.waiters.has_h2_candidate())
|
2044 + | };
|
2045 + | current.then_some(activation)
|
2046 + | }
|
2047 + |
|
2048 + | /// Removes one exact stale route and republishes the requesting demand.
|
2049 + | fn clear_stale_peer_route(cell: &Arc<Self>, route: H2RouteId) {
|
2050 + | let snapshot = {
|
2051 + | let mut state = cell.state.lock();
|
2052 + | if !state.h2.clear_peer_route(route) {
|
2053 + | return;
|
2054 + | }
|
2055 + | let snapshot = state
|
2056 + | .waiters
|
2057 + | .current_demand_snapshot(&cell.eligibility_group);
|
2058 + | state.assert_consistent();
|
2059 + | snapshot
|
2060 + | };
|
2061 + | Self::publish_current_demand(cell, snapshot);
|
2062 + | }
|
2063 + |
|
2064 + | /// Places a post-ALPN waiter behind an already accepting generation.
|
2065 + | pub(in crate::client::pool) fn join_h2_generation(
|
2066 + | cell: &Arc<Self>,
|
2067 + | waiter: WaiterId,
|
2068 + | generation: H2GenerationId,
|
2069 + | ) -> H2GenerationJoin {
|
2070 + | let mut returned_event = None;
|
2071 + | let (install, advertisement) = {
|
2072 + | let mut state = cell.state.lock();
|
2073 + | if !state.waiters.is_launching_h2_candidate(waiter) {
|
2074 + | return H2GenerationJoin::WaiterCompleted;
|
2075 + | }
|
2076 + | if !state.h2.prioritize_waiter(generation, waiter) {
|
2077 + | return H2GenerationJoin::GenerationChanged;
|
2078 + | }
|
2079 + | let install = Self::service_h2_gate_locked(cell, &mut state, &mut returned_event);
|
2080 + | state.assert_consistent();
|
2081 + | let advertisement = state.take_h2_advertisement_update();
|
2082 + | (install, advertisement)
|
2083 + | };
|
2084 + | drop(returned_event);
|
2085 + | Self::publish_h2_advertisement(cell, advertisement);
|
2086 + | Self::finish_h2_install(cell, install);
|
2087 + | H2GenerationJoin::Joined
|
2088 + | }
|
2089 + |
|
2090 + | /// Atomically selects, joins, or installs the cell's post-ALPN flight.
|
2091 + | pub(in crate::client::pool) fn install_or_join_h2_flight(
|
2092 + | &self,
|
2093 + | waiter: WaiterId,
|
2094 + | ) -> H2FlightInstall {
|
2095 + | let mut state = self.state.lock();
|
2096 + | let result = state.h2.install_or_join_flight(waiter);
|
2097 + | state.assert_consistent();
|
2098 + | result
|
2099 + | }
|
2100 + |
|
2101 + | /// Installs one successful flight as the accepting generation.
|
2102 + | pub(in crate::client::pool) fn complete_h2_flight(
|
2103 + | cell: &Arc<Self>,
|
2104 + | flight: H2FlightId,
|
2105 + | connection: Arc<ConnectionState>,
|
2106 + | sender: H2Sender,
|
2107 + | idle_deadline: Option<SystemTime>,
|
2108 + | ) -> Result<H2GenerationId, (Arc<ConnectionState>, H2Sender)> {
|
2109 + | let mut returned_event = None;
|
2110 + | let (completion, install, advertisement) = {
|
2111 + | let mut state = cell.state.lock();
|
2112 + | let completion = state
|
2113 + | .h2
|
2114 + | .complete_flight(flight, connection, sender, idle_deadline);
|
2115 + | let install = if completion.is_ok() {
|
2116 + | let cutoff = state.waiters.publication_cutoff();
|
2117 + | state.h2.prioritize_through(cutoff);
|
2118 + | Self::service_h2_gate_locked(cell, &mut state, &mut returned_event)
|
2119 + | } else {
|
2120 + | None
|
2121 + | };
|
2122 + | state.assert_consistent();
|
2123 + | let advertisement = state.take_h2_advertisement_update();
|
2124 + | (completion, install, advertisement)
|
2125 + | };
|
2126 + | drop(returned_event);
|
2127 + | Self::publish_h2_advertisement(cell, advertisement);
|
2128 + | Self::finish_h2_install(cell, install);
|
2129 + | if completion.is_ok() {
|
2130 + | cell.notify_maintenance(idle_deadline);
|
2131 + | }
|
2132 + | completion
|
2133 + | }
|
2134 + |
|
2135 + | /// Retires one failed flight and returns its retained participants.
|
2136 + | pub(in crate::client::pool) fn fail_h2_flight(
|
2137 + | &self,
|
2138 + | flight: H2FlightId,
|
2139 + | ) -> Option<Vec<WaiterId>> {
|
2140 + | let mut state = self.state.lock();
|
2141 + | let participants = state.h2.fail_flight(flight);
|
2142 + | state.assert_consistent();
|
2143 + | participants
|
2144 + | }
|
2145 + |
|
2146 + | /// Returns the complete current advertisement for an unlocked crossing.
|
2147 + | pub(in crate::client::pool) fn report_h2_advertisement(
|
2148 + | &self,
|
2149 + | ) -> super::super::admission::H2AdvertisementSnapshot {
|
2150 + | let mut state = self.state.lock();
|
2151 + | let snapshot = state.report_h2_advertisement();
|
2152 + | state.assert_consistent();
|
2153 + | snapshot
|
2154 + | }
|
2155 + |
|
2156 + | /// Reclaims one exact idle generation and reports its resulting availability.
|
2157 + | pub(in crate::client::pool) fn reclaim_idle_h2(
|
2158 + | cell: &Arc<Self>,
|
2159 + | generation: H2GenerationId,
|
2160 + | ) -> (
|
2161 + | super::super::admission::H2AdvertisementSnapshot,
|
2162 + | Option<ConnectionId>,
|
2163 + | ) {
|
2164 + | // A zero-request generation may hold the last capacity-owning
|
2165 + | // connection reference. Keep it outside the cell-lock unwind scope.
|
2166 + | let detached;
|
2167 + | let (advertisement, demand) = {
|
2168 + | let mut state = cell.state.lock();
|
2169 + | detached = state.h2.begin_idle_reclaim(generation);
|
2170 + | let advertisement = state.report_h2_advertisement();
|
2171 + | let demand = detached.as_ref().and_then(|_| {
|
2172 + | state
|
2173 + | .waiters
|
2174 + | .current_demand_snapshot(&cell.eligibility_group)
|
2175 + | });
|
2176 + | state.assert_consistent();
|
2177 + | (advertisement, demand)
|
2178 + | };
|
2179 + | let Some(detached) = detached else {
|
2180 + | return (advertisement, None);
|
2181 + | };
|
2182 + | let H2CloseTransition {
|
2183 + | connection,
|
2184 + | pending_waiters,
|
2185 + | removed_generation,
|
2186 + | } = detached;
|
2187 + | assert!(
|
2188 + | pending_waiters.is_empty(),
|
2189 + | "idle HTTP/2 reclaim detached pending generation waiters"
|
2190 + | );
|
2191 + | let connection_id = connection.id();
|
2192 + | drop(removed_generation);
|
2193 + | Self::publish_current_demand(cell, demand);
|
2194 + | let reclaimed = connection.logical_close(super::super::connection::CloseReason::Reclaimed);
|
2195 + | (advertisement, reclaimed.then_some(connection_id))
|
2196 + | }
|
2197 + |
|
2198 + | /// Moves one exact generation to draining and closes its connection.
|
2199 + | pub(super) fn close_h2(
|
2200 + | cell: &Arc<Self>,
|
2201 + | generation: H2GenerationId,
|
2202 + | reason: super::super::connection::CloseReason,
|
2203 + | ) -> bool {
|
2204 + | // A zero-request generation may hold the last capacity-owning
|
2205 + | // connection reference. Keep it outside the cell-lock unwind scope.
|
2206 + | let detached;
|
2207 + | let (advertisement, demand) = {
|
2208 + | let mut state = cell.state.lock();
|
2209 + | detached = state.h2.begin_close(generation);
|
2210 + | let advertisement = detached
|
2211 + | .as_ref()
|
2212 + | .and_then(|_| state.take_h2_advertisement_update());
|
2213 + | let demand = detached.as_ref().and_then(|_| {
|
2214 + | state
|
2215 + | .waiters
|
2216 + | .current_demand_snapshot(&cell.eligibility_group)
|
2217 + | });
|
2218 + | state.assert_consistent();
|
2219 + | (advertisement, demand)
|
2220 + | };
|
2221 + | let Some(detached) = detached else {
|
2222 + | return false;
|
2223 + | };
|
2224 + | let H2CloseTransition {
|
2225 + | connection,
|
2226 + | pending_waiters,
|
2227 + | removed_generation,
|
2228 + | } = detached;
|
2229 + | drop(removed_generation);
|
2230 + | Self::publish_h2_advertisement(cell, advertisement);
|
2231 + | Self::publish_current_demand(cell, demand);
|
2232 + | for waiter in pending_waiters {
|
2233 + | cell.complete_establishment(waiter, AcquisitionResult::Reacquire);
|
2234 + | }
|
2235 + | connection.logical_close(reason)
|
2236 + | }
|
2237 + |
|
2238 + | /// Returns the exact accepting generation for publication.
|
2239 + | #[cfg(test)]
|
2240 + | pub(in crate::client::pool) fn accepting_h2_generation(&self) -> Option<H2GenerationId> {
|
2241 + | self.state.lock().h2.accepting()
|
2242 + | }
|
2243 + |
|
2244 + | /// Returns the prospective and accepted request counts for one generation.
|
2245 + | #[cfg(test)]
|
2246 + | pub(in crate::client::pool) fn h2_request_counts(
|
2247 + | &self,
|
2248 + | generation: H2GenerationId,
|
2249 + | ) -> Option<(usize, usize)> {
|
2250 + | self.state
|
2251 + | .lock()
|
2252 + | .h2
|
2253 + | .generations
|
2254 + | .get(&generation)
|
2255 + | .map(|record| (record.prospective, record.accepted))
|
2256 + | }
|
2257 + |
|
2258 + | /// Installs an accepting generation without a Hyper handshake.
|
2259 + | #[cfg(test)]
|
2260 + | pub(in crate::client::pool) fn install_h2_for_test(
|
2261 + | cell: &Arc<Self>,
|
2262 + | connection: Arc<ConnectionState>,
|
2263 + | sender_id: u64,
|
2264 + | idle_deadline: Option<SystemTime>,
|
2265 + | ) -> H2GenerationId {
|
2266 + | let (generation, advertisement) = {
|
2267 + | let mut state = cell.state.lock();
|
2268 + | let generation = state.h2.install_generation(
|
2269 + | BTreeSet::new(),
|
2270 + | connection,
|
2271 + | H2Sender::test(sender_id),
|
2272 + | idle_deadline,
|
2273 + | );
|
2274 + | state.assert_consistent();
|
2275 + | let advertisement = state.take_h2_advertisement_update();
|
2276 + | (generation, advertisement)
|
2277 + | };
|
2278 + | Self::publish_h2_advertisement(cell, advertisement);
|
2279 + | cell.notify_maintenance(idle_deadline);
|
2280 + | generation
|
2281 + | }
|
2282 + | }
|
2283 + |
|
2284 + | #[cfg(all(test, not(smithy_http_client_loom)))]
|
2285 + | mod tests {
|
2286 + | use super::*;
|
2287 + | use crate::client::pool::admission::ProtocolRequirement;
|
2288 + | use crate::client::pool::connection::{CloseReason, ConnectionInfo};
|
2289 + | use crate::client::pool::origin::OriginKey;
|
2290 + | use crate::client::pool::partition::EligibilityGroup;
|
2291 + | use aws_smithy_runtime_api::client::connection::ConnectionId;
|
2292 + | use http_1x::uri::Scheme;
|
2293 + |
|
2294 + | fn cell() -> Arc<OriginCell> {
|
2295 + | Arc::new(OriginCell::new(
|
2296 + | PartitionId::from_index(1),
|
2297 + | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
2298 + | EligibilityGroup::Pool,
|
2299 + | None,
|
2300 + | None,
|
2301 + | ))
|
2302 + | }
|
2303 + |
|
2304 + | fn bounded_cell(
|
2305 + | admission: &Arc<crate::client::pool::admission::OriginAdmission>,
|
2306 + | partition: usize,
|
2307 + | ) -> Arc<OriginCell> {
|
2308 + | crate::client::pool::admission::OriginAdmission::register_cell(
|
2309 + | admission,
|
2310 + | Arc::new(OriginCell::new(
|
2311 + | PartitionId::from_index(partition),
|
2312 + | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
2313 + | EligibilityGroup::Pool,
|
2314 + | Some(admission.clone()),
|
2315 + | None,
|
2316 + | )),
|
2317 + | )
|
2318 + | }
|
2319 + |
|
2320 + | fn connection(
|
2321 + | id: u64,
|
2322 + | ) -> (
|
2323 + | Arc<ConnectionState>,
|
2324 + | super::super::super::connection::PhysicalConnectionGuard,
|
2325 + | ) {
|
2326 + | ConnectionState::unbounded(ConnectionInfo::for_test(
|
2327 + | ConnectionId::new(id),
|
2328 + | PartitionId::from_index(1),
|
2329 + | ))
|
2330 + | }
|
2331 + |
|
2332 + | fn begin_waiter(cell: &Arc<OriginCell>) -> WaiterId {
|
2333 + | let waiter = OriginCell::register_waiter(cell, ProtocolRequirement::H2Required);
|
2334 + | let event = cell
|
2335 + | .take_ready_event(waiter)
|
2336 + | .expect("unbounded H2 waiter did not receive establishment authority");
|
2337 + | let super::super::AcquisitionEvent::Establish(permit) = event else {
|
2338 + | panic!("new H2 waiter completed before establishment");
|
2339 + | };
|
2340 + | assert!(cell.start_establishment(waiter));
|
2341 + | drop(permit);
|
2342 + | waiter
|
2343 + | }
|
2344 + |
|
2345 + | fn install_generation(
|
2346 + | cell: &Arc<OriginCell>,
|
2347 + | connection_id: u64,
|
2348 + | ) -> (
|
2349 + | H2GenerationId,
|
2350 + | Arc<ConnectionState>,
|
2351 + | super::super::super::connection::PhysicalConnectionGuard,
|
2352 + | ) {
|
2353 + | let waiter = begin_waiter(cell);
|
2354 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(waiter) else {
|
2355 + | panic!("fresh cell did not create an HTTP/2 flight");
|
2356 + | };
|
2357 + | let (connection, physical) = connection(connection_id);
|
2358 + | let generation = OriginCell::complete_h2_flight(
|
2359 + | cell,
|
2360 + | flight,
|
2361 + | connection.clone(),
|
2362 + | H2Sender::test(connection_id),
|
2363 + | None,
|
2364 + | )
|
2365 + | .expect("fresh flight did not install");
|
2366 + | let event = cell
|
2367 + | .take_ready_event(waiter)
|
2368 + | .expect("flight completion did not satisfy its waiter");
|
2369 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2370 + | activation,
|
2371 + | )) = event
|
2372 + | else {
|
2373 + | panic!("flight completion produced a non-H2 result");
|
2374 + | };
|
2375 + | drop(activation);
|
2376 + | (generation, connection, physical)
|
2377 + | }
|
2378 + |
|
2379 + | fn generation_counts(cell: &OriginCell, generation: H2GenerationId) -> (usize, usize) {
|
2380 + | cell.h2_request_counts(generation)
|
2381 + | .expect("generation was not installed")
|
2382 + | }
|
2383 + |
|
2384 + | #[test]
|
2385 + | fn flight_converges_participants_and_cancellation() {
|
2386 + | let mut records = H2Records::default();
|
2387 + | let first = WaiterId(1);
|
2388 + | let second = WaiterId(2);
|
2389 + |
|
2390 + | let H2FlightInstall::Driver(flight) = records.install_or_join_flight(first) else {
|
2391 + | panic!("first participant did not become the flight driver");
|
2392 + | };
|
2393 + | assert!(matches!(
|
2394 + | records.install_or_join_flight(second),
|
2395 + | H2FlightInstall::Joined
|
2396 + | ));
|
2397 + | records.cancel_flight_participant(second);
|
2398 + | assert_eq!(Some(vec![first]), records.fail_flight(flight));
|
2399 + | assert!(records.flight.is_none());
|
2400 + | }
|
2401 + |
|
2402 + | #[test]
|
2403 + | fn accepting_generation_prevents_a_second_flight() {
|
2404 + | let mut records = H2Records::default();
|
2405 + | let (connection, _physical) = connection(1);
|
2406 + | let generation =
|
2407 + | records.install_generation(BTreeSet::new(), connection, H2Sender::test(1), None);
|
2408 + |
|
2409 + | assert!(matches!(
|
2410 + | records.install_or_join_flight(WaiterId(1)),
|
2411 + | H2FlightInstall::Accepting(current) if current == generation
|
2412 + | ));
|
2413 + | assert!(records.flight.is_none());
|
2414 + | }
|
2415 + |
|
2416 + | #[test]
|
2417 + | fn generation_join_does_not_retain_a_waiter_already_served_by_the_gate() {
|
2418 + | let cell = cell();
|
2419 + | let first = begin_waiter(&cell);
|
2420 + | let second = begin_waiter(&cell);
|
2421 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
|
2422 + | panic!("first waiter did not become the flight driver");
|
2423 + | };
|
2424 + | assert!(matches!(
|
2425 + | cell.install_or_join_h2_flight(second),
|
2426 + | H2FlightInstall::Joined
|
2427 + | ));
|
2428 + |
|
2429 + | let (connection, _physical) = connection(1);
|
2430 + | let generation =
|
2431 + | OriginCell::complete_h2_flight(&cell, flight, connection, H2Sender::test(1), None)
|
2432 + | .expect("flight did not install");
|
2433 + |
|
2434 + | assert_eq!(
|
2435 + | H2GenerationJoin::WaiterCompleted,
|
2436 + | OriginCell::join_h2_generation(&cell, first, generation)
|
2437 + | );
|
2438 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2439 + | activation,
|
2440 + | )) = cell
|
2441 + | .take_ready_event(first)
|
2442 + | .expect("generation gate did not serve the first waiter")
|
2443 + | else {
|
2444 + | panic!("first waiter received a non-H2 result");
|
2445 + | };
|
2446 + |
|
2447 + | assert!(OriginCell::close_h2(
|
2448 + | &cell,
|
2449 + | generation,
|
2450 + | CloseReason::PoolDropped,
|
2451 + | ));
|
2452 + | drop(activation);
|
2453 + | assert!(matches!(
|
2454 + | cell.take_ready_event(second),
|
2455 + | Some(super::super::AcquisitionEvent::Complete(
|
2456 + | super::super::AcquisitionResult::Reacquire
|
2457 + | ))
|
2458 + | ));
|
2459 + | assert_eq!(0, cell.retained_waiters_for_test());
|
2460 + | }
|
2461 + |
|
2462 + | #[test]
|
2463 + | fn generation_gate_services_committed_waiters_before_direct_arrivals() {
|
2464 + | let cell = cell();
|
2465 + | let first = begin_waiter(&cell);
|
2466 + | let second = begin_waiter(&cell);
|
2467 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
|
2468 + | panic!("first participant did not become the flight driver");
|
2469 + | };
|
2470 + | assert!(matches!(
|
2471 + | cell.install_or_join_h2_flight(second),
|
2472 + | H2FlightInstall::Joined
|
2473 + | ));
|
2474 + | let (connection, _physical) = connection(1);
|
2475 + | OriginCell::complete_h2_flight(&cell, flight, connection, H2Sender::test(1), None)
|
2476 + | .expect("flight did not install");
|
2477 + |
|
2478 + | assert!(
|
2479 + | OriginCell::select_h2(&cell).is_none(),
|
2480 + | "direct arrival bypassed committed waiters"
|
2481 + | );
|
2482 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2483 + | first_activation,
|
2484 + | )) = cell
|
2485 + | .take_ready_event(first)
|
2486 + | .expect("first waiter was not served")
|
2487 + | else {
|
2488 + | panic!("first waiter received a non-H2 result");
|
2489 + | };
|
2490 + | drop(first_activation);
|
2491 + |
|
2492 + | assert!(
|
2493 + | OriginCell::select_h2(&cell).is_none(),
|
2494 + | "direct arrival bypassed the second committed waiter"
|
2495 + | );
|
2496 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2497 + | second_activation,
|
2498 + | )) = cell
|
2499 + | .take_ready_event(second)
|
2500 + | .expect("second waiter was not served")
|
2501 + | else {
|
2502 + | panic!("second waiter received a non-H2 result");
|
2503 + | };
|
2504 + | drop(second_activation);
|
2505 + |
|
2506 + | assert!(
|
2507 + | OriginCell::select_h2(&cell).is_some(),
|
2508 + | "gate did not open after committed waiters drained"
|
2509 + | );
|
2510 + | }
|
2511 + |
|
2512 + | #[test]
|
2513 + | fn cancelling_an_open_gate_activation_services_the_next_waiter() {
|
2514 + | use crate::client::pool::admission::OriginAdmission;
|
2515 + | use std::num::NonZeroUsize;
|
2516 + |
|
2517 + | let admission = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
2518 + | let cell = OriginAdmission::register_cell(
|
2519 + | &admission,
|
2520 + | Arc::new(OriginCell::new(
|
2521 + | PartitionId::from_index(1),
|
2522 + | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
2523 + | EligibilityGroup::Pool,
|
2524 + | Some(admission.clone()),
|
2525 + | None,
|
2526 + | )),
|
2527 + | );
|
2528 + | let (connection, _physical) = ConnectionState::bounded(
|
2529 + | ConnectionInfo::for_test(ConnectionId::new(1), cell.id().partition()),
|
2530 + | OriginAdmission::lease_for_test(&admission),
|
2531 + | );
|
2532 + | let generation = OriginCell::install_h2_for_test(&cell, connection, 1, None);
|
2533 + | cell.state.lock().h2.prioritize_through(Some(WaiterId(0)));
|
2534 + |
|
2535 + | let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
2536 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2537 + | first_activation,
|
2538 + | )) = cell
|
2539 + | .take_ready_event(first)
|
2540 + | .expect("first waiter was not served")
|
2541 + | else {
|
2542 + | panic!("first waiter received a non-H2 result");
|
2543 + | };
|
2544 + | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
2545 + | let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
2546 + | assert_eq!((1, 0), generation_counts(&cell, generation));
|
2547 + |
|
2548 + | drop(first_activation);
|
2549 + | assert_eq!((1, 0), generation_counts(&cell, generation));
|
2550 + | assert!(OriginCell::cancel_waiter(&cell, second));
|
2551 + | assert_eq!(
|
2552 + | (1, 0),
|
2553 + | generation_counts(&cell, generation),
|
2554 + | "cancelling an open-gate activation stranded its successor"
|
2555 + | );
|
2556 + |
|
2557 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2558 + | third_activation,
|
2559 + | )) = cell
|
2560 + | .take_ready_event(third)
|
2561 + | .expect("successor was not served after cancellation")
|
2562 + | else {
|
2563 + | panic!("successor received a non-H2 result");
|
2564 + | };
|
2565 + | drop(third_activation);
|
2566 + | assert_eq!((0, 0), generation_counts(&cell, generation));
|
2567 + | assert!(OriginCell::close_h2(
|
2568 + | &cell,
|
2569 + | generation,
|
2570 + | CloseReason::PoolDropped,
|
2571 + | ));
|
2572 + | }
|
2573 + |
|
2574 + | #[test]
|
2575 + | fn cancelling_an_unserved_flight_participant_prunes_the_generation() {
|
2576 + | let cell = cell();
|
2577 + | let first = begin_waiter(&cell);
|
2578 + | let second = begin_waiter(&cell);
|
2579 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
|
2580 + | panic!("first participant did not become the flight driver");
|
2581 + | };
|
2582 + | assert!(matches!(
|
2583 + | cell.install_or_join_h2_flight(second),
|
2584 + | H2FlightInstall::Joined
|
2585 + | ));
|
2586 + | let (connection, _physical) = connection(1);
|
2587 + | let generation =
|
2588 + | OriginCell::complete_h2_flight(&cell, flight, connection, H2Sender::test(1), None)
|
2589 + | .expect("flight did not install");
|
2590 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2591 + | first_activation,
|
2592 + | )) = cell
|
2593 + | .take_ready_event(first)
|
2594 + | .expect("first participant was not served")
|
2595 + | else {
|
2596 + | panic!("first participant received a non-H2 result");
|
2597 + | };
|
2598 + |
|
2599 + | assert!(OriginCell::cancel_waiter(&cell, second));
|
2600 + | assert!(OriginCell::close_h2(
|
2601 + | &cell,
|
2602 + | generation,
|
2603 + | CloseReason::PoolDropped,
|
2604 + | ));
|
2605 + | drop(first_activation);
|
2606 + | assert_eq!(0, cell.retained_waiters_for_test());
|
2607 + | }
|
2608 + |
|
2609 + | #[test]
|
2610 + | fn returned_h1_prunes_a_transferred_h2_waiter() {
|
2611 + | let cell = cell();
|
2612 + | let first = begin_waiter(&cell);
|
2613 + | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2614 + | let super::super::AcquisitionEvent::Establish(permit) = cell
|
2615 + | .take_ready_event(second)
|
2616 + | .expect("second participant did not receive establishment authority")
|
2617 + | else {
|
2618 + | panic!("second participant completed before establishment");
|
2619 + | };
|
2620 + | assert!(cell.start_establishment(second));
|
2621 + | drop(permit);
|
2622 + |
|
2623 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
|
2624 + | panic!("first participant did not become the flight driver");
|
2625 + | };
|
2626 + | assert!(matches!(
|
2627 + | cell.install_or_join_h2_flight(second),
|
2628 + | H2FlightInstall::Joined
|
2629 + | ));
|
2630 + | let (h2_connection, _h2_physical) = connection(1);
|
2631 + | let generation =
|
2632 + | OriginCell::complete_h2_flight(&cell, flight, h2_connection, H2Sender::test(1), None)
|
2633 + | .expect("flight did not install");
|
2634 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2635 + | first_activation,
|
2636 + | )) = cell
|
2637 + | .take_ready_event(first)
|
2638 + | .expect("first participant was not served")
|
2639 + | else {
|
2640 + | panic!("first participant received a non-H2 result");
|
2641 + | };
|
2642 + |
|
2643 + | let (h1_connection, _h1_physical) = connection(2);
|
2644 + | let returning = OriginCell::install_selected_h1(
|
2645 + | &cell,
|
2646 + | h1_connection,
|
2647 + | super::super::h1::H1Sender::test(2),
|
2648 + | );
|
2649 + | drop(returning);
|
2650 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H1(
|
2651 + | selection,
|
2652 + | )) = cell
|
2653 + | .take_ready_event(second)
|
2654 + | .expect("returned HTTP/1 sender did not serve the compatible waiter")
|
2655 + | else {
|
2656 + | panic!("compatible waiter received a non-H1 result");
|
2657 + | };
|
2658 + | drop(selection);
|
2659 + |
|
2660 + | assert!(OriginCell::close_h2(
|
2661 + | &cell,
|
2662 + | generation,
|
2663 + | CloseReason::PoolDropped,
|
2664 + | ));
|
2665 + | drop(first_activation);
|
2666 + | assert_eq!(0, cell.retained_waiters_for_test());
|
2667 + | }
|
2668 + |
|
2669 + | #[test]
|
2670 + | fn later_local_activation_does_not_hide_an_open_generation_from_peers() {
|
2671 + | let mut records = H2Records::default();
|
2672 + | let committed = WaiterId(1);
|
2673 + | let later = WaiterId(2);
|
2674 + | let H2FlightInstall::Driver(flight) = records.install_or_join_flight(committed) else {
|
2675 + | panic!("first participant did not become the flight driver");
|
2676 + | };
|
2677 + | let (connection, _physical) = connection(1);
|
2678 + | let generation = records
|
2679 + | .complete_flight(flight, connection, H2Sender::test(1), None)
|
2680 + | .expect("flight did not install");
|
2681 + | records.prioritize_through(Some(committed));
|
2682 + |
|
2683 + | assert_eq!(
|
2684 + | GateTurn::Through(committed),
|
2685 + | records.next_gate_turn(true),
|
2686 + | "committed waiter did not retain initial priority"
|
2687 + | );
|
2688 + | records.begin_gate_activation(committed);
|
2689 + | assert_eq!(None, records.publishable_generation());
|
2690 + | assert!(records.finish_gate_activation(generation, committed));
|
2691 + |
|
2692 + | assert_eq!(GateTurn::Open, records.next_gate_turn(false));
|
2693 + | records.begin_gate_activation(later);
|
2694 + | assert_eq!(
|
2695 + | Some(generation),
|
2696 + | records.publishable_generation(),
|
2697 + | "a post-cutoff local activation hid the generation from peers"
|
2698 + | );
|
2699 + | }
|
2700 + |
|
2701 + | #[test]
|
2702 + | fn generation_close_reacquires_unserved_flight_participant() {
|
2703 + | let cell = cell();
|
2704 + | let first = begin_waiter(&cell);
|
2705 + | let second = begin_waiter(&cell);
|
2706 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
|
2707 + | panic!("first participant did not become the flight driver");
|
2708 + | };
|
2709 + | assert!(matches!(
|
2710 + | cell.install_or_join_h2_flight(second),
|
2711 + | H2FlightInstall::Joined
|
2712 + | ));
|
2713 + | let (connection, _physical) = connection(1);
|
2714 + | let generation =
|
2715 + | OriginCell::complete_h2_flight(&cell, flight, connection, H2Sender::test(1), None)
|
2716 + | .expect("flight did not install");
|
2717 + |
|
2718 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2719 + | first_activation,
|
2720 + | )) = cell
|
2721 + | .take_ready_event(first)
|
2722 + | .expect("first participant was not served")
|
2723 + | else {
|
2724 + | panic!("first participant received a non-H2 result");
|
2725 + | };
|
2726 + | assert!(cell.take_ready_event(second).is_none());
|
2727 + |
|
2728 + | assert!(OriginCell::close_h2(
|
2729 + | &cell,
|
2730 + | generation,
|
2731 + | CloseReason::Poisoned
|
2732 + | ));
|
2733 + | assert!(matches!(
|
2734 + | cell.take_ready_event(second),
|
2735 + | Some(super::super::AcquisitionEvent::Complete(
|
2736 + | super::super::AcquisitionResult::Reacquire
|
2737 + | ))
|
2738 + | ));
|
2739 + | drop(first_activation);
|
2740 + | assert_eq!(0, cell.retained_waiters_for_test());
|
2741 + | }
|
2742 + |
|
2743 + | #[test]
|
2744 + | fn bounded_generation_close_returns_capacity_and_reacquires_participant() {
|
2745 + | use crate::client::pool::admission::OriginAdmission;
|
2746 + | use std::num::NonZeroUsize;
|
2747 + |
|
2748 + | let admission = OriginAdmission::for_test(NonZeroUsize::new(2).unwrap());
|
2749 + | let cell = OriginAdmission::register_cell(
|
2750 + | &admission,
|
2751 + | Arc::new(OriginCell::new(
|
2752 + | PartitionId::from_index(1),
|
2753 + | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
2754 + | EligibilityGroup::Pool,
|
2755 + | Some(admission.clone()),
|
2756 + | None,
|
2757 + | )),
|
2758 + | );
|
2759 + | let mut participants = Vec::new();
|
2760 + | for _ in 0..2 {
|
2761 + | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
2762 + | let super::super::AcquisitionEvent::Establish(permit) = cell
|
2763 + | .take_ready_event(waiter)
|
2764 + | .expect("bounded participant did not receive capacity")
|
2765 + | else {
|
2766 + | panic!("bounded participant completed before establishment");
|
2767 + | };
|
2768 + | assert!(cell.start_establishment(waiter));
|
2769 + | participants.push((waiter, permit));
|
2770 + | }
|
2771 + | let (first, first_permit) = participants.remove(0);
|
2772 + | let (second, second_permit) = participants.remove(0);
|
2773 + | let H2FlightInstall::Driver(flight) = cell.install_or_join_h2_flight(first) else {
|
2774 + | panic!("first participant did not become the flight driver");
|
2775 + | };
|
2776 + | assert!(matches!(
|
2777 + | cell.install_or_join_h2_flight(second),
|
2778 + | H2FlightInstall::Joined
|
2779 + | ));
|
2780 + | drop(second_permit);
|
2781 + | let (connection, _physical) = ConnectionState::bounded(
|
2782 + | ConnectionInfo::for_test(ConnectionId::new(1), cell.id().partition()),
|
2783 + | first_permit
|
2784 + | .into_lease()
|
2785 + | .expect("bounded permit had no lease"),
|
2786 + | );
|
2787 + | let generation =
|
2788 + | OriginCell::complete_h2_flight(&cell, flight, connection, H2Sender::test(1), None)
|
2789 + | .expect("bounded flight did not install");
|
2790 + | let super::super::AcquisitionEvent::Complete(super::super::AcquisitionResult::H2(
|
2791 + | first_activation,
|
2792 + | )) = cell
|
2793 + | .take_ready_event(first)
|
2794 + | .expect("first participant was not served")
|
2795 + | else {
|
2796 + | panic!("first participant received a non-H2 result");
|
2797 + | };
|
2798 + |
|
2799 + | assert!(OriginCell::close_h2(
|
2800 + | &cell,
|
2801 + | generation,
|
2802 + | CloseReason::Poisoned
|
2803 + | ));
|
2804 + | assert!(matches!(
|
2805 + | cell.take_ready_event(second),
|
2806 + | Some(super::super::AcquisitionEvent::Complete(
|
2807 + | super::super::AcquisitionResult::Reacquire
|
2808 + | ))
|
2809 + | ));
|
2810 + | drop(first_activation);
|
2811 + | assert_eq!(2, admission.available_capacity_for_test());
|
2812 + | assert_eq!(0, cell.retained_waiters_for_test());
|
2813 + | }
|
2814 + |
|
2815 + | #[test]
|
2816 + | fn generation_is_reused_only_after_hyper_acceptance() {
|
2817 + | let cell = cell();
|
2818 + | let (_generation, connection, _physical) = install_generation(&cell, 1);
|
2819 + |
|
2820 + | let mut activation =
|
2821 + | OriginCell::select_h2(&cell).expect("accepting generation was not selected");
|
2822 + | assert!(!activation.is_reused());
|
2823 + | let H2DispatchParts {
|
2824 + | sender: _sender,
|
2825 + | send_endpoint: send,
|
2826 + | receive_endpoint: receive,
|
2827 + | } = activation.take_dispatch_parts();
|
2828 + | let dispatch = ConnectionState::try_commit_dispatch(&connection)
|
2829 + | .expect("open connection rejected dispatch");
|
2830 + | activation.accept(dispatch);
|
2831 + | drop(send);
|
2832 + | drop(receive);
|
2833 + |
|
2834 + | let reused = OriginCell::select_h2(&cell).expect("accepted generation was not reusable");
|
2835 + | assert!(reused.is_reused());
|
2836 + | drop(reused);
|
2837 + | }
|
2838 + |
|
2839 + | #[test]
|
2840 + | #[should_panic(expected = "HTTP/2 activation dispatch parts already taken")]
|
2841 + | fn activation_transfers_dispatch_parts_once() {
|
2842 + | let cell = cell();
|
2843 + | let (_generation, _connection, _physical) = install_generation(&cell, 1);
|
2844 + | let mut activation =
|
2845 + | OriginCell::select_h2(&cell).expect("accepting generation was not selected");
|
2846 + |
|
2847 + | let _first = activation.take_dispatch_parts();
|
2848 + | let _second = activation.take_dispatch_parts();
|
2849 + | }
|
2850 + |
|
2851 + | #[test]
|
2852 + | fn prospective_activation_cancellation_returns_its_generation_count() {
|
2853 + | let cell = cell();
|
2854 + | let (generation, _connection, _physical) = install_generation(&cell, 1);
|
2855 + |
|
2856 + | let activation =
|
2857 + | OriginCell::select_h2(&cell).expect("accepting generation was not selected");
|
2858 + | assert_eq!((1, 0), generation_counts(&cell, generation));
|
2859 + | drop(activation);
|
2860 + | assert_eq!((0, 0), generation_counts(&cell, generation));
|
2861 + | }
|
2862 + |
|
2863 + | #[test]
|
2864 + | fn accepted_lease_waits_for_both_endpoints_in_either_order() {
|
2865 + | for send_first in [true, false] {
|
2866 + | let cell = cell();
|
2867 + | let (generation, connection, _physical) = install_generation(&cell, 1);
|
2868 + | let mut activation =
|
2869 + | OriginCell::select_h2(&cell).expect("accepting generation was not selected");
|
2870 + | let H2DispatchParts {
|
2871 + | sender: _sender,
|
2872 + | send_endpoint: send,
|
2873 + | receive_endpoint: receive,
|
2874 + | } = activation.take_dispatch_parts();
|
2875 + | let dispatch = ConnectionState::try_commit_dispatch(&connection)
|
2876 + | .expect("open connection rejected dispatch");
|
2877 + | activation.accept(dispatch);
|
2878 + | assert_eq!((0, 1), generation_counts(&cell, generation));
|
2879 + | assert_eq!(1, connection.snapshot().in_flight);
|
2880 + |
|
2881 + | if send_first {
|
2882 + | drop(send);
|
2883 + | assert_eq!((0, 1), generation_counts(&cell, generation));
|
2884 + | receive.complete();
|
2885 + | } else {
|
2886 + | receive.complete();
|
2887 + | assert_eq!((0, 1), generation_counts(&cell, generation));
|
2888 + | drop(send);
|
2889 + | }
|
2890 + |
|
2891 + | assert_eq!((0, 0), generation_counts(&cell, generation));
|
2892 + | assert_eq!(0, connection.snapshot().in_flight);
|
2893 + | }
|
2894 + | }
|
2895 + |
|
2896 + | #[test]
|
2897 + | fn stale_route_cannot_activate_a_replacement_or_retained_drain() {
|
2898 + | let cell = cell();
|
2899 + | let (first, _first_connection, _first_physical) = install_generation(&cell, 1);
|
2900 + | let stale_route = H2Route::new(&cell, first);
|
2901 + | let retained = OriginCell::select_h2(&cell).expect("first generation was not selectable");
|
2902 + | assert!(OriginCell::close_h2(
|
2903 + | &cell,
|
2904 + | first,
|
2905 + | CloseReason::ProtocolClosed
|
2906 + | ));
|
2907 + |
|
2908 + | let (second, _second_connection, _second_physical) = install_generation(&cell, 2);
|
2909 + | assert_ne!(first, second);
|
2910 + | assert!(
|
2911 + | stale_route.activate(PartitionId::from_index(2)).is_none(),
|
2912 + | "stale route activated a draining or replacement generation"
|
2913 + | );
|
2914 + | drop(retained);
|
2915 + | }
|
2916 + |
|
2917 + | #[test]
|
2918 + | fn accepting_identity_rejects_a_retained_draining_generation() {
|
2919 + | let cell = cell();
|
2920 + | let (first, _first_connection, _first_physical) = install_generation(&cell, 1);
|
2921 + | let retained = OriginCell::select_h2(&cell).expect("first generation was not selectable");
|
2922 + | assert!(OriginCell::close_h2(
|
2923 + | &cell,
|
2924 + | first,
|
2925 + | CloseReason::ProtocolClosed
|
2926 + | ));
|
2927 + | let (second, _second_connection, _second_physical) = install_generation(&cell, 2);
|
2928 + |
|
2929 + | assert!(cell.state.lock().h2.is_accepting(second));
|
2930 + | assert!(!cell.state.lock().h2.is_accepting(first));
|
2931 + | drop(retained);
|
2932 + | }
|
2933 + |
|
2934 + | #[test]
|
2935 + | fn activation_requires_the_exact_accepting_generation() {
|
2936 + | let cell = cell();
|
2937 + | let (first, _first_connection, _first_physical) = install_generation(&cell, 1);
|
2938 + | let retained = OriginCell::select_h2(&cell).expect("first generation was not selectable");
|
2939 + | assert!(OriginCell::close_h2(
|
2940 + | &cell,
|
2941 + | first,
|
2942 + | CloseReason::ProtocolClosed
|
2943 + | ));
|
2944 + | let (second, _second_connection, _second_physical) = install_generation(&cell, 2);
|
2945 + |
|
2946 + | {
|
2947 + | let mut state = cell.state.lock();
|
2948 + | let first_record = state
|
2949 + | .h2
|
2950 + | .generations
|
2951 + | .get_mut(&first)
|
2952 + | .expect("retained generation disappeared");
|
2953 + | assert_eq!(H2Residence::Draining, first_record.residence);
|
2954 + | first_record.residence = H2Residence::Accepting;
|
2955 + | assert!(
|
2956 + | state.h2.activate(first).is_none(),
|
2957 + | "non-current generation accepted an activation"
|
2958 + | );
|
2959 + | state
|
2960 + | .h2
|
2961 + | .generations
|
2962 + | .get_mut(&first)
|
2963 + | .expect("retained generation disappeared")
|
2964 + | .residence = H2Residence::Draining;
|
2965 + | state.assert_consistent();
|
2966 + | }
|
2967 + |
|
2968 + | assert!(cell.state.lock().h2.is_accepting(second));
|
2969 + | drop(retained);
|
2970 + | }
|
2971 + |
|
2972 + | #[test]
|
2973 + | fn activation_requires_accepting_residence() {
|
2974 + | let cell = cell();
|
2975 + | let (first, _first_connection, _first_physical) = install_generation(&cell, 1);
|
2976 + | let retained = OriginCell::select_h2(&cell).expect("first generation was not selectable");
|
2977 + | assert!(OriginCell::close_h2(
|
2978 + | &cell,
|
2979 + | first,
|
2980 + | CloseReason::ProtocolClosed
|
2981 + | ));
|
2982 + | let (second, _second_connection, _second_physical) = install_generation(&cell, 2);
|
2983 + |
|
2984 + | {
|
2985 + | let mut state = cell.state.lock();
|
2986 + | assert_eq!(
|
2987 + | H2Residence::Draining,
|
2988 + | state.h2.generations[&first].residence
|
2989 + | );
|
2990 + | state.h2.accepting = Some(first);
|
2991 + | assert!(
|
2992 + | state.h2.activate(first).is_none(),
|
2993 + | "draining generation accepted an activation"
|
2994 + | );
|
2995 + | state.h2.accepting = Some(second);
|
2996 + | state.assert_consistent();
|
2997 + | }
|
2998 + |
|
2999 + | drop(retained);
|
3000 + | }
|
3001 + |
|
3002 + | #[test]
|
3003 + | fn local_generation_excludes_a_peer_route() {
|
3004 + | let local_cell = cell();
|
3005 + | let (generation, _connection, _physical) = install_generation(&local_cell, 1);
|
3006 + | let peer_cell = Arc::new(OriginCell::new(
|
3007 + | PartitionId::from_index(2),
|
3008 + | local_cell.id().origin().clone(),
|
3009 + | EligibilityGroup::Pool,
|
3010 + | None,
|
3011 + | None,
|
3012 + | ));
|
3013 + | let route = H2Route::new(&peer_cell, H2GenerationId::for_test(99));
|
3014 + |
|
3015 + | let mut state = local_cell.state.lock();
|
3016 + | state.h2.install_peer_route(route, None);
|
3017 + | assert!(state.h2.peer_route.is_none());
|
3018 + | assert_eq!(Some(generation), state.h2.accepting());
|
3019 + | }
|
3020 + |
|
3021 + | #[test]
|
3022 + | fn open_generation_allows_concurrent_prospective_activations() {
|
3023 + | let cell = cell();
|
3024 + | let (generation, _connection, _physical) = install_generation(&cell, 1);
|
3025 + |
|
3026 + | let first = OriginCell::select_h2(&cell).expect("first activation was not selected");
|
3027 + | let second =
|
3028 + | OriginCell::select_h2(&cell).expect("open generation serialized the second activation");
|
3029 + | assert_eq!((2, 0), generation_counts(&cell, generation));
|
3030 + | drop(first);
|
3031 + | drop(second);
|
3032 + | assert_eq!((0, 0), generation_counts(&cell, generation));
|
3033 + | }
|
3034 + |
|
3035 + | #[test]
|
3036 + | fn consistency_checks_do_not_repanic_during_unwind() {
|
3037 + | use std::panic::{catch_unwind, AssertUnwindSafe};
|
3038 + |
|
3039 + | let cell = cell();
|
3040 + | let (generation, _connection, _physical) = install_generation(&cell, 1);
|
3041 + | let activation =
|
3042 + | OriginCell::select_h2(&cell).expect("accepting generation did not activate");
|
3043 + | let (corrupt_connection, corrupt_physical) = connection(2);
|
3044 + | {
|
3045 + | let mut state = cell.state.lock();
|
3046 + | state.h2.generations.insert(
|
3047 + | H2GenerationId(99),
|
3048 + | H2Generation {
|
3049 + | pending_waiters: BTreeSet::new(),
|
3050 + | connection: corrupt_connection,
|
3051 + | sender: H2Sender::test(2),
|
3052 + | residence: H2Residence::Draining,
|
3053 + | prospective: 0,
|
3054 + | has_dispatched: false,
|
3055 + | accepted: 0,
|
3056 + | idle_deadline: None,
|
3057 + | },
|
3058 + | );
|
3059 + | }
|
3060 + |
|
3061 + | let result = catch_unwind(AssertUnwindSafe(move || {
|
3062 + | let _activation = activation;
|
3063 + | panic!("primary test panic");
|
3064 + | }));
|
3065 + | assert!(result.is_err(), "primary panic was not observed");
|
3066 + |
|
3067 + | let removed = {
|
3068 + | let mut state = cell.state.lock();
|
3069 + | state.h2.generations.remove(&H2GenerationId(99))
|
3070 + | };
|
3071 + | drop(removed);
|
3072 + | drop(corrupt_physical);
|
3073 + | assert!(OriginCell::close_h2(
|
3074 + | &cell,
|
3075 + | generation,
|
3076 + | CloseReason::PoolDropped,
|
3077 + | ));
|
3078 + | }
|
3079 + |
|
3080 + | #[test]
|
3081 + | fn removed_generation_id_cannot_activate_or_close_the_current_generation() {
|
3082 + | let cell = cell();
|
3083 + | let (first, first_connection, _first_physical) = install_generation(&cell, 1);
|
3084 + | let stale = H2CloseHandle::new(&cell, first);
|
3085 + | assert!(stale.close(CloseReason::ProtocolClosed));
|
3086 + | assert_eq!(
|
3087 + | Some(CloseReason::ProtocolClosed),
|
3088 + | first_connection.snapshot().close_reason
|
3089 + | );
|
3090 + |
|
3091 + | let (second, second_connection, _second_physical) = install_generation(&cell, 2);
|
3092 + | assert_ne!(first, second);
|
3093 + | assert!(OriginCell::activate_h2(&cell, first).is_none());
|
3094 + | assert!(!stale.close(CloseReason::Poisoned));
|
3095 + | assert_eq!(None, second_connection.snapshot().close_reason);
|
3096 + | assert_eq!(Some(second), cell.accepting_h2_generation());
|
3097 + | }
|
3098 + |
|
3099 + | #[test]
|
3100 + | fn close_drops_detached_capacity_only_after_cell_unlock() {
|
3101 + | use crate::client::pool::admission::OriginAdmission;
|
3102 + | use std::num::NonZeroUsize;
|
3103 + | use std::panic::{catch_unwind, AssertUnwindSafe};
|
3104 + |
|
3105 + | let admission = OriginAdmission::for_test(NonZeroUsize::MIN);
|
3106 + | let cell = OriginAdmission::register_cell(
|
3107 + | &admission,
|
3108 + | Arc::new(OriginCell::new(
|
3109 + | PartitionId::from_index(1),
|
3110 + | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
3111 + | EligibilityGroup::Pool,
|
3112 + | Some(admission.clone()),
|
3113 + | None,
|
3114 + | )),
|
3115 + | );
|
3116 + | let info = ConnectionInfo::for_test(ConnectionId::new(1), PartitionId::from_index(1));
|
3117 + | let (bounded_connection, physical) = ConnectionState::pending_open(info);
|
3118 + | bounded_connection
|
3119 + | .open(Some(OriginAdmission::lease_for_test(&admission)))
|
3120 + | .expect("bounded test connection did not open");
|
3121 + | let generation =
|
3122 + | OriginCell::install_h2_for_test(&cell, bounded_connection.clone(), 1, None);
|
3123 + | drop(bounded_connection);
|
3124 + | drop(physical);
|
3125 + |
|
3126 + | // Keep an unrelated empty drain so the close transition's final
|
3127 + | // consistency check panics after detaching the bounded connection.
|
3128 + | let (corrupt_connection, corrupt_physical) = connection(2);
|
3129 + | {
|
3130 + | let mut state = cell.state.lock();
|
3131 + | state.h2.generations.insert(
|
3132 + | H2GenerationId(99),
|
3133 + | H2Generation {
|
3134 + | pending_waiters: BTreeSet::new(),
|
3135 + | connection: corrupt_connection,
|
3136 + | sender: H2Sender::test(2),
|
3137 + | residence: H2Residence::Draining,
|
3138 + | prospective: 0,
|
3139 + | has_dispatched: false,
|
3140 + | accepted: 0,
|
3141 + | idle_deadline: None,
|
3142 + | },
|
3143 + | );
|
3144 + | }
|
3145 + | drop(corrupt_physical);
|
3146 + |
|
3147 + | let result = catch_unwind(AssertUnwindSafe(|| {
|
3148 + | OriginCell::close_h2(&cell, generation, CloseReason::ProtocolClosed);
|
3149 + | }));
|
3150 + | assert!(
|
3151 + | result.is_err(),
|
3152 + | "corrupt H2 state did not reach its consistency assertion"
|
3153 + | );
|
3154 + |
|
3155 + | // This second lease exists only if unwinding dropped the connection
|
3156 + | // after the cell guard, allowing capacity to return through admission.
|
3157 + | drop(OriginAdmission::lease_for_test(&admission));
|
3158 + | }
|
3159 + |
|
3160 + | #[test]
|
3161 + | fn close_retains_an_accepted_generation_until_both_endpoints_finish() {
|
3162 + | let cell = cell();
|
3163 + | let (generation, connection, _physical) = install_generation(&cell, 1);
|
3164 + | let mut activation =
|
3165 + | OriginCell::select_h2(&cell).expect("accepting generation was not selected");
|
3166 + | let H2DispatchParts {
|
3167 + | sender: _sender,
|
3168 + | send_endpoint: send,
|
3169 + | receive_endpoint: receive,
|
3170 + | } = activation.take_dispatch_parts();
|
3171 + | let dispatch = ConnectionState::try_commit_dispatch(&connection)
|
3172 + | .expect("open connection rejected dispatch");
|
3173 + | activation.accept(dispatch);
|
3174 + |
|
3175 + | assert!(OriginCell::close_h2(
|
3176 + | &cell,
|
3177 + | generation,
|
3178 + | CloseReason::ProtocolClosed
|
3179 + | ));
|
3180 + | assert!(cell.state.lock().h2.generations.contains_key(&generation));
|
3181 + | drop(send);
|
3182 + | assert!(cell.state.lock().h2.generations.contains_key(&generation));
|
3183 + | drop(receive);
|
3184 + | assert!(!cell.state.lock().h2.generations.contains_key(&generation));
|
3185 + | assert_eq!(0, connection.snapshot().in_flight);
|
3186 + | }
|
3187 + |
|
3188 + | #[test]
|
3189 + | fn driver_task_drop_closes_its_exact_generation() {
|
3190 + | let cell = cell();
|
3191 + | let (generation, connection, _physical) = install_generation(&cell, 1);
|
3192 + | let guard = H2DriverGuard::new(H2CloseHandle::new(&cell, generation));
|
3193 + |
|
3194 + | drop(guard);
|
3195 + |
|
3196 + | assert_eq!(
|
3197 + | Some(CloseReason::OwnerRuntimeShutdown),
|
3198 + | connection.snapshot().close_reason
|
3199 + | );
|
3200 + | assert_eq!(None, cell.accepting_h2_generation());
|
3201 + | }
|
3202 + |
|
3203 + | #[test]
|
3204 + | fn local_selection_withdraws_idle_h2_from_reclaim() {
|
3205 + | use crate::client::pool::admission::OriginAdmission;
|
3206 + | use std::num::NonZeroUsize;
|
3207 + |
|
3208 + | let admission = OriginAdmission::for_test(NonZeroUsize::MIN);
|
3209 + | let connection_cell = bounded_cell(&admission, 1);
|
3210 + | let requesting_cell = bounded_cell(&admission, 2);
|
3211 + | let info = ConnectionInfo::for_test(ConnectionId::new(1), connection_cell.id().partition());
|
3212 + | let (connection, _physical) = ConnectionState::pending_open(info);
|
3213 + | connection
|
3214 + | .open(Some(OriginAdmission::lease_for_test(&admission)))
|
3215 + | .expect("bounded HTTP/2 connection did not open");
|
3216 + | let generation =
|
3217 + | OriginCell::install_h2_for_test(&connection_cell, connection.clone(), 1, None);
|
3218 + | let activation = OriginCell::select_h2(&connection_cell)
|
3219 + | .expect("local accepting generation did not activate");
|
3220 + |
|
3221 + | let (waiter, demand) =
|
3222 + | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H1Required);
|
3223 + | let action = OriginAdmission::publish_action_without_driving(
|
3224 + | &admission,
|
3225 + | requesting_cell.id().partition(),
|
3226 + | demand,
|
3227 + | );
|
3228 + | assert!(
|
3229 + | action.is_none(),
|
3230 + | "busy local HTTP/2 generation remained eligible for reclaim"
|
3231 + | );
|
3232 + |
|
3233 + | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
3234 + | drop(activation);
|
3235 + | assert_eq!(None, connection.snapshot().close_reason);
|
3236 + | assert!(OriginCell::close_h2(
|
3237 + | &connection_cell,
|
3238 + | generation,
|
3239 + | CloseReason::PoolDropped,
|
3240 + | ));
|
3241 + | }
|
3242 + |
|
3243 + | #[test]
|
3244 + | fn h1_required_demand_preserves_h2_without_an_http1_guarantee() {
|
3245 + | use crate::client::pool::admission::OriginAdmission;
|
3246 + | use std::num::NonZeroUsize;
|
3247 + |
|
3248 + | let admission = OriginAdmission::for_test_with_h2_reclaim(NonZeroUsize::MIN, false);
|
3249 + | let connection_cell = bounded_cell(&admission, 1);
|
3250 + | let requesting_cell = bounded_cell(&admission, 2);
|
3251 + | let info = ConnectionInfo::for_test(ConnectionId::new(1), connection_cell.id().partition());
|
3252 + | let (connection, _physical) = ConnectionState::pending_open(info);
|
3253 + | connection
|
3254 + | .open(Some(OriginAdmission::lease_for_test(&admission)))
|
3255 + | .expect("bounded HTTP/2 connection did not open");
|
3256 + | let generation =
|
3257 + | OriginCell::install_h2_for_test(&connection_cell, connection.clone(), 1, None);
|
3258 + |
|
3259 + | let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Required);
|
3260 + | assert!(
|
3261 + | requesting_cell.take_ready_event(waiter).is_none(),
|
3262 + | "H1-required demand reclaimed capacity without an HTTP/1 guarantee"
|
3263 + | );
|
3264 + | assert_eq!(None, connection.snapshot().close_reason);
|
3265 + | assert_eq!(Some(generation), connection_cell.accepting_h2_generation());
|
3266 + |
|
3267 + | assert!(OriginCell::close_h2(
|
3268 + | &connection_cell,
|
3269 + | generation,
|
3270 + | CloseReason::ProtocolClosed,
|
3271 + | ));
|
3272 + | let super::super::AcquisitionEvent::Establish(permit) = requesting_cell
|
3273 + | .take_ready_event(waiter)
|
3274 + | .expect("ordinary H2 close did not release capacity")
|
3275 + | else {
|
3276 + | panic!("released capacity produced a non-establishment result");
|
3277 + | };
|
3278 + | drop(permit);
|
3279 + | assert_eq!(1, admission.available_capacity_for_test());
|
3280 + | }
|
3281 + |
|
3282 + | #[test]
|
3283 + | fn h1_required_demand_reclaims_idle_h2_capacity() {
|
3284 + | use crate::client::pool::admission::OriginAdmission;
|
3285 + | use std::num::NonZeroUsize;
|
3286 + |
|
3287 + | let admission = OriginAdmission::for_test(NonZeroUsize::MIN);
|
3288 + | let connection_cell = bounded_cell(&admission, 1);
|
3289 + | let requesting_cell = bounded_cell(&admission, 2);
|
3290 + | let info = ConnectionInfo::for_test(ConnectionId::new(1), connection_cell.id().partition());
|
3291 + | let (connection, _physical) = ConnectionState::pending_open(info);
|
3292 + | connection
|
3293 + | .open(Some(OriginAdmission::lease_for_test(&admission)))
|
3294 + | .expect("bounded HTTP/2 connection did not open");
|
3295 + | OriginCell::install_h2_for_test(&connection_cell, connection.clone(), 1, None);
|
3296 + |
|
3297 + | let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Required);
|
3298 + | let super::super::AcquisitionEvent::Establish(permit) = requesting_cell
|
3299 + | .take_ready_event(waiter)
|
3300 + | .expect("idle HTTP/2 reclaim did not deliver capacity")
|
3301 + | else {
|
3302 + | panic!("idle HTTP/2 reclaim produced a non-capacity result");
|
3303 + | };
|
3304 + |
|
3305 + | assert_eq!(
|
3306 + | Some(CloseReason::Reclaimed),
|
3307 + | connection.snapshot().close_reason
|
3308 + | );
|
3309 + | assert_eq!(None, connection_cell.accepting_h2_generation());
|
3310 + | drop(permit);
|
3311 + | assert_eq!(1, admission.available_capacity_for_test());
|
3312 + | }
|
3313 + |
|
3314 + | #[test]
|
3315 + | fn h1_required_demand_waits_for_active_h2_before_reclaim() {
|
3316 + | use crate::client::pool::admission::OriginAdmission;
|
3317 + | use std::num::NonZeroUsize;
|
3318 + |
|
3319 + | let admission = OriginAdmission::for_test(NonZeroUsize::MIN);
|
3320 + | let connection_cell = bounded_cell(&admission, 1);
|
3321 + | let requesting_cell = bounded_cell(&admission, 2);
|
3322 + | let info = ConnectionInfo::for_test(ConnectionId::new(1), connection_cell.id().partition());
|
3323 + | let (connection, _physical) = ConnectionState::pending_open(info);
|
3324 + | connection
|
3325 + | .open(Some(OriginAdmission::lease_for_test(&admission)))
|
3326 + | .expect("bounded HTTP/2 connection did not open");
|
3327 + | let generation =
|
3328 + | OriginCell::install_h2_for_test(&connection_cell, connection.clone(), 1, None);
|
3329 + | let activation = OriginCell::activate_h2(&connection_cell, generation)
|
3330 + | .expect("accepting HTTP/2 generation did not activate");
|
3331 + |
|
3332 + | let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Required);
|
3333 + | assert!(
|
3334 + | requesting_cell.take_ready_event(waiter).is_none(),
|
3335 + | "active HTTP/2 generation was reclaimed"
|
3336 + | );
|
3337 + | assert_eq!(None, connection.snapshot().close_reason);
|
3338 + |
|
3339 + | drop(activation);
|
3340 + | let super::super::AcquisitionEvent::Establish(permit) = requesting_cell
|
3341 + | .take_ready_event(waiter)
|
3342 + | .expect("idle transition did not resume H1-required demand")
|
3343 + | else {
|
3344 + | panic!("idle HTTP/2 reclaim produced a non-capacity result");
|
3345 + | };
|
3346 + | assert_eq!(
|
3347 + | Some(CloseReason::Reclaimed),
|
3348 + | connection.snapshot().close_reason
|
3349 + | );
|
3350 + | drop(permit);
|
3351 + | assert_eq!(1, admission.available_capacity_for_test());
|
3352 + | }
|
3353 + |
|
3354 + | #[test]
|
3355 + | fn pool_shutdown_closes_an_accepting_generation() {
|
3356 + | let cell = cell();
|
3357 + | let (_generation, connection, _physical) = install_generation(&cell, 1);
|
3358 + |
|
3359 + | OriginCell::close_all(&cell, CloseReason::PoolDropped);
|
3360 + |
|
3361 + | assert_eq!(
|
3362 + | Some(CloseReason::PoolDropped),
|
3363 + | connection.snapshot().close_reason
|
3364 + | );
|
3365 + | assert_eq!(None, cell.accepting_h2_generation());
|
3366 + | }
|
3367 + | }
|
3368 + |
|
3369 + | #[cfg(all(test, smithy_http_client_loom))]
|
3370 + | mod loom_tests {
|
3371 + | use super::*;
|
3372 + | use crate::client::pool::connection::{CloseReason, ConnectionInfo};
|
3373 + | use crate::client::pool::origin::OriginKey;
|
3374 + | use crate::client::pool::partition::EligibilityGroup;
|
3375 + | use aws_smithy_runtime_api::client::connection::ConnectionId;
|
3376 + | use http_1x::uri::Scheme;
|
3377 + |
|
3378 + | fn cell() -> Arc<OriginCell> {
|
3379 + | Arc::new(OriginCell::new(
|
3380 + | PartitionId::from_index(1),
|
3381 + | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
3382 + | EligibilityGroup::Pool,
|
3383 + | None,
|
3384 + | None,
|
3385 + | ))
|
3386 + | }
|
3387 + |
|
3388 + | fn install_generation(cell: &Arc<OriginCell>) -> (H2GenerationId, Arc<ConnectionState>) {
|
3389 + | let (connection, _physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
|
3390 + | ConnectionId::new(1),
|
3391 + | PartitionId::from_index(1),
|
3392 + | ));
|
3393 + | let generation = OriginCell::install_h2_for_test(cell, connection.clone(), 1, None);
|
3394 + | (generation, connection)
|
3395 + | }
|
3396 + |
|
3397 + | #[test]
|
3398 + | fn activation_linearizes_against_generation_close() {
|
3399 + | loom::model(|| {
|
3400 + | let cell = cell();
|
3401 + | let (generation, connection) = install_generation(&cell);
|
3402 + | let activating_cell = cell.clone();
|
3403 + | let activation =
|
3404 + | loom::thread::spawn(move || OriginCell::activate_h2(&activating_cell, generation));
|
3405 + | let closing_cell = cell.clone();
|
3406 + | let close = loom::thread::spawn(move || {
|
3407 + | OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
|
3408 + | });
|
3409 + |
|
3410 + | drop(activation.join().unwrap());
|
3411 + | assert!(close.join().unwrap());
|
3412 + | assert_eq!(
|
3413 + | Some(CloseReason::Poisoned),
|
3414 + | connection.snapshot().close_reason
|
3415 + | );
|
3416 + | assert!(!cell.state.lock().h2.generations.contains_key(&generation));
|
3417 + | });
|
3418 + | }
|
3419 + |
|
3420 + | #[test]
|
3421 + | fn concurrent_lease_endpoint_completion_releases_one_dispatch() {
|
3422 + | loom::model(|| {
|
3423 + | let cell = cell();
|
3424 + | let (generation, connection) = install_generation(&cell);
|
3425 + | let mut activation =
|
3426 + | OriginCell::activate_h2(&cell, generation).expect("generation did not activate");
|
3427 + | let H2DispatchParts {
|
3428 + | sender: _sender,
|
3429 + | send_endpoint: send,
|
3430 + | receive_endpoint: receive,
|
3431 + | } = activation.take_dispatch_parts();
|
3432 + | let dispatch = ConnectionState::try_commit_dispatch(&connection)
|
3433 + | .expect("open connection rejected dispatch");
|
3434 + | activation.accept(dispatch);
|
3435 + |
|
3436 + | let send = loom::thread::spawn(move || drop(send));
|
3437 + | let receive = loom::thread::spawn(move || drop(receive));
|
3438 + | send.join().unwrap();
|
3439 + | receive.join().unwrap();
|
3440 + |
|
3441 + | assert_eq!(0, connection.snapshot().in_flight);
|
3442 + | let state = cell.state.lock();
|
3443 + | let record = state
|
3444 + | .h2
|
3445 + | .generations
|
3446 + | .get(&generation)
|
3447 + | .expect("accepting generation disappeared");
|
3448 + | assert_eq!(0, record.accepted);
|
3449 + | });
|
3450 + | }
|
3451 + |
|
3452 + | #[test]
|
3453 + | fn generation_close_waits_for_both_lease_endpoints() {
|
3454 + | loom::model(|| {
|
3455 + | let cell = cell();
|
3456 + | let (connection, _physical) = ConnectionState::unbounded(ConnectionInfo::for_test(
|
3457 + | ConnectionId::new(1),
|
3458 + | PartitionId::from_index(1),
|
3459 + | ));
|
3460 + | let generation = OriginCell::install_h2_for_test(&cell, connection.clone(), 1, None);
|
3461 + | let mut activation =
|
3462 + | OriginCell::activate_h2(&cell, generation).expect("generation did not activate");
|
3463 + | let H2DispatchParts {
|
3464 + | sender: _sender,
|
3465 + | send_endpoint: send,
|
3466 + | receive_endpoint: receive,
|
3467 + | } = activation.take_dispatch_parts();
|
3468 + | let dispatch = ConnectionState::try_commit_dispatch(&connection)
|
3469 + | .expect("open connection rejected dispatch");
|
3470 + | activation.accept(dispatch);
|
3471 + |
|
3472 + | let first_endpoint = loom::thread::spawn(move || drop(send));
|
3473 + | let closing_cell = cell.clone();
|
3474 + | let close = loom::thread::spawn(move || {
|
3475 + | OriginCell::close_h2(&closing_cell, generation, CloseReason::ProtocolClosed)
|
3476 + | });
|
3477 + | first_endpoint.join().unwrap();
|
3478 + | assert!(close.join().unwrap());
|
3479 + |
|
3480 + | assert_eq!(1, connection.snapshot().in_flight);
|
3481 + | assert_eq!(
|
3482 + | Some((0, 1)),
|
3483 + | cell.h2_request_counts(generation),
|
3484 + | "draining generation released before its second endpoint"
|
3485 + | );
|
3486 + | drop(receive);
|
3487 + | assert_eq!(0, connection.snapshot().in_flight);
|
3488 + | assert_eq!(
|
3489 + | None,
|
3490 + | cell.h2_request_counts(generation),
|
3491 + | "finished draining generation remained installed"
|
3492 + | );
|
3493 + | });
|
3494 + | }
|
3495 + | }
|