byteflow/scheduler/
process.rs

1use std::sync::atomic::{AtomicU64, Ordering};
2use std::sync::Arc;
3use std::time::Duration;
4
5use crate::bytecode::{Cap, CapRights, CapTarget, RevocationCell};
6use crate::vm::Vm;
7
8use super::mailbox::Mailbox;
9use super::oneshot;
10
11/// Identifier of a **flow** — Byteflow's unit of concurrent work.
12///
13/// Host APIs and the directory key on this type. Inside messages it appears
14/// as [`crate::Value::Pid`] (`Message.sender` / `msg_sender`) for **identity**.
15/// Bytecode addressing uses [`crate::Value::Cap`] (FlowCap) — a Pid is not a
16/// Send/Ask authority token.
17///
18/// Backed by a single global, wait-free `AtomicU64` counter rather than
19/// anything derived from memory addresses: ids must stay unique for the
20/// lifetime of the runtime and must **never** be reused, or a stale id in
21/// someone's registers could address a *different* later flow (ABA) via the
22/// host/`Directory` path.
23#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
24pub struct FlowId(pub(crate) u64);
25
26impl FlowId {
27    /// Embedder origin for host `Runtime::send`. Never spawned, never finalized.
28    pub const HOST: FlowId = FlowId(0);
29
30    pub fn as_u64(self) -> u64 {
31        self.0
32    }
33
34    #[inline]
35    pub fn is_host(self) -> bool {
36        self.0 == 0
37    }
38}
39
40impl std::fmt::Display for FlowId {
41    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
42        write!(f, "flow#{}", self.0)
43    }
44}
45
46static NEXT_FLOW_ID: AtomicU64 = AtomicU64::new(1);
47
48pub fn next_flow_id() -> FlowId {
49    FlowId(NEXT_FLOW_ID.fetch_add(1, Ordering::Relaxed))
50}
51
52/// Restart policy consulted by a [`super::supervisor::Supervisor`] when a
53/// supervised flow terminates.
54#[derive(Clone, Copy, Debug, PartialEq, Eq)]
55pub enum RestartPolicy {
56    Always,
57    OnFailure,
58    Never,
59}
60
61/// Live counters for one flow (updated only by the worker currently
62/// running it).
63#[derive(Debug, Default)]
64pub struct FlowMetrics {
65    pub instructions: AtomicU64,
66    /// Atomic hops sent (`Send` of [`crate::Value::Message`]).
67    pub messages_sent: AtomicU64,
68    pub messages_received: AtomicU64,
69    pub reschedules: AtomicU64,
70}
71
72/// A single **flow**: VM state, mailbox, and bookkeeping.
73///
74/// This is the unit of work moved by the scheduler — pushed onto worker
75/// deques, stolen, parked inside a [`Mailbox`] on `Receive`, or held by
76/// the timer wheel while sleeping. Flows talk only via **Atomic Hops**
77/// ([`crate::Value::Message`] on `Send`).
78pub struct Flow {
79    pub id: FlowId,
80    pub vm: Vm,
81    pub mailbox: Arc<Mailbox>,
82    pub metrics: Arc<FlowMetrics>,
83    pub restart_policy: RestartPolicy,
84    /// Completion channel consumed by [`super::handle::FlowHandle::join`].
85    pub(crate) completion: oneshot::Sender<FlowOutcome>,
86    /// Set by [`Mailbox::park`] when a hop wins the park race.
87    pub pending_message: Option<crate::bytecode::Value>,
88    /// Destination register of the most recent `Receive` / `ReceiveTimeout`.
89    pub last_receive_dest: Option<u8>,
90    /// Continuation after a `WAITING_SEND` park is admitted.
91    pub(crate) pending_send: Option<PendingSend>,
92    pub(crate) supervisor: Option<super::supervisor::SupervisorLink>,
93    /// Self-authority (rights + native mask). Derived only via [`Cap::attenuate`].
94    pub(crate) authority: Cap,
95    pub(crate) cell: Arc<RevocationCell>,
96    pub(crate) quota: Arc<super::quota::FlowQuota>,
97}
98
99/// What the worker should do after a parked sender's hop is admitted.
100#[derive(Debug, Clone, Copy)]
101pub(crate) enum PendingSend {
102    FireAndForget,
103    Ask {
104        dest_reg: u8,
105        expect_request_id: u64,
106        expect_sender: u64,
107        timeout: Option<Duration>,
108    },
109}
110
111/// Terminal outcome of a flow, delivered to whoever holds its
112/// [`super::handle::FlowHandle`].
113#[derive(Clone, Debug, PartialEq)]
114pub enum FlowOutcome {
115    Completed(crate::bytecode::Value),
116    Failed(String),
117}
118
119impl Flow {
120    pub fn new(
121        id: FlowId,
122        vm: Vm,
123        mailbox: Arc<Mailbox>,
124        restart_policy: RestartPolicy,
125        completion: oneshot::Sender<FlowOutcome>,
126    ) -> Self {
127        let cell = Arc::new(RevocationCell::new());
128        let authority = Cap::root(
129            CapTarget::Flow(id.as_u64()),
130            CapRights::NONE,
131            None,
132            cell.as_ref(),
133        );
134        Self {
135            id,
136            vm,
137            mailbox,
138            metrics: Arc::new(FlowMetrics::default()),
139            restart_policy,
140            completion,
141            pending_message: None,
142            last_receive_dest: None,
143            pending_send: None,
144            supervisor: None,
145            authority,
146            cell,
147            quota: Arc::new(super::quota::FlowQuota::from_config(
148                super::quota::QuotaConfig::default(),
149            )),
150        }
151    }
152
153    pub(crate) fn complete(self, outcome: FlowOutcome) {
154        self.completion.send(outcome);
155    }
156}
byteflow/scheduler/process.rs

byteflow/scheduler/
process.rs
