Create ProcessManager

tests/supervisors/test_multiprocess.py

@@ -1,7 +1,7 @@
 import signal
 
 from uvicorn import Config
-from uvicorn.supervisors import Multiprocess
+from uvicorn.supervisors.multiprocess import Multiprocess
 
 
 def run(sockets):

uvicorn/server.py

@@ -58,6 +58,7 @@ def __init__(self, config: Config) -> None:
         self.server_state = ServerState()
 
         self.started = False
+        self.terminate_called = 0
         self.should_exit = False
         self.force_exit = False
         self.last_notified = 0.0
@@ -309,8 +310,7 @@ def install_signal_handlers(self) -> None:
                 signal.signal(sig, self.handle_exit)
 
     def handle_exit(self, sig: signal.Signals, frame: FrameType) -> None:
-
-        if self.should_exit:
+        self.terminate_called += 1
+        self.should_exit = True
+        if self.terminate_called > 2:
             self.force_exit = True
-        else:
-            self.should_exit = True

uvicorn/supervisors/__init__.py

@@ -1,7 +1,7 @@
+import sys
 import typing
 
 from uvicorn.supervisors.basereload import BaseReload
-from uvicorn.supervisors.multiprocess import Multiprocess
 
 if typing.TYPE_CHECKING:
     ChangeReload: typing.Type[BaseReload]  # pragma: no cover
@@ -11,4 +11,9 @@
     except ImportError:  # pragma: no cover
         from uvicorn.supervisors.statreload import StatReload as ChangeReload
 
+if sys.platform == "win32":
+    from uvicorn.supervisors.multiprocess import Multiprocess
+else:
+    from uvicorn.supervisors.manager import ProcessManager as Multiprocess
+
 __all__ = ["Multiprocess", "ChangeReload"]

uvicorn/supervisors/manager.py

@@ -0,0 +1,157 @@
+import enum
+import logging
+import multiprocessing as mp
+import os
+import queue
+import signal
+import sys
+import time
+from multiprocessing.context import SpawnProcess
+from socket import socket
+from types import FrameType
+from typing import Callable, List, Optional
+
+from uvicorn.config import Config
+from uvicorn.subprocess import get_subprocess
+
+if sys.version_info >= (3, 8):
+    from typing import Protocol
+else:
+    from typing_extensions import Protocol
+
+logger = logging.getLogger("uvicorn.error")
+
+
+class Target(Protocol):
+    def __call__(self, sockets: Optional[List[socket]] = None) -> None:
+        ...
+
+
+class ExitCode(enum.IntEnum):
+    OK = 0
+    DEFAULT_FAILURE = 1
+    STARTUP_FAILURE = 3
+
+
+class ProcessManager:
+    # Complete list of signals can be found on:
+    # http://manpages.ubuntu.com/manpages/bionic/man7/signal.7.html
+    SIGNALS = {
+        getattr(signal, f"SIG{sig.upper()}"): sig
+        for sig in (
+            "abrt",  # Abort signal from abort(3)
+            "hup",  # Hangup signal generated by terminal close.
+            "quit",  # Quit signal generated by terminal close.
+            "int",  # Interrupt signal generated by Ctrl+C
+            "term",  # Termination signal
+            "winch",  # Window size change signal
+            "chld",  # Child process terminated, stopped, or continued
+        )
+    }
+
+    # TODO(Marcelo): This should be converted into a CLI option.
+    GRACEFUL_TIMEOUT = 30
+
+    def __init__(self, config: Config, target: Target, sockets: List[socket]) -> None:
+        self.config = config
+        self.target = target
+        self.sockets = sockets
+        self.processes: List[SpawnProcess] = []
+        self.sig_queue = mp.Queue()
+
+    def run(self) -> None:
+        self.start()
+
+        try:
+            self.spawn_processes()
+
+            while True:
+                try:
+                    sig = self.sig_queue.get(timeout=0.25)
+                except queue.Empty:
+                    self.reap_processes()
+                    self.spawn_processes()
+                    continue
+
+                if sig not in self.SIGNALS.keys():
+                    logger.info("Ignoring unknown signal: %d", sig)
+                    continue
+
+                handler = self.signal_handler(sig)
+                if handler is None:
+                    logger.info("Unhandled signal: %s", self.SIGNALS.get(sig))
+                    continue
+
+                handler()
+        except StopIteration:
+            self.halt()
+        except Exception:
+            logger.info("Unhandled exception in main loop", exc_info=True)
+            self.stop(signal.SIGTERM)
+            self.halt(ExitCode.DEFAULT_FAILURE)
+
+    def start(self) -> None:
+        self.pid = os.getpid()
+        logger.info("Started manager process [%d]", self.pid)
+        self.init_signals()
+
+    def spawn_processes(self) -> None:
+        for _ in range(self.config.workers - len(self.processes)):
+            self.spawn_process()
+
+    def spawn_process(self) -> None:
+        process = get_subprocess(self.config, target=self.target, sockets=self.sockets)
+        process.start()
+        self.processes.append(process)
+
+    def init_signals(self) -> None:
+        for s in self.SIGNALS.keys():
+            signal.signal(s, self._signal)
+        signal.signal(signal.SIGCHLD, self.handle_chld)
+
+    def _signal(self, sig: signal.Signals, frame: FrameType) -> None:
+        self.sig_queue.put(sig)
+
+    def handle_int(self) -> None:
+        self.stop(signal.SIGINT)
+        raise StopIteration
+
+    def handle_term(self) -> None:
+        self.stop(signal.SIGTERM)
+        raise StopIteration
+
+    def handle_chld(self, sig: signal.Signals, frame: FrameType) -> None:
+        self.reap_processes()
+
+    def signal_handler(self, sig: signal.Signals) -> Optional[Callable[..., None]]:
+        sig_name = self.SIGNALS.get(sig)
+        return getattr(self, f"handle_{sig_name}", None)
+
+    def reap_processes(self) -> None:
+        for process in self.processes:
+            if not process.is_alive():
+                self.processes.remove(process)
+
+    def halt(self, exit_code: int = ExitCode.OK) -> None:
+        logger.info("Stopping parent process [%d]", self.pid)
+        sys.exit(exit_code)
+
+    def kill_processes(self, sig: signal.Signals) -> None:
+        for process in self.processes:
+            self.kill_process(process, sig)
+
+    def kill_process(self, process: SpawnProcess, sig: signal.Signals) -> None:
+        os.kill(process.pid, sig)
+
+    def wait_timeout(self) -> None:
+        limit = time.time() + self.GRACEFUL_TIMEOUT
+        while self.processes and time.time() < limit:
+            time.sleep(0.1)
+
+    def stop(self, sig: signal.Signals) -> None:
+        self.kill_processes(sig)
+        self.wait_timeout()
+        self.kill_processes(signal.SIGKILL)
+
+        for sock in self.sockets:
+            sock.close()

uvicorn/workers.py

@@ -2,6 +2,7 @@
 import logging
 import signal
 import sys
+import warnings
 from typing import Any
 
 from gunicorn.arbiter import Arbiter
@@ -10,6 +11,13 @@
 from uvicorn.config import Config
 from uvicorn.main import Server
 
+warnings.warn(
+    "'workers' module is deprecated since uvicorn 0.16.0, "
+    "and it will be removed in a future release. "
+    "See https://github.com/encode/uvicorn/pull/1205.",
+    DeprecationWarning,
+)
+
 
 class UvicornWorker(Worker):
     """