We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Standardize process work pool type name in the UI and CLI (process vs. subprocess)
### First check

- [X] I added a descriptive title to this issue.
- [X] I used the GitHub search to find a similar request and didn't find it.
- [X] I searched the Prefect documentation for this feature.

### Prefect Version

2.x

### Describe the current behavior

Local Subprocess is the term when creating a work pool in the UI
<img width="913" alt="Screenshot 2023-09-27 at 6 12 28 PM" src="https://github.com/PrefectHQ/prefect/assets/7703961/1b0fbba6-90b4-4305-99a2-7e4f833678e4">

After creation in the UI it is referred to as a Process work pool.
<img width="488" alt="Screenshot 2023-09-27 at 6 13 11 PM" src="https://github.com/PrefectHQ/prefect/assets/7703961/0ac0cb1e-eb30-40f6-b34d-2ed757cca051">

<img width="210" alt="Screenshot 2023-09-27 at 6 13 34 PM" src="https://github.com/PrefectHQ/prefect/assets/7703961/657aea2b-c13e-45af-978e-c48c4bf2463b">

The CLI also refers to it as a process work pool.
<img width="446" alt="Screenshot 2023-09-27 at 6 14 37 PM" src="https://github.com/PrefectHQ/prefect/assets/7703961/66fd13ca-ba1b-4540-bf2c-417db40a1ba9">




### Describe the proposed behavior

Standardizing on process or subprocess would be good for users.

### Example Use

_No response_

### Additional context

_No response_

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/prefect/workers/process.py`
Content:
```
1 """
2 Module containing the Process worker used for executing flow runs as subprocesses.
3 
4 To start a Process worker, run the following command:
5 
6 ```bash
7 prefect worker start --pool 'my-work-pool' --type process
8 ```
9 
10 Replace `my-work-pool` with the name of the work pool you want the worker
11 to poll for flow runs.
12 
13 For more information about work pools and workers,
14 checkout out the [Prefect docs](/concepts/work-pools/).
15 """
16 import asyncio
17 import contextlib
18 import os
19 import signal
20 import socket
21 import subprocess
22 import sys
23 import tempfile
24 from pathlib import Path
25 from typing import TYPE_CHECKING, Dict, Optional, Tuple
26 
27 import anyio
28 import anyio.abc
29 import sniffio
30 
31 from prefect._internal.pydantic import HAS_PYDANTIC_V2
32 
33 if HAS_PYDANTIC_V2:
34     from pydantic.v1 import Field, validator
35 else:
36     from pydantic import Field, validator
37 
38 from prefect._internal.schemas.validators import validate_command
39 from prefect.client.schemas import FlowRun
40 from prefect.exceptions import InfrastructureNotAvailable, InfrastructureNotFound
41 from prefect.utilities.processutils import get_sys_executable, run_process
42 from prefect.workers.base import (
43     BaseJobConfiguration,
44     BaseVariables,
45     BaseWorker,
46     BaseWorkerResult,
47 )
48 
49 if TYPE_CHECKING:
50     from prefect.client.schemas.objects import Flow
51     from prefect.client.schemas.responses import DeploymentResponse
52 
53 if sys.platform == "win32":
54     # exit code indicating that the process was terminated by Ctrl+C or Ctrl+Break
55     STATUS_CONTROL_C_EXIT = 0xC000013A
56 
57 
58 def _use_threaded_child_watcher():
59     if (
60         sys.version_info < (3, 8)
61         and sniffio.current_async_library() == "asyncio"
62         and sys.platform != "win32"
63     ):
64         from prefect.utilities.compat import ThreadedChildWatcher
65 
66         # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can
67         # lead to errors in tests on unix as the previous default `SafeChildWatcher`
68         # is not compatible with threaded event loops.
69         asyncio.get_event_loop_policy().set_child_watcher(ThreadedChildWatcher())
70 
71 
72 def _infrastructure_pid_from_process(process: anyio.abc.Process) -> str:
73     hostname = socket.gethostname()
74     return f"{hostname}:{process.pid}"
75 
76 
77 def _parse_infrastructure_pid(infrastructure_pid: str) -> Tuple[str, int]:
78     hostname, pid = infrastructure_pid.split(":")
79     return hostname, int(pid)
80 
81 
82 class ProcessJobConfiguration(BaseJobConfiguration):
83     stream_output: bool = Field(default=True)
84     working_dir: Optional[Path] = Field(default=None)
85 
86     @validator("working_dir")
87     def validate_command(cls, v):
88         return validate_command(v)
89 
90     def prepare_for_flow_run(
91         self,
92         flow_run: "FlowRun",
93         deployment: Optional["DeploymentResponse"] = None,
94         flow: Optional["Flow"] = None,
95     ):
96         super().prepare_for_flow_run(flow_run, deployment, flow)
97 
98         self.env = {**os.environ, **self.env}
99         self.command = (
100             f"{get_sys_executable()} -m prefect.engine"
101             if self.command == self._base_flow_run_command()
102             else self.command
103         )
104 
105     def _base_flow_run_command(self) -> str:
106         """
107         Override the base flow run command because enhanced cancellation doesn't
108         work with the process worker.
109         """
110         return "python -m prefect.engine"
111 
112 
113 class ProcessVariables(BaseVariables):
114     stream_output: bool = Field(
115         default=True,
116         description=(
117             "If enabled, workers will stream output from flow run processes to "
118             "local standard output."
119         ),
120     )
121     working_dir: Optional[Path] = Field(
122         default=None,
123         title="Working Directory",
124         description=(
125             "If provided, workers will open flow run processes within the "
126             "specified path as the working directory. Otherwise, a temporary "
127             "directory will be created."
128         ),
129     )
130 
131 
132 class ProcessWorkerResult(BaseWorkerResult):
133     """Contains information about the final state of a completed process"""
134 
135 
136 class ProcessWorker(BaseWorker):
137     type = "process"
138     job_configuration = ProcessJobConfiguration
139     job_configuration_variables = ProcessVariables
140 
141     _description = (
142         "Execute flow runs as subprocesses on a worker. Works well for local execution"
143         " when first getting started."
144     )
145     _display_name = "Local Subprocess"
146     _documentation_url = (
147         "https://docs.prefect.io/latest/api-ref/prefect/workers/process/"
148     )
149     _logo_url = "https://cdn.sanity.io/images/3ugk85nk/production/356e6766a91baf20e1d08bbe16e8b5aaef4d8643-48x48.png"
150 
151     async def run(
152         self,
153         flow_run: FlowRun,
154         configuration: ProcessJobConfiguration,
155         task_status: Optional[anyio.abc.TaskStatus] = None,
156     ):
157         command = configuration.command
158         if not command:
159             command = f"{get_sys_executable()} -m prefect.engine"
160 
161         flow_run_logger = self.get_flow_run_logger(flow_run)
162 
163         # We must add creationflags to a dict so it is only passed as a function
164         # parameter on Windows, because the presence of creationflags causes
165         # errors on Unix even if set to None
166         kwargs: Dict[str, object] = {}
167         if sys.platform == "win32":
168             kwargs["creationflags"] = subprocess.CREATE_NEW_PROCESS_GROUP
169 
170         _use_threaded_child_watcher()
171         flow_run_logger.info("Opening process...")
172 
173         working_dir_ctx = (
174             tempfile.TemporaryDirectory(suffix="prefect")
175             if not configuration.working_dir
176             else contextlib.nullcontext(configuration.working_dir)
177         )
178         with working_dir_ctx as working_dir:
179             flow_run_logger.debug(
180                 f"Process running command: {command} in {working_dir}"
181             )
182             process = await run_process(
183                 command.split(" "),
184                 stream_output=configuration.stream_output,
185                 task_status=task_status,
186                 task_status_handler=_infrastructure_pid_from_process,
187                 cwd=working_dir,
188                 env=configuration.env,
189                 **kwargs,
190             )
191 
192         # Use the pid for display if no name was given
193         display_name = f" {process.pid}"
194 
195         if process.returncode:
196             help_message = None
197             if process.returncode == -9:
198                 help_message = (
199                     "This indicates that the process exited due to a SIGKILL signal. "
200                     "Typically, this is either caused by manual cancellation or "
201                     "high memory usage causing the operating system to "
202                     "terminate the process."
203                 )
204             if process.returncode == -15:
205                 help_message = (
206                     "This indicates that the process exited due to a SIGTERM signal. "
207                     "Typically, this is caused by manual cancellation."
208                 )
209             elif process.returncode == 247:
210                 help_message = (
211                     "This indicates that the process was terminated due to high "
212                     "memory usage."
213                 )
214             elif (
215                 sys.platform == "win32" and process.returncode == STATUS_CONTROL_C_EXIT
216             ):
217                 help_message = (
218                     "Process was terminated due to a Ctrl+C or Ctrl+Break signal. "
219                     "Typically, this is caused by manual cancellation."
220                 )
221 
222             flow_run_logger.error(
223                 f"Process{display_name} exited with status code: {process.returncode}"
224                 + (f"; {help_message}" if help_message else "")
225             )
226         else:
227             flow_run_logger.info(f"Process{display_name} exited cleanly.")
228 
229         return ProcessWorkerResult(
230             status_code=process.returncode, identifier=str(process.pid)
231         )
232 
233     async def kill_infrastructure(
234         self,
235         infrastructure_pid: str,
236         configuration: ProcessJobConfiguration,
237         grace_seconds: int = 30,
238     ):
239         hostname, pid = _parse_infrastructure_pid(infrastructure_pid)
240 
241         if hostname != socket.gethostname():
242             raise InfrastructureNotAvailable(
243                 f"Unable to kill process {pid!r}: The process is running on a different"
244                 f" host {hostname!r}."
245             )
246 
247         # In a non-windows environment first send a SIGTERM, then, after
248         # `grace_seconds` seconds have passed subsequent send SIGKILL. In
249         # Windows we use CTRL_BREAK_EVENT as SIGTERM is useless:
250         # https://bugs.python.org/issue26350
251         if sys.platform == "win32":
252             try:
253                 os.kill(pid, signal.CTRL_BREAK_EVENT)
254             except (ProcessLookupError, WindowsError):
255                 raise InfrastructureNotFound(
256                     f"Unable to kill process {pid!r}: The process was not found."
257                 )
258         else:
259             try:
260                 os.kill(pid, signal.SIGTERM)
261             except ProcessLookupError:
262                 raise InfrastructureNotFound(
263                     f"Unable to kill process {pid!r}: The process was not found."
264                 )
265 
266             # Throttle how often we check if the process is still alive to keep
267             # from making too many system calls in a short period of time.
268             check_interval = max(grace_seconds / 10, 1)
269 
270             with anyio.move_on_after(grace_seconds):
271                 while True:
272                     await anyio.sleep(check_interval)
273 
274                     # Detect if the process is still alive. If not do an early
275                     # return as the process respected the SIGTERM from above.
276                     try:
277                         os.kill(pid, 0)
278                     except ProcessLookupError:
279                         return
280 
281             try:
282                 os.kill(pid, signal.SIGKILL)
283             except OSError:
284                 # We shouldn't ever end up here, but it's possible that the
285                 # process ended right after the check above.
286                 return
287
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/prefect/workers/process.py b/src/prefect/workers/process.py
--- a/src/prefect/workers/process.py
+++ b/src/prefect/workers/process.py
@@ -13,6 +13,7 @@
 For more information about work pools and workers,
 checkout out the [Prefect docs](/concepts/work-pools/).
 """
+
 import asyncio
 import contextlib
 import os
@@ -142,7 +143,7 @@
         "Execute flow runs as subprocesses on a worker. Works well for local execution"
         " when first getting started."
     )
-    _display_name = "Local Subprocess"
+    _display_name = "Process"
     _documentation_url = (
         "https://docs.prefect.io/latest/api-ref/prefect/workers/process/"
     )

{"golden_diff": "diff --git a/src/prefect/workers/process.py b/src/prefect/workers/process.py\n--- a/src/prefect/workers/process.py\n+++ b/src/prefect/workers/process.py\n@@ -13,6 +13,7 @@\n For more information about work pools and workers,\n checkout out the [Prefect docs](/concepts/work-pools/).\n \"\"\"\n+\n import asyncio\n import contextlib\n import os\n@@ -142,7 +143,7 @@\n         \"Execute flow runs as subprocesses on a worker. Works well for local execution\"\n         \" when first getting started.\"\n     )\n-    _display_name = \"Local Subprocess\"\n+    _display_name = \"Process\"\n     _documentation_url = (\n         \"https://docs.prefect.io/latest/api-ref/prefect/workers/process/\"\n     )\n", "issue": "Standardize process work pool type name in the UI and CLI (process vs. subprocess)\n### First check\n\n- [X] I added a descriptive title to this issue.\n- [X] I used the GitHub search to find a similar request and didn't find it.\n- [X] I searched the Prefect documentation for this feature.\n\n### Prefect Version\n\n2.x\n\n### Describe the current behavior\n\nLocal Subprocess is the term when creating a work pool in the UI\r\n<img width=\"913\" alt=\"Screenshot 2023-09-27 at 6 12 28 PM\" src=\"https://github.com/PrefectHQ/prefect/assets/7703961/1b0fbba6-90b4-4305-99a2-7e4f833678e4\">\r\n\r\nAfter creation in the UI it is referred to as a Process work pool.\r\n<img width=\"488\" alt=\"Screenshot 2023-09-27 at 6 13 11 PM\" src=\"https://github.com/PrefectHQ/prefect/assets/7703961/0ac0cb1e-eb30-40f6-b34d-2ed757cca051\">\r\n\r\n<img width=\"210\" alt=\"Screenshot 2023-09-27 at 6 13 34 PM\" src=\"https://github.com/PrefectHQ/prefect/assets/7703961/657aea2b-c13e-45af-978e-c48c4bf2463b\">\r\n\r\nThe CLI also refers to it as a process work pool.\r\n<img width=\"446\" alt=\"Screenshot 2023-09-27 at 6 14 37 PM\" src=\"https://github.com/PrefectHQ/prefect/assets/7703961/66fd13ca-ba1b-4540-bf2c-417db40a1ba9\">\r\n\r\n\r\n\n\n### Describe the proposed behavior\n\nStandardizing on process or subprocess would be good for users.\n\n### Example Use\n\n_No response_\n\n### Additional context\n\n_No response_\n", "before_files": [{"content": "\"\"\"\nModule containing the Process worker used for executing flow runs as subprocesses.\n\nTo start a Process worker, run the following command:\n\n```bash\nprefect worker start --pool 'my-work-pool' --type process\n```\n\nReplace `my-work-pool` with the name of the work pool you want the worker\nto poll for flow runs.\n\nFor more information about work pools and workers,\ncheckout out the [Prefect docs](/concepts/work-pools/).\n\"\"\"\nimport asyncio\nimport contextlib\nimport os\nimport signal\nimport socket\nimport subprocess\nimport sys\nimport tempfile\nfrom pathlib import Path\nfrom typing import TYPE_CHECKING, Dict, Optional, Tuple\n\nimport anyio\nimport anyio.abc\nimport sniffio\n\nfrom prefect._internal.pydantic import HAS_PYDANTIC_V2\n\nif HAS_PYDANTIC_V2:\n    from pydantic.v1 import Field, validator\nelse:\n    from pydantic import Field, validator\n\nfrom prefect._internal.schemas.validators import validate_command\nfrom prefect.client.schemas import FlowRun\nfrom prefect.exceptions import InfrastructureNotAvailable, InfrastructureNotFound\nfrom prefect.utilities.processutils import get_sys_executable, run_process\nfrom prefect.workers.base import (\n    BaseJobConfiguration,\n    BaseVariables,\n    BaseWorker,\n    BaseWorkerResult,\n)\n\nif TYPE_CHECKING:\n    from prefect.client.schemas.objects import Flow\n    from prefect.client.schemas.responses import DeploymentResponse\n\nif sys.platform == \"win32\":\n    # exit code indicating that the process was terminated by Ctrl+C or Ctrl+Break\n    STATUS_CONTROL_C_EXIT = 0xC000013A\n\n\ndef _use_threaded_child_watcher():\n    if (\n        sys.version_info < (3, 8)\n        and sniffio.current_async_library() == \"asyncio\"\n        and sys.platform != \"win32\"\n    ):\n        from prefect.utilities.compat import ThreadedChildWatcher\n\n        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can\n        # lead to errors in tests on unix as the previous default `SafeChildWatcher`\n        # is not compatible with threaded event loops.\n        asyncio.get_event_loop_policy().set_child_watcher(ThreadedChildWatcher())\n\n\ndef _infrastructure_pid_from_process(process: anyio.abc.Process) -> str:\n    hostname = socket.gethostname()\n    return f\"{hostname}:{process.pid}\"\n\n\ndef _parse_infrastructure_pid(infrastructure_pid: str) -> Tuple[str, int]:\n    hostname, pid = infrastructure_pid.split(\":\")\n    return hostname, int(pid)\n\n\nclass ProcessJobConfiguration(BaseJobConfiguration):\n    stream_output: bool = Field(default=True)\n    working_dir: Optional[Path] = Field(default=None)\n\n    @validator(\"working_dir\")\n    def validate_command(cls, v):\n        return validate_command(v)\n\n    def prepare_for_flow_run(\n        self,\n        flow_run: \"FlowRun\",\n        deployment: Optional[\"DeploymentResponse\"] = None,\n        flow: Optional[\"Flow\"] = None,\n    ):\n        super().prepare_for_flow_run(flow_run, deployment, flow)\n\n        self.env = {**os.environ, **self.env}\n        self.command = (\n            f\"{get_sys_executable()} -m prefect.engine\"\n            if self.command == self._base_flow_run_command()\n            else self.command\n        )\n\n    def _base_flow_run_command(self) -> str:\n        \"\"\"\n        Override the base flow run command because enhanced cancellation doesn't\n        work with the process worker.\n        \"\"\"\n        return \"python -m prefect.engine\"\n\n\nclass ProcessVariables(BaseVariables):\n    stream_output: bool = Field(\n        default=True,\n        description=(\n            \"If enabled, workers will stream output from flow run processes to \"\n            \"local standard output.\"\n        ),\n    )\n    working_dir: Optional[Path] = Field(\n        default=None,\n        title=\"Working Directory\",\n        description=(\n            \"If provided, workers will open flow run processes within the \"\n            \"specified path as the working directory. Otherwise, a temporary \"\n            \"directory will be created.\"\n        ),\n    )\n\n\nclass ProcessWorkerResult(BaseWorkerResult):\n    \"\"\"Contains information about the final state of a completed process\"\"\"\n\n\nclass ProcessWorker(BaseWorker):\n    type = \"process\"\n    job_configuration = ProcessJobConfiguration\n    job_configuration_variables = ProcessVariables\n\n    _description = (\n        \"Execute flow runs as subprocesses on a worker. Works well for local execution\"\n        \" when first getting started.\"\n    )\n    _display_name = \"Local Subprocess\"\n    _documentation_url = (\n        \"https://docs.prefect.io/latest/api-ref/prefect/workers/process/\"\n    )\n    _logo_url = \"https://cdn.sanity.io/images/3ugk85nk/production/356e6766a91baf20e1d08bbe16e8b5aaef4d8643-48x48.png\"\n\n    async def run(\n        self,\n        flow_run: FlowRun,\n        configuration: ProcessJobConfiguration,\n        task_status: Optional[anyio.abc.TaskStatus] = None,\n    ):\n        command = configuration.command\n        if not command:\n            command = f\"{get_sys_executable()} -m prefect.engine\"\n\n        flow_run_logger = self.get_flow_run_logger(flow_run)\n\n        # We must add creationflags to a dict so it is only passed as a function\n        # parameter on Windows, because the presence of creationflags causes\n        # errors on Unix even if set to None\n        kwargs: Dict[str, object] = {}\n        if sys.platform == \"win32\":\n            kwargs[\"creationflags\"] = subprocess.CREATE_NEW_PROCESS_GROUP\n\n        _use_threaded_child_watcher()\n        flow_run_logger.info(\"Opening process...\")\n\n        working_dir_ctx = (\n            tempfile.TemporaryDirectory(suffix=\"prefect\")\n            if not configuration.working_dir\n            else contextlib.nullcontext(configuration.working_dir)\n        )\n        with working_dir_ctx as working_dir:\n            flow_run_logger.debug(\n                f\"Process running command: {command} in {working_dir}\"\n            )\n            process = await run_process(\n                command.split(\" \"),\n                stream_output=configuration.stream_output,\n                task_status=task_status,\n                task_status_handler=_infrastructure_pid_from_process,\n                cwd=working_dir,\n                env=configuration.env,\n                **kwargs,\n            )\n\n        # Use the pid for display if no name was given\n        display_name = f\" {process.pid}\"\n\n        if process.returncode:\n            help_message = None\n            if process.returncode == -9:\n                help_message = (\n                    \"This indicates that the process exited due to a SIGKILL signal. \"\n                    \"Typically, this is either caused by manual cancellation or \"\n                    \"high memory usage causing the operating system to \"\n                    \"terminate the process.\"\n                )\n            if process.returncode == -15:\n                help_message = (\n                    \"This indicates that the process exited due to a SIGTERM signal. \"\n                    \"Typically, this is caused by manual cancellation.\"\n                )\n            elif process.returncode == 247:\n                help_message = (\n                    \"This indicates that the process was terminated due to high \"\n                    \"memory usage.\"\n                )\n            elif (\n                sys.platform == \"win32\" and process.returncode == STATUS_CONTROL_C_EXIT\n            ):\n                help_message = (\n                    \"Process was terminated due to a Ctrl+C or Ctrl+Break signal. \"\n                    \"Typically, this is caused by manual cancellation.\"\n                )\n\n            flow_run_logger.error(\n                f\"Process{display_name} exited with status code: {process.returncode}\"\n                + (f\"; {help_message}\" if help_message else \"\")\n            )\n        else:\n            flow_run_logger.info(f\"Process{display_name} exited cleanly.\")\n\n        return ProcessWorkerResult(\n            status_code=process.returncode, identifier=str(process.pid)\n        )\n\n    async def kill_infrastructure(\n        self,\n        infrastructure_pid: str,\n        configuration: ProcessJobConfiguration,\n        grace_seconds: int = 30,\n    ):\n        hostname, pid = _parse_infrastructure_pid(infrastructure_pid)\n\n        if hostname != socket.gethostname():\n            raise InfrastructureNotAvailable(\n                f\"Unable to kill process {pid!r}: The process is running on a different\"\n                f\" host {hostname!r}.\"\n            )\n\n        # In a non-windows environment first send a SIGTERM, then, after\n        # `grace_seconds` seconds have passed subsequent send SIGKILL. In\n        # Windows we use CTRL_BREAK_EVENT as SIGTERM is useless:\n        # https://bugs.python.org/issue26350\n        if sys.platform == \"win32\":\n            try:\n                os.kill(pid, signal.CTRL_BREAK_EVENT)\n            except (ProcessLookupError, WindowsError):\n                raise InfrastructureNotFound(\n                    f\"Unable to kill process {pid!r}: The process was not found.\"\n                )\n        else:\n            try:\n                os.kill(pid, signal.SIGTERM)\n            except ProcessLookupError:\n                raise InfrastructureNotFound(\n                    f\"Unable to kill process {pid!r}: The process was not found.\"\n                )\n\n            # Throttle how often we check if the process is still alive to keep\n            # from making too many system calls in a short period of time.\n            check_interval = max(grace_seconds / 10, 1)\n\n            with anyio.move_on_after(grace_seconds):\n                while True:\n                    await anyio.sleep(check_interval)\n\n                    # Detect if the process is still alive. If not do an early\n                    # return as the process respected the SIGTERM from above.\n                    try:\n                        os.kill(pid, 0)\n                    except ProcessLookupError:\n                        return\n\n            try:\n                os.kill(pid, signal.SIGKILL)\n            except OSError:\n                # We shouldn't ever end up here, but it's possible that the\n                # process ended right after the check above.\n                return\n", "path": "src/prefect/workers/process.py"}], "after_files": [{"content": "\"\"\"\nModule containing the Process worker used for executing flow runs as subprocesses.\n\nTo start a Process worker, run the following command:\n\n```bash\nprefect worker start --pool 'my-work-pool' --type process\n```\n\nReplace `my-work-pool` with the name of the work pool you want the worker\nto poll for flow runs.\n\nFor more information about work pools and workers,\ncheckout out the [Prefect docs](/concepts/work-pools/).\n\"\"\"\n\nimport asyncio\nimport contextlib\nimport os\nimport signal\nimport socket\nimport subprocess\nimport sys\nimport tempfile\nfrom pathlib import Path\nfrom typing import TYPE_CHECKING, Dict, Optional, Tuple\n\nimport anyio\nimport anyio.abc\nimport sniffio\n\nfrom prefect._internal.pydantic import HAS_PYDANTIC_V2\n\nif HAS_PYDANTIC_V2:\n    from pydantic.v1 import Field, validator\nelse:\n    from pydantic import Field, validator\n\nfrom prefect._internal.schemas.validators import validate_command\nfrom prefect.client.schemas import FlowRun\nfrom prefect.exceptions import InfrastructureNotAvailable, InfrastructureNotFound\nfrom prefect.utilities.processutils import get_sys_executable, run_process\nfrom prefect.workers.base import (\n    BaseJobConfiguration,\n    BaseVariables,\n    BaseWorker,\n    BaseWorkerResult,\n)\n\nif TYPE_CHECKING:\n    from prefect.client.schemas.objects import Flow\n    from prefect.client.schemas.responses import DeploymentResponse\n\nif sys.platform == \"win32\":\n    # exit code indicating that the process was terminated by Ctrl+C or Ctrl+Break\n    STATUS_CONTROL_C_EXIT = 0xC000013A\n\n\ndef _use_threaded_child_watcher():\n    if (\n        sys.version_info < (3, 8)\n        and sniffio.current_async_library() == \"asyncio\"\n        and sys.platform != \"win32\"\n    ):\n        from prefect.utilities.compat import ThreadedChildWatcher\n\n        # Python < 3.8 does not use a `ThreadedChildWatcher` by default which can\n        # lead to errors in tests on unix as the previous default `SafeChildWatcher`\n        # is not compatible with threaded event loops.\n        asyncio.get_event_loop_policy().set_child_watcher(ThreadedChildWatcher())\n\n\ndef _infrastructure_pid_from_process(process: anyio.abc.Process) -> str:\n    hostname = socket.gethostname()\n    return f\"{hostname}:{process.pid}\"\n\n\ndef _parse_infrastructure_pid(infrastructure_pid: str) -> Tuple[str, int]:\n    hostname, pid = infrastructure_pid.split(\":\")\n    return hostname, int(pid)\n\n\nclass ProcessJobConfiguration(BaseJobConfiguration):\n    stream_output: bool = Field(default=True)\n    working_dir: Optional[Path] = Field(default=None)\n\n    @validator(\"working_dir\")\n    def validate_command(cls, v):\n        return validate_command(v)\n\n    def prepare_for_flow_run(\n        self,\n        flow_run: \"FlowRun\",\n        deployment: Optional[\"DeploymentResponse\"] = None,\n        flow: Optional[\"Flow\"] = None,\n    ):\n        super().prepare_for_flow_run(flow_run, deployment, flow)\n\n        self.env = {**os.environ, **self.env}\n        self.command = (\n            f\"{get_sys_executable()} -m prefect.engine\"\n            if self.command == self._base_flow_run_command()\n            else self.command\n        )\n\n    def _base_flow_run_command(self) -> str:\n        \"\"\"\n        Override the base flow run command because enhanced cancellation doesn't\n        work with the process worker.\n        \"\"\"\n        return \"python -m prefect.engine\"\n\n\nclass ProcessVariables(BaseVariables):\n    stream_output: bool = Field(\n        default=True,\n        description=(\n            \"If enabled, workers will stream output from flow run processes to \"\n            \"local standard output.\"\n        ),\n    )\n    working_dir: Optional[Path] = Field(\n        default=None,\n        title=\"Working Directory\",\n        description=(\n            \"If provided, workers will open flow run processes within the \"\n            \"specified path as the working directory. Otherwise, a temporary \"\n            \"directory will be created.\"\n        ),\n    )\n\n\nclass ProcessWorkerResult(BaseWorkerResult):\n    \"\"\"Contains information about the final state of a completed process\"\"\"\n\n\nclass ProcessWorker(BaseWorker):\n    type = \"process\"\n    job_configuration = ProcessJobConfiguration\n    job_configuration_variables = ProcessVariables\n\n    _description = (\n        \"Execute flow runs as subprocesses on a worker. Works well for local execution\"\n        \" when first getting started.\"\n    )\n    _display_name = \"Process\"\n    _documentation_url = (\n        \"https://docs.prefect.io/latest/api-ref/prefect/workers/process/\"\n    )\n    _logo_url = \"https://cdn.sanity.io/images/3ugk85nk/production/356e6766a91baf20e1d08bbe16e8b5aaef4d8643-48x48.png\"\n\n    async def run(\n        self,\n        flow_run: FlowRun,\n        configuration: ProcessJobConfiguration,\n        task_status: Optional[anyio.abc.TaskStatus] = None,\n    ):\n        command = configuration.command\n        if not command:\n            command = f\"{get_sys_executable()} -m prefect.engine\"\n\n        flow_run_logger = self.get_flow_run_logger(flow_run)\n\n        # We must add creationflags to a dict so it is only passed as a function\n        # parameter on Windows, because the presence of creationflags causes\n        # errors on Unix even if set to None\n        kwargs: Dict[str, object] = {}\n        if sys.platform == \"win32\":\n            kwargs[\"creationflags\"] = subprocess.CREATE_NEW_PROCESS_GROUP\n\n        _use_threaded_child_watcher()\n        flow_run_logger.info(\"Opening process...\")\n\n        working_dir_ctx = (\n            tempfile.TemporaryDirectory(suffix=\"prefect\")\n            if not configuration.working_dir\n            else contextlib.nullcontext(configuration.working_dir)\n        )\n        with working_dir_ctx as working_dir:\n            flow_run_logger.debug(\n                f\"Process running command: {command} in {working_dir}\"\n            )\n            process = await run_process(\n                command.split(\" \"),\n                stream_output=configuration.stream_output,\n                task_status=task_status,\n                task_status_handler=_infrastructure_pid_from_process,\n                cwd=working_dir,\n                env=configuration.env,\n                **kwargs,\n            )\n\n        # Use the pid for display if no name was given\n        display_name = f\" {process.pid}\"\n\n        if process.returncode:\n            help_message = None\n            if process.returncode == -9:\n                help_message = (\n                    \"This indicates that the process exited due to a SIGKILL signal. \"\n                    \"Typically, this is either caused by manual cancellation or \"\n                    \"high memory usage causing the operating system to \"\n                    \"terminate the process.\"\n                )\n            if process.returncode == -15:\n                help_message = (\n                    \"This indicates that the process exited due to a SIGTERM signal. \"\n                    \"Typically, this is caused by manual cancellation.\"\n                )\n            elif process.returncode == 247:\n                help_message = (\n                    \"This indicates that the process was terminated due to high \"\n                    \"memory usage.\"\n                )\n            elif (\n                sys.platform == \"win32\" and process.returncode == STATUS_CONTROL_C_EXIT\n            ):\n                help_message = (\n                    \"Process was terminated due to a Ctrl+C or Ctrl+Break signal. \"\n                    \"Typically, this is caused by manual cancellation.\"\n                )\n\n            flow_run_logger.error(\n                f\"Process{display_name} exited with status code: {process.returncode}\"\n                + (f\"; {help_message}\" if help_message else \"\")\n            )\n        else:\n            flow_run_logger.info(f\"Process{display_name} exited cleanly.\")\n\n        return ProcessWorkerResult(\n            status_code=process.returncode, identifier=str(process.pid)\n        )\n\n    async def kill_infrastructure(\n        self,\n        infrastructure_pid: str,\n        configuration: ProcessJobConfiguration,\n        grace_seconds: int = 30,\n    ):\n        hostname, pid = _parse_infrastructure_pid(infrastructure_pid)\n\n        if hostname != socket.gethostname():\n            raise InfrastructureNotAvailable(\n                f\"Unable to kill process {pid!r}: The process is running on a different\"\n                f\" host {hostname!r}.\"\n            )\n\n        # In a non-windows environment first send a SIGTERM, then, after\n        # `grace_seconds` seconds have passed subsequent send SIGKILL. In\n        # Windows we use CTRL_BREAK_EVENT as SIGTERM is useless:\n        # https://bugs.python.org/issue26350\n        if sys.platform == \"win32\":\n            try:\n                os.kill(pid, signal.CTRL_BREAK_EVENT)\n            except (ProcessLookupError, WindowsError):\n                raise InfrastructureNotFound(\n                    f\"Unable to kill process {pid!r}: The process was not found.\"\n                )\n        else:\n            try:\n                os.kill(pid, signal.SIGTERM)\n            except ProcessLookupError:\n                raise InfrastructureNotFound(\n                    f\"Unable to kill process {pid!r}: The process was not found.\"\n                )\n\n            # Throttle how often we check if the process is still alive to keep\n            # from making too many system calls in a short period of time.\n            check_interval = max(grace_seconds / 10, 1)\n\n            with anyio.move_on_after(grace_seconds):\n                while True:\n                    await anyio.sleep(check_interval)\n\n                    # Detect if the process is still alive. If not do an early\n                    # return as the process respected the SIGTERM from above.\n                    try:\n                        os.kill(pid, 0)\n                    except ProcessLookupError:\n                        return\n\n            try:\n                os.kill(pid, signal.SIGKILL)\n            except OSError:\n                # We shouldn't ever end up here, but it's possible that the\n                # process ended right after the check above.\n                return\n", "path": "src/prefect/workers/process.py"}]}