de4b010 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Improve documentation of C.D.P.Closure

de4b010Edsko de Vries14 years ago

distributed-process/distributed-process.cabal−1

compare with master

unifiedsplit

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

52 Control.Distributed.Process.Internal.CQueue,

53 Control.Distributed.Process.Internal.Dynamic,

54 Control.Distributed.Process.Internal.TypeRep,

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

56 Control.Distributed.Process.Internal.Types,

55 Control.Distributed.Process.Internal.Types,

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

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

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

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

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

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

89 , spawnMonitor

90 , spawnChannel

91 , DidSpawn(..)

92− -- * Local versions of spawn

92+ -- * Local versions of 'spawn'

93 , spawnLocal

94 , spawnChannelLocal

95 ) where

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

101

102import Data.Typeable (Typeable)

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

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

105+import Control.Monad.Reader (ask)

104import Control.Distributed.Process.Internal.Types

106import Control.Distributed.Process.Internal.Types

105 ( RemoteTable

107 ( RemoteTable

106 , NodeId(..)

108 , NodeId(..)

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

124 , SerializableDict(..)

126 , SerializableDict(..)

125 , SendPortId(..)

127 , SendPortId(..)

126 , WhereIsReply(..)

128 , WhereIsReply(..)

127− , procMsg

129+ , LocalProcess(processNode)

128 )

130 )

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

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

130 ( cpSeq

132 ( cpSeq

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

194 , spawnAsync

196 , spawnAsync

195 )

197 )

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

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

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

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

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

199

200

200-- INTERNAL NOTES

201-- INTERNAL NOTES

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

369-- | Spawn a process on the local node

370-- | Spawn a process on the local node

370spawnLocal :: Process () -> Process ProcessId

371spawnLocal :: Process () -> Process ProcessId

371spawnLocal proc = do

372spawnLocal proc = do

372− node <- procMsg $ getLocalNode

373+ node <- processNode <$> ask

373 liftIO $ forkProcess node proc

374 liftIO $ forkProcess node proc

374

375

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

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

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

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

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

379 -> Process (SendPort a)

380 -> Process (SendPort a)

380spawnChannelLocal proc = do

381spawnChannelLocal proc = do

381− node <- procMsg $ getLocalNode

382+ node <- processNode <$> ask

382 (sport, rport) <- newChan

383 (sport, rport) <- newChan

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

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

384 return sport

385 return sport

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

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

12--

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

14--

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

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

16--

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

18-- have a call to 'remotable':

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

35-- > . Mn.__remoteTable

36-- > $ initRemoteTable

37--

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+--

38-- [Composing static values]

61-- [Composing static values]

39--

62--

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

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

41−-- provide

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

65+-- supports

42--

66--

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

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

44--

68--

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+--

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

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

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

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)

47--

109--

48-- [Closures]

110-- [Closures]

49--

111--

50-- Suppose you have a process

112-- Suppose you have a process

51--

113--

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

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

53--

115--

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

116+-- Then

55--

117--

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

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

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

58--

119--

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

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

60−-- remote node compute a factorial number.

121+-- a number is prime.

61--

122--

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

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

63--

124--

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

76--

137--

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

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

78--

139--

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

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

80--

141--

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

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

83−-- > where

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

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

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

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

86-- > `staticCompose`

144-- > `staticCompose`

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

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

88−--

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

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

111--

147--

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

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

113−--

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

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

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

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.

128--

151--

129-- [Combinators on Closures]

152-- [Combinators on Closures]

130--

153--

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

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

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

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

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

135--

158--

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+--

136-- [Notes]

200-- [Notes]

137--

201--

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

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

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

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

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

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

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

149module Control.Distributed.Process.Closure

213module Control.Distributed.Process.Closure

150− ( -- * User-defined closures

214+ ( -- * Creating static values

151 remotable

215 remotable

152 , mkStatic

216 , mkStatic

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

153 , mkClosure

218 , mkClosure

154 , functionSDict

219 , functionSDict

220+ , functionTDict

155 -- * Primitive operations on static values

221 -- * Primitive operations on static values

156 , staticApply

222 , staticApply

157 , staticDuplicate

223 , staticDuplicate

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

207 ( remotable

273 ( remotable

208 , mkStatic

274 , mkStatic

209 , functionSDict

275 , functionSDict

276+ , functionTDict

210 )

277 )

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

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

212 ( -- Static functionals

279 ( -- Static functionals

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

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

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

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

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

34+import Control.Monad.Reader (ask)

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

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

35import Control.Distributed.Process.Internal.Types

36import Control.Distributed.Process.Internal.Types

36 ( Closure(Closure)

37 ( Closure(Closure)

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

43 , typeOfStaticLabel

44 , typeOfStaticLabel

44 , ProcessId

45 , ProcessId

45 , LocalNode(remoteTable)

46 , LocalNode(remoteTable)

46− , procMsg

47+ , LocalProcess(processNode)

47 , SendPort

48 , SendPort

48 , ReceivePort

49 , ReceivePort

49 )

50 )

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

55 , newChan

56 , newChan

56 )

57 )

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

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

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

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

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

61 ( staticCompose

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

118-- | Resolve a closure

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

120unClosure (Static label) env = do

121− rtable <- remoteTable <$> procMsg getLocalNode

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

122 case resolveClosure rtable label env of

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

124 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

7 remotable

8 , mkStatic

9 , functionSDict

10+ , functionTDict

10 ) where

11 ) where

12import Prelude hiding (lookup)

13import Prelude hiding (lookup)

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

47 , StaticLabel(StaticLabel)

48 , StaticLabel(StaticLabel)

48 , remoteTableLabel

49 , remoteTableLabel

49 , SerializableDict(SerializableDict)

50 , SerializableDict(SerializableDict)

51+ , Process

50 )

52 )

51import Control.Distributed.Process.Internal.Dynamic

53import Control.Distributed.Process.Internal.Dynamic

52 ( Dynamic(..)

54 ( Dynamic(..)

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

74mkStatic :: Name -> Q Exp

76mkStatic :: Name -> Q Exp

75mkStatic = varE . staticName

77mkStatic = varE . staticName

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'.

78functionSDict :: Name -> Q Exp

83functionSDict :: Name -> Q Exp

79functionSDict = varE . sdictName

84functionSDict = varE . sdictName

81−{-

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

82−-- | Create a closure

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

83−--

88+--

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

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

85−-- for a detailed explanation and examples.

86−mkClosure :: Name -> Q Exp

90+functionTDict :: Name -> Q Exp

87−mkClosure = varE . closureName

91+functionTDict = varE . tdictName

88−-}

90--------------------------------------------------------------------------------

93--------------------------------------------------------------------------------

91-- Internal (Template Haskell) --

94-- Internal (Template Haskell) --

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

100

103

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

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

102generateDefs n = do

105generateDefs n = do

106+ proc <- [t| Process |]

103− mType <- getType n

107+ mType <- getType n

104− case mType of

108+ case mType of

105− Just (origName, typ) -> do

109+ Just (origName, typ) -> do

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

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

107− _ -> ([], typ)

111+ _ -> ([], typ)

112+

113+ -- The main "static" entry

108− (static, register) <- do

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

109− static <- generateStatic 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

110− let dyn = case typVars of

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

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

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

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

121+ makeDict (sdictName origName) arg

113− (unsafeCoerce# $(varE origName))

122+ _ ->

114− |]

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+

115− return ( static

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

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

134+ , concat [register, registerSDict, registerTDict]

117 )

135 )

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

118

150

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

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

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

152+ makeDict dictName typ = do

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

153+ sdict <- generateDict dictName typ

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

123− sdict <- generateSDict origName arg

124− let dyn' = [| toDyn (SerializableDict

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

125− :: SerializableDict $(return arg)

126− )

127− |]

128− return ( sdict

155+ return ( sdict

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

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

130− $dyn'

131− |] ]

132− )

133− _ ->

134− return ([], [])

135−

136− return ( static ++ sdict

137− , register ++ registerSDict

138 )

157 )

139− _ ->

158+

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

141

159

142registerStatic :: String -> Dynamic -> RemoteTable -> RemoteTable

160registerStatic :: String -> Dynamic -> RemoteTable -> RemoteTable

143registerStatic label dyn = remoteTableLabel label ^= Just dyn

161registerStatic label dyn = remoteTableLabel label ^= Just dyn

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

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

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

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

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

165

183

166−-- | Generate a function serialization dictionary

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

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

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

168−generateSDict n typ = do

186+generateDict n typ = do

169 sequence

187 sequence

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

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

171− , sfnD (sdictName n)

189+ , sfnD n

172 [| Static $ StaticLabel

190 [| Static $ StaticLabel

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

191+ $(stringE (show n))

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

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

175 |]

193 |]

176 ]

194 ]

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

181sdictName :: Name -> Name

199sdictName :: Name -> Name

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

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

183

201

202+tdictName :: Name -> Name

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

204+

184--------------------------------------------------------------------------------

205--------------------------------------------------------------------------------

185-- Generic Template Haskell auxiliary functions --

206-- Generic Template Haskell auxiliary functions --

186--------------------------------------------------------------------------------

207--------------------------------------------------------------------------------

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

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

1module Control.Distributed.Process.Internal.Node

2− ( runLocalProcess

2+ ( -- * Message sending

3+ sendPayload

4+ , sendBinary

5+ , sendMessage

3 ) where

6 ) where

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)

5−import Control.Monad.Reader (runReaderT)

14+import Control.Monad (unless)

15+import qualified Network.Transport as NT

16+ ( Connection

17+ , send

18+ , defaultConnectHints

19+ , connect

20+ , Reliability(ReliableOrdered)

21+ )

6import Control.Distributed.Process.Internal.Types

22import Control.Distributed.Process.Internal.Types

7− ( LocalNode

23+ ( LocalNode(localState, localEndPoint, localCtrlChan)

8− , LocalProcess

24+ , Identifier

9− , Process(unProcess)

25+ , localConnectionBetween

26+ , nodeAddress

27+ , nodeOf

28+ , messageToPayload

29+ , createMessage

30+ , NCMsg(..)

31+ , ProcessSignal(Died)

32+ , DiedReason(DiedDisconnect)

10 )

33 )

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

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

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

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

63+setupConnBetween node from to = do

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

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

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)

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

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

70import Control.Exception (Exception, throw)

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

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

72import Control.Concurrent.MVar (modifyMVar)

73import Control.Concurrent.Chan (writeChan)

74import Control.Concurrent.STM

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

106 , TypedChannel(..)

107 , SendPortId(..)

108 , Identifier(..)

109− , procMsg

110 , DidUnmonitor(..)

109 , DidUnmonitor(..)

111 , DidUnlinkProcess(..)

110 , DidUnlinkProcess(..)

112 , DidUnlinkNode(..)

111 , DidUnlinkNode(..)

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

114 , WhereIsReply(..)

113 , WhereIsReply(..)

115 , createMessage

114 , createMessage

116 , Static(..)

115 , Static(..)

116+ , runLocalProcess

117 )

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

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

119− ( sendMessage

120− , sendBinary

121− , getLocalNode

122− )

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

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

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

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

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

126

121

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

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

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

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

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

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

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

130+send them msg = do

131+ proc <- ask

132+ liftIO $ sendMessage (processNode proc)

133+ (ProcessIdentifier (processId proc))

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

134+ (ProcessIdentifier them)

135+ msg

136

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

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

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

163

164-- | Send a message on a typed channel

165sendChan :: 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

167

172

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

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

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

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

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

261

266

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

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

263getSelfNode :: Process NodeId

268getSelfNode :: Process NodeId

264−getSelfNode = localNodeId <$> procMsg getLocalNode

269+getSelfNode = localNodeId . processNode <$> ask

265

270

266--------------------------------------------------------------------------------

271--------------------------------------------------------------------------------

267-- Monitoring and linking --

272-- Monitoring and linking --

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

318-- | Catch exceptions within a process

323-- | Catch exceptions within a process

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

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

320catch p h = do

325catch p h = do

321− node <- procMsg getLocalNode

322 lproc <- ask

326 lproc <- ask

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

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

324− run proc = runLocalProcess node proc lproc

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

326

328

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

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

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

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

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

460-- | Deserialize a closure

462-- | Deserialize a closure

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

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

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

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

463− rtable <- remoteTable <$> procMsg getLocalNode

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

464 case resolveClosure rtable label env of

466 case resolveClosure rtable label env of

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

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

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

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

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

507 -> ProcessSignal -- ^ Message to send

509 -> ProcessSignal -- ^ Message to send

508 -> Process ()

510 -> Process ()

509sendCtrlMsg mNid signal = do

511sendCtrlMsg mNid signal = do

510− us <- getSelfPid

512+ proc <- ask

511− let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us

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

512 , ctrlMsgSignal = signal

514 , ctrlMsgSignal = signal

513 }

515 }

514 case mNid of

516 case mNid of

515 Nothing -> do

517 Nothing -> do

516− ctrlChan <- localCtrlChan <$> procMsg getLocalNode

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

517 liftIO $ writeChan ctrlChan msg

519 liftIO $ writeChan ctrlChan msg

518 Just nid ->

520 Just nid ->

521+ liftIO $ sendBinary (processNode proc)

522+ (ProcessIdentifier (processId proc))

519− procMsg $ sendBinary (NodeIdentifier nid) msg

523+ (NodeIdentifier nid)

520−

524+ msg

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

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

17 , LocalProcess(..)

18 , LocalProcessState(..)

19 , Process(..)

20− , procMsg

20+ , runLocalProcess

21 -- * Typed channels

22 , LocalSendPortId

23 , SendPortId(..)

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

58 -- * Node controller internal data types

59 , NCMsg(..)

60 , ProcessSignal(..)

61− -- * MessageT monad

62− , MessageT(..)

63− , MessageState(..)

64 -- * Accessors

61 -- * Accessors

65 , localProcesses

62 , localProcesses

66 , localPidCounter

63 , localPidCounter

67 , localPidUnique

64 , localPidUnique

65+ , localConnections

68 , localProcessWithId

66 , localProcessWithId

67+ , localConnectionBetween

69 , monitorCounter

68 , monitorCounter

70 , spawnCounter

69 , spawnCounter

71 , channelCounter

70 , channelCounter

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

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

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

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

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

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

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

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

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

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

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

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

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

104import Control.Distributed.Process.Serializable

101import Control.Distributed.Process.Serializable

105 ( Fingerprint

102 ( Fingerprint

106 , Serializable

103 , Serializable

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

131 { lpidUnique :: Int32

128 { lpidUnique :: Int32

132 , lpidCounter :: Int32

129 , lpidCounter :: Int32

133 }

130 }

134− deriving (Eq, Ord, Typeable)

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

135−

136−instance Show LocalProcessId where

137− show = show . lpidCounter

138

132

139-- | Process identifier

133-- | Process identifier

140data ProcessId = ProcessId

134data ProcessId = ProcessId

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

187

181

188-- | Local node state

182-- | Local node state

189data LocalNodeState = LocalNodeState

183data LocalNodeState = LocalNodeState

190− { _localProcesses :: Map LocalProcessId LocalProcess

184+ { _localProcesses :: Map LocalProcessId LocalProcess

191− , _localPidCounter :: Int32

185+ , _localPidCounter :: Int32

192− , _localPidUnique :: Int32

186+ , _localPidUnique :: Int32

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

193 }

188 }

194

189

195-- | Processes running on our local node

190-- | Processes running on our local node

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

198 , processId :: ProcessId

193 , processId :: ProcessId

199 , processState :: MVar LocalProcessState

194 , processState :: MVar LocalProcessState

200 , processThread :: ThreadId

195 , processThread :: ThreadId

196+ , processNode :: LocalNode

201 }

197 }

202

198

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+

203-- | Local process state

204data LocalProcessState = LocalProcessState

205 { _monitorCounter :: Int32

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

210

211-- | The Cloud Haskell 'Process' type

212newtype Process a = Process {

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

213+ unProcess :: ReaderT LocalProcess IO a

214 }

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

216

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

218−procMsg = Process . Trans.lift

219−

220--------------------------------------------------------------------------------

217--------------------------------------------------------------------------------

221-- Typed channels --

218-- Typed channels --

222--------------------------------------------------------------------------------

219--------------------------------------------------------------------------------

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

574localPidUnique :: Accessor LocalNodeState Int32

571localPidUnique :: Accessor LocalNodeState Int32

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

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

576

573

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

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

576+

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

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

579

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

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

582+

580monitorCounter :: Accessor LocalProcessState Int32

583monitorCounter :: Accessor LocalProcessState Int32

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

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

582

585

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

597

600

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

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

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

602remoteTableLabel 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 ((<$>))

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

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

40import 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)

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

43import qualified Control.Exception as Exception (catch)

44import Control.Concurrent (forkIO)

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

101 , Closure(..)

102 , Static(..)

103 , Message

104− , MessageT

105 , TypedChannel(..)

104 , TypedChannel(..)

106 , Identifier(..)

105 , Identifier(..)

107 , nodeOf

106 , nodeOf

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

110 , WhereIsReply(..)

109 , WhereIsReply(..)

111 , messageToPayload

110 , messageToPayload

112 , RemoteTable(..)

111 , RemoteTable(..)

112+ , payloadToMessage

113+ , createMessage

114+ , runLocalProcess

113 )

115 )

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

116import 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

116 ( sendBinary

120 ( sendBinary

117 , sendMessage

121 , sendMessage

118 , sendPayload

122 , sendPayload

119− , getLocalNode

120− , runMessageT

121− , payloadToMessage

122− , createMessage

123 )

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)

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

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

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

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

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

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

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

156createBareLocalNode endPoint rtable = do

153createBareLocalNode endPoint rtable = do

157 unq <- randomIO

154 unq <- randomIO

158 state <- newMVar LocalNodeState

155 state <- newMVar LocalNodeState

159− { _localProcesses = Map.empty

156+ { _localProcesses = Map.empty

160− , _localPidCounter = 0

157+ , _localPidCounter = 0

161− , _localPidUnique = unq

158+ , _localPidUnique = unq

159+ , _localConnections = Map.empty

162 }

160 }

163 ctrlChan <- newChan

161 ctrlChan <- newChan

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

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

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

214 , processId = pid

212 , processId = pid

215 , processState = pst

213 , processState = pst

216 , processThread = tid

214 , processThread = tid

215+ , processNode = node

217 }

216 }

218 tid' <- forkIO $ do

217 tid' <- forkIO $ do

219 reason <- Exception.catch

218 reason <- Exception.catch

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

219+ (runLocalProcess lproc proc >> return DiedNormal)

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

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

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

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

223 modifyMVar_ (localState node) $

222 modifyMVar_ (localState node) $

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

287 (Map.insert cid proc procs)

286 (Map.insert cid proc procs)

288 chans

287 chans

289 ctrls

288 ctrls

290− Nothing ->

289+ Nothing ->

291 -- Request for an unknown process.

290 -- Request for an unknown process.

292 --

291 --

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

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

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

367--------------------------------------------------------------------------------

366--------------------------------------------------------------------------------

368

367

369runNodeController :: LocalNode -> IO ()

368runNodeController :: LocalNode -> IO ()

370−runNodeController node =

369+runNodeController =

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

370+ runReaderT (evalStateT (unNC nodeController) initNCState)

372

371

373--------------------------------------------------------------------------------

372--------------------------------------------------------------------------------

374-- Internal data types --

373-- Internal data types --

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

383 , _registry :: Map String ProcessId

382 , _registry :: Map String ProcessId

384 }

383 }

385

384

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

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

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

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

388−

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

390−ncMsg = NC . Trans.lift

391

387

392initNCState :: NCState

388initNCState :: NCState

393initNCState = NCState { _links = Map.empty

389initNCState = NCState { _links = Map.empty

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

402-- [Unified: Table 7]

398-- [Unified: Table 7]

403nodeController :: NC ()

399nodeController :: NC ()

404nodeController = do

400nodeController = do

405− node <- ncMsg getLocalNode

401+ node <- ask

406 forever $ do

402 forever $ do

407 msg <- liftIO $ readChan (localCtrlChan node)

403 msg <- liftIO $ readChan (localCtrlChan node)

408

404

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

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

410 case destNid (ctrlMsgSignal msg) of

406 case destNid (ctrlMsgSignal msg) of

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

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

408+ liftIO $ sendBinary node

409+ (ctrlMsgSender msg)

412− ncMsg $ sendBinary (NodeIdentifier nid') msg

410+ (NodeIdentifier nid')

411+ msg

413 _ ->

412 _ ->

414 return ()

413 return ()

415

414

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

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

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

442 -> NC ()

441 -> NC ()

443ncEffectMonitor from them mRef = do

442ncEffectMonitor from them mRef = do

444− node <- ncMsg getLocalNode

443+ node <- ask

445 shouldLink <-

444 shouldLink <-

446 if not (isLocal node them)

445 if not (isLocal node them)

447 then return True

446 then return True

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

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

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

455 notifyDied from them DiedUnknownId mRef

454 notifyDied from them DiedUnknownId mRef

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

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

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)

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

461+ (NodeIdentifier $ processNodeId from) $

462+ NCMsg

458− { ctrlMsgSender = NodeIdentifier (localNodeId node)

463+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

459− , ctrlMsgSignal = Died them DiedUnknownId

464+ , ctrlMsgSignal = Died them DiedUnknownId

460− }

465+ }

461

466

462-- [Unified: Table 11]

467-- [Unified: Table 11]

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

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

464ncEffectUnlink from them = do

469ncEffectUnlink from them = do

465− node <- ncMsg getLocalNode

470+ node <- ask

466 when (isLocal node (ProcessIdentifier from)) $

471 when (isLocal node (ProcessIdentifier from)) $

467 case them of

472 case them of

468 ProcessIdentifier pid ->

473 ProcessIdentifier pid ->

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

476-- [Unified: Table 11]

481-- [Unified: Table 11]

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

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

478ncEffectUnmonitor from ref = do

483ncEffectUnmonitor from ref = do

479− node <- ncMsg getLocalNode

484+ node <- ask

480 when (isLocal node (ProcessIdentifier from)) $

485 when (isLocal node (ProcessIdentifier from)) $

481 postAsMessage from $ DidUnmonitor ref

486 postAsMessage from $ DidUnmonitor ref

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

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

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

484-- [Unified: Table 12]

489-- [Unified: Table 12]

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

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

486ncEffectDied ident reason = do

491ncEffectDied ident reason = do

487− node <- ncMsg getLocalNode

492+ node <- ask

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

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

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

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

490

495

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

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

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

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

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

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

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

512− node <- ncMsg getLocalNode

517+ node <- ask

513 pid' <- liftIO $ forkProcess node proc

518 pid' <- liftIO $ forkProcess node proc

519+ liftIO $ sendMessage node

520+ (NodeIdentifier (localNodeId node))

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

521+ (ProcessIdentifier pid)

522+ (DidSpawn ref pid')

515

523

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

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

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

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

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

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

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

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

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

527ncEffectWhereIs from label = do

535ncEffectWhereIs from label = do

536+ node <- ask

528 mPid <- gets (^. registryFor label)

537 mPid <- gets (^. registryFor label)

538+ liftIO $ sendMessage node

539+ (NodeIdentifier (localNodeId node))

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

540+ (ProcessIdentifier from)

541+ (WhereIsReply label mPid)

530

542

531-- [Unified: Table 14]

543-- [Unified: Table 14]

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

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

533−ncEffectNamedSend _from label msg = do

545+ncEffectNamedSend from label msg = do

534 mPid <- gets (^. registryFor label)

546 mPid <- gets (^. registryFor label)

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

547 -- 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

548+ node <- ask

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

538 forM_ mPid $ \pid ->

549 forM_ mPid $ \pid ->

550+ liftIO $ sendPayload node

551+ from

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

552+ (ProcessIdentifier pid)

553+ (messageToPayload msg)

540

554

541--------------------------------------------------------------------------------

555--------------------------------------------------------------------------------

542-- Auxiliary --

556-- Auxiliary --

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

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

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

549 -> NC ()

563 -> NC ()

550notifyDied dest src reason mRef = do

564notifyDied dest src reason mRef = do

551− node <- ncMsg getLocalNode

565+ node <- ask

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

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

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

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

554 postAsMessage dest $ ProcessMonitorNotification ref pid reason

568 postAsMessage dest $ ProcessMonitorNotification ref pid reason

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

565 (False, _, _) ->

579 (False, _, _) ->

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

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

567 -- (see [Unified: Table 10]

581 -- (see [Unified: Table 10]

582+ liftIO $ sendBinary node

583+ (NodeIdentifier $ localNodeId node)

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

584+ (NodeIdentifier $ processNodeId dest) $

585+ NCMsg

569− { ctrlMsgSender = NodeIdentifier (localNodeId node)

586+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

570− , ctrlMsgSignal = Died src reason

587+ , ctrlMsgSignal = Died src reason

571− }

588+ }

572

589

573-- | [Unified: Table 8]

590-- | [Unified: Table 8]

574destNid :: ProcessSignal -> Maybe NodeId

591destNid :: ProcessSignal -> Maybe NodeId

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

592-- | Lookup a local closure

609-- | Lookup a local closure

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

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

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

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

595− rtable <- remoteTable <$> ncMsg getLocalNode

612+ rtable <- remoteTable <$> ask

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

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

597

614

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

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

599isValidLocalIdentifier :: Identifier -> NC Bool

616isValidLocalIdentifier :: Identifier -> NC Bool

600isValidLocalIdentifier ident = do

617isValidLocalIdentifier ident = do

601− node <- ncMsg getLocalNode

618+ node <- ask

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

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

603 case ident of

620 case ident of

604 NodeIdentifier nid ->

621 NodeIdentifier nid ->

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

631

648

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

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

633withLocalProc pid p = do

650withLocalProc pid p = do

634− node <- ncMsg getLocalNode

651+ node <- ask

635 liftIO $ do

652 liftIO $ do

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

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

637 -- can silently be dropped

654 -- can silently be dropped

distributed-process/tests/TestClosure.hs+15

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

36wait :: Int -> Process ()

37wait = liftIO . threadDelay

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+

39-- | First argument indicates empty closure environment

45-- | First argument indicates empty closure environment

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

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

41typedPingServer () rport = forever $ do

47typedPingServer () rport = forever $ do

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

49 , 'sdictInt

55 , 'sdictInt

50 , 'wait

56 , 'wait

51 , 'typedPingServer

57 , 'typedPingServer

58+ , 'isPrime

52 ]

59 ]

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

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

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

296 sendChan pingServer sendReply

303 sendChan pingServer sendReply

297 receiveChan receiveReply

304 receiveChan receiveReply

298

305

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+

299main :: IO ()

313main :: IO ()

300main = do

314main = do

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

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

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

314 , ("SpawnInvalid", testSpawnInvalid transport rtable)

328 , ("SpawnInvalid", testSpawnInvalid transport rtable)

315 , ("ClosureExpect", testClosureExpect transport rtable)

329 , ("ClosureExpect", testClosureExpect transport rtable)

316 , ("SpawnChannel", testSpawnChannel transport rtable)

330 , ("SpawnChannel", testSpawnChannel transport rtable)

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

317 ]

332 ]