New multiprocess manager (#2183)

* New multiprocess manager

* lint it

* Fixed test

* Fixed `Process`.`__init__`

* Fix signal handling in Multiprocess class

* Update coverage fail_under value

* Remove redundant log message

* Update coverage fail_under value

* Update coverage fail_under value

* Update fail_under value in coverage report

* Remove unused threading event

* lint

* more tests

* More tests and fix bug

* lint

* Add pytest.mark.skipif for SIGHUP test on Windows

* delete unused code

* More tests

* Try fix tests in Windows

* make linter feels great

* delete pytest-xdist

* Try fix test in windows

* Try make mypy happy

* Skip tests in windows

* lint

* Try test basic run in Windows

* Try fix error in Windows

* lint

* Skip tests in window

* Try test in window

* lint

* Add import statement and set current working directory in test_multiprocess.py

* lint

* giveup

* Refactor signal handling in Multiprocess class

* Tests in windows

* lint

* lint

* ignore mypy check in linux

* Add __init__.py

* fix warning

* coverage ignore

* Update coverage

* coverage

* Add documents

* Update docs/deployment.md

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Update uvicorn/supervisors/multiprocess.py

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

* Do not output the PID information repeatedly.

* Fix occasional abnormal exits.

* Update docs

* Change subprocess termination logic in Multiprocess class

* Update new_console_in_windows function to include pragma statement

* Revert coverage to 98.35

* chore: Remove pragma statements from test_multiprocess functions

* chore: Exclude test_multiprocess.py from coverage on Windows

---------

Co-authored-by: Marcelo Trylesinski <marcelotryle@gmail.com>

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()