aio-libs/aiohttp#1821

_README.md

First open three terminals, then (wait for the 1st to be running before attempting 2 and 3):

If you do not wish to use docker, simply use the second lines only.

1.  docker container run -it -v $PWD:/app -w /app --name testcase-1821 --rm python:3.5 \
      sh -c 'sh setup.sh; python3 server.py'
   

2.  docker container exec -it testcase-1821 \
      watch -n 0.1 'netstat -npt | grep $(pgrep python | tail -n1) | wc -l'
   

3.  docker container exec -it testcase-1821 \
      python3 client.py
   

While 3 is running you should see a connection count every 100 requests together with the current # of connections in 2. 3 will first run 10000 reqs using the limit in aiohttp, and then another 10000 reqs using asyncio.Semaphore to limit requests.

At the time of writting, using limit see connections drop dramatically at some point during the last 3000 requests. Very occasionally this will not happen, but a second attempt nearly always shows the connection drop. Conversely, using the semaphore approach consistently maintains connection levels and finishes much faster.

client.py

from asyncio import Semaphore, as_completed, get_event_loop
from collections import Counter
from os import getpid, popen
from time import time

from aiohttp import ClientSession, TCPConnector


async def req(session, i, lock=None):
    if lock:
        async with lock:
            return await req(session, i)
    async with session.request('GET', 'http://127.0.0.1:4040/foo', timeout=None) as res:
        return res.status

async def limit_builtin():
    conn = TCPConnector(limit=LIMIT, loop=loop)
    out = []
    async with ClientSession(connector=conn, loop=loop) as session:
        for res in as_completed(tuple(req(session, i) for i in it)):
            out.append(await res)
            if len(out) % LIMIT == 0:
                print('\033[K  Conns @ {}: {}'.format(
                    len(out),
                    popen(script).read().strip()
                ), end='\r', flush=True)
    return out

async def limit_semaphore():
    lock = Semaphore(LIMIT)
    conn = TCPConnector(limit=None, loop=loop)
    out = []
    async with ClientSession(connector=conn, loop=loop) as session:
        for res in as_completed(tuple(req(session, i, lock=lock) for i in it)):
            out.append(await res)
            if len(out) % LIMIT == 0:
                print('\033[K  Conns @ {}: {}'.format(
                    len(out),
                    popen(script).read().strip()
                ), end='\r', flush=True)
    return out

def present(fn):
    print('{}:'.format(fn.__name__))
    start = time()
    out = loop.run_until_complete(fn())
    took = time() - start
    print('  Responses: {}'.format(Counter(out)))
    print('  Took:      {} seconds'.format(took))

if __name__ == '__main__':
    LIMIT = 100
    loop = get_event_loop()
    script = 'netstat -npt | grep {} | wc -l'.format(getpid())
    it = range(1, LIMIT * 100)

    print(script)
    present(limit_builtin)
    present(limit_semaphore)

fast.pstat

fast.py

from asyncio import Semaphore, as_completed, get_event_loop
from collections import Counter
from time import time

from aiohttp import ClientSession, TCPConnector


async def req(session, i, lock=None):
    if lock:
        async with lock:
            return await req(session, i)
    async with session.request('GET', 'http://localhost:4040/foo', timeout=None) as res:
        return res.status

async def limit_semaphore():
    lock = Semaphore(LIMIT)
    conn = TCPConnector(limit=None, loop=loop)
    out = []
    async with ClientSession(connector=conn, loop=loop) as session:
        for res in as_completed(tuple(req(session, i, lock=lock) for i in it)):
            out.append(await res)
    return out

def present(fn):
    print('{}:'.format(fn.__name__))
    start = time()
    out = loop.run_until_complete(fn())
    took = time() - start
    print('  Responses: {}'.format(Counter(out)))
    print('  Took:      {} seconds'.format(took))

if __name__ == '__main__':
    LIMIT = 100
    loop = get_event_loop()
    it = range(1, LIMIT * 100)
    present(limit_semaphore)

server.py

from asyncio import sleep

from sanic import Sanic
from sanic.response import text


app = Sanic()

@app.route('/foo')
async def foo(request):
    await sleep(0.001)
    return text('hello')

app.run(host='0.0.0.0', port=4040)

setup.sh

#!/bin/sh

if ! which netstat
  then apt-get update && apt-get install -y net-tools
fi

pip3 install aiohttp sanic

slow.pstat

slow.py