etcd网络层(二)——pipeline通道实现

古城微笑少年丶 2022-02-04 12:09 229阅读 0赞

### 一、pipeline结构体 ###

pipeline通道主要负责传递快照数据

pipeline结构体及相关字段如下：

type pipeline struct {
    
           peerID types.ID   //该pipeline对应节点的ID
    
    
    
           tr     *Transport  //关联的rafthttp.Transport实例
    
           picker *urlPicker  //用于选择可用的url
    
           status *peerStatus  //当前peer的状态
    
           raft   Raft   //底层的raft实例
    
           errorc chan error
    
           // deprecate when we depercate v2 API
    
           followerStats *stats.FollowerStats
    
    
    
           msgc chan raftpb.Message  //pipeline实例从该通道中获取待发送的消息
    
           // wait for the handling routines
    
           wg    sync.WaitGroup   //负责同步多个goroutine结束。每个pipeline默认开启4个goroutine来处理msgc中的消息，必须先关闭这些goroutine，才能真正关闭该pipeline
    
           stopc chan struct{}
    
    }

tr 关联的rafthttp.Transport实例

msgc 该channel中存放待发送到对端的消息(快照消息)

raft 底层的raft实例，用于向底层的raft发送消息(如：快照发送成功或失败等)

### 二、start方法 ###

func (p *pipeline) start() {
    
           p.stopc = make(chan struct{})
    
           p.msgc = make(chan raftpb.Message, pipelineBufSize)  //初始化msgc通道，默认缓冲是64个
    
           p.wg.Add(connPerPipeline)
    
           for i := 0; i < connPerPipeline; i++ {  //默认开启4个goroutine来处理msgc中待发送的消息
    
                  go p.handle()
    
           }
    
    
    
           if p.tr != nil && p.tr.Logger != nil {
    
                  p.tr.Logger.Info(
    
                         "started HTTP pipelining with remote peer",
    
                         zap.String("local-member-id", p.tr.ID.String()),
    
                         zap.String("remote-peer-id", p.peerID.String()),
    
                  )
    
           } else {
    
                  plog.Infof("started HTTP pipelining with peer %s", p.peerID)
    
           }
    
    }

msgc通道的默认缓冲是64个，默认开启4个goroutine来处理msgc中待发送的消息，并将消息发送给对端节点。

**handle方法处理msgc中待发送的消息**

func (p *pipeline) handle() {
    
           defer p.wg.Done()
    
    
    
           for {
    
                  select {
    
                  case m := <-p.msgc:  //获取待发送的MsgSnap类型的消息
    
                         start := time.Now()
    
                         err := p.post(pbutil.MustMarshal(&m))  //将消息序列化，然后创建HTTP请求并发送出去
    
                         end := time.Now()
    
    
    
                         if err != nil {
    
                                //通知不可达
    
                                p.status.deactivate(failureType{source: pipelineMsg, action: "write"}, err.Error())
    
    
    
                                if m.Type == raftpb.MsgApp && p.followerStats != nil {
    
                                       p.followerStats.Fail()
    
                                }
    
                                p.raft.ReportUnreachable(m.To)  //通知底层的etcd-raft模块，当前节点与指定的节点无法连通
    
                                if isMsgSnap(m) {  //快照数据则向状态机报告发送失败
    
                                       p.raft.ReportSnapshot(m.To, raft.SnapshotFailure)
    
                                }
    
                                sentFailures.WithLabelValues(types.ID(m.To).String()).Inc()
    
                                continue
    
                         }
    
    
    
                         p.status.activate()  //连接状态为连通active
    
                         if m.Type == raftpb.MsgApp && p.followerStats != nil {
    
                                p.followerStats.Succ(end.Sub(start))
    
                         }
    
                         if isMsgSnap(m) {  //向底层raft-node状态机发送成功的消息
    
                                p.raft.ReportSnapshot(m.To, raft.SnapshotFinish)
    
                         }
    
                         sentBytes.WithLabelValues(types.ID(m.To).String()).Add(float64(m.Size()))
    
                  case <-p.stopc:
    
                         return
    
                  }
    
           }
    
    }

循环处理msgc通道中待发送的消息，然后调用pipeline.post()方法将其发送出去，发送结束之后会调用底层的Raft接口的响应方法报告发送结果。

从handle方法中我们看到，当收到待发送的MsgSnap类型的消息后，调用post方法，使用HTTP POST的请求将消息发送出去。如果发送失败，则通知底层的raft模块，该消息发送失败。如果发送成功则会将当前peer的status设为active，并通知底层的raft模块，该消息发送成功。

func (p *pipeline) post(data []byte) (err error) {
    
           u := p.picker.pick()  //获取对端暴露的url地址
    
           //创建HTTP POST请求的Request
    
           req := createPostRequest(u, RaftPrefix, bytes.NewBuffer(data), "application/protobuf", p.tr.URLs, p.tr.ID, p.tr.ClusterID)
    
    
    
           done := make(chan struct{}, 1)  //通知下面的goroutine请求是否已经发送完成
    
           ctx, cancel := context.WithCancel(context.Background())
    
           req = req.WithContext(ctx)
    
           go func() {  //该goroutine主要用于监听请求是否需要取消
    
                  select {
    
                  case <-done:
    
                  case <-p.stopc:  //如果请求过程中，pipeline被关闭，则取消该请求
    
                         waitSchedule()
    
                         cancel()  //取消请求
    
                  }
    
           }()
    
    
    
           resp, err := p.tr.pipelineRt.RoundTrip(req)  //发送上述HTTP POST请求，并获取到对应的的响应
    
           done <- struct{}{}  //通知上述goroutine，请求已经发送完毕
    
           if err != nil {
    
                  p.picker.unreachable(u)
    
                  return err
    
           }
    
           b, err := ioutil.ReadAll(resp.Body)  //读取HTTP Response.Body内容
    
           if err != nil {
    
                  p.picker.unreachable(u)  //出现异常则将该URL标识为不可用
    
                  return err
    
           }
    
           resp.Body.Close()
    
    
    
           err = checkPostResponse(resp, b, req, p.peerID)  //检测响应的内容
    
           if err != nil {
    
                  p.picker.unreachable(u)
    
                  // errMemberRemoved is a critical error since a removed member should
    
                  // always be stopped. So we use reportCriticalError to report it to errorc.
    
                  if err == errMemberRemoved {
    
                         reportCriticalError(err, p.errorc)
    
                  }
    
                  return err
    
           }
    
    
    
           return nil
    
    }

post方法使用HTTP POST请求将快照数据发送给对端HTTP服务器，如果发送成功则返回nil，否则返回error。

以上就是pipeline的常用方法，我看到pipeline通道主要是通过HTTP POST的方式发送快照数据给对端。