7ec60d6 · typeable/wai

git.iko.soy/typeable/wai

refactoring for enqueueing.

commit7ec60d62a8a75c93ea4f4bba458a28a039b4f49e authorKazu Yamamoto <kazu@iij.ad.jp> dateFri 18 Dec 2015, 12:11 (11 years ago) parent1a3efe9

5 files changed+25−30

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/HTTP2.hs+1−2 warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+10−10 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+9−16 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+4 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+1−2

warp/Network/Wai/Handler/Warp/HTTP2.hs+1−2

@@ -8,7 +8,6 @@ import qualified Control.Exception as E

88import Control.Monad (when, unless, replicateM_)

99import Data.ByteString (ByteString)

1010import Network.HTTP2

11−import Network.HTTP2.Priority

1211import Network.Socket (SockAddr)

1312import Network.Wai

1413import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

@@ -32,6 +31,7 @@ http2 conn ii addr transport settings readN app = do

3231 ok <- checkPreface

3332 when ok $ do

3433 ctx <- newContext

34+ enqueueOutputControl (outputQ ctx) $ OFrame initialFrame

3535 -- Workers, worker manager and timer manager

3636 mgr <- start settings

3737 let responder = response ii settings ctx mgr

@@ -46,7 +46,6 @@ http2 conn ii addr transport settings readN app = do

4646 -- Receiver

4747 let mkreq = mkRequest ii settings addr

4848 tid <- forkIO $ frameReceiver ctx mkreq readN

49− enqueueControl (outputQ ctx) 0 $ OFrame initialFrame

5049 -- Sender

5150 -- frameSender is the main thread because it ensures to send

5251 -- a goway frame.

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

@@ -17,7 +17,7 @@ import Data.ByteString (ByteString)

1717import qualified Data.ByteString as BS

1818import Data.Maybe (isJust)

1919import Network.HTTP2

20−import Network.HTTP2.Priority

20+import Network.HTTP2.Priority (toPrecedence, delete, prepare)

2121import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

2222import Network.Wai.Handler.Warp.HTTP2.HPACK

2323import Network.Wai.Handler.Warp.HTTP2.Request

@@ -42,12 +42,12 @@ frameReceiver ctx mkreq recvN = loop 0 `E.catch` sendGoaway

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

4343 csid <- readIORef currentStreamId

4444 let !frame = goawayFrame csid err msg

45− enqueueControl outputQ 0 $ OGoaway frame

45+ enqueueOutputControl outputQ $ OGoaway frame

4646 | otherwise = return ()

4747

4848 sendReset err sid = do

4949 let !frame = resetFrame err sid

50− enqueueControl outputQ 0 $ OFrame frame

50+ enqueueOutputControl outputQ $ OFrame frame

5151

5252 loop :: Int -> IO ()

5353 loop !n

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

5757 | otherwise = do

5858 hd <- recvN frameHeaderLength

5959 if BS.null hd then

60− enqueueControl outputQ 0 OFinish

60+ enqueueOutputControl outputQ OFinish

6161 else do

6262 cont <- processStreamGuardingError $ decodeFrameHeader hd

6363 when cont $ loop (n + 1)

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

183183 unless (testAck flags) $ do

184184 modifyIORef' http2settings $ \old -> updateSettings old alist

185185 let !frame = settingsFrame setAck []

186− enqueueControl outputQ 0 $ OSettings frame alist

186+ enqueueOutputControl outputQ $ OSettings frame alist

187187 return True

188188

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

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

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

192192 else do

193193 let !frame = pingFrame bs

194− enqueueControl outputQ 0 $ OFrame frame

194+ enqueueOutputControl outputQ $ OFrame frame

195195 return True

196196

197197control FrameGoAway _ _ Context{outputQ} = do

198− enqueueControl outputQ 0 OFinish

198+ enqueueOutputControl outputQ OFinish

199199 return False

200200

201201control FrameWindowUpdate header bs Context{connectionWindow} = do

@@ -269,7 +269,7 @@ stream FrameData

269269 let !frame1 = windowUpdateFrame 0 payloadLength

270270 !frame2 = windowUpdateFrame streamNumber payloadLength

271271 !frame = frame1 `BS.append` frame2

272− enqueueControl outputQ 0 $ OFrame frame

272+ enqueueOutputControl outputQ $ OFrame frame

273273 atomically $ writeTQueue q body

274274 if endOfStream then do

275275 mcl <- readIORef streamContentLength

@@ -328,8 +328,8 @@ stream FramePriority header bs Context{outputQ,priorityTreeSize} s Stream{stream

328328 else do

329329 mx <- delete outputQ streamNumber oldpre

330330 case mx of

331− Nothing -> return ()

332− Just x -> enqueue outputQ streamNumber newpre x

331+ Nothing -> return ()

332+ Just out -> enqueueOutput outputQ out

333333 return s

334334

335335-- this ordering is important

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

@@ -15,7 +15,7 @@ import qualified Data.ByteString.Builder.Extra as B

1515import Foreign.Ptr

1616import Network.HPACK (setLimitForEncoding)

1717import Network.HTTP2

18−import Network.HTTP2.Priority

18+import Network.HTTP2.Priority (dequeue)

1919import Network.Wai

2020import Network.Wai.Handler.Warp.Buffer

2121import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

@@ -46,19 +46,6 @@ data Leftover = LZero

4646

4747----------------------------------------------------------------

4848

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

50−unlessClosed ctx

51− Connection{connSendAll}

52− strm@Stream{streamState,streamNumber}

53− body = E.handle resetStream $ do

54− state <- readIORef streamState

55− unless (isClosed state) body

56− where

57− resetStream e = do

58− closed ctx strm (ResetByMe e)

59− let rst = resetFrame InternalError streamNumber

60− connSendAll rst

61−

6249getConnectionWindowSizeWhenReady :: Context -> IO WindowSize

6350getConnectionWindowSizeWhenReady Context{connectionWindow} = atomically $ do

6451 w <- readTVar connectionWindow

@@ -156,8 +143,9 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}

156143 maybeEnqueueNext strm mnext binfo

157144 loop

158145

159− whenReadyOrEnqueueAgain strm binfo out body =

160− unlessClosed ctx conn strm $ case binfo of

146+ whenReadyOrEnqueueAgain strm binfo out body = E.handle resetStream $ do

147+ state <- readIORef $ streamState strm

148+ unless (isClosed state) $ case binfo of

161149 Persist tbq -> checkStreaming tbq

162150 _ -> checkStreamWindowSize

163151 where

@@ -173,6 +161,11 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}

173161 enqueueWhenReady (waitStreamWindowSize strm) outputQ out

174162 else

175163 body sws

164+ resetStream e = do

165+ closed ctx strm (ResetByMe e)

166+ let rst = resetFrame InternalError $ streamNumber strm

167+ eout = OFrame rst

168+ enqueueOutputControl outputQ eout

176169

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

178171 -- the buffer are filled.

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

@@ -223,3 +223,7 @@ enqueueOutput outQ out = do

223223 let Stream{..} = outputStream out

224224 pre <- readIORef streamPrecedence

225225 enqueue outQ streamNumber pre out

226+

227+{-# INLINE enqueueOutputControl #-}

228+enqueueOutputControl :: PriorityTree Output -> Output -> IO ()

229+enqueueOutputControl outQ out = enqueueControl outQ 0 out

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

@@ -23,7 +23,6 @@ import Control.Monad (when)

2323import Data.ByteString.Builder (byteString)

2424import qualified Network.HTTP.Types as H

2525import Network.HTTP2

26−import Network.HTTP2.Priority

2726import Network.Wai

2827import Network.Wai.Handler.Warp.File

2928import Network.Wai.Handler.Warp.FileInfoCache

@@ -166,7 +165,7 @@ worker ctx@Context{inputQ,outputQ} set app responder tm = do

166165 Just (Input strm req) -> do

167166 closed ctx strm Killed

168167 let frame = resetFrame InternalError (streamNumber strm)

169− enqueueControl outputQ 0 (OFrame frame)

168+ enqueueOutputControl outputQ $ OFrame frame

170169 case me of

171170 Nothing -> return ()

172171 Just e -> S.settingsOnException set (Just req) e