diff --git a/snap-server.cabal b/snap-server.cabal --- a/snap-server.cabal +++ b/snap-server.cabal @@ -1,5 +1,5 @@ name: snap-server -version: 0.2.14.1 +version: 0.2.15 synopsis: A fast, iteratee-based, epoll-enabled web server for the Snap Framework description: This is the first developer prerelease of the Snap framework. Snap is a @@ -122,7 +122,7 @@ murmur-hash >= 0.1 && < 0.2, network == 2.2.1.*, old-locale, - snap-core == 0.2.14, + snap-core == 0.2.15, template-haskell, time, transformers, @@ -139,6 +139,7 @@ other-modules: Snap.Internal.Http.Server.LibevBackend cpp-options: -DLIBEV else + other-modules: Snap.Internal.Http.Server.TimeoutTable build-depends: network-bytestring >= 0.1.2 && < 0.2, PSQueue >= 1.1 && <1.2 diff --git a/src/Data/Concurrent/HashMap.hs b/src/Data/Concurrent/HashMap.hs --- a/src/Data/Concurrent/HashMap.hs +++ b/src/Data/Concurrent/HashMap.hs @@ -17,7 +17,8 @@ , toList , hashString , hashBS - , hashInt ) where + , hashInt + , nextHighestPowerOf2 ) where ------------------------------------------------------------------------------ diff --git a/src/Snap/Internal/Http/Server/SimpleBackend.hs b/src/Snap/Internal/Http/Server/SimpleBackend.hs --- a/src/Snap/Internal/Http/Server/SimpleBackend.hs +++ b/src/Snap/Internal/Http/Server/SimpleBackend.hs @@ -1,11 +1,11 @@ -{-# LANGUAGE BangPatterns #-} -{-# LANGUAGE CPP #-} -{-# LANGUAGE DeriveDataTypeable #-} +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE CPP #-} +{-# LANGUAGE DeriveDataTypeable #-} {-# LANGUAGE ForeignFunctionInterface #-} -{-# LANGUAGE OverloadedStrings #-} -{-# LANGUAGE RankNTypes #-} -{-# LANGUAGE ScopedTypeVariables #-} -{-# LANGUAGE PackageImports #-} +{-# LANGUAGE OverloadedStrings #-} +{-# LANGUAGE PackageImports #-} +{-# LANGUAGE RankNTypes #-} +{-# LANGUAGE ScopedTypeVariables #-} module Snap.Internal.Http.Server.SimpleBackend ( Backend @@ -37,23 +37,20 @@ import Data.ByteString (ByteString) import Data.ByteString.Internal (c2w, w2c) import qualified Data.ByteString as B -import Data.DList (DList) -import qualified Data.DList as D -import Data.IORef import Data.Iteratee.WrappedByteString -import Data.List (foldl') -import qualified Data.PSQueue as PSQ -import Data.PSQueue (PSQ) import Data.Typeable +import Data.Word import Foreign hiding (new) -import Foreign.C.Types (CTime) import GHC.Conc (labelThread, forkOnIO) import Network.Socket import qualified Network.Socket.ByteString as SB import Prelude hiding (catch) ------------------------------------------------------------------------------ +import Data.Concurrent.HashMap (hashString) import Snap.Internal.Debug import Snap.Internal.Http.Server.Date +import qualified Snap.Internal.Http.Server.TimeoutTable as TT +import Snap.Internal.Http.Server.TimeoutTable (TimeoutTable) import Snap.Iteratee hiding (foldl') #if defined(HAS_SENDFILE) @@ -71,27 +68,30 @@ instance Exception BackendTerminatedException -type TimeoutTable = PSQ ThreadId CTime +------------------------------------------------------------------------------ type QueueElem = Maybe (Socket,SockAddr) data Backend = Backend { _acceptSocket :: !Socket , _acceptThread :: !ThreadId - , _timeoutEdits :: !(IORef (DList (TimeoutTable -> TimeoutTable))) + , _timeoutTable :: TimeoutTable , _timeoutThread :: !(MVar ThreadId) , _connectionQueue :: !(Chan QueueElem) } -data Connection = Connection +data Connection = Connection { _backend :: Backend , _socket :: Socket , _remoteAddr :: ByteString , _remotePort :: Int , _localAddr :: ByteString , _localPort :: Int - , _connTid :: MVar ThreadId } + , _connTid :: MVar ThreadId + , _threadHash :: MVar Word + } + {-# INLINE name #-} name :: ByteString name = "simple" @@ -157,13 +157,12 @@ new sock cpu = do debug $ "Backend.new: listening" - ed <- newIORef D.empty + tt <- TT.new t <- newEmptyMVar - connq <- newChan accThread <- forkOnIO cpu $ acceptThread sock connq - let b = Backend sock accThread ed t connq + let b = Backend sock accThread tt t connq tid <- forkIO $ timeoutThread b putMVar t tid @@ -173,59 +172,29 @@ timeoutThread :: Backend -> IO () timeoutThread backend = do - tref <- newIORef $ PSQ.empty - let loop = do - killTooOld tref - threadDelay (5000000) - loop - - loop `catch` (\(_::SomeException) -> killAll tref) + loop `catch` (\(_::SomeException) -> killAll) where - applyEdits table = do - edits <- atomicModifyIORef tedits $ \t -> (D.empty, D.toList t) - return $ foldl' (flip ($)) table edits + table = _timeoutTable backend - killTooOld tref = do - !table <- readIORef tref - -- atomic swap edit list - now <- getCurrentDateTime - table' <- applyEdits table - !t' <- killOlderThan now table' - writeIORef tref t' + loop = do + debug "timeoutThread: waiting for activity on thread table" + TT.waitForActivity table + debug "timeoutThread: woke up, killing old connections" + killTooOld + loop + killTooOld = do + now <- getCurrentDateTime + TT.killOlderThan (now - tIMEOUT) table + -- timeout = 30 seconds tIMEOUT = 30 - killAll !tref = do + killAll = do debug "Backend.timeoutThread: shutdown, killing all connections" - !table <- readIORef tref - !table' <- applyEdits table - go table' - where - go !t = maybe (return ()) - (\m -> (killThread $ PSQ.key m) >> - (go $ PSQ.deleteMin t)) - (PSQ.findMin t) - - killOlderThan now !table = do - debug "Backend.timeoutThread: killing old connections" - let mmin = PSQ.findMin table - maybe (return table) - (\m -> do - debug $ "Backend.timeoutThread: minimum value " - ++ show (PSQ.prio m) ++ ", cutoff=" - ++ show (now - tIMEOUT) - - if now - PSQ.prio m >= tIMEOUT - then do - killThread $ PSQ.key m - killOlderThan now $ PSQ.deleteMin table - else return table) - mmin - - tedits = _timeoutEdits backend + TT.killAll table stop :: Backend -> IO () @@ -286,26 +255,29 @@ return (fromIntegral p, B.pack $ map c2w h') x -> throwIO $ AddressNotSupportedException $ show x - tmvar <- newEmptyMVar + tmvar <- newEmptyMVar + thrhash <- newEmptyMVar - let c = Connection backend sock host port lhost lport tmvar + let c = Connection backend sock host port lhost lport tmvar thrhash tid <- forkOnIO cpu $ do labelMe $ "connHndl " ++ show fd bracket (return c) (\_ -> block $ do debug "thread killed, closing socket" - thr <- readMVar tmvar + thr <- readMVar tmvar + thash <- readMVar thrhash -- remove thread from timeout table - atomicModifyIORef (_timeoutEdits backend) $ - \es -> (D.snoc es (PSQ.delete thr), ()) + TT.delete thash thr $ _timeoutTable backend + eatException $ shutdown sock ShutdownBoth eatException $ sClose sock ) proc putMVar tmvar tid + putMVar thrhash $ hashString $ show tid tickleTimeout c return () @@ -364,24 +336,27 @@ tickleTimeout :: Connection -> IO () tickleTimeout conn = do debug "Backend.tickleTimeout" - now <- getCurrentDateTime - tid <- readMVar $ _connTid conn + now <- getCurrentDateTime + tid <- readMVar $ _connTid conn + thash <- readMVar $ _threadHash conn - atomicModifyIORef tedits $ \es -> (D.snoc es (PSQ.insert tid now), ()) + TT.insert thash tid now table where - tedits = _timeoutEdits $ _backend conn + table = _timeoutTable $ _backend conn _cancelTimeout :: Connection -> IO () _cancelTimeout conn = do debug "Backend.cancelTimeout" - tid <- readMVar $ _connTid conn - atomicModifyIORef tedits $ \es -> (D.snoc es (PSQ.delete tid), ()) + tid <- readMVar $ _connTid conn + thash <- readMVar $ _threadHash conn + TT.delete thash tid table + where - tedits = _timeoutEdits $ _backend conn + table = _timeoutTable $ _backend conn timeoutRecv :: Connection -> Int -> IO ByteString diff --git a/src/Snap/Internal/Http/Server/TimeoutTable.hs b/src/Snap/Internal/Http/Server/TimeoutTable.hs new file mode 100644 --- /dev/null +++ b/src/Snap/Internal/Http/Server/TimeoutTable.hs @@ -0,0 +1,127 @@ +{-# LANGUAGE BangPatterns #-} +{-# LANGUAGE TemplateHaskell #-} + +module Snap.Internal.Http.Server.TimeoutTable + ( TimeoutTable + , new + , null + , insert + , delete + , killAll + , killOlderThan + , waitForActivity + ) +where + +------------------------------------------------------------------------------ +import Control.Concurrent +import Control.Monad +import Data.Bits +import qualified Data.PSQueue as PSQ +import Data.PSQueue (PSQ) +import qualified Data.Vector as V +import Data.Vector (Vector) +import Data.Word +import Foreign.C.Types (CTime) +import GHC.Conc (numCapabilities) +import Prelude hiding (null) +------------------------------------------------------------------------------ +import Data.Concurrent.HashMap (nextHighestPowerOf2) + + +type TT = PSQ ThreadId CTime + + +data TimeoutTable = TimeoutTable { + _maps :: !(Vector (MVar TT)) + , _activity :: !(MVar ()) +} + + +defaultNumberOfLocks :: Word +defaultNumberOfLocks = nextHighestPowerOf2 $ toEnum $ 8 * numCapabilities + + +hashToBucket :: Word -> Word +hashToBucket x = x .&. (defaultNumberOfLocks-1) + + +new :: IO TimeoutTable +new = do + vector <- V.replicateM (fromEnum defaultNumberOfLocks) (newMVar PSQ.empty) + act <- newEmptyMVar + return $ TimeoutTable vector act + + +null :: TimeoutTable -> IO Bool +null (TimeoutTable maps _) = do + nulls <- V.mapM (\mv -> withMVar mv $ return . PSQ.null) maps + return $ V.and nulls + + +insert :: Word -> ThreadId -> CTime -> TimeoutTable -> IO () +insert thash tid time (TimeoutTable maps act) = do + modifyMVar_ psqMv $ \psq -> do + let !psq' = PSQ.insert tid time psq + return $! psq' + + tryPutMVar act () + return () + + where + bucket = hashToBucket thash + psqMv = V.unsafeIndex maps $ fromEnum bucket + + +delete :: Word -> ThreadId -> TimeoutTable -> IO () +delete thash tid (TimeoutTable maps act) = do + modifyMVar_ psqMv $ \psq -> do + let !psq' = PSQ.delete tid psq + return $! psq' + + tryPutMVar act () + return () + + where + bucket = hashToBucket thash + psqMv = V.unsafeIndex maps $ fromEnum bucket + + +killAll :: TimeoutTable -> IO () +killAll (TimeoutTable maps _) = do + V.mapM_ k maps + + where + k psqMV = modifyMVar_ psqMV $ \psq -> do + mapM_ killThread $ PSQ.keys psq + return PSQ.empty + + +killOlderThan :: CTime -> TimeoutTable -> IO () +killOlderThan time (TimeoutTable maps _) = do + V.mapM_ processPSQ maps + + where + processPSQ psqMV = modifyMVar_ psqMV $ \psq -> do + let (psq', threads) = findOlder psq [] + mapM_ killThread threads + return psq' + + findOlder psq l = + let mmin = PSQ.findMin psq + in maybe (psq,l) + (\m -> if PSQ.prio m <= time + then findOlder (PSQ.deleteMin psq) ((PSQ.key m):l) + else (psq,l)) + mmin + + +waitForActivity :: TimeoutTable -> IO () +waitForActivity t@(TimeoutTable _ act) = do + takeMVar act + b <- null t + + -- if the table is not empty, put the activity mvar back + unless b $ (tryPutMVar act () >> return ()) + + threadDelay 2500000