2d3106c · typeable/wai

git.iko.soy/typeable/wai

using the new priority API.

commit2d3106cd200be1e394d81ad297e54d4642f5364a authorKazu Yamamoto <kazu@iij.ad.jp> dateFri 13 Nov 2015, 17:35 (11 years ago) parentc5c577f

4 files changed+58−59

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+21−20 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+18−16 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+13−17 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+6−6

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

@@ -41,17 +41,17 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway

4141 | Just (ConnectionError err msg) <- E.fromException e = do

4242 csid <- readIORef currentStreamId

4343 let frame = goawayFrame csid err msg

44− enqueue outputQ 0 controlPriority $ OGoaway frame

44+ enqueueControl outputQ 0 $ OGoaway frame

4545 | otherwise = return ()

4646

4747 sendReset err sid = do

4848 let frame = resetFrame err sid

49− enqueue outputQ 0 controlPriority $ OFrame frame

49+ enqueueControl outputQ 0 $ OFrame frame

5050

5151 loop = do

5252 hd <- recvN frameHeaderLength

5353 if BS.null hd then

54− enqueue outputQ 0 controlPriority OFinish

54+ enqueueControl outputQ 0 OFinish

5555 else do

5656 cont <- processStreamGuardingError $ decodeFrameHeader hd

5757 when cont loop

@@ -92,7 +92,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway

9292 control ftyp header pl ctx

9393 | otherwise = do

9494 checkContinued

95− strm@Stream{streamState,streamContentLength,streamPriority} <- getStream

95+ strm@Stream{streamState,streamContentLength,streamPrecedence} <- getStream

9696 pl <- recvN payloadLength

9797 state <- readIORef streamState

9898 state' <- stream ftyp header pl ctx state strm

@@ -103,7 +103,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway

103103 Just vh -> do

104104 when (isJust (vhCL vh) && vhCL vh /= Just 0) $

105105 E.throwIO $ StreamError ProtocolError streamId

106− writeIORef streamPriority pri

106+ writeIORef streamPrecedence $ toPrecedence pri

107107 writeIORef streamState HalfClosed

108108 let req = mkreq vh (return "")

109109 atomically $ writeTQueue inputQ $ Input strm req

@@ -113,7 +113,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway

113113 case validateHeaders hdr of

114114 Just vh -> do

115115 q <- newTQueueIO

116− writeIORef streamPriority pri

116+ writeIORef streamPrecedence $ toPrecedence pri

117117 writeIORef streamState (Open (Body q))

118118 writeIORef streamContentLength $ vhCL vh

119119 readQ <- newReadBody q

@@ -186,7 +186,7 @@ control FrameSettings header@FrameHeader{flags} bs Context{http2settings, output

186186 unless (testAck flags) $ do

187187 modifyIORef http2settings $ \old -> updateSettings old alist

188188 let frame = settingsFrame setAck []

189− enqueue outputQ 0 controlPriority $ OSettings frame alist

189+ enqueueControl outputQ 0 $ OSettings frame alist

190190 return True

191191

192192control FramePing FrameHeader{flags} bs Context{outputQ} =

@@ -194,11 +194,11 @@ control FramePing FrameHeader{flags} bs Context{outputQ} =

194194 E.throwIO $ ConnectionError ProtocolError "the ack flag of this ping frame must not be set"

195195 else do

196196 let frame = pingFrame bs

197− enqueue outputQ 0 controlPriority $ OFrame frame

197+ enqueueControl outputQ 0 $ OFrame frame

198198 return True

199199

200200control FrameGoAway _ _ Context{outputQ} = do

201− enqueue outputQ 0 controlPriority OFinish

201+ enqueueControl outputQ 0 OFinish

202202 return False

203203

204204control FrameWindowUpdate header bs Context{connectionWindow} = do

@@ -271,7 +271,7 @@ stream FrameData

271271 let frame1 = windowUpdateFrame 0 payloadLength

272272 frame2 = windowUpdateFrame streamNumber payloadLength

273273 frame = frame1 `BS.append` frame2

274− enqueue outputQ 0 controlPriority $ OFrame frame

274+ enqueueControl outputQ 0 $ OFrame frame

275275 atomically $ writeTQueue q body

276276 if endOfStream then do

277277 mcl <- readIORef streamContentLength

@@ -310,25 +310,26 @@ 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 strm cc ctx

316+ closed strm cc

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

318318

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

320− PriorityFrame newp <- guardIt $ decodePriorityFrame header bs

321− checkPriority newp streamNumber

319+stream FramePriority header bs Context{outputQ} s Stream{streamNumber,streamPrecedence} = do

320+ PriorityFrame newpri <- guardIt $ decodePriorityFrame header bs

321+ checkPriority newpri streamNumber

322322 -- checkme: this should be tested

323− oldp <- readIORef streamPriority

324− writeIORef streamPriority newp

323+ oldpre <- readIORef streamPrecedence

324+ let !newpre = toPrecedence newpri

325+ writeIORef streamPrecedence newpre

325326 if isIdle s then

326− prepare outputQ streamNumber newp

327+ prepare outputQ streamNumber newpri

327328 else do

328− mx <- delete outputQ streamNumber oldp

329+ mx <- delete outputQ streamNumber oldpre

329330 case mx of

330331 Nothing -> return ()

331− Just x -> enqueue outputQ streamNumber newp x

332+ Just x -> enqueue outputQ streamNumber newpre x

332333 return s

333334

334335-- this ordering is important

warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+18−16

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

6060

6161-- | Run the given action if the stream is not closed; handle any exceptions by

6262-- resetting the stream.

63−unlessClosed :: Context -> Connection -> Stream -> IO () -> IO Bool

64−unlessClosed ctx

65− Connection{connSendAll}

63+unlessClosed :: Connection -> Stream -> IO () -> IO Bool

64+unlessClosed Connection{connSendAll}

6665 strm@Stream{streamState,streamNumber}

6766 body = E.handle resetStream $ do

6867 state <- readIORef streamState

6968 if (isClosed state) then return False else body >> return True

7069 where

7170 resetStream e = do

72− closed strm (ResetByMe e) ctx

71+ closed strm (ResetByMe e)

7372 let rst = resetFrame InternalError streamNumber

7473 connSendAll rst

7574 return False

@@ -100,14 +99,14 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}

10099 loop

101100

102101 -- ignoring the old priority because the value might be changed.

103− loop = dequeue outputQ >>= \(_sid,out) -> switch out

102+ loop = dequeue outputQ >>= \(_sid,pre,out) -> switch out pre

104103

105104 ignore :: E.SomeException -> IO ()

106105 ignore _ = return ()

107106

108− switch OFinish = return ()

109− switch (OGoaway frame) = connSendAll frame

110− switch (OSettings frame alist) = do

107+ switch OFinish _ = return ()

108+ switch (OGoaway frame) _ = connSendAll frame

109+ switch (OSettings frame alist) _ = do

111110 connSendAll frame

112111 case lookup SettingsHeaderTableSize alist of

113112 Nothing -> return ()

@@ -115,24 +114,27 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}

115114 dyntbl <- readIORef encodeDynamicTable

116115 setLimitForEncoding siz dyntbl

117116 loop

118− switch (OFrame frame) = do

117+ switch (OFrame frame) _ = do

119118 connSendAll frame

120119 loop

121− switch (OResponse strm s h aux) = do

122− _ <- unlessClosed ctx conn strm $

120+ switch (OResponse strm s h aux) pre = do

121+ writeIORef (streamPrecedence strm) pre -- fixme

122+ _ <- unlessClosed conn strm $

123123 getWindowSize connectionWindow (streamWindow strm) >>=

124124 sendResponse strm s h aux

125125 loop

126− switch (ONext strm curr) = do

127− _ <- unlessClosed ctx conn strm $ do

126+ switch (ONext strm curr) pre = do

127+ writeIORef (streamPrecedence strm) pre

128+ _ <- unlessClosed conn strm $ do

128129 lim <- getWindowSize connectionWindow (streamWindow strm)

129130 -- Data frame payload

130131 Next datPayloadLen mnext <- curr lim

131132 fillDataHeaderSend strm 0 datPayloadLen

132133 dispatchNext strm mnext

133134 loop

134− switch (OPush oldStrm push mvar strm s h aux) = do

135− pushed <- unlessClosed ctx conn oldStrm $ do

135+ switch (OPush oldStrm push mvar strm s h aux) pre = do

136+ writeIORef (streamPrecedence strm) pre -- fixme

137+ pushed <- unlessClosed conn oldStrm $ do

136138 lim <- getWindowSize connectionWindow (streamWindow strm)

137139 -- Write and send the promise.

138140 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push

@@ -184,7 +186,7 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}

184186 -- 'closed' must be before 'flushN'. If not, the context would be

185187 -- switched to the receiver, resulting in the inconsistency of

186188 -- concurrency.

187− closed strm Finished ctx

189+ closed strm Finished

188190 flushN toFlush

189191

190192 -- Flush the connection buffer to the socket, where the first 'n' bytes of

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+13−17

@@ -1,5 +1,5 @@

11{-# LANGUAGE OverloadedStrings, CPP #-}

2−{-# LANGUAGE NamedFieldPuns #-}

2+{-# LANGUAGE NamedFieldPuns, RecordWildCards #-}

44module Network.Wai.Handler.Warp.HTTP2.Types where

@@ -221,7 +221,7 @@ data Stream = Stream {

221221 , streamContentLength :: IORef (Maybe Int)

222222 , streamBodyLength :: IORef Int

223223 , streamWindow :: TVar WindowSize

224− , streamPriority :: IORef Priority

224+ , streamPrecedence :: IORef Precedence

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

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

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

@@ -238,7 +238,7 @@ newStream ref sid win =

238238 <*> newIORef Nothing

239239 <*> newIORef 0

240240 <*> newTVarIO win

241− <*> newIORef defaultPriority

241+ <*> newIORef defaultPrecedence

242242 <*> pure ref

243243

244244----------------------------------------------------------------

@@ -248,14 +248,10 @@ opened Stream{concurrencyRef,streamState} = do

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

249249 writeIORef streamState (Open JustOpened)

250250

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

252−closed Stream{concurrencyRef,streamState,streamNumber,streamPriority}

253− cc

254− Context{outputQ} = do

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

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

255253 atomicModifyIORef' concurrencyRef (\x -> (x-1,()))

256254 writeIORef streamState (Closed cc)

257− pri <- readIORef streamPriority

258− clear outputQ streamNumber pri

259255

260256----------------------------------------------------------------

261257

@@ -296,19 +292,19 @@ search strmtbl k = M.lookup k <$> reaperRead strmtbl

296292-- INVARIANT: streams in the output queue have non-zero window size.

297293enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO ()

298294enqueueWhenWindowIsOpen outQ out = do

299− let strm = outputStream out

295+ let Stream{..} = outputStream out

300296 atomically $ do

301− x <- readTVar $ streamWindow strm

297+ x <- readTVar streamWindow

302298 check (x > 0)

303− pri <- readIORef $ streamPriority strm

304− enqueue outQ (streamNumber strm) pri out

299+ pre <- readIORef streamPrecedence

300+ enqueue outQ streamNumber pre out

305301

306302enqueueOrSpawnTemporaryWaiter :: Stream -> PriorityTree Output -> Output -> IO ()

307−enqueueOrSpawnTemporaryWaiter strm outQ out = do

308− sw <- atomically $ readTVar $ streamWindow strm

303+enqueueOrSpawnTemporaryWaiter Stream{..} outQ out = do

304+ sw <- atomically $ readTVar streamWindow

309305 if sw == 0 then

310306 -- This waiter waits only for the stream window.

311307 void $ forkIO $ enqueueWhenWindowIsOpen outQ out

312308 else do

313− pri <- readIORef $ streamPriority strm

314− enqueue outQ (streamNumber strm) pri out

309+ pre <- readIORef streamPrecedence

310+ enqueue outQ streamNumber pre out

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

@@ -83,7 +83,7 @@ runStream Context{outputQ} mkOutput tickle strm s h strmbdy = do

8383 -- queue, we spawn a thread to ensure that the designated

8484 -- queue is not empty.

8585 void $ forkIO $ waiter tvar sq strm outputQ

86− atomically $ writeTVar tvar (SyncNext out)

86+ atomically $ writeTVar tvar $ SyncNext out

8787 let write chunk = do

8888 atomically $ writeTBQueue sq $ case chunk of

8989 BuilderChunk b -> SBuilder b

@@ -96,11 +96,11 @@ runStream Context{outputQ} mkOutput tickle strm s h strmbdy = do

9696-- | Handle abnormal termination of a stream: mark it as closed, send a reset

9797-- frame, and call the user's 'settingsOnException' handler if applicable.

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

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

100− closed strm Killed ctx

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

100+ closed strm Killed

101101 let sid = streamNumber strm

102102 frame = resetFrame InternalError sid

103− enqueue outputQ sid controlPriority $ OFrame frame

103+ enqueueControl outputQ sid $ OFrame frame

104104 case me of

105105 Nothing -> return ()

106106 Just e -> S.settingsOnException set req e

@@ -144,8 +144,8 @@ actuallyPushResponder ctx set strm promise responder = do

144144 -- Section 5.3.5 of RFC 7540 defines the weight of push promise is 16.

145145 -- But we need not to follow the spec. So, this value would change

146146 -- if necessary.

147− writeIORef (streamPriority newStrm) $

148− defaultPriority { streamDependency = streamNumber strm }

147+ writeIORef (streamPrecedence newStrm) $

148+ toPrecedence $ defaultPriority { streamDependency = streamNumber strm }

149149 opened newStrm

150150 insert streamTable newSid newStrm

151151