boost asio async_receive_from() 当帧连续发送时丢失 udp 帧数据检索答案

【问题标题】：boost asio async_receive_from() missing udp frames data retrieval when frames sent consecutiveboost asio async_receive_from() 当帧连续发送时丢失 udp 帧数据检索
【发布时间】：2019-02-05 21:13:52
【问题描述】：

下面从真实代码库简化的 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 原理的提示

【问题讨论】：

客户端和服务器上的操作系统是什么？你用的是什么版本的 boost.asio？
@Oleg Andriyanov：是的，抱歉。在 RHEL 7 64 位上发现了问题，我在 CentOS 7 64 位上重现了它，两者都使用 boost 1.53

标签： c++ boost-asio

【解决方案1】：

我认为您的代码中存在数据竞争。如果在读取操作完成之前定时器已过期（发生超时），则执行以下代码：

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() 仅将io_service 的stopped 标志设置为false，然后可以通过调用run_one、one .. 方法处理处理程序，您正在使用reset 从队列中恢复处理处理程序。

所以我们在队列中有未处理的处理程序，在主循环中创建了新向量vec，它的所有元素都初始化为0。async_receive_from开始（它正在读入vec并在其中设置ReceivedDataSize handler)，然后reset被调用，run_one可以处理handler并调用handler进行中止操作！并且您正在测试 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，但所有处理程序都已执行。从您的代码中删除breaks 后，您可以确保所有处理程序都将被执行，并保证在调用这些处理程序时本地数据有效。

【讨论】：

我确认删除超时测试中的中断会使错误从 RHEL7 和 CentOS 7 上的 MVCE 中消失（两者都是 GCC 4.8.5）+ 也从真实（完整）代码中消失。非常感谢这一点，也感谢您的解释，这似乎比我阅读的所有（少数）文档更清楚。我仍然没有得到所有的故事，但是有了这些新知识，我将再次进一步阅读文档。暂时我只是想知道如果超时测试的“中断”无效，接收测试的第二个“中断”是否有效！？