client.py

import asyncio
import logging
import time
import random
import aiohttp
import traceback

logging.basicConfig(level=logging.DEBUG)

async def do_stuff(sess):
    cpu_hog = False
    while True:
        condition = ''
        try:
            async with sess.get('http://127.0.0.1:12345/') as resp:
                condition = resp.headers['X-Condition']
                got = 0
                async for chunk in resp.content.iter_chunked(8192):
                    got += len(chunk)
                    lunch = random.random() * 0.001
                    if cpu_hog:
                        time.sleep(lunch)
                    else:
                        await asyncio.sleep(lunch)

                print(condition, '<<<', got, 'bytes')
        except Exception as e:
            print(condition, e)
            traceback.print_exc()

        lunch = random.random() * 0.1
        if cpu_hog:
            time.sleep(lunch)
        else:
            await asyncio.sleep(lunch)

async def main():
    c = aiohttp.TCPConnector(
        limit=3,
        keepalive_timeout=1,
        # timeout_ceil_threshold=0.5,
    )
    async with aiohttp.ClientSession(connector=c) as sess:
        async with asyncio.TaskGroup() as g:
            for _ in range(20):
                g.create_task(do_stuff(sess))

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

server.py

import asyncio
import logging
import random

logging.basicConfig(level=logging.DEBUG)


async def fugazi_web(rd, wr):
    end_chunk_choices = (True, False)
    close_connection_choices = (True, False)

    while True:
        end_chunk = random.choice(end_chunk_choices)
        close_connection = random.choice(close_connection_choices)

        req = await rd.read(8192)

        resp = '\r\n'.join((
            'HTTP/1.1 200 OK',
            'Connection: keep-alive',
            'Transfer-encoding: chunked',
            f'X-Condition: {end_chunk}',
            '',
            '',
        )).encode()

        wr.write(resp)
        await wr.drain()

        while random.random() < 0.95:
            chunk_len = random.randint(1, 8192 + 5)
            chunk = f'{chunk_len:x}\r\n{"A" * chunk_len}\r\n'.encode()
            wr.write(chunk)
            await wr.drain()

        if end_chunk:
            wr.write(b'0\r\n\r\n')
            await wr.drain()

        if close_connection:
            wr.close()
            await wr.wait_closed()

    wr.close()
    await wr.wait_closed()


async def main():
    srv = await asyncio.start_server(
        fugazi_web,
        '127.0.0.1',
        12345,
    )

    async with srv:
        await srv.serve_forever()

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