c++ - boost::asio，为什么我的套接字在调用 async_receive_from 后立即运行回调函数？

Question

我正在尝试使用 3 次握手建立异步连接。在代码的底部有一个 main 方法，它接受一个参数来确定它是充当客户端还是服务器。

如果它充当服务器，它会创建一个套接字并等待它接收到数据，然后调用回调函数 multiplex 来确定如何处理它接收到的数据。

如果它充当客户端，我还将创建一个异步等待接收数据的套接字，但它还会使用 create_connection 方法通过 udp 将 syn 数据包发送到服务器。

在 Socket 的构造函数中，我执行方法 start_receive，它应该在 udp 套接字上调用 async_receive_from。问题是它立即使用 0.0.0.0:0 的端点调用回调函数 multiplex 而不是等待接收数据。

#include <iostream>
#include <string>
#include <boost/asio.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/cstdint.hpp>
#include <boost/function.hpp>
#include <boost/bind.hpp>
#include <unordered_map>
#include "packed_message.h"
#include "segment.pb.h"
#include "rtp.hpp"

#define DEBUG true
#define BUFFER_SIZE 10000

using boost::asio::ip::udp;
typedef boost::shared_ptr<rtp::Segment> SegmentPtr;
typedef std::vector<uint8_t> data_buffer;

// constructor for socket
rtp::Socket::Socket(boost::asio::io_service& io_service_, std::string source_ip, std::string source_port): 
    io_service_(io_service_),
    socket_(io_service_, udp::endpoint(boost::asio::ip::address::from_string(source_ip), std::stoi(source_port))),
    source_port(source_port),
    source_ip(source_ip)
{
    // accept incoming rtp segments
    std::cout << rtp::get_endpoint_str(socket_.local_endpoint()) << std::endl; 
    start_receive();
}

/**
 *  Accept incoming rtp segments
 */
void rtp::Socket::start_receive()
{
    data_buffer tmp_buf;
    socket_.async_receive_from(boost::asio::buffer(tmp_buf), remote_endpoint_, 
        boost::bind(&rtp::Socket::multiplex, this,
            tmp_buf,
            boost::asio::placeholders::error, 
            boost::asio::placeholders::bytes_transferred));

}

void rtp::Socket::multiplex(data_buffer& tmp_buf,
    const boost::system::error_code& error,
    std::size_t /*bytes_transferred*/)
{
    std::string identifier = rtp::get_endpoint_str(remote_endpoint_);

    if (connections.count(identifier) == 0 )
    {
        boost::shared_ptr<Connection> connection(new Connection(remote_endpoint_));
        connections.insert({rtp::get_endpoint_str(remote_endpoint_), connection});
        std::cout << rtp::get_endpoint_str(remote_endpoint_) << std::endl;
        connection_establishment(tmp_buf, connections.at(identifier));


    }
    else if(!(connections.at(identifier)->is_valid())) // connection not in list of connections
    {
        connection_establishment(tmp_buf, connections.at(identifier));


}

boost::shared_ptr<rtp::Connection> rtp::Socket::create_connection(std::string ip, std::string port)
{
    udp::resolver resolver_(io_service_);
    udp::resolver::query query_(ip, port);
    udp::endpoint remote_endpoint_ = *resolver_.resolve(query_);
    boost::shared_ptr<Connection> connection(new Connection(remote_endpoint_));

    connections.insert({rtp::get_endpoint_str(remote_endpoint_), connection});

    PackedMessage<rtp::Segment> m_packed_segment(SegmentPtr(new rtp::Segment()));
    boost::shared_ptr<data_buffer> message(new data_buffer);
    SegmentPtr ackseg(new rtp::Segment());
    ackseg->set_ack(true);
    PackedMessage<rtp::Segment> initialack(ackseg);
    initialack.pack(*message);



    socket_.async_send_to(boost::asio::buffer(*message), remote_endpoint_,
      boost::bind(&rtp::Socket::handle_send, this, message,
        boost::asio::placeholders::error,
        boost::asio::placeholders::bytes_transferred));

    return connection;

}

void rtp::Socket::connection_establishment(data_buffer& m_readbuf, boost::shared_ptr<Connection> connection)
{
    int buffer_position(0);
    PackedMessage<rtp::Segment> m_packed_segment(boost::shared_ptr<rtp::Segment>(new rtp::Segment()));
    boost::shared_ptr<data_buffer> message(new data_buffer);

    unsigned msg_len = m_packed_segment.decode_header(m_readbuf, buffer_position);
    buffer_position += HEADER_SIZE;
    m_packed_segment.unpack(m_readbuf, msg_len, buffer_position);
    buffer_position += msg_len;

    SegmentPtr synackseg = m_packed_segment.get_msg();
    if (synackseg->syn() && synackseg->ack())
    {
        SegmentPtr ackseg(new rtp::Segment());
        ackseg->set_ack(true);
        PackedMessage<rtp::Segment> finalack(ackseg);
        finalack.pack(*message);

        connection->set_valid(true);

        socket_.async_send_to(boost::asio::buffer(*message), remote_endpoint_,
            boost::bind(&rtp::Socket::handle_send, this,
                message,
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));

    }
    else if (synackseg->syn() )
    {
        SegmentPtr synackseg(new rtp::Segment());
        synackseg->set_ack(true);
        synackseg->set_syn(true);
        PackedMessage<rtp::Segment> synack(synackseg);
        synack.pack(*message);

        socket_.async_send_to(boost::asio::buffer(*message), remote_endpoint_,
            boost::bind(&rtp::Socket::handle_send, this,
                message,
                boost::asio::placeholders::error,
                boost::asio::placeholders::bytes_transferred));


    }
    else if (synackseg->ack())
    {
        connection->set_valid(true);
    }

    start_receive();
}

void rtp::Socket::handle_send(boost::shared_ptr<data_buffer> /*message*/,
            const boost::system::error_code& /*error*/, 
            std::size_t /*bytes_transferred*/)
{
}

/**
 *  Get remote peer ip and port in string "<ip>:<port>"
 */
std::string rtp::get_endpoint_str(udp::endpoint remote_endpoint_)
{
    std::string ip = remote_endpoint_.address().to_string();
    std::string port = std::to_string(remote_endpoint_.port());
    return ip + ":" + port;
}


rtp::Connection::Connection(udp::endpoint remote_endpoint_):
    remote_endpoint_(remote_endpoint_),
    dest_ip(remote_endpoint_.address().to_string()),
    dest_port(std::to_string(remote_endpoint_.port())),
    valid(false)
{
}

bool rtp::Connection::is_valid()
{
    return valid;
}

void rtp::Connection::set_valid(bool val)
{
    if(DEBUG) std::cerr << "Connection Created" << std::endl;
    valid = val;
}


int main(int argc, char* argv[])
{
    if (argc == 1)
    {
        std::cerr << "Not enough args" << std::endl;
        return 1;
    }
    boost::asio::io_service io_service_;

    if (std::string(argv[1]) == u8"server")
    {
        rtp::Socket socket(io_service_, u8"127.0.0.1", u8"4545");
    }
    else if (std::string(argv[1]) == u8"client")
    {
        rtp::Socket socket(io_service_, u8"127.0.0.1", u8"4546");
        socket.create_connection(u8"127.0.0.1", u8"4545");

    }
    io_service_.run();
    return 0;

}

Answer 1

问题很可能是这些行:

if (std::string(argv[1]) == u8"server")
{
    rtp::Socket socket(io_service_, u8"127.0.0.1", u8"4545");
}
else if (std::string(argv[1]) == u8"client")
{
    rtp::Socket socket(io_service_, u8"127.0.0.1", u8"4546");
    socket.create_connection(u8"127.0.0.1", u8"4545");

}

在每个 if 语句体中声明一个变量 socket，但它仅在 if 语句的范围内是局部的 body 。一旦该范围结束，变量将被破坏并且不再存在。

您看到的行为可能与该问题有关，即您在这些范围内创建的对象被破坏。

您需要创建一个套接字，其生命周期比任何内部作用域都长，并且会持续到 io_service_.run() 返回。