DevFW-CICD/ingress-nginx-helm
6
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Update go dependencies

Browse source
This commit is contained in:
Manuel de Brito Fontes 2018-05-26 11:27:53 -04:00 • committed by Manuel Alejandro de Brito Fontes
parent 15ffb51394
commit bb4d483837
No known key found for this signature in database
GPG key ID: 786136016A8BA02A
1621 changed files with 86368 additions and 284392 deletions
Download patch file Download diff file Expand all files Collapse all files

9
vendor/google.golang.org/grpc/transport/bdp_estimator.go generated vendored
View file

@ -41,12 +41,9 @@ const (
gamma = 2
)
var (
// Adding arbitrary data to ping so that its ack can be
// identified.
// Easter-egg: what does the ping message say?
bdpPing = &ping{data: [8]byte{2, 4, 16, 16, 9, 14, 7, 7}}
)
// Adding arbitrary data to ping so that its ack can be identified.
// Easter-egg: what does the ping message say?
var bdpPing = &ping{data: [8]byte{2, 4, 16, 16, 9, 14, 7, 7}}
type bdpEstimator struct {
// sentAt is the time when the ping was sent.

769
vendor/google.golang.org/grpc/transport/controlbuf.go generated vendored Normal file
View file

@ -0,0 +1,769 @@
/*
*
* 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"
"fmt"
"runtime"
"sync"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
)
type itemNode struct {
it interface{}
next *itemNode
}
type itemList struct {
head *itemNode
tail *itemNode
}
func (il *itemList) enqueue(i interface{}) {
n := &itemNode{it: i}
if il.tail == nil {
il.head, il.tail = n, n
return
}
il.tail.next = n
il.tail = n
}
// peek returns the first item in the list without removing it from the
// list.
func (il *itemList) peek() interface{} {
return il.head.it
}
func (il *itemList) dequeue() interface{} {
if il.head == nil {
return nil
}
i := il.head.it
il.head = il.head.next
if il.head == nil {
il.tail = nil
}
return i
}
func (il *itemList) dequeueAll() *itemNode {
h := il.head
il.head, il.tail = nil, nil
return h
}
func (il *itemList) isEmpty() bool {
return il.head == nil
}
// The following defines various control items which could flow through
// the control buffer of transport. They represent different aspects of
// control tasks, e.g., flow control, settings, streaming resetting, etc.
type headerFrame struct {
streamID uint32
hf []hpack.HeaderField
endStream bool // Valid on server side.
initStream func(uint32) (bool, error) // Used only on the client side.
onWrite func()
wq *writeQuota // write quota for the stream created.
cleanup *cleanupStream // Valid on the server side.
onOrphaned func(error) // Valid on client-side
}
type cleanupStream struct {
streamID uint32
idPtr *uint32
rst bool
rstCode http2.ErrCode
onWrite func()
}
type dataFrame struct {
streamID uint32
endStream bool
h []byte
d []byte
// onEachWrite is called every time
// a part of d is written out.
onEachWrite func()
}
type incomingWindowUpdate struct {
streamID uint32
increment uint32
}
type outgoingWindowUpdate struct {
streamID uint32
increment uint32
}
type incomingSettings struct {
ss []http2.Setting
}
type outgoingSettings struct {
ss []http2.Setting
}
type settingsAck struct {
}
type incomingGoAway struct {
}
type goAway struct {
code http2.ErrCode
debugData []byte
headsUp bool
closeConn bool
}
type ping struct {
ack bool
data [8]byte
}
type outFlowControlSizeRequest struct {
resp chan uint32
}
type outStreamState int
const (
active outStreamState = iota
empty
waitingOnStreamQuota
)
type outStream struct {
id uint32
state outStreamState
itl *itemList
bytesOutStanding int
wq *writeQuota
next *outStream
prev *outStream
}
func (s *outStream) deleteSelf() {
if s.prev != nil {
s.prev.next = s.next
}
if s.next != nil {
s.next.prev = s.prev
}
s.next, s.prev = nil, nil
}
type outStreamList struct {
// Following are sentinel objects that mark the
// beginning and end of the list. They do not
// contain any item lists. All valid objects are
// inserted in between them.
// This is needed so that an outStream object can
// deleteSelf() in O(1) time without knowing which
// list it belongs to.
head *outStream
tail *outStream
}
func newOutStreamList() *outStreamList {
head, tail := new(outStream), new(outStream)
head.next = tail
tail.prev = head
return &outStreamList{
head: head,
tail: tail,
}
}
func (l *outStreamList) enqueue(s *outStream) {
e := l.tail.prev
e.next = s
s.prev = e
s.next = l.tail
l.tail.prev = s
}
// remove from the beginning of the list.
func (l *outStreamList) dequeue() *outStream {
b := l.head.next
if b == l.tail {
return nil
}
b.deleteSelf()
return b
}
type controlBuffer struct {
ch chan struct{}
done <-chan struct{}
mu sync.Mutex
consumerWaiting bool
list *itemList
err error
}
func newControlBuffer(done <-chan struct{}) *controlBuffer {
return &controlBuffer{
ch: make(chan struct{}, 1),
list: &itemList{},
done: done,
}
}
func (c *controlBuffer) put(it interface{}) error {
_, err := c.executeAndPut(nil, it)
return err
}
func (c *controlBuffer) executeAndPut(f func(it interface{}) bool, it interface{}) (bool, error) {
var wakeUp bool
c.mu.Lock()
if c.err != nil {
c.mu.Unlock()
return false, c.err
}
if f != nil {
if !f(it) { // f wasn't successful
c.mu.Unlock()
return false, nil
}
}
if c.consumerWaiting {
wakeUp = true
c.consumerWaiting = false
}
c.list.enqueue(it)
c.mu.Unlock()
if wakeUp {
select {
case c.ch <- struct{}{}:
default:
}
}
return true, nil
}
func (c *controlBuffer) get(block bool) (interface{}, error) {
for {
c.mu.Lock()
if c.err != nil {
c.mu.Unlock()
return nil, c.err
}
if !c.list.isEmpty() {
h := c.list.dequeue()
c.mu.Unlock()
return h, nil
}
if !block {
c.mu.Unlock()
return nil, nil
}
c.consumerWaiting = true
c.mu.Unlock()
select {
case <-c.ch:
case <-c.done:
c.finish()
return nil, ErrConnClosing
}
}
}
func (c *controlBuffer) finish() {
c.mu.Lock()
if c.err != nil {
c.mu.Unlock()
return
}
c.err = ErrConnClosing
// There may be headers for streams in the control buffer.
// These streams need to be cleaned out since the transport
// is still not aware of these yet.
for head := c.list.dequeueAll(); head != nil; head = head.next {
hdr, ok := head.it.(*headerFrame)
if !ok {
continue
}
if hdr.onOrphaned != nil { // It will be nil on the server-side.
hdr.onOrphaned(ErrConnClosing)
}
}
c.mu.Unlock()
}
type side int
const (
clientSide side = iota
serverSide
)
type loopyWriter struct {
side side
cbuf *controlBuffer
sendQuota uint32
oiws uint32 // outbound initial window size.
estdStreams map[uint32]*outStream // Established streams.
activeStreams *outStreamList // Streams that are sending data.
framer *framer
hBuf *bytes.Buffer // The buffer for HPACK encoding.
hEnc *hpack.Encoder // HPACK encoder.
bdpEst *bdpEstimator
draining bool
// Side-specific handlers
ssGoAwayHandler func(*goAway) (bool, error)
}
func newLoopyWriter(s side, fr *framer, cbuf *controlBuffer, bdpEst *bdpEstimator) *loopyWriter {
var buf bytes.Buffer
l := &loopyWriter{
side: s,
cbuf: cbuf,
sendQuota: defaultWindowSize,
oiws: defaultWindowSize,
estdStreams: make(map[uint32]*outStream),
activeStreams: newOutStreamList(),
framer: fr,
hBuf: &buf,
hEnc: hpack.NewEncoder(&buf),
bdpEst: bdpEst,
}
return l
}
const minBatchSize = 1000
// run should be run in a separate goroutine.
func (l *loopyWriter) run() {
var (
it interface{}
err error
isEmpty bool
)
defer func() {
errorf("transport: loopyWriter.run returning. Err: %v", err)
}()
for {
it, err = l.cbuf.get(true)
if err != nil {
return
}
if err = l.handle(it); err != nil {
return
}
if _, err = l.processData(); err != nil {
return
}
gosched := true
hasdata:
for {
it, err = l.cbuf.get(false)
if err != nil {
return
}
if it != nil {
if err = l.handle(it); err != nil {
return
}
if _, err = l.processData(); err != nil {
return
}
continue hasdata
}
if isEmpty, err = l.processData(); err != nil {
return
}
if !isEmpty {
continue hasdata
}
if gosched {
gosched = false
if l.framer.writer.offset < minBatchSize {
runtime.Gosched()
continue hasdata
}
}
l.framer.writer.Flush()
break hasdata
}
}
}
func (l *loopyWriter) outgoingWindowUpdateHandler(w *outgoingWindowUpdate) error {
return l.framer.fr.WriteWindowUpdate(w.streamID, w.increment)
}
func (l *loopyWriter) incomingWindowUpdateHandler(w *incomingWindowUpdate) error {
// Otherwise update the quota.
if w.streamID == 0 {
l.sendQuota += w.increment
return nil
}
// Find the stream and update it.
if str, ok := l.estdStreams[w.streamID]; ok {
str.bytesOutStanding -= int(w.increment)
if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota > 0 && str.state == waitingOnStreamQuota {
str.state = active
l.activeStreams.enqueue(str)
return nil
}
}
return nil
}
func (l *loopyWriter) outgoingSettingsHandler(s *outgoingSettings) error {
return l.framer.fr.WriteSettings(s.ss...)
}
func (l *loopyWriter) incomingSettingsHandler(s *incomingSettings) error {
if err := l.applySettings(s.ss); err != nil {
return err
}
return l.framer.fr.WriteSettingsAck()
}
func (l *loopyWriter) headerHandler(h *headerFrame) error {
if l.side == serverSide {
if h.endStream { // Case 1.A: Server wants to close stream.
// Make sure it's not a trailers only response.
if str, ok := l.estdStreams[h.streamID]; ok {
if str.state != empty { // either active or waiting on stream quota.
// add it str's list of items.
str.itl.enqueue(h)
return nil
}
}
if err := l.writeHeader(h.streamID, h.endStream, h.hf, h.onWrite); err != nil {
return err
}
return l.cleanupStreamHandler(h.cleanup)
}
// Case 1.B: Server is responding back with headers.
str := &outStream{
state: empty,
itl: &itemList{},
wq: h.wq,
}
l.estdStreams[h.streamID] = str
return l.writeHeader(h.streamID, h.endStream, h.hf, h.onWrite)
}
// Case 2: Client wants to originate stream.
str := &outStream{
id: h.streamID,
state: empty,
itl: &itemList{},
wq: h.wq,
}
str.itl.enqueue(h)
return l.originateStream(str)
}
func (l *loopyWriter) originateStream(str *outStream) error {
hdr := str.itl.dequeue().(*headerFrame)
sendPing, err := hdr.initStream(str.id)
if err != nil {
if err == ErrConnClosing {
return err
}
// Other errors(errStreamDrain) need not close transport.
return nil
}
if err = l.writeHeader(str.id, hdr.endStream, hdr.hf, hdr.onWrite); err != nil {
return err
}
l.estdStreams[str.id] = str
if sendPing {
return l.pingHandler(&ping{data: [8]byte{}})
}
return nil
}
func (l *loopyWriter) writeHeader(streamID uint32, endStream bool, hf []hpack.HeaderField, onWrite func()) error {
if onWrite != nil {
onWrite()
}
l.hBuf.Reset()
for _, f := range hf {
if err := l.hEnc.WriteField(f); err != nil {
warningf("transport: loopyWriter.writeHeader encountered error while encoding headers:", err)
}
}
var (
err error
endHeaders, first bool
)
first = true
for !endHeaders {
size := l.hBuf.Len()
if size > http2MaxFrameLen {
size = http2MaxFrameLen
} else {
endHeaders = true
}
if first {
first = false
err = l.framer.fr.WriteHeaders(http2.HeadersFrameParam{
StreamID: streamID,
BlockFragment: l.hBuf.Next(size),
EndStream: endStream,
EndHeaders: endHeaders,
})
} else {
err = l.framer.fr.WriteContinuation(
streamID,
endHeaders,
l.hBuf.Next(size),
)
}
if err != nil {
return err
}
}
return nil
}
func (l *loopyWriter) preprocessData(df *dataFrame) error {
str, ok := l.estdStreams[df.streamID]
if !ok {
return nil
}
// If we got data for a stream it means that
// stream was originated and the headers were sent out.
str.itl.enqueue(df)
if str.state == empty {
str.state = active
l.activeStreams.enqueue(str)
}
return nil
}
func (l *loopyWriter) pingHandler(p *ping) error {
if !p.ack {
l.bdpEst.timesnap(p.data)
}
return l.framer.fr.WritePing(p.ack, p.data)
}
func (l *loopyWriter) outFlowControlSizeRequestHandler(o *outFlowControlSizeRequest) error {
o.resp <- l.sendQuota
return nil
}
func (l *loopyWriter) cleanupStreamHandler(c *cleanupStream) error {
c.onWrite()
if str, ok := l.estdStreams[c.streamID]; ok {
// On the server side it could be a trailers-only response or
// a RST_STREAM before stream initialization thus the stream might
// not be established yet.
delete(l.estdStreams, c.streamID)
str.deleteSelf()
}
if c.rst { // If RST_STREAM needs to be sent.
if err := l.framer.fr.WriteRSTStream(c.streamID, c.rstCode); err != nil {
return err
}
}
if l.side == clientSide && l.draining && len(l.estdStreams) == 0 {
return ErrConnClosing
}
return nil
}
func (l *loopyWriter) incomingGoAwayHandler(*incomingGoAway) error {
if l.side == clientSide {
l.draining = true
if len(l.estdStreams) == 0 {
return ErrConnClosing
}
}
return nil
}
func (l *loopyWriter) goAwayHandler(g *goAway) error {
// Handling of outgoing GoAway is very specific to side.
if l.ssGoAwayHandler != nil {
draining, err := l.ssGoAwayHandler(g)
if err != nil {
return err
}
l.draining = draining
}
return nil
}
func (l *loopyWriter) handle(i interface{}) error {
switch i := i.(type) {
case *incomingWindowUpdate:
return l.incomingWindowUpdateHandler(i)
case *outgoingWindowUpdate:
return l.outgoingWindowUpdateHandler(i)
case *incomingSettings:
return l.incomingSettingsHandler(i)
case *outgoingSettings:
return l.outgoingSettingsHandler(i)
case *headerFrame:
return l.headerHandler(i)
case *cleanupStream:
return l.cleanupStreamHandler(i)
case *incomingGoAway:
return l.incomingGoAwayHandler(i)
case *dataFrame:
return l.preprocessData(i)
case *ping:
return l.pingHandler(i)
case *goAway:
return l.goAwayHandler(i)
case *outFlowControlSizeRequest:
return l.outFlowControlSizeRequestHandler(i)
default:
return fmt.Errorf("transport: unknown control message type %T", i)
}
}
func (l *loopyWriter) applySettings(ss []http2.Setting) error {
for _, s := range ss {
switch s.ID {
case http2.SettingInitialWindowSize:
o := l.oiws
l.oiws = s.Val
if o < l.oiws {
// If the new limit is greater make all depleted streams active.
for _, stream := range l.estdStreams {
if stream.state == waitingOnStreamQuota {
stream.state = active
l.activeStreams.enqueue(stream)
}
}
}
}
}
return nil
}
func (l *loopyWriter) processData() (bool, error) {
if l.sendQuota == 0 {
return true, nil
}
str := l.activeStreams.dequeue()
if str == nil {
return true, nil
}
dataItem := str.itl.peek().(*dataFrame)
if len(dataItem.h) == 0 && len(dataItem.d) == 0 {
// Client sends out empty data frame with endStream = true
if err := l.framer.fr.WriteData(dataItem.streamID, dataItem.endStream, nil); err != nil {
return false, err
}
str.itl.dequeue()
if str.itl.isEmpty() {
str.state = empty
} else if trailer, ok := str.itl.peek().(*headerFrame); ok { // the next item is trailers.
if err := l.writeHeader(trailer.streamID, trailer.endStream, trailer.hf, trailer.onWrite); err != nil {
return false, err
}
if err := l.cleanupStreamHandler(trailer.cleanup); err != nil {
return false, nil
}
} else {
l.activeStreams.enqueue(str)
}
return false, nil
}
var (
idx int
buf []byte
)
if len(dataItem.h) != 0 { // data header has not been written out yet.
buf = dataItem.h
} else {
idx = 1
buf = dataItem.d
}
size := http2MaxFrameLen
if len(buf) < size {
size = len(buf)
}
if strQuota := int(l.oiws) - str.bytesOutStanding; strQuota <= 0 {
str.state = waitingOnStreamQuota
return false, nil
} else if strQuota < size {
size = strQuota
}
if l.sendQuota < uint32(size) {
size = int(l.sendQuota)
}
// Now that outgoing flow controls are checked we can replenish str's write quota
str.wq.replenish(size)
var endStream bool
// This last data message on this stream and all
// of it can be written in this go.
if dataItem.endStream && size == len(buf) {
// buf contains either data or it contains header but data is empty.
if idx == 1 || len(dataItem.d) == 0 {
endStream = true
}
}
if dataItem.onEachWrite != nil {
dataItem.onEachWrite()
}
if err := l.framer.fr.WriteData(dataItem.streamID, endStream, buf[:size]); err != nil {
return false, err
}
buf = buf[size:]
str.bytesOutStanding += size
l.sendQuota -= uint32(size)
if idx == 0 {
dataItem.h = buf
} else {
dataItem.d = buf
}
if len(dataItem.h) == 0 && len(dataItem.d) == 0 { // All the data from that message was written out.
str.itl.dequeue()
}
if str.itl.isEmpty() {
str.state = empty
} else if trailer, ok := str.itl.peek().(*headerFrame); ok { // The next item is trailers.
if err := l.writeHeader(trailer.streamID, trailer.endStream, trailer.hf, trailer.onWrite); err != nil {
return false, err
}
if err := l.cleanupStreamHandler(trailer.cleanup); err != nil {
return false, err
}
} else if int(l.oiws)-str.bytesOutStanding <= 0 { // Ran out of stream quota.
str.state = waitingOnStreamQuota
} else { // Otherwise add it back to the list of active streams.
l.activeStreams.enqueue(str)
}
return false, nil
}

265
vendor/google.golang.org/grpc/transport/control.go → vendor/google.golang.org/grpc/transport/flowcontrol.go generated vendored
View file

@ -24,9 +24,6 @@ import (
"sync"
"sync/atomic"
"time"
"golang.org/x/net/http2"
"golang.org/x/net/http2/hpack"
)
const (
@ -36,179 +33,109 @@ const (
initialWindowSize = defaultWindowSize // for an RPC
infinity = time.Duration(math.MaxInt64)
defaultClientKeepaliveTime = infinity
defaultClientKeepaliveTimeout = time.Duration(20 * time.Second)
defaultClientKeepaliveTimeout = 20 * time.Second
defaultMaxStreamsClient = 100
defaultMaxConnectionIdle = infinity
defaultMaxConnectionAge = infinity
defaultMaxConnectionAgeGrace = infinity
defaultServerKeepaliveTime = time.Duration(2 * time.Hour)
defaultServerKeepaliveTimeout = time.Duration(20 * time.Second)
defaultKeepalivePolicyMinTime = time.Duration(5 * time.Minute)
defaultServerKeepaliveTime = 2 * time.Hour
defaultServerKeepaliveTimeout = 20 * time.Second
defaultKeepalivePolicyMinTime = 5 * time.Minute
// max window limit set by HTTP2 Specs.
maxWindowSize = math.MaxInt32
// defaultLocalSendQuota sets is default value for number of data
// defaultWriteQuota is the default value for number of data
// bytes that each stream can schedule before some of it being
// flushed out.
defaultLocalSendQuota = 64 * 1024
defaultWriteQuota = 64 * 1024
)
// The following defines various control items which could flow through
// the control buffer of transport. They represent different aspects of
// control tasks, e.g., flow control, settings, streaming resetting, etc.
type headerFrame struct {
streamID uint32
hf []hpack.HeaderField
endStream bool
// writeQuota is a soft limit on the amount of data a stream can
// schedule before some of it is written out.
type writeQuota struct {
quota int32
// get waits on read from when quota goes less than or equal to zero.
// replenish writes on it when quota goes positive again.
ch chan struct{}
// done is triggered in error case.
done <-chan struct{}
}
func (*headerFrame) item() {}
type continuationFrame struct {
streamID uint32
endHeaders bool
headerBlockFragment []byte
}
type dataFrame struct {
streamID uint32
endStream bool
d []byte
f func()
}
func (*dataFrame) item() {}
func (*continuationFrame) item() {}
type windowUpdate struct {
streamID uint32
increment uint32
}
func (*windowUpdate) item() {}
type settings struct {
ack bool
ss []http2.Setting
}
func (*settings) item() {}
type resetStream struct {
streamID uint32
code http2.ErrCode
}
func (*resetStream) item() {}
type goAway struct {
code http2.ErrCode
debugData []byte
headsUp bool
closeConn bool
}
func (*goAway) item() {}
type flushIO struct {
}
func (*flushIO) item() {}
type ping struct {
ack bool
data [8]byte
}
func (*ping) item() {}
// quotaPool is a pool which accumulates the quota and sends it to acquire()
// when it is available.
type quotaPool struct {
c chan int
mu sync.Mutex
version uint32
quota int
}
// newQuotaPool creates a quotaPool which has quota q available to consume.
func newQuotaPool(q int) *quotaPool {
qb := &quotaPool{
c: make(chan int, 1),
}
if q > 0 {
qb.c <- q
} else {
qb.quota = q
}
return qb
}
// add cancels the pending quota sent on acquired, incremented by v and sends
// it back on acquire.
func (qb *quotaPool) add(v int) {
qb.mu.Lock()
defer qb.mu.Unlock()
qb.lockedAdd(v)
}
func (qb *quotaPool) lockedAdd(v int) {
select {
case n := <-qb.c:
qb.quota += n
default:
}
qb.quota += v
if qb.quota <= 0 {
return
}
// After the pool has been created, this is the only place that sends on
// the channel. Since mu is held at this point and any quota that was sent
// on the channel has been retrieved, we know that this code will always
// place any positive quota value on the channel.
select {
case qb.c <- qb.quota:
qb.quota = 0
default:
func newWriteQuota(sz int32, done <-chan struct{}) *writeQuota {
return &writeQuota{
quota: sz,
ch: make(chan struct{}, 1),
done: done,
}
}
func (qb *quotaPool) addAndUpdate(v int) {
qb.mu.Lock()
defer qb.mu.Unlock()
qb.lockedAdd(v)
// Update the version only after having added to the quota
// so that if acquireWithVesrion sees the new vesrion it is
// guaranteed to have seen the updated quota.
// Also, still keep this inside of the lock, so that when
// compareAndExecute is processing, this function doesn't
// get executed partially (quota gets updated but the version
// doesn't).
atomic.AddUint32(&(qb.version), 1)
}
func (qb *quotaPool) acquireWithVersion() (<-chan int, uint32) {
return qb.c, atomic.LoadUint32(&(qb.version))
}
func (qb *quotaPool) compareAndExecute(version uint32, success, failure func()) bool {
qb.mu.Lock()
defer qb.mu.Unlock()
if version == atomic.LoadUint32(&(qb.version)) {
success()
return true
func (w *writeQuota) get(sz int32) error {
for {
if atomic.LoadInt32(&w.quota) > 0 {
atomic.AddInt32(&w.quota, -sz)
return nil
}
select {
case <-w.ch:
continue
case <-w.done:
return errStreamDone
}
}
failure()
return false
}
// acquire returns the channel on which available quota amounts are sent.
func (qb *quotaPool) acquire() <-chan int {
return qb.c
func (w *writeQuota) replenish(n int) {
sz := int32(n)
a := atomic.AddInt32(&w.quota, sz)
b := a - sz
if b <= 0 && a > 0 {
select {
case w.ch <- struct{}{}:
default:
}
}
}
type trInFlow struct {
limit uint32
unacked uint32
effectiveWindowSize uint32
}
func (f *trInFlow) newLimit(n uint32) uint32 {
d := n - f.limit
f.limit = n
f.updateEffectiveWindowSize()
return d
}
func (f *trInFlow) onData(n uint32) uint32 {
f.unacked += n
if f.unacked >= f.limit/4 {
w := f.unacked
f.unacked = 0
f.updateEffectiveWindowSize()
return w
}
f.updateEffectiveWindowSize()
return 0
}
func (f *trInFlow) reset() uint32 {
w := f.unacked
f.unacked = 0
f.updateEffectiveWindowSize()
return w
}
func (f *trInFlow) updateEffectiveWindowSize() {
atomic.StoreUint32(&f.effectiveWindowSize, f.limit-f.unacked)
}
func (f *trInFlow) getSize() uint32 {
return atomic.LoadUint32(&f.effectiveWindowSize)
}
// TODO(mmukhi): Simplify this code.
// inFlow deals with inbound flow control
type inFlow struct {
mu sync.Mutex
@ -229,9 +156,9 @@ type inFlow struct {
// It assumes that n is always greater than the old limit.
func (f *inFlow) newLimit(n uint32) uint32 {
f.mu.Lock()
defer f.mu.Unlock()
d := n - f.limit
f.limit = n
f.mu.Unlock()
return d
}
@ -240,7 +167,6 @@ func (f *inFlow) maybeAdjust(n uint32) uint32 {
n = uint32(math.MaxInt32)
}
f.mu.Lock()
defer f.mu.Unlock()
// estSenderQuota is the receiver's view of the maximum number of bytes the sender
// can send without a window update.
estSenderQuota := int32(f.limit - (f.pendingData + f.pendingUpdate))
@ -252,7 +178,7 @@ func (f *inFlow) maybeAdjust(n uint32) uint32 {
// for this message. Therefore we must send an update over the limit since there's an active read
// request from the application.
if estUntransmittedData > estSenderQuota {
// Sender's window shouldn't go more than 2^31 - 1 as speecified in the HTTP spec.
// Sender's window shouldn't go more than 2^31 - 1 as specified in the HTTP spec.
if f.limit+n > maxWindowSize {
f.delta = maxWindowSize - f.limit
} else {
@ -261,19 +187,24 @@ func (f *inFlow) maybeAdjust(n uint32) uint32 {
// is padded; We will fallback on the current available window(at least a 1/4th of the limit).
f.delta = n
}
f.mu.Unlock()
return f.delta
}
f.mu.Unlock()
return 0
}
// onData is invoked when some data frame is received. It updates pendingData.
func (f *inFlow) onData(n uint32) error {
f.mu.Lock()
defer f.mu.Unlock()
f.pendingData += n
if f.pendingData+f.pendingUpdate > f.limit+f.delta {
return fmt.Errorf("received %d-bytes data exceeding the limit %d bytes", f.pendingData+f.pendingUpdate, f.limit)
limit := f.limit
rcvd := f.pendingData + f.pendingUpdate
f.mu.Unlock()
return fmt.Errorf("received %d-bytes data exceeding the limit %d bytes", rcvd, limit)
}
f.mu.Unlock()
return nil
}
@ -281,8 +212,8 @@ func (f *inFlow) onData(n uint32) error {
// to be sent to the peer.
func (f *inFlow) onRead(n uint32) uint32 {
f.mu.Lock()
defer f.mu.Unlock()
if f.pendingData == 0 {
f.mu.Unlock()
return 0
}
f.pendingData -= n
@ -297,15 +228,9 @@ func (f *inFlow) onRead(n uint32) uint32 {
if f.pendingUpdate >= f.limit/4 {
wu := f.pendingUpdate
f.pendingUpdate = 0
f.mu.Unlock()
return wu
}
f.mu.Unlock()
return 0
}
func (f *inFlow) resetPendingUpdate() uint32 {
f.mu.Lock()
defer f.mu.Unlock()
n := f.pendingUpdate
f.pendingUpdate = 0
return n
}

51
vendor/google.golang.org/grpc/transport/go16.go generated vendored Normal file
View file

@ -0,0 +1,51 @@
// +build go1.6,!go1.7
/*
*
* Copyright 2016 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 (
"net"
"net/http"
"google.golang.org/grpc/codes"
"golang.org/x/net/context"
)
// dialContext connects to the address on the named network.
func dialContext(ctx context.Context, network, address string) (net.Conn, error) {
return (&net.Dialer{Cancel: ctx.Done()}).Dial(network, address)
}
// ContextErr converts the error from context package into a StreamError.
func ContextErr(err error) StreamError {
switch err {
case context.DeadlineExceeded:
return streamErrorf(codes.DeadlineExceeded, "%v", err)
case context.Canceled:
return streamErrorf(codes.Canceled, "%v", err)
}
return streamErrorf(codes.Internal, "Unexpected error from context packet: %v", err)
}
// contextFromRequest returns a background context.
func contextFromRequest(r *http.Request) context.Context {
return context.Background()
}

52
vendor/google.golang.org/grpc/transport/go17.go generated vendored Normal file
View file

@ -0,0 +1,52 @@
// +build go1.7
/*
*
* Copyright 2016 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 (
"context"
"net"
"net/http"
"google.golang.org/grpc/codes"
netctx "golang.org/x/net/context"
)
// dialContext connects to the address on the named network.
func dialContext(ctx context.Context, network, address string) (net.Conn, error) {
return (&net.Dialer{}).DialContext(ctx, network, address)
}
// ContextErr converts the error from context package into a StreamError.
func ContextErr(err error) StreamError {
switch err {
case context.DeadlineExceeded, netctx.DeadlineExceeded:
return streamErrorf(codes.DeadlineExceeded, "%v", err)
case context.Canceled, netctx.Canceled:
return streamErrorf(codes.Canceled, "%v", err)
}
return streamErrorf(codes.Internal, "Unexpected error from context packet: %v", err)
}
// contextFromRequest returns a context from the HTTP Request.
func contextFromRequest(r *http.Request) context.Context {
return r.Context()
}

109
vendor/google.golang.org/grpc/transport/handler_server.go generated vendored
View file

@ -40,20 +40,24 @@ import (
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/peer"
"google.golang.org/grpc/stats"
"google.golang.org/grpc/status"
)
// NewServerHandlerTransport returns a ServerTransport handling gRPC
// from inside an http.Handler. It requires that the http Server
// supports HTTP/2.
func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTransport, error) {
func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request, stats stats.Handler) (ServerTransport, error) {
if r.ProtoMajor != 2 {
return nil, errors.New("gRPC requires HTTP/2")
}
if r.Method != "POST" {
return nil, errors.New("invalid gRPC request method")
}
if !validContentType(r.Header.Get("Content-Type")) {
contentType := r.Header.Get("Content-Type")
// TODO: do we assume contentType is lowercase? we did before
contentSubtype, validContentType := contentSubtype(contentType)
if !validContentType {
return nil, errors.New("invalid gRPC request content-type")
}
if _, ok := w.(http.Flusher); !ok {
@ -64,10 +68,13 @@ func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTr
}
st := &serverHandlerTransport{
rw: w,
req: r,
closedCh: make(chan struct{}),
writes: make(chan func()),
rw: w,
req: r,
closedCh: make(chan struct{}),
writes: make(chan func()),
contentType: contentType,
contentSubtype: contentSubtype,
stats: stats,
}
if v := r.Header.Get("grpc-timeout"); v != "" {
@ -79,19 +86,19 @@ func NewServerHandlerTransport(w http.ResponseWriter, r *http.Request) (ServerTr
st.timeout = to
}
var metakv []string
metakv := []string{"content-type", contentType}
if r.Host != "" {
metakv = append(metakv, ":authority", r.Host)
}
for k, vv := range r.Header {
k = strings.ToLower(k)
if isReservedHeader(k) && !isWhitelistedPseudoHeader(k) {
if isReservedHeader(k) && !isWhitelistedHeader(k) {
continue
}
for _, v := range vv {
v, err := decodeMetadataHeader(k, v)
if err != nil {
return nil, streamErrorf(codes.InvalidArgument, "malformed binary metadata: %v", err)
return nil, streamErrorf(codes.Internal, "malformed binary metadata: %v", err)
}
metakv = append(metakv, k, v)
}
@ -123,10 +130,17 @@ type serverHandlerTransport struct {
// when WriteStatus is called.
writes chan func()
mu sync.Mutex
// streamDone indicates whether WriteStatus has been called and writes channel
// has been closed.
streamDone bool
// block concurrent WriteStatus calls
// e.g. grpc/(*serverStream).SendMsg/RecvMsg
writeStatusMu sync.Mutex
// we just mirror the request content-type
contentType string
// we store both contentType and contentSubtype so we don't keep recreating them
// TODO make sure this is consistent across handler_server and http2_server
contentSubtype string
stats stats.Handler
}
func (ht *serverHandlerTransport) Close() error {
@ -177,13 +191,9 @@ func (ht *serverHandlerTransport) do(fn func()) error {
}
func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status) error {
ht.mu.Lock()
if ht.streamDone {
ht.mu.Unlock()
return nil
}
ht.streamDone = true
ht.mu.Unlock()
ht.writeStatusMu.Lock()
defer ht.writeStatusMu.Unlock()
err := ht.do(func() {
ht.writeCommonHeaders(s)
@ -222,7 +232,14 @@ func (ht *serverHandlerTransport) WriteStatus(s *Stream, st *status.Status) erro
}
}
})
close(ht.writes)
if err == nil { // transport has not been closed
if ht.stats != nil {
ht.stats.HandleRPC(s.Context(), &stats.OutTrailer{})
}
ht.Close()
close(ht.writes)
}
return err
}
@ -236,7 +253,7 @@ func (ht *serverHandlerTransport) writeCommonHeaders(s *Stream) {
h := ht.rw.Header()
h["Date"] = nil // suppress Date to make tests happy; TODO: restore
h.Set("Content-Type", "application/grpc")
h.Set("Content-Type", ht.contentType)
// Predeclare trailers we'll set later in WriteStatus (after the body).
// This is a SHOULD in the HTTP RFC, and the way you add (known)
@ -264,7 +281,7 @@ func (ht *serverHandlerTransport) Write(s *Stream, hdr []byte, data []byte, opts
}
func (ht *serverHandlerTransport) WriteHeader(s *Stream, md metadata.MD) error {
return ht.do(func() {
err := ht.do(func() {
ht.writeCommonHeaders(s)
h := ht.rw.Header()
for k, vv := range md {
@ -280,17 +297,24 @@ func (ht *serverHandlerTransport) WriteHeader(s *Stream, md metadata.MD) error {
ht.rw.WriteHeader(200)
ht.rw.(http.Flusher).Flush()
})
if err == nil {
if ht.stats != nil {
ht.stats.HandleRPC(s.Context(), &stats.OutHeader{})
}
}
return err
}
func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), traceCtx func(context.Context, string) context.Context) {
// With this transport type there will be exactly 1 stream: this HTTP request.
var ctx context.Context
ctx := contextFromRequest(ht.req)
var cancel context.CancelFunc
if ht.timeoutSet {
ctx, cancel = context.WithTimeout(context.Background(), ht.timeout)
ctx, cancel = context.WithTimeout(ctx, ht.timeout)
} else {
ctx, cancel = context.WithCancel(context.Background())
ctx, cancel = context.WithCancel(ctx)
}
// requestOver is closed when either the request's context is done
@ -314,13 +338,14 @@ func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), trace
req := ht.req
s := &Stream{
id: 0, // irrelevant
requestRead: func(int) {},
cancel: cancel,
buf: newRecvBuffer(),
st: ht,
method: req.URL.Path,
recvCompress: req.Header.Get("grpc-encoding"),
id: 0, // irrelevant
requestRead: func(int) {},
cancel: cancel,
buf: newRecvBuffer(),
st: ht,
method: req.URL.Path,
recvCompress: req.Header.Get("grpc-encoding"),
contentSubtype: ht.contentSubtype,
}
pr := &peer.Peer{
Addr: ht.RemoteAddr(),
@ -329,10 +354,18 @@ func (ht *serverHandlerTransport) HandleStreams(startStream func(*Stream), trace
pr.AuthInfo = credentials.TLSInfo{State: *req.TLS}
}
ctx = metadata.NewIncomingContext(ctx, ht.headerMD)
ctx = peer.NewContext(ctx, pr)
s.ctx = newContextWithStream(ctx, s)
s.ctx = peer.NewContext(ctx, pr)
if ht.stats != nil {
s.ctx = ht.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
inHeader := &stats.InHeader{
FullMethod: s.method,
RemoteAddr: ht.RemoteAddr(),
Compression: s.recvCompress,
}
ht.stats.HandleRPC(s.ctx, inHeader)
}
s.trReader = &transportReader{
reader: &recvBufferReader{ctx: s.ctx, recv: s.buf},
reader: &recvBufferReader{ctx: s.ctx, ctxDone: s.ctx.Done(), recv: s.buf},
windowHandler: func(int) {},
}
@ -387,6 +420,10 @@ func (ht *serverHandlerTransport) runStream() {
}
}
func (ht *serverHandlerTransport) IncrMsgSent() {}
func (ht *serverHandlerTransport) IncrMsgRecv() {}
func (ht *serverHandlerTransport) Drain() {
panic("Drain() is not implemented")
}

462
vendor/google.golang.org/grpc/transport/handler_server_test.go generated vendored
View file

@ -1,462 +0,0 @@
/*
*
* Copyright 2016 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 (
"errors"
"fmt"
"io"
"net/http"
"net/http/httptest"
"net/url"
"reflect"
"sync"
"testing"
"time"
"github.com/golang/protobuf/proto"
dpb "github.com/golang/protobuf/ptypes/duration"
"golang.org/x/net/context"
epb "google.golang.org/genproto/googleapis/rpc/errdetails"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/metadata"
"google.golang.org/grpc/status"
)
func TestHandlerTransport_NewServerHandlerTransport(t *testing.T) {
type testCase struct {
name string
req *http.Request
wantErr string
modrw func(http.ResponseWriter) http.ResponseWriter
check func(*serverHandlerTransport, *testCase) error
}
tests := []testCase{
{
name: "http/1.1",
req: &http.Request{
ProtoMajor: 1,
ProtoMinor: 1,
},
wantErr: "gRPC requires HTTP/2",
},
{
name: "bad method",
req: &http.Request{
ProtoMajor: 2,
Method: "GET",
Header: http.Header{},
RequestURI: "/",
},
wantErr: "invalid gRPC request method",
},
{
name: "bad content type",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": {"application/foo"},
},
RequestURI: "/service/foo.bar",
},
wantErr: "invalid gRPC request content-type",
},
{
name: "not flusher",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": {"application/grpc"},
},
RequestURI: "/service/foo.bar",
},
modrw: func(w http.ResponseWriter) http.ResponseWriter {
// Return w without its Flush method
type onlyCloseNotifier interface {
http.ResponseWriter
http.CloseNotifier
}
return struct{ onlyCloseNotifier }{w.(onlyCloseNotifier)}
},
wantErr: "gRPC requires a ResponseWriter supporting http.Flusher",
},
{
name: "not closenotifier",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": {"application/grpc"},
},
RequestURI: "/service/foo.bar",
},
modrw: func(w http.ResponseWriter) http.ResponseWriter {
// Return w without its CloseNotify method
type onlyFlusher interface {
http.ResponseWriter
http.Flusher
}
return struct{ onlyFlusher }{w.(onlyFlusher)}
},
wantErr: "gRPC requires a ResponseWriter supporting http.CloseNotifier",
},
{
name: "valid",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": {"application/grpc"},
},
URL: &url.URL{
Path: "/service/foo.bar",
},
RequestURI: "/service/foo.bar",
},
check: func(t *serverHandlerTransport, tt *testCase) error {
if t.req != tt.req {
return fmt.Errorf("t.req = %p; want %p", t.req, tt.req)
}
if t.rw == nil {
return errors.New("t.rw = nil; want non-nil")
}
return nil
},
},
{
name: "with timeout",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": []string{"application/grpc"},
"Grpc-Timeout": {"200m"},
},
URL: &url.URL{
Path: "/service/foo.bar",
},
RequestURI: "/service/foo.bar",
},
check: func(t *serverHandlerTransport, tt *testCase) error {
if !t.timeoutSet {
return errors.New("timeout not set")
}
if want := 200 * time.Millisecond; t.timeout != want {
return fmt.Errorf("timeout = %v; want %v", t.timeout, want)
}
return nil
},
},
{
name: "with bad timeout",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": []string{"application/grpc"},
"Grpc-Timeout": {"tomorrow"},
},
URL: &url.URL{
Path: "/service/foo.bar",
},
RequestURI: "/service/foo.bar",
},
wantErr: `stream error: code = Internal desc = "malformed time-out: transport: timeout unit is not recognized: \"tomorrow\""`,
},
{
name: "with metadata",
req: &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": []string{"application/grpc"},
"meta-foo": {"foo-val"},
"meta-bar": {"bar-val1", "bar-val2"},
"user-agent": {"x/y a/b"},
},
URL: &url.URL{
Path: "/service/foo.bar",
},
RequestURI: "/service/foo.bar",
},
check: func(ht *serverHandlerTransport, tt *testCase) error {
want := metadata.MD{
"meta-bar": {"bar-val1", "bar-val2"},
"user-agent": {"x/y a/b"},
"meta-foo": {"foo-val"},
}
if !reflect.DeepEqual(ht.headerMD, want) {
return fmt.Errorf("metdata = %#v; want %#v", ht.headerMD, want)
}
return nil
},
},
}
for _, tt := range tests {
rw := newTestHandlerResponseWriter()
if tt.modrw != nil {
rw = tt.modrw(rw)
}
got, gotErr := NewServerHandlerTransport(rw, tt.req)
if (gotErr != nil) != (tt.wantErr != "") || (gotErr != nil && gotErr.Error() != tt.wantErr) {
t.Errorf("%s: error = %v; want %q", tt.name, gotErr, tt.wantErr)
continue
}
if gotErr != nil {
continue
}
if tt.check != nil {
if err := tt.check(got.(*serverHandlerTransport), &tt); err != nil {
t.Errorf("%s: %v", tt.name, err)
}
}
}
}
type testHandlerResponseWriter struct {
*httptest.ResponseRecorder
closeNotify chan bool
}
func (w testHandlerResponseWriter) CloseNotify() <-chan bool { return w.closeNotify }
func (w testHandlerResponseWriter) Flush() {}
func newTestHandlerResponseWriter() http.ResponseWriter {
return testHandlerResponseWriter{
ResponseRecorder: httptest.NewRecorder(),
closeNotify: make(chan bool, 1),
}
}
type handleStreamTest struct {
t *testing.T
bodyw *io.PipeWriter
req *http.Request
rw testHandlerResponseWriter
ht *serverHandlerTransport
}
func newHandleStreamTest(t *testing.T) *handleStreamTest {
bodyr, bodyw := io.Pipe()
req := &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": {"application/grpc"},
},
URL: &url.URL{
Path: "/service/foo.bar",
},
RequestURI: "/service/foo.bar",
Body: bodyr,
}
rw := newTestHandlerResponseWriter().(testHandlerResponseWriter)
ht, err := NewServerHandlerTransport(rw, req)
if err != nil {
t.Fatal(err)
}
return &handleStreamTest{
t: t,
bodyw: bodyw,
ht: ht.(*serverHandlerTransport),
rw: rw,
}
}
func TestHandlerTransport_HandleStreams(t *testing.T) {
st := newHandleStreamTest(t)
handleStream := func(s *Stream) {
if want := "/service/foo.bar"; s.method != want {
t.Errorf("stream method = %q; want %q", s.method, want)
}
st.bodyw.Close() // no body
st.ht.WriteStatus(s, status.New(codes.OK, ""))
}
st.ht.HandleStreams(
func(s *Stream) { go handleStream(s) },
func(ctx context.Context, method string) context.Context { return ctx },
)
wantHeader := http.Header{
"Date": nil,
"Content-Type": {"application/grpc"},
"Trailer": {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
"Grpc-Status": {"0"},
}
if !reflect.DeepEqual(st.rw.HeaderMap, wantHeader) {
t.Errorf("Header+Trailer Map: %#v; want %#v", st.rw.HeaderMap, wantHeader)
}
}
// Tests that codes.Unimplemented will close the body, per comment in handler_server.go.
func TestHandlerTransport_HandleStreams_Unimplemented(t *testing.T) {
handleStreamCloseBodyTest(t, codes.Unimplemented, "thingy is unimplemented")
}
// Tests that codes.InvalidArgument will close the body, per comment in handler_server.go.
func TestHandlerTransport_HandleStreams_InvalidArgument(t *testing.T) {
handleStreamCloseBodyTest(t, codes.InvalidArgument, "bad arg")
}
func handleStreamCloseBodyTest(t *testing.T, statusCode codes.Code, msg string) {
st := newHandleStreamTest(t)
handleStream := func(s *Stream) {
st.ht.WriteStatus(s, status.New(statusCode, msg))
}
st.ht.HandleStreams(
func(s *Stream) { go handleStream(s) },
func(ctx context.Context, method string) context.Context { return ctx },
)
wantHeader := http.Header{
"Date": nil,
"Content-Type": {"application/grpc"},
"Trailer": {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
"Grpc-Status": {fmt.Sprint(uint32(statusCode))},
"Grpc-Message": {encodeGrpcMessage(msg)},
}
if !reflect.DeepEqual(st.rw.HeaderMap, wantHeader) {
t.Errorf("Header+Trailer mismatch.\n got: %#v\nwant: %#v", st.rw.HeaderMap, wantHeader)
}
}
func TestHandlerTransport_HandleStreams_Timeout(t *testing.T) {
bodyr, bodyw := io.Pipe()
req := &http.Request{
ProtoMajor: 2,
Method: "POST",
Header: http.Header{
"Content-Type": {"application/grpc"},
"Grpc-Timeout": {"200m"},
},
URL: &url.URL{
Path: "/service/foo.bar",
},
RequestURI: "/service/foo.bar",
Body: bodyr,
}
rw := newTestHandlerResponseWriter().(testHandlerResponseWriter)
ht, err := NewServerHandlerTransport(rw, req)
if err != nil {
t.Fatal(err)
}
runStream := func(s *Stream) {
defer bodyw.Close()
select {
case <-s.ctx.Done():
case <-time.After(5 * time.Second):
t.Errorf("timeout waiting for ctx.Done")
return
}
err := s.ctx.Err()
if err != context.DeadlineExceeded {
t.Errorf("ctx.Err = %v; want %v", err, context.DeadlineExceeded)
return
}
ht.WriteStatus(s, status.New(codes.DeadlineExceeded, "too slow"))
}
ht.HandleStreams(
func(s *Stream) { go runStream(s) },
func(ctx context.Context, method string) context.Context { return ctx },
)
wantHeader := http.Header{
"Date": nil,
"Content-Type": {"application/grpc"},
"Trailer": {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
"Grpc-Status": {"4"},
"Grpc-Message": {encodeGrpcMessage("too slow")},
}
if !reflect.DeepEqual(rw.HeaderMap, wantHeader) {
t.Errorf("Header+Trailer Map mismatch.\n got: %#v\nwant: %#v", rw.HeaderMap, wantHeader)
}
}
func TestHandlerTransport_HandleStreams_MultiWriteStatus(t *testing.T) {
st := newHandleStreamTest(t)
handleStream := func(s *Stream) {
if want := "/service/foo.bar"; s.method != want {
t.Errorf("stream method = %q; want %q", s.method, want)
}
st.bodyw.Close() // no body
var wg sync.WaitGroup
wg.Add(5)
for i := 0; i < 5; i++ {
go func() {
defer wg.Done()
st.ht.WriteStatus(s, status.New(codes.OK, ""))
}()
}
wg.Wait()
}
st.ht.HandleStreams(
func(s *Stream) { go handleStream(s) },
func(ctx context.Context, method string) context.Context { return ctx },
)
}
func TestHandlerTransport_HandleStreams_ErrDetails(t *testing.T) {
errDetails := []proto.Message{
&epb.RetryInfo{
RetryDelay: &dpb.Duration{Seconds: 60},
},
&epb.ResourceInfo{
ResourceType: "foo bar",
ResourceName: "service.foo.bar",
Owner: "User",
},
}
statusCode := codes.ResourceExhausted
msg := "you are being throttled"
st, err := status.New(statusCode, msg).WithDetails(errDetails...)
if err != nil {
t.Fatal(err)
}
stBytes, err := proto.Marshal(st.Proto())
if err != nil {
t.Fatal(err)
}
hst := newHandleStreamTest(t)
handleStream := func(s *Stream) {
hst.ht.WriteStatus(s, st)
}
hst.ht.HandleStreams(
func(s *Stream) { go handleStream(s) },
func(ctx context.Context, method string) context.Context { return ctx },
)
wantHeader := http.Header{
"Date": nil,
"Content-Type": {"application/grpc"},
"Trailer": {"Grpc-Status", "Grpc-Message", "Grpc-Status-Details-Bin"},
"Grpc-Status": {fmt.Sprint(uint32(statusCode))},
"Grpc-Message": {encodeGrpcMessage(msg)},
"Grpc-Status-Details-Bin": {encodeBinHeader(stBytes)},
}
if !reflect.DeepEqual(hst.rw.HeaderMap, wantHeader) {
t.Errorf("Header+Trailer mismatch.\n got: %#v\nwant: %#v", hst.rw.HeaderMap, wantHeader)
}
}

1110
vendor/google.golang.org/grpc/transport/http2_client.go generated vendored
View file

File diff suppressed because it is too large Load diff

792
vendor/google.golang.org/grpc/transport/http2_server.go generated vendored
View file

File diff suppressed because it is too large Load diff

147
vendor/google.golang.org/grpc/transport/http_util.go generated vendored
View file

@ -23,7 +23,6 @@ import (
"bytes"
"encoding/base64"
"fmt"
"io"
"net"
"net/http"
"strconv"
@ -46,6 +45,12 @@ const (
// http2IOBufSize specifies the buffer size for sending frames.
defaultWriteBufSize = 32 * 1024
defaultReadBufSize = 32 * 1024
// baseContentType is the base content-type for gRPC. This is a valid
// content-type on it's own, but can also include a content-subtype such as
// "proto" as a suffix after "+" or ";". See
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
// for more details.
baseContentType = "application/grpc"
)
var (
@ -64,7 +69,7 @@ var (
http2.ErrCodeConnect: codes.Internal,
http2.ErrCodeEnhanceYourCalm: codes.ResourceExhausted,
http2.ErrCodeInadequateSecurity: codes.PermissionDenied,
http2.ErrCodeHTTP11Required: codes.FailedPrecondition,
http2.ErrCodeHTTP11Required: codes.Internal,
}
statusCodeConvTab = map[codes.Code]http2.ErrCode{
codes.Internal: http2.ErrCodeInternal,
@ -111,9 +116,10 @@ type decodeState struct {
timeout time.Duration
method string
// key-value metadata map from the peer.
mdata map[string][]string
statsTags []byte
statsTrace []byte
mdata map[string][]string
statsTags []byte
statsTrace []byte
contentSubtype string
}
// isReservedHeader checks whether hdr belongs to HTTP2 headers
@ -125,6 +131,7 @@ func isReservedHeader(hdr string) bool {
}
switch hdr {
case "content-type",
"user-agent",
"grpc-message-type",
"grpc-encoding",
"grpc-message",
@ -138,28 +145,55 @@ func isReservedHeader(hdr string) bool {
}
}
// isWhitelistedPseudoHeader checks whether hdr belongs to HTTP2 pseudoheaders
// that should be propagated into metadata visible to users.
func isWhitelistedPseudoHeader(hdr string) bool {
// isWhitelistedHeader checks whether hdr should be propagated
// into metadata visible to users.
func isWhitelistedHeader(hdr string) bool {
switch hdr {
case ":authority":
case ":authority", "user-agent":
return true
default:
return false
}
}
func validContentType(t string) bool {
e := "application/grpc"
if !strings.HasPrefix(t, e) {
return false
// contentSubtype returns the content-subtype for the given content-type. The
// given content-type must be a valid content-type that starts with
// "application/grpc". A content-subtype will follow "application/grpc" after a
// "+" or ";". See
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
// more details.
//
// If contentType is not a valid content-type for gRPC, the boolean
// will be false, otherwise true. If content-type == "application/grpc",
// "application/grpc+", or "application/grpc;", the boolean will be true,
// but no content-subtype will be returned.
//
// contentType is assumed to be lowercase already.
func contentSubtype(contentType string) (string, bool) {
if contentType == baseContentType {
return "", true
}
// Support variations on the content-type
// (e.g. "application/grpc+blah", "application/grpc;blah").
if len(t) > len(e) && t[len(e)] != '+' && t[len(e)] != ';' {
return false
if !strings.HasPrefix(contentType, baseContentType) {
return "", false
}
return true
// guaranteed since != baseContentType and has baseContentType prefix
switch contentType[len(baseContentType)] {
case '+', ';':
// this will return true for "application/grpc+" or "application/grpc;"
// which the previous validContentType function tested to be valid, so we
// just say that no content-subtype is specified in this case
return contentType[len(baseContentType)+1:], true
default:
return "", false
}
}
// contentSubtype is assumed to be lowercase
func contentType(contentSubtype string) string {
if contentSubtype == "" {
return baseContentType
}
return baseContentType + "+" + contentSubtype
}
func (d *decodeState) status() *status.Status {
@ -228,9 +262,9 @@ func (d *decodeState) decodeResponseHeader(frame *http2.MetaHeadersFrame) error
// gRPC status doesn't exist and http status is OK.
// Set rawStatusCode to be unknown and return nil error.
// So that, if the stream has ended this Unknown status
// will be propogated to the user.
// will be propagated to the user.
// Otherwise, it will be ignored. In which case, status from
// a later trailer, that has StreamEnded flag set, is propogated.
// a later trailer, that has StreamEnded flag set, is propagated.
code := int(codes.Unknown)
d.rawStatusCode = &code
return nil
@ -247,9 +281,16 @@ func (d *decodeState) addMetadata(k, v string) {
func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
switch f.Name {
case "content-type":
if !validContentType(f.Value) {
return streamErrorf(codes.FailedPrecondition, "transport: received the unexpected content-type %q", f.Value)
contentSubtype, validContentType := contentSubtype(f.Value)
if !validContentType {
return streamErrorf(codes.Internal, "transport: received the unexpected content-type %q", f.Value)
}
d.contentSubtype = contentSubtype
// TODO: do we want to propagate the whole content-type in the metadata,
// or come up with a way to just propagate the content-subtype if it was set?
// ie {"content-type": "application/grpc+proto"} or {"content-subtype": "proto"}
// in the metadata?
d.addMetadata(f.Name, f.Value)
case "grpc-encoding":
d.encoding = f.Value
case "grpc-status":
@ -299,7 +340,7 @@ func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
d.statsTrace = v
d.addMetadata(f.Name, string(v))
default:
if isReservedHeader(f.Name) && !isWhitelistedPseudoHeader(f.Name) {
if isReservedHeader(f.Name) && !isWhitelistedHeader(f.Name) {
break
}
v, err := decodeMetadataHeader(f.Name, f.Value)
@ -307,7 +348,7 @@ func (d *decodeState) processHeaderField(f hpack.HeaderField) error {
errorf("Failed to decode metadata header (%q, %q): %v", f.Name, f.Value, err)
return nil
}
d.addMetadata(f.Name, string(v))
d.addMetadata(f.Name, v)
}
return nil
}
@ -468,19 +509,63 @@ func decodeGrpcMessageUnchecked(msg string) string {
return buf.String()
}
type bufWriter struct {
buf []byte
offset int
batchSize int
conn net.Conn
err error
onFlush func()
}
func newBufWriter(conn net.Conn, batchSize int) *bufWriter {
return &bufWriter{
buf: make([]byte, batchSize*2),
batchSize: batchSize,
conn: conn,
}
}
func (w *bufWriter) Write(b []byte) (n int, err error) {
if w.err != nil {
return 0, w.err
}
n = copy(w.buf[w.offset:], b)
w.offset += n
if w.offset >= w.batchSize {
err = w.Flush()
}
return n, err
}
func (w *bufWriter) Flush() error {
if w.err != nil {
return w.err
}
if w.offset == 0 {
return nil
}
if w.onFlush != nil {
w.onFlush()
}
_, w.err = w.conn.Write(w.buf[:w.offset])
w.offset = 0
return w.err
}
type framer struct {
numWriters int32
reader io.Reader
writer *bufio.Writer
fr *http2.Framer
writer *bufWriter
fr *http2.Framer
}
func newFramer(conn net.Conn, writeBufferSize, readBufferSize int) *framer {
r := bufio.NewReaderSize(conn, readBufferSize)
w := newBufWriter(conn, writeBufferSize)
f := &framer{
reader: bufio.NewReaderSize(conn, readBufferSize),
writer: bufio.NewWriterSize(conn, writeBufferSize),
writer: w,
fr: http2.NewFramer(w, r),
}
f.fr = http2.NewFramer(f.writer, f.reader)
// Opt-in to Frame reuse API on framer to reduce garbage.
// Frames aren't safe to read from after a subsequent call to ReadFrame.
f.fr.SetReuseFrames()

175
vendor/google.golang.org/grpc/transport/http_util_test.go generated vendored
View file

@ -1,175 +0,0 @@
/*
*
* 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 (
"fmt"
"reflect"
"testing"
"time"
)
func TestTimeoutEncode(t *testing.T) {
for _, test := range []struct {
in string
out string
}{
{"12345678ns", "12345678n"},
{"123456789ns", "123457u"},
{"12345678us", "12345678u"},
{"123456789us", "123457m"},
{"12345678ms", "12345678m"},
{"123456789ms", "123457S"},
{"12345678s", "12345678S"},
{"123456789s", "2057614M"},
{"12345678m", "12345678M"},
{"123456789m", "2057614H"},
} {
d, err := time.ParseDuration(test.in)
if err != nil {
t.Fatalf("failed to parse duration string %s: %v", test.in, err)
}
out := encodeTimeout(d)
if out != test.out {
t.Fatalf("timeoutEncode(%s) = %s, want %s", test.in, out, test.out)
}
}
}
func TestTimeoutDecode(t *testing.T) {
for _, test := range []struct {
// input
s string
// output
d time.Duration
err error
}{
{"1234S", time.Second * 1234, nil},
{"1234x", 0, fmt.Errorf("transport: timeout unit is not recognized: %q", "1234x")},
{"1", 0, fmt.Errorf("transport: timeout string is too short: %q", "1")},
{"", 0, fmt.Errorf("transport: timeout string is too short: %q", "")},
} {
d, err := decodeTimeout(test.s)
if d != test.d || fmt.Sprint(err) != fmt.Sprint(test.err) {
t.Fatalf("timeoutDecode(%q) = %d, %v, want %d, %v", test.s, int64(d), err, int64(test.d), test.err)
}
}
}
func TestValidContentType(t *testing.T) {
tests := []struct {
h string
want bool
}{
{"application/grpc", true},
{"application/grpc+", true},
{"application/grpc+blah", true},
{"application/grpc;", true},
{"application/grpc;blah", true},
{"application/grpcd", false},
{"application/grpd", false},
{"application/grp", false},
}
for _, tt := range tests {
got := validContentType(tt.h)
if got != tt.want {
t.Errorf("validContentType(%q) = %v; want %v", tt.h, got, tt.want)
}
}
}
func TestEncodeGrpcMessage(t *testing.T) {
for _, tt := range []struct {
input string
expected string
}{
{"", ""},
{"Hello", "Hello"},
{"my favorite character is \u0000", "my favorite character is %00"},
{"my favorite character is %", "my favorite character is %25"},
} {
actual := encodeGrpcMessage(tt.input)
if tt.expected != actual {
t.Errorf("encodeGrpcMessage(%v) = %v, want %v", tt.input, actual, tt.expected)
}
}
}
func TestDecodeGrpcMessage(t *testing.T) {
for _, tt := range []struct {
input string
expected string
}{
{"", ""},
{"Hello", "Hello"},
{"H%61o", "Hao"},
{"H%6", "H%6"},
{"%G0", "%G0"},
{"%E7%B3%BB%E7%BB%9F", "系统"},
} {
actual := decodeGrpcMessage(tt.input)
if tt.expected != actual {
t.Errorf("dncodeGrpcMessage(%v) = %v, want %v", tt.input, actual, tt.expected)
}
}
}
const binaryValue = string(128)
func TestEncodeMetadataHeader(t *testing.T) {
for _, test := range []struct {
// input
kin string
vin string
// output
vout string
}{
{"key", "abc", "abc"},
{"KEY", "abc", "abc"},
{"key-bin", "abc", "YWJj"},
{"key-bin", binaryValue, "woA"},
} {
v := encodeMetadataHeader(test.kin, test.vin)
if !reflect.DeepEqual(v, test.vout) {
t.Fatalf("encodeMetadataHeader(%q, %q) = %q, want %q", test.kin, test.vin, v, test.vout)
}
}
}
func TestDecodeMetadataHeader(t *testing.T) {
for _, test := range []struct {
// input
kin string
vin string
// output
vout string
err error
}{
{"a", "abc", "abc", nil},
{"key-bin", "Zm9vAGJhcg==", "foo\x00bar", nil},
{"key-bin", "Zm9vAGJhcg", "foo\x00bar", nil},
{"key-bin", "woA=", binaryValue, nil},
{"a", "abc,efg", "abc,efg", nil},
} {
v, err := decodeMetadataHeader(test.kin, test.vin)
if !reflect.DeepEqual(v, test.vout) || !reflect.DeepEqual(err, test.err) {
t.Fatalf("decodeMetadataHeader(%q, %q) = %q, %v, want %q, %v", test.kin, test.vin, v, err, test.vout, test.err)
}
}
}

406
vendor/google.golang.org/grpc/transport/transport.go generated vendored
View file

@ -17,19 +17,19 @@
*/
// Package transport defines and implements message oriented communication
// channel to complete various transactions (e.g., an RPC).
package transport // import "google.golang.org/grpc/transport"
// channel to complete various transactions (e.g., an RPC). It is meant for
// grpc-internal usage and is not intended to be imported directly by users.
package transport // externally used as import "google.golang.org/grpc/transport"
import (
stdctx "context"
"errors"
"fmt"
"io"
"net"
"sync"
"time"
"sync/atomic"
"golang.org/x/net/context"
"golang.org/x/net/http2"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/credentials"
"google.golang.org/grpc/keepalive"
@ -58,6 +58,7 @@ type recvBuffer struct {
c chan recvMsg
mu sync.Mutex
backlog []recvMsg
err error
}
func newRecvBuffer() *recvBuffer {
@ -69,6 +70,13 @@ func newRecvBuffer() *recvBuffer {
func (b *recvBuffer) put(r recvMsg) {
b.mu.Lock()
if b.err != nil {
b.mu.Unlock()
// An error had occurred earlier, don't accept more
// data or errors.
return
}
b.err = r.err
if len(b.backlog) == 0 {
select {
case b.c <- r:
@ -102,14 +110,15 @@ func (b *recvBuffer) get() <-chan recvMsg {
return b.c
}
//
// recvBufferReader implements io.Reader interface to read the data from
// recvBuffer.
type recvBufferReader struct {
ctx context.Context
goAway chan struct{}
recv *recvBuffer
last []byte // Stores the remaining data in the previous calls.
err error
ctx context.Context
ctxDone <-chan struct{} // cache of ctx.Done() (for performance).
recv *recvBuffer
last []byte // Stores the remaining data in the previous calls.
err error
}
// Read reads the next len(p) bytes from last. If last is drained, it tries to
@ -131,10 +140,8 @@ func (r *recvBufferReader) read(p []byte) (n int, err error) {
return copied, nil
}
select {
case <-r.ctx.Done():
case <-r.ctxDone:
return 0, ContextErr(r.ctx.Err())
case <-r.goAway:
return 0, ErrStreamDrain
case m := <-r.recv.get():
r.recv.load()
if m.err != nil {
@ -146,61 +153,7 @@ func (r *recvBufferReader) read(p []byte) (n int, err error) {
}
}
// All items in an out of a controlBuffer should be the same type.
type item interface {
item()
}
// controlBuffer is an unbounded channel of item.
type controlBuffer struct {
c chan item
mu sync.Mutex
backlog []item
}
func newControlBuffer() *controlBuffer {
b := &controlBuffer{
c: make(chan item, 1),
}
return b
}
func (b *controlBuffer) put(r item) {
b.mu.Lock()
if len(b.backlog) == 0 {
select {
case b.c <- r:
b.mu.Unlock()
return
default:
}
}
b.backlog = append(b.backlog, r)
b.mu.Unlock()
}
func (b *controlBuffer) load() {
b.mu.Lock()
if len(b.backlog) > 0 {
select {
case b.c <- b.backlog[0]:
b.backlog[0] = nil
b.backlog = b.backlog[1:]
default:
}
}
b.mu.Unlock()
}
// get returns the channel that receives an item in the buffer.
//
// Upon receipt of an item, the caller should call load to send another
// item onto the channel if there is any.
func (b *controlBuffer) get() <-chan item {
return b.c
}
type streamState uint8
type streamState uint32
const (
streamActive streamState = iota
@ -211,66 +164,75 @@ const (
// Stream represents an RPC in the transport layer.
type Stream struct {
id uint32
// nil for client side Stream.
st ServerTransport
// ctx is the associated context of the stream.
ctx context.Context
// cancel is always nil for client side Stream.
cancel context.CancelFunc
// done is closed when the final status arrives.
done chan struct{}
// goAway is closed when the server sent GoAways signal before this stream was initiated.
goAway chan struct{}
// method records the associated RPC method of the stream.
method string
id uint32
st ServerTransport // nil for client side Stream
ctx context.Context // the associated context of the stream
cancel context.CancelFunc // always nil for client side Stream
done chan struct{} // closed at the end of stream to unblock writers. On the client side.
ctxDone <-chan struct{} // same as done chan but for server side. Cache of ctx.Done() (for performance)
method string // the associated RPC method of the stream
recvCompress string
sendCompress string
buf *recvBuffer
trReader io.Reader
fc *inFlow
recvQuota uint32
// TODO: Remote this unused variable.
// The accumulated inbound quota pending for window update.
updateQuota uint32
wq *writeQuota
// Callback to state application's intentions to read data. This
// is used to adjust flow control, if need be.
// is used to adjust flow control, if needed.
requestRead func(int)
sendQuotaPool *quotaPool
localSendQuota *quotaPool
// Close headerChan to indicate the end of reception of header metadata.
headerChan chan struct{}
// header caches the received header metadata.
header metadata.MD
// The key-value map of trailer metadata.
trailer metadata.MD
headerChan chan struct{} // closed to indicate the end of header metadata.
headerDone uint32 // set when headerChan is closed. Used to avoid closing headerChan multiple times.
header metadata.MD // the received header metadata.
trailer metadata.MD // the key-value map of trailer metadata.
mu sync.RWMutex // guard the following
// headerOK becomes true from the first header is about to send.
headerOk bool
headerOk bool // becomes true from the first header is about to send
state streamState
// true iff headerChan is closed. Used to avoid closing headerChan
// multiple times.
headerDone bool
// the status error received from the server.
status *status.Status
// rstStream indicates whether a RST_STREAM frame needs to be sent
// to the server to signify that this stream is closing.
rstStream bool
// rstError is the error that needs to be sent along with the RST_STREAM frame.
rstError http2.ErrCode
// bytesSent and bytesReceived indicates whether any bytes have been sent or
// received on this stream.
bytesSent bool
bytesReceived bool
status *status.Status // the status error received from the server
bytesReceived uint32 // indicates whether any bytes have been received on this stream
unprocessed uint32 // set if the server sends a refused stream or GOAWAY including this stream
// contentSubtype is the content-subtype for requests.
// this must be lowercase or the behavior is undefined.
contentSubtype string
}
func (s *Stream) swapState(st streamState) streamState {
return streamState(atomic.SwapUint32((*uint32)(&s.state), uint32(st)))
}
func (s *Stream) compareAndSwapState(oldState, newState streamState) bool {
return atomic.CompareAndSwapUint32((*uint32)(&s.state), uint32(oldState), uint32(newState))
}
func (s *Stream) getState() streamState {
return streamState(atomic.LoadUint32((*uint32)(&s.state)))
}
func (s *Stream) waitOnHeader() error {
if s.headerChan == nil {
// On the server headerChan is always nil since a stream originates
// only after having received headers.
return nil
}
select {
case <-s.ctx.Done():
return ContextErr(s.ctx.Err())
case <-s.headerChan:
return nil
}
}
// RecvCompress returns the compression algorithm applied to the inbound
// message. It is empty string if there is no compression applied.
func (s *Stream) RecvCompress() string {
if err := s.waitOnHeader(); err != nil {
return ""
}
return s.recvCompress
}
@ -285,28 +247,17 @@ func (s *Stream) Done() <-chan struct{} {
return s.done
}
// GoAway returns a channel which is closed when the server sent GoAways signal
// before this stream was initiated.
func (s *Stream) GoAway() <-chan struct{} {
return s.goAway
}
// Header acquires the key-value pairs of header metadata once it
// is available. It blocks until i) the metadata is ready or ii) there is no
// header metadata or iii) the stream is canceled/expired.
func (s *Stream) Header() (metadata.MD, error) {
var err error
select {
case <-s.ctx.Done():
err = ContextErr(s.ctx.Err())
case <-s.goAway:
err = ErrStreamDrain
case <-s.headerChan:
return s.header.Copy(), nil
}
err := s.waitOnHeader()
// Even if the stream is closed, header is returned if available.
select {
case <-s.headerChan:
if s.header == nil {
return nil, nil
}
return s.header.Copy(), nil
default:
}
@ -316,10 +267,10 @@ func (s *Stream) Header() (metadata.MD, error) {
// Trailer returns the cached trailer metedata. Note that if it is not called
// after the entire stream is done, it could return an empty MD. Client
// side only.
// It can be safely read only after stream has ended that is either read
// or write have returned io.EOF.
func (s *Stream) Trailer() metadata.MD {
s.mu.RLock()
c := s.trailer.Copy()
s.mu.RUnlock()
return c
}
@ -329,6 +280,15 @@ func (s *Stream) ServerTransport() ServerTransport {
return s.st
}
// ContentSubtype returns the content-subtype for a request. For example, a
// content-subtype of "proto" will result in a content-type of
// "application/grpc+proto". This will always be lowercase. See
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests for
// more details.
func (s *Stream) ContentSubtype() string {
return s.contentSubtype
}
// Context returns the context of the stream.
func (s *Stream) Context() context.Context {
return s.ctx
@ -340,36 +300,42 @@ func (s *Stream) Method() string {
}
// Status returns the status received from the server.
// Status can be read safely only after the stream has ended,
// that is, read or write has returned io.EOF.
func (s *Stream) Status() *status.Status {
return s.status
}
// SetHeader sets the header metadata. This can be called multiple times.
// Server side only.
// This should not be called in parallel to other data writes.
func (s *Stream) SetHeader(md metadata.MD) error {
s.mu.Lock()
if s.headerOk || s.state == streamDone {
s.mu.Unlock()
return ErrIllegalHeaderWrite
}
if md.Len() == 0 {
s.mu.Unlock()
return nil
}
if s.headerOk || atomic.LoadUint32((*uint32)(&s.state)) == uint32(streamDone) {
return ErrIllegalHeaderWrite
}
s.header = metadata.Join(s.header, md)
s.mu.Unlock()
return nil
}
// SendHeader sends the given header metadata. The given metadata is
// combined with any metadata set by previous calls to SetHeader and
// then written to the transport stream.
func (s *Stream) SendHeader(md metadata.MD) error {
t := s.ServerTransport()
return t.WriteHeader(s, md)
}
// SetTrailer sets the trailer metadata which will be sent with the RPC status
// by the server. This can be called multiple times. Server side only.
// This should not be called parallel to other data writes.
func (s *Stream) SetTrailer(md metadata.MD) error {
if md.Len() == 0 {
return nil
}
s.mu.Lock()
s.trailer = metadata.Join(s.trailer, md)
s.mu.Unlock()
return nil
}
@ -409,28 +375,15 @@ func (t *transportReader) Read(p []byte) (n int, err error) {
return
}
// finish sets the stream's state and status, and closes the done channel.
// s.mu must be held by the caller. st must always be non-nil.
func (s *Stream) finish(st *status.Status) {
s.status = st
s.state = streamDone
close(s.done)
}
// BytesSent indicates whether any bytes have been sent on this stream.
func (s *Stream) BytesSent() bool {
s.mu.Lock()
bs := s.bytesSent
s.mu.Unlock()
return bs
}
// BytesReceived indicates whether any bytes have been received on this stream.
func (s *Stream) BytesReceived() bool {
s.mu.Lock()
br := s.bytesReceived
s.mu.Unlock()
return br
return atomic.LoadUint32(&s.bytesReceived) == 1
}
// Unprocessed indicates whether the server did not process this stream --
// i.e. it sent a refused stream or GOAWAY including this stream ID.
func (s *Stream) Unprocessed() bool {
return atomic.LoadUint32(&s.unprocessed) == 1
}
// GoString is implemented by Stream so context.String() won't
@ -439,21 +392,6 @@ func (s *Stream) GoString() string {
return fmt.Sprintf("<stream: %p, %v>", s, s.method)
}
// The key to save transport.Stream in the context.
type streamKey struct{}
// newContextWithStream creates a new context from ctx and attaches stream
// to it.
func newContextWithStream(ctx context.Context, stream *Stream) context.Context {
return context.WithValue(ctx, streamKey{}, stream)
}
// StreamFromContext returns the stream saved in ctx.
func StreamFromContext(ctx context.Context) (s *Stream, ok bool) {
s, ok = ctx.Value(streamKey{}).(*Stream)
return
}
// state of transport
type transportState int
@ -475,6 +413,7 @@ type ServerConfig struct {
InitialConnWindowSize int32
WriteBufferSize int
ReadBufferSize int
ChannelzParentID int64
}
// NewServerTransport creates a ServerTransport with conn or non-nil error
@ -510,18 +449,21 @@ type ConnectOptions struct {
WriteBufferSize int
// ReadBufferSize sets the size of read buffer, which in turn determines how much data can be read at most for one read syscall.
ReadBufferSize int
// ChannelzParentID sets the addrConn id which initiate the creation of this client transport.
ChannelzParentID int64
}
// TargetInfo contains the information of the target such as network address and metadata.
type TargetInfo struct {
Addr string
Metadata interface{}
Addr string
Metadata interface{}
Authority string
}
// NewClientTransport establishes the transport with the required ConnectOptions
// and returns it to the caller.
func NewClientTransport(ctx context.Context, target TargetInfo, opts ConnectOptions, timeout time.Duration) (ClientTransport, error) {
return newHTTP2Client(ctx, target, opts, timeout)
func NewClientTransport(connectCtx, ctx context.Context, target TargetInfo, opts ConnectOptions, onSuccess func()) (ClientTransport, error) {
return newHTTP2Client(connectCtx, ctx, target, opts, onSuccess)
}
// Options provides additional hints and information for message
@ -545,10 +487,6 @@ type CallHdr struct {
// Method specifies the operation to perform.
Method string
// RecvCompress specifies the compression algorithm applied on
// inbound messages.
RecvCompress string
// SendCompress specifies the compression algorithm applied on
// outbound message.
SendCompress string
@ -563,6 +501,14 @@ type CallHdr struct {
// for performance purposes.
// If it's false, new stream will never be flushed.
Flush bool
// ContentSubtype specifies the content-subtype for a request. For example, a
// content-subtype of "proto" will result in a content-type of
// "application/grpc+proto". The value of ContentSubtype must be all
// lowercase, otherwise the behavior is undefined. See
// https://github.com/grpc/grpc/blob/master/doc/PROTOCOL-HTTP2.md#requests
// for more details.
ContentSubtype string
}
// ClientTransport is the common interface for all gRPC client-side transport
@ -604,6 +550,12 @@ type ClientTransport interface {
// GetGoAwayReason returns the reason why GoAway frame was received.
GetGoAwayReason() GoAwayReason
// IncrMsgSent increments the number of message sent through this transport.
IncrMsgSent()
// IncrMsgRecv increments the number of message received through this transport.
IncrMsgRecv()
}
// ServerTransport is the common interface for all gRPC server-side transport
@ -637,6 +589,12 @@ type ServerTransport interface {
// Drain notifies the client this ServerTransport stops accepting new RPCs.
Drain()
// IncrMsgSent increments the number of message sent through this transport.
IncrMsgSent()
// IncrMsgRecv increments the number of message received through this transport.
IncrMsgRecv()
}
// streamErrorf creates an StreamError with the specified error code and description.
@ -686,9 +644,16 @@ func (e ConnectionError) Origin() error {
var (
// ErrConnClosing indicates that the transport is closing.
ErrConnClosing = connectionErrorf(true, nil, "transport is closing")
// ErrStreamDrain indicates that the stream is rejected by the server because
// the server stops accepting new RPCs.
ErrStreamDrain = streamErrorf(codes.Unavailable, "the server stops accepting new RPCs")
// errStreamDrain indicates that the stream is rejected because the
// connection is draining. This could be caused by goaway or balancer
// removing the address.
errStreamDrain = streamErrorf(codes.Unavailable, "the connection is draining")
// errStreamDone is returned from write at the client side to indiacte application
// layer of an error.
errStreamDone = errors.New("the stream is done")
// StatusGoAway indicates that the server sent a GOAWAY that included this
// stream's ID in unprocessed RPCs.
statusGoAway = status.New(codes.Unavailable, "the stream is rejected because server is draining the connection")
)
// TODO: See if we can replace StreamError with status package errors.
@ -703,75 +668,16 @@ func (e StreamError) Error() string {
return fmt.Sprintf("stream error: code = %s desc = %q", e.Code, e.Desc)
}
// wait blocks until it can receive from one of the provided contexts or channels
func wait(ctx, tctx context.Context, done, goAway <-chan struct{}, proceed <-chan int) (int, error) {
select {
case <-ctx.Done():
return 0, ContextErr(ctx.Err())
case <-done:
return 0, io.EOF
case <-goAway:
return 0, ErrStreamDrain
case <-tctx.Done():
return 0, ErrConnClosing
case i := <-proceed:
return i, nil
}
}
// ContextErr converts the error from context package into a StreamError.
func ContextErr(err error) StreamError {
switch err {
case context.DeadlineExceeded, stdctx.DeadlineExceeded:
return streamErrorf(codes.DeadlineExceeded, "%v", err)
case context.Canceled, stdctx.Canceled:
return streamErrorf(codes.Canceled, "%v", err)
}
return streamErrorf(codes.Internal, "Unexpected error from context packet: %v", err)
}
// GoAwayReason contains the reason for the GoAway frame received.
type GoAwayReason uint8
const (
// Invalid indicates that no GoAway frame is received.
Invalid GoAwayReason = 0
// NoReason is the default value when GoAway frame is received.
NoReason GoAwayReason = 1
// TooManyPings indicates that a GoAway frame with ErrCodeEnhanceYourCalm
// was received and that the debug data said "too_many_pings".
TooManyPings GoAwayReason = 2
// GoAwayInvalid indicates that no GoAway frame is received.
GoAwayInvalid GoAwayReason = 0
// GoAwayNoReason is the default value when GoAway frame is received.
GoAwayNoReason GoAwayReason = 1
// GoAwayTooManyPings indicates that a GoAway frame with
// ErrCodeEnhanceYourCalm was received and that the debug data said
// "too_many_pings".
GoAwayTooManyPings GoAwayReason = 2
)
// loopyWriter is run in a separate go routine. It is the single code path that will
// write data on wire.
func loopyWriter(ctx context.Context, cbuf *controlBuffer, handler func(item) error) {
for {
select {
case i := <-cbuf.get():
cbuf.load()
if err := handler(i); err != nil {
return
}
case <-ctx.Done():
return
}
hasData:
for {
select {
case i := <-cbuf.get():
cbuf.load()
if err := handler(i); err != nil {
return
}
case <-ctx.Done():
return
default:
if err := handler(&flushIO{}); err != nil {
return
}
break hasData
}
}
}
}

2158
vendor/google.golang.org/grpc/transport/transport_test.go generated vendored
View file

File diff suppressed because it is too large Load diff