typha

1.server

1.1.start

• s.Start()作为server向客户端typha提供配置广播服务，本质上基于连接监控及优雅退出避免server端连接突峰及下线过程的连接排空。

• func (s *Server) Start(cxt context.Context) {
  	...
  	go s.serve(cxt)
      // 动态控制连接数不超过MaxConns
  	go s.governNumberOfConnections(cxt)
  	go s.handleGracefulShutDown(cxt, cancelFn)
  }
  
  func (s *Server) governNumberOfConnections(cxt context.Context) {
  	...
      
      // default maxConns
  	maxConns := s.config.MaxConns
      ...
      // [100ms,1s]触发
  	ticker := jitter.NewTicker(dropInterval, dropInterval/10)
  	...
      
  	for {
  		select {
         	// maxConns变化
  		case newMax := <-s.maxConnsC:
  			if newMax == maxConns {
  				continue
  			}
              
              // 更新
  			maxConns = newMax
          // 定时触发
  		case <-ticker.C:
              // 当前活跃连接数
  			numConns := s.NumActiveConnections()
              // 超出最大连接限制
  			if numConns > maxConns {
                  // 随机踢掉一个连接
  				s.TerminateRandomConnection(logCxt, "re-balance load with other Typha instances")
  				...
  			}
  		case <-cxt.Done():
  			return
  		case <-healthTicks:
  			s.reportHealth()
  		}
  	}
  }
  
  func (s *Server) handleGracefulShutDown(cxt context.Context, serverCancelFn context.CancelFunc) {
  	defer s.Finished.Done()
  
  	select {
      // 优雅退出
  	case <-s.shutdownC:
  		logCxt.Info("Graceful shutdown triggered, starting to close connections...")
      // server退出
  	case <-cxt.Done():
  		return
  	}
  
      // 活跃连接数
  	numConns := s.NumActiveConnections()
      
      // 无活跃连接,server直接退出
  	if numConns == 0 {
  		serverCancelFn()
  		return
  	}
  
  	// Aim to close connections within 95% of the allotted time.
  	dropInterval := s.config.ShutdownTimeout * 95 / 100 / time.Duration(numConns)
  	if dropInterval > s.config.ShutdownMaxDropInterval {
  		// We have a long time to shut down (say 5 minutes) but only a few connections.  Cap the delay between
  		// dropping connections.
  		dropInterval = s.config.ShutdownMaxDropInterval
  	}
      
      // 实例化定时器
  	ticker := jitter.NewTicker(dropInterval*95/100, dropInterval*10/100)
  	for {
  		select {
          // 周期触发连接关闭
  		case <-ticker.C:
              // 活跃连接
  			numConns := s.NumActiveConnections()
              // 随机踢掉一个连接
  			dropped := s.TerminateRandomConnection(logCxt, "graceful shutdown in progress")
              
              // 连接排空或无连接可踢
  			if numConns <= 1 || !dropped {
  				serverCancelFn()
  				return
  			}
  		case <-cxt.Done():
  			return
  		}
  	}
  }
  

• 

• dropInterval

  $$DropInterval=\min\left(\frac{285}{NumConns},\ 1\right)\text{ seconds}$$

1.2.serve

• s.serve()会实例化tcp server，基于G->Conn模型实现连接处理，基于context+channel机制管理连接生命周期及优雅关闭。

• func (s *Server) serve(cxt context.Context) {
  	...
  
      // TLS认证
  	if s.config.requiringTLS() {
          // tls config整理
  		pwd, _ := os.Getwd()
  		cert, tlsErr := tls.LoadX509KeyPair(s.config.CertFile, s.config.KeyFile)
  		...
          
  		tlsConfig := calicotls.NewTLSConfig(s.config.FIPSModeEnabled)
  		tlsConfig.Certificates = []tls.Certificate{cert}
  
  		// Arrange for server to verify the clients' certificates.
  		tlsConfig.ClientAuth = tls.RequireAndVerifyClientCert
  		caPEMBlock := os.ReadFile(s.config.CAFile)
  		...
          
  		tlsConfig.ClientCAs = x509.NewCertPool()
  		ok := tlsConfig.ClientCAs.AppendCertsFromPEM(caPEMBlock)
  		...
          
  		tlsConfig.VerifyPeerCertificate = tlsutils.CertificateVerifier(
  			logCxt, tlsConfig.ClientCAs, s.config.ClientCN, s.config.ClientURISAN)
  
          // 初始化listener
  		laddr := fmt.Sprintf("0.0.0.0:%v", s.config.ListenPort())
  		l, err = tls.Listen("tcp", laddr, tlsConfig)
      // 直接初始化listener
  	} else {
  		l, err = net.ListenTCP("tcp", &net.TCPAddr{Port: s.config.ListenPort()})
  	}
  	...
  
      // server退出会关闭listener,避免新连接
  	go func() {
  		select {
  		case <-cxt.Done():
  			log.Info("Context finished, closing listen socket.")
  		case <-s.shutdownC:
  			log.Info("Graceful shutdown triggered, closing listen socket.")
  		}
  		l.Close()
  		...
  	}()
  
  	chosenPort := l.Addr().(*net.TCPAddr).Port
  	s.lock.Lock()
  	s.chosenPort = chosenPort
  	s.lock.Unlock()
  	close(s.listeningC)
      
  	for {
          // 获取连接
  		conn, err := l.Accept()
  		...
  
  		connID := s.nextConnID
  		s.nextConnID++
  		...
  		var connW io.Writer = conn
  		...
          
          // 实例化connection对象
  		connection := &connection{
  			ID:              connID,
  			config:          &s.config,
  			allCaches:       s.caches,
  			allSnapshotters: s.binSnapCaches,
  			cxt:             connCxt,
  			cancelCxt:       cancel,
  			conn:            conn,
  			connW:           connW,
  			...
  			encoder:     gob.NewEncoder(connW),
  			flushWriter: func() error { return nil },
  			readC:       make(chan interface{}),
  			...
  		}
          
  		// 注册到connIDToConn
  		s.recordConnection(connection)
  		...
          
  		go func() {
              // 流量处理
  			connection.handle(&s.Finished)
  			...
  		}()
          
  		// Clean up the entry in connIDToConn as soon as the context is canceled.
  		go func() {
  			<-connCxt.Done()
  			s.discardConnection(connection)
  			s.Finished.Done()
  		}()
  	}
  }
  

• 

• 注意

  tcp server基于listener监听connection，基于wrapper connection异步处理数据推送及优雅关闭

1.3.handle

• h.handle()负责建立连接、握手、同步快照、增量推送及保活检查，基于高性能长连接的客户端同步实现可靠的实时数据连接推送。

• func (h *connection) handle(finishedWG *sync.WaitGroup) (err error) {
  	defer func() {
  		// 退出关闭连接
  		...
  		h.conn.Close()
  		...
  	}()
  	...
  
  	// conn-->msg-->readC
  	go h.readFromClient(h.logCxt.WithField("thread", "read"))
  
  	// handle hello msg and backfill conn info.
  	ih.doHandshake()
      ...
  
  	// client支持重启decoder
  	if h.clientSupportsDecoderRestart {
          // 获取cache实例(前面介绍过)
  		binSnapCache = h.allSnapshotters[h.chosenCompression][h.syncerType]
  		...
          
          // 协商出压缩算法 || cache快照不为空
  		if len(reasonsToRestart) > 0 {
  			// send decoder restart msg.If ack, renew encoder
  			h.restartEncodingIfSupported(strings.Join(reasonsToRestart, ";"))
  			...
  		}
  	}
  	...
  
      // 支持压缩算法
  	if binSnapCache != nil {
          ...
  		// 直接发送cache快照
  		breadcrumb, err = binSnapCache.SendSnapshot(h.cxt, h.connW, h.conn)
  		...
  
  		// wait for the ACK with 60s timeout.
  		h.waitForAckAndRestartEncoder()
  		...
      // cache快照失效/无法协商压缩算法
  	} else {
  		// 流式发送
  		breadcrumb = h.cache.CurrentBreadcrumb()
  		h.streamSnapshotToClient(h.logCxt, breadcrumb)
  		...
  	}
  	...
      
  	// 增量变更同步
  	go h.sendDeltaUpdatesToClient(h.logCxt.WithField("thread", "kv-sender"), breadcrumb)
  	...
      
  	// Ping心跳探测
  	go h.sendPingsToClient(h.logCxt.WithField("thread", "pinger"))
  
  	// receive pongs in a timely fashion.
  	pongTicker := jitter.NewTicker(h.config.PingInterval/2, h.config.PingInterval/10)
  	...
      lastPongReceived := time.Now()
  
  	// wait for client messages and do the ping/pong liveness check.
  	for {
  
  		select {
          // response msg
  		case msg := <-h.readC:
  			if msg == nil {
  				return h.cxt.Err()
  			}
              
  			switch msg := msg.(type) {
              // pong延迟统计
  			case syncproto.MsgPong:
  				lastPongReceived = time.Now()
  				h.summaryPingLatency.Observe(time.Since(msg.PingTimestamp).Seconds())
  			...
  			}
          // 5s延迟,1s抖动
  		case <-pongTicker.C:
              // 超出60s未回复心跳
  			since := time.Since(lastPongReceived)
  			if since > h.config.PongTimeout {
  				return errors.New("No pong received from client")
  			}
  		...
  		}
  	}
  }
  
  // reads messages from client and puts them on the h.readC channel. It is responsible for closing the channel.
  func (h *connection) readFromClient(logCxt *log.Entry) {
  	defer func() {
  		...
  		close(h.readC)
  		...
  	}()
      
  	r := gob.NewDecoder(h.conn)
  	for {
  		...
          // 读取消息
  		r.Decode(&envelope)
  		...
          
  		if envelope.Message == nil {
  			break
  		}
          
  		select {
          // msg-->readC
  		case h.readC <- envelope.Message:
  		case <-h.cxt.Done():
  			return
  		}
  
  	}
  }
  

• 

• 注意

  client刚连接发送hello msg握手，协商压缩算法及订阅资源类型，server将cache内容推一份给client，后续异步发送增量及心跳保活

2.handler

2.1.sendsnap

• s.SendSnapshot()用于初始化snapcache及生成buf压缩内容，将snapcache buf基于进度协调完整发送给刚连接的connection。

• // waits for a binary snapshot to be ready and then sends it as a raw snappy-compressed gob stream
  // on the given connection.
  func (s *SnappySnapshotCache) SendSnapshot(..., w io.Writer, conn WriteDeadlineSetter) (*Breadcrumb, error) {
  	// 获取snapcache
  	snap := s.activeBinarySnapshot()
  	...
      
      // 发送到client
      snap.sendToClient(ctx, s.logCtx, w, conn, s.writeTimeout)
      ...
      
  	return snap.crumb, nil
  }
  
  // returns the current active snapshot (which may still be being created on a background
  // goroutine), or it starts a new snapshot.
  func (s *SnappySnapshotCache) activeBinarySnapshot() *snapshot {
  	...
  
      // 初始化snapcache
  	if s.activeSnapshot == nil {
          // 最新版本数据
  		breadcrumb := s.cache.CurrentBreadcrumb()
          // 预估缓存数据(1.1倍扩容)
  		bufSize := s.lastSnapSize * 110 / 100
  		...
  		if bufSize == 0 {
  			bufSize = 128 * 1024
  		}
          
          // 实例化activeSnapshot
  		s.activeSnapshot = &snapshot{ crumb: breadcrumb, buf: multireadbuf.New(bufSize) }
          // activeSnapshot.breadcrumb.cache压缩写入buf,末尾追加decoder restart msg
          // 基于condition阻塞至cache更新,清理snap buf
  		go s.populateSnapshot(s.activeSnapshot)
  	}
      ...
      
  	return s.activeSnapshot
  }
  
  func (s *snapshot) sendToClient(... w io.Writer, conn WriteDeadlineSetter, writeTimeout time.Duration) error {
  	...
      
  	reader := s.buf.Reader()
  	...
  	for ctx.Err() == nil {
  		// currentWriteDeadline设置
  		if time.Until(currentWriteDeadline) < writeTimeout {
  			newDeadline := time.Now().Add(120 * 110 / 100)
  			conn.SetWriteDeadline(newDeadline)
  			...
              
  			currentWriteDeadline = newDeadline
  		}
  
          // buf-->conn
  		n, err := reader.WriteTo(w)
  		...
          
  		break // WriteTo returns nil if all data was written, not EOF.
  	}
  	...
      
      // 重置conn deadline
  	conn.SetWriteDeadline(zeroTime)
  	...
  
  	return ctx.Err()
  }
  
  func (h *connection) waitForAckAndRestartEncoder() error {
      // readC->msg
  	msg, err := h.waitForMessage(h.logCxt, h.config.PongTimeout)
  	...
      
  	ack, ok := msg.(syncproto.MsgACK)
  	...
      
  	// Upgrade to compressed connection if required.
  	bw := bufio.NewWriter(h.connW)
  	switch h.chosenCompression {
      // 带压缩的encoder
  	case syncproto.CompressionSnappy:
  		w := snappy.NewBufferedWriter(bw)
  		h.encoder = gob.NewEncoder(w) // Need a new Encoder, there's no way to change out the Writer.
  		h.flushWriter = func() error {
  			err := w.Flush()
  			if err != nil {
  				return err
  			}
  			return bw.Flush()
  		}
      // 普通encoder
  	default:
  		h.encoder = gob.NewEncoder(bw) // Need a new Encoder, there's no way to change out the Writer.
  		h.flushWriter = bw.Flush
  	}
      
  	return nil
  }
  

• 

• 注意

  snapcache buffer会边生成边发送，基于chunk模式分段写入encoder缓冲区，触发flush完整发送

2.2.streamsnap

• h.streamSnapshotToClient()未压缩发送，会将snapcache.kvs拆分为多批发送到connection缓冲区，执行flush刷新发送到client。

• // streamSnapshotToClient takes the snapshot contained in the Breadcrumb and streams it to the client in chunks.
  func (h *connection) streamSnapshotToClient(logCxt *log.Entry, breadcrumb *snapcache.Breadcrumb) error {
  	writeSnapshotMessages(h.cxt, ..., breadcrumb, h.sendMsg, h.config.MaxMessageSize)
  	...
  
  	return nil
  }
  
  // chunks the given breadcrumb up into syncproto.MsgKVs objects and calls writeMsg for each one.
  func writeSnapshotMessages(..., breadcrumb *Breadcrumb, writeMsg func(any) error, maxMsgSize int) (err error) {
  	...
  	// writeKVs is a utility function that sends the kvs buffer to the client and clears the buffer.
  	writeKVs := func() error {
  		if len(kvs) == 0 {
  			return nil
  		}
  		...
  		
          // send msg.
          writeMsg(syncproto.MsgKVs{ KVs: kvs })
  		...
          
  		kvs = kvs[:0]
  		return err
  	}
  
  	breadcrumb.KVs.Ascend(func(entry syncproto.SerializedUpdate) bool {
  		...
          
          // 合并至一批再发送
  		kvs = append(kvs, entry)
  		if len(kvs) >= 100 {
  			// Buffer is full, send the next batch.
  			writeKVs()
  			...
  		}
  		return true
  	})
  	...
  	
      // 剩余msg发送
  	writeKVs()
  	...
      
  	return
  }
  
  // sendMsg sends a message to the client.  It may be called from multiple goroutines.
  func (h *connection) sendMsg(msg interface{}) error {
  	...
      
  	envelope := syncproto.Envelope{ Message: msg }
  	...
  
  	// 设置写超时
  	h.maybeResetWriteTimeout()
      ...
  
      // 编码写入缓冲
  	h.encoder.Encode(&envelope)
      ...
      
      // 发送
  	h.flushWriter()
      ...
      
  	return nil
  }
  

• 

• 注意

  snapcache会将当前快照的内容以KV存到KV Tree，未使用压缩算法会将KV整理为多批(100长度)发送到客户端

2.3.senddelta

• h.sendDeltaUpdatesToClient()会持续将cache deltas增量发送给client，同步落后时会基于宽限期追赶发送一批deltas数据。

• // follows the breadcrumbs from the given one, sending delta updates from each subsequent breadcrumb to client.
  func (h *connection) sendDeltaUpdatesToClient(logCxt *log.Entry, breadcrumb *snapcache.Breadcrumb) {
  	defer func() {
          // 断开conn
  		h.cancelCxt()
  		h.shutDownWG.Done()
  	}()
  
  	// finished sending the snapshot, calculate grace time for the client to catch up to a recent breadcrumb.
  	gracePeriodEndTime := time.Now().Add(300s)
  	...
      
  	// Track the sync status reported in each Breadcrumb so we can send an update if it changes.
  	maybeSendStatus := func() (err error) {
          // cache状态变化
  		if lastSentStatus != breadcrumb.SyncStatus {
              // 发送到client
  			h.sendMsg(syncproto.MsgSyncStatus{ SyncStatus: breadcrumb.SyncStatus })
  			...
              
  			lastSentStatus = breadcrumb.SyncStatus
  		}
  		return
  	}
  
  	// The first Breadcrumb may have changed the status.  Send an update if so.
  	maybeSendStatus()
      ...
  
  	loggedClientBehind := false
  	for h.cxt.Err() == nil {
  		...
          
  		for len(deltas) < 100 {
              ...
              
              // 基于condition阻塞获取newCache(last.next)
  			breadcrumb, err = breadcrumb.Next(h.cxt)
  			...
  
  			// cache最新状态
  			latestCrumb := h.cache.CurrentBreadcrumb()
              
              // 间隔有效期
  			crumbAge := latestCrumb.Timestamp.Sub(breadcrumb.Timestamp)
  			...
  			
  			// breadcrumb落后5min
  			if crumbAge > 300s {
                  // 超出最大宽限期(给宽限期是为了适当追赶,避免断开-->重连-->落后-->断开死循环)
  				if time.Now().After(gracePeriodEndTime) {
  					return
  				}
                  ...
  			}
              ...
  
              // breadcrumb基本赶上
  			if crumbAge < 100ms && deltas == nil {
  				// 直接发当前版本deltas
  				deltas = breadcrumb.Deltas
  				break
  			}
  
  			// 合并落后的deltas
  			deltas = append(deltas, breadcrumb.Deltas...)
  			...
  
              // 基本追上再批量发
  			if crumbAge < 100ms {
  				// Caught up, stop batching.
  				break
  			}
  		}
  
          // 发送deltas
  		if len(deltas) > 0 {
              ...
  			// Send the deltas relative to the previous snapshot.
  			h.sendMsg(syncproto.MsgKVs{ KVs: deltas })
  			...
  		}
  
  		// Newest breadcrumb may have updated the sync status, send an update if so.
  		maybeSendStatus()
          ...
  	}
  }
  

• 

• 注意

  gracePeriodEndTime是conn连接的同步宽限期，允许落后太多版本数据时先同步一部分，避免连接陷入同步断连死循环

2.4.sendping

• h.sendPingsToClient()会定期向client发送Ping心跳，用于检测连接存活状态及维持长连接的健康性，超出60s没有Pong会断开连接。

• // sendPingsToClient loops, sending pings to the client at the configured interval.
  func (h *connection) sendPingsToClient(logCxt *log.Entry) {
  	defer func() {
          // conn关闭收尾
  		h.cancelCxt()
  		h.shutDownWG.Done()
  	}()
      
      // 心跳间隔定时器
  	pingTicker := jitter.NewTicker(10s, 1s)
  	defer func() {
  		pingTicker.Stop()
  	}()
      
  	for {
  		select {
          // 发送心跳
  		case <-pingTicker.C:
  			h.sendMsg(syncproto.MsgPing{ Timestamp: time.Now() })
  			...
          // 断开连接
  		case <-h.cxt.Done():
  			return
  		}
  	}
  }
  

• 注意

  Ping心跳间隔10s发送一次，Pong心跳间隔60s必须接收一次，确保连接存活