diff --git a/Main.py b/Main.py index 1e9a75c9..a509fcda 100644 --- a/Main.py +++ b/Main.py @@ -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: diff --git a/gvsds/requirements.txt b/gvsds/requirements.txt index 7a83f1e7..7e3b8cd9 100644 --- a/gvsds/requirements.txt +++ b/gvsds/requirements.txt @@ -41,4 +41,5 @@ redis>=4.0 # === 文档导出（在线 Word 编辑器 .docx 导出） === python-docx>=1.0 -htmldocx>=0.0.3 \ No newline at end of file +htmldocx>=0.0.3 +