Fix deadlock in ring_buffer and queue shutdown_drain (#402)

* Allow timeout version of io_scheduler::schedule with void tasks

If the return_type is void, we cannot get the (incomplete) void type out of
the result. Instead use constexpr dispatch to default-construct the
expected (void) return value in these cases.

* IWYU in queue.hpp and ring_buffer.hpp
* Fix deadlock in ring_buffer and queue shutdown_drain

If a ring_buffer (queue) produces an element such that the buffer is
non-empty and then calls shutdown_drain we enter a draining loop where we
yield until the ring buffer is empty. If, meanwhile, the consumer side of
the buffer decides to forcibly shutdown without consuming the produced
value (for example because it has received an exception that should be
propagated and therefore wants to wake and shutdown the producer) then we
never exit the yield-until-empty loop.

To fix this only loop while the buffer is both non-empty and we are still
in the draining state.

The produced value(s) that are stored in the buffer are now never
processed which I think is the right semantics since the consumer has
explicitly said "I can't do anything with this".

* Use thread_pool rather than io_scheduler to avoid strange link errors

- Fixes #401

Author: wence-

include/coro/io_scheduler.hpp

@@ -7,6 +7,7 @@
 #include "coro/io_notifier.hpp"
 #include "coro/poll.hpp"
 #include "coro/thread_pool.hpp"
+#include <type_traits>
 #include <unistd.h>
 
 #ifdef LIBCORO_FEATURE_NETWORKING
@@ -17,7 +18,6 @@
 #include <functional>
 #include <map>
 #include <memory>
-#include <optional>
 #include <stop_token>
 #include <thread>
 #include <vector>
@@ -225,13 +225,28 @@ class io_scheduler
         auto timeout_ms = std::max(std::chrono::duration_cast<std::chrono::milliseconds>(timeout), 0ms);
         if (timeout_ms == 0ms)
         {
-            co_return coro::expected<return_type, timeout_status>(co_await schedule(std::move(task)));
+            if constexpr (std::is_void_v<return_type>)
+            {
+                co_await schedule(std::move(task));
+                co_return coro::expected<return_type, timeout_status>();
+            }
+            else
+            {
+                co_return coro::expected<return_type, timeout_status>(co_await schedule(std::move(task)));
+            }
         }
 
         auto result = co_await when_any(std::move(task), make_timeout_task(timeout_ms));
         if (!std::holds_alternative<timeout_status>(result))
         {
-            co_return coro::expected<return_type, timeout_status>(std::move(std::get<0>(result)));
+            if constexpr (std::is_void_v<return_type>)
+            {
+                co_return coro::expected<return_type, timeout_status>();
+            }
+            else
+            {
+                co_return coro::expected<return_type, timeout_status>(std::move(std::get<0>(result)));
+            }
         }
         else
         {
@@ -261,13 +276,27 @@ class io_scheduler
         auto timeout_ms = std::max(std::chrono::duration_cast<std::chrono::milliseconds>(timeout), 0ms);
         if (timeout_ms == 0ms)
         {
-            co_return coro::expected<return_type, timeout_status>(co_await schedule(std::move(task)));
+            if constexpr (std::is_void_v<return_type>)
+            {
+                co_return coro::expected<return_type, timeout_status>();
+            }
+            else
+            {
+                co_return coro::expected<return_type, timeout_status>(co_await schedule(std::move(task)));
+            }
         }
 
         auto result = co_await when_any(std::move(stop_source), std::move(task), make_timeout_task(timeout_ms));
         if (!std::holds_alternative<timeout_status>(result))
         {
-            co_return coro::expected<return_type, timeout_status>(std::move(std::get<0>(result)));
+            if constexpr (std::is_void_v<return_type>)
+            {
+                co_return coro::expected<return_type, timeout_status>();
+            }
+            else
+            {
+                co_return coro::expected<return_type, timeout_status>(std::move(std::get<0>(result)));
+            }
         }
         else
         {
@@ -282,7 +311,7 @@ class io_scheduler
      *               Given zero or negative amount of time this behaves identical to schedule().
      */
     template<class rep_type, class period_type>
-     [[nodiscard]] auto schedule_after(std::chrono::duration<rep_type, period_type> amount) -> coro::task<void>
+    [[nodiscard]] auto schedule_after(std::chrono::duration<rep_type, period_type> amount) -> coro::task<void>
     {
         return yield_for_internal(std::chrono::duration_cast<std::chrono::nanoseconds>(amount));
     }

include/coro/queue.hpp

@@ -2,9 +2,11 @@
 
 #include "coro/concepts/executor.hpp"
 #include "coro/expected.hpp"
+#include "coro/mutex.hpp"
 #include "coro/sync_wait.hpp"
 #include "coro/task.hpp"
 
+#include <atomic>
 #include <queue>
 
 namespace coro
@@ -377,7 +379,7 @@ class queue
         }
         lk.unlock();
 
-        while (!empty())
+        while (!empty() && m_running_state.load(std::memory_order::acquire) == running_state_t::draining)
         {
             co_await e->yield();
         }

include/coro/ring_buffer.hpp

@@ -1,13 +1,16 @@
 #pragma once
 
+#include "coro/concepts/executor.hpp"
+#include "coro/detail/awaiter_list.hpp"
 #include "coro/expected.hpp"
 #include "coro/mutex.hpp"
+#include "coro/sync_wait.hpp"
 #include "coro/task.hpp"
 
 #include <array>
 #include <atomic>
 #include <coroutine>
-#include <mutex>
+#include <memory>
 #include <optional>
 
 namespace coro
@@ -301,7 +304,7 @@ class ring_buffer
             produce_waiters = next;
         }
 
-        while (!empty())
+        while (!empty() && m_running_state.load(std::memory_order::acquire) == running_state_t::draining)
         {
             co_await e->yield();
         }

test/test_queue.cpp

@@ -237,6 +237,30 @@ TEST_CASE("queue.try_pop", "[queue]")
     REQUIRE(expected.error() == coro::queue_consume_result::stopped);
 }
 
+TEST_CASE("queue shutdown_drain non-empty consumer shutdown", "[queue]")
+{
+    std::cerr << "BEGIN queue issue-401\n";
+
+    coro::queue<int> q{};
+    auto             exec = coro::thread_pool::make_unique(coro::thread_pool::options{.thread_count = 1});
+
+    const auto producer = [](coro::queue<int>& q, std::unique_ptr<coro::thread_pool>& exec) -> coro::task<void>
+    {
+        auto r = co_await q.push(1);
+        REQUIRE(r == coro::queue_produce_result::produced);
+        co_await q.shutdown_drain(exec);
+        REQUIRE(q.is_shutdown());
+    };
+    const auto consumer = [](coro::queue<int>& q) -> coro::task<void>
+    {
+        co_await q.shutdown();
+        REQUIRE(q.is_shutdown());
+    };
+
+    std::ignore = coro::sync_wait(coro::when_all(exec->schedule(producer(q, exec)), exec->schedule(consumer(q))));
+    std::cerr << "END queue issue-401\n";
+}
+
 TEST_CASE("~queue", "[queue]")
 {
     std::cerr << "[~queue]\n\n";

test/test_ring_buffer.cpp

@@ -2,9 +2,7 @@
 
 #include <coro/coro.hpp>
 
-#include <chrono>
 #include <iostream>
-#include <thread>
 
 TEST_CASE("ring_buffer", "[ring_buffer]")
 {
@@ -413,6 +411,32 @@ TEST_CASE("ring_buffer issue-242 basic type", "[ring_buffer]")
     std::cerr << "END ring_buffer issue-242 basic type\n";
 }
 
+TEST_CASE("ring_buffer shutdown_drain non-empty consumer shutdown", "[ring_buffer]")
+{
+    std::cerr << "BEGIN ring_buffer issue-401\n";
+
+    coro::ring_buffer<int, 1> buffer;
+    auto                      exec = coro::thread_pool::make_unique(coro::thread_pool::options{.thread_count = 1});
+
+    const auto producer = [](coro::ring_buffer<int, 1>&          buffer,
+                             std::unique_ptr<coro::thread_pool>& exec) -> coro::task<void>
+    {
+        auto r = co_await buffer.produce(1);
+        REQUIRE(r == coro::ring_buffer_result::produce::produced);
+        co_await buffer.shutdown_drain(exec);
+        REQUIRE(buffer.is_shutdown());
+    };
+    const auto consumer = [](coro::ring_buffer<int, 1>& buffer) -> coro::task<void>
+    {
+        co_await buffer.shutdown();
+        REQUIRE(buffer.is_shutdown());
+    };
+
+    std::ignore =
+        coro::sync_wait(coro::when_all(exec->schedule(producer(buffer, exec)), exec->schedule(consumer(buffer))));
+    std::cerr << "END ring_buffer issue-401\n";
+}
+
 TEST_CASE("~ring_buffer", "[ring_buffer]")
 {
     std::cerr << "[~ring_buffer]\n\n";