byteflow/scheduler/
handle.rs

1use std::time::{Duration, Instant};
2
3use super::oneshot::{self, JoinState};
4use super::process::{FlowId, FlowOutcome};
5
6/// A reference to a spawned **flow**, returned by
7/// [`super::runtime::Runtime::spawn`].
8///
9/// Holding a `FlowHandle` does not keep the flow alive — it is a way to
10/// (a) read its [`FlowId`] for addressing with `Send`, and (b) block the
11/// *calling native thread* until it finishes via [`FlowHandle::join`].
12/// Dropping without joining is fine (fire-and-forget).
13pub struct FlowHandle {
14    pub(crate) id: FlowId,
15    pub(crate) receiver: oneshot::Receiver<FlowOutcome>,
16}
17
18impl FlowHandle {
19    pub fn id(&self) -> FlowId {
20        self.id
21    }
22
23    /// Block the current (native) thread until the flow terminates.
24    /// **Never call from inside a worker / from bytecode.**
25    ///
26    /// Returns [`FlowOutcome::Failed`] rather than blocking forever if the
27    /// flow was destroyed without producing an outcome — most commonly
28    /// `Runtime::shutdown` while the flow was suspended, since shutdown does
29    /// not drain flows out of the timer or the worker deques. The message
30    /// comes from [`super::error::RuntimeError::Abandoned`], so it is
31    /// distinguishable from a flow that genuinely faulted.
32    pub fn join(self) -> FlowOutcome {
33        match self.receiver.join() {
34            Ok(outcome) => outcome,
35            Err(e) => FlowOutcome::Failed(e.to_string()),
36        }
37    }
38
39    /// The outcome if the flow has already terminated, `None` if it is still
40    /// running. Never waits.
41    ///
42    /// For embedders that drive their own loop and cannot surrender a thread
43    /// to [`FlowHandle::join`]. Takes `&self`, so it can be polled until it
44    /// answers; note that the outcome is handed out exactly once, and a
45    /// further poll after that reports
46    /// [`super::error::RuntimeError::AlreadyCollected`] as
47    /// [`FlowOutcome::Failed`] rather than repeating it.
48    pub fn try_join(&self) -> Option<FlowOutcome> {
49        Self::settle(self.receiver.try_join())
50    }
51
52    /// Wait up to `timeout` for the flow to terminate; `None` if it is still
53    /// running when the bound elapses.
54    ///
55    /// This is the variant to reach for in anything with a deadline — a
56    /// control loop, a watchdog, a test harness — since it is the only join
57    /// whose worst-case duration the caller chooses.
58    pub fn join_timeout(&self, timeout: Duration) -> Option<FlowOutcome> {
59        Self::settle(self.receiver.join_timeout(timeout))
60    }
61
62    /// [`FlowHandle::join_timeout`] against an absolute deadline, for callers
63    /// that already track one and must not have it drift across repeated
64    /// waits.
65    pub fn join_deadline(&self, deadline: Instant) -> Option<FlowOutcome> {
66        Self::settle(self.receiver.join_deadline(deadline))
67    }
68
69    /// Collapse a bounded-wait result into the handle's public vocabulary:
70    /// `None` for "still running", `Some(Failed)` for an infrastructure
71    /// error, since an embedder polling a handle has no separate error
72    /// channel to report one on.
73    fn settle(state: Result<JoinState<FlowOutcome>, super::error::RuntimeError>) -> Option<FlowOutcome> {
74        match state {
75            Ok(JoinState::Ready(outcome)) => Some(outcome),
76            Ok(JoinState::Pending) => None,
77            Err(e) => Some(FlowOutcome::Failed(e.to_string())),
78        }
79    }
80}
byteflow/scheduler/handle.rs

byteflow/scheduler/
handle.rs
