3 * Copyright 2014 gRPC authors.
5 * Licensed under the Apache License, Version 2.0 (the "License");
6 * you may not use this file except in compliance with the License.
7 * You may obtain a copy of the License at
9 * http://www.apache.org/licenses/LICENSE-2.0
11 * Unless required by applicable law or agreed to in writing, software
12 * distributed under the License is distributed on an "AS IS" BASIS,
13 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14 * See the License for the specific language governing permissions and
15 * limitations under the License.
36 "golang.org/x/net/trace"
38 "google.golang.org/grpc/codes"
39 "google.golang.org/grpc/credentials"
40 "google.golang.org/grpc/encoding"
41 "google.golang.org/grpc/encoding/proto"
42 "google.golang.org/grpc/grpclog"
43 "google.golang.org/grpc/internal/binarylog"
44 "google.golang.org/grpc/internal/channelz"
45 "google.golang.org/grpc/internal/transport"
46 "google.golang.org/grpc/keepalive"
47 "google.golang.org/grpc/metadata"
48 "google.golang.org/grpc/peer"
49 "google.golang.org/grpc/stats"
50 "google.golang.org/grpc/status"
51 "google.golang.org/grpc/tap"
55 defaultServerMaxReceiveMessageSize = 1024 * 1024 * 4
56 defaultServerMaxSendMessageSize = math.MaxInt32
59 type methodHandler func(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor UnaryServerInterceptor) (interface{}, error)
61 // MethodDesc represents an RPC service's method specification.
62 type MethodDesc struct {
67 // ServiceDesc represents an RPC service's specification.
68 type ServiceDesc struct {
70 // The pointer to the service interface. Used to check whether the user
71 // provided implementation satisfies the interface requirements.
72 HandlerType interface{}
78 // service consists of the information of the server serving this service and
79 // the methods in this service.
81 server interface{} // the server for service methods
82 md map[string]*MethodDesc
83 sd map[string]*StreamDesc
87 // Server is a gRPC server to serve RPC requests.
91 mu sync.Mutex // guards following
92 lis map[net.Listener]bool
93 conns map[io.Closer]bool
96 cv *sync.Cond // signaled when connections close for GracefulStop
97 m map[string]*service // service name -> service info
104 channelzRemoveOnce sync.Once
105 serveWG sync.WaitGroup // counts active Serve goroutines for GracefulStop
107 channelzID int64 // channelz unique identification number
111 type options struct {
112 creds credentials.TransportCredentials
116 unaryInt UnaryServerInterceptor
117 streamInt StreamServerInterceptor
118 inTapHandle tap.ServerInHandle
119 statsHandler stats.Handler
120 maxConcurrentStreams uint32
121 maxReceiveMessageSize int
122 maxSendMessageSize int
123 unknownStreamDesc *StreamDesc
124 keepaliveParams keepalive.ServerParameters
125 keepalivePolicy keepalive.EnforcementPolicy
126 initialWindowSize int32
127 initialConnWindowSize int32
130 connectionTimeout time.Duration
131 maxHeaderListSize *uint32
134 var defaultServerOptions = options{
135 maxReceiveMessageSize: defaultServerMaxReceiveMessageSize,
136 maxSendMessageSize: defaultServerMaxSendMessageSize,
137 connectionTimeout: 120 * time.Second,
138 writeBufferSize: defaultWriteBufSize,
139 readBufferSize: defaultReadBufSize,
142 // A ServerOption sets options such as credentials, codec and keepalive parameters, etc.
143 type ServerOption func(*options)
145 // WriteBufferSize determines how much data can be batched before doing a write on the wire.
146 // The corresponding memory allocation for this buffer will be twice the size to keep syscalls low.
147 // The default value for this buffer is 32KB.
148 // Zero will disable the write buffer such that each write will be on underlying connection.
149 // Note: A Send call may not directly translate to a write.
150 func WriteBufferSize(s int) ServerOption {
151 return func(o *options) {
152 o.writeBufferSize = s
156 // ReadBufferSize lets you set the size of read buffer, this determines how much data can be read at most
157 // for one read syscall.
158 // The default value for this buffer is 32KB.
159 // Zero will disable read buffer for a connection so data framer can access the underlying
161 func ReadBufferSize(s int) ServerOption {
162 return func(o *options) {
167 // InitialWindowSize returns a ServerOption that sets window size for stream.
168 // The lower bound for window size is 64K and any value smaller than that will be ignored.
169 func InitialWindowSize(s int32) ServerOption {
170 return func(o *options) {
171 o.initialWindowSize = s
175 // InitialConnWindowSize returns a ServerOption that sets window size for a connection.
176 // The lower bound for window size is 64K and any value smaller than that will be ignored.
177 func InitialConnWindowSize(s int32) ServerOption {
178 return func(o *options) {
179 o.initialConnWindowSize = s
183 // KeepaliveParams returns a ServerOption that sets keepalive and max-age parameters for the server.
184 func KeepaliveParams(kp keepalive.ServerParameters) ServerOption {
185 return func(o *options) {
186 o.keepaliveParams = kp
190 // KeepaliveEnforcementPolicy returns a ServerOption that sets keepalive enforcement policy for the server.
191 func KeepaliveEnforcementPolicy(kep keepalive.EnforcementPolicy) ServerOption {
192 return func(o *options) {
193 o.keepalivePolicy = kep
197 // CustomCodec returns a ServerOption that sets a codec for message marshaling and unmarshaling.
199 // This will override any lookups by content-subtype for Codecs registered with RegisterCodec.
200 func CustomCodec(codec Codec) ServerOption {
201 return func(o *options) {
206 // RPCCompressor returns a ServerOption that sets a compressor for outbound
207 // messages. For backward compatibility, all outbound messages will be sent
208 // using this compressor, regardless of incoming message compression. By
209 // default, server messages will be sent using the same compressor with which
210 // request messages were sent.
212 // Deprecated: use encoding.RegisterCompressor instead.
213 func RPCCompressor(cp Compressor) ServerOption {
214 return func(o *options) {
219 // RPCDecompressor returns a ServerOption that sets a decompressor for inbound
220 // messages. It has higher priority than decompressors registered via
221 // encoding.RegisterCompressor.
223 // Deprecated: use encoding.RegisterCompressor instead.
224 func RPCDecompressor(dc Decompressor) ServerOption {
225 return func(o *options) {
230 // MaxMsgSize returns a ServerOption to set the max message size in bytes the server can receive.
231 // If this is not set, gRPC uses the default limit.
233 // Deprecated: use MaxRecvMsgSize instead.
234 func MaxMsgSize(m int) ServerOption {
235 return MaxRecvMsgSize(m)
238 // MaxRecvMsgSize returns a ServerOption to set the max message size in bytes the server can receive.
239 // If this is not set, gRPC uses the default 4MB.
240 func MaxRecvMsgSize(m int) ServerOption {
241 return func(o *options) {
242 o.maxReceiveMessageSize = m
246 // MaxSendMsgSize returns a ServerOption to set the max message size in bytes the server can send.
247 // If this is not set, gRPC uses the default 4MB.
248 func MaxSendMsgSize(m int) ServerOption {
249 return func(o *options) {
250 o.maxSendMessageSize = m
254 // MaxConcurrentStreams returns a ServerOption that will apply a limit on the number
255 // of concurrent streams to each ServerTransport.
256 func MaxConcurrentStreams(n uint32) ServerOption {
257 return func(o *options) {
258 o.maxConcurrentStreams = n
262 // Creds returns a ServerOption that sets credentials for server connections.
263 func Creds(c credentials.TransportCredentials) ServerOption {
264 return func(o *options) {
269 // UnaryInterceptor returns a ServerOption that sets the UnaryServerInterceptor for the
270 // server. Only one unary interceptor can be installed. The construction of multiple
271 // interceptors (e.g., chaining) can be implemented at the caller.
272 func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {
273 return func(o *options) {
274 if o.unaryInt != nil {
275 panic("The unary server interceptor was already set and may not be reset.")
281 // StreamInterceptor returns a ServerOption that sets the StreamServerInterceptor for the
282 // server. Only one stream interceptor can be installed.
283 func StreamInterceptor(i StreamServerInterceptor) ServerOption {
284 return func(o *options) {
285 if o.streamInt != nil {
286 panic("The stream server interceptor was already set and may not be reset.")
292 // InTapHandle returns a ServerOption that sets the tap handle for all the server
293 // transport to be created. Only one can be installed.
294 func InTapHandle(h tap.ServerInHandle) ServerOption {
295 return func(o *options) {
296 if o.inTapHandle != nil {
297 panic("The tap handle was already set and may not be reset.")
303 // StatsHandler returns a ServerOption that sets the stats handler for the server.
304 func StatsHandler(h stats.Handler) ServerOption {
305 return func(o *options) {
310 // UnknownServiceHandler returns a ServerOption that allows for adding a custom
311 // unknown service handler. The provided method is a bidi-streaming RPC service
312 // handler that will be invoked instead of returning the "unimplemented" gRPC
313 // error whenever a request is received for an unregistered service or method.
314 // The handling function has full access to the Context of the request and the
315 // stream, and the invocation bypasses interceptors.
316 func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {
317 return func(o *options) {
318 o.unknownStreamDesc = &StreamDesc{
319 StreamName: "unknown_service_handler",
320 Handler: streamHandler,
321 // We need to assume that the users of the streamHandler will want to use both.
328 // ConnectionTimeout returns a ServerOption that sets the timeout for
329 // connection establishment (up to and including HTTP/2 handshaking) for all
330 // new connections. If this is not set, the default is 120 seconds. A zero or
331 // negative value will result in an immediate timeout.
333 // This API is EXPERIMENTAL.
334 func ConnectionTimeout(d time.Duration) ServerOption {
335 return func(o *options) {
336 o.connectionTimeout = d
340 // MaxHeaderListSize returns a ServerOption that sets the max (uncompressed) size
341 // of header list that the server is prepared to accept.
342 func MaxHeaderListSize(s uint32) ServerOption {
343 return func(o *options) {
344 o.maxHeaderListSize = &s
348 // NewServer creates a gRPC server which has no service registered and has not
349 // started to accept requests yet.
350 func NewServer(opt ...ServerOption) *Server {
351 opts := defaultServerOptions
352 for _, o := range opt {
356 lis: make(map[net.Listener]bool),
358 conns: make(map[io.Closer]bool),
359 m: make(map[string]*service),
360 quit: make(chan struct{}),
361 done: make(chan struct{}),
362 czData: new(channelzData),
364 s.cv = sync.NewCond(&s.mu)
366 _, file, line, _ := runtime.Caller(1)
367 s.events = trace.NewEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
371 s.channelzID = channelz.RegisterServer(&channelzServer{s}, "")
376 // printf records an event in s's event log, unless s has been stopped.
377 // REQUIRES s.mu is held.
378 func (s *Server) printf(format string, a ...interface{}) {
380 s.events.Printf(format, a...)
384 // errorf records an error in s's event log, unless s has been stopped.
385 // REQUIRES s.mu is held.
386 func (s *Server) errorf(format string, a ...interface{}) {
388 s.events.Errorf(format, a...)
392 // RegisterService registers a service and its implementation to the gRPC
393 // server. It is called from the IDL generated code. This must be called before
395 func (s *Server) RegisterService(sd *ServiceDesc, ss interface{}) {
396 ht := reflect.TypeOf(sd.HandlerType).Elem()
397 st := reflect.TypeOf(ss)
398 if !st.Implements(ht) {
399 grpclog.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)
404 func (s *Server) register(sd *ServiceDesc, ss interface{}) {
407 s.printf("RegisterService(%q)", sd.ServiceName)
409 grpclog.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)
411 if _, ok := s.m[sd.ServiceName]; ok {
412 grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
416 md: make(map[string]*MethodDesc),
417 sd: make(map[string]*StreamDesc),
420 for i := range sd.Methods {
422 srv.md[d.MethodName] = d
424 for i := range sd.Streams {
426 srv.sd[d.StreamName] = d
428 s.m[sd.ServiceName] = srv
431 // MethodInfo contains the information of an RPC including its method name and type.
432 type MethodInfo struct {
433 // Name is the method name only, without the service name or package name.
435 // IsClientStream indicates whether the RPC is a client streaming RPC.
437 // IsServerStream indicates whether the RPC is a server streaming RPC.
441 // ServiceInfo contains unary RPC method info, streaming RPC method info and metadata for a service.
442 type ServiceInfo struct {
444 // Metadata is the metadata specified in ServiceDesc when registering service.
448 // GetServiceInfo returns a map from service names to ServiceInfo.
449 // Service names include the package names, in the form of <package>.<service>.
450 func (s *Server) GetServiceInfo() map[string]ServiceInfo {
451 ret := make(map[string]ServiceInfo)
452 for n, srv := range s.m {
453 methods := make([]MethodInfo, 0, len(srv.md)+len(srv.sd))
454 for m := range srv.md {
455 methods = append(methods, MethodInfo{
457 IsClientStream: false,
458 IsServerStream: false,
461 for m, d := range srv.sd {
462 methods = append(methods, MethodInfo{
464 IsClientStream: d.ClientStreams,
465 IsServerStream: d.ServerStreams,
469 ret[n] = ServiceInfo{
477 // ErrServerStopped indicates that the operation is now illegal because of
478 // the server being stopped.
479 var ErrServerStopped = errors.New("grpc: the server has been stopped")
481 func (s *Server) useTransportAuthenticator(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
482 if s.opts.creds == nil {
483 return rawConn, nil, nil
485 return s.opts.creds.ServerHandshake(rawConn)
488 type listenSocket struct {
493 func (l *listenSocket) ChannelzMetric() *channelz.SocketInternalMetric {
494 return &channelz.SocketInternalMetric{
495 SocketOptions: channelz.GetSocketOption(l.Listener),
496 LocalAddr: l.Listener.Addr(),
500 func (l *listenSocket) Close() error {
501 err := l.Listener.Close()
503 channelz.RemoveEntry(l.channelzID)
508 // Serve accepts incoming connections on the listener lis, creating a new
509 // ServerTransport and service goroutine for each. The service goroutines
510 // read gRPC requests and then call the registered handlers to reply to them.
511 // Serve returns when lis.Accept fails with fatal errors. lis will be closed when
512 // this method returns.
513 // Serve will return a non-nil error unless Stop or GracefulStop is called.
514 func (s *Server) Serve(lis net.Listener) error {
519 // Serve called after Stop or GracefulStop.
522 return ErrServerStopped
529 // Stop or GracefulStop called; block until done and return nil.
536 ls := &listenSocket{Listener: lis}
540 ls.channelzID = channelz.RegisterListenSocket(ls, s.channelzID, lis.Addr().String())
546 if s.lis != nil && s.lis[ls] {
553 var tempDelay time.Duration // how long to sleep on accept failure
556 rawConn, err := lis.Accept()
558 if ne, ok := err.(interface {
560 }); ok && ne.Temporary() {
562 tempDelay = 5 * time.Millisecond
566 if max := 1 * time.Second; tempDelay > max {
570 s.printf("Accept error: %v; retrying in %v", err, tempDelay)
572 timer := time.NewTimer(tempDelay)
582 s.printf("done serving; Accept = %v", err)
593 // Start a new goroutine to deal with rawConn so we don't stall this Accept
596 // Make sure we account for the goroutine so GracefulStop doesn't nil out
597 // s.conns before this conn can be added.
600 s.handleRawConn(rawConn)
606 // handleRawConn forks a goroutine to handle a just-accepted connection that
607 // has not had any I/O performed on it yet.
608 func (s *Server) handleRawConn(rawConn net.Conn) {
609 rawConn.SetDeadline(time.Now().Add(s.opts.connectionTimeout))
610 conn, authInfo, err := s.useTransportAuthenticator(rawConn)
613 s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
615 grpclog.Warningf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err)
616 // If serverHandshake returns ErrConnDispatched, keep rawConn open.
617 if err != credentials.ErrConnDispatched {
620 rawConn.SetDeadline(time.Time{})
632 // Finish handshaking (HTTP2)
633 st := s.newHTTP2Transport(conn, authInfo)
638 rawConn.SetDeadline(time.Time{})
648 // newHTTP2Transport sets up a http/2 transport (using the
649 // gRPC http2 server transport in transport/http2_server.go).
650 func (s *Server) newHTTP2Transport(c net.Conn, authInfo credentials.AuthInfo) transport.ServerTransport {
651 config := &transport.ServerConfig{
652 MaxStreams: s.opts.maxConcurrentStreams,
654 InTapHandle: s.opts.inTapHandle,
655 StatsHandler: s.opts.statsHandler,
656 KeepaliveParams: s.opts.keepaliveParams,
657 KeepalivePolicy: s.opts.keepalivePolicy,
658 InitialWindowSize: s.opts.initialWindowSize,
659 InitialConnWindowSize: s.opts.initialConnWindowSize,
660 WriteBufferSize: s.opts.writeBufferSize,
661 ReadBufferSize: s.opts.readBufferSize,
662 ChannelzParentID: s.channelzID,
663 MaxHeaderListSize: s.opts.maxHeaderListSize,
665 st, err := transport.NewServerTransport("http2", c, config)
668 s.errorf("NewServerTransport(%q) failed: %v", c.RemoteAddr(), err)
671 grpclog.Warningln("grpc: Server.Serve failed to create ServerTransport: ", err)
678 func (s *Server) serveStreams(st transport.ServerTransport) {
680 var wg sync.WaitGroup
681 st.HandleStreams(func(stream *transport.Stream) {
685 s.handleStream(st, stream, s.traceInfo(st, stream))
687 }, func(ctx context.Context, method string) context.Context {
691 tr := trace.New("grpc.Recv."+methodFamily(method), method)
692 return trace.NewContext(ctx, tr)
697 var _ http.Handler = (*Server)(nil)
699 // ServeHTTP implements the Go standard library's http.Handler
700 // interface by responding to the gRPC request r, by looking up
701 // the requested gRPC method in the gRPC server s.
703 // The provided HTTP request must have arrived on an HTTP/2
704 // connection. When using the Go standard library's server,
705 // practically this means that the Request must also have arrived
708 // To share one port (such as 443 for https) between gRPC and an
709 // existing http.Handler, use a root http.Handler such as:
711 // if r.ProtoMajor == 2 && strings.HasPrefix(
712 // r.Header.Get("Content-Type"), "application/grpc") {
713 // grpcServer.ServeHTTP(w, r)
715 // yourMux.ServeHTTP(w, r)
718 // Note that ServeHTTP uses Go's HTTP/2 server implementation which is totally
719 // separate from grpc-go's HTTP/2 server. Performance and features may vary
720 // between the two paths. ServeHTTP does not support some gRPC features
721 // available through grpc-go's HTTP/2 server, and it is currently EXPERIMENTAL
722 // and subject to change.
723 func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
724 st, err := transport.NewServerHandlerTransport(w, r, s.opts.statsHandler)
726 http.Error(w, err.Error(), http.StatusInternalServerError)
732 defer s.removeConn(st)
736 // traceInfo returns a traceInfo and associates it with stream, if tracing is enabled.
737 // If tracing is not enabled, it returns nil.
738 func (s *Server) traceInfo(st transport.ServerTransport, stream *transport.Stream) (trInfo *traceInfo) {
739 tr, ok := trace.FromContext(stream.Context())
747 trInfo.firstLine.client = false
748 trInfo.firstLine.remoteAddr = st.RemoteAddr()
750 if dl, ok := stream.Context().Deadline(); ok {
751 trInfo.firstLine.deadline = dl.Sub(time.Now())
756 func (s *Server) addConn(c io.Closer) bool {
764 // Transport added after we drained our existing conns: drain it
766 c.(transport.ServerTransport).Drain()
772 func (s *Server) removeConn(c io.Closer) {
781 func (s *Server) channelzMetric() *channelz.ServerInternalMetric {
782 return &channelz.ServerInternalMetric{
783 CallsStarted: atomic.LoadInt64(&s.czData.callsStarted),
784 CallsSucceeded: atomic.LoadInt64(&s.czData.callsSucceeded),
785 CallsFailed: atomic.LoadInt64(&s.czData.callsFailed),
786 LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&s.czData.lastCallStartedTime)),
790 func (s *Server) incrCallsStarted() {
791 atomic.AddInt64(&s.czData.callsStarted, 1)
792 atomic.StoreInt64(&s.czData.lastCallStartedTime, time.Now().UnixNano())
795 func (s *Server) incrCallsSucceeded() {
796 atomic.AddInt64(&s.czData.callsSucceeded, 1)
799 func (s *Server) incrCallsFailed() {
800 atomic.AddInt64(&s.czData.callsFailed, 1)
803 func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Stream, msg interface{}, cp Compressor, opts *transport.Options, comp encoding.Compressor) error {
804 data, err := encode(s.getCodec(stream.ContentSubtype()), msg)
806 grpclog.Errorln("grpc: server failed to encode response: ", err)
809 compData, err := compress(data, cp, comp)
811 grpclog.Errorln("grpc: server failed to compress response: ", err)
814 hdr, payload := msgHeader(data, compData)
815 // TODO(dfawley): should we be checking len(data) instead?
816 if len(payload) > s.opts.maxSendMessageSize {
817 return status.Errorf(codes.ResourceExhausted, "grpc: trying to send message larger than max (%d vs. %d)", len(payload), s.opts.maxSendMessageSize)
819 err = t.Write(stream, hdr, payload, opts)
820 if err == nil && s.opts.statsHandler != nil {
821 s.opts.statsHandler.HandleRPC(stream.Context(), outPayload(false, msg, data, payload, time.Now()))
826 func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
830 if err != nil && err != io.EOF {
833 s.incrCallsSucceeded()
837 sh := s.opts.statsHandler
839 beginTime := time.Now()
840 begin := &stats.Begin{
841 BeginTime: beginTime,
843 sh.HandleRPC(stream.Context(), begin)
846 BeginTime: beginTime,
849 if err != nil && err != io.EOF {
850 end.Error = toRPCErr(err)
852 sh.HandleRPC(stream.Context(), end)
856 defer trInfo.tr.Finish()
857 trInfo.firstLine.client = false
858 trInfo.tr.LazyLog(&trInfo.firstLine, false)
860 if err != nil && err != io.EOF {
861 trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
867 binlog := binarylog.GetMethodLogger(stream.Method())
869 ctx := stream.Context()
870 md, _ := metadata.FromIncomingContext(ctx)
871 logEntry := &binarylog.ClientHeader{
873 MethodName: stream.Method(),
876 if deadline, ok := ctx.Deadline(); ok {
877 logEntry.Timeout = deadline.Sub(time.Now())
878 if logEntry.Timeout < 0 {
882 if a := md[":authority"]; len(a) > 0 {
883 logEntry.Authority = a[0]
885 if peer, ok := peer.FromContext(ctx); ok {
886 logEntry.PeerAddr = peer.Addr
891 // comp and cp are used for compression. decomp and dc are used for
892 // decompression. If comp and decomp are both set, they are the same;
893 // however they are kept separate to ensure that at most one of the
894 // compressor/decompressor variable pairs are set for use later.
895 var comp, decomp encoding.Compressor
899 // If dc is set and matches the stream's compression, use it. Otherwise, try
900 // to find a matching registered compressor for decomp.
901 if rc := stream.RecvCompress(); s.opts.dc != nil && s.opts.dc.Type() == rc {
903 } else if rc != "" && rc != encoding.Identity {
904 decomp = encoding.GetCompressor(rc)
906 st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
907 t.WriteStatus(stream, st)
912 // If cp is set, use it. Otherwise, attempt to compress the response using
913 // the incoming message compression method.
915 // NOTE: this needs to be ahead of all handling, https://github.com/grpc/grpc-go/issues/686.
916 if s.opts.cp != nil {
918 stream.SetSendCompress(cp.Type())
919 } else if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
920 // Legacy compressor not specified; attempt to respond with same encoding.
921 comp = encoding.GetCompressor(rc)
923 stream.SetSendCompress(rc)
927 var payInfo *payloadInfo
928 if sh != nil || binlog != nil {
929 payInfo = &payloadInfo{}
931 d, err := recvAndDecompress(&parser{r: stream}, stream, dc, s.opts.maxReceiveMessageSize, payInfo, decomp)
933 if st, ok := status.FromError(err); ok {
934 if e := t.WriteStatus(stream, st); e != nil {
935 grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status %v", e)
943 df := func(v interface{}) error {
944 if err := s.getCodec(stream.ContentSubtype()).Unmarshal(d, v); err != nil {
945 return status.Errorf(codes.Internal, "grpc: error unmarshalling request: %v", err)
948 sh.HandleRPC(stream.Context(), &stats.InPayload{
949 RecvTime: time.Now(),
956 binlog.Log(&binarylog.ClientMessage{
961 trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)
965 ctx := NewContextWithServerTransportStream(stream.Context(), stream)
966 reply, appErr := md.Handler(srv.server, ctx, df, s.opts.unaryInt)
968 appStatus, ok := status.FromError(appErr)
970 // Convert appErr if it is not a grpc status error.
971 appErr = status.Error(codes.Unknown, appErr.Error())
972 appStatus, _ = status.FromError(appErr)
975 trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
978 if e := t.WriteStatus(stream, appStatus); e != nil {
979 grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
982 if h, _ := stream.Header(); h.Len() > 0 {
983 // Only log serverHeader if there was header. Otherwise it can
985 binlog.Log(&binarylog.ServerHeader{
989 binlog.Log(&binarylog.ServerTrailer{
990 Trailer: stream.Trailer(),
997 trInfo.tr.LazyLog(stringer("OK"), false)
999 opts := &transport.Options{Last: true}
1001 if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {
1003 // The entire stream is done (for unary RPC only).
1006 if s, ok := status.FromError(err); ok {
1007 if e := t.WriteStatus(stream, s); e != nil {
1008 grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
1011 switch st := err.(type) {
1012 case transport.ConnectionError:
1013 // Nothing to do here.
1015 panic(fmt.Sprintf("grpc: Unexpected error (%T) from sendResponse: %v", st, st))
1019 h, _ := stream.Header()
1020 binlog.Log(&binarylog.ServerHeader{
1023 binlog.Log(&binarylog.ServerTrailer{
1024 Trailer: stream.Trailer(),
1031 h, _ := stream.Header()
1032 binlog.Log(&binarylog.ServerHeader{
1035 binlog.Log(&binarylog.ServerMessage{
1039 if channelz.IsOn() {
1043 trInfo.tr.LazyLog(&payload{sent: true, msg: reply}, true)
1045 // TODO: Should we be logging if writing status failed here, like above?
1046 // Should the logging be in WriteStatus? Should we ignore the WriteStatus
1047 // error or allow the stats handler to see it?
1048 err = t.WriteStatus(stream, status.New(codes.OK, ""))
1050 binlog.Log(&binarylog.ServerTrailer{
1051 Trailer: stream.Trailer(),
1058 func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, sd *StreamDesc, trInfo *traceInfo) (err error) {
1059 if channelz.IsOn() {
1060 s.incrCallsStarted()
1062 if err != nil && err != io.EOF {
1065 s.incrCallsSucceeded()
1069 sh := s.opts.statsHandler
1071 beginTime := time.Now()
1072 begin := &stats.Begin{
1073 BeginTime: beginTime,
1075 sh.HandleRPC(stream.Context(), begin)
1078 BeginTime: beginTime,
1079 EndTime: time.Now(),
1081 if err != nil && err != io.EOF {
1082 end.Error = toRPCErr(err)
1084 sh.HandleRPC(stream.Context(), end)
1087 ctx := NewContextWithServerTransportStream(stream.Context(), stream)
1088 ss := &serverStream{
1092 p: &parser{r: stream},
1093 codec: s.getCodec(stream.ContentSubtype()),
1094 maxReceiveMessageSize: s.opts.maxReceiveMessageSize,
1095 maxSendMessageSize: s.opts.maxSendMessageSize,
1100 ss.binlog = binarylog.GetMethodLogger(stream.Method())
1101 if ss.binlog != nil {
1102 md, _ := metadata.FromIncomingContext(ctx)
1103 logEntry := &binarylog.ClientHeader{
1105 MethodName: stream.Method(),
1108 if deadline, ok := ctx.Deadline(); ok {
1109 logEntry.Timeout = deadline.Sub(time.Now())
1110 if logEntry.Timeout < 0 {
1111 logEntry.Timeout = 0
1114 if a := md[":authority"]; len(a) > 0 {
1115 logEntry.Authority = a[0]
1117 if peer, ok := peer.FromContext(ss.Context()); ok {
1118 logEntry.PeerAddr = peer.Addr
1120 ss.binlog.Log(logEntry)
1123 // If dc is set and matches the stream's compression, use it. Otherwise, try
1124 // to find a matching registered compressor for decomp.
1125 if rc := stream.RecvCompress(); s.opts.dc != nil && s.opts.dc.Type() == rc {
1127 } else if rc != "" && rc != encoding.Identity {
1128 ss.decomp = encoding.GetCompressor(rc)
1129 if ss.decomp == nil {
1130 st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
1131 t.WriteStatus(ss.s, st)
1136 // If cp is set, use it. Otherwise, attempt to compress the response using
1137 // the incoming message compression method.
1139 // NOTE: this needs to be ahead of all handling, https://github.com/grpc/grpc-go/issues/686.
1140 if s.opts.cp != nil {
1142 stream.SetSendCompress(s.opts.cp.Type())
1143 } else if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
1144 // Legacy compressor not specified; attempt to respond with same encoding.
1145 ss.comp = encoding.GetCompressor(rc)
1147 stream.SetSendCompress(rc)
1152 trInfo.tr.LazyLog(&trInfo.firstLine, false)
1155 if err != nil && err != io.EOF {
1156 ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1157 ss.trInfo.tr.SetError()
1159 ss.trInfo.tr.Finish()
1165 var server interface{}
1169 if s.opts.streamInt == nil {
1170 appErr = sd.Handler(server, ss)
1172 info := &StreamServerInfo{
1173 FullMethod: stream.Method(),
1174 IsClientStream: sd.ClientStreams,
1175 IsServerStream: sd.ServerStreams,
1177 appErr = s.opts.streamInt(server, ss, info, sd.Handler)
1180 appStatus, ok := status.FromError(appErr)
1182 appStatus = status.New(codes.Unknown, appErr.Error())
1183 appErr = appStatus.Err()
1187 ss.trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
1188 ss.trInfo.tr.SetError()
1191 t.WriteStatus(ss.s, appStatus)
1192 if ss.binlog != nil {
1193 ss.binlog.Log(&binarylog.ServerTrailer{
1194 Trailer: ss.s.Trailer(),
1198 // TODO: Should we log an error from WriteStatus here and below?
1203 ss.trInfo.tr.LazyLog(stringer("OK"), false)
1206 err = t.WriteStatus(ss.s, status.New(codes.OK, ""))
1207 if ss.binlog != nil {
1208 ss.binlog.Log(&binarylog.ServerTrailer{
1209 Trailer: ss.s.Trailer(),
1216 func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream, trInfo *traceInfo) {
1217 sm := stream.Method()
1218 if sm != "" && sm[0] == '/' {
1221 pos := strings.LastIndex(sm, "/")
1224 trInfo.tr.LazyLog(&fmtStringer{"Malformed method name %q", []interface{}{sm}}, true)
1225 trInfo.tr.SetError()
1227 errDesc := fmt.Sprintf("malformed method name: %q", stream.Method())
1228 if err := t.WriteStatus(stream, status.New(codes.ResourceExhausted, errDesc)); err != nil {
1230 trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1231 trInfo.tr.SetError()
1233 grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
1241 method := sm[pos+1:]
1243 if srv, ok := s.m[service]; ok {
1244 if md, ok := srv.md[method]; ok {
1245 s.processUnaryRPC(t, stream, srv, md, trInfo)
1248 if sd, ok := srv.sd[method]; ok {
1249 s.processStreamingRPC(t, stream, srv, sd, trInfo)
1253 // Unknown service, or known server unknown method.
1254 if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
1255 s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
1259 trInfo.tr.LazyLog(&fmtStringer{"Unknown service %v", []interface{}{service}}, true)
1260 trInfo.tr.SetError()
1262 errDesc := fmt.Sprintf("unknown service %v", service)
1263 if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
1265 trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1266 trInfo.tr.SetError()
1268 grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
1275 // The key to save ServerTransportStream in the context.
1276 type streamKey struct{}
1278 // NewContextWithServerTransportStream creates a new context from ctx and
1279 // attaches stream to it.
1281 // This API is EXPERIMENTAL.
1282 func NewContextWithServerTransportStream(ctx context.Context, stream ServerTransportStream) context.Context {
1283 return context.WithValue(ctx, streamKey{}, stream)
1286 // ServerTransportStream is a minimal interface that a transport stream must
1287 // implement. This can be used to mock an actual transport stream for tests of
1288 // handler code that use, for example, grpc.SetHeader (which requires some
1289 // stream to be in context).
1291 // See also NewContextWithServerTransportStream.
1293 // This API is EXPERIMENTAL.
1294 type ServerTransportStream interface {
1296 SetHeader(md metadata.MD) error
1297 SendHeader(md metadata.MD) error
1298 SetTrailer(md metadata.MD) error
1301 // ServerTransportStreamFromContext returns the ServerTransportStream saved in
1302 // ctx. Returns nil if the given context has no stream associated with it
1303 // (which implies it is not an RPC invocation context).
1305 // This API is EXPERIMENTAL.
1306 func ServerTransportStreamFromContext(ctx context.Context) ServerTransportStream {
1307 s, _ := ctx.Value(streamKey{}).(ServerTransportStream)
1311 // Stop stops the gRPC server. It immediately closes all open
1312 // connections and listeners.
1313 // It cancels all active RPCs on the server side and the corresponding
1314 // pending RPCs on the client side will get notified by connection
1316 func (s *Server) Stop() {
1317 s.quitOnce.Do(func() {
1323 s.doneOnce.Do(func() {
1328 s.channelzRemoveOnce.Do(func() {
1329 if channelz.IsOn() {
1330 channelz.RemoveEntry(s.channelzID)
1339 // interrupt GracefulStop if Stop and GracefulStop are called concurrently.
1343 for lis := range listeners {
1351 if s.events != nil {
1358 // GracefulStop stops the gRPC server gracefully. It stops the server from
1359 // accepting new connections and RPCs and blocks until all the pending RPCs are
1361 func (s *Server) GracefulStop() {
1362 s.quitOnce.Do(func() {
1367 s.doneOnce.Do(func() {
1372 s.channelzRemoveOnce.Do(func() {
1373 if channelz.IsOn() {
1374 channelz.RemoveEntry(s.channelzID)
1383 for lis := range s.lis {
1388 for c := range s.conns {
1389 c.(transport.ServerTransport).Drain()
1394 // Wait for serving threads to be ready to exit. Only then can we be sure no
1395 // new conns will be created.
1400 for len(s.conns) != 0 {
1404 if s.events != nil {
1411 // contentSubtype must be lowercase
1412 // cannot return nil
1413 func (s *Server) getCodec(contentSubtype string) baseCodec {
1414 if s.opts.codec != nil {
1417 if contentSubtype == "" {
1418 return encoding.GetCodec(proto.Name)
1420 codec := encoding.GetCodec(contentSubtype)
1422 return encoding.GetCodec(proto.Name)
1427 // SetHeader sets the header metadata.
1428 // When called multiple times, all the provided metadata will be merged.
1429 // All the metadata will be sent out when one of the following happens:
1430 // - grpc.SendHeader() is called;
1431 // - The first response is sent out;
1432 // - An RPC status is sent out (error or success).
1433 func SetHeader(ctx context.Context, md metadata.MD) error {
1437 stream := ServerTransportStreamFromContext(ctx)
1439 return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
1441 return stream.SetHeader(md)
1444 // SendHeader sends header metadata. It may be called at most once.
1445 // The provided md and headers set by SetHeader() will be sent.
1446 func SendHeader(ctx context.Context, md metadata.MD) error {
1447 stream := ServerTransportStreamFromContext(ctx)
1449 return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
1451 if err := stream.SendHeader(md); err != nil {
1452 return toRPCErr(err)
1457 // SetTrailer sets the trailer metadata that will be sent when an RPC returns.
1458 // When called more than once, all the provided metadata will be merged.
1459 func SetTrailer(ctx context.Context, md metadata.MD) error {
1463 stream := ServerTransportStreamFromContext(ctx)
1465 return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
1467 return stream.SetTrailer(md)
1470 // Method returns the method string for the server context. The returned
1471 // string is in the format of "/service/method".
1472 func Method(ctx context.Context) (string, bool) {
1473 s := ServerTransportStreamFromContext(ctx)
1477 return s.Method(), true
1480 type channelzServer struct {
1484 func (c *channelzServer) ChannelzMetric() *channelz.ServerInternalMetric {
1485 return c.s.channelzMetric()