swaits/dagx
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

WIP: switch to announcement, remove extract input mess

Browse Source
This commit is contained in:
swaits committed 2025-10-14 09:01:28 -06:00
1 parent 82cbf15b85
commit 6de4d43db1
48 files changed
+685 -1187
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

.beads/dagx.db
BIN
View File
Binary file not shown.
CLAUDE.md
+1
View File
@@ -0,0 +1 @@
We track work in Beads instead of Markdown. Run \`bd quickstart\` to see how.
Cargo.lock
Generated
+10
View File
@@ -32,6 +32,15 @@ version = "0.1.6"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "4b46cbb362ab8752921c97e041f5e366ee6297bd428a31275b9fcf1e380f7299"
[[package]]
name = "announcement"
version = "0.1.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "0e58b38a7f1ae19278fb8b7aa4c3ac3957f0e6389a132522789b1dd2e1912189"
dependencies = [
"event-listener 5.4.1",
]
[[package]]
name = "anstyle"
version = "1.0.13"
@@ -436,6 +445,7 @@ dependencies = [
name = "dagx"
version = "0.3.0"
dependencies = [
"announcement",
"async-executor",
"async-std",
"async-trait",
Cargo.toml
+1
View File
@@ -25,6 +25,7 @@ tracing = ["dep:tracing"]
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin)', 'cfg(tarpaulin_include)'] }
[dependencies]
announcement = "0.1.0"
futures = "0.3"
parking_lot = "0.12"
dagx-macros = { version = "0.3.0", path = "./dagx-macros" }
benches/basic/creation.rs
+1 -1
View File
@@ -1,7 +1,7 @@
//! DAG creation benchmarks
use criterion::Criterion;
use dagx::{DagRunner, Task};
use dagx::DagRunner;
use std::hint::black_box;
// Simple task for benchmarking
benches/basic/execution.rs
+1 -1
View File
@@ -1,7 +1,7 @@
//! Basic DAG execution benchmarks
use criterion::Criterion;
use dagx::{task_fn, DagRunner, Task};
use dagx::{task_fn, DagRunner};
pub fn bench_dag_execution(c: &mut Criterion) {
let rt = tokio::runtime::Runtime::new().unwrap();
dagx-macros/src/lib.rs
+135 -172
View File
@@ -1,17 +1,17 @@
//! Procedural macros for dagx
//!
//! This crate provides the `#[task]` attribute macro that automatically implements
//! the `Task` trait by deriving Input and Output types from the `run()` method signature.
//! This crate provides the `#[task]` attribute macro that automatically generates
//! an `execute()` method by deriving Input and Output types from the `run()` method signature.
use proc_macro::TokenStream;
use quote::quote;
use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType, Type};
/// Attribute macro to automatically implement the `Task` trait.
/// Attribute macro to automatically generate the execution logic for a task.
///
/// Apply this to an `impl` block containing a `run()` method (sync or async). The macro:
/// - Derives `Input` and `Output` types from the `run()` signature
/// - Automatically implements the `Task` trait
/// - Derives input types from the `run()` signature parameters
/// - Generates an `execute()` method that handles listener extraction and announcement
/// - Supports both sync and async run methods
/// - Supports stateless (no self) and stateful (&self, &mut self) tasks
/// - Handles various input patterns (no inputs, single input, multiple inputs)
@@ -25,7 +25,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
/// Unit structs for pure computations. Use **no `self` parameter**:
///
/// ```ignore
/// use dagx::{task, Task};
/// use dagx::task;
///
/// struct Add;
///
@@ -42,7 +42,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
/// Tasks that read configuration or constant data. Use **`&self`**:
///
/// ```ignore
/// use dagx::{task, Task};
/// use dagx::task;
///
/// struct Multiplier {
/// factor: i32,
@@ -61,7 +61,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
/// Tasks that accumulate or modify state. Use **`&mut self`**:
///
/// ```ignore
/// use dagx::{task, Task};
/// use dagx::task;
///
/// struct Counter {
/// count: i32,
@@ -81,7 +81,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
/// ## No Inputs (Source Tasks)
///
/// ```ignore
/// use dagx::{task, Task};
/// use dagx::task;
///
/// struct LoadData {
/// value: i32,
@@ -93,110 +93,51 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
/// self.value
/// }
/// }
/// ```ignore
/// ```
///
/// ## Single Input
///
/// ```ignore
/// use dagx::{task, Task};
/// use dagx::task;
///
/// struct Double;
///
/// #[task]
/// impl Double {
/// async fn run(&mut self, input: &i32) -> i32 {
/// async fn run(input: &i32) -> i32 {
/// input * 2
/// }
/// }
/// ```ignore
/// ```
///
/// ## Multiple Inputs (up to 8)
///
/// ```ignore
/// use dagx::{task, Task};
/// use dagx::task;
///
/// struct Combine;
///
/// #[task]
/// impl Combine {
/// async fn run(&mut self, a: &i32, b: &String, c: &bool) -> String {
/// async fn run(a: &i32, b: &String, c: &bool) -> String {
/// format!("{}: {} ({})", b, a, c)
/// }
/// }
/// ```ignore
/// ```
///
/// # Requirements
///
/// - The impl block must contain exactly one `async fn run()` method
/// - The `run()` method can be stateless (no self parameter) or stateful (`&mut self`)
/// - All input parameters must be references (e.g., `&i32`, not `i32`)
/// - The macro requires `Task` to be in scope: `use dagx::Task;`
/// - For stateless tasks, the struct must implement `Default` (e.g., unit structs)
///
/// # Generated Code
///
/// The macro transforms your implementation into a full `Task` trait implementation.
///
/// For stateless tasks (no self parameter):
///
/// ```ignore
/// // Your code:
/// #[task]
/// impl Add {
/// async fn run(a: &i32, b: &i32) -> i32 {
/// a + b
/// }
/// }
///
/// // Generated:
/// impl Task for Add {
/// type Input = (i32, i32);
/// type Output = i32;
///
/// async fn run(&mut self, input: Self::Input) -> Self::Output {
/// let (a, b) = input;
/// Self::run_impl(&a, &b).await
/// }
/// }
///
/// impl Add {
/// #[inline]
/// async fn run_impl(a: &i32, b: &i32) -> i32 {
/// a + b
/// }
/// }
/// ```ignore
///
/// For stateful tasks (with &mut self):
///
/// ```ignore
/// // Your code:
/// #[task]
/// impl Counter {
/// async fn run(&mut self, inc: &i32) -> i32 {
/// self.count += inc;
/// self.count
/// }
/// }
///
/// // Generated:
/// impl Task for Counter {
/// type Input = i32;
/// type Output = i32;
///
/// async fn run(&mut self, input: Self::Input) -> Self::Output {
/// let inc = input;
/// self.run_impl(&inc).await
/// }
/// }
///
/// impl Counter {
/// async fn run_impl(&mut self, inc: &i32) -> i32 {
/// self.count += inc;
/// self.count
/// }
/// }
/// ```ignore
/// The macro transforms your implementation into an `execute()` method that:
/// - Extracts inputs from announcement listeners
/// - Calls your run() method
/// - Announces the result to dependents
/// - Returns the Arc-wrapped output
#[proc_macro_attribute]
pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
let impl_block = parse_macro_input!(item as ItemImpl);
@@ -220,7 +161,7 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
"impl block must contain a run() method\n\n\
Expected signature: fn run(&mut self, ...) -> OutputType\n\
or: async fn run(&mut self, ...) -> OutputType\n\
The #[task] macro requires a run() method to implement the Task trait.",
The #[task] macro requires a run() method to implement task execution.",
)
.to_compile_error()
.into();
@@ -287,31 +228,6 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
ReturnType::Type(_, ty) => ty.clone(),
};
// Build Input type based on parameter count
let input_type = match params.len() {
0 => quote! { () },
1 => {
let (_name, ty) = &params[0];
quote! { #ty }
}
_ => {
let types: Vec<_> = params.iter().map(|(_, ty)| ty).collect();
quote! { ( #(#types),* ) }
}
};
// Generate parameter destructuring for the wrapper run() method
let (param_destructure, param_refs) = if params.is_empty() {
(quote! { _ }, quote! {})
} else if params.len() == 1 {
let (name, _) = &params[0];
(quote! { #name }, quote! { &#name })
} else {
let names: Vec<_> = params.iter().map(|(name, _)| name).collect();
let refs: Vec<_> = params.iter().map(|(name, _)| quote! { &#name }).collect();
(quote! { ( #(#names),* ) }, quote! { #(#refs),* })
};
// Clone the run method and rename it to run_impl
let mut run_impl_method = run_method.clone();
run_impl_method.sig.ident = syn::Ident::new("run_impl", run_method.sig.ident.span());
@@ -323,81 +239,128 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
.iter()
.any(|arg| matches!(arg, FnArg::Receiver(_)));
// Generate the Task trait implementation based on whether we have self and async/sync
let expanded = if has_self_receiver {
// Stateful task - consumes self but delegates to a method that borrows
// Generate listener extraction code
let listener_extractions: Vec<_> = params
.iter()
.enumerate()
.map(|(i, (name, ty))| {
quote! {
let #name = {
if listeners.len() <= #i {
return Err(format!(
"Expected at least {} listener(s), got {}",
#i + 1,
listeners.len()
));
}
let listener = listeners[#i]
.downcast_ref::<announcement::Listener<#ty>>()
.ok_or_else(|| {
format!("Type mismatch: expected Listener<{}> at index {}",
stringify!(#ty), #i)
})?;
listener.listen().await
.map_err(|_| "Listener closed before receiving value")?
};
}
})
.collect();
// Generate parameter refs for the run call
let param_refs: Vec<_> = params.iter().map(|(name, _)| quote! { &#name }).collect();
// Generate the execute method body based on whether we have self and async/sync
let run_call = if has_self_receiver {
if is_async {
// Async with self
quote! {
impl Task for #struct_name {
type Input = #input_type;
type Output = #output_type;
async fn run(mut self, input: Self::Input) -> Self::Output {
let #param_destructure = input;
self.run_impl(#param_refs).await
}
}
impl #struct_name {
#run_impl_method
}
}
quote! { self.run_impl(#(#param_refs),*).await }
} else {
// Sync with self - wrap in async block
quote! {
impl Task for #struct_name {
type Input = #input_type;
type Output = #output_type;
async fn run(mut self, input: Self::Input) -> Self::Output {
let #param_destructure = input;
self.run_impl(#param_refs)
}
}
impl #struct_name {
#run_impl_method
}
}
quote! { self.run_impl(#(#param_refs),*) }
}
} else if is_async {
quote! { Self::run_impl(#(#param_refs),*).await }
} else {
// Stateless task
if is_async {
// Async stateless
quote! {
impl Task for #struct_name {
type Input = #input_type;
type Output = #output_type;
quote! { Self::run_impl(#(#param_refs),*) }
};
async fn run(self, input: Self::Input) -> Self::Output {
let #param_destructure = input;
Self::run_impl(#param_refs).await
}
}
// Generate the self parameter for execute method
let self_param = if has_self_receiver {
quote! { mut self }
} else {
quote! { self }
};
impl #struct_name {
#[inline]
#run_impl_method
}
// Generate the Input type based on parameter count
let input_type = match params.len() {
0 => quote! { () },
1 => {
let (_, ty) = &params[0];
quote! { #ty }
}
_ => {
let types: Vec<_> = params.iter().map(|(_, ty)| ty).collect();
quote! { (#(#types,)*) }
}
};
// Generate the complete execute method and TaskExecute trait implementation
let expanded = quote! {
impl #struct_name {
#[doc(hidden)]
pub async fn execute(
#self_param,
listeners: Vec<Box<dyn std::any::Any + Send>>,
announcer: Box<dyn std::any::Any + Send>,
) -> Result<std::sync::Arc<dyn std::any::Any + Send + Sync>, String> {
// Extract inputs from listeners
#(#listener_extractions)*
// Call the user's run method
let output: #output_type = #run_call;
// Wrap in Arc for sharing
let output_arc = std::sync::Arc::new(output);
// Announce the result
let announcer_typed = announcer
.downcast::<announcement::Announcer<#output_type>>()
.map_err(|_| "Type mismatch: announcer type does not match task output type")?;
let _ = announcer_typed.announce((*output_arc).clone());
// Return Arc-wrapped output
Ok(output_arc as std::sync::Arc<dyn std::any::Any + Send + Sync>)
}
} else {
// Sync stateless
quote! {
impl Task for #struct_name {
type Input = #input_type;
type Output = #output_type;
async fn run(self, input: Self::Input) -> Self::Output {
let #param_destructure = input;
Self::run_impl(#param_refs)
}
}
#run_impl_method
}
impl #struct_name {
#[inline]
#run_impl_method
}
// Implement TaskExecute trait for type erasure
impl dagx::TaskExecute for #struct_name {
type Input = #input_type;
type Output = #output_type;
fn execute(
self,
listeners: Vec<Box<dyn std::any::Any + Send>>,
announcer: Box<dyn std::any::Any + Send>,
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::sync::Arc<dyn std::any::Any + Send + Sync>, String>> + Send>> {
Box::pin(async move {
self.execute(listeners, announcer).await
})
}
fn create_announcement_erased() -> (Box<dyn std::any::Any + Send>, Box<dyn std::any::Any + Send>) {
let (announcer, announcement) = announcement::Announcement::<#output_type>::new();
(
Box::new(announcer) as Box<dyn std::any::Any + Send>,
Box::new(announcement) as Box<dyn std::any::Any + Send>,
)
}
fn create_listener_erased(announcement: &dyn std::any::Any) -> Box<dyn std::any::Any + Send> {
let announcement = announcement
.downcast_ref::<announcement::Announcement<#output_type>>()
.expect("Type mismatch: announcement type does not match task output type");
Box::new(announcement.listener()) as Box<dyn std::any::Any + Send>
}
}
};
examples/01_basic.rs
+1 -1
View File
@@ -30,7 +30,7 @@
//! Result: 5
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
// Step 1: Define a source task (no dependencies)
//
examples/02_fan_out.rs
+1 -1
View File
@@ -55,7 +55,7 @@
//! Squared: 100
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
// Source task with a value field
struct Source {
examples/03_fan_in.rs
+1 -1
View File
@@ -61,7 +61,7 @@
//! Active: true
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
struct FetchName;
examples/04_parallel_computation.rs
+21 -11
View File
@@ -52,7 +52,7 @@
//! This proves dagx runs tasks on multiple CPU cores simultaneously.
//! ```
use dagx::{task, DagResult, DagRunner, Task};
use dagx::{task, DagResult, DagRunner};
use std::time::{Duration, Instant};
use tokio::time::sleep;
@@ -144,22 +144,32 @@ async fn main() -> DagResult<()> {
};
// ============================================================================
// SEQUENTIAL EXECUTION: Run the same 4 tasks one after another
// SEQUENTIAL EXECUTION: Run tasks one at a time in separate DAG executions
// ============================================================================
println!("Running SEQUENTIAL computation (same work, one at a time)...");
let (sequential_time, sequential_result) = {
let start = Instant::now();
// Run tasks sequentially (not using DAG, just raw execution)
let task1 = ComputeSum::new(1, 251, WORK_DELAY);
let task2 = ComputeSum::new(251, 501, WORK_DELAY);
let task3 = ComputeSum::new(501, 751, WORK_DELAY);
let task4 = ComputeSum::new(751, 1001, WORK_DELAY);
// Run each task separately to force sequential execution
let dag1 = DagRunner::new();
let sum1_handle = dag1.add_task(ComputeSum::new(1, 251, WORK_DELAY));
dag1.run(|fut| { tokio::spawn(fut); }).await?;
let r1 = dag1.get(sum1_handle)?;
let r1 = task1.run(()).await;
let r2 = task2.run(()).await;
let r3 = task3.run(()).await;
let r4 = task4.run(()).await;
let dag2 = DagRunner::new();
let sum2_handle = dag2.add_task(ComputeSum::new(251, 501, WORK_DELAY));
dag2.run(|fut| { tokio::spawn(fut); }).await?;
let r2 = dag2.get(sum2_handle)?;
let dag3 = DagRunner::new();
let sum3_handle = dag3.add_task(ComputeSum::new(501, 751, WORK_DELAY));
dag3.run(|fut| { tokio::spawn(fut); }).await?;
let r3 = dag3.get(sum3_handle)?;
let dag4 = DagRunner::new();
let sum4_handle = dag4.add_task(ComputeSum::new(751, 1001, WORK_DELAY));
dag4.run(|fut| { tokio::spawn(fut); }).await?;
let r4 = dag4.get(sum4_handle)?;
let result = r1 + r2 + r3 + r4;
let elapsed = start.elapsed();
examples/circuit_breaker.rs
+1 -1
View File
@@ -150,7 +150,7 @@
//! Result 7: Ok("Success")
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use std::sync::Arc;
use std::time::Instant;
use tokio::sync::Mutex;
examples/complex_dag.rs
+1 -1
View File
@@ -66,7 +66,7 @@
//! Combined: 6 items, sum=57, avg=9.50
//! ```
use dagx::{task, DagResult, DagRunner, Task};
use dagx::{task, DagResult, DagRunner};
// Extract: Load data from different sources
struct LoadData {
examples/conditional_workflow.rs
+1 -1
View File
@@ -55,7 +55,7 @@
//! Expected: Categories: [low, invalid, high]
//! ```
use dagx::{task, DagResult, DagRunner, Task};
use dagx::{task, DagResult, DagRunner};
// Task that validates and processes input
struct ValidateAndDouble {
examples/data_pipeline.rs
+1 -1
View File
@@ -75,7 +75,7 @@
//! ============================
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
// Data source tasks
struct LoadValue1;
examples/debug_tracing.rs
+1 -1
View File
@@ -114,7 +114,7 @@
//! FulfillOrder: 30ms
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use std::time::Instant;
use tokio::time::{sleep, Duration};
examples/error_handling.rs
+1 -1
View File
@@ -114,7 +114,7 @@
//! === Error Handling Example Complete ===
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
// Source task that provides a string input
struct StringSource(&'static str);
examples/large_dag.rs
+1 -1
View File
@@ -109,7 +109,7 @@
//! - Memory overhead is minimal (~ 200 bytes per task)
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use std::time::Instant;
// Simple computation task
examples/parallelism_proof.rs
+1 -1
View File
@@ -47,7 +47,7 @@
//! This proves TRUE PARALLEL EXECUTION!
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use std::time::{Duration, Instant};
use tokio::time::sleep;
examples/retry_strategies.rs
+1 -1
View File
@@ -152,7 +152,7 @@
//! Result: Err("Retry exhausted after 3 attempts: Connection timeout")
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use tokio::time::{sleep, Duration};
// Retry strategy configuration
examples/timeout.rs
+1 -1
View File
@@ -98,7 +98,7 @@
//! === Timeout Example Complete ===
//! ```
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use tokio::time::{sleep, timeout, Duration};
// Fast task that completes quickly
examples/tracing_example.rs
+1 -1
View File
@@ -33,7 +33,7 @@
//! - **ERROR**: Errors, panics, cycle detection
//!
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use tracing_subscriber::{fmt, EnvFilter};
// Simple source task
src/builder.rs
+3 -3
View File
@@ -4,8 +4,8 @@
//! The type parameter tracks whether dependencies have been specified yet.
use crate::deps::DepsTuple;
use crate::node::TaskExecute;
use crate::runner::DagRunner;
use crate::task::Task;
use crate::types::{NodeId, TaskHandle};
#[cfg(feature = "tracing")]
@@ -69,13 +69,13 @@ impl IsUnitType for () {}
/// assert_eq!(dag.get(&b).unwrap(), 20);
/// # };
/// ```
pub struct TaskBuilder<'a, Tk: Task, Deps> {
pub struct TaskBuilder<'a, Tk: TaskExecute, Deps> {
pub(crate) id: NodeId,
pub(crate) dag: &'a DagRunner,
pub(crate) _phantom: std::marker::PhantomData<(Tk, Deps)>,
}
impl<'a, Tk: Task, Deps> TaskBuilder<'a, Tk, Deps> {
impl<'a, Tk: TaskExecute, Deps> TaskBuilder<'a, Tk, Deps> {
/// Provide all dependencies exactly once as a tuple.
///
/// The dependencies must match the task's `Input` type exactly:
src/builder/tests.rs
+3 -1
View File
@@ -1,8 +1,10 @@
//! Unit tests for builder module
// Allow tests to reference dagx:: paths (for proc macro generated code)
extern crate self as dagx;
use crate::builder::IsUnitType;
use crate::runner::DagRunner;
use crate::task::Task;
use crate::types::TaskHandle;
// Initialize tracing subscriber for tests (idempotent)
src/deps.rs
+4 -4
View File
@@ -8,7 +8,7 @@
//! use the provided implementations.
use crate::builder::TaskBuilder;
use crate::task::Task;
use crate::node::TaskExecute;
use crate::types::{NodeId, TaskHandle};
pub(crate) trait DepsTuple<Input> {
@@ -48,13 +48,13 @@ impl<T> DepsTuple<T> for (TaskHandle<T>,) {
}
// Support for &TaskBuilder
impl<'a, Tk: Task, Deps> DepsTuple<Tk::Output> for &TaskBuilder<'a, Tk, Deps> {
impl<'a, Tk: TaskExecute, Deps> DepsTuple<Tk::Output> for &TaskBuilder<'a, Tk, Deps> {
fn to_node_ids(&self) -> Vec<NodeId> {
vec![self.id]
}
}
impl<'a, Tk: Task, Deps> DepsTuple<Tk::Output> for (&TaskBuilder<'a, Tk, Deps>,) {
impl<'a, Tk: TaskExecute, Deps> DepsTuple<Tk::Output> for (&TaskBuilder<'a, Tk, Deps>,) {
fn to_node_ids(&self) -> Vec<NodeId> {
vec![self.0.id]
}
@@ -89,7 +89,7 @@ macro_rules! impl_deps_tuple {
}
// For &TaskBuilder tuples
impl<'a, $($Tk: Task, $D),+> DepsTuple<($($Tk::Output,)+)> for ($(&TaskBuilder<'a, $Tk, $D>,)+) {
impl<'a, $($Tk: TaskExecute, $D),+> DepsTuple<($($Tk::Output,)+)> for ($(&TaskBuilder<'a, $Tk, $D>,)+) {
#[allow(non_snake_case)]
fn to_node_ids(&self) -> Vec<NodeId> {
let ($($T,)+) = self;
src/deps/tests.rs
+3 -1
View File
@@ -1,8 +1,10 @@
//! Unit tests for deps module
// Allow tests to reference dagx:: paths (for proc macro generated code)
extern crate self as dagx;
use crate::deps::DepsTuple;
use crate::runner::DagRunner;
use crate::task::Task;
use crate::types::{NodeId, TaskHandle};
// Test task for dependency testing
src/extract.rs
-291
View File
@@ -1,291 +0,0 @@
//! Input extraction for task dependencies via oneshot channels.
//!
//! This module handles extracting typed inputs from task dependencies via oneshot channels.
//!
//! # How it works
//!
//! When a task executes, it needs to extract its inputs from type-erased oneshot receivers.
//! Outputs are wrapped in Arc for efficient fanout (cheap Arc clones instead of data clones).
//! This module provides the ExtractInput trait with implementations for:
//! - `()` for tasks with no dependencies
//! - Common types (`i32`, `String`, etc.) for tasks with a single dependency
//! - `(A, B, ...)` tuples for tasks with multiple dependencies (awaited concurrently)
//!
//! # Why macros?
//!
//! Rust lacks variadic generics, so we use macros to generate implementations for:
//! - Common primitive/standard types (one impl per type)
//! - Tuples of different sizes (one impl per size, up to 8 elements)
//!
//! This is standard practice in Rust for working with tuples of different arities.
use std::any::Any;
use std::sync::Arc;
use futures::channel::oneshot;
/// Helper trait for async input extraction from oneshot channels.
///
/// Type erasure occurs only at the ExecutableNode trait boundary - by the time
/// we're in ExtractInput, we know the concrete type and can safely downcast.
pub(crate) trait ExtractInput: Sized + Clone {
fn extract_from_channels(
receivers: Vec<Box<dyn Any + Send>>,
) -> impl std::future::Future<Output = Result<Self, String>> + Send;
}
// Unit type - no dependencies
impl ExtractInput for () {
async fn extract_from_channels(_receivers: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
Ok(())
}
}
// Implementations for common primitive and standard library types.
// These allow tasks to receive these types as single inputs without wrapping in tuples.
// Receives Arc<T> from channel, clones the inner value for task consumption.
// For custom types, implement ExtractInput following the same pattern.
macro_rules! impl_extract_single {
($($t:ty),+) => {
$(
impl ExtractInput for $t {
async fn extract_from_channels(mut receivers: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if receivers.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
}
let rx = *receivers.pop()
.unwrap()
.downcast::<oneshot::Receiver<Arc<$t>>>()
.map_err(|_| format!("Type mismatch: expected Arc<{}>", std::any::type_name::<$t>()))?;
let arc_value = rx.await.map_err(|_| "Channel closed before receiving value".to_string())?;
Ok((*arc_value).clone())
}
}
)+
};
}
impl_extract_single!(
i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String
);
// Implement ExtractInput for Result types to support error handling patterns
impl<T, E> ExtractInput for Result<T, E>
where
T: 'static + Clone + Send + Sync,
E: 'static + Clone + Send + Sync,
{
async fn extract_from_channels(
mut receivers: Vec<Box<dyn Any + Send>>,
) -> Result<Self, String> {
if receivers.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
}
let rx = *receivers
.pop()
.unwrap()
.downcast::<oneshot::Receiver<Arc<Result<T, E>>>>()
.map_err(|_| {
format!(
"Type mismatch: expected Arc<Result<{}, {}>>",
std::any::type_name::<T>(),
std::any::type_name::<E>()
)
})?;
let arc_value = rx
.await
.map_err(|_| "Channel closed before receiving value".to_string())?;
Ok((*arc_value).clone())
}
}
// Implement ExtractInput for Option types
impl<T> ExtractInput for Option<T>
where
T: 'static + Clone + Send + Sync,
{
async fn extract_from_channels(
mut receivers: Vec<Box<dyn Any + Send>>,
) -> Result<Self, String> {
if receivers.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
}
let rx = *receivers
.pop()
.unwrap()
.downcast::<oneshot::Receiver<Arc<Option<T>>>>()
.map_err(|_| {
format!(
"Type mismatch: expected Arc<Option<{}>>",
std::any::type_name::<T>()
)
})?;
let arc_value = rx
.await
.map_err(|_| "Channel closed before receiving value".to_string())?;
Ok((*arc_value).clone())
}
}
// Implement ExtractInput for Vec types
impl<T> ExtractInput for Vec<T>
where
T: 'static + Clone + Send + Sync,
{
async fn extract_from_channels(
mut receivers: Vec<Box<dyn Any + Send>>,
) -> Result<Self, String> {
if receivers.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
}
let rx = *receivers
.pop()
.unwrap()
.downcast::<oneshot::Receiver<Arc<Vec<T>>>>()
.map_err(|_| {
format!(
"Type mismatch: expected Arc<Vec<{}>>",
std::any::type_name::<T>()
)
})?;
let arc_value = rx
.await
.map_err(|_| "Channel closed before receiving value".to_string())?;
Ok((*arc_value).clone())
}
}
// Implement ExtractInput for HashMap types
impl<K, V> ExtractInput for std::collections::HashMap<K, V>
where
K: 'static + Clone + Send + Sync + Eq + std::hash::Hash,
V: 'static + Clone + Send + Sync,
{
async fn extract_from_channels(
mut receivers: Vec<Box<dyn Any + Send>>,
) -> Result<Self, String> {
if receivers.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
}
let rx = *receivers
.pop()
.unwrap()
.downcast::<oneshot::Receiver<Arc<std::collections::HashMap<K, V>>>>()
.map_err(|_| {
format!(
"Type mismatch: expected Arc<HashMap<{}, {}>>",
std::any::type_name::<K>(),
std::any::type_name::<V>()
)
})?;
let arc_value = rx
.await
.map_err(|_| "Channel closed before receiving value".to_string())?;
Ok((*arc_value).clone())
}
}
// Implement ExtractInput for Arc types - SPECIAL CASE: just clone the Arc, don't unwrap!
// This allows zero-copy data sharing for users who wrap their data in Arc.
impl<T> ExtractInput for Arc<T>
where
T: 'static + Clone + Send + Sync,
{
async fn extract_from_channels(
mut receivers: Vec<Box<dyn Any + Send>>,
) -> Result<Self, String> {
if receivers.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
}
let rx = *receivers
.pop()
.unwrap()
.downcast::<oneshot::Receiver<Arc<Arc<T>>>>()
.map_err(|_| {
format!(
"Type mismatch: expected Arc<Arc<{}>>",
std::any::type_name::<T>()
)
})?;
let arc_arc_value = rx
.await
.map_err(|_| "Channel closed before receiving value".to_string())?;
// Return the inner Arc (clone is cheap - just refcount increment)
Ok(Arc::clone(&*arc_arc_value))
}
}
/// Macro to implement ExtractInput for tuple types.
///
/// Generates an implementation that extracts values from multiple dependency channels
/// and combines them into a tuple. All channels are awaited concurrently using join_all.
///
/// We need separate implementations for each tuple size (2-8) because Rust doesn't support
/// variadic generics. This is standard practice when working with tuples in Rust.
///
/// # Compile-time Safety
///
/// The number of dependencies is guaranteed to match the tuple size at compile time through:
/// 1. The `DepsTuple<Tk::Input>` trait bound on `depends_on()` ensures type/arity matching
/// 2. The `IsUnitType` constraint on `From<TaskBuilder>` prevents using tasks without calling `depends_on()`
///
/// Therefore, no runtime validation is needed - if it compiles, the dependency count is correct.
macro_rules! impl_extract_tuple {
($($T:ident : $idx:tt),+) => {
impl<$($T: 'static + Clone + Send + Sync),+> ExtractInput for ($($T,)+) {
async fn extract_from_channels(receivers: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
// Verify we have the correct number of receivers
let expected_count = [$($idx),+].len();
if receivers.len() != expected_count {
return Err(format!("Expected {} dependencies, got {}", expected_count, receivers.len()));
}
// Convert Vec to iterator so we can consume it
let mut iter = receivers.into_iter();
// Extract each receiver by consuming from the iterator
// Receivers now contain Arc-wrapped values for efficient fanout
// Dereference Box after downcasting to get the receiver itself
// Use type names as variable names to avoid numeric suffix issues
$(
#[allow(non_snake_case)]
let $T = *iter.next()
.ok_or_else(|| format!("Missing receiver at index {}", $idx))?
.downcast::<oneshot::Receiver<Arc<$T>>>()
.map_err(|_| format!("Type mismatch at index {}: expected Arc<{}>",
$idx, std::any::type_name::<$T>()))?;
)+
// Await all channels concurrently - this is a key optimization!
// We can start receiving as soon as ANY dependency completes, not waiting
// for them sequentially.
let arc_results = futures::join!(
$(
async move {
$T.await.map_err(|_| format!("Channel {} closed", $idx))
}
),+
);
// Clone inner values from Arc (cheap for small types, necessary for ownership)
Ok(($(
(*arc_results.$idx?).clone()
,)+))
}
}
};
}
// Generate ExtractInput implementations for tuples of size 2-8.
// Each invocation creates an impl for a specific tuple size.
// Supporting up to 8 elements covers the vast majority of use cases.
impl_extract_tuple!(A:0, B:1);
impl_extract_tuple!(A:0, B:1, C:2);
impl_extract_tuple!(A:0, B:1, C:2, D:3);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6, H:7);
#[cfg(test)]
mod tests;
src/extract/tests.rs
-410
View File
@@ -1,410 +0,0 @@
//! Unit tests for extract module
#![allow(dead_code)] // Test helpers generate unused code via macro
use crate::extract::ExtractInput;
use crate::task::Task;
use futures::channel::oneshot;
use std::any::Any;
use std::sync::Arc;
// Test task for error scenarios
#[allow(dead_code)]
struct TestTask {
value: i32,
}
#[crate::task]
#[allow(dead_code)]
impl TestTask {
#[cfg(not(tarpaulin_include))]
async fn run(&self) -> i32 {
self.value
}
}
#[allow(dead_code)]
struct TestTaskWithInput;
#[crate::task]
#[allow(dead_code)]
impl TestTaskWithInput {
#[cfg(not(tarpaulin_include))]
async fn run(input: &i32) -> i32 {
input * 2
}
}
#[tokio::test]
async fn test_extract_result_type_single_dependency_error() {
// Test Result<T,E> with wrong dependency count
let deps = vec![
Box::new(oneshot::channel::<Result<i32, String>>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<Result<i32, String>>().1) as Box<dyn Any + Send>,
]; // Two dependencies when expecting 1
let result = Result::<i32, String>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
}
#[tokio::test]
async fn test_extract_result_type_success() {
// Test Result<T,E> successful extraction
let (tx, rx) = oneshot::channel::<Arc<Result<i32, String>>>();
// Send value in background
tokio::spawn(async move {
let _ = tx.send(Arc::new(Ok(42)));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Result::<i32, String>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), Ok(42));
}
#[tokio::test]
async fn test_extract_option_type_single_dependency_error() {
// Test Option<T> with wrong dependency count
let deps = vec![
Box::new(oneshot::channel::<Option<String>>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<Option<String>>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<Option<String>>().1) as Box<dyn Any + Send>,
]; // Three dependencies when expecting 1
let result = Option::<String>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 3"));
}
#[tokio::test]
async fn test_extract_option_type_success() {
// Test Option<T> successful extraction
let (tx, rx) = oneshot::channel::<Arc<Option<i32>>>();
// Send value in background
tokio::spawn(async move {
let _ = tx.send(Arc::new(Some(99)));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Option::<i32>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), Some(99));
}
#[tokio::test]
async fn test_extract_vec_type_single_dependency_error() {
// Test Vec<T> with wrong dependency count
let deps = vec![]; // Zero dependencies when expecting 1
let result = Vec::<u8>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 0"));
}
#[tokio::test]
async fn test_extract_vec_type_success() {
// Test Vec<T> successful extraction
let (tx, rx) = oneshot::channel::<Arc<Vec<i32>>>();
// Send value in background
tokio::spawn(async move {
let _ = tx.send(Arc::new(vec![1, 2, 3]));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Vec::<i32>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), vec![1, 2, 3]);
}
#[tokio::test]
async fn test_extract_primitive_single_dependency_error() {
// Test primitive types with wrong dependency count
let deps = vec![
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
]; // Two dependencies when expecting 1
let result = i32::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
// Test with String
let deps = vec![
Box::new(oneshot::channel::<String>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<String>().1) as Box<dyn Any + Send>,
];
let result = String::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
// Test with f64
let deps = vec![
Box::new(oneshot::channel::<f64>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<f64>().1) as Box<dyn Any + Send>,
];
let result = f64::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
}
#[tokio::test]
async fn test_extract_unit_type() {
// Test that unit type always succeeds regardless of dependencies
let deps = vec![];
let result = <()>::extract_from_channels(deps).await;
assert!(result.is_ok());
// Even with dependencies (ignored), unit type succeeds
let deps = vec![
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
];
let result = <()>::extract_from_channels(deps).await;
assert!(result.is_ok());
}
#[tokio::test]
async fn test_extract_tuple_basic() {
// Test tuple extraction with channels
let (tx1, rx1) = oneshot::channel::<Arc<i32>>();
let (tx2, rx2) = oneshot::channel::<Arc<i32>>();
// Send values in background
tokio::spawn(async move {
let _ = tx1.send(Arc::new(42));
let _ = tx2.send(Arc::new(100));
});
let deps = vec![
Box::new(rx1) as Box<dyn Any + Send>,
Box::new(rx2) as Box<dyn Any + Send>,
];
// Extract a tuple (i32, i32)
let result = <(i32, i32)>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), (42, 100));
}
#[tokio::test]
async fn test_extract_channel_closed() {
// Test error when channel is closed
let (tx, rx) = oneshot::channel::<Arc<i32>>();
// Explicitly drop tx to close the channel
drop(tx);
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = i32::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Channel closed"));
}
#[tokio::test]
async fn test_extract_type_mismatch() {
// Test type mismatch during downcast
let (tx, rx) = oneshot::channel::<String>();
// Send a String
tokio::spawn(async move {
let _ = tx.send("hello".to_string());
});
// Try to extract as wrong type - put wrong receiver type in deps
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = i32::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
#[tokio::test]
async fn test_extract_all_primitive_types() {
// Test that all primitive types in impl_extract_single! work correctly
macro_rules! test_type {
($t:ty, $val:expr) => {
let (tx, rx) = oneshot::channel::<Arc<$t>>();
tokio::spawn(async move {
let _ = tx.send(Arc::new($val));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = <$t>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), $val);
};
}
test_type!(i8, 42i8);
test_type!(i16, 42i16);
test_type!(i32, 42i32);
test_type!(i64, 42i64);
test_type!(String, "test".to_string());
}
#[tokio::test]
async fn test_extract_larger_tuples() {
// Test extraction for larger tuple sizes
let (tx1, rx1) = oneshot::channel::<Arc<i32>>();
let (tx2, rx2) = oneshot::channel::<Arc<i32>>();
let (tx3, rx3) = oneshot::channel::<Arc<i32>>();
let (tx4, rx4) = oneshot::channel::<Arc<i32>>();
let (tx5, rx5) = oneshot::channel::<Arc<i32>>();
// Send values in background
tokio::spawn(async move {
let _ = tx1.send(Arc::new(0));
let _ = tx2.send(Arc::new(1));
let _ = tx3.send(Arc::new(2));
let _ = tx4.send(Arc::new(3));
let _ = tx5.send(Arc::new(4));
});
let deps = vec![
Box::new(rx1) as Box<dyn Any + Send>,
Box::new(rx2) as Box<dyn Any + Send>,
Box::new(rx3) as Box<dyn Any + Send>,
Box::new(rx4) as Box<dyn Any + Send>,
Box::new(rx5) as Box<dyn Any + Send>,
];
// Extract a 5-element tuple
let result = <(i32, i32, i32, i32, i32)>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(result.unwrap(), (0, 1, 2, 3, 4));
}
#[tokio::test]
async fn test_extract_arc_type() {
// Test Arc<T> extraction
let (tx, rx) = oneshot::channel::<Arc<Arc<String>>>();
let test_arc = Arc::new("test".to_string());
let test_arc_clone = test_arc.clone();
tokio::spawn(async move {
let _ = tx.send(Arc::new(test_arc));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Arc::<String>::extract_from_channels(deps).await;
assert!(result.is_ok());
assert_eq!(*result.unwrap(), *test_arc_clone);
}
// Error path tests for ExtractInput implementations
use std::collections::HashMap;
#[tokio::test]
async fn test_hashmap_extract_wrong_dependency_count() {
let deps = vec![
Box::new(oneshot::channel::<Arc<HashMap<String, i32>>>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<Arc<HashMap<String, i32>>>().1) as Box<dyn Any + Send>,
];
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
}
#[tokio::test]
async fn test_hashmap_extract_type_mismatch() {
let (tx, rx) = oneshot::channel::<Arc<String>>();
tokio::spawn(async move {
let _ = tx.send(Arc::new("wrong type".to_string()));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
#[tokio::test]
async fn test_hashmap_extract_channel_closed() {
let (tx, rx) = oneshot::channel::<Arc<HashMap<String, i32>>>();
drop(tx);
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Channel closed"));
}
#[tokio::test]
async fn test_result_extract_downcast_failure() {
let (tx, rx) = oneshot::channel::<Arc<String>>();
tokio::spawn(async move {
let _ = tx.send(Arc::new("wrong".to_string()));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Result::<i32, String>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
#[tokio::test]
async fn test_option_extract_downcast_failure() {
let (tx, rx) = oneshot::channel::<Arc<String>>();
tokio::spawn(async move {
let _ = tx.send(Arc::new("wrong".to_string()));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Option::<i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
#[tokio::test]
async fn test_vec_extract_downcast_failure() {
let (tx, rx) = oneshot::channel::<Arc<String>>();
tokio::spawn(async move {
let _ = tx.send(Arc::new("wrong".to_string()));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Vec::<i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
#[tokio::test]
async fn test_arc_extract_wrong_dependency_count() {
let deps = vec![];
let result = Arc::<i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Expected 1 dependency, got 0"));
}
#[tokio::test]
async fn test_arc_extract_downcast_failure() {
let (tx, rx) = oneshot::channel::<Arc<String>>();
tokio::spawn(async move {
let _ = tx.send(Arc::new("wrong".to_string()));
});
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Arc::<i32>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
#[tokio::test]
async fn test_tuple_wrong_dependency_count() {
let deps = vec![
Box::new(oneshot::channel::<Arc<i32>>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<Arc<i32>>().1) as Box<dyn Any + Send>,
Box::new(oneshot::channel::<Arc<i32>>().1) as Box<dyn Any + Send>,
];
let result = <(i32, i32)>::extract_from_channels(deps).await;
assert!(result.is_err());
assert!(result
.unwrap_err()
.contains("Expected 2 dependencies, got 3"));
}
src/lib.rs
+2 -1
View File
@@ -113,6 +113,7 @@
//! use dagx::{task, Task};
//!
//! // Define a struct to hold your multiple values
//! #[derive(Clone)]
//! struct UserData {
//! name: String,
//! age: i32,
@@ -722,7 +723,6 @@ mod builder;
pub mod cycle_prevention;
mod deps;
mod error;
mod extract;
mod node;
mod runner;
mod task;
@@ -731,6 +731,7 @@ mod types;
// Public re-exports
pub use builder::TaskBuilder;
pub use error::{DagError, DagResult};
pub use node::TaskExecute; // For macro-generated trait implementations
pub use runner::DagRunner;
pub use task::Task;
pub use types::{Pending, TaskHandle};
src/node.rs
+83 -75
View File
@@ -11,138 +11,146 @@ use std::future::Future;
use std::pin::Pin;
use std::sync::Arc;
use futures::channel::oneshot;
use crate::error::DagError;
use crate::extract::ExtractInput;
use crate::task::Task;
use crate::types::NodeId;
/// Type-erased channel endpoints (senders or receivers)
type ErasedChannels = Vec<Box<dyn Any + Send>>;
/// Type-erased channel pair (senders, receivers)
type ChannelPair = (ErasedChannels, ErasedChannels);
/// Type-erased listeners (for receiving inputs from dependencies)
type ErasedListeners = Vec<Box<dyn Any + Send>>;
/// Type-erased async execution result (Arc-wrapped for efficient fanout)
type ExecuteFuture =
Pin<Box<dyn Future<Output = Result<Arc<dyn Any + Send + Sync>, DagError>> + Send>>;
/// Type-erased async execution result for TaskExecute trait (returns String errors)
type TaskExecuteFuture =
Pin<Box<dyn Future<Output = Result<Arc<dyn Any + Send + Sync>, String>> + Send>>;
/// Internal trait for executing heterogeneous tasks.
///
/// This trait provides type erasure necessary for storing different task types in a single
/// collection. While the public API (Task, Handle, Node) is fully type-safe, internally we
/// need dynamic dispatch to execute tasks with different input/output types.
///
/// This is the ONLY place where type erasure occurs - channels are created with full type
/// This is the ONLY place where type erasure occurs - announcements are created with full type
/// information and only erased at this trait boundary.
///
/// This is an implementation detail and not part of the public API.
pub(crate) trait ExecutableNode: Send {
/// Create type-erased output channels for this node's dependents.
/// Create a type-erased announcement for this node's output.
///
/// Returns (senders, receivers) where:
/// - senders: Type-erased senders that this node will use to send outputs
/// - receivers: Type-erased receivers that dependents will use to receive inputs
fn create_output_channels(&self, num_dependents: usize) -> ChannelPair;
/// Returns a tuple of (Announcer, Announcement) both type-erased.
/// The Announcer will be used by this node to broadcast its output.
/// The Announcement can be used to create listeners for dependents.
fn create_announcement(&self) -> (Box<dyn Any + Send>, Box<dyn Any + Send>);
/// Execute the task with type-erased input receivers and output senders.
/// Create a type-erased listener from an announcement.
///
/// This allows dependents to listen to this node's output.
fn create_listener(&self, announcement: &dyn Any) -> Box<dyn Any + Send>;
/// Execute the task with type-erased input listeners and an output announcer.
///
/// The TypedNode implementation knows the concrete types and can safely downcast.
/// Consumes the node and returns the output value.
fn execute_with_channels(
fn execute_with_listeners(
self: Box<Self>,
receivers: ErasedChannels,
senders: ErasedChannels,
listeners: ErasedListeners,
announcer: Box<dyn Any + Send>,
) -> ExecuteFuture;
}
/// Fully-typed node storage for a single task.
///
/// Stores a task of type T. Implements ExecutableNode for type erasure,
/// allowing heterogeneous tasks to be stored in a single Vec.
/// Stores a task that has an `execute()` method (generated by the #[task] macro).
/// Implements ExecutableNode for type erasure, allowing heterogeneous tasks to be
/// stored in a single Vec.
///
/// # Ownership Model
///
/// The task is owned directly (no Mutex needed) and consumed during execution.
/// Each task executes exactly once, taking ownership and producing an output.
///
/// # Oneshot Channels
/// # Announcement Broadcasting
///
/// Outputs are communicated via oneshot channels for inter-task communication.
/// The output is also returned from execute_with_channels for central storage.
pub(crate) struct TypedNode<T: Task> {
/// Outputs are communicated via announcement channels for inter-task communication.
/// Each task has ONE announcer that broadcasts to multiple listeners.
/// The output is also returned from execute_with_listeners for central storage.
pub(crate) struct TypedNode<T> {
pub(crate) id: NodeId,
pub(crate) task: T,
pub(crate) _output_type: std::marker::PhantomData<fn() -> ()>,
}
impl<T: Task + 'static> TypedNode<T> {
impl<T> TypedNode<T> {
pub(crate) fn new(id: NodeId, task: T) -> Self {
Self { id, task }
Self {
id,
task,
_output_type: std::marker::PhantomData,
}
}
}
impl<T: Task + 'static> ExecutableNode for TypedNode<T>
// Implementation for tasks created with the #[task] macro
// These tasks have an execute() method generated by the macro
impl<T: 'static + Send> ExecutableNode for TypedNode<T>
where
T::Input: 'static + Clone + ExtractInput,
T::Output: 'static + Clone,
T: TaskExecute,
{
fn create_output_channels(&self, num_dependents: usize) -> ChannelPair {
let mut senders = Vec::new();
let mut receivers = Vec::new();
for _ in 0..num_dependents {
// Create channels that pass Arc-wrapped outputs (efficient for fanout)
let (tx, rx) = oneshot::channel::<Arc<T::Output>>();
senders.push(Box::new(tx) as Box<dyn Any + Send>);
receivers.push(Box::new(rx) as Box<dyn Any + Send>);
}
(senders, receivers)
fn create_announcement(&self) -> (Box<dyn Any + Send>, Box<dyn Any + Send>) {
T::create_announcement_erased()
}
fn execute_with_channels(
fn create_listener(&self, announcement: &dyn Any) -> Box<dyn Any + Send> {
T::create_listener_erased(announcement)
}
fn execute_with_listeners(
self: Box<Self>,
receivers: ErasedChannels,
senders: ErasedChannels,
listeners: ErasedListeners,
announcer: Box<dyn Any + Send>,
) -> ExecuteFuture {
Box::pin(async move {
// Extract typed input from type-erased receivers
// TypedNode<T> knows T::Input, so it can safely downcast
let input = match T::Input::extract_from_channels(receivers).await {
Ok(input) => input,
Err(_msg) => {
// Failed to extract input - likely channel closed or type mismatch
return Err(DagError::InvalidDependency { task_id: self.id.0 });
}
};
// Call the task's execute() method (generated by the #[task] macro)
// This handles listener extraction, calling run(), and announcing
let result = self.task.execute(listeners, announcer).await;
// Execute task (consumes self.task)
let output = self.task.run(input).await;
// Arc Wrapping Strategy:
// Wrap output in Arc ONCE for efficient fan-out sharing.
// For 1:N dependencies, this enables O(1) sharing (Arc clone = atomic increment)
// instead of O(N) cloning (N deep copies of output data).
// Downstream tasks receive &T after ExtractInput extracts from Arc.
let arc_output = Arc::new(output);
// Send Arc to all dependents via their channels
// TypedNode<T> knows T::Output, so downcast is safe
for sender_any in senders {
if let Ok(sender) = sender_any.downcast::<oneshot::Sender<Arc<T::Output>>>() {
// Clone Arc (just atomic refcount increment, NOT data clone)
// This is the key performance optimization for fan-out patterns
// Ignore send errors - dependent might have been cancelled
let _ = sender.send(Arc::clone(&arc_output));
match result {
Ok(output_arc) => Ok(output_arc),
Err(err_msg) => {
// Execution failed - convert error message to DagError
eprintln!("Task execution failed: {}", err_msg);
Err(DagError::InvalidDependency { task_id: self.id.0 })
}
}
// Return Arc-wrapped output (for storage by runner)
Ok(arc_output as Arc<dyn Any + Send + Sync>)
})
}
}
/// Trait for tasks that have an execute() method (generated by #[task] macro).
///
/// This trait is automatically implemented by the macro and should not be
/// implemented manually.
pub trait TaskExecute: Send {
/// Input type as a tuple of parameter types (for dependency checking)
/// - `()` for no parameters
/// - `T` for single parameter
/// - `(A, B, C, ...)` for multiple parameters
type Input: Send;
/// Output type of the task
type Output: Send + Sync + Clone;
fn execute(
self,
listeners: Vec<Box<dyn Any + Send>>,
announcer: Box<dyn Any + Send>,
) -> TaskExecuteFuture;
fn create_announcement_erased() -> (Box<dyn Any + Send>, Box<dyn Any + Send>);
fn create_listener_erased(announcement: &dyn Any) -> Box<dyn Any + Send>;
}
#[cfg(test)]
mod tests;
src/node/tests.rs
+93 -64
View File
@@ -1,7 +1,9 @@
//! Unit tests for node module
// Allow tests to reference dagx:: paths (for proc macro generated code)
extern crate self as dagx;
use crate::node::{ExecutableNode, TypedNode};
use crate::task::Task;
use crate::types::NodeId;
// Simple test task
@@ -17,25 +19,33 @@ impl TestTask {
}
#[test]
fn test_create_output_channels() {
fn test_create_announcement() {
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
// Create channels for 3 dependents
let (senders, receivers) = node.create_output_channels(3);
// Create ONE announcement that can broadcast to multiple listeners
let (announcer, announcement) = node.create_announcement();
assert_eq!(senders.len(), 3);
assert_eq!(receivers.len(), 3);
// Verify both are created (type-erased)
assert!(std::any::Any::type_id(&*announcer) != std::any::TypeId::of::<()>());
assert!(std::any::Any::type_id(&*announcement) != std::any::TypeId::of::<()>());
}
#[test]
fn test_create_output_channels_zero_dependents() {
fn test_create_multiple_listeners() {
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
// Create channels for 0 dependents (sink node)
let (senders, receivers) = node.create_output_channels(0);
// Create announcement
let (_announcer, announcement) = node.create_announcement();
assert_eq!(senders.len(), 0);
assert_eq!(receivers.len(), 0);
// Create multiple listeners from the same announcement (broadcast pattern)
let listener1 = node.create_listener(announcement.as_ref());
let listener2 = node.create_listener(announcement.as_ref());
let listener3 = node.create_listener(announcement.as_ref());
// All listeners are created from the same announcement
assert!(std::any::Any::type_id(&*listener1) != std::any::TypeId::of::<()>());
assert!(std::any::Any::type_id(&*listener2) != std::any::TypeId::of::<()>());
assert!(std::any::Any::type_id(&*listener3) != std::any::TypeId::of::<()>());
}
struct TaskWithDependency;
@@ -48,15 +58,19 @@ impl TaskWithDependency {
}
#[tokio::test]
async fn test_execute_with_channels_missing_input() {
async fn test_execute_with_listeners_missing_input() {
// Test when task expects input but doesn't receive it
let node = Box::new(TypedNode::new(NodeId(0), TaskWithDependency));
let node = TypedNode::new(NodeId(0), TaskWithDependency);
// TaskWithDependency expects an i32 input, but we provide no receivers
let receivers = vec![];
let senders = vec![];
// Create a real announcer even though we won't use it
let (announcer, _announcement) = node.create_announcement();
let result = node.execute_with_channels(receivers, senders).await;
let node = Box::new(node);
// TaskWithDependency expects an i32 input, but we provide no listeners
let listeners = vec![];
let result = node.execute_with_listeners(listeners, announcer).await;
assert!(result.is_err());
if let Err(crate::error::DagError::InvalidDependency { task_id }) = result {
@@ -67,49 +81,62 @@ async fn test_execute_with_channels_missing_input() {
}
#[tokio::test]
async fn test_execute_with_channels_success() {
// Test successful execution with channels
let dependency_node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 21 }));
async fn test_execute_with_listeners_success() {
// Test successful execution with announcements and listeners
let dependency_node = TypedNode::new(NodeId(0), TestTask { value: 21 });
let dependent_node = Box::new(TypedNode::new(NodeId(1), TaskWithDependency));
// Create a channel from dependency to dependent (Arc-wrapped)
let (tx, rx) = futures::channel::oneshot::channel::<std::sync::Arc<i32>>();
// Create announcement from dependency
let (dep_announcer, dep_announcement) = dependency_node.create_announcement();
// Execute dependency node with sender
let dep_senders = vec![Box::new(tx) as Box<dyn std::any::Any + Send>];
let dep_receivers = vec![];
// Create listener for dependent from dependency's announcement
let listener = dependency_node.create_listener(dep_announcement.as_ref());
// Execute dependency node with announcer
let dependency_node = Box::new(dependency_node);
let dep_listeners = vec![];
// Execute in background
let dep_handle = tokio::spawn(async move {
dependency_node
.execute_with_channels(dep_receivers, dep_senders)
.execute_with_listeners(dep_listeners, dep_announcer)
.await
});
// Execute dependent node with receiver
let receivers = vec![Box::new(rx) as Box<dyn std::any::Any + Send>];
let senders = vec![];
// Execute dependent node with listener
// Create a real announcer for the dependent (even though no one listens to it)
let dep_node_for_announcer = TypedNode::new(NodeId(1), TaskWithDependency);
let (dependent_announcer, _dep_announcement) = dep_node_for_announcer.create_announcement();
let listeners = vec![listener];
let result = dependent_node
.execute_with_channels(receivers, senders)
.execute_with_listeners(listeners, dependent_announcer)
.await;
// Wait for dependency to complete
let _ = dep_handle.await;
assert!(result.is_ok());
let output = result.unwrap();
let arc_value = output.downcast::<i32>().unwrap();
assert_eq!(*arc_value, 42); // 21 * 2 = 42
}
#[tokio::test]
async fn test_execute_with_channels_sink_node() {
async fn test_execute_with_listeners_sink_node() {
// Test that sink nodes (no dependents) return their output
let node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 42 }));
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
// No senders means this is a sink node
let receivers = vec![];
let senders = vec![];
// Create a real announcer (even though no one will listen)
let (announcer, _announcement) = node.create_announcement();
let result = node.execute_with_channels(receivers, senders).await;
let node = Box::new(node);
// Sink node has no listeners
let listeners = vec![];
let result = node.execute_with_listeners(listeners, announcer).await;
assert!(result.is_ok());
// Check output was returned (Arc-wrapped)
@@ -119,29 +146,28 @@ async fn test_execute_with_channels_sink_node() {
}
#[tokio::test]
async fn test_execute_with_channels_non_sink_node() {
// Test that non-sink nodes send output via channels
async fn test_execute_with_listeners_non_sink_node() {
// Test that non-sink nodes broadcast output via announcement
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
// Create output channels for one dependent
let (senders, mut receivers) = node.create_output_channels(1);
// Create announcement and listener
let (announcer, announcement) = node.create_announcement();
let listener = node.create_listener(announcement.as_ref());
// Execute with senders (non-sink node)
// Execute with announcer (non-sink node)
let node = Box::new(node);
let exec_receivers = vec![];
let result = node.execute_with_channels(exec_receivers, senders).await;
let listeners = vec![];
let result = node.execute_with_listeners(listeners, announcer).await;
assert!(result.is_ok());
// Verify output was sent through channel (Arc-wrapped)
let rx = receivers
.pop()
.unwrap()
.downcast::<futures::channel::oneshot::Receiver<std::sync::Arc<i32>>>()
// Verify output was broadcast through announcement
let listener = *listener
.downcast::<announcement::Listener<i32>>()
.unwrap();
let received_arc = rx.await;
assert!(received_arc.is_ok());
assert_eq!(*received_arc.unwrap(), 42);
let received = listener.listen().await;
assert!(received.is_ok());
assert_eq!(received.unwrap(), 42);
// Also verify output was returned (Arc-wrapped)
let output = result.unwrap();
@@ -150,27 +176,30 @@ async fn test_execute_with_channels_non_sink_node() {
}
#[tokio::test]
async fn test_execute_with_multiple_dependents() {
// Test that output is sent to multiple dependents
async fn test_execute_with_multiple_listeners() {
// Test that output is broadcast to multiple listeners
let node = TypedNode::new(NodeId(0), TestTask { value: 100 });
// Create output channels for three dependents
let (senders, receivers) = node.create_output_channels(3);
// Create announcement and multiple listeners
let (announcer, announcement) = node.create_announcement();
let listener1 = node.create_listener(announcement.as_ref());
let listener2 = node.create_listener(announcement.as_ref());
let listener3 = node.create_listener(announcement.as_ref());
// Execute node
let node = Box::new(node);
let exec_receivers = vec![];
let result = node.execute_with_channels(exec_receivers, senders).await;
let listeners = vec![];
let result = node.execute_with_listeners(listeners, announcer).await;
assert!(result.is_ok());
// Verify all three dependents receive the output (Arc-wrapped)
for receiver in receivers {
let rx = receiver
.downcast::<futures::channel::oneshot::Receiver<std::sync::Arc<i32>>>()
// Verify all three listeners receive the output
for listener in [listener1, listener2, listener3] {
let listener = *listener
.downcast::<announcement::Listener<i32>>()
.unwrap();
let received_arc = rx.await;
assert!(received_arc.is_ok());
assert_eq!(*received_arc.unwrap(), 100);
let received = listener.listen().await;
assert!(received.is_ok());
assert_eq!(received.unwrap(), 100);
}
}
src/runner.rs
+44 -40
View File
@@ -22,9 +22,7 @@ use tracing::{debug, error, info, trace};
use crate::builder::TaskBuilder;
use crate::error::{DagError, DagResult};
use crate::extract::ExtractInput;
use crate::node::{ExecutableNode, TypedNode};
use crate::task::Task;
use crate::types::{NodeId, Pending, TaskHandle};
// Guard to ensure run_lock is released even on early return or panic
@@ -195,9 +193,7 @@ impl DagRunner {
/// ```
pub fn add_task<Tk>(&self, task: Tk) -> TaskBuilder<'_, Tk, Pending>
where
Tk: Task + 'static,
Tk::Input: 'static + Clone + ExtractInput,
Tk::Output: 'static + Clone,
Tk: crate::node::TaskExecute + 'static,
{
let id = self.alloc_id();
@@ -321,39 +317,41 @@ impl DagRunner {
let edges = self.edges.lock().clone();
// Create all channels upfront
// Map: (producer_id, consumer_id) -> receiver index
// We store receivers in a Vec per consumer to maintain order
let mut consumer_receivers: HashMap<NodeId, Vec<Box<dyn std::any::Any + Send>>> =
HashMap::new();
let mut producer_senders: HashMap<NodeId, Vec<Box<dyn std::any::Any + Send>>> =
HashMap::new();
// Create ONE announcement per task
// Each task broadcasts its output to all dependents via announcement
let mut announcements: HashMap<NodeId, Box<dyn std::any::Any + Send>> = HashMap::new();
let mut announcers: HashMap<NodeId, Box<dyn std::any::Any + Send>> = HashMap::new();
// Create channels for each producer-consumer relationship
// We need to lock nodes temporarily to call create_output_channels
// Create announcements for all tasks
{
let nodes_lock = self.nodes.lock();
for (node_id, node_opt) in nodes_lock.iter().enumerate() {
if let Some(node) = node_opt {
let (announcer, announcement) = node.create_announcement();
announcers.insert(NodeId(node_id), announcer);
announcements.insert(NodeId(node_id), announcement);
}
}
}
// Create listeners for each consumer from their dependencies
let mut consumer_listeners: HashMap<NodeId, Vec<Box<dyn std::any::Any + Send>>> =
HashMap::new();
{
let nodes_lock = self.nodes.lock();
for (consumer_id, producer_ids) in &edges {
let mut listeners = Vec::new();
for &producer_id in producer_ids {
// Get the producer node (it's still in the vector at this point)
if let Some(Some(producer_node)) = nodes_lock.get(producer_id.0) {
// Ask the producer to create ONE channel for this consumer
let (mut senders, mut receivers) = producer_node.create_output_channels(1);
// Store the sender for the producer
producer_senders
.entry(producer_id)
.or_default()
.push(senders.pop().unwrap());
// Store the receiver for the consumer (order matters!)
consumer_receivers
.entry(*consumer_id)
.or_default()
.push(receivers.pop().unwrap());
if let Some(announcement) = announcements.get(&producer_id) {
if let Some(Some(producer_node)) = nodes_lock.get(producer_id.0) {
// Producer node creates listener from its own announcement
let listener = producer_node.create_listener(announcement.as_ref());
listeners.push(listener);
}
}
}
consumer_listeners.insert(*consumer_id, listeners);
}
}
@@ -412,11 +410,14 @@ impl DagRunner {
};
if let Some(node) = node {
// Take the receivers for this consumer
let receivers = consumer_receivers.remove(&node_id).unwrap_or_default();
// Take the listeners for this consumer
let listeners = consumer_listeners.remove(&node_id).unwrap_or_default();
// Take the senders for this producer
let senders = producer_senders.remove(&node_id).unwrap_or_default();
// Take the announcer for this producer
let announcer = announcers.remove(&node_id).unwrap_or_else(|| {
// Create dummy announcer for tasks with no dependents
Box::new(()) as Box<dyn std::any::Any + Send>
});
// Execute inline with panic handling.
//
@@ -424,7 +425,7 @@ impl DagRunner {
// DagError::TaskPanicked, matching the behavior of async runtimes when
// they spawn tasks. This guarantees consistent error handling whether
// a task executes inline (single-task layer) or spawned (multi-task layer).
let result = AssertUnwindSafe(node.execute_with_channels(receivers, senders))
let result = AssertUnwindSafe(node.execute_with_listeners(listeners, announcer))
.catch_unwind()
.await
.unwrap_or_else(|panic_payload| {
@@ -473,15 +474,18 @@ impl DagRunner {
};
if let Some(node) = node {
// Take the receivers for this consumer
let receivers = consumer_receivers.remove(&node_id).unwrap_or_default();
// Take the listeners for this consumer
let listeners = consumer_listeners.remove(&node_id).unwrap_or_default();
// Take the senders for this producer
let senders = producer_senders.remove(&node_id).unwrap_or_default();
// Take the announcer for this producer
let announcer = announcers.remove(&node_id).unwrap_or_else(|| {
// Create dummy announcer for tasks with no dependents
Box::new(()) as Box<dyn std::any::Any + Send>
});
// Create a 'static future by taking ownership
let inner_future = async move {
let task_future = node.execute_with_channels(receivers, senders);
let task_future = node.execute_with_listeners(listeners, announcer);
let result = task_future.await;
src/runner/tests.rs
+3 -1
View File
@@ -1,8 +1,10 @@
//! Unit tests for runner module
// Allow tests to reference dagx:: paths (for proc macro generated code)
extern crate self as dagx;
use crate::error::DagError;
use crate::runner::DagRunner;
use crate::task::Task;
use crate::types::TaskHandle;
// Initialize tracing subscriber for tests (idempotent)
src/task.rs
+244 -80
View File
@@ -3,94 +3,197 @@
//! This module defines the core Task trait that all DAG nodes must implement,
//! along with convenience functions for creating tasks from closures.
use std::any::Any;
use std::future::Future;
/// Helper trait for extracting inputs from listeners (internal use only)
trait ExtractFromListeners: Sized + Send + Sync + Clone + 'static {
fn extract(
listeners: Vec<Box<dyn Any + Send>>,
) -> impl Future<Output = Result<Self, String>> + Send;
}
// Implementation for unit type (no dependencies)
impl ExtractFromListeners for () {
async fn extract(_listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
Ok(())
}
}
// Implementation for single input types (not tuples)
macro_rules! impl_extract_single {
($($t:ty),+) => {
$(
impl ExtractFromListeners for $t {
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if listeners.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
}
let listener = *listeners.pop()
.unwrap()
.downcast::<announcement::Listener<$t>>()
.map_err(|_| format!("Type mismatch: expected Listener<{}>", std::any::type_name::<$t>()))?;
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
}
}
)+
};
}
impl_extract_single!(i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String);
// Implementation for Result types
impl<T, E> ExtractFromListeners for Result<T, E>
where
T: 'static + Clone + Send + Sync,
E: 'static + Clone + Send + Sync,
{
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if listeners.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
}
let listener = *listeners.pop()
.unwrap()
.downcast::<announcement::Listener<Result<T, E>>>()
.map_err(|_| format!("Type mismatch: expected Listener<Result<{}, {}>>", std::any::type_name::<T>(), std::any::type_name::<E>()))?;
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
}
}
// Implementation for Option types
impl<T> ExtractFromListeners for Option<T>
where
T: 'static + Clone + Send + Sync,
{
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if listeners.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
}
let listener = *listeners.pop()
.unwrap()
.downcast::<announcement::Listener<Option<T>>>()
.map_err(|_| format!("Type mismatch: expected Listener<Option<{}>>", std::any::type_name::<T>()))?;
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
}
}
// Implementation for Vec types
impl<T> ExtractFromListeners for Vec<T>
where
T: 'static + Clone + Send + Sync,
{
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if listeners.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
}
let listener = *listeners.pop()
.unwrap()
.downcast::<announcement::Listener<Vec<T>>>()
.map_err(|_| format!("Type mismatch: expected Listener<Vec<{}>>", std::any::type_name::<T>()))?;
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
}
}
// Implementation for Arc types
impl<T> ExtractFromListeners for std::sync::Arc<T>
where
T: 'static + Clone + Send + Sync,
{
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if listeners.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
}
let listener = *listeners.pop()
.unwrap()
.downcast::<announcement::Listener<std::sync::Arc<T>>>()
.map_err(|_| format!("Type mismatch: expected Listener<Arc<{}>>", std::any::type_name::<T>()))?;
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
}
}
// Implementation for HashMap types
impl<K, V> ExtractFromListeners for std::collections::HashMap<K, V>
where
K: 'static + Clone + Send + Sync + Eq + std::hash::Hash,
V: 'static + Clone + Send + Sync,
{
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
if listeners.len() != 1 {
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
}
let listener = *listeners.pop()
.unwrap()
.downcast::<announcement::Listener<std::collections::HashMap<K, V>>>()
.map_err(|_| format!("Type mismatch: expected Listener<HashMap<{}, {}>>", std::any::type_name::<K>(), std::any::type_name::<V>()))?;
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
}
}
// Macro to implement ExtractFromListeners for tuple types
macro_rules! impl_extract_tuple {
($($T:ident : $idx:tt),+) => {
#[allow(non_snake_case)]
impl<$($T),+> ExtractFromListeners for ($($T,)+)
where
$($T: 'static + Clone + Send + Sync),+
{
async fn extract(listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
let expected_count = [$($idx),+].len();
if listeners.len() != expected_count {
return Err(format!("Expected {} dependencies, got {}", expected_count, listeners.len()));
}
let mut iter = listeners.into_iter();
$(
let $T = *iter.next().unwrap()
.downcast::<announcement::Listener<$T>>()
.map_err(|_| format!("Type mismatch at index {}: expected Listener<{}>", $idx, std::any::type_name::<$T>()))?;
)+
let results = futures::join!(
$(
async move {
$T.listen().await.map_err(|_| format!("Listener {} closed", $idx))
}
),+
);
Ok(($(
results.$idx?
,)+))
}
}
};
}
impl_extract_tuple!(A:0, B:1);
impl_extract_tuple!(A:0, B:1, C:2);
impl_extract_tuple!(A:0, B:1, C:2, D:3);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6);
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6, H:7);
/// A unit of async work with typed inputs and outputs.
///
/// Use the [`task`](crate::task) macro to implement this trait. The macro automatically
/// derives `Input` and `Output` types from your `run()` method signature.
/// **NOTE**: For new code, use the [`task`](crate::task) macro which automatically generates
/// an `execute()` method and handles all the wiring. This trait is primarily for internal use
/// and backward compatibility with `task_fn`.
///
/// # Task Patterns
///
/// dagx supports three main task patterns:
///
/// ## 1. Stateless Tasks (Pure Functions)
///
/// Tasks with no internal state - just transform inputs to outputs:
///
/// ```
/// use dagx::{task, Task};
///
/// struct Add;
///
/// #[task]
/// impl Add {
/// async fn run(a: &i32, b: &i32) -> i32 {
/// a + b
/// }
/// }
/// ```
///
/// ## 2. Stateful Tasks with Shared State (`&self`)
///
/// Tasks that read from internal state but don't modify it:
///
/// ```
/// use dagx::{task, Task};
///
/// struct Multiplier {
/// factor: i32,
/// }
///
/// #[task]
/// impl Multiplier {
/// async fn run(&self, input: &i32) -> i32 {
/// input * self.factor
/// }
/// }
/// ```
///
/// ## 3. Stateful Tasks with Mutable State (`&mut self`)
///
/// Tasks that modify internal state during execution:
///
/// ```
/// use dagx::{task, Task};
///
/// struct Counter {
/// count: i32,
/// }
///
/// #[task]
/// impl Counter {
/// async fn run(&mut self, increment: &i32) -> i32 {
/// self.count += increment;
/// self.count
/// }
/// }
/// ```
///
/// # Input Patterns
///
/// - `()`: No dependencies (source task)
/// - `&T`: Single dependency
/// - `&A, &B, &C, ...`: Multiple dependencies (up to 8, order matters)
///
/// # The Underlying Trait
///
/// The `#[task]` macro generates an implementation of this trait:
/// The `#[task]` macro generates an `execute()` method on your struct that:
/// - Extracts inputs from announcement listeners
/// - Calls your `run()` method
/// - Announces the result to dependents
/// - Returns Arc-wrapped output
///
/// # Output Wrapping
///
/// **Task outputs are automatically wrapped in `Arc<T>` internally** for efficient fan-out.
/// You define `Output = T`, but the framework wraps it in `Arc<T>` before sending to channels.
/// You define output as `T`, but the framework wraps it in `Arc<T>` when announcing.
/// This enables O(1) sharing for 1:N dependencies - Arc is cloned (cheap), not your data.
///
/// Downstream tasks receive `&T` after ExtractInput extracts from the Arc. This is transparent
/// to your task code - you work with `T`, the framework handles Arc wrapping/unwrapping.
pub trait Task: Send {
type Input: Send;
type Output: Send + Sync; // Sync required for Arc-wrapping and cross-thread sharing
type Output: Send + Sync + Clone; // Sync and Clone required for Arc-wrapping and announcement
fn run(self, input: Self::Input) -> impl Future<Output = Self::Output> + Send;
}
@@ -100,12 +203,12 @@ pub trait Task: Send {
/// **INTERNAL USE ONLY**: This function is for testing purposes only and should
/// not be used in production code. Use the `#[task]` macro instead.
#[doc(hidden)]
pub fn task_fn<I, O, Fut, F>(f: F) -> impl Task<Input = I, Output = O>
pub fn task_fn<I, O, Fut, F>(f: F) -> impl Task<Input = I, Output = O> + crate::node::TaskExecute<Input = I, Output = O>
where
F: FnMut(I) -> Fut + Send + 'static,
Fut: Future<Output = O> + Send + 'static,
I: Send + 'static,
O: Send + Sync + 'static,
I: ExtractFromListeners,
O: Send + Sync + Clone + 'static,
{
struct TaskFn<I, O, Fut, F>
where
@@ -123,7 +226,7 @@ where
F: FnMut(I) -> Fut + Send,
Fut: Future<Output = O> + Send,
I: Send,
O: Send + Sync,
O: Send + Sync + Clone,
{
type Input = I;
type Output = O;
@@ -133,6 +236,67 @@ where
}
}
impl<I, O, Fut, F> crate::node::TaskExecute for TaskFn<I, O, Fut, F>
where
F: FnMut(I) -> Fut + Send + 'static,
Fut: Future<Output = O> + Send + 'static,
I: ExtractFromListeners,
O: Send + Sync + Clone + 'static,
{
type Input = I;
type Output = O;
fn execute(
mut self,
listeners: Vec<Box<dyn std::any::Any + Send>>,
announcer: Box<dyn std::any::Any + Send>,
) -> std::pin::Pin<
Box<
dyn Future<Output = Result<std::sync::Arc<dyn std::any::Any + Send + Sync>, String>>
+ Send,
>,
> {
Box::pin(async move {
use std::sync::Arc;
// Extract the input using the ExtractFromListeners trait
let input = I::extract(listeners).await?;
// Call the closure
let output: O = (self.f)(input).await;
// Wrap in Arc
let output_arc = Arc::new(output);
// Announce the result
let announcer_typed = announcer
.downcast::<announcement::Announcer<O>>()
.map_err(|_| {
"Type mismatch: announcer type does not match task output type".to_string()
})?;
let _ = announcer_typed.announce((*output_arc).clone());
// Return Arc-wrapped output
Ok(output_arc as Arc<dyn std::any::Any + Send + Sync>)
})
}
fn create_announcement_erased() -> (Box<dyn std::any::Any + Send>, Box<dyn std::any::Any + Send>) {
let (announcer, announcement) = announcement::Announcement::<O>::new();
(
Box::new(announcer) as Box<dyn std::any::Any + Send>,
Box::new(announcement) as Box<dyn std::any::Any + Send>,
)
}
fn create_listener_erased(announcement: &dyn std::any::Any) -> Box<dyn std::any::Any + Send> {
let announcement = announcement
.downcast_ref::<announcement::Announcement<O>>()
.expect("Type mismatch: announcement type does not match task output type");
Box::new(announcement.listener()) as Box<dyn std::any::Any + Send>
}
}
TaskFn {
f,
_phantom: std::marker::PhantomData,
src/types.rs
+3 -3
View File
@@ -4,7 +4,7 @@
//! including node identifiers, task handles, and type-state markers.
use crate::builder::{IsUnitType, TaskBuilder};
use crate::task::Task;
use crate::node::TaskExecute;
/// Opaque node identifier (internal use)
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
@@ -62,7 +62,7 @@ pub struct Pending;
// TaskBuilder can be converted to a TaskHandle ONLY if Input is () (unit type)
// This enforces at compile time that tasks with non-unit inputs MUST call .depends_on()
impl<'a, Tk: Task, Deps> From<TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
impl<'a, Tk: TaskExecute, Deps> From<TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
where
Tk::Input: IsUnitType,
{
@@ -74,7 +74,7 @@ where
}
}
impl<'a, Tk: Task, Deps> From<&TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
impl<'a, Tk: TaskExecute, Deps> From<&TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
where
Tk::Input: IsUnitType,
{
src/types/tests.rs
+3 -1
View File
@@ -1,6 +1,8 @@
//! Unit tests for types module
use crate::task::Task;
// Allow tests to reference dagx:: paths (for proc macro generated code)
extern crate self as dagx;
use crate::types::{NodeId, TaskHandle};
use std::marker::PhantomData;
tests/arc_tests.rs
+1 -1
View File
@@ -9,7 +9,7 @@
//!
//! These tests serve as both validation and documentation for proper Arc usage.
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
tests/boundaries/zero_tasks.rs
+1 -1
View File
@@ -1,7 +1,7 @@
//! Tests for empty DAGs and minimal task configurations
use crate::common::task_fn;
use dagx::{task, DagResult, DagRunner, Task};
use dagx::{task, DagResult, DagRunner};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
tests/common/mod.rs
+1 -1
View File
@@ -7,7 +7,7 @@ pub use dagx::task_fn;
// Re-export commonly used test structs
#[allow(dead_code)]
pub mod tasks {
use dagx::{task, Task};
use dagx::task;
// Simple seed task with mutable state
pub struct Seed {
tests/cycle_prevention.rs
+1 -1
View File
@@ -3,7 +3,7 @@
//! These tests show that even if you try to create cycles, the Rust compiler
//! prevents it through the type-state pattern.
use dagx::{task, DagRunner, Task, TaskHandle};
use dagx::{task, DagRunner, TaskHandle};
/// Demonstrates that TaskBuilder is consumed by depends_on,
/// preventing you from wiring the same task multiple times
tests/execution/integration.rs
+1 -1
View File
@@ -1,6 +1,6 @@
//! Integration tests for end-to-end functionality
use dagx::{DagResult, DagRunner, Task};
use dagx::{DagResult, DagRunner};
use crate::common::task_fn;
use std::time::Instant;
use tokio::time::{sleep, Duration};
tests/parallelism/timing.rs
+1 -1
View File
@@ -1,7 +1,7 @@
//! Tests for timing and speedup verification
use crate::common::task_fn;
use dagx::{DagResult, DagRunner, Task};
use dagx::{DagResult, DagRunner};
use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant};
tests/runtimes/async_std.rs
+1 -1
View File
@@ -1,6 +1,6 @@
//! Runtime compatibility tests for async-std
use dagx::{DagRunner, Task};
use dagx::DagRunner;
use crate::common::task_fn;
struct Value(i32);
tests/runtimes/compatibility.rs
+1 -1
View File
@@ -1,6 +1,6 @@
//! Runtime integration tests - verify dagx works correctly across different runtimes
use dagx::{DagResult, DagRunner, Task};
use dagx::{DagResult, DagRunner};
struct Value(i32);
#[dagx::task]
tests/runtimes/quirky.rs
+2 -2
View File
@@ -486,7 +486,7 @@ fn test_futures_executor_multi_pool() {
#[test]
fn test_sync_tasks_on_async_executor() {
use dagx::Task;
struct SyncAdd;
@@ -519,7 +519,7 @@ fn test_sync_tasks_on_async_executor() {
#[test]
fn test_sync_tasks_on_pollster() {
use dagx::Task;
struct SyncDouble;
tests/runtimes/smol.rs
+1 -1
View File
@@ -1,6 +1,6 @@
//! Runtime compatibility tests for smol
use dagx::{DagRunner, Task};
use dagx::DagRunner;
use crate::common::task_fn;
struct Value(i32);
tests/tracing/with_tracing.rs
+1 -1
View File
@@ -1,6 +1,6 @@
//! Tests with tracing feature enabled
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
use tracing_subscriber::{fmt, EnvFilter};
struct Value(i32);
tests/tracing/without_tracing.rs
+1 -1
View File
@@ -1,6 +1,6 @@
//! Tests that the library works correctly without tracing feature
use dagx::{task, DagRunner, Task};
use dagx::{task, DagRunner};
struct Value(i32);