prompt: We are currently solving the following issue within our repository. Here is the issue text:
  --- BEGIN ISSUE ---
  restart workers online (graceful restart) to hot reload, in production environment.
  ### Is there an existing issue for this?
  
  - [X] I have searched the existing issues
  
  ### Is your feature request related to a problem? Please describe.
  
  Condition:  keep service available any time.  
  
  # ---- client code----
  
  ```py
  async def run():                                                                                        
      while 1:
          await post('http://127.0.0.1:8000/') 
  ```
  
  when called app.m.restart("__ALL_PROCESSES__") in a worker, sanic crashed.
  
  # ---- server code ----
  ```py
  @app.post("/")
  async def handler(request):
      app.m.restart('__ALL_PROCESSES__')
      return response.text('ok')
  
  
  if __name__ == "__main__":
      app.run(debug=True, workers=2)
  ```
  
  ### Describe the solution you'd like
  
  graceful restarting and reduce the effect when restarting.
  my messy describe:
  1. graceful restart workers; restart all workers will not crash,  if only 1 worker, block a little while (if worker not started yet) is ok.
  2. a way to graceful restart worker one by one, code eg:
      woker_names = tuple(app.m.workers.keys())
      for woker_name in worker_names:
          ret_val = app.m.restart(worker_name)
          # here, the worker has been graceful restarted, ret_val is meaningful
  3. may combine the above 2, when restarting all workers,  50% workers restarting, 50% old workers keep serving
  
  ### Additional context
  
  simplify the api,
  
  ```py
  app.m.restart('__ALL_PROCESSES__')   => app.m.restart_all()
  ```
  
  thanks.
  
  --- END ISSUE ---
  
  Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
  --- BEGIN FILES ---
  Path: `sanic/worker/multiplexer.py`
  Content:
  ```
  1 from multiprocessing.connection import Connection
  2 from os import environ, getpid
  3 from typing import Any, Dict
  4 
  5 from sanic.worker.process import ProcessState
  6 from sanic.worker.state import WorkerState
  7 
  8 
  9 class WorkerMultiplexer:
  10     def __init__(
  11         self,
  12         monitor_publisher: Connection,
  13         worker_state: Dict[str, Any],
  14     ):
  15         self._monitor_publisher = monitor_publisher
  16         self._state = WorkerState(worker_state, self.name)
  17 
  18     def ack(self):
  19         self._state._state[self.name] = {
  20             **self._state._state[self.name],
  21             "state": ProcessState.ACKED.name,
  22         }
  23 
  24     def restart(self, name: str = ""):
  25         if not name:
  26             name = self.name
  27         self._monitor_publisher.send(name)
  28 
  29     reload = restart  # no cov
  30 
  31     def terminate(self, early: bool = False):
  32         message = "__TERMINATE_EARLY__" if early else "__TERMINATE__"
  33         self._monitor_publisher.send(message)
  34 
  35     @property
  36     def pid(self) -> int:
  37         return getpid()
  38 
  39     @property
  40     def name(self) -> str:
  41         return environ.get("SANIC_WORKER_NAME", "")
  42 
  43     @property
  44     def state(self):
  45         return self._state
  46 
  47     @property
  48     def workers(self) -> Dict[str, Any]:
  49         return self.state.full()
  50
  ```
  --- 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()
  
  ```
  
golden_diff: diff --git a/sanic/worker/multiplexer.py b/sanic/worker/multiplexer.py
  --- a/sanic/worker/multiplexer.py
  +++ b/sanic/worker/multiplexer.py
  @@ -21,9 +21,14 @@
               "state": ProcessState.ACKED.name,
           }
   
  -    def restart(self, name: str = ""):
  +    def restart(self, name: str = "", all_workers: bool = False):
  +        if name and all_workers:
  +            raise ValueError(
  +                "Ambiguous restart with both a named process and"
  +                " all_workers=True"
  +            )
           if not name:
  -            name = self.name
  +            name = "__ALL_PROCESSES__:" if all_workers else self.name
           self._monitor_publisher.send(name)
   
       reload = restart  # no cov
  
verification_info: {"golden_diff": "diff --git a/sanic/worker/multiplexer.py b/sanic/worker/multiplexer.py\n--- a/sanic/worker/multiplexer.py\n+++ b/sanic/worker/multiplexer.py\n@@ -21,9 +21,14 @@\n             \"state\": ProcessState.ACKED.name,\n         }\n \n-    def restart(self, name: str = \"\"):\n+    def restart(self, name: str = \"\", all_workers: bool = False):\n+        if name and all_workers:\n+            raise ValueError(\n+                \"Ambiguous restart with both a named process and\"\n+                \" all_workers=True\"\n+            )\n         if not name:\n-            name = self.name\n+            name = \"__ALL_PROCESSES__:\" if all_workers else self.name\n         self._monitor_publisher.send(name)\n \n     reload = restart  # no cov\n", "issue": "restart workers online (graceful restart) to hot reload, in production environment.\n### Is there an existing issue for this?\r\n\r\n- [X] I have searched the existing issues\r\n\r\n### Is your feature request related to a problem? Please describe.\r\n\r\nCondition:  keep service available any time.  \r\n\r\n# ---- client code----\r\n\r\n```py\r\nasync def run():                                                                                        \r\n    while 1:\r\n        await post('http://127.0.0.1:8000/') \r\n```\r\n\r\nwhen called app.m.restart(\"__ALL_PROCESSES__\") in a worker, sanic crashed.\r\n\r\n# ---- server code ----\r\n```py\r\n@app.post(\"/\")\r\nasync def handler(request):\r\n    app.m.restart('__ALL_PROCESSES__')\r\n    return response.text('ok')\r\n\r\n\r\nif __name__ == \"__main__\":\r\n    app.run(debug=True, workers=2)\r\n```\r\n\r\n### Describe the solution you'd like\r\n\r\ngraceful restarting and reduce the effect when restarting.\r\nmy messy describe:\r\n1. graceful restart workers; restart all workers will not crash,  if only 1 worker, block a little while (if worker not started yet) is ok.\r\n2. a way to graceful restart worker one by one, code eg:\r\n    woker_names = tuple(app.m.workers.keys())\r\n    for woker_name in worker_names:\r\n        ret_val = app.m.restart(worker_name)\r\n        # here, the worker has been graceful restarted, ret_val is meaningful\r\n3. may combine the above 2, when restarting all workers,  50% workers restarting, 50% old workers keep serving\r\n\r\n### Additional context\r\n\r\nsimplify the api,\r\n\r\n```py\r\napp.m.restart('__ALL_PROCESSES__')   => app.m.restart_all()\r\n```\r\n\r\nthanks.\n", "before_files": [{"content": "from multiprocessing.connection import Connection\nfrom os import environ, getpid\nfrom typing import Any, Dict\n\nfrom sanic.worker.process import ProcessState\nfrom sanic.worker.state import WorkerState\n\n\nclass WorkerMultiplexer:\n    def __init__(\n        self,\n        monitor_publisher: Connection,\n        worker_state: Dict[str, Any],\n    ):\n        self._monitor_publisher = monitor_publisher\n        self._state = WorkerState(worker_state, self.name)\n\n    def ack(self):\n        self._state._state[self.name] = {\n            **self._state._state[self.name],\n            \"state\": ProcessState.ACKED.name,\n        }\n\n    def restart(self, name: str = \"\"):\n        if not name:\n            name = self.name\n        self._monitor_publisher.send(name)\n\n    reload = restart  # no cov\n\n    def terminate(self, early: bool = False):\n        message = \"__TERMINATE_EARLY__\" if early else \"__TERMINATE__\"\n        self._monitor_publisher.send(message)\n\n    @property\n    def pid(self) -> int:\n        return getpid()\n\n    @property\n    def name(self) -> str:\n        return environ.get(\"SANIC_WORKER_NAME\", \"\")\n\n    @property\n    def state(self):\n        return self._state\n\n    @property\n    def workers(self) -> Dict[str, Any]:\n        return self.state.full()\n", "path": "sanic/worker/multiplexer.py"}], "after_files": [{"content": "from multiprocessing.connection import Connection\nfrom os import environ, getpid\nfrom typing import Any, Dict\n\nfrom sanic.worker.process import ProcessState\nfrom sanic.worker.state import WorkerState\n\n\nclass WorkerMultiplexer:\n    def __init__(\n        self,\n        monitor_publisher: Connection,\n        worker_state: Dict[str, Any],\n    ):\n        self._monitor_publisher = monitor_publisher\n        self._state = WorkerState(worker_state, self.name)\n\n    def ack(self):\n        self._state._state[self.name] = {\n            **self._state._state[self.name],\n            \"state\": ProcessState.ACKED.name,\n        }\n\n    def restart(self, name: str = \"\", all_workers: bool = False):\n        if name and all_workers:\n            raise ValueError(\n                \"Ambiguous restart with both a named process and\"\n                \" all_workers=True\"\n            )\n        if not name:\n            name = \"__ALL_PROCESSES__:\" if all_workers else self.name\n        self._monitor_publisher.send(name)\n\n    reload = restart  # no cov\n\n    def terminate(self, early: bool = False):\n        message = \"__TERMINATE_EARLY__\" if early else \"__TERMINATE__\"\n        self._monitor_publisher.send(message)\n\n    @property\n    def pid(self) -> int:\n        return getpid()\n\n    @property\n    def name(self) -> str:\n        return environ.get(\"SANIC_WORKER_NAME\", \"\")\n\n    @property\n    def state(self):\n        return self._state\n\n    @property\n    def workers(self) -> Dict[str, Any]:\n        return self.state.full()\n", "path": "sanic/worker/multiplexer.py"}]}