From 2950e4015305bafc4a0ba3c9d6c378d97a9ac085 Mon Sep 17 00:00:00 2001 From: lxh2875931338 Date: Mon, 10 Aug 2026 21:47:24 +0800 Subject: [PATCH] =?UTF-8?q?=E9=A6=96=E6=AC=A1=E4=BC=98=E5=8C=96=E7=9A=84?= =?UTF-8?q?=E8=BF=9B=E7=A8=8B=E7=AE=A1=E7=90=86=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/process/manager.go | 444 +++++++++++++++++++++++++++++++----- 1 file changed, 387 insertions(+), 57 deletions(-) diff --git a/internal/process/manager.go b/internal/process/manager.go index 0c2a3a3..4db4d2c 100644 --- a/internal/process/manager.go +++ b/internal/process/manager.go @@ -1,10 +1,7 @@ // internal/process/manager.go // frpc-console 进程管理模块 -// 2.6-preview: 状态机驱动 + 端口检测 + 单实例锁定 - -// 作者留：这个版本可以用了，已经明显看到进程出来发现端口不对立马被杀掉的状态了 -// 但是稍显粗暴，差点意思，到时候让Deepseek稍微软化下就好了 -// 但是现阶段Preview还不能用，再等等 +// 2.6-preview: 状态机驱动 + 冲突检测 + 生命周期管理 +// 2.7-preview: 软化版 - 精确控制 + CONFLICT/STOPPING 状态 package process @@ -44,6 +41,8 @@ const ( StartupMaxAttempts = 50 StartupRetryDelay = 200 * time.Millisecond StartupTimeout = 10 * time.Second + + HealthCheckInterval = 15 * time.Second ) // ================================================================ @@ -59,6 +58,8 @@ const ( PhaseDegraded ProcessPhase = "DEGRADED" PhaseFailed ProcessPhase = "FAILED" PhaseStopped ProcessPhase = "STOPPED" + PhaseStopping ProcessPhase = "STOPPING" // 正在停止中 + PhaseConflict ProcessPhase = "CONFLICT" // 检测到多实例冲突 ) // ================================================================ @@ -70,6 +71,14 @@ type PortCheckResult struct { Err error } +type FrpcInstance struct { + PID int + ParentPID int + ExecPath string + CmdLine string + Owned bool // 是否由当前 ProcessManager 管理 +} + type ProcessState struct { Phase ProcessPhase `json:"phase"` PID int `json:"pid"` @@ -88,8 +97,16 @@ type ProcessState struct { LastOutput string `json:"last_output,omitempty"` LastError string `json:"last_error,omitempty"` Error string `json:"error,omitempty"` + + // 冲突检测信息 + Conflicts []FrpcInstance `json:"conflicts,omitempty"` } +var ( + globalManager *ProcessManager + globalMu sync.Mutex +) + type PortStatus struct { Port int `json:"port"` Occupied bool `json:"occupied"` @@ -131,24 +148,28 @@ type ProcessManager struct { lockFile *os.File locked bool -} -var ( - globalManager *ProcessManager - globalMu sync.Mutex -) + // 当前状态缓存 + currentPhase ProcessPhase + statusMu sync.RWMutex + + // 取消函数（用于停止健康检查） + cancelFunc context.CancelFunc +} // ================================================================ // 构造函数 // ================================================================ func NewManager(dataDir, configPath, frpcBinPath string) *ProcessManager { - return &ProcessManager{ - dataDir: dataDir, - configPath: configPath, - frpcBinPath: frpcBinPath, - adminPort: 0, + pm := &ProcessManager{ + dataDir: dataDir, + configPath: configPath, + frpcBinPath: frpcBinPath, + adminPort: 0, + currentPhase: PhaseUnknown, } + return pm } func SetGlobalManager(pm *ProcessManager) { @@ -167,6 +188,21 @@ func (pm *ProcessManager) AdminPort() int { return pm.adminPort } +func (pm *ProcessManager) CurrentPhase() ProcessPhase { + pm.statusMu.RLock() + defer pm.statusMu.RUnlock() + return pm.currentPhase +} + +func (pm *ProcessManager) setPhase(phase ProcessPhase) { + pm.statusMu.Lock() + defer pm.statusMu.Unlock() + if pm.currentPhase != phase { + log.Printf("[STATUS] %s → %s", pm.currentPhase, phase) + pm.currentPhase = phase + } +} + // ================================================================ // 配置读取 // ================================================================ @@ -469,6 +505,205 @@ func (pm *ProcessManager) isProcessAlive(pid int) bool { return process.Signal(syscall.Signal(0)) == nil } +// ================================================================ +// 实例检测（精确扫描） +// ================================================================ + +// DetectFrpcInstances 检测系统中所有 frpc 实例 +// 三重确认：PID存在 + 进程存活 + Executable/CmdLine匹配 +func (pm *ProcessManager) DetectFrpcInstances() []FrpcInstance { + var instances []FrpcInstance + + // 获取当前 frpc 可执行文件的绝对路径 + targetExe := pm.frpcBinPath + if !filepath.IsAbs(targetExe) { + if abs, err := filepath.Abs(targetExe); err == nil { + targetExe = abs + } + } + + if runtime.GOOS == "windows" { + // Windows: 使用 tasklist + wmic 获取进程信息 + cmd := exec.Command("tasklist", "/FO", "CSV", "/FI", "IMAGENAME eq frpc.exe") + out, err := cmd.Output() + if err != nil { + log.Printf("[WARN] 检测 frpc 实例失败: %v", err) + return instances + } + lines := strings.Split(string(out), "\n") + for _, line := range lines { + if !strings.Contains(line, "frpc.exe") { + continue + } + parts := strings.Split(line, ",") + if len(parts) < 2 { + continue + } + pidStr := strings.Trim(parts[1], `"`) + pid, err := strconv.Atoi(pidStr) + if err != nil || pid <= 0 { + continue + } + if !pm.isProcessAlive(pid) { + continue + } + // 获取 cmdline（Windows 较难获取，先用进程名匹配） + inst := FrpcInstance{ + PID: pid, + ExecPath: "frpc.exe", + CmdLine: "frpc.exe", + Owned: false, + } + // 检查是否 Owned + ownedPid := pm.readPIDFile() + if ownedPid == pid { + inst.Owned = true + } + instances = append(instances, inst) + } + return instances + } + + // Linux: 使用 pgrep + /proc 精确匹配 + cmd := exec.Command("pgrep", "-f", "frpc") + out, err := cmd.Output() + if err != nil { + // 没有找到进程，pgrep 返回非0，属于正常情况 + return instances + } + + pids := strings.Split(strings.TrimSpace(string(out)), "\n") + for _, pidStr := range pids { + pidStr = strings.TrimSpace(pidStr) + if pidStr == "" { + continue + } + pid, err := strconv.Atoi(pidStr) + if err != nil || pid <= 0 { + continue + } + + if !pm.isProcessAlive(pid) { + continue + } + + // 获取命令行 + cmdline, _ := os.ReadFile(fmt.Sprintf("/proc/%d/cmdline", pid)) + cmdLineStr := strings.ReplaceAll(string(cmdline), "\x00", " ") + + // 获取可执行文件路径 + exePath, _ := os.Readlink(fmt.Sprintf("/proc/%d/exe", pid)) + + // 获取父进程 PID + stat, _ := os.ReadFile(fmt.Sprintf("/proc/%d/stat", pid)) + var parentPID int + if len(stat) > 0 { + // stat 格式: pid (comm) state ppid ... + parts := strings.Fields(string(stat)) + if len(parts) > 3 { + parentPID, _ = strconv.Atoi(parts[3]) + } + } + + inst := FrpcInstance{ + PID: pid, + ParentPID: parentPID, + ExecPath: exePath, + CmdLine: cmdLineStr, + Owned: false, + } + + // 三重确认 Owned + ownedPid := pm.readPIDFile() + if ownedPid == pid { + // 条件1: PID 匹配 + if pm.isProcessAlive(pid) { + // 条件2: 进程存在 + // 条件3: 执行路径匹配 + if exePath != "" && (strings.Contains(exePath, "frpc") || strings.Contains(cmdLineStr, "frpc")) { + inst.Owned = true + } + } + } + + instances = append(instances, inst) + } + + return instances +} + +// filterOwned 返回 Owned 实例 +func (pm *ProcessManager) filterOwned(instances []FrpcInstance) []FrpcInstance { + var result []FrpcInstance + for _, inst := range instances { + if inst.Owned { + result = append(result, inst) + } + } + return result +} + +// filterUnknown 返回 Unknown 实例 +func (pm *ProcessManager) filterUnknown(instances []FrpcInstance) []FrpcInstance { + var result []FrpcInstance + for _, inst := range instances { + if !inst.Owned { + result = append(result, inst) + } + } + return result +} + +// ================================================================ +// 冲突检测 +// ================================================================ + +type ConflictInfo struct { + HasConflict bool + Count int + Owned []FrpcInstance + Unknown []FrpcInstance +} + +func (pm *ProcessManager) DetectConflict() ConflictInfo { + instances := pm.DetectFrpcInstances() + owned := pm.filterOwned(instances) + unknown := pm.filterUnknown(instances) + + return ConflictInfo{ + HasConflict: len(unknown) > 0 || len(owned) > 1, + Count: len(instances), + Owned: owned, + Unknown: unknown, + } +} + +// ================================================================ +// 实例健康检查 +// ================================================================ + +func (pm *ProcessManager) isInstanceHealthy(inst FrpcInstance) bool { + // 1. 进程必须存在 + if !pm.isProcessAlive(inst.PID) { + return false + } + // 2. 端口必须可连接 + result := pm.CheckPort() + if !result.Ready { + return false + } + // 3. FRP 就绪（通过 stdout 或 admin API） + stdout := pm.getLastOutput() + if pm.isFRPReady(stdout) { + return true + } + // 4. 尝试 admin API + if info := pm.getFRPCStatus(inst.PID); info != nil && info.Version != "" { + return true + } + return false +} + // ================================================================ // 进程清理 // ================================================================ @@ -490,34 +725,6 @@ func (pm *ProcessManager) killProcess(pid int) error { return process.Kill() } -// killAllFrpc 强制杀死所有 frpc 进程（包括孤儿） -func (pm *ProcessManager) killAllFrpc() { - log.Printf("[INFO] 清理所有 frpc 进程...") - if runtime.GOOS == "windows" { - exec.Command("taskkill", "/F", "/IM", "frpc.exe").Run() - } else { - exec.Command("pkill", "-9", "-f", "frpc").Run() - } - time.Sleep(1 * time.Second) - pm.deletePIDFile() -} - -// countFrpcProcesses 统计当前 frpc 进程数量 -func (pm *ProcessManager) countFrpcProcesses() int { - if runtime.GOOS == "windows" { - cmd := exec.Command("tasklist", "/FI", "IMAGENAME eq frpc.exe") - out, _ := cmd.CombinedOutput() - return strings.Count(string(out), "frpc.exe") - } - cmd := exec.Command("pgrep", "-c", "-f", "frpc") - out, err := cmd.Output() - if err != nil { - return 0 - } - count, _ := strconv.Atoi(strings.TrimSpace(string(out))) - return count -} - func (pm *ProcessManager) cleanupOrphans() { result := pm.CheckPort() if !result.Ready { @@ -673,7 +880,17 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS state.PortError = portErr.Error() } - // ---- 状态判定逻辑 ---- + // ============================================================ + // 状态判定逻辑 + // ============================================================ + + // 1. 如果是主动停止，直接返回 STOPPING + if pm.expectedStop && alive { + state.Phase = PhaseStopping + return state + } + + // 2. 进程不存在 if !alive { if pm.expectedStop { state.Phase = PhaseStopped @@ -695,7 +912,7 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS return state } - // ---- alive == true ---- + // 3. 进程存在，检查超时 if time.Since(pm.startTime) > StartupTimeout { state.Error = fmt.Sprintf("启动超时 (超过 %v)", StartupTimeout) if portErr != nil { @@ -705,22 +922,34 @@ func (pm *ProcessManager) computeState(pid int, result PortCheckResult) ProcessS return state } - if portReady && frpReady { - state.Phase = PhaseRunning - // 检测多进程 - if count := pm.countFrpcProcesses(); count > 1 { - state.Phase = PhaseDegraded - state.Error = fmt.Sprintf("检测到 %d 个 frpc 进程，可能存在多开", count) + // 4. 冲突检测 + conflict := pm.DetectConflict() + if conflict.HasConflict { + state.Phase = PhaseConflict + state.Conflicts = append(conflict.Owned, conflict.Unknown...) + if len(conflict.Unknown) > 0 { + state.Error = fmt.Sprintf("检测到 %d 个未知 frpc 实例", len(conflict.Unknown)) + } + if len(conflict.Owned) > 1 { + state.Error = fmt.Sprintf("检测到 %d 个 Owned frpc 实例（异常）", len(conflict.Owned)) } return state } + // 5. 正常运行状态 + if portReady && frpReady { + state.Phase = PhaseRunning + return state + } + + // 6. 端口就绪但 FRP 未就绪 if portReady && !frpReady { state.Phase = PhaseDegraded state.Error = "端口已监听，但 frpc 未报告就绪" return state } + // 7. 端口未就绪 if !portReady { if portErr != nil && strings.Contains(portErr.Error(), "permission denied") { state.Phase = PhaseDegraded @@ -746,6 +975,7 @@ func (pm *ProcessManager) Status() (*ProcessState, error) { pid := pm.readPIDFile() result := pm.CheckPort() state := pm.computeState(pid, result) + pm.setPhase(state.Phase) return &state, nil } @@ -796,12 +1026,43 @@ func (pm *ProcessManager) Start(ctx context.Context) error { } defer pm.Unlock() - // 先杀所有 frpc 进程（避免多开） - pm.killAllFrpc() + pm.expectedStop = false - // 清理孤儿 + // 1. 检测冲突 + conflict := pm.DetectConflict() + if conflict.HasConflict { + log.Printf("[WARN] 检测到 %d 个 frpc 实例 (Owned: %d, Unknown: %d)，进入 CONFLICT 状态", + conflict.Count, len(conflict.Owned), len(conflict.Unknown)) + + // 清理 Unknown 实例 + for _, inst := range conflict.Unknown { + log.Printf("[INFO] 清理 Unknown frpc 实例 (PID: %d)", inst.PID) + pm.killProcess(inst.PID) + } + + // 如果有且仅有一个 Owned 实例，保留它 + if len(conflict.Owned) == 1 { + pid := conflict.Owned[0].PID + log.Printf("[INFO] 保留 Owned frpc 实例 (PID: %d)，清理完成", pid) + pm.writePIDFile(pid) + pm.setPhase(PhaseRunning) + return nil + } + + // 多个 Owned 或没有 Owned：全部清理，重新启动 + if len(conflict.Owned) > 1 { + log.Printf("[WARN] 多个 Owned 实例冲突，全部清理") + for _, inst := range conflict.Owned { + pm.killProcess(inst.PID) + } + } + pm.deletePIDFile() + } + + // 2. 清理孤儿 pm.cleanupOrphans() + // 3. 正常启动 return pm.startLocked(ctx) } @@ -811,6 +1072,7 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error { pid := pm.readPIDFile() if pid > 0 && pm.isProcessAlive(pid) { log.Printf("[INFO] frpc 已在运行 (PID: %d)", pid) + pm.setPhase(PhaseRunning) return nil } pm.cleanupOrphans() @@ -847,6 +1109,8 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error { } log.Printf("[DEBUG] frpc 进程已启动，PID: %d", cmd.Process.Pid) + pm.setPhase(PhaseStarting) + go func() { err := cmd.Wait() var code int @@ -866,12 +1130,14 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error { if !pm.expectedStop && code != 0 { log.Printf("[WARN] frpc 进程异常退出 (exit code: %d)", code) + pm.setPhase(PhaseFailed) } if !pm.expectedStop { pm.deletePIDFile() } }() + // 等待端口就绪 for attempt := 0; attempt < StartupMaxAttempts; attempt++ { r := pm.CheckPort() if r.Ready { @@ -879,6 +1145,7 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error { if pm.isFRPReady(stdout) { log.Printf("[INFO] frpc 启动成功 (PID: %d, 端口: %d)，耗时 %dms", cmd.Process.Pid, pm.adminPort, attempt*int(StartupRetryDelay/time.Millisecond)) + pm.setPhase(PhaseRunning) return nil } log.Printf("[DEBUG] 端口已就绪，等待 frpc 初始化...") @@ -893,15 +1160,17 @@ func (pm *ProcessManager) startLocked(ctx context.Context) error { if info := pm.getFRPCStatus(cmd.Process.Pid); info != nil && info.Version != "" { log.Printf("[INFO] frpc admin API 可访问，版本: %s", info.Version) pm.writePIDFile(cmd.Process.Pid) + pm.setPhase(PhaseRunning) return nil } - // 启动失败：杀掉进程 pm.killProcess(cmd.Process.Pid) pm.deletePIDFile() + pm.setPhase(PhaseFailed) return fmt.Errorf("frpc 启动超时: 进程存在但端口未就绪") } pm.deletePIDFile() + pm.setPhase(PhaseFailed) return fmt.Errorf("frpc 启动失败: 进程已退出") } @@ -912,6 +1181,7 @@ func (pm *ProcessManager) Stop(ctx context.Context) error { defer pm.Unlock() pm.expectedStop = true + pm.setPhase(PhaseStopping) return pm.stopLocked(ctx) } @@ -924,33 +1194,41 @@ func (pm *ProcessManager) stopLocked(_ context.Context) error { exec.Command("pkill", "-f", "frpc").Run() } pm.deletePIDFile() + pm.setPhase(PhaseStopped) return nil } if !pm.isProcessAlive(pid) { pm.deletePIDFile() + pm.setPhase(PhaseStopped) return nil } + // 发送 SIGTERM process, err := os.FindProcess(pid) if err != nil { pm.deletePIDFile() + pm.setPhase(PhaseStopped) return nil } if err := process.Signal(syscall.SIGTERM); err != nil { pm.deletePIDFile() + pm.setPhase(PhaseStopped) return nil } + // 等待进程退出 start := time.Now() for time.Since(start) < StopMaxWaitTime { if !pm.isProcessAlive(pid) { pm.deletePIDFile() + pm.setPhase(PhaseStopped) return nil } time.Sleep(200 * time.Millisecond) } + // 强制 kill if runtime.GOOS == "windows" { exec.Command("taskkill", "/F", "/PID", strconv.Itoa(pid)).Run() } else { @@ -959,10 +1237,12 @@ func (pm *ProcessManager) stopLocked(_ context.Context) error { time.Sleep(500 * time.Millisecond) if pm.isProcessAlive(pid) { + pm.setPhase(PhaseFailed) return fmt.Errorf("强制停止失败: 进程仍存活 (PID: %d)", pid) } pm.deletePIDFile() + pm.setPhase(PhaseStopped) return nil } @@ -982,6 +1262,54 @@ func (pm *ProcessManager) Restart(ctx context.Context) error { return nil } +// ================================================================ +// 健康检查看门狗（由 main.go 调用） +// ================================================================ + +func (pm *ProcessManager) StartHealthMonitor(ctx context.Context) { + ticker := time.NewTicker(HealthCheckInterval) + defer ticker.Stop() + + for { + select { + case <-ctx.Done(): + return + case <-ticker.C: + pm.runHealthCheck() + } + } +} + +func (pm *ProcessManager) runHealthCheck() { + state, err := pm.Status() + if err != nil { + log.Printf("[WARN] 健康检查失败: %v", err) + return + } + + // 根据状态执行自动恢复 + switch state.Phase { + case PhaseConflict: + log.Printf("[WARN] 检测到冲突，尝试自动恢复...") + // 冲突恢复：清理未知实例 + for _, inst := range state.Conflicts { + if !inst.Owned { + pm.killProcess(inst.PID) + } + } + pm.setPhase(PhaseStarting) + case PhaseDegraded: + log.Printf("[WARN] frpc 处于降级状态，尝试恢复...") + // 降级恢复：重启 + pm.Restart(context.Background()) + case PhaseFailed: + if !state.ExpectedStop { + log.Printf("[WARN] frpc 已失败，自动重启...") + pm.Start(context.Background()) + } + } +} + // ================================================================ // 健康检查 // ================================================================ @@ -989,15 +1317,14 @@ func (pm *ProcessManager) Restart(ctx context.Context) error { func (pm *ProcessManager) HealthCheck() map[string]interface{} { result := make(map[string]interface{}) result["admin_port"] = pm.adminPort + result["phase"] = pm.CurrentPhase() state, err := pm.Status() if err != nil { - result["phase"] = "ERROR" result["error"] = err.Error() return result } - result["phase"] = state.Phase result["pid"] = state.PID result["port"] = state.Port result["alive"] = state.Alive @@ -1012,6 +1339,9 @@ func (pm *ProcessManager) HealthCheck() map[string]interface{} { if state.Version != "" { result["version"] = state.Version } + if len(state.Conflicts) > 0 { + result["conflicts"] = state.Conflicts + } return result }