Repository: sanic-org/sanic
Pull request #2775 (2023-07-09 10:56:56)

Start and restart arbitrary processes
The idea is to provide a better API to allow restarting of arbitrary processes. Also, it is meant to provide a `manage` method to the multiplexer so that endpoints can also start arbitrary workers.

```py
def task(n: int = 10, **kwargs):
    print("TASK STARTED", kwargs)
    for i in range(n):
        print(f"Running task - Step {i+1} of {n}")
        sleep(1)

@app.get("/restart")
async def restart_handler(request: Request):
    request.app.m.restart("Sanic-TEST-0")
    return json({"foo": request.app.m.name})


@app.get("/start")
async def start_handler(request: Request):
    request.app.m.manage("NEW", task, kwargs={"n": 7}, workers=2)
    return json({"foo": request.app.m.name})

@app.main_process_ready
def start_process(app: Sanic):
    app.manager.manage("TEST", task, kwargs={"n": 3}, restartable=True)
```

TODO:
- [x] Cleanup the API (passing messages is a bit messy)
- [x] Finalize the API (`restartable`, `tracked`, etc)
- [ ] ~Possibly add a worker pool option?~
	- Future enhancement. Not this go-around.

ahopkins: Ready for review, I'm still working on tests, but I want to get codecov to run to see what it thinks I'm missing for coverage.
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) Report
Attention: `68 lines` in your changes are missing coverage. Please review.
> Comparison is base [(`78c44ed`)](https://app.codecov.io/gh/sanic-org/sanic/commit/78c44ed5e4947833df3f5a73fa966881bcdebd9f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) 88.254% compared to head [(`dce2131`)](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) 87.434%.

| [Files](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) | Patch % | Lines |
|---|---|---|
| [sanic/worker/manager.py](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org#diff-c2FuaWMvd29ya2VyL21hbmFnZXIucHk=) | 51.515% | [37 Missing and 11 partials :warning: ](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) |
| [sanic/worker/restarter.py](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org#diff-c2FuaWMvd29ya2VyL3Jlc3RhcnRlci5weQ==) | 57.575% | [11 Missing and 3 partials :warning: ](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) |
| [sanic/worker/process.py](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org#diff-c2FuaWMvd29ya2VyL3Byb2Nlc3MucHk=) | 80.000% | [3 Missing and 1 partial :warning: ](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) |
| [sanic/worker/multiplexer.py](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org#diff-c2FuaWMvd29ya2VyL211bHRpcGxleGVyLnB5) | 50.000% | [2 Missing :warning: ](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org) |

<details><summary>Additional details and impacted files</summary>


```diff
@@              Coverage Diff              @@
##              main     #2775       +/-   ##
=============================================
- Coverage   88.254%   87.434%   -0.821%     
=============================================
  Files           93        94        +1     
  Lines         7305      7433      +128     
  Branches      1254      1283       +29     
=============================================
+ Hits          6447      6499       +52     
- Misses         599       658       +59     
- Partials       259       276       +17     
```



</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/sanic-org/sanic/pull/2775?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=sanic-org).


Patch:
diff --git a/guide/content/en/guide/running/manager.md b/guide/content/en/guide/running/manager.md
index 600a3fe0..c021170c 100644
--- a/guide/content/en/guide/running/manager.md
+++ b/guide/content/en/guide/running/manager.md
@@ -239,6 +239,18 @@ The application instance has access to an object that provides access to interac
     }
     ```
 
+The possible states are:
+
+- `NONE` - The worker has been created, but there is no process yet
+- `IDLE` - The process has been created, but is not running yet
+- `STARTING` - The process is starting
+- `STARTED` - The process has started
+- `ACKED` - The process has started and sent an acknowledgement (usually only for server processes)
+- `JOINED` - The process has exited and joined the main process
+- `TERMINATED` - The process has exited and terminated
+- `RESTARTING` - The process is restarting
+- `FAILED` - The process encountered an exception and is no longer running
+- `COMPLETED` - The process has completed its work and exited successfully
 
 ## Built-in non-server processes
 
@@ -324,6 +336,122 @@ To run a managed custom process on Sanic, you must create a callable. If that pr
         app.manager.manage("MyProcess", my_process, {"foo": "bar"})
     ```
 
+### Transient v. durable processes
+
+.. column::
+
+    When you manage a process with the `manage` method, you have the option to make it transient or durable. A transient process will be restarted by the auto-reloader, and a durable process will not.
+    
+    By default, all processes are durable.
+    
+.. column::
+
+    ```python
+    @app.main_process_ready
+    async def ready(app: Sanic, _):
+        app.manager.manage(
+            "MyProcess",
+            my_process,
+            {"foo": "bar"},
+            transient=True,
+        )
+    ```
+
+
+### Tracked v. untracked processes
+
+.. new:: v23.12
+
+    Out of the box, Sanic will track the state of all processes. This means that you can access the state of the process from the [multiplexer](./manager#access-to-the-multiplexer) object, or from the [Inspector](./manager#inspector).
+    
+    See [worker state](./manager#worker-state) for more information.
+    
+    Sometimes it is helpful to run background processes that are not long-running. You run them once until completion and then they exit. Upon completion, they will either be in `FAILED` or `COMPLETED` state.
+    
+
+.. column::
+
+    When you are running a non-long-running process, you can opt out of tracking it by setting `tracked=False` in the `manage` method. This means that upon completion of the process it will be removed from the list of tracked processes. You will only be able to check the state of the process while it is running.
+    
+.. column::
+
+    ```python
+    @app.main_process_ready
+    async def ready(app: Sanic, _):
+        app.manager.manage(
+            "OneAndDone",
+            do_once,
+            {},
+            tracked=False,
+        )
+    ```
+
+*Added in v23.12*
+
+### Restartable custom processes
+
+.. new:: v23.12
+
+    A custom process that is transient will **always** be restartable. That means the auto-restart will work as expected. However, what if you want to be able to *manually* restart a process, but not have it be restarted by the auto-reloader?
+    
+.. column::
+
+    In this scenario, you can set `restartable=True` in the `manage` method. This will allow you to manually restart the process, but it will not be restarted by the auto-reloader.
+    
+.. column::
+
+    ```python
+    @app.main_process_ready
+    async def ready(app: Sanic, _):
+        app.manager.manage(
+            "MyProcess",
+            my_process,
+            {"foo": "bar"},
+            restartable=True,
+        )
+    ```
+    
+.. column::
+
+    You could now manually restart that process from the multiplexer.
+    
+.. column::
+
+    ```python
+    @app.get("/restart")
+    async def restart_handler(request: Request):
+        request.app.m.restart("Sanic-MyProcess-0")
+        return json({"foo": request.app.m.name})
+    ```
+    
+*Added in v23.12*
+
+### On the fly process management
+
+.. new:: v23.12
+
+    Custom processes are usually added in the `main_process_ready` listener. However, there may be times when you want to add a process after the application has started. For example, you may want to add a process from a request handler. The multiplexer provides a method for doing this.
+    
+.. column::
+
+    Once you have a reference to the multiplexer, you can call `manage` to add a process. It works the same as the `manage` method on the Manager.
+    
+.. column::
+
+    ```python
+    @app.post("/start")
+    async def start_handler(request: Request):
+        request.app.m.manage(
+            "MyProcess",
+            my_process,
+            {"foo": "bar"},
+            workers=2,
+        )
+        return json({"foo": request.app.m.name})
+    ```
+    
+*Added in v23.12*
+
 ## Single process mode
 
 .. column::
diff --git a/sanic/app.py b/sanic/app.py
index 17fc925e..0a93e523 100644
--- a/sanic/app.py
+++ b/sanic/app.py
@@ -2510,7 +2510,10 @@ class Sanic(
         """
         if environ.get("SANIC_WORKER_PROCESS") or not self._inspector:
             raise SanicException(
-                "Can only access the inspector from the main process"
+                "Can only access the inspector from the main process "
+                "after main_process_start has run. For example, you most "
+                "likely want to use it inside the @app.main_process_ready "
+                "event listener."
             )
         return self._inspector
 
@@ -2545,6 +2548,9 @@ class Sanic(
 
         if environ.get("SANIC_WORKER_PROCESS") or not self._manager:
             raise SanicException(
-                "Can only access the manager from the main process"
+                "Can only access the manager from the main process "
+                "after main_process_start has run. For example, you most "
+                "likely want to use it inside the @app.main_process_ready "
+                "event listener."
             )
         return self._manager
diff --git a/sanic/mixins/startup.py b/sanic/mixins/startup.py
index 5eb42eca..3f49161a 100644
--- a/sanic/mixins/startup.py
+++ b/sanic/mixins/startup.py
@@ -1166,7 +1166,9 @@ class StartupMixin(metaclass=SanicMeta):
                     ...
                 sock.close()
             socks = []
+
             trigger_events(main_stop, loop, primary)
+
             loop.close()
             cls._cleanup_env_vars()
             cls._cleanup_apps()
@@ -1192,7 +1194,12 @@ class StartupMixin(metaclass=SanicMeta):
     @staticmethod
     def _get_process_states(worker_state) -> List[str]:
         return [
-            state for s in worker_state.values() if (state := s.get("state"))
+            state
+            for s in worker_state.values()
+            if (
+                (state := s.get("state"))
+                and state not in ("TERMINATED", "FAILED", "COMPLETED", "NONE")
+            )
         ]
 
     @classmethod
diff --git a/sanic/worker/constants.py b/sanic/worker/constants.py
index 9a8795c6..b7a7f35d 100644
--- a/sanic/worker/constants.py
+++ b/sanic/worker/constants.py
@@ -13,6 +13,7 @@ class RestartOrder(UpperStrEnum):
 class ProcessState(IntEnum):
     """Process states."""
 
+    NONE = auto()
     IDLE = auto()
     RESTARTING = auto()
     STARTING = auto()
@@ -20,3 +21,5 @@ class ProcessState(IntEnum):
     ACKED = auto()
     JOINED = auto()
     TERMINATED = auto()
+    FAILED = auto()
+    COMPLETED = auto()
diff --git a/sanic/worker/manager.py b/sanic/worker/manager.py
index c3f9dd04..89575ba2 100644
--- a/sanic/worker/manager.py
+++ b/sanic/worker/manager.py
@@ -1,17 +1,19 @@
 import os
 
 from contextlib import suppress
-from itertools import count
+from enum import IntEnum, auto
+from itertools import chain, count
 from random import choice
 from signal import SIGINT, SIGTERM, Signals
 from signal import signal as signal_func
-from typing import Any, Callable, Dict, List, Optional
+from typing import Any, Callable, Dict, Iterable, List, Optional, Tuple
 
 from sanic.compat import OS_IS_WINDOWS
 from sanic.exceptions import ServerKilled
 from sanic.log import error_logger, logger
 from sanic.worker.constants import RestartOrder
 from sanic.worker.process import ProcessState, Worker, WorkerProcess
+from sanic.worker.restarter import Restarter
 
 
 if not OS_IS_WINDOWS:
@@ -20,6 +22,11 @@ else:
     SIGKILL = SIGINT
 
 
+class MonitorCycle(IntEnum):
+    BREAK = auto()
+    CONTINUE = auto()
+
+
 class WorkerManager:
     """Manage all of the processes.
 
@@ -61,6 +68,7 @@ class WorkerManager:
         self.context = context
         self.transient: Dict[str, Worker] = {}
         self.durable: Dict[str, Worker] = {}
+        self.restarter = Restarter()
         self.monitor_publisher, self.monitor_subscriber = monitor_pubsub
         self.worker_state = worker_state
         self.worker_state[self.MAIN_IDENT] = {"pid": self.pid}
@@ -84,6 +92,9 @@ class WorkerManager:
         func: Callable[..., Any],
         kwargs: Dict[str, Any],
         transient: bool = False,
+        restartable: Optional[bool] = None,
+        tracked: bool = True,
+        auto_start: bool = True,
         workers: int = 1,
     ) -> Worker:
         """Instruct Sanic to manage a custom process.
@@ -95,14 +106,37 @@ class WorkerManager:
             transient (bool, optional): Whether to mark the process as transient. If `True`
                 then the Worker Manager will restart the process along
                 with any global restart (ex: auto-reload), defaults to `False`
+            restartable (Optional[bool], optional): Whether to mark the process as restartable. If
+                `True` then the Worker Manager will be able to restart the process
+                if prompted. If `transient=True`, this property will be implied
+                to be `True`, defaults to `None`
+            tracked (bool, optional): Whether to track the process after completion,
+                defaults to `True`
+            auto_start (bool, optional): Whether to start the process immediately, defaults to `True`
             workers (int, optional): The number of worker processes to run. Defaults to `1`.
 
+
         Returns:
             Worker: The Worker instance
         """  # noqa: E501
+        if ident in self.transient or ident in self.durable:
+            raise ValueError(f"Worker {ident} already exists")
+        restartable = restartable if restartable is not None else transient
+        if transient and not restartable:
+            raise ValueError(
+                "Cannot create a transient worker that is not restartable"
+            )
         container = self.transient if transient else self.durable
         worker = Worker(
-            ident, func, kwargs, self.context, self.worker_state, workers
+            ident,
+            func,
+            kwargs,
+            self.context,
+            self.worker_state,
+            workers,
+            restartable,
+            tracked,
+            auto_start,
         )
         container[worker.ident] = worker
         return worker
@@ -119,6 +153,7 @@ class WorkerManager:
             self._serve,
             self._server_settings,
             transient=True,
+            restartable=True,
         )
 
     def shutdown_server(self, ident: Optional[str] = None) -> None:
@@ -158,8 +193,12 @@ class WorkerManager:
 
     def start(self):
         """Start the worker processes."""
-        for process in self.processes:
-            process.start()
+        for worker in self.workers:
+            for process in worker.processes:
+                if not worker.auto_start:
+                    process.set_state(ProcessState.NONE, True)
+                    continue
+                process.start()
 
     def join(self):
         """Join the worker processes."""
@@ -202,9 +241,13 @@ class WorkerManager:
             restart_order (RestartOrder, optional): The order in which to restart the processes.
                 Defaults to `RestartOrder.SHUTDOWN_FIRST`.
         """  # noqa: E501
-        for process in self.transient_processes:
-            if not process_names or process.name in process_names:
-                process.restart(restart_order=restart_order, **kwargs)
+        self.restarter.restart(
+            transient_processes=list(self.transient_processes),
+            durable_processes=list(self.durable_processes),
+            process_names=process_names,
+            restart_order=restart_order,
+            **kwargs,
+        )
 
     def scale(self, num_worker: int):
         if num_worker <= 0:
@@ -243,45 +286,13 @@ class WorkerManager:
         self.wait_for_ack()
         while True:
             try:
-                if self.monitor_subscriber.poll(0.1):
-                    message = self.monitor_subscriber.recv()
-                    logger.debug(
-                        f"Monitor message: {message}", extra={"verbosity": 2}
-                    )
-                    if not message:
-                        break
-                    elif message == "__TERMINATE__":
-                        self.shutdown()
-                        break
-                    logger.debug(
-                        "Incoming monitor message: %s",
-                        message,
-                        extra={"verbosity": 1},
-                    )
-                    split_message = message.split(":", 2)
-                    if message.startswith("__SCALE__"):
-                        self.scale(int(split_message[-1]))
-                        continue
-                    processes = split_message[0]
-                    reloaded_files = (
-                        split_message[1] if len(split_message) > 1 else None
-                    )
-                    process_names = [
-                        name.strip() for name in processes.split(",")
-                    ]
-                    if "__ALL_PROCESSES__" in process_names:
-                        process_names = None
-                    order = (
-                        RestartOrder.STARTUP_FIRST
-                        if "STARTUP_FIRST" in split_message
-                        else RestartOrder.SHUTDOWN_FIRST
-                    )
-                    self.restart(
-                        process_names=process_names,
-                        reloaded_files=reloaded_files,
-                        restart_order=order,
-                    )
+                cycle = self._poll_monitor()
+                if cycle is MonitorCycle.BREAK:
+                    break
+                elif cycle is MonitorCycle.CONTINUE:
+                    continue
                 self._sync_states()
+                self._cleanup_non_tracked_workers()
             except InterruptedError:
                 if not OS_IS_WINDOWS:
                     raise
@@ -326,11 +337,17 @@ class WorkerManager:
         """Get all of the workers."""
         return list(self.transient.values()) + list(self.durable.values())
 
+    @property
+    def all_workers(self) -> Iterable[Tuple[str, Worker]]:
+        return chain(self.transient.items(), self.durable.items())
+
     @property
     def processes(self):
         """Get all of the processes."""
         for worker in self.workers:
             for process in worker.processes:
+                if not process.pid:
+                    continue
                 yield process
 
     @property
@@ -340,11 +357,18 @@ class WorkerManager:
             for process in worker.processes:
                 yield process
 
+    @property
+    def durable_processes(self):
+        for worker in self.durable.values():
+            for process in worker.processes:
+                yield process
+
     def kill(self):
         """Kill all of the processes."""
         for process in self.processes:
             logger.info("Killing %s [%s]", process.name, process.pid)
-            os.kill(process.pid, SIGKILL)
+            with suppress(ProcessLookupError):
+                os.kill(process.pid, SIGKILL)
         raise ServerKilled
 
     def shutdown_signal(self, signal, frame):
@@ -365,6 +389,25 @@ class WorkerManager:
                 process.terminate()
         self._shutting_down = True
 
+    def remove_worker(self, worker: Worker) -> None:
+        if worker.tracked:
+            error_logger.error(
+                f"Worker {worker.ident} is tracked and cannot be removed."
+            )
+            return
+        if worker.has_alive_processes():
+            error_logger.error(
+                f"Worker {worker.ident} has alive processes and cannot be "
+                "removed."
+            )
+            return
+        self.transient.pop(worker.ident, None)
+        self.durable.pop(worker.ident, None)
+        for process in worker.processes:
+            self.worker_state.pop(process.name, None)
+        logger.info("Removed worker %s", worker.ident)
+        del worker
+
     @property
     def pid(self):
         """Get the process ID of the main process."""
@@ -385,5 +428,101 @@ class WorkerManager:
             except KeyError:
                 process.set_state(ProcessState.TERMINATED, True)
                 continue
+            if not process.is_alive():
+                state = "FAILED" if process.exitcode else "COMPLETED"
             if state and process.state.name != state:
                 process.set_state(ProcessState[state], True)
+
+    def _cleanup_non_tracked_workers(self) -> None:
+        to_remove = [
+            worker
+            for worker in self.workers
+            if not worker.tracked and not worker.has_alive_processes()
+        ]
+
+        for worker in to_remove:
+            self.remove_worker(worker)
+
+    def _poll_monitor(self) -> Optional[MonitorCycle]:
+        if self.monitor_subscriber.poll(0.1):
+            message = self.monitor_subscriber.recv()
+            logger.debug(f"Monitor message: {message}", extra={"verbosity": 2})
+            if not message:
+                return MonitorCycle.BREAK
+            elif message == "__TERMINATE__":
+                self._handle_terminate()
+                return MonitorCycle.BREAK
+            elif isinstance(message, tuple) and len(message) == 7:
+                self._handle_manage(*message)  # type: ignore
+                return MonitorCycle.CONTINUE
+            elif not isinstance(message, str):
+                error_logger.error(
+                    "Monitor received an invalid message: %s", message
+                )
+                return MonitorCycle.CONTINUE
+            return self._handle_message(message)
+        return None
+
+    def _handle_terminate(self) -> None:
+        self.shutdown()
+
+    def _handle_message(self, message: str) -> Optional[MonitorCycle]:
+        logger.debug(
+            "Incoming monitor message: %s",
+            message,
+            extra={"verbosity": 1},
+        )
+        split_message = message.split(":", 2)
+        if message.startswith("__SCALE__"):
+            self.scale(int(split_message[-1]))
+            return MonitorCycle.CONTINUE
+
+        processes = split_message[0]
+        reloaded_files = split_message[1] if len(split_message) > 1 else None
+        process_names: Optional[List[str]] = [
+            name.strip() for name in processes.split(",")
+        ]
+        if process_names and "__ALL_PROCESSES__" in process_names:
+            process_names = None
+        order = (
+            RestartOrder.STARTUP_FIRST
+            if "STARTUP_FIRST" in split_message
+            else RestartOrder.SHUTDOWN_FIRST
+        )
+        self.restart(
+            process_names=process_names,
+            reloaded_files=reloaded_files,
+            restart_order=order,
+        )
+
+        return None
+
+    def _handle_manage(
+        self,
+        ident: str,
+        func: Callable[..., Any],
+        kwargs: Dict[str, Any],
+        transient: bool,
+        restartable: Optional[bool],
+        tracked: bool,
+        auto_start: bool,
+        workers: int,
+    ) -> None:
+        try:
+            worker = self.manage(
+                ident,
+                func,
+                kwargs,
+                transient=transient,
+                restartable=restartable,
+                tracked=tracked,
+                auto_start=auto_start,
+                workers=workers,
+            )
+        except Exception:
+            error_logger.exception("Failed to manage worker %s", ident)
+        else:
+            if not auto_start:
+                return
+            for process in worker.processes:
+                process.start()
diff --git a/sanic/worker/multiplexer.py b/sanic/worker/multiplexer.py
index cf2054c5..a9717874 100644
--- a/sanic/worker/multiplexer.py
+++ b/sanic/worker/multiplexer.py
@@ -1,6 +1,6 @@
 from multiprocessing.connection import Connection
 from os import environ, getpid
-from typing import Any, Dict
+from typing import Any, Callable, Dict, Optional
 
 from sanic.log import Colors, logger
 from sanic.worker.process import ProcessState
@@ -39,6 +39,48 @@ class WorkerMultiplexer:
             "state": ProcessState.ACKED.name,
         }
 
+    def manage(
+        self,
+        ident: str,
+        func: Callable[..., Any],
+        kwargs: Dict[str, Any],
+        transient: bool = False,
+        restartable: Optional[bool] = None,
+        tracked: bool = False,
+        auto_start: bool = True,
+        workers: int = 1,
+    ) -> None:
+        """Manages the initiation and monitoring of a worker process.
+
+        Args:
+            ident (str): A unique identifier for the worker process.
+            func (Callable[..., Any]): The function to be executed in the worker process.
+            kwargs (Dict[str, Any]): A dictionary of arguments to be passed to `func`.
+            transient (bool, optional): Flag to mark the process as transient. If `True`,
+                the Worker Manager will restart the process with any global restart
+                (e.g., auto-reload). Defaults to `False`.
+            restartable (Optional[bool], optional): Flag to mark the process as restartable. If `True`,
+                the Worker Manager can restart the process if prompted. Defaults to `None`.
+            tracked (bool, optional): Flag to indicate whether the process should be tracked
+                after its completion. Defaults to `False`.
+            auto_start (bool, optional): Flag to indicate whether the process should be started
+            workers (int, optional): The number of worker processes to run. Defaults to 1.
+
+        This method packages the provided arguments into a bundle and sends them back to the
+        main process to be managed by the Worker Manager.
+        """  # noqa: E501
+        bundle = (
+            ident,
+            func,
+            kwargs,
+            transient,
+            restartable,
+            tracked,
+            auto_start,
+            workers,
+        )
+        self._monitor_publisher.send(bundle)
+
     def set_serving(self, serving: bool) -> None:
         """Set the worker to serving.
 
diff --git a/sanic/worker/process.py b/sanic/worker/process.py
index a66b517e..57c6e5df 100644
--- a/sanic/worker/process.py
+++ b/sanic/worker/process.py
@@ -1,6 +1,7 @@
 import os
 
 from datetime import datetime, timezone
+from inspect import signature
 from multiprocessing.context import BaseContext
 from signal import SIGINT
 from threading import Thread
@@ -22,13 +23,22 @@ class WorkerProcess:
     THRESHOLD = 300  # == 30 seconds
     SERVER_LABEL = "Server"
 
-    def __init__(self, factory, name, target, kwargs, worker_state):
+    def __init__(
+        self,
+        factory,
+        name,
+        target,
+        kwargs,
+        worker_state,
+        restartable: bool = False,
+    ):
         self.state = ProcessState.IDLE
         self.factory = factory
         self.name = name
         self.target = target
         self.kwargs = kwargs
         self.worker_state = worker_state
+        self.restartable = restartable
         if self.name not in self.worker_state:
             self.worker_state[self.name] = {
                 "server": self.SERVER_LABEL in self.name
@@ -108,9 +118,10 @@ class WorkerProcess:
             self._terminate_now()
         else:
             self._old_process = self._current_process
-        self.kwargs.update(
-            {"config": {k.upper(): v for k, v in kwargs.items()}}
-        )
+        if self._add_config():
+            self.kwargs.update(
+                {"config": {k.upper(): v for k, v in kwargs.items()}}
+            )
         try:
             self.spawn()
             self.start()
@@ -147,7 +158,13 @@ class WorkerProcess:
     def pid(self):
         return self._current_process.pid
 
+    @property
+    def exitcode(self):
+        return self._current_process.exitcode
+
     def _terminate_now(self):
+        if not self._current_process.is_alive():
+            return
         logger.debug(
             f"{Colors.BLUE}Begin restart termination: "
             f"{Colors.BOLD}{Colors.SANIC}"
@@ -196,6 +213,15 @@ class WorkerProcess:
             self._old_process.terminate()
         delattr(self, "_old_process")
 
+    def _add_config(self) -> bool:
+        sig = signature(self.target)
+        if "config" in sig.parameters or any(
+            param.kind == param.VAR_KEYWORD
+            for param in sig.parameters.values()
+        ):
+            return True
+        return False
+
 
 class Worker:
     WORKER_PREFIX = "Sanic-"
@@ -208,6 +234,9 @@ class Worker:
         context: BaseContext,
         worker_state: Dict[str, Any],
         num: int = 1,
+        restartable: bool = False,
+        tracked: bool = True,
+        auto_start: bool = True,
     ):
         self.ident = ident
         self.num = num
@@ -216,6 +245,9 @@ class Worker:
         self.server_settings = server_settings
         self.worker_state = worker_state
         self.processes: Set[WorkerProcess] = set()
+        self.restartable = restartable
+        self.tracked = tracked
+        self.auto_start = auto_start
         for _ in range(num):
             self.create_process()
 
@@ -230,6 +262,10 @@ class Worker:
             target=self.serve,
             kwargs={**self.server_settings},
             worker_state=self.worker_state,
+            restartable=self.restartable,
         )
         self.processes.add(process)
         return process
+
+    def has_alive_processes(self) -> bool:
+        return any(process.is_alive() for process in self.processes)
diff --git a/sanic/worker/restarter.py b/sanic/worker/restarter.py
new file mode 100644
index 00000000..5ae50a4b
--- /dev/null
+++ b/sanic/worker/restarter.py
@@ -0,0 +1,90 @@
+from typing import List, Optional, Set
+
+from sanic.log import error_logger
+from sanic.worker.constants import RestartOrder
+from sanic.worker.process import ProcessState, WorkerProcess
+
+
+class Restarter:
+    def restart(
+        self,
+        transient_processes: List[WorkerProcess],
+        durable_processes: List[WorkerProcess],
+        process_names: Optional[List[str]] = None,
+        restart_order=RestartOrder.SHUTDOWN_FIRST,
+        **kwargs,
+    ) -> None:
+        """Restart the worker processes.
+
+        Args:
+            process_names (Optional[List[str]], optional): The names of the processes to restart.
+                If `None`, then all processes will be restarted. Defaults to `None`.
+            restart_order (RestartOrder, optional): The order in which to restart the processes.
+                Defaults to `RestartOrder.SHUTDOWN_FIRST`.
+        """  # noqa: E501
+        restarted = self._restart_transient(
+            transient_processes,
+            process_names or [],
+            restart_order,
+            **kwargs,
+        )
+        restarted |= self._restart_durable(
+            durable_processes,
+            process_names or [],
+            restart_order,
+            **kwargs,
+        )
+
+        if process_names and not restarted:
+            error_logger.error(
+                f"Failed to restart processes: {', '.join(process_names)}"
+            )
+
+    def _restart_transient(
+        self,
+        processes: List[WorkerProcess],
+        process_names: List[str],
+        restart_order: RestartOrder,
+        **kwargs,
+    ) -> Set[str]:
+        restarted: Set[str] = set()
+        for process in processes:
+            if not process.restartable or (
+                process_names and process.name not in process_names
+            ):
+                continue
+            self._restart_process(process, restart_order, **kwargs)
+            restarted.add(process.name)
+        return restarted
+
+    def _restart_durable(
+        self,
+        processes: List[WorkerProcess],
+        process_names: List[str],
+        restart_order: RestartOrder,
+        **kwargs,
+    ) -> Set[str]:
+        restarted: Set[str] = set()
+        if not process_names:
+            return restarted
+        for process in processes:
+            if not process.restartable or process.name not in process_names:
+                continue
+            if process.state not in (
+                ProcessState.COMPLETED,
+                ProcessState.FAILED,
+                ProcessState.NONE,
+            ):
+                error_logger.error(
+                    f"Cannot restart process {process.name} because "
+                    "it is not in a final state. Current state is: "
+                    f"{process.state.name}."
+                )
+                continue
+            self._restart_process(process, restart_order, **kwargs)
+            restarted.add(process.name)
+
+        return restarted
+
+    def _restart_process(self, process, restart_order, **kwargs):
+        process.restart(restart_order=restart_order, **kwargs)


Test patch:
diff --git a/tests/worker/test_manager.py b/tests/worker/test_manager.py
index ae5f4592..76d6cfbd 100644
--- a/tests/worker/test_manager.py
+++ b/tests/worker/test_manager.py
@@ -8,6 +8,7 @@ from sanic.compat import OS_IS_WINDOWS
 from sanic.exceptions import ServerKilled
 from sanic.worker.constants import RestartOrder
 from sanic.worker.manager import WorkerManager
+from sanic.worker.process import Worker
 
 
 if not OS_IS_WINDOWS:
@@ -20,6 +21,17 @@ def fake_serve():
     ...
 
 
+@pytest.fixture
+def manager() -> WorkerManager:
+    p1 = Mock()
+    p1.pid = 1234
+    context = Mock()
+    context.Process.return_value = p1
+    pub = Mock()
+    manager = WorkerManager(1, fake_serve, {}, context, (pub, Mock()), {})
+    return manager
+
+
 def test_manager_no_workers():
     message = "Cannot serve with no workers"
     with pytest.raises(RuntimeError, match=message):
@@ -94,25 +106,25 @@ def test_restart_all():
             call(
                 name="Sanic-Server-0-0",
                 target=fake_serve,
-                kwargs={"config": {}},
+                kwargs={},
                 daemon=True,
             ),
             call(
                 name="Sanic-Server-1-0",
                 target=fake_serve,
-                kwargs={"config": {}},
+                kwargs={},
                 daemon=True,
             ),
             call(
                 name="Sanic-Server-0-0",
                 target=fake_serve,
-                kwargs={"config": {}},
+                kwargs={},
                 daemon=True,
             ),
             call(
                 name="Sanic-Server-1-0",
                 target=fake_serve,
-                kwargs={"config": {}},
+                kwargs={},
                 daemon=True,
             ),
         ]
@@ -303,3 +315,130 @@ def test_scale(caplog):
 
     with pytest.raises(ValueError, match=r"Cannot scale to 0 workers\."):
         manager.scale(0)
+
+
+def test_manage_basic(manager: WorkerManager):
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 0
+    manager.manage("TEST", fake_serve, kwargs={"foo": "bar"})
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+
+    worker_process = manager.durable["TEST"]
+
+    assert isinstance(worker_process, Worker)
+    assert worker_process.server_settings == {"foo": "bar"}
+    assert worker_process.restartable is False
+    assert worker_process.tracked is True
+    assert worker_process.auto_start is True
+    assert worker_process.num == 1
+
+
+def test_manage_transient(manager: WorkerManager):
+    manager.manage(
+        "TEST", fake_serve, kwargs={"foo": "bar"}, workers=3, transient=True
+    )
+    assert len(manager.transient) == 2
+    assert len(manager.durable) == 0
+
+    worker_process = manager.transient["TEST"]
+
+    assert isinstance(worker_process, Worker)
+    assert worker_process.restartable is True
+    assert worker_process.tracked is True
+    assert worker_process.auto_start is True
+    assert worker_process.num == 3
+
+
+def test_manage_restartable(manager: WorkerManager):
+    manager.manage(
+        "TEST",
+        fake_serve,
+        kwargs={"foo": "bar"},
+        restartable=True,
+        auto_start=False,
+    )
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+
+    worker_process = manager.durable["TEST"]
+
+    assert isinstance(worker_process, Worker)
+    assert worker_process.restartable is True
+    assert worker_process.tracked is True
+    assert worker_process.auto_start is False
+
+
+def test_manage_untracked(manager: WorkerManager):
+    manager.manage("TEST", fake_serve, kwargs={"foo": "bar"}, tracked=False)
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+
+    worker_process = manager.durable["TEST"]
+
+    assert isinstance(worker_process, Worker)
+    assert worker_process.restartable is False
+    assert worker_process.tracked is False
+    assert worker_process.auto_start is True
+
+
+def test_manage_duplicate_ident(manager: WorkerManager):
+    manager.manage("TEST", fake_serve, kwargs={"foo": "bar"})
+    message = "Worker TEST already exists"
+    with pytest.raises(ValueError, match=message):
+        manager.manage("TEST", fake_serve, kwargs={"foo": "bar"})
+
+
+def test_transient_not_restartable(manager: WorkerManager):
+    message = "Cannot create a transient worker that is not restartable"
+    with pytest.raises(ValueError, match=message):
+        manager.manage(
+            "TEST",
+            fake_serve,
+            kwargs={"foo": "bar"},
+            transient=True,
+            restartable=False,
+        )
+
+
+def test_remove_worker(manager: WorkerManager, caplog):
+    worker = manager.manage("TEST", fake_serve, kwargs={})
+
+    assert "Sanic-TEST-0" in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+
+    manager.remove_worker(worker)
+    message = "Worker TEST is tracked and cannot be removed."
+
+    assert "Sanic-TEST-0" in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+    assert ("sanic.error", 40, message) in caplog.record_tuples
+
+
+def test_remove_untracked_worker(manager: WorkerManager, caplog):
+    caplog.set_level(20)
+    worker = manager.manage("TEST", fake_serve, kwargs={}, tracked=False)
+    worker.has_alive_processes = Mock(return_value=True)
+
+    assert "Sanic-TEST-0" in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+
+    manager.remove_worker(worker)
+    message = "Worker TEST has alive processes and cannot be removed."
+
+    assert "Sanic-TEST-0" in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 1
+    assert ("sanic.error", 40, message) in caplog.record_tuples
+
+    worker.has_alive_processes = Mock(return_value=False)
+    manager.remove_worker(worker)
+    message = "Removed worker TEST"
+
+    assert "Sanic-TEST-0" not in worker.worker_state
+    assert len(manager.transient) == 1
+    assert len(manager.durable) == 0
+    assert ("sanic.root", 20, message) in caplog.record_tuples
diff --git a/tests/worker/test_multiplexer.py b/tests/worker/test_multiplexer.py
index 8195b094..549ee84a 100644
--- a/tests/worker/test_multiplexer.py
+++ b/tests/worker/test_multiplexer.py
@@ -13,6 +13,10 @@ from sanic.worker.multiplexer import WorkerMultiplexer
 from sanic.worker.state import WorkerState
 
 
+def noop(*args, **kwargs):
+    pass
+
+
 @pytest.fixture
 def monitor_publisher():
     return Mock()
@@ -103,6 +107,13 @@ def test_scale(monitor_publisher: Mock, m: WorkerMultiplexer):
     monitor_publisher.send.assert_called_once_with("__SCALE__:99")
 
 
+def test_manage(monitor_publisher: Mock, m: WorkerMultiplexer):
+    m.manage("NEW", noop, auto_start=False, kwargs={"foo": 99})
+    monitor_publisher.send.assert_called_once_with(
+        ("NEW", noop, {"foo": 99}, False, None, False, False, 1)
+    )
+
+
 def test_properties(
     monitor_publisher: Mock, worker_state: Dict[str, Any], m: WorkerMultiplexer
 ):
diff --git a/tests/worker/test_restarter.py b/tests/worker/test_restarter.py
new file mode 100644
index 00000000..37932fe1
--- /dev/null
+++ b/tests/worker/test_restarter.py
@@ -0,0 +1,89 @@
+from unittest.mock import Mock
+
+import pytest
+
+from sanic.worker.constants import ProcessState, RestartOrder
+from sanic.worker.process import WorkerProcess
+from sanic.worker.restarter import Restarter
+
+
+def noop(*args, **kwargs):
+    pass
+
+
+def make_worker_process(
+    name: str, state: ProcessState = ProcessState.STARTED
+) -> WorkerProcess:
+    worker_process = Mock()
+    worker_process.restart = Mock()
+    worker_process.name = name
+    worker_process.state = state
+    return worker_process
+
+
+def test_restart_transient():
+    transient = make_worker_process("Transient")
+    durable = make_worker_process("Durable")
+    restarter = Restarter()
+
+    restarter.restart([transient], [durable])
+    transient.restart.assert_called_once_with(
+        restart_order=RestartOrder.SHUTDOWN_FIRST
+    )
+    durable.restart.assert_not_called()
+    transient.restart.reset_mock()
+    restarter.restart(
+        [transient], [durable], restart_order=RestartOrder.STARTUP_FIRST
+    )
+    transient.restart.assert_called_once_with(
+        restart_order=RestartOrder.STARTUP_FIRST
+    )
+
+
+@pytest.mark.parametrize(
+    "state,called",
+    (
+        (ProcessState.IDLE, False),
+        (ProcessState.RESTARTING, False),
+        (ProcessState.STARTING, False),
+        (ProcessState.STARTED, False),
+        (ProcessState.ACKED, False),
+        (ProcessState.JOINED, False),
+        (ProcessState.TERMINATED, False),
+        (ProcessState.FAILED, True),
+        (ProcessState.COMPLETED, True),
+        (ProcessState.NONE, True),
+    ),
+)
+def test_restart_durable(caplog, state, called):
+    transient = make_worker_process("Transient")
+    durable = make_worker_process("Durable")
+    restarter = Restarter()
+
+    restarter.restart([transient], [durable], process_names=["Durable"])
+
+    transient.restart.assert_not_called()
+    durable.restart.assert_not_called()
+
+    assert (
+        "sanic.error",
+        40,
+        "Cannot restart process Durable because it is not in a "
+        "final state. Current state is: STARTED.",
+    ) in caplog.record_tuples
+    assert (
+        "sanic.error",
+        40,
+        "Failed to restart processes: Durable",
+    ) in caplog.record_tuples
+
+    durable.state = state
+    restarter.restart([transient], [durable], process_names=["Durable"])
+
+    transient.restart.assert_not_called()
+    if called:
+        durable.restart.assert_called_once_with(
+            restart_order=RestartOrder.SHUTDOWN_FIRST
+        )
+    else:
+        durable.restart.assert_not_called()