git.iko.soy/distributed-process
on master help sign in
Add typed version of the work-pushing skeleton
3930a95Edsko de Vries
distributed-process-demos/distributed-process-demos.cabal+26
compare with master
unifiedsplit
@@ -75,6 +75,32 @@ executable distributed-process-azure-workpushing
75 else
75 else
76 buildable: False
76 buildable: False
77
77
78+executable distributed-process-simplelocalnet-typedworkpushing
79+ main-is: SimpleLocalnet.hs
80+ hs-source-dirs: src/TypedWorkPushing src/Common
81+ build-depends: base ==4.5.*,
82+ distributed-process >= 0.3.2
83+ extensions: BangPatterns, CPP, TemplateHaskell
84+ ghc-options: -Wall
85+ if flag(use-simplelocalnet)
86+ build-depends: distributed-process-simplelocalnet
87+ cpp-options: -DUSE_SIMPLELOCALNET
88+ else
89+ buildable: False
90+
91+executable distributed-process-azure-typedworkpushing
92+ main-is: Azure.hs
93+ hs-source-dirs: src/TypedWorkPushing src/Common
94+ build-depends: base ==4.5.*,
95+ distributed-process >= 0.3.2
96+ extensions: BangPatterns, CPP, TemplateHaskell
97+ ghc-options: -Wall
98+ if flag(use-azure)
99+ build-depends: distributed-process-azure
100+ cpp-options: -DUSE_AZURE
101+ else
102+ buildable: False
103+
78executable distributed-process-simplelocalnet-workstealing
104executable distributed-process-simplelocalnet-workstealing
79 main-is: SimpleLocalnet.hs
105 main-is: SimpleLocalnet.hs
80 hs-source-dirs: src/WorkStealing src/Common
106 hs-source-dirs: src/WorkStealing src/Common
distributed-process-demos/src/TypedWorkPushing/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/TypedWorkPushing/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 TypedWorkPushing
6+
7+rtable :: RemoteTable
8+rtable = TypedWorkPushing.__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 <- TypedWorkPushing.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/TypedWorkPushing/TypedWorkPushing.hsnew file+41
@@ -0,0 +1,41 @@
1+module TypedWorkPushing where
2+
3+import Control.Monad
4+import Control.Distributed.Process
5+import Control.Distributed.Process.Closure
6+import PrimeFactors
7+
8+slave :: SendPort Integer -> ReceivePort Integer -> Process ()
9+slave results todo = forever $ do
10+ n <- receiveChan todo
11+ sendChan results (numPrimeFactors n)
12+
13+sdictInteger :: SerializableDict Integer
14+sdictInteger = SerializableDict
15+
16+remotable ['slave, 'sdictInteger]
17+
18+-- | Wait for n integers and sum them all up
19+sumIntegers :: ReceivePort Integer -> Int -> Process Integer
20+sumIntegers rport = go 0
21+ where
22+ go :: Integer -> Int -> Process Integer
23+ go !acc 0 = return acc
24+ go !acc n = do
25+ m <- receiveChan rport
26+ go (acc + m) (n - 1)
27+
28+master :: Integer -> [NodeId] -> Process Integer
29+master n slaves = do
30+ (sport, rport) <- newChan
31+
32+ -- Start slave processes
33+ slaveProcesses <- forM slaves $ \nid ->
34+ spawnChannel $(mkStatic 'sdictInteger) nid ($(mkClosure 'slave) sport)
35+
36+ -- Distribute 1 .. n amongst the slave processes
37+ spawnLocal $ forM_ (zip [1 .. n] (cycle slaveProcesses)) $
38+ \(m, them) -> sendChan them m
39+
40+ -- Wait for the result
41+ sumIntegers rport (fromIntegral n)