warp 3.1.12 → 3.2.0
raw patch · 22 files changed
+951/−1066 lines, 22 filesdep −unordered-containersdep ~http2dep ~wai
Dependencies removed: unordered-containers
Dependency ranges changed: http2, wai
Files
- Network/Wai/Handler/Warp.hs +1/−55
- Network/Wai/Handler/Warp/FdCache.hs +14/−11
- Network/Wai/Handler/Warp/File.hs +7/−3
- Network/Wai/Handler/Warp/FileInfoCache.hs +33/−19
- Network/Wai/Handler/Warp/HTTP2.hs +13/−12
- Network/Wai/Handler/Warp/HTTP2/HPACK.hs +5/−18
- Network/Wai/Handler/Warp/HTTP2/Manager.hs +46/−35
- Network/Wai/Handler/Warp/HTTP2/Receiver.hs +64/−62
- Network/Wai/Handler/Warp/HTTP2/Request.hs +73/−41
- Network/Wai/Handler/Warp/HTTP2/Sender.hs +346/−305
- Network/Wai/Handler/Warp/HTTP2/Types.hs +112/−166
- Network/Wai/Handler/Warp/HTTP2/Worker.hs +139/−193
- Network/Wai/Handler/Warp/HashMap.hs +24/−0
- Network/Wai/Handler/Warp/Header.hs +4/−0
- Network/Wai/Handler/Warp/Internal.hs +0/−11
- Network/Wai/Handler/Warp/Request.hs +4/−2
- Network/Wai/Handler/Warp/RequestHeader.hs +1/−2
- Network/Wai/Handler/Warp/Response.hs +25/−17
- Network/Wai/Handler/Warp/Run.hs +30/−106
- Network/Wai/Handler/Warp/Types.hs +3/−0
- test/RequestSpec.hs +1/−1
- warp.cabal +6/−7
Network/Wai/Handler/Warp.hs view
@@ -44,15 +44,6 @@ , runEnv , runSettings , runSettingsSocket- -- * Run an HTTP\/2-aware server- -- | Each of these takes an HTTP\/2-aware application as well as a backup- -- 'Application' to be used for HTTP\/1.1 and HTTP\/1 connections. These- -- are only needed if your application needs access to HTTP\/2-specific- -- features such as trailers or pushed streams.- , runHTTP2- , runHTTP2Env- , runHTTP2Settings- , runHTTP2SettingsSocket -- * Settings , Settings , defaultSettings@@ -92,52 +83,13 @@ , defaultOnExceptionResponse , exceptionResponseForDebug -- * Data types- , HostPreference (..)+ , HostPreference , Port , InvalidRequest (..) -- * Utilities , pauseTimeout , FileInfo(..) , getFileInfo- -- * Internal- -- | The following APIs will be removed in Warp 3.2.0. Please use ones exported from Network.Wai.Handler.Warp.Internal.-- -- ** Low level run functions- , runSettingsConnection- , runSettingsConnectionMaker- , runSettingsConnectionMakerSecure- , Transport (..)- -- ** Connection- , Connection (..)- , socketConnection- -- ** Buffer- , Buffer- , BufSize- , bufferSize- , allocateBuffer- , freeBuffer- -- ** Sendfile- , FileId (..)- , SendFile- , sendFile- , readSendFile- -- ** Version- , warpVersion- -- ** Data types- , InternalInfo (..)- , HeaderValue- , IndexedHeader- , requestMaxIndex- -- ** File descriptor cache- , module Network.Wai.Handler.Warp.FdCache- -- ** Date- , module Network.Wai.Handler.Warp.Date- -- ** Request and response- , Source- , recvRequest- , sendResponse- -- ** Time out manager- , module Network.Wai.Handler.Warp.Timeout ) where import Control.Exception (SomeException, throwIO)@@ -148,15 +100,9 @@ import qualified Network.HTTP.Types as H import Network.Socket (SockAddr) import Network.Wai (Request, Response, vault)-import Network.Wai.Handler.Warp.Buffer-import Network.Wai.Handler.Warp.Date-import Network.Wai.Handler.Warp.FdCache import Network.Wai.Handler.Warp.FileInfoCache-import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.Request-import Network.Wai.Handler.Warp.Response import Network.Wai.Handler.Warp.Run-import Network.Wai.Handler.Warp.SendFile import Network.Wai.Handler.Warp.Settings import Network.Wai.Handler.Warp.Timeout import Network.Wai.Handler.Warp.Types
Network/Wai/Handler/Warp/FdCache.hs view
@@ -18,6 +18,7 @@ module Network.Wai.Handler.Warp.FdCache ( withFdCache , getFd+ , getFd' , MutableFdCache , Refresh ) where@@ -69,10 +70,10 @@ type FdCache = MMap Hash FdEntry -- | Mutable Fd cacher.-type MutableFdCache = Reaper FdCache (Hash, FdEntry)+newtype MutableFdCache = MutableFdCache (Reaper FdCache (Hash, FdEntry)) fdCache :: MutableFdCache -> IO FdCache-fdCache mfc = reaperRead mfc+fdCache (MutableFdCache reaper) = reaperRead reaper look :: MutableFdCache -> FilePath -> Hash -> IO (Maybe FdEntry) look mfc path key = searchWith key check <$> fdCache mfc@@ -98,7 +99,7 @@ -- The first argument is a cache duration in second. initialize :: Int -> IO (Maybe MutableFdCache) initialize 0 = return Nothing-initialize duration = Just <$> mkReaper defaultReaperSettings+initialize duration = Just . MutableFdCache <$> mkReaper defaultReaperSettings { reaperAction = clean , reaperDelay = duration , reaperCons = uncurry insert@@ -131,24 +132,26 @@ terminate :: Maybe MutableFdCache -> IO () terminate Nothing = return ()-terminate (Just mfc) = do- !t <- reaperStop mfc+terminate (Just (MutableFdCache reaper)) = do+ !t <- reaperStop reaper mapM_ closeIt $ toList t where closeIt (_, FdEntry _ fd _) = closeFd fd ---------------------------------------------------------------- --- | Getting 'Fd' and 'Refresh' from the mutable Fd cacher. getFd :: MutableFdCache -> FilePath -> IO (Fd, Refresh)-getFd mfc path = look mfc path key >>= getFd'+getFd mfc path = getFd' mfc (hash path) path++-- | Getting 'Fd' and 'Refresh' from the mutable Fd cacher.+getFd' :: MutableFdCache -> Hash -> FilePath -> IO (Fd, Refresh)+getFd' mfc@(MutableFdCache reaper) h path = look mfc path h >>= get where- key = hash path- getFd' Nothing = do+ get Nothing = do ent@(FdEntry _ fd mst) <- newFdEntry path- reaperAdd mfc (key, ent)+ reaperAdd reaper (h, ent) return (fd, refresh mst)- getFd' (Just (FdEntry _ fd mst)) = do+ get (Just (FdEntry _ fd mst)) = do refresh mst return (fd, refresh mst) #endif
Network/Wai/Handler/Warp/File.hs view
@@ -6,6 +6,7 @@ RspFileInfo(..) , conditionalRequest , addContentHeadersForFilePart+ , parseByteRanges ) where import Control.Applicative ((<|>))@@ -13,6 +14,7 @@ import Data.ByteString (ByteString) import qualified Data.ByteString as B hiding (pack) import qualified Data.ByteString.Char8 as B (pack, readInteger)+import Data.Maybe (fromMaybe) import Network.HTTP.Date import qualified Network.HTTP.Types as H import qualified Network.HTTP.Types.Header as H@@ -21,6 +23,10 @@ import Network.Wai.Handler.Warp.Header import Numeric (showInt) +#ifndef MIN_VERSION_http_types+#define MIN_VERSION_http_types(x,y,z) 1+#endif+ ---------------------------------------------------------------- data RspFileInfo = WithoutBody H.Status@@ -44,9 +50,7 @@ mcondition = ifmodified reqidx size mtime <|> ifunmodified reqidx size mtime <|> ifrange reqidx size mtime- condition = case mcondition of- Nothing -> unconditional reqidx size- Just x -> x+ condition = fromMaybe (unconditional reqidx size) mcondition ----------------------------------------------------------------
Network/Wai/Handler/Warp/FileInfoCache.hs view
@@ -2,6 +2,7 @@ module Network.Wai.Handler.Warp.FileInfoCache ( FileInfo(..)+ , Hash , withFileInfoCache , getInfo -- test purpose only ) where@@ -10,13 +11,16 @@ import Control.Monad (void) import Control.Reaper import Data.ByteString (ByteString)-import Data.HashMap.Strict (HashMap)-import qualified Data.HashMap.Strict as M+import Data.Hashable (hash) import Network.HTTP.Date+import Network.Wai.Handler.Warp.HashMap (HashMap)+import qualified Network.Wai.Handler.Warp.HashMap as M import System.PosixCompat.Files ---------------------------------------------------------------- +type Hash = Int+ -- | File information. data FileInfo = FileInfo { fileInfoName :: !FilePath@@ -27,7 +31,7 @@ data Entry = Negative | Positive FileInfo type Cache = HashMap FilePath Entry-type FileInfoCache = Reaper Cache (FilePath,Entry)+type FileInfoCache = Reaper Cache (Int,FilePath,Entry) ---------------------------------------------------------------- @@ -51,25 +55,32 @@ else throwIO (userError "FileInfoCache:getInfo") +getInfo' :: Hash -> FilePath -> IO FileInfo+getInfo' _ = getInfo+ ---------------------------------------------------------------- getAndRegisterInfo :: FileInfoCache -> FilePath -> IO FileInfo-getAndRegisterInfo reaper@Reaper{..} path = do+getAndRegisterInfo reaper path = getAndRegisterInfo' reaper (hash path) path++getAndRegisterInfo' :: FileInfoCache -> Hash -> FilePath -> IO FileInfo+getAndRegisterInfo' reaper@Reaper{..} h path = do cache <- reaperRead- case M.lookup path cache of+ case M.lookup h path cache of Just Negative -> throwIO (userError "FileInfoCache:getAndRegisterInfo") Just (Positive x) -> return x- Nothing -> positive reaper path `E.onException` negative reaper path+ Nothing -> positive reaper h path+ `E.onException` negative reaper h path -positive :: FileInfoCache -> FilePath -> IO FileInfo-positive Reaper{..} path = do+positive :: FileInfoCache -> Hash -> FilePath -> IO FileInfo+positive Reaper{..} h path = do info <- getInfo path- reaperAdd (path, Positive info)+ reaperAdd (h, path, Positive info) return info -negative :: FileInfoCache -> FilePath -> IO FileInfo-negative Reaper{..} path = do- reaperAdd (path,Negative)+negative :: FileInfoCache -> Hash -> FilePath -> IO FileInfo+negative Reaper{..} h path = do+ reaperAdd (h, path,Negative) throwIO (userError "FileInfoCache:negative") ----------------------------------------------------------------@@ -77,19 +88,22 @@ -- | Creating a file information cache -- and executing the action in the second argument. -- The first argument is a cache duration in second.-withFileInfoCache :: Int -> ((FilePath -> IO FileInfo) -> IO a) -> IO a-withFileInfoCache 0 action = action getInfo-withFileInfoCache duration action = E.bracket (initialize duration)- terminate- (\c -> action (getAndRegisterInfo c))+withFileInfoCache :: Int+ -> ((FilePath -> IO FileInfo) -> (Hash -> FilePath -> IO FileInfo) -> IO a)+ -> IO a+withFileInfoCache 0 action = action getInfo getInfo'+withFileInfoCache duration action =+ E.bracket (initialize duration)+ terminate+ (\r -> action (getAndRegisterInfo r) (getAndRegisterInfo' r)) -initialize :: Int -> IO FileInfoCache+initialize :: Hash -> IO FileInfoCache initialize duration = mkReaper settings where settings = defaultReaperSettings { reaperAction = override , reaperDelay = duration * 1000000- , reaperCons = uncurry M.insert+ , reaperCons = \(h,k,v) -> M.insert h k v , reaperNull = M.null , reaperEmpty = M.empty }
Network/Wai/Handler/Warp/HTTP2.hs view
@@ -7,11 +7,9 @@ import qualified Control.Exception as E import Control.Monad (when, unless, replicateM_) import Data.ByteString (ByteString)- import Network.HTTP2 import Network.Socket (SockAddr)--import Network.Wai.HTTP2 (HTTP2Application)+import Network.Wai import Network.Wai.Handler.Warp.HTTP2.EncodeFrame import Network.Wai.Handler.Warp.HTTP2.Manager import Network.Wai.Handler.Warp.HTTP2.Receiver@@ -24,31 +22,34 @@ ---------------------------------------------------------------- -http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> HTTP2Application -> IO ()+http2 :: Connection -> InternalInfo -> SockAddr -> Transport -> S.Settings -> (BufSize -> IO ByteString) -> Application -> IO () http2 conn ii addr transport settings readN app = do checkTLS ok <- checkPreface when ok $ do ctx <- newContext- -- Workers & Manager- mgr <- start- let responder = response ctx mgr- action = worker ctx settings tm app responder+ -- Workers, worker manager and timer manager+ mgr <- start settings+ let responder = response ii settings ctx mgr+ action = worker ctx settings app responder setAction mgr action- -- fixme: hard coding: 10- replicateM_ 10 $ spawnAction mgr+ -- The number of workers is 3.+ -- This was carefully chosen based on a lot of benchmarks.+ -- If it is 1, we cannot avoid head-of-line blocking.+ -- If it is large, huge memory is consumed and many+ -- context switches happen.+ replicateM_ 3 $ spawnAction mgr -- Receiver let mkreq = mkRequest ii settings addr tid <- forkIO $ frameReceiver ctx mkreq readN -- Sender -- frameSender is the main thread because it ensures to send -- a goway frame.- frameSender ctx conn ii settings `E.finally` do+ frameSender ctx conn ii settings mgr `E.finally` do clearContext ctx stop mgr killThread tid where- tm = timeoutManager ii checkTLS = case transport of TCP -> return () -- direct tls -> unless (tls12orLater tls) $ goaway conn InadequateSecurity "Weak TLS"
Network/Wai/Handler/Warp/HTTP2/HPACK.hs view
@@ -24,14 +24,15 @@ -- Set-Cookie: contains only one cookie value. -- So, we don't need to split it.-hpackEncodeHeader :: Context -> InternalInfo -> S.Settings -> H.Status- -> H.ResponseHeaders -> IO Builder+hpackEncodeHeader :: Context -> InternalInfo -> S.Settings+ -> H.Status -> H.ResponseHeaders+ -> IO Builder hpackEncodeHeader ctx ii settings s h = do hdr1 <- addServerAndDate h let hdr2 = (":status", status) : map (first foldedCase) hdr1 hpackEncodeRawHeaders ctx hdr2 where- status = B8.pack $ show $ H.statusCode $ s+ status = B8.pack $ show $ H.statusCode s dc = dateCacher ii rspidxhdr = indexResponseHeader h defServer = S.settingsServerName settings@@ -54,20 +55,6 @@ hdrtbl <- readIORef decodeDynamicTable (hdrtbl', hdr) <- decodeHeader hdrtbl hdrblk `E.onException` cleanup writeIORef decodeDynamicTable hdrtbl'- return $ concatCookie hdr+ return hdr where cleanup = E.throwIO $ ConnectionError CompressionError "cannot decompress the header"---- |------ >>> concatCookie [("foo","bar")]--- [("foo","bar")]--- >>> concatCookie [("cookie","a=b"),("foo","bar"),("cookie","c=d"),("cookie","e=f")]--- [("foo","bar"),("cookie","a=b; c=d; e=f")]-concatCookie :: HeaderList -> HeaderList-concatCookie = collect []- where- collect cookies (("cookie",c):rest) = collect (cookies ++ [c]) rest- collect cookies (h:rest) = h : collect cookies rest- collect [] [] = []- collect cookies [] = [("cookie", B.intercalate "; " cookies)]
Network/Wai/Handler/Warp/HTTP2/Manager.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE CPP #-}+{-# LANGUAGE CPP, BangPatterns #-} -- | A thread pool manager. -- The manager has responsibility to spawn and kill@@ -9,7 +9,8 @@ , setAction , stop , spawnAction- , replaceWithAction+ , addMyId+ , deleteMyId ) where #if __GLASGOW_HASKELL__ < 709@@ -20,42 +21,48 @@ import Control.Monad (void) import Data.Set (Set) import qualified Data.Set as Set+import Data.Foldable import Network.Wai.Handler.Warp.IORef+import Network.Wai.Handler.Warp.Settings+import qualified Network.Wai.Handler.Warp.Timeout as T ---------------------------------------------------------------- -data Command = Stop | Spawn | Replace ThreadId+type Action = T.Manager -> IO () -data Manager = Manager (TQueue Command) (IORef (IO ()))+data Command = Stop | Spawn | Add ThreadId | Delete ThreadId +data Manager = Manager (TQueue Command) (IORef Action)+ -- | Starting a thread pool manager. -- Its action is initially set to 'return ()' and should be set -- by 'setAction'. This allows that the action can include -- the manager itself.-start :: IO Manager-start = do- tset <- newThreadSet+start :: Settings -> IO Manager+start set = do q <- newTQueueIO- ref <- newIORef (return ())- void $ forkIO $ go q tset ref+ ref <- newIORef (\_ -> return ())+ timmgr <- T.initialize $ settingsTimeout set * 1000000+ void $ forkIO $ go q Set.empty ref timmgr return $ Manager q ref where- go q tset ref = do+ go q !tset0 ref timmgr = do x <- atomically $ readTQueue q case x of- Stop -> kill tset- Spawn -> next- Replace oldtid -> do- del tset oldtid- next+ Stop -> kill tset0 >> T.killManager timmgr+ Spawn -> next tset0+ Add newtid -> let !tset = add newtid tset0+ in go q tset ref timmgr+ Delete oldtid -> let !tset = del oldtid tset0+ in go q tset ref timmgr where- next = do+ next tset = do action <- readIORef ref- newtid <- forkIO action- add tset newtid- go q tset ref+ newtid <- forkIO (action timmgr)+ let !tset' = add newtid tset+ go q tset' ref timmgr -setAction :: Manager -> IO () -> IO ()+setAction :: Manager -> Action -> IO () setAction (Manager _ ref) action = writeIORef ref action stop :: Manager -> IO ()@@ -64,23 +71,27 @@ spawnAction :: Manager -> IO () spawnAction (Manager q _) = atomically $ writeTQueue q Spawn -replaceWithAction :: Manager -> ThreadId -> IO ()-replaceWithAction (Manager q _) tid = atomically $ writeTQueue q $ Replace tid------------------------------------------------------------------+addMyId :: Manager -> IO ()+addMyId (Manager q _) = do+ tid <- myThreadId+ atomically $ writeTQueue q $ Add tid -newtype ThreadSet = ThreadSet (IORef (Set ThreadId))+deleteMyId :: Manager -> IO ()+deleteMyId (Manager q _) = do+ tid <- myThreadId+ atomically $ writeTQueue q $ Delete tid -newThreadSet :: IO ThreadSet-newThreadSet = ThreadSet <$> newIORef Set.empty+---------------------------------------------------------------- -add :: ThreadSet -> ThreadId -> IO ()-add (ThreadSet ref) tid =- atomicModifyIORef' ref (\set -> (Set.insert tid set, ()))+add :: ThreadId -> Set ThreadId -> Set ThreadId+add tid set = set'+ where+ !set' = Set.insert tid set -del :: ThreadSet -> ThreadId -> IO ()-del (ThreadSet ref) tid =- atomicModifyIORef' ref (\set -> (Set.delete tid set, ()))+del :: ThreadId -> Set ThreadId -> Set ThreadId+del tid set = set'+ where+ !set' = Set.delete tid set -kill :: ThreadSet -> IO ()-kill (ThreadSet ref) = Set.toList <$> readIORef ref >>= mapM_ killThread+kill :: Set ThreadId -> IO ()+kill set = traverse_ killThread set
Network/Wai/Handler/Warp/HTTP2/Receiver.hs view
@@ -9,6 +9,7 @@ #if __GLASGOW_HASKELL__ < 709 import Control.Applicative #endif+import Control.Concurrent import Control.Concurrent.STM import qualified Control.Exception as E import Control.Monad (when, unless, void)@@ -16,7 +17,7 @@ import qualified Data.ByteString as BS import Data.Maybe (isJust) import Network.HTTP2-import Network.HTTP2.Priority+import Network.HTTP2.Priority (toPrecedence, delete, prepare) import Network.Wai.Handler.Warp.HTTP2.EncodeFrame import Network.Wai.Handler.Warp.HTTP2.HPACK import Network.Wai.Handler.Warp.HTTP2.Request@@ -27,7 +28,7 @@ ---------------------------------------------------------------- frameReceiver :: Context -> MkReq -> (BufSize -> IO ByteString) -> IO ()-frameReceiver ctx mkreq recvN = loop `E.catch` sendGoaway+frameReceiver ctx mkreq recvN = loop 0 `E.catch` sendGoaway where Context{ http2settings , streamTable@@ -35,28 +36,33 @@ , continued , currentStreamId , inputQ- , outputQ+ , controlQ } = ctx sendGoaway e | Just (ConnectionError err msg) <- E.fromException e = do csid <- readIORef currentStreamId- let frame = goawayFrame csid err msg- enqueueControl outputQ 0 $ OGoaway frame+ let !frame = goawayFrame csid err msg+ enqueueControl controlQ $ CGoaway frame | otherwise = return () sendReset err sid = do- let frame = resetFrame err sid- enqueueControl outputQ 0 $ OFrame frame+ let !frame = resetFrame err sid+ enqueueControl controlQ $ CFrame frame - loop = do+ loop :: Int -> IO ()+ loop !n+ | n == 6 = do+ yield+ loop 0+ | otherwise = do hd <- recvN frameHeaderLength if BS.null hd then- enqueueControl outputQ 0 OFinish+ enqueueControl controlQ CFinish else do cont <- processStreamGuardingError $ decodeFrameHeader hd- when cont loop+ when cont $ loop (n + 1) - processStreamGuardingError (FrameHeaders, FrameHeader{streamId})+ processStreamGuardingError (_, FrameHeader{streamId}) | isResponse streamId = E.throwIO $ ConnectionError ProtocolError "stream id should be odd" processStreamGuardingError (FrameUnknown _, FrameHeader{payloadLength}) = do mx <- readIORef continued@@ -92,7 +98,7 @@ control ftyp header pl ctx | otherwise = do checkContinued- strm@Stream{streamState,streamContentLength,streamPrecedence} <- getStream+ !strm@Stream{streamState,streamContentLength,streamPrecedence} <- getStream pl <- recvN payloadLength state <- readIORef streamState state' <- stream ftyp header pl ctx state strm@@ -105,7 +111,7 @@ E.throwIO $ StreamError ProtocolError streamId writeIORef streamPrecedence $ toPrecedence pri writeIORef streamState HalfClosed- let req = mkreq vh (return "")+ let !req = mkreq vh (return "") atomically $ writeTQueue inputQ $ Input strm req Nothing -> E.throwIO $ StreamError ProtocolError streamId Open (HasBody hdr pri) -> do@@ -118,7 +124,7 @@ writeIORef streamContentLength $ vhCL vh readQ <- newReadBody q bodySource <- mkSource readQ- let req = mkreq vh (readSource bodySource)+ let !req = mkreq vh (readSource bodySource) atomically $ writeTQueue inputQ $ Input strm req Nothing -> E.throwIO $ StreamError ProtocolError streamId s@(Open Continued{}) -> do@@ -147,63 +153,60 @@ when (isHalfClosed st) $ E.throwIO $ ConnectionError StreamClosed "header must not be sent to half closed" return strm0 Nothing -> do+ -- checkme when (ftyp `notElem` [FrameHeaders,FramePriority]) $ E.throwIO $ ConnectionError ProtocolError "this frame is not allowed in an idel stream"+ csid <- readIORef currentStreamId+ when (streamId <= csid) $+ E.throwIO $ ConnectionError ProtocolError "stream identifier must not decrease" when (ftyp == FrameHeaders) $ do- csid <- readIORef currentStreamId- if streamId <= csid then- E.throwIO $ ConnectionError ProtocolError "stream identifier must not decrease"- else- writeIORef currentStreamId streamId+ writeIORef currentStreamId streamId cnt <- readIORef concurrency- when (cnt >= recommendedConcurrency) $ do- -- Record that the stream is closed, rather than- -- idle, so that receiving frames on it is only a- -- stream error.- consume payloadLength- strm <- newStream concurrency streamId 0- writeIORef (streamState strm) $ Closed $- ResetByMe $ E.toException $- StreamError RefusedStream streamId- insert streamTable streamId strm+ when (cnt >= recommendedConcurrency) $ E.throwIO $ StreamError RefusedStream streamId ws <- initialWindowSize <$> readIORef http2settings- newstrm <- newStream concurrency streamId (fromIntegral ws)- when (ftyp == FrameHeaders) $ opened newstrm+ newstrm <- newStream streamId (fromIntegral ws)+ when (ftyp == FrameHeaders) $ opened ctx newstrm insert streamTable streamId newstrm return newstrm consume = void . recvN +initialFrame :: ByteString+initialFrame = settingsFrame id [(SettingsMaxConcurrentStreams,recommendedConcurrency)]+ ---------------------------------------------------------------- control :: FrameTypeId -> FrameHeader -> ByteString -> Context -> IO Bool-control FrameSettings header@FrameHeader{flags} bs Context{http2settings, outputQ} = do+control FrameSettings header@FrameHeader{flags} bs Context{http2settings, controlQ,firstSettings} = do SettingsFrame alist <- guardIt $ decodeSettingsFrame header bs case checkSettingsList alist of Just x -> E.throwIO x Nothing -> return () unless (testAck flags) $ do- modifyIORef http2settings $ \old -> updateSettings old alist- let frame = settingsFrame setAck []- enqueueControl outputQ 0 $ OSettings frame alist+ modifyIORef' http2settings $ \old -> updateSettings old alist+ let !frame = settingsFrame setAck []+ sent <- readIORef firstSettings+ let !frame' = if sent then frame else BS.append initialFrame frame+ unless sent $ writeIORef firstSettings True+ enqueueControl controlQ $ CSettings frame' alist return True -control FramePing FrameHeader{flags} bs Context{outputQ} =+control FramePing FrameHeader{flags} bs Context{controlQ} = if testAck flags then E.throwIO $ ConnectionError ProtocolError "the ack flag of this ping frame must not be set" else do- let frame = pingFrame bs- enqueueControl outputQ 0 $ OFrame frame+ let !frame = pingFrame bs+ enqueueControl controlQ $ CFrame frame return True -control FrameGoAway _ _ Context{outputQ} = do- enqueueControl outputQ 0 OFinish+control FrameGoAway _ _ Context{controlQ} = do+ enqueueControl controlQ CFinish return False control FrameWindowUpdate header bs Context{connectionWindow} = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs- w <- (n +) <$> atomically (readTVar connectionWindow)+ !w <- (n +) <$> atomically (readTVar connectionWindow) when (isWindowOverflow w) $ E.throwIO $ ConnectionError FlowControlError "control window should be less than 2^31" atomically $ writeTVar connectionWindow w return True@@ -234,8 +237,8 @@ Just p -> do checkPriority p streamNumber return p- let endOfStream = testEndStream flags- endOfHeader = testEndHeader flags+ let !endOfStream = testEndStream flags+ !endOfHeader = testEndHeader flags if endOfHeader then do hdr <- hpackDecodeHeader frag ctx return $ if endOfStream then@@ -250,7 +253,7 @@ -- trailer is not supported. -- let's read and ignore it. HeadersFrame _ _ <- guardIt $ decodeHeadersFrame header bs- let endOfStream = testEndStream flags+ let !endOfStream = testEndStream flags if endOfStream then do atomically $ writeTQueue q "" return HalfClosed@@ -261,18 +264,18 @@ stream FrameData header@FrameHeader{flags,payloadLength,streamId} bs- Context{outputQ} s@(Open (Body q))+ Context{controlQ} s@(Open (Body q)) Stream{streamNumber,streamBodyLength,streamContentLength} = do DataFrame body <- guardIt $ decodeDataFrame header bs- let endOfStream = testEndStream flags+ let !endOfStream = testEndStream flags len0 <- readIORef streamBodyLength let !len = len0 + payloadLength writeIORef streamBodyLength len when (payloadLength /= 0) $ do- let frame1 = windowUpdateFrame 0 payloadLength- frame2 = windowUpdateFrame streamNumber payloadLength- frame = frame1 `BS.append` frame2- enqueueControl outputQ 0 $ OFrame frame+ let !frame1 = windowUpdateFrame 0 payloadLength+ !frame2 = windowUpdateFrame streamNumber payloadLength+ !frame = frame1 `BS.append` frame2+ enqueueControl controlQ $ CFrame frame atomically $ writeTQueue q body if endOfStream then do mcl <- readIORef streamContentLength@@ -285,16 +288,16 @@ return s stream FrameContinuation FrameHeader{flags} frag ctx (Open (Continued rfrags siz n endOfStream pri)) _ = do- let endOfHeader = testEndHeader flags- rfrags' = frag : rfrags- siz' = siz + BS.length frag- n' = n + 1+ let !endOfHeader = testEndHeader flags+ !rfrags' = frag : rfrags+ !siz' = siz + BS.length frag+ !n' = n + 1 when (siz' > 51200) $ -- fixme: hard coding: 50K E.throwIO $ ConnectionError EnhanceYourCalm "Header is too big" when (n' > 10) $ -- fixme: hard coding E.throwIO $ ConnectionError EnhanceYourCalm "Header is too fragmented" if endOfHeader then do- let hdrblk = BS.concat $ reverse rfrags'+ let !hdrblk = BS.concat $ reverse rfrags' hdr <- hpackDecodeHeader hdrblk ctx return $ if endOfStream then Open (NoBody hdr pri)@@ -305,22 +308,21 @@ stream FrameWindowUpdate header@FrameHeader{streamId} bs _ s Stream{streamWindow} = do WindowUpdateFrame n <- guardIt $ decodeWindowUpdateFrame header bs- w <- (n +) <$> atomically (readTVar streamWindow)+ !w <- (n +) <$> atomically (readTVar streamWindow) when (isWindowOverflow w) $ E.throwIO $ StreamError FlowControlError streamId atomically $ writeTVar streamWindow w return s -stream FrameRSTStream header bs _ _ strm = do+stream FrameRSTStream header bs ctx _ strm = do RSTStreamFrame e <- guardIt $ decoderstStreamFrame header bs- let cc = Reset e- closed strm cc+ let !cc = Reset e+ closed ctx strm cc return $ Closed cc -- will be written to streamState again stream FramePriority header bs Context{outputQ,priorityTreeSize} s Stream{streamNumber,streamPrecedence} = do PriorityFrame newpri <- guardIt $ decodePriorityFrame header bs checkPriority newpri streamNumber- -- checkme: this should be tested oldpre <- readIORef streamPrecedence let !newpre = toPrecedence newpri writeIORef streamPrecedence newpre@@ -332,8 +334,8 @@ else do mx <- delete outputQ streamNumber oldpre case mx of- Nothing -> return ()- Just x -> enqueue outputQ streamNumber newpre x+ Nothing -> return ()+ Just out -> enqueueOutput outputQ out return s -- this ordering is important
Network/Wai/Handler/Warp/HTTP2/Request.hs view
@@ -1,4 +1,5 @@ {-# LANGUAGE OverloadedStrings, CPP #-}+{-# LANGUAGE BangPatterns #-} module Network.Wai.Handler.Warp.HTTP2.Request ( mkRequest@@ -25,7 +26,7 @@ #endif import Data.Word8 (isUpper,_colon) import Network.HPACK-import Network.HTTP.Types (RequestHeaders,hRange)+import Network.HTTP.Types (RequestHeaders) import qualified Network.HTTP.Types as H import Network.Socket (SockAddr) import Network.Wai@@ -38,21 +39,24 @@ import qualified Network.Wai.Handler.Warp.Timeout as Timeout data ValidHeaders = ValidHeaders {- vhMethod :: ByteString- , vhPath :: ByteString- , vhAuth :: Maybe ByteString- , vhCL :: Maybe Int -- ^ Content-Length- , vhHeader :: RequestHeaders- }+ vhMethod :: !ByteString+ , vhPath :: !ByteString+ , vhAuth :: !(Maybe ByteString)+ , vhRange :: !(Maybe ByteString)+ , vhReferer :: !(Maybe ByteString)+ , vhUA :: !(Maybe ByteString)+ , vhCL :: !(Maybe Int)+ , vhHeader :: !RequestHeaders+ } deriving Show type MkReq = ValidHeaders -> IO ByteString -> Request mkRequest :: InternalInfo -> S.Settings -> SockAddr -> MkReq-mkRequest ii settings addr (ValidHeaders m p ma _ hdr) body = req+mkRequest ii settings addr (ValidHeaders m p ma mrng mrr mua _ hdr) body = req where (unparsedPath,query) = B8.break (=='?') p- path = H.extractPath unparsedPath- req = Request {+ !path = H.extractPath unparsedPath+ !req = Request { requestMethod = m , httpVersion = http2ver , rawPathInfo = if S.settingsNoParsePath settings then unparsedPath else path@@ -67,68 +71,96 @@ , requestBody = body , vault = vaultValue , requestBodyLength = ChunkedBody -- fixme- , requestHeaderHost = ma- , requestHeaderRange = lookup hRange hdr+ , requestHeaderHost = ma+ , requestHeaderRange = mrng+ , requestHeaderReferer = mrr+ , requestHeaderUserAgent = mua }- th = threadHandle ii- vaultValue = Vault.insert pauseTimeoutKey (Timeout.pause th)- $ Vault.insert getFileInfoKey (fileInfo ii)- Vault.empty+ !th = threadHandle ii+ !vaultValue = Vault.insert pauseTimeoutKey (Timeout.pause th)+ $ Vault.insert getFileInfoKey (fileInfo ii)+ Vault.empty ---------------------------------------------------------------- -data Pseudo = Pseudo {+data Special = Special { colonMethod :: !(Maybe ByteString) , colonPath :: !(Maybe ByteString) , colonAuth :: !(Maybe ByteString)+ , sRange :: !(Maybe ByteString)+ , sReferer :: !(Maybe ByteString)+ , sUA :: !(Maybe ByteString) , contentLen :: !(Maybe ByteString)- }+ } deriving Show -emptyPseudo :: Pseudo-emptyPseudo = Pseudo Nothing Nothing Nothing Nothing+emptySpecial :: Special+emptySpecial = Special Nothing Nothing Nothing Nothing Nothing Nothing Nothing +-- |+--+-- >>> validateHeaders [(":method","GET"),(":path","path")]+-- Just (ValidHeaders {vhMethod = "GET", vhPath = "path", vhAuth = Nothing, vhRange = Nothing, vhReferer = Nothing, vhUA = Nothing, vhCL = Nothing, vhHeader = []})+-- >>> validateHeaders [(":method","GET"),(":path","path"),(":authority","authority"),("accept-language","en")]+-- Just (ValidHeaders {vhMethod = "GET", vhPath = "path", vhAuth = Just "authority", vhRange = Nothing, vhReferer = Nothing, vhUA = Nothing, vhCL = Nothing, vhHeader = [("accept-language","en")]})+-- >>> validateHeaders [(":method","GET"),(":path","path"),("cookie","a=b"),("accept-language","en"),("cookie","c=d"),("cookie","e=f")]+-- Just (ValidHeaders {vhMethod = "GET", vhPath = "path", vhAuth = Nothing, vhRange = Nothing, vhReferer = Nothing, vhUA = Nothing, vhCL = Nothing, vhHeader = [("accept-language","en"),("cookie","a=b; c=d; e=f")]}) validateHeaders :: HeaderList -> Maybe ValidHeaders-validateHeaders hs = case pseudo hs (emptyPseudo,id) of- Just (Pseudo (Just m) (Just p) ma mcl, h)- -> Just $ ValidHeaders m p ma (readInt <$> mcl) h+validateHeaders hs = case pseudo hs emptySpecial of+ Just (Special (Just m) (Just p) ma mrng mrr mua mcl, !h)+ -> Just $! ValidHeaders m p ma mrng mrr mua (readInt <$> mcl) h _ -> Nothing where- pseudo [] (p,b) = Just (p,b [])- pseudo h@((k,v):kvs) (p,b)- | k == ":method" = if isJust (colonMethod p) then+ pseudo [] !s = Just (s,[])+ pseudo h@((k,v):kvs) !s+ | k == ":method" = if isJust (colonMethod s) then Nothing else- pseudo kvs (p { colonMethod = Just v },b)- | k == ":path" = if isJust (colonPath p) then+ pseudo kvs (s { colonMethod = Just v })+ | k == ":path" = if isJust (colonPath s) then Nothing else- pseudo kvs (p { colonPath = Just v },b)- | k == ":authority" = if isJust (colonAuth p) then+ pseudo kvs (s { colonPath = Just v })+ | k == ":authority" = if isJust (colonAuth s) then Nothing else- pseudo kvs (p { colonAuth = Just v },b)- | k == ":scheme" = pseudo kvs (p,b) -- fixme: how to store :scheme?+ pseudo kvs (s { colonAuth = Just v })+ | k == ":scheme" = pseudo kvs s -- fixme: how to store :scheme? | isPseudo k = Nothing- | otherwise = normal h (p,b)+ | otherwise = normal h (s,id,id) - normal [] (p,b) = Just (p,b [])- normal ((k,v):kvs) (p,b)+ normal [] (!s,b,c) = Just (s, mkH b c)+ normal ((k,v):kvs) (!s,b,c) | isPseudo k = Nothing | k == "connection" = Nothing | k == "te" = if v == "trailers" then- normal kvs (p, b . ((mk k,v) :))+ normal kvs (s, b . ((mk k,v) :), c) else Nothing+ | k == "range"+ = normal kvs (s {sRange = Just v }, b . ((mk k,v) :), c)+ | k == "referer"+ = normal kvs (s { sReferer = Just v }, b . ((mk k,v) :), c)+ | k == "user-agent"+ = normal kvs (s { sUA = Just v }, b . ((mk k,v) :), c) | k == "content-length"- = normal kvs (p { contentLen = Just v }, b . ((mk k,v) :))- | k == "host" = if isJust (colonAuth p) then- normal kvs (p,b)+ = normal kvs (s { contentLen = Just v }, b . ((mk k,v) :), c)+ | k == "host" = if isJust (colonAuth s) then+ normal kvs (s, b, c) else- normal kvs (p { colonAuth = Just v },b)+ normal kvs (s { colonAuth = Just v }, b, c)+ | k == "cookie" = normal kvs (s, b, c . (v:)) | otherwise = case BS.find isUpper k of- Nothing -> normal kvs (p, b . ((mk k,v) :))+ Nothing -> normal kvs (s, b . ((mk k,v) :), c) Just _ -> Nothing + mkH b c = h+ where+ !h = b anchor+ !cookieList = c []+ !anchor+ | null cookieList = []+ | otherwise = let !v = BS.intercalate "; " cookieList+ in [("cookie",v)] isPseudo "" = False isPseudo k = BS.head k == _colon
Network/Wai/Handler/Warp/HTTP2/Sender.hs view
@@ -7,24 +7,22 @@ #if __GLASGOW_HASKELL__ < 709 import Control.Applicative #endif-import Control.Concurrent.MVar (putMVar) import Control.Concurrent.STM import qualified Control.Exception as E import Control.Monad (void, when) import qualified Data.ByteString as BS-import qualified Data.ByteString.Builder as B (int32BE)+import Data.ByteString.Builder (Builder) import qualified Data.ByteString.Builder.Extra as B-import Data.Monoid ((<>))+import Data.Maybe (isNothing) import Foreign.Ptr-import qualified Network.HTTP.Types as H import Network.HPACK (setLimitForEncoding) import Network.HTTP2-import Network.HTTP2.Priority-import Network.Wai (FilePart(..))-import Network.Wai.HTTP2 (Trailers, promiseHeaders)+import Network.HTTP2.Priority (isEmptySTM, dequeueSTM, Precedence)+import Network.Wai import Network.Wai.Handler.Warp.Buffer import Network.Wai.Handler.Warp.HTTP2.EncodeFrame import Network.Wai.Handler.Warp.HTTP2.HPACK+import Network.Wai.Handler.Warp.HTTP2.Manager (Manager) import Network.Wai.Handler.Warp.HTTP2.Types import Network.Wai.Handler.Warp.IORef import qualified Network.Wai.Handler.Warp.Settings as S@@ -33,356 +31,369 @@ #ifdef WINDOWS import qualified System.IO as IO #else-import Network.Wai.Handler.Warp.FdCache (getFd)+import Network.Wai.Handler.Warp.FdCache import Network.Wai.Handler.Warp.SendFile (positionRead) import qualified Network.Wai.Handler.Warp.Timeout as T import System.Posix.IO (openFd, OpenFileFlags(..), defaultFileFlags, OpenMode(ReadOnly), closeFd)-import System.Posix.Types (Fd)+import System.Posix.Types #endif ---------------------------------------------------------------- --- | The platform-specific type of an open file to stream from. On Windows we--- don't have pread, so this is just a Handle; on Unix platforms with pread,--- this is a file descriptor supplied by the fd cache.-#ifdef WINDOWS-type OpenFile = IO.Handle-#else-type OpenFile = Fd-#endif- data Leftover = LZero | LOne B.BufferWriter | LTwo BS.ByteString B.BufferWriter- | LFile OpenFile Integer Integer (IO ()) ---------------------------------------------------------------- --- | Run the given action if the stream is not closed; handle any exceptions by--- resetting the stream.-unlessClosed :: Connection -> Stream -> IO () -> IO Bool-unlessClosed Connection{connSendAll}- strm@Stream{streamState,streamNumber}- body = E.handle resetStream $ do- state <- readIORef streamState- if (isClosed state) then return False else body >> return True- where- resetStream e = do- closed strm (ResetByMe e)- let rst = resetFrame InternalError streamNumber- connSendAll rst- return False+getStreamWindowSize :: Stream -> IO WindowSize+getStreamWindowSize Stream{streamWindow} = atomically $ readTVar streamWindow -getWindowSize :: TVar WindowSize -> TVar WindowSize -> IO WindowSize-getWindowSize connWindow strmWindow = do- -- Waiting that the connection window gets open.- cw <- atomically $ do- w <- readTVar connWindow- check (w > 0)- return w- -- This stream window is greater than 0 thanks to the invariant.- sw <- atomically $ readTVar strmWindow- return $ min cw sw+waitStreamWindowSize :: Stream -> STM ()+waitStreamWindowSize Stream{streamWindow} = do+ w <- readTVar streamWindow+ check (w > 0) -frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> IO ()-frameSender ctx@Context{outputQ,connectionWindow,encodeDynamicTable}- conn@Connection{connWriteBuffer,connBufferSize,connSendAll}- ii settings = go `E.catch` ignore- where- initialSettings = [(SettingsMaxConcurrentStreams,recommendedConcurrency)]- initialFrame = settingsFrame id initialSettings- bufHeaderPayload = connWriteBuffer `plusPtr` frameHeaderLength- headerPayloadLim = connBufferSize - frameHeaderLength+waitStreaming :: TBQueue a -> STM ()+waitStreaming tbq = do+ isEmpty <- isEmptyTBQueue tbq+ check (isEmpty == False) - go = do- connSendAll initialFrame- loop+data Switch = C Control+ | O (StreamId,Precedence,Output)+ | Flush - -- ignoring the old priority because the value might be changed.- loop = dequeue outputQ >>= \(_sid,pre,out) -> switch out pre+frameSender :: Context -> Connection -> InternalInfo -> S.Settings -> Manager -> IO ()+frameSender ctx@Context{outputQ,controlQ,connectionWindow,encodeDynamicTable}+ conn@Connection{connWriteBuffer,connBufferSize,connSendAll}+ ii settings mgr = loop 0 `E.catch` ignore+ where+ dequeue off = do+ isEmpty <- isEmptyTQueue controlQ+ if isEmpty then do+ w <- readTVar connectionWindow+ check (w > 0)+ emp <- isEmptySTM outputQ+ if emp then+ if off /= 0 then return Flush else retry+ else+ O <$> dequeueSTM outputQ+ else+ C <$> readTQueue controlQ - ignore :: E.SomeException -> IO ()- ignore _ = return ()+ loop off = do+ x <- atomically $ dequeue off+ case x of+ C ctl -> do+ when (off /= 0) $ flushN off+ cont <- control ctl+ when cont $ loop 0+ O (_,pre,out) -> do+ let strm = outputStream out+ writeIORef (streamPrecedence strm) pre+ off' <- whenReadyOrEnqueueAgain out off $ output out off+ case off' of+ 0 -> loop 0+ _ | off' > 12288 -> flushN off' >> loop 0 -- fixme: hard-coding+ | otherwise -> loop off'+ Flush -> flushN off >> loop 0 - switch OFinish _ = return ()- switch (OGoaway frame) _ = connSendAll frame- switch (OSettings frame alist) _ = do+ control CFinish = return False+ control (CGoaway frame) = connSendAll frame >> return False+ control (CFrame frame) = connSendAll frame >> return True+ control (CSettings frame alist) = do connSendAll frame case lookup SettingsHeaderTableSize alist of Nothing -> return () Just siz -> do dyntbl <- readIORef encodeDynamicTable setLimitForEncoding siz dyntbl- loop- switch (OFrame frame) _ = do- connSendAll frame- loop- switch (OResponse strm s h aux) pre = do- writeIORef (streamPrecedence strm) pre -- fixme- _ <- unlessClosed conn strm $- getWindowSize connectionWindow (streamWindow strm) >>=- sendResponse strm s h aux- loop- switch (ONext strm curr) pre = do- writeIORef (streamPrecedence strm) pre- _ <- unlessClosed conn strm $ do- lim <- getWindowSize connectionWindow (streamWindow strm)- -- Data frame payload- Next datPayloadLen mnext <- curr lim- fillDataHeaderSend strm 0 datPayloadLen- dispatchNext strm mnext- loop- switch (OPush oldStrm push mvar strm s h aux) pre = do- writeIORef (streamPrecedence strm) pre -- fixme- pushed <- unlessClosed conn oldStrm $ do- lim <- getWindowSize connectionWindow (streamWindow strm)- -- Write and send the promise.- builder <- hpackEncodeCIHeaders ctx $ promiseHeaders push- off <- pushContinue (streamNumber oldStrm) (streamNumber strm) builder- flushN $ off + frameHeaderLength- -- TODO(awpr): refactor sendResponse to be able to handle non-zero- -- initial offsets and use that to potentially avoid the extra syscall.- sendResponse strm s h aux lim- putMVar mvar pushed- loop+ return True - -- Send the response headers and as much of the response as is immediately- -- available; shared by normal responses and pushed streams.- sendResponse :: Stream -> H.Status -> H.ResponseHeaders -> Aux -> WindowSize -> IO ()- sendResponse strm s h (Persist sq tvar) lim = do+ output (ONext strm curr mtbq) off0 lim = do+ -- Data frame payload+ let !buf = connWriteBuffer `plusPtr` off0+ !siz = connBufferSize - off0+ Next datPayloadLen mnext <- curr buf siz lim+ off <- fillDataHeader strm off0 datPayloadLen mnext+ maybeEnqueueNext strm mtbq mnext+ return off++ output (ORspn strm rspn) off0 lim = do -- Header frame and Continuation frame let sid = streamNumber strm- builder <- hpackEncodeHeader ctx ii settings s h- len <- headerContinue sid builder False- let total = len + frameHeaderLength- (off, needSend) <- sendHeadersIfNecessary total- let payloadOff = off + frameHeaderLength- Next datPayloadLen mnext <-- fillStreamBodyGetNext ii conn payloadOff lim sq tvar strm- -- If no data was immediately available, avoid sending an- -- empty data frame.- if datPayloadLen > 0 then- fillDataHeaderSend strm total datPayloadLen- else- when needSend $ flushN off- dispatchNext strm mnext+ endOfStream = case rspn of+ RspnNobody _ _ -> True+ _ -> False+ kvlen <- headerContinue sid rspn endOfStream off0+ off <- sendHeadersIfNecessary $ off0 + frameHeaderLength + kvlen+ case rspn of+ RspnNobody _ _ -> do+ closed ctx strm Finished+ return off+ RspnFile _ _ h path mpart -> do+ -- Data frame payload+ let payloadOff = off + frameHeaderLength+ Next datPayloadLen mnext <-+ fillFileBodyGetNext conn ii payloadOff lim h path mpart+ off' <- fillDataHeader strm off datPayloadLen mnext+ maybeEnqueueNext strm Nothing mnext+ return off'+ RspnBuilder _ _ builder -> do+ -- Data frame payload+ let payloadOff = off + frameHeaderLength+ Next datPayloadLen mnext <-+ fillBuilderBodyGetNext conn ii payloadOff lim builder+ off' <- fillDataHeader strm off datPayloadLen mnext+ maybeEnqueueNext strm Nothing mnext+ return off'+ RspnStreaming _ _ tbq -> do+ let payloadOff = off + frameHeaderLength+ Next datPayloadLen mnext <-+ fillStreamBodyGetNext conn payloadOff lim tbq strm+ off' <- fillDataHeader strm off datPayloadLen mnext+ maybeEnqueueNext strm (Just tbq) mnext+ return off' - -- Send the stream's trailers and close the stream.- sendTrailers :: Stream -> Trailers -> IO ()- sendTrailers strm trailers = do- -- Trailers always indicate the end of a stream; send them in- -- consecutive header+continuation frames and end the stream. Some- -- clients dislike empty headers frames, so end the stream with an- -- empty data frame instead, as recommended by the spec.- toFlush <- case trailers of- [] -> frameHeaderLength <$ fillFrameHeader FrameData 0- (streamNumber strm)- (setEndStream defaultFlags)- connWriteBuffer- _ -> do- builder <- hpackEncodeCIHeaders ctx trailers- off <- headerContinue (streamNumber strm) builder True- return (off + frameHeaderLength)- -- 'closed' must be before 'flushN'. If not, the context would be- -- switched to the receiver, resulting in the inconsistency of- -- concurrency.- closed strm Finished- flushN toFlush+ whenReadyOrEnqueueAgain out off body = E.handle resetStream $ do+ state <- readIORef $ streamState strm+ if isClosed state then+ return off+ else case mtbq of+ Just tbq -> checkStreaming tbq+ _ -> checkStreamWindowSize+ where+ strm = outputStream out+ mtbq = outputMaybeTBQueue out+ checkStreaming tbq = do+ isEmpty <- atomically $ isEmptyTBQueue tbq+ if isEmpty then do+ forkAndEnqueueWhenReady (waitStreaming tbq) outputQ out mgr+ return off+ else+ checkStreamWindowSize+ checkStreamWindowSize = do+ sws <- getStreamWindowSize strm+ if sws == 0 then do+ forkAndEnqueueWhenReady (waitStreamWindowSize strm) outputQ out mgr+ return off+ else do+ cws <- atomically $ readTVar connectionWindow -- not 0+ let !lim = min cws sws+ body lim+ resetStream e = do+ closed ctx strm (ResetByMe e)+ let !rst = resetFrame InternalError $ streamNumber strm+ enqueueControl controlQ $ CFrame rst+ return off -- Flush the connection buffer to the socket, where the first 'n' bytes of -- the buffer are filled. flushN :: Int -> IO () flushN n = bufferIO connWriteBuffer n connSendAll - -- A flags value with the end-header flag set iff the argument is B.Done.- maybeEndHeaders B.Done = setEndHeader defaultFlags- maybeEndHeaders _ = defaultFlags-- -- Write PUSH_PROMISE and possibly CONTINUATION frames into the connection- -- buffer, using the given builder as their contents; flush them to the- -- socket as necessary.- pushContinue sid newSid builder = do- let builder' = B.int32BE (fromIntegral newSid) <> builder- (len, signal) <- B.runBuilder builder' bufHeaderPayload headerPayloadLim- let flag = maybeEndHeaders signal- fillFrameHeader FramePushPromise len sid flag connWriteBuffer- continue sid len signal-- -- Write HEADER and possibly CONTINUATION frames.- headerContinue sid builder endOfStream = do- (len, signal) <- B.runBuilder builder bufHeaderPayload headerPayloadLim- let flag0 = maybeEndHeaders signal+ headerContinue sid rspn endOfStream off = do+ let !s = rspnStatus rspn+ !h = rspnHeaders rspn+ builder <- hpackEncodeHeader ctx ii settings s h+ let !offkv = off + frameHeaderLength+ let !bufkv = connWriteBuffer `plusPtr` offkv+ !limkv = connBufferSize - offkv+ (kvlen, signal) <- B.runBuilder builder bufkv limkv+ let flag0 = case signal of+ B.Done -> setEndHeader defaultFlags+ _ -> defaultFlags flag = if endOfStream then setEndStream flag0 else flag0- fillFrameHeader FrameHeaders len sid flag connWriteBuffer- continue sid len signal+ let buf = connWriteBuffer `plusPtr` off+ fillFrameHeader FrameHeaders kvlen sid flag buf+ continue sid kvlen signal - continue _ len B.Done = return len- continue sid len (B.More _ writer) = do- flushN $ len + frameHeaderLength- (len', signal') <- writer bufHeaderPayload headerPayloadLim- let flag = maybeEndHeaders signal'- fillFrameHeader FrameContinuation len' sid flag connWriteBuffer- continue sid len' signal'- continue sid len (B.Chunk bs writer) = do- flushN $ len + frameHeaderLength+ bufHeaderPayload = connWriteBuffer `plusPtr` frameHeaderLength+ headerPayloadLim = connBufferSize - frameHeaderLength++ continue _ kvlen B.Done = return kvlen+ continue sid kvlen (B.More _ writer) = do+ flushN $ kvlen + frameHeaderLength+ -- Now off is 0+ (kvlen', signal') <- writer bufHeaderPayload headerPayloadLim+ let flag = case signal' of+ B.Done -> setEndHeader defaultFlags+ _ -> defaultFlags+ fillFrameHeader FrameContinuation kvlen' sid flag connWriteBuffer+ continue sid kvlen' signal'+ continue sid kvlen (B.Chunk bs writer) = do+ flushN $ kvlen + frameHeaderLength+ -- Now off is 0 let (bs1,bs2) = BS.splitAt headerPayloadLim bs- len' = BS.length bs1+ kvlen' = BS.length bs1 void $ copy bufHeaderPayload bs1- fillFrameHeader FrameContinuation len' sid defaultFlags connWriteBuffer+ fillFrameHeader FrameContinuation kvlen' sid defaultFlags connWriteBuffer if bs2 == "" then- continue sid len' (B.More 0 writer)+ continue sid kvlen' (B.More 0 writer) else- continue sid len' (B.Chunk bs2 writer)-- -- True if the connection buffer has room for a 1-byte data frame.- canFitDataFrame total = total + frameHeaderLength < connBufferSize-- -- Take the appropriate action based on the given 'Control':- -- - If more output is immediately available, re-enqueue the stream in the- -- output queue.- -- - If the output is over and trailers are available, send them now and- -- end the stream.- -- - If we've drained the queue and handed the stream back to its waiter,- -- do nothing.- --- -- This is done after sending any part of the stream body, so it's shared- -- by 'sendResponse' and @switch (ONext ...)@.- dispatchNext :: Stream -> Control DynaNext -> IO ()- dispatchNext _ CNone = return ()- dispatchNext strm (CFinish trailers) = sendTrailers strm trailers- dispatchNext strm (CNext next) = do- let out = ONext strm next- enqueueOrSpawnTemporaryWaiter strm outputQ out+ continue sid kvlen' (B.Chunk bs2 writer) + -- Re-enqueue the stream in the output queue.+ maybeEnqueueNext :: Stream -> Maybe (TBQueue Sequence) -> Maybe DynaNext -> IO ()+ maybeEnqueueNext _ _ Nothing = return ()+ maybeEnqueueNext strm mtbq (Just next) = enqueueOutput outputQ out+ where+ !out = ONext strm next mtbq -- Send headers if there is not room for a 1-byte data frame, and return- -- the offset of the next frame's first header byte and whether the headers- -- still need to be sent.- sendHeadersIfNecessary total- | canFitDataFrame total = return (total, True)- | otherwise = do- flushN total- return (0, False)+ -- the offset of the next frame's first header byte.+ sendHeadersIfNecessary off+ -- True if the connection buffer has room for a 1-byte data frame.+ | off + frameHeaderLength < connBufferSize = return off+ | otherwise = do+ flushN off+ return 0 - fillDataHeaderSend strm otherLen datPayloadLen = do+ fillDataHeader strm off datPayloadLen mnext = do -- Data frame header- let sid = streamNumber strm- buf = connWriteBuffer `plusPtr` otherLen- total = otherLen + frameHeaderLength + datPayloadLen- fillFrameHeader FrameData datPayloadLen sid defaultFlags buf- flushN total- atomically $ do- modifyTVar' connectionWindow (subtract datPayloadLen)- modifyTVar' (streamWindow strm) (subtract datPayloadLen)+ let !sid = streamNumber strm+ !buf = connWriteBuffer `plusPtr` off+ !off' = off + frameHeaderLength + datPayloadLen+ flag = case mnext of+ Nothing -> setEndStream defaultFlags+ _ -> defaultFlags+ fillFrameHeader FrameData datPayloadLen sid flag buf+ when (isNothing mnext) $ closed ctx strm Finished+ atomically $ modifyTVar' connectionWindow (subtract datPayloadLen)+ atomically $ modifyTVar' (streamWindow strm) (subtract datPayloadLen)+ return off' fillFrameHeader ftyp len sid flag buf = encodeFrameHeaderBuf ftyp hinfo buf where hinfo = FrameHeader len flag sid -------------------------------------------------------------------fillStreamBodyGetNext :: InternalInfo -> Connection -> Int -> WindowSize- -> TBQueue Sequence -> TVar Sync -> Stream -> IO Next-fillStreamBodyGetNext ii Connection{connWriteBuffer,connBufferSize}- off lim sq tvar strm = do- let datBuf = connWriteBuffer `plusPtr` off- room = min (connBufferSize - off) lim- (leftover, cont, len) <- runStreamBuilder ii datBuf room sq- nextForStream ii connWriteBuffer connBufferSize sq tvar strm leftover cont len+ ignore :: E.SomeException -> IO ()+ ignore _ = return () ---------------------------------------------------------------- -runStreamBuilder :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence- -> IO (Leftover, Maybe Trailers, BytesFilled)-runStreamBuilder ii buf0 room0 sq = loop buf0 room0 0- where- loop !buf !room !total = do- mbuilder <- atomically $ tryReadTBQueue sq- case mbuilder of- Nothing -> return (LZero, Nothing, total)- Just (SBuilder builder) -> do- (len, signal) <- B.runBuilder builder buf room- let !total' = total + len- case signal of- B.Done -> loop (buf `plusPtr` len) (room - len) total'- B.More _ writer -> return (LOne writer, Nothing, total')- B.Chunk bs writer -> return (LTwo bs writer, Nothing, total')- Just (SFile path part) -> do- (leftover, len) <- runStreamFile ii buf room path part- let !total' = total + len- case leftover of- LZero -> loop (buf `plusPtr` len) (room - len) total'- _ -> return (leftover, Nothing, total')-- Just SFlush -> return (LZero, Nothing, total)- Just (SFinish trailers) -> return (LZero, Just trailers, total)---- | Open the file and start reading into the send buffer.-runStreamFile :: InternalInfo -> Buffer -> BufSize -> FilePath -> FilePart- -> IO (Leftover, BytesFilled)+{-+ResponseFile Status ResponseHeaders FilePath (Maybe FilePart)+ResponseBuilder Status ResponseHeaders Builder+ResponseStream Status ResponseHeaders StreamingBody+ResponseRaw (IO ByteString -> (ByteString -> IO ()) -> IO ()) Response+-} --- | Read the given (OS-specific) file representation into the buffer. On--- non-Windows systems this uses pread; on Windows this ignores the position--- because we use the Handle's internal read position instead (because it's not--- potentially shared with other readers).-readOpenFile :: OpenFile -> Buffer -> BufSize -> Integer -> IO Int+fillBuilderBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Builder -> IO Next+fillBuilderBodyGetNext Connection{connWriteBuffer,connBufferSize}+ _ off lim bb = do+ let datBuf = connWriteBuffer `plusPtr` off+ room = min (connBufferSize - off) lim+ (len, signal) <- B.runBuilder bb datBuf room+ return $ nextForBuilder len signal +fillFileBodyGetNext :: Connection -> InternalInfo -> Int -> WindowSize -> Int -> FilePath -> Maybe FilePart -> IO Next #ifdef WINDOWS-runStreamFile _ buf room path part = do- let start = filePartOffset part- bytes = filePartByteCount part+fillFileBodyGetNext Connection{connWriteBuffer,connBufferSize}+ _ off lim _ path mpart = do+ let datBuf = connWriteBuffer `plusPtr` off+ room = min (connBufferSize - off) lim+ (start, bytes) <- fileStartEnd path mpart -- fixme: how to close Handle? GC does it at this moment.- h <- IO.openBinaryFile path IO.ReadMode+ hdl <- IO.openBinaryFile path IO.ReadMode IO.hSeek h IO.AbsoluteSeek start- fillBufFile buf room h start bytes (return ())--readOpenFile h buf room _ = IO.hGetBufSome h buf room+ len <- IO.hGetBufSome hdl datBuf (mini room bytes)+ let bytes' = bytes - fromIntegral len+ -- fixme: connWriteBuffer connBufferSize+ return $ nextForFile len hdl bytes' (return ()) #else-runStreamFile ii buf room path part = do- let start = filePartOffset part- bytes = filePartByteCount part+fillFileBodyGetNext Connection{connWriteBuffer,connBufferSize}+ ii off lim h path mpart = do (fd, refresh) <- case fdCacher ii of- Just fdcache -> getFd fdcache path+ Just fdcache -> getFd' fdcache h path Nothing -> do fd' <- openFd path ReadOnly Nothing defaultFileFlags{nonBlock=True} th <- T.register (timeoutManager ii) (closeFd fd') return (fd', T.tickle th)- fillBufFile buf room fd start bytes refresh--readOpenFile = positionRead-#endif---- | Read as much of the file as is currently available into the buffer, then--- return a 'Leftover' to indicate whether this file chunk has more data to--- send. If this read hit the end of the file range, return 'LZero'; otherwise--- return 'LFile' so this stream will continue reading from the file the next--- time it's pulled from the queue.-fillBufFile :: Buffer -> BufSize -> OpenFile -> Integer -> Integer -> (IO ())- -> IO (Leftover, BytesFilled)-fillBufFile buf room f start bytes refresh = do- len <- readOpenFile f buf (mini room bytes) start+ let datBuf = connWriteBuffer `plusPtr` off+ room = min (connBufferSize - off) lim+ (start, bytes) <- fileStartEnd path mpart+ len <- positionRead fd datBuf (mini room bytes) start refresh let len' = fromIntegral len- leftover = if bytes > len' then- LFile f (start + len') (bytes - len') refresh- else- LZero- return (leftover, len)+ return $ nextForFile len fd (start + len') (bytes - len') refresh+#endif -mini :: Int -> Integer -> Int-mini i n- | fromIntegral i < n = i- | otherwise = fromIntegral n+fileStartEnd :: FilePath -> Maybe FilePart -> IO (Integer, Integer)+fileStartEnd _ (Just part) =+ return (filePartOffset part, filePartByteCount part)+fileStartEnd _ _ = error "fileStartEnd" -fillBufStream :: InternalInfo -> Buffer -> BufSize -> Leftover- -> TBQueue Sequence -> TVar Sync -> Stream -> DynaNext-fillBufStream ii buf0 siz0 leftover0 sq tvar strm lim0 = do+----------------------------------------------------------------++fillStreamBodyGetNext :: Connection -> Int -> WindowSize -> TBQueue Sequence -> Stream -> IO Next+fillStreamBodyGetNext Connection{connWriteBuffer,connBufferSize}+ off lim sq strm = do+ let datBuf = connWriteBuffer `plusPtr` off+ room = min (connBufferSize - off) lim+ (leftover, cont, len) <- runStreamBuilder datBuf room sq+ return $ nextForStream sq strm leftover cont len++----------------------------------------------------------------++fillBufBuilder :: Leftover -> DynaNext+fillBufBuilder leftover buf0 siz0 lim = do let payloadBuf = buf0 `plusPtr` frameHeaderLength+ room = min (siz0 - frameHeaderLength) lim+ case leftover of+ LZero -> error "fillBufBuilder: LZero"+ LOne writer -> do+ (len, signal) <- writer payloadBuf room+ getNext len signal+ LTwo bs writer+ | BS.length bs <= room -> do+ buf1 <- copy payloadBuf bs+ let len1 = BS.length bs+ (len2, signal) <- writer buf1 (room - len1)+ getNext (len1 + len2) signal+ | otherwise -> do+ let (bs1,bs2) = BS.splitAt room bs+ void $ copy payloadBuf bs1+ getNext room (B.Chunk bs2 writer)+ where+ getNext l s = return $ nextForBuilder l s++nextForBuilder :: BytesFilled -> B.Next -> Next+nextForBuilder len B.Done+ = Next len Nothing+nextForBuilder len (B.More _ writer)+ = Next len $ Just (fillBufBuilder (LOne writer))+nextForBuilder len (B.Chunk bs writer)+ = Next len $ Just (fillBufBuilder (LTwo bs writer))++----------------------------------------------------------------++runStreamBuilder :: Buffer -> BufSize -> TBQueue Sequence+ -> IO (Leftover, Bool, BytesFilled)+runStreamBuilder buf0 room0 sq = loop buf0 room0 0+ where+ loop !buf !room !total = do+ mbuilder <- atomically $ tryReadTBQueue sq+ case mbuilder of+ Nothing -> return (LZero, True, total)+ Just (SBuilder builder) -> do+ (len, signal) <- B.runBuilder builder buf room+ let !total' = total + len+ case signal of+ B.Done -> loop (buf `plusPtr` len) (room - len) total'+ B.More _ writer -> return (LOne writer, True, total')+ B.Chunk bs writer -> return (LTwo bs writer, True, total')+ Just SFlush -> return (LZero, True, total)+ Just SFinish -> return (LZero, False, total)++fillBufStream :: Leftover -> TBQueue Sequence -> Stream -> DynaNext+fillBufStream leftover0 sq strm buf0 siz0 lim0 = do+ let payloadBuf = buf0 `plusPtr` frameHeaderLength room0 = min (siz0 - frameHeaderLength) lim0 case leftover0 of LZero -> do- (leftover, end, len) <- runStreamBuilder ii payloadBuf room0 sq- getNext leftover end len+ (leftover, cont, len) <- runStreamBuilder payloadBuf room0 sq+ getNext leftover cont len LOne writer -> write writer payloadBuf room0 0 LTwo bs writer | BS.length bs <= room0 -> do@@ -392,35 +403,65 @@ | otherwise -> do let (bs1,bs2) = BS.splitAt room0 bs void $ copy payloadBuf bs1- getNext (LTwo bs2 writer) Nothing room0- LFile fd start bytes refresh -> do- (leftover, len) <- fillBufFile payloadBuf room0 fd start bytes refresh- getNext leftover Nothing len+ getNext (LTwo bs2 writer) True room0 where- getNext = nextForStream ii buf0 siz0 sq tvar strm+ getNext l b r = return $ nextForStream sq strm l b r write writer1 buf room sofar = do (len, signal) <- writer1 buf room case signal of B.Done -> do- (leftover, end, extra) <- runStreamBuilder ii (buf `plusPtr` len) (room - len) sq+ (leftover, cont, extra) <- runStreamBuilder (buf `plusPtr` len) (room - len) sq let !total = sofar + len + extra- getNext leftover end total+ getNext leftover cont total B.More _ writer -> do let !total = sofar + len- getNext (LOne writer) Nothing total+ getNext (LOne writer) True total B.Chunk bs writer -> do let !total = sofar + len- getNext (LTwo bs writer) Nothing total+ getNext (LTwo bs writer) True total -nextForStream :: InternalInfo -> Buffer -> BufSize -> TBQueue Sequence- -> TVar Sync -> Stream -> Leftover -> Maybe Trailers- -> BytesFilled -> IO Next-nextForStream _ _ _ _ tvar _ _ (Just trailers) len = do- atomically $ writeTVar tvar $ SyncFinish- return $ Next len $ CFinish trailers-nextForStream ii buf siz sq tvar strm LZero Nothing len = do- let out = ONext strm (fillBufStream ii buf siz LZero sq tvar strm)- atomically $ writeTVar tvar $ SyncNext out- return $ Next len CNone-nextForStream ii buf siz sq tvar strm leftover Nothing len =- return $ Next len (CNext (fillBufStream ii buf siz leftover sq tvar strm))+nextForStream :: TBQueue Sequence -> Stream+ -> Leftover -> Bool -> BytesFilled+ -> Next+nextForStream _ _ _ False len = Next len Nothing+nextForStream sq strm leftOrZero True len =+ Next len $ Just (fillBufStream leftOrZero sq strm)++----------------------------------------------------------------++#ifdef WINDOWS+fillBufFile :: IO.Handle -> Integer -> IO () -> DynaNext+fillBufFile h bytes refresh buf siz lim = do+ let payloadBuf = buf `plusPtr` frameHeaderLength+ room = min (siz - frameHeaderLength) lim+ len <- IO.hGetBufSome h payloadBuf room+ refresh+ let bytes' = bytes - fromIntegral len+ return $ nextForFile len h bytes' refresh++nextForFile :: BytesFilled -> IO.Handle -> Integer -> IO () -> Next+nextForFile 0 _ _ _ = Next 0 Nothing+nextForFile len _ 0 _ = Next len Nothing+nextForFile len h bytes refresh =+ Next len $ Just (fillBufFile h bytes refresh)+#else+fillBufFile :: Fd -> Integer -> Integer -> IO () -> DynaNext+fillBufFile fd start bytes refresh buf siz lim = do+ let payloadBuf = buf `plusPtr` frameHeaderLength+ room = min (siz - frameHeaderLength) lim+ len <- positionRead fd payloadBuf (mini room bytes) start+ let len' = fromIntegral len+ refresh+ return $ nextForFile len fd (start + len') (bytes - len') refresh++nextForFile :: BytesFilled -> Fd -> Integer -> Integer -> IO () -> Next+nextForFile 0 _ _ _ _ = Next 0 Nothing+nextForFile len _ _ 0 _ = Next len Nothing+nextForFile len fd start bytes refresh =+ Next len $ Just (fillBufFile fd start bytes refresh)+#endif++mini :: Int -> Integer -> Int+mini i n+ | fromIntegral i < n = i+ | otherwise = fromIntegral n
Network/Wai/Handler/Warp/HTTP2/Types.hs view
@@ -1,25 +1,24 @@ {-# LANGUAGE OverloadedStrings, CPP #-} {-# LANGUAGE NamedFieldPuns, RecordWildCards #-}+{-# LANGUAGE BangPatterns #-} module Network.Wai.Handler.Warp.HTTP2.Types where import Data.ByteString.Builder (Builder) #if __GLASGOW_HASKELL__ < 709-import Control.Applicative ((<$>), (<*>), pure)+import Control.Applicative ((<$>),(<*>)) #endif import Control.Concurrent (forkIO)-import Control.Concurrent.MVar (MVar) import Control.Concurrent.STM-import Control.Exception (SomeException)+import Control.Exception (SomeException, bracket) import Control.Monad (void)-import Control.Reaper import Data.ByteString (ByteString) import qualified Data.ByteString as BS import Data.IntMap.Strict (IntMap, IntMap) import qualified Data.IntMap.Strict as M import qualified Network.HTTP.Types as H import Network.Wai (Request, FilePart)-import Network.Wai.HTTP2 (PushPromise, Trailers)+import Network.Wai.Handler.Warp.HTTP2.Manager import Network.Wai.Handler.Warp.IORef import Network.Wai.Handler.Warp.Types @@ -38,7 +37,7 @@ where useHTTP2 = case tlsNegotiatedProtocol tls of Nothing -> False- Just proto -> "h2-" `BS.isPrefixOf` proto || proto == "h2"+ Just proto -> "h2-" `BS.isPrefixOf` proto ---------------------------------------------------------------- @@ -46,151 +45,119 @@ ---------------------------------------------------------------- --- | The result of writing data from a stream's queue into the buffer.-data Control a = CFinish Trailers- -- ^ The stream has ended, and the trailers should be sent.- | CNext a- -- ^ The stream has more data immediately available, and we- -- should re-enqueue it when the stream window becomes open.- | CNone- -- ^ The stream queue has been drained and we've handed it off- -- to its dedicated waiter thread, which will re-enqueue it when- -- more data is available.+type DynaNext = Buffer -> BufSize -> WindowSize -> IO Next -instance Show (Control a) where- show (CFinish _) = "CFinish"- show (CNext _) = "CNext"- show CNone = "CNone"+type BytesFilled = Int -type DynaNext = WindowSize -> IO Next+data Next = Next !BytesFilled (Maybe DynaNext) -type BytesFilled = Int+data Rspn = RspnNobody H.Status H.ResponseHeaders+ | RspnStreaming H.Status H.ResponseHeaders (TBQueue Sequence)+ | RspnBuilder H.Status H.ResponseHeaders Builder+ | RspnFile H.Status H.ResponseHeaders Int FilePath (Maybe FilePart) -data Next = Next BytesFilled (Control DynaNext)+rspnStatus :: Rspn -> H.Status+rspnStatus (RspnNobody s _) = s+rspnStatus (RspnStreaming s _ _) = s+rspnStatus (RspnBuilder s _ _) = s+rspnStatus (RspnFile s _ _ _ _ ) = s -data Output = OFinish- -- ^ Terminate the connection.- | OGoaway ByteString- -- ^ Send a goaway frame and terminate the connection.- | OSettings ByteString SettingsList- -- ^ Update settings and send an ack settings frame.- | OFrame ByteString- -- ^ Send an entire pre-encoded frame.- | OResponse Stream H.Status H.ResponseHeaders Aux- -- ^ Send the headers and as much of the response as is immediately- -- available.- | OPush Stream PushPromise (MVar Bool) Stream H.Status H.ResponseHeaders Aux- -- ^ Send a PUSH_PROMISE frame, then act like OResponse; signal the- -- MVar whether the promise has been sent.- | ONext Stream DynaNext- -- ^ Send a chunk of the response.+rspnHeaders :: Rspn -> H.ResponseHeaders+rspnHeaders (RspnNobody _ h) = h+rspnHeaders (RspnStreaming _ h _) = h+rspnHeaders (RspnBuilder _ h _) = h+rspnHeaders (RspnFile _ h _ _ _ ) = h +data Output = ORspn !Stream !Rspn+ | ONext !Stream !DynaNext !(Maybe (TBQueue Sequence))+ outputStream :: Output -> Stream-outputStream (OResponse strm _ _ _) = strm-outputStream (ONext strm _) = strm-outputStream (OPush strm _ _ _ _ _ _) = strm-outputStream _ = error "outputStream"+outputStream (ORspn strm _) = strm+outputStream (ONext strm _ _) = strm +outputMaybeTBQueue :: Output -> Maybe (TBQueue Sequence)+outputMaybeTBQueue (ORspn _ (RspnStreaming _ _ tbq)) = Just tbq+outputMaybeTBQueue (ORspn _ _) = Nothing+outputMaybeTBQueue (ONext _ _ mtbq) = mtbq++data Control = CFinish+ | CGoaway !ByteString+ | CFrame !ByteString+ | CSettings !ByteString !SettingsList+ ---------------------------------------------------------------- --- | An element on the queue between a running stream and the sender; the order--- should consist of any number of 'SFile', 'SBuilder', and 'SFlush', followed--- by a single 'SFinish'.-data Sequence = SFinish Trailers- -- ^ The stream is over; its trailers are provided.+data Sequence = SFinish | SFlush- -- ^ Any buffered data should be sent immediately. | SBuilder Builder- -- ^ Append a chunk of data to the stream.- | SFile FilePath FilePart- -- ^ Append a chunk of a file's contents to the stream. --- | A message from the sender to a stream's dedicated waiter thread.-data Sync = SyncNone- -- ^ Nothing interesting has happened. Go back to sleep.- | SyncFinish- -- ^ The stream has ended.- | SyncNext Output- -- ^ The stream's queue has been drained; wait for more to be- -- available and re-enqueue the given 'Output'.---- | Auxiliary information needed to communicate with a running stream: a queue--- of stream elements ('Sequence') and a 'TVar' connected to its waiter thread.-data Aux = Persist (TBQueue Sequence) (TVar Sync)- ---------------------------------------------------------------- -- | The context for HTTP/2 connection. data Context = Context {- http2settings :: IORef Settings- , streamTable :: StreamTable- -- | Number of active streams initiated by the client; for enforcing our own- -- max concurrency setting.- , concurrency :: IORef Int- -- | Number of active streams initiated by the server; for respecting the- -- client's max concurrency setting.- , pushConcurrency :: IORef Int- , priorityTreeSize :: IORef Int+ http2settings :: !(IORef Settings)+ , firstSettings :: !(IORef Bool)+ , streamTable :: !StreamTable+ , concurrency :: !(IORef Int)+ , priorityTreeSize :: !(IORef Int) -- | RFC 7540 says "Other frames (from any stream) MUST NOT -- occur between the HEADERS frame and any CONTINUATION -- frames that might follow". This field is used to implement -- this requirement.- , continued :: IORef (Maybe StreamId)- , currentStreamId :: IORef StreamId- -- ^ Last client-initiated stream ID we've handled.- , nextPushStreamId :: IORef StreamId- -- ^ Next available server-initiated stream ID.- , inputQ :: TQueue Input- , outputQ :: PriorityTree Output- , encodeDynamicTable :: IORef DynamicTable- , decodeDynamicTable :: IORef DynamicTable- , connectionWindow :: TVar WindowSize+ , continued :: !(IORef (Maybe StreamId))+ , currentStreamId :: !(IORef StreamId)+ , inputQ :: !(TQueue Input)+ , outputQ :: !(PriorityTree Output)+ , controlQ :: !(TQueue Control)+ , encodeDynamicTable :: !(IORef DynamicTable)+ , decodeDynamicTable :: !(IORef DynamicTable)+ , connectionWindow :: !(TVar WindowSize) } ---------------------------------------------------------------- newContext :: IO Context newContext = Context <$> newIORef defaultSettings- <*> initialize 10 -- fixme: hard coding: 10- <*> newIORef 0+ <*> newIORef False+ <*> newStreamTable <*> newIORef 0 <*> newIORef 0 <*> newIORef Nothing <*> newIORef 0- <*> newIORef 2 -- first server push stream; 0 is reserved <*> newTQueueIO <*> newPriorityTree+ <*> newTQueueIO <*> (newDynamicTableForEncoding defaultDynamicTableSize >>= newIORef) <*> (newDynamicTableForDecoding defaultDynamicTableSize >>= newIORef) <*> newTVarIO defaultInitialWindowSize clearContext :: Context -> IO ()-clearContext ctx = void $ reaperStop $ streamTable ctx+clearContext _ctx = return () ---------------------------------------------------------------- data OpenState = JustOpened | Continued [HeaderBlockFragment]- Int -- Total size- Int -- The number of continuation frames- Bool -- End of stream- Priority- | NoBody HeaderList Priority- | HasBody HeaderList Priority- | Body (TQueue ByteString)+ !Int -- Total size+ !Int -- The number of continuation frames+ !Bool -- End of stream+ !Priority+ | NoBody HeaderList !Priority+ | HasBody HeaderList !Priority+ | Body !(TQueue ByteString) data ClosedCode = Finished | Killed- | Reset ErrorCodeId+ | Reset !ErrorCodeId | ResetByMe SomeException deriving Show data StreamState = Idle- | Open OpenState+ | Open !OpenState | HalfClosed- | Closed ClosedCode+ | Closed !ClosedCode isIdle :: StreamState -> Bool isIdle Idle = True@@ -217,96 +184,75 @@ ---------------------------------------------------------------- data Stream = Stream {- streamNumber :: StreamId- , streamState :: IORef StreamState+ streamNumber :: !StreamId+ , streamState :: !(IORef StreamState) -- Next two fields are for error checking.- , streamContentLength :: IORef (Maybe Int)- , streamBodyLength :: IORef Int- , streamWindow :: TVar WindowSize- , streamPrecedence :: IORef Precedence- -- | The concurrency IORef in which this stream has been counted. The client- -- and server each have separate concurrency values to respect, so pushed- -- streams need to decrement a different count when they're closed. This- -- should be either @concurrency ctx@ or @pushConcurrency ctx@.- , concurrencyRef :: IORef Int+ , streamContentLength :: !(IORef (Maybe Int))+ , streamBodyLength :: !(IORef Int)+ , streamWindow :: !(TVar WindowSize)+ , streamPrecedence :: !(IORef Precedence) } instance Show Stream where show s = show (streamNumber s) -newStream :: IORef Int -> StreamId -> WindowSize -> IO Stream-newStream ref sid win =- Stream sid <$> newIORef Idle- <*> newIORef Nothing- <*> newIORef 0- <*> newTVarIO win- <*> newIORef defaultPrecedence- <*> pure ref+newStream :: StreamId -> WindowSize -> IO Stream+newStream sid win = Stream sid <$> newIORef Idle+ <*> newIORef Nothing+ <*> newIORef 0+ <*> newTVarIO win+ <*> newIORef defaultPrecedence ---------------------------------------------------------------- -opened :: Stream -> IO ()-opened Stream{concurrencyRef,streamState} = do- atomicModifyIORef' concurrencyRef (\x -> (x+1,()))+opened :: Context -> Stream -> IO ()+opened Context{concurrency} Stream{streamState} = do+ atomicModifyIORef' concurrency (\x -> (x+1,())) writeIORef streamState (Open JustOpened) -closed :: Stream -> ClosedCode -> IO ()-closed Stream{concurrencyRef,streamState} cc = do- atomicModifyIORef' concurrencyRef (\x -> (x-1,()))- writeIORef streamState (Closed cc)+closed :: Context -> Stream -> ClosedCode -> IO ()+closed Context{concurrency,streamTable} Stream{streamState,streamNumber} cc = do+ remove streamTable streamNumber+ atomicModifyIORef' concurrency (\x -> (x-1,()))+ writeIORef streamState (Closed cc) -- anyway ---------------------------------------------------------------- -type StreamTable = Reaper (IntMap Stream) (M.Key, Stream)--initialize :: Int -> IO StreamTable-initialize duration = mkReaper settings- where- settings = defaultReaperSettings {- reaperAction = clean- , reaperDelay = duration * 1000000- , reaperCons = uncurry M.insert- , reaperNull = M.null- , reaperEmpty = M.empty- }+newtype StreamTable = StreamTable (IORef (IntMap Stream)) -clean :: IntMap Stream -> IO (IntMap Stream -> IntMap Stream)-clean old = do- new <- M.fromAscList <$> prune oldlist []- return $ M.union new- where- oldlist = M.toDescList old- prune [] lst = return lst- prune (x@(_,s):xs) lst = do- st <- readIORef (streamState s)- if isClosed st then- prune xs lst- else- prune xs (x:lst)+newStreamTable :: IO StreamTable+newStreamTable = StreamTable <$> newIORef M.empty insert :: StreamTable -> M.Key -> Stream -> IO ()-insert strmtbl k v = reaperAdd strmtbl (k,v)+insert (StreamTable ref) k v = atomicModifyIORef' ref $ \m ->+ let !m' = M.insert k v m+ in (m', ()) +remove :: StreamTable -> M.Key -> IO ()+remove (StreamTable ref) k = atomicModifyIORef' ref $ \m ->+ let !m' = M.delete k m+ in (m', ())+ search :: StreamTable -> M.Key -> IO (Maybe Stream)-search strmtbl k = M.lookup k <$> reaperRead strmtbl+search (StreamTable ref) k = M.lookup k <$> readIORef ref +{-# INLINE forkAndEnqueueWhenReady #-}+forkAndEnqueueWhenReady :: STM () -> PriorityTree Output -> Output -> Manager -> IO ()+forkAndEnqueueWhenReady wait outQ out mgr = bracket setup teardown $ \_ ->+ void . forkIO $ do+ atomically wait+ enqueueOutput outQ out+ where+ setup = addMyId mgr+ teardown _ = deleteMyId mgr --- INVARIANT: streams in the output queue have non-zero window size.-enqueueWhenWindowIsOpen :: PriorityTree Output -> Output -> IO ()-enqueueWhenWindowIsOpen outQ out = do+{-# INLINE enqueueOutput #-}+enqueueOutput :: PriorityTree Output -> Output -> IO ()+enqueueOutput outQ out = do let Stream{..} = outputStream out- atomically $ do- x <- readTVar streamWindow- check (x > 0) pre <- readIORef streamPrecedence enqueue outQ streamNumber pre out -enqueueOrSpawnTemporaryWaiter :: Stream -> PriorityTree Output -> Output -> IO ()-enqueueOrSpawnTemporaryWaiter Stream{..} outQ out = do- sw <- atomically $ readTVar streamWindow- if sw == 0 then- -- This waiter waits only for the stream window.- void $ forkIO $ enqueueWhenWindowIsOpen outQ out- else do- pre <- readIORef streamPrecedence- enqueue outQ streamNumber pre out+{-# INLINE enqueueControl #-}+enqueueControl :: TQueue Control -> Control -> IO ()+enqueueControl ctlQ ctl = atomically $ writeTQueue ctlQ ctl
Network/Wai/Handler/Warp/HTTP2/Worker.hs view
@@ -1,252 +1,198 @@-{-# LANGUAGE DeriveDataTypeable #-}-{-# LANGUAGE RankNTypes #-} {-# LANGUAGE NamedFieldPuns #-} {-# LANGUAGE PatternGuards #-} {-# LANGUAGE CPP #-}+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE BangPatterns #-} module Network.Wai.Handler.Warp.HTTP2.Worker (- Respond+ Responder , response , worker ) where #if __GLASGOW_HASKELL__ < 709 import Control.Applicative+import Data.Monoid (mempty) #endif-import Control.Concurrent import Control.Concurrent.STM-import Control.Exception (Exception, SomeException(..), AsyncException(..))+import Control.Exception (SomeException(..), AsyncException(..)) import qualified Control.Exception as E-import Control.Monad (void, when)-import Data.Typeable+import Control.Monad (when)+import Data.ByteString.Builder (byteString)+import Data.Hashable (hash) import qualified Network.HTTP.Types as H import Network.HTTP2-import Network.HTTP2.Priority import Network.Wai+import Network.Wai.Handler.Warp.File+import Network.Wai.Handler.Warp.FileInfoCache import Network.Wai.Handler.Warp.HTTP2.EncodeFrame import Network.Wai.Handler.Warp.HTTP2.Manager import Network.Wai.Handler.Warp.HTTP2.Types+import Network.Wai.Handler.Warp.Header import Network.Wai.Handler.Warp.IORef-import Network.Wai.HTTP2- ( Chunk(..)- , HTTP2Application- , PushPromise- , Responder(runResponder)- , RespondFunc- )+import qualified Network.Wai.Handler.Warp.Response as R import qualified Network.Wai.Handler.Warp.Settings as S import qualified Network.Wai.Handler.Warp.Timeout as T+import Network.Wai.Handler.Warp.Types+import Network.Wai.Internal (Response(..), ResponseReceived(..), ResponseReceived(..)) ---------------------------------------------------------------- --- | An 'HTTP2Application' takes a function of status, headers, trailers, and--- body; this type implements that by currying some internal arguments.------ The token type of the RespondFunc is set to be (). This is a bit--- anti-climactic, but the real benefit of the token type is that the--- application is forced to call the responder, and making it a boring type--- doesn't break that property.------ This is the argument to a 'Responder'.-type Respond = IO () -> Stream -> RespondFunc ()---- | This function is passed to workers. They also pass responses from--- 'HTTP2Application's to this function. This function enqueues commands for--- the HTTP/2 sender.-response :: Context -> Manager -> ThreadContinue -> Respond-response ctx mgr tconf tickle strm s h strmbdy = do- -- TODO(awpr) HEAD requests will still stream.-- -- We must not exit this WAI application.- -- If the application exits, streaming would be also closed.- -- So, this work occupies this thread.- --- -- We need to increase the number of workers.- myThreadId >>= replaceWithAction mgr- -- After this work, this thread stops to decrease the number of workers.- setThreadContinue tconf False-- runStream ctx OResponse tickle strm s h strmbdy---- | Set up a waiter thread and run the stream body with functions to enqueue--- 'Sequence's on the stream's queue.-runStream :: Context- -> (Stream -> H.Status -> H.ResponseHeaders -> Aux -> Output)- -> Respond-runStream Context{outputQ} mkOutput tickle strm s h strmbdy = do- -- Since 'Body' is loop, we cannot control it.- -- So, let's serialize 'Builder' with a designated queue.- sq <- newTBQueueIO 10 -- fixme: hard coding: 10- tvar <- newTVarIO SyncNone- let out = mkOutput strm s h (Persist sq tvar)- -- Since we must not enqueue an empty queue to the priority- -- queue, we spawn a thread to ensure that the designated- -- queue is not empty.- void $ forkIO $ waiter tvar sq strm outputQ- atomically $ writeTVar tvar $ SyncNext out- let write chunk = do- atomically $ writeTBQueue sq $ case chunk of- BuilderChunk b -> SBuilder b- FileChunk path part -> SFile path part- tickle- flush = atomically $ writeTBQueue sq SFlush- trailers <- strmbdy write flush- atomically $ writeTBQueue sq $ SFinish trailers---- | Handle abnormal termination of a stream: mark it as closed, send a reset--- frame, and call the user's 'settingsOnException' handler if applicable.-cleanupStream :: Context -> S.Settings -> Stream -> Maybe Request -> Maybe SomeException -> IO ()-cleanupStream Context{outputQ} set strm req me = do- closed strm Killed- let sid = streamNumber strm- frame = resetFrame InternalError sid- enqueueControl outputQ sid $ OFrame frame- case me of- Nothing -> return ()- Just e -> S.settingsOnException set req e+-- | The wai definition is 'type Application = Request -> (Response -> IO ResponseReceived) -> IO ResponseReceived'.+-- This type implements the second argument (Response -> IO ResponseReceived)+-- with extra arguments.+type Responder = ThreadContinue -> T.Handle -> Stream -> Request ->+ Response -> IO ResponseReceived --- | Push the given 'Responder' to the client if the settings allow it--- (specifically 'enablePush' and 'maxConcurrentStreams'). Returns 'True' if--- the stream was actually pushed.------ This is the push function given to an 'HTTP2Application'.-pushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool-pushResponder ctx set strm promise responder = do- let Context{ http2settings- , pushConcurrency- } = ctx- cnt <- readIORef pushConcurrency- settings <- readIORef http2settings- let enabled = enablePush settings- fits = maybe True (cnt <) $ maxConcurrentStreams settings- canPush = fits && enabled- if canPush then- actuallyPushResponder ctx set strm promise responder- else- return False+-- | This function is passed to workers.+-- They also pass 'Response's from 'Application's to this function.+-- This function enqueues commands for the HTTP/2 sender.+response :: InternalInfo -> S.Settings -> Context -> Manager -> Responder+response ii settings Context{outputQ} mgr tconf th strm req rsp+ | R.hasBody s0 = case rsp of+ ResponseStream _ _ strmbdy+ | isHead -> responseNoBody s0 hs0+ | otherwise -> responseStreaming strmbdy+ ResponseBuilder _ _ b+ | isHead -> responseNoBody s0 hs0+ | otherwise -> responseBuilderBody s0 hs0 b+ ResponseFile _ _ p mp -> responseFileXXX p mp+ ResponseRaw _ _ -> error "HTTP/2 does not support ResponseRaw"+ | otherwise = responseNoBody s0 hs0+ where+ !isHead = requestMethod req == H.methodHead+ !s0 = responseStatus rsp+ !hs0 = responseHeaders rsp+ !logger = S.settingsLogger settings --- | Set up a pushed stream and run the 'Responder' in its own thread. Waits--- for the sender thread to handle the push request. This can fail to push the--- stream and return 'False' if the sender dequeued the push request after the--- associated stream was closed.-actuallyPushResponder :: Context -> S.Settings -> Stream -> PushPromise -> Responder -> IO Bool-actuallyPushResponder ctx set strm promise responder = do- let Context{ http2settings- , nextPushStreamId- , pushConcurrency- , streamTable- } = ctx- -- Claim the next outgoing stream.- newSid <- atomicModifyIORef nextPushStreamId $ \sid -> (sid+2, sid)- ws <- initialWindowSize <$> readIORef http2settings+ -- Ideally, log messages should be written when responses are+ -- actually sent. But there is no way to keep good memory usage+ -- (resist to Request leak) and throughput. By compromise,+ -- log message are written here even the window size of streams+ -- is 0. - newStrm <- newStream pushConcurrency newSid ws- -- Section 5.3.5 of RFC 7540 defines the weight of push promise is 16.- -- But we need not to follow the spec. So, this value would change- -- if necessary.- writeIORef (streamPrecedence newStrm) $- toPrecedence $ defaultPriority { streamDependency = streamNumber strm }- opened newStrm- insert streamTable newSid newStrm+ responseNoBody s hs = do+ logger req s Nothing+ setThreadContinue tconf True+ let rspn = RspnNobody s hs+ out = ORspn strm rspn+ enqueueOutput outputQ out+ return ResponseReceived - -- Set up a channel for the sender to report back whether it pushed the- -- stream.- mvar <- newEmptyMVar+ responseBuilderBody s hs bdy = do+ logger req s Nothing+ setThreadContinue tconf True+ let rspn = RspnBuilder s hs bdy+ out = ORspn strm rspn+ enqueueOutput outputQ out+ return ResponseReceived - let mkOutput = OPush strm promise mvar- tickle = return ()- respond = runStream ctx mkOutput+ responseFileXXX path Nothing = do+ let !h = hash $ rawPathInfo req+ efinfo <- E.try $ fileInfo' ii h path+ case efinfo of+ Left (_ex :: E.IOException) -> response404+ Right finfo -> case conditionalRequest finfo hs0 (indexRequestHeader (requestHeaders req)) of+ WithoutBody s -> responseNoBody s hs0+ WithBody s hs beg len -> responseFile2XX s hs h path (Just (FilePart beg len (fileInfoSize finfo))) - -- TODO(awpr): synthesize a Request for 'settingsOnException'?- _ <- forkIO $ runResponder responder (respond tickle newStrm) `E.catch`- (cleanupStream ctx set strm Nothing . Just)+ responseFileXXX path mpart = responseFile2XX s0 hs0 h path mpart+ where+ !h = hash $ rawPathInfo req - takeMVar mvar+ responseFile2XX s hs h path mpart+ | isHead = do+ logger req s Nothing+ responseNoBody s hs+ | otherwise = do+ logger req s (filePartByteCount <$> mpart)+ setThreadContinue tconf True+ let rspn = RspnFile s hs h path mpart+ out = ORspn strm rspn+ enqueueOutput outputQ out+ return ResponseReceived -data Break = Break deriving (Show, Typeable)+ response404 = responseBuilderBody s hs body+ where+ s = H.notFound404+ hs = R.replaceHeader H.hContentType "text/plain; charset=utf-8" hs0+ body = byteString "File not found" -instance Exception Break+ responseStreaming strmbdy = do+ logger req s0 Nothing+ -- We must not exit this WAI application.+ -- If the application exits, streaming would be also closed.+ -- So, this work occupies this thread.+ --+ -- We need to increase the number of workers.+ spawnAction mgr+ -- After this work, this thread stops to decease+ -- the number of workers.+ setThreadContinue tconf False+ -- Since 'StreamingBody' is loop, we cannot control it.+ -- So, let's serialize 'Builder' with a designated queue.+ tbq <- newTBQueueIO 10 -- fixme: hard coding: 10+ let rspn = RspnStreaming s0 hs0 tbq+ out = ORspn strm rspn+ enqueueOutput outputQ out+ let push b = do+ atomically $ writeTBQueue tbq (SBuilder b)+ T.tickle th+ flush = atomically $ writeTBQueue tbq SFlush+ _ <- strmbdy push flush+ atomically $ writeTBQueue tbq SFinish+ deleteMyId mgr+ return ResponseReceived -worker :: Context- -> S.Settings- -> T.Manager- -> HTTP2Application- -> (ThreadContinue -> Respond)- -> IO ()-worker ctx@Context{inputQ} set tm app respond = do- tid <- myThreadId+worker :: Context -> S.Settings -> Application -> Responder -> T.Manager -> IO ()+worker ctx@Context{inputQ,controlQ} set app responder tm = do sinfo <- newStreamInfo tcont <- newThreadContinue- let setup = T.register tm $ E.throwTo tid Break- E.bracket setup T.cancel $ go sinfo tcont+ E.bracket (T.registerKillThread tm) T.cancel $ go sinfo tcont where go sinfo tcont th = do setThreadContinue tcont True- ex <- E.try $ do T.pause th- Input strm req <- atomically $ readTQueue inputQ- setStreamInfo sinfo strm req+ inp@(Input strm req) <- atomically $ readTQueue inputQ+ setStreamInfo sinfo inp T.resume th T.tickle th- let responder = app req $ pushResponder ctx set strm- runResponder responder $ respond tcont (T.tickle th) strm+ app req $ responder tcont th strm req cont1 <- case ex of- Right () -> return True+ Right ResponseReceived -> return True Left e@(SomeException _)- | Just Break <- E.fromException e -> do+ -- killed by the local worker manager+ | Just ThreadKilled <- E.fromException e -> return False+ -- killed by the local timeout manager+ | Just T.TimeoutThread <- E.fromException e -> do cleanup sinfo Nothing return True- -- killed by the sender- | Just ThreadKilled <- E.fromException e -> do- cleanup sinfo Nothing- return False | otherwise -> do- cleanup sinfo (Just e)+ cleanup sinfo $ Just e return True cont2 <- getThreadContinue tcont+ clearStreamInfo sinfo when (cont1 && cont2) $ go sinfo tcont th cleanup sinfo me = do- m <- getStreamInfo sinfo- case m of- Nothing -> return ()- Just (strm,req) -> do- cleanupStream ctx set strm (Just req) me- clearStreamInfo sinfo---- | A dedicated waiter thread to re-enqueue the stream in the priority tree--- whenever output becomes available. When the sender drains the queue and--- moves on to another stream, it drops a message in the 'TVar', and this--- thread wakes up, waits for more output to become available, and re-enqueues--- the stream.-waiter :: TVar Sync -> TBQueue Sequence -> Stream -> PriorityTree Output -> IO ()-waiter tvar sq strm outQ = do- -- waiting for actions other than SyncNone- mx <- atomically $ do- mout <- readTVar tvar- case mout of- SyncNone -> retry- SyncNext out -> do- writeTVar tvar SyncNone- return $ Just out- SyncFinish -> return Nothing- case mx of- Nothing -> return ()- Just out -> do- -- ensuring that the streaming queue is not empty.- atomically $ do- isEmpty <- isEmptyTBQueue sq- when isEmpty retry- -- ensuring that stream window is greater than 0.- enqueueWhenWindowIsOpen outQ out- waiter tvar sq strm outQ+ minp <- getStreamInfo sinfo+ case minp of+ Nothing -> return ()+ Just (Input strm req) -> do+ closed ctx strm Killed+ let frame = resetFrame InternalError (streamNumber strm)+ enqueueControl controlQ $ CFrame frame+ case me of+ Nothing -> return ()+ Just e -> S.settingsOnException set (Just req) e ---------------------------------------------------------------- -- | It would nice if responders could return values to workers. -- Unfortunately, 'ResponseReceived' is already defined in WAI 2.0. -- It is not wise to change this type.--- So, a reference is shared by a 'Respond' and its worker.+-- So, a reference is shared by a responder and its worker. -- The reference refers a value of this type as a return value. -- If 'True', the worker continue to serve requests. -- Otherwise, the worker get finished.@@ -264,7 +210,7 @@ ---------------------------------------------------------------- -- | The type to store enough information for 'settingsOnException'.-newtype StreamInfo = StreamInfo (IORef (Maybe (Stream,Request)))+newtype StreamInfo = StreamInfo (IORef (Maybe Input)) newStreamInfo :: IO StreamInfo newStreamInfo = StreamInfo <$> newIORef Nothing@@ -272,8 +218,8 @@ clearStreamInfo :: StreamInfo -> IO () clearStreamInfo (StreamInfo ref) = writeIORef ref Nothing -setStreamInfo :: StreamInfo -> Stream -> Request -> IO ()-setStreamInfo (StreamInfo ref) strm req = writeIORef ref $ Just (strm,req)+setStreamInfo :: StreamInfo -> Input -> IO ()+setStreamInfo (StreamInfo ref) inp = writeIORef ref $ Just inp -getStreamInfo :: StreamInfo -> IO (Maybe (Stream, Request))+getStreamInfo :: StreamInfo -> IO (Maybe Input) getStreamInfo (StreamInfo ref) = readIORef ref
+ Network/Wai/Handler/Warp/HashMap.hs view
@@ -0,0 +1,24 @@+module Network.Wai.Handler.Warp.HashMap where++import Data.IntMap.Strict (IntMap)+import qualified Data.IntMap.Strict as I+import Data.Map.Strict (Map)+import qualified Data.Map.Strict as M++type Hash = Int+newtype HashMap k v = HashMap (IntMap (Map k v))++empty :: HashMap k v+empty = HashMap $ I.empty++null :: HashMap k v -> Bool+null (HashMap hm) = I.null hm++insert :: Ord k => Hash -> k -> v -> HashMap k v -> HashMap k v+insert h k v (HashMap hm) = HashMap $ I.insertWith f h m hm+ where+ m = M.singleton k v+ f = M.union -- fimxe++lookup :: Ord k => Hash -> k -> HashMap k v -> Maybe v+lookup h k (HashMap hm) = I.lookup h hm >>= M.lookup k
Network/Wai/Handler/Warp/Header.hs view
@@ -26,6 +26,8 @@ | ReqIfModifiedSince | ReqIfUnmodifiedSince | ReqIfRange+ | ReqReferer+ | ReqUserAgent deriving (Enum,Bounded) -- | The size for 'IndexedHeader' for HTTP Request.@@ -45,6 +47,8 @@ requestKeyIndex "if-modified-since" = fromEnum ReqIfModifiedSince requestKeyIndex "if-unmodified-since" = fromEnum ReqIfUnmodifiedSince requestKeyIndex "if-range" = fromEnum ReqIfRange+requestKeyIndex "referer" = fromEnum ReqReferer+requestKeyIndex "user-agent" = fromEnum ReqUserAgent requestKeyIndex _ = -1 defaultIndexRequestHeader :: IndexedHeader
Network/Wai/Handler/Warp/Internal.hs view
@@ -8,18 +8,7 @@ , runSettingsConnection , runSettingsConnectionMaker , runSettingsConnectionMakerSecure- , runServe- , runServeEnv- , runServeSettings- , runServeSettingsSocket- , runServeSettingsConnection- , runServeSettingsConnectionMaker- , runServeSettingsConnectionMakerSecure , Transport (..)- -- * ServeConnection- , ServeConnection- , serveDefault- , serveHTTP2 -- * Connection , Connection (..) , socketConnection
Network/Wai/Handler/Warp/Request.hs view
@@ -85,8 +85,10 @@ , requestBody = rbody' , vault = vaultValue , requestBodyLength = bodyLength- , requestHeaderHost = idxhdr ! fromEnum ReqHost- , requestHeaderRange = idxhdr ! fromEnum ReqRange+ , requestHeaderHost = idxhdr ! fromEnum ReqHost+ , requestHeaderRange = idxhdr ! fromEnum ReqRange+ , requestHeaderReferer = idxhdr ! fromEnum ReqReferer+ , requestHeaderUserAgent = idxhdr ! fromEnum ReqUserAgent } return (req, remainingRef, idxhdr, rbodyFlush) where
Network/Wai/Handler/Warp/RequestHeader.hs view
@@ -3,7 +3,6 @@ module Network.Wai.Handler.Warp.RequestHeader ( parseHeaderLines- , parseByteRanges ) where import Control.Exception (throwIO)@@ -18,7 +17,7 @@ import Foreign.Storable (peek) import qualified Network.HTTP.Types as H import Network.Wai.Handler.Warp.Types-import Network.Wai.Internal (parseByteRanges)+ -- $setup -- >>> :set -XOverloadedStrings
Network/Wai/Handler/Warp/Response.hs view
@@ -11,6 +11,7 @@ , addDate , addServer , hasBody+ , replaceHeader ) where #ifndef MIN_VERSION_base@@ -35,6 +36,7 @@ import Data.ByteString.Builder.Extra (flush) import qualified Data.CaseInsensitive as CI import Data.Function (on)+import Data.Hashable (hash) import Data.List (deleteBy) import Data.Maybe #if MIN_VERSION_base(4,5,0)@@ -146,14 +148,14 @@ -- and status, the response to HEAD is processed here. -- -- See definition of rsp below for proper body stripping.- (ms, mlen) <- sendRsp conn ii ver s hs rsp+ (ms, mlen) <- sendRsp conn ii req ver s hs rsp case ms of Nothing -> return () Just realStatus -> logger req realStatus mlen T.tickle th return ret else do- _ <- sendRsp conn ii ver s hs RspNoBody+ _ <- sendRsp conn ii req ver s hs RspNoBody logger req s Nothing T.tickle th return isPersist@@ -223,6 +225,7 @@ sendRsp :: Connection -> InternalInfo+ -> Request -> H.HttpVersion -> H.Status -> H.ResponseHeaders@@ -231,7 +234,7 @@ ---------------------------------------------------------------- -sendRsp conn _ ver s hs RspNoBody = do+sendRsp conn _ _ ver s hs RspNoBody = do -- Not adding Content-Length. -- User agents treats it as Content-Length: 0. composeHeader ver s hs >>= connSendAll conn@@ -239,7 +242,7 @@ ---------------------------------------------------------------- -sendRsp conn _ ver s hs (RspBuilder body needsChunked) = do+sendRsp conn _ _ ver s hs (RspBuilder body needsChunked) = do header <- composeHeaderBuilder ver s hs needsChunked let hdrBdy | needsChunked = header <> chunkedTransferEncoding body@@ -252,7 +255,7 @@ ---------------------------------------------------------------- -sendRsp conn _ ver s hs (RspStream streamingBody needsChunked th) = do+sendRsp conn _ _ ver s hs (RspStream streamingBody needsChunked th) = do header <- composeHeaderBuilder ver s hs needsChunked (recv, finish) <- newBlazeRecv $ reuseBufferStrategy $ toBuilderBuffer (connWriteBuffer conn) (connBufferSize conn)@@ -276,7 +279,7 @@ ---------------------------------------------------------------- -sendRsp conn _ _ _ _ (RspRaw withApp src tickle) = do+sendRsp conn _ _ _ _ _ (RspRaw withApp src tickle) = do withApp recv send return (Nothing, Nothing) where@@ -290,9 +293,10 @@ -- Sophisticated WAI applications. -- We respect s0. s0 MUST be a proper value.-sendRsp conn ii ver s0 hs0 (RspFile path (Just part) _ isHead hook) =- sendRspFile2XX conn ii ver s0 hs path beg len isHead hook+sendRsp conn ii req ver s0 hs0 (RspFile path (Just part) _ isHead hook) =+ sendRspFile2XX conn ii req ver s0 hs h path beg len isHead hook where+ h = hash $ rawPathInfo req beg = filePartOffset part len = filePartByteCount part hs = addContentHeadersForFilePart hs0 part@@ -301,33 +305,36 @@ -- Simple WAI applications. -- Status is ignored-sendRsp conn ii ver _ hs0 (RspFile path Nothing idxhdr isHead hook) = do- efinfo <- E.try $ fileInfo ii path+sendRsp conn ii req ver _ hs0 (RspFile path Nothing idxhdr isHead hook) = do+ let h = hash $ rawPathInfo req+ efinfo <- E.try $ fileInfo' ii h path case efinfo of Left (_ex :: E.IOException) -> #ifdef WARP_DEBUG print _ex >> #endif- sendRspFile404 conn ii ver hs0+ sendRspFile404 conn ii req ver hs0 Right finfo -> case conditionalRequest finfo hs0 idxhdr of- WithoutBody s -> sendRsp conn ii ver s hs0 RspNoBody- WithBody s hs beg len -> sendRspFile2XX conn ii ver s hs path beg len isHead hook+ WithoutBody s -> sendRsp conn ii req ver s hs0 RspNoBody+ WithBody s hs beg len -> sendRspFile2XX conn ii req ver s hs h path beg len isHead hook ---------------------------------------------------------------- sendRspFile2XX :: Connection -> InternalInfo+ -> Request -> H.HttpVersion -> H.Status -> H.ResponseHeaders+ -> Hash -> FilePath -> Integer -> Integer -> Bool -> IO () -> IO (Maybe H.Status, Maybe Integer)-sendRspFile2XX conn ii ver s hs path beg len isHead hook- | isHead = sendRsp conn ii ver s hs RspNoBody+sendRspFile2XX conn ii req ver s hs h path beg len isHead hook+ | isHead = sendRsp conn ii req ver s hs RspNoBody | otherwise = do lheader <- composeHeader ver s hs #ifdef WINDOWS@@ -338,7 +345,7 @@ -- settingsFdCacheDuration is 0 Nothing -> return (Nothing, hook) Just fdc -> do- (fd, fresher) <- F.getFd fdc path+ (fd, fresher) <- F.getFd' fdc h path return (Just fd, hook >> fresher) let fid = FileId path mfd #endif@@ -347,10 +354,11 @@ sendRspFile404 :: Connection -> InternalInfo+ -> Request -> H.HttpVersion -> H.ResponseHeaders -> IO (Maybe H.Status, Maybe Integer)-sendRspFile404 conn ii ver hs0 = sendRsp conn ii ver s hs (RspBuilder body True)+sendRspFile404 conn ii req ver hs0 = sendRsp conn ii req ver s hs (RspBuilder body True) where s = H.notFound404 hs = replaceHeader H.hContentType "text/plain; charset=utf-8" hs0
Network/Wai/Handler/Warp/Run.hs view
@@ -25,7 +25,6 @@ import Network.Socket (accept, withSocketsDo, SockAddr(SockAddrInet, SockAddrInet6)) import qualified Network.Socket.ByteString as Sock import Network.Wai-import Network.Wai.HTTP2 (HTTP2Application, promoteApplication) import Network.Wai.Handler.Warp.Buffer import Network.Wai.Handler.Warp.Counter import qualified Network.Wai.Handler.Warp.Date as D@@ -74,23 +73,10 @@ allowInterrupt = unblock $ return () #endif --- Composition over two arguments at once; used for runHTTP2\*.-(.:) :: (c -> d) -> (a -> b -> c) -> a -> b -> d-f .: g = curry $ f . uncurry g- -- | Run an 'Application' on the given port. -- This calls 'runSettings' with 'defaultSettings'. run :: Port -> Application -> IO ()-run port = runServe port . serveDefault---- | Serve an 'HTTP2Application' and an 'Application' together on the given--- port.-runHTTP2 :: Port -> HTTP2Application -> Application -> IO ()-runHTTP2 port = runServe port .: serveHTTP2---- | The generalized form of 'run'.-runServe :: Port -> ServeConnection -> IO ()-runServe p = runServeSettings defaultSettings { settingsPort = p }+run p = runSettings defaultSettings { settingsPort = p } -- | Run an 'Application' on the port present in the @PORT@ -- environment variable. Uses the 'Port' given when the variable is unset.@@ -98,44 +84,28 @@ -- -- Since 3.0.9 runEnv :: Port -> Application -> IO ()-runEnv port = runServeEnv port . serveDefault---- | The HTTP\/2-aware form of 'runEnv'.-runHTTP2Env :: Port -> HTTP2Application -> Application -> IO ()-runHTTP2Env port = runServeEnv port .: serveHTTP2---- | The generalized form of 'runEnv'.-runServeEnv :: Port -> ServeConnection -> IO ()-runServeEnv p serveConn = do+runEnv p app = do mp <- lookup "PORT" <$> getEnvironment - maybe (runServe p serveConn) runReadPort mp+ maybe (run p app) runReadPort mp where runReadPort :: String -> IO () runReadPort sp = case reads sp of- ((p', _):_) -> runServe p' serveConn+ ((p', _):_) -> run p' app _ -> fail $ "Invalid value in $PORT: " ++ sp -- | Run an 'Application' with the given 'Settings'. -- This opens a listen socket on the port defined in 'Settings' and -- calls 'runSettingsSocket'. runSettings :: Settings -> Application -> IO ()-runSettings set = runServeSettings set . serveDefault---- | The HTTP\/2-aware form of 'runSettings'.-runHTTP2Settings :: Settings -> HTTP2Application -> Application -> IO ()-runHTTP2Settings set = runServeSettings set .: serveHTTP2---- | The generalized form of 'runSettings'.-runServeSettings :: Settings -> ServeConnection -> IO ()-runServeSettings set serveConn = withSocketsDo $+runSettings set app = withSocketsDo $ bracket (bindPortTCP (settingsPort set) (settingsHost set)) sClose (\socket -> do setSocketCloseOnExec socket- runServeSettingsSocket set socket serveConn)+ runSettingsSocket set socket app) -- | This installs a shutdown handler for the given socket and -- calls 'runSettingsConnection' with the default connection setup action@@ -149,22 +119,9 @@ -- Note that the 'settingsPort' will still be passed to 'Application's via the -- 'serverPort' record. runSettingsSocket :: Settings -> Socket -> Application -> IO ()-runSettingsSocket set socket = runServeSettingsSocket set socket . serveDefault---- | The HTTP\/2-aware form of 'runSettingsSocket'.-runHTTP2SettingsSocket :: Settings- -> Socket- -> HTTP2Application- -> Application- -> IO ()-runHTTP2SettingsSocket set socket =- runServeSettingsSocket set socket .: serveHTTP2---- | The generalized form of 'runSettingsSocket'.-runServeSettingsSocket :: Settings -> Socket -> ServeConnection -> IO ()-runServeSettingsSocket set socket serveConn = do+runSettingsSocket set socket app = do settingsInstallShutdownHandler set closeListenSocket- runServeSettingsConnection set getConn serveConn+ runSettingsConnection set getConn app where getConn = do #if WINDOWS@@ -188,16 +145,7 @@ -- -- Since 1.3.5 runSettingsConnection :: Settings -> IO (Connection, SockAddr) -> Application -> IO ()-runSettingsConnection set getConn =- runServeSettingsConnection set getConn . serveDefault---- | The generalized form of 'runSettingsConnection'.-runServeSettingsConnection :: Settings- -> IO (Connection, SockAddr)- -> ServeConnection- -> IO ()-runServeSettingsConnection set getConn serveConn =- runServeSettingsConnectionMaker set getConnMaker serveConn+runSettingsConnection set getConn app = runSettingsConnectionMaker set getConnMaker app where getConnMaker = do (conn, sa) <- getConn@@ -206,16 +154,8 @@ -- | This modifies the connection maker so that it returns 'TCP' for 'Transport' -- (i.e. plain HTTP) then calls 'runSettingsConnectionMakerSecure'. runSettingsConnectionMaker :: Settings -> IO (IO Connection, SockAddr) -> Application -> IO ()-runSettingsConnectionMaker set getConnMaker =- runServeSettingsConnectionMaker set getConnMaker . serveDefault---- | The generalized form of 'runSettingsConnectionMaker'.-runServeSettingsConnectionMaker :: Settings- -> IO (IO Connection, SockAddr)- -> ServeConnection- -> IO ()-runServeSettingsConnectionMaker x y =- runServeSettingsConnectionMakerSecure x (toTCP <$> y)+runSettingsConnectionMaker x y =+ runSettingsConnectionMakerSecure x (toTCP <$> y) where toTCP = first ((, TCP) <$>) @@ -228,25 +168,17 @@ -- -- Since 2.1.4 runSettingsConnectionMakerSecure :: Settings -> IO (IO (Connection, Transport), SockAddr) -> Application -> IO ()-runSettingsConnectionMakerSecure set getConnMaker =- runServeSettingsConnectionMakerSecure set getConnMaker . serveDefault---- | The generalized form of 'runSettingsConnectionMakerSecure'.-runServeSettingsConnectionMakerSecure :: Settings- -> IO (IO (Connection, Transport), SockAddr)- -> ServeConnection- -> IO ()-runServeSettingsConnectionMakerSecure set getConnMaker serveConn = do+runSettingsConnectionMakerSecure set getConnMaker app = do settingsBeforeMainLoop set counter <- newCounter- withII $ acceptConnection set getConnMaker serveConn counter+ withII $ acceptConnection set getConnMaker app counter where withII action = D.withDateCache $ \dc -> F.withFdCache fdCacheDurationInSeconds $ \fc ->- I.withFileInfoCache fdFileInfoDurationInSeconds $ \get ->+ I.withFileInfoCache fdFileInfoDurationInSeconds $ \get get' -> withTimeoutManager $ \tm -> do- let ii0 = InternalInfo undefined tm fc get dc -- fixme: undefined+ let ii0 = InternalInfo undefined tm fc get get' dc -- fixme: undefined action ii0 fdCacheDurationInSeconds = settingsFdCacheDuration set * 1000000@@ -273,11 +205,11 @@ -- Our approach is explained in the comments below. acceptConnection :: Settings -> IO (IO (Connection, Transport), SockAddr)- -> ServeConnection+ -> Application -> Counter -> InternalInfo -> IO ()-acceptConnection set getConnMaker serveConn counter ii0 = do+acceptConnection set getConnMaker app counter ii0 = do -- First mask all exceptions in acceptLoop. This is necessary to -- ensure that no async exception is throw between the call to -- acceptNewConnection and the registering of connClose.@@ -299,7 +231,7 @@ case mx of Nothing -> return () Just (mkConn, addr) -> do- fork set mkConn addr serveConn counter ii0+ fork set mkConn addr app counter ii0 acceptLoop acceptNewConnection = do@@ -323,11 +255,11 @@ fork :: Settings -> IO (Connection, Transport) -> SockAddr- -> ServeConnection+ -> Application -> Counter -> InternalInfo -> IO ()-fork set mkConn addr serveConn counter ii0 = settingsFork set $ \ unmask ->+fork set mkConn addr app counter ii0 = settingsFork set $ \ unmask -> -- Run the connection maker to get a new connection, and ensure -- that the connection is closed. If the mkConn call throws an -- exception, we will leak the connection. If the mkConn call is@@ -358,29 +290,21 @@ -- Actually serve this connection. -- bracket with closeConn above ensures the connection is closed.- when goingon $ serveConn conn ii addr transport set+ when goingon $ serveConnection conn ii addr transport set app where closeConn (conn, _transport) = connClose conn onOpen adr = increase counter >> settingsOnOpen set adr onClose adr _ = decrease counter >> settingsOnClose set adr --- The type of a function to serve a fully-prepared connection.-type ServeConnection = Connection- -> InternalInfo- -> SockAddr- -> Transport- -> Settings- -> IO ()---- Serve an HTTP\/2-unaware Application to a connection over any HTTP version.-serveDefault :: Application -> ServeConnection-serveDefault app = serveHTTP2 (promoteApplication app) app---- Serve an HTTP\/2-aware application over HTTP\/2 or a backup 'Application'--- over HTTP\/1.1 or HTTP\/1.-serveHTTP2 :: HTTP2Application -> Application -> ServeConnection-serveHTTP2 app2 app conn ii origAddr transport settings = do+serveConnection :: Connection+ -> InternalInfo+ -> SockAddr+ -> Transport+ -> Settings+ -> Application+ -> IO ()+serveConnection conn ii origAddr transport settings app = do -- fixme: Upgrading to HTTP/2 should be supported. (h2,bs) <- if isHTTP2 transport then return (True, "")@@ -393,7 +317,7 @@ if settingsHTTP2Enabled settings && h2 then do recvN <- makeReceiveN bs (connRecv conn) (connRecvBuf conn) -- fixme: origAddr- http2 conn ii origAddr transport settings recvN app2+ http2 conn ii origAddr transport settings recvN app else do istatus <- newIORef False src <- mkSource (wrappedRecv conn th istatus (settingsSlowlorisSize settings))
Network/Wai/Handler/Warp/Types.hs view
@@ -114,12 +114,15 @@ ---------------------------------------------------------------- +type Hash = Int+ -- | Internal information. data InternalInfo = InternalInfo { threadHandle :: T.Handle , timeoutManager :: T.Manager , fdCacher :: Maybe F.MutableFdCache , fileInfo :: FilePath -> IO I.FileInfo+ , fileInfo' :: Int -> FilePath -> IO I.FileInfo , dateCacher :: D.DateCache }
test/RequestSpec.hs view
@@ -4,8 +4,8 @@ module RequestSpec (main, spec) where +import Network.Wai.Handler.Warp.File (parseByteRanges) import Network.Wai.Handler.Warp.Request-import Network.Wai.Handler.Warp.RequestHeader (parseByteRanges) import Network.Wai.Handler.Warp.Types import Test.Hspec import Test.Hspec.QuickCheck
warp.cabal view
@@ -1,5 +1,5 @@ Name: warp-Version: 3.1.12+Version: 3.2.0 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE@@ -43,17 +43,16 @@ , ghc-prim , http-types >= 0.8.5 , iproute >= 1.3.1- , http2 >= 1.3 && < 1.4+ , http2 >= 1.4 && < 1.5 , simple-sendfile >= 0.2.7 && < 0.3 , unix-compat >= 0.2- , wai >= 3.0.4 && < 3.1+ , wai >= 3.2 && < 3.3 , text , streaming-commons >= 0.1.10 , vault >= 0.3 , stm >= 2.3 , word8 , hashable- , unordered-containers , http-date if flag(network-bytestring) Build-Depends: network >= 2.2.1.5 && < 2.2.3@@ -69,6 +68,7 @@ Network.Wai.Handler.Warp.FdCache Network.Wai.Handler.Warp.File Network.Wai.Handler.Warp.FileInfoCache+ Network.Wai.Handler.Warp.HashMap Network.Wai.Handler.Warp.HTTP2 Network.Wai.Handler.Warp.HTTP2.EncodeFrame Network.Wai.Handler.Warp.HTTP2.HPACK@@ -149,7 +149,7 @@ , simple-sendfile >= 0.2.4 && < 0.3 , transformers >= 0.2.2 , unix-compat >= 0.2- , wai >= 3.0.4 && < 3.1+ , wai >= 3.2 && < 3.3 , network , HUnit , QuickCheck@@ -163,10 +163,9 @@ , directory , process , containers- , http2 >= 1.3+ , http2 >= 1.4 && < 1.5 , word8 , hashable- , unordered-containers , http-date if (os(linux) || os(freebsd) || os(darwin)) && flag(allow-sendfilefd)