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

test: update tests/examples/benches

Browse Source
This commit is contained in:
TechnoPorg committed 2026-02-17 21:31:18 +01:00
1 parent e4fc0b0d44
commit 72a83e29dc
47 files changed
+491 -1239
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/boundaries/deep_chains.rs
+29 -24
View File
@@ -9,13 +9,13 @@ use std::sync::Arc;
#[tokio::test]
async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
// Test 10 parallel chains of depth 500 each
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let mut chain_ends = Vec::new();
for chain_id in 0..8 {
let first = dag.add_task(task_fn::<(), _, _>(move |_: ()| chain_id * 1000));
let mut current: dagx::TaskHandle<i32> = first.into();
let mut current: dagx::TaskHandle<i32> = first;
for _ in 0..500 {
current = dag
@@ -51,7 +51,8 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
&chain_ends[7],
));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Each chain: start + 500
@@ -59,7 +60,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
// Chain 1: 1000 + 500 = 1500
// ...
// Sum = 500*8 + (0+1000+2000+...+7000) = 4000 + 28000 = 32000
assert_eq!(dag.get(final_task)?, 32000);
assert_eq!(output.get(final_task)?, 32000);
Ok(())
}
@@ -67,7 +68,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
#[tokio::test]
async fn test_deep_chain_execution_order() -> DagResult<()> {
// Verify tasks in a deep chain execute in correct order
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let order = Arc::new(AtomicUsize::new(0));
let first = dag.add_task(task_fn::<(), _, _>({
@@ -78,7 +79,7 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
0
}
}));
let mut current: dagx::TaskHandle<i32> = first.into();
let mut current: dagx::TaskHandle<i32> = first;
for expected in 1..100 {
current = dag
@@ -93,11 +94,12 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
.depends_on(current);
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(order.load(Ordering::SeqCst), 100);
assert_eq!(dag.get(current)?, 99);
assert_eq!(output.get(current)?, 99);
Ok(())
}
@@ -105,12 +107,12 @@ async fn test_deep_chain_execution_order() -> DagResult<()> {
#[tokio::test]
async fn test_deep_binary_tree() -> DagResult<()> {
// Test a deep binary tree structure (depth 10 = 1023 nodes)
let dag = DagRunner::new();
let mut dag = DagRunner::new();
fn build_tree(dag: &DagRunner, depth: usize, value: i32) -> dagx::TaskHandle<i32> {
fn build_tree(dag: &mut DagRunner, depth: usize, value: i32) -> dagx::TaskHandle<i32> {
if depth == 0 {
let task = dag.add_task(task_fn::<(), _, _>(move |_: ()| value));
task.into() // Convert TaskBuilder to TaskHandle
task // Convert TaskBuilder to TaskHandle
} else {
let left = build_tree(dag, depth - 1, value * 2);
let right = build_tree(dag, depth - 1, value * 2 + 1);
@@ -122,14 +124,15 @@ async fn test_deep_binary_tree() -> DagResult<()> {
}
}
let root = build_tree(&dag, 10, 1);
let root = build_tree(&mut dag, 10, 1);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.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
let result = dag.get(root)?;
let result = output.get(root)?;
assert!(result > 0);
Ok(())
@@ -138,13 +141,13 @@ async fn test_deep_binary_tree() -> DagResult<()> {
#[tokio::test]
async fn test_fibonacci_chain() -> DagResult<()> {
// Test a Fibonacci-like dependency pattern
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let f0_builder = dag.add_task(task_fn::<(), _, _>(|_: ()| 0i64));
let f1_builder = dag.add_task(task_fn::<(), _, _>(|_: ()| 1i64));
let mut prev2: dagx::TaskHandle<i64> = f0_builder.into();
let mut prev1: dagx::TaskHandle<i64> = f1_builder.into();
let mut prev2: dagx::TaskHandle<i64> = f0_builder;
let mut prev1: dagx::TaskHandle<i64> = f1_builder;
for _ in 2..50 {
let next = dag
@@ -155,11 +158,12 @@ async fn test_fibonacci_chain() -> DagResult<()> {
prev1 = next;
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// 50th Fibonacci number (0-indexed) = 7778742049
assert_eq!(dag.get(prev1)?, 7778742049);
assert_eq!(output.get(prev1)?, 7778742049);
Ok(())
}
@@ -167,13 +171,13 @@ async fn test_fibonacci_chain() -> DagResult<()> {
#[tokio::test]
async fn test_zigzag_dependencies() -> DagResult<()> {
// Test a zigzag pattern: two chains that cross-reference each other
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a_first = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
let b_first = dag.add_task(task_fn::<(), _, _>(|_: ()| 2));
let mut chain_a: dagx::TaskHandle<i32> = a_first.into();
let mut chain_b: dagx::TaskHandle<i32> = b_first.into();
let mut chain_a: dagx::TaskHandle<i32> = a_first;
let mut chain_b: dagx::TaskHandle<i32> = b_first;
for i in 0..100 {
if i % 2 == 0 {
@@ -194,10 +198,11 @@ async fn test_zigzag_dependencies() -> DagResult<()> {
.add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b))
.depends_on((&chain_a, &chain_b));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
let result = dag.get(final_task)?;
let result = output.get(final_task)?;
assert!(result > 100);
Ok(())
tests/boundaries/type_limits.rs
+16 -40
View File
@@ -1,6 +1,6 @@
//! Tests for type system edge cases and limits
use dagx::{DagResult, DagRunner, TaskHandle};
use dagx::{DagResult, DagRunner};
use dagx_test::task_fn;
use std::marker::PhantomData;
@@ -16,13 +16,13 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> {
_phantom: PhantomData<T>,
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// Task producing ZST
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| Empty));
// Task that also produces ZST (independent since we can't extract custom types)
let t2: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into();
let t2 = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
// Task with PhantomData
let _t3 = dag.add_task(task_fn::<(), _, _>(|_: ()| PhantomWrapper::<i32> {
@@ -37,12 +37,13 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> {
}))
.depends_on(t2);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(t1)?, Empty);
assert_eq!(dag.get(t2)?, 42);
assert_eq!(dag.get(t4)?, "success");
assert_eq!(output.get(t1)?, Empty);
assert_eq!(output.get(t2)?, 42);
assert_eq!(output.get(t4)?, "success");
Ok(())
}
@@ -56,7 +57,7 @@ async fn test_large_value_types() -> DagResult<()> {
more_data: Vec<u64>,
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
let large = LargeStruct {
@@ -73,37 +74,11 @@ async fn test_large_value_types() -> DagResult<()> {
}))
.depends_on(t1);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(t2)?, 1042);
Ok(())
}
#[tokio::test]
async fn test_unit_type_chains() -> DagResult<()> {
// Test chains of unit type operations
let dag = DagRunner::new();
let t1: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| {})).into();
let t2 = dag
.add_task(task_fn::<(), _, _>(|_: ()| {}))
.depends_on((t1,));
let t3 = dag
.add_task(task_fn::<(), _, _>(|_: ()| {}))
.depends_on((t2,));
let t4 = dag
.add_task(task_fn::<(), _, _>(|_: ()| "done"))
.depends_on((t3,));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
dag.get(t1)?; // Verify no error
dag.get(t2)?;
dag.get(t3)?;
assert_eq!(dag.get(t4)?, "done");
assert_eq!(output.get(t2)?, 1042);
Ok(())
}
@@ -113,7 +88,7 @@ async fn test_reference_wrapper_types() -> DagResult<()> {
// Test types that wrap references
use std::sync::{Arc, Mutex};
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
let arc = Arc::new(Mutex::new(100));
@@ -126,10 +101,11 @@ async fn test_reference_wrapper_types() -> DagResult<()> {
.add_task(task_fn::<i32, _, _>(|&val: &i32| val * 2))
.depends_on(t1);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(t2)?, 200);
assert_eq!(output.get(t2)?, 200);
Ok(())
}
tests/boundaries/zero_tasks.rs
+12 -9
View File
@@ -8,17 +8,18 @@ use tokio::task::yield_now;
#[tokio::test]
async fn test_dag_with_only_source_tasks() -> DagResult<()> {
// Test a DAG with multiple source tasks but no dependencies
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let tasks: Vec<_> = (0..100)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task)?, i);
assert_eq!(output.get(task)?, i);
}
Ok(())
@@ -38,20 +39,21 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let task = dag.add_task(Counter { value: 21 });
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(task)?, 42);
assert_eq!(output.get(task)?, 42);
Ok(())
}
#[tokio::test]
async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> {
// Test a task that does the absolute minimum async work
let dag = DagRunner::new();
let mut dag = DagRunner::new();
struct YieldTask;
@@ -65,9 +67,10 @@ async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> {
let task = dag.add_task(YieldTask);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(task)?, 1337);
assert_eq!(output.get(task)?, 1337);
Ok(())
}
tests/custom_type_test.rs
+18 -15
View File
@@ -1,5 +1,5 @@
// Test that custom types work with the #[task] macro
use dagx::{task, DagRunner, TaskHandle};
use dagx::{task, DagRunner};
#[derive(Clone, Debug, PartialEq)]
struct Person {
@@ -51,33 +51,35 @@ impl PersonToCompany {
#[tokio::test]
async fn test_custom_type_single_dependency() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let fetch = dag.add_task(FetchPerson);
let process = dag.add_task(ProcessPerson).depends_on(fetch);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
let result = dag.get(process).unwrap();
let result = output.get(process).unwrap();
assert_eq!(result, "Alice is 30 years old");
}
#[tokio::test]
async fn test_custom_type_passthrough() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let fetch: TaskHandle<_> = dag.add_task(FetchPerson).into();
let fetch = dag.add_task(FetchPerson);
let transform = dag.add_task(PersonToCompany).depends_on(fetch);
let process = dag.add_task(ProcessPerson).depends_on(fetch);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
let person_result = dag.get(process).unwrap();
let company_result = dag.get(transform).unwrap();
let person_result = output.get(process).unwrap();
let company_result = output.get(transform).unwrap();
assert_eq!(person_result.as_str(), "Alice is 30 years old");
assert_eq!(company_result.name, "Alice's Company");
@@ -86,20 +88,21 @@ async fn test_custom_type_passthrough() {
#[tokio::test]
async fn test_custom_type_fan_out() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let fetch: TaskHandle<_> = dag.add_task(FetchPerson).into();
let fetch = dag.add_task(FetchPerson);
let process1 = dag.add_task(ProcessPerson).depends_on(fetch);
let process2 = dag.add_task(ProcessPerson).depends_on(fetch);
let process3 = dag.add_task(ProcessPerson).depends_on(fetch);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
let result1 = dag.get(process1).unwrap();
let result2 = dag.get(process2).unwrap();
let result3 = dag.get(process3).unwrap();
let result1 = output.get(process1).unwrap();
let result2 = output.get(process2).unwrap();
let result3 = output.get(process3).unwrap();
assert_eq!(result1, "Alice is 30 years old");
assert_eq!(result2, "Alice is 30 years old");
tests/cycle_prevention.rs
+15 -14
View File
@@ -3,7 +3,7 @@
//! These tests show that even if you try to create cycles, the Rust compiler
//! prevents it through the type-state pattern.
use dagx::{task, DagRunner, TaskHandle};
use dagx::{task, DagRunner};
/// Demonstrates that TaskBuilder is consumed by depends_on,
/// preventing you from wiring the same task multiple times
@@ -25,7 +25,7 @@ fn test_builder_consumed_by_depends_on() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let source = dag.add_task(Source);
let consumer_builder = dag.add_task(Consumer);
@@ -49,11 +49,11 @@ fn test_handle_is_immutable_reference() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let builder = dag.add_task(NoInput);
// Convert to handle (only works because Input = ())
let handle: TaskHandle<_> = builder.into();
let handle = builder;
// TaskHandle is Copy/Clone - just an ID wrapper
let _copy1 = handle;
@@ -85,7 +85,7 @@ fn test_ordering_prevents_cycles() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a_builder = dag.add_task(TaskA);
let b_builder = dag.add_task(TaskB);
@@ -107,7 +107,7 @@ fn test_ordering_prevents_cycles() {
// So we can only create DAGs, never cycles!
// Let's create a valid DAG instead: A→B (B depends on A)
let a_handle: TaskHandle<_> = a_builder.into(); // A has no deps (unit input)
let a_handle = a_builder; // A has no deps (unit input)
let _b_handle = b_builder.depends_on(a_handle);
// Now we could try: a_handle.depends_on(b_handle)?
// But TaskHandle has no such method!
@@ -132,13 +132,13 @@ fn test_unit_tasks_cannot_create_cycles() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let task1 = dag.add_task(Task1);
let task2 = dag.add_task(Task2);
// Both have unit input, so we can convert to handles
let _handle1: TaskHandle<_> = task1.into();
let _handle2: TaskHandle<_> = task2.into();
let _handle1 = task1;
let _handle2 = task2;
// Now we have two handles, but...
// TaskHandle has no methods to add dependencies!
@@ -176,7 +176,7 @@ fn test_three_way_cycle_impossible() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let _a_builder = dag.add_task(A);
let _b_builder = dag.add_task(B);
let _c_builder = dag.add_task(C);
@@ -205,7 +205,7 @@ fn test_self_loop_prevented() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let _builder = dag.add_task(SelfReferential);
// To create A→A (self-loop):
@@ -256,7 +256,7 @@ async fn test_type_safety_prevents_invalid_graphs() {
}
}
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// Build a valid DAG: two sources, two transforms, one aggregator
let s1 = dag.add_task(Source);
@@ -270,11 +270,12 @@ async fn test_type_safety_prevents_invalid_graphs() {
let result = dag.add_task(Aggregate).depends_on((&t1, &t2));
// Execute and verify
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
assert_eq!(dag.get(result).unwrap(), 40); // (10*2) + (10*2)
assert_eq!(output.get(result).unwrap(), 40); // (10*2) + (10*2)
// Now if we wanted to add a cycle, we'd need to:
// - Make one of the Source tasks depend on the Aggregate result
tests/dependencies/resolution.rs
+4 -3
View File
@@ -8,7 +8,7 @@ use std::sync::{Arc, Mutex};
#[tokio::test]
async fn test_dependency_resolution_order() -> DagResult<()> {
// Test that dependencies are resolved in the correct order
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let execution_log = Arc::new(Mutex::new(Vec::new()));
@@ -79,10 +79,11 @@ async fn test_dependency_resolution_order() -> DagResult<()> {
.depends_on(d)
};
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(e)?, 5);
assert_eq!(output.get(e)?, 5);
// Verify execution order
let log = execution_log.lock().unwrap();
tests/dependencies/tuples.rs
+33 -26
View File
@@ -5,7 +5,7 @@ use dagx_test::task_fn;
#[tokio::test]
async fn test_two_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
let b = dag.add_task(task_fn::<(), _, _>(|_: ()| 20));
@@ -13,16 +13,17 @@ async fn test_two_dependencies() -> DagResult<()> {
.add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y))
.depends_on((a, b));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 30);
assert_eq!(output.get(sum)?, 30);
Ok(())
}
#[tokio::test]
async fn test_three_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
let b = dag.add_task(task_fn::<(), _, _>(|_: ()| 2));
@@ -33,19 +34,20 @@ async fn test_three_dependencies() -> DagResult<()> {
))
.depends_on((a, b, c));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 6);
assert_eq!(output.get(sum)?, 6);
Ok(())
}
#[tokio::test]
async fn test_four_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let deps: Vec<_> = (1..=4)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
let sum = dag
@@ -54,19 +56,20 @@ async fn test_four_dependencies() -> DagResult<()> {
))
.depends_on((&deps[0], &deps[1], &deps[2], &deps[3]));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 10); // 1+2+3+4
assert_eq!(output.get(sum)?, 10); // 1+2+3+4
Ok(())
}
#[tokio::test]
async fn test_five_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let deps: Vec<_> = (1..=5)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
let sum = dag
@@ -75,19 +78,20 @@ async fn test_five_dependencies() -> DagResult<()> {
))
.depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4]));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 15); // 1+2+3+4+5
assert_eq!(output.get(sum)?, 15); // 1+2+3+4+5
Ok(())
}
#[tokio::test]
async fn test_six_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let deps: Vec<_> = (1..=6)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
let sum = dag
@@ -96,19 +100,20 @@ async fn test_six_dependencies() -> DagResult<()> {
))
.depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5]));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 21); // 1+2+3+4+5+6
assert_eq!(output.get(sum)?, 21); // 1+2+3+4+5+6
Ok(())
}
#[tokio::test]
async fn test_seven_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let deps: Vec<_> = (1..=7)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
let sum = dag
@@ -121,19 +126,20 @@ async fn test_seven_dependencies() -> DagResult<()> {
&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6],
));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 28); // 1+2+3+4+5+6+7
assert_eq!(output.get(sum)?, 28); // 1+2+3+4+5+6+7
Ok(())
}
#[tokio::test]
async fn test_eight_dependencies() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let deps: Vec<_> = (1..=8)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)).into())
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
let sum = dag
@@ -146,9 +152,10 @@ 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| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(sum)?, 36); // Sum of 1..=8
assert_eq!(output.get(sum)?, 36); // Sum of 1..=8
Ok(())
}
tests/errors/propagation.rs
+48 -72
View File
@@ -1,17 +1,17 @@
//! Tests for error propagation through the DAG
use dagx::{task, DagRunner, TaskHandle};
use dagx::{task, DagError, DagRunner};
use dagx_test::task_fn;
use std::sync::Arc;
use std::sync::{Arc, Mutex};
use std::time::Duration;
use tokio::time::sleep;
#[tokio::test]
async fn test_linear_error_propagation() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let t1: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 1)).into();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
let t2 = dag
.add_task(task_fn::<i32, _, _>(|&x: &i32| -> i32 {
panic!("Error at t2 with input {}", x);
@@ -20,28 +20,19 @@ async fn test_linear_error_propagation() {
let t3 = dag
.add_task(task_fn::<i32, _, _>(|&x: &i32| x + 1))
.depends_on(t2);
let t4 = dag
.add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
dag.add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
.depends_on(t3);
let _ = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await;
// t1 succeeds
assert_eq!(dag.get(t1).unwrap(), 1);
// t2 fails
assert!(dag.get(t2).is_err());
// t3 and t4 fail due to dependency
assert!(dag.get(t3).is_err());
assert!(dag.get(t4).is_err());
assert!(matches!(
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await,
Err(DagError::TaskPanicked { .. })
));
}
#[tokio::test]
async fn test_diamond_error_propagation() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// source
// / \
@@ -49,7 +40,7 @@ async fn test_diamond_error_propagation() {
// \ /
// sink
let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 100)).into();
let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 100));
let left = dag
.add_task(task_fn::<i32, _, _>(|&x: &i32| x / 2))
@@ -61,61 +52,45 @@ async fn test_diamond_error_propagation() {
}))
.depends_on(source);
let sink = dag
.add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r))
dag.add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r))
.depends_on((&left, &right));
let _ = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await;
assert_eq!(dag.get(source).unwrap(), 100);
assert_eq!(dag.get(left).unwrap(), 50);
assert!(dag.get(right).is_err());
assert!(dag.get(sink).is_err()); // Fails due to right dependency
assert!(matches!(
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await,
Err(DagError::TaskPanicked { .. })
));
}
#[tokio::test]
async fn test_error_in_wide_fanout() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let source: TaskHandle<_> = dag
.add_task(task_fn::<(), _, _>(|_: ()| -> i32 {
panic!("Source fails");
}))
.into();
let source = dag.add_task(task_fn::<(), _, _>(|_: ()| -> i32 {
panic!("Source fails");
}));
// Create wide fanout from failing source
let dependents: Vec<_> = (0..20)
.map(|i| {
dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
.depends_on(source)
})
.collect();
let _ = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await;
// Source fails
assert!(dag.get(source).is_err());
// All dependents fail
for dependent in &dependents {
assert!(dag.get(dependent).is_err());
for i in 0..20 {
dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
.depends_on(source);
}
assert!(matches!(
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await,
Err(DagError::TaskPanicked { .. })
));
}
#[tokio::test]
async fn test_selective_error_propagation() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// Source that produces a Result
let source: TaskHandle<_> = dag
.add_task(task_fn::<(), _, _>(|_: ()| {
Result::<i32, String>::Err("Source error".to_string())
}))
.into();
let source = dag.add_task(task_fn::<(), _, _>(|_: ()| {
Result::<i32, String>::Err("Source error".to_string())
}));
// Handler that processes the Result
let handler = dag
@@ -140,34 +115,35 @@ async fn test_selective_error_propagation() {
}))
.depends_on(handler);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// Source returns an Err variant (not a panic)
assert_eq!(dag.get(source).unwrap(), Err("Source error".to_string()));
assert_eq!(output.get(source).unwrap(), Err("Source error".to_string()));
// Handler processes it
assert_eq!(dag.get(handler).unwrap(), -1);
assert_eq!(output.get(handler).unwrap(), -1);
// Final task handles the error case
assert_eq!(dag.get(final_task).unwrap(), "Handled error case");
assert_eq!(output.get(final_task).unwrap(), "Handled error case");
}
#[tokio::test]
async fn test_error_propagation_timing() {
let dag = DagRunner::new();
let propagation_order = Arc::new(parking_lot::Mutex::new(Vec::new()));
let mut dag = DagRunner::new();
let propagation_order = Arc::new(Mutex::new(Vec::new()));
struct T1Task {
order: Arc<parking_lot::Mutex<Vec<&'static str>>>,
order: Arc<Mutex<Vec<&'static str>>>,
}
#[task]
impl T1Task {
async fn run(&self) -> i32 {
sleep(Duration::from_millis(10)).await;
self.order.lock().push("t1_complete");
self.order.lock().unwrap().push("t1_complete");
1
}
}
@@ -177,13 +153,13 @@ async fn test_error_propagation_timing() {
});
struct T2Task {
order: Arc<parking_lot::Mutex<Vec<&'static str>>>,
order: Arc<Mutex<Vec<&'static str>>>,
}
#[task]
impl T2Task {
async fn run(&self, x: &i32) -> i32 {
self.order.lock().push("t2_start");
self.order.lock().unwrap().push("t2_start");
sleep(Duration::from_millis(10)).await;
panic!("t2 fails with {x}");
}
@@ -200,7 +176,7 @@ async fn test_error_propagation_timing() {
let order = propagation_order.clone();
move |x: &i32| {
let order = order.clone();
order.lock().push("t3_should_not_run");
order.lock().unwrap().push("t3_should_not_run");
x + 1
}
}))
@@ -210,7 +186,7 @@ async fn test_error_propagation_timing() {
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await;
let order = propagation_order.lock().clone();
let order = propagation_order.lock().unwrap().clone();
// t1 completes, t2 starts and fails, t3 never runs
assert!(order.contains(&"t1_complete"));
tests/errors/recovery.rs
+26 -19
View File
@@ -4,11 +4,11 @@ use dagx::{DagResult, DagRunner};
use dagx_test::task_fn;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::sync::{Arc, Mutex};
#[tokio::test]
async fn test_retry_pattern_simulation() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let attempt_counter = Arc::new(AtomicUsize::new(0));
// Task that would fail on first attempts (simulated)
@@ -29,18 +29,19 @@ async fn test_retry_pattern_simulation() -> DagResult<()> {
}
}));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Task completes with error on first "attempt"
assert_eq!(dag.get(task)?, Err("Simulated failure"));
assert_eq!(output.get(task)?, Err("Simulated failure"));
Ok(())
}
#[tokio::test]
async fn test_circuit_breaker_pattern() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let failure_count = Arc::new(AtomicUsize::new(0));
let circuit_open = Arc::new(AtomicBool::new(false));
@@ -71,12 +72,13 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> {
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// First few tasks fail and open circuit
for task in tasks.into_iter().take(4) {
assert!(dag.get(task)?.is_err());
assert!(output.get(task)?.is_err());
}
Ok(())
@@ -84,8 +86,8 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> {
#[tokio::test]
async fn test_error_accumulation_pattern() -> DagResult<()> {
let dag = DagRunner::new();
let errors = Arc::new(parking_lot::Mutex::new(Vec::new()));
let mut dag = DagRunner::new();
let errors = Arc::new(Mutex::new(Vec::new()));
// Tasks that may produce errors
let mut tasks: Vec<_> = (0..10)
@@ -94,7 +96,10 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
dag.add_task(task_fn::<(), _, _>(move |_: ()| {
let errors = errors.clone();
if i % 2 == 0 {
errors.lock().push(format!("Error from task {}", i));
errors
.lock()
.unwrap()
.push(format!("Error from task {}", i));
return Err(format!("Task {} error", i));
}
Ok(i)
@@ -102,11 +107,12 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Check accumulated errors
let error_list = errors.lock().clone();
let error_list = errors.lock().unwrap().clone();
assert_eq!(error_list.len(), 5); // Even numbered tasks
tasks.truncate(2);
@@ -114,15 +120,15 @@ async fn test_error_accumulation_pattern() -> DagResult<()> {
let [first, second] = <[_; 2]>::try_from(tasks).map_err(|_e| ()).unwrap();
// Verify specific results
assert!(dag.get(first)?.is_err());
assert_eq!(dag.get(second)?, Ok(1));
assert!(output.get(first)?.is_err());
assert_eq!(output.get(second)?, Ok(1));
Ok(())
}
#[tokio::test]
async fn test_error_boundary_isolation() -> DagResult<()> {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// Group A: contains an error - use Result
let a1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
@@ -143,15 +149,16 @@ async fn test_error_boundary_isolation() -> DagResult<()> {
.add_task(task_fn::<i32, _, _>(|&x: &i32| x + 5))
.depends_on(b2);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Group A fails (returns Err)
assert_eq!(dag.get(a2)?, Err("Group A error"));
assert_eq!(output.get(a2)?, Err("Group A error"));
// Groups B and C succeed
assert_eq!(dag.get(b2)?, 40);
assert_eq!(dag.get(c1)?, 45);
assert_eq!(output.get(b2)?, 40);
assert_eq!(output.get(c1)?, 45);
Ok(())
}
tests/execution/mod.rs
+39 -31
View File
@@ -6,43 +6,46 @@ use dagx::*;
#[tokio::test]
async fn test_empty_dag() {
let dag = DagRunner::new();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let _ = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap(); // Should complete without error
}
#[tokio::test]
async fn test_single_task() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
assert_eq!(dag.get(task).unwrap(), 42);
assert_eq!(output.get(task).unwrap(), 42);
}
#[tokio::test]
async fn test_wide_parallel() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let tasks: Vec<_> = (0..10)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2)))
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i * 2);
assert_eq!(output.get(task).unwrap(), i * 2);
}
}
#[tokio::test]
async fn test_diamond_dependency() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
let b = dag
.add_task(task_fn::<i32, _, _>(|&x: &i32| x + 5))
.depends_on(a);
@@ -53,13 +56,14 @@ async fn test_diamond_dependency() {
.add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y))
.depends_on((&b, &c));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// Note: a, b, and c are not sinks (they have dependents)
// Only d is a sink and can be retrieved
assert_eq!(dag.get(d).unwrap(), 35); // 15 + 20
assert_eq!(output.get(d).unwrap(), 35); // 15 + 20
}
#[derive(Clone)]
@@ -70,7 +74,7 @@ struct User {
#[tokio::test]
async fn test_different_output_types() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let string_task = dag.add_task(task_fn::<(), _, _>(|_: ()| "hello".to_string()));
let vec_task = dag.add_task(task_fn::<(), _, _>(|_: ()| vec![1, 2, 3]));
@@ -79,22 +83,23 @@ async fn test_different_output_types() {
age: 30,
}));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
assert_eq!(&dag.get(string_task).unwrap(), "hello");
assert_eq!(dag.get(vec_task).unwrap(), vec![1, 2, 3]);
let user = dag.get(struct_task).unwrap();
assert_eq!(&output.get(string_task).unwrap(), "hello");
assert_eq!(output.get(vec_task).unwrap(), vec![1, 2, 3]);
let user = output.get(struct_task).unwrap();
assert_eq!(user.name, "Alice");
assert_eq!(user.age, 30);
}
#[tokio::test]
async fn test_multiple_sinks() {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
// Branch 1
let branch1 = dag
@@ -112,21 +117,22 @@ async fn test_multiple_sinks() {
.add_task(task_fn::<i32, _, _>(|&x: &i32| x * 3))
.depends_on(branch2);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.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
assert_eq!(output.get(sink1).unwrap(), 21); // (10 * 2) + 1
assert_eq!(output.get(sink2).unwrap(), 45); // (10 + 5) * 3
}
#[tokio::test]
async fn test_single_task_inline_path() {
// Test single-task layer inline execution path (line 367-410 in runner.rs)
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// Single task in its own layer
let t1: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)).into();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 10));
let t2 = dag
.add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
.depends_on(t1);
@@ -134,19 +140,20 @@ async fn test_single_task_inline_path() {
.add_task(task_fn::<i32, _, _>(|&x: &i32| x + 5))
.depends_on(t2);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
assert_eq!(dag.get(t3).unwrap(), 25); // (10 * 2) + 5
assert_eq!(output.get(t3).unwrap(), 25); // (10 * 2) + 5
}
#[tokio::test]
async fn test_multi_consumer_fanout() {
// Test fanout with multiple consumers from one producer
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let producer: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into();
let producer = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
// Multiple consumers
let c1 = dag
@@ -159,11 +166,12 @@ async fn test_multi_consumer_fanout() {
.add_task(task_fn::<i32, _, _>(|&x: &i32| x - 5))
.depends_on(producer);
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
assert_eq!(dag.get(c1).unwrap(), 84);
assert_eq!(dag.get(c2).unwrap(), 52);
assert_eq!(dag.get(c3).unwrap(), 37);
assert_eq!(output.get(c1).unwrap(), 84);
assert_eq!(output.get(c2).unwrap(), 52);
assert_eq!(output.get(c3).unwrap(), 37);
}
tests/timing/interleaving.rs → tests/interleaving/mod.rs
+35 -32
View File
@@ -4,19 +4,19 @@ use dagx::{task, DagResult, DagRunner};
use dagx_test::task_fn;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::sync::{Arc, Mutex};
use tokio::sync::Barrier;
#[tokio::test]
async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
// Force specific interleaving using barriers
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let barrier = Arc::new(Barrier::new(3));
let order = Arc::new(parking_lot::Mutex::new(Vec::new()));
let order = Arc::new(Mutex::new(Vec::new()));
struct InterleaveTask {
barrier: Arc<Barrier>,
order: Arc<parking_lot::Mutex<Vec<String>>>,
order: Arc<Mutex<Vec<String>>>,
prefix: &'static str,
value: i32,
}
@@ -24,9 +24,15 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
#[task]
impl InterleaveTask {
async fn run(&self) -> i32 {
self.order.lock().push(format!("{}_start", self.prefix));
self.order
.lock()
.unwrap()
.push(format!("{}_start", self.prefix));
self.barrier.wait().await;
self.order.lock().push(format!("{}_end", self.prefix));
self.order
.lock()
.unwrap()
.push(format!("{}_end", self.prefix));
self.value
}
}
@@ -58,13 +64,14 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
))
.depends_on((t1, t2, t3));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(collector)?, 6);
assert_eq!(output.get(collector)?, 6);
// Check interleaving pattern
let events = order.lock().clone();
let events = order.lock().unwrap().clone();
assert_eq!(events.len(), 6);
// All starts should come before all ends due to barrier
@@ -80,14 +87,13 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> {
#[tokio::test]
async fn test_alternating_execution_pattern() -> DagResult<()> {
// Create an alternating execution pattern
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let turn = Arc::new(AtomicUsize::new(0));
let events: Arc<parking_lot::Mutex<Vec<String>>> =
Arc::new(parking_lot::Mutex::new(Vec::new()));
let events: Arc<Mutex<Vec<String>>> = Arc::new(Mutex::new(Vec::new()));
struct FirstA {
turn: Arc<AtomicUsize>,
events: Arc<parking_lot::Mutex<Vec<String>>>,
events: Arc<Mutex<Vec<String>>>,
}
#[task]
@@ -96,22 +102,20 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
while self.turn.load(Ordering::SeqCst) != 0 {
tokio::task::yield_now().await;
}
self.events.lock().push("A0".to_string());
self.events.lock().unwrap().push("A0".to_string());
self.turn.store(1, Ordering::SeqCst);
0
}
}
let first_a = dag.add_task(FirstA {
let mut chain_a = dag.add_task(FirstA {
turn: turn.clone(),
events: events.clone(),
});
let mut chain_a: dagx::TaskHandle<i32> = first_a.into();
struct FirstB {
turn: Arc<AtomicUsize>,
events: Arc<parking_lot::Mutex<Vec<String>>>,
events: Arc<Mutex<Vec<String>>>,
}
#[task]
@@ -120,22 +124,20 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
while self.turn.load(Ordering::SeqCst) != 1 {
tokio::task::yield_now().await;
}
self.events.lock().push("B0".to_string());
self.events.lock().unwrap().push("B0".to_string());
self.turn.store(2, Ordering::SeqCst);
0
}
}
let first_b = dag.add_task(FirstB {
let mut chain_b = dag.add_task(FirstB {
turn: turn.clone(),
events: events.clone(),
});
let mut chain_b: dagx::TaskHandle<i32> = first_b.into();
struct ChainA {
turn: Arc<AtomicUsize>,
events: Arc<parking_lot::Mutex<Vec<String>>>,
events: Arc<Mutex<Vec<String>>>,
idx: usize,
}
@@ -145,7 +147,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
while self.turn.load(Ordering::SeqCst) != self.idx * 2 {
tokio::task::yield_now().await;
}
self.events.lock().push(format!("A{}", self.idx));
self.events.lock().unwrap().push(format!("A{}", self.idx));
self.turn.store(self.idx * 2 + 1, Ordering::SeqCst);
x + 1
}
@@ -153,7 +155,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
struct ChainB {
turn: Arc<AtomicUsize>,
events: Arc<parking_lot::Mutex<Vec<String>>>,
events: Arc<Mutex<Vec<String>>>,
idx: usize,
}
@@ -163,7 +165,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
while self.turn.load(Ordering::SeqCst) != self.idx * 2 + 1 {
tokio::task::yield_now().await;
}
self.events.lock().push(format!("B{}", self.idx));
self.events.lock().unwrap().push(format!("B{}", self.idx));
self.turn.store(self.idx * 2 + 2, Ordering::SeqCst);
x + 1
}
@@ -187,11 +189,12 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
.depends_on(chain_b);
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let _output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Check alternating pattern
let event_list = events.lock().clone();
let event_list = events.lock().unwrap().clone();
assert_eq!(
event_list,
vec!["A0", "B0", "A1", "B1", "A2", "B2", "A3", "B3", "A4", "B4"]
@@ -203,7 +206,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> {
#[tokio::test]
async fn test_layered_interleaving() -> DagResult<()> {
// Test interleaving across DAG layers
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let layer_active = Arc::new(AtomicUsize::new(0));
struct Layer1Task {
@@ -228,7 +231,6 @@ async fn test_layered_interleaving() -> DagResult<()> {
active: layer_active.clone(),
idx: i,
})
.into()
})
.collect();
@@ -264,11 +266,12 @@ async fn test_layered_interleaving() -> DagResult<()> {
})
.depends_on((&layer2[0], &layer2[1]));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Sum of 0+1+2+3 = 6
assert_eq!(dag.get(layer3)?, 6);
assert_eq!(output.get(layer3)?, 6);
Ok(())
}
tests/lib_tests.rs
+1 -1
View File
@@ -8,7 +8,7 @@ mod boundaries;
mod dependencies;
mod errors;
mod execution;
mod interleaving;
mod parallelism;
mod runtimes;
mod timing;
mod tracing;
tests/parallelism/concurrent.rs
-425
View File
@@ -1,425 +0,0 @@
//! Concurrent access tests
use dagx::{task, DagRunner};
use dagx_test::task_fn;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::task::JoinSet;
struct SleepTask(usize);
#[task]
impl SleepTask {
async fn run(&self) -> usize {
tokio::time::sleep(Duration::from_millis(5)).await;
self.0
}
}
#[tokio::test]
async fn test_concurrent_run_calls() {
// Test that multiple run() calls are handled correctly
let dag = Arc::new(DagRunner::new());
// Add some tasks
let tasks: Vec<_> = (0..10).map(|i| dag.add_task(SleepTask(i))).collect();
// Try to run the DAG multiple times concurrently
let mut handles = JoinSet::new();
for _ in 0..5 {
let dag = Arc::clone(&dag);
handles.spawn(async move {
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
});
}
// Only one run should succeed, others should fail with concurrent execution error
let mut success_count = 0;
let mut error_count = 0;
while let Some(result) = handles.join_next().await {
match result.unwrap() {
Ok(_) => success_count += 1,
Err(_) => error_count += 1,
}
}
// Exactly one should succeed, the rest should fail
assert_eq!(success_count, 1);
assert_eq!(error_count, 4);
// Verify results are consistent from the successful run
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i);
}
}
#[tokio::test]
async fn test_concurrent_execution_and_building() {
// Test that we can build and execute concurrently
let dag = Arc::new(DagRunner::new());
// Start with some initial tasks
let initial: Vec<_> = (0..10)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
// Execute the initial DAG
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// Now try to add more tasks while also reading results
let mut handles = JoinSet::new();
// Reader tasks
for task in initial {
let dag = Arc::clone(&dag);
let task_handle: dagx::TaskHandle<i32> = task.into();
handles.spawn(async move { dag.get(task_handle).unwrap() });
}
// Builder tasks (adding new nodes)
for i in 10..20 {
let dag = Arc::clone(&dag);
handles.spawn(async move {
dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
i
});
}
// Wait for all operations
while let Some(result) = handles.join_next().await {
result.unwrap();
}
// Execute again with the new tasks
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
}
#[tokio::test]
async fn test_dag_builder_thread_safety() {
// Test that the DAG builder is thread-safe
let dag = Arc::new(DagRunner::new());
let counter = Arc::new(AtomicUsize::new(0));
let mut handles = JoinSet::new();
// Spawn tasks that build complex sub-graphs concurrently
for thread_id in 0..10 {
let dag = Arc::clone(&dag);
let counter = Arc::clone(&counter);
handles.spawn(async move {
// Each thread creates a small pipeline
let source = dag.add_task(task_fn::<(), _, _>(move |_: ()| thread_id));
let transform1 = dag
.add_task(task_fn::<i32, _, _>(move |&x: &i32| x * 10))
.depends_on(source);
let transform2 = dag
.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + thread_id))
.depends_on(transform1);
counter.fetch_add(1, Ordering::SeqCst);
transform2
});
}
let mut final_tasks = vec![];
while let Some(result) = handles.join_next().await {
final_tasks.push(result.unwrap());
}
// Verify all threads completed
assert_eq!(counter.load(Ordering::SeqCst), 10);
// Execute the combined DAG
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// Verify each pipeline computed correctly
for task in final_tasks.iter() {
let result = dag.get(task).unwrap();
// Each pipeline: thread_id * 10 + thread_id
assert!((0..100).contains(&result));
}
}
#[tokio::test]
async fn test_concurrent_get_operations() {
// Test concurrent get() operations
let dag = Arc::new(DagRunner::new());
// Create and execute tasks
let tasks: Vec<_> = (0..100)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// Now access results concurrently
let mut handles = JoinSet::new();
for task in tasks.into_iter() {
let dag = Arc::clone(&dag);
let task_handle: dagx::TaskHandle<i32> = task.into();
handles.spawn(async move { dag.get(task_handle).unwrap() });
}
// Collect and verify results
let mut results = vec![];
while let Some(result) = handles.join_next().await {
results.push(result.unwrap());
}
// Results should contain all values 0..100
results.sort();
for (i, val) in results.iter().enumerate() {
assert_eq!(*val, i as i32);
}
}
#[tokio::test]
async fn test_no_race_in_dependency_tracking() {
// Test that concurrent dependency modifications don't cause races
let dag = Arc::new(DagRunner::new());
let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
let mut handles = JoinSet::new();
// Multiple threads create tasks depending on the same source
let source_handle: dagx::TaskHandle<i32> = source.into();
for i in 0..50 {
let dag = Arc::clone(&dag);
let source_clone = source_handle;
handles.spawn(async move {
dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
.depends_on(source_clone)
});
}
let mut dependents = vec![];
while let Some(result) = handles.join_next().await {
dependents.push(result.unwrap());
}
// Execute
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// All dependents should have correct values
for dep in dependents.iter() {
let result = dag.get(dep).unwrap();
assert!((42..92).contains(&result)); // 42 + (0..50)
}
}
#[tokio::test]
async fn test_high_concurrency_stress() {
// Stress test with high concurrency
let dag = Arc::new(DagRunner::new());
let completed = Arc::new(AtomicUsize::new(0));
let mut handles = JoinSet::new();
// 1000 concurrent operations
for i in 0..1000 {
let dag = Arc::clone(&dag);
let completed = Arc::clone(&completed);
handles.spawn(async move {
// Mix of operations
match i % 3 {
0 => {
// Add a source task
dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
}
1 => {
// Add a task with a simple dependency
let source = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
dag.add_task(task_fn::<i32, _, _>(move |&x: &i32| x * 2))
.depends_on(source);
}
2 => {
// Add a more complex subgraph
let a = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
let b = dag.add_task(task_fn::<(), _, _>(move |_: ()| i + 1));
dag.add_task(task_fn::<(i32, i32), _, _>(move |(x, y): (&i32, &i32)| {
x + y
}))
.depends_on((a, b));
}
_ => unreachable!(),
}
completed.fetch_add(1, Ordering::SeqCst);
});
}
// Wait for all operations to complete
while let Some(result) = handles.join_next().await {
result.unwrap();
}
assert_eq!(completed.load(Ordering::SeqCst), 1000);
// The DAG should still be executable
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
}
#[tokio::test]
async fn test_lock_contention_measurement() {
// Test to measure lock contention under concurrent load
let dag = Arc::new(DagRunner::new());
let start = tokio::time::Instant::now();
let operations = Arc::new(AtomicUsize::new(0));
let mut handles = JoinSet::new();
// Create many concurrent operations that will contend for locks
for i in 0..100 {
let dag = Arc::clone(&dag);
let ops = Arc::clone(&operations);
handles.spawn(async move {
for j in 0..10 {
dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 10 + j));
ops.fetch_add(1, Ordering::SeqCst);
// Small yield to increase contention likelihood
tokio::task::yield_now().await;
}
});
}
// Wait for completion
while let Some(result) = handles.join_next().await {
result.unwrap();
}
let duration = start.elapsed();
let total_ops = operations.load(Ordering::SeqCst);
println!(
"Completed {} operations in {:?} ({:.0} ops/sec)",
total_ops,
duration,
total_ops as f64 / duration.as_secs_f64()
);
// Should complete reasonably quickly even with contention
assert!(
duration < Duration::from_secs(5),
"Too much lock contention"
);
}
#[tokio::test]
async fn test_atomic_dag_operations() {
// Test that DAG operations are atomic (all-or-nothing)
let dag = Arc::new(DagRunner::new());
// Create a base task
let base = dag.add_task(task_fn::<(), _, _>(|_: ()| 100));
// Try to create dependent tasks concurrently
let mut handles = JoinSet::new();
let success_count = Arc::new(AtomicUsize::new(0));
let base_handle: dagx::TaskHandle<i32> = base.into();
for i in 0..50 {
let dag = Arc::clone(&dag);
let base_clone = base_handle;
let success = Arc::clone(&success_count);
handles.spawn(async move {
let dep = dag
.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + i))
.depends_on(base_clone);
success.fetch_add(1, Ordering::SeqCst);
dep
});
}
let mut deps = vec![];
while let Some(result) = handles.join_next().await {
deps.push(result.unwrap());
}
// All operations should have succeeded
assert_eq!(success_count.load(Ordering::SeqCst), 50);
assert_eq!(deps.len(), 50);
// Execute and verify
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
for dep in deps.iter() {
let result = dag.get(dep).unwrap();
assert!((100..150).contains(&result));
}
}
#[tokio::test]
async fn test_no_deadlock_on_circular_waiting() {
// Test that we don't deadlock even with complex concurrent operations
let dag = Arc::new(DagRunner::new());
// Create multiple threads that might cause circular waiting patterns
let mut handles = JoinSet::new();
for i in 0..10 {
let dag = Arc::clone(&dag);
handles.spawn(async move {
// Each thread creates a chain of tasks
let first = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
let mut chain = vec![first.into()];
for j in 0..5 {
let next = dag
.add_task(task_fn::<i32, _, _>(move |&x: &i32| x + j))
.depends_on(chain.last().unwrap());
chain.push(next);
// Yield to potentially cause interleaving
tokio::task::yield_now().await;
}
*chain.last().unwrap()
});
}
// Set a timeout to detect deadlocks
let result = tokio::time::timeout(Duration::from_secs(5), async {
let mut finals = vec![];
while let Some(result) = handles.join_next().await {
finals.push(result.unwrap());
}
finals
})
.await;
assert!(result.is_ok(), "Deadlock detected - operation timed out");
// Should be able to execute the DAG
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
}
tests/parallelism/mod.rs
-1
View File
@@ -1,6 +1,5 @@
//! Parallelism tests module
pub mod concurrent;
pub mod spawning;
pub mod threads;
pub mod timing;
tests/parallelism/spawning.rs
+4 -3
View File
@@ -9,7 +9,7 @@ use std::sync::Arc;
#[tokio::test]
async fn test_spawner_actually_spawns_tasks() -> DagResult<()> {
// Verify that the spawner function is actually called for each task
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let spawn_count = Arc::new(AtomicUsize::new(0));
// Create 10 independent tasks
@@ -27,12 +27,13 @@ async fn test_spawner_actually_spawns_tasks() -> DagResult<()> {
.collect();
// Custom spawner that counts invocations
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Verify all tasks were spawned
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task)?, i as i32);
assert_eq!(output.get(task)?, i as i32);
}
// The spawner should have been called once for each task
tests/parallelism/threads.rs
+12 -10
View File
@@ -1,6 +1,6 @@
//! Tests for multi-threading and concurrent execution
use dagx::{task, DagResult, DagRunner, TaskHandle};
use dagx::{task, DagResult, DagRunner};
use dagx_test::task_fn;
use std::sync::atomic::{AtomicUsize, Ordering};
@@ -11,7 +11,7 @@ use tokio::time::sleep;
#[tokio::test]
async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
// Prove concurrent execution using atomic counters to track simultaneous tasks
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let concurrent_count = Arc::new(AtomicUsize::new(0));
let max_concurrent = Arc::new(AtomicUsize::new(0));
@@ -62,12 +62,13 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Verify all tasks completed
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task)?, i);
assert_eq!(output.get(task)?, i);
}
// Check that we achieved real parallelism
@@ -90,13 +91,13 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> {
#[tokio::test]
async fn test_massive_parallel_fanout() -> DagResult<()> {
// Test extreme parallelism with 1000 parallel tasks
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let completed = Arc::new(AtomicUsize::new(0));
let max_concurrent = Arc::new(AtomicUsize::new(0));
let current = Arc::new(AtomicUsize::new(0));
let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 1)).into();
let source = dag.add_task(task_fn::<(), _, _>(|_: ()| 1));
struct FanoutTask {
comp: Arc<AtomicUsize>,
@@ -147,7 +148,8 @@ async fn test_massive_parallel_fanout() -> DagResult<()> {
.collect();
let start = Instant::now();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
let elapsed = start.elapsed();
@@ -155,9 +157,9 @@ async fn test_massive_parallel_fanout() -> DagResult<()> {
assert_eq!(completed.load(Ordering::SeqCst), 1000);
// Check some results
assert_eq!(dag.get(tasks[0])?, 1);
assert_eq!(dag.get(tasks[500])?, 501);
assert_eq!(dag.get(tasks[999])?, 1000);
assert_eq!(output.get(tasks[0])?, 1);
assert_eq!(output.get(tasks[500])?, 501);
assert_eq!(output.get(tasks[999])?, 1000);
let max_seen = max_concurrent.load(Ordering::SeqCst);
println!(
tests/parallelism/timing.rs
+9 -7
View File
@@ -22,7 +22,7 @@ impl SleepTask {
#[tokio::test]
async fn test_parallel_execution_speedup() -> DagResult<()> {
// Prove parallel execution by comparing timing
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let work_duration = Duration::from_millis(50);
let num_tasks = 10;
@@ -33,13 +33,14 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
.collect();
let start = Instant::now();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
let elapsed = start.elapsed();
// Verify all tasks completed
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task)?, i);
assert_eq!(output.get(task)?, i);
}
// If tasks ran sequentially: 10 * 50ms = 500ms
@@ -53,7 +54,7 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
// Parallel execution should be significantly faster
assert!(
elapsed < sequential_time / 4,
elapsed < sequential_time / 2,
"Parallel execution too slow: {:?} vs sequential {:?}",
elapsed,
sequential_time
@@ -65,7 +66,7 @@ async fn test_parallel_execution_speedup() -> DagResult<()> {
#[tokio::test]
async fn test_layers_execute_in_parallel() -> DagResult<()> {
// Test that tasks in the same topological layer execute in parallel
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let layer_timing = Arc::new(Mutex::new(Vec::new()));
@@ -142,10 +143,11 @@ async fn test_layers_execute_in_parallel() -> DagResult<()> {
.depends_on((&layer1[0], &layer1[1], &layer1[2]))
};
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(final_task)?, 33); // (0+1) + (10+1) + (20+1)
assert_eq!(output.get(final_task)?, 33); // (0+1) + (10+1) + (20+1)
// Analyze timing
let timing = layer_timing.lock().unwrap();
tests/runtimes/mod.rs
+9 -9
View File
@@ -95,15 +95,15 @@ impl Add {
#[test_case(AsyncExecutorRuntimeTest)]
fn test_basic_dag(runner: impl RuntimeTest) {
runner.run_dagx_test(|spawner| async {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let x = dag.add_task(Value(2));
let y = dag.add_task(Value(3));
let sum = dag.add_task(Add).depends_on((x, y));
dag.run(spawner).await.unwrap();
let mut output = dag.run(spawner).await.unwrap();
assert_eq!(dag.get(sum).unwrap(), 5);
assert_eq!(output.get(sum).unwrap(), 5);
})
}
@@ -113,16 +113,16 @@ fn test_basic_dag(runner: impl RuntimeTest) {
#[test_case(AsyncExecutorRuntimeTest)]
fn test_parallel_execution(runner: impl RuntimeTest) {
runner.run_dagx_test(|spawner| async {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let tasks: Vec<_> = (0..10)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 2)))
.collect();
dag.run(spawner).await.unwrap();
let mut output = dag.run(spawner).await.unwrap();
for (i, task) in tasks.into_iter().enumerate() {
assert_eq!(dag.get(task).unwrap(), i * 2);
assert_eq!(output.get(task).unwrap(), i * 2);
}
});
}
@@ -133,7 +133,7 @@ fn test_parallel_execution(runner: impl RuntimeTest) {
#[test_case(AsyncExecutorRuntimeTest)]
fn test_complex_dependencies(runner: impl RuntimeTest) {
runner.run_dagx_test(|spawner| async {
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a = dag.add_task(Value(10));
let b = dag.add_task(Value(20));
@@ -142,8 +142,8 @@ fn test_complex_dependencies(runner: impl RuntimeTest) {
.add_task(task_fn::<i32, _, _>(|&x: &i32| x * 2))
.depends_on(sum);
dag.run(spawner).await.unwrap();
let mut output = dag.run(spawner).await.unwrap();
assert_eq!(dag.get(double).unwrap().clone(), 60);
assert_eq!(output.get(double).unwrap().clone(), 60);
});
}
tests/timing/concurrent_access.rs
-331
View File
@@ -1,331 +0,0 @@
//! Tests for concurrent access patterns and race conditions
use dagx::{task, DagResult, DagRunner, TaskHandle};
use dagx_test::task_fn;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::RwLock;
#[tokio::test]
async fn test_concurrent_result_retrieval() -> DagResult<()> {
// Test concurrent get() calls after execution
let dag = Arc::new(DagRunner::new());
struct DelayTask(usize, Duration);
#[task]
impl DelayTask {
async fn run(&self) -> usize {
tokio::time::sleep(self.1).await;
self.0
}
}
// Add tasks with minimal delays
let tasks: Vec<dagx::TaskHandle<usize>> = (0..10)
.map(|i| {
let task = dag.add_task(DelayTask(i, Duration::from_millis(1)));
task.into()
})
.collect();
// Execute the DAG
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Now try to get results concurrently from multiple threads
let mut get_handles = Vec::new();
for (i, task) in tasks.iter().enumerate() {
let dag_clone = dag.clone();
let task = *task;
get_handles.push(tokio::spawn(async move {
// Results should be available immediately
dag_clone.get(task).map(|v| (i, v))
}));
}
// Verify all gets succeeded
for handle in get_handles {
let (i, result) = handle.await.unwrap()?;
assert_eq!(result, i);
}
Ok(())
}
#[tokio::test]
async fn test_simultaneous_dag_building_and_execution() -> DagResult<()> {
// Test building DAG while another part is executing
let dag = Arc::new(DagRunner::new());
let building_complete = Arc::new(AtomicBool::new(false));
// Start with a few initial tasks
let initial: Vec<_> = (0..5)
.map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i)))
.collect();
// Start building more tasks in background
let dag_build = dag.clone();
let building = building_complete.clone();
let build_handle = tokio::spawn(async move {
let mut tasks: Vec<dagx::TaskHandle<i32>> = Vec::new();
for i in 5..50 {
let task = dag_build.add_task(task_fn::<(), _, _>(move |_: ()| i));
tasks.push(task.into());
tokio::time::sleep(tokio::time::Duration::from_millis(1)).await;
}
building.store(true, Ordering::SeqCst);
tasks
});
// Wait a bit for some tasks to be added
tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
// Execute the DAG
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Get the additional tasks
let additional_tasks = build_handle.await.unwrap();
// Initial tasks should have results
for (i, task) in initial.into_iter().enumerate() {
assert_eq!(dag.get(task)?, i);
}
// Additional tasks added during execution won't have results yet
// (they need another run cycle)
if building_complete.load(Ordering::SeqCst) {
// Run again for the new tasks
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Now they should have results
for (i, task) in additional_tasks.iter().enumerate() {
assert_eq!(dag.get(task)?, (i + 5) as i32);
}
}
Ok(())
}
#[tokio::test]
async fn test_concurrent_dependency_resolution() -> DagResult<()> {
// Test multiple tasks resolving dependencies concurrently
let dag = DagRunner::new();
let resolution_counter = Arc::new(AtomicUsize::new(0));
// Create a source task
let source: TaskHandle<_> = dag
.add_task(task_fn::<(), _, _>({
let counter = resolution_counter.clone();
move |_: ()| {
let counter = counter.clone();
counter.fetch_add(1, Ordering::SeqCst);
42
}
}))
.into();
// Create many tasks depending on the same source
let dependents: Vec<_> = (0..100)
.map(|i| {
let counter = resolution_counter.clone();
dag.add_task(task_fn::<i32, _, _>(move |val: &i32| {
let counter = counter.clone();
counter.fetch_add(1, Ordering::SeqCst);
val + i
}))
.depends_on(source)
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Check all completed
assert_eq!(resolution_counter.load(Ordering::SeqCst), 101);
// Verify results
for (i, task) in dependents.iter().enumerate() {
assert_eq!(dag.get(task)?, 42 + i as i32);
}
Ok(())
}
#[tokio::test]
async fn test_read_write_race_patterns() -> DagResult<()> {
// Test read-write patterns with shared state
let dag = DagRunner::new();
let shared_state = Arc::new(RwLock::new(Vec::new()));
struct WriterTask {
state: Arc<RwLock<Vec<i32>>>,
idx: i32,
}
#[task]
impl WriterTask {
async fn run(&self) -> i32 {
let mut guard = self.state.write().await;
guard.push(self.idx);
tokio::task::yield_now().await; // Force potential race
guard.push(self.idx * 10);
self.idx
}
}
// Writers
let writers: Vec<_> = (0..10)
.map(|i| {
dag.add_task(WriterTask {
state: shared_state.clone(),
idx: i,
})
.into()
})
.collect();
struct ReaderTask {
state: Arc<RwLock<Vec<i32>>>,
}
struct ReaderTaskOutput((i32, bool, usize));
#[task]
impl ReaderTask {
async fn run(&self, writer_id: &i32) -> ReaderTaskOutput {
let guard = self.state.read().await;
let contains_writer = guard.contains(&writer_id);
let len = guard.len();
ReaderTaskOutput((*writer_id, contains_writer, len))
}
}
// Readers (depend on writers to ensure ordering)
let readers: Vec<_> = writers
.iter()
.map(|writer| {
dag.add_task(ReaderTask {
state: shared_state.clone(),
})
.depends_on(writer)
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Verify readers saw their writer's data
for reader in &readers {
let ReaderTaskOutput((writer_id, saw_writer, _len)) = dag.get(reader)?;
assert!(saw_writer, "Reader didn't see writer {}", writer_id);
}
// Final state should have all writes
let final_state = shared_state.read().await;
assert_eq!(final_state.len(), 20); // 10 writers * 2 writes each
Ok(())
}
#[tokio::test]
async fn test_atomic_counter_races() -> DagResult<()> {
// Test atomic operations under high concurrency
let dag = DagRunner::new();
let counter = Arc::new(AtomicUsize::new(0));
let max_seen = Arc::new(AtomicUsize::new(0));
// Create tasks that increment and track max
let _tasks: Vec<_> = (0..200)
.map(|_| {
let counter = counter.clone();
let max_seen = max_seen.clone();
dag.add_task(task_fn::<(), _, _>(move |_: ()| {
let counter = counter.clone();
let max_seen = max_seen.clone();
let mut local_max = 0;
for _ in 0..100 {
let val = counter.fetch_add(1, Ordering::SeqCst);
local_max = local_max.max(val);
// Update global max
let mut current_max = max_seen.load(Ordering::SeqCst);
while val > current_max {
match max_seen.compare_exchange_weak(
current_max,
val,
Ordering::SeqCst,
Ordering::SeqCst,
) {
Ok(_) => break,
Err(x) => current_max = x,
}
}
}
}))
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Final counter should be 200 * 100 = 20000
assert_eq!(counter.load(Ordering::SeqCst), 20000);
// Max seen should be at least 19999 (last increment - 1)
assert!(max_seen.load(Ordering::SeqCst) >= 19999);
Ok(())
}
#[tokio::test]
async fn test_multiple_dag_runners_interaction() -> DagResult<()> {
// Test multiple DAG runners potentially interfering
let shared_counter = Arc::new(AtomicUsize::new(0));
let mut handles = Vec::new();
for dag_id in 0..5 {
let counter = shared_counter.clone();
let handle = tokio::spawn(async move {
let dag = DagRunner::new();
let tasks: Vec<_> = (0..20)
.map(|task_id| {
let counter = counter.clone();
dag.add_task(task_fn::<(), _, _>(move |_: ()| {
let counter = counter.clone();
counter.fetch_add(1, Ordering::SeqCst);
(dag_id, task_id)
}))
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Collect results
let mut results = Vec::new();
for task in tasks {
results.push(dag.get(task)?);
}
Ok::<_, dagx::DagError>(results)
});
handles.push(handle);
}
// Wait for all DAGs to complete
for handle in handles {
handle.await.unwrap()?;
}
// Should have executed 5 * 20 = 100 tasks total
assert_eq!(shared_counter.load(Ordering::SeqCst), 100);
Ok(())
}
tests/timing/mod.rs
-6
View File
@@ -1,6 +0,0 @@
//! Timing and race condition tests
//!
//! Tests for temporal edge cases, execution ordering, and race conditions.
mod concurrent_access;
mod interleaving;
tests/tracing/with_tracing.rs
+9 -7
View File
@@ -39,17 +39,18 @@ async fn test_tracing_with_subscriber() {
.with_test_writer()
.try_init();
let dag = DagRunner::new();
let mut dag = DagRunner::new();
let a = dag.add_task(Value(2));
let b = dag.add_task(Value(3));
let sum = dag.add_task(Add).depends_on((a, b));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
assert_eq!(dag.get(sum).unwrap(), 5);
assert_eq!(output.get(sum).unwrap(), 5);
}
#[tokio::test(flavor = "multi_thread")]
@@ -60,19 +61,20 @@ async fn test_tracing_inline_execution() {
.with_test_writer()
.try_init();
let dag = DagRunner::new();
let mut dag = DagRunner::new();
// Create a linear chain (should trigger inline execution)
let a = dag.add_task(Value(1)).into();
let a = dag.add_task(Value(1));
let b = dag.add_task(Add).depends_on((&a, &a));
let c = dag.add_task(Multiply).depends_on((&b, &b));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
let mut output = dag
.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await
.unwrap();
// a = 1
// b = 1 + 1 = 2
// c = 2 * 2 = 4
assert_eq!(dag.get(c).unwrap(), 4);
assert_eq!(output.get(c).unwrap(), 4);
}