asyncio's `create_datagram_socket` seems to leak socket objects

README.md

asyncio's create_datagram_socket seems to leak socket objects when Protocol instances go out of scope. The same is not the case with regular socket objects.

Usage:

• Run this (optional pass the open file limit as positional argument, defaut: 256)
• In another terminal run watch 'lsof -p <pid> | grep UDP | wc -l'
• Observe that the number stays at 50 during the sync run
• In the async run the number will increase until a "Too many open files" exception is raised

asyncio_socket_leak.py

import asyncio
import socket
import time
from asyncio import DatagramProtocol

import os

import sys
from math import ceil


async def make_async():
    loop = asyncio.get_running_loop()
    return [
        (await loop.create_datagram_endpoint(
            lambda: DatagramProtocol(),
            local_addr=('::1', 0)
        ))[1]
        for _ in range(25)
    ]


async def main_async(limit):
    res1 = []
    res2 = []
    for _ in range(ceil((limit + 1) / 25)):
        res1, res2 = res2, res1
        res1.extend(await make_async())
        print(len(res1), len(res2))
        await asyncio.sleep(.5)
        print('clear')
        res2.clear()
        await asyncio.sleep(.5)


def make_sync():
    socks = []
    for _ in range(25):
        sock = socket.socket(type=socket.SOCK_DGRAM)
        sock.bind(("127.0.0.1", 0))
        socks.append(sock)
    return socks


def main_sync(limit):
    res1 = []
    res2 = []
    for _ in range(ceil((limit + 1) / 25)):
        res1, res2 = res2, res1
        res1.extend(make_sync())
        print(len(res1), len(res2))
        time.sleep(.5)
        print('clear')
        res2.clear()
        time.sleep(.5)


if __name__ == "__main__":
    if len(sys.argv) > 1:
        limit = int(sys.argv[1])
        print(f'File limit set to {limit}')
    else:
        limit = 256
        print("Assuming file limit of 256. Pass arg to change.")
    print(f"PID: {os.getpid()}")
    print('press enter to start')
    input()
    print("=========== Run sync ===========")
    main_sync(limit)
    print("done\n=========== Run Async ===========")
    asyncio.run(main_async(limit))