CODEWITHSUNDEEP
c++zeromq
sakonque
sakonque

Reputation: 35

zeromq proxy delivered only one reply for each request

I am now developing a messaging infrastructure with ZMQ 4.3.1 C++ binding (cppzmq). My current architecture is like this:

    N Clients (Dealer) <------> [(Router) proxy (Dealer)] <------> N Workers (Dealer)

The worker will process the request and send multiple replies back.

The problem is that my client can only receive one reply. Client thread is blocking at zmq::poll() after receiving the first reply. I have tested one reply for each request works perfect, but when the worker sends back multiple replies, the replies "disappear" except the first one.

Proxy code:

_frontend = zmq::socket_t(_ctx, zmq::socket_type::router);
_backend = zmq::socket_t(_ctx, zmq::socket_type::dealer);
_controller.setsockopt(ZMQ_SUBSCRIBE, "", 0);

_frontend.setsockopt(ZMQ_SNDHWM, 0);
_frontend.setsockopt(ZMQ_RCVHWM, 0);

_backend.setsockopt(ZMQ_SNDHWM, 0);
_backend.setsockopt(ZMQ_RCVHWM, 0);

_frontend.bind("tcp://*:7777");
_backend.bind("tcp://*:7778");

zmq::proxy(_frontend, _backend);

Client code:

void client(int n) {
    zmq::context_t ctx(1);
    zmq::socket_t client(ctx, zmq::socket_type::dealer);
    client.setsockopt(ZMQ_RCVHWM, 0);

    std::string id = "Client-" + std::to_string(n);
    client.setsockopt(ZMQ_IDENTITY, id.c_str(), id.size());

    client.connect("tcp://localhost:7777");

    DataRequest request{utc_now(), utc_now(), "CL", {Timeframe::Five_Minute, Timeframe::Hourly, Timeframe::Daily, Timeframe::Weekly}};
    std::string req_str = request.serialize();

    zmq::message_t msg(req_str.c_str(), req_str.size());
    client.send(msg, zmq::send_flags::none);

    std::vector<zmq::pollitem_t> items = {
        {client, 0, ZMQ_POLLIN, 0}
    };
    while(true) {
        zmq::poll(items, -1);

        if(items[0].revents & ZMQ_POLLIN) {
            zmq::message_t rep;
            client.recv(rep, zmq::recv_flags::none);

            // process the message
        }
    }
}

Server code:

void server(int n) {
    // load data into memory
    std::vector<std::string> data;
    {
        // generate random message and push into data 
    }

    zmq::context_t ctx(1);
    zmq::socket_t server(ctx, zmq::socket_type::dealer);
    server.setsockopt(ZMQ_SNDHWM, 0);

    std::string id = "Server-" + std::to_string(n);
    server.setsockopt(ZMQ_IDENTITY, id.c_str(), id.size()); 

    server.connect("tcp://localhost:7778");

    std::vector<zmq::pollitem_t> items = {
        {server, 0, ZMQ_POLLIN, 0}
    };
    while(is_server_running) {
        zmq::poll(items, POLLER_TIMEOUT);

        if(items[0].revents & ZMQ_POLLIN) {
            zmq::message_t client_id, req;
            server.recv(client_id, zmq::recv_flags::none);
            server.recv(req, zmq::recv_flags::none);

            std::string client_addr = std::string(static_cast<char*>(client_id.data()), client_id.size()),
                        json_req = std::string(static_cast<char*>(req.data()), req.size());

            for(const auto& d: data) {
                zmq::message_t rep(d.c_str(), d.size());

                server.send(client_id, zmq::send_flags::sndmore);
                server.send(rep, zmq::send_flags::none);
                //std::this_thread::sleep_for(std::chrono::milliseconds(100));
            }
        }
    }
}

Upvotes: 1

Views: 157

Answers (1)

sakonque
sakonque

Reputation: 35

Look at here and you see a fool ...

Finally I worked out this problem. The client_id is moved into ZeroMQ I/O queue after being sent so it is an empty message object for the following replies. I double-checked ZeroMQ official guide and found this is referred as "Zero-Copy" feature.

For anyone encountered similar problem, please be reminded to check the IDENTITY at first.

Upvotes: 1

Related Questions