fix: prevent closed pipe errors on closing asyncio transport resources

The proactor pipe transports for subprocess won't be automatically
closed, so "closed pipe" errors (pytest warnings) occur during garbage
collection (upon `__del__`). This results in a bunch of pytest warnings
whenever closing and freeing up fixture Nvim sessions. A solution is to
close all the internal `_ProactorBasePipeTransport` objects later when
closing the asyncio event loop.

Also, `_ProactorBasePipeTransport.close()` does not close immediately,
but rather works asynchronously; therefore the `__del__` finalizer still
can throw if called by GC after the event loop is closed. One solution
for properly closing the pipe transports is to await the graceful
shutdown of these transports.

Example CI output (the pytest warnings that are going to be fixed):

```
Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x00000205288AD1C0>
  Traceback (most recent call last):
    File "C:\hostedtoolcache\windows\Python\3.11.5\x64\Lib\asyncio\proactor_events.py", line 116, in __del__
      _warn(f"unclosed transport {self!r}", ResourceWarning, source=self)
            ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
    File "C:\hostedtoolcache\windows\Python\3.11.5\x64\Lib\asyncio\proactor_events.py", line 80, in __repr__
      info.append(f'fd={self._sock.fileno()}')
                        ^^^^^^^^^^^^^^^^^^^
    File "C:\hostedtoolcache\windows\Python\3.11.5\x64\Lib\asyncio\windows_utils.py", line 102, in fileno
      raise ValueError("I/O operation on closed pipe")
  ValueError: I/O operation on closed pipe
  Exception ignored in: <function _ProactorBasePipeTransport.__del__ at 0x00000205288AD1C0>
  Traceback (most recent call last):
    File "C:\hostedtoolcache\windows\Python\3.11.5\x64\Lib\asyncio\proactor_events.py", line 116, in __del__
    File "C:\hostedtoolcache\windows\Python\3.11.5\x64\Lib\asyncio\proactor_events.py", line 80, in __repr__
    File "C:\hostedtoolcache\windows\Python\3.11.5\x64\Lib\asyncio\windows_utils.py", line 102, in fileno
  ValueError: I/O operation on closed pipe
```

Author: wookayin

pynvim/msgpack_rpc/event_loop/asyncio.py

@@ -117,9 +117,10 @@ def _on_data(data: bytes) -> None:
         )
         self._protocol = None
 
-        # The communication channel (endpoint) created by _connect_*() method.
+        # The communication channel (endpoint) created by _connect_*() methods,
+        # where we write request messages to be sent to neovim
         self._transport = None
-        self._raw_transport = None
+        self._to_close: List[asyncio.BaseTransport] = []
         self._child_watcher = None
 
         super().__init__(transport_type, *args, **kwargs)
@@ -161,7 +162,8 @@ async def connect_stdin():
             transport, protocol = await self._loop.connect_read_pipe(
                 self._protocol_factory, pipe)
             debug("native stdin connection successful")
-            del transport, protocol
+            self._to_close.append(transport)
+            del protocol
         self._loop.run_until_complete(connect_stdin())
 
         # Make sure subprocesses don't clobber stdout,
@@ -200,6 +202,16 @@ async def create_subprocess():
                                    transport.get_pipe_transport(0))  # stdin
             self._protocol = protocol
 
+            # proactor transport implementations do not close the pipes
+            # automatically, so make sure they are closed upon shutdown
+            def _close_later(transport):
+                if transport is not None:
+                    self._to_close.append(transport)
+
+            _close_later(transport.get_pipe_transport(1))
+            _close_later(transport.get_pipe_transport(2))
+            _close_later(transport)
+
         # await until child process have been launched and the transport has
         # been established
         self._loop.run_until_complete(create_subprocess())
@@ -230,10 +242,28 @@ def _stop(self) -> None:
 
     @override
     def _close(self) -> None:
-        # TODO close all the transports
-        if self._raw_transport is not None:
-            self._raw_transport.close()  # type: ignore[unreachable]
+        def _close_transport(transport):
+            transport.close()
+
+            # Windows: for ProactorBasePipeTransport, close() doesn't take in
+            # effect immediately (closing happens asynchronously inside the
+            # event loop), need to wait a bit for completing graceful shutdown.
+            if os.name == 'nt' and hasattr(transport, '_sock'):
+                async def wait_until_closed():
+                    # pylint: disable-next=protected-access
+                    while transport._sock is not None:
+                        await asyncio.sleep(0.01)
+                self._loop.run_until_complete(wait_until_closed())
+
+        if self._transport:
+            _close_transport(self._transport)
+            self._transport = None
+        for transport in self._to_close:
+            _close_transport(transport)
+        self._to_close[:] = []
+
         self._loop.close()
+
         if self._child_watcher is not None:
             self._child_watcher.close()
             self._child_watcher = None