warp 0.3.1 → 0.3.2
raw patch · 3 files changed
+186/−49 lines, 3 files
Files
- Network/Wai/Handler/Warp.hs +125/−48
- Timeout.hs +59/−0
- warp.cabal +2/−1
Network/Wai/Handler/Warp.hs view
@@ -27,6 +27,13 @@ run , runEx , serveConnections+ -- * Run a Warp server with full settings control+ , runSettings+ , Settings+ , defaultSettings+ , settingsPort+ , settingsOnException+ , settingsTimeout -- * Datatypes , Port , InvalidRequest (..)@@ -48,15 +55,16 @@ import qualified Data.ByteString.Unsafe as SU import qualified Data.ByteString.Char8 as B import qualified Data.ByteString.Lazy as L-import Network- ( listenOn, sClose, PortID(PortNumber), Socket- , withSocketsDo)+import Network (listenOn, sClose, PortID(PortNumber), Socket) import Network.Socket ( accept, SockAddr ) import qualified Network.Socket.ByteString as Sock-import Control.Exception (bracket, finally, Exception, SomeException, catch)-import Control.Concurrent (forkIO)+import Control.Exception+ ( bracket, finally, Exception, SomeException, catch+ , fromException+ )+import Control.Concurrent (forkIO, threadWaitWrite) import Data.Maybe (fromMaybe) import Data.Typeable (Typeable)@@ -72,18 +80,18 @@ (copyByteString, Builder, toLazyByteString, toByteStringIO) import Blaze.ByteString.Builder.Char8 (fromChar, fromShow) import Data.Monoid (mappend, mconcat)-import Network.Socket.SendFile (sendFile)-import Network.Socket.Enumerator (iterSocket)+import Network.Socket.SendFile (sendFileIterWith, Iter (..)) import Control.Monad.IO.Class (liftIO)-import System.Timeout (timeout)+import qualified Timeout as T import Data.Word (Word8) import Data.List (foldl')+import Control.Monad (forever) #if WINDOWS import Control.Concurrent (threadDelay) import qualified Control.Concurrent.MVar as MV-import Control.Monad (forever)+import Network.Socket (withSocketsDo) #endif -- | Run an 'Application' on the given port, ignoring all exceptions.@@ -93,23 +101,30 @@ -- | Run an 'Application' on the given port, with the given exception handler. -- Please note that you will also receive 'InvalidRequest' exceptions. runEx :: (SomeException -> IO ()) -> Port -> Application -> IO ()+runEx onE port = runSettings Settings+ { settingsPort = port+ , settingsOnException = onE+ , settingsTimeout = 30+ }++runSettings :: Settings -> Application -> IO () #if WINDOWS-runEx onE port app = withSocketsDo $ do+runSettings set app = withSocketsDo $ do var <- MV.newMVar Nothing let clean = MV.modifyMVar_ var $ \s -> maybe (return ()) sClose s >> return Nothing _ <- forkIO $ bracket- (listenOn $ PortNumber $ fromIntegral port)+ (listenOn $ PortNumber $ fromIntegral $ settingsPort set) (const clean) (\s -> do MV.modifyMVar_ var (\_ -> return $ Just s)- serveConnections onE port app s)+ serveConnections' set app s) forever (threadDelay maxBound) `finally` clean #else-runEx onE port = withSocketsDo . -- FIXME should this be called by client user instead?+runSettings set = bracket- (listenOn $ PortNumber $ fromIntegral port)+ (listenOn $ PortNumber $ fromIntegral $ settingsPort set) sClose .- serveConnections onE port+ serveConnections' set #endif type Port = Int@@ -121,25 +136,43 @@ -- non-TCP socket, this can be a ficticious value. serveConnections :: (SomeException -> IO ()) -> Port -> Application -> Socket -> IO ()-serveConnections onE port app socket = do- (conn, sa) <- accept socket- _ <- forkIO $ serveConnection onE port app conn sa- serveConnections onE port app socket+serveConnections onE port = serveConnections' defaultSettings+ { settingsOnException = onE+ , settingsPort = port+ } -serveConnection :: (SomeException -> IO ())+serveConnections' :: Settings+ -> Application -> Socket -> IO ()+serveConnections' set app socket = do+ let onE = settingsOnException set+ port = settingsPort set+ tm <- T.initialize $ settingsTimeout set * 1000000+ forever $ do+ (conn, sa) <- accept socket+ _ <- forkIO $ do+ th <- T.register tm $ sClose conn+ serveConnection th onE port app conn sa+ T.cancel th+ return ()++serveConnection :: T.Handle+ -> (SomeException -> IO ()) -> Port -> Application -> Socket -> SockAddr -> IO ()-serveConnection onException port app conn remoteHost' = do+serveConnection th onException port app conn remoteHost' = do catch (finally (E.run_ $ fromClient $$ serveConnection') (sClose conn)) onException where- fromClient = enumSocket bytesPerRead conn+ fromClient = enumSocket th bytesPerRead conn serveConnection' = do (enumeratee, env) <- parseRequest port remoteHost'+ -- Let the application run for as long as it wants+ liftIO $ T.pause th res <- E.joinI $ enumeratee $$ app env- keepAlive <- liftIO $ sendResponse env (httpVersion env) conn res+ liftIO $ T.resume th+ keepAlive <- liftIO $ sendResponse th env (httpVersion env) conn res if keepAlive then serveConnection' else return () parseRequest :: Port -> SockAddr -> E.Iteratee S.ByteString IO (E.Enumeratee ByteString ByteString IO a, Request)@@ -148,12 +181,14 @@ parseRequest' port headers' remoteHost' -- FIXME come up with good values here-maxHeaders, maxHeaderLength, bytesPerRead, readTimeout :: Int+maxHeaders, maxHeaderLength, bytesPerRead :: Int maxHeaders = 30 maxHeaderLength = 1024 bytesPerRead = 4096-readTimeout = 3000000 +sendFileCount :: Integer+sendFileCount = 65536+ data InvalidRequest = NotEnoughLines [String] | BadFirstLine String@@ -262,23 +297,37 @@ hasBody :: Status -> Request -> Bool hasBody s req = s /= (Status 204 "") && requestMethod req /= "HEAD" -sendResponse :: Request -> HttpVersion -> Socket -> Response -> IO Bool-sendResponse req hv socket (ResponseFile s hs fp) = do+sendResponse :: T.Handle+ -> Request -> HttpVersion -> Socket -> Response -> IO Bool+sendResponse th req hv socket (ResponseFile s hs fp) = do Sock.sendMany socket $ L.toChunks $ toLazyByteString $ headers hv s hs False if hasBody s req then do- sendFile socket fp+ sendFileIterWith tickler socket fp sendFileCount return $ lookup "content-length" hs /= Nothing else return True-sendResponse req hv socket (ResponseBuilder s hs b)+ where+ tickler iter = do+ r <- iter+ case r of+ Done _ -> return ()+ Sent _ cont -> T.tickle th >> tickler cont+ WouldBlock _ fd cont -> do+ -- FIXME do we want to tickle here?+ threadWaitWrite fd+ tickler cont+sendResponse th req hv socket (ResponseBuilder s hs b) | hasBody s req = do- toByteStringIO (Sock.sendAll socket) b'+ toByteStringIO (\bs -> do+ Sock.sendAll socket bs+ T.tickle th) b' return isKeepAlive | otherwise = do Sock.sendMany socket $ L.toChunks $ toLazyByteString $ headers hv s hs False+ T.tickle th return True where headers' = headers hv s hs isChunked'@@ -291,7 +340,7 @@ hasLength = lookup "content-length" hs /= Nothing isChunked' = isChunked hv && not hasLength isKeepAlive = isChunked' || hasLength-sendResponse req hv socket (ResponseEnumerator res) =+sendResponse th req hv socket (ResponseEnumerator res) = res go where -- FIXME perhaps alloca a buffer per thread and reuse that in all functiosn below. Should lessen greatly the GC burden (I hope)@@ -305,7 +354,7 @@ chunk' $ E.enumList 1 [headers hv s hs isChunked'] $$ E.joinI $ builderToByteString -- FIXME unsafeBuilderToByteString- $$ (iterSocket socket >> return isKeepAlive)+ $$ (iterSocket th socket >> return isKeepAlive) where hasLength = lookup "content-length" hs /= Nothing isChunked' = isChunked hv && not hasLength@@ -331,24 +380,20 @@ else rest in (mkCIByteString k, rest') --- FIXME when we switch to Jeremy's timeout code, we can probably start using--- network-enumerator's enumSocket-enumSocket :: Int -> Socket -> E.Enumerator ByteString IO a-enumSocket len socket (E.Continue k) = do-#if NO_TIMEOUT_PROTECTION- bs <- liftIO $ Sock.recv socket len- go bs-#else- mbs <- liftIO $ timeout readTimeout $ Sock.recv socket len- case mbs of- Nothing -> E.throwError SocketTimeout- Just bs -> go bs-#endif+enumSocket :: T.Handle -> Int -> Socket -> E.Enumerator ByteString IO a+enumSocket th len socket =+ inner where- go bs+ inner (E.Continue k) = do+ bs <- liftIO $ Sock.recv socket len+ liftIO $ T.tickle th+ if S.null bs+ then E.throwError SocketTimeout+ else go k bs+ inner step = E.returnI step+ go k bs | S.length bs == 0 = E.continue k- | otherwise = k (E.Chunks [bs]) >>== enumSocket len socket-enumSocket _ _ step = E.returnI step+ | otherwise = k (E.Chunks [bs]) >>== enumSocket th len socket ------ The functions below are not warp-specific and could be split out into a --separate package.@@ -387,3 +432,35 @@ | otherwise -> do E.yield () $ E.Chunks [B.drop 1 y] return $ B.concat $ front [x']++iterSocket :: T.Handle+ -> Socket+ -> E.Iteratee B.ByteString IO ()+iterSocket th sock =+ E.continue step+ where+ step E.EOF = E.yield () E.EOF+ step (E.Chunks []) = E.continue step+ step (E.Chunks xs) = do+ liftIO $ Sock.sendMany sock xs+ liftIO $ T.tickle th+ E.continue step++data Settings = Settings+ { settingsPort :: Int+ , settingsOnException :: SomeException -> IO ()+ , settingsTimeout :: Int -- ^ seconds+ }++defaultSettings :: Settings+defaultSettings = Settings+ { settingsPort = 3000+ , settingsOnException = \e ->+ case fromException e of+ Just x -> go x+ Nothing -> print e+ , settingsTimeout = 30+ }+ where+ go :: InvalidRequest -> IO ()+ go _ = return ()
+ Timeout.hs view
@@ -0,0 +1,59 @@+module Timeout+ ( Manager+ , Handle+ , initialize+ , register+ , tickle+ , pause+ , resume+ , cancel+ ) where++import qualified Data.IORef as I+import Control.Concurrent (forkIO, threadDelay)+import Control.Monad (forever)+import qualified Control.Exception as E++-- FIXME implement stopManager++newtype Manager = Manager (I.IORef [Handle])+data Handle = Handle (IO ()) (I.IORef State)+data State = Active | Inactive | Paused | Canceled++initialize :: Int -> IO Manager+initialize timeout = do+ ref <- I.newIORef []+ _ <- forkIO $ forever $ do+ threadDelay timeout+ ms <- I.atomicModifyIORef ref (\x -> ([], x))+ ms' <- go ms id+ I.atomicModifyIORef ref (\x -> (ms' x, ()))+ return $ Manager ref+ where+ go [] front = return front+ go (m@(Handle onTimeout iactive):rest) front = do+ state <- I.atomicModifyIORef iactive (\x -> (go' x, x))+ case state of+ Inactive -> do+ onTimeout `E.catch` ignoreAll+ go rest front+ Canceled -> go rest front+ _ -> go rest (front . (:) m)+ go' Active = Inactive+ go' x = x++ignoreAll :: E.SomeException -> IO ()+ignoreAll _ = return ()++register :: Manager -> IO () -> IO Handle+register (Manager ref) onTimeout = do+ iactive <- I.newIORef Active+ let h = Handle onTimeout iactive+ I.atomicModifyIORef ref (\x -> (h : x, ()))+ return h++tickle, pause, resume, cancel :: Handle -> IO ()+tickle (Handle _ iactive) = I.writeIORef iactive Active+pause (Handle _ iactive) = I.writeIORef iactive Paused+resume = tickle+cancel (Handle _ iactive) = I.writeIORef iactive Canceled
warp.cabal view
@@ -1,5 +1,5 @@ Name: warp-Version: 0.3.1+Version: 0.3.2 Synopsis: A fast, light-weight web server for WAI applications. License: BSD3 License-file: LICENSE@@ -32,6 +32,7 @@ else build-depends: network >= 2.3 && < 2.4 Exposed-modules: Network.Wai.Handler.Warp+ Other-modules: Timeout ghc-options: -Wall if !flag(timeout-protection) Cpp-options: -DNO_TIMEOUT_PROTECTION