为啥 boost::asio::async_write 第一次运行良好，第二次出现问题？

Posted 2023-03-23

【中文标题】为啥 boost::asio::async_write 第一次运行良好，第二次出现问题？

【英文标题】：Why boost::asio::async_write works well at the first time and something goes wrong for the second time?为什么 boost::asio::async_write 第一次运行良好，第二次出现问题？

【发布时间】：2021-09-09 09:12:09

【问题描述】：

一开始调用async_write可以通过TCP成功发送数据，但是再次调用async_write就出错了。 这是code snippet：

#include <algorithm>
#include <array>
#include <boost/asio.hpp>
#include <boost/range.hpp>
#include <chrono>
#include <iostream>
#include <random>
#include <vector>

const std::size_t buf_size = 500*1024;
const int test_cycles = 1.024e5;

namespace asio = boost::asio;

int main()

  std::vector<char> send_data(buf_size);

  std::vector<char> recv_buf(buf_size);

  asio::io_service ios;

  asio::ip::tcp::socket socket1(ios);
  asio::ip::tcp::socket socket2(ios);
  asio::ip::tcp::acceptor acceptor(ios, asio::ip::tcp::v4(), 55557);
  socket1.connect(asio::ip::address_v4::loopback(), 55557);
  acceptor.accept(socket2);

  for (std::size_t i = 0; i < 1; ++i)
  
      auto start = std::chrono::steady_clock::now();
      for(int j=0; j < test_cycles; ++j)
      
            size_t written_bytes = 0;
            auto to_send_data = send_data;
            asio::async_write(socket1,
                asio::dynamic_buffer(send_data),
                [&](auto ec, auto n)
                
                    if(!ec)
                    
                        std::cout << "successfully sent " << n << std::endl;
                    
                    else
                    
                        std::cout << ec.message() << std::endl;
                    

                    if(0==n)
                    
                        std::cout << "send error" << std::endl;
                    

                    written_bytes = n;
                );

            asio::async_read(socket2, asio::buffer(recv_buf),
                [&](auto ec, auto n)
                
                    if(!ec)
                    
                        //std::cout << "received " << n << std::endl;
                    
                    else
                    
                        std::cout << ec.message() << std::endl;
                    

                    if(0==n)
                    
                        std::cout << "received error" << std::endl;
                    

                    if(written_bytes != n)
                    
                        std::cout << "received is not same with the sent" << std::endl;
                    
                );
                
                ios.run();
                ios.reset();
        
        
        auto end = std::chrono::steady_clock::now();
        std::chrono::duration<float> elapsed = end - start;
        std::cout << elapsed.count() << " seconds\n";
        std::cout << (buf_size * test_cycles / elapsed.count() / 1024 / 1024/ 1024) << " GB/s\n";
  

这是输出：

successfully sent 512000
successfully sent 0
send error

一些提示

我找到了一个workaround方法，程序运行良好。下面是相关代码sn-p：

        auto to_send_data = send_data;
        asio::async_write(socket1,
            asio::dynamic_buffer(to_send_data),

为什么前面的代码sn-p会出错？

更新：

我尝试通过 VsCode IDE 在 STD::vector::resize() 的实现中设置断点（我在 Ubuntu 上做了这个测试。），但断点确实不起作用（即断点是灰色的。） .我可以保证二进制程序是作为调试模式构建的。我也尝试通过 GDB 设置断点，但是 GDB 输出“Function "std::vector::resize" not defined。”

我在前面提到的operator()的实现中设置了断点，我发现default确实never被触发了，也就是说start总是1。

【参考方案1】：

不幸的是boost documentation on dynamic_buffer 非常简洁：

动态缓冲区封装了可以根据需要自动调整大小的内存存储。

意味着 dynamic_buffer 将在 IO 操作期间操纵底层向量。

Boost ASIO tutorial 更明确：

动态缓冲是一个概念。动态缓冲区是您可以将数据写入或读取的缓冲区。如果缓冲区不足以容纳您的数据，那么它将动态调整大小（增长）。因此，当您写入动态缓冲区时，您不必担心缓冲区中是否有足够的空间。另一方面，当您从动态缓冲区读取数据时，您有责任丢弃（消耗）已读取且不再需要的字节，因此缓冲区不会永久增长。

在 boost 调用中动态缓冲区的写入操作 consume 的 dynamic_buffer 方法从其中删除 bytes_tranferred：

class write_dynbuf_v2_op
   ....
   void operator()(const boost::system::error_code& ec,
       std::size_t bytes_transferred, int start = 0)
   
     switch (start)
    
      case 1:
         // write operation
         async_write(stream_, buffers_.data(0, buffers_.size()),
               BOOST_ASIO_MOVE_CAST(CompletionCondition)(completion_condition_),
               BOOST_ASIO_MOVE_CAST(write_dynbuf_v2_op)(*this));
        return;
     default:
        // write completed, consume transferred bytes and call handler
        buffers_.consume(bytes_transferred);
        handler_(ec, static_cast<const std::size_t&>(bytes_transferred));
  

因此在原始示例中，我们将在向量上排队异步写入，然后在 same 向量上一次又一次地排队异步操作。 在第一个回调完成后，传输了 512000 个字节，第二个回调有一个空向量（因为它已被 dynamic_buffer 擦除）并且回调打印

发送成功0

因为错误码是0然后

发送错误

send error 在传输的字节数为 0 时打印。

io_context.run() 被卡住，因为它有一个挂起的读取操作。

当我们为每个 async_write 操作安排向量的副本时，该解决方法会有所帮助。 另一种方法是在 io_context.run() 之后将向量调整回原始大小

【讨论】：

是，the size of the vector is zero after it has been sent out by async_write()。我不太了解有关 write_dynbuf_v2_op 实现的代码 sn-p（例如：case 1 是什么？）。能否请您为我详细解释一下？

这是来自 boost 实现的引用。异步 io 写入的实现经历了两个阶段 - int start=0 case 1 实际上应该是 bool writeBytes=false 和 if (writeBytes) 和 default 情况，在实际传输字节时处理回调。 default 案例擦除部分动态缓冲区并调用提供给原始 async_write 调用的处理程序

异步写入由两个阶段组成 - 调用操作系统/库函数在某处写入一些字节，以及在操作完成时调用的回调。在 boost 中，它们由同一个操作员处理。 Boost 在 (int start =1) 传输字节时会调用自身或操作系统，否则将调用回调。

1.如果我理解正确的话，应该依次进行两个阶段，第一，调用操作系统/库函数进行写入，第二，在写入函数返回后调用回调函数。我对吗？ 2.据我所知，Linux上的write()函数并不保证数据已经送出，它只是在数据写入到相应的驱动时才返回。你同意我的观点吗？

回调中的断点将引导您到正确的operator()，因为它是直接从它调用的。确实从未调用过调整大小，因为 consume() 使用 erase()