improvements and small fixes to 0057297

include/concurrencpp/executors/thread_pool_executor.h

@@ -52,6 +52,7 @@ namespace concurrencpp::details {
         std::binary_semaphore m_semaphore;
         bool m_idle;
         bool m_abort;
+        std::atomic_bool m_event_found;
         thread m_thread;
 
         void balance_work();
@@ -72,16 +73,16 @@ namespace concurrencpp::details {
 
         void enqueue_foreign(concurrencpp::task& task);
         void enqueue_foreign(std::span<concurrencpp::task> tasks);
-        void enqueue_foreign(const std::deque<concurrencpp::task>& deque,
-                             std::deque<concurrencpp::task>::iterator begin,
-                             std::deque<concurrencpp::task>::iterator end);
+        void enqueue_foreign(std::deque<concurrencpp::task>::iterator begin, std::deque<concurrencpp::task>::iterator end);
 
         void enqueue_local(concurrencpp::task& task);
         void enqueue_local(std::span<concurrencpp::task> tasks);
 
         void shutdown() noexcept;
 
         std::chrono::milliseconds max_worker_idle_time() const noexcept;
+
+        bool appears_empty() const noexcept;
     };
 }  // namespace concurrencpp::details
 

include/concurrencpp/results/impl/producer_context.h

@@ -27,7 +27,7 @@ namespace concurrencpp::details {
             m_result.emplace(std::forward<argument_types>(arguments)...);
         }
 
-        void build_exception(std::exception_ptr exception) noexcept {
+        void build_exception(const std::exception_ptr& exception) noexcept {
             assert(!m_result.has_value());
             assert(!static_cast<bool>(m_exception));
             m_exception = exception;
@@ -77,7 +77,7 @@ namespace concurrencpp::details {
             m_ready = true;
         }
 
-        void build_exception(std::exception_ptr exception) noexcept {
+        void build_exception(const std::exception_ptr& exception) noexcept {
             assert(!m_ready);
             assert(!static_cast<bool>(m_exception));
             m_exception = exception;
@@ -127,7 +127,7 @@ namespace concurrencpp::details {
             m_pointer = std::addressof(reference);
         }
 
-        void build_exception(std::exception_ptr exception) noexcept {
+        void build_exception(const std::exception_ptr& exception) noexcept {
             assert(m_pointer == nullptr);
             assert(!static_cast<bool>(m_exception));
             m_exception = exception;

include/concurrencpp/results/make_result.h

@@ -9,19 +9,15 @@ namespace concurrencpp {
         static_assert(std::is_constructible_v<type, argument_types...> || std::is_same_v<type, void>,
                       "concurrencpp::make_ready_result - <<type>> is not constructible from <<argument_types...>");
 
+        static_assert(std::is_same_v<type, void> ? (sizeof...(argument_types) == 0) : true,
+                      "concurrencpp::make_ready_result<void> - this overload does not accept any argument.");
+
         auto result_state_ptr = std::make_shared<details::result_state<type>>();
         result_state_ptr->set_result(std::forward<argument_types>(arguments)...);
         result_state_ptr->publish_result();
         return {std::move(result_state_ptr)};
     }
 
-    inline result<void> make_ready_result() {
-        auto result_state_ptr = std::make_shared<details::result_state<void>>();
-        result_state_ptr->set_result();
-        result_state_ptr->publish_result();
-        return {std::move(result_state_ptr)};
-    }
-
     template<class type>
     result<type> make_exceptional_result(std::exception_ptr exception_ptr) {
         if (!static_cast<bool>(exception_ptr)) {

include/concurrencpp/utils/bind.h

@@ -37,7 +37,6 @@ namespace concurrencpp::details {
                 tuple = std::make_tuple(std::forward<argument_types>(arguments)...)]() mutable noexcept(inti) -> decltype(auto) {
             try {
                 return std::apply(callable, tuple);
-
             } catch (...) {
                 // do nothing
             }

source/executors/thread_pool_executor.cpp

@@ -116,7 +116,7 @@ void idle_worker_set::find_idle_workers(size_t caller_index, std::vector<size_t>
 
 thread_pool_worker::thread_pool_worker(thread_pool_executor& parent_pool, size_t index, size_t pool_size, std::chrono::milliseconds max_idle_time) :
     m_atomic_abort(false), m_parent_pool(parent_pool), m_index(index), m_pool_size(pool_size), m_max_idle_time(max_idle_time),
-    m_worker_name(details::make_executor_worker_name(parent_pool.name)), m_semaphore(0), m_idle(true), m_abort(false) {
+    m_worker_name(details::make_executor_worker_name(parent_pool.name)), m_semaphore(0), m_idle(true), m_abort(false), m_event_found(false) {
     m_idle_worker_list.reserve(pool_size);
 }
 
@@ -175,7 +175,7 @@ void thread_pool_worker::balance_work() {
         assert(donation_begin_it < m_private_queue.end());
         assert(donation_end_it < m_private_queue.end());
 
-        m_parent_pool.worker_at(idle_worker_index).enqueue_foreign(m_private_queue, donation_begin_it, donation_end_it);
+        m_parent_pool.worker_at(idle_worker_index).enqueue_foreign(donation_begin_it, donation_end_it);
 
         begin = end;
         end += donation_count;
@@ -222,6 +222,10 @@ bool thread_pool_worker::wait_for_task(std::unique_lock<std::mutex>& lock) noexc
             }
         }
 
+        if (!m_event_found.load(std::memory_order_relaxed)) {
+            continue;
+        }
+
         lock.lock();
         if (m_public_queue.empty() && !m_abort) {
             lock.unlock();
@@ -282,6 +286,8 @@ bool thread_pool_worker::drain_queue() {
     assert(lock.owns_lock());
     assert(!m_public_queue.empty() || m_abort);
 
+    m_event_found.store(false, std::memory_order_relaxed);
+
     if (m_abort) {
         m_idle = true;
         return false;
@@ -337,6 +343,8 @@ void thread_pool_worker::enqueue_foreign(concurrencpp::task& task) {
         throw_executor_shutdown_exception(m_parent_pool.name);
     }
 
+    m_event_found.store(true, std::memory_order_relaxed);
+
     const auto is_empty = m_public_queue.empty();
     m_public_queue.emplace_back(std::move(task));
     ensure_worker_active(is_empty, lock);
@@ -348,20 +356,21 @@ void thread_pool_worker::enqueue_foreign(std::span<concurrencpp::task> tasks) {
         throw_executor_shutdown_exception(m_parent_pool.name);
     }
 
+    m_event_found.store(true, std::memory_order_relaxed);
+
     const auto is_empty = m_public_queue.empty();
     m_public_queue.insert(m_public_queue.end(), std::make_move_iterator(tasks.begin()), std::make_move_iterator(tasks.end()));
     ensure_worker_active(is_empty, lock);
 }
 
-void thread_pool_worker::enqueue_foreign(const std::deque<concurrencpp::task>& deque, std::deque<task>::iterator begin, std::deque<task>::iterator end) {
-    assert(begin != end);
+void thread_pool_worker::enqueue_foreign(std::deque<task>::iterator begin, std::deque<task>::iterator end) {
     std::unique_lock<std::mutex> lock(m_lock);
-    assert(&deque != &m_public_queue);
-
     if (m_abort) {
         throw_executor_shutdown_exception(m_parent_pool.name);
     }
 
+    m_event_found.store(true, std::memory_order_relaxed);
+
     const auto is_empty = m_public_queue.empty();
     m_public_queue.insert(m_public_queue.end(), std::make_move_iterator(begin), std::make_move_iterator(end));
     ensure_worker_active(is_empty, lock);
@@ -392,6 +401,8 @@ void thread_pool_worker::shutdown() noexcept {
         m_abort = true;
     }
 
+    m_event_found.store(true, std::memory_order_release);  // make sure the store is finished before notifying the worker.
+
     m_semaphore.release();
 
     if (m_thread.joinable()) {
@@ -415,6 +426,10 @@ std::chrono::milliseconds thread_pool_worker::max_worker_idle_time() const noexc
     return m_max_idle_time;
 }
 
+bool thread_pool_worker::appears_empty() const noexcept {
+    return m_private_queue.empty() && !m_event_found.load(std::memory_order_relaxed);
+}
+
 thread_pool_executor::thread_pool_executor(std::string_view pool_name, size_t pool_size, std::chrono::milliseconds max_idle_time) :
     derivable_executor<concurrencpp::thread_pool_executor>(pool_name), m_round_robin_cursor(0), m_idle_workers(pool_size), m_abort(false) {
     m_workers.reserve(pool_size);
@@ -448,13 +463,20 @@ void thread_pool_executor::mark_worker_active(size_t index) noexcept {
 }
 
 void thread_pool_executor::enqueue(concurrencpp::task task) {
-    const auto idle_worker_pos = m_idle_workers.find_idle_worker(details::s_tl_thread_pool_data.this_thread_index);
+    const auto this_worker = details::s_tl_thread_pool_data.this_worker;
+    const auto this_worker_index = details::s_tl_thread_pool_data.this_thread_index;
+
+    if (this_worker != nullptr && this_worker->appears_empty()) {
+        return this_worker->enqueue_local(task);
+    }
+
+    const auto idle_worker_pos = m_idle_workers.find_idle_worker(this_worker_index);
     if (idle_worker_pos != static_cast<size_t>(-1)) {
         return m_workers[idle_worker_pos].enqueue_foreign(task);
     }
 
-    if (details::s_tl_thread_pool_data.this_worker != nullptr) {
-        return details::s_tl_thread_pool_data.this_worker->enqueue_local(task);
+    if (this_worker != nullptr) {
+        return this_worker->enqueue_local(task);
     }
 
     const auto next_worker = m_round_robin_cursor.fetch_add(1, std::memory_order_relaxed) % m_workers.size();

test/source/tests/coroutine_tests/coroutines_tests.cpp

@@ -89,6 +89,9 @@ result<void> concurrencpp::tests::test_combo_coroutine_impl(std::shared_ptr<thre
 
     assert_equal(string_result, std::to_string(result_factory<int>::get()));
 
+    co_await ex->submit([] {
+    });
+
     auto& int_ref_result = co_await ex->submit([]() -> int& {
         return result_factory<int&>::get();
     });

test/source/tests/executor_tests/manual_executor_tests.cpp

@@ -53,7 +53,6 @@ namespace concurrencpp::tests {
         assert_equal(execution_map.size(), static_cast<size_t>(1));  // only one thread executed the tasks
         assert_equal(execution_map.begin()->first, concurrencpp::details::thread::get_current_virtual_id());  // and it's this thread.
     }
-
 }  // namespace concurrencpp::tests
 
 using namespace std::chrono;