b1854f5 · distributed-process

distributed-process-demos/distributed-process-demos.cabal+26

compare with master

unifiedsplit

@@ -49,6 +49,32 @@ executable distributed-process-azure-masterslave

49 else

50 buildable: False

51

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+

52executable distributed-process-simplelocalnet-workstealing

78executable distributed-process-simplelocalnet-workstealing

53 main-is: SimpleLocalnet.hs

79 main-is: SimpleLocalnet.hs

54 hs-source-dirs: src/WorkStealing

80 hs-source-dirs: src/WorkStealing

distributed-process-demos/src/MasterSlave/MasterSlave.hs+4−9

@@ -10,10 +10,8 @@ fib = go (0, 1)

10 go (!a, !b) !n | n == 0 = a

11 | otherwise = go (b, a + b) (n - 1)

12

13−slave :: () -> Process ()

13+slave :: (ProcessId, Integer) -> Process ()

14−slave () = forever $ do

15− (pid, n) <- expect

16− send pid (fib n)

14+slave (pid, n) = send pid (fib n)

17

15

18remotable ['slave]

16remotable ['slave]

19

17

@@ -21,12 +19,9 @@ master :: Integer -> [NodeId] -> Process Integer

21master n slaves = do

19master n slaves = do

22 us <- getSelfPid

20 us <- getSelfPid

23

21

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

22 -- Distribute 1 .. n amongst the slave processes

28− forM_ (zip [1 .. n] (cycle slaveProcesses)) $ \(m, them) ->

23+ forM_ (zip [1 .. n] (cycle slaves)) $ \(m, them) ->

29− send them (us, m)

24+ spawn them ($(mkClosure 'slave) (us, m))

30

25

31 -- Wait for the result

26 -- Wait for the result

32 partials <- replicateM (fromInteger n) (expect :: Process Integer)

27 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)