多个连接时“尝试对非套接字进行操作”

Question

我正在使用自制的服务器、客户端和序列化实用程序构建简单的客户端-服务器程序。发生的情况是服务器每秒向客户端发送一条消息，同时等待来自客户端的消息。客户端在等待来自服务器的消息的同时，也在等待用户输入：如果输入是“send X”，它会向服务器发送一个字符串“X”；如果输入是“退出”，它将退出。当只有一个连接时一切正常，但是当我尝试同时打开多个客户端程序时，我在服务器控制台上收到此错误：

SENDER -> User [在抛出 'boost::wrapexceptboost::system::system_error' 实例后调用终止 what(): 尝试对不是套接字的东西进行操作

我不知道该尝试什么。我认为问题可能是我在一个和另一个之间打开客户端的时间太短（这应该仍然有效），所以我尝试在几秒钟后打开第二个客户端，但我遇到了同样的问题。我的目标是让它在短时间内即使是多个连接也能完美地工作。

服务器：

#include <thread>
#include <iostream>
#include <chrono>

using namespace std::chrono_literals;

#include "..\..\CppLibraries\Networks\server\physicalserver.hpp"
#include "..\..\CppLibraries\Networks\buffer\buffer.hpp"

#include "..\..\CppLibraries\Networks\misc\buffersocket.hpp"

#include "game.hpp"
#include "messages.hpp"

Game game;
dnw::PhysicalServer server;

void readClientData(dnw::Server::Socket& socket, dnw::Server& server)
{
    dnw::Buffer buffer;

    while(true)
    {
        try
        {
            socket >> buffer;
        } catch(boost::system::system_error& ex)
        {
            std::cout << "RECEIVER -> User ["
                      << socket.remote_endpoint().address().to_string()
                      << "] disconnected...\n"
            ;

            break;
        }

        std::cout << "Message incoming from [" << socket.remote_endpoint().address().to_string() << "]: \"" << buffer.retrieve<std::string>() << "\"\n";
    }
}

// Client Acceptor
void onClientConnection(dnw::Server::Socket& socket, dnw::Server& server)
{
    dnw::Buffer buffer;

    std::thread receive {readClientData, std::ref(socket), std::ref(server)};
    receive.detach();

    while(true)
    {
        std::this_thread::sleep_for(1000ms);

        buffer << (std::string{"Ciao! "} + std::to_string(std::chrono::high_resolution_clock::now().time_since_epoch().count()));

        try
        {
            socket << buffer;
        } catch(boost::system::system_error& ex)
        {
            std::cout << "SENDER -> User ["
                      << socket.remote_endpoint().address().to_string()
                      << "] disconnected...\n"
            ;
            
            break;
        }

        buffer.clear();
    }
}

int main()
{
    using namespace std;
    using namespace chrono_literals;

    // Open Server
    constexpr dnw::Server::Port client_port {50000};

    server.setOnConnection(client_port, onClientConnection);
    server.startListening (client_port);

    // Game Loop
    while(true)
    {
    }
}

客户：

#include "..\..\CppLibraries\Networks\client\physicalclient.hpp"
#include "..\..\CppLibraries\Networks\buffer\buffer.hpp"

#include "..\..\CppLibraries\Networks\misc\buffersocket.hpp"

#include "messages.hpp"

std::string prompt(const std::string_view message)
{
    std::string answer;
    std::cout << message;
    std::cin >> answer;
    return answer;
}

void gameLoop(dnw::Client::Socket& socket)
{
    dnw::Buffer buff;

    while(true)
    {
    }
}

void inputSender(dnw::Client::Socket& socket, dnw::Client& client)
{
    std::string input;
    dnw::Buffer buffer;

    while(true)
    {
        std::getline(std::cin, input);

        if(input.substr(0, 4) == "send")
        {
            buffer << input.substr(5);

            try
            {
                socket << buffer;
            } catch(boost::system::system_error& ex)
            {
                std::cout << "Server closed...\n";

                break;
            }

            buffer.clear();
        } else 
        if(input == "exit")
        {
            exit(0);
        }
    }
}

void onConnection(dnw::Client::Socket& socket, dnw::Client& client)
{
    dnw::Buffer buffer;

    std::thread sender {inputSender, std::ref(socket), std::ref(client)};
    sender.detach();

    while(true)
    {
        try
        {
            socket >> buffer;
        } catch(boost::system::system_error& ex)
        {
            std::cout << "Server closed...\n";

            break;
        }

        std::cout << buffer.retrieve<std::string>() << '\n';
    }
}



int main()
{
    using Client = dnw::PhysicalClient;

    // Connect to Game Server

    Client client;
    
    client.setOnConnection(onConnection);

    std::string address = prompt("Enter server address: ");

    client.connect(address, "50000");

    while(true);
}

物理服务器：

#include "physicalserver.hpp"

#include <algorithm>
#include <thread>

#include <iostream>

void dnw::PhysicalServer::startListening()
{
    for(auto& pair : ports)
    {
        startListening(pair.first);
    }
}

void dnw::PhysicalServer::stopListening()
{
    for(auto& pair : ports)
    {
        stopListening(pair.first);
    }
}

void dnw::PhysicalServer::startListening(const Port port)
{
    std::get<bool>(ports[port]) = true;

    std::thread thread {listen, this, port};

    thread.detach();
}

void dnw::PhysicalServer::stopListening(const Port port)
{
    std::get<bool>(ports[port]) = false;
}

void dnw::PhysicalServer::listen(const Port port)
{
    using namespace boost::asio::ip;

    static tcp::acceptor acceptor {
        context,
        {tcp::v4(), port}
    };

    while(std::get<bool>(ports[port]))
    {
        Socket socket {context};

        acceptor.accept(socket);

        auto& sockets {std::get<std::vector<Socket>>(ports[port])}; 

        sockets.push_back(std::move(socket));

        std::thread thread {std::get<Processor>(ports[port]), std::ref(sockets.back()), std::ref(*this)};
        
        thread.detach();
    }
}

void dnw::PhysicalServer::setOnConnection(const Port port, const Processor& proc)
{
    std::get<Processor>(ports[port]) = proc;
}

void dnw::PhysicalServer::closeConnection()
{
    for(auto& [key, tuple] : ports)
    {
        closeConnection(key);
    }
}

void dnw::PhysicalServer::closeConnection(Port port)
{
    auto& sockets {std::get<std::vector<Socket>>(ports[port])};

    for(auto& socket : sockets)
    {
        socket.close();
    }

    sockets.clear();
}

void dnw::PhysicalServer::closeConnection(Socket& socket)
{
    using std::get;

    auto& sockets {get<std::vector<Socket>>(ports[getSocketPort(socket)])};

    auto socket_iter {
        std::find_if(
            sockets.begin(),
            sockets.end(),
            [&](auto& s2)
            {
                return s2.local_endpoint() == socket.local_endpoint();
            } 
        )
    };

    if(socket_iter != sockets.end())
    {
        socket.close();
        sockets.erase(socket_iter);
    }
}

size_t dnw::PhysicalServer::getClientCount() const
{
    size_t total {};

    for(auto& pair : ports)
    {
        total += getClientCount(pair.first);
    }

    return total;
}

size_t dnw::PhysicalServer::getClientCount(const Port port) const
{
    return std::get<std::vector<Socket>>(ports.at(port)).size();
}

dnw::Server::Port dnw::PhysicalServer::getSocketPort(const Socket& socket)
{
    return socket.local_endpoint().port();
}

void dnw::PhysicalServer::applyOnClient(const std::function<void(Socket&)>& action)
{
    for(auto& [key, port] : ports)
    {
        applyOnClient(key, action);
    }
}

void dnw::PhysicalServer::applyOnClient(const Port port, const std::function<void(Socket&)>& action)
{
    std::for_each(
        std::get<std::vector<Socket>>(ports[port]).begin(),
        std::get<std::vector<Socket>>(ports[port]).end(),
        action
    );
}

实体客户：

#include "physicalclient.hpp"

#include <thread>

dnw::PhysicalClient::PhysicalClient()
:
    socket(context),
    resolver(context)
{

}

void dnw::PhysicalClient::connect(const std::string_view addr, const std::string_view port)
{
    if(socket.is_open())
    {
        return;
    }

    boost::asio::connect(socket, resolver.resolve(addr, port));

    std::thread thread {processor, std::ref(socket), std::ref(*this)};

    thread.detach();
}

void dnw::PhysicalClient::setOnConnection(const Processor& proc)
{
    processor = proc;
}

void dnw::PhysicalClient::disconnect()
{
    socket.close();
}

dnw::PhysicalClient::~PhysicalClient()
{
    disconnect();
}