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 7package http2 8 9import ( 10 "bufio" 11 "bytes" 12 "compress/gzip" 13 "crypto/rand" 14 "crypto/tls" 15 "errors" 16 "fmt" 17 "io" 18 "io/ioutil" 19 "log" 20 "math" 21 mathrand "math/rand" 22 "net" 23 "net/http" 24 "sort" 25 "strconv" 26 "strings" 27 "sync" 28 "time" 29 30 "golang.org/x/net/http2/hpack" 31 "golang.org/x/net/idna" 32 "golang.org/x/net/lex/httplex" 33) 34 35const ( 36 // transportDefaultConnFlow is how many connection-level flow control 37 // tokens we give the server at start-up, past the default 64k. 38 transportDefaultConnFlow = 1 << 30 39 40 // transportDefaultStreamFlow is how many stream-level flow 41 // control tokens we announce to the peer, and how many bytes 42 // we buffer per stream. 43 transportDefaultStreamFlow = 4 << 20 44 45 // transportDefaultStreamMinRefresh is the minimum number of bytes we'll send 46 // a stream-level WINDOW_UPDATE for at a time. 47 transportDefaultStreamMinRefresh = 4 << 10 48 49 defaultUserAgent = "Go-http-client/2.0" 50) 51 52// Transport is an HTTP/2 Transport. 53// 54// A Transport internally caches connections to servers. It is safe 55// for concurrent use by multiple goroutines. 56type Transport struct { 57 // DialTLS specifies an optional dial function for creating 58 // TLS connections for requests. 59 // 60 // If DialTLS is nil, tls.Dial is used. 61 // 62 // If the returned net.Conn has a ConnectionState method like tls.Conn, 63 // it will be used to set http.Response.TLS. 64 DialTLS func(network, addr string, cfg *tls.Config) (net.Conn, error) 65 66 // TLSClientConfig specifies the TLS configuration to use with 67 // tls.Client. If nil, the default configuration is used. 68 TLSClientConfig *tls.Config 69 70 // ConnPool optionally specifies an alternate connection pool to use. 71 // If nil, the default is used. 72 ConnPool ClientConnPool 73 74 // DisableCompression, if true, prevents the Transport from 75 // requesting compression with an "Accept-Encoding: gzip" 76 // request header when the Request contains no existing 77 // Accept-Encoding value. If the Transport requests gzip on 78 // its own and gets a gzipped response, it's transparently 79 // decoded in the Response.Body. However, if the user 80 // explicitly requested gzip it is not automatically 81 // uncompressed. 82 DisableCompression bool 83 84 // AllowHTTP, if true, permits HTTP/2 requests using the insecure, 85 // plain-text "http" scheme. Note that this does not enable h2c support. 86 AllowHTTP bool 87 88 // MaxHeaderListSize is the http2 SETTINGS_MAX_HEADER_LIST_SIZE to 89 // send in the initial settings frame. It is how many bytes 90 // of response headers are allowed. Unlike the http2 spec, zero here 91 // means to use a default limit (currently 10MB). If you actually 92 // want to advertise an ulimited value to the peer, Transport 93 // interprets the highest possible value here (0xffffffff or 1<<32-1) 94 // to mean no limit. 95 MaxHeaderListSize uint32 96 97 // t1, if non-nil, is the standard library Transport using 98 // this transport. Its settings are used (but not its 99 // RoundTrip method, etc). 100 t1 *http.Transport 101 102 connPoolOnce sync.Once 103 connPoolOrDef ClientConnPool // non-nil version of ConnPool 104} 105 106func (t *Transport) maxHeaderListSize() uint32 { 107 if t.MaxHeaderListSize == 0 { 108 return 10 << 20 109 } 110 if t.MaxHeaderListSize == 0xffffffff { 111 return 0 112 } 113 return t.MaxHeaderListSize 114} 115 116func (t *Transport) disableCompression() bool { 117 return t.DisableCompression || (t.t1 != nil && t.t1.DisableCompression) 118} 119 120var errTransportVersion = errors.New("http2: ConfigureTransport is only supported starting at Go 1.6") 121 122// ConfigureTransport configures a net/http HTTP/1 Transport to use HTTP/2. 123// It requires Go 1.6 or later and returns an error if the net/http package is too old 124// or if t1 has already been HTTP/2-enabled. 125func ConfigureTransport(t1 *http.Transport) error { 126 _, err := configureTransport(t1) // in configure_transport.go (go1.6) or not_go16.go 127 return err 128} 129 130func (t *Transport) connPool() ClientConnPool { 131 t.connPoolOnce.Do(t.initConnPool) 132 return t.connPoolOrDef 133} 134 135func (t *Transport) initConnPool() { 136 if t.ConnPool != nil { 137 t.connPoolOrDef = t.ConnPool 138 } else { 139 t.connPoolOrDef = &clientConnPool{t: t} 140 } 141} 142 143// ClientConn is the state of a single HTTP/2 client connection to an 144// HTTP/2 server. 145type ClientConn struct { 146 t *Transport 147 tconn net.Conn // usually *tls.Conn, except specialized impls 148 tlsState *tls.ConnectionState // nil only for specialized impls 149 singleUse bool // whether being used for a single http.Request 150 151 // readLoop goroutine fields: 152 readerDone chan struct{} // closed on error 153 readerErr error // set before readerDone is closed 154 155 idleTimeout time.Duration // or 0 for never 156 idleTimer *time.Timer 157 158 mu sync.Mutex // guards following 159 cond *sync.Cond // hold mu; broadcast on flow/closed changes 160 flow flow // our conn-level flow control quota (cs.flow is per stream) 161 inflow flow // peer's conn-level flow control 162 closed bool 163 wantSettingsAck bool // we sent a SETTINGS frame and haven't heard back 164 goAway *GoAwayFrame // if non-nil, the GoAwayFrame we received 165 goAwayDebug string // goAway frame's debug data, retained as a string 166 streams map[uint32]*clientStream // client-initiated 167 nextStreamID uint32 168 pendingRequests int // requests blocked and waiting to be sent because len(streams) == maxConcurrentStreams 169 pings map[[8]byte]chan struct{} // in flight ping data to notification channel 170 bw *bufio.Writer 171 br *bufio.Reader 172 fr *Framer 173 lastActive time.Time 174 // Settings from peer: (also guarded by mu) 175 maxFrameSize uint32 176 maxConcurrentStreams uint32 177 peerMaxHeaderListSize uint64 178 initialWindowSize uint32 179 180 hbuf bytes.Buffer // HPACK encoder writes into this 181 henc *hpack.Encoder 182 freeBuf [][]byte 183 184 wmu sync.Mutex // held while writing; acquire AFTER mu if holding both 185 werr error // first write error that has occurred 186} 187 188// clientStream is the state for a single HTTP/2 stream. One of these 189// is created for each Transport.RoundTrip call. 190type clientStream struct { 191 cc *ClientConn 192 req *http.Request 193 trace *clientTrace // or nil 194 ID uint32 195 resc chan resAndError 196 bufPipe pipe // buffered pipe with the flow-controlled response payload 197 startedWrite bool // started request body write; guarded by cc.mu 198 requestedGzip bool 199 on100 func() // optional code to run if get a 100 continue response 200 201 flow flow // guarded by cc.mu 202 inflow flow // guarded by cc.mu 203 bytesRemain int64 // -1 means unknown; owned by transportResponseBody.Read 204 readErr error // sticky read error; owned by transportResponseBody.Read 205 stopReqBody error // if non-nil, stop writing req body; guarded by cc.mu 206 didReset bool // whether we sent a RST_STREAM to the server; guarded by cc.mu 207 208 peerReset chan struct{} // closed on peer reset 209 resetErr error // populated before peerReset is closed 210 211 done chan struct{} // closed when stream remove from cc.streams map; close calls guarded by cc.mu 212 213 // owned by clientConnReadLoop: 214 firstByte bool // got the first response byte 215 pastHeaders bool // got first MetaHeadersFrame (actual headers) 216 pastTrailers bool // got optional second MetaHeadersFrame (trailers) 217 218 trailer http.Header // accumulated trailers 219 resTrailer *http.Header // client's Response.Trailer 220} 221 222// awaitRequestCancel waits for the user to cancel a request or for the done 223// channel to be signaled. A non-nil error is returned only if the request was 224// canceled. 225func awaitRequestCancel(req *http.Request, done <-chan struct{}) error { 226 ctx := reqContext(req) 227 if req.Cancel == nil && ctx.Done() == nil { 228 return nil 229 } 230 select { 231 case <-req.Cancel: 232 return errRequestCanceled 233 case <-ctx.Done(): 234 return ctx.Err() 235 case <-done: 236 return nil 237 } 238} 239 240// awaitRequestCancel waits for the user to cancel a request, its context to 241// expire, or for the request to be done (any way it might be removed from the 242// cc.streams map: peer reset, successful completion, TCP connection breakage, 243// etc). If the request is canceled, then cs will be canceled and closed. 244func (cs *clientStream) awaitRequestCancel(req *http.Request) { 245 if err := awaitRequestCancel(req, cs.done); err != nil { 246 cs.cancelStream() 247 cs.bufPipe.CloseWithError(err) 248 } 249} 250 251func (cs *clientStream) cancelStream() { 252 cc := cs.cc 253 cc.mu.Lock() 254 didReset := cs.didReset 255 cs.didReset = true 256 cc.mu.Unlock() 257 258 if !didReset { 259 cc.writeStreamReset(cs.ID, ErrCodeCancel, nil) 260 cc.forgetStreamID(cs.ID) 261 } 262} 263 264// checkResetOrDone reports any error sent in a RST_STREAM frame by the 265// server, or errStreamClosed if the stream is complete. 266func (cs *clientStream) checkResetOrDone() error { 267 select { 268 case <-cs.peerReset: 269 return cs.resetErr 270 case <-cs.done: 271 return errStreamClosed 272 default: 273 return nil 274 } 275} 276 277func (cs *clientStream) getStartedWrite() bool { 278 cc := cs.cc 279 cc.mu.Lock() 280 defer cc.mu.Unlock() 281 return cs.startedWrite 282} 283 284func (cs *clientStream) abortRequestBodyWrite(err error) { 285 if err == nil { 286 panic("nil error") 287 } 288 cc := cs.cc 289 cc.mu.Lock() 290 cs.stopReqBody = err 291 cc.cond.Broadcast() 292 cc.mu.Unlock() 293} 294 295type stickyErrWriter struct { 296 w io.Writer 297 err *error 298} 299 300func (sew stickyErrWriter) Write(p []byte) (n int, err error) { 301 if *sew.err != nil { 302 return 0, *sew.err 303 } 304 n, err = sew.w.Write(p) 305 *sew.err = err 306 return 307} 308 309var ErrNoCachedConn = errors.New("http2: no cached connection was available") 310 311// RoundTripOpt are options for the Transport.RoundTripOpt method. 312type RoundTripOpt struct { 313 // OnlyCachedConn controls whether RoundTripOpt may 314 // create a new TCP connection. If set true and 315 // no cached connection is available, RoundTripOpt 316 // will return ErrNoCachedConn. 317 OnlyCachedConn bool 318} 319 320func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) { 321 return t.RoundTripOpt(req, RoundTripOpt{}) 322} 323 324// authorityAddr returns a given authority (a host/IP, or host:port / ip:port) 325// and returns a host:port. The port 443 is added if needed. 326func authorityAddr(scheme string, authority string) (addr string) { 327 host, port, err := net.SplitHostPort(authority) 328 if err != nil { // authority didn't have a port 329 port = "443" 330 if scheme == "http" { 331 port = "80" 332 } 333 host = authority 334 } 335 if a, err := idna.ToASCII(host); err == nil { 336 host = a 337 } 338 // IPv6 address literal, without a port: 339 if strings.HasPrefix(host, "[") && strings.HasSuffix(host, "]") { 340 return host + ":" + port 341 } 342 return net.JoinHostPort(host, port) 343} 344 345// RoundTripOpt is like RoundTrip, but takes options. 346func (t *Transport) RoundTripOpt(req *http.Request, opt RoundTripOpt) (*http.Response, error) { 347 if !(req.URL.Scheme == "https" || (req.URL.Scheme == "http" && t.AllowHTTP)) { 348 return nil, errors.New("http2: unsupported scheme") 349 } 350 351 addr := authorityAddr(req.URL.Scheme, req.URL.Host) 352 for retry := 0; ; retry++ { 353 cc, err := t.connPool().GetClientConn(req, addr) 354 if err != nil { 355 t.vlogf("http2: Transport failed to get client conn for %s: %v", addr, err) 356 return nil, err 357 } 358 traceGotConn(req, cc) 359 res, gotErrAfterReqBodyWrite, err := cc.roundTrip(req) 360 if err != nil && retry <= 6 { 361 if req, err = shouldRetryRequest(req, err, gotErrAfterReqBodyWrite); err == nil { 362 // After the first retry, do exponential backoff with 10% jitter. 363 if retry == 0 { 364 continue 365 } 366 backoff := float64(uint(1) << (uint(retry) - 1)) 367 backoff += backoff * (0.1 * mathrand.Float64()) 368 select { 369 case <-time.After(time.Second * time.Duration(backoff)): 370 continue 371 case <-reqContext(req).Done(): 372 return nil, reqContext(req).Err() 373 } 374 } 375 } 376 if err != nil { 377 t.vlogf("RoundTrip failure: %v", err) 378 return nil, err 379 } 380 return res, nil 381 } 382} 383 384// CloseIdleConnections closes any connections which were previously 385// connected from previous requests but are now sitting idle. 386// It does not interrupt any connections currently in use. 387func (t *Transport) CloseIdleConnections() { 388 if cp, ok := t.connPool().(clientConnPoolIdleCloser); ok { 389 cp.closeIdleConnections() 390 } 391} 392 393var ( 394 errClientConnClosed = errors.New("http2: client conn is closed") 395 errClientConnUnusable = errors.New("http2: client conn not usable") 396 errClientConnGotGoAway = errors.New("http2: Transport received Server's graceful shutdown GOAWAY") 397) 398 399// shouldRetryRequest is called by RoundTrip when a request fails to get 400// response headers. It is always called with a non-nil error. 401// It returns either a request to retry (either the same request, or a 402// modified clone), or an error if the request can't be replayed. 403func shouldRetryRequest(req *http.Request, err error, afterBodyWrite bool) (*http.Request, error) { 404 if !canRetryError(err) { 405 return nil, err 406 } 407 if !afterBodyWrite { 408 return req, nil 409 } 410 // If the Body is nil (or http.NoBody), it's safe to reuse 411 // this request and its Body. 412 if req.Body == nil || reqBodyIsNoBody(req.Body) { 413 return req, nil 414 } 415 // Otherwise we depend on the Request having its GetBody 416 // func defined. 417 getBody := reqGetBody(req) // Go 1.8: getBody = req.GetBody 418 if getBody == nil { 419 return nil, fmt.Errorf("http2: Transport: cannot retry err [%v] after Request.Body was written; define Request.GetBody to avoid this error", err) 420 } 421 body, err := getBody() 422 if err != nil { 423 return nil, err 424 } 425 newReq := *req 426 newReq.Body = body 427 return &newReq, nil 428} 429 430func canRetryError(err error) bool { 431 if err == errClientConnUnusable || err == errClientConnGotGoAway { 432 return true 433 } 434 if se, ok := err.(StreamError); ok { 435 return se.Code == ErrCodeRefusedStream 436 } 437 return false 438} 439 440func (t *Transport) dialClientConn(addr string, singleUse bool) (*ClientConn, error) { 441 host, _, err := net.SplitHostPort(addr) 442 if err != nil { 443 return nil, err 444 } 445 tconn, err := t.dialTLS()("tcp", addr, t.newTLSConfig(host)) 446 if err != nil { 447 return nil, err 448 } 449 return t.newClientConn(tconn, singleUse) 450} 451 452func (t *Transport) newTLSConfig(host string) *tls.Config { 453 cfg := new(tls.Config) 454 if t.TLSClientConfig != nil { 455 *cfg = *cloneTLSConfig(t.TLSClientConfig) 456 } 457 if !strSliceContains(cfg.NextProtos, NextProtoTLS) { 458 cfg.NextProtos = append([]string{NextProtoTLS}, cfg.NextProtos...) 459 } 460 if cfg.ServerName == "" { 461 cfg.ServerName = host 462 } 463 return cfg 464} 465 466func (t *Transport) dialTLS() func(string, string, *tls.Config) (net.Conn, error) { 467 if t.DialTLS != nil { 468 return t.DialTLS 469 } 470 return t.dialTLSDefault 471} 472 473func (t *Transport) dialTLSDefault(network, addr string, cfg *tls.Config) (net.Conn, error) { 474 cn, err := tls.Dial(network, addr, cfg) 475 if err != nil { 476 return nil, err 477 } 478 if err := cn.Handshake(); err != nil { 479 return nil, err 480 } 481 if !cfg.InsecureSkipVerify { 482 if err := cn.VerifyHostname(cfg.ServerName); err != nil { 483 return nil, err 484 } 485 } 486 state := cn.ConnectionState() 487 if p := state.NegotiatedProtocol; p != NextProtoTLS { 488 return nil, fmt.Errorf("http2: unexpected ALPN protocol %q; want %q", p, NextProtoTLS) 489 } 490 if !state.NegotiatedProtocolIsMutual { 491 return nil, errors.New("http2: could not negotiate protocol mutually") 492 } 493 return cn, nil 494} 495 496// disableKeepAlives reports whether connections should be closed as 497// soon as possible after handling the first request. 498func (t *Transport) disableKeepAlives() bool { 499 return t.t1 != nil && t.t1.DisableKeepAlives 500} 501 502func (t *Transport) expectContinueTimeout() time.Duration { 503 if t.t1 == nil { 504 return 0 505 } 506 return transportExpectContinueTimeout(t.t1) 507} 508 509func (t *Transport) NewClientConn(c net.Conn) (*ClientConn, error) { 510 return t.newClientConn(c, false) 511} 512 513func (t *Transport) newClientConn(c net.Conn, singleUse bool) (*ClientConn, error) { 514 cc := &ClientConn{ 515 t: t, 516 tconn: c, 517 readerDone: make(chan struct{}), 518 nextStreamID: 1, 519 maxFrameSize: 16 << 10, // spec default 520 initialWindowSize: 65535, // spec default 521 maxConcurrentStreams: 1000, // "infinite", per spec. 1000 seems good enough. 522 peerMaxHeaderListSize: 0xffffffffffffffff, // "infinite", per spec. Use 2^64-1 instead. 523 streams: make(map[uint32]*clientStream), 524 singleUse: singleUse, 525 wantSettingsAck: true, 526 pings: make(map[[8]byte]chan struct{}), 527 } 528 if d := t.idleConnTimeout(); d != 0 { 529 cc.idleTimeout = d 530 cc.idleTimer = time.AfterFunc(d, cc.onIdleTimeout) 531 } 532 if VerboseLogs { 533 t.vlogf("http2: Transport creating client conn %p to %v", cc, c.RemoteAddr()) 534 } 535 536 cc.cond = sync.NewCond(&cc.mu) 537 cc.flow.add(int32(initialWindowSize)) 538 539 // TODO: adjust this writer size to account for frame size + 540 // MTU + crypto/tls record padding. 541 cc.bw = bufio.NewWriter(stickyErrWriter{c, &cc.werr}) 542 cc.br = bufio.NewReader(c) 543 cc.fr = NewFramer(cc.bw, cc.br) 544 cc.fr.ReadMetaHeaders = hpack.NewDecoder(initialHeaderTableSize, nil) 545 cc.fr.MaxHeaderListSize = t.maxHeaderListSize() 546 547 // TODO: SetMaxDynamicTableSize, SetMaxDynamicTableSizeLimit on 548 // henc in response to SETTINGS frames? 549 cc.henc = hpack.NewEncoder(&cc.hbuf) 550 551 if cs, ok := c.(connectionStater); ok { 552 state := cs.ConnectionState() 553 cc.tlsState = &state 554 } 555 556 initialSettings := []Setting{ 557 {ID: SettingEnablePush, Val: 0}, 558 {ID: SettingInitialWindowSize, Val: transportDefaultStreamFlow}, 559 } 560 if max := t.maxHeaderListSize(); max != 0 { 561 initialSettings = append(initialSettings, Setting{ID: SettingMaxHeaderListSize, Val: max}) 562 } 563 564 cc.bw.Write(clientPreface) 565 cc.fr.WriteSettings(initialSettings...) 566 cc.fr.WriteWindowUpdate(0, transportDefaultConnFlow) 567 cc.inflow.add(transportDefaultConnFlow + initialWindowSize) 568 cc.bw.Flush() 569 if cc.werr != nil { 570 return nil, cc.werr 571 } 572 573 go cc.readLoop() 574 return cc, nil 575} 576 577func (cc *ClientConn) setGoAway(f *GoAwayFrame) { 578 cc.mu.Lock() 579 defer cc.mu.Unlock() 580 581 old := cc.goAway 582 cc.goAway = f 583 584 // Merge the previous and current GoAway error frames. 585 if cc.goAwayDebug == "" { 586 cc.goAwayDebug = string(f.DebugData()) 587 } 588 if old != nil && old.ErrCode != ErrCodeNo { 589 cc.goAway.ErrCode = old.ErrCode 590 } 591 last := f.LastStreamID 592 for streamID, cs := range cc.streams { 593 if streamID > last { 594 select { 595 case cs.resc <- resAndError{err: errClientConnGotGoAway}: 596 default: 597 } 598 } 599 } 600} 601 602// CanTakeNewRequest reports whether the connection can take a new request, 603// meaning it has not been closed or received or sent a GOAWAY. 604func (cc *ClientConn) CanTakeNewRequest() bool { 605 cc.mu.Lock() 606 defer cc.mu.Unlock() 607 return cc.canTakeNewRequestLocked() 608} 609 610func (cc *ClientConn) canTakeNewRequestLocked() bool { 611 if cc.singleUse && cc.nextStreamID > 1 { 612 return false 613 } 614 return cc.goAway == nil && !cc.closed && 615 int64(cc.nextStreamID)+int64(cc.pendingRequests) < math.MaxInt32 616} 617 618// onIdleTimeout is called from a time.AfterFunc goroutine. It will 619// only be called when we're idle, but because we're coming from a new 620// goroutine, there could be a new request coming in at the same time, 621// so this simply calls the synchronized closeIfIdle to shut down this 622// connection. The timer could just call closeIfIdle, but this is more 623// clear. 624func (cc *ClientConn) onIdleTimeout() { 625 cc.closeIfIdle() 626} 627 628func (cc *ClientConn) closeIfIdle() { 629 cc.mu.Lock() 630 if len(cc.streams) > 0 { 631 cc.mu.Unlock() 632 return 633 } 634 cc.closed = true 635 nextID := cc.nextStreamID 636 // TODO: do clients send GOAWAY too? maybe? Just Close: 637 cc.mu.Unlock() 638 639 if VerboseLogs { 640 cc.vlogf("http2: Transport closing idle conn %p (forSingleUse=%v, maxStream=%v)", cc, cc.singleUse, nextID-2) 641 } 642 cc.tconn.Close() 643} 644 645const maxAllocFrameSize = 512 << 10 646 647// frameBuffer returns a scratch buffer suitable for writing DATA frames. 648// They're capped at the min of the peer's max frame size or 512KB 649// (kinda arbitrarily), but definitely capped so we don't allocate 4GB 650// bufers. 651func (cc *ClientConn) frameScratchBuffer() []byte { 652 cc.mu.Lock() 653 size := cc.maxFrameSize 654 if size > maxAllocFrameSize { 655 size = maxAllocFrameSize 656 } 657 for i, buf := range cc.freeBuf { 658 if len(buf) >= int(size) { 659 cc.freeBuf[i] = nil 660 cc.mu.Unlock() 661 return buf[:size] 662 } 663 } 664 cc.mu.Unlock() 665 return make([]byte, size) 666} 667 668func (cc *ClientConn) putFrameScratchBuffer(buf []byte) { 669 cc.mu.Lock() 670 defer cc.mu.Unlock() 671 const maxBufs = 4 // arbitrary; 4 concurrent requests per conn? investigate. 672 if len(cc.freeBuf) < maxBufs { 673 cc.freeBuf = append(cc.freeBuf, buf) 674 return 675 } 676 for i, old := range cc.freeBuf { 677 if old == nil { 678 cc.freeBuf[i] = buf 679 return 680 } 681 } 682 // forget about it. 683} 684 685// errRequestCanceled is a copy of net/http's errRequestCanceled because it's not 686// exported. At least they'll be DeepEqual for h1-vs-h2 comparisons tests. 687var errRequestCanceled = errors.New("net/http: request canceled") 688 689func commaSeparatedTrailers(req *http.Request) (string, error) { 690 keys := make([]string, 0, len(req.Trailer)) 691 for k := range req.Trailer { 692 k = http.CanonicalHeaderKey(k) 693 switch k { 694 case "Transfer-Encoding", "Trailer", "Content-Length": 695 return "", &badStringError{"invalid Trailer key", k} 696 } 697 keys = append(keys, k) 698 } 699 if len(keys) > 0 { 700 sort.Strings(keys) 701 return strings.Join(keys, ","), nil 702 } 703 return "", nil 704} 705 706func (cc *ClientConn) responseHeaderTimeout() time.Duration { 707 if cc.t.t1 != nil { 708 return cc.t.t1.ResponseHeaderTimeout 709 } 710 // No way to do this (yet?) with just an http2.Transport. Probably 711 // no need. Request.Cancel this is the new way. We only need to support 712 // this for compatibility with the old http.Transport fields when 713 // we're doing transparent http2. 714 return 0 715} 716 717// checkConnHeaders checks whether req has any invalid connection-level headers. 718// per RFC 7540 section 8.1.2.2: Connection-Specific Header Fields. 719// Certain headers are special-cased as okay but not transmitted later. 720func checkConnHeaders(req *http.Request) error { 721 if v := req.Header.Get("Upgrade"); v != "" { 722 return fmt.Errorf("http2: invalid Upgrade request header: %q", req.Header["Upgrade"]) 723 } 724 if vv := req.Header["Transfer-Encoding"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && vv[0] != "chunked") { 725 return fmt.Errorf("http2: invalid Transfer-Encoding request header: %q", vv) 726 } 727 if vv := req.Header["Connection"]; len(vv) > 0 && (len(vv) > 1 || vv[0] != "" && vv[0] != "close" && vv[0] != "keep-alive") { 728 return fmt.Errorf("http2: invalid Connection request header: %q", vv) 729 } 730 return nil 731} 732 733// actualContentLength returns a sanitized version of 734// req.ContentLength, where 0 actually means zero (not unknown) and -1 735// means unknown. 736func actualContentLength(req *http.Request) int64 { 737 if req.Body == nil || reqBodyIsNoBody(req.Body) { 738 return 0 739 } 740 if req.ContentLength != 0 { 741 return req.ContentLength 742 } 743 return -1 744} 745 746func (cc *ClientConn) RoundTrip(req *http.Request) (*http.Response, error) { 747 resp, _, err := cc.roundTrip(req) 748 return resp, err 749} 750 751func (cc *ClientConn) roundTrip(req *http.Request) (res *http.Response, gotErrAfterReqBodyWrite bool, err error) { 752 if err := checkConnHeaders(req); err != nil { 753 return nil, false, err 754 } 755 if cc.idleTimer != nil { 756 cc.idleTimer.Stop() 757 } 758 759 trailers, err := commaSeparatedTrailers(req) 760 if err != nil { 761 return nil, false, err 762 } 763 hasTrailers := trailers != "" 764 765 cc.mu.Lock() 766 if err := cc.awaitOpenSlotForRequest(req); err != nil { 767 cc.mu.Unlock() 768 return nil, false, err 769 } 770 771 body := req.Body 772 contentLen := actualContentLength(req) 773 hasBody := contentLen != 0 774 775 // TODO(bradfitz): this is a copy of the logic in net/http. Unify somewhere? 776 var requestedGzip bool 777 if !cc.t.disableCompression() && 778 req.Header.Get("Accept-Encoding") == "" && 779 req.Header.Get("Range") == "" && 780 req.Method != "HEAD" { 781 // Request gzip only, not deflate. Deflate is ambiguous and 782 // not as universally supported anyway. 783 // See: http://www.gzip.org/zlib/zlib_faq.html#faq38 784 // 785 // Note that we don't request this for HEAD requests, 786 // due to a bug in nginx: 787 // http://trac.nginx.org/nginx/ticket/358 788 // https://golang.org/issue/5522 789 // 790 // We don't request gzip if the request is for a range, since 791 // auto-decoding a portion of a gzipped document will just fail 792 // anyway. See https://golang.org/issue/8923 793 requestedGzip = true 794 } 795 796 // we send: HEADERS{1}, CONTINUATION{0,} + DATA{0,} (DATA is 797 // sent by writeRequestBody below, along with any Trailers, 798 // again in form HEADERS{1}, CONTINUATION{0,}) 799 hdrs, err := cc.encodeHeaders(req, requestedGzip, trailers, contentLen) 800 if err != nil { 801 cc.mu.Unlock() 802 return nil, false, err 803 } 804 805 cs := cc.newStream() 806 cs.req = req 807 cs.trace = requestTrace(req) 808 cs.requestedGzip = requestedGzip 809 bodyWriter := cc.t.getBodyWriterState(cs, body) 810 cs.on100 = bodyWriter.on100 811 812 cc.wmu.Lock() 813 endStream := !hasBody && !hasTrailers 814 werr := cc.writeHeaders(cs.ID, endStream, hdrs) 815 cc.wmu.Unlock() 816 traceWroteHeaders(cs.trace) 817 cc.mu.Unlock() 818 819 if werr != nil { 820 if hasBody { 821 req.Body.Close() // per RoundTripper contract 822 bodyWriter.cancel() 823 } 824 cc.forgetStreamID(cs.ID) 825 // Don't bother sending a RST_STREAM (our write already failed; 826 // no need to keep writing) 827 traceWroteRequest(cs.trace, werr) 828 return nil, false, werr 829 } 830 831 var respHeaderTimer <-chan time.Time 832 if hasBody { 833 bodyWriter.scheduleBodyWrite() 834 } else { 835 traceWroteRequest(cs.trace, nil) 836 if d := cc.responseHeaderTimeout(); d != 0 { 837 timer := time.NewTimer(d) 838 defer timer.Stop() 839 respHeaderTimer = timer.C 840 } 841 } 842 843 readLoopResCh := cs.resc 844 bodyWritten := false 845 ctx := reqContext(req) 846 847 handleReadLoopResponse := func(re resAndError) (*http.Response, bool, error) { 848 res := re.res 849 if re.err != nil || res.StatusCode > 299 { 850 // On error or status code 3xx, 4xx, 5xx, etc abort any 851 // ongoing write, assuming that the server doesn't care 852 // about our request body. If the server replied with 1xx or 853 // 2xx, however, then assume the server DOES potentially 854 // want our body (e.g. full-duplex streaming: 855 // golang.org/issue/13444). If it turns out the server 856 // doesn't, they'll RST_STREAM us soon enough. This is a 857 // heuristic to avoid adding knobs to Transport. Hopefully 858 // we can keep it. 859 bodyWriter.cancel() 860 cs.abortRequestBodyWrite(errStopReqBodyWrite) 861 } 862 if re.err != nil { 863 cc.forgetStreamID(cs.ID) 864 return nil, cs.getStartedWrite(), re.err 865 } 866 res.Request = req 867 res.TLS = cc.tlsState 868 return res, false, nil 869 } 870 871 for { 872 select { 873 case re := <-readLoopResCh: 874 return handleReadLoopResponse(re) 875 case <-respHeaderTimer: 876 if !hasBody || bodyWritten { 877 cc.writeStreamReset(cs.ID, ErrCodeCancel, nil) 878 } else { 879 bodyWriter.cancel() 880 cs.abortRequestBodyWrite(errStopReqBodyWriteAndCancel) 881 } 882 cc.forgetStreamID(cs.ID) 883 return nil, cs.getStartedWrite(), errTimeout 884 case <-ctx.Done(): 885 if !hasBody || bodyWritten { 886 cc.writeStreamReset(cs.ID, ErrCodeCancel, nil) 887 } else { 888 bodyWriter.cancel() 889 cs.abortRequestBodyWrite(errStopReqBodyWriteAndCancel) 890 } 891 cc.forgetStreamID(cs.ID) 892 return nil, cs.getStartedWrite(), ctx.Err() 893 case <-req.Cancel: 894 if !hasBody || bodyWritten { 895 cc.writeStreamReset(cs.ID, ErrCodeCancel, nil) 896 } else { 897 bodyWriter.cancel() 898 cs.abortRequestBodyWrite(errStopReqBodyWriteAndCancel) 899 } 900 cc.forgetStreamID(cs.ID) 901 return nil, cs.getStartedWrite(), errRequestCanceled 902 case <-cs.peerReset: 903 // processResetStream already removed the 904 // stream from the streams map; no need for 905 // forgetStreamID. 906 return nil, cs.getStartedWrite(), cs.resetErr 907 case err := <-bodyWriter.resc: 908 // Prefer the read loop's response, if available. Issue 16102. 909 select { 910 case re := <-readLoopResCh: 911 return handleReadLoopResponse(re) 912 default: 913 } 914 if err != nil { 915 return nil, cs.getStartedWrite(), err 916 } 917 bodyWritten = true 918 if d := cc.responseHeaderTimeout(); d != 0 { 919 timer := time.NewTimer(d) 920 defer timer.Stop() 921 respHeaderTimer = timer.C 922 } 923 } 924 } 925} 926 927// awaitOpenSlotForRequest waits until len(streams) < maxConcurrentStreams. 928// Must hold cc.mu. 929func (cc *ClientConn) awaitOpenSlotForRequest(req *http.Request) error { 930 var waitingForConn chan struct{} 931 var waitingForConnErr error // guarded by cc.mu 932 for { 933 cc.lastActive = time.Now() 934 if cc.closed || !cc.canTakeNewRequestLocked() { 935 return errClientConnUnusable 936 } 937 if int64(len(cc.streams))+1 <= int64(cc.maxConcurrentStreams) { 938 if waitingForConn != nil { 939 close(waitingForConn) 940 } 941 return nil 942 } 943 // Unfortunately, we cannot wait on a condition variable and channel at 944 // the same time, so instead, we spin up a goroutine to check if the 945 // request is canceled while we wait for a slot to open in the connection. 946 if waitingForConn == nil { 947 waitingForConn = make(chan struct{}) 948 go func() { 949 if err := awaitRequestCancel(req, waitingForConn); err != nil { 950 cc.mu.Lock() 951 waitingForConnErr = err 952 cc.cond.Broadcast() 953 cc.mu.Unlock() 954 } 955 }() 956 } 957 cc.pendingRequests++ 958 cc.cond.Wait() 959 cc.pendingRequests-- 960 if waitingForConnErr != nil { 961 return waitingForConnErr 962 } 963 } 964} 965 966// requires cc.wmu be held 967func (cc *ClientConn) writeHeaders(streamID uint32, endStream bool, hdrs []byte) error { 968 first := true // first frame written (HEADERS is first, then CONTINUATION) 969 frameSize := int(cc.maxFrameSize) 970 for len(hdrs) > 0 && cc.werr == nil { 971 chunk := hdrs 972 if len(chunk) > frameSize { 973 chunk = chunk[:frameSize] 974 } 975 hdrs = hdrs[len(chunk):] 976 endHeaders := len(hdrs) == 0 977 if first { 978 cc.fr.WriteHeaders(HeadersFrameParam{ 979 StreamID: streamID, 980 BlockFragment: chunk, 981 EndStream: endStream, 982 EndHeaders: endHeaders, 983 }) 984 first = false 985 } else { 986 cc.fr.WriteContinuation(streamID, endHeaders, chunk) 987 } 988 } 989 // TODO(bradfitz): this Flush could potentially block (as 990 // could the WriteHeaders call(s) above), which means they 991 // wouldn't respond to Request.Cancel being readable. That's 992 // rare, but this should probably be in a goroutine. 993 cc.bw.Flush() 994 return cc.werr 995} 996 997// internal error values; they don't escape to callers 998var ( 999 // abort request body write; don't send cancel 1000 errStopReqBodyWrite = errors.New("http2: aborting request body write") 1001 1002 // abort request body write, but send stream reset of cancel. 1003 errStopReqBodyWriteAndCancel = errors.New("http2: canceling request") 1004) 1005 1006func (cs *clientStream) writeRequestBody(body io.Reader, bodyCloser io.Closer) (err error) { 1007 cc := cs.cc 1008 sentEnd := false // whether we sent the final DATA frame w/ END_STREAM 1009 buf := cc.frameScratchBuffer() 1010 defer cc.putFrameScratchBuffer(buf) 1011 1012 defer func() { 1013 traceWroteRequest(cs.trace, err) 1014 // TODO: write h12Compare test showing whether 1015 // Request.Body is closed by the Transport, 1016 // and in multiple cases: server replies <=299 and >299 1017 // while still writing request body 1018 cerr := bodyCloser.Close() 1019 if err == nil { 1020 err = cerr 1021 } 1022 }() 1023 1024 req := cs.req 1025 hasTrailers := req.Trailer != nil 1026 1027 var sawEOF bool 1028 for !sawEOF { 1029 n, err := body.Read(buf) 1030 if err == io.EOF { 1031 sawEOF = true 1032 err = nil 1033 } else if err != nil { 1034 return err 1035 } 1036 1037 remain := buf[:n] 1038 for len(remain) > 0 && err == nil { 1039 var allowed int32 1040 allowed, err = cs.awaitFlowControl(len(remain)) 1041 switch { 1042 case err == errStopReqBodyWrite: 1043 return err 1044 case err == errStopReqBodyWriteAndCancel: 1045 cc.writeStreamReset(cs.ID, ErrCodeCancel, nil) 1046 return err 1047 case err != nil: 1048 return err 1049 } 1050 cc.wmu.Lock() 1051 data := remain[:allowed] 1052 remain = remain[allowed:] 1053 sentEnd = sawEOF && len(remain) == 0 && !hasTrailers 1054 err = cc.fr.WriteData(cs.ID, sentEnd, data) 1055 if err == nil { 1056 // TODO(bradfitz): this flush is for latency, not bandwidth. 1057 // Most requests won't need this. Make this opt-in or 1058 // opt-out? Use some heuristic on the body type? Nagel-like 1059 // timers? Based on 'n'? Only last chunk of this for loop, 1060 // unless flow control tokens are low? For now, always. 1061 // If we change this, see comment below. 1062 err = cc.bw.Flush() 1063 } 1064 cc.wmu.Unlock() 1065 } 1066 if err != nil { 1067 return err 1068 } 1069 } 1070 1071 if sentEnd { 1072 // Already sent END_STREAM (which implies we have no 1073 // trailers) and flushed, because currently all 1074 // WriteData frames above get a flush. So we're done. 1075 return nil 1076 } 1077 1078 var trls []byte 1079 if hasTrailers { 1080 cc.mu.Lock() 1081 trls, err = cc.encodeTrailers(req) 1082 cc.mu.Unlock() 1083 if err != nil { 1084 cc.writeStreamReset(cs.ID, ErrCodeInternal, err) 1085 cc.forgetStreamID(cs.ID) 1086 return err 1087 } 1088 } 1089 1090 cc.wmu.Lock() 1091 defer cc.wmu.Unlock() 1092 1093 // Two ways to send END_STREAM: either with trailers, or 1094 // with an empty DATA frame. 1095 if len(trls) > 0 { 1096 err = cc.writeHeaders(cs.ID, true, trls) 1097 } else { 1098 err = cc.fr.WriteData(cs.ID, true, nil) 1099 } 1100 if ferr := cc.bw.Flush(); ferr != nil && err == nil { 1101 err = ferr 1102 } 1103 return err 1104} 1105 1106// awaitFlowControl waits for [1, min(maxBytes, cc.cs.maxFrameSize)] flow 1107// control tokens from the server. 1108// It returns either the non-zero number of tokens taken or an error 1109// if the stream is dead. 1110func (cs *clientStream) awaitFlowControl(maxBytes int) (taken int32, err error) { 1111 cc := cs.cc 1112 cc.mu.Lock() 1113 defer cc.mu.Unlock() 1114 for { 1115 if cc.closed { 1116 return 0, errClientConnClosed 1117 } 1118 if cs.stopReqBody != nil { 1119 return 0, cs.stopReqBody 1120 } 1121 if err := cs.checkResetOrDone(); err != nil { 1122 return 0, err 1123 } 1124 if a := cs.flow.available(); a > 0 { 1125 take := a 1126 if int(take) > maxBytes { 1127 1128 take = int32(maxBytes) // can't truncate int; take is int32 1129 } 1130 if take > int32(cc.maxFrameSize) { 1131 take = int32(cc.maxFrameSize) 1132 } 1133 cs.flow.take(take) 1134 return take, nil 1135 } 1136 cc.cond.Wait() 1137 } 1138} 1139 1140type badStringError struct { 1141 what string 1142 str string 1143} 1144 1145func (e *badStringError) Error() string { return fmt.Sprintf("%s %q", e.what, e.str) } 1146 1147// requires cc.mu be held. 1148func (cc *ClientConn) encodeHeaders(req *http.Request, addGzipHeader bool, trailers string, contentLength int64) ([]byte, error) { 1149 cc.hbuf.Reset() 1150 1151 host := req.Host 1152 if host == "" { 1153 host = req.URL.Host 1154 } 1155 host, err := httplex.PunycodeHostPort(host) 1156 if err != nil { 1157 return nil, err 1158 } 1159 1160 var path string 1161 if req.Method != "CONNECT" { 1162 path = req.URL.RequestURI() 1163 if !validPseudoPath(path) { 1164 orig := path 1165 path = strings.TrimPrefix(path, req.URL.Scheme+"://"+host) 1166 if !validPseudoPath(path) { 1167 if req.URL.Opaque != "" { 1168 return nil, fmt.Errorf("invalid request :path %q from URL.Opaque = %q", orig, req.URL.Opaque) 1169 } else { 1170 return nil, fmt.Errorf("invalid request :path %q", orig) 1171 } 1172 } 1173 } 1174 } 1175 1176 // Check for any invalid headers and return an error before we 1177 // potentially pollute our hpack state. (We want to be able to 1178 // continue to reuse the hpack encoder for future requests) 1179 for k, vv := range req.Header { 1180 if !httplex.ValidHeaderFieldName(k) { 1181 return nil, fmt.Errorf("invalid HTTP header name %q", k) 1182 } 1183 for _, v := range vv { 1184 if !httplex.ValidHeaderFieldValue(v) { 1185 return nil, fmt.Errorf("invalid HTTP header value %q for header %q", v, k) 1186 } 1187 } 1188 } 1189 1190 enumerateHeaders := func(f func(name, value string)) { 1191 // 8.1.2.3 Request Pseudo-Header Fields 1192 // The :path pseudo-header field includes the path and query parts of the 1193 // target URI (the path-absolute production and optionally a '?' character 1194 // followed by the query production (see Sections 3.3 and 3.4 of 1195 // [RFC3986]). 1196 f(":authority", host) 1197 f(":method", req.Method) 1198 if req.Method != "CONNECT" { 1199 f(":path", path) 1200 f(":scheme", req.URL.Scheme) 1201 } 1202 if trailers != "" { 1203 f("trailer", trailers) 1204 } 1205 1206 var didUA bool 1207 for k, vv := range req.Header { 1208 if strings.EqualFold(k, "host") || strings.EqualFold(k, "content-length") { 1209 // Host is :authority, already sent. 1210 // Content-Length is automatic, set below. 1211 continue 1212 } else if strings.EqualFold(k, "connection") || strings.EqualFold(k, "proxy-connection") || 1213 strings.EqualFold(k, "transfer-encoding") || strings.EqualFold(k, "upgrade") || 1214 strings.EqualFold(k, "keep-alive") { 1215 // Per 8.1.2.2 Connection-Specific Header 1216 // Fields, don't send connection-specific 1217 // fields. We have already checked if any 1218 // are error-worthy so just ignore the rest. 1219 continue 1220 } else if strings.EqualFold(k, "user-agent") { 1221 // Match Go's http1 behavior: at most one 1222 // User-Agent. If set to nil or empty string, 1223 // then omit it. Otherwise if not mentioned, 1224 // include the default (below). 1225 didUA = true 1226 if len(vv) < 1 { 1227 continue 1228 } 1229 vv = vv[:1] 1230 if vv[0] == "" { 1231 continue 1232 } 1233 1234 } 1235 1236 for _, v := range vv { 1237 f(k, v) 1238 } 1239 } 1240 if shouldSendReqContentLength(req.Method, contentLength) { 1241 f("content-length", strconv.FormatInt(contentLength, 10)) 1242 } 1243 if addGzipHeader { 1244 f("accept-encoding", "gzip") 1245 } 1246 if !didUA { 1247 f("user-agent", defaultUserAgent) 1248 } 1249 } 1250 1251 // Do a first pass over the headers counting bytes to ensure 1252 // we don't exceed cc.peerMaxHeaderListSize. This is done as a 1253 // separate pass before encoding the headers to prevent 1254 // modifying the hpack state. 1255 hlSize := uint64(0) 1256 enumerateHeaders(func(name, value string) { 1257 hf := hpack.HeaderField{Name: name, Value: value} 1258 hlSize += uint64(hf.Size()) 1259 }) 1260 1261 if hlSize > cc.peerMaxHeaderListSize { 1262 return nil, errRequestHeaderListSize 1263 } 1264 1265 // Header list size is ok. Write the headers. 1266 enumerateHeaders(func(name, value string) { 1267 cc.writeHeader(strings.ToLower(name), value) 1268 }) 1269 1270 return cc.hbuf.Bytes(), nil 1271} 1272 1273// shouldSendReqContentLength reports whether the http2.Transport should send 1274// a "content-length" request header. This logic is basically a copy of the net/http 1275// transferWriter.shouldSendContentLength. 1276// The contentLength is the corrected contentLength (so 0 means actually 0, not unknown). 1277// -1 means unknown. 1278func shouldSendReqContentLength(method string, contentLength int64) bool { 1279 if contentLength > 0 { 1280 return true 1281 } 1282 if contentLength < 0 { 1283 return false 1284 } 1285 // For zero bodies, whether we send a content-length depends on the method. 1286 // It also kinda doesn't matter for http2 either way, with END_STREAM. 1287 switch method { 1288 case "POST", "PUT", "PATCH": 1289 return true 1290 default: 1291 return false 1292 } 1293} 1294 1295// requires cc.mu be held. 1296func (cc *ClientConn) encodeTrailers(req *http.Request) ([]byte, error) { 1297 cc.hbuf.Reset() 1298 1299 hlSize := uint64(0) 1300 for k, vv := range req.Trailer { 1301 for _, v := range vv { 1302 hf := hpack.HeaderField{Name: k, Value: v} 1303 hlSize += uint64(hf.Size()) 1304 } 1305 } 1306 if hlSize > cc.peerMaxHeaderListSize { 1307 return nil, errRequestHeaderListSize 1308 } 1309 1310 for k, vv := range req.Trailer { 1311 // Transfer-Encoding, etc.. have already been filtered at the 1312 // start of RoundTrip 1313 lowKey := strings.ToLower(k) 1314 for _, v := range vv { 1315 cc.writeHeader(lowKey, v) 1316 } 1317 } 1318 return cc.hbuf.Bytes(), nil 1319} 1320 1321func (cc *ClientConn) writeHeader(name, value string) { 1322 if VerboseLogs { 1323 log.Printf("http2: Transport encoding header %q = %q", name, value) 1324 } 1325 cc.henc.WriteField(hpack.HeaderField{Name: name, Value: value}) 1326} 1327 1328type resAndError struct { 1329 res *http.Response 1330 err error 1331} 1332 1333// requires cc.mu be held. 1334func (cc *ClientConn) newStream() *clientStream { 1335 cs := &clientStream{ 1336 cc: cc, 1337 ID: cc.nextStreamID, 1338 resc: make(chan resAndError, 1), 1339 peerReset: make(chan struct{}), 1340 done: make(chan struct{}), 1341 } 1342 cs.flow.add(int32(cc.initialWindowSize)) 1343 cs.flow.setConnFlow(&cc.flow) 1344 cs.inflow.add(transportDefaultStreamFlow) 1345 cs.inflow.setConnFlow(&cc.inflow) 1346 cc.nextStreamID += 2 1347 cc.streams[cs.ID] = cs 1348 return cs 1349} 1350 1351func (cc *ClientConn) forgetStreamID(id uint32) { 1352 cc.streamByID(id, true) 1353} 1354 1355func (cc *ClientConn) streamByID(id uint32, andRemove bool) *clientStream { 1356 cc.mu.Lock() 1357 defer cc.mu.Unlock() 1358 cs := cc.streams[id] 1359 if andRemove && cs != nil && !cc.closed { 1360 cc.lastActive = time.Now() 1361 delete(cc.streams, id) 1362 if len(cc.streams) == 0 && cc.idleTimer != nil { 1363 cc.idleTimer.Reset(cc.idleTimeout) 1364 } 1365 close(cs.done) 1366 // Wake up checkResetOrDone via clientStream.awaitFlowControl and 1367 // wake up RoundTrip if there is a pending request. 1368 cc.cond.Broadcast() 1369 } 1370 return cs 1371} 1372 1373// clientConnReadLoop is the state owned by the clientConn's frame-reading readLoop. 1374type clientConnReadLoop struct { 1375 cc *ClientConn 1376 activeRes map[uint32]*clientStream // keyed by streamID 1377 closeWhenIdle bool 1378} 1379 1380// readLoop runs in its own goroutine and reads and dispatches frames. 1381func (cc *ClientConn) readLoop() { 1382 rl := &clientConnReadLoop{ 1383 cc: cc, 1384 activeRes: make(map[uint32]*clientStream), 1385 } 1386 1387 defer rl.cleanup() 1388 cc.readerErr = rl.run() 1389 if ce, ok := cc.readerErr.(ConnectionError); ok { 1390 cc.wmu.Lock() 1391 cc.fr.WriteGoAway(0, ErrCode(ce), nil) 1392 cc.wmu.Unlock() 1393 } 1394} 1395 1396// GoAwayError is returned by the Transport when the server closes the 1397// TCP connection after sending a GOAWAY frame. 1398type GoAwayError struct { 1399 LastStreamID uint32 1400 ErrCode ErrCode 1401 DebugData string 1402} 1403 1404func (e GoAwayError) Error() string { 1405 return fmt.Sprintf("http2: server sent GOAWAY and closed the connection; LastStreamID=%v, ErrCode=%v, debug=%q", 1406 e.LastStreamID, e.ErrCode, e.DebugData) 1407} 1408 1409func isEOFOrNetReadError(err error) bool { 1410 if err == io.EOF { 1411 return true 1412 } 1413 ne, ok := err.(*net.OpError) 1414 return ok && ne.Op == "read" 1415} 1416 1417func (rl *clientConnReadLoop) cleanup() { 1418 cc := rl.cc 1419 defer cc.tconn.Close() 1420 defer cc.t.connPool().MarkDead(cc) 1421 defer close(cc.readerDone) 1422 1423 if cc.idleTimer != nil { 1424 cc.idleTimer.Stop() 1425 } 1426 1427 // Close any response bodies if the server closes prematurely. 1428 // TODO: also do this if we've written the headers but not 1429 // gotten a response yet. 1430 err := cc.readerErr 1431 cc.mu.Lock() 1432 if cc.goAway != nil && isEOFOrNetReadError(err) { 1433 err = GoAwayError{ 1434 LastStreamID: cc.goAway.LastStreamID, 1435 ErrCode: cc.goAway.ErrCode, 1436 DebugData: cc.goAwayDebug, 1437 } 1438 } else if err == io.EOF { 1439 err = io.ErrUnexpectedEOF 1440 } 1441 for _, cs := range rl.activeRes { 1442 cs.bufPipe.CloseWithError(err) 1443 } 1444 for _, cs := range cc.streams { 1445 select { 1446 case cs.resc <- resAndError{err: err}: 1447 default: 1448 } 1449 close(cs.done) 1450 } 1451 cc.closed = true 1452 cc.cond.Broadcast() 1453 cc.mu.Unlock() 1454} 1455 1456func (rl *clientConnReadLoop) run() error { 1457 cc := rl.cc 1458 rl.closeWhenIdle = cc.t.disableKeepAlives() || cc.singleUse 1459 gotReply := false // ever saw a HEADERS reply 1460 gotSettings := false 1461 for { 1462 f, err := cc.fr.ReadFrame() 1463 if err != nil { 1464 cc.vlogf("http2: Transport readFrame error on conn %p: (%T) %v", cc, err, err) 1465 } 1466 if se, ok := err.(StreamError); ok { 1467 if cs := cc.streamByID(se.StreamID, false); cs != nil { 1468 cs.cc.writeStreamReset(cs.ID, se.Code, err) 1469 cs.cc.forgetStreamID(cs.ID) 1470 if se.Cause == nil { 1471 se.Cause = cc.fr.errDetail 1472 } 1473 rl.endStreamError(cs, se) 1474 } 1475 continue 1476 } else if err != nil { 1477 return err 1478 } 1479 if VerboseLogs { 1480 cc.vlogf("http2: Transport received %s", summarizeFrame(f)) 1481 } 1482 if !gotSettings { 1483 if _, ok := f.(*SettingsFrame); !ok { 1484 cc.logf("protocol error: received %T before a SETTINGS frame", f) 1485 return ConnectionError(ErrCodeProtocol) 1486 } 1487 gotSettings = true 1488 } 1489 maybeIdle := false // whether frame might transition us to idle 1490 1491 switch f := f.(type) { 1492 case *MetaHeadersFrame: 1493 err = rl.processHeaders(f) 1494 maybeIdle = true 1495 gotReply = true 1496 case *DataFrame: 1497 err = rl.processData(f) 1498 maybeIdle = true 1499 case *GoAwayFrame: 1500 err = rl.processGoAway(f) 1501 maybeIdle = true 1502 case *RSTStreamFrame: 1503 err = rl.processResetStream(f) 1504 maybeIdle = true 1505 case *SettingsFrame: 1506 err = rl.processSettings(f) 1507 case *PushPromiseFrame: 1508 err = rl.processPushPromise(f) 1509 case *WindowUpdateFrame: 1510 err = rl.processWindowUpdate(f) 1511 case *PingFrame: 1512 err = rl.processPing(f) 1513 default: 1514 cc.logf("Transport: unhandled response frame type %T", f) 1515 } 1516 if err != nil { 1517 if VerboseLogs { 1518 cc.vlogf("http2: Transport conn %p received error from processing frame %v: %v", cc, summarizeFrame(f), err) 1519 } 1520 return err 1521 } 1522 if rl.closeWhenIdle && gotReply && maybeIdle && len(rl.activeRes) == 0 { 1523 cc.closeIfIdle() 1524 } 1525 } 1526} 1527 1528func (rl *clientConnReadLoop) processHeaders(f *MetaHeadersFrame) error { 1529 cc := rl.cc 1530 if f.StreamEnded() { 1531 // Issue 20521: If the stream has ended, streamByID() causes 1532 // clientStream.done to be closed, which causes the request's bodyWriter 1533 // to be closed with an errStreamClosed, which may be received by 1534 // clientConn.RoundTrip before the result of processing these headers. 1535 // Deferring stream closure allows the header processing to occur first. 1536 // clientConn.RoundTrip may still receive the bodyWriter error first, but 1537 // the fix for issue 16102 prioritises any response. 1538 defer cc.streamByID(f.StreamID, true) 1539 } 1540 cs := cc.streamByID(f.StreamID, false) 1541 if cs == nil { 1542 // We'd get here if we canceled a request while the 1543 // server had its response still in flight. So if this 1544 // was just something we canceled, ignore it. 1545 return nil 1546 } 1547 if !cs.firstByte { 1548 if cs.trace != nil { 1549 // TODO(bradfitz): move first response byte earlier, 1550 // when we first read the 9 byte header, not waiting 1551 // until all the HEADERS+CONTINUATION frames have been 1552 // merged. This works for now. 1553 traceFirstResponseByte(cs.trace) 1554 } 1555 cs.firstByte = true 1556 } 1557 if !cs.pastHeaders { 1558 cs.pastHeaders = true 1559 } else { 1560 return rl.processTrailers(cs, f) 1561 } 1562 1563 res, err := rl.handleResponse(cs, f) 1564 if err != nil { 1565 if _, ok := err.(ConnectionError); ok { 1566 return err 1567 } 1568 // Any other error type is a stream error. 1569 cs.cc.writeStreamReset(f.StreamID, ErrCodeProtocol, err) 1570 cs.resc <- resAndError{err: err} 1571 return nil // return nil from process* funcs to keep conn alive 1572 } 1573 if res == nil { 1574 // (nil, nil) special case. See handleResponse docs. 1575 return nil 1576 } 1577 if res.Body != noBody { 1578 rl.activeRes[cs.ID] = cs 1579 } 1580 cs.resTrailer = &res.Trailer 1581 cs.resc <- resAndError{res: res} 1582 return nil 1583} 1584 1585// may return error types nil, or ConnectionError. Any other error value 1586// is a StreamError of type ErrCodeProtocol. The returned error in that case 1587// is the detail. 1588// 1589// As a special case, handleResponse may return (nil, nil) to skip the 1590// frame (currently only used for 100 expect continue). This special 1591// case is going away after Issue 13851 is fixed. 1592func (rl *clientConnReadLoop) handleResponse(cs *clientStream, f *MetaHeadersFrame) (*http.Response, error) { 1593 if f.Truncated { 1594 return nil, errResponseHeaderListSize 1595 } 1596 1597 status := f.PseudoValue("status") 1598 if status == "" { 1599 return nil, errors.New("missing status pseudo header") 1600 } 1601 statusCode, err := strconv.Atoi(status) 1602 if err != nil { 1603 return nil, errors.New("malformed non-numeric status pseudo header") 1604 } 1605 1606 if statusCode == 100 { 1607 traceGot100Continue(cs.trace) 1608 if cs.on100 != nil { 1609 cs.on100() // forces any write delay timer to fire 1610 } 1611 cs.pastHeaders = false // do it all again 1612 return nil, nil 1613 } 1614 1615 header := make(http.Header) 1616 res := &http.Response{ 1617 Proto: "HTTP/2.0", 1618 ProtoMajor: 2, 1619 Header: header, 1620 StatusCode: statusCode, 1621 Status: status + " " + http.StatusText(statusCode), 1622 } 1623 for _, hf := range f.RegularFields() { 1624 key := http.CanonicalHeaderKey(hf.Name) 1625 if key == "Trailer" { 1626 t := res.Trailer 1627 if t == nil { 1628 t = make(http.Header) 1629 res.Trailer = t 1630 } 1631 foreachHeaderElement(hf.Value, func(v string) { 1632 t[http.CanonicalHeaderKey(v)] = nil 1633 }) 1634 } else { 1635 header[key] = append(header[key], hf.Value) 1636 } 1637 } 1638 1639 streamEnded := f.StreamEnded() 1640 isHead := cs.req.Method == "HEAD" 1641 if !streamEnded || isHead { 1642 res.ContentLength = -1 1643 if clens := res.Header["Content-Length"]; len(clens) == 1 { 1644 if clen64, err := strconv.ParseInt(clens[0], 10, 64); err == nil { 1645 res.ContentLength = clen64 1646 } else { 1647 // TODO: care? unlike http/1, it won't mess up our framing, so it's 1648 // more safe smuggling-wise to ignore. 1649 } 1650 } else if len(clens) > 1 { 1651 // TODO: care? unlike http/1, it won't mess up our framing, so it's 1652 // more safe smuggling-wise to ignore. 1653 } 1654 } 1655 1656 if streamEnded || isHead { 1657 res.Body = noBody 1658 return res, nil 1659 } 1660 1661 cs.bufPipe = pipe{b: &dataBuffer{expected: res.ContentLength}} 1662 cs.bytesRemain = res.ContentLength 1663 res.Body = transportResponseBody{cs} 1664 go cs.awaitRequestCancel(cs.req) 1665 1666 if cs.requestedGzip && res.Header.Get("Content-Encoding") == "gzip" { 1667 res.Header.Del("Content-Encoding") 1668 res.Header.Del("Content-Length") 1669 res.ContentLength = -1 1670 res.Body = &gzipReader{body: res.Body} 1671 setResponseUncompressed(res) 1672 } 1673 return res, nil 1674} 1675 1676func (rl *clientConnReadLoop) processTrailers(cs *clientStream, f *MetaHeadersFrame) error { 1677 if cs.pastTrailers { 1678 // Too many HEADERS frames for this stream. 1679 return ConnectionError(ErrCodeProtocol) 1680 } 1681 cs.pastTrailers = true 1682 if !f.StreamEnded() { 1683 // We expect that any headers for trailers also 1684 // has END_STREAM. 1685 return ConnectionError(ErrCodeProtocol) 1686 } 1687 if len(f.PseudoFields()) > 0 { 1688 // No pseudo header fields are defined for trailers. 1689 // TODO: ConnectionError might be overly harsh? Check. 1690 return ConnectionError(ErrCodeProtocol) 1691 } 1692 1693 trailer := make(http.Header) 1694 for _, hf := range f.RegularFields() { 1695 key := http.CanonicalHeaderKey(hf.Name) 1696 trailer[key] = append(trailer[key], hf.Value) 1697 } 1698 cs.trailer = trailer 1699 1700 rl.endStream(cs) 1701 return nil 1702} 1703 1704// transportResponseBody is the concrete type of Transport.RoundTrip's 1705// Response.Body. It is an io.ReadCloser. On Read, it reads from cs.body. 1706// On Close it sends RST_STREAM if EOF wasn't already seen. 1707type transportResponseBody struct { 1708 cs *clientStream 1709} 1710 1711func (b transportResponseBody) Read(p []byte) (n int, err error) { 1712 cs := b.cs 1713 cc := cs.cc 1714 1715 if cs.readErr != nil { 1716 return 0, cs.readErr 1717 } 1718 n, err = b.cs.bufPipe.Read(p) 1719 if cs.bytesRemain != -1 { 1720 if int64(n) > cs.bytesRemain { 1721 n = int(cs.bytesRemain) 1722 if err == nil { 1723 err = errors.New("net/http: server replied with more than declared Content-Length; truncated") 1724 cc.writeStreamReset(cs.ID, ErrCodeProtocol, err) 1725 } 1726 cs.readErr = err 1727 return int(cs.bytesRemain), err 1728 } 1729 cs.bytesRemain -= int64(n) 1730 if err == io.EOF && cs.bytesRemain > 0 { 1731 err = io.ErrUnexpectedEOF 1732 cs.readErr = err 1733 return n, err 1734 } 1735 } 1736 if n == 0 { 1737 // No flow control tokens to send back. 1738 return 1739 } 1740 1741 cc.mu.Lock() 1742 defer cc.mu.Unlock() 1743 1744 var connAdd, streamAdd int32 1745 // Check the conn-level first, before the stream-level. 1746 if v := cc.inflow.available(); v < transportDefaultConnFlow/2 { 1747 connAdd = transportDefaultConnFlow - v 1748 cc.inflow.add(connAdd) 1749 } 1750 if err == nil { // No need to refresh if the stream is over or failed. 1751 // Consider any buffered body data (read from the conn but not 1752 // consumed by the client) when computing flow control for this 1753 // stream. 1754 v := int(cs.inflow.available()) + cs.bufPipe.Len() 1755 if v < transportDefaultStreamFlow-transportDefaultStreamMinRefresh { 1756 streamAdd = int32(transportDefaultStreamFlow - v) 1757 cs.inflow.add(streamAdd) 1758 } 1759 } 1760 if connAdd != 0 || streamAdd != 0 { 1761 cc.wmu.Lock() 1762 defer cc.wmu.Unlock() 1763 if connAdd != 0 { 1764 cc.fr.WriteWindowUpdate(0, mustUint31(connAdd)) 1765 } 1766 if streamAdd != 0 { 1767 cc.fr.WriteWindowUpdate(cs.ID, mustUint31(streamAdd)) 1768 } 1769 cc.bw.Flush() 1770 } 1771 return 1772} 1773 1774var errClosedResponseBody = errors.New("http2: response body closed") 1775 1776func (b transportResponseBody) Close() error { 1777 cs := b.cs 1778 cc := cs.cc 1779 1780 serverSentStreamEnd := cs.bufPipe.Err() == io.EOF 1781 unread := cs.bufPipe.Len() 1782 1783 if unread > 0 || !serverSentStreamEnd { 1784 cc.mu.Lock() 1785 cc.wmu.Lock() 1786 if !serverSentStreamEnd { 1787 cc.fr.WriteRSTStream(cs.ID, ErrCodeCancel) 1788 cs.didReset = true 1789 } 1790 // Return connection-level flow control. 1791 if unread > 0 { 1792 cc.inflow.add(int32(unread)) 1793 cc.fr.WriteWindowUpdate(0, uint32(unread)) 1794 } 1795 cc.bw.Flush() 1796 cc.wmu.Unlock() 1797 cc.mu.Unlock() 1798 } 1799 1800 cs.bufPipe.BreakWithError(errClosedResponseBody) 1801 cc.forgetStreamID(cs.ID) 1802 return nil 1803} 1804 1805func (rl *clientConnReadLoop) processData(f *DataFrame) error { 1806 cc := rl.cc 1807 cs := cc.streamByID(f.StreamID, f.StreamEnded()) 1808 data := f.Data() 1809 if cs == nil { 1810 cc.mu.Lock() 1811 neverSent := cc.nextStreamID 1812 cc.mu.Unlock() 1813 if f.StreamID >= neverSent { 1814 // We never asked for this. 1815 cc.logf("http2: Transport received unsolicited DATA frame; closing connection") 1816 return ConnectionError(ErrCodeProtocol) 1817 } 1818 // We probably did ask for this, but canceled. Just ignore it. 1819 // TODO: be stricter here? only silently ignore things which 1820 // we canceled, but not things which were closed normally 1821 // by the peer? Tough without accumulating too much state. 1822 1823 // But at least return their flow control: 1824 if f.Length > 0 { 1825 cc.mu.Lock() 1826 cc.inflow.add(int32(f.Length)) 1827 cc.mu.Unlock() 1828 1829 cc.wmu.Lock() 1830 cc.fr.WriteWindowUpdate(0, uint32(f.Length)) 1831 cc.bw.Flush() 1832 cc.wmu.Unlock() 1833 } 1834 return nil 1835 } 1836 if !cs.firstByte { 1837 cc.logf("protocol error: received DATA before a HEADERS frame") 1838 rl.endStreamError(cs, StreamError{ 1839 StreamID: f.StreamID, 1840 Code: ErrCodeProtocol, 1841 }) 1842 return nil 1843 } 1844 if f.Length > 0 { 1845 if cs.req.Method == "HEAD" && len(data) > 0 { 1846 cc.logf("protocol error: received DATA on a HEAD request") 1847 rl.endStreamError(cs, StreamError{ 1848 StreamID: f.StreamID, 1849 Code: ErrCodeProtocol, 1850 }) 1851 return nil 1852 } 1853 // Check connection-level flow control. 1854 cc.mu.Lock() 1855 if cs.inflow.available() >= int32(f.Length) { 1856 cs.inflow.take(int32(f.Length)) 1857 } else { 1858 cc.mu.Unlock() 1859 return ConnectionError(ErrCodeFlowControl) 1860 } 1861 // Return any padded flow control now, since we won't 1862 // refund it later on body reads. 1863 var refund int 1864 if pad := int(f.Length) - len(data); pad > 0 { 1865 refund += pad 1866 } 1867 // Return len(data) now if the stream is already closed, 1868 // since data will never be read. 1869 didReset := cs.didReset 1870 if didReset { 1871 refund += len(data) 1872 } 1873 if refund > 0 { 1874 cc.inflow.add(int32(refund)) 1875 cc.wmu.Lock() 1876 cc.fr.WriteWindowUpdate(0, uint32(refund)) 1877 if !didReset { 1878 cs.inflow.add(int32(refund)) 1879 cc.fr.WriteWindowUpdate(cs.ID, uint32(refund)) 1880 } 1881 cc.bw.Flush() 1882 cc.wmu.Unlock() 1883 } 1884 cc.mu.Unlock() 1885 1886 if len(data) > 0 && !didReset { 1887 if _, err := cs.bufPipe.Write(data); err != nil { 1888 rl.endStreamError(cs, err) 1889 return err 1890 } 1891 } 1892 } 1893 1894 if f.StreamEnded() { 1895 rl.endStream(cs) 1896 } 1897 return nil 1898} 1899 1900var errInvalidTrailers = errors.New("http2: invalid trailers") 1901 1902func (rl *clientConnReadLoop) endStream(cs *clientStream) { 1903 // TODO: check that any declared content-length matches, like 1904 // server.go's (*stream).endStream method. 1905 rl.endStreamError(cs, nil) 1906} 1907 1908func (rl *clientConnReadLoop) endStreamError(cs *clientStream, err error) { 1909 var code func() 1910 if err == nil { 1911 err = io.EOF 1912 code = cs.copyTrailers 1913 } 1914 if isConnectionCloseRequest(cs.req) { 1915 rl.closeWhenIdle = true 1916 } 1917 cs.bufPipe.closeWithErrorAndCode(err, code) 1918 delete(rl.activeRes, cs.ID) 1919 1920 select { 1921 case cs.resc <- resAndError{err: err}: 1922 default: 1923 } 1924} 1925 1926func (cs *clientStream) copyTrailers() { 1927 for k, vv := range cs.trailer { 1928 t := cs.resTrailer 1929 if *t == nil { 1930 *t = make(http.Header) 1931 } 1932 (*t)[k] = vv 1933 } 1934} 1935 1936func (rl *clientConnReadLoop) processGoAway(f *GoAwayFrame) error { 1937 cc := rl.cc 1938 cc.t.connPool().MarkDead(cc) 1939 if f.ErrCode != 0 { 1940 // TODO: deal with GOAWAY more. particularly the error code 1941 cc.vlogf("transport got GOAWAY with error code = %v", f.ErrCode) 1942 } 1943 cc.setGoAway(f) 1944 return nil 1945} 1946 1947func (rl *clientConnReadLoop) processSettings(f *SettingsFrame) error { 1948 cc := rl.cc 1949 cc.mu.Lock() 1950 defer cc.mu.Unlock() 1951 1952 if f.IsAck() { 1953 if cc.wantSettingsAck { 1954 cc.wantSettingsAck = false 1955 return nil 1956 } 1957 return ConnectionError(ErrCodeProtocol) 1958 } 1959 1960 err := f.ForeachSetting(func(s Setting) error { 1961 switch s.ID { 1962 case SettingMaxFrameSize: 1963 cc.maxFrameSize = s.Val 1964 case SettingMaxConcurrentStreams: 1965 cc.maxConcurrentStreams = s.Val 1966 case SettingMaxHeaderListSize: 1967 cc.peerMaxHeaderListSize = uint64(s.Val) 1968 case SettingInitialWindowSize: 1969 // Values above the maximum flow-control 1970 // window size of 2^31-1 MUST be treated as a 1971 // connection error (Section 5.4.1) of type 1972 // FLOW_CONTROL_ERROR. 1973 if s.Val > math.MaxInt32 { 1974 return ConnectionError(ErrCodeFlowControl) 1975 } 1976 1977 // Adjust flow control of currently-open 1978 // frames by the difference of the old initial 1979 // window size and this one. 1980 delta := int32(s.Val) - int32(cc.initialWindowSize) 1981 for _, cs := range cc.streams { 1982 cs.flow.add(delta) 1983 } 1984 cc.cond.Broadcast() 1985 1986 cc.initialWindowSize = s.Val 1987 default: 1988 // TODO(bradfitz): handle more settings? SETTINGS_HEADER_TABLE_SIZE probably. 1989 cc.vlogf("Unhandled Setting: %v", s) 1990 } 1991 return nil 1992 }) 1993 if err != nil { 1994 return err 1995 } 1996 1997 cc.wmu.Lock() 1998 defer cc.wmu.Unlock() 1999 2000 cc.fr.WriteSettingsAck() 2001 cc.bw.Flush() 2002 return cc.werr 2003} 2004 2005func (rl *clientConnReadLoop) processWindowUpdate(f *WindowUpdateFrame) error { 2006 cc := rl.cc 2007 cs := cc.streamByID(f.StreamID, false) 2008 if f.StreamID != 0 && cs == nil { 2009 return nil 2010 } 2011 2012 cc.mu.Lock() 2013 defer cc.mu.Unlock() 2014 2015 fl := &cc.flow 2016 if cs != nil { 2017 fl = &cs.flow 2018 } 2019 if !fl.add(int32(f.Increment)) { 2020 return ConnectionError(ErrCodeFlowControl) 2021 } 2022 cc.cond.Broadcast() 2023 return nil 2024} 2025 2026func (rl *clientConnReadLoop) processResetStream(f *RSTStreamFrame) error { 2027 cs := rl.cc.streamByID(f.StreamID, true) 2028 if cs == nil { 2029 // TODO: return error if server tries to RST_STEAM an idle stream 2030 return nil 2031 } 2032 select { 2033 case <-cs.peerReset: 2034 // Already reset. 2035 // This is the only goroutine 2036 // which closes this, so there 2037 // isn't a race. 2038 default: 2039 err := streamError(cs.ID, f.ErrCode) 2040 cs.resetErr = err 2041 close(cs.peerReset) 2042 cs.bufPipe.CloseWithError(err) 2043 cs.cc.cond.Broadcast() // wake up checkResetOrDone via clientStream.awaitFlowControl 2044 } 2045 delete(rl.activeRes, cs.ID) 2046 return nil 2047} 2048 2049// Ping sends a PING frame to the server and waits for the ack. 2050// Public implementation is in go17.go and not_go17.go 2051func (cc *ClientConn) ping(ctx contextContext) error { 2052 c := make(chan struct{}) 2053 // Generate a random payload 2054 var p [8]byte 2055 for { 2056 if _, err := rand.Read(p[:]); err != nil { 2057 return err 2058 } 2059 cc.mu.Lock() 2060 // check for dup before insert 2061 if _, found := cc.pings[p]; !found { 2062 cc.pings[p] = c 2063 cc.mu.Unlock() 2064 break 2065 } 2066 cc.mu.Unlock() 2067 } 2068 cc.wmu.Lock() 2069 if err := cc.fr.WritePing(false, p); err != nil { 2070 cc.wmu.Unlock() 2071 return err 2072 } 2073 if err := cc.bw.Flush(); err != nil { 2074 cc.wmu.Unlock() 2075 return err 2076 } 2077 cc.wmu.Unlock() 2078 select { 2079 case <-c: 2080 return nil 2081 case <-ctx.Done(): 2082 return ctx.Err() 2083 case <-cc.readerDone: 2084 // connection closed 2085 return cc.readerErr 2086 } 2087} 2088 2089func (rl *clientConnReadLoop) processPing(f *PingFrame) error { 2090 if f.IsAck() { 2091 cc := rl.cc 2092 cc.mu.Lock() 2093 defer cc.mu.Unlock() 2094 // If ack, notify listener if any 2095 if c, ok := cc.pings[f.Data]; ok { 2096 close(c) 2097 delete(cc.pings, f.Data) 2098 } 2099 return nil 2100 } 2101 cc := rl.cc 2102 cc.wmu.Lock() 2103 defer cc.wmu.Unlock() 2104 if err := cc.fr.WritePing(true, f.Data); err != nil { 2105 return err 2106 } 2107 return cc.bw.Flush() 2108} 2109 2110func (rl *clientConnReadLoop) processPushPromise(f *PushPromiseFrame) error { 2111 // We told the peer we don't want them. 2112 // Spec says: 2113 // "PUSH_PROMISE MUST NOT be sent if the SETTINGS_ENABLE_PUSH 2114 // setting of the peer endpoint is set to 0. An endpoint that 2115 // has set this setting and has received acknowledgement MUST 2116 // treat the receipt of a PUSH_PROMISE frame as a connection 2117 // error (Section 5.4.1) of type PROTOCOL_ERROR." 2118 return ConnectionError(ErrCodeProtocol) 2119} 2120 2121func (cc *ClientConn) writeStreamReset(streamID uint32, code ErrCode, err error) { 2122 // TODO: map err to more interesting error codes, once the 2123 // HTTP community comes up with some. But currently for 2124 // RST_STREAM there's no equivalent to GOAWAY frame's debug 2125 // data, and the error codes are all pretty vague ("cancel"). 2126 cc.wmu.Lock() 2127 cc.fr.WriteRSTStream(streamID, code) 2128 cc.bw.Flush() 2129 cc.wmu.Unlock() 2130} 2131 2132var ( 2133 errResponseHeaderListSize = errors.New("http2: response header list larger than advertised limit") 2134 errRequestHeaderListSize = errors.New("http2: request header list larger than peer's advertised limit") 2135 errPseudoTrailers = errors.New("http2: invalid pseudo header in trailers") 2136) 2137 2138func (cc *ClientConn) logf(format string, args ...interface{}) { 2139 cc.t.logf(format, args...) 2140} 2141 2142func (cc *ClientConn) vlogf(format string, args ...interface{}) { 2143 cc.t.vlogf(format, args...) 2144} 2145 2146func (t *Transport) vlogf(format string, args ...interface{}) { 2147 if VerboseLogs { 2148 t.logf(format, args...) 2149 } 2150} 2151 2152func (t *Transport) logf(format string, args ...interface{}) { 2153 log.Printf(format, args...) 2154} 2155 2156var noBody io.ReadCloser = ioutil.NopCloser(bytes.NewReader(nil)) 2157 2158func strSliceContains(ss []string, s string) bool { 2159 for _, v := range ss { 2160 if v == s { 2161 return true 2162 } 2163 } 2164 return false 2165} 2166 2167type erringRoundTripper struct{ err error } 2168 2169func (rt erringRoundTripper) RoundTrip(*http.Request) (*http.Response, error) { return nil, rt.err } 2170 2171// gzipReader wraps a response body so it can lazily 2172// call gzip.NewReader on the first call to Read 2173type gzipReader struct { 2174 body io.ReadCloser // underlying Response.Body 2175 zr *gzip.Reader // lazily-initialized gzip reader 2176 zerr error // sticky error 2177} 2178 2179func (gz *gzipReader) Read(p []byte) (n int, err error) { 2180 if gz.zerr != nil { 2181 return 0, gz.zerr 2182 } 2183 if gz.zr == nil { 2184 gz.zr, err = gzip.NewReader(gz.body) 2185 if err != nil { 2186 gz.zerr = err 2187 return 0, err 2188 } 2189 } 2190 return gz.zr.Read(p) 2191} 2192 2193func (gz *gzipReader) Close() error { 2194 return gz.body.Close() 2195} 2196 2197type errorReader struct{ err error } 2198 2199func (r errorReader) Read(p []byte) (int, error) { return 0, r.err } 2200 2201// bodyWriterState encapsulates various state around the Transport's writing 2202// of the request body, particularly regarding doing delayed writes of the body 2203// when the request contains "Expect: 100-continue". 2204type bodyWriterState struct { 2205 cs *clientStream 2206 timer *time.Timer // if non-nil, we're doing a delayed write 2207 fnonce *sync.Once // to call fn with 2208 fn func() // the code to run in the goroutine, writing the body 2209 resc chan error // result of fn's execution 2210 delay time.Duration // how long we should delay a delayed write for 2211} 2212 2213func (t *Transport) getBodyWriterState(cs *clientStream, body io.Reader) (s bodyWriterState) { 2214 s.cs = cs 2215 if body == nil { 2216 return 2217 } 2218 resc := make(chan error, 1) 2219 s.resc = resc 2220 s.fn = func() { 2221 cs.cc.mu.Lock() 2222 cs.startedWrite = true 2223 cs.cc.mu.Unlock() 2224 resc <- cs.writeRequestBody(body, cs.req.Body) 2225 } 2226 s.delay = t.expectContinueTimeout() 2227 if s.delay == 0 || 2228 !httplex.HeaderValuesContainsToken( 2229 cs.req.Header["Expect"], 2230 "100-continue") { 2231 return 2232 } 2233 s.fnonce = new(sync.Once) 2234 2235 // Arm the timer with a very large duration, which we'll 2236 // intentionally lower later. It has to be large now because 2237 // we need a handle to it before writing the headers, but the 2238 // s.delay value is defined to not start until after the 2239 // request headers were written. 2240 const hugeDuration = 365 * 24 * time.Hour 2241 s.timer = time.AfterFunc(hugeDuration, func() { 2242 s.fnonce.Do(s.fn) 2243 }) 2244 return 2245} 2246 2247func (s bodyWriterState) cancel() { 2248 if s.timer != nil { 2249 s.timer.Stop() 2250 } 2251} 2252 2253func (s bodyWriterState) on100() { 2254 if s.timer == nil { 2255 // If we didn't do a delayed write, ignore the server's 2256 // bogus 100 continue response. 2257 return 2258 } 2259 s.timer.Stop() 2260 go func() { s.fnonce.Do(s.fn) }() 2261} 2262 2263// scheduleBodyWrite starts writing the body, either immediately (in 2264// the common case) or after the delay timeout. It should not be 2265// called until after the headers have been written. 2266func (s bodyWriterState) scheduleBodyWrite() { 2267 if s.timer == nil { 2268 // We're not doing a delayed write (see 2269 // getBodyWriterState), so just start the writing 2270 // goroutine immediately. 2271 go s.fn() 2272 return 2273 } 2274 traceWait100Continue(s.cs.trace) 2275 if s.timer.Stop() { 2276 s.timer.Reset(s.delay) 2277 } 2278} 2279 2280// isConnectionCloseRequest reports whether req should use its own 2281// connection for a single request and then close the connection. 2282func isConnectionCloseRequest(req *http.Request) bool { 2283 return req.Close || httplex.HeaderValuesContainsToken(req.Header["Connection"], "close") 2284} 2285