Skip to content

Copy the rest of a suspended text lexeme once, not at every suspension #13

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
robobun wants to merge 1 commit into
base: bun
Choose a base branch
from
Open

Copy the rest of a suspended text lexeme once, not at every suspension #13

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
270 changes: 253 additions & 17 deletions src/rewritable_units/text_decoder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ use crate::AsciiCompatibleEncoding;
use crate::base::{Bytes, SourceLocation, Spanned};
use crate::rewriter::RewritingError;
use encoding_rs::{CoderResult, Decoder, Encoding, UTF_8};
use std::mem;

const DEFAULT_BUFFER_LEN: usize = if cfg!(test) { 13 } else { 1024 };

Expand All @@ -12,17 +13,21 @@ pub(crate) struct TextDecoder {
text_buffer: String,
/// Where the current `feed_text` call was when a text handler suspended
/// (`None` otherwise). Recorded *only* on the error path so the success
/// path pays nothing.
/// path pays nothing. `resume_feed` keeps its record here while it runs.
suspended: Option<TextFeedSuspension>,
}

/// Where a [`TextDecoder::feed_text`] call was when the text handler for one
/// of its chunks suspended. Lets [`TextDecoder::resume_feed`] finish decoding
/// the rest of that text lexeme once the parked chunk completes. The
/// undecoded tail is copied because it pointed into the `write()` call's
/// input buffer, which is gone by the time the resume runs.
/// input buffer, which is gone by the time the resume runs. It is copied
/// once per lexeme: when a later chunk of the same lexeme suspends, the same
/// buffer is parked again and only `offset` moves.
pub(crate) struct TextFeedSuspension {
remaining: Vec<u8>,
remaining: Box<[u8]>,
/// Where the undecoded bytes start in `remaining`.
offset: usize,
next_source_location_bytes_start: usize,
last_in_text_node: bool,
/// `true` if the chunk that suspended was this feed's final one; the
Expand Down Expand Up @@ -78,7 +83,7 @@ impl TextDecoder {
/// right after the chunk that suspended.
pub fn resume_feed(
&mut self,
suspension: TextFeedSuspension,
mut suspension: TextFeedSuspension,
output_handler: &mut OutputHandlerCallback<'_>,
) -> Result<(), RewritingError> {
if suspension.finished {
Expand All @@ -93,12 +98,67 @@ impl TextDecoder {
return Ok(());
}

self.feed_decoder_loop(
&suspension.remaining,
suspension.next_source_location_bytes_start,
suspension.last_in_text_node,
// The loop gets the bytes on loan. The record waits in the slot
// without them, with `offset` at their end. A chunk that suspends
// again finds the record there (see `park`) and moves `offset` back
// to the unread tail, so the rest is not copied a second time.
let remaining = mem::take(&mut suspension.remaining);
let start = mem::replace(&mut suspension.offset, remaining.len());
let next_source_location_bytes_start = suspension.next_source_location_bytes_start;
let last_in_text_node = suspension.last_in_text_node;
self.suspended = Some(suspension);

let res = self.feed_decoder_loop(
remaining.get(start..).unwrap_or_default(),
next_source_location_bytes_start,
last_in_text_node,
output_handler,
)
);

match &mut self.suspended {
Some(parked) if matches!(res, Err(RewritingError::Suspended)) => {
parked.remaining = remaining;
}
_ => self.suspended = None,
}

res
}

/// Records where the feed stopped when the text handler for a chunk
/// suspended. `unread` is the input that the decoder has not read yet.
/// Any other error ends the rewrite, so nothing is recorded for it.
#[cold]
#[inline(never)]
fn park(
slot: &mut Option<TextFeedSuspension>,
res: &Result<(), RewritingError>,
unread: &[u8],
next_source_location_bytes_start: usize,
last_in_text_node: bool,
finished: bool,
) {
if !matches!(res, Err(RewritingError::Suspended)) {
return;
}

let (remaining, offset) = match slot.take() {
// `resume_feed` left its record here: `unread` is the tail of
// the bytes it owns, and it puts them back into the new record.
Some(lent) => {
debug_assert!(lent.remaining.is_empty());
(lent.remaining, lent.offset - unread.len())
}
None => (Box::from(unread), 0),
};

*slot = Some(TextFeedSuspension {
remaining,
offset,
next_source_location_bytes_start,
last_in_text_node,
finished,
});
}

#[inline(never)]
Expand All @@ -123,12 +183,14 @@ impl TextDecoder {

let res = (output_handler)(utf8_text, really_last, encoding, source_location);
if res.is_err() {
self.suspended = Some(TextFeedSuspension {
remaining: rest.to_vec(),
Self::park(
&mut self.suspended,
&res,
rest,
next_source_location_bytes_start,
last_in_text_node,
finished: really_last,
});
really_last,
);
}
res?;

Expand Down Expand Up @@ -189,12 +251,14 @@ impl TextDecoder {
source_location,
);
if res.is_err() {
self.suspended = Some(TextFeedSuspension {
remaining: raw_input.get(read..).unwrap_or_default().to_vec(),
Self::park(
&mut self.suspended,
&res,
raw_input.get(read..).unwrap_or_default(),
next_source_location_bytes_start,
last_in_text_node,
finished: finished_decoding,
});
finished_decoding,
);
}
res?;
}
Expand Down Expand Up @@ -254,3 +318,175 @@ impl TextDecoder {
}
}
}

#[cfg(test)]
mod tests {
use super::*;
use std::ops::Range;

/// What the output handler got for one chunk: the text,
/// `last_in_text_node` and the source range.
type Chunk = (String, bool, Range<usize>);

/// The parked bytes at one suspension: the address and the length of the
/// buffer, and where the undecoded bytes start in it.
type Parked = (*const u8, usize, usize);

fn new_decoder() -> TextDecoder {
TextDecoder::new(AsciiCompatibleEncoding::utf_8())
}

/// Completes a feed whose handler may suspend. Returns the parked bytes
/// at each suspension.
fn resume_until_done(
decoder: &mut TextDecoder,
mut res: Result<(), RewritingError>,
handler: &mut OutputHandlerCallback<'_>,
) -> Vec<Parked> {
let mut parked_at = Vec::new();
while matches!(res, Err(RewritingError::Suspended)) {
let parked = decoder
.take_suspended()
.expect("a suspended feed is parked");
parked_at.push((
parked.remaining.as_ptr(),
parked.remaining.len(),
parked.offset,
));
res = decoder.resume_feed(parked, handler);
}
res.expect("the feed completes");
assert!(decoder.take_suspended().is_none());
parked_at
}

/// Feeds `lexemes` as the lexemes of one text node, then ends the node.
/// The handler suspends on every `every`th chunk (on none for 0).
/// Returns the chunks and, for each lexeme, what `resume_until_done`
/// returns.
fn feed(lexemes: &[Vec<u8>], every: usize) -> (Vec<Chunk>, Vec<Vec<Parked>>) {
let input_len: usize = lexemes.iter().map(Vec::len).sum();
let mut chunks = Vec::new();
let mut handler = |text: &str, last: bool, _: &'static Encoding, at: SourceLocation| {
assert!(chunks.len() <= input_len, "the feed does not advance");
chunks.push((text.to_owned(), last, at.bytes()));
if every != 0 && chunks.len() % every == 0 {
Err(RewritingError::Suspended)
} else {
Ok(())
}
};

let mut decoder = new_decoder();
let mut parked_at = Vec::new();
let mut start = 0;
for lexeme in lexemes {
let res =
decoder.feed_text(Spanned::new(start, Bytes::new(lexeme)), false, &mut handler);
parked_at.push(resume_until_done(&mut decoder, res, &mut handler));
start += lexeme.len();
}
let res = decoder.flush_pending(&mut handler);
resume_until_done(&mut decoder, res, &mut handler);

(chunks, parked_at)
}

/// ASCII text that fills `chunks` chunks of the decoder. Its period is
/// not a multiple of the chunk length, so each chunk has its own text.
fn ascii(chunks: usize) -> Vec<u8> {
let period = b"abcdefghijklmnopqrstuvwxyz0123456789_";
let len = chunks * DEFAULT_BUFFER_LEN;
period.iter().copied().cycle().take(len).collect()
}

/// Text nodes whose lexemes take several chunks of the streaming decoder.
fn text_nodes() -> Vec<Vec<Vec<u8>>> {
let wide = "\u{20AC}\u{2192}\u{221A}\u{2603}\u{2764}\u{2600}\u{266B}".repeat(4);
vec![
// A byte that is not UTF-8 comes first: no fast path.
vec![[&b"\xE9"[..], &ascii(8)].concat()],
// The fast path takes the valid prefix, the decoder takes the rest.
vec![[&ascii(2), &b"\xE9"[..], &ascii(8)].concat()],
// Characters of 3 bytes do not fit the decoder's buffer evenly.
vec![[&b"\xE9"[..], wide.as_bytes()].concat()],
// Later lexemes of a text node always take the decoder. A
// character is split between the last two.
vec![
ascii(3),
[&ascii(5), &b"\xE2\x82"[..]].concat(),
[&b"\xAC"[..], &ascii(5)].concat(),
],
]
}

#[test]
fn suspended_feed_yields_the_same_chunks() {
for lexemes in text_nodes() {
let (expected, _) = feed(&lexemes, 0);
assert!(expected.len() > 8);

for every in [1, 2, 3, 7] {
let (chunks, _) = feed(&lexemes, every);
assert_eq!(chunks, expected, "every: {every}");
}
}
}

#[test]
fn rest_of_lexeme_is_copied_once() {
for lexemes in text_nodes() {
let (_, parked_at) = feed(&lexemes, 1);
let most_suspensions = parked_at.iter().map(Vec::len).max().unwrap();
assert!(most_suspensions > 4);

for parked_at in parked_at {
for pair in parked_at.windows(2) {
let (buffer, len, offset) = pair[0];
let (next_buffer, next_len, next_offset) = pair[1];
assert_eq!(
(next_buffer, next_len),
(buffer, len),
"a later suspension copied or moved the rest of the lexeme"
);
assert!(next_offset > offset);
}
}
}
}

#[test]
fn handler_error_parks_nothing() {
let lexeme = [&b"\xE9"[..], &ascii(8)].concat();

for suspend_first in [false, true] {
let mut calls = 0;
let mut handler = |_: &str, _: bool, _: &'static Encoding, _: SourceLocation| {
calls += 1;
match calls {
1 if suspend_first => Err(RewritingError::Suspended),
2 => Err(RewritingError::ContentHandlerError("failed".into())),
_ => Ok(()),
}
};

let mut decoder = new_decoder();
let mut res =
decoder.feed_text(Spanned::new(0, Bytes::new(&lexeme)), false, &mut handler);
if suspend_first {
assert!(matches!(res, Err(RewritingError::Suspended)));
let parked = decoder.take_suspended().unwrap();
res = decoder.resume_feed(parked, &mut handler);
}

assert!(matches!(res, Err(RewritingError::ContentHandlerError(_))));
assert!(decoder.take_suspended().is_none());
}
}

/// The record is inline in every rewriter, suspended or not.
#[test]
fn suspension_record_size() {
assert!(size_of::<TextFeedSuspension>() <= 5 * size_of::<usize>());
}
}
Loading
Loading