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

Update docs

Browse Source
This commit is contained in:
TechnoPorg committed 2026-01-22 00:30:48 +01:00
1 parent df39dd80b9
commit f4160d2387
13 files changed
+106 -197
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

src/extract.rs
+5 -8
View File
@@ -1,4 +1,4 @@
//! Input extraction for task dependencies via oneshot channels.
//! Input extraction for task dependencies via a list of dependencies.
//!
//! # Legacy Trait for Internal Use Only
//!
@@ -35,7 +35,7 @@
use std::any::Any;
use std::sync::Arc;
/// Helper trait for async input extraction from oneshot channels.
/// Helper trait for async input extraction from list of dependencies.
///
/// 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.
@@ -59,7 +59,7 @@ impl ExtractInput for () {
// 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.
// Receives Arc<T> from runner, clones the inner value for task consumption.
// For custom types, implement ExtractInput following the same pattern.
macro_rules! impl_extract_single {
($($t:ty),+) => {
@@ -217,9 +217,6 @@ where
/// 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.
///
@@ -244,12 +241,12 @@ macro_rules! impl_extract_tuple {
let mut iter = dependencies.into_iter();
// dependencies now contain Arc-wrapped values for efficient fanout
// Dereference Box after downcasting to get the receiver itself
// Dereference Box after downcasting to get the parameter itself
// Use type names as variable names to avoid numeric suffix issues
let arc_results = (
$(
iter.next()
.ok_or_else(|| format!("Missing receiver at index {}", $idx))?
.ok_or_else(|| format!("Missing parameter at index {}", $idx))?
.downcast::<$T>()
.map_err(|_| format!("Type mismatch at index {}: expected Arc<{}>",
$idx, std::any::type_name::<$T>()))?,
src/extract/coverage_tests.rs
+6 -12
View File
@@ -2,7 +2,7 @@
//!
//! These tests specifically target type mismatch error messages that aren't hit
//! in normal usage because the #[task] macro generates type-safe code.
//! We directly call extract_from_channels with mismatched types.
//! We directly call extract_from_deps with mismatched types.
use crate::extract::ExtractInput;
use futures::channel::oneshot;
@@ -141,19 +141,13 @@ async fn test_hashmap_type_mismatch_coverage() {
#[tokio::test]
async fn test_arc_type_mismatch_coverage() {
// Tests line 211 in extract.rs - Arc<T> type mismatch error message
// Arc is special: it expects Arc<Arc<T>> from channel
// Create a channel that sends Arc<String> instead of Arc<Arc<String>>
let (tx, rx) = oneshot::channel::<Arc<String>>();
// Send an Arc<String> value (missing one level of Arc wrapping)
tokio::spawn(async move {
let _ = tx.send(Arc::new("not double arc".to_string()));
});
// Arc is special: it expects Arc<Arc<T>> from dependency
// Create a dependency that provides Arc<String> instead of Arc<Arc<String>>
let deps = vec![Arc::new("not double arc".to_string()) as Arc<dyn Any + Send + Sync>];
// Try to extract as Arc<String> - this will fail with type mismatch
// because it expects Arc<Arc<String>> in the channel
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
let result = Arc::<String>::extract_from_channels(deps).await;
// because it expects Arc<Arc<String>> in the dependency
let result = Arc::<String>::extract_from_deps(deps).await;
// Verify the error message contains the expected type information
assert!(result.is_err());
src/extract/tests.rs
+13 -46
View File
@@ -4,7 +4,6 @@
use crate::extract::ExtractInput;
use crate::task::Task;
use futures::channel::oneshot;
use std::any::Any;
use std::sync::Arc;
@@ -206,16 +205,9 @@ async fn test_extract_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;
// Try to extract as wrong type - put wrong parameter type in deps
let deps = vec![Arc::new("hello".to_string()) as Arc<dyn Any + Send + Sync>];
let result = i32::extract_from_deps(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
@@ -312,13 +304,8 @@ async fn test_hashmap_extract_wrong_dependency_count() {
#[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;
let deps = vec![Arc::new("wrong type".to_string()) as Arc<dyn Any + Send + Sync>];
let result = HashMap::<String, i32>::extract_from_deps(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
@@ -336,39 +323,24 @@ async fn test_hashmap_extract_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;
let deps = vec![Arc::new("wrong".to_string()) as Arc<dyn Any + Send + Sync>];
let result = Result::<i32, String>::extract_from_deps(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;
let deps = vec![Arc::new("wrong".to_string()) as Arc<dyn Any + Send + Sync>];
let result = Option::<i32>::extract_from_deps(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;
let deps = vec![Arc::new("wrong".to_string()) as Arc<dyn Any + Send + Sync>];
let result = Vec::<i32>::extract_from_deps(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
@@ -383,13 +355,8 @@ async fn test_arc_extract_wrong_dependency_count() {
#[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;
let deps = vec![Arc::new("wrong".to_string()) as Arc<dyn Any + Send + Sync>];
let result = Arc::<i32>::extract_from_deps(deps).await;
assert!(result.is_err());
assert!(result.unwrap_err().contains("Type mismatch"));
}
src/lib.rs
+1 -1
View File
@@ -64,7 +64,7 @@
//! dagx automatically optimizes execution for sequential workloads using an inline fast-path.
//! When a layer contains only a single task (common in deep chains and linear pipelines),
//! that task executes inline rather than being spawned. This eliminates spawning overhead,
//! context switching, and channel creation.
//! and context switching.
//!
//! **Panic handling**: To maintain consistent behavior between inline and spawned execution,
//! panics in inline tasks are caught using `FutureExt::catch_unwind()` and converted to
src/node.rs
+5 -7
View File
@@ -25,12 +25,12 @@ type ExecuteFuture =
/// 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 - tasks 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 {
/// Execute the task with type-erased input receivers and output senders.
/// Execute the task with type-erased inputs and output
///
/// The TypedNode implementation knows the concrete types and can safely downcast.
/// Consumes the node and returns the output value.
@@ -50,10 +50,7 @@ pub(crate) trait ExecutableNode: Send {
/// 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
///
/// Outputs are communicated via oneshot channels for inter-task communication.
/// The output is also returned from execute_with_channels for central storage.
/// The output is also returned from execute_with_deps for central storage.
pub(crate) struct TypedNode<T: Task> {
pub(crate) id: NodeId,
pub(crate) task: T,
@@ -81,7 +78,8 @@ where
let output = match self.task.extract_and_run(dependencies).await {
Ok(output) => output,
Err(_msg) => {
// Failed to extract input - likely channel closed or type mismatch
// Failed to extract input - likely type mismatch.
// This should technically be infallible, so it might be worth removing in the future.
return Err(DagError::InvalidDependency { task_id: self.id.0 });
}
};
src/node/tests.rs
+7 -29
View File
@@ -67,17 +67,13 @@ async fn test_execute_with_channels_missing_input() {
}
#[tokio::test]
async fn test_execute_with_channels_success() {
async fn test_execute_with_deps_success() {
// Test successful execution with channels
let dependency_node = Box::new(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>>();
// Execute dependency node with sender
let dep_senders = vec![Box::new(tx) as Box<dyn std::any::Any + Send>];
let dep_receivers = vec![];
let dep_dependencies = vec![];
// Execute in background
let dep_handle = tokio::spawn(async move {
@@ -87,15 +83,9 @@ async fn test_execute_with_channels_success() {
});
// Execute dependent node with receiver
let receivers = vec![Box::new(rx) as Box<dyn std::any::Any + Send>];
let senders = vec![];
let dependencies = vec![dep_result as Arc<dyn std::any::Any + Send + Sync>];
let result = dependent_node
.execute_with_channels(receivers, senders)
.await;
// Wait for dependency to complete
let _ = dep_handle.await;
let result = dependent_node.execute_with_deps(dependencies).await;
assert!(result.is_ok());
}
@@ -106,8 +96,7 @@ async fn test_execute_with_channels_sink_node() {
let node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 42 }));
// No senders means this is a sink node
let receivers = vec![];
let senders = vec![];
let dependencies = vec![];
let result = node.execute_with_channels(receivers, senders).await;
assert!(result.is_ok());
@@ -119,13 +108,10 @@ async fn test_execute_with_channels_sink_node() {
}
#[tokio::test]
async fn test_execute_with_channels_non_sink_node() {
async fn test_execute_with_deps_non_sink_node() {
// Test that non-sink nodes send output via channels
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
// Create output channels for one dependent
let (senders, mut receivers) = node.create_output_channels(1);
// Execute with senders (non-sink node)
let node = Box::new(node);
let exec_receivers = vec![];
@@ -133,15 +119,7 @@ async fn test_execute_with_channels_non_sink_node() {
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>>>()
.unwrap();
let received_arc = rx.await;
assert!(received_arc.is_ok());
assert_eq!(*received_arc.unwrap(), 42);
let arc_value = result.unwrap().downcast::<i32>().unwrap();
// Also verify output was returned (Arc-wrapped)
let output = result.unwrap();
src/runner.rs
+7 -13
View File
@@ -231,10 +231,13 @@ impl DagRunner {
///
/// # Parameters
///
/// - `spawner`: A function that spawns futures on the async runtime. Examples:
/// - Tokio: `|fut| { tokio::spawn(fut); }`
/// - Smol: `|fut| { smol::spawn(fut).detach(); }`
/// - Async-std: `|fut| { async_std::task::spawn(fut); }`
/// - `spawner`: A function that spawns futures on the async runtime
/// and returns a task handle. Examples:
/// - Tokio: `|fut| { tokio::spawn(fut).map(Result::unwrap) }`
/// - Smol: `|fut| { smol::spawn(fut) }`
/// - Async-std: `|fut| { async_std::task::spawn(fut) }`
/// - Single-threaded: `|fut| fut`
/// - For computationally light tasks, concurrency without parallelism can be significantly faster.
///
/// # Errors
///
@@ -276,13 +279,6 @@ impl DagRunner {
/// dag.run().await.unwrap(); // Executes all tasks
/// # };
/// ```
///
/// # Implementation Note
///
/// Tasks communicate via oneshot channels created fresh during run().
/// This eliminates Mutex contention on outputs and enables true streaming execution.
/// Type erasure occurs only at the ExecutableNode trait boundary - channels are created
/// with full type information and type-erased just before passing to execute_with_channels.
#[inline]
#[cfg_attr(feature = "tracing", tracing::instrument(skip(self)))]
pub async fn run(&self) -> DagResult<()> {
@@ -334,7 +330,6 @@ impl DagRunner {
// pipelines), we execute it inline rather than spawning it. This provides
// 10-100x performance improvements for sequential workloads by eliminating:
// - Task spawning overhead
// - Channel creation/destruction for layer coordination
// - Context switching to/from the runtime
//
// CRITICAL: Panic handling is required to maintain behavioral consistency.
@@ -407,7 +402,6 @@ impl DagRunner {
})
});
// Send the output (ignore send errors - receiver may be dropped)
match result {
Ok(output) => {
outputs.insert(node_id, output);
src/task.rs
+2 -2
View File
@@ -87,7 +87,7 @@ use std::{future::Future, sync::Arc};
/// # 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 = T`, but the framework wraps it in `Arc<T>` before sending to dependents.
/// 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
@@ -98,7 +98,7 @@ pub trait Task: Send {
fn run(self, input: Self::Input) -> impl Future<Output = Self::Output> + Send;
/// Internal method: Extract input from type-erased channels and execute the task.
/// Internal method: Extract input from list of type-erased dependencies and execute the task.
///
/// This method is implemented automatically by the #[task] macro with inline
/// extraction logic based on the task's parameter signature. This allows ANY