@@ -7,7 +7,6 @@ module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
77#if __GLASGOW_HASKELL__ < 709
88import Control.Applicative
99#endif
10−import Control.Concurrent (forkIO)
1110import Control.Concurrent.STM
1211import qualified Control.Exception as E
1312import Control.Monad (unless, void, when)
@@ -58,22 +57,16 @@ unlessClosed ctx
5857 let rst = resetFrame InternalError streamNumber
5958 connSendAll rst
6059
61−checkWindowSize :: TVar WindowSize -> TVar WindowSize -> PriorityTree Output -> Output -> Priority -> (WindowSize -> IO ()) -> IO ()
62−checkWindowSize connWindow strmWindow outQ out pri body = do
60+getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
61+getWindowSize connWindow strmWindow = do
62+ -- Waiting that the connection window gets open.
6363 cw <- atomically $ do
6464 w <- readTVar connWindow
6565 check (w > 0)
6666 return w
67+ -- This stream window is greater than 0 thanks to the invariant.
6768 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
7770
7871frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
7972frameSender ctx@Context{outputQ,connectionWindow}
@@ -89,59 +82,60 @@ frameSender ctx@Context{outputQ,connectionWindow}
8982 connSendAll initialFrame
9083 loop
9184
92− loop = dequeue outputQ >>= \(out, pri) -> switch out pri
85+ -- ignoring the old priority because the value might be changed.
86+ loop = dequeue outputQ >>= \(out, _) -> switch out
9387
9488 ignore :: E.SomeException -> IO ()
9589 ignore _ = return ()
9690
97− switch OFinish _ = return ()
98− switch (OGoaway frame) _ = connSendAll frame
99− switch (OFrame frame) _ = do
91+ switch OFinish = return ()
92+ switch (OGoaway frame) = connSendAll frame
93+ switch (OFrame frame) = do
10094 connSendAll frame
10195 loop
102− switch out@(OResponse strm rsp aux) pri = unlessClosed ctx conn strm $ do
103− checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
104− -- Header frame and Continuation frame
105− let sid = streamNumber strm
106− endOfStream = case aux of
107− Persist{} -> False
108− Oneshot hb -> not hb
109− len <- headerContinue sid rsp endOfStream
110− let total = len + frameHeaderLength
111− case aux of
112− Oneshot True -> do -- hasBody
113− -- Data frame payload
114− (off, _) <- sendHeadersIfNecessary total
115− let payloadOff = off + frameHeaderLength
116− Next datPayloadLen mnext <-
117− fillResponseBodyGetNext conn ii payloadOff lim rsp
96+ switch (OResponse strm rsp aux) = unlessClosed ctx conn strm $ do
97+ lim <- getWindowSize connectionWindow (streamWindow strm)
98+ -- Header frame and Continuation frame
99+ let sid = streamNumber strm
100+ endOfStream = case aux of
101+ Persist{} -> False
102+ Oneshot hb -> not hb
103+ len <- headerContinue sid rsp endOfStream
104+ let total = len + frameHeaderLength
105+ case aux of
106+ Oneshot True -> do -- hasBody
107+ -- Data frame payload
108+ (off, _) <- sendHeadersIfNecessary total
109+ let payloadOff = off + frameHeaderLength
110+ Next datPayloadLen mnext <-
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
118128 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
136− when needSend $ flushN off
137− maybeEnqueueNext strm mnext pri
129+ else
130+ when needSend $ flushN off
131+ maybeEnqueueNext strm mnext
138132 loop
139− switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do
140− checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
141− -- Data frame payload
142− Next datPayloadLen mnext <- curr lim
143− fillDataHeaderSend strm 0 datPayloadLen mnext
144− maybeEnqueueNext strm mnext pri
133+ switch (ONext strm curr) = unlessClosed ctx conn strm $ do
134+ lim <- getWindowSize connectionWindow (streamWindow strm)
135+ -- Data frame payload
136+ Next datPayloadLen mnext <- curr lim
137+ fillDataHeaderSend strm 0 datPayloadLen mnext
138+ maybeEnqueueNext strm mnext
145139 loop
146140
147141 -- Flush the connection buffer to the socket, where the first 'n' bytes of
@@ -183,12 +177,15 @@ frameSender ctx@Context{outputQ,connectionWindow}
183177 canFitDataFrame total = total + frameHeaderLength < connBufferSize
184178
185179 -- 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
187− -- already have been written to the 'TVar' owned by 'waiter', which will
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
186+ -- written to the 'TVar' owned by 'waiter', which will
188187 -- put it back into the queue when more output becomes available.
189− maybeEnqueueNext :: Stream -> Control DynaNext -> Priority -> IO ()
190− maybeEnqueueNext strm (CNext next) = enqueue outputQ (ONext strm next)
191− maybeEnqueueNext _ _ = const $ return ()
188+ maybeEnqueueNext _ _ = return ()
192189
193190
194191 -- Send headers if there is not room for a 1-byte data frame, and return
@@ -282,13 +279,13 @@ fileStartEnd _ (Just part) =
282279
283280----------------------------------------------------------------
284281
285−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> IO Next
282+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
286283fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
287− off lim sq tvar = do
284+ off lim sq tvar strm = do
288285 let datBuf = connWriteBuffer `plusPtr` off
289286 room = min (connBufferSize - off) lim
290287 (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
292289
293290----------------------------------------------------------------
294291
@@ -342,8 +339,8 @@ runStreamBuilder buf0 room0 sq = loop buf0 room0 0
342339 Just SFlush -> return (LZero, True, total)
343340 Just SFinish -> return (LZero, False, total)
344341
345−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> DynaNext
346−fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
342+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
343+fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
347344 let payloadBuf = buf0 `plusPtr` frameHeaderLength
348345 room0 = min (siz0 - frameHeaderLength) lim0
349346 case leftover0 of
@@ -361,7 +358,7 @@ fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
361358 void $ copy payloadBuf bs1
362359 getNext (LTwo bs2 writer) True room0
363360 where
364− getNext = nextForStream buf0 siz0 sq tvar
361+ getNext = nextForStream buf0 siz0 sq tvar strm
365362 write writer1 buf room sofar = do
366363 (len, signal) <- writer1 buf room
367364 case signal of
@@ -376,17 +373,18 @@ fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
376373 let !total = sofar + len
377374 getNext (LTwo bs writer) True total
378375
379−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync
376+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
380377 -> Leftover -> Bool -> BytesFilled
381378 -> IO Next
382−nextForStream _ _ _ tvar _ False len = do
379+nextForStream _ _ _ tvar _ _ False len = do
383380 atomically $ writeTVar tvar SyncFinish
384381 return $ Next len CFinish
385−nextForStream buf siz sq tvar LZero True len = do
386− atomically $ writeTVar tvar $ SyncNext (fillBufStream buf siz LZero sq tvar)
382+nextForStream buf siz sq tvar strm LZero True len = do
383+ let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)
384+ atomically $ writeTVar tvar $ SyncNext out
387385 return $ Next len CNone
388−nextForStream buf siz sq tvar leftover True len =
389− return $ Next len (CNext (fillBufStream buf siz leftover sq tvar))
386+nextForStream buf siz sq tvar strm leftover True len =
387+ return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))
390388
391389----------------------------------------------------------------
392390