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

Update tests, examples, benches

Browse Source
This commit is contained in:
TechnoPorg committed 2026-01-22 00:30:48 +01:00
1 parent bacac903b3
commit 85ccd7e507
89 files changed
+942 -1785
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/arc_tests.rs
+54 -91
View File
@@ -10,6 +10,7 @@
//! These tests serve as both validation and documentation for proper Arc usage.
use dagx::{task, DagRunner, Task};
use futures::FutureExt;
use std::collections::{HashMap, HashSet};
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -43,11 +44,9 @@ async fn test_arc_basic_pipeline() {
let producer = dag.add_task(Producer);
let consumer = dag.add_task(Consumer).depends_on(&producer);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: producer is not a sink (consumer depends on it), so we can't get() its value
// Only sink nodes (nodes with no dependents) store their outputs for retrieval
@@ -101,11 +100,9 @@ async fn test_arc_with_primitives() {
.add_task(AllConsumer)
.depends_on((&i32_prod, &f64_prod, &bool_prod));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: producers are not sinks (consumer depends on them), so we can't get() their values
// The consumer output verifies all three producers worked correctly
@@ -177,11 +174,9 @@ async fn test_arc_with_collections() {
.add_task(CollectionConsumer)
.depends_on((&vec_prod, &map_prod, &set_prod));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: producers are not sinks (consumer depends on them), so we can't get() their values
// The consumer output verifies all three collection producers worked correctly
@@ -250,11 +245,9 @@ async fn test_arc_with_custom_structs() {
let age_ext = dag.add_task(AgeExtractor).depends_on(&person_prod);
let email_ext = dag.add_task(EmailCountExtractor).depends_on(&person_prod);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: person_prod is not a sink (3 extractors depend on it), so we can't get() its value
// The extractor outputs verify the person was created and shared correctly
@@ -318,11 +311,9 @@ async fn test_arc_fan_out_three_consumers() {
let c2 = dag.add_task(FirstConsumer).depends_on(&producer);
let c3 = dag.add_task(JoinConsumer).depends_on(&producer);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(c1).unwrap(), 5);
assert_eq!(dag.get(c2).unwrap(), "item1");
@@ -394,11 +385,9 @@ async fn test_arc_fan_out_five_consumers_large_data() {
let c4 = dag.add_task(PatternCounter).depends_on(&producer);
let c5 = dag.add_task(TotalBytesComputer).depends_on(&producer);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: producer is not a sink (has 5 consumers), so we can't get() its value
// We verify correctness through the consumer outputs instead
@@ -486,11 +475,9 @@ async fn test_arc_nested_fan_out_tree() {
let l2_negate = dag.add_task(Level2Negate).depends_on(l1_prod);
let l2_mod = dag.add_task(Level2Mod10).depends_on(l1_prod);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: l1_sum and l1_prod are not sinks (they have level 2 tasks depending on them)
// Only the level 2 tasks are sinks and can be retrieved
@@ -556,11 +543,9 @@ async fn test_arc_performance_vs_clone() {
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let elapsed = start.elapsed();
@@ -658,11 +643,9 @@ async fn test_arc_with_result() {
let consumer3 = dag.add_task(ArcResultConsumer2).depends_on(&arc_result_err);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(consumer1).unwrap(), "Got Ok: Success");
assert_eq!(dag.get(consumer2).unwrap(), "Got Arc: Success");
@@ -767,11 +750,9 @@ async fn test_arc_with_option() {
let c3 = dag.add_task(OptionArcConsumer1).depends_on(&opt_arc_some);
let c4 = dag.add_task(OptionArcConsumer2).depends_on(&opt_arc_none);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(c1).unwrap(), "Some(42)");
assert_eq!(dag.get(c2).unwrap(), "None");
@@ -824,11 +805,9 @@ async fn test_arc_diamond_dependency() {
let right = dag.add_task(RightBranch).depends_on(&top);
let bottom = dag.add_task(Bottom).depends_on((&left, &right));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(bottom).unwrap(), "Sum: 15, Product: 120");
}
@@ -894,11 +873,9 @@ async fn test_arc_empty_collections() {
.add_task(EmptyConsumer)
.depends_on((&vec_task, &string_task, &map_task));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(
dag.get(consumer).unwrap(),
@@ -967,11 +944,9 @@ async fn test_arc_very_large_data() {
let last = dag.add_task(LastSampler).depends_on(&producer);
let middle = dag.add_task(MiddleSampler).depends_on(&producer);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(counter).unwrap(), SIZE);
assert!(dag.get(first).unwrap().contains("Item number 0"));
@@ -1006,11 +981,9 @@ async fn test_arc_of_arc() {
let producer = dag.add_task(NestedProducer);
let consumer = dag.add_task(NestedConsumer).depends_on(&producer);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(consumer).unwrap(), "Value: Nested Arc");
}
@@ -1072,11 +1045,9 @@ async fn test_arc_mixed_with_non_arc() {
let unwrap = dag.add_task(Unwrapper).depends_on(&arc_prod);
let wrap = dag.add_task(Wrapper).depends_on(&non_arc_prod);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(
dag.get(mixed).unwrap(),
@@ -1142,11 +1113,9 @@ async fn test_arc_parallel_execution() {
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let elapsed = start.elapsed();
@@ -1193,11 +1162,9 @@ async fn test_arc_reference_counting() {
let producer = dag.add_task(RefCountProducer);
let checker = dag.add_task(RefCountChecker).depends_on(&producer);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: producer is not a sink (checker depends on it), so we can't get() its value
// The checker task verifies the Arc was passed correctly
@@ -1271,9 +1238,7 @@ async fn test_arc_massive_fanout_performance_comparison() {
let start_without_arc = Instant::now();
dag_without_arc
.run(|fut| {
tokio::spawn(fut);
})
.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let duration_without_arc = start_without_arc.elapsed();
@@ -1325,9 +1290,7 @@ async fn test_arc_massive_fanout_performance_comparison() {
let start_with_arc = Instant::now();
dag_with_arc
.run(|fut| {
tokio::spawn(fut);
})
.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let duration_with_arc = start_with_arc.elapsed();
tests/boundaries/deep_chains.rs
+10 -36
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -19,10 +20,7 @@ async fn test_1000_level_deep_chain() -> DagResult<()> {
.depends_on(current);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(current)?, 999);
Ok(())
@@ -52,10 +50,7 @@ async fn test_deep_chain_with_branches() -> DagResult<()> {
}
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(main_chain)?, 499);
@@ -113,10 +108,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
&chain_ends[7],
));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Each chain: start + 500
// Chain 0: 0 + 500 = 500
@@ -161,10 +153,7 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
.depends_on(current);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(order.load(Ordering::SeqCst), 100);
assert_eq!(dag.get(current)?, 99);
@@ -192,10 +181,7 @@ async fn test_deep_binary_tree() -> DagResult<()> {
let root = build_tree(&dag, 10, 1);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Sum of all values in a complete binary tree with values 1024..2047
// This is sum from 2^10 to 2^11-1 = 1024 * 1023 / 2 + 1024 * 512 = 1047552
@@ -225,10 +211,7 @@ async fn test_fibonacci_chain() -> DagResult<()> {
prev1 = next;
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// 50th Fibonacci number (0-indexed) = 7778742049
assert_eq!(dag.get(prev1)?, 7778742049);
@@ -251,10 +234,7 @@ async fn test_deep_chain_with_accumulator() -> DagResult<()> {
.depends_on(current);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(current)?;
assert_eq!(result, 20100); // sum of 1..=200
@@ -292,10 +272,7 @@ async fn test_zigzag_dependencies() -> DagResult<()> {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a + b }))
.depends_on((&chain_a, &chain_b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(final_task)?;
assert!(result > 100);
@@ -318,10 +295,7 @@ async fn test_10000_level_chain_stress() -> DagResult<()> {
.depends_on(current);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(current)?, 9_999);
Ok(())
tests/boundaries/large_scale.rs
+6 -20
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -25,10 +26,7 @@ async fn test_10000_independent_tasks() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify all tasks completed
assert_eq!(counter.load(Ordering::Relaxed), 10_000);
@@ -103,10 +101,7 @@ async fn test_5000_node_pyramid() -> DagResult<()> {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a + b }))
.depends_on((&current_layer[0], &current_layer[1]));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// We created 1024 ones and 1024 zeros at the bottom
// The sum should propagate up to 1024
@@ -139,10 +134,7 @@ async fn test_wide_dag_1000_sources_1000_sinks() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify some sinks completed correctly
let result0 = dag.get(sinks[0])?;
@@ -195,10 +187,7 @@ async fn test_10000_node_linear_chain_segments() -> DagResult<()> {
chain_ends.push(current);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// All 10,000 tasks should have executed
assert_eq!(counter.load(Ordering::Relaxed), 10_000);
@@ -239,10 +228,7 @@ async fn test_100000_nodes_stress() -> DagResult<()> {
.collect();
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(completed.load(Ordering::Relaxed), 100_000);
Ok(())
tests/boundaries/type_limits.rs
+11 -40
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::marker::PhantomData;
#[tokio::test]
@@ -38,10 +39,7 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> {
}))
.depends_on(&t2);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(t1)?, Empty);
assert_eq!(dag.get(t2)?, 42);
@@ -76,10 +74,7 @@ async fn test_large_value_types() -> DagResult<()> {
}))
.depends_on(&t1);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(t2)?, 1042);
@@ -98,10 +93,7 @@ async fn test_unit_type_chains() -> DagResult<()> {
.add_task(task_fn(|_: ()| async { "done" }))
.depends_on(&t3);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
dag.get(t1)?; // Verify no error
dag.get(t2)?;
@@ -132,10 +124,7 @@ async fn test_nested_option_result_types() -> DagResult<()> {
}))
.depends_on(&t1);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(t2)?, 55);
@@ -159,10 +148,7 @@ async fn test_tuple_of_tuples() -> DagResult<()> {
))
.depends_on((&t1, &t2));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(t3)?, 36);
@@ -196,10 +182,7 @@ async fn test_string_types_variety() -> DagResult<()> {
))
.depends_on((&t1, &t2, &t3, &t4, &t5, &t6));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(combined)?;
assert!(result.contains("static str"));
@@ -228,10 +211,7 @@ async fn test_generic_type_constraints() -> DagResult<()> {
))
.depends_on((&t1, &t2, &t3));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(combined)?;
assert!(result.contains("Wrapper(42)"));
@@ -257,10 +237,7 @@ async fn test_reference_wrapper_types() -> DagResult<()> {
.add_task(task_fn(|val: i32| async move { val * 2 }))
.depends_on(&t1);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(t2)?, 200);
@@ -285,10 +262,7 @@ async fn test_array_types() -> DagResult<()> {
}))
.depends_on((&t1, &t2));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(t3)?, 55);
@@ -327,10 +301,7 @@ async fn test_enum_variants() -> DagResult<()> {
))
.depends_on((&t1, &t2, &t3, &t4));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert!(dag.get(combined)?);
tests/boundaries/zero_tasks.rs
+12 -44
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{task, DagResult, DagRunner, Task};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -10,10 +11,7 @@ async fn test_empty_dag_execution() -> DagResult<()> {
// Test that an empty DAG can be created and run without issues
let dag = DagRunner::new();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Should complete successfully with no tasks
Ok(())
@@ -26,10 +24,7 @@ async fn test_single_task_no_deps() -> DagResult<()> {
let task = dag.add_task(task_fn(|_: ()| async { 42 }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(task)?, 42);
Ok(())
@@ -50,10 +45,7 @@ async fn test_single_task_unit_input_output() -> DagResult<()> {
}
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
dag.get(task)?; // Just verify it doesn't error
assert_eq!(executed.load(Ordering::SeqCst), 1);
@@ -70,10 +62,7 @@ async fn test_zero_sized_type_task() -> DagResult<()> {
let task = dag.add_task(task_fn(|_: ()| async { ZeroSized }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(&task)?, ZeroSized);
Ok(())
@@ -85,10 +74,7 @@ async fn test_empty_dag_multiple_runs() -> DagResult<()> {
let dag = DagRunner::new();
for _ in 0..10 {
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
}
Ok(())
@@ -103,10 +89,7 @@ async fn test_dag_with_only_source_tasks() -> DagResult<()> {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i })))
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task)?, i);
@@ -132,10 +115,7 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> {
let dag = DagRunner::new();
let task = dag.add_task(Counter { value: 21 });
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(task)?, 42);
Ok(())
@@ -149,17 +129,11 @@ async fn test_alternating_empty_and_filled_dag_runs() -> DagResult<()> {
if i % 2 == 0 {
// Empty DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
} else {
// DAG with a task
let task = dag.add_task(task_fn(move |_: ()| async move { i }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(task)?, i);
}
}
@@ -179,10 +153,7 @@ async fn test_dag_with_never_type_simulation() -> DagResult<()> {
// This task completes immediately but returns a Result containing "never" type
let task = dag.add_task(task_fn(|_: ()| async { Result::<i32, Never>::Ok(42) }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(&task)?, Ok(42));
Ok(())
@@ -199,10 +170,7 @@ async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> {
1337
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(task)?, 1337);
Ok(())
tests/custom_type_test.rs
+13 -20
View File
@@ -1,5 +1,6 @@
// Test that custom types work without any trait implementations
use dagx::{task, DagRunner, Task};
use futures::FutureExt;
#[derive(Clone, Debug, PartialEq)]
struct Person {
@@ -56,11 +57,9 @@ async fn test_custom_type_single_dependency() {
let fetch = dag.add_task(FetchPerson);
let process = dag.add_task(ProcessPerson).depends_on(&fetch);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let result = dag.get(process).unwrap();
assert_eq!(result, "Alice is 30 years old");
@@ -74,11 +73,9 @@ async fn test_custom_type_passthrough() {
let transform = dag.add_task(PersonToCompany).depends_on(&fetch);
let process = dag.add_task(ProcessPerson).depends_on(&fetch);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let person_result = dag.get(process).unwrap();
let company_result = dag.get(transform).unwrap();
@@ -99,11 +96,9 @@ async fn test_custom_type_diamond_pattern() {
let path_a = dag.add_task(ProcessPerson).depends_on(&fetch);
let path_b = dag.add_task(PersonToCompany).depends_on(&fetch);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let result_a = dag.get(path_a).unwrap();
let result_b = dag.get(path_b).unwrap();
@@ -121,11 +116,9 @@ async fn test_custom_type_fan_out() {
let process2 = dag.add_task(ProcessPerson).depends_on(&fetch);
let process3 = dag.add_task(ProcessPerson).depends_on(&fetch);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let result1 = dag.get(process1).unwrap();
let result2 = dag.get(process2).unwrap();
tests/cycle_prevention.rs
+4 -5
View File
@@ -4,6 +4,7 @@
//! prevents it through the type-state pattern.
use dagx::{task, DagRunner, Task, TaskHandle};
use futures::FutureExt;
/// Demonstrates that TaskBuilder is consumed by depends_on,
/// preventing you from wiring the same task multiple times
@@ -270,11 +271,9 @@ async fn test_type_safety_prevents_invalid_graphs() {
let result = dag.add_task(Aggregate).depends_on((&t1, &t2));
// Execute and verify
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(result).unwrap(), 40); // (10*2) + (10*2)
tests/dependencies/resolution.rs
+5 -16
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::{Arc, Mutex};
#[tokio::test]
@@ -81,10 +82,7 @@ async fn test_dependency_resolution_order() -> DagResult<()> {
.depends_on(d)
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(e)?, 5);
@@ -118,10 +116,7 @@ async fn test_transitive_dependencies() -> DagResult<()> {
.add_task(task_fn(|x: i32| async move { x + 5 }))
.depends_on(b);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(&a)?, 10);
assert_eq!(dag.get(b)?, 20);
@@ -144,10 +139,7 @@ async fn test_deep_chain_50_levels() -> DagResult<()> {
handles.push(next);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(handles.last().unwrap())?, 50);
Ok(())
@@ -166,10 +158,7 @@ async fn test_deep_chain_200_levels() -> DagResult<()> {
.depends_on(current);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(current)?, 201);
Ok(())
tests/dependencies/single.rs
+6 -20
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
#[tokio::test]
async fn test_single_dependency() -> DagResult<()> {
@@ -12,10 +13,7 @@ async fn test_single_dependency() -> DagResult<()> {
.add_task(task_fn(|x: i32| async move { x + 1 }))
.depends_on(&source);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(dependent)?, 43);
Ok(())
@@ -42,10 +40,7 @@ async fn test_dependency_data_flow() -> DagResult<()> {
.add_task(task_fn(|(s, p): (i32, i32)| async move { s + p }))
.depends_on((&sum, &product));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Note: sum and product are not sinks (final_result depends on them)
// Only final_result is a sink and can be retrieved
@@ -69,10 +64,7 @@ async fn test_dependencies_with_different_types() -> DagResult<()> {
}))
.depends_on((&int_source, &string_source, &bool_source));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(combined)?, "hello: 42 (true)");
Ok(())
@@ -93,10 +85,7 @@ async fn test_shared_dependencies() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify all dependents got the correct value from shared
for (i, handle) in dependents.iter().enumerate() {
@@ -115,10 +104,7 @@ async fn test_multiple_source_nodes() -> DagResult<()> {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i * 10 })))
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
for (i, source) in sources.iter().enumerate() {
assert_eq!(dag.get(source)?, i as i32 * 10);
tests/dependencies/tuples.rs
+8 -28
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
#[tokio::test]
async fn test_two_dependencies() -> DagResult<()> {
@@ -13,10 +14,7 @@ async fn test_two_dependencies() -> DagResult<()> {
.add_task(task_fn(|(x, y): (i32, i32)| async move { x + y }))
.depends_on((&a, &b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 30);
Ok(())
@@ -35,10 +33,7 @@ async fn test_three_dependencies() -> DagResult<()> {
))
.depends_on((&a, &b, &c));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 6);
Ok(())
@@ -58,10 +53,7 @@ async fn test_four_dependencies() -> DagResult<()> {
}))
.depends_on((&deps[0], &deps[1], &deps[2], &deps[3]));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 10); // 1+2+3+4
Ok(())
@@ -81,10 +73,7 @@ async fn test_five_dependencies() -> DagResult<()> {
))
.depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4]));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 15); // 1+2+3+4+5
Ok(())
@@ -106,10 +95,7 @@ async fn test_six_dependencies() -> DagResult<()> {
&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5]
));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 21); // 1+2+3+4+5+6
Ok(())
@@ -133,10 +119,7 @@ async fn test_seven_dependencies() -> DagResult<()> {
&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6],
));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 28); // 1+2+3+4+5+6+7
Ok(())
@@ -160,10 +143,7 @@ async fn test_eight_dependencies() -> DagResult<()> {
&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6], &deps[7],
));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 36); // Sum of 1..=8
Ok(())
tests/errors/error_types.rs
+18 -35
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagError, DagRunner};
use futures::FutureExt;
#[tokio::test]
async fn test_result_not_found_error() {
@@ -14,11 +15,9 @@ async fn test_result_not_found_error() {
assert!(matches!(result, Err(DagError::ResultNotFound { .. })));
// Run and verify it works after
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(&t1).unwrap(), 42);
}
@@ -34,11 +33,9 @@ async fn test_nested_result_error_handling() {
let task2 = dag.add_task(task_fn(|_: ()| async { Result::<i32, String>::Ok(42) }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Both tasks complete successfully (they return Results)
assert_eq!(dag.get(&task1).unwrap(), Err("Internal error".to_string()));
@@ -51,11 +48,9 @@ async fn test_option_none_not_error() {
let task = dag.add_task(task_fn(|_: ()| async { None::<i32> }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// None is a valid result, not an error
assert_eq!(dag.get(&task).unwrap(), None);
@@ -82,11 +77,9 @@ async fn test_custom_error_types() {
})
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Check custom errors are preserved
match dag.get(&task1).unwrap() {
@@ -113,11 +106,7 @@ async fn test_error_in_tuple_dependency() {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a + b }))
.depends_on((&ok_task, &err_task));
let result = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let result = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
assert!(result.is_err());
assert!(dag.get(dependent).is_err());
@@ -135,11 +124,7 @@ async fn test_string_panic_vs_structured_panic() {
panic!("Formatted panic: value={}, code={}", 42, "ERROR");
}));
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// Both should result in errors
assert!(dag.get(&string_panic).is_err());
@@ -159,11 +144,9 @@ async fn test_anyhow_like_error_chain() {
outer
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Get the chained error
let result = dag.get(&task).unwrap();
tests/errors/propagation.rs
+13 -40
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::DagRunner;
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -22,11 +23,7 @@ async fn test_linear_error_propagation() {
.add_task(task_fn(|x: i32| async move { x * 2 }))
.depends_on(t3);
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// t1 succeeds
assert_eq!(dag.get(&t1).unwrap(), 1);
@@ -65,11 +62,7 @@ async fn test_diamond_error_propagation() {
.add_task(task_fn(|(l, r): (i32, i32)| async move { l + r }))
.depends_on((&left, &right));
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
assert_eq!(dag.get(&source).unwrap(), 100);
assert_eq!(dag.get(left).unwrap(), 50);
@@ -119,11 +112,7 @@ async fn test_error_stops_at_boundary() {
}
}));
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// Error branch: source runs but dependent doesn't
assert!(dag.get(&error_source).is_err());
@@ -156,11 +145,7 @@ async fn test_multiple_error_sources_convergence() {
.add_task(task_fn(|(_, _, _): (i32, i32, i32)| async { 100 }))
.depends_on((&error1, &error2, &success));
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// All error sources fail
assert!(dag.get(&error1).is_err());
@@ -189,11 +174,7 @@ async fn test_error_in_wide_fanout() {
})
.collect();
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// Source fails
assert!(dag.get(&source).is_err());
@@ -234,11 +215,9 @@ async fn test_selective_error_propagation() {
}))
.depends_on(handler);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Source returns an Err variant (not a panic)
assert_eq!(dag.get(&source).unwrap(), Err("Source error".to_string()));
@@ -294,11 +273,7 @@ async fn test_error_propagation_timing() {
}))
.depends_on(&t2);
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
let order = propagation_order.lock().clone();
@@ -373,11 +348,9 @@ async fn test_partial_branch_failure_propagation() {
))
.depends_on((&a3, &b3, &c3));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.ok();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.ok();
// Branch A succeeds
assert_eq!(dag.get(a3).unwrap(), 106);
tests/errors/recovery.rs
+9 -35
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
@@ -23,11 +24,7 @@ async fn test_fallback_on_error() -> DagResult<()> {
let _selector = dag.add_task(task_fn(|_: ()| async {}));
// Run DAG
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// Primary should fail
assert!(dag.get(&primary).is_err());
@@ -63,10 +60,7 @@ async fn test_retry_pattern_simulation() -> DagResult<()> {
}
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Task completes with error on first "attempt"
assert_eq!(dag.get(&task)?, Err("Simulated failure"));
@@ -109,10 +103,7 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// First few tasks fail and open circuit
for task in &tasks[0..3] {
@@ -144,11 +135,7 @@ async fn test_partial_recovery_with_defaults() -> DagResult<()> {
vec![0, 10, 20, 0, 40, 50, 0, 70, 80, 0] // 0 for failed tasks
}));
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
let result = dag.get(&aggregator)?;
assert_eq!(result.len(), 10);
@@ -178,10 +165,7 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Check accumulated errors
let error_list = errors.lock().clone();
@@ -215,10 +199,7 @@ async fn test_graceful_degradation() -> DagResult<()> {
}))
.depends_on(&critical);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Critical path works
assert_eq!(dag.get(final_task)?, 15);
@@ -249,10 +230,7 @@ async fn test_error_boundary_isolation() -> DagResult<()> {
.add_task(task_fn(|x: i32| async move { x + 5 }))
.depends_on(b2);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Group A fails (returns Err)
assert_eq!(dag.get(a2)?, Err("Group A error"));
@@ -287,11 +265,7 @@ async fn test_compensation_action() -> DagResult<()> {
}
}));
let _ = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let _ = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
// Action failed
assert!(dag.get(&action).is_err());
tests/execution/basic.rs
+40 -39
View File
@@ -1,23 +1,24 @@
// Basic execution tests for DagRunner
use crate::common::task_fn;
use futures::FutureExt;
use dagx::*;
#[tokio::test]
async fn test_empty_dag() {
let dag = DagRunner::new();
dag.run()
.await
.unwrap(); // Should complete without error
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap(); // Should complete without error
}
#[tokio::test]
async fn test_single_task() {
let dag = DagRunner::new();
let task = dag.add_task(task_fn(|_: ()| async { 42 }));
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(task).unwrap(), 42);
}
@@ -39,9 +40,9 @@ async fn test_deep_chain() {
.add_task(task_fn(|x: i32| async move { x + 1 }))
.depends_on(d);
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: a, b, c, d are not sinks (they have dependents in the chain)
// Only e is a sink and can be retrieved
@@ -56,9 +57,9 @@ async fn test_wide_parallel() {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i * 2 })))
.collect();
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i * 2);
@@ -80,9 +81,9 @@ async fn test_diamond_dependency() {
.add_task(task_fn(|(x, y): (i32, i32)| async move { x + y }))
.depends_on((&b, &c));
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Note: a, b, and c are not sinks (they have dependents)
// Only d is a sink and can be retrieved
@@ -108,9 +109,9 @@ async fn test_different_output_types() {
}
}));
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(&dag.get(string_task).unwrap(), "hello");
assert_eq!(dag.get(vec_task).unwrap(), vec![1, 2, 3]);
@@ -141,9 +142,9 @@ async fn test_multiple_sinks() {
.add_task(task_fn(|x: i32| async move { x * 3 }))
.depends_on(branch2);
dag.run()
.await
.unwrap(); // Should wait for both sinks
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap(); // Should wait for both sinks
assert_eq!(dag.get(sink1).unwrap(), 21); // (10 * 2) + 1
assert_eq!(dag.get(sink2).unwrap(), 45); // (10 + 5) * 3
@@ -163,9 +164,9 @@ async fn test_single_task_inline_path() {
.add_task(task_fn(|x: i32| async move { x + 5 }))
.depends_on(t2);
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(t3).unwrap(), 25); // (10 * 2) + 5
}
@@ -188,9 +189,9 @@ async fn test_multi_task_spawn_path() {
.add_task(task_fn(|x: i32| async move { x + 3 }))
.depends_on(&source);
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(t1).unwrap(), 101);
assert_eq!(dag.get(t2).unwrap(), 102);
@@ -212,9 +213,9 @@ async fn test_producer_consumer_channels() {
))
.depends_on(&producer);
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(consumer).unwrap(), 15);
}
@@ -237,9 +238,9 @@ async fn test_multi_consumer_fanout() {
.add_task(task_fn(|x: i32| async move { x - 5 }))
.depends_on(&producer);
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(c1).unwrap(), 84);
assert_eq!(dag.get(c2).unwrap(), 52);
@@ -265,9 +266,9 @@ 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()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sink).unwrap(), 5); // (1+1) + (1+2) = 5
}
@@ -313,9 +314,9 @@ async fn test_layer_execution_order() {
}))
.depends_on(t2);
dag.run()
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let order = execution_order.lock().clone();
assert_eq!(order, vec![1, 2, 3]);
tests/execution/examples.rs
+13 -20
View File
@@ -3,6 +3,7 @@ use crate::common::task_fn;
use crate::common::tasks::*;
use dagx::*;
use futures::FutureExt;
#[tokio::test]
async fn test_example_a_fan_out() {
@@ -12,11 +13,9 @@ async fn test_example_a_fan_out() {
let plus1 = dag.add_task(Inc::new()).depends_on(&base); // 11
let plus2 = dag.add_task(Inc::new()).depends_on(&base); // 11
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap(); // SPEC API with spawner
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap(); // SPEC API with spawner
assert_eq!(dag.get(plus1).unwrap(), 11);
assert_eq!(dag.get(plus2).unwrap(), 11);
@@ -32,11 +31,9 @@ async fn test_example_b_fan_in() {
let output = dag.add_task(OutputResult::new()).depends_on((&s, &n, &f));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(&dag.get(output).unwrap(), "hi:42:true");
}
@@ -52,11 +49,9 @@ async fn test_example_c_many_to_many() {
let prod_yz = dag.add_task(Mul::new()).depends_on((&y, &z)); // 15
let total = dag.add_task(Add::new()).depends_on((&sum_xy, &prod_yz)); // 20
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(total).unwrap(), 20);
}
@@ -76,11 +71,9 @@ async fn test_single_input_forms() {
.add_task(task_fn(|s: String| async move { s.to_uppercase() }))
.depends_on((&s,));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(&dag.get(upper1).unwrap(), "HELLO");
assert_eq!(&dag.get(upper2).unwrap(), "HELLO");
tests/execution/integration.rs
+34 -27
View File
@@ -1,7 +1,8 @@
//! Integration tests for end-to-end functionality
use dagx::{DagResult, DagRunner, Task};
use crate::common::task_fn;
use dagx::{DagResult, DagRunner, Task};
use futures::FutureExt;
use std::time::Instant;
use tokio::time::{sleep, Duration};
@@ -18,7 +19,7 @@ impl LoadValue {
#[dagx::task]
impl LoadValue {
async fn run(&self) -> i32 {
self.value // Read-only state: use &self
self.value // Read-only state: use &self
}
}
@@ -33,7 +34,7 @@ impl Multiply {
#[dagx::task]
impl Multiply {
async fn run(&self, input: &i32) -> i32 {
input * self.factor // Read-only state: use &self
input * self.factor // Read-only state: use &self
}
}
@@ -41,7 +42,7 @@ struct Add;
#[dagx::task]
impl Add {
async fn run(a: &i32, b: &i32) -> i32 {
a + b // Stateless: no self needed
a + b // Stateless: no self needed
}
}
@@ -67,7 +68,7 @@ async fn integration_test_etl_pipeline() -> DagResult<()> {
let final_result = dag.add_task(Add).depends_on((&combine12, &transform3));
// Execute pipeline
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify
let result = dag.get(final_result)?;
@@ -97,7 +98,7 @@ async fn integration_test_multi_layer_dag() -> DagResult<()> {
// Layer 4: Final result
let final_result = dag.add_task(Add).depends_on((&double_ab, &double_bc));
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(final_result)?;
// (1+2)*2 + (2+3)*2 = 6 + 10 = 16
@@ -118,7 +119,7 @@ async fn integration_test_multiple_sinks() -> DagResult<()> {
let branch2 = dag.add_task(Multiply::new(3)).depends_on(&source);
let branch3 = dag.add_task(Multiply::new(4)).depends_on(&source);
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// All branches should complete
assert_eq!(dag.get(branch1)?, 20);
@@ -144,7 +145,7 @@ async fn integration_test_diamond_cascade() -> DagResult<()> {
let right2 = dag.add_task(Multiply::new(3)).depends_on(sink1);
let sink2 = dag.add_task(Add).depends_on((&left2, &right2));
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(sink2)?;
// Diamond 1: (10*2) + (10*3) = 50
@@ -175,7 +176,7 @@ async fn integration_test_wide_and_deep() -> DagResult<()> {
final_nodes.push(step3);
}
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify each chain
for (i, node) in final_nodes.iter().enumerate() {
@@ -194,7 +195,7 @@ async fn integration_test_error_recovery() -> DagResult<()> {
let source = dag.add_task(LoadValue::new(42));
let transform = dag.add_task(Multiply::new(2)).depends_on(&source);
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Should be able to access results
assert_eq!(dag.get(transform)?, 84);
@@ -230,7 +231,7 @@ async fn integration_test_large_scale() -> DagResult<()> {
})
.collect();
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify some results
assert_eq!(dag.get(aggregations[0])?, 1); // First two
@@ -254,7 +255,7 @@ async fn integration_test_reused_values() -> DagResult<()> {
// Combine some of the results
let combined = dag.add_task(Add).depends_on((&double, &triple));
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(double)?, 200);
assert_eq!(dag.get(triple)?, 300);
@@ -283,7 +284,7 @@ async fn integration_test_parallel_execution_timing() -> DagResult<()> {
.collect();
let start = Instant::now();
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let elapsed = start.elapsed();
// Verify all tasks completed
@@ -322,23 +323,29 @@ async fn integration_test_parallel_vs_sequential_timing() -> DagResult<()> {
// Layer 2: Sequential chain of 3 tasks depending on the first parallel task
// (should add ~150ms)
let seq1 = dag.add_task(task_fn(move |x: i32| async move {
sleep(sleep_duration).await;
x + 1
})).depends_on(&parallel_tasks[0]);
let seq1 = dag
.add_task(task_fn(move |x: i32| async move {
sleep(sleep_duration).await;
x + 1
}))
.depends_on(&parallel_tasks[0]);
let seq2 = dag.add_task(task_fn(move |x: i32| async move {
sleep(sleep_duration).await;
x + 1
})).depends_on(seq1);
let seq2 = dag
.add_task(task_fn(move |x: i32| async move {
sleep(sleep_duration).await;
x + 1
}))
.depends_on(seq1);
let seq3 = dag.add_task(task_fn(move |x: i32| async move {
sleep(sleep_duration).await;
x + 1
})).depends_on(seq2);
let seq3 = dag
.add_task(task_fn(move |x: i32| async move {
sleep(sleep_duration).await;
x + 1
}))
.depends_on(seq2);
let start = Instant::now();
dag.run(|fut| { tokio::spawn(fut); }).await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let elapsed = start.elapsed();
// Verify results
@@ -361,4 +368,4 @@ async fn integration_test_parallel_vs_sequential_timing() -> DagResult<()> {
);
Ok(())
}
}
tests/execution/results.rs
+13 -25
View File
@@ -3,6 +3,7 @@ use crate::common::task_fn;
use crate::common::tasks::Counter;
use dagx::*;
use futures::FutureExt;
#[tokio::test]
async fn test_result_type() {
@@ -13,11 +14,9 @@ async fn test_result_type() {
Err::<i32, String>("error".to_string())
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(success_task).unwrap(), Ok(42));
assert_eq!(dag.get(error_task).unwrap(), Err("error".to_string()));
@@ -30,10 +29,7 @@ async fn test_dag_result_type_works() {
// Test that DagResult<()> works with ? operator pattern
async fn run_dag(dag: &DagRunner) -> DagResult<()> {
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
Ok(())
}
@@ -68,7 +64,7 @@ async fn test_error_messages_are_helpful() {
let err_msg = format!("{}", result.unwrap_err());
assert!(err_msg.contains("Result not found"));
assert!(err_msg.contains("task #0")); // Task ID
assert!(err_msg.contains("Call dag.run()"));
assert!(err_msg.contains("Call dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))"));
}
#[tokio::test]
@@ -77,11 +73,7 @@ async fn test_successful_execution_returns_ok() {
let task = dag.add_task(task_fn(|_: ()| async { 42 }));
// Should return Ok(())
let result = dag
.run(|fut| {
tokio::spawn(fut);
})
.await;
let result = dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await;
assert!(result.is_ok());
// Should be able to get value
@@ -95,11 +87,9 @@ async fn test_stateful_tasks() {
let input = dag.add_task(task_fn(|_: ()| async { 5 }));
let counter = dag.add_task(Counter::new(10)).depends_on(&input);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(counter).unwrap(), 15); // 10 + 5
}
@@ -117,11 +107,9 @@ async fn test_task_fn_with_captured_state() {
.add_task(task_fn(move |x: i32| async move { x * multiplier }))
.depends_on(&base);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(scaled).unwrap(), 35);
}
tests/parallelism/concurrent.rs
+32 -47
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::DagRunner;
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
@@ -14,7 +15,12 @@ async fn test_concurrent_run_calls() {
// Add some tasks
let tasks: Vec<_> = (0..10)
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i })))
.map(|i| {
dag.add_task(task_fn(move |_: ()| async move {
tokio::time::sleep(Duration::from_millis(5)).await;
i
}))
})
.collect();
// Try to run the DAG multiple times concurrently
@@ -22,12 +28,7 @@ async fn test_concurrent_run_calls() {
for _ in 0..5 {
let dag = Arc::clone(&dag);
handles.spawn(async move {
dag.run(|fut| {
tokio::spawn(fut);
})
.await
});
handles.spawn(async move { dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await });
}
// Only one run should succeed, others should fail with concurrent execution error
@@ -62,11 +63,9 @@ async fn test_concurrent_execution_and_building() {
.collect();
// Execute the initial DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Now try to add more tasks while also reading results
let mut handles = JoinSet::new();
@@ -93,11 +92,9 @@ async fn test_concurrent_execution_and_building() {
}
// Execute again with the new tasks
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
}
#[tokio::test]
@@ -139,11 +136,9 @@ async fn test_dag_builder_thread_safety() {
assert_eq!(counter.load(Ordering::SeqCst), 10);
// Execute the combined DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Verify each pipeline computed correctly
for task in final_tasks.iter() {
@@ -163,11 +158,9 @@ async fn test_concurrent_get_operations() {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i })))
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Now access results concurrently
let mut handles = JoinSet::new();
@@ -217,11 +210,9 @@ async fn test_no_race_in_dependency_tracking() {
}
// Execute
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// All dependents should have correct values
for dep in dependents.iter() {
@@ -277,11 +268,9 @@ async fn test_high_concurrency_stress() {
assert_eq!(completed.load(Ordering::SeqCst), 1000);
// The DAG should still be executable
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
}
#[tokio::test]
@@ -369,11 +358,9 @@ async fn test_atomic_dag_operations() {
assert_eq!(deps.len(), 50);
// Execute and verify
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
for dep in deps.iter() {
let result = dag.get(dep).unwrap();
@@ -424,9 +411,7 @@ async fn test_no_deadlock_on_circular_waiting() {
assert!(result.is_ok(), "Deadlock detected - operation timed out");
// Should be able to execute the DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
}
tests/parallelism/execution.rs
+5 -16
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::time::sleep;
@@ -62,10 +63,7 @@ async fn test_layers_execute_in_parallel() -> DagResult<()> {
.depends_on((&layer1[0], &layer1[1], &layer1[2]))
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(final_task)?, 33); // (0+1) + (10+1) + (20+1)
@@ -185,10 +183,7 @@ async fn test_completion_order_matches_dependencies() -> DagResult<()> {
.depends_on(c)
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(d)?, 6); // (1 + 2) * 2
@@ -271,10 +266,7 @@ async fn test_dependent_tasks_respect_ordering() -> DagResult<()> {
.depends_on(c)
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify results
assert_eq!(dag.get(a)?, 1);
@@ -344,10 +336,7 @@ async fn test_independent_tasks_dont_wait() -> DagResult<()> {
}))
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Check results
assert_eq!(dag.get(fast1)?, "fast1");
tests/parallelism/proof.rs
+2 -4
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::time::{Duration, Instant};
use tokio::time::sleep;
@@ -24,10 +25,7 @@ async fn test_parallelism_proof_1000_tasks() -> DagResult<()> {
.collect();
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let elapsed = start.elapsed();
// Verify all tasks completed
tests/parallelism/spawning.rs
+3 -5
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::sync::Mutex;
@@ -31,7 +32,7 @@ async fn test_spawner_actually_spawns_tasks() -> DagResult<()> {
.collect();
// Custom spawner that counts invocations
dag.run().await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify all tasks were spawned
for (i, task) in tasks.iter().enumerate() {
@@ -77,10 +78,7 @@ async fn test_tasks_not_running_inline() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify results
for (i, task) in tasks.iter().enumerate() {
tests/parallelism/threads.rs
+7 -24
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
@@ -40,10 +41,7 @@ async fn test_tasks_run_on_different_threads() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify all tasks completed
for task in tasks.iter() {
@@ -112,10 +110,7 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify all tasks completed
for (i, task) in tasks.iter().enumerate() {
@@ -186,10 +181,7 @@ async fn test_diamond_parallel_execution() -> DagResult<()> {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a + b }))
.depends_on((&parallel1, &parallel2));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sink)?, 500); // (100 * 2) + (100 * 3)
@@ -276,10 +268,7 @@ async fn test_parallel_branches_with_dependencies() -> DagResult<()> {
.depends_on(b2_1)
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify results
assert_eq!(dag.get(b1_2)?, 21); // (10 * 2) + 1
@@ -371,10 +360,7 @@ async fn test_wide_fanout_parallel_execution() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify results
for (i, task) in dependents.iter().enumerate() {
@@ -445,10 +431,7 @@ async fn test_massive_parallel_fanout() -> DagResult<()> {
.collect();
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let elapsed = start.elapsed();
// All should complete
tests/parallelism/timing.rs
+3 -8
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner, Task};
use futures::FutureExt;
use std::sync::Arc;
use std::sync::Mutex;
use std::time::{Duration, Instant};
@@ -26,10 +27,7 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
.collect();
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let elapsed = start.elapsed();
// Verify all tasks completed
@@ -113,10 +111,7 @@ async fn test_sync_tasks_dont_block_async_runtime() -> DagResult<()> {
}))
};
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify results
assert_eq!(dag.get(sync1)?, 20);
tests/patterns/butterfly.rs
+5 -16
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -44,10 +45,7 @@ async fn test_butterfly_network() -> DagResult<()> {
.depends_on((&layer1[3], &layer1[7])),
];
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify some outputs exist
for task in &layer2 {
@@ -100,10 +98,7 @@ async fn test_recursive_butterfly() -> DagResult<()> {
let outputs = build_butterfly(&dag, inputs);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(outputs.len(), 4);
for output in outputs {
@@ -143,10 +138,7 @@ async fn test_benes_network() -> DagResult<()> {
.depends_on((&stage1[1], &stage1[3])),
];
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
for output in outputs {
let _ = dag.get(output)?;
@@ -209,10 +201,7 @@ async fn test_shuffle_exchange_network() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(counter.load(Ordering::Relaxed), 8);
tests/patterns/complex.rs
+6 -20
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -59,10 +60,7 @@ async fn test_workflow_with_conditional_paths() -> DagResult<()> {
}))
.depends_on(combine);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(process)?, 84);
let result = dag.get(final_result)?;
@@ -140,10 +138,7 @@ async fn test_map_reduce_pattern() -> DagResult<()> {
))
.depends_on((&reducers[0], &reducers[1], &reducers[2], &reducers[3]));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(final_task)?;
assert!(result > 0);
@@ -200,10 +195,7 @@ async fn test_pipeline_with_feedback() -> DagResult<()> {
}))
.depends_on(stage3);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(stage1)?, 2);
assert_eq!(dag.get(stage3)?, 12);
@@ -247,10 +239,7 @@ async fn test_scatter_gather_pattern() -> DagResult<()> {
))
.depends_on((&workers[0], &workers[1], &workers[2], &workers[3]));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Sum of 10+20+30+40+50+60+70+80 = 360
assert_eq!(dag.get(gather)?, 360);
@@ -339,10 +328,7 @@ async fn test_fork_join_with_barriers() -> DagResult<()> {
}))
.depends_on(barrier);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let _ = dag.get(join)?;
assert_eq!(phase_counter.load(Ordering::SeqCst), 2);
tests/patterns/grid.rs
+5 -16
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
#[tokio::test]
async fn test_2d_grid_pattern() -> DagResult<()> {
@@ -33,10 +34,7 @@ async fn test_2d_grid_pattern() -> DagResult<()> {
}
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Check corner values
let bottom_right = dag.get(grid[3][3].as_ref().unwrap())?;
@@ -99,10 +97,7 @@ async fn test_3d_grid_pattern() -> DagResult<()> {
}
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Check that all cells have values
for row in &grid {
@@ -178,10 +173,7 @@ async fn test_hexagonal_grid() -> DagResult<()> {
hex_grid.insert((q, r), task);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify center and some ring values
assert_eq!(dag.get(hex_grid.get(&(0, 0)).unwrap())?, 100);
@@ -230,10 +222,7 @@ async fn test_toroidal_grid() -> DagResult<()> {
}
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify all torus tasks completed
for task in torus_tasks {
tests/patterns/tree.rs
+7 -24
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
#[tokio::test]
async fn test_binary_tree_reduction() -> DagResult<()> {
@@ -35,10 +36,7 @@ async fn test_binary_tree_reduction() -> DagResult<()> {
let root = current_level[0];
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Sum of powers of 2 from 0 to 7: 1+2+4+8+16+32+64+128 = 255
assert_eq!(dag.get(root)?, 255);
@@ -69,10 +67,7 @@ async fn test_n_ary_tree() -> DagResult<()> {
let root = build_tree(&dag, 3, 1);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(root)?;
assert!(result > 0);
@@ -119,10 +114,7 @@ async fn test_unbalanced_tree() -> DagResult<()> {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a * b }))
.depends_on((&a, &b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(&f)?, 1);
assert_eq!(dag.get(c)?, 2);
@@ -170,10 +162,7 @@ async fn test_trie_like_structure() -> DagResult<()> {
.add_task(task_fn(|v: String| async move { format!("{}/log", v) }))
.depends_on(var);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(usr_bin)?, "root/usr/bin");
assert_eq!(dag.get(usr_lib)?, "root/usr/lib");
@@ -230,10 +219,7 @@ async fn test_balanced_k_ary_tree() -> DagResult<()> {
let root = create_balanced_tree(&dag, DEPTH, K, 0);
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(root)?;
assert!(result > 0);
@@ -280,10 +266,7 @@ async fn test_merkle_tree_pattern() -> DagResult<()> {
let root_hash = current_level[0];
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
let result = dag.get(root_hash)?;
assert!(result.starts_with("hash"));
tests/performance/memory.rs
+4 -5
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::DagRunner;
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -23,11 +24,9 @@ async fn test_memory_usage_10000_nodes() {
println!("Memory for 10k nodes: ~{} bytes", build_memory);
// Execute
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let after_exec = memory_usage_hint();
let exec_memory = after_exec.saturating_sub(after_build);
tests/performance/stress.rs
+7 -10
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::DagRunner;
use futures::FutureExt;
use std::sync::Arc;
use std::time::{Duration, Instant};
@@ -26,11 +27,9 @@ async fn test_deep_chain_100_levels() {
// Execute
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let exec_time = start.elapsed();
println!("Executed 100-level chain in {:?}", exec_time);
@@ -89,11 +88,9 @@ async fn test_stress_mixed_patterns() {
// Execute
let start = Instant::now();
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let exec_time = start.elapsed();
println!("Executed mixed pattern stress test in {:?}", exec_time);
tests/runtimes/async_std.rs
+5 -5
View File
@@ -1,7 +1,7 @@
//! Runtime compatibility tests for async-std
use dagx::{DagRunner, Task};
use crate::common::task_fn;
use dagx::{DagRunner, Task};
struct Value(i32);
#[dagx::task]
@@ -27,7 +27,7 @@ async fn test_basic_dag_async_std() {
let y = dag.add_task(Value(3));
let sum = dag.add_task(Add).depends_on((&x, &y));
dag.run(|fut| { async_std::task::spawn(fut); }).await.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap();
assert_eq!(dag.get(sum).unwrap(), 5);
}
@@ -40,7 +40,7 @@ async fn test_parallel_execution_async_std() {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i * 2 })))
.collect();
dag.run(|fut| { async_std::task::spawn(fut); }).await.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap();
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i * 2);
@@ -58,7 +58,7 @@ async fn test_complex_dependencies_async_std() {
.add_task(task_fn(|x: i32| async move { x * 2 }))
.depends_on(sum);
dag.run(|fut| { async_std::task::spawn(fut); }).await.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await.unwrap();
assert_eq!(dag.get(double).unwrap(), 60);
}
}
tests/runtimes/compatibility.rs
+2 -4
View File
@@ -1,6 +1,7 @@
//! Runtime integration tests - verify dagx works correctly across different runtimes
use dagx::{DagResult, DagRunner, Task};
use futures::FutureExt;
struct Value(i32);
#[dagx::task]
@@ -26,10 +27,7 @@ async fn runtime_integration_tokio() -> DagResult<()> {
let b = dag.add_task(Value(20));
let sum = dag.add_task(Add).depends_on((&a, &b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(sum)?, 30);
Ok(())
tests/runtimes/quirky.rs
+52 -100
View File
@@ -15,6 +15,8 @@
use crate::common::task_fn;
use dagx::DagRunner;
use futures::task::SpawnExt;
use futures_executor::ThreadPool;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -36,11 +38,7 @@ fn test_async_executor_basic() {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a + b }))
.depends_on((&x, &y));
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(sum).unwrap(), 30);
};
@@ -70,11 +68,7 @@ fn test_async_executor_complex_dag() {
.add_task(task_fn(|(l, r): (i32, i32)| async move { l + r }))
.depends_on((&left, &right));
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(sink).unwrap(), 25); // (5*2) + (5*3) = 25
};
@@ -104,11 +98,7 @@ fn test_async_executor_parallel_tasks() {
})
.collect();
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
// All tasks should have run
assert_eq!(counter.load(Ordering::SeqCst), 10);
@@ -132,11 +122,7 @@ fn test_async_executor_with_local_spawn() {
let x = dag.add_task(task_fn(|_: ()| async { 42 }));
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(x).unwrap(), 42);
};
@@ -163,11 +149,7 @@ fn test_async_executor_deep_chain() {
.depends_on(current);
}
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(current).unwrap(), 50);
};
@@ -190,13 +172,9 @@ fn test_pollster_basic() {
.add_task(task_fn(|x: i32| async move { x * 2 }))
.depends_on(&x);
dag.run(|fut| {
// pollster blocks, so we need to spawn on a different executor
// Use futures::executor for spawning
futures_executor::block_on(fut);
})
.await
.unwrap();
dag.run(|fut| async { pollster::block_on(fut) })
.await
.unwrap();
assert_eq!(dag.get(doubled).unwrap(), 200);
});
@@ -217,11 +195,9 @@ fn test_pollster_complex() {
})
.collect();
dag.run(|fut| {
futures_executor::block_on(fut);
})
.await
.unwrap();
dag.run(|fut| async { pollster::block_on(fut) })
.await
.unwrap();
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), 7 * i as i32);
@@ -237,11 +213,9 @@ fn test_pollster_error_handling() {
// Task that returns Result
let task = dag.add_task(task_fn(|_: ()| async { Ok::<i32, String>(42) }));
dag.run(|fut| {
futures_executor::block_on(fut);
})
.await
.unwrap();
dag.run(|fut| async { pollster::block_on(fut) })
.await
.unwrap();
assert_eq!(dag.get(task).unwrap(), Ok(42));
});
@@ -262,13 +236,11 @@ fn test_futures_executor_basic() {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a * b }))
.depends_on((&x, &y));
dag.run(|fut| {
// Use ThreadPool for spawning
let pool = futures_executor::ThreadPool::new().unwrap();
pool.spawn_ok(fut);
})
.await
.unwrap();
let spawner = ThreadPool::new().unwrap();
dag.run(|fut| spawner.spawn_with_handle(fut).unwrap())
.await
.unwrap();
assert_eq!(dag.get(product).unwrap(), 375); // 15 * 25
});
@@ -276,8 +248,8 @@ fn test_futures_executor_basic() {
#[test]
fn test_futures_executor_threadpool() {
let pool = std::sync::Arc::new(futures_executor::ThreadPool::new().unwrap());
let pool_clone = pool.clone();
let pool = futures_executor::ThreadPool::new().unwrap();
let spawner = pool.clone();
pool.spawn_ok(async move {
let dag = DagRunner::new();
@@ -287,11 +259,9 @@ fn test_futures_executor_threadpool() {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i * i })))
.collect();
dag.run(|fut| {
pool_clone.spawn_ok(fut);
})
.await
.unwrap();
dag.run(|fut| spawner.spawn_with_handle(fut).unwrap())
.await
.unwrap();
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), (i * i) as i32);
@@ -325,12 +295,11 @@ fn test_futures_executor_diamond_pattern() {
.add_task(task_fn(|(x, y): (i32, i32)| async move { x * y }))
.depends_on((&b, &c));
let pool = futures_executor::ThreadPool::new().unwrap();
dag.run(|fut| {
pool.spawn_ok(fut);
})
.await
.unwrap();
let spawner = ThreadPool::new().unwrap();
dag.run(|fut| spawner.spawn_with_handle(fut).unwrap())
.await
.unwrap();
assert_eq!(dag.get(d).unwrap(), 60); // (8+2) * (8-2) = 10 * 6
});
@@ -350,7 +319,7 @@ fn test_all_runtimes_compatibility() {
executor.run(async {
let dag = DagRunner::new();
let x = dag.add_task(task_fn(|_: ()| async { 123 }));
dag.run(|fut| executor.spawn(fut).detach()).await.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(x).unwrap(), 123);
});
}
@@ -360,7 +329,9 @@ fn test_all_runtimes_compatibility() {
pollster::block_on(async {
let dag = DagRunner::new();
let x = dag.add_task(task_fn(|_: ()| async { 123 }));
dag.run(futures_executor::block_on).await.unwrap();
dag.run(|fut| async { pollster::block_on(fut) })
.await
.unwrap();
assert_eq!(dag.get(x).unwrap(), 123);
});
}
@@ -370,8 +341,10 @@ fn test_all_runtimes_compatibility() {
futures_executor::block_on(async {
let dag = DagRunner::new();
let x = dag.add_task(task_fn(|_: ()| async { 123 }));
let pool = futures_executor::ThreadPool::new().unwrap();
dag.run(|fut| pool.spawn_ok(fut)).await.unwrap();
let spawner = ThreadPool::new().unwrap();
dag.run(|fut| spawner.spawn_with_handle(fut).unwrap())
.await
.unwrap();
std::thread::sleep(std::time::Duration::from_millis(10));
assert_eq!(dag.get(x).unwrap(), 123);
});
@@ -393,11 +366,7 @@ fn test_runtime_edge_cases() {
42
}));
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(x).unwrap(), 42);
});
@@ -424,11 +393,7 @@ fn test_async_executor_stress() {
})
.collect();
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i as i32);
@@ -443,13 +408,9 @@ fn test_pollster_nested_blocks() {
let dag1 = DagRunner::new();
let x = dag1.add_task(task_fn(|_: ()| async { 10 }));
dag1.run(|fut| {
pollster::block_on(async {
futures_executor::block_on(fut);
});
})
.await
.unwrap();
dag1.run(|fut| async { pollster::block_on(fut) })
.await
.unwrap();
assert_eq!(dag1.get(x).unwrap(), 10);
});
@@ -459,19 +420,16 @@ fn test_pollster_nested_blocks() {
fn test_futures_executor_multi_pool() {
// Test with multiple thread pools
let pool1 = futures_executor::ThreadPool::new().unwrap();
let pool2 = futures_executor::ThreadPool::new().unwrap();
let spawner = pool1.clone();
pool1.spawn_ok(async move {
let dag = DagRunner::new();
let x = dag.add_task(task_fn(|_: ()| async { 50 }));
dag.run(|fut| {
// Spawn on different pool
pool2.spawn_ok(fut);
})
.await
.unwrap();
dag.run(|fut| spawner.spawn_with_handle(fut).unwrap())
.await
.unwrap();
std::thread::sleep(std::time::Duration::from_millis(50));
assert_eq!(dag.get(x).unwrap(), 50);
@@ -507,11 +465,7 @@ fn test_sync_tasks_on_async_executor() {
let y = dag.add_task(task_fn(|_: ()| async { 7 }));
let sum = dag.add_task(SyncAdd).depends_on((&x, &y));
dag.run(|fut| {
executor.spawn(fut).detach();
})
.await
.unwrap();
dag.run(|fut| executor.spawn(fut)).await.unwrap();
assert_eq!(dag.get(sum).unwrap(), 12);
});
@@ -536,11 +490,9 @@ fn test_sync_tasks_on_pollster() {
let x = dag.add_task(task_fn(|_: ()| async { 21 }));
let doubled = dag.add_task(SyncDouble).depends_on(&x);
dag.run(|fut| {
futures_executor::block_on(fut);
})
.await
.unwrap();
dag.run(|fut| async { pollster::block_on(fut) })
.await
.unwrap();
assert_eq!(dag.get(doubled).unwrap(), 42);
});
tests/runtimes/smol.rs
+11 -5
View File
@@ -1,7 +1,7 @@
//! Runtime compatibility tests for smol
use dagx::{DagRunner, Task};
use crate::common::task_fn;
use dagx::{DagRunner, Task};
struct Value(i32);
#[dagx::task]
@@ -28,7 +28,9 @@ fn test_basic_dag_smol() {
let y = dag.add_task(Value(3));
let sum = dag.add_task(Add).depends_on((&x, &y));
dag.run(|fut| { smol::spawn(fut).detach(); }).await.unwrap();
dag.run(|fut| smol::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 5);
});
@@ -43,7 +45,9 @@ fn test_parallel_execution_smol() {
.map(|i| dag.add_task(task_fn(move |_: ()| async move { i * 2 })))
.collect();
dag.run(|fut| { smol::spawn(fut).detach(); }).await.unwrap();
dag.run(|fut| smol::spawn(fut).map(Result::unwrap))
.await
.unwrap();
for (i, task) in tasks.iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i * 2);
@@ -63,8 +67,10 @@ fn test_complex_dependencies_smol() {
.add_task(task_fn(|x: i32| async move { x * 2 }))
.depends_on(sum);
dag.run(|fut| { smol::spawn(fut).detach(); }).await.unwrap();
dag.run(|fut| smol::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(double).unwrap(), 60);
});
}
}
tests/task_patterns/combined.rs
+10 -15
View File
@@ -3,6 +3,7 @@ use crate::common::task_fn;
// Comprehensive test showing all three patterns working together
use dagx::*;
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
@@ -52,11 +53,9 @@ async fn test_all_three_task_patterns() {
// Pattern 3: Mutable accumulation
let accumulated = dag.add_task(MutableAccumulator(10)).depends_on(doubled); // 10 + 16 = 26
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 8);
assert_eq!(dag.get(doubled).unwrap(), 16);
@@ -97,11 +96,9 @@ async fn test_parallel_execution() {
}
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// All three tasks should have run
assert_eq!(counter.load(Ordering::SeqCst), 3);
@@ -120,11 +117,9 @@ async fn test_task_fn_with_captured_state() {
.add_task(task_fn(move |x: i32| async move { x * multiplier }))
.depends_on(&base);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(scaled).unwrap(), 35);
}
tests/task_patterns/readonly.rs
+10 -15
View File
@@ -3,6 +3,7 @@ use crate::common::task_fn;
// These tests verify the read-only state pattern using &self
use dagx::*;
use futures::FutureExt;
// Read-only state tasks
struct ReadOnlyValue(i32);
@@ -42,11 +43,9 @@ async fn test_readonly_state_basic() {
let value = dag.add_task(ReadOnlyValue(42));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(value).unwrap(), 42);
}
@@ -58,11 +57,9 @@ async fn test_readonly_state_with_input() {
let input = dag.add_task(task_fn(|_: ()| async { 5 }));
let result = dag.add_task(ReadOnlyMultiplier(7)).depends_on(&input);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(result).unwrap(), 35);
}
@@ -87,11 +84,9 @@ async fn test_readonly_state_multiple_uses() {
};
let result2 = dag.add_task(config2).depends_on(&input2);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(result1).unwrap(), 35); // (10 * 3) + 5
assert_eq!(dag.get(result2).unwrap(), 65); // (20 * 3) + 5
}
tests/task_patterns/stateless.rs
+25 -40
View File
@@ -1,5 +1,6 @@
// Stateless task pattern tests
use crate::common::task_fn;
use futures::FutureExt;
use dagx::*;
@@ -85,11 +86,9 @@ async fn test_stateless_basic() {
let y = dag.add_task(task_fn(|_: ()| async { 20 }));
let sum = dag.add_task(StatelessAdd).depends_on((&x, &y));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 30);
}
@@ -101,11 +100,9 @@ async fn test_stateless_single_input() {
let input = dag.add_task(task_fn(|_: ()| async { 21 }));
let doubled = dag.add_task(StatelessDouble).depends_on(&input);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(doubled).unwrap(), 42);
}
@@ -116,11 +113,9 @@ async fn test_stateless_no_input() {
let source = dag.add_task(StatelessSource);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(source).unwrap(), 42);
}
@@ -137,11 +132,9 @@ async fn test_stateless_multiple_inputs() {
.add_task(StatelessFormat)
.depends_on((&num, &text, &flag));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(result).unwrap(), "test: 42 (true)");
}
@@ -178,11 +171,9 @@ async fn test_stateless_mixed_with_stateful() {
// Stateful counter
let counted = dag.add_task(Counter::new(10)).depends_on(product);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(product).unwrap(), 15);
assert_eq!(dag.get(counted).unwrap(), 25); // 10 + 15
@@ -198,11 +189,9 @@ async fn test_stateless_chain() {
let step2 = dag.add_task(StatelessInc).depends_on(step1);
let step3 = dag.add_task(StatelessInc).depends_on(step2);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(step1).unwrap(), 1);
assert_eq!(dag.get(step2).unwrap(), 2);
@@ -221,11 +210,9 @@ async fn test_stateless_parallel() {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(tasks[0]).unwrap(), 1);
assert_eq!(dag.get(tasks[1]).unwrap(), 4);
@@ -244,11 +231,9 @@ async fn test_stateless_with_string_types() {
let greeting = dag.add_task(StatelessConcat).depends_on((&hello, &world));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(greeting).unwrap(), "Hello, World!");
}
tests/timing/concurrent_access.rs
+8 -28
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::RwLock;
@@ -23,10 +24,7 @@ async fn test_concurrent_result_retrieval() -> DagResult<()> {
.collect();
// Execute the DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Now try to get results concurrently from multiple threads
let mut get_handles = Vec::new();
@@ -77,10 +75,7 @@ async fn test_simultaneous_dag_building_and_execution() -> DagResult<()> {
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
// Execute the DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Get the additional tasks
let additional_tasks = build_handle.await.unwrap();
@@ -94,10 +89,7 @@ async fn test_simultaneous_dag_building_and_execution() -> DagResult<()> {
// (they need another run cycle)
if building_complete.load(Ordering::SeqCst) {
// Run again for the new tasks
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Now they should have results
for (i, task) in additional_tasks.iter().enumerate() {
@@ -141,10 +133,7 @@ async fn test_concurrent_dependency_resolution() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Check all completed
assert_eq!(resolution_counter.load(Ordering::SeqCst), 101);
@@ -198,10 +187,7 @@ async fn test_read_write_race_patterns() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Verify readers saw their writer's data
for reader in &readers {
@@ -257,10 +243,7 @@ async fn test_atomic_counter_races() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Final counter should be 200 * 100 = 20000
assert_eq!(counter.load(Ordering::SeqCst), 20000);
@@ -296,10 +279,7 @@ async fn test_multiple_dag_runners_interaction() -> DagResult<()> {
})
.collect();
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Collect results
let mut results = Vec::new();
tests/timing/interleaving.rs
+5 -16
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use futures::FutureExt;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::Barrier;
@@ -66,10 +67,7 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
))
.depends_on((&t1, &t2, &t3));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(collector)?, 6);
@@ -168,10 +166,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
.depends_on(chain_b);
}
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Check alternating pattern
let event_list = events.lock().clone();
@@ -239,10 +234,7 @@ async fn test_layered_interleaving() -> DagResult<()> {
}))
.depends_on((&layer2[0], &layer2[1]));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
// Sum of 0+1+2+3 = 6
assert_eq!(dag.get(layer3)?, 6);
@@ -335,10 +327,7 @@ async fn test_cross_branch_synchronization() -> DagResult<()> {
.add_task(task_fn(|(a, b): (i32, i32)| async move { a + b }))
.depends_on((&a2, &b2));
dag.run(|fut| {
tokio::spawn(fut);
})
.await?;
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap)).await?;
assert_eq!(dag.get(merge)?, 13); // 2 + 11
tests/tracing/with_tracing.rs
+12 -20
View File
@@ -44,11 +44,9 @@ async fn test_tracing_with_subscriber() {
let b = dag.add_task(Value(3));
let sum = dag.add_task(Add).depends_on((&a, &b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 5);
}
@@ -69,11 +67,9 @@ async fn test_tracing_with_complex_dag() {
let right = dag.add_task(Multiply).depends_on((&source, &source));
let sink = dag.add_task(Add).depends_on((&left, &right));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// source = 10
// left = 10 + 10 = 20
@@ -97,11 +93,9 @@ async fn test_tracing_inline_execution() {
let b = dag.add_task(Add).depends_on((&a, &a));
let c = dag.add_task(Multiply).depends_on((&b, &b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// a = 1
// b = 1 + 1 = 2
@@ -133,11 +127,9 @@ async fn test_tracing_multiple_layers() {
// Layer 2
let final_sum = dag.add_task(Add).depends_on((&sum12, &mul23));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// v1 = 1, v2 = 2, v3 = 3
// sum12 = 1 + 2 = 3
tests/tracing/without_tracing.rs
+7 -10
View File
@@ -1,6 +1,7 @@
//! Tests that the library works correctly without tracing feature
use dagx::{task, DagRunner, Task};
use futures::FutureExt;
struct Value(i32);
@@ -28,11 +29,9 @@ async fn test_basic_dag_without_tracing() {
let b = dag.add_task(Value(3));
let sum = dag.add_task(Add).depends_on((&a, &b));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 5);
}
@@ -51,11 +50,9 @@ async fn test_complex_dag_without_tracing() {
let sum34 = dag.add_task(Add).depends_on((&v3, &v4));
let final_sum = dag.add_task(Add).depends_on((&sum12, &sum34));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(final_sum).unwrap(), 10);
}
tests/type_safety/complex.rs
+37 -60
View File
@@ -2,6 +2,7 @@
use crate::common::task_fn;
use dagx::*;
use futures::FutureExt;
#[tokio::test]
async fn test_complex_type_safety() {
@@ -21,11 +22,9 @@ async fn test_complex_type_safety() {
}
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Correct type works
let custom = dag.get(&custom_task).unwrap();
@@ -47,11 +46,9 @@ async fn test_generic_type_safety() {
vec!["a".to_string(), "b".to_string()]
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Correct types work
assert_eq!(dag.get(&vec_i32).unwrap(), vec![1, 2, 3]);
@@ -69,11 +66,9 @@ async fn test_option_result_type_safety() {
// Task returning Result<i32, String>
let result_task = dag.add_task(task_fn(|_: ()| async { Ok::<i32, String>(42) }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Correct wrapped types work
assert_eq!(dag.get(&opt_task).unwrap(), Some(42));
@@ -94,11 +89,9 @@ async fn test_zero_sized_type_safety() {
let zst1_task = dag.add_task(task_fn(|_: ()| async { Zst1 }));
let _zst2_task = dag.add_task(task_fn(|_: ()| async { Zst2 }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Correct ZSTs work
assert!(dag.get(&zst1_task).is_ok());
@@ -117,11 +110,9 @@ async fn test_type_safety_with_many_types() {
let string_task = dag.add_task(task_fn(|_: ()| async { "42".to_string() }));
let bool_task = dag.add_task(task_fn(|_: ()| async { true }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Each type works correctly
assert_eq!(dag.get(&i32_task).unwrap(), 42i32);
@@ -145,11 +136,9 @@ async fn test_type_safety_single_dependency() {
.add_task(task_fn(|x: i32| async move { x * 2 }))
.depends_on(&source);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(transform).unwrap(), 84);
}
@@ -169,11 +158,9 @@ async fn test_type_safety_tuple_dependencies() {
}))
.depends_on((&a, &b, &c));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(combine).unwrap(), "hello: 10 (true)");
}
@@ -186,11 +173,9 @@ async fn test_type_safety_different_output_types() {
let str_source = dag.add_task(task_fn(|_: ()| async { "test".to_string() }));
let vec_source = dag.add_task(task_fn(|_: ()| async { vec![1, 2, 3] }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(&int_source).unwrap(), 42);
assert_eq!(dag.get(&str_source).unwrap(), "test");
@@ -214,11 +199,9 @@ async fn test_type_safety_complex_types() {
vec![Some(1), None, Some(2), Some(3)]
}));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
let tuple_result = dag.get(tuple_task).unwrap();
assert_eq!(tuple_result.0, 42);
@@ -245,11 +228,9 @@ async fn test_type_safety_with_generics() {
))
.depends_on(&source);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(transform).unwrap(), Some(84));
@@ -261,11 +242,9 @@ async fn test_type_safety_with_generics() {
))
.depends_on(&source2);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(transform2).unwrap(), None);
}
@@ -285,11 +264,9 @@ async fn test_large_tuple_input() {
))
.depends_on((&a, &b, &c, &d, &e));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 15);
}
tests/type_safety/handles.rs
+10 -15
View File
@@ -1,5 +1,6 @@
// Type safety tests for TaskHandle
use crate::common::task_fn;
use futures::FutureExt;
// These tests verify dagx's compile-time type safety guarantees.
// Tests cover both runtime behavior (wrong types return None) and
// compile-time checking (type mismatches fail to compile).
@@ -15,11 +16,9 @@ async fn test_get_with_wrong_type_returns_none() {
let int_task = dag.add_task(task_fn(|_: ()| async { 42i32 }));
// Run DAG
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Verify correct type works
assert_eq!(dag.get(&int_task), Ok(42i32));
@@ -38,11 +37,9 @@ async fn test_handle_preserves_type() {
let string_task = dag.add_task(task_fn(|_: ()| async { "hello".to_string() }));
let int_task = dag.add_task(task_fn(|_: ()| async { 42i32 }));
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
// Get with correct types
assert_eq!(dag.get(&string_task).unwrap(), "hello".to_string());
@@ -66,11 +63,9 @@ async fn test_type_safety_handle_reuse() {
.add_task(task_fn(|x: i32| async move { x * 3 }))
.depends_on(&source);
dag.run(|fut| {
tokio::spawn(fut);
})
.await
.unwrap();
dag.run(|fut| tokio::spawn(fut).map(Result::unwrap))
.await
.unwrap();
assert_eq!(dag.get(double).unwrap(), 200);
assert_eq!(dag.get(triple).unwrap(), 300);