channel fixes.

Author: klemens-morgenstern

include/boost/cobalt/impl/channel.hpp

@@ -165,17 +165,21 @@ system::result<T> channel<T>::read_op::await_resume(const struct as_result_tag &
   if (cancelled)
    return {system::in_place_error, asio::error::operation_aborted};
 
-  T value = direct ? std::move(*direct) : std::move(chn->buffer_.front());
-  if (!direct)
+  T value = chn->buffer_.empty() ? std::move(*direct) : std::move(chn->buffer_.front());
+  if (!chn->buffer_.empty())
+  {
     chn->buffer_.pop_front();
+    if (direct)
+      chn->buffer_.push_back(std::move(*direct));
+  }
 
   if (!chn->write_queue_.empty())
   {
     auto &op = chn->write_queue_.front();
     BOOST_ASSERT(chn->read_queue_.empty());
     if (op.await_ready())
     {
-      op.transactional_unlink();
+      op.unlink();
       BOOST_ASSERT(op.awaited_from);
       asio::post(chn->executor_, std::move(op.awaited_from));
     }
@@ -211,6 +215,7 @@ std::coroutine_handle<void> channel<T>::write_op::await_suspend(std::coroutine_h
   if constexpr (requires (Promise p) {p.begin_transaction();})
     begin_transaction = +[](void * p){std::coroutine_handle<Promise>::from_address(p).promise().begin_transaction();};
 
+  BOOST_ASSERT(this->chn->buffer_.full());
   if (chn->read_queue_.empty())
   {
     chn->write_queue_.push_back(*this);
@@ -259,7 +264,7 @@ system::result<void>  channel<T>::write_op::await_resume(const struct as_result_
   if (cancel_slot.is_connected())
     cancel_slot.clear();
   if (cancelled)
-    boost::throw_exception(system::system_error(asio::error::operation_aborted), loc);
+    return {system::in_place_error, asio::error::operation_aborted};
 
   if (!direct)
   {
@@ -281,7 +286,8 @@ system::result<void>  channel<T>::write_op::await_resume(const struct as_result_
     BOOST_ASSERT(chn->write_queue_.empty());
     if (op.await_ready())
     {
-      op.transactional_unlink();
+      // unlink?
+      op.unlink();
       BOOST_ASSERT(op.awaited_from);
       asio::post(chn->executor_, std::move(op.awaited_from));
     }
@@ -297,7 +303,8 @@ struct channel<void>::read_op::cancel_impl
   {
     op->cancelled = true;
     op->unlink();
-    asio::post(op->chn->executor_, std::move(op->awaited_from));
+    if (op->awaited_from)
+      asio::post(op->chn->executor_, std::move(op->awaited_from));
     op->cancel_slot.clear();
   }
 };
@@ -310,7 +317,8 @@ struct channel<void>::write_op::cancel_impl
   {
     op->cancelled = true;
     op->unlink();
-    asio::post(op->chn->executor_, std::move(op->awaited_from));
+    if (op->awaited_from)
+      asio::post(op->chn->executor_, std::move(op->awaited_from));
     op->cancel_slot.clear();
   }
 };
@@ -322,6 +330,8 @@ std::coroutine_handle<void> channel<void>::read_op::await_suspend(std::coroutine
     if ((cancel_slot = h.promise().get_cancellation_slot()).is_connected())
       cancel_slot.emplace<cancel_impl>(this);
 
+  if (awaited_from)
+    boost::throw_exception(std::runtime_error("already-awaited"), loc);
   awaited_from.reset(h.address());
 
   if constexpr (requires (Promise p) {p.begin_transaction();})

src/channel.cpp

@@ -62,8 +62,7 @@ system::result<void>  channel<void>::read_op::await_resume(const struct as_resul
     {
       op.unlink();
       BOOST_ASSERT(op.awaited_from);
-      asio::post(
-          chn->executor_, std::move(op.awaited_from));
+      asio::post(chn->executor_, std::move(op.awaited_from));
     }
   }
   return {system::in_place_value};
@@ -97,8 +96,7 @@ system::result<void> channel<void>::write_op::await_resume(const struct as_resul
     {
       op.unlink();
       BOOST_ASSERT(op.awaited_from);
-      asio::post(
-          chn->executor_, std::move(op.awaited_from));
+      asio::post(chn->executor_, std::move(op.awaited_from));
     }
   }
   return {system::in_place_value};

test/channel.cpp

@@ -363,4 +363,32 @@ CO_TEST_CASE(data_loss)
   }
 }
 
+
+
+CO_TEST_CASE(interrupt_1)
+{
+  cobalt::channel<int> c{1u};
+
+  auto lr = co_await cobalt::left_race(c.write(42), c.read());
+  BOOST_CHECK(lr.index() == 0);
+  BOOST_CHECK(c.read().await_ready());
+  BOOST_CHECK(!c.write(12).await_ready());
+  lr = co_await cobalt::left_race(c.write(43), c.read());
+  BOOST_CHECK(lr.index() == 1);
+  BOOST_CHECK(get<1u>(lr) == 42);
+  auto rl =  co_await cobalt::left_race(c.read(), c.write(42));
+  BOOST_CHECK(rl.index() == 1);
+}
+
+CO_TEST_CASE(interrupt_void_1)
+{
+  cobalt::channel<void> c{1};
+  auto lr = co_await cobalt::left_race(c.write(), c.read());
+  BOOST_CHECK(lr == 0);
+  lr = co_await cobalt::left_race(c.write(), c.read());
+  BOOST_CHECK(lr == 1);
+  auto rl =  co_await cobalt::left_race(c.read(), c.write());
+  BOOST_CHECK(rl == 1);
+}
+
 }