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 if kp.Time > 0 && kp.Time < time.Second {
186 grpclog.Warning("Adjusting keepalive ping interval to minimum period of 1s")
187 kp.Time = time.Second
190 return func(o *options) {
191 o.keepaliveParams = kp
195 // KeepaliveEnforcementPolicy returns a ServerOption that sets keepalive enforcement policy for the server.
196 func KeepaliveEnforcementPolicy(kep keepalive.EnforcementPolicy) ServerOption {
197 return func(o *options) {
198 o.keepalivePolicy = kep
202 // CustomCodec returns a ServerOption that sets a codec for message marshaling and unmarshaling.
204 // This will override any lookups by content-subtype for Codecs registered with RegisterCodec.
205 func CustomCodec(codec Codec) ServerOption {
206 return func(o *options) {
211 // RPCCompressor returns a ServerOption that sets a compressor for outbound
212 // messages. For backward compatibility, all outbound messages will be sent
213 // using this compressor, regardless of incoming message compression. By
214 // default, server messages will be sent using the same compressor with which
215 // request messages were sent.
217 // Deprecated: use encoding.RegisterCompressor instead.
218 func RPCCompressor(cp Compressor) ServerOption {
219 return func(o *options) {
224 // RPCDecompressor returns a ServerOption that sets a decompressor for inbound
225 // messages. It has higher priority than decompressors registered via
226 // encoding.RegisterCompressor.
228 // Deprecated: use encoding.RegisterCompressor instead.
229 func RPCDecompressor(dc Decompressor) ServerOption {
230 return func(o *options) {
235 // MaxMsgSize returns a ServerOption to set the max message size in bytes the server can receive.
236 // If this is not set, gRPC uses the default limit.
238 // Deprecated: use MaxRecvMsgSize instead.
239 func MaxMsgSize(m int) ServerOption {
240 return MaxRecvMsgSize(m)
243 // MaxRecvMsgSize returns a ServerOption to set the max message size in bytes the server can receive.
244 // If this is not set, gRPC uses the default 4MB.
245 func MaxRecvMsgSize(m int) ServerOption {
246 return func(o *options) {
247 o.maxReceiveMessageSize = m
251 // MaxSendMsgSize returns a ServerOption to set the max message size in bytes the server can send.
252 // If this is not set, gRPC uses the default `math.MaxInt32`.
253 func MaxSendMsgSize(m int) ServerOption {
254 return func(o *options) {
255 o.maxSendMessageSize = m
259 // MaxConcurrentStreams returns a ServerOption that will apply a limit on the number
260 // of concurrent streams to each ServerTransport.
261 func MaxConcurrentStreams(n uint32) ServerOption {
262 return func(o *options) {
263 o.maxConcurrentStreams = n
267 // Creds returns a ServerOption that sets credentials for server connections.
268 func Creds(c credentials.TransportCredentials) ServerOption {
269 return func(o *options) {
274 // UnaryInterceptor returns a ServerOption that sets the UnaryServerInterceptor for the
275 // server. Only one unary interceptor can be installed. The construction of multiple
276 // interceptors (e.g., chaining) can be implemented at the caller.
277 func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {
278 return func(o *options) {
279 if o.unaryInt != nil {
280 panic("The unary server interceptor was already set and may not be reset.")
286 // StreamInterceptor returns a ServerOption that sets the StreamServerInterceptor for the
287 // server. Only one stream interceptor can be installed.
288 func StreamInterceptor(i StreamServerInterceptor) ServerOption {
289 return func(o *options) {
290 if o.streamInt != nil {
291 panic("The stream server interceptor was already set and may not be reset.")
297 // InTapHandle returns a ServerOption that sets the tap handle for all the server
298 // transport to be created. Only one can be installed.
299 func InTapHandle(h tap.ServerInHandle) ServerOption {
300 return func(o *options) {
301 if o.inTapHandle != nil {
302 panic("The tap handle was already set and may not be reset.")
308 // StatsHandler returns a ServerOption that sets the stats handler for the server.
309 func StatsHandler(h stats.Handler) ServerOption {
310 return func(o *options) {
315 // UnknownServiceHandler returns a ServerOption that allows for adding a custom
316 // unknown service handler. The provided method is a bidi-streaming RPC service
317 // handler that will be invoked instead of returning the "unimplemented" gRPC
318 // error whenever a request is received for an unregistered service or method.
319 // The handling function has full access to the Context of the request and the
320 // stream, and the invocation bypasses interceptors.
321 func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {
322 return func(o *options) {
323 o.unknownStreamDesc = &StreamDesc{
324 StreamName: "unknown_service_handler",
325 Handler: streamHandler,
326 // We need to assume that the users of the streamHandler will want to use both.
333 // ConnectionTimeout returns a ServerOption that sets the timeout for
334 // connection establishment (up to and including HTTP/2 handshaking) for all
335 // new connections. If this is not set, the default is 120 seconds. A zero or
336 // negative value will result in an immediate timeout.
338 // This API is EXPERIMENTAL.
339 func ConnectionTimeout(d time.Duration) ServerOption {
340 return func(o *options) {
341 o.connectionTimeout = d
345 // MaxHeaderListSize returns a ServerOption that sets the max (uncompressed) size
346 // of header list that the server is prepared to accept.
347 func MaxHeaderListSize(s uint32) ServerOption {
348 return func(o *options) {
349 o.maxHeaderListSize = &s
353 // NewServer creates a gRPC server which has no service registered and has not
354 // started to accept requests yet.
355 func NewServer(opt ...ServerOption) *Server {
356 opts := defaultServerOptions
357 for _, o := range opt {
361 lis: make(map[net.Listener]bool),
363 conns: make(map[io.Closer]bool),
364 m: make(map[string]*service),
365 quit: make(chan struct{}),
366 done: make(chan struct{}),
367 czData: new(channelzData),
369 s.cv = sync.NewCond(&s.mu)
371 _, file, line, _ := runtime.Caller(1)
372 s.events = trace.NewEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
376 s.channelzID = channelz.RegisterServer(&channelzServer{s}, "")
381 // printf records an event in s's event log, unless s has been stopped.
382 // REQUIRES s.mu is held.
383 func (s *Server) printf(format string, a ...interface{}) {
385 s.events.Printf(format, a...)
389 // errorf records an error in s's event log, unless s has been stopped.
390 // REQUIRES s.mu is held.
391 func (s *Server) errorf(format string, a ...interface{}) {
393 s.events.Errorf(format, a...)
397 // RegisterService registers a service and its implementation to the gRPC
398 // server. It is called from the IDL generated code. This must be called before
400 func (s *Server) RegisterService(sd *ServiceDesc, ss interface{}) {
401 ht := reflect.TypeOf(sd.HandlerType).Elem()
402 st := reflect.TypeOf(ss)
403 if !st.Implements(ht) {
404 grpclog.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)
409 func (s *Server) register(sd *ServiceDesc, ss interface{}) {
412 s.printf("RegisterService(%q)", sd.ServiceName)
414 grpclog.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)
416 if _, ok := s.m[sd.ServiceName]; ok {
417 grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
421 md: make(map[string]*MethodDesc),
422 sd: make(map[string]*StreamDesc),
425 for i := range sd.Methods {
427 srv.md[d.MethodName] = d
429 for i := range sd.Streams {
431 srv.sd[d.StreamName] = d
433 s.m[sd.ServiceName] = srv
436 // MethodInfo contains the information of an RPC including its method name and type.
437 type MethodInfo struct {
438 // Name is the method name only, without the service name or package name.
440 // IsClientStream indicates whether the RPC is a client streaming RPC.
442 // IsServerStream indicates whether the RPC is a server streaming RPC.
446 // ServiceInfo contains unary RPC method info, streaming RPC method info and metadata for a service.
447 type ServiceInfo struct {
449 // Metadata is the metadata specified in ServiceDesc when registering service.
453 // GetServiceInfo returns a map from service names to ServiceInfo.
454 // Service names include the package names, in the form of <package>.<service>.
455 func (s *Server) GetServiceInfo() map[string]ServiceInfo {
456 ret := make(map[string]ServiceInfo)
457 for n, srv := range s.m {
458 methods := make([]MethodInfo, 0, len(srv.md)+len(srv.sd))
459 for m := range srv.md {
460 methods = append(methods, MethodInfo{
462 IsClientStream: false,
463 IsServerStream: false,
466 for m, d := range srv.sd {
467 methods = append(methods, MethodInfo{
469 IsClientStream: d.ClientStreams,
470 IsServerStream: d.ServerStreams,
474 ret[n] = ServiceInfo{
482 // ErrServerStopped indicates that the operation is now illegal because of
483 // the server being stopped.
484 var ErrServerStopped = errors.New("grpc: the server has been stopped")
486 func (s *Server) useTransportAuthenticator(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
487 if s.opts.creds == nil {
488 return rawConn, nil, nil
490 return s.opts.creds.ServerHandshake(rawConn)
493 type listenSocket struct {
498 func (l *listenSocket) ChannelzMetric() *channelz.SocketInternalMetric {
499 return &channelz.SocketInternalMetric{
500 SocketOptions: channelz.GetSocketOption(l.Listener),
501 LocalAddr: l.Listener.Addr(),
505 func (l *listenSocket) Close() error {
506 err := l.Listener.Close()
508 channelz.RemoveEntry(l.channelzID)
513 // Serve accepts incoming connections on the listener lis, creating a new
514 // ServerTransport and service goroutine for each. The service goroutines
515 // read gRPC requests and then call the registered handlers to reply to them.
516 // Serve returns when lis.Accept fails with fatal errors. lis will be closed when
517 // this method returns.
518 // Serve will return a non-nil error unless Stop or GracefulStop is called.
519 func (s *Server) Serve(lis net.Listener) error {
524 // Serve called after Stop or GracefulStop.
527 return ErrServerStopped
534 // Stop or GracefulStop called; block until done and return nil.
541 ls := &listenSocket{Listener: lis}
545 ls.channelzID = channelz.RegisterListenSocket(ls, s.channelzID, lis.Addr().String())
551 if s.lis != nil && s.lis[ls] {
558 var tempDelay time.Duration // how long to sleep on accept failure
561 rawConn, err := lis.Accept()
563 if ne, ok := err.(interface {
565 }); ok && ne.Temporary() {
567 tempDelay = 5 * time.Millisecond
571 if max := 1 * time.Second; tempDelay > max {
575 s.printf("Accept error: %v; retrying in %v", err, tempDelay)
577 timer := time.NewTimer(tempDelay)
587 s.printf("done serving; Accept = %v", err)
598 // Start a new goroutine to deal with rawConn so we don't stall this Accept
601 // Make sure we account for the goroutine so GracefulStop doesn't nil out
602 // s.conns before this conn can be added.
605 s.handleRawConn(rawConn)
611 // handleRawConn forks a goroutine to handle a just-accepted connection that
612 // has not had any I/O performed on it yet.
613 func (s *Server) handleRawConn(rawConn net.Conn) {
614 rawConn.SetDeadline(time.Now().Add(s.opts.connectionTimeout))
615 conn, authInfo, err := s.useTransportAuthenticator(rawConn)
618 s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
620 grpclog.Warningf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err)
621 // If serverHandshake returns ErrConnDispatched, keep rawConn open.
622 if err != credentials.ErrConnDispatched {
625 rawConn.SetDeadline(time.Time{})
637 // Finish handshaking (HTTP2)
638 st := s.newHTTP2Transport(conn, authInfo)
643 rawConn.SetDeadline(time.Time{})
653 // newHTTP2Transport sets up a http/2 transport (using the
654 // gRPC http2 server transport in transport/http2_server.go).
655 func (s *Server) newHTTP2Transport(c net.Conn, authInfo credentials.AuthInfo) transport.ServerTransport {
656 config := &transport.ServerConfig{
657 MaxStreams: s.opts.maxConcurrentStreams,
659 InTapHandle: s.opts.inTapHandle,
660 StatsHandler: s.opts.statsHandler,
661 KeepaliveParams: s.opts.keepaliveParams,
662 KeepalivePolicy: s.opts.keepalivePolicy,
663 InitialWindowSize: s.opts.initialWindowSize,
664 InitialConnWindowSize: s.opts.initialConnWindowSize,
665 WriteBufferSize: s.opts.writeBufferSize,
666 ReadBufferSize: s.opts.readBufferSize,
667 ChannelzParentID: s.channelzID,
668 MaxHeaderListSize: s.opts.maxHeaderListSize,
670 st, err := transport.NewServerTransport("http2", c, config)
673 s.errorf("NewServerTransport(%q) failed: %v", c.RemoteAddr(), err)
676 grpclog.Warningln("grpc: Server.Serve failed to create ServerTransport: ", err)
683 func (s *Server) serveStreams(st transport.ServerTransport) {
685 var wg sync.WaitGroup
686 st.HandleStreams(func(stream *transport.Stream) {
690 s.handleStream(st, stream, s.traceInfo(st, stream))
692 }, func(ctx context.Context, method string) context.Context {
696 tr := trace.New("grpc.Recv."+methodFamily(method), method)
697 return trace.NewContext(ctx, tr)
702 var _ http.Handler = (*Server)(nil)
704 // ServeHTTP implements the Go standard library's http.Handler
705 // interface by responding to the gRPC request r, by looking up
706 // the requested gRPC method in the gRPC server s.
708 // The provided HTTP request must have arrived on an HTTP/2
709 // connection. When using the Go standard library's server,
710 // practically this means that the Request must also have arrived
713 // To share one port (such as 443 for https) between gRPC and an
714 // existing http.Handler, use a root http.Handler such as:
716 // if r.ProtoMajor == 2 && strings.HasPrefix(
717 // r.Header.Get("Content-Type"), "application/grpc") {
718 // grpcServer.ServeHTTP(w, r)
720 // yourMux.ServeHTTP(w, r)
723 // Note that ServeHTTP uses Go's HTTP/2 server implementation which is totally
724 // separate from grpc-go's HTTP/2 server. Performance and features may vary
725 // between the two paths. ServeHTTP does not support some gRPC features
726 // available through grpc-go's HTTP/2 server, and it is currently EXPERIMENTAL
727 // and subject to change.
728 func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
729 st, err := transport.NewServerHandlerTransport(w, r, s.opts.statsHandler)
731 http.Error(w, err.Error(), http.StatusInternalServerError)
737 defer s.removeConn(st)
741 // traceInfo returns a traceInfo and associates it with stream, if tracing is enabled.
742 // If tracing is not enabled, it returns nil.
743 func (s *Server) traceInfo(st transport.ServerTransport, stream *transport.Stream) (trInfo *traceInfo) {
744 tr, ok := trace.FromContext(stream.Context())
752 trInfo.firstLine.client = false
753 trInfo.firstLine.remoteAddr = st.RemoteAddr()
755 if dl, ok := stream.Context().Deadline(); ok {
756 trInfo.firstLine.deadline = time.Until(dl)
761 func (s *Server) addConn(c io.Closer) bool {
769 // Transport added after we drained our existing conns: drain it
771 c.(transport.ServerTransport).Drain()
777 func (s *Server) removeConn(c io.Closer) {
786 func (s *Server) channelzMetric() *channelz.ServerInternalMetric {
787 return &channelz.ServerInternalMetric{
788 CallsStarted: atomic.LoadInt64(&s.czData.callsStarted),
789 CallsSucceeded: atomic.LoadInt64(&s.czData.callsSucceeded),
790 CallsFailed: atomic.LoadInt64(&s.czData.callsFailed),
791 LastCallStartedTimestamp: time.Unix(0, atomic.LoadInt64(&s.czData.lastCallStartedTime)),
795 func (s *Server) incrCallsStarted() {
796 atomic.AddInt64(&s.czData.callsStarted, 1)
797 atomic.StoreInt64(&s.czData.lastCallStartedTime, time.Now().UnixNano())
800 func (s *Server) incrCallsSucceeded() {
801 atomic.AddInt64(&s.czData.callsSucceeded, 1)
804 func (s *Server) incrCallsFailed() {
805 atomic.AddInt64(&s.czData.callsFailed, 1)
808 func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Stream, msg interface{}, cp Compressor, opts *transport.Options, comp encoding.Compressor) error {
809 data, err := encode(s.getCodec(stream.ContentSubtype()), msg)
811 grpclog.Errorln("grpc: server failed to encode response: ", err)
814 compData, err := compress(data, cp, comp)
816 grpclog.Errorln("grpc: server failed to compress response: ", err)
819 hdr, payload := msgHeader(data, compData)
820 // TODO(dfawley): should we be checking len(data) instead?
821 if len(payload) > s.opts.maxSendMessageSize {
822 return status.Errorf(codes.ResourceExhausted, "grpc: trying to send message larger than max (%d vs. %d)", len(payload), s.opts.maxSendMessageSize)
824 err = t.Write(stream, hdr, payload, opts)
825 if err == nil && s.opts.statsHandler != nil {
826 s.opts.statsHandler.HandleRPC(stream.Context(), outPayload(false, msg, data, payload, time.Now()))
831 func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
835 if err != nil && err != io.EOF {
838 s.incrCallsSucceeded()
842 sh := s.opts.statsHandler
844 beginTime := time.Now()
845 begin := &stats.Begin{
846 BeginTime: beginTime,
848 sh.HandleRPC(stream.Context(), begin)
851 BeginTime: beginTime,
854 if err != nil && err != io.EOF {
855 end.Error = toRPCErr(err)
857 sh.HandleRPC(stream.Context(), end)
861 defer trInfo.tr.Finish()
862 trInfo.firstLine.client = false
863 trInfo.tr.LazyLog(&trInfo.firstLine, false)
865 if err != nil && err != io.EOF {
866 trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
872 binlog := binarylog.GetMethodLogger(stream.Method())
874 ctx := stream.Context()
875 md, _ := metadata.FromIncomingContext(ctx)
876 logEntry := &binarylog.ClientHeader{
878 MethodName: stream.Method(),
881 if deadline, ok := ctx.Deadline(); ok {
882 logEntry.Timeout = time.Until(deadline)
883 if logEntry.Timeout < 0 {
887 if a := md[":authority"]; len(a) > 0 {
888 logEntry.Authority = a[0]
890 if peer, ok := peer.FromContext(ctx); ok {
891 logEntry.PeerAddr = peer.Addr
896 // comp and cp are used for compression. decomp and dc are used for
897 // decompression. If comp and decomp are both set, they are the same;
898 // however they are kept separate to ensure that at most one of the
899 // compressor/decompressor variable pairs are set for use later.
900 var comp, decomp encoding.Compressor
904 // If dc is set and matches the stream's compression, use it. Otherwise, try
905 // to find a matching registered compressor for decomp.
906 if rc := stream.RecvCompress(); s.opts.dc != nil && s.opts.dc.Type() == rc {
908 } else if rc != "" && rc != encoding.Identity {
909 decomp = encoding.GetCompressor(rc)
911 st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
912 t.WriteStatus(stream, st)
917 // If cp is set, use it. Otherwise, attempt to compress the response using
918 // the incoming message compression method.
920 // NOTE: this needs to be ahead of all handling, https://github.com/grpc/grpc-go/issues/686.
921 if s.opts.cp != nil {
923 stream.SetSendCompress(cp.Type())
924 } else if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
925 // Legacy compressor not specified; attempt to respond with same encoding.
926 comp = encoding.GetCompressor(rc)
928 stream.SetSendCompress(rc)
932 var payInfo *payloadInfo
933 if sh != nil || binlog != nil {
934 payInfo = &payloadInfo{}
936 d, err := recvAndDecompress(&parser{r: stream}, stream, dc, s.opts.maxReceiveMessageSize, payInfo, decomp)
938 if st, ok := status.FromError(err); ok {
939 if e := t.WriteStatus(stream, st); e != nil {
940 grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status %v", e)
948 df := func(v interface{}) error {
949 if err := s.getCodec(stream.ContentSubtype()).Unmarshal(d, v); err != nil {
950 return status.Errorf(codes.Internal, "grpc: error unmarshalling request: %v", err)
953 sh.HandleRPC(stream.Context(), &stats.InPayload{
954 RecvTime: time.Now(),
961 binlog.Log(&binarylog.ClientMessage{
966 trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)
970 ctx := NewContextWithServerTransportStream(stream.Context(), stream)
971 reply, appErr := md.Handler(srv.server, ctx, df, s.opts.unaryInt)
973 appStatus, ok := status.FromError(appErr)
975 // Convert appErr if it is not a grpc status error.
976 appErr = status.Error(codes.Unknown, appErr.Error())
977 appStatus, _ = status.FromError(appErr)
980 trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
983 if e := t.WriteStatus(stream, appStatus); e != nil {
984 grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
987 if h, _ := stream.Header(); h.Len() > 0 {
988 // Only log serverHeader if there was header. Otherwise it can
990 binlog.Log(&binarylog.ServerHeader{
994 binlog.Log(&binarylog.ServerTrailer{
995 Trailer: stream.Trailer(),
1002 trInfo.tr.LazyLog(stringer("OK"), false)
1004 opts := &transport.Options{Last: true}
1006 if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {
1008 // The entire stream is done (for unary RPC only).
1011 if s, ok := status.FromError(err); ok {
1012 if e := t.WriteStatus(stream, s); e != nil {
1013 grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
1016 switch st := err.(type) {
1017 case transport.ConnectionError:
1018 // Nothing to do here.
1020 panic(fmt.Sprintf("grpc: Unexpected error (%T) from sendResponse: %v", st, st))
1024 h, _ := stream.Header()
1025 binlog.Log(&binarylog.ServerHeader{
1028 binlog.Log(&binarylog.ServerTrailer{
1029 Trailer: stream.Trailer(),
1036 h, _ := stream.Header()
1037 binlog.Log(&binarylog.ServerHeader{
1040 binlog.Log(&binarylog.ServerMessage{
1044 if channelz.IsOn() {
1048 trInfo.tr.LazyLog(&payload{sent: true, msg: reply}, true)
1050 // TODO: Should we be logging if writing status failed here, like above?
1051 // Should the logging be in WriteStatus? Should we ignore the WriteStatus
1052 // error or allow the stats handler to see it?
1053 err = t.WriteStatus(stream, status.New(codes.OK, ""))
1055 binlog.Log(&binarylog.ServerTrailer{
1056 Trailer: stream.Trailer(),
1063 func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, sd *StreamDesc, trInfo *traceInfo) (err error) {
1064 if channelz.IsOn() {
1065 s.incrCallsStarted()
1067 if err != nil && err != io.EOF {
1070 s.incrCallsSucceeded()
1074 sh := s.opts.statsHandler
1076 beginTime := time.Now()
1077 begin := &stats.Begin{
1078 BeginTime: beginTime,
1080 sh.HandleRPC(stream.Context(), begin)
1083 BeginTime: beginTime,
1084 EndTime: time.Now(),
1086 if err != nil && err != io.EOF {
1087 end.Error = toRPCErr(err)
1089 sh.HandleRPC(stream.Context(), end)
1092 ctx := NewContextWithServerTransportStream(stream.Context(), stream)
1093 ss := &serverStream{
1097 p: &parser{r: stream},
1098 codec: s.getCodec(stream.ContentSubtype()),
1099 maxReceiveMessageSize: s.opts.maxReceiveMessageSize,
1100 maxSendMessageSize: s.opts.maxSendMessageSize,
1105 ss.binlog = binarylog.GetMethodLogger(stream.Method())
1106 if ss.binlog != nil {
1107 md, _ := metadata.FromIncomingContext(ctx)
1108 logEntry := &binarylog.ClientHeader{
1110 MethodName: stream.Method(),
1113 if deadline, ok := ctx.Deadline(); ok {
1114 logEntry.Timeout = time.Until(deadline)
1115 if logEntry.Timeout < 0 {
1116 logEntry.Timeout = 0
1119 if a := md[":authority"]; len(a) > 0 {
1120 logEntry.Authority = a[0]
1122 if peer, ok := peer.FromContext(ss.Context()); ok {
1123 logEntry.PeerAddr = peer.Addr
1125 ss.binlog.Log(logEntry)
1128 // If dc is set and matches the stream's compression, use it. Otherwise, try
1129 // to find a matching registered compressor for decomp.
1130 if rc := stream.RecvCompress(); s.opts.dc != nil && s.opts.dc.Type() == rc {
1132 } else if rc != "" && rc != encoding.Identity {
1133 ss.decomp = encoding.GetCompressor(rc)
1134 if ss.decomp == nil {
1135 st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
1136 t.WriteStatus(ss.s, st)
1141 // If cp is set, use it. Otherwise, attempt to compress the response using
1142 // the incoming message compression method.
1144 // NOTE: this needs to be ahead of all handling, https://github.com/grpc/grpc-go/issues/686.
1145 if s.opts.cp != nil {
1147 stream.SetSendCompress(s.opts.cp.Type())
1148 } else if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
1149 // Legacy compressor not specified; attempt to respond with same encoding.
1150 ss.comp = encoding.GetCompressor(rc)
1152 stream.SetSendCompress(rc)
1157 trInfo.tr.LazyLog(&trInfo.firstLine, false)
1160 if err != nil && err != io.EOF {
1161 ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1162 ss.trInfo.tr.SetError()
1164 ss.trInfo.tr.Finish()
1170 var server interface{}
1174 if s.opts.streamInt == nil {
1175 appErr = sd.Handler(server, ss)
1177 info := &StreamServerInfo{
1178 FullMethod: stream.Method(),
1179 IsClientStream: sd.ClientStreams,
1180 IsServerStream: sd.ServerStreams,
1182 appErr = s.opts.streamInt(server, ss, info, sd.Handler)
1185 appStatus, ok := status.FromError(appErr)
1187 appStatus = status.New(codes.Unknown, appErr.Error())
1188 appErr = appStatus.Err()
1192 ss.trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
1193 ss.trInfo.tr.SetError()
1196 t.WriteStatus(ss.s, appStatus)
1197 if ss.binlog != nil {
1198 ss.binlog.Log(&binarylog.ServerTrailer{
1199 Trailer: ss.s.Trailer(),
1203 // TODO: Should we log an error from WriteStatus here and below?
1208 ss.trInfo.tr.LazyLog(stringer("OK"), false)
1211 err = t.WriteStatus(ss.s, status.New(codes.OK, ""))
1212 if ss.binlog != nil {
1213 ss.binlog.Log(&binarylog.ServerTrailer{
1214 Trailer: ss.s.Trailer(),
1221 func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream, trInfo *traceInfo) {
1222 sm := stream.Method()
1223 if sm != "" && sm[0] == '/' {
1226 pos := strings.LastIndex(sm, "/")
1229 trInfo.tr.LazyLog(&fmtStringer{"Malformed method name %q", []interface{}{sm}}, true)
1230 trInfo.tr.SetError()
1232 errDesc := fmt.Sprintf("malformed method name: %q", stream.Method())
1233 if err := t.WriteStatus(stream, status.New(codes.ResourceExhausted, errDesc)); err != nil {
1235 trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1236 trInfo.tr.SetError()
1238 grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
1246 method := sm[pos+1:]
1248 if srv, ok := s.m[service]; ok {
1249 if md, ok := srv.md[method]; ok {
1250 s.processUnaryRPC(t, stream, srv, md, trInfo)
1253 if sd, ok := srv.sd[method]; ok {
1254 s.processStreamingRPC(t, stream, srv, sd, trInfo)
1258 // Unknown service, or known server unknown method.
1259 if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
1260 s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
1264 trInfo.tr.LazyLog(&fmtStringer{"Unknown service %v", []interface{}{service}}, true)
1265 trInfo.tr.SetError()
1267 errDesc := fmt.Sprintf("unknown service %v", service)
1268 if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
1270 trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
1271 trInfo.tr.SetError()
1273 grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
1280 // The key to save ServerTransportStream in the context.
1281 type streamKey struct{}
1283 // NewContextWithServerTransportStream creates a new context from ctx and
1284 // attaches stream to it.
1286 // This API is EXPERIMENTAL.
1287 func NewContextWithServerTransportStream(ctx context.Context, stream ServerTransportStream) context.Context {
1288 return context.WithValue(ctx, streamKey{}, stream)
1291 // ServerTransportStream is a minimal interface that a transport stream must
1292 // implement. This can be used to mock an actual transport stream for tests of
1293 // handler code that use, for example, grpc.SetHeader (which requires some
1294 // stream to be in context).
1296 // See also NewContextWithServerTransportStream.
1298 // This API is EXPERIMENTAL.
1299 type ServerTransportStream interface {
1301 SetHeader(md metadata.MD) error
1302 SendHeader(md metadata.MD) error
1303 SetTrailer(md metadata.MD) error
1306 // ServerTransportStreamFromContext returns the ServerTransportStream saved in
1307 // ctx. Returns nil if the given context has no stream associated with it
1308 // (which implies it is not an RPC invocation context).
1310 // This API is EXPERIMENTAL.
1311 func ServerTransportStreamFromContext(ctx context.Context) ServerTransportStream {
1312 s, _ := ctx.Value(streamKey{}).(ServerTransportStream)
1316 // Stop stops the gRPC server. It immediately closes all open
1317 // connections and listeners.
1318 // It cancels all active RPCs on the server side and the corresponding
1319 // pending RPCs on the client side will get notified by connection
1321 func (s *Server) Stop() {
1322 s.quitOnce.Do(func() {
1328 s.doneOnce.Do(func() {
1333 s.channelzRemoveOnce.Do(func() {
1334 if channelz.IsOn() {
1335 channelz.RemoveEntry(s.channelzID)
1344 // interrupt GracefulStop if Stop and GracefulStop are called concurrently.
1348 for lis := range listeners {
1356 if s.events != nil {
1363 // GracefulStop stops the gRPC server gracefully. It stops the server from
1364 // accepting new connections and RPCs and blocks until all the pending RPCs are
1366 func (s *Server) GracefulStop() {
1367 s.quitOnce.Do(func() {
1372 s.doneOnce.Do(func() {
1377 s.channelzRemoveOnce.Do(func() {
1378 if channelz.IsOn() {
1379 channelz.RemoveEntry(s.channelzID)
1388 for lis := range s.lis {
1393 for c := range s.conns {
1394 c.(transport.ServerTransport).Drain()
1399 // Wait for serving threads to be ready to exit. Only then can we be sure no
1400 // new conns will be created.
1405 for len(s.conns) != 0 {
1409 if s.events != nil {
1416 // contentSubtype must be lowercase
1417 // cannot return nil
1418 func (s *Server) getCodec(contentSubtype string) baseCodec {
1419 if s.opts.codec != nil {
1422 if contentSubtype == "" {
1423 return encoding.GetCodec(proto.Name)
1425 codec := encoding.GetCodec(contentSubtype)
1427 return encoding.GetCodec(proto.Name)
1432 // SetHeader sets the header metadata.
1433 // When called multiple times, all the provided metadata will be merged.
1434 // All the metadata will be sent out when one of the following happens:
1435 // - grpc.SendHeader() is called;
1436 // - The first response is sent out;
1437 // - An RPC status is sent out (error or success).
1438 func SetHeader(ctx context.Context, md metadata.MD) error {
1442 stream := ServerTransportStreamFromContext(ctx)
1444 return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
1446 return stream.SetHeader(md)
1449 // SendHeader sends header metadata. It may be called at most once.
1450 // The provided md and headers set by SetHeader() will be sent.
1451 func SendHeader(ctx context.Context, md metadata.MD) error {
1452 stream := ServerTransportStreamFromContext(ctx)
1454 return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
1456 if err := stream.SendHeader(md); err != nil {
1457 return toRPCErr(err)
1462 // SetTrailer sets the trailer metadata that will be sent when an RPC returns.
1463 // When called more than once, all the provided metadata will be merged.
1464 func SetTrailer(ctx context.Context, md metadata.MD) error {
1468 stream := ServerTransportStreamFromContext(ctx)
1470 return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
1472 return stream.SetTrailer(md)
1475 // Method returns the method string for the server context. The returned
1476 // string is in the format of "/service/method".
1477 func Method(ctx context.Context) (string, bool) {
1478 s := ServerTransportStreamFromContext(ctx)
1482 return s.Method(), true
1485 type channelzServer struct {
1489 func (c *channelzServer) ChannelzMetric() *channelz.ServerInternalMetric {
1490 return c.s.channelzMetric()