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

Merge pull request #1076 from apolcyn/account_for_padding

Browse Source 
account for padding in flow control, when received frame uses it
This commit is contained in:
apolcyn
2017-03-06 13:37:14 -08:00
committed by GitHub
parent 0e5ab0098f 3de1621ff6
commit 7b399ed358
2 changed files with 32 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

20
transport/http2_client.go
View File

@ -740,7 +740,7 @@ func (t *http2Client) updateWindow(s *Stream, n uint32) {
} }
func (t *http2Client) handleData(f *http2.DataFrame) { func (t *http2Client) handleData(f *http2.DataFrame) {
size := len(f.Data()) size := f.Header().Length
if err := t.fc.onData(uint32(size)); err != nil { if err := t.fc.onData(uint32(size)); err != nil {
t.notifyError(connectionErrorf(true, err, "%v", err)) t.notifyError(connectionErrorf(true, err, "%v", err))
return return
@ -754,6 +754,11 @@ func (t *http2Client) handleData(f *http2.DataFrame) {
return return
} }
if size > 0 { if size > 0 {
if f.Header().Flags.Has(http2.FlagDataPadded) {
if w := t.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
t.controlBuf.put(&windowUpdate{0, w})
}
}
s.mu.Lock() s.mu.Lock()
if s.state == streamDone { if s.state == streamDone {
s.mu.Unlock() s.mu.Unlock()
@ -773,13 +778,20 @@ func (t *http2Client) handleData(f *http2.DataFrame) {
s.write(recvMsg{err: io.EOF}) s.write(recvMsg{err: io.EOF})
return return
} }
if f.Header().Flags.Has(http2.FlagDataPadded) {
if w := s.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
t.controlBuf.put(&windowUpdate{s.id, w})
}
}
s.mu.Unlock() s.mu.Unlock()
// TODO(bradfitz, zhaoq): A copy is required here because there is no // TODO(bradfitz, zhaoq): A copy is required here because there is no
// guarantee f.Data() is consumed before the arrival of next frame. // guarantee f.Data() is consumed before the arrival of next frame.
// Can this copy be eliminated? // Can this copy be eliminated?
data := make([]byte, size) if len(f.Data()) > 0 {
copy(data, f.Data()) data := make([]byte, len(f.Data()))
s.write(recvMsg{data: data}) copy(data, f.Data())
s.write(recvMsg{data: data})
}
} }
// The server has closed the stream without sending trailers. Record that // The server has closed the stream without sending trailers. Record that
// the read direction is closed, and set the status appropriately. // the read direction is closed, and set the status appropriately.

20
transport/http2_server.go
View File

@ -381,7 +381,7 @@ func (t *http2Server) updateWindow(s *Stream, n uint32) {
} }
func (t *http2Server) handleData(f *http2.DataFrame) { func (t *http2Server) handleData(f *http2.DataFrame) {
size := len(f.Data()) size := f.Header().Length
if err := t.fc.onData(uint32(size)); err != nil { if err := t.fc.onData(uint32(size)); err != nil {
grpclog.Printf("transport: http2Server %v", err) grpclog.Printf("transport: http2Server %v", err)
t.Close() t.Close()
@ -396,6 +396,11 @@ func (t *http2Server) handleData(f *http2.DataFrame) {
return return
} }
if size > 0 { if size > 0 {
if f.Header().Flags.Has(http2.FlagDataPadded) {
if w := t.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
t.controlBuf.put(&windowUpdate{0, w})
}
}
s.mu.Lock() s.mu.Lock()
if s.state == streamDone { if s.state == streamDone {
s.mu.Unlock() s.mu.Unlock()
@ -411,13 +416,20 @@ func (t *http2Server) handleData(f *http2.DataFrame) {
t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl}) t.controlBuf.put(&resetStream{s.id, http2.ErrCodeFlowControl})
return return
} }
if f.Header().Flags.Has(http2.FlagDataPadded) {
if w := s.fc.onRead(uint32(size) - uint32(len(f.Data()))); w > 0 {
t.controlBuf.put(&windowUpdate{s.id, w})
}
}
s.mu.Unlock() s.mu.Unlock()
// TODO(bradfitz, zhaoq): A copy is required here because there is no // TODO(bradfitz, zhaoq): A copy is required here because there is no
// guarantee f.Data() is consumed before the arrival of next frame. // guarantee f.Data() is consumed before the arrival of next frame.
// Can this copy be eliminated? // Can this copy be eliminated?
data := make([]byte, size) if len(f.Data()) > 0 {
copy(data, f.Data()) data := make([]byte, len(f.Data()))
s.write(recvMsg{data: data}) copy(data, f.Data())
s.write(recvMsg{data: data})
}
} }
if f.Header().Flags.Has(http2.FlagDataEndStream) { if f.Header().Flags.Has(http2.FlagDataEndStream) {
// Received the end of stream from the client. // Received the end of stream from the client.