@@ -0,0 +1,536 @@
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