http2: make Transport's Response.Body.Close not wait for buffered data

The Transport's Response.Body.Close call was closing the
Response.Body, but the Reader implementation was yielding its buffered
data before returning the error. Add a new method to force an
immediate close.  An audit of the other CloseWithError callers found
that the after-buffered-data behavior was correct for them.

New tests in the main go repo in net/http/clientserver_test.go:
TestResponseBodyReadAfterClose_h1 and TestResponseBodyReadAfterClose_h2

Updates golang/go#13648

Change-Id: If3a13a20c106b5a7bbe668ccb4e3c704a0e0682b
Reviewed-on: https://go-review.googlesource.com/17937
Reviewed-by: Russ Cox <rsc@golang.org>

http2/pipe.go

@@ -14,12 +14,13 @@ import (
 // io.Pipe except there are no PipeReader/PipeWriter halves, and the
 // underlying buffer is an interface. (io.Pipe is always unbuffered)
 type pipe struct {
-	mu     sync.Mutex
-	c      sync.Cond // c.L must point to
-	b      pipeBuffer
-	err    error         // read error once empty. non-nil means closed.
-	donec  chan struct{} // closed on error
-	readFn func()        // optional code to run in Read before error
+	mu       sync.Mutex
+	c        sync.Cond // c.L lazily initialized to &p.mu
+	b        pipeBuffer
+	err      error         // read error once empty. non-nil means closed.
+	breakErr error         // immediate read error (caller doesn't see rest of b)
+	donec    chan struct{} // closed on error
+	readFn   func()        // optional code to run in Read before error
 }
 
 type pipeBuffer interface {
@@ -37,6 +38,9 @@ func (p *pipe) Read(d []byte) (n int, err error) {
 		p.c.L = &p.mu
 	}
 	for {
+		if p.breakErr != nil {
+			return 0, p.breakErr
+		}
 		if p.b.Len() > 0 {
 			return p.b.Read(d)
 		}
@@ -73,13 +77,20 @@ func (p *pipe) Write(d []byte) (n int, err error) {
 // read.
 //
 // The error must be non-nil.
-func (p *pipe) CloseWithError(err error) { p.closeWithErrorAndCode(err, nil) }
+func (p *pipe) CloseWithError(err error) { p.closeWithError(&p.err, err, nil) }
+
+// BreakWithError causes the next Read (waking up a current blocked
+// Read if needed) to return the provided err immediately, without
+// waiting for unread data.
+func (p *pipe) BreakWithError(err error) { p.closeWithError(&p.breakErr, err, nil) }
 
 // closeWithErrorAndCode is like CloseWithError but also sets some code to run
 // in the caller's goroutine before returning the error.
-func (p *pipe) closeWithErrorAndCode(err error, fn func()) {
+func (p *pipe) closeWithErrorAndCode(err error, fn func()) { p.closeWithError(&p.err, err, fn) }
+
+func (p *pipe) closeWithError(dst *error, err error, fn func()) {
 	if err == nil {
-		panic("CloseWithError err must be non-nil")
+		panic("err must be non-nil")
 	}
 	p.mu.Lock()
 	defer p.mu.Unlock()
@@ -87,22 +98,36 @@ func (p *pipe) closeWithErrorAndCode(err error, fn func()) {
 		p.c.L = &p.mu
 	}
 	defer p.c.Signal()
-	if p.err != nil {
+	if *dst != nil {
 		// Already been done.
 		return
 	}
 	p.readFn = fn
-	p.err = err
-	if p.donec != nil {
+	*dst = err
+	p.closeDoneLocked()
+}
+
+// requires p.mu be held.
+func (p *pipe) closeDoneLocked() {
+	if p.donec == nil {
+		return
+	}
+	// Close if unclosed. This isn't racy since we always
+	// hold p.mu while closing.
+	select {
+	case <-p.donec:
+	default:
 		close(p.donec)
 	}
 }
 
-// Err returns the error (if any) first set with CloseWithError.
-// This is the error which will be returned after the reader is exhausted.
+// Err returns the error (if any) first set by BreakWithError or CloseWithError.
 func (p *pipe) Err() error {
 	p.mu.Lock()
 	defer p.mu.Unlock()
+	if p.breakErr != nil {
+		return p.breakErr
+	}
 	return p.err
 }
 
@@ -113,9 +138,9 @@ func (p *pipe) Done() <-chan struct{} {
 	defer p.mu.Unlock()
 	if p.donec == nil {
 		p.donec = make(chan struct{})
-		if p.err != nil {
+		if p.err != nil || p.breakErr != nil {
 			// Already hit an error.
-			close(p.donec)
+			p.closeDoneLocked()
 		}
 	}
 	return p.donec

http2/pipe_test.go

@@ -8,6 +8,7 @@ import (
 	"bytes"
 	"errors"
 	"io"
+	"io/ioutil"
 	"testing"
 )
 
@@ -50,3 +51,59 @@ func TestPipeDoneChan_ErrFirst(t *testing.T) {
 		t.Fatal("should be done")
 	}
 }
+
+func TestPipeDoneChan_Break(t *testing.T) {
+	var p pipe
+	done := p.Done()
+	select {
+	case <-done:
+		t.Fatal("done too soon")
+	default:
+	}
+	p.BreakWithError(io.EOF)
+	select {
+	case <-done:
+	default:
+		t.Fatal("should be done")
+	}
+}
+
+func TestPipeDoneChan_Break_ErrFirst(t *testing.T) {
+	var p pipe
+	p.BreakWithError(io.EOF)
+	done := p.Done()
+	select {
+	case <-done:
+	default:
+		t.Fatal("should be done")
+	}
+}
+
+func TestPipeCloseWithError(t *testing.T) {
+	p := &pipe{b: new(bytes.Buffer)}
+	const body = "foo"
+	io.WriteString(p, body)
+	a := errors.New("test error")
+	p.CloseWithError(a)
+	all, err := ioutil.ReadAll(p)
+	if string(all) != body {
+		t.Errorf("read bytes = %q; want %q", all, body)
+	}
+	if err != a {
+		t.Logf("read error = %v, %v", err, a)
+	}
+}
+
+func TestPipeBreakWithError(t *testing.T) {
+	p := &pipe{b: new(bytes.Buffer)}
+	io.WriteString(p, "foo")
+	a := errors.New("test err")
+	p.BreakWithError(a)
+	all, err := ioutil.ReadAll(p)
+	if string(all) != "" {
+		t.Errorf("read bytes = %q; want empty string", all)
+	}
+	if err != a {
+		t.Logf("read error = %v, %v", err, a)
+	}
+}

http2/transport.go

@@ -1144,11 +1144,15 @@ func (b transportResponseBody) Read(p []byte) (n int, err error) {
 	return
 }
 
+var errClosedResponseBody = errors.New("http2: response body closed")
+
 func (b transportResponseBody) Close() error {
-	if b.cs.bufPipe.Err() != io.EOF {
+	cs := b.cs
+	if cs.bufPipe.Err() != io.EOF {
 		// TODO: write test for this
-		b.cs.cc.writeStreamReset(b.cs.ID, ErrCodeCancel, nil)
+		cs.cc.writeStreamReset(cs.ID, ErrCodeCancel, nil)
 	}
+	cs.bufPipe.BreakWithError(errClosedResponseBody)
 	return nil
 }