git.iko.soy/distributed-process
on master help sign in
Update ProcessRing benchmark to support 'unsafe' sends
2ccf413Tim Watson
benchmarks/ProcessRing.hs+21−13
compare with master
unifiedsplit
@@ -2,11 +2,12 @@
2
2
3To run the benchmarks, select a value for the ring size (sz) and
3To run the benchmarks, select a value for the ring size (sz) and
4the number of times to send a message around the ring
4the number of times to send a message around the ring
5+
5-}
6-}
6import Control.Monad
7import Control.Monad
7import Control.Distributed.Process hiding (catch)
8import Control.Distributed.Process hiding (catch)
8import Control.Distributed.Process.Node
9import Control.Distributed.Process.Node
9−import Control.Exception (SomeException, catch)
10+import Control.Exception (catch, SomeException)
10import Network.Transport.TCP (createTransport, defaultTCPParameters)
11import Network.Transport.TCP (createTransport, defaultTCPParameters)
11import System.Environment
12import System.Environment
12import System.Console.GetOpt
13import System.Console.GetOpt
@@ -16,6 +17,7 @@ data Options = Options
16 , optIterations :: Int
17 , optIterations :: Int
17 , optForward :: Bool
18 , optForward :: Bool
18 , optParallel :: Bool
19 , optParallel :: Bool
20+ , optUnsafe :: Bool
19 } deriving Show
21 } deriving Show
20
22
21initialProcess :: Options -> Process ()
23initialProcess :: Options -> Process ()
@@ -23,28 +25,30 @@ initialProcess op =
23 let ringSz = optRingSize op
25 let ringSz = optRingSize op
24 msgCnt = optIterations op
26 msgCnt = optIterations op
25 fwd = optForward op
27 fwd = optForward op
28+ unsafe = optUnsafe op
26 msg = ("foobar", "baz")
29 msg = ("foobar", "baz")
27 in do
30 in do
28 self <- getSelfPid
31 self <- getSelfPid
29− ring <- makeRing fwd ringSz self
32+ ring <- makeRing fwd unsafe ringSz self
30 forM_ [1..msgCnt] (\_ -> send ring msg)
33 forM_ [1..msgCnt] (\_ -> send ring msg)
31 collect msgCnt
34 collect msgCnt
32− where relay' pid = do
35+ where relay fsend pid = do
33 msg <- expect :: Process (String, String)
36 msg <- expect :: Process (String, String)
34− send pid msg
37+ fsend pid msg
35− relay' pid
38+ relay fsend pid
36
39
37 forward' pid =
40 forward' pid =
38 receiveWait [ matchAny (\m -> forward m pid) ] >> forward' pid
41 receiveWait [ matchAny (\m -> forward m pid) ] >> forward' pid
39
42
40− makeRing :: Bool -> Int -> ProcessId -> Process ProcessId
43+ makeRing :: Bool -> Bool -> Int -> ProcessId -> Process ProcessId
41− makeRing !f !n !pid
44+ makeRing !f !u !n !pid
42− | n == 0 = go f pid
45+ | n == 0 = go f u pid
43− | otherwise = go f pid >>= makeRing f (n - 1)
46+ | otherwise = go f u pid >>= makeRing f u (n - 1)
44
47
45− go :: Bool -> ProcessId -> Process ProcessId
48+ go :: Bool -> Bool -> ProcessId -> Process ProcessId
49+ go False False next = spawnLocal $ relay send next
46− go False next = spawnLocal $ relay' next
50+ go False True next = spawnLocal $ relay unsafeSend next
47− go True next = spawnLocal $ forward' next
51+ go True _ next = spawnLocal $ forward' next
48
52
49 collect :: Int -> Process ()
53 collect :: Int -> Process ()
50 collect !n
54 collect !n
@@ -63,6 +67,7 @@ defaultOptions = Options
63 , optIterations = 100
67 , optIterations = 100
64 , optForward = False
68 , optForward = False
65 , optParallel = False
69 , optParallel = False
70+ , optUnsafe = False
66 }
71 }
67
72
68options :: [OptDescr (Options -> Options)]
73options :: [OptDescr (Options -> Options)]
@@ -72,6 +77,9 @@ options =
72 , Option ['f'] ["forward"]
77 , Option ['f'] ["forward"]
73 (NoArg (\opts -> opts { optForward = True }))
78 (NoArg (\opts -> opts { optForward = True }))
74 "use `forward' instead of send - default = False"
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"
75 , Option ['p'] ["parallel"]
83 , Option ['p'] ["parallel"]
76 (NoArg (\opts -> opts { optForward = True }))
84 (NoArg (\opts -> opts { optForward = True }))
77 "send in parallel and consume sequentially - default = False"
85 "send in parallel and consume sequentially - default = False"
@@ -100,6 +108,6 @@ main = do
100 putStrLn $ "options: " ++ (show opt)
108 putStrLn $ "options: " ++ (show opt)
101 Right transport <- createTransport "127.0.0.1" "8090" defaultTCPParameters
109 Right transport <- createTransport "127.0.0.1" "8090" defaultTCPParameters
102 node <- newLocalNode transport initRemoteTable
110 node <- newLocalNode transport initRemoteTable
103− catch (runProcess node $ initialProcess opt)
111+ catch (void $ runProcess node $ initialProcess opt)
104 (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))
112 (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))
105
113