Multiprocessing for stream processing

[learnermaxRL]

Any ideas how to use the stream data colected through batch to be utilised with Multiprocessing queue fpr further processing since its a blocking call?

[ask]

Hmm, in Celery we have an async version of the multiprocessing pool. One of the things that I want to do is to port that to asyncio.

multiprocessing.Pool combines threads and fork, this can lead to deadlocks and other issues.
It also uses a POSIX semaphore to protect reads/writes from a shared pipe, and if a child process
is abruptly terminated while the lock is held the pool will deadlock.

I imagine you can already use multiprocessing.Pool with faust, as you can handle the blocking
call in a thread. You could also start a local cluster of Faust workers :) granted this will add some overhead as messages will go through Kafka.

[learnermaxRL]

I was trying to push the data to multiprocessing queue to be processed by class inheriting mp.Process,it works with simple Queue,but with mp.Queue there is deadlock possibly because of the reason you mentioned.
I think you can add this functionality to faust where in kafka stream data can be dumped to mp.queue which can be utilized by other processes while non-blocking the asyncio loop.

[tarasmatsyk]

@learnermaxRL, I ended up using ThreadPool as message processing takes longer than required to speed up things.

CONCURRENCY = cpu_count()
thread_pool = ThreadPoolExecutor(max_workers=CONCURRENCY)

# then you do
@app.agent(new_topic, concurrency=CONCURRENCY)
async def new_message(stream):
    async for msg in stream:
            await app.loop.run_in_executor(thread_pool, method, args)