async_send 数据未发送

Posted 2023-02-21

【中文标题】async_send 数据未发送

【英文标题】：async_send data not sent

【发布时间】：2017-01-14 13:38:02

【问题描述】：

[免责声明] 我是新手。

研究 boost::asio 并尝试创建一个具有以下功能的简单异步 TCP 服务器：

监听端口 13 上的连接 连接后，接收数据 如果接收到的数据 == 时间，则返回当前日期时间，否则返回预定义字符串（“请求了其他内容”）

问题： 虽然，我接受连接并接收数据，但在使用 async_send 传输数据时，虽然我没有收到错误并且 bytes_transferred 的值是正确的，但我在客户端收到了空数据。

如果我尝试从handle_accept（而不是handle_read）中传输数据，这可以正常工作。

实施： 我研究了 here 的 boost asio 教程： 实例化一个 tcp_server 对象，它基本上启动接受器并开始监听。如下图：

int main()

    try
    
        boost::asio::io_service io_service;
        tcp_server server(io_service);
        io_service.run();
    
    catch (std::exception& e)
    
        std::cerr << e.what() << std::endl;
    

    return 0;

在 tcp_server 中：

class tcp_server

public:
    tcp_server(boost::asio::io_service& io_service)
        : acceptor_(io_service, tcp::endpoint(tcp::v4(), 13))
    
        start_accept();
    

private:
    void start_accept()
    
        using std::cout;
        tcp_connection::pointer new_connection =
            tcp_connection::create(acceptor_.get_io_service());

        acceptor_.async_accept(new_connection->socket(),
            boost::bind(&tcp_server::handle_accept, this, new_connection,
                boost::asio::placeholders::error));
        cout << "Done";
    
    ...

一旦连接被接受，我将按如下所示进行处理：

void handle_accept(tcp_connection::pointer new_connection,
        const boost::system::error_code& error)

    if (!error)
    
        new_connection->start();
    

    start_accept();

下面是tcp_connection::start()方法：

void start()


    boost::asio::async_read(socket_, boost::asio::buffer(inputBuffer_),
        boost::bind(&tcp_connection::handle_read, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));

    /* the snippet below works here - but not in handle_read 
    outputBuffer_ = make_daytime_string();

    boost::asio::async_write(socket_, boost::asio::buffer(outputBuffer_),
        boost::bind(&tcp_connection::handle_write, shared_from_this(),
            boost::asio::placeholders::error,
            boost::asio::placeholders::bytes_transferred));*/

在handle_read:

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

    outputBuffer_ = make_daytime_string();
    if (strcmp(inputBuffer_, "time"))
    
        /*this does not work - correct bytes_transferred but nothing shown on receiving end */
        boost::asio::async_write(socket_, boost::asio::buffer(outputBuffer_),
            boost::bind(&tcp_connection::handle_write, shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred)); 
    
    else
    
        outputBuffer_ = "Something else was requested";//, 128);
        boost::asio::async_write(socket_, boost::asio::buffer(outputBuffer_),
            boost::bind(&tcp_connection::handle_write, shared_from_this(),
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));
    

handle_write 如下图所示：

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

    if (!error)
    
        std::cout << "Bytes transferred: " << bytes_transferred;
        std::cout << "Message sent: " << outputBuffer_;
    
    else
    
        std::cout << "Error in writing: " << error.message();
    

请注意以下关于 handle_write 的内容（这真的很奇怪）：

没有错误 bytes_transferred 变量的值正确 outputBuffer_ 的值正确（在 handle_read 中设置）

尽管如此，客户端收到的包 (Packet Sender) 是空的（就数据而言）。

完整代码分享here。

【问题讨论】：

当你说你收到空数据时，你的意思是你得到一个完整的 0 字节 (eof) 接收还是 async_read 永远不会完成？为什么不检查处理函数中的错误代码？

@RichardHodges 我正在使用“Packet Sender”应用程序作为客户端。查看收到的 tcp 数据包的数据部分，它是空的（没有数据部分）。我想澄清一下，在上面的代码中， inputBuffer_ 是正确的。我没有检查错误，因为我收到了 EOF（我猜这是正常的，已经阅读了整个流）。我将检查除 EOF 之外的所有错误。我还将编辑我的问题以包含 handle_write。

小心 asio::async_read 和 eof 错误。 async_read free 函数是一个组合操作。当 bytes_transferred 非零时，它可以指示 eof。在这种情况下，您应该接受数据，然后可选择执行 eof 操作（如果不这样做，您将收到另一个 eof 错误，在下一次读取时传输零字节，因此跳过此可选步骤无关紧要） .

只是为了澄清。我的问题不在于 async_read。我收到的很好。问题是在尝试使用 async_write() 从 handler_read 中发送数据时。数据虽然看起来已正确发送，但在接收端不包含任何数据。

@Lefteris coliru.stacked-crooked.com/a/725c31ef858b80f8

【参考方案1】：

完整的测试程序（c++14）。注意响应接收时异步缓冲的处理 - 可能已经在进行发送。

#include <boost/asio.hpp>
#include <thread>
#include <future>
#include <vector>
#include <array>
#include <memory>
#include <mutex>
#include <condition_variable>
#include <iterator>
#include <iostream>

namespace asio = boost::asio;

asio::io_service        server_service;
asio::io_service::work  server_workserver_service;

bool listening = false;
std::condition_variable cv_listening;
std::mutex              management_mutex;

auto const shared_query = asio::ip::tcp::resolver::query(asio::ip::tcp::v4(), "localhost", "8082");

void client()
try

    asio::io_service      client_service;
    asio::ip::tcp::socket socket(client_service);

    auto lock = std::unique_lock<std::mutex>(management_mutex);
    cv_listening.wait(lock, []  return listening; );
    lock.unlock();

    asio::ip::tcp::resolver resolver(client_service);
    asio::connect(socket, resolver.resolve(shared_query));
    auto s = std::string("time\ntime\ntime\n");
    asio::write(socket, asio::buffer(s));
    socket.shutdown(asio::ip::tcp::socket::shutdown_send);

    asio::streambuf sb;
    boost::system::error_code sink;
    asio::read(socket, sb, sink);
    std::cout << std::addressof(sb);
    socket.close();
    server_service.stop();

catch(const boost::system::system_error& se)

    std::cerr << "client: " << se.code().message() << std::endl;


struct connection
    : std::enable_shared_from_this<connection>

    connection(asio::io_service& ios)
        : strand_(ios)
    

    

    void run()
    
        asio::async_read_until(socket_, buffer_, "\n",
                               strand_.wrap([self = shared_from_this()](auto const&ec, auto size)
        
            if (size == 0 )
            
                // error condition
                boost::system::error_code sink;
                self->socket_.shutdown(asio::ip::tcp::socket::shutdown_receive, sink);
            
            else 
                self->buffer_.commit(size);
                std::istream is(std::addressof(self->buffer_));
                std::string str;
                while (std::getline(is, str))
                
                    if (str == "time") 
                        self->queue_send("eight o clock");
                    
                
                self->run();
            
        ));
    

    void queue_send(std::string s)
    
        assert(strand_.running_in_this_thread());
        s += '\n';
        send_buffers_pending_.push_back(std::move(s));
        nudge_send();
    

    void nudge_send()
    
        assert(strand_.running_in_this_thread());
        if (send_buffers_sending_.empty() and not send_buffers_pending_.empty())
        
            std::swap(send_buffers_pending_, send_buffers_sending_);
            std::vector<asio::const_buffers_1> send_buffers;
            send_buffers.reserve(send_buffers_sending_.size());
            std::transform(send_buffers_sending_.begin(), send_buffers_sending_.end(),
                           std::back_inserter(send_buffers),
            [](auto&& str) 
                return asio::buffer(str);
            );
            asio::async_write(socket_, send_buffers,
                              strand_.wrap([self = shared_from_this()](auto const& ec, auto size)
            
                // should check for errors here...
                self->send_buffers_sending_.clear();
                self->nudge_send();
            ));
        
    

    asio::io_service::strand strand_;
    asio::ip::tcp::socket    socket_strand_.get_io_service();
    asio::streambuf          buffer_;

    std::vector<std::string> send_buffers_pending_;
    std::vector<std::string> send_buffers_sending_;
;

void begin_accepting(asio::ip::tcp::acceptor& acceptor)

    auto candidate = std::make_shared<connection>(acceptor.get_io_service());
    acceptor.async_accept(candidate->socket_, [candidate, &acceptor](auto const& ec)
    
        if (not ec) 
            candidate->run();
            begin_accepting(acceptor);
        
    );


void server()
try

    asio::ip::tcp::acceptor acceptor(server_service);
    asio::ip::tcp::resolver resolver(server_service);

    auto first = resolver.resolve(shared_query);
    acceptor.open(first->endpoint().protocol());
    acceptor.bind(first->endpoint());

    acceptor.listen();

    begin_accepting(acceptor);

    auto lock = std::unique_lock<std::mutex>(management_mutex);
    listening = true;
    lock.unlock();
    cv_listening.notify_all();


    server_service.run();


catch(const boost::system::system_error& se)

    std::cerr << "server: " << se.code().message() << std::endl;


int main()


    using future_type = std::future<void>;

    auto stuff = std::array<future_type, 2> std::async(std::launch::async, client),
                                                 std::async(std::launch::async, server);

    for (auto& f : stuff) f.wait();

【参考方案2】：

此代码中有多个问题。他们中的一些人可能会对您的问题负责：

TCP 没有数据包的定义，所以不能保证你会在handle_read 中一次收到time。为此，您需要一个状态机并尊重 bytes_transferred 信息。如果您只收到了部分消息，则需要以正确的偏移量继续。或者，您可以使用 asio 实用程序函数，例如准确读取字节长度或读取一行。 除了最后一点，您不应该真正将接收到的数据与strcmp 进行比较。这只有在远程也通过连接发送空终止符时才有效 - 是吗？ 您不会检查是否发生了错误，尽管这可能会在其他错误中表现出来。 如果您在短时间内收到多个数据片段，您可能会发出多个并发异步写入。这在 asio 中无效。 更重要的是，您在发送过程中改变了发送缓冲区 (outputBuffer_)。这几乎会导致未定义的行为。 asio 可能会尝试写入一段不再有效的内存。

【讨论】：

谢谢@Matthias247 - 我会调查提出的每一个问题，希望我能深入了解它。您的第 2 和第 3 个项目符号很简单，第 4 个项目我需要花一些时间。 5楼我不是很清楚，你能再解释一下吗？第一点也很清楚。

好的，第 5 点：如果您将数据发送给 boost asio（使用boost::asio::buffer(outputBuffer_)），出于性能原因，asio 不会复制数据并将其存储在内部。它只会复制缓冲区包装结构，这意味着类似于指向数据的指针和数据的长度。您有责任确保数据在整个操作期间保持活动状态。如果在异步写入操作期间您给 asio 的位置的内存发生了变化，那么您就有了未定义的行为。

如果在这里使用 std::string ，则在操作开始时将指针提供给 asio 字符串内部后备内存。当您重新分配字符串 (outputBuffer_=...) 时，std::string 将更改该位置的数据，甚至分配一个新的后备缓冲区并释放旧的。指向旧指针的指针无效。

好的，我知道了。我想这是一个问题，因为 tcp 接收可能会在几个部分被破坏，然后会调用 async_write 几次，而 outputBuffer_ 会在后续调用期间发生变化。

@Matthias247 第 1 点不正确。只有当发生错误或读取到缓冲区大小（在本例中为 128 字节）的数据时，才会调用回调。因此，可以保证在一次调用中读取“时间”。

【参考方案3】：

在问题中提供的 cmets 的集体帮助下，我已经解决了这个问题。我遇到的行为是因为async_read 的功能。更具体地说，在boost asio documentation 中写道：

该函数用于异步读取一定数量的字节 来自流的数据。函数调用总是立即返回。 异步操作将一直持续到以下情况之一 条件为真：

提供的缓冲区已满。也就是说，传输的字节数等于缓冲区大小的总和。 发生错误。

我用来读取输入的 inputBuffer_ 是一个 128 字符数组。我使用的客户端只会传输真实数据（没有填充），因此 async_read 在客户端关闭连接之前不会返回（或传输了 128 个字节的数据）。当客户端关闭连接时，无法发回请求的数据。这也是它与@Arunmu 的简单python tcp 客户端一起工作的原因（因为他总是发送128 字节的数据）。

为了解决这些问题，我进行了以下更改（提供了完整的工作代码here 以供参考）：

在tcp_connection::start：我现在使用async_read_until 来读取传入的数据（并使用\n 作为分隔符）。输入存储在boost::asio::streambuf 中。一旦找到分隔符或发生错误，async_read 保证返回。所以没有机会同时发出多个async_write。 在handle_read 中：我包含了错误检查，这使得调试变得更加简单。