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    pub fn as_u64(self) -> u64 {
28        self.0
29    }
30}
31
32impl std::fmt::Display for FlowId {
33    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
34        write!(f, "flow#{}", self.0)
35    }
36}
37
38static NEXT_FLOW_ID: AtomicU64 = AtomicU64::new(1);
39
40pub fn next_flow_id() -> FlowId {
41    FlowId(NEXT_FLOW_ID.fetch_add(1, Ordering::Relaxed))
42}
43
44/// Restart policy consulted by a [`super::supervisor::Supervisor`] when a
45/// supervised flow terminates.
46#[derive(Clone, Copy, Debug, PartialEq, Eq)]
47pub enum RestartPolicy {
48    Always,
49    OnFailure,
50    Never,
51}
52
53/// Live counters for one flow (updated only by the worker currently
54/// running it).
55#[derive(Debug, Default)]
56pub struct FlowMetrics {
57    pub instructions: AtomicU64,
58    /// Atomic hops sent (`Send` of [`crate::Value::Message`]).
59    pub messages_sent: AtomicU64,
60    pub messages_received: AtomicU64,
61    pub reschedules: AtomicU64,
62}
63
64/// A single **flow**: VM state, mailbox, and bookkeeping.
65///
66/// This is the unit of work moved by the scheduler — pushed onto worker
67/// deques, stolen, parked inside a [`Mailbox`] on `Receive`, or held by
68/// the timer wheel while sleeping. Flows talk only via **Atomic Hops**
69/// ([`crate::Value::Message`] on `Send`).
70pub struct Flow {
71    pub id: FlowId,
72    pub vm: Vm,
73    pub mailbox: Arc<Mailbox>,
74    pub metrics: Arc<FlowMetrics>,
75    pub restart_policy: RestartPolicy,
76    /// Completion channel consumed by [`super::handle::FlowHandle::join`].
77    pub(crate) completion: oneshot::Sender<FlowOutcome>,
78    /// Set by [`Mailbox::park`] when a hop wins the park race.
79    pub pending_message: Option<crate::bytecode::Value>,
80    /// Destination register of the most recent `Receive` / `ReceiveTimeout`.
81    pub last_receive_dest: Option<u8>,
82    /// Continuation after a `WAITING_SEND` park is admitted.
83    pub(crate) pending_send: Option<PendingSend>,
84    pub(crate) supervisor: Option<super::supervisor::SupervisorLink>,
85    /// Self-authority (rights + native mask). Derived only via [`Cap::attenuate`].
86    pub(crate) authority: Cap,
87    pub(crate) cell: Arc<RevocationCell>,
88    pub(crate) quota: Arc<super::quota::FlowQuota>,
89}
90
91/// What the worker should do after a parked sender's hop is admitted.
92#[derive(Debug, Clone, Copy)]
93pub(crate) enum PendingSend {
94    FireAndForget,
95    Ask {
96        dest_reg: u8,
97        expect_request_id: u64,
98        expect_sender: u64,
99        timeout: Option<Duration>,
100    },
101}
102
103/// Terminal outcome of a flow, delivered to whoever holds its
104/// [`super::handle::FlowHandle`].
105#[derive(Clone, Debug, PartialEq)]
106pub enum FlowOutcome {
107    Completed(crate::bytecode::Value),
108    Failed(String),
109}
110
111impl Flow {
112    pub fn new(
113        id: FlowId,
114        vm: Vm,
115        mailbox: Arc<Mailbox>,
116        restart_policy: RestartPolicy,
117        completion: oneshot::Sender<FlowOutcome>,
118    ) -> Self {
119        let cell = Arc::new(RevocationCell::new());
120        let authority = Cap::root(
121            CapTarget::Flow(id.as_u64()),
122            CapRights::NONE,
123            None,
124            cell.as_ref(),
125        );
126        Self {
127            id,
128            vm,
129            mailbox,
130            metrics: Arc::new(FlowMetrics::default()),
131            restart_policy,
132            completion,
133            pending_message: None,
134            last_receive_dest: None,
135            pending_send: None,
136            supervisor: None,
137            authority,
138            cell,
139            quota: Arc::new(super::quota::FlowQuota::from_config(
140                super::quota::QuotaConfig::default(),
141            )),
142        }
143    }
144
145    pub(crate) fn complete(self, outcome: FlowOutcome) {
146        self.completion.send(outcome);
147    }
148}
byteflow/scheduler/process.rs

byteflow/scheduler/
process.rs
