14feac1 · typeable/wai

git.iko.soy/typeable/wai

Honor the client MAX_CONCURRENT_STREAMS

commit14feac1fa8f204dd5b4902f01cb5d1666c55b1d9 authorAndrew Pritchard <awpr@google.com> dateSun 16 Aug 2015, 14:07 (11 years ago) parentd11f7aa

4 files changed+51−34

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+5−5 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+9−10 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+25−12 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+12−7

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+5−5

@@ -161,15 +161,15 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway

161161 -- idle, so that receiving frames on it is only a

162162 -- stream error.

163163 consume payloadLength

164− strm <- newStream streamId 0

164+ strm <- newStream concurrency streamId 0

165165 writeIORef (streamState strm) $ Closed $

166166 ResetByMe $ E.toException $

167167 StreamError RefusedStream streamId

168168 insert streamTable streamId strm

169169 E.throwIO $ StreamError RefusedStream streamId

170170 ws <- initialWindowSize <$> readIORef http2settings

171− newstrm <- newStream streamId (fromIntegral ws)

172− when (ftyp == FrameHeaders) $ opened ctx newstrm

171+ newstrm <- newStream concurrency streamId (fromIntegral ws)

172+ when (ftyp == FrameHeaders) $ opened newstrm

173173 insert streamTable streamId newstrm

174174 return newstrm

175175

@@ -310,10 +310,10 @@ stream FrameWindowUpdate header@FrameHeader{streamId} bs _ s Stream{streamWindow

310310 atomically $ writeTVar streamWindow w

311311 return s

312312

313−stream FrameRSTStream header bs ctx _ strm = do

313+stream FrameRSTStream header bs _ _ strm = do

314314 RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs

315315 let cc = Reset e

316− closed ctx strm cc

316+ closed strm cc

317317 return $ Closed cc -- will be written to streamState again

318318

319319stream FramePriority header bs Context{outputQ} s Stream{streamNumber,streamPriority} = do

warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+9−10

@@ -35,16 +35,15 @@ data Leftover = LZero

3535

3636----------------------------------------------------------------

3737

38−unlessClosed :: Context -> Connection -> Stream -> IO () -> IO ()

39−unlessClosed ctx

40− Connection{connSendAll}

38+unlessClosed :: Connection -> Stream -> IO () -> IO ()

39+unlessClosed Connection{connSendAll}

4140 strm@Stream{streamState,streamNumber}

4241 body = E.handle resetStream $ do

4342 state <- readIORef streamState

4443 unless (isClosed state) body

4544 where

4645 resetStream e = do

47− closed ctx strm (ResetByMe e)

46+ closed strm (ResetByMe e)

4847 let rst = resetFrame InternalError streamNumber

4948 connSendAll rst

5049

@@ -85,12 +84,12 @@ frameSender ctx@Context{outputQ,connectionWindow}

8584 connSendAll frame

8685 loop

8786 switch (OResponse strm s h aux) = do

88− unlessClosed ctx conn strm $

87+ unlessClosed conn strm $

8988 getWindowSize connectionWindow (streamWindow strm) >>=

9089 sendResponse strm s h aux

9190 loop

9291 switch (ONext strm curr) = do

93− unlessClosed ctx conn strm $ do

92+ unlessClosed conn strm $ do

9493 lim <- getWindowSize connectionWindow (streamWindow strm)

9594 -- Data frame payload

9695 Next datPayloadLen mnext <- curr lim

@@ -98,7 +97,7 @@ frameSender ctx@Context{outputQ,connectionWindow}

9897 maybeEnqueueNext strm mnext

9998 loop

10099 switch (OPush oldStrm push strm s h aux) = do

101− unlessClosed ctx conn oldStrm $ do

100+ unlessClosed conn oldStrm $ do

102101 lim <- getWindowSize connectionWindow (streamWindow strm)

103102 -- Write and send the promise.

104103 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push

@@ -111,7 +110,7 @@ frameSender ctx@Context{outputQ,connectionWindow}

111110 switch (OTrailers strm []) = do

112111 let flag = setEndStream defaultFlags

113112 fillFrameHeader FrameData 0 (streamNumber strm) flag connWriteBuffer

114− closed ctx strm Finished

113+ closed strm Finished

115114 flushN frameHeaderLength

116115 loop

117116 switch (OTrailers strm trailers) = do

@@ -119,7 +118,7 @@ frameSender ctx@Context{outputQ,connectionWindow}

119118 -- consecutive header+continuation frames and end the stream.

120119 builder <- hpackEncodeCIHeaders ctx trailers

121120 off <- headerContinue (streamNumber strm) builder True

122− closed ctx strm Finished

121+ closed strm Finished

123122 flushN $ off + frameHeaderLength

124123 loop

125124

@@ -224,7 +223,7 @@ frameSender ctx@Context{outputQ,connectionWindow}

224223 -- the context would be switched to the receiver,

225224 -- resulting the inconsistency of concurrency.

226225 case mnext of

227− CFinish -> closed ctx strm Finished

226+ CFinish -> closed strm Finished

228227 _ -> return ()

229228 flushN total

230229 atomically $ do

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+25−12

@@ -101,7 +101,12 @@ data Aux = Persist (TBQueue Sequence) (TVar Sync)

101101data Context = Context {

102102 http2settings :: IORef Settings

103103 , streamTable :: StreamTable

104+ -- | Number of active streams initiated by the client; for enforcing our own

105+ -- max concurrency setting.

104106 , concurrency :: IORef Int

107+ -- | Number of active streams initiated by the server; for respecting the

108+ -- client's max concurrency setting.

109+ , pushConcurrency :: IORef Int

105110 -- | RFC 7540 says "Other frames (from any stream) MUST NOT

106111 -- occur between the HEADERS frame and any CONTINUATION

107112 -- frames that might follow". This field is used to implement

@@ -124,6 +129,7 @@ newContext :: IO Context

124129newContext = Context <$> newIORef defaultSettings

125130 <*> initialize 10 -- fixme: hard coding: 10

126131 <*> newIORef 0

132+ <*> newIORef 0

127133 <*> newIORef Nothing

128134 <*> newIORef 0

129135 <*> newIORef 2 -- first server push stream; 0 is reserved

@@ -193,28 +199,35 @@ data Stream = Stream {

193199 , streamBodyLength :: IORef Int

194200 , streamWindow :: TVar WindowSize

195201 , streamPriority :: IORef Priority

202+ -- | The concurrency IORef in which this stream has been counted. The client

203+ -- and server each have separate concurrency values to respect, so pushed

204+ -- streams need to decrement a different count when they're closed. This

205+ -- should be either @concurrency ctx@ or @pushConcurrency ctx@.

206+ , concurrencyRef :: IORef Int

196207 }

197208

198209instance Show Stream where

199210 show s = show (streamNumber s)

200211

201−newStream :: StreamId -> WindowSize -> IO Stream

202−newStream sid win = Stream sid <$> newIORef Idle

203− <*> newIORef Nothing

204− <*> newIORef 0

205− <*> newTVarIO win

206− <*> newIORef defaultPriority

212+newStream :: IORef Int -> StreamId -> WindowSize -> IO Stream

213+newStream ref sid win =

214+ Stream sid <$> newIORef Idle

215+ <*> newIORef Nothing

216+ <*> newIORef 0

217+ <*> newTVarIO win

218+ <*> newIORef defaultPriority

219+ <*> pure ref

207220

208221----------------------------------------------------------------

209222

210−opened :: Context -> Stream -> IO ()

211−opened Context{concurrency} Stream{streamState} = do

212− atomicModifyIORef' concurrency (\x -> (x+1,()))

223+opened :: Stream -> IO ()

224+opened Stream{concurrencyRef,streamState} = do

225+ atomicModifyIORef' concurrencyRef (\x -> (x+1,()))

213226 writeIORef streamState (Open JustOpened)

214227

215−closed :: Context -> Stream -> ClosedCode -> IO ()

216−closed Context{concurrency} Stream{streamState} cc = do

217− atomicModifyIORef' concurrency (\x -> (x-1,()))

228+closed :: Stream -> ClosedCode -> IO ()

229+closed Stream{concurrencyRef,streamState} cc = do

230+ atomicModifyIORef' concurrencyRef (\x -> (x-1,()))

218231 writeIORef streamState (Closed cc)

219232

220233----------------------------------------------------------------

warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+12−7

@@ -83,8 +83,8 @@ runResponder responder respond tickle strm = do

8383 atomically $ writeTBQueue sq $ SFinish trailers

8484

8585cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO ()

86−cleanupStream ctx@Context{outputQ} set strm req me = do

87− closed ctx strm Killed

86+cleanupStream Context{outputQ} set strm req me = do

87+ closed strm Killed

8888 let frame = resetFrame InternalError (streamNumber strm)

8989 enqueue outputQ (OFrame frame) highestPriority

9090 case me of

@@ -95,18 +95,23 @@ pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder ->

9595pushResponder ctx set strm promise responder = do

9696 let Context{ http2settings

9797 , nextPushStreamId

98+ , pushConcurrency

9899 , streamTable

99100 } = ctx

100− enabled <- enablePush <$> readIORef http2settings

101− when enabled $ do

101+ cnt <- readIORef pushConcurrency

102+ settings <- readIORef http2settings

103+ let enabled = enablePush settings

104+ fits = maybe True (cnt <) $ maxConcurrentStreams settings

105+ canPush = fits && enabled

106+ when canPush $ do

102107 -- Claim the next outgoing stream.

103108 newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)

104109 ws <- initialWindowSize <$> readIORef http2settings

105110

106− newStrm <- newStream newSid ws

111+ newStrm <- newStream pushConcurrency newSid ws

107112 writeIORef (streamPriority newStrm) $

108113 Priority False (streamNumber strm) 16

109− opened ctx newStrm

114+ opened newStrm

110115 insert streamTable newSid newStrm

111116

112117 let mkOutput = OPush strm promise

@@ -116,7 +121,7 @@ pushResponder ctx set strm promise responder = do

116121 -- TODO(awpr): synthesize a Request for 'settingsOnException'?

117122 runResponder responder respond tickle newStrm `E.catch`

118123 (cleanupStream ctx set strm Nothing . Just)

119− return enabled

124+ return canPush

120125

121126data Break = Break deriving (Show, Typeable)

122127