git.iko.soy/distributed-process
on master help sign in
Various improvements to static/closure/TH support
36ba75dEdsko de Vries
distributed-process-azure/demos/Conway.hsdeleted−51
compare with master
unifiedsplit
@@ -1,51 +0,0 @@
1−{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-}
2−import System.Environment (getArgs)
3−import Data.Data (Typeable, Data)
4−import Data.Binary (Binary(get, put))
5−import Data.Binary.Generic (getGeneric, putGeneric)
6−import Control.Monad (forever)
7−import Control.Monad.IO.Class (liftIO)
8−import Control.Distributed.Process
9− ( Process
10− , expect
11− )
12−import Control.Distributed.Process.Closure
13− ( remotable
14− , mkClosure
15− )
16−import Control.Distributed.Process.Backend.Azure
17−
18−data ControllerMsg =
19− ControllerExit
20− deriving (Typeable, Data)
21−
22−instance Binary ControllerMsg where
23− get = getGeneric
24− put = putGeneric
25−
26−conwayStart :: () -> Backend -> Process ()
27−conwayStart () backend = do
28− vms <- liftIO $ findVMs backend
29− remoteSend (show vms)
30−
31−remotable ['conwayStart]
32−
33−echo :: LocalProcess ()
34−echo = forever $ do
35− msg <- localExpect
36− liftIO $ putStrLn msg
37−
38−main :: IO ()
39−main = do
40− args <- getArgs
41− case args of
42− "onvm":args' -> onVmMain __remoteTable args'
43− cmd:sid:x509:pkey:user:cloudService:virtualMachine:port:_ -> do
44− params <- defaultAzureParameters sid x509 pkey
45− let params' = params { azureSshUserName = user }
46− backend <- initializeBackend params' cloudService
47− Just vm <- findNamedVM backend virtualMachine
48− case cmd of
49− "start" -> callOnVM backend vm port $
50− ProcessPair ($(mkClosure 'conwayStart) ())
51− echo
distributed-process-azure/demos/Fib.hsnew file+140
@@ -0,0 +1,140 @@
1+{-# LANGUAGE TemplateHaskell #-}
2+import System.Environment (getArgs)
3+import System.Random (randomIO)
4+import Data.Binary (encode)
5+import Data.ByteString.Lazy (ByteString)
6+import Control.Monad (forM)
7+import Control.Monad.IO.Class (liftIO)
8+import Control.Distributed.Process
9+ ( Process
10+ , NodeId
11+ , SendPort
12+ , newChan
13+ , sendChan
14+ , Closure
15+ , spawn
16+ , receiveChan
17+ , spawnLocal
18+ )
19+import Control.Distributed.Process.Closure
20+ ( SerializableDict(SerializableDict)
21+ , staticDecode
22+ )
23+import Control.Distributed.Process.Backend.Azure
24+import Control.Distributed.Static
25+ ( Static
26+ , staticLabel
27+ , closure
28+ , RemoteTable
29+ , registerStatic
30+ , staticCompose
31+ )
32+import Data.Rank1Dynamic (toDynamic)
33+import Control.Distributed.Process.Closure (remotable, mkClosure)
34+import qualified Language.Haskell.TH as TH
35+
36+randomElement :: [a] -> IO a
37+randomElement xs = do
38+ ix <- randomIO
39+ return (xs !! (ix `mod` length xs))
40+
41+f :: [TH.Name] -> TH.Q [TH.Dec]
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 ()
65+dfib (_, 0, reply) = sendChan reply 0
66+dfib (_, 1, reply) = sendChan reply 1
67+dfib (nids, n, reply) = do
68+ nid1 <- liftIO $ randomElement nids
69+ nid2 <- liftIO $ randomElement nids
70+ (sport, rport) <- newChan
71+ spawn nid1 $ dfibClosure nids (n - 2) sport
72+ spawn nid2 $ dfibClosure nids (n - 1) sport
73+ n1 <- receiveChan rport
74+ n2 <- receiveChan rport
75+ sendChan reply $ n1 + n2
76+
77+remoteFib :: ([NodeId], Integer) -> Backend -> Process ()
78+remoteFib (nids, n) _backend = do
79+ (sport, rport) <- newChan
80+ spawnLocal $ dfib (nids, n, sport)
81+ fib_n <- receiveChan rport
82+ remoteSend fib_n
83+
84+printResult :: LocalProcess ()
85+printResult = do
86+ result <- localExpect :: LocalProcess Integer
87+ liftIO $ print result
88+
89+main :: IO ()
90+main = do
91+ args <- getArgs
92+ case args of
93+ "onvm":args' -> onVmMain __remoteTable args'
94+ [sid, x509, pkey, user, cloudService, n] -> do
95+ params <- defaultAzureParameters sid x509 pkey
96+ let params' = params { azureSshUserName = user }
97+ backend <- initializeBackend params' cloudService
98+ vms <- findVMs backend
99+ nids <- forM vms $ \vm -> spawnNodeOnVM backend vm "8080"
100+ callOnVM backend (head vms) "8081" $
101+ ProcessPair (remoteFibClosure nids (read n)) printResult
102+ _ ->
103+ 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+-}
distributed-process-azure/distributed-process-azure.cabal+15−10
@@ -30,15 +30,17 @@ Library
30 libssh2 >= 0.2 && < 0.3,
30 libssh2 >= 0.2 && < 0.3,
31 pureMD5 >= 2.1 && < 2.2,
31 pureMD5 >= 2.1 && < 2.2,
32 bytestring >= 0.9 && < 0.11,
32 bytestring >= 0.9 && < 0.11,
33− distributed-process >= 0.2.3 && < 0.3,
33+ distributed-process >= 0.3.2 && < 0.4,
34 binary >= 0.5 && < 0.6,
34 binary >= 0.5 && < 0.6,
35− network-transport-tcp >= 0.2 && < 0.3,
35+ network-transport-tcp >= 0.3 && < 0.4,
36 optparse-applicative >= 0.2 && < 0.4,
36 optparse-applicative >= 0.2 && < 0.4,
37 transformers >= 0.3 && < 0.4,
37 transformers >= 0.3 && < 0.4,
38 certificate >= 1.2.4 && < 1.3,
38 certificate >= 1.2.4 && < 1.3,
39 unix >= 2.5 && < 2.6,
39 unix >= 2.5 && < 2.6,
40− network-transport >= 0.2 && < 0.3,
40+ network-transport >= 0.3 && < 0.4,
41− mtl >= 2.1 && < 2.2
41+ mtl >= 2.1 && < 2.2,
42+ rank1dynamic >= 0.1 && < 0.2,
43+ distributed-static >= 0.2 && < 0.3
42 Exposed-modules: Control.Distributed.Process.Backend.Azure
44 Exposed-modules: Control.Distributed.Process.Backend.Azure
43 Extensions: ViewPatterns,
45 Extensions: ViewPatterns,
44 RankNTypes,
46 RankNTypes,
@@ -53,7 +55,7 @@ Executable cloud-haskell-azure-echo
53 if flag(build-demos)
55 if flag(build-demos)
54 Build-Depends: base >= 4.4 && < 5,
56 Build-Depends: base >= 4.4 && < 5,
55 distributed-process-azure >= 0.1 && < 0.2,
57 distributed-process-azure >= 0.1 && < 0.2,
56− distributed-process >= 0.2 && < 0.3,
58+ distributed-process >= 0.3.2 && < 0.4,
57 transformers >= 0.3 && < 0.4
59 transformers >= 0.3 && < 0.4
58 else
60 else
59 buildable: False
61 buildable: False
@@ -64,7 +66,7 @@ Executable cloud-haskell-azure-ping
64 if flag(build-demos)
66 if flag(build-demos)
65 Build-Depends: base >= 4.4 && < 5,
67 Build-Depends: base >= 4.4 && < 5,
66 distributed-process-azure >= 0.1 && < 0.2,
68 distributed-process-azure >= 0.1 && < 0.2,
67− distributed-process >= 0.2 && < 0.3,
69+ distributed-process >= 0.3.2 && < 0.4,
68 transformers >= 0.3 && < 0.4,
70 transformers >= 0.3 && < 0.4,
69 bytestring >= 0.9 && < 0.11,
71 bytestring >= 0.9 && < 0.11,
70 binary >= 0.5 && < 0.6,
72 binary >= 0.5 && < 0.6,
@@ -75,16 +77,19 @@ Executable cloud-haskell-azure-ping
75 Main-Is: demos/Ping.hs
77 Main-Is: demos/Ping.hs
76 ghc-options: -Wall
78 ghc-options: -Wall
77
79
78−Executable cloud-haskell-azure-conway
80+Executable cloud-haskell-azure-fib
79 if flag(build-demos)
81 if flag(build-demos)
80 Build-Depends: base >= 4.4 && < 5,
82 Build-Depends: base >= 4.4 && < 5,
81 distributed-process-azure >= 0.1 && < 0.2,
83 distributed-process-azure >= 0.1 && < 0.2,
82− distributed-process >= 0.2 && < 0.3,
84+ distributed-process >= 0.3.2 && < 0.4,
83 transformers >= 0.3 && < 0.4,
85 transformers >= 0.3 && < 0.4,
84 bytestring >= 0.9 && < 0.11,
86 bytestring >= 0.9 && < 0.11,
85 binary >= 0.5 && < 0.6,
87 binary >= 0.5 && < 0.6,
86− binary-generic >= 0.2 && < 0.3
88+ binary-generic >= 0.2 && < 0.3,
89+ rank1dynamic >= 0.1 && < 0.2,
90+ distributed-static >= 0.2 && < 0.3,
91+ random >= 1.0 && < 1.1
87 else
92 else
88 buildable: False
93 buildable: False
89− Main-Is: demos/Conway.hs
94+ Main-Is: demos/Fib.hs
90 ghc-options: -Wall
95 ghc-options: -Wall
distributed-process-azure/src/Control/Distributed/Process/Backend/Azure.hs+62−2
@@ -413,6 +413,9 @@ module Control.Distributed.Process.Backend.Azure
413 , localSend
413 , localSend
414 , localExpect
414 , localExpect
415 , remoteSend
415 , remoteSend
416+ -- * High-level API
417+ , spawnNodeOnVM
418+ , terminateNode
416 ) where
419 ) where
417
420
418import Prelude hiding (catch)
421import Prelude hiding (catch)
@@ -431,7 +434,7 @@ import System.IO
431 )
434 )
432import qualified System.Posix.Process as Posix (forkProcess, createSession)
435import qualified System.Posix.Process as Posix (forkProcess, createSession)
433import Data.Maybe (listToMaybe)
436import Data.Maybe (listToMaybe)
434−import Data.Binary (Binary(get, put), encode, decode)
437+import Data.Binary (Binary(get, put), encode, decode, getWord8, putWord8)
435import Data.Digest.Pure.MD5 (md5, MD5Digest)
438import Data.Digest.Pure.MD5 (md5, MD5Digest)
436import qualified Data.ByteString as BSS
439import qualified Data.ByteString as BSS
437 ( ByteString
440 ( ByteString
@@ -510,6 +513,11 @@ import Control.Distributed.Process
510 , unClosure
513 , unClosure
511 , ProcessId
514 , ProcessId
512 , getSelfPid
515 , getSelfPid
516+ , NodeId
517+ , processNodeId
518+ , register
519+ , expect
520+ , nsendRemote
513 )
521 )
514import Control.Distributed.Process.Serializable (Serializable)
522import Control.Distributed.Process.Serializable (Serializable)
515import qualified Control.Distributed.Process.Internal.Types as CH
523import qualified Control.Distributed.Process.Internal.Types as CH
@@ -530,6 +538,15 @@ import Control.Distributed.Process.Internal.CQueue (CQueue, enqueue)
530import Network.Transport.TCP (createTransport, defaultTCPParameters)
538import Network.Transport.TCP (createTransport, defaultTCPParameters)
531import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength)
539import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength)
532
540
541+-- Static
542+import Control.Distributed.Static
543+ ( Static
544+ , registerStatic
545+ , staticClosure
546+ , staticLabel
547+ )
548+import Data.Rank1Dynamic (toDynamic)
549+
533-- | Azure backend
550-- | Azure backend
534data Backend = Backend {
551data Backend = Backend {
535 -- | Find virtual machines
552 -- | Find virtual machines
@@ -855,7 +872,7 @@ onVmMain rtable [host, port, cloudService, bg] = do
855 case mTransport of
872 case mTransport of
856 Left err -> remoteThrow err
873 Left err -> remoteThrow err
857 Right transport -> do
874 Right transport -> do
858− node <- newLocalNode transport (rtable initRemoteTable)
875+ node <- newLocalNode transport (rtable . __remoteTable $ initRemoteTable)
859 void . go node $ do
876 void . go node $ do
860 ask >>= liftIO . putMVar lprocMVar
877 ask >>= liftIO . putMVar lprocMVar
861 proc <- unClosure rproc :: Process (Backend -> Process ())
878 proc <- unClosure rproc :: Process (Backend -> Process ())
@@ -902,3 +919,46 @@ readSizeChannel ch = go []
902readIntChannel :: SSH.Channel -> IO Int
919readIntChannel :: SSH.Channel -> IO Int
903readIntChannel ch =
920readIntChannel ch =
904 decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4
921 decodeInt32 . BSS.concat . BSL.toChunks <$> readSizeChannel ch 4
922+
923+--------------------------------------------------------------------------------
924+-- High-level API --
925+--------------------------------------------------------------------------------
926+
927+data ServiceProcessMsg =
928+ ServiceProcessTerminate
929+ deriving Typeable
930+
931+instance Binary ServiceProcessMsg where
932+ put ServiceProcessTerminate = putWord8 0
933+ get = do
934+ header <- getWord8
935+ case header of
936+ 0 -> return ServiceProcessTerminate
937+ _ -> fail "ServiceProcessMsg.get"
938+
939+serviceProcess :: Backend -> Process ()
940+serviceProcess _backend = do
941+ us <- getSelfPid
942+ register "serviceProcess" us
943+ go
944+ where
945+ go = do
946+ msg <- expect
947+ case msg of
948+ ServiceProcessTerminate ->
949+ return ()
950+
951+serviceProcessStatic :: Static (Backend -> Process ())
952+serviceProcessStatic = staticLabel "serviceProcess"
953+
954+-- | Start a new Cloud Haskell node on the given virtual machine
955+spawnNodeOnVM :: Backend -> VirtualMachine -> String -> IO NodeId
956+spawnNodeOnVM backend vm port =
957+ processNodeId <$> spawnOnVM backend vm port (staticClosure serviceProcessStatic)
958+
959+-- | Terminate a node started with 'spawnNodeOnVM'
960+terminateNode :: NodeId -> Process ()
961+terminateNode nid = nsendRemote nid "serviceProcess" ServiceProcessTerminate
962+
963+__remoteTable :: RemoteTable -> RemoteTable
964+__remoteTable = registerStatic "serviceProcess" (toDynamic serviceProcess)
distributed-process/ChangeLog+3
@@ -2,6 +2,9 @@
2
2
3* Improved treatment of network failure, using new failure semantics of
3* Improved treatment of network failure, using new failure semantics of
4Network.Transport.
4Network.Transport.
5+* Make NodeId Typeable
6+* Extend Template Haskell support with "remotableDec" so that you can refer to
7+$(mkClosure 'f) within the body of "f".
5
8
62012-08-16 Edsko de Vries <edsko@well-typed.com> 0.3.1
92012-08-16 Edsko de Vries <edsko@well-typed.com> 0.3.1
7
10
distributed-process/src/Control/Distributed/Process.hs+2
@@ -66,6 +66,7 @@ module Control.Distributed.Process
66 , Closure
66 , Closure
67 , closure
67 , closure
68 , Static
68 , Static
69+ , unStatic
69 , unClosure
70 , unClosure
70 , RemoteTable
71 , RemoteTable
71 -- * Logging
72 -- * Logging
@@ -200,6 +201,7 @@ import Control.Distributed.Process.Internal.Primitives
200 , whereisRemoteAsync
201 , whereisRemoteAsync
201 , nsendRemote
202 , nsendRemote
202 -- Closures
203 -- Closures
204+ , unStatic
203 , unClosure
205 , unClosure
204 -- Exception handling
206 -- Exception handling
205 , catch
207 , catch
distributed-process/src/Control/Distributed/Process/Closure.hs+2
@@ -179,6 +179,7 @@ module Control.Distributed.Process.Closure
179#ifdef TemplateHaskellSupport
179#ifdef TemplateHaskellSupport
180 -- * Template Haskell support for creating static values and closures
180 -- * Template Haskell support for creating static values and closures
181 , remotable
181 , remotable
182+ , remotableDec
182 , mkStatic
183 , mkStatic
183 , mkClosure
184 , mkClosure
184 , functionSDict
185 , functionSDict
@@ -210,6 +211,7 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn
210#ifdef TemplateHaskellSupport
211#ifdef TemplateHaskellSupport
211import Control.Distributed.Process.Internal.Closure.TH
212import Control.Distributed.Process.Internal.Closure.TH
212 ( remotable
213 ( remotable
214+ , remotableDec
213 , mkStatic
215 , mkStatic
214 , functionSDict
216 , functionSDict
215 , functionTDict
217 , functionTDict
distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hs+79−47
@@ -3,6 +3,7 @@
3module Control.Distributed.Process.Internal.Closure.TH
3module Control.Distributed.Process.Internal.Closure.TH
4 ( -- * User-level API
4 ( -- * User-level API
5 remotable
5 remotable
6+ , remotableDec
6 , mkStatic
7 , mkStatic
7 , functionSDict
8 , functionSDict
8 , functionTDict
9 , functionTDict
@@ -20,7 +21,7 @@ import Language.Haskell.TH
20 , mkName
21 , mkName
21 , nameBase
22 , nameBase
22 -- Algebraic data types
23 -- Algebraic data types
23− , Dec
24+ , Dec(SigD)
24 , Exp
25 , Exp
25 , Type(AppT, ForallT, VarT, ArrowT)
26 , Type(AppT, ForallT, VarT, ArrowT)
26 , Info(VarI)
27 , Info(VarI)
@@ -39,6 +40,7 @@ import Language.Haskell.TH
39 , funD
40 , funD
40 , sigD
41 , sigD
41 )
42 )
43+import Data.Maybe (catMaybes)
42import Data.Binary (encode)
44import Data.Binary (encode)
43import Data.Generics (everywhereM, mkM, gmapM)
45import Data.Generics (everywhereM, mkM, gmapM)
44import Data.Rank1Dynamic (toDynamic)
46import Data.Rank1Dynamic (toDynamic)
@@ -69,9 +71,42 @@ import Control.Distributed.Process.Internal.Closure.BuiltIn (staticDecode)
69-- of functions
71-- of functions
70remotable :: [Name] -> Q [Dec]
72remotable :: [Name] -> Q [Dec]
71remotable ns = do
73remotable ns = do
72− (closures, inserts) <- unzip <$> mapM generateDefs ns
74+ types <- mapM getType ns
75+ (closures, inserts) <- unzip <$> mapM generateDefs types
73− rtable <- createMetaData (concat inserts)
76+ rtable <- createMetaData (mkName "__remoteTable") (concat inserts)
77+ return $ concat closures ++ rtable
78+
79+-- | Like 'remotable', but parameterized by the declaration of a function
80+-- instead of the function name. So where for 'remotable' you'd do
81+--
82+-- > f :: T1 -> T2
83+-- > f = ...
84+-- >
85+-- > remotable ['f]
86+--
87+-- with 'remotableDec' you would instead do
88+--
89+-- > remotableDec [
90+-- > [d| f :: T1 -> T2 ;
91+-- > f = ...
92+-- > |]
93+-- > ]
94+--
95+-- 'remotableDec' creates the function specified as well as the various
96+-- dictionaries and static versions that 'remotable' also creates.
97+-- 'remotableDec' is sometimes necessary when you want to refer to, say,
98+-- @$(mkClosure 'f)@ within the definition of @f@ itself.
99+remotableDec :: [Q [Dec]] -> Q [Dec]
100+remotableDec qDecs = do
101+ decs <- concat <$> sequence qDecs
102+ let types = catMaybes (map typeOf decs)
103+ (closures, inserts) <- unzip <$> mapM generateDefs types
104+ rtable <- createMetaData (mkName "__remoteTableDec") (concat inserts)
74− return $ concat closures ++ rtable
105+ return $ decs ++ concat closures ++ rtable
106+ where
107+ typeOf :: Dec -> Maybe (Name, Type)
108+ typeOf (SigD name typ) = Just (name, typ)
109+ typeOf _ = Nothing
75
110
76-- | Construct a static value.
111-- | Construct a static value.
77--
112--
@@ -95,6 +130,10 @@ functionSDict = varE . sdictName
95functionTDict :: Name -> Q Exp
130functionTDict :: Name -> Q Exp
96functionTDict = varE . tdictName
131functionTDict = varE . tdictName
97
132
133+-- | If @f : T1 -> T2@ then @$(mkClosure 'f) :: T1 -> Closure T2@.
134+--
135+-- TODO: The current version of mkClosure is too polymorphic
136+-- (@forall a. Binary a => a -> Closure T2).
98mkClosure :: Name -> Q Exp
137mkClosure :: Name -> Q Exp
99mkClosure n =
138mkClosure n =
100 [| closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n))
139 [| closure ($(mkStatic n) `staticCompose` staticDecode $(functionSDict n))
@@ -106,49 +145,44 @@ mkClosure n =
106--------------------------------------------------------------------------------
145--------------------------------------------------------------------------------
107
146
108-- | Generate the code to add the metadata to the CH runtime
147-- | Generate the code to add the metadata to the CH runtime
109−createMetaData :: [Q Exp] -> Q [Dec]
148+createMetaData :: Name -> [Q Exp] -> Q [Dec]
110−createMetaData is =
149+createMetaData name is =
111− [d| __remoteTable :: RemoteTable -> RemoteTable ;
150+ sequence [ sigD name [t| RemoteTable -> RemoteTable |]
112− __remoteTable = $(compose is)
151+ , sfnD name (compose is)
113− |]
152+ ]
114
153
115−generateDefs :: Name -> Q ([Dec], [Q Exp])
154+generateDefs :: (Name, Type) -> Q ([Dec], [Q Exp])
116−generateDefs n = do
155+generateDefs (origName, fullType) = do
117 proc <- [t| Process |]
156 proc <- [t| Process |]
118− mType <- getType n
119− case mType of
120− Just (origName, typ) -> do
121− let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)
157+ let (typVars, typ') = case fullType of ForallT vars [] mono -> (vars, mono)
122− _ -> ([], typ)
158+ _ -> ([], fullType)
123
159
124− -- The main "static" entry
160+ -- The main "static" entry
125− (static, register) <- makeStatic origName typVars typ'
161+ (static, register) <- makeStatic typVars typ'
126−
162+
127− -- If n :: T1 -> T2, static serializable dictionary for T1
163+ -- If n :: T1 -> T2, static serializable dictionary for T1
128− -- TODO: we should check if arg is an instance of Serializable, but we cannot
164+ -- TODO: we should check if arg is an instance of Serializable, but we cannot
129− -- http://hackage.haskell.org/trac/ghc/ticket/7066
165+ -- http://hackage.haskell.org/trac/ghc/ticket/7066
130− (sdict, registerSDict) <- case (typVars, typ') of
166+ (sdict, registerSDict) <- case (typVars, typ') of
131− ([], ArrowT `AppT` arg `AppT` _res) ->
167+ ([], ArrowT `AppT` arg `AppT` _res) ->
132− makeDict (sdictName origName) arg
168+ makeDict (sdictName origName) arg
133− _ ->
134− return ([], [])
135−
136− -- If n :: T1 -> Process T2, static serializable dictionary for T2
137− -- TODO: check if T2 is serializable (same as above)
138− (tdict, registerTDict) <- case (typVars, typ') of
139− ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc ->
140− makeDict (tdictName origName) res
141− _ ->
142− return ([], [])
143−
144− return ( concat [static, sdict, tdict]
145− , concat [register, registerSDict, registerTDict]
146− )
147 _ ->
169 _ ->
148− fail $ "remotable: " ++ show n ++ " not found"
170+ return ([], [])
171+
172+ -- If n :: T1 -> Process T2, static serializable dictionary for T2
173+ -- TODO: check if T2 is serializable (same as above)
174+ (tdict, registerTDict) <- case (typVars, typ') of
175+ ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc ->
176+ makeDict (tdictName origName) res
177+ _ ->
178+ return ([], [])
179+
180+ return ( concat [static, sdict, tdict]
181+ , concat [register, registerSDict, registerTDict]
182+ )
149 where
183 where
150− makeStatic :: Name -> [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])
184+ makeStatic :: [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])
151− makeStatic origName typVars typ = do
185+ makeStatic typVars typ = do
152 static <- generateStatic origName typVars typ
186 static <- generateStatic origName typVars typ
153 let dyn = case typVars of
187 let dyn = case typVars of
154 [] -> [| toDynamic $(varE origName) |]
188 [] -> [| toDynamic $(varE origName) |]
@@ -190,8 +224,6 @@ monomorphize tvs =
190 Just t -> t
224 Just t -> t
191 applySubst s t = gmapM (mkM (applySubst s)) t
225 applySubst s t = gmapM (mkM (applySubst s)) t
192
226
193−
194−
195-- | Generate a static value
227-- | Generate a static value
196generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec]
228generateStatic :: Name -> [TyVarBndr] -> Type -> Q [Dec]
197generateStatic n xs typ = do
229generateStatic n xs typ = do
@@ -240,12 +272,12 @@ stringE :: String -> Q Exp
240stringE = litE . stringL
272stringE = litE . stringL
241
273
242-- | Look up the "original name" (module:name) and type of a top-level function
274-- | Look up the "original name" (module:name) and type of a top-level function
243−getType :: Name -> Q (Maybe (Name, Type))
275+getType :: Name -> Q (Name, Type)
244getType name = do
276getType name = do
245 info <- reify name
277 info <- reify name
246 case info of
278 case info of
247− VarI origName typ _ _ -> return $ Just (origName, typ)
279+ VarI origName typ _ _ -> return (origName, typ)
248− _ -> return Nothing
280+ _ -> fail $ show name ++ " not found"
249
281
250-- | Variation on 'funD' which takes a single expression to define the function
282-- | Variation on 'funD' which takes a single expression to define the function
251sfnD :: Name -> Q Exp -> Q Dec
283sfnD :: Name -> Q Exp -> Q Dec
distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+13−4
@@ -42,6 +42,7 @@ module Control.Distributed.Process.Internal.Primitives
42 , nsendRemote
42 , nsendRemote
43 -- * Closures
43 -- * Closures
44 , unClosure
44 , unClosure
45+ , unStatic
45 -- * Exception handling
46 -- * Exception handling
46 , catch
47 , catch
47 , mask
48 , mask
@@ -91,9 +92,9 @@ import Control.Concurrent.STM
91import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))
92import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))
92import Control.Distributed.Process.Serializable (Serializable, fingerprint)
93import Control.Distributed.Process.Serializable (Serializable, fingerprint)
93import Data.Accessor ((^.), (^:), (^=))
94import Data.Accessor ((^.), (^:), (^=))
94−import Control.Distributed.Static (Closure)
95+import Control.Distributed.Static (Closure, Static)
95import Data.Rank1Typeable (Typeable)
96import Data.Rank1Typeable (Typeable)
96−import qualified Control.Distributed.Static as Static (unclosure)
97+import qualified Control.Distributed.Static as Static (unstatic, unclosure)
97import Control.Distributed.Process.Internal.Types
98import Control.Distributed.Process.Internal.Types
98 ( NodeId(..)
99 ( NodeId(..)
99 , ProcessId(..)
100 , ProcessId(..)
@@ -516,8 +517,16 @@ nsendRemote nid label msg =
516-- Closures --
517-- Closures --
517--------------------------------------------------------------------------------
518--------------------------------------------------------------------------------
518
519
520+-- | Resolve a static value
521+unStatic :: Typeable a => Static a -> Process a
522+unStatic static = do
523+ rtable <- remoteTable . processNode <$> ask
524+ case Static.unstatic rtable static of
525+ Left err -> fail $ "Could not resolve static value: " ++ err
526+ Right x -> return x
527+
519−-- | Deserialize a closure
528+-- | Resolve a closure
520−unClosure :: forall a. Typeable a => Closure a -> Process a
529+unClosure :: Typeable a => Closure a -> Process a
521unClosure closure = do
530unClosure closure = do
522 rtable <- remoteTable . processNode <$> ask
531 rtable <- remoteTable . processNode <$> ask
523 case Static.unclosure rtable closure of
532 case Static.unclosure rtable closure of
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+1−1
@@ -105,7 +105,7 @@ import Control.Distributed.Static (RemoteTable, Closure)
105
105
106-- | Node identifier
106-- | Node identifier
107newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
107newtype NodeId = NodeId { nodeAddress :: NT.EndPointAddress }
108− deriving (Eq, Ord, Binary)
108+ deriving (Eq, Ord, Binary, Typeable)
109
109
110instance Show NodeId where
110instance Show NodeId where
111 show (NodeId addr) = "nid://" ++ show addr
111 show (NodeId addr) = "nid://" ++ show addr
distributed-process/tests/TestClosure.hs+37−1
@@ -16,6 +16,7 @@ import Control.Concurrent.MVar
16 , newMVar
16 , newMVar
17 )
17 )
18import Control.Applicative ((<$>))
18import Control.Applicative ((<$>))
19+import System.Random (randomIO)
19import Network.Transport (Transport)
20import Network.Transport (Transport)
20import Network.Transport.TCP
21import Network.Transport.TCP
21 ( createTransportExposeInternals
22 ( createTransportExposeInternals
@@ -82,6 +83,27 @@ remotable [ 'factorial
82 , 'signal
83 , 'signal
83 ]
84 ]
84
85
86+randomElement :: [a] -> IO a
87+randomElement xs = do
88+ ix <- randomIO
89+ return (xs !! (ix `mod` length xs))
90+
91+remotableDec [
92+ [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;
93+ dfib (_, reply, 0) = sendChan reply 0
94+ dfib (_, reply, 1) = sendChan reply 1
95+ dfib (nids, reply, n) = do
96+ nid1 <- liftIO $ randomElement nids
97+ nid2 <- liftIO $ randomElement nids
98+ (sport, rport) <- newChan
99+ spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)
100+ spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)
101+ n1 <- receiveChan rport
102+ n2 <- receiveChan rport
103+ sendChan reply $ n1 + n2
104+ |]
105+ ]
106+
85-- Just try creating a static polymorphic value
107-- Just try creating a static polymorphic value
86staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)
108staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)
87 => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))
109 => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))
@@ -362,6 +384,19 @@ simulateNetworkFailure transportInternals fr to = liftIO $ do
362 sClose sock
384 sClose sock
363 threadDelay 10000
385 threadDelay 10000
364
386
387+testFib :: Transport -> RemoteTable -> IO ()
388+testFib transport rtable = do
389+ nodes <- replicateM 4 $ newLocalNode transport rtable
390+ done <- newEmptyMVar
391+
392+ forkProcess (head nodes) $ do
393+ (sport, rport) <- newChan
394+ spawnLocal $ dfib (map localNodeId nodes, sport, 10)
395+ 55 <- receiveChan rport :: Process Integer
396+ liftIO $ putMVar done ()
397+
398+ takeMVar done
399+
365testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()
400testSpawnReconnect :: Transport -> RemoteTable -> TransportInternals -> IO ()
366testSpawnReconnect transport rtable transportInternals = do
401testSpawnReconnect transport rtable transportInternals = do
367 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
402 [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
@@ -392,7 +427,7 @@ testSpawnReconnect transport rtable transportInternals = do
392main :: IO ()
427main :: IO ()
393main = do
428main = do
394 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
429 Right (transport, transportInternals) <- createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters
395− let rtable = __remoteTable initRemoteTable
430+ let rtable = __remoteTable . __remoteTableDec $ initRemoteTable
396 runTests
431 runTests
397 [ ("Unclosure", testUnclosure transport rtable)
432 [ ("Unclosure", testUnclosure transport rtable)
398 , ("Bind", testBind transport rtable)
433 , ("Bind", testBind transport rtable)
@@ -408,5 +443,6 @@ main = do
408 , ("ClosureExpect", testClosureExpect transport rtable)
443 , ("ClosureExpect", testClosureExpect transport rtable)
409 , ("SpawnChannel", testSpawnChannel transport rtable)
444 , ("SpawnChannel", testSpawnChannel transport rtable)
410 , ("TDict", testTDict transport rtable)
445 , ("TDict", testTDict transport rtable)
446+ , ("Fib", testFib transport rtable)
411 , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)
447 , ("SpawnReconnect", testSpawnReconnect transport rtable transportInternals)
412 ]
448 ]