前端之家

  1. 首页
  2. 问答

两对共享相同代码的TCP服务器/客户端，但是只有一个客户端接收消息

• 问答

我有两对简单的TCP服务器/客户端，即在Windows上运行的每个服务器一个客户端：

  • 服务器在进程（应用程序）中运行。
  • 客户端在另一个过程中运行。
  • 服务器不断向其配对的客户端发送心跳（字符串）。
  • 第一对服务器/客户端在各自的线程中运行其主循环。
  • 一旦第一个服务器/客户端与第一个心跳握手，第二对服务器/客户端就会在自己的线程中启动其主循环。
  • 对于此测试，它们在具有不同端口的同一台计算机上运行：2345和2346。

现在是我的问题

  • 第一个客户端收到其服务器的心跳。
  • 第二个客户端不会不，尽管第二个服务器发出了没有错误的心跳。

这是服务器代码：

// hello_dualchannel_server.cpp : This file contains the 'main' function. 

#include "pch.h"

#include <iostream>
#include <thread>

#include <winsock2.h>
#include <ws2tcpip.h>

#include <spdlog/spdlog.h>


#define SPDLOG_WCHAR_TO_UTF8_SUPPORT
#ifdef _DEBUG
#if !defined(SPDLOG_actIVE_LEVEL)
#define SPDLOG_actIVE_LEVEL SPDLOG_LEVEL_TRACE
#endif  // #if !defined(SPDLOG_actIVE_LEVEL)
#define SPDLOG_DEBUG_ON
#define SPDLOG_TRACE_ON
#define _trace SPDLOG_TRACE 
#endif  // #ifdef _DEBUG
using namespace spdlog;

SOCKET g_sockFirst = 0;
SOCKET g_sockClientFirst = 0;
std::thread g_threadFirst;
uint32_t g_timeLatestHeartBeatFirst = 0;

SOCKET g_sockSecond = 0;
SOCKET g_sockClientSecond = 0;
std::thread g_threadSecond;
uint32_t g_timeLatestHeartBeatSecond = 0;

void SetupLogger() {
#ifdef _DEBUG
    spdlog::set_level(spdlog::level::trace);
    spdlog::set_pattern("[%H:%M:%S%z][%^%L%$][%t:%s:%#] %v");
#else
    spdlog::set_level(spdlog::level::info);
    spdlog::set_pattern("[%H:%M:%S][%^%L%$][%t] %v");
#endif  // #ifdef _DEBUG
}

int InitWinSock() {
    WORD wVersionRequested;
    WSADATA wsaData;

    /* Use the MAKEWORD(lowbyte,highbyte) macro declared in Windef.h */
    wVersionRequested = MAKEWORD(2,2);

    int err = WSAStartup(wVersionRequested,&wsaData);
    if (err != 0) {
        /* Tell the user that we could not find a usable */
        /* Winsock DLL.                                  */
        printf("WSAStartup failed with error: %d\n",err);
        return 1;
    }

    /* Confirm that the WinSock DLL supports 2.2.*/
    /* Note that if the DLL supports versions greater    */
    /* than 2.2 in addition to 2.2,it will still return */
    /* 2.2 in wVersion since that is the version we      */
    /* requested.                                        */

    if (LOBYTE(wsaData.wVersion) != 2 || HIBYTE(wsaData.wVersion) != 2) {
        /* Tell the user that we could not find a usable */
        /* WinSock DLL.                                  */
        printf("Could not find a usable version of Winsock.dll\n");
        WSACleanup();
        return 1;
    }
    else
        printf("The Winsock 2.2 dll was found okay\n");
    return 0;
}

bool Init(int host_port,SOCKET* p_sockServer,SOCKET*p_sockClient) {
    int err = 0;
    int* p_int = 0;
    std::string host_name("127.0.0.1");
    struct sockaddr_in my_addr;
    int addr_size = 0;
    sockaddr_in sadr_client;

    if (!*p_sockServer) {
        *p_sockServer = socket(AF_INET,SOCK_STREAM,0);
        if (*p_sockServer == -1) {
            char log[MAX_PATH];
            strerror_s(log,MAX_PATH,errno);
            error("Server Error initializing socket: {}",log);
            goto FINISH;
        }

        p_int = (int*)malloc(sizeof(int));
        *p_int = 1;
        if ((setsockopt(*p_sockServer,SOL_SOCKET,SO_REUSEADDR,(char*)p_int,sizeof(int)) == -1)
            || (setsockopt(*p_sockServer,SO_KEEPALIVE,sizeof(int)) == -1)) {
            char log[MAX_PATH];
            strerror_s(log,errno);
            error("Server Error setting options: {}",log);
            free(p_int);
            goto FINISH;
        }
        free(p_int);
        info("Server socket is set up.");

        my_addr.sin_family = AF_INET;
        my_addr.sin_port = htons(host_port);
        memset(&(my_addr.sin_zero),8);
        my_addr.sin_addr.s_addr = INADDR_ANY;

        if (bind(*p_sockServer,(sockaddr*)&my_addr,sizeof(my_addr)) == -1) {
            char log[MAX_PATH];
            strerror_s(log,errno);
            error("Server Error binding to socket,make sure nothing else is listening on this port: {}",log);
            goto FINISH;
        }
        if (listen(*p_sockServer,10) == -1) {
            char log[MAX_PATH];
            strerror_s(log,errno);
            error("Server Error listening: {}",log);
            goto FINISH;
        }
        info("SUCCESS: Server socket listening ...");
    }

    info("Server accepting connection ...");
    addr_size = sizeof(sockaddr_in);
    char sAddress[MAX_PATH];
    *p_sockClient = accept(*p_sockServer,(sockaddr*)&sadr_client,&addr_size);
    if (*p_sockClient == INVALID_SOCKET) {
        char log[MAX_PATH];
        strerror_s(log,errno);
        error("Server error accepting client connection: {}",log);
        // DO NOT close sockets here.
        return false;
    }
    inet_ntop(sadr_client.sin_family,&sadr_client.sin_addr,sAddress,MAX_PATH);
    g_timeLatestHeartBeatFirst = getcurrentTime();
    info("SUCCESS: Server accepted client connection.");
    return true;
FINISH:
    closesocket(*p_sockServer);
    return false;
}

bool IsConnected(uint32_t timeLatestHeartBeat) {
    // CAUTION: denser than client for sure catch
    const unsigned long ConnTimeoutMs = 300;
    auto cur = getcurrentTime();
    auto latest = timeLatestHeartBeat;
    return cur - latest < ConnTimeoutMs;
}

bool StayInTouch(const char* name,SOCKET* pSockClient,uint32_t* pTimeLatestHeartBeat) {
    if (IsConnected(*pTimeLatestHeartBeat))
        return true;
    char heartBeat[] = "biku";
    int nBytesSent = 0;
    int flags = 0;
    int res = send(*pSockClient,heartBeat,sizeof(heartBeat),flags);
    if (res == SOCKET_ERROR) {
        char log[MAX_PATH];
        strerror_s(log,errno);
        error("{}: Server failed to send heartbeat: {},Windows error: {}",name,log,GetLastError());
        return false;
    }
    else if (res == 0) {
        char log[MAX_PATH];
        strerror_s(log,errno);
        error("{}: Server sent zerobyte heartbeat: {}",log);
        return false;
    }
    debug("{}: Heartbeat sent: {}",heartBeat);
    *pTimeLatestHeartBeat = getcurrentTime();
    return true;
}

void Close(SOCKET* pSock) {
    closesocket(*pSock);
    *pSock = 0;
}

bool Connect() {
    if (g_threadFirst.joinable()) {
        warn("FirstTunnel already running. Skipped.");
        return true;
    }
    g_threadFirst = std::thread([&]() {
        bool isConnected = false;
        while (true) {
            while (!isConnected) {
                isConnected = Init(2345,&g_sockFirst,&g_sockClientFirst);
            }
            isConnected = StayInTouch("FirstTunnel",&g_sockClientFirst,&g_timeLatestHeartBeatFirst);
            if (!isConnected) {
                // We don't close as client.
                // We keep connecting               
                error("About to reconnect ...");
                Sleep(1000);
                continue;
            }
            if (!g_threadSecond.joinable()) {
                g_threadSecond = std::thread([&]() {
                    while (true) {
                        while (!isConnected) {
                            isConnected = Init(2346,&g_sockSecond,&g_sockClientSecond);
                        }
                        isConnected = StayInTouch("SecondTunnel",&g_sockClientSecond,&g_timeLatestHeartBeatSecond);
                        if (!isConnected) {
                            // We don't close as client.
                            // We keep connecting               
                            error("About to reconnect ...");
                            Sleep(1000);
                            continue;
                        }
                    }
                    info("SecondTunnel quitting...");
                    Close(&g_sockSecond);
                });
            }
        }
        info("FirstTunnel quitting...");
        Close(&g_sockFirst);
    });

    while (true) {
        //info("main thread ...");
        Sleep(3000);
    }

    return g_threadFirst.joinable() ? true : false;
}


int main() {
    SetupLogger();
    info("Hello World!\n");

    if (InitWinSock()) {
        critical("Failed to initialize Window socket. Aborted.");
    }

    Connect();

    if (g_threadSecond.joinable()) {
        g_threadSecond.join();
    }
    if (g_threadFirst.joinable()) {
        g_threadFirst.join();
    }
    WSACleanup();
    info("Bye!");
}

这是客户端代码

// hello_dualchannel_client.cpp : This file contains the 'main' function. 
//

#include "pch.h"

#include <iostream>
#include <thread>

#include <winsock2.h>
#include <ws2tcpip.h>

#include <spdlog/spdlog.h>

#define SPDLOG_WCHAR_TO_UTF8_SUPPORT
#ifdef _DEBUG
#if !defined(SPDLOG_actIVE_LEVEL)
#define SPDLOG_actIVE_LEVEL SPDLOG_LEVEL_TRACE
#endif  // #if !defined(SPDLOG_actIVE_LEVEL)
#define SPDLOG_DEBUG_ON
#define SPDLOG_TRACE_ON
#define _trace SPDLOG_TRACE 
#endif  // #ifdef _DEBUG
using namespace spdlog;

SOCKET g_sockFirst = 0;
std::thread g_threadFirst;
uint32_t g_timeLatestHeartBeatFirst;

SOCKET g_sockSecond = 0;
std::thread g_threadSecond;
uint32_t g_timeLatestHeartBeatSecond;

void SetupLogger() {
#ifdef _DEBUG
    spdlog::set_level(spdlog::level::trace);
    spdlog::set_pattern("[%H:%M:%S%z][%^%L%$][%t:%s:%#] %v");
#else
    spdlog::set_level(spdlog::level::info);
    spdlog::set_pattern("[%H:%M:%S][%^%L%$][%t] %v");
#endif  // #ifdef _DEBUG
}

int InitWinSock() {
    WORD wVersionRequested;
    WSADATA wsaData;

    /* Use the MAKEWORD(lowbyte,SOCKET* p_sock) {
    int err = 0;
    int* p_int = 0;
    std::string host_name("127.0.0.1");
    struct sockaddr_in my_addr;
    char handshake[] = "hello";
    //int nBytesSent;
    if (!*p_sock) {
        *p_sock = socket(AF_INET,0);
        if (*p_sock == -1) {
            char log[MAX_PATH];
            strerror_s(log,errno);
            error("Client Error initializing socket {}",log);
            goto FINISH;
        }

        p_int = (int*)malloc(sizeof(int));
        *p_int = 1;
        if ((setsockopt(*p_sock,sizeof(int)) == -1)
            || (setsockopt(*p_sock,errno);
            error("Client Error setting options {}",log);
            free(p_int);
            goto FINISH;
        }
        free(p_int);
        info("SUCCESS: Client socket is set up.");
    }

    my_addr.sin_family = AF_INET;
    my_addr.sin_port = htons(host_port);
    memset(&(my_addr.sin_zero),8);
    inet_pton(my_addr.sin_family,host_name.c_str(),&my_addr.sin_addr);
    if (connect(*p_sock,(struct sockaddr*)&my_addr,sizeof(my_addr)) == SOCKET_ERROR) {
        char log[MAX_PATH];
        strerror_s(log,errno);
        error("Client Error connecting socket {}",log);
        Sleep(1000);
        goto FINISH;
    }
    /*nBytesSent = send(g_sockFirst,handshake,sizeof(handshake),0);
    if (nBytesSent <= 0) {
        char log[MAX_PATH];
        strerror_s(log,errno);
        error("Client error sending handshake: {}",log);
        goto FINISH;
    }*/
    g_timeLatestHeartBeatFirst = getcurrentTime();
    info("SUCCESS: Client connected to server.");
    return true;
FINISH:
    closesocket(*p_sock);
    *p_sock = 0;
    return false;
}

bool IsConnected(uint32_t timeLatestHeartBeat) {
    const unsigned long ConnTimeoutMs = 3000;
    auto cur = getcurrentTime();
    auto latest = timeLatestHeartBeat;
    //if (cur - latest > ConnTimeoutMs)
    //{
    //  debug("cur: {},late: {}",cur,latest);
    //} 
    return cur - latest < ConnTimeoutMs;
}

bool StayInTouch(const char* name,SOCKET* pSock,uint32_t* pTimeLatestHeartBeat) {
    // Client checks inbox right away and measure timeout later.
    char heartBeat[MAX_PATH] = { 0 };
    // CAUTION: min 100ms required for receiving heartbeat.
    //const uint32_t TimeoutMS = 100;
    int flags = 0;
    int nBytesRecved = recv(*pSock,flags);
    bool gotHeartbeat = nBytesRecved > 0;
    if (gotHeartbeat) {
        debug("{}: Heartbeat received: {}",heartBeat);
        *pTimeLatestHeartBeat = getcurrentTime();
    }
    return IsConnected(*pTimeLatestHeartBeat);
}

void Close(SOCKET* pSock) {
    closesocket(*pSock);
    *pSock = 0;
}

bool Connect() {
    if (g_threadFirst.joinable()) {
        warn("FirstTunnel already running. Skipped.");
        return true;
    }
    g_threadFirst = std::thread([&]() {
        bool isConnected = false;
        while (true) {
            while (!isConnected) {
                isConnected = Init(2345,&g_sockFirst);
            }
            isConnected = StayInTouch("FirstTunnel",&g_timeLatestHeartBeatFirst);
            if (!isConnected) {
                // We don't close as client.
                // We keep connecting
                Close(&g_sockFirst);
                error("About to reconnect ...");
                continue;
            }
            if (!g_threadSecond.joinable()) {
                g_threadSecond = std::thread([&]() {
                    while (true) {
                        while (!isConnected) {
                            isConnected = Init(2346,&g_sockSecond);
                        }
                        isConnected = StayInTouch("SecondTunnel",&g_timeLatestHeartBeatSecond);
                        if (!isConnected) {
                            // We don't close as client.
                            // We keep connecting               
                            error("About to reconnect ...");
                            Sleep(1000);
                            continue;
                        }
                    }
                    info("SecondTunnel quitting...");
                    Close(&g_sockSecond);
                });
            }
        }
        info("FirstTunnel quitting.");
        Close(&g_sockFirst);
    });

    while (true) {
        //info("main thread ...");
        Sleep(3000);
    }

    return g_threadFirst.joinable() ? true : false;
}

int main() {
    SetupLogger();
    info("Hello World!\n");

    if (InitWinSock()) {
        critical("Failed to initialize Window socket. Aborted.");
    }

    Connect();

    if (g_threadSecond.joinable()) {
        g_threadSecond.join();
    }
    if (g_threadFirst.joinable()) {
        g_threadFirst.join();
    }
    WSACleanup();
    info("Bye!");
}

主要连接逻辑在函数Connect()中。 我会感谢我错在哪里的提示。

要按原样运行代码，您需要一个依赖项spdlog 

vcpkg install spdlog:x64-Windows

您还可以用自己的代码替换所有基于spdlog的日志记录代码。

更新

观察＃1

我进入代码并确认

  • 所有循环都在运行。这样就产生了所有线程。
  • 由于joinable()的保护，没有产生多余的线程。
  • 唯一的失败点是第二个客户端的recv调用。

结论如此

  • 没有防火墙
  • 没有多余的线程
  • 这两个线程都是根据设计在客户端和服务器端运行的。

观察＃2

在运行服务器和客户端程序并让问题发生时，我尝试了

  • 保持两个程序都打开。
  • 像这样netcat运行netcat（nmap的C:\Apps\nmap\ncat.exe 127.0.0.1 2346端口）

这实际上有助于解决问题。它告诉我存在连接问题，而我肯定可以进入客户端代码并看到连接仍然存在。

观察＃3

  • 仅在客户端程序的第二个连接代码中保留断点并运行该程序之后，我无法进入第一个连接代码。请注意，它们位于单独的线程中。
  • 在两个连接中都保留断点，并在它们之间快速过渡，我可以继续进行而不会陷入一个线程中。 这是我注意到客户端开始接收其应收到的消息的时候。

因此，我怀疑那里存在线程问题。如果那是问题，那么如何解决呢？

dmayahot777 回答：两对共享相同代码的TCP服务器/客户端，但是只有一个客户端接收消息

我自己发现了问题。对我而言，这是一个愚蠢的错误：

两个线程错误地共享一个状态：isConnected。因此，其中一个原始程序获得新的连接或心跳后，它们会继续互相踢。他们应该为此使用不同的状态。

要提供更多细节，由isConnected初始化的g_threadFirst属于g_threadSecond匿名函数的关闭。

c++multithreadingnetworkingsockets
本文链接：https://www.f2er.com/3139199.html

大家都在问