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

View as plain text