6 6 | //! State local to one partition and origin.
|
7 7 | //!
|
8 8 | //! An [`OriginCell`] owns its acquisition order and protocol records. A
|
9 9 | //! bounded cell publishes aggregate demand to [`OriginAdmission`] after
|
10 10 | //! releasing its lock, and admission returns capacity through the same
|
11 11 | //! unlocked boundary.
|
12 12 | //!
|
13 13 | //! [`AcquisitionQueue`] owns every live acquisition attempt, including the
|
14 14 | //! bounded FIFO, delivery crossings, establishment launch, cancellation, and
|
15 15 | //! terminal results. Its module documents the complete transition model.
|
16 - | //! [`CellState`] couples that model to HTTP/1 sender residence, HTTP/2
|
17 - | //! flight and generation residence, generation gates, and the cross-cell reuse
|
18 - | //! reservation. A local HTTP/1 return either satisfies one acquisition or
|
19 - | //! becomes idle, never both. An HTTP/2 activation becomes visible only with a
|
20 - | //! prospective generation lease and one recorded gate opportunity.
|
16 + | //! [`CellState`] couples that model to HTTP/1 sender ownership, HTTP/2 flights,
|
17 + | //! connection generations, activation gates, and cross-cell reuse. A local
|
18 + | //! HTTP/1 return either satisfies one acquisition or becomes idle, never both.
|
19 + | //! An HTTP/2 activation becomes visible only with a prospective request claim
|
20 + | //! and one recorded activation turn.
|
21 21 | //!
|
22 22 | //! Permits, senders, admission updates, and task wakes are detached from
|
23 23 | //! mutable state before their fallback or callback runs. This keeps drop and
|
24 24 | //! wake paths outside the cell lock.
|
25 25 |
|
26 26 | pub(super) mod h1;
|
27 27 | pub(super) mod h2;
|
28 28 | mod waiters;
|
29 29 |
|
30 30 | #[cfg(test)]
|
31 31 | use self::h1::H1CloseHandle;
|
32 32 | #[cfg(all(test, not(smithy_http_client_loom)))]
|
33 33 | use self::h1::H1DriverGuard;
|
34 34 | #[cfg(test)]
|
35 35 | use self::h1::H1Sender;
|
36 - | use self::h1::{H1Records, H1ReuseReservation, H1Selection, OwnedH1};
|
36 + | use self::h1::{H1CellState, H1Selection, OwnedH1Sender};
|
37 37 | #[cfg(test)]
|
38 - | use self::waiters::CellSnapshot;
|
38 + | use self::waiters::AcquisitionProbe;
|
39 39 | pub(in crate::client::pool) use self::waiters::WaiterId;
|
40 - | use self::waiters::{AcquisitionQueue, DeliveryReservation, ResultInstallError};
|
41 - | use super::admission::reuse::{H1Availability, H1AvailabilitySnapshot, ReuseId};
|
40 + | use self::waiters::{AcquisitionQueue, CellCommitError, CellCommitOutcome, DeliveryReservation};
|
42 41 | use super::admission::{
|
43 - | AdmissionAction, CapacityLease, DeliveryGuard, DemandSnapshot, H2AdvertisementSnapshot,
|
44 - | OriginAdmission, ProtocolRequirement,
|
42 + | AdmissionAction, CapacityLease, DeliveryGuard, DemandSnapshot, H1MatchId, H1SupplyStatus,
|
43 + | H2SupplyStatus, OriginAdmission, ProtocolRequirement, SupplyRevision,
|
45 44 | };
|
46 45 | use super::connection::CloseReason;
|
47 46 | #[cfg(test)]
|
48 47 | use super::connection::ConnectionInfo;
|
49 48 | #[cfg(test)]
|
50 49 | use super::connection::ConnectionState;
|
51 50 | use super::maintenance::PartitionMaintenance;
|
52 51 | use super::origin::OriginKey;
|
53 52 | use super::partition::{EligibilityGroup, PartitionId};
|
54 53 | use crate::sync::{Arc, Mutex};
|
55 54 | #[cfg(test)]
|
56 55 | use aws_smithy_runtime_api::client::connection::ConnectionId;
|
57 56 | use aws_smithy_runtime_api::client::result::ConnectorError;
|
58 57 | use std::task::{Context, Poll};
|
59 58 | use std::time::SystemTime;
|
60 59 |
|
61 60 | /// Stable identity of an [`OriginCell`].
|
62 61 | #[derive(Clone, Debug, Eq, Hash, PartialEq)]
|
63 - | pub(crate) struct CellId {
|
62 + | pub(in crate::client::pool) struct CellId {
|
64 63 | /// Partition that owns this cell's connection work.
|
65 64 | partition: PartitionId,
|
66 65 | /// Canonical origin whose requests share this cell.
|
67 66 | origin: OriginKey,
|
68 67 | }
|
69 68 |
|
70 69 | impl CellId {
|
71 70 | pub(super) fn new(partition: PartitionId, origin: OriginKey) -> Self {
|
72 71 | Self { partition, origin }
|
73 72 | }
|
74 73 |
|
75 - | pub(crate) fn partition(&self) -> PartitionId {
|
74 + | pub(in crate::client::pool) fn partition(&self) -> PartitionId {
|
76 75 | self.partition
|
77 76 | }
|
78 77 |
|
79 - | pub(crate) fn origin(&self) -> &OriginKey {
|
78 + | pub(in crate::client::pool) fn origin(&self) -> &OriginKey {
|
80 79 | &self.origin
|
81 80 | }
|
82 81 | }
|
83 82 |
|
84 83 | /// Stable state shared by requests for one partition and canonical origin.
|
85 - | pub(crate) struct OriginCell {
|
84 + | pub(in crate::client::pool) struct OriginCell {
|
86 85 | /// Complete identity used by partition and admission indexes.
|
87 86 | id: CellId,
|
88 87 | /// Partitions whose connections may satisfy this cell's demand.
|
89 88 | eligibility_group: EligibilityGroup,
|
90 89 | /// Origin-wide authority present only when capacity is bounded.
|
91 90 | admission: Option<Arc<OriginAdmission>>,
|
92 91 | /// Partition-owned idle deadline scheduler.
|
93 92 | maintenance: Option<Arc<PartitionMaintenance>>,
|
94 93 | /// Waiter order and protocol records protected by one cell-local lock.
|
95 94 | state: Mutex<CellState>,
|
96 95 | }
|
97 96 |
|
98 97 | /// Mutable state protected by one partition-origin cell lock.
|
99 98 | ///
|
100 - | /// Waiter outcomes and protocol residence share this lock. HTTP/1 returns,
|
99 + | /// Waiter outcomes and protocol state share this lock. HTTP/1 returns,
|
101 100 | /// HTTP/2 activation gates, flight participants, and cancellation therefore
|
102 101 | /// commit against one acquisition order.
|
103 - | #[derive(Debug, Default)]
|
102 + | #[derive(Debug)]
|
104 103 | struct CellState {
|
105 104 | /// Cell-local acquisition order and delivered results.
|
106 - | waiters: AcquisitionQueue,
|
107 - | /// Cell-owned HTTP/1 records and reusable sender order.
|
108 - | h1: H1Records,
|
109 - | /// Cell-owned HTTP/2 flight and installed generations.
|
110 - | h2: h2::H2Records,
|
111 - | /// One cross-cell reuse reservation and its local fairness debt.
|
112 - | reuse: H1ReuseReservation,
|
113 - | /// Last HTTP/1 availability published to bounded-origin admission.
|
114 - | published_h1_availability: Option<H1Availability>,
|
115 - | /// Revision used to reject availability reports that cross out of order.
|
116 - | h1_availability_revision: u64,
|
117 - | /// Last complete H2 advertisement reported to origin admission.
|
118 - | published_h2_advertisement: Option<(h2::H2GenerationId, bool)>,
|
119 - | /// Revision used to reject H2 advertisements that cross out of order.
|
120 - | h2_advertisement_revision: u64,
|
105 + | acquisitions: AcquisitionQueue,
|
106 + | /// Complete HTTP/1 sender and peer-reservation state.
|
107 + | h1: H1CellState,
|
108 + | /// Complete HTTP/2 acquisition state for this cell.
|
109 + | h2: h2::H2CellState,
|
110 + | /// Last admission-facing HTTP/1 status and its monotonic revision.
|
111 + | h1_supply_revision: SupplyRevision<H1SupplyStatus>,
|
112 + | /// Last admission-facing HTTP/2 status and its monotonic revision.
|
113 + | h2_supply_revision: SupplyRevision<H2SupplyStatus>,
|
114 + | }
|
115 + |
|
116 + | impl Default for CellState {
|
117 + | fn default() -> Self {
|
118 + | Self {
|
119 + | acquisitions: AcquisitionQueue::default(),
|
120 + | h1: H1CellState::default(),
|
121 + | h2: h2::H2CellState::default(),
|
122 + | h1_supply_revision: SupplyRevision::new(
|
123 + | 0,
|
124 + | H1SupplyStatus {
|
125 + | has_returnable_connection: false,
|
126 + | peer_use_blocked: false,
|
127 + | },
|
128 + | ),
|
129 + | h2_supply_revision: SupplyRevision::new(0, H2SupplyStatus::Unavailable),
|
130 + | }
|
131 + | }
|
121 132 | }
|
122 133 |
|
123 134 | impl CellState {
|
124 - | /// Checks the coupled waiter, sender, and reuse-reservation state machines.
|
135 + | /// Checks the coupled waiter and protocol state machines.
|
125 136 | fn assert_consistent(&self) {
|
126 137 | #[cfg(any(debug_assertions, test))]
|
138 + | {
|
127 139 | if std::thread::panicking() {
|
128 140 | return;
|
129 141 | }
|
130 - | self.waiters.assert_consistent();
|
142 + | self.acquisitions.assert_consistent();
|
131 143 | self.h1.assert_consistent();
|
132 144 | self.h2.assert_consistent();
|
133 - | #[cfg(any(debug_assertions, test))]
|
134 - | self.h2.assert_pending_waiters(&self.waiters);
|
135 - | self.reuse
|
136 - | .assert_consistent(self.h1.supports_installed_reuse());
|
145 + | self.h2.assert_pending_waiters(&self.acquisitions);
|
146 + | }
|
137 147 | }
|
138 148 |
|
139 - | /// Reports the HTTP/1 availability admission needs without exposing cell internals.
|
140 - | fn h1_availability(&self) -> H1Availability {
|
141 - | let local_h1_demand = self.waiters.can_accept_h1();
|
142 - | H1Availability {
|
143 - | advertised: self.h1.has_returnable(),
|
144 - | blocked: !self.reuse.is_available()
|
145 - | || self.reuse.blocks_peer_reuse(local_h1_demand)
|
146 - | || self.waiters.has_prior_h1_candidate(),
|
149 + | /// Derives the complete HTTP/1 status admission needs from cell-owned state.
|
150 + | fn h1_supply_status(&self) -> H1SupplyStatus {
|
151 + | let local_h1_demand = self.acquisitions.has_h1_compatible_waiter();
|
152 + | H1SupplyStatus {
|
153 + | has_returnable_connection: self.h1.has_returnable(),
|
154 + | peer_use_blocked: !self.h1.peer_reservation_available()
|
155 + | || self.h1.blocks_peer_selection(local_h1_demand)
|
156 + | || self.acquisitions.has_prior_h1_waiter(),
|
147 157 | }
|
148 158 | }
|
149 159 |
|
150 - | /// Returns an availability update only when admission's view must change.
|
151 - | fn take_h1_availability_update(&mut self) -> Option<H1AvailabilitySnapshot> {
|
152 - | let current = self.h1_availability();
|
153 - | if self.published_h1_availability == Some(current) {
|
160 + | /// Returns an HTTP/1 revision only when admission's view must change.
|
161 + | ///
|
162 + | /// The first submission advances to revision one even for the initial
|
163 + | /// unavailable status because it also drives retained H1 matching.
|
164 + | fn take_h1_supply_update(&mut self) -> Option<SupplyRevision<H1SupplyStatus>> {
|
165 + | let current = self.h1_supply_status();
|
166 + | if self.h1_supply_revision.revision != 0 && self.h1_supply_revision.status == current {
|
154 167 | return None;
|
155 168 | }
|
156 - | Some(self.record_h1_availability(current))
|
169 + | Some(self.advance_h1_supply_revision(current))
|
157 170 | }
|
158 171 |
|
159 - | /// Returns the current versioned availability for a reuse transition.
|
160 - | fn report_h1_availability(&mut self) -> H1AvailabilitySnapshot {
|
161 - | let current = self.h1_availability();
|
162 - | if self.published_h1_availability == Some(current) {
|
163 - | return H1AvailabilitySnapshot::new(self.h1_availability_revision, current);
|
172 + | /// Returns the current HTTP/1 revision for an exact crossing fallback.
|
173 + | fn current_h1_supply_revision(&mut self) -> SupplyRevision<H1SupplyStatus> {
|
174 + | let current = self.h1_supply_status();
|
175 + | if self.h1_supply_revision.revision != 0 && self.h1_supply_revision.status == current {
|
176 + | return self.h1_supply_revision;
|
164 177 | }
|
165 - | self.record_h1_availability(current)
|
178 + | self.advance_h1_supply_revision(current)
|
166 179 | }
|
167 180 |
|
168 - | /// Advances the revision and records changed HTTP/1 availability.
|
169 - | fn record_h1_availability(&mut self, availability: H1Availability) -> H1AvailabilitySnapshot {
|
170 - | self.h1_availability_revision = self
|
171 - | .h1_availability_revision
|
181 + | fn advance_h1_supply_revision(
|
182 + | &mut self,
|
183 + | status: H1SupplyStatus,
|
184 + | ) -> SupplyRevision<H1SupplyStatus> {
|
185 + | let revision = self
|
186 + | .h1_supply_revision
|
187 + | .revision
|
172 188 | .checked_add(1)
|
173 - | .expect("HTTP/1 availability revision exhausted");
|
174 - | self.published_h1_availability = Some(availability);
|
175 - | H1AvailabilitySnapshot::new(self.h1_availability_revision, availability)
|
189 + | .expect("HTTP/1 supply revision exhausted");
|
190 + | self.h1_supply_revision = SupplyRevision::new(revision, status);
|
191 + | self.h1_supply_revision
|
176 192 | }
|
177 193 |
|
178 - | /// Returns an H2 advertisement only when admission's view must change.
|
179 - | fn take_h2_advertisement_update(&mut self) -> Option<H2AdvertisementSnapshot> {
|
180 - | let current = self.h2_advertisement();
|
181 - | if self.published_h2_advertisement == current {
|
194 + | /// Returns an HTTP/2 revision only when admission's view must change.
|
195 + | ///
|
196 + | /// Initial unavailability is suppressed because absence already represents
|
197 + | /// that state in admission.
|
198 + | fn take_h2_supply_update(&mut self) -> Option<SupplyRevision<H2SupplyStatus>> {
|
199 + | let current = self.h2_supply_status();
|
200 + | if self.h2_supply_revision.status == current {
|
182 201 | return None;
|
183 202 | }
|
184 - | Some(self.record_h2_advertisement(current))
|
203 + | Some(self.advance_h2_supply_revision(current))
|
185 204 | }
|
186 205 |
|
187 - | /// Returns the current versioned H2 advertisement for a reclaim transition.
|
188 - | fn report_h2_advertisement(&mut self) -> H2AdvertisementSnapshot {
|
189 - | let current = self.h2_advertisement();
|
190 - | if self.published_h2_advertisement == current {
|
191 - | return Self::h2_advertisement_snapshot(self.h2_advertisement_revision, current);
|
206 + | /// Returns the current HTTP/2 revision for an exact reclaim fallback.
|
207 + | fn current_h2_supply_revision(&mut self) -> SupplyRevision<H2SupplyStatus> {
|
208 + | let current = self.h2_supply_status();
|
209 + | if self.h2_supply_revision.status == current {
|
210 + | return self.h2_supply_revision;
|
192 211 | }
|
193 - | self.record_h2_advertisement(current)
|
212 + | self.advance_h2_supply_revision(current)
|
194 213 | }
|
195 214 |
|
196 - | /// Returns the exact publishable generation and its idle state.
|
197 - | fn h2_advertisement(&self) -> Option<(h2::H2GenerationId, bool)> {
|
198 - | self.h2
|
199 - | .publishable_generation()
|
200 - | .map(|generation| (generation, self.h2.is_idle(generation)))
|
215 + | /// Derives the exact generation admission may route or reclaim.
|
216 + | fn h2_supply_status(&self) -> H2SupplyStatus {
|
217 + | match self.h2.peer_routable_generation() {
|
218 + | Some(generation) => H2SupplyStatus::Accepting {
|
219 + | generation,
|
220 + | idle: self.h2.is_idle(generation),
|
221 + | },
|
222 + | None => H2SupplyStatus::Unavailable,
|
223 + | }
|
201 224 | }
|
202 225 |
|
203 - | /// Advances the revision and records changed H2 availability.
|
204 - | fn record_h2_advertisement(
|
226 + | fn advance_h2_supply_revision(
|
205 227 | &mut self,
|
206 - | advertisement: Option<(h2::H2GenerationId, bool)>,
|
207 - | ) -> H2AdvertisementSnapshot {
|
208 - | self.h2_advertisement_revision = self
|
209 - | .h2_advertisement_revision
|
228 + | status: H2SupplyStatus,
|
229 + | ) -> SupplyRevision<H2SupplyStatus> {
|
230 + | let revision = self
|
231 + | .h2_supply_revision
|
232 + | .revision
|
210 233 | .checked_add(1)
|
211 - | .expect("HTTP/2 advertisement revision exhausted");
|
212 - | self.published_h2_advertisement = advertisement;
|
213 - | Self::h2_advertisement_snapshot(self.h2_advertisement_revision, advertisement)
|
214 - | }
|
215 - |
|
216 - | fn h2_advertisement_snapshot(
|
217 - | revision: u64,
|
218 - | advertisement: Option<(h2::H2GenerationId, bool)>,
|
219 - | ) -> H2AdvertisementSnapshot {
|
220 - | match advertisement {
|
221 - | Some((generation, true)) => H2AdvertisementSnapshot::idle(revision, generation),
|
222 - | Some((generation, false)) => H2AdvertisementSnapshot::accepting(revision, generation),
|
223 - | None => H2AdvertisementSnapshot::unavailable(revision),
|
224 - | }
|
234 + | .expect("HTTP/2 supply revision exhausted");
|
235 + | self.h2_supply_revision = SupplyRevision::new(revision, status);
|
236 + | self.h2_supply_revision
|
225 237 | }
|
226 238 |
|
227 239 | /// Atomically decides whether a peer may reserve this cell's connection.
|
228 - | fn install_reuse(&mut self, reuse_id: ReuseId) -> ReuseInstall {
|
229 - | let local_h1_demand = self.waiters.can_accept_h1();
|
230 - | if self.reuse.blocks_peer_reuse(local_h1_demand)
|
231 - | || self.waiters.has_prior_h1_candidate()
|
232 - | || !self.reuse.is_available()
|
240 + | fn commit_h1_reservation(
|
241 + | &mut self,
|
242 + | match_id: H1MatchId,
|
243 + | ) -> H1ReservationDecision<OwnedH1Sender> {
|
244 + | let local_h1_demand = self.acquisitions.has_h1_compatible_waiter();
|
245 + | if self.h1.blocks_peer_selection(local_h1_demand)
|
246 + | || self.acquisitions.has_prior_h1_waiter()
|
247 + | || !self.h1.peer_reservation_available()
|
233 248 | {
|
234 - | let report = self.report_h1_availability();
|
249 + | let report = self.current_h1_supply_revision();
|
235 250 | self.assert_consistent();
|
236 - | return ReuseInstall::Rejected(report);
|
251 + | return H1ReservationDecision::Rejected(report);
|
237 252 | }
|
238 253 |
|
239 - | if let Some(owner) = self.h1.take_idle_for_reuse() {
|
254 + | if let Some(owner) = self.h1.take_idle_candidate() {
|
240 255 | assert!(
|
241 - | self.reuse.install_resolving(reuse_id),
|
242 - | "available HTTP/1 connection could not install its reuse reservation"
|
256 + | self.h1.reserve_resolving(match_id),
|
257 + | "available HTTP/1 connection could not reserve its peer match"
|
243 258 | );
|
244 259 | self.assert_consistent();
|
245 - | return ReuseInstall::Candidate(owner);
|
260 + | return H1ReservationDecision::Candidate(owner);
|
246 261 | }
|
247 262 | if self.h1.has_returnable() {
|
248 263 | assert!(
|
249 - | self.reuse.install(reuse_id),
|
250 - | "available HTTP/1 connection could not install its reuse reservation"
|
264 + | self.h1.reserve_waiting(match_id),
|
265 + | "available HTTP/1 connection could not reserve its peer match"
|
251 266 | );
|
252 267 | self.assert_consistent();
|
253 - | return ReuseInstall::Installed;
|
268 + | return H1ReservationDecision::Installed;
|
254 269 | }
|
255 270 |
|
256 - | let report = self.report_h1_availability();
|
271 + | let report = self.current_h1_supply_revision();
|
257 272 | self.assert_consistent();
|
258 - | ReuseInstall::Rejected(report)
|
273 + | H1ReservationDecision::Rejected(report)
|
259 274 | }
|
260 275 |
|
261 - | /// Clears a reuse reservation and returns the cell's complete availability.
|
262 - | fn cancel_reuse(&mut self, reuse_id: ReuseId) -> H1AvailabilitySnapshot {
|
263 - | self.reuse.reject(reuse_id);
|
264 - | let local_h1_demand = self.waiters.can_accept_h1();
|
265 - | self.reuse.clear_unused_turn(local_h1_demand);
|
276 + | /// Clears a peer reservation and returns the cell's complete H1 status.
|
277 + | fn cancel_h1_reservation(&mut self, match_id: H1MatchId) -> SupplyRevision<H1SupplyStatus> {
|
278 + | self.h1.release_peer_reservation(match_id);
|
279 + | let local_h1_demand = self.acquisitions.has_h1_compatible_waiter();
|
280 + | self.h1.clear_unused_turn(local_h1_demand);
|
266 281 | self.assert_consistent();
|
267 - | self.report_h1_availability()
|
282 + | self.current_h1_supply_revision()
|
268 283 | }
|
269 284 |
|
270 285 | /// Clears a reservation before its provisional sender follows ordinary return.
|
271 - | fn reject_reuse_candidate(&mut self, reuse_id: ReuseId) {
|
272 - | self.reuse.reject(reuse_id);
|
286 + | fn reject_h1_match(&mut self, match_id: H1MatchId) {
|
287 + | self.h1.release_peer_reservation(match_id);
|
273 288 | self.assert_consistent();
|
274 289 | }
|
275 290 |
|
276 - | /// Revalidates a reuse operation and sender residence as one cell transition.
|
277 - | fn commit_reuse(&mut self, reuse_id: ReuseId, owner: &OwnedH1) -> bool {
|
278 - | let committed = self.reuse.names(reuse_id) && self.h1.commit_return_to_waiter(owner);
|
291 + | /// Revalidates one retained match and sender residence as one transition.
|
292 + | fn commit_h1_match(&mut self, match_id: H1MatchId, owner: &OwnedH1Sender) -> bool {
|
293 + | let committed =
|
294 + | self.h1.names_peer_reservation(match_id) && self.h1.commit_return_to_waiter(owner);
|
279 295 | self.assert_consistent();
|
280 296 | committed
|
281 297 | }
|
282 298 |
|
283 299 | /// Completes or rejects a reservation after its external action resolves.
|
284 - | fn finish_reuse(&mut self, reuse_id: ReuseId, transferred: bool) -> H1AvailabilitySnapshot {
|
300 + | fn complete_h1_match(
|
301 + | &mut self,
|
302 + | match_id: H1MatchId,
|
303 + | transferred: bool,
|
304 + | ) -> SupplyRevision<H1SupplyStatus> {
|
285 305 | if transferred {
|
286 - | let local_h1_demand = self.waiters.can_accept_h1();
|
287 - | self.reuse.complete_transfer(reuse_id, local_h1_demand);
|
306 + | let local_h1_demand = self.acquisitions.has_h1_compatible_waiter();
|
307 + | self.h1.complete_peer_transfer(match_id, local_h1_demand);
|
288 308 | } else {
|
289 - | self.reuse.reject(reuse_id);
|
309 + | self.h1.release_peer_reservation(match_id);
|
290 310 | }
|
291 311 | self.assert_consistent();
|
292 - | self.report_h1_availability()
|
312 + | self.current_h1_supply_revision()
|
293 313 | }
|
294 314 |
|
295 315 | /// Removes active HTTP/2-compatible demand already served by visible H2 state.
|
296 316 | fn publishable_demand_updates(
|
297 317 | &self,
|
298 318 | updates: [Option<DemandSnapshot>; 2],
|
299 319 | ) -> [Option<DemandSnapshot>; 2] {
|
300 320 | let h2_visible = self.h2.has_visible_h2();
|
301 321 | updates.map(|snapshot| {
|
302 322 | snapshot
|
303 323 | .filter(|snapshot| !h2_visible || !snapshot.is_active() || !snapshot.accepts_h2())
|
304 324 | })
|
305 325 | }
|
306 326 | }
|
307 327 |
|
308 - | /// Cell-local result of attempting to install one peer reuse reservation.
|
309 - | enum ReuseInstall {
|
310 - | /// The reservation will intercept a future sender return.
|
328 + | /// Supplier-cell decision for one HTTP/1 reservation attempt.
|
329 + | pub(in crate::client::pool) enum H1ReservationDecision<C> {
|
330 + | /// A future reusable sender return will satisfy the retained match.
|
311 331 | Installed,
|
312 - | /// An idle sender was detached for immediate resolution.
|
313 - | Candidate(OwnedH1),
|
314 - | /// The cell could not reserve a connection for the operation.
|
315 - | Rejected(H1AvailabilitySnapshot),
|
332 + | /// An idle sender was extracted immediately.
|
333 + | Candidate(C),
|
334 + | /// The supplier cell could not reserve sender ownership for the match.
|
335 + | Rejected(SupplyRevision<H1SupplyStatus>),
|
336 + | }
|
337 + |
|
338 + | impl<C> H1ReservationDecision<C> {
|
339 + | /// Maps only the detached sender while preserving the supplier decision.
|
340 + | fn map_candidate<D>(self, map: impl FnOnce(C) -> D) -> H1ReservationDecision<D> {
|
341 + | match self {
|
342 + | Self::Installed => H1ReservationDecision::Installed,
|
343 + | Self::Candidate(candidate) => H1ReservationDecision::Candidate(map(candidate)),
|
344 + | Self::Rejected(revision) => H1ReservationDecision::Rejected(revision),
|
345 + | }
|
346 + | }
|
316 347 | }
|
317 348 |
|
318 349 | /// A terminal result that may satisfy an acquisition waiter.
|
319 350 | #[derive(Debug)]
|
320 - | pub(super) enum AcquisitionResult {
|
351 + | pub(super) enum AcquisitionOutcome {
|
321 352 | /// An exclusive HTTP/1 sender selected from an installed cell-owned record.
|
322 353 | H1(H1Selection),
|
323 354 | /// A prospective request lease against one HTTP/2 generation.
|
324 355 | H2(h2::H2Activation),
|
325 356 | /// Establishment failed before producing a dispatchable connection.
|
326 357 | Failed(ConnectorError),
|
327 358 | /// An HTTP/2 generation closed before serving a transferred attempt.
|
328 - | Reacquire,
|
359 + | RetryAcquisition,
|
329 360 | }
|
330 361 |
|
331 362 | /// One event observed while driving an acquisition attempt.
|
332 363 | #[derive(Debug)]
|
333 - | pub(super) enum AcquisitionEvent {
|
364 + | pub(super) enum AcquisitionStep {
|
334 365 | /// Capacity is available and one establishment attempt may start.
|
335 - | Establish(EstablishmentPermit),
|
366 + | StartEstablishment(EstablishmentPermit),
|
336 367 | /// A returned sender or establishment produced the terminal result.
|
337 - | Complete(AcquisitionResult),
|
368 + | Resolved(AcquisitionOutcome),
|
338 369 | }
|
339 370 |
|
340 371 | /// Optional bounded-origin capacity for one establishment attempt.
|
341 372 | ///
|
342 373 | /// Unbounded origins carry no lease. For bounded origins, dropping this value
|
343 374 | /// before connection installation returns the permit to admission.
|
344 375 | #[derive(Debug)]
|
345 376 | pub(super) struct EstablishmentPermit {
|
346 377 | /// Bounded-origin slot transferred into a connection on success.
|
347 378 | lease: Option<CapacityLease>,
|
461 492 | (state.h1.connection_ids(), state.h2.generation_ids())
|
462 493 | };
|
463 494 | for id in h1 {
|
464 495 | Self::close_h1(cell, id, reason);
|
465 496 | }
|
466 497 | for generation in h2 {
|
467 498 | Self::close_h2(cell, generation, reason);
|
468 499 | }
|
469 500 | }
|
470 501 |
|
471 - | /// Applies one materialized acquisition delivery after admission unlock.
|
502 + | /// Applies one resolved acquisition delivery after admission unlock.
|
472 503 | ///
|
473 504 | /// Connection-cell H1 revalidation has already completed, so reserving
|
474 505 | /// the requesting cell cannot be followed by another fallible connection-
|
475 506 | /// cell transition.
|
476 507 | ///
|
477 508 | /// # Panics
|
478 509 | ///
|
479 510 | /// Panics if a waiter reserved by this function disappears or enters a
|
480 - | /// state that cannot accept the committed acquisition event.
|
511 + | /// state that cannot accept the committed acquisition step.
|
481 512 | pub(in crate::client::pool) fn receive_delivery(
|
482 513 | cell: &Arc<Self>,
|
483 514 | delivery: DeliveryGuard,
|
484 515 | ) -> Option<AdmissionAction> {
|
485 516 | let reservation = {
|
486 517 | let mut state = cell.state.lock();
|
487 518 | state
|
488 - | .waiters
|
519 + | .acquisitions
|
489 520 | .reserve_delivery_waiter(delivery.demand(), &cell.eligibility_group)
|
490 521 | };
|
491 522 |
|
492 523 | let DeliveryReservation::Reserved { waiter, successor } = reservation else {
|
493 - | return delivery.reject(None);
|
524 + | return delivery.refuse(None);
|
494 525 | };
|
495 526 |
|
496 - | let (event, mut acknowledgement) = delivery.commit(successor);
|
497 - | let (installation, suppress_successor) = {
|
527 + | let (step, mut settlement) = delivery.into_step(successor);
|
528 + | let (commit, suppress_successor) = {
|
498 529 | let mut state = cell.state.lock();
|
499 - | let installation = match event {
|
500 - | AcquisitionEvent::Establish(permit) => {
|
501 - | state.waiters.install_capacity(waiter, permit)
|
530 + | let commit = match step {
|
531 + | AcquisitionStep::StartEstablishment(permit) => {
|
532 + | state.acquisitions.commit_capacity(waiter, permit)
|
502 533 | }
|
503 - | AcquisitionEvent::Complete(result) => {
|
504 - | state.waiters.install_borrowed_h1(waiter, result)
|
534 + | AcquisitionStep::Resolved(outcome) => {
|
535 + | state.acquisitions.commit_borrowed_h1(waiter, outcome)
|
505 536 | }
|
506 537 | };
|
507 - | (installation, state.h2.has_visible_h2())
|
538 + | (commit, state.h2.has_visible_h2())
|
508 539 | };
|
509 540 | if suppress_successor {
|
510 - | acknowledgement.suppress_h2_successor();
|
541 + | settlement.suppress_h2_successor();
|
511 542 | }
|
512 543 |
|
513 - | let accepted = installation.accepted;
|
514 - | let error = installation.error;
|
515 - | let waker = installation.waker;
|
516 - | let next = if accepted {
|
517 - | drop(installation.returned_events);
|
518 - | acknowledgement.accept()
|
519 - | } else {
|
520 - | acknowledgement.reject(installation.returned_events)
|
544 + | let (next, waker, error) = match commit {
|
545 + | CellCommitOutcome::Committed { waker } => (settlement.accept(), waker, None),
|
546 + | CellCommitOutcome::Refused { returned, waker } => {
|
547 + | (settlement.refuse(returned), waker, None)
|
548 + | }
|
549 + | CellCommitOutcome::Invalid {
|
550 + | returned,
|
551 + | waker,
|
552 + | error,
|
553 + | } => (settlement.refuse(returned), waker, Some(error)),
|
521 554 | };
|
522 555 | if let Some(waker) = waker {
|
523 556 | waker.wake();
|
524 557 | }
|
525 558 | if let Some(error) = error {
|
526 559 | match error {
|
527 - | ResultInstallError::MissingWaiter => {
|
560 + | CellCommitError::MissingWaiter => {
|
528 561 | panic!("reserved waiter disappeared before acquisition delivery")
|
529 562 | }
|
530 - | ResultInstallError::UnexpectedState => {
|
563 + | CellCommitError::UnexpectedState => {
|
531 564 | panic!("reserved waiter entered an invalid acquisition-delivery state")
|
532 565 | }
|
533 566 | }
|
534 567 | }
|
535 568 | next
|
536 569 | }
|
537 570 |
|
538 571 | /// Returns the number of retained acquisition waiters for boundary tests.
|
539 572 | #[cfg(test)]
|
540 573 | pub(super) fn retained_waiters_for_test(&self) -> usize {
|
541 - | self.state.lock().waiters.snapshot().retained
|
574 + | self.state.lock().acquisitions.probe().retained
|
542 575 | }
|
543 576 |
|
544 577 | /// Registers one acquisition waiter in cell-local arrival order.
|
545 578 | ///
|
546 579 | /// The waiter and its demand snapshot are committed under the cell lock.
|
547 580 | /// A bounded cell publishes the snapshot to origin admission only after
|
548 581 | /// that lock is released. An unbounded cell retains only local order.
|
549 582 | pub(super) fn register_waiter(cell: &Arc<Self>, requirement: ProtocolRequirement) -> WaiterId {
|
550 583 | let (waiter, snapshot, h2_visible) = {
|
551 584 | let mut state = cell.state.lock();
|
552 - | let (waiter, snapshot) = state.waiters.register_waiter(
|
585 + | let (waiter, snapshot) = state.acquisitions.register_waiter(
|
553 586 | requirement,
|
554 587 | &cell.eligibility_group,
|
555 588 | cell.admission.is_some(),
|
556 589 | );
|
557 590 | (
|
558 591 | waiter,
|
559 592 | snapshot,
|
560 593 | state.h2.has_visible_h2() && requirement.accepts_h2(),
|
561 594 | )
|
562 595 | };
|
563 596 |
|
564 597 | if let (Some(admission), Some(snapshot), false) = (&cell.admission, snapshot, h2_visible) {
|
565 - | OriginAdmission::publish_demand(admission, cell.id.partition(), snapshot);
|
598 + | OriginAdmission::submit_demand_snapshot(admission, cell.id.partition(), snapshot);
|
566 599 | }
|
567 - | Self::service_h2_waiters(cell);
|
568 - | Self::service_peer_h2_waiters(cell);
|
600 + | Self::offer_local_h2(cell);
|
601 + | Self::offer_peer_h2(cell);
|
569 602 | waiter
|
570 603 | }
|
571 604 |
|
572 605 | /// Cancels a waiter and re-examines local idle H1 state atomically.
|
573 606 | ///
|
574 607 | /// A newly compatible successor may be serviceable by an H1 that an older
|
575 608 | /// HTTP/2-only head could not use. The local sender is selected under the
|
576 609 | /// same cell lock as cancellation, so it never needs a same-cell peer
|
577 - | /// reuse operation. Demand publication, fallback, and waking remain outside the lock.
|
610 + | /// H1 match. Demand publication, fallback, and waking remain outside the lock.
|
578 611 | pub(super) fn cancel_waiter(cell: &Arc<Self>, waiter: WaiterId) -> bool {
|
579 612 | let transition = {
|
580 613 | let mut state = cell.state.lock();
|
581 614 | state.h2.cancel_flight_participant(waiter);
|
582 615 | state.h2.cancel_peer_activation(waiter);
|
583 616 | state.h2.cancel_pending_waiter(waiter);
|
584 617 | state
|
585 - | .waiters
|
618 + | .acquisitions
|
586 619 | .cancel_waiter(waiter, &cell.eligibility_group)
|
587 620 | .map(|mut cancelled| {
|
588 - | let mut local_install = if state.waiters.can_accept_h1() {
|
621 + | let mut local_install = if state.acquisitions.has_h1_compatible_waiter() {
|
589 622 | state.h1.select_idle().map(|owner| {
|
590 - | state.reuse.consume_local_turn();
|
591 - | state.waiters.install_returned_h1(
|
592 - | || AcquisitionResult::H1(H1Selection::new(cell, owner)),
|
623 + | state.h1.consume_local_turn();
|
624 + | let (waiter, resolution) = state.acquisitions.offer_returned_h1(
|
625 + | || AcquisitionOutcome::H1(H1Selection::new(cell, owner)),
|
593 626 | &cell.eligibility_group,
|
627 + | );
|
628 + | (
|
629 + | waiter.expect(
|
630 + | "compatible HTTP/1 waiter disappeared during local offer",
|
631 + | ),
|
632 + | resolution,
|
594 633 | )
|
595 634 | })
|
596 635 | } else {
|
597 636 | None
|
598 637 | };
|
599 638 | cancelled.demand_updates =
|
600 639 | state.publishable_demand_updates(cancelled.demand_updates);
|
601 - | if let Some(install) = &mut local_install {
|
602 - | if let Some(waiter) = install.waiter {
|
603 - | state.h2.cancel_pending_waiter(waiter);
|
604 - | }
|
640 + | if let Some((waiter, install)) = &mut local_install {
|
641 + | state.h2.cancel_pending_waiter(*waiter);
|
605 642 | install.demand_updates = state.publishable_demand_updates(std::mem::take(
|
606 643 | &mut install.demand_updates,
|
607 644 | ));
|
608 645 | }
|
609 646 | (cancelled, local_install)
|
610 647 | })
|
611 648 | };
|
612 649 | let Some((cancelled, local_install)) = transition else {
|
613 650 | return false;
|
614 651 | };
|
615 652 |
|
616 653 | if let Some(admission) = &cell.admission {
|
617 654 | for snapshot in cancelled
|
618 655 | .demand_updates
|
619 656 | .into_iter()
|
620 657 | .chain(
|
621 658 | local_install
|
622 659 | .as_ref()
|
623 660 | .into_iter()
|
624 - | .flat_map(|install| install.demand_updates.iter().cloned()),
|
661 + | .flat_map(|(_, install)| install.demand_updates.iter().cloned()),
|
625 662 | )
|
626 663 | .flatten()
|
627 664 | {
|
628 - | OriginAdmission::publish_demand(admission, cell.id.partition(), snapshot);
|
665 + | OriginAdmission::submit_demand_snapshot(admission, cell.id.partition(), snapshot);
|
629 666 | }
|
630 667 | }
|
631 668 |
|
632 669 | // Ready results and any locally rejected event cross the lock boundary
|
633 670 | // before their fallback can re-enter the pool.
|
634 - | drop(cancelled.returned_events);
|
635 - | if let Some(install) = local_install {
|
636 - | drop(install.returned_event);
|
671 + | drop(cancelled.returned_steps);
|
672 + | if let Some((_, install)) = local_install {
|
673 + | drop(install.returned_step);
|
637 674 | if let Some(waker) = install.waker {
|
638 675 | waker.wake();
|
639 676 | }
|
640 677 | }
|
641 678 |
|
642 - | let availability = {
|
679 + | let supply_update = {
|
643 680 | let mut state = cell.state.lock();
|
644 - | let local_h1_demand = state.waiters.can_accept_h1();
|
645 - | state.reuse.clear_unused_turn(local_h1_demand);
|
681 + | let local_h1_demand = state.acquisitions.has_h1_compatible_waiter();
|
682 + | state.h1.clear_unused_turn(local_h1_demand);
|
646 683 | state.assert_consistent();
|
647 - | state.take_h1_availability_update()
|
684 + | state.take_h1_supply_update()
|
648 685 | };
|
649 - | cell.publish_h1_availability(availability);
|
650 - | Self::service_h2_waiters(cell);
|
651 - | Self::service_peer_h2_waiters(cell);
|
686 + | cell.submit_h1_supply_update(supply_update);
|
687 + | Self::offer_local_h2(cell);
|
688 + | Self::offer_peer_h2(cell);
|
652 689 | true
|
653 690 | }
|
654 691 |
|
655 692 | /// Polls the next event for one acquisition waiter.
|
656 693 | ///
|
694 + | /// The cell lock is held only for this synchronous state-machine poll. It
|
695 + | /// is released before `Poll::Pending` returns to the async caller and is
|
696 + | /// never held while polling Hyper, connector I/O, or user code.
|
697 + | ///
|
657 698 | /// # Panics
|
658 699 | ///
|
659 700 | /// Panics if `waiter` is unknown, was cancelled, or is polled again after
|
660 701 | /// its ready result was consumed.
|
661 702 | pub(super) fn poll_waiter(
|
662 703 | &self,
|
663 704 | waiter: WaiterId,
|
664 705 | cx: &mut Context<'_>,
|
665 - | ) -> Poll<AcquisitionEvent> {
|
706 + | ) -> Poll<AcquisitionStep> {
|
666 707 | let mut state = self.state.lock();
|
667 - | state.waiters.poll_waiter(waiter, cx)
|
708 + | state.acquisitions.poll_waiter(waiter, cx)
|
668 709 | }
|
669 710 |
|
670 711 | /// Marks an establishment attempt as started before its first connector poll.
|
671 712 | pub(super) fn start_establishment(&self, waiter: WaiterId) -> bool {
|
672 - | self.state.lock().waiters.start_establishment(waiter)
|
713 + | self.state.lock().acquisitions.start_establishment(waiter)
|
673 714 | }
|
674 715 |
|
675 716 | /// Commits one terminal establishment result to its launching waiter.
|
676 717 | ///
|
677 718 | /// A returned H1 may already have served or cancelled the waiter. In that
|
678 719 | /// case the losing result leaves the cell lock and follows its ordinary
|
679 720 | /// sender-return or error-drop fallback.
|
680 721 | ///
|
681 722 | /// # Panics
|
682 723 | ///
|
683 724 | /// Panics if the launching waiter disappeared or left the launching state
|
684 725 | /// before its owned establishment completed.
|
685 - | pub(super) fn complete_establishment(&self, waiter: WaiterId, result: AcquisitionResult) {
|
686 - | let served_with_h1 = matches!(&result, AcquisitionResult::H1(_));
|
687 - | let installation = {
|
726 + | pub(super) fn complete_establishment(&self, waiter: WaiterId, result: AcquisitionOutcome) {
|
727 + | let served_with_h1 = matches!(&result, AcquisitionOutcome::H1(_));
|
728 + | let commit = {
|
688 729 | let mut state = self.state.lock();
|
689 - | let installation = state.waiters.install_establishment_result(waiter, result);
|
690 - | let served_with_h1 = served_with_h1
|
691 - | && !installation.returned_events.iter().any(|event| {
|
692 - | matches!(
|
693 - | event,
|
694 - | Some(AcquisitionEvent::Complete(AcquisitionResult::H1(_)))
|
695 - | )
|
696 - | });
|
697 - | if installation.accepted {
|
730 + | let commit = state.acquisitions.commit_establishment(waiter, result);
|
731 + | let accepted = matches!(commit, CellCommitOutcome::Committed { .. });
|
732 + | if accepted {
|
698 733 | if served_with_h1 {
|
699 - | state.reuse.consume_local_turn();
|
734 + | state.h1.consume_local_turn();
|
700 735 | } else {
|
701 - | let local_h1_demand = state.waiters.can_accept_h1();
|
702 - | state.reuse.clear_unused_turn(local_h1_demand);
|
736 + | let local_h1_demand = state.acquisitions.has_h1_compatible_waiter();
|
737 + | state.h1.clear_unused_turn(local_h1_demand);
|
703 738 | }
|
704 739 | state.assert_consistent();
|
705 740 | }
|
706 - | installation
|
741 + | commit
|
707 742 | };
|
708 - | drop(installation.returned_events);
|
709 - | if let Some(waker) = installation.waker {
|
743 + | let (returned, waker, error) = match commit {
|
744 + | CellCommitOutcome::Committed { waker } => ([None, None], waker, None),
|
745 + | CellCommitOutcome::Refused { returned, waker } => (returned, waker, None),
|
746 + | CellCommitOutcome::Invalid {
|
747 + | returned,
|
748 + | waker,
|
749 + | error,
|
750 + | } => (returned, waker, Some(error)),
|
751 + | };
|
752 + | drop(returned);
|
753 + | if let Some(waker) = waker {
|
710 754 | waker.wake();
|
711 755 | }
|
712 - | if let Some(error) = installation.error {
|
756 + | if let Some(error) = error {
|
713 757 | match error {
|
714 - | ResultInstallError::MissingWaiter => {
|
758 + | CellCommitError::MissingWaiter => {
|
715 759 | unreachable!("establishment result reported a missing waiter")
|
716 760 | }
|
717 - | ResultInstallError::UnexpectedState => {
|
761 + | CellCommitError::UnexpectedState => {
|
718 762 | panic!("establishment completed for a waiter that was not launching")
|
719 763 | }
|
720 764 | }
|
721 765 | }
|
722 - | let availability = {
|
766 + | let supply_update = {
|
723 767 | let mut state = self.state.lock();
|
724 768 | state.assert_consistent();
|
725 - | state.take_h1_availability_update()
|
769 + | state.take_h1_supply_update()
|
726 770 | };
|
727 - | self.publish_h1_availability(availability);
|
771 + | self.submit_h1_supply_update(supply_update);
|
728 772 | }
|
729 773 |
|
730 774 | #[cfg(test)]
|
731 775 | pub(super) fn take_ready_lease(cell: &Arc<Self>, waiter: WaiterId) -> Option<CapacityLease> {
|
732 776 | let permit = match cell.take_ready_event(waiter)? {
|
733 - | AcquisitionEvent::Establish(permit) => permit,
|
734 - | AcquisitionEvent::Complete(_) => {
|
735 - | panic!("capacity test received a terminal acquisition result")
|
777 + | AcquisitionStep::StartEstablishment(permit) => permit,
|
778 + | AcquisitionStep::Resolved(_) => {
|
779 + | panic!("capacity test received a terminal acquisition outcome")
|
736 780 | }
|
737 781 | };
|
738 782 | assert!(Self::cancel_waiter(cell, waiter));
|
739 783 | permit.into_lease()
|
740 784 | }
|
741 785 |
|
742 786 | #[cfg(test)]
|
743 787 | fn take_ready_h1(&self, waiter: WaiterId) -> Option<H1Selection> {
|
744 788 | match self.take_ready_event(waiter)? {
|
745 - | AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => Some(selection),
|
746 - | AcquisitionEvent::Complete(AcquisitionResult::H2(_)) => {
|
789 + | AcquisitionStep::Resolved(AcquisitionOutcome::H1(selection)) => Some(selection),
|
790 + | AcquisitionStep::Resolved(AcquisitionOutcome::H2(_)) => {
|
747 791 | panic!("HTTP/1 ownership test received an HTTP/2 activation")
|
748 792 | }
|
749 - | AcquisitionEvent::Complete(AcquisitionResult::Failed(_)) => {
|
793 + | AcquisitionStep::Resolved(AcquisitionOutcome::Failed(_)) => {
|
750 794 | panic!("HTTP/1 ownership test received establishment failure")
|
751 795 | }
|
752 - | AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
|
796 + | AcquisitionStep::Resolved(AcquisitionOutcome::RetryAcquisition) => {
|
753 797 | panic!("HTTP/1 ownership test received an internal reacquisition")
|
754 798 | }
|
755 - | AcquisitionEvent::Establish(_) => {
|
799 + | AcquisitionStep::StartEstablishment(_) => {
|
756 800 | panic!("HTTP/1 ownership test received establishment capacity")
|
757 801 | }
|
758 802 | }
|
759 803 | }
|
760 804 |
|
761 805 | #[cfg(test)]
|
762 - | fn take_ready_event(&self, waiter: WaiterId) -> Option<AcquisitionEvent> {
|
806 + | fn take_ready_event(&self, waiter: WaiterId) -> Option<AcquisitionStep> {
|
763 807 | let mut state = self.state.lock();
|
764 808 | match state
|
765 - | .waiters
|
809 + | .acquisitions
|
766 810 | .poll_waiter(waiter, &mut Context::from_waker(std::task::Waker::noop()))
|
767 811 | {
|
768 812 | Poll::Ready(event) => Some(event),
|
769 813 | Poll::Pending => None,
|
770 814 | }
|
771 815 | }
|
772 816 |
|
773 817 | #[cfg(test)]
|
774 818 | pub(super) fn register_waiter_without_publish(
|
775 819 | &self,
|
776 820 | requirement: ProtocolRequirement,
|
777 821 | ) -> (WaiterId, DemandSnapshot) {
|
778 - | let (waiter, snapshot) =
|
779 - | self.state
|
780 - | .lock()
|
781 - | .waiters
|
782 - | .register_waiter(requirement, &self.eligibility_group, true);
|
822 + | let (waiter, snapshot) = self.state.lock().acquisitions.register_waiter(
|
823 + | requirement,
|
824 + | &self.eligibility_group,
|
825 + | true,
|
826 + | );
|
783 827 | (
|
784 828 | waiter,
|
785 829 | snapshot.expect("first unpublished waiter did not create demand"),
|
786 830 | )
|
787 831 | }
|
788 832 |
|
789 833 | #[cfg(test)]
|
790 - | fn snapshot(&self) -> CellSnapshot {
|
791 - | self.state.lock().waiters.snapshot()
|
834 + | fn probe(&self) -> AcquisitionProbe {
|
835 + | self.state.lock().acquisitions.probe()
|
792 836 | }
|
793 837 | }
|
794 838 |
|
795 839 | impl std::fmt::Debug for OriginCell {
|
796 840 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
797 841 | f.debug_struct("OriginCell")
|
798 842 | .field("id", &self.id)
|
799 843 | .field("eligibility_group", &self.eligibility_group)
|
800 844 | .field("admission", &self.admission)
|
801 845 | .field("state", &self.state)
|
894 938 | fn unbounded_connection(
|
895 939 | id: u64,
|
896 940 | ) -> (
|
897 941 | Arc<ConnectionState>,
|
898 942 | super::super::connection::PhysicalConnectionGuard,
|
899 943 | ) {
|
900 944 | ConnectionState::unbounded(connection_info(id))
|
901 945 | }
|
902 946 |
|
903 947 | #[test]
|
904 - | fn h1_availability_revision_advances_only_when_availability_changes() {
|
948 + | fn h1_supply_revision_advances_only_when_status_changes() {
|
905 949 | let mut state = CellState::default();
|
906 - | assert!(state.take_h1_availability_update().is_some());
|
907 - | assert_eq!(1, state.h1_availability_revision);
|
908 - | assert!(state.take_h1_availability_update().is_none());
|
909 - | assert_eq!(1, state.h1_availability_revision);
|
950 + | assert!(state.take_h1_supply_update().is_some());
|
951 + | assert_eq!(1, state.h1_supply_revision.revision);
|
952 + | assert!(state.take_h1_supply_update().is_none());
|
953 + | assert_eq!(1, state.h1_supply_revision.revision);
|
910 954 |
|
911 955 | let (connection, _physical) = unbounded_connection(1);
|
912 956 | state
|
913 957 | .h1
|
914 - | .install_idle(connection, H1Sender::test(11), None)
|
958 + | .insert_idle(connection, H1Sender::test(11), None)
|
915 959 | .unwrap();
|
916 - | assert!(state.take_h1_availability_update().is_some());
|
917 - | assert_eq!(2, state.h1_availability_revision);
|
960 + | assert!(state.take_h1_supply_update().is_some());
|
961 + | assert_eq!(2, state.h1_supply_revision.revision);
|
918 962 | }
|
919 963 |
|
920 964 | #[test]
|
921 - | fn active_reuse_reservation_blocks_peer_availability() {
|
965 + | fn active_peer_reservation_blocks_h1_supply() {
|
922 966 | let mut state = CellState::default();
|
923 967 | let (connection, _physical) = unbounded_connection(1);
|
924 968 | let owner = state
|
925 969 | .h1
|
926 - | .install_selected(connection, H1Sender::test(11))
|
970 + | .insert_selected(connection, H1Sender::test(11))
|
927 971 | .expect("fresh HTTP/1 record was rejected");
|
928 - | let reuse_id = ReuseId::for_test(1);
|
929 - | assert!(state.reuse.install(reuse_id));
|
972 + | let match_id = H1MatchId::for_test(1);
|
973 + | assert!(state.h1.reserve_waiting(match_id));
|
930 974 |
|
931 - | assert!(state.h1_availability().blocked);
|
975 + | assert!(state.h1_supply_status().peer_use_blocked);
|
932 976 |
|
933 - | assert!(state.reuse.reject(reuse_id));
|
977 + | assert!(state.h1.release_peer_reservation(match_id));
|
934 978 | assert!(state.h1.close_owned(&owner));
|
935 979 | drop(owner);
|
936 980 | }
|
937 981 |
|
938 982 | #[test]
|
939 983 | fn idle_h1_selection_returns_to_its_owning_cell() {
|
940 984 | let cell = unbounded_cell();
|
941 985 | let (connection, _physical) = unbounded_connection(1);
|
942 - | OriginCell::install_idle_h1(&cell, connection, H1Sender::test(11));
|
986 + | OriginCell::insert_idle_h1(&cell, connection, H1Sender::test(11));
|
943 987 |
|
944 988 | let selection = OriginCell::select_h1(&cell).expect("idle sender was not selected");
|
945 989 | assert_eq!(ConnectionId::new(1), selection.connection_id());
|
946 990 | assert_eq!(11, selection.test_sender_id());
|
947 991 | assert!(selection.is_reused());
|
948 992 | assert_eq!((1, 0), cell.h1_counts());
|
949 993 |
|
950 994 | drop(selection);
|
951 995 | assert_eq!((1, 1), cell.h1_counts());
|
952 996 | assert_eq!(
|
953 997 | 11,
|
954 998 | OriginCell::select_h1(&cell)
|
955 999 | .expect("returned sender was not reusable")
|
956 1000 | .test_sender_id()
|
957 1001 | );
|
958 1002 | }
|
959 1003 |
|
960 1004 | #[test]
|
961 1005 | fn selected_h1_closes_when_its_owning_cell_is_gone() {
|
962 1006 | let cell = unbounded_cell();
|
963 1007 | let (connection, _physical) = unbounded_connection(1);
|
964 1008 | let selection =
|
965 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1009 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
966 1010 |
|
967 1011 | drop(cell);
|
968 1012 | drop(selection);
|
969 1013 |
|
970 1014 | assert_eq!(
|
971 1015 | Some(CloseReason::PoolDropped),
|
972 - | connection.snapshot().close_reason
|
1016 + | connection.probe().close_reason
|
973 1017 | );
|
974 1018 | }
|
975 1019 |
|
976 1020 | #[test]
|
977 1021 | fn complete_h1_exchange_returns_sender_to_idle() {
|
978 1022 | let cell = unbounded_cell();
|
979 1023 | let (connection, _physical) = unbounded_connection(1);
|
980 - | let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1024 + | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
981 1025 | assert!(!selection.is_reused());
|
982 1026 |
|
983 1027 | selection.into_exchange().offer_for_reuse();
|
984 1028 | assert_eq!((1, 1), cell.h1_counts());
|
985 1029 | }
|
986 1030 |
|
987 1031 | #[test]
|
988 1032 | fn incomplete_h1_return_closes_and_releases_capacity() {
|
989 1033 | let (admission, cell) = bounded_cell();
|
990 1034 | let lease = OriginAdmission::lease_for_test(&admission);
|
991 1035 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
992 1036 | let selection =
|
993 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1037 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
994 1038 |
|
995 1039 | drop(selection.into_exchange());
|
996 1040 |
|
997 1041 | assert_eq!((0, 0), cell.h1_counts());
|
998 1042 | assert_eq!(1, admission.available_capacity_for_test());
|
999 1043 | assert_eq!(
|
1000 1044 | Some(CloseReason::IncompleteH1Exchange),
|
1001 - | connection.snapshot().close_reason
|
1045 + | connection.probe().close_reason
|
1002 1046 | );
|
1003 1047 | }
|
1004 1048 |
|
1005 1049 | #[test]
|
1006 1050 | fn returned_h1_satisfies_an_unbounded_waiter() {
|
1007 1051 | let cell = unbounded_cell();
|
1008 1052 | let (connection, _physical) = unbounded_connection(1);
|
1009 - | let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1053 + | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1010 1054 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1011 1055 |
|
1012 1056 | drop(selection);
|
1013 1057 |
|
1014 1058 | let selected = cell
|
1015 1059 | .take_ready_h1(waiter)
|
1016 1060 | .expect("returned sender did not satisfy the waiter");
|
1017 1061 | assert_eq!(11, selected.test_sender_id());
|
1018 1062 | assert!(selected.is_reused());
|
1019 - | assert_eq!(0, cell.snapshot().retained);
|
1063 + | assert_eq!(0, cell.probe().retained);
|
1020 1064 | drop(selected);
|
1021 1065 | assert_eq!((1, 1), cell.h1_counts());
|
1022 1066 | }
|
1023 1067 |
|
1024 1068 | #[test]
|
1025 1069 | fn returned_h1_satisfies_bounded_demand_without_releasing_capacity() {
|
1026 1070 | let (admission, cell) = bounded_cell();
|
1027 1071 | let lease = OriginAdmission::lease_for_test(&admission);
|
1028 1072 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1029 - | let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1073 + | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1030 1074 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1031 1075 | assert_eq!(1, admission.ordered_demand_count_for_test());
|
1032 1076 |
|
1033 1077 | drop(selection);
|
1034 1078 |
|
1035 1079 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
1036 1080 | assert_eq!(0, admission.available_capacity_for_test());
|
1037 1081 | let selected = cell
|
1038 1082 | .take_ready_h1(waiter)
|
1039 1083 | .expect("returned sender did not satisfy bounded demand");
|
1040 1084 | drop(selected);
|
1041 1085 | assert_eq!((1, 1), cell.h1_counts());
|
1042 1086 | }
|
1043 1087 |
|
1044 1088 | #[test]
|
1045 1089 | fn eligible_peer_borrows_an_idle_h1_without_moving_capacity() {
|
1046 1090 | let (admission, connection_cell, requesting_cell) =
|
1047 1091 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
1048 1092 | let lease = OriginAdmission::lease_for_test(&admission);
|
1049 1093 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1050 - | OriginCell::install_idle_h1(&connection_cell, connection, H1Sender::test(11));
|
1094 + | OriginCell::insert_idle_h1(&connection_cell, connection, H1Sender::test(11));
|
1051 1095 |
|
1052 1096 | let waiter =
|
1053 1097 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
1054 1098 | let borrowed = requesting_cell
|
1055 1099 | .take_ready_h1(waiter)
|
1056 1100 | .expect("eligible peer did not receive the idle HTTP/1 sender");
|
1057 1101 |
|
1058 1102 | assert_eq!(11, borrowed.test_sender_id());
|
1059 1103 | assert_eq!(0, admission.available_capacity_for_test());
|
1060 1104 | assert_eq!((1, 0), connection_cell.h1_counts());
|
1061 1105 | drop(borrowed);
|
1062 1106 | assert_eq!((1, 1), connection_cell.h1_counts());
|
1063 1107 | }
|
1064 1108 |
|
1065 1109 | #[test]
|
1066 1110 | fn ineligible_peer_reclaims_h1_capacity_without_moving_the_sender() {
|
1067 1111 | let (admission, connection_cell, requesting_cell) = bounded_peer_cells(
|
1068 1112 | 1,
|
1069 1113 | EligibilityGroup::Partition(PartitionId::from_index(1)),
|
1070 1114 | EligibilityGroup::Partition(PartitionId::from_index(2)),
|
1071 1115 | );
|
1072 1116 | let lease = OriginAdmission::lease_for_test(&admission);
|
1073 1117 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1074 - | OriginCell::install_idle_h1(&connection_cell, connection.clone(), H1Sender::test(11));
|
1118 + | OriginCell::insert_idle_h1(&connection_cell, connection.clone(), H1Sender::test(11));
|
1075 1119 |
|
1076 1120 | let waiter =
|
1077 1121 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
1078 1122 | let replacement = OriginCell::take_ready_lease(&requesting_cell, waiter)
|
1079 1123 | .expect("ineligible peer did not receive reclaimed capacity");
|
1080 1124 |
|
1081 1125 | assert_eq!(
|
1082 1126 | Some(CloseReason::Reclaimed),
|
1083 - | connection.snapshot().close_reason
|
1127 + | connection.probe().close_reason
|
1084 1128 | );
|
1085 1129 | assert_eq!((0, 0), connection_cell.h1_counts());
|
1086 1130 | assert_eq!(0, admission.available_capacity_for_test());
|
1087 1131 | drop(replacement);
|
1088 1132 | assert_eq!(1, admission.available_capacity_for_test());
|
1089 1133 | }
|
1090 1134 |
|
1091 1135 | #[test]
|
1092 - | fn installed_peer_reuse_intercepts_the_next_active_h1_return() {
|
1136 + | fn installed_peer_reservation_intercepts_the_next_active_h1_return() {
|
1093 1137 | let (admission, connection_cell, requesting_cell) =
|
1094 1138 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
1095 1139 | let lease = OriginAdmission::lease_for_test(&admission);
|
1096 1140 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1097 1141 | let selected =
|
1098 - | OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
1142 + | OriginCell::insert_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
1099 1143 | let waiter =
|
1100 1144 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
1101 1145 |
|
1102 1146 | drop(selected);
|
1103 1147 |
|
1104 1148 | let borrowed = requesting_cell
|
1105 1149 | .take_ready_h1(waiter)
|
1106 - | .expect("installed reuse operation did not intercept the active return");
|
1150 + | .expect("installed peer reservation did not intercept the active return");
|
1107 1151 | assert_eq!(11, borrowed.test_sender_id());
|
1108 1152 | drop(borrowed);
|
1109 1153 | assert_eq!((1, 1), connection_cell.h1_counts());
|
1110 1154 | }
|
1111 1155 |
|
1112 1156 | #[test]
|
1113 1157 | fn cancelling_requesting_cell_releases_connection_reservation() {
|
1114 1158 | let (admission, connection_cell, requesting_cell) =
|
1115 1159 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
1116 1160 | let lease = OriginAdmission::lease_for_test(&admission);
|
1117 1161 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1118 1162 | let selected =
|
1119 - | OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
1163 + | OriginCell::insert_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
1120 1164 | let waiter =
|
1121 1165 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
1122 1166 |
|
1123 1167 | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
1124 1168 | assert!(
|
1125 - | connection_cell.state.lock().reuse.is_available(),
|
1126 - | "requesting cell cancellation did not reconcile the installed connection-owning cell reuse operation"
|
1169 + | connection_cell
|
1170 + | .state
|
1171 + | .lock()
|
1172 + | .h1
|
1173 + | .peer_reservation_available(),
|
1174 + | "requesting cell cancellation did not reconcile the installed supplier-cell reservation"
|
1127 1175 | );
|
1128 1176 | drop(selected);
|
1129 1177 |
|
1130 1178 | assert_eq!((1, 1), connection_cell.h1_counts());
|
1131 1179 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
1132 1180 | }
|
1133 1181 |
|
1134 1182 | #[test]
|
1135 1183 | fn cross_cell_borrow_owes_one_usable_turn_to_the_owning_cell() {
|
1136 1184 | let (admission, connection_cell, requesting_cell) =
|
1137 1185 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
1138 1186 | let lease = OriginAdmission::lease_for_test(&admission);
|
1139 1187 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1140 1188 | let selected =
|
1141 - | OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
1189 + | OriginCell::insert_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
1142 1190 |
|
1143 1191 | let requesting_waiter =
|
1144 1192 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
1145 1193 | let local_waiter =
|
1146 1194 | OriginCell::register_waiter(&connection_cell, ProtocolRequirement::H1Compatible);
|
1147 1195 | drop(selected);
|
1148 1196 |
|
1149 1197 | let borrowed = requesting_cell
|
1150 1198 | .take_ready_h1(requesting_waiter)
|
1151 1199 | .expect("older peer demand did not receive the intercepted sender");
|
1152 - | assert!(connection_cell.state.lock().reuse.local_turn_owed());
|
1200 + | assert!(connection_cell.state.lock().h1.local_turn_owed());
|
1153 1201 | drop(borrowed);
|
1154 1202 |
|
1155 1203 | let local = connection_cell
|
1156 1204 | .take_ready_h1(local_waiter)
|
1157 1205 | .expect("connection-owning cell local demand did not receive its owed turn");
|
1158 - | assert!(!connection_cell.state.lock().reuse.local_turn_owed());
|
1206 + | assert!(!connection_cell.state.lock().h1.local_turn_owed());
|
1159 1207 | drop(local);
|
1160 1208 | assert_eq!((1, 1), connection_cell.h1_counts());
|
1161 1209 | }
|
1162 1210 |
|
1163 1211 | #[test]
|
1164 1212 | fn cancelling_ready_h1_returns_it_after_unlock() {
|
1165 1213 | let cell = unbounded_cell();
|
1166 1214 | let (connection, _physical) = unbounded_connection(1);
|
1167 - | let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1215 + | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1168 1216 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1169 1217 | drop(selection);
|
1170 1218 |
|
1171 1219 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1172 1220 | assert_eq!((1, 1), cell.h1_counts());
|
1173 1221 | assert_eq!(
|
1174 1222 | 11,
|
1175 1223 | OriginCell::select_h1(&cell)
|
1176 1224 | .expect("cancelled result did not return its sender")
|
1177 1225 | .test_sender_id()
|
1178 1226 | );
|
1179 1227 | }
|
1180 1228 |
|
1181 1229 | #[test]
|
1182 1230 | fn returned_h1_and_establishment_complete_one_acquisition_attempt() {
|
1183 1231 | let cell = unbounded_cell();
|
1184 1232 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1185 - | let AcquisitionEvent::Establish(permit) = cell
|
1233 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1186 1234 | .take_ready_event(waiter)
|
1187 1235 | .expect("unbounded miss did not start establishment")
|
1188 1236 | else {
|
1189 1237 | panic!("unbounded miss completed before establishment started");
|
1190 1238 | };
|
1191 1239 | assert!(cell.start_establishment(waiter));
|
1192 1240 | assert!(permit.into_lease().is_none());
|
1193 1241 |
|
1194 1242 | let (returning_connection, _returning_physical) = unbounded_connection(1);
|
1195 1243 | let returning =
|
1196 - | OriginCell::install_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
1244 + | OriginCell::insert_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
1197 1245 | let (fresh_connection, _fresh_physical) = unbounded_connection(2);
|
1198 - | let fresh = OriginCell::install_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
1246 + | let fresh = OriginCell::insert_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
1199 1247 |
|
1200 1248 | drop(returning);
|
1201 - | cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
|
1249 + | cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
1202 1250 |
|
1203 1251 | assert_eq!((2, 1), cell.h1_counts());
|
1204 1252 | let winner = cell
|
1205 1253 | .take_ready_h1(waiter)
|
1206 1254 | .expect("acquisition attempt had no terminal H1");
|
1207 1255 | assert_eq!(11, winner.test_sender_id());
|
1208 1256 | drop(winner);
|
1209 1257 | assert_eq!((2, 2), cell.h1_counts());
|
1210 1258 | }
|
1211 1259 |
|
1212 1260 | #[test]
|
1213 1261 | fn bounded_attempt_remains_eligible_for_a_returned_h1() {
|
1214 1262 | let (admission, cell) = bounded_cell_with_limit(2);
|
1215 1263 | let installed_lease = OriginAdmission::lease_for_test(&admission);
|
1216 1264 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), installed_lease);
|
1217 - | let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1265 + | let returning = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1218 1266 |
|
1219 1267 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1220 - | let AcquisitionEvent::Establish(permit) = cell
|
1268 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1221 1269 | .take_ready_event(waiter)
|
1222 1270 | .expect("bounded miss received no capacity")
|
1223 1271 | else {
|
1224 1272 | panic!("bounded miss completed before establishment started");
|
1225 1273 | };
|
1226 1274 | let attempt_lease = permit
|
1227 1275 | .into_lease()
|
1228 1276 | .expect("bounded establishment carried no capacity");
|
1229 1277 | assert_eq!(0, admission.available_capacity_for_test());
|
1230 1278 |
|
1231 1279 | drop(returning);
|
1232 1280 | let selected = cell
|
1233 1281 | .take_ready_h1(waiter)
|
1234 1282 | .expect("returned H1 did not beat the pending establishment");
|
1235 1283 | drop(attempt_lease);
|
1236 1284 | assert_eq!(1, admission.available_capacity_for_test());
|
1237 1285 | drop(selected);
|
1238 1286 | assert_eq!((1, 1), cell.h1_counts());
|
1239 1287 | }
|
1240 1288 |
|
1241 1289 | #[test]
|
1242 1290 | fn older_launching_waiter_precedes_younger_capacity_waiter_for_returned_h1() {
|
1243 1291 | let (admission, cell) = bounded_cell_with_limit(2);
|
1244 1292 | let installed_lease = OriginAdmission::lease_for_test(&admission);
|
1245 1293 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), installed_lease);
|
1246 - | let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1294 + | let returning = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1247 1295 |
|
1248 1296 | let older = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1249 - | let AcquisitionEvent::Establish(permit) = cell
|
1297 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1250 1298 | .take_ready_event(older)
|
1251 1299 | .expect("older waiter received no establishment capacity")
|
1252 1300 | else {
|
1253 1301 | panic!("older waiter completed before establishment started");
|
1254 1302 | };
|
1255 1303 | let attempt_lease = permit
|
1256 1304 | .into_lease()
|
1257 1305 | .expect("bounded establishment carried no capacity");
|
1258 1306 |
|
1259 1307 | let younger = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1260 1308 | assert!(cell.take_ready_event(younger).is_none());
|
1261 1309 |
|
1262 1310 | drop(returning);
|
1263 1311 |
|
1264 1312 | let selected = cell
|
1265 1313 | .take_ready_h1(older)
|
1266 1314 | .expect("younger capacity waiter overtook the older launching waiter");
|
1267 1315 | assert_eq!(11, selected.test_sender_id());
|
1268 1316 | assert!(cell.take_ready_event(younger).is_none());
|
1269 1317 |
|
1270 1318 | assert!(OriginCell::cancel_waiter(&cell, younger));
|
1271 1319 | drop(attempt_lease);
|
1272 1320 | drop(selected);
|
1273 1321 | assert_eq!((1, 1), cell.h1_counts());
|
1274 1322 | assert_eq!(1, admission.available_capacity_for_test());
|
1275 1323 | }
|
1276 1324 |
|
1277 1325 | #[test]
|
1278 - | fn establishment_failure_is_a_terminal_acquisition_result() {
|
1326 + | fn establishment_failure_is_a_terminal_acquisition_outcome() {
|
1279 1327 | let cell = unbounded_cell();
|
1280 1328 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1281 - | let AcquisitionEvent::Establish(permit) = cell
|
1329 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1282 1330 | .take_ready_event(waiter)
|
1283 1331 | .expect("unbounded miss did not start establishment")
|
1284 1332 | else {
|
1285 1333 | panic!("unbounded miss completed before establishment started");
|
1286 1334 | };
|
1287 1335 | assert!(cell.start_establishment(waiter));
|
1288 1336 | drop(permit);
|
1289 1337 |
|
1290 1338 | cell.complete_establishment(
|
1291 1339 | waiter,
|
1292 - | AcquisitionResult::Failed(ConnectorError::io(
|
1340 + | AcquisitionOutcome::Failed(ConnectorError::io(
|
1293 1341 | std::io::Error::other("synthetic establishment failure").into(),
|
1294 1342 | )),
|
1295 1343 | );
|
1296 1344 |
|
1297 - | let AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) = cell
|
1345 + | let AcquisitionStep::Resolved(AcquisitionOutcome::Failed(error)) = cell
|
1298 1346 | .take_ready_event(waiter)
|
1299 1347 | .expect("establishment failure was not delivered")
|
1300 1348 | else {
|
1301 - | panic!("establishment failure produced the wrong acquisition event");
|
1349 + | panic!("establishment failure produced the wrong acquisition step");
|
1302 1350 | };
|
1303 1351 | assert!(error.is_io());
|
1304 - | assert_eq!(0, cell.snapshot().retained);
|
1352 + | assert_eq!(0, cell.probe().retained);
|
1305 1353 | }
|
1306 1354 |
|
1307 1355 | #[test]
|
1308 1356 | fn establishment_completion_clears_the_served_waiters_local_turn() {
|
1309 1357 | let cell = unbounded_cell();
|
1310 - | let reuse_id = ReuseId::for_test(1);
|
1358 + | let match_id = H1MatchId::for_test(1);
|
1311 1359 | {
|
1312 1360 | let mut state = cell.state.lock();
|
1313 - | assert!(state.reuse.install_resolving(reuse_id));
|
1314 - | assert!(state.reuse.complete_transfer(reuse_id, true));
|
1361 + | assert!(state.h1.reserve_resolving(match_id));
|
1362 + | assert!(state.h1.complete_peer_transfer(match_id, true));
|
1315 1363 | }
|
1316 1364 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1317 - | let AcquisitionEvent::Establish(permit) = cell
|
1365 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1318 1366 | .take_ready_event(waiter)
|
1319 1367 | .expect("unbounded miss did not prepare establishment")
|
1320 1368 | else {
|
1321 1369 | panic!("unbounded miss completed before establishment was prepared");
|
1322 1370 | };
|
1323 1371 | assert!(cell.start_establishment(waiter));
|
1324 1372 | drop(permit);
|
1325 1373 | let (connection, _physical) = unbounded_connection(1);
|
1326 - | let fresh = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1374 + | let fresh = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1327 1375 |
|
1328 - | cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
|
1376 + | cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
1329 1377 |
|
1330 - | assert!(!cell.state.lock().reuse.local_turn_owed());
|
1378 + | assert!(!cell.state.lock().h1.local_turn_owed());
|
1331 1379 | drop(
|
1332 1380 | cell.take_ready_h1(waiter)
|
1333 1381 | .expect("successful establishment result was not retained"),
|
1334 1382 | );
|
1335 1383 | }
|
1336 1384 |
|
1337 1385 | #[test]
|
1338 1386 | fn failed_establishment_preserves_a_turn_for_compatible_successor() {
|
1339 1387 | let cell = unbounded_cell();
|
1340 - | let reuse_id = ReuseId::for_test(1);
|
1388 + | let match_id = H1MatchId::for_test(1);
|
1341 1389 | {
|
1342 1390 | let mut state = cell.state.lock();
|
1343 - | assert!(state.reuse.install_resolving(reuse_id));
|
1344 - | assert!(state.reuse.complete_transfer(reuse_id, true));
|
1391 + | assert!(state.h1.reserve_resolving(match_id));
|
1392 + | assert!(state.h1.complete_peer_transfer(match_id, true));
|
1345 1393 | }
|
1346 1394 | let failed = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1347 - | let AcquisitionEvent::Establish(permit) = cell
|
1395 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1348 1396 | .take_ready_event(failed)
|
1349 1397 | .expect("unbounded miss did not prepare establishment")
|
1350 1398 | else {
|
1351 1399 | panic!("unbounded miss completed before establishment was prepared");
|
1352 1400 | };
|
1353 1401 | assert!(cell.start_establishment(failed));
|
1354 1402 | drop(permit);
|
1355 1403 | let successor = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1356 1404 |
|
1357 1405 | cell.complete_establishment(
|
1358 1406 | failed,
|
1359 - | AcquisitionResult::Failed(ConnectorError::io(
|
1407 + | AcquisitionOutcome::Failed(ConnectorError::io(
|
1360 1408 | std::io::Error::other("synthetic establishment failure").into(),
|
1361 1409 | )),
|
1362 1410 | );
|
1363 1411 |
|
1364 - | assert!(cell.state.lock().reuse.local_turn_owed());
|
1412 + | assert!(cell.state.lock().h1.local_turn_owed());
|
1365 1413 | assert!(matches!(
|
1366 1414 | cell.take_ready_event(failed),
|
1367 - | Some(AcquisitionEvent::Complete(AcquisitionResult::Failed(_)))
|
1415 + | Some(AcquisitionStep::Resolved(AcquisitionOutcome::Failed(_)))
|
1368 1416 | ));
|
1369 1417 | assert!(OriginCell::cancel_waiter(&cell, successor));
|
1370 - | assert!(!cell.state.lock().reuse.local_turn_owed());
|
1418 + | assert!(!cell.state.lock().h1.local_turn_owed());
|
1371 1419 | }
|
1372 1420 |
|
1373 1421 | #[test]
|
1374 1422 | fn failed_establishment_clears_a_turn_after_local_demand_drains() {
|
1375 1423 | let cell = unbounded_cell();
|
1376 - | let reuse_id = ReuseId::for_test(1);
|
1424 + | let match_id = H1MatchId::for_test(1);
|
1377 1425 | {
|
1378 1426 | let mut state = cell.state.lock();
|
1379 - | assert!(state.reuse.install_resolving(reuse_id));
|
1380 - | assert!(state.reuse.complete_transfer(reuse_id, true));
|
1427 + | assert!(state.h1.reserve_resolving(match_id));
|
1428 + | assert!(state.h1.complete_peer_transfer(match_id, true));
|
1381 1429 | }
|
1382 1430 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1383 - | let AcquisitionEvent::Establish(permit) = cell
|
1431 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1384 1432 | .take_ready_event(waiter)
|
1385 1433 | .expect("unbounded miss did not prepare establishment")
|
1386 1434 | else {
|
1387 1435 | panic!("unbounded miss completed before establishment was prepared");
|
1388 1436 | };
|
1389 1437 | assert!(cell.start_establishment(waiter));
|
1390 1438 | drop(permit);
|
1391 1439 |
|
1392 1440 | cell.complete_establishment(
|
1393 1441 | waiter,
|
1394 - | AcquisitionResult::Failed(ConnectorError::io(
|
1442 + | AcquisitionOutcome::Failed(ConnectorError::io(
|
1395 1443 | std::io::Error::other("synthetic establishment failure").into(),
|
1396 1444 | )),
|
1397 1445 | );
|
1398 1446 |
|
1399 - | assert!(!cell.state.lock().reuse.local_turn_owed());
|
1447 + | assert!(!cell.state.lock().h1.local_turn_owed());
|
1400 1448 | assert!(matches!(
|
1401 1449 | cell.take_ready_event(waiter),
|
1402 - | Some(AcquisitionEvent::Complete(AcquisitionResult::Failed(_)))
|
1450 + | Some(AcquisitionStep::Resolved(AcquisitionOutcome::Failed(_)))
|
1403 1451 | ));
|
1404 1452 | }
|
1405 1453 |
|
1406 1454 | #[test]
|
1407 1455 | fn completion_after_waiter_cancellation_returns_the_new_h1() {
|
1408 1456 | let cell = unbounded_cell();
|
1409 1457 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1410 - | let AcquisitionEvent::Establish(permit) = cell
|
1458 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1411 1459 | .take_ready_event(waiter)
|
1412 1460 | .expect("unbounded miss did not start establishment")
|
1413 1461 | else {
|
1414 1462 | panic!("unbounded miss completed before establishment started");
|
1415 1463 | };
|
1416 1464 | assert!(cell.start_establishment(waiter));
|
1417 1465 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1418 1466 | drop(permit);
|
1419 1467 |
|
1420 1468 | let (connection, _physical) = unbounded_connection(1);
|
1421 - | let fresh = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1422 - | cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
|
1469 + | let fresh = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1470 + | cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
1423 1471 |
|
1424 1472 | assert_eq!((1, 1), cell.h1_counts());
|
1425 1473 | assert_eq!(
|
1426 1474 | 11,
|
1427 1475 | OriginCell::select_h1(&cell)
|
1428 1476 | .expect("cancelled acquisition lost its completed H1")
|
1429 1477 | .test_sender_id()
|
1430 1478 | );
|
1431 1479 | }
|
1432 1480 |
|
1433 1481 | #[test]
|
1434 1482 | fn returned_h1_prevents_the_first_establishment_poll() {
|
1435 1483 | let cell = unbounded_cell();
|
1436 1484 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1437 - | let AcquisitionEvent::Establish(permit) = cell
|
1485 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1438 1486 | .take_ready_event(waiter)
|
1439 1487 | .expect("unbounded miss did not prepare establishment")
|
1440 1488 | else {
|
1441 1489 | panic!("unbounded miss completed before establishment was prepared");
|
1442 1490 | };
|
1443 1491 | let (connection, _physical) = unbounded_connection(1);
|
1444 - | let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1492 + | let returning = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1445 1493 |
|
1446 1494 | drop(returning);
|
1447 1495 | assert!(!cell.start_establishment(waiter));
|
1448 1496 | drop(permit);
|
1449 1497 | let selected = cell
|
1450 1498 | .take_ready_h1(waiter)
|
1451 1499 | .expect("returned H1 did not win before establishment started");
|
1452 1500 | assert_eq!(11, selected.test_sender_id());
|
1453 1501 | }
|
1454 1502 |
|
1455 1503 | #[test]
|
1456 1504 | fn h2_required_head_does_not_take_h1() {
|
1457 1505 | let cell = unbounded_cell();
|
1458 1506 | let (connection, _physical) = unbounded_connection(1);
|
1459 - | let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1507 + | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1460 1508 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
1461 1509 |
|
1462 1510 | drop(selection);
|
1463 1511 |
|
1464 1512 | assert_eq!((1, 1), cell.h1_counts());
|
1465 - | let AcquisitionEvent::Establish(permit) = cell
|
1513 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1466 1514 | .take_ready_event(waiter)
|
1467 1515 | .expect("unbounded miss did not start establishment")
|
1468 1516 | else {
|
1469 1517 | panic!("H2-required waiter accepted an HTTP/1 selection");
|
1470 1518 | };
|
1471 1519 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1472 1520 | drop(permit);
|
1473 1521 | }
|
1474 1522 |
|
1475 1523 | #[test]
|
1476 1524 | fn cancelling_h2_head_refunnels_reclaimed_capacity_to_successor() {
|
1477 1525 | let (admission, cell) = bounded_cell();
|
1478 1526 | let lease = OriginAdmission::lease_for_test(&admission);
|
1479 1527 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1480 - | OriginCell::install_idle_h1(&cell, connection, H1Sender::test(11));
|
1528 + | OriginCell::insert_idle_h1(&cell, connection, H1Sender::test(11));
|
1481 1529 | let h2_head = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
1482 1530 | let h1_successor = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1483 1531 |
|
1484 - | let AcquisitionEvent::Establish(permit) = cell
|
1532 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
1485 1533 | .take_ready_event(h2_head)
|
1486 1534 | .expect("HTTP/2 demand did not receive reclaimed local capacity")
|
1487 1535 | else {
|
1488 1536 | panic!("HTTP/2 demand received a protocol handle instead of capacity");
|
1489 1537 | };
|
1490 1538 | assert!(OriginCell::cancel_waiter(&cell, h2_head));
|
1491 1539 | drop(permit);
|
1492 1540 |
|
1493 1541 | let successor = OriginCell::take_ready_lease(&cell, h1_successor)
|
1494 1542 | .expect("reclaimed capacity did not reach the compatible successor");
|
1495 1543 | assert_eq!((0, 0), cell.h1_counts());
|
1496 - | assert!(cell.state.lock().reuse.is_available());
|
1544 + | assert!(cell.state.lock().h1.peer_reservation_available());
|
1497 1545 | drop(successor);
|
1498 1546 | }
|
1499 1547 |
|
1500 1548 | #[test]
|
1501 1549 | fn close_marks_selected_record_until_sender_returns() {
|
1502 1550 | let cell = unbounded_cell();
|
1503 1551 | let (connection, _physical) = unbounded_connection(1);
|
1504 1552 | let selection =
|
1505 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1553 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1506 1554 | let close = H1CloseHandle::new(&cell, selection.connection());
|
1507 1555 |
|
1508 1556 | assert!(close.close(CloseReason::Poisoned));
|
1509 1557 | assert!(OriginCell::select_h1(&cell).is_none());
|
1510 1558 | assert_eq!((1, 0), cell.h1_counts());
|
1511 1559 |
|
1512 1560 | drop(selection);
|
1513 1561 | assert_eq!((0, 0), cell.h1_counts());
|
1514 - | assert_eq!(
|
1515 - | Some(CloseReason::Poisoned),
|
1516 - | connection.snapshot().close_reason
|
1517 - | );
|
1562 + | assert_eq!(Some(CloseReason::Poisoned), connection.probe().close_reason);
|
1518 1563 | }
|
1519 1564 |
|
1520 1565 | #[test]
|
1521 1566 | fn confirmed_upgrade_refines_an_earlier_driver_close() {
|
1522 1567 | let (admission, cell) = bounded_cell();
|
1523 1568 | let lease = OriginAdmission::lease_for_test(&admission);
|
1524 1569 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
1525 1570 | let selection =
|
1526 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1571 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1527 1572 | let driver = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));
|
1528 1573 | let return_task = selection.into_exchange();
|
1529 1574 |
|
1530 1575 | driver.protocol_closed();
|
1531 1576 | return_task.retire_connection(CloseReason::Upgraded);
|
1532 1577 |
|
1533 - | assert_eq!(
|
1534 - | Some(CloseReason::Upgraded),
|
1535 - | connection.snapshot().close_reason
|
1536 - | );
|
1578 + | assert_eq!(Some(CloseReason::Upgraded), connection.probe().close_reason);
|
1537 1579 | assert_eq!(1, admission.available_capacity_for_test());
|
1538 1580 | assert_eq!((0, 0), cell.h1_counts());
|
1539 1581 | }
|
1540 1582 |
|
1541 1583 | #[test]
|
1542 1584 | fn driver_drop_closes_its_installed_generation() {
|
1543 1585 | let cell = unbounded_cell();
|
1544 1586 | let (connection, _physical) = unbounded_connection(1);
|
1545 1587 | let selection =
|
1546 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1588 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1547 1589 | let driver = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));
|
1548 1590 |
|
1549 1591 | drop(driver);
|
1550 1592 | assert_eq!(
|
1551 1593 | Some(CloseReason::OwnerRuntimeShutdown),
|
1552 - | connection.snapshot().close_reason
|
1594 + | connection.probe().close_reason
|
1553 1595 | );
|
1554 1596 | drop(selection);
|
1555 1597 | assert_eq!((0, 0), cell.h1_counts());
|
1556 1598 | }
|
1557 1599 |
|
1558 1600 | #[test]
|
1559 1601 | fn owner_runtime_shutdown_drops_live_driver_guard() {
|
1560 1602 | let cell = unbounded_cell();
|
1561 1603 | let (connection, _physical) = unbounded_connection(1);
|
1562 1604 | let selection =
|
1563 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1605 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
1564 1606 | let driver = H1DriverGuard::new(H1CloseHandle::new(&cell, &connection));
|
1565 1607 | let runtime = tokio::runtime::Builder::new_multi_thread()
|
1566 1608 | .worker_threads(1)
|
1567 1609 | .enable_all()
|
1568 1610 | .build()
|
1569 1611 | .unwrap();
|
1570 1612 | let (started, observed) = std::sync::mpsc::sync_channel(0);
|
1571 1613 |
|
1572 1614 | runtime.spawn(async move {
|
1573 1615 | let _driver = driver;
|
1574 1616 | started.send(()).unwrap();
|
1575 1617 | std::future::pending::<()>().await;
|
1576 1618 | });
|
1577 1619 | observed
|
1578 1620 | .recv_timeout(std::time::Duration::from_secs(1))
|
1579 1621 | .expect("driver task did not start");
|
1580 1622 | runtime.shutdown_timeout(std::time::Duration::from_secs(1));
|
1581 1623 |
|
1582 1624 | assert_eq!(
|
1583 1625 | Some(CloseReason::OwnerRuntimeShutdown),
|
1584 - | connection.snapshot().close_reason
|
1626 + | connection.probe().close_reason
|
1585 1627 | );
|
1586 1628 | drop(selection);
|
1587 1629 | assert_eq!((0, 0), cell.h1_counts());
|
1588 1630 | }
|
1589 1631 |
|
1590 1632 | #[test]
|
1591 1633 | fn three_waiters_receive_capacity_in_fifo_order() {
|
1592 1634 | let (_admission, cell) = bounded_cell();
|
1593 1635 | let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1594 1636 | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1595 1637 | let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1596 1638 |
|
1597 1639 | let first_lease = OriginCell::take_ready_lease(&cell, first).unwrap();
|
1598 1640 | assert!(OriginCell::take_ready_lease(&cell, second).is_none());
|
1599 1641 | assert!(OriginCell::take_ready_lease(&cell, third).is_none());
|
1600 1642 |
|
1601 1643 | drop(first_lease);
|
1602 1644 | let second_lease = OriginCell::take_ready_lease(&cell, second).unwrap();
|
1603 1645 | assert!(OriginCell::take_ready_lease(&cell, third).is_none());
|
1604 1646 |
|
1605 1647 | drop(second_lease);
|
1606 1648 | let third_lease = OriginCell::take_ready_lease(&cell, third).unwrap();
|
1607 1649 | drop(third_lease);
|
1608 - | assert_eq!(0, cell.snapshot().retained);
|
1650 + | assert_eq!(0, cell.probe().retained);
|
1609 1651 | }
|
1610 1652 |
|
1611 1653 | #[test]
|
1612 1654 | fn cancelling_middle_and_tail_waiters_repairs_fifo_links() {
|
1613 1655 | let (_admission, cell, held) = saturated_bounded_cell();
|
1614 1656 | let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1615 1657 | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H2Required);
|
1616 1658 | let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1617 1659 | let fourth = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1618 - | let demand = cell.snapshot().demand;
|
1660 + | let demand = cell.probe().demand;
|
1619 1661 |
|
1620 1662 | assert!(OriginCell::cancel_waiter(&cell, second));
|
1621 1663 | assert!(OriginCell::cancel_waiter(&cell, fourth));
|
1622 - | assert_eq!(demand, cell.snapshot().demand);
|
1623 - | assert_eq!(2, cell.snapshot().waiting);
|
1664 + | assert_eq!(demand, cell.probe().demand);
|
1665 + | assert_eq!(2, cell.probe().waiting);
|
1624 1666 | let fifth = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1625 - | assert_eq!(3, cell.snapshot().waiting);
|
1667 + | assert_eq!(3, cell.probe().waiting);
|
1626 1668 |
|
1627 1669 | drop(held);
|
1628 1670 | let first_lease = OriginCell::take_ready_lease(&cell, first).unwrap();
|
1629 1671 | assert!(OriginCell::take_ready_lease(&cell, third).is_none());
|
1630 1672 | assert!(OriginCell::take_ready_lease(&cell, fifth).is_none());
|
1631 1673 | drop(first_lease);
|
1632 1674 |
|
1633 1675 | let third_lease = OriginCell::take_ready_lease(&cell, third).unwrap();
|
1634 1676 | assert!(OriginCell::take_ready_lease(&cell, fifth).is_none());
|
1635 1677 | drop(third_lease);
|
1636 1678 |
|
1637 1679 | drop(OriginCell::take_ready_lease(&cell, fifth).unwrap());
|
1638 - | assert_eq!(0, cell.snapshot().retained);
|
1680 + | assert_eq!(0, cell.probe().retained);
|
1639 1681 | }
|
1640 1682 |
|
1641 1683 | #[test]
|
1642 1684 | fn cancelling_head_waiter_preserves_remaining_fifo_order() {
|
1643 1685 | let (_admission, cell, held) = saturated_bounded_cell();
|
1644 1686 | let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1645 1687 | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1646 1688 | let third = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1647 1689 | let fourth = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1648 1690 |
|
1649 1691 | assert!(OriginCell::cancel_waiter(&cell, first));
|
1650 1692 | drop(held);
|
1651 1693 |
|
1652 1694 | let second_lease = OriginCell::take_ready_lease(&cell, second).unwrap();
|
1653 1695 | assert!(OriginCell::take_ready_lease(&cell, third).is_none());
|
1654 1696 | assert!(OriginCell::take_ready_lease(&cell, fourth).is_none());
|
1655 1697 | drop(second_lease);
|
1656 1698 |
|
1657 1699 | let third_lease = OriginCell::take_ready_lease(&cell, third).unwrap();
|
1658 1700 | assert!(OriginCell::take_ready_lease(&cell, fourth).is_none());
|
1659 1701 | drop(third_lease);
|
1660 1702 |
|
1661 1703 | drop(OriginCell::take_ready_lease(&cell, fourth).unwrap());
|
1662 - | assert_eq!(0, cell.snapshot().retained);
|
1704 + | assert_eq!(0, cell.probe().retained);
|
1663 1705 | }
|
1664 1706 |
|
1665 1707 | #[test]
|
1666 1708 | fn cancelling_ready_waiter_refunnels_capacity() {
|
1667 1709 | let (_admission, cell) = bounded_cell();
|
1668 1710 | let first = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1669 1711 | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1670 1712 |
|
1671 1713 | assert!(OriginCell::cancel_waiter(&cell, first));
|
1672 1714 | let lease = OriginCell::take_ready_lease(&cell, second).unwrap();
|
1673 1715 | drop(lease);
|
1674 1716 | }
|
1675 1717 |
|
1676 1718 | #[test]
|
1677 1719 | fn cancelling_during_delivery_refunnels_capacity_after_unlock() {
|
1678 1720 | let (admission, cell) = bounded_cell();
|
1679 1721 | let (waiter, demand) =
|
1680 1722 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1681 1723 | let mut delivery =
|
1682 - | OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
|
1724 + | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
1683 1725 | .expect("published demand did not reserve capacity");
|
1684 - | assert!(delivery.materialize_for_test());
|
1726 + | assert!(delivery.resolve_payload_for_test());
|
1685 1727 | let reservation = {
|
1686 1728 | let mut state = cell.state.lock();
|
1687 1729 | state
|
1688 - | .waiters
|
1730 + | .acquisitions
|
1689 1731 | .reserve_delivery_waiter(delivery.demand(), &cell.eligibility_group)
|
1690 1732 | };
|
1691 1733 | let DeliveryReservation::Reserved {
|
1692 1734 | waiter: reserved,
|
1693 1735 | successor,
|
1694 1736 | } = reservation
|
1695 1737 | else {
|
1696 1738 | panic!("current delivery was rejected");
|
1697 1739 | };
|
1698 1740 | assert_eq!(waiter, reserved);
|
1699 - | let (event, acknowledgement) = delivery.commit(successor);
|
1700 - | let AcquisitionEvent::Establish(permit) = event else {
|
1741 + | let (event, settlement) = delivery.into_step(successor);
|
1742 + | let AcquisitionStep::StartEstablishment(permit) = event else {
|
1701 1743 | panic!("capacity delivery materialized a non-capacity event");
|
1702 1744 | };
|
1703 1745 |
|
1704 1746 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1705 - | let installation = {
|
1747 + | let commitment = {
|
1706 1748 | let mut state = cell.state.lock();
|
1707 - | state.waiters.install_capacity(waiter, permit)
|
1749 + | state.acquisitions.commit_capacity(waiter, permit)
|
1708 1750 | };
|
1709 - | assert!(installation.returned_events[0].is_none());
|
1751 + | let CellCommitOutcome::Refused { returned, .. } = commitment else {
|
1752 + | panic!("cancelled delivery accepted its capacity");
|
1753 + | };
|
1754 + | assert!(returned[0].is_none());
|
1710 1755 | assert!(matches!(
|
1711 - | installation.returned_events[1],
|
1712 - | Some(AcquisitionEvent::Establish(_))
|
1756 + | returned[1],
|
1757 + | Some(AcquisitionStep::StartEstablishment(_))
|
1713 1758 | ));
|
1714 - | assert!(installation.error.is_none());
|
1715 - | assert_eq!(0, cell.snapshot().retained);
|
1759 + | assert_eq!(0, cell.probe().retained);
|
1716 1760 |
|
1717 - | drop(installation.returned_events);
|
1761 + | drop(returned);
|
1718 1762 | assert_eq!(1, admission.available_capacity_for_test());
|
1719 - | assert!(acknowledgement.accept().is_none());
|
1763 + | assert!(settlement.accept().is_none());
|
1720 1764 | assert_eq!(1, admission.available_capacity_for_test());
|
1721 1765 | }
|
1722 1766 |
|
1723 1767 | #[test]
|
1724 1768 | fn h1_return_during_capacity_delivery_refunnels_the_permit() {
|
1725 1769 | let (admission, cell) = bounded_cell_with_limit(3);
|
1726 1770 | let first_lease = OriginAdmission::lease_for_test(&admission);
|
1727 1771 | let (first_connection, _first_physical) =
|
1728 1772 | ConnectionState::bounded(connection_info(1), first_lease);
|
1729 1773 | let first_return =
|
1730 - | OriginCell::install_selected_h1(&cell, first_connection, H1Sender::test(11));
|
1774 + | OriginCell::insert_selected_h1(&cell, first_connection, H1Sender::test(11));
|
1731 1775 | let second_lease = OriginAdmission::lease_for_test(&admission);
|
1732 1776 | let (second_connection, _second_physical) =
|
1733 1777 | ConnectionState::bounded(connection_info(2), second_lease);
|
1734 1778 | let second_return =
|
1735 - | OriginCell::install_selected_h1(&cell, second_connection, H1Sender::test(22));
|
1779 + | OriginCell::insert_selected_h1(&cell, second_connection, H1Sender::test(22));
|
1736 1780 | let (waiter, demand) =
|
1737 1781 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1738 1782 | let mut delivery =
|
1739 - | OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
|
1783 + | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
1740 1784 | .expect("published demand did not reserve capacity");
|
1741 - | assert!(delivery.materialize_for_test());
|
1785 + | assert!(delivery.resolve_payload_for_test());
|
1742 1786 | let reservation = {
|
1743 1787 | let mut state = cell.state.lock();
|
1744 1788 | state
|
1745 - | .waiters
|
1789 + | .acquisitions
|
1746 1790 | .reserve_delivery_waiter(delivery.demand(), &cell.eligibility_group)
|
1747 1791 | };
|
1748 1792 | let DeliveryReservation::Reserved {
|
1749 1793 | waiter: reserved,
|
1750 1794 | successor,
|
1751 1795 | } = reservation
|
1752 1796 | else {
|
1753 1797 | panic!("current delivery was rejected");
|
1754 1798 | };
|
1755 1799 | assert_eq!(waiter, reserved);
|
1756 1800 |
|
1757 1801 | drop(first_return);
|
1758 1802 | drop(second_return);
|
1759 1803 | assert_eq!((2, 1), cell.h1_counts());
|
1760 - | let (event, acknowledgement) = delivery.commit(successor);
|
1761 - | let AcquisitionEvent::Establish(permit) = event else {
|
1804 + | let (event, settlement) = delivery.into_step(successor);
|
1805 + | let AcquisitionStep::StartEstablishment(permit) = event else {
|
1762 1806 | panic!("capacity delivery materialized a non-capacity event");
|
1763 1807 | };
|
1764 - | let installation = {
|
1808 + | let commitment = {
|
1765 1809 | let mut state = cell.state.lock();
|
1766 - | state.waiters.install_capacity(waiter, permit)
|
1810 + | state.acquisitions.commit_capacity(waiter, permit)
|
1811 + | };
|
1812 + | let CellCommitOutcome::Refused { returned, .. } = commitment else {
|
1813 + | panic!("capacity delivery displaced a local HTTP/1 result");
|
1767 1814 | };
|
1768 - | assert!(installation.returned_events[0].is_some());
|
1769 - | drop(installation.returned_events);
|
1770 - | assert!(acknowledgement.accept().is_none());
|
1815 + | assert!(returned[0].is_some());
|
1816 + | drop(returned);
|
1817 + | assert!(settlement.accept().is_none());
|
1771 1818 |
|
1772 1819 | assert_eq!(1, admission.available_capacity_for_test());
|
1773 1820 | let selected = cell
|
1774 1821 | .take_ready_h1(waiter)
|
1775 1822 | .expect("returning H1 did not win the crossing delivery");
|
1776 1823 | assert_eq!(11, selected.test_sender_id());
|
1777 1824 | drop(selected);
|
1778 1825 | }
|
1779 1826 |
|
1780 1827 | #[test]
|
1781 1828 | fn cancellation_during_h1_and_capacity_crossing_returns_both() {
|
1782 1829 | let (admission, cell) = bounded_cell_with_limit(2);
|
1783 1830 | let installed_lease = OriginAdmission::lease_for_test(&admission);
|
1784 1831 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), installed_lease);
|
1785 - | let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
1832 + | let returning = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
1786 1833 | let (waiter, demand) =
|
1787 1834 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1788 1835 | let mut delivery =
|
1789 - | OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
|
1836 + | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
1790 1837 | .expect("published demand did not reserve capacity");
|
1791 - | assert!(delivery.materialize_for_test());
|
1838 + | assert!(delivery.resolve_payload_for_test());
|
1792 1839 | let reservation = {
|
1793 1840 | let mut state = cell.state.lock();
|
1794 1841 | state
|
1795 - | .waiters
|
1842 + | .acquisitions
|
1796 1843 | .reserve_delivery_waiter(delivery.demand(), &cell.eligibility_group)
|
1797 1844 | };
|
1798 1845 | let DeliveryReservation::Reserved {
|
1799 1846 | waiter: reserved,
|
1800 1847 | successor,
|
1801 1848 | } = reservation
|
1802 1849 | else {
|
1803 1850 | panic!("current delivery was rejected");
|
1804 1851 | };
|
1805 1852 | assert_eq!(waiter, reserved);
|
1806 1853 |
|
1807 1854 | drop(returning);
|
1808 1855 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1809 - | let (event, acknowledgement) = delivery.commit(successor);
|
1810 - | let AcquisitionEvent::Establish(permit) = event else {
|
1856 + | let (event, settlement) = delivery.into_step(successor);
|
1857 + | let AcquisitionStep::StartEstablishment(permit) = event else {
|
1811 1858 | panic!("capacity delivery materialized a non-capacity event");
|
1812 1859 | };
|
1813 - | let installation = {
|
1860 + | let commitment = {
|
1814 1861 | let mut state = cell.state.lock();
|
1815 - | state.waiters.install_capacity(waiter, permit)
|
1862 + | state.acquisitions.commit_capacity(waiter, permit)
|
1816 1863 | };
|
1817 - | assert!(installation.returned_events.iter().all(Option::is_some));
|
1818 - | drop(installation.returned_events);
|
1819 - | assert!(acknowledgement.accept().is_none());
|
1864 + | let CellCommitOutcome::Refused { returned, .. } = commitment else {
|
1865 + | panic!("cancelled crossing retained one of its returned resources");
|
1866 + | };
|
1867 + | assert!(returned.iter().all(Option::is_some));
|
1868 + | drop(returned);
|
1869 + | assert!(settlement.accept().is_none());
|
1820 1870 |
|
1821 1871 | assert_eq!(1, admission.available_capacity_for_test());
|
1822 - | assert_eq!(0, cell.snapshot().retained);
|
1872 + | assert_eq!(0, cell.probe().retained);
|
1823 1873 | assert_eq!((1, 1), cell.h1_counts());
|
1824 1874 | assert_eq!(
|
1825 1875 | 11,
|
1826 1876 | OriginCell::select_h1(&cell)
|
1827 1877 | .expect("cancelled crossing lost its H1 sender")
|
1828 1878 | .test_sender_id()
|
1829 1879 | );
|
1830 1880 | }
|
1831 1881 |
|
1832 1882 | #[test]
|
1833 1883 | fn stale_delivery_is_rejected_after_head_cancellation() {
|
1834 1884 | let (admission, cell) = bounded_cell();
|
1835 1885 | let (first, first_demand) =
|
1836 1886 | cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
1837 1887 | let second = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1838 - | let stale = OriginAdmission::publish_without_driving(
|
1888 + | let stale = OriginAdmission::submit_without_running(
|
1839 1889 | &admission,
|
1840 1890 | cell.id().partition(),
|
1841 1891 | first_demand,
|
1842 1892 | )
|
1843 1893 | .expect("first demand did not reserve capacity");
|
1844 1894 |
|
1845 1895 | assert!(OriginCell::cancel_waiter(&cell, first));
|
1846 - | OriginAdmission::drive(Some(AdmissionAction::Delivery(stale)));
|
1896 + | OriginAdmission::run_action_chain(Some(AdmissionAction::Deliver(stale)));
|
1847 1897 |
|
1848 1898 | let lease = OriginCell::take_ready_lease(&cell, second)
|
1849 1899 | .expect("replacement demand did not receive refunnelled capacity");
|
1850 1900 | drop(lease);
|
1851 1901 | }
|
1852 1902 |
|
1853 1903 | #[test]
|
1854 1904 | fn stale_delivery_returns_its_successor_to_the_driver() {
|
1855 1905 | let admission = OriginAdmission::for_test(NonZeroUsize::new(1).unwrap());
|
1856 1906 | let stale = OriginAdmission::register_cell(
|
1857 1907 | &admission,
|
1858 1908 | Arc::new(OriginCell::new(
|
1859 1909 | PartitionId::from_index(1),
|
1860 1910 | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
1861 1911 | EligibilityGroup::Pool,
|
1862 1912 | Some(admission.clone()),
|
1863 1913 | None,
|
1864 1914 | )),
|
1865 1915 | );
|
1866 1916 | let target = OriginAdmission::register_cell(
|
1867 1917 | &admission,
|
1868 1918 | Arc::new(OriginCell::new(
|
1869 1919 | PartitionId::from_index(2),
|
1870 1920 | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
1871 1921 | EligibilityGroup::Pool,
|
1872 1922 | Some(admission.clone()),
|
1873 1923 | None,
|
1874 1924 | )),
|
1875 1925 | );
|
1876 1926 |
|
1877 1927 | let (stale_waiter, stale_demand) =
|
1878 1928 | stale.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1879 - | let mut delivery = OriginAdmission::publish_without_driving(
|
1929 + | let mut delivery = OriginAdmission::submit_without_running(
|
1880 1930 | &admission,
|
1881 1931 | stale.id().partition(),
|
1882 1932 | stale_demand,
|
1883 1933 | )
|
1884 1934 | .expect("stale demand did not reserve capacity");
|
1885 1935 | let (target_waiter, target_demand) =
|
1886 1936 | target.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
1887 - | OriginAdmission::publish_demand(&admission, target.id().partition(), target_demand);
|
1937 + | OriginAdmission::submit_demand_snapshot(&admission, target.id().partition(), target_demand);
|
1888 1938 |
|
1889 1939 | // Model cancellation winning in the cell before its retirement
|
1890 1940 | // publication reaches admission.
|
1891 1941 | let cancelled = stale
|
1892 1942 | .state
|
1893 1943 | .lock()
|
1894 - | .waiters
|
1944 + | .acquisitions
|
1895 1945 | .cancel_waiter(stale_waiter, &stale.eligibility_group)
|
1896 1946 | .expect("stale waiter was not cancelled");
|
1897 - | drop(cancelled.returned_events);
|
1947 + | drop(cancelled.returned_steps);
|
1898 1948 |
|
1899 - | assert!(delivery.materialize_for_test());
|
1949 + | assert!(delivery.resolve_payload_for_test());
|
1900 1950 | let next = OriginCell::receive_delivery(&stale, delivery)
|
1901 1951 | .expect("stale rejection did not return the successor delivery");
|
1902 1952 | assert!(
|
1903 1953 | OriginCell::take_ready_lease(&target, target_waiter).is_none(),
|
1904 1954 | "stale rejection recursively drove its successor"
|
1905 1955 | );
|
1906 1956 |
|
1907 - | OriginAdmission::drive(Some(next));
|
1957 + | OriginAdmission::run_action_chain(Some(next));
|
1908 1958 | drop(
|
1909 1959 | OriginCell::take_ready_lease(&target, target_waiter)
|
1910 1960 | .expect("existing driver did not deliver the successor"),
|
1911 1961 | );
|
1912 1962 | }
|
1913 1963 |
|
1914 1964 | #[test]
|
1915 1965 | fn cancelling_the_only_waiter_retires_admission_demand() {
|
1916 1966 | let (admission, cell, held) = saturated_bounded_cell();
|
1917 1967 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1918 1968 | assert_eq!(1, admission.ordered_demand_count_for_test());
|
1919 1969 |
|
1920 1970 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1921 1971 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
1922 1972 | drop(held);
|
1923 1973 | }
|
1924 1974 |
|
1925 1975 | #[test]
|
1926 1976 | fn unknown_waiter_cancellation_is_a_noop() {
|
1927 1977 | let (_admission, cell) = bounded_cell();
|
1928 1978 | assert!(!OriginCell::cancel_waiter(&cell, WaiterId(99)));
|
1929 1979 | }
|
1930 1980 |
|
1931 1981 | #[test]
|
1932 1982 | fn unbounded_waiter_registers_and_cancels_locally() {
|
1933 1983 | let cell = Arc::new(OriginCell::new(
|
1934 1984 | PartitionId::from_index(1),
|
1935 1985 | OriginKey::from_parts(Scheme::HTTPS, "example.com", None).unwrap(),
|
1936 1986 | EligibilityGroup::Pool,
|
1937 1987 | None,
|
1938 1988 | None,
|
1939 1989 | ));
|
1940 1990 |
|
1941 1991 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1942 - | assert_eq!(0, cell.snapshot().waiting);
|
1943 - | assert_eq!(1, cell.snapshot().retained);
|
1992 + | assert_eq!(0, cell.probe().waiting);
|
1993 + | assert_eq!(1, cell.probe().retained);
|
1944 1994 | assert!(OriginCell::cancel_waiter(&cell, waiter));
|
1945 - | assert_eq!(0, cell.snapshot().retained);
|
1995 + | assert_eq!(0, cell.probe().retained);
|
1946 1996 | }
|
1947 1997 |
|
1948 1998 | #[test]
|
1949 1999 | fn delivered_waiter_is_woken_once() {
|
1950 2000 | let (_admission, cell, held) = saturated_bounded_cell();
|
1951 2001 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
1952 2002 | let counter = StdArc::new(WakeCounter(AtomicUsize::new(0)));
|
1953 2003 | let waker = Waker::from(counter.clone());
|
1954 2004 | let mut context = Context::from_waker(&waker);
|
1955 2005 | assert!(cell.poll_waiter(waiter, &mut context).is_pending());
|
1956 2006 |
|
1957 2007 | drop(held);
|
1958 2008 | assert_eq!(1, counter.0.load(Ordering::Relaxed));
|
1959 2009 | drop(OriginCell::take_ready_lease(&cell, waiter).expect("woken waiter had no capacity"));
|
1960 2010 | assert_eq!(1, counter.0.load(Ordering::Relaxed));
|
1961 2011 | }
|
1962 2012 |
|
1963 2013 | fn install_bounded_h2(
|
1964 2014 | admission: &Arc<OriginAdmission>,
|
1965 2015 | cell: &Arc<OriginCell>,
|
1966 2016 | connection_id: u64,
|
1967 2017 | ) -> (
|
1968 2018 | h2::H2GenerationId,
|
1969 2019 | Arc<ConnectionState>,
|
1970 2020 | super::super::connection::PhysicalConnectionGuard,
|
1971 2021 | ) {
|
1972 2022 | let lease = OriginAdmission::lease_for_test(admission);
|
1973 2023 | let (connection, physical) = ConnectionState::bounded(
|
1974 2024 | ConnectionInfo::for_test(ConnectionId::new(connection_id), cell.id().partition()),
|
1975 2025 | lease,
|
1976 2026 | );
|
1977 2027 | let generation =
|
1978 2028 | OriginCell::install_h2_for_test(cell, connection.clone(), connection_id, None);
|
1979 2029 | (generation, connection, physical)
|
1980 2030 | }
|
1981 2031 |
|
1982 2032 | fn take_ready_h2(cell: &OriginCell, waiter: WaiterId) -> h2::H2Activation {
|
1983 2033 | match cell
|
1984 2034 | .take_ready_event(waiter)
|
1985 2035 | .expect("HTTP/2 waiter did not receive an activation")
|
1986 2036 | {
|
1987 - | AcquisitionEvent::Complete(AcquisitionResult::H2(activation)) => activation,
|
1988 - | AcquisitionEvent::Establish(_) => {
|
2037 + | AcquisitionStep::Resolved(AcquisitionOutcome::H2(activation)) => activation,
|
2038 + | AcquisitionStep::StartEstablishment(_) => {
|
1989 2039 | panic!("HTTP/2 waiter received establishment capacity")
|
1990 2040 | }
|
1991 - | AcquisitionEvent::Complete(AcquisitionResult::H1(_)) => {
|
2041 + | AcquisitionStep::Resolved(AcquisitionOutcome::H1(_)) => {
|
1992 2042 | panic!("HTTP/2 waiter received an HTTP/1 sender")
|
1993 2043 | }
|
1994 - | AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
|
2044 + | AcquisitionStep::Resolved(AcquisitionOutcome::RetryAcquisition) => {
|
1995 2045 | panic!("HTTP/2 waiter was returned for reacquisition")
|
1996 2046 | }
|
1997 - | AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
|
2047 + | AcquisitionStep::Resolved(AcquisitionOutcome::Failed(error)) => {
|
1998 2048 | panic!("HTTP/2 waiter received an establishment failure: {error}")
|
1999 2049 | }
|
2000 2050 | }
|
2001 2051 | }
|
2002 2052 |
|
2003 2053 | #[test]
|
2004 2054 | fn eligible_peer_h2_generation_satisfies_bounded_demand() {
|
2005 2055 | let (admission, connection_cell, requesting_cell) =
|
2006 2056 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2007 2057 | let (generation, connection, _physical) =
|
2008 2058 | install_bounded_h2(&admission, &connection_cell, 1);
|
2009 2059 |
|
2010 2060 | let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
|
2011 2061 | let activation = take_ready_h2(&requesting_cell, waiter);
|
2012 2062 |
|
2013 2063 | assert_eq!(generation, activation.generation());
|
2014 2064 | assert!(Arc::ptr_eq(&connection, activation.connection()));
|
2015 2065 | assert_eq!(0, admission.available_capacity_for_test());
|
2016 2066 | assert_eq!(Some(generation), connection_cell.accepting_h2_generation());
|
2017 2067 | drop(activation);
|
2018 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2068 + | assert_eq!(0, requesting_cell.probe().retained);
|
2019 2069 | assert!(OriginCell::close_h2(
|
2020 2070 | &connection_cell,
|
2021 2071 | generation,
|
2022 2072 | CloseReason::PoolDropped,
|
2023 2073 | ));
|
2024 2074 | assert_eq!(1, admission.available_capacity_for_test());
|
2025 2075 | }
|
2026 2076 |
|
2027 2077 | #[test]
|
2028 - | fn peer_h2_publication_respects_eligibility_group() {
|
2078 + | fn peer_h2_route_respects_eligibility_group() {
|
2029 2079 | let groups = [
|
2030 2080 | (
|
2031 2081 | EligibilityGroup::Partition(PartitionId::from_index(1)),
|
2032 2082 | EligibilityGroup::Partition(PartitionId::from_index(2)),
|
2033 2083 | ),
|
2034 2084 | (
|
2035 2085 | EligibilityGroup::NetworkInterface(Some(Arc::from("eth0"))),
|
2036 2086 | EligibilityGroup::NetworkInterface(Some(Arc::from("eth1"))),
|
2037 2087 | ),
|
2038 2088 | ];
|
2039 2089 | for (connection_group, requesting_group) in groups {
|
2040 2090 | let (admission, connection_cell, requesting_cell) =
|
2041 2091 | bounded_peer_cells(1, connection_group, requesting_group);
|
2042 2092 | let (generation, _connection, _physical) =
|
2043 2093 | install_bounded_h2(&admission, &connection_cell, 1);
|
2044 2094 | let waiter =
|
2045 2095 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
|
2046 2096 |
|
2047 2097 | assert!(requesting_cell.take_ready_event(waiter).is_none());
|
2048 2098 | assert_eq!(1, admission.ordered_demand_count_for_test());
|
2049 2099 | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
2050 2100 | assert!(OriginCell::close_h2(
|
2051 2101 | &connection_cell,
|
2052 2102 | generation,
|
2053 2103 | CloseReason::PoolDropped,
|
2054 2104 | ));
|
2055 2105 | assert_eq!(1, admission.available_capacity_for_test());
|
2056 2106 | }
|
2057 2107 | }
|
2058 2108 |
|
2059 2109 | #[test]
|
2060 - | fn dropped_publication_guard_retries_the_same_demand() {
|
2110 + | fn dropped_route_guard_retries_the_same_demand() {
|
2061 2111 | let (admission, connection_cell, requesting_cell) =
|
2062 2112 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2063 2113 | let (generation, _connection, _physical) =
|
2064 2114 | install_bounded_h2(&admission, &connection_cell, 1);
|
2065 2115 | let (waiter, demand) =
|
2066 2116 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2067 - | let action = OriginAdmission::publish_action_without_driving(
|
2117 + | let action = OriginAdmission::submit_action_without_running(
|
2068 2118 | &admission,
|
2069 2119 | requesting_cell.id().partition(),
|
2070 2120 | demand,
|
2071 2121 | )
|
2072 - | .expect("peer demand did not prepare publication");
|
2122 + | .expect("peer demand did not prepare a route");
|
2073 2123 |
|
2074 2124 | drop(action);
|
2075 2125 |
|
2076 2126 | drop(take_ready_h2(&requesting_cell, waiter));
|
2077 2127 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2078 2128 | assert!(OriginCell::close_h2(
|
2079 2129 | &connection_cell,
|
2080 2130 | generation,
|
2081 2131 | CloseReason::PoolDropped,
|
2082 2132 | ));
|
2083 2133 | }
|
2084 2134 |
|
2085 2135 | #[test]
|
2086 - | fn requesting_cell_cancellation_closes_an_in_flight_publication_fence() {
|
2136 + | fn requesting_cell_cancellation_closes_an_in_flight_route_assignment() {
|
2087 2137 | let (admission, connection_cell, requesting_cell) =
|
2088 2138 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2089 2139 | let (generation, _connection, _physical) =
|
2090 2140 | install_bounded_h2(&admission, &connection_cell, 1);
|
2091 2141 | let (waiter, demand) =
|
2092 2142 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2093 - | let action = OriginAdmission::publish_action_without_driving(
|
2143 + | let action = OriginAdmission::submit_action_without_running(
|
2094 2144 | &admission,
|
2095 2145 | requesting_cell.id().partition(),
|
2096 2146 | demand,
|
2097 2147 | )
|
2098 - | .expect("peer demand did not prepare publication");
|
2148 + | .expect("peer demand did not prepare a route");
|
2099 2149 |
|
2100 2150 | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
2101 - | OriginAdmission::drive(Some(action));
|
2151 + | OriginAdmission::run_action_chain(Some(action));
|
2102 2152 |
|
2103 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2153 + | assert_eq!(0, requesting_cell.probe().retained);
|
2104 2154 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2105 2155 | assert_eq!(0, admission.available_capacity_for_test());
|
2106 2156 | assert_eq!(Some(generation), connection_cell.accepting_h2_generation());
|
2107 2157 | assert!(OriginCell::close_h2(
|
2108 2158 | &connection_cell,
|
2109 2159 | generation,
|
2110 2160 | CloseReason::PoolDropped,
|
2111 2161 | ));
|
2112 2162 | }
|
2113 2163 |
|
2114 2164 | #[test]
|
2115 - | fn stale_publication_retries_against_a_replacement_generation() {
|
2165 + | fn stale_route_retries_against_a_replacement_generation() {
|
2116 2166 | let (admission, connection_cell, requesting_cell) =
|
2117 2167 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2118 2168 | let (first, _first_connection, _first_physical) =
|
2119 2169 | install_bounded_h2(&admission, &connection_cell, 1);
|
2120 2170 | let (waiter, demand) =
|
2121 2171 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2122 - | let action = OriginAdmission::publish_action_without_driving(
|
2172 + | let action = OriginAdmission::submit_action_without_running(
|
2123 2173 | &admission,
|
2124 2174 | requesting_cell.id().partition(),
|
2125 2175 | demand,
|
2126 2176 | )
|
2127 - | .expect("peer demand did not prepare publication");
|
2177 + | .expect("peer demand did not prepare a route");
|
2128 2178 |
|
2129 2179 | assert!(OriginCell::close_h2(
|
2130 2180 | &connection_cell,
|
2131 2181 | first,
|
2132 2182 | CloseReason::ProtocolClosed
|
2133 2183 | ));
|
2134 2184 | let (second, _second_connection, _second_physical) =
|
2135 2185 | install_bounded_h2(&admission, &connection_cell, 2);
|
2136 - | OriginAdmission::drive(Some(action));
|
2186 + | OriginAdmission::run_action_chain(Some(action));
|
2137 2187 |
|
2138 2188 | let activation = take_ready_h2(&requesting_cell, waiter);
|
2139 2189 | assert_eq!(second, activation.generation());
|
2140 2190 | drop(activation);
|
2141 2191 | assert!(OriginCell::close_h2(
|
2142 2192 | &connection_cell,
|
2143 2193 | second,
|
2144 2194 | CloseReason::PoolDropped
|
2145 2195 | ));
|
2146 2196 | }
|
2147 2197 |
|
2148 2198 | #[test]
|
2149 2199 | fn stale_peer_route_republishes_demand_to_a_replacement_generation() {
|
2150 2200 | let (admission, connection_cell, requesting_cell) =
|
2151 2201 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2152 2202 | let (first, _first_connection, _first_physical) =
|
2153 2203 | install_bounded_h2(&admission, &connection_cell, 1);
|
2154 2204 | let (waiter, demand) =
|
2155 2205 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2156 - | assert!(OriginCell::install_h2_route(
|
2206 + | assert!(OriginCell::attach_h2_route(
|
2157 2207 | &requesting_cell,
|
2158 2208 | h2::H2Route::new(&connection_cell, first),
|
2159 2209 | &EligibilityGroup::Pool,
|
2160 2210 | demand.id_for_test(),
|
2161 2211 | ));
|
2162 2212 |
|
2163 2213 | assert!(OriginCell::close_h2(
|
2164 2214 | &connection_cell,
|
2165 2215 | first,
|
2166 2216 | CloseReason::ProtocolClosed
|
2167 2217 | ));
|
2168 2218 | let (second, _second_connection, _second_physical) =
|
2169 2219 | install_bounded_h2(&admission, &connection_cell, 2);
|
2170 - | OriginCell::service_peer_h2_waiters(&requesting_cell);
|
2220 + | OriginCell::offer_peer_h2(&requesting_cell);
|
2171 2221 | let activation = take_ready_h2(&requesting_cell, waiter);
|
2172 2222 | assert_eq!(second, activation.generation());
|
2173 2223 | drop(activation);
|
2174 2224 | assert!(OriginCell::close_h2(
|
2175 2225 | &connection_cell,
|
2176 2226 | second,
|
2177 2227 | CloseReason::PoolDropped
|
2178 2228 | ));
|
2179 2229 | }
|
2180 2230 |
|
2181 2231 | #[test]
|
2182 2232 | fn direct_selection_removes_a_stale_peer_route() {
|
2183 2233 | let (admission, connection_cell, requesting_cell) =
|
2184 2234 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2185 2235 | let (generation, _connection, _physical) =
|
2186 2236 | install_bounded_h2(&admission, &connection_cell, 1);
|
2187 2237 | let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
|
2188 2238 | drop(take_ready_h2(&requesting_cell, waiter));
|
2189 2239 |
|
2190 2240 | assert!(OriginCell::close_h2(
|
2191 2241 | &connection_cell,
|
2192 2242 | generation,
|
2193 2243 | CloseReason::ProtocolClosed,
|
2194 2244 | ));
|
2195 2245 | assert!(OriginCell::select_h2(&requesting_cell).is_none());
|
2196 2246 | assert!(
|
2197 2247 | !requesting_cell.state.lock().h2.has_visible_h2(),
|
2198 2248 | "direct selection retained a stale peer route"
|
2199 2249 | );
|
2200 2250 | }
|
2201 2251 |
|
2202 2252 | #[test]
|
2203 2253 | fn open_peer_route_allows_concurrent_prospective_activations() {
|
2204 2254 | let (admission, connection_cell, requesting_cell) =
|
2205 2255 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2206 2256 | let (generation, _connection, _physical) =
|
2207 2257 | install_bounded_h2(&admission, &connection_cell, 1);
|
2208 2258 | let waiter = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
|
2209 2259 | drop(take_ready_h2(&requesting_cell, waiter));
|
2210 2260 |
|
2211 2261 | let first = OriginCell::select_h2(&requesting_cell)
|
2212 2262 | .expect("first direct peer activation was not selected");
|
2213 2263 | let second = OriginCell::select_h2(&requesting_cell)
|
2214 2264 | .expect("open peer route serialized the second activation");
|
2215 2265 | assert_eq!(generation, first.generation());
|
2216 2266 | assert_eq!(generation, second.generation());
|
2217 2267 | assert_eq!(Some((2, 0)), connection_cell.h2_request_counts(generation));
|
2218 2268 | drop(first);
|
2219 2269 | drop(second);
|
2220 2270 | assert!(OriginCell::close_h2(
|
2221 2271 | &connection_cell,
|
2222 2272 | generation,
|
2223 2273 | CloseReason::PoolDropped
|
2224 2274 | ));
|
2225 2275 | }
|
2226 2276 |
|
2227 2277 | #[test]
|
2228 - | fn peer_generation_gate_preserves_prepublication_order() {
|
2278 + | fn peer_generation_gate_preserves_pre_route_order() {
|
2229 2279 | let (admission, connection_cell, requesting_cell) =
|
2230 2280 | bounded_peer_cells(1, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2231 2281 | let (generation, _connection, _physical) =
|
2232 2282 | install_bounded_h2(&admission, &connection_cell, 1);
|
2233 2283 | let (first, demand) =
|
2234 2284 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2235 2285 | let second = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
|
2236 - | let action = OriginAdmission::publish_action_without_driving(
|
2286 + | let action = OriginAdmission::submit_action_without_running(
|
2237 2287 | &admission,
|
2238 2288 | requesting_cell.id().partition(),
|
2239 2289 | demand,
|
2240 2290 | )
|
2241 - | .expect("peer demand did not prepare publication");
|
2242 - | OriginAdmission::drive(Some(action));
|
2291 + | .expect("peer demand did not prepare a route");
|
2292 + | OriginAdmission::run_action_chain(Some(action));
|
2243 2293 | let third = OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H2Required);
|
2244 2294 |
|
2245 2295 | let first_activation = take_ready_h2(&requesting_cell, first);
|
2246 2296 | assert!(requesting_cell.take_ready_event(second).is_none());
|
2247 2297 | assert!(requesting_cell.take_ready_event(third).is_none());
|
2248 2298 | drop(first_activation);
|
2249 2299 |
|
2250 2300 | let second_activation = take_ready_h2(&requesting_cell, second);
|
2251 2301 | assert!(requesting_cell.take_ready_event(third).is_none());
|
2252 2302 | drop(second_activation);
|
2331 2381 |
|
2332 2382 | fn connection_info(id: u64) -> Arc<ConnectionInfo> {
|
2333 2383 | ConnectionInfo::for_test(ConnectionId::new(id), PartitionId::from_index(1))
|
2334 2384 | }
|
2335 2385 |
|
2336 2386 | #[test]
|
2337 2387 | fn h1_selection_linearizes_against_close() {
|
2338 2388 | loom::model(|| {
|
2339 2389 | let cell = unbounded_cell();
|
2340 2390 | let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
|
2341 - | OriginCell::install_idle_h1(&cell, connection.clone(), H1Sender::test(11));
|
2391 + | OriginCell::insert_idle_h1(&cell, connection.clone(), H1Sender::test(11));
|
2342 2392 | let close = H1CloseHandle::new(&cell, &connection);
|
2343 2393 |
|
2344 2394 | let select_cell = cell.clone();
|
2345 2395 | let select = loom::thread::spawn(move || OriginCell::select_h1(&select_cell));
|
2346 2396 | let close = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2347 2397 |
|
2348 2398 | let selected = select.join().unwrap();
|
2349 2399 | assert!(close.join().unwrap());
|
2350 2400 | drop(selected);
|
2351 2401 |
|
2352 2402 | assert_eq!((0, 0), cell.h1_counts());
|
2353 - | assert_eq!(
|
2354 - | Some(CloseReason::Poisoned),
|
2355 - | connection.snapshot().close_reason
|
2356 - | );
|
2403 + | assert_eq!(Some(CloseReason::Poisoned), connection.probe().close_reason);
|
2357 2404 | });
|
2358 2405 | }
|
2359 2406 |
|
2360 2407 | #[test]
|
2361 2408 | fn h1_return_linearizes_against_close() {
|
2362 2409 | loom::model(|| {
|
2363 2410 | let cell = unbounded_cell();
|
2364 2411 | let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
|
2365 2412 | let selection =
|
2366 - | OriginCell::install_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
2413 + | OriginCell::insert_selected_h1(&cell, connection.clone(), H1Sender::test(11));
|
2367 2414 | let exchange = selection.into_exchange();
|
2368 2415 | let close = H1CloseHandle::new(&cell, &connection);
|
2369 2416 |
|
2370 2417 | let returning = loom::thread::spawn(move || exchange.offer_for_reuse());
|
2371 2418 | let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2372 2419 | returning.join().unwrap();
|
2373 2420 | assert!(closing.join().unwrap());
|
2374 2421 |
|
2375 2422 | assert_eq!((0, 0), cell.h1_counts());
|
2376 - | assert_eq!(
|
2377 - | Some(CloseReason::Poisoned),
|
2378 - | connection.snapshot().close_reason
|
2379 - | );
|
2423 + | assert_eq!(Some(CloseReason::Poisoned), connection.probe().close_reason);
|
2380 2424 | });
|
2381 2425 | }
|
2382 2426 |
|
2383 2427 | #[test]
|
2384 2428 | fn h1_delivery_and_waiter_cancellation_preserve_the_sender() {
|
2385 2429 | loom::model(|| {
|
2386 2430 | let cell = unbounded_cell();
|
2387 2431 | let (connection, _physical) = ConnectionState::unbounded(connection_info(1));
|
2388 - | let selection = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
2432 + | let selection = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
2389 2433 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2390 2434 |
|
2391 2435 | let returning = loom::thread::spawn(move || drop(selection));
|
2392 2436 | let cancel_cell = cell.clone();
|
2393 2437 | let cancelling =
|
2394 2438 | loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter));
|
2395 2439 | returning.join().unwrap();
|
2396 2440 | assert!(cancelling.join().unwrap());
|
2397 2441 |
|
2398 2442 | assert_eq!((1, 1), cell.h1_counts());
|
2399 - | assert_eq!(0, cell.snapshot().retained);
|
2443 + | assert_eq!(0, cell.probe().retained);
|
2400 2444 | assert_eq!(
|
2401 2445 | 11,
|
2402 2446 | OriginCell::select_h1(&cell)
|
2403 2447 | .expect("sender was lost during return/cancel")
|
2404 2448 | .test_sender_id()
|
2405 2449 | );
|
2406 2450 | });
|
2407 2451 | }
|
2408 2452 |
|
2409 2453 | #[test]
|
2410 2454 | fn returned_h1_and_establishment_race_for_one_waiter() {
|
2411 2455 | loom::model(|| {
|
2412 2456 | let cell = unbounded_cell();
|
2413 2457 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2414 - | let AcquisitionEvent::Establish(permit) = cell
|
2458 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
2415 2459 | .take_ready_event(waiter)
|
2416 2460 | .expect("unbounded miss did not start establishment")
|
2417 2461 | else {
|
2418 2462 | panic!("unbounded miss completed before establishment started");
|
2419 2463 | };
|
2420 2464 | assert!(cell.start_establishment(waiter));
|
2421 2465 | drop(permit);
|
2422 2466 |
|
2423 2467 | let (returning_connection, _returning_physical) =
|
2424 2468 | ConnectionState::unbounded(connection_info(1));
|
2425 2469 | let returning =
|
2426 - | OriginCell::install_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
2470 + | OriginCell::insert_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
2427 2471 | let (fresh_connection, _fresh_physical) =
|
2428 2472 | ConnectionState::unbounded(connection_info(2));
|
2429 - | let fresh =
|
2430 - | OriginCell::install_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
2473 + | let fresh = OriginCell::insert_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
2431 2474 |
|
2432 2475 | let returning = loom::thread::spawn(move || drop(returning));
|
2433 2476 | let completion_cell = cell.clone();
|
2434 2477 | let completing = loom::thread::spawn(move || {
|
2435 - | completion_cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
|
2478 + | completion_cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
2436 2479 | });
|
2437 2480 | returning.join().unwrap();
|
2438 2481 | completing.join().unwrap();
|
2439 2482 |
|
2440 2483 | assert_eq!((2, 1), cell.h1_counts());
|
2441 2484 | let winner = cell
|
2442 2485 | .take_ready_h1(waiter)
|
2443 - | .expect("both acquisition results were lost");
|
2486 + | .expect("both acquisition outcomes were lost");
|
2444 2487 | assert!(matches!(winner.test_sender_id(), 11 | 22));
|
2445 2488 | drop(winner);
|
2446 2489 | assert_eq!((2, 2), cell.h1_counts());
|
2447 2490 | });
|
2448 2491 | }
|
2449 2492 |
|
2450 2493 | #[test]
|
2451 2494 | fn first_establishment_poll_races_returned_h1() {
|
2452 2495 | loom::model(|| {
|
2453 2496 | let cell = unbounded_cell();
|
2454 2497 | let waiter = OriginCell::register_waiter(&cell, ProtocolRequirement::H1Compatible);
|
2455 - | let AcquisitionEvent::Establish(permit) = cell
|
2498 + | let AcquisitionStep::StartEstablishment(permit) = cell
|
2456 2499 | .take_ready_event(waiter)
|
2457 2500 | .expect("unbounded miss did not prepare establishment")
|
2458 2501 | else {
|
2459 2502 | panic!("unbounded miss completed before establishment was prepared");
|
2460 2503 | };
|
2461 2504 |
|
2462 2505 | let (returning_connection, _returning_physical) =
|
2463 2506 | ConnectionState::unbounded(connection_info(1));
|
2464 2507 | let returning =
|
2465 - | OriginCell::install_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
2508 + | OriginCell::insert_selected_h1(&cell, returning_connection, H1Sender::test(11));
|
2466 2509 | let return_thread = loom::thread::spawn(move || drop(returning));
|
2467 2510 | let start_cell = cell.clone();
|
2468 2511 | let start_thread = loom::thread::spawn(move || start_cell.start_establishment(waiter));
|
2469 2512 |
|
2470 2513 | return_thread.join().unwrap();
|
2471 2514 | let started = start_thread.join().unwrap();
|
2472 2515 | drop(permit);
|
2473 2516 |
|
2474 2517 | if started {
|
2475 2518 | let (fresh_connection, _fresh_physical) =
|
2476 2519 | ConnectionState::unbounded(connection_info(2));
|
2477 2520 | let fresh =
|
2478 - | OriginCell::install_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
2479 - | cell.complete_establishment(waiter, AcquisitionResult::H1(fresh));
|
2521 + | OriginCell::insert_selected_h1(&cell, fresh_connection, H1Sender::test(22));
|
2522 + | cell.complete_establishment(waiter, AcquisitionOutcome::H1(fresh));
|
2480 2523 | }
|
2481 2524 |
|
2482 2525 | let winner = cell
|
2483 2526 | .take_ready_h1(waiter)
|
2484 2527 | .expect("returned H1 was lost at the first-poll boundary");
|
2485 2528 | assert_eq!(11, winner.test_sender_id());
|
2486 2529 | drop(winner);
|
2487 2530 | assert_eq!(
|
2488 2531 | (usize::from(started) + 1, usize::from(started) + 1),
|
2489 2532 | cell.h1_counts()
|
2490 2533 | );
|
2491 2534 | });
|
2492 2535 | }
|
2493 2536 |
|
2494 2537 | #[test]
|
2495 2538 | fn h1_return_and_capacity_delivery_complete_one_waiter() {
|
2496 2539 | loom::model(|| {
|
2497 2540 | let (admission, cell) = bounded_cell_with_limit(2);
|
2498 2541 | let installed_lease = OriginAdmission::lease_for_test(&admission);
|
2499 2542 | let (connection, _physical) =
|
2500 2543 | ConnectionState::bounded(connection_info(1), installed_lease);
|
2501 - | let returning = OriginCell::install_selected_h1(&cell, connection, H1Sender::test(11));
|
2544 + | let returning = OriginCell::insert_selected_h1(&cell, connection, H1Sender::test(11));
|
2502 2545 | let (waiter, demand) =
|
2503 2546 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2504 2547 | let mut delivery =
|
2505 - | OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
|
2548 + | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
2506 2549 | .expect("published demand did not reserve capacity");
|
2507 2550 | assert!(
|
2508 - | delivery.materialize_for_test(),
|
2551 + | delivery.resolve_payload_for_test(),
|
2509 2552 | "capacity delivery did not materialize"
|
2510 2553 | );
|
2511 2554 |
|
2512 2555 | let delivery_cell = cell.clone();
|
2513 2556 | let delivering = loom::thread::spawn(move || {
|
2514 2557 | drop(OriginCell::receive_delivery(&delivery_cell, delivery));
|
2515 2558 | });
|
2516 2559 | let returning = loom::thread::spawn(move || drop(returning));
|
2517 2560 | delivering.join().unwrap();
|
2518 2561 | returning.join().unwrap();
|
2519 2562 |
|
2520 2563 | assert_eq!(1, admission.available_capacity_for_test());
|
2521 2564 | let selection = cell
|
2522 2565 | .take_ready_h1(waiter)
|
2523 2566 | .expect("return and capacity delivery lost the HTTP/1 sender");
|
2524 2567 | assert_eq!(11, selection.test_sender_id());
|
2525 2568 | drop(selection);
|
2526 2569 | admission.clear_modeled_cells_for_test();
|
2527 2570 | });
|
2528 2571 | }
|
2529 2572 |
|
2530 2573 | #[test]
|
2531 - | fn peer_reuse_and_request_cancellation_preserve_the_sender() {
|
2574 + | fn peer_match_and_request_cancellation_preserve_the_sender() {
|
2532 2575 | let mut model = loom::model::Builder::new();
|
2533 2576 | model.preemption_bound = Some(2);
|
2534 2577 | model.check(|| {
|
2535 2578 | let (admission, connection_cell, requesting_cell) =
|
2536 2579 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2537 2580 | let lease = OriginAdmission::lease_for_test(&admission);
|
2538 2581 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2539 2582 | let returning =
|
2540 - | OriginCell::install_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
2583 + | OriginCell::insert_selected_h1(&connection_cell, connection, H1Sender::test(11));
|
2541 2584 | let waiter =
|
2542 2585 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
2543 2586 |
|
2544 - | let returning = loom::thread::spawn(move || drop(returning));
|
2587 + | // Loom's default 4 KiB coroutine stack is too small for the full
|
2588 + | // sender-return and admission-settlement path modeled here.
|
2589 + | let returning = loom::thread::Builder::new()
|
2590 + | .stack_size(16 * 1024)
|
2591 + | .spawn(move || drop(returning))
|
2592 + | .unwrap();
|
2545 2593 | let cancel_requesting_cell = requesting_cell.clone();
|
2546 - | let cancelling = loom::thread::spawn(move || {
|
2547 - | OriginCell::cancel_waiter(&cancel_requesting_cell, waiter)
|
2548 - | });
|
2594 + | let cancelling = loom::thread::Builder::new()
|
2595 + | .stack_size(16 * 1024)
|
2596 + | .spawn(move || OriginCell::cancel_waiter(&cancel_requesting_cell, waiter))
|
2597 + | .unwrap();
|
2549 2598 | returning.join().unwrap();
|
2550 2599 | assert!(cancelling.join().unwrap());
|
2551 2600 |
|
2552 2601 | assert_eq!((1, 1), connection_cell.h1_counts());
|
2553 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2602 + | assert_eq!(0, requesting_cell.probe().retained);
|
2554 2603 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2555 2604 | admission.clear_modeled_cells_for_test();
|
2556 2605 | });
|
2557 2606 | }
|
2558 2607 |
|
2559 2608 | #[test]
|
2560 2609 | fn borrowed_delivery_racing_a_local_return_preserves_both_senders() {
|
2561 2610 | let mut model = loom::model::Builder::new();
|
2562 2611 | model.preemption_bound = Some(2);
|
2563 2612 | model.check(|| {
|
2564 2613 | let (admission, connection_cell, requesting_cell) =
|
2565 2614 | bounded_peer_cells_with_limit(2, EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2566 2615 | let owning_lease = OriginAdmission::lease_for_test(&admission);
|
2567 2616 | let (owned_connection, _owning_physical) =
|
2568 2617 | ConnectionState::bounded(connection_info(1), owning_lease);
|
2569 - | OriginCell::install_idle_h1(&connection_cell, owned_connection, H1Sender::test(11));
|
2618 + | OriginCell::insert_idle_h1(&connection_cell, owned_connection, H1Sender::test(11));
|
2570 2619 | let requesting_lease = OriginAdmission::lease_for_test(&admission);
|
2571 2620 | let (requesting_connection, _requesting_physical) =
|
2572 2621 | ConnectionState::bounded(connection_info(2), requesting_lease);
|
2573 - | let local = OriginCell::install_selected_h1(
|
2622 + | let local = OriginCell::insert_selected_h1(
|
2574 2623 | &requesting_cell,
|
2575 2624 | requesting_connection,
|
2576 2625 | H1Sender::test(22),
|
2577 2626 | );
|
2578 2627 | let (waiter, demand) =
|
2579 2628 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2580 - | let install = OriginAdmission::publish_action_without_driving(
|
2629 + | let install = OriginAdmission::submit_action_without_running(
|
2581 2630 | &admission,
|
2582 2631 | requesting_cell.id().partition(),
|
2583 2632 | demand,
|
2584 2633 | )
|
2585 - | .expect("peer demand did not prepare a reuse operation");
|
2634 + | .expect("peer demand did not prepare an H1 match");
|
2586 2635 | let delivery = install
|
2587 - | .drive_once_for_test()
|
2588 - | .expect("reuse operation installation did not prepare a delivery");
|
2636 + | .run_once_for_test()
|
2637 + | .expect("H1 supplier reservation did not prepare a delivery");
|
2589 2638 |
|
2590 2639 | let delivering = loom::thread::spawn(move || {
|
2591 - | OriginAdmission::drive(Some(delivery));
|
2640 + | OriginAdmission::run_action_chain(Some(delivery));
|
2592 2641 | });
|
2593 2642 | let returning = loom::thread::spawn(move || drop(local));
|
2594 2643 | delivering.join().unwrap();
|
2595 2644 | returning.join().unwrap();
|
2596 2645 |
|
2597 2646 | let selected = requesting_cell
|
2598 2647 | .take_ready_h1(waiter)
|
2599 2648 | .expect("delivery/return race did not satisfy the requesting cell waiter");
|
2600 2649 | drop(selected);
|
2601 2650 | assert_eq!(
|
2602 2651 | 2,
|
2603 2652 | connection_cell.h1_counts().0 + requesting_cell.h1_counts().0
|
2604 2653 | );
|
2605 2654 | assert_eq!(
|
2606 2655 | 2,
|
2607 2656 | connection_cell.h1_counts().1 + requesting_cell.h1_counts().1
|
2608 2657 | );
|
2609 - | assert!(connection_cell.state.lock().reuse.is_available());
|
2658 + | assert!(connection_cell.state.lock().h1.peer_reservation_available());
|
2610 2659 | admission.clear_modeled_cells_for_test();
|
2611 2660 | });
|
2612 2661 | }
|
2613 2662 |
|
2614 2663 | #[test]
|
2615 2664 | fn borrow_materialization_races_owning_cell_close_without_stranding_request() {
|
2616 2665 | let mut model = loom::model::Builder::new();
|
2617 2666 | model.preemption_bound = Some(2);
|
2618 2667 | model.check(|| {
|
2619 2668 | let (admission, connection_cell, requesting_cell) =
|
2620 2669 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2621 2670 | let lease = OriginAdmission::lease_for_test(&admission);
|
2622 2671 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2623 - | OriginCell::install_idle_h1(&connection_cell, connection.clone(), H1Sender::test(11));
|
2672 + | OriginCell::insert_idle_h1(&connection_cell, connection.clone(), H1Sender::test(11));
|
2624 2673 | let close = H1CloseHandle::new(&connection_cell, &connection);
|
2625 2674 | let (waiter, demand) =
|
2626 2675 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2627 - | let install = OriginAdmission::publish_action_without_driving(
|
2676 + | let install = OriginAdmission::submit_action_without_running(
|
2628 2677 | &admission,
|
2629 2678 | requesting_cell.id().partition(),
|
2630 2679 | demand,
|
2631 2680 | )
|
2632 - | .expect("peer demand did not prepare a reuse operation");
|
2681 + | .expect("peer demand did not prepare an H1 match");
|
2633 2682 | let delivery = install
|
2634 - | .drive_once_for_test()
|
2635 - | .expect("reuse operation installation did not prepare a delivery");
|
2683 + | .run_once_for_test()
|
2684 + | .expect("H1 supplier reservation did not prepare a delivery");
|
2636 2685 |
|
2637 2686 | let delivering = loom::thread::spawn(move || {
|
2638 - | OriginAdmission::drive(Some(delivery));
|
2687 + | OriginAdmission::run_action_chain(Some(delivery));
|
2639 2688 | });
|
2640 2689 | let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2641 2690 | delivering.join().unwrap();
|
2642 2691 | closing.join().unwrap();
|
2643 2692 |
|
2644 2693 | let event = requesting_cell
|
2645 2694 | .take_ready_event(waiter)
|
2646 2695 | .expect("borrow/close race stranded its requesting cell waiter");
|
2647 2696 | match event {
|
2648 - | AcquisitionEvent::Complete(AcquisitionResult::H1(selection)) => {
|
2697 + | AcquisitionStep::Resolved(AcquisitionOutcome::H1(selection)) => {
|
2649 2698 | drop(selection);
|
2650 2699 | }
|
2651 - | AcquisitionEvent::Establish(permit) => {
|
2700 + | AcquisitionStep::StartEstablishment(permit) => {
|
2652 2701 | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
2653 2702 | drop(permit);
|
2654 2703 | }
|
2655 - | AcquisitionEvent::Complete(AcquisitionResult::Failed(error)) => {
|
2704 + | AcquisitionStep::Resolved(AcquisitionOutcome::Failed(error)) => {
|
2656 2705 | panic!("unexpected establishment failure: {error}")
|
2657 2706 | }
|
2658 - | AcquisitionEvent::Complete(AcquisitionResult::Reacquire) => {
|
2707 + | AcquisitionStep::Resolved(AcquisitionOutcome::RetryAcquisition) => {
|
2659 2708 | panic!("HTTP/1 reuse model requested reacquisition")
|
2660 2709 | }
|
2661 - | AcquisitionEvent::Complete(AcquisitionResult::H2(_)) => {
|
2710 + | AcquisitionStep::Resolved(AcquisitionOutcome::H2(_)) => {
|
2662 2711 | panic!("HTTP/1 reuse model received an HTTP/2 activation")
|
2663 2712 | }
|
2664 2713 | }
|
2665 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2714 + | assert_eq!(0, requesting_cell.probe().retained);
|
2666 2715 | assert_eq!(1, admission.available_capacity_for_test());
|
2667 2716 | admission.clear_modeled_cells_for_test();
|
2668 2717 | });
|
2669 2718 | }
|
2670 2719 |
|
2671 2720 | #[test]
|
2672 2721 | fn reclaim_and_connection_close_release_exactly_one_capacity_slot() {
|
2673 2722 | let mut model = loom::model::Builder::new();
|
2674 2723 | model.preemption_bound = Some(2);
|
2675 2724 | model.check(|| {
|
2676 2725 | let (admission, connection_cell, requesting_cell) = bounded_peer_cells(
|
2677 2726 | EligibilityGroup::Partition(PartitionId::from_index(1)),
|
2678 2727 | EligibilityGroup::Partition(PartitionId::from_index(2)),
|
2679 2728 | );
|
2680 2729 | let lease = OriginAdmission::lease_for_test(&admission);
|
2681 2730 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2682 - | let returning = OriginCell::install_selected_h1(
|
2731 + | let returning = OriginCell::insert_selected_h1(
|
2683 2732 | &connection_cell,
|
2684 2733 | connection.clone(),
|
2685 2734 | H1Sender::test(11),
|
2686 2735 | );
|
2687 2736 | let close = H1CloseHandle::new(&connection_cell, &connection);
|
2688 2737 | let waiter =
|
2689 2738 | OriginCell::register_waiter(&requesting_cell, ProtocolRequirement::H1Compatible);
|
2690 2739 | let local_waiter =
|
2691 2740 | OriginCell::register_waiter(&connection_cell, ProtocolRequirement::H1Compatible);
|
2692 2741 |
|
2693 2742 | let returning = loom::thread::spawn(move || drop(returning));
|
2694 2743 | let closing = loom::thread::spawn(move || close.close(CloseReason::Poisoned));
|
2695 2744 | returning.join().unwrap();
|
2696 2745 | let close_won = closing.join().unwrap();
|
2697 2746 |
|
2698 2747 | assert_eq!(
|
2699 2748 | !close_won,
|
2700 - | connection_cell.state.lock().reuse.local_turn_owed(),
|
2749 + | connection_cell.state.lock().h1.local_turn_owed(),
|
2701 2750 | "only a completed reclaim may earn the connection-owning cell a local fairness turn"
|
2702 2751 | );
|
2703 2752 |
|
2704 2753 | let replacement = OriginCell::take_ready_lease(&requesting_cell, waiter)
|
2705 2754 | .expect("close/reclaim race did not deliver released capacity");
|
2706 2755 | assert_eq!((0, 0), connection_cell.h1_counts());
|
2707 2756 | assert_eq!(0, admission.available_capacity_for_test());
|
2708 2757 | assert!(OriginCell::cancel_waiter(&connection_cell, local_waiter));
|
2709 2758 | drop(replacement);
|
2710 2759 | assert_eq!(1, admission.available_capacity_for_test());
|
2711 2760 | admission.clear_modeled_cells_for_test();
|
2712 2761 | });
|
2713 2762 | }
|
2714 2763 |
|
2715 2764 | #[test]
|
2716 2765 | fn delivery_and_cancellation_race_refunnels_capacity() {
|
2717 2766 | loom::model(|| {
|
2718 2767 | let (admission, cell) = bounded_cell();
|
2719 2768 | let (first, demand) =
|
2720 2769 | cell.register_waiter_without_publish(ProtocolRequirement::H1Compatible);
|
2721 2770 | let mut delivery =
|
2722 - | OriginAdmission::publish_without_driving(&admission, cell.id().partition(), demand)
|
2771 + | OriginAdmission::submit_without_running(&admission, cell.id().partition(), demand)
|
2723 2772 | .unwrap();
|
2724 2773 | assert!(
|
2725 - | delivery.materialize_for_test(),
|
2774 + | delivery.resolve_payload_for_test(),
|
2726 2775 | "capacity delivery did not materialize"
|
2727 2776 | );
|
2728 2777 |
|
2729 2778 | let delivery_cell = cell.clone();
|
2730 2779 | let deliver = loom::thread::spawn(move || {
|
2731 2780 | drop(OriginCell::receive_delivery(&delivery_cell, delivery));
|
2732 2781 | });
|
2733 2782 | let cancel_cell = cell.clone();
|
2734 2783 | let cancel =
|
2735 2784 | loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, first));
|
2736 2785 | deliver.join().unwrap();
|
2737 2786 | cancel.join().unwrap();
|
2738 2787 |
|
2739 2788 | assert_eq!(1, admission.available_capacity_for_test());
|
2740 2789 | admission.clear_modeled_cells_for_test();
|
2741 2790 | });
|
2742 2791 | }
|
2743 2792 |
|
2744 2793 | #[test]
|
2745 2794 | fn h2_route_installation_races_requesting_cell_cancellation() {
|
2746 2795 | let mut model = loom::model::Builder::new();
|
2747 2796 | model.preemption_bound = Some(2);
|
2748 2797 | model.check(|| {
|
2749 2798 | let (admission, connection_cell, requesting_cell) =
|
2750 2799 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2751 2800 | let lease = OriginAdmission::lease_for_test(&admission);
|
2752 2801 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2753 2802 | let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
|
2754 2803 | let (waiter, demand) =
|
2755 2804 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2756 2805 | let route = h2::H2Route::new(&connection_cell, generation);
|
2757 2806 |
|
2758 2807 | let install_cell = requesting_cell.clone();
|
2759 2808 | let installing = loom::thread::spawn(move || {
|
2760 - | OriginCell::install_h2_route(
|
2809 + | OriginCell::attach_h2_route(
|
2761 2810 | &install_cell,
|
2762 2811 | route,
|
2763 2812 | &EligibilityGroup::Pool,
|
2764 2813 | demand.id_for_test(),
|
2765 2814 | )
|
2766 2815 | });
|
2767 2816 | let cancel_cell = requesting_cell.clone();
|
2768 2817 | let cancelling =
|
2769 2818 | loom::thread::spawn(move || OriginCell::cancel_waiter(&cancel_cell, waiter));
|
2770 2819 | let _installed = installing.join().unwrap();
|
2771 2820 | assert!(cancelling.join().unwrap());
|
2772 2821 |
|
2773 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2822 + | assert_eq!(0, requesting_cell.probe().retained);
|
2774 2823 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2775 2824 | assert_eq!(0, admission.available_capacity_for_test());
|
2776 2825 | assert!(OriginCell::close_h2(
|
2777 2826 | &connection_cell,
|
2778 2827 | generation,
|
2779 2828 | CloseReason::PoolDropped,
|
2780 2829 | ));
|
2781 2830 | assert_eq!(1, admission.available_capacity_for_test());
|
2782 2831 | admission.clear_modeled_cells_for_test();
|
2783 2832 | });
|
2784 2833 | }
|
2785 2834 |
|
2786 2835 | #[test]
|
2787 - | fn h2_publication_acknowledgement_races_generation_close_and_route_service() {
|
2836 + | fn h2_route_acknowledgement_races_generation_close_and_route_service() {
|
2788 2837 | let mut model = loom::model::Builder::new();
|
2789 2838 | model.preemption_bound = Some(3);
|
2790 2839 | model.check(|| {
|
2791 2840 | let (admission, connection_cell, requesting_cell) =
|
2792 2841 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2793 2842 | let lease = OriginAdmission::lease_for_test(&admission);
|
2794 2843 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2795 2844 | let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
|
2796 2845 | let (waiter, demand) =
|
2797 2846 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2798 - | let action = OriginAdmission::publish_action_without_driving(
|
2847 + | let action = OriginAdmission::submit_action_without_running(
|
2799 2848 | &admission,
|
2800 2849 | requesting_cell.id().partition(),
|
2801 2850 | demand,
|
2802 2851 | )
|
2803 - | .expect("peer demand did not prepare publication");
|
2852 + | .expect("peer demand did not prepare a route");
|
2804 2853 |
|
2805 2854 | let publishing = loom::thread::spawn(move || {
|
2806 - | OriginAdmission::drive(Some(action));
|
2855 + | OriginAdmission::run_action_chain(Some(action));
|
2807 2856 | });
|
2808 2857 | let closing_cell = connection_cell.clone();
|
2809 2858 | let closing = loom::thread::spawn(move || {
|
2810 2859 | OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
|
2811 2860 | });
|
2812 2861 | let servicing_cell = requesting_cell.clone();
|
2813 2862 | let servicing = loom::thread::spawn(move || {
|
2814 - | OriginCell::service_peer_h2_waiters(&servicing_cell);
|
2863 + | OriginCell::offer_peer_h2(&servicing_cell);
|
2815 2864 | });
|
2816 2865 |
|
2817 2866 | publishing.join().unwrap();
|
2818 2867 | assert!(closing.join().unwrap());
|
2819 2868 | servicing.join().unwrap();
|
2820 2869 |
|
2821 2870 | assert!(
|
2822 2871 | OriginCell::cancel_waiter(&requesting_cell, waiter),
|
2823 - | "publication acknowledgement lost the live demand"
|
2872 + | "route acknowledgement lost the live demand"
|
2824 2873 | );
|
2825 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2874 + | assert_eq!(0, requesting_cell.probe().retained);
|
2826 2875 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2827 2876 | assert_eq!(1, admission.available_capacity_for_test());
|
2828 2877 | admission.clear_modeled_cells_for_test();
|
2829 2878 | });
|
2830 2879 | }
|
2831 2880 |
|
2832 2881 | #[test]
|
2833 - | fn h2_publication_close_race_preserves_capacity_ownership() {
|
2882 + | fn h2_route_close_race_preserves_capacity_ownership() {
|
2834 2883 | let mut model = loom::model::Builder::new();
|
2835 2884 | model.preemption_bound = Some(2);
|
2836 2885 | model.check(|| {
|
2837 2886 | let (admission, connection_cell, requesting_cell) =
|
2838 2887 | bounded_peer_cells(EligibilityGroup::Pool, EligibilityGroup::Pool);
|
2839 2888 | let lease = OriginAdmission::lease_for_test(&admission);
|
2840 2889 | let (connection, _physical) = ConnectionState::bounded(connection_info(1), lease);
|
2841 2890 | let generation = OriginCell::install_h2_for_test(&connection_cell, connection, 1, None);
|
2842 2891 | let (waiter, demand) =
|
2843 2892 | requesting_cell.register_waiter_without_publish(ProtocolRequirement::H2Required);
|
2844 - | let action = OriginAdmission::publish_action_without_driving(
|
2893 + | let action = OriginAdmission::submit_action_without_running(
|
2845 2894 | &admission,
|
2846 2895 | requesting_cell.id().partition(),
|
2847 2896 | demand,
|
2848 2897 | )
|
2849 - | .expect("peer demand did not prepare publication");
|
2898 + | .expect("peer demand did not prepare a route");
|
2850 2899 |
|
2851 2900 | let publishing = loom::thread::spawn(move || {
|
2852 - | OriginAdmission::drive(Some(action));
|
2901 + | OriginAdmission::run_action_chain(Some(action));
|
2853 2902 | });
|
2854 2903 | let closing_cell = connection_cell.clone();
|
2855 2904 | let closing = loom::thread::spawn(move || {
|
2856 2905 | OriginCell::close_h2(&closing_cell, generation, CloseReason::Poisoned)
|
2857 2906 | });
|
2858 2907 | publishing.join().unwrap();
|
2859 2908 | assert!(closing.join().unwrap());
|
2860 2909 |
|
2861 2910 | assert!(OriginCell::cancel_waiter(&requesting_cell, waiter));
|
2862 - | assert_eq!(0, requesting_cell.snapshot().retained);
|
2911 + | assert_eq!(0, requesting_cell.probe().retained);
|
2863 2912 | assert_eq!(0, admission.ordered_demand_count_for_test());
|
2864 2913 | assert_eq!(1, admission.available_capacity_for_test());
|
2865 2914 | admission.clear_modeled_cells_for_test();
|
2866 2915 | });
|
2867 2916 | }
|
2868 2917 | }
|