Improve robustness of subprocess text streaming

[zanieb]

Summary

Previously, this utility would call blocking writes to stderr and stdout directly. This appears to introduce the possibility of race conditions and blocked event loops when many processes are run concurrently. Since we use these utilities to launch flows in parallel processes from the agent, it is important that they are robust to concurrency. wrap_file may not be thread safe, but this is still an improvement from where we were. There is not a clear suggested pattern for this, see extensive discussion at python-trio/trio#174.

Attempts to address #6335

Steps Taken to QA Changes

I wrote a recursive subprocess test

from prefect.utilities.processutils import run_process
import anyio
import sys


async def main(count):
    if count <= 0:
        return

    print(f"Running {count} children")

    async with anyio.create_task_group() as tg:
        for _ in range(count):
            await tg.start(
                run_process, [sys.executable, "./process-test.py", str(count - 2)], True
            )

if __name__ == "__main__":
    anyio.run(main, int(sys.argv[1]))

Checklist

This pull request is:

• A documentation / typographical error fix
  • No tests or issue needed
• A short code fix
  • Please reference the related issue by including "closes <link to issue>" in this Pull Request's summary section.
    • If no issue exists, please create a bug report issue
  • Please include tests. One-line fixes without tests will not be accepted unless it's related to the documentation only.
• A new feature implementation
  • Please reference the related issue by including "closes <link to issue>" in this Pull Request's summary section.
    • If no issue exists, please create a feature enhancement issue
  • Please include tests
  • Please make sure that your QA steps are both thorough and easy to reproduce by somebody with limited knowledge of the feature that you are submitting

Happy engineering!

[chrisguidry]

Excellent find!

[peytonrunyan]

lgtm!