8a9b8f9 · typeable/wai

git.iko.soy/typeable/wai

on master help sign in

Support sending file ranges in stream bodies

8a9b8f9Andrew Pritchard11 years ago

wai/Network/Wai/HTTP2.hs+15−3

compare with master

unifiedsplit

@@ -1,11 +1,14 @@

1{-# LANGUAGE OverloadedStrings, RankNTypes #-}

2module Network.Wai.HTTP2

3+ ( Chunk(..)

3− ( HTTP2Application

4+ , HTTP2Application

4 , PushPromise(..)

5 , PushPromise(..)

5 , RespondFunc

6 , RespondFunc

6 , Responder

7 , Responder

7 , Trailers

8 , Trailers

8 , promiseHeaders

9 , promiseHeaders

10+ , sendBuilder

11+ , sendFilePart

9 ) where

12 ) where

11import Data.ByteString (ByteString)

14import Data.ByteString (ByteString)

@@ -14,7 +17,7 @@ import qualified Data.ByteString.Lazy as L

14import Data.ByteString.Builder (Builder)

17import Data.ByteString.Builder (Builder)

15import qualified Network.HTTP.Types as H

18import qualified Network.HTTP.Types as H

17−import Network.Wai.Internal (Request)

20+import Network.Wai.Internal (FilePart, Request)

19-- | Headers sent after the end of a data stream, as defined by section 4.1.2 of

22-- | Headers sent after the end of a data stream, as defined by section 4.1.2 of

20-- the HTTP\/1.1 spec (RFC 7230), and section 8.1 of the HTTP\/2 spec.

23-- the HTTP\/1.1 spec (RFC 7230), and section 8.1 of the HTTP\/2 spec.

@@ -32,7 +35,16 @@ data PushPromise = PushPromise

32-- | The HTTP\/2-aware equivalent of 'Network.Wai.Application'.

35-- | The HTTP\/2-aware equivalent of 'Network.Wai.Application'.

33type HTTP2Application = Request -> PushFunc -> Responder

36type HTTP2Application = Request -> PushFunc -> Responder

38+data Chunk = FileChunk FilePath (Maybe FilePart) | BuilderChunk Builder

39+

40+sendBuilder :: (Chunk -> IO ()) -> Builder -> IO ()

41+sendBuilder = (. BuilderChunk)

42+

43+sendFilePart :: (Chunk -> IO ()) -> FilePath -> Maybe FilePart -> IO ()

44+sendFilePart = (.: FileChunk)

45+ where f .: g = curry $ f . uncurry g

46+

35−type Body a = (Builder -> IO ()) -> IO () -> IO a

47+type Body a = (Chunk -> IO ()) -> IO () -> IO a

37type RespondFunc = forall a. H.Status -> H.ResponseHeaders -> Body a -> IO a

49type RespondFunc = forall a. H.Status -> H.ResponseHeaders -> Body a -> IO a

warp/Network/Wai/Handler/Warp/HTTP2.hs+3−2

@@ -17,7 +17,7 @@ import Network.HTTP2

17import Network.Socket (SockAddr)

19import Network.Wai (Application, responseToStream)

20−import Network.Wai.HTTP2 (HTTP2Application)

20+import Network.Wai.HTTP2 (HTTP2Application, Chunk(BuilderChunk))

21import Network.Wai.Internal (ResponseReceived(..))

22import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

23import Network.Wai.Handler.Warp.HTTP2.Manager

@@ -79,5 +79,6 @@ goaway Connection{..} etype debugmsg = connSendAll bytestream

79promoteApplication :: Application -> HTTP2Application

80promoteApplication app req _ respond = [] <$ app req respond'

81 where

82− respond' r = ResponseReceived <$ (withBody $ respond s h)

82+ respond' r = ResponseReceived <$ (withBody $ respond s h . convBody)

83 where (s, h, withBody) = responseToStream r

84+ convBody b = \send flush -> b (send . BuilderChunk) flush

warp/Network/Wai/Handler/Warp/HTTP2/Sender.hs+107−20

@@ -18,20 +18,34 @@ import Foreign.Ptr

18import qualified Network.HTTP.Types as H

19import Network.HTTP2

20import Network.HTTP2.Priority

21+import Network.Wai (FilePart(..))

21import Network.Wai.HTTP2 (Trailers, promiseHeaders)

22import Network.Wai.HTTP2 (Trailers, promiseHeaders)

22import Network.Wai.Handler.Warp.Buffer

23import Network.Wai.Handler.Warp.Buffer

24+import Network.Wai.Handler.Warp.FdCache (getFd)

25+import Network.Wai.Handler.Warp.SendFile (positionRead)

23import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

26import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

24import Network.Wai.Handler.Warp.HTTP2.HPACK

27import Network.Wai.Handler.Warp.HTTP2.HPACK

25import Network.Wai.Handler.Warp.HTTP2.Types

28import Network.Wai.Handler.Warp.HTTP2.Types

26import Network.Wai.Handler.Warp.IORef

29import Network.Wai.Handler.Warp.IORef

27import qualified Network.Wai.Handler.Warp.Settings as S

30import qualified Network.Wai.Handler.Warp.Settings as S

31+import qualified Network.Wai.Handler.Warp.Timeout as T

28import Network.Wai.Handler.Warp.Types

32import Network.Wai.Handler.Warp.Types

33+import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)

34+import System.Posix.Types (Fd)

35+import qualified System.PosixCompat.Files as P

30----------------------------------------------------------------

37----------------------------------------------------------------

32data Leftover = LZero

39data Leftover = LZero

33 | LOne B.BufferWriter

40 | LOne B.BufferWriter

34 | LTwo BS.ByteString B.BufferWriter

41 | LTwo BS.ByteString B.BufferWriter

42+ | LFile OpenFile Integer Integer (IO ())

43+

44+#ifdef WINDOWS

45+type OpenFile = IO.Handle

46+#else

47+type OpenFile = Fd

48+#endif

36----------------------------------------------------------------

50----------------------------------------------------------------

@@ -138,7 +152,7 @@ frameSender ctx@Context{outputQ,connectionWindow}

138 (off, needSend) <- sendHeadersIfNecessary total

152 (off, needSend) <- sendHeadersIfNecessary total

139 let payloadOff = off + frameHeaderLength

153 let payloadOff = off + frameHeaderLength

140 Next datPayloadLen mnext <-

154 Next datPayloadLen mnext <-

141− fillStreamBodyGetNext conn payloadOff lim sq tvar strm

155+ fillStreamBodyGetNext ii conn payloadOff lim sq tvar strm

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

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

143 -- empty data frame.

157 -- empty data frame.

144 if datPayloadLen > 0 then

158 if datPayloadLen > 0 then

@@ -240,19 +254,20 @@ frameSender ctx@Context{outputQ,connectionWindow}

240

254

241----------------------------------------------------------------

255----------------------------------------------------------------

242

256

243−fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next

257+fillStreamBodyGetNext :: InternalInfo -> Connection -> Int -> WindowSize

258+ -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next

244−fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}

259+fillStreamBodyGetNext ii Connection{connWriteBuffer,connBufferSize}

245 off lim sq tvar strm = do

260 off lim sq tvar strm = do

246 let datBuf = connWriteBuffer `plusPtr` off

261 let datBuf = connWriteBuffer `plusPtr` off

247 room = min (connBufferSize - off) lim

262 room = min (connBufferSize - off) lim

248− (leftover, cont, len) <- runStreamBuilder datBuf room sq

263+ (leftover, cont, len) <- runStreamBuilder ii datBuf room sq

249− nextForStream connWriteBuffer connBufferSize sq tvar strm leftover cont len

264+ nextForStream ii connWriteBuffer connBufferSize sq tvar strm leftover cont len

250

265

251----------------------------------------------------------------

266----------------------------------------------------------------

252

267

253−runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence

268+runStreamBuilder :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence

254 -> IO (Leftover, Maybe Trailers, BytesFilled)

269 -> IO (Leftover, Maybe Trailers, BytesFilled)

255−runStreamBuilder buf0 room0 sq = loop buf0 room0 0

270+runStreamBuilder ii buf0 room0 sq = loop buf0 room0 0

256 where

271 where

257 loop !buf !room !total = do

272 loop !buf !room !total = do

258 mbuilder <- atomically $ tryReadTBQueue sq

273 mbuilder <- atomically $ tryReadTBQueue sq

@@ -265,16 +280,85 @@ runStreamBuilder buf0 room0 sq = loop buf0 room0 0

265 B.Done -> loop (buf `plusPtr` len) (room - len) total'

280 B.Done -> loop (buf `plusPtr` len) (room - len) total'

266 B.More _ writer -> return (LOne writer, Nothing, total')

281 B.More _ writer -> return (LOne writer, Nothing, total')

267 B.Chunk bs writer -> return (LTwo bs writer, Nothing, total')

282 B.Chunk bs writer -> return (LTwo bs writer, Nothing, total')

283+ Just (SFile path mpart) -> do

284+ (leftover, len) <- runStreamFile ii buf room path mpart

285+ let !total' = total + len

286+ return (leftover, Nothing, total')

287+ -- TODO if file part is done, go back to loop

268 Just SFlush -> return (LZero, Nothing, total)

288 Just SFlush -> return (LZero, Nothing, total)

269 Just (SFinish trailers) -> return (LZero, Just trailers, total)

289 Just (SFinish trailers) -> return (LZero, Just trailers, total)

270

290

291+-- | Open the file, determine the range we should send, and start reading into

292+-- the send buffer.

293+runStreamFile :: InternalInfo -> Buffer -> BufSize -> FilePath -> Maybe FilePart

294+ -> IO (Leftover, BytesFilled)

295+

296+-- | Read the given (OS-specific) file representation into the buffer. On

297+-- non-Windows systems this uses pread; on Windows this ignores the position

298+-- because we use the Handle's internal read position instead (because it's not

299+-- potentially shared with other readers).

300+readOpenFile :: OpenFile -> Buffer -> BufSize -> Integer -> IO Int

301+

302+#ifdef WINDOWS

303+runStreamFile ii buf room path mpart = do

304+ (start, bytes) <- fileStartEnd path mpart

305+ -- fixme: how to close Handle? GC does it at this moment.

306+ h <- IO.openBinaryFile path IO.ReadMode

307+ IO.hSeek h IO.AbsoluteSeek start

308+ fillBufFile buf room h start bytes (return ())

309+

310+readOpenFile h buf room _ = IO.hGetBufSome h buf room

311+#else

312+runStreamFile ii buf room path mpart = do

313+ (start, bytes) <- fileStartEnd path mpart

314+ (fd, refresh) <- case fdCacher ii of

315+ Just fdcache -> getFd fdcache path

316+ Nothing -> do

317+ fd' <- openFd path ReadOnly Nothing defaultFileFlags{nonBlock=True}

318+ th <- T.register (timeoutManager ii) (closeFd fd')

319+ return (fd', T.tickle th)

320+ fillBufFile buf room fd start bytes refresh

321+

322+readOpenFile = positionRead

323+#endif

324+

325+-- | Read as much of the file as is currently available into the buffer, then

326+-- return a 'Leftover' to indicate whether this file chunk has more data to

327+-- send. If this read hit the end of the file range, return 'LZero'; otherwise

328+-- return 'LFile' so this stream will continue reading from the file the next

329+-- time it's pulled from the queue.

330+fillBufFile :: Buffer -> BufSize -> OpenFile -> Integer -> Integer -> (IO ())

331+ -> IO (Leftover, BytesFilled)

332+fillBufFile buf room f start bytes refresh = do

333+ len <- readOpenFile f buf (mini room bytes) start

334+ refresh

335+ let len' = fromIntegral len

336+ leftover = if bytes > len' then

337+ LFile f (start + len') (bytes - len') refresh

338+ else

339+ LZero

340+ return (leftover, len)

341+

342+fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)

343+fileStartEnd path Nothing = do

344+ end <- fromIntegral . P.fileSize <$> P.getFileStatus path

345+ return (0, end)

346+fileStartEnd _ (Just part) =

347+ return (filePartOffset part, filePartByteCount part)

348+

349+mini :: Int -> Integer -> Int

350+mini i n

351+ | fromIntegral i < n = i

352+ | otherwise = fromIntegral n

353+

354+fillBufStream :: InternalInfo -> Buffer -> BufSize -> Leftover

271−fillBufStream :: Buffer -> BufSize -> Leftover -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext

355+ -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext

272−fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do

356+fillBufStream ii buf0 siz0 leftover0 sq tvar strm lim0 = do

273 let payloadBuf = buf0 `plusPtr` frameHeaderLength

357 let payloadBuf = buf0 `plusPtr` frameHeaderLength

274 room0 = min (siz0 - frameHeaderLength) lim0

358 room0 = min (siz0 - frameHeaderLength) lim0

275 case leftover0 of

359 case leftover0 of

276 LZero -> do

360 LZero -> do

277− (leftover, end, len) <- runStreamBuilder payloadBuf room0 sq

361+ (leftover, end, len) <- runStreamBuilder ii payloadBuf room0 sq

278 getNext leftover end len

362 getNext leftover end len

279 LOne writer -> write writer payloadBuf room0 0

363 LOne writer -> write writer payloadBuf room0 0

280 LTwo bs writer

364 LTwo bs writer

@@ -286,13 +370,16 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do

286 let (bs1,bs2) = BS.splitAt room0 bs

370 let (bs1,bs2) = BS.splitAt room0 bs

287 void $ copy payloadBuf bs1

371 void $ copy payloadBuf bs1

288 getNext (LTwo bs2 writer) Nothing room0

372 getNext (LTwo bs2 writer) Nothing room0

373+ LFile fd start bytes refresh -> do

374+ (leftover, len) <- fillBufFile payloadBuf room0 fd start bytes refresh

375+ getNext leftover Nothing len

289 where

376 where

290− getNext = nextForStream buf0 siz0 sq tvar strm

377+ getNext = nextForStream ii buf0 siz0 sq tvar strm

291 write writer1 buf room sofar = do

378 write writer1 buf room sofar = do

292 (len, signal) <- writer1 buf room

379 (len, signal) <- writer1 buf room

293 case signal of

380 case signal of

294 B.Done -> do

381 B.Done -> do

295− (leftover, end, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq

382+ (leftover, end, extra) <- runStreamBuilder ii (buf `plusPtr` len) (room - len) sq

296 let !total = sofar + len + extra

383 let !total = sofar + len + extra

297 getNext leftover end total

384 getNext leftover end total

298 B.More _ writer -> do

385 B.More _ writer -> do

@@ -302,15 +389,15 @@ fillBufStream buf0 siz0 leftover0 sq tvar strm lim0 = do

302 let !total = sofar + len

389 let !total = sofar + len

303 getNext (LTwo bs writer) Nothing total

390 getNext (LTwo bs writer) Nothing total

304

391

305−nextForStream :: Buffer -> BufSize -> TBQueue Sequence -> TVar Sync -> Stream

392+nextForStream :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence

306− -> Leftover -> Maybe Trailers -> BytesFilled

393+ -> TVar Sync -> Stream -> Leftover -> Maybe Trailers

307− -> IO Next

394+ -> BytesFilled -> IO Next

308−nextForStream _ _ _ tvar _ _ (Just trailers) len = do

395+nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do

309 atomically $ writeTVar tvar $ SyncFinish trailers

396 atomically $ writeTVar tvar $ SyncFinish trailers

310 return $ Next len CNone

397 return $ Next len CNone

311−nextForStream buf siz sq tvar strm LZero Nothing len = do

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

312− let out = ONext strm (fillBufStream buf siz LZero sq tvar strm)

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

313 atomically $ writeTVar tvar $ SyncNext out

400 atomically $ writeTVar tvar $ SyncNext out

314 return $ Next len CNone

401 return $ Next len CNone

315−nextForStream buf siz sq tvar strm leftover Nothing len =

402+nextForStream ii buf siz sq tvar strm leftover Nothing len =

316− return $ Next len (CNext (fillBufStream buf siz leftover sq tvar strm))

403+ return $ Next len (CNext (fillBufStream ii buf siz leftover sq tvar strm))

warp/Network/Wai/Handler/Warp/HTTP2/Types.hs+2−1

@@ -17,7 +17,7 @@ import qualified Data.ByteString as BS

17import Data.IntMap.Strict (IntMap, IntMap)

18import qualified Data.IntMap.Strict as M

19import qualified Network.HTTP.Types as H

20−import Network.Wai (Request)

20+import Network.Wai (Request, FilePart)

21import Network.Wai.HTTP2 (PushPromise, Trailers)

22import Network.Wai.Handler.Warp.IORef

23import Network.Wai.Handler.Warp.Types

@@ -88,6 +88,7 @@ outputStream _ = error "outputStream"

88data Sequence = SFinish Trailers

89 | SFlush

90 | SBuilder Builder

91+ | SFile FilePath (Maybe FilePart)

92data Sync = SyncNone

93data Sync = SyncNone

93 | SyncFinish Trailers

94 | SyncFinish Trailers

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

@@ -27,7 +27,7 @@ import Network.Wai.Handler.Warp.HTTP2.EncodeFrame

27import Network.Wai.Handler.Warp.HTTP2.Manager

28import Network.Wai.Handler.Warp.HTTP2.Types

29import Network.Wai.Handler.Warp.IORef

30−import Network.Wai.HTTP2 (HTTP2Application, PushPromise, Responder, RespondFunc)

30+import Network.Wai.HTTP2 (Chunk(..), HTTP2Application, PushPromise, Responder, RespondFunc)

31import qualified Network.Wai.Handler.Warp.Settings as S

32import qualified Network.Wai.Handler.Warp.Timeout as T

@@ -68,8 +68,10 @@ runStream Context{outputQ} mkOutput tickle strm sq s h strmbdy = do

68 -- queue is not empty.

69 void $ forkIO $ waiter tvar sq strm outputQ

70 atomically $ writeTVar tvar (SyncNext out)

71− let push b = do

71+ let push chunk = do

72− atomically $ writeTBQueue sq (SBuilder b)

72+ atomically $ writeTBQueue sq $ case chunk of

73+ BuilderChunk b -> SBuilder b

74+ FileChunk path part -> SFile path part

73 tickle

75 tickle

74 flush = atomically $ writeTBQueue sq SFlush

76 flush = atomically $ writeTBQueue sq SFlush

75 strmbdy push flush

77 strmbdy push flush