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

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

View as plain text