1 1 | /*
|
2 2 | * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
|
3 3 | * SPDX-License-Identifier: Apache-2.0
|
4 4 | */
|
5 5 |
|
6 6 | //! Versioned cross-cell demand scheduling for one bounded origin.
|
7 7 | //!
|
8 - | //! Each retained cell publishes a complete demand snapshot. Active snapshots
|
9 - | //! occupy one origin-wide FIFO, and a delivery remains at the head as a fence
|
10 - | //! until its requesting cell acknowledges ownership or rejection.
|
8 + | //! Each cell publishes at most one active demand: its acquisition queue's
|
9 + | //! current head. That demand may be satisfied either by origin-wide connection
|
10 + | //! capacity or by an existing HTTP/2 connection in its eligibility group.
|
11 + | //! These resources require different orderings.
|
12 + | //!
|
13 + | //! For example, the origin order may contain `A(group X), B(group Y)` while
|
14 + | //! only group Y has a reusable HTTP/2 connection. A remains first for the next
|
15 + | //! available connection permit, while the group-Y connection can serve B
|
16 + | //! without consuming that permit or delaying A. The group order finds B
|
17 + | //! directly instead of scanning every partition.
|
18 + | //!
|
19 + | //! The same demand occupies both orders because capacity and HTTP/2 publication
|
20 + | //! can race to satisfy it. Reserving either position retains a fence in both
|
21 + | //! orders until the requesting cell accepts or rejects the handoff. The other
|
22 + | //! resource therefore cannot serve the same demand while the first handoff is
|
23 + | //! running outside the admission lock.
|
11 24 |
|
12 25 | use super::{
|
13 26 | DeliveryAckResult, DeliveryId, DemandId, DemandSnapshot, DemandState, IntrusiveLinks,
|
14 27 | IntrusiveOrder, Permit, ProtocolRequirement,
|
15 28 | };
|
16 29 | use crate::client::pool::partition::{EligibilityGroup, PartitionId};
|
17 30 | use std::collections::HashMap;
|
18 31 |
|
19 - | /// Cross-cell demand records and their origin-wide scheduling order.
|
32 + | /// Cross-cell demand records and their scheduling orders.
|
20 33 | ///
|
21 34 | /// At every completed transition:
|
22 35 | ///
|
23 36 | /// - `records` owns the newest snapshot for every retained cell.
|
24 - | /// - `IntrusiveOrder::Active` contains exactly the `Queued` and `Delivering`
|
25 - | /// records.
|
26 - | /// - Queue links live inside those ordered residence variants.
|
27 - | /// - A `Delivering` record remains at the head as a scheduling fence.
|
37 + | /// - the origin order and the record's eligibility-group order contain exactly
|
38 + | /// the `Queued` and `Delivering` records;
|
39 + | /// - links for both views live inside those ordered residence variants;
|
40 + | /// - an origin delivery fences the origin head; and
|
41 + | /// - an HTTP/2 publication fences its eligibility-group head.
|
28 42 | ///
|
29 43 | /// Admission coordinates capacity extraction with this schedule while holding
|
30 44 | /// the same origin lock.
|
31 45 | #[derive(Debug, Default)]
|
32 46 | pub(super) struct DemandSchedule {
|
33 47 | /// Latest demand and scheduling residence for each retained cell.
|
34 48 | records: HashMap<PartitionId, DemandRecord>,
|
35 - | /// Origin-wide order, including an outstanding delivery fence.
|
36 - | order: IntrusiveOrder<PartitionId>,
|
49 + | /// Origin-wide order used by one-to-one capacity and HTTP/1 reuse.
|
50 + | origin_order: IntrusiveOrder<PartitionId>,
|
51 + | /// Reusable-HTTP/2 demand order for each connection-reuse group.
|
52 + | group_orders: HashMap<EligibilityGroup, IntrusiveOrder<PartitionId>>,
|
37 53 | }
|
38 54 |
|
39 55 | /// Complete origin-order head used to choose one HTTP/1 reuse action.
|
40 56 | #[derive(Clone, Debug)]
|
41 57 | pub(super) struct QueuedDemand {
|
42 58 | /// Cell whose oldest waiter owns this demand generation.
|
43 59 | pub(super) requesting_partition: PartitionId,
|
44 60 | /// Demand identity revalidated at each crossing.
|
45 61 | pub(super) demand: DemandId,
|
46 62 | /// Protocol capability required by the requesting cell waiter.
|
47 63 | pub(super) requirement: ProtocolRequirement,
|
48 64 | /// Connection-owning cells whose H1 senders may satisfy this demand.
|
49 65 | pub(super) eligibility_group: EligibilityGroup,
|
50 66 | }
|
51 67 |
|
52 68 | /// Latest snapshot and scheduling residence for one stable cell.
|
53 69 | #[derive(Debug)]
|
54 70 | struct DemandRecord {
|
55 71 | /// Newest complete publication observed for the cell.
|
56 72 | pub(super) latest: DemandSnapshot,
|
73 + | /// Stable group retained while an inactive replacement crosses a fence.
|
74 + | group: Option<EligibilityGroup>,
|
57 75 | /// Scheduling residence, including links while ordered.
|
58 76 | residence: DemandResidence,
|
59 77 | }
|
60 78 |
|
79 + | /// Links retained by one demand in both scheduling views.
|
80 + | #[derive(Clone, Debug)]
|
81 + | struct DemandLinks {
|
82 + | /// Position in origin-wide capacity and HTTP/1 order.
|
83 + | origin: IntrusiveLinks<PartitionId>,
|
84 + | /// Position in the demand's all-protocol eligibility-group order.
|
85 + | group: IntrusiveLinks<PartitionId>,
|
86 + | }
|
87 + |
|
88 + | /// Order whose head is fenced by one delivery crossing.
|
89 + | #[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
90 + | enum DeliveryView {
|
91 + | /// Capacity or HTTP/1 delivery selected from origin order.
|
92 + | Origin,
|
93 + | /// HTTP/2 publication selected from eligibility-group order.
|
94 + | Group,
|
95 + | }
|
96 + |
|
97 + | /// Selects one of a demand residence's intrusive link sets.
|
98 + | #[derive(Clone, Copy)]
|
99 + | enum DemandOrderView {
|
100 + | /// Origin-wide capacity and HTTP/1 order.
|
101 + | Origin,
|
102 + | /// Eligibility-group all-protocol order.
|
103 + | Group,
|
104 + | }
|
105 + |
|
61 106 | /// Admission residence for one partition's latest demand.
|
62 107 | ///
|
63 108 | /// ```text
|
64 109 | /// Idle -- active publication ------------------------------> Queued
|
65 110 | /// Queued -- reserve head ----------------------------------> Delivering
|
66 111 | /// Queued -- inactive or replacement publication ----------> Idle
|
67 112 | /// Delivering -- retry unchanged demand --------------------> Queued
|
68 113 | /// Delivering -- accept, reject, or replacement retry ------> Idle
|
69 114 | /// Idle -- active successor remains ------------------------> Queued
|
70 115 | /// ```
|
71 116 | #[derive(Clone, Debug)]
|
72 117 | enum DemandResidence {
|
73 118 | /// The cell has no demand represented in scheduling.
|
74 119 | Idle,
|
75 120 | /// The demand is waiting in origin order.
|
76 121 | Queued {
|
77 122 | /// Demand generation represented by this residence.
|
78 123 | demand: DemandId,
|
79 - | /// Origin-wide scheduling links and arrival sequence.
|
80 - | links: IntrusiveLinks<PartitionId>,
|
124 + | /// Origin-wide and eligibility-group scheduling links.
|
125 + | links: DemandLinks,
|
81 126 | },
|
82 - | /// One delivery guard owns capacity for this demand.
|
127 + | /// One delivery or publication guard fences this demand.
|
83 128 | Delivering {
|
84 - | /// Demand generation fenced at the order head.
|
129 + | /// Demand generation fenced at one scheduling head.
|
85 130 | demand: DemandId,
|
131 + | /// Snapshot version current when admission created the fence.
|
132 + | version: super::SnapshotVersion,
|
86 133 | /// Delivery allowed to complete this fence.
|
87 134 | delivery: DeliveryId,
|
88 - | /// Origin-wide scheduling links retained until acknowledgement.
|
89 - | links: IntrusiveLinks<PartitionId>,
|
135 + | /// Scheduling view whose head owns this crossing.
|
136 + | #[cfg_attr(
|
137 + | not(debug_assertions),
|
138 + | allow(
|
139 + | dead_code,
|
140 + | reason = "the scheduling view is retained for debug invariant checks"
|
141 + | )
|
142 + | )]
|
143 + | view: DeliveryView,
|
144 + | /// Both scheduling positions retained until acknowledgement.
|
145 + | links: DemandLinks,
|
90 146 | },
|
91 147 | }
|
92 148 |
|
93 149 | impl DemandResidence {
|
94 - | #[cfg(debug_assertions)]
|
95 - | /// Borrows order links while this record is queued or delivering.
|
96 - | fn links(&self) -> Option<&IntrusiveLinks<PartitionId>> {
|
97 - | match self {
|
98 - | Self::Idle => None,
|
99 - | Self::Queued { links, .. } | Self::Delivering { links, .. } => Some(links),
|
150 + | #[cfg(any(debug_assertions, test))]
|
151 + | /// Borrows one view's links while this record is ordered.
|
152 + | fn links(&self, view: DemandOrderView) -> Option<&IntrusiveLinks<PartitionId>> {
|
153 + | let links = match self {
|
154 + | Self::Idle => return None,
|
155 + | Self::Queued { links, .. } | Self::Delivering { links, .. } => links,
|
156 + | };
|
157 + | match view {
|
158 + | DemandOrderView::Origin => Some(&links.origin),
|
159 + | DemandOrderView::Group => Some(&links.group),
|
100 160 | }
|
101 161 | }
|
102 162 |
|
103 - | /// Mutably borrows links from a record known to be ordered.
|
163 + | /// Mutably borrows one view's links from an ordered record.
|
104 164 | ///
|
105 165 | /// # Panics
|
106 166 | ///
|
107 167 | /// Panics when called for an idle record.
|
108 - | fn links_mut(&mut self) -> &mut IntrusiveLinks<PartitionId> {
|
109 - | match self {
|
110 - | Self::Idle => panic!("idle demand has no order links"),
|
168 + | fn links_mut(&mut self, view: DemandOrderView) -> &mut IntrusiveLinks<PartitionId> {
|
169 + | let links = match self {
|
170 + | Self::Idle => panic!("idle demand has no scheduling links"),
|
111 171 | Self::Queued { links, .. } | Self::Delivering { links, .. } => links,
|
172 + | };
|
173 + | match view {
|
174 + | DemandOrderView::Origin => &mut links.origin,
|
175 + | DemandOrderView::Group => &mut links.group,
|
112 176 | }
|
113 177 | }
|
114 178 |
|
115 - | /// Detaches links while moving a record out of origin-wide order.
|
116 - | fn into_links(self) -> Option<IntrusiveLinks<PartitionId>> {
|
179 + | /// Detaches both link sets while moving a record out of scheduling.
|
180 + | fn into_links(self) -> Option<DemandLinks> {
|
117 181 | match self {
|
118 182 | Self::Idle => None,
|
119 183 | Self::Queued { links, .. } | Self::Delivering { links, .. } => Some(links),
|
120 184 | }
|
121 185 | }
|
122 186 | }
|
123 187 |
|
124 188 | /// Demand reserved at the order head for one capacity delivery.
|
125 189 | pub(super) struct ScheduledDemand {
|
126 190 | /// Selected destination cell.
|
127 191 | pub(super) requesting_partition: PartitionId,
|
128 192 | /// Demand identity current when the crossing was prepared.
|
129 193 | pub(super) demand: DemandId,
|
130 194 | }
|
131 195 |
|
132 196 | /// Connection capacity extracted under admission lock before crossing to a cell.
|
133 197 | #[derive(Debug)]
|
134 198 | pub(super) struct PreparedCapacityDelivery {
|
135 199 | /// Permit transferred into the delivery guard.
|
136 200 | pub(super) permit: Permit,
|
137 201 | /// Fence identity allocated for this crossing.
|
138 202 | pub(super) delivery: DeliveryId,
|
139 203 | /// Selected destination cell.
|
140 204 | pub(super) requesting_partition: PartitionId,
|
141 205 | /// Demand identity current when the crossing was prepared.
|
142 206 | pub(super) demand: DemandId,
|
143 207 | }
|
144 208 |
|
145 209 | impl DemandSchedule {
|
146 210 | /// Applies a complete snapshot and updates its cell's scheduling residence.
|
147 211 | pub(super) fn publish(&mut self, requesting_partition: PartitionId, snapshot: DemandSnapshot) {
|
148 212 | if let Some(current) = self.records.get(&requesting_partition) {
|
149 213 | if !snapshot.is_newer_than(¤t.latest) {
|
150 214 | return;
|
151 215 | }
|
152 216 | } else {
|
153 217 | self.records.insert(
|
154 218 | requesting_partition,
|
155 219 | DemandRecord {
|
156 220 | latest: snapshot.clone(),
|
221 + | group: None,
|
157 222 | residence: DemandResidence::Idle,
|
158 223 | },
|
159 224 | );
|
160 225 | }
|
161 226 |
|
162 227 | let should_remove = matches!(
|
163 228 | &self
|
164 229 | .records
|
165 230 | .get(&requesting_partition)
|
166 231 | .expect("published demand record disappeared")
|
167 232 | .residence,
|
168 233 | DemandResidence::Queued { demand, .. }
|
169 234 | if *demand != snapshot.id || !snapshot.is_active()
|
170 235 | );
|
171 236 | if should_remove {
|
172 237 | self.remove_from_order(&requesting_partition);
|
173 238 | }
|
174 239 |
|
240 + | if let DemandState::Active {
|
241 + | eligibility_group, ..
|
242 + | } = &snapshot.state
|
243 + | {
|
244 + | self.records
|
245 + | .get_mut(&requesting_partition)
|
246 + | .expect("published demand record disappeared")
|
247 + | .group = Some(eligibility_group.clone());
|
248 + | }
|
175 249 | self.records
|
176 250 | .get_mut(&requesting_partition)
|
177 251 | .expect("published demand record disappeared")
|
178 252 | .latest = snapshot;
|
179 253 |
|
180 254 | let record = self
|
181 255 | .records
|
182 256 | .get(&requesting_partition)
|
183 257 | .expect("published demand record disappeared");
|
184 258 | if record.latest.is_active() && matches!(&record.residence, DemandResidence::Idle) {
|
185 259 | self.enqueue(requesting_partition);
|
186 260 | }
|
187 261 | self.assert_consistent();
|
188 262 | }
|
189 263 |
|
190 - | /// Appends an idle active demand to the origin-wide order.
|
264 + | /// Appends an idle active demand to both scheduling orders.
|
191 265 | fn enqueue(&mut self, requesting_partition: PartitionId) {
|
192 - | let links = self.order.push_back(requesting_partition);
|
193 - | if let Some(previous) = links.previous {
|
266 + | let group = self
|
267 + | .group_for(&requesting_partition)
|
268 + | .expect("enqueued demand was inactive");
|
269 + | let origin = self.origin_order.push_back(requesting_partition);
|
270 + | let group_links = self
|
271 + | .group_orders
|
272 + | .entry(group)
|
273 + | .or_default()
|
274 + | .push_back(requesting_partition);
|
275 + | if let Some(previous) = origin.previous {
|
194 276 | self.records
|
195 277 | .get_mut(&previous)
|
196 - | .expect("order tail disappeared")
|
278 + | .expect("origin demand tail disappeared")
|
197 279 | .residence
|
198 - | .links_mut()
|
280 + | .links_mut(DemandOrderView::Origin)
|
281 + | .next = Some(requesting_partition);
|
282 + | }
|
283 + | if let Some(previous) = group_links.previous {
|
284 + | self.records
|
285 + | .get_mut(&previous)
|
286 + | .expect("group demand tail disappeared")
|
287 + | .residence
|
288 + | .links_mut(DemandOrderView::Group)
|
199 289 | .next = Some(requesting_partition);
|
200 290 | }
|
201 291 |
|
202 292 | let record = self
|
203 293 | .records
|
204 294 | .get_mut(&requesting_partition)
|
205 295 | .expect("queued demand record disappeared");
|
206 296 | debug_assert!(matches!(record.residence, DemandResidence::Idle));
|
207 297 | let demand = record.latest.id;
|
208 - | record.residence = DemandResidence::Queued { demand, links };
|
298 + | record.residence = DemandResidence::Queued {
|
299 + | demand,
|
300 + | links: DemandLinks {
|
301 + | origin,
|
302 + | group: group_links,
|
303 + | },
|
304 + | };
|
209 305 | }
|
210 306 |
|
211 - | /// Removes an ordered demand and leaves its retained record idle.
|
307 + | /// Removes an ordered demand from both views and leaves its record idle.
|
212 308 | fn remove_from_order(&mut self, requesting_partition: &PartitionId) {
|
309 + | let group = self
|
310 + | .group_for(requesting_partition)
|
311 + | .expect("removed demand had no eligibility group");
|
213 312 | let residence = {
|
214 313 | let record = self
|
215 314 | .records
|
216 315 | .get_mut(requesting_partition)
|
217 316 | .expect("removed demand record disappeared");
|
218 317 | std::mem::replace(&mut record.residence, DemandResidence::Idle)
|
219 318 | };
|
220 319 | let links = residence
|
221 320 | .into_links()
|
222 - | .expect("removed demand had no order links");
|
321 + | .expect("removed demand had no scheduling links");
|
223 322 |
|
224 - | if let Some(previous) = links.previous {
|
323 + | if let Some(previous) = links.origin.previous {
|
225 324 | self.records
|
226 325 | .get_mut(&previous)
|
227 - | .expect("previous demand disappeared")
|
326 + | .expect("previous origin demand disappeared")
|
228 327 | .residence
|
229 - | .links_mut()
|
230 - | .next = links.next;
|
328 + | .links_mut(DemandOrderView::Origin)
|
329 + | .next = links.origin.next;
|
231 330 | }
|
232 - | if let Some(next) = links.next {
|
331 + | if let Some(next) = links.origin.next {
|
233 332 | self.records
|
234 333 | .get_mut(&next)
|
235 - | .expect("next demand disappeared")
|
334 + | .expect("next origin demand disappeared")
|
236 335 | .residence
|
237 - | .links_mut()
|
238 - | .previous = links.previous;
|
336 + | .links_mut(DemandOrderView::Origin)
|
337 + | .previous = links.origin.previous;
|
239 338 | }
|
339 + | self.origin_order
|
340 + | .remove(*requesting_partition, links.origin);
|
240 341 |
|
241 - | self.order.remove(*requesting_partition, links);
|
342 + | if let Some(previous) = links.group.previous {
|
343 + | self.records
|
344 + | .get_mut(&previous)
|
345 + | .expect("previous group demand disappeared")
|
346 + | .residence
|
347 + | .links_mut(DemandOrderView::Group)
|
348 + | .next = links.group.next;
|
349 + | }
|
350 + | if let Some(next) = links.group.next {
|
351 + | self.records
|
352 + | .get_mut(&next)
|
353 + | .expect("next group demand disappeared")
|
354 + | .residence
|
355 + | .links_mut(DemandOrderView::Group)
|
356 + | .previous = links.group.previous;
|
357 + | }
|
358 + | let order = self
|
359 + | .group_orders
|
360 + | .get_mut(&group)
|
361 + | .expect("ordered demand lost its eligibility-group order");
|
362 + | order.remove(*requesting_partition, links.group);
|
363 + | if order.len() == 0 {
|
364 + | self.group_orders.remove(&group);
|
365 + | }
|
242 366 | }
|
243 367 |
|
244 368 | /// Returns whether the head can begin a new one-to-one delivery.
|
245 369 | pub(super) fn head_is_queued(&self) -> bool {
|
246 - | let Some(head) = self.order.head() else {
|
370 + | let Some(head) = self.origin_order.head() else {
|
247 371 | return false;
|
248 372 | };
|
249 373 | matches!(
|
250 374 | &self
|
251 375 | .records
|
252 376 | .get(&head)
|
253 377 | .expect("order head disappeared")
|
254 378 | .residence,
|
255 379 | DemandResidence::Queued { .. }
|
256 380 | )
|
257 381 | }
|
258 382 |
|
259 383 | /// Returns the complete origin-order head when it may begin reuse.
|
260 384 | pub(super) fn queued_head(&self) -> Option<QueuedDemand> {
|
261 - | let head = self.order.head()?;
|
385 + | let head = self.origin_order.head()?;
|
262 386 | let record = self.records.get(&head).expect("order head disappeared");
|
263 387 | let DemandResidence::Queued { demand, .. } = &record.residence else {
|
264 388 | return None;
|
265 389 | };
|
266 390 | let DemandState::Active {
|
267 391 | head: requirement,
|
268 392 | eligibility_group,
|
269 393 | } = &record.latest.state
|
270 394 | else {
|
271 395 | unreachable!("queued demand became inactive");
|
272 396 | };
|
273 397 | Some(QueuedDemand {
|
274 398 | requesting_partition: head,
|
275 399 | demand: *demand,
|
276 400 | requirement: *requirement,
|
277 401 | eligibility_group: eligibility_group.clone(),
|
278 402 | })
|
279 403 | }
|
280 404 |
|
405 + | /// Returns one eligibility-group head when it may receive H2 visibility.
|
406 + | pub(super) fn queued_group_head(&self, group: &EligibilityGroup) -> Option<QueuedDemand> {
|
407 + | let head = self.group_orders.get(group)?.head()?;
|
408 + | let record = self
|
409 + | .records
|
410 + | .get(&head)
|
411 + | .expect("group demand head disappeared");
|
412 + | let DemandResidence::Queued { demand, .. } = &record.residence else {
|
413 + | return None;
|
414 + | };
|
415 + | let DemandState::Active {
|
416 + | head: requirement,
|
417 + | eligibility_group,
|
418 + | } = &record.latest.state
|
419 + | else {
|
420 + | unreachable!("queued group demand became inactive");
|
421 + | };
|
422 + | debug_assert_eq!(eligibility_group, group);
|
423 + | Some(QueuedDemand {
|
424 + | requesting_partition: head,
|
425 + | demand: *demand,
|
426 + | requirement: *requirement,
|
427 + | eligibility_group: eligibility_group.clone(),
|
428 + | })
|
429 + | }
|
430 + |
|
431 + | /// Returns the latest eligibility group retained for one cell.
|
432 + | pub(super) fn group_for(&self, requesting_partition: &PartitionId) -> Option<EligibilityGroup> {
|
433 + | self.records
|
434 + | .get(requesting_partition)
|
435 + | .and_then(|record| record.group.clone())
|
436 + | }
|
437 + |
|
281 438 | /// Returns whether `requesting_partition` still has this demand queued for a new action.
|
282 439 | pub(super) fn is_current_queued(
|
283 440 | &self,
|
284 441 | requesting_partition: &PartitionId,
|
285 442 | demand: DemandId,
|
286 443 | ) -> bool {
|
287 444 | self.records
|
288 445 | .get(requesting_partition)
|
289 446 | .is_some_and(|record| {
|
290 447 | record.latest.id == demand
|
291 448 | && record.latest.is_active()
|
292 449 | && matches!(
|
293 450 | record.residence,
|
294 451 | DemandResidence::Queued {
|
295 452 | demand: current,
|
296 453 | ..
|
297 454 | } if current == demand
|
298 455 | )
|
299 456 | })
|
300 457 | }
|
301 458 |
|
302 459 | /// Fences one reuse operation's demand while it remains the origin-order head.
|
303 460 | pub(super) fn reserve_reuse_demand(
|
304 461 | &mut self,
|
305 462 | requesting_partition: &PartitionId,
|
306 463 | demand: DemandId,
|
307 464 | delivery: DeliveryId,
|
308 465 | ) -> Option<ScheduledDemand> {
|
309 466 | if !self.is_current_queued(requesting_partition, demand) {
|
310 467 | return None;
|
311 468 | }
|
312 - | let head = self.order.head()?;
|
469 + | let head = self.origin_order.head()?;
|
313 470 | if head != *requesting_partition {
|
314 471 | return None;
|
315 472 | }
|
316 - | self.reserve_head(delivery)
|
473 + | self.reserve_origin_head(delivery)
|
317 474 | }
|
318 475 |
|
319 - | /// Changes the queued head into a delivery fence at the same order position.
|
320 - | pub(super) fn reserve_head(&mut self, delivery: DeliveryId) -> Option<ScheduledDemand> {
|
321 - | let head = self.order.head()?;
|
322 - | let requesting_partition = head;
|
476 + | /// Fences one eligibility-group head for an H2 publication.
|
477 + | pub(super) fn reserve_group_head(
|
478 + | &mut self,
|
479 + | group: &EligibilityGroup,
|
480 + | requesting_partition: &PartitionId,
|
481 + | demand: DemandId,
|
482 + | delivery: DeliveryId,
|
483 + | ) -> Option<ScheduledDemand> {
|
484 + | if !self.is_current_queued(requesting_partition, demand)
|
485 + | || self.group_orders.get(group)?.head() != Some(*requesting_partition)
|
486 + | {
|
487 + | return None;
|
488 + | }
|
489 + | self.reserve(*requesting_partition, delivery, DeliveryView::Group)
|
490 + | }
|
491 + |
|
492 + | /// Changes the origin head into a delivery fence at the same positions.
|
493 + | pub(super) fn reserve_origin_head(&mut self, delivery: DeliveryId) -> Option<ScheduledDemand> {
|
494 + | let head = self.origin_order.head()?;
|
495 + | self.reserve(head, delivery, DeliveryView::Origin)
|
496 + | }
|
497 + |
|
498 + | /// Changes one queued demand into a delivery fence.
|
499 + | fn reserve(
|
500 + | &mut self,
|
501 + | requesting_partition: PartitionId,
|
502 + | delivery: DeliveryId,
|
503 + | view: DeliveryView,
|
504 + | ) -> Option<ScheduledDemand> {
|
323 505 | let record = self
|
324 506 | .records
|
325 507 | .get_mut(&requesting_partition)
|
326 508 | .expect("order head disappeared");
|
327 509 | let residence = std::mem::replace(&mut record.residence, DemandResidence::Idle);
|
328 510 | match residence {
|
329 511 | DemandResidence::Queued { demand, links } => {
|
330 512 | debug_assert_eq!(record.latest.id, demand);
|
331 513 | debug_assert!(record.latest.is_active());
|
332 514 | record.residence = DemandResidence::Delivering {
|
333 515 | demand,
|
516 + | version: record.latest.version,
|
334 517 | delivery,
|
518 + | view,
|
335 519 | links,
|
336 520 | };
|
337 521 | self.assert_consistent();
|
338 522 | Some(ScheduledDemand {
|
339 523 | requesting_partition,
|
340 524 | demand,
|
341 525 | })
|
342 526 | }
|
343 527 | residence => {
|
344 528 | record.residence = residence;
|
389 573 | delivery: DeliveryId,
|
390 574 | requesting_partition: &PartitionId,
|
391 575 | result: DeliveryAckResult,
|
392 576 | ) {
|
393 577 | let Some(record) = self.records.get(requesting_partition) else {
|
394 578 | return;
|
395 579 | };
|
396 580 | let delivered_demand = match &record.residence {
|
397 581 | DemandResidence::Delivering {
|
398 582 | demand,
|
583 + | version,
|
399 584 | delivery: current,
|
400 585 | ..
|
401 - | } if *current == delivery => *demand,
|
586 + | } if *current == delivery => (*demand, *version),
|
402 587 | _ => return,
|
403 588 | };
|
589 + | let (delivered_demand, delivered_version) = delivered_demand;
|
404 590 |
|
405 591 | match result {
|
406 592 | DeliveryAckResult::RetrySameResidence => {
|
407 593 | let record = self
|
408 594 | .records
|
409 595 | .get(requesting_partition)
|
410 596 | .expect("delivery demand record disappeared");
|
411 597 | if record.latest.id == delivered_demand && record.latest.is_active() {
|
412 598 | let record = self
|
413 599 | .records
|
414 600 | .get_mut(requesting_partition)
|
415 601 | .expect("delivery demand record disappeared");
|
416 602 | let residence = std::mem::replace(&mut record.residence, DemandResidence::Idle);
|
417 603 | let DemandResidence::Delivering {
|
418 604 | demand,
|
605 + | version: _,
|
419 606 | delivery: current,
|
607 + | view: _,
|
420 608 | links,
|
421 609 | } = residence
|
422 610 | else {
|
423 611 | unreachable!("delivery fence disappeared");
|
424 612 | };
|
425 613 | debug_assert_eq!(current, delivery);
|
426 614 | record.residence = DemandResidence::Queued { demand, links };
|
427 615 | self.assert_consistent();
|
428 616 | return;
|
429 617 | }
|
430 618 |
|
431 619 | self.remove_from_order(requesting_partition);
|
432 620 | if self
|
433 621 | .records
|
434 622 | .get(requesting_partition)
|
435 623 | .expect("delivery demand record disappeared")
|
436 624 | .latest
|
437 625 | .is_active()
|
438 626 | {
|
439 627 | self.enqueue(*requesting_partition);
|
440 628 | }
|
441 629 | }
|
442 630 | DeliveryAckResult::Accepted { successor }
|
443 631 | | DeliveryAckResult::Rejected { successor } => {
|
444 632 | self.remove_from_order(requesting_partition);
|
445 633 |
|
446 634 | let install_successor = successor.as_ref().is_some_and(|successor| {
|
447 635 | successor.id > delivered_demand
|
448 636 | && successor.is_newer_than(
|
449 637 | &self
|
450 638 | .records
|
451 639 | .get(requesting_partition)
|
452 640 | .expect("delivery demand record disappeared")
|
453 641 | .latest,
|
454 642 | )
|
455 643 | });
|
456 644 | if install_successor {
|
457 645 | self.records
|
458 646 | .get_mut(requesting_partition)
|
459 647 | .expect("delivery demand record disappeared")
|
460 648 | .latest = successor.expect("validated successor disappeared");
|
461 - | } else if self
|
462 - | .records
|
463 - | .get(requesting_partition)
|
464 - | .expect("delivery demand record disappeared")
|
465 - | .latest
|
466 - | .id
|
467 - | == delivered_demand
|
468 - | {
|
649 + | } else {
|
650 + | let retirement =
|
651 + | DemandSnapshot::inactive(delivered_demand, delivered_version.next());
|
469 652 | let record = self
|
470 653 | .records
|
471 654 | .get_mut(requesting_partition)
|
472 655 | .expect("delivery demand record disappeared");
|
473 - | record.latest =
|
474 - | DemandSnapshot::inactive(delivered_demand, record.latest.version.next());
|
656 + | if retirement.is_newer_than(&record.latest) {
|
657 + | record.latest = retirement;
|
658 + | }
|
475 659 | }
|
476 660 |
|
477 661 | if self
|
478 662 | .records
|
479 663 | .get(requesting_partition)
|
480 664 | .expect("delivery demand record disappeared")
|
481 665 | .latest
|
482 666 | .is_active()
|
483 667 | {
|
484 668 | self.enqueue(*requesting_partition);
|
485 669 | }
|
486 670 | }
|
487 671 | }
|
488 672 | self.assert_consistent();
|
489 673 | }
|
490 674 |
|
491 675 | /// Returns the latest complete snapshot retained for `requesting_partition`.
|
492 676 | #[cfg(test)]
|
493 677 | pub(super) fn latest_for_test(
|
494 678 | &self,
|
495 679 | requesting_partition: &PartitionId,
|
496 680 | ) -> Option<&DemandSnapshot> {
|
497 681 | self.records
|
498 682 | .get(requesting_partition)
|
499 683 | .map(|record| &record.latest)
|
500 684 | }
|
501 685 |
|
502 686 | #[cfg(test)]
|
503 687 | pub(super) fn len(&self) -> usize {
|
504 - | self.order.len()
|
688 + | self.origin_order.len()
|
505 689 | }
|
506 690 |
|
507 691 | #[cfg(test)]
|
508 692 | pub(super) fn queued_len(&self) -> usize {
|
509 693 | self.records
|
510 694 | .values()
|
511 695 | .filter(|record| matches!(&record.residence, DemandResidence::Queued { .. }))
|
512 696 | .count()
|
513 697 | }
|
514 698 |
|
515 699 | #[cfg(test)]
|
516 700 | pub(super) fn delivering_len(&self) -> usize {
|
517 701 | self.records
|
518 702 | .values()
|
519 703 | .filter(|record| matches!(&record.residence, DemandResidence::Delivering { .. }))
|
520 704 | .count()
|
521 705 | }
|
522 706 |
|
523 - | /// Checks record residence, FIFO links, length, and head-fence relationships.
|
707 + | /// Checks residence, link, length, group, and fence relationships.
|
524 708 | fn assert_consistent(&self) {
|
525 - | #[cfg(debug_assertions)]
|
526 - | self.assert_consistent_debug();
|
709 + | #[cfg(any(debug_assertions, test))]
|
710 + | {
|
711 + | if std::thread::panicking() {
|
712 + | return;
|
713 + | }
|
714 + | self.assert_consistent_debug();
|
715 + | }
|
527 716 | }
|
528 717 |
|
529 - | #[cfg(debug_assertions)]
|
718 + | #[cfg(any(debug_assertions, test))]
|
530 719 | fn assert_consistent_debug(&self) {
|
531 720 | let ordered_records = self
|
532 721 | .records
|
533 722 | .values()
|
534 - | .filter(|record| record.residence.links().is_some())
|
723 + | .filter(|record| record.residence.links(DemandOrderView::Origin).is_some())
|
724 + | .count();
|
725 + | let origin_deliveries = self
|
726 + | .records
|
727 + | .values()
|
728 + | .filter(|record| {
|
729 + | matches!(
|
730 + | record.residence,
|
731 + | DemandResidence::Delivering {
|
732 + | view: DeliveryView::Origin,
|
733 + | ..
|
734 + | }
|
735 + | )
|
736 + | })
|
535 737 | .count();
|
536 - | let head = self.order.head();
|
537 - | let mut delivering = 0;
|
538 - | self.order.assert_consistent(
|
738 + | assert!(
|
739 + | origin_deliveries <= 1,
|
740 + | "more than one origin delivery fence was active"
|
741 + | );
|
742 + | self.origin_order.assert_consistent(
|
539 743 | ordered_records,
|
540 744 | self.records.len(),
|
541 - | "demand order",
|
745 + | "origin demand order",
|
542 746 | |requesting_partition| {
|
543 747 | let record = self
|
544 748 | .records
|
545 749 | .get(&requesting_partition)
|
546 - | .expect("ordered demand disappeared");
|
750 + | .expect("origin-ordered demand disappeared");
|
547 751 | match &record.residence {
|
548 - | DemandResidence::Idle => unreachable!("ordered demand became idle"),
|
752 + | DemandResidence::Idle => {
|
753 + | unreachable!("origin-ordered demand became idle")
|
754 + | }
|
549 755 | DemandResidence::Queued { demand, .. } => {
|
550 756 | assert!(record.latest.is_active(), "queued demand became inactive");
|
551 757 | assert_eq!(
|
552 758 | record.latest.id, *demand,
|
553 759 | "queued residence did not match its latest demand"
|
554 760 | );
|
555 761 | }
|
556 - | DemandResidence::Delivering { demand, .. } => {
|
557 - | delivering += 1;
|
558 - | assert_eq!(
|
559 - | Some(requesting_partition),
|
560 - | head,
|
561 - | "delivery fence moved away from the order head"
|
562 - | );
|
762 + | DemandResidence::Delivering { demand, view, .. } => {
|
763 + | if *view == DeliveryView::Origin {
|
764 + | assert_eq!(
|
765 + | Some(requesting_partition),
|
766 + | self.origin_order.head(),
|
767 + | "origin delivery fence moved away from its head"
|
768 + | );
|
769 + | }
|
563 770 | assert!(
|
564 771 | record.latest.id >= *demand,
|
565 772 | "delivery fence named a future demand"
|
566 773 | );
|
567 774 | }
|
568 775 | }
|
569 776 | *record
|
570 777 | .residence
|
571 - | .links()
|
572 - | .expect("ordered demand lost its links")
|
778 + | .links(DemandOrderView::Origin)
|
779 + | .expect("origin-ordered demand lost its links")
|
573 780 | },
|
574 781 | );
|
575 - | assert!(delivering <= 1, "more than one delivery fence was active");
|
782 + |
|
783 + | let group_ordered_records = self
|
784 + | .records
|
785 + | .values()
|
786 + | .filter(|record| record.residence.links(DemandOrderView::Group).is_some())
|
787 + | .count();
|
788 + | for record in self
|
789 + | .records
|
790 + | .values()
|
791 + | .filter(|record| record.residence.links(DemandOrderView::Group).is_some())
|
792 + | {
|
793 + | let group = record
|
794 + | .group
|
795 + | .as_ref()
|
796 + | .expect("group-ordered demand lost its eligibility group");
|
797 + | assert!(
|
798 + | self.group_orders.contains_key(group),
|
799 + | "ordered demand lost its eligibility-group order"
|
800 + | );
|
801 + | }
|
802 + | assert_eq!(
|
803 + | group_ordered_records,
|
804 + | self.group_orders
|
805 + | .values()
|
806 + | .map(IntrusiveOrder::len)
|
807 + | .sum::<usize>(),
|
808 + | "eligibility-group orders did not contain every ordered demand"
|
809 + | );
|
810 + |
|
811 + | for (group, order) in &self.group_orders {
|
812 + | let group_deliveries = self
|
813 + | .records
|
814 + | .values()
|
815 + | .filter(|record| {
|
816 + | record.group.as_ref() == Some(group)
|
817 + | && matches!(
|
818 + | record.residence,
|
819 + | DemandResidence::Delivering {
|
820 + | view: DeliveryView::Group,
|
821 + | ..
|
822 + | }
|
823 + | )
|
824 + | })
|
825 + | .count();
|
826 + | assert!(
|
827 + | group_deliveries <= 1,
|
828 + | "more than one group delivery fence was active"
|
829 + | );
|
830 + | let expected = self
|
831 + | .records
|
832 + | .values()
|
833 + | .filter(|record| {
|
834 + | record.residence.links(DemandOrderView::Group).is_some()
|
835 + | && record.group.as_ref() == Some(group)
|
836 + | })
|
837 + | .count();
|
838 + | order.assert_consistent(
|
839 + | expected,
|
840 + | self.records.len(),
|
841 + | "eligibility-group demand order",
|
842 + | |requesting_partition| {
|
843 + | let record = self
|
844 + | .records
|
845 + | .get(&requesting_partition)
|
846 + | .expect("group-ordered demand disappeared");
|
847 + | assert_eq!(
|
848 + | record.group.as_ref(),
|
849 + | Some(group),
|
850 + | "demand occupied the wrong eligibility-group order"
|
851 + | );
|
852 + | if matches!(
|
853 + | record.residence,
|
854 + | DemandResidence::Delivering {
|
855 + | view: DeliveryView::Group,
|
856 + | ..
|
857 + | }
|
858 + | ) {
|
859 + | assert_eq!(
|
860 + | Some(requesting_partition),
|
861 + | order.head(),
|
862 + | "group publication fence moved away from its head"
|
863 + | );
|
864 + | }
|
865 + | *record
|
866 + | .residence
|
867 + | .links(DemandOrderView::Group)
|
868 + | .expect("group-ordered demand lost its links")
|
869 + | },
|
870 + | );
|
871 + | }
|
872 + | }
|
873 + | }
|
874 + |
|
875 + | #[cfg(all(test, not(smithy_http_client_loom)))]
|
876 + | mod tests {
|
877 + | use super::*;
|
878 + |
|
879 + | fn partition(index: usize) -> PartitionId {
|
880 + | PartitionId::from_index(index)
|
881 + | }
|
882 + |
|
883 + | fn active(id: u64, group: EligibilityGroup) -> DemandSnapshot {
|
884 + | DemandSnapshot::active(
|
885 + | DemandId::from_u64(id),
|
886 + | super::super::SnapshotVersion::INITIAL,
|
887 + | ProtocolRequirement::H1Compatible,
|
888 + | group,
|
889 + | )
|
890 + | }
|
891 + |
|
892 + | #[test]
|
893 + | fn origin_and_group_orders_repair_independently() {
|
894 + | let pool = EligibilityGroup::Pool;
|
895 + | let isolated = EligibilityGroup::Partition(partition(2));
|
896 + | let mut schedule = DemandSchedule::default();
|
897 + | schedule.publish(partition(1), active(1, pool.clone()));
|
898 + | schedule.publish(partition(2), active(2, isolated.clone()));
|
899 + | schedule.publish(partition(3), active(3, pool.clone()));
|
900 + |
|
901 + | assert_eq!(
|
902 + | Some(partition(1)),
|
903 + | schedule.queued_head().map(|head| head.requesting_partition)
|
904 + | );
|
905 + | assert_eq!(
|
906 + | Some(partition(1)),
|
907 + | schedule
|
908 + | .queued_group_head(&pool)
|
909 + | .map(|head| head.requesting_partition)
|
910 + | );
|
911 + | assert_eq!(
|
912 + | Some(partition(2)),
|
913 + | schedule
|
914 + | .queued_group_head(&isolated)
|
915 + | .map(|head| head.requesting_partition)
|
916 + | );
|
917 + |
|
918 + | let delivery = DeliveryId(7);
|
919 + | schedule
|
920 + | .reserve_group_head(&pool, &partition(1), DemandId::from_u64(1), delivery)
|
921 + | .expect("pool group head should reserve");
|
922 + | assert!(schedule.queued_group_head(&pool).is_none());
|
923 + | assert_eq!(
|
924 + | Some(partition(2)),
|
925 + | schedule
|
926 + | .queued_group_head(&isolated)
|
927 + | .map(|head| head.requesting_partition)
|
928 + | );
|
929 + | assert!(schedule.queued_head().is_none());
|
930 + |
|
931 + | schedule.finish_delivery(
|
932 + | delivery,
|
933 + | &partition(1),
|
934 + | DeliveryAckResult::Accepted { successor: None },
|
935 + | );
|
936 + | assert_eq!(
|
937 + | Some(partition(2)),
|
938 + | schedule.queued_head().map(|head| head.requesting_partition)
|
939 + | );
|
940 + | assert_eq!(
|
941 + | Some(partition(3)),
|
942 + | schedule
|
943 + | .queued_group_head(&pool)
|
944 + | .map(|head| head.requesting_partition)
|
945 + | );
|
946 + | }
|
947 + |
|
948 + | #[test]
|
949 + | fn retry_preserves_both_order_positions() {
|
950 + | let group = EligibilityGroup::Pool;
|
951 + | let mut schedule = DemandSchedule::default();
|
952 + | schedule.publish(partition(1), active(1, group.clone()));
|
953 + | schedule.publish(partition(2), active(2, group.clone()));
|
954 + |
|
955 + | let delivery = DeliveryId(9);
|
956 + | schedule
|
957 + | .reserve_group_head(&group, &partition(1), DemandId::from_u64(1), delivery)
|
958 + | .expect("group head should reserve");
|
959 + | schedule.finish_delivery(
|
960 + | delivery,
|
961 + | &partition(1),
|
962 + | DeliveryAckResult::RetrySameResidence,
|
963 + | );
|
964 + |
|
965 + | assert_eq!(
|
966 + | Some(partition(1)),
|
967 + | schedule.queued_head().map(|head| head.requesting_partition)
|
968 + | );
|
969 + | assert_eq!(
|
970 + | Some(partition(1)),
|
971 + | schedule
|
972 + | .queued_group_head(&group)
|
973 + | .map(|head| head.requesting_partition)
|
974 + | );
|
975 + | }
|
976 + |
|
977 + | #[test]
|
978 + | fn inactive_snapshot_during_group_fence_retires_without_losing_group() {
|
979 + | let group = EligibilityGroup::Pool;
|
980 + | let mut schedule = DemandSchedule::default();
|
981 + | schedule.publish(partition(1), active(1, group.clone()));
|
982 + | let delivery = DeliveryId(11);
|
983 + | schedule
|
984 + | .reserve_group_head(&group, &partition(1), DemandId::from_u64(1), delivery)
|
985 + | .expect("group head should reserve");
|
986 + | schedule.publish(
|
987 + | partition(1),
|
988 + | DemandSnapshot::inactive(
|
989 + | DemandId::from_u64(1),
|
990 + | super::super::SnapshotVersion::INITIAL.next(),
|
991 + | ),
|
992 + | );
|
993 + |
|
994 + | schedule.finish_delivery(
|
995 + | delivery,
|
996 + | &partition(1),
|
997 + | DeliveryAckResult::RetrySameResidence,
|
998 + | );
|
999 + | assert_eq!(0, schedule.len());
|
1000 + | assert!(schedule.queued_group_head(&group).is_none());
|
1001 + | }
|
1002 + |
|
1003 + | #[test]
|
1004 + | fn accepted_fence_does_not_retire_a_newer_active_publication() {
|
1005 + | let group = EligibilityGroup::Pool;
|
1006 + | let partition = partition(1);
|
1007 + | let demand = DemandId::from_u64(1);
|
1008 + | let initial = super::super::SnapshotVersion::INITIAL;
|
1009 + | let mut schedule = DemandSchedule::default();
|
1010 + | schedule.publish(
|
1011 + | partition,
|
1012 + | DemandSnapshot::active(
|
1013 + | demand,
|
1014 + | initial,
|
1015 + | ProtocolRequirement::H2Required,
|
1016 + | group.clone(),
|
1017 + | ),
|
1018 + | );
|
1019 + | let delivery = DeliveryId(12);
|
1020 + | schedule
|
1021 + | .reserve_group_head(&group, &partition, demand, delivery)
|
1022 + | .expect("group head should reserve");
|
1023 + | let republished = DemandSnapshot::active(
|
1024 + | demand,
|
1025 + | initial.next().next(),
|
1026 + | ProtocolRequirement::H2Required,
|
1027 + | group.clone(),
|
1028 + | );
|
1029 + | schedule.publish(partition, republished.clone());
|
1030 + |
|
1031 + | schedule.finish_delivery(
|
1032 + | delivery,
|
1033 + | &partition,
|
1034 + | DeliveryAckResult::Accepted { successor: None },
|
1035 + | );
|
1036 + |
|
1037 + | assert_eq!(Some(&republished), schedule.latest_for_test(&partition));
|
1038 + | assert_eq!(
|
1039 + | Some(partition),
|
1040 + | schedule.queued_head().map(|head| head.requesting_partition)
|
1041 + | );
|
1042 + | assert_eq!(
|
1043 + | Some(partition),
|
1044 + | schedule
|
1045 + | .queued_group_head(&group)
|
1046 + | .map(|head| head.requesting_partition)
|
1047 + | );
|
1048 + | }
|
1049 + |
|
1050 + | #[test]
|
1051 + | fn group_reservation_rejects_a_stale_demand_identity() {
|
1052 + | let group = EligibilityGroup::Pool;
|
1053 + | let partition = partition(1);
|
1054 + | let mut schedule = DemandSchedule::default();
|
1055 + | schedule.publish(partition, active(1, group.clone()));
|
1056 + | schedule.publish(partition, active(2, group.clone()));
|
1057 + |
|
1058 + | assert!(schedule
|
1059 + | .reserve_group_head(&group, &partition, DemandId::from_u64(1), DeliveryId(13),)
|
1060 + | .is_none());
|
1061 + | assert_eq!(
|
1062 + | Some(DemandId::from_u64(2)),
|
1063 + | schedule
|
1064 + | .queued_group_head(&group)
|
1065 + | .map(|queued| queued.demand)
|
1066 + | );
|
1067 + | }
|
1068 + |
|
1069 + | #[test]
|
1070 + | fn stale_acknowledgement_does_not_close_a_newer_delivery_fence() {
|
1071 + | let group = EligibilityGroup::Pool;
|
1072 + | let partition = partition(1);
|
1073 + | let demand = DemandId::from_u64(1);
|
1074 + | let mut schedule = DemandSchedule::default();
|
1075 + | schedule.publish(partition, active(1, group.clone()));
|
1076 + | schedule
|
1077 + | .reserve_group_head(&group, &partition, demand, DeliveryId(14))
|
1078 + | .expect("group head should reserve");
|
1079 + |
|
1080 + | schedule.finish_delivery(
|
1081 + | DeliveryId(15),
|
1082 + | &partition,
|
1083 + | DeliveryAckResult::Accepted { successor: None },
|
1084 + | );
|
1085 + |
|
1086 + | assert!(schedule.delivery_is_current(DeliveryId(14), &partition, demand));
|
1087 + | assert!(!schedule.delivery_is_current(DeliveryId(15), &partition, demand));
|
1088 + | assert_eq!(
|
1089 + | Some(&active(1, group)),
|
1090 + | schedule.latest_for_test(&partition)
|
1091 + | );
|
1092 + | }
|
1093 + |
|
1094 + | #[test]
|
1095 + | #[should_panic(expected = "ordered demand lost its eligibility-group order")]
|
1096 + | fn consistency_check_rejects_an_orphaned_group_link() {
|
1097 + | let group = EligibilityGroup::Pool;
|
1098 + | let mut schedule = DemandSchedule::default();
|
1099 + | schedule.publish(partition(1), active(1, group.clone()));
|
1100 + |
|
1101 + | schedule.group_orders.remove(&group);
|
1102 + | schedule.assert_consistent();
|
576 1103 | }
|
577 1104 | }
|