fusevm 0.14.14

//! Cooperative goroutine scheduler and channels.
//!
//! fusevm's dispatch loop runs one VM to completion; this module layers a
//! green-thread scheduler on top so a frontend (e.g. Go) can express
//! goroutines and channels. Each goroutine is its own [`VM`] instance sharing
//! the program `Chunk` and the frontend's thread-local heap; the scheduler
//! owns the channel table and a run queue.
//!
//! The mechanism reuses the same "op raises a request, halts, driver reads it
//! after `run()`" pattern as `Op::AwkSignal` — see [`crate::op::Op::Go`] and the
//! channel ops. A goroutine/channel op stashes a [`SchedReq`] in the VM and
//! halts; the scheduler reads it via [`VM::take_sched`], services it (delivering
//! any result by pushing onto the target VM's stack), and resumes that VM or
//! switches to another. Because the op has already advanced the VM's `ip`, a
//! resumed VM continues *past* the op — the scheduler never rewinds or
//! re-executes, so no result is lost.
//!
//! Channels are host-opaque `Value::Int(id)` handles into the scheduler's
//! channel table. Send/receive follow CSP semantics: a buffered channel holds up
//! to `cap` values; an unbuffered channel hands a value directly from a blocked
//! sender to a blocked receiver. When every goroutine is blocked, the scheduler
//! reports a deadlock (Go's "all goroutines are asleep").
//!
//! This is single-threaded and cooperative: goroutines yield only at channel
//! operations (and completion). That is a faithful model for channel-driven Go
//! programs — the common shape — without the data races an OS-thread model would
//! introduce over the frontend's thread-local heap.

use crate::{Frame, VMResult, Value, VM};
use std::collections::{HashSet, VecDeque};

/// A scheduling request a goroutine/channel op raises in the VM (`self.sched`),
/// read by the [`Scheduler`] after `run()` returns.
#[derive(Debug, Clone)]
pub enum SchedReq {
    /// `go f(args)` — spawn a goroutine running sub `name_idx` with `args`.
    Go { name_idx: u16, args: Vec<Value> },
    /// `make(chan T, cap)` — allocate a channel; its id is pushed to the caller.
    Make { cap: usize },
    /// `ch <- val` — send `val` on channel `ch` (may block).
    Send { ch: i64, val: Value },
    /// `<-ch` — receive from channel `ch`, pushing the value (may block).
    Recv { ch: i64 },
    /// `close(ch)` — close channel `ch`.
    Close { ch: i64 },
}

/// Why the scheduler stopped without the main goroutine finishing.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SchedError {
    /// Every live goroutine is blocked on a channel — no progress is possible.
    Deadlock,
    /// A goroutine faulted (a channel misuse, or a VM runtime error).
    Panic(String),
}

impl std::fmt::Display for SchedError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            SchedError::Deadlock => {
                write!(f, "fatal error: all goroutines are asleep - deadlock!")
            }
            SchedError::Panic(m) => write!(f, "{m}"),
        }
    }
}

/// A channel: a bounded value buffer plus queues of goroutines blocked sending
/// (with their pending value) or receiving.
struct Channel {
    cap: usize,
    buf: VecDeque<Value>,
    closed: bool,
    send_q: VecDeque<(usize, Value)>,
    recv_q: VecDeque<usize>,
}

impl Channel {
    fn new(cap: usize) -> Self {
        Channel {
            cap,
            buf: VecDeque::new(),
            closed: false,
            send_q: VecDeque::new(),
            recv_q: VecDeque::new(),
        }
    }
}

/// The cooperative goroutine scheduler. Construct with [`Scheduler::new`], then
/// [`Scheduler::run`].
pub struct Scheduler<F: FnMut() -> VM> {
    /// Fresh, fully-configured VM factory (same chunk + builtins + hooks) used to
    /// materialize each spawned goroutine. The scheduler positions it at the
    /// goroutine's entry.
    make_vm: F,
    vms: Vec<VM>,
    ready: VecDeque<usize>,
    blocked: HashSet<usize>,
    done: HashSet<usize>,
    chans: Vec<Channel>,
    /// The zero value a receive yields on a closed, empty channel. Frontends set
    /// this to the element type's zero (`Value::Int(0)` by default).
    recv_zero: Value,
}

impl<F: FnMut() -> VM> Scheduler<F> {
    /// Create a scheduler whose goroutines are built by `make_vm`.
    pub fn new(make_vm: F) -> Self {
        Scheduler {
            make_vm,
            vms: Vec::new(),
            ready: VecDeque::new(),
            blocked: HashSet::new(),
            done: HashSet::new(),
            chans: Vec::new(),
            recv_zero: Value::Int(0),
        }
    }

    /// Set the value a receive on a closed, empty channel yields (default `0`).
    pub fn with_recv_zero(mut self, zero: Value) -> Self {
        self.recv_zero = zero;
        self
    }

    /// Drive `main_vm` (goroutine 0) and every goroutine it spawns to completion.
    /// Returns when the main goroutine finishes (Go semantics: the program exits
    /// when `main` returns) or on deadlock / panic.
    pub fn run(mut self, main_vm: VM) -> Result<(), SchedError> {
        self.vms.push(main_vm);
        self.ready.push_back(0);
        self.drive()
    }

    /// Like [`Scheduler::run`], but afterward returns the value of `global` from
    /// the main goroutine (`None` if the program has no such global). Lets a
    /// frontend read `main`'s result out of a scheduled run.
    pub fn run_capturing(mut self, main_vm: VM, global: &str) -> Result<Option<Value>, SchedError> {
        self.vms.push(main_vm);
        self.ready.push_back(0);
        self.drive()?;
        let main = &self.vms[0];
        let val = main
            .chunk
            .names
            .iter()
            .position(|n| n == global)
            .and_then(|i| main.globals.get(i).cloned());
        Ok(val)
    }

    /// The scheduling loop over the ready queue.
    fn drive(&mut self) -> Result<(), SchedError> {
        while let Some(gid) = self.ready.pop_front() {
            self.vms[gid].clear_halt();
            // `VM::run` pops the top of the value stack on exit to return a
            // result. When a goroutine/channel op halts the VM mid-expression,
            // that popped value is live data the op left below its operands —
            // capture it and restore it before servicing the park.
            let popped = match self.vms[gid].run() {
                VMResult::Error(e) => return Err(SchedError::Panic(e)),
                VMResult::Ok(v) => Some(v),
                VMResult::Halted => None,
            };
            match self.vms[gid].take_sched() {
                Some(req) => {
                    if let Some(v) = popped {
                        self.vms[gid].stack.push(v);
                    }
                    self.service(gid, req)?;
                }
                None => {
                    // The goroutine finished (or halted for a frontend reason);
                    // `popped` is its final result, which the caller ignores.
                    self.done.insert(gid);
                    if gid == 0 {
                        // main returned — the program exits.
                        return Ok(());
                    }
                }
            }

            if self.ready.is_empty() && !self.blocked.is_empty() {
                return Err(SchedError::Deadlock);
            }
        }
        Ok(())
    }

    /// Service one scheduling request from goroutine `gid`.
    fn service(&mut self, gid: usize, req: SchedReq) -> Result<(), SchedError> {
        match req {
            SchedReq::Make { cap } => {
                let id = self.chans.len();
                self.chans.push(Channel::new(cap));
                self.vms[gid].stack.push(Value::Int(id as i64));
                self.ready.push_back(gid);
            }
            SchedReq::Go { name_idx, args } => {
                let ngid = self.vms.len();
                let mut vm = (self.make_vm)();
                self.position_goroutine(&mut vm, name_idx, &args)?;
                self.vms.push(vm);
                self.ready.push_back(ngid);
                // The spawner continues immediately after `go`.
                self.ready.push_back(gid);
            }
            SchedReq::Send { ch, val } => self.send(gid, ch, val)?,
            SchedReq::Recv { ch } => self.recv(gid, ch)?,
            SchedReq::Close { ch } => {
                let c = self.chan_mut(ch)?;
                c.closed = true;
                // Wake every blocked receiver with the zero value.
                let woken: Vec<usize> = c.recv_q.drain(..).collect();
                for r in woken {
                    self.vms[r].stack.push(self.recv_zero.clone());
                    self.wake(r);
                }
                self.ready.push_back(gid);
            }
        }
        Ok(())
    }

    /// `ch <- val` from goroutine `gid`.
    fn send(&mut self, gid: usize, ch: i64, val: Value) -> Result<(), SchedError> {
        // Validate + capture what to do without holding the borrow across a wake.
        let (deliver_to, buffered, park) = {
            let c = self.chan_mut(ch)?;
            if c.closed {
                return Err(SchedError::Panic(
                    "panic: send on closed channel".to_string(),
                ));
            }
            if let Some(r) = c.recv_q.pop_front() {
                // A receiver is waiting — hand the value straight to it.
                (Some(r), false, false)
            } else if c.buf.len() < c.cap {
                c.buf.push_back(val.clone());
                (None, true, false)
            } else {
                // No room and no receiver — block the sender with its value.
                c.send_q.push_back((gid, val.clone()));
                (None, false, true)
            }
        };
        if let Some(r) = deliver_to {
            self.vms[r].stack.push(val);
            self.wake(r);
            self.ready.push_back(gid);
        } else if buffered {
            self.ready.push_back(gid);
        } else if park {
            self.blocked.insert(gid);
        }
        Ok(())
    }

    /// `<-ch` from goroutine `gid`.
    fn recv(&mut self, gid: usize, ch: i64) -> Result<(), SchedError> {
        enum Outcome {
            Value(Value),
            WakeSender(usize, Value),
            Zero,
            Park,
        }
        let outcome = {
            let c = self.chan_mut(ch)?;
            if let Some(v) = c.buf.pop_front() {
                // Room freed up — pull a blocked sender's value into the buffer.
                if let Some((s, sv)) = c.send_q.pop_front() {
                    c.buf.push_back(sv);
                    Outcome::WakeSender(s, v)
                } else {
                    Outcome::Value(v)
                }
            } else if let Some((s, sv)) = c.send_q.pop_front() {
                // Unbuffered handoff: take straight from a blocked sender.
                Outcome::WakeSender(s, sv)
            } else if c.closed {
                Outcome::Zero
            } else {
                c.recv_q.push_back(gid);
                Outcome::Park
            }
        };
        match outcome {
            Outcome::Value(v) => {
                self.vms[gid].stack.push(v);
                self.ready.push_back(gid);
            }
            Outcome::WakeSender(s, v) => {
                self.vms[gid].stack.push(v);
                self.wake(s);
                self.ready.push_back(gid);
            }
            Outcome::Zero => {
                let z = self.recv_zero.clone();
                self.vms[gid].stack.push(z);
                self.ready.push_back(gid);
            }
            Outcome::Park => {
                self.blocked.insert(gid);
            }
        }
        Ok(())
    }

    /// Move a blocked goroutine back to the ready queue.
    fn wake(&mut self, gid: usize) {
        self.blocked.remove(&gid);
        self.ready.push_back(gid);
    }

    fn chan_mut(&mut self, ch: i64) -> Result<&mut Channel, SchedError> {
        self.chans
            .get_mut(ch as usize)
            .ok_or_else(|| SchedError::Panic(format!("panic: invalid channel {ch}")))
    }

    /// Position a fresh goroutine VM to start executing sub `name_idx` with
    /// `args`, returning to a terminal ip when the sub returns (so `run()` ends).
    fn position_goroutine(
        &self,
        vm: &mut VM,
        name_idx: u16,
        args: &[Value],
    ) -> Result<(), SchedError> {
        let entry = vm.chunk.find_sub(name_idx).ok_or_else(|| {
            SchedError::Panic(format!("panic: goroutine target sub {name_idx} not found"))
        })?;
        let base = vm.stack.len();
        for a in args {
            vm.stack.push(a.clone());
        }
        vm.frames.push(Frame {
            return_ip: vm.chunk.ops.len(), // returning ends the goroutine's run()
            stack_base: base,
            slots: Vec::new(),
        });
        vm.ip = entry;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{Chunk, ChunkBuilder, Op};

    /// Run `chunk` under a fresh scheduler and read back `global` from main.
    fn run(chunk: Chunk, global: &str) -> Result<Option<Value>, SchedError> {
        let fc = chunk.clone();
        Scheduler::new(move || VM::new(fc.clone())).run_capturing(VM::new(chunk), global)
    }

    #[test]
    fn producer_consumer_over_unbuffered_channel() {
        // main:
        //   ch := make(chan int)          // cap 0 (unbuffered)
        //   go producer(ch)
        //   sum := <-ch + <-ch + <-ch     // synchronizes with the sender
        // producer(ch): send 1, 2, 3
        let mut b = ChunkBuilder::new();
        let prod = b.add_name("producer");
        let ch = b.add_name("ch");
        let sum = b.add_name("sum");

        b.emit(Op::LoadInt(0), 1);
        b.emit(Op::ChanMake, 1);
        b.emit(Op::SetVar(ch), 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::Go(prod, 1), 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::ChanRecv, 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::ChanRecv, 1);
        b.emit(Op::Add, 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::ChanRecv, 1);
        b.emit(Op::Add, 1);
        b.emit(Op::SetVar(sum), 1);
        let skip = b.emit(Op::Jump(0), 1);

        let entry = b.current_pos();
        b.add_sub_entry(prod, entry);
        b.emit(Op::SetSlot(0), 2); // bind the channel arg
        for v in 1..=3 {
            b.emit(Op::GetSlot(0), 2);
            b.emit(Op::LoadInt(v), 2);
            b.emit(Op::ChanSend, 2);
        }
        b.emit(Op::LoadUndef, 2);
        b.emit(Op::ReturnValue, 2);
        b.patch_jump(skip, b.current_pos());

        assert_eq!(run(b.build(), "sum"), Ok(Some(Value::Int(6))));
    }

    #[test]
    fn buffered_channel_holds_values_without_a_receiver() {
        // ch := make(chan int, 2); ch <- 10; ch <- 20; sum := <-ch + <-ch
        // The two sends do not block (buffer capacity 2), all in one goroutine.
        let mut b = ChunkBuilder::new();
        let ch = b.add_name("ch");
        let sum = b.add_name("sum");
        b.emit(Op::LoadInt(2), 1);
        b.emit(Op::ChanMake, 1);
        b.emit(Op::SetVar(ch), 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::LoadInt(10), 1);
        b.emit(Op::ChanSend, 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::LoadInt(20), 1);
        b.emit(Op::ChanSend, 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::ChanRecv, 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::ChanRecv, 1);
        b.emit(Op::Add, 1);
        b.emit(Op::SetVar(sum), 1);
        assert_eq!(run(b.build(), "sum"), Ok(Some(Value::Int(30))));
    }

    #[test]
    fn deadlock_is_detected() {
        // main receives on a channel nobody ever sends to.
        let mut b = ChunkBuilder::new();
        let ch = b.add_name("ch");
        b.emit(Op::LoadInt(0), 1);
        b.emit(Op::ChanMake, 1);
        b.emit(Op::SetVar(ch), 1);
        b.emit(Op::GetVar(ch), 1);
        b.emit(Op::ChanRecv, 1);
        b.emit(Op::Pop, 1);
        assert_eq!(run(b.build(), "x"), Err(SchedError::Deadlock));
    }
}