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