From 253cb47ad8fc28b17049be1bad95c1700badc456 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Micka=C3=ABl=20R=C3=A9mond?= Date: Sat, 14 Nov 2026 14:50:49 +0100 Subject: [PATCH] fix(runtime): bound and account for binary copy work --- CHANGELOG.md | 16 ++ crates/nera-runtime/src/ffi/binary_ops.rs | 28 +- .../nera-runtime/src/ffi/binary_ops/copy.rs | 265 ++++++++++++++++++ crates/nera-runtime/src/native_work.rs | 7 +- tests/README.md | 2 + tests/binary_copy.erl | 64 +++++ tests/binary_copy_scheduler.erl | 45 +++ 7 files changed, 398 insertions(+), 29 deletions(-) create mode 100644 crates/nera-runtime/src/ffi/binary_ops/copy.rs create mode 100644 tests/binary_copy.erl create mode 100644 tests/binary_copy_scheduler.erl diff --git a/CHANGELOG.md b/CHANGELOG.md index dced86f..fd5fae5 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,22 @@ All notable milestones for the nera project are documented in this file. +## Oversized binary copies raise Erlang errors + +- `binary:copy/2` checks result-size multiplication and the binary header's + bit-length limit before reserving storage. Impossible sizes and refused + reservations raise `system_limit` instead of panicking across the native + boundary and aborting the program. +- Counts accept unsigned machine-word integers, including boxed integers; + invalid counts and non-binary inputs raise `badarg`, as on OTP, even when + the result would otherwise be empty. +- Copies charge reductions for output bytes and yield between bounded chunks. + A small input repeated many times no longer monopolizes a scheduler. The + completed output reuses its allocation rather than copying it again. +- Allocation and garbage collection remain synchronous. A successful virtual + memory reservation does not guarantee sufficient physical memory; this + change does not contain operating-system out-of-memory termination. + ## Stacktraces list the raising function first - Captured exception stacktraces now list the raising function before its diff --git a/crates/nera-runtime/src/ffi/binary_ops.rs b/crates/nera-runtime/src/ffi/binary_ops.rs index 25bf63d..2306275 100644 --- a/crates/nera-runtime/src/ffi/binary_ops.rs +++ b/crates/nera-runtime/src/ffi/binary_ops.rs @@ -13,7 +13,9 @@ use crate::term::*; use super::helpers::*; mod bit_window; +mod copy; mod float_segment; +pub use copy::nera_rt_binary_copy_2; use float_segment::{encode_half, decode_half}; use bit_window::{BitReader, BitWriter}; @@ -1904,32 +1906,6 @@ pub unsafe extern "C" fn nera_rt_binary_copy_1( with_ctx_heap(ctx, |heap| heap.alloc_binary_auto(bytes).0) } -/// binary:copy/2 — return a binary consisting of Subject repeated N times. -/// copy(<<>>, _) → <<>>, copy(_, 0) → <<>>. -#[no_mangle] -pub unsafe extern "C" fn nera_rt_binary_copy_2( - ctx: *mut ProcessContext, - subject_term: u64, - n_term: u64, -) -> u64 { - let bytes = match unsafe { binary_data(ErlTerm(subject_term)) } { - Some((ptr, len)) => unsafe { std::slice::from_raw_parts(ptr, len) }, - None => return raise_badarg(ctx), - }; - let n = match ErlTerm(n_term).as_small_int() { - Some(n) if n >= 0 => n as usize, - _ => return raise_badarg(ctx), - }; - if n == 0 || bytes.is_empty() { - return with_ctx_heap(ctx, |heap| heap.alloc_binary_auto(&[]).0); - } - let mut result = Vec::with_capacity(bytes.len() * n); - for _ in 0..n { - result.extend_from_slice(bytes); - } - with_ctx_heap(ctx, |heap| heap.alloc_binary_auto(&result).0) -} - // ===================================================================== // UTF-8/16/32 binary segment operations // ===================================================================== diff --git a/crates/nera-runtime/src/ffi/binary_ops/copy.rs b/crates/nera-runtime/src/ffi/binary_ops/copy.rs new file mode 100644 index 0000000..7d60080 --- /dev/null +++ b/crates/nera-runtime/src/ffi/binary_ops/copy.rs @@ -0,0 +1,265 @@ +//! Checked, cooperative binary:copy/2. The shared adapter roots the input +//! across suspension and adopts the completed output without another copy. +use super::super::{helpers::*, native_work}; +use crate::{native_work::BinaryWork, process::ProcessContext, term::*}; + +const BYTES_PER_REDUCTION: usize = 64; +const CHUNK_BYTES: usize = 64 * 1024; + +/// OTP accepts an unsigned machine-word count, including boxed integers, but +/// rejects larger integers even when the subject is empty. +/// # Safety +/// The count is a live, initialized, immutable and unmoved term. +unsafe fn repetitions(term: ErlTerm) -> Option { + if let Some(n) = term.as_small_int() { + return usize::try_from(n).ok(); + } + let ptr = term.as_boxed_ptr()?; + let header = BoxedHeader(*ptr); + if header.subtype() != BOXED_BIGINT + || header.arity() != 1 + || *ptr.add(BIGINT_SIGN_WORD_INDEX) != BIGINT_SIGN_NON_NEGATIVE + { + return None; + } + usize::try_from(*ptr.add(BIGINT_FIRST_LIMB_WORD_INDEX)).ok() +} + +fn result_len(input_len: usize, count: usize) -> Option { + let len = input_len.checked_mul(count)?; + // Refc binary descriptors encode the bit length in the boxed header. + let bits = u64::try_from(len).ok()?.checked_mul(8)?; + (bits <= u64::MAX >> BOXED_TAG_BITS).then_some(len) +} + +struct Copy { + output: Vec, + len: usize, +} + +impl Copy { + fn new(input_len: usize, count: usize) -> Option { + let len = result_len(input_len, count)?; + let mut output = Vec::new(); + // Capacity overflow and allocator refusal must become Erlang errors, + // not panics/aborts at the extern "C" boundary. Allocation itself is + // synchronous; a successful reservation is not a physical-memory bound. + output.try_reserve_exact(len).ok()?; + Some(Self { output, len }) + } +} + +impl BinaryWork for Copy { + fn bytes_per_reduction(&self) -> usize { + BYTES_PER_REDUCTION + } + fn max_bytes(&self) -> usize { + CHUNK_BYTES + } + fn offset(&self) -> usize { + self.output.len() + } + fn output(&mut self) -> &mut Vec { + &mut self.output + } + fn step(&mut self, input: &[u8], limit: usize) -> Result { + let end = self.output.len() + limit.min(self.len - self.output.len()); + while self.output.len() < end { + let offset = self.output.len() % input.len(); + let count = (end - self.output.len()).min(input.len() - offset); + self.output + .extend_from_slice(&input[offset..offset + count]); + } + Ok(self.output.len() == self.len) + } +} + +/// Repeat a byte-aligned binary, validating both arguments before empty-result +/// shortcuts. Work is charged by output bytes, not the (possibly tiny) input. +/// # Safety +/// Generated code supplies an exclusive live context and live, unmoved terms. +#[no_mangle] +pub unsafe extern "C" fn nera_rt_binary_copy_2( + ctx: *mut ProcessContext, + subject: u64, + count: u64, +) -> u64 { + if !ErlTerm(subject).is_binary() { + return raise_badarg(ctx); + } + let Some(count) = repetitions(ErlTerm(count)) else { + return raise_badarg(ctx); + }; + let (_, len) = binary_data(ErlTerm(subject)).unwrap(); + let Some(work) = Copy::new(len, count) else { + return raise_error(ctx, ErlTerm::atom(crate::atom_table::ATOM_SYSTEM_LIMIT).0); + }; + native_work::start(ctx, subject, work) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::{ + binary::BinaryData, + native_work::budget, + process::{Process, ProcessId, STATUS_OK, STATUS_YIELD}, + }; + use std::sync::Arc; + + // Intent: oversized copies must raise rather than overflow or truncate. + // These are actual request sizes; no large input allocation is needed. + #[test] + fn request_size_checks_multiplication_and_binary_header() { + assert_eq!(result_len(b"123456789".len(), SMALL_INT_MAX as usize), None); + assert_eq!( + result_len(b"0123456789abcdef0".len(), SMALL_INT_MAX as usize), + None + ); + let max_bytes = ((u64::MAX >> BOXED_TAG_BITS) / 8) as usize; + assert_eq!(result_len(b"x".len(), max_bytes), Some(max_bytes)); + assert_eq!(result_len(b"x".len(), max_bytes + 1), None); + } + + // Intent: every copy, including its final/empty slice, pays for output work. + #[test] + fn charges_output_and_obeys_remaining_reductions() { + let mut process = Process::new(ProcessId::new(1)); + process.context.heap_ptr = &mut process.heap; + let input = process.heap.alloc_binary(b"x"); + for len in [ + 0, + 1, + BYTES_PER_REDUCTION, + BYTES_PER_REDUCTION + 1, + CHUNK_BYTES, + ] { + process.context.reductions = 4000; + unsafe { + nera_rt_binary_copy_2( + &mut process.context, + input.0, + ErlTerm::small_int(len as i64).0, + ); + } + assert_eq!(process.context.status, STATUS_OK); + let cost = len.div_ceil(BYTES_PER_REDUCTION).max(1) as i64; + assert_eq!(process.context.reductions, 4000 - cost); + } + for remaining in [0, 1] { + let mut work = Copy::new(1, CHUNK_BYTES + 1).unwrap(); + let limit = budget(&work, remaining); + assert_eq!(limit, remaining as usize * BYTES_PER_REDUCTION); + work.step(b"x", limit).unwrap(); + assert_eq!(work.offset(), limit); + } + } + + // Intent: suspended copying preserves input roots, partial repetition offsets + // and caller state through full GC, then adopts (rather than recopies) output. + #[test] + fn suspended_copy_survives_gc_and_adopts_output() { + let mut process = Process::new(ProcessId::new(2)); + let bytes: Vec = (0..=255).cycle().take(CHUNK_BYTES + 19).collect(); + let data = Arc::new(BinaryData::new( + [b"prefix".as_slice(), &bytes, b"suffix"].concat(), + )); + let owner = Arc::downgrade(&data); + let input = process + .heap + .alloc_refc_binary_from_arc(data, b"prefix".len(), bytes.len()); + process.context.heap_ptr = &mut process.heap; + process.context.reductions = 1; + unsafe { + nera_rt_binary_copy_2(&mut process.context, input.0, ErlTerm::small_int(3).0); + } + assert_eq!(process.context.status, STATUS_YIELD); + let work = process.context.native_work.as_mut().unwrap(); + assert_eq!(work.offset(), BYTES_PER_REDUCTION); + let output_ptr = work.output().as_ptr(); + process.resume_continuations = super::super::super::receive::take_pending_resumes(); + let mut slices = 0; + while !process.resume_continuations.is_empty() { + unsafe { + process.collect_garbage_full(); + process.reductions = 4000; + process.run_slice(); + } + assert!(owner.upgrade().is_some()); + slices += 1; + assert!(slices <= (bytes.len() * 3).div_ceil(CHUNK_BYTES)); + } + assert_eq!(process.context.status, STATUS_OK); + assert!(process.context.native_work.is_none()); + let (ptr, len) = unsafe { binary_data(ErlTerm(process.context.result)).unwrap() }; + assert_eq!(ptr, output_ptr); + assert_eq!( + unsafe { std::slice::from_raw_parts(ptr, len) }, + bytes.repeat(3) + ); + let last_bytes = (len - BYTES_PER_REDUCTION) % CHUNK_BYTES; + assert_eq!( + process.context.reductions, + 4000 - last_bytes.div_ceil(BYTES_PER_REDUCTION) as i64 + ); + drop(process); + assert!(owner.upgrade().is_none()); + } + + // Intent: a second process must run during a single large copy, even with + // only one worker. Observe actual copy progress, not a timing threshold. + #[test] + fn single_worker_observes_partial_copy() { + use crate::scheduler::Scheduler; + use std::sync::atomic::{AtomicUsize, Ordering}; + extern "C" fn entry(ctx: *mut ProcessContext, args: *const u64, _: u32) { + unsafe { + (*ctx).result = nera_rt_binary_copy_2(ctx, *args, *args.add(1)); + } + } + let scheduler = Scheduler::new(1); + let len = 4 * 1024 * 1024; + let mut producer = Process::with_entry(scheduler.next_pid(), entry, Vec::new()); + let input = producer.heap.alloc_binary(b"x"); + producer.initial_args = vec![input, ErlTerm::small_int(len as i64)]; + let pid = scheduler.spawn_process(producer); + let entry = scheduler.get_process(pid).unwrap(); + let progress = Arc::new(AtomicUsize::new(0)); + let observed = Arc::clone(&progress); + let mut observer = Process::new(scheduler.next_pid()); + observer.rust_fn = Some(Box::new(move |inner| { + let producer = entry.lock_for_runtime(); + if let Some(work) = &producer.context.native_work { + observed.store(work.offset(), Ordering::SeqCst); + } + inner.finish(); + })); + scheduler.spawn_process(observer); + assert!(unsafe { scheduler.run() }); + let copied = progress.load(Ordering::SeqCst); + eprintln!("observer ran after {copied}/{len} output bytes"); + assert!(copied > 0 && copied < len); + } + + // Intent: cancellation must release native work and its only input root. + #[test] + fn dropping_suspended_copy_releases_input() { + let mut process = Process::new(ProcessId::new(3)); + let data = Arc::new(BinaryData::new(b"repeated source".repeat(100))); + let owner = Arc::downgrade(&data); + let input = + process + .heap + .alloc_refc_binary_from_arc(data, 0, b"repeated source".len() * 100); + process.context.heap_ptr = &mut process.heap; + process.context.reductions = 0; + unsafe { + nera_rt_binary_copy_2(&mut process.context, input.0, ErlTerm::small_int(100).0); + } + assert_eq!(process.context.status, STATUS_YIELD); + assert_eq!(process.context.native_work.as_ref().unwrap().offset(), 0); + process.resume_continuations = super::super::super::receive::take_pending_resumes(); + drop(process); + assert!(owner.upgrade().is_none()); + } +} diff --git a/crates/nera-runtime/src/native_work.rs b/crates/nera-runtime/src/native_work.rs index 81da191..49059de 100644 --- a/crates/nera-runtime/src/native_work.rs +++ b/crates/nera-runtime/src/native_work.rs @@ -16,16 +16,17 @@ pub(crate) enum WorkError { } pub(crate) trait BinaryWork: Send { - /// Logical input bytes per reduction, calibrated per operation. + /// Logical work bytes per reduction, calibrated per operation (input bytes + /// for codecs, output bytes for repetition). fn bytes_per_reduction(&self) -> usize; - /// Non-final steps must end at this input boundary. + /// Non-final steps must end at this work boundary. fn quantum(&self) -> usize { 1 } fn max_bytes(&self) -> usize; fn offset(&self) -> usize; fn output(&mut self) -> &mut Vec; - /// Process at most `limit` input bytes. Return Ok(true) when complete, or a terminal error. + /// Process at most `limit` work bytes. Return Ok(true) when complete, or a terminal error. fn step(&mut self, input: &[u8], limit: usize) -> Result; } diff --git a/tests/README.md b/tests/README.md index 976fbaf..844a347 100644 --- a/tests/README.md +++ b/tests/README.md @@ -199,6 +199,8 @@ These are automatically skipped by the runner and show up as SKIP in output. | `rand_regression.erl` | Seeded rand against OTP 29 for every algorithm and seed form: uniform, uniform_real, bytes, normal, shuffle and jump draws, exported states, the process dictionary forms, mwc59 and splitmix64 utilities and error reasons; expectations generated by `scripts/generate-rand-regression.escript`; also run on BEAM | Table-driven | | `math_exp_log_pow.erl` | math:exp/1, log/1 and pow/2: values, overflow and domain errors, badarg for nonnumbers, direct, remote fun and variable module calls; also run on BEAM | Table-driven | | `float_integer_large.erl` | trunc/1, round/1, ceil/1 and floor/1 on floats beyond 2^63 return the exact integer, through direct, apply and remote fun calls; also run on BEAM | Table-driven | +| `binary_copy.erl` | binary:copy/2: size overflow, unsigned-word count validation, monitored errors, chunked/sub-binary results and dynamic calls; also run on BEAM | Table-driven | +| `binary_copy_scheduler.erl` | binary:copy/2: large and repeated near-chunk copies share a scheduler; strict ordering through single_scheduler/0 with NERA_SCHEDULERS=1 | 4 workloads | | `binary_split.erl` | binary:split/2,3 with [global] | ~8 | | `binary_split_patterns.erl` | binary:split/2,3 with a list of patterns: one-element lists, Cowlib's server-sent events line and field splits, longest match at the same offset, non-overlapping global splits, trim_all, a 1,000-line stream and a view into it, apply and fun calls, badarg patterns; also run on BEAM | Table-driven | | `binary_longest_common_prefix.erl` | Complete proper-list/binary validation, shared/shifted prefixes and dynamic calls | 17 | diff --git a/tests/binary_copy.erl b/tests/binary_copy.erl new file mode 100644 index 0000000..0f1d224 --- /dev/null +++ b/tests/binary_copy.erl @@ -0,0 +1,64 @@ +%% EXPECT: binary_copy_ok +%% Intent: binary:copy/2 must raise OTP-compatible errors instead of aborting, +%% and preserve ordinary results and caller state when copying suspends. +-module(binary_copy). +-export([test/0]). + +test() -> + lists:foreach(fun({Bin, Count, Expected}) -> + Expected = binary:copy(id(Bin), id(Count)), + Expected = apply(binary, copy, [Bin, Count]), + Copy = fun binary:copy/2, + Expected = Copy(Bin, Count) + end, [{<<"abc">>, 3, <<"abcabcabc">>}, + {<<>>, (1 bsl 64)-1, <<>>}, + {<<"source">>, 0, <<>>}]), + lists:foreach(fun({Bin, Count, Reason}) -> + expect_error(fun() -> binary:copy(Bin, Count) end, Reason), + expect_error(fun() -> apply(binary, copy, [Bin, Count]) end, Reason), + Copy = fun binary:copy/2, + expect_error(fun() -> Copy(Bin, Count) end, Reason) + end, [{<<"123456789">>, (1 bsl 60)-1, system_limit}, + {<<"0123456789abcdef0">>, (1 bsl 60)-1, system_limit}, + {<<"x">>, 1 bsl 61, system_limit}, + {<<>>, 1 bsl 64, badarg}, + {<<"x">>, 1 bsl 64, badarg}, + {<<>>, -1, badarg}, {<<"x">>, -(1 bsl 80), badarg}, + {<<>>, not_integer, badarg}, {<<"x">>, 1.0, badarg}, + {not_binary, 0, badarg}, {<<1:1>>, 0, badarg}]), + %% The original crash must become an ordinary monitored child error, with + %% the parent reaching the completion marker rather than aborting or hanging. + {Pid, Ref} = spawn_monitor(fun() -> + binary:copy(id(<<"123456789">>), id((1 bsl 60)-1)) + end), + receive + {'DOWN', Ref, process, Pid, {system_limit, Stack}} when is_list(Stack) -> ok + after 5000 -> error(missing_system_limit_down) + end, + %% Real outputs straddle native chunks and repetition boundaries. The live + %% parent binary and try handler must survive all suspensions. + lists:foreach(fun(Count) -> + Expected = list_to_binary(lists:duplicate(Count, "abc")), + Wrapped = <<42,Expected/binary,43>>, + Subject = binary:part(Wrapped, 1, byte_size(Expected)), + Triple = <>, + Triple = try binary:copy(Subject, 3) + catch C:R -> error({unexpected_copy_error,C,R}) end, + Triple = apply(binary, copy, [Subject, 3]), + 42 = binary:at(Wrapped, 0), + 43 = binary:at(Wrapped, byte_size(Wrapped)-1) + end, [21845,21846,43691]), + io:format("binary_copy_ok~n"), + ok. + +expect_error(Fun, Expected) -> + Parent = self(), + Expected = try + Fun(), + Parent ! copy_returned_after_error, + no_error + catch error:Reason -> Reason end, + receive copy_returned_after_error -> error(copy_did_not_raise) + after 0 -> ok end. + +id(Value) -> Value. diff --git a/tests/binary_copy_scheduler.erl b/tests/binary_copy_scheduler.erl new file mode 100644 index 0000000..0b89b87 --- /dev/null +++ b/tests/binary_copy_scheduler.erl @@ -0,0 +1,45 @@ +%% EXPECT: binary_copy_scheduler_ok +%% Intent: binary:copy/2 must account for output work and yield while copying. +%% The dedicated single-scheduler entry asserts ordering, not elapsed time. +-module(binary_copy_scheduler). +-export([test/0, single_scheduler/0]). + +test() -> run(false). +single_scheduler() -> run(true). + +run(Strict) -> + %% One large call and repeated near-chunk calls must both share a worker. + lists:foreach(fun({Size, Count}) -> compete(Size, Count, Strict) end, + [{65535,128}, {65536,128}, {65537,128}, {4*1024*1024,1}]), + io:format("binary_copy_scheduler_ok~n"), + ok. + +compete(Size, Count, Strict) -> + Parent = self(), + Observer = spawn(fun() -> + Parent ! observer_ready, + receive go -> Parent ! observed end + end), + receive observer_ready -> ok after 5000 -> error(observer_not_ready) end, + Worker = spawn(fun() -> + receive go -> + Observer ! go, + copy_loop(Size, Count), + Parent ! copied + end + end), + Worker ! go, + receive + observed -> receive copied -> ok after 5000 -> error(copy_timeout) end; + copied when Strict =:= false -> + receive observed -> ok after 5000 -> error(observer_timeout) end; + copied -> error(observer_starved) + after 5000 -> error(no_copy_progress) + end. + +copy_loop(_, 0) -> ok; +copy_loop(Size, Count) -> + Result = binary:copy(<<42>>, Size), + Size = byte_size(Result), + 42 = binary:at(Result, Size-1), + copy_loop(Size, Count-1).