Does calling thread.join() blocks the event loop in an asynchronous context?

user_6501050 0
Viewed 1924

I'm implementing a web API using aiohttp, deployed using gunicorn with UVloop enabled --worker-class aiohttp.GunicornUVLoopWebWorker. Therefore, my code always runs in an asynchronous context. I had the ideia of implementing parallel jobs in the handling of requests for better performance.

I'm not using asyncio because i want Parallelism, not Concurrency.

I'm aware of multiprocessing and the GIL problem in python. But joining a process also applies to my question.

Here is an example:

from aiohttp.web import middleware

@middleware
async def context_init(request, handler):
    request.context = {}
    request.context['threads'] = []
    ret = await handler(request)
    for thread in request.context['threads']:
        thread.join()
    return ret

Taking into account that thread.join() or process.join() blocks the current thread, this will block the event loop (As far as my knowledge goes). How can I join asynchronously? What I want can be represented figuratively as this: await thread.join() or await process.join().

Update:

Thanks to @user4815162342 I was able to write proper code for my project:

Middleware:

from aiohttp.web import middleware
from util.process_session import ProcessSession

@middleware
async def context_init(request, handler):
    request.context = {}
    request.context['process_session'] = ProcessSession()
    request.context['processes'] = {}
    ret = await handler(request)
    await request.context['process_session'].wait_for_all()
    return ret

Util:

import asyncio
import concurrent.futures
from functools import partial

class ProcessSession():

    def __init__(self):
        self.loop = asyncio.get_running_loop()
        self.pool = concurrent.futures.ProcessPoolExecutor()
        self.futures = []

    async def wait_for_all(self):
        await asyncio.wait(self.futures)

    def add_process(self, f, *args, **kwargs):
        ret = self.loop.run_in_executor(self.pool, partial(f, *args, **kwargs))
        self.futures.append(ret)
        return ret

class ProcessBase():

    def __init__(self, process_session, f, *args, **kwargs):
        self.future = process_session.add_process(f, *args, **kwargs)

    async def wait(self):
        await asyncio.wait([self.future])
        return self.future.result()
python-asyncio multithreading python
2 Answers

Answering your question: Yes, it does block the event loop.

I found that ThreadPoolExecutor works pretty well on this situations.

from util.process_session import ProcessSession
from concurrent.futures.thread import ThreadPoolExecutor
import asyncio

from aiohttp.web import middleware

@middleware
async def context_init(request, handler):
    request.context = {}
    request.context['threads'] = []
    ret = await handler(request)
    with ThreadPoolExecutor(1) as executor:
           await asyncio.get_event_loop().run_in_executor(executor, 
           functools.partial(join_threads, data={
             'threads': request.context['threads']
           }))
    return ret

def join_threads(threads):
    for t in threads:
        t.join()
user_105268690

I found a solution using multiprocesses. It can be done using a Pool. The standard lib provides some "async" methods (It's not really async, it just separates the initialization of the process from the process' output): apply_async

Using a simple async wrapper, I managed to deliver what I wanted:

from multiprocessing import Pool
from async_converter import sync_to_async
import asyncio

def f(x):
    i = 0
    while i < 10000000 * x:
       i = i + 1
    print("Finished: " + str(x))
    return i


async def run():
    print("Started with run")
    with Pool(processes=4) as pool:         # start 4 worker processes
        result1 = pool.apply_async(f, (10,)) # evaluate "f(10)" asynchronously
        result2 = pool.apply_async(f, (2,))
        res1 = await sync_to_async(result1.get)()
        print(res1)
        res2 = await sync_to_async(result2.get)()
        print(res2)

async def dummy(output):
    print(output)

async def main():
    # Schedule three calls *concurrently*:
    await asyncio.gather(
        run(),
        dummy("Nice"),
        dummy("Async"),
        dummy("Loop"),
        dummy("Perfect"),
        dummy("Dummy1"),
        dummy("Dummy2"),
        dummy("Dummy3"),
        dummy("Dummy4"),
        dummy("Dummy5"),
        dummy("Dummy6"),
        dummy("Dummy7"),
        dummy("Dummy8"),
        dummy("Dummy9"),
        dummy("Dummy10"),
    )

loop = asyncio.get_event_loop()
loop.run_until_complete(main())
loop.close()

outputs:

Perfect
Dummy6
Nice
Dummy1
Dummy7
Started with run
Dummy2
Dummy8
Dummy3
Dummy9
Async
Dummy4
Dummy10
Loop
Dummy5
Finished: 2
Finished: 10
100000000
20000000

Parallelism with asyncio :)

user_65010500
Related