@@ -0,0 +1,387 @@
1+{-# LANGUAGE OverloadedStrings #-}
2+{-# LANGUAGE BangPatterns, CPP #-}
3+{-# LANGUAGE RecordWildCards, NamedFieldPuns #-}
4+
5+module Network.Wai.Handler.Warp.HTTP2.Sender (frameSender) where
6+
7+#if __GLASGOW_HASKELL__ < 709
8+import Control.Applicative
9+#endif
10+import Control.Concurrent (putMVar, forkIO)
11+import Control.Concurrent.STM
12+import qualified Control.Exception as E
13+import Control.Monad (void, unless)
14+import qualified Data.ByteString as BS
15+import qualified Data.ByteString.Builder.Extra as B
16+import Foreign.Ptr
17+import Network.HTTP2
18+import Network.HTTP2.Priority
19+import Network.Wai
20+import Network.Wai.Handler.Warp.Buffer
21+import Network.Wai.Handler.Warp.HTTP2.EncodeFrame
22+import Network.Wai.Handler.Warp.HTTP2.HPACK
23+import Network.Wai.Handler.Warp.HTTP2.Types
24+import Network.Wai.Handler.Warp.IORef
25+import qualified Network.Wai.Handler.Warp.Settings as S
26+import Network.Wai.Handler.Warp.Types
27+import Network.Wai.Internal (Response(..))
28+import qualified System.PosixCompat.Files as P
29+
30+#ifdef WINDOWS
31+import qualified System.IO as IO
32+#else
33+import Network.Wai.Handler.Warp.FdCache
34+import Network.Wai.Handler.Warp.SendFile (positionRead)
35+import System.Posix.Types
36+#endif
37+
38+----------------------------------------------------------------
39+
40+unlessClosed :: Context -> Connection -> Stream -> IO () -> IO ()
41+unlessClosed ctx
42+ Connection{connSendAll}
43+ strm@Stream{streamState,streamNumber}
44+ body = E.handle resetStream $ do
45+ state <- readIORef streamState
46+ unless (isClosed state) body
47+ where
48+ resetStream e = do
49+ closed ctx strm (ResetByMe e)
50+ let rst = resetFrame InternalError streamNumber
51+ connSendAll rst
52+
53+checkWindowSize :: TVar WindowSize -> TVar WindowSize -> PriorityTree Output -> Output -> Priority -> (WindowSize -> IO ()) -> IO ()
54+checkWindowSize connWindow strmWindow outQ out pri body = do
55+ cw <- atomically $ do
56+ w <- readTVar connWindow
57+ check (w > 0)
58+ return w
59+ sw <- atomically $ readTVar strmWindow
60+ if sw == 0 then
61+ void $ forkIO $ do
62+ -- checkme: if connection is closed? GC throws dead lock error?
63+ atomically $ do
64+ x <- readTVar strmWindow
65+ check (x > 0)
66+ enqueue outQ out pri
67+ else
68+ body (min cw sw)
69+
70+frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()
71+frameSender ctx@Context{outputQ,connectionWindow,wait}
72+ conn@Connection{connWriteBuffer,connBufferSize,connSendAll}
73+ ii settings = do
74+ connSendAll initialFrame
75+ loop `E.finally` putMVar wait ()
76+ where
77+ initialSettings = [(SettingsMaxConcurrentStreams,recommendedConcurrency)]
78+ initialFrame = settingsFrame id initialSettings
79+ bufHeaderPayload = connWriteBuffer `plusPtr` frameHeaderLength
80+ headerPayloadLim = connBufferSize - frameHeaderLength
81+
82+ loop = dequeue outputQ >>= \(out, pri) -> switch out pri
83+
84+ switch OFinish _ = return ()
85+ switch (OGoaway frame) _ = connSendAll frame
86+ switch (OFrame frame) _ = do
87+ connSendAll frame
88+ loop
89+ switch out@(OResponse strm rsp aux) pri = unlessClosed ctx conn strm $ do
90+ checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
91+ -- Header frame and Continuation frame
92+ let sid = streamNumber strm
93+ endOfStream = case aux of
94+ Persist{} -> False
95+ Oneshot hb -> not hb
96+ len <- headerContinue sid rsp endOfStream
97+ let total = len + frameHeaderLength
98+ case aux of
99+ Oneshot True -> do -- hasBody
100+ -- Data frame payload
101+ off <- sendHeadersIfNecessary total
102+ Next datPayloadLen mnext <- fillResponseBodyGetNext conn ii off lim rsp
103+ fillDataHeaderSend strm total datPayloadLen mnext pri
104+ Oneshot False -> do
105+ -- "closed" must be before "connSendAll". If not,
106+ -- the context would be switched to the receiver,
107+ -- resulting the inconsistency of concurrency.
108+ closed ctx strm Finished
109+ bufferIO connWriteBuffer total connSendAll
110+ Persist sq tvar -> do
111+ off <- sendHeadersIfNecessary total
112+ Next datPayloadLen mnext <- fillStreamBodyGetNext conn off lim sq tvar
113+ fillDataHeaderSend strm total datPayloadLen mnext pri
114+ loop
115+ switch out@(ONext strm curr) pri = unlessClosed ctx conn strm $ do
116+ checkWindowSize connectionWindow (streamWindow strm) outputQ out pri $ \lim -> do
117+ -- Data frame payload
118+ Next datPayloadLen mnext <- curr lim
119+ fillDataHeaderSend strm 0 datPayloadLen mnext pri
120+ loop
121+
122+ headerContinue sid rsp endOfStream = do
123+ builder <- hpackEncodeHeader ctx ii settings rsp
124+ (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim
125+ let flag0 = case signal of
126+ B.Done -> setEndHeader defaultFlags
127+ _ -> defaultFlags
128+ flag = if endOfStream then setEndStream flag0 else flag0
129+ fillFrameHeader FrameHeaders len sid flag connWriteBuffer
130+ continue sid len signal
131+
132+ continue _ len B.Done = return len
133+ continue sid len (B.More _ writer) = do
134+ let total = len + frameHeaderLength
135+ bufferIO connWriteBuffer total connSendAll
136+ (len', signal') <- writer bufHeaderPayload headerPayloadLim
137+ let flag = case signal' of
138+ B.Done -> setEndHeader defaultFlags
139+ _ -> defaultFlags
140+ fillFrameHeader FrameContinuation len' sid flag connWriteBuffer
141+ continue sid len' signal'
142+ continue sid len (B.Chunk bs writer) = do
143+ let total = len + frameHeaderLength
144+ bufferIO connWriteBuffer total connSendAll
145+ let (bs1,bs2) = BS.splitAt headerPayloadLim bs
146+ len' = BS.length bs1
147+ void $ copy bufHeaderPayload bs1
148+ fillFrameHeader FrameContinuation len' sid defaultFlags connWriteBuffer
149+ if bs2 == "" then
150+ continue sid len' (B.More 0 writer)
151+ else
152+ continue sid len' (B.Chunk bs2 writer)
153+
154+ sendHeadersIfNecessary total = do
155+ let datPayloadOff = total + frameHeaderLength
156+ if datPayloadOff < connBufferSize then
157+ return datPayloadOff
158+ else do
159+ bufferIO connWriteBuffer total connSendAll
160+ return frameHeaderLength
161+
162+ fillDataHeaderSend strm otherLen datPayloadLen mnext pri = do
163+ -- Data frame header
164+ let sid = streamNumber strm
165+ buf = connWriteBuffer `plusPtr` otherLen
166+ total = otherLen + frameHeaderLength + datPayloadLen
167+ flag = case mnext of
168+ CFinish -> setEndStream defaultFlags
169+ _ -> defaultFlags
170+ fillFrameHeader FrameData datPayloadLen sid flag buf
171+ -- "closed" must be before "connSendAll". If not,
172+ -- the context would be switched to the receiver,
173+ -- resulting the inconsistency of concurrency.
174+ case mnext of
175+ CFinish -> closed ctx strm Finished
176+ _ -> return ()
177+ bufferIO connWriteBuffer total connSendAll
178+ atomically $ do
179+ modifyTVar' connectionWindow (subtract datPayloadLen)
180+ modifyTVar' (streamWindow strm) (subtract datPayloadLen)
181+ case mnext of
182+ CFinish -> return ()
183+ CNext next -> enqueue outputQ (ONext strm next) pri
184+ CNone -> return ()
185+
186+ fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf
187+ where
188+ hinfo = FrameHeader len flag sid
189+
190+----------------------------------------------------------------
191+
192+{-
193+ResponseFile Status ResponseHeaders FilePath (Maybe FilePart)
194+ResponseBuilder Status ResponseHeaders Builder
195+ResponseStream Status ResponseHeaders StreamingBody
196+ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response
197+-}
198+
199+fillResponseBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Response -> IO Next
200+fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
201+ _ off lim (ResponseBuilder _ _ bb) = do
202+ let datBuf = connWriteBuffer `plusPtr` off
203+ room = min (connBufferSize - off) lim
204+ (len, signal) <- B.runBuilder bb datBuf room
205+ return $ nextForBuilder connWriteBuffer connBufferSize len signal
206+
207+#ifdef WINDOWS
208+fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
209+ _ off lim (ResponseFile _ _ path mpart) = do
210+ let datBuf = connWriteBuffer `plusPtr` off
211+ room = min (connBufferSize - off) lim
212+ (start, bytes) <- fileStartEnd path mpart
213+ -- fixme: how to close Handle? GC does it at this moment.
214+ h <- IO.openBinaryFile path IO.ReadMode
215+ IO.hSeek h IO.AbsoluteSeek start
216+ len <- IO.hGetBufSome h datBuf (mini room bytes)
217+ let bytes' = bytes - fromIntegral len
218+ return $ nextForFile len connWriteBuffer connBufferSize h bytes' (return ())
219+#else
220+fillResponseBodyGetNext Connection{connWriteBuffer,connBufferSize}
221+ ii off lim (ResponseFile _ _ path mpart) = do
222+ let Just fdcache = fdCacher ii
223+ (fd, refresh) <- getFd fdcache path
224+ let datBuf = connWriteBuffer `plusPtr` off
225+ room = min (connBufferSize - off) lim
226+ (start, bytes) <- fileStartEnd path mpart
227+ len <- positionRead fd datBuf (mini room bytes) start
228+ refresh
229+ let len' = fromIntegral len
230+ return $ nextForFile len connWriteBuffer connBufferSize fd (start + len') (bytes - len') refresh
231+#endif
232+
233+fillResponseBodyGetNext _ _ _ _ _ = error "fillResponseBodyGetNext"
234+
235+fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)
236+fileStartEnd path Nothing = do
237+ end <- fromIntegral . P.fileSize <$> P.getFileStatus path
238+ return (0, end)
239+fileStartEnd _ (Just part) =
240+ return (filePartOffset part, filePartByteCount part)
241+
242+----------------------------------------------------------------
243+
244+fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> IO Next
245+fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}
246+ off lim sq tvar = do
247+ let datBuf = connWriteBuffer `plusPtr` off
248+ room = min (connBufferSize - off) lim
249+ (leftover, cont, len) <- runStreamBuilder datBuf room sq
250+ nextForStream connWriteBuffer connBufferSize sq tvar leftover cont len
251+
252+----------------------------------------------------------------
253+
254+fillBufBuilder :: Buffer -> BufSize -> Leftover -> DynaNext
255+fillBufBuilder buf0 siz0 leftover lim = do
256+ let payloadBuf = buf0 `plusPtr` frameHeaderLength
257+ room = min (siz0 - frameHeaderLength) lim
258+ case leftover of
259+ LZero -> error "fillBufBuilder: LZero"
260+ LOne writer -> do
261+ (len, signal) <- writer payloadBuf room
262+ getNext len signal
263+ LTwo bs writer
264+ | BS.length bs <= room -> do
265+ buf1 <- copy payloadBuf bs
266+ let len1 = BS.length bs
267+ (len2, signal) <- writer buf1 (room - len1)
268+ getNext (len1 + len2) signal
269+ | otherwise -> do
270+ let (bs1,bs2) = BS.splitAt room bs
271+ void $ copy payloadBuf bs1
272+ getNext room (B.Chunk bs2 writer)
273+ where
274+ getNext l s = return $ nextForBuilder buf0 siz0 l s
275+
276+nextForBuilder :: Buffer -> BufSize -> BytesFilled -> B.Next -> Next
277+nextForBuilder _ _ len B.Done
278+ = Next len CFinish
279+nextForBuilder buf siz len (B.More _ writer)
280+ = Next len (CNext (fillBufBuilder buf siz (LOne writer)))
281+nextForBuilder buf siz len (B.Chunk bs writer)
282+ = Next len (CNext (fillBufBuilder buf siz (LTwo bs writer)))
283+
284+----------------------------------------------------------------
285+
286+runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence
287+ -> IO (Leftover, Bool, BytesFilled)
288+runStreamBuilder buf0 room0 sq = loop buf0 room0 0
289+ where
290+ loop !buf !room !total = do
291+ mbuilder <- atomically $ tryReadTBQueue sq
292+ case mbuilder of
293+ Nothing -> return (LZero, True, total)
294+ Just (SBuilder builder) -> do
295+ (len, signal) <- B.runBuilder builder buf room
296+ let !total' = total + len
297+ case signal of
298+ B.Done -> loop (buf `plusPtr` len) (room - len) total'
299+ B.More _ writer -> return (LOne writer, True, total')
300+ B.Chunk bs writer -> return (LTwo bs writer, True, total')
301+ Just SFlush -> return (LZero, True, total)
302+ Just SFinish -> return (LZero, False, total)
303+
304+fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> DynaNext
305+fillBufStream buf0 siz0 leftover0 sq tvar lim0 = do
306+ let payloadBuf = buf0 `plusPtr` frameHeaderLength
307+ room0 = min (siz0 - frameHeaderLength) lim0
308+ case leftover0 of
309+ LZero -> do
310+ (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq
311+ getNext leftover cont len
312+ LOne writer -> write writer payloadBuf room0 0
313+ LTwo bs writer
314+ | BS.length bs <= room0 -> do
315+ buf1 <- copy payloadBuf bs
316+ let len = BS.length bs
317+ write writer buf1 (room0 - len) len
318+ | otherwise -> do
319+ let (bs1,bs2) = BS.splitAt room0 bs
320+ void $ copy payloadBuf bs1
321+ getNext (LTwo bs2 writer) True room0
322+ where
323+ getNext = nextForStream buf0 siz0 sq tvar
324+ write writer1 buf room sofar = do
325+ (len, signal) <- writer1 buf room
326+ case signal of
327+ B.Done -> do
328+ (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq
329+ let !total = sofar + len + extra
330+ getNext leftover cont total
331+ B.More _ writer -> do
332+ let !total = sofar + len
333+ getNext (LOne writer) True total
334+ B.Chunk bs writer -> do
335+ let !total = sofar + len
336+ getNext (LTwo bs writer) True total
337+
338+nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync
339+ -> Leftover -> Bool -> BytesFilled
340+ -> IO Next
341+nextForStream _ _ _ tvar _ False len = do
342+ atomically $ writeTVar tvar SyncFinish
343+ return $ Next len CFinish
344+nextForStream buf siz sq tvar LZero True len = do
345+ atomically $ writeTVar tvar $ SyncNext (fillBufStream buf siz LZero sq tvar)
346+ return $ Next len CNone
347+nextForStream buf siz sq tvar leftover True len =
348+ return $ Next len (CNext (fillBufStream buf siz leftover sq tvar))
349+
350+----------------------------------------------------------------
351+
352+#ifdef WINDOWS
353+fillBufFile :: Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> DynaNext
354+fillBufFile buf siz h bytes refresh lim = do
355+ let payloadBuf = buf `plusPtr` frameHeaderLength
356+ room = min (siz - frameHeaderLength) lim
357+ len <- IO.hGetBufSome h payloadBuf room
358+ refresh
359+ let bytes' = bytes - fromIntegral len
360+ return $ nextForFile len buf siz h bytes' refresh
361+
362+nextForFile :: BytesFilled -> Buffer -> BufSize -> IO.Handle -> Integer -> IO () -> Next
363+nextForFile 0 _ _ _ _ _ = Next 0 CFinish
364+nextForFile len _ _ _ 0 _ = Next len CFinish
365+nextForFile len buf siz h bytes refresh =
366+ Next len (CNext (fillBufFile buf siz h bytes refresh))
367+#else
368+fillBufFile :: Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> DynaNext
369+fillBufFile buf siz fd start bytes refresh lim = do
370+ let payloadBuf = buf `plusPtr` frameHeaderLength
371+ room = min (siz - frameHeaderLength) lim
372+ len <- positionRead fd payloadBuf (mini room bytes) start
373+ let len' = fromIntegral len
374+ refresh
375+ return $ nextForFile len buf siz fd (start + len') (bytes - len') refresh
376+
377+nextForFile :: BytesFilled -> Buffer -> BufSize -> Fd -> Integer -> Integer -> IO () -> Next
378+nextForFile 0 _ _ _ _ _ _ = Next 0 CFinish
379+nextForFile len _ _ _ _ 0 _ = Next len CFinish
380+nextForFile len buf siz fd start bytes refresh =
381+ Next len (CNext (fillBufFile buf siz fd start bytes refresh))
382+#endif
383+
384+mini :: Int -> Integer -> Int
385+mini i n
386+ | fromIntegral i < n = i
387+ | otherwise = fromIntegral n