@@ -1,536 +0,0 @@
1−{-# LANGUAGE OverloadedStrings #-}
2−{-# LANGUAGE BangPatterns, CPP #-}
3−{-# LANGUAGE NamedFieldPuns #-}
5−module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
7−import Control.Concurrent.STM
8−import qualified Control.Exception as E
9−import qualified Data.ByteString as BS
10−import Data.ByteString.Builder (Builder)
11−import qualified Data.ByteString.Builder.Extra as B
13−import Foreign.Ptr (Ptr, plusPtr)
14−import Foreign.Storable (poke)
15−import Network.HPACK (setLimitForEncoding, toHeaderTable)
17−import Network.HTTP2.Priority (isEmptySTM, dequeueSTM, Precedence)
21−import qualified System.IO as IO
23−import Network.Wai.Handler.Warp.FdCache
24−import Network.Wai.Handler.Warp.SendFile (positionRead)
25−import qualified System.TimeManager as T
28−import Network.Wai.Handler.Warp.Buffer
29−import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
30−import Network.Wai.Handler.Warp.HTTP2.HPACK
31−import Network.Wai.Handler.Warp.HTTP2.Manager (Manager)
32−import Network.Wai.Handler.Warp.HTTP2.Types
33−import Network.Wai.Handler.Warp.Imports hiding (readInt)
34−import qualified Network.Wai.Handler.Warp.Settings as S
35−import Network.Wai.Handler.Warp.Types
37−----------------------------------------------------------------
40− | LOne B.BufferWriter
41− | LTwo ByteString B.BufferWriter
43−----------------------------------------------------------------
45−{-# INLINE getStreamWindowSize #-}
46−getStreamWindowSize :: Stream -> IO WindowSize
47−getStreamWindowSize Stream{streamWindow} = atomically $ readTVar streamWindow
49−{-# INLINE waitStreamWindowSize #-}
50−waitStreamWindowSize :: Stream -> IO ()
51−waitStreamWindowSize Stream{streamWindow} = atomically $ do
52− w <- readTVar streamWindow
55−{-# INLINE waitStreaming #-}
56−waitStreaming :: TBQueue a -> IO ()
57−waitStreaming tbq = atomically $ do
58− isEmpty <- isEmptyTBQueue tbq
61−data Switch = C Control
62− | O (StreamId,Precedence,Output)
65−frameSender :: Context -> Connection -> S.Settings -> Manager -> IO ()
66−frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}
67− conn@Connection{connWriteBuffer,connBufferSize,connSendAll}
68− settings mgr = loop 0 `E.catch` ignore
71− isEmpty <- isEmptyTQueue controlQ
73− w <- readTVar connectionWindow
75− emp <- isEmptySTM outputQ
77− if off /= 0 then return Flush else retry
79− O <$> dequeueSTM outputQ
81− C <$> readTQueue controlQ
84− x <- atomically $ dequeue off
87− when (off /= 0) $ flushN off
88− off' <- control ctl off
89− when (off' >= 0) $ loop off'
91− let strm = outputStream out
92− writeIORef (streamPrecedence strm) pre
93− off' <- outputOrEnqueueAgain out off
96− _ | off' > 15872 -> flushN off' >> loop 0 -- fixme: hard-coding
97− | otherwise -> loop off'
98− Flush -> flushN off >> loop 0
100− control CFinish _ = return (-1)
101− control (CGoaway frame) _ = connSendAll frame >> return (-1)
102− control (CFrame frame) _ = connSendAll frame >> return 0
103− control (CSettings frame alist) _ = do
107− control (CSettings0 frame1 frame2 alist) off = do -- off == 0, just in case
108− let !buf = connWriteBuffer `plusPtr` off
109− !off' = off + BS.length frame1 + BS.length frame2
110− buf' <- copy buf frame1
111− void $ copy buf' frame2
115− {-# INLINE setLimit #-}
116− setLimit alist = case lookup SettingsHeaderTableSize alist of
117− Nothing -> return ()
118− Just siz -> setLimitForEncoding siz encodeDynamicTable
120− output out@(Output strm _ _ tell getH2D (ONext curr)) off0 lim = do
121− -- Data frame payload
122− let !buf = connWriteBuffer `plusPtr` off0
123− !siz = connBufferSize - off0
124− Next datPayloadLen mnext <- curr buf siz lim
125− off <- fillDataHeader strm off0 datPayloadLen mnext tell getH2D
126− maybeEnqueueNext out mnext
129− output out@(Output strm rspn ii tell getH2D ORspn) off0 lim = do
130− -- Header frame and Continuation frame
131− let !sid = streamNumber strm
132− !endOfStream = case rspn of
133− RspnNobody _ _ -> True
135− ths <- fixHeaders ctx rspn ii settings
136− kvlen <- headerContinue sid ths endOfStream off0
137− off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + kvlen
139− RspnNobody _ _ -> do
140− halfClosedLocal ctx strm Finished
142− RspnFile _ _ path mpart -> do
143− -- Data frame payload
144− let payloadOff = off + frameHeaderLength
145− Next datPayloadLen mnext <-
146− fillFileBodyGetNext conn ii payloadOff lim path mpart
147− off' <- fillDataHeader strm off datPayloadLen mnext tell getH2D
148− maybeEnqueueNext out mnext
150− RspnBuilder _ _ builder -> do
151− -- Data frame payload
152− let payloadOff = off + frameHeaderLength
153− Next datPayloadLen mnext <-
154− fillBuilderBodyGetNext conn ii payloadOff lim builder
155− off' <- fillDataHeader strm off datPayloadLen mnext tell getH2D
156− maybeEnqueueNext out mnext
158− RspnStreaming _ _ tbq -> do
159− let payloadOff = off + frameHeaderLength
160− Next datPayloadLen mnext <-
161− fillStreamBodyGetNext conn payloadOff lim tbq strm
162− off' <- fillDataHeader strm off datPayloadLen mnext tell getH2D
163− maybeEnqueueNext out mnext
166− output out@(Output strm _ _ _ _ (OPush ths pid)) off0 lim = do
167− -- Creating a push promise header
168− -- Frame id should be associated stream id from the client.
169− let !sid = streamNumber strm
170− len <- pushPromise pid sid ths off0
171− off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + len
172− output out{ outputType = ORspn } off lim
174− output _ _ _ = undefined -- never reach
176− outputOrEnqueueAgain out off = E.handle resetStream $ do
177− state <- readStreamState strm
178− if isHalfClosedLocal state then
181− Output _ _ _ wait _ OWait -> do
182− -- Checking if all push are done.
184− outputHook = return ()
185− , outputType = ORspn
187− forkAndEnqueueWhenReady wait outputQ out' mgr
190− Just tbq -> checkStreaming tbq
191− _ -> checkStreamWindowSize
193− strm = outputStream out
194− mtbq = outputMaybeTBQueue out
195− checkStreaming tbq = do
196− isEmpty <- atomically $ isEmptyTBQueue tbq
198− forkAndEnqueueWhenReady (waitStreaming tbq) outputQ out mgr
201− checkStreamWindowSize
202− checkStreamWindowSize = do
203− sws <- getStreamWindowSize strm
205− forkAndEnqueueWhenReady (waitStreamWindowSize strm) outputQ out mgr
208− cws <- atomically $ readTVar connectionWindow -- not 0
209− let !lim = min cws sws
212− closed ctx strm (ResetByMe e)
213− let !rst = resetFrame InternalError $ streamNumber strm
214− enqueueControl controlQ $ CFrame rst
217− {-# INLINE flushN #-}
218− -- Flush the connection buffer to the socket, where the first 'n' bytes of
219− -- the buffer are filled.
220− flushN :: Int -> IO ()
221− flushN n = bufferIO connWriteBuffer n connSendAll
223− headerContinue sid ths endOfStream off = do
224− let !offkv = off + frameHeaderLength
225− let !bufkv = connWriteBuffer `plusPtr` offkv
226− !limkv = connBufferSize - offkv
227− (hs,kvlen) <- hpackEncodeHeader ctx bufkv limkv ths
228− let flag0 = case hs of
229− [] -> setEndHeader defaultFlags
231− flag = if endOfStream then setEndStream flag0 else flag0
232− let buf = connWriteBuffer `plusPtr` off
233− fillFrameHeader FrameHeaders kvlen sid flag buf
234− continue sid kvlen hs
236− !bufHeaderPayload = connWriteBuffer `plusPtr` frameHeaderLength
237− !headerPayloadLim = connBufferSize - frameHeaderLength
239− continue _ kvlen [] = return kvlen
240− continue sid kvlen ths = do
241− flushN $ kvlen + frameHeaderLength
243− (ths', kvlen') <- hpackEncodeHeaderLoop ctx bufHeaderPayload headerPayloadLim ths
244− when (ths == ths') $ E.throwIO $ ConnectionError CompressionError "cannot compress the header"
245− let flag = case ths' of
246− [] -> setEndHeader defaultFlags
248− fillFrameHeader FrameContinuation kvlen' sid flag connWriteBuffer
249− continue sid kvlen' ths'
251− {-# INLINE maybeEnqueueNext #-}
252− -- Re-enqueue the stream in the output queue.
253− maybeEnqueueNext :: Output -> Maybe DynaNext -> IO ()
254− maybeEnqueueNext _ Nothing = return ()
255− maybeEnqueueNext out (Just next) = enqueueOutput outputQ out'
257− !out' = out { outputType = ONext next }
259− {-# INLINE sendHeadersIfNecessary #-}
260− -- Send headers if there is not room for a 1-byte data frame, and return
261− -- the offset of the next frame's first header byte.
262− sendHeadersIfNecessary off
263− -- True if the connection buffer has room for a 1-byte data frame.
264− | off + frameHeaderLength < connBufferSize = return off
269− fillDataHeader strm@Stream{streamWindow,streamNumber}
270− off datPayloadLen mnext tell getH2D = do
271− -- Data frame header
273− let (!trailers,!noTrailers) = case http2dataTrailers <$> mh2d of
274− Nothing -> ([], True)
275− Just ts -> (ts, null ts)
276− !buf = connWriteBuffer `plusPtr` off
277− !off' = off + frameHeaderLength + datPayloadLen
278− !noMoreBody = isNothing mnext
279− flag | noMoreBody && noTrailers = setEndStream defaultFlags
280− | otherwise = defaultFlags
281− fillFrameHeader FrameData datPayloadLen streamNumber flag buf
282− off'' <- handleEndOfBody noMoreBody off' noTrailers trailers
283− atomically $ modifyTVar' connectionWindow (subtract datPayloadLen)
284− atomically $ modifyTVar' streamWindow (subtract datPayloadLen)
287− handleTrailers True off0 _ = return off0
288− handleTrailers _ off0 trailers = do
289− (ths,_) <- toHeaderTable trailers
290− kvlen <- headerContinue streamNumber ths True off0
291− sendHeadersIfNecessary $ off0 + frameHeaderLength + kvlen
292− handleEndOfBody True off0 noTrailers trailers = do
293− off1 <- handleTrailers noTrailers off0 trailers
295− halfClosedLocal ctx strm Finished
297− handleEndOfBody False off0 _ _ = return off0
300− pushPromise pid sid ths off = do
301− let !offsid = off + frameHeaderLength
302− !bufsid = connWriteBuffer `plusPtr` offsid
303− poke32 bufsid $ fromIntegral sid
304− let !offkv = offsid + 4
305− !bufkv = connWriteBuffer `plusPtr` offkv
306− !limkv = connBufferSize - offkv
307− (_,kvlen) <- hpackEncodeHeader ctx bufkv limkv ths
308− let !flag = setEndHeader defaultFlags -- No EndStream flag
309− !buf = connWriteBuffer `plusPtr` off
311− fillFrameHeader FramePushPromise len pid flag buf
314− {-# INLINE fillFrameHeader #-}
315− fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf
317− hinfo = FrameHeader len flag sid
319− {-# INLINE ignore #-}
320− ignore :: E.SomeException -> IO ()
321− ignore _ = return ()
323−----------------------------------------------------------------
326−ResponseFile Status ResponseHeaders FilePath (Maybe FilePart)
327−ResponseBuilder Status ResponseHeaders Builder
328−ResponseStream Status ResponseHeaders StreamingBody
329−ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response
332−fillBuilderBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Builder -> IO Next
333−fillBuilderBodyGetNext Connection{connWriteBuffer,connBufferSize}
335− let datBuf = connWriteBuffer `plusPtr` off
336− room = min (connBufferSize - off) lim
337− (len, signal) <- B.runBuilder bb datBuf room
338− return $ nextForBuilder len signal
340−fillFileBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> FilePath -> Maybe FilePart -> IO Next
342−fillFileBodyGetNext Connection{connWriteBuffer,connBufferSize}
343− _ off lim path mpart = do
344− let datBuf = connWriteBuffer `plusPtr` off
345− room = min (connBufferSize - off) lim
346− (start, bytes) <- fileStartEnd path mpart
347− -- fixme: how to close Handle? GC does it at this moment.
348− hdl <- IO.openBinaryFile path IO.ReadMode
349− IO.hSeek hdl IO.AbsoluteSeek start
350− len <- IO.hGetBufSome hdl datBuf (mini room bytes)
351− let bytes' = bytes - fromIntegral len
352− -- fixme: connWriteBuffer connBufferSize
353− return $ nextForFile len hdl bytes' (return ())
355−fillFileBodyGetNext Connection{connWriteBuffer,connBufferSize}
356− ii off lim path mpart = do
357− (mfd, refresh') <- getFd ii path
358− (fd, refresh) <- case mfd of
360− fd' <- openFile path
361− th <- T.register (timeoutManager ii) (closeFile fd')
362− return (fd', T.tickle th)
363− Just fd -> return (fd, refresh')
364− let datBuf = connWriteBuffer `plusPtr` off
365− room = min (connBufferSize - off) lim
366− (start, bytes) <- fileStartEnd path mpart
367− len <- positionRead fd datBuf (mini room bytes) start
369− let len' = fromIntegral len
370− return $ nextForFile len fd (start + len') (bytes - len') refresh
373−fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)
374−fileStartEnd _ (Just part) =
375− return (filePartOffset part, filePartByteCount part)
376−fileStartEnd _ _ = error "fileStartEnd"
378−----------------------------------------------------------------
380−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> Stream -> IO Next
381−fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
382− off lim sq strm = do
383− let datBuf = connWriteBuffer `plusPtr` off
384− room = min (connBufferSize - off) lim
385− (leftover, cont, len) <- runStreamBuilder datBuf room sq
386− return $ nextForStream sq strm leftover cont len
388−----------------------------------------------------------------
390−fillBufBuilder :: Leftover -> DynaNext
391−fillBufBuilder leftover buf0 siz0 lim = do
392− let payloadBuf = buf0 `plusPtr` frameHeaderLength
393− room = min (siz0 - frameHeaderLength) lim
395− LZero -> error "fillBufBuilder: LZero"
397− (len, signal) <- writer payloadBuf room
400− | BS.length bs <= room -> do
401− buf1 <- copy payloadBuf bs
402− let len1 = BS.length bs
403− (len2, signal) <- writer buf1 (room - len1)
404− getNext (len1 + len2) signal
406− let (bs1,bs2) = BS.splitAt room bs
407− void $ copy payloadBuf bs1
408− getNext room (B.Chunk bs2 writer)
410− getNext l s = return $ nextForBuilder l s
412−nextForBuilder :: BytesFilled -> B.Next -> Next
413−nextForBuilder len B.Done
415−nextForBuilder len (B.More _ writer)
416− = Next len $ Just (fillBufBuilder (LOne writer))
417−nextForBuilder len (B.Chunk bs writer)
418− = Next len $ Just (fillBufBuilder (LTwo bs writer))
420−----------------------------------------------------------------
422−runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence
423− -> IO (Leftover, Bool, BytesFilled)
424−runStreamBuilder buf0 room0 sq = loop buf0 room0 0
426− loop !buf !room !total = do
427− mbuilder <- atomically $ tryReadTBQueue sq
429− Nothing -> return (LZero, True, total)
430− Just (SBuilder builder) -> do
431− (len, signal) <- B.runBuilder builder buf room
432− let !total' = total + len
434− B.Done -> loop (buf `plusPtr` len) (room - len) total'
435− B.More _ writer -> return (LOne writer, True, total')
436− B.Chunk bs writer -> return (LTwo bs writer, True, total')
437− Just SFlush -> return (LZero, True, total)
438− Just SFinish -> return (LZero, False, total)
440−fillBufStream :: Leftover -> TBQueue Sequence -> Stream -> DynaNext
441−fillBufStream leftover0 sq strm buf0 siz0 lim0 = do
442− let payloadBuf = buf0 `plusPtr` frameHeaderLength
443− room0 = min (siz0 - frameHeaderLength) lim0
446− (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq
447− getNext leftover cont len
448− LOne writer -> write writer payloadBuf room0 0
450− | BS.length bs <= room0 -> do
451− buf1 <- copy payloadBuf bs
452− let len = BS.length bs
453− write writer buf1 (room0 - len) len
455− let (bs1,bs2) = BS.splitAt room0 bs
456− void $ copy payloadBuf bs1
457− getNext (LTwo bs2 writer) True room0
459− getNext l b r = return $ nextForStream sq strm l b r
460− write writer1 buf room sofar = do
461− (len, signal) <- writer1 buf room
464− (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq
465− let !total = sofar + len + extra
466− getNext leftover cont total
467− B.More _ writer -> do
468− let !total = sofar + len
469− getNext (LOne writer) True total
470− B.Chunk bs writer -> do
471− let !total = sofar + len
472− getNext (LTwo bs writer) True total
474−nextForStream :: TBQueue Sequence -> Stream
475− -> Leftover -> Bool -> BytesFilled
477−nextForStream _ _ _ False len = Next len Nothing
478−nextForStream sq strm leftOrZero True len =
479− Next len $ Just (fillBufStream leftOrZero sq strm)
481−----------------------------------------------------------------
484−fillBufFile :: IO.Handle -> Integer -> IO () -> DynaNext
485−fillBufFile h bytes refresh buf siz lim = do
486− let payloadBuf = buf `plusPtr` frameHeaderLength
487− room = min (siz - frameHeaderLength) lim
488− len <- IO.hGetBufSome h payloadBuf room
490− let bytes' = bytes - fromIntegral len
491− return $ nextForFile len h bytes' refresh
493−nextForFile :: BytesFilled -> IO.Handle -> Integer -> IO () -> Next
494−nextForFile 0 _ _ _ = Next 0 Nothing
495−nextForFile len _ 0 _ = Next len Nothing
496−nextForFile len h bytes refresh =
497− Next len $ Just (fillBufFile h bytes refresh)
499−fillBufFile :: Fd -> Integer -> Integer -> IO () -> DynaNext
500−fillBufFile fd start bytes refresh buf siz lim = do
501− let payloadBuf = buf `plusPtr` frameHeaderLength
502− room = min (siz - frameHeaderLength) lim
503− len <- positionRead fd payloadBuf (mini room bytes) start
504− let len' = fromIntegral len
506− return $ nextForFile len fd (start + len') (bytes - len') refresh
508−nextForFile :: BytesFilled -> Fd -> Integer -> Integer -> IO () -> Next
509−nextForFile 0 _ _ _ _ = Next 0 Nothing
510−nextForFile len _ _ 0 _ = Next len Nothing
511−nextForFile len fd start bytes refresh =
512− Next len $ Just (fillBufFile fd start bytes refresh)
516−mini :: Int -> Integer -> Int
518− | fromIntegral i < n = i
519− | otherwise = fromIntegral n
522−----------------------------------------------------------------
524−poke32 :: Ptr Word8 -> Word32 -> IO ()
531− w0 = fromIntegral ((i `shiftR` 24) .&. 0xff)
532− w1 = fromIntegral ((i `shiftR` 16) .&. 0xff)
533− w2 = fromIntegral ((i `shiftR` 8) .&. 0xff)
534− w3 = fromIntegral (i .&. 0xff)
535− poke8 :: Ptr Word8 -> Int -> Word8 -> IO ()
536− poke8 ptr0 n w = poke (ptr0 `plusPtr` n) w