How to create a multi-threaded ROUTER server in ZMQ?

54 Views Asked by At

I have a number of DEALER clients and a ROUTER server that communicate asynchronously.

Currently, the server runs a loop that checks for incoming requests, dispatches them to a thread pool of workers, and when results are computed, sends them back to the corresponding clients.

  [client 1] <-->          <--> [worker 1]
  [client 2] <--> [server] <--> [worker 2]
  [client 3] <-->          <--> [worker 3]
  ...                           ...

However, sending results to a batch of clients is quite slow (~40 ms) and slows down the server loop, preventing it from processing results from other clients in the meantime.

To remove the sending bottleneck, I would like for each worker to send its own results back to the clients, either directly or via the server socket (but at least not going through Python logic for it).

def client():
  socket = zmq.Context.instance().socket(zmq.DEALER)
  socket.connect('tcp://localhost:4444')
  while True:
    socket.send_multipart([b'', b'request'])
    while True:
      try:
        _, response = socket.recv_multipart(zmq.NOBLOCK)
        break
      except zmq.Again:
        continue

def server():
  pool = concurrent.futures.TreadPoolExecutor(8)
  promises = collections.deque()
  socket = zmq.Context.instance().socket(zmq.ROUTER)
  socket.bind('tcp://*:4444')
  while True:
    try:
      addr, empty, payload = socket.recv_multipart(zmq.NOBLOCK)
      promises.append(pool.submit(worker, addr, payload))
    except zmq.Again:
      time.sleep(0.0001)
    # TODO: Sending in the server loop is too slow.
    # if promises and promises[0].done():
    #   addr, result = promises.popleft().result()
    #   socket.send_multipart([addr, b'', result])

def worker(addr, payload):
  # Do work...
  # return addr, result
  # TODO: Instead of returning the result back to the server loop,
  # the worker should send the result itself through a socket. 

What socket type do I need for the worker and how can I set it up to send to the DEALER clients or forward its messages to the ROUTER server?

0

There are 0 best solutions below