quic-go源码一---server启动

news2024/10/22 9:38:06

前言：

走马观花地看了RFC 9000:QUIC: A UDP-Based Multiplexed and Secure Transport，
感受不是那么直观，所以再来看看这个协议的golang语言实现：quic-go,加强学习。

https://quic-go.net/docs/quic/quic-go文档
在这里插入图片描述

本篇准备的代码片断如下：

const addr = "127.0.0.1:9000"

func main() {
	quicConf := &quic.Config{
		InitialStreamReceiveWindow:     1 << 20,  // 1 MB
		MaxStreamReceiveWindow:         6 << 20,  // 6 MB
		InitialConnectionReceiveWindow: 2 << 20,  // 2 MB
		MaxConnectionReceiveWindow:     12 << 20, // 12 MB
	}
	// 学习点1
	listener, err := quic.ListenAddr(addr, generateTLSConfig(), quicConf)
	if err != nil {
		log.Fatalf("Error listening on address: %v", err)
	}
	defer listener.Close()

	for {
		// 学习点2
		conn, err := listener.Accept(context.Background())
		if err != nil {
			log.Printf("Error accepting connection: %v", err)
			continue
		}

		go handleConnection(conn)
		fmt.Println("New client connected")
	}
}

func handleConnection(conn quic.Connection) {
	for {
		// 接收数据流
		stream, err := conn.AcceptStream(context.Background())
		// 。。。。。。
	}
}

func generateTLSConfig() *tls.Config {
	key, err := rsa.GenerateKey(rand.Reader, 1024)
	if err != nil {
		panic(err)
	}
	template := x509.Certificate{SerialNumber: big.NewInt(1)}
	certDER, err := x509.CreateCertificate(rand.Reader, &template, &template, &key.PublicKey, key)
	if err != nil {
		panic(err)
	}
	keyPEM := pem.EncodeToMemory(&pem.Block{Type: "RSA PRIVATE KEY", Bytes: x509.MarshalPKCS1PrivateKey(key)})
	certPEM := pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: certDER})

	tlsCert, err := tls.X509KeyPair(certPEM, keyPEM)
	if err != nil {
		panic(err)
	}
	return &tls.Config{
		Certificates: []tls.Certificate{tlsCert},
		NextProtos:   []string{"TEST"},
	}
}

看点1

listener, err := quic.ListenAddr(addr, generateTLSConfig(), quicConf)

先记住参数

createdConn: true,
isSingleUse: true,

在这里插入图片描述
其中conn, err := listenUDP(addr)是net udp常规操作，略过不提。
所以核心是调用了Transport.Listener(tlsConf, quicConf)方法。但是注意其中的参数，呃，我们先看官网文档：

使用简写方式：我们demo正是如此：

看看官方的实现：正好印证了文档的准确性：
在这里插入图片描述

接着看 Transport.Listen(…)

在这里插入图片描述

上述allow0RTT为true/false的区别在于，所以当使用ListenEarly(。。。)（allow0RTT为true）时，服务端会开辟一个map：

官网文档如下：To allow clients to use 0-RTT resumption, the Allow0RTT flag needs to be set on the quic.Config.（为了能够使用0-RTT, 需要在 quic.Config设置Allow0RTT标识）
当然也得用ListenEarly() 而不是Listen()
在这里插入图片描述
allow0RTT在本篇只是带过，知道有个概念，后续有机会学习分析。。。

接着看 Transport.createServer(…)

在这里插入图片描述

validateConfig(…)：

在这里插入图片描述

populateConfig(config *Config)

在这里插入图片描述

t.init(false)

在这里插入图片描述
wrapConn(..)核心如下：

setReceiveBuffer(pc net.PacketConn)
setSendBuffer(pc)

supportsDF, err = setDF(rawConn)

// 优化点在这里！
c, ok := pc.(OOBCapablePacketConn)
if !ok {
	return &basicConn{PacketConn: pc, supportsDF: supportsDF}, nil
}
return newConn(c, supportsDF)

其中前2项性质都一样:

_ = conn.SetReadBuffer(protocol.DesiredReceiveBufferSize) // 7 MB

err := fd.pfd.SetsockoptInt(syscall.SOL_SOCKET, syscall.SO_SNDBUF, bytes)
size, serr = unix.GetsockoptInt(int(fd), unix.SOL_SOCKET, unix.SO_SNDBUF)

在这里插入图片描述
其中setDF(rawConn syscall.RawConn)实现主要逻辑如下：

errDFIPv4 = unix.SetsockoptInt(int(fd), unix.IPPROTO_IP, unix.IP_DONTFRAG, 1)
errDFIPv6 = unix.SetsockoptInt(int(fd), unix.IPPROTO_IPV6, unix.IPV6_DONTFRAG, 1)

在这里插入图片描述
再看看newConn(...)

只激活了读取ipv4的ECN：

最后返回的conn如下：

说完了wrapConn(...)后再回到init(...)

// is closed when listen returns
t.listening = make(chan struct{})

在这里插入图片描述
DefaultConnectionIDGenerator初始化：

getMultiplexer().AddConn(t.Conn)：

func getMultiplexer() multiplexer {
	connMuxerOnce.Do(func() {
		connMuxer = &connMultiplexer{
			conns:  make(map[string]indexableConn),
			logger: utils.DefaultLogger.WithPrefix("muxer"),
		}
	})
	return connMuxer
}

func (m *connMultiplexer) AddConn(c indexableConn) {
	m.mutex.Lock()
	defer m.mutex.Unlock()

	connIndex := m.index(c.LocalAddr())
	p, ok := m.conns[connIndex]
	if ok {
		// 后续会去掉panic(...)
		panic("connection already exists") // TODO: write a nice message
	}
	m.conns[connIndex] = p
}

还有最后2行代码：协程启动的：
在这里插入图片描述

go t.listen(conn)
go t.runSendQueue()

在这里插入图片描述
这里先不展开如何处理数据包的逻辑，先把整体流程熟悉起来。

init方法终于结束了！还差最后一个newServer(...):

go s.run()
go s.runSendQueue()

在这里插入图片描述
正是transport.handlePacket(p receivedPacket)调用server.handlePacket(p receivedPacket)

func (s *baseServer) handlePacket(p receivedPacket) {
	select {
	case s.receivedPackets <- p: // 【看这里看这里看这里看这里】
	default:
		s.logger.Debugf("Dropping packet from %s (%d bytes). Server receive queue full.", p.remoteAddr, p.Size())
		if s.tracer != nil && s.tracer.DroppedPacket != nil {
			s.tracer.DroppedPacket(p.remoteAddr, logging.PacketTypeNotDetermined, p.Size(), logging.PacketDropDOSPrevention)
		}
	}
}

但是上面第299行的s.handlePacketImpl(p)很重要，后续篇再分析。
在这里插入图片描述

func (s *baseServer) sendRetry(p rejectedPacket) {
	if err := s.sendRetryPacket(p); err != nil {
		s.logger.Debugf("Error sending Retry packet: %s", err)
	}
}