boost:asio::read or boost:asio::async_read with timeout

boost:asio::read or boost:asio::async_read with timeout

本文关键字：boost read asio timeout with or async      更新时间：2023-10-16

是的。我知道在boost::asio中围绕这个time_out有几个问题。我的问题对于asio人来说可能太简单了，无法在这里解决。

我正在使用 TCP 协议上的 boost::asio 以尽可能快的速度在循环中连续读取网络数据。

在 while 循环中，从工作线程std::thread连续调用ReadData()函数。

std::size_t ReadData(std::vector<unsigned char> & buffer, unsigned int size_to_read) {
 boost::system::error_code error_code;
 buffer.resize(size_to_read);
 // Receive body
 std::size_t bytes_read = boost::asio::read(*m_socket, boost::asio::buffer(buffer), error_code);
 if (bytes_read == 0) {
   // log error
   return;
 }
 return bytes_read;
}

它工作正常。返回数据。一切都很好。

我想要的，就是为boost::asio::read使用time_out。我了解到我需要将boost::asio::async_read与boost::asio::async_wait一起使用，time_out技术才能起作用。

一个提升示例建议使用 boost::asio::async_read_until ？

我应该使用boost::asio::async_read还是boost::asio::async_read_until？

我使用boost::asio::async_read、boost::asio::async_read_until或boost::asio::read都没有关系。但是我希望asio::read调用在调用我的方法ReadData中触发并完成，以便客户端代码不会受到影响。

我怎样才能做到这一点？请建议

好的，这样的东西应该适合你的目的：

理由：

您似乎想要使用阻止操作。既然是这种情况，你很有可能没有运行线程来抽取io循环。

因此，我们在套接字的 io 循环上同时启动两个异步任务，然后生成一个线程来：

a( 重置(实际重新启动(循环，以防它已经耗尽

b( 运行

循环到耗尽(我们在这里可以更聪明，只在处理程序指示满足某些条件之前运行它，但这是另一天的教训(

#include <type_traits>
template<class Stream, class ConstBufferSequence, class Handler>
auto async_read_with_timeout(Stream& stream, ConstBufferSequence&& sequence, std::size_t millis, Handler&& handler)
{
    using handler_type = std::decay_t<Handler>;
    using buffer_sequence_type = std::decay_t<ConstBufferSequence>;
    using stream_type = Stream;
    struct state_machine : std::enable_shared_from_this<state_machine>
    {
        state_machine(stream_type& stream, buffer_sequence_type sequence, handler_type handler)
                : stream_(stream)
                , sequence_(std::move(sequence))
                , handler_(std::move(handler))
        {}
        void start(std::size_t millis)
        {
            timer_.expires_from_now(boost::posix_time::milliseconds(millis));
            timer_.async_wait(strand_.wrap([self = this->shared_from_this()](auto&& ec) {
                self->handle_timeout(ec);
            }));
            boost::asio::async_read(stream_, sequence_,
                                    strand_.wrap([self = this->shared_from_this()](auto&& ec, auto size){
                self->handle_read(ec, size);
            }));
        }
        void handle_timeout(boost::system::error_code const& ec)
        {
            if (not ec and not completed_)
            {
                boost::system::error_code sink;
                stream_.cancel(sink);
            }
        }
        void handle_read(boost::system::error_code const& ec, std::size_t size)
        {
            assert(not completed_);
            boost::system::error_code sink;
            timer_.cancel(sink);
            completed_ = true;
            handler_(ec, size);
        }
        stream_type& stream_;
        buffer_sequence_type sequence_;
        handler_type handler_;
        boost::asio::io_service::strand strand_ { stream_.get_io_service() };
        boost::asio::deadline_timer timer_ { stream_.get_io_service() };
        bool completed_ = false;
    };
    auto psm = std::make_shared<state_machine>(stream,
                                               std::forward<ConstBufferSequence>(sequence),
                                               std::forward<Handler>(handler));
    psm->start(millis);
}
std::size_t ReadData(boost::asio::ip::tcp::socket& socket,
                     std::vector<unsigned char> & buffer,
                     unsigned int size_to_read,
                     boost::system::error_code& ec) {
    buffer.resize(size_to_read);
    ec.clear();
    std::size_t bytes_read = 0;
    auto& executor = socket.get_io_service();
    async_read_with_timeout(socket, boost::asio::buffer(buffer),
                            2000, // 2 seconds for example
                            [&](auto&& err, auto size){
        ec = err;
        bytes_read = size;
    });
    // todo: use a more scalable executor than spawning threads
    auto future = std::async(std::launch::async, [&] {
        if (executor.stopped()) {
            executor.reset();
        }
        executor.run();
    });
    future.wait();
    return bytes_read;
}
相关文章：