ZzzzHeJ commented 2 years ago

I got a demo about Load Balancing Broker which use zhelpers, but I rebuild it with cppzmq. this demo can be runing and the broker can receive the msg from worker and client. And the question is that the worker and client can not receive the msg from broker. I guess the the problem is about the router id.

`#include "zmq.hpp"

include

include // random() RAND_MAX

define within(num) (int) ((float) (num) / (RAND_MAX + 1.0))

static std::atomic i; inline std::string s_set_id(std::string type,zmq::socket_t & socket) { std::stringstream ss; ss << std::hex << std::uppercase << std::setw(4) << std::setfill('0') << std::to_string(i++) << within(0x10000) << "-" << std::setw(4) << std::setfill('0') << std::to_string(i) << within(0x10000); socket.set(zmq::sockopt::routing_id, type + std::to_string(i) + "|" + ss.str()); return ss.str(); } // Basic request-reply client using REQ socket // static void client_thread(void arg) { zmq::context_t context(1); zmq::socket_t client(context, ZMQ_REQ);

s_set_id("client",client); // Set a printable identity
client.connect("ipc://frontend.ipc");
//  Send request, get reply
client.send(zmq::buffer("HELLO"));
std::cout << "client send hello" << "\n";
zmq::message_t msg;
client.recv(msg);
std::string reply = msg.to_string();
std::cout << "Client: " << reply << std::endl;
return (NULL);

}

// Worker using REQ socket to do LRU routing // static void worker_thread(void arg) { zmq::context_t context(1); zmq::socket_t worker(context, ZMQ_REQ);

if (defined (WIN32))

s_set_id(worker, (intptr_t)arg);
worker.connect("tcp://localhost:5673"); // backend

else

s_set_id("worker",worker);
worker.connect("ipc://backend.ipc");

endif

while (1) {
    //  Tell backend we're ready for work
    worker.send(zmq::buffer("READY"));
    //  Read and save all frames until we get an empty frame
    //  In this example there is only 1 but it could be more
    zmq::message_t msg;
    std::cout << "work waite" << "\n";
    worker.recv(msg);
    std::cout << "work recv" << "\n";
    std::string address = msg.to_string();
    {
        worker.recv(msg);
        std::string empty = msg.to_string();
        assert(empty.size() == 0);
    }

    //  Get request, send reply
    worker.recv(msg);
    std::string request = msg.to_string();
    std::cout << "Worker: " << request << std::endl;

    worker.send(zmq::buffer(address), zmq::send_flags::sndmore);
    worker.send(zmq::buffer(""), zmq::send_flags::sndmore);
    worker.send(zmq::buffer("OK"));
}
return (NULL);

}

int main(int argc, char argv[]) { // Prepare our context and sockets zmq::context_t context(1); zmq::socket_t frontend(context, ZMQ_ROUTER); zmq::socket_t backend(context, ZMQ_ROUTER); std::vector<std::thread>threadList;

frontend.bind("ipc://frontend.ipc");
backend.bind("ipc://backend.ipc");
int client_nbr;
for (client_nbr = 0; client_nbr < 10; client_nbr++) {
    pthread_t client;
    pthread_create(&client, NULL, client_thread, (void*)(intptr_t)client_nbr);
}
int worker_nbr;
for (worker_nbr = 0; worker_nbr < 3; worker_nbr++) {
    pthread_t worker;
    pthread_create(&worker, NULL, worker_thread, (void*)(intptr_t)worker_nbr);
}

zmq::poller_t<> in_poller, out_poller;
// Our application has two input sockets and one output.
in_poller.add(frontend, zmq::event_flags::pollin);
in_poller.add(backend, zmq::event_flags::pollin);
//out_poller.add(frontend, zmq::event_flags::pollout);
//out_poller.add(backend, zmq::event_flags::pollout);

//std::vector<zmq::poller_event<>> in_events(2);
//std::vector<zmq::poller_event<>> out_events(2);
std::queue<std::string> worker_queue;
zmq::message_t msg;
const std::chrono::milliseconds timeout{ 100 };
while (1) 
{
    zmq::pollitem_t items[] = { { backend, 0, ZMQ_POLLIN, 0 },
                                { frontend, 0, ZMQ_POLLIN, 0 }, };
    if (worker_queue.size())
        zmq::poll(&items[0], 2);
    else
        zmq::poll(&items[0], 1);
    if (items[0].revents & ZMQ_POLLIN) 
    {
        backend.recv(msg);
        std::string workAddr = msg.to_string();
        {
            //  Second frame is empty
            backend.recv(msg);
            std::string empty = msg.to_string();
            assert(empty.size() == 0);
        }
        backend.recv(msg);
        std::string client_addr = msg.to_string();
        if (client_addr.compare("READY") == 1)
        {
            std::cout << workAddr << "  say  " << client_addr << "\n";
            worker_queue.push(workAddr);
        }
        else
        {
            backend.recv(msg);
            std::string reply = msg.to_string();
            std::cout << " work rely to " << client_addr << reply << "\n";
            frontend.send(zmq::buffer(client_addr), zmq::send_flags::sndmore);
            frontend.send(zmq::buffer(""), zmq::send_flags::sndmore);
            frontend.send(zmq::buffer(reply));
        }

        if (--client_nbr == 0)
            break;
    }
    if (items[1].revents & ZMQ_POLLIN)
    {

        //  Now get next client request, route to LRU worker
        //  Client request is [address][empty][request]
        frontend.recv(msg);
        std::string client_addr = msg.to_string();
        {
            frontend.recv(msg);
            std::string empty = msg.to_string();
            assert(empty.size() == 0);
        }
        frontend.recv(msg);
        std::string request = msg.to_string();
        std::cout << client_addr << "  say  " << request << "\n";
        std::string worker_addr = worker_queue.front();//worker_queue [0];
        worker_queue.pop();
        backend.send(zmq::buffer(worker_addr), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(""), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(client_addr), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(""), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(request));
    }
}
return 0;

}`

ZzzzHeJ commented 2 years ago

broker.zip demo file

ZzzzHeJ commented 2 years ago

I think I was fund the bug. send(zmq::buffer("")) and send(zmq::messge_t("")) is different. former will receive a "/0" ,but latter will receive "" maybe cause the router fail?

ZzzzHeJ commented 2 years ago

when i change any send(zmq::buffer(client_addr), zmq::send_flags::sndmore); to backend.send(zmq::buffer(std::string("")), zmq::send_flags::sndmore); the demo is work

ZzzzHeJ commented 2 years ago

this is ok `// Least - recently used(LRU) queue device // Clients and workers are shown here in-process //

include "zmq.hpp"

include

include // random() RAND_MAX

define within(num) (int) ((float) (num) / (RAND_MAX + 1.0))

static std::atomic i; inline std::string s_set_id(std::string type, zmq::socket_t& socket) { std::stringstream ss; ss << std::hex << std::uppercase << std::setw(4) << std::setfill('0') << std::to_string(i++) << within(0x10000) << "-" << std::setw(4) << std::setfill('0') << std::to_string(i) << within(0x10000); socket.set(zmq::sockopt::routing_id, type + std::to_string(i) + "|" + ss.str()); return ss.str(); } // Basic request-reply client using REQ socket // static void client_thread(void arg) { zmq::context_t context(1); zmq::socket_t client(context, ZMQ_REQ);

s_set_id("client", client); // Set a printable identity
client.connect("ipc://frontend.ipc");
//  Send request, get reply
client.send(zmq::buffer("HELLO"));
std::cout << "client send hello" << "\n";
zmq::message_t msg;
client.recv(msg);
std::string reply = msg.to_string();
std::cout << "Client: " << reply << std::endl;
return (NULL);

}

// Worker using REQ socket to do LRU routing // static void worker_thread(void arg) { zmq::context_t context(1); zmq::socket_t worker(context, ZMQ_REQ);

if (defined (WIN32))

s_set_id(worker, (intptr_t)arg);
worker.connect("tcp://localhost:5673"); // backend

else

s_set_id("worker", worker);
worker.connect("ipc://backend.ipc");

endif

worker.send(zmq::buffer(std::string("READY")));
while (1) {
    //  Tell backend we're ready for work
    //  Read and save all frames until we get an empty frame
    //  In this example there is only 1 but it could be more
    zmq::message_t msg;
    std::cout << "work waite" << "\n";
    worker.recv(msg);
    std::cout << "work recv" << "\n";
    std::string address = msg.to_string();
    {
        worker.recv(msg);
        std::string empty = msg.to_string();
        assert(empty.size() == 0);
    }

    //  Get request, send reply
    worker.recv(msg);
    std::string request = msg.to_string();
    std::cout << "Worker: " << request << std::endl;

    worker.send(zmq::buffer(address), zmq::send_flags::sndmore);
    worker.send(zmq::buffer(std::string("")), zmq::send_flags::sndmore);
    worker.send(zmq::buffer("OK"));
}
return (NULL);

}

int main(int argc, char argv[]) { // Prepare our context and sockets zmq::context_t context(1); zmq::socket_t frontend(context, ZMQ_ROUTER); zmq::socket_t backend(context, ZMQ_ROUTER); std::vector<std::thread>threadList;

frontend.bind("ipc://frontend.ipc");
backend.bind("ipc://backend.ipc");
int client_nbr;
for (client_nbr = 0; client_nbr < 10; client_nbr++) {
    pthread_t client;
    pthread_create(&client, NULL, client_thread, (void*)(intptr_t)client_nbr);
}
int worker_nbr;
for (worker_nbr = 0; worker_nbr < 3; worker_nbr++) {
    pthread_t worker;
    pthread_create(&worker, NULL, worker_thread, (void*)(intptr_t)worker_nbr);
}

zmq::poller_t<> in_poller, out_poller;
// Our application has two input sockets and one output.
in_poller.add(frontend, zmq::event_flags::pollin);
in_poller.add(backend, zmq::event_flags::pollin);
//out_poller.add(frontend, zmq::event_flags::pollout);
//out_poller.add(backend, zmq::event_flags::pollout);

//std::vector<zmq::poller_event<>> in_events(2);
//std::vector<zmq::poller_event<>> out_events(2);
std::queue<std::string> worker_queue;
zmq::message_t msg;
const std::chrono::milliseconds timeout{ 100 };
while (1)
{
    zmq::pollitem_t items[] = { { backend, 0, ZMQ_POLLIN, 0 },
                                { frontend, 0, ZMQ_POLLIN, 0 }, };
    if (worker_queue.size())
        zmq::poll(&items[0], 2);
    else
        zmq::poll(&items[0], 1);
    if (items[0].revents & ZMQ_POLLIN)
    {
        backend.recv(msg);
        std::string workAddr = msg.to_string();
        worker_queue.push(workAddr);
        {
            //  Second frame is empty
            backend.recv(msg);
            std::string empty = msg.to_string();
            assert(empty.size() == 0);
        }
        backend.recv(msg);
        std::string client_addr = msg.to_string();
        if (client_addr.compare("READY") == 0)
        {
            std::cout << workAddr << "  say  " << client_addr << "\n";
        }
        else
        {
            backend.recv(msg);
            std::string empty = msg.to_string();
            backend.recv(msg);
            std::string reply = msg.to_string();
            std::cout << " work rely to " << client_addr << reply << "\n";
            frontend.send(zmq::buffer(client_addr), zmq::send_flags::sndmore);
            frontend.send(zmq::buffer(std::string("")), zmq::send_flags::sndmore);
            frontend.send(zmq::buffer(reply));
        }

        if (--client_nbr == 0)
            break;
    }
    if (items[1].revents & ZMQ_POLLIN)
    {

        //  Now get next client request, route to LRU worker
        //  Client request is [address][empty][request]
        frontend.recv(msg);
        std::string client_addr = msg.to_string();
        {
            frontend.recv(msg);
            std::string empty = msg.to_string();
            assert(empty.size() == 0);
        }
        frontend.recv(msg);
        std::string request = msg.to_string();
        std::cout << client_addr << "  say  " << request << "\n";
        std::string worker_addr = worker_queue.front();//worker_queue [0];
        worker_queue.pop();
        backend.send(zmq::buffer(worker_addr), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(std::string("")), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(client_addr), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(std::string("")), zmq::send_flags::sndmore);
        backend.send(zmq::buffer(request));
    }
}
return 0;

}`

ZzzzHeJ commented 2 years ago

but i'm still confused, way the str or c_str will cause the worker can not receive msg? Can someone tell me why

zeromq / cppzmq

question about the rep-router-rep #540

include

include

include

include

include

include

include // random() RAND_MAX

define within(num) (int) ((float) (num) / (RAND_MAX + 1.0))

if (defined (WIN32))

else

endif

include "zmq.hpp"

include

include

include

include

include

include

include // random() RAND_MAX

define within(num) (int) ((float) (num) / (RAND_MAX + 1.0))

if (defined (WIN32))

else

endif