c10k 0.2.0 → 0.5.0
raw patch · 3 files changed
Files
- Network/C10kServer.hs +148/−60
- Network/TCPInfo.hs +44/−0
- c10k.cabal +10/−6
Network/C10kServer.hs view
@@ -1,14 +1,14 @@ {-| Network server library to handle over 10,000 connections. Since- GHC 6.10.4 or earlier uses select(), it cannot handle connections over+ GHC 6.12 or earlier uses select(), it cannot handle connections over 1,024. This library uses the \"prefork\" technique to get over the barrier. Each process handles 'threadNumberPerProcess' connections. 'preforkProcessNumber' child server processes are preforked. So, this server can handle 'preforkProcessNumber' * 'threadNumberPerProcess' connections. - Even if GHC supports kqueue or epoll(), it is difficult for RTS- to balance over multi-cores. So, this library can be used to+ GHC 7 supports kqueue or epoll() but it is difficult+ to balance over multi-core. So, this library can be used to make a process for each core and to set limitation of the number to accept connections. @@ -16,19 +16,27 @@ (e.g. @kill `cat PIDFILE`@ where the PID file name is specified by 'pidFile') -}-module Network.C10kServer (C10kServer, C10kConfig(..),- runC10kServer) where+module Network.C10kServer (+ C10kConfig(..)+ , C10kServer, runC10kServer+ , C10kServerH, runC10kServerH+ ) where import Control.Concurrent import Control.Exception import Control.Monad-import Network hiding (accept)+import Data.IORef+import Network.BSD import Network.Socket+import qualified Network.TCPInfo as T (accept)+import Network.TCPInfo hiding (accept) import Prelude hiding (catch)+import System.Exit+import System.IO import System.Posix.Process import System.Posix.Signals+import System.Posix.Types import System.Posix.User-import System.Exit ---------------------------------------------------------------- @@ -38,6 +46,11 @@ type C10kServer = Socket -> IO () {-|+ The type of the first argument of 'runC10kServerH'.+-}+type C10kServerH = Handle -> TCPInfo -> IO ()++{-| The type of configuration given to 'runC10kServer' as the second argument. -}@@ -63,6 +76,8 @@ , threadNumberPerProcess :: Int -- | A port name. e.g. \"http\" or \"80\" , portName :: ServiceName+ -- | A numeric IP address. e.g. \"127.0.0.1\"+ , ipAddr :: Maybe HostName -- | A file where the process ID of the parent process is written. , pidFile :: FilePath -- | A user name. When the program linked with this library runs@@ -81,83 +96,135 @@ Run 'C10kServer' with 'C10kConfig'. -} runC10kServer :: C10kServer -> C10kConfig -> IO ()-runC10kServer srv cnf = do+runC10kServer srv cnf = runC10kServer' (dispatchS srv) cnf `catch` errorHandle cnf++{-|+ Run 'C10kServerH' with 'C10kConfig'.+-}+runC10kServerH :: C10kServerH -> C10kConfig -> IO ()+runC10kServerH srv cnf = runC10kServer' (dispatchH srv) cnf `catch` errorHandle cnf++errorHandle :: C10kConfig -> SomeException -> IO ()+errorHandle cnf e = do+ exitHook cnf (show e)+ exitFailure++----------------------------------------------------------------++runC10kServer' :: (Socket -> Dispatch) -> C10kConfig -> IO ()+runC10kServer' sDispatch cnf = do initHook cnf `catch` ignore- initServer srv cnf `catch` errorHandle+ if onlyOne+ then stay sDispatch cnf+ else prefork sDispatch cnf+ where+ onlyOne = preforkProcessNumber cnf == 1++----------------------------------------------------------------++stay :: (Socket -> Dispatch) -> C10kConfig -> IO ()+stay sDispatch cnf = do parentStartedHook cnf `catch` ignore- doNothing+ s <- listenTo addr port+ writePidFile cnf+ setGroupUser cnf+ runServer (sDispatch s) cnf+ sClose s where- errorHandle :: SomeException -> IO ()- errorHandle e = do- exitHook cnf (show e)- exitFailure- doNothing = do- threadDelay $ 5 * microseconds- doNothing+ port = portName cnf+ addr = ipAddr cnf ---------------------------------------------------------------- -initServer :: C10kServer -> C10kConfig -> IO ()-initServer srv cnf = do- let port = Service $ portName cnf- n = preforkProcessNumber cnf- pidf = pidFile cnf- s <- listenOn port- setGroupUser- preFork s n srv cnf+prefork :: (Socket -> Dispatch) -> C10kConfig -> IO ()+prefork sDispatch cnf = do+ cids <- doPrefork sDispatch cnf+ parentStartedHook cnf `catch` ignore+ handleSignal cids+ pause+ terminateChildren cids+ where+ handleSignal cids = do+ ignoreSigChild+ mapM_ (terminator cids) [sigTERM,sigINT]+ pause = blockSignals reservedSignals >> awaitSignal Nothing >> yield+ initHandler func sig = installHandler sig func Nothing+ ignoreSigChild = initHandler Ignore sigCHLD+ terminator cids = initHandler (Catch (terminateChildren cids))+ terminateChildren cids = do+ ignoreSigChild+ mapM_ terminateChild cids+ terminateChild cid = signalProcess sigTERM cid `catch` ignore++----------------------------------------------------------------++doPrefork :: (Socket -> Dispatch) -> C10kConfig -> IO [ProcessID]+doPrefork sDispatch cnf = do+ s <- listenTo addr port+ writePidFile cnf+ setGroupUser cnf+ cids <- fork (sDispatch s) sClose s- writePidFile pidf+ return cids where- writePidFile pidf = do- pid <- getProcessID- writeFile pidf $ show pid ++ "\n"- setGroupUser = do- uid <- getRealUserID- when (uid == 0) $ do+ port = portName cnf+ addr = ipAddr cnf+ n = preforkProcessNumber cnf+ fork dispatch = replicateM n . forkProcess $ runServer dispatch cnf++----------------------------------------------------------------++setGroupUser :: C10kConfig -> IO ()+setGroupUser cnf = do+ uid <- getRealUserID+ when (uid == 0) $ do getGroupEntryForName (group cnf) >>= setGroupID . groupID getUserEntryForName (user cnf) >>= setUserID . userID -preFork :: Socket -> Int -> C10kServer -> C10kConfig -> IO ()-preFork s n srv cnf = do- ignoreSigChild+writePidFile :: C10kConfig -> IO ()+writePidFile cnf = do pid <- getProcessID- cids <- replicateM n $ forkProcess (runServer s srv cnf)- mapM_ (terminator pid cids) [sigTERM,sigINT]- where- ignoreSigChild = installHandler sigCHLD Ignore Nothing- terminator pid cids sig = installHandler sig (Catch (terminate pid cids)) Nothing- terminate pid cids = do- mapM_ terminateChild cids- signalProcess killProcess pid- terminateChild cid = signalProcess sigTERM cid `catch` ignore+ writeFile (pidFile cnf) $ show pid ++ "\n" ---------------------------------------------------------------- -runServer :: Socket -> C10kServer -> C10kConfig -> IO ()-runServer s srv cnf = do+runServer :: Dispatch -> C10kConfig -> IO ()+runServer dispatch cnf = do startedHook cnf- mvar <- newMVar 0- dispatchOrSleep mvar s srv cnf+ ref <- newIORef 0 :: IO (IORef Int)+ dispatchOrSleep ref dispatch cnf -dispatchOrSleep :: MVar Int -> Socket -> C10kServer -> C10kConfig -> IO ()-dispatchOrSleep mvar s srv cnf = do+dispatchOrSleep :: IORef Int -> Dispatch -> C10kConfig -> IO ()+dispatchOrSleep ref dispatch cnf = do n <- howMany if n > threadNumberPerProcess cnf then sleep (sleepTimer cnf * microseconds)- else dispatch- dispatchOrSleep mvar s srv cnf+ else dispatch increase decrease+ dispatchOrSleep ref dispatch cnf where- dispatch = do- (sock,_) <- accept s- increase- forkIO $ srv sock `finally` (decrease >> (sClose sock `catch` ignore))- return ()- howMany = readMVar mvar- increase = modifyMVar_ mvar (return . succ)- decrease = modifyMVar_ mvar (return . pred)+ howMany = readIORef ref+ increase = atomicModifyIORef ref (\n -> (n+1,()))+ decrease = atomicModifyIORef ref (\n -> (n-1,())) sleep = threadDelay +---------------------------------------------------------------- +type Dispatch = IO () -> IO () -> IO ()++dispatchS :: C10kServer -> Socket -> Dispatch+dispatchS srv sock inc dec = do+ (connsock,_) <- accept sock+ inc+ forkIO $ srv connsock `finally` (dec >> sClose connsock)+ return ()++dispatchH :: C10kServerH -> Socket -> Dispatch+dispatchH srv sock inc dec = do+ (hdl,tcpi) <- T.accept sock+ inc+ forkIO $ srv hdl tcpi `finally` (dec >> hClose hdl)+ return ()+ ---------------------------------------------------------------- ignore :: SomeException -> IO ()@@ -165,3 +232,24 @@ microseconds :: Int microseconds = 1000000++----------------------------------------------------------------++listenTo :: Maybe HostName -> ServiceName -> IO Socket+listenTo maddr serv = do+ proto <- getProtocolNumber "tcp"+ let hints = defaultHints {+ addrFlags = [ AI_ADDRCONFIG, AI_NUMERICHOST+ , AI_NUMERICSERV, AI_PASSIVE+ ]+ , addrSocketType = Stream+ , addrProtocol = proto+ }+ addrs <- getAddrInfo (Just hints) maddr (Just serv)+ let addrs' = filter (\x -> addrFamily x == AF_INET6) addrs+ addr = if null addrs' then head addrs else head addrs'+ sock <- socket (addrFamily addr) (addrSocketType addr) (addrProtocol addr)+ setSocketOption sock ReuseAddr 1+ bindSocket sock (addrAddress addr)+ listen sock maxListenQueue+ return sock
+ Network/TCPInfo.hs view
@@ -0,0 +1,44 @@+{-|+ Yet another accept() to tell TCP information.+-}+module Network.TCPInfo where++import Data.List+import Network.Socket+import System.IO++{-|+ A Type to carry TCP information.+-}+data TCPInfo = TCPInfo {+ -- | Local IP address+ myAddr :: HostName+ -- | Local port number+ , myPort :: ServiceName+ -- | Remote IP address+ , peerAddr :: HostName+ -- | Remote port number+ , peerPort :: ServiceName+ } deriving (Eq,Show)++----------------------------------------------------------------++{-|+ Yet another accept() to return both 'Handle' and 'TCPInfo'.+-}+accept :: Socket -> IO (Handle, TCPInfo)+accept sock = do+ (sock', sockaddr) <- Network.Socket.accept sock+ let getInfo = getNameInfo [NI_NUMERICHOST, NI_NUMERICSERV] True True+ (Just vMyAddr, Just vMyPort) <- getSocketName sock' >>= getInfo+ (Just vPeerAddr, Just vPeerPort) <- getInfo sockaddr+ hdl <- socketToHandle sock' ReadWriteMode+ let info = TCPInfo { myAddr = strip vMyAddr+ , myPort = vMyPort+ , peerAddr = strip vPeerAddr+ , peerPort = vPeerPort}+ return (hdl, info)+ where+ strip x+ | "::ffff:" `isPrefixOf` x = drop 7 x+ | otherwise = x
c10k.cabal view
@@ -1,16 +1,16 @@ Name: c10k-Version: 0.2.0+Version: 0.5.0 Author: Kazu Yamamoto <kazu@iij.ad.jp> Maintainer: Kazu Yamamoto <kazu@iij.ad.jp> License: BSD3 License-File: LICENSE-Synopsis: C10k server library+Synopsis: C10k server library using prefork Description: Network server library to handle- over 10,000 connections. Since GHC 6.10.4 or+ over 10,000 connections. Since GHC 6.12 or earlier uses select(), it cannot handle connections over 1,024. This library uses the prefork technique to get over the barrier.- Programs complied by GHC 6.10.4 or earlier+ Programs complied by GHC 6.10 or earlier with the -threaded option kill the IO thread when forkProcess is used. So, don't specify the -threaded option to use this library.@@ -19,9 +19,13 @@ Cabal-Version: >= 1.6 Build-Type: Simple Library- GHC-Options: -Wall -O2+ if impl(ghc >= 6.12)+ GHC-Options: -Wall -fno-warn-unused-do-bind+ else+ GHC-Options: -Wall Exposed-Modules: Network.C10kServer- Build-Depends: base >= 4 && < 10, haskell98, network, unix, hdaemonize+ Network.TCPInfo+ Build-Depends: base >= 4 && < 5, network, unix Source-Repository head Type: git Location: git://github.com/kazu-yamamoto/c10k.git