Source file src/net/http/internal/http2/transport.go

     1  // Copyright 2015 The Go Authors. All rights reserved.
     2  // Use of this source code is governed by a BSD-style
     3  // license that can be found in the LICENSE file.
     4  
     5  // Transport code.
     6  
     7  package http2
     8  
     9  import (
    10  	"bufio"
    11  	"bytes"
    12  	"compress/flate"
    13  	"compress/gzip"
    14  	"context"
    15  	"crypto/rand"
    16  	"crypto/tls"
    17  	"errors"
    18  	"fmt"
    19  	"io"
    20  	"io/fs"
    21  	"log"
    22  	"math"
    23  	"math/bits"
    24  	mathrand "math/rand"
    25  	"net"
    26  	"net/http/httptrace"
    27  	"net/http/internal"
    28  	"net/http/internal/httpcommon"
    29  	"net/textproto"
    30  	"slices"
    31  	"strconv"
    32  	"strings"
    33  	"sync"
    34  	"sync/atomic"
    35  	"time"
    36  
    37  	"golang.org/x/net/http/httpguts"
    38  	"golang.org/x/net/http2/hpack"
    39  	"golang.org/x/net/idna"
    40  )
    41  
    42  const (
    43  	// transportDefaultConnFlow is how many connection-level flow control
    44  	// tokens we give the server at start-up, past the default 64k.
    45  	transportDefaultConnFlow = 1 << 30
    46  
    47  	// transportDefaultStreamFlow is how many stream-level flow
    48  	// control tokens we announce to the peer, and how many bytes
    49  	// we buffer per stream.
    50  	transportDefaultStreamFlow = 4 << 20
    51  
    52  	defaultUserAgent = "Go-http-client/2.0"
    53  
    54  	// initialMaxConcurrentStreams is a connections maxConcurrentStreams until
    55  	// it's received servers initial SETTINGS frame, which corresponds with the
    56  	// spec's minimum recommended value.
    57  	initialMaxConcurrentStreams = 100
    58  
    59  	// defaultMaxConcurrentStreams is a connections default maxConcurrentStreams
    60  	// if the server doesn't include one in its initial SETTINGS frame.
    61  	defaultMaxConcurrentStreams = 1000
    62  )
    63  
    64  // Transport is an HTTP/2 Transport.
    65  //
    66  // A Transport internally caches connections to servers. It is safe
    67  // for concurrent use by multiple goroutines.
    68  type Transport struct {
    69  	t1       TransportConfig
    70  	connPool noDialClientConnPool
    71  	*transportTestHooks
    72  }
    73  
    74  // Hook points used for testing.
    75  // Outside of tests, t.transportTestHooks is nil and these all have minimal implementations.
    76  // Inside tests, see the testSyncHooks function docs.
    77  
    78  type transportTestHooks struct {
    79  	newclientconn func(*ClientConn)
    80  }
    81  
    82  func (t *Transport) maxHeaderListSize() uint32 {
    83  	n := t.t1.MaxHeaderListSize()
    84  	if b := t.t1.MaxResponseHeaderBytes(); b != 0 {
    85  		n = b
    86  		if n > 0 {
    87  			n = adjustHTTP1MaxHeaderSize(n)
    88  		}
    89  	}
    90  	if n <= 0 {
    91  		return 10 << 20
    92  	}
    93  	if n >= 0xffffffff {
    94  		return 0
    95  	}
    96  	return uint32(n)
    97  }
    98  
    99  func (t *Transport) disableCompression() bool {
   100  	return t.t1 != nil && t.t1.DisableCompression()
   101  }
   102  
   103  func NewTransport(t1 TransportConfig) *Transport {
   104  	connPool := new(clientConnPool)
   105  	t2 := &Transport{
   106  		connPool: noDialClientConnPool{connPool},
   107  		t1:       t1,
   108  	}
   109  	connPool.t = t2
   110  	return t2
   111  }
   112  
   113  func (t *Transport) AddConn(scheme, authority string, c net.Conn) error {
   114  	addr := authorityAddr(scheme, authority)
   115  	used, err := t.connPool.addConnIfNeeded(addr, t, c)
   116  	if !used {
   117  		go c.Close()
   118  	}
   119  	return err
   120  }
   121  
   122  // unencryptedTransport is a Transport with a RoundTrip method that
   123  // always permits http:// URLs.
   124  type unencryptedTransport Transport
   125  
   126  func (t *unencryptedTransport) RoundTrip(req *ClientRequest) (*ClientResponse, error) {
   127  	return (*Transport)(t).RoundTripOpt(req, RoundTripOpt{})
   128  }
   129  
   130  // ClientConn is the state of a single HTTP/2 client connection to an
   131  // HTTP/2 server.
   132  type ClientConn struct {
   133  	t             *Transport
   134  	tconn         net.Conn             // usually *tls.Conn, except specialized impls
   135  	tlsState      *tls.ConnectionState // nil only for specialized impls
   136  	atomicReused  uint32               // whether conn is being reused; atomic
   137  	singleUse     bool                 // whether being used for a single http.Request
   138  	getConnCalled bool                 // used by clientConnPool
   139  
   140  	// readLoop goroutine fields:
   141  	readerDone chan struct{} // closed on error
   142  	readerErr  error         // set before readerDone is closed
   143  
   144  	idleTimeout time.Duration // or 0 for never
   145  	idleTimer   *time.Timer
   146  
   147  	mu               sync.Mutex // guards following
   148  	cond             *sync.Cond // hold mu; broadcast on flow/closed changes
   149  	flow             outflow    // our conn-level flow control quota (cs.outflow is per stream)
   150  	inflow           inflow     // peer's conn-level flow control
   151  	doNotReuse       bool       // whether conn is marked to not be reused for any future requests
   152  	closing          bool
   153  	closed           bool
   154  	closedOnIdle     bool                     // true if conn was closed for idleness
   155  	seenSettings     bool                     // true if we've seen a settings frame, false otherwise
   156  	seenSettingsChan chan struct{}            // closed when seenSettings is true or frame reading fails
   157  	wantSettingsAck  bool                     // we sent a SETTINGS frame and haven't heard back
   158  	goAway           *GoAwayFrame             // if non-nil, the GoAwayFrame we received
   159  	goAwayDebug      string                   // goAway frame's debug data, retained as a string
   160  	streams          map[uint32]*clientStream // client-initiated
   161  	streamsReserved  int                      // incr by ReserveNewRequest; decr on RoundTrip
   162  	nextStreamID     uint32
   163  	pendingRequests  int                       // requests blocked and waiting to be sent because len(streams) == maxConcurrentStreams
   164  	pings            map[[8]byte]chan struct{} // in flight ping data to notification channel
   165  	br               *bufio.Reader
   166  	lastActive       time.Time
   167  	lastIdle         time.Time // time last idle
   168  	// Settings from peer: (also guarded by wmu)
   169  	maxFrameSize                uint32
   170  	maxConcurrentStreams        uint32
   171  	peerMaxHeaderListSize       uint64
   172  	peerMaxHeaderTableSize      uint32
   173  	initialWindowSize           uint32
   174  	initialStreamRecvWindowSize int32
   175  	readIdleTimeout             time.Duration
   176  	pingTimeout                 time.Duration
   177  	extendedConnectAllowed      bool
   178  	strictMaxConcurrentStreams  bool
   179  
   180  	// rstStreamPingsBlocked works around an unfortunate gRPC behavior.
   181  	// gRPC strictly limits the number of PING frames that it will receive.
   182  	// The default is two pings per two hours, but the limit resets every time
   183  	// the gRPC endpoint sends a HEADERS or DATA frame. See golang/go#70575.
   184  	//
   185  	// rstStreamPingsBlocked is set after receiving a response to a PING frame
   186  	// bundled with an RST_STREAM (see pendingResets below), and cleared after
   187  	// receiving a HEADERS or DATA frame.
   188  	rstStreamPingsBlocked bool
   189  
   190  	// pendingResets is the number of RST_STREAM frames we have sent to the peer,
   191  	// without confirming that the peer has received them. When we send a RST_STREAM,
   192  	// we bundle it with a PING frame, unless a PING is already in flight. We count
   193  	// the reset stream against the connection's concurrency limit until we get
   194  	// a PING response. This limits the number of requests we'll try to send to a
   195  	// completely unresponsive connection.
   196  	pendingResets int
   197  
   198  	// readBeforeStreamID is the smallest stream ID that has not been followed by
   199  	// a frame read from the peer. We use this to determine when a request may
   200  	// have been sent to a completely unresponsive connection:
   201  	// If the request ID is less than readBeforeStreamID, then we have had some
   202  	// indication of life on the connection since sending the request.
   203  	readBeforeStreamID uint32
   204  
   205  	// reqHeaderMu is a 1-element semaphore channel controlling access to sending new requests.
   206  	// Write to reqHeaderMu to lock it, read from it to unlock.
   207  	// Lock reqmu BEFORE mu or wmu.
   208  	reqHeaderMu chan struct{}
   209  
   210  	// internalStateHook reports state changes back to the net/http.ClientConn.
   211  	// Note that this is different from the user state hook registered by
   212  	// net/http.ClientConn.SetStateHook: The internal hook calls ClientConn,
   213  	// which calls the user hook.
   214  	internalStateHook func()
   215  
   216  	// wmu is held while writing.
   217  	// Acquire BEFORE mu when holding both, to avoid blocking mu on network writes.
   218  	// Only acquire both at the same time when changing peer settings.
   219  	wmu  sync.Mutex
   220  	bw   *bufio.Writer
   221  	fr   *Framer
   222  	werr error        // first write error that has occurred
   223  	hbuf bytes.Buffer // HPACK encoder writes into this
   224  	henc *hpack.Encoder
   225  }
   226  
   227  // clientStream is the state for a single HTTP/2 stream. One of these
   228  // is created for each Transport.RoundTrip call.
   229  type clientStream struct {
   230  	cc *ClientConn
   231  
   232  	// Fields of Request that we may access even after the response body is closed.
   233  	ctx       context.Context
   234  	reqCancel <-chan struct{}
   235  
   236  	trace         *httptrace.ClientTrace // or nil
   237  	ID            uint32
   238  	bufPipe       pipe // buffered pipe with the flow-controlled response payload
   239  	requestedGzip bool
   240  	isHead        bool
   241  
   242  	abortOnce sync.Once
   243  	abort     chan struct{} // closed to signal stream should end immediately
   244  	abortErr  error         // set if abort is closed
   245  
   246  	peerClosed chan struct{} // closed when the peer sends an END_STREAM flag
   247  	donec      chan struct{} // closed after the stream is in the closed state
   248  	on100      chan struct{} // buffered; written to if a 100 is received
   249  
   250  	respHeaderRecv chan struct{}   // closed when headers are received
   251  	res            *ClientResponse // set if respHeaderRecv is closed
   252  
   253  	flow        outflow // guarded by cc.mu
   254  	inflow      inflow  // guarded by cc.mu
   255  	bytesRemain int64   // -1 means unknown; owned by transportResponseBody.Read
   256  	readErr     error   // sticky read error; owned by transportResponseBody.Read
   257  
   258  	reqBody              io.ReadCloser
   259  	reqBodyContentLength int64         // -1 means unknown
   260  	reqBodyClosed        chan struct{} // guarded by cc.mu; non-nil on Close, closed when done
   261  
   262  	// owned by writeRequest:
   263  	sentEndStream bool // sent an END_STREAM flag to the peer
   264  	sentHeaders   bool
   265  
   266  	// owned by clientConnReadLoop:
   267  	firstByte       bool  // got the first response byte
   268  	pastHeaders     bool  // got first MetaHeadersFrame (actual headers)
   269  	pastTrailers    bool  // got optional second MetaHeadersFrame (trailers)
   270  	readClosed      bool  // peer sent an END_STREAM flag
   271  	readAborted     bool  // read loop reset the stream
   272  	totalHeaderSize int64 // total size of 1xx headers seen
   273  
   274  	trailer    Header  // accumulated trailers
   275  	resTrailer *Header // client's Response.Trailer
   276  
   277  	staticResp ClientResponse
   278  }
   279  
   280  var got1xxFuncForTests func(int, textproto.MIMEHeader) error
   281  
   282  // get1xxTraceFunc returns the value of request's httptrace.ClientTrace.Got1xxResponse func,
   283  // if any. It returns nil if not set or if the Go version is too old.
   284  func (cs *clientStream) get1xxTraceFunc() func(int, textproto.MIMEHeader) error {
   285  	if fn := got1xxFuncForTests; fn != nil {
   286  		return fn
   287  	}
   288  	return traceGot1xxResponseFunc(cs.trace)
   289  }
   290  
   291  func (cs *clientStream) abortStream(err error) {
   292  	cs.cc.mu.Lock()
   293  	defer cs.cc.mu.Unlock()
   294  	cs.abortStreamLocked(err)
   295  }
   296  
   297  func (cs *clientStream) abortStreamLocked(err error) {
   298  	cs.abortOnce.Do(func() {
   299  		cs.abortErr = err
   300  		close(cs.abort)
   301  	})
   302  	if cs.reqBody != nil {
   303  		cs.closeReqBodyLocked()
   304  	}
   305  	// TODO(dneil): Clean up tests where cs.cc.cond is nil.
   306  	if cs.cc.cond != nil {
   307  		// Wake up writeRequestBody if it is waiting on flow control.
   308  		cs.cc.cond.Broadcast()
   309  	}
   310  }
   311  
   312  func (cs *clientStream) abortRequestBodyWrite() {
   313  	cc := cs.cc
   314  	cc.mu.Lock()
   315  	defer cc.mu.Unlock()
   316  	if cs.reqBody != nil && cs.reqBodyClosed == nil {
   317  		cs.closeReqBodyLocked()
   318  		cc.cond.Broadcast()
   319  	}
   320  }
   321  
   322  func (cs *clientStream) closeReqBodyLocked() {
   323  	if cs.reqBodyClosed != nil {
   324  		return
   325  	}
   326  	cs.reqBodyClosed = make(chan struct{})
   327  	reqBodyClosed := cs.reqBodyClosed
   328  	go func() {
   329  		cs.reqBody.Close()
   330  		close(reqBodyClosed)
   331  	}()
   332  }
   333  
   334  type stickyErrWriter struct {
   335  	conn    net.Conn
   336  	timeout time.Duration
   337  	err     *error
   338  }
   339  
   340  func (sew stickyErrWriter) Write(p []byte) (n int, err error) {
   341  	if *sew.err != nil {
   342  		return 0, *sew.err
   343  	}
   344  	n, err = writeWithByteTimeout(sew.conn, sew.timeout, p)
   345  	*sew.err = err
   346  	return n, err
   347  }
   348  
   349  // noCachedConnError is the concrete type of ErrNoCachedConn, which
   350  // needs to be detected by net/http regardless of whether it's its
   351  // bundled version (in h2_bundle.go with a rewritten type name) or
   352  // from a user's x/net/http2. As such, as it has a unique method name
   353  // (IsHTTP2NoCachedConnError) that net/http sniffs for via func
   354  // isNoCachedConnError.
   355  type noCachedConnError struct{}
   356  
   357  func (noCachedConnError) IsHTTP2NoCachedConnError() {}
   358  func (noCachedConnError) Error() string             { return "http2: no cached connection was available" }
   359  
   360  // isNoCachedConnError reports whether err is of type noCachedConnError
   361  // or its equivalent renamed type in net/http2's h2_bundle.go. Both types
   362  // may coexist in the same running program.
   363  func isNoCachedConnError(err error) bool {
   364  	_, ok := err.(interface{ IsHTTP2NoCachedConnError() })
   365  	return ok
   366  }
   367  
   368  var ErrNoCachedConn error = noCachedConnError{}
   369  
   370  // RoundTripOpt are options for the Transport.RoundTripOpt method.
   371  type RoundTripOpt struct {
   372  	// OnlyCachedConn controls whether RoundTripOpt may
   373  	// create a new TCP connection. If set true and
   374  	// no cached connection is available, RoundTripOpt
   375  	// will return ErrNoCachedConn.
   376  	OnlyCachedConn bool
   377  }
   378  
   379  func (t *Transport) RoundTrip(req *ClientRequest) (*ClientResponse, error) {
   380  	return t.RoundTripOpt(req, RoundTripOpt{})
   381  }
   382  
   383  // authorityAddr returns a given authority (a host/IP, or host:port / ip:port)
   384  // and returns a host:port. The port 443 is added if needed.
   385  func authorityAddr(scheme string, authority string) (addr string) {
   386  	host, port, err := net.SplitHostPort(authority)
   387  	if err != nil { // authority didn't have a port
   388  		host = authority
   389  		port = ""
   390  	}
   391  	if port == "" { // authority's port was empty
   392  		port = "443"
   393  		if scheme == "http" {
   394  			port = "80"
   395  		}
   396  	}
   397  	if a, err := idna.ToASCII(host); err == nil {
   398  		host = a
   399  	}
   400  	// IPv6 address literal, without a port:
   401  	if strings.HasPrefix(host, "[") && strings.HasSuffix(host, "]") {
   402  		return host + ":" + port
   403  	}
   404  	return net.JoinHostPort(host, port)
   405  }
   406  
   407  // RoundTripOpt is like RoundTrip, but takes options.
   408  func (t *Transport) RoundTripOpt(req *ClientRequest, opt RoundTripOpt) (*ClientResponse, error) {
   409  	switch req.URL.Scheme {
   410  	case "https":
   411  	case "http":
   412  	default:
   413  		return nil, errors.New("http2: unsupported scheme")
   414  	}
   415  
   416  	addr := authorityAddr(req.URL.Scheme, req.URL.Host)
   417  	for retry := 0; ; retry++ {
   418  		cc, err := t.connPool.GetClientConn(req, addr)
   419  		if err != nil {
   420  			t.vlogf("http2: Transport failed to get client conn for %s: %v", addr, err)
   421  			return nil, err
   422  		}
   423  		reused := !atomic.CompareAndSwapUint32(&cc.atomicReused, 0, 1)
   424  		traceGotConn(req, cc, reused)
   425  		res, err := cc.RoundTrip(req)
   426  		if err != nil && retry <= 6 {
   427  			roundTripErr := err
   428  			if req, err = shouldRetryRequest(req, err); err == nil {
   429  				// After the first retry, do exponential backoff with 10% jitter.
   430  				if retry == 0 {
   431  					t.vlogf("RoundTrip retrying after failure: %v", roundTripErr)
   432  					continue
   433  				}
   434  				backoff := float64(uint(1) << (uint(retry) - 1))
   435  				backoff += backoff * (0.1 * mathrand.Float64())
   436  				d := time.Second * time.Duration(backoff)
   437  				tm := time.NewTimer(d)
   438  				select {
   439  				case <-tm.C:
   440  					t.vlogf("RoundTrip retrying after failure: %v", roundTripErr)
   441  					continue
   442  				case <-req.Context.Done():
   443  					tm.Stop()
   444  					err = req.Context.Err()
   445  				}
   446  			}
   447  		}
   448  		if err == errClientConnNotEstablished {
   449  			// This ClientConn was created recently,
   450  			// this is the first request to use it,
   451  			// and the connection is closed and not usable.
   452  			//
   453  			// In this state, cc.idleTimer will remove the conn from the pool
   454  			// when it fires. Stop the timer and remove it here so future requests
   455  			// won't try to use this connection.
   456  			//
   457  			// If the timer has already fired and we're racing it, the redundant
   458  			// call to MarkDead is harmless.
   459  			if cc.idleTimer != nil {
   460  				cc.idleTimer.Stop()
   461  			}
   462  			t.connPool.MarkDead(cc)
   463  		}
   464  		if err != nil {
   465  			t.vlogf("RoundTrip failure: %v", err)
   466  			return nil, err
   467  		}
   468  		return res, nil
   469  	}
   470  }
   471  
   472  func (t *Transport) IdleConnStrsForTesting() []string {
   473  	var ret []string
   474  	t.connPool.mu.Lock()
   475  	defer t.connPool.mu.Unlock()
   476  	for k, ccs := range t.connPool.conns {
   477  		for _, cc := range ccs {
   478  			if cc.idleState().canTakeNewRequest {
   479  				ret = append(ret, k)
   480  			}
   481  		}
   482  	}
   483  	slices.Sort(ret)
   484  	return ret
   485  }
   486  
   487  // CloseIdleConnections closes any connections which were previously
   488  // connected from previous requests but are now sitting idle.
   489  // It does not interrupt any connections currently in use.
   490  func (t *Transport) CloseIdleConnections() {
   491  	t.connPool.closeIdleConnections()
   492  }
   493  
   494  var (
   495  	errClientConnClosed         = errors.New("http2: client conn is closed")
   496  	errClientConnUnusable       = errors.New("http2: client conn not usable")
   497  	errClientConnNotEstablished = errors.New("http2: client conn could not be established")
   498  	errClientConnGotGoAway      = errors.New("http2: Transport received Server's graceful shutdown GOAWAY")
   499  	errClientConnForceClosed    = errors.New("http2: client connection force closed via ClientConn.Close")
   500  )
   501  
   502  // shouldRetryRequest is called by RoundTrip when a request fails to get
   503  // response headers. It is always called with a non-nil error.
   504  // It returns either a request to retry or an error if the request can't be replayed.
   505  // If the request is retried, it always clones the request (since requests
   506  // contain an unreusable clientStream).
   507  func shouldRetryRequest(req *ClientRequest, err error) (*ClientRequest, error) {
   508  	if !canRetryError(err) {
   509  		return nil, err
   510  	}
   511  	// If the Body is nil (or http.NoBody), it's safe to reuse this request's Body.
   512  	if req.Body == nil || req.Body == NoBody {
   513  		return req.Clone(), nil
   514  	}
   515  
   516  	// If the request body can be reset back to its original
   517  	// state via the optional req.GetBody, do that.
   518  	if req.GetBody != nil {
   519  		body, err := req.GetBody()
   520  		if err != nil {
   521  			return nil, err
   522  		}
   523  		newReq := req.Clone()
   524  		newReq.Body = body
   525  		return newReq, nil
   526  	}
   527  
   528  	// The Request.Body can't reset back to the beginning, but we
   529  	// don't seem to have started to read from it yet, so reuse the body.
   530  	if err == errClientConnUnusable {
   531  		return req.Clone(), nil
   532  	}
   533  
   534  	return nil, fmt.Errorf("http2: Transport: cannot retry err [%v] after Request.Body was written; define Request.GetBody to avoid this error", err)
   535  }
   536  
   537  func canRetryError(err error) bool {
   538  	if err == errClientConnUnusable || err == errClientConnGotGoAway {
   539  		return true
   540  	}
   541  	if se, ok := err.(StreamError); ok {
   542  		return se.Code == ErrCodeRefusedStream
   543  	}
   544  	return false
   545  }
   546  
   547  func (t *Transport) dialClientConn(ctx context.Context, addr string, singleUse bool) (*ClientConn, error) {
   548  	if t.transportTestHooks != nil {
   549  		return t.newClientConn(nil, singleUse, nil)
   550  	}
   551  	host, _, err := net.SplitHostPort(addr)
   552  	if err != nil {
   553  		return nil, err
   554  	}
   555  	tconn, err := t.dialTLS(ctx, "tcp", addr, t.newTLSConfig(host))
   556  	if err != nil {
   557  		return nil, err
   558  	}
   559  	return t.newClientConn(tconn, singleUse, nil)
   560  }
   561  
   562  func (t *Transport) newTLSConfig(host string) *tls.Config {
   563  	cfg := new(tls.Config)
   564  	if !slices.Contains(cfg.NextProtos, NextProtoTLS) {
   565  		cfg.NextProtos = append([]string{NextProtoTLS}, cfg.NextProtos...)
   566  	}
   567  	if cfg.ServerName == "" {
   568  		cfg.ServerName = host
   569  	}
   570  	return cfg
   571  }
   572  
   573  func (t *Transport) dialTLS(ctx context.Context, network, addr string, tlsCfg *tls.Config) (net.Conn, error) {
   574  	tlsCn, err := t.dialTLSWithContext(ctx, network, addr, tlsCfg)
   575  	if err != nil {
   576  		return nil, err
   577  	}
   578  	state := tlsCn.ConnectionState()
   579  	if p := state.NegotiatedProtocol; p != NextProtoTLS {
   580  		return nil, fmt.Errorf("http2: unexpected ALPN protocol %q; want %q", p, NextProtoTLS)
   581  	}
   582  	if !state.NegotiatedProtocolIsMutual {
   583  		return nil, errors.New("http2: could not negotiate protocol mutually")
   584  	}
   585  	return tlsCn, nil
   586  }
   587  
   588  // disableKeepAlives reports whether connections should be closed as
   589  // soon as possible after handling the first request.
   590  func (t *Transport) disableKeepAlives() bool {
   591  	return t.t1 != nil && t.t1.DisableKeepAlives()
   592  }
   593  
   594  func (t *Transport) expectContinueTimeout() time.Duration {
   595  	if t.t1 == nil {
   596  		return 0
   597  	}
   598  	return t.t1.ExpectContinueTimeout()
   599  }
   600  
   601  func (t *Transport) NewClientConn(c net.Conn, internalStateHook func()) (NetHTTPClientConn, error) {
   602  	cc, err := t.newClientConn(c, t.disableKeepAlives(), internalStateHook)
   603  	if err != nil {
   604  		return NetHTTPClientConn{}, err
   605  	}
   606  
   607  	// RoundTrip should block when the conn is at its concurrency limit,
   608  	// not return an error. Setting strictMaxConcurrentStreams enables this.
   609  	cc.strictMaxConcurrentStreams = true
   610  
   611  	return NetHTTPClientConn{cc}, nil
   612  }
   613  
   614  func (t *Transport) newClientConn(c net.Conn, singleUse bool, internalStateHook func()) (*ClientConn, error) {
   615  	conf := configFromTransport(t)
   616  	cc := &ClientConn{
   617  		t:                           t,
   618  		tconn:                       c,
   619  		readerDone:                  make(chan struct{}),
   620  		nextStreamID:                1,
   621  		maxFrameSize:                16 << 10, // spec default
   622  		initialWindowSize:           65535,    // spec default
   623  		initialStreamRecvWindowSize: int32(conf.MaxReceiveBufferPerStream),
   624  		maxConcurrentStreams:        initialMaxConcurrentStreams, // "infinite", per spec. Use a smaller value until we have received server settings.
   625  		strictMaxConcurrentStreams:  conf.StrictMaxConcurrentRequests,
   626  		peerMaxHeaderListSize:       0xffffffffffffffff, // "infinite", per spec. Use 2^64-1 instead.
   627  		streams:                     make(map[uint32]*clientStream),
   628  		singleUse:                   singleUse,
   629  		seenSettingsChan:            make(chan struct{}),
   630  		wantSettingsAck:             true,
   631  		readIdleTimeout:             conf.SendPingTimeout,
   632  		pingTimeout:                 conf.PingTimeout,
   633  		pings:                       make(map[[8]byte]chan struct{}),
   634  		reqHeaderMu:                 make(chan struct{}, 1),
   635  		lastActive:                  time.Now(),
   636  		internalStateHook:           internalStateHook,
   637  	}
   638  	if t.transportTestHooks != nil {
   639  		t.transportTestHooks.newclientconn(cc)
   640  		c = cc.tconn
   641  	}
   642  	if VerboseLogs {
   643  		t.vlogf("http2: Transport creating client conn %p to %v", cc, c.RemoteAddr())
   644  	}
   645  
   646  	cc.cond = sync.NewCond(&cc.mu)
   647  	cc.flow.add(int32(initialWindowSize))
   648  
   649  	// TODO: adjust this writer size to account for frame size +
   650  	// MTU + crypto/tls record padding.
   651  	cc.bw = bufio.NewWriter(stickyErrWriter{
   652  		conn:    c,
   653  		timeout: conf.WriteByteTimeout,
   654  		err:     &cc.werr,
   655  	})
   656  	cc.br = bufio.NewReader(c)
   657  	cc.fr = NewFramer(cc.bw, cc.br)
   658  	cc.fr.SetMaxReadFrameSize(uint32(conf.MaxReadFrameSize))
   659  	if conf.CountError != nil {
   660  		cc.fr.countError = conf.CountError
   661  	}
   662  	maxHeaderTableSize := uint32(conf.MaxDecoderHeaderTableSize)
   663  	cc.fr.ReadMetaHeaders = hpack.NewDecoder(maxHeaderTableSize, nil)
   664  	cc.fr.MaxHeaderListSize = t.maxHeaderListSize()
   665  
   666  	cc.henc = hpack.NewEncoder(&cc.hbuf)
   667  	cc.henc.SetMaxDynamicTableSizeLimit(uint32(conf.MaxEncoderHeaderTableSize))
   668  	cc.peerMaxHeaderTableSize = initialHeaderTableSize
   669  
   670  	if cs, ok := c.(connectionStater); ok {
   671  		state := cs.ConnectionState()
   672  		cc.tlsState = &state
   673  	}
   674  
   675  	initialSettings := []Setting{
   676  		{ID: SettingEnablePush, Val: 0},
   677  		{ID: SettingInitialWindowSize, Val: uint32(cc.initialStreamRecvWindowSize)},
   678  	}
   679  	initialSettings = append(initialSettings, Setting{ID: SettingMaxFrameSize, Val: uint32(conf.MaxReadFrameSize)})
   680  	if max := t.maxHeaderListSize(); max != 0 {
   681  		initialSettings = append(initialSettings, Setting{ID: SettingMaxHeaderListSize, Val: max})
   682  	}
   683  	if maxHeaderTableSize != initialHeaderTableSize {
   684  		initialSettings = append(initialSettings, Setting{ID: SettingHeaderTableSize, Val: maxHeaderTableSize})
   685  	}
   686  
   687  	cc.bw.Write(clientPreface)
   688  	cc.fr.WriteSettings(initialSettings...)
   689  	cc.fr.WriteWindowUpdate(0, uint32(conf.MaxReceiveBufferPerConnection))
   690  	cc.inflow.init(int32(conf.MaxReceiveBufferPerConnection) + initialWindowSize)
   691  	cc.bw.Flush()
   692  	if cc.werr != nil {
   693  		cc.Close()
   694  		return nil, cc.werr
   695  	}
   696  
   697  	// Start the idle timer after the connection is fully initialized.
   698  	if d := t.idleConnTimeout(); d != 0 {
   699  		cc.idleTimeout = d
   700  		cc.idleTimer = time.AfterFunc(d, cc.onIdleTimeout)
   701  	}
   702  
   703  	go cc.readLoop()
   704  	return cc, nil
   705  }
   706  
   707  func (cc *ClientConn) healthCheck() {
   708  	pingTimeout := cc.pingTimeout
   709  	// We don't need to periodically ping in the health check, because the readLoop of ClientConn will
   710  	// trigger the healthCheck again if there is no frame received.
   711  	ctx, cancel := context.WithTimeout(context.Background(), pingTimeout)
   712  	defer cancel()
   713  	cc.vlogf("http2: Transport sending health check")
   714  	err := cc.Ping(ctx)
   715  	if err != nil {
   716  		cc.vlogf("http2: Transport health check failure: %v", err)
   717  		cc.closeForLostPing()
   718  	} else {
   719  		cc.vlogf("http2: Transport health check success")
   720  	}
   721  }
   722  
   723  // SetDoNotReuse marks cc as not reusable for future HTTP requests.
   724  func (cc *ClientConn) SetDoNotReuse() {
   725  	cc.mu.Lock()
   726  	defer cc.mu.Unlock()
   727  	cc.doNotReuse = true
   728  }
   729  
   730  // CanTakeNewRequest reports whether the connection can take a new request,
   731  // meaning it has not been closed or received or sent a GOAWAY.
   732  //
   733  // If the caller is going to immediately make a new request on this
   734  // connection, use ReserveNewRequest instead.
   735  func (cc *ClientConn) CanTakeNewRequest() bool {
   736  	cc.mu.Lock()
   737  	defer cc.mu.Unlock()
   738  	return cc.canTakeNewRequestLocked()
   739  }
   740  
   741  // ReserveNewRequest is like CanTakeNewRequest but also reserves a
   742  // concurrent stream in cc. The reservation is decremented on the
   743  // next call to RoundTrip.
   744  func (cc *ClientConn) ReserveNewRequest() bool {
   745  	cc.mu.Lock()
   746  	defer cc.mu.Unlock()
   747  	if st := cc.idleStateLocked(); !st.canTakeNewRequest {
   748  		return false
   749  	}
   750  	cc.streamsReserved++
   751  	return true
   752  }
   753  
   754  // ClientConnState describes the state of a ClientConn.
   755  type ClientConnState struct {
   756  	// Closed is whether the connection is closed.
   757  	Closed bool
   758  
   759  	// Closing is whether the connection is in the process of
   760  	// closing. It may be closing due to shutdown, being a
   761  	// single-use connection, being marked as DoNotReuse, or
   762  	// having received a GOAWAY frame.
   763  	Closing bool
   764  
   765  	// StreamsActive is how many streams are active.
   766  	StreamsActive int
   767  
   768  	// StreamsReserved is how many streams have been reserved via
   769  	// ClientConn.ReserveNewRequest.
   770  	StreamsReserved int
   771  
   772  	// StreamsPending is how many requests have been sent in excess
   773  	// of the peer's advertised MaxConcurrentStreams setting and
   774  	// are waiting for other streams to complete.
   775  	StreamsPending int
   776  
   777  	// MaxConcurrentStreams is how many concurrent streams the
   778  	// peer advertised as acceptable. Zero means no SETTINGS
   779  	// frame has been received yet.
   780  	MaxConcurrentStreams uint32
   781  
   782  	// LastIdle, if non-zero, is when the connection last
   783  	// transitioned to idle state.
   784  	LastIdle time.Time
   785  }
   786  
   787  // State returns a snapshot of cc's state.
   788  func (cc *ClientConn) State() ClientConnState {
   789  	cc.wmu.Lock()
   790  	maxConcurrent := cc.maxConcurrentStreams
   791  	if !cc.seenSettings {
   792  		maxConcurrent = 0
   793  	}
   794  	cc.wmu.Unlock()
   795  
   796  	cc.mu.Lock()
   797  	defer cc.mu.Unlock()
   798  	return ClientConnState{
   799  		Closed:               cc.closed,
   800  		Closing:              cc.closing || cc.singleUse || cc.doNotReuse || cc.goAway != nil,
   801  		StreamsActive:        len(cc.streams) + cc.pendingResets,
   802  		StreamsReserved:      cc.streamsReserved,
   803  		StreamsPending:       cc.pendingRequests,
   804  		LastIdle:             cc.lastIdle,
   805  		MaxConcurrentStreams: maxConcurrent,
   806  	}
   807  }
   808  
   809  // clientConnIdleState describes the suitability of a client
   810  // connection to initiate a new RoundTrip request.
   811  type clientConnIdleState struct {
   812  	canTakeNewRequest bool
   813  }
   814  
   815  func (cc *ClientConn) idleState() clientConnIdleState {
   816  	cc.mu.Lock()
   817  	defer cc.mu.Unlock()
   818  	return cc.idleStateLocked()
   819  }
   820  
   821  func (cc *ClientConn) idleStateLocked() (st clientConnIdleState) {
   822  	if cc.singleUse && cc.nextStreamID > 1 {
   823  		return
   824  	}
   825  	var maxConcurrentOkay bool
   826  	if cc.strictMaxConcurrentStreams {
   827  		// We'll tell the caller we can take a new request to
   828  		// prevent the caller from dialing a new TCP
   829  		// connection, but then we'll block later before
   830  		// writing it.
   831  		maxConcurrentOkay = true
   832  	} else {
   833  		// We can take a new request if the total of
   834  		//   - active streams;
   835  		//   - reservation slots for new streams; and
   836  		//   - streams for which we have sent a RST_STREAM and a PING,
   837  		//     but received no subsequent frame
   838  		// is less than the concurrency limit.
   839  		maxConcurrentOkay = cc.currentRequestCountLocked() < int(cc.maxConcurrentStreams)
   840  	}
   841  
   842  	st.canTakeNewRequest = maxConcurrentOkay && cc.isUsableLocked()
   843  
   844  	// If this connection has never been used for a request and is closed,
   845  	// then let it take a request (which will fail).
   846  	// If the conn was closed for idleness, we're racing the idle timer;
   847  	// don't try to use the conn. (Issue #70515.)
   848  	//
   849  	// This avoids a situation where an error early in a connection's lifetime
   850  	// goes unreported.
   851  	if cc.nextStreamID == 1 && cc.streamsReserved == 0 && cc.closed && !cc.closedOnIdle {
   852  		st.canTakeNewRequest = true
   853  	}
   854  
   855  	return
   856  }
   857  
   858  func (cc *ClientConn) isUsableLocked() bool {
   859  	return cc.goAway == nil &&
   860  		!cc.closed &&
   861  		!cc.closing &&
   862  		!cc.doNotReuse &&
   863  		int64(cc.nextStreamID)+2*int64(cc.pendingRequests) < math.MaxInt32 &&
   864  		!cc.tooIdleLocked()
   865  }
   866  
   867  // canReserveLocked reports whether a net/http.ClientConn can reserve a slot on this conn.
   868  //
   869  // This follows slightly different rules than clientConnIdleState.canTakeNewRequest.
   870  // We only permit reservations up to the conn's concurrency limit.
   871  // This differs from ClientConn.ReserveNewRequest, which permits reservations
   872  // past the limit when StrictMaxConcurrentStreams is set.
   873  func (cc *ClientConn) canReserveLocked() bool {
   874  	if cc.currentRequestCountLocked() >= int(cc.maxConcurrentStreams) {
   875  		return false
   876  	}
   877  	if !cc.isUsableLocked() {
   878  		return false
   879  	}
   880  	return true
   881  }
   882  
   883  // currentRequestCountLocked reports the number of concurrency slots currently in use,
   884  // including active streams, reserved slots, and reset streams waiting for acknowledgement.
   885  func (cc *ClientConn) currentRequestCountLocked() int {
   886  	return len(cc.streams) + cc.streamsReserved + cc.pendingResets
   887  }
   888  
   889  func (cc *ClientConn) canTakeNewRequestLocked() bool {
   890  	st := cc.idleStateLocked()
   891  	return st.canTakeNewRequest
   892  }
   893  
   894  // availableLocked reports the number of concurrency slots available.
   895  func (cc *ClientConn) availableLocked() int {
   896  	if !cc.canTakeNewRequestLocked() {
   897  		return 0
   898  	}
   899  	return max(0, int(cc.maxConcurrentStreams)-cc.currentRequestCountLocked())
   900  }
   901  
   902  // tooIdleLocked reports whether this connection has been been sitting idle
   903  // for too much wall time.
   904  func (cc *ClientConn) tooIdleLocked() bool {
   905  	// The Round(0) strips the monontonic clock reading so the
   906  	// times are compared based on their wall time. We don't want
   907  	// to reuse a connection that's been sitting idle during
   908  	// VM/laptop suspend if monotonic time was also frozen.
   909  	return cc.idleTimeout != 0 && !cc.lastIdle.IsZero() && time.Since(cc.lastIdle.Round(0)) > cc.idleTimeout
   910  }
   911  
   912  // onIdleTimeout is called from a time.AfterFunc goroutine. It will
   913  // only be called when we're idle, but because we're coming from a new
   914  // goroutine, there could be a new request coming in at the same time,
   915  // so this simply calls the synchronized closeIfIdle to shut down this
   916  // connection. The timer could just call closeIfIdle, but this is more
   917  // clear.
   918  func (cc *ClientConn) onIdleTimeout() {
   919  	cc.closeIfIdle()
   920  }
   921  
   922  func (cc *ClientConn) closeConn() {
   923  	t := time.AfterFunc(250*time.Millisecond, cc.forceCloseConn)
   924  	defer t.Stop()
   925  	cc.tconn.Close()
   926  	cc.maybeCallStateHook()
   927  }
   928  
   929  // A tls.Conn.Close can hang for a long time if the peer is unresponsive.
   930  // Try to shut it down more aggressively.
   931  func (cc *ClientConn) forceCloseConn() {
   932  	tc, ok := cc.tconn.(*tls.Conn)
   933  	if !ok {
   934  		return
   935  	}
   936  	if nc := tc.NetConn(); nc != nil {
   937  		nc.Close()
   938  	}
   939  }
   940  
   941  func (cc *ClientConn) closeIfIdle() {
   942  	cc.mu.Lock()
   943  	if len(cc.streams) > 0 || cc.streamsReserved > 0 {
   944  		cc.mu.Unlock()
   945  		return
   946  	}
   947  	cc.closed = true
   948  	cc.closedOnIdle = true
   949  	nextID := cc.nextStreamID
   950  	// TODO: do clients send GOAWAY too? maybe? Just Close:
   951  	cc.mu.Unlock()
   952  
   953  	if VerboseLogs {
   954  		cc.vlogf("http2: Transport closing idle conn %p (forSingleUse=%v, maxStream=%v)", cc, cc.singleUse, nextID-2)
   955  	}
   956  	cc.closeConn()
   957  }
   958  
   959  func (cc *ClientConn) isDoNotReuseAndIdle() bool {
   960  	cc.mu.Lock()
   961  	defer cc.mu.Unlock()
   962  	return cc.doNotReuse && len(cc.streams) == 0
   963  }
   964  
   965  var shutdownEnterWaitStateHook = func() {}
   966  
   967  // Shutdown gracefully closes the client connection, waiting for running streams to complete.
   968  func (cc *ClientConn) Shutdown(ctx context.Context) error {
   969  	if err := cc.sendGoAway(); err != nil {
   970  		return err
   971  	}
   972  	// Wait for all in-flight streams to complete or connection to close
   973  	done := make(chan struct{})
   974  	cancelled := false // guarded by cc.mu
   975  	go func() {
   976  		cc.mu.Lock()
   977  		defer cc.mu.Unlock()
   978  		for {
   979  			if len(cc.streams) == 0 || cc.closed {
   980  				cc.closed = true
   981  				close(done)
   982  				break
   983  			}
   984  			if cancelled {
   985  				break
   986  			}
   987  			cc.cond.Wait()
   988  		}
   989  	}()
   990  	shutdownEnterWaitStateHook()
   991  	select {
   992  	case <-done:
   993  		cc.closeConn()
   994  		return nil
   995  	case <-ctx.Done():
   996  		cc.mu.Lock()
   997  		// Free the goroutine above
   998  		cancelled = true
   999  		cc.cond.Broadcast()
  1000  		cc.mu.Unlock()
  1001  		return ctx.Err()
  1002  	}
  1003  }
  1004  
  1005  func (cc *ClientConn) sendGoAway() error {
  1006  	cc.mu.Lock()
  1007  	closing := cc.closing
  1008  	cc.closing = true
  1009  	maxStreamID := cc.nextStreamID
  1010  	cc.mu.Unlock()
  1011  	if closing {
  1012  		// GOAWAY sent already
  1013  		return nil
  1014  	}
  1015  
  1016  	cc.wmu.Lock()
  1017  	defer cc.wmu.Unlock()
  1018  	// Send a graceful shutdown frame to server
  1019  	if err := cc.fr.WriteGoAway(maxStreamID, ErrCodeNo, nil); err != nil {
  1020  		return err
  1021  	}
  1022  	if err := cc.bw.Flush(); err != nil {
  1023  		return err
  1024  	}
  1025  	// Prevent new requests
  1026  	return nil
  1027  }
  1028  
  1029  // closes the client connection immediately. In-flight requests are interrupted.
  1030  // err is sent to streams.
  1031  func (cc *ClientConn) closeForError(err error) {
  1032  	cc.mu.Lock()
  1033  	cc.closed = true
  1034  	for _, cs := range cc.streams {
  1035  		cs.abortStreamLocked(err)
  1036  	}
  1037  	cc.cond.Broadcast()
  1038  	cc.mu.Unlock()
  1039  	cc.closeConn()
  1040  }
  1041  
  1042  // Close closes the client connection immediately.
  1043  //
  1044  // In-flight requests are interrupted. For a graceful shutdown, use Shutdown instead.
  1045  func (cc *ClientConn) Close() error {
  1046  	cc.closeForError(errClientConnForceClosed)
  1047  	return nil
  1048  }
  1049  
  1050  // closes the client connection immediately. In-flight requests are interrupted.
  1051  func (cc *ClientConn) closeForLostPing() {
  1052  	err := errors.New("http2: client connection lost")
  1053  	if f := cc.fr.countError; f != nil {
  1054  		f("conn_close_lost_ping")
  1055  	}
  1056  	cc.closeForError(err)
  1057  }
  1058  
  1059  // errRequestCanceled is a copy of net/http's errRequestCanceled because it's not
  1060  // exported. At least they'll be DeepEqual for h1-vs-h2 comparisons tests.
  1061  var errRequestCanceled = internal.ErrRequestCanceled
  1062  
  1063  func (cc *ClientConn) responseHeaderTimeout() time.Duration {
  1064  	if cc.t.t1 != nil {
  1065  		return cc.t.t1.ResponseHeaderTimeout()
  1066  	}
  1067  	// No way to do this (yet?) with just an http2.Transport. Probably
  1068  	// no need. Request.Cancel this is the new way. We only need to support
  1069  	// this for compatibility with the old http.Transport fields when
  1070  	// we're doing transparent http2.
  1071  	return 0
  1072  }
  1073  
  1074  // actualContentLength returns a sanitized version of
  1075  // req.ContentLength, where 0 actually means zero (not unknown) and -1
  1076  // means unknown.
  1077  func actualContentLength(req *ClientRequest) int64 {
  1078  	if req.Body == nil || req.Body == NoBody {
  1079  		return 0
  1080  	}
  1081  	if req.ContentLength != 0 {
  1082  		return req.ContentLength
  1083  	}
  1084  	return -1
  1085  }
  1086  
  1087  func (cc *ClientConn) decrStreamReservations() {
  1088  	cc.mu.Lock()
  1089  	defer cc.mu.Unlock()
  1090  	cc.decrStreamReservationsLocked()
  1091  }
  1092  
  1093  func (cc *ClientConn) decrStreamReservationsLocked() {
  1094  	if cc.streamsReserved > 0 {
  1095  		cc.streamsReserved--
  1096  	}
  1097  }
  1098  
  1099  func (cc *ClientConn) RoundTrip(req *ClientRequest) (*ClientResponse, error) {
  1100  	return cc.roundTrip(req, nil)
  1101  }
  1102  
  1103  func (cc *ClientConn) roundTrip(req *ClientRequest, streamf func(*clientStream)) (*ClientResponse, error) {
  1104  	ctx := req.Context
  1105  	req.stream = clientStream{
  1106  		cc:                   cc,
  1107  		ctx:                  ctx,
  1108  		reqCancel:            req.Cancel,
  1109  		isHead:               req.Method == "HEAD",
  1110  		reqBody:              req.Body,
  1111  		reqBodyContentLength: actualContentLength(req),
  1112  		trace:                httptrace.ContextClientTrace(ctx),
  1113  		peerClosed:           make(chan struct{}),
  1114  		abort:                make(chan struct{}),
  1115  		respHeaderRecv:       make(chan struct{}),
  1116  		donec:                make(chan struct{}),
  1117  		resTrailer:           req.ResTrailer,
  1118  	}
  1119  	cs := &req.stream
  1120  
  1121  	cs.requestedGzip = httpcommon.IsRequestGzip(req.Method, req.Header, cc.t.disableCompression())
  1122  
  1123  	go cs.doRequest(req, streamf)
  1124  
  1125  	waitDone := func() error {
  1126  		select {
  1127  		case <-cs.donec:
  1128  			return nil
  1129  		case <-ctx.Done():
  1130  			return ctx.Err()
  1131  		case <-cs.reqCancel:
  1132  			return errRequestCanceled
  1133  		}
  1134  	}
  1135  
  1136  	handleResponseHeaders := func() (*ClientResponse, error) {
  1137  		res := cs.res
  1138  		if res.StatusCode > 299 {
  1139  			// On error or status code 3xx, 4xx, 5xx, etc abort any
  1140  			// ongoing write, assuming that the server doesn't care
  1141  			// about our request body. If the server replied with 1xx or
  1142  			// 2xx, however, then assume the server DOES potentially
  1143  			// want our body (e.g. full-duplex streaming:
  1144  			// golang.org/issue/13444). If it turns out the server
  1145  			// doesn't, they'll RST_STREAM us soon enough. This is a
  1146  			// heuristic to avoid adding knobs to Transport. Hopefully
  1147  			// we can keep it.
  1148  			cs.abortRequestBodyWrite()
  1149  		}
  1150  		res.TLS = cc.tlsState
  1151  		if res.Body == NoBody && actualContentLength(req) == 0 {
  1152  			// If there isn't a request or response body still being
  1153  			// written, then wait for the stream to be closed before
  1154  			// RoundTrip returns.
  1155  			if err := waitDone(); err != nil {
  1156  				return nil, err
  1157  			}
  1158  		}
  1159  		return res, nil
  1160  	}
  1161  
  1162  	cancelRequest := func(cs *clientStream, err error) error {
  1163  		cs.cc.mu.Lock()
  1164  		bodyClosed := cs.reqBodyClosed
  1165  		cs.cc.mu.Unlock()
  1166  		// Wait for the request body to be closed.
  1167  		//
  1168  		// If nothing closed the body before now, abortStreamLocked
  1169  		// will have started a goroutine to close it.
  1170  		//
  1171  		// Closing the body before returning avoids a race condition
  1172  		// with net/http checking its readTrackingBody to see if the
  1173  		// body was read from or closed. See golang/go#60041.
  1174  		//
  1175  		// The body is closed in a separate goroutine without the
  1176  		// connection mutex held, but dropping the mutex before waiting
  1177  		// will keep us from holding it indefinitely if the body
  1178  		// close is slow for some reason.
  1179  		if bodyClosed != nil {
  1180  			<-bodyClosed
  1181  		}
  1182  		return err
  1183  	}
  1184  
  1185  	for {
  1186  		select {
  1187  		case <-cs.respHeaderRecv:
  1188  			return handleResponseHeaders()
  1189  		case <-cs.abort:
  1190  			select {
  1191  			case <-cs.respHeaderRecv:
  1192  				// If both cs.respHeaderRecv and cs.abort are signaling,
  1193  				// pick respHeaderRecv. The server probably wrote the
  1194  				// response and immediately reset the stream.
  1195  				// golang.org/issue/49645
  1196  				return handleResponseHeaders()
  1197  			default:
  1198  				waitDone()
  1199  				return nil, cs.abortErr
  1200  			}
  1201  		case <-ctx.Done():
  1202  			err := ctx.Err()
  1203  			cs.abortStream(err)
  1204  			return nil, cancelRequest(cs, err)
  1205  		case <-cs.reqCancel:
  1206  			cs.abortStream(errRequestCanceled)
  1207  			return nil, cancelRequest(cs, errRequestCanceled)
  1208  		}
  1209  	}
  1210  }
  1211  
  1212  // doRequest runs for the duration of the request lifetime.
  1213  //
  1214  // It sends the request and performs post-request cleanup (closing Request.Body, etc.).
  1215  func (cs *clientStream) doRequest(req *ClientRequest, streamf func(*clientStream)) {
  1216  	err := cs.writeRequest(req, streamf)
  1217  	cs.cleanupWriteRequest(err)
  1218  }
  1219  
  1220  var errExtendedConnectNotSupported = errors.New("net/http: extended connect not supported by peer")
  1221  
  1222  // writeRequest sends a request.
  1223  //
  1224  // It returns nil after the request is written, the response read,
  1225  // and the request stream is half-closed by the peer.
  1226  //
  1227  // It returns non-nil if the request ends otherwise.
  1228  // If the returned error is StreamError, the error Code may be used in resetting the stream.
  1229  func (cs *clientStream) writeRequest(req *ClientRequest, streamf func(*clientStream)) (err error) {
  1230  	cc := cs.cc
  1231  	ctx := cs.ctx
  1232  
  1233  	// wait for setting frames to be received, a server can change this value later,
  1234  	// but we just wait for the first settings frame
  1235  	var isExtendedConnect bool
  1236  	if req.Method == "CONNECT" && req.Header.Get(":protocol") != "" {
  1237  		isExtendedConnect = true
  1238  	}
  1239  
  1240  	// Acquire the new-request lock by writing to reqHeaderMu.
  1241  	// This lock guards the critical section covering allocating a new stream ID
  1242  	// (requires mu) and creating the stream (requires wmu).
  1243  	if cc.reqHeaderMu == nil {
  1244  		panic("RoundTrip on uninitialized ClientConn") // for tests
  1245  	}
  1246  	if isExtendedConnect {
  1247  		select {
  1248  		case <-cs.reqCancel:
  1249  			return errRequestCanceled
  1250  		case <-ctx.Done():
  1251  			return ctx.Err()
  1252  		case <-cc.seenSettingsChan:
  1253  			if !cc.extendedConnectAllowed {
  1254  				return errExtendedConnectNotSupported
  1255  			}
  1256  		}
  1257  	}
  1258  	select {
  1259  	case cc.reqHeaderMu <- struct{}{}:
  1260  	case <-cs.reqCancel:
  1261  		return errRequestCanceled
  1262  	case <-ctx.Done():
  1263  		return ctx.Err()
  1264  	}
  1265  
  1266  	cc.mu.Lock()
  1267  	if cc.idleTimer != nil {
  1268  		cc.idleTimer.Stop()
  1269  	}
  1270  	cc.decrStreamReservationsLocked()
  1271  	if err := cc.awaitOpenSlotForStreamLocked(cs); err != nil {
  1272  		cc.mu.Unlock()
  1273  		<-cc.reqHeaderMu
  1274  		return err
  1275  	}
  1276  	cc.addStreamLocked(cs) // assigns stream ID
  1277  	if isConnectionCloseRequest(req) {
  1278  		cc.doNotReuse = true
  1279  	}
  1280  	cc.mu.Unlock()
  1281  
  1282  	if streamf != nil {
  1283  		streamf(cs)
  1284  	}
  1285  
  1286  	continueTimeout := cc.t.expectContinueTimeout()
  1287  	if continueTimeout != 0 {
  1288  		if !httpguts.HeaderValuesContainsToken(req.Header["Expect"], "100-continue") {
  1289  			continueTimeout = 0
  1290  		} else {
  1291  			cs.on100 = make(chan struct{}, 1)
  1292  		}
  1293  	}
  1294  
  1295  	// Past this point (where we send request headers), it is possible for
  1296  	// RoundTrip to return successfully. Since the RoundTrip contract permits
  1297  	// the caller to "mutate or reuse" the Request after closing the Response's Body,
  1298  	// we must take care when referencing the Request from here on.
  1299  	err = cs.encodeAndWriteHeaders(req)
  1300  	<-cc.reqHeaderMu
  1301  	if err != nil {
  1302  		return err
  1303  	}
  1304  
  1305  	hasBody := cs.reqBodyContentLength != 0
  1306  	if !hasBody {
  1307  		cs.sentEndStream = true
  1308  	} else {
  1309  		if continueTimeout != 0 {
  1310  			traceWait100Continue(cs.trace)
  1311  			timer := time.NewTimer(continueTimeout)
  1312  			select {
  1313  			case <-timer.C:
  1314  				err = nil
  1315  			case <-cs.on100:
  1316  				err = nil
  1317  			case <-cs.abort:
  1318  				err = cs.abortErr
  1319  			case <-ctx.Done():
  1320  				err = ctx.Err()
  1321  			case <-cs.reqCancel:
  1322  				err = errRequestCanceled
  1323  			}
  1324  			timer.Stop()
  1325  			if err != nil {
  1326  				traceWroteRequest(cs.trace, err)
  1327  				return err
  1328  			}
  1329  		}
  1330  
  1331  		if err = cs.writeRequestBody(req); err != nil {
  1332  			if err != errStopReqBodyWrite {
  1333  				traceWroteRequest(cs.trace, err)
  1334  				return err
  1335  			}
  1336  		} else {
  1337  			cs.sentEndStream = true
  1338  		}
  1339  	}
  1340  
  1341  	traceWroteRequest(cs.trace, err)
  1342  
  1343  	var respHeaderTimer <-chan time.Time
  1344  	var respHeaderRecv chan struct{}
  1345  	if d := cc.responseHeaderTimeout(); d != 0 {
  1346  		timer := time.NewTimer(d)
  1347  		defer timer.Stop()
  1348  		respHeaderTimer = timer.C
  1349  		respHeaderRecv = cs.respHeaderRecv
  1350  	}
  1351  	// Wait until the peer half-closes its end of the stream,
  1352  	// or until the request is aborted (via context, error, or otherwise),
  1353  	// whichever comes first.
  1354  	for {
  1355  		select {
  1356  		case <-cs.peerClosed:
  1357  			return nil
  1358  		case <-respHeaderTimer:
  1359  			return errTimeout
  1360  		case <-respHeaderRecv:
  1361  			respHeaderRecv = nil
  1362  			respHeaderTimer = nil // keep waiting for END_STREAM
  1363  		case <-cs.abort:
  1364  			return cs.abortErr
  1365  		case <-ctx.Done():
  1366  			return ctx.Err()
  1367  		case <-cs.reqCancel:
  1368  			return errRequestCanceled
  1369  		}
  1370  	}
  1371  }
  1372  
  1373  func (cs *clientStream) encodeAndWriteHeaders(req *ClientRequest) error {
  1374  	cc := cs.cc
  1375  	ctx := cs.ctx
  1376  
  1377  	cc.wmu.Lock()
  1378  	defer cc.wmu.Unlock()
  1379  
  1380  	// If the request was canceled while waiting for cc.mu, just quit.
  1381  	select {
  1382  	case <-cs.abort:
  1383  		return cs.abortErr
  1384  	case <-ctx.Done():
  1385  		return ctx.Err()
  1386  	case <-cs.reqCancel:
  1387  		return errRequestCanceled
  1388  	default:
  1389  	}
  1390  
  1391  	// Encode headers.
  1392  	//
  1393  	// we send: HEADERS{1}, CONTINUATION{0,} + DATA{0,} (DATA is
  1394  	// sent by writeRequestBody below, along with any Trailers,
  1395  	// again in form HEADERS{1}, CONTINUATION{0,})
  1396  	cc.hbuf.Reset()
  1397  	res, err := encodeRequestHeaders(req, cs.requestedGzip, cc.peerMaxHeaderListSize, func(name, value string) {
  1398  		cc.writeHeader(name, value)
  1399  	})
  1400  	if err != nil {
  1401  		return fmt.Errorf("http2: %w", err)
  1402  	}
  1403  	hdrs := cc.hbuf.Bytes()
  1404  
  1405  	// Write the request.
  1406  	endStream := !res.HasBody && !res.HasTrailers
  1407  	cs.sentHeaders = true
  1408  	err = cc.writeHeaders(cs.ID, endStream, int(cc.maxFrameSize), hdrs)
  1409  	traceWroteHeaders(cs.trace)
  1410  	return err
  1411  }
  1412  
  1413  func encodeRequestHeaders(req *ClientRequest, addGzipHeader bool, peerMaxHeaderListSize uint64, headerf func(name, value string)) (httpcommon.EncodeHeadersResult, error) {
  1414  	return httpcommon.EncodeHeaders(req.Context, httpcommon.EncodeHeadersParam{
  1415  		Request: httpcommon.Request{
  1416  			Header:              req.Header,
  1417  			Trailer:             req.Trailer,
  1418  			URL:                 req.URL,
  1419  			Host:                req.Host,
  1420  			Method:              req.Method,
  1421  			ActualContentLength: actualContentLength(req),
  1422  		},
  1423  		AddGzipHeader:         addGzipHeader,
  1424  		PeerMaxHeaderListSize: peerMaxHeaderListSize,
  1425  		DefaultUserAgent:      defaultUserAgent,
  1426  	}, headerf)
  1427  }
  1428  
  1429  // cleanupWriteRequest performs post-request tasks.
  1430  //
  1431  // If err (the result of writeRequest) is non-nil and the stream is not closed,
  1432  // cleanupWriteRequest will send a reset to the peer.
  1433  func (cs *clientStream) cleanupWriteRequest(err error) {
  1434  	cc := cs.cc
  1435  
  1436  	if cs.ID == 0 {
  1437  		// We were canceled before creating the stream, so return our reservation.
  1438  		cc.decrStreamReservations()
  1439  	}
  1440  
  1441  	// TODO: write h12Compare test showing whether
  1442  	// Request.Body is closed by the Transport,
  1443  	// and in multiple cases: server replies <=299 and >299
  1444  	// while still writing request body
  1445  	cc.mu.Lock()
  1446  	mustCloseBody := false
  1447  	if cs.reqBody != nil && cs.reqBodyClosed == nil {
  1448  		mustCloseBody = true
  1449  		cs.reqBodyClosed = make(chan struct{})
  1450  	}
  1451  	bodyClosed := cs.reqBodyClosed
  1452  	closeOnIdle := cc.singleUse || cc.doNotReuse || cc.t.disableKeepAlives() || cc.goAway != nil
  1453  	// Have we read any frames from the connection since sending this request?
  1454  	readSinceStream := cc.readBeforeStreamID > cs.ID
  1455  	cc.mu.Unlock()
  1456  	if mustCloseBody {
  1457  		cs.reqBody.Close()
  1458  		close(bodyClosed)
  1459  	}
  1460  	if bodyClosed != nil {
  1461  		<-bodyClosed
  1462  	}
  1463  
  1464  	if err != nil && cs.sentEndStream {
  1465  		// If the connection is closed immediately after the response is read,
  1466  		// we may be aborted before finishing up here. If the stream was closed
  1467  		// cleanly on both sides, there is no error.
  1468  		select {
  1469  		case <-cs.peerClosed:
  1470  			err = nil
  1471  		default:
  1472  		}
  1473  	}
  1474  	if err != nil {
  1475  		cs.abortStream(err) // possibly redundant, but harmless
  1476  		if cs.sentHeaders {
  1477  			if se, ok := err.(StreamError); ok {
  1478  				if se.Cause != errFromPeer {
  1479  					cc.writeStreamReset(cs.ID, se.Code, false, err)
  1480  				}
  1481  			} else {
  1482  				// We're cancelling an in-flight request.
  1483  				//
  1484  				// This could be due to the server becoming unresponsive.
  1485  				// To avoid sending too many requests on a dead connection,
  1486  				// if we haven't read any frames from the connection since
  1487  				// sending this request, we let it continue to consume
  1488  				// a concurrency slot until we can confirm the server is
  1489  				// still responding.
  1490  				// We do this by sending a PING frame along with the RST_STREAM
  1491  				// (unless a ping is already in flight).
  1492  				//
  1493  				// For simplicity, we don't bother tracking the PING payload:
  1494  				// We reset cc.pendingResets any time we receive a PING ACK.
  1495  				//
  1496  				// We skip this if the conn is going to be closed on idle,
  1497  				// because it's short lived and will probably be closed before
  1498  				// we get the ping response.
  1499  				ping := false
  1500  				if !closeOnIdle && !readSinceStream {
  1501  					cc.mu.Lock()
  1502  					// rstStreamPingsBlocked works around a gRPC behavior:
  1503  					// see comment on the field for details.
  1504  					if !cc.rstStreamPingsBlocked {
  1505  						if cc.pendingResets == 0 {
  1506  							ping = true
  1507  						}
  1508  						cc.pendingResets++
  1509  					}
  1510  					cc.mu.Unlock()
  1511  				}
  1512  				cc.writeStreamReset(cs.ID, ErrCodeCancel, ping, err)
  1513  			}
  1514  		}
  1515  		cs.bufPipe.CloseWithError(err) // no-op if already closed
  1516  	} else {
  1517  		if cs.sentHeaders && !cs.sentEndStream {
  1518  			cc.writeStreamReset(cs.ID, ErrCodeNo, false, nil)
  1519  		}
  1520  		cs.bufPipe.CloseWithError(errRequestCanceled)
  1521  	}
  1522  	if cs.ID != 0 {
  1523  		cc.forgetStreamID(cs.ID)
  1524  	}
  1525  
  1526  	cc.wmu.Lock()
  1527  	werr := cc.werr
  1528  	cc.wmu.Unlock()
  1529  	if werr != nil {
  1530  		cc.Close()
  1531  	}
  1532  
  1533  	close(cs.donec)
  1534  	cc.maybeCallStateHook()
  1535  }
  1536  
  1537  // awaitOpenSlotForStreamLocked waits until len(streams) < maxConcurrentStreams.
  1538  // Must hold cc.mu.
  1539  func (cc *ClientConn) awaitOpenSlotForStreamLocked(cs *clientStream) error {
  1540  	for {
  1541  		if cc.closed && cc.nextStreamID == 1 && cc.streamsReserved == 0 {
  1542  			// This is the very first request sent to this connection.
  1543  			// Return a fatal error which aborts the retry loop.
  1544  			return errClientConnNotEstablished
  1545  		}
  1546  		cc.lastActive = time.Now()
  1547  		if cc.closed || !cc.canTakeNewRequestLocked() {
  1548  			return errClientConnUnusable
  1549  		}
  1550  		cc.lastIdle = time.Time{}
  1551  		if cc.currentRequestCountLocked() < int(cc.maxConcurrentStreams) {
  1552  			return nil
  1553  		}
  1554  		cc.pendingRequests++
  1555  		cc.cond.Wait()
  1556  		cc.pendingRequests--
  1557  		select {
  1558  		case <-cs.abort:
  1559  			return cs.abortErr
  1560  		default:
  1561  		}
  1562  	}
  1563  }
  1564  
  1565  // requires cc.wmu be held
  1566  func (cc *ClientConn) writeHeaders(streamID uint32, endStream bool, maxFrameSize int, hdrs []byte) error {
  1567  	first := true // first frame written (HEADERS is first, then CONTINUATION)
  1568  	for len(hdrs) > 0 && cc.werr == nil {
  1569  		chunk := hdrs
  1570  		if len(chunk) > maxFrameSize {
  1571  			chunk = chunk[:maxFrameSize]
  1572  		}
  1573  		hdrs = hdrs[len(chunk):]
  1574  		endHeaders := len(hdrs) == 0
  1575  		if first {
  1576  			cc.fr.WriteHeaders(HeadersFrameParam{
  1577  				StreamID:      streamID,
  1578  				BlockFragment: chunk,
  1579  				EndStream:     endStream,
  1580  				EndHeaders:    endHeaders,
  1581  			})
  1582  			first = false
  1583  		} else {
  1584  			cc.fr.WriteContinuation(streamID, endHeaders, chunk)
  1585  		}
  1586  	}
  1587  	cc.bw.Flush()
  1588  	return cc.werr
  1589  }
  1590  
  1591  // internal error values; they don't escape to callers
  1592  var (
  1593  	// abort request body write; don't send cancel
  1594  	errStopReqBodyWrite = errors.New("http2: aborting request body write")
  1595  
  1596  	// abort request body write, but send stream reset of cancel.
  1597  	errStopReqBodyWriteAndCancel = errors.New("http2: canceling request")
  1598  
  1599  	errReqBodyTooLong = errors.New("http2: request body larger than specified content length")
  1600  )
  1601  
  1602  // frameScratchBufferLen returns the length of a buffer to use for
  1603  // outgoing request bodies to read/write to/from.
  1604  //
  1605  // It returns max(1, min(peer's advertised max frame size,
  1606  // Request.ContentLength+1, 512KB)).
  1607  func (cs *clientStream) frameScratchBufferLen(maxFrameSize int) int {
  1608  	const max = 512 << 10
  1609  	n := min(int64(maxFrameSize), max)
  1610  	if cl := cs.reqBodyContentLength; cl != -1 && cl+1 < n {
  1611  		// Add an extra byte past the declared content-length to
  1612  		// give the caller's Request.Body io.Reader a chance to
  1613  		// give us more bytes than they declared, so we can catch it
  1614  		// early.
  1615  		n = cl + 1
  1616  	}
  1617  	if n < 1 {
  1618  		return 1
  1619  	}
  1620  	return int(n) // doesn't truncate; max is 512K
  1621  }
  1622  
  1623  // Seven bufPools manage different frame sizes. This helps to avoid scenarios where long-running
  1624  // streaming requests using small frame sizes occupy large buffers initially allocated for prior
  1625  // requests needing big buffers. The size ranges are as follows:
  1626  // {0 KB, 16 KB], {16 KB, 32 KB], {32 KB, 64 KB], {64 KB, 128 KB], {128 KB, 256 KB],
  1627  // {256 KB, 512 KB], {512 KB, infinity}
  1628  // In practice, the maximum scratch buffer size should not exceed 512 KB due to
  1629  // frameScratchBufferLen(maxFrameSize), thus the "infinity pool" should never be used.
  1630  // It exists mainly as a safety measure, for potential future increases in max buffer size.
  1631  var bufPools [7]sync.Pool // of *[]byte
  1632  func bufPoolIndex(size int) int {
  1633  	if size <= 16384 {
  1634  		return 0
  1635  	}
  1636  	size -= 1
  1637  	bits := bits.Len(uint(size))
  1638  	index := bits - 14
  1639  	if index >= len(bufPools) {
  1640  		return len(bufPools) - 1
  1641  	}
  1642  	return index
  1643  }
  1644  
  1645  func (cs *clientStream) writeRequestBody(req *ClientRequest) (err error) {
  1646  	cc := cs.cc
  1647  	body := cs.reqBody
  1648  	sentEnd := false // whether we sent the final DATA frame w/ END_STREAM
  1649  
  1650  	hasTrailers := req.Trailer != nil
  1651  	remainLen := cs.reqBodyContentLength
  1652  	hasContentLen := remainLen != -1
  1653  
  1654  	cc.mu.Lock()
  1655  	maxFrameSize := int(cc.maxFrameSize)
  1656  	cc.mu.Unlock()
  1657  
  1658  	// Scratch buffer for reading into & writing from.
  1659  	scratchLen := cs.frameScratchBufferLen(maxFrameSize)
  1660  	var buf []byte
  1661  	index := bufPoolIndex(scratchLen)
  1662  	if bp, ok := bufPools[index].Get().(*[]byte); ok && len(*bp) >= scratchLen {
  1663  		defer bufPools[index].Put(bp)
  1664  		buf = *bp
  1665  	} else {
  1666  		buf = make([]byte, scratchLen)
  1667  		defer bufPools[index].Put(&buf)
  1668  	}
  1669  
  1670  	var sawEOF bool
  1671  	for !sawEOF {
  1672  		n, err := body.Read(buf)
  1673  		if hasContentLen {
  1674  			remainLen -= int64(n)
  1675  			if remainLen == 0 && err == nil {
  1676  				// The request body's Content-Length was predeclared and
  1677  				// we just finished reading it all, but the underlying io.Reader
  1678  				// returned the final chunk with a nil error (which is one of
  1679  				// the two valid things a Reader can do at EOF). Because we'd prefer
  1680  				// to send the END_STREAM bit early, double-check that we're actually
  1681  				// at EOF. Subsequent reads should return (0, EOF) at this point.
  1682  				// If either value is different, we return an error in one of two ways below.
  1683  				var scratch [1]byte
  1684  				var n1 int
  1685  				n1, err = body.Read(scratch[:])
  1686  				remainLen -= int64(n1)
  1687  			}
  1688  			if remainLen < 0 {
  1689  				err = errReqBodyTooLong
  1690  				return err
  1691  			}
  1692  		}
  1693  		if err != nil {
  1694  			cc.mu.Lock()
  1695  			bodyClosed := cs.reqBodyClosed != nil
  1696  			cc.mu.Unlock()
  1697  			switch {
  1698  			case bodyClosed:
  1699  				return errStopReqBodyWrite
  1700  			case err == io.EOF:
  1701  				sawEOF = true
  1702  				err = nil
  1703  			default:
  1704  				return err
  1705  			}
  1706  		}
  1707  
  1708  		remain := buf[:n]
  1709  		for len(remain) > 0 && err == nil {
  1710  			var allowed int32
  1711  			allowed, err = cs.awaitFlowControl(len(remain))
  1712  			if err != nil {
  1713  				return err
  1714  			}
  1715  			cc.wmu.Lock()
  1716  			data := remain[:allowed]
  1717  			remain = remain[allowed:]
  1718  			sentEnd = sawEOF && len(remain) == 0 && !hasTrailers
  1719  			err = cc.fr.WriteData(cs.ID, sentEnd, data)
  1720  			if err == nil {
  1721  				// TODO(bradfitz): this flush is for latency, not bandwidth.
  1722  				// Most requests won't need this. Make this opt-in or
  1723  				// opt-out?  Use some heuristic on the body type? Nagel-like
  1724  				// timers?  Based on 'n'? Only last chunk of this for loop,
  1725  				// unless flow control tokens are low? For now, always.
  1726  				// If we change this, see comment below.
  1727  				err = cc.bw.Flush()
  1728  			}
  1729  			cc.wmu.Unlock()
  1730  		}
  1731  		if err != nil {
  1732  			return err
  1733  		}
  1734  	}
  1735  
  1736  	if sentEnd {
  1737  		// Already sent END_STREAM (which implies we have no
  1738  		// trailers) and flushed, because currently all
  1739  		// WriteData frames above get a flush. So we're done.
  1740  		return nil
  1741  	}
  1742  
  1743  	// Since the RoundTrip contract permits the caller to "mutate or reuse"
  1744  	// a request after the Response's Body is closed, verify that this hasn't
  1745  	// happened before accessing the trailers.
  1746  	cc.mu.Lock()
  1747  	trailer := req.Trailer
  1748  	err = cs.abortErr
  1749  	cc.mu.Unlock()
  1750  	if err != nil {
  1751  		return err
  1752  	}
  1753  
  1754  	cc.wmu.Lock()
  1755  	defer cc.wmu.Unlock()
  1756  	var trls []byte
  1757  	if len(trailer) > 0 {
  1758  		trls, err = cc.encodeTrailers(trailer)
  1759  		if err != nil {
  1760  			return err
  1761  		}
  1762  	}
  1763  
  1764  	// Two ways to send END_STREAM: either with trailers, or
  1765  	// with an empty DATA frame.
  1766  	if len(trls) > 0 {
  1767  		err = cc.writeHeaders(cs.ID, true, maxFrameSize, trls)
  1768  	} else {
  1769  		err = cc.fr.WriteData(cs.ID, true, nil)
  1770  	}
  1771  	if ferr := cc.bw.Flush(); ferr != nil && err == nil {
  1772  		err = ferr
  1773  	}
  1774  	return err
  1775  }
  1776  
  1777  // awaitFlowControl waits for [1, min(maxBytes, cc.cs.maxFrameSize)] flow
  1778  // control tokens from the server.
  1779  // It returns either the non-zero number of tokens taken or an error
  1780  // if the stream is dead.
  1781  func (cs *clientStream) awaitFlowControl(maxBytes int) (taken int32, err error) {
  1782  	cc := cs.cc
  1783  	ctx := cs.ctx
  1784  	cc.mu.Lock()
  1785  	defer cc.mu.Unlock()
  1786  	for {
  1787  		if cc.closed {
  1788  			return 0, errClientConnClosed
  1789  		}
  1790  		if cs.reqBodyClosed != nil {
  1791  			return 0, errStopReqBodyWrite
  1792  		}
  1793  		select {
  1794  		case <-cs.abort:
  1795  			return 0, cs.abortErr
  1796  		case <-ctx.Done():
  1797  			return 0, ctx.Err()
  1798  		case <-cs.reqCancel:
  1799  			return 0, errRequestCanceled
  1800  		default:
  1801  		}
  1802  		if a := cs.flow.available(); a > 0 {
  1803  			take := a
  1804  			if int(take) > maxBytes {
  1805  
  1806  				take = int32(maxBytes) // can't truncate int; take is int32
  1807  			}
  1808  			if take > int32(cc.maxFrameSize) {
  1809  				take = int32(cc.maxFrameSize)
  1810  			}
  1811  			cs.flow.take(take)
  1812  			return take, nil
  1813  		}
  1814  		cc.cond.Wait()
  1815  	}
  1816  }
  1817  
  1818  // requires cc.wmu be held.
  1819  func (cc *ClientConn) encodeTrailers(trailer Header) ([]byte, error) {
  1820  	cc.hbuf.Reset()
  1821  
  1822  	hlSize := uint64(0)
  1823  	for k, vv := range trailer {
  1824  		for _, v := range vv {
  1825  			hf := hpack.HeaderField{Name: k, Value: v}
  1826  			hlSize += uint64(hf.Size())
  1827  		}
  1828  	}
  1829  	if hlSize > cc.peerMaxHeaderListSize {
  1830  		return nil, errRequestHeaderListSize
  1831  	}
  1832  
  1833  	for k, vv := range trailer {
  1834  		lowKey, ascii := httpcommon.LowerHeader(k)
  1835  		if !ascii {
  1836  			// Skip writing invalid headers. Per RFC 7540, Section 8.1.2, header
  1837  			// field names have to be ASCII characters (just as in HTTP/1.x).
  1838  			continue
  1839  		}
  1840  		// Transfer-Encoding, etc.. have already been filtered at the
  1841  		// start of RoundTrip
  1842  		for _, v := range vv {
  1843  			cc.writeHeader(lowKey, v)
  1844  		}
  1845  	}
  1846  	return cc.hbuf.Bytes(), nil
  1847  }
  1848  
  1849  func (cc *ClientConn) writeHeader(name, value string) {
  1850  	if VerboseLogs {
  1851  		log.Printf("http2: Transport encoding header %q = %q", name, value)
  1852  	}
  1853  	cc.henc.WriteField(hpack.HeaderField{Name: name, Value: value})
  1854  }
  1855  
  1856  type resAndError struct {
  1857  	_   incomparable
  1858  	res *ClientResponse
  1859  	err error
  1860  }
  1861  
  1862  // requires cc.mu be held.
  1863  func (cc *ClientConn) addStreamLocked(cs *clientStream) {
  1864  	cs.flow.add(int32(cc.initialWindowSize))
  1865  	cs.flow.setConnFlow(&cc.flow)
  1866  	cs.inflow.init(cc.initialStreamRecvWindowSize)
  1867  	cs.ID = cc.nextStreamID
  1868  	cc.nextStreamID += 2
  1869  	cc.streams[cs.ID] = cs
  1870  	if cs.ID == 0 {
  1871  		panic("assigned stream ID 0")
  1872  	}
  1873  }
  1874  
  1875  func (cc *ClientConn) forgetStreamID(id uint32) {
  1876  	cc.mu.Lock()
  1877  	slen := len(cc.streams)
  1878  	delete(cc.streams, id)
  1879  	if len(cc.streams) != slen-1 {
  1880  		panic("forgetting unknown stream id")
  1881  	}
  1882  	cc.lastActive = time.Now()
  1883  	if len(cc.streams) == 0 && cc.idleTimer != nil {
  1884  		cc.idleTimer.Reset(cc.idleTimeout)
  1885  		cc.lastIdle = time.Now()
  1886  	}
  1887  	// Wake up writeRequestBody via clientStream.awaitFlowControl and
  1888  	// wake up RoundTrip if there is a pending request.
  1889  	cc.cond.Broadcast()
  1890  
  1891  	closeOnIdle := cc.singleUse || cc.doNotReuse || cc.t.disableKeepAlives() || cc.goAway != nil
  1892  	if closeOnIdle && cc.streamsReserved == 0 && len(cc.streams) == 0 {
  1893  		if VerboseLogs {
  1894  			cc.vlogf("http2: Transport closing idle conn %p (forSingleUse=%v, maxStream=%v)", cc, cc.singleUse, cc.nextStreamID-2)
  1895  		}
  1896  		cc.closed = true
  1897  		defer cc.closeConn()
  1898  	}
  1899  
  1900  	cc.mu.Unlock()
  1901  }
  1902  
  1903  // clientConnReadLoop is the state owned by the clientConn's frame-reading readLoop.
  1904  type clientConnReadLoop struct {
  1905  	_  incomparable
  1906  	cc *ClientConn
  1907  }
  1908  
  1909  // readLoop runs in its own goroutine and reads and dispatches frames.
  1910  func (cc *ClientConn) readLoop() {
  1911  	rl := &clientConnReadLoop{cc: cc}
  1912  	defer rl.cleanup()
  1913  	cc.readerErr = rl.run()
  1914  	if ce, ok := cc.readerErr.(ConnectionError); ok {
  1915  		cc.wmu.Lock()
  1916  		cc.fr.WriteGoAway(0, ErrCode(ce), nil)
  1917  		cc.wmu.Unlock()
  1918  	}
  1919  }
  1920  
  1921  // GoAwayError is returned by the Transport when the server closes the
  1922  // TCP connection after sending a GOAWAY frame.
  1923  type GoAwayError struct {
  1924  	LastStreamID uint32
  1925  	ErrCode      ErrCode
  1926  	DebugData    string
  1927  }
  1928  
  1929  func (e GoAwayError) Error() string {
  1930  	return fmt.Sprintf("http2: server sent GOAWAY and closed the connection; LastStreamID=%v, ErrCode=%v, debug=%q",
  1931  		e.LastStreamID, e.ErrCode, e.DebugData)
  1932  }
  1933  
  1934  func isEOFOrNetReadError(err error) bool {
  1935  	if err == io.EOF {
  1936  		return true
  1937  	}
  1938  	ne, ok := err.(*net.OpError)
  1939  	return ok && ne.Op == "read"
  1940  }
  1941  
  1942  func (rl *clientConnReadLoop) cleanup() {
  1943  	cc := rl.cc
  1944  	defer cc.closeConn()
  1945  	defer close(cc.readerDone)
  1946  
  1947  	if cc.idleTimer != nil {
  1948  		cc.idleTimer.Stop()
  1949  	}
  1950  
  1951  	// Close any response bodies if the server closes prematurely.
  1952  	// TODO: also do this if we've written the headers but not
  1953  	// gotten a response yet.
  1954  	err := cc.readerErr
  1955  	cc.mu.Lock()
  1956  	if cc.goAway != nil && isEOFOrNetReadError(err) {
  1957  		err = GoAwayError{
  1958  			LastStreamID: cc.goAway.LastStreamID,
  1959  			ErrCode:      cc.goAway.ErrCode,
  1960  			DebugData:    cc.goAwayDebug,
  1961  		}
  1962  	} else if err == io.EOF {
  1963  		err = io.ErrUnexpectedEOF
  1964  	}
  1965  	cc.closed = true
  1966  
  1967  	// If the connection has never been used, and has been open for only a short time,
  1968  	// leave it in the connection pool for a little while.
  1969  	//
  1970  	// This avoids a situation where new connections are constantly created,
  1971  	// added to the pool, fail, and are removed from the pool, without any error
  1972  	// being surfaced to the user.
  1973  	unusedWaitTime := 5 * time.Second
  1974  	if cc.idleTimeout > 0 && unusedWaitTime > cc.idleTimeout {
  1975  		unusedWaitTime = cc.idleTimeout
  1976  	}
  1977  	idleTime := time.Now().Sub(cc.lastActive)
  1978  	if atomic.LoadUint32(&cc.atomicReused) == 0 && idleTime < unusedWaitTime && !cc.closedOnIdle {
  1979  		cc.idleTimer = time.AfterFunc(unusedWaitTime-idleTime, func() {
  1980  			cc.t.connPool.MarkDead(cc)
  1981  		})
  1982  	} else {
  1983  		cc.mu.Unlock() // avoid any deadlocks in MarkDead
  1984  		cc.t.connPool.MarkDead(cc)
  1985  		cc.mu.Lock()
  1986  	}
  1987  
  1988  	for _, cs := range cc.streams {
  1989  		select {
  1990  		case <-cs.peerClosed:
  1991  			// The server closed the stream before closing the conn,
  1992  			// so no need to interrupt it.
  1993  		default:
  1994  			cs.abortStreamLocked(err)
  1995  		}
  1996  	}
  1997  	cc.cond.Broadcast()
  1998  	cc.mu.Unlock()
  1999  
  2000  	if !cc.seenSettings {
  2001  		// If we have a pending request that wants extended CONNECT,
  2002  		// let it continue and fail with the connection error.
  2003  		cc.extendedConnectAllowed = true
  2004  		close(cc.seenSettingsChan)
  2005  	}
  2006  }
  2007  
  2008  // countReadFrameError calls ClientConn.fr.countError with a string
  2009  // representing err.
  2010  func (cc *ClientConn) countReadFrameError(err error) {
  2011  	f := cc.fr.countError
  2012  	if f == nil || err == nil {
  2013  		return
  2014  	}
  2015  	if ce, ok := err.(ConnectionError); ok {
  2016  		errCode := ErrCode(ce)
  2017  		f(fmt.Sprintf("read_frame_conn_error_%s", errCode.stringToken()))
  2018  		return
  2019  	}
  2020  	if errors.Is(err, io.EOF) {
  2021  		f("read_frame_eof")
  2022  		return
  2023  	}
  2024  	if errors.Is(err, io.ErrUnexpectedEOF) {
  2025  		f("read_frame_unexpected_eof")
  2026  		return
  2027  	}
  2028  	if errors.Is(err, ErrFrameTooLarge) {
  2029  		f("read_frame_too_large")
  2030  		return
  2031  	}
  2032  	f("read_frame_other")
  2033  }
  2034  
  2035  // errStopReadLoop is an error which cleanly exits the client's read loop.
  2036  var errStopReadLoop = errors.New("client connection is closing (BUG: this is not user visible)")
  2037  
  2038  func (rl *clientConnReadLoop) run() error {
  2039  	cc := rl.cc
  2040  	gotSettings := false
  2041  	readIdleTimeout := cc.readIdleTimeout
  2042  	var t *time.Timer
  2043  	if readIdleTimeout != 0 {
  2044  		t = time.AfterFunc(readIdleTimeout, cc.healthCheck)
  2045  	}
  2046  	for {
  2047  		f, err := cc.fr.ReadFrame()
  2048  		if t != nil {
  2049  			t.Reset(readIdleTimeout)
  2050  		}
  2051  		if err != nil {
  2052  			cc.vlogf("http2: Transport readFrame error on conn %p: (%T) %v", cc, err, err)
  2053  		}
  2054  		if se, ok := err.(StreamError); ok {
  2055  			if cs := rl.streamByID(se.StreamID, notHeaderOrDataFrame); cs != nil {
  2056  				if se.Cause == nil {
  2057  					se.Cause = cc.fr.errDetail
  2058  				}
  2059  				rl.endStreamError(cs, se)
  2060  			}
  2061  			continue
  2062  		} else if err != nil {
  2063  			cc.countReadFrameError(err)
  2064  			return err
  2065  		}
  2066  		if VerboseLogs {
  2067  			cc.vlogf("http2: Transport received %s", summarizeFrame(f))
  2068  		}
  2069  		if !gotSettings {
  2070  			if _, ok := f.(*SettingsFrame); !ok {
  2071  				cc.logf("protocol error: received %T before a SETTINGS frame", f)
  2072  				return ConnectionError(ErrCodeProtocol)
  2073  			}
  2074  			gotSettings = true
  2075  		}
  2076  
  2077  		switch f := f.(type) {
  2078  		case *MetaHeadersFrame:
  2079  			err = rl.processHeaders(f)
  2080  		case *DataFrame:
  2081  			err = rl.processData(f)
  2082  		case *GoAwayFrame:
  2083  			err = rl.processGoAway(f)
  2084  		case *RSTStreamFrame:
  2085  			err = rl.processResetStream(f)
  2086  		case *SettingsFrame:
  2087  			err = rl.processSettings(f)
  2088  		case *PushPromiseFrame:
  2089  			err = rl.processPushPromise(f)
  2090  		case *WindowUpdateFrame:
  2091  			err = rl.processWindowUpdate(f)
  2092  		case *PingFrame:
  2093  			err = rl.processPing(f)
  2094  		default:
  2095  			cc.logf("Transport: unhandled response frame type %T", f)
  2096  		}
  2097  		if err != nil {
  2098  			if VerboseLogs && err != errStopReadLoop {
  2099  				cc.vlogf("http2: Transport conn %p received error from processing frame %v: %v", cc, summarizeFrame(f), err)
  2100  			}
  2101  			return err
  2102  		}
  2103  	}
  2104  }
  2105  
  2106  func (rl *clientConnReadLoop) processHeaders(f *MetaHeadersFrame) error {
  2107  	cs := rl.streamByID(f.StreamID, headerOrDataFrame)
  2108  	if cs == nil {
  2109  		// We'd get here if we canceled a request while the
  2110  		// server had its response still in flight. So if this
  2111  		// was just something we canceled, ignore it.
  2112  		return nil
  2113  	}
  2114  	if cs.readClosed {
  2115  		rl.endStreamError(cs, StreamError{
  2116  			StreamID: f.StreamID,
  2117  			Code:     ErrCodeProtocol,
  2118  			Cause:    errors.New("protocol error: headers after END_STREAM"),
  2119  		})
  2120  		return nil
  2121  	}
  2122  	if !cs.firstByte {
  2123  		if cs.trace != nil {
  2124  			// TODO(bradfitz): move first response byte earlier,
  2125  			// when we first read the 9 byte header, not waiting
  2126  			// until all the HEADERS+CONTINUATION frames have been
  2127  			// merged. This works for now.
  2128  			traceFirstResponseByte(cs.trace)
  2129  		}
  2130  		cs.firstByte = true
  2131  	}
  2132  	if !cs.pastHeaders {
  2133  		cs.pastHeaders = true
  2134  	} else {
  2135  		return rl.processTrailers(cs, f)
  2136  	}
  2137  
  2138  	res, err := rl.handleResponse(cs, f)
  2139  	if err != nil {
  2140  		if _, ok := err.(ConnectionError); ok {
  2141  			return err
  2142  		}
  2143  		// Any other error type is a stream error.
  2144  		rl.endStreamError(cs, StreamError{
  2145  			StreamID: f.StreamID,
  2146  			Code:     ErrCodeProtocol,
  2147  			Cause:    err,
  2148  		})
  2149  		return nil // return nil from process* funcs to keep conn alive
  2150  	}
  2151  	if res == nil {
  2152  		// (nil, nil) special case. See handleResponse docs.
  2153  		return nil
  2154  	}
  2155  	cs.res = res
  2156  	close(cs.respHeaderRecv)
  2157  	if f.StreamEnded() {
  2158  		rl.endStream(cs)
  2159  	}
  2160  	return nil
  2161  }
  2162  
  2163  // may return error types nil, or ConnectionError. Any other error value
  2164  // is a StreamError of type ErrCodeProtocol. The returned error in that case
  2165  // is the detail.
  2166  //
  2167  // As a special case, handleResponse may return (nil, nil) to skip the
  2168  // frame (currently only used for 1xx responses).
  2169  func (rl *clientConnReadLoop) handleResponse(cs *clientStream, f *MetaHeadersFrame) (*ClientResponse, error) {
  2170  	if f.Truncated {
  2171  		return nil, errResponseHeaderListSize
  2172  	}
  2173  
  2174  	status := f.PseudoValue("status")
  2175  	if status == "" {
  2176  		return nil, errors.New("malformed response from server: missing status pseudo header")
  2177  	}
  2178  	statusCode, err := strconv.Atoi(status)
  2179  	if err != nil {
  2180  		return nil, errors.New("malformed response from server: malformed non-numeric status pseudo header")
  2181  	}
  2182  
  2183  	regularFields := f.RegularFields()
  2184  	strs := make([]string, len(regularFields))
  2185  	header := make(Header, len(regularFields))
  2186  	res := &cs.staticResp
  2187  	cs.staticResp = ClientResponse{
  2188  		Header:     header,
  2189  		StatusCode: statusCode,
  2190  		Status:     status,
  2191  	}
  2192  	for _, hf := range regularFields {
  2193  		key := httpcommon.CanonicalHeader(hf.Name)
  2194  		if key == "Trailer" {
  2195  			t := res.Trailer
  2196  			if t == nil {
  2197  				t = make(Header)
  2198  				res.Trailer = t
  2199  			}
  2200  			foreachHeaderElement(hf.Value, func(v string) {
  2201  				t[httpcommon.CanonicalHeader(v)] = nil
  2202  			})
  2203  		} else {
  2204  			vv := header[key]
  2205  			if vv == nil && len(strs) > 0 {
  2206  				// More than likely this will be a single-element key.
  2207  				// Most headers aren't multi-valued.
  2208  				// Set the capacity on strs[0] to 1, so any future append
  2209  				// won't extend the slice into the other strings.
  2210  				vv, strs = strs[:1:1], strs[1:]
  2211  				vv[0] = hf.Value
  2212  				header[key] = vv
  2213  			} else {
  2214  				header[key] = append(vv, hf.Value)
  2215  			}
  2216  		}
  2217  	}
  2218  
  2219  	if statusCode >= 100 && statusCode <= 199 {
  2220  		if f.StreamEnded() {
  2221  			return nil, errors.New("1xx informational response with END_STREAM flag")
  2222  		}
  2223  		if fn := cs.get1xxTraceFunc(); fn != nil {
  2224  			// If the 1xx response is being delivered to the user,
  2225  			// then they're responsible for limiting the number
  2226  			// of responses.
  2227  			if err := fn(statusCode, textproto.MIMEHeader(header)); err != nil {
  2228  				return nil, err
  2229  			}
  2230  		} else {
  2231  			// If the user didn't examine the 1xx response, then we
  2232  			// limit the size of all 1xx headers.
  2233  			//
  2234  			// This differs a bit from the HTTP/1 implementation, which
  2235  			// limits the size of all 1xx headers plus the final response.
  2236  			// Use the larger limit of MaxHeaderListSize and
  2237  			// net/http.Transport.MaxResponseHeaderBytes.
  2238  			limit := int64(cs.cc.t.maxHeaderListSize())
  2239  			if t1 := cs.cc.t.t1; t1 != nil && t1.MaxResponseHeaderBytes() > limit {
  2240  				limit = t1.MaxResponseHeaderBytes()
  2241  			}
  2242  			for _, h := range f.Fields {
  2243  				cs.totalHeaderSize += int64(h.Size())
  2244  			}
  2245  			if cs.totalHeaderSize > limit {
  2246  				if VerboseLogs {
  2247  					log.Printf("http2: 1xx informational responses too large")
  2248  				}
  2249  				return nil, errors.New("header list too large")
  2250  			}
  2251  		}
  2252  		if statusCode == 100 {
  2253  			traceGot100Continue(cs.trace)
  2254  			select {
  2255  			case cs.on100 <- struct{}{}:
  2256  			default:
  2257  			}
  2258  		}
  2259  		cs.pastHeaders = false // do it all again
  2260  		return nil, nil
  2261  	}
  2262  
  2263  	res.ContentLength = -1
  2264  	if clens := res.Header["Content-Length"]; len(clens) == 1 {
  2265  		if cl, err := strconv.ParseUint(clens[0], 10, 63); err == nil {
  2266  			res.ContentLength = int64(cl)
  2267  		} else {
  2268  			// TODO: care? unlike http/1, it won't mess up our framing, so it's
  2269  			// more safe smuggling-wise to ignore.
  2270  		}
  2271  	} else if len(clens) > 1 {
  2272  		// TODO: care? unlike http/1, it won't mess up our framing, so it's
  2273  		// more safe smuggling-wise to ignore.
  2274  	} else if f.StreamEnded() && !cs.isHead {
  2275  		res.ContentLength = 0
  2276  	}
  2277  
  2278  	if cs.isHead {
  2279  		res.Body = NoBody
  2280  		return res, nil
  2281  	}
  2282  
  2283  	if f.StreamEnded() {
  2284  		if res.ContentLength > 0 {
  2285  			res.Body = missingBody{}
  2286  		} else {
  2287  			res.Body = NoBody
  2288  		}
  2289  		return res, nil
  2290  	}
  2291  
  2292  	cs.bufPipe.setBuffer(&dataBuffer{expected: res.ContentLength})
  2293  	cs.bytesRemain = res.ContentLength
  2294  	res.Body = transportResponseBody{cs}
  2295  
  2296  	if cs.requestedGzip && asciiEqualFold(res.Header.Get("Content-Encoding"), "gzip") {
  2297  		res.Header.Del("Content-Encoding")
  2298  		res.Header.Del("Content-Length")
  2299  		res.ContentLength = -1
  2300  		res.Body = &gzipReader{body: res.Body}
  2301  		res.Uncompressed = true
  2302  	}
  2303  	return res, nil
  2304  }
  2305  
  2306  func (rl *clientConnReadLoop) processTrailers(cs *clientStream, f *MetaHeadersFrame) error {
  2307  	if cs.pastTrailers {
  2308  		// Too many HEADERS frames for this stream.
  2309  		return ConnectionError(ErrCodeProtocol)
  2310  	}
  2311  	cs.pastTrailers = true
  2312  	if !f.StreamEnded() {
  2313  		// We expect that any headers for trailers also
  2314  		// has END_STREAM.
  2315  		return ConnectionError(ErrCodeProtocol)
  2316  	}
  2317  	if len(f.PseudoFields()) > 0 {
  2318  		// No pseudo header fields are defined for trailers.
  2319  		// TODO: ConnectionError might be overly harsh? Check.
  2320  		return ConnectionError(ErrCodeProtocol)
  2321  	}
  2322  	if f.Truncated {
  2323  		rl.endStreamError(cs, StreamError{
  2324  			StreamID: f.StreamID,
  2325  			Code:     ErrCodeProtocol,
  2326  			Cause:    errResponseHeaderListSize,
  2327  		})
  2328  		return nil
  2329  	}
  2330  
  2331  	trailer := make(Header)
  2332  	for _, hf := range f.RegularFields() {
  2333  		key := httpcommon.CanonicalHeader(hf.Name)
  2334  		trailer[key] = append(trailer[key], hf.Value)
  2335  	}
  2336  	cs.trailer = trailer
  2337  
  2338  	rl.endStream(cs)
  2339  	return nil
  2340  }
  2341  
  2342  // transportResponseBody is the concrete type of Transport.RoundTrip's
  2343  // Response.Body. It is an io.ReadCloser.
  2344  type transportResponseBody struct {
  2345  	cs *clientStream
  2346  }
  2347  
  2348  func (b transportResponseBody) Read(p []byte) (n int, err error) {
  2349  	cs := b.cs
  2350  	cc := cs.cc
  2351  
  2352  	if cs.readErr != nil {
  2353  		return 0, cs.readErr
  2354  	}
  2355  	n, err = b.cs.bufPipe.Read(p)
  2356  	if cs.bytesRemain != -1 {
  2357  		if int64(n) > cs.bytesRemain {
  2358  			n = int(cs.bytesRemain)
  2359  			if err == nil {
  2360  				err = errors.New("net/http: server replied with more than declared Content-Length; truncated")
  2361  				cs.abortStream(err)
  2362  			}
  2363  			cs.readErr = err
  2364  			return int(cs.bytesRemain), err
  2365  		}
  2366  		cs.bytesRemain -= int64(n)
  2367  		if err == io.EOF && cs.bytesRemain > 0 {
  2368  			err = io.ErrUnexpectedEOF
  2369  			cs.readErr = err
  2370  			return n, err
  2371  		}
  2372  	}
  2373  	if n == 0 {
  2374  		// No flow control tokens to send back.
  2375  		return
  2376  	}
  2377  
  2378  	cc.mu.Lock()
  2379  	connAdd := cc.inflow.add(n)
  2380  	var streamAdd int32
  2381  	if err == nil { // No need to refresh if the stream is over or failed.
  2382  		streamAdd = cs.inflow.add(n)
  2383  	}
  2384  	cc.mu.Unlock()
  2385  
  2386  	if connAdd != 0 || streamAdd != 0 {
  2387  		cc.wmu.Lock()
  2388  		defer cc.wmu.Unlock()
  2389  		if connAdd != 0 {
  2390  			cc.fr.WriteWindowUpdate(0, mustUint31(connAdd))
  2391  		}
  2392  		if streamAdd != 0 {
  2393  			cc.fr.WriteWindowUpdate(cs.ID, mustUint31(streamAdd))
  2394  		}
  2395  		cc.bw.Flush()
  2396  	}
  2397  	return
  2398  }
  2399  
  2400  var errClosedResponseBody = errors.New("http2: response body closed")
  2401  
  2402  func (b transportResponseBody) Close() error {
  2403  	cs := b.cs
  2404  	cc := cs.cc
  2405  
  2406  	cs.bufPipe.BreakWithError(errClosedResponseBody)
  2407  	cs.abortStream(errClosedResponseBody)
  2408  
  2409  	unread := cs.bufPipe.Len()
  2410  	if unread > 0 {
  2411  		cc.mu.Lock()
  2412  		// Return connection-level flow control.
  2413  		connAdd := cc.inflow.add(unread)
  2414  		cc.mu.Unlock()
  2415  
  2416  		// TODO(dneil): Acquiring this mutex can block indefinitely.
  2417  		// Move flow control return to a goroutine?
  2418  		cc.wmu.Lock()
  2419  		// Return connection-level flow control.
  2420  		if connAdd > 0 {
  2421  			cc.fr.WriteWindowUpdate(0, uint32(connAdd))
  2422  		}
  2423  		cc.bw.Flush()
  2424  		cc.wmu.Unlock()
  2425  	}
  2426  
  2427  	select {
  2428  	case <-cs.donec:
  2429  	case <-cs.ctx.Done():
  2430  		// See golang/go#49366: The net/http package can cancel the
  2431  		// request context after the response body is fully read.
  2432  		// Don't treat this as an error.
  2433  		return nil
  2434  	case <-cs.reqCancel:
  2435  		return errRequestCanceled
  2436  	}
  2437  	return nil
  2438  }
  2439  
  2440  func (rl *clientConnReadLoop) processData(f *DataFrame) error {
  2441  	cc := rl.cc
  2442  	cs := rl.streamByID(f.StreamID, headerOrDataFrame)
  2443  	data := f.Data()
  2444  	if cs == nil {
  2445  		cc.mu.Lock()
  2446  		neverSent := cc.nextStreamID
  2447  		cc.mu.Unlock()
  2448  		if f.StreamID >= neverSent {
  2449  			// We never asked for this.
  2450  			cc.logf("http2: Transport received unsolicited DATA frame; closing connection")
  2451  			return ConnectionError(ErrCodeProtocol)
  2452  		}
  2453  		// We probably did ask for this, but canceled. Just ignore it.
  2454  		// TODO: be stricter here? only silently ignore things which
  2455  		// we canceled, but not things which were closed normally
  2456  		// by the peer? Tough without accumulating too much state.
  2457  
  2458  		// But at least return their flow control:
  2459  		if f.Length > 0 {
  2460  			cc.mu.Lock()
  2461  			ok := cc.inflow.take(f.Length)
  2462  			connAdd := cc.inflow.add(int(f.Length))
  2463  			cc.mu.Unlock()
  2464  			if !ok {
  2465  				return ConnectionError(ErrCodeFlowControl)
  2466  			}
  2467  			if connAdd > 0 {
  2468  				cc.wmu.Lock()
  2469  				cc.fr.WriteWindowUpdate(0, uint32(connAdd))
  2470  				cc.bw.Flush()
  2471  				cc.wmu.Unlock()
  2472  			}
  2473  		}
  2474  		return nil
  2475  	}
  2476  	if cs.readClosed {
  2477  		cc.logf("protocol error: received DATA after END_STREAM")
  2478  		rl.endStreamError(cs, StreamError{
  2479  			StreamID: f.StreamID,
  2480  			Code:     ErrCodeProtocol,
  2481  		})
  2482  		return nil
  2483  	}
  2484  	if !cs.pastHeaders {
  2485  		cc.logf("protocol error: received DATA before a HEADERS frame")
  2486  		rl.endStreamError(cs, StreamError{
  2487  			StreamID: f.StreamID,
  2488  			Code:     ErrCodeProtocol,
  2489  		})
  2490  		return nil
  2491  	}
  2492  	if f.Length > 0 {
  2493  		if cs.isHead && len(data) > 0 {
  2494  			cc.logf("protocol error: received DATA on a HEAD request")
  2495  			rl.endStreamError(cs, StreamError{
  2496  				StreamID: f.StreamID,
  2497  				Code:     ErrCodeProtocol,
  2498  			})
  2499  			return nil
  2500  		}
  2501  		// Check connection-level flow control.
  2502  		cc.mu.Lock()
  2503  		if !takeInflows(&cc.inflow, &cs.inflow, f.Length) {
  2504  			cc.mu.Unlock()
  2505  			return ConnectionError(ErrCodeFlowControl)
  2506  		}
  2507  		// Return any padded flow control now, since we won't
  2508  		// refund it later on body reads.
  2509  		var refund int
  2510  		if pad := int(f.Length) - len(data); pad > 0 {
  2511  			refund += pad
  2512  		}
  2513  
  2514  		didReset := false
  2515  		var err error
  2516  		if len(data) > 0 {
  2517  			if _, err = cs.bufPipe.Write(data); err != nil {
  2518  				// Return len(data) now if the stream is already closed,
  2519  				// since data will never be read.
  2520  				didReset = true
  2521  				refund += len(data)
  2522  			}
  2523  		}
  2524  
  2525  		sendConn := cc.inflow.add(refund)
  2526  		var sendStream int32
  2527  		if !didReset {
  2528  			sendStream = cs.inflow.add(refund)
  2529  		}
  2530  		cc.mu.Unlock()
  2531  
  2532  		if sendConn > 0 || sendStream > 0 {
  2533  			cc.wmu.Lock()
  2534  			if sendConn > 0 {
  2535  				cc.fr.WriteWindowUpdate(0, uint32(sendConn))
  2536  			}
  2537  			if sendStream > 0 {
  2538  				cc.fr.WriteWindowUpdate(cs.ID, uint32(sendStream))
  2539  			}
  2540  			cc.bw.Flush()
  2541  			cc.wmu.Unlock()
  2542  		}
  2543  
  2544  		if err != nil {
  2545  			rl.endStreamError(cs, err)
  2546  			return nil
  2547  		}
  2548  	}
  2549  
  2550  	if f.StreamEnded() {
  2551  		rl.endStream(cs)
  2552  	}
  2553  	return nil
  2554  }
  2555  
  2556  func (rl *clientConnReadLoop) endStream(cs *clientStream) {
  2557  	// TODO: check that any declared content-length matches, like
  2558  	// server.go's (*stream).endStream method.
  2559  	if !cs.readClosed {
  2560  		cs.readClosed = true
  2561  		// Close cs.bufPipe and cs.peerClosed with cc.mu held to avoid a
  2562  		// race condition: The caller can read io.EOF from Response.Body
  2563  		// and close the body before we close cs.peerClosed, causing
  2564  		// cleanupWriteRequest to send a RST_STREAM.
  2565  		rl.cc.mu.Lock()
  2566  		defer rl.cc.mu.Unlock()
  2567  		cs.bufPipe.closeWithErrorAndCode(io.EOF, cs.copyTrailers)
  2568  		close(cs.peerClosed)
  2569  	}
  2570  }
  2571  
  2572  func (rl *clientConnReadLoop) endStreamError(cs *clientStream, err error) {
  2573  	cs.readAborted = true
  2574  	cs.abortStream(err)
  2575  }
  2576  
  2577  func (rl *clientConnReadLoop) endStreamErrorLocked(cs *clientStream, err error) {
  2578  	cs.readAborted = true
  2579  	cs.abortStreamLocked(err)
  2580  }
  2581  
  2582  // Constants passed to streamByID for documentation purposes.
  2583  const (
  2584  	headerOrDataFrame    = true
  2585  	notHeaderOrDataFrame = false
  2586  )
  2587  
  2588  // streamByID returns the stream with the given id, or nil if no stream has that id.
  2589  // If headerOrData is true, it clears rst.StreamPingsBlocked.
  2590  func (rl *clientConnReadLoop) streamByID(id uint32, headerOrData bool) *clientStream {
  2591  	rl.cc.mu.Lock()
  2592  	defer rl.cc.mu.Unlock()
  2593  	if headerOrData {
  2594  		// Work around an unfortunate gRPC behavior.
  2595  		// See comment on ClientConn.rstStreamPingsBlocked for details.
  2596  		rl.cc.rstStreamPingsBlocked = false
  2597  	}
  2598  	rl.cc.readBeforeStreamID = rl.cc.nextStreamID
  2599  	cs := rl.cc.streams[id]
  2600  	if cs != nil && !cs.readAborted {
  2601  		return cs
  2602  	}
  2603  	return nil
  2604  }
  2605  
  2606  func (cs *clientStream) copyTrailers() {
  2607  	for k, vv := range cs.trailer {
  2608  		t := cs.resTrailer
  2609  		if *t == nil {
  2610  			*t = make(Header)
  2611  		}
  2612  		(*t)[k] = vv
  2613  	}
  2614  }
  2615  
  2616  func (rl *clientConnReadLoop) processGoAway(f *GoAwayFrame) error {
  2617  	cc := rl.cc
  2618  	cc.t.connPool.MarkDead(cc)
  2619  	if f.ErrCode != 0 {
  2620  		// TODO: deal with GOAWAY more. particularly the error code
  2621  		cc.vlogf("transport got GOAWAY with error code = %v", f.ErrCode)
  2622  		if fn := cc.fr.countError; fn != nil {
  2623  			fn("recv_goaway_" + f.ErrCode.stringToken())
  2624  		}
  2625  	}
  2626  
  2627  	cc.mu.Lock()
  2628  	defer cc.mu.Unlock()
  2629  
  2630  	old := cc.goAway
  2631  	cc.goAway = f
  2632  
  2633  	// Merge the previous and current GoAway error frames.
  2634  	if cc.goAwayDebug == "" {
  2635  		cc.goAwayDebug = string(f.DebugData())
  2636  	}
  2637  	if old != nil && old.ErrCode != ErrCodeNo {
  2638  		cc.goAway.ErrCode = old.ErrCode
  2639  	}
  2640  	last := f.LastStreamID
  2641  	if len(cc.streams) == 0 {
  2642  		// Received a GOAWAY and no streams active, just close the conn.
  2643  		return errStopReadLoop
  2644  	}
  2645  	for streamID, cs := range cc.streams {
  2646  		if streamID <= last {
  2647  			// The server's GOAWAY indicates that it received this stream.
  2648  			// It will either finish processing it, or close the connection
  2649  			// without doing so. Either way, leave the stream alone for now.
  2650  			continue
  2651  		}
  2652  		if streamID == 1 && cc.goAway.ErrCode != ErrCodeNo {
  2653  			// Don't retry the first stream on a connection if we get a non-NO error.
  2654  			// If the server is sending an error on a new connection,
  2655  			// retrying the request on a new one probably isn't going to work.
  2656  			cs.abortStreamLocked(fmt.Errorf("http2: Transport received GOAWAY from server ErrCode:%v", cc.goAway.ErrCode))
  2657  		} else {
  2658  			// Aborting the stream with errClentConnGotGoAway indicates that
  2659  			// the request should be retried on a new connection.
  2660  			cs.abortStreamLocked(errClientConnGotGoAway)
  2661  		}
  2662  	}
  2663  	return nil
  2664  }
  2665  
  2666  func (rl *clientConnReadLoop) processSettings(f *SettingsFrame) error {
  2667  	cc := rl.cc
  2668  	// Locking both mu and wmu here allows frame encoding to read settings with only wmu held.
  2669  	// Acquiring wmu when f.IsAck() is unnecessary, but convenient and mostly harmless.
  2670  	cc.wmu.Lock()
  2671  	defer cc.wmu.Unlock()
  2672  
  2673  	if err := rl.processSettingsNoWrite(f); err != nil {
  2674  		return err
  2675  	}
  2676  	if !f.IsAck() {
  2677  		cc.fr.WriteSettingsAck()
  2678  		cc.bw.Flush()
  2679  	}
  2680  	return nil
  2681  }
  2682  
  2683  func (rl *clientConnReadLoop) processSettingsNoWrite(f *SettingsFrame) error {
  2684  	cc := rl.cc
  2685  	defer cc.maybeCallStateHook()
  2686  	cc.mu.Lock()
  2687  	defer cc.mu.Unlock()
  2688  
  2689  	if f.IsAck() {
  2690  		if cc.wantSettingsAck {
  2691  			cc.wantSettingsAck = false
  2692  			return nil
  2693  		}
  2694  		return ConnectionError(ErrCodeProtocol)
  2695  	}
  2696  
  2697  	var seenMaxConcurrentStreams bool
  2698  	err := f.ForeachSetting(func(s Setting) error {
  2699  		if err := s.Valid(); err != nil {
  2700  			return err
  2701  		}
  2702  		switch s.ID {
  2703  		case SettingMaxFrameSize:
  2704  			cc.maxFrameSize = s.Val
  2705  		case SettingMaxConcurrentStreams:
  2706  			cc.maxConcurrentStreams = s.Val
  2707  			seenMaxConcurrentStreams = true
  2708  		case SettingMaxHeaderListSize:
  2709  			cc.peerMaxHeaderListSize = uint64(s.Val)
  2710  		case SettingInitialWindowSize:
  2711  			// Adjust flow control of currently-open
  2712  			// frames by the difference of the old initial
  2713  			// window size and this one.
  2714  			delta := int32(s.Val) - int32(cc.initialWindowSize)
  2715  			for _, cs := range cc.streams {
  2716  				if !cs.flow.add(delta) {
  2717  					return ConnectionError(ErrCodeFlowControl)
  2718  				}
  2719  			}
  2720  			cc.cond.Broadcast()
  2721  
  2722  			cc.initialWindowSize = s.Val
  2723  		case SettingHeaderTableSize:
  2724  			cc.henc.SetMaxDynamicTableSize(s.Val)
  2725  			cc.peerMaxHeaderTableSize = s.Val
  2726  		case SettingEnableConnectProtocol:
  2727  			// If the peer wants to send us SETTINGS_ENABLE_CONNECT_PROTOCOL,
  2728  			// we require that it do so in the first SETTINGS frame.
  2729  			//
  2730  			// When we attempt to use extended CONNECT, we wait for the first
  2731  			// SETTINGS frame to see if the server supports it. If we let the
  2732  			// server enable the feature with a later SETTINGS frame, then
  2733  			// users will see inconsistent results depending on whether we've
  2734  			// seen that frame or not.
  2735  			if !cc.seenSettings {
  2736  				cc.extendedConnectAllowed = s.Val == 1
  2737  			}
  2738  		default:
  2739  			cc.vlogf("Unhandled Setting: %v", s)
  2740  		}
  2741  		return nil
  2742  	})
  2743  	if err != nil {
  2744  		return err
  2745  	}
  2746  
  2747  	if !cc.seenSettings {
  2748  		if !seenMaxConcurrentStreams {
  2749  			// This was the servers initial SETTINGS frame and it
  2750  			// didn't contain a MAX_CONCURRENT_STREAMS field so
  2751  			// increase the number of concurrent streams this
  2752  			// connection can establish to our default.
  2753  			cc.maxConcurrentStreams = defaultMaxConcurrentStreams
  2754  		}
  2755  		close(cc.seenSettingsChan)
  2756  		cc.seenSettings = true
  2757  	}
  2758  
  2759  	return nil
  2760  }
  2761  
  2762  func (rl *clientConnReadLoop) processWindowUpdate(f *WindowUpdateFrame) error {
  2763  	cc := rl.cc
  2764  	cs := rl.streamByID(f.StreamID, notHeaderOrDataFrame)
  2765  	if f.StreamID != 0 && cs == nil {
  2766  		return nil
  2767  	}
  2768  
  2769  	cc.mu.Lock()
  2770  	defer cc.mu.Unlock()
  2771  
  2772  	fl := &cc.flow
  2773  	if cs != nil {
  2774  		fl = &cs.flow
  2775  	}
  2776  	if !fl.add(int32(f.Increment)) {
  2777  		// For stream, the sender sends RST_STREAM with an error code of FLOW_CONTROL_ERROR
  2778  		if cs != nil {
  2779  			rl.endStreamErrorLocked(cs, StreamError{
  2780  				StreamID: f.StreamID,
  2781  				Code:     ErrCodeFlowControl,
  2782  			})
  2783  			return nil
  2784  		}
  2785  
  2786  		return ConnectionError(ErrCodeFlowControl)
  2787  	}
  2788  	cc.cond.Broadcast()
  2789  	return nil
  2790  }
  2791  
  2792  func (rl *clientConnReadLoop) processResetStream(f *RSTStreamFrame) error {
  2793  	cs := rl.streamByID(f.StreamID, notHeaderOrDataFrame)
  2794  	if cs == nil {
  2795  		// TODO: return error if server tries to RST_STREAM an idle stream
  2796  		return nil
  2797  	}
  2798  	serr := streamError(cs.ID, f.ErrCode)
  2799  	serr.Cause = errFromPeer
  2800  	if f.ErrCode == ErrCodeProtocol {
  2801  		rl.cc.SetDoNotReuse()
  2802  	}
  2803  	if fn := cs.cc.fr.countError; fn != nil {
  2804  		fn("recv_rststream_" + f.ErrCode.stringToken())
  2805  	}
  2806  	cs.abortStream(serr)
  2807  
  2808  	cs.bufPipe.CloseWithError(serr)
  2809  	return nil
  2810  }
  2811  
  2812  // Ping sends a PING frame to the server and waits for the ack.
  2813  func (cc *ClientConn) Ping(ctx context.Context) error {
  2814  	c := make(chan struct{})
  2815  	// Generate a random payload
  2816  	var p [8]byte
  2817  	for {
  2818  		if _, err := rand.Read(p[:]); err != nil {
  2819  			return err
  2820  		}
  2821  		cc.mu.Lock()
  2822  		// check for dup before insert
  2823  		if _, found := cc.pings[p]; !found {
  2824  			cc.pings[p] = c
  2825  			cc.mu.Unlock()
  2826  			break
  2827  		}
  2828  		cc.mu.Unlock()
  2829  	}
  2830  	var pingError error
  2831  	errc := make(chan struct{})
  2832  	go func() {
  2833  		cc.wmu.Lock()
  2834  		defer cc.wmu.Unlock()
  2835  		if pingError = cc.fr.WritePing(false, p); pingError != nil {
  2836  			close(errc)
  2837  			return
  2838  		}
  2839  		if pingError = cc.bw.Flush(); pingError != nil {
  2840  			close(errc)
  2841  			return
  2842  		}
  2843  	}()
  2844  	select {
  2845  	case <-c:
  2846  		return nil
  2847  	case <-errc:
  2848  		return pingError
  2849  	case <-ctx.Done():
  2850  		return ctx.Err()
  2851  	case <-cc.readerDone:
  2852  		// connection closed
  2853  		return cc.readerErr
  2854  	}
  2855  }
  2856  
  2857  func (rl *clientConnReadLoop) processPing(f *PingFrame) error {
  2858  	if f.IsAck() {
  2859  		cc := rl.cc
  2860  		defer cc.maybeCallStateHook()
  2861  		cc.mu.Lock()
  2862  		defer cc.mu.Unlock()
  2863  		// If ack, notify listener if any
  2864  		if c, ok := cc.pings[f.Data]; ok {
  2865  			close(c)
  2866  			delete(cc.pings, f.Data)
  2867  		}
  2868  		if cc.pendingResets > 0 {
  2869  			// See clientStream.cleanupWriteRequest.
  2870  			cc.pendingResets = 0
  2871  			cc.rstStreamPingsBlocked = true
  2872  			cc.cond.Broadcast()
  2873  		}
  2874  		return nil
  2875  	}
  2876  	cc := rl.cc
  2877  	cc.wmu.Lock()
  2878  	defer cc.wmu.Unlock()
  2879  	if err := cc.fr.WritePing(true, f.Data); err != nil {
  2880  		return err
  2881  	}
  2882  	return cc.bw.Flush()
  2883  }
  2884  
  2885  func (rl *clientConnReadLoop) processPushPromise(f *PushPromiseFrame) error {
  2886  	// We told the peer we don't want them.
  2887  	// Spec says:
  2888  	// "PUSH_PROMISE MUST NOT be sent if the SETTINGS_ENABLE_PUSH
  2889  	// setting of the peer endpoint is set to 0. An endpoint that
  2890  	// has set this setting and has received acknowledgement MUST
  2891  	// treat the receipt of a PUSH_PROMISE frame as a connection
  2892  	// error (Section 5.4.1) of type PROTOCOL_ERROR."
  2893  	return ConnectionError(ErrCodeProtocol)
  2894  }
  2895  
  2896  // writeStreamReset sends a RST_STREAM frame.
  2897  // When ping is true, it also sends a PING frame with a random payload.
  2898  func (cc *ClientConn) writeStreamReset(streamID uint32, code ErrCode, ping bool, err error) {
  2899  	// TODO: map err to more interesting error codes, once the
  2900  	// HTTP community comes up with some. But currently for
  2901  	// RST_STREAM there's no equivalent to GOAWAY frame's debug
  2902  	// data, and the error codes are all pretty vague ("cancel").
  2903  	cc.wmu.Lock()
  2904  	cc.fr.WriteRSTStream(streamID, code)
  2905  	if ping {
  2906  		var payload [8]byte
  2907  		rand.Read(payload[:])
  2908  		cc.fr.WritePing(false, payload)
  2909  	}
  2910  	cc.bw.Flush()
  2911  	cc.wmu.Unlock()
  2912  }
  2913  
  2914  var (
  2915  	errResponseHeaderListSize = errors.New("http2: response header list larger than advertised limit")
  2916  	errRequestHeaderListSize  = httpcommon.ErrRequestHeaderListSize
  2917  )
  2918  
  2919  func (cc *ClientConn) logf(format string, args ...any) {
  2920  	cc.t.logf(format, args...)
  2921  }
  2922  
  2923  func (cc *ClientConn) vlogf(format string, args ...any) {
  2924  	cc.t.vlogf(format, args...)
  2925  }
  2926  
  2927  func (t *Transport) vlogf(format string, args ...any) {
  2928  	if VerboseLogs {
  2929  		t.logf(format, args...)
  2930  	}
  2931  }
  2932  
  2933  func (t *Transport) logf(format string, args ...any) {
  2934  	log.Printf(format, args...)
  2935  }
  2936  
  2937  type missingBody struct{}
  2938  
  2939  func (missingBody) Close() error             { return nil }
  2940  func (missingBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF }
  2941  
  2942  type erringRoundTripper struct{ err error }
  2943  
  2944  func (rt erringRoundTripper) RoundTripErr() error                               { return rt.err }
  2945  func (rt erringRoundTripper) RoundTrip(*ClientRequest) (*ClientResponse, error) { return nil, rt.err }
  2946  
  2947  var errConcurrentReadOnResBody = errors.New("http2: concurrent read on response body")
  2948  
  2949  // gzipReader wraps a response body so it can lazily
  2950  // get gzip.Reader from the pool on the first call to Read.
  2951  // After Close is called it puts gzip.Reader to the pool immediately
  2952  // if there is no Read in progress or later when Read completes.
  2953  type gzipReader struct {
  2954  	_    incomparable
  2955  	body io.ReadCloser // underlying Response.Body
  2956  	mu   sync.Mutex    // guards zr and zerr
  2957  	zr   *gzip.Reader  // stores gzip reader from the pool between reads
  2958  	zerr error         // sticky gzip reader init error or sentinel value to detect concurrent read and read after close
  2959  }
  2960  
  2961  type eofReader struct{}
  2962  
  2963  func (eofReader) Read([]byte) (int, error) { return 0, io.EOF }
  2964  func (eofReader) ReadByte() (byte, error)  { return 0, io.EOF }
  2965  
  2966  var gzipPool = sync.Pool{New: func() any { return new(gzip.Reader) }}
  2967  
  2968  // gzipPoolGet gets a gzip.Reader from the pool and resets it to read from r.
  2969  func gzipPoolGet(r io.Reader) (*gzip.Reader, error) {
  2970  	zr := gzipPool.Get().(*gzip.Reader)
  2971  	if err := zr.Reset(r); err != nil {
  2972  		gzipPoolPut(zr)
  2973  		return nil, err
  2974  	}
  2975  	return zr, nil
  2976  }
  2977  
  2978  // gzipPoolPut puts a gzip.Reader back into the pool.
  2979  func gzipPoolPut(zr *gzip.Reader) {
  2980  	// Reset will allocate bufio.Reader if we pass it anything
  2981  	// other than a flate.Reader, so ensure that it's getting one.
  2982  	var r flate.Reader = eofReader{}
  2983  	zr.Reset(r)
  2984  	gzipPool.Put(zr)
  2985  }
  2986  
  2987  // acquire returns a gzip.Reader for reading response body.
  2988  // The reader must be released after use.
  2989  func (gz *gzipReader) acquire() (*gzip.Reader, error) {
  2990  	gz.mu.Lock()
  2991  	defer gz.mu.Unlock()
  2992  	if gz.zerr != nil {
  2993  		return nil, gz.zerr
  2994  	}
  2995  	if gz.zr == nil {
  2996  		// gzipPoolGet might block indefinitely since it reads the gzip header.
  2997  		// Therefore, drop mu temporarily when using gzipPoolGet.
  2998  		// We set zerr to errConcurrentReadOnResBody to prevent concurrent read
  2999  		// even when mu is temporarily dropped.
  3000  		gz.zerr = errConcurrentReadOnResBody
  3001  		gz.mu.Unlock()
  3002  		zr, err := gzipPoolGet(gz.body)
  3003  		gz.mu.Lock()
  3004  		// Guard against Close being called while gzipPoolGet is running.
  3005  		if gz.zerr != errConcurrentReadOnResBody {
  3006  			if zr != nil {
  3007  				gzipPoolPut(zr)
  3008  			}
  3009  			return nil, gz.zerr
  3010  		}
  3011  		gz.zr, gz.zerr = zr, err
  3012  		if gz.zerr != nil {
  3013  			return nil, gz.zerr
  3014  		}
  3015  	}
  3016  	ret := gz.zr
  3017  	gz.zr, gz.zerr = nil, errConcurrentReadOnResBody
  3018  	return ret, nil
  3019  }
  3020  
  3021  // release returns the gzip.Reader to the pool if Close was called during Read.
  3022  func (gz *gzipReader) release(zr *gzip.Reader) {
  3023  	gz.mu.Lock()
  3024  	defer gz.mu.Unlock()
  3025  	if gz.zerr == errConcurrentReadOnResBody {
  3026  		gz.zr, gz.zerr = zr, nil
  3027  	} else { // fs.ErrClosed
  3028  		gzipPoolPut(zr)
  3029  	}
  3030  }
  3031  
  3032  // close returns the gzip.Reader to the pool immediately or
  3033  // signals release to do so after Read completes.
  3034  func (gz *gzipReader) close() {
  3035  	gz.mu.Lock()
  3036  	defer gz.mu.Unlock()
  3037  	if gz.zerr == nil && gz.zr != nil {
  3038  		gzipPoolPut(gz.zr)
  3039  		gz.zr = nil
  3040  	}
  3041  	gz.zerr = fs.ErrClosed
  3042  }
  3043  
  3044  func (gz *gzipReader) Read(p []byte) (n int, err error) {
  3045  	zr, err := gz.acquire()
  3046  	if err != nil {
  3047  		return 0, err
  3048  	}
  3049  	defer gz.release(zr)
  3050  
  3051  	return zr.Read(p)
  3052  }
  3053  
  3054  func (gz *gzipReader) Close() error {
  3055  	gz.close()
  3056  
  3057  	return gz.body.Close()
  3058  }
  3059  
  3060  // isConnectionCloseRequest reports whether req should use its own
  3061  // connection for a single request and then close the connection.
  3062  func isConnectionCloseRequest(req *ClientRequest) bool {
  3063  	return req.Close || httpguts.HeaderValuesContainsToken(req.Header["Connection"], "close")
  3064  }
  3065  
  3066  // netHTTPClientConn wraps ClientConn and implements the interface net/http expects from
  3067  // the RoundTripper returned by NewClientConn.
  3068  type NetHTTPClientConn struct {
  3069  	cc *ClientConn
  3070  }
  3071  
  3072  func (cc NetHTTPClientConn) RoundTrip(req *ClientRequest) (*ClientResponse, error) {
  3073  	return cc.cc.RoundTrip(req)
  3074  }
  3075  
  3076  func (cc NetHTTPClientConn) Close() error {
  3077  	return cc.cc.Close()
  3078  }
  3079  
  3080  func (cc NetHTTPClientConn) Err() error {
  3081  	cc.cc.mu.Lock()
  3082  	defer cc.cc.mu.Unlock()
  3083  	if cc.cc.closed {
  3084  		return errors.New("connection closed")
  3085  	}
  3086  	return nil
  3087  }
  3088  
  3089  func (cc NetHTTPClientConn) Reserve() error {
  3090  	defer cc.cc.maybeCallStateHook()
  3091  	cc.cc.mu.Lock()
  3092  	defer cc.cc.mu.Unlock()
  3093  	if !cc.cc.canReserveLocked() {
  3094  		return errors.New("connection is unavailable")
  3095  	}
  3096  	cc.cc.streamsReserved++
  3097  	return nil
  3098  }
  3099  
  3100  func (cc NetHTTPClientConn) Release() {
  3101  	defer cc.cc.maybeCallStateHook()
  3102  	cc.cc.mu.Lock()
  3103  	defer cc.cc.mu.Unlock()
  3104  	// We don't complain if streamsReserved is 0.
  3105  	//
  3106  	// This is consistent with RoundTrip: both Release and RoundTrip will
  3107  	// consume a reservation iff one exists.
  3108  	if cc.cc.streamsReserved > 0 {
  3109  		cc.cc.streamsReserved--
  3110  	}
  3111  }
  3112  
  3113  func (cc NetHTTPClientConn) Available() int {
  3114  	cc.cc.mu.Lock()
  3115  	defer cc.cc.mu.Unlock()
  3116  	return cc.cc.availableLocked()
  3117  }
  3118  
  3119  func (cc NetHTTPClientConn) InFlight() int {
  3120  	cc.cc.mu.Lock()
  3121  	defer cc.cc.mu.Unlock()
  3122  	return cc.cc.currentRequestCountLocked()
  3123  }
  3124  
  3125  func (cc NetHTTPClientConn) Ping(ctx context.Context) error {
  3126  	return cc.cc.Ping(ctx)
  3127  }
  3128  
  3129  func (cc *ClientConn) maybeCallStateHook() {
  3130  	if cc.internalStateHook != nil {
  3131  		cc.internalStateHook()
  3132  	}
  3133  }
  3134  
  3135  func (t *Transport) idleConnTimeout() time.Duration {
  3136  	if t.t1 != nil {
  3137  		return t.t1.IdleConnTimeout()
  3138  	}
  3139  
  3140  	return 0
  3141  }
  3142  
  3143  func traceGetConn(req *ClientRequest, hostPort string) {
  3144  	trace := httptrace.ContextClientTrace(req.Context)
  3145  	if trace == nil || trace.GetConn == nil {
  3146  		return
  3147  	}
  3148  	trace.GetConn(hostPort)
  3149  }
  3150  
  3151  func traceGotConn(req *ClientRequest, cc *ClientConn, reused bool) {
  3152  	trace := httptrace.ContextClientTrace(req.Context)
  3153  	if trace == nil || trace.GotConn == nil {
  3154  		return
  3155  	}
  3156  	ci := httptrace.GotConnInfo{Conn: cc.tconn}
  3157  	ci.Reused = reused
  3158  	cc.mu.Lock()
  3159  	ci.WasIdle = len(cc.streams) == 0 && reused
  3160  	if ci.WasIdle && !cc.lastActive.IsZero() {
  3161  		ci.IdleTime = time.Since(cc.lastActive)
  3162  	}
  3163  	cc.mu.Unlock()
  3164  
  3165  	trace.GotConn(ci)
  3166  }
  3167  
  3168  func traceWroteHeaders(trace *httptrace.ClientTrace) {
  3169  	if trace != nil && trace.WroteHeaders != nil {
  3170  		trace.WroteHeaders()
  3171  	}
  3172  }
  3173  
  3174  func traceGot100Continue(trace *httptrace.ClientTrace) {
  3175  	if trace != nil && trace.Got100Continue != nil {
  3176  		trace.Got100Continue()
  3177  	}
  3178  }
  3179  
  3180  func traceWait100Continue(trace *httptrace.ClientTrace) {
  3181  	if trace != nil && trace.Wait100Continue != nil {
  3182  		trace.Wait100Continue()
  3183  	}
  3184  }
  3185  
  3186  func traceWroteRequest(trace *httptrace.ClientTrace, err error) {
  3187  	if trace != nil && trace.WroteRequest != nil {
  3188  		trace.WroteRequest(httptrace.WroteRequestInfo{Err: err})
  3189  	}
  3190  }
  3191  
  3192  func traceFirstResponseByte(trace *httptrace.ClientTrace) {
  3193  	if trace != nil && trace.GotFirstResponseByte != nil {
  3194  		trace.GotFirstResponseByte()
  3195  	}
  3196  }
  3197  
  3198  func traceGot1xxResponseFunc(trace *httptrace.ClientTrace) func(int, textproto.MIMEHeader) error {
  3199  	if trace != nil {
  3200  		return trace.Got1xxResponse
  3201  	}
  3202  	return nil
  3203  }
  3204  
  3205  // dialTLSWithContext uses tls.Dialer, added in Go 1.15, to open a TLS
  3206  // connection.
  3207  func (t *Transport) dialTLSWithContext(ctx context.Context, network, addr string, cfg *tls.Config) (*tls.Conn, error) {
  3208  	dialer := &tls.Dialer{
  3209  		Config: cfg,
  3210  	}
  3211  	cn, err := dialer.DialContext(ctx, network, addr)
  3212  	if err != nil {
  3213  		return nil, err
  3214  	}
  3215  	tlsCn := cn.(*tls.Conn) // DialContext comment promises this will always succeed
  3216  	return tlsCn, nil
  3217  }
  3218  

View as plain text