Linux

fsnotify原理探究

October 10, 2023
技术总结
Golang, Linux, Fsnotify, Inotify

本文如果没有特殊说明，所有的内容都是指 linux 系统

起因是从 kratos 群里看到有人问：“测了下kratos的config watch，好像对软链不生效”，他提供的屏幕截图如下类似：

$ pwd
/tmp/testconfig
$ ls -l
drwxr-xr-x  3 root root 4096 Oct 10 19:48 .
drwxr-xr-x 10 root root 4096 Oct 10 19:48 ..
drwxr-xr-x  1 root root   11 Oct 10 19:48 ..ver1
drwxr-xr-x  1 root root   11 Oct 10 19:48 ..ver2
lrwxr-xr-x  1 root root   11 Oct 10 19:48 ..data -> ..ver1
drwxr-xr-x  1 root root   11 Oct 10 19:48 data
$
$ ll -a data
drwxr-xr-x  3 root root 4096 Oct 10 19:48 .
drwxr-xr-x 10 root root 4096 Oct 10 19:48 ..
lrwxrwxrwx  1 root root   11 Oct 10 19:48 registry.yaml -> /tmp/testconfig/..data/registry.yaml

然后触发更新的动作其实是把 ..data 的源改成了 ..ver2，但是发现并没有触发更新，于是就问了一下。

...

Tcp 长连接服务优雅重启的秘密

August 17, 2023
技术总结
Golang, Linux, Graceful Restart, Fork, Exec, TCP

假设我们有一个长连接服务，我们想要对它进行升级，但是不想让客户端受到影响应该怎么做？这个问题其实是一个很常见的问题，比如我们的游戏服务器，我们的 IM 服务器，推送服务器等等，诸如此类使用tcp长连接的服务，都会遇到这个问题。那么我们应该怎么做呢？

需求分析 #

我们可以先来看下这个场景下的需求：

  • 客户端必须要对这个操作没有感知，也就是说客户端不需要做任何的修改，在服务器升级的过程中不需要配合。
  • 服务器在升级的过程中，不能丢失任何的连接，也就是说，如果有新的连接进来，那么这个连接必须要被接受，如果有旧的连接，那么客户端不能够触发重连。

基本思路 #

实现思路的讨论范围限制在 linux 服务器上

为了实现上述的要求，首先在升级流程中我们需要做到以下几点：

  • 旧的服务器进程在处理完请求前不能退出，而且一旦升级开始就不能再接受新的连接。
  • 旧的服务器进程在所有连接都处理完毕后才能退出。
  • 新的服务器进程在启动时需要继承旧的服务器进程的所有连接，新的连接也应该被新的服务器进程接受。
  • 新的服务器进程也必须监听旧的服务器进程的监听端口，否则新的连接无法被接受。

那么通过 Google 和 ChatGPT 的帮助，我们可以找到一些思路：

新进程继承旧进程的（监听）套接字，而不是创建新的。

为什么不创建新的（监听）套接字呢？在 linux 中内核会把处在不同握手阶段的TCP连接放在不同的队列中（半连接/全连接）。服务器的监听套接字会有自己的队列，如果创建新的套接字，那么旧的套接字队列中的连接就会丢失。为了做到客户端无感知，我们需要继承旧的套接字（主要是为了连接队列中的连接不丢失）。

半连接队列：当客户端发送 SYN 包时，服务器会把这个连接放在半连接队列中，等待服务器的 ACK 包，这个时候连接处于半连接状态。当服务器发送 ACK 包时，这个连接就会从半连接队列中移除，放到全连接队列中，这个时候连接处于全连接状态。当服务器调用 accept 时，就会从全连接队列中取出一个连接，这个时候连接处于 ESTABLISHED 状态。

实现方式 #

那么在 linux 中，我们可以通过以如下方式实现：

  1. 通过 fork 创建子进程，子进程继承父进程的所有资源，包括监听套接字;
  2. 子进程通过 exec 加载最新的二进制程序执行，这样就实现了新进程继承旧进程的监听套接字。
  3. 新进程启动完成后，通知父进程退出。
  4. 父进程受到信号后，停止接受新的连接，等待所有的连接处理完毕后退出。

在 Go 里面，我们可以通过如下方式实现：

type gracefulTcpServer struct {
	listener     *net.TCPListener
	shutdownChan chan struct{}
	conns        map[net.Conn]struct{}

	servingConnCount atomic.Int32
	serveRunning     atomic.Bool
}


// 普通启动方式
func start(port int) (*gracefulTcpServer, error) {
	ln, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
    // handle error ignored

	s := &gracefulTcpServer{
		listener:         ln.(*net.TCPListener),
		shutdownChan:     make(chan struct{}, 1),
		conns:            make(map[net.Conn]struct{}, 16),
		servingConnCount: atomic.Int32{},
		serveRunning:     atomic.Bool{},
	}

	return s, nil
}

// 优雅重启启动方式
func startFromFork() (*gracefulTcpServer, error) {
    // ... ignored code

    // 从环境变量中获取 父进程的处理的连接数，用来恢复连接
	if nfdStr := os.Getenv(__GRACE_ENV_NFDS); nfdStr == "" {
		panic("not nfds env")
	} else if nfd, err = strconv.Atoi(nfdStr); err != nil {
		panic(err)
	}

	// restore conn fds, 0, 1, 2 has been used by os.Stdin, os.Stdout, os.Stderr
	lfd := os.NewFile(3, filepath.Join(tmpdir, "graceful"))
	ln, err := net.FileListener(lfd)
	// handle error ignored

	s := &gracefulTcpServer{
		listener:         ln.(*net.TCPListener),
		shutdownChan:     make(chan struct{}, 1),
		conns:            make(map[net.Conn]struct{}, 16),
		servingConnCount: atomic.Int32{},
		serveRunning:     atomic.Bool{},
	}

    // 从父进程继承的套接字中恢复连接
	for i := 0; i < nfd; i++ {
		fd := os.NewFile(uintptr(4+i), filepath.Join(tmpdir, strconv.Itoa(4+i)))
		conn, err := net.FileConn(fd)
		// handle error ignored
		go s.handleConn(conn)
	}

	return s, nil
}

func (s *gracefulTcpServer) gracefulRestart() {
	_ = s.listener.SetDeadline(time.Now())
	lfd, err := s.listener.File()

    // 给子进程设置 优雅重启 相关的环境变量
	os.Setenv(__GRACE_ENV_FLAG, "true")
	os.Setenv(__GRACE_ENV_NFDS, strconv.Itoa(len(s.conns)))

    // 将父进程的监听套接字传递给子进程
	files := make([]uintptr, 4, 3+1+len(s.conns))
	copy(files[:4], []uintptr{
		os.Stdin.Fd(),
		os.Stdout.Fd(),
		os.Stderr.Fd(),
		lfd.Fd(),
	})
    // 将父进程的套接字传递给子进程 
	for conn := range s.conns {
		connFd, _ := conn.(*net.TCPConn).File()
		files = append(files, connFd.Fd())
	}
	procAttr := &syscall.ProcAttr{
		Env:   os.Environ(),
		Files: files,
		Sys:   nil,
	}

    // 执行 fork + exec 调用
	childPid, err := syscall.ForkExec(os.Args[0], os.Args, procAttr)
}


func main() {
    // ...
    
    // 根据环境变量判断是 fork 还是新启动
	if v := os.Getenv(__GRACE_ENV_FLAG); v != "" {
		s, err = startFromFork()
	} else {
		s, err = start(*port)
	}

	go s.serve()

    // 处理信号，如果是 SIGHUP 信号，则执行 gracefulRestart 方法后再退出
	s.waitForSignals()
}

完整代码可以在 https://github.com/yeqown/playground/golang/tcp-graceful-restart 中找到。

...

访问量 访客数