优雅关闭：停止准入，再排空工作

from concurrent.futures import ThreadPoolExecutor
from http.client import HTTPConnection, RemoteDisconnected
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import json
import selectors
import signal
import subprocess
import sys
from threading import Condition, Event, Thread
import time


def child(grace):
    state = {"draining": False, "active": 0}
    condition, release = Condition(), Event()
    stop_requested = False

    def stop(signum, frame):
        nonlocal stop_requested
        stop_requested = True  # No lock, I/O, or shutdown call in the signal handler.

    class Handler(BaseHTTPRequestHandler):
        def log_message(self, *args):
            pass

        def reply(self, status, value):
            body = json.dumps(value).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json")
            self.send_header("Content-Length", str(len(body)))
            self.send_header("Cache-Control", "no-store")
            self.end_headers()
            self.wfile.write(body)

        def do_GET(self):
            if self.path == "/state":
                with condition:
                    snapshot = dict(state)
                self.reply(200, snapshot)
            elif self.path == "/readyz":
                with condition:
                    ready = not state["draining"]
                self.reply(200 if ready else 503, {"ready": ready})
            elif self.path == "/work":
                with condition:
                    admitted = not state["draining"]
                    if admitted:
                        state["active"] += 1
                if not admitted:
                    self.reply(503, {"error": "draining"})
                    return
                try:
                    if release.wait(timeout=5):
                        self.reply(200, {"done": True})
                    else:
                        self.reply(503, {"error": "fixture-timeout"})
                finally:
                    with condition:
                        state["active"] -= 1
                        condition.notify_all()
            else:
                self.reply(404, {"error": "not-found"})

        def do_POST(self):
            if self.path == "/release":  # Loopback-only test coordination, not workload admission.
                self.reply(200, {"released": True})
                release.set()  # Finish the control reply before work can complete and the child exits.
            else:
                self.reply(404, {"error": "not-found"})

    signal.signal(signal.SIGTERM, stop)
    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
    server.daemon_threads = True  # Explicitly permits abandonment in the deadline case.
    serving = Thread(target=server.serve_forever, kwargs={"poll_interval": 0.01})
    serving.start()
    print(json.dumps({"port": server.server_port}), flush=True)
    while not stop_requested:
        time.sleep(0.005)
    deadline = time.monotonic() + grace
    with condition:
        state["draining"] = True
        while state["active"] and time.monotonic() < deadline:
            condition.wait(timeout=max(0, deadline - time.monotonic()))
        unfinished = state["active"]
    server.shutdown()  # Called outside the thread running serve_forever.
    serving.join(timeout=2)
    server.server_close()  # Does not wait for daemon request threads.
    assert not serving.is_alive()
    print(json.dumps({"unfinished": unfinished, "deadline_expired": unfinished > 0}), flush=True)
    return 3 if unfinished else 0  # Fixture policy, not a universal exit-code convention.


def request(port, route, method="GET"):
    connection = HTTPConnection("127.0.0.1", port, timeout=3)
    try:
        connection.request(method, route)
        response = connection.getresponse()
        return response.status, json.loads(response.read())
    finally:
        connection.close()


def wait_state(port, predicate):
    deadline = time.monotonic() + 3
    while time.monotonic() < deadline:
        status, state = request(port, "/state")
        assert status == 200
        if predicate(state):
            return state
        time.sleep(0.01)
    raise AssertionError("child did not reach the expected state")


def exercise(finish):
    process = subprocess.Popen(
        [sys.executable, __file__, "child", "3" if finish else "1"],
        stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
    )
    try:
        with selectors.DefaultSelector() as selector:
            selector.register(process.stdout, selectors.EVENT_READ)
            assert selector.select(timeout=5), "child did not announce its local port"
        port = json.loads(process.stdout.readline())["port"]
        assert request(port, "/readyz") == (200, {"ready": True})
        with ThreadPoolExecutor(max_workers=1) as pool:
            active = pool.submit(request, port, "/work")
            wait_state(port, lambda state: state["active"] == 1)
            process.send_signal(signal.SIGTERM)
            wait_state(port, lambda state: state["draining"])
            assert request(port, "/readyz") == (503, {"ready": False})
            assert request(port, "/work") == (503, {"error": "draining"})
            assert request(port, "/state")[1]["active"] == 1
            if finish:
                assert request(port, "/release", "POST")[0] == 200
                assert active.result(timeout=4) == (200, {"done": True})
            output, errors = process.communicate(timeout=5)
            assert errors == ""
            record = json.loads(output)
            assert process.returncode == (0 if finish else 3)
            assert record == {"unfinished": 0 if finish else 1, "deadline_expired": not finish}
            if not finish:
                try:
                    active.result(timeout=4)
                except (OSError, RemoteDisconnected):
                    pass
                else:
                    raise AssertionError("abandoned work must not report a successful reply")
        print("completed drain" if finish else "expired drain", record, "exit", process.returncode)
    finally:
        if process.poll() is None:
            process.kill()  # Safety cleanup targets only this fixture's child.
            process.wait(timeout=5)
        process.stdout.close()
        process.stderr.close()


if __name__ == "__main__":
    if sys.argv[1:2] == ["child"]:
        raise SystemExit(child(float(sys.argv[2])))
    exercise(True)
    exercise(False)