1 1 | /*
|
2 2 | * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
3 3 | * SPDX-License-Identifier: Apache-2.0
|
4 4 | */
|
5 5 |
|
6 - | //! Capacity and demand shared by the cells of one bounded origin.
|
6 + | //! Demand, connection supply, and capacity shared by one bounded origin.
|
7 7 | //!
|
8 8 | //! [`OriginAdmission`] stores the origin once and keys its partition cells,
|
9 - | //! demand records, and availability indexes by [`PartitionId`]. Its lock is the
|
10 - | //! sole authority for available capacity, origin-wide demand order, and reuse
|
11 - | //! operations. Cell locks are never held with the admission lock.
|
9 + | //! demand records, and supply indexes by [`PartitionId`]. Its lock is the sole
|
10 + | //! authority for admission-resident capacity, origin-wide demand order, and
|
11 + | //! retained HTTP/1 matching. Cell locks are never held with the admission lock.
|
12 12 | //!
|
13 13 | //! A bounded acquisition moves through the layer in this order:
|
14 14 | //!
|
15 - | //! 1. A cell publishes a complete [`DemandSnapshot`] after releasing its lock.
|
15 + | //! 1. A cell submits a complete [`DemandSnapshot`] after releasing its lock.
|
16 16 | //! 2. [`DemandSchedule`] replaces that partition snapshot and links active
|
17 17 | //! demand in the origin order and its eligibility-group order.
|
18 - | //! 3. Capacity delivery and HTTP/1 reuse select from the origin head. HTTP/2
|
19 - | //! publication pairs a group head with an advertised peer generation.
|
20 - | //! 4. A [`DeliveryGuard`] materializes its one capacity or HTTP/1 payload before
|
21 - | //! reserving the requesting waiter. Payload materialization failure
|
22 - | //! therefore leaves the waiter in its existing residence.
|
23 - | //! 5. An [`H2PublicationGuard`] carries identities through the same unlocked
|
18 + | //! 3. Capacity delivery and HTTP/1 matching select from the origin head.
|
19 + | //! HTTP/2 route matching pairs a group head with peer supply.
|
20 + | //! 4. The selected demand becomes a [`DemandAssignment`] while a resource
|
21 + | //! crosses lock domains. Both scheduling positions remain attached, so
|
22 + | //! refusal can restore the demand without changing its order.
|
23 + | //! 5. A [`DeliveryGuard`] resolves its capacity or HTTP/1 payload before
|
24 + | //! committing the exact assigned waiter.
|
25 + | //! 6. An [`H2RouteGuard`] carries identities through the same unlocked
|
24 26 | //! handoff. It revalidates the connection-owning generation, then installs
|
25 27 | //! a route and activation opportunity in the requesting cell.
|
26 - | //! 6. The requesting cell becomes authoritative before either guard submits its
|
27 - | //! admission acknowledgement. Rejection and drop execute the same terminal
|
28 - | //! acknowledgement paths.
|
28 + | //! 7. The requesting cell becomes authoritative before either guard settles the
|
29 + | //! assignment. Refusal and drop execute the same fallback paths.
|
29 30 | //!
|
30 31 | //! Capacity and HTTP/1 handoffs own the payload they must return on failure.
|
31 - | //! HTTP/2 publication moves no payload: the connection-owning cell retains its
|
32 - | //! request handle, driver, socket, and capacity. No cell lock is held with
|
33 - | //! the admission lock, and connection-owning and requesting cell locks are
|
34 - | //! never held together.
|
32 + | //! HTTP/2 route installation moves no payload: the connection-owning cell
|
33 + | //! retains its request handle, driver, socket, and capacity. No cell lock is
|
34 + | //! held with the admission lock, and connection-owning and requesting cell
|
35 + | //! locks are never held together.
|
35 36 |
|
36 - | use super::cell::OriginCell;
|
37 + | use super::cell::{H1ReservationDecision, OriginCell};
|
37 38 | use super::origin::OriginKey;
|
38 39 | use super::partition::{EligibilityGroup, PartitionId};
|
40 + | use super::registry::AdmissionPolicy;
|
39 41 | use crate::sync::{Arc, Mutex, Weak};
|
40 42 | use std::collections::HashMap;
|
41 43 | use std::fmt;
|
42 44 | use std::num::NonZeroUsize;
|
43 45 |
|
44 46 | mod delivery;
|
45 47 | mod demand;
|
48 + | mod h1;
|
49 + | mod h2;
|
46 50 | mod order;
|
47 - | mod publication;
|
48 - | pub(in crate::client::pool) mod reuse;
|
49 51 |
|
50 - | use self::demand::{DemandSchedule, PreparedCapacityDelivery};
|
51 - | use self::order::{IntrusiveLinks, IntrusiveOrder};
|
52 - | use self::publication::{
|
53 - | H2Publication, H2PublicationGuard, H2ReclaimAction, PreparedH2Publication, PreparedH2Reclaim,
|
52 + | use self::demand::{
|
53 + | DemandAssignment, DemandAssignmentId, DemandAssignmentOutcome, DemandSchedule,
|
54 + | PreparedCapacityDelivery,
|
55 + | };
|
56 + | pub(in crate::client::pool) use self::demand::{
|
57 + | DemandId, DemandSnapshot, ProtocolRequirement, SnapshotVersion,
|
54 58 | };
|
59 + | use self::h1::{
|
60 + | H1CancellationAction, H1CapacityReclaim, H1ReservationAction, H1SupplierSettlement, H1Supply,
|
61 + | H1SupplyOutcome,
|
62 + | };
|
63 + | use self::h2::{H2CapacityReclaim, H2RouteGuard, H2Supply, PreparedH2Reclaim, PreparedH2Route};
|
64 + | use self::order::{IntrusiveLinks, IntrusiveOrder};
|
55 65 | pub(in crate::client::pool) use delivery::DeliveryGuard;
|
56 - | pub(in crate::client::pool) use publication::H2AdvertisementSnapshot;
|
66 + | pub(in crate::client::pool) use h1::{
|
67 + | H1Candidate, H1MatchId, H1SupplyStatus, PreparedH1Reservation,
|
68 + | };
|
69 + | pub(in crate::client::pool) use h2::H2SupplyStatus;
|
57 70 |
|
58 - | /// Protocol capability required by the head waiter in a cell.
|
71 + | /// Complete admission-facing protocol status at one cell-owned revision.
|
59 72 | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
60 - | pub(crate) enum ProtocolRequirement {
|
61 - | /// The waiter requires HTTP/1 wire semantics.
|
62 - | H1Required,
|
63 - | /// The waiter may dispatch over HTTP/1 or HTTP/2.
|
64 - | H1Compatible,
|
65 - | /// The waiter requires HTTP/2.
|
66 - | H2Required,
|
73 + | pub(in crate::client::pool) struct SupplyRevision<T> {
|
74 + | /// Monotonic sequence number allocated under the cell lock.
|
75 + | pub(in crate::client::pool) revision: u64,
|
76 + | /// Complete status represented by this revision.
|
77 + | pub(in crate::client::pool) status: T,
|
67 78 | }
|
68 79 |
|
69 - | impl ProtocolRequirement {
|
70 - | /// Returns whether an HTTP/1 sender can satisfy this requirement.
|
71 - | pub(crate) fn accepts_h1(self) -> bool {
|
72 - | self != Self::H2Required
|
73 - | }
|
74 - |
|
75 - | /// Returns whether an HTTP/2 activation can satisfy this requirement.
|
76 - | pub(crate) fn accepts_h2(self) -> bool {
|
77 - | self != Self::H1Required
|
78 - | }
|
79 - | }
|
80 - |
|
81 - | /// Identity of one cell-local demand generation.
|
82 - | ///
|
83 - | /// A generation begins when a waiter becomes the cell's bounded-demand head.
|
84 - | /// The cell identity and this value together identify its publications.
|
85 - | #[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
|
86 - | pub(crate) struct DemandId(u64);
|
87 - |
|
88 - | impl DemandId {
|
89 - | /// Constructs an identity allocated by one cell's monotonic demand counter.
|
90 - | pub(crate) const fn from_u64(value: u64) -> Self {
|
91 - | Self(value)
|
92 - | }
|
93 - | }
|
94 - |
|
95 - | /// Strict ordering of complete publications within one [`DemandId`].
|
96 - | ///
|
97 - | /// Versions never wrap. Preserving strict order prevents a delayed publication
|
98 - | /// from becoming current again after counter reuse (an ABA). A new FIFO head
|
99 - | /// receives a new [`DemandId`] and starts again at [`SnapshotVersion::INITIAL`].
|
100 - | #[derive(Clone, Copy, Debug, Eq, Ord, PartialEq, PartialOrd)]
|
101 - | pub(crate) struct SnapshotVersion(u64);
|
102 - |
|
103 - | impl SnapshotVersion {
|
104 - | /// Version assigned to the first snapshot for a demand generation.
|
105 - | pub(crate) const INITIAL: Self = Self(0);
|
106 - |
|
107 - | /// Advances the publication version for the same demand generation.
|
108 - | ///
|
109 - | /// # Panics
|
110 - | ///
|
111 - | /// Panics after `u64::MAX` replacements of one generation. Wrapping would
|
112 - | /// break stale-publication rejection; a new head waiter starts a
|
113 - | /// new generation and resets this counter.
|
114 - | pub(crate) fn next(self) -> Self {
|
115 - | Self(
|
116 - | self.0
|
117 - | .checked_add(1)
|
118 - | .expect("demand snapshot version exhausted"),
|
119 - | )
|
120 - | }
|
121 - | }
|
122 - |
|
123 - | /// Complete state published for one demand identity.
|
124 - | #[derive(Clone, Debug, Eq, PartialEq)]
|
125 - | pub(crate) enum DemandState {
|
126 - | /// The cell has a waiter that still needs a connection.
|
127 - | Active {
|
128 - | /// Protocol capability required by the cell's oldest waiter.
|
129 - | head: ProtocolRequirement,
|
130 - | /// Partitions whose connections may satisfy this demand.
|
131 - | eligibility_group: EligibilityGroup,
|
132 - | },
|
133 - | /// The demand has ended without a successor in this snapshot.
|
134 - | Inactive,
|
135 - | }
|
136 - |
|
137 - | /// Versioned replacement state for one cell's current demand generation.
|
138 - | ///
|
139 - | /// A publication replaces the complete previous snapshot. This prevents a
|
140 - | /// delayed active publication from reviving demand retired by a newer version.
|
141 - | #[derive(Clone, Debug, Eq, PartialEq)]
|
142 - | pub(crate) struct DemandSnapshot {
|
143 - | /// Cell-local identity of the head-waiter generation.
|
144 - | id: DemandId,
|
145 - | /// Ordering among publications for the same demand identity.
|
146 - | version: SnapshotVersion,
|
147 - | /// Complete current state of the demand generation.
|
148 - | state: DemandState,
|
149 - | }
|
150 - |
|
151 - | impl DemandSnapshot {
|
152 - | /// Describes the current head waiter for an active demand generation.
|
153 - | pub(crate) fn active(
|
154 - | id: DemandId,
|
155 - | version: SnapshotVersion,
|
156 - | head: ProtocolRequirement,
|
157 - | eligibility_group: EligibilityGroup,
|
158 - | ) -> Self {
|
159 - | Self {
|
160 - | id,
|
161 - | version,
|
162 - | state: DemandState::Active {
|
163 - | head,
|
164 - | eligibility_group,
|
165 - | },
|
166 - | }
|
167 - | }
|
168 - |
|
169 - | /// Returns whether this active demand can use an HTTP/2 generation.
|
170 - | pub(crate) fn accepts_h2(&self) -> bool {
|
171 - | matches!(
|
172 - | self.state,
|
173 - | DemandState::Active { head, .. } if head.accepts_h2()
|
174 - | )
|
175 - | }
|
176 - |
|
177 - | /// Retires a demand generation at the supplied publication version.
|
178 - | pub(crate) fn inactive(id: DemandId, version: SnapshotVersion) -> Self {
|
179 - | Self {
|
180 - | id,
|
181 - | version,
|
182 - | state: DemandState::Inactive,
|
183 - | }
|
184 - | }
|
185 - |
|
186 - | /// Returns the demand identity for cross-module transition tests.
|
187 - | #[cfg(test)]
|
188 - | pub(in crate::client::pool) fn id_for_test(&self) -> DemandId {
|
189 - | self.id
|
190 - | }
|
191 - |
|
192 - | /// Returns whether this snapshot still requests capacity.
|
193 - | pub(in crate::client::pool) fn is_active(&self) -> bool {
|
194 - | matches!(self.state, DemandState::Active { .. })
|
195 - | }
|
196 - |
|
197 - | /// Returns whether this snapshot may replace `current`.
|
198 - | ///
|
199 - | /// A newer generation supersedes every snapshot of an older generation;
|
200 - | /// publications within one generation are ordered by snapshot version.
|
201 - | fn is_newer_than(&self, current: &Self) -> bool {
|
202 - | self.id > current.id || (self.id == current.id && self.version > current.version)
|
80 + | impl<T> SupplyRevision<T> {
|
81 + | pub(in crate::client::pool) fn new(revision: u64, status: T) -> Self {
|
82 + | Self { revision, status }
|
203 83 | }
|
204 84 | }
|
205 85 |
|
206 86 | /// One conserved unit of bounded-origin connection capacity.
|
207 87 | ///
|
208 88 | /// This non-`Copy` value proves that admission removed one unit from its
|
209 89 | /// available count. Identities are never reused so diagnostics remain
|
210 90 | /// unambiguous.
|
211 - | pub(super) struct Permit(u64);
|
91 + | pub(super) struct CapacityPermit(u64);
|
212 92 |
|
213 - | impl fmt::Debug for Permit {
|
93 + | impl fmt::Debug for CapacityPermit {
|
214 94 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
215 - | f.debug_tuple("Permit").field(&self.0).finish()
|
95 + | f.debug_tuple("CapacityPermit").field(&self.0).finish()
|
216 96 | }
|
217 97 | }
|
218 98 |
|
219 - | /// Stable identity of one admission-to-cell delivery.
|
220 - | #[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
|
221 - | pub(super) struct DeliveryId(u64);
|
222 - |
|
223 99 | /// Shared admission authority for one bounded origin.
|
224 100 | ///
|
225 101 | /// The state lock is the only authority for available capacity while it
|
226 102 | /// resides in admission, and for demand order while no delivery is crossing
|
227 103 | /// to a cell. Delivery work is detached before this lock is released.
|
228 - | pub(crate) struct OriginAdmission {
|
104 + | pub(in crate::client::pool) struct OriginAdmission {
|
229 105 | /// Canonical origin shared by every partition represented in `state`.
|
230 106 | origin: OriginKey,
|
231 107 | /// Whether admission may close idle H2 capacity for H1-required demand.
|
232 - | allow_h2_reclaim_for_h1: bool,
|
233 - | /// Capacity, demand, and delivery-fence state for this origin.
|
108 + | can_reclaim_h2_for_h1: bool,
|
109 + | /// Capacity, demand, and supply state for this origin.
|
234 110 | state: Mutex<AdmissionState>,
|
235 111 | }
|
236 112 |
|
237 113 | impl OriginAdmission {
|
238 - | /// Creates admission for at most `limit` logically open connections.
|
239 - | pub(crate) fn new(
|
240 - | origin: OriginKey,
|
241 - | limit: NonZeroUsize,
|
242 - | allow_h2_reclaim_for_h1: bool,
|
243 - | ) -> Arc<Self> {
|
114 + | /// Creates admission for one bounded origin policy.
|
115 + | pub(in crate::client::pool) fn new(origin: OriginKey, policy: AdmissionPolicy) -> Arc<Self> {
|
244 116 | Arc::new(Self {
|
245 117 | origin,
|
246 - | allow_h2_reclaim_for_h1,
|
247 - | state: Mutex::new(AdmissionState::new(limit)),
|
118 + | can_reclaim_h2_for_h1: policy.can_reclaim_h2_for_h1(),
|
119 + | state: Mutex::new(AdmissionState::new(policy.connection_limit())),
|
248 120 | })
|
249 121 | }
|
250 122 |
|
251 123 | #[cfg(test)]
|
252 124 | pub(super) fn for_test(limit: NonZeroUsize) -> Arc<Self> {
|
253 125 | Self::for_test_with_h2_reclaim(limit, true)
|
254 126 | }
|
255 127 |
|
256 128 | #[cfg(test)]
|
257 129 | pub(super) fn for_test_with_h2_reclaim(
|
258 130 | limit: NonZeroUsize,
|
259 - | allow_h2_reclaim_for_h1: bool,
|
131 + | can_reclaim_h2_for_h1: bool,
|
260 132 | ) -> Arc<Self> {
|
261 133 | Self::new(
|
262 134 | OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None)
|
263 135 | .expect("test origin is valid"),
|
264 - | limit,
|
265 - | allow_h2_reclaim_for_h1,
|
136 + | AdmissionPolicy::new(limit, can_reclaim_h2_for_h1),
|
266 137 | )
|
267 138 | }
|
268 139 |
|
269 140 | /// Returns the canonical origin coordinated by this admission authority.
|
270 141 | fn origin(&self) -> &OriginKey {
|
271 142 | &self.origin
|
272 143 | }
|
273 144 |
|
274 145 | /// Registers or returns the unique retained cell for an identity.
|
275 - | pub(crate) fn register_cell(origin: &Arc<Self>, candidate: Arc<OriginCell>) -> Arc<OriginCell> {
|
146 + | pub(in crate::client::pool) fn register_cell(
|
147 + | origin: &Arc<Self>,
|
148 + | candidate: Arc<OriginCell>,
|
149 + | ) -> Arc<OriginCell> {
|
276 150 | let partition = candidate.id().partition();
|
277 151 | assert_eq!(
|
278 152 | origin.origin(),
|
279 153 | candidate.id().origin(),
|
280 154 | "cell origin did not match its admission authority"
|
281 155 | );
|
282 156 | let existing = {
|
283 157 | let mut state = origin.state.lock();
|
284 158 | match state.cells.get(&partition).and_then(Weak::upgrade) {
|
285 159 | Some(existing) => Some(existing),
|
286 160 | None => {
|
287 161 | state.cells.insert(partition, Weak::from_arc(&candidate));
|
288 162 | None
|
289 163 | }
|
290 164 | }
|
291 165 | };
|
292 166 | if let Some(existing) = existing {
|
293 167 | // Dropping the losing cell may return capacity to this admission
|
294 168 | // authority, so release the state lock first.
|
295 169 | drop(candidate);
|
296 170 | existing
|
297 171 | } else {
|
298 172 | candidate
|
299 173 | }
|
300 174 | }
|
301 175 |
|
302 - | /// Publishes a complete demand snapshot and drives any resulting delivery.
|
303 - | pub(crate) fn publish_demand(
|
304 - | origin: &Arc<Self>,
|
305 - | requesting_partition: PartitionId,
|
176 + | /// Submits a complete demand snapshot and runs resulting detached actions.
|
177 + | pub(in crate::client::pool) fn submit_demand_snapshot(
|
178 + | admission: &Arc<Self>,
|
179 + | requester: PartitionId,
|
306 180 | snapshot: DemandSnapshot,
|
307 181 | ) {
|
308 182 | let action = {
|
309 - | let mut state = origin.state.lock();
|
310 - | state.publish_demand(requesting_partition, snapshot);
|
311 - | Self::prepare_action(origin, &mut state)
|
183 + | let mut state = admission.state.lock();
|
184 + | state.apply_demand_snapshot(requester, snapshot);
|
185 + | Self::prepare_action(admission, &mut state)
|
312 186 | };
|
313 - | Self::drive(action);
|
187 + | Self::run_action_chain(action);
|
314 188 | }
|
315 189 |
|
316 190 | /// Selects at most one action while admission is locked.
|
317 191 | ///
|
318 192 | /// Cancellation cleanup runs first, followed by available capacity,
|
319 - | /// compatible HTTP/2 publication, HTTP/1 reuse, and idle-H2 reclaim. The
|
193 + | /// a compatible peer HTTP/2 route, HTTP/1 reuse, and idle-H2 reclaim. The
|
320 194 | /// returned action owns everything needed to run after releasing the
|
321 - | /// admission lock. Its completion prepares the next action, forming an
|
322 - | /// iterative pump without nesting admission and cell locks.
|
195 + | /// admission lock. Its completion may prepare a successor for the caller's
|
196 + | /// action chain. A guard dropped outside that chain may start another
|
197 + | /// chain, but admission and cell locks are never nested.
|
323 198 | fn prepare_action(origin: &Arc<Self>, state: &mut AdmissionState) -> Option<AdmissionAction> {
|
324 - | if let Some(cancellation) = state.h1.prepare_cancellation() {
|
325 - | return Some(AdmissionAction::H1(reuse::H1ReuseAction::cancel(
|
326 - | origin.clone(),
|
327 - | cancellation,
|
328 - | )));
|
199 + | if let Some(cancellation) = state.h1_supply.prepare_cancellation() {
|
200 + | return Some(AdmissionAction::CancelH1Reservation(
|
201 + | H1CancellationAction::new(origin.clone(), cancellation),
|
202 + | ));
|
329 203 | }
|
330 - | if let Some(pending) = state.schedule_one() {
|
331 - | return Some(AdmissionAction::Delivery(DeliveryGuard::capacity(
|
204 + | if let Some(prepared) = state.prepare_capacity_delivery() {
|
205 + | return Some(AdmissionAction::Deliver(DeliveryGuard::capacity(
|
332 206 | origin.clone(),
|
333 - | pending.delivery,
|
334 - | pending.requesting_partition,
|
335 - | pending.demand,
|
336 - | pending.permit,
|
207 + | prepared.assignment,
|
208 + | prepared.permit,
|
337 209 | )));
|
338 210 | }
|
339 - | if let Some(publication) = state.prepare_h2_publication() {
|
340 - | return Some(AdmissionAction::H2(H2PublicationGuard::new(
|
211 + | if let Some(route) = state.prepare_h2_route() {
|
212 + | return Some(AdmissionAction::AttachH2Route(H2RouteGuard::new(
|
341 213 | origin.clone(),
|
342 - | publication,
|
214 + | route,
|
343 215 | )));
|
344 216 | }
|
345 - | if let Some(reuse) = state.h1.prepare_reuse(&state.demand_schedule) {
|
346 - | return Some(AdmissionAction::H1(reuse::H1ReuseAction::install(
|
347 - | origin.clone(),
|
348 - | reuse,
|
349 - | )));
|
217 + | if let Some(reservation) = state.h1_supply.prepare_match(&state.demand) {
|
218 + | return Some(AdmissionAction::ReserveH1Supplier(
|
219 + | H1ReservationAction::new(origin.clone(), reservation),
|
220 + | ));
|
350 221 | }
|
351 - | if !origin.allow_h2_reclaim_for_h1 {
|
222 + | if !origin.can_reclaim_h2_for_h1 {
|
352 223 | return None;
|
353 224 | }
|
354 225 | state
|
355 - | .h2
|
356 - | .prepare_reclaim(&state.demand_schedule)
|
226 + | .h2_supply
|
227 + | .prepare_reclaim(&state.demand)
|
357 228 | .map(|reclaim| {
|
358 - | AdmissionAction::H2Reclaim(H2ReclaimAction::new(origin.clone(), reclaim))
|
229 + | AdmissionAction::ReclaimCapacity(CapacityReclaim::FromH2(H2CapacityReclaim::new(
|
230 + | origin.clone(),
|
231 + | reclaim,
|
232 + | )))
|
359 233 | })
|
360 234 | }
|
361 235 |
|
362 236 | /// Upgrades a registered requesting cell without holding the admission lock.
|
363 237 | fn cell(&self, id: &PartitionId) -> Option<Arc<OriginCell>> {
|
364 238 | let cell = {
|
365 239 | let state = self.state.lock();
|
366 240 | state.cells.get(id).cloned()
|
367 241 | };
|
368 242 | cell.and_then(|cell| cell.upgrade())
|
369 243 | }
|
370 244 |
|
371 - | /// Drives sequential unlocked actions until no completion schedules another.
|
372 - | pub(in crate::client::pool) fn drive(mut action: Option<AdmissionAction>) {
|
245 + | /// Runs one detached action chain until no completion prepares a successor.
|
246 + | pub(in crate::client::pool) fn run_action_chain(mut action: Option<AdmissionAction>) {
|
373 247 | while let Some(current) = action {
|
374 - | action = current.drive_once();
|
248 + | action = match current {
|
249 + | AdmissionAction::Deliver(delivery) => delivery.deliver(),
|
250 + | AdmissionAction::ReserveH1Supplier(reservation) => reservation.reserve_supplier(),
|
251 + | AdmissionAction::CancelH1Reservation(cancellation) => {
|
252 + | cancellation.cancel_reservation()
|
253 + | }
|
254 + | AdmissionAction::SettleH1Supplier(settlement) => settlement.settle_supplier(),
|
255 + | AdmissionAction::AttachH2Route(route) => route.attach_route(),
|
256 + | AdmissionAction::ReclaimCapacity(reclaim) => reclaim.reclaim_capacity(),
|
257 + | };
|
375 258 | }
|
376 259 | }
|
377 260 |
|
378 - | /// Returns whether the named delivery still owns the requesting cell's demand fence.
|
261 + | /// Returns whether admission still recognizes this exact assignment.
|
379 262 | #[cfg(test)]
|
380 - | fn delivery_is_current(
|
381 - | &self,
|
382 - | delivery: DeliveryId,
|
383 - | requesting_partition: &PartitionId,
|
384 - | demand: DemandId,
|
385 - | ) -> bool {
|
386 - | self.state
|
387 - | .lock()
|
388 - | .delivery_is_current(delivery, requesting_partition, demand)
|
263 + | fn assignment_is_current(&self, assignment: &DemandAssignment) -> bool {
|
264 + | self.state.lock().demand.assignment_is_current(assignment)
|
389 265 | }
|
390 266 |
|
391 267 | /// Returns `permit` to admission and serves ordered demand when possible.
|
392 - | fn return_permit(origin: &Arc<Self>, permit: Permit) {
|
268 + | fn return_permit(origin: &Arc<Self>, permit: CapacityPermit) {
|
393 269 | let action = {
|
394 270 | let mut state = origin.state.lock();
|
395 271 | state.return_permit(permit);
|
396 272 | Self::prepare_action(origin, &mut state)
|
397 273 | };
|
398 - | Self::drive(action);
|
274 + | Self::run_action_chain(action);
|
399 275 | }
|
400 276 |
|
401 - | /// Closes a delivery fence and prepares at most one successor.
|
402 - | ///
|
403 - | /// `permit` is present when an undelivered payload is refunnelled. A
|
404 - | /// committed delivery leaves the permit in its installed
|
405 - | /// [`CapacityLease`] and acknowledges with `None`.
|
406 - | fn finish_delivery(
|
407 - | origin: &Arc<Self>,
|
408 - | delivery: DeliveryId,
|
409 - | requesting_partition: &PartitionId,
|
410 - | permit: Option<Permit>,
|
411 - | result: DeliveryAckResult,
|
277 + | /// Settles one demand assignment and returns an unused capacity permit.
|
278 + | fn settle_delivery(
|
279 + | admission: &Arc<Self>,
|
280 + | assignment: &DemandAssignment,
|
281 + | permit: Option<CapacityPermit>,
|
282 + | outcome: DemandAssignmentOutcome,
|
412 283 | ) -> Option<AdmissionAction> {
|
413 - | let mut state = origin.state.lock();
|
284 + | let mut state = admission.state.lock();
|
414 285 | if let Some(permit) = permit {
|
415 286 | state.return_permit(permit);
|
416 287 | }
|
417 - | state.finish_delivery(delivery, requesting_partition, result);
|
418 - | Self::prepare_action(origin, &mut state)
|
288 + | state.settle_assignment(assignment, outcome);
|
289 + | Self::prepare_action(admission, &mut state)
|
419 290 | }
|
420 291 |
|
421 - | /// Replaces one connection cell's complete HTTP/2 advertisement.
|
422 - | pub(in crate::client::pool) fn update_h2_advertisement(
|
423 - | origin: &Arc<Self>,
|
424 - | connection_partition: PartitionId,
|
425 - | group: EligibilityGroup,
|
426 - | snapshot: H2AdvertisementSnapshot,
|
292 + | /// Applies one cell's complete HTTP/1 supply revision.
|
293 + | pub(in crate::client::pool) fn apply_h1_supply_revision(
|
294 + | admission: &Arc<Self>,
|
295 + | supplier: PartitionId,
|
296 + | eligibility_group: EligibilityGroup,
|
297 + | revision: SupplyRevision<H1SupplyStatus>,
|
427 298 | ) {
|
428 - | let action = {
|
429 - | let mut state = origin.state.lock();
|
430 - | let AdmissionState {
|
431 - | h2,
|
432 - | demand_schedule,
|
433 - | ..
|
434 - | } = &mut *state;
|
435 - | h2.update(connection_partition, group, snapshot, demand_schedule);
|
436 - | Self::prepare_action(origin, &mut state)
|
437 - | };
|
438 - | Self::drive(action);
|
299 + | h1::apply_supply_revision(admission, supplier, eligibility_group, revision);
|
439 300 | }
|
440 301 |
|
441 - | /// Applies one H2 publication acknowledgement and prepares its successor.
|
442 - | fn finish_h2_publication(
|
443 - | origin: &Arc<Self>,
|
444 - | prepared: &PreparedH2Publication,
|
302 + | /// Rejects a returned sender whose retained match is no longer current.
|
303 + | pub(in crate::client::pool) fn reject_returned_h1_match(
|
304 + | admission: &Arc<Self>,
|
305 + | match_id: H1MatchId,
|
306 + | supplier: PartitionId,
|
307 + | revision: SupplyRevision<H1SupplyStatus>,
|
308 + | ) {
|
309 + | h1::reject_returned_match(admission, match_id, supplier, revision);
|
310 + | }
|
311 + |
|
312 + | /// Settles installation of one supplier-cell reservation.
|
313 + | pub(in crate::client::pool) fn settle_h1_reservation(
|
314 + | admission: &Arc<Self>,
|
315 + | match_id: H1MatchId,
|
316 + | supplier: PartitionId,
|
317 + | decision: H1ReservationDecision<H1Candidate>,
|
318 + | ) -> Option<AdmissionAction> {
|
319 + | h1::settle_reservation(admission, match_id, supplier, decision)
|
320 + | }
|
321 + |
|
322 + | /// Resolves a provisional sender against its retained match and demand.
|
323 + | pub(in crate::client::pool) fn resolve_h1_match(
|
324 + | admission: &Arc<Self>,
|
325 + | match_id: H1MatchId,
|
326 + | candidate: H1Candidate,
|
327 + | ) -> Option<AdmissionAction> {
|
328 + | h1::resolve_match(admission, match_id, candidate)
|
329 + | }
|
330 + |
|
331 + | /// Removes one retained H1 match and applies its supplier outcome.
|
332 + | fn settle_h1_match(
|
333 + | admission: &Arc<Self>,
|
334 + | match_id: H1MatchId,
|
335 + | outcome: H1SupplyOutcome,
|
336 + | ) -> Option<AdmissionAction> {
|
337 + | h1::settle_match(admission, match_id, outcome)
|
338 + | }
|
339 + |
|
340 + | /// Settles a borrowed-sender assignment and its retained H1 match.
|
341 + | fn settle_borrow_delivery(
|
342 + | admission: &Arc<Self>,
|
343 + | match_id: H1MatchId,
|
344 + | assignment: &DemandAssignment,
|
345 + | outcome: DemandAssignmentOutcome,
|
346 + | transferred_supplier: Option<PartitionId>,
|
347 + | refused_outcome: Option<H1SupplyOutcome>,
|
348 + | ) -> Option<AdmissionAction> {
|
349 + | h1::settle_borrow_delivery(
|
350 + | admission,
|
351 + | match_id,
|
352 + | assignment,
|
353 + | outcome,
|
354 + | transferred_supplier,
|
355 + | refused_outcome,
|
356 + | )
|
357 + | }
|
358 + |
|
359 + | /// Applies one cell's complete HTTP/2 supply revision.
|
360 + | pub(in crate::client::pool) fn apply_h2_supply_revision(
|
361 + | admission: &Arc<Self>,
|
362 + | supplier: PartitionId,
|
363 + | eligibility_group: EligibilityGroup,
|
364 + | revision: SupplyRevision<H2SupplyStatus>,
|
365 + | ) {
|
366 + | h2::apply_supply_revision(admission, supplier, eligibility_group, revision);
|
367 + | }
|
368 + |
|
369 + | /// Settles one H2 route assignment and repairs stale supply if needed.
|
370 + | fn settle_h2_route(
|
371 + | admission: &Arc<Self>,
|
372 + | prepared: &PreparedH2Route,
|
445 373 | stale_generation: Option<super::cell::h2::H2GenerationId>,
|
446 - | result: DeliveryAckResult,
|
374 + | outcome: DemandAssignmentOutcome,
|
447 375 | ) -> Option<AdmissionAction> {
|
448 - | let mut state = origin.state.lock();
|
449 - | if let Some(generation) = stale_generation {
|
450 - | let AdmissionState {
|
451 - | h2,
|
452 - | demand_schedule,
|
453 - | ..
|
454 - | } = &mut *state;
|
455 - | h2.remove_if_exact(&prepared.connection_partition, generation, demand_schedule);
|
456 - | }
|
457 - | state.finish_delivery(prepared.delivery, &prepared.requesting_partition, result);
|
458 - | Self::prepare_action(origin, &mut state)
|
376 + | h2::settle_route(admission, prepared, stale_generation, outcome)
|
459 377 | }
|
460 378 |
|
461 379 | /// Completes one exact idle-H2 reclaim crossing.
|
462 - | fn finish_h2_reclaim(
|
463 - | origin: &Arc<Self>,
|
380 + | fn settle_h2_reclaim(
|
381 + | admission: &Arc<Self>,
|
464 382 | prepared: &PreparedH2Reclaim,
|
465 - | snapshot: Option<H2AdvertisementSnapshot>,
|
383 + | revision: Option<SupplyRevision<H2SupplyStatus>>,
|
466 384 | ) -> Option<AdmissionAction> {
|
467 - | let mut state = origin.state.lock();
|
468 - | let AdmissionState {
|
469 - | h2,
|
470 - | demand_schedule,
|
471 - | ..
|
472 - | } = &mut *state;
|
473 - | h2.finish_reclaim(prepared, snapshot, demand_schedule);
|
474 - | Self::prepare_action(origin, &mut state)
|
385 + | h2::settle_reclaim(admission, prepared, revision)
|
475 386 | }
|
387 + |
|
476 388 | #[cfg(test)]
|
477 - | pub(super) fn publish_action_without_driving(
|
478 - | origin: &Arc<Self>,
|
479 - | requesting_partition: PartitionId,
|
389 + | pub(super) fn submit_action_without_running(
|
390 + | admission: &Arc<Self>,
|
391 + | requester: PartitionId,
|
480 392 | snapshot: DemandSnapshot,
|
481 393 | ) -> Option<AdmissionAction> {
|
482 - | let mut state = origin.state.lock();
|
483 - | state.publish_demand(requesting_partition, snapshot);
|
484 - | Self::prepare_action(origin, &mut state)
|
394 + | let mut state = admission.state.lock();
|
395 + | state.apply_demand_snapshot(requester, snapshot);
|
396 + | Self::prepare_action(admission, &mut state)
|
485 397 | }
|
486 398 |
|
487 399 | #[cfg(test)]
|
488 - | pub(super) fn publish_without_driving(
|
489 - | origin: &Arc<Self>,
|
490 - | requesting_partition: PartitionId,
|
400 + | pub(super) fn submit_without_running(
|
401 + | admission: &Arc<Self>,
|
402 + | requester: PartitionId,
|
491 403 | snapshot: DemandSnapshot,
|
492 404 | ) -> Option<DeliveryGuard> {
|
493 - | match Self::publish_action_without_driving(origin, requesting_partition, snapshot) {
|
494 - | Some(AdmissionAction::Delivery(delivery)) => Some(delivery),
|
495 - | Some(AdmissionAction::H1(_)) => {
|
496 - | panic!("capacity-only test unexpectedly prepared an HTTP/1 action")
|
497 - | }
|
498 - | Some(AdmissionAction::H2(_)) => {
|
499 - | panic!("capacity-only test unexpectedly prepared an HTTP/2 action")
|
500 - | }
|
501 - | Some(AdmissionAction::H2Reclaim(_)) => {
|
502 - | panic!("capacity-only test unexpectedly prepared an HTTP/2 reclaim action")
|
503 - | }
|
405 + | match Self::submit_action_without_running(admission, requester, snapshot) {
|
406 + | Some(AdmissionAction::Deliver(delivery)) => Some(delivery),
|
407 + | Some(_) => panic!("capacity-only test unexpectedly prepared another action"),
|
504 408 | None => None,
|
505 409 | }
|
506 410 | }
|
507 411 |
|
508 412 | #[cfg(test)]
|
509 - | fn counts(&self) -> AdmissionCounts {
|
413 + | fn probe(&self) -> AdmissionProbe {
|
510 414 | let state = self.state.lock();
|
511 - | AdmissionCounts {
|
512 - | limit: state.limit,
|
415 + | AdmissionProbe {
|
416 + | limit: state.capacity.limit,
|
513 417 | available: state.available_capacity(),
|
514 - | ordered: state.demand_schedule.len(),
|
515 - | queued: state.demand_schedule.queued_len(),
|
516 - | delivering: state.demand_schedule.delivering_len(),
|
418 + | ordered: state.demand.len(),
|
419 + | queued: state.demand.queued_len(),
|
420 + | assigned: state.demand.pending_assignment_count(),
|
517 421 | }
|
518 422 | }
|
519 423 |
|
520 424 | #[cfg(test)]
|
521 425 | pub(super) fn lease_for_test(origin: &Arc<Self>) -> CapacityLease {
|
522 426 | let permit = origin
|
523 427 | .state
|
524 428 | .lock()
|
525 429 | .take_permit()
|
526 430 | .expect("test origin had no available capacity");
|
527 431 | CapacityLease::new(origin.clone(), permit)
|
528 432 | }
|
529 433 |
|
530 434 | #[cfg(test)]
|
531 435 | pub(super) fn available_capacity_for_test(&self) -> usize {
|
532 436 | self.state.lock().available_capacity()
|
533 437 | }
|
534 438 |
|
535 439 | #[cfg(test)]
|
536 440 | pub(super) fn ordered_demand_count_for_test(&self) -> usize {
|
537 - | self.state.lock().demand_schedule.len()
|
441 + | self.state.lock().demand.len()
|
538 442 | }
|
539 443 |
|
540 444 | #[cfg(all(test, smithy_http_client_loom))]
|
541 445 | pub(super) fn clear_modeled_cells_for_test(&self) {
|
542 446 | // Loom has no modeled Weak, so its synchronization facade retains
|
543 447 | // cells strongly. Explicit teardown prevents that model-only
|
544 448 | // substitution from appearing as an Arc leak.
|
545 449 | self.state.lock().cells.clear();
|
546 450 | }
|
547 451 | }
|
548 452 |
|
549 453 | /// One detached step prepared while holding the bounded-origin lock.
|
550 454 | pub(super) enum AdmissionAction {
|
551 455 | /// One capacity or borrowed-H1 payload handed to a requesting cell.
|
552 - | Delivery(DeliveryGuard),
|
553 - | /// HTTP/1 availability, reservation, or borrowed-sender work.
|
554 - | H1(reuse::H1ReuseAction),
|
555 - | /// HTTP/2 generation visibility handed to one requesting cell.
|
556 - | H2(H2PublicationGuard),
|
557 - | /// Exact idle HTTP/2 generation reserved at its connection cell.
|
558 - | H2Reclaim(H2ReclaimAction),
|
456 + | Deliver(DeliveryGuard),
|
457 + | /// Reserve the selected supplier at its owning HTTP/1 cell.
|
458 + | ReserveH1Supplier(H1ReservationAction),
|
459 + | /// Cancel a retained HTTP/1 supplier reservation.
|
460 + | CancelH1Reservation(H1CancellationAction),
|
461 + | /// Settle sender ownership at the original HTTP/1 supplier cell.
|
462 + | SettleH1Supplier(H1SupplierSettlement),
|
463 + | /// HTTP/2 generation route handed to one requesting cell.
|
464 + | AttachH2Route(H2RouteGuard),
|
465 + | /// Close selected connection supply outside admission to recover capacity.
|
466 + | ReclaimCapacity(CapacityReclaim),
|
559 467 | }
|
560 468 |
|
561 469 | impl AdmissionAction {
|
562 - | /// Executes one detached action and returns the next prepared step.
|
563 - | fn drive_once(self) -> Option<Self> {
|
470 + | /// Advances one crossing without recursively driving its successor.
|
471 + | #[cfg(all(test, smithy_http_client_loom))]
|
472 + | pub(super) fn run_once_for_test(self) -> Option<Self> {
|
564 473 | match self {
|
565 - | Self::Delivery(delivery) => delivery.deliver_once(),
|
566 - | Self::H1(action) => action.drive_once(),
|
567 - | Self::H2(publication) => publication.publish_once(),
|
568 - | Self::H2Reclaim(reclaim) => reclaim.drive_once(),
|
474 + | Self::Deliver(delivery) => delivery.deliver(),
|
475 + | Self::ReserveH1Supplier(reservation) => reservation.reserve_supplier(),
|
476 + | Self::CancelH1Reservation(cancellation) => cancellation.cancel_reservation(),
|
477 + | Self::SettleH1Supplier(settlement) => settlement.settle_supplier(),
|
478 + | Self::AttachH2Route(route) => route.attach_route(),
|
479 + | Self::ReclaimCapacity(reclaim) => reclaim.reclaim_capacity(),
|
569 480 | }
|
570 481 | }
|
482 + | }
|
571 483 |
|
572 - | /// Advances one crossing without recursively driving its successor.
|
573 - | #[cfg(all(test, smithy_http_client_loom))]
|
574 - | pub(super) fn drive_once_for_test(self) -> Option<Self> {
|
575 - | self.drive_once()
|
484 + | /// Protocol-specific detached reclaim with one shared admission action.
|
485 + | pub(super) enum CapacityReclaim {
|
486 + | FromH1(H1CapacityReclaim),
|
487 + | FromH2(H2CapacityReclaim),
|
488 + | }
|
489 + |
|
490 + | impl CapacityReclaim {
|
491 + | fn reclaim_capacity(self) -> Option<AdmissionAction> {
|
492 + | match self {
|
493 + | Self::FromH1(reclaim) => reclaim.reclaim_capacity(),
|
494 + | Self::FromH2(reclaim) => reclaim.reclaim_capacity(),
|
495 + | }
|
576 496 | }
|
577 497 | }
|
578 498 |
|
579 499 | impl fmt::Debug for OriginAdmission {
|
580 500 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
581 501 | f.debug_struct("OriginAdmission")
|
582 502 | .field("state", &self.state)
|
583 503 | .finish()
|
584 504 | }
|
585 505 | }
|
586 506 |
|
587 507 | /// One admitted connection slot for a bounded origin.
|
588 508 | ///
|
589 509 | /// While this lease exists, one place in the origin's configured connection
|
590 510 | /// limit is occupied. The lease moves through capacity delivery, waiter
|
591 511 | /// resolution, and establishment into the installed
|
592 512 | /// [`super::connection::ConnectionState`]. Logical close drops the installed
|
593 513 | /// lease and makes that place available again.
|
594 514 | ///
|
595 515 | /// Dropping a lease may synchronously deliver capacity to another waiter, so
|
596 516 | /// callers must move it out of protected state before drop.
|
597 - | pub(crate) struct CapacityLease {
|
517 + | pub(in crate::client::pool) struct CapacityLease {
|
598 518 | /// Admission state to which this slot returns when the lease ends.
|
599 - | origin: Arc<OriginAdmission>,
|
519 + | admission: Arc<OriginAdmission>,
|
600 520 | /// Permit returned to admission when this lease ends.
|
601 - | permit: Option<Permit>,
|
521 + | permit: Option<CapacityPermit>,
|
602 522 | }
|
603 523 |
|
604 524 | impl CapacityLease {
|
605 525 | /// Takes ownership of a permit removed from admission's available set.
|
606 - | fn new(origin: Arc<OriginAdmission>, permit: Permit) -> Self {
|
526 + | fn new(admission: Arc<OriginAdmission>, permit: CapacityPermit) -> Self {
|
607 527 | Self {
|
608 - | origin,
|
528 + | admission,
|
609 529 | permit: Some(permit),
|
610 530 | }
|
611 531 | }
|
612 532 | }
|
613 533 |
|
614 534 | impl fmt::Debug for CapacityLease {
|
615 535 | fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
616 536 | f.debug_struct("CapacityLease")
|
617 537 | .field("permit", &self.permit)
|
618 538 | .finish_non_exhaustive()
|
619 539 | }
|
620 540 | }
|
621 541 |
|
622 542 | impl Drop for CapacityLease {
|
623 543 | fn drop(&mut self) {
|
624 544 | if let Some(permit) = self.permit.take() {
|
625 - | OriginAdmission::return_permit(&self.origin, permit);
|
545 + | OriginAdmission::return_permit(&self.admission, permit);
|
626 546 | }
|
627 547 | }
|
628 548 | }
|
629 549 |
|
630 - | /// Admission's acknowledgement of a requesting cell-side delivery attempt.
|
631 - | #[derive(Clone, Debug, Eq, PartialEq)]
|
632 - | enum DeliveryAckResult {
|
633 - | /// The requesting cell accepted the payload and optionally published its successor demand.
|
634 - | Accepted { successor: Option<DemandSnapshot> },
|
635 - | /// The same demand remains live at its existing order position.
|
636 - | RetrySameResidence,
|
637 - | /// The old demand ended and may have a successor.
|
638 - | Rejected { successor: Option<DemandSnapshot> },
|
639 - | }
|
640 - |
|
641 - | /// Mutable bounded-origin state protected by one admission lock.
|
550 + | /// Capacity not currently owned by a delivery, attempt, or connection.
|
642 551 | #[derive(Debug)]
|
643 - | struct AdmissionState {
|
644 - | /// Requesting cells, held weakly to avoid an ownership cycle.
|
645 - | cells: HashMap<PartitionId, Weak<OriginCell>>,
|
646 - | /// Permits not owned by a delivery, attempt, or connection.
|
552 + | struct CapacityBudget {
|
553 + | /// Configured connection bound used to check capacity conservation.
|
554 + | limit: usize,
|
555 + | /// Permits available for a new establishment attempt.
|
647 556 | available: usize,
|
648 557 | /// Next never-reused permit identity.
|
649 558 | next_permit_id: u64,
|
650 - | /// Cross-cell demand records and their origin-wide scheduling order.
|
651 - | demand_schedule: DemandSchedule,
|
652 - | /// HTTP/1 availability reports and cross-cell reuse operations.
|
653 - | h1: reuse::H1Reuse,
|
654 - | /// HTTP/2 advertisements and eligibility-group publication turns.
|
655 - | h2: H2Publication,
|
656 - | /// Next never-reused delivery identity.
|
657 - | next_delivery: u64,
|
658 - | /// Configured connection bound used to check capacity conservation.
|
659 - | limit: usize,
|
660 559 | }
|
661 560 |
|
662 - | impl AdmissionState {
|
663 - | /// Creates a ledger with all configured capacity available.
|
561 + | impl CapacityBudget {
|
664 562 | fn new(limit: NonZeroUsize) -> Self {
|
665 563 | let limit = limit.get();
|
666 564 | Self {
|
667 - | cells: HashMap::new(),
|
565 + | limit,
|
668 566 | available: limit,
|
669 567 | next_permit_id: 0,
|
670 - | demand_schedule: DemandSchedule::default(),
|
671 - | h1: reuse::H1Reuse::default(),
|
672 - | h2: H2Publication::default(),
|
673 - | next_delivery: 0,
|
674 - | limit,
|
675 568 | }
|
676 569 | }
|
677 570 |
|
678 - | /// Removes one available permit from admission.
|
679 - | fn take_permit(&mut self) -> Option<Permit> {
|
571 + | fn take_permit(&mut self) -> Option<CapacityPermit> {
|
680 572 | if self.available == 0 {
|
681 573 | return None;
|
682 574 | }
|
683 575 | let id = self.next_permit_id;
|
684 - | let next_permit_id = id.checked_add(1).expect("permit identity exhausted");
|
576 + | self.next_permit_id = id.checked_add(1).expect("permit identity exhausted");
|
685 577 | self.available -= 1;
|
686 - | self.next_permit_id = next_permit_id;
|
687 - | Some(Permit(id))
|
578 + | Some(CapacityPermit(id))
|
688 579 | }
|
689 580 |
|
690 - | /// Consumes a returned permit and restores one available slot.
|
691 - | fn return_permit(&mut self, _permit: Permit) {
|
581 + | fn return_permit(&mut self, _permit: CapacityPermit) {
|
692 582 | let available = self
|
693 583 | .available
|
694 584 | .checked_add(1)
|
695 585 | .expect("available capacity count overflowed");
|
696 586 | assert!(
|
697 587 | available <= self.limit,
|
698 588 | "available capacity exceeded the configured limit"
|
699 589 | );
|
700 590 | self.available = available;
|
701 591 | }
|
592 + | }
|
593 + |
|
594 + | /// Mutable bounded-origin state protected by one admission lock.
|
595 + | #[derive(Debug)]
|
596 + | struct AdmissionState {
|
597 + | /// Requesting cells, held weakly to avoid an ownership cycle.
|
598 + | cells: HashMap<PartitionId, Weak<OriginCell>>,
|
599 + | /// Conserved bounded-origin connection capacity.
|
600 + | capacity: CapacityBudget,
|
601 + | /// Canonical cross-cell demand and scheduling orders.
|
602 + | demand: DemandSchedule,
|
603 + | /// Admission's indexed view of HTTP/1 connection supply.
|
604 + | h1_supply: H1Supply,
|
605 + | /// Admission's indexed view of HTTP/2 connection supply.
|
606 + | h2_supply: H2Supply,
|
607 + | /// Next never-reused resource-to-demand assignment identity.
|
608 + | next_assignment_id: u64,
|
609 + | }
|
610 + |
|
611 + | impl AdmissionState {
|
612 + | /// Creates admission with all configured capacity available.
|
613 + | fn new(limit: NonZeroUsize) -> Self {
|
614 + | Self {
|
615 + | cells: HashMap::new(),
|
616 + | capacity: CapacityBudget::new(limit),
|
617 + | demand: DemandSchedule::default(),
|
618 + | h1_supply: H1Supply::default(),
|
619 + | h2_supply: H2Supply::default(),
|
620 + | next_assignment_id: 0,
|
621 + | }
|
622 + | }
|
623 + |
|
624 + | /// Removes one available permit from admission.
|
625 + | fn take_permit(&mut self) -> Option<CapacityPermit> {
|
626 + | self.capacity.take_permit()
|
627 + | }
|
628 + |
|
629 + | /// Consumes a returned permit and restores one available slot.
|
630 + | fn return_permit(&mut self, permit: CapacityPermit) {
|
631 + | self.capacity.return_permit(permit);
|
632 + | }
|
702 633 |
|
703 634 | #[cfg(test)]
|
704 635 | fn available_capacity(&self) -> usize {
|
705 - | self.available
|
636 + | self.capacity.available
|
706 637 | }
|
707 638 |
|
708 - | /// Applies one complete cell publication to cross-cell scheduling.
|
709 - | fn publish_demand(&mut self, requesting_partition: PartitionId, snapshot: DemandSnapshot) {
|
710 - | let old_group = self.demand_schedule.group_for(&requesting_partition);
|
711 - | self.demand_schedule.publish(requesting_partition, snapshot);
|
712 - | self.reconcile_demand_indexes(&requesting_partition, old_group);
|
639 + | /// Applies one complete cell snapshot to cross-cell demand scheduling.
|
640 + | fn apply_demand_snapshot(&mut self, requester: PartitionId, snapshot: DemandSnapshot) {
|
641 + | let old_group = self.demand.group_for(&requester);
|
642 + | self.demand.apply_snapshot(requester, snapshot);
|
643 + | self.reconcile_demand_indexes(&requester, old_group);
|
713 644 | }
|
714 645 |
|
715 646 | /// Refreshes protocol indexes derived from the canonical demand schedule.
|
716 647 | ///
|
717 648 | /// HTTP/1 indexes depend on the requesting partition's origin position.
|
718 - | /// HTTP/2 publication depends on both the previous and current eligibility
|
649 + | /// HTTP/2 route scheduling depends on both previous and current eligibility
|
719 650 | /// groups, so a group change must repair each view. The derived indexes do
|
720 651 | /// not own demand or ordering.
|
721 652 | fn reconcile_demand_indexes(
|
722 653 | &mut self,
|
723 654 | requesting_partition: &PartitionId,
|
724 655 | old_group: Option<EligibilityGroup>,
|
725 656 | ) {
|
726 - | self.h1
|
727 - | .reconcile_requesting_cell(requesting_partition, &self.demand_schedule);
|
657 + | self.h1_supply
|
658 + | .reconcile_requester(requesting_partition, &self.demand);
|
728 659 | if let Some(old_group) = old_group {
|
729 - | self.h2.reconcile_group(&old_group, &self.demand_schedule);
|
660 + | self.h2_supply.reconcile_group(&old_group, &self.demand);
|
730 661 | }
|
731 - | if let Some(group) = self.demand_schedule.group_for(requesting_partition) {
|
732 - | self.h2.reconcile_group(&group, &self.demand_schedule);
|
662 + | if let Some(group) = self.demand.group_for(requesting_partition) {
|
663 + | self.h2_supply.reconcile_group(&group, &self.demand);
|
733 664 | }
|
734 665 | }
|
735 666 |
|
736 667 | /// Pairs the oldest deliverable demand with one available permit.
|
737 - | fn schedule_one(&mut self) -> Option<PreparedCapacityDelivery> {
|
738 - | if !self.demand_schedule.head_is_queued() {
|
668 + | fn prepare_capacity_delivery(&mut self) -> Option<PreparedCapacityDelivery> {
|
669 + | if !self.demand.head_is_queued() {
|
739 670 | return None;
|
740 671 | }
|
741 672 |
|
742 673 | let permit = self.take_permit()?;
|
743 - | let delivery = self.take_delivery_id();
|
674 + | let assignment_id = self.take_assignment_id();
|
744 675 | let old_group = self
|
745 - | .demand_schedule
|
676 + | .demand
|
746 677 | .queued_head()
|
747 - | .and_then(|head| self.demand_schedule.group_for(&head.requesting_partition));
|
748 - | let scheduled = self
|
749 - | .demand_schedule
|
750 - | .reserve_origin_head(delivery)
|
678 + | .and_then(|head| self.demand.group_for(&head.requester));
|
679 + | let assignment = self
|
680 + | .demand
|
681 + | .prepare_origin_assignment(assignment_id)
|
751 682 | .expect("queued demand head disappeared");
|
752 - | self.reconcile_demand_indexes(&scheduled.requesting_partition, old_group);
|
753 - | Some(PreparedCapacityDelivery {
|
754 - | permit,
|
755 - | delivery,
|
756 - | requesting_partition: scheduled.requesting_partition,
|
757 - | demand: scheduled.demand,
|
758 - | })
|
759 - | }
|
760 - |
|
761 - | /// Delegates delivery-fence revalidation to the demand schedule.
|
762 - | #[cfg(test)]
|
763 - | fn delivery_is_current(
|
764 - | &self,
|
765 - | delivery: DeliveryId,
|
766 - | requesting_partition: &PartitionId,
|
767 - | demand: DemandId,
|
768 - | ) -> bool {
|
769 - | self.demand_schedule
|
770 - | .delivery_is_current(delivery, requesting_partition, demand)
|
683 + | self.reconcile_demand_indexes(&assignment.requester, old_group);
|
684 + | Some(PreparedCapacityDelivery { permit, assignment })
|
771 685 | }
|
772 686 |
|
773 - | /// Applies a requesting cell acknowledgement to the demand schedule.
|
774 - | fn finish_delivery(
|
687 + | /// Applies one detached assignment outcome to canonical demand.
|
688 + | fn settle_assignment(
|
775 689 | &mut self,
|
776 - | delivery: DeliveryId,
|
777 - | requesting_partition: &PartitionId,
|
778 - | result: DeliveryAckResult,
|
690 + | assignment: &DemandAssignment,
|
691 + | outcome: DemandAssignmentOutcome,
|
779 692 | ) {
|
780 - | let old_group = self.demand_schedule.group_for(requesting_partition);
|
781 - | self.demand_schedule
|
782 - | .finish_delivery(delivery, requesting_partition, result);
|
783 - | self.reconcile_demand_indexes(requesting_partition, old_group);
|
693 + | let old_group = self.demand.group_for(&assignment.requester);
|
694 + | self.demand.settle_assignment(assignment, outcome);
|
695 + | self.reconcile_demand_indexes(&assignment.requester, old_group);
|
784 696 | }
|
785 697 |
|
786 - | /// Reserves one eligibility-group head for identity-only H2 publication.
|
787 - | fn prepare_h2_publication(&mut self) -> Option<PreparedH2Publication> {
|
788 - | if !self.h2.has_ready_group() {
|
698 + | /// Reserves one eligibility-group head for an identity-only H2 route.
|
699 + | fn prepare_h2_route(&mut self) -> Option<PreparedH2Route> {
|
700 + | if !self.h2_supply.has_route_ready_group() {
|
789 701 | return None;
|
790 702 | }
|
791 - | let delivery = self.take_delivery_id();
|
792 - | self.h2.prepare(&mut self.demand_schedule, delivery)
|
703 + | let assignment_id = self.take_assignment_id();
|
704 + | self.h2_supply
|
705 + | .prepare_route(&mut self.demand, assignment_id)
|
793 706 | }
|
794 707 |
|
795 - | /// Allocates a delivery-fence identity that is never reused by this origin.
|
796 - | fn take_delivery_id(&mut self) -> DeliveryId {
|
797 - | let value = self.next_delivery;
|
798 - | self.next_delivery = value.checked_add(1).expect("delivery identity exhausted");
|
799 - | DeliveryId(value)
|
708 + | /// Allocates an assignment identity that is never reused by this origin.
|
709 + | fn take_assignment_id(&mut self) -> DemandAssignmentId {
|
710 + | let value = self.next_assignment_id;
|
711 + | self.next_assignment_id = value
|
712 + | .checked_add(1)
|
713 + | .expect("demand assignment identity exhausted");
|
714 + | DemandAssignmentId(value)
|
800 715 | }
|
801 716 | }
|
802 717 |
|
803 718 | /// Test snapshot of the bounded-origin capacity and demand ledger.
|
804 719 | #[cfg(test)]
|
805 720 | #[derive(Debug, Eq, PartialEq)]
|
806 - | struct AdmissionCounts {
|
721 + | struct AdmissionProbe {
|
807 722 | /// Configured connection capacity.
|
808 723 | limit: usize,
|
809 724 | /// Permits not currently owned by a connection or delivery.
|
810 725 | available: usize,
|
811 - | /// Demands linked in origin order, including a delivery fence.
|
726 + | /// Demands linked in origin order, including a pending assignment.
|
812 727 | ordered: usize,
|
813 728 | /// Demands eligible to start a delivery.
|
814 729 | queued: usize,
|
815 - | /// Demands currently fenced by a crossing delivery.
|
816 - | delivering: usize,
|
730 + | /// Demands currently owned by a detached assignment.
|
731 + | assigned: usize,
|
817 732 | }
|
818 733 |
|
819 734 | #[cfg(all(test, not(smithy_http_client_loom)))]
|
820 735 | mod tests {
|
821 736 | use super::*;
|
822 737 | use crate::client::pool::partition::PartitionId;
|
823 738 |
|
824 739 | fn cell(origin: &Arc<OriginAdmission>, partition: usize) -> Arc<OriginCell> {
|
825 740 | let cell = Arc::new(OriginCell::new(
|
826 741 | PartitionId::from_index(partition),
|
827 742 | OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "example.com", None).unwrap(),
|
828 743 | EligibilityGroup::Pool,
|
829 744 | Some(origin.clone()),
|
830 745 | None,
|
831 746 | ));
|
832 747 | OriginAdmission::register_cell(origin, cell)
|
833 748 | }
|
834 749 |
|
835 750 | fn demand(id: u64) -> DemandSnapshot {
|
836 751 | DemandSnapshot::active(
|
837 752 | DemandId::from_u64(id),
|
838 753 | SnapshotVersion::INITIAL,
|
839 754 | ProtocolRequirement::H1Compatible,
|
840 755 | EligibilityGroup::Pool,
|
841 756 | )
|
842 757 | }
|
843 758 |
|
844 759 | #[test]
|
845 760 | #[should_panic(expected = "cell origin did not match its admission authority")]
|
846 761 | fn registration_rejects_a_cell_from_another_origin() {
|
847 762 | let admission = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
848 763 | let candidate = Arc::new(OriginCell::new(
|
849 764 | PartitionId::from_index(1),
|
850 765 | OriginKey::from_parts(http_1x::uri::Scheme::HTTPS, "other.example.com", None).unwrap(),
|
851 766 | EligibilityGroup::Pool,
|
852 767 | Some(admission.clone()),
|
853 768 | None,
|
854 769 | ));
|
855 770 |
|
856 771 | OriginAdmission::register_cell(&admission, candidate);
|
857 772 | }
|
858 773 |
|
859 774 | #[test]
|
860 775 | fn permits_are_linear_and_never_reused() {
|
861 776 | let mut state = AdmissionState::new(NonZeroUsize::new(2).unwrap());
|
862 777 |
|
863 - | assert_eq!(2, state.available);
|
778 + | assert_eq!(2, state.available_capacity());
|
864 779 | let first = state.take_permit().unwrap();
|
865 780 | let first_id = first.0;
|
866 781 | let second = state.take_permit().unwrap();
|
867 - | assert_eq!(0, state.available);
|
782 + | assert_eq!(0, state.available_capacity());
|
868 783 | assert!(state.take_permit().is_none());
|
869 784 |
|
870 785 | state.return_permit(first);
|
871 786 | let third = state.take_permit().unwrap();
|
872 787 | assert_ne!(first_id, third.0);
|
873 788 | state.return_permit(second);
|
874 789 | state.return_permit(third);
|
875 - | assert_eq!(2, state.available);
|
790 + | assert_eq!(2, state.available_capacity());
|
876 791 | }
|
877 792 |
|
878 793 | #[test]
|
879 794 | fn equal_or_older_snapshots_do_not_replace_current_demand() {
|
880 795 | let mut state = AdmissionState::new(NonZeroUsize::new(1).unwrap());
|
881 796 | let requesting_partition = PartitionId::from_index(1);
|
882 797 | let current = demand(2);
|
883 - | state.publish_demand(requesting_partition, current.clone());
|
884 - | state.publish_demand(
|
798 + | state.apply_demand_snapshot(requesting_partition, current.clone());
|
799 + | state.apply_demand_snapshot(
|
885 800 | requesting_partition,
|
886 801 | DemandSnapshot::inactive(DemandId::from_u64(2), SnapshotVersion::INITIAL),
|
887 802 | );
|
888 - | state.publish_demand(requesting_partition, demand(1));
|
803 + | state.apply_demand_snapshot(requesting_partition, demand(1));
|
889 804 |
|
890 - | assert_eq!(1, state.demand_schedule.len());
|
805 + | assert_eq!(1, state.demand.len());
|
891 806 | assert_eq!(
|
892 807 | Some(¤t),
|
893 - | state.demand_schedule.latest_for_test(&requesting_partition)
|
808 + | state.demand.latest_for_test(&requesting_partition)
|
894 809 | );
|
895 810 | }
|
896 811 |
|
897 812 | #[test]
|
898 813 | fn cancellation_churn_does_not_retain_order_entries() {
|
899 814 | let mut state = AdmissionState::new(NonZeroUsize::new(1).unwrap());
|
900 815 | let held = state.take_permit().unwrap();
|
901 816 | let requesting_partition = PartitionId::from_index(1);
|
902 817 |
|
903 818 | for id in 0..2_000 {
|
904 819 | let id = DemandId::from_u64(id);
|
905 - | state.publish_demand(
|
820 + | state.apply_demand_snapshot(
|
906 821 | requesting_partition,
|
907 822 | DemandSnapshot::active(
|
908 823 | id,
|
909 824 | SnapshotVersion::INITIAL,
|
910 825 | ProtocolRequirement::H1Compatible,
|
911 826 | EligibilityGroup::Pool,
|
912 827 | ),
|
913 828 | );
|
914 - | state.publish_demand(
|
829 + | state.apply_demand_snapshot(
|
915 830 | requesting_partition,
|
916 831 | DemandSnapshot::inactive(id, SnapshotVersion::INITIAL.next()),
|
917 832 | );
|
918 833 | }
|
919 834 |
|
920 - | assert_eq!(0, state.demand_schedule.len());
|
835 + | assert_eq!(0, state.demand.len());
|
921 836 | state.return_permit(held);
|
922 837 | }
|
923 838 |
|
924 839 | #[test]
|
925 840 | fn removing_middle_and_tail_demands_repairs_order() {
|
926 841 | let mut state = AdmissionState::new(NonZeroUsize::new(1).unwrap());
|
927 842 | let held = state.take_permit().unwrap();
|
928 843 | let targets: Vec<_> = (1..=5).map(PartitionId::from_index).collect();
|
929 844 |
|
930 845 | for (index, requesting_partition) in targets[..4].iter().enumerate() {
|
931 - | state.publish_demand(*requesting_partition, demand(index as u64 + 1));
|
846 + | state.apply_demand_snapshot(*requesting_partition, demand(index as u64 + 1));
|
932 847 | }
|
933 - | state.publish_demand(
|
848 + | state.apply_demand_snapshot(
|
934 849 | targets[1],
|
935 850 | DemandSnapshot::inactive(DemandId::from_u64(2), SnapshotVersion::INITIAL.next()),
|
936 851 | );
|
937 - | state.publish_demand(
|
852 + | state.apply_demand_snapshot(
|
938 853 | targets[3],
|
939 854 | DemandSnapshot::inactive(DemandId::from_u64(4), SnapshotVersion::INITIAL.next()),
|
940 855 | );
|
941 - | state.publish_demand(targets[4], demand(5));
|
942 - | assert_eq!(3, state.demand_schedule.len());
|
856 + | state.apply_demand_snapshot(targets[4], demand(5));
|
857 + | assert_eq!(3, state.demand.len());
|
943 858 |
|
944 859 | state.return_permit(held);
|
945 860 | for expected in [&targets[0], &targets[2], &targets[4]] {
|
946 - | let pending = state.schedule_one().unwrap();
|
947 - | assert_eq!(expected, &pending.requesting_partition);
|
948 - | state.finish_delivery(
|
949 - | pending.delivery,
|
950 - | &pending.requesting_partition,
|
951 - | DeliveryAckResult::Accepted { successor: None },
|
861 + | let pending = state.prepare_capacity_delivery().unwrap();
|
862 + | assert_eq!(expected, &pending.assignment.requester);
|
863 + | state.settle_assignment(
|
864 + | &pending.assignment,
|
865 + | DemandAssignmentOutcome::Accepted { successor: None },
|
952 866 | );
|
953 867 | state.return_permit(pending.permit);
|
954 868 | }
|
955 - | assert_eq!(0, state.demand_schedule.len());
|
869 + | assert_eq!(0, state.demand.len());
|
956 870 | }
|
957 871 |
|
958 872 | #[test]
|
959 873 | fn new_demand_moves_queued_cell_to_the_tail() {
|
960 874 | let mut state = AdmissionState::new(NonZeroUsize::new(1).unwrap());
|
961 875 | let held = state.take_permit().unwrap();
|
962 876 | let first = PartitionId::from_index(1);
|
963 877 | let second = PartitionId::from_index(2);
|
964 - | state.publish_demand(first, demand(1));
|
965 - | state.publish_demand(second, demand(2));
|
966 - | state.publish_demand(first, demand(3));
|
878 + | state.apply_demand_snapshot(first, demand(1));
|
879 + | state.apply_demand_snapshot(second, demand(2));
|
880 + | state.apply_demand_snapshot(first, demand(3));
|
967 881 | state.return_permit(held);
|
968 882 |
|
969 - | assert_eq!(second, state.schedule_one().unwrap().requesting_partition);
|
970 - | }
|
971 - |
|
972 - | #[test]
|
973 - | fn delivery_currency_includes_the_demand_id() {
|
974 - | let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
975 - | let requesting_partition = cell(&origin, 1);
|
976 - | let delivery = OriginAdmission::publish_without_driving(
|
977 - | &origin,
|
978 - | requesting_partition.id().partition(),
|
979 - | demand(1),
|
980 - | )
|
981 - | .unwrap();
|
982 - | assert!(delivery.is_current());
|
983 - |
|
984 - | origin
|
985 - | .state
|
986 - | .lock()
|
987 - | .publish_demand(requesting_partition.id().partition(), demand(2));
|
988 - | assert!(!delivery.is_current());
|
989 - | delivery.reject(None);
|
990 - | }
|
991 - |
|
992 - | #[test]
|
993 - | fn stale_successor_cannot_leave_active_demand_idle() {
|
994 - | let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
995 - | let requesting_partition = cell(&origin, 1);
|
996 - | let delivery = OriginAdmission::publish_without_driving(
|
997 - | &origin,
|
998 - | requesting_partition.id().partition(),
|
999 - | demand(1),
|
1000 - | )
|
1001 - | .unwrap();
|
1002 - | origin.state.lock().publish_demand(
|
1003 - | requesting_partition.id().partition(),
|
1004 - | DemandSnapshot::active(
|
1005 - | DemandId::from_u64(1),
|
1006 - | SnapshotVersion::INITIAL.next(),
|
1007 - | ProtocolRequirement::H1Compatible,
|
1008 - | EligibilityGroup::Pool,
|
1009 - | ),
|
883 + | assert_eq!(
|
884 + | second,
|
885 + | state
|
886 + | .prepare_capacity_delivery()
|
887 + | .unwrap()
|
888 + | .assignment
|
889 + | .requester
|
1010 890 | );
|
1011 - | delivery.reject(Some(demand(1)));
|
1012 - |
|
1013 - | assert_eq!(1, origin.counts().available);
|
1014 - | assert_eq!(0, origin.counts().ordered);
|
1015 - | }
|
1016 - |
|
1017 - | #[test]
|
1018 - | fn dropped_delivery_refunnels_capacity_and_preserves_order() {
|
1019 - | let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1020 - | let first = cell(&origin, 1);
|
1021 - | let second = cell(&origin, 2);
|
1022 - | let (first_waiter, first_demand) =
|
1023 - | first.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1024 - | let (second_waiter, second_demand) =
|
1025 - | second.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1026 - | let delivery =
|
1027 - | OriginAdmission::publish_without_driving(&origin, first.id().partition(), first_demand)
|
1028 - | .unwrap();
|
1029 - | {
|
1030 - | let mut state = origin.state.lock();
|
1031 - | state.publish_demand(second.id().partition(), second_demand);
|
1032 - | }
|
1033 - | drop(delivery);
|
1034 - |
|
1035 - | let first_lease = OriginCell::take_ready_lease(&first, first_waiter)
|
1036 - | .expect("dropped delivery did not retry the original head");
|
1037 - | assert!(OriginCell::take_ready_lease(&second, second_waiter).is_none());
|
1038 - | assert_eq!(1, origin.counts().ordered);
|
1039 - |
|
1040 - | drop(first_lease);
|
1041 - | let second_lease = OriginCell::take_ready_lease(&second, second_waiter)
|
1042 - | .expect("younger demand did not run after the original head");
|
1043 - | drop(second_lease);
|
1044 891 | }
|
1045 892 |
|
1046 893 | #[test]
|
1047 894 | fn losing_registered_cell_returns_capacity_after_admission_unlocks() {
|
1048 895 | let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1049 896 | let retained = cell(&origin, 1);
|
1050 897 | let candidate = Arc::new(OriginCell::new(
|
1051 898 | retained.id().partition(),
|
1052 899 | retained.id().origin().clone(),
|
1053 900 | EligibilityGroup::Pool,
|
1054 901 | Some(origin.clone()),
|
1055 902 | None,
|
1056 903 | ));
|
1057 904 | assert_eq!(retained.id(), candidate.id());
|
1058 905 |
|
1059 906 | let (_waiter, snapshot) =
|
1060 907 | candidate.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1061 908 | let mut delivery =
|
1062 - | OriginAdmission::publish_without_driving(&origin, candidate.id().partition(), snapshot)
|
909 + | OriginAdmission::submit_without_running(&origin, candidate.id().partition(), snapshot)
|
1063 910 | .expect("candidate demand did not reserve capacity");
|
1064 - | assert!(delivery.materialize_for_test());
|
911 + | assert!(delivery.resolve_payload_for_test());
|
1065 912 | assert!(OriginCell::receive_delivery(&candidate, delivery).is_none());
|
1066 913 | assert_eq!(0, origin.available_capacity_for_test());
|
1067 914 |
|
1068 915 | let winner = OriginAdmission::register_cell(&origin, candidate);
|
1069 916 | assert!(Arc::ptr_eq(&retained, &winner));
|
1070 917 | assert_eq!(
|
1071 918 | 1,
|
1072 919 | origin.available_capacity_for_test(),
|
1073 920 | "losing candidate did not return capacity after registration"
|
1074 921 | );
|
1075 922 | }
|
1076 923 |
|
1077 - | #[test]
|
1078 - | fn expired_requesting_cell_refunnels_capacity() {
|
1079 - | let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1080 - | let requesting_partition = cell(&origin, 1);
|
1081 - | let requesting_cell_id = requesting_partition.id().partition();
|
1082 - | let (_waiter, snapshot) =
|
1083 - | requesting_partition.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1084 - | let delivery =
|
1085 - | OriginAdmission::publish_without_driving(&origin, requesting_cell_id, snapshot)
|
1086 - | .unwrap();
|
1087 - |
|
1088 - | drop(requesting_partition);
|
1089 - | assert!(origin.cell(&requesting_cell_id).is_none());
|
1090 - | OriginAdmission::drive(Some(AdmissionAction::Delivery(delivery)));
|
1091 - |
|
1092 - | let counts = origin.counts();
|
1093 - | assert_eq!(1, counts.available);
|
1094 - | assert_eq!(0, counts.delivering);
|
1095 - | assert_eq!(0, counts.ordered);
|
1096 - | }
|
1097 - |
|
1098 924 | #[test]
|
1099 925 | fn separate_origins_conserve_capacity_independently() {
|
1100 926 | let first = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1101 927 | let second = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1102 928 | let first_cell = cell(&first, 1);
|
1103 929 | let second_cell = cell(&second, 1);
|
1104 930 |
|
1105 - | let first_delivery = OriginAdmission::publish_without_driving(
|
1106 - | &first,
|
1107 - | first_cell.id().partition(),
|
1108 - | demand(1),
|
1109 - | )
|
1110 - | .unwrap();
|
1111 - | let second_delivery = OriginAdmission::publish_without_driving(
|
931 + | let first_delivery =
|
932 + | OriginAdmission::submit_without_running(&first, first_cell.id().partition(), demand(1))
|
933 + | .unwrap();
|
934 + | let second_delivery = OriginAdmission::submit_without_running(
|
1112 935 | &second,
|
1113 936 | second_cell.id().partition(),
|
1114 937 | demand(1),
|
1115 938 | )
|
1116 939 | .unwrap();
|
1117 940 | assert_eq!(0, first.available_capacity_for_test());
|
1118 941 | assert_eq!(0, second.available_capacity_for_test());
|
1119 942 |
|
1120 943 | drop(first_delivery);
|
1121 944 | assert_eq!(1, first.available_capacity_for_test());
|
1122 945 | assert_eq!(0, second.available_capacity_for_test());
|
1123 946 |
|
1124 947 | drop(second_delivery);
|
1125 948 | assert_eq!(1, second.available_capacity_for_test());
|
1126 949 | }
|
1127 950 | }
|
1128 951 |
|
1129 952 | #[cfg(all(test, smithy_http_client_loom))]
|
1130 953 | mod loom_tests {
|
1131 954 | use super::*;
|
1132 955 | use crate::client::pool::partition::PartitionId;
|
1133 956 |
|
1134 957 | fn id() -> PartitionId {
|
1135 958 | PartitionId::from_index(1)
|
1136 959 | }
|
1137 960 |
|
1138 961 | fn demand() -> DemandSnapshot {
|
1139 962 | DemandSnapshot::active(
|
1140 963 | DemandId::from_u64(1),
|
1141 964 | SnapshotVersion::INITIAL,
|
1142 965 | ProtocolRequirement::H1Compatible,
|
1143 966 | EligibilityGroup::Pool,
|
1144 967 | )
|
1145 968 | }
|
1146 969 |
|
1147 970 | #[test]
|
1148 - | fn release_and_demand_publication_conserve_one_permit() {
|
971 + | fn release_and_demand_submission_conserve_one_permit() {
|
1149 972 | loom::model(|| {
|
1150 973 | let origin = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1151 974 | let permit = origin.state.lock().take_permit().unwrap();
|
1152 975 | let lease = CapacityLease::new(origin.clone(), permit);
|
1153 976 |
|
1154 977 | let release = loom::thread::spawn(move || drop(lease));
|
1155 978 | let publish_origin = origin.clone();
|
1156 979 | let publish = loom::thread::spawn(move || {
|
1157 980 | let delivery =
|
1158 - | OriginAdmission::publish_without_driving(&publish_origin, id(), demand());
|
981 + | OriginAdmission::submit_without_running(&publish_origin, id(), demand());
|
1159 982 | drop(delivery);
|
1160 983 | });
|
1161 984 | release.join().unwrap();
|
1162 985 | publish.join().unwrap();
|
1163 986 |
|
1164 - | assert_eq!(1, origin.counts().available);
|
1165 - | assert!(origin.counts().ordered <= 1);
|
987 + | assert_eq!(1, origin.probe().available);
|
988 + | assert!(origin.probe().ordered <= 1);
|
1166 989 | });
|
1167 990 | }
|
1168 991 |
|
1169 992 | #[test]
|
1170 - | fn cancellation_preserves_an_outstanding_delivery_fence() {
|
993 + | fn cancellation_preserves_an_outstanding_demand_assignment() {
|
1171 994 | loom::model(|| {
|
1172 995 | use loom::sync::atomic::{AtomicBool, Ordering};
|
1173 996 |
|
1174 997 | let origin = OriginAdmission::for_test(NonZeroUsize::new(2).unwrap());
|
1175 998 | let requesting_partition = id();
|
1176 999 | let delivery =
|
1177 - | OriginAdmission::publish_without_driving(&origin, requesting_partition, demand())
|
1000 + | OriginAdmission::submit_without_running(&origin, requesting_partition, demand())
|
1178 1001 | .unwrap();
|
1179 1002 | let release = Arc::new(AtomicBool::new(false));
|
1180 1003 |
|
1181 1004 | let delivery_release = release.clone();
|
1182 1005 | let dropped = loom::thread::spawn(move || {
|
1183 1006 | while !delivery_release.load(Ordering::Acquire) {
|
1184 1007 | loom::thread::yield_now();
|
1185 1008 | }
|
1186 1009 | drop(delivery);
|
1187 1010 | });
|
1188 1011 |
|
1189 - | origin.state.lock().publish_demand(
|
1012 + | origin.state.lock().apply_demand_snapshot(
|
1190 1013 | requesting_partition,
|
1191 1014 | DemandSnapshot::inactive(DemandId::from_u64(1), SnapshotVersion::INITIAL.next()),
|
1192 1015 | );
|
1193 - | let duplicate = OriginAdmission::publish_without_driving(
|
1016 + | let duplicate = OriginAdmission::submit_without_running(
|
1194 1017 | &origin,
|
1195 1018 | requesting_partition,
|
1196 1019 | DemandSnapshot::active(
|
1197 1020 | DemandId::from_u64(2),
|
1198 1021 | SnapshotVersion::INITIAL,
|
1199 1022 | ProtocolRequirement::H1Compatible,
|
1200 1023 | EligibilityGroup::Pool,
|
1201 1024 | ),
|
1202 1025 | );
|
1203 1026 |
|
1204 1027 | release.store(true, Ordering::Release);
|
1205 1028 | dropped.join().unwrap();
|
1206 1029 | assert!(
|
1207 1030 | duplicate.is_none(),
|
1208 - | "outstanding delivery fence admitted a second delivery"
|
1031 + | "outstanding demand assignment admitted a second delivery"
|
1209 1032 | );
|
1210 1033 |
|
1211 - | let counts = origin.counts();
|
1212 - | assert_eq!(2, counts.available);
|
1213 - | assert_eq!(0, counts.delivering);
|
1214 - | assert_eq!(0, counts.ordered);
|
1034 + | let probe = origin.probe();
|
1035 + | assert_eq!(2, probe.available);
|
1036 + | assert_eq!(0, probe.assigned);
|
1037 + | assert_eq!(0, probe.ordered);
|
1215 1038 | });
|
1216 1039 | }
|
1217 1040 | }
|