How to implement multi-threaded asyncio outside a function?

Question

I am trying to run a multi-threaded program with asyncio, but I am failing at the part of adding threads. The program runs ok with asyncio as it is:

async def main(var1, var2):
    tasks = list()
    for z in var1:
        for x in range(5):
            tasks.append(get_ip(z, var2))
    return await asyncio.gather(*tasks)


loop = asyncio.get_event_loop()
start_time = time.time()
for x in list:
    result = loop.run_until_complete(main(x, list2))
    loop.run_until_complete(release_main(result))
loop.close()

I want to have the for x in list: in threads, I have 8 CPUs so I would want to run it with 8 threads like for example using: with concurrent.futures.ThreadPoolExecutor(max_workers=8) as executor:.

I have been reading posts and everything but I either mess the result from result, or break something and doesn't work. Help/tips needed.

await loop.run_in_executor() doesn't work if I don't have that in a function, is it really needed? when I add the above code in a function and call it it breaks everything

Asyncio loops are single threaded. You cannot share a single loop between multiple threads, except for a very few threadsafe functions like call_soon_threadsafe. So you would need to manually spawn threads, manually spawn a separate loop in each thread, run whatever you need and finally synchronize results afterwards. But really, due to GIL this doesn't seem to be worth it anyway. Python is quite bad at parallel execution. — freakish
– freakish, Commented Mar 10, 2022 at 9:07
If I am able to get differente asyncio loops in different threads, it should make my code x5 or x6 times faster, as right now it runs with 1 of 8 cpu cores, running throw the loop woth 7-8 cores should make a big difference, and that's what I want. Any idea how to code it? I'm a begginer and this kind of things really confuse me — Tomi Begher
– Tomi Begher, Commented Mar 10, 2022 at 16:28
What makes you think it will run x5 or x6 times faster? What makes you think it will run faster at all? Again: Python has a big limitation called Global Interpreter Lock (GIL), which makes Python threads only usable for i/o. But asyncio already handles i/o efficiently. It is extremely unlikely that threads will give you any performance benefit compared to single-threaded asyncio. — freakish
– freakish, Commented Mar 10, 2022 at 17:04
I run this same exact program using multi-proccesing and it took 150 seconds (using 8 of 8 cpu cores). running the same program with asyncio took 300 seconds (using 1 of 8 cores). I really think combining asyncio with multi-threading using all cores would be much much faster, but I don't know how to implement it correctly. I tried asyncio.run_coroutine_threadsafe() but it just doesn't run my program with that. — Tomi Begher
– Tomi Begher, Commented Mar 10, 2022 at 17:19
At the end of the day I am not sure if it would be faster or not, but I want to try it myself somehow :) — Tomi Begher
– Tomi Begher, Commented Mar 10, 2022 at 17:20

freakish · Accepted Answer · 2022-03-10 18:23:22Z

0

Asyncio loops are not thread safe and are supposed to be used in a single thread, except for few thread safe function e.g. run_coroutine_threadsafe. And so one thing you can do is to spawn multiple threads and then create a separate loop in each thread:

import asyncio
import multiprocessing
import threading
import queue

# Spawn threads and run new loop in each thread
loops = queue.Queue()
def thread_job():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    loops.put(loop)
    loop.run_forever()

for _ in range(multiprocessing.cpu_count()):
    t = threading.Thread(target=thread_job)
    t.start()

# Our main
async def main(x, ev):
    try:
        await asyncio.sleep(1)
        print(x, threading.current_thread().ident)
    finally:
        ev.set()  # events are used to signalize that we are done

# Our job, that only runs main
def job(x, ev):
    loop = asyncio.get_event_loop()
    loop.create_task(main(x, ev))


# Main thread that synchronizes everything
lst = [i for i in range(3)]

events = []
for x in lst:
    ev = threading.Event()
    events.append(ev)

    # by putting the loop back to queue
    # we effectively shift the queue and
    # distribute the load equally
    loop = loops.get()
    loops.put(loop)

    loop.call_soon_threadsafe(job, x, ev)

for ev in events:
    ev.wait()

# Once all events are set, i.e. we are done, cleanup
while not loops.empty():
    loop = loops.get()
    loop.call_soon_threadsafe(loop.stop)

I use a FIFO queue for loops here, in order to distribute load equally between threads.

answered Mar 10, 2022 at 18:23

freakish

57k12 gold badges141 silver badges181 bronze badges

Sign up to request clarification or add additional context in comments.

3 Comments

Tomi Begher Over a year ago

That looks perfect, yet I still can't understand most of it (due to lack of knowledge, not because of your code). if it doesn't bother you, can you explain/rewrite how would you use this code with the 5 lines of code I wrote in the question? that way would be much easier for me to understand the code. This looks like what I am looking for, but im sure I will break something when implementing it.. (gonna update in half an hour).

Tomi Begher Over a year ago

I added main(), to show more of what it looks like

freakish Over a year ago

@TomiBegher I don't know how to do that in 5 lines, sorry. With my code you have to replace my main with your main (and add an additional ev parameter).

Collectives™ on Stack Overflow

How to implement multi-threaded asyncio outside a function?

1 Answer 1

3 Comments

Your Answer

Linked

Hot Network Questions

Collectives™ on Stack Overflow

1 Answer 1

3 Comments

Your Answer

Sign up or log in

Post as a guest

Linked

Related