Skip to content

Commit ce2d2d3

Browse files
committed
refactor(read): separate framing actions from consumption
1 parent 2b43c03 commit ce2d2d3

5 files changed

Lines changed: 593 additions & 534 deletions

File tree

‎README.md‎

Lines changed: 45 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -10,11 +10,9 @@ for defining frame boundaries, decoding frames, and encoding values, together
1010
with buffered readers and writers that use those implementations.
1111

1212
The default configuration supports stable Rust. The `nightly` feature switches
13-
`FramedRead` to `Read::read_buf` and requires nightly Rust. This uses the
14-
standard-library features `read_buf`, `core_io_borrowed_buf`, and
15-
`borrowed_buf_init` to read into uninitialized space and track its
16-
initialization state. This can avoid initializing the read destination for
17-
sources that support this API.
13+
`FramedRead` to `Read::read_buf` and requires nightly Rust. Sources with their
14+
own `read_buf` implementation can avoid initializing the destination before
15+
reading.
1816

1917
To enable the nightly read path, use
2018
`cargo +nightly build --features nightly`.
@@ -85,24 +83,46 @@ To decode with `FramedRead`, implement `Framer` and at least one of
8583
boundaries can depend on format state, such as quoting or length prefixes.
8684
Implement `Encoder<Item>` to write an Item's encoded bytes.
8785

88-
`Framer::frame` returns a `Frame` describing the next action.
89-
90-
- `ready_split(range, addition)`: Selects `range` and consumes through its end
91-
plus `addition` bytes; `decode_bytes()` transfers the selection without
92-
copying.
93-
- `ready_copy(range, consume)`: Selects `range` independently of consuming the
94-
first `consume` bytes, allowing adjacent frames to overlap; `decode_bytes()`
95-
copies the framed bytes.
96-
- `skip(amount)`: Skips the first `amount` bytes of the current input without
97-
producing an Item.
98-
- `need_more(consume)`: Consumes the first `consume` bytes of the current input
99-
before requesting more data.
100-
101-
Both ready actions can be used with any of the three decoder traits. All
102-
selections and consumption counts are relative to the input passed to `frame`
103-
and must fit within it, including counts passed to `skip` and `need_more`.
104-
Empty Items and zero consumption are allowed. `skip(amount)` immediately
105-
resumes framing. `need_more(consume)` attempts a read before framing again.
86+
`Framer::frame` returns a `Frame` with a consumption setting and a `FrameAction`:
87+
`Ready(start, end)` decodes one Item, `Reframe` immediately frames the remaining
88+
input, and `NeedMore` attempts a read before framing again. Inspect the complete
89+
decision through `consume()` and `action()`.
90+
91+
- `ready(range)`: Requests selection of the range and consumption through its
92+
end by default.
93+
- `reframe()`: Creates a `Reframe` decision with zero consumption.
94+
- `need_more()`: Creates a `NeedMore` decision with zero consumption.
95+
- `with_consume(consume)`: Returns the decision with an explicit consumption count,
96+
independently of the selection, allowing adjacent frames to overlap.
97+
- `with_action(action)`: Returns the decision with another action, keeping any
98+
explicit consumption count.
99+
100+
Use `FrameAction::ready(range)` to construct a ready action, including when
101+
replacing an action with `with_action`.
102+
103+
For example, `ready(1..4)` consumes 4 bytes, and `ready(..)` consumes all input
104+
supplied to the framer when the reader applies the decision. `consume()` returns
105+
`None` for the action's default, or
106+
`Some(n)` after setting an explicit count. Set it separately when needed:
107+
108+
```rust
109+
use decen::read::Frame;
110+
111+
let frame = Frame::need_more().with_consume(4);
112+
113+
let frame = Frame::ready(..4).with_consume(0); // Retain all input.
114+
```
115+
116+
`Ready` works with all three decoder traits. `decode_bytes()` transfers the
117+
selection without copying when consumption covers the prefix through its end;
118+
otherwise, it copies the selection into a new `BytesMut`. `decode_view()` and
119+
`decode_owned()` always pass a slice without this copy.
120+
121+
The reader interprets selections relative to the input before consumption and
122+
rejects ranges or consumption counts outside that input. Empty Items and zero
123+
consumption are allowed. Reading or decoding errors do not undo consumption.
124+
For example, selecting `1..4` and requesting consumption of 5 bytes selects
125+
`abc` from `<abc>tail` and leaves `tail`.
106126

107127
The decoder receives framed bytes. A view Item can borrow fields directly;
108128
converted fields, such as unescaped text, can use storage owned by the Item.
@@ -131,9 +151,9 @@ impl Framer for ByteCodec {
131151

132152
unsafe fn frame(&mut self, src: &[u8]) -> io::Result<Frame> {
133153
Ok(if src.is_empty() {
134-
Frame::need_more(0)
154+
Frame::need_more()
135155
} else {
136-
Frame::ready_split(..1, 0)
156+
Frame::ready(..1)
137157
})
138158
}
139159
}

‎benches/README.md‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,10 @@ Oversized Items repeat 16 KiB + 3 B, 64 KiB + 5 B, 100 KiB + 7 B,
3333
Read sizes include the final newline. Write sizes are the bytes written per
3434
Item. Initial buffer capacity is 8 KiB.
3535

36+
Each workload mixes its listed sizes in one dataset. Its table reports the
37+
whole mixture, not individual results for each size. The oversized workload
38+
exercises Items larger than the initial buffer, including buffer growth.
39+
3640
To check read results without timing, append `-- --test` to either read command.
3741

3842
Reading the tables:

‎benches/read.rs‎

Lines changed: 42 additions & 54 deletions
Original file line numberDiff line numberDiff line change
@@ -16,7 +16,9 @@ use std::mem;
1616
use std::time::{Instant, SystemTime, UNIX_EPOCH};
1717

1818
use bytes::{Buf, BytesMut};
19-
use decen::read::{BytesDecoder, Frame, FrameKind, FramedRead, Framer, OwnedDecoder, ViewDecoder};
19+
use decen::read::{
20+
BytesDecoder, Frame, FrameAction, FramedRead, Framer, OwnedDecoder, ViewDecoder,
21+
};
2022
use memchr::memchr;
2123

2224
#[path = "src/table.rs"]
@@ -26,11 +28,10 @@ const CAPACITY: usize = 8 << 10;
2628
// One complete rotation gives every method each timing position once.
2729
const SAMPLES: usize = METHODS.len();
2830
const ROUNDS: usize = 4;
29-
const METHODS: [&str; 13] = [
31+
const METHODS: [&str; 12] = [
3032
"FramedRead::decode_bytes()",
31-
"Frame::ready_split(..end, 0); FramedRead::decode_view()",
32-
"Frame::ready_copy(..end, end); FramedRead::decode_view()",
33-
"FramedRead<_, Lines<false>>::decode_owned()",
33+
"FramedRead::decode_view()",
34+
"FramedRead::decode_owned()",
3435
"BufReader::fill_buf() + consume(n)",
3536
"BufReader::fill_buf() + finish_owned::<true>() + consume(n)",
3637
"BufReader::read_until(b'\\n', &mut Vec::new())",
@@ -74,17 +75,17 @@ fn finish_owned<const REUSE: bool>(partial: &mut Vec<u8>, frame: &[u8]) -> Vec<u
7475

7576
// Remember how much of the pending frame has already been searched for LF.
7677
#[derive(Default)]
77-
struct Lines<const COPY: bool>(usize);
78-
impl<const COPY: bool> Framer for Lines<COPY> {
78+
struct Lines(usize);
79+
impl Framer for Lines {
7980
type Error = io::Error;
8081
unsafe fn frame(&mut self, src: &[u8]) -> io::Result<Frame> {
8182
Ok(match self.line_end(src) {
82-
Some(end) => Self::ready(end),
83-
None => Frame::need_more(0),
83+
Some(end) => Frame::ready(..end),
84+
None => Frame::need_more(),
8485
})
8586
}
8687
}
87-
impl<const COPY: bool> Lines<COPY> {
88+
impl Lines {
8889
// Share this scanner with the direct BytesMut baseline so only buffering
8990
// and frame delivery differ, not the LF search or partial-line bookkeeping.
9091
#[inline(always)]
@@ -101,16 +102,6 @@ impl<const COPY: bool> Lines<COPY> {
101102
None
102103
}
103104
}
104-
#[inline(always)]
105-
fn ready(end: usize) -> Frame {
106-
// Fix the decision for the whole run. Both choices select and consume
107-
// the same bytes; decode_view borrows them in either case.
108-
if COPY {
109-
Frame::ready_copy(..end, end)
110-
} else {
111-
Frame::ready_split(..end, 0)
112-
}
113-
}
114105

115106
// Inline so the framing loop can simplify the returned decision's bounds.
116107
#[inline(always)]
@@ -127,19 +118,19 @@ impl<const COPY: bool> Lines<COPY> {
127118
}
128119
}
129120
}
130-
impl<const COPY: bool> BytesDecoder for Lines<COPY> {
121+
impl BytesDecoder for Lines {
131122
type Item = BytesMut;
132123
unsafe fn decode_bytes(&mut self, frame: BytesMut) -> io::Result<Self::Item> {
133124
Ok(frame)
134125
}
135126
}
136-
impl<const COPY: bool> ViewDecoder for Lines<COPY> {
127+
impl ViewDecoder for Lines {
137128
type Item<'a> = &'a [u8];
138129
unsafe fn decode_view<'a>(&mut self, frame: &'a [u8]) -> io::Result<Self::Item<'a>> {
139130
Ok(frame)
140131
}
141132
}
142-
impl<const COPY: bool> OwnedDecoder for Lines<COPY> {
133+
impl OwnedDecoder for Lines {
143134
type Item = Vec<u8>;
144135
unsafe fn decode_owned(&mut self, frame: &[u8]) -> io::Result<Self::Item> {
145136
Ok(copy_frame(frame))
@@ -159,18 +150,18 @@ impl<const REUSE: bool> Framer for AccumulatingLines<REUSE> {
159150

160151
unsafe fn frame(&mut self, src: &[u8]) -> io::Result<Frame> {
161152
if let Some(end) = find_line_end(src) {
162-
Ok(Frame::ready_split(..end, 0))
153+
Ok(Frame::ready(..end))
163154
} else {
164155
self.partial.extend_from_slice(src);
165-
Ok(Frame::need_more(src.len()))
156+
Ok(Frame::need_more().with_consume(src.len()))
166157
}
167158
}
168159

169160
unsafe fn frame_eof(&mut self, src: &[u8]) -> io::Result<Frame> {
170161
// SAFETY: frame_eof forwards the same input and framing state.
171162
let frame = unsafe { self.frame(src)? };
172163
// The default EOF check cannot see bytes already consumed into partial.
173-
if frame.kind() == FrameKind::NeedMore && !self.partial.is_empty() {
164+
if frame.action() == &FrameAction::NeedMore && !self.partial.is_empty() {
174165
return Err(io::Error::new(
175166
io::ErrorKind::UnexpectedEof,
176167
"incomplete line",
@@ -202,10 +193,8 @@ fn observe<const VERIFY: bool>(frame: &[u8], totals: &mut (usize, usize)) {
202193
}
203194

204195
#[inline(never)]
205-
fn framed<R: Read, const MODE: usize, const COPY: bool, const VERIFY: bool>(
206-
source: R,
207-
) -> (usize, usize) {
208-
let mut reader = FramedRead::with_capacity(source, Lines::<COPY>::default(), CAPACITY);
196+
fn framed<R: Read, const MODE: usize, const VERIFY: bool>(source: R) -> (usize, usize) {
197+
let mut reader = FramedRead::with_capacity(source, Lines::default(), CAPACITY);
209198
let mut totals = (0, 0);
210199
match MODE {
211200
0 => {
@@ -291,7 +280,7 @@ fn standard<R: Read, const MODE: usize, const REUSE: bool, const VERIFY: bool>(
291280
#[inline(never)]
292281
fn bytes_mut<R: Read, const MODE: usize, const VERIFY: bool>(mut source: R) -> (usize, usize) {
293282
let mut buffer = BytesMut::with_capacity(CAPACITY);
294-
let mut lines = Lines::<false>::default();
283+
let mut lines = Lines::default();
295284
let mut totals = (0, 0);
296285
loop {
297286
if let Some(end) = lines.line_end(&buffer) {
@@ -350,19 +339,18 @@ fn bytes_mut<R: Read, const MODE: usize, const VERIFY: bool>(mut source: R) -> (
350339

351340
fn run<R: Read, const VERIFY: bool>(source: R, mode: usize) -> (usize, usize) {
352341
match mode {
353-
0 => framed::<_, 0, false, VERIFY>(source),
354-
1 => framed::<_, 1, false, VERIFY>(source),
355-
2 => framed::<_, 1, true, VERIFY>(source),
356-
3 => framed::<_, 2, false, VERIFY>(source),
357-
4 => standard::<_, 0, true, VERIFY>(source),
358-
5 => standard::<_, 1, true, VERIFY>(source),
359-
6 => standard::<_, 2, true, VERIFY>(source),
360-
7 => bytes_mut::<_, 0, VERIFY>(source),
361-
8 => bytes_mut::<_, 1, VERIFY>(source),
362-
9 => bytes_mut::<_, 2, VERIFY>(source),
363-
10 => accumulating::<_, false, VERIFY>(source),
364-
11 => accumulating::<_, true, VERIFY>(source),
365-
12 => standard::<_, 1, false, VERIFY>(source),
342+
0 => framed::<_, 0, VERIFY>(source),
343+
1 => framed::<_, 1, VERIFY>(source),
344+
2 => framed::<_, 2, VERIFY>(source),
345+
3 => standard::<_, 0, true, VERIFY>(source),
346+
4 => standard::<_, 1, true, VERIFY>(source),
347+
5 => standard::<_, 2, true, VERIFY>(source),
348+
6 => bytes_mut::<_, 0, VERIFY>(source),
349+
7 => bytes_mut::<_, 1, VERIFY>(source),
350+
8 => bytes_mut::<_, 2, VERIFY>(source),
351+
9 => accumulating::<_, false, VERIFY>(source),
352+
10 => accumulating::<_, true, VERIFY>(source),
353+
11 => standard::<_, 1, false, VERIFY>(source),
366354
_ => unreachable!(),
367355
}
368356
}
@@ -409,23 +397,23 @@ fn compare<R: Read + Seek>(
409397
}
410398
// Compare identical output types; Vec and BytesMut have different
411399
// ownership and allocation costs even though both own their bytes.
412-
let borrowed_best = [1, 2, 4, 7]
400+
let borrowed_best = [1, 3, 6]
413401
.into_iter()
414402
.map(|mode| times[mode][SAMPLES / 2])
415403
.fold(f64::INFINITY, f64::min);
416-
let vec_best = [3, 5, 6, 8, 10, 11, 12]
404+
let vec_best = [2, 4, 5, 7, 9, 10, 11]
417405
.into_iter()
418406
.map(|mode| times[mode][SAMPLES / 2])
419407
.fold(f64::INFINITY, f64::min);
420-
let bytes_best = times[0][SAMPLES / 2].min(times[9][SAMPLES / 2]);
408+
let bytes_best = times[0][SAMPLES / 2].min(times[8][SAMPLES / 2]);
421409
let mut rows = Vec::new();
422-
// Keep the three Lines variants adjacent without changing timing order.
423-
for mode in [1, 2, 4, 7, 3, 10, 11, 5, 12, 8, 6, 0, 9] {
410+
// Group matching result types without changing timing order.
411+
for mode in [1, 3, 6, 2, 9, 10, 4, 11, 7, 5, 0, 8] {
424412
let method = METHODS[mode];
425413
let values = &times[mode];
426414
let (item, best) = match mode {
427-
1 | 2 | 4 | 7 => ("borrowed", borrowed_best),
428-
0 | 9 => ("BytesMut", bytes_best),
415+
1 | 3 | 6 => ("borrowed", borrowed_best),
416+
0 | 8 => ("BytesMut", bytes_best),
429417
_ => ("Vec<u8>", vec_best),
430418
};
431419
rows.push(table::Row {
@@ -568,9 +556,9 @@ fn main() -> io::Result<()> {
568556
Read request sizes can differ, and no artificial short-read limit is applied."
569557
);
570558
println!(
571-
"Framing: decode_view compares ready_split(..end, 0) with \
572-
ready_copy(..end, end); both return borrowed slices without \
573-
copying. decode_bytes and decode_owned use ready_split(..end, 0)."
559+
"Framing: all FramedRead paths select ..end and request consumption of end bytes. \
560+
decode_bytes transfers the selection; decode_view and \
561+
decode_owned receive a slice without copying."
574562
);
575563
println!(
576564
"All explicit framing paths share find_line_end(); all fresh Vec \

0 commit comments

Comments
 (0)