diff --git a/CHANGELOG.md b/CHANGELOG.md index 62a335b..8f4b278 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -13,6 +13,19 @@ All notable milestones for the nera project are documented in this file. covering message and timeout wakes, tail position, deeper nesting and successive inner receives. +## Initial process handles signals queued during its final slice + +- When the initial entry function returns with links still held, the scheduler + now checks its inbox after publishing the wait. A message or exit signal sent + while the process was still running no longer remains stranded and hangs + program termination. +- EXIT fragments carry their signal classification in the same inbox push. + A worker that wakes itself before the sender's wake attempt now handles a + fatal exit as an exit, rather than completing normally. +- Runtime regressions exercise both deliveries through the worker loop, + including a sender paused after its push, and both inbox drain paths. An + empty-inbox control retains the initial process's wait and links. + ## Oversized binary copies raise Erlang errors - `binary:copy/2` checks result-size multiplication and the binary header's diff --git a/crates/nera-runtime/src/ffi/process.rs b/crates/nera-runtime/src/ffi/process.rs index 1fb0839..f7ce5fc 100644 --- a/crates/nera-runtime/src/ffi/process.rs +++ b/crates/nera-runtime/src/ffi/process.rs @@ -1893,8 +1893,7 @@ pub unsafe extern "C" fn nera_rt_exit_2(ctx: *mut ProcessContext, pid_term: u64, heap.alloc_tuple(&[exit_atom, from_pid, ErlTerm(reason)]) }); let fragment = crate::message::MessageFragment::from_term(exit_msg); - entry.inbox.push(fragment); - entry.has_exit_signals.store(true, std::sync::atomic::Ordering::Release); + entry.inbox.push(crate::message::MessageFragment::ExitSignal(Box::new(fragment))); // Try waking the process. try_wake() handles receive-waiting // (ASTATUS_WAITING), try_wake_io() handles IO-waiting // (ASTATUS_IO_WAIT). Only one CAS can succeed. diff --git a/crates/nera-runtime/src/message.rs b/crates/nera-runtime/src/message.rs index d4ddebc..9ed87d7 100644 --- a/crates/nera-runtime/src/message.rs +++ b/crates/nera-runtime/src/message.rs @@ -161,6 +161,10 @@ pub struct HeapFragment { /// inbox (`SegQueue`). The receiving worker drains fragments into the process /// mailbox under its own Mutex. pub enum MessageFragment { + /// EXIT classification travels with the payload in the same queue push. + /// The worker marks the mailbox for signal handling when it drains this, + /// before a self-wake can treat the payload as ordinary mail. + ExitSignal(Box), /// Immediate value (atom, small int, pid, nil) — no heap needed. Immediate(ImmediateTerm), /// Heap-backed term with its own mini-heap (boxed to shrink the enum). @@ -309,6 +313,7 @@ impl MessageFragment { /// Consumes the fragment. pub fn copy_to_heap(self, target: &mut ProcessHeap) -> ErlTerm { match self { + MessageFragment::ExitSignal(signal) => signal.copy_to_heap(target), MessageFragment::Immediate(term) => term.get(), MessageFragment::RefcBinary(range) => { // Move the validated window and its Arc without another check or clone. diff --git a/crates/nera-runtime/src/process.rs b/crates/nera-runtime/src/process.rs index cc78527..53f06e7 100644 --- a/crates/nera-runtime/src/process.rs +++ b/crates/nera-runtime/src/process.rs @@ -320,9 +320,9 @@ pub struct ProcessEntry { /// Atomic mirror of `ProcessStatus` for lock-free wake detection. /// Senders CAS this from `WAITING → READY` to wake a process. atomic_status: AtomicU32, - /// Set when an EXIT signal is pushed to the inbox. The worker checks - /// this before scanning the mailbox for EXIT tuples, avoiding O(N) - /// scans on every wake when no signals are pending. + /// Set by the worker when it drains a classified EXIT fragment. The + /// classification is queued with the payload, never published separately + /// by the sender. Avoids O(N) mailbox scans when no signals are pending. pub has_exit_signals: AtomicBool, /// The actual process state, protected by a Mutex. proc: Mutex, @@ -580,8 +580,10 @@ fn drain_inbox_into(entry: &ProcessEntry, proc: &mut Process) { } } other => { + let is_exit = matches!(&other, MessageFragment::ExitSignal(_)); let msg = other.copy_to_heap(&mut proc.heap); proc.mailbox.send(msg); + if is_exit { entry.has_exit_signals.store(true, Ordering::Relaxed); } } } return; @@ -610,8 +612,10 @@ fn drain_inbox_into(entry: &ProcessEntry, proc: &mut Process) { } } other => { + let is_exit = matches!(&other, MessageFragment::ExitSignal(_)); let msg = other.copy_to_heap(&mut proc.heap); proc.mailbox.send(msg); + if is_exit { entry.has_exit_signals.store(true, Ordering::Relaxed); } } } } @@ -3593,6 +3597,39 @@ mod tests { // ASTATUS_IO_WAIT and try_wake / try_wake_io separation // --------------------------------------------------------------- + // EXIT publication must be indivisible even if its sender stops before + // waking the target. Exercise both the single and batched drain paths. + #[test] + fn inbox_drain_keeps_exit_classification_with_its_payload() { + use crate::message::{FragmentElem, ImmediateTerm}; + for ordinary_count in [0, 1, 2] { + let entry = ProcessEntry::new(Process::new(ProcessId::new(1))); + let ordinary: Vec<_> = (0..ordinary_count).map(ErlTerm::small_int).collect(); + for &msg in &ordinary { + entry.inbox.push(MessageFragment::Immediate(ImmediateTerm::new(msg))); + } + entry.inbox.push(MessageFragment::ExitSignal(Box::new( + MessageFragment::SimpleTuple(vec![ + FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom( + crate::atom_table::ATOM_EXIT_SIGNAL))), + FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::pid(ProcessId::new(2)))), + FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom( + crate::atom_table::ATOM_BADARG))), + ])))); + // No sender-side flag store or wake has occurred. + assert!(!entry.has_exit_signals.load(Ordering::Relaxed)); + let mut proc = entry.lock_for_runtime(); + proc.drain_inbox(); + assert!(entry.has_exit_signals.swap(false, Ordering::Acquire)); + // SAFETY: the guard exclusively owns every copied mailbox term; + // no collection or mutation intervenes before signal processing. + assert_eq!(unsafe { process_exit_signals(&mut proc) }, + Some(ErlTerm::atom(crate::atom_table::ATOM_BADARG))); + assert_eq!(proc.mailbox.messages().collect::>(), ordinary); + assert!(entry.inbox.is_empty()); + } + } + #[test] fn guard_drains_and_publishes_only_its_own_entry() { use crate::message::ImmediateTerm; diff --git a/crates/nera-runtime/src/scheduler.rs b/crates/nera-runtime/src/scheduler.rs index f99223d..74a612c 100644 --- a/crates/nera-runtime/src/scheduler.rs +++ b/crates/nera-runtime/src/scheduler.rs @@ -724,8 +724,7 @@ fn propagate_exit_signals( }; drop(proc); - entry.inbox.push(fragment); - entry.has_exit_signals.store(true, Ordering::Release); + entry.inbox.push(MessageFragment::ExitSignal(Box::new(fragment))); // Try waking the process. try_wake() handles receive-waiting, // try_wake_io() handles IO-waiting. Only one CAS can succeed. // The worker that runs the process cancels its receive timer; @@ -1594,12 +1593,15 @@ fn worker_loop( // from a linked process or when the runtime shuts down. let is_initial = proc.id == initial_pid.load(Ordering::Relaxed); if is_initial && !proc.otp.links.is_empty() { - proc.transition(ProcessState::Waiting(WaitReason::Receive { - timeout_deadline_ms: None, - })); - runnable_count.fetch_sub(1, Ordering::AcqRel); - drop(proc); - // Do not re-inject — the process sleeps until an EXIT + // A signal can arrive during the final Running slice + // and fail its wake. Publish the wait before checking + // the inbox, just as for an explicit receive. + if publish_wait(proc, &proc_entry, ProcessState::Waiting( + WaitReason::Receive { timeout_deadline_ms: None }), + &timer_wheel, &runnable_count) { + local.push(proc_entry); + } + // Otherwise the process sleeps until an EXIT // signal or message wakes it, at which point it will // exit normally (no more code to run). continue; @@ -2423,6 +2425,149 @@ mod tests { expired.len() } + #[derive(Clone, Copy, Debug)] + enum FinalSliceDelivery { + Message, + Exit, + ExitBeforeSenderWake, + Empty, + } + + // The initial entry returns while it still owns a link. Deliver in its + // actual Running slice, after the worker's inbox drain, so the sender's + // wake fails. The worker must publish its wait and recover that wake. + fn check_initial_return_with_links(delivery: FinalSliceDelivery) { + use crate::message::{FragmentElem, ImmediateTerm, MessageFragment}; + use std::cell::RefCell; + + thread_local! { + static FINAL_SLICE_SEND: RefCell>> = const { RefCell::new(None) }; + } + unsafe extern "C" fn returning_entry( + _ctx: *mut crate::process::ProcessContext, _args: *const u64, _nargs: u32, + ) { + FINAL_SLICE_SEND.with(|send| { + if let Some(send) = send.borrow_mut().take() { send(); } + }); + } + + let scheduler = Scheduler::new(1); + let initial_pid = scheduler.next_pid(); + let linked_pid = scheduler.next_pid(); + let mut initial = Process::with_entry(initial_pid, returning_entry, vec![]); + initial.add_link(linked_pid); + scheduler.spawn_process(initial); + let Steal::Success(initial) = scheduler.injector.steal() else { + panic!("initial process must be queued"); + }; + let mut linked = Process::new(linked_pid); + linked.add_link(initial_pid); + scheduler.spawn_process(linked); + let Steal::Success(linked) = scheduler.injector.steal() else { + panic!("linked process must be queued"); + }; + assert!(!publish_wait(linked.lock_for_runtime(), &linked, + ProcessState::Waiting(WaitReason::Receive { timeout_deadline_ms: None }), + &scheduler.timer_wheel, &scheduler.runnable_count)); + scheduler.injector.push(initial.clone()); + + let observed = Arc::new(AtomicU32::new(0)); + let observation = observed.clone(); + let target = initial.clone(); + FINAL_SLICE_SEND.with(|send| { + *send.borrow_mut() = Some(Box::new(move || { + observation.store(target.published_status(), Ordering::Release); + let fragment = match delivery { + FinalSliceDelivery::Message => Some(MessageFragment::Immediate( + ImmediateTerm::new(ErlTerm::atom(crate::atom_table::ATOM_OK)))), + FinalSliceDelivery::Exit | FinalSliceDelivery::ExitBeforeSenderWake => Some( + MessageFragment::ExitSignal(Box::new(MessageFragment::SimpleTuple(vec![ + FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom( + crate::atom_table::ATOM_EXIT_SIGNAL))), + FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::pid(linked_pid))), + FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom( + crate::atom_table::ATOM_BADARG))), + ])))), + FinalSliceDelivery::Empty => None, + }; + if let Some(fragment) = fragment { + target.inbox.push(fragment); + // Record an unexpected successful sender wake without + // panicking across the native entry boundary. + if !matches!(delivery, FinalSliceDelivery::ExitBeforeSenderWake) + && target.try_wake() { observation.store(u32::MAX, Ordering::Release); } + } + })); + }); + let worker = Worker::new_fifo(); + let stealers = Arc::new(vec![worker.stealer()]); + // Controlled parks bound the old hang without sleeps or retries. + let parks = std::cell::Cell::new(0); + let forced_stop = std::cell::Cell::new(false); + #[cfg(mio_supported)] + let (reactor, join) = crate::reactor::spawn_reactor(); + worker_loop(0, worker, scheduler.injector.clone(), stealers, + scheduler.process_table.clone(), scheduler.shutdown.clone(), + scheduler.runnable_count.clone(), scheduler.alive_count.clone(), + scheduler.any_crash.clone(), scheduler.initial_pid.clone(), 1, + scheduler.registered_names.clone(), scheduler.timer_wheel.clone(), + scheduler.io_pool.clone(), scheduler.worker_stats[0].clone(), + scheduler.worker_inboxes[0].clone(), Arc::new(vec![]), + |_| { + parks.set(parks.get() + 1); + if parks.get() > 16 { + forced_stop.set(true); + scheduler.shutdown.store(true, Ordering::SeqCst); + } + }, + #[cfg(mio_supported)] + reactor.clone()); + #[cfg(mio_supported)] + { reactor.shutdown(); join.join().unwrap(); } + FINAL_SLICE_SEND.with(|send| assert!(send.borrow().is_none())); + eprintln!("initial return: delivery={delivery:?} state_at_send={} inbox_empty={} \ + runnable={} forced_stop={}", observed.load(Ordering::Acquire), + initial.inbox.is_empty(), scheduler.runnable_count.load(Ordering::Acquire), + forced_stop.get()); + assert_eq!(observed.load(Ordering::Acquire), ASTATUS_RUNNING); + assert!(!forced_stop.get(), "the final-slice delivery must not strand the worker"); + assert!(initial.inbox.is_empty()); + assert_eq!(scheduler.runnable_count.load(Ordering::Acquire), 0); + assert!(scheduler.shutdown.load(Ordering::Acquire)); + assert_eq!(scheduler.any_crash.load(Ordering::Acquire), + matches!(delivery, FinalSliceDelivery::Exit | FinalSliceDelivery::ExitBeforeSenderWake)); + if matches!(delivery, FinalSliceDelivery::Empty) { + assert_eq!(initial.published_status(), ASTATUS_WAITING); + assert_eq!(initial.lock_for_runtime().otp.links.len(), 1); + } else { + assert_eq!(initial.published_status(), ASTATUS_EXITED); + assert!(scheduler.get_process(initial_pid).is_none()); + } + assert!(processes_are_blocked(&scheduler.process_table)); + } + + #[test] + fn initial_return_with_links_observes_final_slice_message() { + check_initial_return_with_links(FinalSliceDelivery::Message); + } + + #[test] + fn initial_return_with_links_observes_final_slice_exit() { + check_initial_return_with_links(FinalSliceDelivery::Exit); + } + + #[test] + fn initial_return_with_links_observes_exit_before_sender_wake() { + // Pause the sender after inbox publication, before its wake attempt. + // Classification must already be visible to the receiver's self-wake. + check_initial_return_with_links(FinalSliceDelivery::ExitBeforeSenderWake); + } + + #[test] + fn initial_return_with_links_without_delivery_stays_waiting() { + check_initial_return_with_links(FinalSliceDelivery::Empty); + } + #[test] fn stale_receive_timer_cannot_replace_a_later_one() { let scheduler = Scheduler::new(1); @@ -2580,14 +2725,13 @@ mod tests { use crate::message::{FragmentElem, ImmediateTerm, MessageFragment}; // exit/2 can wake the target before another process's // linked exit kills it; the queued kill is now stale. - linked.inbox.push(MessageFragment::SimpleTuple(vec![ + linked.inbox.push(MessageFragment::ExitSignal(Box::new(MessageFragment::SimpleTuple(vec![ FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom( crate::atom_table::ATOM_EXIT_SIGNAL))), FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::pid(dead_pid))), FragmentElem::Immediate(ImmediateTerm::new(ErlTerm::atom( crate::atom_table::ATOM_KILL))), - ])); - linked.has_exit_signals.store(true, Ordering::Release); + ])))); } assert!(if state == io { linked.try_wake_io() } else { linked.try_wake() }); scheduler.runnable_count.fetch_add(1, Ordering::AcqRel);