延迟后未调用 async_write_some 回调

Posted 2023-02-21

【中文标题】延迟后未调用 async_write_some 回调

【英文标题】：async_write_some callback not called after delay

【发布时间】：2018-04-30 03:09:46

【问题描述】：

我对@9​​87654321@ 的回调在睡眠一秒后没有被调用。如果我为每次写入启动一个io_service 工作线程，为什么不调用回调？

标题

boost::system::error_code error_1;
boost::shared_ptr <boost::asio::io_service> io_service_1;
boost::shared_ptr <boost::asio::ip::tcp::socket> socket_1;

连接

void eth_socket::open_eth_socket (void)

    // 1. reset io services
    io_service_1.reset();
    io_service_1 = boost::make_shared <boost::asio::io_service> ();

    // 2. create endpoint
    boost::asio::ip::tcp::endpoint remote_endpoint(
        boost::asio::ip::address::from_string("10.0.0.3"), 
        socket_1_port
    );

    // 3. reset socket
    socket_1.reset(new boost::asio::ip::tcp::socket(*io_service_1));                

    // 4. connect socket
    socket_1->async_connect(remote_endpoint,
        boost::bind(
            &eth_socket::socket_1_connect_callback,
            this, boost::asio::placeholders::error
        )
    );

    // 5. start io_service_1 run thread after giving it work
    boost::thread t(boost::bind(&boost::asio::io_service::run, *&io_service_1));                
    return;

写

void eth_socket::write_data (std::string data)
   
    // 1. check socket status
    if (!socket_1->is_open())
    
        WARNING << "socket_1 is not open";
        throw -3;
    

    // 2. start asynchronous write
    socket_1->async_write_some(
        boost::asio::buffer(data.c_str(), data.size()),
        boost::bind(
            &eth_socket::socket_1_write_data_callback,
            this, boost::asio::placeholders::error, 
            boost::asio::placeholders::bytes_transferred
        )
    );

    // 3. start io_service_1 run thread after giving it work
    boost::thread t(boost::bind(&boost::asio::io_service::run, *&io_service_1));
    return;

回调

void eth_socket::socket_1_write_data_callback (const boost::system::error_code& error, size_t bytes_transferred)

    // 1. check for errors
    if (error) 
    
        ERROR << "error.message() >> " << error.message().c_str();
        return;
    
    if (socket_1.get() == NULL || !socket_1->is_open())
    
        WARNING << "serial_port_1 is not open";
        return;
    
    INFO << "data written to 10.0.0.3:1337 succeeded; bytes_transferred = " << bytes_transferred;
    return;

测试

open_eth_socket();
write_data("Hello");    // callback called
write_data("Hello");    // callback called
write_data("Hello");    // callback called
sleep(1);
write_data("Hello");    // callback not called after sleep

【参考方案1】：

boost::thread t(boost::bind(&boost::asio::io_service::run, *&io_service_1));                

这很奇怪，原因有很多。

您不应为每个操作“运行”io_services。相反，在可能发布操作的同时稳定地运行它们。可以选择使用io_service::work 来防止运行返回。 您不应该（必须）为每个操作创建线程。如果有的话，这是同步问题的秘诀 (Why do I need strand per connection when using boost::asio?) 当 io_service 返回后再次运行时（没有错误），您应该首先调用 reset()，根据文档 (Why must io_service::reset() be called?) 您破坏了一个未分离的线程 - 可能在它完成之前。如果您使用了std::thread，这甚至会导致程序立即异常终止。不加入非分离线程是不好的做法（我补充说，在线程终止时没有显式同步的情况下使用分离线程是很困难的）。见Why is destructor of boost::thread detaching joinable thread instead of calling terminate() as standard suggests?

我会添加到这些***关注点

使用诸如socket_1 之类的名称的气味（只需将其称为socket_ 并用描述性名称实例化另一个对象以包含另一个socket_）。我不确定，但这个问题确实让人怀疑这些甚至可能是全局变量。 （我希望不是这样） throw-ing 原始整数，真的吗？ 您在销毁 io_service 的同时从不检查工作线程是否已完成，这可能会导致数据争用。

更多Undefined Behaviour在这里：

_sock.async_write_some(
        ba::buffer(data.c_str(), data.size()),

您传递了对超出范围的参数data 的引用。当异步操作完成时，它将是一个悬空引用

这里有一些明显的复制/粘贴问题：

if (socket_1.get() == NULL || !socket_1->is_open())

    WARNING << "serial_port_1 is not open";
    return;

我实际上想说这源于导致变量名称为 serial_port_1 和 socket_1 的完全相同的来源

一些清理

简化。没有独立的代码，所以这里没有完整的，但至少看到了许多简化点：

Live On Coliru

#include <boost/asio.hpp>
#include <boost/thread.hpp>
#include <iostream>

namespace ba = boost::asio;
using ba::ip::tcp;
using boost::system::error_code;

#define ERROR   std::cerr
#define WARNING std::cerr
#define INFO    std::cerr

struct eth_socket 

    ~eth_socket() 
        _work.reset();
        if (_worker.joinable())
            _worker.join(); // wait
    

    void open(std::string address);
    void write_data(std::string data);

  private:
    void connected(error_code error) 
        if (error)
            ERROR << "Connect failed: " << error << "\n";
        else
            INFO << "Connected to " << _sock.remote_endpoint() << "\n";
    
    void written(error_code error, size_t bytes_transferred);

  private:
    ba::io_service _svc;
    boost::optional<ba::io_service::work> _work _svc ;
    boost::thread _worker [this]  _svc.run();  ;

    std::string _data;

    unsigned short _port = 6767;
    tcp::socket _sock _svc ;
;

void eth_socket::open(std::string address) 
    tcp::endpoint remote_endpoint(ba::ip::address::from_string(address), _port);

    _sock.async_connect(remote_endpoint, boost::bind(&eth_socket::connected, this, _1));


void eth_socket::write_data(std::string data) 
    _data = data;

    _sock.async_write_some(ba::buffer(_data), boost::bind(&eth_socket::written, this, _1, _2));


void eth_socket::written(error_code error, size_t bytes_transferred) 
    INFO << "data written to " << _sock.remote_endpoint() << " " << error.message() << ";"
         << "bytes_transferred = " << bytes_transferred << "\n";


int main() 
    
        eth_socket s;
        s.open("127.0.0.1");

        s.write_data("Hello"); // callback called
        s.write_data("Hello"); // callback called
        s.write_data("Hello"); // callback called
        boost::this_thread::sleep_for(boost::chrono::seconds(1));
        s.write_data("Hello"); // callback not called after sleep

     // orderly worker thread join here

【讨论】：

请注意，当write_data 与先前的读取重叠时，这会导致数据争用问题。稍后我将链接到通用解决方案。

谢谢。我想我的问题出在io_service 和线程上。那么_work 线程一直在运行吗？在回调中我应该有io_service_1->reset(); 并在我连接后启动io_service 工作线程？我认为工作线程在回调上耗尽了工作，因此我需要为每次写入启动它。

...请阅读示例代码/我没有无缘无故地编写示例。额外的背景链接可能会以防你错过：Why do we need to use boos::asio::io_service::work（不过它指的是第一个项目符号）

哦。我以前没有看到这个，但*&io_service_1 绝对是错误的。这是一个重言式。您可能（无用地）表示&*io_service_。但是boost::bind 直接支持shared_ptr<> 就好了——这也让它更加安全）

【参考方案2】：

感谢 sehe 的帮助和祈祷，我的问题现在得到了解决。

open_eth_socket中的这一行：

boost::thread t(boost::bind(&boost::asio::io_service::run, *&io_service_1));

现在是这样的：

boost::shared_ptr <boost::thread>  io_service_1_thread;    // in header

if (io_service_1_thread.get()) io_service_1_thread->interrupt();
io_service_1_thread.reset(new boost::thread (boost::bind(&eth_socket::run_io_service_1, this)));

我添加了这个功能：

void eth_socket::run_io_service_1 (void)

  while (true)  // work forever
  
    boost::asio::io_service::work work(*io_service_1);
    io_service_1->run();
    io_service_1->reset();   // not sure if this will cause problems yet
    INFO << "io_service_1 run complete";
    boost::this_thread::sleep (boost::posix_time::milliseconds (100));
  

    return;