c++ - 两对 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, SOL_SOCKET, SO_KEEPALIVE, (char*)p_int, sizeof(int)) == -1)) {
char log[MAX_PATH];
strerror_s(log, MAX_PATH, 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), 0, 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, MAX_PATH, 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, MAX_PATH, 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, MAX_PATH, 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, MAX_PATH, 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, MAX_PATH, errno);
error("{}: Server sent zerobyte heartbeat: {}", name, log);
return false;
}
debug("{}: Heartbeat sent: {}", name, 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, 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_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, SOCK_STREAM, 0);
if (*p_sock == -1) {
char log[MAX_PATH];
strerror_s(log, MAX_PATH, errno);
error("Client Error initializing socket {}", log);
goto FINISH;
}
p_int = (int*)malloc(sizeof(int));
*p_int = 1;
if ((setsockopt(*p_sock, SOL_SOCKET, SO_REUSEADDR, (char*)p_int, sizeof(int)) == -1)
|| (setsockopt(*p_sock, SOL_SOCKET, SO_KEEPALIVE, (char*)p_int, sizeof(int)) == -1)) {
char log[MAX_PATH];
strerror_s(log, MAX_PATH, 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), 0, 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, MAX_PATH, 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, MAX_PATH, 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, heartBeat, sizeof(heartBeat), flags);
bool gotHeartbeat = nBytesRecved > 0;
if (gotHeartbeat) {
debug("{}: Heartbeat received: {}", name, 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_sockFirst, &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_sockSecond, &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(NMap的
netcat
端口)C:\Apps\Nmap\ncat.exe 127.0.0.1 2346
这实际上有助于解决问题。它告诉我存在连接问题,而我绝对可以进入客户端代码并看到连接仍然存在。
观察#3
- 在我只在客户端程序的第二个连接代码中留下断点并运行程序后,我无法进入第一个连接代码。请注意,它们位于不同的线程中。
- 在两个连接中都留下断点,并在它们之间快速跨过,我可以继续前进,而不会被困在一个线程中。这是我注意到客户端开始接收它应该接收的消息的时候。
所以我怀疑那里存在线程问题。如果这是问题所在,那么如何解决这个问题?
解决方案
我自己发现了问题。这是我的一个愚蠢的错误:
两个线程错误地共享一个状态:isConnected
. 因此,在其中一个获得新的连接或心跳之后,原始程序会继续互相踢。他们应该为此使用不同的状态。
提供更多细节:isConnected
由 初始化的,通过的匿名函数g_threadFirst
的闭包。g_threadSecond
推荐阅读
- docker - 有没有办法自动保存 Docker 容器日志?
- svn - 为什么 svn update 会丢弃本地更改?
- pivot - Vlookup 查找两个项目与数据透视表一起出现的实例数
- java - 使用 cellEditor() 删除带有 jButton 的 jTable 行;
- python - 时间范围检查应该包含还是排除?
- android - 如何在 Android 上的 RxJava 中通过 flatmap 发送自定义的 throwable?
- spring - 使用 Spring Data JPA - 如何在保存父实体和使用 CascadeType=PERSIST 的同时仅保存子实体中的唯一项目
- php - 有时缺少对象索引的填充数据
- reactjs - 我无法使用 react-select 处理更改
- android - 如何使用 Android 的 ImageDecoder 类获取图像大小?