From 49be497964110d000ec395fb0b220d50098ce9de Mon Sep 17 00:00:00 2001 From: TechnoPorg Date: Thu, 12 Feb 2026 20:49:10 +0100 Subject: [PATCH] perf: don't double-box task outputs --- src/runner.rs | 13 ++++++------- 1 file changed, 6 insertions(+), 7 deletions(-) diff --git a/src/runner.rs b/src/runner.rs index b70f37b..e596300 100644 --- a/src/runner.rs +++ b/src/runner.rs @@ -10,12 +10,12 @@ use std::collections::{HashMap, HashSet, VecDeque}; use std::future::Future; use std::hash::{BuildHasher, Hasher}; use std::panic::AssertUnwindSafe; -use std::pin::Pin; use std::sync::{ atomic::{AtomicBool, Ordering}, Arc, }; +use futures_util::future::BoxFuture; use futures_util::{stream::FuturesUnordered, FutureExt, StreamExt, TryFutureExt}; use parking_lot::Mutex; @@ -28,8 +28,6 @@ use crate::node::{ExecutableNode, TypedNode}; use crate::task::Task; use crate::types::{NodeId, Pending, TaskHandle}; -type BoxFuture<'a, T> = Pin + Send + 'a>>; - // Guard to ensure run_lock is released even on early return or panic struct RunGuard<'a> { lock: &'a AtomicBool, @@ -319,8 +317,8 @@ impl DagRunner { #[cfg_attr(feature = "tracing", tracing::instrument(skip(self, spawner)))] pub async fn run(&self, spawner: S) -> DagResult<()> where - S: Fn(BoxFuture<'static, DagResult<(NodeId, Arc)>>) -> F, - F: Future)>>, + S: Fn(BoxFuture<'static, DagResult>>) -> F, + F: Future>>, { #[cfg(feature = "tracing")] info!("starting DAG execution"); @@ -474,14 +472,15 @@ impl DagRunner { .cloned() .collect(); + let inner_future = spawner(node.execute_with_deps(dependencies)); // Spawn the task using the provided spawner let inner_future = async move { - let result = node.execute_with_deps(dependencies).await?; + let result = inner_future.await?; Ok((node_id, result)) }; Some( - AssertUnwindSafe(spawner(Box::pin(inner_future))) + AssertUnwindSafe(inner_future) .catch_unwind() .unwrap_or_else(move |panic_payload| { // Convert panic to error