opensource/grpc-go
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

do right counting for max-streams flow control

Browse Source
This commit is contained in:
iamqizhao
2015-07-28 16:41:46 -07:00
parent 31fa21984e
commit 4da1327a66
3 changed files with 19 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
transport/control.go
View File

@ -105,7 +105,9 @@ type quotaPool struct {
// newQuotaPool creates a quotaPool which has quota q available to consume. // newQuotaPool creates a quotaPool which has quota q available to consume.
func newQuotaPool(q int) *quotaPool { func newQuotaPool(q int) *quotaPool {
qb := &quotaPool{c: make(chan int, 1)} qb := &quotaPool{c: make(chan int, 1)}
qb.c <- q if q > 0 {
qb.c <- q
}
return qb return qb
} }

19
transport/http2_client.go
View File

@ -196,17 +196,16 @@ func newHTTP2Client(addr string, opts *ConnectOptions) (_ ClientTransport, err e
return t, nil return t, nil
} }
func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr, sq bool) *Stream { func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream {
fc := &inFlow{ fc := &inFlow{
limit: initialWindowSize, limit: initialWindowSize,
conn: t.fc, conn: t.fc,
} }
// TODO(zhaoq): Handle uint32 overflow of Stream.id. // TODO(zhaoq): Handle uint32 overflow of Stream.id.
s := &Stream{ s := &Stream{
id: t.nextID, id: t.nextID,
method: callHdr.Method, method: callHdr.Method,
buf: newRecvBuffer(), buf: newRecvBuffer(),
updateStreams: sq,
fc: fc, fc: fc,
sendQuotaPool: newQuotaPool(int(t.streamSendQuota)), sendQuotaPool: newQuotaPool(int(t.streamSendQuota)),
headerChan: make(chan struct{}), headerChan: make(chan struct{}),
@ -267,7 +266,7 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
return nil, err return nil, err
} }
t.mu.Lock() t.mu.Lock()
s := t.newStream(ctx, callHdr, checkStreamsQuota) s := t.newStream(ctx, callHdr)
t.activeStreams[s.id] = s t.activeStreams[s.id] = s
t.mu.Unlock() t.mu.Unlock()
// HPACK encodes various headers. Note that once WriteField(...) is // HPACK encodes various headers. Note that once WriteField(...) is
@ -336,10 +335,14 @@ func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Strea
// CloseStream clears the footprint of a stream when the stream is not needed any more. // CloseStream clears the footprint of a stream when the stream is not needed any more.
// This must not be executed in reader's goroutine. // This must not be executed in reader's goroutine.
func (t *http2Client) CloseStream(s *Stream, err error) { func (t *http2Client) CloseStream(s *Stream, err error) {
var updateStreams bool
t.mu.Lock() t.mu.Lock()
if t.streamsQuota != nil {
updateStreams = true
}
delete(t.activeStreams, s.id) delete(t.activeStreams, s.id)
t.mu.Unlock() t.mu.Unlock()
if s.updateStreams { if updateStreams {
t.streamsQuota.add(1) t.streamsQuota.add(1)
} }
s.mu.Lock() s.mu.Lock()
@ -737,7 +740,7 @@ func (t *http2Client) applySettings(ss []http2.Setting) {
t.mu.Lock() t.mu.Lock()
reset := t.streamsQuota != nil reset := t.streamsQuota != nil
if !reset { if !reset {
t.streamsQuota = newQuotaPool(int(s.Val)) t.streamsQuota = newQuotaPool(int(s.Val) - len(t.activeStreams))
} }
ms := t.maxStreams ms := t.maxStreams
t.maxStreams = int(s.Val) t.maxStreams = int(s.Val)

16
transport/transport.go
View File

@ -169,17 +169,11 @@ type Stream struct {
ctx context.Context ctx context.Context
cancel context.CancelFunc cancel context.CancelFunc
// method records the associated RPC method of the stream. // method records the associated RPC method of the stream.
method string method string
buf *recvBuffer buf *recvBuffer
dec io.Reader dec io.Reader
fc *inFlow
// updateStreams indicates whether the transport's streamsQuota needed recvQuota uint32
// to be updated when this stream is closed. It is false when the transport
// sticks to the initial infinite value of the number of concurrent streams.
// Ture otherwise.
updateStreams bool
fc *inFlow
recvQuota uint32
// The accumulated inbound quota pending for window update. // The accumulated inbound quota pending for window update.
updateQuota uint32 updateQuota uint32
// The handler to control the window update procedure for both this // The handler to control the window update procedure for both this