Add advanced coroutine features: yield, work-stealing, cache-aligned allocator

Co-authored-by: caixuf <130882544+caixuf@users.noreply.github.com>

Author: Copilot

examples/advanced_features_demo.cpp

@@ -0,0 +1,125 @@
+/**
+ * @file advanced_features_demo.cpp
+ * @brief 演示FlowCoro的高级架构特性
+ */
+
+#include <flowcoro.hpp>
+#include <iostream>
+
+using namespace flowcoro;
+
+// 演示1：使用yield进行协作式调度
+Task<int> cooperative_task(int id) {
+    int result = 0;
+    for (int i = 0; i < 1000; ++i) {
+        result += i;
+        
+        // 每100次迭代yield一次，让其他协程有机会执行
+        if (i % 100 == 0) {
+            co_await yield();  // 轻量级yield，不需要完整的重新调度
+        }
+    }
+    co_return result;
+}
+
+// 演示2：批量处理与周期性yield
+Task<void> batch_processing_task() {
+    std::vector<int> data(10000);
+    size_t counter = 0;
+    
+    for (size_t i = 0; i < data.size(); ++i) {
+        data[i] = i * 2;
+        
+        // 使用BatchYieldAwaiter自动管理yield频率
+        co_await BatchYieldAwaiter(counter, 500);
+    }
+    
+    std::cout << "Batch processing completed with " << counter << " yields\n";
+}
+
+// 演示3：高性能并发任务
+Task<void> concurrent_workers() {
+    std::vector<Task<int>> workers;
+    
+    // 创建多个协作式工作任务
+    for (int i = 0; i < 10; ++i) {
+        workers.push_back(cooperative_task(i));
+    }
+    
+    // 等待所有任务完成
+    int total = 0;
+    for (auto& worker : workers) {
+        total += co_await worker;
+    }
+    
+    std::cout << "Total result from " << workers.size() << " workers: " << total << "\n";
+}
+
+// 演示4：立即执行的void任务（利用suspend_never）
+Task<void> immediate_void_task() {
+    // 这个任务会立即执行，因为Task<void>现在使用suspend_never
+    std::cout << "Immediate void task executing!\n";
+    co_return;
+}
+
+// 演示5：组合使用不同的异步原语
+Task<void> combined_async_operations() {
+    std::cout << "Starting combined operations...\n";
+    
+    // 1. 立即执行的任务
+    co_await immediate_void_task();
+    
+    // 2. 协作式yield
+    co_await yield();
+    
+    // 3. 延时操作
+    co_await sleep_for(std::chrono::milliseconds(10));
+    
+    // 4. 并发任务
+    auto task1 = cooperative_task(1);
+    auto task2 = cooperative_task(2);
+    
+    auto [value, index] = co_await when_any(std::move(task1), std::move(task2));
+    std::cout << "First completed task (index " << index << ") returned: " << value << "\n";
+    
+    std::cout << "Combined operations completed!\n";
+}
+
+int main() {
+    std::cout << "=== FlowCoro Advanced Features Demo ===\n\n";
+    
+    try {
+        // 演示1: 协作式任务
+        std::cout << "Demo 1: Cooperative Task\n";
+        auto result1 = sync_wait(cooperative_task(0));
+        std::cout << "Result: " << result1 << "\n\n";
+        
+        // 演示2: 批量处理
+        std::cout << "Demo 2: Batch Processing\n";
+        sync_wait(batch_processing_task());
+        std::cout << "\n";
+        
+        // 演示3: 并发工作器
+        std::cout << "Demo 3: Concurrent Workers\n";
+        sync_wait(concurrent_workers());
+        std::cout << "\n";
+        
+        // 演示4: 立即执行的void任务
+        std::cout << "Demo 4: Immediate Void Task\n";
+        sync_wait(immediate_void_task());
+        std::cout << "\n";
+        
+        // 演示5: 组合操作
+        std::cout << "Demo 5: Combined Async Operations\n";
+        sync_wait(combined_async_operations());
+        std::cout << "\n";
+        
+        std::cout << "=== All demos completed successfully! ===\n";
+        
+    } catch (const std::exception& e) {
+        std::cerr << "Exception: " << e.what() << "\n";
+        return 1;
+    }
+    
+    return 0;
+}

include/flowcoro.hpp

@@ -49,6 +49,8 @@
 #include "flowcoro/simple_db.h"
 #include "flowcoro/rpc.h"
 #include "flowcoro/channel.h"
+#include "flowcoro/yield.h"           // 新增：轻量级yield支持
+#include "flowcoro/task_allocator.h"  // 新增：缓存友好的任务分配器
 
 // #include "flowcoro/rpc.h" // 暂时注释掉复杂的RPC实现
 

include/flowcoro/task_allocator.h

@@ -0,0 +1,66 @@
+#pragma once
+#include <memory>
+#include <new>
+#include <cstddef>
+
+namespace flowcoro {
+
+// 缓存对齐的协程promise分配器
+// 提高缓存局部性，减少false sharing
+template<typename T>
+class CacheAlignedAllocator {
+public:
+    using value_type = T;
+    using size_type = std::size_t;
+    using difference_type = std::ptrdiff_t;
+    
+    // C++20 alignas支持
+    static constexpr size_t alignment = std::hardware_destructive_interference_size;
+    
+    CacheAlignedAllocator() noexcept = default;
+    
+    template<typename U>
+    CacheAlignedAllocator(const CacheAlignedAllocator<U>&) noexcept {}
+    
+    T* allocate(size_t n) {
+        if (n == 0) return nullptr;
+        
+        size_t size = n * sizeof(T);
+        // 对齐到缓存行
+        void* ptr = ::operator new(size, std::align_val_t{alignment});
+        
+        if (!ptr) {
+            throw std::bad_alloc();
+        }
+        
+        return static_cast<T*>(ptr);
+    }
+    
+    void deallocate(T* ptr, size_t) noexcept {
+        ::operator delete(ptr, std::align_val_t{alignment});
+    }
+    
+    template<typename U>
+    bool operator==(const CacheAlignedAllocator<U>&) const noexcept {
+        return true;
+    }
+    
+    template<typename U>
+    bool operator!=(const CacheAlignedAllocator<U>&) const noexcept {
+        return false;
+    }
+};
+
+// 为promise_type提供自定义分配器支持
+// 可以在promise_type中添加以下方法来使用：
+// void* operator new(std::size_t size) {
+//     CacheAlignedAllocator<promise_type> alloc;
+//     return alloc.allocate(1);
+// }
+// 
+// void operator delete(void* ptr) noexcept {
+//     CacheAlignedAllocator<promise_type> alloc;
+//     alloc.deallocate(static_cast<promise_type*>(ptr), 1);
+// }
+
+} // namespace flowcoro

include/flowcoro/yield.h

@@ -0,0 +1,50 @@
+#pragma once
+#include <coroutine>
+#include "coroutine_manager.h"
+
+namespace flowcoro {
+
+// 轻量级yield awaiter - 立即恢复，但给其他协程执行机会
+struct YieldAwaiter {
+    bool await_ready() const noexcept { 
+        return false; // 总是挂起以给其他协程机会
+    }
+    
+    void await_suspend(std::coroutine_handle<> h) noexcept {
+        // 立即重新调度当前协程，但允许其他协程先执行
+        auto& manager = CoroutineManager::get_instance();
+        manager.schedule_resume(h);
+    }
+    
+    void await_resume() const noexcept {}
+};
+
+// 便捷函数：让出执行权
+inline YieldAwaiter yield() noexcept {
+    return {};
+}
+
+// 优化的批量操作awaiter - 在循环中周期性yield
+class BatchYieldAwaiter {
+private:
+    size_t& counter_;
+    const size_t yield_interval_;
+    
+public:
+    BatchYieldAwaiter(size_t& counter, size_t yield_interval = 100) 
+        : counter_(counter), yield_interval_(yield_interval) {}
+    
+    bool await_ready() const noexcept {
+        // 只在达到间隔时才挂起
+        return (++counter_ % yield_interval_) != 0;
+    }
+    
+    void await_suspend(std::coroutine_handle<> h) noexcept {
+        auto& manager = CoroutineManager::get_instance();
+        manager.schedule_resume(h);
+    }
+    
+    void await_resume() const noexcept {}
+};
+
+} // namespace flowcoro

src/coroutine_pool.cpp

@@ -250,6 +250,29 @@ class CoroutineScheduler {
         return queue_size_.load(std::memory_order_relaxed);
     }
 
+    // 工作窃取：尝试从队列中获取一批任务给其他调度器
+    size_t try_steal_work(std::vector<std::coroutine_handle<>>& stolen_tasks, size_t max_steal = 32) {
+        if (queue_size_.load(std::memory_order_relaxed) <= 1) {
+            return 0; // 队列太小，不值得窃取
+        }
+        
+        size_t stolen_count = 0;
+        std::coroutine_handle<> handle;
+        
+        // 窃取一半的任务（最多max_steal个）
+        size_t target_steal = std::min(queue_size_.load() / 2, max_steal);
+        
+        while (stolen_count < target_steal && coroutine_queue_.dequeue(handle)) {
+            if (handle && !handle.done()) {
+                stolen_tasks.push_back(handle);
+                stolen_count++;
+                queue_size_.fetch_sub(1, std::memory_order_relaxed);
+            }
+        }
+        
+        return stolen_count;
+    }
+    
     size_t get_total_coroutines() const { return total_coroutines_.load(); }
     size_t get_completed_coroutines() const { return completed_coroutines_.load(); }
     size_t get_scheduler_id() const { return scheduler_id_; }