db9ae02 · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

Cleanup: handle trailers directly in sender.

db9ae02Andrew Pritchard11 years ago

warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+40−34

compare with master

unifiedsplit

@@ -119,8 +119,8 @@ frameSender ctx@Context{outputQ,connectionWindow}

119 lim <- getWindowSize connectionWindow (streamWindow strm)

120 -- Data frame payload

121 Next datPayloadLen mnext <- curr lim

122− fillDataHeaderSend strm 0 datPayloadLen mnext

122+ fillDataHeaderSend strm 0 datPayloadLen

123− maybeEnqueueNext strm mnext

123+ dispatchNext strm mnext

124 loop

125 switch (OPush oldStrm push mvar strm s h aux) = do

126 pushed <- unlessClosed conn oldStrm $ do

@@ -134,23 +134,26 @@ frameSender ctx@Context{outputQ,connectionWindow}

134 sendResponse strm s h aux lim

135 putMVar mvar pushed

136 loop

137− switch (OTrailers strm []) = do

138− _ <- unlessClosed conn strm $ do

139− -- An empty data frame should be used to end the stream instead

140− -- when there are no trailers.

141− let flag = setEndStream defaultFlags

142− fillFrameHeader FrameData 0 (streamNumber strm) flag connWriteBuffer

143− closed strm Finished

144− flushN frameHeaderLength

145− loop

146 switch (OTrailers strm trailers) = do

137 switch (OTrailers strm trailers) = do

147 _ <- unlessClosed conn strm $ do

138 _ <- unlessClosed conn strm $ do

148 -- Trailers always indicate the end of a stream; send them in

139 -- Trailers always indicate the end of a stream; send them in

149− -- consecutive header+continuation frames and end the stream.

140+ -- consecutive header+continuation frames and end the stream. Some

141+ -- clients dislike empty headers frames, so end the stream with an

142+ -- empty data frame instead, as recommended by the spec.

143+ toFlush <- case trailers of

144+ [] -> frameHeaderLength <$ fillFrameHeader FrameData 0

145+ (streamNumber strm)

146+ (setEndStream defaultFlags)

147+ connWriteBuffer

148+ _ -> do

150− builder <- hpackEncodeCIHeaders ctx trailers

149+ builder <- hpackEncodeCIHeaders ctx trailers

151− off <- headerContinue (streamNumber strm) builder True

150+ off <- headerContinue (streamNumber strm) builder True

151+ return (off + frameHeaderLength)

152+ -- 'closed' must be before 'flushN'. If not, the context would be

153+ -- switched to the receiver, resulting in the inconsistency of

154+ -- concurrency.

152 closed strm Finished

155 closed strm Finished

153− flushN $ off + frameHeaderLength

156+ flushN toFlush

154 loop

157 loop

155

158

156 -- Send the response headers and as much of the response as is immediately

159 -- Send the response headers and as much of the response as is immediately

@@ -169,10 +172,10 @@ frameSender ctx@Context{outputQ,connectionWindow}

169 -- If no data was immediately available, avoid sending an

172 -- If no data was immediately available, avoid sending an

170 -- empty data frame.

173 -- empty data frame.

171 if datPayloadLen > 0 then

174 if datPayloadLen > 0 then

172− fillDataHeaderSend strm total datPayloadLen mnext

175+ fillDataHeaderSend strm total datPayloadLen

173 else

176 else

174 when needSend $ flushN off

177 when needSend $ flushN off

175− maybeEnqueueNext strm mnext

178+ dispatchNext strm mnext

176

179

177

180

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

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

@@ -223,16 +226,25 @@ frameSender ctx@Context{outputQ,connectionWindow}

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

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

224 canFitDataFrame total = total + frameHeaderLength < connBufferSize

227 canFitDataFrame total = total + frameHeaderLength < connBufferSize

225

228

226− -- Re-enqueue the stream in the output queue if more output is immediately

229+ -- Take the appropriate action based on the given 'Control':

230+ -- - If more output is immediately available, re-enqueue the stream in the

231+ -- output queue.

232+ -- - If the output is over and trailers are available, send them now and

233+ -- end the stream.

234+ -- - If we've drained the queue and handed the stream back to its waiter,

227− -- available; do nothing otherwise.

235+ -- do nothing.

236+ --

237+ -- This is done after sending any part of the stream body, so it's shared

238+ -- by 'sendResponse' and @switch (ONext ...)@.

228− maybeEnqueueNext :: Stream -> Control DynaNext -> IO ()

239+ dispatchNext :: Stream -> Control DynaNext -> IO ()

229− maybeEnqueueNext strm (CNext next) = do

240+ dispatchNext strm (CNext next) = do

230 let out = ONext strm next

241 let out = ONext strm next

231 enqueueOrSpawnTemporaryWaiter strm outputQ out

242 enqueueOrSpawnTemporaryWaiter strm outputQ out

232− -- If the streaming is not finished, it must already have been

243+ dispatchNext strm (CFinish trailers) = do

233− -- written to the 'TVar' owned by 'waiter', which will

244+ let out = OTrailers strm trailers

234− -- put it back into the queue when more output becomes available.

245+ pri <- readIORef $ streamPriority strm

235− maybeEnqueueNext _ _ = return ()

246+ enqueue outputQ out pri

247+ dispatchNext _ _ = return ()

236

248

237

249

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

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

@@ -244,18 +256,12 @@ frameSender ctx@Context{outputQ,connectionWindow}

244 flushN total

256 flushN total

245 return (0, False)

257 return (0, False)

246

258

247− fillDataHeaderSend strm otherLen datPayloadLen mnext = do

259+ fillDataHeaderSend strm otherLen datPayloadLen = do

248 -- Data frame header

260 -- Data frame header

249 let sid = streamNumber strm

261 let sid = streamNumber strm

250 buf = connWriteBuffer `plusPtr` otherLen

262 buf = connWriteBuffer `plusPtr` otherLen

251 total = otherLen + frameHeaderLength + datPayloadLen

263 total = otherLen + frameHeaderLength + datPayloadLen

252 fillFrameHeader FrameData datPayloadLen sid defaultFlags buf

264 fillFrameHeader FrameData datPayloadLen sid defaultFlags buf

253− -- "closed" must be before "connSendAll". If not,

254− -- the context would be switched to the receiver,

255− -- resulting the inconsistency of concurrency.

256− case mnext of

257− CFinish -> closed strm Finished

258− _ -> return ()

259 flushN total

265 flushN total

260 atomically $ do

266 atomically $ do

261 modifyTVar' connectionWindow (subtract datPayloadLen)

267 modifyTVar' connectionWindow (subtract datPayloadLen)

@@ -406,8 +412,8 @@ nextForStream :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence

406 -> TVar Sync -> Stream -> Leftover -> Maybe Trailers

412 -> TVar Sync -> Stream -> Leftover -> Maybe Trailers

407 -> BytesFilled -> IO Next

413 -> BytesFilled -> IO Next

408nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do

414nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do

409− atomically $ writeTVar tvar $ SyncFinish trailers

415+ atomically $ writeTVar tvar $ SyncFinish

410− return $ Next len CNone

416+ return $ Next len $ CFinish trailers

411nextForStream ii buf siz sq tvar strm LZero Nothing len = do

417nextForStream ii buf siz sq tvar strm LZero Nothing len = do

412 let out = ONext strm (fillBufStream ii buf siz LZero sq tvar strm)

418 let out = ONext strm (fillBufStream ii buf siz LZero sq tvar strm)

413 atomically $ writeTVar tvar $ SyncNext out

419 atomically $ writeTVar tvar $ SyncNext out

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+13−6

@@ -46,14 +46,21 @@ data Input = Input Stream Request

47----------------------------------------------------------------

49+-- | The result of writing data from a stream's queue into the buffer.

49−data Control a = CFinish

50+data Control a = CFinish Trailers

51+ -- ^ The stream has ended, and the trailers should be sent.

50 | CNext a

52 | CNext a

53+ -- ^ The stream has more data immediately available, and we

54+ -- should re-enqueue it when the stream window becomes open.

51 | CNone

55 | CNone

56+ -- ^ The stream queue has been drained and we've handed it off

57+ -- to its dedicated waiter thread, which will re-enqueue it when

58+ -- more data is available.

53instance Show (Control a) where

60instance Show (Control a) where

54− show CFinish = "CFinish"

61+ show (CFinish _) = "CFinish"

55− show (CNext _) = "CNext"

62+ show (CNext _) = "CNext"

56− show CNone = "CNone"

63+ show CNone = "CNone"

58type DynaNext = WindowSize -> IO Next

65type DynaNext = WindowSize -> IO Next

@@ -102,8 +109,8 @@ data Sequence = SFinish Trailers

102-- | A message from the sender to a stream's dedicated waiter thread.

109-- | A message from the sender to a stream's dedicated waiter thread.

103data Sync = SyncNone

110data Sync = SyncNone

104 -- ^ Nothing interesting has happened. Go back to sleep.

111 -- ^ Nothing interesting has happened. Go back to sleep.

105− | SyncFinish Trailers

112+ | SyncFinish

106− -- ^ The stream has ended; enqueue 'OTrailers' and quit.

113+ -- ^ The stream has ended.

107 | SyncNext Output

114 | SyncNext Output

108 -- ^ The stream's queue has been drained; wait for more to be

115 -- ^ The stream's queue has been drained; wait for more to be

109 -- available and re-enqueue the given 'Output'.

116 -- available and re-enqueue the given 'Output'.

warp/Network/Wai/Handler/Warp/HTTP2/Worker.hs+6−6

@@ -217,14 +217,14 @@ waiter tvar sq strm outQ = do

217 mx <- atomically $ do

218 mout <- readTVar tvar

219 case mout of

220− SyncNone -> retry

220+ SyncNone -> retry

221− SyncNext out -> do

221+ SyncNext out -> do

222 writeTVar tvar SyncNone

223− return $ Right out

223+ return $ Just out

224− SyncFinish trailers -> return $ Left trailers

224+ SyncFinish -> return Nothing

225 case mx of

226− Left trailers -> enqueueWhenWindowIsOpen outQ (OTrailers strm trailers)

226+ Nothing -> return ()

227− Right out -> do

227+ Just out -> do

228 -- ensuring that the streaming queue is not empty.

229 atomically $ do

230 isEmpty <- isEmptyTBQueue sq