subetha_cxc/ring_contract.rs
1//! `RingContract` - the declared operation envelope for a ring.
2//!
3//! A [`RingContract`] is the USER OVERRIDE on the otherwise fully
4//! automatic ring: peer-count ceilings, an ordering contract, and a
5//! capacity bound, declared as one validated artifact every attaching
6//! party agrees on. An `AdaptiveRing` WITHOUT a declared contract is
7//! unbounded - peers grow the ring on demand and registration never
8//! fails; declaring a contract is the only thing that makes
9//! `TooManyProducers` / `TooManyConsumers` possible.
10//!
11//! (Lineage: path expressions - Campbell & Habermann, 1974 - declare
12//! the legal operation histories of a shared type, with enforcement
13//! derived rather than hand-written. This module is the
14//! counter-compilable fragment of that idea, expressed entirely as
15//! data the rings already track.)
16//!
17//! Two jobs:
18//!
19//! 1. **One validated pin.** The peer ceilings
20//! ([`from_counts`](RingContract::from_counts)) and the ordering /
21//! capacity constraints live in a single declared artifact instead
22//! of scattered flags.
23//! 2. **Give the adaptive policy a feasible-region filter.** A policy
24//! proposes a candidate `(shape, capacity)`; [`permits_config`]
25//! rejects any move that would violate the declared envelope, so an
26//! aggressive auto-morph cannot break the contract by construction.
27//!
28//! Enforcement cost on the hot path is zero - the contract is consulted
29//! only at attach time and at policy-tick time. The
30//! ordering-contract-to-shape rule is where it earns its keep: the
31//! sharded [`Mpmc`](RingShape::Mpmc) shape (per-producer lanes)
32//! delivers only per-producer FIFO, so it is illegal under a `Fifo`
33//! (global-total-order) contract - which is exactly the
34//! `GlobalFifo -> Vyukov` rule the QoS shape policy also applies,
35//! derived here from the declared contract.
36//!
37//! [`permits_config`]: RingContract::permits_config
38
39use crate::adaptive_ring::RingShape;
40
41/// The ordering envelope a ring's consumers may observe.
42#[derive(Debug, Clone, Copy, PartialEq, Eq)]
43pub enum OrderingContract {
44 /// Global total order across all producers (the strongest).
45 Fifo,
46 /// Per-producer FIFO; items from different producers may interleave
47 /// arbitrarily, but each producer's items arrive in push order.
48 FifoPerProducer,
49 /// Bounded reordering: an item is delivered at most `k` positions
50 /// from its global arrival order.
51 KOutOfOrder(u32),
52 /// No ordering guarantee.
53 Unordered,
54}
55
56/// Declared operation envelope for a ring. The two count bounds use
57/// `0` to mean "unbounded".
58#[derive(Debug, Clone, Copy, PartialEq, Eq)]
59pub struct RingContract {
60 /// Max concurrently-registered producers (`0` = unbounded).
61 pub max_concurrent_push: u8,
62 /// Max concurrently-registered consumers (`0` = unbounded).
63 pub max_concurrent_pop: u8,
64 /// The ordering envelope the consumers may observe.
65 pub ordering: OrderingContract,
66 /// Optional hard ceiling on ring capacity (`None` = unbounded).
67 pub capacity_bound: Option<u32>,
68}
69
70impl RingContract {
71 /// A contract that PINS the peer counts: registration past
72 /// `max_producers` / `max_consumers` returns `TooManyProducers` /
73 /// `TooManyConsumers` instead of growing the ring. No ordering
74 /// constraint, no capacity bound. The common fixed-topology
75 /// declaration.
76 pub fn from_counts(max_producers: usize, max_consumers: usize) -> Self {
77 Self {
78 max_concurrent_push: max_producers.min(u8::MAX as usize) as u8,
79 max_concurrent_pop: max_consumers.min(u8::MAX as usize) as u8,
80 ordering: OrderingContract::Unordered,
81 capacity_bound: None,
82 }
83 }
84
85 /// The fully-unbounded contract: any peer counts, any capacity,
86 /// no ordering constraint. The DEFAULT for rings with no declared
87 /// contract - registration never fails under it (peers grow the
88 /// ring on demand up to the substrate slot ceilings).
89 pub fn unbounded() -> Self {
90 Self {
91 max_concurrent_push: 0,
92 max_concurrent_pop: 0,
93 ordering: OrderingContract::Unordered,
94 capacity_bound: None,
95 }
96 }
97
98 /// May another producer attach, given `active` are registered?
99 #[inline]
100 pub fn permits_producer(&self, active: usize) -> bool {
101 self.max_concurrent_push == 0 || active < self.max_concurrent_push as usize
102 }
103
104 /// May another consumer attach, given `active` are registered?
105 #[inline]
106 pub fn permits_consumer(&self, active: usize) -> bool {
107 self.max_concurrent_pop == 0 || active < self.max_concurrent_pop as usize
108 }
109
110 /// Is a ring of `capacity` slots legal under the contract?
111 #[inline]
112 pub fn permits_capacity(&self, capacity: usize) -> bool {
113 match self.capacity_bound {
114 None => true,
115 Some(bound) => capacity <= bound as usize,
116 }
117 }
118
119 /// Is `shape` legal under the ordering contract? Global total order
120 /// is preserved only by the single-stream [`Spsc`](RingShape::Spsc)
121 /// and the shared-sequence [`Vyukov`](RingShape::Vyukov); the
122 /// partitioned per-producer-lane shapes
123 /// ([`Mpsc`](RingShape::Mpsc), [`Mpmc`](RingShape::Mpmc)) interleave
124 /// producers, so both are illegal under a `Fifo` contract. Every
125 /// other contract permits every shape (`FifoPerProducer` is exactly
126 /// what the lanes deliver).
127 #[inline]
128 pub fn permits_shape(&self, shape: RingShape) -> bool {
129 match self.ordering {
130 OrderingContract::Fifo => {
131 matches!(shape, RingShape::Spsc | RingShape::Vyukov)
132 }
133 _ => true,
134 }
135 }
136
137 /// Feasible-region oracle for an adaptive policy: is a candidate
138 /// `(shape, capacity)` configuration legal? A policy filters every
139 /// move it proposes through this, so an aggressive auto-morph
140 /// cannot violate the declared envelope.
141 #[inline]
142 pub fn permits_config(&self, shape: RingShape, capacity: usize) -> bool {
143 self.permits_shape(shape) && self.permits_capacity(capacity)
144 }
145}
146
147#[cfg(test)]
148mod tests {
149 use super::*;
150
151 #[test]
152 fn from_counts_pins_the_peer_ceilings() {
153 let g = RingContract::from_counts(4, 2);
154 assert_eq!(g.max_concurrent_push, 4);
155 assert_eq!(g.max_concurrent_pop, 2);
156 assert_eq!(g.ordering, OrderingContract::Unordered);
157 assert_eq!(g.capacity_bound, None);
158 // The pin: ids 0..4 permitted, a 5th producer rejected.
159 assert!(g.permits_producer(0) && g.permits_producer(3));
160 assert!(!g.permits_producer(4));
161 assert!(g.permits_consumer(0) && g.permits_consumer(1));
162 assert!(!g.permits_consumer(2));
163 }
164
165 #[test]
166 fn unbounded_counts_permit_any() {
167 let g = RingContract {
168 max_concurrent_push: 0,
169 max_concurrent_pop: 0,
170 ordering: OrderingContract::Unordered,
171 capacity_bound: None,
172 };
173 assert!(g.permits_producer(1000));
174 assert!(g.permits_consumer(1000));
175 }
176
177 #[test]
178 fn fifo_permits_only_order_preserving_shapes() {
179 let fifo = RingContract {
180 max_concurrent_push: 0,
181 max_concurrent_pop: 0,
182 ordering: OrderingContract::Fifo,
183 capacity_bound: None,
184 };
185 // Single-stream and shared-sequence preserve global total order.
186 assert!(fifo.permits_shape(RingShape::Spsc));
187 assert!(fifo.permits_shape(RingShape::Vyukov));
188 // Both partitioned per-producer-lane shapes reorder producers.
189 assert!(!fifo.permits_shape(RingShape::Mpsc));
190 assert!(!fifo.permits_shape(RingShape::Mpmc));
191 }
192
193 #[test]
194 fn relaxed_contracts_permit_every_shape() {
195 for ordering in [
196 OrderingContract::FifoPerProducer,
197 OrderingContract::KOutOfOrder(8),
198 OrderingContract::Unordered,
199 ] {
200 let g = RingContract {
201 max_concurrent_push: 0,
202 max_concurrent_pop: 0,
203 ordering,
204 capacity_bound: None,
205 };
206 for shape in [
207 RingShape::Spsc,
208 RingShape::Mpsc,
209 RingShape::Mpmc,
210 RingShape::Vyukov,
211 ] {
212 assert!(g.permits_shape(shape), "{ordering:?} should permit {shape:?}");
213 }
214 }
215 }
216
217 #[test]
218 fn capacity_bound_enforced() {
219 let g = RingContract {
220 max_concurrent_push: 0,
221 max_concurrent_pop: 0,
222 ordering: OrderingContract::Unordered,
223 capacity_bound: Some(1024),
224 };
225 assert!(g.permits_capacity(512) && g.permits_capacity(1024));
226 assert!(!g.permits_capacity(2048));
227 // permits_config combines shape + capacity.
228 assert!(g.permits_config(RingShape::Mpmc, 1024));
229 assert!(!g.permits_config(RingShape::Mpmc, 2048));
230 }
231
232 #[test]
233 fn fifo_with_capacity_combined_in_permits_config() {
234 let g = RingContract {
235 max_concurrent_push: 4,
236 max_concurrent_pop: 4,
237 ordering: OrderingContract::Fifo,
238 capacity_bound: Some(4096),
239 };
240 // Mpmc illegal regardless of capacity under Fifo.
241 assert!(!g.permits_config(RingShape::Mpmc, 1024));
242 // Vyukov legal at/under the capacity bound, illegal above it.
243 assert!(g.permits_config(RingShape::Vyukov, 4096));
244 assert!(!g.permits_config(RingShape::Vyukov, 8192));
245 }
246}