From f4160d2387565f0168ef39c3af73fbdd53f795fd Mon Sep 17 00:00:00 2001 From: TechnoPorg Date: Tue, 20 Jan 2026 21:03:07 +0100 Subject: [PATCH] Update docs --- README.md | 12 ++++--- dagx-macros/src/lib.rs | 4 +-- docs/DESIGN_PHILOSOPHY.md | 49 ++++++++++++++--------------- src/extract.rs | 13 +++----- src/extract/coverage_tests.rs | 18 ++++------- src/extract/tests.rs | 59 ++++++++--------------------------- src/lib.rs | 2 +- src/node.rs | 12 +++---- src/node/tests.rs | 36 +++++---------------- src/runner.rs | 20 +++++------- src/task.rs | 4 +-- tests/execution/basic.rs | 54 +++++++++----------------------- tests/parallelism/spawning.rs | 20 +++++++----- 13 files changed, 106 insertions(+), 197 deletions(-) diff --git a/README.md b/README.md index 2272c57..4fb48b9 100644 --- a/README.md +++ b/README.md @@ -104,7 +104,6 @@ dagx provides true parallel execution with sub-microsecond overhead per task. **Why is dagx so fast?** - **Inline fast-path**: Sequential chains execute inline without spawning (8.9-129x faster) -- **Primitives as scheduler**: No custom scheduler — channels coordinate execution - **Adaptive execution**: Inline for sequential work, true parallelism for concurrent work - **Zero-cost abstractions**: Generics and monomorphization eliminate overhead @@ -223,13 +222,18 @@ dagx works with any async runtime. Provide a spawner function to `run()`: ```rust // With Tokio -dag.run().await.unwrap(); +// The join handle result can be unwrapped because dagx catches panics internally +dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap(); // With async-std -dag.run().await.unwrap(); +dag.run(|fut| async_std::task::spawn(fut)).await.unwrap(); // With smol -dag.run().await.unwrap(); +dag.run(|fut| smol::spawn(fut)).await.unwrap(); + +// Single-threaded on the invoking runtime +// Faster for situations with many computationally light tasks +dag.run(|fut| fut).await.unwrap() ``` ## Tutorials & Examples diff --git a/dagx-macros/src/lib.rs b/dagx-macros/src/lib.rs index 4339bb3..a75bab1 100644 --- a/dagx-macros/src/lib.rs +++ b/dagx-macros/src/lib.rs @@ -394,10 +394,10 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream { let mut iter = dependencies.into_iter(); - // Extract each receiver + // Extract each parameter let arc_results = (#( iter.next() - .ok_or_else(|| format!("Missing receiver at index {}", #indices))? + .ok_or_else(|| format!("Missing parameter at index {}", #indices))? .downcast::<#param_types>() .map_err(|_| format!("Type mismatch at index {}: expected Arc<{}>", #indices, std::any::type_name::<#param_types>()))?, diff --git a/docs/DESIGN_PHILOSOPHY.md b/docs/DESIGN_PHILOSOPHY.md index 59cf447..7c2d29d 100644 --- a/docs/DESIGN_PHILOSOPHY.md +++ b/docs/DESIGN_PHILOSOPHY.md @@ -8,28 +8,23 @@ Traditional DAG executors contain substantial scheduling code—algorithms that dagx eliminates this entirely: -1. **Wire up tasks with primitives**: During DAG construction, tasks are connected using channels and organized into topological layers based on their dependencies +1. **Wire up tasks with primitives**: During DAG construction, tasks are organized into topological layers based on their dependencies 2. **Start everything simultaneously**: When you call `run()`, all tasks spawn at once—there's no scheduler deciding when each task should start -3. **Let the primitives handle coordination**: Channels naturally enforce execution order—tasks wait on their input channels until upstream tasks send data. No custom orchestration needed. -4. **Runtime joins on completion**: The runtime simply spawns all tasks and waits for the completion channel to close. That's it. +3. **Let the primitives handle coordination**: Layers naturally enforce execution order—tasks wait on their dependencies until . No custom orchestration needed. +4. **Runtime joins on completion**: The runtime simply spawns all tasks and waits for them to finish. That's it. ## The Implementation Under the hood, dagx uses: -- **Oneshot channels**: Each edge in the DAG gets a `futures::channel::oneshot` - producer sends once, consumer receives once - **Ownership model**: Tasks take ownership (`self`) and are consumed during execution - no Mutex needed for task state -- **Direct data flow**: Outputs flow from producer to consumer via channels, never stored in shared memory during execution +- **Direct data flow**: Outputs flow from producer(s) to consumer(s) through a shared hash map - **Fast-path optimization**: Single-task layers execute inline without spawning overhead The core execution logic in `run()` is remarkably simple: ```rust -// Create oneshot channel for each edge -for (producer, consumer) in edges { - let (tx, rx) = oneshot::channel(); - // Wire tx to producer, rx to consumer -} +let outputs = HashMap::new(); // Execute layer by layer for layer in layers { @@ -40,25 +35,27 @@ for layer in layers { // Convert panic to error to match spawned task behavior } else { // Parallel path: Spawn all tasks in layer - for task in layer { - spawner(async { - let input = await_on_input_channels(); // Blocks until deps complete + let tasks = layers + .map(async { + let input = get_inputs(map); // Deps are guaranteed to be in the map let output = task.run(input); // Consumes task - send_to_output_channels(output); // Sends to dependents - }); + output + }) + .collect::>(); // Run all tasks in parallel + + while let Some(out) = tasks.next().await { + outputs.insert(out.task_id, out.output); // Make result available for dependents } - // Wait for layer completion } + // Wait for layer completion } -// Tasks coordinate themselves via channels - no scheduler needed +// FuturesUnordered coordinates task execution - no custom scheduler needed ``` -No state machines. No task queues. No wake-up logic. No Mutex locks. Just channels doing what channels do, with an inline fast-path for sequential execution. - ## Inline Execution Fast-Path -**Performance optimization for sequential workloads**: When a layer contains only a single task (common in deep chains and linear pipelines), dagx executes it inline rather than spawning it. This eliminates spawning overhead, context switching, and channel creation, resulting in 10-100x performance improvements for sequential patterns. +**Performance optimization for sequential workloads**: When a layer contains only a single task (common in deep chains and linear pipelines), dagx executes it inline rather than spawning it. This eliminates spawning overhead and context switching, resulting in a 10-100x performance improvements for sequential patterns. **Panic handling guarantee**: To maintain behavioral consistency between inline and spawned execution, panics in inline tasks are caught using `FutureExt::catch_unwind()` and converted to errors. This matches the behavior of all major async runtimes (Tokio, async-std, smol, embassy-rs), which catch panics in spawned tasks and convert them to `JoinError` or equivalent. @@ -72,19 +69,19 @@ This ensures your code behaves the same whether a task runs inline or spawned, m ## Benefits -**Simplicity**: The runtime is straightforward: create channels, spawn tasks, let them coordinate via awaiting. No complex scheduler code to maintain, debug, or optimize. +**Simplicity**: The runtime is straightforward: spawn tasks, coordinate via compile-time dependency guarantees. No complex scheduler code to maintain, debug, or optimize. -**Reliability**: Built on battle-tested primitives (oneshot channels, async/await) from Rust's standard library and the futures crate. These have been used in production by thousands of projects and are orders of magnitude more reliable than custom scheduling logic. +**Reliability**: Built on battle-tested primitives from Rust's standard library and the futures crate. These have been used in production by thousands of projects and are orders of magnitude more reliable than custom scheduling logic. -**Bug resistance**: Fewer moving parts means fewer places for bugs to hide. The type system enforces correct wiring at compile time. Channels handle synchronization. The ownership model prevents data races. What's left to break? +**Bug resistance**: Fewer moving parts means fewer places for bugs to hide. The type system enforces correct wiring at compile time. The ownership model prevents data races. What's left to break? -**Performance**: Near zero-overhead. No Mutex locks during execution. Arc reference counting for efficient fanout (atomic operations, not locks). Oneshot channels are often lock-free. Tasks start as soon as their dependencies complete - maximum parallelism. +**Performance**: Near zero-overhead. No Mutex locks during execution. Arc reference counting for efficient fanout (atomic operations, not locks). Tasks start as soon as their dependencies complete - maximum parallelism. -**Auditability**: Want to verify correctness? Check the channel wiring, verify tasks await their inputs, done. No need to trace through complex state machine transitions or wake-up cascades. +**Auditability**: Want to verify correctness? Check the dependency wiring, verify tasks await their inputs, done. No need to trace through complex state machine transitions or wake-up cascades. ## The Insight -The key insight is that **dependencies ARE the schedule**. If task B depends on task A's output, a channel naturally enforces that B waits for A. The dependency graph already encodes all the scheduling information—we just need to wire up channels to match it. +The key insight is that **dependencies ARE the schedule**. If task B depends on task A's output, the topological sort naturally enforces that B waits for A. The dependency graph already encodes all the scheduling information—we just need to wire outputs to match it. This is dagx's core philosophy: leverage the type system for correctness, use primitives for coordination, and let the compiler optimize everything else away. diff --git a/src/extract.rs b/src/extract.rs index 7c9f06f..a338e12 100644 --- a/src/extract.rs +++ b/src/extract.rs @@ -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 from channel, clones the inner value for task consumption. +// Receives Arc 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>()))?, diff --git a/src/extract/coverage_tests.rs b/src/extract/coverage_tests.rs index 15ec72e..60ea9f8 100644 --- a/src/extract/coverage_tests.rs +++ b/src/extract/coverage_tests.rs @@ -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 type mismatch error message - // Arc is special: it expects Arc> from channel - // Create a channel that sends Arc instead of Arc> - let (tx, rx) = oneshot::channel::>(); - - // Send an Arc 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> from dependency + // Create a dependency that provides Arc instead of Arc> + let deps = vec![Arc::new("not double arc".to_string()) as Arc]; // Try to extract as Arc - this will fail with type mismatch - // because it expects Arc> in the channel - let deps = vec![Box::new(rx) as Box]; - let result = Arc::::extract_from_channels(deps).await; + // because it expects Arc> in the dependency + let result = Arc::::extract_from_deps(deps).await; // Verify the error message contains the expected type information assert!(result.is_err()); diff --git a/src/extract/tests.rs b/src/extract/tests.rs index dacbbe9..3102d59 100644 --- a/src/extract/tests.rs +++ b/src/extract/tests.rs @@ -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::(); - - // 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]; - 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]; + 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::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong type".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = HashMap::::extract_from_channels(deps).await; + let deps = vec![Arc::new("wrong type".to_string()) as Arc]; + let result = HashMap::::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::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Result::::extract_from_channels(deps).await; + let deps = vec![Arc::new("wrong".to_string()) as Arc]; + let result = Result::::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::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Option::::extract_from_channels(deps).await; + let deps = vec![Arc::new("wrong".to_string()) as Arc]; + let result = Option::::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::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Vec::::extract_from_channels(deps).await; + let deps = vec![Arc::new("wrong".to_string()) as Arc]; + let result = Vec::::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::>(); - tokio::spawn(async move { - let _ = tx.send(Arc::new("wrong".to_string())); - }); - - let deps = vec![Box::new(rx) as Box]; - let result = Arc::::extract_from_channels(deps).await; + let deps = vec![Arc::new("wrong".to_string()) as Arc]; + let result = Arc::::extract_from_deps(deps).await; assert!(result.is_err()); assert!(result.unwrap_err().contains("Type mismatch")); } diff --git a/src/lib.rs b/src/lib.rs index 903fe3b..7495f44 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -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 diff --git a/src/node.rs b/src/node.rs index 41d3b27..bd9da25 100644 --- a/src/node.rs +++ b/src/node.rs @@ -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 { 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 }); } }; diff --git a/src/node/tests.rs b/src/node/tests.rs index c990071..c74a0b2 100644 --- a/src/node/tests.rs +++ b/src/node/tests.rs @@ -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::>(); - // Execute dependency node with sender - let dep_senders = vec![Box::new(tx) as Box]; - 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]; - let senders = vec![]; + let dependencies = vec![dep_result as Arc]; - 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::>>() - .unwrap(); - - let received_arc = rx.await; - assert!(received_arc.is_ok()); - assert_eq!(*received_arc.unwrap(), 42); + let arc_value = result.unwrap().downcast::().unwrap(); // Also verify output was returned (Arc-wrapped) let output = result.unwrap(); diff --git a/src/runner.rs b/src/runner.rs index c1ab82c..cadbf78 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -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); diff --git a/src/task.rs b/src/task.rs index 54ee361..32883b0 100644 --- a/src/task.rs +++ b/src/task.rs @@ -87,7 +87,7 @@ use std::{future::Future, sync::Arc}; /// # Output Wrapping /// /// **Task outputs are automatically wrapped in `Arc` internally** for efficient fan-out. -/// You define `Output = T`, but the framework wraps it in `Arc` before sending to channels. +/// You define `Output = T`, but the framework wraps it in `Arc` 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 + 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 diff --git a/tests/execution/basic.rs b/tests/execution/basic.rs index b9e343d..565a803 100644 --- a/tests/execution/basic.rs +++ b/tests/execution/basic.rs @@ -6,9 +6,7 @@ use dagx::*; #[tokio::test] async fn test_empty_dag() { let dag = DagRunner::new(); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); // Should complete without error } @@ -17,9 +15,7 @@ async fn test_empty_dag() { async fn test_single_task() { let dag = DagRunner::new(); let task = dag.add_task(task_fn(|_: ()| async { 42 })); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); assert_eq!(dag.get(task).unwrap(), 42); @@ -43,9 +39,7 @@ async fn test_deep_chain() { .add_task(task_fn(|x: i32| async move { x + 1 })) .depends_on(d); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -62,9 +56,7 @@ async fn test_wide_parallel() { .map(|i| dag.add_task(task_fn(move |_: ()| async move { i * 2 }))) .collect(); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -88,9 +80,7 @@ async fn test_diamond_dependency() { .add_task(task_fn(|(x, y): (i32, i32)| async move { x + y })) .depends_on((&b, &c)); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -118,9 +108,7 @@ async fn test_different_output_types() { } })); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -153,9 +141,7 @@ async fn test_multiple_sinks() { .add_task(task_fn(|x: i32| async move { x * 3 })) .depends_on(branch2); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); // Should wait for both sinks @@ -177,9 +163,7 @@ async fn test_single_task_inline_path() { .add_task(task_fn(|x: i32| async move { x + 5 })) .depends_on(t2); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -204,9 +188,7 @@ async fn test_multi_task_spawn_path() { .add_task(task_fn(|x: i32| async move { x + 3 })) .depends_on(&source); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -230,9 +212,7 @@ async fn test_producer_consumer_channels() { )) .depends_on(&producer); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -241,7 +221,7 @@ async fn test_producer_consumer_channels() { #[tokio::test] async fn test_multi_consumer_fanout() { - // Test fanout with multiple consumers from one producer (channel creation) + // Test fanout with multiple consumers from one producer let dag = DagRunner::new(); let producer = dag.add_task(task_fn(|_: ()| async { 42 })); @@ -257,9 +237,7 @@ async fn test_multi_consumer_fanout() { .add_task(task_fn(|x: i32| async move { x - 5 })) .depends_on(&producer); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -287,9 +265,7 @@ async fn test_compute_layers_with_dependents_tracking() { .add_task(task_fn(|(l, r): (i32, i32)| async move { l + r })) .depends_on((&left, &right)); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); @@ -337,9 +313,7 @@ async fn test_layer_execution_order() { })) .depends_on(t2); - dag.run(|fut| { - tokio::spawn(fut); - }) + dag.run() .await .unwrap(); diff --git a/tests/parallelism/spawning.rs b/tests/parallelism/spawning.rs index 01a0bca..634ad45 100644 --- a/tests/parallelism/spawning.rs +++ b/tests/parallelism/spawning.rs @@ -16,16 +16,22 @@ async fn test_spawner_actually_spawns_tasks() -> DagResult<()> { // Create 10 independent tasks let tasks: Vec<_> = (0..10) - .map(|i| dag.add_task(task_fn(move |_: ()| async move { i }))) + .map(|i| { + dag.add_task(task_fn({ + let value = spawn_count.clone(); + move |_: ()| { + let value = value.clone(); + async move { + value.fetch_add(1, Ordering::SeqCst); + i + } + } + })) + }) .collect(); // Custom spawner that counts invocations - let counter = spawn_count.clone(); - dag.run(move |fut| { - counter.fetch_add(1, Ordering::SeqCst); - tokio::spawn(fut); - }) - .await?; + dag.run().await?; // Verify all tasks were spawned for (i, task) in tasks.iter().enumerate() {