cnbugs
47714c9622
回归 (eba7ae8 引入):
sync loop 周期调 sync_instances, 它只看 DB 里 enabled=True 的实例。
但 stop_instance 只停进程 + 设 running=False, 不改 enabled。
结果: 用户手动停止实例后, 30 秒内 sync loop 会把它当'死掉的实例'
自动拉回来——手动停止完全失效。
语义修正:
enabled = 期望运行状态, 是 sync loop 判断'该实例是否应该在跑'的唯一权威。
- stop_instance: enabled=False (手动停止, sync loop 不得自动重启)
- start_instance: enabled=True (手动启动, 否则下轮 sync 会把它当多余停掉)
- start 失败: enabled 保持 True (端口冲突常是暂时的, sync loop 自动重试)
测试:
+ tests/e2e_stop_semantics.py
stop 后等 2 个 loop 周期 (5s), 断言端口不被自动拉回 + DB.enabled=False;
start 后断言端口恢复 + DB.enabled=True + DB.running=True
本地 PASS
全套回归:
smoke tunnel / e2e lifecycle / e2e health / e2e sync loop / e2e stop semantics
全部 PASS
360 lines
13 KiB
Python
360 lines
13 KiB
Python
"""多实例管理器：每个实例是一个独立的 asyncio 服务器。"""
|
||
import asyncio
|
||
import logging
|
||
import threading
|
||
import time
|
||
from dataclasses import dataclass, field
|
||
|
||
from engine.server import ConnectionHandler
|
||
|
||
log = logging.getLogger("socks.instances")
|
||
|
||
|
||
@dataclass
|
||
class InstanceConfig:
|
||
"""单个实例的配置快照（运行时使用）。"""
|
||
instance_id: int
|
||
name: str
|
||
listen_host: str
|
||
listen_port: int
|
||
timeout: int
|
||
auth_method: str
|
||
bandwidth_down: float
|
||
bandwidth_up: float
|
||
max_concurrent: int
|
||
|
||
def match(self, new_name, new_host, new_port):
|
||
return (self.name == new_name and
|
||
self.listen_host == new_host and
|
||
self.listen_port == new_port)
|
||
|
||
|
||
class Socks5Server:
|
||
"""单个 SOCKS5 服务器的 asyncio 事件循环。"""
|
||
|
||
def __init__(self, config: InstanceConfig, user_service, flask_app=None):
|
||
self.config = config
|
||
self.user_service = user_service
|
||
self.flask_app = flask_app
|
||
self.server = None
|
||
self.loop = None
|
||
self.thread = None
|
||
self.running = False
|
||
self._active_connections = 0
|
||
self._lock = threading.Lock()
|
||
|
||
def start(self):
|
||
"""启动服务器（在新线程中运行独立事件循环）。"""
|
||
if self.running:
|
||
return
|
||
self.loop = asyncio.new_event_loop()
|
||
asyncio.set_event_loop(self.loop)
|
||
|
||
self._active_connections = 0
|
||
self.running = True
|
||
self.server = None
|
||
self._start_error = None # 用于在主线程透传子线程异常
|
||
|
||
def run_loop():
|
||
asyncio.set_event_loop(self.loop)
|
||
try:
|
||
self.server = self.loop.run_until_complete(
|
||
self._create_server()
|
||
)
|
||
self.loop.run_forever()
|
||
except Exception as e:
|
||
# 端口占用等异常：保存异常并停止循环，避免主线程死等
|
||
self._start_error = e
|
||
log.error("[%s] SOCKS5 服务器启动失败: %s",
|
||
self.config.name, e)
|
||
try:
|
||
self.loop.stop()
|
||
except Exception:
|
||
pass
|
||
|
||
self.thread = threading.Thread(target=run_loop, daemon=True,
|
||
name=f"socks5-{self.config.name}")
|
||
self.thread.start()
|
||
# 等 server 创建完成 (或失败) — 设了上限避免永久死等
|
||
deadline = time.time() + 5.0
|
||
while self.server is None and self.running and self._start_error is None:
|
||
time.sleep(0.05)
|
||
if time.time() > deadline:
|
||
log.error("[%s] SOCKS5 服务器启动超时", self.config.name)
|
||
self.running = False
|
||
break
|
||
# 如果子线程失败, 把异常抛回调用方
|
||
if self._start_error is not None:
|
||
self.running = False
|
||
raise self._start_error
|
||
if self.server:
|
||
log.info("[%s] SOCKS5 服务器已启动: %s:%d",
|
||
self.config.name, self.config.listen_host, self.config.listen_port)
|
||
|
||
def stop(self):
|
||
"""停止服务器。"""
|
||
if not self.running:
|
||
return
|
||
self.running = False
|
||
if self.server and self.loop:
|
||
# 在子线程事件循环中优雅关闭服务器
|
||
import concurrent.futures
|
||
async def _shutdown():
|
||
self.server.close()
|
||
await self.server.wait_closed()
|
||
try:
|
||
future = asyncio.run_coroutine_threadsafe(_shutdown(), self.loop)
|
||
future.result(timeout=5) # 最多等待5秒
|
||
except (concurrent.futures.TimeoutError, Exception):
|
||
pass
|
||
if self.loop:
|
||
try:
|
||
self.loop.call_soon_threadsafe(self.loop.stop)
|
||
except Exception:
|
||
pass
|
||
if self.thread:
|
||
self.thread.join(timeout=5)
|
||
log.info("[%s] SOCKS5 服务器已停止", self.config.name)
|
||
|
||
async def _create_server(self):
|
||
return await asyncio.start_server(
|
||
self._accept_connection,
|
||
self.config.listen_host, self.config.listen_port
|
||
)
|
||
|
||
async def _accept_connection(self, reader, writer):
|
||
try:
|
||
with self._lock:
|
||
self._active_connections += 1
|
||
handler = ConnectionHandler(
|
||
reader, writer, self.config, self.user_service,
|
||
flask_app=self.flask_app,
|
||
on_close_cb=self._on_connection_close
|
||
)
|
||
await handler.handle()
|
||
except (asyncio.CancelledError, GeneratorExit):
|
||
# 服务器关闭时正常取消，不记录错误
|
||
pass
|
||
finally:
|
||
try:
|
||
writer.close()
|
||
except Exception:
|
||
pass
|
||
|
||
def _on_connection_close(self, handler):
|
||
with self._lock:
|
||
self._active_connections = max(0, self._active_connections - 1)
|
||
|
||
@property
|
||
def active_connections(self):
|
||
with self._lock:
|
||
return self._active_connections
|
||
|
||
def is_alive(self):
|
||
"""真实健康检查: 线程活 + server socket 在服务。
|
||
|
||
self.running 是乐观标记, start() 之后一直为 True 直到 stop() 被显式调用。
|
||
如果子线程 event loop 崩了 (例如 _record_stats 抛 TypeError 把 worker 卡死,
|
||
上游 gunicorn timeout kill 进程, 或端口被外部抢占), self.running 还是
|
||
True 但 server 实际不再 accept。
|
||
|
||
调用方应据此重启实例, 否则 DB 里 running=true 是谎言。
|
||
|
||
判定:
|
||
- thread 死了 -> 死 (无法恢复)
|
||
- server is None -> 死 (还没起来)
|
||
- server is_serving() == False -> 死 (socket 已关, asyncio.Server
|
||
不能复用同一个对象重新 serve, 必须新 Socks5Server)
|
||
"""
|
||
if not self.running:
|
||
return False
|
||
if self.thread is None or not self.thread.is_alive():
|
||
return False
|
||
if self.server is None:
|
||
return False
|
||
if not self.server.is_serving():
|
||
return False
|
||
return True
|
||
|
||
|
||
import time # noqa: E402
|
||
|
||
|
||
class InstanceManager:
|
||
"""管理所有 SOCKS5 实例。"""
|
||
|
||
def __init__(self, user_service, flask_app=None):
|
||
self.user_service = user_service
|
||
self.flask_app = flask_app
|
||
self._instances = {} # name -> Socks5Server
|
||
self._config_cache = {} # name -> InstanceConfig
|
||
self._lock = threading.Lock()
|
||
|
||
def reload_config(self):
|
||
"""从数据库重新加载实例配置。"""
|
||
from models import Instance
|
||
from database import db
|
||
new_configs = {}
|
||
with db.app.app_context():
|
||
for inst in Instance.query.filter_by(enabled=True).all():
|
||
new_configs[inst.name] = InstanceConfig(
|
||
instance_id=inst.id,
|
||
name=inst.name,
|
||
listen_host=inst.listen_host,
|
||
listen_port=inst.listen_port,
|
||
timeout=inst.timeout,
|
||
auth_method=inst.auth_method,
|
||
bandwidth_down=inst.bandwidth_down,
|
||
bandwidth_up=inst.bandwidth_up,
|
||
max_concurrent=inst.max_concurrent,
|
||
)
|
||
self._config_cache = new_configs
|
||
|
||
def sync_instances(self):
|
||
"""根据数据库配置同步实例运行状态。
|
||
|
||
健康检查: 对每个 _instances[name], 调用 is_alive() 验证线程 + server
|
||
socket 都活着。如果死了, 从 _instances 删掉, 用同一个 config 重新启动。
|
||
避免 40080 那种'DB 写 running=true 但端口没人 listen'的谎言状态。
|
||
"""
|
||
from models import Instance
|
||
from database import db
|
||
self.reload_config()
|
||
|
||
current_names = set(self._config_cache.keys())
|
||
running_names = set(self._instances.keys())
|
||
|
||
# 健康检查: 把死的从 _instances 摘掉, 让下面的'启动缺失'逻辑接管。
|
||
# _config_cache 是从 DB reload 的权威配置, 不能动它, 否则重启循环拿不到 cfg。
|
||
dead = []
|
||
for name, srv in self._instances.items():
|
||
if not srv.is_alive():
|
||
log.warning("[%s] SOCKS5 进程不健康 (thread/socket dead), 标记待重启", name)
|
||
dead.append(name)
|
||
for name in dead:
|
||
self._instances.pop(name, None)
|
||
running_names = set(self._instances.keys())
|
||
|
||
# 启动缺失的实例 (含刚被健康检查踢出来的)
|
||
for name in current_names - running_names:
|
||
cfg = self._config_cache[name]
|
||
srv = Socks5Server(cfg, self.user_service, self.flask_app)
|
||
try:
|
||
srv.start()
|
||
except OSError as e:
|
||
log.error("[%s] 跳过启动（端口冲突?）: %s", name, e)
|
||
continue
|
||
with self._lock:
|
||
self._instances[name] = srv
|
||
|
||
# 停止多余的实例
|
||
for name in running_names - current_names:
|
||
srv = self._instances.pop(name)
|
||
srv.stop()
|
||
|
||
# 更新实例运行状态到数据库 (用 is_alive() 而不是乐观 running 标记)
|
||
with db.app.app_context():
|
||
for inst in Instance.query.all():
|
||
srv = self._instances.get(inst.name)
|
||
inst.running = srv is not None and srv.is_alive()
|
||
if srv:
|
||
inst.active_connections = srv.active_connections
|
||
else:
|
||
inst.active_connections = 0
|
||
db.session.commit()
|
||
|
||
def start_instance(self, instance_id):
|
||
"""启动单个实例。"""
|
||
from models import Instance
|
||
from database import db
|
||
with db.app.app_context():
|
||
inst = Instance.query.get(instance_id)
|
||
if not inst:
|
||
return False
|
||
cfg = InstanceConfig(
|
||
instance_id=inst.id,
|
||
name=inst.name,
|
||
listen_host=inst.listen_host,
|
||
listen_port=inst.listen_port,
|
||
timeout=inst.timeout,
|
||
auth_method=inst.auth_method,
|
||
bandwidth_down=inst.bandwidth_down,
|
||
bandwidth_up=inst.bandwidth_up,
|
||
max_concurrent=inst.max_concurrent,
|
||
)
|
||
self._config_cache[inst.name] = cfg
|
||
srv = Socks5Server(cfg, self.user_service, self.flask_app)
|
||
try:
|
||
srv.start()
|
||
except OSError as e:
|
||
# 端口占用等启动失败: 回滚状态, 清理缓存, 返回 False
|
||
log.error("[%s] 启动实例失败: %s", inst.name, e)
|
||
self._config_cache.pop(inst.name, None)
|
||
# enabled 保持 True: 用户意图是运行, sync loop 下周期会重试
|
||
# (端口冲突常是暂时的, 自动重试比让用户手动点更合理)
|
||
inst.enabled = True
|
||
inst.running = False
|
||
db.session.commit()
|
||
raise
|
||
with self._lock:
|
||
self._instances[inst.name] = srv
|
||
# enabled = 期望运行状态, 是 sync loop 的权威依据。
|
||
# 手动启动必须置 True, 否则下轮 sync 会把它当"多余实例"停掉。
|
||
inst.enabled = True
|
||
inst.running = True
|
||
db.session.commit()
|
||
return True
|
||
|
||
def stop_instance(self, instance_id):
|
||
"""停止单个实例。
|
||
|
||
关键: 必须同时置 enabled=False。enabled 是 sync loop 判断"该实例是否
|
||
应该在跑"的唯一依据——只停进程不改 enabled, 30 秒内 sync loop 会把它
|
||
当"死掉的实例"自动重启, 用户的手动停止就失效了。
|
||
"""
|
||
from models import Instance
|
||
from database import db
|
||
with db.app.app_context():
|
||
inst = Instance.query.get(instance_id)
|
||
if not inst:
|
||
return False
|
||
srv = self._instances.pop(inst.name, None)
|
||
if srv:
|
||
srv.stop()
|
||
self._config_cache.pop(inst.name, None)
|
||
inst.enabled = False # 手动停止: sync loop 不得自动重启
|
||
inst.running = False
|
||
inst.active_connections = 0
|
||
db.session.commit()
|
||
return True
|
||
|
||
def restart_instance(self, instance_id):
|
||
"""重启单个实例。"""
|
||
self.stop_instance(instance_id)
|
||
return self.start_instance(instance_id)
|
||
|
||
def get_status(self):
|
||
"""获取所有实例状态。"""
|
||
from models import Instance
|
||
from database import db
|
||
with db.app.app_context():
|
||
result = []
|
||
for inst in Instance.query.all():
|
||
srv = self._instances.get(inst.name)
|
||
result.append({
|
||
"id": inst.id,
|
||
"name": inst.name,
|
||
"host": inst.listen_host,
|
||
"port": inst.listen_port,
|
||
"enabled": inst.enabled,
|
||
"running": bool(srv and srv.running),
|
||
"active_connections": srv.active_connections if srv else 0,
|
||
"auth_method": inst.auth_method,
|
||
"timeout": inst.timeout,
|
||
"bandwidth_down": inst.bandwidth_down,
|
||
"bandwidth_up": inst.bandwidth_up,
|
||
"max_concurrent": inst.max_concurrent,
|
||
"notes": inst.notes,
|
||
})
|
||
return result
|