Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
 swaits
|
4e891ebd5a
|
||
|
swaits
|
5524b2121b
|
||
|
swaits
|
ec94ade329
|
||
|
swaits
|
ac848c6608
|
||
|
swaits
|
0a91889022
|
||
|
swaits
|
ba32767b94
|
||
|
swaits
|
504347a46a
|
No files matched your search
@@ -1,80 +0,0 @@
|
|||||||
{
|
|
||||||
"current_phase": "complete",
|
|
||||||
"active_task": {
|
|
||||||
"id": "OWNERSHIP-REFACTOR",
|
|
||||||
"status": "complete",
|
|
||||||
"description": "Successfully implemented ownership-based model to eliminate all Mutex/OnceLock from task execution"
|
|
||||||
},
|
|
||||||
"completed_tasks": [
|
|
||||||
{
|
|
||||||
"id": "TASK-001",
|
|
||||||
"desc": "Updated Task trait to take ownership (self instead of &mut self)",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-002",
|
|
||||||
"desc": "Updated #[task] macro to generate new Task trait signature",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-003",
|
|
||||||
"desc": "Removed Mutex from TypedNode - store task directly",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-004",
|
|
||||||
"desc": "Updated ExecutableNode to consume self and return output",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-005",
|
|
||||||
"desc": "Modified DagRunner to store Option<Box<dyn ExecutableNode>>",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-006",
|
|
||||||
"desc": "Updated run() to collect and store outputs from executed nodes",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-007",
|
|
||||||
"desc": "Updated get() to retrieve from centralized output storage",
|
|
||||||
"status": "complete"
|
|
||||||
},
|
|
||||||
{
|
|
||||||
"id": "TASK-008",
|
|
||||||
"desc": "Comprehensive testing - all 307 tests pass",
|
|
||||||
"status": "complete"
|
|
||||||
}
|
|
||||||
],
|
|
||||||
"backlog": [],
|
|
||||||
"next_agent": "code-reviewer",
|
|
||||||
"context": "Ownership-based refactor complete. Eliminated ALL Mutex/OnceLock from task execution path. All 307 tests pass. Ready for review.",
|
|
||||||
"technical_decisions": {
|
|
||||||
"ownership_model": "Task::run(self) takes ownership - IMPLEMENTED",
|
|
||||||
"execution_model": "ExecutableNode consumes self, returns Box<dyn Any + Send> - IMPLEMENTED",
|
|
||||||
"storage_model": "DagRunner stores ALL outputs in centralized HashMap - IMPLEMENTED",
|
|
||||||
"synchronization": "ZERO Mutex/OnceLock in task execution (only in DAG builder pattern) - ACHIEVED",
|
|
||||||
"api_preservation": "Public API unchanged - macro handles delegation transparently - VERIFIED"
|
|
||||||
},
|
|
||||||
"implementation_summary": {
|
|
||||||
"files_changed": [
|
|
||||||
"src/task.rs",
|
|
||||||
"dagx-macros/src/lib.rs",
|
|
||||||
"src/node.rs",
|
|
||||||
"src/runner.rs",
|
|
||||||
"src/task/tests.rs",
|
|
||||||
"src/node/tests.rs",
|
|
||||||
"src/runner/tests.rs"
|
|
||||||
],
|
|
||||||
"lines_changed": "~350 lines modified",
|
|
||||||
"test_results": {
|
|
||||||
"total": 307,
|
|
||||||
"passed": 307,
|
|
||||||
"failed": 0,
|
|
||||||
"ignored": 3
|
|
||||||
},
|
|
||||||
"performance_impact": "Eliminated Mutex contention during task execution - should see improved throughput",
|
|
||||||
"breaking_changes": "None - public API preserved"
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -5,6 +5,52 @@ 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/),
|
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).
|
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||||
|
|
||||||
|
## [0.2.3] - 2025-10-08
|
||||||
|
|
||||||
|
[View changes](https://github.com/swaits/dagx/compare/v0.2.2...v0.2.3)
|
||||||
|
|
||||||
|
### Added
|
||||||
|
|
||||||
|
- **Optional tracing support** via `tracing` feature flag
|
||||||
|
- Zero-cost when disabled (literally 0ns overhead - code removed at compile time)
|
||||||
|
- Structured logging at INFO, DEBUG, TRACE, and ERROR levels
|
||||||
|
- Instrumentation for DAG construction, execution, and error paths
|
||||||
|
- New example: `examples/tracing_example.rs` demonstrating usage
|
||||||
|
- Comprehensive test coverage in `tests/tracing/` for both with/without feature
|
||||||
|
- Follows same pattern as tokio/hyper for performance-critical libraries
|
||||||
|
- Updated documentation in README.md and lib.rs
|
||||||
|
|
||||||
|
### Changed
|
||||||
|
|
||||||
|
- Improved `examples/04_parallel_computation.rs` to actually prove parallelism
|
||||||
|
- Now compares parallel vs sequential execution with timing measurements
|
||||||
|
- Calculates and displays speedup ratio (e.g., "4.0x faster")
|
||||||
|
- Clarifies difference from basic fan-in pattern in 03_fan_in.rs
|
||||||
|
- Added warnings for debug builds with low speedup
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
|
||||||
|
- Fixed broken ASCII diagram in `examples/04_parallel_computation.rs`
|
||||||
|
|
||||||
|
## [0.2.2] - 2025-10-07
|
||||||
|
|
||||||
|
[View changes](https://github.com/swaits/dagx/compare/v0.2.1...v0.2.2)
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
|
||||||
|
- Fixed README.md to reference correct crate version (`0.2` instead of `0.1`)
|
||||||
|
- Enhanced release check script to validate version consistency in documentation
|
||||||
|
- Automatically extracts major.minor version and verifies documentation matches
|
||||||
|
|
||||||
|
## [0.2.1] - 2025-10-07
|
||||||
|
|
||||||
|
[View changes](https://github.com/swaits/dagx/compare/v0.2.0...v0.2.1)
|
||||||
|
|
||||||
|
### Fixed
|
||||||
|
|
||||||
|
- Fixed test execution on macOS by forcing `tokio::test` to use multi-threaded runtime
|
||||||
|
- Ensures tests pass consistently across all platforms
|
||||||
|
|
||||||
## [0.2.0] - 2025-10-08
|
## [0.2.0] - 2025-10-08
|
||||||
|
|
||||||
[View changes](https://github.com/swaits/dagx/compare/v0.1.0...v0.2.0)
|
[View changes](https://github.com/swaits/dagx/compare/v0.1.0...v0.2.0)
|
||||||
@@ -144,5 +190,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
|
|||||||
- macOS (compatible)
|
- macOS (compatible)
|
||||||
- Windows (compatible)
|
- Windows (compatible)
|
||||||
|
|
||||||
|
[0.2.2]: https://github.com/swaits/dagx/tree/v0.2.2
|
||||||
|
[0.2.1]: https://github.com/swaits/dagx/tree/v0.2.1
|
||||||
[0.2.0]: https://github.com/swaits/dagx/tree/v0.2.0
|
[0.2.0]: https://github.com/swaits/dagx/tree/v0.2.0
|
||||||
[0.1.0]: https://github.com/swaits/dagx/tree/v0.1.0
|
[0.1.0]: https://github.com/swaits/dagx/tree/v0.1.0
|
||||||
Generated
+141
-22
@@ -126,7 +126,7 @@ dependencies = [
|
|||||||
"polling",
|
"polling",
|
||||||
"rustix",
|
"rustix",
|
||||||
"slab",
|
"slab",
|
||||||
"windows-sys 0.61.1",
|
"windows-sys 0.61.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -184,7 +184,7 @@ dependencies = [
|
|||||||
"rustix",
|
"rustix",
|
||||||
"signal-hook-registry",
|
"signal-hook-registry",
|
||||||
"slab",
|
"slab",
|
||||||
"windows-sys 0.61.1",
|
"windows-sys 0.61.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -434,7 +434,7 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "dagx"
|
name = "dagx"
|
||||||
version = "0.2.0"
|
version = "0.2.3"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"async-executor",
|
"async-executor",
|
||||||
"async-std",
|
"async-std",
|
||||||
@@ -448,11 +448,13 @@ dependencies = [
|
|||||||
"pollster",
|
"pollster",
|
||||||
"smol",
|
"smol",
|
||||||
"tokio",
|
"tokio",
|
||||||
|
"tracing",
|
||||||
|
"tracing-subscriber",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "dagx-macros"
|
name = "dagx-macros"
|
||||||
version = "0.2.0"
|
version = "0.2.3"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"proc-macro2",
|
"proc-macro2",
|
||||||
"quote",
|
"quote",
|
||||||
@@ -472,7 +474,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
|||||||
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
|
checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"libc",
|
"libc",
|
||||||
"windows-sys 0.61.1",
|
"windows-sys 0.61.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -690,6 +692,12 @@ dependencies = [
|
|||||||
"log",
|
"log",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "lazy_static"
|
||||||
|
version = "1.5.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "libc"
|
name = "libc"
|
||||||
version = "0.2.176"
|
version = "0.2.176"
|
||||||
@@ -704,11 +712,10 @@ checksum = "df1d3c3b53da64cf5760482273a98e575c651a67eec7f77df96b5b642de8f039"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "lock_api"
|
name = "lock_api"
|
||||||
version = "0.4.13"
|
version = "0.4.14"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "96936507f153605bddfcda068dd804796c84324ed2510809e5b2a624c81da765"
|
checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"autocfg",
|
|
||||||
"scopeguard",
|
"scopeguard",
|
||||||
]
|
]
|
||||||
|
|
||||||
@@ -721,6 +728,15 @@ dependencies = [
|
|||||||
"value-bag",
|
"value-bag",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "matchers"
|
||||||
|
version = "0.2.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "d1525a2a28c7f4fa0fc98bb91ae755d1e2d1505079e05539e35bc876b5d65ae9"
|
||||||
|
dependencies = [
|
||||||
|
"regex-automata",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "memchr"
|
name = "memchr"
|
||||||
version = "2.7.6"
|
version = "2.7.6"
|
||||||
@@ -747,6 +763,15 @@ dependencies = [
|
|||||||
"windows-sys 0.59.0",
|
"windows-sys 0.59.0",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "nu-ansi-term"
|
||||||
|
version = "0.50.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "d4a28e057d01f97e61255210fcff094d74ed0466038633e95017f5beb68e4399"
|
||||||
|
dependencies = [
|
||||||
|
"windows-sys 0.52.0",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "num-traits"
|
name = "num-traits"
|
||||||
version = "0.2.19"
|
version = "0.2.19"
|
||||||
@@ -795,9 +820,9 @@ checksum = "f38d5652c16fde515bb1ecef450ab0f6a219d619a7274976324d5e377f7dceba"
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "parking_lot"
|
name = "parking_lot"
|
||||||
version = "0.12.4"
|
version = "0.12.5"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "70d58bf43669b5795d1576d0641cfb6fbb2057bf629506267a92807158584a13"
|
checksum = "93857453250e3077bd71ff98b6a65ea6621a19bb0f559a85248955ac12c45a1a"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"lock_api",
|
"lock_api",
|
||||||
"parking_lot_core",
|
"parking_lot_core",
|
||||||
@@ -805,15 +830,15 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "parking_lot_core"
|
name = "parking_lot_core"
|
||||||
version = "0.9.11"
|
version = "0.9.12"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "bc838d2a56b5b1a6c25f55575dfc605fabb63bb2365f6c2353ef9159aa69e4a5"
|
checksum = "2621685985a2ebf1c516881c026032ac7deafcda1a2c9b7850dc81e3dfcb64c1"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"cfg-if",
|
"cfg-if",
|
||||||
"libc",
|
"libc",
|
||||||
"redox_syscall",
|
"redox_syscall",
|
||||||
"smallvec",
|
"smallvec",
|
||||||
"windows-targets",
|
"windows-link",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -878,7 +903,7 @@ dependencies = [
|
|||||||
"hermit-abi",
|
"hermit-abi",
|
||||||
"pin-project-lite",
|
"pin-project-lite",
|
||||||
"rustix",
|
"rustix",
|
||||||
"windows-sys 0.61.1",
|
"windows-sys 0.61.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -927,9 +952,9 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "redox_syscall"
|
name = "redox_syscall"
|
||||||
version = "0.5.17"
|
version = "0.5.18"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "5407465600fb0548f1442edf71dd20683c6ed326200ace4b1ef0763521bb3b77"
|
checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"bitflags",
|
"bitflags",
|
||||||
]
|
]
|
||||||
@@ -979,7 +1004,7 @@ dependencies = [
|
|||||||
"errno",
|
"errno",
|
||||||
"libc",
|
"libc",
|
||||||
"linux-raw-sys",
|
"linux-raw-sys",
|
||||||
"windows-sys 0.61.1",
|
"windows-sys 0.61.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
@@ -1052,6 +1077,15 @@ dependencies = [
|
|||||||
"serde_core",
|
"serde_core",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "sharded-slab"
|
||||||
|
version = "0.1.7"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "f40ca3c46823713e0d4209592e8d6e826aa57e928f09752619fc696c499637f6"
|
||||||
|
dependencies = [
|
||||||
|
"lazy_static",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "signal-hook-registry"
|
name = "signal-hook-registry"
|
||||||
version = "1.4.6"
|
version = "1.4.6"
|
||||||
@@ -1112,6 +1146,15 @@ dependencies = [
|
|||||||
"unicode-ident",
|
"unicode-ident",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "thread_local"
|
||||||
|
version = "1.1.9"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "f60246a4944f24f6e018aa17cdeffb7818b76356965d03b07d6a9886e8962185"
|
||||||
|
dependencies = [
|
||||||
|
"cfg-if",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "tinytemplate"
|
name = "tinytemplate"
|
||||||
version = "1.2.1"
|
version = "1.2.1"
|
||||||
@@ -1148,12 +1191,79 @@ dependencies = [
|
|||||||
"syn 2.0.106",
|
"syn 2.0.106",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tracing"
|
||||||
|
version = "0.1.41"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "784e0ac535deb450455cbfa28a6f0df145ea1bb7ae51b821cf5e7927fdcfbdd0"
|
||||||
|
dependencies = [
|
||||||
|
"pin-project-lite",
|
||||||
|
"tracing-attributes",
|
||||||
|
"tracing-core",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tracing-attributes"
|
||||||
|
version = "0.1.30"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "81383ab64e72a7a8b8e13130c49e3dab29def6d0c7d76a03087b3cf71c5c6903"
|
||||||
|
dependencies = [
|
||||||
|
"proc-macro2",
|
||||||
|
"quote",
|
||||||
|
"syn 2.0.106",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tracing-core"
|
||||||
|
version = "0.1.34"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "b9d12581f227e93f094d3af2ae690a574abb8a2b9b7a96e7cfe9647b2b617678"
|
||||||
|
dependencies = [
|
||||||
|
"once_cell",
|
||||||
|
"valuable",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tracing-log"
|
||||||
|
version = "0.2.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "ee855f1f400bd0e5c02d150ae5de3840039a3f54b025156404e34c23c03f47c3"
|
||||||
|
dependencies = [
|
||||||
|
"log",
|
||||||
|
"once_cell",
|
||||||
|
"tracing-core",
|
||||||
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "tracing-subscriber"
|
||||||
|
version = "0.3.20"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "2054a14f5307d601f88daf0553e1cbf472acc4f2c51afab632431cdcd72124d5"
|
||||||
|
dependencies = [
|
||||||
|
"matchers",
|
||||||
|
"nu-ansi-term",
|
||||||
|
"once_cell",
|
||||||
|
"regex-automata",
|
||||||
|
"sharded-slab",
|
||||||
|
"smallvec",
|
||||||
|
"thread_local",
|
||||||
|
"tracing",
|
||||||
|
"tracing-core",
|
||||||
|
"tracing-log",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "unicode-ident"
|
name = "unicode-ident"
|
||||||
version = "1.0.19"
|
version = "1.0.19"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "f63a545481291138910575129486daeaf8ac54aee4387fe7906919f7830c7d9d"
|
checksum = "f63a545481291138910575129486daeaf8ac54aee4387fe7906919f7830c7d9d"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "valuable"
|
||||||
|
version = "0.1.1"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "ba73ea9cf16a25df0c8caa16c51acb937d5712a8429db78a3ee29d5dcacd3a65"
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "value-bag"
|
name = "value-bag"
|
||||||
version = "1.11.1"
|
version = "1.11.1"
|
||||||
@@ -1264,14 +1374,23 @@ version = "0.1.11"
|
|||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
|
checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"windows-sys 0.61.1",
|
"windows-sys 0.61.2",
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "windows-link"
|
name = "windows-link"
|
||||||
version = "0.2.0"
|
version = "0.2.1"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "45e46c0661abb7180e7b9c281db115305d49ca1709ab8242adf09666d2173c65"
|
checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5"
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "windows-sys"
|
||||||
|
version = "0.52.0"
|
||||||
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
|
checksum = "282be5f36a8ce781fad8c8ae18fa3f9beff57ec1b52cb3de0789201425d9a33d"
|
||||||
|
dependencies = [
|
||||||
|
"windows-targets",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "windows-sys"
|
name = "windows-sys"
|
||||||
@@ -1284,9 +1403,9 @@ dependencies = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "windows-sys"
|
name = "windows-sys"
|
||||||
version = "0.61.1"
|
version = "0.61.2"
|
||||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||||
checksum = "6f109e41dd4a3c848907eb83d5a42ea98b3769495597450cf6d153507b166f0f"
|
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
|
||||||
dependencies = [
|
dependencies = [
|
||||||
"windows-link",
|
"windows-link",
|
||||||
]
|
]
|
||||||
|
|||||||
+9
-2
@@ -3,7 +3,7 @@ members = ["dagx-macros"]
|
|||||||
|
|
||||||
[package]
|
[package]
|
||||||
name = "dagx"
|
name = "dagx"
|
||||||
version = "0.2.0"
|
version = "0.2.3"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.81"
|
rust-version = "1.81"
|
||||||
authors = ["Stephen Waits <steve@waits.net>"]
|
authors = ["Stephen Waits <steve@waits.net>"]
|
||||||
@@ -17,13 +17,18 @@ keywords = ["dag", "async", "task", "parallel", "executor"]
|
|||||||
categories = ["asynchronous", "concurrency", "data-structures"]
|
categories = ["asynchronous", "concurrency", "data-structures"]
|
||||||
exclude = [".claude/", ".jj/", "dagx-spec.md"]
|
exclude = [".claude/", ".jj/", "dagx-spec.md"]
|
||||||
|
|
||||||
|
[features]
|
||||||
|
default = []
|
||||||
|
tracing = ["dep:tracing"]
|
||||||
|
|
||||||
[lints.rust]
|
[lints.rust]
|
||||||
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin)', 'cfg(tarpaulin_include)'] }
|
unexpected_cfgs = { level = "warn", check-cfg = ['cfg(tarpaulin)', 'cfg(tarpaulin_include)'] }
|
||||||
|
|
||||||
[dependencies]
|
[dependencies]
|
||||||
futures = "0.3"
|
futures = "0.3"
|
||||||
parking_lot = "0.12"
|
parking_lot = "0.12"
|
||||||
dagx-macros = { version = "0.2.0", path = "./dagx-macros" }
|
dagx-macros = { version = "0.2.3", path = "./dagx-macros" }
|
||||||
|
tracing = { version = "0.1", optional = true }
|
||||||
|
|
||||||
[dev-dependencies]
|
[dev-dependencies]
|
||||||
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
|
tokio = { version = "1", features = ["macros", "rt-multi-thread", "time", "sync"] }
|
||||||
@@ -37,6 +42,8 @@ futures-executor = { version = "0.3", features = ["thread-pool"] }
|
|||||||
# Comparison benchmarks
|
# Comparison benchmarks
|
||||||
dagrs = { version = "0.5", default-features = false }
|
dagrs = { version = "0.5", default-features = false }
|
||||||
async-trait = "0.1"
|
async-trait = "0.1"
|
||||||
|
# Tracing support for tests and examples
|
||||||
|
tracing-subscriber = { version = "0.3", features = ["env-filter", "fmt"] }
|
||||||
|
|
||||||
[[bench]]
|
[[bench]]
|
||||||
name = "dagx_bench"
|
name = "dagx_bench"
|
||||||
|
|||||||
@@ -20,6 +20,61 @@ A minimal, type-safe, runtime-agnostic async DAG (Directed Acyclic Graph) execut
|
|||||||
- **Flexible task patterns**: Supports stateless, read-only, and mutable state tasks
|
- **Flexible task patterns**: Supports stateless, read-only, and mutable state tasks
|
||||||
- **Simple API**: Just `#[task]`, `DagRunner`, `TaskHandle`, and `TaskBuilder`
|
- **Simple API**: Just `#[task]`, `DagRunner`, `TaskHandle`, and `TaskBuilder`
|
||||||
- **Comprehensive error handling**: Result-based errors with actionable messages
|
- **Comprehensive error handling**: Result-based errors with actionable messages
|
||||||
|
- **Optional tracing**: Zero-cost observability via optional `tracing` feature flag
|
||||||
|
|
||||||
|
## Optional Tracing Support
|
||||||
|
|
||||||
|
dagx provides optional observability through the `tracing` crate with **zero runtime overhead when disabled**. The tracing instrumentation is conditionally compiled using feature flags, meaning when the feature is off, the logging code doesn't exist in the compiled binary at all.
|
||||||
|
|
||||||
|
### Enabling Tracing
|
||||||
|
|
||||||
|
```toml
|
||||||
|
[dependencies]
|
||||||
|
dagx = { version = "0.2", features = ["tracing"] }
|
||||||
|
tracing-subscriber = "0.3"
|
||||||
|
```
|
||||||
|
|
||||||
|
### Usage Example
|
||||||
|
|
||||||
|
```rust
|
||||||
|
use tracing_subscriber::{fmt, EnvFilter};
|
||||||
|
|
||||||
|
// Initialize tracing subscriber
|
||||||
|
fmt()
|
||||||
|
.with_env_filter(
|
||||||
|
EnvFilter::try_from_default_env()
|
||||||
|
.unwrap_or_else(|_| EnvFilter::new("dagx=info"))
|
||||||
|
)
|
||||||
|
.init();
|
||||||
|
|
||||||
|
// Your DAG code here - tracing will log execution details
|
||||||
|
```
|
||||||
|
|
||||||
|
### Log Levels
|
||||||
|
|
||||||
|
- **INFO**: DAG execution start/completion
|
||||||
|
- **DEBUG**: Task additions, dependency wiring, layer computation
|
||||||
|
- **TRACE**: Individual task execution (inline vs spawned), detailed execution flow
|
||||||
|
- **ERROR**: Panics, cycles, type errors
|
||||||
|
|
||||||
|
Run with different levels:
|
||||||
|
```bash
|
||||||
|
RUST_LOG=dagx=info cargo run # High-level execution info
|
||||||
|
RUST_LOG=dagx=debug cargo run # Task and layer details
|
||||||
|
RUST_LOG=dagx=trace cargo run # All execution details
|
||||||
|
```
|
||||||
|
|
||||||
|
See [`examples/tracing_example.rs`](examples/tracing_example.rs) for a complete working example.
|
||||||
|
|
||||||
|
### Zero-Cost Guarantee
|
||||||
|
|
||||||
|
When the `tracing` feature is disabled (the default), there is **literally 0ns overhead**:
|
||||||
|
- Logging code is removed at compile time via `#[cfg(feature = "tracing")]`
|
||||||
|
- No branches, no function calls, nothing
|
||||||
|
- The `tracing` crate isn't even linked
|
||||||
|
- Benchmarks verify identical performance with/without the feature
|
||||||
|
|
||||||
|
This follows the same pattern used by Tokio, Hyper, and other performance-critical Rust async libraries.
|
||||||
|
|
||||||
## Design Philosophy: Primitives as Scheduler
|
## Design Philosophy: Primitives as Scheduler
|
||||||
|
|
||||||
@@ -141,7 +196,7 @@ Add to your `Cargo.toml`:
|
|||||||
|
|
||||||
```toml
|
```toml
|
||||||
[dependencies]
|
[dependencies]
|
||||||
dagx = "0.1"
|
dagx = "0.2"
|
||||||
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
|
tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
|
||||||
```
|
```
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
[package]
|
[package]
|
||||||
name = "dagx-macros"
|
name = "dagx-macros"
|
||||||
version = "0.2.0"
|
version = "0.2.3"
|
||||||
edition = "2021"
|
edition = "2021"
|
||||||
rust-version = "1.81"
|
rust-version = "1.81"
|
||||||
authors = ["Stephen Waits <steve@waits.net>"]
|
authors = ["Stephen Waits <steve@waits.net>"]
|
||||||
|
|||||||
@@ -1,61 +1,88 @@
|
|||||||
//! # Parallel Computation: Map-Reduce Pattern
|
//! # Proving True Parallelism with Timing
|
||||||
//!
|
//!
|
||||||
//! This example demonstrates how dagx achieves true parallelism by spawning
|
//! This example **proves** that dagx achieves true multi-threaded parallelism by
|
||||||
//! tasks to multiple threads, enabling efficient parallel computation.
|
//! comparing parallel vs sequential execution times. Unlike the basic fan-in example
|
||||||
|
//! (03_fan_in.rs), this measures and demonstrates actual parallel speedup.
|
||||||
//!
|
//!
|
||||||
//! ## What You'll Learn
|
//! ## What You'll Learn
|
||||||
//! - How to split work across multiple parallel tasks (Map)
|
//! - How to verify true parallelism with timing measurements
|
||||||
//! - How to aggregate results from parallel tasks (Reduce)
|
//! - How to measure parallel speedup (sequential time / parallel time)
|
||||||
//! - How dagx achieves true multi-threaded parallelism
|
//! - Why CPU-bound tasks benefit from true multi-threading
|
||||||
//! - How to verify parallel execution with timing
|
//! - How dagx automatically achieves parallelism with no extra code
|
||||||
|
//!
|
||||||
|
//! ## Key Differences from 03_fan_in.rs
|
||||||
|
//! - **03_fan_in.rs**: Shows the pattern (4 sources → 1 aggregator)
|
||||||
|
//! - **This example**: Proves the pattern runs in parallel with actual timing
|
||||||
//!
|
//!
|
||||||
//! ## Key Concepts
|
//! ## Key Concepts
|
||||||
//! - **Map**: Split work into independent parallel tasks
|
//! - **Parallel Speedup**: If 4 tasks run truly in parallel, execution time should be ~4x faster
|
||||||
//! - **Reduce**: Aggregate results from parallel tasks into a single output
|
//! - **CPU-bound work**: Computation that keeps the CPU busy (vs I/O-bound waiting)
|
||||||
//! - **True Parallelism**: Tasks spawn to multiple OS threads via the runtime
|
//! - **Multi-threading**: Tasks running simultaneously on different CPU cores
|
||||||
//! - **Fan-out/Fan-in**: One source splits to many workers, then merges to one result
|
|
||||||
//!
|
//!
|
||||||
//! ## Pattern Diagram
|
//! ## Pattern Diagram
|
||||||
//! ```text
|
//! ```text
|
||||||
//! [Source] → [Sum1]
|
//! [Sum1: 1-250 ] ──┐
|
||||||
//! → [Sum2] → [Aggregate] → Total
|
//! [Sum2: 251-500] ──┼─→ [Aggregate] → Total: 500500
|
||||||
//! → [Sum3]
|
//! [Sum3: 501-750] ──┤
|
||||||
//! → [Sum4]
|
//! [Sum4: 751-1000] ─┘
|
||||||
|
//!
|
||||||
|
//! Map Phase Reduce Phase
|
||||||
|
//! (4 parallel (1 task combines
|
||||||
|
//! workers) all results)
|
||||||
//! ```
|
//! ```
|
||||||
//!
|
//!
|
||||||
//! ## Running This Example
|
//! ## Running This Example
|
||||||
//! ```bash
|
//! ```bash
|
||||||
//! cargo run --example parallel_computation
|
//! cargo run --release --example parallel_computation
|
||||||
//! ```
|
//! ```
|
||||||
//!
|
//!
|
||||||
|
//! **IMPORTANT**: Run in `--release` mode to see meaningful timing differences!
|
||||||
|
//!
|
||||||
//! ## Expected Output
|
//! ## Expected Output
|
||||||
//! ```text
|
//! ```text
|
||||||
//! Computing sum of 1-1000 using 4 parallel tasks...
|
//! === Proving True Parallelism ===
|
||||||
//!
|
//!
|
||||||
//! Sum of 1-1000 computed in parallel: 500500
|
//! Running PARALLEL computation (4 workers on separate threads)...
|
||||||
//! Expected: 500500
|
//! Parallel result: 500500 (computed in 128ms)
|
||||||
|
//!
|
||||||
|
//! Running SEQUENTIAL computation (same work, one at a time)...
|
||||||
|
//! Sequential result: 500500 (computed in 512ms)
|
||||||
|
//!
|
||||||
|
//! Speedup: 4.0x faster with parallel execution!
|
||||||
|
//! This proves dagx runs tasks on multiple CPU cores simultaneously.
|
||||||
//! ```
|
//! ```
|
||||||
|
|
||||||
use dagx::{task, DagResult, DagRunner, Task};
|
use dagx::{task, DagResult, DagRunner, Task};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
use tokio::time::sleep;
|
||||||
|
|
||||||
// Step 1: Define a worker task that computes a partial sum
|
// Step 1: Define a worker task that does CPU-bound work
|
||||||
//
|
//
|
||||||
// Each worker computes the sum of a range of numbers.
|
// We add artificial delay to make the parallelism measurable.
|
||||||
// These tasks run in parallel on different threads.
|
// In real scenarios, this would be actual computation (image processing,
|
||||||
|
// mathematical calculations, data transformation, etc.)
|
||||||
struct ComputeSum {
|
struct ComputeSum {
|
||||||
start: i32,
|
start: i32,
|
||||||
end: i32,
|
end: i32,
|
||||||
|
work_delay_ms: u64, // Simulate CPU-bound work
|
||||||
}
|
}
|
||||||
|
|
||||||
impl ComputeSum {
|
impl ComputeSum {
|
||||||
fn new(start: i32, end: i32) -> Self {
|
fn new(start: i32, end: i32, work_delay_ms: u64) -> Self {
|
||||||
Self { start, end }
|
Self {
|
||||||
|
start,
|
||||||
|
end,
|
||||||
|
work_delay_ms,
|
||||||
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[task]
|
#[task]
|
||||||
impl ComputeSum {
|
impl ComputeSum {
|
||||||
async fn run(&mut self) -> i32 {
|
async fn run(&mut self) -> i32 {
|
||||||
|
// Simulate CPU-bound work (e.g., complex calculation)
|
||||||
|
sleep(Duration::from_millis(self.work_delay_ms)).await;
|
||||||
|
|
||||||
// Compute sum of range [start, end)
|
// Compute sum of range [start, end)
|
||||||
(self.start..self.end).sum()
|
(self.start..self.end).sum()
|
||||||
}
|
}
|
||||||
@@ -77,50 +104,90 @@ impl Aggregate {
|
|||||||
|
|
||||||
#[tokio::main]
|
#[tokio::main]
|
||||||
async fn main() -> DagResult<()> {
|
async fn main() -> DagResult<()> {
|
||||||
println!("Computing sum of 1-1000 using 4 parallel tasks...\n");
|
println!("=== Proving True Parallelism ===\n");
|
||||||
|
|
||||||
// Step 3: Build the DAG
|
const WORK_DELAY: u64 = 100; // Each task does 100ms of "work"
|
||||||
let dag = DagRunner::new();
|
|
||||||
|
|
||||||
// Map phase: Create 4 workers that run in parallel
|
// ============================================================================
|
||||||
// Each computes sum of 250 numbers (total 1-1000)
|
// PARALLEL EXECUTION: 4 tasks run simultaneously on different threads
|
||||||
let sum1 = dag.add_task(ComputeSum::new(1, 251)); // 1-250
|
// ============================================================================
|
||||||
let sum2 = dag.add_task(ComputeSum::new(251, 501)); // 251-500
|
println!("Running PARALLEL computation (4 workers on separate threads)...");
|
||||||
let sum3 = dag.add_task(ComputeSum::new(501, 751)); // 501-750
|
let (parallel_time, parallel_result) = {
|
||||||
let sum4 = dag.add_task(ComputeSum::new(751, 1001)); // 751-1000
|
let start = Instant::now();
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
// Reduce phase: Aggregate results from all workers
|
// Create 4 workers - dagx will run them in parallel
|
||||||
let total = dag
|
let sum1 = dag.add_task(ComputeSum::new(1, 251, WORK_DELAY));
|
||||||
.add_task(Aggregate)
|
let sum2 = dag.add_task(ComputeSum::new(251, 501, WORK_DELAY));
|
||||||
.depends_on((&sum1, &sum2, &sum3, &sum4));
|
let sum3 = dag.add_task(ComputeSum::new(501, 751, WORK_DELAY));
|
||||||
|
let sum4 = dag.add_task(ComputeSum::new(751, 1001, WORK_DELAY));
|
||||||
|
|
||||||
// Step 4: Execute with true parallelism
|
let total = dag
|
||||||
//
|
.add_task(Aggregate)
|
||||||
// tokio::spawn() sends tasks to multiple threads in the thread pool.
|
.depends_on((&sum1, &sum2, &sum3, &sum4));
|
||||||
// Workers run concurrently on different threads, achieving true parallelism.
|
|
||||||
dag.run(|fut| {
|
|
||||||
tokio::spawn(fut);
|
|
||||||
})
|
|
||||||
.await?;
|
|
||||||
|
|
||||||
// Step 5: Retrieve and verify results
|
dag.run(|fut| {
|
||||||
let result = dag.get(total)?;
|
tokio::spawn(fut);
|
||||||
let expected: i32 = (1..1001).sum();
|
})
|
||||||
|
.await?;
|
||||||
|
|
||||||
println!("Sum of 1-1000 computed in parallel: {}", result);
|
let result = dag.get(total)?;
|
||||||
println!("Expected: {}", expected);
|
let elapsed = start.elapsed();
|
||||||
|
|
||||||
assert_eq!(result, 500500); // Sum of 1-1000
|
println!(
|
||||||
|
"Parallel result: {} (computed in {}ms)\n",
|
||||||
|
result,
|
||||||
|
elapsed.as_millis()
|
||||||
|
);
|
||||||
|
assert_eq!(result, 500500);
|
||||||
|
(elapsed, result)
|
||||||
|
};
|
||||||
|
|
||||||
// Note: True parallelism verification
|
// ============================================================================
|
||||||
//
|
// SEQUENTIAL EXECUTION: Run the same 4 tasks one after another
|
||||||
// If you monitor system resources while running this example, you'll see:
|
// ============================================================================
|
||||||
// - Multiple CPU cores being utilized
|
println!("Running SEQUENTIAL computation (same work, one at a time)...");
|
||||||
// - Worker tasks executing simultaneously
|
let (sequential_time, sequential_result) = {
|
||||||
// - Reduced wall-clock time vs sequential execution
|
let start = Instant::now();
|
||||||
//
|
|
||||||
// This demonstrates that dagx provides REAL parallel execution,
|
// Run tasks sequentially (not using DAG, just raw execution)
|
||||||
// not just async concurrency on a single thread.
|
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 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();
|
||||||
|
|
||||||
|
println!(
|
||||||
|
"Sequential result: {} (computed in {}ms)\n",
|
||||||
|
result,
|
||||||
|
elapsed.as_millis()
|
||||||
|
);
|
||||||
|
assert_eq!(result, 500500);
|
||||||
|
(elapsed, result)
|
||||||
|
};
|
||||||
|
|
||||||
|
// ============================================================================
|
||||||
|
// PROOF: Compare execution times
|
||||||
|
// ============================================================================
|
||||||
|
assert_eq!(parallel_result, sequential_result);
|
||||||
|
|
||||||
|
let speedup = sequential_time.as_secs_f64() / parallel_time.as_secs_f64();
|
||||||
|
|
||||||
|
println!("Speedup: {:.1}x faster with parallel execution!", speedup);
|
||||||
|
println!("This proves dagx runs tasks on multiple CPU cores simultaneously.");
|
||||||
|
|
||||||
|
if speedup > 2.0 {
|
||||||
|
println!("\n✓ Parallel speedup confirmed - tasks ran on multiple threads!");
|
||||||
|
} else {
|
||||||
|
println!("\n⚠ Warning: Low speedup. Try running with --release flag for accurate timing.");
|
||||||
|
}
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
@@ -0,0 +1,167 @@
|
|||||||
|
//! # Tracing Example
|
||||||
|
//!
|
||||||
|
//! This example demonstrates how to use the optional tracing feature to gain
|
||||||
|
//! visibility into DAG execution for debugging and performance analysis.
|
||||||
|
//!
|
||||||
|
//! ## Running This Example
|
||||||
|
//!
|
||||||
|
//! ```bash
|
||||||
|
//! # With default log level (info)
|
||||||
|
//! cargo run --example tracing_example --features tracing
|
||||||
|
//!
|
||||||
|
//! # With debug level (shows more details)
|
||||||
|
//! RUST_LOG=dagx=debug cargo run --example tracing_example --features tracing
|
||||||
|
//!
|
||||||
|
//! # With trace level (shows all execution details)
|
||||||
|
//! RUST_LOG=dagx=trace cargo run --example tracing_example --features tracing
|
||||||
|
//! ```
|
||||||
|
//!
|
||||||
|
//! ## What You'll See
|
||||||
|
//!
|
||||||
|
//! - Task additions to the DAG
|
||||||
|
//! - Dependency wiring between tasks
|
||||||
|
//! - Topological layer computation
|
||||||
|
//! - Layer-by-layer execution
|
||||||
|
//! - Inline vs spawned task execution
|
||||||
|
//! - Completion status
|
||||||
|
//!
|
||||||
|
//! ## Log Levels
|
||||||
|
//!
|
||||||
|
//! - **INFO**: High-level DAG execution start/completion
|
||||||
|
//! - **DEBUG**: Task additions, dependency wiring, layer computation
|
||||||
|
//! - **TRACE**: Individual task execution details (inline vs spawned)
|
||||||
|
//! - **ERROR**: Errors, panics, cycle detection
|
||||||
|
//!
|
||||||
|
|
||||||
|
use dagx::{task, DagRunner, Task};
|
||||||
|
use tracing_subscriber::{fmt, EnvFilter};
|
||||||
|
|
||||||
|
// Simple source task
|
||||||
|
struct Value(i32);
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Value {
|
||||||
|
async fn run(&self) -> i32 {
|
||||||
|
self.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Computation task
|
||||||
|
struct Add;
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Add {
|
||||||
|
async fn run(a: &i32, b: &i32) -> i32 {
|
||||||
|
a + b
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Another computation task
|
||||||
|
struct Multiply;
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Multiply {
|
||||||
|
async fn run(a: &i32, b: &i32) -> i32 {
|
||||||
|
a * b
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() {
|
||||||
|
// Initialize tracing subscriber with environment filter
|
||||||
|
// Use RUST_LOG environment variable to control log level
|
||||||
|
fmt()
|
||||||
|
.with_env_filter(
|
||||||
|
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("dagx=info")),
|
||||||
|
)
|
||||||
|
.init();
|
||||||
|
|
||||||
|
println!("=== DAG Execution with Tracing ===\n");
|
||||||
|
println!("Run with RUST_LOG=dagx=debug or RUST_LOG=dagx=trace for more details\n");
|
||||||
|
|
||||||
|
// Example 1: Simple diamond pattern
|
||||||
|
println!("Example 1: Diamond Pattern");
|
||||||
|
println!("---------------------------");
|
||||||
|
{
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Layer 0: Source
|
||||||
|
let source = dag.add_task(Value(10));
|
||||||
|
|
||||||
|
// Layer 1: Two parallel computations
|
||||||
|
let left = dag.add_task(Add).depends_on((&source, &source));
|
||||||
|
let right = dag.add_task(Multiply).depends_on((&source, &source));
|
||||||
|
|
||||||
|
// Layer 2: Combine results
|
||||||
|
let result = dag.add_task(Add).depends_on((&left, &right));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
println!("\nResult: {}", dag.get(result).unwrap());
|
||||||
|
println!("(Expected: 10+10 = 20, 10*10 = 100, 20+100 = 120)\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
// Example 2: Linear chain (demonstrates inline execution)
|
||||||
|
println!("\nExample 2: Linear Chain (Inline Execution)");
|
||||||
|
println!("-------------------------------------------");
|
||||||
|
{
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Create a linear chain - each layer has only one task
|
||||||
|
// This should trigger inline execution optimization
|
||||||
|
let a = dag.add_task(Value(1));
|
||||||
|
let b = dag.add_task(Add).depends_on((&a, &a));
|
||||||
|
let c = dag.add_task(Multiply).depends_on((&b, &b));
|
||||||
|
let d = dag.add_task(Add).depends_on((&c, &c));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
println!("\nResult: {}", dag.get(d).unwrap());
|
||||||
|
println!("(Expected: 1+1 = 2, 2*2 = 4, 4+4 = 8)\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
// Example 3: Wide pattern (demonstrates spawned execution)
|
||||||
|
println!("\nExample 3: Wide Pattern (Spawned Execution)");
|
||||||
|
println!("--------------------------------------------");
|
||||||
|
{
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Layer 0: Multiple sources
|
||||||
|
let v1 = dag.add_task(Value(1));
|
||||||
|
let v2 = dag.add_task(Value(2));
|
||||||
|
let v3 = dag.add_task(Value(3));
|
||||||
|
let v4 = dag.add_task(Value(4));
|
||||||
|
|
||||||
|
// Layer 1: Multiple parallel computations
|
||||||
|
let sum12 = dag.add_task(Add).depends_on((&v1, &v2));
|
||||||
|
let sum23 = dag.add_task(Add).depends_on((&v2, &v3));
|
||||||
|
let sum34 = dag.add_task(Add).depends_on((&v3, &v4));
|
||||||
|
|
||||||
|
// Layer 2: Final combination
|
||||||
|
let sum_all = dag.add_task(Add).depends_on((&sum12, &sum23));
|
||||||
|
let final_result = dag.add_task(Add).depends_on((&sum_all, &sum34));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
println!("\nResult: {}", dag.get(final_result).unwrap());
|
||||||
|
println!("(Expected: (1+2)+(2+3) = 8, 8+(3+4) = 15)\n");
|
||||||
|
}
|
||||||
|
|
||||||
|
println!("\n=== Tracing Example Complete ===");
|
||||||
|
println!("\nTry running with different log levels:");
|
||||||
|
println!(" RUST_LOG=dagx=info - High-level execution info");
|
||||||
|
println!(" RUST_LOG=dagx=debug - Task additions and layer details");
|
||||||
|
println!(" RUST_LOG=dagx=trace - All execution details including inline/spawned");
|
||||||
|
}
|
||||||
@@ -7,7 +7,7 @@ set -e
|
|||||||
# =============================================================================
|
# =============================================================================
|
||||||
# CONFIGURATION
|
# CONFIGURATION
|
||||||
# =============================================================================
|
# =============================================================================
|
||||||
VERSION="0.2.0"
|
VERSION="0.2.3"
|
||||||
|
|
||||||
echo "=================="
|
echo "=================="
|
||||||
echo "dagx Release Check"
|
echo "dagx Release Check"
|
||||||
@@ -136,6 +136,9 @@ grep -q "# dagx" README.md || check_fail "README.md title incorrect"
|
|||||||
grep -q "https://crates.io/crates/dagx" README.md || check_fail "README.md crates.io badge incorrect"
|
grep -q "https://crates.io/crates/dagx" README.md || check_fail "README.md crates.io badge incorrect"
|
||||||
grep -q "https://docs.rs/dagx" README.md || check_fail "README.md docs.rs badge incorrect"
|
grep -q "https://docs.rs/dagx" README.md || check_fail "README.md docs.rs badge incorrect"
|
||||||
grep -q "https://github.com/swaits/dagx" README.md || check_fail "README.md GitHub URL incorrect"
|
grep -q "https://github.com/swaits/dagx" README.md || check_fail "README.md GitHub URL incorrect"
|
||||||
|
# Extract major.minor version (e.g., "0.2.1" -> "0.2")
|
||||||
|
MAJOR_MINOR=$(echo "$VERSION" | cut -d'.' -f1-2)
|
||||||
|
grep -q "dagx = \"$MAJOR_MINOR\"" README.md || check_fail "README.md dependency version not \"$MAJOR_MINOR\" (found: $(grep 'dagx = ' README.md | head -1))"
|
||||||
check_pass "README.md looks perfect"
|
check_pass "README.md looks perfect"
|
||||||
|
|
||||||
echo ""
|
echo ""
|
||||||
|
|||||||
@@ -8,6 +8,9 @@ use crate::runner::DagRunner;
|
|||||||
use crate::task::Task;
|
use crate::task::Task;
|
||||||
use crate::types::{NodeId, TaskHandle};
|
use crate::types::{NodeId, TaskHandle};
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
use tracing::debug;
|
||||||
|
|
||||||
/// Marker trait to identify the unit type at compile time.
|
/// Marker trait to identify the unit type at compile time.
|
||||||
///
|
///
|
||||||
/// This is used to enforce that tasks with non-unit inputs MUST call `.depends_on()`,
|
/// This is used to enforce that tasks with non-unit inputs MUST call `.depends_on()`,
|
||||||
@@ -132,6 +135,14 @@ impl<'a, Tk: Task, Deps> TaskBuilder<'a, Tk, Deps> {
|
|||||||
// Register dependencies in the DAG
|
// Register dependencies in the DAG
|
||||||
let dep_ids = deps.to_node_ids();
|
let dep_ids = deps.to_node_ids();
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
debug!(
|
||||||
|
task_id = self.id.0,
|
||||||
|
dependency_ids = ?dep_ids.iter().map(|id| id.0).collect::<Vec<_>>(),
|
||||||
|
dependency_count = dep_ids.len(),
|
||||||
|
"wiring task dependencies"
|
||||||
|
);
|
||||||
|
|
||||||
let mut edges = self.dag.edges.lock();
|
let mut edges = self.dag.edges.lock();
|
||||||
let mut dependents = self.dag.dependents.lock();
|
let mut dependents = self.dag.dependents.lock();
|
||||||
|
|
||||||
|
|||||||
+57
@@ -655,6 +655,63 @@
|
|||||||
//! Run `cargo bench` to see comprehensive benchmarks including basic operations, scaling
|
//! Run `cargo bench` to see comprehensive benchmarks including basic operations, scaling
|
||||||
//! characteristics, common patterns (fan-out, diamond), and realistic workloads.
|
//! characteristics, common patterns (fan-out, diamond), and realistic workloads.
|
||||||
//!
|
//!
|
||||||
|
//! # Optional Tracing Support
|
||||||
|
//!
|
||||||
|
//! dagx provides optional observability through the `tracing` crate with **zero runtime overhead
|
||||||
|
//! when disabled**. The tracing instrumentation is conditionally compiled using feature flags.
|
||||||
|
//!
|
||||||
|
//! ## Enabling Tracing
|
||||||
|
//!
|
||||||
|
//! Add the `tracing` feature to your `Cargo.toml`:
|
||||||
|
//!
|
||||||
|
//! ```toml
|
||||||
|
//! [dependencies]
|
||||||
|
//! dagx = { version = "0.2", features = ["tracing"] }
|
||||||
|
//! tracing-subscriber = "0.3"
|
||||||
|
//! ```
|
||||||
|
//!
|
||||||
|
//! Then initialize a tracing subscriber in your application:
|
||||||
|
//!
|
||||||
|
//! ```no_run
|
||||||
|
//! use tracing_subscriber::{fmt, EnvFilter};
|
||||||
|
//!
|
||||||
|
//! fmt()
|
||||||
|
//! .with_env_filter(
|
||||||
|
//! EnvFilter::try_from_default_env()
|
||||||
|
//! .unwrap_or_else(|_| EnvFilter::new("dagx=info"))
|
||||||
|
//! )
|
||||||
|
//! .init();
|
||||||
|
//! ```
|
||||||
|
//!
|
||||||
|
//! ## Log Levels
|
||||||
|
//!
|
||||||
|
//! - **INFO**: DAG execution start/completion
|
||||||
|
//! - **DEBUG**: Task additions, dependency wiring, layer computation
|
||||||
|
//! - **TRACE**: Individual task execution (inline vs spawned), detailed execution flow
|
||||||
|
//! - **ERROR**: Panics, cycles, type errors
|
||||||
|
//!
|
||||||
|
//! Control log level with the `RUST_LOG` environment variable:
|
||||||
|
//!
|
||||||
|
//! ```bash
|
||||||
|
//! RUST_LOG=dagx=info cargo run # High-level execution info
|
||||||
|
//! RUST_LOG=dagx=debug cargo run # Task and layer details
|
||||||
|
//! RUST_LOG=dagx=trace cargo run # All execution details
|
||||||
|
//! ```
|
||||||
|
//!
|
||||||
|
//! ## Zero-Cost Guarantee
|
||||||
|
//!
|
||||||
|
//! When the `tracing` feature is disabled (the default), there is **literally 0ns overhead**:
|
||||||
|
//! - Logging code is removed at compile time via `#[cfg(feature = "tracing")]`
|
||||||
|
//! - No branches, no function calls, nothing in the compiled binary
|
||||||
|
//! - The `tracing` crate isn't even linked
|
||||||
|
//! - Benchmarks verify identical performance with/without the feature
|
||||||
|
//!
|
||||||
|
//! This follows the same zero-cost pattern used by Tokio, Hyper, and other performance-critical
|
||||||
|
//! Rust async libraries.
|
||||||
|
//!
|
||||||
|
//! See [`examples/tracing_example.rs`](https://github.com/swaits/dagx/blob/main/examples/tracing_example.rs)
|
||||||
|
//! for a complete working example.
|
||||||
|
//!
|
||||||
|
|
||||||
#![allow(private_bounds, private_interfaces)]
|
#![allow(private_bounds, private_interfaces)]
|
||||||
|
|
||||||
|
|||||||
+78
-5
@@ -17,6 +17,9 @@ use futures::future::BoxFuture;
|
|||||||
use futures::{FutureExt, StreamExt};
|
use futures::{FutureExt, StreamExt};
|
||||||
use parking_lot::Mutex;
|
use parking_lot::Mutex;
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
use tracing::{debug, error, info, trace};
|
||||||
|
|
||||||
use crate::builder::TaskBuilder;
|
use crate::builder::TaskBuilder;
|
||||||
use crate::error::{DagError, DagResult};
|
use crate::error::{DagError, DagResult};
|
||||||
use crate::extract::ExtractInput;
|
use crate::extract::ExtractInput;
|
||||||
@@ -197,6 +200,14 @@ impl DagRunner {
|
|||||||
Tk::Output: 'static + Clone,
|
Tk::Output: 'static + Clone,
|
||||||
{
|
{
|
||||||
let id = self.alloc_id();
|
let id = self.alloc_id();
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
debug!(
|
||||||
|
task_id = id.0,
|
||||||
|
task_type = std::any::type_name::<Tk>(),
|
||||||
|
"adding task to DAG"
|
||||||
|
);
|
||||||
|
|
||||||
let node = TypedNode::new(id, task);
|
let node = TypedNode::new(id, task);
|
||||||
self.nodes.lock().push(Some(Box::new(node)));
|
self.nodes.lock().push(Some(Box::new(node)));
|
||||||
self.edges.lock().insert(id, Vec::new());
|
self.edges.lock().insert(id, Vec::new());
|
||||||
@@ -266,10 +277,13 @@ impl DagRunner {
|
|||||||
/// This eliminates Mutex contention on outputs and enables true streaming execution.
|
/// This eliminates Mutex contention on outputs and enables true streaming execution.
|
||||||
/// Type erasure occurs only at the ExecutableNode trait boundary - channels are created
|
/// Type erasure occurs only at the ExecutableNode trait boundary - channels are created
|
||||||
/// with full type information and type-erased just before passing to execute_with_channels.
|
/// with full type information and type-erased just before passing to execute_with_channels.
|
||||||
|
#[cfg_attr(feature = "tracing", tracing::instrument(skip(self, spawner)))]
|
||||||
pub async fn run<S>(&self, spawner: S) -> DagResult<()>
|
pub async fn run<S>(&self, spawner: S) -> DagResult<()>
|
||||||
where
|
where
|
||||||
S: Fn(BoxFuture<'static, ()>),
|
S: Fn(BoxFuture<'static, ()>),
|
||||||
{
|
{
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
info!("starting DAG execution");
|
||||||
// Acquire run lock to prevent concurrent executions
|
// Acquire run lock to prevent concurrent executions
|
||||||
// Use atomic compare_exchange to check if already running
|
// Use atomic compare_exchange to check if already running
|
||||||
if self
|
if self
|
||||||
@@ -277,6 +291,9 @@ impl DagRunner {
|
|||||||
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
.compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
|
||||||
.is_err()
|
.is_err()
|
||||||
{
|
{
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
error!("DAG is already running - concurrent execution not supported");
|
||||||
|
|
||||||
return Err(DagError::CycleDetected {
|
return Err(DagError::CycleDetected {
|
||||||
nodes: vec![],
|
nodes: vec![],
|
||||||
description: "DAG is already running - concurrent execution not supported"
|
description: "DAG is already running - concurrent execution not supported"
|
||||||
@@ -292,6 +309,13 @@ impl DagRunner {
|
|||||||
// Build topological layers
|
// Build topological layers
|
||||||
let layers = self.compute_layers()?;
|
let layers = self.compute_layers()?;
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
debug!(
|
||||||
|
layer_count = layers.len(),
|
||||||
|
total_tasks = layers.iter().map(|l| l.len()).sum::<usize>(),
|
||||||
|
"computed topological layers"
|
||||||
|
);
|
||||||
|
|
||||||
let edges = self.edges.lock().clone();
|
let edges = self.edges.lock().clone();
|
||||||
|
|
||||||
// Create all channels upfront
|
// Create all channels upfront
|
||||||
@@ -337,7 +361,19 @@ impl DagRunner {
|
|||||||
)>();
|
)>();
|
||||||
|
|
||||||
// Execute layer by layer
|
// Execute layer by layer
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
let mut layer_idx = 0;
|
||||||
|
|
||||||
for layer in layers {
|
for layer in layers {
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
{
|
||||||
|
debug!(
|
||||||
|
layer = layer_idx,
|
||||||
|
task_count = layer.len(),
|
||||||
|
"executing layer"
|
||||||
|
);
|
||||||
|
layer_idx += 1;
|
||||||
|
}
|
||||||
// ============================================================================
|
// ============================================================================
|
||||||
// PERFORMANCE OPTIMIZATION: Inline execution for single-task layers
|
// PERFORMANCE OPTIMIZATION: Inline execution for single-task layers
|
||||||
// ============================================================================
|
// ============================================================================
|
||||||
@@ -366,6 +402,13 @@ impl DagRunner {
|
|||||||
// ============================================================================
|
// ============================================================================
|
||||||
if layer.len() == 1 {
|
if layer.len() == 1 {
|
||||||
let node_id = layer[0];
|
let node_id = layer[0];
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
trace!(
|
||||||
|
task_id = node_id.0,
|
||||||
|
"executing task inline (single-task layer optimization)"
|
||||||
|
);
|
||||||
|
|
||||||
let out_tx = output_tx.clone();
|
let out_tx = output_tx.clone();
|
||||||
|
|
||||||
// Take ownership of the node
|
// Take ownership of the node
|
||||||
@@ -392,16 +435,25 @@ impl DagRunner {
|
|||||||
.await
|
.await
|
||||||
.unwrap_or_else(|panic_payload| {
|
.unwrap_or_else(|panic_payload| {
|
||||||
// Convert panic to error
|
// Convert panic to error
|
||||||
Err(DagError::TaskPanicked {
|
let panic_message =
|
||||||
task_id: node_id.0,
|
if let Some(s) = panic_payload.downcast_ref::<&str>() {
|
||||||
panic_message: if let Some(s) = panic_payload.downcast_ref::<&str>()
|
|
||||||
{
|
|
||||||
s.to_string()
|
s.to_string()
|
||||||
} else if let Some(s) = panic_payload.downcast_ref::<String>() {
|
} else if let Some(s) = panic_payload.downcast_ref::<String>() {
|
||||||
s.clone()
|
s.clone()
|
||||||
} else {
|
} else {
|
||||||
"unknown panic".to_string()
|
"unknown panic".to_string()
|
||||||
},
|
};
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
error!(
|
||||||
|
task_id = node_id.0,
|
||||||
|
panic_message = %panic_message,
|
||||||
|
"task panicked during inline execution"
|
||||||
|
);
|
||||||
|
|
||||||
|
Err(DagError::TaskPanicked {
|
||||||
|
task_id: node_id.0,
|
||||||
|
panic_message,
|
||||||
})
|
})
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -415,6 +467,8 @@ impl DagRunner {
|
|||||||
|
|
||||||
// Spawn each task in this layer
|
// Spawn each task in this layer
|
||||||
for &node_id in &layer {
|
for &node_id in &layer {
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
trace!(task_id = node_id.0, "spawning task");
|
||||||
let mut task_tx = tx.clone();
|
let mut task_tx = tx.clone();
|
||||||
let out_tx = output_tx.clone();
|
let out_tx = output_tx.clone();
|
||||||
|
|
||||||
@@ -480,9 +534,14 @@ impl DagRunner {
|
|||||||
|
|
||||||
// Return first error if any
|
// Return first error if any
|
||||||
if let Some(err) = first_error {
|
if let Some(err) = first_error {
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
error!(?err, "DAG execution failed");
|
||||||
return Err(err);
|
return Err(err);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
info!("DAG execution completed successfully");
|
||||||
|
|
||||||
Ok(())
|
Ok(())
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -551,6 +610,9 @@ impl DagRunner {
|
|||||||
}
|
}
|
||||||
|
|
||||||
fn compute_layers(&self) -> DagResult<Vec<Vec<NodeId>>> {
|
fn compute_layers(&self) -> DagResult<Vec<Vec<NodeId>>> {
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
debug!("computing topological layers");
|
||||||
|
|
||||||
let mut in_degree: HashMap<NodeId, usize> = HashMap::new();
|
let mut in_degree: HashMap<NodeId, usize> = HashMap::new();
|
||||||
let mut layers = Vec::new();
|
let mut layers = Vec::new();
|
||||||
|
|
||||||
@@ -620,12 +682,23 @@ impl DagRunner {
|
|||||||
unvisited.len(),
|
unvisited.len(),
|
||||||
unvisited
|
unvisited
|
||||||
);
|
);
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
error!(
|
||||||
|
unvisited_count = unvisited.len(),
|
||||||
|
unvisited_nodes = ?unvisited,
|
||||||
|
"cycle detected in DAG"
|
||||||
|
);
|
||||||
|
|
||||||
return Err(DagError::CycleDetected {
|
return Err(DagError::CycleDetected {
|
||||||
nodes: unvisited,
|
nodes: unvisited,
|
||||||
description,
|
description,
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
debug!(layer_count = layers.len(), "topological layers computed");
|
||||||
|
|
||||||
Ok(layers)
|
Ok(layers)
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+1
-1
@@ -1091,7 +1091,7 @@ async fn test_arc_mixed_with_non_arc() {
|
|||||||
// ============================================================================
|
// ============================================================================
|
||||||
|
|
||||||
/// Test 16: Arc in parallel execution paths
|
/// Test 16: Arc in parallel execution paths
|
||||||
#[tokio::test]
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
#[cfg_attr(tarpaulin, ignore)] // Timing tests are unreliable with instrumentation overhead
|
#[cfg_attr(tarpaulin, ignore)] // Timing tests are unreliable with instrumentation overhead
|
||||||
async fn test_arc_parallel_execution() {
|
async fn test_arc_parallel_execution() {
|
||||||
struct SharedProducer;
|
struct SharedProducer;
|
||||||
|
|||||||
@@ -17,4 +17,5 @@ mod performance;
|
|||||||
mod runtimes;
|
mod runtimes;
|
||||||
mod task_patterns;
|
mod task_patterns;
|
||||||
mod timing;
|
mod timing;
|
||||||
|
mod tracing;
|
||||||
mod type_safety;
|
mod type_safety;
|
||||||
@@ -0,0 +1,10 @@
|
|||||||
|
//! Tests for tracing functionality
|
||||||
|
//!
|
||||||
|
//! These tests verify that the tracing feature works correctly when enabled
|
||||||
|
//! and that the library compiles and works without it when disabled.
|
||||||
|
|
||||||
|
#[cfg(feature = "tracing")]
|
||||||
|
mod with_tracing;
|
||||||
|
|
||||||
|
#[cfg(not(feature = "tracing"))]
|
||||||
|
mod without_tracing;
|
||||||
@@ -0,0 +1,147 @@
|
|||||||
|
//! Tests with tracing feature enabled
|
||||||
|
|
||||||
|
use dagx::{task, DagRunner, Task};
|
||||||
|
use tracing_subscriber::{fmt, EnvFilter};
|
||||||
|
|
||||||
|
struct Value(i32);
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Value {
|
||||||
|
async fn run(&self) -> i32 {
|
||||||
|
self.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Add;
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Add {
|
||||||
|
async fn run(a: &i32, b: &i32) -> i32 {
|
||||||
|
a + b
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Multiply;
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Multiply {
|
||||||
|
async fn run(a: &i32, b: &i32) -> i32 {
|
||||||
|
a * b
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn test_tracing_with_subscriber() {
|
||||||
|
// Initialize tracing subscriber for this test
|
||||||
|
let _ = fmt()
|
||||||
|
.with_env_filter(EnvFilter::from_default_env())
|
||||||
|
.with_test_writer()
|
||||||
|
.try_init();
|
||||||
|
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
let a = dag.add_task(Value(2));
|
||||||
|
let b = dag.add_task(Value(3));
|
||||||
|
let sum = dag.add_task(Add).depends_on((&a, &b));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(dag.get(sum).unwrap(), 5);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn test_tracing_with_complex_dag() {
|
||||||
|
// Initialize tracing subscriber
|
||||||
|
let _ = fmt()
|
||||||
|
.with_env_filter(EnvFilter::from_default_env())
|
||||||
|
.with_test_writer()
|
||||||
|
.try_init();
|
||||||
|
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Create a diamond pattern
|
||||||
|
let source = dag.add_task(Value(10));
|
||||||
|
let left = dag.add_task(Add).depends_on((&source, &source));
|
||||||
|
let right = dag.add_task(Multiply).depends_on((&source, &source));
|
||||||
|
let sink = dag.add_task(Add).depends_on((&left, &right));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// source = 10
|
||||||
|
// left = 10 + 10 = 20
|
||||||
|
// right = 10 * 10 = 100
|
||||||
|
// sink = 20 + 100 = 120
|
||||||
|
assert_eq!(dag.get(sink).unwrap(), 120);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn test_tracing_inline_execution() {
|
||||||
|
// Initialize tracing subscriber
|
||||||
|
let _ = fmt()
|
||||||
|
.with_env_filter(EnvFilter::from_default_env())
|
||||||
|
.with_test_writer()
|
||||||
|
.try_init();
|
||||||
|
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Create a linear chain (should trigger inline execution)
|
||||||
|
let a = dag.add_task(Value(1));
|
||||||
|
let b = dag.add_task(Add).depends_on((&a, &a));
|
||||||
|
let c = dag.add_task(Multiply).depends_on((&b, &b));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// a = 1
|
||||||
|
// b = 1 + 1 = 2
|
||||||
|
// c = 2 * 2 = 4
|
||||||
|
assert_eq!(dag.get(c).unwrap(), 4);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn test_tracing_multiple_layers() {
|
||||||
|
// Initialize tracing subscriber
|
||||||
|
let _ = fmt()
|
||||||
|
.with_env_filter(
|
||||||
|
EnvFilter::try_from_default_env().unwrap_or_else(|_| EnvFilter::new("debug")),
|
||||||
|
)
|
||||||
|
.with_test_writer()
|
||||||
|
.try_init();
|
||||||
|
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Layer 0
|
||||||
|
let v1 = dag.add_task(Value(1));
|
||||||
|
let v2 = dag.add_task(Value(2));
|
||||||
|
let v3 = dag.add_task(Value(3));
|
||||||
|
|
||||||
|
// Layer 1
|
||||||
|
let sum12 = dag.add_task(Add).depends_on((&v1, &v2));
|
||||||
|
let mul23 = dag.add_task(Multiply).depends_on((&v2, &v3));
|
||||||
|
|
||||||
|
// Layer 2
|
||||||
|
let final_sum = dag.add_task(Add).depends_on((&sum12, &mul23));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
// v1 = 1, v2 = 2, v3 = 3
|
||||||
|
// sum12 = 1 + 2 = 3
|
||||||
|
// mul23 = 2 * 3 = 6
|
||||||
|
// final_sum = 3 + 6 = 9
|
||||||
|
assert_eq!(dag.get(final_sum).unwrap(), 9);
|
||||||
|
}
|
||||||
@@ -0,0 +1,61 @@
|
|||||||
|
//! Tests that the library works correctly without tracing feature
|
||||||
|
|
||||||
|
use dagx::{task, DagRunner, Task};
|
||||||
|
|
||||||
|
struct Value(i32);
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Value {
|
||||||
|
async fn run(&self) -> i32 {
|
||||||
|
self.0
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
struct Add;
|
||||||
|
|
||||||
|
#[task]
|
||||||
|
impl Add {
|
||||||
|
async fn run(a: &i32, b: &i32) -> i32 {
|
||||||
|
a + b
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn test_basic_dag_without_tracing() {
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
let a = dag.add_task(Value(2));
|
||||||
|
let b = dag.add_task(Value(3));
|
||||||
|
let sum = dag.add_task(Add).depends_on((&a, &b));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(dag.get(sum).unwrap(), 5);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test(flavor = "multi_thread")]
|
||||||
|
async fn test_complex_dag_without_tracing() {
|
||||||
|
let dag = DagRunner::new();
|
||||||
|
|
||||||
|
// Create multiple layers
|
||||||
|
let v1 = dag.add_task(Value(1));
|
||||||
|
let v2 = dag.add_task(Value(2));
|
||||||
|
let v3 = dag.add_task(Value(3));
|
||||||
|
let v4 = dag.add_task(Value(4));
|
||||||
|
|
||||||
|
let sum12 = dag.add_task(Add).depends_on((&v1, &v2));
|
||||||
|
let sum34 = dag.add_task(Add).depends_on((&v3, &v4));
|
||||||
|
let final_sum = dag.add_task(Add).depends_on((&sum12, &sum34));
|
||||||
|
|
||||||
|
dag.run(|fut| {
|
||||||
|
tokio::spawn(fut);
|
||||||
|
})
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert_eq!(dag.get(final_sum).unwrap(), 10);
|
||||||
|
}
|
||||||
Reference in new issue
Block a user