git.iko.soy/distributed-process
on 263cb13 help sign in
distributed-process/benchmarks/ProcessRing.hs
105 lines 3.5 kB history raw
1{- ProcessRing benchmarks.
3To run the benchmarks, select a value for the ring size (sz) and
4the number of times to send a message around the ring
5-}
6import Control.Monad
7import Control.Distributed.Process hiding (catch)
8import Control.Distributed.Process.Node
9import Control.Exception (SomeException, catch)
10import Network.Transport.TCP (createTransport, defaultTCPParameters)
11import System.Environment
12import System.Console.GetOpt
14data Options = Options
15 { optRingSize :: Int
16 , optIterations :: Int
17 , optForward :: Bool
18 , optParallel :: Bool
19 } deriving Show
21initialProcess :: Options -> Process ()
22initialProcess op =
23 let ringSz = optRingSize op
24 msgCnt = optIterations op
25 fwd = optForward op
26 msg = ("foobar", "baz")
27 in do
28 self <- getSelfPid
29 ring <- makeRing fwd ringSz self
30 forM_ [1..msgCnt] (\_ -> send ring msg)
31 collect msgCnt
32 where relay' pid = do
33 msg <- expect :: Process (String, String)
34 send pid msg
35 relay' pid
37 forward' pid =
38 receiveWait [ matchAny (\m -> forward m pid) ] >> forward' pid
40 makeRing :: Bool -> Int -> ProcessId -> Process ProcessId
41 makeRing !f !n !pid
42 | n == 0 = go f pid
43 | otherwise = go f pid >>= makeRing f (n - 1)
45 go :: Bool -> ProcessId -> Process ProcessId
46 go False next = spawnLocal $ relay' next
47 go True next = spawnLocal $ forward' next
49 collect :: Int -> Process ()
50 collect !n
51 | n == 0 = return ()
52 | otherwise = do
53 receiveWait [
54 matchIf (\(a, b) -> a == "foobar" && b == "baz")
55 (\_ -> return ())
56 , matchAny (\_ -> error "unexpected input!")
57 ]
58 collect (n - 1)
60defaultOptions :: Options
61defaultOptions = Options
62 { optRingSize = 10
63 , optIterations = 100
64 , optForward = False
65 , optParallel = False
66 }
68options :: [OptDescr (Options -> Options)]
69options =
70 [ Option ['s'] ["ring-size"] (OptArg optSz "SIZE") "# of processes in ring"
71 , Option ['i'] ["iterations"] (OptArg optMsgCnt "ITER") "# of times to send"
72 , Option ['f'] ["forward"]
73 (NoArg (\opts -> opts { optForward = True }))
74 "use `forward' instead of send - default = False"
75 , Option ['p'] ["parallel"]
76 (NoArg (\opts -> opts { optForward = True }))
77 "send in parallel and consume sequentially - default = False"
78 ]
80optMsgCnt :: Maybe String -> Options -> Options
81optMsgCnt Nothing opts = opts
82optMsgCnt (Just c) opts = opts { optIterations = ((read c) :: Int) }
84optSz :: Maybe String -> Options -> Options
85optSz Nothing opts = opts
86optSz (Just s) opts = opts { optRingSize = ((read s) :: Int) }
88parseArgv :: [String] -> IO (Options, [String])
89parseArgv argv = do
90 pn <- getProgName
91 case getOpt Permute options argv of
92 (o,n,[] ) -> return (foldl (flip id) defaultOptions o, n)
93 (_,_,errs) -> ioError (userError (concat errs ++ usageInfo (header pn) options))
94 where header pn' = "Usage: " ++ pn' ++ " [OPTION...]"
96main :: IO ()
97main = do
98 argv <- getArgs
99 (opt, _) <- parseArgv argv
100 putStrLn $ "options: " ++ (show opt)
101 Right transport <- createTransport "127.0.0.1" "8090" defaultTCPParameters
102 node <- newLocalNode transport initRemoteTable
103 catch (runProcess node $ initialProcess opt)
104 (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))