type kv struct {
    remote   pb.KVClient
    callOpts []grpc.CallOption
}

func NewKV(c *Client) KV {
    api := &kv{remote: RetryKVClient(c)}
    if c != nil {
        api.callOpts = c.callOpts
    }
    return api
}

func (kv *kv) Put(ctx context.Context, key, val string, opts ...OpOption) (*PutResponse, error) {
    r, err := kv.Do(ctx, OpPut(key, val, opts...))
    return r.put, toErr(ctx, err)
}

var resp *pb.PutResponse
r := &pb.PutRequest{Key: op.key, Value: op.val, Lease: int64(op.leaseID), PrevKv: op.prevKV, IgnoreValue: op.ignoreValue, IgnoreLease: op.ignoreLease}
resp, err = kv.remote.Put(ctx, r, kv.callOpts...)
if err == nil {
    return OpResponse{put: (*PutResponse)(resp)}, nil
}

type RaftKV interface {
    Range(ctx context.Context, r *pb.RangeRequest) (*pb.RangeResponse, error)
    Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error)
    DeleteRange(ctx context.Context, r *pb.DeleteRangeRequest) (*pb.DeleteRangeResponse, error)
    Txn(ctx context.Context, r *pb.TxnRequest) (*pb.TxnResponse, error)
    Compact(ctx context.Context, r *pb.CompactionRequest) (*pb.CompactionResponse, error)
}

func (s *EtcdServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {
    resp, err := s.raftRequest(ctx, pb.InternalRaftRequest{Put: r})
    if err != nil {
        return nil, err
    }
    return resp.(*pb.PutResponse), nil
}

func (s *EtcdServer) processInternalRaftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (*applyResult, error) {
    ai := s.getAppliedIndex()	//获取当前Applied Index
    ci := s.getCommittedIndex() //获取当前Committed Index

    // 如果当前Applied Index大于maxGapBetweenApplyAndCommitIndex（5000），证明集群处于不健康状态，因为请求都累积起来了，
    // 正常情况下即使有个时间间隔来响应请求，但是还是可以消费的，Applied Index不会太大
    if ci > ai+maxGapBetweenApplyAndCommitIndex {
        return nil, ErrTooManyRequests
    }

    //生成一个有序递增且唯一的id作为每个请求的id
    r.Header = &pb.RequestHeader{
        ID: s.reqIDGen.Next(),
    }

    // grpc，请求auth模块获得auth信息。
    // Username是与gRPC连接的验证令牌关联的用户名
    // Revision是auth.authStore的修订号。与mvcc无关
    // 每个请求走grpc，都会验证节点身份
    authInfo, err := s.AuthInfoFromCtx(ctx)
    if err != nil {
        return nil, err
    }
    if authInfo != nil {
        r.Header.Username = authInfo.Username
        r.Header.AuthRevision = authInfo.Revision
    }

    data, err := r.Marshal()
    if err != nil {
        return nil, err
    }

    //每次请求的数据大小不能超过MaxRequestBytes
    if len(data) > int(s.Cfg.MaxRequestBytes) {
        return nil, ErrRequestTooLarge
    }

    id := r.ID
    if id == 0 {
        id = r.Header.ID
    }

    //为每个请求的id注册等待和触发事件，等待事件触发
    ch := s.w.Register(id)

    //计算请求等待通过超时时间，ReqTimeout计算有点特别，看下面
    cctx, cancel := context.WithTimeout(ctx, s.Cfg.ReqTimeout())
    defer cancel()

    start := time.Now()

    //提议每个raftNode节点通过请求
    s.r.Propose(cctx, data)

    //提议数自增1，结束自减1。为了记录每次提议
    proposalsPending.Inc()
    defer proposalsPending.Dec()

    select {
    case x := <-ch:
        //等待提议经过大多数节点的通过
        return x.(*applyResult), nil
    case <-cctx.Done():
        //超时失败，提议失败原子+1
        proposalsFailed.Inc()
        s.w.Trigger(id, nil) // GC wait
        return nil, s.parseProposeCtxErr(cctx.Err(), start)
    case <-s.done:
        return nil, ErrStopped
    }
}

func (c *ServerConfig) ReqTimeout() time.Duration {
    // 5s for queue waiting, computation and disk IO delay
    // + 2 * election timeout for possible leader election
    return 5*time.Second + 2*time.Duration(c.ElectionTicks*int(c.TickMs))*time.Millisecond
}