Spawning subprocesses on Windows fails

[zanieb]

First check

• I added a descriptive title to this issue.
• I used the GitHub search to find a similar issue and didn't find it.
• I searched the Prefect documentation for this issue.
• I checked that this issue is related to Prefect and not one of its dependencies.

Bug summary

Spawning subproceses fails with a NotImplementedError. We believe this is due to a change in the default event loop policy by Dask as this cannot be reproduced without prefect-dask installed.

For example, this affects:

• Running flow runs with Process infrastructure on agents
• prefect orion start

Reproduction

import prefect_dask

from prefect.infrastructure import Process

Process(command=["echo", "hello world"]).run()

Error

Traceback (most recent call last):
  File "C:\...", line 5, in <module>
    Process(command=["echo", "hello world"]).run()
  File "C:\Python310\lib\site-packages\prefect\utilities\asyncutils.py", line 230, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "C:\Python310\lib\site-packages\prefect\utilities\asyncutils.py", line 181, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "C:\Python310\lib\site-packages\anyio\_core\_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "C:\Python310\lib\site-packages\anyio\_backends\_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "C:\Python310\lib\asyncio\runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "C:\Python310\lib\asyncio\base_events.py", line 646, in run_until_complete
    return future.result()
  File "C:\Python310\lib\site-packages\anyio\_backends\_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "C:\Python310\lib\site-packages\prefect\infrastructure\process.py", line 114, in run
    process = await run_process(
  File "C:\Python310\lib\site-packages\prefect\utilities\processutils.py", line 257, in run_process
    async with open_process(
  File "C:\Python310\lib\contextlib.py", line 199, in __aenter__
    return await anext(self.gen)
  File "C:\Python310\lib\site-packages\prefect\utilities\processutils.py", line 199, in open_process
    process = await _open_anyio_process(command, **kwargs)
  File "C:\Python310\lib\site-packages\prefect\utilities\processutils.py", line 161, in _open_anyio_process
    process = await asyncio.create_subprocess_shell(
  File "C:\Python310\lib\asyncio\subprocess.py", line 205, in create_subprocess_shell
    transport, protocol = await loop.subprocess_shell(
  File "C:\Python310\lib\asyncio\base_events.py", line 1634, in subprocess_shell
    transport = await self._make_subprocess_transport(
  File "C:\Python310\lib\asyncio\base_events.py", line 498, in _make_subprocess_transport
    raise NotImplementedError
NotImplementedError

Versions

All Python versions should be effected

Dask version information may be relevant.

Additional context

No response

[zanieb]

Related: #8111 #8184

[ccueto36]

I can confirm the issue. Ran the Reproduction code and getting the following error:

Traceback (most recent call last):
  File "C:\Users\ccufdet\AppData\Local\Temp\tempCodeRunnerFile.python", line 5, in <module>
    Process(command=["echo", "hello world"]).run()
  File "C:\Python310\lib\site-packages\prefect\utilities\asyncutils.py", line 230, in coroutine_wrapper
    return run_async_in_new_loop(async_fn, *args, **kwargs)
  File "C:\Python310\lib\site-packages\prefect\utilities\asyncutils.py", line 181, in run_async_in_new_loop
    return anyio.run(partial(__fn, *args, **kwargs))
  File "C:\Python310\lib\site-packages\anyio\_core\_eventloop.py", line 70, in run
    return asynclib.run(func, *args, **backend_options)
  File "C:\Python310\lib\site-packages\anyio\_backends\_asyncio.py", line 292, in run
    return native_run(wrapper(), debug=debug)
  File "C:\Python310\lib\asyncio\runners.py", line 44, in run
    return loop.run_until_complete(main)
  File "C:\Python310\lib\asyncio\base_events.py", line 646, in run_until_complete
    return future.result()
  File "C:\Python310\lib\site-packages\anyio\_backends\_asyncio.py", line 287, in wrapper
    return await func(*args)
  File "C:\Python310\lib\site-packages\prefect\infrastructure\process.py", line 114, in run
    process = await run_process(
  File "C:\Python310\lib\site-packages\prefect\utilities\processutils.py", line 257, in run_process
    async with open_process(
  File "C:\Python310\lib\contextlib.py", line 199, in __aenter__
    return await anext(self.gen)
  File "C:\Python310\lib\site-packages\prefect\utilities\processutils.py", line 199, in open_process
    process = await _open_anyio_process(command, **kwargs)
  File "C:\Python310\lib\site-packages\prefect\utilities\processutils.py", line 161, in _open_anyio_process
    process = await asyncio.create_subprocess_shell(
  File "C:\Python310\lib\asyncio\subprocess.py", line 205, in create_subprocess_shell
    transport, protocol = await loop.subprocess_shell(
  File "C:\Python310\lib\asyncio\base_events.py", line 1634, in subprocess_shell
    transport = await self._make_subprocess_transport(
  File "C:\Python310\lib\asyncio\base_events.py", line 498, in _make_subprocess_transport
    raise NotImplementedError
NotImplementedError

[zanieb]

Tracking in Dask at dask/distributed#7492

[zanieb]

We've resolved this when using our CLI by setting the entrypoint again in #8184

The Dask change is blocked by tornadoweb/tornado#3173

[github-actions]

This issue is stale because it has been open 30 days with no activity. To keep this issue open remove stale label or comment.

[github-actions]

This issue was closed because it has been stale for 14 days with no activity. If this issue is important or you have more to add feel free to re-open it.

[zanieb]

@zhen0 fwiw this is still a valid issue and can be tracked by the Dask issue

[zhen0]

Thanks @zanieb 🙏

[zanieb]

Should be resolved by dask/distributed#7969 / dask/distributed#7492