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).

Suraaj