From b6ff1f81e6da15a524ba978ab86f4a9488de016c Mon Sep 17 00:00:00 2001 From: TechnoPorg Date: Wed, 11 Feb 2026 23:06:33 +0100 Subject: [PATCH] chore: fix formatting for CI --- benches/basic/execution.rs | 1 - benches/basic/scaling.rs | 1 - benches/comparisons/deep_chain.rs | 1 - benches/comparisons/diamond_pattern.rs | 1 - benches/comparisons/etl_pipeline.rs | 1 - benches/comparisons/large_scale.rs | 1 - benches/comparisons/linear_pipeline.rs | 1 - benches/comparisons/wide_fanout.rs | 1 - benches/patterns/breakdown.rs | 1 - benches/patterns/diamond.rs | 1 - benches/patterns/fanout.rs | 1 - benches/realistic/etl.rs | 1 - benches/realistic/mixed.rs | 1 - examples/01_basic.rs | 1 - examples/02_fan_out.rs | 1 - examples/03_fan_in.rs | 1 - examples/04_parallel_computation.rs | 3 ++- examples/complex_dag.rs | 4 ++-- examples/conditional_workflow.rs | 4 ++-- examples/custom_types.rs | 1 - examples/data_pipeline.rs | 1 - examples/error_handling.rs | 1 - tests/boundaries/deep_chains.rs | 27 ++++++++++++++------- tests/boundaries/large_scale.rs | 15 ++++++++---- tests/boundaries/type_limits.rs | 24 ++++++++++++------- tests/boundaries/zero_tasks.rs | 33 +++++++++++++++++--------- tests/dependencies/resolution.rs | 12 ++++++---- tests/dependencies/single.rs | 16 ++++++++----- tests/dependencies/tuples.rs | 22 ++++++++++------- tests/parallelism/concurrent.rs | 5 +++- tests/parallelism/execution.rs | 9 ++++--- tests/runtimes/compatibility.rs | 4 ++-- 32 files changed, 116 insertions(+), 81 deletions(-) diff --git a/benches/basic/execution.rs b/benches/basic/execution.rs index 70a895c..c044bfc 100644 --- a/benches/basic/execution.rs +++ b/benches/basic/execution.rs @@ -3,7 +3,6 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner}; - pub fn bench_dag_execution(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/benches/basic/scaling.rs b/benches/basic/scaling.rs index 61d1904..c4d6a17 100644 --- a/benches/basic/scaling.rs +++ b/benches/basic/scaling.rs @@ -3,7 +3,6 @@ use criterion::{BenchmarkId, Criterion}; use dagx::{task_fn, DagRunner}; - pub fn bench_dag_scaling(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("scaling"); diff --git a/benches/comparisons/deep_chain.rs b/benches/comparisons/deep_chain.rs index 324c965..c93084b 100644 --- a/benches/comparisons/deep_chain.rs +++ b/benches/comparisons/deep_chain.rs @@ -5,7 +5,6 @@ use criterion::Criterion; use dagx::TaskHandle; - pub fn bench_deep_chain(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("deep_chain_100_tasks"); diff --git a/benches/comparisons/diamond_pattern.rs b/benches/comparisons/diamond_pattern.rs index 6d92606..c0f15bd 100644 --- a/benches/comparisons/diamond_pattern.rs +++ b/benches/comparisons/diamond_pattern.rs @@ -5,7 +5,6 @@ use criterion::Criterion; use dagx::TaskHandle; - pub fn bench_diamond_pattern(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("diamond_pattern"); diff --git a/benches/comparisons/etl_pipeline.rs b/benches/comparisons/etl_pipeline.rs index fcb92da..ef9f70c 100644 --- a/benches/comparisons/etl_pipeline.rs +++ b/benches/comparisons/etl_pipeline.rs @@ -4,7 +4,6 @@ use criterion::Criterion; - pub fn bench_etl_pipeline(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("etl_pipeline"); diff --git a/benches/comparisons/large_scale.rs b/benches/comparisons/large_scale.rs index ff03076..53c2bb4 100644 --- a/benches/comparisons/large_scale.rs +++ b/benches/comparisons/large_scale.rs @@ -4,7 +4,6 @@ use criterion::Criterion; - pub fn bench_large_scale(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("large_scale_10k_tasks"); diff --git a/benches/comparisons/linear_pipeline.rs b/benches/comparisons/linear_pipeline.rs index 543ce24..63d243c 100644 --- a/benches/comparisons/linear_pipeline.rs +++ b/benches/comparisons/linear_pipeline.rs @@ -4,7 +4,6 @@ use criterion::Criterion; - pub fn bench_linear_pipeline(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("linear_pipeline_5_tasks"); diff --git a/benches/comparisons/wide_fanout.rs b/benches/comparisons/wide_fanout.rs index da434a2..960956d 100644 --- a/benches/comparisons/wide_fanout.rs +++ b/benches/comparisons/wide_fanout.rs @@ -5,7 +5,6 @@ use criterion::Criterion; use dagx::TaskHandle; - pub fn bench_wide_fanout(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("wide_fanout_100_tasks"); diff --git a/benches/patterns/breakdown.rs b/benches/patterns/breakdown.rs index 401d762..e58d77e 100644 --- a/benches/patterns/breakdown.rs +++ b/benches/patterns/breakdown.rs @@ -3,7 +3,6 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner}; - pub fn bench_10k_breakdown(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("10k_tasks"); diff --git a/benches/patterns/diamond.rs b/benches/patterns/diamond.rs index ce215ad..5806bb9 100644 --- a/benches/patterns/diamond.rs +++ b/benches/patterns/diamond.rs @@ -3,7 +3,6 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner, TaskHandle}; - pub fn bench_diamond(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/benches/patterns/fanout.rs b/benches/patterns/fanout.rs index c68a60f..9e29a85 100644 --- a/benches/patterns/fanout.rs +++ b/benches/patterns/fanout.rs @@ -3,7 +3,6 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner, TaskHandle}; - pub fn bench_fanout(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/benches/realistic/etl.rs b/benches/realistic/etl.rs index 80a9610..06c0e80 100644 --- a/benches/realistic/etl.rs +++ b/benches/realistic/etl.rs @@ -3,7 +3,6 @@ use criterion::{BenchmarkId, Criterion}; use dagx::{task_fn, DagRunner, TaskHandle}; - pub fn bench_etl_pipeline(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); let mut group = c.benchmark_group("etl_scaling"); diff --git a/benches/realistic/mixed.rs b/benches/realistic/mixed.rs index f0d7637..9b706d1 100644 --- a/benches/realistic/mixed.rs +++ b/benches/realistic/mixed.rs @@ -3,7 +3,6 @@ use criterion::Criterion; use dagx::{task_fn, DagRunner, TaskHandle}; - pub fn bench_mixed_patterns(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/examples/01_basic.rs b/examples/01_basic.rs index 5e7ba3c..5bd065c 100644 --- a/examples/01_basic.rs +++ b/examples/01_basic.rs @@ -32,7 +32,6 @@ use dagx::{task, DagRunner}; - // Step 1: Define a source task (no dependencies) // // Source tasks produce initial data for the DAG. diff --git a/examples/02_fan_out.rs b/examples/02_fan_out.rs index 0cc78cd..f387a79 100644 --- a/examples/02_fan_out.rs +++ b/examples/02_fan_out.rs @@ -57,7 +57,6 @@ use dagx::{task, DagRunner, TaskHandle}; - // Source task with a value field struct Source { initial_value: i32, diff --git a/examples/03_fan_in.rs b/examples/03_fan_in.rs index cbc3761..48596c1 100644 --- a/examples/03_fan_in.rs +++ b/examples/03_fan_in.rs @@ -63,7 +63,6 @@ use dagx::{task, DagRunner}; - struct FetchName; #[task] diff --git a/examples/04_parallel_computation.rs b/examples/04_parallel_computation.rs index 65fc54a..1c57389 100644 --- a/examples/04_parallel_computation.rs +++ b/examples/04_parallel_computation.rs @@ -125,7 +125,8 @@ async fn main() -> DagResult<()> { let total = dag.add_task(Aggregate).depends_on((sum1, sum2, sum3, sum4)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(total)?; let elapsed = start.elapsed(); diff --git a/examples/complex_dag.rs b/examples/complex_dag.rs index b5fff8c..4ac8aad 100644 --- a/examples/complex_dag.rs +++ b/examples/complex_dag.rs @@ -68,7 +68,6 @@ use dagx::{task, DagResult, DagRunner}; - // Extract: Load data from different sources struct LoadData { source: &'static str, @@ -202,7 +201,8 @@ async fn main() -> DagResult<()> { // Execute the entire ETL pipeline println!("Executing pipeline...\n"); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Display final report println!("\n{}", dag.get(report)?); diff --git a/examples/conditional_workflow.rs b/examples/conditional_workflow.rs index c47e4a6..74b98ea 100644 --- a/examples/conditional_workflow.rs +++ b/examples/conditional_workflow.rs @@ -57,7 +57,6 @@ use dagx::{task, DagResult, DagRunner}; - // Task that validates and processes input struct ValidateAndDouble { value: i32, @@ -138,7 +137,8 @@ async fn main() -> DagResult<()> { .depends_on((&cat1, &cat2, &cat3)); // Execute - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Display results println!("\nFinal summary: {}", dag.get(summary)?); diff --git a/examples/custom_types.rs b/examples/custom_types.rs index d61743c..4c3d8ed 100644 --- a/examples/custom_types.rs +++ b/examples/custom_types.rs @@ -36,7 +36,6 @@ use dagx::{task, DagRunner, TaskHandle}; - #[derive(Debug)] struct User { name: String, diff --git a/examples/data_pipeline.rs b/examples/data_pipeline.rs index c290ff1..67cc0ce 100644 --- a/examples/data_pipeline.rs +++ b/examples/data_pipeline.rs @@ -77,7 +77,6 @@ use dagx::{task, DagRunner}; - // Data source tasks struct LoadValue1; diff --git a/examples/error_handling.rs b/examples/error_handling.rs index 3290a40..c4dde6b 100644 --- a/examples/error_handling.rs +++ b/examples/error_handling.rs @@ -116,7 +116,6 @@ use dagx::{task, DagRunner}; - // Source task that provides a string input struct StringSource(&'static str); diff --git a/tests/boundaries/deep_chains.rs b/tests/boundaries/deep_chains.rs index 1a4c9fe..8ef1dde 100644 --- a/tests/boundaries/deep_chains.rs +++ b/tests/boundaries/deep_chains.rs @@ -20,7 +20,8 @@ async fn test_1000_level_deep_chain() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(current)?, 999); Ok(()) @@ -50,7 +51,8 @@ async fn test_deep_chain_with_branches() -> DagResult<()> { } } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(main_chain)?, 499); @@ -108,7 +110,8 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> { &chain_ends[7], )); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Each chain: start + 500 // Chain 0: 0 + 500 = 500 @@ -149,7 +152,8 @@ async fn test_deep_chain_execution_order() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(order.load(Ordering::SeqCst), 100); assert_eq!(dag.get(current)?, 99); @@ -179,7 +183,8 @@ async fn test_deep_binary_tree() -> DagResult<()> { let root = build_tree(&dag, 10, 1); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + 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 @@ -209,7 +214,8 @@ async fn test_fibonacci_chain() -> DagResult<()> { prev1 = next; } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // 50th Fibonacci number (0-indexed) = 7778742049 assert_eq!(dag.get(prev1)?, 7778742049); @@ -232,7 +238,8 @@ async fn test_deep_chain_with_accumulator() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(current)?; assert_eq!(result, 20100); // sum of 1..=200 @@ -270,7 +277,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(final_task)?; assert!(result > 100); @@ -293,7 +301,8 @@ async fn test_10000_level_chain_stress() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(current)?, 9_999); Ok(()) diff --git a/tests/boundaries/large_scale.rs b/tests/boundaries/large_scale.rs index ab75d8f..9d092a2 100644 --- a/tests/boundaries/large_scale.rs +++ b/tests/boundaries/large_scale.rs @@ -23,7 +23,8 @@ async fn test_10000_independent_tasks() -> DagResult<()> { }) .collect(); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify all tasks completed assert_eq!(counter.load(Ordering::Relaxed), 10_000); @@ -100,7 +101,8 @@ async fn test_5000_node_pyramid() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) .depends_on((¤t_layer[0], ¤t_layer[1])); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + 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 @@ -209,7 +211,8 @@ async fn test_wide_dag_1000_sources_1000_sinks() -> DagResult<()> { }) .collect(); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify some sinks completed correctly let result0 = dag.get(sinks[0])?; @@ -258,7 +261,8 @@ async fn test_10000_node_linear_chain_segments() -> DagResult<()> { chain_ends.push(current); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + 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); @@ -297,7 +301,8 @@ async fn test_100000_nodes_stress() -> DagResult<()> { .collect(); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(completed.load(Ordering::Relaxed), 100_000); Ok(()) diff --git a/tests/boundaries/type_limits.rs b/tests/boundaries/type_limits.rs index 1e3a544..3cae04b 100644 --- a/tests/boundaries/type_limits.rs +++ b/tests/boundaries/type_limits.rs @@ -37,7 +37,8 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> { })) .depends_on(t2); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(t1)?, Empty); assert_eq!(dag.get(t2)?, 42); @@ -72,7 +73,8 @@ async fn test_large_value_types() -> DagResult<()> { })) .depends_on(t1); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(t2)?, 1042); @@ -95,7 +97,8 @@ async fn test_unit_type_chains() -> DagResult<()> { .add_task(task_fn::<(), _, _>(|_: ()| "done")) .depends_on((t3,)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; dag.get(t1)?; // Verify no error dag.get(t2)?; @@ -145,7 +148,8 @@ async fn test_string_types_variety() -> DagResult<()> { )) .depends_on((t1, t2, t3, t4, t5, t6)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(combined)?; assert!(result.contains("static str")); @@ -178,7 +182,8 @@ async fn test_generic_type_constraints() -> DagResult<()> { )) .depends_on((t1, t2, t3)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; let result = dag.get(combined)?; assert!(result.contains("Wrapper(42)")); @@ -204,7 +209,8 @@ async fn test_reference_wrapper_types() -> DagResult<()> { .add_task(task_fn::(|&val: &i32| val * 2)) .depends_on(t1); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(t2)?, 200); @@ -231,7 +237,8 @@ async fn test_array_types() -> DagResult<()> { )) .depends_on((t1, t2)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(t3)?, 55); @@ -270,7 +277,8 @@ async fn test_enum_variants() -> DagResult<()> { )) .depends_on((t1, t2, t3, t4)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert!(dag.get(combined)?); diff --git a/tests/boundaries/zero_tasks.rs b/tests/boundaries/zero_tasks.rs index f884d8d..d17dddc 100644 --- a/tests/boundaries/zero_tasks.rs +++ b/tests/boundaries/zero_tasks.rs @@ -12,7 +12,8 @@ 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?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Should complete successfully with no tasks Ok(()) @@ -25,7 +26,8 @@ async fn test_single_task_no_deps() -> DagResult<()> { let task = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(task)?, 42); Ok(()) @@ -44,7 +46,8 @@ async fn test_single_task_unit_input_output() -> DagResult<()> { exec.fetch_add(1, Ordering::SeqCst); })); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + 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); @@ -61,7 +64,8 @@ async fn test_zero_sized_type_task() -> DagResult<()> { let task = dag.add_task(task_fn::<(), _, _>(|_: ()| ZeroSized)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(task)?, ZeroSized); Ok(()) @@ -73,7 +77,8 @@ async fn test_empty_dag_multiple_runs() -> DagResult<()> { let dag = DagRunner::new(); for _ in 0..10 { - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; } Ok(()) @@ -88,7 +93,8 @@ async fn test_dag_with_only_source_tasks() -> DagResult<()> { .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i))) .collect(); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + 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); @@ -114,7 +120,8 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> { let dag = DagRunner::new(); let task = dag.add_task(Counter { value: 21 }); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(task)?, 42); Ok(()) @@ -128,11 +135,13 @@ async fn test_alternating_empty_and_filled_dag_runs() -> DagResult<()> { if i % 2 == 0 { // Empty DAG - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + 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?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(task)?, i); } } @@ -152,7 +161,8 @@ 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::<(), _, _>(|_: ()| Result::::Ok(42))); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(task)?, Ok(42)); Ok(()) @@ -175,7 +185,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(task)?, 1337); Ok(()) diff --git a/tests/dependencies/resolution.rs b/tests/dependencies/resolution.rs index 5f70e5c..3200e97 100644 --- a/tests/dependencies/resolution.rs +++ b/tests/dependencies/resolution.rs @@ -79,7 +79,8 @@ async fn test_dependency_resolution_order() -> DagResult<()> { .depends_on(d) }; - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(e)?, 5); @@ -113,7 +114,8 @@ async fn test_transitive_dependencies() -> DagResult<()> { .add_task(task_fn::(|&x: &i32| x + 5)) .depends_on(b); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(a)?, 10); assert_eq!(dag.get(b)?, 20); @@ -136,7 +138,8 @@ async fn test_deep_chain_50_levels() -> DagResult<()> { handles.push(next); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(handles.last().unwrap())?, 50); Ok(()) @@ -155,7 +158,8 @@ async fn test_deep_chain_200_levels() -> DagResult<()> { .depends_on(current); } - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(current)?, 201); Ok(()) diff --git a/tests/dependencies/single.rs b/tests/dependencies/single.rs index 290e88f..d728dd2 100644 --- a/tests/dependencies/single.rs +++ b/tests/dependencies/single.rs @@ -3,7 +3,6 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; - #[tokio::test] async fn test_single_dependency() -> DagResult<()> { let dag = DagRunner::new(); @@ -13,7 +12,8 @@ async fn test_single_dependency() -> DagResult<()> { .add_task(task_fn::(|&x: &i32| x + 1)) .depends_on(source); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(dependent)?, 43); Ok(()) @@ -44,7 +44,8 @@ async fn test_dependency_data_flow() -> DagResult<()> { .add_task(task_fn::<(i32, i32), _, _>(|(s, p): (&i32, &i32)| s + p)) .depends_on((&sum, &product)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Note: sum and product are not sinks (final_result depends on them) // Only final_result is a sink and can be retrieved @@ -68,7 +69,8 @@ async fn test_dependencies_with_different_types() -> DagResult<()> { )) .depends_on((int_source, string_source, bool_source)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(combined)?, "hello: 42 (true)"); Ok(()) @@ -89,7 +91,8 @@ async fn test_shared_dependencies() -> DagResult<()> { }) .collect(); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify all dependents got the correct value from shared for (i, handle) in dependents.iter().enumerate() { @@ -108,7 +111,8 @@ async fn test_multiple_source_nodes() -> DagResult<()> { .map(|i| dag.add_task(task_fn::<(), _, _>(move |_: ()| i * 10))) .collect(); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; for (i, source) in sources.into_iter().enumerate() { assert_eq!(dag.get(source)?, i as i32 * 10); diff --git a/tests/dependencies/tuples.rs b/tests/dependencies/tuples.rs index 9d40635..1b4f141 100644 --- a/tests/dependencies/tuples.rs +++ b/tests/dependencies/tuples.rs @@ -3,7 +3,6 @@ use crate::common::task_fn; use dagx::{DagResult, DagRunner}; - #[tokio::test] async fn test_two_dependencies() -> DagResult<()> { let dag = DagRunner::new(); @@ -14,7 +13,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 30); Ok(()) @@ -33,7 +33,8 @@ async fn test_three_dependencies() -> DagResult<()> { )) .depends_on((a, b, c)); - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 6); Ok(()) @@ -53,7 +54,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 10); // 1+2+3+4 Ok(()) @@ -73,7 +75,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 15); // 1+2+3+4+5 Ok(()) @@ -93,7 +96,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 21); // 1+2+3+4+5+6 Ok(()) @@ -117,7 +121,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 28); // 1+2+3+4+5+6+7 Ok(()) @@ -141,7 +146,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 36); // Sum of 1..=8 Ok(()) diff --git a/tests/parallelism/concurrent.rs b/tests/parallelism/concurrent.rs index 64b0cc4..d7d70b1 100644 --- a/tests/parallelism/concurrent.rs +++ b/tests/parallelism/concurrent.rs @@ -31,7 +31,10 @@ async fn test_concurrent_run_calls() { for _ in 0..5 { let dag = Arc::clone(&dag); - handles.spawn(async move { dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await }); + 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 diff --git a/tests/parallelism/execution.rs b/tests/parallelism/execution.rs index 6c25331..c2f1cc0 100644 --- a/tests/parallelism/execution.rs +++ b/tests/parallelism/execution.rs @@ -87,7 +87,8 @@ 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() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(final_task)?, 33); // (0+1) + (10+1) + (20+1) @@ -213,7 +214,8 @@ async fn test_completion_order_matches_dependencies() -> DagResult<()> { .depends_on(c) }; - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(d)?, 6); // (1 + 2) * 2 @@ -313,7 +315,8 @@ async fn test_dependent_tasks_respect_ordering() -> DagResult<()> { .depends_on(c) }; - dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; // Verify results assert_eq!(dag.get(a)?, 1); diff --git a/tests/runtimes/compatibility.rs b/tests/runtimes/compatibility.rs index 5073235..4a0b3f1 100644 --- a/tests/runtimes/compatibility.rs +++ b/tests/runtimes/compatibility.rs @@ -2,7 +2,6 @@ use dagx::{DagResult, DagRunner}; - struct Value(i32); #[dagx::task] impl Value { @@ -27,7 +26,8 @@ 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| async move { tokio::spawn(fut).await.unwrap() }).await?; + dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) + .await?; assert_eq!(dag.get(sum)?, 30); Ok(())