e3fd9a2 · typeable/wai

10import Control.Concurrent (forkIO)

11import Control.Concurrent.STM

12import qualified Control.Exception as E

13−import Control.Monad (void, unless)

13+import Control.Monad (unless, void, when)

14import qualified Data.ByteString as BS

15import qualified Data.ByteString.Builder.Extra as B

16import Foreign.Ptr

111 case aux of

112 Oneshot True -> do -- hasBody

113 -- Data frame payload

114− off <- sendHeadersIfNecessary total

114+ (off, _) <- sendHeadersIfNecessary total

115+ let payloadOff = off + frameHeaderLength

116+ Next datPayloadLen mnext <-

115− Next datPayloadLen mnext <- fillResponseBodyGetNext conn ii off lim rsp

117+ fillResponseBodyGetNext conn ii payloadOff lim rsp

116− fillDataHeaderSend strm total datPayloadLen mnext pri

118+ fillDataHeaderSend strm total datPayloadLen mnext

119+ maybeEnqueueNext strm mnext pri

117 Oneshot False -> do

120 Oneshot False -> do

118 -- "closed" must be before "connSendAll". If not,

121 -- "closed" must be before "connSendAll". If not,

119 -- the context would be switched to the receiver,

122 -- the context would be switched to the receiver,

120 -- resulting the inconsistency of concurrency.

123 -- resulting the inconsistency of concurrency.

121 closed ctx strm Finished

124 closed ctx strm Finished

122− bufferIO connWriteBuffer total connSendAll

125+ flushN total

123 Persist sq tvar -> do

126 Persist sq tvar -> do

124− off <- sendHeadersIfNecessary total

127+ (off, needSend) <- sendHeadersIfNecessary total

128+ let payloadOff = off + frameHeaderLength

129+ Next datPayloadLen mnext <-

125− Next datPayloadLen mnext <- fillStreamBodyGetNext conn off lim sq tvar

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

126− fillDataHeaderSend strm total datPayloadLen mnext pri

134+ fillDataHeaderSend strm total datPayloadLen mnext

135+ else

136+ when needSend $ flushN off

137+ maybeEnqueueNext strm mnext pri

127 loop

138 loop

128 switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do

139 switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do

129 checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do

140 checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do

130 -- Data frame payload

141 -- Data frame payload

131 Next datPayloadLen mnext <- curr lim

142 Next datPayloadLen mnext <- curr lim

132− fillDataHeaderSend strm 0 datPayloadLen mnext pri

143+ fillDataHeaderSend strm 0 datPayloadLen mnext

144+ maybeEnqueueNext strm mnext pri

133 loop

145 loop

134

146

147+ -- Flush the connection buffer to the socket, where the first 'n' bytes of

148+ -- the buffer are filled.

149+ flushN :: Int -> IO ()

150+ flushN n = bufferIO connWriteBuffer n connSendAll

151+

135 headerContinue sid rsp endOfStream = do

152 headerContinue sid rsp endOfStream = do

136 builder <- hpackEncodeHeader ctx ii settings rsp

153 builder <- hpackEncodeHeader ctx ii settings rsp

137 (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim

154 (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim

144

161

145 continue _ len B.Done = return len

162 continue _ len B.Done = return len

146 continue sid len (B.More _ writer) = do

163 continue sid len (B.More _ writer) = do

147− let total = len + frameHeaderLength

164+ flushN $ len + frameHeaderLength

148− bufferIO connWriteBuffer total connSendAll

149 (len', signal') <- writer bufHeaderPayload headerPayloadLim

165 (len', signal') <- writer bufHeaderPayload headerPayloadLim

150 let flag = case signal' of

166 let flag = case signal' of

151 B.Done -> setEndHeader defaultFlags

167 B.Done -> setEndHeader defaultFlags

153 fillFrameHeader FrameContinuation len' sid flag connWriteBuffer

169 fillFrameHeader FrameContinuation len' sid flag connWriteBuffer

154 continue sid len' signal'

170 continue sid len' signal'

155 continue sid len (B.Chunk bs writer) = do

171 continue sid len (B.Chunk bs writer) = do

156− let total = len + frameHeaderLength

172+ flushN $ len + frameHeaderLength

157− bufferIO connWriteBuffer total connSendAll

158 let (bs1,bs2) = BS.splitAt headerPayloadLim bs

173 let (bs1,bs2) = BS.splitAt headerPayloadLim bs

159 len' = BS.length bs1

174 len' = BS.length bs1

160 void $ copy bufHeaderPayload bs1

175 void $ copy bufHeaderPayload bs1

164 else

179 else

165 continue sid len' (B.Chunk bs2 writer)

180 continue sid len' (B.Chunk bs2 writer)

166

181

167− sendHeadersIfNecessary total = do

182+ -- True if the connection buffer has room for a 1-byte data frame.

168− let datPayloadOff = total + frameHeaderLength

183+ canFitDataFrame total = total + frameHeaderLength < connBufferSize

169− if datPayloadOff < connBufferSize then

184+

170− return datPayloadOff

185+ -- Re-enqueue the stream in the output queue if more output is immediately

171− else do

186+ -- available; do nothing otherwise. If the stream is not finished, it must

172− bufferIO connWriteBuffer total connSendAll

187+ -- already have been written to the 'TVar' owned by 'waiter', which will

173− return frameHeaderLength

188+ -- 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 ()

192+

174

193

194+ -- Send headers if there is not room for a 1-byte data frame, and return

195+ -- the offset of the next frame's first header byte and whether the headers

196+ -- still need to be sent.

197+ sendHeadersIfNecessary total

198+ | canFitDataFrame total = return (total, True)

199+ | otherwise = do

200+ flushN total

201+ return (0, False)

202+

175− fillDataHeaderSend strm otherLen datPayloadLen mnext pri = do

203+ fillDataHeaderSend strm otherLen datPayloadLen mnext = do

176 -- Data frame header

204 -- Data frame header

177 let sid = streamNumber strm

205 let sid = streamNumber strm

178 buf = connWriteBuffer `plusPtr` otherLen

206 buf = connWriteBuffer `plusPtr` otherLen

187 case mnext of

215 case mnext of

188 CFinish -> closed ctx strm Finished

216 CFinish -> closed ctx strm Finished

189 _ -> return ()

217 _ -> return ()

190− bufferIO connWriteBuffer total connSendAll

218+ flushN total

191 atomically $ do

219 atomically $ do

192 modifyTVar' connectionWindow (subtract datPayloadLen)

220 modifyTVar' connectionWindow (subtract datPayloadLen)

193 modifyTVar' (streamWindow strm) (subtract datPayloadLen)

221 modifyTVar' (streamWindow strm) (subtract datPayloadLen)

194− case mnext of

195− CFinish -> return ()

196− CNext next -> enqueue outputQ (ONext strm next) pri

197− CNone -> return ()

198

222

199 fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf

223 fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf

200 where

224 where