pipe_io/
driver.rs

1//! Pipeline execution drivers.
2//!
3//! [`SyncDriver`] runs the pipeline single-threaded on the calling
4//! thread. [`ThreadedDriver`] (under `std`) drives the pipeline on a
5//! background OS thread. Custom executors can implement the [`Driver`]
6//! trait and be selected at run time via
7//! [`crate::Pipeline::run_with`].
8//!
9//! The [`Driver`] trait carries the stricter `Send` bound (matching
10//! [`ThreadedDriver`]). The inherent `SyncDriver::run` method keeps the
11//! looser bound for callers that drive non-`Send` sources on the
12//! current thread.
13
14#[cfg(feature = "std")]
15use crate::error::Error;
16use crate::error::Result;
17use crate::pipeline::Pipeline;
18use crate::source::Source;
19
20#[cfg(feature = "std")]
21use core::time::Duration;
22
23/// Statistics returned by a successful pipeline run.
24#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
25pub struct RunStats {
26    /// Number of items pulled from the source.
27    pub items_in: u64,
28    /// Wall-clock duration of the run. Always `Duration::ZERO` under
29    /// `no_std` (no monotonic clock available).
30    #[cfg(feature = "std")]
31    pub duration: Duration,
32}
33
34/// Generic executor for built pipelines.
35///
36/// Implement for custom executors (tokio's runtime, a rayon thread
37/// pool, a sharded worker farm, etc.). Built-in implementations are
38/// [`SyncDriver`] and [`ThreadedDriver`].
39///
40/// The trait requires every part of the pipeline to be `Send`: the
41/// source, its item type, and its error type. This matches
42/// [`ThreadedDriver`]'s natural bounds and lets a custom executor
43/// move a pipeline to another thread without extra constraints. If
44/// you need to drive a non-`Send` source on the calling thread, use
45/// [`SyncDriver::run`] (the inherent method) directly; that path
46/// keeps the looser bound.
47///
48/// # Example
49///
50/// ```
51/// use pipe_io::driver::{Driver, RunStats, SyncDriver};
52/// use pipe_io::{sink::NullSink, Pipeline, Result};
53///
54/// fn run_anything<D: Driver>(driver: D) -> Result<RunStats> {
55///     let pipeline = Pipeline::from_iter(0..5).sink(NullSink::<i32>::new());
56///     driver.run(pipeline)
57/// }
58///
59/// run_anything(SyncDriver::new()).unwrap();
60/// ```
61pub trait Driver {
62    /// Drive a pipeline to completion.
63    ///
64    /// # Errors
65    ///
66    /// Returns the first error produced by the source, any stage, or
67    /// the sink.
68    fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
69    where
70        S: Source + Send + 'static,
71        S::Item: Send + 'static,
72        S::Error: Send + 'static;
73}
74
75/// Single-threaded pipeline driver.
76#[derive(Debug, Default, Clone, Copy)]
77pub struct SyncDriver;
78
79impl SyncDriver {
80    /// Construct a new sync driver.
81    #[must_use]
82    pub const fn new() -> Self {
83        Self
84    }
85
86    /// Drive a pipeline to completion on the calling thread.
87    ///
88    /// This inherent method carries a looser bound than the
89    /// [`Driver`] trait impl: the source and its item/error types do
90    /// not need to be `Send`. Use this method directly when the
91    /// pipeline cannot satisfy `Send` (for example, a source holding
92    /// an `Rc`). When the `Send` bounds hold, the trait impl and the
93    /// inherent method behave identically.
94    ///
95    /// # Errors
96    ///
97    /// Returns the first error produced by the source, any stage, or
98    /// the sink.
99    pub fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
100    where
101        S: Source + 'static,
102        S::Item: 'static,
103        S::Error: 'static,
104    {
105        crate::pipeline::run_sync(pipeline)
106    }
107}
108
109impl Driver for SyncDriver {
110    fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
111    where
112        S: Source + Send + 'static,
113        S::Item: Send + 'static,
114        S::Error: Send + 'static,
115    {
116        crate::pipeline::run_sync(pipeline)
117    }
118}
119
120/// Background-thread pipeline driver. Spawns a single OS thread that
121/// runs the pipeline; the calling thread blocks on join.
122#[cfg(feature = "std")]
123#[cfg_attr(docsrs, doc(cfg(feature = "std")))]
124#[derive(Debug, Default, Clone, Copy)]
125pub struct ThreadedDriver;
126
127#[cfg(feature = "std")]
128impl ThreadedDriver {
129    /// Construct a new threaded driver.
130    #[must_use]
131    pub const fn new() -> Self {
132        Self
133    }
134
135    /// Drive a pipeline to completion on a spawned thread.
136    ///
137    /// # Errors
138    ///
139    /// Returns the first error produced by the source, any stage, or
140    /// the sink. Returns [`Error::Cancelled`] if the worker thread
141    /// panics.
142    pub fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
143    where
144        S: Source + Send + 'static,
145        S::Item: Send + 'static,
146        S::Error: Send + 'static,
147    {
148        let handle = std::thread::spawn(move || crate::pipeline::run_sync(pipeline));
149        match handle.join() {
150            Ok(result) => result,
151            Err(_) => Err(Error::Cancelled),
152        }
153    }
154}
155
156#[cfg(feature = "std")]
157impl Driver for ThreadedDriver {
158    fn run<S>(self, pipeline: Pipeline<S>) -> Result<RunStats>
159    where
160        S: Source + Send + 'static,
161        S::Item: Send + 'static,
162        S::Error: Send + 'static,
163    {
164        Self::run(self, pipeline)
165    }
166}
pipe_io/driver.rs

pipe_io/
driver.rs
