634ae01 · distributed-process

git.iko.soy/distributed-process

Final cleanup and documentation

commit634ae017e074523abb3812d0066d8b50b6fa822a authorEdsko de Vries <edsko@edsko.net> dateTue 10 Jul 2012, 13:24 (14 years ago) parentf4d9f03

9 files changed+128−166

compare with master

unifiedsplit

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

@@ -117,15 +117,15 @@ import Control.Distributed.Process.Internal.Types

117117 , WhereIsReply(..)

118118 )

119119import Control.Distributed.Process.Internal.Closure.BuiltIn

120− ( linkClosure

121− , unlinkClosure

122− , sendClosure

120+ ( sendClosure

123121 , expectClosure

124− , serializableDictUnit

125122 )

126123import Control.Distributed.Process.Internal.Closure.Derived

127124 ( cpSeq

128125 , cpBind

126+ , linkClosure

127+ , unlinkClosure

128+ , serializableDictUnit

129129 )

130130import Control.Distributed.Process.Internal.Primitives

131131 ( -- Basic messaging

distributed-process/src/Control/Distributed/Process/Closure.hs+53−30

@@ -1,24 +1,6 @@

11-- | Implementation of 'Closure' that works around the absence of 'static'.

22--

3−-- [Built-in closures]

4−--

5−-- We offer a number of standard commonly useful closures.

6−--

7−-- [Closure combinators]

8−--

9−-- Closures combinators allow to create closures from other closures. For

10−-- example, 'spawnSupervised' is defined as follows:

11−--

12−-- > spawnSupervised :: NodeId

13−-- > -> Closure (Process ())

14−-- > -> Process (ProcessId, MonitorRef)

15−-- > spawnSupervised nid proc = do

16−-- > us <- getSelfPid

17−-- > them <- spawn nid (linkClosure us `cpSeq` proc)

18−-- > ref <- monitor them

19−-- > return (them, ref)

20−--

21−-- [User-defined closures]

3+-- [User-defined monomorphic closures]

224--

235-- Suppose we have a monomorphic function

246--

@@ -38,8 +20,8 @@

3820--

3921-- > unClosure :: Typeable a => Closure a -> Process a

4022--

41−-- In general, given a monomorphic function @f :: a -> b@ the corresponding

42−-- function @$(mkClosure 'f)@ will have type @a -> Closure b@.

23+-- In general, given a monomorphic function @f :: T1 -> T2@ the corresponding

24+-- function @$(mkClosure 'f)@ will have type @T1 -> Closure T2@.

4325--

4426-- The call to 'remotable' will also generate a function

4527--

@@ -57,6 +39,44 @@

5739--

5840-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.

5941--

42+-- [User-defined polymorphic closures]

43+--

44+-- Suppose we have a polymorphic function

45+--

46+-- > first :: forall a b. (a, b) -> a

47+-- > first (x, _) = x

48+--

49+-- Then the Template Haskell splice

50+--

51+-- > remotable ['first]

52+--

53+-- creates a function

54+--

55+-- > $(mkClosure 'first) :: forall a b. (Typeable a, Typeable b)

56+-- > => Closure (a -> b -> a)

57+--

58+-- In general, given a polymorphic function @f :: forall a1 .. an. T@ the

59+-- corresponding function @$(mkClosure 'f)@ will have type

60+-- @forall a1 .. an. (Typeable a1, .., Typeable an) => Closure T@.

61+--

62+-- [Built-in closures]

63+--

64+-- We offer a number of standard commonly useful closures.

65+--

66+-- [Closure combinators]

67+--

68+-- Closures combinators allow to create closures from other closures. For

69+-- example, 'spawnSupervised' is defined as follows:

70+--

71+-- > spawnSupervised :: NodeId

72+-- > -> Closure (Process ())

73+-- > -> Process (ProcessId, MonitorRef)

74+-- > spawnSupervised nid proc = do

75+-- > us <- getSelfPid

76+-- > them <- spawn nid (linkClosure us `cpSeq` proc)

77+-- > ref <- monitor them

78+-- > return (them, ref)

79+--

6080-- [Serializable Dictionaries]

6181--

6282-- Some functions (such as 'sendClosure' or 'returnClosure') require an

@@ -71,8 +91,7 @@ module Control.Distributed.Process.Closure

7191 ( -- * User-defined closures

7292 remotable

7393 , mkClosure

74− , SerializableDict(..)

75− -- * Built-in closures

94+ -- * Process primitives

7695 , linkClosure

7796 , unlinkClosure

7897 , sendClosure

@@ -100,24 +119,27 @@ module Control.Distributed.Process.Closure

100119 , cpEither

101120 , cpUntag

102121 , cpFanIn

103− , cpApply

104122 -- * Derived combinators for processes

105123 , cpBind

106124 , cpSeq

125+ -- * Serialization dictionaries

126+ , SerializableDict(..)

127+ , serializableDictUnit

107128 ) where

108129

109130import Control.Distributed.Process.Internal.Types (SerializableDict(..))

110131import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)

111132import Control.Distributed.Process.Internal.Closure.BuiltIn

112− ( linkClosure

113− , unlinkClosure

114− , sendClosure

133+ ( sendClosure

115134 , returnClosure

116135 , expectClosure

117136 )

118137import Control.Distributed.Process.Internal.Closure.Derived

119− ( -- Generic combinators

120− closureApply

138+ ( -- Process primitives

139+ linkClosure

140+ , unlinkClosure

141+ -- Generic combinators

142+ , closureApply

121143 , closureConst

122144 , closureUnit

123145 -- Arrow combinators for processes

@@ -138,8 +160,9 @@ import Control.Distributed.Process.Internal.Closure.Derived

138160 , cpEither

139161 , cpUntag

140162 , cpFanIn

141− , cpApply

142163 -- Derived process operators

143164 , cpBind

144165 , cpSeq

166+ -- Serialization dictionaries

167+ , serializableDictUnit

145168 )

distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+8−27

@@ -1,12 +1,8 @@

1−{-# LANGUAGE TemplateHaskell #-}

1+-- | Built-in closures

22module Control.Distributed.Process.Internal.Closure.BuiltIn

3− ( -- * Runtime support for the builtin closures

4− remoteTable

5− -- * Serialization dictionaries

6− , serializableDictUnit

7− -- * Closures

8− , linkClosure

9− , unlinkClosure

3+ ( -- Combinators

4+ closureApply

5+ -- TODO

106 , sendClosure

117 , returnClosure

128 , expectClosure

@@ -14,39 +10,24 @@ module Control.Distributed.Process.Internal.Closure.BuiltIn

1410

1511import Data.Binary (encode)

1612import Data.Typeable (Typeable, typeOf)

17−import Control.Distributed.Process.Internal.Primitives (link, unlink)

1813import Control.Distributed.Process.Internal.Types

1914 ( ProcessId

2015 , Closure

2116 , Process

22− , RemoteTable

2317 , SerializableDict(..)

2418 , Closure(..)

2519 , Static(..)

2620 , StaticLabel(..)

2721 )

28−import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)

2922import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

3023

31−serializableDictUnit :: SerializableDict ()

32−serializableDictUnit = SerializableDict

33−

34−remotable ['link, 'unlink, 'serializableDictUnit]

35−

36−remoteTable :: RemoteTable -> RemoteTable

37−remoteTable = __remoteTable

38−

3924--------------------------------------------------------------------------------

40−-- TH generated closures --

25+-- Combinators --

4126--------------------------------------------------------------------------------

4227

43−-- | Closure version of 'link'

44−linkClosure :: ProcessId -> Closure (Process ())

45−linkClosure = $(mkClosure 'link)

46−

47−-- | Closure version of 'unlink'

48−unlinkClosure :: ProcessId -> Closure (Process ())

49−unlinkClosure = $(mkClosure 'unlink)

28+closureApply :: Closure (a -> b) -> Closure a -> Closure b

29+closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =

30+ Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)

5031

5132--------------------------------------------------------------------------------

5233-- Polymorphic closures --

distributed-process/src/Control/Distributed/Process/Internal/Closure/Derived.hs+42−44

@@ -1,6 +1,9 @@

11-- | Derived (TH generated) closures

22module Control.Distributed.Process.Internal.Closure.Derived

33 ( remoteTable

4+ -- * Process primitives

5+ , linkClosure

6+ , unlinkClosure

47 -- * Generic combinators

58 , closureApply

69 , closureConst

@@ -23,28 +26,32 @@ module Control.Distributed.Process.Internal.Closure.Derived

2326 , cpEither

2427 , cpUntag

2528 , cpFanIn

26− , cpApply

2729 -- * Derived process operators

2830 , cpBind

2931 , cpSeq

32+ -- * Serialization dictionaries

33+ , serializableDictUnit

3034 ) where

3135

3236import Prelude hiding (lookup)

33−import Data.Binary (encode)

34−import Data.Typeable (typeOf, Typeable)

37+import Data.Typeable (Typeable)

3538import Data.Tuple (swap)

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

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

3841import Control.Distributed.Process.Internal.Types

3942 ( Closure(..)

40− , Static(..)

41− , StaticLabel(..)

4243 , Process

4344 , RemoteTable

45+ , SerializableDict(..)

46+ , ProcessId

4447 )

4548import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

4649import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)

47−import Control.Distributed.Process.Internal.Primitives (unClosure)

50+import Control.Distributed.Process.Internal.Primitives

51+ ( link

52+ , unlink

53+ )

54+import Control.Distributed.Process.Internal.Closure.BuiltIn (closureApply)

4855

4956--------------------------------------------------------------------------------

5057-- TH stuff --

@@ -80,10 +87,8 @@ untagProcess :: Either a a -> Process a

8087untagProcess (Left x) = return x

8188untagProcess (Right y) = return y

8289

83−applyClosureProcess :: (Typeable a, Typeable b) => (Closure (a -> Process b), a) -> Process b

84−applyClosureProcess (procClosure, a) = do

85− proc <- unClosure procClosure

86− proc a

90+serializableDictUnit :: SerializableDict ()

91+serializableDictUnit = SerializableDict

8792

8893remotable [ 'const

8994 , 'idUnit

@@ -95,18 +100,29 @@ remotable [ 'const

95100 , 'leftProcess

96101 , 'mirrorProcess

97102 , 'untagProcess

103+ , 'link

104+ , 'unlink

105+ , 'serializableDictUnit

98106 ]

99107

100108remoteTable :: RemoteTable -> RemoteTable

101109remoteTable = __remoteTable

102110

103111--------------------------------------------------------------------------------

104−-- Generic closure combinators --

112+-- Process primitives --

105113--------------------------------------------------------------------------------

106114

107−closureApply :: Closure (a -> b) -> Closure a -> Closure b

108−closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =

109− Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)

115+-- | Closure version of 'link'

116+linkClosure :: ProcessId -> Closure (Process ())

117+linkClosure = $(mkClosure 'link)

118+

119+-- | Closure version of 'unlink'

120+unlinkClosure :: ProcessId -> Closure (Process ())

121+unlinkClosure = $(mkClosure 'unlink)

122+

123+--------------------------------------------------------------------------------

124+-- Generic closure combinators --

125+--------------------------------------------------------------------------------

110126

111127closureConst :: (Typeable a, Typeable b) => Closure (a -> b -> a)

112128closureConst = $(mkClosure 'const)

@@ -128,47 +144,40 @@ cpElim :: Typeable a

128144 => CP () a -> Closure (Process a)

129145cpElim = flip closureApply closureUnit

130146

131−cpId :: forall a. Typeable a

132− => CP a a

147+cpId :: Typeable a => CP a a

133148cpId = $(mkClosure 'returnProcess)

134149

135−cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)

136− => CP a b -> CP b c -> CP a c

150+cpComp :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP b c -> CP a c

137151cpComp f g = $(mkClosure 'kleisliComposeProcess) `closureApply` f `closureApply` g

138152

139−cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)

140− => CP a b -> CP (a, c) (b, c)

153+cpFirst :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP (a, c) (b, c)

141154cpFirst = closureApply $(mkClosure 'firstProcess)

142155

143−cpSwap :: forall a b. (Typeable a, Typeable b)

144− => CP (a, b) (b, a)

156+cpSwap :: (Typeable a, Typeable b) => CP (a, b) (b, a)

145157cpSwap = $(mkClosure 'swapProcess)

146158

147−cpSecond :: (Typeable a, Typeable b, Typeable c)

148− => CP a b -> CP (c, a) (c, b)

159+cpSecond :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP (c, a) (c, b)

149160cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap

150161

151162cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')

152− => CP a b -> CP a' b' -> CP (a, a') (b, b')

163+ => CP a b -> CP a' b' -> CP (a, a') (b, b')

153164cpPair f g = cpFirst f `cpComp` cpSecond g

154165

155−cpCopy :: forall a. Typeable a

156− => CP a (a, a)

166+cpCopy :: Typeable a => CP a (a, a)

157167cpCopy = $(mkClosure 'copyProcess)

158168

159−cpFanOut :: (Typeable a, Typeable b, Typeable c)

169+cpFanOut :: (Typeable a, Typeable b, Typeable c)

160170 => CP a b -> CP a c -> CP a (b, c)

161171cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)

162172

163−cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)

173+cpLeft :: (Typeable a, Typeable b, Typeable c)

164174 => CP a b -> CP (Either a c) (Either b c)

165175cpLeft = closureApply $(mkClosure 'leftProcess)

166176

167−cpMirror :: forall a b. (Typeable a, Typeable b)

168− => CP (Either a b) (Either b a)

177+cpMirror :: (Typeable a, Typeable b) => CP (Either a b) (Either b a)

169178cpMirror = $(mkClosure 'mirrorProcess)

170179

171−cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)

180+cpRight :: (Typeable a, Typeable b, Typeable c)

172181 => CP a b -> CP (Either c a) (Either c b)

173182cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror

174183

@@ -176,24 +185,13 @@ cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')

176185 => CP a b -> CP a' b' -> CP (Either a a') (Either b b')

177186cpEither f g = cpLeft f `cpComp` cpRight g

178187

179−cpUntag :: forall a. Typeable a

180− => CP (Either a a) a

188+cpUntag :: Typeable a => CP (Either a a) a

181189cpUntag = $(mkClosure 'untagProcess)

182190

183191cpFanIn :: (Typeable a, Typeable b, Typeable c)

184192 => CP a c -> CP b c -> CP (Either a b) c

185193cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag

186194

187−cpApply :: forall a b. (Typeable a, Typeable b)

188− => CP (CP a b, a) b

189−cpApply = Closure (Static CpApply) $ encode ( typeOf aux

190− , typeOf (undefined :: a)

191− , typeOf (undefined :: Process b)

192− )

193− where

194− aux :: (Closure (a -> Process b), a) -> Process b

195− aux = undefined

196−

197195--------------------------------------------------------------------------------

198196-- Some derived operators for processes --

199197--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Closure/Resolution.hs+7−27

@@ -12,21 +12,23 @@ import Control.Distributed.Process.Internal.Types

1212 , remoteTableDict

1313 , StaticLabel(..)

1414 , ProcessId

15− , Process

16− , Closure(Closure)

17− , Static(Static)

1815 , RuntimeSerializableSupport(..)

1916 )

2017import Control.Distributed.Process.Internal.Dynamic

2118 ( Dynamic(Dynamic)

2219 , toDyn

23− , dynApp

2420 , dynApply

25− , unsafeCoerce#

2621 )

2722import Control.Distributed.Process.Internal.TypeRep () -- Binary instances

2823

2924resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic

25+-- Generic closure combinators

26+resolveClosure rtable ClosureApply env = do

27+ f <- resolveClosure rtable labelf envf

28+ x <- resolveClosure rtable labelx envx

29+ f `dynApply` x

30+ where

31+ (labelf, envf, labelx, envx) = decode env

3032-- Built-in closures

3133resolveClosure rtable ClosureSend env = do

3234 rss <- rtable ^. remoteTableDict typ

@@ -42,28 +44,6 @@ resolveClosure rtable ClosureExpect env =

4244 rssExpect <$> rtable ^. remoteTableDict typ

4345 where

4446 typ = decode env :: TypeRep

45−-- Generic closure combinators

46−resolveClosure rtable ClosureApply env = do

47− f <- resolveClosure rtable labelf envf

48− x <- resolveClosure rtable labelx envx

49− f `dynApply` x

50− where

51− (labelf, envf, labelx, envx) = decode env

52−resolveClosure rtable CpApply env =

53− return $ Dynamic typApply (unsafeCoerce# cpApply)

54− where

55− cpApply :: forall a b. (Closure (a -> Process b), a) -> Process b

56− cpApply (Closure (Static flabel) fenv, x) = do

57− let Just f = resolveClosure rtable flabel fenv

58− Dynamic typResult val = f `dynApp` Dynamic typA (unsafeCoerce# x)

59− if typResult == typProcB

60− then unsafeCoerce# val

61− else error $ "Type error in cpApply: "

62− ++ "mismatch between " ++ show typResult

63− ++ " and " ++ show typProcB

64−

65− (typApply, typA, typProcB) = decode env

66−

6747-- User defined closures

6848resolveClosure rtable (UserStatic label) env = do

6949 val <- rtable ^. remoteTableLabel label

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

@@ -75,8 +75,8 @@ remotable ns = do

7575

7676-- | Create a closure

7777--

78−-- If @f :: a -> b@ then @mkClosure :: a -> Closure b@. Make sure to pass 'f'

79−-- as an argument to 'remotable' too.

78+-- See module documentation header for "Control.Distributed.Process.Closure"

79+-- for a detailed explanation and examples.

8080mkClosure :: Name -> Q Exp

8181mkClosure = varE . closureName

8282

distributed-process/src/Control/Distributed/Process/Internal/Types.hs+10−13

@@ -262,13 +262,12 @@ data ReceivePort a =

262262data StaticLabel =

263263 UserStatic String

264264 | PolyStatic String

265+ -- Closure combinators

266+ | ClosureApply

265267 -- Built-in closures

266268 | ClosureReturn

267269 | ClosureSend

268270 | ClosureExpect

269− -- Generic closure combinators

270− | ClosureApply

271− | CpApply

272271 deriving (Typeable, Show)

273272

274273-- | A static value is one that is bound at top-level.

@@ -547,21 +546,19 @@ instance Binary Identifier where

547546instance Binary StaticLabel where

548547 put (UserStatic string) = putWord8 0 >> put string

549548 put (PolyStatic string) = putWord8 1 >> put string

550− put ClosureReturn = putWord8 2

551− put ClosureSend = putWord8 3

552− put ClosureExpect = putWord8 4

553− put ClosureApply = putWord8 6

554− put CpApply = putWord8 17

549+ put ClosureApply = putWord8 2

550+ put ClosureReturn = putWord8 3

551+ put ClosureSend = putWord8 4

552+ put ClosureExpect = putWord8 5

555553 get = do

556554 header <- getWord8

557555 case header of

558556 0 -> UserStatic <$> get

559557 1 -> PolyStatic <$> get

560− 2 -> return ClosureReturn

561− 3 -> return ClosureSend

562− 4 -> return ClosureExpect

563− 6 -> return ClosureApply

564− 17 -> return CpApply

558+ 2 -> return ClosureApply

559+ 3 -> return ClosureReturn

560+ 4 -> return ClosureSend

561+ 5 -> return ClosureExpect

565562 _ -> fail "StaticLabel.get: invalid"

566563

567564instance Binary WhereIsReply where

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

@@ -121,17 +121,14 @@ import Control.Distributed.Process.Internal.Dynamic (fromDynamic)

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

122122import Control.Distributed.Process.Internal.Node (runLocalProcess)

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

124−import qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn (remoteTable)

125−import qualified Control.Distributed.Process.Internal.Closure.Derived as Combinators (remoteTable)

124+import qualified Control.Distributed.Process.Internal.Closure.Derived as Derived (remoteTable)

126125

127126--------------------------------------------------------------------------------

128127-- Initialization --

129128--------------------------------------------------------------------------------

130129

131130initRemoteTable :: RemoteTable

132−initRemoteTable = BuiltIn.remoteTable

133− . Combinators.remoteTable

134− $ RemoteTable Map.empty Map.empty

131+initRemoteTable = Derived.remoteTable $ RemoteTable Map.empty Map.empty

135132

136133-- | Initialize a new local node.

137134--

distributed-process/tests/TestClosure.hs−14

@@ -212,19 +212,6 @@ testSeq transport rtable = do

212212 720 :: Int <- expect

213213 return ()

214214

215−testApply :: Transport -> RemoteTable -> IO ()

216−testApply transport rtable = do

217− node <- newLocalNode transport rtable

218− runProcess node $ do

219− (120 :: Int) <- join (unClosure bar)

220− return ()

221− where

222− bar :: Closure (Process Int)

223− bar = closureApply cpApply foo

224−

225− foo :: Closure (Closure (Int -> Process Int), Int)

226− foo = $(mkClosure 'returnForTestApply) ($(mkClosure 'factorialOf) (), 5)

227−

228215-- Test 'spawnSupervised'

229216--

230217-- Set up a supervisor, spawn a child, then have a third process monitor the

@@ -294,7 +281,6 @@ main = do

294281 , ("Bind", testBind transport rtable)

295282 , ("CallBind", testCallBind transport rtable)

296283 , ("Seq", testSeq transport rtable)

297− , ("Apply", testApply transport rtable)

298284 , ("SpawnSupervised", testSpawnSupervised transport rtable)

299285 , ("SpawnInvalid", testSpawnInvalid transport rtable)

300286 , ("ClosureExpect", testClosureExpect transport rtable)