git.iko.soy/typeable/wai
on master help sign in
Merge pull request #423 from yesodweb/refactor-http2
9c4ac89Kazu Yamamoto
warp/Network/Wai/Handler/Warp/HTTP2/Receiver.hs+15−5
compare with master
unifiedsplit
@@ -92,7 +92,7 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
92 control ftyp header pl ctx
92 control ftyp header pl ctx
93 | otherwise = do
93 | otherwise = do
94 checkContinued
94 checkContinued
95− strm@Stream{streamState,streamContentLength} <- getStream
95+ strm@Stream{streamState,streamContentLength,streamPriority} <- getStream
96 pl <- recvN payloadLength
96 pl <- recvN payloadLength
97 state <- readIORef streamState
97 state <- readIORef streamState
98 state' <- stream ftyp header pl ctx state strm
98 state' <- stream ftyp header pl ctx state strm
@@ -103,21 +103,23 @@ frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway
103 Just vh -> do
103 Just vh -> do
104 when (isJust (vhCL vh) && vhCL vh /= Just 0) $
104 when (isJust (vhCL vh) && vhCL vh /= Just 0) $
105 E.throwIO $ StreamError ProtocolError streamId
105 E.throwIO $ StreamError ProtocolError streamId
106+ writeIORef streamPriority pri
106 writeIORef streamState HalfClosed
107 writeIORef streamState HalfClosed
107 let req = mkreq vh (return "")
108 let req = mkreq vh (return "")
108− atomically $ writeTQueue inputQ $ Input strm req pri
109+ atomically $ writeTQueue inputQ $ Input strm req
109 Nothing -> E.throwIO $ StreamError ProtocolError streamId
110 Nothing -> E.throwIO $ StreamError ProtocolError streamId
110 Open (HasBody hdr pri) -> do
111 Open (HasBody hdr pri) -> do
111 resetContinued
112 resetContinued
112 case validateHeaders hdr of
113 case validateHeaders hdr of
113 Just vh -> do
114 Just vh -> do
114 q <- newTQueueIO
115 q <- newTQueueIO
116+ writeIORef streamPriority pri
115 writeIORef streamState (Open (Body q))
117 writeIORef streamState (Open (Body q))
116 writeIORef streamContentLength $ vhCL vh
118 writeIORef streamContentLength $ vhCL vh
117 readQ <- newReadBody q
119 readQ <- newReadBody q
118 bodySource <- mkSource readQ
120 bodySource <- mkSource readQ
119 let req = mkreq vh (readSource bodySource)
121 let req = mkreq vh (readSource bodySource)
120− atomically $ writeTQueue inputQ $ Input strm req pri
122+ atomically $ writeTQueue inputQ $ Input strm req
121 Nothing -> E.throwIO $ StreamError ProtocolError streamId
123 Nothing -> E.throwIO $ StreamError ProtocolError streamId
122 s@(Open Continued{}) -> do
124 s@(Open Continued{}) -> do
123 setContinued
125 setContinued
@@ -314,10 +316,18 @@ stream FrameRSTStream header bs ctx _ strm = do
314 closed ctx strm cc
316 closed ctx strm cc
315 return $ Closed cc -- will be written to streamState again
317 return $ Closed cc -- will be written to streamState again
316
318
317−stream FramePriority header bs Context{outputQ} s Stream{streamNumber} = do
319+stream FramePriority header bs Context{outputQ} s Stream{streamNumber,streamPriority} = do
318 PriorityFrame p <- guardIt $ decodePriorityFrame header bs
320 PriorityFrame p <- guardIt $ decodePriorityFrame header bs
319 checkPriority p streamNumber
321 checkPriority p streamNumber
322+ -- 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
320− prepare outputQ streamNumber p
330+ when (isIdle s) $ prepare outputQ streamNumber p
321 return s
331 return s
322
332
323-- this ordering is important
333-- this ordering is important
warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+72−74
@@ -7,7 +7,6 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
7#if __GLASGOW_HASKELL__ < 709
7#if __GLASGOW_HASKELL__ < 709
8import Control.Applicative
8import Control.Applicative
9#endif
9#endif
10−import Control.Concurrent (forkIO)
11import Control.Concurrent.STM
10import Control.Concurrent.STM
12import qualified Control.Exception as E
11import qualified Control.Exception as E
13import Control.Monad (unless, void, when)
12import Control.Monad (unless, void, when)
@@ -58,22 +57,16 @@ unlessClosed ctx
58 let rst = resetFrame InternalError streamNumber
57 let rst = resetFrame InternalError streamNumber
59 connSendAll rst
58 connSendAll rst
60
59
61−checkWindowSize :: TVar WindowSize -> TVar WindowSize -> PriorityTree Output -> Output -> Priority -> (WindowSize -> IO ()) -> IO ()
60+getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
62−checkWindowSize connWindow strmWindow outQ out pri body = do
61+getWindowSize connWindow strmWindow = do
62+ -- Waiting that the connection window gets open.
63 cw <- atomically $ do
63 cw <- atomically $ do
64 w <- readTVar connWindow
64 w <- readTVar connWindow
65 check (w > 0)
65 check (w > 0)
66 return w
66 return w
67+ -- This stream window is greater than 0 thanks to the invariant.
67 sw <- atomically $ readTVar strmWindow
68 sw <- atomically $ readTVar strmWindow
68− if sw == 0 then
69− void $ forkIO $ do
70− -- checkme: if connection is closed? GC throws dead lock error?
71− atomically $ do
72− x <- readTVar strmWindow
73− check (x > 0)
74− enqueue outQ out pri
75− else
76− body (min cw sw)
69+ return $ min cw sw
77
70
78frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
71frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
79frameSender ctx@Context{outputQ,connectionWindow}
72frameSender ctx@Context{outputQ,connectionWindow}
@@ -89,59 +82,60 @@ frameSender ctx@Context{outputQ,connectionWindow}
89 connSendAll initialFrame
82 connSendAll initialFrame
90 loop
83 loop
91
84
85+ -- ignoring the old priority because the value might be changed.
92− loop = dequeue outputQ >>= \(out, pri) -> switch out pri
86+ loop = dequeue outputQ >>= \(out, _) -> switch out
93
87
94 ignore :: E.SomeException -> IO ()
88 ignore :: E.SomeException -> IO ()
95 ignore _ = return ()
89 ignore _ = return ()
96
90
97− switch OFinish _ = return ()
91+ switch OFinish = return ()
98− switch (OGoaway frame) _ = connSendAll frame
92+ switch (OGoaway frame) = connSendAll frame
99− switch (OFrame frame) _ = do
93+ switch (OFrame frame) = do
100 connSendAll frame
94 connSendAll frame
101 loop
95 loop
102− switch out@(OResponse strm rsp aux) pri = unlessClosed ctx conn strm $ do
96+ switch (OResponse strm rsp aux) = unlessClosed ctx conn strm $ do
103− checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
97+ lim <- getWindowSize connectionWindow (streamWindow strm)
104− -- Header frame and Continuation frame
98+ -- Header frame and Continuation frame
105− let sid = streamNumber strm
99+ let sid = streamNumber strm
106− endOfStream = case aux of
100+ endOfStream = case aux of
107− Persist{} -> False
101+ Persist{} -> False
108− Oneshot hb -> not hb
102+ Oneshot hb -> not hb
109− len <- headerContinue sid rsp endOfStream
103+ len <- headerContinue sid rsp endOfStream
110− let total = len + frameHeaderLength
104+ let total = len + frameHeaderLength
111− case aux of
105+ case aux of
112− Oneshot True -> do -- hasBody
106+ Oneshot True -> do -- hasBody
113− -- Data frame payload
107+ -- Data frame payload
114− (off, _) <- sendHeadersIfNecessary total
108+ (off, _) <- sendHeadersIfNecessary total
115− let payloadOff = off + frameHeaderLength
109+ let payloadOff = off + frameHeaderLength
116− Next datPayloadLen mnext <-
110+ Next datPayloadLen mnext <-
117− fillResponseBodyGetNext conn ii payloadOff lim rsp
111+ fillResponseBodyGetNext conn ii payloadOff lim rsp
112+ fillDataHeaderSend strm total datPayloadLen mnext
113+ maybeEnqueueNext strm mnext
114+ Oneshot False -> do
115+ -- "closed" must be before "connSendAll". If not,
116+ -- the context would be switched to the receiver,
117+ -- resulting the inconsistency of concurrency.
118+ closed ctx strm Finished
119+ flushN total
120+ Persist sq tvar -> do
121+ (off, needSend) <- sendHeadersIfNecessary total
122+ let payloadOff = off + frameHeaderLength
123+ Next datPayloadLen mnext <-
124+ fillStreamBodyGetNext conn payloadOff lim sq tvar strm
125+ -- If no data was immediately available, avoid sending an
126+ -- empty data frame.
127+ if datPayloadLen > 0 then
118 fillDataHeaderSend strm total datPayloadLen mnext
128 fillDataHeaderSend strm total datPayloadLen mnext
119− maybeEnqueueNext strm mnext pri
120− Oneshot False -> do
121− -- "closed" must be before "connSendAll". If not,
122− -- the context would be switched to the receiver,
123− -- resulting the inconsistency of concurrency.
124− closed ctx strm Finished
125− flushN total
126− Persist sq tvar -> do
127− (off, needSend) <- sendHeadersIfNecessary total
128− let payloadOff = off + frameHeaderLength
129− Next datPayloadLen mnext <-
130− fillStreamBodyGetNext conn payloadOff lim sq tvar
131− -- If no data was immediately available, avoid sending an
132− -- empty data frame.
133− if datPayloadLen > 0 then
134− fillDataHeaderSend strm total datPayloadLen mnext
135− else
129+ else
136− when needSend $ flushN off
130+ when needSend $ flushN off
137− maybeEnqueueNext strm mnext pri
131+ maybeEnqueueNext strm mnext
138 loop
132 loop
139− switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do
133+ switch (ONext strm curr) = unlessClosed ctx conn strm $ do
140− checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
134+ lim <- getWindowSize connectionWindow (streamWindow strm)
141− -- Data frame payload
135+ -- Data frame payload
142− Next datPayloadLen mnext <- curr lim
136+ Next datPayloadLen mnext <- curr lim
143− fillDataHeaderSend strm 0 datPayloadLen mnext
137+ fillDataHeaderSend strm 0 datPayloadLen mnext
144− maybeEnqueueNext strm mnext pri
138+ maybeEnqueueNext strm mnext
145 loop
139 loop
146
140
147 -- Flush the connection buffer to the socket, where the first 'n' bytes of
141 -- Flush the connection buffer to the socket, where the first 'n' bytes of
@@ -183,12 +177,15 @@ frameSender ctx@Context{outputQ,connectionWindow}
183 canFitDataFrame total = total + frameHeaderLength < connBufferSize
177 canFitDataFrame total = total + frameHeaderLength < connBufferSize
184
178
185 -- Re-enqueue the stream in the output queue if more output is immediately
179 -- Re-enqueue the stream in the output queue if more output is immediately
186− -- available; do nothing otherwise. If the stream is not finished, it must
180+ -- available; do nothing otherwise.
181+ maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()
182+ maybeEnqueueNext strm (CNext next) = do
183+ let out = ONext strm next
184+ enqueueOrSpawnTemporaryWaiter strm outputQ out
185+ -- If the streaming is not finished, it must already have been
187− -- already have been written to the 'TVar' owned by 'waiter', which will
186+ -- written to the 'TVar' owned by 'waiter', which will
188 -- put it back into the queue when more output becomes available.
187 -- put it back into the queue when more output becomes available.
189− maybeEnqueueNext :: Stream -> Control DynaNext -> Priority -> IO ()
188+ maybeEnqueueNext _ _ = return ()
190− maybeEnqueueNext strm (CNext next) = enqueue outputQ (ONext strm next)
191− maybeEnqueueNext _ _ = const $ return ()
192
189
193
190
194 -- Send headers if there is not room for a 1-byte data frame, and return
191 -- Send headers if there is not room for a 1-byte data frame, and return
@@ -282,13 +279,13 @@ fileStartEnd _ (Just part) =
282
279
283----------------------------------------------------------------
280----------------------------------------------------------------
284
281
285−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> IO Next
282+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
286fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
283fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
287− off lim sq tvar = do
284+ off lim sq tvar strm = do
288 let datBuf = connWriteBuffer `plusPtr` off
285 let datBuf = connWriteBuffer `plusPtr` off
289 room = min (connBufferSize - off) lim
286 room = min (connBufferSize - off) lim
290 (leftover, cont, len) <- runStreamBuilder datBuf room sq
287 (leftover, cont, len) <- runStreamBuilder datBuf room sq
291− nextForStream connWriteBuffer connBufferSize sq tvar leftover cont len
288+ nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len
292
289
293----------------------------------------------------------------
290----------------------------------------------------------------
294
291
@@ -342,8 +339,8 @@ runStreamBuilder buf0 room0 sq = loop buf0 room0 0
342 Just SFlush -> return (LZero, True, total)
339 Just SFlush -> return (LZero, True, total)
343 Just SFinish -> return (LZero, False, total)
340 Just SFinish -> return (LZero, False, total)
344
341
345−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> DynaNext
342+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
346−fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
343+fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
347 let payloadBuf = buf0 `plusPtr` frameHeaderLength
344 let payloadBuf = buf0 `plusPtr` frameHeaderLength
348 room0 = min (siz0 - frameHeaderLength) lim0
345 room0 = min (siz0 - frameHeaderLength) lim0
349 case leftover0 of
346 case leftover0 of
@@ -361,7 +358,7 @@ fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
361 void $ copy payloadBuf bs1
358 void $ copy payloadBuf bs1
362 getNext (LTwo bs2 writer) True room0
359 getNext (LTwo bs2 writer) True room0
363 where
360 where
364− getNext = nextForStream buf0 siz0 sq tvar
361+ getNext = nextForStream buf0 siz0 sq tvar strm
365 write writer1 buf room sofar = do
362 write writer1 buf room sofar = do
366 (len, signal) <- writer1 buf room
363 (len, signal) <- writer1 buf room
367 case signal of
364 case signal of
@@ -376,17 +373,18 @@ fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
376 let !total = sofar + len
373 let !total = sofar + len
377 getNext (LTwo bs writer) True total
374 getNext (LTwo bs writer) True total
378
375
379−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync
376+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
380 -> Leftover -> Bool -> BytesFilled
377 -> Leftover -> Bool -> BytesFilled
381 -> IO Next
378 -> IO Next
382−nextForStream _ _ _ tvar _ False len = do
379+nextForStream _ _ _ tvar _ _ False len = do
383 atomically $ writeTVar tvar SyncFinish
380 atomically $ writeTVar tvar SyncFinish
384 return $ Next len CFinish
381 return $ Next len CFinish
385−nextForStream buf siz sq tvar LZero True len = do
382+nextForStream buf siz sq tvar strm LZero True len = do
386− atomically $ writeTVar tvar $ SyncNext (fillBufStream buf siz LZero sq tvar)
383+ let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)
384+ atomically $ writeTVar tvar $ SyncNext out
387 return $ Next len CNone
385 return $ Next len CNone
388−nextForStream buf siz sq tvar leftover True len =
386+nextForStream buf siz sq tvar strm leftover True len =
389− return $ Next len (CNext (fillBufStream buf siz leftover sq tvar))
387+ return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))
390
388
391----------------------------------------------------------------
389----------------------------------------------------------------
392
390
warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+31−2
@@ -7,6 +7,7 @@ import Data.ByteString.Builder (Builder)
7#if __GLASGOW_HASKELL__ < 709
7#if __GLASGOW_HASKELL__ < 709
8import Control.Applicative ((<$>),(<*>))
8import Control.Applicative ((<$>),(<*>))
9#endif
9#endif
10+import Control.Concurrent (forkIO)
10import Control.Concurrent.STM
11import Control.Concurrent.STM
11import Control.Exception (SomeException)
12import Control.Exception (SomeException)
12import Control.Monad (void)
13import Control.Monad (void)
@@ -39,7 +40,7 @@ isHTTP2 tls = useHTTP2
39
40
40----------------------------------------------------------------
41----------------------------------------------------------------
41
42
42−data Input = Input Stream Request Priority
43+data Input = Input Stream Request
43
44
44----------------------------------------------------------------
45----------------------------------------------------------------
45
46
@@ -64,6 +65,11 @@ data Output = OFinish
64 | OResponse Stream Response Aux
65 | OResponse Stream Response Aux
65 | ONext Stream DynaNext
66 | ONext Stream DynaNext
66
67
68+outputStream :: Output -> Stream
69+outputStream (OResponse strm _ _) = strm
70+outputStream (ONext strm _) = strm
71+outputStream _ = error "outputStream"
72+
67----------------------------------------------------------------
73----------------------------------------------------------------
68
74
69data Sequence = SFinish
75data Sequence = SFinish
@@ -72,7 +78,7 @@ data Sequence = SFinish
72
78
73data Sync = SyncNone
79data Sync = SyncNone
74 | SyncFinish
80 | SyncFinish
75− | SyncNext DynaNext
81+ | SyncNext Output
76
82
77data Aux = Oneshot Bool
83data Aux = Oneshot Bool
78 | Persist (TBQueue Sequence) (TVar Sync)
84 | Persist (TBQueue Sequence) (TVar Sync)
@@ -170,6 +176,7 @@ data Stream = Stream {
170 , streamContentLength :: IORef (Maybe Int)
176 , streamContentLength :: IORef (Maybe Int)
171 , streamBodyLength :: IORef Int
177 , streamBodyLength :: IORef Int
172 , streamWindow :: TVar WindowSize
178 , streamWindow :: TVar WindowSize
179+ , streamPriority :: IORef Priority
173 }
180 }
174
181
175instance Show Stream where
182instance Show Stream where
@@ -180,6 +187,7 @@ newStream sid win = Stream sid <$> newIORef Idle
180 <*> newIORef Nothing
187 <*> newIORef Nothing
181 <*> newIORef 0
188 <*> newIORef 0
182 <*> newTVarIO win
189 <*> newTVarIO win
190+ <*> newIORef defaultPriority
183
191
184----------------------------------------------------------------
192----------------------------------------------------------------
185
193
@@ -227,3 +235,24 @@ insert strmtbl k v = reaperAdd strmtbl (k,v)
227
235
228search :: StreamTable -> M.Key -> IO (Maybe Stream)
236search :: StreamTable -> M.Key -> IO (Maybe Stream)
229search strmtbl k = M.lookup k <$> reaperRead strmtbl
237search strmtbl k = M.lookup k <$> reaperRead strmtbl
238+
239+
240+-- INVARIANT: streams in the output queue have non-zero window size.
241+enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO ()
242+enqueueWhenWindowIsOpen outQ out = do
243+ let strm = outputStream out
244+ atomically $ do
245+ x <- readTVar $ streamWindow strm
246+ check (x > 0)
247+ pri <- readIORef $ streamPriority strm
248+ enqueue outQ out pri
249+
250+enqueueOrSpawnTemporaryWaiter :: Stream -> PriorityTree Output -> Output -> IO ()
251+enqueueOrSpawnTemporaryWaiter strm outQ out = do
252+ sw <- atomically $ readTVar $ streamWindow strm
253+ if sw == 0 then
254+ -- This waiter waits only for the stream window.
255+ void $ forkIO $ enqueueWhenWindowIsOpen outQ out
256+ else do
257+ pri <- readIORef $ streamPriority strm
258+ enqueue outQ out pri
warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+22−19
@@ -36,14 +36,14 @@ import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceive
36-- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'.
36-- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'.
37-- This type implements the second argument (Response -> IO ResponseReceived)
37-- This type implements the second argument (Response -> IO ResponseReceived)
38-- with extra arguments.
38-- with extra arguments.
39−type Responder = ThreadContinue -> T.Handle -> Stream -> Priority -> Request ->
39+type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->
40 Response -> IO ResponseReceived
40 Response -> IO ResponseReceived
41
41
42-- | This function is passed to workers.
42-- | This function is passed to workers.
43-- They also pass 'Response's from 'Application's to this function.
43-- They also pass 'Response's from 'Application's to this function.
44-- This function enqueues commands for the HTTP/2 sender.
44-- This function enqueues commands for the HTTP/2 sender.
45response :: Context -> Manager -> Responder
45response :: 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
47 case rsp of
47 case rsp of
48 ResponseStream _ _ strmbdy -> do
48 ResponseStream _ _ strmbdy -> do
49 -- We must not exit this WAI application.
49 -- We must not exit this WAI application.
@@ -59,22 +59,24 @@ response Context{outputQ} mgr tconf th strm pri req rsp = do
59 -- So, let's serialize 'Builder' with a designated queue.
59 -- So, let's serialize 'Builder' with a designated queue.
60 sq <- newTBQueueIO 10 -- fixme: hard coding: 10
60 sq <- newTBQueueIO 10 -- fixme: hard coding: 10
61 tvar <- newTVarIO SyncNone
61 tvar <- newTVarIO SyncNone
62− enqueue outputQ (OResponse strm rsp (Persist sq tvar)) pri
62+ let out = OResponse strm rsp (Persist sq tvar)
63+ -- Since we must not enqueue an empty queue to the priority
64+ -- queue, we spawn a thread to ensure that the designated
65+ -- queue is not empty.
66+ void $ forkIO $ waiter tvar sq outputQ
67+ atomically $ writeTVar tvar (SyncNext out)
63 let push b = do
68 let push b = do
64 atomically $ writeTBQueue sq (SBuilder b)
69 atomically $ writeTBQueue sq (SBuilder b)
65 T.tickle th
70 T.tickle th
66 flush = atomically $ writeTBQueue sq SFlush
71 flush = atomically $ writeTBQueue sq SFlush
67− -- Since we must not enqueue an empty queue to the priority
68− -- queue, we spawn a thread to ensure that the designated
69− -- queue is not empty.
70− void $ forkIO $ waiter tvar sq (enqueue outputQ) strm pri
71 strmbdy push flush
72 strmbdy push flush
72 atomically $ writeTBQueue sq SFinish
73 atomically $ writeTBQueue sq SFinish
73 _ -> do
74 _ -> do
74 setThreadContinue tconf True
75 setThreadContinue tconf True
75 let hasBody = requestMethod req /= H.methodHead
76 let hasBody = requestMethod req /= H.methodHead
76 && R.hasBody (responseStatus rsp)
77 && R.hasBody (responseStatus rsp)
77− enqueue outputQ (OResponse strm rsp (Oneshot hasBody)) pri
78+ out = OResponse strm rsp (Oneshot hasBody)
79+ enqueueOrSpawnTemporaryWaiter strm outputQ out
78 return ResponseReceived
80 return ResponseReceived
79
81
80data Break = Break deriving (Show, Typeable)
82data Break = Break deriving (Show, Typeable)
@@ -93,11 +95,11 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
93 setThreadContinue tcont True
95 setThreadContinue tcont True
94 ex <- E.try $ do
96 ex <- E.try $ do
95 T.pause th
97 T.pause th
96− Input strm req pri <- atomically $ readTQueue inputQ
98+ Input strm req <- atomically $ readTQueue inputQ
97 setStreamInfo sinfo strm req
99 setStreamInfo sinfo strm req
98 T.resume th
100 T.resume th
99 T.tickle th
101 T.tickle th
100− app req $ responder tcont th strm pri req
102+ app req $ responder tcont th strm req
101 cont1 <- case ex of
103 cont1 <- case ex of
102 Right ResponseReceived -> return True
104 Right ResponseReceived -> return True
103 Left e@(SomeException _)
105 Left e@(SomeException _)
@@ -126,26 +128,27 @@ worker ctx@Context{inputQ,outputQ} set tm app responder = do
126 Just e -> S.settingsOnException set (Just req) e
128 Just e -> S.settingsOnException set (Just req) e
127 clearStreamInfo sinfo
129 clearStreamInfo sinfo
128
130
129−waiter :: TVar Sync -> TBQueue Sequence
131+waiter :: TVar Sync -> TBQueue Sequence -> PriorityTree Output -> IO ()
130− -> (Output -> Priority -> IO ()) -> Stream -> Priority
131− -> IO ()
132−waiter tvar sq enq strm pri = do
132+waiter tvar sq outQ = do
133+ -- waiting for actions other than SyncNone
133 mx <- atomically $ do
134 mx <- atomically $ do
134 mout <- readTVar tvar
135 mout <- readTVar tvar
135 case mout of
136 case mout of
136 SyncNone -> retry
137 SyncNone -> retry
137− SyncNext nxt -> do
138+ SyncNext out -> do
138 writeTVar tvar SyncNone
139 writeTVar tvar SyncNone
139− return $ Just nxt
140+ return $ Just out
140 SyncFinish -> return Nothing
141 SyncFinish -> return Nothing
141 case mx of
142 case mx of
142 Nothing -> return ()
143 Nothing -> return ()
143− Just next -> do
144+ Just out -> do
145+ -- ensuring that the streaming queue is not empty.
144 atomically $ do
146 atomically $ do
145 isEmpty <- isEmptyTBQueue sq
147 isEmpty <- isEmptyTBQueue sq
146 when isEmpty retry
148 when isEmpty retry
147− enq (ONext strm next) pri
149+ -- ensuring that stream window is greater than 0.
150+ enqueueWhenWindowIsOpen outQ out
148− waiter tvar sq enq strm pri
151+ waiter tvar sq outQ
149
152
150----------------------------------------------------------------
153----------------------------------------------------------------
151
154