// closing a net.Conn that is now owned by the caller.
var errCallerOwnsConn = errors.New("read loop ending; caller owns writable underlying conn")
+// maxPostCloseReadBytes is the max number of bytes that a client is willing to
+// read when draining the response body of any unread bytes after it has been
+// closed. This number is chosen for consistency with maxPostHandlerReadBytes.
+const maxPostCloseReadBytes = 256 << 10
+
+// maxPostCloseReadTime defines the maximum amount of time that a client is
+// willing to spend on draining a response body of any unread bytes after it
+// has been closed.
+const maxPostCloseReadTime = 50 * time.Millisecond
+
+func maybeDrainBody(body io.Reader) bool {
+ drainedCh := make(chan bool, 1)
+ go func() {
+ if _, err := io.CopyN(io.Discard, body, maxPostCloseReadBytes+1); err == io.EOF {
+ drainedCh <- true
+ } else {
+ drainedCh <- false
+ }
+ }()
+ select {
+ case drained := <-drainedCh:
+ return drained
+ case <-time.After(maxPostCloseReadTime):
+ return false
+ }
+}
+
func (pc *persistConn) readLoop() {
closeErr := errReadLoopExiting // default value, if not changed below
defer func() {
// reading the response body. (or for cancellation or death)
select {
case bodyEOF := <-waitForBodyRead:
+ if !bodyEOF && resp.ContentLength <= maxPostCloseReadBytes {
+ bodyEOF = maybeDrainBody(body.body)
+ }
alive = alive &&
bodyEOF &&
!pc.sawEOF &&
"fmt"
"go/token"
"internal/nettrace"
- "internal/synctest"
"io"
"log"
mrand "math/rand"
"net"
+ "net/http"
. "net/http"
"net/http/httptest"
"net/http/httptrace"
"sync/atomic"
"testing"
"testing/iotest"
+ "testing/synctest"
"time"
"golang.org/x/net/http/httpguts"
}
}
+// In HTTP/1, if a response body has not been fully read by the time it is
+// closed, we try to drain it, up to a maximum byte and time limit. If we
+// manage to drain it before the next request, the connection is re-used;
+// otherwise, a new connection is made.
+func TestTransportNotReadToEndConnectionReuse(t *testing.T) {
+ run(t, testTransportNotReadToEndConnectionReuse, []testMode{http1Mode, https1Mode})
+}
+func testTransportNotReadToEndConnectionReuse(t *testing.T, mode testMode) {
+ tests := []struct {
+ name string
+ bodyLen int
+ contentLenKnown bool
+ headRequest bool
+ timeBetweenReqs time.Duration
+ responseTime time.Duration
+ wantReuse bool
+ }{
+ {
+ name: "unconsumed body within drain limit",
+ bodyLen: 200 * 1024,
+ timeBetweenReqs: http.MaxPostCloseReadTime,
+ wantReuse: true,
+ },
+ {
+ name: "unconsumed body within drain limit with known length",
+ bodyLen: 200 * 1024,
+ contentLenKnown: true,
+ timeBetweenReqs: http.MaxPostCloseReadTime,
+ wantReuse: true,
+ },
+ {
+ name: "unconsumed body larger than drain limit",
+ bodyLen: 500 * 1024,
+ timeBetweenReqs: http.MaxPostCloseReadTime,
+ wantReuse: false,
+ },
+ {
+ name: "unconsumed body larger than drain limit with known length",
+ bodyLen: 500 * 1024,
+ contentLenKnown: true,
+ timeBetweenReqs: http.MaxPostCloseReadTime,
+ wantReuse: false,
+ },
+ {
+ name: "new requests start before drain for old requests are finished",
+ bodyLen: 200 * 1024,
+ timeBetweenReqs: 0,
+ responseTime: time.Minute,
+ wantReuse: false,
+ },
+ {
+ // Server handler will always return no body when handling a HEAD
+ // request, which should always allow connection re-use.
+ name: "unconsumed body larger than drain limit for HEAD request",
+ bodyLen: 500 * 1024,
+ headRequest: true,
+ wantReuse: true,
+ },
+ }
+
+ for _, tc := range tests {
+ subtest := func(t *testing.T) {
+ addrSeen := make(map[string]int)
+ ts := newClientServerTest(t, mode, HandlerFunc(func(w ResponseWriter, r *Request) {
+ addrSeen[r.RemoteAddr]++
+ time.Sleep(tc.responseTime)
+ if tc.contentLenKnown {
+ w.Header().Add("Content-Length", strconv.Itoa(tc.bodyLen))
+ }
+ w.Write(slices.Repeat([]byte("a"), tc.bodyLen))
+ }), optFakeNet).ts
+
+ var wg sync.WaitGroup
+ for range 10 {
+ wg.Go(func() {
+ method := http.MethodGet
+ if tc.headRequest {
+ method = http.MethodHead
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ req, err := http.NewRequestWithContext(ctx, method, ts.URL, nil)
+ if err != nil {
+ log.Fatal(err)
+ }
+ resp, err := ts.Client().Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if resp.StatusCode != http.StatusOK {
+ t.Errorf("expected HTTP 200, got: %v", resp.StatusCode)
+ }
+ resp.Body.Close()
+ // Context cancellation and body read after the body has been
+ // closed should not affect connection re-use.
+ cancel()
+ if n, err := resp.Body.Read([]byte{}); n != 0 || err == nil {
+ t.Errorf("read after body has been closed should not succeed, but read %v byte with %v error", n, err)
+ }
+ })
+ time.Sleep(tc.timeBetweenReqs)
+ synctest.Wait()
+ }
+ wg.Wait()
+ if (len(addrSeen) == 1) != tc.wantReuse {
+ t.Errorf("want connection reuse to be %v, but %v connections were created", tc.wantReuse, len(addrSeen))
+ }
+ }
+ t.Run(tc.name, func(t *testing.T) {
+ synctest.Test(t, subtest)
+ })
+ }
+}
+
func TestTransportMaxPerHostIdleConns(t *testing.T) {
run(t, testTransportMaxPerHostIdleConns, []testMode{http1Mode})
}