de4b010 · distributed-process

git.iko.soy/distributed-process

Improve documentation of C.D.P.Closure

commitde4b0105b8b2386155760c3ba0a18c1efa1853ef authorEdsko de Vries <edsko@edsko.net> dateSat 21 Jul 2012, 16:09 (14 years ago) parent56c3794

10 files changed+414−224

compare with master

unifiedsplit

distributed-process/distributed-process.cabal−1

@@ -52,7 +52,6 @@ Library

5252 Control.Distributed.Process.Internal.CQueue,

5353 Control.Distributed.Process.Internal.Dynamic,

5454 Control.Distributed.Process.Internal.TypeRep,

55− Control.Distributed.Process.Internal.MessageT,

5655 Control.Distributed.Process.Internal.Types,

5756 Control.Distributed.Process.Internal.Closure.Static,

5857 Control.Distributed.Process.Internal.Closure.MkClosure,

distributed-process/src/Control/Distributed/Process.hs+6−5

@@ -89,7 +89,7 @@ module Control.Distributed.Process

8989 , spawnMonitor

9090 , spawnChannel

9191 , DidSpawn(..)

92− -- * Local versions of spawn

92+ -- * Local versions of 'spawn'

9393 , spawnLocal

9494 , spawnChannelLocal

9595 ) where

@@ -101,6 +101,8 @@ import Prelude hiding (catch)

101101

102102import Data.Typeable (Typeable)

103103import Control.Monad.IO.Class (liftIO)

104+import Control.Applicative ((<$>))

105+import Control.Monad.Reader (ask)

104106import Control.Distributed.Process.Internal.Types

105107 ( RemoteTable

106108 , NodeId(..)

@@ -124,7 +126,7 @@ import Control.Distributed.Process.Internal.Types

124126 , SerializableDict(..)

125127 , SendPortId(..)

126128 , WhereIsReply(..)

127− , procMsg

129+ , LocalProcess(processNode)

128130 )

129131import Control.Distributed.Process.Internal.Closure.CP

130132 ( cpSeq

@@ -194,7 +196,6 @@ import Control.Distributed.Process.Internal.Primitives

194196 , spawnAsync

195197 )

196198import Control.Distributed.Process.Serializable (Serializable)

197−import Control.Distributed.Process.Internal.MessageT (getLocalNode)

198199import Control.Distributed.Process.Node (forkProcess)

199200

200201-- INTERNAL NOTES

@@ -369,7 +370,7 @@ spawnChannel dict nid proc = do

369370-- | Spawn a process on the local node

370371spawnLocal :: Process () -> Process ProcessId

371372spawnLocal proc = do

372− node <- procMsg $ getLocalNode

373+ node <- processNode <$> ask

373374 liftIO $ forkProcess node proc

374375

375376-- | Create a new typed channel, spawn a process on the local node, passing it

@@ -378,7 +379,7 @@ spawnChannelLocal :: Serializable a

378379 => (ReceivePort a -> Process ())

379380 -> Process (SendPort a)

380381spawnChannelLocal proc = do

381− node <- procMsg $ getLocalNode

382+ node <- processNode <$> ask

382383 (sport, rport) <- newChan

383384 _ <- liftIO $ forkProcess node (proc rport)

384385 return sport

distributed-process/src/Control/Distributed/Process/Closure.hs+125−58

@@ -12,7 +12,7 @@

1212--

1313-- you can use a Template Haskell splice to create a static version of 'f':

1414--

15−-- > $(mkStatic 'f) :: forall a1 .. an. Static T

15+-- > $(mkStatic 'f) :: forall a1 .. an. (Typeable a1, .., Typeable an) => Static T

1616--

1717-- Every module that you write that contains calls to 'mkStatic' needs to

1818-- have a call to 'remotable':

@@ -35,29 +35,90 @@

3535-- > . Mn.__remoteTable

3636-- > $ initRemoteTable

3737--

38+-- [Dealing with type class qualifiers]

39+--

40+-- Although 'mkStatic' supports polymorphic types, it does not support

41+-- qualified types. For instance, you cannot call 'mkStatic' on

42+--

43+-- > decode :: Serializable a => ByteString -> a

44+--

45+-- Instead, you will need to reify the type class dictionary. Cloud Haskell

46+-- comes with a reified version of 'Serializable':

47+--

48+-- > data SerializableDict a where

49+-- > SerializableDict :: Serializable a => SerializableDict a

50+--

51+-- Using the reified dictionary you can define

52+--

53+-- > decodeDict :: SerializableDict a -> ByteString -> a

54+-- > decodeDict SerializableDict = decode

55+--

56+-- where 'decodeDict' is a normal (unqualified) polymorphic value and hence

57+-- can be passed as an argument to remotable:

58+--

59+-- > $(mkStatic 'decodeDict) :: Typeable a => Static (SerializableDict a -> ByteString -> a)

60+--

3861-- [Composing static values]

3962--

40−-- We generalize the notion of 'static' as described in the paper, and also

41−-- provide

63+-- The version of 'static' provided by this implementation of Cloud Haskell is

64+-- strictly more expressive than the one proposed in the paper, and additionally

65+-- supports

4266--

4367-- > staticApply :: Static (a -> b) -> Static a -> Static b

4468--

45−-- This makes it possible to define a rich set of combinators on 'static'

46−-- values, a number of which are provided in this module.

69+-- This is extremely useful. For example, Cloud Haskell comes with

70+-- 'staticDecode' defined as

71+--

72+-- > staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)

73+-- > staticDecode dict = $(mkStatic 'decodeDict) `staticApply` dict

74+--

75+-- 'staticDecode' is used when defining closures (see below), and makes

76+-- essential use of 'staticApply'.

77+--

78+-- Support for 'staticApply' also makes it possible to define a rich set of

79+-- combinators on 'static' values, a number of which are provided in this

80+-- module.

81+--

82+-- [Static serialization dictionaries]

83+--

84+-- Many Cloud Haskell primitives (like 'staticDecode', above) require static

85+-- serialization dictionaries. In principle these dictionaries require nothing

86+-- special; for instance, given some serializable type 'T' you can define

87+--

88+-- > sdictT :: SerializableDict T

89+-- > sdictT = SerializableDict

90+--

91+-- and then have

92+--

93+-- > $(mkStatic 'sdictT) :: Static (SerializableDict T)

94+--

95+-- However, since these dictionaries are so frequently required Cloud Haskell

96+-- provides special support for them. When you call 'remotable' on a

97+-- /monomorphic/ function @f :: T1 -> T2@

98+--

99+-- > remotable ['f]

100+--

101+-- then a serialization dictionary is automatically created for you, which you

102+-- can access with

103+--

104+-- > $(functionSDict 'f) :: Static (SerializableDict T1)

105+--

106+-- In addition, if @f :: T1 -> Process T2@, then a second dictionary is created

107+--

108+-- > $(functionTDict 'f) :: Static (SerializableDict T2)

47109--

48110-- [Closures]

49111--

50112-- Suppose you have a process

51113--

52−-- > factorial :: Int -> Process Int

114+-- > isPrime :: Integer -> Process Bool

53115--

54−-- Then you can use the supplied Template Haskell function 'mkClosure' to define

116+-- Then

55117--

56−-- > factorialClosure :: Int -> Closure (Process Int)

57−-- > factorialClosure = $(mkClosure 'factorial)

118+-- > $(mkClosure 'isPrime) :: Integer -> Closure (Process Bool)

58119--

59−-- You can then pass 'factorialClosure n' to 'spawn', for example, to have a

60−-- remote node compute a factorial number.

120+-- which you can then 'call', for example, to have a remote node check if

121+-- a number is prime.

61122--

62123-- In general, if you have a /monomorphic/ function

63124--

@@ -76,55 +137,17 @@

76137--

77138-- > data Closure a = Closure (Static (ByteString -> a)) ByteString

78139--

79−-- The splice @$(mkClosure 'factorial)@ above expands to (prettified a bit):

140+-- The splice @$(mkClosure 'isPrime)@ above expands to (prettified a bit):

80141--

81−-- > factorialClosure :: Int -> Closure (Process Int)

82−-- > factorialClosure n = Closure decoder (encode n)

83−-- > where

84−-- > decoder :: Static (ByteString -> Process Int)

85−-- > decoder = $(mkStatic 'factorial)

142+-- > let decoder :: Static (ByteString -> Process Bool)

143+-- > decoder = $(mkStatic 'isPrime)

86144-- > `staticCompose`

87−-- > staticDecode $(functionSDict 'factorial)

88−--

89−-- 'mkStatic' we have already seen:

90−--

91−-- > $(mkStatic 'factorial) :: Static (Int -> Process Int)

92−--

93−-- 'staticCompose' is function composition on static functions. 'staticDecode'

94−-- has type (**)

95−--

96−-- > staticDecode :: Typeable a

97−-- > => Static (SerializableDict a) -> Static (ByteString -> a)

98−--

99−-- and gives you a static decoder, given a static Serializable dictionary.

100−-- 'SerializableDict' is a reified type class dictionary, and defined simply as

101−--

102−-- > data SerializableDict a where

103−-- > SerializableDict :: Serializable a => SerializableDict a

104−--

105−-- That means that for any serialziable type 'T', you can define

106−--

107−-- > sdictForMyType :: SerializableDict T

108−-- > sdictForMyType = SerializableDict

109−--

110−-- and then use

145+-- > staticDecode $(functionSDict 'isPrime)

146+-- > in Closure decoder (encode n)

111147--

112−-- > $(mkStatic 'sdictForMyType) :: Static (SerializableDict T)

113−--

114−-- to obtain a static serializable dictionary for 'T' (make sure to pass

115−-- 'sdictForMyType' to 'remotable').

116−--

117−-- However, since these serialization dictionaries are so frequently required,

118−-- when you call 'remotable' on a monomorphic function @f : T1 -> T2@

119−--

120−-- > remotable ['f]

121−--

122−-- then a serialization dictionary is automatically created for you, which you

123−-- can access with

124−--

125−-- > $(functionSDict 'f) :: Static (SerializableDict T1)

126−--

127−-- This is the dictionary that 'mkClosure' uses.

148+-- where 'staticCompose' is composition of static functions. Note that

149+-- 'mkClosure' makes use of the static serialization dictionary

150+-- ('functionSDict') created by 'remotable'.

128151--

129152-- [Combinators on Closures]

130153--

@@ -133,6 +156,47 @@

133156-- the most important of which is 'cpBind'. Have a look at the implementation

134157-- of 'Control.Distributed.Process.call' for an example use.

135158--

159+-- [Example]

160+--

161+-- Here is a small self-contained example that uses closures and serialization

162+-- dictionaries. It makes use of the Control.Distributed.Process.SimpleLocalnet

163+-- Cloud Haskell backend.

164+--

165+-- > {-# LANGUAGE TemplateHaskell #-}

166+-- > import System.Environment (getArgs)

167+-- > import Control.Distributed.Process

168+-- > import Control.Distributed.Process.Closure

169+-- > import Control.Distributed.Process.Backend.SimpleLocalnet

170+-- > import Control.Distributed.Process.Node (initRemoteTable)

171+-- >

172+-- > isPrime :: Integer -> Process Bool

173+-- > isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]

174+-- > where

175+-- > sieve :: [Integer] -> [Integer]

176+-- > sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]

177+-- >

178+-- > remotable ['isPrime]

179+-- >

180+-- > master :: [NodeId] -> Process ()

181+-- > master [] = liftIO $ putStrLn "no slaves"

182+-- > master (slave:_) = do

183+-- > isPrime79 <- call $(functionTDict 'isPrime) slave ($(mkClosure 'isPrime) (79 :: Integer))

184+-- > liftIO $ print isPrime79

185+-- >

186+-- > main :: IO ()

187+-- > main = do

188+-- > args <- getArgs

189+-- > case args of

190+-- > ["master", host, port] -> do

191+-- > backend <- initializeBackend host port rtable

192+-- > startMaster backend master

193+-- > ["slave", host, port] -> do

194+-- > backend <- initializeBackend host port rtable

195+-- > startSlave backend

196+-- > where

197+-- > rtable :: RemoteTable

198+-- > rtable = __remoteTable initRemoteTable

199+--

136200-- [Notes]

137201--

138202-- (*) If 'T1' is not serializable you will get a type error in the generated

@@ -147,11 +211,13 @@

147211-- instance into scope, but unless proper 'static' support is added to

148212-- ghc we need both the type class argument and the explicit dictionary.

149213module Control.Distributed.Process.Closure

150− ( -- * User-defined closures

214+ ( -- * Creating static values

151215 remotable

152216 , mkStatic

217+ -- * Template-Haskell support for creating closures

153218 , mkClosure

154219 , functionSDict

220+ , functionTDict

155221 -- * Primitive operations on static values

156222 , staticApply

157223 , staticDuplicate

@@ -207,6 +273,7 @@ import Control.Distributed.Process.Internal.Closure.TH

207273 ( remotable

208274 , mkStatic

209275 , functionSDict

276+ , functionTDict

210277 )

211278import Control.Distributed.Process.Internal.Closure.Static

212279 ( -- Static functionals

distributed-process/src/Control/Distributed/Process/Internal/Closure/CP.hs+3−3

@@ -31,6 +31,7 @@ import Data.ByteString.Lazy (ByteString)

3131import Data.Typeable (Typeable, typeOf, typeRepTyCon, TyCon)

3232import Control.Applicative ((<$>))

3333import Control.Monad ((>=>))

34+import Control.Monad.Reader (ask)

3435import Control.Distributed.Process.Serializable (Serializable)

3536import Control.Distributed.Process.Internal.Types

3637 ( Closure(Closure)

@@ -43,7 +44,7 @@ import Control.Distributed.Process.Internal.Types

4344 , typeOfStaticLabel

4445 , ProcessId

4546 , LocalNode(remoteTable)

46− , procMsg

47+ , LocalProcess(processNode)

4748 , SendPort

4849 , ReceivePort

4950 )

@@ -55,7 +56,6 @@ import Control.Distributed.Process.Internal.Primitives

5556 , newChan

5657 )

5758import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)

58−import Control.Distributed.Process.Internal.MessageT (getLocalNode)

5959import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)

6060import Control.Distributed.Process.Internal.Closure.Static

6161 ( staticCompose

@@ -118,7 +118,7 @@ sdictComp = SerializableDict

118118-- | Resolve a closure

119119unClosure :: Static a -> ByteString -> Process Dynamic

120120unClosure (Static label) env = do

121− rtable <- remoteTable <$> procMsg getLocalNode

121+ rtable <- remoteTable . processNode <$> ask

122122 case resolveClosure rtable label env of

123123 Nothing -> fail "Derived.unClosure: resolveClosure failed"

124124 Just dyn -> return dyn

distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hs+71−50

@@ -7,6 +7,7 @@ module Control.Distributed.Process.Internal.Closure.TH

77 remotable

88 , mkStatic

99 , functionSDict

10+ , functionTDict

1011 ) where

1112

1213import Prelude hiding (lookup)

@@ -47,6 +48,7 @@ import Control.Distributed.Process.Internal.Types

4748 , StaticLabel(StaticLabel)

4849 , remoteTableLabel

4950 , SerializableDict(SerializableDict)

51+ , Process

5052 )

5153import Control.Distributed.Process.Internal.Dynamic

5254 ( Dynamic(..)

@@ -74,18 +76,19 @@ remotable ns = do

7476mkStatic :: Name -> Q Exp

7577mkStatic = varE . staticName

7678

77−-- | Serialization dictionary for a function argument (see module header)

79+-- | If @f : T1 -> T2@ is a monomorphic function

80+-- then @$(functionSDict 'f) :: Static (SerializableDict T1)@.

81+--

82+-- Be sure to pass 'f' to 'remotable'.

7883functionSDict :: Name -> Q Exp

7984functionSDict = varE . sdictName

8085

81−{-

82−-- | Create a closure

83−--

84−-- See module documentation header for "Control.Distributed.Process.Closure"

85−-- for a detailed explanation and examples.

86−mkClosure :: Name -> Q Exp

87−mkClosure = varE . closureName

88−-}

86+-- | If @f : T1 -> Process T2@ is a monomorphic function

87+-- then @$(functionTDict 'f) :: Static (SerializableDict T2)@.

88+--

89+-- Be sure to pass 'f' to 'remotable'.

90+functionTDict :: Name -> Q Exp

91+functionTDict = varE . tdictName

8992

9093--------------------------------------------------------------------------------

9194-- Internal (Template Haskell) --

@@ -100,44 +103,59 @@ createMetaData is =

100103

101104generateDefs :: Name -> Q ([Dec], [Q Exp])

102105generateDefs n = do

103− mType <- getType n

104− case mType of

105− Just (origName, typ) -> do

106− let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)

107− _ -> ([], typ)

108− (static, register) <- do

109− static <- generateStatic origName typVars typ'

110− let dyn = case typVars of

111− [] -> [| toDyn $(varE origName) |]

112− _ -> [| Dynamic (error "Polymorphic value")

113− (unsafeCoerce# $(varE origName))

114− |]

115− return ( static

116− , [ [| registerStatic $(stringE (show origName)) $dyn |] ]

106+ proc <- [t| Process |]

107+ mType <- getType n

108+ case mType of

109+ Just (origName, typ) -> do

110+ let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)

111+ _ -> ([], typ)

112+

113+ -- The main "static" entry

114+ (static, register) <- makeStatic origName typVars typ'

115+

116+ -- If n :: T1 -> T2, static serializable dictionary for T1

117+ -- TODO: we should check if arg is an instance of Serializable, but we cannot

118+ -- http://hackage.haskell.org/trac/ghc/ticket/7066

119+ (sdict, registerSDict) <- case (typVars, typ') of

120+ ([], ArrowT `AppT` arg `AppT` _res) ->

121+ makeDict (sdictName origName) arg

122+ _ ->

123+ return ([], [])

124+

125+ -- If n :: T1 -> Process T2, static serializable dictionary for T2

126+ -- TODO: check if T2 is serializable (same as above)

127+ (tdict, registerTDict) <- case (typVars, typ') of

128+ ([], ArrowT `AppT` _arg `AppT` (proc' `AppT` res)) | proc' == proc ->

129+ makeDict (tdictName origName) res

130+ _ ->

131+ return ([], [])

132+

133+ return ( concat [static, sdict, tdict]

134+ , concat [register, registerSDict, registerTDict]

117135 )

136+ _ ->

137+ fail $ "remotable: " ++ show n ++ " not found"

138+ where

139+ makeStatic :: Name -> [TyVarBndr] -> Type -> Q ([Dec], [Q Exp])

140+ makeStatic origName typVars typ = do

141+ static <- generateStatic origName typVars typ

142+ let dyn = case typVars of

143+ [] -> [| toDyn $(varE origName) |]

144+ _ -> [| Dynamic (error "Polymorphic value")

145+ (unsafeCoerce# $(varE origName))

146+ |]

147+ return ( static

148+ , [ [| registerStatic $(stringE (show origName)) $dyn |] ]

149+ )

118150

119− (sdict, registerSDict) <- case (typVars, typ') of

120− ([], ArrowT `AppT` arg `AppT` _res) -> do

121− -- TODO: we should check if arg is an instance of Serializable, but we cannot

122− -- http://hackage.haskell.org/trac/ghc/ticket/7066

123− sdict <- generateSDict origName arg

124− let dyn' = [| toDyn (SerializableDict

125− :: SerializableDict $(return arg)

126− )

127− |]

128− return ( sdict

129− , [ [| registerStatic $(stringE (show (sdictName origName)))

130− $dyn'

131− |] ]

132− )

133− _ ->

134− return ([], [])

135−

136− return ( static ++ sdict

137− , register ++ registerSDict

151+ makeDict :: Name -> Type -> Q ([Dec], [Q Exp])

152+ makeDict dictName typ = do

153+ sdict <- generateDict dictName typ

154+ let dyn = [| toDyn (SerializableDict :: SerializableDict $(return typ)) |]

155+ return ( sdict

156+ , [ [| registerStatic $(stringE (show dictName)) $dyn |] ]

138157 )

139− _ ->

140− fail $ "remotable: " ++ show n ++ " not found"

158+

141159

142160registerStatic :: String -> Dynamic -> RemoteTable -> RemoteTable

143161registerStatic label dyn = remoteTableLabel label ^= Just dyn

@@ -163,14 +181,14 @@ generateStatic n xs typ = do

163181 typeable (PlainTV v) = ClassP (mkName "Typeable") [VarT v]

164182 typeable (KindedTV v _) = ClassP (mkName "Typeable") [VarT v]

165183

166−-- | Generate a function serialization dictionary

167−generateSDict :: Name -> Type -> Q [Dec]

168−generateSDict n typ = do

184+-- | Generate a serialization dictionary with name 'n' for type 'typ'

185+generateDict :: Name -> Type -> Q [Dec]

186+generateDict n typ = do

169187 sequence

170− [ sigD (sdictName n) $ [t| Static (SerializableDict $(return typ)) |]

171− , sfnD (sdictName n)

188+ [ sigD n $ [t| Static (SerializableDict $(return typ)) |]

189+ , sfnD n

172190 [| Static $ StaticLabel

173− $(stringE (show (sdictName n)))

191+ $(stringE (show n))

174192 (typeOf (undefined :: SerializableDict $(return typ)))

175193 |]

176194 ]

@@ -181,6 +199,9 @@ staticName n = mkName $ nameBase n ++ "__static"

181199sdictName :: Name -> Name

182200sdictName n = mkName $ nameBase n ++ "__sdict"

183201

202+tdictName :: Name -> Name

203+tdictName n = mkName $ nameBase n ++ "__tdict"

204+

184205--------------------------------------------------------------------------------

185206-- Generic Template Haskell auxiliary functions --

186207--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Node.hs+84−9

@@ -1,16 +1,91 @@

11module Control.Distributed.Process.Internal.Node

2− ( runLocalProcess

2+ ( -- * Message sending

3+ sendPayload

4+ , sendBinary

5+ , sendMessage

36 ) where

5−import Control.Monad.Reader (runReaderT)

8+import Data.Accessor ((^.), (^=))

9+import Data.Binary (Binary, encode)

10+import qualified Data.ByteString.Lazy as BSL (toChunks)

11+import qualified Data.ByteString as BSS (ByteString)

12+import Control.Concurrent.MVar (withMVar, modifyMVar_)

13+import Control.Concurrent.Chan (writeChan)

14+import Control.Monad (unless)

15+import qualified Network.Transport as NT

16+ ( Connection

17+ , send

18+ , defaultConnectHints

19+ , connect

20+ , Reliability(ReliableOrdered)

21+ )

622import Control.Distributed.Process.Internal.Types

7− ( LocalNode

8− , LocalProcess

9− , Process(unProcess)

23+ ( LocalNode(localState, localEndPoint, localCtrlChan)

24+ , Identifier

25+ , localConnectionBetween

26+ , nodeAddress

27+ , nodeOf

28+ , messageToPayload

29+ , createMessage

30+ , NCMsg(..)

31+ , ProcessSignal(Died)

32+ , DiedReason(DiedDisconnect)

1033 )

11−import Control.Distributed.Process.Internal.MessageT (runMessageT)

34+import Control.Distributed.Process.Serializable (Serializable)

35+

36+--------------------------------------------------------------------------------

37+-- Message sending --

38+--------------------------------------------------------------------------------

39+

40+sendPayload :: LocalNode -> Identifier -> Identifier -> [BSS.ByteString] -> IO ()

41+sendPayload node from to payload = do

42+ mConn <- connBetween node from to

43+ didSend <- case mConn of

44+ Just conn -> do

45+ didSend <- NT.send conn payload

46+ case didSend of

47+ Left _ -> return False

48+ Right _ -> return True

49+ Nothing -> return False

50+ unless didSend $ do

51+ writeChan (localCtrlChan node) $ NCMsg

52+ { ctrlMsgSender = to

53+ , ctrlMsgSignal = Died to DiedDisconnect

54+ }

55+

56+sendBinary :: Binary a => LocalNode -> Identifier -> Identifier -> a -> IO ()

57+sendBinary node from to = sendPayload node from to . BSL.toChunks . encode

58+

59+sendMessage :: Serializable a => LocalNode -> Identifier -> Identifier -> a -> IO ()

60+sendMessage node from to = sendPayload node from to . messageToPayload . createMessage

1261

13−-- | Deconstructor for 'Process' (not exported to the public API)

14−runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a

15−runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)

62+setupConnBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)

63+setupConnBetween node from to = do

64+ mConn <- NT.connect endPoint

65+ (nodeAddress . nodeOf $ to)

66+ NT.ReliableOrdered

67+ NT.defaultConnectHints

68+ case mConn of

69+ Right conn -> do

70+ didSend <- NT.send conn (BSL.toChunks . encode $ to)

71+ case didSend of

72+ Left _ ->

73+ return Nothing

74+ Right () -> do

75+ modifyMVar_ nodeState $ return .

76+ (localConnectionBetween from to ^= Just conn)

77+ return $ Just conn

78+ Left _ ->

79+ return Nothing

80+ where

81+ endPoint = localEndPoint node

82+ nodeState = localState node

1683

84+connBetween :: LocalNode -> Identifier -> Identifier -> IO (Maybe NT.Connection)

85+connBetween node from to = do

86+ mConn <- withMVar nodeState $ return . (^. localConnectionBetween from to)

87+ case mConn of

88+ Just conn -> return $ Just conn

89+ Nothing -> setupConnBetween node from to

90+ where

91+ nodeState = localState node

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+25−21

@@ -68,7 +68,7 @@ import Control.Monad.Reader (ask)

6868import Control.Monad.IO.Class (MonadIO, liftIO)

6969import Control.Applicative ((<$>))

7070import Control.Exception (Exception, throw)

71−import qualified Control.Exception as Exception (catch)

71+import qualified Control.Exception as Ex (catch)

7272import Control.Concurrent.MVar (modifyMVar)

7373import Control.Concurrent.Chan (writeChan)

7474import Control.Concurrent.STM

@@ -106,7 +106,6 @@ import Control.Distributed.Process.Internal.Types

106106 , TypedChannel(..)

107107 , SendPortId(..)

108108 , Identifier(..)

109− , procMsg

110109 , DidUnmonitor(..)

111110 , DidUnlinkProcess(..)

112111 , DidUnlinkNode(..)

@@ -114,13 +113,9 @@ import Control.Distributed.Process.Internal.Types

114113 , WhereIsReply(..)

115114 , createMessage

116115 , Static(..)

116+ , runLocalProcess

117117 )

118−import Control.Distributed.Process.Internal.MessageT

119− ( sendMessage

120− , sendBinary

121− , getLocalNode

122− )

123−import Control.Distributed.Process.Internal.Node (runLocalProcess)

118+import Control.Distributed.Process.Internal.Node (sendMessage, sendBinary)

124119import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)

125120import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)

126121

@@ -132,7 +127,12 @@ import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)

132127send :: Serializable a => ProcessId -> a -> Process ()

133128-- This requires a lookup on every send. If we want to avoid that we need to

134129-- modify serializable to allow for stateful (IO) deserialization

135−send them msg = procMsg $ sendMessage (ProcessIdentifier them) msg

130+send them msg = do

131+ proc <- ask

132+ liftIO $ sendMessage (processNode proc)

133+ (ProcessIdentifier (processId proc))

134+ (ProcessIdentifier them)

135+ msg

136136

137137-- | Wait for a message of a specific type

138138expect :: forall a. Serializable a => Process a

@@ -163,7 +163,12 @@ newChan = do

163163

164164-- | Send a message on a typed channel

165165sendChan :: Serializable a => SendPort a -> a -> Process ()

166−sendChan (SendPort cid) msg = procMsg $ sendBinary (SendPortIdentifier cid) msg

166+sendChan (SendPort cid) msg = do

167+ proc <- ask

168+ liftIO $ sendBinary (processNode proc)

169+ (ProcessIdentifier (processId proc))

170+ (SendPortIdentifier cid)

171+ msg

167172

168173-- | Wait for a message on a typed channel

169174receiveChan :: Serializable a => ReceivePort a -> Process a

@@ -261,7 +266,7 @@ getSelfPid = processId <$> ask

261266

262267-- | Get the node ID of our local node

263268getSelfNode :: Process NodeId

264−getSelfNode = localNodeId <$> procMsg getLocalNode

269+getSelfNode = localNodeId . processNode <$> ask

265270

266271--------------------------------------------------------------------------------

267272-- Monitoring and linking --

@@ -318,11 +323,8 @@ unmonitor ref = do

318323-- | Catch exceptions within a process

319324catch :: Exception e => Process a -> (e -> Process a) -> Process a

320325catch p h = do

321− node <- procMsg getLocalNode

322326 lproc <- ask

323− let run :: Process a -> IO a

324− run proc = runLocalProcess node proc lproc

325− liftIO $ Exception.catch (run p) (run . h)

327+ liftIO $ Ex.catch (runLocalProcess lproc p) (runLocalProcess lproc . h)

326328

327329-- | Like 'expect' but with a timeout

328330expectTimeout :: forall a. Serializable a => Int -> Process (Maybe a)

@@ -460,7 +462,7 @@ nsendRemote nid label msg =

460462-- | Deserialize a closure

461463unClosure :: forall a. Typeable a => Closure a -> Process a

462464unClosure (Closure (Static label) env) = do

463− rtable <- remoteTable <$> procMsg getLocalNode

465+ rtable <- remoteTable . processNode <$> ask

464466 case resolveClosure rtable label env of

465467 Nothing -> throw . userError $ "Unregistered closure " ++ show label

466468 Just dyn -> return $ fromDyn dyn (throw (typeError dyn))

@@ -507,14 +509,16 @@ sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node

507509 -> ProcessSignal -- ^ Message to send

508510 -> Process ()

509511sendCtrlMsg mNid signal = do

510− us <- getSelfPid

511− let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us

512+ proc <- ask

513+ let msg = NCMsg { ctrlMsgSender = ProcessIdentifier (processId proc)

512514 , ctrlMsgSignal = signal

513515 }

514516 case mNid of

515517 Nothing -> do

516− ctrlChan <- localCtrlChan <$> procMsg getLocalNode

518+ ctrlChan <- localCtrlChan . processNode <$> ask

517519 liftIO $ writeChan ctrlChan msg

518520 Just nid ->

519− procMsg $ sendBinary (NodeIdentifier nid) msg

520−

521+ liftIO $ sendBinary (processNode proc)

522+ (ProcessIdentifier (processId proc))

523+ (NodeIdentifier nid)

524+ msg

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+21−30

@@ -17,7 +17,7 @@ module Control.Distributed.Process.Internal.Types

1717 , LocalProcess(..)

1818 , LocalProcessState(..)

1919 , Process(..)

20− , procMsg

20+ , runLocalProcess

2121 -- * Typed channels

2222 , LocalSendPortId

2323 , SendPortId(..)

@@ -58,14 +58,13 @@ module Control.Distributed.Process.Internal.Types

5858 -- * Node controller internal data types

5959 , NCMsg(..)

6060 , ProcessSignal(..)

61− -- * MessageT monad

62− , MessageT(..)

63− , MessageState(..)

6461 -- * Accessors

6562 , localProcesses

6663 , localPidCounter

6764 , localPidUnique

65+ , localConnections

6866 , localProcessWithId

67+ , localConnectionBetween

6968 , monitorCounter

7069 , spawnCounter

7170 , channelCounter

@@ -97,10 +96,8 @@ import Control.Concurrent.Chan (Chan)

9796import Control.Concurrent.STM (TChan, TVar)

9897import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)

9998import Control.Applicative (Applicative, (<$>), (<*>))

100−import Control.Monad.Reader (MonadReader(..), ReaderT)

99+import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

101100import Control.Monad.IO.Class (MonadIO)

102−import Control.Monad.State (MonadState, StateT)

103−import qualified Control.Monad.Trans.Class as Trans (lift)

104101import Control.Distributed.Process.Serializable

105102 ( Fingerprint

106103 , Serializable

@@ -131,10 +128,7 @@ data LocalProcessId = LocalProcessId

131128 { lpidUnique :: Int32

132129 , lpidCounter :: Int32

133130 }

134− deriving (Eq, Ord, Typeable)

135−

136−instance Show LocalProcessId where

137− show = show . lpidCounter

131+ deriving (Eq, Ord, Typeable, Show)

138132

139133-- | Process identifier

140134data ProcessId = ProcessId

@@ -187,9 +181,10 @@ data LocalNode = LocalNode

187181

188182-- | Local node state

189183data LocalNodeState = LocalNodeState

190− { _localProcesses :: Map LocalProcessId LocalProcess

191− , _localPidCounter :: Int32

192− , _localPidUnique :: Int32

184+ { _localProcesses :: Map LocalProcessId LocalProcess

185+ , _localPidCounter :: Int32

186+ , _localPidUnique :: Int32

187+ , _localConnections :: Map (Identifier, Identifier) NT.Connection

193188 }

194189

195190-- | Processes running on our local node

@@ -198,8 +193,13 @@ data LocalProcess = LocalProcess

198193 , processId :: ProcessId

199194 , processState :: MVar LocalProcessState

200195 , processThread :: ThreadId

196+ , processNode :: LocalNode

201197 }

202198

199+-- | Deconstructor for 'Process' (not exported to the public API)

200+runLocalProcess :: LocalProcess -> Process a -> IO a

201+runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

202+

203203-- | Local process state

204204data LocalProcessState = LocalProcessState

205205 { _monitorCounter :: Int32

@@ -210,13 +210,10 @@ data LocalProcessState = LocalProcessState

210210

211211-- | The Cloud Haskell 'Process' type

212212newtype Process a = Process {

213− unProcess :: ReaderT LocalProcess (MessageT IO) a

213+ unProcess :: ReaderT LocalProcess IO a

214214 }

215215 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)

216216

217−procMsg :: MessageT IO a -> Process a

218−procMsg = Process . Trans.lift

219−

220217--------------------------------------------------------------------------------

221218-- Typed channels --

222219--------------------------------------------------------------------------------

@@ -574,9 +571,15 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =

574571localPidUnique :: Accessor LocalNodeState Int32

575572localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

576573

574+localConnections :: Accessor LocalNodeState (Map (Identifier, Identifier) NT.Connection)

575+localConnections = accessor _localConnections (\conns st -> st { _localConnections = conns })

576+

577577localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

578578localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

579579

580+localConnectionBetween :: Identifier -> Identifier -> Accessor LocalNodeState (Maybe NT.Connection)

581+localConnectionBetween from to = localConnections >>> DAC.mapMaybe (from, to)

582+

580583monitorCounter :: Accessor LocalProcessState Int32

581584monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt })

582585

@@ -597,15 +600,3 @@ remoteTableLabels = accessor _remoteTableLabels (\ls tbl -> tbl { _remoteTableLa

597600

598601remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic)

599602remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label

600−

601−--------------------------------------------------------------------------------

602−-- MessageT monad --

603−--------------------------------------------------------------------------------

604−

605−newtype MessageT m a = MessageT { unMessageT :: StateT MessageState m a }

606− deriving (Functor, Monad, MonadIO, MonadState MessageState, Applicative)

607−

608−data MessageState = MessageState {

609− messageLocalNode :: LocalNode

610− , _messageConnections :: Map Identifier NT.Connection

611− }

distributed-process/src/Control/Distributed/Process/Node.hs+64−47

@@ -38,7 +38,7 @@ import Control.Applicative ((<$>))

3838import Control.Monad (void, when, forever)

3939import Control.Monad.IO.Class (MonadIO, liftIO)

4040import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)

41−import qualified Control.Monad.Trans.Class as Trans (lift)

41+import Control.Monad.Reader (MonadReader, ReaderT, runReaderT, ask)

4242import Control.Exception (throwIO, SomeException, Exception, throwTo)

4343import qualified Control.Exception as Exception (catch)

4444import Control.Concurrent (forkIO)

@@ -101,7 +101,6 @@ import Control.Distributed.Process.Internal.Types

101101 , Closure(..)

102102 , Static(..)

103103 , Message

104− , MessageT

105104 , TypedChannel(..)

106105 , Identifier(..)

107106 , nodeOf

@@ -110,20 +109,18 @@ import Control.Distributed.Process.Internal.Types

110109 , WhereIsReply(..)

111110 , messageToPayload

112111 , RemoteTable(..)

112+ , payloadToMessage

113+ , createMessage

114+ , runLocalProcess

113115 )

114116import Control.Distributed.Process.Serializable (Serializable)

115−import Control.Distributed.Process.Internal.MessageT

117+import Control.Distributed.Process.Internal.Dynamic (fromDynamic)

118+import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)

119+import Control.Distributed.Process.Internal.Node

116120 ( sendBinary

117121 , sendMessage

118122 , sendPayload

119− , getLocalNode

120− , runMessageT

121− , payloadToMessage

122− , createMessage

123123 )

124−import Control.Distributed.Process.Internal.Dynamic (fromDynamic)

125−import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)

126−import Control.Distributed.Process.Internal.Node (runLocalProcess)

127124import Control.Distributed.Process.Internal.Primitives (expect, register)

128125import qualified Control.Distributed.Process.Internal.Closure.Static as Static (__remoteTable)

129126import qualified Control.Distributed.Process.Internal.Closure.CP as CP (__remoteTable)

@@ -156,9 +153,10 @@ createBareLocalNode :: NT.EndPoint -> RemoteTable -> IO LocalNode

156153createBareLocalNode endPoint rtable = do

157154 unq <- randomIO

158155 state <- newMVar LocalNodeState

159− { _localProcesses = Map.empty

160− , _localPidCounter = 0

161− , _localPidUnique = unq

156+ { _localProcesses = Map.empty

157+ , _localPidCounter = 0

158+ , _localPidUnique = unq

159+ , _localConnections = Map.empty

162160 }

163161 ctrlChan <- newChan

164162 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint

@@ -214,10 +212,11 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do

214212 , processId = pid

215213 , processState = pst

216214 , processThread = tid

215+ , processNode = node

217216 }

218217 tid' <- forkIO $ do

219218 reason <- Exception.catch

220− (runLocalProcess node proc lproc >> return DiedNormal)

219+ (runLocalProcess lproc proc >> return DiedNormal)

221220 (return . DiedException . (show :: SomeException -> String))

222221 -- [Unified: Table 4, rules termination and exiting]

223222 modifyMVar_ (localState node) $

@@ -287,7 +286,7 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

287286 (Map.insert cid proc procs)

288287 chans

289288 ctrls

290− Nothing ->

289+ Nothing ->

291290 -- Request for an unknown process.

292291 --

293292 -- TODO: We should treat this as a fatal error on the

@@ -367,8 +366,8 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty

367366--------------------------------------------------------------------------------

368367

369368runNodeController :: LocalNode -> IO ()

370−runNodeController node =

371− runMessageT node (evalStateT (unNC nodeController) initNCState)

369+runNodeController =

370+ runReaderT (evalStateT (unNC nodeController) initNCState)

372371

373372--------------------------------------------------------------------------------

374373-- Internal data types --

@@ -383,11 +382,8 @@ data NCState = NCState

383382 , _registry :: Map String ProcessId

384383 }

385384

386−newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }

387− deriving (Functor, Monad, MonadIO, MonadState NCState)

388−

389−ncMsg :: MessageT IO a -> NC a

390−ncMsg = NC . Trans.lift

385+newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

386+ deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)

391387

392388initNCState :: NCState

393389initNCState = NCState { _links = Map.empty

@@ -402,14 +398,17 @@ initNCState = NCState { _links = Map.empty

402398-- [Unified: Table 7]

403399nodeController :: NC ()

404400nodeController = do

405− node <- ncMsg getLocalNode

401+ node <- ask

406402 forever $ do

407403 msg <- liftIO $ readChan (localCtrlChan node)

408404

409405 -- [Unified: Table 7, rule nc_forward]

410406 case destNid (ctrlMsgSignal msg) of

411407 Just nid' | nid' /= localNodeId node ->

412− ncMsg $ sendBinary (NodeIdentifier nid') msg

408+ liftIO $ sendBinary node

409+ (ctrlMsgSender msg)

410+ (NodeIdentifier nid')

411+ msg

413412 _ ->

414413 return ()

415414

@@ -441,7 +440,7 @@ ncEffectMonitor :: ProcessId -- ^ Who's watching?

441440 -> Maybe MonitorRef -- ^ 'Nothing' to link

442441 -> NC ()

443442ncEffectMonitor from them mRef = do

444− node <- ncMsg getLocalNode

443+ node <- ask

445444 shouldLink <-

446445 if not (isLocal node them)

447446 then return True

@@ -454,15 +453,21 @@ ncEffectMonitor from them mRef = do

454453 (False, True) -> -- [Unified: second rule]

455454 notifyDied from them DiedUnknownId mRef

456455 (False, False) -> -- [Unified: third rule]

457− ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg

458− { ctrlMsgSender = NodeIdentifier (localNodeId node)

459− , ctrlMsgSignal = Died them DiedUnknownId

460− }

456+ -- TODO: this is the right sender according to the Unified semantics,

457+ -- but perhaps having 'them' as the sender would make more sense

458+ -- (see also: notifyDied)

459+ liftIO $ sendBinary node

460+ (NodeIdentifier $ localNodeId node)

461+ (NodeIdentifier $ processNodeId from) $

462+ NCMsg

463+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

464+ , ctrlMsgSignal = Died them DiedUnknownId

465+ }

461466

462467-- [Unified: Table 11]

463468ncEffectUnlink :: ProcessId -> Identifier -> NC ()

464469ncEffectUnlink from them = do

465− node <- ncMsg getLocalNode

470+ node <- ask

466471 when (isLocal node (ProcessIdentifier from)) $

467472 case them of

468473 ProcessIdentifier pid ->

@@ -476,7 +481,7 @@ ncEffectUnlink from them = do

476481-- [Unified: Table 11]

477482ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()

478483ncEffectUnmonitor from ref = do

479− node <- ncMsg getLocalNode

484+ node <- ask

480485 when (isLocal node (ProcessIdentifier from)) $

481486 postAsMessage from $ DidUnmonitor ref

482487 modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)

@@ -484,7 +489,7 @@ ncEffectUnmonitor from ref = do

484489-- [Unified: Table 12]

485490ncEffectDied :: Identifier -> DiedReason -> NC ()

486491ncEffectDied ident reason = do

487− node <- ncMsg getLocalNode

492+ node <- ask

488493 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))

489494 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))

490495

@@ -509,9 +514,12 @@ ncEffectSpawn pid cProc ref = do

509514 -- If the closure does not exist, we spawn a process that throws an exception

510515 -- This allows the remote node to find out what's happening

511516 let proc = fromMaybe (fail $ "Error: unknown closure " ++ show cProc) mProc

512− node <- ncMsg getLocalNode

517+ node <- ask

513518 pid' <- liftIO $ forkProcess node proc

514− ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

519+ liftIO $ sendMessage node

520+ (NodeIdentifier (localNodeId node))

521+ (ProcessIdentifier pid)

522+ (DidSpawn ref pid')

515523

516524-- Unified semantics does not explicitly describe how to implement 'register',

517525-- but mentions it's "very similar to nsend" (Table 14)

@@ -525,18 +533,24 @@ ncEffectRegister label mPid =

525533-- Unified semantics does not explicitly describe 'whereis'

526534ncEffectWhereIs :: ProcessId -> String -> NC ()

527535ncEffectWhereIs from label = do

536+ node <- ask

528537 mPid <- gets (^. registryFor label)

529− ncMsg $ sendMessage (ProcessIdentifier from) (WhereIsReply label mPid)

538+ liftIO $ sendMessage node

539+ (NodeIdentifier (localNodeId node))

540+ (ProcessIdentifier from)

541+ (WhereIsReply label mPid)

530542

531543-- [Unified: Table 14]

532544ncEffectNamedSend :: Identifier -> String -> Message -> NC ()

533−ncEffectNamedSend _from label msg = do

545+ncEffectNamedSend from label msg = do

534546 mPid <- gets (^. registryFor label)

535547 -- If mPid is Nothing, we just ignore the named send (as per Table 14)

536− -- TODO: messages don't carry a "from", but if they do, we should set it

537− -- to be the 'from' of the original named send, not this node controller

548+ node <- ask

538549 forM_ mPid $ \pid ->

539− ncMsg $ sendPayload (ProcessIdentifier pid) (messageToPayload msg)

550+ liftIO $ sendPayload node

551+ from

552+ (ProcessIdentifier pid)

553+ (messageToPayload msg)

540554

541555--------------------------------------------------------------------------------

542556-- Auxiliary --

@@ -548,7 +562,7 @@ notifyDied :: ProcessId -- ^ Who to notify?

548562 -> Maybe MonitorRef -- ^ 'Nothing' for linking

549563 -> NC ()

550564notifyDied dest src reason mRef = do

551− node <- ncMsg getLocalNode

565+ node <- ask

552566 case (isLocal node (ProcessIdentifier dest), mRef, src) of

553567 (True, Just ref, ProcessIdentifier pid) ->

554568 postAsMessage dest $ ProcessMonitorNotification ref pid reason

@@ -565,10 +579,13 @@ notifyDied dest src reason mRef = do

565579 (False, _, _) ->

566580 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

567581 -- (see [Unified: Table 10]

568− ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg

569− { ctrlMsgSender = NodeIdentifier (localNodeId node)

570− , ctrlMsgSignal = Died src reason

571− }

582+ liftIO $ sendBinary node

583+ (NodeIdentifier $ localNodeId node)

584+ (NodeIdentifier $ processNodeId dest) $

585+ NCMsg

586+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

587+ , ctrlMsgSignal = Died src reason

588+ }

572589

573590-- | [Unified: Table 8]

574591destNid :: ProcessSignal -> Maybe NodeId

@@ -592,13 +609,13 @@ isLocal nid ident = nodeOf ident == localNodeId nid

592609-- | Lookup a local closure

593610unClosure :: Typeable a => Closure a -> NC (Maybe a)

594611unClosure (Closure (Static label) env) = do

595− rtable <- remoteTable <$> ncMsg getLocalNode

612+ rtable <- remoteTable <$> ask

596613 return (resolveClosure rtable label env >>= fromDynamic)

597614

598615-- | Check if an identifier refers to a valid local object

599616isValidLocalIdentifier :: Identifier -> NC Bool

600617isValidLocalIdentifier ident = do

601− node <- ncMsg getLocalNode

618+ node <- ask

602619 liftIO . withMVar (localState node) $ \nSt ->

603620 case ident of

604621 NodeIdentifier nid ->

@@ -631,7 +648,7 @@ throwException pid e = withLocalProc pid $ \p ->

631648

632649withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()

633650withLocalProc pid p = do

634− node <- ncMsg getLocalNode

651+ node <- ask

635652 liftIO $ do

636653 -- By [Unified: table 6, rule missing_process] messages to dead processes

637654 -- can silently be dropped

distributed-process/tests/TestClosure.hs+15

@@ -36,6 +36,12 @@ sendPid toPid = do

3636wait :: Int -> Process ()

3737wait = liftIO . threadDelay

3838

39+isPrime :: Integer -> Process Bool

40+isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]

41+ where

42+ sieve :: [Integer] -> [Integer]

43+ sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]

44+

3945-- | First argument indicates empty closure environment

4046typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()

4147typedPingServer () rport = forever $ do

@@ -49,6 +55,7 @@ remotable [ 'factorial

4955 , 'sdictInt

5056 , 'wait

5157 , 'typedPingServer

58+ , 'isPrime

5259 ]

5360

5461factorialClosure :: Int -> Closure (Process Int)

@@ -296,6 +303,13 @@ testSpawnChannel transport rtable = do

296303 sendChan pingServer sendReply

297304 receiveChan receiveReply

298305

306+testTDict :: Transport -> RemoteTable -> IO ()

307+testTDict transport rtable = do

308+ [node1, node2] <- replicateM 2 $ newLocalNode transport rtable

309+ runProcess node1 $ do

310+ True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))

311+ return ()

312+

299313main :: IO ()

300314main = do

301315 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

@@ -314,4 +328,5 @@ main = do

314328 , ("SpawnInvalid", testSpawnInvalid transport rtable)

315329 , ("ClosureExpect", testClosureExpect transport rtable)

316330 , ("SpawnChannel", testSpawnChannel transport rtable)

331+ , ("TDict", testTDict transport rtable)

317332 ]