WIP: perf: test regression, needs Arc<T>

PERFORMANCE_REGRESSION_ANALYSIS.md

@@ -0,0 +1,134 @@
+# 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

benches/realistic/etl.rs

@@ -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: Vec<String>| async move {
+                            .add_task(task_fn(move |data: std::sync::Arc<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: Vec<String>| async move {
+                            .add_task(task_fn(move |data: std::sync::Arc<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: Vec<String>| async move {
+                            .add_task(task_fn(move |data: std::sync::Arc<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): (usize, Vec<i32>, Vec<String>)| async move {
+                            move |(valid, scores, names): (std::sync::Arc<usize>, std::sync::Arc<Vec<i32>>, std::sync::Arc<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
                                 )

dagx-macros/src/lib.rs

@@ -254,9 +254,9 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
                         ));
                     }
                     let listener = listeners[#i]
-                        .downcast_ref::<announcement::Listener<#ty>>()
+                        .downcast_ref::<announcement::Listener<std::sync::Arc<#ty>>>()
                         .ok_or_else(|| {
-                            format!("Type mismatch: expected Listener<{}> at index {}",
+                            format!("Type mismatch: expected Listener<Arc<{}>> at index {}",
                                 stringify!(#ty), #i)
                         })?;
                     listener.listen().await
@@ -267,7 +267,8 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
         .collect();
 
     // Generate parameter refs for the run call
-    let param_refs: Vec<_> = params.iter().map(|(name, _)| quote! { &#name }).collect();
+    // 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 {
@@ -320,11 +321,11 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
                 // Wrap in Arc for sharing
                 let output_arc = std::sync::Arc::new(output);
 
-                // Announce the result
+                // Announce the result (announce the Arc, not a clone of the data)
                 let announcer_typed = announcer
-                    .downcast::<announcement::Announcer<#output_type>>()
+                    .downcast::<announcement::Announcer<std::sync::Arc<#output_type>>>()
                     .map_err(|_| "Type mismatch: announcer type does not match task output type")?;
-                let _ = announcer_typed.announce((*output_arc).clone());
+                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>)
@@ -349,7 +350,7 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
             }
 
             fn create_announcement_erased() -> (Box<dyn std::any::Any + Send>, Box<dyn std::any::Any + Send>) {
-                let (announcer, announcement) = announcement::Announcement::<#output_type>::new();
+                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>,
@@ -358,7 +359,7 @@ pub fn task(_attr: TokenStream, item: TokenStream) -> TokenStream {
 
             fn create_listener_erased(announcement: &dyn std::any::Any) -> Box<dyn std::any::Any + Send> {
                 let announcement = announcement
-                    .downcast_ref::<announcement::Announcement<#output_type>>()
+                    .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>
             }

src/task.rs

@@ -5,6 +5,7 @@
 
 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 {
@@ -21,18 +22,19 @@ impl ExtractFromListeners for () {
 }
 
 // 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 $t {
+            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<$t>>()
+                        .downcast::<announcement::Listener<std::sync::Arc<$t>>>()
-                        .map_err(|_| format!("Type mismatch: expected Listener<{}>", std::any::type_name::<$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())
                 }
             }
@@ -42,8 +44,8 @@ macro_rules! impl_extract_single {
 
 impl_extract_single!(i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool, char, String);
 
-// Implementation for Result types
+// Implementation for Result types (wrapped in Arc)
-impl<T, E> ExtractFromListeners for Result<T, E>
+impl<T, E> ExtractFromListeners for std::sync::Arc<Result<T, E>>
 where
     T: 'static + Clone + Send + Sync,
     E: 'static + Clone + Send + Sync,
@@ -54,14 +56,14 @@ where
         }
         let listener = *listeners.pop()
             .unwrap()
-            .downcast::<announcement::Listener<Result<T, E>>>()
+            .downcast::<announcement::Listener<std::sync::Arc<Result<T, E>>>>()
-            .map_err(|_| format!("Type mismatch: expected Listener<Result<{}, {}>>", std::any::type_name::<T>(), std::any::type_name::<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
+// Implementation for Option types (wrapped in Arc)
-impl<T> ExtractFromListeners for Option<T>
+impl<T> ExtractFromListeners for std::sync::Arc<Option<T>>
 where
     T: 'static + Clone + Send + Sync,
 {
@@ -71,14 +73,14 @@ where
         }
         let listener = *listeners.pop()
             .unwrap()
-            .downcast::<announcement::Listener<Option<T>>>()
+            .downcast::<announcement::Listener<std::sync::Arc<Option<T>>>>()
-            .map_err(|_| format!("Type mismatch: expected Listener<Option<{}>>", std::any::type_name::<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
+// Implementation for Vec types (wrapped in Arc)
-impl<T> ExtractFromListeners for Vec<T>
+impl<T> ExtractFromListeners for std::sync::Arc<Vec<T>>
 where
     T: 'static + Clone + Send + Sync,
 {
@@ -88,14 +90,14 @@ where
         }
         let listener = *listeners.pop()
             .unwrap()
-            .downcast::<announcement::Listener<Vec<T>>>()
+            .downcast::<announcement::Listener<std::sync::Arc<Vec<T>>>>()
-            .map_err(|_| format!("Type mismatch: expected Listener<Vec<{}>>", std::any::type_name::<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
+// Implementation for Arc types (Arc<Arc<T>> - useful when T itself is Arc-wrapped by user)
-impl<T> ExtractFromListeners for std::sync::Arc<T>
+impl<T> ExtractFromListeners for std::sync::Arc<std::sync::Arc<T>>
 where
     T: 'static + Clone + Send + Sync,
 {
@@ -105,14 +107,14 @@ where
         }
         let listener = *listeners.pop()
             .unwrap()
-            .downcast::<announcement::Listener<std::sync::Arc<T>>>()
+            .downcast::<announcement::Listener<std::sync::Arc<std::sync::Arc<T>>>>()
-            .map_err(|_| format!("Type mismatch: expected Listener<Arc<{}>>", std::any::type_name::<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
+// Implementation for HashMap types (wrapped in Arc)
-impl<K, V> ExtractFromListeners for std::collections::HashMap<K, V>
+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,
@@ -123,17 +125,18 @@ where
         }
         let listener = *listeners.pop()
             .unwrap()
-            .downcast::<announcement::Listener<std::collections::HashMap<K, V>>>()
+            .downcast::<announcement::Listener<std::sync::Arc<std::collections::HashMap<K, V>>>>()
-            .map_err(|_| format!("Type mismatch: expected Listener<HashMap<{}, {}>>", std::any::type_name::<K>(), std::any::type_name::<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 ($($T,)+)
+        impl<$($T),+> ExtractFromListeners for ($(std::sync::Arc<$T>,)+)
         where
             $($T: 'static + Clone + Send + Sync),+
         {
@@ -146,8 +149,8 @@ macro_rules! impl_extract_tuple {
                 let mut iter = listeners.into_iter();
                 $(
                     let $T = *iter.next().unwrap()
-                        .downcast::<announcement::Listener<$T>>()
+                        .downcast::<announcement::Listener<std::sync::Arc<$T>>>()
-                        .map_err(|_| format!("Type mismatch at index {}: expected Listener<{}>", $idx, std::any::type_name::<$T>()))?;
+                        .map_err(|_| format!("Type mismatch at index {}: expected Listener<Arc<{}>>", $idx, std::any::type_name::<$T>()))?;
                 )+
 
                 let results = futures::join!(
@@ -268,13 +271,13 @@ where
                 // Wrap in Arc
                 let output_arc = Arc::new(output);
 
-                // Announce the result
+                // Announce the result (announce the Arc, not a clone of the data)
                 let announcer_typed = announcer
-                    .downcast::<announcement::Announcer<O>>()
+                    .downcast::<announcement::Announcer<Arc<O>>>()
                     .map_err(|_| {
                         "Type mismatch: announcer type does not match task output type".to_string()
                     })?;
-                let _ = announcer_typed.announce((*output_arc).clone());
+                let _ = announcer_typed.announce(Arc::clone(&output_arc));
 
                 // Return Arc-wrapped output
                 Ok(output_arc as Arc<dyn std::any::Any + Send + Sync>)
@@ -282,7 +285,7 @@ where
         }
 
         fn create_announcement_erased() -> (Box<dyn std::any::Any + Send>, Box<dyn std::any::Any + Send>) {
-            let (announcer, announcement) = announcement::Announcement::<O>::new();
+            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>,
@@ -291,7 +294,7 @@ where
 
         fn create_listener_erased(announcement: &dyn std::any::Any) -> Box<dyn std::any::Any + Send> {
             let announcement = announcement
-                .downcast_ref::<announcement::Announcement<O>>()
+                .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>
         }