I have an event that I am listening to every minute that returns a list ; it could be empty, 1 element, or more. And with those elements in that list, I'd like to run a function that would monitor an event on that element every minute for 10 minute.

For that I wrote that script

from concurrent.futures import ThreadPoolExecutor
from time import sleep
import asyncio
import Client

client = Client()

def handle_event(event):
    for i in range(10):
        client.get_info(event)
        sleep(60)

async def main():
    while True:
        entires = client.get_new_entry()

        if len(entires) > 0:
            with ThreadPoolExecutor(max_workers=len(entires)) as executor:
                executor.map(handle_event, entires)

        await asyncio.sleep(60)

if __name__ == "__main__":
    loop = asyncio.new_event_loop()
    loop.run_until_complete(main())

However, instead of keep monitoring the entries, it blocks while the previous entries are still being monitors.

Any idea how I could do that please?

CodePudding user response：

First let me explain why your program doesn't work the way you want it to: It's because you use the ThreadPoolExecutor as a context manager, which will not close until all the threads started by the call to map are finished. So main() waits there, and the next iteration of the loop can't happen until all the work is finished.

There are ways around this. Since you are using asyncio already, one approach is to move the creation of the Executor to a separate task. Each iteration of the main loop starts one copy of this task, which runs as long as it takes to finish. It's a async def function so many copies of this task can run concurrently.

I changed a few things in your code. Instead of Client I just used some simple print statements. I pass a list of integers, of random length, to handle_event. I increment a counter each time through the while True: loop, and add 10 times the counter to every integer in the list. This makes it easy to see how old calls continue for a time, mixing with new calls. I also shortened your time delays. All of these changes were for convenience and are not important.

The important change is to move ThreadPoolExecutor creation into a task. To make it cooperate with other tasks, it must contain an await expression, and for that reason I use executor.submit rather than executor.map. submit returns a concurrent.futures.Future, which provides a convenient way to await the completion of all the calls. executor.map, on the other hand, returns an iterator; I couldn't think of any good way to convert it to an awaitable object.

To convert a concurrent.futures.Future to an asyncio.Future, an awaitable, there is a function asyncio.wrap_future. When all the futures are complete, I exit from the ThreadPoolExecutor context manager. That will be very fast since all of the Executor's work is finished, so it does not block other tasks.

import random
from concurrent.futures import ThreadPoolExecutor
from time import sleep
import asyncio

def handle_event(event):
    for i in range(10):
        print("Still here", event)
        sleep(2)
        
async def process_entires(counter, entires):
    print("Counter", counter, "Entires", entires)
    x = [counter * 10   a for a in entires]
    with ThreadPoolExecutor(max_workers=len(entires)) as executor:
        futs = []
        for z in x:
            futs.append(executor.submit(handle_event, z))
        await asyncio.gather(*(asyncio.wrap_future(f) for f in futs))

async def main():
    counter = 0
    while True:
        entires = [0, 1, 2, 3, 4][:random.randrange(5)]

        if len(entires) > 0:
            counter  = 1
            asyncio.create_task(process_entires(counter, entires))

        await asyncio.sleep(3)

if __name__ == "__main__":
    asyncio.run(main())