/home/noah/src/trueno/src/brick/buffer.rs
Line | Count | Source |
1 | | //! Buffer Management with Flow Control |
2 | | //! |
3 | | //! Two-tier watermarks for back-pressure control. |
4 | | |
5 | | // ---------------------------------------------------------------------------- |
6 | | // AWP-01: Two-Tier Buffer Watermarks |
7 | | // ---------------------------------------------------------------------------- |
8 | | |
9 | | /// Two-tier buffer watermarks for back-pressure control. |
10 | | /// |
11 | | /// # Example |
12 | | /// ```rust |
13 | | /// use trueno::brick::BufferWatermarks; |
14 | | /// |
15 | | /// let wm = BufferWatermarks::default(); |
16 | | /// assert!(!wm.should_backpressure(1000)); // Below high watermark |
17 | | /// assert!(wm.should_backpressure(10000)); // Above high watermark |
18 | | /// ``` |
19 | | #[derive(Debug, Clone, Copy)] |
20 | | pub struct BufferWatermarks { |
21 | | /// Low watermark: resume writing when buffer drops below this |
22 | | pub low: usize, |
23 | | /// High watermark: apply back-pressure when buffer exceeds this |
24 | | pub high: usize, |
25 | | } |
26 | | |
27 | | impl Default for BufferWatermarks { |
28 | 0 | fn default() -> Self { |
29 | 0 | Self { |
30 | 0 | low: 1024, // 1KB |
31 | 0 | high: 8 * 1024, // 8KB |
32 | 0 | } |
33 | 0 | } |
34 | | } |
35 | | |
36 | | impl BufferWatermarks { |
37 | | /// Create new watermarks. |
38 | | /// |
39 | | /// # Panics |
40 | | /// Panics if low >= high. |
41 | 0 | pub fn new(low: usize, high: usize) -> Self { |
42 | 0 | assert!(low < high, "low watermark must be less than high"); |
43 | 0 | Self { low, high } |
44 | 0 | } |
45 | | |
46 | | /// Check if back-pressure should be applied. |
47 | | #[must_use] |
48 | 0 | pub fn should_backpressure(&self, current: usize) -> bool { |
49 | 0 | current >= self.high |
50 | 0 | } |
51 | | |
52 | | /// Check if writing can resume. |
53 | | #[must_use] |
54 | 0 | pub fn can_write(&self, current: usize) -> bool { |
55 | 0 | current < self.low |
56 | 0 | } |
57 | | |
58 | | /// Get pressure level (0.0 = empty, 1.0 = at high watermark). |
59 | | #[must_use] |
60 | 0 | pub fn pressure_level(&self, current: usize) -> f64 { |
61 | 0 | (current as f64 / self.high as f64).min(1.0) |
62 | 0 | } |
63 | | } |
64 | | |
65 | | /// Buffer with watermark-based flow control. |
66 | | #[derive(Debug)] |
67 | | pub struct WatermarkedBuffer { |
68 | | data: Vec<u8>, |
69 | | watermarks: BufferWatermarks, |
70 | | } |
71 | | |
72 | | impl WatermarkedBuffer { |
73 | | /// Create a new watermarked buffer. |
74 | 0 | pub fn new(watermarks: BufferWatermarks) -> Self { |
75 | 0 | Self { |
76 | 0 | data: Vec::with_capacity(watermarks.high), |
77 | 0 | watermarks, |
78 | 0 | } |
79 | 0 | } |
80 | | |
81 | | /// Check if back-pressure should be applied. |
82 | | #[must_use] |
83 | 0 | pub fn should_backpressure(&self) -> bool { |
84 | 0 | self.watermarks.should_backpressure(self.data.len()) |
85 | 0 | } |
86 | | |
87 | | /// Check if writing can resume. |
88 | | #[must_use] |
89 | 0 | pub fn can_write(&self) -> bool { |
90 | 0 | self.watermarks.can_write(self.data.len()) |
91 | 0 | } |
92 | | |
93 | | /// Get current buffer length. |
94 | | #[must_use] |
95 | 0 | pub fn len(&self) -> usize { |
96 | 0 | self.data.len() |
97 | 0 | } |
98 | | |
99 | | /// Check if buffer is empty. |
100 | | #[must_use] |
101 | 0 | pub fn is_empty(&self) -> bool { |
102 | 0 | self.data.is_empty() |
103 | 0 | } |
104 | | |
105 | | /// Write data to the buffer. |
106 | 0 | pub fn write(&mut self, data: &[u8]) { |
107 | 0 | self.data.extend_from_slice(data); |
108 | 0 | } |
109 | | |
110 | | /// Drain data from the buffer. |
111 | 0 | pub fn drain(&mut self, amount: usize) -> Vec<u8> { |
112 | 0 | let amount = amount.min(self.data.len()); |
113 | 0 | self.data.drain(..amount).collect() |
114 | 0 | } |
115 | | |
116 | | /// Clear the buffer. |
117 | 0 | pub fn clear(&mut self) { |
118 | 0 | self.data.clear(); |
119 | 0 | } |
120 | | |
121 | | /// Get the watermarks configuration. |
122 | | #[must_use] |
123 | 0 | pub fn watermarks(&self) -> BufferWatermarks { |
124 | 0 | self.watermarks |
125 | 0 | } |
126 | | |
127 | | /// Get current pressure level. |
128 | | #[must_use] |
129 | 0 | pub fn pressure_level(&self) -> f64 { |
130 | 0 | self.watermarks.pressure_level(self.data.len()) |
131 | 0 | } |
132 | | } |
133 | | |
134 | | impl Default for WatermarkedBuffer { |
135 | 0 | fn default() -> Self { |
136 | 0 | Self::new(BufferWatermarks::default()) |
137 | 0 | } |
138 | | } |
139 | | |
140 | | #[cfg(test)] |
141 | | mod tests { |
142 | | use super::*; |
143 | | |
144 | | #[test] |
145 | | fn test_buffer_watermarks_default() { |
146 | | let wm = BufferWatermarks::default(); |
147 | | assert_eq!(wm.low, 1024); |
148 | | assert_eq!(wm.high, 8 * 1024); |
149 | | } |
150 | | |
151 | | #[test] |
152 | | fn test_buffer_watermarks_new() { |
153 | | let wm = BufferWatermarks::new(100, 1000); |
154 | | assert_eq!(wm.low, 100); |
155 | | assert_eq!(wm.high, 1000); |
156 | | } |
157 | | |
158 | | #[test] |
159 | | #[should_panic(expected = "low watermark must be less than high")] |
160 | | fn test_buffer_watermarks_invalid() { |
161 | | BufferWatermarks::new(1000, 100); |
162 | | } |
163 | | |
164 | | #[test] |
165 | | fn test_buffer_watermarks_backpressure() { |
166 | | let wm = BufferWatermarks::new(100, 1000); |
167 | | assert!(!wm.should_backpressure(500)); |
168 | | assert!(!wm.should_backpressure(999)); |
169 | | assert!(wm.should_backpressure(1000)); |
170 | | assert!(wm.should_backpressure(1500)); |
171 | | } |
172 | | |
173 | | #[test] |
174 | | fn test_buffer_watermarks_can_write() { |
175 | | let wm = BufferWatermarks::new(100, 1000); |
176 | | assert!(wm.can_write(50)); |
177 | | assert!(wm.can_write(99)); |
178 | | assert!(!wm.can_write(100)); |
179 | | assert!(!wm.can_write(500)); |
180 | | } |
181 | | |
182 | | #[test] |
183 | | fn test_buffer_watermarks_pressure_level() { |
184 | | let wm = BufferWatermarks::new(100, 1000); |
185 | | assert!((wm.pressure_level(0) - 0.0).abs() < 0.001); |
186 | | assert!((wm.pressure_level(500) - 0.5).abs() < 0.001); |
187 | | assert!((wm.pressure_level(1000) - 1.0).abs() < 0.001); |
188 | | assert!((wm.pressure_level(2000) - 1.0).abs() < 0.001); // Capped at 1.0 |
189 | | } |
190 | | |
191 | | #[test] |
192 | | fn test_watermarked_buffer_new() { |
193 | | let wm = BufferWatermarks::new(100, 1000); |
194 | | let buffer = WatermarkedBuffer::new(wm); |
195 | | assert!(buffer.is_empty()); |
196 | | assert_eq!(buffer.len(), 0); |
197 | | } |
198 | | |
199 | | #[test] |
200 | | fn test_watermarked_buffer_write() { |
201 | | let mut buffer = WatermarkedBuffer::default(); |
202 | | buffer.write(&[1, 2, 3, 4, 5]); |
203 | | assert_eq!(buffer.len(), 5); |
204 | | assert!(!buffer.is_empty()); |
205 | | } |
206 | | |
207 | | #[test] |
208 | | fn test_watermarked_buffer_drain() { |
209 | | let mut buffer = WatermarkedBuffer::default(); |
210 | | buffer.write(&[1, 2, 3, 4, 5]); |
211 | | let drained = buffer.drain(3); |
212 | | assert_eq!(drained, vec![1, 2, 3]); |
213 | | assert_eq!(buffer.len(), 2); |
214 | | } |
215 | | |
216 | | #[test] |
217 | | fn test_watermarked_buffer_drain_more_than_available() { |
218 | | let mut buffer = WatermarkedBuffer::default(); |
219 | | buffer.write(&[1, 2, 3]); |
220 | | let drained = buffer.drain(10); |
221 | | assert_eq!(drained, vec![1, 2, 3]); |
222 | | assert!(buffer.is_empty()); |
223 | | } |
224 | | |
225 | | #[test] |
226 | | fn test_watermarked_buffer_clear() { |
227 | | let mut buffer = WatermarkedBuffer::default(); |
228 | | buffer.write(&[1, 2, 3, 4, 5]); |
229 | | buffer.clear(); |
230 | | assert!(buffer.is_empty()); |
231 | | } |
232 | | |
233 | | #[test] |
234 | | fn test_watermarked_buffer_backpressure() { |
235 | | let wm = BufferWatermarks::new(10, 100); |
236 | | let mut buffer = WatermarkedBuffer::new(wm); |
237 | | |
238 | | assert!(!buffer.should_backpressure()); |
239 | | assert!(buffer.can_write()); |
240 | | |
241 | | buffer.write(&[0u8; 50]); |
242 | | assert!(!buffer.should_backpressure()); |
243 | | assert!(!buffer.can_write()); |
244 | | |
245 | | buffer.write(&[0u8; 50]); |
246 | | assert!(buffer.should_backpressure()); |
247 | | } |
248 | | |
249 | | #[test] |
250 | | fn test_watermarked_buffer_pressure_level() { |
251 | | let wm = BufferWatermarks::new(10, 100); |
252 | | let mut buffer = WatermarkedBuffer::new(wm); |
253 | | |
254 | | assert!((buffer.pressure_level() - 0.0).abs() < 0.001); |
255 | | |
256 | | buffer.write(&[0u8; 50]); |
257 | | assert!((buffer.pressure_level() - 0.5).abs() < 0.001); |
258 | | |
259 | | buffer.write(&[0u8; 50]); |
260 | | assert!((buffer.pressure_level() - 1.0).abs() < 0.001); |
261 | | } |
262 | | |
263 | | #[test] |
264 | | fn test_watermarked_buffer_watermarks_getter() { |
265 | | let wm = BufferWatermarks::new(100, 1000); |
266 | | let buffer = WatermarkedBuffer::new(wm); |
267 | | let retrieved = buffer.watermarks(); |
268 | | assert_eq!(retrieved.low, 100); |
269 | | assert_eq!(retrieved.high, 1000); |
270 | | } |
271 | | |
272 | | /// FALSIFICATION TEST (Section 4.3) |
273 | | /// |
274 | | /// Verifies the buffer's behavior when pushed past its high watermark. |
275 | | /// The WatermarkedBuffer is a SIGNALING mechanism, not a blocking mechanism. |
276 | | /// It will accept writes beyond the limit but MUST correctly signal backpressure. |
277 | | /// |
278 | | /// Failure mode: If this test passes but the buffer doesn't signal, |
279 | | /// the caller would never know to stop writing, leading to OOM. |
280 | | #[test] |
281 | | fn test_falsify_buffer_limit_signaling() { |
282 | | let wm = BufferWatermarks::new(10, 100); |
283 | | let mut buffer = WatermarkedBuffer::new(wm); |
284 | | |
285 | | // Write incrementally and verify signaling at each step |
286 | | let mut backpressure_signaled_at = None; |
287 | | |
288 | | for i in 0..200 { |
289 | | let chunk = [i as u8; 10]; // 10 bytes per write |
290 | | buffer.write(&chunk); |
291 | | |
292 | | if buffer.should_backpressure() && backpressure_signaled_at.is_none() { |
293 | | backpressure_signaled_at = Some(buffer.len()); |
294 | | } |
295 | | } |
296 | | |
297 | | // CRITICAL ASSERTION: Backpressure MUST have been signaled |
298 | | assert!( |
299 | | backpressure_signaled_at.is_some(), |
300 | | "FALSIFICATION FAILED: Buffer accepted 2000 bytes without signaling backpressure" |
301 | | ); |
302 | | |
303 | | // Verify it was signaled at the right threshold (at or above high watermark) |
304 | | let signal_point = backpressure_signaled_at.unwrap(); |
305 | | assert!( |
306 | | signal_point >= 100, |
307 | | "FALSIFICATION FAILED: Backpressure signaled too early at {} bytes (high=100)", |
308 | | signal_point |
309 | | ); |
310 | | |
311 | | // Verify buffer actually accepted all the data (it's non-blocking) |
312 | | assert_eq!( |
313 | | buffer.len(), |
314 | | 2000, |
315 | | "Buffer should accept all writes (signaling is advisory)" |
316 | | ); |
317 | | |
318 | | // Verify pressure level is capped at 1.0 even when way over |
319 | | assert!( |
320 | | (buffer.pressure_level() - 1.0).abs() < 0.001, |
321 | | "Pressure level should cap at 1.0 when over limit" |
322 | | ); |
323 | | } |
324 | | |
325 | | /// FALSIFICATION TEST: Verify drain restores write capability |
326 | | #[test] |
327 | | fn test_falsify_drain_restores_writability() { |
328 | | let wm = BufferWatermarks::new(10, 100); |
329 | | let mut buffer = WatermarkedBuffer::new(wm); |
330 | | |
331 | | // Fill past high watermark |
332 | | buffer.write(&[0u8; 150]); |
333 | | assert!(buffer.should_backpressure()); |
334 | | assert!(!buffer.can_write()); |
335 | | |
336 | | // Drain to below low watermark |
337 | | buffer.drain(145); // Should leave 5 bytes |
338 | | assert_eq!(buffer.len(), 5); |
339 | | |
340 | | // CRITICAL: Must restore write capability |
341 | | assert!( |
342 | | buffer.can_write(), |
343 | | "FALSIFICATION FAILED: Draining below low watermark did not restore writability" |
344 | | ); |
345 | | assert!( |
346 | | !buffer.should_backpressure(), |
347 | | "FALSIFICATION FAILED: Draining below low watermark did not clear backpressure" |
348 | | ); |
349 | | } |
350 | | } |