启动Boost.Asio TCP客户端触发boost::system::system_error异常求助

问题现象 #

基于Boost.Asio开发类Redis的TCP客户端服务器应用，服务器启动后可正常监听（netstat显示处于LISTEN状态），但客户端启动后立即抛出错误：

terminate called after throwing an instance of 'boost::wrapexcept<boost::system::system_error>'
  what():  close: Bad file descriptor [system:9]

相关代码 #

client.h #

#ifndef CLIENT_IMPL_H
#define CLIENT_IMPL_H

#include <boost/asio.hpp>
#include <boost/enable_shared_from_this.hpp>
#include <cstddef>
#include <string>

namespace io = boost::asio;
using tcp = io::ip::tcp;
using error_code = boost::system::error_code;
using namespace std::placeholders;

class client : public boost::enable_shared_from_this<client> {

public:

    client(io::io_context &io_context, std::string const &addr, uint16_t port) 
    : socket_(io_context) {
        start(tcp::endpoint(
            io::ip::address::from_string(addr),
            port
        ));
    }

    void start(tcp::endpoint ep);

private:

    void on_connect(error_code const &err);

    void on_read(error_code const &err, std::size_t bytes_transmitted);
    void async_read();

    void on_write(error_code const &err);
    void async_write();

private:

    tcp::socket socket_;
    io::streambuf streambuf_;
    std::string request_;
    std::string response_;

};

#endif

client.cpp #

#include <iostream>

#include "client_impl.h"

void client::start(tcp::endpoint ep) {
    socket_.async_connect(
        ep,
        [this](error_code err) {
            on_connect(err);
        }
    );
}

void client::on_connect(error_code const &err) {
    if (!err) {
        async_write();
    } else {
        socket_.close();
    }
}

void client::on_read(error_code const &err, std::size_t bytes_transmitted) {
    if (!err) {
        std::stringstream message(response_);
        message << std::istream(&streambuf_).rdbuf();
        streambuf_.consume(bytes_transmitted);

        std::cout << response_ << std::endl;
        response_.clear();

        async_write();
    } else {
        socket_.close();
    }
}

void client::async_read() {
    auto self = shared_from_this();
    io::async_read_until(
        socket_,
        streambuf_,
        "\n",
        [self](error_code err, std::size_t bytes_transmitted) {
            self->on_read(err, bytes_transmitted);
        }
    );
}

void client::on_write(error_code const &err) {
    if (!err) {
        async_read();
    } else {
        socket_.close();
    }
}

void client::async_write() {
    std::getline(std::cin, request_);
    request_ += "\n";
    io::async_write(
        socket_,
        io::buffer(request_),
        [this](error_code err, std::size_t bytes_transmitted) {
            on_write(err);
        }
    );
}

server.h #

#ifndef SERVER_IMPL_H
#define SERVER_IMPL_H

#include <boost/asio.hpp>
#include <boost/asio/io_context.hpp>
#include <memory>
#include <optional>
#include <string>
#include <unordered_map>
#include <utility>
#include <regex>


namespace io = boost::asio;
using tcp = io::ip::tcp;
using error_code = boost::system::error_code;
using namespace std::placeholders;

class session : public std::enable_shared_from_this<session> {

public:
    session(tcp::socket socket, std::unordered_map<std::string, std::string> &storage) : 
        socket_(std::move(socket)),
        storage_ {storage} {}

    void start();

private:

    void async_read();
    void on_read(error_code err, std::size_t bytes_transmitted);

    void async_write();
    void on_write(error_code err, std::size_t bytes_transmitted);

    void message_handler(std::string const &message);
    void on_put(std::string const &message);
    void on_get(std::string const &message);
    void on_del(std::string const &message);
    void on_count(std::string const &message);
    void on_dump(std::string const &message);
    bool is_valid_key(std::string const &key);
    bool is_valid_txt(std::string const &fname);

    std::string get_from_storage(std::string const &key);
    void put_from_storage(std::string const &key, std::string const &value);

    tcp::socket socket_;
    io::streambuf streambuf_;
    std::string buffer_;
    std::string response_;

    std::unordered_map<std::string, std::string> &storage_;
    const static std::regex key_regex_;
    const static std::regex txt_regex_;

};

class server {

public:

    server(io::io_context& io_context, std::uint16_t port) :
        io_context_(io_context),
        acceptor_(io_context_, tcp::endpoint(tcp::v4(), port)) {
        async_accept();
    }

    void async_accept();

  private:

    io::io_context& io_context_;
    tcp::acceptor acceptor_;

    std::unordered_map<std::string, std::string> storage_;

};

#endif

server.cpp #

#include "server_impl.h"

#include <boost/asio.hpp>
#include <cstddef>
#include <functional>
#include <memory>
#include <regex>
#include <sstream>
#include <string>
#include <utility>

/* session implementation */

const std::regex session::key_regex_("[A-Za-z0-9]+");
const std::regex session::txt_regex_("[A-Za-z0-9]+\\.txt");

void session::start() {
    async_read();
}

void session::async_read() {
    io::async_read_until(
        socket_,
        streambuf_,
        "\n",
        std::bind(&session::on_read, shared_from_this(), _1, _2)
    );
}

void session::on_read(error_code err, std::size_t bytes_transmitted) {
    if (!err) {
        std::stringstream message(buffer_);
        message << std::istream(&streambuf_).rdbuf();
        streambuf_.consume(bytes_transmitted);

        message_handler(buffer_);
        async_write();
        buffer_.clear();

        async_read();
    } else {
        socket_.close();
    }
}

void session::async_write() {
    io::async_write(
        socket_,
        io::buffer(response_),
        std::bind(&session::on_write, shared_from_this(), _1, _2)
    );
}

void session::on_write(error_code err, std::size_t bytes_transmitted) {
    if (!err) {
        async_write();
    } else {
        socket_.close();
    }
}

void session::message_handler(std::string const &message) {

    if (message.starts_with("PUT ")) {
        on_put(message);
    } else if (message.starts_with("GET ")) {
        on_get(message);
    } else if (message.starts_with("DEL ")) {
        on_del(message);
    } else if (message.starts_with("COUNT ")) {
        on_count(message);
    } else if (message.starts_with("DUMP ")) {
        on_dump(message);
    } else {
        response_ = "NE";
    }

}

void session::on_put(std::string const &message) {
    std::istringstream ss(message);
    std::string tmp;
    std::string key;
    int cnt {};

    ss >> tmp;
    while (ss >> tmp) {
        if (++cnt == 3) {
            response_ = "NE\n";
            return;
        }
        if (cnt == 1) {
            if (!is_valid_key(tmp)) {
                response_ = "NE\n";
                return;
            }
            key = tmp;
        }
    }

    if (cnt != 2) { 
        response_ = "NE\n";
        return;
    }

    if (std::string value = get_from_storage(key); value != "") {
        response_ = "OK " + value + "\n";
    }
    storage_[key] = tmp;
}

void session::on_get(std::string const &message) {
    std::istringstream ss(message);
    std::string tmp;
    std::string value;
    int cnt {};

    ss >> tmp;
    while (ss >> tmp) {
        if (++cnt == 2) {
            response_ = "NE\n";
            return;
        }
        if (!is_valid_key(tmp)) {
            response_ = "NE\n";
            return;
        }
        value = get_from_storage(tmp);
    }

    if (cnt != 1 || value == "") { 
        response_ = "NE\n";
        return;
    }

    response_ = "OK " + value + "\n";
}

void session::on_del(std::string const &message) {
    std::istringstream ss(message);
    std::string tmp;
    std::string value;
    int cnt {};

    ss >> tmp;
    while (ss >> tmp) {
        if (++cnt == 2) {
            response_ = "NE\n";
            return;
        }
        if (!is_valid_key(tmp)) {
            response_ = "NE\n";
            return;
        }
        value = get_from_storage(tmp);
    }

    if (cnt != 1 || value == "") { 
        response_ = "NE\n";
        return;
    }

    storage_.erase(tmp);
    response_ = "OK " + value + "\n";
}

void session::on_count(std::string const &message) {
    std::istringstream ss(message);
    std::string tmp;

    ss >> tmp;
    while (ss >> tmp) {
        response_ = "NE\n";
        return;
    }

    size_t cnt {storage_.size()};
    response_ = "OK " + std::to_string(cnt) + "\n";
}

void session::on_dump(std::string const &message) {
    std::istringstream ss(message);
    std::string tmp;
    int cnt {};

    ss >> tmp;
    while (ss >> tmp) {
        if (++cnt == 2) {
            response_ = "NE\n";
            return;
        }
    }

    if (cnt != 1 || !is_valid_txt(tmp)) { 
        response_ = "NE\n";
        return;
    }

    // dumping
    response_ = "OK\n";
}

bool session::is_valid_key(std::string const &key) {
    return std::regex_match(key, key_regex_);
}

bool session::is_valid_txt(std::string const &fname) {
    return std::regex_match(fname, txt_regex_);
}

std::string session::get_from_storage(std::string const &key) {
    if (auto it = storage_.find(key); it != storage_.end()) {
        return it->second;
    }
    return "";
}

void session::put_from_storage(std::string const &key, std::string const &value) {
    storage_[key] = value;
}


/* server implemetation */

void server::async_accept() {
    acceptor_.async_accept([this](error_code err, tcp::socket socket) {
        if (!err) {
            std::make_shared<session>(std::move(socket), storage_)->start();
        }
    });
}

启动代码 #

// 服务器启动代码
int main(int argc, char* argv[]) {

    if (argc < 3 || argc > 4) {
        std::cout << "usage: server <port> <limit_connections> [<dump_file>]" << std::endl;
        return 1;
    }

    io::io_context io_context;
    server srv(io_context, static_cast<uint16_t>(std::stoi(argv[1])));
    srv.async_accept();
    io_context.run();
    return 0;
}

// 客户端启动代码
int main(int argc, char* argv[]) {
    if (argc != 3) {
        std::cout << "usage: client <ip> <port>" << std::endl;
        return 1;
    }

    io::io_context io_context;
    std::make_shared<client>(io_context, "127.0.0.1", std::stoi(argv[2]));
    io_context.run();

    return 0;
}

问题根源及修复方案 #

1. 客户端对象生命周期失效 #

客户端启动代码中，std::make_shared<client>创建的对象未被持有，导致对象创建后立即销毁，socket被提前关闭，后续异步操作触发文件描述符错误。
修复： 保存客户端的shared_ptr引用：

// 客户端启动代码修改
int main(int argc, char* argv[]) {
    if (argc != 3) {
        std::cout << "usage: client <ip> <port>" << std::endl;
        return 1;
    }

    io::io_context io_context;
    auto client_ptr = std::make_shared<client>(io_context, argv[1], std::stoi(argv[2]));
    io_context.run();

    return 0;
}

2. 服务器accept循环中断 #

服务器async_accept仅调用一次，处理完第一个连接后停止监听新连接，后续客户端连接失败。
修复： 在accept回调中再次调用async_accept，保持监听状态：

// server.cpp中async_accept修改
void server::async_accept() {
    acceptor_.async_accept([this](error_code err, tcp::socket socket) {
        if (!err) {
            std::make_shared<session>(std::move(socket), storage_)->start();
        }
        async_accept(); // 持续监听新连接
    });
}

3. Session写操作无限递归 #

session::on_write中无错误时重复调用async_write，导致无限发起写操作，最终触发错误。
修复： 写完响应后启动下一次读操作，等待客户端新请求：

void session::on_write(error_code err, std::size_t bytes_transmitted) {
    if (!err) {
        response_.clear();
        async_read();
    } else {
        if (socket_.is_open()) {
            socket_.close();
        }
    }
}

4. 数据读取逻辑错误 #

session::on_read和client::on_read中，未正确从streambuf读取数据到缓冲区，导致后续业务逻辑处理空字符串。
修复： 用std::getline从streambuf读取完整一行：

// session::on_read修改
void session::on_read(error_code err, std::size_t bytes_transmitted) {
    if (!err) {
        std::istream is(&streambuf_);
        std::getline(is, buffer_);
        streambuf_.consume(bytes_transmitted);

        message_handler(buffer_);
        async_write();
        buffer_.clear();

        async_read();
    } else {
        if (err != boost::asio::error::eof && socket_.is_open()) {
            socket_.close();
        }
    }
}

// client::on_read修改
void client::on_read(error_code const &err, std::size_t bytes_transmitted) {
    if (!err) {
        std::istream is(&streambuf_);
        std::getline(is, response_);
        streambuf_.consume(bytes_transmitted);

        std::cout << response_ << std::endl;
        response_.clear();

        async_write();
    } else {
        if (err != boost::asio::error::eof && socket_.is_open()) {
            socket_.close();
        }
    }
}

5. 重复关闭socket #

当错误为EOF时，socket已自动关闭，再次调用socket_.close()会触发"Bad file descriptor"错误。
修复： 关闭前检查socket状态，或忽略EOF错误：

// 所有调用socket_.close()的地方统一修改
if (socket_.is_open()) {
    socket_.close();
}

内容的提问来源于stack exchange，提问作者loaf4