git.iko.soy/distributed-process
on master help sign in
Distinguish between master/slave and work-pushing
commitb1854f5e6d18a3406e1467b64491e493aface1ad authorEdsko de Vries <edsko@edsko.net> dateWed 26 Sep 2012, 14:06 (14 years ago) parent6ad0728
5 files changed+124−9
compare with master
unifiedsplit
distributed-process-demos/distributed-process-demos.cabal+26 distributed-process-demos/src/MasterSlave/MasterSlave.hs+4−9 distributed-process-demos/src/WorkPushing/Azure.hs+37 distributed-process-demos/src/WorkPushing/SimpleLocalnet.hs+22 distributed-process-demos/src/WorkPushing/WorkPushing.hs+35
distributed-process-demos/distributed-process-demos.cabal+26
@@ -49,6 +49,32 @@ executable distributed-process-azure-masterslave
4949 else
5050 buildable: False
5151
52+executable distributed-process-simplelocalnet-workpushing
53+ main-is: SimpleLocalnet.hs
54+ hs-source-dirs: src/WorkPushing
55+ build-depends: base ==4.5.*,
56+ distributed-process >= 0.3.2
57+ extensions: BangPatterns, CPP, TemplateHaskell
58+ ghc-options: -Wall
59+ if flag(use-simplelocalnet)
60+ build-depends: distributed-process-simplelocalnet
61+ cpp-options: -DUSE_SIMPLELOCALNET
62+ else
63+ buildable: False
64+
65+executable distributed-process-azure-workpushing
66+ main-is: Azure.hs
67+ hs-source-dirs: src/WorkPushing
68+ build-depends: base ==4.5.*,
69+ distributed-process >= 0.3.2
70+ extensions: BangPatterns, CPP, TemplateHaskell
71+ ghc-options: -Wall
72+ if flag(use-azure)
73+ build-depends: distributed-process-azure
74+ cpp-options: -DUSE_AZURE
75+ else
76+ buildable: False
77+
5278executable distributed-process-simplelocalnet-workstealing
5379 main-is: SimpleLocalnet.hs
5480 hs-source-dirs: src/WorkStealing
distributed-process-demos/src/MasterSlave/MasterSlave.hs+4−9
@@ -10,10 +10,8 @@ fib = go (0, 1)
1010 go (!a, !b) !n | n == 0 = a
1111 | otherwise = go (b, a + b) (n - 1)
1212
13−slave :: () -> Process ()
14−slave () = forever $ do
15− (pid, n) <- expect
16− send pid (fib n)
13+slave :: (ProcessId, Integer) -> Process ()
14+slave (pid, n) = send pid (fib n)
1715
1816remotable ['slave]
1917
@@ -21,12 +19,9 @@ master :: Integer -> [NodeId] -> Process Integer
2119master n slaves = do
2220 us <- getSelfPid
2321
24− -- Start processes on the slaves that compute Fibonacci numbers
25− slaveProcesses <- forM slaves $ flip spawn ($(mkClosure 'slave) ())
26−
2722 -- Distribute 1 .. n amongst the slave processes
28− forM_ (zip [1 .. n] (cycle slaveProcesses)) $ \(m, them) ->
29− send them (us, m)
23+ forM_ (zip [1 .. n] (cycle slaves)) $ \(m, them) ->
24+ spawn them ($(mkClosure 'slave) (us, m))
3025
3126 -- Wait for the result
3227 partials <- replicateM (fromInteger n) (expect :: Process Integer)
distributed-process-demos/src/WorkPushing/Azure.hsnew file+37
@@ -0,0 +1,37 @@
1+import System.Environment (getArgs)
2+import Control.Monad
3+import Control.Distributed.Process
4+import Control.Distributed.Process.Closure
5+import Control.Distributed.Process.Backend.Azure
6+import qualified MasterSlave
7+
8+azureMaster :: ([NodeId], Integer) -> Backend -> Process ()
9+azureMaster (slaves, n) _backend = do
10+ result <- MasterSlave.master n slaves
11+ mapM_ terminateNode slaves
12+ remoteSend result
13+
14+remotable ['azureMaster]
15+
16+printResult :: LocalProcess ()
17+printResult = do
18+ result <- localExpect :: LocalProcess Integer
19+ liftIO $ print result
20+
21+main :: IO ()
22+main = do
23+ args <- getArgs
24+ case args of
25+ "onvm":args' -> onVmMain (__remoteTable . MasterSlave.__remoteTable) args'
26+ [sid, x509, pkey, user, cloudService, n] -> do
27+ params <- defaultAzureParameters sid x509 pkey
28+ let params' = params { azureSshUserName = user }
29+ backend <- initializeBackend params' cloudService
30+ vms <- findVMs backend
31+ print vms
32+ nids <- forM vms $ \vm -> spawnNodeOnVM backend vm "8080"
33+ callOnVM backend (head vms) "8081" $
34+ ProcessPair ($(mkClosure 'azureMaster) (nids, read n :: Integer))
35+ printResult
36+ _ ->
37+ error "Invalid command line arguments"
distributed-process-demos/src/WorkPushing/SimpleLocalnet.hsnew file+22
@@ -0,0 +1,22 @@
1+import System.Environment (getArgs)
2+import Control.Distributed.Process
3+import Control.Distributed.Process.Node (initRemoteTable)
4+import Control.Distributed.Process.Backend.SimpleLocalnet
5+import qualified WorkPushing
6+
7+rtable :: RemoteTable
8+rtable = WorkPushing.__remoteTable initRemoteTable
9+
10+main :: IO ()
11+main = do
12+ args <- getArgs
13+
14+ case args of
15+ ["master", host, port, n] -> do
16+ backend <- initializeBackend host port rtable
17+ startMaster backend $ \slaves -> do
18+ result <- WorkPushing.master (read n) slaves
19+ liftIO $ print result
20+ ["slave", host, port] -> do
21+ backend <- initializeBackend host port rtable
22+ startSlave backend
distributed-process-demos/src/WorkPushing/WorkPushing.hsnew file+35
@@ -0,0 +1,35 @@
1+module WorkPushing where
2+
3+import Control.Monad
4+import Control.Distributed.Process
5+import Control.Distributed.Process.Closure
6+
7+fib :: Integer -> Integer
8+fib = go (0, 1)
9+ where
10+ go (!a, !b) !n | n == 0 = a
11+ | otherwise = go (b, a + b) (n - 1)
12+
13+slave :: () -> Process ()
14+slave () = forever $ do
15+ (pid, n) <- expect
16+ send pid (fib n)
17+
18+remotable ['slave]
19+
20+master :: Integer -> [NodeId] -> Process Integer
21+master n slaves = do
22+ us <- getSelfPid
23+
24+ -- Start processes on the slaves that compute Fibonacci numbers
25+ slaveProcesses <- forM slaves $ flip spawn ($(mkClosure 'slave) ())
26+
27+ -- Distribute 1 .. n amongst the slave processes
28+ forM_ (zip [1 .. n] (cycle slaveProcesses)) $ \(m, them) ->
29+ send them (us, m)
30+
31+ -- Wait for the result
32+ partials <- replicateM (fromInteger n) (expect :: Process Integer)
33+
34+ -- And return the sum
35+ return (sum partials)