diff --git a/g/net/ghttp/ghttp_server_comm.go b/g/net/ghttp/ghttp_server_comm.go index c4240296b..0ab92295d 100644 --- a/g/net/ghttp/ghttp_server_comm.go +++ b/g/net/ghttp/ghttp_server_comm.go @@ -12,6 +12,7 @@ import ( "syscall" "os/signal" "gitee.com/johng/gf/g/os/gproc" + "gitee.com/johng/gf/g/os/gtime" "gitee.com/johng/gf/g/util/gconv" "gitee.com/johng/gf/g/encoding/gjson" "gitee.com/johng/gf/g/container/gtype" @@ -59,7 +60,13 @@ func handleProcessMsg() { act := gbinary.DecodeToUint(msg.Data[0 : 1]) data := msg.Data[1 : ] if gproc.IsChild() { + // =============== // 子进程 + // =============== + // 任何与父进程的通信都会更新最后通信时间 + if msg.Pid == gproc.Ppid() { + lastHeartbeatTime.Set(int(gtime.Millisecond())) + } switch act { case gMSG_START: onCommChildStart(msg.Pid, data) case gMSG_RESTART: onCommChildRestart(msg.Pid, data) @@ -69,7 +76,13 @@ func handleProcessMsg() { return } } else { + // =============== // 父进程 + // =============== + // 任何进程消息都会自动更新最后通信时间记录 + if msg.Pid != gproc.Pid() { + updateProcessCommTime(msg.Pid) + } switch act { case gMSG_START: onCommMainStart(msg.Pid, data) case gMSG_RESTART: onCommMainRestart(msg.Pid, data) diff --git a/g/net/ghttp/ghttp_server_comm_child.go b/g/net/ghttp/ghttp_server_comm_child.go index f771d689a..6a7bbdc7b 100644 --- a/g/net/ghttp/ghttp_server_comm_child.go +++ b/g/net/ghttp/ghttp_server_comm_child.go @@ -18,6 +18,7 @@ import ( "gitee.com/johng/gf/g/encoding/gjson" "gitee.com/johng/gf/g/container/gtype" "gitee.com/johng/gf/g/encoding/gbinary" + "gitee.com/johng/gf/g/os/glog" ) // (子进程)上一次从主进程接收心跳的时间戳 @@ -37,12 +38,17 @@ func onCommChildStart(pid int, data []byte) { } }) } + // 进程创建成功之后(开始执行服务时间点为准),通知主进程自身的存在,并开始执行心跳机制 + sendProcessMsg(gproc.Ppid(), gMSG_NEW_FORK, nil) + // 如果创建自己的父进程非gproc父进程,那么表示该进程为重启创建的进程,创建成功之后需要通知父进程销毁 + if os.Getppid() != gproc.Ppid() { + sendProcessMsg(os.Getppid(), gMSG_SHUTDOWN, nil) + } heartbeatStarted.Set(true) } // 心跳消息 func onCommChildHeartbeat(pid int, data []byte) { - //glog.Printfln("%d: child heartbeat", gproc.Pid()) lastHeartbeatTime.Set(int(gtime.Millisecond())) } @@ -70,9 +76,7 @@ func onCommChildRestart(pid int, data []byte) { p.Run() // 编码,通信 b, _ := gjson.Encode(sfm) - sendProcessMsg(p.Pid(), gMSG_START, b) - sendProcessMsg(gproc.Ppid(), gMSG_NEW_FORK, gbinary.EncodeInt(p.Pid())) - sendProcessMsg(gproc.Pid(), gMSG_SHUTDOWN, nil) + sendProcessMsg(p.Pid(), gMSG_START, b) } // 友好关闭服务链接并退出 @@ -95,7 +99,8 @@ func handleChildProcessHeartbeat() { // 超过时间没有接收到主进程心跳,自动关闭退出 if heartbeatStarted.Val() && (int(gtime.Millisecond()) - lastHeartbeatTime.Val() > gPROC_HEARTBEAT_TIMEOUT) { sendProcessMsg(gproc.Pid(), gMSG_SHUTDOWN, nil) - // 子进程有时会无法退出,这里直接使用exit,而不是return + // 子进程有时会无法退出(僵尸?),这里直接使用exit,而不是return + glog.Errorfln("%d heartbeat timeout, exit", gproc.Pid()) os.Exit(0) } } diff --git a/g/net/ghttp/ghttp_server_comm_main.go b/g/net/ghttp/ghttp_server_comm_main.go index 273b10f93..4ff63b8cc 100644 --- a/g/net/ghttp/ghttp_server_comm_main.go +++ b/g/net/ghttp/ghttp_server_comm_main.go @@ -27,8 +27,7 @@ func onCommMainStart(pid int, data []byte) { // 心跳处理 func onCommMainHeartbeat(pid int, data []byte) { - //glog.Printfln("%d: main heartbeat", gproc.Pid()) - procUpdateMap.Set(pid, int(gtime.Millisecond())) + updateProcessCommTime(pid) } // 重启服务 @@ -39,7 +38,7 @@ func onCommMainRestart(pid int, data []byte) { // 新建子进程通知 func onCommMainNewFork(pid int, data []byte) { - procManager.AddProcess(gbinary.DecodeToInt(data)) + procManager.AddProcess(pid) heartbeatStarted.Set(true) } @@ -53,6 +52,11 @@ func onCommMainShutdown(pid int, data []byte) { procManager.Send(formatMsgBuffer(gMSG_SHUTDOWN, nil)) } +// 更新指定进程的通信时间记录 +func updateProcessCommTime(pid int) { + procUpdateMap.Set(pid, int(gtime.Millisecond())) +} + // 主进程与子进程相互异步方式发送心跳信息,保持活跃状态 func handleMainProcessHeartbeat() { for { diff --git a/g/os/gproc/gproc.go b/g/os/gproc/gproc.go index 1131b6533..353be58fb 100644 --- a/g/os/gproc/gproc.go +++ b/g/os/gproc/gproc.go @@ -21,9 +21,19 @@ func Pid() int { return os.Getpid() } -// 获取父进程ID +// 获取父进程ID(gproc父进程,不存在时则使用系统父进程) func Ppid() int { - return gconv.Int(os.Getenv(gPROC_ENV_KEY_PPID_KEY)) + // gPROC_ENV_KEY_PPID_KEY为gproc包自定义的父进程 + ppidValue := os.Getenv(gPROC_ENV_KEY_PPID_KEY) + if ppidValue != "" { + return gconv.Int(ppidValue) + } + return os.Getppid() +} + +// 获取父进程ID(系统父进程) +func PpidOfOs() int { + return os.Getppid() } // 判断当前进程是否为gproc创建的子进程 diff --git a/g/os/gproc/gproc_proccess.go b/g/os/gproc/gproc_proccess.go index 1fa3e6de4..c414335d0 100644 --- a/g/os/gproc/gproc_proccess.go +++ b/g/os/gproc/gproc_proccess.go @@ -27,7 +27,9 @@ func (p *Process) Run() (int, error) { if p.process != nil { return p.Pid(), nil } - p.attr.Env = append(p.attr.Env, fmt.Sprintf("%s=%d", gPROC_ENV_KEY_PPID_KEY, p.ppid)) + if p.ppid > 0 { + p.attr.Env = append(p.attr.Env, fmt.Sprintf("%s=%d", gPROC_ENV_KEY_PPID_KEY, p.ppid)) + } if process, err := os.StartProcess(p.path, p.args, p.attr); err == nil { p.process = process if p.pm != nil {