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