Ufdjrw

I want to do some calculation on lines a huge file, and I find reading the lines take less than 5% of the time, calculation each lines takes 95%.

Then I find asyncio.Queue seems a good abstract for my need.
However, it needs to use concurrent.futures.ThreadPoolExecutor or ProcessPoolExecutor to achieve parallel. Without them, asyncio is single threaded.

I cannot find an example to know how to write it.
Most examples only prints in consumer, without return any values.

Would you please write an example for me ?

For demonstration, the producer put many lines of 10 numbers to queue, the 5 consumers split the line and calculate sum and average for a single line, and the main() adds up the sums and the averages to print two results in the end.

The default example is single threded. Just change await asyncio.sleep(sleep_for) to for i in range(10000000):pass.

And then, only "worker-0" shown up. The CPU% is always less than 100%.

import asyncio
import random
import time


async def worker(name, queue):
 while True:
 # Get a "work item" out of the queue.
 sleep_for = await queue.get()
 print(f'name has Begin')

 # Sleep for the "sleep_for" seconds.
 #await asyncio.sleep(sleep_for)
 for i in range(10000000):pass

 # Notify the queue that the "work item" has been processed.
 queue.task_done()

 print(f'name has slept for sleep_for:.2f seconds')


async def main():
 # Create a queue that we will use to store our "workload".
 queue = asyncio.Queue()

 # Generate random timings and put them into the queue.
 total_sleep_time = 0
 for _ in range(20):
 sleep_for = random.uniform(0.05, 1.0)
 total_sleep_time += sleep_for
 queue.put_nowait(sleep_for)

 # Create three worker tasks to process the queue concurrently.
 tasks = []
 for i in range(3):
 task = asyncio.create_task(worker(f'worker-i', queue))
 tasks.append(task)

 # Wait until the queue is fully processed.
 started_at = time.monotonic()
 await queue.join()
 total_slept_for = time.monotonic() - started_at

 # Cancel our worker tasks.
 for task in tasks:
 task.cancel()
 # Wait until all worker tasks are cancelled.
 await asyncio.gather(*tasks, return_exceptions=True)

 print('====')
 print(f'3 workers slept in parallel for total_slept_for:.2f seconds')
 print(f'total expected sleep time: total_sleep_time:.2f seconds')


asyncio.run(main())

Also, I hope the code is backward compatible with Python 3.5.3, since PyPy seems faster.