c++ - 并发异步写入。有免等待的解决方案吗？

Question

async_write() 禁止从不同线程同时调用。它使用 async_write_some 按 block 发送数据，这样的 block 可以交错。因此，用户需要注意不要同时调用 async_write()。

有比这个伪代码更好的解决方案吗？

void send(shared_ptr<char> p) {
  boost::mutex::scoped_lock lock(m_write_mutex);
  async_write(p, handler);
}

我不喜欢长时间阻塞其他线程的想法(我的应用程序中有大约 50Mb 的发送)。

这样的东西可能行得通吗？

void handler(const boost::system::error_code& e) {
  if(!e) {
    bool empty = lockfree_pop_front(m_queue);
    if(!empty) {
      shared_ptr<char> p = lockfree_queue_get_first(m_queue);
      async_write(p, handler);
    }
  }
}

void send(shared_ptr<char> p) {
  bool q_was_empty = lockfree_queue_push_back(m_queue, p)
  if(q_was_empty)
    async_write(p, handler);
}

我更愿意找到一个现成的食谱食谱。处理无锁并不容易，会出现很多细微的错误。

Answer 1

async_write() is forbidden to be called concurrently from different threads

这个说法不太正确。应用程序可以自由地并发调用 async_write，只要它们在不同的 socket 对象上即可。

Is there a nicer solution than this pseudocode?

void send(shared_ptr<char> p) {
  boost::mutex::scoped_lock lock(m_write_mutex);
  async_write(p, handler);
}

这可能无法实现您的预​​期，因为 async_write 会立即返回。如果您打算在整个写操作期间锁定互斥量，则需要将 scoped_lock 保持在范围内，直到调用完成处理程序。

这个问题有更好的解决方案，该库使用 strand 的概念内置支持.很适合这种场景。

A strand is defined as a strictly sequential invocation of event handlers (i.e. no concurrent invocation). Use of strands allows execution of code in a multithreaded program without the need for explicit locking (e.g. using mutexes).

在此处使用显式链将确保您的处理程序仅由调用了 io_service::run() 的单个线程调用。在您的示例中，m_queue 成员将受到链的保护，确保对传出消息队列的原子访问。将条目添加到队列后，如果大小为 1，则表示没有未完成的 async_write 操作正在进行，应用程序可以通过链启动一个包装。如果队列大小大于 1，应用程序应等待 async_write 完成。在 async_write 完成处理程序中，从队列中弹出一个条目并根据需要处理任何错误。如果队列不为空，完成处理程序应该从队列的前面启动另一个 async_write。

这是一种更简洁的设计，它在您的类中散布了互斥锁，因为它按预期使用了内置的 Asio 结构。这other answer I wrote有一些代码实现了这个设计。