@@ -71,6 +71,11 @@ waitStreamWindowSize Stream{streamWindow} = do
7171 w <- readTVar streamWindow
7272 check (w > 0)
7373
74+waitStreaming :: TBQueue a -> STM ()
75+waitStreaming tbq = do
76+ isEmpty <- isEmptyTBQueue tbq
77+ check (isEmpty == False)
78+
7479frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
7580frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
7681 conn@Connection{connWriteBuffer,connBufferSize,connSendAll}
@@ -104,65 +109,75 @@ frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}
104109 dyntbl <- readIORef encodeDynamicTable
105110 setLimitForEncoding siz dyntbl
106111 loop
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
107122 switch out@(OResponse strm rsp binfo) pre = do
108123 writeIORef (streamPrecedence strm) pre
109− unlessClosed ctx conn strm $ do
110− sws <- getStreamWindowSize strm
111− if sws == 0 then do
112− enqueueWhenReady (waitStreamWindowSize strm) outputQ out
113− else do
114− cws <- getConnectionWindowSizeWhenReady ctx
115− let !lim = min cws sws
116− -- Header frame and Continuation frame
117− let sid = streamNumber strm
118− endOfStream = case binfo of
119− Persist{} -> False
120− OneshotWithBody -> False
121− OneshotWithoutBody -> True
122− len <- headerContinue sid rsp endOfStream
123− let total = len + frameHeaderLength
124− case binfo of
125− OneshotWithBody -> do
126− -- Data frame payload
127− (off, _) <- sendHeadersIfNecessary total
128− let payloadOff = off + frameHeaderLength
129− Next datPayloadLen mnext <-
130− fillResponseBodyGetNext conn ii payloadOff lim rsp
124+ whenReadyOrEnqueueAgain strm binfo out $ \sws -> do
125+ cws <- getConnectionWindowSizeWhenReady ctx
126+ let !lim = min cws sws
127+ -- Header frame and Continuation frame
128+ let sid = streamNumber strm
129+ endOfStream = case binfo of
130+ Persist{} -> False
131+ OneshotWithBody -> False
132+ OneshotWithoutBody -> True
133+ len <- headerContinue sid rsp endOfStream
134+ let total = len + frameHeaderLength
135+ case binfo of
136+ OneshotWithBody -> do
137+ -- Data frame payload
138+ (off, _) <- sendHeadersIfNecessary total
139+ let payloadOff = off + frameHeaderLength
140+ Next datPayloadLen mnext <-
141+ fillResponseBodyGetNext conn ii payloadOff lim rsp
142+ fillDataHeaderSend strm total datPayloadLen mnext
143+ maybeEnqueueNext strm mnext binfo
144+ OneshotWithoutBody -> do
145+ -- "closed" must be before "connSendAll". If not,
146+ -- the context would be switched to the receiver,
147+ -- resulting the inconsistency of concurrency.
148+ closed ctx strm Finished
149+ flushN total
150+ Persist sq -> do
151+ (off, needSend) <- sendHeadersIfNecessary total
152+ let payloadOff = off + frameHeaderLength
153+ Next datPayloadLen mnext <-
154+ fillStreamBodyGetNext conn payloadOff lim sq strm
155+ -- If no data was immediately available, avoid sending an
156+ -- empty data frame.
157+ if datPayloadLen > 0 then
131158 fillDataHeaderSend strm total datPayloadLen mnext
132− maybeEnqueueNext strm mnext binfo
133− OneshotWithoutBody -> do
134− -- "closed" must be before "connSendAll". If not,
135− -- the context would be switched to the receiver,
136− -- resulting the inconsistency of concurrency.
137− closed ctx strm Finished
138− flushN total
139− Persist sq -> do
140− (off, needSend) <- sendHeadersIfNecessary total
141− let payloadOff = off + frameHeaderLength
142− Next datPayloadLen mnext <-
143− fillStreamBodyGetNext conn payloadOff lim sq strm
144− -- If no data was immediately available, avoid sending an
145− -- empty data frame.
146− if datPayloadLen > 0 then
147− fillDataHeaderSend strm total datPayloadLen mnext
148− else
149− when needSend $ flushN off
150− maybeEnqueueNext strm mnext binfo
159+ else
160+ when needSend $ flushN off
161+ maybeEnqueueNext strm mnext binfo
151162 loop
152− switch out@(ONext strm curr binfo) pre = do
153− writeIORef (streamPrecedence strm) pre
154− unlessClosed ctx conn strm $ do
163+
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
155176 sws <- getStreamWindowSize strm
156177 if sws == 0 then do
157178 enqueueWhenReady (waitStreamWindowSize strm) outputQ out
158− else do
159− cws <- getConnectionWindowSizeWhenReady ctx
160− let !lim = min cws sws
161− -- Data frame payload
162− Next datPayloadLen mnext <- curr lim
163− fillDataHeaderSend strm 0 datPayloadLen mnext
164− maybeEnqueueNext strm mnext binfo
165− loop
179+ else
180+ body sws
166181
167182 -- Flush the connection buffer to the socket, where the first 'n' bytes of
168183 -- the buffer are filled.