135 lines
4.8 KiB
Markdown
135 lines
4.8 KiB
Markdown
# 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
|