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

refactor: consolidate some tests/examples/benchmarks

Browse Source
This commit is contained in:
TechnoPorg committed 2026-02-12 23:31:19 +01:00
1 parent 49be497964
commit b8204c4ed9
83 files changed
+569 -10068
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

tests/boundaries/deep_chains.rs
+1 -106
View File
@@ -1,70 +1,11 @@
//! Tests for extremely deep dependency chains
use crate::common::task_fn;
use dagx::task_fn;
use dagx::{DagResult, DagRunner};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
#[tokio::test]
async fn test_1000_level_deep_chain() -> DagResult<()> {
// Test a linear chain of 1000 tasks
let dag = DagRunner::new();
let first = dag.add_task(task_fn::<(), _, _>(|_: ()| 0));
let mut current = first.into(); // Convert to TaskHandle
for _i in 1..1000 {
current = dag
.add_task(task_fn::<i32, _, _>(move |&prev: &i32| prev + 1))
.depends_on(current);
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(current)?, 999);
Ok(())
}
#[tokio::test]
async fn test_deep_chain_with_branches() -> DagResult<()> {
// Test a deep chain where each node has a side branch
let dag = DagRunner::new();
let first = dag.add_task(task_fn::<(), _, _>(|_: ()| 0));
let mut main_chain: dagx::TaskHandle<i32> = first.into();
let mut branches = Vec::new();
for i in 1..500 {
// Main chain continues
main_chain = dag
.add_task(task_fn::<i32, _, _>(move |&prev: &i32| prev + 1))
.depends_on(main_chain);
// Create a branch at every 10th node
if i % 10 == 0 {
let branch = dag
.add_task(task_fn::<i32, _, _>(move |main_val: &i32| main_val * 2))
.depends_on(main_chain);
branches.push(branch);
}
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(main_chain)?, 499);
// Check some branches
if !branches.is_empty() {
assert_eq!(dag.get(branches[0])?, 20); // 10 * 2
assert_eq!(dag.get(branches[4])?, 100); // 50 * 2
}
Ok(())
}
#[tokio::test]
async fn test_multiple_parallel_deep_chains() -> DagResult<()> {
// Test 10 parallel chains of depth 500 each
@@ -223,30 +164,6 @@ async fn test_fibonacci_chain() -> DagResult<()> {
Ok(())
}
#[tokio::test]
async fn test_deep_chain_with_accumulator() -> DagResult<()> {
// Test a deep chain where each node accumulates a value
let dag = DagRunner::new();
// Use a simpler accumulation pattern - just sum
let first = dag.add_task(task_fn::<(), _, _>(|_: ()| 0i64));
let mut current: dagx::TaskHandle<i64> = first.into();
for i in 1..=200 {
current = dag
.add_task(task_fn::<i64, _, _>(move |sum: &i64| sum + i as i64))
.depends_on(current);
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
let result = dag.get(current)?;
assert_eq!(result, 20100); // sum of 1..=200
Ok(())
}
#[tokio::test]
async fn test_zigzag_dependencies() -> DagResult<()> {
// Test a zigzag pattern: two chains that cross-reference each other
@@ -285,25 +202,3 @@ async fn test_zigzag_dependencies() -> DagResult<()> {
Ok(())
}
#[tokio::test]
#[cfg_attr(tarpaulin, ignore)] // This test is extremely resource intensive
async fn test_10000_level_chain_stress() -> DagResult<()> {
// Ultimate deep chain test
let dag = DagRunner::new();
let first = dag.add_task(task_fn::<(), _, _>(|_: ()| 0u64));
let mut current: dagx::TaskHandle<u64> = first.into();
for _ in 1..10_000 {
current = dag
.add_task(task_fn::<u64, _, _>(|prev: &u64| prev + 1))
.depends_on(current);
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(current)?, 9_999);
Ok(())
}
tests/boundaries/large_scale.rs
-309
View File
@@ -1,309 +0,0 @@
//! Large scale stress tests with thousands of nodes and edges
use crate::common::task_fn;
use dagx::{DagResult, DagRunner};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
#[tokio::test]
async fn test_10000_independent_tasks() -> DagResult<()> {
// Test with 10,000 independent tasks
let dag = DagRunner::new();
let counter = Arc::new(AtomicUsize::new(0));
let mut tasks: Vec<_> = (0..10_000)
.map(|i| {
let counter = counter.clone();
dag.add_task(task_fn::<(), _, _>(move |_: ()| {
counter.fetch_add(1, Ordering::Relaxed);
i
}))
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Verify all tasks completed
assert_eq!(counter.load(Ordering::Relaxed), 10_000);
// Spot check some results
assert_eq!(dag.get(tasks.swap_remove(9_999))?, 9_999);
assert_eq!(dag.get(tasks.swap_remove(5_000))?, 5_000);
assert_eq!(dag.get(tasks.swap_remove(0))?, 0);
Ok(())
}
#[tokio::test]
async fn test_5000_node_pyramid() -> DagResult<()> {
// Create a pyramid structure with ~5000 nodes
// Layer sizes: 2048 -> 512 -> 128 -> 32 -> 8 -> 2 -> 1
let dag = DagRunner::new();
// Bottom layer: 2048 nodes - convert to TaskHandles
let mut current_layer: Vec<dagx::TaskHandle<i32>> = (0..2048)
.map(|i| {
let task = dag.add_task(task_fn::<(), _, _>(
move |_: ()| {
if i < 1024 {
1
} else {
0
}
},
));
task.into()
})
.collect();
let layer_sizes = vec![512, 128, 32, 8, 2];
for layer_size in layer_sizes {
let mut next_layer = Vec::with_capacity(layer_size);
let group_size = current_layer.len() / layer_size;
for i in 0..layer_size {
let start = i * group_size;
let deps: Vec<_> = current_layer[start..start + group_size.min(4)]
.iter()
.take(4)
.collect();
let task = match deps.len() {
1 => dag
.add_task(task_fn::<i32, _, _>(|&a: &i32| a))
.depends_on(deps[0]),
2 => dag
.add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b))
.depends_on((deps[0], deps[1])),
3 => dag
.add_task(task_fn::<(i32, i32, i32), _, _>(
|(a, b, c): (&i32, &i32, &i32)| a + b + c,
))
.depends_on((deps[0], deps[1], deps[2])),
4 => dag
.add_task(task_fn::<(i32, i32, i32, i32), _, _>(
|(a, b, c, d): (&i32, &i32, &i32, &i32)| a + b + c + d,
))
.depends_on((deps[0], deps[1], deps[2], deps[3])),
_ => unreachable!(),
};
next_layer.push(task);
}
current_layer = next_layer;
}
// Final layer combines the last 2
let root = dag
.add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b))
.depends_on((&current_layer[0], &current_layer[1]));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// We created 1024 ones and 1024 zeros at the bottom
// The sum should propagate up to 1024
let result = dag.get(root)?;
assert!(result > 0, "Expected positive result, got {}", result);
Ok(())
}
#[tokio::test]
async fn test_5000_node_pyramid_single_threaded() -> DagResult<()> {
// Create a pyramid structure with ~5000 nodes
// Layer sizes: 2048 -> 512 -> 128 -> 32 -> 8 -> 2 -> 1
let dag = DagRunner::new();
// Bottom layer: 2048 nodes - convert to TaskHandles
let mut current_layer: Vec<dagx::TaskHandle<i32>> = (0..2048)
.map(|i| {
let task = dag.add_task(task_fn::<(), _, _>(
move |_: ()| {
if i < 1024 {
1
} else {
0
}
},
));
task.into()
})
.collect();
let layer_sizes = vec![512, 128, 32, 8, 2];
for layer_size in layer_sizes {
let mut next_layer = Vec::with_capacity(layer_size);
let group_size = current_layer.len() / layer_size;
for i in 0..layer_size {
let start = i * group_size;
let deps: Vec<_> = current_layer[start..start + group_size.min(4)]
.iter()
.take(4)
.collect();
let task = match deps.len() {
1 => dag
.add_task(task_fn::<i32, _, _>(|&a: &i32| a))
.depends_on(deps[0]),
2 => dag
.add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b))
.depends_on((deps[0], deps[1])),
3 => dag
.add_task(task_fn::<(i32, i32, i32), _, _>(
|(a, b, c): (&i32, &i32, &i32)| a + b + c,
))
.depends_on((deps[0], deps[1], deps[2])),
4 => dag
.add_task(task_fn::<(i32, i32, i32, i32), _, _>(
|(a, b, c, d): (&i32, &i32, &i32, &i32)| a + b + c + d,
))
.depends_on((deps[0], deps[1], deps[2], deps[3])),
_ => unreachable!(),
};
next_layer.push(task);
}
current_layer = next_layer;
}
// Final layer combines the last 2
let root = dag
.add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b))
.depends_on((&current_layer[0], &current_layer[1]));
dag.run(|fut| fut).await?;
// We created 1024 ones and 1024 zeros at the bottom
// The sum should propagate up to 1024
let result = dag.get(root)?;
assert!(result > 0, "Expected positive result, got {}", result);
Ok(())
}
#[tokio::test]
async fn test_wide_dag_1000_sources_1000_sinks() -> DagResult<()> {
// Test a very wide DAG with 1000 sources and 1000 sinks
let dag = DagRunner::new();
// Create 1000 source tasks
let sources: Vec<_> = (0..1000)
.map(|i| {
dag.add_task(task_fn::<(), _, _>(move |_: ()| i % 10))
.into()
})
.collect();
// Create 1000 sink tasks, each depending on 3 random sources
let sinks: Vec<_> = (0..1000)
.map(|i| {
let idx1 = (i * 7) % 1000;
let idx2 = (i * 13) % 1000;
let idx3 = (i * 17) % 1000;
dag.add_task(task_fn::<(i32, i32, i32), _, _>(
move |(a, b, c): (&i32, &i32, &i32)| a + b + c + i as i32,
))
.depends_on((&sources[idx1], &sources[idx2], &sources[idx3]))
})
.collect();
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Verify some sinks completed correctly
let result0 = dag.get(sinks[0])?;
let result500 = dag.get(sinks[500])?;
let result999 = dag.get(sinks[999])?;
assert!(result0 >= 0);
assert!(result500 >= 500);
assert!(result999 >= 999);
Ok(())
}
#[tokio::test]
async fn test_10000_node_linear_chain_segments() -> DagResult<()> {
// Test 10000 nodes arranged as 100 parallel chains of 100 nodes each
let dag = DagRunner::new();
let counter = Arc::new(AtomicUsize::new(0));
let mut chain_ends = Vec::new();
for chain_id in 0..100 {
let first = dag.add_task(task_fn::<(), _, _>({
let counter = counter.clone();
move |_: ()| {
let counter = counter.clone();
counter.fetch_add(1, Ordering::Relaxed);
chain_id * 100
}
}));
let mut current: dagx::TaskHandle<i32> = first.into();
for _i in 1..100 {
current = dag
.add_task(task_fn::<i32, _, _>({
let counter = counter.clone();
move |&prev: &i32| {
let counter = counter.clone();
counter.fetch_add(1, Ordering::Relaxed);
prev + 1
}
}))
.depends_on(current);
}
chain_ends.push(current);
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// All 10,000 tasks should have executed
assert_eq!(counter.load(Ordering::Relaxed), 10_000);
// Each chain should end with start + 99
for (chain_id, &chain_end) in chain_ends.iter().enumerate() {
let expected = chain_id as i32 * 100 + 99;
assert_eq!(dag.get(chain_end)?, expected);
}
Ok(())
}
#[tokio::test]
#[cfg_attr(tarpaulin, ignore)] // This test is very resource intensive
async fn test_100000_nodes_stress() -> DagResult<()> {
// Ultimate stress test: 100,000 nodes
let dag = DagRunner::new();
let completed = Arc::new(AtomicUsize::new(0));
// Create 100,000 tasks in batches
let batch_size = 1000;
let num_batches = 100;
for batch in 0..num_batches {
let _: Vec<_> = (0..batch_size)
.map(|i| {
let completed = completed.clone();
let task_id = batch * batch_size + i;
dag.add_task(task_fn::<(), _, _>(move |_: ()| {
let completed = completed.clone();
completed.fetch_add(1, Ordering::Relaxed);
task_id
}))
})
.collect();
}
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(completed.load(Ordering::Relaxed), 100_000);
Ok(())
}
tests/boundaries/mod.rs
-1
View File
@@ -4,6 +4,5 @@
//! by normal usage patterns.
mod deep_chains;
mod large_scale;
mod type_limits;
mod zero_tasks;
tests/boundaries/type_limits.rs
+1 -152
View File
@@ -1,6 +1,6 @@
//! Tests for type system edge cases and limits
use crate::common::task_fn;
use dagx::task_fn;
use dagx::{DagResult, DagRunner, TaskHandle};
use std::marker::PhantomData;
@@ -108,89 +108,6 @@ async fn test_unit_type_chains() -> DagResult<()> {
Ok(())
}
#[tokio::test]
async fn test_string_types_variety() -> DagResult<()> {
// Test various string types
let dag = DagRunner::new();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| "static str"));
let t2 = dag.add_task(task_fn::<(), _, _>(|_: ()| String::from("owned string")));
let t3 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
std::borrow::Cow::Borrowed("cow str")
}));
let t4 = dag.add_task(task_fn::<(), _, _>(|_: ()| Box::new("boxed str")));
let t5 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
std::sync::Arc::new("arc str 2")
}));
let t6 = dag.add_task(task_fn::<(), _, _>(|_: ()| std::sync::Arc::new("arc str")));
let combined = dag
.add_task(task_fn::<
(
&str,
String,
std::borrow::Cow<str>,
Box<&str>,
std::sync::Arc<&str>,
std::sync::Arc<&str>,
),
_,
_,
>(
|(s1, s2, s3, s4, s5, s6): (
&&str,
&String,
&std::borrow::Cow<str>,
&Box<&str>,
&std::sync::Arc<&str>,
&std::sync::Arc<&str>,
)| { format!("{} {} {} {} {} {}", s1, s2, s3, s4, s5, s6) },
))
.depends_on((t1, t2, t3, t4, t5, t6));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
let result = dag.get(combined)?;
assert!(result.contains("static str"));
assert!(result.contains("owned string"));
Ok(())
}
#[tokio::test]
async fn test_generic_type_constraints() -> DagResult<()> {
// Test with types that have trait bounds
#[derive(Debug, Clone, PartialEq)]
struct Wrapper<T: Clone + Send + 'static>(T);
let dag = DagRunner::new();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| Wrapper(42)));
let t2 = dag.add_task(task_fn::<(), _, _>(|_: ()| Wrapper("hello")));
let t3 = dag.add_task(task_fn::<(), _, _>(|_: ()| Wrapper(vec![1, 2, 3])));
let combined = dag
.add_task(task_fn::<
(Wrapper<i32>, Wrapper<&str>, Wrapper<Vec<i32>>),
_,
_,
>(
|(w1, w2, w3): (&Wrapper<i32>, &Wrapper<&str>, &Wrapper<Vec<i32>>)| {
format!("{:?} {:?} {:?}", w1, w2, w3)
},
))
.depends_on((t1, t2, t3));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
let result = dag.get(combined)?;
assert!(result.contains("Wrapper(42)"));
Ok(())
}
#[tokio::test]
async fn test_reference_wrapper_types() -> DagResult<()> {
// Test types that wrap references
@@ -216,71 +133,3 @@ async fn test_reference_wrapper_types() -> DagResult<()> {
Ok(())
}
#[tokio::test]
async fn test_array_types() -> DagResult<()> {
// Test with fixed-size arrays
let dag = DagRunner::new();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| [1, 2, 3, 4, 5]));
let t2 = dag.add_task(task_fn::<(), _, _>(|_: ()| [6, 7, 8, 9, 10]));
let t3 = dag
.add_task(task_fn::<([i32; 5], [i32; 5]), _, _>(
|(a1, a2): (&[i32; 5], &[i32; 5])| {
let mut sum = 0;
for i in 0..5 {
sum += a1[i] + a2[i];
}
sum
},
))
.depends_on((t1, t2));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(t3)?, 55);
Ok(())
}
#[tokio::test]
async fn test_enum_variants() -> DagResult<()> {
// Test with various enum types
#[derive(Debug, Clone, PartialEq)]
#[allow(clippy::large_enum_variant)]
enum MyEnum {
Unit,
Tuple(i32, String),
Struct { x: i32, y: i32 },
LargeVariant([u8; 1024]),
}
let dag = DagRunner::new();
let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| MyEnum::Unit));
let t2 = dag.add_task(task_fn::<(), _, _>(|_: ()| {
MyEnum::Tuple(42, "test".to_string())
}));
let t3 = dag.add_task(task_fn::<(), _, _>(|_: ()| MyEnum::Struct { x: 10, y: 20 }));
let t4 = dag.add_task(task_fn::<(), _, _>(|_: ()| MyEnum::LargeVariant([0; 1024])));
let combined = dag
.add_task(task_fn::<(MyEnum, MyEnum, MyEnum, MyEnum), _, _>(
|(e1, e2, e3, e4): (&MyEnum, &MyEnum, &MyEnum, &MyEnum)| {
matches!(e1, MyEnum::Unit)
&& matches!(e2, MyEnum::Tuple(42, _))
&& matches!(e3, MyEnum::Struct { x: 10, y: 20 })
&& matches!(e4, MyEnum::LargeVariant(_))
},
))
.depends_on((t1, t2, t3, t4));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert!(dag.get(combined)?);
Ok(())
}
tests/boundaries/zero_tasks.rs
+1 -121
View File
@@ -1,89 +1,10 @@
//! Tests for empty DAGs and minimal task configurations
use crate::common::task_fn;
use dagx::task_fn;
use dagx::{task, DagResult, DagRunner};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::task::yield_now;
#[tokio::test]
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| async move { tokio::spawn(fut).await.unwrap() })
.await?;
// Should complete successfully with no tasks
Ok(())
}
#[tokio::test]
async fn test_single_task_no_deps() -> DagResult<()> {
// Test the absolute minimum: one task with no dependencies
let dag = DagRunner::new();
let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(task)?, 42);
Ok(())
}
#[tokio::test]
async fn test_single_task_unit_input_output() -> DagResult<()> {
// Test a task that takes unit and returns unit
let dag = DagRunner::new();
let executed = Arc::new(AtomicUsize::new(0));
let exec_clone = executed.clone();
let task = dag.add_task(task_fn::<(), _, _>(move |_: ()| {
let exec = exec_clone.clone();
exec.fetch_add(1, Ordering::SeqCst);
}));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
dag.get(task)?; // Just verify it doesn't error
assert_eq!(executed.load(Ordering::SeqCst), 1);
Ok(())
}
#[tokio::test]
async fn test_zero_sized_type_task() -> DagResult<()> {
// Test with zero-sized types
#[derive(Debug, Clone, PartialEq)]
struct ZeroSized;
let dag = DagRunner::new();
let task = dag.add_task(task_fn::<(), _, _>(|_: ()| ZeroSized));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(task)?, ZeroSized);
Ok(())
}
#[tokio::test]
async fn test_empty_dag_multiple_runs() -> DagResult<()> {
// Test that an empty DAG can be run multiple times
let dag = DagRunner::new();
for _ in 0..10 {
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
}
Ok(())
}
#[tokio::test]
async fn test_dag_with_only_source_tasks() -> DagResult<()> {
// Test a DAG with multiple source tasks but no dependencies
@@ -127,47 +48,6 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> {
Ok(())
}
#[tokio::test]
async fn test_alternating_empty_and_filled_dag_runs() -> DagResult<()> {
// Test alternating between empty DAG and DAG with tasks
for i in 0..5 {
let dag = DagRunner::new();
if i % 2 == 0 {
// Empty DAG
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
} else {
// DAG with a task
let task = dag.add_task(task_fn::<(), _, _>(move |_: ()| i));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(task)?, i);
}
}
Ok(())
}
#[tokio::test]
async fn test_dag_with_never_type_simulation() -> DagResult<()> {
// Test handling of tasks that conceptually "never" complete
// (simulated by immediate completion with special marker)
#[derive(Debug, Clone, PartialEq)]
enum Never {}
let dag = DagRunner::new();
// This task completes immediately but returns a Result containing "never" type
let task = dag.add_task(task_fn::<(), _, _>(|_: ()| Result::<i32, Never>::Ok(42)));
dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })
.await?;
assert_eq!(dag.get(task)?, Ok(42));
Ok(())
}
#[tokio::test]
async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> {
// Test a task that does the absolute minimum async work