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