src/net/http/internal/http2/server.go | 41 +++++++++++++++++++++++++---------------- src/net/http/internal/http2/server_test.go | 174 ++++++++++++++++++++++++++++++++++++++++++++++++++++- diff --git a/src/net/http/internal/http2/server.go b/src/net/http/internal/http2/server.go index c1fb306977dfe8d154dfcea97fefc8738d989aec..c51bc590897f43f9a08e16f9cb29b2c083e69326 100644 --- a/src/net/http/internal/http2/server.go +++ b/src/net/http/internal/http2/server.go @@ -510,7 +510,8 @@ type stream struct { // immutable: sc *serverConn id uint32 - body *pipe // non-nil if expecting DATA frames + body *pipe // non-nil if expecting DATA frames + reqBody *requestBody cw closeWaiter // closed wait stream transitions to closed state ctx context.Context cancelCtx func() @@ -1584,10 +1585,6 @@ sc.startGracefulShutdownInternal() } } if p := st.body; p != nil { - // Return any buffered unread bytes worth of conn-level flow control. - // See golang.org/issue/16481 - sc.sendWindowUpdate(nil, p.Len()) - p.CloseWithError(err) } if e, ok := err.(StreamError); ok { @@ -1952,7 +1949,7 @@ st.reqTrailer = req.Trailer if st.reqTrailer != nil { st.trailer = make(Header) } - st.body = req.Body.(*requestBody).pipe // may be nil + st.body = st.reqBody.pipe // may be nil st.declBodyBytes = req.ContentLength handler := sc.handler.ServeHTTP @@ -1967,7 +1964,7 @@ if sc.hs.ReadTimeout() > 0 { st.readDeadline = time.AfterFunc(sc.hs.ReadTimeout(), st.onReadTimeout) } - return sc.scheduleHandler(id, rw, req, handler) + return sc.scheduleHandler(st, rw, req, handler) } func (sc *serverConn) upgradeRequest(req *ServerRequest) { @@ -2172,7 +2169,7 @@ if bodyOpen { if _, ok := rp.Header["Content-Length"]; !ok { req.ContentLength = -1 } - req.Body.(*requestBody).pipe = &pipe{ + st.reqBody.pipe = &pipe{ b: &dataBuffer{expected: req.ContentLength}, } } @@ -2187,7 +2184,7 @@ if res.InvalidReason != "" { return nil, nil, sc.countError(res.InvalidReason, streamError(st.id, ErrCodeProtocol)) } - body := &requestBody{ + st.reqBody = &requestBody{ conn: sc, stream: st, needsContinue: res.NeedsContinue, @@ -2205,7 +2202,7 @@ ProtoMajor: 2, ProtoMinor: 0, TLS: sc.tlsState, Host: rp.Authority, - Body: body, + Body: st.reqBody, Trailer: res.Trailer, } return rw, &rw.rws.req, nil @@ -2227,11 +2224,12 @@ streamID uint32 rw *responseWriter req *ServerRequest handler func(*ResponseWriter, *ServerRequest) + body *pipe } // scheduleHandler starts a handler goroutine, // or schedules one to start as soon as an existing handler finishes. -func (sc *serverConn) scheduleHandler(streamID uint32, rw *responseWriter, req *ServerRequest, handler func(*ResponseWriter, *ServerRequest)) error { +func (sc *serverConn) scheduleHandler(st *stream, rw *responseWriter, req *ServerRequest, handler func(*ResponseWriter, *ServerRequest)) error { sc.serveG.check() maxHandlers := sc.advMaxStreams if sc.curHandlers < maxHandlers { @@ -2243,10 +2241,11 @@ if len(sc.unstartedHandlers) > int(4*sc.advMaxStreams) { return sc.countError("too_many_early_resets", ConnectionError(ErrCodeEnhanceYourCalm)) } sc.unstartedHandlers = append(sc.unstartedHandlers, unstartedHandler{ - streamID: streamID, + streamID: st.id, rw: rw, req: req, handler: handler, + body: st.body, }) return nil } @@ -2260,6 +2259,10 @@ for ; i < len(sc.unstartedHandlers); i++ { u := sc.unstartedHandlers[i] if sc.streams[u.streamID] == nil { // This stream was reset before its goroutine had a chance to start. + if u.body != nil { + u.body.BreakWithError(errClosedBody) + sc.sendWindowUpdate(nil, u.body.Len()) + } continue } if sc.curHandlers >= maxHandlers { @@ -2284,6 +2287,12 @@ rw.rws.stream.cancelCtx() if req.MultipartForm != nil { req.MultipartForm.RemoveAll() } + if b := rw.rws.stream.reqBody; b != nil { + // Closing the body refunds flow control credit for any unconsumed data. + // (reqBody is nil for Upgrade: h2c requests, but those do not use flow + // control for the request body.) + b.Close() + } if didPanic { e := recover() sc.writeFrameFromHandler(FrameWriteRequest{ @@ -2379,7 +2388,7 @@ func (sc *serverConn) noteBodyRead(st *stream, n int) { sc.serveG.check() sc.sendWindowUpdate(nil, n) // conn-level - if st.state != stateHalfClosedRemote && st.state != stateClosed { + if st != nil && st.state != stateHalfClosedRemote && st.state != stateClosed { // Don't send this WINDOW_UPDATE if the stream is closed // remotely. sc.sendWindowUpdate(st, n) @@ -2427,6 +2436,9 @@ func (b *requestBody) Close() error { b.closeOnce.Do(func() { if b.pipe != nil { b.pipe.BreakWithError(errClosedBody) + if unread := b.pipe.Len(); unread > 0 { + b.conn.noteBodyReadFromHandler(nil, unread, errClosedBody) + } } }) return nil @@ -2443,9 +2455,6 @@ } n, err = b.pipe.Read(p) if err == io.EOF { b.sawEOF = true - } - if b.conn == nil { - return } b.conn.noteBodyReadFromHandler(b.stream, n, err) return diff --git a/src/net/http/internal/http2/server_test.go b/src/net/http/internal/http2/server_test.go index 38ea7afac3ff70870beef3cdb219e28586d5229d..fb2b84e04663bd0422006d1e5dbe425f7907df04 100644 --- a/src/net/http/internal/http2/server_test.go +++ b/src/net/http/internal/http2/server_test.go @@ -3559,6 +3559,166 @@ endStream: true, }) } +func TestServerResetStreamUnreadBody(t *testing.T) { + synctestSubtest(t, "read_after_reset", func(t *testing.T) { + st := newServerTester(t, nil) + defer st.Close() + + st.greet() + st.writeHeaders(HeadersFrameParam{ + StreamID: 1, + BlockFragment: st.encodeHeader(":method", "POST"), + EndHeaders: true, + }) + call := st.nextHandlerCall() + + const size = InflowMinRefresh + st.writeData(1, false, make([]byte, size)) + st.writeRSTStream(1, ErrCodeCancel) + st.sync() + + // The handler reads the body after the stream has been reset. + call.do(func(w http.ResponseWriter, req *http.Request) { + io.ReadAll(req.Body) + }) + st.wantWindowUpdate(0, size) + + // Handler exits; no second refund. + call.exit() + st.wantIdle() + }) + + synctestSubtest(t, "exit_without_reading", func(t *testing.T) { + st := newServerTester(t, nil) + defer st.Close() + + st.greet() + st.writeHeaders(HeadersFrameParam{ + StreamID: 1, + BlockFragment: st.encodeHeader(":method", "POST"), + EndHeaders: true, + }) + call := st.nextHandlerCall() + + const size = InflowMinRefresh + st.writeData(1, false, make([]byte, size)) + st.writeRSTStream(1, ErrCodeCancel) + st.sync() + + // The handler exits without reading the body. Flow control is returned on exit. + call.exit() + st.wantWindowUpdate(0, size) + st.wantIdle() + }) + + synctestSubtest(t, "read_after_handler_exit", func(t *testing.T) { + st := newServerTester(t, nil) + defer st.Close() + + st.greet() + st.writeHeaders(HeadersFrameParam{ + StreamID: 1, + BlockFragment: st.encodeHeader(":method", "POST"), + EndHeaders: true, + }) + call := st.nextHandlerCall() + + const size = InflowMinRefresh + st.writeData(1, false, make([]byte, size)) + synctest.Wait() + + // The handler exits without reading the body. + call.exit() + + // Flow control is returned when the handler exits. + st.wantUnorderedFrames( + func(f *WindowUpdateFrame) bool { + return f.StreamID == 0 && f.Increment == size + }, + func(f *HeadersFrame) bool { + return f.StreamID == 1 && f.StreamEnded() + }, + func(f *RSTStreamFrame) bool { + return f.StreamID == 1 && f.ErrCode == ErrCodeNo + }, + ) + + // Reading the body after the handler exits must not cause a double refund. + io.ReadAll(call.req.Body) + st.wantIdle() + }) + + synctestSubtest(t, "unstarted_handler", func(t *testing.T) { + st := newServerTester(t, nil, func(h2 *http.HTTP2Config) { + h2.MaxConcurrentStreams = 1 + }) + defer st.Close() + + st.greet() + + // Stream 1 uses the single handler slot. + st.writeHeaders(HeadersFrameParam{ + StreamID: 1, + BlockFragment: st.encodeHeader(), + EndStream: true, + EndHeaders: true, + }) + call := st.nextHandlerCall() + + // Reset stream 1 so the client can open another stream without + // exceeding the concurrent stream limit. + st.writeRSTStream(1, ErrCodeCancel) + + // Stream 3 is queued in unstartedHandlers because the handler for + // stream 1 is still executing. + st.writeHeaders(HeadersFrameParam{ + StreamID: 3, + BlockFragment: st.encodeHeader(":method", "POST"), + EndHeaders: true, + }) + + const size = InflowMinRefresh + st.writeData(3, false, make([]byte, size)) + st.writeRSTStream(3, ErrCodeCancel) + st.wantIdle() + + // Stream 1 handler exits. Stream 3 is removed from unstartedHandlers + // and its flow control is returned. + call.exit() + st.wantWindowUpdate(0, size) + st.wantIdle() + }) + + // Same thing as exit_without_reading, but some imaginary middleware + // replaced Request.Body first. + synctestSubtest(t, "middleware_replaces_body", func(t *testing.T) { + st := newServerTester(t, nil) + defer st.Close() + + st.greet() + st.writeHeaders(HeadersFrameParam{ + StreamID: 1, + BlockFragment: st.encodeHeader(":method", "POST"), + EndHeaders: true, + }) + call := st.nextHandlerCall() + + call.do(func(w http.ResponseWriter, r *http.Request) { + // Wrap r.Body like some middleware might. + r.Body = http.MaxBytesReader(w, r.Body, 1000) + }) + + const size = InflowMinRefresh + st.writeData(1, false, make([]byte, size)) + st.writeRSTStream(1, ErrCodeCancel) + st.sync() + + call.exit() + st.wantWindowUpdate(0, size) + st.wantIdle() + }) +} + func TestServerIdleTimeout(t *testing.T) { synctest.Test(t, testServerIdleTimeout) } func testServerIdleTimeout(t *testing.T) { if testing.Short() { @@ -4064,9 +4224,21 @@ } } } + st.wantHeaders(wantHeader{ + streamID: 1, + endStream: true, + }) + // Writing data after the stream is reset immediately returns flow control credit. st.writeData(1, false, content[windowSize/2:]) - st.wantWindowUpdate(0, windowSize/2) + st.wantUnorderedFrames( + func(f *WindowUpdateFrame) bool { + return f.StreamID == 0 && f.Increment == windowSize/2 + }, + func(f *RSTStreamFrame) bool { + return f.StreamID == 1 && f.ErrCode == ErrCodeStreamClosed + }, + ) } func TestNoErrorLoggedOnPostAfterGOAWAY(t *testing.T) {