vendor/google.golang.org/grpc/transport/http2_server.go

   1 /*
   2  *
   3  * Copyright 2014 gRPC authors.
   4  *
   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
   8  *
   9  *     http://www.apache.org/licenses/LICENSE-2.0
  10  *
  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.
  16  *
  17  */
  18
  19 package transport
  20
  21 import (
  22         "bytes"
  23         "errors"
  24         "fmt"
  25         "io"
  26         "math"
  27         "math/rand"
  28         "net"
  29         "strconv"
  30         "sync"
  31         "sync/atomic"
  32         "time"
  33
  34         "github.com/golang/protobuf/proto"
  35         "golang.org/x/net/context"
  36         "golang.org/x/net/http2"
  37         "golang.org/x/net/http2/hpack"
  38         "google.golang.org/grpc/codes"
  39         "google.golang.org/grpc/credentials"
  40         "google.golang.org/grpc/keepalive"
  41         "google.golang.org/grpc/metadata"
  42         "google.golang.org/grpc/peer"
  43         "google.golang.org/grpc/stats"
  44         "google.golang.org/grpc/status"
  45         "google.golang.org/grpc/tap"
  46 )
  47
  48 // ErrIllegalHeaderWrite indicates that setting header is illegal because of
  49 // the stream's state.
  50 var ErrIllegalHeaderWrite = errors.New("transport: the stream is done or WriteHeader was already called")
  51
  52 // http2Server implements the ServerTransport interface with HTTP2.
  53 type http2Server struct {
  54         ctx         context.Context
  55         cancel      context.CancelFunc
  56         conn        net.Conn
  57         remoteAddr  net.Addr
  58         localAddr   net.Addr
  59         maxStreamID uint32               // max stream ID ever seen
  60         authInfo    credentials.AuthInfo // auth info about the connection
  61         inTapHandle tap.ServerInHandle
  62         framer      *framer
  63         hBuf        *bytes.Buffer  // the buffer for HPACK encoding
  64         hEnc        *hpack.Encoder // HPACK encoder
  65         // The max number of concurrent streams.
  66         maxStreams uint32
  67         // controlBuf delivers all the control related tasks (e.g., window
  68         // updates, reset streams, and various settings) to the controller.
  69         controlBuf *controlBuffer
  70         fc         *inFlow
  71         // sendQuotaPool provides flow control to outbound message.
  72         sendQuotaPool *quotaPool
  73         stats         stats.Handler
  74         // Flag to keep track of reading activity on transport.
  75         // 1 is true and 0 is false.
  76         activity uint32 // Accessed atomically.
  77         // Keepalive and max-age parameters for the server.
  78         kp keepalive.ServerParameters
  79
  80         // Keepalive enforcement policy.
  81         kep keepalive.EnforcementPolicy
  82         // The time instance last ping was received.
  83         lastPingAt time.Time
  84         // Number of times the client has violated keepalive ping policy so far.
  85         pingStrikes uint8
  86         // Flag to signify that number of ping strikes should be reset to 0.
  87         // This is set whenever data or header frames are sent.
  88         // 1 means yes.
  89         resetPingStrikes  uint32 // Accessed atomically.
  90         initialWindowSize int32
  91         bdpEst            *bdpEstimator
  92
  93         mu sync.Mutex // guard the following
  94
  95         // drainChan is initialized when drain(...) is called the first time.
  96         // After which the server writes out the first GoAway(with ID 2^31-1) frame.
  97         // Then an independent goroutine will be launched to later send the second GoAway.
  98         // During this time we don't want to write another first GoAway(with ID 2^31 -1) frame.
  99         // Thus call to drain(...) will be a no-op if drainChan is already initialized since draining is
 100         // already underway.
 101         drainChan     chan struct{}
 102         state         transportState
 103         activeStreams map[uint32]*Stream
 104         // the per-stream outbound flow control window size set by the peer.
 105         streamSendQuota uint32
 106         // idle is the time instant when the connection went idle.
 107         // This is either the beginning of the connection or when the number of
 108         // RPCs go down to 0.
 109         // When the connection is busy, this value is set to 0.
 110         idle time.Time
 111 }
 112
 113 // newHTTP2Server constructs a ServerTransport based on HTTP2. ConnectionError is
 114 // returned if something goes wrong.
 115 func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) {
 116         writeBufSize := defaultWriteBufSize
 117         if config.WriteBufferSize > 0 {
 118                 writeBufSize = config.WriteBufferSize
 119         }
 120         readBufSize := defaultReadBufSize
 121         if config.ReadBufferSize > 0 {
 122                 readBufSize = config.ReadBufferSize
 123         }
 124         framer := newFramer(conn, writeBufSize, readBufSize)
 125         // Send initial settings as connection preface to client.
 126         var isettings []http2.Setting
 127         // TODO(zhaoq): Have a better way to signal "no limit" because 0 is
 128         // permitted in the HTTP2 spec.
 129         maxStreams := config.MaxStreams
 130         if maxStreams == 0 {
 131                 maxStreams = math.MaxUint32
 132         } else {
 133                 isettings = append(isettings, http2.Setting{
 134                         ID:  http2.SettingMaxConcurrentStreams,
 135                         Val: maxStreams,
 136                 })
 137         }
 138         dynamicWindow := true
 139         iwz := int32(initialWindowSize)
 140         if config.InitialWindowSize >= defaultWindowSize {
 141                 iwz = config.InitialWindowSize
 142                 dynamicWindow = false
 143         }
 144         icwz := int32(initialWindowSize)
 145         if config.InitialConnWindowSize >= defaultWindowSize {
 146                 icwz = config.InitialConnWindowSize
 147                 dynamicWindow = false
 148         }
 149         if iwz != defaultWindowSize {
 150                 isettings = append(isettings, http2.Setting{
 151                         ID:  http2.SettingInitialWindowSize,
 152                         Val: uint32(iwz)})
 153         }
 154         if err := framer.fr.WriteSettings(isettings...); err != nil {
 155                 return nil, connectionErrorf(true, err, "transport: %v", err)
 156         }
 157         // Adjust the connection flow control window if needed.
 158         if delta := uint32(icwz - defaultWindowSize); delta > 0 {
 159                 if err := framer.fr.WriteWindowUpdate(0, delta); err != nil {
 160                         return nil, connectionErrorf(true, err, "transport: %v", err)
 161                 }
 162         }
 163         kp := config.KeepaliveParams
 164         if kp.MaxConnectionIdle == 0 {
 165                 kp.MaxConnectionIdle = defaultMaxConnectionIdle
 166         }
 167         if kp.MaxConnectionAge == 0 {
 168                 kp.MaxConnectionAge = defaultMaxConnectionAge
 169         }
 170         // Add a jitter to MaxConnectionAge.
 171         kp.MaxConnectionAge += getJitter(kp.MaxConnectionAge)
 172         if kp.MaxConnectionAgeGrace == 0 {
 173                 kp.MaxConnectionAgeGrace = defaultMaxConnectionAgeGrace
 174         }
 175         if kp.Time == 0 {
 176                 kp.Time = defaultServerKeepaliveTime
 177         }
 178         if kp.Timeout == 0 {
 179                 kp.Timeout = defaultServerKeepaliveTimeout
 180         }
 181         kep := config.KeepalivePolicy
 182         if kep.MinTime == 0 {
 183                 kep.MinTime = defaultKeepalivePolicyMinTime
 184         }
 185         var buf bytes.Buffer
 186         ctx, cancel := context.WithCancel(context.Background())
 187         t := &http2Server{
 188                 ctx:               ctx,
 189                 cancel:            cancel,
 190                 conn:              conn,
 191                 remoteAddr:        conn.RemoteAddr(),
 192                 localAddr:         conn.LocalAddr(),
 193                 authInfo:          config.AuthInfo,
 194                 framer:            framer,
 195                 hBuf:              &buf,
 196                 hEnc:              hpack.NewEncoder(&buf),
 197                 maxStreams:        maxStreams,
 198                 inTapHandle:       config.InTapHandle,
 199                 controlBuf:        newControlBuffer(),
 200                 fc:                &inFlow{limit: uint32(icwz)},
 201                 sendQuotaPool:     newQuotaPool(defaultWindowSize),
 202                 state:             reachable,
 203                 activeStreams:     make(map[uint32]*Stream),
 204                 streamSendQuota:   defaultWindowSize,
 205                 stats:             config.StatsHandler,
 206                 kp:                kp,
 207                 idle:              time.Now(),
 208                 kep:               kep,
 209                 initialWindowSize: iwz,
 210         }
 211         if dynamicWindow {
 212                 t.bdpEst = &bdpEstimator{
 213                         bdp:               initialWindowSize,
 214                         updateFlowControl: t.updateFlowControl,
 215                 }
 216         }
 217         if t.stats != nil {
 218                 t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{
 219                         RemoteAddr: t.remoteAddr,
 220                         LocalAddr:  t.localAddr,
 221                 })
 222                 connBegin := &stats.ConnBegin{}
 223                 t.stats.HandleConn(t.ctx, connBegin)
 224         }
 225         t.framer.writer.Flush()
 226         go func() {
 227                 loopyWriter(t.ctx, t.controlBuf, t.itemHandler)
 228                 t.Close()
 229         }()
 230         go t.keepalive()
 231         return t, nil
 232 }
 233
 234 // operateHeader takes action on the decoded headers.
 235 func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (close bool) {
 236         streamID := frame.Header().StreamID
 237
 238         var state decodeState
 239         for _, hf := range frame.Fields {
 240                 if err := state.processHeaderField(hf); err != nil {
 241                         if se, ok := err.(StreamError); ok {
 242                                 t.controlBuf.put(&resetStream{streamID, statusCodeConvTab[se.Code]})
 243                         }
 244                         return
 245                 }
 246         }
 247
 248         buf := newRecvBuffer()
 249         s := &Stream{
 250                 id:           streamID,
 251                 st:           t,
 252                 buf:          buf,
 253                 fc:           &inFlow{limit: uint32(t.initialWindowSize)},
 254                 recvCompress: state.encoding,
 255                 method:       state.method,
 256         }
 257
 258         if frame.StreamEnded() {
 259                 // s is just created by the caller. No lock needed.
 260                 s.state = streamReadDone
 261         }
 262         if state.timeoutSet {
 263                 s.ctx, s.cancel = context.WithTimeout(t.ctx, state.timeout)
 264         } else {
 265                 s.ctx, s.cancel = context.WithCancel(t.ctx)
 266         }
 267         pr := &peer.Peer{
 268                 Addr: t.remoteAddr,
 269         }
 270         // Attach Auth info if there is any.
 271         if t.authInfo != nil {
 272                 pr.AuthInfo = t.authInfo
 273         }
 274         s.ctx = peer.NewContext(s.ctx, pr)
 275         // Cache the current stream to the context so that the server application
 276         // can find out. Required when the server wants to send some metadata
 277         // back to the client (unary call only).
 278         s.ctx = newContextWithStream(s.ctx, s)
 279         // Attach the received metadata to the context.
 280         if len(state.mdata) > 0 {
 281                 s.ctx = metadata.NewIncomingContext(s.ctx, state.mdata)
 282         }
 283         if state.statsTags != nil {
 284                 s.ctx = stats.SetIncomingTags(s.ctx, state.statsTags)
 285         }
 286         if state.statsTrace != nil {
 287                 s.ctx = stats.SetIncomingTrace(s.ctx, state.statsTrace)
 288         }
 289         if t.inTapHandle != nil {
 290                 var err error
 291                 info := &tap.Info{
 292                         FullMethodName: state.method,
 293                 }
 294                 s.ctx, err = t.inTapHandle(s.ctx, info)
 295                 if err != nil {
 296                         warningf("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
 297                         t.controlBuf.put(&resetStream{s.id, http2.ErrCodeRefusedStream})
 298                         return
 299                 }
 300         }
 301         t.mu.Lock()
 302         if t.state != reachable {
 303                 t.mu.Unlock()
 304                 return
 305         }
 306         if uint32(len(t.activeStreams)) >= t.maxStreams {
 307                 t.mu.Unlock()
 308                 t.controlBuf.put(&resetStream{streamID, http2.ErrCodeRefusedStream})
 309                 return
 310         }
 311         if streamID%2 != 1 || streamID <= t.maxStreamID {
 312                 t.mu.Unlock()
 313                 // illegal gRPC stream id.
 314                 errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID)
 315                 return true
 316         }
 317         t.maxStreamID = streamID
 318         s.sendQuotaPool = newQuotaPool(int(t.streamSendQuota))
 319         s.localSendQuota = newQuotaPool(defaultLocalSendQuota)
 320         t.activeStreams[streamID] = s
 321         if len(t.activeStreams) == 1 {
 322                 t.idle = time.Time{}
 323         }
 324         t.mu.Unlock()
 325         s.requestRead = func(n int) {
 326                 t.adjustWindow(s, uint32(n))
 327         }
 328         s.ctx = traceCtx(s.ctx, s.method)
 329         if t.stats != nil {
 330                 s.ctx = t.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
 331                 inHeader := &stats.InHeader{
 332                         FullMethod:  s.method,
 333                         RemoteAddr:  t.remoteAddr,
 334                         LocalAddr:   t.localAddr,
 335                         Compression: s.recvCompress,
 336                         WireLength:  int(frame.Header().Length),
 337                 }
 338                 t.stats.HandleRPC(s.ctx, inHeader)
 339         }
 340         s.trReader = &transportReader{
 341                 reader: &recvBufferReader{
 342                         ctx:  s.ctx,
 343                         recv: s.buf,
 344                 },
 345                 windowHandler: func(n int) {
 346                         t.updateWindow(s, uint32(n))
 347                 },
 348         }
 349         handle(s)
 350         return
 351 }
 352
 353 // HandleStreams receives incoming streams using the given handler. This is
 354 // typically run in a separate goroutine.
 355 // traceCtx attaches trace to ctx and returns the new context.
 356 func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.Context, string) context.Context) {
 357         // Check the validity of client preface.
 358         preface := make([]byte, len(clientPreface))
 359         if _, err := io.ReadFull(t.conn, preface); err != nil {
 360                 // Only log if it isn't a simple tcp accept check (ie: tcp balancer doing open/close socket)
 361                 if err != io.EOF {
 362                         errorf("transport: http2Server.HandleStreams failed to receive the preface from client: %v", err)
 363                 }
 364                 t.Close()
 365                 return
 366         }
 367         if !bytes.Equal(preface, clientPreface) {
 368                 errorf("transport: http2Server.HandleStreams received bogus greeting from client: %q", preface)
 369                 t.Close()
 370                 return
 371         }
 372
 373         frame, err := t.framer.fr.ReadFrame()
 374         if err == io.EOF || err == io.ErrUnexpectedEOF {
 375                 t.Close()
 376                 return
 377         }
 378         if err != nil {
 379                 errorf("transport: http2Server.HandleStreams failed to read initial settings frame: %v", err)
 380                 t.Close()
 381                 return
 382         }
 383         atomic.StoreUint32(&t.activity, 1)
 384         sf, ok := frame.(*http2.SettingsFrame)
 385         if !ok {
 386                 errorf("transport: http2Server.HandleStreams saw invalid preface type %T from client", frame)
 387                 t.Close()
 388                 return
 389         }
 390         t.handleSettings(sf)
 391
 392         for {
 393                 frame, err := t.framer.fr.ReadFrame()
 394                 atomic.StoreUint32(&t.activity, 1)
 395                 if err != nil {
 396                         if se, ok := err.(http2.StreamError); ok {
 397                                 t.mu.Lock()
 398                                 s := t.activeStreams[se.StreamID]
 399                                 t.mu.Unlock()
 400                                 if s != nil {
 401                                         t.closeStream(s)
 402                                 }
 403                                 t.controlBuf.put(&resetStream{se.StreamID, se.Code})
 404                                 continue
 405                         }
 406                         if err == io.EOF || err == io.ErrUnexpectedEOF {
 407                                 t.Close()
 408                                 return
 409                         }
 410                         warningf("transport: http2Server.HandleStreams failed to read frame: %v", err)
 411                         t.Close()
 412                         return
 413                 }
 414                 switch frame := frame.(type) {
 415                 case *http2.MetaHeadersFrame:
 416                         if t.operateHeaders(frame, handle, traceCtx) {
 417                                 t.Close()
 418                                 break
 419                         }
 420                 case *http2.DataFrame:
 421                         t.handleData(frame)
 422                 case *http2.RSTStreamFrame:
 423                         t.handleRSTStream(frame)
 424                 case *http2.SettingsFrame:
 425                         t.handleSettings(frame)
 426                 case *http2.PingFrame:
 427                         t.handlePing(frame)
 428                 case *http2.WindowUpdateFrame:
 429                         t.handleWindowUpdate(frame)
 430                 case *http2.GoAwayFrame:
 431                         // TODO: Handle GoAway from the client appropriately.
 432                 default:
 433                         errorf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame)
 434                 }
 435         }
 436 }
 437
 438 func (t *http2Server) getStream(f http2.Frame) (*Stream, bool) {
 439         t.mu.Lock()
 440         defer t.mu.Unlock()
 441         if t.activeStreams == nil {
 442                 // The transport is closing.
 443                 return nil, false
 444         }
 445         s, ok := t.activeStreams[f.Header().StreamID]
 446         if !ok {
 447                 // The stream is already done.
 448                 return nil, false
 449         }
 450         return s, true
 451 }
 452
 453 // adjustWindow sends out extra window update over the initial window size
 454 // of stream if the application is requesting data larger in size than
 455 // the window.
 456 func (t *http2Server) adjustWindow(s *Stream, n uint32) {
 457         s.mu.Lock()
 458         defer s.mu.Unlock()
 459         if s.state == streamDone {
 460                 return
 461         }
 462         if w := s.fc.maybeAdjust(n); w > 0 {
 463                 if cw := t.fc.resetPendingUpdate(); cw > 0 {
 464                         t.controlBuf.put(&windowUpdate{0, cw})
 465                 }
 466                 t.controlBuf.put(&windowUpdate{s.id, w})
 467         }
 468 }
 469
 470 // updateWindow adjusts the inbound quota for the stream and the transport.
 471 // Window updates will deliver to the controller for sending when
 472 // the cumulative quota exceeds the corresponding threshold.
 473 func (t *http2Server) updateWindow(s *Stream, n uint32) {
 474         s.mu.Lock()
 475         defer s.mu.Unlock()
 476         if s.state == streamDone {
 477                 return
 478         }
 479         if w := s.fc.onRead(n); w > 0 {
 480                 if cw := t.fc.resetPendingUpdate(); cw > 0 {
 481                         t.controlBuf.put(&windowUpdate{0, cw})
 482                 }
 483                 t.controlBuf.put(&windowUpdate{s.id, w})
 484         }
 485 }
 486
 487 // updateFlowControl updates the incoming flow control windows
 488 // for the transport and the stream based on the current bdp
 489 // estimation.
 490 func (t *http2Server) updateFlowControl(n uint32) {
 491         t.mu.Lock()
 492         for _, s := range t.activeStreams {
 493                 s.fc.newLimit(n)
 494         }
 495         t.initialWindowSize = int32(n)
 496         t.mu.Unlock()
 497         t.controlBuf.put(&windowUpdate{0, t.fc.newLimit(n)})
 498         t.controlBuf.put(&settings{
 499                 ack: false,
 500                 ss: []http2.Setting{
 501                         {
 502                                 ID:  http2.SettingInitialWindowSize,
 503                                 Val: uint32(n),
 504                         },
 505                 },
 506         })
 507
 508 }
 509
 510 func (t *http2Server) handleData(f *http2.DataFrame) {
 511         size := f.Header().Length
 512         var sendBDPPing bool
 513         if t.bdpEst != nil {
 514                 sendBDPPing = t.bdpEst.add(uint32(size))
 515         }
 516         // Decouple connection's flow control from application's read.
 517         // An update on connection's flow control should not depend on
 518         // whether user application has read the data or not. Such a
 519         // restriction is already imposed on the stream's flow control,
 520         // and therefore the sender will be blocked anyways.
 521         // Decoupling the connection flow control will prevent other
 522         // active(fast) streams from starving in presence of slow or
 523         // inactive streams.
 524         //
 525         // Furthermore, if a bdpPing is being sent out we can piggyback
 526         // connection's window update for the bytes we just received.
 527         if sendBDPPing {
 528                 if size != 0 { // Could be an empty frame.
 529                         t.controlBuf.put(&windowUpdate{0, uint32(size)})
 530                 }
 531                 t.controlBuf.put(bdpPing)
 532         } else {
 533                 if err := t.fc.onData(uint32(size)); err != nil {
 534                         errorf("transport: http2Server %v", err)
 535                         t.Close()
 536                         return
 537                 }
 538                 if w := t.fc.onRead(uint32(size)); w > 0 {
 539                         t.controlBuf.put(&windowUpdate{0, w})
 540                 }
 541         }
 542         // Select the right stream to dispatch.
 543         s, ok := t.getStream(f)
 544         if !ok {
 545                 return
 546         }
 547         if size > 0 {
 548                 s.mu.Lock()
 549                 if s.state == streamDone {
 550                         s.mu.Unlock()
 551                         return
 552                 }
 553                 if err := s.fc.onData(uint32(size)); err != nil {
 554                         s.mu.Unlock()
 555                         t.closeStream(s)
 556                         t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl})
 557                         return
 558                 }
 559                 if f.Header().Flags.Has(http2.FlagDataPadded) {
 560                         if w := s.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
 561                                 t.controlBuf.put(&windowUpdate{s.id, w})
 562                         }
 563                 }
 564                 s.mu.Unlock()
 565                 // TODO(bradfitz, zhaoq): A copy is required here because there is no
 566                 // guarantee f.Data() is consumed before the arrival of next frame.
 567                 // Can this copy be eliminated?
 568                 if len(f.Data()) > 0 {
 569                         data := make([]byte, len(f.Data()))
 570                         copy(data, f.Data())
 571                         s.write(recvMsg{data: data})
 572                 }
 573         }
 574         if f.Header().Flags.Has(http2.FlagDataEndStream) {
 575                 // Received the end of stream from the client.
 576                 s.mu.Lock()
 577                 if s.state != streamDone {
 578                         s.state = streamReadDone
 579                 }
 580                 s.mu.Unlock()
 581                 s.write(recvMsg{err: io.EOF})
 582         }
 583 }
 584
 585 func (t *http2Server) handleRSTStream(f *http2.RSTStreamFrame) {
 586         s, ok := t.getStream(f)
 587         if !ok {
 588                 return
 589         }
 590         t.closeStream(s)
 591 }
 592
 593 func (t *http2Server) handleSettings(f *http2.SettingsFrame) {
 594         if f.IsAck() {
 595                 return
 596         }
 597         var ss []http2.Setting
 598         f.ForeachSetting(func(s http2.Setting) error {
 599                 ss = append(ss, s)
 600                 return nil
 601         })
 602         t.controlBuf.put(&settings{ack: true, ss: ss})
 603 }
 604
 605 func (t *http2Server) applySettings(ss []http2.Setting) {
 606         for _, s := range ss {
 607                 if s.ID == http2.SettingInitialWindowSize {
 608                         t.mu.Lock()
 609                         for _, stream := range t.activeStreams {
 610                                 stream.sendQuotaPool.addAndUpdate(int(s.Val) - int(t.streamSendQuota))
 611                         }
 612                         t.streamSendQuota = s.Val
 613                         t.mu.Unlock()
 614                 }
 615
 616         }
 617 }
 618
 619 const (
 620         maxPingStrikes     = 2
 621         defaultPingTimeout = 2 * time.Hour
 622 )
 623
 624 func (t *http2Server) handlePing(f *http2.PingFrame) {
 625         if f.IsAck() {
 626                 if f.Data == goAwayPing.data && t.drainChan != nil {
 627                         close(t.drainChan)
 628                         return
 629                 }
 630                 // Maybe it's a BDP ping.
 631                 if t.bdpEst != nil {
 632                         t.bdpEst.calculate(f.Data)
 633                 }
 634                 return
 635         }
 636         pingAck := &ping{ack: true}
 637         copy(pingAck.data[:], f.Data[:])
 638         t.controlBuf.put(pingAck)
 639
 640         now := time.Now()
 641         defer func() {
 642                 t.lastPingAt = now
 643         }()
 644         // A reset ping strikes means that we don't need to check for policy
 645         // violation for this ping and the pingStrikes counter should be set
 646         // to 0.
 647         if atomic.CompareAndSwapUint32(&t.resetPingStrikes, 1, 0) {
 648                 t.pingStrikes = 0
 649                 return
 650         }
 651         t.mu.Lock()
 652         ns := len(t.activeStreams)
 653         t.mu.Unlock()
 654         if ns < 1 && !t.kep.PermitWithoutStream {
 655                 // Keepalive shouldn't be active thus, this new ping should
 656                 // have come after at least defaultPingTimeout.
 657                 if t.lastPingAt.Add(defaultPingTimeout).After(now) {
 658                         t.pingStrikes++
 659                 }
 660         } else {
 661                 // Check if keepalive policy is respected.
 662                 if t.lastPingAt.Add(t.kep.MinTime).After(now) {
 663                         t.pingStrikes++
 664                 }
 665         }
 666
 667         if t.pingStrikes > maxPingStrikes {
 668                 // Send goaway and close the connection.
 669                 errorf("transport: Got to too many pings from the client, closing the connection.")
 670                 t.controlBuf.put(&goAway{code: http2.ErrCodeEnhanceYourCalm, debugData: []byte("too_many_pings"), closeConn: true})
 671         }
 672 }
 673
 674 func (t *http2Server) handleWindowUpdate(f *http2.WindowUpdateFrame) {
 675         id := f.Header().StreamID
 676         incr := f.Increment
 677         if id == 0 {
 678                 t.sendQuotaPool.add(int(incr))
 679                 return
 680         }
 681         if s, ok := t.getStream(f); ok {
 682                 s.sendQuotaPool.add(int(incr))
 683         }
 684 }
 685
 686 // WriteHeader sends the header metedata md back to the client.
 687 func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error {
 688         select {
 689         case <-s.ctx.Done():
 690                 return ContextErr(s.ctx.Err())
 691         case <-t.ctx.Done():
 692                 return ErrConnClosing
 693         default:
 694         }
 695
 696         s.mu.Lock()
 697         if s.headerOk || s.state == streamDone {
 698                 s.mu.Unlock()
 699                 return ErrIllegalHeaderWrite
 700         }
 701         s.headerOk = true
 702         if md.Len() > 0 {
 703                 if s.header.Len() > 0 {
 704                         s.header = metadata.Join(s.header, md)
 705                 } else {
 706                         s.header = md
 707                 }
 708         }
 709         md = s.header
 710         s.mu.Unlock()
 711         // TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields
 712         // first and create a slice of that exact size.
 713         headerFields := make([]hpack.HeaderField, 0, 2) // at least :status, content-type will be there if none else.
 714         headerFields = append(headerFields, hpack.HeaderField{Name: ":status", Value: "200"})
 715         headerFields = append(headerFields, hpack.HeaderField{Name: "content-type", Value: "application/grpc"})
 716         if s.sendCompress != "" {
 717                 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-encoding", Value: s.sendCompress})
 718         }
 719         for k, vv := range md {
 720                 if isReservedHeader(k) {
 721                         // Clients don't tolerate reading restricted headers after some non restricted ones were sent.
 722                         continue
 723                 }
 724                 for _, v := range vv {
 725                         headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
 726                 }
 727         }
 728         t.controlBuf.put(&headerFrame{
 729                 streamID:  s.id,
 730                 hf:        headerFields,
 731                 endStream: false,
 732         })
 733         if t.stats != nil {
 734                 outHeader := &stats.OutHeader{
 735                 //WireLength: // TODO(mmukhi): Revisit this later, if needed.
 736                 }
 737                 t.stats.HandleRPC(s.Context(), outHeader)
 738         }
 739         return nil
 740 }
 741
 742 // WriteStatus sends stream status to the client and terminates the stream.
 743 // There is no further I/O operations being able to perform on this stream.
 744 // TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early
 745 // OK is adopted.
 746 func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error {
 747         select {
 748         case <-t.ctx.Done():
 749                 return ErrConnClosing
 750         default:
 751         }
 752
 753         var headersSent, hasHeader bool
 754         s.mu.Lock()
 755         if s.state == streamDone {
 756                 s.mu.Unlock()
 757                 return nil
 758         }
 759         if s.headerOk {
 760                 headersSent = true
 761         }
 762         if s.header.Len() > 0 {
 763                 hasHeader = true
 764         }
 765         s.mu.Unlock()
 766
 767         if !headersSent && hasHeader {
 768                 t.WriteHeader(s, nil)
 769                 headersSent = true
 770         }
 771
 772         // TODO(mmukhi): Benchmark if the performance gets better if count the metadata and other header fields
 773         // first and create a slice of that exact size.
 774         headerFields := make([]hpack.HeaderField, 0, 2) // grpc-status and grpc-message will be there if none else.
 775         if !headersSent {
 776                 headerFields = append(headerFields, hpack.HeaderField{Name: ":status", Value: "200"})
 777                 headerFields = append(headerFields, hpack.HeaderField{Name: "content-type", Value: "application/grpc"})
 778         }
 779         headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-status", Value: strconv.Itoa(int(st.Code()))})
 780         headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(st.Message())})
 781
 782         if p := st.Proto(); p != nil && len(p.Details) > 0 {
 783                 stBytes, err := proto.Marshal(p)
 784                 if err != nil {
 785                         // TODO: return error instead, when callers are able to handle it.
 786                         panic(err)
 787                 }
 788
 789                 headerFields = append(headerFields, hpack.HeaderField{Name: "grpc-status-details-bin", Value: encodeBinHeader(stBytes)})
 790         }
 791
 792         // Attach the trailer metadata.
 793         for k, vv := range s.trailer {
 794                 // Clients don't tolerate reading restricted headers after some non restricted ones were sent.
 795                 if isReservedHeader(k) {
 796                         continue
 797                 }
 798                 for _, v := range vv {
 799                         headerFields = append(headerFields, hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)})
 800                 }
 801         }
 802         t.controlBuf.put(&headerFrame{
 803                 streamID:  s.id,
 804                 hf:        headerFields,
 805                 endStream: true,
 806         })
 807         if t.stats != nil {
 808                 t.stats.HandleRPC(s.Context(), &stats.OutTrailer{})
 809         }
 810         t.closeStream(s)
 811         return nil
 812 }
 813
 814 // Write converts the data into HTTP2 data frame and sends it out. Non-nil error
 815 // is returns if it fails (e.g., framing error, transport error).
 816 func (t *http2Server) Write(s *Stream, hdr []byte, data []byte, opts *Options) error {
 817         select {
 818         case <-s.ctx.Done():
 819                 return ContextErr(s.ctx.Err())
 820         case <-t.ctx.Done():
 821                 return ErrConnClosing
 822         default:
 823         }
 824
 825         var writeHeaderFrame bool
 826         s.mu.Lock()
 827         if s.state == streamDone {
 828                 s.mu.Unlock()
 829                 return streamErrorf(codes.Unknown, "the stream has been done")
 830         }
 831         if !s.headerOk {
 832                 writeHeaderFrame = true
 833         }
 834         s.mu.Unlock()
 835         if writeHeaderFrame {
 836                 t.WriteHeader(s, nil)
 837         }
 838         // Add data to header frame so that we can equally distribute data across frames.
 839         emptyLen := http2MaxFrameLen - len(hdr)
 840         if emptyLen > len(data) {
 841                 emptyLen = len(data)
 842         }
 843         hdr = append(hdr, data[:emptyLen]...)
 844         data = data[emptyLen:]
 845         var (
 846                 streamQuota    int
 847                 streamQuotaVer uint32
 848                 localSendQuota int
 849                 err            error
 850                 sqChan         <-chan int
 851         )
 852         for _, r := range [][]byte{hdr, data} {
 853                 for len(r) > 0 {
 854                         size := http2MaxFrameLen
 855                         if size > len(r) {
 856                                 size = len(r)
 857                         }
 858                         if streamQuota == 0 { // Used up all the locally cached stream quota.
 859                                 sqChan, streamQuotaVer = s.sendQuotaPool.acquireWithVersion()
 860                                 // Wait until the stream has some quota to send the data.
 861                                 streamQuota, err = wait(s.ctx, t.ctx, nil, nil, sqChan)
 862                                 if err != nil {
 863                                         return err
 864                                 }
 865                         }
 866                         if localSendQuota <= 0 {
 867                                 localSendQuota, err = wait(s.ctx, t.ctx, nil, nil, s.localSendQuota.acquire())
 868                                 if err != nil {
 869                                         return err
 870                                 }
 871                         }
 872                         if size > streamQuota {
 873                                 size = streamQuota
 874                         } // No need to do that for localSendQuota since that's only a soft limit.
 875                         // Wait until the transport has some quota to send the data.
 876                         tq, err := wait(s.ctx, t.ctx, nil, nil, t.sendQuotaPool.acquire())
 877                         if err != nil {
 878                                 return err
 879                         }
 880                         if tq < size {
 881                                 size = tq
 882                         }
 883                         if tq > size {
 884                                 t.sendQuotaPool.add(tq - size)
 885                         }
 886                         streamQuota -= size
 887                         localSendQuota -= size
 888                         p := r[:size]
 889                         // Reset ping strikes when sending data since this might cause
 890                         // the peer to send ping.
 891                         atomic.StoreUint32(&t.resetPingStrikes, 1)
 892                         success := func() {
 893                                 sz := size
 894                                 t.controlBuf.put(&dataFrame{streamID: s.id, endStream: false, d: p, f: func() {
 895                                         s.localSendQuota.add(sz)
 896                                 }})
 897                                 r = r[size:]
 898                         }
 899                         failure := func() { // The stream quota version must have changed.
 900                                 // Our streamQuota cache is invalidated now, so give it back.
 901                                 s.sendQuotaPool.lockedAdd(streamQuota + size)
 902                         }
 903                         if !s.sendQuotaPool.compareAndExecute(streamQuotaVer, success, failure) {
 904                                 // Couldn't send this chunk out.
 905                                 t.sendQuotaPool.add(size)
 906                                 localSendQuota += size
 907                                 streamQuota = 0
 908                         }
 909                 }
 910         }
 911         if streamQuota > 0 {
 912                 // ADd the left over quota back to stream.
 913                 s.sendQuotaPool.add(streamQuota)
 914         }
 915         if localSendQuota > 0 {
 916                 s.localSendQuota.add(localSendQuota)
 917         }
 918         return nil
 919 }
 920
 921 // keepalive running in a separate goroutine does the following:
 922 // 1. Gracefully closes an idle connection after a duration of keepalive.MaxConnectionIdle.
 923 // 2. Gracefully closes any connection after a duration of keepalive.MaxConnectionAge.
 924 // 3. Forcibly closes a connection after an additive period of keepalive.MaxConnectionAgeGrace over keepalive.MaxConnectionAge.
 925 // 4. Makes sure a connection is alive by sending pings with a frequency of keepalive.Time and closes a non-responsive connection
 926 // after an additional duration of keepalive.Timeout.
 927 func (t *http2Server) keepalive() {
 928         p := &ping{}
 929         var pingSent bool
 930         maxIdle := time.NewTimer(t.kp.MaxConnectionIdle)
 931         maxAge := time.NewTimer(t.kp.MaxConnectionAge)
 932         keepalive := time.NewTimer(t.kp.Time)
 933         // NOTE: All exit paths of this function should reset their
 934         // respective timers. A failure to do so will cause the
 935         // following clean-up to deadlock and eventually leak.
 936         defer func() {
 937                 if !maxIdle.Stop() {
 938                         <-maxIdle.C
 939                 }
 940                 if !maxAge.Stop() {
 941                         <-maxAge.C
 942                 }
 943                 if !keepalive.Stop() {
 944                         <-keepalive.C
 945                 }
 946         }()
 947         for {
 948                 select {
 949                 case <-maxIdle.C:
 950                         t.mu.Lock()
 951                         idle := t.idle
 952                         if idle.IsZero() { // The connection is non-idle.
 953                                 t.mu.Unlock()
 954                                 maxIdle.Reset(t.kp.MaxConnectionIdle)
 955                                 continue
 956                         }
 957                         val := t.kp.MaxConnectionIdle - time.Since(idle)
 958                         t.mu.Unlock()
 959                         if val <= 0 {
 960                                 // The connection has been idle for a duration of keepalive.MaxConnectionIdle or more.
 961                                 // Gracefully close the connection.
 962                                 t.drain(http2.ErrCodeNo, []byte{})
 963                                 // Reseting the timer so that the clean-up doesn't deadlock.
 964                                 maxIdle.Reset(infinity)
 965                                 return
 966                         }
 967                         maxIdle.Reset(val)
 968                 case <-maxAge.C:
 969                         t.drain(http2.ErrCodeNo, []byte{})
 970                         maxAge.Reset(t.kp.MaxConnectionAgeGrace)
 971                         select {
 972                         case <-maxAge.C:
 973                                 // Close the connection after grace period.
 974                                 t.Close()
 975                                 // Reseting the timer so that the clean-up doesn't deadlock.
 976                                 maxAge.Reset(infinity)
 977                         case <-t.ctx.Done():
 978                         }
 979                         return
 980                 case <-keepalive.C:
 981                         if atomic.CompareAndSwapUint32(&t.activity, 1, 0) {
 982                                 pingSent = false
 983                                 keepalive.Reset(t.kp.Time)
 984                                 continue
 985                         }
 986                         if pingSent {
 987                                 t.Close()
 988                                 // Reseting the timer so that the clean-up doesn't deadlock.
 989                                 keepalive.Reset(infinity)
 990                                 return
 991                         }
 992                         pingSent = true
 993                         t.controlBuf.put(p)
 994                         keepalive.Reset(t.kp.Timeout)
 995                 case <-t.ctx.Done():
 996                         return
 997                 }
 998         }
 999 }
1000
1001 var goAwayPing = &ping{data: [8]byte{1, 6, 1, 8, 0, 3, 3, 9}}
1002
1003 // TODO(mmukhi): A lot of this code(and code in other places in the tranpsort layer)
1004 // is duplicated between the client and the server.
1005 // The transport layer needs to be refactored to take care of this.
1006 func (t *http2Server) itemHandler(i item) error {
1007         switch i := i.(type) {
1008         case *dataFrame:
1009                 if err := t.framer.fr.WriteData(i.streamID, i.endStream, i.d); err != nil {
1010                         return err
1011                 }
1012                 i.f()
1013                 return nil
1014         case *headerFrame:
1015                 t.hBuf.Reset()
1016                 for _, f := range i.hf {
1017                         t.hEnc.WriteField(f)
1018                 }
1019                 first := true
1020                 endHeaders := false
1021                 for !endHeaders {
1022                         size := t.hBuf.Len()
1023                         if size > http2MaxFrameLen {
1024                                 size = http2MaxFrameLen
1025                         } else {
1026                                 endHeaders = true
1027                         }
1028                         var err error
1029                         if first {
1030                                 first = false
1031                                 err = t.framer.fr.WriteHeaders(http2.HeadersFrameParam{
1032                                         StreamID:      i.streamID,
1033                                         BlockFragment: t.hBuf.Next(size),
1034                                         EndStream:     i.endStream,
1035                                         EndHeaders:    endHeaders,
1036                                 })
1037                         } else {
1038                                 err = t.framer.fr.WriteContinuation(
1039                                         i.streamID,
1040                                         endHeaders,
1041                                         t.hBuf.Next(size),
1042                                 )
1043                         }
1044                         if err != nil {
1045                                 return err
1046                         }
1047                 }
1048                 atomic.StoreUint32(&t.resetPingStrikes, 1)
1049                 return nil
1050         case *windowUpdate:
1051                 return t.framer.fr.WriteWindowUpdate(i.streamID, i.increment)
1052         case *settings:
1053                 if i.ack {
1054                         t.applySettings(i.ss)
1055                         return t.framer.fr.WriteSettingsAck()
1056                 }
1057                 return t.framer.fr.WriteSettings(i.ss...)
1058         case *resetStream:
1059                 return t.framer.fr.WriteRSTStream(i.streamID, i.code)
1060         case *goAway:
1061                 t.mu.Lock()
1062                 if t.state == closing {
1063                         t.mu.Unlock()
1064                         // The transport is closing.
1065                         return fmt.Errorf("transport: Connection closing")
1066                 }
1067                 sid := t.maxStreamID
1068                 if !i.headsUp {
1069                         // Stop accepting more streams now.
1070                         t.state = draining
1071                         t.mu.Unlock()
1072                         if err := t.framer.fr.WriteGoAway(sid, i.code, i.debugData); err != nil {
1073                                 return err
1074                         }
1075                         if i.closeConn {
1076                                 // Abruptly close the connection following the GoAway (via
1077                                 // loopywriter).  But flush out what's inside the buffer first.
1078                                 t.framer.writer.Flush()
1079                                 return fmt.Errorf("transport: Connection closing")
1080                         }
1081                         return nil
1082                 }
1083                 t.mu.Unlock()
1084                 // For a graceful close, send out a GoAway with stream ID of MaxUInt32,
1085                 // Follow that with a ping and wait for the ack to come back or a timer
1086                 // to expire. During this time accept new streams since they might have
1087                 // originated before the GoAway reaches the client.
1088                 // After getting the ack or timer expiration send out another GoAway this
1089                 // time with an ID of the max stream server intends to process.
1090                 if err := t.framer.fr.WriteGoAway(math.MaxUint32, http2.ErrCodeNo, []byte{}); err != nil {
1091                         return err
1092                 }
1093                 if err := t.framer.fr.WritePing(false, goAwayPing.data); err != nil {
1094                         return err
1095                 }
1096                 go func() {
1097                         timer := time.NewTimer(time.Minute)
1098                         defer timer.Stop()
1099                         select {
1100                         case <-t.drainChan:
1101                         case <-timer.C:
1102                         case <-t.ctx.Done():
1103                                 return
1104                         }
1105                         t.controlBuf.put(&goAway{code: i.code, debugData: i.debugData})
1106                 }()
1107                 return nil
1108         case *flushIO:
1109                 return t.framer.writer.Flush()
1110         case *ping:
1111                 if !i.ack {
1112                         t.bdpEst.timesnap(i.data)
1113                 }
1114                 return t.framer.fr.WritePing(i.ack, i.data)
1115         default:
1116                 err := status.Errorf(codes.Internal, "transport: http2Server.controller got unexpected item type %t", i)
1117                 errorf("%v", err)
1118                 return err
1119         }
1120 }
1121
1122 // Close starts shutting down the http2Server transport.
1123 // TODO(zhaoq): Now the destruction is not blocked on any pending streams. This
1124 // could cause some resource issue. Revisit this later.
1125 func (t *http2Server) Close() error {
1126         t.mu.Lock()
1127         if t.state == closing {
1128                 t.mu.Unlock()
1129                 return errors.New("transport: Close() was already called")
1130         }
1131         t.state = closing
1132         streams := t.activeStreams
1133         t.activeStreams = nil
1134         t.mu.Unlock()
1135         t.cancel()
1136         err := t.conn.Close()
1137         // Cancel all active streams.
1138         for _, s := range streams {
1139                 s.cancel()
1140         }
1141         if t.stats != nil {
1142                 connEnd := &stats.ConnEnd{}
1143                 t.stats.HandleConn(t.ctx, connEnd)
1144         }
1145         return err
1146 }
1147
1148 // closeStream clears the footprint of a stream when the stream is not needed
1149 // any more.
1150 func (t *http2Server) closeStream(s *Stream) {
1151         t.mu.Lock()
1152         delete(t.activeStreams, s.id)
1153         if len(t.activeStreams) == 0 {
1154                 t.idle = time.Now()
1155         }
1156         if t.state == draining && len(t.activeStreams) == 0 {
1157                 defer t.Close()
1158         }
1159         t.mu.Unlock()
1160         // In case stream sending and receiving are invoked in separate
1161         // goroutines (e.g., bi-directional streaming), cancel needs to be
1162         // called to interrupt the potential blocking on other goroutines.
1163         s.cancel()
1164         s.mu.Lock()
1165         if s.state == streamDone {
1166                 s.mu.Unlock()
1167                 return
1168         }
1169         s.state = streamDone
1170         s.mu.Unlock()
1171 }
1172
1173 func (t *http2Server) RemoteAddr() net.Addr {
1174         return t.remoteAddr
1175 }
1176
1177 func (t *http2Server) Drain() {
1178         t.drain(http2.ErrCodeNo, []byte{})
1179 }
1180
1181 func (t *http2Server) drain(code http2.ErrCode, debugData []byte) {
1182         t.mu.Lock()
1183         defer t.mu.Unlock()
1184         if t.drainChan != nil {
1185                 return
1186         }
1187         t.drainChan = make(chan struct{})
1188         t.controlBuf.put(&goAway{code: code, debugData: debugData, headsUp: true})
1189 }
1190
1191 var rgen = rand.New(rand.NewSource(time.Now().UnixNano()))
1192
1193 func getJitter(v time.Duration) time.Duration {
1194         if v == infinity {
1195                 return 0
1196         }
1197         // Generate a jitter between +/- 10% of the value.
1198         r := int64(v / 10)
1199         j := rgen.Int63n(2*r) - r
1200         return time.Duration(j)
1201 }