@@ -57,16 +57,19 @@ unlessClosed ctx
5757 let rst = resetFrame InternalError streamNumber
5858 connSendAll rst
5959
60−getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize
61−getWindowSize connWindow strmWindow = do
62− -- Waiting that the connection window gets open.
63− cw <- atomically $ do
64− w <- readTVar connWindow
65− check (w > 0)
66− return w
67− -- This stream window is greater than 0 thanks to the invariant.
68− sw <- atomically $ readTVar strmWindow
69− return $ min cw sw
60+getConnectionWindowSizeWhenReady :: Context -> IO WindowSize
61+getConnectionWindowSizeWhenReady Context{connectionWindow} = atomically $ do
62+ w <- readTVar connectionWindow
63+ check (w > 0)
64+ return w
65+
66+getStreamWindowSize :: Stream -> IO WindowSize
67+getStreamWindowSize Stream{streamWindow} = atomically $ readTVar streamWindow
68+
69+waitStreamWindowSize :: Stream -> STM ()
70+waitStreamWindowSize Stream{streamWindow} = do
71+ w <- readTVar streamWindow
72+ check (w > 0)
7073
7174frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
7275frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
@@ -101,53 +104,63 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
101104 dyntbl <- readIORef encodeDynamicTable
102105 setLimitForEncoding siz dyntbl
103106 loop
104− switch (OResponse strm rsp aux) pre = do
107+ switch out@(OResponse strm rsp aux) pre = do
105108 writeIORef (streamPrecedence strm) pre
106109 unlessClosed ctx conn strm $ do
107− lim <- getWindowSize connectionWindow (streamWindow strm)
108− -- Header frame and Continuation frame
109− let sid = streamNumber strm
110− endOfStream = case aux of
111− Persist{} -> False
112− Oneshot hb -> not hb
113− len <- headerContinue sid rsp endOfStream
114− let total = len + frameHeaderLength
115− case aux of
116− Oneshot True -> do -- hasBody
117− -- Data frame payload
118− (off, _) <- sendHeadersIfNecessary total
119− let payloadOff = off + frameHeaderLength
120− Next datPayloadLen mnext <-
121− fillResponseBodyGetNext conn ii payloadOff lim rsp
122− fillDataHeaderSend strm total datPayloadLen mnext
123− maybeEnqueueNext strm mnext
124− Oneshot False -> do
125− -- "closed" must be before "connSendAll". If not,
126− -- the context would be switched to the receiver,
127− -- resulting the inconsistency of concurrency.
128− closed ctx strm Finished
129− flushN total
130− Persist sq tvar -> do
131− (off, needSend) <- sendHeadersIfNecessary total
132− let payloadOff = off + frameHeaderLength
133− Next datPayloadLen mnext <-
134− fillStreamBodyGetNext conn payloadOff lim sq tvar strm
135− -- If no data was immediately available, avoid sending an
136− -- empty data frame.
137− if datPayloadLen > 0 then
110+ cws <- getConnectionWindowSizeWhenReady ctx
111+ sws <- getStreamWindowSize strm
112+ if sws == 0 then do
113+ enqueueWhenReady (waitStreamWindowSize strm) outputQ out
114+ else do
115+ let !lim = min cws sws
116+ -- Header frame and Continuation frame
117+ let sid = streamNumber strm
118+ endOfStream = case aux of
119+ Persist{} -> False
120+ Oneshot hb -> not hb
121+ len <- headerContinue sid rsp endOfStream
122+ let total = len + frameHeaderLength
123+ case aux of
124+ Oneshot True -> do -- hasBody
125+ -- Data frame payload
126+ (off, _) <- sendHeadersIfNecessary total
127+ let payloadOff = off + frameHeaderLength
128+ Next datPayloadLen mnext <-
129+ fillResponseBodyGetNext conn ii payloadOff lim rsp
138130 fillDataHeaderSend strm total datPayloadLen mnext
139− else
140− when needSend $ flushN off
141− maybeEnqueueNext strm mnext
131+ maybeEnqueueNext strm mnext
132+ Oneshot False -> do
133+ -- "closed" must be before "connSendAll". If not,
134+ -- the context would be switched to the receiver,
135+ -- resulting the inconsistency of concurrency.
136+ closed ctx strm Finished
137+ flushN total
138+ Persist sq -> do
139+ (off, needSend) <- sendHeadersIfNecessary total
140+ let payloadOff = off + frameHeaderLength
141+ Next datPayloadLen mnext <-
142+ fillStreamBodyGetNext conn payloadOff lim sq strm
143+ -- If no data was immediately available, avoid sending an
144+ -- empty data frame.
145+ if datPayloadLen > 0 then
146+ fillDataHeaderSend strm total datPayloadLen mnext
147+ else
148+ when needSend $ flushN off
149+ maybeEnqueueNext strm mnext
142150 loop
143− switch (ONext strm curr) pre = do
151+ switch out@(ONext strm curr) pre = do
144152 writeIORef (streamPrecedence strm) pre
145153 unlessClosed ctx conn strm $ do
146− lim <- getWindowSize connectionWindow (streamWindow strm)
147− -- Data frame payload
148− Next datPayloadLen mnext <- curr lim
149− fillDataHeaderSend strm 0 datPayloadLen mnext
150− maybeEnqueueNext strm mnext
154+ cws <- getConnectionWindowSizeWhenReady ctx
155+ sws <- getStreamWindowSize strm
156+ if sws == 0 then do
157+ enqueueWhenReady (waitStreamWindowSize strm) outputQ out
158+ else do
159+ let !lim = min cws sws
160+ -- Data frame payload
161+ Next datPayloadLen mnext <- curr lim
162+ fillDataHeaderSend strm 0 datPayloadLen mnext
163+ maybeEnqueueNext strm mnext
151164 loop
152165
153166 -- Flush the connection buffer to the socket, where the first 'n' bytes of
@@ -193,7 +206,7 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
193206 maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()
194207 maybeEnqueueNext strm (CNext next) = do
195208 let out = ONext strm next
196− enqueueOrSpawnTemporaryWaiter strm outputQ out
209+ enqueueOutput outputQ out
197210 -- If the streaming is not finished, it must already have been
198211 -- written to the 'TVar' owned by 'waiter', which will
199212 -- put it back into the queue when more output becomes available.
@@ -289,13 +302,13 @@ fileStartEnd _ _ = error "fileStartEnd"
289302
290303----------------------------------------------------------------
291304
292−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
305+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> Stream -> IO Next
293306fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
294− off lim sq tvar strm = do
307+ off lim sq strm = do
295308 let datBuf = connWriteBuffer `plusPtr` off
296309 room = min (connBufferSize - off) lim
297310 (leftover, cont, len) <- runStreamBuilder datBuf room sq
298− nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len
311+ return $ nextForStream connWriteBuffer connBufferSize sq strm leftover cont len
299312
300313----------------------------------------------------------------
301314
@@ -349,8 +362,8 @@ runStreamBuilder buf0 room0 sq = loop buf0 room0 0
349362 Just SFlush -> return (LZero, True, total)
350363 Just SFinish -> return (LZero, False, total)
351364
352−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
353−fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
365+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> Stream -> DynaNext
366+fillBufStream buf0 siz0 leftover0 sq strm lim0 = do
354367 let payloadBuf = buf0 `plusPtr` frameHeaderLength
355368 room0 = min (siz0 - frameHeaderLength) lim0
356369 case leftover0 of
@@ -368,7 +381,7 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
368381 void $ copy payloadBuf bs1
369382 getNext (LTwo bs2 writer) True room0
370383 where
371− getNext = nextForStream buf0 siz0 sq tvar strm
384+ getNext l b r = return $ nextForStream buf0 siz0 sq strm l b r
372385 write writer1 buf room sofar = do
373386 (len, signal) <- writer1 buf room
374387 case signal of
@@ -383,18 +396,12 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
383396 let !total = sofar + len
384397 getNext (LTwo bs writer) True total
385398
386−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
399+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> Stream
387400 -> Leftover -> Bool -> BytesFilled
388− -> IO Next
389−nextForStream _ _ _ tvar _ _ False len = do
390− atomically $ writeTVar tvar SyncFinish
391− return $ Next len CFinish
392−nextForStream buf siz sq tvar strm LZero True len = do
393− let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)
394− atomically $ writeTVar tvar $ SyncNext out
395− return $ Next len CNone
396−nextForStream buf siz sq tvar strm leftover True len =
397− return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))
401+ -> Next
402+nextForStream _ _ _ _ _ False len = Next len CFinish
403+nextForStream buf siz sq strm leftOrZero True len =
404+ Next len (CNext (fillBufStream buf siz leftOrZero sq strm))
398405
399406----------------------------------------------------------------
400407