boost::asio 兼容等待条件变量

Question

我想创建一个与

boost::asio::awaitable

兼容的类，允许异步等待条件变量。

这是我的尝试，但老实说我不知道我在做什么

template <class T>
class ConditionVariableAwaiter : public boost::asio::awaitable<T> {
public:
   ConditionVariableAwaiter(std::condition_variable& cv, std::mutex& mtx, std::function<bool(void)> && ready, const std::chrono::seconds timeout = std::chrono::seconds(0))
      : _cv(cv), _mutex(mtx), _ready(std::move(ready)), _handle(nullptr), _timeout(timeout) {}

   ConditionVariableAwaiter(ConditionVariableAwaiter && other) noexcept :
   _cv(std::move(other._cv)),
   _mutex(std::move(other._cv)),
   _ready(std::move(other._ready)),
   _handle(std::move(other._handle)),
   _timeout(std::move(other._timeout)) {}

   ConditionVariableAwaiter &operator=(ConditionVariableAwaiter && other) noexcept {
      _cv = std::move(other._cv);
      _mutex = std::move(other._cv);
      _ready = std::move(other._ready);
      _handle = std::move(other._handle);
      _timeout = std::move(other._timeout);
   }

   ConditionVariableAwaiter(const ConditionVariableAwaiter & other) = delete;
   ConditionVariableAwaiter &operator=(const ConditionVariableAwaiter & other) = delete;

   bool await_ready() const noexcept {
      return _ready();
   }

   void await_suspend(std::coroutine_handle<> handle) {
      std::unique_lock lock(_mutex);
      _handle = handle;
      if (_timeout.count() == 0) {
         _cv.wait(lock, [this] { return _ready(); });
         _handle.resume();
      }
      else {
         if (_cv.wait_for(lock, _timeout, [this] { return _ready(); })) {
            _handle.resume();
         }
         else {
            _timedOut = true;
            _handle.resume();
         }
      }
   }

   bool await_resume() const noexcept {
      return _timedOut;
   }

   void notify() {
      std::lock_guard lock(_mutex);
      _cv.notify_all();
      if (_handle && _ready()) {
         _handle.resume();
      }
   }

   bool timedOut() const {
      return _timedOut;
   }

private:
   std::condition_variable& _cv;
   std::mutex& _mutex;
   std::function<bool(void)> _ready;
   std::coroutine_handle<> _handle;
   std::chrono::seconds _timeout;
   bool _timedOut = false;
};

当我尝试在函数中调用 co_await 时

boost::asio::awaitable<std::optional<ValueType>> waitCoro(const KeyType &key)

，我收到错误：

error: use of deleted function ‘boost::asio::awaitable<T, Executor>::awaitable(const boost::asio::awaitable<T, Executor>&) [with T = std::optional<int>; Executor = boost::asio::any_io_executor]’
[build]   143 |          co_await awaiter;

所讨论的等待变量是这样创建的：

ConditionVariableAwaiter<std::optional<ValueType>> awaiter(_cv, _mutex, [this] { return !_map.empty() || _cancelled; });

编辑

因此，我没有尝试在条件变量上进行等待（正如有人指出的那样，对 wait 的调用仍然是阻塞的），而是使用 boost stable_timers 来异步等待，然后当我需要向等待线程发出信号以唤醒时，我只需取消定时器。

唯一的小问题是

co_await _timer->async_wait(boost::asio::use_awaitable);

必须包含在 try catch 中，因为取消时它会抛出 operation_aborted 。有什么办法可以避免异常吗？我尝试传递一个 system_error 对象，但它不喜欢它。

Answer 1

唯一的小问题是

co_await _timer->async_wait(boost::asio::use_awaitable);

必须包含在 try catch 中，因为取消时它会抛出 operation_aborted。有什么办法可以避免异常吗？我尝试传递一个 system_error 对象，但它不喜欢它。

您可以使用

asio::as_tuple

或

asio::redirect_error

等令牌适配器来获取错误代码，而不是让

co_await

抛出

system_error

异常。

关于使用计时器：请务必注意竞争条件窗口：安全取消 boost asio 截止时间计时器