etcd网络层(三)——stream实现过程

桃扇骨 2022-02-04 12:11 305阅读 0赞

stream消息通道主要是通过长连接的方式和对端进行数据交互，peer结构体中的stream相关数据如下(忽略V2版本)：

  1. type peer struct {
  3.        writer         *streamWriter   //负责向Stream消息通道中写消息
  5.        msgAppReader   *streamReader  //负责从Stream消息通道中读消息
  7.        .......其他字段省略
  9. }

从中我们看到stream是读写分离的，主要有streamWriter和streamReader结构体实现相关的功能。

stream实现的相关源码在 etcdserver/api/rafthttp/stream.go文件中。

#

一、streamWriter

streamWriter结构体如下：

  1.  type streamWriter struct {
  3.        lg *zap.Logger
  7.        localID types.ID   //本端的ID
  9.        peerID  types.ID  //对端节点的ID
  13.        status *peerStatus
  15.        fs     *stats.FollowerStats
  17.        r      Raft   //底层的Raft实例
  21.        mu      sync.Mutex // guard field working and closer
  23.        closer  io.Closer    //负责关闭底层的长连接
  25.        working bool   //负责标识当前的streamWriter是否可用
  29.        msgc  chan raftpb.Message   //Peer会将待发送的消息写入到该通道，streamWriter则从该通道中读取消息并发送出去
  31.        connc chan *outgoingConn   //通过该通道获取当前streamWriter实例关联的底层网络连接，  outgoingConn其实是对网络连接的一层封装，其中记录了当前连接使用的协议版本，以及用于关闭连接的Flusher和Closer等信息。
  33.        stopc chan struct{}
  35.        done  chan struct{}
  37. }

从中我们看到:

localID和peerID分别记录本端ID和对端ID

status记录当前peer的可用性状态

r 是底层的Raft实例，用于和底层的raft模块交互

closer负责关闭底层的长连接

msgc peer会将待发送到对端的消息(非快照数据)写入到该通道，streamWriter则从该通道中读取并发送到对端节点。

connc outgoingConn其实是对网络连接的一层封装，connc用于等待对端的主动连接，当对端主动和当前节点建立连接之后，本节点才会向对端发送数据。

1. streamWriter的初始化

streamWriter的初始化过程在startStreamWriter函数中

  1. func startStreamWriter(lg *zap.Logger, local, id types.ID, status *peerStatus, fs *stats.FollowerStats, r Raft) *streamWriter {
  3.        w := &streamWriter{
  5.               lg: lg,
  9.               localID: local,
  11.               peerID:  id,
  15.               status: status,
  17.               fs:     fs,
  19.               r:      r,
  21.               msgc:   make(chan raftpb.Message, streamBufSize),
  23.               connc:  make(chan *outgoingConn),
  25.               stopc:  make(chan struct{}),
  27.               done:   make(chan struct{}),
  29.        }
  31.        go w.run()
  33.        return w
  35. }

该方法主要是用于初始化streamWriter中的相关字段，并开启一个协程执行run方法。

2.run方法

run方法是streamWriter的核心，源码如下：

  1. func (cw *streamWriter) run() {
  3.        var (
  5.               msgc       chan raftpb.Message   //指向当前streamWriter.msgc字段
  7.               heartbeatc <-chan time.Time  //定时器会定时向该通道发送信号，触发心跳消息的发送，该心跳消息与后台介绍的Raft的心跳消息有所不同，该心跳的主要目的是为了防止连接长时间不用断开的
  9.               t          streamType    //用来记录消息的版本信息
  11.               enc        encoder  //编码器，负责将消息序列化并写入连接的缓冲区
  13.               flusher    http.Flusher  //负责刷新底层连接，将数据真正发送出去
  15.               batched    int   //当前未Flush的消息个数
  17.        )
  19.        tickc := time.NewTicker(ConnReadTimeout / 3)  //发送心跳的定时器
  21.        defer tickc.Stop()
  23.        unflushed := 0   //为Flush的字节数
  27.        if cw.lg != nil {
  29.               cw.lg.Info(
  31.                      "started stream writer with remote peer",
  33.                      zap.String("local-member-id", cw.localID.String()),
  35.                      zap.String("remote-peer-id", cw.peerID.String()),
  37.               )
  39.        } else {
  41.               plog.Infof("started streaming with peer %s (writer)", cw.peerID)
  43.        }
  47.        for {
  49.               select {
  51.               case <-heartbeatc:   //定时器到期，触发心跳消息
  53.                      err := enc.encode(&linkHeartbeatMessage)  //通过encoder将心跳编码并写入到writer
  55.                      unflushed += linkHeartbeatMessage.Size()  //增加未Flush出去的字节数
  57.                      if err == nil { //若没有异常，则使用flusher将缓存的消息全部发送出去，并重置batched和unflushed两个统计变量
  59.                             flusher.Flush()
  61.                             batched = 0
  63.                             sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
  65.                             unflushed = 0
  67.                             continue
  69.                      }
  71.                      //如果有异常则设置peer的status 为  不连通状态
  73.                      cw.status.deactivate(failureType{source: t.String(), action: "heartbeat"}, err.Error())
  77.                      sentFailures.WithLabelValues(cw.peerID.String()).Inc()
  79.                      cw.close()  //若发生异常，则关闭streamWriter,会导致底层连接的关闭
  81.                      if cw.lg != nil {
  83.                             cw.lg.Warn(
  85.                                    "lost TCP streaming connection with remote peer",
  87.                                    zap.String("stream-writer-type", t.String()),
  89.                                    zap.String("local-member-id", cw.localID.String()),
  91.                                    zap.String("remote-peer-id", cw.peerID.String()),
  93.                             )
  95.                      } else {
  97.                             plog.Warningf("lost the TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
  99.                      }
  101.                      //将heartbeatc和msgc两个通道清空，后续就不会在发送心跳消息和其他类型的消息了
  103.                      heartbeatc, msgc = nil, nil
  107.               case m := <-msgc:  //peer向streamWriter.msgc写入待发送的消息
  109.                      err := enc.encode(&m)  //将消息序列化并写入底层连接
  111.                      if err == nil {  //若没有异常，则递增unflushed变量
  113.                             unflushed += m.Size()
  117.                             //msgc通道中的消息全部发送完成或是未Flush的消息较多（超过streamBufSize的一半），则触发Flush，否则只是递增batched变量
  119.                             if len(msgc) == 0 || batched > streamBufSize/2 {
  121.                                    flusher.Flush()
  123.                                    sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
  125.                                    unflushed = 0
  127.                                    batched = 0
  129.                             } else {
  131.                                    batched++
  133.                             }
  137.                             continue
  139.                      }
  143.                      //若发生异常，则关闭streamWriter，清空heartbeatc和msgc两个通道
  147.                      cw.status.deactivate(failureType{source: t.String(), action: "write"}, err.Error())
  149.                      cw.close()
  151.                      if cw.lg != nil {
  153.                             cw.lg.Warn(
  155.                                    "lost TCP streaming connection with remote peer",
  157.                                    zap.String("stream-writer-type", t.String()),
  159.                                    zap.String("local-member-id", cw.localID.String()),
  161.                                    zap.String("remote-peer-id", cw.peerID.String()),
  163.                             )
  165.                      } else {
  167.                             plog.Warningf("lost the TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
  169.                      }
  171.                      heartbeatc, msgc = nil, nil
  173.                      cw.r.ReportUnreachable(m.To)
  175.                      sentFailures.WithLabelValues(cw.peerID.String()).Inc()
  181.               /*
  183.                      当其他节点(对端)主动与当前节点创建Stream消息通道时，会先通过StreamHandler的处理，
  185.                      StreamHandler会通过attach()方法将连接写入对应的peer.writer.connc通道，
  187.                      而当前的goroutine会通过该通道获取连接，然后开始发送消息
  189.               */
  191.               case conn := <-cw.connc: //获取与当前streamWriter实例绑定的底层连接
  193.                      cw.mu.Lock()
  195.                      closed := cw.closeUnlocked()
  197.                      t = conn.t    //获取该连接底层发送的消息版本，并创建相应的encoder实例
  199.                      switch conn.t {    //不同的版本创建不同的编码器，将http.ResponseWriter封装成messageEncoder，上层调用通过messageEncoder实例完成消息发送
  201.                      case streamTypeMsgAppV2:
  203.                             enc = newMsgAppV2Encoder(conn.Writer, cw.fs)
  205.                      case streamTypeMessage:
  207.                             enc = &messageEncoder{w: conn.Writer}
  209.                      default:
  211.                             plog.Panicf("unhandled stream type %s", conn.t)
  213.                      }
  215.                      if cw.lg != nil {
  217.                             cw.lg.Info(
  219.                                    "set message encoder",
  221.                                    zap.String("from", conn.localID.String()),
  223.                                    zap.String("to", conn.peerID.String()),
  225.                                    zap.String("stream-type", t.String()),
  227.                             )
  229.                      }
  231.                      flusher = conn.Flusher   //记录底层连接对应的Flusher
  233.                      unflushed = 0  //重置未Flush的字节数
  235.                      cw.status.activate()  //peerStatus.active设置为true
  237.                      cw.closer = conn.Closer  //负责关闭底层的长连接
  239.                      cw.working = true //标识当前streamWriter正在运行
  241.                      cw.mu.Unlock()
  245.                      if closed {
  247.                             if cw.lg != nil {
  249.                                    cw.lg.Warn(
  251.                                           "closed TCP streaming connection with remote peer",
  253.                                           zap.String("stream-writer-type", t.String()),
  255.                                           zap.String("local-member-id", cw.localID.String()),
  257.                                           zap.String("remote-peer-id", cw.peerID.String()),
  259.                                    )
  261.                             } else {
  263.                                    plog.Warningf("closed an existing TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
  265.                             }
  267.                      }
  269.                      if cw.lg != nil {
  271.                             cw.lg.Warn(
  273.                                    "established TCP streaming connection with remote peer",
  275.                                    zap.String("stream-writer-type", t.String()),
  277.                                    zap.String("local-member-id", cw.localID.String()),
  279.                                    zap.String("remote-peer-id", cw.peerID.String()),
  281.                             )
  283.                      } else {
  285.                             plog.Infof("established a TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
  287.                      }
  291.                      //更新heartbeatc和msgc两个通道，自此之后，才能发送消息
  293.                      heartbeatc, msgc = tickc.C, cw.msgc
  297.               case <-cw.stopc:
  299.                      if cw.close() {
  301.                             if cw.lg != nil {
  303.                                    cw.lg.Warn(
  305.                                           "closed TCP streaming connection with remote peer",
  307.                                           zap.String("stream-writer-type", t.String()),
  309.                                           zap.String("remote-peer-id", cw.peerID.String()),
  311.                                    )
  313.                             } else {
  315.                                    plog.Infof("closed the TCP streaming connection with peer %s (%s writer)", cw.peerID, t)
  317.                             }
  319.                      }
  321.                      if cw.lg != nil {
  323.                             cw.lg.Warn(
  325.                                    "stopped TCP streaming connection with remote peer",
  327.                                    zap.String("stream-writer-type", t.String()),
  329.                                    zap.String("remote-peer-id", cw.peerID.String()),
  331.                             )
  333.                      } else {
  335.                             plog.Infof("stopped streaming with peer %s (writer)", cw.peerID)
  337.                      }
  339.                      close(cw.done)
  341.                      return
  343.               }
  345.        }
  347. }

run方法的主要作用如下：

1.当其他节点主动与当前节点创建连接（即Stream消息通道底层使用的网络连接）时，该连接实例会写入对应的peer.writer.connc通道，

在streamWriter.run()方法中会通过该通道获取该连接实例并进行绑定，之后才能开始后续的消息发送

2.定时发送心跳，该心跳消息并不是前面介绍的etcd-raft模块时提到的MsgHeartbeat消息，而是为了防止底层连接超时的消息，为了维护长连接。

3.发送消息给对端，即msgc中等待发送给对端节点的消息。

3.attach方法

  1. func (cw *streamWriter) attach(conn *outgoingConn) bool {
  3.        select {
  5.        case cw.connc <- conn:  //将conn实例写入streamWriter.connc通道中
  7.               return true
  9.        case <-cw.done:
  11.               return false
  13.        }
  15. }

该方法主要用于接收对端的连接（该连接被封装成outgoingConn），并将outgoingConn实例并写入streamWriter.connc通道中。

#

二、streamReader实例

  1. type streamReader struct {
  3.        lg *zap.Logger
  7.        peerID types.ID   //对端节点的ID
  9.        typ    streamType    //关联的底层连接使用的协议版本信息
  13.        tr     *Transport   //关联的rafthttp.Transport实例
  15.        picker *urlPicker  //用于获取对端节点的可用的URL
  17.        status *peerStatus  //
  19.        recvc  chan<- raftpb.Message    //创建streamReader时会启动一个后台goroutine从peer.recvc通道中读取消息。从对端节点发送来的非MsgProp类型消息会写入到该通道中，然后由新建的goroutine读取出来，交给底层的etcd-raft模块进行处理
  21.        propc  chan<- raftpb.Message   //功能基本同上，只不过用于接收MsgProp类型的消息。
  25.        rl *rate.Limiter // alters the frequency of dial retrial attempts
  29.        errorc chan<- error
  33.        mu     sync.Mutex
  35.        paused bool   //是否暂停
  37.        closer io.Closer   //负责关闭底层的长连接
  41.        ctx    context.Context
  43.        cancel context.CancelFunc
  45.        done   chan struct{}
  47. }

主要功能是从Stream通道(底层连接)中读取消息

recvc 创建streamReader时会启动一个后台goroutine从peer.recvc通道中读取消息。从对端节点发送来的非MsgProp类型消息会写入到该通道中，然后由新建的goroutine读取出来，交给底层的etcd-raft模块进行处理。

propc 功能基本同上，只不过用于接收MsgProp类型的消息。

streamReader的初始化

  1.  func (cr *streamReader) start() {
  3.        cr.done = make(chan struct{})
  5.        if cr.errorc == nil {
  7.               cr.errorc = cr.tr.ErrorC
  9.        }
  11.        if cr.ctx == nil {
  13.               cr.ctx, cr.cancel = context.WithCancel(context.Background())
  15.        }
  17.        go cr.run()
  19. }

初始化 done、errorc及ctx、cancel等相关字段，然后开启一个协程执行run方法

  1. func (cr *streamReader) run() {
  3.        t := cr.typ   //获取使用的消息版本
  7.        //省略日志相关
  11.        for {
  13.               rc, err := cr.dial(t)  //向对端节点发送一个GET请求，然后获取并返回响应的ReadCloser，主要用于和对端建立连接
  15.               if err != nil {
  17.                      if err != errUnsupportedStreamType {
  19.                             cr.status.deactivate(failureType{source: t.String(), action: "dial"}, err.Error())
  21.                      }
  23.               } else {//如果未出现异常，则开始读取对端返回的消息，并将读取到的消息写入streamReader.recvc通道中
  25.                      cr.status.activate()
  29.                      err = cr.decodeLoop(rc, t)  //轮询读取消息
  31.                      //省略日志相关......                    
  35.                      switch {
  37.                      // all data is read out
  39.                      case err == io.EOF:
  41.                      // connection is closed by the remote
  43.                      case transport.IsClosedConnError(err):
  45.                      default:
  47.                             cr.status.deactivate(failureType{source: t.String(), action: "read"}, err.Error())
  49.                      }
  51.               }
  53.               // Wait for a while before new dial attempt
  55.               err = cr.rl.Wait(cr.ctx)
  57.               if cr.ctx.Err() != nil {
  59.                      if cr.lg != nil {
  61.                             cr.lg.Info(
  63.                                    "stopped stream reader with remote peer",
  65.                                    zap.String("stream-reader-type", t.String()),
  67.                                    zap.String("local-member-id", cr.tr.ID.String()),
  69.                                    zap.String("remote-peer-id", cr.peerID.String()),
  71.                             )
  73.                      } else {
  75.                             plog.Infof("stopped streaming with peer %s (%s reader)", cr.peerID, t)
  77.                      }
  79.                      close(cr.done)
  81.                      return
  83.               }
  85.               if err != nil {
  87.                      if cr.lg != nil {
  89.                             cr.lg.Warn(
  91.                                    "rate limit on stream reader with remote peer",
  93.                                    zap.String("stream-reader-type", t.String()),
  95.                                    zap.String("local-member-id", cr.tr.ID.String()),
  97.                                    zap.String("remote-peer-id", cr.peerID.String()),
  99.                                    zap.Error(err),
  101.                             )
  103.                      } else {
  105.                             plog.Errorf("streaming with peer %s (%s reader) rate limiter error: %v", cr.peerID, t, err)
  107.                      }
  109.               }
  111.        }
  113. }

run方法时streamReader的核心，首先调用dial方法给对端发送一个GET请求，从而和对端建立长连接。对端的Header收到该连接后，会封装成outgoingConn结构，并发送给streamWriter，即上面介绍的，当streamWriter收到新建的连接后，会用心跳维持该连接。

连接建立成功之后会调用decodeLoop方法来轮询读取对端节点发送过来的消息，并将收到的字节流转成raftpb.Message消息类型的格式。如果消息是心跳类型则忽略，如果是raftpb.MsgProp类型的消息，则发送到propc通道中，会有专门的协程发送给本节点底层的raft模块。如果是其他消息(注意这些消息里没有快照类型的消息，因为快照类型的消息是专门的通道)，则发送到recvc通道中，会有专门的协程将数据发送给底层的raft模块。

dial方法

  1. func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) {
  3.        u := cr.picker.pick()   //获取对端节点暴露的一个URL
  5.        uu := u
  7.        //根据使用的协议版本号和节点ID创建最终的URL地址
  9.        uu.Path = path.Join(t.endpoint(), cr.tr.ID.String())
  15.        //创建一个GET请求
  17.        req, err := http.NewRequest("GET", uu.String(), nil)
  19.        if err != nil {
  21.               cr.picker.unreachable(u)
  23.               return nil, fmt.Errorf("failed to make http request to %v (%v)", u, err)
  25.        }
  27.        //设置HTTP请求头
  29.        req.Header.Set("X-Server-From", cr.tr.ID.String())
  31.        req.Header.Set("X-Server-Version", version.Version)
  33.        req.Header.Set("X-Min-Cluster-Version", version.MinClusterVersion)
  35.        req.Header.Set("X-Etcd-Cluster-ID", cr.tr.ClusterID.String())
  37.        req.Header.Set("X-Raft-To", cr.peerID.String())
  41.        //通过设置Header，将当前节点暴露的URL也一起发送给对端节点
  43.        setPeerURLsHeader(req, cr.tr.URLs)
  47.        req = req.WithContext(cr.ctx)
  51.        cr.mu.Lock()
  53.        select {
  55.        case <-cr.ctx.Done():
  57.               cr.mu.Unlock()
  59.               return nil, fmt.Errorf("stream reader is stopped")
  61.        default:
  63.        }
  65.        cr.mu.Unlock()
  69.        //发送请求  ping-pone
  71.        resp, err := cr.tr.streamRt.RoundTrip(req)
  73.        if err != nil {
  75.               cr.picker.unreachable(u)
  77.               return nil, err
  79.        }
  83.        rv := serverVersion(resp.Header)
  85.        lv := semver.Must(semver.NewVersion(version.Version))
  87.        if compareMajorMinorVersion(rv, lv) == -1 && !checkStreamSupport(rv, t) {
  89.               httputil.GracefulClose(resp)
  91.               cr.picker.unreachable(u)
  93.               return nil, errUnsupportedStreamType
  95.        }
  99.        switch resp.StatusCode {
  103.        case http.StatusOK:
  105.               return resp.Body, nil
  107.      //省略其他错误处理
  109.        }
  111. }

主要负责与对端节点建立连接，其中包含了多种异常情况的处理。

看到创建GET请求，并设置HTTP请求头信息，调用RoundTrip发送请求到对端。

当返回成功时，将resp.Body返回。

decodeLoop方法

  1. func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
  3.        var dec decoder
  5.        cr.mu.Lock()
  7.        //根据使用的协议版本创建对应的decoder实例
  9.        switch t {
  11.        case streamTypeMsgAppV2:
  13.               dec = newMsgAppV2Decoder(rc, cr.tr.ID, cr.peerID)
  15.        case streamTypeMessage:
  17.               dec = &messageDecoder{r: rc}   //主要负责从连接中读取消息
  19.        default:
  21.               if cr.lg != nil {
  23.                      cr.lg.Panic("unknown stream type", zap.String("type", t.String()))
  25.               } else {
  27.                      plog.Panicf("unhandled stream type %s", t)
  29.               }
  31.        }
  35.        //检测是否已经暂停从该连接上读取消息
  37.        select {
  39.        case <-cr.ctx.Done():
  41.               cr.mu.Unlock()
  43.               if err := rc.Close(); err != nil {
  45.                      return err
  47.               }
  49.               return io.EOF
  51.        default:
  53.               cr.closer = rc
  55.        }
  57.        cr.mu.Unlock()
  61.        // gofail: labelRaftDropHeartbeat:
  63.        for {
  65.               //从底层连接中读取数据，并反序列化成raftpb.Message实例
  67.               m, err := dec.decode()
  69.               if err != nil {
  71.                      cr.mu.Lock()
  73.                      cr.close()
  75.                      cr.mu.Unlock()
  77.                      return err
  79.               }
  83.               // gofail-go: var raftDropHeartbeat struct{}
  85.               // continue labelRaftDropHeartbeat
  87.               receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(m.Size()))
  91.               cr.mu.Lock()
  93.               paused := cr.paused
  95.               cr.mu.Unlock()
  99.               if paused {
  101.                      continue
  103.               }
  107.               // 忽略连接层的心跳消息，注意与后面介绍的Raft的心跳消息进行区分
  109.               if isLinkHeartbeatMessage(&m) {
  111.                      // raft is not interested in link layer
  113.                      // heartbeat message, so we should ignore
  115.                      // it.
  117.                      continue
  119.               }
  123.               //根据消息类型获取要写入的通道
  125.               recvc := cr.recvc
  127.               if m.Type == raftpb.MsgProp {
  129.                      recvc = cr.propc
  131.               }
  135.               select {
  137.               case recvc <- m:  //将消息写入到对应的通道中，之后交给底层的Raft状态机进行处理
  139.               default:
  141.               //省略异常处理
  143.               }
  145.        }
  147. }

该方法的主要作用就是 从底层的网络连接读取数据并进行反序列化，之前将得到的消息写入到recvc通道或propc通道中，等待Peer进行处理。

发表评论

表情：
评论列表 （有 0 条评论，305人围观）

还没有评论，来说两句吧...

相关阅读