ZMQ 中的 EAGAIN 扩展请求回复
EAGAIN in ZMQ extended request reply
我正在尝试用 C++ 创建一个 REQ <--> 路由器 <--> 经销商 <--> REP 通信。子进程绑定路由器和经销商,路由器和经销商之间的代理,将 REP 连接到经销商并等待 zmq_recv.
的消息
父进程将 REQ 连接到路由器并尝试发送消息,但我收到 zmq_send error in parent: Resource temporarily unavailable
(即 EAGAIN
)。根据 zmq_send 文档,EAGAIN
表示:
Non-blocking mode was requested and the message cannot be sent at the moment.
然而,消息 确实 被发送了,因为它是在子进程中接收到的。为什么 return 那个错误号?
这里是 MCVE:
#include <zmq.h>
#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <assert.h>
#include <thread>
#include <stdio.h>
int main() {
char connect_path[35];
int rc;
int msg;
pid_t child_pid = fork();
if (child_pid == 0) {
// Child
void* child_context = zmq_ctx_new ();
if (child_context == NULL) {
std::cerr << "\nChild context error\n";
}
void* router = zmq_socket(child_context, ZMQ_ROUTER);
if (router == NULL) {
perror("zmq_socket of type router error");
}
char bind_path[35];
snprintf(bind_path, sizeof(bind_path), "ipc:///tmp/zmqtest%d-router", getpid());
rc = zmq_bind(router, bind_path);
assert (rc == 0);
void* dealer = zmq_socket(child_context, ZMQ_DEALER);
if (dealer == NULL) {
perror("zmq_socket of type dealer error");
}
snprintf(bind_path, sizeof(bind_path), "ipc:///tmp/zmqtest%d-dealer", getpid());
rc = zmq_bind(dealer, bind_path);
assert (rc == 0);
std::thread z_proxy (zmq_proxy, router, dealer, nullptr);
z_proxy.detach();
void* rep_socket = zmq_socket (child_context, ZMQ_REP);
if (rep_socket == NULL) {
perror("zmq_socket of type rep error");
}
snprintf(connect_path, sizeof(connect_path), "ipc:///tmp/zmqtest%d-dealer", getpid());
rc = zmq_connect(rep_socket, connect_path);
assert (rc == 0);
while(1) {
if (zmq_recv (rep_socket, &msg, sizeof(msg), 0) != 0) {
perror("zmq_recv error");
}
printf("\nReceived msg %d in process %d\n", msg, getpid());
break;
}
if (zmq_close(rep_socket) != 0) {
perror("zmq_close of rep_socket in child error");
}
if (zmq_ctx_term(child_context) != 0) {
perror("zmq_ctx_term of child_context error");
}
} else {
// Parent
sleep(1);
void* parent_context = zmq_ctx_new ();
if (parent_context == NULL) {
std::cerr << "\nParent ctx error\n";
}
void* req_socket = zmq_socket (parent_context, ZMQ_REQ);
if (req_socket == NULL) {
perror("zmq_socket of type req error in parent");
}
snprintf(connect_path, sizeof(connect_path), "ipc:///tmp/zmqtest%d-router", child_pid);
rc = zmq_connect(req_socket, connect_path);
assert (rc == 0);
msg = 30;
if (zmq_send (req_socket, &msg, sizeof(msg), 0) != 0) {
perror("zmq_send error in parent");
}
if (zmq_close(req_socket) != 0) {
perror("zmq_close of req_socket in parent error");
}
if (zmq_ctx_term(parent_context) != 0) {
perror("zmq_ctx_term of parent_context error");
}
}
}
第 1 步:进行简单测试:
嗯,至少,首先应该有这样的test-en-Queueing:
rc = zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT );
printf ( "INF: zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT )\nZMQ: returned rc == %d\nZMQ: zmq_errno ~ %s\n",
rc,
zmq_strerror ( zmq_errno() )
);
.
第 2 步:post 打印输出
接下来,如果有任何 "missed" 镜头,error-analysis 可能会建议可能的原因
(当且仅当 parent_ctx
确实拒绝 甚至接受来自最简单的 zmq_send()
调用它的内部排队工具 的数据这样做的明确原因).
否则我们一无所知(ZMQ_DONTWAIT
标志不是这里的原因)。
作为测试
INF: zmq_send ( req_socket, 'A_TEST_BLOCK', 12, ZMQ_DONTWAIT )
ZMQ: returned rc == 12
ZMQ: zmq_errno ~ Resource temporarily unavailable
第 3 步:
根据文档,测试已确认:
The zmq_send()
function shall return number of bytes in the message if successful.
那么,让我们更深入地挖掘一下:
int major, minor, patch;
zmq_version ( &major, &minor, &patch );
printf ( "INF: current ØMQ version is %d.%d.%d\nZMQ: zmq_errno ~ %s\n",
major, minor, patch,
zmq_strerror ( zmq_errno() )
);
第 4 步:
如果 bleeding-edge API-updates 与已发布的 API-specification 不符,记录事件:
printf ( "EXPECT( NO ERROR, ON START ): zmq_errno ~ %s\n",
zmq_strerror ( zmq_errno() )
);
printf ( "EXPECT( <major>.<minor>.<patch> ): zmq_version ~\n" );
int major, minor, patch
zmq_version ( &major, &minor, &patch );
printf ( "INF: current ØMQ version is %d.%d.%d\nZMQ: zmq_errno ~ %s\n",
major, minor, patch
)
printf ( "EXPECT( NO ERROR ): zmq_errno ~ %s\n",
zmq_strerror ( zmq_errno() )
);
printf ( "EXPECT( NO ERROR ): zmq_send() ~ %s\n" );
rc = zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT );
printf ( "INF: zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT )\nZMQ: returned rc == %d which ouhgt be == 12, is it?\n",
rc
);
printf ( "EXPECT( NO ERROR ): zmq_errno ~ %s\n",
zmq_strerror ( zmq_errno() )
);
如果出现意外结果,请随时提出问题。
我正在尝试用 C++ 创建一个 REQ <--> 路由器 <--> 经销商 <--> REP 通信。子进程绑定路由器和经销商,路由器和经销商之间的代理,将 REP 连接到经销商并等待 zmq_recv.
的消息父进程将 REQ 连接到路由器并尝试发送消息,但我收到 zmq_send error in parent: Resource temporarily unavailable
(即 EAGAIN
)。根据 zmq_send 文档,EAGAIN
表示:
Non-blocking mode was requested and the message cannot be sent at the moment.
然而,消息 确实 被发送了,因为它是在子进程中接收到的。为什么 return 那个错误号?
这里是 MCVE:
#include <zmq.h>
#include <iostream>
#include <sys/types.h>
#include <unistd.h>
#include <assert.h>
#include <thread>
#include <stdio.h>
int main() {
char connect_path[35];
int rc;
int msg;
pid_t child_pid = fork();
if (child_pid == 0) {
// Child
void* child_context = zmq_ctx_new ();
if (child_context == NULL) {
std::cerr << "\nChild context error\n";
}
void* router = zmq_socket(child_context, ZMQ_ROUTER);
if (router == NULL) {
perror("zmq_socket of type router error");
}
char bind_path[35];
snprintf(bind_path, sizeof(bind_path), "ipc:///tmp/zmqtest%d-router", getpid());
rc = zmq_bind(router, bind_path);
assert (rc == 0);
void* dealer = zmq_socket(child_context, ZMQ_DEALER);
if (dealer == NULL) {
perror("zmq_socket of type dealer error");
}
snprintf(bind_path, sizeof(bind_path), "ipc:///tmp/zmqtest%d-dealer", getpid());
rc = zmq_bind(dealer, bind_path);
assert (rc == 0);
std::thread z_proxy (zmq_proxy, router, dealer, nullptr);
z_proxy.detach();
void* rep_socket = zmq_socket (child_context, ZMQ_REP);
if (rep_socket == NULL) {
perror("zmq_socket of type rep error");
}
snprintf(connect_path, sizeof(connect_path), "ipc:///tmp/zmqtest%d-dealer", getpid());
rc = zmq_connect(rep_socket, connect_path);
assert (rc == 0);
while(1) {
if (zmq_recv (rep_socket, &msg, sizeof(msg), 0) != 0) {
perror("zmq_recv error");
}
printf("\nReceived msg %d in process %d\n", msg, getpid());
break;
}
if (zmq_close(rep_socket) != 0) {
perror("zmq_close of rep_socket in child error");
}
if (zmq_ctx_term(child_context) != 0) {
perror("zmq_ctx_term of child_context error");
}
} else {
// Parent
sleep(1);
void* parent_context = zmq_ctx_new ();
if (parent_context == NULL) {
std::cerr << "\nParent ctx error\n";
}
void* req_socket = zmq_socket (parent_context, ZMQ_REQ);
if (req_socket == NULL) {
perror("zmq_socket of type req error in parent");
}
snprintf(connect_path, sizeof(connect_path), "ipc:///tmp/zmqtest%d-router", child_pid);
rc = zmq_connect(req_socket, connect_path);
assert (rc == 0);
msg = 30;
if (zmq_send (req_socket, &msg, sizeof(msg), 0) != 0) {
perror("zmq_send error in parent");
}
if (zmq_close(req_socket) != 0) {
perror("zmq_close of req_socket in parent error");
}
if (zmq_ctx_term(parent_context) != 0) {
perror("zmq_ctx_term of parent_context error");
}
}
}
第 1 步:进行简单测试:
嗯,至少,首先应该有这样的test-en-Queueing:
rc = zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT );
printf ( "INF: zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT )\nZMQ: returned rc == %d\nZMQ: zmq_errno ~ %s\n",
rc,
zmq_strerror ( zmq_errno() )
);
.
第 2 步:post 打印输出
接下来,如果有任何 "missed" 镜头,error-analysis 可能会建议可能的原因
(当且仅当 parent_ctx
确实拒绝 甚至接受来自最简单的 zmq_send()
调用它的内部排队工具 的数据这样做的明确原因).
否则我们一无所知(ZMQ_DONTWAIT
标志不是这里的原因)。
作为测试
INF: zmq_send ( req_socket, 'A_TEST_BLOCK', 12, ZMQ_DONTWAIT )
ZMQ: returned rc == 12
ZMQ: zmq_errno ~ Resource temporarily unavailable
第 3 步:
根据文档,测试已确认:
The
zmq_send()
function shall return number of bytes in the message if successful.
那么,让我们更深入地挖掘一下:
int major, minor, patch;
zmq_version ( &major, &minor, &patch );
printf ( "INF: current ØMQ version is %d.%d.%d\nZMQ: zmq_errno ~ %s\n",
major, minor, patch,
zmq_strerror ( zmq_errno() )
);
第 4 步:
如果 bleeding-edge API-updates 与已发布的 API-specification 不符,记录事件:
printf ( "EXPECT( NO ERROR, ON START ): zmq_errno ~ %s\n",
zmq_strerror ( zmq_errno() )
);
printf ( "EXPECT( <major>.<minor>.<patch> ): zmq_version ~\n" );
int major, minor, patch
zmq_version ( &major, &minor, &patch );
printf ( "INF: current ØMQ version is %d.%d.%d\nZMQ: zmq_errno ~ %s\n",
major, minor, patch
)
printf ( "EXPECT( NO ERROR ): zmq_errno ~ %s\n",
zmq_strerror ( zmq_errno() )
);
printf ( "EXPECT( NO ERROR ): zmq_send() ~ %s\n" );
rc = zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT );
printf ( "INF: zmq_send ( req_socket, "A_TEST_BLOCK", 12, ZMQ_DONTWAIT )\nZMQ: returned rc == %d which ouhgt be == 12, is it?\n",
rc
);
printf ( "EXPECT( NO ERROR ): zmq_errno ~ %s\n",
zmq_strerror ( zmq_errno() )
);
如果出现意外结果,请随时提出问题。