4e0731d · typeable/wai

git.iko.soy/typeable/wai

including Priority in Stream.

commit4e0731d9f9a1709429792087d17ca0e1c530cac6 authorKazu Yamamoto <kazu@iij.ad.jp> dateTue 11 Aug 2015, 16:03 (11 years ago) parent615e384

4 files changed+28−18

compare with master

unifiedsplit

warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+5−3 warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+9−6 warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+3−1 warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+11−8

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

@@ -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} <- getStream

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

9696 pl <- recvN payloadLength

9797 state <- readIORef streamState

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

@@ -103,21 +103,23 @@ 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

106107 writeIORef streamState HalfClosed

107108 let req = mkreq vh (return "")

108− atomically $ writeTQueue inputQ $ Input strm req pri

109+ atomically $ writeTQueue inputQ $ Input strm req

109110 Nothing -> E.throwIO $ StreamError ProtocolError streamId

110111 Open (HasBody hdr pri) -> do

111112 resetContinued

112113 case validateHeaders hdr of

113114 Just vh -> do

114115 q <- newTQueueIO

116+ writeIORef streamPriority pri

115117 writeIORef streamState (Open (Body q))

116118 writeIORef streamContentLength $ vhCL vh

117119 readQ <- newReadBody q

118120 bodySource <- mkSource readQ

119121 let req = mkreq vh (readSource bodySource)

120− atomically $ writeTQueue inputQ $ Input strm req pri

122+ atomically $ writeTQueue inputQ $ Input strm req

121123 Nothing -> E.throwIO $ StreamError ProtocolError streamId

122124 s@(Open Continued{}) -> do

123125 setContinued

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

@@ -89,17 +89,19 @@ frameSender ctx@Context{outputQ,connectionWindow}

8989 connSendAll initialFrame

9090 loop

9191

92− loop = dequeue outputQ >>= \(out, pri) -> switch out pri

92+ -- ignoring the old priority because the value might be changed.

93+ loop = dequeue outputQ >>= \(out, _) -> switch out

9394

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

9596 ignore _ = return ()

9697

97− switch OFinish _ = return ()

98− switch (OGoaway frame) _ = connSendAll frame

99− switch (OFrame frame) _ = do

98+ switch OFinish = return ()

99+ switch (OGoaway frame) = connSendAll frame

100+ switch (OFrame frame) = do

100101 connSendAll frame

101102 loop

102− switch out@(OResponse strm rsp aux) pri = unlessClosed ctx conn strm $ do

103+ switch out@(OResponse strm rsp aux) = unlessClosed ctx conn strm $ do

104+ pri <- readIORef $ streamPriority strm

103105 checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do

104106 -- Header frame and Continuation frame

105107 let sid = streamNumber strm

@@ -136,7 +138,8 @@ frameSender ctx@Context{outputQ,connectionWindow}

136138 when needSend $ flushN off

137139 maybeEnqueueNext strm mnext pri

138140 loop

139− switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do

141+ switch out@(ONext strm curr) = unlessClosed ctx conn strm $ do

142+ pri <- readIORef $ streamPriority strm

140143 checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do

141144 -- Data frame payload

142145 Next datPayloadLen mnext <- curr lim

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

@@ -39,7 +39,7 @@ isHTTP2 tls = useHTTP2

3939

4040----------------------------------------------------------------

4141

42−data Input = Input Stream Request Priority

42+data Input = Input Stream Request

4343

4444----------------------------------------------------------------

4545

@@ -170,6 +170,7 @@ data Stream = Stream {

170170 , streamContentLength :: IORef (Maybe Int)

171171 , streamBodyLength :: IORef Int

172172 , streamWindow :: TVar WindowSize

173+ , streamPriority :: IORef Priority

173174 }

174175

175176instance Show Stream where

@@ -180,6 +181,7 @@ newStream sid win = Stream sid <$> newIORef Idle

180181 <*> newIORef Nothing

181182 <*> newIORef 0

182183 <*> newTVarIO win

184+ <*> newIORef defaultPriority

183185

184186----------------------------------------------------------------

185187

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

@@ -36,14 +36,14 @@ import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceive

3636-- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'.

3737-- This type implements the second argument (Response -> IO ResponseReceived)

3838-- with extra arguments.

39−type Responder = ThreadContinue -> T.Handle -> Stream -> Priority -> Request ->

39+type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->

4040 Response -> IO ResponseReceived

4141

4242-- | This function is passed to workers.

4343-- They also pass 'Response's from 'Application's to this function.

4444-- This function enqueues commands for the HTTP/2 sender.

4545response :: Context -> Manager -> Responder

46−response Context{outputQ} mgr tconf th strm pri req rsp = do

46+response Context{outputQ} mgr tconf th strm req rsp = do

4747 case rsp of

4848 ResponseStream _ _ strmbdy -> do

4949 -- We must not exit this WAI application.

@@ -59,6 +59,7 @@ response Context{outputQ} mgr tconf th strm pri req rsp = do

5959 -- So, let's serialize 'Builder' with a designated queue.

6060 sq <- newTBQueueIO 10 -- fixme: hard coding: 10

6161 tvar <- newTVarIO SyncNone

62+ pri <- readIORef $ streamPriority strm

6263 enqueue outputQ (OResponse strm rsp (Persist sq tvar)) pri

6364 let push b = do

6465 atomically $ writeTBQueue sq (SBuilder b)

@@ -67,13 +68,14 @@ response Context{outputQ} mgr tconf th strm pri req rsp = do

6768 -- Since we must not enqueue an empty queue to the priority

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

6970 -- queue is not empty.

70− void $ forkIO $ waiter tvar sq (enqueue outputQ) strm pri

71+ void $ forkIO $ waiter tvar sq (enqueue outputQ) strm

7172 strmbdy push flush

7273 atomically $ writeTBQueue sq SFinish

7374 _ -> do

7475 setThreadContinue tconf True

7576 let hasBody = requestMethod req /= H.methodHead

7677 && R.hasBody (responseStatus rsp)

78+ pri <- readIORef $ streamPriority strm

7779 enqueue outputQ (OResponse strm rsp (Oneshot hasBody)) pri

7880 return ResponseReceived

7981

@@ -93,11 +95,11 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do

9395 setThreadContinue tcont True

9496 ex <- E.try $ do

9597 T.pause th

96− Input strm req pri <- atomically $ readTQueue inputQ

98+ Input strm req <- atomically $ readTQueue inputQ

9799 setStreamInfo sinfo strm req

98100 T.resume th

99101 T.tickle th

100− app req $ responder tcont th strm pri req

102+ app req $ responder tcont th strm req

101103 cont1 <- case ex of

102104 Right ResponseReceived -> return True

103105 Left e@(SomeException _)

@@ -127,9 +129,9 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do

127129 clearStreamInfo sinfo

128130

129131waiter :: TVar Sync -> TBQueue Sequence

130− -> (Output -> Priority -> IO ()) -> Stream -> Priority

132+ -> (Output -> Priority -> IO ()) -> Stream

131133 -> IO ()

132−waiter tvar sq enq strm pri = do

134+waiter tvar sq enq strm = do

133135 mx <- atomically $ do

134136 mout <- readTVar tvar

135137 case mout of

@@ -144,8 +146,9 @@ waiter tvar sq enq strm pri = do

144146 atomically $ do

145147 isEmpty <- isEmptyTBQueue sq

146148 when isEmpty retry

149+ pri <- readIORef $ streamPriority strm

147150 enq (ONext strm next) pri

148− waiter tvar sq enq strm pri

151+ waiter tvar sq enq strm

149152

150153----------------------------------------------------------------

151154