perf(runtime): transfer existing tasks directly

CHANGELOG.md

@@ -52,6 +52,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Changed
 
+- **Direct task spawn handoff**: `go`, `go_to`, `go_joinable`, and
+  `go_joinable_to` now accept an already-constructed rvalue `task<T>` without
+  creating a callable-wrapper coroutine. Callable overloads retain their
+  existing lifetime-safe wrapper behavior (#1030).
 - **Bounded scheduler service under runnable load**: Workers now check
   cross-thread submissions every 256 completed worker-loop task dispatches and
   give pending I/O a non-blocking service opportunity every 16,384 dispatches

README.md

@@ -397,9 +397,12 @@ elio::coro::task<int> compute() {
 elio::runtime::scheduler sched(num_threads);
 sched.start();
 
-// Spawn coroutines (pass callable, not invoked task)
+// Callable form safely keeps the callable and arguments in a wrapper frame
 sched.go(my_coroutine);  // fire-and-forget
-// Or for joinable: auto handle = sched.go_joinable(my_coroutine);
+
+// Direct form transfers an uncompleted lazy task without a wrapper
+sched.go(my_coroutine());
+auto handle = sched.go_joinable(my_coroutine());
 
 // Dynamic thread adjustment
 sched.set_thread_count(8);
@@ -430,6 +433,9 @@ elio::go(some_task);
 // With arguments
 elio::go(task_with_args, arg1, arg2);
 
+// Directly transfer an already-constructed task (no callable wrapper)
+elio::go(task_with_args(arg1, arg2));
+
 // Lambda with captures (safe - copied into coroutine frame)
 int value = 42;
 elio::go([value]() -> coro::task<void> {
@@ -439,6 +445,7 @@ elio::go([value]() -> coro::task<void> {
 
 // Joinable spawn: get a handle to await later
 auto handle = elio::spawn(compute_value);
+auto direct_handle = elio::spawn(compute_value());
 // ... do other work ...
 int result = co_await handle;  // Wait and get result
 

examples/quick_benchmark.cpp

@@ -16,6 +16,7 @@ using namespace std::chrono;
 
 // Minimum benchmark duration - reduced for quick testing
 constexpr auto MIN_BENCH_DURATION = seconds(3);
+constexpr auto SPAWN_BATCH_TIMEOUT = seconds(30);
 
 // Statistics helper
 struct bench_stats {
@@ -113,8 +114,10 @@ void benchmark_spawn_overhead() {
             sched.go(empty_task);
         }
 
-        while (sched.pending_tasks() > 0) {
-            std::this_thread::sleep_for(microseconds(1));
+        if (!sched.wait_for_idle(SPAWN_BATCH_TIMEOUT)) {
+            std::cerr << "Timed out waiting for callable-spawn batch"
+                      << std::endl;
+            std::abort();
         }
 
         auto batch_end = high_resolution_clock::now();
@@ -131,13 +134,63 @@ void benchmark_spawn_overhead() {
 
     auto stats = bench_stats::compute(samples);
 
-    std::cout << "Task Spawn: " << std::fixed << std::setprecision(2)
+    std::cout << "Task Spawn (callable): " << std::fixed << std::setprecision(2)
               << stats.avg << " ns/task (min=" << stats.min
               << ", max=" << stats.max << ")" << std::endl;
     std::cout << "  Throughput: " << std::fixed << std::setprecision(0)
               << (total_tasks / total_sec) << " tasks/sec" << std::endl;
 }
 
+// Same work with an already-constructed lazy task transferred directly. This
+// isolates the callable-wrapper frame and control-state cost.
+void benchmark_direct_task_spawn_overhead() {
+    const int batch_size = 10000;
+    std::vector<double> samples;
+    size_t total_tasks = 0;
+
+    auto bench_start = high_resolution_clock::now();
+
+    while (duration_cast<seconds>(high_resolution_clock::now() - bench_start) <
+           MIN_BENCH_DURATION) {
+        runtime::scheduler sched(4);
+        sched.start();
+
+        auto batch_start = high_resolution_clock::now();
+
+        for (int i = 0; i < batch_size; ++i) {
+            sched.go(empty_task());
+        }
+
+        if (!sched.wait_for_idle(SPAWN_BATCH_TIMEOUT)) {
+            std::cerr << "Timed out waiting for direct-spawn batch"
+                      << std::endl;
+            std::abort();
+        }
+
+        auto batch_end = high_resolution_clock::now();
+        auto batch_ns = duration_cast<nanoseconds>(
+            batch_end - batch_start).count();
+
+        samples.push_back(static_cast<double>(batch_ns) / batch_size);
+        total_tasks += batch_size;
+
+        sched.shutdown();
+    }
+
+    auto bench_end = high_resolution_clock::now();
+    auto total_sec = duration_cast<milliseconds>(
+        bench_end - bench_start).count() / 1000.0;
+
+    auto stats = bench_stats::compute(samples);
+
+    std::cout << "Task Spawn (direct task): " << std::fixed
+              << std::setprecision(2) << stats.avg
+              << " ns/task (min=" << stats.min
+              << ", max=" << stats.max << ")" << std::endl;
+    std::cout << "  Throughput: " << std::fixed << std::setprecision(0)
+              << (total_tasks / total_sec) << " tasks/sec" << std::endl;
+}
+
 // Time-based context switch benchmark
 void benchmark_context_switch() {
     const int batch_size = 5000;
@@ -327,6 +380,7 @@ int main() {
               << duration_cast<seconds>(MIN_BENCH_DURATION).count() << "s each) ===" << std::endl;
 
     benchmark_spawn_overhead();
+    benchmark_direct_task_spawn_overhead();
     benchmark_context_switch();
     benchmark_yield();
     benchmark_scheduler_reschedule();

include/elio/coro/promise_base.hpp

@@ -13,6 +13,12 @@ class scheduler;
 
 namespace elio::coro {
 
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+namespace detail {
+inline std::atomic<size_t> promise_constructions_for_test{0};
+}  // namespace detail
+#endif
+
 /// Coroutine state for debugging
 enum class coroutine_state : uint8_t {
     created = 0,    // Just created, not started
@@ -98,6 +104,10 @@ class promise_base {
 #endif
         , execution_context_(std::make_shared<task_execution_context>())
     {
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+        detail::promise_constructions_for_test.fetch_add(
+            1, std::memory_order_relaxed);
+#endif
         current_frame_ = this;
     }
 

include/elio/runtime/scheduler.hpp

@@ -19,6 +19,7 @@
 #include <algorithm>
 #include <functional>
 #include <exception>
+#include <new>
 #include <stdexcept>
 #include <vector>
 
@@ -38,6 +39,7 @@ namespace detail {
 #ifdef ELIO_RUNTIME_TEST_HOOKS
     inline std::atomic<bool> reject_next_schedule_for_test{false};
     inline std::atomic<bool> reject_next_spawn_for_test{false};
+    inline std::atomic<bool> fail_next_join_state_allocation_for_test{false};
     inline std::atomic<size_t> local_schedule_fast_paths_for_test{0};
     inline std::atomic<size_t> local_schedule_fallbacks_for_test{0};
     inline std::atomic<bool> pause_shutdown_teardown_for_test{false};
@@ -638,6 +640,13 @@ class scheduler {
         do_go_<false, false>(0, std::forward<F>(f), std::forward<Args>(args)...);
     }
 
+    /// Transfer an already-constructed lazy task directly to this scheduler.
+    /// Unlike the callable overload, this does not need a lifetime wrapper.
+    template<typename T>
+    void go(coro::task<T>&& task) {
+        do_go_task_<false, false>(0, std::move(task));
+    }
+
     /// High-level API: fire-and-forget, with affinity to a specific worker.
     /// Affinity is set before first resume; steal attempts that observe the
     /// task on another queue bounce it back to the target worker instead of
@@ -649,6 +658,12 @@ class scheduler {
         do_go_<false, true>(worker_id, std::forward<F>(f), std::forward<Args>(args)...);
     }
 
+    /// Transfer an already-constructed lazy task toward a specific worker.
+    template<typename T>
+    void go_to(size_t worker_id, coro::task<T>&& task) {
+        do_go_task_<false, true>(worker_id, std::move(task));
+    }
+
     /// High-level API: spawn + join, spawn to this scheduler
     template<typename F, typename... Args>
         requires (std::invocable<F, Args...> && detail::is_task_v<std::invoke_result_t<F, Args...>>)
@@ -658,6 +673,12 @@ class scheduler {
         return do_go_<true, false>(0, std::forward<F>(f), std::forward<Args>(args)...);
     }
 
+    /// Transfer an already-constructed lazy task and retain join authority.
+    template<typename T>
+    auto go_joinable(coro::task<T>&& task) -> coro::join_handle<T> {
+        return do_go_task_<true, false>(0, std::move(task));
+    }
+
     /// High-level API: spawn + join, with affinity to a specific worker.
     /// Affinity is set before first resume; I/O is bound to that worker's
     /// io_context, and steal attempts bounce the task back to the affinity
@@ -672,6 +693,14 @@ class scheduler {
         return do_go_<true, true>(worker_id, std::forward<F>(f), std::forward<Args>(args)...);
     }
 
+    /// Transfer an already-constructed lazy task toward a worker and retain
+    /// join authority.
+    template<typename T>
+    auto go_joinable_to(size_t worker_id, coro::task<T>&& task)
+        -> coro::join_handle<T> {
+        return do_go_task_<true, true>(worker_id, std::move(task));
+    }
+
     /// Spawn an existing coroutine toward worker_id. Exact placement requires
     /// worker_id < num_threads(); otherwise scheduling is best-effort fallback.
     /// This is an initial ownership handoff and detaches construction-time
@@ -1139,29 +1168,39 @@ class scheduler {
         return scheduled;
     }
 
-    template<bool Joinable, bool Pinned, typename F, typename... Args>
-    auto do_go_(size_t worker_id, F&& f, Args&&... args) {
-        using ResultTask = std::invoke_result_t<F, Args...>;
-        using T = detail::task_value_t<ResultTask>;
-
-        auto wrapper = [&]() {
-            if constexpr (Joinable) {
-                return detail::callable_wrapper(std::forward<F>(f), std::forward<Args>(args)...);
-            } else {
-                return detail::callable_wrapper_void(std::forward<F>(f), std::forward<Args>(args)...);
-            }
-        }();
+    template<bool Joinable, bool Pinned, typename T>
+    auto do_go_task_(size_t worker_id, coro::task<T>&& task) {
+        if (!task) {
+            throw std::invalid_argument(
+                "cannot transfer an empty task to the scheduler");
+        }
+        if (coro::detail::task_access::handle(task).done()) {
+            throw std::invalid_argument(
+                "cannot transfer a completed task to the scheduler");
+        }
 
-        auto handle = coro::detail::task_access::release(std::move(wrapper));
+        auto handle = coro::detail::task_access::release(std::move(task));
         handle.promise().detached_ = true;
         if constexpr (Pinned) {
             handle.promise().set_affinity(worker_id);
         }
         handle.promise().detach_from_parent();
 
         if constexpr (Joinable) {
-            auto state = std::make_shared<coro::detail::join_state<T>>(
-                handle.promise().execution_context());
+            std::shared_ptr<coro::detail::join_state<T>> state;
+            try {
+#ifdef ELIO_RUNTIME_TEST_HOOKS
+                if (detail::fail_next_join_state_allocation_for_test.exchange(
+                        false, std::memory_order_acq_rel)) {
+                    throw std::bad_alloc{};
+                }
+#endif
+                state = std::make_shared<coro::detail::join_state<T>>(
+                    handle.promise().execution_context());
+            } catch (...) {
+                handle.destroy();
+                throw;
+            }
             handle.promise().join_state_ = state;
             bool scheduled = false;
             try {
@@ -1203,6 +1242,21 @@ class scheduler {
         }
     }
 
+    template<bool Joinable, bool Pinned, typename F, typename... Args>
+    auto do_go_(size_t worker_id, F&& f, Args&&... args) {
+        auto wrapper = [&]() {
+            if constexpr (Joinable) {
+                return detail::callable_wrapper(
+                    std::forward<F>(f), std::forward<Args>(args)...);
+            } else {
+                return detail::callable_wrapper_void(
+                    std::forward<F>(f), std::forward<Args>(args)...);
+            }
+        }();
+        return do_go_task_<Joinable, Pinned>(
+            worker_id, std::move(wrapper));
+    }
+
     void reap_draining_workers_() noexcept {
         auto it = draining_workers_.begin();
         while (it != draining_workers_.end()) {

include/elio/runtime/spawn.hpp

@@ -36,6 +36,18 @@ void go(F&& f, Args&&... args) {
     std::abort();
 }
 
+/// Fire-and-forget direct transfer of an already-constructed lazy task.
+template<typename T>
+void go(coro::task<T>&& task) {
+    auto* sched = runtime::scheduler::current();
+    if (sched && sched->is_running()) {
+        sched->go(std::move(task));
+        return;
+    }
+    ELIO_LOG_ERROR("elio::go() called without a running scheduler — aborting");
+    std::abort();
+}
+
 /// Fire-and-forget: spawn a coroutine with affinity to a specific worker.
 /// The task is bound to the given worker before first resume. Steal attempts
 /// that observe it on another queue bounce it back instead of executing it.
@@ -63,6 +75,19 @@ void go_to(size_t worker_id, F&& f, Args&&... args) {
     std::abort();
 }
 
+/// Fire-and-forget direct transfer of an already-constructed lazy task toward
+/// a specific worker.
+template<typename T>
+void go_to(size_t worker_id, coro::task<T>&& task) {
+    auto* sched = runtime::scheduler::current();
+    if (sched && sched->is_running()) {
+        sched->go_to(worker_id, std::move(task));
+        return;
+    }
+    ELIO_LOG_ERROR("elio::go_to() called without a running scheduler — aborting");
+    std::abort();
+}
+
 /// Spawn a coroutine and return a join_handle to await its result.
 /// The coroutine runs concurrently and the result can be retrieved via co_await.
 ///
@@ -92,6 +117,34 @@ auto spawn(F&& f, Args&&... args)
     return coro::join_handle<T>{std::move(state)};
 }
 
+/// Directly transfer an already-constructed lazy task and retain join
+/// authority without creating a callable-wrapper coroutine.
+template<typename T>
+auto spawn(coro::task<T>&& task) -> coro::join_handle<T> {
+    if (!task) {
+        throw std::invalid_argument(
+            "cannot transfer an empty task to the scheduler");
+    }
+    auto task_handle = coro::detail::task_access::handle(task);
+    if (task_handle.done()) {
+        throw std::invalid_argument(
+            "cannot transfer a completed task to the scheduler");
+    }
+    auto* sched = runtime::scheduler::current();
+    if (sched && sched->is_running()) {
+        return sched->go_joinable(std::move(task));
+    }
+    auto state = std::make_shared<coro::detail::join_state<T>>(
+        task_handle.promise().execution_context());
+    task_handle = coro::detail::task_access::release(std::move(task));
+    task_handle.promise().detach_from_parent();
+    task_handle.destroy();
+    state->set_exception(std::make_exception_ptr(
+        std::logic_error("elio::spawn() called without a running scheduler")));
+    state->mark_destroyed();
+    return coro::join_handle<T>{std::move(state)};
+}
+
 } // namespace elio
 
 // Macros — syntactic sugar for inline lambda coroutines.