PipeServer closure race

[hartytp]

I'm tracking down a race in the asyncio.windows_events.PipeServers close method. AFAICT the issue is this:

• the PipeSever stores weakrefs to each PipeHandle it opens
• when we call close, it calls the close method for any live PipeHandles
• however, the IocpProactor runs in a different thread and can store references to the PipeHandles
• when those references are released (e.g. here https://github.com/harvimt/quamash/blob/e513b30f137415c5e098602fa383e45debab85e7/quamash/init.py#L210 ) it can trigger the PipeHandle.__del__ to be called in a different thread to the user code. Thus, there is a race between __del__ and PipeServer.close

Possibly related to https://github.com/harvimt/quamash/issues/55

Any suggestions for a fix for this? Thanks!

Asyncio PipeServer code below for reference...

class PipeServer(object):
    """Class representing a pipe server.

    This is much like a bound, listening socket.
    """
    def __init__(self, address):
        self._address = address
        self._free_instances = weakref.WeakSet()
        # initialize the pipe attribute before calling _server_pipe_handle()
        # because this function can raise an exception and the destructor calls
        # the close() method
        self._pipe = None
        self._accept_pipe_future = None
        self._pipe = self._server_pipe_handle(True)


    def _get_unconnected_pipe(self):
        # Create new instance and return previous one.  This ensures
        # that (until the server is closed) there is always at least
        # one pipe handle for address.  Therefore if a client attempt
        # to connect it will not fail with FileNotFoundError.
        tmp, self._pipe = self._pipe, self._server_pipe_handle(False)
        return tmp


    def _server_pipe_handle(self, first):
        # Return a wrapper for a new pipe handle.
        if self.closed():
            return None
        flags = _winapi.PIPE_ACCESS_DUPLEX | _winapi.FILE_FLAG_OVERLAPPED
        if first:
            flags |= _winapi.FILE_FLAG_FIRST_PIPE_INSTANCE
        h = _winapi.CreateNamedPipe(
            self._address, flags,
            _winapi.PIPE_TYPE_MESSAGE | _winapi.PIPE_READMODE_MESSAGE |
            _winapi.PIPE_WAIT,
            _winapi.PIPE_UNLIMITED_INSTANCES,
            windows_utils.BUFSIZE, windows_utils.BUFSIZE,
            _winapi.NMPWAIT_WAIT_FOREVER, _winapi.NULL)
        pipe = windows_utils.PipeHandle(h)
        self._free_instances.add(pipe)
        return pipe


    def closed(self):
        return (self._address is None)


    def close(self):
        if self._accept_pipe_future is not None:
            self._accept_pipe_future.cancel()
            self._accept_pipe_future = None
        # Close all instances which have not been connected to by a client.
        if self._address is not None:
            for pipe in self._free_instances:
                pipe.close()
            self._pipe = None
            self._address = None
            self._free_instances.clear()


    __del__ = close

https://github.com/python/cpython/blob/bfba8c373e362d48d4ee0e0cf55b8d9c169344ae/Lib/asyncio/windows_events.py#L241-L297