Repository: agronholm/anyio

Issue:
Closing a process's stdout when there's more data to be read results in internal asyncio error
When using anyio with asyncio (This works fine with trio), if a subprocess is started with piped stdout and the stdout is closed (either explicitly or via context manager) before all data is read, asyncio will crash with a horrible traceback.

MVCE:

```python
import sys
import anyio

CODE = """\
import sys
for _ in range(1000):
    sys.stdout.buffer.write(bytes(range(256)))
"""

async def amain():
    async with await anyio.open_process([sys.executable, "-c", CODE]) as p:
        print(repr(await p.stdout.receive()))
        await p.stdout.aclose()  # The error even happens if you comment out this line!

if __name__ == "__main__":
    anyio.run(amain)
```

<details>
<summary>Output (after skipping the <code>print</code>'s output)</summary>

```
handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')>
Traceback (most recent call last):
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/subprocess.py", line 72, in pipe_data_received
    reader.feed_data(data)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/streams.py", line 456, in feed_data
    assert not self._eof, 'feed_data after feed_eof'
AssertionError: feed_data after feed_eof
Exception in callback SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')
handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')>
Traceback (most recent call last):
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/subprocess.py", line 72, in pipe_data_received
    reader.feed_data(data)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/streams.py", line 456, in feed_data
    assert not self._eof, 'feed_data after feed_eof'
AssertionError: feed_data after feed_eof
Exception in callback SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')
handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')>
Traceback (most recent call last):
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/subprocess.py", line 72, in pipe_data_received
    reader.feed_data(data)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/streams.py", line 456, in feed_data
    assert not self._eof, 'feed_data after feed_eof'
AssertionError: feed_data after feed_eof
Exception in callback SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')
handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')>
Traceback (most recent call last):
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/subprocess.py", line 72, in pipe_data_received
    reader.feed_data(data)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/streams.py", line 456, in feed_data
    assert not self._eof, 'feed_data after feed_eof'
AssertionError: feed_data after feed_eof
Exception in callback SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')
handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\x00\x01\x0...c\xfd\xfe\xff')>
Traceback (most recent call last):
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/subprocess.py", line 72, in pipe_data_received
    reader.feed_data(data)
  File "/usr/local/Cellar/python@3.10/3.10.8/Frameworks/Python.framework/Versions/3.10/lib/python3.10/asyncio/streams.py", line 456, in feed_data
    assert not self._eof, 'feed_data after feed_eof'
AssertionError: feed_data after feed_eof
```

</details>

This behavior has been observed in the following environments:

* macOS 11.6.8, Python 3.10.8, anyio 3.6.2
* Debian 11 (bullseye), Python 3.8.13, anyio 3.6.1

Discussion:
dolamroth: Test #1

process.py
```
import time
import random

while True:
    print(random.random(), flush=True)
    time.sleep(0.01)
```

test_subprocess.py
```
import sys
from pathlib import Path

import anyio
from anyio.streams.buffered import BufferedByteReceiveStream


async def main():
    _script_path = str(Path(__file__).parent / "process.py")
    async with await anyio.open_process([sys.executable, _script_path]) as process:
        stream = BufferedByteReceiveStream(process.stdout)
        while True:
            try:
                output = await stream.receive_until(b"\n", 6)
                print(output, flush=True)
            except (anyio.ClosedResourceError, anyio.BrokenResourceError, anyio.EndOfStream):
                break


if __name__ == "__main__":
    anyio.run(main)
```

WITHOUT fix -- starts producing these after some time
```
Exception in callback SubprocessStreamProtocol.pipe_data_received(1, b'\r\n0.85027...539023763\r\n')
handle: <Handle SubprocessStreamProtocol.pipe_data_received(1, b'\r\n0.85027...539023763\r\n')>
Traceback (most recent call last):
  File "C:\Users\...\AppData\Local\Programs\Python\Python312\Lib\asyncio\events.py", line 88, in _run
    self._context.run(self._callback, *self._args)
  File "C:\Users\...\AppData\Local\Programs\Python\Python312\Lib\asyncio\subprocess.py", line 72, in pipe_data_received
    reader.feed_data(data)
  File "C:\Users\...\AppData\Local\Programs\Python\Python312\Lib\asyncio\streams.py", line 500, in feed_data
    assert not self._eof, 'feed_data after feed_eof'
           ^^^^^^^^^^^^^
AssertionError: feed_data after feed_eof
```

WITH fix
Outputs numbers without issues. Works on both Windows and Linux.
dolamroth: I think, the second test is not linked to the `feed_eof` issue. Rather, it may be due to a bit contrintuitive behavior of `Process.aclose`.

It turns out, that (in the case, if error pops in user's code and not in subprocess) you need to explicitly terminate or kill process before calling `Process.aclose`, because it calls `await Process.wait()` before `Process.terminate()`.

If we substitute `__aexit__` in the 2nd test with the following code, it works and raises IndexError as expected

```py
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        self._proc.terminate()
        await self._proc.aclose()
        return False
```
agronholm: @graingert correctly pointed out that if the stream is aclosed _while_ another task is reading from it, it will hang, as there is no signal to stop waiting on the read. I'm struggling to write a proper test for this though.
agronholm: I've pushed an alternate solution, and an accompanying test to cover the case pointed out by @graingert.
graingert: Grumbles about asyncio streams

Patch:
diff --git a/docs/versionhistory.rst b/docs/versionhistory.rst
index 72903ae..d99d30a 100644
--- a/docs/versionhistory.rst
+++ b/docs/versionhistory.rst
@@ -29,6 +29,9 @@ This library adheres to `Semantic Versioning 2.0 <http://semver.org/>`_.
 - Fixed ``to_process.run_sync()`` failing to initialize if ``__main__.__file__`` pointed
   to a file in a nonexistent directory
   (`#696 <https://github.com/agronholm/anyio/issues/696>`_)
+- Fixed ``AssertionError: feed_data after feed_eof`` on asyncio when a subprocess is
+  closed early, before its output has been read
+  (`#490 <https://github.com/agronholm/anyio/issues/490>`_)
 - Fixed ``TaskInfo.has_pending_cancellation()`` on asyncio not respecting shielded
   scopes (`#771 <https://github.com/agronholm/anyio/issues/771>`_; PR by @gschaffner)
 - Fixed ``SocketStream.receive()`` returning ``bytearray`` instead of ``bytes`` when
diff --git a/src/anyio/_backends/_asyncio.py b/src/anyio/_backends/_asyncio.py
index b88b9cc..97edb8a 100644
--- a/src/anyio/_backends/_asyncio.py
+++ b/src/anyio/_backends/_asyncio.py
@@ -930,7 +930,7 @@ class StreamReaderWrapper(abc.ByteReceiveStream):
             raise EndOfStream
 
     async def aclose(self) -> None:
-        self._stream.feed_eof()
+        self._stream.set_exception(ClosedResourceError())
         await AsyncIOBackend.checkpoint()
 
 


Test patch:
diff --git a/tests/test_subprocesses.py b/tests/test_subprocesses.py
index 84c4b4d..e8020a7 100644
--- a/tests/test_subprocesses.py
+++ b/tests/test_subprocesses.py
@@ -13,7 +13,13 @@ from typing import Any
 import pytest
 from pytest import FixtureRequest
 
-from anyio import CancelScope, ClosedResourceError, open_process, run_process
+from anyio import (
+    CancelScope,
+    ClosedResourceError,
+    create_task_group,
+    open_process,
+    run_process,
+)
 from anyio.streams.buffered import BufferedByteReceiveStream
 
 pytestmark = pytest.mark.anyio
@@ -289,3 +295,33 @@ async def test_py39_arguments(
                 pytest.skip(f"the {argname!r} argument is not supported by uvloop yet")
 
             raise
+
+
+async def test_close_early() -> None:
+    """Regression test for #490."""
+    code = dedent("""\
+    import sys
+    for _ in range(100):
+        sys.stdout.buffer.write(bytes(range(256)))
+    """)
+
+    async with await open_process([sys.executable, "-c", code]):
+        pass
+
+
+async def test_close_while_reading() -> None:
+    code = dedent("""\
+    import time
+
+    time.sleep(3)
+    """)
+
+    async with await open_process(
+        [sys.executable, "-c", code]
+    ) as process, create_task_group() as tg:
+        assert process.stdout
+        tg.start_soon(process.stdout.aclose)
+        with pytest.raises(ClosedResourceError):
+            await process.stdout.receive()
+
+        process.terminate()