grpc server源码学习
如何实现一个最简略的grpc server
// 创立listener
lis, err := net.Listen("tcp", port)
if err != nil {
log.Fatalf("failed to listen: %v", err)
}
// 创立server示例
s := grpc.NewServer()
// 注册服务
pb.RegisterGreeterServer(s, &server{})
reflection.Register(s)
// 启动服务端监听
if err := s.Serve(lis); err != nil {
log.Fatalf("failed to serve: %v", err)
}grpc.NewServer()
grpc.NewServer()会返回一个grpc.Server，它的构造如下：
// Server is a gRPC server to serve RPC requests.
type Server struct {
opts serverOptions
mu sync.Mutex // guards following
lis map[net.Listener]bool
conns map[transport.ServerTransport]bool
serve bool
drain bool
cv *sync.Cond // signaled when connections close for GracefulStop
m map[string]*service // service name -> service info
events trace.EventLog
quit *grpcsync.Event
done *grpcsync.Event
channelzRemoveOnce sync.Once
serveWG sync.WaitGroup // counts active Serve goroutines for GracefulStop
channelzID int64 // channelz unique identification number
czData *channelzData
}由这个构造，咱们能够略知一二，它应用了一个容器conns用来保留以后的所有连贯；也有和优雅退出的waitgroup，猜想应该是须要期待所有申请解决完后退出；cond猜想是用来告诉所有以后的连贯，服务将被进行了；其余字段的用处临时无奈很显著的猜想进去，咱们将在前面持续剖析。
pb.RegisterGreeterServer
func RegisterGreeterServer(s *grpc.Server, srv GreeterServer) {
s.RegisterService(&_Greeter_serviceDesc, srv)
}最终是调用了grpc.Server的RegisterService进行注册，第一个参数是pb生成代码生成的_Greeter_serviceDesc，它形容了rpc service的一些属性信息，内容如下：
var _Greeter_serviceDesc = grpc.ServiceDesc{
ServiceName: "helloworld.Greeter",
HandlerType: (*GreeterServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "SayHello",
Handler: _Greeter_SayHello_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "demo.proto",
}这外面形容了
- 服务名
- HandlerType（这个是做什么的？存疑）
- 办法列表，办法列表包含办法名和一个Handler（Handler做什么的？存疑）
- Streams，这个应该是只有stream类型的rpc service才会领有具体的值
- Metadata，这个代表生成这个go文件的原始pb文件的文件名
办法列表里，每一项是一个MethodDesc构造体，它的定义如下：
type methodHandler func(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor UnaryServerInterceptor) (interface{}, error)
// MethodDesc represents an RPC service's method specification.
type MethodDesc struct {
MethodName string
Handler methodHandler
}能够发现Handler其实是一个函数；
那么实在的_Greeter_SayHello_Handler蕴含了哪些信息呢？蕴含的信息如下：
func _Greeter_SayHello_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error)
in := new(HelloRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(GreeterServer).SayHello(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/helloworld.Greeter/SayHello",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(GreeterServer).SayHello(ctx, req.(*HelloRequest))
}
return interceptor(ctx, in, info, handler)
}通过这段代码能够看出，它其实是真正调用解决rpc定义的接口办法的入口。
- 有一个参数dec，这个首先会将其解码到request构造体上
- 如果没有设置拦截器，那么间接调用srv实现的SayHello办法进行解决并返回。
- 如果设置了拦截器，会通过传入的拦截器做一些非凡的解决。
能够猜想，在grpc server实在调用时，会将具体实现了rpc service的构造体作为第一个参数传入，将通过解码的request数据作为第三个参数传入，由实现了rpc service接口的对象来进行解决。
注册服务
func (s *Server) register(sd *ServiceDesc, ss interface{}) {
s.mu.Lock()
defer s.mu.Unlock()
s.printf("RegisterService(%q)", sd.ServiceName)
if s.serve {
grpclog.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)
}
if _, ok := s.m[sd.ServiceName]; ok {
grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
}
srv := &service{
server: ss,
md: make(map[string]*MethodDesc),
sd: make(map[string]*StreamDesc),
mdata: sd.Metadata,
}
for i := range sd.Methods {
d := &sd.Methods[i]
srv.md[d.MethodName] = d
}
for i := range sd.Streams {
d := &sd.Streams[i]
srv.sd[d.StreamName] = d
}
s.m[sd.ServiceName] = srv
}能够依据之前形容的server构造体能够看出，server有一个属性s，这个汇合能够存储多个pb定义的service。key为服务名，value为具体的服务构造体。意思是一个grpc server启动时，能够有多个pb定义的rpc service被注册到其中。并且同一个rpc service不能被注册两次。
注册服务时，将之前提到的pb生成的_Greeter_serviceDesc 相干信息注册到了service汇合中。把service和method都注册到了server中。
grpcServer.Serve
当初服务能够正式启动起来了，首先看一下官网对Serve办法的正文：
// Serve accepts incoming connections on the listener lis, creating a new
// ServerTransport and service goroutine for each. The service goroutines
// read gRPC requests and then call the registered handlers to reply to them.
// Serve returns when lis.Accept fails with fatal errors. lis will be closed when
// this method returns.
// Serve will return a non-nil error unless Stop or GracefulStop is called.大抵翻译一下：
Serve对每一个listener监听到的连贯创立一个新的ServerTransport和协程，这个协程读取grpc申请而后调用被注册的handler来响应这些申请。Serve办法在lis.Accept失败时会返回error，当这个办法返回时，lis会被敞开。当应用失常退出或者优雅退出时，Serve办法不会返回error。
正文说得十分明确，也能看到，的确是调用了被注册的handler来响应申请。
Serve办法的大抵工作逻辑：
for {
rawConn, err := lis.Accept()
if err != nil {
//错误处理
//...
}
s.serveWG.Add(1)
go func() {
s.handleRawConn(rawConn)
s.serveWG.Done()
}()
}整个Serve办法在一个大的for循环中一直获取listener监听到的申请，而后对每一个监听到的申请开了一个协程去解决。
也就是说外围其实是handleRawConn办法
// handleRawConn forks a goroutine to handle a just-accepted connection that
// has not had any I/O performed on it yet.
func (s *Server) handleRawConn(rawConn net.Conn) {
if s.quit.HasFired() {
rawConn.Close()
return
}
//设置连贯超时，若没有的话会有默认的超时工夫，默认两分钟
rawConn.SetDeadline(time.Now().Add(s.opts.connectionTimeout))
//查看证书认证相干信息
conn, authInfo, err := s.useTransportAuthenticator(rawConn)
if err != nil {
// ErrConnDispatched means that the connection was dispatched away from
// gRPC; those connections should be left open.
if err != credentials.ErrConnDispatched {
s.mu.Lock()
s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
s.mu.Unlock()
grpclog.Warningf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err)
rawConn.Close()
}
rawConn.SetDeadline(time.Time{})
return
}
//进行http2的传输
// Finish handshaking (HTTP2)
st := s.newHTTP2Transport(conn, authInfo)//获取到传输st
if st == nil {
return
}
rawConn.SetDeadline(time.Time{})
//把http2传输连贯缓存到server的conns构造中
//为何要缓存呢？
if !s.addConn(st) {//如果server的conns构造为空时，会返回false，并敞开掉传输st
return
}
//间接go一个协程，异步解决申请
go func() {
s.serveStreams(st)
//最初从连贯缓存conns构造中删除掉传输st
//并通过cond，发动一个播送
s.removeConn(st)
}()//这里能够看到，起了一个协程去解决这个http2的stream，这是因为http2是长连贯，能够复用这个流一直接收数据，所以须要起一个协程来做这件事。
这样看来，实际上咱们应该重点关注下serveStreams办法，它做了什么呢？
func (s *Server) serveStreams(st transport.ServerTransport) {
defer st.Close()
var wg sync.WaitGroup
st.HandleStreams(func(stream *transport.Stream) {
wg.Add(1)
go func() {
defer wg.Done()
s.handleStream(st, stream, s.traceInfo(st, stream))
}()
}, func(ctx context.Context, method string) context.Context {
if !EnableTracing {
return ctx
}
tr := trace.New("grpc.Recv."+methodFamily(method), method)
return trace.NewContext(ctx, tr)
})
wg.Wait()
}能够看到，这里实际上的解决办法是：
func(stream *transport.Stream) {
wg.Add(1)
go func() {
defer wg.Done()
s.handleStream(st, stream, s.traceInfo(st, stream))
}()
}即HandleStream，为何这里要应用waitGroup，并期待这个实现，而不是同步的形式呢？。这是因为，咱们要理解一个前提，grpc是能够反对长连贯的，因而severStreams办法解决的其实并不仅仅是单个申请，而是一个源源不断的http2流。所以咱们应该查看st的HandleStream干了什么事件。
st是transport.ServerTransport类型的interface，而实际上这里的对象是http2Server。
// HandleStreams receives incoming streams using the given handler. This is
// typically run in a separate goroutine.
// traceCtx attaches trace to ctx and returns the new context.
func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.Context, string) context.Context) {
defer close(t.readerDone)
for {
t.controlBuf.throttle()
frame, err := t.framer.fr.ReadFrame()
atomic.StoreInt64(&t.lastRead, time.Now().UnixNano())
if err != nil {
//错误处理
//...
}
switch frame := frame.(type) {
case *http2.MetaHeadersFrame:
if t.operateHeaders(frame, handle, traceCtx) {
t.Close()
break
}
case *http2.DataFrame:
t.handleData(frame)
case *http2.RSTStreamFrame:
t.handleRSTStream(frame)
case *http2.SettingsFrame:
t.handleSettings(frame)
case *http2.PingFrame:
t.handlePing(frame)
case *http2.WindowUpdateFrame:
t.handleWindowUpdate(frame)
case *http2.GoAwayFrame:
// TODO: Handle GoAway from the client appropriately.
default:
errorf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame)
}
}
}因而，整个办法是在一个for循环中，一直解决二进制流，所以对每次获取到的数据都应用一个协程去解决，这就是为何handle参数要应用waitgroup。
handleStream做了什么事呢？这里只截取要害代码：
sm := stream.Method()
service := sm[:pos]
method := sm[pos+1:]
srv, knownService := s.m[service]
if knownService {
if md, ok := srv.md[method]; ok {
s.processUnaryRPC(t, stream, srv, md, trInfo)
return
}
if sd, ok := srv.sd[method]; ok {
s.processStreamingRPC(t, stream, srv, sd, trInfo)
return
}
}总结起来是：
- 从stream里取出method，service信息
- 如果service与之前注册胜利的rpc service名字匹配，则查看method是否与注册胜利的rpc service的method相匹配，若匹配的话，间接调用
processUnaryRPC或processStreamingRPC进行解决。
接着看processUnaryRPC，发现了咱们最开始提到的pb注册的Handler: _Greeter_SayHello_Handler，它是这样被调用的：
df := func(v interface{}) error {
if err := s.getCodec(stream.ContentSubtype()).Unmarshal(d, v); err != nil {
return status.Errorf(codes.Internal, "grpc: error unmarshalling request: %v", err)
}
if sh != nil {
sh.HandleRPC(stream.Context(), &stats.InPayload{
RecvTime: time.Now(),
Payload: v,
WireLength: payInfo.wireLength,
Data: d,
Length: len(d),
})
}
if binlog != nil {
binlog.Log(&binarylog.ClientMessage{
Message: d,
})
}
if trInfo != nil {
trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)
}
return nil
}
ctx := NewContextWithServerTransportStream(stream.Context(), stream)
reply, appErr := md.Handler(srv.server, ctx, df, s.opts.unaryInt)最初解决实现后，会将resp写回：
if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {
...
}至此，整个grpc就解决完了一条申请，并返回了响应。
优雅退出是如何实现的？
之前咱们有一个疑难，为什么要缓存所有的conns呢？这就是因为在优雅退出时，咱们须要告诉/期待所有以后的连贯都开释再退出。
看一下优雅退出是如何实现的：
- 第一阶段
s.mu.Lock()
if s.conns == nil {
s.mu.Unlock()
return
}
for lis := range s.lis {
lis.Close()
}
s.lis = nil
if !s.drain {
for st := range s.conns {
st.Drain()
}
s.drain = true
}
// Wait for serving threads to be ready to exit. Only then can we be sure no
// new conns will be created.
s.mu.Unlock()第一阶段次要是首先将所有tcp listener敞开掉，并且若server状态不处于 进行接管申请的状态，那么就告诉以后所有的连贯进行再接管申请（Drain），最初将状态drain置为true。
- 第二阶段
s.serveWG.Wait()
s.mu.Lock()
for len(s.conns) != 0 {
s.cv.Wait()
}
s.conns = nil
if s.events != nil {
s.events.Finish()
s.events = nil
}
s.mu.Unlock()首先期待所有的申请都解决完，而后加锁，期待缓存的conns连贯都被敞开掉，若都被敞开掉，会有协程发动Broadcase进行告诉，告诉结束后Wait()不会再阻塞，能够接着往下走，最初发送finish event，解锁即可。
什么时候会发动Broadcast呢？有两个中央：
- 解决完一个流，会将流从conns中删除
go func() {
s.serveStreams(st)
s.removeConn(st)
}()而删除时，则会将st从conns中删除掉，并且发一个Broadcast
func (s *Server) removeConn(st transport.ServerTransport) {
s.mu.Lock()
defer s.mu.Unlock()
if s.conns != nil {
delete(s.conns, st)
s.cv.Broadcast()
}
}依据优雅退出代码，尽管发动了播送，s.cv.Wait不再阻塞，然而若还有别的流没有开释，那么s.conns依然不为0，因而又会进入到s.cv.Wait的阻塞中。
- server失常退出（
Stop）
s.mu.Lock()
listeners := s.lis
s.lis = nil
st := s.conns
s.conns = nil
// interrupt GracefulStop if Stop and GracefulStop are called concurrently.
s.cv.Broadcast()
s.mu.Unlock()
for lis := range listeners {
lis.Close()
}
for c := range st {
c.Close()
}
s.mu.Lock()
if s.events != nil {
s.events.Finish()
s.events = nil
}
s.mu.Unlock()能够看到，在失常退出server时，间接将conns置为了nil，而后发送了Broadcasr，因而优雅退出的Wait也不会再期待了。
发表回复