guarded_process.go

package daphne

import (
	"context"
	"log"
	"os"
	"os/exec"
	"sync"
	"time"
)

// 守护进程，进程退出后立即启动
type GuardedProcess struct {
	cmd       string
	args      []string
	errCh     chan error
	awakeCh   chan bool
	command   *exec.Cmd
	mux       sync.RWMutex
	ctx       context.Context
	cancel    context.CancelFunc
	restartCb func()
}

func (p *GuardedProcess) SetRestartCb(restartCb func()) {
	p.restartCb = restartCb
}

func (p *GuardedProcess) getCommand() *exec.Cmd {
	p.mux.RLock()
	defer p.mux.RUnlock()
	return p.command
}

func (p *GuardedProcess) setCommand(command *exec.Cmd) {
	p.mux.Lock()
	defer p.mux.Unlock()
	p.command = command
}

func (p *GuardedProcess) wait() {
	for {
		select {
		case p.errCh <- p.getCommand().Wait():
			<-p.awakeCh
		case <-p.ctx.Done():
			log.Printf("guard_process: wait done, process: %s", p.cmd)
			return
		}
	}
}

func (p *GuardedProcess) start() (e error) {
	command := exec.Command(p.cmd, p.args...)
	command.Stdout = os.Stdout
	command.Stderr = os.Stderr
	p.setCommand(command)
	return command.Start()
}

func (p *GuardedProcess) guard() {
	log.Printf("guard_process: process: %s", p.cmd)
	for {
		select {
		case e := <-p.errCh:
			log.Printf("guard_process: prev process error: %v, restart process: %s", e, p.cmd)
			e = p.start()
			if e != nil {
				log.Printf("guard_process: restart process: %s, error: %s", p.cmd, e)
				time.Sleep(time.Second)
			} else if p.restartCb != nil {
				log.Printf("guard_process: process: %s, run callback now", p.cmd)
				p.restartCb()
			}
			p.awakeCh <- true // 通知等待协程重启完成
		case <-p.ctx.Done():
			log.Printf("guard_process: guard done, process: %s", p.cmd)
			return
		}
	}
}

// 新建需守护的进程
func NewGuardedProcess(command string, args ...string) *GuardedProcess {
	ctx, cancel := context.WithCancel(context.Background())
	return &GuardedProcess{
		cmd:     command,
		args:    args,
		errCh:   make(chan error, 1),
		awakeCh: make(chan bool, 1),
		ctx:     ctx,
		cancel:  cancel,
	}
}

// 运行守护进程
func (p *GuardedProcess) Exec() (e error) {
	e = p.start()
	if e != nil {
		return
	}
	go p.wait()
	go p.guard()
	return nil
}

// 杀掉守护进程
func (p *GuardedProcess) Kill() {
	p.cancel()
	e := p.command.Process.Kill()
	if e != nil {
		log.Printf("kill process %s error:", p.cmd)
	}
}

GuardedProcess：

• 守护进程，一旦退出立即拉起
• 拉起失败后，睡眠1秒再次拉起