ZeroMQ 经销商 - 与 winsock 相比，对经销商的高延迟
ZeroMQ dealer--to-dealer high latency compared to winsock
我的公司正在考虑使用ZeroMQ作为传输机制。首先，我对性能进行了基准测试，只是为了预感我正在玩什么。
因此，我创建了一个应用程序，将 zmq 经销商到经销商的设置与 winsock 进行比较。我测量了从客户端向服务器发送同步消息然后计算平均值的往返行程。
这里运行 winsock 的服务器：
DWORD RunServerWINSOCKTest(DWORD dwPort)
{
WSADATA wsaData;
int iRet = WSAStartup(MAKEWORD(2, 2), &wsaData);
if (iRet != NO_ERROR)
{
printf("WSAStartup failed with error: %dn", iRet);
return iRet;
}
struct addrinfo hints;
ZeroMemory(&hints, sizeof(hints));
hints.ai_family = AF_INET;
hints.ai_socktype = SOCK_STREAM;
hints.ai_protocol = IPPROTO_TCP;
hints.ai_flags = AI_PASSIVE;
struct addrinfo *result = NULL;
iRet = getaddrinfo(NULL, std::to_string(dwPort).c_str(), &hints, &result);
if (iRet != 0)
{
WSACleanup();
return iRet;
}
SOCKET ListenSocket = socket(result->ai_family, result->ai_socktype, result->ai_protocol);
if (ListenSocket == INVALID_SOCKET)
{
freeaddrinfo(result);
WSACleanup();
return WSAGetLastError();
}
iRet = bind(ListenSocket, result->ai_addr, (int)result->ai_addrlen);
if (iRet == SOCKET_ERROR)
{
freeaddrinfo(result);
closesocket(ListenSocket);
WSACleanup();
return WSAGetLastError();
}
freeaddrinfo(result);
iRet = listen(ListenSocket, SOMAXCONN);
if (iRet == SOCKET_ERROR)
{
closesocket(ListenSocket);
WSACleanup();
return WSAGetLastError();
}
while (true)
{
SOCKET ClientSocket = accept(ListenSocket, NULL, NULL);
if (ClientSocket == INVALID_SOCKET)
{
closesocket(ListenSocket);
WSACleanup();
return WSAGetLastError();
}
char value = 0;
setsockopt(ClientSocket, IPPROTO_TCP, TCP_NODELAY, &value, sizeof(value));
char recvbuf[DEFAULT_BUFLEN];
int recvbuflen = DEFAULT_BUFLEN;
do {
iRet = recv(ClientSocket, recvbuf, recvbuflen, 0);
if (iRet > 0) {
// Echo the buffer back to the sender
int iSendResult = send(ClientSocket, recvbuf, iRet, 0);
if (iSendResult == SOCKET_ERROR)
{
closesocket(ClientSocket);
WSACleanup();
return WSAGetLastError();
}
}
else if (iRet == 0)
printf("Connection closing...n");
else {
closesocket(ClientSocket);
WSACleanup();
return 1;
}
} while (iRet > 0);
iRet = shutdown(ClientSocket, SD_SEND);
if (iRet == SOCKET_ERROR)
{
closesocket(ClientSocket);
WSACleanup();
return WSAGetLastError();
}
closesocket(ClientSocket);
}
closesocket(ListenSocket);
return WSACleanup();
}
下面是运行 winsock 的客户端：
DWORD RunClientWINSOCKTest(std::string strAddress, DWORD dwPort, DWORD dwMessageSize)
{
WSADATA wsaData;
int iRet = WSAStartup(MAKEWORD(2, 2), &wsaData);
if (iRet != NO_ERROR)
{
return iRet;
}
SOCKET ConnectSocket = INVALID_SOCKET;
struct addrinfo *result = NULL, *ptr = NULL, hints;
ZeroMemory(&hints, sizeof(hints));
hints.ai_family = AF_UNSPEC;
hints.ai_socktype = SOCK_STREAM;
hints.ai_protocol = IPPROTO_TCP;
int iResult = getaddrinfo(strAddress.c_str(), std::to_string(dwPort).c_str(), &hints, &result);
if (iResult != 0) {
WSACleanup();
return 1;
}
for (ptr = result; ptr != NULL; ptr = ptr->ai_next) {
ConnectSocket = socket(ptr->ai_family, ptr->ai_socktype, ptr->ai_protocol);
if (ConnectSocket == INVALID_SOCKET) {
WSACleanup();
return 1;
}
iResult = connect(ConnectSocket, ptr->ai_addr, (int)ptr->ai_addrlen);
if (iResult == SOCKET_ERROR) {
closesocket(ConnectSocket);
ConnectSocket = INVALID_SOCKET;
continue;
}
break;
}
freeaddrinfo(result);
if (ConnectSocket == INVALID_SOCKET) {
WSACleanup();
return 1;
}
// Statistics
UINT64 uint64BytesTransmitted = 0;
UINT64 uint64StartTime = s_TimeStampGenerator.GetHighResolutionTimeStamp();
UINT64 uint64WaitForResponse = 0;
DWORD dwMessageCount = 1000000;
CHAR cRecvMsg[DEFAULT_BUFLEN];
SecureZeroMemory(&cRecvMsg, DEFAULT_BUFLEN);
std::string strSendMsg(dwMessageSize, 'X');
for (DWORD dwI = 0; dwI < dwMessageCount; dwI++)
{
int iRet = send(ConnectSocket, strSendMsg.data(), strSendMsg.size(), 0);
if (iRet == SOCKET_ERROR) {
closesocket(ConnectSocket);
WSACleanup();
return 1;
}
uint64BytesTransmitted += strSendMsg.size();
UINT64 uint64BeforeRespone = s_TimeStampGenerator.GetHighResolutionTimeStamp();
iRet = recv(ConnectSocket, cRecvMsg, DEFAULT_BUFLEN, 0);
if (iRet < 1)
{
closesocket(ConnectSocket);
WSACleanup();
return 1;
}
std::string strMessage(cRecvMsg);
if (strMessage.compare(strSendMsg) == 0)
{
uint64WaitForResponse += (s_TimeStampGenerator.GetHighResolutionTimeStamp() - uint64BeforeRespone);
}
else
{
return NO_ERROR;
}
}
UINT64 uint64ElapsedTime = s_TimeStampGenerator.GetHighResolutionTimeStamp() - uint64StartTime;
PrintResult(uint64ElapsedTime, uint64WaitForResponse, dwMessageCount, uint64BytesTransmitted, dwMessageSize);
iResult = shutdown(ConnectSocket, SD_SEND);
if (iResult == SOCKET_ERROR) {
closesocket(ConnectSocket);
WSACleanup();
return 1;
}
closesocket(ConnectSocket);
return WSACleanup();
}
这是运行ZMQ(经销商(的服务器
DWORD RunServerZMQTest(DWORD dwPort)
{
try
{
zmq::context_t context(1);
zmq::socket_t server(context, ZMQ_DEALER);
// Set options here
std::string strIdentity = s_set_id(server);
printf("Created server connection with ID: %sn", strIdentity.c_str());
std::string strConnect = "tcp://*:" + std::to_string(dwPort);
server.bind(strConnect.c_str());
bool bRunning = true;
while (bRunning)
{
std::string strMessage = s_recv(server);
if (!s_send(server, strMessage))
{
return NO_ERROR;
}
}
}
catch (zmq::error_t& e)
{
return (DWORD)e.num();
}
return NO_ERROR;
}
这是运行ZMQ的客户端(经销商(
DWORD RunClientZMQTest(std::string strAddress, DWORD dwPort, DWORD dwMessageSize)
{
try
{
zmq::context_t ctx(1);
zmq::socket_t client(ctx, ZMQ_DEALER); // ZMQ_REQ
// Set options here
std::string strIdentity = s_set_id(client);
std::string strConnect = "tcp://" + strAddress + ":" + std::to_string(dwPort);
client.connect(strConnect.c_str());
if(s_send(client, "INIT"))
{
std::string strMessage = s_recv(client);
if (strMessage.compare("INIT") == 0)
{
printf("Client[%s] connected to: %sn", strIdentity.c_str(), strConnect.c_str());
}
else
{
return NO_ERROR;
}
}
else
{
return NO_ERROR;
}
// Statistics
UINT64 uint64BytesTransmitted = 0;
UINT64 uint64StartTime = s_TimeStampGenerator.GetHighResolutionTimeStamp();
UINT64 uint64WaitForResponse = 0;
DWORD dwMessageCount = 10000000;
std::string strSendMsg(dwMessageSize, 'X');
for (DWORD dwI = 0; dwI < dwMessageCount; dwI++)
{
if (s_send(client, strSendMsg))
{
uint64BytesTransmitted += strSendMsg.size();
UINT64 uint64BeforeRespone = s_TimeStampGenerator.GetHighResolutionTimeStamp();
std::string strRecvMsg = s_recv(client);
if (strRecvMsg.compare(strSendMsg) == 0)
{
uint64WaitForResponse += (s_TimeStampGenerator.GetHighResolutionTimeStamp() - uint64BeforeRespone);
}
else
{
return NO_ERROR;
}
}
else
{
return NO_ERROR;
}
}
UINT64 uint64ElapsedTime = s_TimeStampGenerator.GetHighResolutionTimeStamp() - uint64StartTime;
PrintResult(uint64ElapsedTime, uint64WaitForResponse, dwMessageCount, uint64BytesTransmitted, dwMessageSize);
}
catch (zmq::error_t& e)
{
return (DWORD)e.num();
}
return NO_ERROR;
}
我在本地运行基准测试，消息大小为 5 字节，我得到以下结果：
温索克
Messages sent: 1 000 000
Time elapsed (us): 48 019 415
Time elapsed (s): 48.019 415
Message size (bytes): 5
Msg/s: 20 825
Bytes/s: 104 125
Mb/s: 0.099
Total response time (us): 24 537 376
Average repsonse time (us): 24.0
和
零MQ
Messages sent: 1 000 000
Time elapsed (us): 158 290 708
Time elapsed (s): 158.290 708
Message size (bytes): 5
Msg/s: 6 317
Bytes/s: 31 587
Mb/s: 0.030
Total response time (us): 125 524 178
Average response time (us): 125.0
谁能解释为什么使用 ZMQ 时平均响应时间要长得多？
目标是找到一个设置，我可以异步发送和接收消息，而无需回复。如果这可以通过与经销商-经销商不同的设置来实现，请告诉我！
这只是对你问题的一小部分的回答，但这里是 -
为什么需要经销商/经销商？我假设是因为沟通可以从任何一点开始？您不依赖于经销商/经销商，特别是它将您限制为只能两个端点，如果您在通信的任一端添加另一个端点，例如第二个客户端，那么每个客户端将只收到一半的消息，因为经销商是严格的循环。
异步通信需要的是经销商和/或路由器套接字的某种组合。 两者都不需要响应，主要区别在于它们如何选择将消息发送到哪个连接的对等体：
- 庄家，如前所述，是严格的循环，它会串联发送到每个连接的对等体
- 路由器严格来说是一条寻址消息，您必须知道要发送到的对等方的"名称"才能在那里获取消息。
，经销商是"请求类型"套接字(将其"名称"作为消息的一部分发送，路由器套接字可以使用该套接字发回数据。这是一个请求/回复范式，您将在指南中的所有示例中看到强制执行的这种范式，但您可以将该范式弯曲到您要查找的内容，特别是经销商和路由器都不需要回复。
在不知道您的全部要求的情况下，我无法告诉您我会选择哪种 ZMQ 架构，但总的来说，我更喜欢路由器套接字的可扩展性，处理适当的寻址比将所有内容硬塞到单个对等体中更容易......你会看到关于做路由器/路由器的警告，我同意他们的观点，在尝试之前你应该了解你在做什么，但了解你在做什么，实现并不难。
如果
符合您的要求，您还可以选择将每一端设置一个酒吧插座，如果实际上没有回复，则每个端都有一个子插座。 如果严格来说，它是从源到目标的数据馈送，并且两个对等方都不需要任何关于它发送内容的反馈，那么这可能是最好的选择，即使这意味着你每端要处理两个套接字而不是一个套接字。
这些都不直接解决性能问题，但重要的是要了解 zmq 套接字针对特定用例进行了优化，正如 John Jefferies 的回答中所指出的，通过使测试中的消息传递严格同步，您正在打破经销商套接字的用例。 首先要开始的是完成 ZMQ 架构，然后模拟实际的消息流，特别是不添加任意等待和同步性，这必然会改变吞吐量在您测试它时的外观，几乎是按照定义。
您说要异步发送和接收消息而无需回复。然而，到目前为止所做的测试都是完全同步的，基本上是请求-回复，但在经销商-经销商插槽上。那里有些东西没有计算。为什么不运行更能模拟您目标设计的测试呢？
ZeroMQ通过将排队的消息聚合到单个消息中来获得相当多的"比TCP更快"的性能。显然，该机制不能在一次只有一个消息在传输的纯同步设计中激活。
至于为什么这种非常小的消息纯粹同步发送和接收的特殊测试相对较慢，我不能说。您是否进行了剖析？我要再次说的是，运行这个测试并基于它做出决策是没有意义的，如果它看起来不像你的最终设计。
看起来很奇怪的一件事是 ZeroMQ 代码中的 try/catch 块。这看起来不公平，因为winsock测试不是这样写的。众所周知，在尝试/捕获中存在/曾经有相当多的开销。
OP 问题是吞吐量问题，而不是延迟问题，并且可能是所提供示例中使用的模式问题。但是，您可能总是会发现ZeroMQ具有更高的延迟，我将对此进行解释，尽管在这种情况下它可能对 OP 没有用。
ZeroMQ通过缓冲消息来工作。想象一下(作为一个基本的例证(创建一个std::string并附加许多小字符串(数千个，每个都包含一个小标头以了解这些小段的大小(，然后以100us、 1000us、 10ms或其他间隔发送这个较大的字符串。在接收端，接收大字符串，并根据随其发送的大小标头一次删除一条较小的消息。这允许您批量发送数百万条消息(尽管std::string显然是一个糟糕的选择(，而无需一次发送数百万条非常小的度量值。因此，您可以充分利用网络资源并提高吞吐量，还可以创建基本的FIFO行为。但是，您还会创建延迟以允许缓冲区填充，这意味着延迟增加。
想象一下(再次，作为一个基本说明(：如果你花半秒钟(包括字符串操作等(缓冲一百万条消息，这将导致一个几兆字节的更大字符串。现代网络可以在剩下的半秒内轻松发送这个更大的字符串。 1000000us(1 秒(/每条消息将1us 1000000 条消息，对吗？错误 - 所有消息都有半秒的延迟以允许队列填满，从而导致所有消息的延迟增加长达半秒。 ZeroMQ发送批处理的速度比每500ms快得多，但这说明延迟的增加仍然发生在ZeroMQ中，尽管它通常沿着几个ms的路线。
- 如何仅为一个函数添加延迟
- 计算每个节点的树高，帮助我解释这个代码解决方案
- 当使用比格式支持的精度更高的精度来显示数字时，会写出什么数据
- 以在Qt中的IF语句中设置时间延迟
- C++高阶模板
- 获取 SFML 窗口的 HWND 和高可用性?
- 模板化类中静态成员的延迟初始化
- 在 Windows 8/10 技术中完全实时的屏幕捕获，没有延迟
- Opencv 恢复到比我设置的更高的分辨率
- 将自定义函数传递到基抽象类中以延迟执行
- 输入较高值时的分段代码
- 高精度双精度的 Sprintf 格式化问题
- 我希望改进或要求我目前的延迟/睡眠方法。C++
- 如何在 c++ 中延迟?
- 在Windows(C++）中使用USB相机拍摄高分辨率照片
- 较高值 n 的分割错误(例如 n=999997）
- 在 C++11 及更高版本中，有没有办法初始化初始值设定项列表中的向量?
- 为什么延迟太高，即使只是 RGB 到灰色转换 (Vivado HLS)？
- STM32(使用Mbed在线）在较高模拟输入频率下显示延迟
- ZeroMQ 经销商 - 与 winsock 相比，对经销商的高延迟
