rename inner types again

Author: apolcyn

transport/control.go

@@ -175,18 +175,19 @@ type inFlow struct {
 	// window update for them. Used to reduce window update frequency.
 	pendingUpdate uint32
 
-	// Pre-consumed window space. This is temporary extra space in the incoming flow control
-	// window that is ok to for incoming data to fill up.
-	preConsumedWindowSpace uint32
+	// This is temporary space in the incoming flow control that can be granted at convenient times
+	// to prevent the sender from stalling for lack flow control space.
+	// If present, it is paid back when data is consumed from the window.
+	loanedWindowSpace uint32
 }
 
 // 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.preConsumedWindowSpace {
-		return fmt.Errorf("received %d-bytes data exceeding the limit %d bytes", f.pendingData+f.pendingUpdate, f.limit+f.preConsumedWindowSpace)
+	if f.pendingData+f.pendingUpdate > f.limit+f.loanedWindowSpace {
+		return fmt.Errorf("received %d-bytes data exceeding the limit %d bytes", f.pendingData+f.pendingUpdate, f.limit+f.loanedWindowSpace)
 	}
 	return nil
 }
@@ -207,12 +208,11 @@ func (f *inFlow) onRead(n uint32) uint32 {
 		grpclog.Fatalf("potential window update too large. onRead(n) where n is %v; max n is %v", f.pendingUpdate, http2MaxWindowUpdate)
 	}
 	f.pendingData -= n
-	// add to "pendingUpdate" only if all data in "preConsumedWindowSpace" has been "consumed"
-	if f.preConsumedWindowSpace > 0 {
-		p := min(n, f.preConsumedWindowSpace)
-		f.preConsumedWindowSpace -= p
-		n -= p
-	}
+	// first use up remaining "loanedWindowSpace", add remaining Read to "pendingUpdate"
+	windowSpaceDebtPayment := min(n, f.loanedWindowSpace)
+	f.loanedWindowSpace -= windowSpaceDebtPayment
+	n -= windowSpaceDebtPayment
+
 	f.pendingUpdate += n
 	if f.pendingUpdate >= f.limit/4 {
 		wu := f.pendingUpdate
@@ -222,18 +222,16 @@ func (f *inFlow) onRead(n uint32) uint32 {
 	return 0
 }
 
-func (f *inFlow) preConsumeWindowSpace(n uint32) uint32 {
+func (f *inFlow) loanWindowSpace(n uint32) uint32 {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	if f.preConsumedWindowSpace > 0 {
-		grpclog.Fatalf("pre-consuming window space while there is pre-consumed window space stil  outstanding")
+	if f.loanedWindowSpace > 0 {
+		grpclog.Fatalf("pre-consuming window space while there is pre-consumed window space still  outstanding")
 	}
-	f.preConsumedWindowSpace = n
-	if f.preConsumedWindowSpace+f.pendingData > http2MaxWindowUpdate {
-		grpclog.Fatalf("potential window of %v update too large. http2 max window update is %v", f.preConsumedWindowSpace+f.pendingData, http2MaxWindowUpdate)
-	}
-	if f.preConsumedWindowSpace+f.pendingUpdate >= f.limit/4 {
-		wu := f.pendingUpdate + f.preConsumedWindowSpace
+	f.loanedWindowSpace = n
+
+	if f.loanedWindowSpace+f.pendingUpdate >= f.limit/4 {
+		wu := f.pendingUpdate + f.loanedWindowSpace
 		f.pendingUpdate = 0
 		return wu
 	}

transport/http2_client.go

@@ -300,13 +300,13 @@ func (t *http2Client) newStream(ctx context.Context, callHdr *CallHdr) *Stream {
 	// That means, s.ctx should be read-only. And s.ctx is done iff ctx is done.
 	// So we use the original context here instead of creating a copy.
 	s.ctx = ctx
-	consumeConnectionWindow := func(n uint32) {
+	consumeConnAndStreamWindows := func(n uint32) {
 		t.updateWindow(s, n)
 	}
-	preConsumeStreamWindow := func(n uint32) {
-		t.preConsumeWindowSpace(s, n)
+	loanSpaceInStreamWindow := func(n uint32) {
+		t.loanSpaceInStreamWindow(s, n)
 	}
-	s.setClientStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
+	s.setClientStreamReader(consumeConnAndStreamWindows, loanSpaceInStreamWindow)
 	return s
 }
 
@@ -771,13 +771,13 @@ func (t *http2Client) updateWindow(s *Stream, n uint32) {
 	}
 }
 
-func (t *http2Client) preConsumeWindowSpace(s *Stream, n uint32) {
+func (t *http2Client) loanSpaceInStreamWindow(s *Stream, n uint32) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.state == streamDone {
 		return
 	}
-	if w := s.fc.preConsumeWindowSpace(n); w > 0 {
+	if w := s.fc.loanWindowSpace(n); w > 0 {
 		t.controlBuf.put(&windowUpdate{s.id, w})
 	}
 }

transport/http2_server.go

@@ -296,13 +296,13 @@ func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(
 		}
 		t.stats.HandleRPC(s.ctx, inHeader)
 	}
-	consumeConnectionWindow := func(n uint32) {
+	consumeConnAndStreamWindows := func(n uint32) {
 		t.updateWindow(s, n)
 	}
-	preConsumeStreamWindow := func(n uint32) {
-		t.preConsumeWindowSpace(s, n)
+	loanSpaceInStreamWindow := func(n uint32) {
+		t.loanSpaceInStreamWindow(s, n)
 	}
-	s.setServerStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
+	s.setServerStreamReader(consumeConnAndStreamWindows, loanSpaceInStreamWindow)
 	handle(s)
 	return
 }
@@ -421,13 +421,13 @@ func (t *http2Server) updateWindow(s *Stream, n uint32) {
 	}
 }
 
-func (t *http2Server) preConsumeWindowSpace(s *Stream, n uint32) {
+func (t *http2Server) loanSpaceInStreamWindow(s *Stream, n uint32) {
 	s.mu.Lock()
 	defer s.mu.Unlock()
 	if s.state == streamDone {
 		return
 	}
-	if w := s.fc.preConsumeWindowSpace(n); w > 0 {
+	if w := s.fc.loanWindowSpace(n); w > 0 {
 		t.controlBuf.put(&windowUpdate{s.id, w})
 	}
 }

transport/transport.go

@@ -221,33 +221,33 @@ type Stream struct {
 	rstError http2.ErrCode
 }
 
-func (s *Stream) setClientStreamReader(consumeConnectionWindow func(uint32), preConsumeStreamWindow func(uint32)) {
+func (s *Stream) setClientStreamReader(consumeConnAndStreamWindows func(uint32), loanSpaceInStreamWindow func(uint32)) {
 	if s.ctx == nil || s.goAway == nil || s.buf == nil {
 		grpclog.Fatalf("Uninitialized stream. s.ctx == nil: %v; s.goAway == nil: %v; s.buf == nil: %v", s.ctx == nil, s.goAway == nil, s.buf == nil)
 	}
-	s.setStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
+	s.setStreamReader(consumeConnAndStreamWindows, loanSpaceInStreamWindow)
 }
 
-func (s *Stream) setServerStreamReader(consumeConnectionWindow func(uint32), preConsumeStreamWindow func(uint32)) {
+func (s *Stream) setServerStreamReader(consumeConnAndStreamWindows func(uint32), loanSpaceInStreamWindow func(uint32)) {
 	if s.ctx == nil || s.buf == nil {
 		grpclog.Fatalf("Uninitialized stream. s.ctx == nil: %v; s.buf == nil: %v", s.ctx == nil, s.buf == nil)
 	}
 	if s.goAway != nil {
 		grpclog.Fatalf("Unexpected initialization of s.goAway; got %v; want nil", s.goAway)
 	}
-	s.setStreamReader(consumeConnectionWindow, preConsumeStreamWindow)
+	s.setStreamReader(consumeConnAndStreamWindows, loanSpaceInStreamWindow)
 }
 
-func (s *Stream) setStreamReader(consumeConnectionWindow func(uint32), preConsumeStreamWindow func(uint32)) {
-	s.streamReader = &streamWindowPreconsumingReader{
-		preConsumeStreamWindow: preConsumeStreamWindow,
-		cwr: &connectionWindowConsumingReader{
+func (s *Stream) setStreamReader(consumeConnAndStreamWindows func(uint32), loanSpaceInStreamWindow func(uint32)) {
+	s.streamReader = &streamWindowSpaceLoaningReader{
+		loanSpaceInStreamWindow: loanSpaceInStreamWindow,
+		wr: &connAndStreamWindowConsumingReader{
 			dec: &recvBufferReader{
 				ctx:    s.ctx,
 				goAway: s.goAway,
 				recv:   s.buf,
 			},
-			consumeConnectionWindow: consumeConnectionWindow,
+			consumeConnAndStreamWindows: consumeConnAndStreamWindows,
 		},
 	}
 }
@@ -273,39 +273,39 @@ func (s *Stream) ReadFull(p []byte) (n int, err error) {
 	return io.ReadFull(s.streamReader, p)
 }
 
-type streamWindowPreconsumingReader struct {
+type streamWindowSpaceLoaningReader struct {
 	// The handler to control the window update procedure for both this
 	// particular stream and the associated transport.
-	preConsumeStreamWindow func(uint32)
-	cwr                    *connectionWindowConsumingReader
-	readFullErr            error
+	loanSpaceInStreamWindow func(uint32)
+	wr                      *connAndStreamWindowConsumingReader
+	readFullErr             error
 }
 
-func (r *streamWindowPreconsumingReader) Read(p []byte) (n int, err error) {
+func (r *streamWindowSpaceLoaningReader) Read(p []byte) (n int, err error) {
 	if r.readFullErr != nil {
 		return 0, r.readFullErr
 	}
 	defer func() { r.readFullErr = err }()
-	readAmount := min(maxSingleStreamWindowUpdate, uint32(len(p)))
-	r.preConsumeStreamWindow(readAmount)
-	return io.ReadFull(r.cwr, p[:readAmount])
+	committedReadAmount := min(maxSingleStreamWindowUpdate, uint32(len(p)))
+	r.loanSpaceInStreamWindow(committedReadAmount)
+	return io.ReadFull(r.wr, p[:committedReadAmount])
 }
 
-type connectionWindowConsumingReader struct {
-	dec                     io.Reader
-	consumeConnectionWindow func(uint32)
+type connAndStreamWindowConsumingReader struct {
+	dec                         io.Reader
+	consumeConnAndStreamWindows func(uint32)
 }
 
 // Read reads all the data available for this Stream from the transport and
 // passes them into the decoder, which converts them into a gRPC message stream.
 // The error is io.EOF when the stream is done or another non-nil error if
 // the stream broke.
-func (s *connectionWindowConsumingReader) Read(p []byte) (n int, err error) {
+func (s *connAndStreamWindowConsumingReader) Read(p []byte) (n int, err error) {
 	n, err = s.dec.Read(p)
 	if err != nil {
 		return
 	}
-	s.consumeConnectionWindow(uint32(n))
+	s.consumeConnAndStreamWindows(uint32(n))
 	return
 }
 

transport/transport_test.go

@@ -1112,7 +1112,7 @@ func TestClientWithMisbehavedServer(t *testing.T) {
 		t.Fatalf("Failed to write: %v", err)
 	}
 	// reflect to get the inner recvBufferReader, which reads without doing window updates
-	recvBufferReader := s.streamReader.(*streamWindowPreconsumingReader).cwr.dec
+	recvBufferReader := s.streamReader.(*streamWindowSpaceLoaningReader).wr.dec
 	// Read without window update.
 	for {
 		p := make([]byte, http2MaxFrameLen)
@@ -1174,7 +1174,7 @@ func TestEncodingRequiredStatus(t *testing.T) {
 	}
 	p := make([]byte, http2MaxFrameLen)
 	// reflect to get the plain recvBufferReader from the stream's stream reader, which doesn't do window updates
-	recvBufferReader := s.streamReader.(*streamWindowPreconsumingReader).cwr.dec
+	recvBufferReader := s.streamReader.(*streamWindowSpaceLoaningReader).wr.dec
 	if _, err := recvBufferReader.Read(p); err != io.EOF {
 		t.Fatalf("Read got error %v, want %v", err, io.EOF)
 	}
@@ -1204,7 +1204,7 @@ func TestInvalidHeaderField(t *testing.T) {
 	}
 	p := make([]byte, http2MaxFrameLen)
 	// reflect to get the inner recvBufferReader, which reads without doing window updates
-	_, err = s.streamReader.(*streamWindowPreconsumingReader).cwr.dec.Read(p)
+	_, err = s.streamReader.(*streamWindowSpaceLoaningReader).wr.dec.Read(p)
 	if se, ok := err.(StreamError); !ok || se.Code != codes.FailedPrecondition || !strings.Contains(err.Error(), expectedInvalidHeaderField) {
 		t.Fatalf("Read got error %v, want error with code %s and contains %q", err, codes.FailedPrecondition, expectedInvalidHeaderField)
 	}