Fix Process.stdin.send exceptions and checkpointing on asyncio

[gschaffner]

Changes

Fixes the remaining issue reported in #671:

_asyncio.Process.stdin.send not raising ClosedResourceError or BrokenResourceError, but rather non-AnyIO exceptions. (The analogous std{out,err} case was already fixed in #752.)

I also fixed a missing checkpoint before writing to a subprocess send buffer on asyncio.

I also made _close = False unconditional in asyncio UDP aclose, similar to #980. I'm not sure that this is strictly necessary, but it seems safer and I don't see any downsides. See details in commit message.

Checklist

• You've added tests (in tests/) which would fail without your patch
• You've updated the documentation (in docs/), in case of behavior changes or new
  features
• You've added a new changelog entry (in docs/versionhistory.rst).

[gschaffner]

Off topic: I noticed that we don't have ResourceGuards in _asyncio.Stream{Reader,Writer}Wrapper and _trio.{Receive,Send}StreamWrapper. Should we add guards?

[agronholm]

Perhaps that would be wise. I'm not sure if Trio doesn't already have a mechanism in place to guard against concurrent access. It would be good to check first.

[gschaffner]

Trio already has its own ConflictDetectors in standard streams (FdStream, PipeSendStream, and PipeReceiveStream), but they raise trio.BusyResourceError, not anyio.BusyResourceError, so I think we should to add our own guards on the Trio backend too.

Perhaps let's handle this in a separate PR after this one. It looks like there are also a few other places where one can currently get a ConflictDetector's trio.BusyResourceError on AnyIO instead of an anyio.BusyResourceError.

[gschaffner]

@Vizonex there's a new test here that is only failing on Winloop—would you be able to take a look? I will take a look tomorrow too.

[agronholm]

Looks good. I left some formatting changes and other comments.

[agronholm]

We already have self._stream.drain(), so I don't think we need a second full checkpoint here. What we should add instead is AsyncIOBackend.checkpoint_if_cancelled() to respond to cancellation before the synchronous stream.write().

[gschaffner]

It looks like drain() might not checkpoint in all branches: https://github.com/python/cpython/blob/v3.14.0/Lib/asyncio/streams.py#L167-L168

[agronholm]

OK, how about checkpointing based on self._stream._paused?

   async def send(self, item: bytes) -> None:
        await checkpoint_if_cancelled()
        stream_paused = self._stream._paused
        try:
            self._stream.write(item)
            await self._stream.drain()
        except (ConnectionResetError, BrokenPipeError, RuntimeError) as exc:
            # If closed by us and/or the peer:
            # * on stdlib, drain() raises ConnectionResetError or BrokenPipeError
            # * on uvloop, write() raises RuntimeError and drain() doesn't appear to
            #   raise anything
            if self._closed:
                raise ClosedResourceError from exc
            elif self._stream.is_closing():
                raise BrokenResourceError from exc

            raise

        if not stream_paused:
            await cancel_shielded_checkpoint()

I know it's a private attribute but our CI should alert us if it's ever changed.

[gschaffner]

Sounds good. I'm assuming that you meant self._stream._protocol._paused since self._stream._paused does not exist.

However: when I do make this change, test_exceptions_after_subprocess_closes_standard_streams[asyncio+uvloop] starts failing. (It's fine on stdlib asyncio.) Without the checkpoint, it seems like we might be hitting a uvloop bug that is similar to the Winloop bug. Looking into it.

[gschaffner]

Done.

[agronholm]

Perhaps that would be wise. I'm not sure if Trio doesn't already have a mechanism in place to guard against concurrent access. It would be good to check first.

[agronholm]

Note that a test on Windows + Python 3.9 is genuinely failing.

[gschaffner]

@Vizonex there's a new test here that is only failing on Winloop—would you be able to take a look? I will take a look tomorrow too.

Note that a test on Windows + Python 3.9 is genuinely failing.

See Vizonex/Winloop#84

[Vizonex]

@Vizonex there's a new test here that is only failing on Winloop—would you be able to take a look? I will take a look tomorrow too.

Note that a test on Windows + Python 3.9 is genuinely failing.

See Vizonex/Winloop#84

@gschaffner @agronholm I saw it and I have now made a fork. Thanks for writing me a reproducible test those really help me understand what it is that is going on. I'll try and come up with a template as well when the bug for winloop is fixed so people know I want reproducible tests =)

[Vizonex]

@agronholm I fixed the bug on winloop's end there should be a release for 0.3.1 shortly here later today.

[agronholm]

@agronholm I fixed the bug on winloop's end there should be a release for 0.3.1 shortly here later today.

There seems to be a failure on 3.13 now with winloop. 3.9 passes though.

[gschaffner]

It looks like test_exceptions_after_subprocess_closes_standard_streams[asyncio+winloop] is still failing sporadically on Winloop 0.3.1, on both 3.9 and 3.13. Looking into it.

[Vizonex]

It looks like test_exceptions_after_subprocess_closes_standard_streams[asyncio+winloop] is still failing sporadically on Winloop 0.3.1, on both 3.9 and 3.13. Looking into it.

@gschaffner If your lucky I might be able to do another bugfix tomorrow or make better tests available. (This might be related/unrelated) but winloop & uvloop are suffering from rounding bugs as well...

[gschaffner]

When I remove the schedule point at the start of send(), the test that is sporadically failing on asyncio+winloop (test_exceptions_after_subprocess_closes_standard_streams) starts failing on asyncio+uvloop too (but not stdlib asyncio). I'm wondering if there is a problem with the approach or the test here.

I suppose it's just a race? Perhaps it's fine for subprocess.stdin.send() to return success once or twice after the peer has closed their end of the pipe—even when the event loop¹ already knows that the peer closed their end of the pipe—as long as send() eventually starts raising BrokenResourceError?

Footnotes

1. or at least some parts of the event loop ↩

[gschaffner]

I.e. perhaps the test is too strict, and we should amend it?:

diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index c18df94..736ef8c 100644
--- a/tests/test_subprocesses.py
+++ b/tests/test_subprocesses.py
@@ -275,7 +275,8 @@ async def test_exceptions_after_subprocess_closes_standard_streams() -> None:
         assert process.stderr is not None
         assert process.stdout is not None
         with pytest.raises(BrokenResourceError):
-            await process.stdin.send(b"foo")
+            for _ in range(5):
+                await process.stdin.send(b"foo")
 
         with pytest.raises(EndOfStream):
             await process.stdout.receive(1)

[gschaffner]

On the other hand:

• On Trio, subprocess.stdin.send() will always raise after the peer closed the pipe, because on Trio subprocess.stdin.send() uses os.write directly.

• On stdlib asyncio, subprocess.stdin.send() will (IIUC) always raise after the peer closed the pipe if subprocess.wait() was called, because of this checkpoint: https://github.com/python/cpython/blob/v3.14.0/Lib/asyncio/streams.py#L385

[gschaffner]

I've gone ahead and weakened the test to accommodate uvloop's and Winloop's transmission delays. Let me know what you think.

[agronholm]

Thanks!

[Vizonex]

@agronholm @gschaffner Thank both of you, glad I could be of assistance and that you helped me hunt down some bugs with my library.