byteflow/scheduler/
process.rs

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

byteflow/scheduler/
process.rs
