http2: improved Request.Body.Close not to hold lock on connection

The existing implementation holds a lock on a connection which causes
issues on a slow Request.Body.Close call.

Unlock before Request.Body.Close call. The abortStream closes the request
body after unlocking a mutex. The abortStreamLocked returns reqBody as
io.ReadCloser if the caller needs to close the body after unlocking the
mutex.

Fixes golang/go#52853

Change-Id: I0b74ba5263f65393c0e69e1c645d10c4db048903
Reviewed-on: https://go-review.googlesource.com/c/net/+/424755
Reviewed-by: Damien Neil <dneil@google.com>
Run-TryBot: Damien Neil <dneil@google.com>
Reviewed-by: Dmitri Shuralyov <dmitshur@google.com>
TryBot-Result: Gopher Robot <gobot@golang.org>

Author: nikola-jokic

http2/transport.go

@@ -376,25 +376,33 @@ func (cs *clientStream) get1xxTraceFunc() func(int, textproto.MIMEHeader) error
 }
 
 func (cs *clientStream) abortStream(err error) {
+	var reqBody io.ReadCloser
+	defer func() {
+		if reqBody != nil {
+			reqBody.Close()
+		}
+	}()
 	cs.cc.mu.Lock()
 	defer cs.cc.mu.Unlock()
-	cs.abortStreamLocked(err)
+	reqBody = cs.abortStreamLocked(err)
 }
 
-func (cs *clientStream) abortStreamLocked(err error) {
+func (cs *clientStream) abortStreamLocked(err error) io.ReadCloser {
 	cs.abortOnce.Do(func() {
 		cs.abortErr = err
 		close(cs.abort)
 	})
+	var reqBody io.ReadCloser
 	if cs.reqBody != nil && !cs.reqBodyClosed {
-		cs.reqBody.Close()
 		cs.reqBodyClosed = true
+		reqBody = cs.reqBody
 	}
 	// TODO(dneil): Clean up tests where cs.cc.cond is nil.
 	if cs.cc.cond != nil {
 		// Wake up writeRequestBody if it is waiting on flow control.
 		cs.cc.cond.Broadcast()
 	}
+	return reqBody
 }
 
 func (cs *clientStream) abortRequestBodyWrite() {
@@ -763,6 +771,12 @@ func (cc *ClientConn) SetDoNotReuse() {
 }
 
 func (cc *ClientConn) setGoAway(f *GoAwayFrame) {
+	var reqBodiesToClose []io.ReadCloser
+	defer func() {
+		for _, reqBody := range reqBodiesToClose {
+			reqBody.Close()
+		}
+	}()
 	cc.mu.Lock()
 	defer cc.mu.Unlock()
 
@@ -779,7 +793,10 @@ func (cc *ClientConn) setGoAway(f *GoAwayFrame) {
 	last := f.LastStreamID
 	for streamID, cs := range cc.streams {
 		if streamID > last {
-			cs.abortStreamLocked(errClientConnGotGoAway)
+			reqBody := cs.abortStreamLocked(errClientConnGotGoAway)
+			if reqBody != nil {
+				reqBodiesToClose = append(reqBodiesToClose, reqBody)
+			}
 		}
 	}
 }
@@ -1032,11 +1049,19 @@ func (cc *ClientConn) sendGoAway() error {
 func (cc *ClientConn) closeForError(err error) {
 	cc.mu.Lock()
 	cc.closed = true
+
+	var reqBodiesToClose []io.ReadCloser
 	for _, cs := range cc.streams {
-		cs.abortStreamLocked(err)
+		reqBody := cs.abortStreamLocked(err)
+		if reqBody != nil {
+			reqBodiesToClose = append(reqBodiesToClose, reqBody)
+		}
 	}
 	cc.cond.Broadcast()
 	cc.mu.Unlock()
+	for _, reqBody := range reqBodiesToClose {
+		reqBody.Close()
+	}
 	cc.closeConn()
 }
 
@@ -2084,17 +2109,25 @@ func (rl *clientConnReadLoop) cleanup() {
 		err = io.ErrUnexpectedEOF
 	}
 	cc.closed = true
+
+	var reqBodiesToClose []io.ReadCloser
 	for _, cs := range cc.streams {
 		select {
 		case <-cs.peerClosed:
 			// The server closed the stream before closing the conn,
 			// so no need to interrupt it.
 		default:
-			cs.abortStreamLocked(err)
+			reqBody := cs.abortStreamLocked(err)
+			if reqBody != nil {
+				reqBodiesToClose = append(reqBodiesToClose, reqBody)
+			}
 		}
 	}
 	cc.cond.Broadcast()
 	cc.mu.Unlock()
+	for _, reqBody := range reqBodiesToClose {
+		reqBody.Close()
+	}
 }
 
 // countReadFrameError calls Transport.CountError with a string

http2/transport_test.go

@@ -5999,3 +5999,52 @@ func testTransportClosesConnAfterGoAway(t *testing.T, lastStream uint32) {
 
 	ct.run()
 }
+
+type slowCloser struct {
+	closing chan struct{}
+	closed  chan struct{}
+}
+
+func (r *slowCloser) Read([]byte) (int, error) {
+	return 0, io.EOF
+}
+
+func (r *slowCloser) Close() error {
+	close(r.closing)
+	<-r.closed
+	return nil
+}
+
+func TestTransportSlowClose(t *testing.T) {
+	st := newServerTester(t, func(w http.ResponseWriter, r *http.Request) {
+	}, optOnlyServer)
+	defer st.Close()
+
+	client := st.ts.Client()
+	body := &slowCloser{
+		closing: make(chan struct{}),
+		closed:  make(chan struct{}),
+	}
+
+	reqc := make(chan struct{})
+	go func() {
+		defer close(reqc)
+		res, err := client.Post(st.ts.URL, "text/plain", body)
+		if err != nil {
+			t.Error(err)
+		}
+		res.Body.Close()
+	}()
+	defer func() {
+		close(body.closed)
+		<-reqc // wait for POST request to finish
+	}()
+
+	<-body.closing // wait for POST request to call body.Close
+	// This GET request should not be blocked by the in-progress POST.
+	res, err := client.Get(st.ts.URL)
+	if err != nil {
+		t.Fatal(err)
+	}
+	res.Body.Close()
+}