c++ - boost asio tcp - 套接字写入数据不同于缓冲区中的数据 - 某处可能存在线程不安全

Question

我正在探索 boost asio 产品

客户端发送一个 1 字节的 header ，指示后面的字节长度。

相关服务器代码:

  enum {max_length=1};     

  void handle_read(const boost::system::error_code & error, const size_t & bytes_transferred){
if (! error){
  ++ctr;
  std::string inc_data_str(this->inc_data.begin(),this->inc_data.end());
  std::cout<<"received string: "<<inc_data_str<<" with size "<<inc_data_str.size()
           <<" bytes_transferred: "<<bytes_transferred<<" ctr: "<<ctr<<std::endl;
  int size_inc_next = boost::lexical_cast<int>(inc_data_str);
  int offset = 0;
  //std::cout<<"incoming integer of size "<<size_inc_next<<" processed from string: "<<inc_data_str<<std::endl;                    
  std::vector<char> next_inc_data(size_inc_next+offset);
  boost::asio::read(this->socket,boost::asio::buffer(next_inc_data),boost::asio::transfer_exactly(size_inc_next+offset));
  std::string int_recvd(next_inc_data.begin(),next_inc_data.begin()+size_inc_next);
  //std::cout<<boost::posix_time::microsec_clock::local_time()<<std::endl;                                                         
  //std::cout<<"received integer: "<<int_recvd<<" from string "<<int_recvd<<" of size "<<int_recvd.size()<<std::endl;              
  this->process_connection();
 } // ! error
} // handle_read

void process_connection(){
                                                   boost::asio::async_read(this->socket,boost::asio::buffer(this->inc_data),boost::asio::transfer_exactly(max_length),
                    boost::bind(&Connection::handle_read,shared_from_this(),boost::asio::placeholders::error,
                                    boost::asio::placeholders::bytes_transferred));
}

相关客户端代码:

  void on_write(const boost::system::error_code & error_code){
if (! error_code){
  std::string transfer_data("15");
  std::vector<char> v_td(transfer_data.begin(),transfer_data.end());
  ++ctr;
  for (std::vector<char>::iterator iter = v_td.begin(); iter != v_td.end(); ++iter) std::cout<<*iter;
  std::cout<<" ctr: "<<ctr;
  std::endl(std::cout);
        boost::asio::async_write(this->socket,boost::asio::buffer(v_td),boost::asio::transfer_exactly(2),
                           boost::bind(&Client::on_write,shared_from_this(),
                                       boost::asio::placeholders::error));
  }
 }

服务器进程的预期示例打印输出:

  received string: 1 with size 1 bytes_transferred: 1 ctr: 159685

客户端进程的预期示例打印输出:

  15 ctr: 356293

这样的预期输出会产生一段时间，但是在 356293 次客户端迭代之后(这个 ctr 数字从重复的进程试验中肉眼看是不确定的)，服务器中断并出现以下错误:

 received string:  with size 1 bytes_transferred: 1 ctr: 159686
 terminate called after throwing an instance of 'boost::exception_detail::clone_impl<boost::exception_detail::error_info_injector<boost::bad_lexical_cast> >'
 what():  bad lexical cast: source type value could not be interpreted as target

中止(核心转储)

请注意，收到的字符串是“空白”。有时它也会因替代消息而中断:

 received string: X with size 1 bytes_transferred: 1 ctr: 159686

这是怎么回事，为什么以及我该如何解决？

strace 之后的进一步编辑:

客户端跟踪:

 sendmsg(6, {msg_name(0)=NULL, msg_iov(1)=[{"15", 2}], msg_controllen=0, msg_flags=0}, MSG_NOSIGNAL) = 2
 epoll_wait(4, {}, 128, 0)               = 0
 write(1, "15 ctr: 204441\n", 1515 ctr: 204441)        = 15
 sendmsg(6, {msg_name(0)=NULL, msg_iov(1)=[{"15", 2}], msg_controllen=0, msg_flags=0},    MSG_NOSIGNAL) = 2
 epoll_wait(4, {}, 128, 0)               = 0
 write(1, "15 ctr: 204442\n", 1515 ctr: 204442)        = 15
 sendmsg(6, {msg_name(0)=NULL, msg_iov(1)=[{"15", 2}], msg_controllen=0, msg_flags=0}, MSG_NOSIGNAL) = -1 EAGAIN (Resource temporarily \
 unavailable)
 epoll_wait(4, {{EPOLLOUT, {u32=167539936, u64=167539936}}}, 128, -1) = 1
 sendmsg(6, {msg_name(0)=NULL, msg_iov(1)=[{"\0\0", 2}], msg_controllen=0,   msg_flags=0}, MSG_NOSIGNAL) = 2
 write(1, "15 ctr: 204443\n", 1515 ctr: 204443)        = 15
 sendmsg(6, {msg_name(0)=NULL, msg_iov(1)=[{"15", 2}], msg_controllen=0, msg_flags=0},     MSG_NOSIGNAL) = 2
 epoll_wait(4, {}, 128, 0)               = 0
 write(1, "15 ctr: 204444\n", 1515 ctr: 204444)        = 15
 sendmsg(6, {msg_name(0)=NULL, msg_iov(1)=[{"15", 2}], msg_controllen=0, msg_flags=0}, MSG_NOSIGNAL) = 2
 epoll_wait(4, {}, 128, 0)               = 0
 write(1, "15 ctr: 204445\n", 1515 ctr: 204445)        = 15

服务器跟踪:

 write(1, "received string: 1 with size 1 b"..., 64received string: 1 with size 1   bytes_transferred: 1 ctr: 204441) = 64
 write(1, "incoming integer of size 1 proce"..., 52incoming integer of size 1 processed from string: 1) = 52
 recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"5", 1}], msg_controllen=0, msg_flags=0},0) = 1
 write(1, "received integer: 5 from string "..., 44received integer: 5 from string 5 of size 1) = 44
 recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"1", 1}], msg_controllen=0, msg_flags=0},0) = 1
 epoll_wait(4, {}, 128, 0)               = 0
 write(1, "received string: 1 with size 1 b"..., 64received string: 1 with size 1 bytes_transferred: 1 ctr: 204442) = 64
 write(1, "incoming integer of size 1 proce"..., 52incoming integer of size 1 processed from string: 1) = 52
 recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"5", 1}], msg_controllen=0, msg_flags=0}, 0) = 1
 write(1, "received integer: 5 from string "..., 44received integer: 5 from string 5 of size 1) = 44
 recvmsg(7, {msg_name(0)=NULL, msg_iov(1)=[{"\0", 1}], msg_controllen=0, msg_flags=0}, 0) = 1
 epoll_wait(4, {}, 128, 0)               = 0
 write(1, "received string: \0 with size 1 b"..., 64received string: ^@ with size 1 bytes_transferred: 1 ctr: 204443) = 64
 futex(0xb76640fc, FUTEX_WAKE_PRIVATE, 2147483647) = 0
 write(1, "inc_data_str\n", 13inc_data_str)          = 13

对于客户端进程，错误的“\0\0”发送之前的 epoll_wait 与其他 epoll_wait 调用不同(u32=...., u64=....)...不知道这意味着什么

总结一下令人困惑的部分，strace 表示正在传输空值，但下一行的 strace 表示对标准输出的写系统调用，文字为“15”，这意味着这就是 transfer_data vector 中的内容

重新编辑:

最后我插入了一个

  boost::this_thread::sleep(boost::posix_time::microseconds(200));

就在客户端 on_write 函数中的 write 语句之前。

有了这个，没有遇到任何问题。那么 asio 对象会出现什么样的并发问题呢？是 socket 吗？

Answer 1

您的客户端因缓冲区生命周期而损坏

void 
on_write(
    const boost::system::error_code& error_code
)
{
    if ( !error_code ) {
        std::string transfer_data("15");
        std::vector<char> v_td(transfer_data.begin(), transfer_data.end());
        //                 ^
        //                  \------ goes out of scope before async_write() returns

        boost::asio::async_write(
            this->socket,
            boost::asio::buffer(v_td),
            boost::asio::transfer_exactly(2),
            boost::bind( 
                &Client::on_write,
                shared_from_this(),
                boost::asio::placeholders::error
            )
        );
     }
 }

您需要确保给async_write() 的缓冲区在调用完成处理程序之前保持有效:

buffers One or more buffers containing the data to be written. Although the buffers object may be copied as necessary, ownership of the underlying memory blocks is retained by the caller, which must guarantee that they remain valid until the handler is called. handler

---