Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
2 changes: 1 addition & 1 deletion packages/process-daemon/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ shell-words = "1.1"

[target.'cfg(unix)'.dependencies]
libc = "0.2"
nix = { version = "0.28", features = ["process", "term", "user"] }
nix = { version = "0.28", features = ["poll", "process", "term", "user"] }
sha2 = "0.10"

[target.'cfg(windows)'.dependencies]
Expand Down
4 changes: 4 additions & 0 deletions packages/process-daemon/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,10 @@ daemon; runtime-directory/registration wiring is platform-specific. The Unix
backend preserves half-close when completing subscriptions so queued final
frames are not discarded on macOS. Shutdown cancellation is a separate operation
that wakes partial requests and blocked subscription writes before joining.
On Unix the daemon loop sleeps in `poll(2)` on the listener and a private wake
socket pair until a connection arrives, a stop or ownership change wakes it, an
ownership deadline is due, or a two-second safety net elapses. Windows still
polls every 10 ms.

The Windows named-pipe backend keeps protocol 7 and the registration fields
`instance_id`, `pid`, `protocol`, `socket`, and `token`. Treat `socket` as an opaque
Expand Down
103 changes: 94 additions & 9 deletions packages/process-daemon/src/daemon/control.rs
Original file line number Diff line number Diff line change
@@ -1,26 +1,48 @@
#[cfg(all(test, unix))]
use std::sync::atomic::AtomicU64;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Mutex, MutexGuard, PoisonError};
use std::time::Instant;

use anyhow::{Result, anyhow};
use anyhow::{Result, anyhow, bail};

use crate::ownership::Ownership;
use crate::ownership::{Ownership, Snapshot};
use crate::transport::Wake;

/// Daemon stop flag and ownership state, shared by the main loop and handlers.
///
/// Every write to either goes through a method here; the fields stay private to
/// this module so no caller can change them behind the main loop's back. Both
/// writes happen under the ownership lock.
/// this module so no caller can change them behind the main loop's back. Each
/// write happens under the ownership lock and notifies `wake` before releasing
/// it, which is what makes the missed-wake check in `tick` exact.
pub(super) struct Control {
stop: AtomicBool,
ownership: Mutex<Ownership>,
wake: Wake,
#[cfg(all(test, unix))]
pub stats: Stats,
}

#[cfg(all(test, unix))]
#[derive(Default)]
pub(super) struct Stats {
pub iterations: AtomicU64,
pub missed_wakes: AtomicU64,
}

pub(super) struct Tick {
pub stopped: bool,
pub deadline: Option<Instant>,
}

impl Control {
pub fn new(ownership: Ownership) -> Self {
pub fn new(wake: Wake, ownership: Ownership) -> Self {
Self {
stop: AtomicBool::new(false),
ownership: Mutex::new(ownership),
wake,
#[cfg(all(test, unix))]
stats: Stats::default(),
}
}

Expand All @@ -32,31 +54,94 @@ impl Control {
.lock()
.unwrap_or_else(PoisonError::into_inner);
self.stop.store(true, Ordering::Release);
self.wake.notify();
}

pub fn stopping(&self) -> bool {
self.stop.load(Ordering::Acquire)
}

/// Notifies even when `f` changes nothing, so every mutation wakes the loop.
/// `f` must not call back into `Control`; the lock is not reentrant.
pub fn update_ownership<T>(&self, f: impl FnOnce(&mut Ownership) -> T) -> Result<T> {
let mut ownership = self.lock()?;
Ok(f(&mut ownership))
let value = f(&mut ownership);
self.wake.notify();
Ok(value)
}

pub fn is_owner(&self, generation: u64) -> Result<bool> {
Ok(self.lock()?.is_owner(generation))
}

/// Main loop only. Returns whether the daemon should stop.
pub fn tick(&self, now: Instant) -> Result<bool> {
/// Wakes the loop without changing state, e.g. when a handler finishes.
pub fn notify(&self) {
self.wake.notify();
}

/// Main loop only.
pub fn wake(&self) -> &Wake {
&self.wake
}

/// Main loop only. Drains wakes, detects a missed wake, and advances ownership in one
/// critical section. Must never notify, or the loop would wake itself every pass.
///
/// `last` is the snapshot left by the previous tick; `timed_out` is whether the wait
/// before this tick slept for its whole timeout.
pub fn tick(&self, now: Instant, last: &mut Snapshot, timed_out: bool) -> Result<Tick> {
let mut ownership = self.lock()?;
Ok(self.stop.load(Ordering::Acquire) || ownership.tick(now))
let woken = self.wake.drain();
let stopping = self.stop.load(Ordering::Acquire);
let observed = ownership.snapshot();
// Every writer notifies before releasing this lock, so a change since `last` with
// nothing drained means some writer skipped its wake.
let missed = timed_out && !woken && (stopping || observed != *last);
if missed {
#[cfg(all(test, unix))]
self.stats.missed_wakes.fetch_add(1, Ordering::Relaxed);
eprintln!(
"opencode-pty missed wake (stopping={stopping}, {last:?} -> {observed:?}); acted on via safety net"
);
}
let stopped = stopping || ownership.tick(now);
*last = ownership.snapshot();
let deadline = ownership.deadline();
drop(ownership);
// An error rather than a panic, so cleanup still runs.
if missed && cfg!(debug_assertions) {
bail!("missed wake detected");
}
Ok(Tick { stopped, deadline })
}

fn lock(&self) -> Result<MutexGuard<'_, Ownership>> {
self.ownership
.lock()
.map_err(|_| anyhow!("ownership lock poisoned"))
}

/// Detector tests only: sets the stop flag without a wake. Acts only when no wake is
/// pending, which proves the loop has ticked past every earlier change. Returns whether
/// it acted.
#[cfg(all(test, unix))]
pub fn set_stop_without_wake(&self) -> bool {
let _ownership = self.ownership.lock().unwrap();
if self.wake.pending() {
return false;
}
self.stop.store(true, Ordering::Release);
true
}

/// Detector tests only: like `set_stop_without_wake`, for an ownership change.
#[cfg(all(test, unix))]
pub fn update_ownership_without_wake(&self, f: impl FnOnce(&mut Ownership)) -> bool {
let mut ownership = self.ownership.lock().unwrap();
if self.wake.pending() {
return false;
}
f(&mut ownership);
true
}
}
Loading
Loading