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