fix(Main): 修复热重载卡死事件循环并改造为单实例

--reload 改为纯通知指令:只通过 127.0.0.1:57321 控制端口通知既有实例重启,
本进程立即退出;找不到实例也直接退出,不再新开服务器抢占 80/443。

根因修复:monkey.patch_all() 把 threading 一并打补丁,导致 watchdog 的
Observer 变成 gevent 协程,其阻塞的 ReadDirectoryChangesW 不让出 hub,
整个事件循环被卡死(端口在监听却不再 accept)。改为轮询 .py 的 mtime/size,
并从 requirements.txt 移除 watchdog。

另修两处启动隐患:预置 WSGI SERVER_NAME 跳过 gevent 对 0.0.0.0 的反向 DNS
(实测 1.3s 起,期间服务器不 started);WaitForPortFree 用裸 bind 探测端口,
避免 SO_REUSEADDR 造成的"假绑定→静默停服"。
This commit is contained in:
TermiNexus committed 2026-10-01 21:16:12 +08:00
1 parent 45a710783b
commit 5a8d29d31a
2 files changed
+373 -74

No files matched your search

+371 -73
View File
@@ -8,6 +8,7 @@ import logging.handlers
import datetime
import traceback
import signal
import socket
import subprocess
from pathlib import Path
@@ -29,6 +30,21 @@ CPU_CORES = os.cpu_count() or 4
MAX_COROUTINES = CPU_CORES * 50
MAX_THREADS = CPU_CORES * 2
# 单实例控制端口(仅回环):全局只允许一个 Main.py 实例,同时作为 --reload 的通知通道
# 行协议:PING\n → PONG\n ;RELOAD [json-argv]\n → OK\n(收到后由既有实例按该参数自重启)
CONTROL_HOST = '127.0.0.1'
CONTROL_PORT = int(os.environ.get('MAIN_CONTROL_PORT') or 57321)
# 抢绑端口的宽限时间:覆盖"上一个实例刚 execv、端口尚未释放"的瞬间
INSTANCE_TAKEOVER_SECONDS = 5.0
# 抢绑重试间隔 / 探测既有实例的连接与应答超时(都要短,避免启动慢)
INSTANCE_RETRY_INTERVAL = 0.2
CONTROL_TIMEOUT = 0.5
# WSGI 环境的 SERVER_NAME:必须预置,否则 gevent 的 WSGIServer.init_socket() 会对监听地址
# 做反向 DNS(socket.getfqdn('0.0.0.0'))—— 实测本机要 1.3s 起,网络抖动时更久,期间服务器
# 一直没 started(表现为"启动很慢"甚至误判成启动失败)。用本机名预置即可跳过该查询(纯本地调用,不走 DNS)。
SERVER_NAME = socket.gethostname()
from gvsds.secrets import get as _secret_get
SQL_IP = _secret_get('database', 'host', fallback='127.0.0.1')
@@ -554,85 +570,289 @@ def get_django_wsgi_application():
class HotReloader:
def __init__(self, watch_dirs=None, extensions=None, debounce=1.0):
"""文件变更探测(轮询 .py 的 mtime/size,**不用 watchdog**)
**为什么不用 watchdog**:`monkey.patch_all()` 把 threading 一起打了补丁,watchdog 的
Observer/BaseThread 于是变成 gevent 协程(实测 MRO:BaseThread → gevent.threading.Thread);
它内部阻塞的 ReadDirectoryChangesW 是 Win32 阻塞调用、不会让出 hub,一个协程就能把整个
事件循环卡死 —— 端口全在监听却不再 accept,80/443 与控制端口一起静默失联。
轮询是无副作用的纯检查,代价可忽略(约 400 个文件),且不引入任何线程/dependency。
"""
def __init__(self, watch_dirs=None, extensions=None):
self.watch_dirs = watch_dirs or ['./gvsds']
self.extensions = extensions or {'.py'}
self.debounce = debounce
self._observer = None
self._changed = gevent.event.Event()
self._last_change_time = 0
self._lock = gevent.lock.BoundedSemaphore(1)
self._snapshot = {}
def _should_watch(self, path):
def _ShouldWatch(self, path):
ext = os.path.splitext(path)[1].lower()
if ext not in self.extensions:
return False
parts = os.path.normpath(path).split(os.sep)
if '__pycache__' in parts or '.git' in parts:
return False
return True
return '__pycache__' not in parts and '.git' not in parts
def _on_change(self, event):
if event.is_directory:
return
if not self._should_watch(event.src_path):
return
now = time.time()
with self._lock:
if now - self._last_change_time < self.debounce:
return
self._last_change_time = now
rel_path = os.path.relpath(event.src_path, '.')
logger.warning(f"[热重启] 检测到文件变更: {rel_path}", extra={'domain': 'system', 'status_code': 0})
self._changed.set()
def _Scan(self):
"""扫描监视目标,返回 {路径: (mtime, size)}(目录递归、单文件直取)"""
snapshot = {}
for target in self.watch_dirs:
node = Path(target)
if node.is_dir():
candidates = node.rglob('*')
elif node.is_file():
candidates = [node]
else:
continue
for item in candidates:
if not self._ShouldWatch(str(item)):
continue
try:
stat = item.stat()
except OSError:
continue
snapshot[str(item)] = (stat.st_mtime, stat.st_size)
return snapshot
def start(self):
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
"""记录初始快照作为基线(之后只对比变化)"""
self._snapshot = self._Scan()
logger.info(
f"[热重启] 文件监听已启动(轮询 mtime): {self.watch_dirs},共 {len(self._snapshot)} 个文件",
extra={'domain': 'system', 'status_code': 0},
)
class Handler(FileSystemEventHandler):
def __init__(self, callback):
super().__init__()
self.callback = callback
def on_modified(self, event):
self.callback(event)
def on_created(self, event):
self.callback(event)
self._observer = Observer()
handler = Handler(self._on_change)
for d in self.watch_dirs:
abs_dir = os.path.abspath(d)
if os.path.isdir(abs_dir):
self._observer.schedule(handler, abs_dir, recursive=True)
self._observer.daemon = True
self._observer.start()
logger.info(f"[热重启] 文件监听已启动: {self.watch_dirs}", extra={'domain': 'system', 'status_code': 0})
def stop(self):
if self._observer:
self._observer.stop()
self._observer.join(timeout=3)
def _Poll(self):
snapshot = self._Scan()
detail = None
for path, fingerprint in snapshot.items():
if self._snapshot.get(path) != fingerprint:
detail = os.path.relpath(path, '.')
break
if detail is None and len(snapshot) != len(self._snapshot):
detail = '文件新增或删除'
self._snapshot = snapshot
if detail is None:
return
logger.warning(f"[热重启] 检测到文件变更: {detail}", extra={'domain': 'system', 'status_code': 0})
self._changed.set()
def wait_for_change(self, timeout=None):
"""等一个轮询周期后检查一次变更;返回是否已置位"""
self._changed.clear()
return self._changed.wait(timeout)
gevent.sleep(timeout if timeout is not None else 1.0)
self._Poll()
return self._changed.is_set()
@property
def is_changed(self):
return self._changed.is_set()
class ServerManager:
class ControlServer:
"""单实例控制端口(仅 127.0.0.1)
两个职责:
1. **单实例闸门** —— 绑定该端口即"我是唯一实例",后来者绑不上就有据可查;
2. **热重载通道** —— 后来者先 PING 探测,再发 RELOAD 让既有实例自重启,
从而做到"全局只有一个 Main.py 实例,重启是通知而不是新起一个抢 80/443"。
行协议(客户端发一行,服务端回一行):
PING\\n → PONG\\n
RELOAD\\n → OK\\n(随后由主协程按本进程自己的启动参数重启,不在本协程里做)
两个必须遵守的实现约束:
- **不能设 SO_REUSEADDR** —— Windows 下它允许重复绑定同一端口,单实例闸门会失效;
- **连接就地处理、不另开协程** —— 消息只有几十字节,直接 accept 后同步收发即可;
一旦改成 spawn,处理协程若因任何原因没被调度,连接就会堆在 CloseWait 里占满
backlog,后来者被"连接拒绝",闸门与通知通道一起失效(实测踩过)。
"""
def __init__(self):
self.host = CONTROL_HOST
self.port = CONTROL_PORT
self.on_reload = None
self.socket = None
self.greenlet = None
self.last_error = None
def SetReloadHandler(self, handler):
self.on_reload = handler
return self
def TryBind(self):
"""尝试绑定控制端口;成功返回 True,被占用返回 False(原因存 last_error)"""
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.bind((self.host, self.port))
sock.listen(8)
except OSError as e:
self.last_error = e
try:
sock.close()
except Exception:
pass
return False
sock.settimeout(1)
self.socket = sock
return True
def start(self):
self.greenlet = gevent.spawn(self._Serve)
return self
def _Serve(self):
while True:
try:
conn, addr = self.socket.accept()
except socket.timeout:
continue
except (OSError, gevent.GreenletExit):
return
self._Handle(conn, addr)
def _Handle(self, conn, addr):
"""逐行收发(一个连接可以连发多条命令:PING 探活 + RELOAD 通知)"""
try:
conn.settimeout(CONTROL_TIMEOUT)
buffer = b''
while True:
try:
chunk = conn.recv(256)
except socket.timeout:
return
if not chunk:
return
buffer += chunk
while b'\n' in buffer:
line, buffer = buffer.split(b'\n', 1)
if not self._Dispatch(conn, addr, line.decode('utf-8', 'replace').strip()):
return
except Exception:
pass
finally:
try:
conn.close()
except Exception:
pass
def _Dispatch(self, conn, addr, line):
"""处理一条命令;返回 False 表示可以收尾(收到重启通知)"""
command = line.split(' ', 1)[0].upper()
if command == 'PING':
conn.sendall(b'PONG\n')
return True
if command in ('RELOAD', 'RESTART'):
logger.warning(f"[单实例] 收到来自 {addr[0]} 的重启通知", extra={'domain': 'system', 'status_code': 0})
conn.sendall(b'OK\n')
# 只置标志,真正重启交给主协程(本协程若在此 stop_all 会被连带停掉)
if self.on_reload:
self.on_reload()
return False
conn.sendall(b'ERR\n')
return True
def stop(self):
greenlet, self.greenlet = self.greenlet, None
if greenlet is not None:
try:
greenlet.kill()
except Exception:
pass
sock, self.socket = self.socket, None
if sock is not None:
try:
sock.close()
except Exception:
pass
def ProbeRunningInstance():
"""PING 探测是否已有 Main.py 实例(**不发重启指令**,只用于给出准确报错信息)"""
try:
with socket.create_connection((CONTROL_HOST, CONTROL_PORT), timeout=CONTROL_TIMEOUT) as conn:
conn.settimeout(CONTROL_TIMEOUT)
conn.sendall(b'PING\n')
return conn.recv(64).strip().upper().startswith(b'PONG')
except OSError:
return False
def NotifyRunningInstance():
"""探测既有 Main.py 实例并通知它重启;通知成功返回 True
连不上 / 不应答 / 回的既不是 PONG,都说明"没有既有实例",一律返回 False,
由调用方决定怎么办(`--reload` 直接退出,不带参数则继续抢绑端口)。
"""
try:
with socket.create_connection((CONTROL_HOST, CONTROL_PORT), timeout=CONTROL_TIMEOUT) as conn:
conn.settimeout(CONTROL_TIMEOUT)
conn.sendall(b'PING\n')
if not conn.recv(64).strip().upper().startswith(b'PONG'):
return False
conn.sendall(b'RELOAD\n')
conn.recv(64)
return True
except OSError:
return False
def WaitForPortFree(port, host='0.0.0.0', timeout=INSTANCE_TAKEOVER_SECONDS):
"""等待端口真正可绑(execv 重启时旧进程会短暂占着 80/443)
**探测用的临时 socket 不能设 SO_REUSEADDR**:Windows 下带该选项的 bind 在端口被占用时
也会"成功",于是新进程会在旧进程尚未退出时抢绑 80/443 —— 表面绑上了,实际没有任何
连接进来(静默停服)。必须用裸 bind 才能如实反映"是否仍被占用"。
"""
deadline = time.time() + timeout
while True:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
sock.bind((host, port))
return True
except OSError:
pass
finally:
try:
sock.close()
except Exception:
pass
if time.time() >= deadline:
return False
time.sleep(INSTANCE_RETRY_INTERVAL)
def AcquireInstanceControl():
"""抢绑单实例控制端口(仅"启动服务器"的路径调用):返回 (control, status)
status 取值:
'primary' —— 本进程拿到控制端口,是唯一实例(control 已绑定,尚未 start)
'occupied' —— 端口被占(既有实例或别的程序),调用方应报错退出
**顺序很关键:先抢绑端口(瞬时),绑不上再 PING 看是不是既有实例(≤0.5s)。**
反着来(先 PING 再绑)会在"上一个实例刚 execv、端口半开"时白等十几秒 —— 启动慢的根源。
"""
control = ControlServer()
deadline = time.time() + INSTANCE_TAKEOVER_SECONDS
while True:
if control.TryBind():
return control, 'primary'
if time.time() >= deadline:
return None, 'occupied'
time.sleep(INSTANCE_RETRY_INTERVAL)
class ServerManager:
def __init__(self, control=None):
global logger
if logger is None:
SetupLogging()
self.pool_manager = ConnectionPoolManager()
self.servers = []
# 控制端口(单实例闸门 + --reload 通知通道):由 __main__ 抢绑后传入
self.control_server = control
# 控制端口请求的重启信号:由控制协程置位,主协程执行重启
self.reload_signal = gevent.event.Event()
self.ssl_context = CreateSslContext(
os.environ.get('SSL_CERT_PATH', './SSL/gvsds.com_public.crt'),
os.environ.get('SSL_KEY_PATH', './SSL/gvsds.com.key')
@@ -646,7 +866,8 @@ class ServerManager:
server = CustomWSGIServer(
('0.0.0.0', port), RedirectApp,
spawn=self.pool_manager.coroutine_pool,
log=NoOpLog(), error_log=NoOpLog()
log=NoOpLog(), error_log=NoOpLog(),
environ={'SERVER_NAME': SERVER_NAME, 'SERVER_PORT': str(port)},
)
self.servers.append(('HTTP Redirect', port, server, None))
return self
@@ -660,6 +881,7 @@ class ServerManager:
'error_log': NoOpLog(),
'suppress_ragged_eofs': True,
'do_handshake_on_connect': False,
'environ': {'SERVER_NAME': SERVER_NAME, 'SERVER_PORT': str(port)},
}
if is_ssl:
kwargs['ssl_context'] = self.ssl_context
@@ -668,46 +890,95 @@ class ServerManager:
self.servers.append((app_name, port, server, None))
return self
def start_all(self, hot_reload=False):
def RequestReload(self):
"""控制端口收到重启通知:置位重启信号(真正的重启由主协程按本进程自身参数执行)"""
self.reload_signal.set()
def start_all(self):
logger.info("=== 启动 Django 服务器集群 ===", extra={'domain': 'system', 'status_code': 0})
logger.info(f"协程池: {MAX_COROUTINES} | 线程池: {MAX_THREADS}", extra={'domain': 'system', 'status_code': 0})
# 启动单实例控制端口:绑定动作已在 __main__ 的 AcquireInstanceControl() 完成,
# 这里只挂上重启回调并开接受循环(80/443 之前启动,闸门才算真正生效)
if self.control_server is None:
self.control_server = ControlServer()
if not self.control_server.TryBind():
logger.critical(
f"控制端口 {CONTROL_HOST}:{CONTROL_PORT} 绑定失败"
f"({self.control_server.last_error}):已有 Main.py 实例或该端口被占用,本进程退出",
extra={'domain': 'system', 'status_code': 0},
)
sys.exit(1)
self.control_server.SetReloadHandler(self.RequestReload).start()
logger.info(f"单实例控制端口: {CONTROL_HOST}:{CONTROL_PORT}", extra={'domain': 'system', 'status_code': 0})
from django.urls import get_resolver
resolver = get_resolver()
logger.info("=== Django URL 路由已加载 ===", extra={'domain': 'system', 'status_code': 0})
running_servers = []
for name, port, server, _ in self.servers:
# 等端口真正空闲再绑:execv 重启时旧进程可能还没退完,抢绑会变成"静默停服"
if not WaitForPortFree(port):
logger.critical(
f"{name} 端口 {port} 仍被占用,无法监听,本进程退出",
extra={'domain': 'system', 'status_code': 0},
)
sys.exit(1)
logger.info(f"启动 {name} 服务器 on 0.0.0.0:{port}", extra={'domain': 'system', 'status_code': 0})
gevent.sleep(0.1)
g = gevent.spawn(server.serve_forever)
running_servers.append((name, port, server, g))
self.servers = running_servers
# 绑成功与否只有 gevent 内部知道(异常发生在协程里、不进日志),这里复核一次。
# 用有界轮询而不是固定 sleep:固定等待在机器繁忙时会把"正在启动"误判成"启动失败"(实测踩过)。
deadline = time.time() + INSTANCE_TAKEOVER_SECONDS
while True:
pending = [item for item in self.servers if not getattr(item[2], 'started', False)]
if not pending:
break
if time.time() >= deadline:
for name, port, _server, _g in pending:
logger.critical(
f"{name} 服务器未成功监听 0.0.0.0:{port},本进程退出",
extra={'domain': 'system', 'status_code': 0},
)
sys.exit(1)
gevent.sleep(0.1)
reloader = None
if hot_reload:
reloader = HotReloader(watch_dirs=['./gvsds', './Main.py'])
reloader.start()
# 文件热重载常开:监视 gvsds(业务代码)+ gvsdsdk(SDK,业务大量逻辑在这里)
# + Main.py 自身,任一 .py 变更即 os.execv 全量重启。
reloader = HotReloader(watch_dirs=['./gvsds', './gvsdsdk', './Main.py'])
reloader.start()
try:
while True:
if hot_reload and reloader:
reloader.wait_for_change(timeout=1)
if reloader.is_changed:
logger.warning("[热重启] 正在重启服务器...", extra={'domain': 'system', 'status_code': 0})
self.stop_all()
reloader.stop()
os.execv(sys.executable, [sys.executable] + sys.argv)
# 每轮最多等 1s:既轮询文件变更,也等控制端口(单实例通知)的重启信号
reloader.wait_for_change(timeout=1)
if self.reload_signal.is_set():
logger.warning("[单实例] 收到 TCP 热重载通知,正在重启服务器...", extra={'domain': 'system', 'status_code': 0})
elif reloader.is_changed:
logger.warning("[热重启] 检测到文件变更,正在重启服务器...", extra={'domain': 'system', 'status_code': 0})
else:
gevent.sleep(3600)
continue
self.stop_all()
# 重启沿用本进程自己的启动参数(--reload 那条路径不会走到这里,它只负责通知)
os.execv(sys.executable, [sys.executable] + sys.argv)
except KeyboardInterrupt:
if reloader:
reloader.stop()
self.stop_all()
def stop_all(self):
logger.info("=== 停止所有服务器 ===", extra={'domain': 'system', 'status_code': 0})
# 先释放单实例控制端口:execv 重启时新进程要能立刻绑上它
if self.control_server is not None:
try:
self.control_server.stop()
except:
pass
self.control_server = None
for item in self.servers:
if len(item) == 4:
_, _, server, g = item
@@ -727,17 +998,44 @@ if __name__ == "__main__":
colorama.init(autoreset=True)
os.environ['GEVENT_THREADPOOL_SIZE'] = str(MAX_THREADS)
hot_reload = '--reload' in sys.argv
server_manager = None
try:
logger = SetupLogging()
if hot_reload:
logger.info("[热重启] 模式已启用 (--reload)", extra={'domain': 'system', 'status_code': 0})
server_manager = ServerManager()
# `--reload` 是**纯指令**:只通知正在运行的实例重启,自己不新开服务器。
# 有实例 → 通知并退出;没有实例 → 直接退出(要启动服务器请不带参数运行)。
if '--reload' in sys.argv:
if NotifyRunningInstance():
logger.warning(
f"[单实例] 已通知正在运行的 Main.py 实例重启,本进程退出"
f"(控制端口 {CONTROL_HOST}:{CONTROL_PORT})",
extra={'domain': 'system', 'status_code': 0},
)
else:
logger.warning(
"[单实例] 没有正在运行的 Main.py 实例,--reload 无事可做,本进程退出"
"(要启动服务器请直接运行 python Main.py)",
extra={'domain': 'system', 'status_code': 0},
)
sys.exit(0)
# 单实例:全局只允许一个 Main.py。先抢绑控制端口(瞬时),抢不到按占用报错退出。
control, status = AcquireInstanceControl()
if status != 'primary':
if ProbeRunningInstance():
detail = "已有 Main.py 实例在运行(要让它重启请执行 python Main.py --reload)"
else:
detail = f"该端口被其它程序占用(可换端口:设置环境变量 MAIN_CONTROL_PORT)"
logger.critical(
f"控制端口 {CONTROL_HOST}:{CONTROL_PORT} 绑定失败:{detail},本进程退出",
extra={'domain': 'system', 'status_code': 0},
)
sys.exit(1)
logger.info("[热重启] 文件监听已启用(改动 gvsds / gvsdsdk / Main.py 自动重启)", extra={'domain': 'system', 'status_code': 0})
server_manager = ServerManager(control)
server_manager.add_http_redirect_server(80)
server_manager.add_django_server(443, is_ssl=True, app_name="Main Django App")
server_manager.start_all(hot_reload=hot_reload)
server_manager.start_all()
except Exception as e:
traceback.print_exc()
if logger:
+2 -1
View File
@@ -41,4 +41,5 @@ redis>=4.0
# === 文档导出(在线 Word 编辑器 .docx 导出) ===
python-docx>=1.0
htmldocx>=0.0.3
htmldocx>=0.0.3