c++ - 在清理过程中，zmq卡在zmq_proxy()中

我在Ubuntu机器(4.4.0-28-通用＃47〜14.04.1-Ubuntu)上使用4.2.4版本的ZeroMQ。

我从https://github.com/booksbyus/zguide/blob/master/examples/C/mtserver.c提取了 mtserver.c 。
模式是REQ-ROUTER-DEALER-REP。下面是相同代码的略微修改版本。

问题-当我尝试在所有工作线程完成后干净退出时，代码卡在zmq_proxy( clients, workers, NULL )中。

我尝试在“客户端”套接字上使用LINGER计时器选项，但没有帮助。感谢任何帮助我进行调试的帮助。谢谢。

(gdb) bt
#0  0x00007fd6a4febc9d in poll () at ../sysdeps/unix/syscall-template.S:81
#1  0x00007fd6a58232fa in poll (__timeout=-1, __nfds=1, __fds=0x7ffd692d80e0) at /usr/include/x86_64-linux-gnu/bits/poll2.h:46
#2  zmq::signaler_t::wait (this=this@entry=0x1f09e28, timeout_=timeout_@entry=-1) at src/signaler.cpp:232
#3  0x00007fd6a5809f95 in zmq::mailbox_t::recv (this=0x1f09dc0, cmd_=0x7ffd692d8140, timeout_=-1) at src/mailbox.cpp:81
#4  0x00007fd6a582497d in zmq::socket_base_t::process_commands (this=this@entry=0x1f09850, timeout_=timeout_@entry=-1,
throttle_=throttle_@entry=false) at src/socket_base.cpp:1341
#5  0x00007fd6a5824ee3 in zmq::socket_base_t::send (this=this@entry=0x1f09850, msg_=msg_@entry=0x7ffd692d83d0, flags_=<optimized out>)
at src/socket_base.cpp:1156
#6  0x00007fd6a5819d24 in forward (from_=from_@entry=0x1f077a0, from_stats=from_stats@entry=0x7ffd692d8330, to_=to_@entry=0x1f09850,
to_stats=to_stats@entry=0x7ffd692d8350, capture_=capture_@entry=0x0, msg_=...) at src/proxy.cpp:147
#7  0x00007fd6a581a7d3 in zmq::proxy (frontend_=0x1f077a0, backend_=0x1f09850, capture_=0x0, control_=0x0) at src/proxy.cpp:462
#8  0x00000000004018b2 in zmqMTServer::Start (this=this@entry=0x7ffd692d84e0) at mtserver.cpp:75
#9  0x0000000000401207 in main () at mtserver.cpp:89

(gdb

#include <thread>
#include <iostream>
#include <zmq.h>
#include <assert.h>
#include <string.h>

using namespace std;
bool zmqServerShutdown = false;

class zmqMTServer {
private:
    ushort workerThreads;
void* zmqContext;
unique_ptr<thread[]> workerThreadIDs;
void workerRoutine(void);

public:
    zmqMTServer(ushort threads) : workerThreads(threads)
    {
        zmqContext = zmq_ctx_new();
        assert(zmqContext);
        auto rc = zmq_ctx_set(zmqContext, ZMQ_IO_THREADS, 1);
        assert(rc == 0);
        rc = zmq_ctx_set(zmqContext, ZMQ_MAX_SOCKETS, ZMQ_MAX_SOCKETS_DFLT);
        assert(rc == 0);
    }
    int Start(void);
    ~zmqMTServer(void)
    {
        for (auto i = 0; i < workerThreads; i++)
            workerThreadIDs[i].join();
        zmq_ctx_destroy(zmqContext);
    }
};

void zmqMTServer::workerRoutine(void)
{
    void *receiver = zmq_socket(zmqContext, ZMQ_REP);
    assert(receiver);
    auto rc = zmq_connect(receiver, "inproc://workerThreads");
    assert(rc == 0);
    cout << "Worker thread(" << this_thread::get_id() << ") started \n";
    while (!zmqServerShutdown) {
        char buf[256];
        auto size = zmq_recv(receiver, buf, 255, 0);
        if (size == -1) {
            cout << "workerRoutine(): zmq_recv size = " << size << "\n";
            continue; // some thing went wrong
        }
        buf[size] = '\0';
        cout << "Worker thread(" << this_thread::get_id() << ") Received request: " << buf << "\n";

        size = zmq_send(receiver, buf, strlen(buf), 0);
    }
    cout << "zmqMTServer worker thread exiting " << this_thread::get_id() << "\n";
    rc = zmq_close(receiver);
}

int zmqMTServer::Start(void)
{
    void *clients = zmq_socket(zmqContext, ZMQ_ROUTER);
    assert(clients);

    auto rc = zmq_bind(clients, "tcp://*:10051");
    assert(rc == 0);
    void *workers = zmq_socket(zmqContext, ZMQ_DEALER);
    rc = zmq_bind (workers, "inproc://workerThreads");
    assert(rc == 0);

    //  Launch pool of worker threads
    workerThreadIDs = unique_ptr<thread[]>(new thread[workerThreads]);
    for(auto i = 0; i < workerThreads; i++) {
        workerThreadIDs[i] = thread(&zmqMTServer::workerRoutine, this);
    }

    zmq_proxy(clients, workers, NULL);

    zmq_close(clients);
    zmq_close(workers);
    return 0;
}

int main(void)
{
    int major, minor, patch;
    zmq_version(&major, &minor, &patch);
    cout << "ZMQ version : " << major << "." << minor << "." << patch << "\n";

    zmqMTServer server(3);
    server.Start();
}

最佳答案

[+1]了解有关使用ZMQ_LINGER的分析思路

哦，那些便宜又差劲的例子，相对于现实世界分布式系统

快速浏览一下，从教科书示例中借来的代码由于使用 zmq_recv(...) 操作的阻止模式而遭受的困扰。

原则上以非阻塞方式更好地设计分布式系统。这样，您就不会失去控制。是的，大多数示例和教科书代码片段仍会出现阻塞读取的情况，但这绝对不能出现在严肃的生产代码中。

为什么？因为如果没有别的，阻塞状态会将您从游戏中移出，您可能只是站着祈祷外部事件的发生。如果这不会发生，那么您宝贵的代码将一直风靡一时...

#define ZMQ_DONTWAIT 1
...
#define ZMQ_NOBLOCK ZMQ_DONTWAIT

因此，重新设计一个消息等待部分，以便使用zmq_poll()，它配备了一些相当短的(或连续零)超时，并在每个POSACKed消息到达之后进行非阻塞读取:

zmq_recv( ...., ZMQ_NOBLOCK );

这样，您的代码将永远不会在任何阻塞的context-instance(s)线程内等待。

distributed-systems根本无法挂断电话。

关于c++ - 在清理过程中，zmq卡在zmq_proxy()中，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/48899046/