RFC 016 Chunk 2a: per-actor timeslice overrun counter
Tally overruns at the slice-expiry site in preempt.rs (the RFC 006 signal), surfaced as ActorInfo.overruns. - Counter is a hot-region AtomicU64 on Slot, single-writer Relaxed load+store (no atomic RMW), read Relaxed by the snapshot (D5). - Reached from the rare expiry branch via a stashed *const Slot in a preempt thread-local, set/cleared on the same resume/return boundary as CURRENT_STOP — one TLS load, no runtime lookup. Shared infra for the messages-received counter next. - Reset in all three slot-lifecycle sites: Slot::vacant, reclaim_slot, install_actor (standing invariant, D7) — per-incarnation counts.
This commit is contained in:
@@ -83,6 +83,10 @@ pub struct ActorInfo {
|
|||||||
/// install / spawn_addr / gen_server). 0 for an actor that holds only a
|
/// install / spawn_addr / gen_server). 0 for an actor that holds only a
|
||||||
/// private `channel()` receiver — those are invisible to the registry.
|
/// private `channel()` receiver — those are invisible to the registry.
|
||||||
pub mailbox_depth: u32,
|
pub mailbox_depth: u32,
|
||||||
|
/// Timeslice overruns tallied for this incarnation (RFC 016 Chunk 2): how
|
||||||
|
/// many times the actor was preempted for exceeding its slice. Resets on
|
||||||
|
/// restart (per-incarnation, D7).
|
||||||
|
pub overruns: u64,
|
||||||
}
|
}
|
||||||
|
|
||||||
/// A whole-runtime snapshot. See the module docs for the D2 tearing model.
|
/// A whole-runtime snapshot. See the module docs for the D2 tearing model.
|
||||||
@@ -159,6 +163,9 @@ fn read_slot(slot: &Slot, idx: u32, mail: Option<&MailboxInfo>) -> Option<ActorI
|
|||||||
let joiners = cold.waiters.len() as u32;
|
let joiners = cold.waiters.len() as u32;
|
||||||
drop(cold);
|
drop(cold);
|
||||||
|
|
||||||
|
// Counters are hot-region atomics, read lock-free (RFC 016 Chunk 2).
|
||||||
|
let overruns = slot.overruns();
|
||||||
|
|
||||||
// Names + depth belong to this incarnation only if the registry mailbox's
|
// Names + depth belong to this incarnation only if the registry mailbox's
|
||||||
// pid matches the slab generation; a stale registry entry (dead prior
|
// pid matches the slab generation; a stale registry entry (dead prior
|
||||||
// occupant, not yet pruned) contributes nothing.
|
// occupant, not yet pruned) contributes nothing.
|
||||||
@@ -177,6 +184,7 @@ fn read_slot(slot: &Slot, idx: u32, mail: Option<&MailboxInfo>) -> Option<ActorI
|
|||||||
links,
|
links,
|
||||||
joiners,
|
joiners,
|
||||||
mailbox_depth,
|
mailbox_depth,
|
||||||
|
overruns,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -58,6 +58,16 @@ thread_local! {
|
|||||||
/// resume path free of atomic ref-count traffic; see `check_cancelled` for
|
/// resume path free of atomic ref-count traffic; see `check_cancelled` for
|
||||||
/// the safety argument.
|
/// the safety argument.
|
||||||
static CURRENT_STOP: Cell<*const AtomicBool> = const { Cell::new(std::ptr::null()) };
|
static CURRENT_STOP: Cell<*const AtomicBool> = const { Cell::new(std::ptr::null()) };
|
||||||
|
|
||||||
|
/// Raw pointer to the on-CPU actor's slot, set/cleared by the scheduler on
|
||||||
|
/// the same resume/return boundary as `CURRENT_STOP` (RFC 016 Chunk 2).
|
||||||
|
/// Lets the rare slice-expiry site bump that actor's overrun counter with
|
||||||
|
/// one TLS load and no runtime lookup. Null while no actor is on-CPU. The
|
||||||
|
/// slot lives in the fixed slab and is never reclaimed while the actor is
|
||||||
|
/// running, so the pointer is valid for the whole resume (same lifetime
|
||||||
|
/// argument as `CURRENT_STOP`).
|
||||||
|
static CURRENT_SLOT: Cell<*const crate::runtime::Slot> =
|
||||||
|
const { Cell::new(std::ptr::null()) };
|
||||||
}
|
}
|
||||||
|
|
||||||
// ---------------------------------------------------------------------------
|
// ---------------------------------------------------------------------------
|
||||||
@@ -77,6 +87,33 @@ pub(crate) fn clear_current_stop() {
|
|||||||
CURRENT_STOP.with(|c| c.set(std::ptr::null()));
|
CURRENT_STOP.with(|c| c.set(std::ptr::null()));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Bind the on-CPU actor's slot. Called by the scheduler immediately before
|
||||||
|
/// `switch_to_actor`, beside `set_current_stop`.
|
||||||
|
pub(crate) fn set_current_slot(slot: *const crate::runtime::Slot) {
|
||||||
|
CURRENT_SLOT.with(|c| c.set(slot));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Unbind the slot pointer on the return path, beside `clear_current_stop`.
|
||||||
|
pub(crate) fn clear_current_slot() {
|
||||||
|
CURRENT_SLOT.with(|c| c.set(std::ptr::null()));
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Tally a timeslice overrun against the on-CPU actor (RFC 016 Chunk 2). A
|
||||||
|
/// no-op if no actor is bound (the scheduler's own stack). Reached only from
|
||||||
|
/// the slice-expiry branch, which is already the yield path, so its cost is
|
||||||
|
/// irrelevant.
|
||||||
|
#[inline]
|
||||||
|
fn note_overrun() {
|
||||||
|
let p = CURRENT_SLOT.with(|c| c.get());
|
||||||
|
// SAFETY: `p` is null (no actor on-CPU) or a pointer to the on-CPU actor's
|
||||||
|
// slot in the fixed slab. The slot is not reclaimed while the actor runs
|
||||||
|
// (finalize/reclaim happen only after it yields back), so the deref is
|
||||||
|
// valid for the whole resume — the same argument as `check_cancelled`.
|
||||||
|
if !p.is_null() {
|
||||||
|
unsafe { (*p).record_overrun() };
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/// Observation point for cooperative cancellation. If the on-CPU actor has
|
/// Observation point for cooperative cancellation. If the on-CPU actor has
|
||||||
/// been flagged for stop, raise the sentinel panic so the trampoline's
|
/// been flagged for stop, raise the sentinel panic so the trampoline's
|
||||||
/// `catch_unwind` tears the stack down (running Drop) and reports
|
/// `catch_unwind` tears the stack down (running Drop) and reports
|
||||||
@@ -189,6 +226,10 @@ pub fn maybe_preempt() {
|
|||||||
check_cancelled();
|
check_cancelled();
|
||||||
let start = TIMESLICE_START.with(|s| s.get());
|
let start = TIMESLICE_START.with(|s| s.get());
|
||||||
if rdtsc().saturating_sub(start) > CONFIGURED_TIMESLICE_CYCLES.with(|t| t.get()) {
|
if rdtsc().saturating_sub(start) > CONFIGURED_TIMESLICE_CYCLES.with(|t| t.get()) {
|
||||||
|
// Tally the overrun (RFC 016 Chunk 2) before handing back —
|
||||||
|
// this is the slice-expiry site RFC 006 wanted, and it's
|
||||||
|
// already the yield path, so the counter is near-free.
|
||||||
|
note_overrun();
|
||||||
// SAFETY: reachable only inside an actor (the scheduler
|
// SAFETY: reachable only inside an actor (the scheduler
|
||||||
// sets PREEMPTION_ENABLED on resume and clears it on
|
// sets PREEMPTION_ENABLED on resume and clears it on
|
||||||
// return). The scheduler stack is therefore valid.
|
// return). The scheduler stack is therefore valid.
|
||||||
|
|||||||
@@ -426,6 +426,14 @@ pub(crate) struct Slot {
|
|||||||
/// First-resume closure, double-boxed so it fits an `AtomicPtr`
|
/// First-resume closure, double-boxed so it fits an `AtomicPtr`
|
||||||
/// (`Box<Closure>` is a thin pointer). Swap-to-take; null when absent.
|
/// (`Box<Closure>` is a thin pointer). Swap-to-take; null when absent.
|
||||||
closure: AtomicPtr<Closure>,
|
closure: AtomicPtr<Closure>,
|
||||||
|
/// RFC 016 Chunk 2 — per-actor timeslice overrun tally. Single-writer: only
|
||||||
|
/// the on-CPU actor's scheduler thread increments it (at the slice-expiry
|
||||||
|
/// site in `preempt.rs`, reached via the stashed slot pointer), so the
|
||||||
|
/// writes are plain Relaxed load+store with no atomic-RMW traffic; the
|
||||||
|
/// snapshot reads it Relaxed from any thread. Lives in the hot region rather
|
||||||
|
/// than `SlotCold` so the increment needs no lock; reset across reuse like
|
||||||
|
/// every other slot field (`vacant` / `reclaim_slot` / `install_actor`).
|
||||||
|
overruns: AtomicU64,
|
||||||
/// Cold lifecycle data. See [`SlotCold`].
|
/// Cold lifecycle data. See [`SlotCold`].
|
||||||
pub(crate) cold: RawMutex<SlotCold>,
|
pub(crate) cold: RawMutex<SlotCold>,
|
||||||
}
|
}
|
||||||
@@ -437,6 +445,7 @@ impl Slot {
|
|||||||
sp: AtomicUsize::new(0),
|
sp: AtomicUsize::new(0),
|
||||||
stop_ptr: AtomicPtr::new(std::ptr::null_mut()),
|
stop_ptr: AtomicPtr::new(std::ptr::null_mut()),
|
||||||
closure: AtomicPtr::new(std::ptr::null_mut()),
|
closure: AtomicPtr::new(std::ptr::null_mut()),
|
||||||
|
overruns: AtomicU64::new(0),
|
||||||
cold: RawMutex::new(SlotCold {
|
cold: RawMutex::new(SlotCold {
|
||||||
actor: None,
|
actor: None,
|
||||||
waiters: Vec::new(),
|
waiters: Vec::new(),
|
||||||
@@ -465,6 +474,31 @@ impl Slot {
|
|||||||
self.word.load()
|
self.word.load()
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/// Tally one timeslice overrun (RFC 016 Chunk 2). Single-writer: only the
|
||||||
|
/// on-CPU actor's own thread calls this, at the slice-expiry site, so a
|
||||||
|
/// Relaxed load+store is sufficient and avoids the cache-line lock of an
|
||||||
|
/// atomic RMW.
|
||||||
|
#[inline]
|
||||||
|
pub(crate) fn record_overrun(&self) {
|
||||||
|
let v = self.overruns.load(Ordering::Relaxed);
|
||||||
|
self.overruns.store(v.wrapping_add(1), Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Read the overrun tally (Relaxed; the snapshot reads cross-thread).
|
||||||
|
#[inline]
|
||||||
|
pub(crate) fn overruns(&self) -> u64 {
|
||||||
|
self.overruns.load(Ordering::Relaxed)
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Zero the per-actor introspection counters. Called at every point a slot
|
||||||
|
/// is recycled or freshly occupied (`reclaim_slot`, `install_actor`) so a
|
||||||
|
/// reused slot never carries a previous incarnation's counts — the standing
|
||||||
|
/// slot-lifecycle reset invariant (RFC 016 D7).
|
||||||
|
#[inline]
|
||||||
|
pub(crate) fn reset_counters(&self) {
|
||||||
|
self.overruns.store(0, Ordering::Relaxed);
|
||||||
|
}
|
||||||
|
|
||||||
/// A pid's-eye snapshot of the slot. Cold paths re-read this under the
|
/// A pid's-eye snapshot of the slot. Cold paths re-read this under the
|
||||||
/// cold lock (generation can't change while it is held).
|
/// cold lock (generation can't change while it is held).
|
||||||
#[inline]
|
#[inline]
|
||||||
@@ -1003,6 +1037,7 @@ pub(crate) fn install_actor(
|
|||||||
}
|
}
|
||||||
slot.sp.store(sp, Ordering::Relaxed);
|
slot.sp.store(sp, Ordering::Relaxed);
|
||||||
slot.store_closure(closure);
|
slot.store_closure(closure);
|
||||||
|
slot.reset_counters();
|
||||||
inner.live_actors.fetch_add(1, Ordering::Relaxed);
|
inner.live_actors.fetch_add(1, Ordering::Relaxed);
|
||||||
|
|
||||||
// Publish: only now can pops, unparks, or stops find the actor. The
|
// Publish: only now can pops, unparks, or stops find the actor. The
|
||||||
@@ -1043,6 +1078,7 @@ pub(crate) fn reclaim_slot(inner: &RuntimeInner, pid: Pid) {
|
|||||||
cold.waiters.clear();
|
cold.waiters.clear();
|
||||||
cold.monitors.clear();
|
cold.monitors.clear();
|
||||||
cold.links.clear();
|
cold.links.clear();
|
||||||
|
slot.reset_counters();
|
||||||
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
slot.stop_ptr.store(std::ptr::null_mut(), Ordering::Release);
|
||||||
// The generation bump IS the reclaim: every stale pid is dead from
|
// The generation bump IS the reclaim: every stale pid is dead from
|
||||||
// this store onwards (unpark protocol, pops, cold-path re-verifies).
|
// this store onwards (unpark protocol, pops, cold-path re-verifies).
|
||||||
@@ -1451,6 +1487,7 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
|||||||
set_actor_sp(sp);
|
set_actor_sp(sp);
|
||||||
set_current_pid(pid);
|
set_current_pid(pid);
|
||||||
crate::preempt::set_current_stop(stop_flag);
|
crate::preempt::set_current_stop(stop_flag);
|
||||||
|
crate::preempt::set_current_slot(slot as *const Slot);
|
||||||
reset_actor_done();
|
reset_actor_done();
|
||||||
YIELD_INTENT.with(|c| c.set(YieldIntent::Yield));
|
YIELD_INTENT.with(|c| c.set(YieldIntent::Yield));
|
||||||
// RFC 005 timeslice inheritance: a slot-popped actor does NOT get a
|
// RFC 005 timeslice inheritance: a slot-popped actor does NOT get a
|
||||||
@@ -1474,6 +1511,7 @@ fn schedule_loop(inner: &Arc<RuntimeInner>, slot_idx: usize) {
|
|||||||
stats.current_pid_index.store(u32::MAX, Ordering::Relaxed);
|
stats.current_pid_index.store(u32::MAX, Ordering::Relaxed);
|
||||||
clear_current_pid();
|
clear_current_pid();
|
||||||
crate::preempt::clear_current_stop();
|
crate::preempt::clear_current_stop();
|
||||||
|
crate::preempt::clear_current_slot();
|
||||||
|
|
||||||
let intent = YIELD_INTENT.with(|c| c.get());
|
let intent = YIELD_INTENT.with(|c| c.get());
|
||||||
slot.sp.store(get_actor_sp(), Ordering::Relaxed);
|
slot.sp.store(get_actor_sp(), Ordering::Relaxed);
|
||||||
|
|||||||
@@ -234,6 +234,7 @@ fn tree_from_nests_children_and_reroots_orphans() {
|
|||||||
links: 0,
|
links: 0,
|
||||||
joiners: 0,
|
joiners: 0,
|
||||||
mailbox_depth: 0,
|
mailbox_depth: 0,
|
||||||
|
overruns: 0,
|
||||||
};
|
};
|
||||||
|
|
||||||
let snap = RuntimeSnapshot {
|
let snap = RuntimeSnapshot {
|
||||||
@@ -258,3 +259,30 @@ fn tree_from_nests_children_and_reroots_orphans() {
|
|||||||
assert!(o.orphaned, "an actor whose parent is absent must be flagged orphaned");
|
assert!(o.orphaned, "an actor whose parent is absent must be flagged orphaned");
|
||||||
assert!(o.children.is_empty());
|
assert!(o.children.is_empty());
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn overrun_count_increments_on_forced_preemption() {
|
||||||
|
run(|| {
|
||||||
|
// A worker that forces its slice to expire, then hits an observation
|
||||||
|
// point so the slice-expiry site fires and tallies one overrun.
|
||||||
|
let (ready_tx, ready_rx) = channel::<()>();
|
||||||
|
let (gate_tx, gate_rx) = channel::<()>();
|
||||||
|
let h = spawn(move || {
|
||||||
|
smarm::preempt::expire_timeslice_for_test();
|
||||||
|
smarm::check!(); // preempt-yield here → one overrun tallied
|
||||||
|
ready_tx.send(()).unwrap();
|
||||||
|
gate_rx.recv().unwrap();
|
||||||
|
});
|
||||||
|
ready_rx.recv().unwrap(); // worker is past the forced preemption
|
||||||
|
|
||||||
|
let info = actor_info(h.pid()).expect("worker present");
|
||||||
|
assert!(
|
||||||
|
info.overruns >= 1,
|
||||||
|
"forced timeslice expiry should tally at least one overrun, got {}",
|
||||||
|
info.overruns
|
||||||
|
);
|
||||||
|
|
||||||
|
gate_tx.send(()).unwrap();
|
||||||
|
h.join().unwrap();
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user