/*

*

* Copyright 2014 gRPC authors.

*

* Licensed under the Apache License, Version 2.0 (the "License");

* you may not use this file except in compliance with the License.

* You may obtain a copy of the License at

*

* http://www.apache.org/licenses/LICENSE-2.0

*

* Unless required by applicable law or agreed to in writing, software

* distributed under the License is distributed on an "AS IS" BASIS,

* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

* See the License for the specific language governing permissions and

* limitations under the License.

*

*/

package grpc

import (

"context"

"errors"

"fmt"

"io"

"math"

"net"

"net/http"

"reflect"

"runtime"

"strings"

"sync"

"sync/atomic"

"time"

"google.golang.org/grpc/codes"

"google.golang.org/grpc/credentials"

"google.golang.org/grpc/encoding"

"google.golang.org/grpc/encoding/proto"

"google.golang.org/grpc/grpclog"

"google.golang.org/grpc/internal"

"google.golang.org/grpc/internal/binarylog"

"google.golang.org/grpc/internal/channelz"

"google.golang.org/grpc/internal/grpcsync"

"google.golang.org/grpc/internal/grpcutil"

"google.golang.org/grpc/internal/transport"

"google.golang.org/grpc/keepalive"

"google.golang.org/grpc/metadata"

"google.golang.org/grpc/peer"

"google.golang.org/grpc/stats"

"google.golang.org/grpc/status"

"google.golang.org/grpc/tap"

)

const (

defaultServerMaxReceiveMessageSize = 1024 * 1024 * 4

defaultServerMaxSendMessageSize = math.MaxInt32

// Server transports are tracked in a map which is keyed on listener

// address. For regular gRPC traffic, connections are accepted in Serve()

// through a call to Accept(), and we use the actual listener address as key

// when we add it to the map. But for connections received through

// ServeHTTP(), we do not have a listener and hence use this dummy value.

listenerAddressForServeHTTP = "listenerAddressForServeHTTP"

)

func init() {

internal.GetServerCredentials = func(srv *Server) credentials.TransportCredentials {

return srv.opts.creds

}

internal.IsRegisteredMethod = func(srv *Server, method string) bool {

return srv.isRegisteredMethod(method)

}

internal.ServerFromContext = serverFromContext

internal.AddGlobalServerOptions = func(opt ...ServerOption) {

globalServerOptions = append(globalServerOptions, opt...)

}

internal.ClearGlobalServerOptions = func() {

globalServerOptions = nil

}

internal.BinaryLogger = binaryLogger

internal.JoinServerOptions = newJoinServerOption

internal.RecvBufferPool = recvBufferPool

}

var statusOK = status.New(codes.OK, "")

var logger = grpclog.Component("core")

type methodHandler func(srv any, ctx context.Context, dec func(any) error, interceptor UnaryServerInterceptor) (any, error)

// MethodDesc represents an RPC service's method specification.

type MethodDesc struct {

MethodName string

Handler methodHandler

}

// ServiceDesc represents an RPC service's specification.

type ServiceDesc struct {

ServiceName string

// The pointer to the service interface. Used to check whether the user

// provided implementation satisfies the interface requirements.

HandlerType any

Methods []MethodDesc

Streams []StreamDesc

Metadata any

}

// serviceInfo wraps information about a service. It is very similar to

// ServiceDesc and is constructed from it for internal purposes.

type serviceInfo struct {

// Contains the implementation for the methods in this service.

serviceImpl any

methods map[string]*MethodDesc

streams map[string]*StreamDesc

mdata any

}

// 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 contains all active server transports. It is a map keyed on a

// listener address with the value being the set of active transports

// belonging to that listener.

conns map[string]map[transport.ServerTransport]bool

serve bool

drain bool

cv *sync.Cond // signaled when connections close for GracefulStop

services map[string]*serviceInfo // service name -> service info

events traceEventLog

quit *grpcsync.Event

done *grpcsync.Event

channelzRemoveOnce sync.Once

serveWG sync.WaitGroup // counts active Serve goroutines for Stop/GracefulStop

handlersWG sync.WaitGroup // counts active method handler goroutines

channelz *channelz.Server

serverWorkerChannel chan func()

serverWorkerChannelClose func()

}

type serverOptions struct {

creds credentials.TransportCredentials

codec baseCodec

cp Compressor

dc Decompressor

unaryInt UnaryServerInterceptor

streamInt StreamServerInterceptor

chainUnaryInts []UnaryServerInterceptor

chainStreamInts []StreamServerInterceptor

binaryLogger binarylog.Logger

inTapHandle tap.ServerInHandle

statsHandlers []stats.Handler

maxConcurrentStreams uint32

maxReceiveMessageSize int

maxSendMessageSize int

unknownStreamDesc *StreamDesc

keepaliveParams keepalive.ServerParameters

keepalivePolicy keepalive.EnforcementPolicy

initialWindowSize int32

initialConnWindowSize int32

writeBufferSize int

readBufferSize int

sharedWriteBuffer bool

connectionTimeout time.Duration

maxHeaderListSize *uint32

headerTableSize *uint32

numServerWorkers uint32

recvBufferPool SharedBufferPool

waitForHandlers bool

}

var defaultServerOptions = serverOptions{

maxConcurrentStreams: math.MaxUint32,

maxReceiveMessageSize: defaultServerMaxReceiveMessageSize,

maxSendMessageSize: defaultServerMaxSendMessageSize,

connectionTimeout: 120 * time.Second,

writeBufferSize: defaultWriteBufSize,

readBufferSize: defaultReadBufSize,

recvBufferPool: nopBufferPool{},

}

var globalServerOptions []ServerOption

// A ServerOption sets options such as credentials, codec and keepalive parameters, etc.

type ServerOption interface {

apply(*serverOptions)

}

// EmptyServerOption does not alter the server configuration. It can be embedded

// in another structure to build custom server options.

//

// # Experimental

//

// Notice: This type is EXPERIMENTAL and may be changed or removed in a

// later release.

type EmptyServerOption struct{}

func (EmptyServerOption) apply(*serverOptions) {}

// funcServerOption wraps a function that modifies serverOptions into an

// implementation of the ServerOption interface.

type funcServerOption struct {

f func(*serverOptions)

}

func (fdo *funcServerOption) apply(do *serverOptions) {

fdo.f(do)

}

func newFuncServerOption(f func(*serverOptions)) *funcServerOption {

return &funcServerOption{

f: f,

}

}

// joinServerOption provides a way to combine arbitrary number of server

// options into one.

type joinServerOption struct {

opts []ServerOption

}

func (mdo *joinServerOption) apply(do *serverOptions) {

for _, opt := range mdo.opts {

opt.apply(do)

}

}

func newJoinServerOption(opts ...ServerOption) ServerOption {

return &joinServerOption{opts: opts}

}

// SharedWriteBuffer allows reusing per-connection transport write buffer.

// If this option is set to true every connection will release the buffer after

// flushing the data on the wire.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func SharedWriteBuffer(val bool) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.sharedWriteBuffer = val

})

}

// WriteBufferSize determines how much data can be batched before doing a write

// on the wire. The default value for this buffer is 32KB. Zero or negative

// values will disable the write buffer such that each write will be on underlying

// connection. Note: A Send call may not directly translate to a write.

func WriteBufferSize(s int) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.writeBufferSize = s

})

}

// ReadBufferSize lets you set the size of read buffer, this determines how much

// data can be read at most for one read syscall. The default value for this

// buffer is 32KB. Zero or negative values will disable read buffer for a

// connection so data framer can access the underlying conn directly.

func ReadBufferSize(s int) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.readBufferSize = s

})

}

// InitialWindowSize returns a ServerOption that sets window size for stream.

// The lower bound for window size is 64K and any value smaller than that will be ignored.

func InitialWindowSize(s int32) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.initialWindowSize = s

})

}

// InitialConnWindowSize returns a ServerOption that sets window size for a connection.

// The lower bound for window size is 64K and any value smaller than that will be ignored.

func InitialConnWindowSize(s int32) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.initialConnWindowSize = s

})

}

// KeepaliveParams returns a ServerOption that sets keepalive and max-age parameters for the server.

func KeepaliveParams(kp keepalive.ServerParameters) ServerOption {

if kp.Time > 0 && kp.Time < internal.KeepaliveMinServerPingTime {

logger.Warning("Adjusting keepalive ping interval to minimum period of 1s")

kp.Time = internal.KeepaliveMinServerPingTime

}

return newFuncServerOption(func(o *serverOptions) {

o.keepaliveParams = kp

})

}

// KeepaliveEnforcementPolicy returns a ServerOption that sets keepalive enforcement policy for the server.

func KeepaliveEnforcementPolicy(kep keepalive.EnforcementPolicy) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.keepalivePolicy = kep

})

}

// CustomCodec returns a ServerOption that sets a codec for message marshaling and unmarshaling.

//

// This will override any lookups by content-subtype for Codecs registered with RegisterCodec.

//

// Deprecated: register codecs using encoding.RegisterCodec. The server will

// automatically use registered codecs based on the incoming requests' headers.

// See also

// https://github.com/grpc/grpc-go/blob/master/Documentation/encoding.md#using-a-codec.

// Will be supported throughout 1.x.

func CustomCodec(codec Codec) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.codec = codec

})

}

// ForceServerCodec returns a ServerOption that sets a codec for message

// marshaling and unmarshaling.

//

// This will override any lookups by content-subtype for Codecs registered

// with RegisterCodec.

//

// See Content-Type on

// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for

// more details. Also see the documentation on RegisterCodec and

// CallContentSubtype for more details on the interaction between encoding.Codec

// and content-subtype.

//

// This function is provided for advanced users; prefer to register codecs

// using encoding.RegisterCodec.

// The server will automatically use registered codecs based on the incoming

// requests' headers. See also

// https://github.com/grpc/grpc-go/blob/master/Documentation/encoding.md#using-a-codec.

// Will be supported throughout 1.x.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func ForceServerCodec(codec encoding.Codec) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.codec = codec

})

}

// RPCCompressor returns a ServerOption that sets a compressor for outbound

// messages. For backward compatibility, all outbound messages will be sent

// using this compressor, regardless of incoming message compression. By

// default, server messages will be sent using the same compressor with which

// request messages were sent.

//

// Deprecated: use encoding.RegisterCompressor instead. Will be supported

// throughout 1.x.

func RPCCompressor(cp Compressor) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.cp = cp

})

}

// RPCDecompressor returns a ServerOption that sets a decompressor for inbound

// messages. It has higher priority than decompressors registered via

// encoding.RegisterCompressor.

//

// Deprecated: use encoding.RegisterCompressor instead. Will be supported

// throughout 1.x.

func RPCDecompressor(dc Decompressor) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.dc = dc

})

}

// MaxMsgSize returns a ServerOption to set the max message size in bytes the server can receive.

// If this is not set, gRPC uses the default limit.

//

// Deprecated: use MaxRecvMsgSize instead. Will be supported throughout 1.x.

func MaxMsgSize(m int) ServerOption {

return MaxRecvMsgSize(m)

}

// MaxRecvMsgSize returns a ServerOption to set the max message size in bytes the server can receive.

// If this is not set, gRPC uses the default 4MB.

func MaxRecvMsgSize(m int) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.maxReceiveMessageSize = m

})

}

// MaxSendMsgSize returns a ServerOption to set the max message size in bytes the server can send.

// If this is not set, gRPC uses the default `math.MaxInt32`.

func MaxSendMsgSize(m int) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.maxSendMessageSize = m

})

}

// MaxConcurrentStreams returns a ServerOption that will apply a limit on the number

// of concurrent streams to each ServerTransport.

func MaxConcurrentStreams(n uint32) ServerOption {

if n == 0 {

n = math.MaxUint32

}

return newFuncServerOption(func(o *serverOptions) {

o.maxConcurrentStreams = n

})

}

// Creds returns a ServerOption that sets credentials for server connections.

func Creds(c credentials.TransportCredentials) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.creds = c

})

}

// UnaryInterceptor returns a ServerOption that sets the UnaryServerInterceptor for the

// server. Only one unary interceptor can be installed. The construction of multiple

// interceptors (e.g., chaining) can be implemented at the caller.

func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

if o.unaryInt != nil {

panic("The unary server interceptor was already set and may not be reset.")

}

o.unaryInt = i

})

}

// ChainUnaryInterceptor returns a ServerOption that specifies the chained interceptor

// for unary RPCs. The first interceptor will be the outer most,

// while the last interceptor will be the inner most wrapper around the real call.

// All unary interceptors added by this method will be chained.

func ChainUnaryInterceptor(interceptors ...UnaryServerInterceptor) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.chainUnaryInts = append(o.chainUnaryInts, interceptors...)

})

}

// StreamInterceptor returns a ServerOption that sets the StreamServerInterceptor for the

// server. Only one stream interceptor can be installed.

func StreamInterceptor(i StreamServerInterceptor) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

if o.streamInt != nil {

panic("The stream server interceptor was already set and may not be reset.")

}

o.streamInt = i

})

}

// ChainStreamInterceptor returns a ServerOption that specifies the chained interceptor

// for streaming RPCs. The first interceptor will be the outer most,

// while the last interceptor will be the inner most wrapper around the real call.

// All stream interceptors added by this method will be chained.

func ChainStreamInterceptor(interceptors ...StreamServerInterceptor) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.chainStreamInts = append(o.chainStreamInts, interceptors...)

})

}

// InTapHandle returns a ServerOption that sets the tap handle for all the server

// transport to be created. Only one can be installed.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func InTapHandle(h tap.ServerInHandle) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

if o.inTapHandle != nil {

panic("The tap handle was already set and may not be reset.")

}

o.inTapHandle = h

})

}

// StatsHandler returns a ServerOption that sets the stats handler for the server.

func StatsHandler(h stats.Handler) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

if h == nil {

logger.Error("ignoring nil parameter in grpc.StatsHandler ServerOption")

// Do not allow a nil stats handler, which would otherwise cause

// panics.

return

}

o.statsHandlers = append(o.statsHandlers, h)

})

}

// binaryLogger returns a ServerOption that can set the binary logger for the

// server.

func binaryLogger(bl binarylog.Logger) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.binaryLogger = bl

})

}

// UnknownServiceHandler returns a ServerOption that allows for adding a custom

// unknown service handler. The provided method is a bidi-streaming RPC service

// handler that will be invoked instead of returning the "unimplemented" gRPC

// error whenever a request is received for an unregistered service or method.

// The handling function and stream interceptor (if set) have full access to

// the ServerStream, including its Context.

func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.unknownStreamDesc = &StreamDesc{

StreamName: "unknown_service_handler",

Handler: streamHandler,

// We need to assume that the users of the streamHandler will want to use both.

ClientStreams: true,

ServerStreams: true,

}

})

}

// ConnectionTimeout returns a ServerOption that sets the timeout for

// connection establishment (up to and including HTTP/2 handshaking) for all

// new connections. If this is not set, the default is 120 seconds. A zero or

// negative value will result in an immediate timeout.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func ConnectionTimeout(d time.Duration) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.connectionTimeout = d

})

}

// MaxHeaderListSizeServerOption is a ServerOption that sets the max

// (uncompressed) size of header list that the server is prepared to accept.

type MaxHeaderListSizeServerOption struct {

MaxHeaderListSize uint32

}

func (o MaxHeaderListSizeServerOption) apply(so *serverOptions) {

so.maxHeaderListSize = &o.MaxHeaderListSize

}

// MaxHeaderListSize returns a ServerOption that sets the max (uncompressed) size

// of header list that the server is prepared to accept.

func MaxHeaderListSize(s uint32) ServerOption {

return MaxHeaderListSizeServerOption{

MaxHeaderListSize: s,

}

}

// HeaderTableSize returns a ServerOption that sets the size of dynamic

// header table for stream.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func HeaderTableSize(s uint32) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.headerTableSize = &s

})

}

// NumStreamWorkers returns a ServerOption that sets the number of worker

// goroutines that should be used to process incoming streams. Setting this to

// zero (default) will disable workers and spawn a new goroutine for each

// stream.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func NumStreamWorkers(numServerWorkers uint32) ServerOption {

// TODO: If/when this API gets stabilized (i.e. stream workers become the

// only way streams are processed), change the behavior of the zero value to

// a sane default. Preliminary experiments suggest that a value equal to the

// number of CPUs available is most performant; requires thorough testing.

return newFuncServerOption(func(o *serverOptions) {

o.numServerWorkers = numServerWorkers

})

}

// WaitForHandlers cause Stop to wait until all outstanding method handlers have

// exited before returning. If false, Stop will return as soon as all

// connections have closed, but method handlers may still be running. By

// default, Stop does not wait for method handlers to return.

//

// # Experimental

//

// Notice: This API is EXPERIMENTAL and may be changed or removed in a

// later release.

func WaitForHandlers(w bool) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.waitForHandlers = w

})

}

// RecvBufferPool returns a ServerOption that configures the server

// to use the provided shared buffer pool for parsing incoming messages. Depending

// on the application's workload, this could result in reduced memory allocation.

//

// If you are unsure about how to implement a memory pool but want to utilize one,

// begin with grpc.NewSharedBufferPool.

//

// Note: The shared buffer pool feature will not be active if any of the following

// options are used: StatsHandler, EnableTracing, or binary logging. In such

// cases, the shared buffer pool will be ignored.

//

// Deprecated: use experimental.WithRecvBufferPool instead. Will be deleted in

// v1.60.0 or later.

func RecvBufferPool(bufferPool SharedBufferPool) ServerOption {

return recvBufferPool(bufferPool)

}

func recvBufferPool(bufferPool SharedBufferPool) ServerOption {

return newFuncServerOption(func(o *serverOptions) {

o.recvBufferPool = bufferPool

})

}

// serverWorkerResetThreshold defines how often the stack must be reset. Every

// N requests, by spawning a new goroutine in its place, a worker can reset its

// stack so that large stacks don't live in memory forever. 2^16 should allow

// each goroutine stack to live for at least a few seconds in a typical

// workload (assuming a QPS of a few thousand requests/sec).

const serverWorkerResetThreshold = 1 << 16

// serverWorkers blocks on a *transport.Stream channel forever and waits for

// data to be fed by serveStreams. This allows multiple requests to be

// processed by the same goroutine, removing the need for expensive stack

// re-allocations (see the runtime.morestack problem [1]).

//

// [1] https://github.com/golang/go/issues/18138

func (s *Server) serverWorker() {

for completed := 0; completed < serverWorkerResetThreshold; completed++ {

f, ok := <-s.serverWorkerChannel

if !ok {

return

}

f()

}

go s.serverWorker()

}

// initServerWorkers creates worker goroutines and a channel to process incoming

// connections to reduce the time spent overall on runtime.morestack.

func (s *Server) initServerWorkers() {

s.serverWorkerChannel = make(chan func())

s.serverWorkerChannelClose = grpcsync.OnceFunc(func() {

close(s.serverWorkerChannel)

})

for i := uint32(0); i < s.opts.numServerWorkers; i++ {

go s.serverWorker()

}

}

// NewServer creates a gRPC server which has no service registered and has not

// started to accept requests yet.

func NewServer(opt ...ServerOption) *Server {

opts := defaultServerOptions

for _, o := range globalServerOptions {

o.apply(&opts)

}

for _, o := range opt {

o.apply(&opts)

}

s := &Server{

lis: make(map[net.Listener]bool),

opts: opts,

conns: make(map[string]map[transport.ServerTransport]bool),

services: make(map[string]*serviceInfo),

quit: grpcsync.NewEvent(),

done: grpcsync.NewEvent(),

channelz: channelz.RegisterServer(""),

}

chainUnaryServerInterceptors(s)

chainStreamServerInterceptors(s)

s.cv = sync.NewCond(&s.mu)

if EnableTracing {

_, file, line, _ := runtime.Caller(1)

s.events = newTraceEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))

}

if s.opts.numServerWorkers > 0 {

s.initServerWorkers()

}

channelz.Info(logger, s.channelz, "Server created")

return s

}

// printf records an event in s's event log, unless s has been stopped.

// REQUIRES s.mu is held.

func (s *Server) printf(format string, a ...any) {

if s.events != nil {

s.events.Printf(format, a...)

}

}

// errorf records an error in s's event log, unless s has been stopped.

// REQUIRES s.mu is held.

func (s *Server) errorf(format string, a ...any) {

if s.events != nil {

s.events.Errorf(format, a...)

}

}

// ServiceRegistrar wraps a single method that supports service registration. It

// enables users to pass concrete types other than grpc.Server to the service

// registration methods exported by the IDL generated code.

type ServiceRegistrar interface {

// RegisterService registers a service and its implementation to the

// concrete type implementing this interface. It may not be called

// once the server has started serving.

// desc describes the service and its methods and handlers. impl is the

// service implementation which is passed to the method handlers.

RegisterService(desc *ServiceDesc, impl any)

}

// RegisterService registers a service and its implementation to the gRPC

// server. It is called from the IDL generated code. This must be called before

// invoking Serve. If ss is non-nil (for legacy code), its type is checked to

// ensure it implements sd.HandlerType.

func (s *Server) RegisterService(sd *ServiceDesc, ss any) {

if ss != nil {

ht := reflect.TypeOf(sd.HandlerType).Elem()

st := reflect.TypeOf(ss)

if !st.Implements(ht) {

logger.Fatalf("grpc: Server.RegisterService found the handler of type %v that does not satisfy %v", st, ht)

}

}

s.register(sd, ss)

}

func (s *Server) register(sd *ServiceDesc, ss any) {

s.mu.Lock()

defer s.mu.Unlock()

s.printf("RegisterService(%q)", sd.ServiceName)

if s.serve {

logger.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)

}

if _, ok := s.services[sd.ServiceName]; ok {

logger.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)

}

info := &serviceInfo{

serviceImpl: ss,

methods: make(map[string]*MethodDesc),

streams: make(map[string]*StreamDesc),

mdata: sd.Metadata,

}

for i := range sd.Methods {

d := &sd.Methods[i]

info.methods[d.MethodName] = d

}

for i := range sd.Streams {

d := &sd.Streams[i]

info.streams[d.StreamName] = d

}

s.services[sd.ServiceName] = info

}

// MethodInfo contains the information of an RPC including its method name and type.

type MethodInfo struct {

// Name is the method name only, without the service name or package name.

Name string

// IsClientStream indicates whether the RPC is a client streaming RPC.

IsClientStream bool

// IsServerStream indicates whether the RPC is a server streaming RPC.

IsServerStream bool

}

// ServiceInfo contains unary RPC method info, streaming RPC method info and metadata for a service.

type ServiceInfo struct {

Methods []MethodInfo

// Metadata is the metadata specified in ServiceDesc when registering service.

Metadata any

}

// GetServiceInfo returns a map from service names to ServiceInfo.

// Service names include the package names, in the form of <package>.<service>.

func (s *Server) GetServiceInfo() map[string]ServiceInfo {

ret := make(map[string]ServiceInfo)

for n, srv := range s.services {

methods := make([]MethodInfo, 0, len(srv.methods)+len(srv.streams))

for m := range srv.methods {

methods = append(methods, MethodInfo{

Name: m,

IsClientStream: false,

IsServerStream: false,

})

}

for m, d := range srv.streams {

methods = append(methods, MethodInfo{

Name: m,

IsClientStream: d.ClientStreams,

IsServerStream: d.ServerStreams,

})

}

ret[n] = ServiceInfo{

Methods: methods,

Metadata: srv.mdata,

}

}

return ret

}

// ErrServerStopped indicates that the operation is now illegal because of

// the server being stopped.

var ErrServerStopped = errors.New("grpc: the server has been stopped")

type listenSocket struct {

net.Listener

channelz *channelz.Socket

}

func (l *listenSocket) Close() error {

err := l.Listener.Close()

channelz.RemoveEntry(l.channelz.ID)

channelz.Info(logger, l.channelz, "ListenSocket deleted")

return err

}

// 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.

//

// Note: All supported releases of Go (as of December 2023) override the OS

// defaults for TCP keepalive time and interval to 15s. To enable TCP keepalive

// with OS defaults for keepalive time and interval, callers need to do the

// following two things:

// - pass a net.Listener created by calling the Listen method on a

// net.ListenConfig with the `KeepAlive` field set to a negative value. This

// will result in the Go standard library not overriding OS defaults for TCP

// keepalive interval and time. But this will also result in the Go standard

// library not enabling TCP keepalives by default.

// - override the Accept method on the passed in net.Listener and set the

// SO_KEEPALIVE socket option to enable TCP keepalives, with OS defaults.

func (s *Server) Serve(lis net.Listener) error {

s.mu.Lock()

s.printf("serving")

s.serve = true

if s.lis == nil {

// Serve called after Stop or GracefulStop.

s.mu.Unlock()

lis.Close()

return ErrServerStopped

}

s.serveWG.Add(1)

defer func() {

s.serveWG.Done()

if s.quit.HasFired() {

// Stop or GracefulStop called; block until done and return nil.

<-s.done.Done()

}

}()

ls := &listenSocket{

Listener: lis,

channelz: channelz.RegisterSocket(&channelz.Socket{

SocketType: channelz.SocketTypeListen,

Parent: s.channelz,

RefName: lis.Addr().String(),

LocalAddr: lis.Addr(),

SocketOptions: channelz.GetSocketOption(lis)},

),

}

s.lis[ls] = true

defer func() {

s.mu.Lock()

if s.lis != nil && s.lis[ls] {

ls.Close()

delete(s.lis, ls)

}

s.mu.Unlock()

}()

s.mu.Unlock()

channelz.Info(logger, ls.channelz, "ListenSocket created")

var tempDelay time.Duration // how long to sleep on accept failure

for {

rawConn, err := lis.Accept()

if err != nil {

if ne, ok := err.(interface {

Temporary() bool

}); ok && ne.Temporary() {

if tempDelay == 0 {

tempDelay = 5 * time.Millisecond

} else {

tempDelay *= 2

}

if max := 1 * time.Second; tempDelay > max {

tempDelay = max

}

s.mu.Lock()

s.printf("Accept error: %v; retrying in %v", err, tempDelay)

s.mu.Unlock()

timer := time.NewTimer(tempDelay)

select {

case <-timer.C:

case <-s.quit.Done():

timer.Stop()

return nil

}

continue

}

s.mu.Lock()

s.printf("done serving; Accept = %v", err)

s.mu.Unlock()

if s.quit.HasFired() {

return nil

}

return err

}

tempDelay = 0

// Start a new goroutine to deal with rawConn so we don't stall this Accept

// loop goroutine.

//

// Make sure we account for the goroutine so GracefulStop doesn't nil out

// s.conns before this conn can be added.

s.serveWG.Add(1)

go func() {

s.handleRawConn(lis.Addr().String(), rawConn)

s.serveWG.Done()

}()

}

}

// 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(lisAddr string, rawConn net.Conn) {

if s.quit.HasFired() {

rawConn.Close()

return

}

rawConn.SetDeadline(time.Now().Add(s.opts.connectionTimeout))

// Finish handshaking (HTTP2)

st := s.newHTTP2Transport(rawConn)

rawConn.SetDeadline(time.Time{})

if st == nil {

return

}

if cc, ok := rawConn.(interface {

PassServerTransport(transport.ServerTransport)

}); ok {

cc.PassServerTransport(st)

}

if !s.addConn(lisAddr, st) {

return

}

go func() {

s.serveStreams(context.Background(), st, rawConn)

s.removeConn(lisAddr, st)

}()

}

// newHTTP2Transport sets up a http/2 transport (using the

// gRPC http2 server transport in transport/http2_server.go).

func (s *Server) newHTTP2Transport(c net.Conn) transport.ServerTransport {

config := &transport.ServerConfig{

MaxStreams: s.opts.maxConcurrentStreams,

ConnectionTimeout: s.opts.connectionTimeout,

Credentials: s.opts.creds,

InTapHandle: s.opts.inTapHandle,

StatsHandlers: s.opts.statsHandlers,

KeepaliveParams: s.opts.keepaliveParams,

KeepalivePolicy: s.opts.keepalivePolicy,

InitialWindowSize: s.opts.initialWindowSize,

InitialConnWindowSize: s.opts.initialConnWindowSize,

WriteBufferSize: s.opts.writeBufferSize,

ReadBufferSize: s.opts.readBufferSize,

SharedWriteBuffer: s.opts.sharedWriteBuffer,

ChannelzParent: s.channelz,

MaxHeaderListSize: s.opts.maxHeaderListSize,

HeaderTableSize: s.opts.headerTableSize,

}

st, err := transport.NewServerTransport(c, config)

if err != nil {

s.mu.Lock()

s.errorf("NewServerTransport(%q) failed: %v", c.RemoteAddr(), err)

s.mu.Unlock()

// ErrConnDispatched means that the connection was dispatched away from

// gRPC; those connections should be left open.

if err != credentials.ErrConnDispatched {

// Don't log on ErrConnDispatched and io.EOF to prevent log spam.

if err != io.EOF {

channelz.Info(logger, s.channelz, "grpc: Server.Serve failed to create ServerTransport: ", err)

}

c.Close()

}

return nil

}