/* * * 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 transport import ( "bytes" "errors" "io" "math" "math/rand" "net" "strconv" "sync" "sync/atomic" "time" "github.com/golang/protobuf/proto" "golang.org/x/net/context" "golang.org/x/net/http2" "golang.org/x/net/http2/hpack" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials" "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" ) // ErrIllegalHeaderWrite indicates that setting header is illegal because of // the stream's state. var ErrIllegalHeaderWrite = errors.New("transport: the stream is done or WriteHeader was already called") // http2Server implements the ServerTransport interface with HTTP2. type http2Server struct { ctx context.Context conn net.Conn remoteAddr net.Addr localAddr net.Addr maxStreamID uint32 // max stream ID ever seen authInfo credentials.AuthInfo // auth info about the connection inTapHandle tap.ServerInHandle // writableChan synchronizes write access to the transport. // A writer acquires the write lock by receiving a value on writableChan // and releases it by sending on writableChan. writableChan chan int // shutdownChan is closed when Close is called. // Blocking operations should select on shutdownChan to avoid // blocking forever after Close. shutdownChan chan struct{} framer *framer hBuf *bytes.Buffer // the buffer for HPACK encoding hEnc *hpack.Encoder // HPACK encoder // The max number of concurrent streams. maxStreams uint32 // controlBuf delivers all the control related tasks (e.g., window // updates, reset streams, and various settings) to the controller. controlBuf *controlBuffer fc *inFlow // sendQuotaPool provides flow control to outbound message. sendQuotaPool *quotaPool stats stats.Handler // Flag to keep track of reading activity on transport. // 1 is true and 0 is false. activity uint32 // Accessed atomically. // Keepalive and max-age parameters for the server. kp keepalive.ServerParameters // Keepalive enforcement policy. kep keepalive.EnforcementPolicy // The time instance last ping was received. lastPingAt time.Time // Number of times the client has violated keepalive ping policy so far. pingStrikes uint8 // Flag to signify that number of ping strikes should be reset to 0. // This is set whenever data or header frames are sent. // 1 means yes. resetPingStrikes uint32 // Accessed atomically. initialWindowSize int32 bdpEst *bdpEstimator outQuotaVersion uint32 mu sync.Mutex // guard the following // drainChan is initialized when drain(...) is called the first time. // After which the server writes out the first GoAway(with ID 2^31-1) frame. // Then an independent goroutine will be launched to later send the second GoAway. // During this time we don't want to write another first GoAway(with ID 2^31 -1) frame. // Thus call to drain(...) will be a no-op if drainChan is already initialized since draining is // already underway. drainChan chan struct{} state transportState activeStreams map[uint32]*Stream // the per-stream outbound flow control window size set by the peer. streamSendQuota uint32 // idle is the time instant when the connection went idle. // This is either the begining of the connection or when the number of // RPCs go down to 0. // When the connection is busy, this value is set to 0. idle time.Time } // newHTTP2Server constructs a ServerTransport based on HTTP2. ConnectionError is // returned if something goes wrong. func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) { framer := newFramer(conn) // Send initial settings as connection preface to client. var isettings []http2.Setting // TODO(zhaoq): Have a better way to signal "no limit" because 0 is // permitted in the HTTP2 spec. maxStreams := config.MaxStreams if maxStreams == 0 { maxStreams = math.MaxUint32 } else { isettings = append(isettings, http2.Setting{ ID: http2.SettingMaxConcurrentStreams, Val: maxStreams, }) } dynamicWindow := true iwz := int32(initialWindowSize) if config.InitialWindowSize >= defaultWindowSize { iwz = config.InitialWindowSize dynamicWindow = false } icwz := int32(initialWindowSize) if config.InitialConnWindowSize >= defaultWindowSize { icwz = config.InitialConnWindowSize dynamicWindow = false } if iwz != defaultWindowSize { isettings = append(isettings, http2.Setting{ ID: http2.SettingInitialWindowSize, Val: uint32(iwz)}) } if err := framer.writeSettings(true, isettings...); err != nil { return nil, connectionErrorf(true, err, "transport: %v", err) } // Adjust the connection flow control window if needed. if delta := uint32(icwz - defaultWindowSize); delta > 0 { if err := framer.writeWindowUpdate(true, 0, delta); err != nil { return nil, connectionErrorf(true, err, "transport: %v", err) } } kp := config.KeepaliveParams if kp.MaxConnectionIdle == 0 { kp.MaxConnectionIdle = defaultMaxConnectionIdle } if kp.MaxConnectionAge == 0 { kp.MaxConnectionAge = defaultMaxConnectionAge } // Add a jitter to MaxConnectionAge. kp.MaxConnectionAge += getJitter(kp.MaxConnectionAge) if kp.MaxConnectionAgeGrace == 0 { kp.MaxConnectionAgeGrace = defaultMaxConnectionAgeGrace } if kp.Time == 0 { kp.Time = defaultServerKeepaliveTime } if kp.Timeout == 0 { kp.Timeout = defaultServerKeepaliveTimeout } kep := config.KeepalivePolicy if kep.MinTime == 0 { kep.MinTime = defaultKeepalivePolicyMinTime } var buf bytes.Buffer t := &http2Server{ ctx: context.Background(), conn: conn, remoteAddr: conn.RemoteAddr(), localAddr: conn.LocalAddr(), authInfo: config.AuthInfo, framer: framer, hBuf: &buf, hEnc: hpack.NewEncoder(&buf), maxStreams: maxStreams, inTapHandle: config.InTapHandle, controlBuf: newControlBuffer(), fc: &inFlow{limit: uint32(icwz)}, sendQuotaPool: newQuotaPool(defaultWindowSize), state: reachable, writableChan: make(chan int, 1), shutdownChan: make(chan struct{}), activeStreams: make(map[uint32]*Stream), streamSendQuota: defaultWindowSize, stats: config.StatsHandler, kp: kp, idle: time.Now(), kep: kep, initialWindowSize: iwz, } if dynamicWindow { t.bdpEst = &bdpEstimator{ bdp: initialWindowSize, updateFlowControl: t.updateFlowControl, } } if t.stats != nil { t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{ RemoteAddr: t.remoteAddr, LocalAddr: t.localAddr, }) connBegin := &stats.ConnBegin{} t.stats.HandleConn(t.ctx, connBegin) } go t.controller() go t.keepalive() t.writableChan <- 0 return t, nil } // operateHeader takes action on the decoded headers. func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (close bool) { streamID := frame.Header().StreamID var state decodeState for _, hf := range frame.Fields { if err := state.processHeaderField(hf); err != nil { if se, ok := err.(StreamError); ok { t.controlBuf.put(&resetStream{streamID, statusCodeConvTab[se.Code]}) } return } } buf := newRecvBuffer() s := &Stream{ id: streamID, st: t, buf: buf, fc: &inFlow{limit: uint32(t.initialWindowSize)}, recvCompress: state.encoding, method: state.method, } if frame.StreamEnded() { // s is just created by the caller. No lock needed. s.state = streamReadDone } if state.timeoutSet { s.ctx, s.cancel = context.WithTimeout(t.ctx, state.timeout) } else { s.ctx, s.cancel = context.WithCancel(t.ctx) } pr := &peer.Peer{ Addr: t.remoteAddr, } // Attach Auth info if there is any. if t.authInfo != nil { pr.AuthInfo = t.authInfo } s.ctx = peer.NewContext(s.ctx, pr) // Cache the current stream to the context so that the server application // can find out. Required when the server wants to send some metadata // back to the client (unary call only). s.ctx = newContextWithStream(s.ctx, s) // Attach the received metadata to the context. if len(state.mdata) > 0 { s.ctx = metadata.NewIncomingContext(s.ctx, state.mdata) } if state.statsTags != nil { s.ctx = stats.SetIncomingTags(s.ctx, state.statsTags) } if state.statsTrace != nil { s.ctx = stats.SetIncomingTrace(s.ctx, state.statsTrace) } if t.inTapHandle != nil { var err error info := &tap.Info{ FullMethodName: state.method, } s.ctx, err = t.inTapHandle(s.ctx, info) if err != nil { warningf("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err) t.controlBuf.put(&resetStream{s.id, http2.ErrCodeRefusedStream}) return } } t.mu.Lock() if t.state != reachable { t.mu.Unlock() return } if uint32(len(t.activeStreams)) >= t.maxStreams { t.mu.Unlock() t.controlBuf.put(&resetStream{streamID, http2.ErrCodeRefusedStream}) return } if streamID%2 != 1 || streamID <= t.maxStreamID { t.mu.Unlock() // illegal gRPC stream id. errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID) return true } t.maxStreamID = streamID s.sendQuotaPool = newQuotaPool(int(t.streamSendQuota)) t.activeStreams[streamID] = s if len(t.activeStreams) == 1 { t.idle = time.Time{} } t.mu.Unlock() s.requestRead = func(n int) { t.adjustWindow(s, uint32(n)) } s.ctx = traceCtx(s.ctx, s.method) if t.stats != nil { s.ctx = t.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method}) inHeader := &stats.InHeader{ FullMethod: s.method, RemoteAddr: t.remoteAddr, LocalAddr: t.localAddr, Compression: s.recvCompress, WireLength: int(frame.Header().Length), } t.stats.HandleRPC(s.ctx, inHeader) } s.trReader = &transportReader{ reader: &recvBufferReader{ ctx: s.ctx, recv: s.buf, }, windowHandler: func(n int) { t.updateWindow(s, uint32(n)) }, } handle(s) return } // HandleStreams receives incoming streams using the given handler. This is // typically run in a separate goroutine. // traceCtx attaches trace to ctx and returns the new context. func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.Context, string) context.Context) { // Check the validity of client preface. preface := make([]byte, len(clientPreface)) if _, err := io.ReadFull(t.conn, preface); err != nil { // Only log if it isn't a simple tcp accept check (ie: tcp balancer doing open/close socket) if err != io.EOF { errorf("transport: http2Server.HandleStreams failed to receive the preface from client: %v", err) } t.Close() return } if !bytes.Equal(preface, clientPreface) { errorf("transport: http2Server.HandleStreams received bogus greeting from client: %q", preface) t.Close() return } frame, err := t.framer.readFrame() if err == io.EOF || err == io.ErrUnexpectedEOF { t.Close() return } if err != nil { errorf("transport: http2Server.HandleStreams failed to read initial settings frame: %v", err) t.Close() return } atomic.StoreUint32(&t.activity, 1) sf, ok := frame.(*http2.SettingsFrame) if !ok { errorf("transport: http2Server.HandleStreams saw invalid preface type %T from client", frame) t.Close() return } t.handleSettings(sf) for { frame, err := t.framer.readFrame() atomic.StoreUint32(&t.activity, 1) if err != nil { if se, ok := err.(http2.StreamError); ok { t.mu.Lock() s := t.activeStreams[se.StreamID] t.mu.Unlock() if s != nil { t.closeStream(s) } t.controlBuf.put(&resetStream{se.StreamID, se.Code}) continue } if err == io.EOF || err == io.ErrUnexpectedEOF { t.Close() return } warningf("transport: http2Server.HandleStreams failed to read frame: %v", err) t.Close() return } switch frame := frame.(type) { case *http2.MetaHeadersFrame: if t.operateHeaders(frame, handle, traceCtx) { t.Close() break } case *http2.DataFrame: t.handleData(frame) case *http2.RSTStreamFrame: t.handleRSTStream(frame) case *http2.SettingsFrame: t.handleSettings(frame) case *http2.PingFrame: t.handlePing(frame) case *http2.WindowUpdateFrame: t.handleWindowUpdate(frame) case *http2.GoAwayFrame: // TODO: Handle GoAway from the client appropriately. default: errorf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame) } } } func (t *http2Server) getStream(f http2.Frame) (*Stream, bool) { t.mu.Lock() defer t.mu.Unlock() if t.activeStreams == nil { // The transport is closing. return nil, false } s, ok := t.activeStreams[f.Header().StreamID] if !ok { // The stream is already done. return nil, false } return s, true } // adjustWindow sends out extra window update over the initial window size // of stream if the application is requesting data larger in size than // the window. func (t *http2Server) adjustWindow(s *Stream, n uint32) { s.mu.Lock() defer s.mu.Unlock() if s.state == streamDone { return } if w := s.fc.maybeAdjust(n); w > 0 { if cw := t.fc.resetPendingUpdate(); cw > 0 { t.controlBuf.put(&windowUpdate{0, cw, false}) } t.controlBuf.put(&windowUpdate{s.id, w, true}) } } // updateWindow adjusts the inbound quota for the stream and the transport. // Window updates will deliver to the controller for sending when // the cumulative quota exceeds the corresponding threshold. func (t *http2Server) updateWindow(s *Stream, n uint32) { s.mu.Lock() defer s.mu.Unlock() if s.state == streamDone { return } if w := s.fc.onRead(n); w > 0 { if cw := t.fc.resetPendingUpdate(); cw > 0 { t.controlBuf.put(&windowUpdate{0, cw, false}) } t.controlBuf.put(&windowUpdate{s.id, w, true}) } } // updateFlowControl updates the incoming flow control windows // for the transport and the stream based on the current bdp // estimation. func (t *http2Server) updateFlowControl(n uint32) { t.mu.Lock() for _, s := range t.activeStreams { s.fc.newLimit(n) } t.initialWindowSize = int32(n) t.mu.Unlock() t.controlBuf.put(&windowUpdate{0, t.fc.newLimit(n), false}) t.controlBuf.put(&settings{ ack: false, ss: []http2.Setting{ { ID: http2.SettingInitialWindowSize, Val: uint32(n), }, }, }) } func (t *http2Server) handleData(f *http2.DataFrame) { size := f.Header().Length var sendBDPPing bool if t.bdpEst != nil { sendBDPPing = t.bdpEst.add(uint32(size)) } // Decouple connection's flow control from application's read. // An update on connection's flow control should not depend on // whether user application has read the data or not. Such a // restriction is already imposed on the stream's flow control, // and therefore the sender will be blocked anyways. // Decoupling the connection flow control will prevent other // active(fast) streams from starving in presence of slow or // inactive streams. // // Furthermore, if a bdpPing is being sent out we can piggyback // connection's window update for the bytes we just received. if sendBDPPing { t.controlBuf.put(&windowUpdate{0, uint32(size), false}) t.controlBuf.put(bdpPing) } else { if err := t.fc.onData(uint32(size)); err != nil { errorf("transport: http2Server %v", err) t.Close() return } if w := t.fc.onRead(uint32(size)); w > 0 { t.controlBuf.put(&windowUpdate{0, w, true}) } } // Select the right stream to dispatch. s, ok := t.getStream(f) if !ok { return } if size > 0 { s.mu.Lock() if s.state == streamDone { s.mu.Unlock() return } if err := s.fc.onData(uint32(size)); err != nil { s.mu.Unlock() t.closeStream(s) t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl}) return } if f.Header().Flags.Has(http2.FlagDataPadded) { if w := s.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 { t.controlBuf.put(&windowUpdate{s.id, w, true}) } } s.mu.Unlock() // TODO(bradfitz, zhaoq): A copy is required here because there is no // guarantee f.Data() is consumed before the arrival of next frame. // Can this copy be eliminated? if len(f.Data()) > 0 { data := make([]byte, len(f.Data())) copy(data, f.Data()) s.write(recvMsg{data: data}) } } if f.Header().Flags.Has(http2.FlagDataEndStream) { // Received the end of stream from the client. s.mu.Lock() if s.state != streamDone { s.state = streamReadDone } s.mu.Unlock() s.write(recvMsg{err: io.EOF}) } } func (t *http2Server) handleRSTStream(f *http2.RSTStreamFrame) { s, ok := t.getStream(f) if !ok { return } t.closeStream(s) } func (t *http2Server) handleSettings(f *http2.SettingsFrame) { if f.IsAck() { return } var ss []http2.Setting f.ForeachSetting(func(s http2.Setting) error { ss = append(ss, s) return nil }) // The settings will be applied once the ack is sent. t.controlBuf.put(&settings{ack: true, ss: ss}) } const ( maxPingStrikes = 2 defaultPingTimeout = 2 * time.Hour ) func (t *http2Server) handlePing(f *http2.PingFrame) { if f.IsAck() { if f.Data == goAwayPing.data && t.drainChan != nil { close(t.drainChan) return } // Maybe it's a BDP ping. if t.bdpEst != nil { t.bdpEst.calculate(f.Data) } return } pingAck := &ping{ack: true} copy(pingAck.data[:], f.Data[:]) t.controlBuf.put(pingAck) now := time.Now() defer func() { t.lastPingAt = now }() // A reset ping strikes means that we don't need to check for policy // violation for this ping and the pingStrikes counter should be set // to 0. if atomic.CompareAndSwapUint32(&t.resetPingStrikes, 1, 0) { t.pingStrikes = 0 return } t.mu.Lock() ns := len(t.activeStreams) t.mu.Unlock() if ns < 1 && !t.kep.PermitWithoutStream { // Keepalive shouldn't be active thus, this new ping should // have come after atleast defaultPingTimeout. if t.lastPingAt.Add(defaultPingTimeout).After(now) { t.pingStrikes++ } } else { // Check if keepalive policy is respected. if t.lastPingAt.Add(t.kep.MinTime).After(now) { t.pingStrikes++ } } if t.pingStrikes > maxPingStrikes { // Send goaway and close the connection. t.controlBuf.put(&goAway{code: http2.ErrCodeEnhanceYourCalm, debugData: []byte("too_many_pings"), closeConn: true}) } } func (t *http2Server) handleWindowUpdate(f *http2.WindowUpdateFrame) { id := f.Header().StreamID incr := f.Increment if id == 0 { t.sendQuotaPool.add(int(incr)) return } if s, ok := t.getStream(f); ok { s.sendQuotaPool.add(int(incr)) } } func (t *http2Server) writeHeaders(s *Stream, b *bytes.Buffer, endStream bool) error { first := true endHeaders := false var err error defer func() { if err == nil { // Reset ping strikes when seding headers since that might cause the // peer to send ping. atomic.StoreUint32(&t.resetPingStrikes, 1) } }() // Sends the headers in a single batch. for !endHeaders { size := t.hBuf.Len() if size > http2MaxFrameLen { size = http2MaxFrameLen } else { endHeaders = true } if first { p := http2.HeadersFrameParam{ StreamID: s.id, BlockFragment: b.Next(size), EndStream: endStream, EndHeaders: endHeaders, } err = t.framer.writeHeaders(endHeaders, p) first = false } else { err = t.framer.writeContinuation(endHeaders, s.id, endHeaders, b.Next(size)) } if err != nil { t.Close() return connectionErrorf(true, err, "transport: %v", err) } } return nil } // WriteHeader sends the header metedata md back to the client. func (t *http2Server) WriteHeader(s *Stream, md metadata.MD) error { s.mu.Lock() if s.headerOk || s.state == streamDone { s.mu.Unlock() return ErrIllegalHeaderWrite } s.headerOk = true if md.Len() > 0 { if s.header.Len() > 0 { s.header = metadata.Join(s.header, md) } else { s.header = md } } md = s.header s.mu.Unlock() if _, err := wait(s.ctx, nil, nil, t.shutdownChan, t.writableChan); err != nil { return err } t.hBuf.Reset() t.hEnc.WriteField(hpack.HeaderField{Name: ":status", Value: "200"}) t.hEnc.WriteField(hpack.HeaderField{Name: "content-type", Value: "application/grpc"}) if s.sendCompress != "" { t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-encoding", Value: s.sendCompress}) } for k, vv := range md { if isReservedHeader(k) { // Clients don't tolerate reading restricted headers after some non restricted ones were sent. continue } for _, v := range vv { t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)}) } } bufLen := t.hBuf.Len() if err := t.writeHeaders(s, t.hBuf, false); err != nil { return err } if t.stats != nil { outHeader := &stats.OutHeader{ WireLength: bufLen, } t.stats.HandleRPC(s.Context(), outHeader) } t.writableChan <- 0 return nil } // WriteStatus sends stream status to the client and terminates the stream. // There is no further I/O operations being able to perform on this stream. // TODO(zhaoq): Now it indicates the end of entire stream. Revisit if early // OK is adopted. func (t *http2Server) WriteStatus(s *Stream, st *status.Status) error { var headersSent, hasHeader bool s.mu.Lock() if s.state == streamDone { s.mu.Unlock() return nil } if s.headerOk { headersSent = true } if s.header.Len() > 0 { hasHeader = true } s.mu.Unlock() if !headersSent && hasHeader { t.WriteHeader(s, nil) headersSent = true } // Always write a status regardless of context cancellation unless the stream // is terminated (e.g. by a RST_STREAM, GOAWAY, or transport error). The // server's application code is already done so it is fine to ignore s.ctx. select { case <-t.shutdownChan: return ErrConnClosing case <-t.writableChan: } t.hBuf.Reset() if !headersSent { t.hEnc.WriteField(hpack.HeaderField{Name: ":status", Value: "200"}) t.hEnc.WriteField(hpack.HeaderField{Name: "content-type", Value: "application/grpc"}) } t.hEnc.WriteField( hpack.HeaderField{ Name: "grpc-status", Value: strconv.Itoa(int(st.Code())), }) t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-message", Value: encodeGrpcMessage(st.Message())}) if p := st.Proto(); p != nil && len(p.Details) > 0 { stBytes, err := proto.Marshal(p) if err != nil { // TODO: return error instead, when callers are able to handle it. panic(err) } t.hEnc.WriteField(hpack.HeaderField{Name: "grpc-status-details-bin", Value: encodeBinHeader(stBytes)}) } // Attach the trailer metadata. for k, vv := range s.trailer { // Clients don't tolerate reading restricted headers after some non restricted ones were sent. if isReservedHeader(k) { continue } for _, v := range vv { t.hEnc.WriteField(hpack.HeaderField{Name: k, Value: encodeMetadataHeader(k, v)}) } } bufLen := t.hBuf.Len() if err := t.writeHeaders(s, t.hBuf, true); err != nil { t.Close() return err } if t.stats != nil { outTrailer := &stats.OutTrailer{ WireLength: bufLen, } t.stats.HandleRPC(s.Context(), outTrailer) } t.closeStream(s) t.writableChan <- 0 return nil } // Write converts the data into HTTP2 data frame and sends it out. Non-nil error // is returns if it fails (e.g., framing error, transport error). func (t *http2Server) Write(s *Stream, hdr []byte, data []byte, opts *Options) (err error) { // TODO(zhaoq): Support multi-writers for a single stream. secondStart := http2MaxFrameLen - len(hdr)%http2MaxFrameLen if len(data) < secondStart { secondStart = len(data) } hdr = append(hdr, data[:secondStart]...) data = data[secondStart:] isLastSlice := (len(data) == 0) var writeHeaderFrame bool s.mu.Lock() if s.state == streamDone { s.mu.Unlock() return streamErrorf(codes.Unknown, "the stream has been done") } if !s.headerOk { writeHeaderFrame = true } s.mu.Unlock() if writeHeaderFrame { t.WriteHeader(s, nil) } r := bytes.NewBuffer(hdr) var ( p []byte oqv uint32 ) for { if r.Len() == 0 && p == nil { return nil } oqv = atomic.LoadUint32(&t.outQuotaVersion) size := http2MaxFrameLen // Wait until the stream has some quota to send the data. sq, err := wait(s.ctx, nil, nil, t.shutdownChan, s.sendQuotaPool.acquire()) if err != nil { return err } // Wait until the transport has some quota to send the data. tq, err := wait(s.ctx, nil, nil, t.shutdownChan, t.sendQuotaPool.acquire()) if err != nil { return err } if sq < size { size = sq } if tq < size { size = tq } if p == nil { p = r.Next(size) } ps := len(p) if ps < sq { // Overbooked stream quota. Return it back. s.sendQuotaPool.add(sq - ps) } if ps < tq { // Overbooked transport quota. Return it back. t.sendQuotaPool.add(tq - ps) } t.framer.adjustNumWriters(1) // Got some quota. Try to acquire writing privilege on the // transport. if _, err := wait(s.ctx, nil, nil, t.shutdownChan, t.writableChan); err != nil { if _, ok := err.(StreamError); ok { // Return the connection quota back. t.sendQuotaPool.add(ps) } if t.framer.adjustNumWriters(-1) == 0 { // This writer is the last one in this batch and has the // responsibility to flush the buffered frames. It queues // a flush request to controlBuf instead of flushing directly // in order to avoid the race with other writing or flushing. t.controlBuf.put(&flushIO{}) } return err } select { case <-s.ctx.Done(): t.sendQuotaPool.add(ps) if t.framer.adjustNumWriters(-1) == 0 { t.controlBuf.put(&flushIO{}) } t.writableChan <- 0 return ContextErr(s.ctx.Err()) default: } if oqv != atomic.LoadUint32(&t.outQuotaVersion) { // InitialWindowSize settings frame must have been received after we // acquired send quota but before we got the writable channel. // We must forsake this write. t.sendQuotaPool.add(ps) s.sendQuotaPool.add(ps) if t.framer.adjustNumWriters(-1) == 0 { t.controlBuf.put(&flushIO{}) } t.writableChan <- 0 continue } var forceFlush bool if r.Len() == 0 { if isLastSlice { if t.framer.adjustNumWriters(0) == 1 && !opts.Last { forceFlush = true } } else { r = bytes.NewBuffer(data) isLastSlice = true } } // Reset ping strikes when sending data since this might cause // the peer to send ping. atomic.StoreUint32(&t.resetPingStrikes, 1) if err := t.framer.writeData(forceFlush, s.id, false, p); err != nil { t.Close() return connectionErrorf(true, err, "transport: %v", err) } p = nil if t.framer.adjustNumWriters(-1) == 0 { t.framer.flushWrite() } t.writableChan <- 0 } } func (t *http2Server) applySettings(ss []http2.Setting) { for _, s := range ss { if s.ID == http2.SettingInitialWindowSize { t.mu.Lock() defer t.mu.Unlock() for _, stream := range t.activeStreams { stream.sendQuotaPool.add(int(s.Val) - int(t.streamSendQuota)) } t.streamSendQuota = s.Val atomic.AddUint32(&t.outQuotaVersion, 1) } } } // keepalive running in a separate goroutine does the following: // 1. Gracefully closes an idle connection after a duration of keepalive.MaxConnectionIdle. // 2. Gracefully closes any connection after a duration of keepalive.MaxConnectionAge. // 3. Forcibly closes a connection after an additive period of keepalive.MaxConnectionAgeGrace over keepalive.MaxConnectionAge. // 4. Makes sure a connection is alive by sending pings with a frequency of keepalive.Time and closes a non-responsive connection // after an additional duration of keepalive.Timeout. func (t *http2Server) keepalive() { p := &ping{} var pingSent bool maxIdle := time.NewTimer(t.kp.MaxConnectionIdle) maxAge := time.NewTimer(t.kp.MaxConnectionAge) keepalive := time.NewTimer(t.kp.Time) // NOTE: All exit paths of this function should reset their // respecitve timers. A failure to do so will cause the // following clean-up to deadlock and eventually leak. defer func() { if !maxIdle.Stop() { <-maxIdle.C } if !maxAge.Stop() { <-maxAge.C } if !keepalive.Stop() { <-keepalive.C } }() for { select { case <-maxIdle.C: t.mu.Lock() idle := t.idle if idle.IsZero() { // The connection is non-idle. t.mu.Unlock() maxIdle.Reset(t.kp.MaxConnectionIdle) continue } val := t.kp.MaxConnectionIdle - time.Since(idle) t.mu.Unlock() if val <= 0 { // The connection has been idle for a duration of keepalive.MaxConnectionIdle or more. // Gracefully close the connection. t.drain(http2.ErrCodeNo, []byte{}) // Reseting the timer so that the clean-up doesn't deadlock. maxIdle.Reset(infinity) return } maxIdle.Reset(val) case <-maxAge.C: t.drain(http2.ErrCodeNo, []byte{}) maxAge.Reset(t.kp.MaxConnectionAgeGrace) select { case <-maxAge.C: // Close the connection after grace period. t.Close() // Reseting the timer so that the clean-up doesn't deadlock. maxAge.Reset(infinity) case <-t.shutdownChan: } return case <-keepalive.C: if atomic.CompareAndSwapUint32(&t.activity, 1, 0) { pingSent = false keepalive.Reset(t.kp.Time) continue } if pingSent { t.Close() // Reseting the timer so that the clean-up doesn't deadlock. keepalive.Reset(infinity) return } pingSent = true t.controlBuf.put(p) keepalive.Reset(t.kp.Timeout) case <-t.shutdownChan: return } } } var goAwayPing = &ping{data: [8]byte{1, 6, 1, 8, 0, 3, 3, 9}} // controller running in a separate goroutine takes charge of sending control // frames (e.g., window update, reset stream, setting, etc.) to the server. func (t *http2Server) controller() { for { select { case i := <-t.controlBuf.get(): t.controlBuf.load() select { case <-t.writableChan: switch i := i.(type) { case *windowUpdate: t.framer.writeWindowUpdate(i.flush, i.streamID, i.increment) case *settings: if i.ack { t.framer.writeSettingsAck(true) t.applySettings(i.ss) } else { t.framer.writeSettings(true, i.ss...) } case *resetStream: t.framer.writeRSTStream(true, i.streamID, i.code) case *goAway: t.mu.Lock() if t.state == closing { t.mu.Unlock() // The transport is closing. return } sid := t.maxStreamID if !i.headsUp { // Stop accepting more streams now. t.state = draining activeStreams := len(t.activeStreams) t.mu.Unlock() t.framer.writeGoAway(true, sid, i.code, i.debugData) if i.closeConn || activeStreams == 0 { // Abruptly close the connection following the GoAway. t.Close() } t.writableChan <- 0 continue } t.mu.Unlock() // For a graceful close, send out a GoAway with stream ID of MaxUInt32, // Follow that with a ping and wait for the ack to come back or a timer // to expire. During this time accept new streams since they might have // originated before the GoAway reaches the client. // After getting the ack or timer expiration send out another GoAway this // time with an ID of the max stream server intends to process. t.framer.writeGoAway(true, math.MaxUint32, http2.ErrCodeNo, []byte{}) t.framer.writePing(true, false, goAwayPing.data) go func() { timer := time.NewTimer(time.Minute) defer timer.Stop() select { case <-t.drainChan: case <-timer.C: case <-t.shutdownChan: return } t.controlBuf.put(&goAway{code: i.code, debugData: i.debugData}) }() case *flushIO: t.framer.flushWrite() case *ping: if !i.ack { t.bdpEst.timesnap(i.data) } t.framer.writePing(true, i.ack, i.data) default: errorf("transport: http2Server.controller got unexpected item type %v\n", i) } t.writableChan <- 0 continue case <-t.shutdownChan: return } case <-t.shutdownChan: return } } } // Close starts shutting down the http2Server transport. // TODO(zhaoq): Now the destruction is not blocked on any pending streams. This // could cause some resource issue. Revisit this later. func (t *http2Server) Close() (err error) { t.mu.Lock() if t.state == closing { t.mu.Unlock() return errors.New("transport: Close() was already called") } t.state = closing streams := t.activeStreams t.activeStreams = nil t.mu.Unlock() close(t.shutdownChan) err = t.conn.Close() // Cancel all active streams. for _, s := range streams { s.cancel() } if t.stats != nil { connEnd := &stats.ConnEnd{} t.stats.HandleConn(t.ctx, connEnd) } return } // closeStream clears the footprint of a stream when the stream is not needed // any more. func (t *http2Server) closeStream(s *Stream) { t.mu.Lock() delete(t.activeStreams, s.id) if len(t.activeStreams) == 0 { t.idle = time.Now() } if t.state == draining && len(t.activeStreams) == 0 { defer t.Close() } t.mu.Unlock() // In case stream sending and receiving are invoked in separate // goroutines (e.g., bi-directional streaming), cancel needs to be // called to interrupt the potential blocking on other goroutines. s.cancel() s.mu.Lock() if s.state == streamDone { s.mu.Unlock() return } s.state = streamDone s.mu.Unlock() } func (t *http2Server) RemoteAddr() net.Addr { return t.remoteAddr } func (t *http2Server) Drain() { t.drain(http2.ErrCodeNo, []byte{}) } func (t *http2Server) drain(code http2.ErrCode, debugData []byte) { t.mu.Lock() defer t.mu.Unlock() if t.drainChan != nil { return } t.drainChan = make(chan struct{}) t.controlBuf.put(&goAway{code: code, debugData: debugData, headsUp: true}) } var rgen = rand.New(rand.NewSource(time.Now().UnixNano())) func getJitter(v time.Duration) time.Duration { if v == infinity { return 0 } // Generate a jitter between +/- 10% of the value. r := int64(v / 10) j := rgen.Int63n(2*r) - r return time.Duration(j) }