Boost ASIO: Problems building a conditional variable using io_context / strand

From: Suraaj K S (kssuraaj28_at_[hidden])
Date: 2023-05-22 22:39:33

Hi all,

In boost-asio, I realized that there was no easy way to have something that
resembles a condition variable.

However, I realized that I could get something very similar using stackful
coroutines (
https://www.boost.org/doc/libs/1_66_0/doc/html/boost_asio/overview/core/spawn.html),
while using an io_context as a 'queue' for completion tokens. Below is my
approach:

*#include "asio_types.hpp"#include <atomic>#include <iostream>class
async_pending_queue {public:  async_pending_queue()      :
pending_handlers_(0), strand_(pending_queue_),
wg_(asio::make_work_guard(pending_queue_)) {}  template <typename
CompletionToken>  auto async_submit(      CompletionToken &&token,
std::function<void(void)> atomic_action = [] {}) {    auto init = [this,
&atomic_action](auto completion_handler) {      auto posted_lambda =
[handler = std::move(completion_handler),
this]() mutable {        pending_handlers_--;        asio_sys_err ec;
  handler(ec);        };      post(strand_,std::move(posted_lambda));
pending_handlers_++;      atomic_action();    };    return
asio::async_initiate<CompletionToken, void(asio_sys_err)>(init,
                                                         token);  }  int
pending_count() { return pending_handlers_.load(); }  // It may not run 1
and run 0  bool try_run_one() {    auto cnt = pending_queue_.poll_one();
std::cout << "completion token result" << cnt << std::endl;    bool ret =
(cnt == 1);    return ret;  }private:  std::atomic<unsigned int>
pending_handlers_;  asio_ioctx pending_queue_;  asio_ioctx::strand
strand_;  decltype(asio::make_work_guard(pending_queue_)) wg_;};*
Here, one simply uses calls `my_async_pending_queue.async_submit(yield)`,
if calling from a stackful coroutine. The coroutine can be continued by
calling `my_async_pending_queue.try_run_one()`.

Using this, I wanted to build a 'memory checker'. It has two functions ->
`request_space` and `free_space`. A coroutine calls `request_space`, which
may block if there is no space left. Meanwhile, another thread / coroutine
can call `free_space`, which will run blocked coroutines if possible.

I built a toy memory checker wrapper as follows:

*#ifndef MEM_CHECK_HPP#define MEM_CHECK_HPP#include <cstddef>#include
<mutex>#include <queue>#include <boost/asio.hpp>#include
<boost/asio/spawn.hpp>namespace asio = boost::asio;using asio_ioctx =
asio::io_context;using asio_sys_err = boost::system::error_code;using
asio::yield_context;#include "async_pending_queue.hpp"class MemoryChecker
{public:  using bytes_cnt = size_t;  MemoryChecker(asio_ioctx &ioctx,
bytes_cnt total_mem = 1024, bytes_cnt initial_fill = 0);  // This requests
some space, and possibly yields;  void request_space(bytes_cnt cnt,
yield_context yield);  void free_space(bytes_cnt cnt);private:  bytes_cnt
get_available_mem();  const bytes_cnt total_mem_;  bytes_cnt mem_used_;
std::queue<bytes_cnt> request_queue_;  async_pending_queue
request_routines_;  std::mutex lock_;  asio_ioctx::strand fifo_strand_;
asio_ioctx &completion_ioctx_;};inline
MemoryChecker::MemoryChecker(asio_ioctx &ioctx, bytes_cnt total_mem,
                     bytes_cnt initial_fill)    : total_mem_(total_mem),
mem_used_(initial_fill), request_queue_{}, lock_{},
request_routines_{}, completion_ioctx_(ioctx), fifo_strand_(ioctx) {}inline
MemoryChecker::bytes_cnt MemoryChecker::get_available_mem() {
assert(total_mem_ >= mem_used_);  return total_mem_ - mem_used_;}inline
void MemoryChecker::request_space(bytes_cnt cnt, yield_context yield) {  //
if (cnt > total_mem_) throw logic_error  lock_.lock();  assert(cnt <=
total_mem_);  assert(cnt > 0);  if (request_queue_.empty()) {
assert(request_routines_.pending_count() == 0);    if (get_available_mem()
>= cnt) {      // We bypass the pending queue      mem_used_ += cnt;
lock_.unlock();      return;    }  }  assert(request_queue_.size() ==
request_routines_.pending_count());  std::cout << "Pushing " << cnt <<
std::endl;  request_queue_.push(cnt);  auto wg =
asio::make_work_guard(completion_ioctx_);
request_routines_.async_submit(yield, [this] { lock_.unlock(); });  auto
oldest_req{request_queue_.front()};  assert(cnt == oldest_req);
request_queue_.pop();   mem_used_ += cnt;  assert(request_queue_.size() ==
request_routines_.pending_count());  asio::post(fifo_strand_,
yield);}inline void MemoryChecker::free_space(bytes_cnt cnt) {  {
std::lock_guard<std::mutex> lg{lock_};    mem_used_ -= cnt;    // Here, we
own the lock, and free as many coroutines as we can    while (true) {
if (request_queue_.size() == 0) {        std::cout << "No pending requests.
Bailing" << std::endl;        break;      }
assert(request_queue_.size() == request_routines_.pending_count());
auto oldest_req{request_queue_.front()};      auto
available_mem{get_available_mem()};      if (available_mem < oldest_req) {
      std::cout << "Oldest request is larger than available_mem. Bailing"
<< std::endl;        break;      }
assert(request_routines_.try_run_one() == true);    }  }}#endif /*
MEM_CHECK_HPP */*

Here is a test program that can run it:

*#include "mem_check.hpp"#include <thread>#include <unistd.h>constexpr
size_t mc_size{4};asio_ioctx ioctx;size_t total{0};MemoryChecker mc{ioctx,
mc_size};void requestor_coroutine(size_t rq,yield_context yield) {
asio::steady_timer t(ioctx);  while (true) {    total += rq;
mc.request_space(rq, yield);    std::cout << "Got requested space ";
asio_sys_err ec;  }}int main() {  asio::spawn(ioctx, [](yield_context
yield) { requestor_coroutine(1,yield); });  asio::spawn(ioctx,
[](yield_context yield) { requestor_coroutine(2,yield); });
asio::spawn(ioctx, [](yield_context yield) { requestor_coroutine(3,yield);
});  asio::spawn(ioctx, [](yield_context yield) {
requestor_coroutine(4,yield); });  std::thread t([] { ioctx.run(); });
while (true) {    getchar();    std::cout << total << std::endl;    if
(total > 0) {      std::cout << "freeing" << std::endl;
mc.free_space(1);      total -= 1;    }  }  t.join();}*

Finally, the problem we face is as follows. When we run the program, the
assertion `total_mem >= mem_used_` fails. On some further investigation, I
realized that our completion token was being called even when we do not
call `try_run_one`, which was very weird.

Finally, somewhat more surprisingly, If I replace
*post(stand_,std::move(posted_lambda));* by
*post(pending_queue_,std::move(posted_lambda));*, things seem to work.
However, the asio documentation says that only strands guarantee a FIFO
execution order. I am not sure if using a simple `io_context` will work as
a FIFO queue (even though it seems to in these examples).

Any inputs would be helpful - I am happy to hear the problem in this
implementation, as well as other implementations (for example, using a
std::queue as a proper queue instead of this io_context hack).

This question has also been posted here:
https://stackoverflow.com/questions/76310252/c-boost-asio-building-a-conditional-variable-using-io-context-strand
.

Thanks,
Suraaj