diff --git a/Cargo.lock b/Cargo.lock index 9fe16ff..fc6ce91 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -378,6 +378,7 @@ dependencies = [ "criterion", "dagrs", "dagx-macros", + "dagx-test", "futures-executor", "futures-util", "parking_lot", @@ -398,6 +399,13 @@ dependencies = [ "syn", ] +[[package]] +name = "dagx-test" +version = "0.3.1" +dependencies = [ + "dagx", +] + [[package]] name = "either" version = "1.15.0" diff --git a/Cargo.toml b/Cargo.toml index 4c25fd3..c53da27 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["dagx-macros"] +members = ["dagx-macros", "dagx-test"] [package] name = "dagx" @@ -21,12 +21,15 @@ exclude = [".claude/", ".jj/", "dagx-spec.md"] default = [] tracing = ["dep:tracing"] -[lints.rust] +[workspace.lints.rust] unexpected_cfgs = { level = "warn", check-cfg = [ 'cfg(tarpaulin)', 'cfg(tarpaulin_include)', ] } +[lints] +workspace = true + [dependencies] futures-util = { version = "0.3", default-features = false, features = ["std"] } parking_lot = "0.12" @@ -34,6 +37,7 @@ dagx-macros = { version = "0.3.1", path = "./dagx-macros" } tracing = { version = "0.1", optional = true } [dev-dependencies] +dagx-test = { path = "./dagx-test" } tokio = { version = "1", features = [ "macros", "rt-multi-thread", diff --git a/benches/basic/scaling.rs b/benches/basic/scaling.rs index c4d6a17..1be3a1c 100644 --- a/benches/basic/scaling.rs +++ b/benches/basic/scaling.rs @@ -1,7 +1,8 @@ //! DAG scaling benchmarks - how performance scales with task count use criterion::{BenchmarkId, Criterion}; -use dagx::{task_fn, DagRunner}; +use dagx::DagRunner; +use dagx_test::task_fn; pub fn bench_dag_scaling(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/benches/comparisons/deep_chain.rs b/benches/comparisons/deep_chain.rs index c93084b..932601d 100644 --- a/benches/comparisons/deep_chain.rs +++ b/benches/comparisons/deep_chain.rs @@ -13,7 +13,8 @@ pub fn bench_deep_chain(c: &mut Criterion) { group.bench_function("dagx", |b| { b.iter(|| { rt.block_on(async { - use dagx::{task_fn, DagRunner}; + use dagx::DagRunner; + use dagx_test::task_fn; let dag = DagRunner::new(); diff --git a/benches/comparisons/diamond_pattern.rs b/benches/comparisons/diamond_pattern.rs index c0f15bd..5aae415 100644 --- a/benches/comparisons/diamond_pattern.rs +++ b/benches/comparisons/diamond_pattern.rs @@ -13,7 +13,8 @@ pub fn bench_diamond_pattern(c: &mut Criterion) { group.bench_function("dagx", |b| { b.iter(|| { rt.block_on(async { - use dagx::{task_fn, DagRunner}; + use dagx::DagRunner; + use dagx_test::task_fn; let dag = DagRunner::new(); diff --git a/benches/comparisons/etl_pipeline.rs b/benches/comparisons/etl_pipeline.rs index ef9f70c..1363546 100644 --- a/benches/comparisons/etl_pipeline.rs +++ b/benches/comparisons/etl_pipeline.rs @@ -12,7 +12,8 @@ pub fn bench_etl_pipeline(c: &mut Criterion) { group.bench_function("dagx", |b| { b.iter(|| { rt.block_on(async { - use dagx::{task_fn, DagRunner}; + use dagx::DagRunner; + use dagx_test::task_fn; let dag = DagRunner::new(); diff --git a/benches/comparisons/large_scale.rs b/benches/comparisons/large_scale.rs index 53c2bb4..b23a562 100644 --- a/benches/comparisons/large_scale.rs +++ b/benches/comparisons/large_scale.rs @@ -14,7 +14,8 @@ pub fn bench_large_scale(c: &mut Criterion) { group.bench_function("dagx", |b| { b.iter(|| { rt.block_on(async { - use dagx::{task_fn, DagRunner}; + use dagx::DagRunner; + use dagx_test::task_fn; let dag = DagRunner::new(); for i in 0..10_000 { diff --git a/benches/comparisons/linear_pipeline.rs b/benches/comparisons/linear_pipeline.rs index 63d243c..6e787c0 100644 --- a/benches/comparisons/linear_pipeline.rs +++ b/benches/comparisons/linear_pipeline.rs @@ -12,7 +12,8 @@ pub fn bench_linear_pipeline(c: &mut Criterion) { group.bench_function("dagx", |b| { b.iter(|| { rt.block_on(async { - use dagx::{task_fn, DagRunner}; + use dagx::DagRunner; + use dagx_test::task_fn; let dag = DagRunner::new(); diff --git a/benches/comparisons/wide_fanout.rs b/benches/comparisons/wide_fanout.rs index 960956d..fa5149c 100644 --- a/benches/comparisons/wide_fanout.rs +++ b/benches/comparisons/wide_fanout.rs @@ -13,7 +13,8 @@ pub fn bench_wide_fanout(c: &mut Criterion) { group.bench_function("dagx", |b| { b.iter(|| { rt.block_on(async { - use dagx::{task_fn, DagRunner}; + use dagx::DagRunner; + use dagx_test::task_fn; let dag = DagRunner::new(); let source: TaskHandle<_> = dag.add_task(task_fn::<(), _, _>(|_: ()| 42)).into(); diff --git a/benches/patterns/breakdown.rs b/benches/patterns/breakdown.rs index 3d84b7d..196d857 100644 --- a/benches/patterns/breakdown.rs +++ b/benches/patterns/breakdown.rs @@ -1,7 +1,8 @@ //! 10,000 task breakdown benchmarks - construction vs execution use criterion::{BatchSize, Criterion}; -use dagx::{task_fn, DagRunner}; +use dagx::DagRunner; +use dagx_test::task_fn; pub fn bench_10k_breakdown(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/benches/patterns/diamond.rs b/benches/patterns/diamond.rs index 5806bb9..75a2263 100644 --- a/benches/patterns/diamond.rs +++ b/benches/patterns/diamond.rs @@ -1,8 +1,8 @@ //! Diamond pattern benchmarks (fan-out then fan-in) use criterion::Criterion; -use dagx::{task_fn, DagRunner, TaskHandle}; - +use dagx::{DagRunner, TaskHandle}; +use dagx_test::task_fn; 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 91c62cc..0b49035 100644 --- a/benches/patterns/fanout.rs +++ b/benches/patterns/fanout.rs @@ -1,8 +1,8 @@ //! Fan-out pattern benchmarks (1 → N dependencies) use criterion::Criterion; -use dagx::{task_fn, DagRunner, TaskHandle}; - +use dagx::{DagRunner, TaskHandle}; +use dagx_test::task_fn; 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 06c0e80..2be6428 100644 --- a/benches/realistic/etl.rs +++ b/benches/realistic/etl.rs @@ -1,8 +1,8 @@ //! ETL (Extract, Transform, Load) pipeline benchmark use criterion::{BenchmarkId, Criterion}; -use dagx::{task_fn, DagRunner, TaskHandle}; - +use dagx::{DagRunner, TaskHandle}; +use dagx_test::task_fn; 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/memory_efficiency.rs b/benches/realistic/memory_efficiency.rs index 64b5e69..61ae2c8 100644 --- a/benches/realistic/memory_efficiency.rs +++ b/benches/realistic/memory_efficiency.rs @@ -1,8 +1,8 @@ //! Memory efficiency benchmark with large data structures use criterion::{BenchmarkId, Criterion}; -use dagx::{task_fn, DagRunner, TaskHandle}; - +use dagx::{DagRunner, TaskHandle}; +use dagx_test::task_fn; use std::hint::black_box; pub fn bench_memory_efficiency(c: &mut Criterion) { diff --git a/benches/realistic/mixed.rs b/benches/realistic/mixed.rs index 9b706d1..5f34d8d 100644 --- a/benches/realistic/mixed.rs +++ b/benches/realistic/mixed.rs @@ -1,8 +1,8 @@ //! Mixed realistic patterns - fan-out, fan-in, and processing use criterion::Criterion; -use dagx::{task_fn, DagRunner, TaskHandle}; - +use dagx::{DagRunner, TaskHandle}; +use dagx_test::task_fn; pub fn bench_mixed_patterns(c: &mut Criterion) { let rt = tokio::runtime::Runtime::new().unwrap(); diff --git a/dagx-test/Cargo.toml b/dagx-test/Cargo.toml new file mode 100644 index 0000000..3b4ec4a --- /dev/null +++ b/dagx-test/Cargo.toml @@ -0,0 +1,17 @@ +[package] +name = "dagx-test" +version = "0.3.1" +edition = "2021" +rust-version = "1.81" +authors = ["TechnoPorg"] +license = "MIT" +description = "Utilities for dagx's internal tests" +repository = "https://github.com/swaits/dagx" +homepage = "https://github.com/swaits/dagx" +documentation = "https://docs.rs/dagx-test" + +[lints] +workspace = true + +[dependencies] +dagx = { version = "0.3.1", path = "../" } diff --git a/src/extract.rs b/dagx-test/src/extract.rs similarity index 65% rename from src/extract.rs rename to dagx-test/src/extract.rs index ad3b0ba..d3c0dcf 100644 --- a/src/extract.rs +++ b/dagx-test/src/extract.rs @@ -1,45 +1,10 @@ -//! Input extraction for task dependencies via a list of dependencies. -//! -//! # Legacy Trait for Internal Use Only -//! -//! **NOTE**: This module is now primarily for internal use. The `ExtractInput` trait is only -//! used by the internal `task_fn` test helper function. -//! -//! **For regular tasks**: The `#[task]` macro generates inline `run()` methods with -//! type-specific extraction logic. This means **ANY type** implementing `Send + Sync` -//! works automatically without needing `ExtractInput` implementations! -//! -//! # How it works -//! -//! When a task executes, it needs to extract its inputs from type-erased oneshot dependencies. -//! Outputs are wrapped in Arc for efficient fanout (cheap Arc clones instead of data clones). -//! -//! **Modern approach (via `#[task]` macro)**: -//! - The macro generates custom extraction logic inline in `run()` -//! - Works with ANY type (just needs `Send + Sync + 'static`) -//! - No trait implementations needed! -//! -//! **Legacy approach (via `ExtractInput` trait)**: -//! - Used only by the internal `task_fn` test helper -//! - Provides implementations for primitives, standard types, and tuples -//! - Requires explicit trait implementations for custom types (not recommended) -//! -//! # Why macros? -//! -//! Rust lacks variadic generics, so we use macros to generate implementations for: -//! - Common primitive/standard types (one impl per type) -//! - Tuples of different sizes (one impl per size, up to 8 elements) -//! -//! This is standard practice in Rust for working with tuples of different arities. +//! Input extraction for tasks from a list of dependencies. use std::{collections::HashMap, hash::Hash, sync::Arc}; use crate::TaskInput; /// Helper trait for input extraction from list of dependencies. -/// -/// Type erasure occurs only at the ExecutableNode trait boundary - by the time -/// we're in ExtractInput, we know the concrete type and can safely downcast. pub trait ExtractInput: Sized + Clone { type Input: Send; type Retv<'input>; @@ -157,18 +122,6 @@ where } } -/// Macro to implement ExtractInput for tuple types. -/// -/// We need separate implementations for each tuple size (2-8) because Rust doesn't support -/// variadic generics. This is standard practice when working with tuples in Rust. -/// -/// # Compile-time Safety -/// -/// The number of dependencies is guaranteed to match the tuple size at compile time through: -/// 1. The `DepsTuple` trait bound on `depends_on()` ensures type/arity matching -/// 2. The `IsUnitType` constraint on `From` prevents using tasks without calling `depends_on()` -/// -/// Therefore, no runtime validation is needed - if it compiles, the dependency count is correct. macro_rules! impl_extract_tuple { ($($T:ident),+) => { impl<$($T: 'static + Clone + Send + Sync),+> ExtractInput for ($($T,)+) { diff --git a/dagx-test/src/lib.rs b/dagx-test/src/lib.rs new file mode 100644 index 0000000..226b891 --- /dev/null +++ b/dagx-test/src/lib.rs @@ -0,0 +1,50 @@ +//! DAG builder helpers for internal use in dagx tests and benchmarks. +//! +//! This crate is not meant for public use and offers no stability guarantees. +//! The ergonomics are also such that you don't want to use it anyways. **Use the `#[task]` macro instead.** + +#![cfg(not(tarpaulin_include))] + +use crate::extract::ExtractInput; + +use dagx::{Task, TaskInput}; + +mod extract; + +pub struct TaskFn +where + for<'input> F: FnMut(I::Retv<'input>) -> O + Send, + I: ExtractInput + Send + Sync + 'static, + O: Send, +{ + f: F, + _phantom: std::marker::PhantomData O>, +} + +impl Task for TaskFn +where + for<'input> F: FnMut(I::Retv<'input>) -> O + Send, + I: ExtractInput + Send + Sync + 'static, + O: Send + Sync, +{ + type Input = I::Input; + type Output = O; + + async fn run(mut self, input: TaskInput<'_, I::Input>) -> O { + (self.f)(I::extract_from_task_input(input).unwrap()) + } +} + +/// Convenience function to create a task from a closure. +pub fn task_fn(f: F) -> TaskFn +where + for<'input> F: FnMut(I::Retv<'input>) -> O + Send, + I: ExtractInput + Send + Sync + 'static, + I::Input: Send + Sync + 'static, + O: Send + Sync + 'static, +{ + TaskFn { + f, + _phantom: std::marker::PhantomData, + } +} diff --git a/src/lib.rs b/src/lib.rs index d09e7d0..443e15e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -770,7 +770,6 @@ extern crate self as dagx; mod builder; mod deps; mod error; -mod extract; mod node; mod runner; mod task; @@ -785,7 +784,3 @@ pub use types::{Pending, TaskHandle}; // Re-export the procedural macro pub use dagx_macros::task; - -// Internal testing support - not part of the public API -#[doc(hidden)] -pub use task::task_fn; diff --git a/src/task.rs b/src/task.rs index 604c2aa..0b40a18 100644 --- a/src/task.rs +++ b/src/task.rs @@ -5,8 +5,6 @@ use std::{any::Any, future::Future, marker::PhantomData, slice::Iter, sync::Arc}; -use crate::extract::ExtractInput; - /// A unit of async work with typed inputs and outputs. /// /// Use the [`task`](crate::task) macro to implement this trait. The macro automatically @@ -149,50 +147,5 @@ impl TaskInput<'static, ()> { } } -/// **INTERNAL USE ONLY**: This struct is for testing purposes only and should -/// not be used in production code. Use the `#[task]` macro instead. -#[doc(hidden)] -pub struct TaskFn -where - for<'input> F: FnMut(I::Retv<'input>) -> O + Send, - I: ExtractInput + Send + Sync + 'static, - O: Send, -{ - f: F, - _phantom: std::marker::PhantomData O>, -} - -impl Task for TaskFn -where - for<'input> F: FnMut(I::Retv<'input>) -> O + Send, - I: ExtractInput + Send + Sync + 'static, - O: Send + Sync, -{ - type Input = I::Input; - type Output = O; - - async fn run(mut self, input: TaskInput<'_, I::Input>) -> O { - (self.f)(I::extract_from_task_input(input).unwrap()) - } -} - -/// Convenience function to create a task from a closure. -/// -/// **INTERNAL USE ONLY**: This function is for testing purposes only and should -/// not be used in production code. Use the `#[task]` macro instead. -#[doc(hidden)] -pub fn task_fn(f: F) -> TaskFn -where - for<'input> F: FnMut(I::Retv<'input>) -> O + Send, - I: ExtractInput + Send + Sync + 'static, - I::Input: Send + Sync + 'static, - O: Send + Sync + 'static, -{ - TaskFn { - f, - _phantom: std::marker::PhantomData, - } -} - #[cfg(test)] mod tests; diff --git a/tests/boundaries/deep_chains.rs b/tests/boundaries/deep_chains.rs index a4b32dc..a6bb29e 100644 --- a/tests/boundaries/deep_chains.rs +++ b/tests/boundaries/deep_chains.rs @@ -1,7 +1,7 @@ //! Tests for extremely deep dependency chains -use dagx::task_fn; use dagx::{DagResult, DagRunner}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/boundaries/type_limits.rs b/tests/boundaries/type_limits.rs index 6187928..79a3f21 100644 --- a/tests/boundaries/type_limits.rs +++ b/tests/boundaries/type_limits.rs @@ -1,7 +1,7 @@ //! Tests for type system edge cases and limits -use dagx::task_fn; use dagx::{DagResult, DagRunner, TaskHandle}; +use dagx_test::task_fn; use std::marker::PhantomData; diff --git a/tests/boundaries/zero_tasks.rs b/tests/boundaries/zero_tasks.rs index a455243..3548a63 100644 --- a/tests/boundaries/zero_tasks.rs +++ b/tests/boundaries/zero_tasks.rs @@ -1,7 +1,7 @@ //! Tests for empty DAGs and minimal task configurations -use dagx::task_fn; use dagx::{task, DagResult, DagRunner}; +use dagx_test::task_fn; use tokio::task::yield_now; diff --git a/tests/dependencies/resolution.rs b/tests/dependencies/resolution.rs index 232042b..31ac727 100644 --- a/tests/dependencies/resolution.rs +++ b/tests/dependencies/resolution.rs @@ -1,7 +1,7 @@ //! Tests for dependency resolution order and transitive dependencies -use dagx::task_fn; use dagx::{task, DagResult, DagRunner}; +use dagx_test::task_fn; use std::sync::{Arc, Mutex}; diff --git a/tests/dependencies/tuples.rs b/tests/dependencies/tuples.rs index acce64e..aafbd3a 100644 --- a/tests/dependencies/tuples.rs +++ b/tests/dependencies/tuples.rs @@ -1,7 +1,7 @@ //! Tests for tuple dependencies (2-8 dependencies) -use dagx::task_fn; use dagx::{DagResult, DagRunner}; +use dagx_test::task_fn; #[tokio::test] async fn test_two_dependencies() -> DagResult<()> { diff --git a/tests/errors/propagation.rs b/tests/errors/propagation.rs index 56f28f3..44501c9 100644 --- a/tests/errors/propagation.rs +++ b/tests/errors/propagation.rs @@ -1,7 +1,7 @@ //! Tests for error propagation through the DAG -use dagx::task_fn; use dagx::{task, DagRunner, TaskHandle}; +use dagx_test::task_fn; use std::sync::Arc; use std::time::Duration; diff --git a/tests/errors/recovery.rs b/tests/errors/recovery.rs index 1dbbcd8..e1dc5b9 100644 --- a/tests/errors/recovery.rs +++ b/tests/errors/recovery.rs @@ -1,7 +1,7 @@ //! Tests for error recovery patterns -use dagx::task_fn; use dagx::{DagResult, DagRunner}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/execution/mod.rs b/tests/execution/mod.rs index 95b3cea..9f29dd3 100644 --- a/tests/execution/mod.rs +++ b/tests/execution/mod.rs @@ -1,5 +1,5 @@ // Basic execution tests for DagRunner -use dagx::task_fn; +use dagx_test::task_fn; use dagx::*; diff --git a/tests/parallelism/concurrent.rs b/tests/parallelism/concurrent.rs index 5dbd95d..fd746a6 100644 --- a/tests/parallelism/concurrent.rs +++ b/tests/parallelism/concurrent.rs @@ -1,7 +1,7 @@ //! Concurrent access tests -use dagx::task_fn; use dagx::{task, DagRunner}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/parallelism/spawning.rs b/tests/parallelism/spawning.rs index 32478c2..1ddf148 100644 --- a/tests/parallelism/spawning.rs +++ b/tests/parallelism/spawning.rs @@ -1,7 +1,7 @@ //! Tests for spawner functionality -use dagx::task_fn; use dagx::{DagResult, DagRunner}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/parallelism/threads.rs b/tests/parallelism/threads.rs index 1ebd61e..59cbd29 100644 --- a/tests/parallelism/threads.rs +++ b/tests/parallelism/threads.rs @@ -1,7 +1,7 @@ //! Tests for multi-threading and concurrent execution -use dagx::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/parallelism/timing.rs b/tests/parallelism/timing.rs index cc2b24b..8903ea0 100644 --- a/tests/parallelism/timing.rs +++ b/tests/parallelism/timing.rs @@ -1,6 +1,7 @@ //! Tests for timing and speedup verification -use dagx::{task, task_fn, DagResult, DagRunner}; +use dagx::{task, DagResult, DagRunner}; +use dagx_test::task_fn; use std::{ sync::{Arc, Mutex}, diff --git a/tests/runtimes/mod.rs b/tests/runtimes/mod.rs index 093d4c6..740a1da 100644 --- a/tests/runtimes/mod.rs +++ b/tests/runtimes/mod.rs @@ -2,7 +2,8 @@ use std::{any::Any, sync::Arc}; -use dagx::{task_fn, DagResult, DagRunner}; +use dagx::{DagResult, DagRunner}; +use dagx_test::task_fn; use futures_util::{future::BoxFuture, task::SpawnExt, Future}; use test_case::test_case; diff --git a/tests/timing/concurrent_access.rs b/tests/timing/concurrent_access.rs index 1a4570a..ad5b147 100644 --- a/tests/timing/concurrent_access.rs +++ b/tests/timing/concurrent_access.rs @@ -1,7 +1,7 @@ //! Tests for concurrent access patterns and race conditions -use dagx::task_fn; use dagx::{task, DagResult, DagRunner, TaskHandle}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::Arc; diff --git a/tests/timing/interleaving.rs b/tests/timing/interleaving.rs index 3cf9ad6..cb430d9 100644 --- a/tests/timing/interleaving.rs +++ b/tests/timing/interleaving.rs @@ -1,7 +1,7 @@ //! Tests for forced execution interleaving patterns -use dagx::task_fn; use dagx::{task, DagResult, DagRunner}; +use dagx_test::task_fn; use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc;