1 + | /*
|
2 + | * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
3 + | * SPDX-License-Identifier: Apache-2.0
|
4 + | */
|
5 + |
|
6 + | //! Admission's view of HTTP/1 connection supply for one bounded origin.
|
7 + | //!
|
8 + | //! [`H1Supply`] indexes cells with selectable HTTP/1 senders and pairs the oldest
|
9 + | //! origin demand with one supplier. Compatible demand borrows the sender.
|
10 + | //! Incompatible demand closes the selected connection to recover capacity.
|
11 + | //!
|
12 + | //! Admission retains an [`H1Match`] while supplier reservation crosses locks or
|
13 + | //! waits for a sender return. The match owns identities and phase flags only.
|
14 + | //! Detached values own resources and their drop fallback.
|
15 + | //!
|
16 + | //! Admission advances each match through these phases:
|
17 + | //!
|
18 + | //! ```text
|
19 + | //! Reserving -- reservation installed -----------------------> WaitingForSender
|
20 + | //! Reserving -- sender extracted ----------------------------> Resolving
|
21 + | //! Reserving(cancelled) -- reservation installed ------------> Cancelling
|
22 + | //! Reserving(cancelled) -- sender extracted -----------------> Resolving
|
23 + | //! Reserving -- reservation rejected or cell expired --------> removed
|
24 + | //! WaitingForSender -- sender return intercepted ------------> Resolving
|
25 + | //! WaitingForSender -- demand cancelled ---------------------> Cancelling
|
26 + | //! WaitingForSender -- terminal supplier outcome ------------> removed
|
27 + | //! Cancelling -- sender return intercepted ------------------> Resolving
|
28 + | //! Cancelling -- reservation cleared -------------------------> removed
|
29 + | //! Resolving -- repeated candidate handoff ------------------> Resolving
|
30 + | //! Resolving -- borrow or reclaim completes ------------------> removed
|
31 + | //! ```
|
32 + | //!
|
33 + | //! A sender extracted during installation takes the `Resolving` path even if
|
34 + | //! cancellation raced with the install acknowledgement. The extracted sender
|
35 + | //! must complete its terminal transition before cancellation can finish.
|
36 + |
|
37 + | use super::{
|
38 + | AdmissionAction, DeliveryGuard, DemandAssignmentOutcome, DemandId, DemandSchedule,
|
39 + | IntrusiveLinks, IntrusiveOrder, OriginAdmission, SupplyRevision,
|
40 + | };
|
41 + | use crate::client::pool::cell::h1::{H1Selection, ProvisionalH1};
|
42 + | use crate::client::pool::cell::{H1ReservationDecision, OriginCell};
|
43 + | use crate::client::pool::partition::{EligibilityGroup, PartitionId};
|
44 + | use crate::sync::Arc;
|
45 + | use aws_smithy_runtime_api::client::connection::ConnectionId;
|
46 + | use std::collections::{HashMap, VecDeque};
|
47 + | use std::fmt;
|
48 + |
|
49 + | /// Identity of one retained HTTP/1 demand-to-supplier match.
|
50 + | #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
|
51 + | pub(in crate::client::pool) struct H1MatchId(u64);
|
52 + |
|
53 + | impl H1MatchId {
|
54 + | /// Creates a deterministic identity for focused transition tests.
|
55 + | #[cfg(test)]
|
56 + | pub(in crate::client::pool) const fn for_test(value: u64) -> Self {
|
57 + | Self(value)
|
58 + | }
|
59 + | }
|
60 + |
|
61 + | /// How one retained HTTP/1 match satisfies waiting demand.
|
62 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
63 + | pub(in crate::client::pool) enum H1MatchKind {
|
64 + | /// Move the selected sender to the requesting cell for dispatch.
|
65 + | BorrowSender,
|
66 + | /// Close the selected connection and deliver its capacity instead.
|
67 + | ReclaimCapacity,
|
68 + | }
|
69 + |
|
70 + | /// Origin-locked HTTP/1 supply indexes and retained cross-lock matches.
|
71 + | #[derive(Debug, Default)]
|
72 + | pub(super) struct H1Supply {
|
73 + | index: H1SupplyIndex,
|
74 + | matches: HashMap<H1MatchId, H1Match>,
|
75 + | by_requester: HashMap<PartitionId, H1MatchId>,
|
76 + | /// Lazy cancellation queue; resolution may leave a tombstone to discard.
|
77 + | cancellations: VecDeque<H1MatchId>,
|
78 + | next_match_id: u64,
|
79 + | }
|
80 + |
|
81 + | /// Admission-facing HTTP/1 status derived under one cell lock.
|
82 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
83 + | pub(in crate::client::pool) struct H1SupplyStatus {
|
84 + | pub(in crate::client::pool) has_returnable_connection: bool,
|
85 + | pub(in crate::client::pool) peer_use_blocked: bool,
|
86 + | }
|
87 + |
|
88 + | /// Terminal HTTP/1 supply observed at a supplier cell.
|
89 + | pub(in crate::client::pool) enum H1SupplyOutcome {
|
90 + | SupplierLive {
|
91 + | supplier: PartitionId,
|
92 + | revision: SupplyRevision<H1SupplyStatus>,
|
93 + | },
|
94 + | SupplierExpired {
|
95 + | supplier: PartitionId,
|
96 + | },
|
97 + | }
|
98 + |
|
99 + | impl H1SupplyOutcome {
|
100 + | pub(in crate::client::pool) fn supplier_live(
|
101 + | supplier: PartitionId,
|
102 + | revision: SupplyRevision<H1SupplyStatus>,
|
103 + | ) -> Self {
|
104 + | Self::SupplierLive { supplier, revision }
|
105 + | }
|
106 + |
|
107 + | pub(in crate::client::pool) fn supplier_expired(supplier: PartitionId) -> Self {
|
108 + | Self::SupplierExpired { supplier }
|
109 + | }
|
110 + |
|
111 + | fn supplier(&self) -> &PartitionId {
|
112 + | match self {
|
113 + | Self::SupplierLive { supplier, .. } | Self::SupplierExpired { supplier } => supplier,
|
114 + | }
|
115 + | }
|
116 + | }
|
117 + |
|
118 + | /// Admission's indexed view of supplier cells exposing selectable H1 supply.
|
119 + | #[derive(Debug, Default)]
|
120 + | struct H1SupplyIndex {
|
121 + | records: HashMap<PartitionId, H1SupplyRecord>,
|
122 + | reclaim_order: IntrusiveOrder<PartitionId>,
|
123 + | suppliers_by_group: HashMap<EligibilityGroup, IntrusiveOrder<PartitionId>>,
|
124 + | }
|
125 + |
|
126 + | /// Admission's current view of one HTTP/1 supplier cell.
|
127 + | #[derive(Debug)]
|
128 + | struct H1SupplyRecord {
|
129 + | /// Reuse group whose peers may borrow this connection cell's sender.
|
130 + | eligibility_group: EligibilityGroup,
|
131 + | /// Newest connection cell report accepted by admission.
|
132 + | revision: u64,
|
133 + | /// Whether the cell has an H1 record that can return or be reclaimed.
|
134 + | status: H1SupplyStatus,
|
135 + | reserved_by: Option<H1MatchId>,
|
136 + | index_state: H1SupplyIndexState,
|
137 + | }
|
138 + |
|
139 + | impl H1SupplyRecord {
|
140 + | /// Returns whether this connection cell must occupy both scheduling views.
|
141 + | fn is_selectable(&self) -> bool {
|
142 + | self.status.has_returnable_connection
|
143 + | && !self.status.peer_use_blocked
|
144 + | && self.reserved_by.is_none()
|
145 + | }
|
146 + | }
|
147 + |
|
148 + | /// Whether a connection cell is linked in both scheduling views.
|
149 + | #[derive(Debug, Default)]
|
150 + | enum H1SupplyIndexState {
|
151 + | /// The connection cell is absent from connection cell-selection order.
|
152 + | #[default]
|
153 + | Unlinked,
|
154 + | /// The connection cell is linked once in origin and group order.
|
155 + | Linked {
|
156 + | /// Links in origin-wide reclaim order.
|
157 + | reclaim: IntrusiveLinks<PartitionId>,
|
158 + | /// Links in eligibility-group borrow order.
|
159 + | group: IntrusiveLinks<PartitionId>,
|
160 + | },
|
161 + | }
|
162 + |
|
163 + | impl H1SupplyIndexState {
|
164 + | fn is_linked(&self) -> bool {
|
165 + | matches!(self, Self::Linked { .. })
|
166 + | }
|
167 + | }
|
168 + |
|
169 + | /// Admission-owned state for one nonterminal demand/supplier match.
|
170 + | #[derive(Clone, Debug)]
|
171 + | struct H1Match {
|
172 + | /// Cell whose connection is reserved by this match.
|
173 + | supplier: PartitionId,
|
174 + | /// Cell whose demand caused this match.
|
175 + | requester: PartitionId,
|
176 + | /// Exact requesting-cell demand generation that caused this match.
|
177 + | demand: DemandId,
|
178 + | /// Whether resolution borrows a sender or reclaims its capacity.
|
179 + | kind: H1MatchKind,
|
180 + | /// Admission-side progress of the match.
|
181 + | state: H1MatchState,
|
182 + | /// Whether requesting cell demand became stale before connection-cell
|
183 + | /// resolution.
|
184 + | cancelled: bool,
|
185 + | }
|
186 + |
|
187 + | /// Origin-side progress of one retained H1 match.
|
188 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
189 + | enum H1MatchState {
|
190 + | /// Admission selected a connection, but the cell has not reserved it.
|
191 + | Reserving,
|
192 + | /// The cell will intercept the connection's next reusable return.
|
193 + | WaitingForSender,
|
194 + | /// A provisional sender is resolving outside admission.
|
195 + | Resolving,
|
196 + | /// Request cancellation must clear the installed reservation.
|
197 + | Cancelling,
|
198 + | }
|
199 + |
|
200 + | /// Work required to reserve one supplier outside the admission lock.
|
201 + | #[derive(Debug)]
|
202 + | pub(in crate::client::pool) struct PreparedH1Reservation {
|
203 + | /// Match whose supplier is being reserved.
|
204 + | pub(in crate::client::pool) match_id: H1MatchId,
|
205 + | /// Cell whose connection was selected while admission was locked.
|
206 + | pub(in crate::client::pool) supplier: PartitionId,
|
207 + | }
|
208 + |
|
209 + | /// Work required to cancel a reservation outside the admission lock.
|
210 + | pub(super) struct PreparedH1Cancellation {
|
211 + | /// Match whose supplier reservation must be cleared.
|
212 + | match_id: H1MatchId,
|
213 + | /// Cell that owns the reservation.
|
214 + | supplier: PartitionId,
|
215 + | }
|
216 + |
|
217 + | impl H1SupplyIndex {
|
218 + | fn apply_revision(
|
219 + | &mut self,
|
220 + | supplier: PartitionId,
|
221 + | eligibility_group: EligibilityGroup,
|
222 + | revision: SupplyRevision<H1SupplyStatus>,
|
223 + | ) {
|
224 + | if self
|
225 + | .records
|
226 + | .get(&supplier)
|
227 + | .is_some_and(|record| record.revision >= revision.revision)
|
228 + | {
|
229 + | return;
|
230 + | }
|
231 + | self.unlink_supplier(&supplier);
|
232 + | let record = self
|
233 + | .records
|
234 + | .entry(supplier)
|
235 + | .or_insert_with(|| H1SupplyRecord {
|
236 + | eligibility_group: eligibility_group.clone(),
|
237 + | revision: revision.revision,
|
238 + | status: revision.status,
|
239 + | reserved_by: None,
|
240 + | index_state: H1SupplyIndexState::Unlinked,
|
241 + | });
|
242 + | record.eligibility_group = eligibility_group;
|
243 + | record.revision = revision.revision;
|
244 + | record.status = revision.status;
|
245 + | self.link_supplier_if_selectable(&supplier);
|
246 + | }
|
247 + |
|
248 + | fn select_borrow_supplier(
|
249 + | &mut self,
|
250 + | eligibility_group: &EligibilityGroup,
|
251 + | requester: PartitionId,
|
252 + | ) -> Option<PartitionId> {
|
253 + | let supplier = {
|
254 + | let order = self.suppliers_by_group.get(eligibility_group)?;
|
255 + | self.first_peer_supplier(order, requester, false)?
|
256 + | };
|
257 + | self.unlink_supplier(&supplier);
|
258 + | Some(supplier)
|
259 + | }
|
260 + |
|
261 + | fn select_peer_reclaim_supplier(&mut self, requester: PartitionId) -> Option<PartitionId> {
|
262 + | let supplier = self.first_peer_supplier(&self.reclaim_order, requester, true)?;
|
263 + | self.unlink_supplier(&supplier);
|
264 + | Some(supplier)
|
265 + | }
|
266 + |
|
267 + | fn select_oldest_reclaim_supplier(&mut self) -> Option<PartitionId> {
|
268 + | let supplier = self.reclaim_order.head()?;
|
269 + | self.unlink_supplier(&supplier);
|
270 + | Some(supplier)
|
271 + | }
|
272 + |
|
273 + | fn first_peer_supplier(
|
274 + | &self,
|
275 + | order: &IntrusiveOrder<PartitionId>,
|
276 + | requester: PartitionId,
|
277 + | reclaim: bool,
|
278 + | ) -> Option<PartitionId> {
|
279 + | let head = order.head()?;
|
280 + | if head != requester {
|
281 + | return Some(head);
|
282 + | }
|
283 + | let record = self
|
284 + | .records
|
285 + | .get(&head)
|
286 + | .expect("H1 supply order head disappeared");
|
287 + | let H1SupplyIndexState::Linked {
|
288 + | reclaim: reclaim_links,
|
289 + | group,
|
290 + | } = &record.index_state
|
291 + | else {
|
292 + | unreachable!("ordered H1 supplier was unlinked");
|
293 + | };
|
294 + | if reclaim {
|
295 + | reclaim_links.next
|
296 + | } else {
|
297 + | group.next
|
298 + | }
|
299 + | }
|
300 + |
|
301 + | fn link_supplier_if_selectable(&mut self, supplier: &PartitionId) {
|
302 + | let Some(record) = self.records.get(supplier) else {
|
303 + | return;
|
304 + | };
|
305 + | if !record.is_selectable() || record.index_state.is_linked() {
|
306 + | return;
|
307 + | }
|
308 + | let eligibility_group = record.eligibility_group.clone();
|
309 + | let reclaim = self.reclaim_order.push_back(*supplier);
|
310 + | let group = self
|
311 + | .suppliers_by_group
|
312 + | .entry(eligibility_group)
|
313 + | .or_default()
|
314 + | .push_back(*supplier);
|
315 + |
|
316 + | if let Some(previous) = reclaim.previous {
|
317 + | let H1SupplyIndexState::Linked { reclaim, .. } = &mut self
|
318 + | .records
|
319 + | .get_mut(&previous)
|
320 + | .expect("reclaim order tail disappeared")
|
321 + | .index_state
|
322 + | else {
|
323 + | unreachable!("reclaim order tail was unlinked");
|
324 + | };
|
325 + | reclaim.next = Some(*supplier);
|
326 + | }
|
327 + | if let Some(previous) = group.previous {
|
328 + | let H1SupplyIndexState::Linked { group, .. } = &mut self
|
329 + | .records
|
330 + | .get_mut(&previous)
|
331 + | .expect("group supplier order tail disappeared")
|
332 + | .index_state
|
333 + | else {
|
334 + | unreachable!("group supplier order tail was unlinked");
|
335 + | };
|
336 + | group.next = Some(*supplier);
|
337 + | }
|
338 + | self.records
|
339 + | .get_mut(supplier)
|
340 + | .expect("linked H1 supplier disappeared")
|
341 + | .index_state = H1SupplyIndexState::Linked { reclaim, group };
|
342 + | }
|
343 + |
|
344 + | fn unlink_supplier(&mut self, supplier: &PartitionId) {
|
345 + | let Some(record) = self.records.get_mut(supplier) else {
|
346 + | return;
|
347 + | };
|
348 + | let index_state = std::mem::take(&mut record.index_state);
|
349 + | let H1SupplyIndexState::Linked { reclaim, group } = index_state else {
|
350 + | return;
|
351 + | };
|
352 + | let eligibility_group = record.eligibility_group.clone();
|
353 + | self.repair_reclaim_links(supplier, &reclaim);
|
354 + | self.reclaim_order.remove(*supplier, reclaim);
|
355 + | self.repair_group_links(supplier, &group);
|
356 + | let order = self
|
357 + | .suppliers_by_group
|
358 + | .get_mut(&eligibility_group)
|
359 + | .expect("linked H1 supplier lost its group order");
|
360 + | order.remove(*supplier, group);
|
361 + | }
|
362 + |
|
363 + | fn repair_reclaim_links(
|
364 + | &mut self,
|
365 + | supplier: &PartitionId,
|
366 + | links: &IntrusiveLinks<PartitionId>,
|
367 + | ) {
|
368 + | if let Some(previous) = links.previous {
|
369 + | let H1SupplyIndexState::Linked { reclaim, .. } = &mut self
|
370 + | .records
|
371 + | .get_mut(&previous)
|
372 + | .expect("previous reclaim supplier disappeared")
|
373 + | .index_state
|
374 + | else {
|
375 + | unreachable!("previous reclaim supplier was unlinked");
|
376 + | };
|
377 + | reclaim.next = links.next;
|
378 + | }
|
379 + | if let Some(next) = links.next {
|
380 + | let H1SupplyIndexState::Linked { reclaim, .. } = &mut self
|
381 + | .records
|
382 + | .get_mut(&next)
|
383 + | .expect("next reclaim supplier disappeared")
|
384 + | .index_state
|
385 + | else {
|
386 + | unreachable!("next reclaim supplier was unlinked");
|
387 + | };
|
388 + | reclaim.previous = links.previous;
|
389 + | }
|
390 + | debug_assert_ne!(links.previous.as_ref(), Some(supplier));
|
391 + | debug_assert_ne!(links.next.as_ref(), Some(supplier));
|
392 + | }
|
393 + |
|
394 + | fn repair_group_links(&mut self, supplier: &PartitionId, links: &IntrusiveLinks<PartitionId>) {
|
395 + | if let Some(previous) = links.previous {
|
396 + | let H1SupplyIndexState::Linked { group, .. } = &mut self
|
397 + | .records
|
398 + | .get_mut(&previous)
|
399 + | .expect("previous group supplier disappeared")
|
400 + | .index_state
|
401 + | else {
|
402 + | unreachable!("previous group supplier was unlinked");
|
403 + | };
|
404 + | group.next = links.next;
|
405 + | }
|
406 + | if let Some(next) = links.next {
|
407 + | let H1SupplyIndexState::Linked { group, .. } = &mut self
|
408 + | .records
|
409 + | .get_mut(&next)
|
410 + | .expect("next group supplier disappeared")
|
411 + | .index_state
|
412 + | else {
|
413 + | unreachable!("next group supplier was unlinked");
|
414 + | };
|
415 + | group.previous = links.previous;
|
416 + | }
|
417 + | debug_assert_ne!(links.previous.as_ref(), Some(supplier));
|
418 + | debug_assert_ne!(links.next.as_ref(), Some(supplier));
|
419 + | }
|
420 + |
|
421 + | #[cfg(any(debug_assertions, test))]
|
422 + | fn assert_consistent(&self) {
|
423 + | for record in self.records.values() {
|
424 + | assert_eq!(
|
425 + | record.index_state.is_linked(),
|
426 + | record.is_selectable(),
|
427 + | "H1 supplier index state did not match selectability"
|
428 + | );
|
429 + | }
|
430 + | self.assert_order(&self.reclaim_order, None, true);
|
431 + | for (group, order) in &self.suppliers_by_group {
|
432 + | self.assert_order(order, Some(group), false);
|
433 + | }
|
434 + | }
|
435 + |
|
436 + | #[cfg(any(debug_assertions, test))]
|
437 + | fn assert_order(
|
438 + | &self,
|
439 + | order: &IntrusiveOrder<PartitionId>,
|
440 + | expected_group: Option<&EligibilityGroup>,
|
441 + | reclaim_order: bool,
|
442 + | ) {
|
443 + | let expected = self
|
444 + | .records
|
445 + | .values()
|
446 + | .filter(|record| {
|
447 + | record.index_state.is_linked()
|
448 + | && expected_group.is_none_or(|group| &record.eligibility_group == group)
|
449 + | })
|
450 + | .count();
|
451 + | order.assert_consistent(
|
452 + | expected,
|
453 + | self.records.len(),
|
454 + | "HTTP/1 supplier order",
|
455 + | |supplier| {
|
456 + | let record = self
|
457 + | .records
|
458 + | .get(&supplier)
|
459 + | .expect("ordered H1 supplier disappeared");
|
460 + | if let Some(group) = expected_group {
|
461 + | assert_eq!(&record.eligibility_group, group);
|
462 + | }
|
463 + | let H1SupplyIndexState::Linked { reclaim, group } = &record.index_state else {
|
464 + | unreachable!("ordered H1 supplier was unlinked");
|
465 + | };
|
466 + | if reclaim_order {
|
467 + | *reclaim
|
468 + | } else {
|
469 + | *group
|
470 + | }
|
471 + | },
|
472 + | );
|
473 + | }
|
474 + | }
|
475 + |
|
476 + | impl H1Supply {
|
477 + | fn apply_revision(
|
478 + | &mut self,
|
479 + | supplier: PartitionId,
|
480 + | eligibility_group: EligibilityGroup,
|
481 + | revision: SupplyRevision<H1SupplyStatus>,
|
482 + | ) {
|
483 + | self.index
|
484 + | .apply_revision(supplier, eligibility_group, revision);
|
485 + | self.assert_consistent();
|
486 + | }
|
487 + |
|
488 + | pub(super) fn reconcile_requester(&mut self, requester: &PartitionId, demand: &DemandSchedule) {
|
489 + | let Some(match_id) = self.by_requester.get(requester).copied() else {
|
490 + | return;
|
491 + | };
|
492 + | let Some(retained) = self.matches.get_mut(&match_id) else {
|
493 + | self.by_requester.remove(requester);
|
494 + | self.assert_consistent();
|
495 + | return;
|
496 + | };
|
497 + | if demand.is_current_queued(&retained.requester, retained.demand) {
|
498 + | return;
|
499 + | }
|
500 + | retained.cancelled = true;
|
501 + | if retained.state == H1MatchState::WaitingForSender {
|
502 + | retained.state = H1MatchState::Cancelling;
|
503 + | self.cancellations.push_back(match_id);
|
504 + | }
|
505 + | self.assert_consistent();
|
506 + | }
|
507 + |
|
508 + | pub(super) fn prepare_match(
|
509 + | &mut self,
|
510 + | demand: &DemandSchedule,
|
511 + | ) -> Option<PreparedH1Reservation> {
|
512 + | let queued = demand.queued_head()?;
|
513 + | if self.by_requester.contains_key(&queued.requester) {
|
514 + | return None;
|
515 + | }
|
516 + | let (supplier, kind) = if queued.requirement.accepts_h1() {
|
517 + | match self
|
518 + | .index
|
519 + | .select_borrow_supplier(&queued.eligibility_group, queued.requester)
|
520 + | {
|
521 + | Some(supplier) => (supplier, H1MatchKind::BorrowSender),
|
522 + | None => (
|
523 + | self.index.select_peer_reclaim_supplier(queued.requester)?,
|
524 + | H1MatchKind::ReclaimCapacity,
|
525 + | ),
|
526 + | }
|
527 + | } else {
|
528 + | (
|
529 + | self.index.select_oldest_reclaim_supplier()?,
|
530 + | H1MatchKind::ReclaimCapacity,
|
531 + | )
|
532 + | };
|
533 + | let match_id = self.take_match_id();
|
534 + | let record = self
|
535 + | .index
|
536 + | .records
|
537 + | .get_mut(&supplier)
|
538 + | .expect("selected H1 supplier disappeared");
|
539 + | debug_assert!(record.reserved_by.is_none());
|
540 + | record.reserved_by = Some(match_id);
|
541 + | self.by_requester.insert(queued.requester, match_id);
|
542 + | self.matches.insert(
|
543 + | match_id,
|
544 + | H1Match {
|
545 + | supplier,
|
546 + | requester: queued.requester,
|
547 + | demand: queued.demand,
|
548 + | kind,
|
549 + | state: H1MatchState::Reserving,
|
550 + | cancelled: false,
|
551 + | },
|
552 + | );
|
553 + | self.assert_consistent();
|
554 + | Some(PreparedH1Reservation { match_id, supplier })
|
555 + | }
|
556 + |
|
557 + | pub(super) fn prepare_cancellation(&mut self) -> Option<PreparedH1Cancellation> {
|
558 + | while let Some(match_id) = self.cancellations.pop_front() {
|
559 + | let Some(retained) = self.matches.get(&match_id) else {
|
560 + | continue;
|
561 + | };
|
562 + | if retained.state != H1MatchState::Cancelling {
|
563 + | continue;
|
564 + | }
|
565 + | return Some(PreparedH1Cancellation {
|
566 + | match_id,
|
567 + | supplier: retained.supplier,
|
568 + | });
|
569 + | }
|
570 + | None
|
571 + | }
|
572 + |
|
573 + | fn settle_reservation(&mut self, match_id: H1MatchId, resolved: bool) -> Option<H1Match> {
|
574 + | let retained = self.matches.get_mut(&match_id)?;
|
575 + | if retained.state != H1MatchState::Reserving {
|
576 + | return None;
|
577 + | }
|
578 + | retained.state = if resolved {
|
579 + | H1MatchState::Resolving
|
580 + | } else if retained.cancelled {
|
581 + | H1MatchState::Cancelling
|
582 + | } else {
|
583 + | H1MatchState::WaitingForSender
|
584 + | };
|
585 + | if retained.state == H1MatchState::Cancelling {
|
586 + | self.cancellations.push_back(match_id);
|
587 + | }
|
588 + | let retained = retained.clone();
|
589 + | self.assert_consistent();
|
590 + | Some(retained)
|
591 + | }
|
592 + |
|
593 + | fn begin_resolution(&mut self, match_id: H1MatchId) -> Option<H1Match> {
|
594 + | let retained = self.matches.get_mut(&match_id)?;
|
595 + | if !matches!(
|
596 + | retained.state,
|
597 + | H1MatchState::WaitingForSender | H1MatchState::Resolving | H1MatchState::Cancelling
|
598 + | ) {
|
599 + | return None;
|
600 + | }
|
601 + | retained.state = H1MatchState::Resolving;
|
602 + | let retained = retained.clone();
|
603 + | self.assert_consistent();
|
604 + | Some(retained)
|
605 + | }
|
606 + |
|
607 + | fn settle_match(&mut self, match_id: H1MatchId, outcome: H1SupplyOutcome) -> Option<H1Match> {
|
608 + | let outcome_supplier = *outcome.supplier();
|
609 + | self.apply_h1_supply_outcome(outcome);
|
610 + | let retained = self.matches.remove(&match_id);
|
611 + | if let Some(retained) = retained.as_ref() {
|
612 + | debug_assert_eq!(retained.supplier, outcome_supplier);
|
613 + | if self.by_requester.get(&retained.requester) == Some(&match_id) {
|
614 + | self.by_requester.remove(&retained.requester);
|
615 + | }
|
616 + | if let Some(record) = self.index.records.get_mut(&retained.supplier) {
|
617 + | if record.reserved_by == Some(match_id) {
|
618 + | record.reserved_by = None;
|
619 + | }
|
620 + | }
|
621 + | }
|
622 + | self.index.link_supplier_if_selectable(&outcome_supplier);
|
623 + | self.assert_consistent();
|
624 + | retained
|
625 + | }
|
626 + |
|
627 + | fn apply_h1_supply_outcome(&mut self, outcome: H1SupplyOutcome) {
|
628 + | match outcome {
|
629 + | H1SupplyOutcome::SupplierLive { supplier, revision } => {
|
630 + | let Some(eligibility_group) = self
|
631 + | .index
|
632 + | .records
|
633 + | .get(&supplier)
|
634 + | .map(|record| record.eligibility_group.clone())
|
635 + | else {
|
636 + | return;
|
637 + | };
|
638 + | self.index
|
639 + | .apply_revision(supplier, eligibility_group, revision);
|
640 + | }
|
641 + | H1SupplyOutcome::SupplierExpired { supplier } => {
|
642 + | self.index.unlink_supplier(&supplier);
|
643 + | self.index.records.remove(&supplier);
|
644 + | }
|
645 + | }
|
646 + | }
|
647 + |
|
648 + | fn take_match_id(&mut self) -> H1MatchId {
|
649 + | let value = self.next_match_id;
|
650 + | self.next_match_id = value
|
651 + | .checked_add(1)
|
652 + | .expect("HTTP/1 match identity exhausted");
|
653 + | H1MatchId(value)
|
654 + | }
|
655 + |
|
656 + | fn assert_consistent(&self) {
|
657 + | #[cfg(any(debug_assertions, test))]
|
658 + | {
|
659 + | if std::thread::panicking() {
|
660 + | return;
|
661 + | }
|
662 + | self.index.assert_consistent();
|
663 + | for (match_id, retained) in &self.matches {
|
664 + | assert_eq!(self.by_requester.get(&retained.requester), Some(match_id));
|
665 + | assert_eq!(
|
666 + | self.index
|
667 + | .records
|
668 + | .get(&retained.supplier)
|
669 + | .and_then(|record| record.reserved_by),
|
670 + | Some(*match_id),
|
671 + | "H1 supplier back-pointer did not name its retained match"
|
672 + | );
|
673 + | }
|
674 + | for (requester, match_id) in &self.by_requester {
|
675 + | assert_eq!(
|
676 + | self.matches
|
677 + | .get(match_id)
|
678 + | .map(|retained| &retained.requester),
|
679 + | Some(requester),
|
680 + | "H1 requester index named a missing match"
|
681 + | );
|
682 + | }
|
683 + | for (supplier, record) in &self.index.records {
|
684 + | if let Some(match_id) = record.reserved_by {
|
685 + | assert_eq!(
|
686 + | self.matches
|
687 + | .get(&match_id)
|
688 + | .map(|retained| &retained.supplier),
|
689 + | Some(supplier),
|
690 + | "H1 supplier record named a missing match"
|
691 + | );
|
692 + | }
|
693 + | }
|
694 + | for (index, match_id) in self.cancellations.iter().enumerate() {
|
695 + | if let Some(retained) = self.matches.get(match_id) {
|
696 + | assert!(
|
697 + | retained.cancelled
|
698 + | && matches!(
|
699 + | retained.state,
|
700 + | H1MatchState::Cancelling | H1MatchState::Resolving
|
701 + | ),
|
702 + | "live H1 cancellation named a match outside cancellation or resolution"
|
703 + | );
|
704 + | }
|
705 + | assert!(
|
706 + | !self
|
707 + | .cancellations
|
708 + | .iter()
|
709 + | .skip(index + 1)
|
710 + | .any(|id| id == match_id),
|
711 + | "H1 cancellation was queued more than once"
|
712 + | );
|
713 + | }
|
714 + | }
|
715 + | }
|
716 + | }
|
717 + |
|
718 + | /// Reservation crossing from admission to one supplier cell.
|
719 + | pub(in crate::client::pool) struct H1ReservationAction {
|
720 + | admission: Arc<OriginAdmission>,
|
721 + | prepared: Option<PreparedH1Reservation>,
|
722 + | }
|
723 + |
|
724 + | impl H1ReservationAction {
|
725 + | pub(super) fn new(admission: Arc<OriginAdmission>, prepared: PreparedH1Reservation) -> Self {
|
726 + | Self {
|
727 + | admission,
|
728 + | prepared: Some(prepared),
|
729 + | }
|
730 + | }
|
731 + |
|
732 + | pub(super) fn reserve_supplier(mut self) -> Option<AdmissionAction> {
|
733 + | let prepared = self
|
734 + | .prepared
|
735 + | .take()
|
736 + | .expect("HTTP/1 reservation consumed more than once");
|
737 + | let Some(supplier) = self.admission.cell(&prepared.supplier) else {
|
738 + | return OriginAdmission::settle_h1_match(
|
739 + | &self.admission,
|
740 + | prepared.match_id,
|
741 + | H1SupplyOutcome::supplier_expired(prepared.supplier),
|
742 + | );
|
743 + | };
|
744 + | OriginCell::commit_h1_reservation(&supplier, self.admission.clone(), prepared)
|
745 + | }
|
746 + | }
|
747 + |
|
748 + | impl Drop for H1ReservationAction {
|
749 + | fn drop(&mut self) {
|
750 + | if let Some(prepared) = self.prepared.take() {
|
751 + | let outcome = h1_supply_outcome(&self.admission, &prepared.supplier, |supplier| {
|
752 + | supplier.cancel_h1_reservation(prepared.match_id)
|
753 + | });
|
754 + | let next =
|
755 + | OriginAdmission::settle_h1_match(&self.admission, prepared.match_id, outcome);
|
756 + | OriginAdmission::run_action_chain(next);
|
757 + | }
|
758 + | }
|
759 + | }
|
760 + |
|
761 + | /// Cancellation crossing for an installed supplier reservation.
|
762 + | pub(in crate::client::pool) struct H1CancellationAction {
|
763 + | admission: Arc<OriginAdmission>,
|
764 + | prepared: Option<PreparedH1Cancellation>,
|
765 + | }
|
766 + |
|
767 + | impl H1CancellationAction {
|
768 + | pub(super) fn new(admission: Arc<OriginAdmission>, prepared: PreparedH1Cancellation) -> Self {
|
769 + | Self {
|
770 + | admission,
|
771 + | prepared: Some(prepared),
|
772 + | }
|
773 + | }
|
774 + |
|
775 + | pub(super) fn cancel_reservation(mut self) -> Option<AdmissionAction> {
|
776 + | let prepared = self
|
777 + | .prepared
|
778 + | .take()
|
779 + | .expect("HTTP/1 cancellation consumed more than once");
|
780 + | let outcome = h1_supply_outcome(&self.admission, &prepared.supplier, |supplier| {
|
781 + | supplier.cancel_h1_reservation(prepared.match_id)
|
782 + | });
|
783 + | OriginAdmission::settle_h1_match(&self.admission, prepared.match_id, outcome)
|
784 + | }
|
785 + | }
|
786 + |
|
787 + | impl Drop for H1CancellationAction {
|
788 + | fn drop(&mut self) {
|
789 + | if let Some(prepared) = self.prepared.take() {
|
790 + | let outcome = h1_supply_outcome(&self.admission, &prepared.supplier, |supplier| {
|
791 + | supplier.cancel_h1_reservation(prepared.match_id)
|
792 + | });
|
793 + | let next =
|
794 + | OriginAdmission::settle_h1_match(&self.admission, prepared.match_id, outcome);
|
795 + | OriginAdmission::run_action_chain(next);
|
796 + | }
|
797 + | }
|
798 + | }
|
799 + |
|
800 + | /// Provisional sender owned by one resolving H1 match.
|
801 + | pub(in crate::client::pool) struct H1Candidate {
|
802 + | admission: Arc<OriginAdmission>,
|
803 + | match_id: H1MatchId,
|
804 + | supplier: PartitionId,
|
805 + | provisional: Option<ProvisionalH1>,
|
806 + | }
|
807 + |
|
808 + | impl H1Candidate {
|
809 + | pub(in crate::client::pool) fn new(
|
810 + | admission: Arc<OriginAdmission>,
|
811 + | match_id: H1MatchId,
|
812 + | supplier: PartitionId,
|
813 + | provisional: ProvisionalH1,
|
814 + | ) -> Self {
|
815 + | Self {
|
816 + | admission,
|
817 + | match_id,
|
818 + | supplier,
|
819 + | provisional: Some(provisional),
|
820 + | }
|
821 + | }
|
822 + |
|
823 + | pub(super) fn connection_id(&self) -> ConnectionId {
|
824 + | self.provisional
|
825 + | .as_ref()
|
826 + | .expect("HTTP/1 candidate consumed more than once")
|
827 + | .connection_id()
|
828 + | }
|
829 + |
|
830 + | pub(in crate::client::pool) fn commit(mut self) -> Result<H1Selection, Self> {
|
831 + | let Some(supplier) = self.admission.cell(&self.supplier) else {
|
832 + | return Err(self);
|
833 + | };
|
834 + | let provisional = self
|
835 + | .provisional
|
836 + | .take()
|
837 + | .expect("HTTP/1 candidate consumed more than once");
|
838 + | match OriginCell::commit_h1_match(&supplier, self.match_id, provisional) {
|
839 + | Ok(selection) => Ok(selection),
|
840 + | Err(provisional) => {
|
841 + | self.provisional = Some(provisional);
|
842 + | Err(self)
|
843 + | }
|
844 + | }
|
845 + | }
|
846 + |
|
847 + | /// Returns the provisional sender to its supplier without independently
|
848 + | /// running the admission action chain.
|
849 + | pub(super) fn reject(mut self) -> H1SupplyOutcome {
|
850 + | let Some(provisional) = self.provisional.take() else {
|
851 + | unreachable!("HTTP/1 candidate consumed more than once");
|
852 + | };
|
853 + | match self.admission.cell(&self.supplier) {
|
854 + | Some(supplier) => H1SupplyOutcome::supplier_live(
|
855 + | self.supplier,
|
856 + | OriginCell::reject_h1_match(&supplier, self.match_id, provisional),
|
857 + | ),
|
858 + | None => {
|
859 + | drop(provisional);
|
860 + | H1SupplyOutcome::supplier_expired(self.supplier)
|
861 + | }
|
862 + | }
|
863 + | }
|
864 + |
|
865 + | fn close_for_capacity(mut self) -> (H1SupplyOutcome, bool) {
|
866 + | let Some(supplier) = self.admission.cell(&self.supplier) else {
|
867 + | drop(self.provisional.take());
|
868 + | return (H1SupplyOutcome::supplier_expired(self.supplier), false);
|
869 + | };
|
870 + | let provisional = self
|
871 + | .provisional
|
872 + | .take()
|
873 + | .expect("HTTP/1 candidate consumed more than once");
|
874 + | let (revision, reclaimed) =
|
875 + | match OriginCell::reclaim_h1_candidate(&supplier, self.match_id, provisional) {
|
876 + | Ok(result) => result,
|
877 + | Err(provisional) => (
|
878 + | OriginCell::reject_h1_match(&supplier, self.match_id, provisional),
|
879 + | false,
|
880 + | ),
|
881 + | };
|
882 + | (
|
883 + | H1SupplyOutcome::supplier_live(self.supplier, revision),
|
884 + | reclaimed,
|
885 + | )
|
886 + | }
|
887 + | }
|
888 + |
|
889 + | impl fmt::Debug for H1Candidate {
|
890 + | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
891 + | f.debug_struct("H1Candidate")
|
892 + | .field("match_id", &self.match_id)
|
893 + | .field("supplier", &self.supplier)
|
894 + | .field("provisional", &self.provisional)
|
895 + | .finish_non_exhaustive()
|
896 + | }
|
897 + | }
|
898 + |
|
899 + | impl Drop for H1Candidate {
|
900 + | fn drop(&mut self) {
|
901 + | let Some(provisional) = self.provisional.take() else {
|
902 + | return;
|
903 + | };
|
904 + | let outcome = match self.admission.cell(&self.supplier) {
|
905 + | Some(supplier) => H1SupplyOutcome::supplier_live(
|
906 + | self.supplier,
|
907 + | OriginCell::reject_h1_match(&supplier, self.match_id, provisional),
|
908 + | ),
|
909 + | None => {
|
910 + | drop(provisional);
|
911 + | H1SupplyOutcome::supplier_expired(self.supplier)
|
912 + | }
|
913 + | };
|
914 + | let next = OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome);
|
915 + | OriginAdmission::run_action_chain(next);
|
916 + | }
|
917 + | }
|
918 + |
|
919 + | /// Exact provisional H1 sender selected to close for bounded capacity.
|
920 + | pub(in crate::client::pool) struct H1CapacityReclaim {
|
921 + | admission: Arc<OriginAdmission>,
|
922 + | requester: PartitionId,
|
923 + | match_id: H1MatchId,
|
924 + | candidate: Option<H1Candidate>,
|
925 + | }
|
926 + |
|
927 + | impl H1CapacityReclaim {
|
928 + | pub(super) fn new(
|
929 + | admission: Arc<OriginAdmission>,
|
930 + | requester: PartitionId,
|
931 + | match_id: H1MatchId,
|
932 + | candidate: H1Candidate,
|
933 + | ) -> Self {
|
934 + | Self {
|
935 + | admission,
|
936 + | requester,
|
937 + | match_id,
|
938 + | candidate: Some(candidate),
|
939 + | }
|
940 + | }
|
941 + |
|
942 + | pub(super) fn reclaim_capacity(mut self) -> Option<AdmissionAction> {
|
943 + | let candidate = self
|
944 + | .candidate
|
945 + | .take()
|
946 + | .expect("HTTP/1 capacity reclaim consumed more than once");
|
947 + | let connection_id = candidate.connection_id();
|
948 + | let supplier = candidate.supplier;
|
949 + | let (outcome, reclaimed) = candidate.close_for_capacity();
|
950 + | if reclaimed {
|
951 + | tracing::trace!(
|
952 + | connection_id = %connection_id,
|
953 + | request_partition = ?self.requester,
|
954 + | connection_partition = ?supplier,
|
955 + | origin_scheme = %self.admission.origin().scheme(),
|
956 + | origin_host = self.admission.origin().host(),
|
957 + | origin_port = ?self.admission.origin().port(),
|
958 + | "HTTP/1 connection reclaimed for peer demand"
|
959 + | );
|
960 + | }
|
961 + | OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome)
|
962 + | }
|
963 + | }
|
964 + |
|
965 + | /// Supplier-cell settlement after an irreversible sender transfer.
|
966 + | pub(in crate::client::pool) struct H1SupplierSettlement {
|
967 + | admission: Arc<OriginAdmission>,
|
968 + | match_id: H1MatchId,
|
969 + | supplier: PartitionId,
|
970 + | transferred: bool,
|
971 + | active: bool,
|
972 + | }
|
973 + |
|
974 + | impl H1SupplierSettlement {
|
975 + | pub(super) fn new(
|
976 + | admission: Arc<OriginAdmission>,
|
977 + | match_id: H1MatchId,
|
978 + | supplier: PartitionId,
|
979 + | transferred: bool,
|
980 + | ) -> Self {
|
981 + | Self {
|
982 + | admission,
|
983 + | match_id,
|
984 + | supplier,
|
985 + | transferred,
|
986 + | active: true,
|
987 + | }
|
988 + | }
|
989 + |
|
990 + | pub(super) fn settle_supplier(mut self) -> Option<AdmissionAction> {
|
991 + | let outcome = h1_supply_outcome(&self.admission, &self.supplier, |supplier| {
|
992 + | supplier.complete_h1_match(self.match_id, self.transferred)
|
993 + | });
|
994 + | let action = OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome);
|
995 + | self.active = false;
|
996 + | action
|
997 + | }
|
998 + | }
|
999 + |
|
1000 + | impl Drop for H1SupplierSettlement {
|
1001 + | fn drop(&mut self) {
|
1002 + | if !self.active {
|
1003 + | return;
|
1004 + | }
|
1005 + | let outcome = h1_supply_outcome(&self.admission, &self.supplier, |supplier| {
|
1006 + | supplier.complete_h1_match(self.match_id, self.transferred)
|
1007 + | });
|
1008 + | let next = OriginAdmission::settle_h1_match(&self.admission, self.match_id, outcome);
|
1009 + | OriginAdmission::run_action_chain(next);
|
1010 + | }
|
1011 + | }
|
1012 + |
|
1013 + | fn h1_supply_outcome(
|
1014 + | admission: &OriginAdmission,
|
1015 + | supplier: &PartitionId,
|
1016 + | current: impl FnOnce(&Arc<OriginCell>) -> SupplyRevision<H1SupplyStatus>,
|
1017 + | ) -> H1SupplyOutcome {
|
1018 + | match admission.cell(supplier) {
|
1019 + | Some(cell) => H1SupplyOutcome::supplier_live(*supplier, current(&cell)),
|
1020 + | None => H1SupplyOutcome::supplier_expired(*supplier),
|
1021 + | }
|
1022 + | }
|
1023 + |
|
1024 + | pub(super) fn apply_supply_revision(
|
1025 + | admission: &Arc<OriginAdmission>,
|
1026 + | supplier: PartitionId,
|
1027 + | eligibility_group: EligibilityGroup,
|
1028 + | revision: SupplyRevision<H1SupplyStatus>,
|
1029 + | ) {
|
1030 + | let action = {
|
1031 + | let mut state = admission.state.lock();
|
1032 + | state
|
1033 + | .h1_supply
|
1034 + | .apply_revision(supplier, eligibility_group, revision);
|
1035 + | OriginAdmission::prepare_action(admission, &mut state)
|
1036 + | };
|
1037 + | OriginAdmission::run_action_chain(action);
|
1038 + | }
|
1039 + |
|
1040 + | pub(super) fn reject_returned_match(
|
1041 + | admission: &Arc<OriginAdmission>,
|
1042 + | match_id: H1MatchId,
|
1043 + | supplier: PartitionId,
|
1044 + | revision: SupplyRevision<H1SupplyStatus>,
|
1045 + | ) {
|
1046 + | let action = settle_match(
|
1047 + | admission,
|
1048 + | match_id,
|
1049 + | H1SupplyOutcome::supplier_live(supplier, revision),
|
1050 + | );
|
1051 + | OriginAdmission::run_action_chain(action);
|
1052 + | }
|
1053 + |
|
1054 + | pub(super) fn settle_reservation(
|
1055 + | admission: &Arc<OriginAdmission>,
|
1056 + | match_id: H1MatchId,
|
1057 + | supplier: PartitionId,
|
1058 + | decision: H1ReservationDecision<H1Candidate>,
|
1059 + | ) -> Option<AdmissionAction> {
|
1060 + | match decision {
|
1061 + | H1ReservationDecision::Rejected(revision) => settle_match(
|
1062 + | admission,
|
1063 + | match_id,
|
1064 + | H1SupplyOutcome::supplier_live(supplier, revision),
|
1065 + | ),
|
1066 + | H1ReservationDecision::Installed => {
|
1067 + | let mut state = admission.state.lock();
|
1068 + | state.h1_supply.settle_reservation(match_id, false);
|
1069 + | OriginAdmission::prepare_action(admission, &mut state)
|
1070 + | }
|
1071 + | H1ReservationDecision::Candidate(candidate) => {
|
1072 + | {
|
1073 + | let mut state = admission.state.lock();
|
1074 + | if state.h1_supply.settle_reservation(match_id, true).is_none() {
|
1075 + | drop(state);
|
1076 + | return reject_candidate(admission, match_id, candidate);
|
1077 + | }
|
1078 + | }
|
1079 + | resolve_match(admission, match_id, candidate)
|
1080 + | }
|
1081 + | }
|
1082 + | }
|
1083 + |
|
1084 + | pub(super) fn resolve_match(
|
1085 + | admission: &Arc<OriginAdmission>,
|
1086 + | match_id: H1MatchId,
|
1087 + | candidate: H1Candidate,
|
1088 + | ) -> Option<AdmissionAction> {
|
1089 + | let mut state = admission.state.lock();
|
1090 + | let Some(retained) = state.h1_supply.begin_resolution(match_id) else {
|
1091 + | drop(state);
|
1092 + | return reject_candidate(admission, match_id, candidate);
|
1093 + | };
|
1094 + | if retained.cancelled
|
1095 + | || !state
|
1096 + | .demand
|
1097 + | .is_current_queued(&retained.requester, retained.demand)
|
1098 + | {
|
1099 + | drop(state);
|
1100 + | return reject_candidate(admission, match_id, candidate);
|
1101 + | }
|
1102 + | match retained.kind {
|
1103 + | H1MatchKind::BorrowSender => {
|
1104 + | let assignment_id = state.take_assignment_id();
|
1105 + | let old_group = state.demand.group_for(&retained.requester);
|
1106 + | let Some(assignment) = state.demand.prepare_h1_assignment(
|
1107 + | &retained.requester,
|
1108 + | retained.demand,
|
1109 + | assignment_id,
|
1110 + | ) else {
|
1111 + | drop(state);
|
1112 + | return reject_candidate(admission, match_id, candidate);
|
1113 + | };
|
1114 + | state.reconcile_demand_indexes(&assignment.requester, old_group);
|
1115 + | Some(AdmissionAction::Deliver(DeliveryGuard::borrowed_h1(
|
1116 + | admission.clone(),
|
1117 + | assignment,
|
1118 + | match_id,
|
1119 + | retained.supplier,
|
1120 + | candidate,
|
1121 + | )))
|
1122 + | }
|
1123 + | H1MatchKind::ReclaimCapacity => Some(AdmissionAction::ReclaimCapacity(
|
1124 + | super::CapacityReclaim::FromH1(H1CapacityReclaim::new(
|
1125 + | admission.clone(),
|
1126 + | retained.requester,
|
1127 + | match_id,
|
1128 + | candidate,
|
1129 + | )),
|
1130 + | )),
|
1131 + | }
|
1132 + | }
|
1133 + |
|
1134 + | pub(super) fn settle_match(
|
1135 + | admission: &Arc<OriginAdmission>,
|
1136 + | match_id: H1MatchId,
|
1137 + | outcome: H1SupplyOutcome,
|
1138 + | ) -> Option<AdmissionAction> {
|
1139 + | let mut state = admission.state.lock();
|
1140 + | state.h1_supply.settle_match(match_id, outcome);
|
1141 + | OriginAdmission::prepare_action(admission, &mut state)
|
1142 + | }
|
1143 + |
|
1144 + | /// Returns a provisional sender and keeps successor work in the caller's
|
1145 + | /// action chain instead of re-entering it through `H1Candidate::drop`.
|
1146 + | fn reject_candidate(
|
1147 + | admission: &Arc<OriginAdmission>,
|
1148 + | match_id: H1MatchId,
|
1149 + | candidate: H1Candidate,
|
1150 + | ) -> Option<AdmissionAction> {
|
1151 + | let outcome = candidate.reject();
|
1152 + | settle_match(admission, match_id, outcome)
|
1153 + | }
|
1154 + |
|
1155 + | pub(super) fn settle_borrow_delivery(
|
1156 + | admission: &Arc<OriginAdmission>,
|
1157 + | match_id: H1MatchId,
|
1158 + | assignment: &super::DemandAssignment,
|
1159 + | outcome: DemandAssignmentOutcome,
|
1160 + | transferred_supplier: Option<PartitionId>,
|
1161 + | refused_outcome: Option<H1SupplyOutcome>,
|
1162 + | ) -> Option<AdmissionAction> {
|
1163 + | let mut state = admission.state.lock();
|
1164 + | state.settle_assignment(assignment, outcome);
|
1165 + | if let Some(supplier) = transferred_supplier {
|
1166 + | return Some(AdmissionAction::SettleH1Supplier(
|
1167 + | H1SupplierSettlement::new(admission.clone(), match_id, supplier, true),
|
1168 + | ));
|
1169 + | }
|
1170 + | state.h1_supply.settle_match(
|
1171 + | match_id,
|
1172 + | refused_outcome.expect("refused H1 borrow had no supplier outcome"),
|
1173 + | );
|
1174 + | OriginAdmission::prepare_action(admission, &mut state)
|
1175 + | }
|
1176 + |
|
1177 + | #[cfg(all(test, not(smithy_http_client_loom)))]
|
1178 + | mod tests {
|
1179 + | use super::*;
|
1180 + | use crate::client::pool::admission::{DemandSnapshot, ProtocolRequirement, SnapshotVersion};
|
1181 + | use crate::client::pool::partition::PartitionId;
|
1182 + |
|
1183 + | fn cell(index: usize) -> PartitionId {
|
1184 + | PartitionId::from_index(index)
|
1185 + | }
|
1186 + |
|
1187 + | fn schedule(requesting_partition: PartitionId, group: EligibilityGroup) -> DemandSchedule {
|
1188 + | let mut schedule = DemandSchedule::default();
|
1189 + | schedule.apply_snapshot(
|
1190 + | requesting_partition,
|
1191 + | DemandSnapshot::active(
|
1192 + | DemandId::from_u64(1),
|
1193 + | SnapshotVersion::INITIAL,
|
1194 + | ProtocolRequirement::H1Compatible,
|
1195 + | group,
|
1196 + | ),
|
1197 + | );
|
1198 + | schedule
|
1199 + | }
|
1200 + |
|
1201 + | fn supply_revision(
|
1202 + | revision: u64,
|
1203 + | has_returnable_connection: bool,
|
1204 + | peer_use_blocked: bool,
|
1205 + | ) -> SupplyRevision<H1SupplyStatus> {
|
1206 + | SupplyRevision::new(
|
1207 + | revision,
|
1208 + | H1SupplyStatus {
|
1209 + | has_returnable_connection,
|
1210 + | peer_use_blocked,
|
1211 + | },
|
1212 + | )
|
1213 + | }
|
1214 + |
|
1215 + | #[test]
|
1216 + | fn supply_revision_keeps_one_bounded_residence() {
|
1217 + | let supplier = cell(1);
|
1218 + | let group = EligibilityGroup::Pool;
|
1219 + | let mut supply = H1Supply::default();
|
1220 + |
|
1221 + | for revision in 1..=500 {
|
1222 + | supply.apply_revision(
|
1223 + | supplier,
|
1224 + | group.clone(),
|
1225 + | supply_revision(revision, true, false),
|
1226 + | );
|
1227 + | }
|
1228 + |
|
1229 + | assert_eq!(1, supply.index.records.len());
|
1230 + | assert_eq!(1, supply.index.reclaim_order.len());
|
1231 + | assert_eq!(
|
1232 + | 1,
|
1233 + | supply
|
1234 + | .index
|
1235 + | .suppliers_by_group
|
1236 + | .get(&group)
|
1237 + | .expect("supplier group was not indexed")
|
1238 + | .len()
|
1239 + | );
|
1240 + |
|
1241 + | supply.apply_revision(supplier, group.clone(), supply_revision(501, false, false));
|
1242 + | assert_eq!(0, supply.index.reclaim_order.len());
|
1243 + | assert_eq!(
|
1244 + | 0,
|
1245 + | supply
|
1246 + | .index
|
1247 + | .suppliers_by_group
|
1248 + | .get(&group)
|
1249 + | .expect("supplier group disappeared")
|
1250 + | .len()
|
1251 + | );
|
1252 + | }
|
1253 + |
|
1254 + | #[test]
|
1255 + | fn h2_required_demand_reclaims_local_h1_capacity() {
|
1256 + | let requesting_partition = cell(1);
|
1257 + | let group = EligibilityGroup::Pool;
|
1258 + | let mut schedule = DemandSchedule::default();
|
1259 + | schedule.apply_snapshot(
|
1260 + | requesting_partition,
|
1261 + | DemandSnapshot::active(
|
1262 + | DemandId::from_u64(1),
|
1263 + | SnapshotVersion::INITIAL,
|
1264 + | ProtocolRequirement::H2Required,
|
1265 + | group.clone(),
|
1266 + | ),
|
1267 + | );
|
1268 + | let mut supply = H1Supply::default();
|
1269 + | supply.apply_revision(requesting_partition, group, supply_revision(1, true, false));
|
1270 + |
|
1271 + | let prepared = supply
|
1272 + | .prepare_match(&schedule)
|
1273 + | .expect("local HTTP/1 capacity was not selected for reclaim");
|
1274 + |
|
1275 + | assert_eq!(requesting_partition, prepared.supplier);
|
1276 + | assert_eq!(
|
1277 + | H1MatchKind::ReclaimCapacity,
|
1278 + | supply.matches[&prepared.match_id].kind
|
1279 + | );
|
1280 + | }
|
1281 + |
|
1282 + | #[test]
|
1283 + | fn borrow_supplier_selection_skips_the_requesting_cell() {
|
1284 + | let requesting_partition = cell(1);
|
1285 + | let peer = cell(2);
|
1286 + | let group = EligibilityGroup::Pool;
|
1287 + | let mut supply = H1Supply::default();
|
1288 + | supply.apply_revision(
|
1289 + | requesting_partition,
|
1290 + | group.clone(),
|
1291 + | supply_revision(1, true, false),
|
1292 + | );
|
1293 + | supply.apply_revision(peer, group.clone(), supply_revision(1, true, false));
|
1294 + |
|
1295 + | let prepared = supply
|
1296 + | .prepare_match(&schedule(requesting_partition, group))
|
1297 + | .expect("peer connection cell was not selected");
|
1298 + | assert_eq!(peer, prepared.supplier);
|
1299 + | assert_ne!(requesting_partition, prepared.supplier);
|
1300 + | }
|
1301 + |
|
1302 + | #[test]
|
1303 + | fn resolving_match_discards_its_lazy_cancellation_entry() {
|
1304 + | let supplier = cell(1);
|
1305 + | let requester = cell(2);
|
1306 + | let group = EligibilityGroup::Pool;
|
1307 + | let mut schedule = schedule(requester, group.clone());
|
1308 + | let mut supply = H1Supply::default();
|
1309 + | supply.apply_revision(supplier, group.clone(), supply_revision(1, true, false));
|
1310 + | let prepared = supply
|
1311 + | .prepare_match(&schedule)
|
1312 + | .expect("supplier cell did not produce a retained match");
|
1313 + | supply.settle_reservation(prepared.match_id, false);
|
1314 + |
|
1315 + | schedule.apply_snapshot(
|
1316 + | requester,
|
1317 + | DemandSnapshot::inactive(DemandId::from_u64(1), SnapshotVersion::INITIAL.next()),
|
1318 + | );
|
1319 + | supply.reconcile_requester(&requester, &schedule);
|
1320 + | assert_eq!(
|
1321 + | H1MatchState::Cancelling,
|
1322 + | supply.matches[&prepared.match_id].state
|
1323 + | );
|
1324 + |
|
1325 + | supply.begin_resolution(prepared.match_id);
|
1326 + | assert_eq!(
|
1327 + | H1MatchState::Resolving,
|
1328 + | supply.matches[&prepared.match_id].state
|
1329 + | );
|
1330 + | assert!(
|
1331 + | supply.prepare_cancellation().is_none(),
|
1332 + | "resolving match produced stale cancellation work"
|
1333 + | );
|
1334 + | assert!(supply.cancellations.is_empty());
|
1335 + | }
|
1336 + |
|
1337 + | #[test]
|
1338 + | fn expired_supplier_cell_terminates_its_match_without_resubmission() {
|
1339 + | let connection_partition = cell(1);
|
1340 + | let requesting_partition = cell(2);
|
1341 + | let group = EligibilityGroup::Pool;
|
1342 + | let schedule = schedule(requesting_partition, group.clone());
|
1343 + | let mut supply = H1Supply::default();
|
1344 + | supply.apply_revision(connection_partition, group, supply_revision(1, true, false));
|
1345 + | let prepared = supply
|
1346 + | .prepare_match(&schedule)
|
1347 + | .expect("supplier cell did not produce a retained match");
|
1348 + |
|
1349 + | supply.settle_match(
|
1350 + | prepared.match_id,
|
1351 + | H1SupplyOutcome::supplier_expired(connection_partition),
|
1352 + | );
|
1353 + |
|
1354 + | assert!(!supply.index.records.contains_key(&connection_partition));
|
1355 + | assert!(supply.prepare_match(&schedule).is_none());
|
1356 + | assert!(supply.matches.is_empty());
|
1357 + | }
|
1358 + |
|
1359 + | #[test]
|
1360 + | fn stale_supply_outcome_cannot_hide_newer_status() {
|
1361 + | let connection_partition = cell(1);
|
1362 + | let requesting_partition = cell(2);
|
1363 + | let group = EligibilityGroup::Pool;
|
1364 + | let schedule = schedule(requesting_partition, group.clone());
|
1365 + | let mut supply = H1Supply::default();
|
1366 + | supply.apply_revision(connection_partition, group, supply_revision(1, true, false));
|
1367 + | let prepared = supply
|
1368 + | .prepare_match(&schedule)
|
1369 + | .expect("supplier cell did not produce a retained match");
|
1370 + |
|
1371 + | supply.settle_match(
|
1372 + | prepared.match_id,
|
1373 + | H1SupplyOutcome::supplier_live(connection_partition, supply_revision(3, true, false)),
|
1374 + | );
|
1375 + | supply.settle_match(
|
1376 + | prepared.match_id,
|
1377 + | H1SupplyOutcome::supplier_live(connection_partition, supply_revision(2, false, false)),
|
1378 + | );
|
1379 + |
|
1380 + | let supplier = supply
|
1381 + | .index
|
1382 + | .records
|
1383 + | .get(&connection_partition)
|
1384 + | .expect("stale revision removed the supplier cell");
|
1385 + | assert!(supplier.status.has_returnable_connection);
|
1386 + | assert!(supplier.index_state.is_linked());
|
1387 + | assert_eq!(3, supplier.revision);
|
1388 + | }
|
1389 + |
|
1390 + | #[test]
|
1391 + | fn duplicate_supply_outcome_still_refreshes_status() {
|
1392 + | let connection_partition = cell(1);
|
1393 + | let requesting_partition = cell(2);
|
1394 + | let group = EligibilityGroup::Pool;
|
1395 + | let schedule = schedule(requesting_partition, group.clone());
|
1396 + | let mut supply = H1Supply::default();
|
1397 + | supply.apply_revision(connection_partition, group, supply_revision(1, true, false));
|
1398 + | let prepared = supply
|
1399 + | .prepare_match(&schedule)
|
1400 + | .expect("supplier cell did not produce a retained match");
|
1401 + |
|
1402 + | supply.settle_match(
|
1403 + | prepared.match_id,
|
1404 + | H1SupplyOutcome::supplier_live(connection_partition, supply_revision(2, true, false)),
|
1405 + | );
|
1406 + | supply.settle_match(
|
1407 + | prepared.match_id,
|
1408 + | H1SupplyOutcome::supplier_live(connection_partition, supply_revision(3, false, false)),
|
1409 + | );
|
1410 + |
|
1411 + | let supplier = supply
|
1412 + | .index
|
1413 + | .records
|
1414 + | .get(&connection_partition)
|
1415 + | .expect("live supply revision removed the supplier cell");
|
1416 + | assert!(!supplier.status.has_returnable_connection);
|
1417 + | assert!(!supplier.index_state.is_linked());
|
1418 + | }
|
1419 + | }
|