14feac1 · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

Honor the client MAX_CONCURRENT_STREAMS

14feac1Andrew Pritchard11 years ago

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

compare with master

unifiedsplit

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

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

162 -- stream error.

163 consume payloadLength

164− strm <- newStream streamId 0

164+ strm <- newStream concurrency streamId 0

165 writeIORef (streamState strm) $ Closed $

166 ResetByMe $ E.toException $

167 StreamError RefusedStream streamId

168 insert streamTable streamId strm

169 E.throwIO $ StreamError RefusedStream streamId

170 ws <- initialWindowSize <$> readIORef http2settings

171− newstrm <- newStream streamId (fromIntegral ws)

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

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

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

173 insert streamTable streamId newstrm

174 return newstrm

175

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

310 atomically $ writeTVar streamWindow w

311 return s

312

313−stream FrameRSTStream header bs ctx _ strm = do

313+stream FrameRSTStream header bs _ _ strm = do

314 RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs

315 let cc = Reset e

316− closed ctx strm cc

316+ closed strm cc

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

318

319stream 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

36----------------------------------------------------------------

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

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

39−unlessClosed ctx

40− Connection{connSendAll}

39+unlessClosed Connection{connSendAll}

41 strm@Stream{streamState,streamNumber}

40 strm@Stream{streamState,streamNumber}

42 body = E.handle resetStream $ do

41 body = E.handle resetStream $ do

43 state <- readIORef streamState

42 state <- readIORef streamState

44 unless (isClosed state) body

43 unless (isClosed state) body

45 where

44 where

46 resetStream e = do

45 resetStream e = do

47− closed ctx strm (ResetByMe e)

46+ closed strm (ResetByMe e)

48 let rst = resetFrame InternalError streamNumber

47 let rst = resetFrame InternalError streamNumber

49 connSendAll rst

48 connSendAll rst

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

85 connSendAll frame

84 connSendAll frame

86 loop

85 loop

87 switch (OResponse strm s h aux) = do

86 switch (OResponse strm s h aux) = do

88− unlessClosed ctx conn strm $

87+ unlessClosed conn strm $

89 getWindowSize connectionWindow (streamWindow strm) >>=

88 getWindowSize connectionWindow (streamWindow strm) >>=

90 sendResponse strm s h aux

89 sendResponse strm s h aux

91 loop

90 loop

92 switch (ONext strm curr) = do

91 switch (ONext strm curr) = do

93− unlessClosed ctx conn strm $ do

92+ unlessClosed conn strm $ do

94 lim <- getWindowSize connectionWindow (streamWindow strm)

93 lim <- getWindowSize connectionWindow (streamWindow strm)

95 -- Data frame payload

94 -- Data frame payload

96 Next datPayloadLen mnext <- curr lim

95 Next datPayloadLen mnext <- curr lim

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

98 maybeEnqueueNext strm mnext

97 maybeEnqueueNext strm mnext

99 loop

98 loop

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

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

101− unlessClosed ctx conn oldStrm $ do

100+ unlessClosed conn oldStrm $ do

102 lim <- getWindowSize connectionWindow (streamWindow strm)

101 lim <- getWindowSize connectionWindow (streamWindow strm)

103 -- Write and send the promise.

102 -- Write and send the promise.

104 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push

103 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push

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

111 switch (OTrailers strm []) = do

110 switch (OTrailers strm []) = do

112 let flag = setEndStream defaultFlags

111 let flag = setEndStream defaultFlags

113 fillFrameHeader FrameData 0 (streamNumber strm) flag connWriteBuffer

112 fillFrameHeader FrameData 0 (streamNumber strm) flag connWriteBuffer

114− closed ctx strm Finished

113+ closed strm Finished

115 flushN frameHeaderLength

114 flushN frameHeaderLength

116 loop

115 loop

117 switch (OTrailers strm trailers) = do

116 switch (OTrailers strm trailers) = do

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

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

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

120 builder <- hpackEncodeCIHeaders ctx trailers

119 builder <- hpackEncodeCIHeaders ctx trailers

121 off <- headerContinue (streamNumber strm) builder True

120 off <- headerContinue (streamNumber strm) builder True

122− closed ctx strm Finished

121+ closed strm Finished

123 flushN $ off + frameHeaderLength

122 flushN $ off + frameHeaderLength

124 loop

123 loop

125

124

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

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

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

225 -- resulting the inconsistency of concurrency.

224 -- resulting the inconsistency of concurrency.

226 case mnext of

225 case mnext of

227− CFinish -> closed ctx strm Finished

226+ CFinish -> closed strm Finished

228 _ -> return ()

227 _ -> return ()

229 flushN total

228 flushN total

230 atomically $ do

229 atomically $ do

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

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

101data Context = Context {

102 http2settings :: IORef Settings

103 , streamTable :: StreamTable

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

105+ -- max concurrency setting.

104 , concurrency :: IORef Int

106 , 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

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

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

106 -- occur between the HEADERS frame and any CONTINUATION

111 -- occur between the HEADERS frame and any CONTINUATION

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

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

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

124newContext = Context <$> newIORef defaultSettings

129newContext = Context <$> newIORef defaultSettings

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

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

126 <*> newIORef 0

131 <*> newIORef 0

132+ <*> newIORef 0

127 <*> newIORef Nothing

133 <*> newIORef Nothing

128 <*> newIORef 0

134 <*> newIORef 0

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

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

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

193 , streamBodyLength :: IORef Int

199 , streamBodyLength :: IORef Int

194 , streamWindow :: TVar WindowSize

200 , streamWindow :: TVar WindowSize

195 , streamPriority :: IORef Priority

201 , 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

196 }

207 }

197

208

198instance Show Stream where

209instance Show Stream where

199 show s = show (streamNumber s)

210 show s = show (streamNumber s)

200

211

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

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

213+newStream ref sid win =

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

214+ Stream sid <$> newIORef Idle

203− <*> newIORef Nothing

215+ <*> newIORef Nothing

204− <*> newIORef 0

216+ <*> newIORef 0

205− <*> newTVarIO win

217+ <*> newTVarIO win

206− <*> newIORef defaultPriority

218+ <*> newIORef defaultPriority

219+ <*> pure ref

207

220

208----------------------------------------------------------------

221----------------------------------------------------------------

209

222

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

223+opened :: Stream -> IO ()

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

224+opened Stream{concurrencyRef,streamState} = do

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

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

213 writeIORef streamState (Open JustOpened)

226 writeIORef streamState (Open JustOpened)

214

227

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

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

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

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

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

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

218 writeIORef streamState (Closed cc)

231 writeIORef streamState (Closed cc)

219

232

220----------------------------------------------------------------

233----------------------------------------------------------------

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

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

83 atomically $ writeTBQueue sq $ SFinish trailers

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

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

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

87− closed ctx strm Killed

87+ closed strm Killed

88 let frame = resetFrame InternalError (streamNumber strm)

89 enqueue outputQ (OFrame frame) highestPriority

90 case me of

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

95pushResponder ctx set strm promise responder = do

96 let Context{ http2settings

97 , nextPushStreamId

98+ , pushConcurrency

98 , streamTable

99 , streamTable

99 } = ctx

100 } = ctx

101+ cnt <- readIORef pushConcurrency

102+ settings <- readIORef http2settings

100− enabled <- enablePush <$> readIORef http2settings

103+ let enabled = enablePush settings

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

105+ canPush = fits && enabled

101− when enabled $ do

106+ when canPush $ do

102 -- Claim the next outgoing stream.

107 -- Claim the next outgoing stream.

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

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

104 ws <- initialWindowSize <$> readIORef http2settings

109 ws <- initialWindowSize <$> readIORef http2settings

105

110

106− newStrm <- newStream newSid ws

111+ newStrm <- newStream pushConcurrency newSid ws

107 writeIORef (streamPriority newStrm) $

112 writeIORef (streamPriority newStrm) $

108 Priority False (streamNumber strm) 16

113 Priority False (streamNumber strm) 16

109− opened ctx newStrm

114+ opened newStrm

110 insert streamTable newSid newStrm

115 insert streamTable newSid newStrm

111

116

112 let mkOutput = OPush strm promise

117 let mkOutput = OPush strm promise

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

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

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

117 runResponder responder respond tickle newStrm `E.catch`

122 runResponder responder respond tickle newStrm `E.catch`

118 (cleanupStream ctx set strm Nothing . Just)

123 (cleanupStream ctx set strm Nothing . Just)

119− return enabled

124+ return canPush

120

125

121data Break = Break deriving (Show, Typeable)

126data Break = Break deriving (Show, Typeable)

122

127