fix threapool local, introduce resume_now() instead of handle().resume()

benchmark/benchmark_thread_pool.cpp

@@ -13,11 +13,10 @@ using namespace asyncpp;
 
 struct noop_promise : schedulable_promise {
     noop_promise(std::atomic_size_t* counter = nullptr) : counter(counter) {}
-    std::coroutine_handle<> handle() override {
+    void resume_now() override {
         if (counter) {
             counter->fetch_sub(1, std::memory_order_relaxed);
         }
-        return std::noop_coroutine();
     }
     std::atomic_size_t* counter = nullptr;
 };

include/asyncpp/promise.hpp

@@ -72,7 +72,7 @@ struct resumable_promise {
 
 struct schedulable_promise {
     virtual ~schedulable_promise() = default;
-    virtual std::coroutine_handle<> handle() = 0;
+    virtual void resume_now() = 0;
     schedulable_promise* m_scheduler_next = nullptr;
     schedulable_promise* m_scheduler_prev = nullptr;
     scheduler* m_scheduler = nullptr;

include/asyncpp/stream.hpp

@@ -100,16 +100,18 @@ namespace impl_stream {
             m_result = std::nullopt;
         }
 
-        auto handle() -> std::coroutine_handle<> override {
-            return std::coroutine_handle<promise>::from_promise(*this);
+        void resume() final {
+            return m_scheduler ? m_scheduler->schedule(*this) : resume_now();
         }
 
-        void resume() override {
-            return m_scheduler ? m_scheduler->schedule(*this) : handle().resume();
+        void resume_now() final {
+            const auto handle = std::coroutine_handle<promise>::from_promise(*this);
+            handle.resume();
         }
 
         void destroy() noexcept {
-            handle().destroy();
+            const auto handle = std::coroutine_handle<promise>::from_promise(*this);
+            handle.destroy();
         }
 
         auto await() noexcept;

include/asyncpp/task.hpp

@@ -40,12 +40,13 @@ namespace impl_task {
             return final_awaitable{};
         }
 
-        auto handle() -> std::coroutine_handle<> final {
-            return std::coroutine_handle<promise>::from_promise(*this);
+        void resume_now() final {
+            const auto handle = std::coroutine_handle<promise>::from_promise(*this);
+            handle.resume();
         }
 
         void resume() final {
-            return m_scheduler ? m_scheduler->schedule(*this) : handle().resume();
+            return m_scheduler ? m_scheduler->schedule(*this) : resume_now();
         }
 
         void start() {
@@ -62,7 +63,8 @@ namespace impl_task {
         }
 
         void destroy() {
-            handle().destroy();
+            const auto handle = std::coroutine_handle<promise>::from_promise(*this);
+            handle.destroy();
         }
 
     private:

include/asyncpp/thread_pool.hpp

@@ -107,12 +107,13 @@ class thread_pool_3 : public scheduler {
 
     private:
         void run(pack& pack) {
+            m_local = this;
             size_t stealing_attempt = pack.workers.size();
             bool exit_loop = false;
             while (!exit_loop) {
                 const auto promise = try_get_promise(pack, stealing_attempt, exit_loop);
                 if (promise) {
-                    promise->handle().resume();
+                    promise->resume_now();
                 }
             }
         }
@@ -161,8 +162,8 @@ class thread_pool_3 : public scheduler {
     }
 
 private:
-    pack m_pack;
-    std::atomic_ptrdiff_t m_next_in_schedule;
+    alignas(64) pack m_pack;
+    alignas(64) std::atomic_ptrdiff_t m_next_in_schedule;
     inline static thread_local worker* m_local = nullptr;
 };
 

src/thread_pool.cpp

@@ -70,7 +70,7 @@ void thread_pool::execute(worker& local,
                           std::span<worker> workers) {
     do {
         if (const auto item = INTERLEAVED(local.worklist.pop())) {
-            item->handle().resume();
+            item->resume_now();
             continue;
         }
         else if (const auto item = INTERLEAVED(global_worklist.pop())) {

test/helper_schedulers.hpp

@@ -22,7 +22,7 @@ class thread_locked_scheduler : public asyncpp::scheduler {
     void resume() {
         const auto promise = m_promise.exchange(nullptr);
         assert(promise != nullptr);
-        promise->handle().resume();
+        promise->resume_now();
     }
 
 private:

test/monitor_task.hpp

@@ -37,17 +37,19 @@ class [[nodiscard]] monitor_task {
             m_counters->exception = std::current_exception();
         }
 
-        void resume() override {
+        void resume() final {
             m_counters->suspensions.fetch_add(1);
-            m_scheduler ? m_scheduler->schedule(*this) : handle().resume();
+            m_scheduler ? m_scheduler->schedule(*this) : resume_now();
         }
 
-        std::coroutine_handle<> handle() override {
-            return std::coroutine_handle<promise>::from_promise(*this);
+        void resume_now() final {
+            const auto handle = std::coroutine_handle<promise>::from_promise(*this);
+            handle.resume();
         }
 
         void destroy() noexcept {
-            handle().destroy();
+            const auto handle = std::coroutine_handle<promise>::from_promise(*this);
+            handle.destroy();
         }
 
         const counters& get_counters() const {

test/test_thread_pool.cpp

@@ -21,9 +21,8 @@ constexpr size_t branching = 10;
 
 
 struct test_promise : schedulable_promise {
-    std::coroutine_handle<> handle() override {
+    void resume_now() final {
         ++num_queried;
-        return std::noop_coroutine();
     }
     std::atomic_size_t num_queried = 0;
 };