Winsock API - 在两个线程中同时发送和接收数据

Question

我正在尝试在同一个程序中构建客户端和服务器。例如，用户 1 向用户 2 发送了一个数据包，用户 2 收到数据包后向用户 1 发回不同的数据包。问题是，在运行程序后，用户都没有收到数据包。如果我将客户端构建在将程序与其工作的服务器分开

#pragma comment(lib,"ws2_32.lib")
#include <WinSock2.h>
#include <iostream>
#include <thread>

static const int num_threads = 2;

static char buffer[8096 + 1];
char a[256] = {};
char MOTD[256];

void call_from_thread(int tid) {
//  ---------- Server code: ---------- //
if( tid == 0 ){

WSAData wsaData;
WORD DllVersion = MAKEWORD(2, 1);
if (WSAStartup(DllVersion, &wsaData) != 0){MessageBoxA(NULL, "WinSock startup failed", "Error", MB_OK | MB_ICONERROR);}

    SOCKADDR_IN addr;
    int addrlen = sizeof(addr);
    addr.sin_addr.s_addr = inet_addr("127.0.0.1");
    addr.sin_port = htons(1112);
    addr.sin_family = AF_INET;

    SOCKET sListen = socket(AF_INET, SOCK_STREAM, NULL);
    bind(sListen, (SOCKADDR*)&addr, sizeof(addr));
    listen(sListen, SOMAXCONN);

    SOCKET newConnection;
    newConnection = accept(sListen, (SOCKADDR*)&addr, &addrlen);
    if (newConnection == 0){
        std::cout << "Failed to accept the client's connection." << std::endl;
    }else{
        std::cout << "Client Connected!" << std::endl;
    }

    while (true){
        std::cin >> a;
        send(newConnection, a, sizeof(a), NULL);
    }
}else if( tid == 1 ){


//  ---------- Client code: ---------- //
    WSAData wsaData;
    WORD DllVersion = MAKEWORD(2, 1);
    if (WSAStartup(DllVersion, &wsaData) != 0){MessageBoxA(NULL, "Winsock startup failed", "Error", MB_OK | MB_ICONERROR);}

    SOCKADDR_IN addr;
    int sizeofaddr = sizeof(addr);
    addr.sin_addr.s_addr = inet_addr("127.0.0.1");
    addr.sin_port = htons(1111);
    addr.sin_family = AF_INET;

    SOCKET Connection = socket(AF_INET, SOCK_STREAM, NULL); //Set Connection socket
    if (connect(Connection, (SOCKADDR*)&addr, sizeofaddr) != 0) //If we are unable to connect...
    {
        MessageBoxA(NULL, "Failed to Connect", "Error", MB_OK | MB_ICONERROR);
        //return 0; //Failed to Connect
    }

    std::cout << "Connected!" << std::endl;

    char MOTD[256];
    while (true){
        recv(Connection, MOTD, sizeof(MOTD), NULL); //Receive Message of the Day buffer into MOTD array
        std::cout << "MOTD:" << MOTD << std::endl;
    }
}



//  ---------- Thread selection: ---------- //
int main() {
std::thread t[num_threads];

for (int i = 0; i < num_threads; ++i) {
    t[i] = std::thread(call_from_thread, i);
 }

 for (int i = 0; i < num_threads; ++i) {
     t[i].join();
 }

}

Answer 1

服务器正在监听 1112 端口，但客户端正在连接到 1111 端口。这意味着客户端和服务器根本不可能相互连接，无论它们是否在同一个应用程序中。

我还发现您的代码存在许多其他问题，包括：

• WSAStartup() 应该在应用启动时调用一次，而不是每个线程。

• 如果任何 WinSock 函数失败，两个线程都不会完全退出。

• 客户端线程不会等待服务器线程打开其侦听端口，然后客户端才能连接到它。

• 潜在的缓冲区溢出。

• 资源泄漏。

• 缺乏体面的错误处理。

  • 在服务器端，您完全忽略了socket()、bind()、listen() 和send() 报告的错误，并且您没有正确检查accept() 的返回值。
  • 在客户端，您完全忽略了socket() 和recv() 报告的错误。而且，在将 std::cout 打印到 std::cout 之前，您不能确保 MOTD 缓冲区是空终止的。

试试这样的：

#include <winsock2.h>
#include <windows.h>
#pragma comment(lib,"ws2_32.lib")

#include <iostream>
#include <thread>
#include <string>
#include <cstdint>
#include <stdexcept>
#include <limits>
#include <algorithm>

// The below variables are used to let the client wait for the server
// port to be opened. If you don't want to use these, especially if the
// client and server are ever run on different machines, you can omit
// these and instead just have the client call connect() in a loop
// until successful...
//
#include <mutex>
#include <condition_variable>
#include <chrono>
static std::condition_variable server_cv;
static std::mutex server_mtx;
static bool server_is_listening = false;
//

static const int num_threads = 2;
static const u_short server_port = 1111;

class winsock_error : public std::runtime_error
{
public:
    int errCode;
    std::string funcName;
    winsock_error(int errCode, const char *funcName) : std::runtime_error("WinSock error"), errCode(errCode), funcName(funcName) {}
};

void WinSockError(const char *funcName, int errCode = WSAGetLastError())
{
    throw winsock_error(errCode, funcName);
}

class connection_closed : public std::runtime_error
{
public:
    connection_closed() : std::runtime_error("Connection closed") {}
};

class socket_ptr
{
public:
    socket_ptr(SOCKET s) : m_sckt(s) {}
    socket_ptr(const socket_ptr &) = delete;
    socket_ptr(socket_ptr &&src) : m_sckt(src.m_sckt) { src.m_sckt = INVALID_SOCKET; }
    ~socket_ptr() { if (m_sckt != INVALID_SOCKET) closesocket(m_sckt); }

    socket_ptr& operator=(const socket_ptr &) = delete;
    socket_ptr& operator=(socket_ptr &&rhs) { m_sckt = rhs.m_sckt; rhs.m_sckt = INVALID_SOCKET; return *this; }

    operator SOCKET() { return m_sckt; }
    bool operator!() const { return (m_sckt == INVALID_SOCKET); }

private:
    SOCKET m_sckt;
}

template <typename T>
T LimitBufferSize(size_t size)
{
    return (T) std::min(size, (size_t) std::numeric_limits<T>::max());
}

void sendRaw(SOCKET sckt, const void *buffer, size_t buflen)
{
    const char *ptr = static_cast<const char*>(buffer);
    while (buflen > 0)
    {
        int numToSend = LimitBufferSize<int>(buflen);
        int numSent = ::send(sckt, ptr, numToSend, 0);
        if (numSent == SOCKET_ERROR)
            WinSockError("send");
        ptr += numSent;
        buflen -= numSent;
    }
}

void recvRaw(SOCKET sckt, void *buffer, size_t buflen)
{
    char *ptr = static_cast<char*>(buffer);
    while (buflen > 0)
    {
        int numToRecv = LimitBufferSize<int>(buflen);
        int numRecvd = ::recv(sckt, ptr, numToRecv, 0);
        if (numRecvd == SOCKET_ERROR)
            WinSockError("recv");
        if (numRecvd == 0)
            throw connection_closed();
        ptr += numRecvd;
        buflen -= numRecvd;
    }
}

void sendStr(SOCKET sckt, const std::string &str)
{
    uint32_t len = LimitBufferSize<uint32_t>(str.size());
    uint32_t tmp = htonl(len);
    sendRaw(sckt, &tmp, sizeof(tmp));
    if (len > 0)
        sendRaw(sckt, str.c_str(), len);
}

std::string recvStr(SOCKET sckt)
{
    std::string str;
    uint32_t len;
    recvRaw(sckt, &len, sizeof(len));
    len = ntohl(len);
    if (len > 0)
    {
        str.resize(len);
        recvRaw(sckt, &str[0], len);
    }
    return str;
}

void server_thread()
{
    std::cout << "[Server] Starting ..." << std::endl;

    try
    {
        socket_ptr sListen(::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
        if (!sListen)
            WinSockError("socket");

        SOCKADDR_IN addr = {};
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = ::inet_addr("127.0.0.1");
        addr.sin_port = ::htons(server_port);

        int addrlen = sizeof(addr);

        if (::bind(sListen, (SOCKADDR*)&addr, sizeof(addr)) == SOCKET_ERROR)
            WinSockError("bind");

        if (::listen(sListen, 1) == SOCKET_ERROR)
            WinSockError("listen");

        // this is optional...
        {
        std::unique_lock<std::mutex> lk(server_mtx);
        server_is_listening = true;
        }
        server_cv.notify_all();
        //

        std::cout << "[Server] Listening for Client ..." << std::endl;

        socket_ptr newConnection(::accept(sListen, (SOCKADDR*)&addr, &addrlen));
        if (!newConnection)
            WinSockError("accept");

        std::cout << "[Server] Client Connected!" << std::endl;

        try
        {
            std::string a;
            while (std::cin >> a)
                sendStr(newConnection, a);
        }
        catch (const connection_closed &)
        {
        }
        catch (const winsock_error &e)
        {
            std::cerr << "[Server] Client error: " << e.errCode << " on WinSock function: " << e.funcName << std::endl;
        }

        std::cout << "[Server] Client Disconnected!" << std::endl;
    }
    catch (const winsock_error &e)
    {
        std::cerr << "[Server] Error: " << e.errCode << " on WinSock function: " << e.funcName << std::endl;
    }
    catch (const std::exception &e)
    {
        std::cerr << "[Server] Unexpected Error! " << e.what() << std::endl;
    }

    std::cout << "[Server] Stopped!" << std::endl;
}

void client_thread()
{
    std::cout << "[Client] Starting ..." << std::endl;

    try
    {
        // this is optional, could call connect() below in a loop instead...
        std::cout << "[Client] Waiting for Server ..." << std::endl;
        {
        std::unique_lock<std::mutex> lk(server_mtx);
        if (!server_cv.wait_for(lk, std::chrono::seconds(5), []{ return server_is_listening; }))
            throw std::runtime_error("Server not listening after 5 seconds!");
        }
        //

        socket_ptr sConnection(::socket(AF_INET, SOCK_STREAM, IPPROTO_TCP));
        if (!sConnection)
            WinSockError("socket");

        SOCKADDR_IN addr = {};
        addr.sin_family = AF_INET;
        addr.sin_addr.s_addr = ::inet_addr("127.0.0.1");
        addr.sin_port = ::htons(server_port);

        if (::connect(sConnection, (SOCKADDR*)&addr, sizeof(addr)) == SOCKET_ERROR)
            WinSockError("connect");

        std::cout << "[Client] Connected!" << std::endl;

        try
        {
            std::string MOTD;
            while (true)
            {
                MOTD = recvStr(sConnection);
                std::cout << "[Client] MOTD: " << MOTD << std::endl;
            }
        }
        catch (const connection_closed &)
        {
        }
        catch (const winsock_error &e)
        {
           std::cerr << "[Client] Server error: " << e.errCode << " on WinSock function: " << e.funcName << std::endl;
        } 

        std::cout << "[Client] Disconnected!" << std::endl;
    }
    catch (const winsock_error &e)
    {
        std::cerr << "[Client] Error: " << e.errCode << " on WinSock function: " << e.funcName << std::endl;
    }
    catch (const std::exception &e)
    {
        std::cerr << "[Client] Unexpected Error! " << e.what() << std::endl;
    }

    std::cout << "[Client] Stopped!" << std::endl;
}

int main()
{
    WSADATA wsaData;

    int ret = ::WSAStartup(MAKEWORD(2, 1), &wsaData);
    if (ret != 0)
    {
        std::cerr << "WinSock Startup failed with error: " << ret << std::endl;
        return -1;
    }

    std::cout << "WinSock Startup successful" << std::endl;

    std::thread t[num_threads];

    t[0] = std::thread(server_thread);
    t[1] = std::thread(client_thread);

    for (int i = 0; i < num_threads; ++i) {
        t[i].join();
    }

    ::WSACleanup();
    return 0;
}