ETCD中raft如何实现线性一致性

func (kv *kv) Get(ctx context.Context, key string, opts ...OpOption) (*GetResponse, error) {
	r, err := kv.Do(ctx, OpGet(key, opts...))
	return r.get, toErr(ctx, err)
}

// OpGet returns "get" operation based on given key and operation options.
func OpGet(key string, opts ...OpOption) Op {
	// WithPrefix and WithFromKey are not supported together
	if IsOptsWithPrefix(opts) && IsOptsWithFromKey(opts) {
		panic("`WithPrefix` and `WithFromKey` cannot be set at the same time, choose one")
	}
	ret := Op{t: tRange, key: []byte(key)}
	ret.applyOpts(opts)
	return ret
}

func (kv *kv) Do(ctx context.Context, op Op) (OpResponse, error) {
	var err error
	switch op.t {
	case tRange:
		var resp *pb.RangeResponse
		resp, err = kv.remote.Range(ctx, op.toRangeRequest(), kv.callOpts...)
		if err == nil {
			return OpResponse{get: (*GetResponse)(resp)}, nil
		}
		...
	}
	return OpResponse{}, toErr(ctx, err)
}

func (c *kVClient) Range(ctx context.Context, in *RangeRequest, opts ...grpc.CallOption) (*RangeResponse, error) {
	out := new(RangeResponse)
	err := c.cc.Invoke(ctx, "/etcdserverpb.KV/Range", in, out, opts...)
	if err != nil {
		return nil, err
	}
	return out, nil
}

service KV {
  // Range gets the keys in the range from the key-value store.
  rpc Range(RangeRequest) returns (RangeResponse) {
      option (google.api.http) = {
        post: "/v3/kv/range"
        body: "*"
    };
  }
}

// etcd/server/etcdserver/v3_server.go
func (s *EtcdServer) Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error) {
	...
	// 判断是否需要serializable read  
	// Serializable为true表示需要serializable read
	// serializable read 会直接读取当前节点的数据返回给客户端，它并不能保证返回给客户端的数据是最新的  
	// Serializable为false表示需要linearizable read
	// Linearizable Read 需要阻塞等待直到读到最新的数据
	if !r.Serializable {
		err = s.linearizableReadNotify(ctx)
		trace.Step("agreement among raft nodes before linearized reading")
		if err != nil {
			return nil, err
		}
	}
	...
	return resp, err
}

// etcd/server/etcdserver/v3_server.go
func (s *EtcdServer) linearizableReadNotify(ctx context.Context) error {
	s.readMu.RLock()
	nc := s.readNotifier
	s.readMu.RUnlock()

	// linearizableReadLoop会阻塞监听readwaitc
	// 这边写入一个空结构体到readwaitc中，linearizableReadLoop就会开始结束阻塞开始工作  
	select {
	case s.readwaitc <- struct{}{}:
	default:
	}

	// wait for read state notification
	select {
	case <-nc.c:
		return nc.err
	case <-ctx.Done():
		return ctx.Err()
	case <-s.done:
		return ErrStopped
	}
}

// start会启动一个linearizableReadLoop
func (s *EtcdServer) Start() {
	...
	s.GoAttach(s.linearizableReadLoop)
	...
}

// etcd/server/etcdserver/v3_server.go
func (s *EtcdServer) linearizableReadLoop() {
	for {
		requestId := s.reqIDGen.Next()
		leaderChangedNotifier := s.LeaderChangedNotify()
		select {
		case <-leaderChangedNotifier:
			continue
		// 在client发起一次Linearizable Read的时候，会向readwaitc写入一个空的结构体作为信号
		case <-s.readwaitc:
		case <-s.stopping:
			return
		}
		...
		// 处理不同的消息
		// 这里会监听readwaitc，等待MsgReadIndex信息的处理结果 
		// 同时获取当前已提交的日志索引
		confirmedIndex, err := s.requestCurrentIndex(leaderChangedNotifier, requestId)
		if isStopped(err) {
			return
		}
		if err != nil {
			nr.notify(err)
			continue
		}

		...
		// 此处是重点
		// 等待 apply index >= read index
		if appliedIndex < confirmedIndex {
			select {
			case <-s.applyWait.Wait(confirmedIndex):
			case <-s.stopping:
				return
			}
		}
		// 发出可以进行读取状态机的信号
		nr.notify(nil)
		...
	}
}

// etcd/server/etcdserver/v3_server.go
func (s *EtcdServer) sendReadIndex(requestIndex uint64) error {
	...
	err := s.r.ReadIndex(cctx, ctxToSend)
	...
	return nil
}

// etcd/raft/node.go
func (n *node) ReadIndex(ctx context.Context, rctx []byte) error {
	return n.step(ctx, pb.Message{Type: pb.MsgReadIndex, Entries: []pb.Entry})
}

// etcd/raft/raft.go
func stepFollower(r *raft, m pb.Message) error {
	switch m.Type {
		...
	case pb.MsgReadIndex:
		if r.lead == None {
			r.logger.Infof("%x no leader at term %d; dropping index reading msg", r.id, r.Term)
			return nil
		}
		// 目标为leader
		m.To = r.lead
		// 转发信息
		r.send(m)
	}
	...
	return nil
}

// etcd/raft/raft.go
func stepLeader(r *raft, m pb.Message) error {
	// These message types do not require any progress for m.From.
	switch m.Type {
	case pb.MsgReadIndex:
		...
		sendMsgReadIndexResponse(r, m)

		return nil
	}

	return nil
}

// raft结构体中的readOnly作用是批量处理只读请求，只读请求有两种模式，分别是ReadOnlySafe和ReadOnlyLeaseBased
// ReadOnlySafe是ETCD作者推荐的模式，因为这种模式不受节点之间时钟差异和网络分区的影响
// 线性一致性读用的就是ReadOnlySafe
func sendMsgReadIndexResponse(r *raft, m pb.Message) {
	switch r.readOnly.option {
	// If more than the local vote is needed, go through a full broadcast.
	case ReadOnlySafe:
		// 清空readOnly中指定消息ID及之前的所有记录
		// 开启leader向follower的确认机制
		r.readOnly.addRequest(r.raftLog.committed, m)
		// recvAck通知只读结构raft状态机已收到对附加只读请求上下文的心跳信号的确认。
		// 也就是记录下只读的请求
		r.readOnly.recvAck(r.id, m.Entries[0].Data)
		// leader 节点向其他节点发起广播
		r.bcastHeartbeatWithCtx(m.Entries[0].Data)
	case ReadOnlyLeaseBased:
		if resp := r.responseToReadIndexReq(m, r.raftLog.committed); resp.To != None {
			r.send(resp)
		}
	}
}

func stepLeader(r *raft, m pb.Message) error {
	// All other message types require a progress for m.From (pr).
	pr := r.prs.Progress[m.From]
	if pr == nil {
		r.logger.Debugf("%x no progress available for %x", r.id, m.From)
		return nil
	}
	switch m.Type {
	case pb.MsgHeartbeatResp:
		...
		if r.readOnly.option != ReadOnlySafe || len(m.Context) == 0 {
			return nil
		}

		// 判断leader有没有收到大多数节点的确认
		// 也就是ReadIndex算法中，leader节点得到follower的确认，证明自己目前还是Leader
		if r.prs.Voters.VoteResult(r.readOnly.recvAck(m.From, m.Context)) != quorum.VoteWon {
			return nil
		}

		// 收到了响应节点超过半数，会清空readOnly中指定消息ID及之前的所有记录
		rss := r.readOnly.advance(m)
		// 返回follower的心跳回执
		for _, rs := range rss {
			if resp := r.responseToReadIndexReq(rs.req, rs.index); resp.To != None {
				r.send(resp)
			}
		}
	}
	return nil
}

// responseToReadIndexReq 为 `req` 构造一个响应。如果`req`来自对等方
// 本身，将返回一个空值。
func (r *raft) responseToReadIndexReq(req pb.Message, readIndex uint64) pb.Message {
	// 通过from来判断该消息是否是follower节点转发到leader中的
	...
	// 如果是其他follower节点转发到leader节点的MsgReadIndex消息
	// leader会回向follower节点返回响应的MsgReadIndexResp消息，follower会响应给client
	return pb.Message{
		Type:    pb.MsgReadIndexResp,
		To:      req.From,
		Index:   readIndex,
		Entries: req.Entries,
	}
}

func stepFollower(r *raft, m pb.Message) error {
	switch m.Type {
	case pb.MsgReadIndexResp:
		if len(m.Entries) != 1 {
			r.logger.Errorf("%x invalid format of MsgReadIndexResp from %x, entries count: %d", r.id, m.From, len(m.Entries))
			return nil
		}
		// 将MsgReadIndex消息中的已提交位置和消息id封装成ReadState实例，添加到readStates
		// raft 模块也有一个 for-loop 的 goroutine，来读取该数组，并对MsgReadIndex进行响应
		r.readStates = append(r.readStates, ReadState{Index: m.Index, RequestCtx: m.Entries[0].Data})
	}
	return nil
}

// etcd/server/etcdserver/raft.goetcd/raft/node.go
func (r *raftNode) start(rh *raftReadyHandler) {
	internalTimeout := time.Second

	go func() {
		defer r.onStop()
		islead := false

		for {
			select {
			case rd := <-r.Ready():
				...
				if len(rd.ReadStates) != 0 {
					select {
					// ReadStates中最后意向将会被写入到readStateC中
					// linearizableReadLoop会监听readStateC，获取MsgReadIndex的处理信息
					case r.readStateC <- rd.ReadStates[len(rd.ReadStates)-1]:
					case <-time.After(internalTimeout):
						r.lg.Warn("timed out sending read state", zap.Duration("timeout", internalTimeout))
					case <-r.stopped:
						return
					}
				}
				...
			}
		}
	}()
}

func stepLeader(r *raft, m pb.Message) error {
	// All other message types require a progress for m.From (pr).
	pr := r.prs.Progress[m.From]
	if pr == nil {
		r.logger.Debugf("%x no progress available for %x", r.id, m.From)
		return nil
	}
	switch m.Type {
	case pb.MsgReadIndex:
		// 表示当前只有一个节点，当前节点就是leader
		if r.prs.IsSingleton() {
			if resp := r.responseToReadIndexReq(m, r.raftLog.committed); resp.To != None {
				r.send(resp)
			}
			return nil
		}
		...
		return nil
	}
	return nil
}

// responseToReadIndexReq 为 `req` 构造一个响应。如果`req`来自对等方
// 本身，将返回一个空值。
func (r *raft) responseToReadIndexReq(req pb.Message, readIndex uint64) pb.Message {
	// 通过from来判断该消息是否是follower节点转发到leader中的

	// 如果是客户端直接发到leader节点的消息，将MsgReadIndex消息中的已提交位置和消息id封装成ReadState实例，添加到readStates
	// raft 模块也有一个 for-loop 的 goroutine，来读取该数组，并对MsgReadIndex进行响应
	if req.From == None || req.From == r.id {
		r.readStates = append(r.readStates, ReadState{
			Index:      readIndex,
			RequestCtx: req.Entries[0].Data,
		})
		return pb.Message{}
	}
	...
}