New multiprocess manager

docs/deployment.md

@@ -178,7 +178,23 @@ Running Uvicorn using a process manager ensures that you can run multiple proces
 
 A process manager will handle the socket setup, start-up multiple server processes, monitor process aliveness, and listen for signals to provide for processes restarts, shutdowns, or dialing up and down the number of running processes.
 
-Uvicorn provides a lightweight way to run multiple worker processes, for example `--workers 4`, but does not provide any process monitoring.
+### Built-in
+
+Uvicorn includes a `--workers` option that allows you to run multiple worker processes.
+
+```bash
+$ uvicorn main:app --workers 4
+```
+
+Unlike gunicorn, uvicorn does not use pre-fork, but uses [`spawn`](https://docs.python.org/3/library/multiprocessing.html#contexts-and-start-methods), which allows uvicorn's multiprocess manager to still work well on Windows.
+
+The default process manager monitors the status of child processes and automatically restarts child processes that die unexpectedly. Not only that, it will also monitor the status of the child process through the pipeline. When the child process is accidentally stuck, the corresponding child process will be killed through an unstoppable system signal or interface.
+
+You can also manage child processes by sending specific signals to the main process. (Not supported on Windows.)
+
+- `SIGHUP`: Work processeses are graceful restarted one after another. If you update the code, the new worker process will use the new code.
+- `SIGTTIN`: Increase the number of worker processes by one.
+- `SIGTTOU`: Decrease the number of worker processes by one.
 
 ### Gunicorn
 

pyproject.toml

@@ -115,7 +115,11 @@ exclude_lines = [
 ]
 
 [tool.coverage.coverage_conditional_plugin.omit]
-"sys_platform == 'win32'" = ["uvicorn/loops/uvloop.py"]
+"sys_platform == 'win32'" = [
+    "uvicorn/loops/uvloop.py",
+    "uvicorn/supervisors/multiprocess.py",
+    "tests/supervisors/test_multiprocess.py",
+]
 "sys_platform != 'win32'" = ["uvicorn/loops/asyncio.py"]
 
 [tool.coverage.coverage_conditional_plugin.rules]

tests/supervisors/test_multiprocess.py

@@ -1,21 +1,66 @@
 from __future__ import annotations
 
+import functools
+import os
 import signal
 import socket
+import threading
+import time
+from typing import Any, Callable
+
+import pytest
 
 from uvicorn import Config
 from uvicorn._types import ASGIReceiveCallable, ASGISendCallable, Scope
 from uvicorn.supervisors import Multiprocess
+from uvicorn.supervisors.multiprocess import Process
+
+
+def new_console_in_windows(test_function: Callable[[], Any]) -> Callable[[], Any]:  # pragma: no cover
+    if os.name != "nt":
+        return test_function
+
+    @functools.wraps(test_function)
+    def new_function():
+        import subprocess
+        import sys
+
+        module = test_function.__module__
+        name = test_function.__name__
+
+        subprocess.check_call(
+            [
+                sys.executable,
+                "-c",
+                f"from {module} import {name}; {name}.__wrapped__()",
+            ],
+            creationflags=subprocess.CREATE_NO_WINDOW,  # type: ignore[attr-defined]
+        )
+
+    return new_function
 
 
 async def app(scope: Scope, receive: ASGIReceiveCallable, send: ASGISendCallable) -> None:
     pass  # pragma: no cover
 
 
 def run(sockets: list[socket.socket] | None) -> None:
-    pass  # pragma: no cover
+    while True:
+        time.sleep(1)
+
+
+def test_process_ping_pong() -> None:
+    process = Process(Config(app=app), target=lambda x: None, sockets=[])
+    threading.Thread(target=process.always_pong, daemon=True).start()
+    assert process.ping()
+
+
+def test_process_ping_pong_timeout() -> None:
+    process = Process(Config(app=app), target=lambda x: None, sockets=[])
+    assert not process.ping(0.1)
 
 
+@new_console_in_windows
 def test_multiprocess_run() -> None:
     """
     A basic sanity check.
@@ -25,5 +70,102 @@ def test_multiprocess_run() -> None:
     """
     config = Config(app=app, workers=2)
     supervisor = Multiprocess(config, target=run, sockets=[])
-    supervisor.signal_handler(sig=signal.SIGINT, frame=None)
-    supervisor.run()
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    supervisor.signal_queue.append(signal.SIGINT)
+    supervisor.join_all()
+
+
+@new_console_in_windows
+def test_multiprocess_health_check() -> None:
+    """
+    Ensure that the health check works as expected.
+    """
+    config = Config(app=app, workers=2)
+    supervisor = Multiprocess(config, target=run, sockets=[])
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    time.sleep(1)
+    process = supervisor.processes[0]
+    process.kill()
+    assert not process.is_alive()
+    time.sleep(1)
+    for p in supervisor.processes:
+        assert p.is_alive()
+    supervisor.signal_queue.append(signal.SIGINT)
+    supervisor.join_all()
+
+
+@new_console_in_windows
+def test_multiprocess_sigterm() -> None:
+    """
+    Ensure that the SIGTERM signal is handled as expected.
+    """
+    config = Config(app=app, workers=2)
+    supervisor = Multiprocess(config, target=run, sockets=[])
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    time.sleep(1)
+    supervisor.signal_queue.append(signal.SIGTERM)
+    supervisor.join_all()
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGBREAK"), reason="platform unsupports SIGBREAK")
+@new_console_in_windows
+def test_multiprocess_sigbreak() -> None:  # pragma: py-not-win32
+    """
+    Ensure that the SIGBREAK signal is handled as expected.
+    """
+    config = Config(app=app, workers=2)
+    supervisor = Multiprocess(config, target=run, sockets=[])
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    time.sleep(1)
+    supervisor.signal_queue.append(getattr(signal, "SIGBREAK"))
+    supervisor.join_all()
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGHUP"), reason="platform unsupports SIGHUP")
+def test_multiprocess_sighup() -> None:
+    """
+    Ensure that the SIGHUP signal is handled as expected.
+    """
+    config = Config(app=app, workers=2)
+    supervisor = Multiprocess(config, target=run, sockets=[])
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    time.sleep(1)
+    pids = [p.pid for p in supervisor.processes]
+    supervisor.signal_queue.append(signal.SIGHUP)
+    time.sleep(1)
+    assert pids != [p.pid for p in supervisor.processes]
+    supervisor.signal_queue.append(signal.SIGINT)
+    supervisor.join_all()
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGTTIN"), reason="platform unsupports SIGTTIN")
+def test_multiprocess_sigttin() -> None:
+    """
+    Ensure that the SIGTTIN signal is handled as expected.
+    """
+    config = Config(app=app, workers=2)
+    supervisor = Multiprocess(config, target=run, sockets=[])
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    supervisor.signal_queue.append(signal.SIGTTIN)
+    time.sleep(1)
+    assert len(supervisor.processes) == 3
+    supervisor.signal_queue.append(signal.SIGINT)
+    supervisor.join_all()
+
+
+@pytest.mark.skipif(not hasattr(signal, "SIGTTOU"), reason="platform unsupports SIGTTOU")
+def test_multiprocess_sigttou() -> None:
+    """
+    Ensure that the SIGTTOU signal is handled as expected.
+    """
+    config = Config(app=app, workers=2)
+    supervisor = Multiprocess(config, target=run, sockets=[])
+    threading.Thread(target=supervisor.run, daemon=True).start()
+    supervisor.signal_queue.append(signal.SIGTTOU)
+    time.sleep(1)
+    assert len(supervisor.processes) == 1
+    supervisor.signal_queue.append(signal.SIGTTOU)
+    time.sleep(1)
+    assert len(supervisor.processes) == 1
+    supervisor.signal_queue.append(signal.SIGINT)
+    supervisor.join_all()