@@ -57,16 +57,24 @@ 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)
73+
74+waitStreaming :: TBQueue a -> STM ()
75+waitStreaming tbq = do
76+ isEmpty <- isEmptyTBQueue tbq
77+ check (isEmpty == False)
7078
7179frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
7280frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
@@ -101,55 +109,76 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
101109 dyntbl <- readIORef encodeDynamicTable
102110 setLimitForEncoding siz dyntbl
103111 loop
104− switch (OResponse strm rsp aux) pre = do
112+ switch out@(ONext strm curr binfo) pre = do
113+ writeIORef (streamPrecedence strm) pre
114+ whenReadyOrEnqueueAgain strm binfo out $ \sws -> do
115+ cws <- getConnectionWindowSizeWhenReady ctx
116+ let !lim = min cws sws
117+ -- Data frame payload
118+ Next datPayloadLen mnext <- curr lim
119+ fillDataHeaderSend strm 0 datPayloadLen mnext
120+ maybeEnqueueNext strm mnext binfo
121+ loop
122+ switch out@(OResponse strm rsp binfo) pre = do
105123 writeIORef (streamPrecedence strm) pre
106− unlessClosed ctx conn strm $ do
107− lim <- getWindowSize connectionWindow (streamWindow strm)
124+ whenReadyOrEnqueueAgain strm binfo out $ \sws -> do
125+ cws <- getConnectionWindowSizeWhenReady ctx
126+ let !lim = min cws sws
108127 -- Header frame and Continuation frame
109128 let sid = streamNumber strm
110− endOfStream = case aux of
111− Persist{} -> False
112− Oneshot hb -> not hb
129+ endOfStream = case binfo of
130+ Persist{} -> False
131+ OneshotWithBody -> False
132+ OneshotWithoutBody -> True
113133 len <- headerContinue sid rsp endOfStream
114134 let total = len + frameHeaderLength
115− case aux of
116− Oneshot True -> do -- hasBody
135+ case binfo of
136+ OneshotWithBody -> do
117137 -- Data frame payload
118138 (off, _) <- sendHeadersIfNecessary total
119139 let payloadOff = off + frameHeaderLength
120140 Next datPayloadLen mnext <-
121141 fillResponseBodyGetNext conn ii payloadOff lim rsp
122142 fillDataHeaderSend strm total datPayloadLen mnext
123− maybeEnqueueNext strm mnext
124− Oneshot False -> do
143+ maybeEnqueueNext strm mnext binfo
144+ OneshotWithoutBody -> do
125145 -- "closed" must be before "connSendAll". If not,
126146 -- the context would be switched to the receiver,
127147 -- resulting the inconsistency of concurrency.
128148 closed ctx strm Finished
129149 flushN total
130− Persist sq tvar -> do
150+ Persist sq -> do
131151 (off, needSend) <- sendHeadersIfNecessary total
132152 let payloadOff = off + frameHeaderLength
133153 Next datPayloadLen mnext <-
134− fillStreamBodyGetNext conn payloadOff lim sq tvar strm
154+ fillStreamBodyGetNext conn payloadOff lim sq strm
135155 -- If no data was immediately available, avoid sending an
136156 -- empty data frame.
137157 if datPayloadLen > 0 then
138158 fillDataHeaderSend strm total datPayloadLen mnext
139159 else
140160 when needSend $ flushN off
141− maybeEnqueueNext strm mnext
142− loop
143− switch (ONext strm curr) pre = do
144− writeIORef (streamPrecedence strm) pre
145− 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
161+ maybeEnqueueNext strm mnext binfo
151162 loop
152163
164+ whenReadyOrEnqueueAgain strm binfo out body =
165+ unlessClosed ctx conn strm $ case binfo of
166+ Persist tbq -> checkStreaming tbq
167+ _ -> checkStreamWindowSize
168+ where
169+ checkStreaming tbq = do
170+ isEmpty <- atomically $ isEmptyTBQueue tbq
171+ if isEmpty then
172+ enqueueWhenReady (waitStreaming tbq) outputQ out
173+ else
174+ checkStreamWindowSize
175+ checkStreamWindowSize = do
176+ sws <- getStreamWindowSize strm
177+ if sws == 0 then do
178+ enqueueWhenReady (waitStreamWindowSize strm) outputQ out
179+ else
180+ body sws
181+
153182 -- Flush the connection buffer to the socket, where the first 'n' bytes of
154183 -- the buffer are filled.
155184 flushN :: Int -> IO ()
@@ -188,16 +217,12 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
188217 -- True if the connection buffer has room for a 1-byte data frame.
189218 canFitDataFrame total = total + frameHeaderLength < connBufferSize
190219
191− -- Re-enqueue the stream in the output queue if more output is immediately
192− -- available; do nothing otherwise.
193− maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()
194− maybeEnqueueNext strm (CNext next) = do
195− let out = ONext strm next
196− enqueueOrSpawnTemporaryWaiter strm outputQ out
197− -- If the streaming is not finished, it must already have been
198− -- written to the 'TVar' owned by 'waiter', which will
199− -- put it back into the queue when more output becomes available.
200− maybeEnqueueNext _ _ = return ()
220+ -- Re-enqueue the stream in the output queue.
221+ maybeEnqueueNext :: Stream -> Maybe DynaNext -> BodyInfo -> IO ()
222+ maybeEnqueueNext strm (Just next) binfo = do
223+ let !out = ONext strm next binfo
224+ enqueueOutput outputQ out
225+ maybeEnqueueNext _ _ _ = return ()
201226
202227
203228 -- Send headers if there is not room for a 1-byte data frame, and return
@@ -215,19 +240,18 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
215240 buf = connWriteBuffer `plusPtr` otherLen
216241 total = otherLen + frameHeaderLength + datPayloadLen
217242 flag = case mnext of
218− CFinish -> setEndStream defaultFlags
243+ Nothing -> setEndStream defaultFlags
219244 _ -> defaultFlags
220245 fillFrameHeader FrameData datPayloadLen sid flag buf
221246 -- "closed" must be before "connSendAll". If not,
222247 -- the context would be switched to the receiver,
223248 -- resulting the inconsistency of concurrency.
224249 case mnext of
225− CFinish -> closed ctx strm Finished
226− _ -> return ()
250+ Nothing -> closed ctx strm Finished
251+ _ -> return ()
227252 flushN total
228− atomically $ do
229− modifyTVar' connectionWindow (subtract datPayloadLen)
230− modifyTVar' (streamWindow strm) (subtract datPayloadLen)
253+ atomically $ modifyTVar' connectionWindow (subtract datPayloadLen)
254+ atomically $ modifyTVar' (streamWindow strm) (subtract datPayloadLen)
231255
232256 fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf
233257 where
@@ -289,13 +313,13 @@ fileStartEnd _ _ = error "fileStartEnd"
289313
290314----------------------------------------------------------------
291315
292−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next
316+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> Stream -> IO Next
293317fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
294− off lim sq tvar strm = do
318+ off lim sq strm = do
295319 let datBuf = connWriteBuffer `plusPtr` off
296320 room = min (connBufferSize - off) lim
297321 (leftover, cont, len) <- runStreamBuilder datBuf room sq
298− nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len
322+ return $ nextForStream connWriteBuffer connBufferSize sq strm leftover cont len
299323
300324----------------------------------------------------------------
301325
@@ -323,11 +347,11 @@ fillBufBuilder buf0 siz0 leftover lim = do
323347
324348nextForBuilder :: Buffer -> BufSize -> BytesFilled -> B.Next -> Next
325349nextForBuilder _ _ len B.Done
326− = Next len CFinish
350+ = Next len Nothing
327351nextForBuilder buf siz len (B.More _ writer)
328− = Next len (CNext (fillBufBuilder buf siz (LOne writer)))
352+ = Next len $ Just (fillBufBuilder buf siz (LOne writer))
329353nextForBuilder buf siz len (B.Chunk bs writer)
330− = Next len (CNext (fillBufBuilder buf siz (LTwo bs writer)))
354+ = Next len $ Just (fillBufBuilder buf siz (LTwo bs writer))
331355
332356----------------------------------------------------------------
333357
@@ -349,8 +373,8 @@ runStreamBuilder buf0 room0 sq = loop buf0 room0 0
349373 Just SFlush -> return (LZero, True, total)
350374 Just SFinish -> return (LZero, False, total)
351375
352−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext
353−fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
376+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> Stream -> DynaNext
377+fillBufStream buf0 siz0 leftover0 sq strm lim0 = do
354378 let payloadBuf = buf0 `plusPtr` frameHeaderLength
355379 room0 = min (siz0 - frameHeaderLength) lim0
356380 case leftover0 of
@@ -368,7 +392,7 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
368392 void $ copy payloadBuf bs1
369393 getNext (LTwo bs2 writer) True room0
370394 where
371− getNext = nextForStream buf0 siz0 sq tvar strm
395+ getNext l b r = return $ nextForStream buf0 siz0 sq strm l b r
372396 write writer1 buf room sofar = do
373397 (len, signal) <- writer1 buf room
374398 case signal of
@@ -383,18 +407,12 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do
383407 let !total = sofar + len
384408 getNext (LTwo bs writer) True total
385409
386−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream
410+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> Stream
387411 -> 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))
412+ -> Next
413+nextForStream _ _ _ _ _ False len = Next len Nothing
414+nextForStream buf siz sq strm leftOrZero True len =
415+ Next len $ Just (fillBufStream buf siz leftOrZero sq strm)
398416
399417----------------------------------------------------------------
400418
@@ -409,10 +427,10 @@ fillBufFile buf siz h bytes refresh lim = do
409427 return $ nextForFile len buf siz h bytes' refresh
410428
411429nextForFile :: BytesFilled -> Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> Next
412−nextForFile 0 _ _ _ _ _ = Next 0 CFinish
413−nextForFile len _ _ _ 0 _ = Next len CFinish
430+nextForFile 0 _ _ _ _ _ = Next 0 Nothing
431+nextForFile len _ _ _ 0 _ = Next len Nothing
414432nextForFile len buf siz h bytes refresh =
415− Next len (CNext (fillBufFile buf siz h bytes refresh))
433+ Next len $ Just (fillBufFile buf siz h bytes refresh)
416434#else
417435fillBufFile :: Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> DynaNext
418436fillBufFile buf siz fd start bytes refresh lim = do
@@ -424,10 +442,10 @@ fillBufFile buf siz fd start bytes refresh lim = do
424442 return $ nextForFile len buf siz fd (start + len') (bytes - len') refresh
425443
426444nextForFile :: BytesFilled -> Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> Next
427−nextForFile 0 _ _ _ _ _ _ = Next 0 CFinish
428−nextForFile len _ _ _ _ 0 _ = Next len CFinish
445+nextForFile 0 _ _ _ _ _ _ = Next 0 Nothing
446+nextForFile len _ _ _ _ 0 _ = Next len Nothing
429447nextForFile len buf siz fd start bytes refresh =
430− Next len (CNext (fillBufFile buf siz fd start bytes refresh))
448+ Next len $ Just (fillBufFile buf siz fd start bytes refresh)
431449#endif
432450
433451mini :: Int -> Integer -> Int