idleLRU connLRU
reqMu sync.Mutex
- reqCanceler map[cancelKey]func(error)
+ reqCanceler map[*Request]context.CancelCauseFunc
altMu sync.Mutex // guards changing altProto only
altProto atomic.Value // of nil or map[string]RoundTripper, key is URI scheme
ForceAttemptHTTP2 bool
}
-// A cancelKey is the key of the reqCanceler map.
-// We wrap the *Request in this type since we want to use the original request,
-// not any transient one created by roundTrip.
-type cancelKey struct {
- req *Request
-}
-
func (t *Transport) writeBufferSize() int {
if t.WriteBufferSize > 0 {
return t.WriteBufferSize
// optional extra headers to write and stores any error to return
// from roundTrip.
type transportRequest struct {
- *Request // original request, not to be mutated
- extra Header // extra headers to write, or nil
- trace *httptrace.ClientTrace // optional
- cancelKey cancelKey
+ *Request // original request, not to be mutated
+ extra Header // extra headers to write, or nil
+ trace *httptrace.ClientTrace // optional
+
+ ctx context.Context // canceled when we are done with the request
+ cancel context.CancelCauseFunc
mu sync.Mutex // guards err
err error // first setError value for mapRoundTripError to consider
}
// roundTrip implements a RoundTripper over HTTP.
-func (t *Transport) roundTrip(req *Request) (*Response, error) {
+func (t *Transport) roundTrip(req *Request) (_ *Response, err error) {
t.nextProtoOnce.Do(t.onceSetNextProtoDefaults)
ctx := req.Context()
trace := httptrace.ContextClientTrace(ctx)
}
origReq := req
- cancelKey := cancelKey{origReq}
req = setupRewindBody(req)
if altRT := t.alternateRoundTripper(req); altRT != nil {
return nil, errors.New("http: no Host in request URL")
}
+ // Transport request context.
+ //
+ // If RoundTrip returns an error, it cancels this context before returning.
+ //
+ // If RoundTrip returns no error:
+ // - For an HTTP/1 request, persistConn.readLoop cancels this context
+ // after reading the request body.
+ // - For an HTTP/2 request, RoundTrip cancels this context after the HTTP/2
+ // RoundTripper returns.
+ ctx, cancel := context.WithCancelCause(req.Context())
+
+ // Convert Request.Cancel into context cancelation.
+ if origReq.Cancel != nil {
+ go awaitLegacyCancel(ctx, cancel, origReq)
+ }
+
+ // Convert Transport.CancelRequest into context cancelation.
+ //
+ // This is lamentably expensive. CancelRequest has been deprecated for a long time
+ // and doesn't work on HTTP/2 requests. Perhaps we should drop support for it entirely.
+ cancel = t.prepareTransportCancel(origReq, cancel)
+
+ defer func() {
+ if err != nil {
+ cancel(err)
+ }
+ }()
+
for {
select {
case <-ctx.Done():
req.closeBody()
- return nil, ctx.Err()
+ return nil, context.Cause(ctx)
default:
}
// treq gets modified by roundTrip, so we need to recreate for each retry.
- treq := &transportRequest{Request: req, trace: trace, cancelKey: cancelKey}
+ treq := &transportRequest{Request: req, trace: trace, ctx: ctx, cancel: cancel}
cm, err := t.connectMethodForRequest(treq)
if err != nil {
req.closeBody()
// to send it requests.
pconn, err := t.getConn(treq, cm)
if err != nil {
- t.setReqCanceler(cancelKey, nil)
req.closeBody()
return nil, err
}
var resp *Response
if pconn.alt != nil {
// HTTP/2 path.
- t.setReqCanceler(cancelKey, nil) // not cancelable with CancelRequest
resp, err = pconn.alt.RoundTrip(req)
} else {
resp, err = pconn.roundTrip(treq)
}
if err == nil {
+ if pconn.alt != nil {
+ // HTTP/2 requests are not cancelable with CancelRequest,
+ // so we have no further need for the request context.
+ //
+ // On the HTTP/1 path, roundTrip takes responsibility for
+ // canceling the context after the response body is read.
+ cancel(errRequestDone)
+ }
resp.Request = origReq
return resp, nil
}
}
}
+func awaitLegacyCancel(ctx context.Context, cancel context.CancelCauseFunc, req *Request) {
+ select {
+ case <-req.Cancel:
+ cancel(errRequestCanceled)
+ case <-ctx.Done():
+ }
+}
+
var errCannotRewind = errors.New("net/http: cannot rewind body after connection loss")
type readTrackingBody struct {
}
}
+// prepareTransportCancel sets up state to convert Transport.CancelRequest into context cancelation.
+func (t *Transport) prepareTransportCancel(req *Request, origCancel context.CancelCauseFunc) context.CancelCauseFunc {
+ // Historically, RoundTrip has not modified the Request in any way.
+ // We could avoid the need to keep a map of all in-flight requests by adding
+ // a field to the Request containing its cancel func, and setting that field
+ // while the request is in-flight. Callers aren't supposed to reuse a Request
+ // until after the response body is closed, so this wouldn't violate any
+ // concurrency guarantees.
+ cancel := func(err error) {
+ origCancel(err)
+ t.reqMu.Lock()
+ delete(t.reqCanceler, req)
+ t.reqMu.Unlock()
+ }
+ t.reqMu.Lock()
+ if t.reqCanceler == nil {
+ t.reqCanceler = make(map[*Request]context.CancelCauseFunc)
+ }
+ t.reqCanceler[req] = cancel
+ t.reqMu.Unlock()
+ return cancel
+}
+
// CancelRequest cancels an in-flight request by closing its connection.
// CancelRequest should only be called after [Transport.RoundTrip] has returned.
//
// Deprecated: Use [Request.WithContext] to create a request with a
// cancelable context instead. CancelRequest cannot cancel HTTP/2
-// requests.
+// requests. This may become a no-op in a future release of Go.
func (t *Transport) CancelRequest(req *Request) {
- t.cancelRequest(cancelKey{req}, errRequestCanceled)
-}
-
-// Cancel an in-flight request, recording the error value.
-// Returns whether the request was canceled.
-func (t *Transport) cancelRequest(key cancelKey, err error) bool {
- // This function must not return until the cancel func has completed.
- // See: https://golang.org/issue/34658
t.reqMu.Lock()
- defer t.reqMu.Unlock()
- cancel := t.reqCanceler[key]
- delete(t.reqCanceler, key)
+ cancel := t.reqCanceler[req]
+ t.reqMu.Unlock()
if cancel != nil {
- cancel(err)
+ cancel(errRequestCanceled)
}
-
- return cancel != nil
}
//
return removed
}
-func (t *Transport) setReqCanceler(key cancelKey, fn func(error)) {
- t.reqMu.Lock()
- defer t.reqMu.Unlock()
- if t.reqCanceler == nil {
- t.reqCanceler = make(map[cancelKey]func(error))
- }
- if fn != nil {
- t.reqCanceler[key] = fn
- } else {
- delete(t.reqCanceler, key)
- }
-}
-
-// replaceReqCanceler replaces an existing cancel function. If there is no cancel function
-// for the request, we don't set the function and return false.
-// Since CancelRequest will clear the canceler, we can use the return value to detect if
-// the request was canceled since the last setReqCancel call.
-func (t *Transport) replaceReqCanceler(key cancelKey, fn func(error)) bool {
- t.reqMu.Lock()
- defer t.reqMu.Unlock()
- _, ok := t.reqCanceler[key]
- if !ok {
- return false
- }
- if fn != nil {
- t.reqCanceler[key] = fn
- } else {
- delete(t.reqCanceler, key)
- }
- return true
-}
-
var zeroDialer net.Dialer
func (t *Transport) dial(ctx context.Context, network, addr string) (net.Conn, error) {
}
}()
- var cancelc chan error
-
// Queue for idle connection.
- if delivered := t.queueForIdleConn(w); delivered {
- // set request canceler to some non-nil function so we
- // can detect whether it was cleared between now and when
- // we enter roundTrip
- t.setReqCanceler(treq.cancelKey, func(error) {})
- } else {
- cancelc = make(chan error, 1)
- t.setReqCanceler(treq.cancelKey, func(err error) { cancelc <- err })
-
- // Queue for permission to dial.
+ if delivered := t.queueForIdleConn(w); !delivered {
t.queueForDial(w)
}
// what caused r.err; if so, prefer to return the
// cancellation error (see golang.org/issue/16049).
select {
- case <-req.Cancel:
- return nil, errRequestCanceledConn
- case <-req.Context().Done():
- return nil, req.Context().Err()
- case err := <-cancelc:
+ case <-treq.ctx.Done():
+ err := context.Cause(treq.ctx)
if err == errRequestCanceled {
err = errRequestCanceledConn
}
}
}
return r.pc, r.err
- case <-req.Cancel:
- return nil, errRequestCanceledConn
- case <-req.Context().Done():
- return nil, req.Context().Err()
- case err := <-cancelc:
+ case <-treq.ctx.Done():
+ err := context.Cause(treq.ctx)
if err == errRequestCanceled {
err = errRequestCanceledConn
}
pc.t.removeIdleConn(pc)
}()
- tryPutIdleConn := func(trace *httptrace.ClientTrace) bool {
+ tryPutIdleConn := func(treq *transportRequest) bool {
+ trace := treq.trace
if err := pc.t.tryPutIdleConn(pc); err != nil {
closeErr = err
if trace != nil && trace.PutIdleConn != nil && err != errKeepAlivesDisabled {
pc.mu.Unlock()
rc := <-pc.reqch
- trace := httptrace.ContextClientTrace(rc.req.Context())
+ trace := rc.treq.trace
var resp *Response
if err == nil {
pc.mu.Unlock()
bodyWritable := resp.bodyIsWritable()
- hasBody := rc.req.Method != "HEAD" && resp.ContentLength != 0
+ hasBody := rc.treq.Request.Method != "HEAD" && resp.ContentLength != 0
- if resp.Close || rc.req.Close || resp.StatusCode <= 199 || bodyWritable {
+ if resp.Close || rc.treq.Request.Close || resp.StatusCode <= 199 || bodyWritable {
// Don't do keep-alive on error if either party requested a close
// or we get an unexpected informational (1xx) response.
// StatusCode 100 is already handled above.
}
if !hasBody || bodyWritable {
- replaced := pc.t.replaceReqCanceler(rc.cancelKey, nil)
-
// Put the idle conn back into the pool before we send the response
// so if they process it quickly and make another request, they'll
// get this same conn. But we use the unbuffered channel 'rc'
alive = alive &&
!pc.sawEOF &&
pc.wroteRequest() &&
- replaced && tryPutIdleConn(trace)
+ tryPutIdleConn(rc.treq)
if bodyWritable {
closeErr = errCallerOwnsConn
return
}
+ rc.treq.cancel(errRequestDone)
+
// Now that they've read from the unbuffered channel, they're safely
// out of the select that also waits on this goroutine to die, so
// we're allowed to exit now if needed (if alive is false)
// reading the response body. (or for cancellation or death)
select {
case bodyEOF := <-waitForBodyRead:
- replaced := pc.t.replaceReqCanceler(rc.cancelKey, nil) // before pc might return to idle pool
alive = alive &&
bodyEOF &&
!pc.sawEOF &&
pc.wroteRequest() &&
- replaced && tryPutIdleConn(trace)
+ tryPutIdleConn(rc.treq)
if bodyEOF {
eofc <- struct{}{}
}
- case <-rc.req.Cancel:
- alive = false
- pc.t.cancelRequest(rc.cancelKey, errRequestCanceled)
- case <-rc.req.Context().Done():
+ case <-rc.treq.ctx.Done():
alive = false
- pc.t.cancelRequest(rc.cancelKey, rc.req.Context().Err())
+ pc.cancelRequest(errRequestCanceled)
case <-pc.closech:
alive = false
- pc.t.setReqCanceler(rc.cancelKey, nil)
}
+ rc.treq.cancel(errRequestDone)
testHookReadLoopBeforeNextRead()
}
}
continueCh := rc.continueCh
for {
- resp, err = ReadResponse(pc.br, rc.req)
+ resp, err = ReadResponse(pc.br, rc.treq.Request)
if err != nil {
return
}
}
type requestAndChan struct {
- _ incomparable
- req *Request
- cancelKey cancelKey
- ch chan responseAndError // unbuffered; always send in select on callerGone
+ _ incomparable
+ treq *transportRequest
+ ch chan responseAndError // unbuffered; always send in select on callerGone
// whether the Transport (as opposed to the user client code)
// added the Accept-Encoding gzip header. If the Transport
var errRequestCanceled = http2errRequestCanceled
var errRequestCanceledConn = errors.New("net/http: request canceled while waiting for connection") // TODO: unify?
+// errRequestDone is used to cancel the round trip Context after a request is successfully done.
+// It should not be seen by the user.
+var errRequestDone = errors.New("net/http: request completed")
+
func nop() {}
// testHooks. Always non-nil.
func (pc *persistConn) roundTrip(req *transportRequest) (resp *Response, err error) {
testHookEnterRoundTrip()
- if !pc.t.replaceReqCanceler(req.cancelKey, pc.cancelRequest) {
- pc.t.putOrCloseIdleConn(pc)
- return nil, errRequestCanceled
- }
pc.mu.Lock()
pc.numExpectedResponses++
headerFn := pc.mutateHeaderFunc
gone := make(chan struct{})
defer close(gone)
- defer func() {
- if err != nil {
- pc.t.setReqCanceler(req.cancelKey, nil)
- }
- }()
-
const debugRoundTrip = false
// Write the request concurrently with waiting for a response,
resc := make(chan responseAndError)
pc.reqch <- requestAndChan{
- req: req.Request,
- cancelKey: req.cancelKey,
+ treq: req,
ch: resc,
addedGzip: requestedGzip,
continueCh: continueCh,
callerGone: gone,
}
+ handleResponse := func(re responseAndError) (*Response, error) {
+ if (re.res == nil) == (re.err == nil) {
+ panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
+ }
+ if debugRoundTrip {
+ req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
+ }
+ if re.err != nil {
+ return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
+ }
+ return re.res, nil
+ }
+
var respHeaderTimer <-chan time.Time
- cancelChan := req.Request.Cancel
- ctxDoneChan := req.Context().Done()
+ ctxDoneChan := req.ctx.Done()
pcClosed := pc.closech
- canceled := false
for {
testHookWaitResLoop()
select {
respHeaderTimer = timer.C
}
case <-pcClosed:
- pcClosed = nil
- if canceled || pc.t.replaceReqCanceler(req.cancelKey, nil) {
- if debugRoundTrip {
- req.logf("closech recv: %T %#v", pc.closed, pc.closed)
- }
- return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
+ select {
+ case re := <-resc:
+ // The pconn closing raced with the response to the request,
+ // probably after the server wrote a response and immediately
+ // closed the connection. Use the response.
+ return handleResponse(re)
+ default:
}
+ if debugRoundTrip {
+ req.logf("closech recv: %T %#v", pc.closed, pc.closed)
+ }
+ return nil, pc.mapRoundTripError(req, startBytesWritten, pc.closed)
case <-respHeaderTimer:
if debugRoundTrip {
req.logf("timeout waiting for response headers.")
pc.close(errTimeout)
return nil, errTimeout
case re := <-resc:
- if (re.res == nil) == (re.err == nil) {
- panic(fmt.Sprintf("internal error: exactly one of res or err should be set; nil=%v", re.res == nil))
- }
- if debugRoundTrip {
- req.logf("resc recv: %p, %T/%#v", re.res, re.err, re.err)
- }
- if re.err != nil {
- return nil, pc.mapRoundTripError(req, startBytesWritten, re.err)
- }
- return re.res, nil
- case <-cancelChan:
- canceled = pc.t.cancelRequest(req.cancelKey, errRequestCanceled)
- cancelChan = nil
+ return handleResponse(re)
case <-ctxDoneChan:
- canceled = pc.t.cancelRequest(req.cancelKey, req.Context().Err())
- cancelChan = nil
- ctxDoneChan = nil
+ select {
+ case re := <-resc:
+ // readLoop is responsible for canceling req.ctx after
+ // it reads the response body. Check for a response racing
+ // the context close, and use the response if available.
+ return handleResponse(re)
+ default:
+ }
+ pc.cancelRequest(context.Cause(req.ctx))
}
}
}