c++ - boost asio async_receive_from() 当帧连续发送时丢失 udp 帧数据检索
问题描述
下面从真实代码库简化的 MVCE 显示了同样的问题。
服务器连续发送 5 个 UDP 帧的“突发”,其中填充了 150 个字节的值 0xA5,其间延迟很小或没有延迟。暂停 1 秒。
客户端将boost::asio async_receive_from()函数与 1 秒计时器并行使用。客户端工作得相对较好,除非 UDP 帧之间的延迟“太”小。似乎检索到了正确的大小(此处为 150 字节),但缓冲区/向量似乎没有更新。
- 5 x 150 字节的 UDP 帧似乎并不多。
- Wireshark 确实看到发送的完整和正确的帧。
- 如果我使用同步 boost asio 套接字同步 receive_from() 我没有遇到任何问题
我尝试了大约六次来深入研究 boost asio,但没有成功找到一个事实或理由。SO上的相同帖子显示了非常不同的代码,因此很难将它们转换为当前代码
这是代码 客户端(client_with_timer.cc)
#include <iostream>
#include <vector>
#include <string>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
using namespace boost::asio;
void asyncReadHandler( const boost::system::error_code& error, std::size_t bytesTransferred );
void timeoutHandler( const boost::system::error_code& error, bool* ptime_out );
size_t ReceivedDataSize;
std::string ReadError;
int main(int argc, char * argv[])
{
io_service io;
ip::udp::socket socket(io, ip::udp::endpoint(ip::udp::v4(), 1620));
size_t num = 0;
while (true)
{
std::vector<unsigned char> vec(1500);
ip::udp::endpoint from;
socket.async_receive_from(
boost::asio::buffer( vec ),
from,
boost::bind(
asyncReadHandler,
boost::asio::placeholders::error,
boost::asio::placeholders::bytes_transferred ) );
bool timeout = false;
ReceivedDataSize = 0;
ReadError = "";
// Creating and starting timer (by registering timeout handler)
deadline_timer timer( io, boost::posix_time::seconds( 1 ) );
timer.async_wait(
boost::bind( timeoutHandler, boost::asio::placeholders::error, &timeout ) );
// Resetting IO service instance
io.reset();
while(io.run_one())
{
if ( timeout ) {
socket.cancel();
timer.cancel();
//Leave the io run_one loop
break;
}
else if ( (0 != ReceivedDataSize ) || (!ReadError.empty())) {
timer.cancel();
socket.cancel();
std::cout << "Received n°" << num++ << ": " << ReceivedDataSize << "\r" << std::flush;
if (0 != ReceivedDataSize )
vec.resize(ReceivedDataSize);
if (!ReadError.empty())
std::cout << "Error: " << ReadError << std::endl;
bool result = true;
for ( auto x : vec )
if ( 0xA5 != x ) { result = false; break; }
if ( false == result ) {
std::cout << std::endl << "Bad reception" << std::endl << std::hex;
for ( auto x : vec )
std::cout << (int)x << " ";
std::cout << std::dec << "\n";
}
//Leave the io run_one loop
break;
}
else {
//What shall I do here ???
//another potential io.reset () did not bring much
}
}
}
return 0;
}
void asyncReadHandler( const boost::system::error_code& error, std::size_t bytesTransferred )
{
// If read canceled, simply returning...
if( error == boost::asio::error::operation_aborted ) return;
ReceivedDataSize = 0;
// If no error
if( !error ) {
ReceivedDataSize = bytesTransferred;
}
else {
ReadError = error.message();
}
}
void timeoutHandler( const boost::system::error_code& error, bool* ptime_out )
{
// If timer canceled, simply returning...
if( error == boost::asio::error::operation_aborted ) return;
// Setting timeout flag
*ptime_out = true;
}
这是服务器(server.cc),因此您不必自己动手
#include <iostream>
#include <vector>
#include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <unistd.h>
using namespace boost::asio;
int main(int argc, char * argv[])
{
io_service io;
ip::udp::socket socket(io, ip::udp::endpoint(ip::udp::v4(), 0));
std::vector<char> vec(150,0xA5);
#if 1
int separator = 1 * 1000;
#else
int separator = 0;
#endif
while (true)
{
socket.send_to(buffer(vec), ip::udp::endpoint(ip::udp::v4(), 1620));
if ( separator ) usleep(separator);
socket.send_to(buffer(vec), ip::udp::endpoint(ip::udp::v4(), 1620));
if ( separator ) usleep(separator);
socket.send_to(buffer(vec), ip::udp::endpoint(ip::udp::v4(), 1620));
if ( separator ) usleep(separator);
socket.send_to(buffer(vec), ip::udp::endpoint(ip::udp::v4(), 1620));
if ( separator ) usleep(separator);
socket.send_to(buffer(vec), ip::udp::endpoint(ip::udp::v4(), 1620));
usleep(1000*1000);
}
return 0;
}
我用下面的天真的命令编译了这两个:
g++ client_with_timer.cc -std=c++11 -O2 -Wall -o client_with_timer -lboost_system
g++ server.cc -std=c++11 -O2 -Wall -o server -lboost_system
当延迟太小时,它会产生如下输出
nils@localhost ASIO_C]$ ./client_with_timer
Received n°21: 150
Bad reception
0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
Received n°148: 150
Bad reception
0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
Received n°166: 150
Bad reception
0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0
Received n°194: 150
如何更正客户端代码以避免丢失帧?欢迎任何有助于更好地理解 boost asio 原理的提示
解决方案
我认为您的代码中存在数据竞争。如果在读取操作完成之前计时器已过期(发生超时),则执行以下代码:
if ( timeout ) {
socket.cancel();
timer.cancel();
//Leave the io run_one loop
break; // [1]
}
您正在从 while 循环中断,socket.cancel()
取消异步读取操作,其operation_aborted
错误处理程序排队并等待事件循环中的处理。因为你从 while 循环中跳出来,run_one
没有被调用,而且这个处理程序还在队列中。
io_service.reset()
不清除队列。中止操作的处理程序仍然存在。并等待被调用。reset()
仅将stopped
标志设置为io_service
to false
,然后处理程序可以通过调用处理.. 方法run_one
,one
您reset
用于从队列中恢复处理处理程序。
所以我们在队列中有未处理的处理程序,在 main while 循环vec
中创建了新向量,它的所有元素都初始化为 0。async_receive_from
启动(它正在读取vec
并ReceivedDataSize
在其处理程序中设置),然后reset
被调用,run_one
可以处理处理程序并调用处理程序以中止手术!并且您正在测试ReceivedDataSize
和vec
中止操作...但是您应该为上次启动的异步操作执行此操作。
我会将带有超时的子句重写为:
if ( timeout ) {
socket.cancel();
timer.cancel();
} // no break
删除 break 后,我们保证处理中止的操作,run_one
并且在启动新的异步操作时不会调用未完成的处理程序。bad reception
进行此修改后,我在测试您的代码时没有看到。
编辑
关于您的评论,是的,其他break
语句也应该从代码中删除。
程序的输出是不可预测的,因为您正在启动异步操作,该操作引用局部变量(vec
由 修改),处理程序排队,局部变量被销毁,然后在已经被销毁async_receive_from
的情况下从 io_service 调用处理程序。vec
你可以测试下面的代码,看看会发生什么:
boost::asio::io_context io; // alias on io_service
boost::asio::system_timer t1{io};
t1.expires_from_now(std::chrono::seconds(1));
boost::asio::system_timer t2{io};
t2.expires_from_now(std::chrono::seconds(1));
boost::asio::system_timer t3{io};
t3.expires_from_now(std::chrono::seconds(1));
t1.async_wait ([](const boost::system::error_code& ec){ cout << "[1]" << endl;});
t2.async_wait ([](const boost::system::error_code& ec){ cout << "[2]" << endl;});
t3.async_wait ([](const boost::system::error_code& ec){ cout << "[3]" << endl;});
// 3 handlers are queueud
cout << "num of handlers executed " << io.run_one() << endl; // wait for handler, print 1
io.reset(); // RESET is called
cout << "num of handlers executed " << io.run_one() << endl; // wait for handler, print 1
io.reset(); // RESET is called
cout << "num of handlers executed " << io.run_one() << endl; // wait for handler, print 1
cout << "executed: " << io.poll_one() << endl; // call handler if any ready, print 0
我们正在调用io_service::reset
,但所有处理程序都已执行。从代码中删除break
s 后,您确保将执行所有处理程序,并保证在调用这些处理程序时本地数据有效。
推荐阅读
- ldap - 无法创建新条目。PHPLDAP管理员
- android - 如何将当前数据从设置活动传递到火力基地
- javascript - firebase.auth().signInWithPopup() 不适用于 Linux/Mac 上的 Chrome 扩展
- python - Python paramiko chdir 不起作用。getcwd() 返回无
- android - Android AppAuth 将密码发送到 Chrome 选项卡
- sql - Hue 无法加载 Presto Schema dbc 不存在
- python - 与第二页不同的页面模板
- raspberry-pi3 - Raspberry pi 3 B 通过 RTSP (Wifi) 与无线 IP 摄像头连接
- php - 从 db (Codeigniter) 加载页面信息的重定向
- docker - 尝试连接到 Redis 会返回 ECONNREFUSED