Compare commits
5
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 swaits
|
a34786774f
|
||
|
swaits
|
9ad3bd174f
|
||
|
swaits
|
979232bd67
|
||
|
swaits
|
8ee4092f9a
|
||
|
swaits
|
fad97dbf16
|
No files matched your search
Binary file not shown.
@@ -1,142 +0,0 @@
|
||||
# cargo-mutants configuration for the dagx crate
|
||||
#
|
||||
# This configuration optimizes mutation testing by:
|
||||
# - Focusing on correctness-critical code
|
||||
# - Excluding low-value mutations
|
||||
# - Setting appropriate timeouts
|
||||
# - Configuring parallel execution
|
||||
|
||||
# Overall timeout for each mutant test
|
||||
# Increased from default to handle async tests
|
||||
timeout = 300 # 5 minutes per mutant
|
||||
|
||||
# Number of parallel jobs
|
||||
# Set to number of CPU cores for faster execution
|
||||
# jobs = 8 # Uncomment and adjust for your system
|
||||
|
||||
# Test tool to use
|
||||
# nextest is faster than cargo test
|
||||
test_tool = "nextest"
|
||||
|
||||
# Additional arguments to pass to the test command
|
||||
# --no-fail-fast ensures all tests run even if one fails
|
||||
test_args = ["--no-fail-fast"]
|
||||
|
||||
# Exclude patterns for files that shouldn't be mutated
|
||||
# These are low-value targets that produce noise
|
||||
exclude_globs = [
|
||||
# Test files - we test the tests, not mutate them
|
||||
"tests/**",
|
||||
"**/tests.rs",
|
||||
"benches/**",
|
||||
"examples/**",
|
||||
|
||||
# Proc macro crate
|
||||
"dagx-macros/**",
|
||||
|
||||
# Build scripts
|
||||
"build.rs",
|
||||
]
|
||||
|
||||
# Exclude specific functions/methods that shouldn't be mutated
|
||||
# These are low-value mutations that don't test correctness
|
||||
[[exclude_functions]]
|
||||
# Display trait implementations - not correctness-critical
|
||||
regex = ".*::fmt"
|
||||
reason = "Display/Debug implementations don't affect correctness"
|
||||
|
||||
[[exclude_functions]]
|
||||
# Clone implementations - trivial mutations
|
||||
regex = ".*::clone"
|
||||
reason = "Clone implementations are trivial"
|
||||
|
||||
[[exclude_functions]]
|
||||
# Default implementations - trivial
|
||||
regex = ".*::default"
|
||||
reason = "Default implementations are trivial"
|
||||
|
||||
# Specific functions to focus on (include)
|
||||
# These are the correctness-critical paths
|
||||
[[examine_functions]]
|
||||
# Core DAG execution logic
|
||||
regex = "DagRunner::run"
|
||||
reason = "Critical: Core DAG execution orchestration"
|
||||
|
||||
[[examine_functions]]
|
||||
# Task addition and dependency tracking
|
||||
regex = "DagRunner::add_task"
|
||||
reason = "Critical: Task registration and node creation"
|
||||
|
||||
[[examine_functions]]
|
||||
# Dependency management
|
||||
regex = "TaskBuilder::depends_on"
|
||||
reason = "Critical: Dependency edge creation"
|
||||
|
||||
[[examine_functions]]
|
||||
# Output extraction and type checking
|
||||
regex = "DagRunner::get"
|
||||
reason = "Critical: Type-safe output retrieval"
|
||||
|
||||
[[examine_functions]]
|
||||
# Node execution with listeners
|
||||
regex = "ExecutableNode::execute_with_listeners"
|
||||
reason = "Critical: Task execution and value propagation"
|
||||
|
||||
[[examine_functions]]
|
||||
# Listener extraction from announcements
|
||||
regex = "ExtractFromListeners::extract"
|
||||
reason = "Critical: Input extraction for dependent tasks"
|
||||
|
||||
[[examine_functions]]
|
||||
# Dependency tuple conversion
|
||||
regex = "DepsTuple::to_node_ids"
|
||||
reason = "Critical: Dependency tracking correctness"
|
||||
|
||||
# Minimum test time threshold
|
||||
# If tests complete faster than this, warn about insufficient testing
|
||||
# minimum_test_time = 1.0 # Uncomment to enable
|
||||
|
||||
# Output formats
|
||||
# Generate both text and JSON for tracking
|
||||
# output_formats = ["text", "json"] # Uncomment to enable
|
||||
|
||||
# Baseline
|
||||
# Compare mutations against a baseline to detect regressions
|
||||
# baseline = "main" # Uncomment to enable baseline comparison
|
||||
|
||||
# Notes on configuration strategy:
|
||||
#
|
||||
# 1. FOCUS: We explicitly list critical functions to mutate
|
||||
# - DAG execution: run(), execute_with_listeners()
|
||||
# - Dependency tracking: add_task(), depends_on(), to_node_ids()
|
||||
# - Type safety: get() with downcasting
|
||||
# - Value propagation: extract() from listeners
|
||||
#
|
||||
# 2. EXCLUDE: We exclude low-value targets
|
||||
# - Display/Debug impls: String formatting doesn't affect correctness
|
||||
# - Clone/Default impls: Trivial, unlikely to have bugs
|
||||
# - Tests themselves: We test production code, not tests
|
||||
# - Proc macros: Generated code tested via integration
|
||||
#
|
||||
# 3. TIMEOUT: 5 minutes is generous
|
||||
# - Most mutants complete in seconds
|
||||
# - Allows for async tests with real execution
|
||||
# - Catches actual deadlocks without false positives
|
||||
#
|
||||
# 4. NEXTEST: Faster test execution
|
||||
# - Parallel test execution
|
||||
# - Better output formatting
|
||||
# - Improved reliability
|
||||
#
|
||||
# 5. NO-FAIL-FAST: See all failures
|
||||
# - Don't stop at first test failure
|
||||
# - Get complete picture of mutation impact
|
||||
# - Better for debugging
|
||||
#
|
||||
# Expected mutation counts with this configuration:
|
||||
# - Target: ~30-50 mutants for core logic
|
||||
# - Score: 90%+ (high bar for DAG correctness)
|
||||
#
|
||||
# Runtime with this configuration:
|
||||
# - Current: ~5-10 minutes
|
||||
# - With nextest: ~30% faster
|
||||
@@ -5,6 +5,49 @@ All notable changes to dagx will be documented in this file.
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [Unreleased]
|
||||
|
||||
## [0.3.2] - 2025-10-10
|
||||
|
||||
[View changes](https://github.com/swaits/dagx/compare/v0.3.1...v0.3.2)
|
||||
|
||||
### Changed
|
||||
|
||||
- **Removed Cargo.lock from version control**
|
||||
- Library crates should not commit Cargo.lock files
|
||||
- Only application crates need Cargo.lock for reproducible builds
|
||||
- Added Cargo.lock to .gitignore
|
||||
- Follows Rust best practices as documented in the Cargo book
|
||||
|
||||
## [0.3.1] - 2025-10-10
|
||||
|
||||
[View changes](https://github.com/swaits/dagx/compare/v0.3.0...v0.3.1)
|
||||
|
||||
### Fixed
|
||||
|
||||
- **Custom types now work automatically!** The `#[task]` macro now generates inline `extract_and_run()`
|
||||
methods with type-specific extraction logic. This means **ANY type** implementing `Clone + Send + Sync + 'static`
|
||||
works automatically without implementing any special traits.
|
||||
- Previously, only types with `ExtractInput` implementations worked (primitives, String, Vec, HashMap, etc.)
|
||||
- Now, custom user types work seamlessly - just derive `Clone` and you're good to go!
|
||||
- No `ExtractInput` trait implementation needed for custom types
|
||||
- Nested structs, collections of custom types, and complex type hierarchies all work out of the box
|
||||
- New example: `examples/custom_types.rs` demonstrating custom types flowing through a DAG
|
||||
|
||||
### Changed
|
||||
|
||||
- **Documentation improvements**
|
||||
- Updated README.md to highlight custom type support as a key feature
|
||||
- Added "Custom Types" section to Core Concepts with examples
|
||||
- Added `custom_types.rs` to the examples list
|
||||
- Updated feature list to emphasize "Works with ANY type"
|
||||
|
||||
### Internal
|
||||
|
||||
- `ExtractInput` trait is now only used by the internal `task_fn` test helper
|
||||
- The `#[task]` macro generates custom extraction logic per task, bypassing `ExtractInput` entirely
|
||||
- This architectural change enables ANY type to work without requiring trait implementations
|
||||
|
||||
## [0.3.0] - 2025-10-08
|
||||
|
||||
[View changes](https://github.com/swaits/dagx/compare/v0.2.3...v0.3.0)
|
||||
@@ -255,6 +298,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
||||
- macOS (compatible)
|
||||
- Windows (compatible)
|
||||
|
||||
[0.3.2]: https://github.com/swaits/dagx/tree/v0.3.2
|
||||
[0.3.1]: https://github.com/swaits/dagx/tree/v0.3.1
|
||||
[0.3.0]: https://github.com/swaits/dagx/tree/v0.3.0
|
||||
[0.2.3]: https://github.com/swaits/dagx/tree/v0.2.3
|
||||
[0.2.2]: https://github.com/swaits/dagx/tree/v0.2.2
|
||||
|
||||
@@ -1 +0,0 @@
|
||||
We track work in Beads instead of Markdown. Run \`bd quickstart\` to see how.
|
||||
+2
-5
@@ -3,7 +3,7 @@ members = ["dagx-macros"]
|
||||
|
||||
[package]
|
||||
name = "dagx"
|
||||
version = "0.3.0"
|
||||
version = "0.3.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.81"
|
||||
authors = ["Stephen Waits <steve@waits.net>"]
|
||||
@@ -25,10 +25,9 @@ tracing = ["dep:tracing"]
|
||||
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin)', 'cfg(tarpaulin_include)'] }
|
||||
|
||||
[dependencies]
|
||||
announcement = "0.1.0"
|
||||
futures = "0.3"
|
||||
parking_lot = "0.12"
|
||||
dagx-macros = { version = "0.3.0", path = "./dagx-macros" }
|
||||
dagx-macros = { version = "0.3.2", path = "./dagx-macros" }
|
||||
tracing = { version = "0.1", optional = true }
|
||||
|
||||
[dev-dependencies]
|
||||
@@ -36,8 +35,6 @@ tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"
|
||||
async-std = { version = "1", features = ["attributes"] }
|
||||
smol = "2"
|
||||
criterion = "0.7"
|
||||
# Property-based testing
|
||||
proptest = "1"
|
||||
# Quirky async runtimes for edge-case testing
|
||||
async-executor = "1.13"
|
||||
pollster = "0.4"
|
||||
|
||||
@@ -0,0 +1,109 @@
|
||||
# Documentation Update Summary - Custom Type Support
|
||||
|
||||
## Context
|
||||
|
||||
Fixed a significant issue where dagx only worked with types that had `ExtractInput` trait implementations (primitives, String, Vec, HashMap, etc.). Custom user types didn't work.
|
||||
|
||||
**The Fix**: The `#[task]` macro now generates inline `extract_and_run()` methods with type-specific extraction logic. This means **ANY type** implementing `Clone + Send + Sync + 'static` works automatically - no trait implementations required!
|
||||
|
||||
## Files Updated
|
||||
|
||||
### 1. **README.md**
|
||||
- **Features section**: Added "Works with ANY type" as a key feature (line 117)
|
||||
- **Core Concepts**: Added new "Custom Types" section with example (lines 190-218)
|
||||
- **Examples**: Added `custom_types.rs` to advanced examples list (line 232)
|
||||
|
||||
### 2. **CHANGELOG.md**
|
||||
- **[Unreleased] section**: Added comprehensive documentation of this improvement (lines 8-33)
|
||||
- Explained the change from ExtractInput to macro-generated extraction
|
||||
- Highlighted that custom types now work seamlessly
|
||||
- Noted new example file
|
||||
|
||||
### 3. **src/lib.rs** (Main library documentation)
|
||||
- **Features**: Added custom type support as third bullet point (lines 13-15)
|
||||
- **Examples section**: Added new "Custom Types" example showing usage (lines 439-484)
|
||||
|
||||
### 4. **src/task.rs** (Task trait documentation)
|
||||
- **Trait docs**: Updated to mention macro generates extraction logic (lines 10-15)
|
||||
- Added note that custom types work automatically
|
||||
|
||||
### 5. **src/extract.rs** (ExtractInput trait documentation)
|
||||
- **Module docs**: Updated to clarify this is now legacy/internal only (lines 1-33)
|
||||
- Explained modern approach via `#[task]` macro vs legacy ExtractInput trait
|
||||
- Noted ExtractInput only used by internal `task_fn` helper
|
||||
|
||||
### 6. **dagx-macros/src/lib.rs** (Macro documentation)
|
||||
- **Macro docs**: Added emphasis on type-specific extraction generation (lines 15-22)
|
||||
- Highlighted that custom types work without trait implementations
|
||||
|
||||
### 7. **examples/custom_types.rs** (NEW FILE)
|
||||
- **Created comprehensive example** demonstrating custom type support
|
||||
- Shows nested custom types (User, Project, TeamMember, Team)
|
||||
- Demonstrates types flowing through multiple task layers
|
||||
- Full working example with detailed documentation comments
|
||||
- Expected output documented in header
|
||||
|
||||
## Key Messages in Documentation
|
||||
|
||||
### What Changed
|
||||
1. **Before**: Only types with ExtractInput implementations worked (primitives + standard types)
|
||||
2. **After**: ANY type with `Clone + Send + Sync + 'static` works automatically
|
||||
|
||||
### Why This Matters
|
||||
- No trait boilerplate for custom types
|
||||
- Nested structs, collections of custom types all work seamlessly
|
||||
- Simpler mental model - just derive Clone and go!
|
||||
- More ergonomic API
|
||||
|
||||
### How It Works
|
||||
- `#[task]` macro generates custom `extract_and_run()` implementation per task
|
||||
- Type-specific extraction logic inlined at compile time
|
||||
- No runtime overhead compared to ExtractInput approach
|
||||
- ExtractInput now only used internally by test helper
|
||||
|
||||
## Documentation Gaps Remaining
|
||||
|
||||
None found. The following were checked and updated:
|
||||
- ✅ README.md (main documentation)
|
||||
- ✅ CHANGELOG.md (version history)
|
||||
- ✅ src/lib.rs (API documentation)
|
||||
- ✅ src/task.rs (Task trait docs)
|
||||
- ✅ src/extract.rs (ExtractInput trait docs)
|
||||
- ✅ dagx-macros/src/lib.rs (macro documentation)
|
||||
- ✅ examples/ (new example added)
|
||||
- ✅ CONTRIBUTING.md (no changes needed - general guidelines)
|
||||
- ✅ docs/*.md (no changes needed - architectural docs)
|
||||
|
||||
## Example Test Status
|
||||
|
||||
The new `custom_types.rs` example:
|
||||
- Created at `/home/swaits/Code/dagx/examples/custom_types.rs`
|
||||
- Demonstrates User, Project, TeamMember, Team custom types
|
||||
- Shows nested types and collections
|
||||
- Needs to be compiled and tested: `cargo run --example custom_types`
|
||||
|
||||
## Recommendations
|
||||
|
||||
### Before Release
|
||||
1. **Compile and test** the new example: `cargo run --example custom_types`
|
||||
2. **Update lib.rs last audited date** (currently shows 2025-10-06 for v0.1.0)
|
||||
3. **Review CHANGELOG** to ensure version number and date are correct when releasing
|
||||
4. Consider adding a **migration guide** if there are any breaking changes
|
||||
|
||||
### Future Documentation Enhancements
|
||||
1. Consider adding a "Type Requirements" section to README explaining `Clone + Send + Sync + 'static`
|
||||
2. Could add more examples showing:
|
||||
- Enums as task outputs
|
||||
- Generic custom types
|
||||
- Types with lifetimes (showing why 'static is needed)
|
||||
|
||||
## Summary
|
||||
|
||||
All documentation has been updated to accurately reflect the new custom type support. The key improvement - that ANY type works automatically with just `Clone + Send + Sync` - is now prominently featured in:
|
||||
- Main features list
|
||||
- Core concepts with examples
|
||||
- API documentation
|
||||
- Macro documentation
|
||||
- A comprehensive new example
|
||||
|
||||
The documentation emphasizes this is a **major ergonomic improvement** that makes dagx significantly easier to use with custom types.
|
||||
@@ -1,134 +0,0 @@
|
||||
# Performance Regression Analysis: ETL Benchmark
|
||||
|
||||
## Summary
|
||||
|
||||
The ETL scaling benchmark regressed by ~17% (3.24ms -> 3.30ms for 10k records) due to expensive data cloning when using the announcement crate.
|
||||
|
||||
## Root Cause
|
||||
|
||||
In both `dagx-macros/src/lib.rs:327` and `src/task.rs:253`:
|
||||
|
||||
```rust
|
||||
let _ = announcer_typed.announce((*output_arc).clone());
|
||||
```
|
||||
|
||||
This dereferences the Arc and clones the actual data, which is extremely expensive for large types.
|
||||
|
||||
### For ETL benchmark with 10k strings:
|
||||
|
||||
**Current (Broken) Flow:**
|
||||
1. Task produces `Vec<String>` (10k items)
|
||||
2. Wrap in `Arc<Vec<String>>`
|
||||
3. **Announce `(*arc).clone()`** → Full Vec clone #1 (EXPENSIVE!)
|
||||
4. Announcement stores cloned Vec in `OnceLock<Vec<String>>`
|
||||
5. Each of 3 listeners calls `.listen()` → Vec clone #2, #3, #4 (EXPENSIVE!)
|
||||
|
||||
**Total**: 4 full Vec clones = 40k string clones
|
||||
|
||||
### Why other benchmarks improved:
|
||||
|
||||
Most benchmarks use primitive types (i32, etc.) where cloning is cheap. The announcement crate's broadcast pattern is more efficient than multiple oneshot channels for small types.
|
||||
|
||||
## The Fix
|
||||
|
||||
Change announcements to use `Arc<T>` instead of `T`:
|
||||
|
||||
**New (Correct) Flow:**
|
||||
1. Task produces `Vec<String>` (10k items)
|
||||
2. Wrap in `Arc<Vec<String>>`
|
||||
3. **Announce `Arc::clone(&arc)`** → Arc clone (CHEAP!)
|
||||
4. Announcement stores Arc in `OnceLock<Arc<Vec<String>>>`
|
||||
5. Each of 3 listeners calls `.listen()` → Arc clone #2, #3, #4 (CHEAP!)
|
||||
|
||||
**Total**: 4 Arc clones = just refcount increments (~40ns total)
|
||||
|
||||
## Required Changes
|
||||
|
||||
### 1. dagx-macros/src/lib.rs
|
||||
|
||||
```rust
|
||||
// Line 327: Change announcement call
|
||||
- let _ = announcer_typed.announce((*output_arc).clone());
|
||||
+ let _ = announcer_typed.announce(Arc::clone(&output_arc));
|
||||
|
||||
// Line 325: Change announcer type
|
||||
- .downcast::<announcement::Announcer<#output_type>>()
|
||||
+ .downcast::<announcement::Announcer<Arc<#output_type>>>()
|
||||
|
||||
// Line 352: Change announcement creation
|
||||
- let (announcer, announcement) = announcement::Announcement::<#output_type>::new();
|
||||
+ let (announcer, announcement) = announcement::Announcement::<Arc<#output_type>>::new();
|
||||
|
||||
// Line 257: Change listener type in parameter extraction
|
||||
- .downcast_ref::<announcement::Listener<#ty>>()
|
||||
+ .downcast_ref::<announcement::Listener<Arc<#ty>>>()
|
||||
|
||||
// Line 361: Change announcement downcast
|
||||
- .downcast_ref::<announcement::Announcement<#output_type>>()
|
||||
+ .downcast_ref::<announcement::Announcement<Arc<#output_type>>>()
|
||||
```
|
||||
|
||||
### 2. src/task.rs
|
||||
|
||||
```rust
|
||||
// Line 253: Change announcement call
|
||||
- let _ = announcer_typed.announce((*output_arc).clone());
|
||||
+ let _ = announcer_typed.announce(Arc::clone(&output_arc));
|
||||
|
||||
// Line 249: Change announcer type
|
||||
- .downcast::<announcement::Announcer<O>>()
|
||||
+ .downcast::<announcement::Announcer<Arc<O>>>()
|
||||
|
||||
// Line 261: Change announcement creation
|
||||
- let (announcer, announcement) = announcement::Announcement::<O>::new();
|
||||
+ let (announcer, announcement) = announcement::Announcement::<Arc<O>>::new();
|
||||
|
||||
// Line 270: Change announcement downcast
|
||||
- .downcast_ref::<announcement::Announcement<O>>()
|
||||
+ .downcast_ref::<announcement::Announcement<Arc<O>>>()
|
||||
|
||||
// Lines 34, 57, 67, 84, 102, 125: Change all listener downcasts from:
|
||||
- .downcast::<announcement::Listener<T>>()
|
||||
+ .downcast::<announcement::Listener<Arc<T>>>()
|
||||
|
||||
// Then unwrap the Arc after listening:
|
||||
let arc_value = listener.listen().await?;
|
||||
let value = (*arc_value).clone(); // Only clone primitives, Arc is already cheap for complex types
|
||||
```
|
||||
|
||||
### 3. Handle Arc unwrapping in ExtractFromListeners
|
||||
|
||||
For each type implementation, extract from `Listener<Arc<T>>` and then dereference:
|
||||
|
||||
```rust
|
||||
impl ExtractFromListeners for Vec<T> {
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
let listener = *listeners.pop().unwrap()
|
||||
.downcast::<announcement::Listener<Arc<Vec<T>>>>() // Now Arc<Vec<T>>
|
||||
.map_err(...)?;
|
||||
let arc_vec = listener.listen().await.map_err(...)?;
|
||||
Ok((*arc_vec).clone()) // Clone the Vec out of the Arc
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
Actually, a better approach: keep the Arc! For types that are expensive to clone, just return the Arc:
|
||||
|
||||
```rust
|
||||
// For expensive types, use Arc directly in function signatures
|
||||
async fn run(&self, data: &Arc<Vec<String>>) -> Result { ... }
|
||||
```
|
||||
|
||||
Or even simpler: change ExtractFromListeners to extract as the Arc type directly, and modify the macro's parameter extraction to work with Arc types.
|
||||
|
||||
## Expected Performance Impact
|
||||
|
||||
- **ETL 10k benchmark**: Should improve by ~17% back to baseline (or better)
|
||||
- **Other benchmarks**: Should remain the same or slightly improve (Arc clone overhead is minimal)
|
||||
- **Memory efficiency**: Better! Arc refcounting instead of deep copies
|
||||
|
||||
## Why This Wasn't Caught
|
||||
|
||||
- Most tests use small primitive types where clone() is fast
|
||||
- The regression only shows up with large data structures (Vec, HashMap, etc.)
|
||||
- The ETL benchmark with 10k records is the perfect stress test for this issue
|
||||
@@ -114,6 +114,7 @@ See [design philosophy](docs/DESIGN_PHILOSOPHY.md) for details.
|
||||
|
||||
- **Compile-time cycle prevention**: Type system makes cycles impossible — no runtime checks
|
||||
- **Compile-time type safety**: Dependencies validated at compile time, no runtime type errors
|
||||
- **Works with ANY type**: Custom types work automatically — just `Clone + Send + Sync`, no trait implementations needed
|
||||
- **Runtime-agnostic**: Works with Tokio, async-std, smol, or any async runtime
|
||||
- **True parallelism**: Tasks spawn to multiple threads for genuine parallel execution
|
||||
- **Type-state pattern**: API prevents incorrect wiring through compile-time errors
|
||||
@@ -186,6 +187,36 @@ let task = dag.add_task(my_task).depends_on(&upstream);
|
||||
let task = dag.add_task(my_task).depends_on((&upstream1, &upstream2));
|
||||
```
|
||||
|
||||
### Custom Types
|
||||
|
||||
**dagx works with ANY type automatically!** As long as your type implements `Clone + Send + Sync + 'static`, it works seamlessly:
|
||||
|
||||
```rust
|
||||
#[derive(Clone)] // Just derive Clone!
|
||||
struct User {
|
||||
name: String,
|
||||
age: u32,
|
||||
}
|
||||
|
||||
#[task]
|
||||
impl CreateUser {
|
||||
async fn run(&self) -> User {
|
||||
User { name: "Alice".to_string(), age: 30 }
|
||||
}
|
||||
}
|
||||
|
||||
#[task]
|
||||
impl ProcessUser {
|
||||
async fn run(user: &User) -> String {
|
||||
format!("{} is {} years old", user.name, user.age)
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**No trait implementations needed!** The `#[task]` macro generates type-specific extraction logic automatically. This includes nested structs, collections, enums, and any other type you define.
|
||||
|
||||
See [`custom_types.rs`](examples/custom_types.rs) for a complete example with nested custom types.
|
||||
|
||||
### Runtime Agnostic
|
||||
|
||||
dagx works with any async runtime. Provide a spawner function to `run()`:
|
||||
@@ -228,6 +259,7 @@ Real-world patterns:
|
||||
- [`circuit_breaker.rs`](examples/circuit_breaker.rs) - Circuit breaker pattern for resilient systems
|
||||
- [`complex_dag.rs`](examples/complex_dag.rs) - Multi-layer workflows with complex dependencies
|
||||
- [`conditional_workflow.rs`](examples/conditional_workflow.rs) - Conditional execution with Result types
|
||||
- [`custom_types.rs`](examples/custom_types.rs) - Using your own custom types (no trait implementations needed!)
|
||||
- [`data_pipeline.rs`](examples/data_pipeline.rs) - ETL data processing pipeline
|
||||
- [`debug_tracing.rs`](examples/debug_tracing.rs) - Debug metadata and task naming
|
||||
- [`error_handling.rs`](examples/error_handling.rs) - Error propagation and recovery
|
||||
@@ -237,7 +269,7 @@ Real-world patterns:
|
||||
- [`timeout.rs`](examples/timeout.rs) - Managing task timeouts
|
||||
- [`tracing_example.rs`](examples/tracing_example.rs) - Observability with tracing support
|
||||
|
||||
Run any example: `cargo run --example data_pipeline`
|
||||
Run any example: `cargo run --example custom_types`
|
||||
|
||||
## Advanced Topics
|
||||
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
//! DAG creation benchmarks
|
||||
|
||||
use criterion::Criterion;
|
||||
use dagx::DagRunner;
|
||||
use dagx::{DagRunner, Task};
|
||||
use std::hint::black_box;
|
||||
|
||||
// Simple task for benchmarking
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
//! Basic DAG execution benchmarks
|
||||
|
||||
use criterion::Criterion;
|
||||
use dagx::{task_fn, DagRunner};
|
||||
use dagx::{task_fn, DagRunner, Task};
|
||||
|
||||
pub fn bench_dag_execution(c: &mut Criterion) {
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
|
||||
@@ -37,7 +37,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
|
||||
|
||||
// T1: Parse and validate
|
||||
let validate = dag
|
||||
.add_task(task_fn(move |data: std::sync::Arc<Vec<String>>| async move {
|
||||
.add_task(task_fn(move |data: Vec<String>| async move {
|
||||
data.iter()
|
||||
.filter(|s| {
|
||||
s.contains("\"id\"")
|
||||
@@ -50,7 +50,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
|
||||
|
||||
// T2: Extract scores
|
||||
let scores = dag
|
||||
.add_task(task_fn(move |data: std::sync::Arc<Vec<String>>| async move {
|
||||
.add_task(task_fn(move |data: Vec<String>| async move {
|
||||
data.iter()
|
||||
.filter_map(|s| {
|
||||
s.split("\"score\":").nth(1).and_then(|part| {
|
||||
@@ -63,7 +63,7 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
|
||||
|
||||
// T3: Extract names
|
||||
let names = dag
|
||||
.add_task(task_fn(move |data: std::sync::Arc<Vec<String>>| async move {
|
||||
.add_task(task_fn(move |data: Vec<String>| async move {
|
||||
data.iter()
|
||||
.filter_map(|s| {
|
||||
s.split("\"name\":\"")
|
||||
@@ -77,12 +77,12 @@ pub fn bench_etl_pipeline(c: &mut Criterion) {
|
||||
|
||||
// Load: Aggregate all transformations
|
||||
dag.add_task(task_fn(
|
||||
move |(valid, scores, names): (std::sync::Arc<usize>, std::sync::Arc<Vec<i32>>, std::sync::Arc<Vec<String>>)| async move {
|
||||
move |(valid, scores, names): (usize, Vec<i32>, Vec<String>)| async move {
|
||||
let avg_score =
|
||||
scores.iter().sum::<i32>() / scores.len().max(1) as i32;
|
||||
format!(
|
||||
"Processed {} valid records, {} names, avg score: {}",
|
||||
*valid,
|
||||
valid,
|
||||
names.len(),
|
||||
avg_score
|
||||
)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[package]
|
||||
name = "dagx-macros"
|
||||
version = "0.3.0"
|
||||
version = "0.3.2"
|
||||
edition = "2021"
|
||||
rust-version = "1.81"
|
||||
authors = ["Stephen Waits <steve@waits.net>"]
|
||||
|
||||
+279
-124
@@ -1,21 +1,26 @@
|
||||
//! Procedural macros for dagx
|
||||
//!
|
||||
//! This crate provides the `#[task]` attribute macro that automatically generates
|
||||
//! an `execute()` method by deriving Input and Output types from the `run()` method signature.
|
||||
//! This crate provides the `#[task]` attribute macro that automatically implements
|
||||
//! the `Task` trait by deriving Input and Output types from the `run()` method signature.
|
||||
|
||||
use proc_macro::TokenStream;
|
||||
use quote::quote;
|
||||
use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType, Type};
|
||||
|
||||
/// Attribute macro to automatically generate the execution logic for a task.
|
||||
/// Attribute macro to automatically implement the `Task` trait.
|
||||
///
|
||||
/// Apply this to an `impl` block containing a `run()` method (sync or async). The macro:
|
||||
/// - Derives input types from the `run()` signature parameters
|
||||
/// - Generates an `execute()` method that handles listener extraction and announcement
|
||||
/// - Derives `Input` and `Output` types from the `run()` signature
|
||||
/// - Automatically implements the `Task` trait
|
||||
/// - **Generates type-specific extraction logic** - works with ANY type (Clone + Send + Sync)!
|
||||
/// - Supports both sync and async run methods
|
||||
/// - Supports stateless (no self) and stateful (&self, &mut self) tasks
|
||||
/// - Handles various input patterns (no inputs, single input, multiple inputs)
|
||||
///
|
||||
/// **Key Feature**: Custom types work automatically without implementing any traits!
|
||||
/// The macro generates inline extraction logic in `extract_and_run()` specific to your
|
||||
/// task's parameter types. Just derive `Clone` on your types and they'll work seamlessly.
|
||||
///
|
||||
/// # Task Patterns
|
||||
///
|
||||
/// The `#[task]` macro supports three patterns based on state requirements:
|
||||
@@ -25,7 +30,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
|
||||
/// Unit structs for pure computations. Use **no `self` parameter**:
|
||||
///
|
||||
/// ```ignore
|
||||
/// use dagx::task;
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Add;
|
||||
///
|
||||
@@ -42,7 +47,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
|
||||
/// Tasks that read configuration or constant data. Use **`&self`**:
|
||||
///
|
||||
/// ```ignore
|
||||
/// use dagx::task;
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Multiplier {
|
||||
/// factor: i32,
|
||||
@@ -61,7 +66,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
|
||||
/// Tasks that accumulate or modify state. Use **`&mut self`**:
|
||||
///
|
||||
/// ```ignore
|
||||
/// use dagx::task;
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Counter {
|
||||
/// count: i32,
|
||||
@@ -81,7 +86,7 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
|
||||
/// ## No Inputs (Source Tasks)
|
||||
///
|
||||
/// ```ignore
|
||||
/// use dagx::task;
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct LoadData {
|
||||
/// value: i32,
|
||||
@@ -93,51 +98,110 @@ use syn::{parse_macro_input, FnArg, ImplItem, ItemImpl, Pat, PatType, ReturnType
|
||||
/// self.value
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
/// ```ignore
|
||||
///
|
||||
/// ## Single Input
|
||||
///
|
||||
/// ```ignore
|
||||
/// use dagx::task;
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Double;
|
||||
///
|
||||
/// #[task]
|
||||
/// impl Double {
|
||||
/// async fn run(input: &i32) -> i32 {
|
||||
/// async fn run(&mut self, input: &i32) -> i32 {
|
||||
/// input * 2
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
/// ```ignore
|
||||
///
|
||||
/// ## Multiple Inputs (up to 8)
|
||||
///
|
||||
/// ```ignore
|
||||
/// use dagx::task;
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Combine;
|
||||
///
|
||||
/// #[task]
|
||||
/// impl Combine {
|
||||
/// async fn run(a: &i32, b: &String, c: &bool) -> String {
|
||||
/// async fn run(&mut self, a: &i32, b: &String, c: &bool) -> String {
|
||||
/// format!("{}: {} ({})", b, a, c)
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
/// ```ignore
|
||||
///
|
||||
/// # Requirements
|
||||
///
|
||||
/// - The impl block must contain exactly one `async fn run()` method
|
||||
/// - The `run()` method can be stateless (no self parameter) or stateful (`&mut self`)
|
||||
/// - All input parameters must be references (e.g., `&i32`, not `i32`)
|
||||
/// - The macro requires `Task` to be in scope: `use dagx::Task;`
|
||||
/// - For stateless tasks, the struct must implement `Default` (e.g., unit structs)
|
||||
///
|
||||
/// # Generated Code
|
||||
///
|
||||
/// The macro transforms your implementation into an `execute()` method that:
|
||||
/// - Extracts inputs from announcement listeners
|
||||
/// - Calls your run() method
|
||||
/// - Announces the result to dependents
|
||||
/// - Returns the Arc-wrapped output
|
||||
/// The macro transforms your implementation into a full `Task` trait implementation.
|
||||
///
|
||||
/// For stateless tasks (no self parameter):
|
||||
///
|
||||
/// ```ignore
|
||||
/// // Your code:
|
||||
/// #[task]
|
||||
/// impl Add {
|
||||
/// async fn run(a: &i32, b: &i32) -> i32 {
|
||||
/// a + b
|
||||
/// }
|
||||
/// }
|
||||
///
|
||||
/// // Generated:
|
||||
/// impl Task for Add {
|
||||
/// type Input = (i32, i32);
|
||||
/// type Output = i32;
|
||||
///
|
||||
/// async fn run(&mut self, input: Self::Input) -> Self::Output {
|
||||
/// let (a, b) = input;
|
||||
/// Self::run_impl(&a, &b).await
|
||||
/// }
|
||||
/// }
|
||||
///
|
||||
/// impl Add {
|
||||
/// #[inline]
|
||||
/// async fn run_impl(a: &i32, b: &i32) -> i32 {
|
||||
/// a + b
|
||||
/// }
|
||||
/// }
|
||||
/// ```ignore
|
||||
///
|
||||
/// For stateful tasks (with &mut self):
|
||||
///
|
||||
/// ```ignore
|
||||
/// // Your code:
|
||||
/// #[task]
|
||||
/// impl Counter {
|
||||
/// async fn run(&mut self, inc: &i32) -> i32 {
|
||||
/// self.count += inc;
|
||||
/// self.count
|
||||
/// }
|
||||
/// }
|
||||
///
|
||||
/// // Generated:
|
||||
/// impl Task for Counter {
|
||||
/// type Input = i32;
|
||||
/// type Output = i32;
|
||||
///
|
||||
/// async fn run(&mut self, input: Self::Input) -> Self::Output {
|
||||
/// let inc = input;
|
||||
/// self.run_impl(&inc).await
|
||||
/// }
|
||||
/// }
|
||||
///
|
||||
/// impl Counter {
|
||||
/// async fn run_impl(&mut self, inc: &i32) -> i32 {
|
||||
/// self.count += inc;
|
||||
/// self.count
|
||||
/// }
|
||||
/// }
|
||||
/// ```ignore
|
||||
#[proc_macro_attribute]
|
||||
pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
|
||||
let impl_block = parse_macro_input!(item as ItemImpl);
|
||||
@@ -161,7 +225,7 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
|
||||
"impl block must contain a run() method\n\n\
|
||||
Expected signature: fn run(&mut self, ...) -> OutputType\n\
|
||||
or: async fn run(&mut self, ...) -> OutputType\n\
|
||||
The #[task] macro requires a run() method to implement task execution.",
|
||||
The #[task] macro requires a run() method to implement the Task trait.",
|
||||
)
|
||||
.to_compile_error()
|
||||
.into();
|
||||
@@ -228,6 +292,31 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
|
||||
ReturnType::Type(_, ty) => ty.clone(),
|
||||
};
|
||||
|
||||
// Build Input type based on parameter count
|
||||
let input_type = match params.len() {
|
||||
0 => quote! { () },
|
||||
1 => {
|
||||
let (_name, ty) = ¶ms[0];
|
||||
quote! { #ty }
|
||||
}
|
||||
_ => {
|
||||
let types: Vec<_> = params.iter().map(|(_, ty)| ty).collect();
|
||||
quote! { ( #(#types),* ) }
|
||||
}
|
||||
};
|
||||
|
||||
// Generate parameter destructuring for the wrapper run() method
|
||||
let (param_destructure, param_refs) = if params.is_empty() {
|
||||
(quote! { _ }, quote! {})
|
||||
} else if params.len() == 1 {
|
||||
let (name, _) = ¶ms[0];
|
||||
(quote! { #name }, quote! { &#name })
|
||||
} else {
|
||||
let names: Vec<_> = params.iter().map(|(name, _)| name).collect();
|
||||
let refs: Vec<_> = params.iter().map(|(name, _)| quote! { &#name }).collect();
|
||||
(quote! { ( #(#names),* ) }, quote! { #(#refs),* })
|
||||
};
|
||||
|
||||
// Clone the run method and rename it to run_impl
|
||||
let mut run_impl_method = run_method.clone();
|
||||
run_impl_method.sig.ident = syn::Ident::new("run_impl", run_method.sig.ident.span());
|
||||
@@ -239,129 +328,195 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
|
||||
.iter()
|
||||
.any(|arg| matches!(arg, FnArg::Receiver(_)));
|
||||
|
||||
// Generate listener extraction code
|
||||
let listener_extractions: Vec<_> = params
|
||||
.iter()
|
||||
.enumerate()
|
||||
.map(|(i, (name, ty))| {
|
||||
// Generate extract_and_run implementation based on parameter count
|
||||
let extract_and_run_impl = match params.len() {
|
||||
0 => {
|
||||
// Zero parameters - no extraction needed
|
||||
quote! {
|
||||
let #name = {
|
||||
if listeners.len() <= #i {
|
||||
return Err(format!(
|
||||
"Expected at least {} listener(s), got {}",
|
||||
#i + 1,
|
||||
listeners.len()
|
||||
));
|
||||
fn extract_and_run(
|
||||
self,
|
||||
_receivers: Vec<Box<dyn std::any::Any + Send>>,
|
||||
) -> impl std::future::Future<Output = Result<Self::Output, String>> + Send {
|
||||
async move {
|
||||
let input = ();
|
||||
Ok(self.run(input).await)
|
||||
}
|
||||
let listener = listeners[#i]
|
||||
.downcast_ref::<announcement::Listener<std::sync::Arc<#ty>>>()
|
||||
.ok_or_else(|| {
|
||||
format!("Type mismatch: expected Listener<Arc<{}>> at index {}",
|
||||
stringify!(#ty), #i)
|
||||
})?;
|
||||
listener.listen().await
|
||||
.map_err(|_| "Listener closed before receiving value")?
|
||||
};
|
||||
}
|
||||
}
|
||||
})
|
||||
.collect();
|
||||
|
||||
// Generate parameter refs for the run call
|
||||
// Dereference the Arc to get &T for the user's function
|
||||
let param_refs: Vec<_> = params.iter().map(|(name, _)| quote! { &*#name }).collect();
|
||||
|
||||
// Generate the execute method body based on whether we have self and async/sync
|
||||
let run_call = if has_self_receiver {
|
||||
if is_async {
|
||||
quote! { self.run_impl(#(#param_refs),*).await }
|
||||
} else {
|
||||
quote! { self.run_impl(#(#param_refs),*) }
|
||||
}
|
||||
} else if is_async {
|
||||
quote! { Self::run_impl(#(#param_refs),*).await }
|
||||
} else {
|
||||
quote! { Self::run_impl(#(#param_refs),*) }
|
||||
};
|
||||
|
||||
// Generate the self parameter for execute method
|
||||
let self_param = if has_self_receiver {
|
||||
quote! { mut self }
|
||||
} else {
|
||||
quote! { self }
|
||||
};
|
||||
|
||||
// Generate the Input type based on parameter count
|
||||
let input_type = match params.len() {
|
||||
0 => quote! { () },
|
||||
1 => {
|
||||
let (_, ty) = ¶ms[0];
|
||||
quote! { #ty }
|
||||
// Single parameter
|
||||
let param_type = ¶ms[0].1;
|
||||
quote! {
|
||||
fn extract_and_run(
|
||||
self,
|
||||
mut receivers: Vec<Box<dyn std::any::Any + Send>>,
|
||||
) -> impl std::future::Future<Output = Result<Self::Output, String>> + Send {
|
||||
async move {
|
||||
use futures::channel::oneshot;
|
||||
use std::sync::Arc;
|
||||
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
|
||||
let rx = *receivers.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<#param_type>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Arc<{}>", std::any::type_name::<#param_type>()))?;
|
||||
|
||||
let arc_value = rx.await
|
||||
.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
|
||||
let input = (*arc_value).clone();
|
||||
Ok(self.run(input).await)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
let types: Vec<_> = params.iter().map(|(_, ty)| ty).collect();
|
||||
quote! { (#(#types,)*) }
|
||||
// Multiple parameters
|
||||
let param_count = params.len();
|
||||
let param_types: Vec<_> = params.iter().map(|(_, ty)| ty).collect();
|
||||
let indices: Vec<_> = (0..param_count).collect();
|
||||
|
||||
// Generate unique receiver variable names
|
||||
let rx_vars: Vec<_> = (0..param_count)
|
||||
.map(|i| syn::Ident::new(&format!("rx_{}", i), proc_macro2::Span::call_site()))
|
||||
.collect();
|
||||
|
||||
// Create syn::Index for tuple field access (avoids the suffix warning)
|
||||
let syn_indices: Vec<_> = (0..param_count).map(syn::Index::from).collect();
|
||||
|
||||
quote! {
|
||||
fn extract_and_run(
|
||||
self,
|
||||
receivers: Vec<Box<dyn std::any::Any + Send>>,
|
||||
) -> impl std::future::Future<Output = Result<Self::Output, String>> + Send {
|
||||
async move {
|
||||
use futures::channel::oneshot;
|
||||
use std::sync::Arc;
|
||||
|
||||
let expected_count = #param_count;
|
||||
if receivers.len() != expected_count {
|
||||
return Err(format!("Expected {} dependencies, got {}", expected_count, receivers.len()));
|
||||
}
|
||||
|
||||
let mut iter = receivers.into_iter();
|
||||
|
||||
// Extract each receiver
|
||||
#(
|
||||
let #rx_vars = *iter.next()
|
||||
.ok_or_else(|| format!("Missing receiver at index {}", #indices))?
|
||||
.downcast::<oneshot::Receiver<Arc<#param_types>>>()
|
||||
.map_err(|_| format!("Type mismatch at index {}: expected Arc<{}>",
|
||||
#indices, std::any::type_name::<#param_types>()))?;
|
||||
)*
|
||||
|
||||
// Await all channels concurrently
|
||||
let arc_results = futures::join!(
|
||||
#(
|
||||
async move {
|
||||
#rx_vars.await.map_err(|_| format!("Channel {} closed", #indices))
|
||||
}
|
||||
),*
|
||||
);
|
||||
|
||||
// Clone inner values and build tuple
|
||||
let input = (#(
|
||||
(*arc_results.#syn_indices?).clone()
|
||||
),*);
|
||||
|
||||
Ok(self.run(input).await)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
// Generate the complete execute method and TaskExecute trait implementation
|
||||
let expanded = quote! {
|
||||
impl #struct_name {
|
||||
#[doc(hidden)]
|
||||
pub async fn execute(
|
||||
#self_param,
|
||||
listeners: Vec<Box<dyn std::any::Any + Send>>,
|
||||
announcer: Box<dyn std::any::Any + Send>,
|
||||
) -> Result<std::sync::Arc<dyn std::any::Any + Send + Sync>, String> {
|
||||
// Extract inputs from listeners
|
||||
#(#listener_extractions)*
|
||||
// Generate the Task trait implementation based on whether we have self and async/sync
|
||||
let expanded = if has_self_receiver {
|
||||
// Stateful task - consumes self but delegates to a method that borrows
|
||||
if is_async {
|
||||
// Async with self
|
||||
quote! {
|
||||
impl Task for #struct_name {
|
||||
type Input = #input_type;
|
||||
type Output = #output_type;
|
||||
|
||||
// Call the user's run method
|
||||
let output: #output_type = #run_call;
|
||||
async fn run(mut self, input: Self::Input) -> Self::Output {
|
||||
let #param_destructure = input;
|
||||
self.run_impl(#param_refs).await
|
||||
}
|
||||
|
||||
// Wrap in Arc for sharing
|
||||
let output_arc = std::sync::Arc::new(output);
|
||||
#extract_and_run_impl
|
||||
}
|
||||
|
||||
// Announce the result (announce the Arc, not a clone of the data)
|
||||
let announcer_typed = announcer
|
||||
.downcast::<announcement::Announcer<std::sync::Arc<#output_type>>>()
|
||||
.map_err(|_| "Type mismatch: announcer type does not match task output type")?;
|
||||
let _ = announcer_typed.announce(std::sync::Arc::clone(&output_arc));
|
||||
|
||||
// Return Arc-wrapped output
|
||||
Ok(output_arc as std::sync::Arc<dyn std::any::Any + Send + Sync>)
|
||||
impl #struct_name {
|
||||
#run_impl_method
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Sync with self - wrap in async block
|
||||
quote! {
|
||||
impl Task for #struct_name {
|
||||
type Input = #input_type;
|
||||
type Output = #output_type;
|
||||
|
||||
#run_impl_method
|
||||
async fn run(mut self, input: Self::Input) -> Self::Output {
|
||||
let #param_destructure = input;
|
||||
self.run_impl(#param_refs)
|
||||
}
|
||||
|
||||
#extract_and_run_impl
|
||||
}
|
||||
|
||||
impl #struct_name {
|
||||
#run_impl_method
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Stateless task
|
||||
if is_async {
|
||||
// Async stateless
|
||||
quote! {
|
||||
impl Task for #struct_name {
|
||||
type Input = #input_type;
|
||||
type Output = #output_type;
|
||||
|
||||
// Implement TaskExecute trait for type erasure
|
||||
impl dagx::TaskExecute for #struct_name {
|
||||
type Input = #input_type;
|
||||
type Output = #output_type;
|
||||
async fn run(self, input: Self::Input) -> Self::Output {
|
||||
let #param_destructure = input;
|
||||
Self::run_impl(#param_refs).await
|
||||
}
|
||||
|
||||
fn execute(
|
||||
self,
|
||||
listeners: Vec<Box<dyn std::any::Any + Send>>,
|
||||
announcer: Box<dyn std::any::Any + Send>,
|
||||
) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<std::sync::Arc<dyn std::any::Any + Send + Sync>, String>> + Send>> {
|
||||
Box::pin(async move {
|
||||
self.execute(listeners, announcer).await
|
||||
})
|
||||
#extract_and_run_impl
|
||||
}
|
||||
|
||||
impl #struct_name {
|
||||
#[inline]
|
||||
#run_impl_method
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Sync stateless
|
||||
quote! {
|
||||
impl Task for #struct_name {
|
||||
type Input = #input_type;
|
||||
type Output = #output_type;
|
||||
|
||||
fn create_announcement_erased() -> (Box<dyn std::any::Any + Send>, Box<dyn std::any::Any + Send>) {
|
||||
let (announcer, announcement) = announcement::Announcement::<std::sync::Arc<#output_type>>::new();
|
||||
(
|
||||
Box::new(announcer) as Box<dyn std::any::Any + Send>,
|
||||
Box::new(announcement) as Box<dyn std::any::Any + Send>,
|
||||
)
|
||||
}
|
||||
async fn run(self, input: Self::Input) -> Self::Output {
|
||||
let #param_destructure = input;
|
||||
Self::run_impl(#param_refs)
|
||||
}
|
||||
|
||||
fn create_listener_erased(announcement: &dyn std::any::Any) -> Box<dyn std::any::Any + Send> {
|
||||
let announcement = announcement
|
||||
.downcast_ref::<announcement::Announcement<std::sync::Arc<#output_type>>>()
|
||||
.expect("Type mismatch: announcement type does not match task output type");
|
||||
Box::new(announcement.listener()) as Box<dyn std::any::Any + Send>
|
||||
#extract_and_run_impl
|
||||
}
|
||||
|
||||
impl #struct_name {
|
||||
#[inline]
|
||||
#run_impl_method
|
||||
}
|
||||
}
|
||||
}
|
||||
};
|
||||
|
||||
@@ -30,7 +30,7 @@
|
||||
//! Result: 5
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
// Step 1: Define a source task (no dependencies)
|
||||
//
|
||||
|
||||
@@ -55,7 +55,7 @@
|
||||
//! Squared: 100
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
// Source task with a value field
|
||||
struct Source {
|
||||
|
||||
@@ -61,7 +61,7 @@
|
||||
//! Active: true
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
struct FetchName;
|
||||
|
||||
|
||||
@@ -52,7 +52,7 @@
|
||||
//! This proves dagx runs tasks on multiple CPU cores simultaneously.
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagResult, DagRunner};
|
||||
use dagx::{task, DagResult, DagRunner, Task};
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::time::sleep;
|
||||
|
||||
@@ -144,32 +144,22 @@ async fn main() -> DagResult<()> {
|
||||
};
|
||||
|
||||
// ============================================================================
|
||||
// SEQUENTIAL EXECUTION: Run tasks one at a time in separate DAG executions
|
||||
// SEQUENTIAL EXECUTION: Run the same 4 tasks one after another
|
||||
// ============================================================================
|
||||
println!("Running SEQUENTIAL computation (same work, one at a time)...");
|
||||
let (sequential_time, sequential_result) = {
|
||||
let start = Instant::now();
|
||||
|
||||
// Run each task separately to force sequential execution
|
||||
let dag1 = DagRunner::new();
|
||||
let sum1_handle = dag1.add_task(ComputeSum::new(1, 251, WORK_DELAY));
|
||||
dag1.run(|fut| { tokio::spawn(fut); }).await?;
|
||||
let r1 = dag1.get(sum1_handle)?;
|
||||
// Run tasks sequentially (not using DAG, just raw execution)
|
||||
let task1 = ComputeSum::new(1, 251, WORK_DELAY);
|
||||
let task2 = ComputeSum::new(251, 501, WORK_DELAY);
|
||||
let task3 = ComputeSum::new(501, 751, WORK_DELAY);
|
||||
let task4 = ComputeSum::new(751, 1001, WORK_DELAY);
|
||||
|
||||
let dag2 = DagRunner::new();
|
||||
let sum2_handle = dag2.add_task(ComputeSum::new(251, 501, WORK_DELAY));
|
||||
dag2.run(|fut| { tokio::spawn(fut); }).await?;
|
||||
let r2 = dag2.get(sum2_handle)?;
|
||||
|
||||
let dag3 = DagRunner::new();
|
||||
let sum3_handle = dag3.add_task(ComputeSum::new(501, 751, WORK_DELAY));
|
||||
dag3.run(|fut| { tokio::spawn(fut); }).await?;
|
||||
let r3 = dag3.get(sum3_handle)?;
|
||||
|
||||
let dag4 = DagRunner::new();
|
||||
let sum4_handle = dag4.add_task(ComputeSum::new(751, 1001, WORK_DELAY));
|
||||
dag4.run(|fut| { tokio::spawn(fut); }).await?;
|
||||
let r4 = dag4.get(sum4_handle)?;
|
||||
let r1 = task1.run(()).await;
|
||||
let r2 = task2.run(()).await;
|
||||
let r3 = task3.run(()).await;
|
||||
let r4 = task4.run(()).await;
|
||||
|
||||
let result = r1 + r2 + r3 + r4;
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
@@ -150,7 +150,7 @@
|
||||
//! Result 7: Ok("Success")
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use std::sync::Arc;
|
||||
use std::time::Instant;
|
||||
use tokio::sync::Mutex;
|
||||
|
||||
@@ -66,7 +66,7 @@
|
||||
//! Combined: 6 items, sum=57, avg=9.50
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagResult, DagRunner};
|
||||
use dagx::{task, DagResult, DagRunner, Task};
|
||||
|
||||
// Extract: Load data from different sources
|
||||
struct LoadData {
|
||||
|
||||
@@ -55,7 +55,7 @@
|
||||
//! Expected: Categories: [low, invalid, high]
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagResult, DagRunner};
|
||||
use dagx::{task, DagResult, DagRunner, Task};
|
||||
|
||||
// Task that validates and processes input
|
||||
struct ValidateAndDouble {
|
||||
|
||||
@@ -18,7 +18,7 @@
|
||||
//! - `Send + Sync` (required for async tasks)
|
||||
//! - `'static` (no lifetime constraints)
|
||||
//!
|
||||
//! That's it! No special traits, no annotations, no boilerplate.
|
||||
//! That's it! No ExtractInput, no special annotations, no trait boilerplate.
|
||||
//!
|
||||
//! ## Running This Example
|
||||
//! ```bash
|
||||
@@ -36,7 +36,7 @@
|
||||
//! Validation: Alice is valid (age 30 >= 18)
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
// Custom type - just derive Clone!
|
||||
#[derive(Clone, Debug)]
|
||||
|
||||
@@ -75,7 +75,7 @@
|
||||
//! ============================
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
// Data source tasks
|
||||
struct LoadValue1;
|
||||
|
||||
@@ -114,7 +114,7 @@
|
||||
//! FulfillOrder: 30ms
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use std::time::Instant;
|
||||
use tokio::time::{sleep, Duration};
|
||||
|
||||
|
||||
@@ -114,7 +114,7 @@
|
||||
//! === Error Handling Example Complete ===
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
// Source task that provides a string input
|
||||
struct StringSource(&'static str);
|
||||
|
||||
@@ -109,7 +109,7 @@
|
||||
//! - Memory overhead is minimal (~ 200 bytes per task)
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use std::time::Instant;
|
||||
|
||||
// Simple computation task
|
||||
|
||||
@@ -47,7 +47,7 @@
|
||||
//! This proves TRUE PARALLEL EXECUTION!
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::time::sleep;
|
||||
|
||||
|
||||
@@ -152,7 +152,7 @@
|
||||
//! Result: Err("Retry exhausted after 3 attempts: Connection timeout")
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use tokio::time::{sleep, Duration};
|
||||
|
||||
// Retry strategy configuration
|
||||
|
||||
+1
-1
@@ -98,7 +98,7 @@
|
||||
//! === Timeout Example Complete ===
|
||||
//! ```
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use tokio::time::{sleep, timeout, Duration};
|
||||
|
||||
// Fast task that completes quickly
|
||||
|
||||
@@ -33,7 +33,7 @@
|
||||
//! - **ERROR**: Errors, panics, cycle detection
|
||||
//!
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use tracing_subscriber::{fmt, EnvFilter};
|
||||
|
||||
// Simple source task
|
||||
|
||||
@@ -1,4 +0,0 @@
|
||||
target
|
||||
corpus
|
||||
artifacts
|
||||
Cargo.lock
|
||||
@@ -1,43 +0,0 @@
|
||||
[package]
|
||||
name = "dagx-fuzz"
|
||||
version = "0.0.0"
|
||||
publish = false
|
||||
edition = "2021"
|
||||
|
||||
[package.metadata]
|
||||
cargo-fuzz = true
|
||||
|
||||
[dependencies]
|
||||
libfuzzer-sys = "0.4"
|
||||
tokio = { version = "1", features = ["rt", "macros"] }
|
||||
|
||||
[dependencies.dagx]
|
||||
path = ".."
|
||||
|
||||
[[bin]]
|
||||
name = "fuzz_dag_execution"
|
||||
path = "fuzz_targets/fuzz_dag_execution.rs"
|
||||
test = false
|
||||
doc = false
|
||||
bench = false
|
||||
|
||||
[[bin]]
|
||||
name = "fuzz_dependencies"
|
||||
path = "fuzz_targets/fuzz_dependencies.rs"
|
||||
test = false
|
||||
doc = false
|
||||
bench = false
|
||||
|
||||
[[bin]]
|
||||
name = "fuzz_type_safety"
|
||||
path = "fuzz_targets/fuzz_type_safety.rs"
|
||||
test = false
|
||||
doc = false
|
||||
bench = false
|
||||
|
||||
[[bin]]
|
||||
name = "fuzz_concurrent"
|
||||
path = "fuzz_targets/fuzz_concurrent.rs"
|
||||
test = false
|
||||
doc = false
|
||||
bench = false
|
||||
-191
@@ -1,191 +0,0 @@
|
||||
# Fuzz Testing for dagx
|
||||
|
||||
This directory contains fuzz testing infrastructure using [cargo-fuzz](https://github.com/rust-fuzz/cargo-fuzz) and libFuzzer.
|
||||
|
||||
## Overview
|
||||
|
||||
We have 4 fuzz targets that test different aspects of the `dagx` library:
|
||||
|
||||
1. **fuzz_dag_execution** - Tests DAG execution with various task configurations
|
||||
2. **fuzz_dependencies** - Tests dependency patterns (linear, diamond, fan-out, complex)
|
||||
3. **fuzz_type_safety** - Tests type-safe output retrieval with various types
|
||||
4. **fuzz_concurrent** - Tests concurrent task execution and thread safety
|
||||
|
||||
## Prerequisites
|
||||
|
||||
Fuzzing requires nightly Rust and cargo-fuzz:
|
||||
|
||||
```bash
|
||||
rustup install nightly
|
||||
cargo install cargo-fuzz
|
||||
```
|
||||
|
||||
## Running Fuzz Tests
|
||||
|
||||
Run a single fuzz target:
|
||||
|
||||
```bash
|
||||
cargo +nightly fuzz run fuzz_dag_execution
|
||||
```
|
||||
|
||||
Run with a time limit (recommended for CI):
|
||||
|
||||
```bash
|
||||
cargo +nightly fuzz run fuzz_dag_execution -- -max_total_time=60
|
||||
```
|
||||
|
||||
Run with a specific number of iterations:
|
||||
|
||||
```bash
|
||||
cargo +nightly fuzz run fuzz_dag_execution -- -runs=10000
|
||||
```
|
||||
|
||||
Run all fuzz targets sequentially:
|
||||
|
||||
```bash
|
||||
for target in fuzz_dag_execution fuzz_dependencies fuzz_type_safety fuzz_concurrent; do
|
||||
echo "Running $target..."
|
||||
cargo +nightly fuzz run $target -- -runs=10000
|
||||
done
|
||||
```
|
||||
|
||||
## Continuous Fuzzing
|
||||
|
||||
For deeper testing, run fuzz targets for extended periods:
|
||||
|
||||
```bash
|
||||
# Run overnight (8 hours)
|
||||
cargo +nightly fuzz run fuzz_concurrent -- -max_total_time=28800
|
||||
|
||||
# Run until crash (or Ctrl+C)
|
||||
cargo +nightly fuzz run fuzz_dag_execution
|
||||
```
|
||||
|
||||
## Understanding Output
|
||||
|
||||
Fuzzer output includes:
|
||||
|
||||
- `cov: X` - Code coverage (number of edges covered)
|
||||
- `ft: X` - Feature coverage (number of unique code paths)
|
||||
- `corp: X/Yb` - Corpus size (X inputs, Y bytes total)
|
||||
- `exec/s: X` - Executions per second
|
||||
|
||||
If a crash is found:
|
||||
|
||||
- Crash input is saved to `artifacts/FUZZ_TARGET/crash-*`
|
||||
- Reproduce with: `cargo +nightly fuzz run FUZZ_TARGET artifacts/FUZZ_TARGET/crash-*`
|
||||
|
||||
## Fuzz Target Details
|
||||
|
||||
### fuzz_dag_execution
|
||||
|
||||
Tests basic DAG execution:
|
||||
- Simple linear task chains
|
||||
- Fan-out patterns (one task → multiple dependents)
|
||||
- Various data types (i32, Vec<u8>, custom structs)
|
||||
- DAG execution correctness
|
||||
- Output value verification
|
||||
|
||||
### fuzz_dependencies
|
||||
|
||||
Tests dependency management:
|
||||
- Linear dependency chains
|
||||
- Diamond patterns (source → two paths → merge)
|
||||
- Multiple independent sources
|
||||
- Complex dependency graphs
|
||||
- Dependency tracking correctness
|
||||
|
||||
### fuzz_type_safety
|
||||
|
||||
Tests type system correctness:
|
||||
- Various output types (i32, String, Vec, custom structs)
|
||||
- Correct type retrieval with dag.get()
|
||||
- Option<T> and Result<T, E> handling
|
||||
- Custom struct preservation
|
||||
- Type mismatch error handling
|
||||
|
||||
### fuzz_concurrent
|
||||
|
||||
Tests thread safety and concurrency:
|
||||
- Multiple independent concurrent tasks
|
||||
- Concurrent dependent tasks (fan-out)
|
||||
- Arc<DagRunner> sharing across threads
|
||||
- Task timing variations
|
||||
- Result access from multiple threads
|
||||
|
||||
## Adding New Fuzz Targets
|
||||
|
||||
1. Create a new file in `fuzz_targets/` (e.g., `fuzz_new_feature.rs`)
|
||||
2. Use the template:
|
||||
|
||||
```rust
|
||||
#![no_main]
|
||||
use libfuzzer_sys::fuzz_target;
|
||||
use dagx::{task, DagRunner};
|
||||
|
||||
fuzz_target!(|data: &[u8]| {
|
||||
// Your fuzz test here
|
||||
});
|
||||
```
|
||||
|
||||
3. Add to `fuzz/Cargo.toml`:
|
||||
|
||||
```toml
|
||||
[[bin]]
|
||||
name = "fuzz_new_feature"
|
||||
path = "fuzz_targets/fuzz_new_feature.rs"
|
||||
test = false
|
||||
doc = false
|
||||
bench = false
|
||||
```
|
||||
|
||||
4. Build and run:
|
||||
|
||||
```bash
|
||||
cargo +nightly fuzz build
|
||||
cargo +nightly fuzz run fuzz_new_feature
|
||||
```
|
||||
|
||||
## CI Integration
|
||||
|
||||
Fuzzing can run automatically in CI on every push/PR:
|
||||
- Each target runs for 60-120 seconds
|
||||
- Targets run 1,000-10,000 iterations minimum
|
||||
- Any crashes fail the build
|
||||
|
||||
See your CI configuration for details.
|
||||
|
||||
## Corpus Management
|
||||
|
||||
The fuzzer builds a corpus of interesting inputs in `corpus/FUZZ_TARGET/`. These inputs:
|
||||
- Are automatically discovered during fuzzing
|
||||
- Can be committed to Git for regression testing
|
||||
- Help future fuzzing sessions start with good coverage
|
||||
|
||||
To minimize the corpus:
|
||||
|
||||
```bash
|
||||
cargo +nightly fuzz cmin fuzz_dag_execution
|
||||
```
|
||||
|
||||
## Tips
|
||||
|
||||
1. **Start with short runs** - Use `-max_total_time=60` to catch obvious bugs quickly
|
||||
2. **Run overnight** - For thorough testing, run for hours or days
|
||||
3. **Check coverage** - Monitor `cov:` values to ensure good code coverage
|
||||
4. **Save crashes** - Commit crash artifacts as regression tests
|
||||
5. **Update regularly** - Re-run fuzzing when adding new features
|
||||
6. **Limit data size** - Fuzz targets filter out oversized inputs to avoid timeouts
|
||||
|
||||
## Performance Notes
|
||||
|
||||
- DAG execution involves async runtime overhead
|
||||
- Each fuzz iteration creates a new Tokio runtime
|
||||
- Typical execution rate: 100-1000 runs/second
|
||||
- Adjust `-runs` or `-max_total_time` based on your needs
|
||||
|
||||
## References
|
||||
|
||||
- [cargo-fuzz book](https://rust-fuzz.github.io/book/cargo-fuzz.html)
|
||||
- [libFuzzer documentation](https://llvm.org/docs/LibFuzzer.html)
|
||||
- [Fuzzing in Rust](https://rust-fuzz.github.io/book/)
|
||||
@@ -1,137 +0,0 @@
|
||||
#![no_main]
|
||||
|
||||
use libfuzzer_sys::fuzz_target;
|
||||
use dagx::{task, DagRunner};
|
||||
use std::sync::Arc;
|
||||
|
||||
fuzz_target!(|data: &[u8]| {
|
||||
if data.is_empty() || data.len() > 50 {
|
||||
return;
|
||||
}
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
|
||||
// Test concurrent task execution
|
||||
struct SlowTask { val: i32, delay_ms: u64 }
|
||||
#[task]
|
||||
impl SlowTask {
|
||||
async fn run(&self) -> i32 {
|
||||
tokio::time::sleep(tokio::time::Duration::from_millis(self.delay_ms)).await;
|
||||
self.val
|
||||
}
|
||||
}
|
||||
|
||||
struct FastTask { val: i32 }
|
||||
#[task]
|
||||
impl FastTask {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
// Test concurrent independent tasks
|
||||
{
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Create multiple independent tasks
|
||||
let num_tasks = (data[0] % 5 + 1) as usize;
|
||||
let mut tasks = Vec::new();
|
||||
|
||||
for i in 0..num_tasks {
|
||||
let val = data.get(i).copied().unwrap_or(0) as i32;
|
||||
let delay = (val.abs() as u64 % 10) + 1;
|
||||
|
||||
let task = dag.add_task(SlowTask { val, delay_ms: delay });
|
||||
tasks.push((task, val));
|
||||
}
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
// Verify all tasks completed with correct values
|
||||
for (handle, expected) in tasks {
|
||||
if let Ok(result) = dag.get(handle) {
|
||||
assert_eq!(result, expected);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test concurrent dependent tasks
|
||||
{
|
||||
let dag = DagRunner::new();
|
||||
|
||||
let source = dag.add_task(FastTask { val: data[0] as i32 });
|
||||
|
||||
// Create multiple tasks that depend on source
|
||||
struct AddOne;
|
||||
#[task]
|
||||
impl AddOne {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(1)
|
||||
}
|
||||
}
|
||||
|
||||
struct AddTwo;
|
||||
#[task]
|
||||
impl AddTwo {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(2)
|
||||
}
|
||||
}
|
||||
|
||||
struct AddThree;
|
||||
#[task]
|
||||
impl AddThree {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(3)
|
||||
}
|
||||
}
|
||||
|
||||
let t1 = dag.add_task(AddOne).depends_on(&source);
|
||||
let t2 = dag.add_task(AddTwo).depends_on(&source);
|
||||
let t3 = dag.add_task(AddThree).depends_on(&source);
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
let base = data[0] as i32;
|
||||
if let Ok(r1) = dag.get(t1) {
|
||||
assert_eq!(r1, base.wrapping_add(1));
|
||||
}
|
||||
if let Ok(r2) = dag.get(t2) {
|
||||
assert_eq!(r2, base.wrapping_add(2));
|
||||
}
|
||||
if let Ok(r3) = dag.get(t3) {
|
||||
assert_eq!(r3, base.wrapping_add(3));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test Arc sharing across concurrent tasks
|
||||
{
|
||||
let dag = Arc::new(DagRunner::new());
|
||||
|
||||
struct SharedTask { val: i32 }
|
||||
#[task]
|
||||
impl SharedTask {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
let task = dag.add_task(SharedTask { val: data[0] as i32 });
|
||||
let task_copy = task;
|
||||
|
||||
let dag1 = Arc::clone(&dag);
|
||||
let dag2 = Arc::clone(&dag);
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
// Multiple threads can access result
|
||||
if let Ok(r1) = dag1.get(task_copy) {
|
||||
if let Ok(r2) = dag2.get(task_copy) {
|
||||
assert_eq!(r1, r2);
|
||||
assert_eq!(r1, data[0] as i32);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1,120 +0,0 @@
|
||||
#![no_main]
|
||||
|
||||
use libfuzzer_sys::fuzz_target;
|
||||
use dagx::{task, DagRunner};
|
||||
|
||||
fuzz_target!(|data: &[u8]| {
|
||||
// Only fuzz with reasonable data sizes to avoid timeouts
|
||||
if data.is_empty() || data.len() > 1000 {
|
||||
return;
|
||||
}
|
||||
|
||||
// Convert bytes to values for tasks
|
||||
let value1 = data.get(0).copied().unwrap_or(0) as i32;
|
||||
let value2 = data.get(1).copied().unwrap_or(0) as i32;
|
||||
let value3 = data.get(2).copied().unwrap_or(0) as i32;
|
||||
|
||||
// Test simple linear DAG execution
|
||||
{
|
||||
struct Source1 { val: i32 }
|
||||
#[task]
|
||||
impl Source1 {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct Source2 { val: i32 }
|
||||
#[task]
|
||||
impl Source2 {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct Compute;
|
||||
#[task]
|
||||
impl Compute {
|
||||
async fn run(a: &i32, b: &i32) -> i32 {
|
||||
a.wrapping_add(*b)
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let s1 = dag.add_task(Source1 { val: value1 });
|
||||
let s2 = dag.add_task(Source2 { val: value2 });
|
||||
let compute = dag.add_task(Compute).depends_on((&s1, &s2));
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
if let Ok(()) = rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}) {
|
||||
// Verify output
|
||||
if let Ok(result) = dag.get(compute) {
|
||||
assert_eq!(result, value1.wrapping_add(value2));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test fan-out pattern
|
||||
{
|
||||
struct Source { val: i32 }
|
||||
#[task]
|
||||
impl Source {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct Double;
|
||||
#[task]
|
||||
impl Double {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_mul(2)
|
||||
}
|
||||
}
|
||||
|
||||
struct Triple;
|
||||
#[task]
|
||||
impl Triple {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_mul(3)
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(Source { val: value3 });
|
||||
let double = dag.add_task(Double).depends_on(&source);
|
||||
let triple = dag.add_task(Triple).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
if let Ok(()) = rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}) {
|
||||
// Verify outputs
|
||||
if let (Ok(d), Ok(t)) = (dag.get(double), dag.get(triple)) {
|
||||
assert_eq!(d, value3.wrapping_mul(2));
|
||||
assert_eq!(t, value3.wrapping_mul(3));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test error-free execution with various data
|
||||
if data.len() >= 4 {
|
||||
struct DataTask { data: Vec<u8> }
|
||||
#[task]
|
||||
impl DataTask {
|
||||
async fn run(&self) -> Vec<u8> {
|
||||
self.data.clone()
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let task = dag.add_task(DataTask {
|
||||
data: data[..4.min(data.len())].to_vec()
|
||||
});
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
if let Ok(()) = rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}) {
|
||||
if let Ok(result) = dag.get(task) {
|
||||
assert_eq!(result.len(), 4.min(data.len()));
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1,115 +0,0 @@
|
||||
#![no_main]
|
||||
|
||||
use libfuzzer_sys::fuzz_target;
|
||||
use dagx::{task, DagRunner};
|
||||
|
||||
fuzz_target!(|data: &[u8]| {
|
||||
// Only test with reasonable sizes
|
||||
if data.is_empty() || data.len() > 100 {
|
||||
return;
|
||||
}
|
||||
|
||||
// Test various dependency patterns based on input data
|
||||
let pattern = data[0] % 4;
|
||||
|
||||
struct SimpleTask { val: i32 }
|
||||
#[task]
|
||||
impl SimpleTask {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct DependentTask;
|
||||
#[task]
|
||||
impl DependentTask {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(1)
|
||||
}
|
||||
}
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
|
||||
match pattern {
|
||||
0 => {
|
||||
// Linear dependency chain
|
||||
let dag = DagRunner::new();
|
||||
let t1 = dag.add_task(SimpleTask { val: data[0] as i32 });
|
||||
let t2 = dag.add_task(DependentTask).depends_on(&t1);
|
||||
let t3 = dag.add_task(DependentTask).depends_on(&t2);
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
if let Ok(result) = dag.get(t3) {
|
||||
assert_eq!(result, (data[0] as i32).wrapping_add(2));
|
||||
}
|
||||
}
|
||||
}
|
||||
1 => {
|
||||
// Diamond pattern
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(SimpleTask { val: data[0] as i32 });
|
||||
let left = dag.add_task(DependentTask).depends_on(&source);
|
||||
let right = dag.add_task(DependentTask).depends_on(&source);
|
||||
|
||||
struct MergeTask;
|
||||
#[task]
|
||||
impl MergeTask {
|
||||
async fn run(a: &i32, b: &i32) -> i32 {
|
||||
a.wrapping_add(*b)
|
||||
}
|
||||
}
|
||||
|
||||
let merge = dag.add_task(MergeTask).depends_on((&left, &right));
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
if let Ok(result) = dag.get(merge) {
|
||||
let expected = (data[0] as i32).wrapping_add(1);
|
||||
assert_eq!(result, expected.wrapping_add(expected));
|
||||
}
|
||||
}
|
||||
}
|
||||
2 => {
|
||||
// Multiple independent sources
|
||||
let dag = DagRunner::new();
|
||||
let t1 = dag.add_task(SimpleTask { val: data[0] as i32 });
|
||||
let t2 = dag.add_task(SimpleTask {
|
||||
val: data.get(1).copied().unwrap_or(0) as i32
|
||||
});
|
||||
let t3 = dag.add_task(SimpleTask {
|
||||
val: data.get(2).copied().unwrap_or(0) as i32
|
||||
});
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
if let (Ok(r1), Ok(r2), Ok(r3)) = (dag.get(t1), dag.get(t2), dag.get(t3)) {
|
||||
assert_eq!(r1, data[0] as i32);
|
||||
assert_eq!(r2, data.get(1).copied().unwrap_or(0) as i32);
|
||||
assert_eq!(r3, data.get(2).copied().unwrap_or(0) as i32);
|
||||
}
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// Complex dependency graph
|
||||
let dag = DagRunner::new();
|
||||
let s1 = dag.add_task(SimpleTask { val: data[0] as i32 });
|
||||
let s2 = dag.add_task(SimpleTask {
|
||||
val: data.get(1).copied().unwrap_or(0) as i32
|
||||
});
|
||||
|
||||
let d1 = dag.add_task(DependentTask).depends_on(&s1);
|
||||
let d2 = dag.add_task(DependentTask).depends_on(&s2);
|
||||
let d3 = dag.add_task(DependentTask).depends_on(&d1);
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
let _ = dag.get(d1);
|
||||
let _ = dag.get(d2);
|
||||
let _ = dag.get(d3);
|
||||
}
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1,138 +0,0 @@
|
||||
#![no_main]
|
||||
|
||||
use libfuzzer_sys::fuzz_target;
|
||||
use dagx::{task, DagRunner};
|
||||
|
||||
fuzz_target!(|data: &[u8]| {
|
||||
if data.is_empty() || data.len() > 100 {
|
||||
return;
|
||||
}
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
|
||||
// Test type safety with various output types
|
||||
struct IntTask { val: i32 }
|
||||
#[task]
|
||||
impl IntTask {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct StringTask { val: String }
|
||||
#[task]
|
||||
impl StringTask {
|
||||
async fn run(&self) -> String { self.val.clone() }
|
||||
}
|
||||
|
||||
struct VecTask { data: Vec<u8> }
|
||||
#[task]
|
||||
impl VecTask {
|
||||
async fn run(&self) -> Vec<u8> { self.data.clone() }
|
||||
}
|
||||
|
||||
// Test correct type retrieval
|
||||
{
|
||||
let dag = DagRunner::new();
|
||||
let int_task = dag.add_task(IntTask { val: data[0] as i32 });
|
||||
|
||||
if let Ok(string_data) = std::str::from_utf8(data) {
|
||||
let string_task = dag.add_task(StringTask {
|
||||
val: string_data.to_string()
|
||||
});
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
// Correct type should work
|
||||
let _ = dag.get(int_task);
|
||||
let _ = dag.get(string_task);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test Vec type
|
||||
{
|
||||
let dag = DagRunner::new();
|
||||
let vec_task = dag.add_task(VecTask {
|
||||
data: data.to_vec()
|
||||
});
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
if let Ok(result) = dag.get(vec_task) {
|
||||
assert_eq!(result, data);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test custom struct types
|
||||
{
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct CustomData {
|
||||
value: i32,
|
||||
flag: bool,
|
||||
}
|
||||
|
||||
struct CustomTask { data: CustomData }
|
||||
#[task]
|
||||
impl CustomTask {
|
||||
async fn run(&self) -> CustomData {
|
||||
self.data.clone()
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let custom = CustomData {
|
||||
value: data[0] as i32,
|
||||
flag: data.get(1).map(|&b| b % 2 == 0).unwrap_or(false),
|
||||
};
|
||||
let task = dag.add_task(CustomTask { data: custom.clone() });
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
if let Ok(result) = dag.get(task) {
|
||||
assert_eq!(result, custom);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Test Option and Result types
|
||||
{
|
||||
struct OptionTask { val: Option<i32> }
|
||||
#[task]
|
||||
impl OptionTask {
|
||||
async fn run(&self) -> Option<i32> { self.val }
|
||||
}
|
||||
|
||||
struct ResultTask { val: Result<i32, String> }
|
||||
#[task]
|
||||
impl ResultTask {
|
||||
async fn run(&self) -> Result<i32, String> { self.val.clone() }
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let opt_task = dag.add_task(OptionTask {
|
||||
val: if data[0] % 2 == 0 {
|
||||
Some(data[0] as i32)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
|
||||
let result_task = dag.add_task(ResultTask {
|
||||
val: if data.get(1).map(|&b| b % 2 == 0).unwrap_or(false) {
|
||||
Ok(data.get(1).copied().unwrap_or(0) as i32)
|
||||
} else {
|
||||
Err("error".to_string())
|
||||
}
|
||||
});
|
||||
|
||||
if rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).is_ok() {
|
||||
let _ = dag.get(opt_task);
|
||||
let _ = dag.get(result_task);
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -1,257 +0,0 @@
|
||||
# justfile for dagx crate
|
||||
# High-level development and CI workflows
|
||||
|
||||
# =============================================================================
|
||||
# Configuration
|
||||
# =============================================================================
|
||||
|
||||
# Treat all warnings as errors
|
||||
rustflags := "-D warnings"
|
||||
|
||||
# Code coverage target percentage (fail if below this threshold)
|
||||
coverage_target := "80"
|
||||
|
||||
# Timeout for coverage analysis in seconds
|
||||
coverage_timeout := "600"
|
||||
|
||||
# Benchmark timeout in seconds
|
||||
bench_timeout := "300"
|
||||
|
||||
# =============================================================================
|
||||
|
||||
# Default recipe - show available recipes
|
||||
default:
|
||||
@just --list
|
||||
|
||||
# =============================================================================
|
||||
# Primary Development Workflows
|
||||
# =============================================================================
|
||||
|
||||
# Quick development loop: format + lint + fast tests
|
||||
dev:
|
||||
#!/usr/bin/env bash
|
||||
set -e
|
||||
echo "🔄 Running development checks..."
|
||||
cargo fmt
|
||||
RUSTFLAGS="{{rustflags}}" cargo clippy --all-targets --all-features &
|
||||
RUSTFLAGS="{{rustflags}}" cargo nextest run --lib --all-features &
|
||||
wait
|
||||
echo "✅ Development checks passed"
|
||||
|
||||
# Pre-commit checks: format-check + lint + typecheck
|
||||
check:
|
||||
@echo "Running format check..."
|
||||
cargo fmt -- --check
|
||||
@echo ""
|
||||
@echo "Running clippy..."
|
||||
RUSTFLAGS="{{rustflags}}" cargo clippy --all-targets --all-features
|
||||
RUSTFLAGS="{{rustflags}}" cargo clippy --all-targets --no-default-features
|
||||
@echo ""
|
||||
@echo "Verifying compilation..."
|
||||
RUSTFLAGS="{{rustflags}}" cargo test --lib --all-features --no-run
|
||||
RUSTFLAGS="{{rustflags}}" cargo test --lib --no-default-features --no-run
|
||||
@echo ""
|
||||
@echo "✅ All checks passed"
|
||||
|
||||
# Run all tests
|
||||
test:
|
||||
@echo "Running all tests..."
|
||||
RUSTFLAGS="{{rustflags}}" cargo nextest run --lib --all-features
|
||||
@echo "✅ All tests passed"
|
||||
|
||||
# Before pushing: check + test + coverage
|
||||
pre-push: check test coverage
|
||||
@echo ""
|
||||
@echo "✅ Pre-push checks passed - safe to push!"
|
||||
|
||||
# Comprehensive pre-release validation
|
||||
pre-release: check test coverage audit examples bench-check
|
||||
#!/usr/bin/env bash
|
||||
set -e
|
||||
|
||||
# Verify documentation version matches crate version
|
||||
echo ""
|
||||
echo "📋 Verifying documentation version..."
|
||||
VERSION=$(grep '^version = ' Cargo.toml | head -1 | sed 's/.*"\(.*\)".*/\1/')
|
||||
MAJOR_MINOR=$(echo $VERSION | cut -d. -f1-2)
|
||||
|
||||
if ! grep -q "dagx = \"$MAJOR_MINOR\"" README.md; then
|
||||
echo "❌ ERROR: README.md should reference version $MAJOR_MINOR (crate is $VERSION)"
|
||||
echo " Update installation examples in README.md"
|
||||
exit 1
|
||||
fi
|
||||
echo "✅ Documentation version correct: $MAJOR_MINOR"
|
||||
|
||||
# Package verification
|
||||
echo ""
|
||||
echo "📦 Verifying package..."
|
||||
cargo publish --dry-run
|
||||
cargo package --list > /tmp/dagx-files.txt
|
||||
echo "✅ Package verified"
|
||||
|
||||
echo ""
|
||||
echo "✅ All pre-release checks passed!"
|
||||
echo " - Code formatted and linted"
|
||||
echo " - All tests passing"
|
||||
echo " - Code coverage ≥ {{coverage_target}}%"
|
||||
echo " - No security vulnerabilities"
|
||||
echo " - Dependencies verified"
|
||||
echo " - All examples run successfully"
|
||||
echo " - All benchmarks compile"
|
||||
echo " - Documentation version verified"
|
||||
echo " - Package ready for publication"
|
||||
echo ""
|
||||
echo "📦 Ready to publish! Run: cargo publish"
|
||||
|
||||
# =============================================================================
|
||||
# Quality & Security
|
||||
# =============================================================================
|
||||
|
||||
# Run code coverage analysis
|
||||
coverage:
|
||||
@echo "Running code coverage analysis (target: {{coverage_target}}%)..."
|
||||
cargo llvm-cov nextest --lib --html --output-dir coverage --fail-under-lines {{coverage_target}}
|
||||
@echo "Coverage report: coverage/html/index.html"
|
||||
|
||||
# Mutation testing with cargo-mutants
|
||||
mutants:
|
||||
@echo "Running mutation testing..."
|
||||
@echo "⚠️ This will use significant CPU and memory resources!"
|
||||
cargo mutants \
|
||||
--no-shuffle \
|
||||
--test-tool nextest \
|
||||
--jobs 3 \
|
||||
--all-features \
|
||||
-vV
|
||||
|
||||
# Security audit + dependency checks
|
||||
audit:
|
||||
@echo "🔒 Running security checks..."
|
||||
cargo audit
|
||||
@echo "✅ Security checks passed"
|
||||
|
||||
# Run Miri for undefined behavior detection
|
||||
miri:
|
||||
@echo "Running Miri tests for UB detection..."
|
||||
cargo +nightly miri test --lib
|
||||
|
||||
# =============================================================================
|
||||
# Benchmarks
|
||||
# =============================================================================
|
||||
|
||||
# Verify benchmarks compile
|
||||
bench-check:
|
||||
@echo "Verifying benchmarks compile..."
|
||||
cargo bench --no-run
|
||||
@echo "✅ All benchmarks compile"
|
||||
|
||||
# Run all benchmarks
|
||||
bench:
|
||||
cargo bench
|
||||
|
||||
# Run benchmarks and save as baseline
|
||||
bench-baseline:
|
||||
cargo bench -- --save-baseline baseline
|
||||
|
||||
# Compare benchmarks against baseline
|
||||
bench-compare:
|
||||
cargo bench -- --baseline baseline
|
||||
|
||||
# Run specific benchmark suite
|
||||
bench-suite suite:
|
||||
cargo bench --bench {{suite}}
|
||||
|
||||
# =============================================================================
|
||||
# Fuzzing
|
||||
# =============================================================================
|
||||
|
||||
# Run a single fuzz target for 60 seconds
|
||||
fuzz target:
|
||||
@echo "Fuzzing {{target}} for 60 seconds..."
|
||||
cargo +nightly fuzz run {{target}} -- -max_total_time=60
|
||||
|
||||
# Run all fuzz targets sequentially for quick smoke test
|
||||
fuzz-all:
|
||||
#!/usr/bin/env bash
|
||||
set -e
|
||||
echo "🔬 Running all fuzz targets..."
|
||||
for target in fuzz_dag_execution fuzz_dependencies fuzz_type_safety fuzz_concurrent; do
|
||||
echo ""
|
||||
echo "Running $target..."
|
||||
cargo +nightly fuzz run $target -- -max_total_time=30 -runs=10000
|
||||
done
|
||||
echo ""
|
||||
echo "✅ All fuzz targets completed"
|
||||
|
||||
# Run extended fuzzing session (for deeper testing)
|
||||
fuzz-extended target:
|
||||
@echo "Running extended fuzzing session for {{target}}..."
|
||||
@echo "⚠️ This will run for 1 hour. Press Ctrl+C to stop early."
|
||||
cargo +nightly fuzz run {{target}} -- -max_total_time=3600
|
||||
|
||||
# Build all fuzz targets
|
||||
fuzz-build:
|
||||
@echo "Building fuzz targets..."
|
||||
cargo +nightly fuzz build
|
||||
@echo "✅ All fuzz targets built"
|
||||
|
||||
# Minimize corpus for a fuzz target
|
||||
fuzz-cmin target:
|
||||
@echo "Minimizing corpus for {{target}}..."
|
||||
cargo +nightly fuzz cmin {{target}}
|
||||
|
||||
# Run property-based tests
|
||||
proptest:
|
||||
@echo "Running property-based tests..."
|
||||
cargo nextest run property_tests --all-features
|
||||
@echo "✅ Property tests passed"
|
||||
|
||||
# =============================================================================
|
||||
# Documentation & Utilities
|
||||
# =============================================================================
|
||||
|
||||
# Build documentation and open in browser
|
||||
doc:
|
||||
cargo doc --no-deps --all-features --open
|
||||
|
||||
# Format code
|
||||
fmt:
|
||||
cargo fmt
|
||||
|
||||
# Show project information
|
||||
info:
|
||||
@echo "dagx crate - Type-safe DAG execution framework for async Rust"
|
||||
@echo ""
|
||||
@echo "Rust: $(rustc --version)"
|
||||
@echo "Cargo: $(cargo --version)"
|
||||
@echo ""
|
||||
@echo "Dependencies:"
|
||||
@cargo tree --depth 1
|
||||
|
||||
# Run all examples
|
||||
examples:
|
||||
@echo "Running all examples..."
|
||||
cargo run --example 01_basic
|
||||
cargo run --example 02_fan_out
|
||||
cargo run --example 03_fan_in
|
||||
cargo run --example 04_parallel_computation
|
||||
cargo run --example circuit_breaker
|
||||
cargo run --example complex_dag
|
||||
cargo run --example conditional_workflow
|
||||
cargo run --example custom_types
|
||||
cargo run --example data_pipeline
|
||||
cargo run --example debug_tracing
|
||||
cargo run --example error_handling
|
||||
cargo run --example large_dag
|
||||
cargo run --example parallelism_proof
|
||||
cargo run --example retry_strategies
|
||||
cargo run --example timeout
|
||||
cargo run --example tracing_example
|
||||
@echo "✅ All examples completed successfully"
|
||||
|
||||
# Clean build artifacts
|
||||
clean:
|
||||
cargo clean
|
||||
rm -rf mutants.out/ mutants.out.old/ mutation_history.json
|
||||
rm -rf target/criterion
|
||||
rm -rf coverage/
|
||||
Whitespace-only changes.
@@ -1,5 +0,0 @@
|
||||
0.038627406s DEBUG cargo_mutants::copy_tree: /home/swaits/.local/share/cargo/registry/src/index.crates.io-1949cf8c6b5b557f/cargo-mutants-25.3.1/src/copy_tree.rs:69: walk_builder=WalkBuilder { paths: ["/home/swaits/Code/dagx"], ig_builder: IgnoreBuilder { dir: "", overrides: Override(Gitignore { set: GlobSet { len: 0, strats: [] }, root: "", globs: [], num_ignores: 0, num_whitelists: 0, matches: None }), types: Types { defs: [], selections: [], has_selected: false, glob_to_selection: [], set: GlobSet { len: 0, strats: [] }, matches: Pool(Pool { stacks: [CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. }), CacheLine(Mutex { data: [], poisoned: false, .. })], owner: 0, owner_val: UnsafeCell { .. } }) }, explicit_ignores: [], custom_ignore_filenames: [], opts: IgnoreOptions { hidden: false, ignore: false, parents: true, git_global: false, git_ignore: false, git_exclude: false, ignore_case_insensitive: false, require_git: true } }, max_depth: None, max_filesize: None, follow_links: false, threads: 0, skip: None }
|
||||
0.041495573s DEBUG cargo_mutants::copy_tree: /home/swaits/.local/share/cargo/registry/src/index.crates.io-1949cf8c6b5b557f/cargo-mutants-25.3.1/src/copy_tree.rs:115: Copied source tree total_bytes=967004 total_files=153 temp_dir="/tmp/cargo-mutants-dagx-o2Mc3y.tmp"
|
||||
0.041758047s DEBUG cargo_mutants::lab: /home/swaits/.local/share/cargo/registry/src/index.crates.io-1949cf8c6b5b557f/cargo-mutants-25.3.1/src/lab.rs:55: starting jobserver n_tasks=16
|
||||
0.041839390s DEBUG cargo_mutants::lab: /home/swaits/.local/share/cargo/registry/src/index.crates.io-1949cf8c6b5b557f/cargo-mutants-25.3.1/src/lab.rs:267: test_packages=Explicit([Package { name: "dagx", version: "0.3.0", relative_dir: "", top_sources: ["src/lib.rs"] }])
|
||||
0.041870839s DEBUG run{phase=Build}: cargo_mutants::process: /home/swaits/.local/share/cargo/registry/src/index.crates.io-1949cf8c6b5b557f/cargo-mutants-25.3.1/src/process.rs:81: start process quoted_argv=/home/swaits/.local/share/rustup/toolchains/1.90.0-x86_64-unknown-linux-gnu/bin/cargo nextest run --no-run --verbose --package=dagx@0.3.0 --all-features
|
||||
@@ -1,6 +0,0 @@
|
||||
{
|
||||
"cargo_mutants_version": "25.3.1",
|
||||
"start_time": "2025-10-14T16:44:19.225601093Z",
|
||||
"hostname": "swaits-fw13",
|
||||
"username": "swaits"
|
||||
}
|
||||
File diff suppressed because it is too large.
Load diff
Whitespace-only changes.
File diff suppressed because it is too large.
Load diff
Whitespace-only changes.
Whitespace-only changes.
@@ -7,7 +7,7 @@ set -e
|
||||
# =============================================================================
|
||||
# CONFIGURATION
|
||||
# =============================================================================
|
||||
VERSION="0.3.0"
|
||||
VERSION="0.3.2"
|
||||
|
||||
echo "=================="
|
||||
echo "dagx Release Check"
|
||||
|
||||
+3
-3
@@ -4,8 +4,8 @@
|
||||
//! The type parameter tracks whether dependencies have been specified yet.
|
||||
|
||||
use crate::deps::DepsTuple;
|
||||
use crate::node::TaskExecute;
|
||||
use crate::runner::DagRunner;
|
||||
use crate::task::Task;
|
||||
use crate::types::{NodeId, TaskHandle};
|
||||
|
||||
#[cfg(feature = "tracing")]
|
||||
@@ -69,13 +69,13 @@ impl IsUnitType for () {}
|
||||
/// assert_eq!(dag.get(&b).unwrap(), 20);
|
||||
/// # };
|
||||
/// ```
|
||||
pub struct TaskBuilder<'a, Tk: TaskExecute, Deps> {
|
||||
pub struct TaskBuilder<'a, Tk: Task, Deps> {
|
||||
pub(crate) id: NodeId,
|
||||
pub(crate) dag: &'a DagRunner,
|
||||
pub(crate) _phantom: std::marker::PhantomData<(Tk, Deps)>,
|
||||
}
|
||||
|
||||
impl<'a, Tk: TaskExecute, Deps> TaskBuilder<'a, Tk, Deps> {
|
||||
impl<'a, Tk: Task, Deps> TaskBuilder<'a, Tk, Deps> {
|
||||
/// Provide all dependencies exactly once as a tuple.
|
||||
///
|
||||
/// The dependencies must match the task's `Input` type exactly:
|
||||
|
||||
@@ -1,10 +1,8 @@
|
||||
//! Unit tests for builder module
|
||||
|
||||
// Allow tests to reference dagx:: paths (for proc macro generated code)
|
||||
extern crate self as dagx;
|
||||
|
||||
use crate::builder::IsUnitType;
|
||||
use crate::runner::DagRunner;
|
||||
use crate::task::Task;
|
||||
use crate::types::TaskHandle;
|
||||
|
||||
// Initialize tracing subscriber for tests (idempotent)
|
||||
|
||||
+4
-4
@@ -8,7 +8,7 @@
|
||||
//! use the provided implementations.
|
||||
|
||||
use crate::builder::TaskBuilder;
|
||||
use crate::node::TaskExecute;
|
||||
use crate::task::Task;
|
||||
use crate::types::{NodeId, TaskHandle};
|
||||
|
||||
pub(crate) trait DepsTuple<Input> {
|
||||
@@ -48,13 +48,13 @@ impl<T> DepsTuple<T> for (TaskHandle<T>,) {
|
||||
}
|
||||
|
||||
// Support for &TaskBuilder
|
||||
impl<'a, Tk: TaskExecute, Deps> DepsTuple<Tk::Output> for &TaskBuilder<'a, Tk, Deps> {
|
||||
impl<'a, Tk: Task, Deps> DepsTuple<Tk::Output> for &TaskBuilder<'a, Tk, Deps> {
|
||||
fn to_node_ids(&self) -> Vec<NodeId> {
|
||||
vec![self.id]
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, Tk: TaskExecute, Deps> DepsTuple<Tk::Output> for (&TaskBuilder<'a, Tk, Deps>,) {
|
||||
impl<'a, Tk: Task, Deps> DepsTuple<Tk::Output> for (&TaskBuilder<'a, Tk, Deps>,) {
|
||||
fn to_node_ids(&self) -> Vec<NodeId> {
|
||||
vec![self.0.id]
|
||||
}
|
||||
@@ -89,7 +89,7 @@ macro_rules! impl_deps_tuple {
|
||||
}
|
||||
|
||||
// For &TaskBuilder tuples
|
||||
impl<'a, $($Tk: TaskExecute, $D),+> DepsTuple<($($Tk::Output,)+)> for ($(&TaskBuilder<'a, $Tk, $D>,)+) {
|
||||
impl<'a, $($Tk: Task, $D),+> DepsTuple<($($Tk::Output,)+)> for ($(&TaskBuilder<'a, $Tk, $D>,)+) {
|
||||
#[allow(non_snake_case)]
|
||||
fn to_node_ids(&self) -> Vec<NodeId> {
|
||||
let ($($T,)+) = self;
|
||||
|
||||
+2
-4
@@ -1,10 +1,8 @@
|
||||
//! Unit tests for deps module
|
||||
|
||||
// Allow tests to reference dagx:: paths (for proc macro generated code)
|
||||
extern crate self as dagx;
|
||||
|
||||
use crate::deps::DepsTuple;
|
||||
use crate::runner::DagRunner;
|
||||
use crate::task::Task;
|
||||
use crate::types::{NodeId, TaskHandle};
|
||||
|
||||
// Test task for dependency testing
|
||||
@@ -20,11 +18,11 @@ impl TestTask {
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
struct TestTaskWithInput;
|
||||
|
||||
#[crate::task]
|
||||
impl TestTaskWithInput {
|
||||
#[allow(dead_code)]
|
||||
#[cfg(not(tarpaulin_include))]
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input * 2
|
||||
|
||||
+312
@@ -0,0 +1,312 @@
|
||||
//! Input extraction for task dependencies via oneshot channels.
|
||||
//!
|
||||
//! # 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 `extract_and_run()` methods with
|
||||
//! type-specific extraction logic. This means **ANY type** implementing `Clone + 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 receivers.
|
||||
//! 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 `extract_and_run()`
|
||||
//! - Works with ANY type (just needs `Clone + Send + Sync`)
|
||||
//! - 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.
|
||||
|
||||
use std::any::Any;
|
||||
use std::sync::Arc;
|
||||
|
||||
use futures::channel::oneshot;
|
||||
|
||||
/// Helper trait for async input extraction from oneshot channels.
|
||||
///
|
||||
/// 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.
|
||||
///
|
||||
/// NOTE: This trait is now primarily used by the internal `task_fn` test helper.
|
||||
/// The `#[task]` macro generates inline extraction logic via `extract_and_run()`,
|
||||
/// which allows ANY type to work without requiring ExtractInput implementations.
|
||||
#[allow(dead_code)]
|
||||
pub(crate) trait ExtractInput: Sized + Clone {
|
||||
fn extract_from_channels(
|
||||
receivers: Vec<Box<dyn Any + Send>>,
|
||||
) -> impl std::future::Future<Output = Result<Self, String>> + Send;
|
||||
}
|
||||
|
||||
// Unit type - no dependencies
|
||||
impl ExtractInput for () {
|
||||
async fn extract_from_channels(_receivers: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// Implementations for common primitive and standard library types.
|
||||
// These allow tasks to receive these types as single inputs without wrapping in tuples.
|
||||
// Receives Arc<T> from channel, clones the inner value for task consumption.
|
||||
// For custom types, implement ExtractInput following the same pattern.
|
||||
macro_rules! impl_extract_single {
|
||||
($($t:ty),+) => {
|
||||
$(
|
||||
impl ExtractInput for $t {
|
||||
async fn extract_from_channels(mut receivers: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
let rx = *receivers.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<$t>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Arc<{}>", std::any::type_name::<$t>()))?;
|
||||
let arc_value = rx.await.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
Ok((*arc_value).clone())
|
||||
}
|
||||
}
|
||||
)+
|
||||
};
|
||||
}
|
||||
|
||||
impl_extract_single!(
|
||||
i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String
|
||||
);
|
||||
|
||||
// Implement ExtractInput for Result types to support error handling patterns
|
||||
impl<T, E> ExtractInput for Result<T, E>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
E: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract_from_channels(
|
||||
mut receivers: Vec<Box<dyn Any + Send>>,
|
||||
) -> Result<Self, String> {
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
let rx = *receivers
|
||||
.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<Result<T, E>>>>()
|
||||
.map_err(|_| {
|
||||
format!(
|
||||
"Type mismatch: expected Arc<Result<{}, {}>>",
|
||||
std::any::type_name::<T>(),
|
||||
std::any::type_name::<E>()
|
||||
)
|
||||
})?;
|
||||
let arc_value = rx
|
||||
.await
|
||||
.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
Ok((*arc_value).clone())
|
||||
}
|
||||
}
|
||||
|
||||
// Implement ExtractInput for Option types
|
||||
impl<T> ExtractInput for Option<T>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract_from_channels(
|
||||
mut receivers: Vec<Box<dyn Any + Send>>,
|
||||
) -> Result<Self, String> {
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
let rx = *receivers
|
||||
.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<Option<T>>>>()
|
||||
.map_err(|_| {
|
||||
format!(
|
||||
"Type mismatch: expected Arc<Option<{}>>",
|
||||
std::any::type_name::<T>()
|
||||
)
|
||||
})?;
|
||||
let arc_value = rx
|
||||
.await
|
||||
.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
Ok((*arc_value).clone())
|
||||
}
|
||||
}
|
||||
|
||||
// Implement ExtractInput for Vec types
|
||||
impl<T> ExtractInput for Vec<T>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract_from_channels(
|
||||
mut receivers: Vec<Box<dyn Any + Send>>,
|
||||
) -> Result<Self, String> {
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
let rx = *receivers
|
||||
.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<Vec<T>>>>()
|
||||
.map_err(|_| {
|
||||
format!(
|
||||
"Type mismatch: expected Arc<Vec<{}>>",
|
||||
std::any::type_name::<T>()
|
||||
)
|
||||
})?;
|
||||
let arc_value = rx
|
||||
.await
|
||||
.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
Ok((*arc_value).clone())
|
||||
}
|
||||
}
|
||||
|
||||
// Implement ExtractInput for HashMap types
|
||||
impl<K, V> ExtractInput for std::collections::HashMap<K, V>
|
||||
where
|
||||
K: 'static + Clone + Send + Sync + Eq + std::hash::Hash,
|
||||
V: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract_from_channels(
|
||||
mut receivers: Vec<Box<dyn Any + Send>>,
|
||||
) -> Result<Self, String> {
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
let rx = *receivers
|
||||
.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<std::collections::HashMap<K, V>>>>()
|
||||
.map_err(|_| {
|
||||
format!(
|
||||
"Type mismatch: expected Arc<HashMap<{}, {}>>",
|
||||
std::any::type_name::<K>(),
|
||||
std::any::type_name::<V>()
|
||||
)
|
||||
})?;
|
||||
let arc_value = rx
|
||||
.await
|
||||
.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
Ok((*arc_value).clone())
|
||||
}
|
||||
}
|
||||
|
||||
// Implement ExtractInput for Arc types - SPECIAL CASE: just clone the Arc, don't unwrap!
|
||||
// This allows zero-copy data sharing for users who wrap their data in Arc.
|
||||
impl<T> ExtractInput for Arc<T>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract_from_channels(
|
||||
mut receivers: Vec<Box<dyn Any + Send>>,
|
||||
) -> Result<Self, String> {
|
||||
if receivers.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", receivers.len()));
|
||||
}
|
||||
let rx = *receivers
|
||||
.pop()
|
||||
.unwrap()
|
||||
.downcast::<oneshot::Receiver<Arc<Arc<T>>>>()
|
||||
.map_err(|_| {
|
||||
format!(
|
||||
"Type mismatch: expected Arc<Arc<{}>>",
|
||||
std::any::type_name::<T>()
|
||||
)
|
||||
})?;
|
||||
let arc_arc_value = rx
|
||||
.await
|
||||
.map_err(|_| "Channel closed before receiving value".to_string())?;
|
||||
// Return the inner Arc (clone is cheap - just refcount increment)
|
||||
Ok(Arc::clone(&*arc_arc_value))
|
||||
}
|
||||
}
|
||||
|
||||
/// Macro to implement ExtractInput for tuple types.
|
||||
///
|
||||
/// Generates an implementation that extracts values from multiple dependency channels
|
||||
/// and combines them into a tuple. All channels are awaited concurrently using join_all.
|
||||
///
|
||||
/// 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<Tk::Input>` trait bound on `depends_on()` ensures type/arity matching
|
||||
/// 2. The `IsUnitType` constraint on `From<TaskBuilder>` 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 : $idx:tt),+) => {
|
||||
impl<$($T: 'static + Clone + Send + Sync),+> ExtractInput for ($($T,)+) {
|
||||
async fn extract_from_channels(receivers: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
// Verify we have the correct number of receivers
|
||||
let expected_count = [$($idx),+].len();
|
||||
if receivers.len() != expected_count {
|
||||
return Err(format!("Expected {} dependencies, got {}", expected_count, receivers.len()));
|
||||
}
|
||||
|
||||
// Convert Vec to iterator so we can consume it
|
||||
let mut iter = receivers.into_iter();
|
||||
|
||||
// Extract each receiver by consuming from the iterator
|
||||
// Receivers now contain Arc-wrapped values for efficient fanout
|
||||
// Dereference Box after downcasting to get the receiver itself
|
||||
// Use type names as variable names to avoid numeric suffix issues
|
||||
$(
|
||||
#[allow(non_snake_case)]
|
||||
let $T = *iter.next()
|
||||
.ok_or_else(|| format!("Missing receiver at index {}", $idx))?
|
||||
.downcast::<oneshot::Receiver<Arc<$T>>>()
|
||||
.map_err(|_| format!("Type mismatch at index {}: expected Arc<{}>",
|
||||
$idx, std::any::type_name::<$T>()))?;
|
||||
)+
|
||||
|
||||
// Await all channels concurrently - this is a key optimization!
|
||||
// We can start receiving as soon as ANY dependency completes, not waiting
|
||||
// for them sequentially.
|
||||
let arc_results = futures::join!(
|
||||
$(
|
||||
async move {
|
||||
$T.await.map_err(|_| format!("Channel {} closed", $idx))
|
||||
}
|
||||
),+
|
||||
);
|
||||
|
||||
// Clone inner values from Arc (cheap for small types, necessary for ownership)
|
||||
Ok(($(
|
||||
(*arc_results.$idx?).clone()
|
||||
,)+))
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
// Generate ExtractInput implementations for tuples of size 2-8.
|
||||
// Each invocation creates an impl for a specific tuple size.
|
||||
// Supporting up to 8 elements covers the vast majority of use cases.
|
||||
impl_extract_tuple!(A:0, B:1);
|
||||
impl_extract_tuple!(A:0, B:1, C:2);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6, H:7);
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod coverage_tests;
|
||||
@@ -0,0 +1,378 @@
|
||||
//! Coverage tests for uncovered error paths in extract.rs
|
||||
//!
|
||||
//! These tests specifically target type mismatch error messages that aren't hit
|
||||
//! in normal usage because the #[task] macro generates type-safe code.
|
||||
//! We directly call extract_from_channels with mismatched types.
|
||||
|
||||
use crate::extract::ExtractInput;
|
||||
use futures::channel::oneshot;
|
||||
use std::any::Any;
|
||||
use std::collections::HashMap;
|
||||
use std::sync::Arc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_result_type_mismatch_coverage() {
|
||||
// Tests line 95 in extract.rs - Result<T,E> type mismatch error message
|
||||
// Create a channel that sends the WRONG type (String instead of Result<i32, String>)
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
|
||||
// Send a String value
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("hello".to_string()));
|
||||
});
|
||||
|
||||
// Try to extract as Result<i32, String> - this will fail with type mismatch
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Result::<i32, String>::extract_from_channels(deps).await;
|
||||
|
||||
// Verify the error message contains the expected type information
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
|
||||
// The error should mention Result type mismatch (line 95)
|
||||
assert!(
|
||||
err_msg.contains("Result"),
|
||||
"Expected type mismatch error mentioning Result, got: {}",
|
||||
err_msg
|
||||
);
|
||||
assert!(
|
||||
err_msg.contains("Type mismatch"),
|
||||
"Expected 'Type mismatch' in error, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_option_type_mismatch_coverage() {
|
||||
// Tests line 124 in extract.rs - Option<T> type mismatch error message
|
||||
// Create a channel that sends the WRONG type (i32 instead of Option<i32>)
|
||||
let (tx, rx) = oneshot::channel::<Arc<i32>>();
|
||||
|
||||
// Send an i32 value
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(42));
|
||||
});
|
||||
|
||||
// Try to extract as Option<i32> - this will fail with type mismatch
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Option::<i32>::extract_from_channels(deps).await;
|
||||
|
||||
// Verify the error message contains the expected type information
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
|
||||
// The error should mention Option type mismatch (line 124)
|
||||
assert!(
|
||||
err_msg.contains("Option"),
|
||||
"Expected type mismatch error mentioning Option, got: {}",
|
||||
err_msg
|
||||
);
|
||||
assert!(
|
||||
err_msg.contains("Type mismatch"),
|
||||
"Expected 'Type mismatch' in error, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_vec_type_mismatch_coverage() {
|
||||
// Tests line 152 in extract.rs - Vec<T> type mismatch error message
|
||||
// Create a channel that sends the WRONG type (String instead of Vec<u8>)
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
|
||||
// Send a String value
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("not a vec".to_string()));
|
||||
});
|
||||
|
||||
// Try to extract as Vec<u8> - this will fail with type mismatch
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Vec::<u8>::extract_from_channels(deps).await;
|
||||
|
||||
// Verify the error message contains the expected type information
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
|
||||
// The error should mention Vec type mismatch (line 152)
|
||||
assert!(
|
||||
err_msg.contains("Vec"),
|
||||
"Expected type mismatch error mentioning Vec, got: {}",
|
||||
err_msg
|
||||
);
|
||||
assert!(
|
||||
err_msg.contains("Type mismatch"),
|
||||
"Expected 'Type mismatch' in error, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hashmap_type_mismatch_coverage() {
|
||||
// Tests line 181 in extract.rs - HashMap<K,V> type mismatch error message
|
||||
// Create a channel that sends the WRONG type (Vec<i32> instead of HashMap)
|
||||
let (tx, rx) = oneshot::channel::<Arc<Vec<i32>>>();
|
||||
|
||||
// Send a Vec value
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(vec![1, 2, 3]));
|
||||
});
|
||||
|
||||
// Try to extract as HashMap<String, i32> - this will fail with type mismatch
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
|
||||
|
||||
// Verify the error message contains the expected type information
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
|
||||
// The error should mention HashMap type mismatch (line 181)
|
||||
assert!(
|
||||
err_msg.contains("HashMap"),
|
||||
"Expected type mismatch error mentioning HashMap, got: {}",
|
||||
err_msg
|
||||
);
|
||||
assert!(
|
||||
err_msg.contains("Type mismatch"),
|
||||
"Expected 'Type mismatch' in error, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_arc_type_mismatch_coverage() {
|
||||
// Tests line 211 in extract.rs - Arc<T> type mismatch error message
|
||||
// Arc is special: it expects Arc<Arc<T>> from channel
|
||||
// Create a channel that sends Arc<String> instead of Arc<Arc<String>>
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
|
||||
// Send an Arc<String> value (missing one level of Arc wrapping)
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("not double arc".to_string()));
|
||||
});
|
||||
|
||||
// Try to extract as Arc<String> - this will fail with type mismatch
|
||||
// because it expects Arc<Arc<String>> in the channel
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Arc::<String>::extract_from_channels(deps).await;
|
||||
|
||||
// Verify the error message contains the expected type information
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
|
||||
// The error should mention Arc type mismatch (line 211)
|
||||
assert!(
|
||||
err_msg.contains("Arc"),
|
||||
"Expected type mismatch error mentioning Arc, got: {}",
|
||||
err_msg
|
||||
);
|
||||
assert!(
|
||||
err_msg.contains("Type mismatch"),
|
||||
"Expected 'Type mismatch' in error, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
// Additional tests for better coverage of complex scenarios
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_result_with_wrong_ok_type() {
|
||||
// Test Result with wrong Ok type
|
||||
let (tx, rx) = oneshot::channel::<Arc<Result<String, String>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Ok("test".to_string())));
|
||||
});
|
||||
|
||||
// Try to extract as Result<i32, String> - wrong Ok type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Result::<i32, String>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Result") && err_msg.contains("Type mismatch"),
|
||||
"Expected Result type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_result_with_wrong_err_type() {
|
||||
// Test Result with wrong Err type
|
||||
let (tx, rx) = oneshot::channel::<Arc<Result<i32, i32>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Err(42)));
|
||||
});
|
||||
|
||||
// Try to extract as Result<i32, String> - wrong Err type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Result::<i32, String>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Result") && err_msg.contains("Type mismatch"),
|
||||
"Expected Result type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_option_with_wrong_inner_type() {
|
||||
// Test Option with wrong inner type
|
||||
let (tx, rx) = oneshot::channel::<Arc<Option<String>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Some("test".to_string())));
|
||||
});
|
||||
|
||||
// Try to extract as Option<i32> - wrong inner type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Option::<i32>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Option") && err_msg.contains("Type mismatch"),
|
||||
"Expected Option type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_vec_with_wrong_element_type() {
|
||||
// Test Vec with wrong element type
|
||||
let (tx, rx) = oneshot::channel::<Arc<Vec<String>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(vec!["a".to_string(), "b".to_string()]));
|
||||
});
|
||||
|
||||
// Try to extract as Vec<i32> - wrong element type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Vec::<i32>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Vec") && err_msg.contains("Type mismatch"),
|
||||
"Expected Vec type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hashmap_with_wrong_key_type() {
|
||||
// Test HashMap with wrong key type
|
||||
let (tx, rx) = oneshot::channel::<Arc<HashMap<i32, String>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut map = HashMap::new();
|
||||
map.insert(1, "one".to_string());
|
||||
let _ = tx.send(Arc::new(map));
|
||||
});
|
||||
|
||||
// Try to extract as HashMap<String, String> - wrong key type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = HashMap::<String, String>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("HashMap") && err_msg.contains("Type mismatch"),
|
||||
"Expected HashMap type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hashmap_with_wrong_value_type() {
|
||||
// Test HashMap with wrong value type
|
||||
let (tx, rx) = oneshot::channel::<Arc<HashMap<String, i32>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut map = HashMap::new();
|
||||
map.insert("key".to_string(), 42);
|
||||
let _ = tx.send(Arc::new(map));
|
||||
});
|
||||
|
||||
// Try to extract as HashMap<String, String> - wrong value type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = HashMap::<String, String>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("HashMap") && err_msg.contains("Type mismatch"),
|
||||
"Expected HashMap type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_arc_with_wrong_inner_type() {
|
||||
// Test Arc with completely wrong inner type
|
||||
// Send Arc<i32> but expect Arc<String>
|
||||
let (tx, rx) = oneshot::channel::<Arc<Arc<i32>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Arc::new(42)));
|
||||
});
|
||||
|
||||
// Try to extract as Arc<String> - wrong inner type
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Arc::<String>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Arc") && err_msg.contains("Type mismatch"),
|
||||
"Expected Arc type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_primitive_type_mismatch_for_completeness() {
|
||||
// Test primitive type mismatch (already covered in existing tests but good for completeness)
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("not an i32".to_string()));
|
||||
});
|
||||
|
||||
// Try to extract as i32 - type mismatch
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = i32::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Type mismatch"),
|
||||
"Expected type mismatch error, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_complex_nested_result_option() {
|
||||
// Test complex nested type: send Result<String, String>, expect Result<Option<i32>, String>
|
||||
let (tx, rx) = oneshot::channel::<Arc<Result<String, String>>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Ok("test".to_string())));
|
||||
});
|
||||
|
||||
// Try to extract as Result<Option<i32>, String> - nested type mismatch
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Result::<Option<i32>, String>::extract_from_channels(deps).await;
|
||||
|
||||
assert!(result.is_err());
|
||||
let err_msg = result.unwrap_err();
|
||||
assert!(
|
||||
err_msg.contains("Result") && err_msg.contains("Type mismatch"),
|
||||
"Expected Result type mismatch, got: {}",
|
||||
err_msg
|
||||
);
|
||||
}
|
||||
@@ -0,0 +1,410 @@
|
||||
//! Unit tests for extract module
|
||||
|
||||
#![allow(dead_code)] // Test helpers generate unused code via macro
|
||||
|
||||
use crate::extract::ExtractInput;
|
||||
use crate::task::Task;
|
||||
use futures::channel::oneshot;
|
||||
use std::any::Any;
|
||||
use std::sync::Arc;
|
||||
|
||||
// Test task for error scenarios
|
||||
#[allow(dead_code)]
|
||||
struct TestTask {
|
||||
value: i32,
|
||||
}
|
||||
|
||||
#[crate::task]
|
||||
#[allow(dead_code)]
|
||||
impl TestTask {
|
||||
#[cfg(not(tarpaulin_include))]
|
||||
async fn run(&self) -> i32 {
|
||||
self.value
|
||||
}
|
||||
}
|
||||
|
||||
#[allow(dead_code)]
|
||||
struct TestTaskWithInput;
|
||||
|
||||
#[crate::task]
|
||||
#[allow(dead_code)]
|
||||
impl TestTaskWithInput {
|
||||
#[cfg(not(tarpaulin_include))]
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input * 2
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_result_type_single_dependency_error() {
|
||||
// Test Result<T,E> with wrong dependency count
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<Result<i32, String>>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<Result<i32, String>>().1) as Box<dyn Any + Send>,
|
||||
]; // Two dependencies when expecting 1
|
||||
|
||||
let result = Result::<i32, String>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_result_type_success() {
|
||||
// Test Result<T,E> successful extraction
|
||||
let (tx, rx) = oneshot::channel::<Arc<Result<i32, String>>>();
|
||||
|
||||
// Send value in background
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Ok(42)));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Result::<i32, String>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), Ok(42));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_option_type_single_dependency_error() {
|
||||
// Test Option<T> with wrong dependency count
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<Option<String>>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<Option<String>>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<Option<String>>().1) as Box<dyn Any + Send>,
|
||||
]; // Three dependencies when expecting 1
|
||||
|
||||
let result = Option::<String>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 3"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_option_type_success() {
|
||||
// Test Option<T> successful extraction
|
||||
let (tx, rx) = oneshot::channel::<Arc<Option<i32>>>();
|
||||
|
||||
// Send value in background
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(Some(99)));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Option::<i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), Some(99));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_vec_type_single_dependency_error() {
|
||||
// Test Vec<T> with wrong dependency count
|
||||
let deps = vec![]; // Zero dependencies when expecting 1
|
||||
|
||||
let result = Vec::<u8>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 0"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_vec_type_success() {
|
||||
// Test Vec<T> successful extraction
|
||||
let (tx, rx) = oneshot::channel::<Arc<Vec<i32>>>();
|
||||
|
||||
// Send value in background
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(vec![1, 2, 3]));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Vec::<i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), vec![1, 2, 3]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_primitive_single_dependency_error() {
|
||||
// Test primitive types with wrong dependency count
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
|
||||
]; // Two dependencies when expecting 1
|
||||
|
||||
let result = i32::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
|
||||
|
||||
// Test with String
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<String>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<String>().1) as Box<dyn Any + Send>,
|
||||
];
|
||||
let result = String::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
|
||||
|
||||
// Test with f64
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<f64>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<f64>().1) as Box<dyn Any + Send>,
|
||||
];
|
||||
let result = f64::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_unit_type() {
|
||||
// Test that unit type always succeeds regardless of dependencies
|
||||
let deps = vec![];
|
||||
let result = <()>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
|
||||
// Even with dependencies (ignored), unit type succeeds
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<i32>().1) as Box<dyn Any + Send>,
|
||||
];
|
||||
let result = <()>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_tuple_basic() {
|
||||
// Test tuple extraction with channels
|
||||
let (tx1, rx1) = oneshot::channel::<Arc<i32>>();
|
||||
let (tx2, rx2) = oneshot::channel::<Arc<i32>>();
|
||||
|
||||
// Send values in background
|
||||
tokio::spawn(async move {
|
||||
let _ = tx1.send(Arc::new(42));
|
||||
let _ = tx2.send(Arc::new(100));
|
||||
});
|
||||
|
||||
let deps = vec![
|
||||
Box::new(rx1) as Box<dyn Any + Send>,
|
||||
Box::new(rx2) as Box<dyn Any + Send>,
|
||||
];
|
||||
|
||||
// Extract a tuple (i32, i32)
|
||||
let result = <(i32, i32)>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), (42, 100));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_channel_closed() {
|
||||
// Test error when channel is closed
|
||||
let (tx, rx) = oneshot::channel::<Arc<i32>>();
|
||||
// Explicitly drop tx to close the channel
|
||||
drop(tx);
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = i32::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Channel closed"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_type_mismatch() {
|
||||
// Test type mismatch during downcast
|
||||
let (tx, rx) = oneshot::channel::<String>();
|
||||
|
||||
// Send a String
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send("hello".to_string());
|
||||
});
|
||||
|
||||
// Try to extract as wrong type - put wrong receiver type in deps
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = i32::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Type mismatch"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_all_primitive_types() {
|
||||
// Test that all primitive types in impl_extract_single! work correctly
|
||||
macro_rules! test_type {
|
||||
($t:ty, $val:expr) => {
|
||||
let (tx, rx) = oneshot::channel::<Arc<$t>>();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new($val));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = <$t>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), $val);
|
||||
};
|
||||
}
|
||||
|
||||
test_type!(i8, 42i8);
|
||||
test_type!(i16, 42i16);
|
||||
test_type!(i32, 42i32);
|
||||
test_type!(i64, 42i64);
|
||||
test_type!(String, "test".to_string());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_larger_tuples() {
|
||||
// Test extraction for larger tuple sizes
|
||||
let (tx1, rx1) = oneshot::channel::<Arc<i32>>();
|
||||
let (tx2, rx2) = oneshot::channel::<Arc<i32>>();
|
||||
let (tx3, rx3) = oneshot::channel::<Arc<i32>>();
|
||||
let (tx4, rx4) = oneshot::channel::<Arc<i32>>();
|
||||
let (tx5, rx5) = oneshot::channel::<Arc<i32>>();
|
||||
|
||||
// Send values in background
|
||||
tokio::spawn(async move {
|
||||
let _ = tx1.send(Arc::new(0));
|
||||
let _ = tx2.send(Arc::new(1));
|
||||
let _ = tx3.send(Arc::new(2));
|
||||
let _ = tx4.send(Arc::new(3));
|
||||
let _ = tx5.send(Arc::new(4));
|
||||
});
|
||||
|
||||
let deps = vec![
|
||||
Box::new(rx1) as Box<dyn Any + Send>,
|
||||
Box::new(rx2) as Box<dyn Any + Send>,
|
||||
Box::new(rx3) as Box<dyn Any + Send>,
|
||||
Box::new(rx4) as Box<dyn Any + Send>,
|
||||
Box::new(rx5) as Box<dyn Any + Send>,
|
||||
];
|
||||
|
||||
// Extract a 5-element tuple
|
||||
let result = <(i32, i32, i32, i32, i32)>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(result.unwrap(), (0, 1, 2, 3, 4));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_extract_arc_type() {
|
||||
// Test Arc<T> extraction
|
||||
let (tx, rx) = oneshot::channel::<Arc<Arc<String>>>();
|
||||
|
||||
let test_arc = Arc::new("test".to_string());
|
||||
let test_arc_clone = test_arc.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new(test_arc));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Arc::<String>::extract_from_channels(deps).await;
|
||||
assert!(result.is_ok());
|
||||
assert_eq!(*result.unwrap(), *test_arc_clone);
|
||||
}
|
||||
|
||||
// Error path tests for ExtractInput implementations
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hashmap_extract_wrong_dependency_count() {
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<Arc<HashMap<String, i32>>>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<Arc<HashMap<String, i32>>>().1) as Box<dyn Any + Send>,
|
||||
];
|
||||
|
||||
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 2"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hashmap_extract_type_mismatch() {
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("wrong type".to_string()));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Type mismatch"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_hashmap_extract_channel_closed() {
|
||||
let (tx, rx) = oneshot::channel::<Arc<HashMap<String, i32>>>();
|
||||
drop(tx);
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = HashMap::<String, i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Channel closed"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_result_extract_downcast_failure() {
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("wrong".to_string()));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Result::<i32, String>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Type mismatch"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_option_extract_downcast_failure() {
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("wrong".to_string()));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Option::<i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Type mismatch"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_vec_extract_downcast_failure() {
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("wrong".to_string()));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Vec::<i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Type mismatch"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_arc_extract_wrong_dependency_count() {
|
||||
let deps = vec![];
|
||||
let result = Arc::<i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Expected 1 dependency, got 0"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_arc_extract_downcast_failure() {
|
||||
let (tx, rx) = oneshot::channel::<Arc<String>>();
|
||||
tokio::spawn(async move {
|
||||
let _ = tx.send(Arc::new("wrong".to_string()));
|
||||
});
|
||||
|
||||
let deps = vec![Box::new(rx) as Box<dyn Any + Send>];
|
||||
let result = Arc::<i32>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result.unwrap_err().contains("Type mismatch"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_tuple_wrong_dependency_count() {
|
||||
let deps = vec![
|
||||
Box::new(oneshot::channel::<Arc<i32>>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<Arc<i32>>().1) as Box<dyn Any + Send>,
|
||||
Box::new(oneshot::channel::<Arc<i32>>().1) as Box<dyn Any + Send>,
|
||||
];
|
||||
|
||||
let result = <(i32, i32)>::extract_from_channels(deps).await;
|
||||
assert!(result.is_err());
|
||||
assert!(result
|
||||
.unwrap_err()
|
||||
.contains("Expected 2 dependencies, got 3"));
|
||||
}
|
||||
+51
-2
@@ -10,6 +10,9 @@
|
||||
//! - **Compile-time type safety**: Dependencies are validated at compile time through the type
|
||||
//! system. The public API is fully type-safe with no runtime type errors. Internal execution
|
||||
//! uses type erasure for heterogeneous task storage, but this is never exposed to users.
|
||||
//! - **Works with ANY type**: Custom types work automatically with the `#[task]` macro—just
|
||||
//! implement `Clone + Send + Sync`, no trait implementations needed! The macro generates
|
||||
//! type-specific extraction logic for seamless integration.
|
||||
//! - **Runtime-agnostic**: Works with any async runtime (Tokio, async-std, smol, Embassy, etc.)
|
||||
//! - **Type-state pattern**: The API guides you with compile-time errors if you wire dependencies
|
||||
//! incorrectly. This is what prevents cycles at compile time.
|
||||
@@ -113,7 +116,6 @@
|
||||
//! use dagx::{task, Task};
|
||||
//!
|
||||
//! // Define a struct to hold your multiple values
|
||||
//! #[derive(Clone)]
|
||||
//! struct UserData {
|
||||
//! name: String,
|
||||
//! age: i32,
|
||||
@@ -434,6 +436,53 @@
|
||||
//! # };
|
||||
//! ```
|
||||
//!
|
||||
//! ## Custom Types
|
||||
//!
|
||||
//! **dagx works with ANY type automatically!** As long as your type implements `Clone + Send + Sync + 'static`,
|
||||
//! the `#[task]` macro generates the necessary extraction logic:
|
||||
//!
|
||||
//! ```no_run
|
||||
//! use dagx::{task, DagRunner, Task};
|
||||
//!
|
||||
//! // Just derive Clone - that's all you need!
|
||||
//! #[derive(Clone)]
|
||||
//! struct User {
|
||||
//! name: String,
|
||||
//! age: u32,
|
||||
//! }
|
||||
//!
|
||||
//! struct CreateUser;
|
||||
//! #[task]
|
||||
//! impl CreateUser {
|
||||
//! async fn run() -> User {
|
||||
//! User { name: "Alice".to_string(), age: 30 }
|
||||
//! }
|
||||
//! }
|
||||
//!
|
||||
//! struct FormatUser;
|
||||
//! #[task]
|
||||
//! impl FormatUser {
|
||||
//! async fn run(user: &User) -> String {
|
||||
//! format!("{} is {} years old", user.name, user.age)
|
||||
//! }
|
||||
//! }
|
||||
//!
|
||||
//! # async {
|
||||
//! let dag = DagRunner::new();
|
||||
//!
|
||||
//! let user = dag.add_task(CreateUser);
|
||||
//! let formatted = dag.add_task(FormatUser).depends_on(&user);
|
||||
//!
|
||||
//! dag.run(|fut| { tokio::spawn(fut); }).await.unwrap();
|
||||
//!
|
||||
//! assert_eq!(dag.get(formatted).unwrap(), "Alice is 30 years old");
|
||||
//! # };
|
||||
//! ```
|
||||
//!
|
||||
//! **No trait implementations needed!** Works with nested structs, collections, enums, and any
|
||||
//! other custom type. See [`examples/custom_types.rs`](https://github.com/swaits/dagx/blob/main/examples/custom_types.rs)
|
||||
//! for a complete example with complex nested types.
|
||||
//!
|
||||
//! # Runtime Agnostic
|
||||
//!
|
||||
//! dagx works with any async runtime. The library has been tested with:
|
||||
@@ -723,6 +772,7 @@ mod builder;
|
||||
pub mod cycle_prevention;
|
||||
mod deps;
|
||||
mod error;
|
||||
mod extract;
|
||||
mod node;
|
||||
mod runner;
|
||||
mod task;
|
||||
@@ -731,7 +781,6 @@ mod types;
|
||||
// Public re-exports
|
||||
pub use builder::TaskBuilder;
|
||||
pub use error::{DagError, DagResult};
|
||||
pub use node::TaskExecute; // For macro-generated trait implementations
|
||||
pub use runner::DagRunner;
|
||||
pub use task::Task;
|
||||
pub use types::{Pending, TaskHandle};
|
||||
|
||||
+72
-83
@@ -11,146 +11,135 @@ use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
use std::sync::Arc;
|
||||
|
||||
use futures::channel::oneshot;
|
||||
|
||||
use crate::error::DagError;
|
||||
use crate::task::Task;
|
||||
use crate::types::NodeId;
|
||||
|
||||
/// Type-erased listeners (for receiving inputs from dependencies)
|
||||
type ErasedListeners = Vec<Box<dyn Any + Send>>;
|
||||
/// Type-erased channel endpoints (senders or receivers)
|
||||
type ErasedChannels = Vec<Box<dyn Any + Send>>;
|
||||
|
||||
/// Type-erased channel pair (senders, receivers)
|
||||
type ChannelPair = (ErasedChannels, ErasedChannels);
|
||||
|
||||
/// Type-erased async execution result (Arc-wrapped for efficient fanout)
|
||||
type ExecuteFuture =
|
||||
Pin<Box<dyn Future<Output = Result<Arc<dyn Any + Send + Sync>, DagError>> + Send>>;
|
||||
|
||||
/// Type-erased async execution result for TaskExecute trait (returns String errors)
|
||||
type TaskExecuteFuture =
|
||||
Pin<Box<dyn Future<Output = Result<Arc<dyn Any + Send + Sync>, String>> + Send>>;
|
||||
|
||||
/// Internal trait for executing heterogeneous tasks.
|
||||
///
|
||||
/// This trait provides type erasure necessary for storing different task types in a single
|
||||
/// collection. While the public API (Task, Handle, Node) is fully type-safe, internally we
|
||||
/// need dynamic dispatch to execute tasks with different input/output types.
|
||||
///
|
||||
/// This is the ONLY place where type erasure occurs - announcements are created with full type
|
||||
/// This is the ONLY place where type erasure occurs - channels are created with full type
|
||||
/// information and only erased at this trait boundary.
|
||||
///
|
||||
/// This is an implementation detail and not part of the public API.
|
||||
pub(crate) trait ExecutableNode: Send {
|
||||
/// Create a type-erased announcement for this node's output.
|
||||
/// Create type-erased output channels for this node's dependents.
|
||||
///
|
||||
/// Returns a tuple of (Announcer, Announcement) both type-erased.
|
||||
/// The Announcer will be used by this node to broadcast its output.
|
||||
/// The Announcement can be used to create listeners for dependents.
|
||||
fn create_announcement(&self) -> (Box<dyn Any + Send>, Box<dyn Any + Send>);
|
||||
/// Returns (senders, receivers) where:
|
||||
/// - senders: Type-erased senders that this node will use to send outputs
|
||||
/// - receivers: Type-erased receivers that dependents will use to receive inputs
|
||||
fn create_output_channels(&self, num_dependents: usize) -> ChannelPair;
|
||||
|
||||
/// Create a type-erased listener from an announcement.
|
||||
///
|
||||
/// This allows dependents to listen to this node's output.
|
||||
fn create_listener(&self, announcement: &dyn Any) -> Box<dyn Any + Send>;
|
||||
|
||||
/// Execute the task with type-erased input listeners and an output announcer.
|
||||
/// Execute the task with type-erased input receivers and output senders.
|
||||
///
|
||||
/// The TypedNode implementation knows the concrete types and can safely downcast.
|
||||
/// Consumes the node and returns the output value.
|
||||
fn execute_with_listeners(
|
||||
fn execute_with_channels(
|
||||
self: Box<Self>,
|
||||
listeners: ErasedListeners,
|
||||
announcer: Box<dyn Any + Send>,
|
||||
receivers: ErasedChannels,
|
||||
senders: ErasedChannels,
|
||||
) -> ExecuteFuture;
|
||||
}
|
||||
|
||||
/// Fully-typed node storage for a single task.
|
||||
///
|
||||
/// Stores a task that has an `execute()` method (generated by the #[task] macro).
|
||||
/// Implements ExecutableNode for type erasure, allowing heterogeneous tasks to be
|
||||
/// stored in a single Vec.
|
||||
/// Stores a task of type T. Implements ExecutableNode for type erasure,
|
||||
/// allowing heterogeneous tasks to be stored in a single Vec.
|
||||
///
|
||||
/// # Ownership Model
|
||||
///
|
||||
/// The task is owned directly (no Mutex needed) and consumed during execution.
|
||||
/// Each task executes exactly once, taking ownership and producing an output.
|
||||
///
|
||||
/// # Announcement Broadcasting
|
||||
/// # Oneshot Channels
|
||||
///
|
||||
/// Outputs are communicated via announcement channels for inter-task communication.
|
||||
/// Each task has ONE announcer that broadcasts to multiple listeners.
|
||||
/// The output is also returned from execute_with_listeners for central storage.
|
||||
pub(crate) struct TypedNode<T> {
|
||||
/// Outputs are communicated via oneshot channels for inter-task communication.
|
||||
/// The output is also returned from execute_with_channels for central storage.
|
||||
pub(crate) struct TypedNode<T: Task> {
|
||||
pub(crate) id: NodeId,
|
||||
pub(crate) task: T,
|
||||
pub(crate) _output_type: std::marker::PhantomData<fn() -> ()>,
|
||||
}
|
||||
|
||||
impl<T> TypedNode<T> {
|
||||
impl<T: Task + 'static> TypedNode<T> {
|
||||
pub(crate) fn new(id: NodeId, task: T) -> Self {
|
||||
Self {
|
||||
id,
|
||||
task,
|
||||
_output_type: std::marker::PhantomData,
|
||||
}
|
||||
Self { id, task }
|
||||
}
|
||||
}
|
||||
|
||||
// Implementation for tasks created with the #[task] macro
|
||||
// These tasks have an execute() method generated by the macro
|
||||
impl<T: 'static + Send> ExecutableNode for TypedNode<T>
|
||||
impl<T: Task + 'static> ExecutableNode for TypedNode<T>
|
||||
where
|
||||
T: TaskExecute,
|
||||
T::Input: 'static + Clone,
|
||||
T::Output: 'static + Clone,
|
||||
{
|
||||
fn create_announcement(&self) -> (Box<dyn Any + Send>, Box<dyn Any + Send>) {
|
||||
T::create_announcement_erased()
|
||||
fn create_output_channels(&self, num_dependents: usize) -> ChannelPair {
|
||||
let mut senders = Vec::new();
|
||||
let mut receivers = Vec::new();
|
||||
|
||||
for _ in 0..num_dependents {
|
||||
// Create channels that pass Arc-wrapped outputs (efficient for fanout)
|
||||
let (tx, rx) = oneshot::channel::<Arc<T::Output>>();
|
||||
senders.push(Box::new(tx) as Box<dyn Any + Send>);
|
||||
receivers.push(Box::new(rx) as Box<dyn Any + Send>);
|
||||
}
|
||||
|
||||
(senders, receivers)
|
||||
}
|
||||
|
||||
fn create_listener(&self, announcement: &dyn Any) -> Box<dyn Any + Send> {
|
||||
T::create_listener_erased(announcement)
|
||||
}
|
||||
|
||||
fn execute_with_listeners(
|
||||
fn execute_with_channels(
|
||||
self: Box<Self>,
|
||||
listeners: ErasedListeners,
|
||||
announcer: Box<dyn Any + Send>,
|
||||
receivers: ErasedChannels,
|
||||
senders: ErasedChannels,
|
||||
) -> ExecuteFuture {
|
||||
Box::pin(async move {
|
||||
// Call the task's execute() method (generated by the #[task] macro)
|
||||
// This handles listener extraction, calling run(), and announcing
|
||||
let result = self.task.execute(listeners, announcer).await;
|
||||
// Use the task's extract_and_run method which has inline extraction logic
|
||||
// generated by the #[task] macro. This allows ANY type to work without
|
||||
// requiring ExtractInput trait implementations.
|
||||
let output = match self.task.extract_and_run(receivers).await {
|
||||
Ok(output) => output,
|
||||
Err(_msg) => {
|
||||
// Failed to extract input - likely channel closed or type mismatch
|
||||
return Err(DagError::InvalidDependency { task_id: self.id.0 });
|
||||
}
|
||||
};
|
||||
|
||||
match result {
|
||||
Ok(output_arc) => Ok(output_arc),
|
||||
Err(err_msg) => {
|
||||
// Execution failed - convert error message to DagError
|
||||
eprintln!("Task execution failed: {}", err_msg);
|
||||
Err(DagError::InvalidDependency { task_id: self.id.0 })
|
||||
// Arc Wrapping Strategy:
|
||||
// Wrap output in Arc ONCE for efficient fan-out sharing.
|
||||
// For 1:N dependencies, this enables O(1) sharing (Arc clone = atomic increment)
|
||||
// instead of O(N) cloning (N deep copies of output data).
|
||||
// Downstream tasks receive &T after extraction from Arc.
|
||||
let arc_output = Arc::new(output);
|
||||
|
||||
// Send Arc to all dependents via their channels
|
||||
// TypedNode<T> knows T::Output, so downcast is safe
|
||||
for sender_any in senders {
|
||||
if let Ok(sender) = sender_any.downcast::<oneshot::Sender<Arc<T::Output>>>() {
|
||||
// Clone Arc (just atomic refcount increment, NOT data clone)
|
||||
// This is the key performance optimization for fan-out patterns
|
||||
// Ignore send errors - dependent might have been cancelled
|
||||
let _ = sender.send(Arc::clone(&arc_output));
|
||||
}
|
||||
}
|
||||
|
||||
// Return Arc-wrapped output (for storage by runner)
|
||||
Ok(arc_output as Arc<dyn Any + Send + Sync>)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
/// Trait for tasks that have an execute() method (generated by #[task] macro).
|
||||
///
|
||||
/// This trait is automatically implemented by the macro and should not be
|
||||
/// implemented manually.
|
||||
pub trait TaskExecute: Send {
|
||||
/// Input type as a tuple of parameter types (for dependency checking)
|
||||
/// - `()` for no parameters
|
||||
/// - `T` for single parameter
|
||||
/// - `(A, B, C, ...)` for multiple parameters
|
||||
type Input: Send;
|
||||
|
||||
/// Output type of the task
|
||||
type Output: Send + Sync + Clone;
|
||||
|
||||
fn execute(
|
||||
self,
|
||||
listeners: Vec<Box<dyn Any + Send>>,
|
||||
announcer: Box<dyn Any + Send>,
|
||||
) -> TaskExecuteFuture;
|
||||
|
||||
fn create_announcement_erased() -> (Box<dyn Any + Send>, Box<dyn Any + Send>);
|
||||
|
||||
fn create_listener_erased(announcement: &dyn Any) -> Box<dyn Any + Send>;
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
+64
-93
@@ -1,9 +1,7 @@
|
||||
//! Unit tests for node module
|
||||
|
||||
// Allow tests to reference dagx:: paths (for proc macro generated code)
|
||||
extern crate self as dagx;
|
||||
|
||||
use crate::node::{ExecutableNode, TypedNode};
|
||||
use crate::task::Task;
|
||||
use crate::types::NodeId;
|
||||
|
||||
// Simple test task
|
||||
@@ -19,33 +17,25 @@ impl TestTask {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_create_announcement() {
|
||||
fn test_create_output_channels() {
|
||||
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
|
||||
|
||||
// Create ONE announcement that can broadcast to multiple listeners
|
||||
let (announcer, announcement) = node.create_announcement();
|
||||
// Create channels for 3 dependents
|
||||
let (senders, receivers) = node.create_output_channels(3);
|
||||
|
||||
// Verify both are created (type-erased)
|
||||
assert!(std::any::Any::type_id(&*announcer) != std::any::TypeId::of::<()>());
|
||||
assert!(std::any::Any::type_id(&*announcement) != std::any::TypeId::of::<()>());
|
||||
assert_eq!(senders.len(), 3);
|
||||
assert_eq!(receivers.len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_create_multiple_listeners() {
|
||||
fn test_create_output_channels_zero_dependents() {
|
||||
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
|
||||
|
||||
// Create announcement
|
||||
let (_announcer, announcement) = node.create_announcement();
|
||||
// Create channels for 0 dependents (sink node)
|
||||
let (senders, receivers) = node.create_output_channels(0);
|
||||
|
||||
// Create multiple listeners from the same announcement (broadcast pattern)
|
||||
let listener1 = node.create_listener(announcement.as_ref());
|
||||
let listener2 = node.create_listener(announcement.as_ref());
|
||||
let listener3 = node.create_listener(announcement.as_ref());
|
||||
|
||||
// All listeners are created from the same announcement
|
||||
assert!(std::any::Any::type_id(&*listener1) != std::any::TypeId::of::<()>());
|
||||
assert!(std::any::Any::type_id(&*listener2) != std::any::TypeId::of::<()>());
|
||||
assert!(std::any::Any::type_id(&*listener3) != std::any::TypeId::of::<()>());
|
||||
assert_eq!(senders.len(), 0);
|
||||
assert_eq!(receivers.len(), 0);
|
||||
}
|
||||
|
||||
struct TaskWithDependency;
|
||||
@@ -58,19 +48,15 @@ impl TaskWithDependency {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_with_listeners_missing_input() {
|
||||
async fn test_execute_with_channels_missing_input() {
|
||||
// Test when task expects input but doesn't receive it
|
||||
let node = TypedNode::new(NodeId(0), TaskWithDependency);
|
||||
let node = Box::new(TypedNode::new(NodeId(0), TaskWithDependency));
|
||||
|
||||
// Create a real announcer even though we won't use it
|
||||
let (announcer, _announcement) = node.create_announcement();
|
||||
// TaskWithDependency expects an i32 input, but we provide no receivers
|
||||
let receivers = vec![];
|
||||
let senders = vec![];
|
||||
|
||||
let node = Box::new(node);
|
||||
|
||||
// TaskWithDependency expects an i32 input, but we provide no listeners
|
||||
let listeners = vec![];
|
||||
|
||||
let result = node.execute_with_listeners(listeners, announcer).await;
|
||||
let result = node.execute_with_channels(receivers, senders).await;
|
||||
assert!(result.is_err());
|
||||
|
||||
if let Err(crate::error::DagError::InvalidDependency { task_id }) = result {
|
||||
@@ -81,62 +67,49 @@ async fn test_execute_with_listeners_missing_input() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_with_listeners_success() {
|
||||
// Test successful execution with announcements and listeners
|
||||
let dependency_node = TypedNode::new(NodeId(0), TestTask { value: 21 });
|
||||
async fn test_execute_with_channels_success() {
|
||||
// Test successful execution with channels
|
||||
let dependency_node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 21 }));
|
||||
let dependent_node = Box::new(TypedNode::new(NodeId(1), TaskWithDependency));
|
||||
|
||||
// Create announcement from dependency
|
||||
let (dep_announcer, dep_announcement) = dependency_node.create_announcement();
|
||||
// Create a channel from dependency to dependent (Arc-wrapped)
|
||||
let (tx, rx) = futures::channel::oneshot::channel::<std::sync::Arc<i32>>();
|
||||
|
||||
// Create listener for dependent from dependency's announcement
|
||||
let listener = dependency_node.create_listener(dep_announcement.as_ref());
|
||||
|
||||
// Execute dependency node with announcer
|
||||
let dependency_node = Box::new(dependency_node);
|
||||
let dep_listeners = vec![];
|
||||
// Execute dependency node with sender
|
||||
let dep_senders = vec![Box::new(tx) as Box<dyn std::any::Any + Send>];
|
||||
let dep_receivers = vec![];
|
||||
|
||||
// Execute in background
|
||||
let dep_handle = tokio::spawn(async move {
|
||||
dependency_node
|
||||
.execute_with_listeners(dep_listeners, dep_announcer)
|
||||
.execute_with_channels(dep_receivers, dep_senders)
|
||||
.await
|
||||
});
|
||||
|
||||
// Execute dependent node with listener
|
||||
// Create a real announcer for the dependent (even though no one listens to it)
|
||||
let dep_node_for_announcer = TypedNode::new(NodeId(1), TaskWithDependency);
|
||||
let (dependent_announcer, _dep_announcement) = dep_node_for_announcer.create_announcement();
|
||||
|
||||
let listeners = vec![listener];
|
||||
// Execute dependent node with receiver
|
||||
let receivers = vec![Box::new(rx) as Box<dyn std::any::Any + Send>];
|
||||
let senders = vec![];
|
||||
|
||||
let result = dependent_node
|
||||
.execute_with_listeners(listeners, dependent_announcer)
|
||||
.execute_with_channels(receivers, senders)
|
||||
.await;
|
||||
|
||||
// Wait for dependency to complete
|
||||
let _ = dep_handle.await;
|
||||
|
||||
assert!(result.is_ok());
|
||||
let output = result.unwrap();
|
||||
let arc_value = output.downcast::<i32>().unwrap();
|
||||
assert_eq!(*arc_value, 42); // 21 * 2 = 42
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_with_listeners_sink_node() {
|
||||
async fn test_execute_with_channels_sink_node() {
|
||||
// Test that sink nodes (no dependents) return their output
|
||||
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
|
||||
let node = Box::new(TypedNode::new(NodeId(0), TestTask { value: 42 }));
|
||||
|
||||
// Create a real announcer (even though no one will listen)
|
||||
let (announcer, _announcement) = node.create_announcement();
|
||||
// No senders means this is a sink node
|
||||
let receivers = vec![];
|
||||
let senders = vec![];
|
||||
|
||||
let node = Box::new(node);
|
||||
|
||||
// Sink node has no listeners
|
||||
let listeners = vec![];
|
||||
|
||||
let result = node.execute_with_listeners(listeners, announcer).await;
|
||||
let result = node.execute_with_channels(receivers, senders).await;
|
||||
assert!(result.is_ok());
|
||||
|
||||
// Check output was returned (Arc-wrapped)
|
||||
@@ -146,28 +119,29 @@ async fn test_execute_with_listeners_sink_node() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_with_listeners_non_sink_node() {
|
||||
// Test that non-sink nodes broadcast output via announcement
|
||||
async fn test_execute_with_channels_non_sink_node() {
|
||||
// Test that non-sink nodes send output via channels
|
||||
let node = TypedNode::new(NodeId(0), TestTask { value: 42 });
|
||||
|
||||
// Create announcement and listener
|
||||
let (announcer, announcement) = node.create_announcement();
|
||||
let listener = node.create_listener(announcement.as_ref());
|
||||
// Create output channels for one dependent
|
||||
let (senders, mut receivers) = node.create_output_channels(1);
|
||||
|
||||
// Execute with announcer (non-sink node)
|
||||
// Execute with senders (non-sink node)
|
||||
let node = Box::new(node);
|
||||
let listeners = vec![];
|
||||
let result = node.execute_with_listeners(listeners, announcer).await;
|
||||
let exec_receivers = vec![];
|
||||
let result = node.execute_with_channels(exec_receivers, senders).await;
|
||||
assert!(result.is_ok());
|
||||
|
||||
// Verify output was broadcast through announcement
|
||||
let listener = *listener
|
||||
.downcast::<announcement::Listener<std::sync::Arc<i32>>>()
|
||||
// Verify output was sent through channel (Arc-wrapped)
|
||||
let rx = receivers
|
||||
.pop()
|
||||
.unwrap()
|
||||
.downcast::<futures::channel::oneshot::Receiver<std::sync::Arc<i32>>>()
|
||||
.unwrap();
|
||||
|
||||
let received = listener.listen().await;
|
||||
assert!(received.is_ok());
|
||||
assert_eq!(*received.unwrap(), 42);
|
||||
let received_arc = rx.await;
|
||||
assert!(received_arc.is_ok());
|
||||
assert_eq!(*received_arc.unwrap(), 42);
|
||||
|
||||
// Also verify output was returned (Arc-wrapped)
|
||||
let output = result.unwrap();
|
||||
@@ -176,30 +150,27 @@ async fn test_execute_with_listeners_non_sink_node() {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_execute_with_multiple_listeners() {
|
||||
// Test that output is broadcast to multiple listeners
|
||||
async fn test_execute_with_multiple_dependents() {
|
||||
// Test that output is sent to multiple dependents
|
||||
let node = TypedNode::new(NodeId(0), TestTask { value: 100 });
|
||||
|
||||
// Create announcement and multiple listeners
|
||||
let (announcer, announcement) = node.create_announcement();
|
||||
let listener1 = node.create_listener(announcement.as_ref());
|
||||
let listener2 = node.create_listener(announcement.as_ref());
|
||||
let listener3 = node.create_listener(announcement.as_ref());
|
||||
// Create output channels for three dependents
|
||||
let (senders, receivers) = node.create_output_channels(3);
|
||||
|
||||
// Execute node
|
||||
let node = Box::new(node);
|
||||
let listeners = vec![];
|
||||
let result = node.execute_with_listeners(listeners, announcer).await;
|
||||
let exec_receivers = vec![];
|
||||
let result = node.execute_with_channels(exec_receivers, senders).await;
|
||||
assert!(result.is_ok());
|
||||
|
||||
// Verify all three listeners receive the output
|
||||
for listener in [listener1, listener2, listener3] {
|
||||
let listener = *listener
|
||||
.downcast::<announcement::Listener<std::sync::Arc<i32>>>()
|
||||
// Verify all three dependents receive the output (Arc-wrapped)
|
||||
for receiver in receivers {
|
||||
let rx = receiver
|
||||
.downcast::<futures::channel::oneshot::Receiver<std::sync::Arc<i32>>>()
|
||||
.unwrap();
|
||||
|
||||
let received = listener.listen().await;
|
||||
assert!(received.is_ok());
|
||||
assert_eq!(*received.unwrap(), 100);
|
||||
let received_arc = rx.await;
|
||||
assert!(received_arc.is_ok());
|
||||
assert_eq!(*received_arc.unwrap(), 100);
|
||||
}
|
||||
}
|
||||
+42
-44
@@ -23,6 +23,7 @@ use tracing::{debug, error, info, trace};
|
||||
use crate::builder::TaskBuilder;
|
||||
use crate::error::{DagError, DagResult};
|
||||
use crate::node::{ExecutableNode, TypedNode};
|
||||
use crate::task::Task;
|
||||
use crate::types::{NodeId, Pending, TaskHandle};
|
||||
|
||||
// Guard to ensure run_lock is released even on early return or panic
|
||||
@@ -193,7 +194,9 @@ impl DagRunner {
|
||||
/// ```
|
||||
pub fn add_task<Tk>(&self, task: Tk) -> TaskBuilder<'_, Tk, Pending>
|
||||
where
|
||||
Tk: crate::node::TaskExecute + 'static,
|
||||
Tk: Task + 'static,
|
||||
Tk::Input: 'static + Clone,
|
||||
Tk::Output: 'static + Clone,
|
||||
{
|
||||
let id = self.alloc_id();
|
||||
|
||||
@@ -317,41 +320,39 @@ impl DagRunner {
|
||||
|
||||
let edges = self.edges.lock().clone();
|
||||
|
||||
// Create ONE announcement per task
|
||||
// Each task broadcasts its output to all dependents via announcement
|
||||
let mut announcements: HashMap<NodeId, Box<dyn std::any::Any + Send>> = HashMap::new();
|
||||
let mut announcers: HashMap<NodeId, Box<dyn std::any::Any + Send>> = HashMap::new();
|
||||
|
||||
// Create announcements for all tasks
|
||||
{
|
||||
let nodes_lock = self.nodes.lock();
|
||||
for (node_id, node_opt) in nodes_lock.iter().enumerate() {
|
||||
if let Some(node) = node_opt {
|
||||
let (announcer, announcement) = node.create_announcement();
|
||||
announcers.insert(NodeId(node_id), announcer);
|
||||
announcements.insert(NodeId(node_id), announcement);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Create listeners for each consumer from their dependencies
|
||||
let mut consumer_listeners: HashMap<NodeId, Vec<Box<dyn std::any::Any + Send>>> =
|
||||
// Create all channels upfront
|
||||
// Map: (producer_id, consumer_id) -> receiver index
|
||||
// We store receivers in a Vec per consumer to maintain order
|
||||
let mut consumer_receivers: HashMap<NodeId, Vec<Box<dyn std::any::Any + Send>>> =
|
||||
HashMap::new();
|
||||
let mut producer_senders: HashMap<NodeId, Vec<Box<dyn std::any::Any + Send>>> =
|
||||
HashMap::new();
|
||||
|
||||
// Create channels for each producer-consumer relationship
|
||||
// We need to lock nodes temporarily to call create_output_channels
|
||||
{
|
||||
let nodes_lock = self.nodes.lock();
|
||||
|
||||
for (consumer_id, producer_ids) in &edges {
|
||||
let mut listeners = Vec::new();
|
||||
for &producer_id in producer_ids {
|
||||
if let Some(announcement) = announcements.get(&producer_id) {
|
||||
if let Some(Some(producer_node)) = nodes_lock.get(producer_id.0) {
|
||||
// Producer node creates listener from its own announcement
|
||||
let listener = producer_node.create_listener(announcement.as_ref());
|
||||
listeners.push(listener);
|
||||
}
|
||||
// Get the producer node (it's still in the vector at this point)
|
||||
if let Some(Some(producer_node)) = nodes_lock.get(producer_id.0) {
|
||||
// Ask the producer to create ONE channel for this consumer
|
||||
let (mut senders, mut receivers) = producer_node.create_output_channels(1);
|
||||
|
||||
// Store the sender for the producer
|
||||
producer_senders
|
||||
.entry(producer_id)
|
||||
.or_default()
|
||||
.push(senders.pop().unwrap());
|
||||
|
||||
// Store the receiver for the consumer (order matters!)
|
||||
consumer_receivers
|
||||
.entry(*consumer_id)
|
||||
.or_default()
|
||||
.push(receivers.pop().unwrap());
|
||||
}
|
||||
}
|
||||
consumer_listeners.insert(*consumer_id, listeners);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -410,14 +411,11 @@ impl DagRunner {
|
||||
};
|
||||
|
||||
if let Some(node) = node {
|
||||
// Take the listeners for this consumer
|
||||
let listeners = consumer_listeners.remove(&node_id).unwrap_or_default();
|
||||
// Take the receivers for this consumer
|
||||
let receivers = consumer_receivers.remove(&node_id).unwrap_or_default();
|
||||
|
||||
// Take the announcer for this producer
|
||||
let announcer = announcers.remove(&node_id).unwrap_or_else(|| {
|
||||
// Create dummy announcer for tasks with no dependents
|
||||
Box::new(()) as Box<dyn std::any::Any + Send>
|
||||
});
|
||||
// Take the senders for this producer
|
||||
let senders = producer_senders.remove(&node_id).unwrap_or_default();
|
||||
|
||||
// Execute inline with panic handling.
|
||||
//
|
||||
@@ -425,7 +423,7 @@ impl DagRunner {
|
||||
// DagError::TaskPanicked, matching the behavior of async runtimes when
|
||||
// they spawn tasks. This guarantees consistent error handling whether
|
||||
// a task executes inline (single-task layer) or spawned (multi-task layer).
|
||||
let result = AssertUnwindSafe(node.execute_with_listeners(listeners, announcer))
|
||||
let result = AssertUnwindSafe(node.execute_with_channels(receivers, senders))
|
||||
.catch_unwind()
|
||||
.await
|
||||
.unwrap_or_else(|panic_payload| {
|
||||
@@ -474,18 +472,15 @@ impl DagRunner {
|
||||
};
|
||||
|
||||
if let Some(node) = node {
|
||||
// Take the listeners for this consumer
|
||||
let listeners = consumer_listeners.remove(&node_id).unwrap_or_default();
|
||||
// Take the receivers for this consumer
|
||||
let receivers = consumer_receivers.remove(&node_id).unwrap_or_default();
|
||||
|
||||
// Take the announcer for this producer
|
||||
let announcer = announcers.remove(&node_id).unwrap_or_else(|| {
|
||||
// Create dummy announcer for tasks with no dependents
|
||||
Box::new(()) as Box<dyn std::any::Any + Send>
|
||||
});
|
||||
// Take the senders for this producer
|
||||
let senders = producer_senders.remove(&node_id).unwrap_or_default();
|
||||
|
||||
// Create a 'static future by taking ownership
|
||||
let inner_future = async move {
|
||||
let task_future = node.execute_with_listeners(listeners, announcer);
|
||||
let task_future = node.execute_with_channels(receivers, senders);
|
||||
|
||||
let result = task_future.await;
|
||||
|
||||
@@ -701,3 +696,6 @@ impl DagRunner {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod coverage_tests;
|
||||
@@ -0,0 +1,249 @@
|
||||
//! Coverage tests for runner module - specifically targeting uncovered error paths
|
||||
//!
|
||||
//! These tests ensure we hit the type mismatch error path in get() (line 601)
|
||||
|
||||
use crate::error::DagError;
|
||||
use crate::runner::DagRunner;
|
||||
use crate::task::Task;
|
||||
use crate::types::TaskHandle;
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_error() {
|
||||
// Test the type mismatch error path in get() - line 601 in runner.rs
|
||||
// This occurs when Arc downcast fails due to type mismatch
|
||||
|
||||
use crate::task_fn;
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Task outputs i32
|
||||
let source = dag.add_task(task_fn(|_: ()| async { 42i32 }));
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as String - TYPE MISMATCH!
|
||||
// Create a handle with the same ID but wrong type
|
||||
let wrong_handle: TaskHandle<String> = TaskHandle {
|
||||
id: source.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
|
||||
assert!(result.is_err());
|
||||
let err = result.unwrap_err();
|
||||
match err {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<String>());
|
||||
assert_eq!(found, "unknown"); // This covers line 601!
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error, got: {:?}", err),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_i32_to_vec() {
|
||||
// Different type mismatch: i32 to Vec<u8>
|
||||
use crate::task_fn;
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Task outputs i32
|
||||
let int_task = dag.add_task(task_fn(|_: ()| async { 123i32 }));
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as Vec<u8>
|
||||
let wrong_handle: TaskHandle<Vec<u8>> = TaskHandle {
|
||||
id: int_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<Vec<u8>>());
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_struct_to_hashmap() {
|
||||
// Complex type mismatch: custom struct to HashMap
|
||||
|
||||
#[derive(Clone)]
|
||||
#[allow(dead_code)]
|
||||
struct CustomOutput {
|
||||
value: i32,
|
||||
}
|
||||
|
||||
struct CustomTask;
|
||||
|
||||
#[crate::task]
|
||||
impl CustomTask {
|
||||
async fn run(&self) -> CustomOutput {
|
||||
CustomOutput { value: 42 }
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let task_handle = dag.add_task(CustomTask);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as HashMap
|
||||
let wrong_handle: TaskHandle<HashMap<String, i32>> = TaskHandle {
|
||||
id: task_handle.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert!(expected.contains("HashMap"));
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_vec_different_types() {
|
||||
// Type mismatch between Vec<i32> and Vec<String>
|
||||
use crate::task_fn;
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Task outputs Vec<i32>
|
||||
let vec_task = dag.add_task(task_fn(|_: ()| async { vec![1, 2, 3] }));
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as Vec<String>
|
||||
let wrong_handle: TaskHandle<Vec<String>> = TaskHandle {
|
||||
id: vec_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<Vec<String>>());
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_option_types() {
|
||||
// Type mismatch with Option types
|
||||
use crate::task_fn;
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Task outputs Option<i32>
|
||||
let opt_task = dag.add_task(task_fn(|_: ()| async { Some(42) }));
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as Option<String>
|
||||
let wrong_handle: TaskHandle<Option<String>> = TaskHandle {
|
||||
id: opt_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<Option<String>>());
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_multiple_type_mismatches_same_dag() {
|
||||
// Test multiple type mismatches in the same DAG
|
||||
use crate::task_fn;
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Create tasks with different output types
|
||||
let int_task = dag.add_task(task_fn(|_: ()| async { 100i32 }));
|
||||
let string_task = dag.add_task(task_fn(|_: ()| async { "hello".to_string() }));
|
||||
let bool_task = dag.add_task(task_fn(|_: ()| async { true }));
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get int_task as String
|
||||
let wrong_int_handle: TaskHandle<String> = TaskHandle {
|
||||
id: int_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
assert!(matches!(
|
||||
dag.get(wrong_int_handle).unwrap_err(),
|
||||
DagError::TypeMismatch { found, .. } if found == "unknown"
|
||||
));
|
||||
|
||||
// Try to get string_task as bool
|
||||
let wrong_string_handle: TaskHandle<bool> = TaskHandle {
|
||||
id: string_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
assert!(matches!(
|
||||
dag.get(wrong_string_handle).unwrap_err(),
|
||||
DagError::TypeMismatch { found, .. } if found == "unknown"
|
||||
));
|
||||
|
||||
// Try to get bool_task as i32
|
||||
let wrong_bool_handle: TaskHandle<i32> = TaskHandle {
|
||||
id: bool_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
assert!(matches!(
|
||||
dag.get(wrong_bool_handle).unwrap_err(),
|
||||
DagError::TypeMismatch { found, .. } if found == "unknown"
|
||||
));
|
||||
|
||||
// Verify correct types still work
|
||||
assert_eq!(dag.get(int_task).unwrap(), 100);
|
||||
assert_eq!(dag.get(string_task).unwrap(), "hello");
|
||||
assert!(dag.get(bool_task).unwrap());
|
||||
}
|
||||
+1
-172
@@ -1,10 +1,8 @@
|
||||
//! Unit tests for runner module
|
||||
|
||||
// Allow tests to reference dagx:: paths (for proc macro generated code)
|
||||
extern crate self as dagx;
|
||||
|
||||
use crate::error::DagError;
|
||||
use crate::runner::DagRunner;
|
||||
use crate::task::Task;
|
||||
use crate::types::TaskHandle;
|
||||
|
||||
// Initialize tracing subscriber for tests (idempotent)
|
||||
@@ -301,172 +299,3 @@ async fn test_task_panic_in_multi_task_layer() {
|
||||
// This test mainly ensures we don't crash
|
||||
assert!(result.is_ok() || result.is_err());
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Coverage tests for DagRunner::get() type mismatch errors
|
||||
// These tests target the error path at src/runner.rs:601-607
|
||||
// ============================================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_i32_to_vec() {
|
||||
init_tracing();
|
||||
// Test type mismatch: i32 to Vec<u8>
|
||||
|
||||
struct IntTask;
|
||||
#[crate::task]
|
||||
impl IntTask {
|
||||
async fn run(&self) -> i32 {
|
||||
123
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let int_task = dag.add_task(IntTask);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as Vec<u8> - TYPE MISMATCH!
|
||||
let wrong_handle: TaskHandle<Vec<u8>> = TaskHandle {
|
||||
id: int_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<Vec<u8>>());
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_struct_to_hashmap() {
|
||||
init_tracing();
|
||||
// Test type mismatch: custom struct to HashMap
|
||||
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[derive(Clone)]
|
||||
struct CustomOutput {
|
||||
#[allow(dead_code)]
|
||||
value: i32,
|
||||
}
|
||||
|
||||
struct CustomTask;
|
||||
#[crate::task]
|
||||
impl CustomTask {
|
||||
async fn run(&self) -> CustomOutput {
|
||||
CustomOutput { value: 42 }
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let task_handle = dag.add_task(CustomTask);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as HashMap - TYPE MISMATCH!
|
||||
let wrong_handle: TaskHandle<HashMap<String, i32>> = TaskHandle {
|
||||
id: task_handle.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert!(expected.contains("HashMap"));
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_vec_different_types() {
|
||||
init_tracing();
|
||||
// Test type mismatch: Vec<i32> to Vec<String>
|
||||
|
||||
struct VecTask;
|
||||
#[crate::task]
|
||||
impl VecTask {
|
||||
async fn run(&self) -> Vec<i32> {
|
||||
vec![1, 2, 3]
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let vec_task = dag.add_task(VecTask);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as Vec<String> - TYPE MISMATCH!
|
||||
let wrong_handle: TaskHandle<Vec<String>> = TaskHandle {
|
||||
id: vec_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<Vec<String>>());
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_type_mismatch_option_types() {
|
||||
init_tracing();
|
||||
// Test type mismatch: Option<i32> to Option<String>
|
||||
|
||||
struct OptionTask;
|
||||
#[crate::task]
|
||||
impl OptionTask {
|
||||
async fn run(&self) -> Option<i32> {
|
||||
Some(42)
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let opt_task = dag.add_task(OptionTask);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Try to get as Option<String> - TYPE MISMATCH!
|
||||
let wrong_handle: TaskHandle<Option<String>> = TaskHandle {
|
||||
id: opt_task.id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
|
||||
let result = dag.get(wrong_handle);
|
||||
assert!(result.is_err());
|
||||
match result.unwrap_err() {
|
||||
DagError::TypeMismatch { expected, found } => {
|
||||
assert_eq!(expected, std::any::type_name::<Option<String>>());
|
||||
assert_eq!(found, "unknown");
|
||||
}
|
||||
_ => panic!("Expected TypeMismatch error"),
|
||||
}
|
||||
}
|
||||
+105
-245
@@ -3,202 +3,111 @@
|
||||
//! This module defines the core Task trait that all DAG nodes must implement,
|
||||
//! along with convenience functions for creating tasks from closures.
|
||||
|
||||
use std::any::Any;
|
||||
use std::future::Future;
|
||||
use std::sync::Arc;
|
||||
|
||||
/// Helper trait for extracting inputs from listeners (internal use only)
|
||||
trait ExtractFromListeners: Sized + Send + Sync + Clone + 'static {
|
||||
fn extract(
|
||||
listeners: Vec<Box<dyn Any + Send>>,
|
||||
) -> impl Future<Output = Result<Self, String>> + Send;
|
||||
}
|
||||
|
||||
// Implementation for unit type (no dependencies)
|
||||
impl ExtractFromListeners for () {
|
||||
async fn extract(_listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
// Implementation for single input types (not tuples)
|
||||
// Note: These now return Arc<T> instead of T for efficiency
|
||||
macro_rules! impl_extract_single {
|
||||
($($t:ty),+) => {
|
||||
$(
|
||||
impl ExtractFromListeners for std::sync::Arc<$t> {
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if listeners.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
|
||||
}
|
||||
let listener = *listeners.pop()
|
||||
.unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<$t>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Listener<Arc<{}>>", std::any::type_name::<$t>()))?;
|
||||
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
|
||||
}
|
||||
}
|
||||
)+
|
||||
};
|
||||
}
|
||||
|
||||
impl_extract_single!(i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String);
|
||||
|
||||
// Implementation for Result types (wrapped in Arc)
|
||||
impl<T, E> ExtractFromListeners for std::sync::Arc<Result<T, E>>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
E: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if listeners.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
|
||||
}
|
||||
let listener = *listeners.pop()
|
||||
.unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<Result<T, E>>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Listener<Arc<Result<{}, {}>>>", std::any::type_name::<T>(), std::any::type_name::<E>()))?;
|
||||
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
// Implementation for Option types (wrapped in Arc)
|
||||
impl<T> ExtractFromListeners for std::sync::Arc<Option<T>>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if listeners.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
|
||||
}
|
||||
let listener = *listeners.pop()
|
||||
.unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<Option<T>>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Listener<Arc<Option<{}>>>", std::any::type_name::<T>()))?;
|
||||
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
// Implementation for Vec types (wrapped in Arc)
|
||||
impl<T> ExtractFromListeners for std::sync::Arc<Vec<T>>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if listeners.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
|
||||
}
|
||||
let listener = *listeners.pop()
|
||||
.unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<Vec<T>>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Listener<Arc<Vec<{}>>>", std::any::type_name::<T>()))?;
|
||||
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
// Implementation for Arc types (Arc<Arc<T>> - useful when T itself is Arc-wrapped by user)
|
||||
impl<T> ExtractFromListeners for std::sync::Arc<std::sync::Arc<T>>
|
||||
where
|
||||
T: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if listeners.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
|
||||
}
|
||||
let listener = *listeners.pop()
|
||||
.unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<std::sync::Arc<T>>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Listener<Arc<Arc<{}>>>", std::any::type_name::<T>()))?;
|
||||
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
// Implementation for HashMap types (wrapped in Arc)
|
||||
impl<K, V> ExtractFromListeners for std::sync::Arc<std::collections::HashMap<K, V>>
|
||||
where
|
||||
K: 'static + Clone + Send + Sync + Eq + std::hash::Hash,
|
||||
V: 'static + Clone + Send + Sync,
|
||||
{
|
||||
async fn extract(mut listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
if listeners.len() != 1 {
|
||||
return Err(format!("Expected 1 dependency, got {}", listeners.len()));
|
||||
}
|
||||
let listener = *listeners.pop()
|
||||
.unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<std::collections::HashMap<K, V>>>>()
|
||||
.map_err(|_| format!("Type mismatch: expected Listener<Arc<HashMap<{}, {}>>>", std::any::type_name::<K>(), std::any::type_name::<V>()))?;
|
||||
listener.listen().await.map_err(|_| "Listener closed before receiving value".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
// Macro to implement ExtractFromListeners for tuple types
|
||||
// Note: Tuples now contain Arc-wrapped elements for efficiency
|
||||
macro_rules! impl_extract_tuple {
|
||||
($($T:ident : $idx:tt),+) => {
|
||||
#[allow(non_snake_case)]
|
||||
impl<$($T),+> ExtractFromListeners for ($(std::sync::Arc<$T>,)+)
|
||||
where
|
||||
$($T: 'static + Clone + Send + Sync),+
|
||||
{
|
||||
async fn extract(listeners: Vec<Box<dyn Any + Send>>) -> Result<Self, String> {
|
||||
let expected_count = [$($idx),+].len();
|
||||
if listeners.len() != expected_count {
|
||||
return Err(format!("Expected {} dependencies, got {}", expected_count, listeners.len()));
|
||||
}
|
||||
|
||||
let mut iter = listeners.into_iter();
|
||||
$(
|
||||
let $T = *iter.next().unwrap()
|
||||
.downcast::<announcement::Listener<std::sync::Arc<$T>>>()
|
||||
.map_err(|_| format!("Type mismatch at index {}: expected Listener<Arc<{}>>", $idx, std::any::type_name::<$T>()))?;
|
||||
)+
|
||||
|
||||
let results = futures::join!(
|
||||
$(
|
||||
async move {
|
||||
$T.listen().await.map_err(|_| format!("Listener {} closed", $idx))
|
||||
}
|
||||
),+
|
||||
);
|
||||
|
||||
Ok(($(
|
||||
results.$idx?
|
||||
,)+))
|
||||
}
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
impl_extract_tuple!(A:0, B:1);
|
||||
impl_extract_tuple!(A:0, B:1, C:2);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6);
|
||||
impl_extract_tuple!(A:0, B:1, C:2, D:3, E:4, F:5, G:6, H:7);
|
||||
|
||||
/// A unit of async work with typed inputs and outputs.
|
||||
///
|
||||
/// **NOTE**: For new code, use the [`task`](crate::task) macro which automatically generates
|
||||
/// an `execute()` method and handles all the wiring. This trait is primarily for internal use
|
||||
/// and backward compatibility with `task_fn`.
|
||||
/// Use the [`task`](crate::task) macro to implement this trait. The macro automatically
|
||||
/// derives `Input` and `Output` types from your `run()` method signature and generates
|
||||
/// type-specific extraction logic in `extract_and_run()`.
|
||||
///
|
||||
/// The `#[task]` macro generates an `execute()` method on your struct that:
|
||||
/// - Extracts inputs from announcement listeners
|
||||
/// - Calls your `run()` method
|
||||
/// - Announces the result to dependents
|
||||
/// - Returns Arc-wrapped output
|
||||
/// **Custom types work automatically!** As long as your output type implements
|
||||
/// `Clone + Send + Sync + 'static`, the macro handles everything. No trait implementations needed!
|
||||
///
|
||||
/// # Task Patterns
|
||||
///
|
||||
/// dagx supports three main task patterns:
|
||||
///
|
||||
/// ## 1. Stateless Tasks (Pure Functions)
|
||||
///
|
||||
/// Tasks with no internal state - just transform inputs to outputs:
|
||||
///
|
||||
/// ```
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Add;
|
||||
///
|
||||
/// #[task]
|
||||
/// impl Add {
|
||||
/// async fn run(a: &i32, b: &i32) -> i32 {
|
||||
/// a + b
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// ## 2. Stateful Tasks with Shared State (`&self`)
|
||||
///
|
||||
/// Tasks that read from internal state but don't modify it:
|
||||
///
|
||||
/// ```
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Multiplier {
|
||||
/// factor: i32,
|
||||
/// }
|
||||
///
|
||||
/// #[task]
|
||||
/// impl Multiplier {
|
||||
/// async fn run(&self, input: &i32) -> i32 {
|
||||
/// input * self.factor
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// ## 3. Stateful Tasks with Mutable State (`&mut self`)
|
||||
///
|
||||
/// Tasks that modify internal state during execution:
|
||||
///
|
||||
/// ```
|
||||
/// use dagx::{task, Task};
|
||||
///
|
||||
/// struct Counter {
|
||||
/// count: i32,
|
||||
/// }
|
||||
///
|
||||
/// #[task]
|
||||
/// impl Counter {
|
||||
/// async fn run(&mut self, increment: &i32) -> i32 {
|
||||
/// self.count += increment;
|
||||
/// self.count
|
||||
/// }
|
||||
/// }
|
||||
/// ```
|
||||
///
|
||||
/// # Input Patterns
|
||||
///
|
||||
/// - `()`: No dependencies (source task)
|
||||
/// - `&T`: Single dependency
|
||||
/// - `&A, &B, &C, ...`: Multiple dependencies (up to 8, order matters)
|
||||
///
|
||||
/// # The Underlying Trait
|
||||
///
|
||||
/// The `#[task]` macro generates an implementation of this trait:
|
||||
///
|
||||
/// # Output Wrapping
|
||||
///
|
||||
/// **Task outputs are automatically wrapped in `Arc<T>` internally** for efficient fan-out.
|
||||
/// You define output as `T`, but the framework wraps it in `Arc<T>` when announcing.
|
||||
/// You define `Output = T`, but the framework wraps it in `Arc<T>` before sending to channels.
|
||||
/// This enables O(1) sharing for 1:N dependencies - Arc is cloned (cheap), not your data.
|
||||
///
|
||||
/// Downstream tasks receive `&T` after ExtractInput extracts from the Arc. This is transparent
|
||||
/// to your task code - you work with `T`, the framework handles Arc wrapping/unwrapping.
|
||||
pub trait Task: Send {
|
||||
type Input: Send;
|
||||
type Output: Send + Sync + Clone; // Sync and Clone required for Arc-wrapping and announcement
|
||||
type Output: Send + Sync; // Sync required for Arc-wrapping and cross-thread sharing
|
||||
|
||||
fn run(self, input: Self::Input) -> impl Future<Output = Self::Output> + Send;
|
||||
|
||||
/// Internal method: Extract input from type-erased channels and execute the task.
|
||||
///
|
||||
/// This method is implemented automatically by the #[task] macro with inline
|
||||
/// extraction logic based on the task's parameter signature. This allows ANY
|
||||
/// type to work without requiring trait implementations.
|
||||
#[doc(hidden)]
|
||||
fn extract_and_run(
|
||||
self,
|
||||
receivers: Vec<Box<dyn std::any::Any + Send>>,
|
||||
) -> impl Future<Output = Result<Self::Output, String>> + Send;
|
||||
}
|
||||
|
||||
/// Convenience function to create a task from a closure.
|
||||
@@ -206,12 +115,12 @@ pub trait Task: Send {
|
||||
/// **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<I, O, Fut, F>(f: F) -> impl Task<Input = I, Output = O> + crate::node::TaskExecute<Input = I, Output = O>
|
||||
pub fn task_fn<I, O, Fut, F>(f: F) -> impl Task<Input = I, Output = O>
|
||||
where
|
||||
F: FnMut(I) -> Fut + Send + 'static,
|
||||
Fut: Future<Output = O> + Send + 'static,
|
||||
I: ExtractFromListeners,
|
||||
O: Send + Sync + Clone + 'static,
|
||||
I: Send + crate::extract::ExtractInput + 'static,
|
||||
O: Send + Sync + 'static,
|
||||
{
|
||||
struct TaskFn<I, O, Fut, F>
|
||||
where
|
||||
@@ -228,8 +137,8 @@ where
|
||||
where
|
||||
F: FnMut(I) -> Fut + Send,
|
||||
Fut: Future<Output = O> + Send,
|
||||
I: Send,
|
||||
O: Send + Sync + Clone,
|
||||
I: Send + crate::extract::ExtractInput,
|
||||
O: Send + Sync,
|
||||
{
|
||||
type Input = I;
|
||||
type Output = O;
|
||||
@@ -237,66 +146,17 @@ where
|
||||
async fn run(mut self, input: I) -> O {
|
||||
(self.f)(input).await
|
||||
}
|
||||
}
|
||||
|
||||
impl<I, O, Fut, F> crate::node::TaskExecute for TaskFn<I, O, Fut, F>
|
||||
where
|
||||
F: FnMut(I) -> Fut + Send + 'static,
|
||||
Fut: Future<Output = O> + Send + 'static,
|
||||
I: ExtractFromListeners,
|
||||
O: Send + Sync + Clone + 'static,
|
||||
{
|
||||
type Input = I;
|
||||
type Output = O;
|
||||
async fn extract_and_run(
|
||||
self,
|
||||
receivers: Vec<Box<dyn std::any::Any + Send>>,
|
||||
) -> Result<Self::Output, String> {
|
||||
// For task_fn, we use ExtractInput since we don't have macro-generated extraction
|
||||
let input = I::extract_from_channels(receivers)
|
||||
.await
|
||||
.map_err(|e| format!("Failed to extract input: {}", e))?;
|
||||
|
||||
fn execute(
|
||||
mut self,
|
||||
listeners: Vec<Box<dyn std::any::Any + Send>>,
|
||||
announcer: Box<dyn std::any::Any + Send>,
|
||||
) -> std::pin::Pin<
|
||||
Box<
|
||||
dyn Future<Output = Result<std::sync::Arc<dyn std::any::Any + Send + Sync>, String>>
|
||||
+ Send,
|
||||
>,
|
||||
> {
|
||||
Box::pin(async move {
|
||||
use std::sync::Arc;
|
||||
|
||||
// Extract the input using the ExtractFromListeners trait
|
||||
let input = I::extract(listeners).await?;
|
||||
|
||||
// Call the closure
|
||||
let output: O = (self.f)(input).await;
|
||||
|
||||
// Wrap in Arc
|
||||
let output_arc = Arc::new(output);
|
||||
|
||||
// Announce the result (announce the Arc, not a clone of the data)
|
||||
let announcer_typed = announcer
|
||||
.downcast::<announcement::Announcer<Arc<O>>>()
|
||||
.map_err(|_| {
|
||||
"Type mismatch: announcer type does not match task output type".to_string()
|
||||
})?;
|
||||
let _ = announcer_typed.announce(Arc::clone(&output_arc));
|
||||
|
||||
// Return Arc-wrapped output
|
||||
Ok(output_arc as Arc<dyn std::any::Any + Send + Sync>)
|
||||
})
|
||||
}
|
||||
|
||||
fn create_announcement_erased() -> (Box<dyn std::any::Any + Send>, Box<dyn std::any::Any + Send>) {
|
||||
let (announcer, announcement) = announcement::Announcement::<Arc<O>>::new();
|
||||
(
|
||||
Box::new(announcer) as Box<dyn std::any::Any + Send>,
|
||||
Box::new(announcement) as Box<dyn std::any::Any + Send>,
|
||||
)
|
||||
}
|
||||
|
||||
fn create_listener_erased(announcement: &dyn std::any::Any) -> Box<dyn std::any::Any + Send> {
|
||||
let announcement = announcement
|
||||
.downcast_ref::<announcement::Announcement<Arc<O>>>()
|
||||
.expect("Type mismatch: announcement type does not match task output type");
|
||||
Box::new(announcement.listener()) as Box<dyn std::any::Any + Send>
|
||||
Ok(self.run(input).await)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
+126
-1
@@ -1,9 +1,72 @@
|
||||
//! Unit tests for task module
|
||||
|
||||
use crate::task::Task;
|
||||
use crate::task::{task_fn, Task};
|
||||
use std::future::Future;
|
||||
use std::pin::Pin;
|
||||
|
||||
// Test the task_fn function and TaskFn struct
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_creation_and_execution() {
|
||||
// Test line 111 in task.rs - Task implementation for TaskFn
|
||||
|
||||
// Create a simple task function
|
||||
let task = task_fn(|input: i32| async move { input * 2 });
|
||||
|
||||
// Test the Task trait implementation
|
||||
let result = task.run(42).await;
|
||||
assert_eq!(result, 84);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_with_unit_input() {
|
||||
// Test task_fn with unit input
|
||||
let task = task_fn(|_: ()| async { 100 });
|
||||
|
||||
let result = task.run(()).await;
|
||||
assert_eq!(result, 100);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_with_tuple_input() {
|
||||
// Test task_fn with tuple input
|
||||
let task = task_fn(|(a, b): (i32, i32)| async move { a + b });
|
||||
|
||||
let result = task.run((10, 20)).await;
|
||||
assert_eq!(result, 30);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_with_string_input() {
|
||||
// Test task_fn with String input
|
||||
let task = task_fn(|input: String| {
|
||||
let len = input.len();
|
||||
async move { len }
|
||||
});
|
||||
|
||||
let test_string = String::from("hello");
|
||||
let result = task.run(test_string).await;
|
||||
assert_eq!(result, 5);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_with_complex_output() {
|
||||
// Test task_fn with complex output type
|
||||
let task = task_fn(|input: i32| async move { vec![input, input * 2, input * 3] });
|
||||
|
||||
let result = task.run(5).await;
|
||||
assert_eq!(result, vec![5, 10, 15]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_stateful_closure() {
|
||||
// Test that task_fn can capture state
|
||||
let multiplier = 3;
|
||||
let task = task_fn(move |input: i32| async move { input * multiplier });
|
||||
|
||||
let result = task.run(7).await;
|
||||
assert_eq!(result, 21);
|
||||
}
|
||||
|
||||
// Test custom Task implementation
|
||||
struct CustomTask {
|
||||
state: i32,
|
||||
@@ -20,6 +83,17 @@ impl Task for CustomTask {
|
||||
self.state
|
||||
})
|
||||
}
|
||||
|
||||
#[allow(refining_impl_trait)]
|
||||
fn extract_and_run(
|
||||
self,
|
||||
_receivers: Vec<Box<dyn std::any::Any + Send>>,
|
||||
) -> Pin<Box<dyn Future<Output = Result<Self::Output, String>> + Send>> {
|
||||
Box::pin(async move {
|
||||
// This is a test helper - just return an error
|
||||
Err("extract_and_run not implemented for CustomTask (test helper)".to_string())
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -48,4 +122,55 @@ fn assert_task_bounds<T: Task>() {
|
||||
fn test_task_trait_bounds() {
|
||||
// Verify that our test tasks meet the Task trait bounds
|
||||
assert_task_bounds::<CustomTask>();
|
||||
|
||||
// We can't directly test TaskFn because it's a private type and needs a concrete closure type,
|
||||
// but the fact that task_fn works in the tests above proves it's correct
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_single_call() {
|
||||
// Test that task_fn works (can only call once due to ownership)
|
||||
let task = task_fn(|x: i32| async move { x + 1 });
|
||||
|
||||
assert_eq!(task.run(1).await, 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_with_result_type() {
|
||||
// Test task_fn returning Result
|
||||
let task = task_fn(|x: i32| async move {
|
||||
if x > 0 {
|
||||
Ok(x * 2)
|
||||
} else {
|
||||
Err("negative input")
|
||||
}
|
||||
});
|
||||
|
||||
assert_eq!(task.run(5).await, Ok(10));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_with_option_type() {
|
||||
// Test task_fn returning Option
|
||||
let task = task_fn(|x: i32| async move {
|
||||
if x > 0 {
|
||||
Some(x * 2)
|
||||
} else {
|
||||
None
|
||||
}
|
||||
});
|
||||
|
||||
assert_eq!(task.run(5).await, Some(10));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_task_fn_explicit_trait_call() {
|
||||
// Test explicit Task trait method call to cover line 121
|
||||
use crate::task::Task;
|
||||
|
||||
let task = task_fn(|x: i32| async move { x * 3 });
|
||||
|
||||
// Explicitly call the Task trait's run method
|
||||
let result = Task::run(task, 10).await;
|
||||
assert_eq!(result, 30);
|
||||
}
|
||||
+6
-3
@@ -4,7 +4,7 @@
|
||||
//! including node identifiers, task handles, and type-state markers.
|
||||
|
||||
use crate::builder::{IsUnitType, TaskBuilder};
|
||||
use crate::node::TaskExecute;
|
||||
use crate::task::Task;
|
||||
|
||||
/// Opaque node identifier (internal use)
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, PartialOrd, Ord)]
|
||||
@@ -62,7 +62,7 @@ pub struct Pending;
|
||||
|
||||
// TaskBuilder can be converted to a TaskHandle ONLY if Input is () (unit type)
|
||||
// This enforces at compile time that tasks with non-unit inputs MUST call .depends_on()
|
||||
impl<'a, Tk: TaskExecute, Deps> From<TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
|
||||
impl<'a, Tk: Task, Deps> From<TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
|
||||
where
|
||||
Tk::Input: IsUnitType,
|
||||
{
|
||||
@@ -74,7 +74,7 @@ where
|
||||
}
|
||||
}
|
||||
|
||||
impl<'a, Tk: TaskExecute, Deps> From<&TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
|
||||
impl<'a, Tk: Task, Deps> From<&TaskBuilder<'a, Tk, Deps>> for TaskHandle<Tk::Output>
|
||||
where
|
||||
Tk::Input: IsUnitType,
|
||||
{
|
||||
@@ -95,3 +95,6 @@ impl<T> From<&TaskHandle<T>> for TaskHandle<T> {
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests;
|
||||
|
||||
#[cfg(test)]
|
||||
mod coverage_tests;
|
||||
@@ -0,0 +1,283 @@
|
||||
//! Coverage tests for types module - specifically targeting the TaskBuilder to TaskHandle conversion
|
||||
//!
|
||||
//! These tests ensure we hit line 71 in the From<TaskBuilder> implementation
|
||||
|
||||
use crate::types::TaskHandle;
|
||||
use crate::{task, DagRunner, Task};
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_unit_type_taskbuilder_to_handle_conversion() {
|
||||
// Test that explicitly uses the From<TaskBuilder> conversion for unit-type tasks
|
||||
// This covers line 71 in types.rs
|
||||
|
||||
// Unit struct (no fields) - its Input type is ()
|
||||
struct UnitTask;
|
||||
|
||||
#[task]
|
||||
impl UnitTask {
|
||||
async fn run(&self) -> i32 {
|
||||
42
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// This creates a TaskBuilder<UnitTask, Pending>
|
||||
// Since UnitTask has Input = (), it can be converted to TaskHandle
|
||||
let builder = dag.add_task(UnitTask);
|
||||
|
||||
// Explicitly convert to TaskHandle using From implementation
|
||||
// This triggers the From<TaskBuilder> implementation at line 69-74,
|
||||
// specifically covering line 71
|
||||
let handle: TaskHandle<i32> = builder.into();
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Verify the task ran and we can get its output
|
||||
assert_eq!(dag.get(handle).unwrap(), 42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_unit_type_conversion_with_multiple_tasks() {
|
||||
// Test conversion with multiple unit-type tasks
|
||||
|
||||
struct FirstTask;
|
||||
#[task]
|
||||
impl FirstTask {
|
||||
async fn run(&self) -> String {
|
||||
"first".to_string()
|
||||
}
|
||||
}
|
||||
|
||||
struct SecondTask;
|
||||
#[task]
|
||||
impl SecondTask {
|
||||
async fn run(&self) -> i32 {
|
||||
100
|
||||
}
|
||||
}
|
||||
|
||||
struct ThirdTask;
|
||||
#[task]
|
||||
impl ThirdTask {
|
||||
async fn run(&self) -> bool {
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Create builders for all tasks
|
||||
let first_builder = dag.add_task(FirstTask);
|
||||
let second_builder = dag.add_task(SecondTask);
|
||||
let third_builder = dag.add_task(ThirdTask);
|
||||
|
||||
// Convert each builder to handle explicitly
|
||||
let first_handle: TaskHandle<String> = first_builder.into();
|
||||
let second_handle: TaskHandle<i32> = second_builder.into();
|
||||
let third_handle: TaskHandle<bool> = third_builder.into();
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Verify all outputs
|
||||
assert_eq!(dag.get(first_handle).unwrap(), "first");
|
||||
assert_eq!(dag.get(second_handle).unwrap(), 100);
|
||||
assert!(dag.get(third_handle).unwrap());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_unit_struct_with_state() {
|
||||
// Even structs with fields can have unit input type
|
||||
|
||||
struct StatefulTask {
|
||||
value: i32,
|
||||
}
|
||||
|
||||
impl StatefulTask {
|
||||
fn new(value: i32) -> Self {
|
||||
Self { value }
|
||||
}
|
||||
}
|
||||
|
||||
#[task]
|
||||
impl StatefulTask {
|
||||
async fn run(&self) -> i32 {
|
||||
self.value * 2
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Create task with initial state
|
||||
let builder = dag.add_task(StatefulTask::new(21));
|
||||
|
||||
// Explicit conversion - triggers line 71
|
||||
let handle: TaskHandle<i32> = builder.into();
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(dag.get(handle).unwrap(), 42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_taskbuilder_conversion_assignment() {
|
||||
// Test the conversion through assignment (different from .into())
|
||||
|
||||
struct SimpleTask;
|
||||
#[task]
|
||||
impl SimpleTask {
|
||||
async fn run(&self) -> Vec<i32> {
|
||||
vec![1, 2, 3]
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let builder = dag.add_task(SimpleTask);
|
||||
|
||||
// Conversion through type annotation and assignment
|
||||
// This also uses From<TaskBuilder> implementation
|
||||
let handle: TaskHandle<Vec<i32>> = builder.into();
|
||||
let id = handle.id;
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let result = dag.get(handle).unwrap();
|
||||
assert_eq!(result, vec![1, 2, 3]);
|
||||
|
||||
// Verify the ID was preserved correctly through conversion
|
||||
let another_handle = TaskHandle::<Vec<i32>> {
|
||||
id,
|
||||
_phantom: std::marker::PhantomData,
|
||||
};
|
||||
assert_eq!(dag.get(another_handle).unwrap(), vec![1, 2, 3]);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_mixed_unit_and_dependent_tasks() {
|
||||
// Test that unit-type tasks work alongside tasks with dependencies
|
||||
|
||||
struct Source;
|
||||
#[task]
|
||||
impl Source {
|
||||
async fn run(&self) -> i32 {
|
||||
10
|
||||
}
|
||||
}
|
||||
|
||||
struct Doubler;
|
||||
#[task]
|
||||
impl Doubler {
|
||||
async fn run(&self, input: &i32) -> i32 {
|
||||
input * 2
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Source is unit-type, can be converted
|
||||
let source_builder = dag.add_task(Source);
|
||||
let source_handle: TaskHandle<i32> = source_builder.into(); // Covers line 71
|
||||
|
||||
// Doubler needs dependencies, so we use it differently
|
||||
let doubler = dag.add_task(Doubler).depends_on(source_handle);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(dag.get(source_handle).unwrap(), 10);
|
||||
assert_eq!(dag.get(doubler).unwrap(), 20);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_taskbuilder_conversion_without_running() {
|
||||
// Test that conversion works even without running the DAG
|
||||
// This ensures the conversion is purely a type-level operation
|
||||
|
||||
struct TestTask;
|
||||
#[task]
|
||||
impl TestTask {
|
||||
async fn run(&self) -> String {
|
||||
"test".to_string()
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let builder = dag.add_task(TestTask);
|
||||
let builder_id = builder.id;
|
||||
|
||||
// Convert without running the DAG
|
||||
let handle: TaskHandle<String> = builder.into();
|
||||
|
||||
// Verify the ID was transferred correctly
|
||||
assert_eq!(handle.id, builder_id);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_multiple_conversions_preserve_ids() {
|
||||
// Ensure that multiple conversions maintain correct IDs
|
||||
|
||||
struct Task1;
|
||||
#[task]
|
||||
impl Task1 {
|
||||
async fn run(&self) -> i32 {
|
||||
1
|
||||
}
|
||||
}
|
||||
|
||||
struct Task2;
|
||||
#[task]
|
||||
impl Task2 {
|
||||
async fn run(&self) -> i32 {
|
||||
2
|
||||
}
|
||||
}
|
||||
|
||||
struct Task3;
|
||||
#[task]
|
||||
impl Task3 {
|
||||
async fn run(&self) -> i32 {
|
||||
3
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
let builder1 = dag.add_task(Task1);
|
||||
let builder2 = dag.add_task(Task2);
|
||||
let builder3 = dag.add_task(Task3);
|
||||
|
||||
let id1 = builder1.id;
|
||||
let id2 = builder2.id;
|
||||
let id3 = builder3.id;
|
||||
|
||||
// Convert all builders
|
||||
let handle1: TaskHandle<i32> = builder1.into();
|
||||
let handle2: TaskHandle<i32> = builder2.into();
|
||||
let handle3: TaskHandle<i32> = builder3.into();
|
||||
|
||||
// Verify IDs are preserved and sequential
|
||||
assert_eq!(handle1.id, id1);
|
||||
assert_eq!(handle2.id, id2);
|
||||
assert_eq!(handle3.id, id3);
|
||||
assert_eq!(handle1.id.0 + 1, handle2.id.0);
|
||||
assert_eq!(handle2.id.0 + 1, handle3.id.0);
|
||||
}
|
||||
+1
-158
@@ -1,8 +1,6 @@
|
||||
//! Unit tests for types module
|
||||
|
||||
// Allow tests to reference dagx:: paths (for proc macro generated code)
|
||||
extern crate self as dagx;
|
||||
|
||||
use crate::task::Task;
|
||||
use crate::types::{NodeId, TaskHandle};
|
||||
use std::marker::PhantomData;
|
||||
|
||||
@@ -172,158 +170,3 @@ fn test_task_builder_ref_from_conversion() {
|
||||
let handle: TaskHandle<String> = (&builder).into();
|
||||
assert_eq!(handle.id, builder_id);
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Additional TaskBuilder conversion coverage tests
|
||||
// Port from commit sq (sqpzloqrlykp) - src/types/coverage_tests.rs
|
||||
// ============================================================================
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_taskbuilder_conversion_with_multiple_tasks() {
|
||||
// Test conversion with multiple unit-type tasks to ensure IDs are preserved
|
||||
use crate::runner::DagRunner;
|
||||
|
||||
struct FirstTask;
|
||||
#[crate::task]
|
||||
impl FirstTask {
|
||||
async fn run(&self) -> String {
|
||||
"first".to_string()
|
||||
}
|
||||
}
|
||||
|
||||
struct SecondTask;
|
||||
#[crate::task]
|
||||
impl SecondTask {
|
||||
async fn run(&self) -> i32 {
|
||||
100
|
||||
}
|
||||
}
|
||||
|
||||
struct ThirdTask;
|
||||
#[crate::task]
|
||||
impl ThirdTask {
|
||||
async fn run(&self) -> bool {
|
||||
true
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Create builders for all tasks
|
||||
let first_builder = dag.add_task(FirstTask);
|
||||
let second_builder = dag.add_task(SecondTask);
|
||||
let third_builder = dag.add_task(ThirdTask);
|
||||
|
||||
// Save IDs before conversion
|
||||
let first_id = first_builder.id;
|
||||
let second_id = second_builder.id;
|
||||
let third_id = third_builder.id;
|
||||
|
||||
// Convert each builder to handle explicitly
|
||||
let first_handle: TaskHandle<String> = first_builder.into();
|
||||
let second_handle: TaskHandle<i32> = second_builder.into();
|
||||
let third_handle: TaskHandle<bool> = third_builder.into();
|
||||
|
||||
// Verify IDs are preserved through conversion
|
||||
assert_eq!(first_handle.id, first_id);
|
||||
assert_eq!(second_handle.id, second_id);
|
||||
assert_eq!(third_handle.id, third_id);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
// Verify all outputs are correct
|
||||
assert_eq!(dag.get(first_handle).unwrap(), "first");
|
||||
assert_eq!(dag.get(second_handle).unwrap(), 100);
|
||||
assert!(dag.get(third_handle).unwrap());
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_taskbuilder_conversion_with_stateful_task() {
|
||||
// Test that stateful tasks (with fields) can have unit input and convert correctly
|
||||
use crate::runner::DagRunner;
|
||||
|
||||
struct StatefulTask {
|
||||
value: i32,
|
||||
}
|
||||
|
||||
impl StatefulTask {
|
||||
fn new(value: i32) -> Self {
|
||||
Self { value }
|
||||
}
|
||||
}
|
||||
|
||||
#[crate::task]
|
||||
impl StatefulTask {
|
||||
async fn run(&self) -> i32 {
|
||||
self.value * 2
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Create task with initial state
|
||||
let builder = dag.add_task(StatefulTask::new(21));
|
||||
let builder_id = builder.id;
|
||||
|
||||
// Explicit conversion - ensures From<TaskBuilder> works for stateful tasks
|
||||
let handle: TaskHandle<i32> = builder.into();
|
||||
assert_eq!(handle.id, builder_id);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(dag.get(handle).unwrap(), 42);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_taskbuilder_conversion_with_complex_output_types() {
|
||||
// Test conversion works with complex output types (Vec, Option, etc.)
|
||||
use crate::runner::DagRunner;
|
||||
|
||||
struct VecTask;
|
||||
#[crate::task]
|
||||
impl VecTask {
|
||||
async fn run(&self) -> Vec<i32> {
|
||||
vec![1, 2, 3]
|
||||
}
|
||||
}
|
||||
|
||||
struct OptionTask;
|
||||
#[crate::task]
|
||||
impl OptionTask {
|
||||
async fn run(&self) -> Option<String> {
|
||||
Some("value".to_string())
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
|
||||
let vec_builder = dag.add_task(VecTask);
|
||||
let option_builder = dag.add_task(OptionTask);
|
||||
|
||||
let vec_id = vec_builder.id;
|
||||
let option_id = option_builder.id;
|
||||
|
||||
// Convert to handles with complex types
|
||||
let vec_handle: TaskHandle<Vec<i32>> = vec_builder.into();
|
||||
let option_handle: TaskHandle<Option<String>> = option_builder.into();
|
||||
|
||||
assert_eq!(vec_handle.id, vec_id);
|
||||
assert_eq!(option_handle.id, option_id);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(dag.get(vec_handle).unwrap(), vec![1, 2, 3]);
|
||||
assert_eq!(dag.get(option_handle).unwrap(), Some("value".to_string()));
|
||||
}
|
||||
+1
-1
@@ -9,7 +9,7 @@
|
||||
//!
|
||||
//! These tests serve as both validation and documentation for proper Arc usage.
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use std::collections::{HashMap, HashSet};
|
||||
use std::sync::Arc;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
//! Tests for empty DAGs and minimal task configurations
|
||||
|
||||
use crate::common::task_fn;
|
||||
use dagx::{task, DagResult, DagRunner};
|
||||
use dagx::{task, DagResult, DagRunner, Task};
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::sync::Arc;
|
||||
|
||||
|
||||
+1
-1
@@ -7,7 +7,7 @@ pub use dagx::task_fn;
|
||||
// Re-export commonly used test structs
|
||||
#[allow(dead_code)]
|
||||
pub mod tasks {
|
||||
use dagx::task;
|
||||
use dagx::{task, Task};
|
||||
|
||||
// Simple seed task with mutable state
|
||||
pub struct Seed {
|
||||
|
||||
@@ -0,0 +1,137 @@
|
||||
// Test that custom types work without any trait implementations
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct Person {
|
||||
name: String,
|
||||
age: u32,
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct Company {
|
||||
name: String,
|
||||
employees: Vec<Person>,
|
||||
}
|
||||
|
||||
// Source task returning custom type
|
||||
struct FetchPerson;
|
||||
|
||||
#[task]
|
||||
impl FetchPerson {
|
||||
async fn run(&self) -> Person {
|
||||
Person {
|
||||
name: "Alice".to_string(),
|
||||
age: 30,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Task consuming custom type
|
||||
struct ProcessPerson;
|
||||
|
||||
#[task]
|
||||
impl ProcessPerson {
|
||||
async fn run(person: &Person) -> String {
|
||||
format!("{} is {} years old", person.name, person.age)
|
||||
}
|
||||
}
|
||||
|
||||
// Task transforming custom type to another custom type
|
||||
struct PersonToCompany;
|
||||
|
||||
#[task]
|
||||
impl PersonToCompany {
|
||||
async fn run(person: &Person) -> Company {
|
||||
Company {
|
||||
name: format!("{}'s Company", person.name),
|
||||
employees: vec![person.clone()],
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_custom_type_single_dependency() {
|
||||
let dag = DagRunner::new();
|
||||
|
||||
let fetch = dag.add_task(FetchPerson);
|
||||
let process = dag.add_task(ProcessPerson).depends_on(&fetch);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let result = dag.get(process).unwrap();
|
||||
assert_eq!(result, "Alice is 30 years old");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_custom_type_passthrough() {
|
||||
let 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);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let person_result = dag.get(process).unwrap();
|
||||
let company_result = dag.get(transform).unwrap();
|
||||
|
||||
assert_eq!(person_result, "Alice is 30 years old");
|
||||
assert_eq!(company_result.name, "Alice's Company");
|
||||
assert_eq!(company_result.employees.len(), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_custom_type_diamond_pattern() {
|
||||
let dag = DagRunner::new();
|
||||
|
||||
// Source
|
||||
let fetch = dag.add_task(FetchPerson);
|
||||
|
||||
// Two paths from source
|
||||
let path_a = dag.add_task(ProcessPerson).depends_on(&fetch);
|
||||
let path_b = dag.add_task(PersonToCompany).depends_on(&fetch);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let result_a = dag.get(path_a).unwrap();
|
||||
let result_b = dag.get(path_b).unwrap();
|
||||
|
||||
assert_eq!(result_a, "Alice is 30 years old");
|
||||
assert_eq!(result_b.name, "Alice's Company");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_custom_type_fan_out() {
|
||||
let 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);
|
||||
|
||||
dag.run(|fut| {
|
||||
tokio::spawn(fut);
|
||||
})
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let result1 = dag.get(process1).unwrap();
|
||||
let result2 = dag.get(process2).unwrap();
|
||||
let result3 = dag.get(process3).unwrap();
|
||||
|
||||
assert_eq!(result1, "Alice is 30 years old");
|
||||
assert_eq!(result2, "Alice is 30 years old");
|
||||
assert_eq!(result3, "Alice is 30 years old");
|
||||
}
|
||||
@@ -3,7 +3,7 @@
|
||||
//! These tests show that even if you try to create cycles, the Rust compiler
|
||||
//! prevents it through the type-state pattern.
|
||||
|
||||
use dagx::{task, DagRunner, TaskHandle};
|
||||
use dagx::{task, DagRunner, Task, TaskHandle};
|
||||
|
||||
/// Demonstrates that TaskBuilder is consumed by depends_on,
|
||||
/// preventing you from wiring the same task multiple times
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Integration tests for end-to-end functionality
|
||||
|
||||
use dagx::{DagResult, DagRunner};
|
||||
use dagx::{DagResult, DagRunner, Task};
|
||||
use crate::common::task_fn;
|
||||
use std::time::Instant;
|
||||
use tokio::time::{sleep, Duration};
|
||||
|
||||
@@ -309,8 +309,10 @@ async fn test_parallel_branches_with_dependencies() -> DagResult<()> {
|
||||
b2_1_start - b1_1_start
|
||||
};
|
||||
|
||||
// Use a more generous threshold to account for coverage instrumentation overhead
|
||||
// 50ms is enough to prove they're parallel (not sequential) while being robust to overhead
|
||||
assert!(
|
||||
branch_start_diff < Duration::from_millis(10),
|
||||
branch_start_diff < Duration::from_millis(50),
|
||||
"Branches should start in parallel, diff was {:?}",
|
||||
branch_start_diff
|
||||
);
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
//! Tests for timing and speedup verification
|
||||
|
||||
use crate::common::task_fn;
|
||||
use dagx::{DagResult, DagRunner};
|
||||
use dagx::{DagResult, DagRunner, Task};
|
||||
use std::sync::Arc;
|
||||
use std::sync::Mutex;
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
@@ -1,533 +0,0 @@
|
||||
//! Property-based testing for the dagx crate.
|
||||
//! Uses proptest to generate random test cases and verify invariants.
|
||||
|
||||
use dagx::DagRunner;
|
||||
use proptest::prelude::*;
|
||||
use std::collections::HashMap;
|
||||
|
||||
// Re-export task macro for use in tests
|
||||
use dagx::task;
|
||||
|
||||
// Strategy for generating various types
|
||||
fn arb_string() -> impl Strategy<Value = String> {
|
||||
prop::string::string_regex("[a-zA-Z0-9]{0,100}").unwrap()
|
||||
}
|
||||
|
||||
fn arb_vec() -> impl Strategy<Value = Vec<i32>> {
|
||||
prop::collection::vec(any::<i32>(), 0..100)
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Single task execution preserves value
|
||||
///
|
||||
/// Verifies: Output matches input for identity tasks
|
||||
/// Strategy: Random i32 values
|
||||
#[test]
|
||||
fn prop_single_task_preserves_value(value: i32) {
|
||||
struct IdentityTask { val: i32 }
|
||||
#[task]
|
||||
impl IdentityTask {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let task = dag.add_task(IdentityTask { val: value });
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
assert_eq!(dag.get(task).unwrap(), value);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Linear dependency chain maintains computation
|
||||
///
|
||||
/// Verifies: Values flow correctly through chain
|
||||
/// Strategy: Random starting value with fixed operations
|
||||
#[test]
|
||||
fn prop_linear_chain_correctness(start: i32) {
|
||||
struct Source { val: i32 }
|
||||
#[task]
|
||||
impl Source {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct AddOne;
|
||||
#[task]
|
||||
impl AddOne {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(1)
|
||||
}
|
||||
}
|
||||
|
||||
struct Double;
|
||||
#[task]
|
||||
impl Double {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_mul(2)
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(Source { val: start });
|
||||
let add_one = dag.add_task(AddOne).depends_on(&source);
|
||||
let double = dag.add_task(Double).depends_on(&add_one);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let expected = start.wrapping_add(1).wrapping_mul(2);
|
||||
assert_eq!(dag.get(double).unwrap(), expected);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Fan-out delivers same value to all dependents
|
||||
///
|
||||
/// Verifies: All dependent tasks receive identical input
|
||||
/// Strategy: Random source value with multiple computations
|
||||
#[test]
|
||||
fn prop_fan_out_delivers_same_value(value: i32) {
|
||||
struct Source { val: i32 }
|
||||
#[task]
|
||||
impl Source {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct AddOne;
|
||||
#[task]
|
||||
impl AddOne {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(1)
|
||||
}
|
||||
}
|
||||
|
||||
struct AddTwo;
|
||||
#[task]
|
||||
impl AddTwo {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(2)
|
||||
}
|
||||
}
|
||||
|
||||
struct AddThree;
|
||||
#[task]
|
||||
impl AddThree {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_add(3)
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(Source { val: value });
|
||||
let t1 = dag.add_task(AddOne).depends_on(&source);
|
||||
let t2 = dag.add_task(AddTwo).depends_on(&source);
|
||||
let t3 = dag.add_task(AddThree).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
assert_eq!(dag.get(t1).unwrap(), value.wrapping_add(1));
|
||||
assert_eq!(dag.get(t2).unwrap(), value.wrapping_add(2));
|
||||
assert_eq!(dag.get(t3).unwrap(), value.wrapping_add(3));
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Diamond pattern computes correctly
|
||||
///
|
||||
/// Verifies: Fork and join operations maintain correctness
|
||||
/// Strategy: Random base value through diamond computation
|
||||
#[test]
|
||||
fn prop_diamond_pattern_correctness(base: i32) {
|
||||
struct Source { val: i32 }
|
||||
#[task]
|
||||
impl Source {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
struct Left;
|
||||
#[task]
|
||||
impl Left {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_mul(2)
|
||||
}
|
||||
}
|
||||
|
||||
struct Right;
|
||||
#[task]
|
||||
impl Right {
|
||||
async fn run(input: &i32) -> i32 {
|
||||
input.wrapping_mul(3)
|
||||
}
|
||||
}
|
||||
|
||||
struct Merge;
|
||||
#[task]
|
||||
impl Merge {
|
||||
async fn run(left: &i32, right: &i32) -> i32 {
|
||||
left.wrapping_add(*right)
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(Source { val: base });
|
||||
let left = dag.add_task(Left).depends_on(&source);
|
||||
let right = dag.add_task(Right).depends_on(&source);
|
||||
let merge = dag.add_task(Merge).depends_on((&left, &right));
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let expected = base.wrapping_mul(2).wrapping_add(base.wrapping_mul(3));
|
||||
assert_eq!(dag.get(merge).unwrap(), expected);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: String data preserved through DAG
|
||||
///
|
||||
/// Verifies: Non-trivial types flow correctly
|
||||
/// Strategy: Random strings with transformations
|
||||
#[test]
|
||||
fn prop_string_data_preserved(s in arb_string()) {
|
||||
struct StringSource { val: String }
|
||||
#[task]
|
||||
impl StringSource {
|
||||
async fn run(&self) -> String { self.val.clone() }
|
||||
}
|
||||
|
||||
struct Uppercase;
|
||||
#[task]
|
||||
impl Uppercase {
|
||||
async fn run(input: &String) -> String {
|
||||
input.to_uppercase()
|
||||
}
|
||||
}
|
||||
|
||||
struct Lowercase;
|
||||
#[task]
|
||||
impl Lowercase {
|
||||
async fn run(input: &String) -> String {
|
||||
input.to_lowercase()
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(StringSource { val: s.clone() });
|
||||
let upper = dag.add_task(Uppercase).depends_on(&source);
|
||||
let lower = dag.add_task(Lowercase).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
assert_eq!(dag.get(upper).unwrap(), s.to_uppercase());
|
||||
assert_eq!(dag.get(lower).unwrap(), s.to_lowercase());
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Vec operations maintain length and contents
|
||||
///
|
||||
/// Verifies: Collection types work correctly in DAG
|
||||
/// Strategy: Random Vec<i32> with transformations
|
||||
#[test]
|
||||
fn prop_vec_operations_correct(vec in arb_vec()) {
|
||||
struct VecSource { data: Vec<i32> }
|
||||
#[task]
|
||||
impl VecSource {
|
||||
async fn run(&self) -> Vec<i32> { self.data.clone() }
|
||||
}
|
||||
|
||||
struct Sum;
|
||||
#[task]
|
||||
impl Sum {
|
||||
async fn run(input: &Vec<i32>) -> i32 {
|
||||
input.iter().fold(0i32, |acc, &x| acc.wrapping_add(x))
|
||||
}
|
||||
}
|
||||
|
||||
struct Len;
|
||||
#[task]
|
||||
impl Len {
|
||||
async fn run(input: &Vec<i32>) -> usize {
|
||||
input.len()
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(VecSource { data: vec.clone() });
|
||||
let sum = dag.add_task(Sum).depends_on(&source);
|
||||
let len = dag.add_task(Len).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let expected_sum = vec.iter().fold(0i32, |acc, &x| acc.wrapping_add(x));
|
||||
assert_eq!(dag.get(sum).unwrap(), expected_sum);
|
||||
assert_eq!(dag.get(len).unwrap(), vec.len());
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Option<T> values preserved correctly
|
||||
///
|
||||
/// Verifies: Both Some and None variants handled properly
|
||||
/// Strategy: Random Option<i32> values
|
||||
#[test]
|
||||
fn prop_option_handling(opt in prop::option::of(any::<i32>())) {
|
||||
struct OptionSource { val: Option<i32> }
|
||||
#[task]
|
||||
impl OptionSource {
|
||||
async fn run(&self) -> Option<i32> { self.val }
|
||||
}
|
||||
|
||||
struct MapOption;
|
||||
#[task]
|
||||
impl MapOption {
|
||||
async fn run(input: &Option<i32>) -> Option<i32> {
|
||||
input.map(|x| x.wrapping_mul(2))
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(OptionSource { val: opt });
|
||||
let mapped = dag.add_task(MapOption).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let expected = opt.map(|x| x.wrapping_mul(2));
|
||||
assert_eq!(dag.get(mapped).unwrap(), expected);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Result<T, E> values preserved correctly
|
||||
///
|
||||
/// Verifies: Both Ok and Err variants handled properly
|
||||
/// Strategy: Random Result<i32, String> values
|
||||
#[test]
|
||||
fn prop_result_handling(
|
||||
result in prop::result::maybe_ok(any::<i32>(), arb_string())
|
||||
) {
|
||||
struct ResultSource { val: Result<i32, String> }
|
||||
#[task]
|
||||
impl ResultSource {
|
||||
async fn run(&self) -> Result<i32, String> { self.val.clone() }
|
||||
}
|
||||
|
||||
struct MapResult;
|
||||
#[task]
|
||||
impl MapResult {
|
||||
async fn run(input: &Result<i32, String>) -> Result<i32, String> {
|
||||
input.as_ref().map(|&x| x.wrapping_add(1)).map_err(|e| e.clone())
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(ResultSource { val: result.clone() });
|
||||
let mapped = dag.add_task(MapResult).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let expected = result.map(|x| x.wrapping_add(1));
|
||||
assert_eq!(dag.get(mapped).unwrap(), expected);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: HashMap structure preserved through DAG
|
||||
///
|
||||
/// Verifies: Complex types maintain structure
|
||||
/// Strategy: Random HashMap with string keys and i32 values
|
||||
#[test]
|
||||
fn prop_hashmap_preservation(
|
||||
data in prop::collection::hash_map(arb_string(), any::<i32>(), 0..20)
|
||||
) {
|
||||
struct MapSource { data: HashMap<String, i32> }
|
||||
#[task]
|
||||
impl MapSource {
|
||||
async fn run(&self) -> HashMap<String, i32> { self.data.clone() }
|
||||
}
|
||||
|
||||
struct MapLen;
|
||||
#[task]
|
||||
impl MapLen {
|
||||
async fn run(input: &HashMap<String, i32>) -> usize {
|
||||
input.len()
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(MapSource { data: data.clone() });
|
||||
let len = dag.add_task(MapLen).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let result_map = dag.get(source).unwrap();
|
||||
assert_eq!(result_map.len(), data.len());
|
||||
for (k, v) in &data {
|
||||
assert_eq!(result_map.get(k), Some(v));
|
||||
}
|
||||
|
||||
assert_eq!(dag.get(len).unwrap(), data.len());
|
||||
}
|
||||
}
|
||||
|
||||
// Custom struct for testing
|
||||
#[derive(Clone, Debug, PartialEq)]
|
||||
struct CustomData {
|
||||
id: i32,
|
||||
name: String,
|
||||
values: Vec<i32>,
|
||||
}
|
||||
|
||||
// Strategy for generating CustomData
|
||||
fn arb_custom_data() -> impl Strategy<Value = CustomData> {
|
||||
(any::<i32>(), arb_string(), arb_vec())
|
||||
.prop_map(|(id, name, values)| CustomData { id, name, values })
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Custom structs preserved through DAG
|
||||
///
|
||||
/// Verifies: User-defined types work correctly
|
||||
/// Strategy: Random custom struct instances
|
||||
#[test]
|
||||
fn prop_custom_struct_preservation(custom in arb_custom_data()) {
|
||||
struct CustomSource { data: CustomData }
|
||||
#[task]
|
||||
impl CustomSource {
|
||||
async fn run(&self) -> CustomData { self.data.clone() }
|
||||
}
|
||||
|
||||
struct ExtractId;
|
||||
#[task]
|
||||
impl ExtractId {
|
||||
async fn run(input: &CustomData) -> i32 {
|
||||
input.id
|
||||
}
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let source = dag.add_task(CustomSource { data: custom.clone() });
|
||||
let id = dag.add_task(ExtractId).depends_on(&source);
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
let result = dag.get(source).unwrap();
|
||||
assert_eq!(result, custom);
|
||||
assert_eq!(dag.get(id).unwrap(), custom.id);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(256))]
|
||||
|
||||
/// Property: Multiple independent DAGs are isolated
|
||||
///
|
||||
/// Verifies: DAG instances don't interfere with each other
|
||||
/// Strategy: Two random values in separate DAGs
|
||||
#[test]
|
||||
fn prop_dag_isolation(value1: i32, value2: i32) {
|
||||
struct Source { val: i32 }
|
||||
#[task]
|
||||
impl Source {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
let dag1 = DagRunner::new();
|
||||
let dag2 = DagRunner::new();
|
||||
|
||||
let task1 = dag1.add_task(Source { val: value1 });
|
||||
let task2 = dag2.add_task(Source { val: value2 });
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag1.run(|fut| { tokio::spawn(fut); }).await.unwrap();
|
||||
dag2.run(|fut| { tokio::spawn(fut); }).await.unwrap();
|
||||
});
|
||||
|
||||
assert_eq!(dag1.get(task1).unwrap(), value1);
|
||||
assert_eq!(dag2.get(task2).unwrap(), value2);
|
||||
}
|
||||
}
|
||||
|
||||
proptest! {
|
||||
#![proptest_config(ProptestConfig::with_cases(128))]
|
||||
|
||||
/// Property: N independent tasks all execute correctly
|
||||
///
|
||||
/// Verifies: Concurrent execution scales correctly
|
||||
/// Strategy: Variable task count (1-20) with unique values
|
||||
#[test]
|
||||
fn prop_concurrent_independent_tasks(count in 1usize..20) {
|
||||
struct IndependentTask { val: i32 }
|
||||
#[task]
|
||||
impl IndependentTask {
|
||||
async fn run(&self) -> i32 { self.val }
|
||||
}
|
||||
|
||||
let dag = DagRunner::new();
|
||||
let mut tasks = Vec::new();
|
||||
|
||||
for i in 0..count {
|
||||
let task = dag.add_task(IndependentTask { val: i as i32 });
|
||||
tasks.push((task, i as i32));
|
||||
}
|
||||
|
||||
let rt = tokio::runtime::Runtime::new().unwrap();
|
||||
rt.block_on(async {
|
||||
dag.run(|fut| { tokio::spawn(fut); }).await
|
||||
}).unwrap();
|
||||
|
||||
for (handle, expected) in tasks {
|
||||
assert_eq!(dag.get(handle).unwrap(), expected);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Runtime compatibility tests for async-std
|
||||
|
||||
use dagx::DagRunner;
|
||||
use dagx::{DagRunner, Task};
|
||||
use crate::common::task_fn;
|
||||
|
||||
struct Value(i32);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Runtime integration tests - verify dagx works correctly across different runtimes
|
||||
|
||||
use dagx::{DagResult, DagRunner};
|
||||
use dagx::{DagResult, DagRunner, Task};
|
||||
|
||||
struct Value(i32);
|
||||
#[dagx::task]
|
||||
|
||||
@@ -486,7 +486,7 @@ fn test_futures_executor_multi_pool() {
|
||||
|
||||
#[test]
|
||||
fn test_sync_tasks_on_async_executor() {
|
||||
|
||||
use dagx::Task;
|
||||
|
||||
struct SyncAdd;
|
||||
|
||||
@@ -519,7 +519,7 @@ fn test_sync_tasks_on_async_executor() {
|
||||
|
||||
#[test]
|
||||
fn test_sync_tasks_on_pollster() {
|
||||
|
||||
use dagx::Task;
|
||||
|
||||
struct SyncDouble;
|
||||
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Runtime compatibility tests for smol
|
||||
|
||||
use dagx::DagRunner;
|
||||
use dagx::{DagRunner, Task};
|
||||
use crate::common::task_fn;
|
||||
|
||||
struct Value(i32);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Tests with tracing feature enabled
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
use tracing_subscriber::{fmt, EnvFilter};
|
||||
|
||||
struct Value(i32);
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
//! Tests that the library works correctly without tracing feature
|
||||
|
||||
use dagx::{task, DagRunner};
|
||||
use dagx::{task, DagRunner, Task};
|
||||
|
||||
struct Value(i32);
|
||||
|
||||
|
||||
Reference in new issue
Block a user