diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/prefork.cabal b/prefork.cabal new file mode 100644 --- /dev/null +++ b/prefork.cabal @@ -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 diff --git a/sample/simple.hs b/sample/simple.hs new file mode 100644 --- /dev/null +++ b/sample/simple.hs @@ -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]) diff --git a/sample/various-workers.hs b/sample/various-workers.hs new file mode 100644 --- /dev/null +++ b/sample/various-workers.hs @@ -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]) diff --git a/sample/warp.hs b/sample/warp.hs new file mode 100644 --- /dev/null +++ b/sample/warp.hs @@ -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 + ) + + diff --git a/src/System/Prefork.hs b/src/System/Prefork.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork.hs @@ -0,0 +1,190 @@ +{- | + Module : System.Prefork + Copyright : (c) 2013 GREE, Inc. + License : MIT-style + + Maintainer : Kiyoshi Ikehara + 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]) +@ + +-} diff --git a/src/System/Prefork/Class.hs b/src/System/Prefork/Class.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork/Class.hs @@ -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 _ = [] diff --git a/src/System/Prefork/Main.hs b/src/System/Prefork/Main.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork/Main.hs @@ -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 + diff --git a/src/System/Prefork/Main/Internal.hs b/src/System/Prefork/Main/Internal.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork/Main/Internal.hs @@ -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 + diff --git a/src/System/Prefork/Settings.hs b/src/System/Prefork/Settings.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork/Settings.hs @@ -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 () + diff --git a/src/System/Prefork/Types.hs b/src/System/Prefork/Types.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork/Types.hs @@ -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 diff --git a/src/System/Prefork/Worker.hs b/src/System/Prefork/Worker.hs new file mode 100644 --- /dev/null +++ b/src/System/Prefork/Worker.hs @@ -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 [] + diff --git a/test/test-prefork-server.hs b/test/test-prefork-server.hs new file mode 100644 --- /dev/null +++ b/test/test-prefork-server.hs @@ -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 + + diff --git a/test/test-prefork.hs b/test/test-prefork.hs new file mode 100644 --- /dev/null +++ b/test/test-prefork.hs @@ -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