How to use concurrency as well as multiprocessing to handle the data received through websocket in python?

Question

CODE:

url = 'ws://xx.xx.xx.xx:1234'
ws = create_connection(url)
ws.send(json.dumps(subscribe_msg))
ws.recv()
while True:
    result = ws.recv()
    # handle the result using a different core each time
    handle_parallely(result)

The while loop result=ws.recv() needs to be concurrent, so that ws.recv can be repeatedly called without waiting for handle_parallely to return. handle_parallely needs to run parallely when it is called.

The data received and its processing is independent of any previous or future data.

Answer 1

You can use a ProcessPoolExecutor from the concurrent futures module. This could look like

from concurrent.futures import ProcessPoolExecutor
max_number_of_processes = 4 # just put any number here
futures = []
with ProcessPoolExecutor(max_worker=max_number_of_processes) as executor:
    while True:
        result = ws.recv()
        # handle the result using a different core each time
        future = executor.submit(handle_parallely, result)
        futures.append(future) 
        futures = [f for f in futures if not f.done()]

This of course only works if result and handle_parallely are pickable, see this for which types are pickable by default if you run into issues with PickelingError.

Storing the futures in that list is optional, but maybe you want to keep track of references to them.