当用std::thread创建worker时，ZeroMQ崩溃。

Question

我正在移植 lbbroker.c 示例 从ZeroMQ指南书到更高层次的API。czmq.

出于某种原因，a std::thread 创建并连接到一个端点，并使用 zsock 总是崩溃。下面是用C++编写的负载平衡代理的代码。

//  Load-balancing broker
//  Demonstrates use of the CZMQ API

#include <czmq.h>
#include <cstdio>
#include <thread>
#include <iostream>
#include <vector>

#define NBR_CLIENTS 10
#define NBR_WORKERS 3
#define WORKER_READY   "READY"      //  Signals worker is ready

//  Basic request-reply client using REQ socket
//
void client_task()
{
    std::cout << "Start client...\n";
    zsock_t *client = zsock_new_req("tcp://*:5672");
    // zsock_t *client = zsock_new_req("ipc://frontend.ipc");
    std::cout << "Client connect...\n";
    // zsock_connect(client, "ipc://frontend.ipc");

    //  Send request, get reply
    std::cout << "Client send HELLO\n";
    zstr_send(client, "HELLO");
    char *reply = zstr_recv(client);
    if (reply) {
        printf("Client: %s\n", reply);
        free(reply);
    }
}

//  Worker using REQ socket to do load-balancing
//
void worker_task()
{
    std::cout << "Start work task...\n";
    zsock_t *worker = zsock_new(ZMQ_REQ);
#if (defined (WIN32))
    // worker = zsock_new_req("tcp://localhost:5673"); // backend
#else
    // worker = zsock_new_req("ipc://backend.ipc");
#endif
    std::cout << "Worker connect\n";
    zsock_connect(worker, "tcp::/localhost:5673");
    // zsock_connect(worker, "ipc://backend.ipc");
    //  Tell broker we're ready for work
    zframe_t *frame = zframe_new(WORKER_READY, strlen(WORKER_READY));
    zframe_send(&frame, worker, 0);

    //  Process messages as they arrive
    while (true) {
        zmsg_t *msg = zmsg_recv(worker);
        if (!msg)
            break;              //  Interrupted
        zframe_print(zmsg_last(msg), "Worker: ");
        zframe_reset(zmsg_last(msg), "OK", 2);
        zmsg_send(&msg, worker);
    }
}

//  .split main task
//  Now we come to the main task. This has the identical functionality to
//  the previous {{lbbroker}} broker example, but uses CZMQ to start child 
//  threads, to hold the list of workers, and to read and send messages:

int main(void)
{
    zsock_t *frontend = zsock_new_router("tcp://*:5672");
    zsock_t *backend = zsock_new_router ("tcp://*:5673");
    // zsock_t *frontend = zsock_new_router("ipc://frontend.ipc");
    // zsock_t *backend = zsock_new_router ("ipc://backend.ipc");

    // printf("Create server threads");
    for (int worker_nbr = 0; worker_nbr < NBR_WORKERS; worker_nbr++) {
        std::thread t(worker_task);
        std::cout << "Created worker...\n";
    //     pthread_t worker;
    //     pthread_create(&worker, NULL, worker_task, (void *)(intptr_t)worker_nbr);
    }
    int client_nbr;

    std::cout << "Create client threads\n";
    for (client_nbr = 0; client_nbr < NBR_CLIENTS; client_nbr++) {
        std::thread t(client_task);
    //     pthread_t client;
    //     printf("Create client thread %d", client_nbr);
    //     pthread_create(&client, NULL, client_task, (void *)(intptr_t)client_nbr);
    }

    //  Queue of available workers
    zlist_t *workers = zlist_new();

    //  .split main load-balancer loop
    //  Here is the main loop for the load balancer. It works the same way
    //  as the previous example, but is a lot shorter because CZMQ gives
    //  us an API that does more with fewer calls:
    while (true) {
        zmq_pollitem_t items[] = {
                                  { backend, 0, ZMQ_POLLIN, 0 },
                                  { frontend, 0, ZMQ_POLLIN, 0 }
        };
        std::cout << "Running\n";
        //  Poll frontend only if we have available workers
        int rc = zmq_poll(items, zlist_size(workers) ? 2 : 1, -1);
        if (rc == -1)
            break;              //  Interrupted

        //  Handle worker activity on backend
        if (items[0].revents & ZMQ_POLLIN) {
            //  Use worker identity for load-balancing
            zmsg_t *msg = zmsg_recv(backend);
            if (!msg)
                break;          //  Interrupted

#if 0
            // zmsg_unwrap is DEPRECATED as over-engineered, poor style
            zframe_t *identity = zmsg_unwrap(msg);
#else
            zframe_t *identity = zmsg_pop(msg);
            zframe_t *delimiter = zmsg_pop(msg);
            zframe_destroy(&delimiter); 
#endif

            zlist_append(workers, identity);

            //  Forward message to client if it's not a READY
            zframe_t *frame = zmsg_first(msg);
            if (memcmp(zframe_data(frame), WORKER_READY, strlen(WORKER_READY)) == 0) {
                zmsg_destroy(&msg);
            } else {
                zmsg_send(&msg, frontend);
                if (--client_nbr == 0)
                    break; // Exit after N messages
            }
        }
        if (items[1].revents & ZMQ_POLLIN) {
            //  Get client request, route to first available worker
            zmsg_t *msg = zmsg_recv(frontend);
            if (msg) {
#if 0
                // zmsg_wrap is DEPRECATED as unsafe
                zmsg_wrap(msg, (zframe_t *)zlist_pop(workers));
#else
                zmsg_pushmem(msg, NULL, 0); // delimiter
                zmsg_push(msg, (zframe_t *)zlist_pop(workers));
#endif

                zmsg_send(&msg, backend);
            }
        }
    }
    //  When we're done, clean up properly
    while (zlist_size(workers)) {
        zframe_t *frame = (zframe_t *)zlist_pop(workers);
        zframe_destroy(&frame);
    }
    zlist_destroy(&workers);
    zsock_destroy(&frontend);
    zsock_destroy(&backend);
    return 0;
}

你可以用 g++ lbbroker.cpp -o lbbroker -lzmq -lpthread. 它成功地编译了，然而，当运行时，我得到了一个异常和一个核心转储。

Created worker...
terminate called without an active exception
Start work task...
aborted (core dumped)

崩溃的原因是... worker_task 函数。根据GDB，好像有1个线程成功创建，然后程序就崩溃了。C版本工作正常。

Answer 1

C++有destructor，它在声明变量的作用域结束时被调用。

在 std::thread如果在可连接的线程中调用了destructor，那么就可以使用该线程。std::terminate 将被调用。你应该决定是否要 join 线 detach 在你到达destructor之前，它就已经存在了。

所以，要么你分离线程（因此它们会自由运行，没有一个简单的方法来等待它们完成），要么你增加对象的寿命，并在以后加入它们。