javascript - Boost::asio::async_read_until 从不调用处理程序

Question

我正在尝试在本地网络上的两台计算机之间建立连接，一台使用 Boost Asio C++ TCP 异步服务器示例的略微修改版本，另一台使用 NodeJS。

tcp_client.js :

var net = require('net');
var HOST = '127.0.0.1';
var PORT = 14002;

var client = new net.Socket();
client.connect(PORT, HOST, function() {
    console.log('CONNECTED TO: ' + HOST + ':' + PORT);
    // Write a message to the socket as soon as the client is connected
    //the server will receive it as message from the client 
    client.write('Hello');
});

// Add a 'data' event handler for the client socket
// data is what the server sent to this socket
client.on('data', function(data) {
    var fs = require('fs');
    fs.writeFile("test.txt", data, function(err) {
        if(err) {
            return console.log(err);
        }
        client.write("Data written"); // returns successful
        console.log("The file was saved!");
    }); 
});

// Add a 'close' event handler for the client socket
client.on('close', function() {
    console.log('Connection closed');
});

tcpServer.cpp :

#include <ctime>
#include <iostream>
#include <fstream>
#include <string>
#include <unistd.h>
#include <boost/bind.hpp>
#include <boost/shared_ptr.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <boost/asio.hpp>

extern string _coordinates;

using namespace std;
using boost::asio::ip::tcp;

std::string inline make_daytime_string() {
    std:: time_t now = std::time(0);
    return std::ctime(&now);
}

class tcp_connection
    // Using shared_ptr and enable_shared_from_this
    // because we want to keep the tcp_connection object alive
    // as long as there is an operation that refers to it.
    : public boost::enable_shared_from_this<tcp_connection>
    {
    public:
        typedef boost::shared_ptr<tcp_connection> pointer;

        static pointer create(boost::asio::io_service& io_service) {
            cout << "Creates a pointer for the tcp connection" <<endl;
            return pointer(new tcp_connection(io_service));
        }

        tcp::socket& socket() {
            return socket_;
        }

        // Call boost::asio::async_write() to serve the data to the client.
        // We are using boost::asio::async_write(),
        // rather than ip::tcp::socket::async_write_some(),
        // to ensure that the entire block of data is sent.

        void start() {
            while(1) {
                start_read();

                // This is going to read after every 1ms the _coordinates variable
                usleep(1000);

                m_message = _coordinates;

                boost::asio::async_write(
                    socket_,
                    boost::asio::buffer(m_message),
                    boost::bind(
                        &tcp_connection::handle_write,
                        shared_from_this(),
                        boost::asio::placeholders::error,
                        boost::asio::placeholders::bytes_transferred
                    )
                );
            }
        }

private:
    tcp_connection(boost::asio::io_service& io_service)
        : socket_(io_service)
        {
        }

    void start_read() {
        // Start an asynchronous operation to read a newline-delimited message.
        // When read, handle_read should kick in
        boost::asio::async_read_until(
            socket_,
            input_buffer_,
            '\n',
            boost::bind(
                &tcp_connection::handle_read,
                shared_from_this(),
                boost::asio::placeholders::error
            )
        );
    }

    // When stream is received, handle the message from the client
    void handle_read(const boost::system::error_code& ec) {

        std::cout << "HANDLE_READ - line 101" << "\n";
        messageFromClient_ = "";
        if (!ec) {
            // Extract the newline-delimited message from the buffer.
            std::string line;
            std::istream is(&input_buffer_);
            std::getline(is, line);

            // Empty messages are heartbeats and so ignored.
            if (!line.empty()) {
                messageFromClient_ += line;
                std::cout << "Received: " << line << "\n";
            }
            start_read();
        }
        else {
            std::cout << "Error on receive: " << ec.message() << "\n";
        }
    }

    // handle_write() is responsible for any further actions
    // for this client connection.
    void handle_write(const boost::system::error_code& /*error*/, size_t /*bytes_transferred*/) {
        m_message += "helloo\n";
    }

    tcp::socket socket_;
    std::string m_message;
    boost::asio::streambuf input_buffer_;
    std::string messageFromClient_;
};

class tcp_server {
    public:
        // Constructor: initialises an acceptor to listen on TCP port 14002.
        tcp_server(boost::asio::io_service& io_service)
        : acceptor_(io_service, tcp::endpoint(tcp::v4(), 14002))
        {
            // start_accept() creates a socket and
            // initiates an asynchronous accept operation
            // to wait for a new connection.
            start_accept();
        }

private:
    void start_accept() {
        // creates a socket
        cout << "creating a new socket for the communication" <<endl;
        tcp_connection::pointer new_connection = tcp_connection::create(acceptor_.get_io_service());

        // initiates an asynchronous accept operation
        // to wait for a new connection.
        acceptor_.async_accept(
            new_connection->socket(),
            boost::bind(
                &tcp_server::handle_accept,
                this,
                new_connection,
                boost::asio::placeholders::error
            )
        );
    }

    // handle_accept() is called when the asynchronous accept operation
    // initiated by start_accept() finishes. It services the client request
    void handle_accept(tcp_connection::pointer new_connection, const boost::system::error_code& error) {

        if (!error) {
            cout << "Starting the new tcp connection" <<endl;
            new_connection->start();
        }

        // Call start_accept() to initiate the next accept operation.
        start_accept();
    }

    tcp::acceptor acceptor_;
};


int inline launch_server() {
    try {
        boost::asio::io_service io_service;
        tcp_server server(io_service);

        // Run the io_service object to perform asynchronous operations.
        io_service.run();
    }
    catch (std::exception& e) {
        std::cerr << e.what() << std::endl;
    }
    return 0;
}

将消息从 C++ 发送到 NodeJS 是可行的(start() while 循环中的 async_write(..)，它发送 _coordinates 每 1 毫秒)，但是我无法处理来 self 的 NodeJS 程序的消息:

运行时(两个程序都在我的电脑上，在本地主机上)，ss -tp | 的输出grep 14002(14002为端口)，NodeJS进程的Recv-Q/Send-Q为空(socket.write(...)返回成功)，而，对于C++部分，Recv-Q不断增长，Send-Q为空而且，在运行时，handler_read()的所有cout ..都没有打印出来，也就是说async_read_until()函数永远不会调用处理程序。

我尝试了 async_read_until() 的所有重载版本，但没有一个有效。而且，由于消息的大小不是恒定的，看来我别无选择，只能使用 read_until。

希望我没有忘记任何有用的信息。感谢您的帮助!

Answer 1

您基本上是在使用 tcp_connection::start 方法中的无限 while 循环使您的 CPU 饱和。它不仅使 CPU 饱和，而且还是您设计中的错误。为什么要在无限循环中连续附加读取处理程序以及向套接字发送/写入数据？很可能您想在收到来自客户端的请求后写入到套接字。

以下是我更改的方法，使其像常规客户端-服务器一样工作:

void start() {
          start_read();

          // This is going to read after every 1ms the _coordinates variable
          usleep(1000);
          m_message = _coordinates;
        }

void start_read() {
        // Start an asynchronous operation to read a newline-delimited message.
        // When read, handle_read should kick in
        boost::asio::async_read_until(
            socket_,
            input_buffer_,
            '\n',
            boost::bind(
                &tcp_connection::handle_read,
                shared_from_this(),
                boost::asio::placeholders::error
            )
        );
    }

void handle_read(const boost::system::error_code& ec) {

        std::cout << "HANDLE_READ - line 101" << "\n";
        messageFromClient_ = "";
        if (!ec) {
            // Extract the newline-delimited message from the buffer.
            std::string line;
            std::istream is(&input_buffer_);
            std::getline(is, line);

            // Empty messages are heartbeats and so ignored.
            if (!line.empty()) {
                messageFromClient_ += line;
                std::cout << "Received: " << line << "\n";
            }
            start_read();
        }
        else {
            std::cout << "Error on receive: " << ec.message() << "\n";
        }

        start_read();

        boost::asio::async_write(
                    socket_,
                    boost::asio::buffer(m_message),
                    boost::bind(
                        &tcp_connection::handle_write,
                        shared_from_this(),
                        boost::asio::placeholders::error,
                        boost::asio::placeholders::bytes_transferred
                    )
                );
    }

根据目前的问题，我不确定您“实际上”想做什么，但上述更改应该是一个很好的起点。