diff --git a/Network/Wai/Handler/Warp.hs b/Network/Wai/Handler/Warp.hs --- a/Network/Wai/Handler/Warp.hs +++ b/Network/Wai/Handler/Warp.hs @@ -44,27 +44,32 @@ -- * Datatypes , Port , InvalidRequest (..) - -- * Internal (Manager) + -- * Internal + -- ** Time out manager , Manager , Handle + , TimeoutAction , initialize , withManager , register , registerKillThread - , pause - , resume + , tickle , cancel - -- * Internal + , resume + , pause + -- ** Cleaner + , Cleaner + , dummyCleaner + -- ** Request and response , parseRequest , sendResponse - , dummyCleaner , socketConnection #if TEST , takeHeaders , parseFirst , readInt #endif - -- * Misc + -- ** Version , warpVersion ) where diff --git a/Network/Wai/Handler/Warp/FdCache.hs b/Network/Wai/Handler/Warp/FdCache.hs --- a/Network/Wai/Handler/Warp/FdCache.hs +++ b/Network/Wai/Handler/Warp/FdCache.hs @@ -7,14 +7,13 @@ ) where import Control.Applicative ((<$>), (<*>)) -import Control.Concurrent +import Control.Concurrent (forkIO, threadDelay, ThreadId, killThread) import Control.Exception (mask_) -import Control.Monad -import Data.Hashable -import Data.IORef +import Data.Hashable (hash) +import Data.IORef (IORef, newIORef, readIORef, writeIORef, atomicModifyIORef, mkWeakIORef) import Network.Wai.Handler.Warp.MultiMap -import System.Posix.IO -import System.Posix.Types +import System.Posix.IO (openFd, defaultFileFlags, OpenMode(ReadOnly), closeFd) +import System.Posix.Types (Fd) ---------------------------------------------------------------- @@ -81,6 +80,9 @@ initialize duration = do mfc <- newMutableFdCache tid <- forkIO $ loop mfc + -- Registering finalizer to this IORef. + -- When Warp is finished in GHCi, this IORef is GCed. + -- At that time, we should close all opened file descriptors. _ <- mkWeakIORef (unMutableFdCache mfc) $ terminate mfc tid return mfc where diff --git a/Network/Wai/Handler/Warp/Request.hs b/Network/Wai/Handler/Warp/Request.hs --- a/Network/Wai/Handler/Warp/Request.hs +++ b/Network/Wai/Handler/Warp/Request.hs @@ -16,7 +16,6 @@ import qualified Data.IORef as I import Data.Maybe (fromMaybe) import Data.Monoid (mempty) -import Data.Void (Void) import Data.Word (Word8) import qualified Network.HTTP.Types as H import Network.Socket (SockAddr) diff --git a/Network/Wai/Handler/Warp/Run.hs b/Network/Wai/Handler/Warp/Run.hs --- a/Network/Wai/Handler/Warp/Run.hs +++ b/Network/Wai/Handler/Warp/Run.hs @@ -1,10 +1,11 @@ {-# LANGUAGE CPP #-} +{-# LANGUAGE OverloadedStrings #-} {-# LANGUAGE ScopedTypeVariables #-} module Network.Wai.Handler.Warp.Run where import Control.Concurrent (threadDelay, forkIOWithUnmask) -import Control.Exception +import Control.Exception as E import Control.Monad (forever, when, unless, void) import Control.Monad.IO.Class (MonadIO, liftIO) import Control.Monad.Trans.Class (lift) @@ -18,13 +19,14 @@ import Network.Sendfile import Network.Socket (accept, SockAddr) import qualified Network.Socket.ByteString as Sock +import Data.Monoid (mempty) +import qualified Network.HTTP.Types as H import Network.Wai import Network.Wai.Handler.Warp.Request import Network.Wai.Handler.Warp.Response import Network.Wai.Handler.Warp.Settings import qualified Network.Wai.Handler.Warp.Timeout as T import Network.Wai.Handler.Warp.Types -import Prelude hiding (catch) -- Sock.recv first tries to call recvfrom() optimistically. -- If EAGAIN returns, it polls incoming data with epoll/kqueue. @@ -115,7 +117,7 @@ where getter = do (conn, sa) <- accept socket - setSocketCloseOnExec socket + setSocketCloseOnExec conn return (socketConnection conn, sa) runSettingsConnection :: Settings -> IO (Connection, SockAddr) -> Application -> IO () @@ -209,7 +211,7 @@ serveConnection th set cleaner port app conn addr where -- FIXME: only IOEception is caught. What about other exceptions? - getConnLoop = getConn `catch` \(e :: IOException) -> do + getConnLoop = getConn `E.catch` \(e :: IOException) -> do onE (toException e) -- "resource exhausted (Too many open files)" may happen by accept(). -- Wait a second hoping that resource will be available. @@ -233,7 +235,8 @@ -> Cleaner -> Port -> Application -> Connection -> SockAddr-> IO () serveConnection timeoutHandle settings cleaner port app conn remoteHost' = - runResourceT serveConnection' + respondOnException settings cleaner conn remoteHost' $ + runResourceT serveConnection' where innerRunResourceT | settingsResourceTPerRequest settings = lift . runResourceT @@ -264,6 +267,14 @@ liftIO $ T.pause th ResumableSource fromClient' _ <- liftIO getSource intercept fromClient' conn + +respondOnException :: Settings -> Cleaner -> Connection -> SockAddr -> IO () -> IO () +respondOnException settings cleaner conn remoteHost' io = io `E.catch` \e@(SomeException _) -> do + _ <- runResourceT $ sendResponse settings cleaner blankRequest conn internalError + throwIO e + where + blankRequest = Request H.methodGet H.http10 mempty mempty mempty 0 [] False remoteHost' [] [] (return mempty) mempty (KnownLength 0) + internalError = responseLBS H.internalServerError500 [(H.hContentType, "text/plain")] "Something went wrong" connSource :: Connection -> T.Handle -> Source (ResourceT IO) ByteString connSource Connection { connRecv = recv } th = src diff --git a/Network/Wai/Handler/Warp/Timeout.hs b/Network/Wai/Handler/Warp/Timeout.hs --- a/Network/Wai/Handler/Warp/Timeout.hs +++ b/Network/Wai/Handler/Warp/Timeout.hs @@ -5,6 +5,7 @@ module Network.Wai.Handler.Warp.Timeout ( Manager , Handle + , TimeoutAction , initialize , stopManager , register @@ -17,88 +18,118 @@ , dummyHandle ) where -import System.Mem.Weak (deRefWeak) #if MIN_VERSION_base(4,6,0) -import Control.Concurrent (mkWeakThreadId) +import Control.Concurrent (mkWeakThreadId, ThreadId) #else -import GHC.Weak (Weak (..)) -import GHC.Conc.Sync (ThreadId (..)) -import GHC.IO (IO (IO)) +import Control.Concurrent (ThreadId(..)) import GHC.Exts (mkWeak#) +import GHC.IO (IO (IO)) #endif +import GHC.Weak (Weak (..)) import Control.Concurrent (forkIO, threadDelay, myThreadId, killThread) import qualified Control.Exception as E import Control.Monad (forever, void) +import Data.IORef (IORef) import qualified Data.IORef as I -import System.IO.Unsafe (unsafePerformIO) import Data.Typeable (Typeable) +import System.IO.Unsafe (unsafePerformIO) +import System.Mem.Weak (deRefWeak) +---------------------------------------------------------------- + -- | A timeout manager -newtype Manager = Manager (I.IORef [Handle]) +newtype Manager = Manager (IORef [Handle]) +-- | An action to be performed on timeout. +type TimeoutAction = IO () + -- | A handle used by 'Manager' --- --- First field is action to be performed on timeout. -data Handle = Handle (IO ()) (I.IORef State) +data Handle = Handle TimeoutAction (IORef State) +data State = Active -- Manager turns it to Inactive. + | Inactive -- Manager removes it with timeout action. + | Paused -- Manager does not change it. + | Canceled -- Manager removes it without timeout action. + +---------------------------------------------------------------- + -- | A dummy @Handle@. dummyHandle :: Handle dummyHandle = Handle (return ()) (unsafePerformIO $ I.newIORef Active) -data State = Active | Inactive | Paused | Canceled +---------------------------------------------------------------- +data TimeoutManagerStopped = TimeoutManagerStopped + deriving (Show, Typeable) +instance E.Exception TimeoutManagerStopped + +---------------------------------------------------------------- + +-- | Creating timeout manager which works every N micro seconds +-- where N is the first argument. initialize :: Int -> IO Manager initialize timeout = do ref <- I.newIORef [] void . forkIO $ E.handle ignoreStop $ forever $ do threadDelay timeout - ms <- I.atomicModifyIORef ref (\x -> ([], x)) - ms' <- go ms id - I.atomicModifyIORef ref (\x -> (ms' x, ())) + -- FIXME: isn't mask_ necessary? + old <- I.atomicModifyIORef ref (\x -> ([], x)) + merge <- prune old id + I.atomicModifyIORef ref (\new -> (merge new, ())) return $ Manager ref where ignoreStop TimeoutManagerStopped = return () - go [] front = return front - go (m@(Handle onTimeout iactive):rest) front = do - state <- I.atomicModifyIORef iactive (\x -> (go' x, x)) + prune [] front = return front + prune (m@(Handle onTimeout iactive):rest) front = do + state <- I.atomicModifyIORef iactive (\x -> (inactivate 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 + prune rest front + Canceled -> prune rest front + _ -> prune rest (front . (:) m) + inactivate Active = Inactive + inactivate x = x -data TimeoutManagerStopped = TimeoutManagerStopped - deriving (Show, Typeable) -instance E.Exception TimeoutManagerStopped +---------------------------------------------------------------- stopManager :: Manager -> IO () stopManager (Manager ihandles) = E.mask_ $ do -- Put an undefined value in the IORef to kill the worker thread (yes, it's -- a bit of a hack) !handles <- I.atomicModifyIORef ihandles $ \h -> (E.throw TimeoutManagerStopped, h) - mapM_ go handles + mapM_ fire handles where - go (Handle onTimeout _) = onTimeout `E.catch` ignoreAll + fire (Handle onTimeout _) = onTimeout `E.catch` ignoreAll ignoreAll :: E.SomeException -> IO () ignoreAll _ = return () -register :: Manager -> IO () -> IO Handle +---------------------------------------------------------------- + +-- | Registering a timeout action. +register :: Manager -> TimeoutAction -> IO Handle register (Manager ref) onTimeout = do iactive <- I.newIORef Active let h = Handle onTimeout iactive I.atomicModifyIORef ref (\x -> (h : x, ())) return h +-- | Registering a timeout action of killing this thread. registerKillThread :: Manager -> IO Handle registerKillThread m = do wtid <- myThreadId >>= mkWeakThreadId - register m $ deRefWeak wtid >>= maybe (return ()) killThread + register m $ killIfExist wtid +-- If ThreadId is hold referred by a strong reference, +-- it leaks even after the thread is killed. +-- So, let's use a weak reference so that CG can throw ThreadId away. +-- deRefWeak checks if ThreadId referenced by the weak reference +-- exists. If exists, it means that the thread is alive. +killIfExist :: Weak ThreadId -> TimeoutAction +killIfExist wtid = deRefWeak wtid >>= maybe (return ()) killThread + #if !MIN_VERSION_base(4,6,0) mkWeakThreadId :: ThreadId -> IO (Weak ThreadId) mkWeakThreadId t@(ThreadId t#) = IO $ \s -> @@ -106,11 +137,29 @@ (# s1, w #) -> (# s1, Weak w #) #endif -tickle, pause, resume, cancel :: Handle -> IO () +---------------------------------------------------------------- + +-- | Setting the state to active. +-- 'Manager' turns active to inactive repeatedly. +tickle :: Handle -> IO () tickle (Handle _ iactive) = I.writeIORef iactive Active + +-- | Setting the state to canceled. +-- 'Manager' eventually removes this without timeout action. +cancel :: Handle -> IO () +cancel (Handle _ iactive) = I.writeIORef iactive Canceled + +-- | Setting the state to paused. +-- 'Manager' does not change the value. +pause :: Handle -> IO () pause (Handle _ iactive) = I.writeIORef iactive Paused + +-- | Setting the state to active. +-- This is an alias to 'ticle'. +resume :: Handle -> IO () resume = tickle -cancel (Handle _ iactive) = I.writeIORef iactive Canceled + +---------------------------------------------------------------- -- | Call the inner function with a timeout manager. withManager :: Int -- ^ timeout in microseconds diff --git a/Network/Wai/Handler/Warp/Types.hs b/Network/Wai/Handler/Warp/Types.hs --- a/Network/Wai/Handler/Warp/Types.hs +++ b/Network/Wai/Handler/Warp/Types.hs @@ -87,6 +87,7 @@ #if SENDFILEFD dummyCleaner = Cleaner T.dummyHandle Nothing +-- | A type used to clean up file descriptor caches. data Cleaner = Cleaner { threadHandle :: T.Handle , fdCacher :: Maybe F.MutableFdCache diff --git a/warp.cabal b/warp.cabal --- a/warp.cabal +++ b/warp.cabal @@ -1,5 +1,5 @@ Name: warp -Version: 1.3.9.2 +Version: 1.3.10 Synopsis: A fast, light-weight web server for WAI applications. License: MIT License-file: LICENSE @@ -84,6 +84,7 @@ , case-insensitive >= 0.2 , conduit >= 0.5 , ghc-prim + , HTTP , http-types >= 0.7 , lifted-base >= 0.1 , network-conduit