Multi-core ZeroMQ?
ZeroMQ is used to accept input parameters.
def server():
rep = context.socket(zmq.REP)
rep.bind('tcp://*:{}'.format(PORT))
while True:
data = rep.recv_json()
result = calculate(data)
rep.send_json(result)
The calculation method is called calculate
, upon completion, it result
will be sent to the client via ZMQ.
Basics of my test, it currently only uses 1 core of the machine, now I wanted to use other cores. I've read several docs about multiprocessing
and multithreading
, but they mostly focus on fixed inputs, which is not my case.
So, I need some help now.
source to share
Here's what you can use multiprocessing
to have multiple worker processes that can handle concurrent clients connecting:
import zmq
from multiprocessing import Process
def calculate(data):
return {"output": data['ok']}
def handle_request(url):
context = zmq.Context()
socket = context.socket(zmq.REP)
socket.connect(url)
while True:
data = socket.recv_json()
print("received {}".format(data))
out = calculate(data)
socket.send_json(out)
def server():
context = zmq.Context()
# Set up socket for clients to connect to.
clients = context.socket(zmq.ROUTER)
clients.bind('tcp://*:{}'.format(5556))
# Set up ipc socket for workers to connect to
url_worker = 'ipc:///tmp/workers'
workers = context.socket(zmq.DEALER)
workers.bind(url_worker)
# Start 4 worker processes
for _ in range(4):
p = Process(target=handle_request, args=(url_worker,))
p.start()
# Forward requests from clients to the workers via a Queue
zmq.device(zmq.QUEUE, clients, workers)
if __name__ == "__main__":
server()
Now if you point it to the client example:
import zmq
from threading import Thread
def send_req(request):
context = zmq.Context()
print("Connecting to hello world server...")
socket = context.socket(zmq.REQ)
socket.connect("tcp://localhost:5556")
print("Sending request %s ..." % request)
socket.send_json({"ok" : "Hello"})
message = socket.recv()
print("Received reply %s [ %s ]" % (request, message))
# Do 10 requests in parallel
for request in range(10):
Thread(target=send_req, args=(request,)).start()
You get the following output:
Connecting to hello world server...
Sending request 0 ...
Connecting to hello world server...
Sending request 1 ...
Connecting to hello world server...
Sending request 2 ...
Connecting to hello world server...
Sending request 3 ...
Connecting to hello world server...
Sending request 4 ...
Connecting to hello world server...
Sending request 5 ...
Connecting to hello world server...
Sending request 6 ...
Connecting to hello world server...
Sending request 7 ...
Connecting to hello world server...
Sending request 8 ...
Connecting to hello world server...
Sending request 9 ...
<5 second delay>
Received reply 0 [ {"output":"Hello"} ]
Received reply 1 [ {"output":"Hello"} ]
Received reply 3 [ {"output":"Hello"} ]
Received reply 2 [ {"output":"Hello"} ]
<5 second delay>
Received reply 4 [ {"output":"Hello"} ]
Received reply 5 [ {"output":"Hello"} ]
Received reply 6 [ {"output":"Hello"} ]
Received reply 7 [ {"output":"Hello"} ]
< 5 second delay>
Received reply 8 [ {"output":"Hello"} ]
Received reply 9 [ {"output":"Hello"} ]
source to share