diff --git a/README.md b/README.md index 4ffc2f7..dabd9e6 100644 --- a/README.md +++ b/README.md @@ -79,7 +79,7 @@ async fn main() { let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); // Retrieve results - assert_eq!(output.get(sum).unwrap(), 5); + assert_eq!(output.get(sum), 5); } ``` diff --git a/benches/comparisons/deep_chain.rs b/benches/comparisons/deep_chain.rs index 10f8101..e365831 100644 --- a/benches/comparisons/deep_chain.rs +++ b/benches/comparisons/deep_chain.rs @@ -20,19 +20,19 @@ pub fn bench_deep_chain(c: &mut Criterion) { let first = dag.add_task(task_fn::<(), _, _>(|_: ()| 0)); let mut prev = dag .add_task(task_fn::(|&x: &i32| x + 1)) - .depends_on(first); + .depends_on(&first); for i in 2..100 { prev = dag .add_task(task_fn::(move |&x: &i32| x + i)) - .depends_on(prev); + .depends_on(&prev); } let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - output.get(prev).unwrap() + output.get(prev) }) }); }); diff --git a/benches/comparisons/diamond_pattern.rs b/benches/comparisons/diamond_pattern.rs index a6b7bc4..2597dee 100644 --- a/benches/comparisons/diamond_pattern.rs +++ b/benches/comparisons/diamond_pattern.rs @@ -20,19 +20,19 @@ pub fn bench_diamond_pattern(c: &mut Criterion) { let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)); let b = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(a); + .depends_on(&a); let c = dag .add_task(task_fn::(|&x: &i32| x + 5)) - .depends_on(a); + .depends_on(&a); let d = dag .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y)) - .depends_on((b, c)); + .depends_on((&b, &c)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - output.get(d).unwrap() + output.get(d) }) }); }); diff --git a/benches/comparisons/etl_pipeline.rs b/benches/comparisons/etl_pipeline.rs index 766630e..90b601e 100644 --- a/benches/comparisons/etl_pipeline.rs +++ b/benches/comparisons/etl_pipeline.rs @@ -27,19 +27,19 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { .add_task(task_fn::, _, _>(|data: &Vec| { data.iter().map(|x| x * 2).collect::>() })) - .depends_on(source1); + .depends_on(&source1); let transform2 = dag .add_task(task_fn::, _, _>(|data: &Vec| { data.iter().map(|x| x * 2).collect::>() })) - .depends_on(source2); + .depends_on(&source2); let transform3 = dag .add_task(task_fn::, _, _>(|data: &Vec| { data.iter().map(|x| x * 2).collect::>() })) - .depends_on(source3); + .depends_on(&source3); // Merge all transformed data let merge = dag @@ -51,7 +51,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { result }, )) - .depends_on((transform1, transform2, transform3)); + .depends_on((&transform1, &transform2, &transform3)); // Validate let validate = dag @@ -61,20 +61,20 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { .filter(|&x| x > 0) .collect::>() })) - .depends_on(merge); + .depends_on(&merge); // Load (compute sum) let load = dag .add_task(task_fn::, _, _>(|data: &Vec| { data.iter().sum::() })) - .depends_on(validate); + .depends_on(&validate); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - output.get(load).unwrap() + output.get(load) }) }); }); diff --git a/benches/comparisons/linear_pipeline.rs b/benches/comparisons/linear_pipeline.rs index 3cf1a90..98733f9 100644 --- a/benches/comparisons/linear_pipeline.rs +++ b/benches/comparisons/linear_pipeline.rs @@ -20,22 +20,22 @@ pub fn bench_linear_pipeline(c: &mut Criterion) { let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 1)); let b = dag .add_task(task_fn::(|&x: &i32| x + 1)) - .depends_on(a); + .depends_on(&a); let c = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(b); + .depends_on(&b); let d = dag .add_task(task_fn::(|&x: &i32| x - 1)) - .depends_on(c); + .depends_on(&c); let e = dag .add_task(task_fn::(|&x: &i32| x * 3)) - .depends_on(d); + .depends_on(&d); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - output.get(e).unwrap() + output.get(e) }) }); }); diff --git a/benches/comparisons/wide_fanout.rs b/benches/comparisons/wide_fanout.rs index b43dcad..442c130 100644 --- a/benches/comparisons/wide_fanout.rs +++ b/benches/comparisons/wide_fanout.rs @@ -21,7 +21,7 @@ pub fn bench_wide_fanout(c: &mut Criterion) { // Create 100 tasks that all depend on source for i in 0..100 { dag.add_task(task_fn::(move |&x: &i32| x + i)) - .depends_on(source); + .depends_on(&source); } let _output = dag diff --git a/benches/patterns/diamond.rs b/benches/patterns/diamond.rs index e3685de..99756e4 100644 --- a/benches/patterns/diamond.rs +++ b/benches/patterns/diamond.rs @@ -16,13 +16,13 @@ pub fn bench_diamond(c: &mut Criterion) { let left = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(source); + .depends_on(&source); let right = dag .add_task(task_fn::(|&x: &i32| x * 3)) - .depends_on(source); + .depends_on(&source); dag.add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r)) - .depends_on((left, right)); + .depends_on((&left, &right)); } let _output = dag diff --git a/benches/patterns/fanout.rs b/benches/patterns/fanout.rs index e411a9e..e9fee19 100644 --- a/benches/patterns/fanout.rs +++ b/benches/patterns/fanout.rs @@ -15,7 +15,7 @@ pub fn bench_fanout(c: &mut Criterion) { for i in 0..100 { dag.add_task(task_fn::(move |&x: &i32| x + i)) - .depends_on(source); + .depends_on(&source); } let _output = dag @@ -41,7 +41,7 @@ pub fn bench_fanout(c: &mut Criterion) { // Each consumer gets data extracted from Arc data.iter().filter(|s| s.contains(&i.to_string())).count() })) - .depends_on(source); + .depends_on(&source); } let _output = dag diff --git a/benches/realistic/etl.rs b/benches/realistic/etl.rs index 00f63f2..9b35e63 100644 --- a/benches/realistic/etl.rs +++ b/benches/realistic/etl.rs @@ -46,7 +46,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { }) .count() })) - .depends_on(extract); + .depends_on(&extract); // T2: Extract scores let scores = dag @@ -59,7 +59,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { }) .collect::>() })) - .depends_on(extract); + .depends_on(&extract); // T3: Extract names let names = dag @@ -73,7 +73,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { }) .collect::>() })) - .depends_on(extract); + .depends_on(&extract); // Load: Aggregate all transformations dag.add_task(task_fn::<(usize, Vec<_>, Vec<_>), _, _>( @@ -88,7 +88,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { ) }, )) - .depends_on((validate, scores, names)); + .depends_on((&validate, &scores, &names)); let _output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) diff --git a/benches/realistic/memory_efficiency.rs b/benches/realistic/memory_efficiency.rs index 3ea0e35..949330c 100644 --- a/benches/realistic/memory_efficiency.rs +++ b/benches/realistic/memory_efficiency.rs @@ -36,7 +36,7 @@ pub fn bench_memory_efficiency(c: &mut Criterion) { .sum(); black_box(sample_sum) })) - .depends_on(source); + .depends_on(&source); } let _output = dag diff --git a/benches/realistic/mixed.rs b/benches/realistic/mixed.rs index 7b55ba9..7ac5f24 100644 --- a/benches/realistic/mixed.rs +++ b/benches/realistic/mixed.rs @@ -23,17 +23,17 @@ pub fn bench_mixed_patterns(c: &mut Criterion) { .add_task(task_fn::, _, _>(|d: &Vec| { d.iter().filter(|s| s.contains("00")).count() })) - .depends_on(data); + .depends_on(&data); let analysis2 = dag .add_task(task_fn::, _, _>(|d: &Vec| { d.iter().map(|s| s.len()).sum::() })) - .depends_on(data); + .depends_on(&data); let analysis3 = dag .add_task(task_fn::, _, _>(|d: &Vec| d.len())) - .depends_on(data); + .depends_on(&data); // Stage 3: Fan-in to aggregate let summary = dag @@ -42,14 +42,14 @@ pub fn bench_mixed_patterns(c: &mut Criterion) { format!("Matches: {}, Total bytes: {}, Count: {}", a1, a2, a3) }, )) - .depends_on((analysis1, analysis2, analysis3)); + .depends_on((&analysis1, &analysis2, &analysis3)); // Stage 4: Fan-out final report to multiple destinations for i in 0..3 { dag.add_task(task_fn::(move |report: &String| { format!("Destination {}: {}", i, report) })) - .depends_on(summary); + .depends_on(&summary); } let _output = dag diff --git a/docs/CYCLE_PREVENTION.md b/docs/CYCLE_PREVENTION.md index a233770..0b00a7a 100644 --- a/docs/CYCLE_PREVENTION.md +++ b/docs/CYCLE_PREVENTION.md @@ -11,8 +11,8 @@ Most DAG libraries check for cycles when you execute the graph: let mut dag = DagRunner::new(); let a = dag.add_task(TaskA); let b = dag.add_task(TaskB); -a.depends_on(b); -b.depends_on(a); // Compiles fine... +a.depends_on(&b); +b.depends_on(&a); // Compiles fine... dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) ``` @@ -26,12 +26,12 @@ let b_builder = dag.add_task(TaskB); // To make A depend on B, we need B as a TaskHandle. // But calling depends_on() consumes the builder! -let b_handle = b_builder.depends_on(some_source); +let b_handle = b_builder.depends_on(&some_source); // Now we can't make B depend on A because b_builder was moved! // This won't compile: -// a_builder.depends_on(b_handle); // ✓ OK: A→B -// b_handle.depends_on(a_handle); // ❌ ERROR: TaskHandle has no depends_on() method! +// a_builder.depends_on(&b_handle); // ✓ OK: A→B +// b_handle.depends_on(&a_handle); // ❌ ERROR: TaskHandle has no depends_on() method! ``` ## How It Works: The Type-State Pattern @@ -58,7 +58,7 @@ let b_builder = dag.add_task(TaskB); // TaskBuilder let b_handle = b_builder.into(); // Convert B to TaskHandle (consumes builder) // To wire B→A, we need A as a TaskHandle -let a_handle = a_builder.depends_on(b_handle); // A now depends on B +let a_handle = a_builder.depends_on(&b_handle); // A now depends on B // Now if we try to complete the cycle: // a_handle.depends_on(...); // ❌ ERROR: no method named `depends_on` found diff --git a/examples/01_basic.rs b/examples/01_basic.rs index 8044ede..93f69b3 100644 --- a/examples/01_basic.rs +++ b/examples/01_basic.rs @@ -93,8 +93,8 @@ async fn main() { .add_task(Add { label: "Sum".to_string(), }) - .depends_on((x, y)); // Pass handles as a tuple - // sum is now TaskHandle + .depends_on((&x, &y)); // Pass handles as a tuple + // sum is now TaskHandle // Step 6: Execute the DAG // @@ -114,9 +114,9 @@ async fn main() { // // Use .get() with a TaskHandle to retrieve the output. // Results are not cloned, so you can only retrieve them once. - println!("\nResult: {}", output.get(sum).unwrap()); + println!("\nResult: {}", output.get(sum)); // Note: You can also retrieve intermediate results - assert_eq!(output.get(x).unwrap(), 2); - assert_eq!(output.get(y).unwrap(), 3); + assert_eq!(output.get(x), 2); + assert_eq!(output.get(y), 3); } diff --git a/examples/02_fan_out.rs b/examples/02_fan_out.rs index 25513cb..ab44525 100644 --- a/examples/02_fan_out.rs +++ b/examples/02_fan_out.rs @@ -134,9 +134,9 @@ async fn main() { // Multiple downstream tasks consuming the same value // Each constructed differently to show flexibility - let plus_one = dag.add_task(AddOffset::new(1)).depends_on(source); - let times_two = dag.add_task(Multiply::new(2)).depends_on(source); - let squared = dag.add_task(Power { exponent: 2 }).depends_on(source); + let plus_one = dag.add_task(AddOffset::new(1)).depends_on(&source); + let times_two = dag.add_task(Multiply::new(2)).depends_on(&source); + let squared = dag.add_task(Power { exponent: 2 }).depends_on(&source); // Run the DAG println!("Running fan-out DAG...\n"); @@ -147,8 +147,8 @@ async fn main() { // Print all results println!("\nResults:"); - println!(" Source: {}", output.get(source).unwrap()); - println!(" Plus One: {}", output.get(plus_one).unwrap()); - println!(" Times Two: {}", output.get(times_two).unwrap()); - println!(" Squared: {}", output.get(squared).unwrap()); + println!(" Source: {}", output.get(source)); + println!(" Plus One: {}", output.get(plus_one)); + println!(" Times Two: {}", output.get(times_two)); + println!(" Squared: {}", output.get(squared)); } diff --git a/examples/03_fan_in.rs b/examples/03_fan_in.rs index ad915e5..86f5401 100644 --- a/examples/03_fan_in.rs +++ b/examples/03_fan_in.rs @@ -29,7 +29,7 @@ //! ## Key Concepts //! - **Fan-in**: Multiple tasks' outputs are consumed by a single downstream task //! - **Aggregation**: Combining multiple values into a single result -//! - **Parameter order**: The order in `.depends_on((a, b, c))` must match +//! - **Parameter order**: The order in `.depends_on((&a, &b, &c))` must match //! the function signature `fn run(a: &T1, b: &T2, c: &T3)` //! //! ## Use Cases @@ -129,7 +129,7 @@ async fn main() { // Single downstream task consuming all values let profile = dag .add_task(BuildProfile) - .depends_on((name, age, city, active)); + .depends_on((&name, &age, &city, &active)); // Run the DAG println!("Running fan-in DAG...\n"); @@ -139,5 +139,5 @@ async fn main() { .unwrap(); // Print result - println!("\n{}", output.get(profile).unwrap()); + println!("\n{}", output.get(profile)); } diff --git a/examples/04_parallel_computation.rs b/examples/04_parallel_computation.rs index 13f9527..45864a8 100644 --- a/examples/04_parallel_computation.rs +++ b/examples/04_parallel_computation.rs @@ -123,13 +123,15 @@ async fn main() -> DagResult<()> { let sum3 = dag.add_task(ComputeSum::new(501, 751, WORK_DELAY)); let sum4 = dag.add_task(ComputeSum::new(751, 1001, WORK_DELAY)); - let total = dag.add_task(Aggregate).depends_on((sum1, sum2, sum3, sum4)); + let total = dag + .add_task(Aggregate) + .depends_on((&sum1, &sum2, &sum3, &sum4)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - let result = output.get(total)?; + let result = output.get(total); let elapsed = start.elapsed(); println!( diff --git a/examples/circuit_breaker.rs b/examples/circuit_breaker.rs index 1c65b1e..fc27646 100644 --- a/examples/circuit_breaker.rs +++ b/examples/circuit_breaker.rs @@ -395,7 +395,7 @@ async fn main() { .await .unwrap(); - println!("Result 1: {:?}\n", output.get(call1).unwrap()); + println!("Result 1: {:?}\n", output.get(call1)); // Request 2: Failure (circuit closed, count = 1) let mut dag = DagRunner::new(); @@ -410,7 +410,7 @@ async fn main() { .await .unwrap(); - println!("Result 2: {:?}\n", output.get(call2).unwrap()); + println!("Result 2: {:?}\n", output.get(call2)); // Request 3: Failure (circuit closed, count = 2) let mut dag = DagRunner::new(); @@ -425,7 +425,7 @@ async fn main() { .await .unwrap(); - println!("Result 3: {:?}\n", output.get(call3).unwrap()); + println!("Result 3: {:?}\n", output.get(call3)); // Request 4: Failure (circuit closed, count = 3, opens!) let mut dag = DagRunner::new(); @@ -440,7 +440,7 @@ async fn main() { .await .unwrap(); - println!("Result 4: {:?}\n", output.get(call4).unwrap()); + println!("Result 4: {:?}\n", output.get(call4)); // Request 5: Circuit is now open, fails fast with fallback let mut dag = DagRunner::new(); @@ -452,14 +452,14 @@ async fn main() { let fallback5 = dag .add_task(FallbackService::new("ServiceCall-5")) - .depends_on(call5); + .depends_on(&call5); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - println!("Result 5: {:?}\n", output.get(fallback5).unwrap()); + println!("Result 5: {:?}\n", output.get(fallback5)); // Example 2: Circuit recovery after timeout println!("Example 2: Circuit recovery after timeout\n"); @@ -479,7 +479,7 @@ async fn main() { .await .unwrap(); - println!("Result 6: {:?}\n", output.get(call6).unwrap()); + println!("Result 6: {:?}\n", output.get(call6)); // Request 7: Circuit is now closed again, normal operation let mut dag = DagRunner::new(); @@ -494,7 +494,7 @@ async fn main() { .await .unwrap(); - println!("Result 7: {:?}\n", output.get(call7).unwrap()); + println!("Result 7: {:?}\n", output.get(call7)); } println!("=== Circuit Breaker Example Complete ==="); diff --git a/examples/data_pipeline.rs b/examples/data_pipeline.rs index ab38b17..cec71ab 100644 --- a/examples/data_pipeline.rs +++ b/examples/data_pipeline.rs @@ -190,31 +190,33 @@ async fn main() { // Stage 2: Data transformation (with stateful tasks) let transformed_1 = dag .add_task(DataTransformer::new(1.5)) - .depends_on(raw_value_1); + .depends_on(&raw_value_1); let transformed_2 = dag .add_task(DataTransformer::new(2.0)) - .depends_on(raw_value_2); + .depends_on(&raw_value_2); let transformed_3 = dag .add_task(DataTransformer::new(0.5)) - .depends_on(raw_value_3); + .depends_on(&raw_value_3); // Stage 3: Statistical analysis (parallel operations on transformed data) let sum = dag .add_task(ComputeSum) - .depends_on((transformed_1, transformed_2, transformed_3)); + .depends_on((&transformed_1, &transformed_2, &transformed_3)); let product = dag.add_task(ComputeProduct) - .depends_on((transformed_1, transformed_2, transformed_3)); + .depends_on((&transformed_1, &transformed_2, &transformed_3)); let max = dag .add_task(FindMax) - .depends_on((transformed_1, transformed_2, transformed_3)); + .depends_on((&transformed_1, &transformed_2, &transformed_3)); // Stage 4: Final report - let report = dag.add_task(GenerateReport).depends_on((sum, product, max)); + let report = dag + .add_task(GenerateReport) + .depends_on((&sum, &product, &max)); // Execute the pipeline println!("Executing pipeline...\n"); @@ -224,5 +226,5 @@ async fn main() { .unwrap(); // Print final report - println!("{}", output.get(report).unwrap()); + println!("{}", output.get(report)); } diff --git a/examples/error_handling.rs b/examples/error_handling.rs index 67cea84..8d4ab77 100644 --- a/examples/error_handling.rs +++ b/examples/error_handling.rs @@ -213,21 +213,21 @@ async fn main() { let input = dag.add_task(StringSource("42")); - let parsed = dag.add_task(ParseInt).depends_on(input); + let parsed = dag.add_task(ParseInt).depends_on(&input); let validated = dag .add_task(ValidateRange { min: 0, max: 100 }) - .depends_on(parsed); + .depends_on(&parsed); - let processed = dag.add_task(ProcessData).depends_on(validated); - let logged = dag.add_task(LogError).depends_on(processed); + let processed = dag.add_task(ProcessData).depends_on(&validated); + let logged = dag.add_task(LogError).depends_on(&processed); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - match output.get(logged).unwrap().as_ref() { + match output.get(logged).as_ref() { Ok(result) => println!("Final result: {}\n", result), Err(e) => eprintln!("Final error: {}\n", e), } @@ -240,21 +240,21 @@ async fn main() { let input = dag.add_task(StringSource("not a number")); - let parsed = dag.add_task(ParseInt).depends_on(input); + let parsed = dag.add_task(ParseInt).depends_on(&input); let validated = dag .add_task(ValidateRange { min: 0, max: 100 }) - .depends_on(parsed); + .depends_on(&parsed); - let processed = dag.add_task(ProcessData).depends_on(validated); - let logged = dag.add_task(LogError).depends_on(processed); + let processed = dag.add_task(ProcessData).depends_on(&validated); + let logged = dag.add_task(LogError).depends_on(&processed); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - match output.get(logged).unwrap().as_ref() { + match output.get(logged).as_ref() { Ok(result) => println!("Final result: {}\n", result), Err(e) => eprintln!("Final error: {}\n", e), } @@ -267,21 +267,21 @@ async fn main() { let input = dag.add_task(StringSource("150")); - let parsed = dag.add_task(ParseInt).depends_on(input); + let parsed = dag.add_task(ParseInt).depends_on(&input); let validated = dag .add_task(ValidateRange { min: 0, max: 100 }) - .depends_on(parsed); + .depends_on(&parsed); - let processed = dag.add_task(ProcessData).depends_on(validated); - let logged = dag.add_task(LogError).depends_on(processed); + let processed = dag.add_task(ProcessData).depends_on(&validated); + let logged = dag.add_task(LogError).depends_on(&processed); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - match output.get(logged).unwrap().as_ref() { + match output.get(logged).as_ref() { Ok(result) => println!("Final result: {}\n", result), Err(e) => eprintln!("Final error: {}\n", e), } @@ -294,18 +294,18 @@ async fn main() { let input = dag.add_task(StringSource("invalid")); - let parsed = dag.add_task(ParseInt).depends_on(input); + let parsed = dag.add_task(ParseInt).depends_on(&input); let with_fallback = dag .add_task(WithFallback { fallback: 0 }) - .depends_on(parsed); + .depends_on(&parsed); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - let result = output.get(with_fallback).unwrap(); + let result = output.get(with_fallback); println!("Recovered with fallback: {}\n", result); } diff --git a/src/builder.rs b/src/builder.rs index 71e5f3e..0fd475d 100644 --- a/src/builder.rs +++ b/src/builder.rs @@ -50,11 +50,11 @@ use tracing::debug; /// let b = dag.add_task(Multiplier::new(2)); /// // b is TaskBuilder until we call depends_on() /// -/// let b = b.depends_on(a); +/// let b = b.depends_on(&a); /// // Now b is a TaskHandle /// ///let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); -/// assert_eq!(output.get(b).unwrap(), 20); +/// assert_eq!(output.get(b), 20); /// # }; /// ``` #[must_use] @@ -76,11 +76,16 @@ where /// Provide all dependencies exactly once as a tuple. /// /// The dependencies must match the task's `Input` type exactly: - /// - `Input = A`: Pass `TaskHandle` - /// - `Input = (A, B, ...)`: Pass `(TaskHandle, TaskHandle, ...)` + /// - `Input = A`: Pass `&TaskHandle` + /// - `Input = (A, B, ...)`: Pass `(&TaskHandle, &TaskHandle, ...)` /// /// The order of dependencies in the tuple must match the order in `Input`. /// + /// # Panics + /// + /// This function will cause a panic in [`DagRunner::run`] if called with a [`TaskHandle`] from a `DagRunner` instance + /// other than the one which created this TaskBuilder. + /// /// # Examples /// /// ```no_run @@ -117,10 +122,10 @@ where /// let y = dag.add_task(Value(3)); /// /// // Single dependency - /// let double = dag.add_task(Scale(2)).depends_on(x); + /// let double = dag.add_task(Scale(2)).depends_on(&x); /// /// // Multiple dependencies: tuple form - /// let sum = dag.add_task(Add).depends_on((x, y)); + /// let sum = dag.add_task(Add).depends_on((&x, &y)); /// ///let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// # }; @@ -188,7 +193,7 @@ pub struct NodeId(pub u32); /// ///let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// -/// assert_eq!(output.get(node).unwrap(), 42); +/// assert_eq!(output.get(node), 42); /// # }; /// ``` pub struct TaskHandle { @@ -196,14 +201,6 @@ pub struct TaskHandle { pub(crate) _phantom: PhantomData T>, } -impl Clone for TaskHandle { - fn clone(&self) -> Self { - *self - } -} - -impl Copy for TaskHandle {} - /// Takes a task and converts it to either a TaskBuilder or a TaskHandle, /// depending on whether it has inputs or not. /// @@ -235,7 +232,6 @@ where /// Macro to implement TaskWire for different tuple sizes. macro_rules! impl_wire_tuple { ($($T:ident),+) => { - // For tuples of any combination of &TaskHandle, TaskHandle, and TaskBuilder impl TaskWire<($($T,)+)> for Tk where Tk: Task<($($T,)+)> + Sync + 'static diff --git a/src/builder/coverage_tests.rs b/src/builder/coverage_tests.rs index e3e9565..ae74d0a 100644 --- a/src/builder/coverage_tests.rs +++ b/src/builder/coverage_tests.rs @@ -38,7 +38,7 @@ async fn test_unit_struct_with_state() { .await .unwrap(); - assert_eq!(output.get(handle).unwrap(), 42); + assert_eq!(output.get(handle), 42); } #[test] diff --git a/src/builder/tests.rs b/src/builder/tests.rs index 164a669..0c2e3d4 100644 --- a/src/builder/tests.rs +++ b/src/builder/tests.rs @@ -62,8 +62,8 @@ fn test_task_builder_chain() { // Test chaining multiple tasks let t1 = dag.add_task(TestTask { value: 1 }); - let t2 = dag.add_task(TestTaskWithInput).depends_on(t1); - let t3 = dag.add_task(TestTaskWithInput).depends_on(t2); + let t2 = dag.add_task(TestTaskWithInput).depends_on(&t1); + let t3 = dag.add_task(TestTaskWithInput).depends_on(&t2); // All should return TaskHandles let _ = t1; @@ -89,7 +89,7 @@ fn test_multiple_dependencies() { let b = dag.add_task(TestTask { value: 20 }); // Test multiple dependencies - let sum = dag.add_task(AddTask).depends_on((a, b)); + let sum = dag.add_task(AddTask).depends_on((&a, &b)); // Should return a TaskHandle let _ = sum; diff --git a/src/deps.rs b/src/deps.rs index e70e75c..fb13ef7 100644 --- a/src/deps.rs +++ b/src/deps.rs @@ -17,7 +17,7 @@ impl DepsTuple<()> for () { } // Implementation for single dependency (all forms) -impl DepsTuple<(O,)> for TaskHandle { +impl DepsTuple<(O,)> for &TaskHandle { fn to_node_ids(self) -> Vec { vec![self.id] } @@ -28,14 +28,14 @@ impl DepsTuple<(O,)> for TaskHandle { /// This allows flexible dependency specification: /// ```ignore /// // Using handles -/// task.depends_on((handle_a, handle_b)) +/// task.depends_on((&handle_a, &handle_b)) /// /// // Using builders directly /// task.depends_on((dag.add_task(TaskA), dag.add_task(TaskB))) /// ``` macro_rules! impl_deps_tuple { ($($O:ident),+) => { - impl<$($O),+> DepsTuple<($($O,)+)> for ($(TaskHandle<$O>,)+) + impl<$($O),+> DepsTuple<($($O,)+)> for ($(&TaskHandle<$O>,)+) { #[allow(non_snake_case)] fn to_node_ids(self) -> Vec { diff --git a/src/deps/tests.rs b/src/deps/tests.rs index d5ba1ff..cc788d5 100644 --- a/src/deps/tests.rs +++ b/src/deps/tests.rs @@ -37,19 +37,6 @@ fn test_deps_tuple_single_handle_ref() { assert_eq!(node_ids[0], NodeId(42)); } -#[test] -fn test_deps_tuple_single_handle_owned() { - // Test lines 32-36 in deps.rs - TaskHandle - let handle: TaskHandle = TaskHandle { - id: NodeId(99), - _phantom: std::marker::PhantomData, - }; - let handle_copy = handle; // Copy for use - let node_ids = handle_copy.to_node_ids(); - assert_eq!(node_ids.len(), 1); - assert_eq!(node_ids[0], NodeId(99)); -} - #[test] fn test_deps_tuple_single_handle_tuple_ref() { // Test lines 38-42 in deps.rs - (&TaskHandle,) @@ -57,24 +44,11 @@ fn test_deps_tuple_single_handle_tuple_ref() { id: NodeId(7), _phantom: std::marker::PhantomData, }; - let node_ids = (handle,).to_node_ids(); + let node_ids = (&handle,).to_node_ids(); assert_eq!(node_ids.len(), 1); assert_eq!(node_ids[0], NodeId(7)); } -#[test] -fn test_deps_tuple_single_handle_tuple_owned() { - // Test lines 44-48 in deps.rs - (TaskHandle,) - let handle: TaskHandle> = TaskHandle { - id: NodeId(13), - _phantom: std::marker::PhantomData, - }; - let handle_copy = handle; // Copy for use - let node_ids = (handle_copy,).to_node_ids(); - assert_eq!(node_ids.len(), 1); - assert_eq!(node_ids[0], NodeId(13)); -} - #[test] fn test_deps_tuple_task_builder_ref() { // Test lines 51-55 in deps.rs - &TaskBuilder @@ -96,7 +70,7 @@ fn test_deps_tuple_task_builder_tuple() { let builder = dag.add_task(TestTask { value: 100 }); let builder_id = builder.id; - let node_ids = (builder,).to_node_ids(); + let node_ids = (&builder,).to_node_ids(); assert_eq!(node_ids.len(), 1); assert_eq!(node_ids[0], builder_id); } @@ -114,7 +88,7 @@ fn test_deps_tuple_multiple_handles() { }; // Test 2-tuple - let node_ids = (handle1, handle2).to_node_ids(); + let node_ids = (&handle1, &handle2).to_node_ids(); assert_eq!(node_ids.len(), 2); assert_eq!(node_ids[0], NodeId(1)); assert_eq!(node_ids[1], NodeId(2)); @@ -124,32 +98,13 @@ fn test_deps_tuple_multiple_handles() { id: NodeId(3), _phantom: std::marker::PhantomData, }; - let node_ids = (handle1, handle2, handle3).to_node_ids(); + let node_ids = (&handle1, &handle2, &handle3).to_node_ids(); assert_eq!(node_ids.len(), 3); assert_eq!(node_ids[0], NodeId(1)); assert_eq!(node_ids[1], NodeId(2)); assert_eq!(node_ids[2], NodeId(3)); } -#[test] -fn test_deps_tuple_multiple_builders() { - // Test macro-generated implementations for multiple builders - let mut dag = DagRunner::new(); - let builder1 = dag.add_task(TestTask { value: 10 }); - let builder1_id = builder1.id; - let builder2 = dag.add_task(TestTask { value: 20 }); - let builder2_id = builder2.id; - let builder3 = dag.add_task(TestTask { value: 30 }); - let builder3_id = builder3.id; - - // Test 3-tuple of builders - let node_ids = (builder1, builder2, builder3).to_node_ids(); - assert_eq!(node_ids.len(), 3); - assert_eq!(node_ids[0], builder1_id); - assert_eq!(node_ids[1], builder2_id); - assert_eq!(node_ids[2], builder3_id); -} - #[test] fn test_deps_tuple_large_tuples() { // Test that larger tuples work (up to 16 elements) @@ -162,7 +117,7 @@ fn test_deps_tuple_large_tuples() { } // Test 4-tuple - let node_ids = (handles[0], handles[1], handles[2], handles[3]).to_node_ids(); + let node_ids = (&handles[0], &handles[1], &handles[2], &handles[3]).to_node_ids(); assert_eq!(node_ids.len(), 4); for (i, node_id) in node_ids.iter().enumerate().take(4) { assert_eq!(*node_id, NodeId(i as u32)); @@ -170,8 +125,14 @@ fn test_deps_tuple_large_tuples() { // Test 8-tuple let node_ids = ( - handles[0], handles[1], handles[2], handles[3], handles[4], handles[5], handles[6], - handles[7], + &handles[0], + &handles[1], + &handles[2], + &handles[3], + &handles[4], + &handles[5], + &handles[6], + &handles[7], ) .to_node_ids(); assert_eq!(node_ids.len(), 8); diff --git a/src/error.rs b/src/error.rs index 8bc24c4..5f4869c 100644 --- a/src/error.rs +++ b/src/error.rs @@ -6,42 +6,13 @@ #[derive(Debug, Clone, PartialEq)] #[non_exhaustive] pub enum DagError { - /// Invalid dependency: task does not exist - InvalidDependency { task_id: u32 }, - /// Type mismatch in task dependencies - TypeMismatch { - expected: &'static str, - found: &'static str, - }, /// Task panicked during execution TaskPanicked { task_id: u32, panic_message: String }, - /// Result not found for task - ResultNotFound { task_id: u32 }, } impl std::fmt::Display for DagError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - DagError::InvalidDependency { task_id } => { - write!( - f, - "Invalid dependency: task #{} does not exist.\n\ - \n\ - Ensure all task handles reference tasks that have been added to the DAG.", - task_id - ) - } - DagError::TypeMismatch { expected, found } => { - write!( - f, - "Type mismatch in task dependencies.\n\ - Expected: {}\n\ - Found: {}\n\ - \n\ - Verify that dependency types match the task's Input type exactly.", - expected, found - ) - } DagError::TaskPanicked { task_id, panic_message, @@ -54,15 +25,6 @@ impl std::fmt::Display for DagError { task_id, panic_message ) } - DagError::ResultNotFound { task_id } => { - write!( - f, - "Result not found for task #{}.\n\ - \n\ - Call dag.run(|fut| async move {{ tokio::spawn(fut).await.unwrap() }}).", - task_id - ) - } } } } diff --git a/src/error/tests.rs b/src/error/tests.rs index 3e9d680..d58283b 100644 --- a/src/error/tests.rs +++ b/src/error/tests.rs @@ -2,33 +2,6 @@ use crate::error::DagError; -#[test] -fn test_dag_error_display_invalid_dependency() { - // Test lines 47-54 in error.rs - let err = DagError::InvalidDependency { task_id: 42 }; - let display = format!("{}", err); - - assert!(display.contains("Invalid dependency")); - assert!(display.contains("task #42")); - assert!(display.contains("does not exist")); - assert!(display.contains("Ensure all task handles")); -} - -#[test] -fn test_dag_error_display_type_mismatch() { - // Test lines 56-65 in error.rs - let err = DagError::TypeMismatch { - expected: "i32", - found: "String", - }; - let display = format!("{}", err); - - assert!(display.contains("Type mismatch")); - assert!(display.contains("Expected: i32")); - assert!(display.contains("Found: String")); - assert!(display.contains("Verify that dependency types")); -} - #[test] fn test_dag_error_display_task_panicked() { // Test lines 67-77 in error.rs @@ -44,20 +17,13 @@ fn test_dag_error_display_task_panicked() { assert!(display.contains("entire DAG execution is aborted")); } -#[test] -fn test_dag_error_display_result_not_found() { - let err = DagError::ResultNotFound { task_id: 7 }; - let display = format!("{}", err); - - assert!(display.contains("Result not found")); - assert!(display.contains("task #7")); - assert!(display.contains("Call dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() })")); -} - #[test] fn test_dag_error_std_error_impl() { // Test that DagError implements std::error::Error - let err = DagError::InvalidDependency { task_id: 1 }; + let err = DagError::TaskPanicked { + task_id: 1, + panic_message: "test panic".to_string(), + }; let err_ref: &dyn std::error::Error = &err; // Should be able to call Error trait methods diff --git a/src/lib.rs b/src/lib.rs index 08972df..ad9d683 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -42,11 +42,11 @@ //! let y = dag.add_task(Value(3)); //! //! // Add task with dependencies -//! let sum = dag.add_task(Add).depends_on((x, y)); +//! let sum = dag.add_task(Add).depends_on((&x, &y)); //! //! // Execute and retrieve results //!let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); -//! assert_eq!(output.get(sum).unwrap(), 5); +//! assert_eq!(output.get(sum), 5); //! # }; //! ``` //! @@ -184,7 +184,7 @@ //! # async { //! # let mut dag = DagRunner::new(); //! # let source = dag.add_task(Value(42)); -//! let doubled = dag.add_task(Double).depends_on(source); +//! let doubled = dag.add_task(Double).depends_on(&source); //! # }; //! ``` //! @@ -210,7 +210,7 @@ //! # 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)); +//! let sum = dag.add_task(Add).depends_on((&x, &y)); //! # }; //! ``` //! @@ -252,13 +252,13 @@ //! let mut dag = DagRunner::new(); //! //! let base = dag.add_task(Value(10)); -//! let plus1 = dag.add_task(Add(1)).depends_on(base); -//! let times2 = dag.add_task(Scale(2)).depends_on(base); +//! let plus1 = dag.add_task(Add(1)).depends_on(&base); +//! let times2 = dag.add_task(Scale(2)).depends_on(&base); //! //!let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! -//! assert_eq!(output.get(plus1).unwrap(), 11); -//! assert_eq!(output.get(times2).unwrap(), 20); +//! assert_eq!(output.get(plus1), 11); +//! assert_eq!(output.get(times2), 20); //! # }; //! ``` //! @@ -310,11 +310,11 @@ //! let name = dag.add_task(Name("Alice".to_string())); //! let age = dag.add_task(Age(30)); //! let active = dag.add_task(Active(true)); -//! let result = dag.add_task(FormatUser).depends_on((name, age, active)); +//! let result = dag.add_task(FormatUser).depends_on((&name, &age, &active)); //! //!let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! -//! assert_eq!(output.get(result).unwrap(), "User: Alice, Age: 30, Active: true"); +//! assert_eq!(output.get(result), "User: Alice, Age: 30, Active: true"); //! # }; //! ``` //! @@ -368,7 +368,7 @@ //! # let node = dag.add_task(Value(42)); //! // Simple approach with .unwrap() //! let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); -//! let result = output.get(node).unwrap(); +//! let result = output.get(node); //! # let mut dag = DagRunner::new(); //! //! // Or handle errors explicitly @@ -428,9 +428,9 @@ //! let data = dag.add_task(FetchData); //! //! // All three tasks get efficient Arc-wrapped sharing automatically -//! let task1 = dag.add_task(ProcessData).depends_on(data); -//! let task2 = dag.add_task(ProcessData).depends_on(data); -//! let task3 = dag.add_task(ProcessData).depends_on(data); +//! let task1 = dag.add_task(ProcessData).depends_on(&data); +//! let task2 = dag.add_task(ProcessData).depends_on(&data); +//! let task3 = dag.add_task(ProcessData).depends_on(&data); //! //!let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); //! # }; diff --git a/src/output.rs b/src/output.rs index 7431335..2bb68a3 100644 --- a/src/output.rs +++ b/src/output.rs @@ -1,6 +1,6 @@ use std::{any::Any, sync::Arc}; -use crate::{builder::NodeId, runner::PassThroughHashMap, DagError, DagResult, TaskHandle}; +use crate::{builder::NodeId, runner::PassThroughHashMap, TaskHandle}; pub struct DagOutput { outputs: PassThroughHashMap>, @@ -12,14 +12,11 @@ impl DagOutput { /// # Behavior /// /// All task outputs are stored after execution and can be retrieved via get(). - /// Each task's output can only be retrieved once, after which it will return `DagError::ResultNotFound`. /// - /// # Errors + /// # Panics /// - /// Returns `DagError::ResultNotFound` if: - /// - The task hasn't been executed yet - /// - The handle is invalid - /// - The output has already been retrieved + /// This function panics if called with a [`TaskHandle`] from a [`crate::DagRunner`] instance other than the + /// one which produced this DagOutput. /// /// # Examples /// @@ -49,26 +46,19 @@ impl DagOutput { /// /// let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// - /// assert_eq!(output.get(task).unwrap(), 42); + /// assert_eq!(output.get(task), 42); /// # }; /// ``` - pub fn get(&mut self, handle: TaskHandle) -> DagResult { - let arc_output = self - .outputs - .remove(&handle.id) - .ok_or(DagError::ResultNotFound { - task_id: handle.id.0, - })?; + pub fn get(&mut self, handle: TaskHandle) -> T { + // Because this consumes the task handle, it can only be retrieved once, so is certain to be there + let arc_output = self.outputs.remove(&handle.id).unwrap(); // Downcast Arc to Arc - let output = arc_output - .downcast::() - .map_err(|_| DagError::TypeMismatch { - expected: std::any::type_name::(), - found: "unknown", - })?; + let output = arc_output.downcast::().unwrap(); - Ok(Arc::into_inner(output).unwrap()) + // Because we never give the user access to anything but a reference to the Arc, + // we can guarantee that the strong reference count is exactly one here + Arc::into_inner(output).unwrap() } pub(super) fn new( diff --git a/src/runner.rs b/src/runner.rs index 219feab..7c0f0c2 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -94,11 +94,11 @@ pub(crate) type PassThroughHashMap = HashMap; /// // Construct instances using ::new() pattern /// let x = dag.add_task(LoadValue::new(2)); /// let y = dag.add_task(LoadValue::new(3)); -/// let sum = dag.add_task(Add).depends_on((x, y)); +/// let sum = dag.add_task(Add).depends_on((&x, &y)); /// ///let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); /// -/// assert_eq!(output.get(sum).unwrap(), 5); +/// assert_eq!(output.get(sum), 5); /// # }; /// ``` pub struct DagRunner { @@ -183,10 +183,10 @@ impl DagRunner { /// let base = dag.add_task(LoadValue::new(10)); /// /// // Construct task with offset of 1 - /// let inc = dag.add_task(AddOffset::new(1)).depends_on(base); + /// let inc = dag.add_task(AddOffset::new(1)).depends_on(&base); /// ///let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); - /// assert_eq!(output.get(inc).unwrap(), 11); + /// assert_eq!(output.get(inc), 11); /// # }; /// ``` pub fn add_task<'dag, Input, Tk>(&'dag mut self, task: Tk) -> Tk::Retval<'dag> @@ -257,7 +257,7 @@ impl DagRunner { /// /// let a = dag.add_task(Value(1)); /// let b = dag.add_task(Value(2)); - /// let sum = dag.add_task(Add).depends_on((a, b)); + /// let sum = dag.add_task(Add).depends_on((&a, &b)); /// ///let mut output = dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }).await.unwrap(); // Executes all tasks /// # }; diff --git a/src/runner/tests.rs b/src/runner/tests.rs index ceef534..d724aa9 100644 --- a/src/runner/tests.rs +++ b/src/runner/tests.rs @@ -4,9 +4,7 @@ use std::collections::HashMap; use std::hash::Hasher; use crate::builder::TaskHandle; -use crate::error::DagError; use crate::runner::{DagRunner, PassThroughHasher}; -use crate::DagOutput; // Initialize tracing subscriber for tests (idempotent) #[cfg(feature = "tracing")] @@ -65,9 +63,10 @@ fn test_dag_runner_default() { } #[tokio::test] +#[should_panic] async fn test_get_wrong_type() { init_tracing(); - // Test getting with wrong type - this should return TypeMismatch + // Test getting with wrong type - this should panic let mut dag = DagRunner::new(); let handle = dag.add_task(TestTask { value: 42 }); @@ -82,38 +81,11 @@ async fn test_get_wrong_type() { _phantom: std::marker::PhantomData, }; - let result = output.get(fake_handle); - assert!(result.is_err()); - - // When downcast fails, we get TypeMismatch - match result.unwrap_err() { - DagError::TypeMismatch { expected, .. } => { - assert_eq!(expected, std::any::type_name::()); - } - _ => panic!("Expected TypeMismatch error"), - } -} - -#[tokio::test] -async fn test_get_result_not_found() { - init_tracing(); - // Test getting a result before running the DAG - let mut dag = DagRunner::new(); - let handle = dag.add_task(TestTask { value: 42 }); - let handle_id = handle.id.0; // Save ID before moving handle - let mut output = DagOutput::new(HashMap::default()); - - let result = output.get(handle); - assert!(result.is_err()); - match result.unwrap_err() { - DagError::ResultNotFound { task_id } => { - assert_eq!(task_id, handle_id); - } - _ => panic!("Expected ResultNotFound error"), - } + let _result = output.get(fake_handle); } #[tokio::test] +#[should_panic] async fn test_invalid_node_id_in_get() { init_tracing(); // Test getting with an invalid node ID @@ -130,16 +102,7 @@ async fn test_invalid_node_id_in_get() { .await .unwrap(); - let result = output.get(invalid_handle); - assert!(result.is_err()); - - // With centralized output storage, we get ResultNotFound for invalid node IDs - match result.unwrap_err() { - DagError::ResultNotFound { task_id } => { - assert_eq!(task_id, 999); - } - _ => panic!("Expected ResultNotFound error"), - } + let _result = output.get(invalid_handle); } #[tokio::test] @@ -175,8 +138,8 @@ async fn test_task_panic_in_multi_task_layer() { let source = dag.add_task(Source); // Create two tasks in the same layer - one panics, one doesn't - let _panic_task = dag.add_task(PanicTask).depends_on(source); - let _good_task = dag.add_task(GoodTask).depends_on(source); + let _panic_task = dag.add_task(PanicTask).depends_on(&source); + let _good_task = dag.add_task(GoodTask).depends_on(&source); // Run the DAG - the panic should be caught let result = dag diff --git a/src/task.rs b/src/task.rs index 4bbdd31..2afa538 100644 --- a/src/task.rs +++ b/src/task.rs @@ -47,9 +47,6 @@ where /// TaskInput with the remaining dependencies. /// /// No dependencies can be retrieved from an empty TaskInput (Input = ()). -/// -/// **Note:** Calling [`std::mem::forget`] on a `TaskInput` instance will make it impossible to retrieve outputs of dependencies. -/// There's no reason to ever do this. pub struct TaskInput<'inputs, Input: Send> { inputs: Iter<'inputs, Arc>, phantom: PhantomData, diff --git a/tests/boundaries/deep_chains.rs b/tests/boundaries/deep_chains.rs index 8a70432..e2c1f96 100644 --- a/tests/boundaries/deep_chains.rs +++ b/tests/boundaries/deep_chains.rs @@ -20,7 +20,7 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> { for _ in 0..500 { current = dag .add_task(task_fn::(|&prev: &i32| prev + 1)) - .depends_on(current); + .depends_on(¤t); } chain_ends.push(current); @@ -41,14 +41,14 @@ async fn test_multiple_parallel_deep_chains() -> DagResult<()> { }, )) .depends_on(( - chain_ends[0], - chain_ends[1], - chain_ends[2], - chain_ends[3], - chain_ends[4], - chain_ends[5], - chain_ends[6], - chain_ends[7], + &chain_ends[0], + &chain_ends[1], + &chain_ends[2], + &chain_ends[3], + &chain_ends[4], + &chain_ends[5], + &chain_ends[6], + &chain_ends[7], )); let mut output = dag @@ -60,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!(output.get(final_task)?, 32000); + assert_eq!(output.get(final_task), 32000); Ok(()) } @@ -91,7 +91,7 @@ async fn test_deep_chain_execution_order() -> DagResult<()> { prev + 1 } })) - .depends_on(current); + .depends_on(¤t); } let mut output = dag @@ -99,7 +99,7 @@ async fn test_deep_chain_execution_order() -> DagResult<()> { .await?; assert_eq!(order.load(Ordering::SeqCst), 100); - assert_eq!(output.get(current)?, 99); + assert_eq!(output.get(current), 99); Ok(()) } @@ -120,7 +120,7 @@ async fn test_deep_binary_tree() -> DagResult<()> { dag.add_task(task_fn::<(i32, i32), _, _>(move |(l, r): (&i32, &i32)| { l + r })) - .depends_on((left, right)) + .depends_on((&left, &right)) } } @@ -132,7 +132,7 @@ async fn test_deep_binary_tree() -> DagResult<()> { // 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 = output.get(root)?; + let result = output.get(root); assert!(result > 0); Ok(()) @@ -152,7 +152,7 @@ async fn test_fibonacci_chain() -> DagResult<()> { for _ in 2..50 { let next = dag .add_task(task_fn::<(i64, i64), _, _>(|(a, b): (&i64, &i64)| a + b)) - .depends_on((prev2, prev1)); + .depends_on((&prev2, &prev1)); prev2 = prev1; prev1 = next; @@ -163,7 +163,7 @@ async fn test_fibonacci_chain() -> DagResult<()> { .await?; // 50th Fibonacci number (0-indexed) = 7778742049 - assert_eq!(output.get(prev1)?, 7778742049); + assert_eq!(output.get(prev1), 7778742049); Ok(()) } @@ -184,25 +184,25 @@ async fn test_zigzag_dependencies() -> DagResult<()> { // A depends on B chain_a = dag .add_task(task_fn::(move |b_val: &i32| b_val + 1)) - .depends_on(chain_b); + .depends_on(&chain_b); } else { // B depends on A chain_b = dag .add_task(task_fn::(move |a_val: &i32| a_val + 2)) - .depends_on(chain_a); + .depends_on(&chain_a); } } // Final task depends on both chains let final_task = dag .add_task(task_fn::<(i32, i32), _, _>(|(a, b): (&i32, &i32)| a + b)) - .depends_on((chain_a, chain_b)); + .depends_on((&chain_a, &chain_b)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - let result = output.get(final_task)?; + let result = output.get(final_task); assert!(result > 100); Ok(()) diff --git a/tests/boundaries/type_limits.rs b/tests/boundaries/type_limits.rs index 11d1e30..71387ce 100644 --- a/tests/boundaries/type_limits.rs +++ b/tests/boundaries/type_limits.rs @@ -35,15 +35,15 @@ async fn test_zero_sized_types_throughout() -> DagResult<()> { assert_eq!(val, 42); "success" })) - .depends_on(t2); + .depends_on(&t2); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(t1)?, Empty); - assert_eq!(output.get(t2)?, 42); - assert_eq!(output.get(t4)?, "success"); + assert_eq!(output.get(t1), Empty); + assert_eq!(output.get(t2), 42); + assert_eq!(output.get(t4), "success"); Ok(()) } @@ -72,13 +72,13 @@ async fn test_large_value_types() -> DagResult<()> { .add_task(task_fn::(|&val: &usize| { val // Just pass through to verify })) - .depends_on(t1); + .depends_on(&t1); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(t2)?, 1042); + assert_eq!(output.get(t2), 1042); Ok(()) } @@ -99,13 +99,13 @@ async fn test_reference_wrapper_types() -> DagResult<()> { let t2 = dag .add_task(task_fn::(|&val: &i32| val * 2)) - .depends_on(t1); + .depends_on(&t1); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(t2)?, 200); + assert_eq!(output.get(t2), 200); Ok(()) } diff --git a/tests/boundaries/zero_tasks.rs b/tests/boundaries/zero_tasks.rs index b9c2100..599ca9c 100644 --- a/tests/boundaries/zero_tasks.rs +++ b/tests/boundaries/zero_tasks.rs @@ -19,7 +19,7 @@ async fn test_dag_with_only_source_tasks() -> DagResult<()> { .await?; for (i, task) in tasks.into_iter().enumerate() { - assert_eq!(output.get(task)?, i); + assert_eq!(output.get(task), i); } Ok(()) @@ -46,7 +46,7 @@ async fn test_single_self_contained_stateful_task() -> DagResult<()> { .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(task)?, 42); + assert_eq!(output.get(task), 42); Ok(()) } @@ -71,6 +71,6 @@ async fn test_minimal_task_with_minimal_async_work() -> DagResult<()> { .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(task)?, 1337); + assert_eq!(output.get(task), 1337); Ok(()) } diff --git a/tests/custom_type_test.rs b/tests/custom_type_test.rs index 607c4cc..22ffb4b 100644 --- a/tests/custom_type_test.rs +++ b/tests/custom_type_test.rs @@ -54,14 +54,14 @@ async fn test_custom_type_single_dependency() { let mut dag = DagRunner::new(); let fetch = dag.add_task(FetchPerson); - let process = dag.add_task(ProcessPerson).depends_on(fetch); + let process = dag.add_task(ProcessPerson).depends_on(&fetch); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - let result = output.get(process).unwrap(); + let result = output.get(process); assert_eq!(result, "Alice is 30 years old"); } @@ -70,16 +70,16 @@ async fn test_custom_type_passthrough() { let mut dag = DagRunner::new(); let fetch = dag.add_task(FetchPerson); - let transform = dag.add_task(PersonToCompany).depends_on(fetch); - let process = dag.add_task(ProcessPerson).depends_on(fetch); + let transform = dag.add_task(PersonToCompany).depends_on(&fetch); + let process = dag.add_task(ProcessPerson).depends_on(&fetch); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - let person_result = output.get(process).unwrap(); - let company_result = output.get(transform).unwrap(); + let person_result = output.get(process); + let company_result = output.get(transform); assert_eq!(person_result.as_str(), "Alice is 30 years old"); assert_eq!(company_result.name, "Alice's Company"); @@ -91,18 +91,18 @@ async fn test_custom_type_fan_out() { let mut dag = DagRunner::new(); 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); + 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); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - let result1 = output.get(process1).unwrap(); - let result2 = output.get(process2).unwrap(); - let result3 = output.get(process3).unwrap(); + let result1 = output.get(process1); + let result2 = output.get(process2); + let result3 = output.get(process3); assert_eq!(result1, "Alice is 30 years old"); assert_eq!(result2, "Alice is 30 years old"); diff --git a/tests/cycle_prevention.rs b/tests/cycle_prevention.rs index a5c2ed6..92e9dae 100644 --- a/tests/cycle_prevention.rs +++ b/tests/cycle_prevention.rs @@ -30,40 +30,11 @@ fn test_builder_consumed_by_depends_on() { let consumer_builder = dag.add_task(Consumer); // This consumes consumer_builder and returns a TaskHandle - let _consumer_handle = consumer_builder.depends_on(source); + let _consumer_handle = consumer_builder.depends_on(&source); // We CANNOT use consumer_builder again because it was moved // This would fail to compile: - // let _consumer2 = consumer_builder.depends_on(source); // ERROR: use of moved value -} - -/// Demonstrates that you can only get a TaskHandle by finalizing a builder, -/// and handles are immutable references with no methods to add dependencies -#[test] -fn test_handle_is_immutable_reference() { - struct NoInput; - #[task] - impl NoInput { - async fn run(&self) -> i32 { - 100 - } - } - - let mut dag = DagRunner::new(); - let builder = dag.add_task(NoInput); - - // Convert to handle (only works because Input = ()) - let handle = builder; - - // TaskHandle is Copy/Clone - just an ID wrapper - let _copy1 = handle; - let _copy2 = handle; - let _copy3 = handle; - - // But TaskHandle has NO methods to modify the DAG - // There's no way to add dependencies to it - // This would fail to compile: - // handle.depends_on(something); // ERROR: no method named `depends_on` + // let _consumer2 = consumer_builder.depends_on(&source); // ERROR: use of moved value } /// Documents the ordering constraint that prevents cycles @@ -108,8 +79,8 @@ fn test_ordering_prevents_cycles() { // Let's create a valid DAG instead: A→B (B depends on A) 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)? + 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! } @@ -182,9 +153,9 @@ fn test_three_way_cycle_impossible() { let _c_builder = dag.add_task(C); // To create A→C→B→A: - // 1. a_builder.depends_on(c_handle) // Need c_handle - // 2. c_builder.depends_on(b_handle) // Need b_handle - // 3. b_builder.depends_on(a_handle) // Need a_handle (from step 1!) + // 1. a_builder.depends_on(&c_handle) // Need c_handle + // 2. c_builder.depends_on(&b_handle) // Need b_handle + // 3. b_builder.depends_on(&a_handle) // Need a_handle (from step 1!) // // But step 3 needs a_handle which doesn't exist until step 1 completes. // And step 1 needs c_handle which doesn't exist until step 2 completes. @@ -209,7 +180,7 @@ fn test_self_loop_prevented() { let _builder = dag.add_task(SelfReferential); // To create A→A (self-loop): - // builder.depends_on(builder) + // builder.depends_on(&builder) // // But this would fail because: // 1. depends_on() takes `self` by value (moves builder) @@ -217,7 +188,7 @@ fn test_self_loop_prevented() { // 3. We can't also use &builder as an argument because it's being moved // // This would fail to compile: - // let _handle = builder.depends_on(builder); // ERROR: use of moved value + // let _handle = builder.depends_on(&builder); // ERROR: use of moved value } /// Comprehensive test showing all the guardrails @@ -263,11 +234,11 @@ async fn test_type_safety_prevents_invalid_graphs() { let s2 = dag.add_task(Source); // Transform depends on sources - let t1 = dag.add_task(Transform).depends_on(s1); - let t2 = dag.add_task(Transform).depends_on(s2); + let t1 = dag.add_task(Transform).depends_on(&s1); + let t2 = dag.add_task(Transform).depends_on(&s2); // Aggregate depends on transforms - let result = dag.add_task(Aggregate).depends_on((t1, t2)); + let result = dag.add_task(Aggregate).depends_on((&t1, &t2)); // Execute and verify let mut output = dag @@ -275,7 +246,7 @@ async fn test_type_safety_prevents_invalid_graphs() { .await .unwrap(); - assert_eq!(output.get(result).unwrap(), 40); // (10*2) + (10*2) + assert_eq!(output.get(result), 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 diff --git a/tests/dependencies/resolution.rs b/tests/dependencies/resolution.rs index 4aecc5c..c860048 100644 --- a/tests/dependencies/resolution.rs +++ b/tests/dependencies/resolution.rs @@ -40,7 +40,7 @@ async fn test_dependency_resolution_order() -> DagResult<()> { log.lock().unwrap().push(("B_end", x + 1)); x + 1 })) - .depends_on(a) + .depends_on(&a) }; let c = { @@ -52,7 +52,7 @@ async fn test_dependency_resolution_order() -> DagResult<()> { log.lock().unwrap().push(("C_end", x + 1)); x + 1 })) - .depends_on(b) + .depends_on(&b) }; let d = { @@ -64,7 +64,7 @@ async fn test_dependency_resolution_order() -> DagResult<()> { log.lock().unwrap().push(("D_end", x + 1)); x + 1 })) - .depends_on(c) + .depends_on(&c) }; let e = { @@ -76,14 +76,14 @@ async fn test_dependency_resolution_order() -> DagResult<()> { log.lock().unwrap().push(("E_end", x + 1)); x + 1 })) - .depends_on(d) + .depends_on(&d) }; let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(e)?, 5); + assert_eq!(output.get(e), 5); // Verify execution order let log = execution_log.lock().unwrap(); diff --git a/tests/dependencies/tuples.rs b/tests/dependencies/tuples.rs index dbdd8f8..ab3e790 100644 --- a/tests/dependencies/tuples.rs +++ b/tests/dependencies/tuples.rs @@ -11,13 +11,13 @@ async fn test_two_dependencies() -> DagResult<()> { let b = dag.add_task(task_fn::<(), _, _>(|_: ()| 20)); let sum = dag .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y)) - .depends_on((a, b)); + .depends_on((&a, &b)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 30); + assert_eq!(output.get(sum), 30); Ok(()) } @@ -32,13 +32,13 @@ async fn test_three_dependencies() -> DagResult<()> { .add_task(task_fn::<(i32, i32, i32), _, _>( |(x, y, z): (&i32, &i32, &i32)| x + y + z, )) - .depends_on((a, b, c)); + .depends_on((&a, &b, &c)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 6); + assert_eq!(output.get(sum), 6); Ok(()) } @@ -54,13 +54,13 @@ async fn test_four_dependencies() -> DagResult<()> { .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])); + .depends_on((&deps[0], &deps[1], &deps[2], &deps[3])); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 10); // 1+2+3+4 + assert_eq!(output.get(sum), 10); // 1+2+3+4 Ok(()) } @@ -76,13 +76,13 @@ async fn test_five_dependencies() -> DagResult<()> { .add_task(task_fn::<(i32, i32, i32, i32, i32), _, _>( |(a, b, c, d, e): (&i32, &i32, &i32, &i32, &i32)| a + b + c + d + e, )) - .depends_on((deps[0], deps[1], deps[2], deps[3], deps[4])); + .depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4])); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 15); // 1+2+3+4+5 + assert_eq!(output.get(sum), 15); // 1+2+3+4+5 Ok(()) } @@ -98,13 +98,13 @@ async fn test_six_dependencies() -> DagResult<()> { .add_task(task_fn::<(i32, i32, i32, i32, i32, i32), _, _>( |(a, b, c, d, e, f): (&i32, &i32, &i32, &i32, &i32, &i32)| a + b + c + d + e + f, )) - .depends_on((deps[0], deps[1], deps[2], deps[3], deps[4], deps[5])); + .depends_on((&deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5])); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 21); // 1+2+3+4+5+6 + assert_eq!(output.get(sum), 21); // 1+2+3+4+5+6 Ok(()) } @@ -123,14 +123,14 @@ async fn test_seven_dependencies() -> DagResult<()> { }, )) .depends_on(( - deps[0], deps[1], deps[2], deps[3], deps[4], deps[5], deps[6], + &deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6], )); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 28); // 1+2+3+4+5+6+7 + assert_eq!(output.get(sum), 28); // 1+2+3+4+5+6+7 Ok(()) } @@ -149,13 +149,13 @@ async fn test_eight_dependencies() -> DagResult<()> { }, )) .depends_on(( - deps[0], deps[1], deps[2], deps[3], deps[4], deps[5], deps[6], deps[7], + &deps[0], &deps[1], &deps[2], &deps[3], &deps[4], &deps[5], &deps[6], &deps[7], )); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(sum)?, 36); // Sum of 1..=8 + assert_eq!(output.get(sum), 36); // Sum of 1..=8 Ok(()) } diff --git a/tests/errors/propagation.rs b/tests/errors/propagation.rs index d4d7461..e66d3d7 100644 --- a/tests/errors/propagation.rs +++ b/tests/errors/propagation.rs @@ -16,12 +16,12 @@ async fn test_linear_error_propagation() { .add_task(task_fn::(|&x: &i32| -> i32 { panic!("Error at t2 with input {}", x); })) - .depends_on(t1); + .depends_on(&t1); let t3 = dag .add_task(task_fn::(|&x: &i32| x + 1)) - .depends_on(t2); + .depends_on(&t2); dag.add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(t3); + .depends_on(&t3); assert!(matches!( dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) @@ -44,16 +44,16 @@ async fn test_diamond_error_propagation() { let left = dag .add_task(task_fn::(|&x: &i32| x / 2)) - .depends_on(source); + .depends_on(&source); let right = dag .add_task(task_fn::(|&x: &i32| { panic!("Right path fails with {}", x); })) - .depends_on(source); + .depends_on(&source); dag.add_task(task_fn::<(i32, i32), _, _>(|(l, r): (&i32, &i32)| l + r)) - .depends_on((left, right)); + .depends_on((&left, &right)); assert!(matches!( dag.run(|fut| async move { tokio::spawn(fut).await.unwrap() }) @@ -73,7 +73,7 @@ async fn test_error_in_wide_fanout() { // Create wide fanout from failing source for i in 0..20 { dag.add_task(task_fn::(move |&x: &i32| x + i)) - .depends_on(source); + .depends_on(&source); } assert!(matches!( @@ -102,7 +102,7 @@ async fn test_selective_error_propagation() { } }, )) - .depends_on(source); + .depends_on(&source); // Further processing let final_task = dag @@ -113,7 +113,7 @@ async fn test_selective_error_propagation() { format!("Success: {}", val) } })) - .depends_on(handler); + .depends_on(&handler); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) @@ -121,13 +121,13 @@ async fn test_selective_error_propagation() { .unwrap(); // Source returns an Err variant (not a panic) - assert_eq!(output.get(source).unwrap(), Err("Source error".to_string())); + assert_eq!(output.get(source), Err("Source error".to_string())); // Handler processes it - assert_eq!(output.get(handler).unwrap(), -1); + assert_eq!(output.get(handler), -1); // Final task handles the error case - assert_eq!(output.get(final_task).unwrap(), "Handled error case"); + assert_eq!(output.get(final_task), "Handled error case"); } #[tokio::test] @@ -169,7 +169,7 @@ async fn test_error_propagation_timing() { .add_task(T2Task { order: propagation_order.clone(), }) - .depends_on(t1); + .depends_on(&t1); let _t3 = dag .add_task(task_fn::({ @@ -180,7 +180,7 @@ async fn test_error_propagation_timing() { x + 1 } })) - .depends_on(t2); + .depends_on(&t2); let _ = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) diff --git a/tests/errors/recovery.rs b/tests/errors/recovery.rs index ef96102..8a82eb6 100644 --- a/tests/errors/recovery.rs +++ b/tests/errors/recovery.rs @@ -34,7 +34,7 @@ async fn test_retry_pattern_simulation() -> DagResult<()> { .await?; // Task completes with error on first "attempt" - assert_eq!(output.get(task)?, Err("Simulated failure")); + assert_eq!(output.get(task), Err("Simulated failure")); Ok(()) } @@ -78,7 +78,7 @@ async fn test_circuit_breaker_pattern() -> DagResult<()> { // First few tasks fail and open circuit for task in tasks.into_iter().take(4) { - assert!(output.get(task)?.is_err()); + assert!(output.get(task).is_err()); } Ok(()) @@ -120,8 +120,8 @@ async fn test_error_accumulation_pattern() -> DagResult<()> { let [first, second] = <[_; 2]>::try_from(tasks).map_err(|_e| ()).unwrap(); // Verify specific results - assert!(output.get(first)?.is_err()); - assert_eq!(output.get(second)?, Ok(1)); + assert!(output.get(first).is_err()); + assert_eq!(output.get(second), Ok(1)); Ok(()) } @@ -136,29 +136,29 @@ async fn test_error_boundary_isolation() -> DagResult<()> { .add_task(task_fn::(|_x: &i32| { Err::("Group A error") })) - .depends_on(a1); + .depends_on(&a1); // Group B: should be isolated from Group A's error let b1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 20)); let b2 = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(b1); + .depends_on(&b1); // Group C: depends on B but not A let c1 = dag .add_task(task_fn::(|&x: &i32| x + 5)) - .depends_on(b2); + .depends_on(&b2); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; // Group A fails (returns Err) - assert_eq!(output.get(a2)?, Err("Group A error")); + assert_eq!(output.get(a2), Err("Group A error")); // Groups B and C succeed - assert_eq!(output.get(b2)?, 40); - assert_eq!(output.get(c1)?, 45); + assert_eq!(output.get(b2), 40); + assert_eq!(output.get(c1), 45); Ok(()) } diff --git a/tests/execution/mod.rs b/tests/execution/mod.rs index 4e3e11c..a878c41 100644 --- a/tests/execution/mod.rs +++ b/tests/execution/mod.rs @@ -20,7 +20,7 @@ async fn test_single_task() { .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - assert_eq!(output.get(task).unwrap(), 42); + assert_eq!(output.get(task), 42); } #[tokio::test] @@ -37,7 +37,7 @@ async fn test_wide_parallel() { .unwrap(); for (i, task) in tasks.into_iter().enumerate() { - assert_eq!(output.get(task).unwrap(), i * 2); + assert_eq!(output.get(task), i * 2); } } @@ -48,13 +48,13 @@ async fn test_diamond_dependency() { let a = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)); let b = dag .add_task(task_fn::(|&x: &i32| x + 5)) - .depends_on(a); + .depends_on(&a); let c = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(a); + .depends_on(&a); let d = dag .add_task(task_fn::<(i32, i32), _, _>(|(x, y): (&i32, &i32)| x + y)) - .depends_on((b, c)); + .depends_on((&b, &c)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) @@ -63,7 +63,7 @@ async fn test_diamond_dependency() { // Note: a, b, and c are not sinks (they have dependents) // Only d is a sink and can be retrieved - assert_eq!(output.get(d).unwrap(), 35); // 15 + 20 + assert_eq!(output.get(d), 35); // 15 + 20 } #[derive(Clone)] @@ -88,9 +88,9 @@ async fn test_different_output_types() { .await .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!(&output.get(string_task), "hello"); + assert_eq!(output.get(vec_task), vec![1, 2, 3]); + let user = output.get(struct_task); assert_eq!(user.name, "Alice"); assert_eq!(user.age, 30); } @@ -104,26 +104,26 @@ async fn test_multiple_sinks() { // Branch 1 let branch1 = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(source); + .depends_on(&source); let sink1 = dag .add_task(task_fn::(|&x: &i32| x + 1)) - .depends_on(branch1); + .depends_on(&branch1); // Branch 2 let branch2 = dag .add_task(task_fn::(|&x: &i32| x + 5)) - .depends_on(source); + .depends_on(&source); let sink2 = dag .add_task(task_fn::(|&x: &i32| x * 3)) - .depends_on(branch2); + .depends_on(&branch2); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); // Should wait for both sinks - assert_eq!(output.get(sink1).unwrap(), 21); // (10 * 2) + 1 - assert_eq!(output.get(sink2).unwrap(), 45); // (10 + 5) * 3 + assert_eq!(output.get(sink1), 21); // (10 * 2) + 1 + assert_eq!(output.get(sink2), 45); // (10 + 5) * 3 } #[tokio::test] @@ -135,17 +135,17 @@ async fn test_single_task_inline_path() { let t1 = dag.add_task(task_fn::<(), _, _>(|_: ()| 10)); let t2 = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(t1); + .depends_on(&t1); let t3 = dag .add_task(task_fn::(|&x: &i32| x + 5)) - .depends_on(t2); + .depends_on(&t2); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - assert_eq!(output.get(t3).unwrap(), 25); // (10 * 2) + 5 + assert_eq!(output.get(t3), 25); // (10 * 2) + 5 } #[tokio::test] @@ -158,20 +158,20 @@ async fn test_multi_consumer_fanout() { // Multiple consumers let c1 = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(producer); + .depends_on(&producer); let c2 = dag .add_task(task_fn::(|&x: &i32| x + 10)) - .depends_on(producer); + .depends_on(&producer); let c3 = dag .add_task(task_fn::(|&x: &i32| x - 5)) - .depends_on(producer); + .depends_on(&producer); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - assert_eq!(output.get(c1).unwrap(), 84); - assert_eq!(output.get(c2).unwrap(), 52); - assert_eq!(output.get(c3).unwrap(), 37); + assert_eq!(output.get(c1), 84); + assert_eq!(output.get(c2), 52); + assert_eq!(output.get(c3), 37); } diff --git a/tests/interleaving/mod.rs b/tests/interleaving/mod.rs index 63177f7..7b1aae2 100644 --- a/tests/interleaving/mod.rs +++ b/tests/interleaving/mod.rs @@ -62,13 +62,13 @@ async fn test_forced_interleaving_with_barriers() -> DagResult<()> { .add_task(task_fn::<(i32, i32, i32), _, _>( |(a, b, c): (&i32, &i32, &i32)| a + b + c, )) - .depends_on((t1, t2, t3)); + .depends_on((&t1, &t2, &t3)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.get(collector)?, 6); + assert_eq!(output.get(collector), 6); // Check interleaving pattern let events = order.lock().unwrap().clone(); @@ -178,7 +178,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> { turn: turn.clone(), idx: i, }) - .depends_on(chain_a); + .depends_on(&chain_a); chain_b = dag .add_task(ChainB { @@ -186,7 +186,7 @@ async fn test_alternating_execution_pattern() -> DagResult<()> { events: events.clone(), idx: i, }) - .depends_on(chain_b); + .depends_on(&chain_b); } let _output = dag @@ -255,7 +255,7 @@ async fn test_layered_interleaving() -> DagResult<()> { dag.add_task(Layer23Task { active: layer_active.clone(), }) - .depends_on((layer1[idx], layer1[idx + 1])) + .depends_on((&layer1[idx], &layer1[idx + 1])) }) .collect(); @@ -264,14 +264,14 @@ async fn test_layered_interleaving() -> DagResult<()> { .add_task(Layer23Task { active: layer_active.clone(), }) - .depends_on((layer2[0], layer2[1])); + .depends_on((&layer2[0], &layer2[1])); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; // Sum of 0+1+2+3 = 6 - assert_eq!(output.get(layer3)?, 6); + assert_eq!(output.get(layer3), 6); Ok(()) } diff --git a/tests/parallelism/spawning.rs b/tests/parallelism/spawning.rs index a6070c4..cb4fc1e 100644 --- a/tests/parallelism/spawning.rs +++ b/tests/parallelism/spawning.rs @@ -33,7 +33,7 @@ async fn test_spawner_actually_spawns_tasks() -> DagResult<()> { // Verify all tasks were spawned for (i, task) in tasks.into_iter().enumerate() { - assert_eq!(output.get(task)?, i as i32); + assert_eq!(output.get(task), i as i32); } // The spawner should have been called once for each task diff --git a/tests/parallelism/threads.rs b/tests/parallelism/threads.rs index 6636ebb..685a72c 100644 --- a/tests/parallelism/threads.rs +++ b/tests/parallelism/threads.rs @@ -68,7 +68,7 @@ async fn test_concurrent_execution_with_atomic_counter() -> DagResult<()> { // Verify all tasks completed for (i, task) in tasks.into_iter().enumerate() { - assert_eq!(output.get(task)?, i); + assert_eq!(output.get(task), i); } // Check that we achieved real parallelism @@ -135,7 +135,7 @@ async fn test_massive_parallel_fanout() -> DagResult<()> { } // Create 1000 tasks all depending on the same source - let tasks: Vec<_> = (0..1000) + let mut tasks: Vec<_> = (0..1000) .map(|i| { dag.add_task(FanoutTask { comp: completed.clone(), @@ -143,7 +143,7 @@ async fn test_massive_parallel_fanout() -> DagResult<()> { curr: current.clone(), idx: i, }) - .depends_on(source) + .depends_on(&source) }) .collect(); @@ -157,9 +157,9 @@ async fn test_massive_parallel_fanout() -> DagResult<()> { assert_eq!(completed.load(Ordering::SeqCst), 1000); // Check some results - assert_eq!(output.get(tasks[0])?, 1); - assert_eq!(output.get(tasks[500])?, 501); - assert_eq!(output.get(tasks[999])?, 1000); + assert_eq!(output.get(tasks.remove(999)), 1000); + assert_eq!(output.get(tasks.remove(500)), 501); + assert_eq!(output.get(tasks.remove(0)), 1); let max_seen = max_concurrent.load(Ordering::SeqCst); println!( diff --git a/tests/parallelism/timing.rs b/tests/parallelism/timing.rs index 1e0e978..088c44d 100644 --- a/tests/parallelism/timing.rs +++ b/tests/parallelism/timing.rs @@ -40,7 +40,7 @@ async fn test_parallel_execution_speedup() -> DagResult<()> { // Verify all tasks completed for (i, task) in tasks.into_iter().enumerate() { - assert_eq!(output.get(task)?, i); + assert_eq!(output.get(task), i); } // If tasks ran sequentially: 10 * 50ms = 500ms @@ -125,7 +125,7 @@ async fn test_layers_execute_in_parallel() -> DagResult<()> { idx: i, timing: layer_timing.clone(), }) - .depends_on(source) + .depends_on(&source) }) .collect(); @@ -140,14 +140,14 @@ async fn test_layers_execute_in_parallel() -> DagResult<()> { a + b + c }, )) - .depends_on((layer1[0], layer1[1], layer1[2])) + .depends_on((&layer1[0], &layer1[1], &layer1[2])) }; let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await?; - assert_eq!(output.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(); diff --git a/tests/runtimes/mod.rs b/tests/runtimes/mod.rs index ca424fe..ad83257 100644 --- a/tests/runtimes/mod.rs +++ b/tests/runtimes/mod.rs @@ -99,11 +99,11 @@ fn test_basic_dag(runner: impl RuntimeTest) { let x = dag.add_task(Value(2)); let y = dag.add_task(Value(3)); - let sum = dag.add_task(Add).depends_on((x, y)); + let sum = dag.add_task(Add).depends_on((&x, &y)); let mut output = dag.run(spawner).await.unwrap(); - assert_eq!(output.get(sum).unwrap(), 5); + assert_eq!(output.get(sum), 5); }) } @@ -122,7 +122,7 @@ fn test_parallel_execution(runner: impl RuntimeTest) { let mut output = dag.run(spawner).await.unwrap(); for (i, task) in tasks.into_iter().enumerate() { - assert_eq!(output.get(task).unwrap(), i * 2); + assert_eq!(output.get(task), i * 2); } }); } @@ -137,13 +137,13 @@ fn test_complex_dependencies(runner: impl RuntimeTest) { let a = dag.add_task(Value(10)); let b = dag.add_task(Value(20)); - let sum = dag.add_task(Add).depends_on((a, b)); + let sum = dag.add_task(Add).depends_on((&a, &b)); let double = dag .add_task(task_fn::(|&x: &i32| x * 2)) - .depends_on(sum); + .depends_on(&sum); let mut output = dag.run(spawner).await.unwrap(); - assert_eq!(output.get(double).unwrap().clone(), 60); + assert_eq!(output.get(double).clone(), 60); }); } diff --git a/tests/tracing/with_tracing.rs b/tests/tracing/with_tracing.rs index c952d2b..9bf9eea 100644 --- a/tests/tracing/with_tracing.rs +++ b/tests/tracing/with_tracing.rs @@ -43,14 +43,14 @@ async fn test_tracing_with_subscriber() { let a = dag.add_task(Value(2)); let b = dag.add_task(Value(3)); - let sum = dag.add_task(Add).depends_on((a, b)); + let sum = dag.add_task(Add).depends_on((&a, &b)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) .await .unwrap(); - assert_eq!(output.get(sum).unwrap(), 5); + assert_eq!(output.get(sum), 5); } #[tokio::test(flavor = "multi_thread")] @@ -65,8 +65,8 @@ async fn test_tracing_inline_execution() { // Create a linear chain (should trigger inline execution) 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)); + let b = dag.add_task(Add).depends_on((&a, &a)); + let c = dag.add_task(Multiply).depends_on((&b, &b)); let mut output = dag .run(|fut| async move { tokio::spawn(fut).await.unwrap() }) @@ -76,5 +76,5 @@ async fn test_tracing_inline_execution() { // a = 1 // b = 1 + 1 = 2 // c = 2 * 2 = 4 - assert_eq!(output.get(c).unwrap(), 4); + assert_eq!(output.get(c), 4); }