Reproduce socket not being closed

gistfile1.py

import asyncio
import os
import signal
import functools
import socket
import traceback
from aiohttp import *
import aiohttp
from aiohttp import web

def find_unused_port():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('127.0.0.1', 0))
    port = s.getsockname()[1]
    s.close()
    return port

@asyncio.coroutine
def create_server(loop, method="GET", path="/", port=None, handler=None):
    app = web.Application()
    if handler:
        app.router.add_route(method, path, handler)

    port = port or find_unused_port()
    srv = yield from loop.create_server(
        app.make_handler(), '127.0.0.1', port)
    url = "http://127.0.0.1:{}".format(port) + path
    return app, srv, url

@asyncio.coroutine
def origin_handler(request):
    body = yield from request.read()
    return web.Response(body=b'OK' * 100000)

# create origin server
loop = asyncio.get_event_loop()
app, srv, origin_url = loop.run_until_complete(create_server(loop, origin_handler))

connector = TCPConnector()
@asyncio.coroutine
def proxy_handler(request):
    # call origin
    try:
        r = yield from aiohttp.request('get', origin_url, connector=connector)
        body = yield from r.read()
        return web.Response(body=body)
    finally:
        if r:
            yield from r.release()
            r.close()

origin_pid = os.fork()
if origin_pid:
    # origin
    print("origin pid: {}, url: {}".format(origin_pid, origin_url))
    loop.run_forever()
else:
    # proxy
    app, srv, proxy_url = loop.run_until_complete(create_server(loop, port=1234, handler=proxy_handler))
    print("proxy pid: {}, url: {}".format(os.getpid(), proxy_url))
    loop.run_forever()
    

Once you start this code, hammer the proxy server with requests, like so:

ab -n 10000000 -c 1000 http://127.0.0.1:1234/

Then inspect open sockets on the proxy process.