git.iko.soy/distributed-process
on master help sign in
Distributed Fib on Azure working!
129db28Edsko de Vries
distributed-process-azure/demos/Fib.hs+26−92
compare with master
unifiedsplit
@@ -1,8 +1,6 @@
1{-# LANGUAGE TemplateHaskell #-}
1{-# LANGUAGE TemplateHaskell #-}
2import System.Environment (getArgs)
2import System.Environment (getArgs)
3import System.Random (randomIO)
3import System.Random (randomIO)
4−import Data.Binary (encode)
5−import Data.ByteString.Lazy (ByteString)
6import Control.Monad (forM)
4import Control.Monad (forM)
7import Control.Monad.IO.Class (liftIO)
5import Control.Monad.IO.Class (liftIO)
8import Control.Distributed.Process
6import Control.Distributed.Process
@@ -11,76 +9,48 @@ import Control.Distributed.Process
11 , SendPort
9 , SendPort
12 , newChan
10 , newChan
13 , sendChan
11 , sendChan
14− , Closure
15 , spawn
12 , spawn
16 , receiveChan
13 , receiveChan
17 , spawnLocal
14 , spawnLocal
18 )
15 )
19−import Control.Distributed.Process.Closure
20− ( SerializableDict(SerializableDict)
21− , staticDecode
22− )
23import Control.Distributed.Process.Backend.Azure
16import Control.Distributed.Process.Backend.Azure
24−import Control.Distributed.Static
17+import Control.Distributed.Process.Closure
25− ( Static
18+ ( remotable
26− , staticLabel
19+ , remotableDecl
27− , closure
20+ , mkClosure
28− , RemoteTable
29− , registerStatic
30− , staticCompose
31 )
21 )
32−import Data.Rank1Dynamic (toDynamic)
33−import Control.Distributed.Process.Closure (remotable, mkClosure)
34−import qualified Language.Haskell.TH as TH
35
22
36randomElement :: [a] -> IO a
23randomElement :: [a] -> IO a
37randomElement xs = do
24randomElement xs = do
38 ix <- randomIO
25 ix <- randomIO
39 return (xs !! (ix `mod` length xs))
26 return (xs !! (ix `mod` length xs))
40
27
41−f :: [TH.Name] -> TH.Q [TH.Dec]
28+remotableDecl [
42−f = remotable
43−
44−remotableDec :: [TH.Dec] -> TH.Q [TH.Dec]
45−remotableDec = undefined
46−
47−remotableDec [ [d| f :: Int |] ]
48−
49−main :: IO ()
50−main = undefined
51−
52−--------------------------------------------------------------------------------
53−-- Auxiliary --
54−--------------------------------------------------------------------------------
55−
56−
57−
58−{-
59−
60−--------------------------------------------------------------------------------
61−-- Distributed Fibonacci --
62−--------------------------------------------------------------------------------
63−
64−dfib :: ([NodeId], Integer, SendPort Integer) -> Process ()
29+ [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;
65−dfib (_, 0, reply) = sendChan reply 0
30+ dfib (_, reply, 0) = sendChan reply 0
66−dfib (_, 1, reply) = sendChan reply 1
31+ dfib (_, reply, 1) = sendChan reply 1
67−dfib (nids, n, reply) = do
32+ dfib (nids, reply, n) = do
68− nid1 <- liftIO $ randomElement nids
33+ nid1 <- liftIO $ randomElement nids
69− nid2 <- liftIO $ randomElement nids
34+ nid2 <- liftIO $ randomElement nids
70− (sport, rport) <- newChan
35+ (sport, rport) <- newChan
71− spawn nid1 $ dfibClosure nids (n - 2) sport
36+ _ <- spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)
72− spawn nid2 $ dfibClosure nids (n - 1) sport
37+ _ <- spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)
73− n1 <- receiveChan rport
38+ n1 <- receiveChan rport
74− n2 <- receiveChan rport
39+ n2 <- receiveChan rport
75− sendChan reply $ n1 + n2
40+ sendChan reply $ n1 + n2
41+ |]
42+ ]
76
43
77remoteFib :: ([NodeId], Integer) -> Backend -> Process ()
44remoteFib :: ([NodeId], Integer) -> Backend -> Process ()
78remoteFib (nids, n) _backend = do
45remoteFib (nids, n) _backend = do
79 (sport, rport) <- newChan
46 (sport, rport) <- newChan
80− spawnLocal $ dfib (nids, n, sport)
47+ _ <- spawnLocal $ dfib (nids, sport, n)
81 fib_n <- receiveChan rport
48 fib_n <- receiveChan rport
49+ mapM_ terminateNode nids
82 remoteSend fib_n
50 remoteSend fib_n
83
51
52+remotable ['remoteFib]
53+
84printResult :: LocalProcess ()
54printResult :: LocalProcess ()
85printResult = do
55printResult = do
86 result <- localExpect :: LocalProcess Integer
56 result <- localExpect :: LocalProcess Integer
@@ -90,7 +60,7 @@ main :: IO ()
90main = do
60main = do
91 args <- getArgs
61 args <- getArgs
92 case args of
62 case args of
93− "onvm":args' -> onVmMain __remoteTable args'
63+ "onvm":args' -> onVmMain (__remoteTable . __remoteTableDecl) args'
94 [sid, x509, pkey, user, cloudService, n] -> do
64 [sid, x509, pkey, user, cloudService, n] -> do
95 params <- defaultAzureParameters sid x509 pkey
65 params <- defaultAzureParameters sid x509 pkey
96 let params' = params { azureSshUserName = user }
66 let params' = params { azureSshUserName = user }
@@ -98,43 +68,7 @@ main = do
98 vms <- findVMs backend
68 vms <- findVMs backend
99 nids <- forM vms $ \vm -> spawnNodeOnVM backend vm "8080"
69 nids <- forM vms $ \vm -> spawnNodeOnVM backend vm "8080"
100 callOnVM backend (head vms) "8081" $
70 callOnVM backend (head vms) "8081" $
101− ProcessPair (remoteFibClosure nids (read n)) printResult
71+ ProcessPair ($(mkClosure 'remoteFib) (nids, read n :: Integer))
72+ printResult
102 _ ->
73 _ ->
103 error "Invalid command line arguments"
74 error "Invalid command line arguments"
104−
105−--------------------------------------------------------------------------------
106−-- Static plumping --
107−--------------------------------------------------------------------------------
108−
109−dfibStatic :: Static (([NodeId], Integer, SendPort Integer) -> Process ())
110−dfibStatic = staticLabel "dfib"
111−
112−dfibDict :: Static (SerializableDict ([NodeId], Integer, SendPort Integer))
113−dfibDict = staticLabel "dfibDict"
114−
115−dfibClosure :: [NodeId] -> Integer -> SendPort Integer -> Closure (Process ())
116−dfibClosure nids n reply = closure decoder (encode (nids, n, reply))
117− where
118− decoder :: Static (ByteString -> Process ())
119− decoder = dfibStatic `staticCompose` staticDecode dfibDict
120−
121−remoteFibStatic :: Static (([NodeId], Integer) -> Backend -> Process ())
122−remoteFibStatic = staticLabel "remoteFib"
123−
124−remoteFibDict :: Static (SerializableDict ([NodeId], Integer))
125−remoteFibDict = staticLabel "remoteFibDict"
126−
127−remoteFibClosure :: [NodeId] -> Integer -> RemoteProcess ()
128−remoteFibClosure nids n = closure decoder (encode (nids, n))
129− where
130− decoder :: Static (ByteString -> Backend -> Process ())
131− decoder = remoteFibStatic `staticCompose` staticDecode remoteFibDict
132−
133−__remoteTable :: RemoteTable -> RemoteTable
134−__remoteTable =
135− registerStatic "dfib" (toDynamic dfib)
136− . registerStatic "dfibDict" (toDynamic (SerializableDict :: SerializableDict ([NodeId], Integer, SendPort Integer)))
137− . registerStatic "remoteFib" (toDynamic remoteFib)
138− . registerStatic "remoteFibDict" (toDynamic (SerializableDict :: SerializableDict ([NodeId], Integer)))
139−
140−-}