byteflow/scheduler/
process.rs

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

byteflow/scheduler/
process.rs
