Adjust io_scheduler to use moodycamel::ConcurrentQueue

Large benchmark gains on io_scheduler tcp/tls.

Author: jbaldwin

include/coro/io_scheduler.hpp

@@ -1,5 +1,6 @@
 #pragma once
 
+#include "coro/detail/vendor/cameron314/concurrentqueue/concurrentqueue.h"
 #include "coro/detail/poll_info.hpp"
 #include "coro/detail/timer_handle.hpp"
 #include "coro/expected.hpp"
@@ -145,10 +146,7 @@ class io_scheduler
             if (m_scheduler.m_opts.execution_strategy == execution_strategy_t::process_tasks_inline)
             {
                 m_scheduler.m_size.fetch_add(1, std::memory_order::release);
-                {
-                    std::scoped_lock lk{m_scheduler.m_scheduled_tasks_mutex};
-                    m_scheduler.m_scheduled_tasks.emplace_back(awaiting_coroutine);
-                }
+                m_scheduler.m_scheduled_tasks.enqueue(awaiting_coroutine);
 
                 // Trigger the event to wake-up the scheduler if this event isn't currently triggered.
                 bool expected{false};
@@ -393,10 +391,7 @@ class io_scheduler
         if (m_opts.execution_strategy == execution_strategy_t::process_tasks_inline)
         {
             m_size.fetch_add(1, std::memory_order::release);
-            {
-                std::scoped_lock lk{m_scheduled_tasks_mutex};
-                m_scheduled_tasks.emplace_back(handle);
-            }
+            m_scheduled_tasks.enqueue(handle);
 
             bool expected{false};
             if (m_schedule_fd_triggered.compare_exchange_strong(
@@ -481,8 +476,7 @@ class io_scheduler
     static auto       event_to_poll_status(uint32_t events) -> poll_status;
 
     auto                                 process_scheduled_execute_inline() -> void;
-    std::mutex                           m_scheduled_tasks_mutex{};
-    std::vector<std::coroutine_handle<>> m_scheduled_tasks{};
+    moodycamel::ConcurrentQueue<std::coroutine_handle<>> m_scheduled_tasks;
 
     static constexpr const int   m_shutdown_object{0};
     static constexpr const void* m_shutdown_ptr = &m_shutdown_object;

include/coro/thread_pool.hpp

@@ -239,13 +239,13 @@ class thread_pool
     /// The background executor threads.
     std::vector<std::thread> m_threads;
     /// Local executor worker thread queues.
-    std::vector<moodycamel::BlockingConcurrentQueue<std::coroutine_handle<>>> m_local_queues;
+    std::vector<moodycamel::ConcurrentQueue<std::coroutine_handle<>>> m_local_queues;
     /// Global queue.
     moodycamel::BlockingConcurrentQueue<std::coroutine_handle<>> m_global_queue;
 
     std::vector<std::unique_ptr<executor_state>> m_executor_state;
 
-    auto try_steal_work(std::size_t my_idx, std::array<std::coroutine_handle<>, 4>& handles) -> bool;
+    auto try_steal_work(std::size_t my_idx, std::array<std::coroutine_handle<>, 2>& handles) -> bool;
 
     /**
      * Each background thread runs from this function.

src/io_scheduler.cpp

@@ -290,26 +290,40 @@ auto io_scheduler::process_events_execute(std::chrono::milliseconds timeout) ->
 
 auto io_scheduler::process_scheduled_execute_inline() -> void
 {
-    std::vector<std::coroutine_handle<>> tasks{};
+    // Clear the schedule eventfd if this is a scheduled task.
+    int control = 0;
+    ::read(m_schedule_fd[1], reinterpret_cast<void*>(&control), sizeof(control));
+
+    // Clear the in memory flag to reduce eventfd_* calls on scheduling.
+    m_schedule_fd_triggered.exchange(false, std::memory_order::release);
+
+    constexpr std::size_t MAX{32};
+    std::size_t processed{0};
+    while (true)
     {
-        // Acquire the entire list, and then reset it.
-        std::scoped_lock lk{m_scheduled_tasks_mutex};
-        tasks.swap(m_scheduled_tasks);
+        std::array<std::coroutine_handle<>, MAX> tasks{nullptr};
+        if (!m_scheduled_tasks.try_dequeue_bulk(tasks.data(), MAX))
+        {
+            break;;
+        }
 
-        // Clear the schedule eventfd if this is a scheduled task.
-        int control = 0;
-        ::read(m_schedule_fd[1], reinterpret_cast<void*>(&control), sizeof(control));
+        for (std::size_t i = 0; i < MAX; ++i)
+        {
+            auto& handle = tasks[i];
+            if (handle == nullptr)
+            {
+                break;
+            }
 
-        // Clear the in memory flag to reduce eventfd_* calls on scheduling.
-        m_schedule_fd_triggered.exchange(false, std::memory_order::release);
+            handle.resume();
+            ++processed;
+        }
     }
 
-    // This set of handles can be safely resumed now since they do not have a corresponding timeout event.
-    for (auto& task : tasks)
+    if (processed > 0)
     {
-        task.resume();
+        m_size.fetch_sub(processed, std::memory_order::release);
     }
-    m_size.fetch_sub(tasks.size(), std::memory_order::release);
 }
 
 auto io_scheduler::process_event_execute(detail::poll_info* pi, poll_status status) -> void

src/thread_pool.cpp

@@ -5,7 +5,7 @@
 
 namespace coro
 {
-static constexpr std::size_t MAX_HANDLES{4};
+static constexpr std::size_t MAX_HANDLES{2};
 
 thread_pool::schedule_operation::schedule_operation(thread_pool& tp) noexcept : m_thread_pool(tp)
 {

test/bench.cpp

@@ -141,7 +141,6 @@ TEST_CASE("benchmark thread_pool{1} counter task", "[benchmark]")
 
     auto make_task = [](std::unique_ptr<coro::thread_pool>& tp, std::atomic<uint64_t>& c) -> coro::task<void>
     {
-        co_await tp->schedule();
         c.fetch_add(1, std::memory_order::relaxed);
         co_return;
     };