b0eeb67 · typeable/wai

git.iko.soy/typeable/wai

using new priority tree.

commitb0eeb67ed03f978b977f398c165c56d38657350f authorKazu Yamamoto <kazu@iij.ad.jp> dateMon 26 Oct 2015, 17:05 (11 years ago) parent7f0d4ca

4 files changed+42−35

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+20−19 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+9−8 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+8−4 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+5−4

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

@@ -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 (OGoaway frame) highestPriority

44+ enqueue outputQ 0 controlPriority $ OGoaway frame

4545 | otherwise = return ()

4646

4747 sendReset err sid = do

4848 let frame = resetFrame err sid

49− enqueue outputQ (OFrame frame) highestPriority

49+ enqueue outputQ 0 controlPriority $ OFrame frame

5050

5151 loop = do

5252 hd <- recvN frameHeaderLength

5353 if BS.null hd then

54− enqueue outputQ OFinish highestPriority

54+ enqueue outputQ 0 controlPriority OFinish

5555 else do

5656 cont <- processStreamGuardingError $ decodeFrameHeader hd

5757 when cont loop

@@ -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 (OSettings frame alist) highestPriority

189+ enqueue outputQ 0 controlPriority $ 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 (OFrame frame) defaultPriority

197+ enqueue outputQ 0 controlPriority $ OFrame frame

198198 return True

199199

200200control FrameGoAway _ _ Context{outputQ} = do

201− enqueue outputQ OFinish highestPriority

201+ enqueue outputQ 0 controlPriority 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 (OFrame frame) highestPriority

274+ enqueue outputQ 0 controlPriority $ OFrame frame

275275 atomically $ writeTQueue q body

276276 if endOfStream then do

277277 mcl <- readIORef streamContentLength

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

310310 atomically $ writeTVar streamWindow w

311311 return s

312312

313−stream FrameRSTStream header bs _ _ strm = do

313+stream FrameRSTStream header bs ctx _ strm = do

314314 RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs

315315 let cc = Reset e

316− closed strm cc

316+ closed strm cc ctx

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

318318

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

320− PriorityFrame p <- guardIt $ decodePriorityFrame header bs

321− checkPriority p streamNumber

320+ PriorityFrame newp <- guardIt $ decodePriorityFrame header bs

321+ checkPriority newp streamNumber

322322 -- checkme: this should be tested

323− -- fixme: This works well when the priority gets lower because

324− -- the old higher priority value comes out from the queue quickly

325− -- and the new lower priority is used when enqueuing again.

326− -- But when the priority get higher, it takes time to use the new

327− -- priority.

328− writeIORef streamPriority p

329− -- checkme: this should be tested

330− when (isIdle s) $ prepare outputQ streamNumber p

323+ oldp <- readIORef streamPriority

324+ writeIORef streamPriority newp

325+ if isIdle s then

326+ prepare outputQ streamNumber newp

327+ else do

328+ mx <- delete outputQ streamNumber oldp

329+ case mx of

330+ Nothing -> return ()

331+ Just x -> enqueue outputQ streamNumber newp x

331332 return s

332333

333334-- this ordering is important

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

@@ -60,15 +60,16 @@ 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 :: Connection -> Stream -> IO () -> IO Bool

64−unlessClosed Connection{connSendAll}

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

64+unlessClosed ctx

65+ Connection{connSendAll}

6566 strm@Stream{streamState,streamNumber}

6667 body = E.handle resetStream $ do

6768 state <- readIORef streamState

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

6970 where

7071 resetStream e = do

71− closed strm (ResetByMe e)

72+ closed strm (ResetByMe e) ctx

7273 let rst = resetFrame InternalError streamNumber

7374 connSendAll rst

7475 return False

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

99100 loop

100101

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

102− loop = dequeue outputQ >>= \(out, _) -> switch out

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

103104

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

105106 ignore _ = return ()

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

118119 connSendAll frame

119120 loop

120121 switch (OResponse strm s h aux) = do

121− _ <- unlessClosed conn strm $

122+ _ <- unlessClosed ctx conn strm $

122123 getWindowSize connectionWindow (streamWindow strm) >>=

123124 sendResponse strm s h aux

124125 loop

125126 switch (ONext strm curr) = do

126− _ <- unlessClosed conn strm $ do

127+ _ <- unlessClosed ctx conn strm $ do

127128 lim <- getWindowSize connectionWindow (streamWindow strm)

128129 -- Data frame payload

129130 Next datPayloadLen mnext <- curr lim

@@ -131,7 +132,7 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}

131132 dispatchNext strm mnext

132133 loop

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

134− pushed <- unlessClosed conn oldStrm $ do

135+ pushed <- unlessClosed ctx conn oldStrm $ do

135136 lim <- getWindowSize connectionWindow (streamWindow strm)

136137 -- Write and send the promise.

137138 builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push

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

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

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

185186 -- concurrency.

186− closed strm Finished

187+ closed strm Finished ctx

187188 flushN toFlush

188189

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

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+8−4

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

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

249249 writeIORef streamState (Open JustOpened)

250250

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

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

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

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

253+ cc

254+ Context{outputQ} = do

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

254256 writeIORef streamState (Closed cc)

257+ pri <- readIORef streamPriority

258+ clear outputQ streamNumber pri

255259

256260----------------------------------------------------------------

257261

@@ -297,7 +301,7 @@ enqueueWhenWindowIsOpen outQ out = do

297301 x <- readTVar $ streamWindow strm

298302 check (x > 0)

299303 pri <- readIORef $ streamPriority strm

300− enqueue outQ out pri

304+ enqueue outQ (streamNumber strm) pri out

301305

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

303307enqueueOrSpawnTemporaryWaiter strm outQ out = do

@@ -307,4 +311,4 @@ enqueueOrSpawnTemporaryWaiter strm outQ out = do

307311 void $ forkIO $ enqueueWhenWindowIsOpen outQ out

308312 else do

309313 pri <- readIORef $ streamPriority strm

310− enqueue outQ out pri

314+ enqueue outQ (streamNumber strm) pri out

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

@@ -96,10 +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 Context{outputQ} set strm req me = do

100− closed strm Killed

101− let frame = resetFrame InternalError (streamNumber strm)

102− enqueue outputQ (OFrame frame) highestPriority

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

100+ closed strm Killed ctx

101+ let sid = streamNumber strm

102+ frame = resetFrame InternalError sid

103+ enqueue outputQ sid controlPriority $ OFrame frame

103104 case me of

104105 Nothing -> return ()

105106 Just e -> S.settingsOnException set req e