server.go 44 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112111311141115111611171118111911201121112211231124112511261127112811291130113111321133113411351136113711381139114011411142114311441145114611471148114911501151115211531154115511561157115811591160116111621163116411651166116711681169117011711172117311741175117611771178117911801181118211831184118511861187118811891190119111921193119411951196119711981199120012011202120312041205120612071208120912101211121212131214121512161217121812191220122112221223122412251226122712281229123012311232123312341235123612371238123912401241124212431244124512461247124812491250125112521253125412551256125712581259126012611262126312641265126612671268126912701271127212731274127512761277127812791280128112821283128412851286128712881289129012911292129312941295129612971298129913001301130213031304130513061307130813091310131113121313131413151316131713181319132013211322132313241325132613271328132913301331133213331334133513361337133813391340134113421343134413451346134713481349135013511352135313541355135613571358135913601361136213631364136513661367136813691370137113721373137413751376137713781379138013811382138313841385138613871388138913901391139213931394139513961397139813991400140114021403140414051406140714081409141014111412141314141415141614171418141914201421142214231424142514261427142814291430143114321433143414351436143714381439144014411442144314441445144614471448144914501451145214531454145514561457145814591460146114621463146414651466146714681469147014711472147314741475147614771478147914801481148214831484148514861487148814891490149114921493149414951496149714981499150015011502150315041505150615071508150915101511151215131514151515161517151815191520152115221523152415251526152715281529153015311532153315341535153615371538153915401541154215431544154515461547154815491550155115521553155415551556155715581559156015611562156315641565156615671568156915701571157215731574157515761577157815791580158115821583
  1. // Copyright 2014 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. // See https://code.google.com/p/go/source/browse/CONTRIBUTORS
  5. // Licensed under the same terms as Go itself:
  6. // https://code.google.com/p/go/source/browse/LICENSE
  7. package http2
  8. import (
  9. "bufio"
  10. "bytes"
  11. "crypto/tls"
  12. "errors"
  13. "fmt"
  14. "io"
  15. "log"
  16. "net"
  17. "net/http"
  18. "net/url"
  19. "strconv"
  20. "strings"
  21. "sync"
  22. "time"
  23. "github.com/bradfitz/http2/hpack"
  24. )
  25. const (
  26. prefaceTimeout = 5 * time.Second
  27. firstSettingsTimeout = 2 * time.Second // should be in-flight with preface anyway
  28. )
  29. // TODO: finish GOAWAY support. Consider each incoming frame type and
  30. // whether it should be ignored during a shutdown race.
  31. // TODO: (edge case?) if peer sends a SETTINGS frame with e.g. a
  32. // SETTINGS_MAX_FRAME_SIZE that's lower than what we had before,
  33. // before we ACK it we have to make sure all currently-active streams
  34. // know about that and don't have existing too-large frames in flight?
  35. // Perhaps the settings processing should just wait for new frame to
  36. // be in-flight and then the frame scheduler in the serve goroutine
  37. // will be responsible for splitting things.
  38. // TODO: send PING frames to idle clients and disconnect them if no
  39. // reply
  40. // Server is an HTTP/2 server.
  41. type Server struct {
  42. // MaxStreams optionally ...
  43. MaxStreams int
  44. // MaxReadFrameSize optionally specifies the largest frame
  45. // this server is willing to read. A valid value is between
  46. // 16k and 16M, inclusive.
  47. MaxReadFrameSize uint32
  48. }
  49. func (s *Server) maxReadFrameSize() uint32 {
  50. if v := s.MaxReadFrameSize; v >= minMaxFrameSize && v <= maxFrameSize {
  51. return v
  52. }
  53. return defaultMaxReadFrameSize
  54. }
  55. var testHookOnConn func() // for testing
  56. // ConfigureServer adds HTTP/2 support to a net/http Server.
  57. //
  58. // The configuration conf may be nil.
  59. //
  60. // ConfigureServer must be called before s begins serving.
  61. func ConfigureServer(s *http.Server, conf *Server) {
  62. if conf == nil {
  63. conf = new(Server)
  64. }
  65. if s.TLSConfig == nil {
  66. s.TLSConfig = new(tls.Config)
  67. }
  68. haveNPN := false
  69. for _, p := range s.TLSConfig.NextProtos {
  70. if p == npnProto {
  71. haveNPN = true
  72. break
  73. }
  74. }
  75. if !haveNPN {
  76. s.TLSConfig.NextProtos = append(s.TLSConfig.NextProtos, npnProto)
  77. }
  78. if s.TLSNextProto == nil {
  79. s.TLSNextProto = map[string]func(*http.Server, *tls.Conn, http.Handler){}
  80. }
  81. s.TLSNextProto[npnProto] = func(hs *http.Server, c *tls.Conn, h http.Handler) {
  82. if testHookOnConn != nil {
  83. testHookOnConn()
  84. }
  85. conf.handleConn(hs, c, h)
  86. }
  87. }
  88. var testHookGetServerConn func(*serverConn)
  89. func (srv *Server) handleConn(hs *http.Server, c net.Conn, h http.Handler) {
  90. // TODO: write to a (custom?) buffered writer that can
  91. // alternate when it's in buffered mode.
  92. fr := NewFramer(c, c)
  93. fr.SetMaxReadFrameSize(srv.maxReadFrameSize())
  94. sc := &serverConn{
  95. srv: srv,
  96. hs: hs,
  97. conn: c,
  98. handler: h,
  99. framer: fr,
  100. streams: make(map[uint32]*stream),
  101. readFrameCh: make(chan frameAndGate),
  102. readFrameErrCh: make(chan error, 1), // must be buffered for 1
  103. wantWriteFrameCh: make(chan frameWriteMsg, 8),
  104. writeFrameCh: make(chan frameWriteMsg, 1), // may be 0 or 1, but more is useless. (max 1 in flight)
  105. wroteFrameCh: make(chan struct{}, 1), // TODO: consider 0. will deadlock currently in sendFrameWrite in sentReset case
  106. flow: newFlow(initialWindowSize),
  107. doneServing: make(chan struct{}),
  108. maxWriteFrameSize: initialMaxFrameSize,
  109. initialWindowSize: initialWindowSize,
  110. headerTableSize: initialHeaderTableSize,
  111. maxConcurrentStreams: -1, // no limit
  112. serveG: newGoroutineLock(),
  113. pushEnabled: true,
  114. }
  115. sc.hpackEncoder = hpack.NewEncoder(&sc.headerWriteBuf)
  116. sc.hpackDecoder = hpack.NewDecoder(initialHeaderTableSize, sc.onNewHeaderField)
  117. if hook := testHookGetServerConn; hook != nil {
  118. hook(sc)
  119. }
  120. sc.serve()
  121. }
  122. // frameAndGates coordinates the readFrames and serve
  123. // goroutines. Because the Framer interface only permits the most
  124. // recently-read Frame from being accessed, the readFrames goroutine
  125. // blocks until it has a frame, passes it to serve, and then waits for
  126. // serve to be done with it before reading the next one.
  127. type frameAndGate struct {
  128. f Frame
  129. g gate
  130. }
  131. type serverConn struct {
  132. // Immutable:
  133. srv *Server
  134. hs *http.Server
  135. conn net.Conn
  136. handler http.Handler
  137. framer *Framer
  138. hpackDecoder *hpack.Decoder
  139. doneServing chan struct{} // closed when serverConn.serve ends
  140. readFrameCh chan frameAndGate // written by serverConn.readFrames
  141. readFrameErrCh chan error
  142. wantWriteFrameCh chan frameWriteMsg // from handlers -> serve
  143. writeFrameCh chan frameWriteMsg // from serve -> writeFrames
  144. wroteFrameCh chan struct{} // from writeFrames -> serve, tickles more sends on writeFrameCh
  145. testHookCh chan func() // code to run on the serve loop
  146. serveG goroutineLock // used to verify funcs are on serve()
  147. writeG goroutineLock // used to verify things running on writeLoop
  148. flow *flow // connection-wide (not stream-specific) flow control
  149. // Everything following is owned by the serve loop; use serveG.check():
  150. pushEnabled bool
  151. sawFirstSettings bool // got the initial SETTINGS frame after the preface
  152. needToSendSettingsAck bool
  153. maxStreamID uint32 // max ever seen
  154. streams map[uint32]*stream
  155. maxWriteFrameSize uint32
  156. initialWindowSize int32
  157. headerTableSize uint32
  158. maxHeaderListSize uint32 // zero means unknown (default)
  159. maxConcurrentStreams int64 // negative means no limit.
  160. canonHeader map[string]string // http2-lower-case -> Go-Canonical-Case
  161. req requestParam // non-zero while reading request headers
  162. writingFrame bool // sent on writeFrameCh but haven't heard back on wroteFrameCh yet
  163. writeQueue []frameWriteMsg // TODO: proper scheduler, not a queue
  164. inGoAway bool // we've started to or sent GOAWAY
  165. needToSendGoAway bool // we need to schedule a GOAWAY frame write
  166. goAwayCode ErrCode
  167. shutdownTimerCh <-chan time.Time
  168. shutdownTimer *time.Timer
  169. // Owned by the writeFrames goroutine; use writeG.check():
  170. headerWriteBuf bytes.Buffer
  171. hpackEncoder *hpack.Encoder
  172. }
  173. // requestParam is the state of the next request, initialized over
  174. // potentially several frames HEADERS + zero or more CONTINUATION
  175. // frames.
  176. type requestParam struct {
  177. // stream is non-nil if we're reading (HEADER or CONTINUATION)
  178. // frames for a request (but not DATA).
  179. stream *stream
  180. header http.Header
  181. method, path string
  182. scheme, authority string
  183. sawRegularHeader bool // saw a non-pseudo header already
  184. invalidHeader bool // an invalid header was seen
  185. }
  186. // stream represents a stream. This is the minimal metadata needed by
  187. // the serve goroutine. Most of the actual stream state is owned by
  188. // the http.Handler's goroutine in the responseWriter. Because the
  189. // responseWriter's responseWriterState is recycled at the end of a
  190. // handler, this struct intentionally has no pointer to the
  191. // *responseWriter{,State} itself, as the Handler ending nils out the
  192. // responseWriter's state field.
  193. type stream struct {
  194. // immutable:
  195. id uint32
  196. conn *serverConn
  197. flow *flow // limits writing from Handler to client
  198. body *pipe // non-nil if expecting DATA frames
  199. cw closeWaiter // closed wait stream transitions to closed state
  200. // owned by serverConn's serve loop:
  201. state streamState
  202. bodyBytes int64 // body bytes seen so far
  203. declBodyBytes int64 // or -1 if undeclared
  204. sentReset bool // only true once detached from streams map
  205. gotReset bool // only true once detacted from streams map
  206. }
  207. func (sc *serverConn) state(streamID uint32) streamState {
  208. sc.serveG.check()
  209. // http://http2.github.io/http2-spec/#rfc.section.5.1
  210. if st, ok := sc.streams[streamID]; ok {
  211. return st.state
  212. }
  213. // "The first use of a new stream identifier implicitly closes all
  214. // streams in the "idle" state that might have been initiated by
  215. // that peer with a lower-valued stream identifier. For example, if
  216. // a client sends a HEADERS frame on stream 7 without ever sending a
  217. // frame on stream 5, then stream 5 transitions to the "closed"
  218. // state when the first frame for stream 7 is sent or received."
  219. if streamID <= sc.maxStreamID {
  220. return stateClosed
  221. }
  222. return stateIdle
  223. }
  224. func (sc *serverConn) vlogf(format string, args ...interface{}) {
  225. if VerboseLogs {
  226. sc.logf(format, args...)
  227. }
  228. }
  229. func (sc *serverConn) logf(format string, args ...interface{}) {
  230. if lg := sc.hs.ErrorLog; lg != nil {
  231. lg.Printf(format, args...)
  232. } else {
  233. log.Printf(format, args...)
  234. }
  235. }
  236. func (sc *serverConn) condlogf(err error, format string, args ...interface{}) {
  237. if err == nil {
  238. return
  239. }
  240. str := err.Error()
  241. if err == io.EOF || strings.Contains(str, "use of closed network connection") {
  242. // Boring, expected errors.
  243. sc.vlogf(format, args...)
  244. } else {
  245. sc.logf(format, args...)
  246. }
  247. }
  248. func (sc *serverConn) onNewHeaderField(f hpack.HeaderField) {
  249. sc.serveG.check()
  250. switch {
  251. case !validHeader(f.Name):
  252. sc.req.invalidHeader = true
  253. case strings.HasPrefix(f.Name, ":"):
  254. if sc.req.sawRegularHeader {
  255. sc.logf("pseudo-header after regular header")
  256. sc.req.invalidHeader = true
  257. return
  258. }
  259. var dst *string
  260. switch f.Name {
  261. case ":method":
  262. dst = &sc.req.method
  263. case ":path":
  264. dst = &sc.req.path
  265. case ":scheme":
  266. dst = &sc.req.scheme
  267. case ":authority":
  268. dst = &sc.req.authority
  269. default:
  270. // 8.1.2.1 Pseudo-Header Fields
  271. // "Endpoints MUST treat a request or response
  272. // that contains undefined or invalid
  273. // pseudo-header fields as malformed (Section
  274. // 8.1.2.6)."
  275. sc.logf("invalid pseudo-header %q", f.Name)
  276. sc.req.invalidHeader = true
  277. return
  278. }
  279. if *dst != "" {
  280. sc.logf("duplicate pseudo-header %q sent", f.Name)
  281. sc.req.invalidHeader = true
  282. return
  283. }
  284. *dst = f.Value
  285. case f.Name == "cookie":
  286. sc.req.sawRegularHeader = true
  287. if s, ok := sc.req.header["Cookie"]; ok && len(s) == 1 {
  288. s[0] = s[0] + "; " + f.Value
  289. } else {
  290. sc.req.header.Add("Cookie", f.Value)
  291. }
  292. default:
  293. sc.req.sawRegularHeader = true
  294. sc.req.header.Add(sc.canonicalHeader(f.Name), f.Value)
  295. }
  296. }
  297. func (sc *serverConn) canonicalHeader(v string) string {
  298. sc.serveG.check()
  299. cv, ok := commonCanonHeader[v]
  300. if ok {
  301. return cv
  302. }
  303. cv, ok = sc.canonHeader[v]
  304. if ok {
  305. return cv
  306. }
  307. if sc.canonHeader == nil {
  308. sc.canonHeader = make(map[string]string)
  309. }
  310. cv = http.CanonicalHeaderKey(v)
  311. sc.canonHeader[v] = cv
  312. return cv
  313. }
  314. // readFrames is the loop that reads incoming frames.
  315. // It's run on its own goroutine.
  316. func (sc *serverConn) readFrames() {
  317. g := make(gate, 1)
  318. for {
  319. f, err := sc.framer.ReadFrame()
  320. if err != nil {
  321. sc.readFrameErrCh <- err // BEFORE the close
  322. close(sc.readFrameCh)
  323. return
  324. }
  325. sc.readFrameCh <- frameAndGate{f, g}
  326. g.Wait()
  327. }
  328. }
  329. // writeFrames is the loop that writes frames to the peer
  330. // and is responsible for prioritization and buffering.
  331. // It's run on its own goroutine.
  332. func (sc *serverConn) writeFrames() {
  333. sc.writeG = newGoroutineLock()
  334. for wm := range sc.writeFrameCh {
  335. err := wm.write(sc, wm.v)
  336. if ch := wm.done; ch != nil {
  337. select {
  338. case ch <- err:
  339. default:
  340. panic(fmt.Sprintf("unbuffered done channel passed in for type %T", wm.v))
  341. }
  342. }
  343. sc.wroteFrameCh <- struct{}{} // tickle frame selection scheduler
  344. }
  345. }
  346. var errClientDisconnected = errors.New("client disconnected")
  347. func (sc *serverConn) closeAllStreamsOnConnClose() {
  348. sc.serveG.check()
  349. for _, st := range sc.streams {
  350. sc.closeStream(st, errClientDisconnected)
  351. }
  352. }
  353. func (sc *serverConn) stopShutdownTimer() {
  354. sc.serveG.check()
  355. if t := sc.shutdownTimer; t != nil {
  356. t.Stop()
  357. }
  358. }
  359. func (sc *serverConn) serve() {
  360. sc.serveG.check()
  361. defer sc.conn.Close()
  362. defer sc.closeAllStreamsOnConnClose()
  363. defer sc.stopShutdownTimer()
  364. defer close(sc.doneServing) // unblocks handlers trying to send
  365. defer close(sc.writeFrameCh) // stop the writeFrames loop
  366. sc.vlogf("HTTP/2 connection from %v on %p", sc.conn.RemoteAddr(), sc.hs)
  367. sc.enqueueFrameWrite(frameWriteMsg{write: (*serverConn).sendInitialSettings})
  368. if err := sc.readPreface(); err != nil {
  369. sc.condlogf(err, "Error reading preface from client %v: %v", sc.conn.RemoteAddr(), err)
  370. return
  371. }
  372. go sc.readFrames() // closed by defer sc.conn.Close above
  373. go sc.writeFrames() // source closed in stopServing
  374. settingsTimer := time.NewTimer(firstSettingsTimeout)
  375. for {
  376. select {
  377. case wm := <-sc.wantWriteFrameCh:
  378. sc.enqueueFrameWrite(wm)
  379. case <-sc.wroteFrameCh:
  380. sc.writingFrame = false
  381. sc.scheduleFrameWrite()
  382. case fg, ok := <-sc.readFrameCh:
  383. if !ok {
  384. sc.readFrameCh = nil
  385. }
  386. if !sc.processFrameFromReader(fg, ok) {
  387. return
  388. }
  389. if settingsTimer.C != nil {
  390. settingsTimer.Stop()
  391. settingsTimer.C = nil
  392. }
  393. case <-settingsTimer.C:
  394. sc.logf("timeout waiting for SETTINGS frames from %v", sc.conn.RemoteAddr())
  395. return
  396. case <-sc.shutdownTimerCh:
  397. sc.vlogf("GOAWAY close timer fired; closing conn from %v", sc.conn.RemoteAddr())
  398. return
  399. case fn := <-sc.testHookCh:
  400. fn()
  401. }
  402. }
  403. }
  404. func (sc *serverConn) sendInitialSettings(_ interface{}) error {
  405. sc.writeG.check()
  406. return sc.framer.WriteSettings(
  407. Setting{SettingMaxFrameSize, sc.srv.maxReadFrameSize()},
  408. Setting{SettingMaxConcurrentStreams, 250}, // TODO: tunable?
  409. /* TODO: more actual settings */
  410. )
  411. }
  412. // readPreface reads the ClientPreface greeting from the peer
  413. // or returns an error on timeout or an invalid greeting.
  414. func (sc *serverConn) readPreface() error {
  415. errc := make(chan error, 1)
  416. go func() {
  417. // Read the client preface
  418. buf := make([]byte, len(ClientPreface))
  419. // TODO: timeout reading from the client
  420. if _, err := io.ReadFull(sc.conn, buf); err != nil {
  421. errc <- err
  422. } else if !bytes.Equal(buf, clientPreface) {
  423. errc <- fmt.Errorf("bogus greeting %q", buf)
  424. } else {
  425. errc <- nil
  426. }
  427. }()
  428. timer := time.NewTimer(5 * time.Second) // TODO: configurable on *Server?
  429. defer timer.Stop()
  430. select {
  431. case <-timer.C:
  432. return errors.New("timeout waiting for client preface")
  433. case err := <-errc:
  434. if err == nil {
  435. sc.vlogf("client %v said hello", sc.conn.RemoteAddr())
  436. }
  437. return err
  438. }
  439. }
  440. // should be called from non-serve() goroutines, otherwise the ends may deadlock
  441. // the serve loop. (it's only buffered a little bit).
  442. func (sc *serverConn) writeFrame(wm frameWriteMsg) {
  443. sc.serveG.checkNotOn() // NOT
  444. select {
  445. case sc.wantWriteFrameCh <- wm:
  446. case <-sc.doneServing:
  447. // Client has closed their connection to the server.
  448. }
  449. }
  450. func (sc *serverConn) enqueueFrameWrite(wm frameWriteMsg) {
  451. sc.serveG.check()
  452. // Fast path for common case:
  453. if !sc.writingFrame {
  454. sc.sendFrameWrite(wm)
  455. return
  456. }
  457. sc.writeQueue = append(sc.writeQueue, wm) // TODO: proper scheduler
  458. }
  459. // sendFrameWrite sends a frame to the writeFrames goroutine.
  460. // Only one frame can be in-flight at a time.
  461. // sendFrameWrite also updates stream state right before the frame is
  462. // sent to be written.
  463. func (sc *serverConn) sendFrameWrite(wm frameWriteMsg) {
  464. sc.serveG.check()
  465. if sc.writingFrame {
  466. panic("invariant")
  467. }
  468. st := wm.stream
  469. if st != nil {
  470. switch st.state {
  471. case stateHalfClosedLocal:
  472. panic("internal error: attempt to send frame on half-closed-local stream")
  473. case stateClosed:
  474. if st.sentReset || st.gotReset {
  475. // Skip this frame. But fake the frame write to reschedule:
  476. sc.wroteFrameCh <- struct{}{}
  477. return
  478. }
  479. panic("internal error: attempt to send a frame on a closed stream")
  480. }
  481. }
  482. sc.writingFrame = true
  483. if wm.endStream {
  484. if st == nil {
  485. panic("nil stream with endStream set")
  486. }
  487. switch st.state {
  488. case stateOpen:
  489. st.state = stateHalfClosedLocal
  490. case stateHalfClosedRemote:
  491. sc.closeStream(st, nil)
  492. }
  493. }
  494. sc.writeFrameCh <- wm
  495. }
  496. func (sc *serverConn) enqueueSettingsAck() {
  497. sc.serveG.check()
  498. if !sc.writingFrame {
  499. sc.needToSendSettingsAck = false
  500. sc.writeFrameCh <- frameWriteMsg{write: (*serverConn).writeSettingsAck}
  501. return
  502. }
  503. sc.needToSendSettingsAck = true
  504. }
  505. func (sc *serverConn) scheduleFrameWrite() {
  506. sc.serveG.check()
  507. if sc.writingFrame {
  508. return
  509. }
  510. if sc.needToSendGoAway {
  511. sc.needToSendGoAway = false
  512. sc.sendFrameWrite(frameWriteMsg{
  513. write: (*serverConn).writeGoAwayFrame,
  514. v: &goAwayParams{
  515. maxStreamID: sc.maxStreamID,
  516. code: sc.goAwayCode,
  517. },
  518. })
  519. return
  520. }
  521. if sc.inGoAway {
  522. // No more frames after we've sent GOAWAY.
  523. return
  524. }
  525. if sc.needToSendSettingsAck {
  526. sc.enqueueSettingsAck()
  527. return
  528. }
  529. if len(sc.writeQueue) == 0 {
  530. // TODO: flush Framer's underlying buffered writer, once that's added
  531. return
  532. }
  533. // TODO: proper scheduler
  534. wm := sc.writeQueue[0]
  535. // shift it all down. kinda lame. will be removed later anyway.
  536. copy(sc.writeQueue, sc.writeQueue[1:])
  537. sc.writeQueue = sc.writeQueue[:len(sc.writeQueue)-1]
  538. // TODO: if wm is a data frame, make sure it's not too big
  539. // (because a SETTINGS frame changed our max frame size while
  540. // a stream was open and writing) and cut it up into smaller
  541. // bits.
  542. sc.sendFrameWrite(wm)
  543. }
  544. func (sc *serverConn) goAway(code ErrCode) {
  545. sc.serveG.check()
  546. if sc.inGoAway {
  547. return
  548. }
  549. if code != ErrCodeNo {
  550. sc.shutDownIn(250 * time.Millisecond)
  551. } else {
  552. // TODO: configurable
  553. sc.shutDownIn(1 * time.Second)
  554. }
  555. sc.inGoAway = true
  556. sc.needToSendGoAway = true
  557. sc.goAwayCode = code
  558. sc.scheduleFrameWrite()
  559. }
  560. func (sc *serverConn) shutDownIn(d time.Duration) {
  561. sc.serveG.check()
  562. sc.shutdownTimer = time.NewTimer(d)
  563. sc.shutdownTimerCh = sc.shutdownTimer.C
  564. }
  565. type goAwayParams struct {
  566. maxStreamID uint32
  567. code ErrCode
  568. }
  569. func (sc *serverConn) writeGoAwayFrame(v interface{}) error {
  570. sc.writeG.check()
  571. p := v.(*goAwayParams)
  572. err := sc.framer.WriteGoAway(p.maxStreamID, p.code, nil)
  573. if p.code != 0 {
  574. // TODO: flush any buffer, if we add a buffering writing
  575. // Sleep a bit to give the peer a bit of time to read the
  576. // GOAWAY before potentially getting a TCP RST packet:
  577. time.Sleep(50 * time.Millisecond)
  578. sc.conn.Close()
  579. }
  580. return err
  581. }
  582. func (sc *serverConn) resetStreamInLoop(se StreamError) {
  583. sc.serveG.check()
  584. st, ok := sc.streams[se.StreamID]
  585. if !ok {
  586. panic(fmt.Sprintf("invariant. closing non-open stream %d", se.StreamID))
  587. }
  588. sc.enqueueFrameWrite(frameWriteMsg{
  589. write: (*serverConn).writeRSTStreamFrame,
  590. v: &se,
  591. })
  592. st.sentReset = true
  593. sc.closeStream(st, se)
  594. }
  595. func (sc *serverConn) writeRSTStreamFrame(v interface{}) error {
  596. sc.writeG.check()
  597. se := v.(*StreamError)
  598. return sc.framer.WriteRSTStream(se.StreamID, se.Code)
  599. }
  600. func (sc *serverConn) curHeaderStreamID() uint32 {
  601. sc.serveG.check()
  602. st := sc.req.stream
  603. if st == nil {
  604. return 0
  605. }
  606. return st.id
  607. }
  608. // processFrameFromReader processes the serve loop's read from readFrameCh from the
  609. // frame-reading goroutine.
  610. // processFrameFromReader returns whether the connection should be kept open.
  611. func (sc *serverConn) processFrameFromReader(fg frameAndGate, fgValid bool) bool {
  612. sc.serveG.check()
  613. if !fgValid {
  614. err := <-sc.readFrameErrCh
  615. if err == ErrFrameTooLarge {
  616. sc.goAway(ErrCodeFrameSize)
  617. return true // goAway will close the loop
  618. }
  619. if err != io.EOF {
  620. errstr := err.Error()
  621. if !strings.Contains(errstr, "use of closed network connection") {
  622. sc.logf("client %s stopped sending frames: %v", sc.conn.RemoteAddr(), errstr)
  623. }
  624. }
  625. // TODO: could we also get into this state if the peer does a half close (e.g. CloseWrite)
  626. // because they're done sending frames but they're still wanting our open replies?
  627. // Investigate.
  628. return false
  629. }
  630. f := fg.f
  631. sc.vlogf("got %v: %#v", f.Header(), f)
  632. err := sc.processFrame(f)
  633. fg.g.Done() // unblock the readFrames goroutine
  634. if err == nil {
  635. return true
  636. }
  637. switch ev := err.(type) {
  638. case StreamError:
  639. sc.resetStreamInLoop(ev)
  640. return true
  641. case goAwayFlowError:
  642. sc.goAway(ErrCodeFlowControl)
  643. return true
  644. case ConnectionError:
  645. sc.logf("%v: %v", sc.conn.RemoteAddr(), ev)
  646. sc.goAway(ErrCode(ev))
  647. return true // goAway will handle shutdown
  648. default:
  649. sc.logf("Disconnection due to other error: %v", err)
  650. }
  651. return false
  652. }
  653. func (sc *serverConn) processFrame(f Frame) error {
  654. sc.serveG.check()
  655. // First frame received must be SETTINGS.
  656. if !sc.sawFirstSettings {
  657. if _, ok := f.(*SettingsFrame); !ok {
  658. return ConnectionError(ErrCodeProtocol)
  659. }
  660. sc.sawFirstSettings = true
  661. }
  662. if s := sc.curHeaderStreamID(); s != 0 {
  663. if cf, ok := f.(*ContinuationFrame); !ok {
  664. return ConnectionError(ErrCodeProtocol)
  665. } else if cf.Header().StreamID != s {
  666. return ConnectionError(ErrCodeProtocol)
  667. }
  668. }
  669. switch f := f.(type) {
  670. case *SettingsFrame:
  671. return sc.processSettings(f)
  672. case *HeadersFrame:
  673. return sc.processHeaders(f)
  674. case *ContinuationFrame:
  675. return sc.processContinuation(f)
  676. case *WindowUpdateFrame:
  677. return sc.processWindowUpdate(f)
  678. case *PingFrame:
  679. return sc.processPing(f)
  680. case *DataFrame:
  681. return sc.processData(f)
  682. case *RSTStreamFrame:
  683. return sc.processResetStream(f)
  684. default:
  685. log.Printf("Ignoring frame: %v", f.Header())
  686. return nil
  687. }
  688. }
  689. func (sc *serverConn) processPing(f *PingFrame) error {
  690. sc.serveG.check()
  691. if f.Flags.Has(FlagSettingsAck) {
  692. // 6.7 PING: " An endpoint MUST NOT respond to PING frames
  693. // containing this flag."
  694. return nil
  695. }
  696. if f.StreamID != 0 {
  697. // "PING frames are not associated with any individual
  698. // stream. If a PING frame is received with a stream
  699. // identifier field value other than 0x0, the recipient MUST
  700. // respond with a connection error (Section 5.4.1) of type
  701. // PROTOCOL_ERROR."
  702. return ConnectionError(ErrCodeProtocol)
  703. }
  704. sc.enqueueFrameWrite(frameWriteMsg{
  705. write: (*serverConn).writePingAck,
  706. v: f,
  707. })
  708. return nil
  709. }
  710. func (sc *serverConn) writePingAck(v interface{}) error {
  711. sc.writeG.check()
  712. pf := v.(*PingFrame) // contains the data we need to write back
  713. return sc.framer.WritePing(true, pf.Data)
  714. }
  715. func (sc *serverConn) processWindowUpdate(f *WindowUpdateFrame) error {
  716. sc.serveG.check()
  717. switch {
  718. case f.StreamID != 0: // stream-level flow control
  719. st := sc.streams[f.StreamID]
  720. if st == nil {
  721. // "WINDOW_UPDATE can be sent by a peer that has sent a
  722. // frame bearing the END_STREAM flag. This means that a
  723. // receiver could receive a WINDOW_UPDATE frame on a "half
  724. // closed (remote)" or "closed" stream. A receiver MUST
  725. // NOT treat this as an error, see Section 5.1."
  726. return nil
  727. }
  728. if !st.flow.add(int32(f.Increment)) {
  729. return StreamError{f.StreamID, ErrCodeFlowControl}
  730. }
  731. default: // connection-level flow control
  732. if !sc.flow.add(int32(f.Increment)) {
  733. return goAwayFlowError{}
  734. }
  735. }
  736. return nil
  737. }
  738. func (sc *serverConn) processResetStream(f *RSTStreamFrame) error {
  739. sc.serveG.check()
  740. if sc.state(f.StreamID) == stateIdle {
  741. // 6.4 "RST_STREAM frames MUST NOT be sent for a
  742. // stream in the "idle" state. If a RST_STREAM frame
  743. // identifying an idle stream is received, the
  744. // recipient MUST treat this as a connection error
  745. // (Section 5.4.1) of type PROTOCOL_ERROR.
  746. return ConnectionError(ErrCodeProtocol)
  747. }
  748. st, ok := sc.streams[f.StreamID]
  749. if ok {
  750. st.gotReset = true
  751. sc.closeStream(st, StreamError{f.StreamID, f.ErrCode})
  752. }
  753. return nil
  754. }
  755. func (sc *serverConn) closeStream(st *stream, err error) {
  756. sc.serveG.check()
  757. if st.state == stateIdle || st.state == stateClosed {
  758. panic("invariant")
  759. }
  760. st.state = stateClosed
  761. delete(sc.streams, st.id)
  762. st.flow.close()
  763. if p := st.body; p != nil {
  764. p.Close(err)
  765. }
  766. st.cw.Close() // signals Handler's CloseNotifier goroutine (if any) to send
  767. }
  768. func (sc *serverConn) processSettings(f *SettingsFrame) error {
  769. sc.serveG.check()
  770. if f.IsAck() {
  771. // TODO: do we need to do anything?
  772. return nil
  773. }
  774. if err := f.ForeachSetting(sc.processSetting); err != nil {
  775. return err
  776. }
  777. sc.enqueueSettingsAck()
  778. return nil
  779. }
  780. func (sc *serverConn) writeSettingsAck(_ interface{}) error {
  781. return sc.framer.WriteSettingsAck()
  782. }
  783. func (sc *serverConn) processSetting(s Setting) error {
  784. sc.serveG.check()
  785. if err := s.Valid(); err != nil {
  786. return err
  787. }
  788. sc.vlogf("processing setting %v", s)
  789. switch s.ID {
  790. case SettingHeaderTableSize:
  791. sc.headerTableSize = s.Val
  792. return nil
  793. case SettingEnablePush:
  794. sc.pushEnabled = s.Val != 0
  795. return nil
  796. case SettingMaxConcurrentStreams:
  797. sc.maxConcurrentStreams = int64(s.Val)
  798. return nil
  799. case SettingInitialWindowSize:
  800. return sc.processSettingInitialWindowSize(s.Val)
  801. case SettingMaxFrameSize:
  802. sc.maxWriteFrameSize = s.Val
  803. return nil
  804. case SettingMaxHeaderListSize:
  805. sc.maxHeaderListSize = s.Val
  806. return nil
  807. }
  808. // Unknown setting: "An endpoint that receives a SETTINGS
  809. // frame with any unknown or unsupported identifier MUST
  810. // ignore that setting."
  811. return nil
  812. }
  813. func (sc *serverConn) processSettingInitialWindowSize(val uint32) error {
  814. sc.serveG.check()
  815. // Note: val already validated to be within range by
  816. // processSetting's Valid call.
  817. // "A SETTINGS frame can alter the initial flow control window
  818. // size for all current streams. When the value of
  819. // SETTINGS_INITIAL_WINDOW_SIZE changes, a receiver MUST
  820. // adjust the size of all stream flow control windows that it
  821. // maintains by the difference between the new value and the
  822. // old value."
  823. old := sc.initialWindowSize
  824. sc.initialWindowSize = int32(val)
  825. growth := sc.initialWindowSize - old // may be negative
  826. for _, st := range sc.streams {
  827. if !st.flow.add(growth) {
  828. // 6.9.2 Initial Flow Control Window Size
  829. // "An endpoint MUST treat a change to
  830. // SETTINGS_INITIAL_WINDOW_SIZE that causes any flow
  831. // control window to exceed the maximum size as a
  832. // connection error (Section 5.4.1) of type
  833. // FLOW_CONTROL_ERROR."
  834. return ConnectionError(ErrCodeFlowControl)
  835. }
  836. }
  837. return nil
  838. }
  839. func (sc *serverConn) processData(f *DataFrame) error {
  840. sc.serveG.check()
  841. // "If a DATA frame is received whose stream is not in "open"
  842. // or "half closed (local)" state, the recipient MUST respond
  843. // with a stream error (Section 5.4.2) of type STREAM_CLOSED."
  844. id := f.Header().StreamID
  845. st, ok := sc.streams[id]
  846. if !ok || (st.state != stateOpen && st.state != stateHalfClosedLocal) {
  847. return StreamError{id, ErrCodeStreamClosed}
  848. }
  849. if st.body == nil {
  850. // Not expecting data.
  851. // TODO: which error code?
  852. return StreamError{id, ErrCodeStreamClosed}
  853. }
  854. data := f.Data()
  855. // Sender sending more than they'd declared?
  856. if st.declBodyBytes != -1 && st.bodyBytes+int64(len(data)) > st.declBodyBytes {
  857. st.body.Close(fmt.Errorf("Sender tried to send more than declared Content-Length of %d bytes", st.declBodyBytes))
  858. return StreamError{id, ErrCodeStreamClosed}
  859. }
  860. if len(data) > 0 {
  861. // TODO: verify they're allowed to write with the flow control
  862. // window we'd advertised to them.
  863. // TODO: verify n from Write
  864. if _, err := st.body.Write(data); err != nil {
  865. return StreamError{id, ErrCodeStreamClosed}
  866. }
  867. st.bodyBytes += int64(len(data))
  868. }
  869. if f.StreamEnded() {
  870. if st.declBodyBytes != -1 && st.declBodyBytes != st.bodyBytes {
  871. st.body.Close(fmt.Errorf("Request declared a Content-Length of %d but only wrote %d bytes",
  872. st.declBodyBytes, st.bodyBytes))
  873. } else {
  874. st.body.Close(io.EOF)
  875. }
  876. switch st.state {
  877. case stateOpen:
  878. st.state = stateHalfClosedRemote
  879. case stateHalfClosedLocal:
  880. st.state = stateClosed
  881. }
  882. }
  883. return nil
  884. }
  885. func (sc *serverConn) processHeaders(f *HeadersFrame) error {
  886. sc.serveG.check()
  887. id := f.Header().StreamID
  888. if sc.inGoAway {
  889. // Ignore.
  890. return nil
  891. }
  892. // http://http2.github.io/http2-spec/#rfc.section.5.1.1
  893. if id%2 != 1 || id <= sc.maxStreamID || sc.req.stream != nil {
  894. // Streams initiated by a client MUST use odd-numbered
  895. // stream identifiers. [...] The identifier of a newly
  896. // established stream MUST be numerically greater than all
  897. // streams that the initiating endpoint has opened or
  898. // reserved. [...] An endpoint that receives an unexpected
  899. // stream identifier MUST respond with a connection error
  900. // (Section 5.4.1) of type PROTOCOL_ERROR.
  901. return ConnectionError(ErrCodeProtocol)
  902. }
  903. if id > sc.maxStreamID {
  904. sc.maxStreamID = id
  905. }
  906. st := &stream{
  907. conn: sc,
  908. id: id,
  909. state: stateOpen,
  910. flow: newFlow(sc.initialWindowSize),
  911. }
  912. st.cw.Init() // make Cond use its Mutex, without heap-promoting them separately
  913. if f.StreamEnded() {
  914. st.state = stateHalfClosedRemote
  915. }
  916. sc.streams[id] = st
  917. sc.req = requestParam{
  918. stream: st,
  919. header: make(http.Header),
  920. }
  921. return sc.processHeaderBlockFragment(st, f.HeaderBlockFragment(), f.HeadersEnded())
  922. }
  923. func (sc *serverConn) processContinuation(f *ContinuationFrame) error {
  924. sc.serveG.check()
  925. st := sc.streams[f.Header().StreamID]
  926. if st == nil || sc.curHeaderStreamID() != st.id {
  927. return ConnectionError(ErrCodeProtocol)
  928. }
  929. return sc.processHeaderBlockFragment(st, f.HeaderBlockFragment(), f.HeadersEnded())
  930. }
  931. func (sc *serverConn) processHeaderBlockFragment(st *stream, frag []byte, end bool) error {
  932. sc.serveG.check()
  933. if _, err := sc.hpackDecoder.Write(frag); err != nil {
  934. // TODO: convert to stream error I assume?
  935. return err
  936. }
  937. if !end {
  938. return nil
  939. }
  940. if err := sc.hpackDecoder.Close(); err != nil {
  941. // TODO: convert to stream error I assume?
  942. return err
  943. }
  944. rw, req, err := sc.newWriterAndRequest()
  945. sc.req = requestParam{}
  946. if err != nil {
  947. return err
  948. }
  949. st.body = req.Body.(*requestBody).pipe // may be nil
  950. st.declBodyBytes = req.ContentLength
  951. go sc.runHandler(rw, req)
  952. return nil
  953. }
  954. func (sc *serverConn) newWriterAndRequest() (*responseWriter, *http.Request, error) {
  955. sc.serveG.check()
  956. rp := &sc.req
  957. if rp.invalidHeader || rp.method == "" || rp.path == "" ||
  958. (rp.scheme != "https" && rp.scheme != "http") {
  959. // See 8.1.2.6 Malformed Requests and Responses:
  960. //
  961. // Malformed requests or responses that are detected
  962. // MUST be treated as a stream error (Section 5.4.2)
  963. // of type PROTOCOL_ERROR."
  964. //
  965. // 8.1.2.3 Request Pseudo-Header Fields
  966. // "All HTTP/2 requests MUST include exactly one valid
  967. // value for the :method, :scheme, and :path
  968. // pseudo-header fields"
  969. return nil, nil, StreamError{rp.stream.id, ErrCodeProtocol}
  970. }
  971. var tlsState *tls.ConnectionState // make this non-nil if https
  972. if rp.scheme == "https" {
  973. tlsState = &tls.ConnectionState{}
  974. if tc, ok := sc.conn.(*tls.Conn); ok {
  975. *tlsState = tc.ConnectionState()
  976. if tlsState.Version < tls.VersionTLS12 {
  977. // 9.2 Use of TLS Features
  978. // An implementation of HTTP/2 over TLS MUST use TLS
  979. // 1.2 or higher with the restrictions on feature set
  980. // and cipher suite described in this section. Due to
  981. // implementation limitations, it might not be
  982. // possible to fail TLS negotiation. An endpoint MUST
  983. // immediately terminate an HTTP/2 connection that
  984. // does not meet the TLS requirements described in
  985. // this section with a connection error (Section
  986. // 5.4.1) of type INADEQUATE_SECURITY.
  987. return nil, nil, ConnectionError(ErrCodeInadequateSecurity)
  988. }
  989. // TODO: verify cipher suites. (9.2.1, 9.2.2)
  990. }
  991. }
  992. authority := rp.authority
  993. if authority == "" {
  994. authority = rp.header.Get("Host")
  995. }
  996. needsContinue := rp.header.Get("Expect") == "100-continue"
  997. if needsContinue {
  998. rp.header.Del("Expect")
  999. }
  1000. bodyOpen := rp.stream.state == stateOpen
  1001. body := &requestBody{
  1002. sc: sc,
  1003. stream: rp.stream,
  1004. needsContinue: needsContinue,
  1005. }
  1006. url, err := url.ParseRequestURI(rp.path)
  1007. if err != nil {
  1008. // TODO: find the right error code?
  1009. return nil, nil, StreamError{rp.stream.id, ErrCodeProtocol}
  1010. }
  1011. req := &http.Request{
  1012. Method: rp.method,
  1013. URL: url,
  1014. RemoteAddr: sc.conn.RemoteAddr().String(),
  1015. Header: rp.header,
  1016. RequestURI: rp.path,
  1017. Proto: "HTTP/2.0",
  1018. ProtoMajor: 2,
  1019. ProtoMinor: 0,
  1020. TLS: tlsState,
  1021. Host: authority,
  1022. Body: body,
  1023. }
  1024. if bodyOpen {
  1025. body.pipe = &pipe{
  1026. b: buffer{buf: make([]byte, 65536)}, // TODO: share/remove
  1027. }
  1028. body.pipe.c.L = &body.pipe.m
  1029. if vv, ok := rp.header["Content-Length"]; ok {
  1030. req.ContentLength, _ = strconv.ParseInt(vv[0], 10, 64)
  1031. } else {
  1032. req.ContentLength = -1
  1033. }
  1034. }
  1035. rws := responseWriterStatePool.Get().(*responseWriterState)
  1036. bwSave := rws.bw
  1037. *rws = responseWriterState{} // zero all the fields
  1038. rws.bw = bwSave
  1039. rws.bw.Reset(chunkWriter{rws})
  1040. rws.stream = rp.stream
  1041. rws.req = req
  1042. rws.body = body
  1043. rws.chunkWrittenCh = make(chan error, 1)
  1044. rw := &responseWriter{rws: rws}
  1045. return rw, req, nil
  1046. }
  1047. const handlerChunkWriteSize = 4 << 10
  1048. var responseWriterStatePool = sync.Pool{
  1049. New: func() interface{} {
  1050. rws := &responseWriterState{}
  1051. rws.bw = bufio.NewWriterSize(chunkWriter{rws}, handlerChunkWriteSize)
  1052. return rws
  1053. },
  1054. }
  1055. // Run on its own goroutine.
  1056. func (sc *serverConn) runHandler(rw *responseWriter, req *http.Request) {
  1057. defer rw.handlerDone()
  1058. // TODO: catch panics like net/http.Server
  1059. sc.handler.ServeHTTP(rw, req)
  1060. }
  1061. type frameWriteMsg struct {
  1062. // write runs on the writeFrames goroutine.
  1063. write func(sc *serverConn, v interface{}) error
  1064. v interface{} // passed to write
  1065. cost uint32 // number of flow control bytes required
  1066. stream *stream // used for prioritization
  1067. endStream bool // streamID is being closed locally
  1068. // done, if non-nil, must be a buffered channel with space for
  1069. // 1 message and is sent the return value from write (or an
  1070. // earlier error) when the frame has been written.
  1071. done chan error
  1072. }
  1073. // headerWriteReq is a request to write an HTTP response header from a server Handler.
  1074. type headerWriteReq struct {
  1075. stream *stream
  1076. httpResCode int
  1077. h http.Header // may be nil
  1078. endStream bool
  1079. contentType string
  1080. contentLength string
  1081. }
  1082. // called from handler goroutines.
  1083. // h may be nil.
  1084. func (sc *serverConn) writeHeaders(req headerWriteReq) {
  1085. sc.serveG.checkNotOn() // NOT on
  1086. var errc chan error
  1087. if req.h != nil {
  1088. // If there's a header map (which we don't own), so we have to block on
  1089. // waiting for this frame to be written, so an http.Flush mid-handler
  1090. // writes out the correct value of keys, before a handler later potentially
  1091. // mutates it.
  1092. errc = make(chan error, 1)
  1093. }
  1094. sc.writeFrame(frameWriteMsg{
  1095. write: (*serverConn).writeHeadersFrame,
  1096. v: req,
  1097. stream: req.stream,
  1098. done: errc,
  1099. endStream: req.endStream,
  1100. })
  1101. if errc != nil {
  1102. select {
  1103. case <-errc:
  1104. // Ignore. Just for synchronization.
  1105. // Any error will be handled in the writing goroutine.
  1106. case <-sc.doneServing:
  1107. // Client has closed the connection.
  1108. }
  1109. }
  1110. }
  1111. func (sc *serverConn) writeHeadersFrame(v interface{}) error {
  1112. sc.writeG.check()
  1113. req := v.(headerWriteReq)
  1114. sc.headerWriteBuf.Reset()
  1115. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: ":status", Value: httpCodeString(req.httpResCode)})
  1116. for k, vv := range req.h {
  1117. k = lowerHeader(k)
  1118. for _, v := range vv {
  1119. // TODO: more of "8.1.2.2 Connection-Specific Header Fields"
  1120. if k == "transfer-encoding" && v != "trailers" {
  1121. continue
  1122. }
  1123. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: k, Value: v})
  1124. }
  1125. }
  1126. if req.contentType != "" {
  1127. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: "content-type", Value: req.contentType})
  1128. }
  1129. if req.contentLength != "" {
  1130. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: "content-length", Value: req.contentLength})
  1131. }
  1132. headerBlock := sc.headerWriteBuf.Bytes()
  1133. if len(headerBlock) > int(sc.maxWriteFrameSize) {
  1134. // we'll need continuation ones.
  1135. panic("TODO")
  1136. }
  1137. return sc.framer.WriteHeaders(HeadersFrameParam{
  1138. StreamID: req.stream.id,
  1139. BlockFragment: headerBlock,
  1140. EndStream: req.endStream,
  1141. EndHeaders: true, // no continuation yet
  1142. })
  1143. }
  1144. // called from handler goroutines.
  1145. func (sc *serverConn) write100ContinueHeaders(st *stream) {
  1146. sc.serveG.checkNotOn() // NOT
  1147. sc.writeFrame(frameWriteMsg{
  1148. write: (*serverConn).write100ContinueHeadersFrame,
  1149. v: st,
  1150. stream: st,
  1151. })
  1152. }
  1153. func (sc *serverConn) write100ContinueHeadersFrame(v interface{}) error {
  1154. sc.writeG.check()
  1155. st := v.(*stream)
  1156. sc.headerWriteBuf.Reset()
  1157. sc.hpackEncoder.WriteField(hpack.HeaderField{Name: ":status", Value: "100"})
  1158. return sc.framer.WriteHeaders(HeadersFrameParam{
  1159. StreamID: st.id,
  1160. BlockFragment: sc.headerWriteBuf.Bytes(),
  1161. EndStream: false,
  1162. EndHeaders: true,
  1163. })
  1164. }
  1165. func (sc *serverConn) writeDataFrame(v interface{}) error {
  1166. sc.writeG.check()
  1167. rws := v.(*responseWriterState)
  1168. return sc.framer.WriteData(rws.stream.id, rws.curChunkIsFinal, rws.curChunk)
  1169. }
  1170. type windowUpdateReq struct {
  1171. stream *stream
  1172. n uint32
  1173. }
  1174. // called from handler goroutines
  1175. func (sc *serverConn) sendWindowUpdate(st *stream, n int) {
  1176. if st == nil {
  1177. panic("no stream")
  1178. }
  1179. const maxUint32 = 2147483647
  1180. for n >= maxUint32 {
  1181. sc.writeFrame(frameWriteMsg{
  1182. write: (*serverConn).sendWindowUpdateInLoop,
  1183. v: windowUpdateReq{st, maxUint32},
  1184. stream: st,
  1185. })
  1186. n -= maxUint32
  1187. }
  1188. if n > 0 {
  1189. sc.writeFrame(frameWriteMsg{
  1190. write: (*serverConn).sendWindowUpdateInLoop,
  1191. v: windowUpdateReq{st, uint32(n)},
  1192. stream: st,
  1193. })
  1194. }
  1195. }
  1196. func (sc *serverConn) sendWindowUpdateInLoop(v interface{}) error {
  1197. sc.writeG.check()
  1198. wu := v.(windowUpdateReq)
  1199. if err := sc.framer.WriteWindowUpdate(0, wu.n); err != nil {
  1200. return err
  1201. }
  1202. if err := sc.framer.WriteWindowUpdate(wu.stream.id, wu.n); err != nil {
  1203. return err
  1204. }
  1205. return nil
  1206. }
  1207. type requestBody struct {
  1208. sc *serverConn
  1209. stream *stream
  1210. closed bool
  1211. pipe *pipe // non-nil if we have a HTTP entity message body
  1212. needsContinue bool // need to send a 100-continue
  1213. }
  1214. var errClosedBody = errors.New("body closed by handler")
  1215. func (b *requestBody) Close() error {
  1216. if b.pipe != nil {
  1217. b.pipe.Close(errClosedBody)
  1218. }
  1219. b.closed = true
  1220. return nil
  1221. }
  1222. func (b *requestBody) Read(p []byte) (n int, err error) {
  1223. if b.needsContinue {
  1224. b.needsContinue = false
  1225. b.sc.write100ContinueHeaders(b.stream)
  1226. }
  1227. if b.pipe == nil {
  1228. return 0, io.EOF
  1229. }
  1230. n, err = b.pipe.Read(p)
  1231. if n > 0 {
  1232. b.sc.sendWindowUpdate(b.stream, n)
  1233. // TODO: tell b.sc to send back 'n' flow control quota credits to the sender
  1234. }
  1235. return
  1236. }
  1237. // responseWriter is the http.ResponseWriter implementation. It's
  1238. // intentionally small (1 pointer wide) to minimize garbage. The
  1239. // responseWriterState pointer inside is zeroed at the end of a
  1240. // request (in handlerDone) and calls on the responseWriter thereafter
  1241. // simply crash (caller's mistake), but the much larger responseWriterState
  1242. // and buffers are reused between multiple requests.
  1243. type responseWriter struct {
  1244. rws *responseWriterState
  1245. }
  1246. // Optional http.ResponseWriter interfaces implemented.
  1247. var (
  1248. _ http.CloseNotifier = (*responseWriter)(nil)
  1249. _ http.Flusher = (*responseWriter)(nil)
  1250. _ stringWriter = (*responseWriter)(nil)
  1251. // TODO: hijacker for websockets?
  1252. )
  1253. type responseWriterState struct {
  1254. // immutable within a request:
  1255. stream *stream
  1256. req *http.Request
  1257. body *requestBody // to close at end of request, if DATA frames didn't
  1258. // TODO: adjust buffer writing sizes based on server config, frame size updates from peer, etc
  1259. bw *bufio.Writer // writing to a chunkWriter{this *responseWriterState}
  1260. // mutated by http.Handler goroutine:
  1261. handlerHeader http.Header // nil until called
  1262. snapHeader http.Header // snapshot of handlerHeader at WriteHeader time
  1263. wroteHeader bool // WriteHeader called (explicitly or implicitly). Not necessarily sent to user yet.
  1264. status int // status code passed to WriteHeader
  1265. sentHeader bool // have we sent the header frame?
  1266. handlerDone bool // handler has finished
  1267. curChunk []byte // current chunk we're writing
  1268. curChunkIsFinal bool
  1269. chunkWrittenCh chan error
  1270. closeNotifierMu sync.Mutex // guards closeNotifierCh
  1271. closeNotifierCh chan bool // nil until first used
  1272. }
  1273. type chunkWriter struct{ rws *responseWriterState }
  1274. func (cw chunkWriter) Write(p []byte) (n int, err error) { return cw.rws.writeChunk(p) }
  1275. // writeChunk writes chunks from the bufio.Writer. But because
  1276. // bufio.Writer may bypass its chunking, sometimes p may be
  1277. // arbitrarily large.
  1278. //
  1279. // writeChunk is also responsible (on the first chunk) for sending the
  1280. // HEADER response.
  1281. func (rws *responseWriterState) writeChunk(p []byte) (n int, err error) {
  1282. if !rws.wroteHeader {
  1283. rws.writeHeader(200)
  1284. }
  1285. if !rws.sentHeader {
  1286. rws.sentHeader = true
  1287. var ctype, clen string // implicit ones, if we can calculate it
  1288. if rws.handlerDone && rws.snapHeader.Get("Content-Length") == "" {
  1289. clen = strconv.Itoa(len(p))
  1290. }
  1291. if rws.snapHeader.Get("Content-Type") == "" {
  1292. ctype = http.DetectContentType(p)
  1293. }
  1294. endStream := rws.handlerDone && len(p) == 0
  1295. rws.stream.conn.writeHeaders(headerWriteReq{
  1296. stream: rws.stream,
  1297. httpResCode: rws.status,
  1298. h: rws.snapHeader,
  1299. endStream: endStream,
  1300. contentType: ctype,
  1301. contentLength: clen,
  1302. })
  1303. if endStream {
  1304. return
  1305. }
  1306. }
  1307. if len(p) == 0 {
  1308. if rws.handlerDone {
  1309. rws.curChunk = nil
  1310. rws.curChunkIsFinal = true
  1311. rws.stream.conn.writeFrame(frameWriteMsg{
  1312. write: (*serverConn).writeDataFrame,
  1313. cost: 0,
  1314. stream: rws.stream,
  1315. endStream: true,
  1316. v: rws, // writeDataInLoop uses only rws.curChunk and rws.curChunkIsFinal
  1317. })
  1318. }
  1319. return
  1320. }
  1321. for len(p) > 0 {
  1322. chunk := p
  1323. if len(chunk) > handlerChunkWriteSize {
  1324. chunk = chunk[:handlerChunkWriteSize]
  1325. }
  1326. p = p[len(chunk):]
  1327. rws.curChunk = chunk
  1328. rws.curChunkIsFinal = rws.handlerDone && len(p) == 0
  1329. // TODO: await flow control tokens for both stream and conn
  1330. rws.stream.conn.writeFrame(frameWriteMsg{
  1331. write: (*serverConn).writeDataFrame,
  1332. cost: uint32(len(chunk)),
  1333. stream: rws.stream,
  1334. endStream: rws.curChunkIsFinal,
  1335. done: rws.chunkWrittenCh,
  1336. v: rws, // writeDataInLoop uses only rws.curChunk and rws.curChunkIsFinal
  1337. })
  1338. // Block until it's written, or if the client disconnects.
  1339. select {
  1340. case err = <-rws.chunkWrittenCh:
  1341. case <-rws.stream.conn.doneServing:
  1342. // Client disconnected.
  1343. err = errClientDisconnected
  1344. }
  1345. if err != nil {
  1346. break
  1347. }
  1348. n += len(chunk)
  1349. }
  1350. return
  1351. }
  1352. func (w *responseWriter) Flush() {
  1353. rws := w.rws
  1354. if rws == nil {
  1355. panic("Header called after Handler finished")
  1356. }
  1357. if rws.bw.Buffered() > 0 {
  1358. if err := rws.bw.Flush(); err != nil {
  1359. // Ignore the error. The frame writer already knows.
  1360. return
  1361. }
  1362. } else {
  1363. // The bufio.Writer won't call chunkWriter.Write
  1364. // (writeChunk with zero bytes, so we have to do it
  1365. // ourselves to force the HTTP response header and/or
  1366. // final DATA frame (with END_STREAM) to be sent.
  1367. rws.writeChunk(nil)
  1368. }
  1369. }
  1370. func (w *responseWriter) CloseNotify() <-chan bool {
  1371. rws := w.rws
  1372. if rws == nil {
  1373. panic("CloseNotify called after Handler finished")
  1374. }
  1375. rws.closeNotifierMu.Lock()
  1376. ch := rws.closeNotifierCh
  1377. if ch == nil {
  1378. ch = make(chan bool, 1)
  1379. rws.closeNotifierCh = ch
  1380. go func() {
  1381. rws.stream.cw.Wait() // wait for close
  1382. ch <- true
  1383. }()
  1384. }
  1385. rws.closeNotifierMu.Unlock()
  1386. return ch
  1387. }
  1388. func (w *responseWriter) Header() http.Header {
  1389. rws := w.rws
  1390. if rws == nil {
  1391. panic("Header called after Handler finished")
  1392. }
  1393. if rws.handlerHeader == nil {
  1394. rws.handlerHeader = make(http.Header)
  1395. }
  1396. return rws.handlerHeader
  1397. }
  1398. func (w *responseWriter) WriteHeader(code int) {
  1399. rws := w.rws
  1400. if rws == nil {
  1401. panic("WriteHeader called after Handler finished")
  1402. }
  1403. rws.writeHeader(code)
  1404. }
  1405. func (rws *responseWriterState) writeHeader(code int) {
  1406. if !rws.wroteHeader {
  1407. rws.wroteHeader = true
  1408. rws.status = code
  1409. if len(rws.handlerHeader) > 0 {
  1410. rws.snapHeader = cloneHeader(rws.handlerHeader)
  1411. }
  1412. }
  1413. }
  1414. func cloneHeader(h http.Header) http.Header {
  1415. h2 := make(http.Header, len(h))
  1416. for k, vv := range h {
  1417. vv2 := make([]string, len(vv))
  1418. copy(vv2, vv)
  1419. h2[k] = vv2
  1420. }
  1421. return h2
  1422. }
  1423. // The Life Of A Write is like this:
  1424. //
  1425. // TODO: copy/adapt the similar comment from Go's http server.go
  1426. func (w *responseWriter) Write(p []byte) (n int, err error) {
  1427. return w.write(len(p), p, "")
  1428. }
  1429. func (w *responseWriter) WriteString(s string) (n int, err error) {
  1430. return w.write(len(s), nil, s)
  1431. }
  1432. // either dataB or dataS is non-zero.
  1433. func (w *responseWriter) write(lenData int, dataB []byte, dataS string) (n int, err error) {
  1434. rws := w.rws
  1435. if rws == nil {
  1436. panic("Write called after Handler finished")
  1437. }
  1438. if !rws.wroteHeader {
  1439. w.WriteHeader(200)
  1440. }
  1441. if dataB != nil {
  1442. return rws.bw.Write(dataB)
  1443. } else {
  1444. return rws.bw.WriteString(dataS)
  1445. }
  1446. }
  1447. func (w *responseWriter) handlerDone() {
  1448. rws := w.rws
  1449. if rws == nil {
  1450. panic("handlerDone called twice")
  1451. }
  1452. rws.handlerDone = true
  1453. w.Flush()
  1454. w.rws = nil
  1455. responseWriterStatePool.Put(rws)
  1456. }