ETCD数据库源码分析——服务端PUT流程

在ETCD数据库源码分析——etcd gRPC 服务 API注册服务小节&＃xff0c;有如下grpc服务注册代码&＃xff0c;包含了服务端PUT流程所属的KVServer。

RegisterKVServer函数定义在/api/etcdserverpb/rpc.pb.go文件中&＃xff0c;这里pb.RegisterKVServer(grpcServer, NewQutaKVServer(s))中的s指的就是ETCD数据库中的最重要的etcdserver结构体。

func RegisterKVServer(s *grpc.Server, srv KVServer) {s.RegisterService(&_KV_serviceDesc, srv)
}


在/api/etcdserverpb/rpc.pb.go里面&＃xff0c;可以看到上面定义的ServiceName和MethodName&＃xff0c;可以找到Put方法对应的函数_KV_Put_Handler。

从_KV_Put_Handler函数可以看到其最重要的语句就是srv.(KVServer).Put(ctx, in)&＃xff0c;其实就是调用的quotaKVServer.(KVServer).Put(ctx, in)。

func _KV_Put_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {in :&＃61; new(PutRequest)if err :&＃61; dec(in); err !&＃61; nil { return nil, err }if interceptor &＃61;&＃61; nil { return srv.(KVServer).Put(ctx, in) }info :&＃61; &grpc.UnaryServerInfo{Server: srv,FullMethod: "/etcdserverpb.KV/Put",}handler :&＃61; func(ctx context.Context, req interface{}) (interface{}, error) { return srv.(KVServer).Put(ctx, req.(*PutRequest)) }return interceptor(ctx, in, info, handler)
}


因此这里从NewQuotaKVServer看出&＃xff0c;quotaKVServer和pb.KVServer接口是组合关系&＃xff0c;NewQuotaKVServer函数中调用NewKVServer函数初始化继承了pb.KVServer接口的kvServer结构体。

// server/etcdserver/api/v3rpc/quota.go
type quotaKVServer struct {pb.KVServerqa quotaAlarmer
}
func NewQuotaKVServer(s *etcdserver.EtcdServer) pb.KVServer {return &quotaKVServer{NewKVServer(s),quotaAlarmer{newBackendQuota(s, "kv"), s, s.MemberId()},}
}


etcdserver.RaftKV接口继承了KVServer接口。而pb.RegisterKVServer(grpcServer, NewQutaKVServer(s))中的s指的就是ETCD数据库中的最重要的etcdserver结构体&＃xff0c;所以这里的etcdserver.RaftKV指向的就是etcdserver结构体。

type kvServer struct {hdr header // 用于填充响应消息的头信息kv etcdserver.RaftKV// maxTxnOps is the max operations per txn.// e.g suppose maxTxnOps &＃61; 128.// Txn.Success can have at most 128 operations,// and Txn.Failure can have at most 128 operations.maxTxnOps uint
}
func NewKVServer(s *etcdserver.EtcdServer) pb.KVServer {return &kvServer{hdr: newHeader(s), kv: s, maxTxnOps: s.Cfg.MaxTxnOps}
}
func (s *kvServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {if err :&＃61; checkPutRequest(r); err !&＃61; nil { return nil, err }resp, err :&＃61; s.kv.Put(ctx, r)if err !&＃61; nil { return nil, togRPCError(err) }s.hdr.fill(resp.Header)return resp, nil
}


kvServer结构体Put函数处理流程如下&＃xff1a;首先会对请求消息进行各方面的检查&＃xff0c;检查完之后会将所有的请求交给其内封装的RaftKV接口进行处理&＃xff0c;待处理完成得到响应消息之后&＃xff0c;会通过header.fill()方法填充响应的头信息&＃xff0c;最后将完整的响应消息返回给客户端。因此&＃xff0c;往下追踪 srv.(KVServer).Put(ctx, in)其实就是调用(s *EtcdServer) Put()函数。

// server/etcdserver/v3_server.go
func (s *EtcdServer) Put(ctx context.Context, r *pb.PutRequest) (*pb.PutResponse, error) {ctx &＃61; context.WithValue(ctx, traceutil.StartTimeKey, time.Now())resp, err :&＃61; s.raftRequest(ctx, pb.InternalRaftRequest{Put: r})if err !&＃61; nil { return nil, err }return resp.(*pb.PutResponse), nil
}
func (s *EtcdServer) raftRequest(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error) {return s.raftRequestOnce(ctx, r)
}
func (s *EtcdServer) raftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (proto.Message, error) {result, err :&＃61; s.processInternalRaftRequestOnce(ctx, r)if err !&＃61; nil { return nil, err }if result.Err !&＃61; nil { return nil, result.Err }if startTime, ok :&＃61; ctx.Value(traceutil.StartTimeKey).(time.Time); ok && result.Trace !&＃61; nil {applyStart :&＃61; result.Trace.GetStartTime()// The trace object is created in toApply. Here reset the start time to trace// the raft request time by the difference between the request start time// and toApply start timeresult.Trace.SetStartTime(startTime)result.Trace.InsertStep(0, applyStart, "process raft request")result.Trace.LogIfLong(traceThreshold)}return result.Resp, nil
}


从上面流程看出最终调用的是 (s *EtcdServer) processInternalRaftRequestOnce(…)函数&＃xff0c;在该函数里面有一句关键调用 s.r.Propose(cctx, data)。s是EtcdServer, r是其里面的成员变量raftNode, 这就是进入raft协议相关的节奏了。

func (s *EtcdServer) processInternalRaftRequestOnce(ctx context.Context, r pb.InternalRaftRequest) (*apply2.Result, error) {ai :&＃61; s.getAppliedIndex()ci :&＃61; s.getCommittedIndex()// 如果apply和commit的raft log差距太大需要拒绝客户的请求if ci > ai&＃43;maxGapBetweenApplyAndCommitIndex { return nil, errors.ErrTooManyRequests }r.Header &＃61; &pb.RequestHeader{ ID: s.reqIDGen.Next(), } // 利用reqIDGen产生递增标号// check authinfo if it is not InternalAuthenticateRequest 对于非InternalAuthenticateRequest的请求&＃xff0c;需要检查authinfo的信息if r.Authenticate &＃61;&＃61; nil {authInfo, err :&＃61; s.AuthInfoFromCtx(ctx)if err !&＃61; nil { return nil, err }if authInfo !&＃61; nil {r.Header.Username &＃61; authInfo.Usernamer.Header.AuthRevision &＃61; authInfo.Revision}}data, err :&＃61; r.Marshal()if err !&＃61; nil { return nil, err }if len(data) > int(s.Cfg.MaxRequestBytes) { return nil, errors.ErrRequestTooLarge } // 发送的数据太大id :&＃61; r.IDif id &＃61;&＃61; 0 { id &＃61; r.Header.ID }ch :&＃61; s.w.Register(id) // Register waits returns a chan that waits on the given ID// The chan will be triggered when Trigger is called with the same ID.// 这里将id注册到etcdsever结构体中的Wait中&＃xff0c;等待Wait.Trigger(id, nil)// 正常情况下当raft log被应用后会调用Wait.Trigger(id, nil)触发管道cctx, cancel :&＃61; context.WithTimeout(ctx, s.Cfg.ReqTimeout())defer cancel()start :&＃61; time.Now()err &＃61; s.r.Propose(cctx, data) // 调用raft模块的s.r.Propose(cctx, data)if err !&＃61; nil {proposalsFailed.Inc()s.w.Trigger(id, nil) // GC waitreturn nil, err}proposalsPending.Inc()defer proposalsPending.Dec()select {case x :&＃61; <-ch: // Wait.Trigger(id, nil)管道被触发return x.(*apply2.Result), nilcase <-cctx.Done():proposalsFailed.Inc()s.w.Trigger(id, nil) // GC waitreturn nil, s.parseProposeCtxErr(cctx.Err(), start)case <-s.done:return nil, errors.ErrStopped}
}


欢迎关注微信公众号肥叔菌&＃xff0c;最新内容优先在微信公众号发布。