I am trying to create a server with fastapi with multiprocessing, where you can send an image to a given endpoint, and it will be processed concurrently via 3 different ocr models to recognize text. Then when all 3 models will end, the endpoint will response with all 3 versions of recognized text.

I don’t want to create new processes every time because it takes very long time to load the ocr models into memory.
So I want to initialize them and start sending them images via a different queue for all three model and receive text from shared dict.

Here you can see an example with requests library instead of OCR models for simplicity of what I tried with comments, but it wont work. Can you help me please? 🙂

import asyncio
import multiprocessing
from multiprocessing import Manager
from multiprocessing.managers import DictProxy  # noqa
from multiprocessing.queues import Queue

manager = Manager()

url_queue = manager.Queue()
url_shared_dict = manager.dict()


def tess_process(url_queue: Queue, text_shared_dict: DictProxy):
    # Some heavy initialization
    import requests

    # starts waiting for tasks
    while True:
        if url_queue.empty():
            continue
        url, future = url_queue.get()
        
        result = requests.get(url)

        # adding result to a shared dict, to get it in main process afterward
        text_shared_dict[url] = result
        
        # Here is the main problem
        # I need to somehow notify a main process that a job is done 
        # and main process can take it.
        # While subprocess is doing its job, 
        # the main process need to be able to await
        future.set_result('Task completed')


async def async_func():
    loop = asyncio.get_event_loop()
    future = loop.create_future()
    
    url = 'https://google.com'
    url_queue.put((url, future))
    
    await future
    
    print(url_shared_dict[url])

if __name__ == '__main__':
    p = multiprocessing.Process(
        target=tess_process, args=(url_queue, url_shared_dict)
    )
    p.start()
    asyncio.run(async_func())

Its getting stack, infinite loop maybe
Now I know, that async loop wont work between processes.
But maybe, there any other way way to await for result of subprocesses in shared dict,

I know i can do it with asyncio.sleep(some_time) but its a bit hacky