question about the rep-router-rep #540
Description
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
#include
#include
#include <pthread.h>
#include
#include <stdlib.h> // 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::vectorstd::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;
}`