transport.go 68 KB

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