$ gunicorn -w 4 -k uvicorn_worker.UvicornWorker app:app

[pid 160604] <... epoll_wait resumed>[{events=EPOLLIN, data={u32=5, u64=139977279143941}}], 2, 20000) = 1
[pid 160603] <... epoll_wait resumed>[{events=EPOLLIN, data={u32=5, u64=139977279143941}}], 2, 20000) = 1
[pid 160602] <... epoll_wait resumed>[{events=EPOLLIN, data={u32=5, u64=139977279143941}}], 2, 20000) = 1
[pid 160601] <... epoll_wait resumed>[{events=EPOLLIN, data={u32=5, u64=139977279143941}}], 2, 20000) = 1
[pid 160604] accept4(5,  <unfinished ...>
[pid 160603] accept4(5,  <unfinished ...>
[pid 160602] accept4(5,  <unfinished ...>
[pid 160601] accept4(5,  <unfinished ...>
[pid 160604] <... accept4 resumed>{sa_family=AF_INET, sin_port=htons(43136), sin_addr=inet_addr("127.0.0.1")}, [16], SOCK_CLOEXEC) = 12
[pid 160603] <... accept4 resumed>0x7ffe449e2d50, [16], SOCK_CLOEXEC) = -1 EAGAIN (Resource temporarily unavailable)
[pid 160602] <... accept4 resumed>0x7ffe449e2d50, [16], SOCK_CLOEXEC) = -1 EAGAIN (Resource temporarily unavailable)
[pid 160601] <... accept4 resumed>0x7ffe449e2d50, [16], SOCK_CLOEXEC) = -1 EAGAIN (Resource temporarily unavailable)

import time
time.sleep(10)

async def app(scope, receive, send):
    assert scope['type'] == 'http'
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [(b'content-type', b'text/plain')],
    })
    await send({
        'type': 'http.response.body',
        'body': b"Hello world!",
    })

$ uvicorn --workers 4 app:app
$ gunicorn -w 4 -k uvicorn_worker.UvicornWorker app:app

from collections import defaultdict
import gc
import sys

gc.collect()
gc_objects = gc.get_objects()
total = 0
counts = defaultdict(int)
for obj in gc_objects:
    size = sys.getsizeof(obj)
    counts[size] += 1
    total += size

print("GC-tracked objects")
print("  count: {:,}".format(len(gc_objects)))
print("  total size: {:,} bytes".format(total))
if gc_objects:
    print("  average: {}".format(total / len(gc_objects)))
print("  distribution: {}".format(sorted(counts.items())))

#!/usr/bin/env -S systemd-run --user --scope -p MemoryMax=1200M python
import gc
import os
import sys
from typing import Self
from dataclasses import dataclass

@dataclass(frozen=True)
class Node:
    cdr: Self | None

BIG_DATA = None
for i in range(12 * 1024 * 1024):
    BIG_DATA = Node(BIG_DATA)

gc.freeze()
try:
    gc.immortalize_heap()
except AttributeError:
    print(
        "couldn't immortalize heap: will consume extra memory",
        file=sys.stderr
    )
os.fork()

# As you can see, we never mutate any part of this linked list, but stock
# CPython will mutate its reference counts, which is enough to cause double
# the memory usage.
head = BIG_DATA
while head is not None:
    head = head.cdr

# Copyright 2024 Josh Snyder
# Licensed under https://www.apache.org/licenses/LICENSE-2.0.txt
from functools import partial
from pathlib import Path
from typing import Callable, Iterator, Iterable
import asyncio
import contextlib
import dataclasses
import gc
import os
import shutil
import signal
import socket
import sys
import tempfile
import time
import uvicorn


# -- the webapp --

async def app(scope, receive, send):
    """A basic ASGI app."""

    if scope["type"] == "lifespan":
        while True:
            message = await receive()
            if message["type"] == "lifespan.startup":
                await send({"type": "lifespan.startup.complete"})
            elif message["type"] == "lifespan.shutdown":
                await send({"type": "lifespan.shutdown.complete"})

    if scope["type"] == "http":
        await send(
            {
                "type": "http.response.start",
                "status": 200,
                "headers": [(b"content-type", b"text/plain")],
            }
        )
        await send(
            {
                "type": "http.response.body",
                "body": b"Hello world!\n",
            }
        )
        return

    raise NotImplementedError("scope.type", scope["type"])


uvicorn_server = uvicorn.Server(uvicorn.Config(app, lifespan="on"))

# -- basic utilities --

# All of the other signals get the default disposition
HANDLED_SIGNALS = set([signal.SIGCHLD, signal.SIGHUP])


@contextlib.contextmanager
def mask_signals(sigset):
    old_mask = signal.pthread_sigmask(signal.SIG_BLOCK, sigset)
    try:
        yield
    finally:
        signal.pthread_sigmask(signal.SIG_SETMASK, old_mask)


@contextlib.contextmanager
def tempdir_for_pid(pid: int) -> Iterator[Path]:
    """A tempdir that only cleans up when it hasn't been forked"""
    path = Path(tempfile.mkdtemp())
    try:
        yield path
    finally:
        if os.getpid() == pid:
            shutil.rmtree(path)

# -- management of child processes --

@dataclasses.dataclass(frozen=True)
class ChildDispatch:
    """Encapsulates a task for the child to run."""

    # The file descriptors that the child should keep open
    fds_to_keep: tuple[int, ...]
    # What to actually run
    call: Callable[[], int]


def reap_zombies(pids: dict[int, ChildDispatch]) -> ChildDispatch | None:
    """If any of our child proceses dies (becomes a zombie), this will reap
    them and restart the appropriate callable."""

    while True:
        # Wait for child processes. There must be at least one, but we
        # don't know how many there will be.
        pid, status, rusage = os.wait3(os.WNOHANG)
        if pid == 0:
            break

        try:
            dispatch = pids.pop(pid)
        except KeyError:
            # We reaped a process that we didn't launch
            continue

        dispatch = dataclasses.replace(
            dispatch,
            # Change the started_at kwarg on the previous partial
            call=partial(dispatch.call, started_at=time.perf_counter()),
        )

        # Relaunch any children that died
        pid = os.fork()
        if pid == 0:
            return dispatch

        pids[pid] = dispatch
        print(f"Spawned {pid}", file=sys.stderr)

    return None


def poll_once(pids: dict[int, ChildDispatch]) -> ChildDispatch | None:
    # We could also use signalfd and a selectors.poll/epoll object
    seen_signal = signal.sigwait(HANDLED_SIGNALS)

    if seen_signal == signal.SIGHUP:
        for pid in pids:
            os.kill(pid, signal.SIGHUP)

    if seen_signal == signal.SIGCHLD:
        # The return is needed for the case where reap_zombies decides that it
        # needs to fork.
        return reap_zombies(pids)

    return None


def bind_sockets(path: Path, count: int) -> list[int]:
    ret = []
    for i in range(count):
        name = str(path / f"{i:03x}")
        s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        try:
            s.bind(name)
            # We start listening immediately, but nobody will accept() until
            # we fork into processes.
            s.listen(1)

            ret.append(s.detach())
        finally:
            s.close()

    return ret



def fork_children(
    sockets: list[int], self_pidfd: int
) -> ChildDispatch | dict[int, ChildDispatch]:
    """Launches child processes.

    If we are now in a child process, we return a ChildDispatch.
    Otherwise we return a dict[pid, ChildDispatch] in case a child needs to
    be respawned."""

    ret = dict()
    for sock in sockets:
        # the format we return is:
        # (file descriptors to keep, callable to run)
        child_ret = ChildDispatch(
            (self_pidfd, sock),
            partial(
                child_process, self_pidfd, sock, started_at=time.perf_counter()
            ),
        )
        pid = os.fork()
        if pid == 0:
            return child_ret

        ret[pid] = child_ret

    return ret


def spawn_loop(stack: contextlib.ExitStack) -> ChildDispatch:
    # Mask the normal delivery of signals
    stack.enter_context(mask_signals(HANDLED_SIGNALS))
    my_pid = os.getpid()

    # Set up a pidfd. We'll give this pidfd to our children, and they'll use
    # it to monitor us (the parent process)
    self_pidfd = os.pidfd_open(my_pid)

    path = stack.enter_context(tempdir_for_pid(my_pid))
    cpu_count = os.cpu_count()
    assert cpu_count
    sockets = bind_sockets(path, cpu_count)
    fork_result = fork_children(sockets, self_pidfd)
    if isinstance(fork_result, ChildDispatch):
        return fork_result

    while True:
        ret = poll_once(fork_result)
        if ret is not None:
            return ret

async def run_uvicorn(listen_fd: int, started_at: float) -> None:
    # Our actual activity. This runs in the child (worker) processes.
    sock = socket.socket(fileno=listen_fd)
    elapsed = time.perf_counter() - started_at
    name = sock.getsockname()
    print(
        f"PID {os.getpid()} fork took {elapsed} seconds.\n"
        f"Query me at `curl --unix-socket \"{name}\" http://localhost`",
        file=sys.stderr
    )
    await uvicorn_server.serve([sock])


async def watch_parent(parent_fd: int) -> None:
    """Watches an fd `parent_fd`, and exits when it reports readiness.

    The intent of this function is to prevent us from outlasting our parent
    process.
    """
    loop = asyncio.get_running_loop()
    future = loop.create_future()
    loop.add_reader(parent_fd, future.set_result, None)
    try:
        await future
    finally:
        loop.remove_reader(parent_fd)


async def async_child_process(
    parent_fd: int, listen_fd: int, *, started_at: float
) -> int:
    loop = asyncio.get_running_loop()
    tasks = [
        loop.create_task(watch_parent(parent_fd)),
        loop.create_task(run_uvicorn(listen_fd, started_at)),
    ]
    await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
    return 0


def child_process(parent_fd: int, listen_fd: int, *, started_at: float) -> int:
    # The child process's contract is to:
    # 1) monitor the parent_fd for readiness, which indicates the parent has
    # exited.
    # 2) handle connections arriving to listen_fd.
    try:
        return asyncio.run(
            async_child_process(parent_fd, listen_fd, started_at=started_at)
        )
    except KeyboardInterrupt:
        pass

    # This will skip any atexit handlers
    signal.signal(signal.SIGINT, signal.SIG_DFL)
    os.kill(os.getpid(), signal.SIGINT)
    return 0


def close_fds(keep: Iterable[int] = ()) -> None:
    """Closes all of the FDs not needed to do our job."""
    # This is necessary because we inherit all of the parent's FDs, but we
    # only want to be responsible for the ones it has assigned us.

    keep = iter(sorted(keep))
    from_fd = 3
    for to_fd in keep:
        # os.closerange is exclusive of its second argument, matching range()
        if to_fd > from_fd:
            os.closerange(from_fd, to_fd)
        from_fd = to_fd + 1
    else:
        os.closerange(from_fd, 0)


def wait_for_work() -> Callable[[], int]:
    # Runs in the main process and waits for it to fork into a child process.
    # It'll return a callback when it does, which we execute as our child
    # process.
    try:
        with contextlib.ExitStack() as stack:
            dispatch = spawn_loop(stack)
            if dispatch is None:
                # No work to do. We're done!
                raise SystemExit

        close_fds(dispatch.fds_to_keep)
        return dispatch.call
    except KeyboardInterrupt:
        raise


def main() -> int:
    # It's not sufficient to prevent de-CoWing, but it's slightly worthwhile to
    # freeze the existing heap.
    gc.freeze()
    # By exiting all the way out to the outermost stack frame, we ensure that
    # none of the parent's code (within `wait_for_work()`) is seen in stack
    # traces of the child, which makes them simpler to understand.
    callback = wait_for_work()
    return callback()


if __name__ == "__main__":
    raise SystemExit(main())