sark_core/body/
flow.rs

1use core::{marker, num::NonZero};
2
3use o3::buffer::segments;
4
5use crate::body::{
6    self, credit, ingress,
7    input::Parts,
8    stream::{Admission, Drive, Stream},
9};
10
11enum State<B, O> {
12    Building(B),
13    Ready(O),
14}
15
16#[derive(Clone, Copy, Debug, PartialEq, Eq)]
17pub enum Error<E> {
18    LengthMismatch,
19    Credit(credit::Error),
20    Build(E),
21    Release { build: E, error: credit::Error },
22}
23
24pub(super) enum Push<B, E> {
25    Accepted(B),
26    Failed { build: B, error: body::Error },
27    Rejected { build: B, error: E },
28}
29
30/// Linear request construction whose ready payload replaces its builder.
31pub struct Flow<'d, B: body::Build<'d>> {
32    state: State<B, B::Output>,
33    pub(in crate::body) lifetime: marker::PhantomData<&'d ()>,
34}
35
36/// Completed output retaining its exact finished ingress and unsettled credit.
37#[must_use = "admit the output or discard its retained owner"]
38pub struct Ready<'d, B: body::Build<'d>, O: credit::Origin = credit::raw::Stored> {
39    pub(in crate::body) state: ingress::Finished<O>,
40    pub(in crate::body) output: B::Output,
41    pub(in crate::body) lifetime: marker::PhantomData<&'d ()>,
42}
43
44impl<'d, B: body::Build<'d>, O: credit::Origin> Ready<'d, B, O> {
45    pub fn try_admit<E: credit::Evidence>(
46        mut self,
47        ledger: &mut credit::raw::Ledger<'_, O>,
48        mut event: E,
49    ) -> Admission<'d, B, E, O> {
50        match self.state.try_drain(ledger, &mut event) {
51            Ok(()) => Admission::Accepted(self.output),
52            Err(error) => Admission::Rejected {
53                ready: self,
54                evidence: event,
55                error,
56            },
57        }
58    }
59
60    pub fn discard<E: credit::Evidence>(
61        self,
62        ledger: &mut credit::raw::Ledger<'_, O>,
63        event: E,
64    ) -> credit::Discard<O> {
65        self.into_release().try_discard(ledger, event)
66    }
67
68    pub fn into_release(self) -> credit::Release<O> {
69        let Self { state, output, .. } = self;
70        drop(output);
71        state.into_state().abandon()
72    }
73}
74
75impl<'d, B: body::Build<'d>> Flow<'d, B> {
76    pub const fn new(build: B) -> Self {
77        Self {
78            state: State::Building(build),
79            lifetime: marker::PhantomData,
80        }
81    }
82
83    pub(super) fn from_done(build: B, len: usize) -> Self {
84        Self::new(build.finish_input(len))
85    }
86
87    fn from_output(output: B::Output) -> Self {
88        Self {
89            state: State::Ready(output),
90            lifetime: marker::PhantomData,
91        }
92    }
93
94    pub(super) fn try_parts<const COUNT: bool, I>(
95        mut build: B,
96        parts: I,
97        bytes: usize,
98        chunks: segments::Count,
99    ) -> Push<Self, <B as body::Push<'d>>::Error>
100    where
101        B: body::Push<'d>,
102        I: Parts<'d>,
103    {
104        if I::EXACT {
105            let size = parts.dimensions();
106            if size.bytes() != bytes || COUNT && size.chunks() != chunks {
107                return Push::Failed {
108                    build: Self::new(build),
109                    error: body::Error::SegmentMismatch,
110                };
111            }
112        }
113        let mut seen = 0usize;
114        let mut count = segments::Count::EMPTY;
115        for part in parts {
116            if !I::EXACT && part.is_empty() {
117                continue;
118            }
119            if !I::EXACT {
120                let Some(total) = seen.checked_add(part.len()).filter(|total| *total <= bytes)
121                else {
122                    return Push::Failed {
123                        build: Self::new(build),
124                        error: body::Error::SegmentMismatch,
125                    };
126                };
127                seen = total;
128                if COUNT {
129                    let Some(tally) = count
130                        .checked_add(segments::Count::ONE)
131                        .filter(|tally| *tally <= chunks)
132                    else {
133                        return Push::Failed {
134                            build: Self::new(build),
135                            error: body::Error::SegmentMismatch,
136                        };
137                    };
138                    count = tally;
139                }
140            }
141            build = match body::Push::push(build, part) {
142                body::Transfer::Accepted(build) => build,
143                body::Transfer::Rejected { build, error } => {
144                    return Push::Rejected {
145                        build: Self::new(build),
146                        error,
147                    };
148                }
149            };
150        }
151        if !I::EXACT && (seen != bytes || COUNT && count != chunks) {
152            return Push::Failed {
153                build: Self::new(build),
154                error: body::Error::SegmentMismatch,
155            };
156        }
157        Push::Accepted(Self::new(build))
158    }
159
160    pub(super) fn try_push<I, E, O: credit::Origin>(
161        self,
162        turn: &mut ingress::Turn<'_, '_, E, O>,
163        parts: I,
164        bytes: usize,
165        chunks: segments::Count,
166        rmax: credit::Limits,
167        cmax: credit::Limits,
168    ) -> Push<Self, <B as body::Push<'d>>::Error>
169    where
170        B: body::Push<'d>,
171        I: Parts<'d>,
172        E: credit::Evidence,
173    {
174        let State::Building(build) = self.state else {
175            return Push::Failed {
176                build: self,
177                error: body::Error::InputFinished,
178            };
179        };
180        let next = match turn.try_advance(bytes) {
181            Ok(next) => next,
182            Err(error) => {
183                return Push::Failed {
184                    build: Self::new(build),
185                    error,
186                };
187            }
188        };
189        let amount = credit::Amount::new(bytes, chunks);
190        let next = match next.try_reserve(amount, rmax, cmax) {
191            Ok(next) => next.commit(),
192            Err(error) => {
193                return Push::Failed {
194                    build: Self::new(build),
195                    error: body::Error::Credit(error),
196                };
197            }
198        };
199
200        let transfer = Self::try_parts::<true, _>(build, parts, bytes, chunks);
201        if matches!(transfer, Push::Accepted(_)) {
202            next.commit();
203        }
204        transfer
205    }
206
207    /// Finishes with the final logical total, including previously counted input.
208    /// Discarded input may defer accounting until its carrier knows that total.
209    pub fn try_exact<E: credit::Evidence, O: credit::Origin>(
210        self,
211        turn: &mut ingress::Turn<'_, '_, E, O>,
212        len: usize,
213    ) -> body::Transfer<Self, body::Error> {
214        if let Err(error) = turn.state_mut().try_exact(len) {
215            return body::Transfer::Rejected { build: self, error };
216        }
217        self.try_finish(turn.state_mut())
218    }
219
220    pub(super) fn try_finish<O: credit::Origin>(
221        self,
222        state: &mut ingress::State<O>,
223    ) -> body::Transfer<Self, body::Error> {
224        self.try_end_with(|| state.try_end())
225    }
226
227    pub(super) fn try_complete<O: credit::Origin>(
228        self,
229        state: &mut ingress::State<O>,
230    ) -> body::Transfer<Self, body::Error> {
231        self.try_end_with(|| {
232            if state.is_finished() {
233                Ok(state.logical_len())
234            } else {
235                state.try_end()
236            }
237        })
238    }
239
240    fn try_end_with(
241        self,
242        finish: impl FnOnce() -> Result<usize, body::Error>,
243    ) -> body::Transfer<Self, body::Error> {
244        let State::Building(build) = self.state else {
245            return body::Transfer::Rejected {
246                build: self,
247                error: body::Error::InputFinished,
248            };
249        };
250        let len = match finish() {
251            Ok(len) => len,
252            Err(error) => {
253                return body::Transfer::Rejected {
254                    build: Self::new(build),
255                    error,
256                };
257            }
258        };
259        body::Transfer::Accepted(Self::new(build.finish_input(len)))
260    }
261
262    pub(super) fn try_drive<E: credit::Evidence, O: credit::Origin>(
263        self,
264        mut state: ingress::State<O>,
265        ledger: &mut credit::raw::Ledger<'_, O>,
266        event: &mut E,
267        budget: NonZero<usize>,
268    ) -> Drive<'d, B, O> {
269        let build = match self.state {
270            State::Building(build) => build,
271            State::Ready(output) => {
272                return match state.try_ready() {
273                    Ok(state) => Drive::Ready(Ready::<B, O> {
274                        state,
275                        output,
276                        lifetime: self.lifetime,
277                    }),
278                    Err(state) => Drive::Rejected {
279                        stream: Stream::from_parts(state, Self::from_output(output)),
280                        error: body::Error::InputIncomplete,
281                    },
282                };
283            }
284        };
285        match build.step(budget) {
286            body::Step::NeedInput { build, released } => {
287                let flow = Self::new(build);
288                let result = ingress::Turn::new(&mut state, ledger, &mut *event)
289                    .try_release(released, credit::Disposition::Discarded);
290                let error = match result {
291                    Ok(()) if state.is_finished() => Error::LengthMismatch,
292                    Ok(()) => return Drive::NeedInput(Stream::from_parts(state, flow)),
293                    Err(error) => Error::Credit(error),
294                };
295                Drive::Failed {
296                    stream: Stream::from_parts(state, flow),
297                    error,
298                }
299            }
300            body::Step::Pending { build, released } => {
301                let flow = Self::new(build);
302                let result = ingress::Turn::new(&mut state, ledger, &mut *event)
303                    .try_release(released, credit::Disposition::Discarded);
304                match result {
305                    Ok(()) => Drive::Pending(Stream::from_parts(state, flow)),
306                    Err(error) => Drive::Failed {
307                        stream: Stream::from_parts(state, flow),
308                        error: Error::Credit(error),
309                    },
310                }
311            }
312            body::Step::Ready { output, released } => {
313                if let Err(error) = ingress::Turn::new(&mut state, ledger, &mut *event)
314                    .try_release(released, credit::Disposition::Discarded)
315                {
316                    return Drive::Failed {
317                        stream: Stream::from_parts(state, Self::from_output(output)),
318                        error: Error::Credit(error),
319                    };
320                }
321                let state = match state.try_ready() {
322                    Ok(state) => state,
323                    Err(state) => {
324                        return Drive::Rejected {
325                            stream: Stream::from_parts(state, Self::from_output(output)),
326                            error: body::Error::InputIncomplete,
327                        };
328                    }
329                };
330                event.build_ready();
331                Drive::Ready(Ready::<B, O> {
332                    state,
333                    output,
334                    lifetime: self.lifetime,
335                })
336            }
337            body::Step::Failed {
338                build,
339                error,
340                released,
341            } => {
342                let flow = Self::new(build);
343                let error = match ingress::Turn::new(&mut state, ledger, &mut *event)
344                    .try_release(released, credit::Disposition::Discarded)
345                {
346                    Ok(()) => Error::Build(error),
347                    Err(release) => Error::Release {
348                        build: error,
349                        error: release,
350                    },
351                };
352                Drive::Failed {
353                    stream: Stream::from_parts(state, flow),
354                    error,
355                }
356            }
357        }
358    }
359}
360
361const _: () = {
362    assert!(core::mem::size_of::<Error<u8>>() == 4 * core::mem::size_of::<usize>());
363};
sark_core/body/flow.rs

sark_core/body/
flow.rs
