12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502 |
- package grpc
- import (
- "bytes"
- "errors"
- "fmt"
- "io"
- "math"
- "net"
- "net/http"
- "reflect"
- "runtime"
- "strings"
- "sync"
- "time"
- "io/ioutil"
- "golang.org/x/net/context"
- "golang.org/x/net/http2"
- "golang.org/x/net/trace"
- "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/channelz"
- "google.golang.org/grpc/internal/transport"
- "google.golang.org/grpc/keepalive"
- "google.golang.org/grpc/metadata"
- "google.golang.org/grpc/stats"
- "google.golang.org/grpc/status"
- "google.golang.org/grpc/tap"
- )
- const (
- defaultServerMaxReceiveMessageSize = 1024 * 1024 * 4
- defaultServerMaxSendMessageSize = math.MaxInt32
- )
- type methodHandler func(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor UnaryServerInterceptor) (interface{}, error)
- type MethodDesc struct {
- MethodName string
- Handler methodHandler
- }
- type ServiceDesc struct {
- ServiceName string
-
-
- HandlerType interface{}
- Methods []MethodDesc
- Streams []StreamDesc
- Metadata interface{}
- }
- type service struct {
- server interface{}
- md map[string]*MethodDesc
- sd map[string]*StreamDesc
- mdata interface{}
- }
- type Server struct {
- opts options
- mu sync.Mutex
- lis map[net.Listener]bool
- conns map[io.Closer]bool
- serve bool
- drain bool
- cv *sync.Cond
- m map[string]*service
- events trace.EventLog
- quit chan struct{}
- done chan struct{}
- quitOnce sync.Once
- doneOnce sync.Once
- channelzRemoveOnce sync.Once
- serveWG sync.WaitGroup
- channelzID int64
- czmu sync.RWMutex
- callsStarted int64
- callsFailed int64
- callsSucceeded int64
- lastCallStartedTime time.Time
- }
- type options struct {
- creds credentials.TransportCredentials
- codec baseCodec
- cp Compressor
- dc Decompressor
- unaryInt UnaryServerInterceptor
- streamInt StreamServerInterceptor
- inTapHandle tap.ServerInHandle
- statsHandler stats.Handler
- maxConcurrentStreams uint32
- maxReceiveMessageSize int
- maxSendMessageSize int
- useHandlerImpl bool
- unknownStreamDesc *StreamDesc
- keepaliveParams keepalive.ServerParameters
- keepalivePolicy keepalive.EnforcementPolicy
- initialWindowSize int32
- initialConnWindowSize int32
- writeBufferSize int
- readBufferSize int
- connectionTimeout time.Duration
- maxHeaderListSize *uint32
- }
- var defaultServerOptions = options{
- maxReceiveMessageSize: defaultServerMaxReceiveMessageSize,
- maxSendMessageSize: defaultServerMaxSendMessageSize,
- connectionTimeout: 120 * time.Second,
- writeBufferSize: defaultWriteBufSize,
- readBufferSize: defaultReadBufSize,
- }
- type ServerOption func(*options)
- func WriteBufferSize(s int) ServerOption {
- return func(o *options) {
- o.writeBufferSize = s
- }
- }
- func ReadBufferSize(s int) ServerOption {
- return func(o *options) {
- o.readBufferSize = s
- }
- }
- func InitialWindowSize(s int32) ServerOption {
- return func(o *options) {
- o.initialWindowSize = s
- }
- }
- func InitialConnWindowSize(s int32) ServerOption {
- return func(o *options) {
- o.initialConnWindowSize = s
- }
- }
- func KeepaliveParams(kp keepalive.ServerParameters) ServerOption {
- return func(o *options) {
- o.keepaliveParams = kp
- }
- }
- func KeepaliveEnforcementPolicy(kep keepalive.EnforcementPolicy) ServerOption {
- return func(o *options) {
- o.keepalivePolicy = kep
- }
- }
- func CustomCodec(codec Codec) ServerOption {
- return func(o *options) {
- o.codec = codec
- }
- }
- func RPCCompressor(cp Compressor) ServerOption {
- return func(o *options) {
- o.cp = cp
- }
- }
- func RPCDecompressor(dc Decompressor) ServerOption {
- return func(o *options) {
- o.dc = dc
- }
- }
- func MaxMsgSize(m int) ServerOption {
- return MaxRecvMsgSize(m)
- }
- func MaxRecvMsgSize(m int) ServerOption {
- return func(o *options) {
- o.maxReceiveMessageSize = m
- }
- }
- func MaxSendMsgSize(m int) ServerOption {
- return func(o *options) {
- o.maxSendMessageSize = m
- }
- }
- func MaxConcurrentStreams(n uint32) ServerOption {
- return func(o *options) {
- o.maxConcurrentStreams = n
- }
- }
- func Creds(c credentials.TransportCredentials) ServerOption {
- return func(o *options) {
- o.creds = c
- }
- }
- func UnaryInterceptor(i UnaryServerInterceptor) ServerOption {
- return func(o *options) {
- if o.unaryInt != nil {
- panic("The unary server interceptor was already set and may not be reset.")
- }
- o.unaryInt = i
- }
- }
- func StreamInterceptor(i StreamServerInterceptor) ServerOption {
- return func(o *options) {
- if o.streamInt != nil {
- panic("The stream server interceptor was already set and may not be reset.")
- }
- o.streamInt = i
- }
- }
- func InTapHandle(h tap.ServerInHandle) ServerOption {
- return func(o *options) {
- if o.inTapHandle != nil {
- panic("The tap handle was already set and may not be reset.")
- }
- o.inTapHandle = h
- }
- }
- func StatsHandler(h stats.Handler) ServerOption {
- return func(o *options) {
- o.statsHandler = h
- }
- }
- func UnknownServiceHandler(streamHandler StreamHandler) ServerOption {
- return func(o *options) {
- o.unknownStreamDesc = &StreamDesc{
- StreamName: "unknown_service_handler",
- Handler: streamHandler,
-
- ClientStreams: true,
- ServerStreams: true,
- }
- }
- }
- func ConnectionTimeout(d time.Duration) ServerOption {
- return func(o *options) {
- o.connectionTimeout = d
- }
- }
- func MaxHeaderListSize(s uint32) ServerOption {
- return func(o *options) {
- o.maxHeaderListSize = &s
- }
- }
- func NewServer(opt ...ServerOption) *Server {
- opts := defaultServerOptions
- for _, o := range opt {
- o(&opts)
- }
- s := &Server{
- lis: make(map[net.Listener]bool),
- opts: opts,
- conns: make(map[io.Closer]bool),
- m: make(map[string]*service),
- quit: make(chan struct{}),
- done: make(chan struct{}),
- }
- s.cv = sync.NewCond(&s.mu)
- if EnableTracing {
- _, file, line, _ := runtime.Caller(1)
- s.events = trace.NewEventLog("grpc.Server", fmt.Sprintf("%s:%d", file, line))
- }
- if channelz.IsOn() {
- s.channelzID = channelz.RegisterServer(s, "")
- }
- return s
- }
- func (s *Server) printf(format string, a ...interface{}) {
- if s.events != nil {
- s.events.Printf(format, a...)
- }
- }
- func (s *Server) errorf(format string, a ...interface{}) {
- if s.events != nil {
- s.events.Errorf(format, a...)
- }
- }
- func (s *Server) RegisterService(sd *ServiceDesc, ss interface{}) {
- ht := reflect.TypeOf(sd.HandlerType).Elem()
- st := reflect.TypeOf(ss)
- if !st.Implements(ht) {
- grpclog.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 interface{}) {
- s.mu.Lock()
- defer s.mu.Unlock()
- s.printf("RegisterService(%q)", sd.ServiceName)
- if s.serve {
- grpclog.Fatalf("grpc: Server.RegisterService after Server.Serve for %q", sd.ServiceName)
- }
- if _, ok := s.m[sd.ServiceName]; ok {
- grpclog.Fatalf("grpc: Server.RegisterService found duplicate service registration for %q", sd.ServiceName)
- }
- srv := &service{
- server: ss,
- md: make(map[string]*MethodDesc),
- sd: make(map[string]*StreamDesc),
- mdata: sd.Metadata,
- }
- for i := range sd.Methods {
- d := &sd.Methods[i]
- srv.md[d.MethodName] = d
- }
- for i := range sd.Streams {
- d := &sd.Streams[i]
- srv.sd[d.StreamName] = d
- }
- s.m[sd.ServiceName] = srv
- }
- type MethodInfo struct {
-
- Name string
-
- IsClientStream bool
-
- IsServerStream bool
- }
- type ServiceInfo struct {
- Methods []MethodInfo
-
- Metadata interface{}
- }
- func (s *Server) GetServiceInfo() map[string]ServiceInfo {
- ret := make(map[string]ServiceInfo)
- for n, srv := range s.m {
- methods := make([]MethodInfo, 0, len(srv.md)+len(srv.sd))
- for m := range srv.md {
- methods = append(methods, MethodInfo{
- Name: m,
- IsClientStream: false,
- IsServerStream: false,
- })
- }
- for m, d := range srv.sd {
- methods = append(methods, MethodInfo{
- Name: m,
- IsClientStream: d.ClientStreams,
- IsServerStream: d.ServerStreams,
- })
- }
- ret[n] = ServiceInfo{
- Methods: methods,
- Metadata: srv.mdata,
- }
- }
- return ret
- }
- var ErrServerStopped = errors.New("grpc: the server has been stopped")
- func (s *Server) useTransportAuthenticator(rawConn net.Conn) (net.Conn, credentials.AuthInfo, error) {
- if s.opts.creds == nil {
- return rawConn, nil, nil
- }
- return s.opts.creds.ServerHandshake(rawConn)
- }
- type listenSocket struct {
- net.Listener
- channelzID int64
- }
- func (l *listenSocket) ChannelzMetric() *channelz.SocketInternalMetric {
- return &channelz.SocketInternalMetric{
- SocketOptions: channelz.GetSocketOption(l.Listener),
- LocalAddr: l.Listener.Addr(),
- }
- }
- func (l *listenSocket) Close() error {
- err := l.Listener.Close()
- if channelz.IsOn() {
- channelz.RemoveEntry(l.channelzID)
- }
- return err
- }
- func (s *Server) Serve(lis net.Listener) error {
- s.mu.Lock()
- s.printf("serving")
- s.serve = true
- if s.lis == nil {
-
- s.mu.Unlock()
- lis.Close()
- return ErrServerStopped
- }
- s.serveWG.Add(1)
- defer func() {
- s.serveWG.Done()
- select {
-
- case <-s.quit:
- <-s.done
- default:
- }
- }()
- ls := &listenSocket{Listener: lis}
- s.lis[ls] = true
- if channelz.IsOn() {
- ls.channelzID = channelz.RegisterListenSocket(ls, s.channelzID, "")
- }
- s.mu.Unlock()
- defer func() {
- s.mu.Lock()
- if s.lis != nil && s.lis[ls] {
- ls.Close()
- delete(s.lis, ls)
- }
- s.mu.Unlock()
- }()
- var tempDelay time.Duration
- 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:
- timer.Stop()
- return nil
- }
- continue
- }
- s.mu.Lock()
- s.printf("done serving; Accept = %v", err)
- s.mu.Unlock()
- select {
- case <-s.quit:
- return nil
- default:
- }
- return err
- }
- tempDelay = 0
-
-
-
-
-
- s.serveWG.Add(1)
- go func() {
- s.handleRawConn(rawConn)
- s.serveWG.Done()
- }()
- }
- }
- func (s *Server) handleRawConn(rawConn net.Conn) {
- rawConn.SetDeadline(time.Now().Add(s.opts.connectionTimeout))
- conn, authInfo, err := s.useTransportAuthenticator(rawConn)
- if err != nil {
- s.mu.Lock()
- s.errorf("ServerHandshake(%q) failed: %v", rawConn.RemoteAddr(), err)
- s.mu.Unlock()
- grpclog.Warningf("grpc: Server.Serve failed to complete security handshake from %q: %v", rawConn.RemoteAddr(), err)
-
- if err != credentials.ErrConnDispatched {
- rawConn.Close()
- }
- rawConn.SetDeadline(time.Time{})
- return
- }
- s.mu.Lock()
- if s.conns == nil {
- s.mu.Unlock()
- conn.Close()
- return
- }
- s.mu.Unlock()
- var serve func()
- c := conn.(io.Closer)
- if s.opts.useHandlerImpl {
- serve = func() { s.serveUsingHandler(conn) }
- } else {
-
- st := s.newHTTP2Transport(conn, authInfo)
- if st == nil {
- return
- }
- c = st
- serve = func() { s.serveStreams(st) }
- }
- rawConn.SetDeadline(time.Time{})
- if !s.addConn(c) {
- return
- }
- go func() {
- serve()
- s.removeConn(c)
- }()
- }
- func (s *Server) newHTTP2Transport(c net.Conn, authInfo credentials.AuthInfo) transport.ServerTransport {
- config := &transport.ServerConfig{
- MaxStreams: s.opts.maxConcurrentStreams,
- AuthInfo: authInfo,
- InTapHandle: s.opts.inTapHandle,
- StatsHandler: s.opts.statsHandler,
- KeepaliveParams: s.opts.keepaliveParams,
- KeepalivePolicy: s.opts.keepalivePolicy,
- InitialWindowSize: s.opts.initialWindowSize,
- InitialConnWindowSize: s.opts.initialConnWindowSize,
- WriteBufferSize: s.opts.writeBufferSize,
- ReadBufferSize: s.opts.readBufferSize,
- ChannelzParentID: s.channelzID,
- MaxHeaderListSize: s.opts.maxHeaderListSize,
- }
- st, err := transport.NewServerTransport("http2", c, config)
- if err != nil {
- s.mu.Lock()
- s.errorf("NewServerTransport(%q) failed: %v", c.RemoteAddr(), err)
- s.mu.Unlock()
- c.Close()
- grpclog.Warningln("grpc: Server.Serve failed to create ServerTransport: ", err)
- return nil
- }
- return st
- }
- func (s *Server) serveStreams(st transport.ServerTransport) {
- defer st.Close()
- var wg sync.WaitGroup
- st.HandleStreams(func(stream *transport.Stream) {
- wg.Add(1)
- go func() {
- defer wg.Done()
- s.handleStream(st, stream, s.traceInfo(st, stream))
- }()
- }, func(ctx context.Context, method string) context.Context {
- if !EnableTracing {
- return ctx
- }
- tr := trace.New("grpc.Recv."+methodFamily(method), method)
- return trace.NewContext(ctx, tr)
- })
- wg.Wait()
- }
- var _ http.Handler = (*Server)(nil)
- func (s *Server) serveUsingHandler(conn net.Conn) {
- h2s := &http2.Server{
- MaxConcurrentStreams: s.opts.maxConcurrentStreams,
- }
- h2s.ServeConn(conn, &http2.ServeConnOpts{
- Handler: s,
- })
- }
- func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) {
- st, err := transport.NewServerHandlerTransport(w, r, s.opts.statsHandler)
- if err != nil {
- http.Error(w, err.Error(), http.StatusInternalServerError)
- return
- }
- if !s.addConn(st) {
- return
- }
- defer s.removeConn(st)
- s.serveStreams(st)
- }
- func (s *Server) traceInfo(st transport.ServerTransport, stream *transport.Stream) (trInfo *traceInfo) {
- tr, ok := trace.FromContext(stream.Context())
- if !ok {
- return nil
- }
- trInfo = &traceInfo{
- tr: tr,
- }
- trInfo.firstLine.client = false
- trInfo.firstLine.remoteAddr = st.RemoteAddr()
- if dl, ok := stream.Context().Deadline(); ok {
- trInfo.firstLine.deadline = dl.Sub(time.Now())
- }
- return trInfo
- }
- func (s *Server) addConn(c io.Closer) bool {
- s.mu.Lock()
- defer s.mu.Unlock()
- if s.conns == nil {
- c.Close()
- return false
- }
- if s.drain {
-
-
- c.(transport.ServerTransport).Drain()
- }
- s.conns[c] = true
- return true
- }
- func (s *Server) removeConn(c io.Closer) {
- s.mu.Lock()
- defer s.mu.Unlock()
- if s.conns != nil {
- delete(s.conns, c)
- s.cv.Broadcast()
- }
- }
- func (s *Server) ChannelzMetric() *channelz.ServerInternalMetric {
- s.czmu.RLock()
- defer s.czmu.RUnlock()
- return &channelz.ServerInternalMetric{
- CallsStarted: s.callsStarted,
- CallsSucceeded: s.callsSucceeded,
- CallsFailed: s.callsFailed,
- LastCallStartedTimestamp: s.lastCallStartedTime,
- }
- }
- func (s *Server) incrCallsStarted() {
- s.czmu.Lock()
- s.callsStarted++
- s.lastCallStartedTime = time.Now()
- s.czmu.Unlock()
- }
- func (s *Server) incrCallsSucceeded() {
- s.czmu.Lock()
- s.callsSucceeded++
- s.czmu.Unlock()
- }
- func (s *Server) incrCallsFailed() {
- s.czmu.Lock()
- s.callsFailed++
- s.czmu.Unlock()
- }
- func (s *Server) sendResponse(t transport.ServerTransport, stream *transport.Stream, msg interface{}, cp Compressor, opts *transport.Options, comp encoding.Compressor) error {
- data, err := encode(s.getCodec(stream.ContentSubtype()), msg)
- if err != nil {
- grpclog.Errorln("grpc: server failed to encode response: ", err)
- return err
- }
- compData, err := compress(data, cp, comp)
- if err != nil {
- grpclog.Errorln("grpc: server failed to compress response: ", err)
- return err
- }
- hdr, payload := msgHeader(data, compData)
-
- if len(payload) > s.opts.maxSendMessageSize {
- return status.Errorf(codes.ResourceExhausted, "grpc: trying to send message larger than max (%d vs. %d)", len(payload), s.opts.maxSendMessageSize)
- }
- err = t.Write(stream, hdr, payload, opts)
- if err == nil && s.opts.statsHandler != nil {
- s.opts.statsHandler.HandleRPC(stream.Context(), outPayload(false, msg, data, payload, time.Now()))
- }
- return err
- }
- func (s *Server) processUnaryRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, md *MethodDesc, trInfo *traceInfo) (err error) {
- if channelz.IsOn() {
- s.incrCallsStarted()
- defer func() {
- if err != nil && err != io.EOF {
- s.incrCallsFailed()
- } else {
- s.incrCallsSucceeded()
- }
- }()
- }
- sh := s.opts.statsHandler
- if sh != nil {
- beginTime := time.Now()
- begin := &stats.Begin{
- BeginTime: beginTime,
- }
- sh.HandleRPC(stream.Context(), begin)
- defer func() {
- end := &stats.End{
- BeginTime: beginTime,
- EndTime: time.Now(),
- }
- if err != nil && err != io.EOF {
- end.Error = toRPCErr(err)
- }
- sh.HandleRPC(stream.Context(), end)
- }()
- }
- if trInfo != nil {
- defer trInfo.tr.Finish()
- trInfo.firstLine.client = false
- trInfo.tr.LazyLog(&trInfo.firstLine, false)
- defer func() {
- if err != nil && err != io.EOF {
- trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
- trInfo.tr.SetError()
- }
- }()
- }
-
-
-
-
- var comp, decomp encoding.Compressor
- var cp Compressor
- var dc Decompressor
-
-
- if rc := stream.RecvCompress(); s.opts.dc != nil && s.opts.dc.Type() == rc {
- dc = s.opts.dc
- } else if rc != "" && rc != encoding.Identity {
- decomp = encoding.GetCompressor(rc)
- if decomp == nil {
- st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
- t.WriteStatus(stream, st)
- return st.Err()
- }
- }
-
-
-
-
- if s.opts.cp != nil {
- cp = s.opts.cp
- stream.SetSendCompress(cp.Type())
- } else if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
-
- comp = encoding.GetCompressor(rc)
- if comp != nil {
- stream.SetSendCompress(rc)
- }
- }
- p := &parser{r: stream}
- pf, req, err := p.recvMsg(s.opts.maxReceiveMessageSize)
- if err == io.EOF {
-
- return err
- }
- if err == io.ErrUnexpectedEOF {
- err = status.Errorf(codes.Internal, io.ErrUnexpectedEOF.Error())
- }
- if err != nil {
- if st, ok := status.FromError(err); ok {
- if e := t.WriteStatus(stream, st); e != nil {
- grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status %v", e)
- }
- } else {
- switch st := err.(type) {
- case transport.ConnectionError:
-
- case transport.StreamError:
- if e := t.WriteStatus(stream, status.New(st.Code, st.Desc)); e != nil {
- grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status %v", e)
- }
- default:
- panic(fmt.Sprintf("grpc: Unexpected error (%T) from recvMsg: %v", st, st))
- }
- }
- return err
- }
- if channelz.IsOn() {
- t.IncrMsgRecv()
- }
- if st := checkRecvPayload(pf, stream.RecvCompress(), dc != nil || decomp != nil); st != nil {
- if e := t.WriteStatus(stream, st); e != nil {
- grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status %v", e)
- }
- return st.Err()
- }
- var inPayload *stats.InPayload
- if sh != nil {
- inPayload = &stats.InPayload{
- RecvTime: time.Now(),
- }
- }
- df := func(v interface{}) error {
- if inPayload != nil {
- inPayload.WireLength = len(req)
- }
- if pf == compressionMade {
- var err error
- if dc != nil {
- req, err = dc.Do(bytes.NewReader(req))
- if err != nil {
- return status.Errorf(codes.Internal, err.Error())
- }
- } else {
- tmp, _ := decomp.Decompress(bytes.NewReader(req))
- req, err = ioutil.ReadAll(tmp)
- if err != nil {
- return status.Errorf(codes.Internal, "grpc: failed to decompress the received message %v", err)
- }
- }
- }
- if len(req) > s.opts.maxReceiveMessageSize {
-
-
- return status.Errorf(codes.ResourceExhausted, "grpc: received message larger than max (%d vs. %d)", len(req), s.opts.maxReceiveMessageSize)
- }
- if err := s.getCodec(stream.ContentSubtype()).Unmarshal(req, v); err != nil {
- return status.Errorf(codes.Internal, "grpc: error unmarshalling request: %v", err)
- }
- if inPayload != nil {
- inPayload.Payload = v
- inPayload.Data = req
- inPayload.Length = len(req)
- sh.HandleRPC(stream.Context(), inPayload)
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(&payload{sent: false, msg: v}, true)
- }
- return nil
- }
- ctx := NewContextWithServerTransportStream(stream.Context(), stream)
- reply, appErr := md.Handler(srv.server, ctx, df, s.opts.unaryInt)
- if appErr != nil {
- appStatus, ok := status.FromError(appErr)
- if !ok {
-
- appErr = status.Error(codes.Unknown, appErr.Error())
- appStatus, _ = status.FromError(appErr)
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
- trInfo.tr.SetError()
- }
- if e := t.WriteStatus(stream, appStatus); e != nil {
- grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
- }
- return appErr
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(stringer("OK"), false)
- }
- opts := &transport.Options{Last: true}
- if err := s.sendResponse(t, stream, reply, cp, opts, comp); err != nil {
- if err == io.EOF {
-
- return err
- }
- if s, ok := status.FromError(err); ok {
- if e := t.WriteStatus(stream, s); e != nil {
- grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status: %v", e)
- }
- } else {
- switch st := err.(type) {
- case transport.ConnectionError:
-
- case transport.StreamError:
- if e := t.WriteStatus(stream, status.New(st.Code, st.Desc)); e != nil {
- grpclog.Warningf("grpc: Server.processUnaryRPC failed to write status %v", e)
- }
- default:
- panic(fmt.Sprintf("grpc: Unexpected error (%T) from sendResponse: %v", st, st))
- }
- }
- return err
- }
- if channelz.IsOn() {
- t.IncrMsgSent()
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(&payload{sent: true, msg: reply}, true)
- }
-
-
-
- return t.WriteStatus(stream, status.New(codes.OK, ""))
- }
- func (s *Server) processStreamingRPC(t transport.ServerTransport, stream *transport.Stream, srv *service, sd *StreamDesc, trInfo *traceInfo) (err error) {
- if channelz.IsOn() {
- s.incrCallsStarted()
- defer func() {
- if err != nil && err != io.EOF {
- s.incrCallsFailed()
- } else {
- s.incrCallsSucceeded()
- }
- }()
- }
- sh := s.opts.statsHandler
- if sh != nil {
- beginTime := time.Now()
- begin := &stats.Begin{
- BeginTime: beginTime,
- }
- sh.HandleRPC(stream.Context(), begin)
- defer func() {
- end := &stats.End{
- BeginTime: beginTime,
- EndTime: time.Now(),
- }
- if err != nil && err != io.EOF {
- end.Error = toRPCErr(err)
- }
- sh.HandleRPC(stream.Context(), end)
- }()
- }
- ctx := NewContextWithServerTransportStream(stream.Context(), stream)
- ss := &serverStream{
- ctx: ctx,
- t: t,
- s: stream,
- p: &parser{r: stream},
- codec: s.getCodec(stream.ContentSubtype()),
- maxReceiveMessageSize: s.opts.maxReceiveMessageSize,
- maxSendMessageSize: s.opts.maxSendMessageSize,
- trInfo: trInfo,
- statsHandler: sh,
- }
-
-
- if rc := stream.RecvCompress(); s.opts.dc != nil && s.opts.dc.Type() == rc {
- ss.dc = s.opts.dc
- } else if rc != "" && rc != encoding.Identity {
- ss.decomp = encoding.GetCompressor(rc)
- if ss.decomp == nil {
- st := status.Newf(codes.Unimplemented, "grpc: Decompressor is not installed for grpc-encoding %q", rc)
- t.WriteStatus(ss.s, st)
- return st.Err()
- }
- }
-
-
-
-
- if s.opts.cp != nil {
- ss.cp = s.opts.cp
- stream.SetSendCompress(s.opts.cp.Type())
- } else if rc := stream.RecvCompress(); rc != "" && rc != encoding.Identity {
-
- ss.comp = encoding.GetCompressor(rc)
- if ss.comp != nil {
- stream.SetSendCompress(rc)
- }
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(&trInfo.firstLine, false)
- defer func() {
- ss.mu.Lock()
- if err != nil && err != io.EOF {
- ss.trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
- ss.trInfo.tr.SetError()
- }
- ss.trInfo.tr.Finish()
- ss.trInfo.tr = nil
- ss.mu.Unlock()
- }()
- }
- var appErr error
- var server interface{}
- if srv != nil {
- server = srv.server
- }
- if s.opts.streamInt == nil {
- appErr = sd.Handler(server, ss)
- } else {
- info := &StreamServerInfo{
- FullMethod: stream.Method(),
- IsClientStream: sd.ClientStreams,
- IsServerStream: sd.ServerStreams,
- }
- appErr = s.opts.streamInt(server, ss, info, sd.Handler)
- }
- if appErr != nil {
- appStatus, ok := status.FromError(appErr)
- if !ok {
- switch err := appErr.(type) {
- case transport.StreamError:
- appStatus = status.New(err.Code, err.Desc)
- default:
- appStatus = status.New(codes.Unknown, appErr.Error())
- }
- appErr = appStatus.Err()
- }
- if trInfo != nil {
- ss.mu.Lock()
- ss.trInfo.tr.LazyLog(stringer(appStatus.Message()), true)
- ss.trInfo.tr.SetError()
- ss.mu.Unlock()
- }
- t.WriteStatus(ss.s, appStatus)
-
- return appErr
- }
- if trInfo != nil {
- ss.mu.Lock()
- ss.trInfo.tr.LazyLog(stringer("OK"), false)
- ss.mu.Unlock()
- }
- return t.WriteStatus(ss.s, status.New(codes.OK, ""))
- }
- func (s *Server) handleStream(t transport.ServerTransport, stream *transport.Stream, trInfo *traceInfo) {
- sm := stream.Method()
- if sm != "" && sm[0] == '/' {
- sm = sm[1:]
- }
- pos := strings.LastIndex(sm, "/")
- if pos == -1 {
- if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"Malformed method name %q", []interface{}{sm}}, true)
- trInfo.tr.SetError()
- }
- errDesc := fmt.Sprintf("malformed method name: %q", stream.Method())
- if err := t.WriteStatus(stream, status.New(codes.ResourceExhausted, errDesc)); err != nil {
- if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
- trInfo.tr.SetError()
- }
- grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
- }
- if trInfo != nil {
- trInfo.tr.Finish()
- }
- return
- }
- service := sm[:pos]
- method := sm[pos+1:]
- srv, ok := s.m[service]
- if !ok {
- if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
- s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
- return
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"Unknown service %v", []interface{}{service}}, true)
- trInfo.tr.SetError()
- }
- errDesc := fmt.Sprintf("unknown service %v", service)
- if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
- if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
- trInfo.tr.SetError()
- }
- grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
- }
- if trInfo != nil {
- trInfo.tr.Finish()
- }
- return
- }
-
- if md, ok := srv.md[method]; ok {
- s.processUnaryRPC(t, stream, srv, md, trInfo)
- return
- }
- if sd, ok := srv.sd[method]; ok {
- s.processStreamingRPC(t, stream, srv, sd, trInfo)
- return
- }
- if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"Unknown method %v", []interface{}{method}}, true)
- trInfo.tr.SetError()
- }
- if unknownDesc := s.opts.unknownStreamDesc; unknownDesc != nil {
- s.processStreamingRPC(t, stream, nil, unknownDesc, trInfo)
- return
- }
- errDesc := fmt.Sprintf("unknown method %v", method)
- if err := t.WriteStatus(stream, status.New(codes.Unimplemented, errDesc)); err != nil {
- if trInfo != nil {
- trInfo.tr.LazyLog(&fmtStringer{"%v", []interface{}{err}}, true)
- trInfo.tr.SetError()
- }
- grpclog.Warningf("grpc: Server.handleStream failed to write status: %v", err)
- }
- if trInfo != nil {
- trInfo.tr.Finish()
- }
- }
- type streamKey struct{}
- func NewContextWithServerTransportStream(ctx context.Context, stream ServerTransportStream) context.Context {
- return context.WithValue(ctx, streamKey{}, stream)
- }
- type ServerTransportStream interface {
- Method() string
- SetHeader(md metadata.MD) error
- SendHeader(md metadata.MD) error
- SetTrailer(md metadata.MD) error
- }
- func ServerTransportStreamFromContext(ctx context.Context) ServerTransportStream {
- s, _ := ctx.Value(streamKey{}).(ServerTransportStream)
- return s
- }
- func (s *Server) Stop() {
- s.quitOnce.Do(func() {
- close(s.quit)
- })
- defer func() {
- s.serveWG.Wait()
- s.doneOnce.Do(func() {
- close(s.done)
- })
- }()
- s.channelzRemoveOnce.Do(func() {
- if channelz.IsOn() {
- channelz.RemoveEntry(s.channelzID)
- }
- })
- s.mu.Lock()
- listeners := s.lis
- s.lis = nil
- st := s.conns
- s.conns = nil
-
- s.cv.Broadcast()
- s.mu.Unlock()
- for lis := range listeners {
- lis.Close()
- }
- for c := range st {
- c.Close()
- }
- s.mu.Lock()
- if s.events != nil {
- s.events.Finish()
- s.events = nil
- }
- s.mu.Unlock()
- }
- func (s *Server) GracefulStop() {
- s.quitOnce.Do(func() {
- close(s.quit)
- })
- defer func() {
- s.doneOnce.Do(func() {
- close(s.done)
- })
- }()
- s.channelzRemoveOnce.Do(func() {
- if channelz.IsOn() {
- channelz.RemoveEntry(s.channelzID)
- }
- })
- s.mu.Lock()
- if s.conns == nil {
- s.mu.Unlock()
- return
- }
- for lis := range s.lis {
- lis.Close()
- }
- s.lis = nil
- if !s.drain {
- for c := range s.conns {
- c.(transport.ServerTransport).Drain()
- }
- s.drain = true
- }
-
-
- s.mu.Unlock()
- s.serveWG.Wait()
- s.mu.Lock()
- for len(s.conns) != 0 {
- s.cv.Wait()
- }
- s.conns = nil
- if s.events != nil {
- s.events.Finish()
- s.events = nil
- }
- s.mu.Unlock()
- }
- func init() {
- internal.TestingUseHandlerImpl = func(arg interface{}) {
- arg.(*Server).opts.useHandlerImpl = true
- }
- }
- func (s *Server) getCodec(contentSubtype string) baseCodec {
- if s.opts.codec != nil {
- return s.opts.codec
- }
- if contentSubtype == "" {
- return encoding.GetCodec(proto.Name)
- }
- codec := encoding.GetCodec(contentSubtype)
- if codec == nil {
- return encoding.GetCodec(proto.Name)
- }
- return codec
- }
- func SetHeader(ctx context.Context, md metadata.MD) error {
- if md.Len() == 0 {
- return nil
- }
- stream := ServerTransportStreamFromContext(ctx)
- if stream == nil {
- return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
- }
- return stream.SetHeader(md)
- }
- func SendHeader(ctx context.Context, md metadata.MD) error {
- stream := ServerTransportStreamFromContext(ctx)
- if stream == nil {
- return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
- }
- if err := stream.SendHeader(md); err != nil {
- return toRPCErr(err)
- }
- return nil
- }
- func SetTrailer(ctx context.Context, md metadata.MD) error {
- if md.Len() == 0 {
- return nil
- }
- stream := ServerTransportStreamFromContext(ctx)
- if stream == nil {
- return status.Errorf(codes.Internal, "grpc: failed to fetch the stream from the context %v", ctx)
- }
- return stream.SetTrailer(md)
- }
- func Method(ctx context.Context) (string, bool) {
- s := ServerTransportStreamFromContext(ctx)
- if s == nil {
- return "", false
- }
- return s.Method(), true
- }
|