|
@@ -1,33 +1,18 @@
|
|
|
/*
|
|
/*
|
|
|
*
|
|
*
|
|
|
- * Copyright 2014, Google Inc.
|
|
|
|
|
- * All rights reserved.
|
|
|
|
|
|
|
+ * Copyright 2014 gRPC authors.
|
|
|
*
|
|
*
|
|
|
- * Redistribution and use in source and binary forms, with or without
|
|
|
|
|
- * modification, are permitted provided that the following conditions are
|
|
|
|
|
- * met:
|
|
|
|
|
|
|
+ * 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
|
|
|
*
|
|
*
|
|
|
- * * Redistributions of source code must retain the above copyright
|
|
|
|
|
- * notice, this list of conditions and the following disclaimer.
|
|
|
|
|
- * * Redistributions in binary form must reproduce the above
|
|
|
|
|
- * copyright notice, this list of conditions and the following disclaimer
|
|
|
|
|
- * in the documentation and/or other materials provided with the
|
|
|
|
|
- * distribution.
|
|
|
|
|
- * * Neither the name of Google Inc. nor the names of its
|
|
|
|
|
- * contributors may be used to endorse or promote products derived from
|
|
|
|
|
- * this software without specific prior written permission.
|
|
|
|
|
|
|
+ * http://www.apache.org/licenses/LICENSE-2.0
|
|
|
*
|
|
*
|
|
|
- * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
|
|
|
|
|
- * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
|
|
|
|
|
- * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
|
|
|
|
|
- * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
|
|
|
|
|
- * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
|
|
|
|
|
- * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
|
|
|
|
|
- * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
|
|
|
|
|
- * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
|
|
|
|
|
- * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
|
|
|
|
- * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
|
|
|
|
|
- * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
|
|
|
|
|
|
+ * 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.
|
|
|
*
|
|
*
|
|
|
*/
|
|
*/
|
|
|
|
|
|
|
@@ -51,7 +36,6 @@ import (
|
|
|
"golang.org/x/net/http2/hpack"
|
|
"golang.org/x/net/http2/hpack"
|
|
|
"google.golang.org/grpc/codes"
|
|
"google.golang.org/grpc/codes"
|
|
|
"google.golang.org/grpc/credentials"
|
|
"google.golang.org/grpc/credentials"
|
|
|
- "google.golang.org/grpc/grpclog"
|
|
|
|
|
"google.golang.org/grpc/keepalive"
|
|
"google.golang.org/grpc/keepalive"
|
|
|
"google.golang.org/grpc/metadata"
|
|
"google.golang.org/grpc/metadata"
|
|
|
"google.golang.org/grpc/peer"
|
|
"google.golang.org/grpc/peer"
|
|
@@ -89,7 +73,7 @@ type http2Server struct {
|
|
|
maxStreams uint32
|
|
maxStreams uint32
|
|
|
// controlBuf delivers all the control related tasks (e.g., window
|
|
// controlBuf delivers all the control related tasks (e.g., window
|
|
|
// updates, reset streams, and various settings) to the controller.
|
|
// updates, reset streams, and various settings) to the controller.
|
|
|
- controlBuf *recvBuffer
|
|
|
|
|
|
|
+ controlBuf *controlBuffer
|
|
|
fc *inFlow
|
|
fc *inFlow
|
|
|
// sendQuotaPool provides flow control to outbound message.
|
|
// sendQuotaPool provides flow control to outbound message.
|
|
|
sendQuotaPool *quotaPool
|
|
sendQuotaPool *quotaPool
|
|
@@ -115,6 +99,8 @@ type http2Server struct {
|
|
|
|
|
|
|
|
initialWindowSize int32
|
|
initialWindowSize int32
|
|
|
|
|
|
|
|
|
|
+ bdpEst *bdpEstimator
|
|
|
|
|
+
|
|
|
mu sync.Mutex // guard the following
|
|
mu sync.Mutex // guard the following
|
|
|
state transportState
|
|
state transportState
|
|
|
activeStreams map[uint32]*Stream
|
|
activeStreams map[uint32]*Stream
|
|
@@ -132,32 +118,35 @@ type http2Server struct {
|
|
|
func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) {
|
|
func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err error) {
|
|
|
framer := newFramer(conn)
|
|
framer := newFramer(conn)
|
|
|
// Send initial settings as connection preface to client.
|
|
// Send initial settings as connection preface to client.
|
|
|
- var settings []http2.Setting
|
|
|
|
|
|
|
+ var isettings []http2.Setting
|
|
|
// TODO(zhaoq): Have a better way to signal "no limit" because 0 is
|
|
// TODO(zhaoq): Have a better way to signal "no limit" because 0 is
|
|
|
// permitted in the HTTP2 spec.
|
|
// permitted in the HTTP2 spec.
|
|
|
maxStreams := config.MaxStreams
|
|
maxStreams := config.MaxStreams
|
|
|
if maxStreams == 0 {
|
|
if maxStreams == 0 {
|
|
|
maxStreams = math.MaxUint32
|
|
maxStreams = math.MaxUint32
|
|
|
} else {
|
|
} else {
|
|
|
- settings = append(settings, http2.Setting{
|
|
|
|
|
|
|
+ isettings = append(isettings, http2.Setting{
|
|
|
ID: http2.SettingMaxConcurrentStreams,
|
|
ID: http2.SettingMaxConcurrentStreams,
|
|
|
Val: maxStreams,
|
|
Val: maxStreams,
|
|
|
})
|
|
})
|
|
|
}
|
|
}
|
|
|
|
|
+ dynamicWindow := true
|
|
|
iwz := int32(initialWindowSize)
|
|
iwz := int32(initialWindowSize)
|
|
|
if config.InitialWindowSize >= defaultWindowSize {
|
|
if config.InitialWindowSize >= defaultWindowSize {
|
|
|
iwz = config.InitialWindowSize
|
|
iwz = config.InitialWindowSize
|
|
|
|
|
+ dynamicWindow = false
|
|
|
}
|
|
}
|
|
|
- icwz := int32(initialConnWindowSize)
|
|
|
|
|
|
|
+ icwz := int32(initialWindowSize)
|
|
|
if config.InitialConnWindowSize >= defaultWindowSize {
|
|
if config.InitialConnWindowSize >= defaultWindowSize {
|
|
|
icwz = config.InitialConnWindowSize
|
|
icwz = config.InitialConnWindowSize
|
|
|
|
|
+ dynamicWindow = false
|
|
|
}
|
|
}
|
|
|
if iwz != defaultWindowSize {
|
|
if iwz != defaultWindowSize {
|
|
|
- settings = append(settings, http2.Setting{
|
|
|
|
|
|
|
+ isettings = append(isettings, http2.Setting{
|
|
|
ID: http2.SettingInitialWindowSize,
|
|
ID: http2.SettingInitialWindowSize,
|
|
|
Val: uint32(iwz)})
|
|
Val: uint32(iwz)})
|
|
|
}
|
|
}
|
|
|
- if err := framer.writeSettings(true, settings...); err != nil {
|
|
|
|
|
|
|
+ if err := framer.writeSettings(true, isettings...); err != nil {
|
|
|
return nil, connectionErrorf(true, err, "transport: %v", err)
|
|
return nil, connectionErrorf(true, err, "transport: %v", err)
|
|
|
}
|
|
}
|
|
|
// Adjust the connection flow control window if needed.
|
|
// Adjust the connection flow control window if needed.
|
|
@@ -200,7 +189,7 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
|
|
|
hEnc: hpack.NewEncoder(&buf),
|
|
hEnc: hpack.NewEncoder(&buf),
|
|
|
maxStreams: maxStreams,
|
|
maxStreams: maxStreams,
|
|
|
inTapHandle: config.InTapHandle,
|
|
inTapHandle: config.InTapHandle,
|
|
|
- controlBuf: newRecvBuffer(),
|
|
|
|
|
|
|
+ controlBuf: newControlBuffer(),
|
|
|
fc: &inFlow{limit: uint32(icwz)},
|
|
fc: &inFlow{limit: uint32(icwz)},
|
|
|
sendQuotaPool: newQuotaPool(defaultWindowSize),
|
|
sendQuotaPool: newQuotaPool(defaultWindowSize),
|
|
|
state: reachable,
|
|
state: reachable,
|
|
@@ -214,6 +203,12 @@ func newHTTP2Server(conn net.Conn, config *ServerConfig) (_ ServerTransport, err
|
|
|
kep: kep,
|
|
kep: kep,
|
|
|
initialWindowSize: iwz,
|
|
initialWindowSize: iwz,
|
|
|
}
|
|
}
|
|
|
|
|
+ if dynamicWindow {
|
|
|
|
|
+ t.bdpEst = &bdpEstimator{
|
|
|
|
|
+ bdp: initialWindowSize,
|
|
|
|
|
+ updateFlowControl: t.updateFlowControl,
|
|
|
|
|
+ }
|
|
|
|
|
+ }
|
|
|
if t.stats != nil {
|
|
if t.stats != nil {
|
|
|
t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{
|
|
t.ctx = t.stats.TagConn(t.ctx, &stats.ConnTagInfo{
|
|
|
RemoteAddr: t.remoteAddr,
|
|
RemoteAddr: t.remoteAddr,
|
|
@@ -292,7 +287,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
|
|
|
}
|
|
}
|
|
|
s.ctx, err = t.inTapHandle(s.ctx, info)
|
|
s.ctx, err = t.inTapHandle(s.ctx, info)
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
- grpclog.Printf("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
|
|
|
|
|
|
|
+ warningf("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
|
|
|
t.controlBuf.put(&resetStream{s.id, http2.ErrCodeRefusedStream})
|
|
t.controlBuf.put(&resetStream{s.id, http2.ErrCodeRefusedStream})
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
@@ -310,7 +305,7 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
|
|
|
if s.id%2 != 1 || s.id <= t.maxStreamID {
|
|
if s.id%2 != 1 || s.id <= t.maxStreamID {
|
|
|
t.mu.Unlock()
|
|
t.mu.Unlock()
|
|
|
// illegal gRPC stream id.
|
|
// illegal gRPC stream id.
|
|
|
- grpclog.Println("transport: http2Server.HandleStreams received an illegal stream id: ", s.id)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", s.id)
|
|
|
return true
|
|
return true
|
|
|
}
|
|
}
|
|
|
t.maxStreamID = s.id
|
|
t.maxStreamID = s.id
|
|
@@ -348,13 +343,13 @@ func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.
|
|
|
if _, err := io.ReadFull(t.conn, preface); err != nil {
|
|
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)
|
|
// Only log if it isn't a simple tcp accept check (ie: tcp balancer doing open/close socket)
|
|
|
if err != io.EOF {
|
|
if err != io.EOF {
|
|
|
- grpclog.Printf("transport: http2Server.HandleStreams failed to receive the preface from client: %v", err)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.HandleStreams failed to receive the preface from client: %v", err)
|
|
|
}
|
|
}
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
if !bytes.Equal(preface, clientPreface) {
|
|
if !bytes.Equal(preface, clientPreface) {
|
|
|
- grpclog.Printf("transport: http2Server.HandleStreams received bogus greeting from client: %q", preface)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.HandleStreams received bogus greeting from client: %q", preface)
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
@@ -365,14 +360,14 @@ func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
if err != nil {
|
|
if err != nil {
|
|
|
- grpclog.Printf("transport: http2Server.HandleStreams failed to read initial settings frame: %v", err)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.HandleStreams failed to read initial settings frame: %v", err)
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
atomic.StoreUint32(&t.activity, 1)
|
|
atomic.StoreUint32(&t.activity, 1)
|
|
|
sf, ok := frame.(*http2.SettingsFrame)
|
|
sf, ok := frame.(*http2.SettingsFrame)
|
|
|
if !ok {
|
|
if !ok {
|
|
|
- grpclog.Printf("transport: http2Server.HandleStreams saw invalid preface type %T from client", frame)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.HandleStreams saw invalid preface type %T from client", frame)
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
@@ -396,7 +391,7 @@ func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
- grpclog.Printf("transport: http2Server.HandleStreams failed to read frame: %v", err)
|
|
|
|
|
|
|
+ warningf("transport: http2Server.HandleStreams failed to read frame: %v", err)
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
@@ -419,7 +414,7 @@ func (t *http2Server) HandleStreams(handle func(*Stream), traceCtx func(context.
|
|
|
case *http2.GoAwayFrame:
|
|
case *http2.GoAwayFrame:
|
|
|
// TODO: Handle GoAway from the client appropriately.
|
|
// TODO: Handle GoAway from the client appropriately.
|
|
|
default:
|
|
default:
|
|
|
- grpclog.Printf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.HandleStreams found unhandled frame type %v.", frame)
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
@@ -473,12 +468,34 @@ func (t *http2Server) updateWindow(s *Stream, n uint32) {
|
|
|
}
|
|
}
|
|
|
}
|
|
}
|
|
|
|
|
|
|
|
|
|
+// 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) {
|
|
func (t *http2Server) handleData(f *http2.DataFrame) {
|
|
|
size := f.Header().Length
|
|
size := f.Header().Length
|
|
|
- if err := t.fc.onData(uint32(size)); err != nil {
|
|
|
|
|
- grpclog.Printf("transport: http2Server %v", err)
|
|
|
|
|
- t.Close()
|
|
|
|
|
- return
|
|
|
|
|
|
|
+ var sendBDPPing bool
|
|
|
|
|
+ if t.bdpEst != nil {
|
|
|
|
|
+ sendBDPPing = t.bdpEst.add(uint32(size))
|
|
|
}
|
|
}
|
|
|
// Decouple connection's flow control from application's read.
|
|
// Decouple connection's flow control from application's read.
|
|
|
// An update on connection's flow control should not depend on
|
|
// An update on connection's flow control should not depend on
|
|
@@ -488,8 +505,21 @@ func (t *http2Server) handleData(f *http2.DataFrame) {
|
|
|
// Decoupling the connection flow control will prevent other
|
|
// Decoupling the connection flow control will prevent other
|
|
|
// active(fast) streams from starving in presence of slow or
|
|
// active(fast) streams from starving in presence of slow or
|
|
|
// inactive streams.
|
|
// inactive streams.
|
|
|
- if w := t.fc.onRead(uint32(size)); w > 0 {
|
|
|
|
|
- t.controlBuf.put(&windowUpdate{0, w, true})
|
|
|
|
|
|
|
+ //
|
|
|
|
|
+ // 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.
|
|
// Select the right stream to dispatch.
|
|
|
s, ok := t.getStream(f)
|
|
s, ok := t.getStream(f)
|
|
@@ -561,7 +591,11 @@ const (
|
|
|
)
|
|
)
|
|
|
|
|
|
|
|
func (t *http2Server) handlePing(f *http2.PingFrame) {
|
|
func (t *http2Server) handlePing(f *http2.PingFrame) {
|
|
|
- if f.IsAck() { // Do nothing.
|
|
|
|
|
|
|
+ if f.IsAck() {
|
|
|
|
|
+ // Maybe it's a BDP ping.
|
|
|
|
|
+ if t.bdpEst != nil {
|
|
|
|
|
+ t.bdpEst.calculate(f.Data)
|
|
|
|
|
+ }
|
|
|
return
|
|
return
|
|
|
}
|
|
}
|
|
|
pingAck := &ping{ack: true}
|
|
pingAck := &ping{ack: true}
|
|
@@ -793,13 +827,6 @@ func (t *http2Server) Write(s *Stream, data []byte, opts *Options) (err error) {
|
|
|
if writeHeaderFrame {
|
|
if writeHeaderFrame {
|
|
|
t.WriteHeader(s, nil)
|
|
t.WriteHeader(s, nil)
|
|
|
}
|
|
}
|
|
|
- defer func() {
|
|
|
|
|
- if err == nil {
|
|
|
|
|
- // Reset ping strikes when sending data since this might cause
|
|
|
|
|
- // the peer to send ping.
|
|
|
|
|
- atomic.StoreUint32(&t.resetPingStrikes, 1)
|
|
|
|
|
- }
|
|
|
|
|
- }()
|
|
|
|
|
r := bytes.NewBuffer(data)
|
|
r := bytes.NewBuffer(data)
|
|
|
for {
|
|
for {
|
|
|
if r.Len() == 0 {
|
|
if r.Len() == 0 {
|
|
@@ -863,6 +890,9 @@ func (t *http2Server) Write(s *Stream, data []byte, opts *Options) (err error) {
|
|
|
if r.Len() == 0 && t.framer.adjustNumWriters(0) == 1 && !opts.Last {
|
|
if r.Len() == 0 && t.framer.adjustNumWriters(0) == 1 && !opts.Last {
|
|
|
forceFlush = true
|
|
forceFlush = 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 {
|
|
if err := t.framer.writeData(forceFlush, s.id, false, p); err != nil {
|
|
|
t.Close()
|
|
t.Close()
|
|
|
return connectionErrorf(true, err, "transport: %v", err)
|
|
return connectionErrorf(true, err, "transport: %v", err)
|
|
@@ -1012,9 +1042,12 @@ func (t *http2Server) controller() {
|
|
|
case *flushIO:
|
|
case *flushIO:
|
|
|
t.framer.flushWrite()
|
|
t.framer.flushWrite()
|
|
|
case *ping:
|
|
case *ping:
|
|
|
|
|
+ if !i.ack {
|
|
|
|
|
+ t.bdpEst.timesnap(i.data)
|
|
|
|
|
+ }
|
|
|
t.framer.writePing(true, i.ack, i.data)
|
|
t.framer.writePing(true, i.ack, i.data)
|
|
|
default:
|
|
default:
|
|
|
- grpclog.Printf("transport: http2Server.controller got unexpected item type %v\n", i)
|
|
|
|
|
|
|
+ errorf("transport: http2Server.controller got unexpected item type %v\n", i)
|
|
|
}
|
|
}
|
|
|
t.writableChan <- 0
|
|
t.writableChan <- 0
|
|
|
continue
|
|
continue
|