Making 100 million requests with Python aiohttp

Andy Balaam
artificialworlds.net/blog

What is async?

Concurrency without threads

What is async?

Concurrency without threads
"Everything runs in parallel except your code"*

*Felix Geisendörfer, "Understanding node.js"

What is async?

Concurrency without threads
"Everything runs in parallel except your code"*
Launch some IO, and get a callback when it's done

What is async?

Concurrency without threads
"Everything runs in parallel except your code"*
Launch some IO, and get a callback when it's done
Except: in Python the callback is hidden

What is async?

import asyncio
async def mycoro(number):
    print("Starting %d" % number)   # Runs now
    await asyncio.sleep(1)
    print("Finishing %d" % number)
    return str(number)

What is async?

import asyncio
async def mycoro(number):
    print("Starting %d" % number)
    await asyncio.sleep(1)
    print("Finishing %d" % number)  # Runs later
    return str(number)

async in Python

coroutines, futures and tasks are things you can wait for

async in Python

coroutines, futures and tasks are things you can wait for
the event loop actually runs them (not you)

async in Python

import asyncio
async def mycoro(number):
    print("Starting %d" % number)
    await asyncio.sleep(1)
    print("Finishing %d" % number)
    return str(number)

Define a coroutine function with async def

async in Python

import asyncio
async def mycoro(number):
    print("Starting %d" % number)
    await asyncio.sleep(1)
    print("Finishing %d" % number)
    return str(number)

Inside your coroutine, wait for something with await

async in Python

>>> mycoro(3)
<coroutine object mycoro at 0x7ff70b91f0a0>

Create a coroutine object by calling the coroutine function
This does not run it!

async in Python

c = mycoro(3)
task = asyncio.ensure_future(c)

# or, Python 3.7+:
task = asyncio.create_task(c)

Run it with ensure_future or (better) create_task
This does run it!

async in Python

$ python3 python-async.py
Task was destroyed but it is pending!
task: <Task pending coro=<mycoro() running at python-async.py:10>>
sys:1: RuntimeWarning: coroutine 'mycoro' was never awaited

You must wait for your tasks

async in Python

task = asyncio.ensure_future(mycoro(1))
loop = asyncio.get_event_loop()
loop.run_until_complete(task)
loop.close()

Wait for something using run_until_complete

async in Python

# or, Python 3.7+:
asyncio.run(mycoro(1))

Hurrah Python 3.7+!

async in Python

$ python3 python-async.py
Starting 1
Finishing 1

async in Python

many = asyncio.gather(
    mycoro(1),
    mycoro(2),
    mycoro(3)
)

asyncio.run(many)

Combine tasks together with gather

async in Python

$ python3 python-async.py
Starting 3
Starting 1
Starting 2
Finishing 3
Finishing 1
Finishing 2
['1', '2', '3']

async in Python

import asyncio
async def f2():
    print("start f2")
    await asyncio.sleep(1)
    print("stop f2")
async def f1():
    print("start f1")
    await f2()
    print("stop f1")
asyncio.run(f1())

Call one coroutine from inside another with await

async in Python

$ python3 python-async.py
start f1
start f2
stop f2
stop f1

async in Python

HTTP requests

from aiohttp import ClientSession
import asyncio

async def fetch(url):
    async with ClientSession() as s, s.get(url) as res:
        ret = await res.read()
        print(ret)
        return ret

asyncio.run(fetch("http://example.com"))

100 million requests!

I want to make lots of HTTP requests
I want to deal with results as they appear
How about asyncio.as_completed?

100 million requests!

async def print_when_done(tasks):
    for res in asyncio.as_completed(tasks):
        print(await res)

coros = [
    fetch("http://example.com")
    for i in range(100_000_000)
]
asyncio.run(print_when_done(coros))

100 million requests!

Problems with using as_completed:
- I can't launch them all simultaneously
- I can't make a list of all the work to do

100 million requests!

async def print_when_done(tasks):
    for res in limited_as_completed(tasks, 1000):
        print(await res)

coros = (
    fetch("http://example.com")
    for i in range(100_000_000)
)
asyncio.run(print_when_done(coros))

100 million requests!

limited_as_completed:
- Takes a generator of awaitables instead of a list
- Limits the number of concurrent tasks

100 million requests!

def limited_as_completed(coros, limit):
    futures = [
        asyncio.ensure_future(c)
        for c in islice(coros, 0, limit)
    ]
    async def first_to_finish():
        # Wait until something finishes.
        # Remove it from futures.
        # Add a new task to futures.
        # Return the finished one.

    while len(futures) > 0:
        yield first_to_finish()