prefork (empty) → 0.0.9
raw patch · 15 files changed
+1158/−0 lines, 15 filesdep +asyncdep +basedep +blaze-buildersetup-changed
Dependencies added: async, base, blaze-builder, bytestring, cab, cmdargs, containers, data-default, directory, filepath, hspec, http-types, network, prefork, process, stm, system-argv0, system-filepath, unix, wai, warp
Files
- LICENSE +22/−0
- Setup.hs +2/−0
- prefork.cabal +133/−0
- sample/simple.hs +30/−0
- sample/various-workers.hs +32/−0
- sample/warp.hs +128/−0
- src/System/Prefork.hs +190/−0
- src/System/Prefork/Class.hs +14/−0
- src/System/Prefork/Main.hs +31/−0
- src/System/Prefork/Main/Internal.hs +131/−0
- src/System/Prefork/Settings.hs +92/−0
- src/System/Prefork/Types.hs +48/−0
- src/System/Prefork/Worker.hs +66/−0
- test/test-prefork-server.hs +111/−0
- test/test-prefork.hs +128/−0
+ LICENSE view
@@ -0,0 +1,22 @@+Copyright (c) 2014 GREE, Inc.++MIT License++Permission is hereby granted, free of charge, to any person obtaining+a copy of this software and associated documentation files (the+"Software"), to deal in the Software without restriction, including+without limitation the rights to use, copy, modify, merge, publish,+distribute, sublicense, and/or sell copies of the Software, and to+permit persons to whom the Software is furnished to do so, subject to+the following conditions:++The above copyright notice and this permission notice shall be+included in all copies or substantial portions of the Software.++THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,+EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF+MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND+NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE+LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION+OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION+WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ prefork.cabal view
@@ -0,0 +1,133 @@+-- Initial haskell-prefork.cabal generated by cabal init. For further +-- documentation, see http://haskell.org/cabal/users-guide/++name: prefork+version: 0.0.9+synopsis: A library for building a prefork-style server quickly+license: MIT+license-file: LICENSE+author: Kiyoshi Ikehara, Benjamin Sruma+maintainer: kiyoshi.ikehara@gree.net+copyright: (c) 2013 GREE, Inc.+category: System+build-type: Simple+cabal-version: >=1.8+description: Prefork is a library for building a prefork-style server quickly.+ ++Source-Repository head+ type: git+ location: https://github.com/gree/haskell-prefork.git+++Flag sample+ Description: Build sample programs+ Default: False++Library+ exposed-modules: System.Prefork+ , System.Prefork.Main+ , System.Prefork.Main.Internal+ , System.Prefork.Types+ , System.Prefork.Class+ , System.Prefork.Worker+ , System.Prefork.Settings+ -- other-modules: + hs-source-dirs: src+ build-depends: base >= 4.6 && < 4.8+ , stm >= 2.4+ , containers+ , unix+ , system-argv0+ , system-filepath+ , data-default+ , process >= 1.2 && < 1.3++Executable prefork-sample-simple+ if flag(sample)+ Buildable: True+ else+ Buildable: False+ Ghc-Options: -threaded -Wall -rtsopts+ Build-Depends: base >=4 && <5+ , bytestring+ , containers+ , network+ , unix+ , prefork+ Hs-source-dirs: sample+ Main-is: simple.hs+ Extensions: DeriveDataTypeable++Executable prefork-sample-various-workers+ if flag(sample)+ Buildable: True+ else+ Buildable: False+ Ghc-Options: -threaded -Wall -rtsopts+ Build-Depends: base >=4 && <5+ , bytestring+ , containers+ , network+ , unix+ , prefork+ Hs-source-dirs: sample+ Main-is: various-workers.hs+ Extensions: DeriveDataTypeable++Executable prefork-sample-warp+ if flag(sample)+ Buildable: True+ else+ Buildable: False+ Ghc-Options: -threaded -Wall -rtsopts+ Build-Depends: base >=4 && <5+ , bytestring+ , containers+ , network+ , wai >= 2.0 && < 3.0+ , warp >= 2.0 && < 3.0 + , stm+ , blaze-builder+ , http-types+ , unix+ , prefork+ , cmdargs+ , async+ Hs-source-dirs: sample+ Main-is: warp.hs+ Extensions: DeriveDataTypeable++test-suite test-prefork+ type: exitcode-stdio-1.0+ build-depends: base >= 4.6 && < 4.8+ , prefork+ , hspec+ , containers+ , stm + , unix+ , process >= 1.2 && < 1.3+ , cab < 0.2.10+ , directory+ , filepath+ hs-source-dirs: test+ main-is: test-prefork.hs+ extensions: DeriveDataTypeable+ ghc-options: -O2 -Wall+ default-language: Haskell2010++test-suite test-prefork-server+ Ghc-Options: -threaded -Wall -rtsopts+ type: exitcode-stdio-1.0+ build-depends: base >= 4.6 && < 4.8+ , prefork+ , hspec+ , containers+ , stm + , unix+ , process >= 1.2 && < 1.3+ hs-source-dirs: test+ main-is: test-prefork-server.hs+ extensions: DeriveDataTypeable+ ghc-options: -O2 -Wall+ default-language: Haskell2010
+ sample/simple.hs view
@@ -0,0 +1,30 @@+-- Copyright: (c) 2013 Gree, Inc.+-- License: MIT-style++import System.Prefork+import System.Posix+import System.Exit (exitSuccess)++data ServerConfig = ServerConfig+data Worker = Worker1 String deriving (Show, Read)++instance WorkerContext Worker++main :: IO ()+main = defaultMain defaultSettings {+ psUpdateConfig = updateConfig+ , psUpdateServer = updateServer+ , psOnStart = \_ -> do+ pid <- getProcessID+ putStrLn $ "Please send SIGHUP to " ++ show pid ++ " to relaunch a worker"+ } $ \so -> case so of+ Worker1 msg -> print msg >> exitSuccess++updateConfig :: IO (Maybe ServerConfig)+updateConfig = do+ return (Just ServerConfig)++updateServer :: ServerConfig -> IO ([ProcessID])+updateServer ServerConfig = do+ pid <- forkWorkerProcess (Worker1 "Hello. I'm a worker.")+ return ([pid])
+ sample/various-workers.hs view
@@ -0,0 +1,32 @@+-- Copyright: (c) 2013 Gree, Inc.+-- License: MIT-style++import System.Prefork+import System.Posix+import System.Exit (exitSuccess)++data ServerConfig = ServerConfig+data Worker = Worker1 String | Worker2 String deriving (Show, Read, Eq)++instance WorkerContext Worker++main :: IO ()+main = defaultMain defaultSettings {+ psUpdateConfig = updateConfig+ , psUpdateServer = updateServer+ , psOnStart = \_ -> do+ pid <- getProcessID+ putStrLn $ "Please send SIGHUP to " ++ show pid ++ " to relaunch workers"+ } $ \w -> case w of+ Worker1 msg -> putStrLn msg >> exitSuccess+ Worker2 msg -> putStrLn msg >> exitSuccess++updateConfig :: IO (Maybe ServerConfig)+updateConfig = do+ return (Just ServerConfig)++updateServer :: ServerConfig -> IO ([ProcessID])+updateServer ServerConfig = do+ pid1 <- forkWorkerProcess (Worker1 "Hello. I'm a worker 1.")+ pid2 <- forkWorkerProcess (Worker2 "Hello. I'm a worker 2.")+ return ([pid1, pid2])
+ sample/warp.hs view
@@ -0,0 +1,128 @@+-- Copyright: (c) 2013 Gree, Inc.+-- License: MIT-style++{-# LANGUAGE OverloadedStrings #-}++-- This is a simple web server based on Warp++import Blaze.ByteString.Builder.Char.Utf8+import Foreign.C.Types+import Network.BSD+import Network.Socket+import Control.Exception+import Control.Concurrent.STM+import Control.Concurrent.Async+import Network.Wai+import qualified Network.Wai.Handler.Warp as Warp+import Network.HTTP.Types+import System.Posix+import System.Prefork+import System.Console.CmdArgs++-- Application specific configuration+data Config = Config {+ cWarpSettings :: Warp.Settings+ }++-- Worker context passed by the parent+data Worker = Worker {+ wId :: Int+ , wPort :: Int+ , wSocketFd :: CInt+ , wHost :: String+ , wCap :: Int+ } deriving (Show, Read)++instance WorkerContext Worker where+ rtsOptions w = ["-N" ++ show (wCap w)]++instance Eq Worker where+ (==) a b = wId a == wId b++instance Ord Worker where+ compare a b = compare (wId a) (wId b)++-- Server states+data Server = Server {+ sServerSoc :: TVar (Maybe Socket)+ , sPort :: Int+ , sWorkers :: Int+ }++-- Command line options+data Warp = Warp {+ port :: Int+ , workers :: Int+ , extraArgs :: [String]+ } deriving (Show, Data, Typeable, Eq)++cmdLineOptions :: Warp+cmdLineOptions = Warp {+ port = 11111 &= name "p" &= help "Port number" &= typ "PORT"+ , workers = 4 &= name "w" &= help "Number of workers" &= typ "NUM"+ , extraArgs = def &= args+ } &=+ help "Preforking Warp Server Sample" &=+ summary ("Preforking Warp Server Sample, (C) GREE, Inc") &=+ details ["Web Server"]++main :: IO ()+main = do+ option <- cmdArgs cmdLineOptions+ resource <- emptyPreforkResource+ mSoc <- newTVarIO Nothing+ let s = Server mSoc (port option) (workers option)+ defaultMain (relaunchSettings resource (update s) (fork s)) $ \(Worker { wId = i, wSocketFd = fd }) -> do+ -- worker action+ soc <- mkSocket fd AF_INET Stream defaultProtocol Listening+ mConfig <- updateConfig s+ case mConfig of+ Just config -> do+ a <- asyncOn i $ Warp.runSettingsSocket (cWarpSettings config) soc $ serverApp+ wait a+ Nothing -> return ()+ where+ update :: Server -> PreforkResource Worker -> IO (Maybe Config)+ update s resource = do+ mConfig <- updateConfig s+ updateWorkerSet resource $ flip map [0..(sWorkers s - 1)] $ \i ->+ Worker { wId = i, wPort = (sPort s), wSocketFd = -1, wHost = "localhost", wCap = sWorkers s }+ return (mConfig)++ fork :: Server -> Worker -> IO (ProcessID)+ fork Server { sServerSoc = socVar } w = do+ msoc <- readTVarIO socVar+ soc <- case msoc of+ Just soc -> return (soc)+ Nothing -> do+ hentry <- getHostByName (wHost w)+ soc <- listenOnAddr (SockAddrInet (fromIntegral (wPort w)) (head $ hostAddresses hentry))+ atomically $ writeTVar socVar (Just soc)+ return (soc)+ let w' = w { wSocketFd = fdSocket soc }+ forkWorkerProcessWithArgs (w') ["id=" ++ show (wId w') ]++-- Load config+updateConfig :: Server -> IO (Maybe Config)+updateConfig s = return (Just $ Config Warp.defaultSettings { Warp.settingsPort = fromIntegral (sPort s) })++-- Web application+serverApp :: Application+serverApp _ = return $ responseBuilder status200 [] $ fromString "hello"++-- Create a server socket with SockAddr+listenOnAddr :: SockAddr -> IO Socket+listenOnAddr sockAddr = do+ let backlog = 1024+ proto <- getProtocolNumber "tcp"+ bracketOnError+ (socket AF_INET Stream proto)+ (sClose)+ (\sock -> do+ setSocketOption sock ReuseAddr 1+ bindSocket sock sockAddr+ listen sock backlog+ return sock+ )++
+ src/System/Prefork.hs view
@@ -0,0 +1,190 @@+{- |+ Module : System.Prefork+ Copyright : (c) 2013 GREE, Inc.+ License : MIT-style+ + Maintainer : Kiyoshi Ikehara <kiyoshi.ikehara@gree.net>+ Stability : experimental+ Portability : portable+-}++{-# LANGUAGE NoMonomorphismRestriction #-}++module System.Prefork(+ -- * Overview+ -- $overview+ + -- * Tutorial+ -- ** A typical preforking server+ -- $relaunch_tutorial+ + -- ** Using low-level interface+ -- $rawlevel_tutorial+ module System.Prefork.Class+ , module System.Prefork.Types+ , module System.Prefork.Main+ , module System.Prefork.Worker+ , module System.Prefork.Settings+ ) where++import System.Prefork.Class+import System.Prefork.Types+import System.Prefork.Main+import System.Prefork.Worker+import System.Prefork.Settings++{- $overview+This is a library for servers based on worker process model (preforking).+-}++{- $relaunch_tutorial+Import System.Prefork in your Main module.++@+import ...+import System.Posix+import System.Prefork+import System.Console.CmdArgs+@++Define data type used for server configuration.++@+-- Application specific configuration+data Config = Config {+ cWarpSettings :: Warp.Settings+ }+@++Define workers as a data type that belongs to 'WorkerContext' class.+In this case, the field 'wId' is a ID number for idenfitying single worker process and other fields are+parameters for a worker process.++@+-- Worker context passed by the parent+data Worker = Worker {+ wId :: Int+ , wPort :: Int+ , wSocketFd :: CInt+ , wHost :: String+ , wCap :: Int+ } deriving (Show, Read)++instance WorkerContext Worker where+ rtsOptions w = [\"-N\" ++ show (wCap w)]+@++Define Eq and Ord instances for Worker. These are required for using 'relaunchSettings'.++@+instance Eq Worker where+ (==) a b = wId a == wId b++instance Ord Worker where+ compare a b = compare (wId a) (wId b)+@++Call 'defaultMain' with 'update' and 'fork' functions in your 'main' function.+'relaunchSettings' is a function that creates comvenient settings for a typical prefork server.++@+main :: IO ()+main = do+ option <- cmdArgs cmdLineOptions+ resource <- emptyPreforkResource+ mSoc <- newTVarIO Nothing+ let s = Server mSoc (port option) (workers option)+ defaultMain (relaunchSettings resource (update s) (fork s)) $ \(Worker { wId = i, wSocketFd = fd, wHost = _host }) -> do+ -- worker action+ soc <- mkSocket fd AF_INET Stream defaultProtocol Listening+ mConfig <- updateConfig s+ case mConfig of+ Just config -> do+ a <- asyncOn i $ Warp.runSettingsSocket (cWarpSettings config) soc $ serverApp+ wait a+ Nothing -> return ()+ where+ ...+@++'update' function is used for modifying the worker process configuration.+If you want to increase or decrease the number of workers, change worker parameters, and etc,+you can use 'updateWorkerSet' function here.++@+ update :: Server -> PreforkResource Worker -> IO (Maybe Config)+ update s resource = do+ mConfig <- updateConfig s+ updateWorkerSet resource $ flip map [0..(sWorkers s - 1)] $ \i ->+ Worker { wId = i, wPort = (sPort s), wSocketFd = -1, wHost = \"localhost\", wCap = sWorkers s }+ return (mConfig)+@++'fork' function simply creates a worker process with the given parameters.+You should call 'forkWorkerProcess' or 'forkWorkerProcessWithArgs' in this function to invoke a child process as a worker.+In this case, the arguments of 'forkWorkerProcessWithArgs' are just for displaying id number and not used.++@+ fork :: Server -> Worker -> IO (ProcessID)+ fork Server { sServerSoc = socVar } w = do+ msoc <- readTVarIO socVar+ soc <- case msoc of+ Just soc -> return (soc)+ Nothing -> do+ hentry <- getHostByName (wHost w)+ soc <- listenOnAddr (SockAddrInet (fromIntegral (wPort w)) (head $ hostAddresses hentry))+ atomically $ writeTVar socVar (Just soc)+ return (soc)+ let w' = w { wSocketFd = fdSocket soc }+ forkWorkerProcessWithArgs (w') [\"id=\" ++ show (wId w') ]+@++-}++{- $rawlevel_tutorial+Import System.Prefork in your Main module.++@+import System.Prefork+import System.Posix+import System.Exit (exitSuccess)+@++Define data type used for server configuration.++@+data ServerConfig = ServerConfig+@++Define workers as a data type.++@+data Worker = Worker1 String deriving (Show, Read)++instance WorkerContext Worker+@++Call System.Prefork.defaultMain function with settings in your main function.+ +@+main :: IO ()+main = defaultMain defaultSettings {+ psUpdateConfig = updateConfig+ , psUpdateServer = updateServer+ , psOnStart = \_ -> do+ pid <- getProcessID+ putStrLn $ \"Please send SIGHUP to \" ++ show pid ++ \" to relaunch a worker\"+ } $ \so -> case so of+ Worker1 msg -> print msg >> exitSuccess++updateConfig :: IO (Maybe ServerConfig)+updateConfig = do+ return (Just ServerConfig)++updateServer :: ServerConfig -> IO ([ProcessID])+updateServer ServerConfig = do+ pid <- forkWorkerProcess (Worker1 \"Hello. I'm a worker.\")+ return ([pid])+@++-}
+ src/System/Prefork/Class.hs view
@@ -0,0 +1,14 @@+-- Copyright: (c) 2013 GREE, Inc.+-- License: MIT-style++module System.Prefork.Class where++class (Show a, Read a) => WorkerContext a where+ encodeToString :: a -> String+ encodeToString = show+ + decodeFromString :: String -> a+ decodeFromString = read++ rtsOptions :: a -> [String]+ rtsOptions _ = []
+ src/System/Prefork/Main.hs view
@@ -0,0 +1,31 @@+-- Copyright: (c) 2013 GREE, Inc.+-- License: MIT-style++module System.Prefork.Main (+ defaultMain+ , runSettings+ ) where++import System.Environment (lookupEnv)+import System.Posix.Env (setEnv)++import System.Prefork.Class+import System.Prefork.Types+import System.Prefork.Worker+import System.Prefork.Main.Internal++defaultMain :: (WorkerContext w) => PreforkSettings sc -> (w -> IO ()) -> IO ()+defaultMain = runSettings++runSettings :: (WorkerContext w) => PreforkSettings sc -> (w -> IO ()) -> IO ()+runSettings settings workerAction = do+ mPrefork <- lookupEnv preforkEnvKey+ case mPrefork of+ Just "parent" -> do+ setEnv preforkEnvKey "server" True+ masterMain settings+ Just _ -> workerMain workerAction+ Nothing -> do+ setEnv preforkEnvKey "parent" True+ rootMain settings+
+ src/System/Prefork/Main/Internal.hs view
@@ -0,0 +1,131 @@+-- Copyright: (c) 2013 GREE, Inc.+-- License: MIT-style++module System.Prefork.Main.Internal (+ masterMain+ , workerMain+ ) where++import Data.Maybe (listToMaybe, catMaybes)+import Control.Monad (unless, forM_, void)+import Control.Concurrent.STM+import System.Posix+import System.Environment (getArgs)+import System.Exit (exitSuccess)++import System.Prefork.Class+import System.Prefork.Types+import System.Prefork.Worker++data ControlMessage = + TerminateCM+ | InterruptCM+ | HungupCM+ | QuitCM+ | ChildCM+ deriving (Eq, Show, Read)++data Prefork sc = Prefork {+ pServerConfig :: !(TVar (Maybe sc))+ , pCtrlChan :: !(TChan ControlMessage)+ , pProcs :: !(TVar [ProcessID])+ , pSettings :: !(PreforkSettings sc)+ }++workerMain :: (WorkerContext so) => (so -> IO ()) -> IO ()+workerMain act = do+ rawOpt <- getLine+ act $ decodeFromString rawOpt+ exitSuccess++masterMain :: PreforkSettings sc -> IO ()+masterMain settings = do+ ctrlChan <- newTChanIO+ procs <- newTVarIO []+ mConfig <- psUpdateConfig settings+ soptVar <- newTVarIO mConfig+ atomically $ writeTChan ctrlChan HungupCM + setupServer ctrlChan+ (psOnStart settings) mConfig+ masterMainLoop (Prefork soptVar ctrlChan procs settings)+ mConfig' <- readTVarIO soptVar+ (psOnFinish settings) mConfig'+ return ()++masterMainLoop :: Prefork sc -> IO ()+masterMainLoop prefork@Prefork { pSettings = settings } = loop False+ where+ loop :: Bool -> IO ()+ loop finishing = do+ (msg, cids) <- atomically $ do+ msg <- readTChan $ pCtrlChan prefork+ procs <- readTVar $ pProcs prefork+ return (msg, procs)+ finRequested <- dispatch msg cids finishing+ childIds <- readTVarIO (pProcs prefork)+ unless ((finishing || finRequested) && null childIds) $ loop (finishing || finRequested)++ dispatch :: ControlMessage -> [CPid] -> Bool -> IO (Bool)+ dispatch msg cids finishing = case msg of+ TerminateCM -> do+ m <- readTVarIO $ pServerConfig prefork+ whenJust m $ flip (psOnTerminate settings) cids+ return (True)+ InterruptCM -> do+ m <- readTVarIO $ pServerConfig prefork+ whenJust m $ flip (psOnInterrupt settings) cids+ return (True)+ HungupCM -> do+ mConfig <- psUpdateConfig (pSettings prefork)+ whenJust mConfig $ \config -> do+ newProcs <- (psUpdateServer (pSettings prefork)) config+ atomically $ do+ writeTVar (pServerConfig prefork) (Just config)+ modifyTVar' (pProcs prefork) $ (++) newProcs+ return (False)+ QuitCM -> do+ m <- readTVarIO $ pServerConfig prefork+ whenJust m $ psOnQuit settings+ return (False)+ ChildCM -> do+ finished <- cleanupChildren cids (pProcs prefork)+ mConfig <- readTVarIO $ pServerConfig prefork+ case mConfig of+ Just config -> do+ forM_ finished $ (psCleanupChild settings) config+ unless finishing $ do+ newProcs <- (psOnChildFinished settings) config+ atomically $ modifyTVar' (pProcs prefork) $ (++) newProcs+ Nothing -> return ()+ return (False)++whenJust :: Monad m => Maybe a -> (a -> m ()) -> m ()+whenJust mg f = maybe (return ()) f mg++cleanupChildren :: [ProcessID] -> TVar [ProcessID] -> IO ([ProcessID])+cleanupChildren cids procs = do+ r <- mapM (getProcessStatus False False) cids -- WNOHANG is true, WUNTRACED is false+ let finished = catMaybes $ flip map (zip cids r) $ checkFinished+ atomically $ modifyTVar' procs $ filter (flip notElem finished)+ return (finished)+ where+ checkFinished (pid, x) = case x of+ Just (Exited {}) -> Just pid+ Just (Terminated {}) -> Just pid+ Just (Stopped {}) -> Nothing+ _ -> Nothing++setupServer :: TChan ControlMessage -> IO ()+setupServer chan = do+ let delegate = \sig msg -> setSignalHandler sig $ Catch $ atomically $ writeTChan chan msg+ delegate sigCHLD ChildCM+ delegate sigTERM TerminateCM+ delegate sigINT InterruptCM+ delegate sigQUIT QuitCM+ delegate sigHUP HungupCM+ setSignalHandler sigPIPE $ Ignore+ return ()+ where+ setSignalHandler :: Signal -> System.Posix.Handler -> IO ()+ setSignalHandler sig func = void $ installHandler sig func Nothing+
+ src/System/Prefork/Settings.hs view
@@ -0,0 +1,92 @@+-- Copyright: (c) 2013 GREE, Inc.+-- License: MIT-style++module System.Prefork.Settings (defaultSettings, relaunchSettings) where++import Control.Exception (SomeException, catch)+import System.Posix++import System.Prefork.Class+import System.Prefork.Types+import System.Prefork.Worker+import Control.Concurrent.STM+import qualified Data.Set as S+import qualified Data.Map as M+import Data.List+import Control.Monad++{- | default settings for defaultMain+ + This just sends signals to child processes.+-}+defaultSettings :: PreforkSettings sc+defaultSettings = PreforkSettings {+ psOnTerminate = \_config -> mapM_ (sendSignal sigTERM)+ , psOnInterrupt = \_config -> mapM_ (sendSignal sigINT)+ , psOnQuit = \_config -> return ()+ , psOnChildFinished = \_config -> return ([])+ , psOnStart = \_mConfig -> return ()+ , psOnFinish = \_mConfig -> return ()+ , psUpdateServer = \_config -> return ([])+ , psCleanupChild = \_config _pid -> return ()+ , psUpdateConfig = return (Nothing)+ }++{- | relaunch settings+ + This requires 'PreforkResource' that describes resouces used by workers.+ 'relaunchSettings' takes two functions.+ The one is 'update' and the other is 'fork'.+ 'update' function is used for reading server configuration (usually from a file)+ and update 'sc' type.+ 'fork' function is used for launching a new worker with a worker context.+ The worker context should be the instance of 'Eq' and 'Ord' classes because it + will be a element type of 'Set'.+-}+relaunchSettings :: (Ord w, Eq w)+ => PreforkResource w+ -> (PreforkResource w -> IO (Maybe sc))+ -> (w -> IO (ProcessID))+ -> PreforkSettings sc+relaunchSettings resource updateAction forkAction = defaultSettings {+ psUpdateConfig = updateAction resource+ , psUpdateServer = updateWorkers resource forkAction+ , psCleanupChild = cleanupChild resource+ , psOnChildFinished = relaunchWorkers resource forkAction+ }+ where+ -- Clean up application specific resources associated to a child process+ cleanupChild :: (Ord w, Eq w) => PreforkResource w -> sc -> ProcessID -> IO ()+ cleanupChild resource _config pid = atomically $ modifyTVar' (prProcs resource) $ M.delete pid+ + -- Update the entire state of a server+ updateWorkers :: (Ord w, Eq w) => PreforkResource w -> (w -> IO (ProcessID)) -> sc -> IO ([ProcessID])+ updateWorkers resource forkAction _config = do+ workers <- readTVarIO (prWorkers resource)+ newPids <- forM (S.toList workers) $ \w -> do+ pid <- forkAction w+ return (pid, w)+ oldPids <- atomically $ swapTVar (prProcs resource) (M.fromList newPids)+ forM_ (M.keys oldPids) $ sendSignal sigTERM+ return (map fst newPids)+ + -- Relaunch workers after some of them terminate+ relaunchWorkers :: (Ord w, Eq w) => PreforkResource w -> (w -> IO (ProcessID)) -> sc -> IO ([ProcessID])+ relaunchWorkers resource@PreforkResource { prProcs = procs, prWorkers = workers } forkAction _config = do+ (live, workers) <- atomically $ do+ live <- readTVar procs+ workers <- readTVar workers+ return (live, workers)+ newPids <- fmap M.fromList $ forM (S.toList workers \\ M.elems live) $ \w -> do+ pid <- forkAction w+ return (pid, w)+ atomically $ modifyTVar' procs $ M.union newPids+ return (M.keys newPids)++-- private+sendSignal :: Signal -> ProcessID -> IO ()+sendSignal sig cid = signalProcess sig cid `catch` ignore+ where+ ignore :: SomeException -> IO ()+ ignore _ = return ()+
+ src/System/Prefork/Types.hs view
@@ -0,0 +1,48 @@+-- Copyright: (c) 2013 GREE, Inc.+-- License: MIT-style++module System.Prefork.Types (+ PreforkSettings(..)+ , PreforkResource(..)+ , makePreforkResource+ , emptyPreforkResource+ , updateWorkerSet+ ) where++import qualified Data.Set as S+import qualified Data.Map as M+import System.Posix (ProcessID)+import System.Prefork.Class+import Control.Concurrent.STM+import Control.Applicative++{- | This represents handlers for controlling child processes.+ + The type parameter 'sc' is a server configuration (application specific).+-}+data PreforkSettings sc = PreforkSettings {+ psOnTerminate :: sc -> [ProcessID] -> IO () -- ^ This is called on TERM signal.+ , psOnInterrupt :: sc -> [ProcessID] -> IO () -- ^ This is called on INT signal.+ , psOnQuit :: sc -> IO () -- ^ This is called on QUIT signal.+ , psOnChildFinished :: sc -> IO ([ProcessID]) -- ^ This is called on CHILD signal.+ , psOnStart :: Maybe sc -> IO () -- ^ This is called on the start of a parent.+ , psOnFinish :: Maybe sc -> IO () -- ^ This is called on the finish of a parent.+ , psUpdateServer :: sc -> IO ([ProcessID]) -- ^ This is called when the server state has changed.+ , psCleanupChild :: sc -> ProcessID -> IO () -- ^ This is called when one of child process has finished.+ , psUpdateConfig :: IO (Maybe sc) -- ^ This is called when a configuration is needed.+ }++data PreforkResource w = PreforkResource {+ prProcs :: TVar (M.Map ProcessID w)+ , prWorkers :: TVar (S.Set w)+ }++makePreforkResource :: (Ord w, Eq w) => [w] -> IO (PreforkResource w)+makePreforkResource workers = PreforkResource <$> newTVarIO M.empty+ <*> newTVarIO (S.fromList workers)++emptyPreforkResource :: (Ord w, Eq w) => IO (PreforkResource w)+emptyPreforkResource = makePreforkResource []++updateWorkerSet :: (Ord w, Eq w) => PreforkResource w -> [w] -> IO ()+updateWorkerSet resource workers = atomically $ writeTVar (prWorkers resource) $ S.fromList workers
+ src/System/Prefork/Worker.hs view
@@ -0,0 +1,66 @@+-- Copyright: (c) 2013 GREE, Inc.+-- License: MIT-style++{-# LANGUAGE ScopedTypeVariables #-}++module System.Prefork.Worker (+ forkWorkerProcess+ , forkWorkerProcessWithArgs+ , preforkEnvKey+ ) where++import Control.Monad+import Control.Exception+import System.Process+import System.Process.Internals (withProcessHandle, ProcessHandle__(OpenHandle))+import Filesystem.Path.CurrentOS(encodeString)+import System.Posix hiding (version)+import Foreign.C.Types+import System.IO (hPutStrLn)+import System.Argv0+import Data.Maybe+import System.Environment (lookupEnv)+import System.IO+import Control.Concurrent++import System.Prefork.Class++preforkEnvKey :: String+preforkEnvKey = "PREFORK"++{- | create a new worker with arguments+-}+forkWorkerProcessWithArgs :: (WorkerContext a)+ => a -- ^ a worker context+ -> [String] -- ^ command line arguments+ -> IO ProcessID -- ^ a process id of a created worker+forkWorkerProcessWithArgs opt args = do+ exe <- liftM encodeString getArgv0+ (Just hIn, Just hOut, Just hErr, ph) <- createProcess $ (proc exe options)+ { std_in = CreatePipe+ , std_out = CreatePipe+ , std_err = CreatePipe+ }+ forkIO $ hPutStr stdout =<< hGetContents hOut+ forkIO $ hPutStr stderr =<< hGetContents hErr+ hPutStrLn hIn $ encodeToString opt+ hFlush hIn+ extractProcessID ph+ where+ options :: [String]+ options = case rtsOptions opt of+ [] -> args+ rtsopts -> args ++ ["+RTS"] ++ rtsopts ++ ["-RTS"]++ extractProcessID :: ProcessHandle -> IO ProcessID+ extractProcessID h = withProcessHandle h $ \x -> case x of+ OpenHandle pid -> return pid+ _ -> throwIO $ userError "Unable to retrieve child process ID."++{- | create a new worker+-}+forkWorkerProcess :: (WorkerContext a)+ => a -- ^ a worker context+ -> IO ProcessID -- ^ a process id of a created worker+forkWorkerProcess opt = forkWorkerProcessWithArgs opt []+
+ test/test-prefork-server.hs view
@@ -0,0 +1,111 @@+import System.Prefork+import System.Posix+import System.Exit (exitSuccess)+import System.IO +import System.Environment+import Control.Monad+import Control.Concurrent++import Constant (+ Worker(..)+ , workerNum+ , serverOption+ , masterOutputFile+ , workerOutputFile + , settingDefault+ , settingRelaunch+ )++data ServerConfig = ServerConfig++main :: IO ()+main = do+ -- not to run in cabal test+ args <- getArgs+ case args of+ ("keep-alive":_) -> runServer+ _ -> exitSuccess++runServer :: IO ()+runServer = do+ args <- getArgs+ case args of+ ("keep-alive":settings:_) -> case settings of+ "defaultSettings" -> runByDefaultSettings+ "relaunchSettings" -> runByRelaunchSettings+ _ -> exitSuccess+ _ -> exitSuccess++++runByDefaultSettings :: IO ()+runByDefaultSettings = defaultMain defaultSettings {+ psOnTerminate = onTerminate+ , psOnInterrupt = onInterrupt+ , psOnQuit = onQuit+ , psOnChildFinished = \_config -> return ([])+ , psOnStart = onStart+ , psOnFinish = onFinish+ , psUpdateServer = updateServer+ , psCleanupChild = \_config _pid -> return ()+ , psUpdateConfig = updateConfig++ } $ \so -> case so of+ Worker msg -> withFile workerOutputFile WriteMode $ \hdl -> do + hPutStr hdl msg+ exitSuccess+ where+ onTerminate :: ServerConfig -> [ProcessID] -> IO ()+ onTerminate _ cids = do+ forM_ cids $ \cid -> signalProcess sigTERM cid++ onInterrupt :: ServerConfig -> [ProcessID] -> IO ()+ onInterrupt _ cids = do+ forM_ cids $ \cid -> signalProcess sigINT cid++ onQuit :: ServerConfig -> IO ()+ onQuit _ = do+ withFile workerOutputFile WriteMode $ \hdl -> do+ hPutStr hdl "onQuit"++ onStart :: Maybe ServerConfig -> IO ()+ onStart _ = do+ withFile masterOutputFile WriteMode $ \hdl -> do+ hPutStr hdl "onStart"++ onFinish :: Maybe ServerConfig -> IO ()+ onFinish _ = do+ withFile masterOutputFile WriteMode $ \hdl -> do+ hPutStr hdl "onFinish"++ updateServer :: ServerConfig -> IO ([ProcessID])+ updateServer ServerConfig = do+ pid <- forkWorkerProcessWithArgs (Worker "updateServer") [serverOption, settingDefault]+ return ([pid])++ updateConfig :: IO (Maybe ServerConfig)+ updateConfig = do+ return (Just ServerConfig)+++runByRelaunchSettings :: IO ()+runByRelaunchSettings = do+ resource <- makePreforkResource []+ defaultMain (relaunchSettings resource updateAction forkAction) $ \(Worker _) -> do+ threadDelay 10000000000++ where+ updateAction :: PreforkResource Worker -> IO (Maybe ServerConfig)+ updateAction resource = do+ updateWorkerSet resource $ flip map [0..(workerNum-1)] $ \i ->+ Worker $ "worker" ++ show i+ return (Just ServerConfig)++ forkAction :: Worker -> IO (ProcessID)+ forkAction worker = do+ pid <- forkWorkerProcessWithArgs worker [serverOption, settingRelaunch]+ withFile "/tmp/relaunch_workers" AppendMode $ \hdl -> do+ hPutStrLn hdl $ show pid+ return pid++
+ test/test-prefork.hs view
@@ -0,0 +1,128 @@+{-# LANGUAGE OverloadedStrings #-}++module Main where++import Test.Hspec+import qualified Data.Set as S+import qualified Data.Map as M+import Control.Exception+import Control.Concurrent+import Control.Concurrent.STM+import System.Posix+import System.Process+import System.Process.Internals (withProcessHandle, ProcessHandle__(OpenHandle))+import System.IO+import System.Directory+import System.FilePath+import Data.Functor++import Util+import System.Prefork.Class+import System.Prefork.Types++import Constant (+ Worker(..)+ , workerNum+ , serverOption+ , masterOutputFile+ , workerOutputFile + , relaunchWorkerFile+ , settingDefault+ , settingRelaunch+ ) +++main :: IO ()+main = do+ hspec $ do+ describe "Class" $ do+ let worker = Worker "test"+ it "translate worker to string" $ encodeToString worker `shouldBe` show worker+ it "translate string to worker" $ decodeFromString "Worker \"test\"" `shouldBe` worker+ it "returns default options" $ rtsOptions worker `shouldBe` []++ describe "Types" $ do+ let w1 = Worker "test1"+ w2 = Worker "test2"++ it "makes PreforkResource" $ do+ resource <- makePreforkResource [w1, w2]+ workerMap <- atomically $ readTVar $ prProcs resource+ workerSet <- atomically $ readTVar $ prWorkers resource+ M.size workerMap `shouldBe` 0+ workerSet `shouldBe` S.fromList [w1, w2]++ it "updates workers" $ do+ resource <- makePreforkResource [w1, w2]+ updateWorkerSet resource [w1]+ workerSet <- atomically $ readTVar $ prWorkers resource+ workerSet `shouldBe` S.fromList [w1]++ describe "Main" $ do+ it "makes test server" $ do+ (_, ph) <- createTestServer settingDefault+ withFile masterOutputFile ReadMode $ \hdl -> do+ threadDelay 1000000+ flip shouldBe "onStart" =<< hGetContents hdl+ terminateProcess ph++ it "sends sigHUP" $ do+ checkOutputOnSignal sigHUP workerOutputFile "updateServer"++ it "sends sigTERM" $ do+ checkOutputOnSignal sigTERM masterOutputFile "onFinish"++ it "sends sigINT" $ do+ checkOutputOnSignal sigINT masterOutputFile "onFinish"++ it "sends sigQUIT" $ do+ checkOutputOnSignal sigQUIT workerOutputFile "onQuit"+++ it "sends sigHUP to relauch settings server" $ do+ (pid, ph) <- createTestServer settingRelaunch+ testActionBySignal sigHUP pid relaunchWorkerFile $ \hdl -> do+ terminateProcess ph+ workerPids <- lines <$> hGetContents hdl+ length workerPids `shouldBe` workerNum++ it "sends sigTERM to worker in relauch settings" $ do+ writeFile relaunchWorkerFile ""+ (_, ph) <- createTestServer settingRelaunch+ h <- openFile relaunchWorkerFile ReadMode+ workerPid <- hGetLine h+ hClose h+ testActionBySignal sigTERM (read workerPid) relaunchWorkerFile $ \hdl -> do+ terminateProcess ph+ workerPids <- lines <$> hGetContents hdl+ length workerPids `shouldBe` 1++createTestServer :: String -> IO (ProcessID, ProcessHandle)+createTestServer settings = do+ cDir <- getCurrentDirectory+ distDir <- getDistDir cDir+ let exePath = cDir </> distDir </> "build" </> "test-prefork-server" </> "test-prefork-server"+ (_, Just hOut, _, ph) <- createProcess $ (proc exePath [serverOption, settings]) { std_out = CreatePipe }+ _ <- forkIO $ hPutStr stdout =<< hGetContents hOut+ pid <- withProcessHandle ph $ \x -> case x of+ OpenHandle pid' -> return pid'+ _ -> throwIO $ userError "Unable to retrieve child process ID."++ threadDelay 1000000+ return (pid, ph)++checkOutputOnSignal :: Signal -> String -> String -> IO ()+checkOutputOnSignal sig file expected = do+ (pid, ph) <- createTestServer settingDefault+ testActionBySignal sig pid file $ \hdl -> do+ flip shouldBe expected =<< hGetContents hdl+ terminateProcess ph+++testActionBySignal :: Signal -> ProcessID -> String -> (Handle -> IO ()) -> IO ()+testActionBySignal sig pid file testAction = do+ writeFile file ""+ signalProcess sig pid+ withFile file ReadMode $ \hdl -> do+ threadDelay 1000000+ testAction hdl