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.
36
"golang.org/x/net/http/httpguts"
37
"golang.org/x/net/http2/hpack"
38
"golang.org/x/net/idna"
42
// transportDefaultConnFlow is how many connection-level flow control
43
// tokens we give the server at start-up, past the default 64k.
44
transportDefaultConnFlow = 1 << 30
46
// transportDefaultStreamFlow is how many stream-level flow
47
// control tokens we announce to the peer, and how many bytes
48
// we buffer per stream.
49
transportDefaultStreamFlow = 4 << 20
51
defaultUserAgent = "Go-http-client/2.0"
53
// initialMaxConcurrentStreams is a connections maxConcurrentStreams until
54
// it's received servers initial SETTINGS frame, which corresponds with the
55
// spec's minimum recommended value.
56
initialMaxConcurrentStreams = 100
58
// defaultMaxConcurrentStreams is a connections default maxConcurrentStreams
59
// if the server doesn't include one in its initial SETTINGS frame.
60
defaultMaxConcurrentStreams = 1000
63
// Transport is an HTTP/2 Transport.
65
// A Transport internally caches connections to servers. It is safe
66
// for concurrent use by multiple goroutines.
67
type Transport struct {68
// DialTLSContext specifies an optional dial function with context for
69
// creating TLS connections for requests.
71
// If DialTLSContext and DialTLS is nil, tls.Dial is used.
73
// If the returned net.Conn has a ConnectionState method like tls.Conn,
74
// it will be used to set http.Response.TLS.
75
DialTLSContext func(ctx context.Context, network, addr string, cfg *tls.Config) (net.Conn, error)
77
// DialTLS specifies an optional dial function for creating
78
// TLS connections for requests.
80
// If DialTLSContext and DialTLS is nil, tls.Dial is used.
82
// Deprecated: Use DialTLSContext instead, which allows the transport
83
// to cancel dials as soon as they are no longer needed.
84
// If both are set, DialTLSContext takes priority.
85
DialTLS func(network, addr string, cfg *tls.Config) (net.Conn, error)
87
// TLSClientConfig specifies the TLS configuration to use with
88
// tls.Client. If nil, the default configuration is used.
89
TLSClientConfig *tls.Config
91
// ConnPool optionally specifies an alternate connection pool to use.
92
// If nil, the default is used.
93
ConnPool ClientConnPool
95
// DisableCompression, if true, prevents the Transport from
96
// requesting compression with an "Accept-Encoding: gzip"
97
// request header when the Request contains no existing
98
// Accept-Encoding value. If the Transport requests gzip on
99
// its own and gets a gzipped response, it's transparently
100
// decoded in the Response.Body. However, if the user
101
// explicitly requested gzip it is not automatically
103
DisableCompression bool
105
// AllowHTTP, if true, permits HTTP/2 requests using the insecure,
106
// plain-text "http" scheme. Note that this does not enable h2c support.
109
// MaxHeaderListSize is the http2 SETTINGS_MAX_HEADER_LIST_SIZE to
110
// send in the initial settings frame. It is how many bytes
111
// of response headers are allowed. Unlike the http2 spec, zero here
112
// means to use a default limit (currently 10MB). If you actually
113
// want to advertise an unlimited value to the peer, Transport
114
// interprets the highest possible value here (0xffffffff or 1<<32-1)
116
MaxHeaderListSize uint32
118
// MaxReadFrameSize is the http2 SETTINGS_MAX_FRAME_SIZE to send in the
119
// initial settings frame. It is the size in bytes of the largest frame
120
// payload that the sender is willing to receive. If 0, no setting is
121
// sent, and the value is provided by the peer, which should be 16384
122
// according to the spec:
123
// https://datatracker.ietf.org/doc/html/rfc7540#section-6.5.2.
124
// Values are bounded in the range 16k to 16M.
125
MaxReadFrameSize uint32
127
// MaxDecoderHeaderTableSize optionally specifies the http2
128
// SETTINGS_HEADER_TABLE_SIZE to send in the initial settings frame. It
129
// informs the remote endpoint of the maximum size of the header compression
130
// table used to decode header blocks, in octets. If zero, the default value
132
MaxDecoderHeaderTableSize uint32
134
// MaxEncoderHeaderTableSize optionally specifies an upper limit for the
135
// header compression table used for encoding request headers. Received
136
// SETTINGS_HEADER_TABLE_SIZE settings are capped at this limit. If zero,
137
// the default value of 4096 is used.
138
MaxEncoderHeaderTableSize uint32
140
// StrictMaxConcurrentStreams controls whether the server's
141
// SETTINGS_MAX_CONCURRENT_STREAMS should be respected
142
// globally. If false, new TCP connections are created to the
143
// server as needed to keep each under the per-connection
144
// SETTINGS_MAX_CONCURRENT_STREAMS limit. If true, the
145
// server's SETTINGS_MAX_CONCURRENT_STREAMS is interpreted as
146
// a global limit and callers of RoundTrip block when needed,
147
// waiting for their turn.
148
StrictMaxConcurrentStreams bool
150
// IdleConnTimeout is the maximum amount of time an idle
151
// (keep-alive) connection will remain idle before closing
153
// Zero means no limit.
154
IdleConnTimeout time.Duration
156
// ReadIdleTimeout is the timeout after which a health check using ping
157
// frame will be carried out if no frame is received on the connection.
158
// Note that a ping response will is considered a received frame, so if
159
// there is no other traffic on the connection, the health check will
160
// be performed every ReadIdleTimeout interval.
161
// If zero, no health check is performed.
162
ReadIdleTimeout time.Duration
164
// PingTimeout is the timeout after which the connection will be closed
165
// if a response to Ping is not received.
167
PingTimeout time.Duration
169
// WriteByteTimeout is the timeout after which the connection will be
170
// closed no data can be written to it. The timeout begins when data is
171
// available to write, and is extended whenever any bytes are written.
172
WriteByteTimeout time.Duration
174
// CountError, if non-nil, is called on HTTP/2 transport errors.
175
// It's intended to increment a metric for monitoring, such
176
// as an expvar or Prometheus metric.
177
// The errType consists of only ASCII word characters.
178
CountError func(errType string)
180
// t1, if non-nil, is the standard library Transport using
181
// this transport. Its settings are used (but not its
182
// RoundTrip method, etc).
185
connPoolOnce sync.Once
186
connPoolOrDef ClientConnPool // non-nil version of ConnPool
188
syncHooks *testSyncHooks
191
func (t *Transport) maxHeaderListSize() uint32 {192
if t.MaxHeaderListSize == 0 {195
if t.MaxHeaderListSize == 0xffffffff {198
return t.MaxHeaderListSize
201
func (t *Transport) maxFrameReadSize() uint32 {202
if t.MaxReadFrameSize == 0 {203
return 0 // use the default provided by the peer
205
if t.MaxReadFrameSize < minMaxFrameSize {206
return minMaxFrameSize
208
if t.MaxReadFrameSize > maxFrameSize {211
return t.MaxReadFrameSize
214
func (t *Transport) disableCompression() bool {215
return t.DisableCompression || (t.t1 != nil && t.t1.DisableCompression)
218
func (t *Transport) pingTimeout() time.Duration {219
if t.PingTimeout == 0 {220
return 15 * time.Second
226
// ConfigureTransport configures a net/http HTTP/1 Transport to use HTTP/2.
227
// It returns an error if t1 has already been HTTP/2-enabled.
229
// Use ConfigureTransports instead to configure the HTTP/2 Transport.
230
func ConfigureTransport(t1 *http.Transport) error {231
_, err := ConfigureTransports(t1)
235
// ConfigureTransports configures a net/http HTTP/1 Transport to use HTTP/2.
236
// It returns a new HTTP/2 Transport for further configuration.
237
// It returns an error if t1 has already been HTTP/2-enabled.
238
func ConfigureTransports(t1 *http.Transport) (*Transport, error) {239
return configureTransports(t1)
242
func configureTransports(t1 *http.Transport) (*Transport, error) {243
connPool := new(clientConnPool)
245
ConnPool: noDialClientConnPool{connPool},249
if err := registerHTTPSProtocol(t1, noDialH2RoundTripper{t2}); err != nil {252
if t1.TLSClientConfig == nil {253
t1.TLSClientConfig = new(tls.Config)
255
if !strSliceContains(t1.TLSClientConfig.NextProtos, "h2") {256
t1.TLSClientConfig.NextProtos = append([]string{"h2"}, t1.TLSClientConfig.NextProtos...)258
if !strSliceContains(t1.TLSClientConfig.NextProtos, "http/1.1") {259
t1.TLSClientConfig.NextProtos = append(t1.TLSClientConfig.NextProtos, "http/1.1")
261
upgradeFn := func(authority string, c *tls.Conn) http.RoundTripper {262
addr := authorityAddr("https", authority)263
if used, err := connPool.addConnIfNeeded(addr, t2, c); err != nil {265
return erringRoundTripper{err}267
// Turns out we don't need this c.
268
// For example, two goroutines made requests to the same host
269
// at the same time, both kicking off TCP dials. (since protocol
275
if m := t1.TLSNextProto; len(m) == 0 {276
t1.TLSNextProto = map[string]func(string, *tls.Conn) http.RoundTripper{285
func (t *Transport) connPool() ClientConnPool {286
t.connPoolOnce.Do(t.initConnPool)
287
return t.connPoolOrDef
290
func (t *Transport) initConnPool() {291
if t.ConnPool != nil {292
t.connPoolOrDef = t.ConnPool
294
t.connPoolOrDef = &clientConnPool{t: t}298
// ClientConn is the state of a single HTTP/2 client connection to an
300
type ClientConn struct {302
tconn net.Conn // usually *tls.Conn, except specialized impls
303
tlsState *tls.ConnectionState // nil only for specialized impls
304
reused uint32 // whether conn is being reused; atomic
305
singleUse bool // whether being used for a single http.Request
306
getConnCalled bool // used by clientConnPool
308
// readLoop goroutine fields:
309
readerDone chan struct{} // closed on error310
readerErr error // set before readerDone is closed
312
idleTimeout time.Duration // or 0 for never
315
mu sync.Mutex // guards following
316
cond *sync.Cond // hold mu; broadcast on flow/closed changes
317
flow outflow // our conn-level flow control quota (cs.outflow is per stream)
318
inflow inflow // peer's conn-level flow control
319
doNotReuse bool // whether conn is marked to not be reused for any future requests
322
seenSettings bool // true if we've seen a settings frame, false otherwise
323
wantSettingsAck bool // we sent a SETTINGS frame and haven't heard back
324
goAway *GoAwayFrame // if non-nil, the GoAwayFrame we received
325
goAwayDebug string // goAway frame's debug data, retained as a string
326
streams map[uint32]*clientStream // client-initiated
327
streamsReserved int // incr by ReserveNewRequest; decr on RoundTrip
329
pendingRequests int // requests blocked and waiting to be sent because len(streams) == maxConcurrentStreams
330
pings map[[8]byte]chan struct{} // in flight ping data to notification channel333
lastIdle time.Time // time last idle
334
// Settings from peer: (also guarded by wmu)
336
maxConcurrentStreams uint32
337
peerMaxHeaderListSize uint64
338
peerMaxHeaderTableSize uint32
339
initialWindowSize uint32
341
// reqHeaderMu is a 1-element semaphore channel controlling access to sending new requests.
342
// Write to reqHeaderMu to lock it, read from it to unlock.
343
// Lock reqmu BEFORE mu or wmu.
344
reqHeaderMu chan struct{}346
// wmu is held while writing.
347
// Acquire BEFORE mu when holding both, to avoid blocking mu on network writes.
348
// Only acquire both at the same time when changing peer settings.
352
werr error // first write error that has occurred
353
hbuf bytes.Buffer // HPACK encoder writes into this
356
syncHooks *testSyncHooks // can be nil
359
// Hook points used for testing.
360
// Outside of tests, cc.syncHooks is nil and these all have minimal implementations.
361
// Inside tests, see the testSyncHooks function docs.
363
// goRun starts a new goroutine.
364
func (cc *ClientConn) goRun(f func()) {365
if cc.syncHooks != nil {366
cc.syncHooks.goRun(f)
372
// condBroadcast is cc.cond.Broadcast.
373
func (cc *ClientConn) condBroadcast() {374
if cc.syncHooks != nil {375
cc.syncHooks.condBroadcast(cc.cond)
380
// condWait is cc.cond.Wait.
381
func (cc *ClientConn) condWait() {382
if cc.syncHooks != nil {383
cc.syncHooks.condWait(cc.cond)
388
// newTimer creates a new time.Timer, or a synthetic timer in tests.
389
func (cc *ClientConn) newTimer(d time.Duration) timer {390
if cc.syncHooks != nil {391
return cc.syncHooks.newTimer(d)
393
return newTimeTimer(d)
396
// afterFunc creates a new time.AfterFunc timer, or a synthetic timer in tests.
397
func (cc *ClientConn) afterFunc(d time.Duration, f func()) timer {398
if cc.syncHooks != nil {399
return cc.syncHooks.afterFunc(d, f)
401
return newTimeAfterFunc(d, f)
404
func (cc *ClientConn) contextWithTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) {405
if cc.syncHooks != nil {406
return cc.syncHooks.contextWithTimeout(ctx, d)
408
return context.WithTimeout(ctx, d)
411
// clientStream is the state for a single HTTP/2 stream. One of these
412
// is created for each Transport.RoundTrip call.
413
type clientStream struct {416
// Fields of Request that we may access even after the response body is closed.
418
reqCancel <-chan struct{}420
trace *httptrace.ClientTrace // or nil
422
bufPipe pipe // buffered pipe with the flow-controlled response payload
427
abort chan struct{} // closed to signal stream should end immediately428
abortErr error // set if abort is closed
430
peerClosed chan struct{} // closed when the peer sends an END_STREAM flag431
donec chan struct{} // closed after the stream is in the closed state432
on100 chan struct{} // buffered; written to if a 100 is received434
respHeaderRecv chan struct{} // closed when headers are received435
res *http.Response // set if respHeaderRecv is closed
437
flow outflow // guarded by cc.mu
438
inflow inflow // guarded by cc.mu
439
bytesRemain int64 // -1 means unknown; owned by transportResponseBody.Read
440
readErr error // sticky read error; owned by transportResponseBody.Read
442
reqBody io.ReadCloser
443
reqBodyContentLength int64 // -1 means unknown
444
reqBodyClosed chan struct{} // guarded by cc.mu; non-nil on Close, closed when done446
// owned by writeRequest:
447
sentEndStream bool // sent an END_STREAM flag to the peer
450
// owned by clientConnReadLoop:
451
firstByte bool // got the first response byte
452
pastHeaders bool // got first MetaHeadersFrame (actual headers)
453
pastTrailers bool // got optional second MetaHeadersFrame (trailers)
454
num1xx uint8 // number of 1xx responses seen
455
readClosed bool // peer sent an END_STREAM flag
456
readAborted bool // read loop reset the stream
458
trailer http.Header // accumulated trailers
459
resTrailer *http.Header // client's Response.Trailer
462
var got1xxFuncForTests func(int, textproto.MIMEHeader) error
464
// get1xxTraceFunc returns the value of request's httptrace.ClientTrace.Got1xxResponse func,
465
// if any. It returns nil if not set or if the Go version is too old.
466
func (cs *clientStream) get1xxTraceFunc() func(int, textproto.MIMEHeader) error {467
if fn := got1xxFuncForTests; fn != nil {470
return traceGot1xxResponseFunc(cs.trace)
473
func (cs *clientStream) abortStream(err error) {475
defer cs.cc.mu.Unlock()
476
cs.abortStreamLocked(err)
479
func (cs *clientStream) abortStreamLocked(err error) {480
cs.abortOnce.Do(func() {484
if cs.reqBody != nil {485
cs.closeReqBodyLocked()
487
// TODO(dneil): Clean up tests where cs.cc.cond is nil.
488
if cs.cc.cond != nil {489
// Wake up writeRequestBody if it is waiting on flow control.
490
cs.cc.condBroadcast()
494
func (cs *clientStream) abortRequestBodyWrite() {498
if cs.reqBody != nil && cs.reqBodyClosed == nil {499
cs.closeReqBodyLocked()
504
func (cs *clientStream) closeReqBodyLocked() {505
if cs.reqBodyClosed != nil {508
cs.reqBodyClosed = make(chan struct{})509
reqBodyClosed := cs.reqBodyClosed
516
type stickyErrWriter struct {518
timeout time.Duration
522
func (sew stickyErrWriter) Write(p []byte) (n int, err error) {527
if sew.timeout != 0 {528
sew.conn.SetWriteDeadline(time.Now().Add(sew.timeout))
530
nn, err := sew.conn.Write(p[n:])
532
if n < len(p) && nn > 0 && errors.Is(err, os.ErrDeadlineExceeded) {533
// Keep extending the deadline so long as we're making progress.
536
if sew.timeout != 0 {537
sew.conn.SetWriteDeadline(time.Time{})544
// noCachedConnError is the concrete type of ErrNoCachedConn, which
545
// needs to be detected by net/http regardless of whether it's its
546
// bundled version (in h2_bundle.go with a rewritten type name) or
547
// from a user's x/net/http2. As such, as it has a unique method name
548
// (IsHTTP2NoCachedConnError) that net/http sniffs for via func
549
// isNoCachedConnError.
550
type noCachedConnError struct{}552
func (noCachedConnError) IsHTTP2NoCachedConnError() {}553
func (noCachedConnError) Error() string { return "http2: no cached connection was available" }555
// isNoCachedConnError reports whether err is of type noCachedConnError
556
// or its equivalent renamed type in net/http2's h2_bundle.go. Both types
557
// may coexist in the same running program.
558
func isNoCachedConnError(err error) bool {559
_, ok := err.(interface{ IsHTTP2NoCachedConnError() })563
var ErrNoCachedConn error = noCachedConnError{}565
// RoundTripOpt are options for the Transport.RoundTripOpt method.
566
type RoundTripOpt struct {567
// OnlyCachedConn controls whether RoundTripOpt may
568
// create a new TCP connection. If set true and
569
// no cached connection is available, RoundTripOpt
570
// will return ErrNoCachedConn.
574
func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {575
return t.RoundTripOpt(req, RoundTripOpt{})578
// authorityAddr returns a given authority (a host/IP, or host:port / ip:port)
579
// and returns a host:port. The port 443 is added if needed.
580
func authorityAddr(scheme string, authority string) (addr string) {581
host, port, err := net.SplitHostPort(authority)
582
if err != nil { // authority didn't have a port586
if port == "" { // authority's port was empty588
if scheme == "http" {592
if a, err := idna.ToASCII(host); err == nil {595
// IPv6 address literal, without a port:
596
if strings.HasPrefix(host, "[") && strings.HasSuffix(host, "]") {597
return host + ":" + port
599
return net.JoinHostPort(host, port)
602
// RoundTripOpt is like RoundTrip, but takes options.
603
func (t *Transport) RoundTripOpt(req *http.Request, opt RoundTripOpt) (*http.Response, error) {604
if !(req.URL.Scheme == "https" || (req.URL.Scheme == "http" && t.AllowHTTP)) {605
return nil, errors.New("http2: unsupported scheme")608
addr := authorityAddr(req.URL.Scheme, req.URL.Host)
609
for retry := 0; ; retry++ {610
cc, err := t.connPool().GetClientConn(req, addr)
612
t.vlogf("http2: Transport failed to get client conn for %s: %v", addr, err)615
reused := !atomic.CompareAndSwapUint32(&cc.reused, 0, 1)
616
traceGotConn(req, cc, reused)
617
res, err := cc.RoundTrip(req)
618
if err != nil && retry <= 6 {620
if req, err = shouldRetryRequest(req, err); err == nil {621
// After the first retry, do exponential backoff with 10% jitter.
623
t.vlogf("RoundTrip retrying after failure: %v", roundTripErr)626
backoff := float64(uint(1) << (uint(retry) - 1))
627
backoff += backoff * (0.1 * mathrand.Float64())
628
d := time.Second * time.Duration(backoff)
630
if t.syncHooks != nil {631
tm = t.syncHooks.newTimer(d)
632
t.syncHooks.blockUntil(func() bool {635
case <-req.Context().Done():
646
t.vlogf("RoundTrip retrying after failure: %v", roundTripErr)648
case <-req.Context().Done():
650
err = req.Context().Err()
655
t.vlogf("RoundTrip failure: %v", err)662
// CloseIdleConnections closes any connections which were previously
663
// connected from previous requests but are now sitting idle.
664
// It does not interrupt any connections currently in use.
665
func (t *Transport) CloseIdleConnections() {666
if cp, ok := t.connPool().(clientConnPoolIdleCloser); ok {667
cp.closeIdleConnections()
672
errClientConnClosed = errors.New("http2: client conn is closed")673
errClientConnUnusable = errors.New("http2: client conn not usable")674
errClientConnGotGoAway = errors.New("http2: Transport received Server's graceful shutdown GOAWAY")677
// shouldRetryRequest is called by RoundTrip when a request fails to get
678
// response headers. It is always called with a non-nil error.
679
// It returns either a request to retry (either the same request, or a
680
// modified clone), or an error if the request can't be replayed.
681
func shouldRetryRequest(req *http.Request, err error) (*http.Request, error) {682
if !canRetryError(err) {685
// If the Body is nil (or http.NoBody), it's safe to reuse
686
// this request and its Body.
687
if req.Body == nil || req.Body == http.NoBody {691
// If the request body can be reset back to its original
692
// state via the optional req.GetBody, do that.
693
if req.GetBody != nil {694
body, err := req.GetBody()
703
// The Request.Body can't reset back to the beginning, but we
704
// don't seem to have started to read from it yet, so reuse
705
// the request directly.
706
if err == errClientConnUnusable {710
return nil, fmt.Errorf("http2: Transport: cannot retry err [%v] after Request.Body was written; define Request.GetBody to avoid this error", err)713
func canRetryError(err error) bool {714
if err == errClientConnUnusable || err == errClientConnGotGoAway {717
if se, ok := err.(StreamError); ok {718
if se.Code == ErrCodeProtocol && se.Cause == errFromPeer {719
// See golang/go#47635, golang/go#42777
722
return se.Code == ErrCodeRefusedStream
727
func (t *Transport) dialClientConn(ctx context.Context, addr string, singleUse bool) (*ClientConn, error) {728
if t.syncHooks != nil {729
return t.newClientConn(nil, singleUse, t.syncHooks)
731
host, _, err := net.SplitHostPort(addr)
735
tconn, err := t.dialTLS(ctx, "tcp", addr, t.newTLSConfig(host))
739
return t.newClientConn(tconn, singleUse, nil)
742
func (t *Transport) newTLSConfig(host string) *tls.Config {743
cfg := new(tls.Config)
744
if t.TLSClientConfig != nil {745
*cfg = *t.TLSClientConfig.Clone()
747
if !strSliceContains(cfg.NextProtos, NextProtoTLS) {748
cfg.NextProtos = append([]string{NextProtoTLS}, cfg.NextProtos...)750
if cfg.ServerName == "" {751
cfg.ServerName = host
756
func (t *Transport) dialTLS(ctx context.Context, network, addr string, tlsCfg *tls.Config) (net.Conn, error) {757
if t.DialTLSContext != nil {758
return t.DialTLSContext(ctx, network, addr, tlsCfg)
759
} else if t.DialTLS != nil {760
return t.DialTLS(network, addr, tlsCfg)
763
tlsCn, err := t.dialTLSWithContext(ctx, network, addr, tlsCfg)
767
state := tlsCn.ConnectionState()
768
if p := state.NegotiatedProtocol; p != NextProtoTLS {769
return nil, fmt.Errorf("http2: unexpected ALPN protocol %q; want %q", p, NextProtoTLS)771
if !state.NegotiatedProtocolIsMutual {772
return nil, errors.New("http2: could not negotiate protocol mutually")777
// disableKeepAlives reports whether connections should be closed as
778
// soon as possible after handling the first request.
779
func (t *Transport) disableKeepAlives() bool {780
return t.t1 != nil && t.t1.DisableKeepAlives
783
func (t *Transport) expectContinueTimeout() time.Duration {787
return t.t1.ExpectContinueTimeout
790
func (t *Transport) maxDecoderHeaderTableSize() uint32 {791
if v := t.MaxDecoderHeaderTableSize; v > 0 {794
return initialHeaderTableSize
797
func (t *Transport) maxEncoderHeaderTableSize() uint32 {798
if v := t.MaxEncoderHeaderTableSize; v > 0 {801
return initialHeaderTableSize
804
func (t *Transport) NewClientConn(c net.Conn) (*ClientConn, error) {805
return t.newClientConn(c, t.disableKeepAlives(), nil)
808
func (t *Transport) newClientConn(c net.Conn, singleUse bool, hooks *testSyncHooks) (*ClientConn, error) {812
readerDone: make(chan struct{}),814
maxFrameSize: 16 << 10, // spec default
815
initialWindowSize: 65535, // spec default
816
maxConcurrentStreams: initialMaxConcurrentStreams, // "infinite", per spec. Use a smaller value until we have received server settings.
817
peerMaxHeaderListSize: 0xffffffffffffffff, // "infinite", per spec. Use 2^64-1 instead.
818
streams: make(map[uint32]*clientStream),
819
singleUse: singleUse,
820
wantSettingsAck: true,
821
pings: make(map[[8]byte]chan struct{}),822
reqHeaderMu: make(chan struct{}, 1),826
hooks.newclientconn(cc)
829
if d := t.idleConnTimeout(); d != 0 {831
cc.idleTimer = cc.afterFunc(d, cc.onIdleTimeout)
834
t.vlogf("http2: Transport creating client conn %p to %v", cc, c.RemoteAddr())837
cc.cond = sync.NewCond(&cc.mu)
838
cc.flow.add(int32(initialWindowSize))
840
// TODO: adjust this writer size to account for frame size +
841
// MTU + crypto/tls record padding.
842
cc.bw = bufio.NewWriter(stickyErrWriter{844
timeout: t.WriteByteTimeout,
847
cc.br = bufio.NewReader(c)
848
cc.fr = NewFramer(cc.bw, cc.br)
849
if t.maxFrameReadSize() != 0 {850
cc.fr.SetMaxReadFrameSize(t.maxFrameReadSize())
852
if t.CountError != nil {853
cc.fr.countError = t.CountError
855
maxHeaderTableSize := t.maxDecoderHeaderTableSize()
856
cc.fr.ReadMetaHeaders = hpack.NewDecoder(maxHeaderTableSize, nil)
857
cc.fr.MaxHeaderListSize = t.maxHeaderListSize()
859
cc.henc = hpack.NewEncoder(&cc.hbuf)
860
cc.henc.SetMaxDynamicTableSizeLimit(t.maxEncoderHeaderTableSize())
861
cc.peerMaxHeaderTableSize = initialHeaderTableSize
867
if cs, ok := c.(connectionStater); ok {868
state := cs.ConnectionState()
872
initialSettings := []Setting{873
{ID: SettingEnablePush, Val: 0},874
{ID: SettingInitialWindowSize, Val: transportDefaultStreamFlow},876
if max := t.maxFrameReadSize(); max != 0 {877
initialSettings = append(initialSettings, Setting{ID: SettingMaxFrameSize, Val: max})879
if max := t.maxHeaderListSize(); max != 0 {880
initialSettings = append(initialSettings, Setting{ID: SettingMaxHeaderListSize, Val: max})882
if maxHeaderTableSize != initialHeaderTableSize {883
initialSettings = append(initialSettings, Setting{ID: SettingHeaderTableSize, Val: maxHeaderTableSize})886
cc.bw.Write(clientPreface)
887
cc.fr.WriteSettings(initialSettings...)
888
cc.fr.WriteWindowUpdate(0, transportDefaultConnFlow)
889
cc.inflow.init(transportDefaultConnFlow + initialWindowSize)
896
cc.goRun(cc.readLoop)
900
func (cc *ClientConn) healthCheck() {901
pingTimeout := cc.t.pingTimeout()
902
// We don't need to periodically ping in the health check, because the readLoop of ClientConn will
903
// trigger the healthCheck again if there is no frame received.
904
ctx, cancel := cc.contextWithTimeout(context.Background(), pingTimeout)
906
cc.vlogf("http2: Transport sending health check")909
cc.vlogf("http2: Transport health check failure: %v", err)910
cc.closeForLostPing()
912
cc.vlogf("http2: Transport health check success")916
// SetDoNotReuse marks cc as not reusable for future HTTP requests.
917
func (cc *ClientConn) SetDoNotReuse() {923
func (cc *ClientConn) setGoAway(f *GoAwayFrame) {930
// Merge the previous and current GoAway error frames.
931
if cc.goAwayDebug == "" {932
cc.goAwayDebug = string(f.DebugData())
934
if old != nil && old.ErrCode != ErrCodeNo {935
cc.goAway.ErrCode = old.ErrCode
937
last := f.LastStreamID
938
for streamID, cs := range cc.streams {940
cs.abortStreamLocked(errClientConnGotGoAway)
945
// CanTakeNewRequest reports whether the connection can take a new request,
946
// meaning it has not been closed or received or sent a GOAWAY.
948
// If the caller is going to immediately make a new request on this
949
// connection, use ReserveNewRequest instead.
950
func (cc *ClientConn) CanTakeNewRequest() bool {953
return cc.canTakeNewRequestLocked()
956
// ReserveNewRequest is like CanTakeNewRequest but also reserves a
957
// concurrent stream in cc. The reservation is decremented on the
958
// next call to RoundTrip.
959
func (cc *ClientConn) ReserveNewRequest() bool {962
if st := cc.idleStateLocked(); !st.canTakeNewRequest {969
// ClientConnState describes the state of a ClientConn.
970
type ClientConnState struct {971
// Closed is whether the connection is closed.
974
// Closing is whether the connection is in the process of
975
// closing. It may be closing due to shutdown, being a
976
// single-use connection, being marked as DoNotReuse, or
977
// having received a GOAWAY frame.
980
// StreamsActive is how many streams are active.
983
// StreamsReserved is how many streams have been reserved via
984
// ClientConn.ReserveNewRequest.
987
// StreamsPending is how many requests have been sent in excess
988
// of the peer's advertised MaxConcurrentStreams setting and
989
// are waiting for other streams to complete.
992
// MaxConcurrentStreams is how many concurrent streams the
993
// peer advertised as acceptable. Zero means no SETTINGS
994
// frame has been received yet.
995
MaxConcurrentStreams uint32
997
// LastIdle, if non-zero, is when the connection last
998
// transitioned to idle state.
1002
// State returns a snapshot of cc's state.
1003
func (cc *ClientConn) State() ClientConnState {1005
maxConcurrent := cc.maxConcurrentStreams
1006
if !cc.seenSettings {1012
defer cc.mu.Unlock()
1013
return ClientConnState{1015
Closing: cc.closing || cc.singleUse || cc.doNotReuse || cc.goAway != nil,
1016
StreamsActive: len(cc.streams),
1017
StreamsReserved: cc.streamsReserved,
1018
StreamsPending: cc.pendingRequests,
1019
LastIdle: cc.lastIdle,
1020
MaxConcurrentStreams: maxConcurrent,
1024
// clientConnIdleState describes the suitability of a client
1025
// connection to initiate a new RoundTrip request.
1026
type clientConnIdleState struct {1027
canTakeNewRequest bool
1030
func (cc *ClientConn) idleState() clientConnIdleState {1032
defer cc.mu.Unlock()
1033
return cc.idleStateLocked()
1036
func (cc *ClientConn) idleStateLocked() (st clientConnIdleState) {1037
if cc.singleUse && cc.nextStreamID > 1 {1040
var maxConcurrentOkay bool
1041
if cc.t.StrictMaxConcurrentStreams {1042
// We'll tell the caller we can take a new request to
1043
// prevent the caller from dialing a new TCP
1044
// connection, but then we'll block later before
1046
maxConcurrentOkay = true
1048
maxConcurrentOkay = int64(len(cc.streams)+cc.streamsReserved+1) <= int64(cc.maxConcurrentStreams)
1051
st.canTakeNewRequest = cc.goAway == nil && !cc.closed && !cc.closing && maxConcurrentOkay &&
1053
int64(cc.nextStreamID)+2*int64(cc.pendingRequests) < math.MaxInt32 &&
1058
func (cc *ClientConn) canTakeNewRequestLocked() bool {1059
st := cc.idleStateLocked()
1060
return st.canTakeNewRequest
1063
// tooIdleLocked reports whether this connection has been been sitting idle
1064
// for too much wall time.
1065
func (cc *ClientConn) tooIdleLocked() bool {1066
// The Round(0) strips the monontonic clock reading so the
1067
// times are compared based on their wall time. We don't want
1068
// to reuse a connection that's been sitting idle during
1069
// VM/laptop suspend if monotonic time was also frozen.
1070
return cc.idleTimeout != 0 && !cc.lastIdle.IsZero() && time.Since(cc.lastIdle.Round(0)) > cc.idleTimeout
1073
// onIdleTimeout is called from a time.AfterFunc goroutine. It will
1074
// only be called when we're idle, but because we're coming from a new
1075
// goroutine, there could be a new request coming in at the same time,
1076
// so this simply calls the synchronized closeIfIdle to shut down this
1077
// connection. The timer could just call closeIfIdle, but this is more
1079
func (cc *ClientConn) onIdleTimeout() {1083
func (cc *ClientConn) closeConn() {1084
t := time.AfterFunc(250*time.Millisecond, cc.forceCloseConn)
1089
// A tls.Conn.Close can hang for a long time if the peer is unresponsive.
1090
// Try to shut it down more aggressively.
1091
func (cc *ClientConn) forceCloseConn() {1092
tc, ok := cc.tconn.(*tls.Conn)
1096
if nc := tc.NetConn(); nc != nil {1101
func (cc *ClientConn) closeIfIdle() {1103
if len(cc.streams) > 0 || cc.streamsReserved > 0 {1108
nextID := cc.nextStreamID
1109
// TODO: do clients send GOAWAY too? maybe? Just Close:
1113
cc.vlogf("http2: Transport closing idle conn %p (forSingleUse=%v, maxStream=%v)", cc, cc.singleUse, nextID-2)1118
func (cc *ClientConn) isDoNotReuseAndIdle() bool {1120
defer cc.mu.Unlock()
1121
return cc.doNotReuse && len(cc.streams) == 0
1124
var shutdownEnterWaitStateHook = func() {}1126
// Shutdown gracefully closes the client connection, waiting for running streams to complete.
1127
func (cc *ClientConn) Shutdown(ctx context.Context) error {1128
if err := cc.sendGoAway(); err != nil {1131
// Wait for all in-flight streams to complete or connection to close
1132
done := make(chan struct{})1133
cancelled := false // guarded by cc.mu
1136
defer cc.mu.Unlock()
1138
if len(cc.streams) == 0 || cc.closed {1149
shutdownEnterWaitStateHook()
1156
// Free the goroutine above
1164
func (cc *ClientConn) sendGoAway() error {1166
closing := cc.closing
1168
maxStreamID := cc.nextStreamID
1171
// GOAWAY sent already
1176
defer cc.wmu.Unlock()
1177
// Send a graceful shutdown frame to server
1178
if err := cc.fr.WriteGoAway(maxStreamID, ErrCodeNo, nil); err != nil {1181
if err := cc.bw.Flush(); err != nil {1184
// Prevent new requests
1188
// closes the client connection immediately. In-flight requests are interrupted.
1189
// err is sent to streams.
1190
func (cc *ClientConn) closeForError(err error) {1193
for _, cs := range cc.streams {1194
cs.abortStreamLocked(err)
1201
// Close closes the client connection immediately.
1203
// In-flight requests are interrupted. For a graceful shutdown, use Shutdown instead.
1204
func (cc *ClientConn) Close() error {1205
err := errors.New("http2: client connection force closed via ClientConn.Close")1206
cc.closeForError(err)
1210
// closes the client connection immediately. In-flight requests are interrupted.
1211
func (cc *ClientConn) closeForLostPing() {1212
err := errors.New("http2: client connection lost")1213
if f := cc.t.CountError; f != nil {1214
f("conn_close_lost_ping")1216
cc.closeForError(err)
1219
// errRequestCanceled is a copy of net/http's errRequestCanceled because it's not
1220
// exported. At least they'll be DeepEqual for h1-vs-h2 comparisons tests.
1221
var errRequestCanceled = errors.New("net/http: request canceled")1223
func commaSeparatedTrailers(req *http.Request) (string, error) {1224
keys := make([]string, 0, len(req.Trailer))
1225
for k := range req.Trailer {1226
k = canonicalHeader(k)
1228
case "Transfer-Encoding", "Trailer", "Content-Length":
1229
return "", fmt.Errorf("invalid Trailer key %q", k)1231
keys = append(keys, k)
1235
return strings.Join(keys, ","), nil
1240
func (cc *ClientConn) responseHeaderTimeout() time.Duration {1242
return cc.t.t1.ResponseHeaderTimeout
1244
// No way to do this (yet?) with just an http2.Transport. Probably
1245
// no need. Request.Cancel this is the new way. We only need to support
1246
// this for compatibility with the old http.Transport fields when
1247
// we're doing transparent http2.
1251
// checkConnHeaders checks whether req has any invalid connection-level headers.
1252
// per RFC 7540 section 8.1.2.2: Connection-Specific Header Fields.
1253
// Certain headers are special-cased as okay but not transmitted later.
1254
func checkConnHeaders(req *http.Request) error {1255
if v := req.Header.Get("Upgrade"); v != "" {1256
return fmt.Errorf("http2: invalid Upgrade request header: %q", req.Header["Upgrade"])1258
if vv := req.Header["Transfer-Encoding"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && vv[0] != "chunked") {1259
return fmt.Errorf("http2: invalid Transfer-Encoding request header: %q", vv)1261
if vv := req.Header["Connection"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && !asciiEqualFold(vv[0], "close") && !asciiEqualFold(vv[0], "keep-alive")) {1262
return fmt.Errorf("http2: invalid Connection request header: %q", vv)1267
// actualContentLength returns a sanitized version of
1268
// req.ContentLength, where 0 actually means zero (not unknown) and -1
1270
func actualContentLength(req *http.Request) int64 {1271
if req.Body == nil || req.Body == http.NoBody {1274
if req.ContentLength != 0 {1275
return req.ContentLength
1280
func (cc *ClientConn) decrStreamReservations() {1282
defer cc.mu.Unlock()
1283
cc.decrStreamReservationsLocked()
1286
func (cc *ClientConn) decrStreamReservationsLocked() {1287
if cc.streamsReserved > 0 {1288
cc.streamsReserved--
1292
func (cc *ClientConn) RoundTrip(req *http.Request) (*http.Response, error) {1293
return cc.roundTrip(req, nil)
1296
func (cc *ClientConn) roundTrip(req *http.Request, streamf func(*clientStream)) (*http.Response, error) {1297
ctx := req.Context()
1298
cs := &clientStream{1301
reqCancel: req.Cancel,
1302
isHead: req.Method == "HEAD",
1304
reqBodyContentLength: actualContentLength(req),
1305
trace: httptrace.ContextClientTrace(ctx),
1306
peerClosed: make(chan struct{}),1307
abort: make(chan struct{}),1308
respHeaderRecv: make(chan struct{}),1309
donec: make(chan struct{}),1315
waitDone := func() error {1316
if cc.syncHooks != nil {1317
cc.syncHooks.blockUntil(func() bool {1321
case <-cs.reqCancel:
1333
case <-cs.reqCancel:
1334
return errRequestCanceled
1338
handleResponseHeaders := func() (*http.Response, error) {1340
if res.StatusCode > 299 {1341
// On error or status code 3xx, 4xx, 5xx, etc abort any
1342
// ongoing write, assuming that the server doesn't care
1343
// about our request body. If the server replied with 1xx or
1344
// 2xx, however, then assume the server DOES potentially
1345
// want our body (e.g. full-duplex streaming:
1346
// golang.org/issue/13444). If it turns out the server
1347
// doesn't, they'll RST_STREAM us soon enough. This is a
1348
// heuristic to avoid adding knobs to Transport. Hopefully
1350
cs.abortRequestBodyWrite()
1353
res.TLS = cc.tlsState
1354
if res.Body == noBody && actualContentLength(req) == 0 {1355
// If there isn't a request or response body still being
1356
// written, then wait for the stream to be closed before
1357
// RoundTrip returns.
1358
if err := waitDone(); err != nil {1365
cancelRequest := func(cs *clientStream, err error) error {1367
bodyClosed := cs.reqBodyClosed
1369
// Wait for the request body to be closed.
1371
// If nothing closed the body before now, abortStreamLocked
1372
// will have started a goroutine to close it.
1374
// Closing the body before returning avoids a race condition
1375
// with net/http checking its readTrackingBody to see if the
1376
// body was read from or closed. See golang/go#60041.
1378
// The body is closed in a separate goroutine without the
1379
// connection mutex held, but dropping the mutex before waiting
1380
// will keep us from holding it indefinitely if the body
1381
// close is slow for some reason.
1382
if bodyClosed != nil {1393
if cc.syncHooks != nil {1394
cc.syncHooks.blockUntil(func() bool {1396
case <-cs.respHeaderRecv:
1399
case <-cs.reqCancel:
1407
case <-cs.respHeaderRecv:
1408
return handleResponseHeaders()
1411
case <-cs.respHeaderRecv:
1412
// If both cs.respHeaderRecv and cs.abort are signaling,
1413
// pick respHeaderRecv. The server probably wrote the
1414
// response and immediately reset the stream.
1415
// golang.org/issue/49645
1416
return handleResponseHeaders()
1419
return nil, cs.abortErr
1424
return nil, cancelRequest(cs, err)
1425
case <-cs.reqCancel:
1426
cs.abortStream(errRequestCanceled)
1427
return nil, cancelRequest(cs, errRequestCanceled)
1432
// doRequest runs for the duration of the request lifetime.
1434
// It sends the request and performs post-request cleanup (closing Request.Body, etc.).
1435
func (cs *clientStream) doRequest(req *http.Request) {1436
err := cs.writeRequest(req)
1437
cs.cleanupWriteRequest(err)
1440
// writeRequest sends a request.
1442
// It returns nil after the request is written, the response read,
1443
// and the request stream is half-closed by the peer.
1445
// It returns non-nil if the request ends otherwise.
1446
// If the returned error is StreamError, the error Code may be used in resetting the stream.
1447
func (cs *clientStream) writeRequest(req *http.Request) (err error) {1451
if err := checkConnHeaders(req); err != nil {1455
// Acquire the new-request lock by writing to reqHeaderMu.
1456
// This lock guards the critical section covering allocating a new stream ID
1457
// (requires mu) and creating the stream (requires wmu).
1458
if cc.reqHeaderMu == nil {1459
panic("RoundTrip on uninitialized ClientConn") // for tests1461
var newStreamHook func(*clientStream)
1462
if cc.syncHooks != nil {1463
newStreamHook = cc.syncHooks.newstream
1464
cc.syncHooks.blockUntil(func() bool {1466
case cc.reqHeaderMu <- struct{}{}:1468
case <-cs.reqCancel:
1477
case cc.reqHeaderMu <- struct{}{}:1478
case <-cs.reqCancel:
1479
return errRequestCanceled
1485
if cc.idleTimer != nil {1488
cc.decrStreamReservationsLocked()
1489
if err := cc.awaitOpenSlotForStreamLocked(cs); err != nil {1494
cc.addStreamLocked(cs) // assigns stream ID
1495
if isConnectionCloseRequest(req) {1496
cc.doNotReuse = true
1500
if newStreamHook != nil {1504
// TODO(bradfitz): this is a copy of the logic in net/http. Unify somewhere?
1505
if !cc.t.disableCompression() &&
1506
req.Header.Get("Accept-Encoding") == "" &&1507
req.Header.Get("Range") == "" &&1509
// Request gzip only, not deflate. Deflate is ambiguous and
1510
// not as universally supported anyway.
1511
// See: https://zlib.net/zlib_faq.html#faq39
1513
// Note that we don't request this for HEAD requests,
1514
// due to a bug in nginx:
1515
// http://trac.nginx.org/nginx/ticket/358
1516
// https://golang.org/issue/5522
1518
// We don't request gzip if the request is for a range, since
1519
// auto-decoding a portion of a gzipped document will just fail
1520
// anyway. See https://golang.org/issue/8923
1521
cs.requestedGzip = true
1524
continueTimeout := cc.t.expectContinueTimeout()
1525
if continueTimeout != 0 {1526
if !httpguts.HeaderValuesContainsToken(req.Header["Expect"], "100-continue") {1529
cs.on100 = make(chan struct{}, 1)1533
// Past this point (where we send request headers), it is possible for
1534
// RoundTrip to return successfully. Since the RoundTrip contract permits
1535
// the caller to "mutate or reuse" the Request after closing the Response's Body,
1536
// we must take care when referencing the Request from here on.
1537
err = cs.encodeAndWriteHeaders(req)
1543
hasBody := cs.reqBodyContentLength != 0
1545
cs.sentEndStream = true
1547
if continueTimeout != 0 {1548
traceWait100Continue(cs.trace)
1549
timer := time.NewTimer(continueTimeout)
1559
case <-cs.reqCancel:
1560
err = errRequestCanceled
1564
traceWroteRequest(cs.trace, err)
1569
if err = cs.writeRequestBody(req); err != nil {1570
if err != errStopReqBodyWrite {1571
traceWroteRequest(cs.trace, err)
1575
cs.sentEndStream = true
1579
traceWroteRequest(cs.trace, err)
1581
var respHeaderTimer <-chan time.Time
1582
var respHeaderRecv chan struct{}1583
if d := cc.responseHeaderTimeout(); d != 0 {1584
timer := cc.newTimer(d)
1586
respHeaderTimer = timer.C()
1587
respHeaderRecv = cs.respHeaderRecv
1589
// Wait until the peer half-closes its end of the stream,
1590
// or until the request is aborted (via context, error, or otherwise),
1591
// whichever comes first.
1593
if cc.syncHooks != nil {1594
cc.syncHooks.blockUntil(func() bool {1596
case <-cs.peerClosed:
1597
case <-respHeaderTimer:
1598
case <-respHeaderRecv:
1601
case <-cs.reqCancel:
1609
case <-cs.peerClosed:
1611
case <-respHeaderTimer:
1613
case <-respHeaderRecv:
1614
respHeaderRecv = nil
1615
respHeaderTimer = nil // keep waiting for END_STREAM
1620
case <-cs.reqCancel:
1621
return errRequestCanceled
1626
func (cs *clientStream) encodeAndWriteHeaders(req *http.Request) error {1631
defer cc.wmu.Unlock()
1633
// If the request was canceled while waiting for cc.mu, just quit.
1639
case <-cs.reqCancel:
1640
return errRequestCanceled
1646
// we send: HEADERS{1}, CONTINUATION{0,} + DATA{0,} (DATA is1647
// sent by writeRequestBody below, along with any Trailers,
1648
// again in form HEADERS{1}, CONTINUATION{0,})1649
trailers, err := commaSeparatedTrailers(req)
1653
hasTrailers := trailers != ""
1654
contentLen := actualContentLength(req)
1655
hasBody := contentLen != 0
1656
hdrs, err := cc.encodeHeaders(req, cs.requestedGzip, trailers, contentLen)
1661
// Write the request.
1662
endStream := !hasBody && !hasTrailers
1663
cs.sentHeaders = true
1664
err = cc.writeHeaders(cs.ID, endStream, int(cc.maxFrameSize), hdrs)
1665
traceWroteHeaders(cs.trace)
1669
// cleanupWriteRequest performs post-request tasks.
1671
// If err (the result of writeRequest) is non-nil and the stream is not closed,
1672
// cleanupWriteRequest will send a reset to the peer.
1673
func (cs *clientStream) cleanupWriteRequest(err error) {1677
// We were canceled before creating the stream, so return our reservation.
1678
cc.decrStreamReservations()
1681
// TODO: write h12Compare test showing whether
1682
// Request.Body is closed by the Transport,
1683
// and in multiple cases: server replies <=299 and >299
1684
// while still writing request body
1686
mustCloseBody := false
1687
if cs.reqBody != nil && cs.reqBodyClosed == nil {1688
mustCloseBody = true
1689
cs.reqBodyClosed = make(chan struct{})1691
bodyClosed := cs.reqBodyClosed
1697
if bodyClosed != nil {1701
if err != nil && cs.sentEndStream {1702
// If the connection is closed immediately after the response is read,
1703
// we may be aborted before finishing up here. If the stream was closed
1704
// cleanly on both sides, there is no error.
1706
case <-cs.peerClosed:
1712
cs.abortStream(err) // possibly redundant, but harmless
1714
if se, ok := err.(StreamError); ok {1715
if se.Cause != errFromPeer {1716
cc.writeStreamReset(cs.ID, se.Code, err)
1719
cc.writeStreamReset(cs.ID, ErrCodeCancel, err)
1722
cs.bufPipe.CloseWithError(err) // no-op if already closed
1724
if cs.sentHeaders && !cs.sentEndStream {1725
cc.writeStreamReset(cs.ID, ErrCodeNo, nil)
1727
cs.bufPipe.CloseWithError(errRequestCanceled)
1730
cc.forgetStreamID(cs.ID)
1743
// awaitOpenSlotForStreamLocked waits until len(streams) < maxConcurrentStreams.
1745
func (cc *ClientConn) awaitOpenSlotForStreamLocked(cs *clientStream) error {1747
cc.lastActive = time.Now()
1748
if cc.closed || !cc.canTakeNewRequestLocked() {1749
return errClientConnUnusable
1751
cc.lastIdle = time.Time{}1752
if int64(len(cc.streams)) < int64(cc.maxConcurrentStreams) {1755
cc.pendingRequests++
1757
cc.pendingRequests--
1766
// requires cc.wmu be held
1767
func (cc *ClientConn) writeHeaders(streamID uint32, endStream bool, maxFrameSize int, hdrs []byte) error {1768
first := true // first frame written (HEADERS is first, then CONTINUATION)
1769
for len(hdrs) > 0 && cc.werr == nil {1771
if len(chunk) > maxFrameSize {1772
chunk = chunk[:maxFrameSize]
1774
hdrs = hdrs[len(chunk):]
1775
endHeaders := len(hdrs) == 0
1777
cc.fr.WriteHeaders(HeadersFrameParam{1779
BlockFragment: chunk,
1780
EndStream: endStream,
1781
EndHeaders: endHeaders,
1785
cc.fr.WriteContinuation(streamID, endHeaders, chunk)
1792
// internal error values; they don't escape to callers
1794
// abort request body write; don't send cancel
1795
errStopReqBodyWrite = errors.New("http2: aborting request body write")1797
// abort request body write, but send stream reset of cancel.
1798
errStopReqBodyWriteAndCancel = errors.New("http2: canceling request")1800
errReqBodyTooLong = errors.New("http2: request body larger than specified content length")1803
// frameScratchBufferLen returns the length of a buffer to use for
1804
// outgoing request bodies to read/write to/from.
1806
// It returns max(1, min(peer's advertised max frame size,
1807
// Request.ContentLength+1, 512KB)).
1808
func (cs *clientStream) frameScratchBufferLen(maxFrameSize int) int {1809
const max = 512 << 10
1810
n := int64(maxFrameSize)
1814
if cl := cs.reqBodyContentLength; cl != -1 && cl+1 < n {1815
// Add an extra byte past the declared content-length to
1816
// give the caller's Request.Body io.Reader a chance to
1817
// give us more bytes than they declared, so we can catch it
1824
return int(n) // doesn't truncate; max is 512K
1827
// Seven bufPools manage different frame sizes. This helps to avoid scenarios where long-running
1828
// streaming requests using small frame sizes occupy large buffers initially allocated for prior
1829
// requests needing big buffers. The size ranges are as follows:
1830
// {0 KB, 16 KB], {16 KB, 32 KB], {32 KB, 64 KB], {64 KB, 128 KB], {128 KB, 256 KB],1831
// {256 KB, 512 KB], {512 KB, infinity}1832
// In practice, the maximum scratch buffer size should not exceed 512 KB due to
1833
// frameScratchBufferLen(maxFrameSize), thus the "infinity pool" should never be used.
1834
// It exists mainly as a safety measure, for potential future increases in max buffer size.
1835
var bufPools [7]sync.Pool // of *[]byte
1836
func bufPoolIndex(size int) int {1841
bits := bits.Len(uint(size))
1843
if index >= len(bufPools) {1844
return len(bufPools) - 1
1849
func (cs *clientStream) writeRequestBody(req *http.Request) (err error) {1852
sentEnd := false // whether we sent the final DATA frame w/ END_STREAM
1854
hasTrailers := req.Trailer != nil
1855
remainLen := cs.reqBodyContentLength
1856
hasContentLen := remainLen != -1
1859
maxFrameSize := int(cc.maxFrameSize)
1862
// Scratch buffer for reading into & writing from.
1863
scratchLen := cs.frameScratchBufferLen(maxFrameSize)
1865
index := bufPoolIndex(scratchLen)
1866
if bp, ok := bufPools[index].Get().(*[]byte); ok && len(*bp) >= scratchLen {1867
defer bufPools[index].Put(bp)
1870
buf = make([]byte, scratchLen)
1871
defer bufPools[index].Put(&buf)
1876
n, err := body.Read(buf)
1878
remainLen -= int64(n)
1879
if remainLen == 0 && err == nil {1880
// The request body's Content-Length was predeclared and
1881
// we just finished reading it all, but the underlying io.Reader
1882
// returned the final chunk with a nil error (which is one of
1883
// the two valid things a Reader can do at EOF). Because we'd prefer
1884
// to send the END_STREAM bit early, double-check that we're actually
1885
// at EOF. Subsequent reads should return (0, EOF) at this point.
1886
// If either value is different, we return an error in one of two ways below.
1889
n1, err = body.Read(scratch[:])
1890
remainLen -= int64(n1)
1893
err = errReqBodyTooLong
1899
bodyClosed := cs.reqBodyClosed != nil
1903
return errStopReqBodyWrite
1913
for len(remain) > 0 && err == nil {1915
allowed, err = cs.awaitFlowControl(len(remain))
1920
data := remain[:allowed]
1921
remain = remain[allowed:]
1922
sentEnd = sawEOF && len(remain) == 0 && !hasTrailers
1923
err = cc.fr.WriteData(cs.ID, sentEnd, data)
1925
// TODO(bradfitz): this flush is for latency, not bandwidth.
1926
// Most requests won't need this. Make this opt-in or
1927
// opt-out? Use some heuristic on the body type? Nagel-like
1928
// timers? Based on 'n'? Only last chunk of this for loop,
1929
// unless flow control tokens are low? For now, always.
1930
// If we change this, see comment below.
1941
// Already sent END_STREAM (which implies we have no
1942
// trailers) and flushed, because currently all
1943
// WriteData frames above get a flush. So we're done.
1947
// Since the RoundTrip contract permits the caller to "mutate or reuse"
1948
// a request after the Response's Body is closed, verify that this hasn't
1949
// happened before accessing the trailers.
1951
trailer := req.Trailer
1959
defer cc.wmu.Unlock()
1961
if len(trailer) > 0 {1962
trls, err = cc.encodeTrailers(trailer)
1968
// Two ways to send END_STREAM: either with trailers, or
1969
// with an empty DATA frame.
1971
err = cc.writeHeaders(cs.ID, true, maxFrameSize, trls)
1973
err = cc.fr.WriteData(cs.ID, true, nil)
1975
if ferr := cc.bw.Flush(); ferr != nil && err == nil {1981
// awaitFlowControl waits for [1, min(maxBytes, cc.cs.maxFrameSize)] flow
1982
// control tokens from the server.
1983
// It returns either the non-zero number of tokens taken or an error
1984
// if the stream is dead.
1985
func (cs *clientStream) awaitFlowControl(maxBytes int) (taken int32, err error) {1989
defer cc.mu.Unlock()
1992
return 0, errClientConnClosed
1994
if cs.reqBodyClosed != nil {1995
return 0, errStopReqBodyWrite
1999
return 0, cs.abortErr
2002
case <-cs.reqCancel:
2003
return 0, errRequestCanceled
2006
if a := cs.flow.available(); a > 0 {2008
if int(take) > maxBytes {2010
take = int32(maxBytes) // can't truncate int; take is int32
2012
if take > int32(cc.maxFrameSize) {2013
take = int32(cc.maxFrameSize)
2022
func validateHeaders(hdrs http.Header) string {2023
for k, vv := range hdrs {2024
if !httpguts.ValidHeaderFieldName(k) {2025
return fmt.Sprintf("name %q", k)2027
for _, v := range vv {2028
if !httpguts.ValidHeaderFieldValue(v) {2029
// Don't include the value in the error,
2030
// because it may be sensitive.
2031
return fmt.Sprintf("value for header %q", k)2038
var errNilRequestURL = errors.New("http2: Request.URI is nil")2040
// requires cc.wmu be held.
2041
func (cc *ClientConn) encodeHeaders(req *http.Request, addGzipHeader bool, trailers string, contentLength int64) ([]byte, error) {2044
return nil, errNilRequestURL
2051
host, err := httpguts.PunycodeHostPort(host)
2055
if !httpguts.ValidHostHeader(host) {2056
return nil, errors.New("http2: invalid Host header")2060
if req.Method != "CONNECT" {2061
path = req.URL.RequestURI()
2062
if !validPseudoPath(path) {2064
path = strings.TrimPrefix(path, req.URL.Scheme+"://"+host)
2065
if !validPseudoPath(path) {2066
if req.URL.Opaque != "" {2067
return nil, fmt.Errorf("invalid request :path %q from URL.Opaque = %q", orig, req.URL.Opaque)2069
return nil, fmt.Errorf("invalid request :path %q", orig)2075
// Check for any invalid headers+trailers and return an error before we
2076
// potentially pollute our hpack state. (We want to be able to
2077
// continue to reuse the hpack encoder for future requests)
2078
if err := validateHeaders(req.Header); err != "" {2079
return nil, fmt.Errorf("invalid HTTP header %s", err)2081
if err := validateHeaders(req.Trailer); err != "" {2082
return nil, fmt.Errorf("invalid HTTP trailer %s", err)2085
enumerateHeaders := func(f func(name, value string)) {2086
// 8.1.2.3 Request Pseudo-Header Fields
2087
// The :path pseudo-header field includes the path and query parts of the
2088
// target URI (the path-absolute production and optionally a '?' character
2089
// followed by the query production, see Sections 3.3 and 3.4 of
2091
f(":authority", host)2097
if req.Method != "CONNECT" {2099
f(":scheme", req.URL.Scheme)2102
f("trailer", trailers)2106
for k, vv := range req.Header {2107
if asciiEqualFold(k, "host") || asciiEqualFold(k, "content-length") {2108
// Host is :authority, already sent.
2109
// Content-Length is automatic, set below.
2111
} else if asciiEqualFold(k, "connection") ||
2112
asciiEqualFold(k, "proxy-connection") ||
2113
asciiEqualFold(k, "transfer-encoding") ||
2114
asciiEqualFold(k, "upgrade") ||
2115
asciiEqualFold(k, "keep-alive") {2116
// Per 8.1.2.2 Connection-Specific Header
2117
// Fields, don't send connection-specific
2118
// fields. We have already checked if any
2119
// are error-worthy so just ignore the rest.
2121
} else if asciiEqualFold(k, "user-agent") {2122
// Match Go's http1 behavior: at most one
2123
// User-Agent. If set to nil or empty string,
2124
// then omit it. Otherwise if not mentioned,
2125
// include the default (below).
2134
} else if asciiEqualFold(k, "cookie") {2135
// Per 8.1.2.5 To allow for better compression efficiency, the
2136
// Cookie header field MAY be split into separate header fields,
2137
// each with one or more cookie-pairs.
2138
for _, v := range vv {2140
p := strings.IndexByte(v, ';')
2146
// strip space after semicolon if any.
2147
for p+1 <= len(v) && v[p] == ' ' {2159
for _, v := range vv {2163
if shouldSendReqContentLength(req.Method, contentLength) {2164
f("content-length", strconv.FormatInt(contentLength, 10))2167
f("accept-encoding", "gzip")2170
f("user-agent", defaultUserAgent)2174
// Do a first pass over the headers counting bytes to ensure
2175
// we don't exceed cc.peerMaxHeaderListSize. This is done as a
2176
// separate pass before encoding the headers to prevent
2177
// modifying the hpack state.
2179
enumerateHeaders(func(name, value string) {2180
hf := hpack.HeaderField{Name: name, Value: value}2181
hlSize += uint64(hf.Size())
2184
if hlSize > cc.peerMaxHeaderListSize {2185
return nil, errRequestHeaderListSize
2188
trace := httptrace.ContextClientTrace(req.Context())
2189
traceHeaders := traceHasWroteHeaderField(trace)
2191
// Header list size is ok. Write the headers.
2192
enumerateHeaders(func(name, value string) {2193
name, ascii := lowerHeader(name)
2195
// Skip writing invalid headers. Per RFC 7540, Section 8.1.2, header
2196
// field names have to be ASCII characters (just as in HTTP/1.x).
2199
cc.writeHeader(name, value)
2201
traceWroteHeaderField(trace, name, value)
2205
return cc.hbuf.Bytes(), nil
2208
// shouldSendReqContentLength reports whether the http2.Transport should send
2209
// a "content-length" request header. This logic is basically a copy of the net/http
2210
// transferWriter.shouldSendContentLength.
2211
// The contentLength is the corrected contentLength (so 0 means actually 0, not unknown).
2213
func shouldSendReqContentLength(method string, contentLength int64) bool {2214
if contentLength > 0 {2217
if contentLength < 0 {2220
// For zero bodies, whether we send a content-length depends on the method.
2221
// It also kinda doesn't matter for http2 either way, with END_STREAM.
2223
case "POST", "PUT", "PATCH":
2230
// requires cc.wmu be held.
2231
func (cc *ClientConn) encodeTrailers(trailer http.Header) ([]byte, error) {2235
for k, vv := range trailer {2236
for _, v := range vv {2237
hf := hpack.HeaderField{Name: k, Value: v}2238
hlSize += uint64(hf.Size())
2241
if hlSize > cc.peerMaxHeaderListSize {2242
return nil, errRequestHeaderListSize
2245
for k, vv := range trailer {2246
lowKey, ascii := lowerHeader(k)
2248
// Skip writing invalid headers. Per RFC 7540, Section 8.1.2, header
2249
// field names have to be ASCII characters (just as in HTTP/1.x).
2252
// Transfer-Encoding, etc.. have already been filtered at the
2253
// start of RoundTrip
2254
for _, v := range vv {2255
cc.writeHeader(lowKey, v)
2258
return cc.hbuf.Bytes(), nil
2261
func (cc *ClientConn) writeHeader(name, value string) {2263
log.Printf("http2: Transport encoding header %q = %q", name, value)2265
cc.henc.WriteField(hpack.HeaderField{Name: name, Value: value})2268
type resAndError struct {2274
// requires cc.mu be held.
2275
func (cc *ClientConn) addStreamLocked(cs *clientStream) {2276
cs.flow.add(int32(cc.initialWindowSize))
2277
cs.flow.setConnFlow(&cc.flow)
2278
cs.inflow.init(transportDefaultStreamFlow)
2279
cs.ID = cc.nextStreamID
2280
cc.nextStreamID += 2
2281
cc.streams[cs.ID] = cs
2283
panic("assigned stream ID 0")2287
func (cc *ClientConn) forgetStreamID(id uint32) {2289
slen := len(cc.streams)
2290
delete(cc.streams, id)
2291
if len(cc.streams) != slen-1 {2292
panic("forgetting unknown stream id")2294
cc.lastActive = time.Now()
2295
if len(cc.streams) == 0 && cc.idleTimer != nil {2296
cc.idleTimer.Reset(cc.idleTimeout)
2297
cc.lastIdle = time.Now()
2299
// Wake up writeRequestBody via clientStream.awaitFlowControl and
2300
// wake up RoundTrip if there is a pending request.
2303
closeOnIdle := cc.singleUse || cc.doNotReuse || cc.t.disableKeepAlives() || cc.goAway != nil
2304
if closeOnIdle && cc.streamsReserved == 0 && len(cc.streams) == 0 {2306
cc.vlogf("http2: Transport closing idle conn %p (forSingleUse=%v, maxStream=%v)", cc, cc.singleUse, cc.nextStreamID-2)2309
defer cc.closeConn()
2315
// clientConnReadLoop is the state owned by the clientConn's frame-reading readLoop.
2316
type clientConnReadLoop struct {2321
// readLoop runs in its own goroutine and reads and dispatches frames.
2322
func (cc *ClientConn) readLoop() {2323
rl := &clientConnReadLoop{cc: cc}2325
cc.readerErr = rl.run()
2326
if ce, ok := cc.readerErr.(ConnectionError); ok {2328
cc.fr.WriteGoAway(0, ErrCode(ce), nil)
2333
// GoAwayError is returned by the Transport when the server closes the
2334
// TCP connection after sending a GOAWAY frame.
2335
type GoAwayError struct {2341
func (e GoAwayError) Error() string {2342
return fmt.Sprintf("http2: server sent GOAWAY and closed the connection; LastStreamID=%v, ErrCode=%v, debug=%q",2343
e.LastStreamID, e.ErrCode, e.DebugData)
2346
func isEOFOrNetReadError(err error) bool {2350
ne, ok := err.(*net.OpError)
2351
return ok && ne.Op == "read"
2354
func (rl *clientConnReadLoop) cleanup() {2356
cc.t.connPool().MarkDead(cc)
2357
defer cc.closeConn()
2358
defer close(cc.readerDone)
2360
if cc.idleTimer != nil {2364
// Close any response bodies if the server closes prematurely.
2365
// TODO: also do this if we've written the headers but not
2366
// gotten a response yet.
2369
if cc.goAway != nil && isEOFOrNetReadError(err) {2371
LastStreamID: cc.goAway.LastStreamID,
2372
ErrCode: cc.goAway.ErrCode,
2373
DebugData: cc.goAwayDebug,
2375
} else if err == io.EOF {2376
err = io.ErrUnexpectedEOF
2380
for _, cs := range cc.streams {2382
case <-cs.peerClosed:
2383
// The server closed the stream before closing the conn,
2384
// so no need to interrupt it.
2386
cs.abortStreamLocked(err)
2393
// countReadFrameError calls Transport.CountError with a string
2395
func (cc *ClientConn) countReadFrameError(err error) {2396
f := cc.t.CountError
2397
if f == nil || err == nil {2400
if ce, ok := err.(ConnectionError); ok {2401
errCode := ErrCode(ce)
2402
f(fmt.Sprintf("read_frame_conn_error_%s", errCode.stringToken()))2405
if errors.Is(err, io.EOF) {2409
if errors.Is(err, io.ErrUnexpectedEOF) {2410
f("read_frame_unexpected_eof")2413
if errors.Is(err, ErrFrameTooLarge) {2414
f("read_frame_too_large")2417
f("read_frame_other")2420
func (rl *clientConnReadLoop) run() error {2422
gotSettings := false
2423
readIdleTimeout := cc.t.ReadIdleTimeout
2425
if readIdleTimeout != 0 {2426
t = cc.afterFunc(readIdleTimeout, cc.healthCheck)
2429
f, err := cc.fr.ReadFrame()
2431
t.Reset(readIdleTimeout)
2434
cc.vlogf("http2: Transport readFrame error on conn %p: (%T) %v", cc, err, err)2436
if se, ok := err.(StreamError); ok {2437
if cs := rl.streamByID(se.StreamID); cs != nil {2438
if se.Cause == nil {2439
se.Cause = cc.fr.errDetail
2441
rl.endStreamError(cs, se)
2444
} else if err != nil {2445
cc.countReadFrameError(err)
2449
cc.vlogf("http2: Transport received %s", summarizeFrame(f))2452
if _, ok := f.(*SettingsFrame); !ok {2453
cc.logf("protocol error: received %T before a SETTINGS frame", f)2454
return ConnectionError(ErrCodeProtocol)
2459
switch f := f.(type) {2460
case *MetaHeadersFrame:
2461
err = rl.processHeaders(f)
2463
err = rl.processData(f)
2465
err = rl.processGoAway(f)
2466
case *RSTStreamFrame:
2467
err = rl.processResetStream(f)
2468
case *SettingsFrame:
2469
err = rl.processSettings(f)
2470
case *PushPromiseFrame:
2471
err = rl.processPushPromise(f)
2472
case *WindowUpdateFrame:
2473
err = rl.processWindowUpdate(f)
2475
err = rl.processPing(f)
2477
cc.logf("Transport: unhandled response frame type %T", f)2481
cc.vlogf("http2: Transport conn %p received error from processing frame %v: %v", cc, summarizeFrame(f), err)2488
func (rl *clientConnReadLoop) processHeaders(f *MetaHeadersFrame) error {2489
cs := rl.streamByID(f.StreamID)
2491
// We'd get here if we canceled a request while the
2492
// server had its response still in flight. So if this
2493
// was just something we canceled, ignore it.
2497
rl.endStreamError(cs, StreamError{2498
StreamID: f.StreamID,
2499
Code: ErrCodeProtocol,
2500
Cause: errors.New("protocol error: headers after END_STREAM"),2505
if cs.trace != nil {2506
// TODO(bradfitz): move first response byte earlier,
2507
// when we first read the 9 byte header, not waiting
2508
// until all the HEADERS+CONTINUATION frames have been
2509
// merged. This works for now.
2510
traceFirstResponseByte(cs.trace)
2514
if !cs.pastHeaders {2515
cs.pastHeaders = true
2517
return rl.processTrailers(cs, f)
2520
res, err := rl.handleResponse(cs, f)
2522
if _, ok := err.(ConnectionError); ok {2525
// Any other error type is a stream error.
2526
rl.endStreamError(cs, StreamError{2527
StreamID: f.StreamID,
2528
Code: ErrCodeProtocol,
2531
return nil // return nil from process* funcs to keep conn alive
2534
// (nil, nil) special case. See handleResponse docs.
2537
cs.resTrailer = &res.Trailer
2539
close(cs.respHeaderRecv)
2540
if f.StreamEnded() {2546
// may return error types nil, or ConnectionError. Any other error value
2547
// is a StreamError of type ErrCodeProtocol. The returned error in that case
2550
// As a special case, handleResponse may return (nil, nil) to skip the
2551
// frame (currently only used for 1xx responses).
2552
func (rl *clientConnReadLoop) handleResponse(cs *clientStream, f *MetaHeadersFrame) (*http.Response, error) {2554
return nil, errResponseHeaderListSize
2557
status := f.PseudoValue("status")2559
return nil, errors.New("malformed response from server: missing status pseudo header")2561
statusCode, err := strconv.Atoi(status)
2563
return nil, errors.New("malformed response from server: malformed non-numeric status pseudo header")2566
regularFields := f.RegularFields()
2567
strs := make([]string, len(regularFields))
2568
header := make(http.Header, len(regularFields))
2569
res := &http.Response{2573
StatusCode: statusCode,
2574
Status: status + " " + http.StatusText(statusCode),
2576
for _, hf := range regularFields {2577
key := canonicalHeader(hf.Name)
2578
if key == "Trailer" {2581
t = make(http.Header)
2584
foreachHeaderElement(hf.Value, func(v string) {2585
t[canonicalHeader(v)] = nil
2589
if vv == nil && len(strs) > 0 {2590
// More than likely this will be a single-element key.
2591
// Most headers aren't multi-valued.
2592
// Set the capacity on strs[0] to 1, so any future append
2593
// won't extend the slice into the other strings.
2594
vv, strs = strs[:1:1], strs[1:]
2598
header[key] = append(vv, hf.Value)
2603
if statusCode >= 100 && statusCode <= 199 {2604
if f.StreamEnded() {2605
return nil, errors.New("1xx informational response with END_STREAM flag")2608
const max1xxResponses = 5 // arbitrary bound on number of informational responses, same as net/http
2609
if cs.num1xx > max1xxResponses {2610
return nil, errors.New("http2: too many 1xx informational responses")2612
if fn := cs.get1xxTraceFunc(); fn != nil {2613
if err := fn(statusCode, textproto.MIMEHeader(header)); err != nil {2617
if statusCode == 100 {2618
traceGot100Continue(cs.trace)
2620
case cs.on100 <- struct{}{}:2624
cs.pastHeaders = false // do it all again
2628
res.ContentLength = -1
2629
if clens := res.Header["Content-Length"]; len(clens) == 1 {2630
if cl, err := strconv.ParseUint(clens[0], 10, 63); err == nil {2631
res.ContentLength = int64(cl)
2633
// TODO: care? unlike http/1, it won't mess up our framing, so it's
2634
// more safe smuggling-wise to ignore.
2636
} else if len(clens) > 1 {2637
// TODO: care? unlike http/1, it won't mess up our framing, so it's
2638
// more safe smuggling-wise to ignore.
2639
} else if f.StreamEnded() && !cs.isHead {2640
res.ContentLength = 0
2648
if f.StreamEnded() {2649
if res.ContentLength > 0 {2650
res.Body = missingBody{}2657
cs.bufPipe.setBuffer(&dataBuffer{expected: res.ContentLength})2658
cs.bytesRemain = res.ContentLength
2659
res.Body = transportResponseBody{cs}2661
if cs.requestedGzip && asciiEqualFold(res.Header.Get("Content-Encoding"), "gzip") {2662
res.Header.Del("Content-Encoding")2663
res.Header.Del("Content-Length")2664
res.ContentLength = -1
2665
res.Body = &gzipReader{body: res.Body}2666
res.Uncompressed = true
2671
func (rl *clientConnReadLoop) processTrailers(cs *clientStream, f *MetaHeadersFrame) error {2672
if cs.pastTrailers {2673
// Too many HEADERS frames for this stream.
2674
return ConnectionError(ErrCodeProtocol)
2676
cs.pastTrailers = true
2677
if !f.StreamEnded() {2678
// We expect that any headers for trailers also
2680
return ConnectionError(ErrCodeProtocol)
2682
if len(f.PseudoFields()) > 0 {2683
// No pseudo header fields are defined for trailers.
2684
// TODO: ConnectionError might be overly harsh? Check.
2685
return ConnectionError(ErrCodeProtocol)
2688
trailer := make(http.Header)
2689
for _, hf := range f.RegularFields() {2690
key := canonicalHeader(hf.Name)
2691
trailer[key] = append(trailer[key], hf.Value)
2693
cs.trailer = trailer
2699
// transportResponseBody is the concrete type of Transport.RoundTrip's
2700
// Response.Body. It is an io.ReadCloser.
2701
type transportResponseBody struct {2705
func (b transportResponseBody) Read(p []byte) (n int, err error) {2709
if cs.readErr != nil {2710
return 0, cs.readErr
2712
n, err = b.cs.bufPipe.Read(p)
2713
if cs.bytesRemain != -1 {2714
if int64(n) > cs.bytesRemain {2715
n = int(cs.bytesRemain)
2717
err = errors.New("net/http: server replied with more than declared Content-Length; truncated")2721
return int(cs.bytesRemain), err
2723
cs.bytesRemain -= int64(n)
2724
if err == io.EOF && cs.bytesRemain > 0 {2725
err = io.ErrUnexpectedEOF
2731
// No flow control tokens to send back.
2736
connAdd := cc.inflow.add(n)
2738
if err == nil { // No need to refresh if the stream is over or failed.2739
streamAdd = cs.inflow.add(n)
2743
if connAdd != 0 || streamAdd != 0 {2745
defer cc.wmu.Unlock()
2747
cc.fr.WriteWindowUpdate(0, mustUint31(connAdd))
2750
cc.fr.WriteWindowUpdate(cs.ID, mustUint31(streamAdd))
2757
var errClosedResponseBody = errors.New("http2: response body closed")2759
func (b transportResponseBody) Close() error {2763
cs.bufPipe.BreakWithError(errClosedResponseBody)
2764
cs.abortStream(errClosedResponseBody)
2766
unread := cs.bufPipe.Len()
2769
// Return connection-level flow control.
2770
connAdd := cc.inflow.add(unread)
2773
// TODO(dneil): Acquiring this mutex can block indefinitely.
2774
// Move flow control return to a goroutine?
2776
// Return connection-level flow control.
2778
cc.fr.WriteWindowUpdate(0, uint32(connAdd))
2786
case <-cs.ctx.Done():
2787
// See golang/go#49366: The net/http package can cancel the
2788
// request context after the response body is fully read.
2789
// Don't treat this as an error.
2791
case <-cs.reqCancel:
2792
return errRequestCanceled
2797
func (rl *clientConnReadLoop) processData(f *DataFrame) error {2799
cs := rl.streamByID(f.StreamID)
2803
neverSent := cc.nextStreamID
2805
if f.StreamID >= neverSent {2806
// We never asked for this.
2807
cc.logf("http2: Transport received unsolicited DATA frame; closing connection")2808
return ConnectionError(ErrCodeProtocol)
2810
// We probably did ask for this, but canceled. Just ignore it.
2811
// TODO: be stricter here? only silently ignore things which
2812
// we canceled, but not things which were closed normally
2813
// by the peer? Tough without accumulating too much state.
2815
// But at least return their flow control:
2818
ok := cc.inflow.take(f.Length)
2819
connAdd := cc.inflow.add(int(f.Length))
2822
return ConnectionError(ErrCodeFlowControl)
2826
cc.fr.WriteWindowUpdate(0, uint32(connAdd))
2834
cc.logf("protocol error: received DATA after END_STREAM")2835
rl.endStreamError(cs, StreamError{2836
StreamID: f.StreamID,
2837
Code: ErrCodeProtocol,
2841
if !cs.pastHeaders {2842
cc.logf("protocol error: received DATA before a HEADERS frame")2843
rl.endStreamError(cs, StreamError{2844
StreamID: f.StreamID,
2845
Code: ErrCodeProtocol,
2850
if cs.isHead && len(data) > 0 {2851
cc.logf("protocol error: received DATA on a HEAD request")2852
rl.endStreamError(cs, StreamError{2853
StreamID: f.StreamID,
2854
Code: ErrCodeProtocol,
2858
// Check connection-level flow control.
2860
if !takeInflows(&cc.inflow, &cs.inflow, f.Length) {2862
return ConnectionError(ErrCodeFlowControl)
2864
// Return any padded flow control now, since we won't
2865
// refund it later on body reads.
2867
if pad := int(f.Length) - len(data); pad > 0 {2874
if _, err = cs.bufPipe.Write(data); err != nil {2875
// Return len(data) now if the stream is already closed,
2876
// since data will never be read.
2882
sendConn := cc.inflow.add(refund)
2883
var sendStream int32
2885
sendStream = cs.inflow.add(refund)
2889
if sendConn > 0 || sendStream > 0 {2892
cc.fr.WriteWindowUpdate(0, uint32(sendConn))
2895
cc.fr.WriteWindowUpdate(cs.ID, uint32(sendStream))
2902
rl.endStreamError(cs, err)
2907
if f.StreamEnded() {2913
func (rl *clientConnReadLoop) endStream(cs *clientStream) {2914
// TODO: check that any declared content-length matches, like
2915
// server.go's (*stream).endStream method.
2917
cs.readClosed = true
2918
// Close cs.bufPipe and cs.peerClosed with cc.mu held to avoid a
2919
// race condition: The caller can read io.EOF from Response.Body
2920
// and close the body before we close cs.peerClosed, causing
2921
// cleanupWriteRequest to send a RST_STREAM.
2923
defer rl.cc.mu.Unlock()
2924
cs.bufPipe.closeWithErrorAndCode(io.EOF, cs.copyTrailers)
2925
close(cs.peerClosed)
2929
func (rl *clientConnReadLoop) endStreamError(cs *clientStream, err error) {2930
cs.readAborted = true
2934
func (rl *clientConnReadLoop) streamByID(id uint32) *clientStream {2936
defer rl.cc.mu.Unlock()
2937
cs := rl.cc.streams[id]
2938
if cs != nil && !cs.readAborted {2944
func (cs *clientStream) copyTrailers() {2945
for k, vv := range cs.trailer {2948
*t = make(http.Header)
2954
func (rl *clientConnReadLoop) processGoAway(f *GoAwayFrame) error {2956
cc.t.connPool().MarkDead(cc)
2958
// TODO: deal with GOAWAY more. particularly the error code
2959
cc.vlogf("transport got GOAWAY with error code = %v", f.ErrCode)2960
if fn := cc.t.CountError; fn != nil {2961
fn("recv_goaway_" + f.ErrCode.stringToken())2968
func (rl *clientConnReadLoop) processSettings(f *SettingsFrame) error {2970
// Locking both mu and wmu here allows frame encoding to read settings with only wmu held.
2971
// Acquiring wmu when f.IsAck() is unnecessary, but convenient and mostly harmless.
2973
defer cc.wmu.Unlock()
2975
if err := rl.processSettingsNoWrite(f); err != nil {2979
cc.fr.WriteSettingsAck()
2985
func (rl *clientConnReadLoop) processSettingsNoWrite(f *SettingsFrame) error {2988
defer cc.mu.Unlock()
2991
if cc.wantSettingsAck {2992
cc.wantSettingsAck = false
2995
return ConnectionError(ErrCodeProtocol)
2998
var seenMaxConcurrentStreams bool
2999
err := f.ForeachSetting(func(s Setting) error {3001
case SettingMaxFrameSize:
3002
cc.maxFrameSize = s.Val
3003
case SettingMaxConcurrentStreams:
3004
cc.maxConcurrentStreams = s.Val
3005
seenMaxConcurrentStreams = true
3006
case SettingMaxHeaderListSize:
3007
cc.peerMaxHeaderListSize = uint64(s.Val)
3008
case SettingInitialWindowSize:
3009
// Values above the maximum flow-control
3010
// window size of 2^31-1 MUST be treated as a
3011
// connection error (Section 5.4.1) of type
3012
// FLOW_CONTROL_ERROR.
3013
if s.Val > math.MaxInt32 {3014
return ConnectionError(ErrCodeFlowControl)
3017
// Adjust flow control of currently-open
3018
// frames by the difference of the old initial
3019
// window size and this one.
3020
delta := int32(s.Val) - int32(cc.initialWindowSize)
3021
for _, cs := range cc.streams {3026
cc.initialWindowSize = s.Val
3027
case SettingHeaderTableSize:
3028
cc.henc.SetMaxDynamicTableSize(s.Val)
3029
cc.peerMaxHeaderTableSize = s.Val
3031
cc.vlogf("Unhandled Setting: %v", s)3039
if !cc.seenSettings {3040
if !seenMaxConcurrentStreams {3041
// This was the servers initial SETTINGS frame and it
3042
// didn't contain a MAX_CONCURRENT_STREAMS field so
3043
// increase the number of concurrent streams this
3044
// connection can establish to our default.
3045
cc.maxConcurrentStreams = defaultMaxConcurrentStreams
3047
cc.seenSettings = true
3053
func (rl *clientConnReadLoop) processWindowUpdate(f *WindowUpdateFrame) error {3055
cs := rl.streamByID(f.StreamID)
3056
if f.StreamID != 0 && cs == nil {3061
defer cc.mu.Unlock()
3067
if !fl.add(int32(f.Increment)) {3068
// For stream, the sender sends RST_STREAM with an error code of FLOW_CONTROL_ERROR
3070
rl.endStreamError(cs, StreamError{3071
StreamID: f.StreamID,
3072
Code: ErrCodeFlowControl,
3077
return ConnectionError(ErrCodeFlowControl)
3083
func (rl *clientConnReadLoop) processResetStream(f *RSTStreamFrame) error {3084
cs := rl.streamByID(f.StreamID)
3086
// TODO: return error if server tries to RST_STREAM an idle stream
3089
serr := streamError(cs.ID, f.ErrCode)
3090
serr.Cause = errFromPeer
3091
if f.ErrCode == ErrCodeProtocol {3092
rl.cc.SetDoNotReuse()
3094
if fn := cs.cc.t.CountError; fn != nil {3095
fn("recv_rststream_" + f.ErrCode.stringToken())3097
cs.abortStream(serr)
3099
cs.bufPipe.CloseWithError(serr)
3103
// Ping sends a PING frame to the server and waits for the ack.
3104
func (cc *ClientConn) Ping(ctx context.Context) error {3105
c := make(chan struct{})3106
// Generate a random payload
3109
if _, err := rand.Read(p[:]); err != nil {3113
// check for dup before insert
3114
if _, found := cc.pings[p]; !found {3122
errc := make(chan struct{})3125
defer cc.wmu.Unlock()
3126
if pingError = cc.fr.WritePing(false, p); pingError != nil {3130
if pingError = cc.bw.Flush(); pingError != nil {3135
if cc.syncHooks != nil {3136
cc.syncHooks.blockUntil(func() bool {3141
case <-cc.readerDone:
3155
case <-cc.readerDone:
3156
// connection closed
3161
func (rl *clientConnReadLoop) processPing(f *PingFrame) error {3165
defer cc.mu.Unlock()
3166
// If ack, notify listener if any
3167
if c, ok := cc.pings[f.Data]; ok {3169
delete(cc.pings, f.Data)
3175
defer cc.wmu.Unlock()
3176
if err := cc.fr.WritePing(true, f.Data); err != nil {3179
return cc.bw.Flush()
3182
func (rl *clientConnReadLoop) processPushPromise(f *PushPromiseFrame) error {3183
// We told the peer we don't want them.
3185
// "PUSH_PROMISE MUST NOT be sent if the SETTINGS_ENABLE_PUSH
3186
// setting of the peer endpoint is set to 0. An endpoint that
3187
// has set this setting and has received acknowledgement MUST
3188
// treat the receipt of a PUSH_PROMISE frame as a connection
3189
// error (Section 5.4.1) of type PROTOCOL_ERROR."
3190
return ConnectionError(ErrCodeProtocol)
3193
func (cc *ClientConn) writeStreamReset(streamID uint32, code ErrCode, err error) {3194
// TODO: map err to more interesting error codes, once the
3195
// HTTP community comes up with some. But currently for
3196
// RST_STREAM there's no equivalent to GOAWAY frame's debug
3197
// data, and the error codes are all pretty vague ("cancel").3199
cc.fr.WriteRSTStream(streamID, code)
3205
errResponseHeaderListSize = errors.New("http2: response header list larger than advertised limit")3206
errRequestHeaderListSize = errors.New("http2: request header list larger than peer's advertised limit")3209
func (cc *ClientConn) logf(format string, args ...interface{}) {3210
cc.t.logf(format, args...)
3213
func (cc *ClientConn) vlogf(format string, args ...interface{}) {3214
cc.t.vlogf(format, args...)
3217
func (t *Transport) vlogf(format string, args ...interface{}) {3219
t.logf(format, args...)
3223
func (t *Transport) logf(format string, args ...interface{}) {3224
log.Printf(format, args...)
3227
var noBody io.ReadCloser = noBodyReader{}3229
type noBodyReader struct{}3231
func (noBodyReader) Close() error { return nil }3232
func (noBodyReader) Read([]byte) (int, error) { return 0, io.EOF }3234
type missingBody struct{}3236
func (missingBody) Close() error { return nil }3237
func (missingBody) Read([]byte) (int, error) { return 0, io.ErrUnexpectedEOF }3239
func strSliceContains(ss []string, s string) bool {3240
for _, v := range ss {3248
type erringRoundTripper struct{ err error }3250
func (rt erringRoundTripper) RoundTripErr() error { return rt.err }3251
func (rt erringRoundTripper) RoundTrip(*http.Request) (*http.Response, error) { return nil, rt.err }3253
// gzipReader wraps a response body so it can lazily
3254
// call gzip.NewReader on the first call to Read
3255
type gzipReader struct {3257
body io.ReadCloser // underlying Response.Body
3258
zr *gzip.Reader // lazily-initialized gzip reader
3259
zerr error // sticky error
3262
func (gz *gzipReader) Read(p []byte) (n int, err error) {3267
gz.zr, err = gzip.NewReader(gz.body)
3273
return gz.zr.Read(p)
3276
func (gz *gzipReader) Close() error {3277
if err := gz.body.Close(); err != nil {3280
gz.zerr = fs.ErrClosed
3284
type errorReader struct{ err error }3286
func (r errorReader) Read(p []byte) (int, error) { return 0, r.err }3288
// isConnectionCloseRequest reports whether req should use its own
3289
// connection for a single request and then close the connection.
3290
func isConnectionCloseRequest(req *http.Request) bool {3291
return req.Close || httpguts.HeaderValuesContainsToken(req.Header["Connection"], "close")
3294
// registerHTTPSProtocol calls Transport.RegisterProtocol but
3295
// converting panics into errors.
3296
func registerHTTPSProtocol(t *http.Transport, rt noDialH2RoundTripper) (err error) {3298
if e := recover(); e != nil {3299
err = fmt.Errorf("%v", e)3302
t.RegisterProtocol("https", rt)3306
// noDialH2RoundTripper is a RoundTripper which only tries to complete the request
3307
// if there's already has a cached connection to the host.
3308
// (The field is exported so it can be accessed via reflect from net/http; tested
3309
// by TestNoDialH2RoundTripperType)
3310
type noDialH2RoundTripper struct{ *Transport }3312
func (rt noDialH2RoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {3313
res, err := rt.Transport.RoundTrip(req)
3314
if isNoCachedConnError(err) {3315
return nil, http.ErrSkipAltProtocol
3320
func (t *Transport) idleConnTimeout() time.Duration {3321
// to keep things backwards compatible, we use non-zero values of
3322
// IdleConnTimeout, followed by using the IdleConnTimeout on the underlying
3323
// http1 transport, followed by 0
3324
if t.IdleConnTimeout != 0 {3325
return t.IdleConnTimeout
3329
return t.t1.IdleConnTimeout
3335
func traceGetConn(req *http.Request, hostPort string) {3336
trace := httptrace.ContextClientTrace(req.Context())
3337
if trace == nil || trace.GetConn == nil {3340
trace.GetConn(hostPort)
3343
func traceGotConn(req *http.Request, cc *ClientConn, reused bool) {3344
trace := httptrace.ContextClientTrace(req.Context())
3345
if trace == nil || trace.GotConn == nil {3348
ci := httptrace.GotConnInfo{Conn: cc.tconn}3351
ci.WasIdle = len(cc.streams) == 0 && reused
3352
if ci.WasIdle && !cc.lastActive.IsZero() {3353
ci.IdleTime = time.Since(cc.lastActive)
3360
func traceWroteHeaders(trace *httptrace.ClientTrace) {3361
if trace != nil && trace.WroteHeaders != nil {3362
trace.WroteHeaders()
3366
func traceGot100Continue(trace *httptrace.ClientTrace) {3367
if trace != nil && trace.Got100Continue != nil {3368
trace.Got100Continue()
3372
func traceWait100Continue(trace *httptrace.ClientTrace) {3373
if trace != nil && trace.Wait100Continue != nil {3374
trace.Wait100Continue()
3378
func traceWroteRequest(trace *httptrace.ClientTrace, err error) {3379
if trace != nil && trace.WroteRequest != nil {3380
trace.WroteRequest(httptrace.WroteRequestInfo{Err: err})3384
func traceFirstResponseByte(trace *httptrace.ClientTrace) {3385
if trace != nil && trace.GotFirstResponseByte != nil {3386
trace.GotFirstResponseByte()
3390
func traceHasWroteHeaderField(trace *httptrace.ClientTrace) bool {3391
return trace != nil && trace.WroteHeaderField != nil
3394
func traceWroteHeaderField(trace *httptrace.ClientTrace, k, v string) {3395
if trace != nil && trace.WroteHeaderField != nil {3396
trace.WroteHeaderField(k, []string{v})3400
func traceGot1xxResponseFunc(trace *httptrace.ClientTrace) func(int, textproto.MIMEHeader) error {3402
return trace.Got1xxResponse
3407
// dialTLSWithContext uses tls.Dialer, added in Go 1.15, to open a TLS
3409
func (t *Transport) dialTLSWithContext(ctx context.Context, network, addr string, cfg *tls.Config) (*tls.Conn, error) {3410
dialer := &tls.Dialer{3413
cn, err := dialer.DialContext(ctx, network, addr)
3417
tlsCn := cn.(*tls.Conn) // DialContext comment promises this will always succeed