systemdの再起動中、ソケットアクティベーションなら接続は拒否されず、待たされる

#!/usr/bin/env python3
"""A small line-based server that can be replaced while it serves.
Env: VERSION (label in replies), WORK (seconds of "work" per request), REUSEPORT=1 (bind with SO_REUSEPORT),
     EXTEND=1 (ask the service manager for more stop time while requests are still running).
Under socket activation (LISTEN_FDS) the listening socket is inherited and never re-created."""
import os, signal, socket, sys, threading, time

VERSION = os.environ.get("VERSION", "v1")
WORK = float(os.environ.get("WORK", "0"))
PORT = int(os.environ.get("PORT", "9100"))

def notify(msg):                                   # minimal sd_notify(3)
    path = os.environ.get("NOTIFY_SOCKET")
    if not path: return
    s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
    s.connect("\0" + path[1:] if path[0] == "@" else path); s.send(msg.encode()); s.close()

def listener():
    if os.environ.get("LISTEN_PID") == str(os.getpid()) and int(os.environ.get("LISTEN_FDS", "0")) >= 1:
        return socket.socket(fileno=3), "inherited from the service manager"
    s = socket.socket(); s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    if os.environ.get("REUSEPORT"): s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)
    s.bind(("127.0.0.1", PORT)); s.listen(128)
    return s, "bound by myself"

lock = threading.Lock(); inflight = 0; stopping = False

def handle(c):
    global inflight
    try:
        c.recv(100); time.sleep(WORK); c.sendall(f"{VERSION} {os.getpid()}\n".encode())
    except OSError: pass
    finally:
        c.close()
        with lock: inflight -= 1

def on_term(*_):
    global stopping; stopping = True
signal.signal(signal.SIGTERM, on_term)

srv, how = listener(); srv.settimeout(0.05)
print(f"{VERSION} pid={os.getpid()} listening ({how})", flush=True)
notify("READY=1")
while not stopping:
    try: c, _ = srv.accept()
    except socket.timeout: continue
    with lock: inflight += 1
    threading.Thread(target=handle, args=(c,), daemon=True).start()
srv.close()                                         # stop accepting, then drain
print(f"{VERSION} pid={os.getpid()} SIGTERM: draining {inflight} request(s)", flush=True)
notify("STOPPING=1")
while inflight:
    if os.environ.get("EXTEND"): notify("EXTEND_TIMEOUT_USEC=2000000")
    time.sleep(0.2)
print(f"{VERSION} pid={os.getpid()} drained, exiting", flush=True)

#!/usr/bin/env python3
"""Open a new connection ~100x/s from 4 threads for DURATION seconds; count what happened."""
import collections, socket, sys, threading, time
DURATION = float(sys.argv[1]); PORT = 9100
res = collections.Counter(); first_err = []; slowest = [0.0]; t0 = time.monotonic(); lock = threading.Lock()
def one():
    c = socket.socket(); c.settimeout(8); t = time.monotonic()
    try:
        c.connect(("127.0.0.1", PORT)); c.sendall(b"GET\n")
        data = c.recv(100)
        out = data.decode().split()[0] if data else "EOF-without-reply"
    except Exception as e: out = type(e).__name__
    finally: c.close()
    with lock:
        slowest[0] = max(slowest[0], time.monotonic() - t)
        res[out] += 1
        if out not in ("v1", "v2", "v3"): first_err.append(round(time.monotonic() - t0, 2))
def loop():
    while time.monotonic() - t0 < DURATION: one(); time.sleep(0.03)
ts = [threading.Thread(target=loop) for _ in range(4)]; [t.start() for t in ts]; [t.join() for t in ts]
print("  results:", dict(res), "| errors between", (min(first_err), max(first_err)) if first_err else "-", "s | slowest request", round(slowest[0]*1000), "ms")

#!/bin/bash
# usage: drain.sh <label> <TimeoutStopSec> <EXTEND 0|1>   : one 3-second request is in flight when the service is restarted
echo "=== $1  (TimeoutStopSec=$2, EXTEND_TIMEOUT_USEC sent: $( [ "$3" = 1 ] && echo yes || echo no ))"
systemctl stop web.socket web.service web-b.service 2>/dev/null
cat > /run/systemd/system/web.service <<EOT
[Unit]
Description=swapdemo drain
[Service]
Type=notify
Environment=VERSION=v1 WORK=3 $( [ "$3" = 1 ] && echo EXTEND=1 )
TimeoutStopSec=$2
ExecStart=/usr/bin/python3 /run/swapdemo/srv.py
EOT
systemctl daemon-reload; systemctl start web.service; sleep 0.5
python3 - <<'PY' &
import socket, time
t = time.monotonic(); c = socket.socket(); c.connect(("127.0.0.1", 9100)); c.sendall(b"GET\n")
try: d = c.recv(100); print(f"  client: {'reply ' + d.decode().strip() if d else 'connection closed with no reply'} after {time.monotonic()-t:.1f}s")
except Exception as e: print(f"  client: {type(e).__name__} after {time.monotonic()-t:.1f}s")
PY
P=$!
sleep 1; t=$(date +%s.%N); systemctl restart web.service; echo "  restart returned after $(python3 -c "import time;print(f'{time.time()-$t:.1f}')")s"
wait $P
journalctl -u web --no-pager -o cat --since "-8s" | grep -E 'SIGTERM|drained|killed|timed out|Failed|SIGKILL|signal' | sed 's/^/  journal: /'

=== default timeout  (TimeoutStopSec=90, EXTEND_TIMEOUT_USEC sent: no)
  client: reply v1 47537 after 3.0s
  restart returned after 2.1s
=== short timeout, no extension  (TimeoutStopSec=1, EXTEND_TIMEOUT_USEC sent: no)
  client: connection closed with no reply after 2.1s
  restart returned after 1.2s
=== short timeout, extended while draining  (TimeoutStopSec=1, EXTEND_TIMEOUT_USEC sent: yes)
  client: reply v1 47611 after 3.0s
  restart returned after 2.1s

#!/bin/bash
# plain service, bind itself
cat > /run/systemd/system/web.service <<EOT
[Unit]
Description=swapdemo app
[Service]
Type=notify
Environment=VERSION=v1 REUSEPORT=1
ExecStart=/usr/bin/python3 /run/swapdemo/srv.py
EOT
cat > /run/systemd/system/web-b.service <<EOT
[Unit]
Description=swapdemo app (new version, same port)
[Service]
Type=notify
Environment=VERSION=v2 REUSEPORT=1
ExecStart=/usr/bin/python3 /run/swapdemo/srv.py
EOT
cat > /run/systemd/system/web.socket <<EOT
[Socket]
ListenStream=127.0.0.1:9100
EOT
systemctl daemon-reload

#!/bin/bash
# usage: run.sh <label> '<start command>' '<what to do at t=2s>'
echo "=== $1"
systemctl stop web.socket web.service web-b.service 2>/dev/null; sleep 0.5
eval "$2"; sleep 1
python3 /run/swapdemo/hammer.py 5 &
H=$!
sleep 2
eval "$3"
wait $H
echo "  journal:"; journalctl -u web -u web-b --no-pager -o cat --since "-12s" 2>/dev/null | grep -v -E 'Started|Stopped|Starting|Stopping|Deactivated|Consumed|Succeeded|Finished' | sed 's/^/    /'

#!/bin/bash
# usage: bench.sh plain|overlap|socket [rounds]     (units.sh must have been run once; run.sh and hammer.py are in the same directory)
D=$(dirname "$0")
for i in $(seq 1 "${2:-5}"); do
  case $1 in
    plain)   $D/run.sh "plain restart, run $i"          'systemctl start web.service'  'systemctl restart web.service' ;;
    overlap) $D/run.sh "start v2 beside v1, run $i"     'systemctl start web.service'  'systemctl start web-b.service; sleep 0.3; systemctl stop web.service' ;;
    socket)  $D/run.sh "socket activation, run $i"      'systemctl start web.socket'   'systemctl restart web.service' ;;
  esac
done

#!/bin/bash
# usage: migrate.sh <0|1>      Run as root inside a fresh network namespace, e.g.
#   ip netns add mig; ip netns exec mig bash migrate.sh 0
# Sets net.ipv4.tcp_migrate_req (per network namespace), then replaces v1 by v2 eight times under load,
# with both listening on one port via SO_REUSEPORT. srv.py and hammer.py must be in the same directory.
HERE=$(cd "$(dirname "$0")" && pwd)
ip link set lo up
sysctl -qw net.ipv4.tcp_migrate_req=$1
echo "tcp_migrate_req=$(cat /proc/sys/net/ipv4/tcp_migrate_req)"
for i in 1 2 3 4 5 6 7 8; do
  VERSION=v1 REUSEPORT=1 python3 $HERE/srv.py >/dev/null & P1=$!
  sleep 0.7
  python3 $HERE/hammer.py 4 & H=$!
  sleep 2
  VERSION=v2 REUSEPORT=1 python3 $HERE/srv.py >/dev/null & P2=$!
  sleep 0.3
  kill -TERM $P1                       # v1 closes its listener; v2 keeps listening
  wait $H
  kill -TERM $P2; wait $P1 $P2 2>/dev/null
done

#!/bin/bash
# usage: swap.sh <release-name>      Releases live in /run/swapdemo/rel/<name>/srv.py ; /run/swapdemo/current is a symlink.
set -u
new=$1; base=/run/swapdemo; unit=web.service
old=$(basename "$(readlink $base/current)")
echo "swap: $old -> $new"
healthy() {   # healthy = the service answers one request AND says it is the release we expect
  for i in $(seq 1 20); do
    out=$(python3 - <<PY 2>/dev/null
import socket
c = socket.socket(); c.settimeout(1); c.connect(("127.0.0.1", 9100)); c.sendall(b"GET\n"); print(c.recv(100).decode().split()[0])
PY
)
    [ "$out" = "$1" ] && return 0
    sleep 0.25
  done
  return 1
}
ln -sfn "$base/rel/$new" "$base/current.tmp" && mv -T "$base/current.tmp" "$base/current"   # atomic switch
systemctl restart $unit 2>/dev/null
if healthy "$new"; then echo "swap: $new is healthy, keeping it"; exit 0; fi
echo "swap: $new did not become healthy within 5 s, rolling back to $old"
ln -sfn "$base/rel/$old" "$base/current.tmp" && mv -T "$base/current.tmp" "$base/current"
systemctl reset-failed $unit 2>/dev/null; systemctl restart $unit
healthy "$old" && echo "swap: $old is serving again" || echo "swap: ROLLBACK FAILED"
exit 1

#!/bin/bash
b=/run/swapdemo; mkdir -p $b/rel/{v1,v2,v3,v4}
for v in v1 v2 v3 v4; do cp $b/srv.py $b/rel/$v/srv.py; sed -i "s/^VERSION = .*/VERSION = \"$v\"/" $b/rel/$v/srv.py; done
# v3 crashes at start; v4 starts, but answers with the wrong label
sed -i 's/^srv, how = listener().*/import sys; sys.exit("v3 cannot start: bad config")/' $b/rel/v3/srv.py
sed -i 's/^VERSION = .*/VERSION = "v1-but-actually-broken"/' $b/rel/v4/srv.py
ln -sfn $b/rel/v1 $b/current
cat > /run/systemd/system/web.service <<EOT
[Unit]
Description=swapdemo release swap
[Service]
Type=notify
Environment=REUSEPORT=1
ExecStart=/usr/bin/python3 /run/swapdemo/current/srv.py
TimeoutStartSec=3
EOT
systemctl daemon-reload; systemctl stop web.socket web-b.service 2>/dev/null; systemctl restart web.service

swap: v1 -> v2
swap: v2 is healthy, keeping it
swap: v2 -> v3
swap: v3 did not become healthy within 5 s, rolling back to v2
swap: v2 is serving again
swap: v2 -> v4
swap: v4 did not become healthy within 5 s, rolling back to v2
swap: v2 is serving again