git.iko.soy/distributed-process
on master help sign in
Add support for (fake) Closures
ab70aa5Edsko de Vries
distributed-process/distributed-process.cabal+27−3
compare with master
unifiedsplit
@@ -18,9 +18,12 @@ Library
18 transformers,
18 transformers,
19 data-accessor,
19 data-accessor,
20 random,
20 random,
21+ template-haskell,
21 network-transport >= 0.1
22 network-transport >= 0.1
22− Exposed-modules: Control.Distributed.Process
23+ Exposed-modules: Control.Distributed.Process,
24+ Control.Distributed.Process.Serializable,
25+ Control.Distributed.Process.Closure
23− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable
26+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell
24 ghc-options: -Wall
27 ghc-options: -Wall
25 HS-Source-Dirs: src
28 HS-Source-Dirs: src
26
29
@@ -36,7 +39,28 @@ Test-Suite TestCH
36 data-accessor,
39 data-accessor,
37 random,
40 random,
38 ansi-terminal,
41 ansi-terminal,
42+ template-haskell,
39 network-transport >= 0.1
43 network-transport >= 0.1
40− Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable
44+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell
45+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
46+ HS-Source-Dirs: tests src
47+
48+-- Separate test suite for tests that involve closures, as this will be replaced
49+-- once we have proper support for 'static'
50+Test-Suite TestClosure
51+ Type: exitcode-stdio-1.0
52+ Main-Is: TestClosure.hs
53+ Build-Depends: base,
54+ bytestring,
55+ containers,
56+ mtl,
57+ binary,
58+ transformers,
59+ data-accessor,
60+ random,
61+ ansi-terminal,
62+ template-haskell,
63+ network-transport >= 0.1
64+ Extensions: ExplicitForAll, ScopedTypeVariables, GeneralizedNewtypeDeriving, FlexibleInstances, UndecidableInstances, MultiParamTypeClasses, DeriveDataTypeable, TemplateHaskell
41 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
65 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
42 HS-Source-Dirs: tests src
66 HS-Source-Dirs: tests src
distributed-process/src/Control/Distributed/Process.hs+9−4
@@ -63,6 +63,7 @@ module Control.Distributed.Process
63 , expect
63 , expect
64 , send
64 , send
65 , getSelfPid
65 , getSelfPid
66+ , RemoteCallMetaData
66 -- * Monitoring and linking
67 -- * Monitoring and linking
67 , link
68 , link
68 , unlink
69 , unlink
@@ -95,6 +96,8 @@ import qualified Data.List as List (delete)
95import Data.Set (Set)
96import Data.Set (Set)
96import qualified Data.Set as Set (empty, insert, delete, member)
97import qualified Data.Set as Set (empty, insert, delete, member)
97import Data.Foldable (forM_)
98import Data.Foldable (forM_)
99+import Data.Dynamic (Dynamic)
100+import Data.Typeable (Typeable)
98import Control.Monad (void)
101import Control.Monad (void)
99import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
102import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
100import Control.Monad.IO.Class (MonadIO, liftIO)
103import Control.Monad.IO.Class (MonadIO, liftIO)
@@ -136,7 +139,8 @@ import qualified Network.Transport as NT ( Transport
136import Data.Accessor ((^.), (^=), (^:))
139import Data.Accessor ((^.), (^=), (^:))
137import System.Random (randomIO)
140import System.Random (randomIO)
138import Control.Distributed.Process.Internal.NodeController (runNodeController)
141import Control.Distributed.Process.Internal.NodeController (runNodeController)
139−import Control.Distributed.Process.Internal ( NodeId(..)
142+import Control.Distributed.Process.Internal ( RemoteCallMetaData
143+ , NodeId(..)
140 , LocalProcessId(..)
144 , LocalProcessId(..)
141 , ProcessId(..)
145 , ProcessId(..)
142 , LocalNode(..)
146 , LocalNode(..)
@@ -166,7 +170,7 @@ import Control.Distributed.Process.Internal ( NodeId(..)
166
170
167-- The Cloud Haskell 'Process' type
171-- The Cloud Haskell 'Process' type
168newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
172newtype Process a = Process { unProcess :: ReaderT LocalProcess IO a }
169− deriving (Functor, Monad, MonadIO, MonadReader LocalProcess)
173+ deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable)
170
174
171--------------------------------------------------------------------------------
175--------------------------------------------------------------------------------
172-- Basic Cloud Haskell API --
176-- Basic Cloud Haskell API --
@@ -241,8 +245,8 @@ expectTimeout timeout = do
241-- Initialization --
245-- Initialization --
242--------------------------------------------------------------------------------
246--------------------------------------------------------------------------------
243
247
244−newLocalNode :: NT.Transport -> IO LocalNode
248+newLocalNode :: NT.Transport -> Map String Dynamic -> IO LocalNode
245−newLocalNode transport = do
249+newLocalNode transport metaData = do
246 mEndPoint <- NT.newEndPoint transport
250 mEndPoint <- NT.newEndPoint transport
247 case mEndPoint of
251 case mEndPoint of
248 Left ex -> throwIO ex
252 Left ex -> throwIO ex
@@ -259,6 +263,7 @@ newLocalNode transport = do
259 , localEndPoint = endPoint
263 , localEndPoint = endPoint
260 , localState = state
264 , localState = state
261 , localCtrlChan = ctrlChan
265 , localCtrlChan = ctrlChan
266+ , localMetaData = metaData
262 }
267 }
263 void . forkIO $ runNodeController node
268 void . forkIO $ runNodeController node
264 void . forkIO $ handleIncomingMessages node
269 void . forkIO $ handleIncomingMessages node
distributed-process/src/Control/Distributed/Process/Closure.hsnew file+214
@@ -0,0 +1,214 @@
1+-- | Implementation of 'Closure' that works around the absence of 'static'.
2+--
3+-- [Example usage]
4+--
5+-- Suppose we have a monomorphic function
6+--
7+-- > addInt :: Int -> Int -> Int
8+-- > addInt x y = x + y
9+--
10+-- Then
11+--
12+-- > remotable ['addInt]
13+--
14+-- creates a function
15+--
16+-- > $(mkClosure 'addInt) :: Int -> Closure (Int -> Int)
17+--
18+-- which can be used to partially apply 'addInt' and turn it into a 'Closure',
19+-- which can be sent across the network. Closures can be deserialized with
20+--
21+-- > unClosure :: Typeable a => Closure a -> Process a
22+--
23+-- The call to 'remotable' will also generate a function
24+--
25+-- > __remoteCallMetaData :: RemoteCallMetaData -> RemoteCallMetaData
26+--
27+-- which can be used to construct the 'RemoteCallMetaData' used to initialize
28+-- Cloud Haskell. You should have one call to 'remotable' per module, and
29+-- compose all created functions when initializing Cloud Haskell:
30+--
31+-- > let metaData = M1.__remoteCallMetaData
32+-- > . M2.__remoteCallMetaData
33+-- > . ...
34+-- > . Mn.__remoteCallMetaData
35+-- > $ Map.empty
36+--
37+-- See Section 6, Faking It, of Towards Haskell in the Cloud for more info.
38+{-# LANGUAGE TemplateHaskell #-}
39+module Control.Distributed.Process.Closure
40+ ( unClosure
41+ , remotable
42+ , mkClosure
43+ ) where
44+
45+import Prelude hiding (lookup)
46+import Data.ByteString.Lazy (ByteString)
47+import Data.Binary (Binary(..), encode, decode)
48+import qualified Data.Map as Map (insert, (!))
49+import Data.Dynamic (toDyn, fromDyn)
50+import Data.Typeable (Typeable)
51+import Control.Applicative ((<$>), (<*>))
52+import Control.Monad.Reader (ask)
53+import Control.Exception (throw)
54+import Language.Haskell.TH ( -- Q monad and operations
55+ Q
56+ , report
57+ , reify
58+ -- Names
59+ , Name
60+ , mkName
61+ , nameBase
62+ -- Algebraic data types
63+ , Dec
64+ , Exp
65+ , Type(AppT, ArrowT)
66+ , Info(VarI)
67+ -- Lifted constructors
68+ -- .. Literals
69+ , stringL
70+ -- .. Patterns
71+ , normalB
72+ , clause
73+ -- .. Expressions
74+ , varE
75+ , litE
76+ -- .. Top-level declarations
77+ , funD
78+ , sigD
79+ )
80+import Control.Distributed.Process (Process)
81+import Control.Distributed.Process.Internal ( RemoteCallMetaData
82+ , LocalProcess(processNode)
83+ , LocalNode(localMetaData)
84+ )
85+
86+-- | A static value is one that is bound at top-level.
87+-- We represent it simply by a string
88+newtype Static a = Static String
89+ deriving (Typeable)
90+
91+-- | A closure is a static value and an encoded environment
92+data Closure a = Closure (Static (ByteString -> a)) ByteString
93+ deriving (Typeable)
94+
95+--------------------------------------------------------------------------------
96+-- Top-level API --
97+--------------------------------------------------------------------------------
98+
99+-- | Create the closure, decoder, and metadata definitions for the given list
100+-- of functions
101+remotable :: [Name] -> Q [Dec]
102+remotable ns = do
103+ (closures, decoders, inserts) <- unzip3 <$> mapM process ns
104+ metaData <- createMetaData inserts
105+ report False "Cloud Haskell '__remoteCallMetaData' generated"
106+ return $ concat closures ++ concat decoders ++ metaData
107+
108+-- | Deserialize a closure
109+unClosure :: Typeable a => Closure a -> Process a
110+unClosure (Closure (Static label) env) = do
111+ dec <- lookupStatic label
112+ return (dec env)
113+
114+-- | Create a closure
115+mkClosure :: Name -> Q Exp
116+mkClosure = varE . closureName
117+
118+--------------------------------------------------------------------------------
119+-- Internal (Cloud Haskell state) --
120+--------------------------------------------------------------------------------
121+
122+lookupStatic :: Typeable a => String -> Process a
123+lookupStatic label = do
124+ metaData <- localMetaData . processNode <$> ask
125+ return $ fromDyn (metaData Map.! label) (throw typeError)
126+ where
127+ typeError = userError "lookupStatic type error"
128+
129+instance Binary (Closure a) where
130+ put (Closure (Static label) env) = put label >> put env
131+ get = Closure <$> (Static <$> get) <*> get
132+
133+--------------------------------------------------------------------------------
134+-- Internal (Template Haskell) --
135+--------------------------------------------------------------------------------
136+
137+-- | Generate the code to add the metadata to the CH runtime
138+createMetaData :: [Q Exp] -> Q [Dec]
139+createMetaData is =
140+ [d| __remoteCallMetaData :: RemoteCallMetaData -> RemoteCallMetaData ;
141+ __remoteCallMetaData = $(compose is)
142+ |]
143+
144+-- | Generate the necessary definitions for one function
145+--
146+-- Given an (f :: a -> b) in module M, create:
147+-- 1. f__closure :: a -> Closure b,
148+-- 2. f__dec :: ByteString -> b
149+-- 3. Map.insert "M.f" (toDyn f__dec)
150+process :: Name -> Q ([Dec], [Dec], Q Exp)
151+process n = do
152+ mType <- getType n
153+ case mType of
154+ Just (origName, ArrowT `AppT` arg `AppT` res) -> do
155+ (closure, label) <- generateClosure origName (return arg) (return res)
156+ decoder <- generateDecoder origName (return res)
157+ let decoderE = varE (decoderName n)
158+ insert = [| Map.insert $(stringE label) (toDyn $decoderE) |]
159+ return (closure, decoder, insert)
160+ _ ->
161+ fail $ "remotable: " ++ show n ++ " is not a function"
162+
163+-- | Generate the closure creator (see 'process')
164+generateClosure :: Name -> Q Type -> Q Type -> Q ([Dec], String)
165+generateClosure n arg res = do
166+ closure <- sequence
167+ [ sigD (closureName n) [t| $arg -> Closure $res |]
168+ , sfnD (closureName n) [| Closure (Static $(stringE label)) . encode |]
169+ ]
170+ return (closure, label)
171+ where
172+ label :: String
173+ label = show $ n
174+
175+-- | Generate the decoder (see 'process')
176+generateDecoder :: Name -> Q Type -> Q [Dec]
177+generateDecoder n res = do
178+ sequence [ sigD (decoderName n) [t| ByteString -> $res |]
179+ , sfnD (decoderName n) [| $(varE n) . decode |]
180+ ]
181+
182+-- | The name for the function that generates the closure
183+closureName :: Name -> Name
184+closureName n = mkName $ nameBase n ++ "__closure"
185+
186+-- | The name for the decoder
187+decoderName :: Name -> Name
188+decoderName n = mkName $ nameBase n ++ "__dec"
189+
190+--------------------------------------------------------------------------------
191+-- Generic Template Haskell auxiliary functions --
192+--------------------------------------------------------------------------------
193+
194+-- | Compose a set of expressions
195+compose :: [Q Exp] -> Q Exp
196+compose [] = [| id |]
197+compose [e] = e
198+compose (e:es) = [| $e . $(compose es) |]
199+
200+-- | Literal string as an expression
201+stringE :: String -> Q Exp
202+stringE = litE . stringL
203+
204+-- | Look up the "original name" (module:name) and type of a top-level function
205+getType :: Name -> Q (Maybe (Name, Type))
206+getType name = do
207+ info <- reify name
208+ case info of
209+ VarI origName typ _ _ -> return $ Just (origName, typ)
210+ _ -> return Nothing
211+
212+-- | Variation on 'funD' which takes a single expression to define the function
213+sfnD :: Name -> Q Exp -> Q Dec
214+sfnD n e = funD n [clause [] (normalB e) []]
distributed-process/src/Control/Distributed/Process/Internal.hs+9−1
@@ -1,5 +1,7 @@
1module Control.Distributed.Process.Internal
1module Control.Distributed.Process.Internal
2+ ( -- * Global CH state
3+ RemoteCallMetaData
2− ( -- * Node and process identifiers
4+ , -- * Node and process identifiers
3 NodeId(..)
5 NodeId(..)
4 , LocalProcessId(..)
6 , LocalProcessId(..)
5 , ProcessId(..)
7 , ProcessId(..)
@@ -45,6 +47,7 @@ import Control.Exception (Exception)
45import Data.Map (Map)
47import Data.Map (Map)
46import Data.Int (Int32)
48import Data.Int (Int32)
47import Data.Typeable (Typeable)
49import Data.Typeable (Typeable)
50+import Data.Dynamic (Dynamic)
48import Data.Binary (Binary, encode, put, get, putWord8, getWord8)
51import Data.Binary (Binary, encode, put, get, putWord8, getWord8)
49import Data.Accessor (Accessor, accessor)
52import Data.Accessor (Accessor, accessor)
50import qualified Data.Accessor.Container as DAC (mapMaybe)
53import qualified Data.Accessor.Container as DAC (mapMaybe)
@@ -91,12 +94,17 @@ data ProcessId = ProcessId
91instance Show ProcessId where
94instance Show ProcessId where
92 show pid = show (processNodeId pid) ++ ":" ++ show (processLocalId pid)
95 show pid = show (processNodeId pid) ++ ":" ++ show (processLocalId pid)
93
96
97+-- | Used to fake 'static' (see paper)
98+type RemoteCallMetaData = Map String Dynamic
99+
94-- | Local nodes
100-- | Local nodes
95data LocalNode = LocalNode
101data LocalNode = LocalNode
96 { localNodeId :: NodeId
102 { localNodeId :: NodeId
97 , localEndPoint :: NT.EndPoint
103 , localEndPoint :: NT.EndPoint
98 , localState :: MVar LocalNodeState
104 , localState :: MVar LocalNodeState
99 , localCtrlChan :: Chan NCMsg
105 , localCtrlChan :: Chan NCMsg
106+ -- TODO: this should be part of the CH state, not the local endpoint state
107+ , localMetaData :: RemoteCallMetaData
100 }
108 }
101
109
102-- | Local node state
110-- | Local node state
distributed-process/tests/TestCH.hs+20−19
@@ -4,6 +4,7 @@ import Prelude hiding (catch)
4import Data.Binary (Binary)
4import Data.Binary (Binary)
5import Data.Typeable (Typeable)
5import Data.Typeable (Typeable)
6import Data.Foldable (forM_)
6import Data.Foldable (forM_)
7+import qualified Data.Map as Map (empty)
7import Control.Concurrent (forkIO, threadDelay)
8import Control.Concurrent (forkIO, threadDelay)
8import Control.Concurrent.MVar ( MVar
9import Control.Concurrent.MVar ( MVar
9 , newEmptyMVar
10 , newEmptyMVar
@@ -43,13 +44,13 @@ testPing transport = do
43
44
44 -- Server
45 -- Server
45 forkIO $ do
46 forkIO $ do
46− localNode <- newLocalNode transport
47+ localNode <- newLocalNode transport Map.empty
47 addr <- forkProcess localNode ping
48 addr <- forkProcess localNode ping
48 putMVar serverAddr addr
49 putMVar serverAddr addr
49
50
50 -- Client
51 -- Client
51 forkIO $ do
52 forkIO $ do
52− localNode <- newLocalNode transport
53+ localNode <- newLocalNode transport Map.empty
53 pingServer <- readMVar serverAddr
54 pingServer <- readMVar serverAddr
54
55
55 let numPings = 10000
56 let numPings = 10000
@@ -117,13 +118,13 @@ testMonitorUnreachable transport mOrL un = do
117 done <- newEmptyMVar
118 done <- newEmptyMVar
118
119
119 forkIO $ do
120 forkIO $ do
120− localNode <- newLocalNode transport
121+ localNode <- newLocalNode transport Map.empty
121 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
122 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
122 closeLocalNode localNode
123 closeLocalNode localNode
123 putMVar deadProcess addr
124 putMVar deadProcess addr
124
125
125 forkIO $ do
126 forkIO $ do
126− localNode <- newLocalNode transport
127+ localNode <- newLocalNode transport Map.empty
127 theirAddr <- readMVar deadProcess
128 theirAddr <- readMVar deadProcess
128 runProcess localNode $
129 runProcess localNode $
129 monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
130 monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done
@@ -138,13 +139,13 @@ testMonitorNormalTermination transport mOrL un = do
138 done <- newEmptyMVar
139 done <- newEmptyMVar
139
140
140 forkIO $ do
141 forkIO $ do
141− localNode <- newLocalNode transport
142+ localNode <- newLocalNode transport Map.empty
142 addr <- forkProcess localNode $
143 addr <- forkProcess localNode $
143 liftIO $ readMVar monitorSetup
144 liftIO $ readMVar monitorSetup
144 putMVar monitoredProcess addr
145 putMVar monitoredProcess addr
145
146
146 forkIO $ do
147 forkIO $ do
147− localNode <- newLocalNode transport
148+ localNode <- newLocalNode transport Map.empty
148 theirAddr <- readMVar monitoredProcess
149 theirAddr <- readMVar monitoredProcess
149 runProcess localNode $
150 runProcess localNode $
150 monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
151 monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done
@@ -161,14 +162,14 @@ testMonitorAbnormalTermination transport mOrL un = do
161 let err = userError "Abnormal termination"
162 let err = userError "Abnormal termination"
162
163
163 forkIO $ do
164 forkIO $ do
164− localNode <- newLocalNode transport
165+ localNode <- newLocalNode transport Map.empty
165 addr <- forkProcess localNode . liftIO $ do
166 addr <- forkProcess localNode . liftIO $ do
166 readMVar monitorSetup
167 readMVar monitorSetup
167 throwIO err
168 throwIO err
168 putMVar monitoredProcess addr
169 putMVar monitoredProcess addr
169
170
170 forkIO $ do
171 forkIO $ do
171− localNode <- newLocalNode transport
172+ localNode <- newLocalNode transport Map.empty
172 theirAddr <- readMVar monitoredProcess
173 theirAddr <- readMVar monitoredProcess
173 runProcess localNode $
174 runProcess localNode $
174 monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
175 monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done
@@ -180,7 +181,7 @@ testMonitorLocalDeadProcess :: NT.Transport -> Bool -> Bool -> IO ()
180testMonitorLocalDeadProcess transport mOrL un = do
181testMonitorLocalDeadProcess transport mOrL un = do
181 processDead <- newEmptyMVar
182 processDead <- newEmptyMVar
182 processAddr <- newEmptyMVar
183 processAddr <- newEmptyMVar
183− localNode <- newLocalNode transport
184+ localNode <- newLocalNode transport Map.empty
184 done <- newEmptyMVar
185 done <- newEmptyMVar
185
186
186 forkIO $ do
187 forkIO $ do
@@ -203,12 +204,12 @@ testMonitorRemoteDeadProcess transport mOrL un = do
203 done <- newEmptyMVar
204 done <- newEmptyMVar
204
205
205 forkIO $ do
206 forkIO $ do
206− localNode <- newLocalNode transport
207+ localNode <- newLocalNode transport Map.empty
207 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
208 addr <- forkProcess localNode . liftIO $ putMVar processDead ()
208 putMVar processAddr addr
209 putMVar processAddr addr
209
210
210 forkIO $ do
211 forkIO $ do
211− localNode <- newLocalNode transport
212+ localNode <- newLocalNode transport Map.empty
212 theirAddr <- readMVar processAddr
213 theirAddr <- readMVar processAddr
213 readMVar processDead
214 readMVar processDead
214 runProcess localNode $ do
215 runProcess localNode $ do
@@ -224,14 +225,14 @@ testMonitorDisconnect transport mOrL un = do
224 done <- newEmptyMVar
225 done <- newEmptyMVar
225
226
226 forkIO $ do
227 forkIO $ do
227− localNode <- newLocalNode transport
228+ localNode <- newLocalNode transport Map.empty
228 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
229 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
229 putMVar processAddr addr
230 putMVar processAddr addr
230 readMVar monitorSetup
231 readMVar monitorSetup
231 NT.closeEndPoint (localEndPoint localNode)
232 NT.closeEndPoint (localEndPoint localNode)
232
233
233 forkIO $ do
234 forkIO $ do
234− localNode <- newLocalNode transport
235+ localNode <- newLocalNode transport Map.empty
235 theirAddr <- readMVar processAddr
236 theirAddr <- readMVar processAddr
236 runProcess localNode $ do
237 runProcess localNode $ do
237 monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
238 monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done
@@ -247,13 +248,13 @@ testMonitor2 transport = do
247 done <- newEmptyMVar
248 done <- newEmptyMVar
248
249
249 forkIO $ do
250 forkIO $ do
250− localNode <- newLocalNode transport
251+ localNode <- newLocalNode transport Map.empty
251 addr <- forkProcess localNode $ return ()
252 addr <- forkProcess localNode $ return ()
252 closeLocalNode localNode
253 closeLocalNode localNode
253 putMVar deadProcess addr
254 putMVar deadProcess addr
254
255
255 forkIO $ do
256 forkIO $ do
256− localNode <- newLocalNode transport
257+ localNode <- newLocalNode transport Map.empty
257 theirAddr <- readMVar deadProcess
258 theirAddr <- readMVar deadProcess
258 runProcess localNode $ do
259 runProcess localNode $ do
259 monitor theirAddr MaLink
260 monitor theirAddr MaLink
@@ -277,7 +278,7 @@ testMonitor3 transport = do
277 done <- newEmptyMVar
278 done <- newEmptyMVar
278
279
279 forkIO $ do
280 forkIO $ do
280− localNode <- newLocalNode transport
281+ localNode <- newLocalNode transport Map.empty
281 -- TODO: what happens when processes terminate?
282 -- TODO: what happens when processes terminate?
282 addr <- forkProcess localNode $ return ()
283 addr <- forkProcess localNode $ return ()
283 putMVar serverAddr addr
284 putMVar serverAddr addr
@@ -287,7 +288,7 @@ testMonitor3 transport = do
287 putMVar serverDead ()
288 putMVar serverDead ()
288
289
289 forkIO $ do
290 forkIO $ do
290− localNode <- newLocalNode transport
291+ localNode <- newLocalNode transport Map.empty
291 theirAddr <- readMVar serverAddr
292 theirAddr <- readMVar serverAddr
292 runProcess localNode $ do
293 runProcess localNode $ do
293 monitor theirAddr MaMonitor
294 monitor theirAddr MaMonitor
@@ -315,7 +316,7 @@ testMonitor4 transport = do
315 done <- newEmptyMVar
316 done <- newEmptyMVar
316
317
317 forkIO $ do
318 forkIO $ do
318− localNode <- newLocalNode transport
319+ localNode <- newLocalNode transport Map.empty
319 -- TODO: what happens when processes terminate?
320 -- TODO: what happens when processes terminate?
320 addr <- forkProcess localNode $ return ()
321 addr <- forkProcess localNode $ return ()
321 putMVar serverAddr addr
322 putMVar serverAddr addr
@@ -325,7 +326,7 @@ testMonitor4 transport = do
325 putMVar serverDead ()
326 putMVar serverDead ()
326
327
327 forkIO $ do
328 forkIO $ do
328− localNode <- newLocalNode transport
329+ localNode <- newLocalNode transport Map.empty
329 theirAddr <- readMVar serverAddr
330 theirAddr <- readMVar serverAddr
330 runProcess localNode $ do
331 runProcess localNode $ do
331 monitor theirAddr MaMonitor
332 monitor theirAddr MaMonitor
distributed-process/tests/TestClosure.hsnew file+101
@@ -0,0 +1,101 @@
1+module Main where
2+
3+import qualified Data.Map as Map (empty)
4+import Control.Monad.IO.Class (liftIO)
5+import Control.Concurrent (forkIO)
6+import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)
7+import Network.Transport (Transport)
8+import Network.Transport.TCP (createTransport, defaultTCPParameters)
9+import Control.Distributed.Process
10+import Control.Distributed.Process.Closure
11+import TestAuxiliary
12+
13+addInt :: Int -> Int -> Int
14+addInt x y = x + y
15+
16+putInt :: Int -> MVar Int -> IO ()
17+putInt = flip putMVar
18+
19+sendInt :: ProcessId -> Int -> Process ()
20+sendInt = send
21+
22+$(remotable ['addInt, 'putInt, 'sendInt])
23+
24+testSendPureClosure :: Transport -> RemoteCallMetaData -> IO ()
25+testSendPureClosure transport metaData = do
26+ serverAddr <- newEmptyMVar
27+ serverDone <- newEmptyMVar
28+
29+ forkIO $ do
30+ node <- newLocalNode transport metaData
31+ addr <- forkProcess node $ do
32+ cl <- expect
33+ fn <- unClosure cl :: Process (Int -> Int)
34+ 11 <- return $ fn 6
35+ liftIO $ putMVar serverDone ()
36+ putMVar serverAddr addr
37+
38+ forkIO $ do
39+ node <- newLocalNode transport metaData
40+ theirAddr <- readMVar serverAddr
41+ runProcess node $ send theirAddr ($(mkClosure 'addInt) 5)
42+
43+ takeMVar serverDone
44+
45+testSendIOClosure :: Transport -> RemoteCallMetaData -> IO ()
46+testSendIOClosure transport metaData = do
47+ serverAddr <- newEmptyMVar
48+ serverDone <- newEmptyMVar
49+
50+ forkIO $ do
51+ node <- newLocalNode transport metaData
52+ addr <- forkProcess node $ do
53+ cl <- expect
54+ io <- unClosure cl :: Process (MVar Int -> IO ())
55+ liftIO $ do
56+ someMVar <- newEmptyMVar
57+ io someMVar
58+ 5 <- readMVar someMVar
59+ putMVar serverDone ()
60+ putMVar serverAddr addr
61+
62+ forkIO $ do
63+ node <- newLocalNode transport metaData
64+ theirAddr <- readMVar serverAddr
65+ runProcess node $ send theirAddr ($(mkClosure 'putInt) 5)
66+
67+ takeMVar serverDone
68+
69+testSendProcClosure :: Transport -> RemoteCallMetaData -> IO ()
70+testSendProcClosure transport metaData = do
71+ serverAddr <- newEmptyMVar
72+ clientDone <- newEmptyMVar
73+
74+ forkIO $ do
75+ node <- newLocalNode transport metaData
76+ addr <- forkProcess node $ do
77+ cl <- expect
78+ pr <- unClosure cl :: Process (Int -> Process ())
79+ pr 5
80+ putMVar serverAddr addr
81+
82+ forkIO $ do
83+ node <- newLocalNode transport metaData
84+ theirAddr <- readMVar serverAddr
85+ runProcess node $ do
86+ pid <- getSelfPid
87+ send theirAddr ($(mkClosure 'sendInt) pid)
88+ 5 <- expect :: Process Int
89+ liftIO $ putMVar clientDone ()
90+
91+ takeMVar clientDone
92+
93+main :: IO ()
94+main = do
95+ Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
96+ let metaData = __remoteCallMetaData Map.empty
97+ runTests
98+ [ ("SendPureClosure", testSendPureClosure transport metaData)
99+ , ("SendIOClosure", testSendIOClosure transport metaData)
100+ , ("SendProcClosure", testSendProcClosure transport metaData)
101+ ]