git.iko.soy/distributed-process
on master help sign in
Focus on 'Static', not 'Closure'
55dfe3aEdsko de Vries
distributed-process/src/Control/Distributed/Process.hs+16−14
compare with master
unifiedsplit
@@ -7,6 +7,7 @@
7-- although some of the details are different. The precise message passing
7-- although some of the details are different. The precise message passing
8-- semantics are based on /A unified semantics for future Erlang/ by Hans
8-- semantics are based on /A unified semantics for future Erlang/ by Hans
9-- Svensson, Lars-Åke Fredlund and Clara Benac Earle.
9-- Svensson, Lars-Åke Fredlund and Clara Benac Earle.
10+{-# LANGUAGE TemplateHaskell #-}
10module Control.Distributed.Process
11module Control.Distributed.Process
11 ( -- * Basic types
12 ( -- * Basic types
12 ProcessId
13 ProcessId
@@ -116,16 +117,15 @@ import Control.Distributed.Process.Internal.Types
116 , SendPortId(..)
117 , SendPortId(..)
117 , WhereIsReply(..)
118 , WhereIsReply(..)
118 )
119 )
119−import Control.Distributed.Process.Internal.Closure.BuiltIn
120− ( sendClosure
121− , expectClosure
122− )
123import Control.Distributed.Process.Internal.Closure.Derived
120import Control.Distributed.Process.Internal.Closure.Derived
124 ( cpSeq
121 ( cpSeq
125 , cpBind
122 , cpBind
126− , linkClosure
123+ , cpSend
127− , unlinkClosure
124+ , cpExpect
128− , serializableDictUnit
125+ , cpLink
126+ , cpUnlink
127+ , sdictUnit
128+ , sdictUnit__static
129 )
129 )
130import Control.Distributed.Process.Internal.Primitives
130import Control.Distributed.Process.Internal.Primitives
131 ( -- Basic messaging
131 ( -- Basic messaging
@@ -179,6 +179,8 @@ import Control.Distributed.Process.Internal.Primitives
179 , expectTimeout
179 , expectTimeout
180 , spawnAsync
180 , spawnAsync
181 )
181 )
182+import Control.Distributed.Process.Serializable (Serializable)
183+import Control.Distributed.Process.Internal.Closure.TH (mkStatic)
182
184
183-- INTERNAL NOTES
185-- INTERNAL NOTES
184--
186--
@@ -253,9 +255,9 @@ spawn nid proc = do
253 -- we call linkNode here, and unlinkNode after, then we might remove a link
255 -- we call linkNode here, and unlinkNode after, then we might remove a link
254 -- that was already set up
256 -- that was already set up
255 mRef <- monitorNode nid
257 mRef <- monitorNode nid
256− sRef <- spawnAsync nid $ linkClosure us
258+ sRef <- spawnAsync nid $ cpLink us
257− `cpSeq` expectClosure serializableDictUnit
259+ `cpSeq` cpExpect $(mkStatic 'sdictUnit)
258− `cpSeq` unlinkClosure us
260+ `cpSeq` cpUnlink us
259 `cpSeq` proc
261 `cpSeq` proc
260 mPid <- receiveWait
262 mPid <- receiveWait
261 [ matchIf (\(DidSpawn ref _) -> ref == sRef)
263 [ matchIf (\(DidSpawn ref _) -> ref == sRef)
@@ -290,10 +292,10 @@ spawnMonitor nid proc = do
290--
292--
291-- We monitor the remote process; if it dies before it can send a reply, we die
293-- We monitor the remote process; if it dies before it can send a reply, we die
292-- too
294-- too
293−call :: SerializableDict a -> NodeId -> Closure (Process a) -> Process a
295+call :: Serializable a => Static (SerializableDict a) -> NodeId -> Closure (Process a) -> Process a
294−call sdict@SerializableDict nid proc = do
296+call dict nid proc = do
295 us <- getSelfPid
297 us <- getSelfPid
296− (_, mRef) <- spawnMonitor nid (proc `cpBind` sendClosure sdict us)
298+ (_, mRef) <- spawnMonitor nid (proc `cpBind` cpSend dict us)
297 -- We are guaranteed to receive the reply before the monitor notification
299 -- We are guaranteed to receive the reply before the monitor notification
298 -- (if a reply is sent at all)
300 -- (if a reply is sent at all)
299 -- NOTE: This might not be true if we switch to unreliable delivery.
301 -- NOTE: This might not be true if we switch to unreliable delivery.
@@ -313,7 +315,7 @@ spawnSupervised :: NodeId
313 -> Process (ProcessId, MonitorRef)
315 -> Process (ProcessId, MonitorRef)
314spawnSupervised nid proc = do
316spawnSupervised nid proc = do
315 us <- getSelfPid
317 us <- getSelfPid
316− them <- spawn nid (linkClosure us `cpSeq` proc)
318+ them <- spawn nid (cpLink us `cpSeq` proc)
317 ref <- monitor them
319 ref <- monitor them
318 return (them, ref)
320 return (them, ref)
319
321
distributed-process/src/Control/Distributed/Process/Closure.hs+54−70
@@ -82,87 +82,71 @@
82-- Some functions (such as 'sendClosure' or 'returnClosure') require an
82-- Some functions (such as 'sendClosure' or 'returnClosure') require an
83-- explicit (reified) serializable dictionary. To create such a dictionary do
83-- explicit (reified) serializable dictionary. To create such a dictionary do
84--
84--
85−-- > serializableDictInt :: SerializableDict Int
85+-- > sdictInt :: SerializableDict Int
86−-- > serializableDictInt = SerializableDict
86+-- > sdictInt = SerializableDict
87--
87--
88−-- and then pass @'serializableDictInt@ to 'remotable'. This will fail if the
88+-- and then pass @'sdictInt@ to 'remotable'. This will fail if the
89-- type is not serializable.
89-- type is not serializable.
90module Control.Distributed.Process.Closure
90module Control.Distributed.Process.Closure
91 ( -- * User-defined closures
91 ( -- * User-defined closures
92 remotable
92 remotable
93− , mkClosure
93+ , mkStatic
94+ -- * Static functionals
95+ , staticConst
96+ , staticCompose
97+ , staticFirst
98+ , staticSecond
99+ , staticSplit
100+ -- * Creating closures
101+ , staticDecode
102+ , staticClosure
103+ , toClosure
94− -- * Process primitives
104+ -- * Closure versions of CH primitives
95− , linkClosure
105+ , cpLink
96− , unlinkClosure
106+ , cpUnlink
97− , sendClosure
107+ , cpSend
98− , returnClosure
108+ , cpExpect
99− , expectClosure
109+ -- * @Closure (Process a)@ as a not-quite-monad
100− -- * Generic closure combinators
110+ , cpReturn
101− , closureApply
102− , closureConst
103− , closureUnit
104− -- * Arrow combinators for processes
105− , CP
106− , cpIntro
107− , cpElim
108− , cpId
109− , cpComp
110− , cpFirst
111− , cpSwap
112− , cpSecond
113− , cpPair
114− , cpCopy
115− , cpFanOut
116− , cpLeft
117− , cpMirror
118− , cpRight
119− , cpEither
120− , cpUntag
121− , cpFanIn
122− -- * Derived combinators for processes
123 , cpBind
111 , cpBind
124 , cpSeq
112 , cpSeq
125− -- * Serialization dictionaries
113+ -- * Serialization dictionaries (and their static versions)
126 , SerializableDict(..)
114 , SerializableDict(..)
127− , serializableDictUnit
115+ , staticApply
116+ , sdictUnit
117+ , sdictUnit__static
118+ , sdictProcessId
119+ , sdictProcessId__static
128 ) where
120 ) where
129
121
130−import Control.Distributed.Process.Internal.Types (SerializableDict(..))
122+import Control.Distributed.Process.Internal.Types
131−import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
123+ ( SerializableDict(..)
132−import Control.Distributed.Process.Internal.Closure.BuiltIn
124+ , staticApply
133− ( sendClosure
134− , returnClosure
135− , expectClosure
136 )
125 )
137−import Control.Distributed.Process.Internal.Closure.Derived
126+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)
127+import Control.Distributed.Process.Internal.Closure.Derived
128+ ( -- Static functionals
129+ staticConst
130+ , staticCompose
131+ , staticFirst
132+ , staticSecond
133+ , staticSplit
134+ -- Creating closures
135+ , staticDecode
136+ , staticClosure
137+ , toClosure
138− ( -- Process primitives
138+ -- Closure versions of CH primitives
139− linkClosure
139+ , cpLink
140− , unlinkClosure
140+ , cpUnlink
141− -- Generic combinators
141+ , cpSend
142− , closureApply
142+ , cpExpect
143− , closureConst
143+ -- @Closure (Process a)@ as a not-quite-monad
144− , closureUnit
144+ , cpReturn
145− -- Arrow combinators for processes
146− , CP
147− , cpIntro
148− , cpElim
149− , cpId
150− , cpComp
151− , cpFirst
152− , cpSwap
153− , cpSecond
154− , cpPair
155− , cpCopy
156− , cpFanOut
157− , cpLeft
158− , cpMirror
159− , cpRight
160− , cpEither
161− , cpUntag
162− , cpFanIn
163− -- Derived process operators
164 , cpBind
145 , cpBind
165 , cpSeq
146 , cpSeq
166− -- Serialization dictionaries
147+ -- Serialization dictionaries (and their static versions)
167− , serializableDictUnit
148+ , sdictUnit
149+ , sdictUnit__static
150+ , sdictProcessId
151+ , sdictProcessId__static
168 )
152 )
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs+11−1
@@ -1,5 +1,7 @@
1-- | Built-in closures
1-- | Built-in closures
2−module Control.Distributed.Process.Internal.Closure.BuiltIn
2+module Control.Distributed.Process.Internal.Closure.BuiltIn () where
3+
4+{-
3 ( -- Combinators
5 ( -- Combinators
4 closureApply
6 closureApply
5 -- TODO
7 -- TODO
@@ -20,6 +22,7 @@ import Control.Distributed.Process.Internal.Types
20 , StaticLabel(..)
22 , StaticLabel(..)
21 )
23 )
22import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
24import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
25+import Control.Distributed.Process.Serializable (Serializable)
23
26
24--------------------------------------------------------------------------------
27--------------------------------------------------------------------------------
25-- Combinators --
28-- Combinators --
@@ -29,6 +32,12 @@ closureApply :: Closure (a -> b) -> Closure a -> Closure b
29closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =
32closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =
30 Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)
33 Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)
31
34
35+closureReturn :: Serializable a => Static (SerializableDict a) -> a -> Closure a
36+closureReturn (Static dict) x = Closure (Static ClosureReturn) (encode (dict, x))
37+
38+closureStatic :: Static a -> Closure a
39+closureStatic (Static dict) = Closure (Static ClosureStatic) (encode dict)
40+
32--------------------------------------------------------------------------------
41--------------------------------------------------------------------------------
33-- Polymorphic closures --
42-- Polymorphic closures --
34-- --
43-- --
@@ -56,3 +65,4 @@ returnClosure SerializableDict val =
56expectClosure :: forall a. SerializableDict a -> Closure (Process a)
65expectClosure :: forall a. SerializableDict a -> Closure (Process a)
57expectClosure SerializableDict =
66expectClosure SerializableDict =
58 Closure (Static ClosureExpect) (encode (typeOf (undefined :: a)))
67 Closure (Static ClosureExpect) (encode (typeOf (undefined :: a)))
68+-}
distributed-process/src/Control/Distributed/Process/Internal/Closure/Derived.hs+289−155
@@ -1,204 +1,338 @@
1-- | Derived (TH generated) closures
1-- | Derived (TH generated) closures
2+{-# LANGUAGE MagicHash #-}
2−module Control.Distributed.Process.Internal.Closure.Derived
3+module Control.Distributed.Process.Internal.Closure.Derived
3− ( remoteTable
4+ ( -- * Static functionals
5+ staticConst
6+ , staticCompose
7+ , staticFirst
8+ , staticSecond
9+ , staticSplit
10+ -- * Creating closures
11+ , staticDecode
12+ , staticClosure
13+ , toClosure
4− -- * Process primitives
14+ -- * Closure versions of CH primitives
5− , linkClosure
15+ , cpLink
6− , unlinkClosure
16+ , cpUnlink
7− -- * Generic combinators
17+ , cpSend
8− , closureApply
18+ , cpExpect
9− , closureConst
19+ -- * @Closure (Process a)@ as a not-quite-monad
10− , closureUnit
20+ , cpReturn
11− -- * Arrow combinators for processes
12− , CP
13− , cpIntro
14− , cpElim
15− , cpId
16− , cpComp
17− , cpFirst
18− , cpSwap
19− , cpSecond
20− , cpPair
21− , cpCopy
22− , cpFanOut
23− , cpLeft
24− , cpMirror
25− , cpRight
26− , cpEither
27− , cpUntag
28− , cpFanIn
29− -- * Derived process operators
30 , cpBind
21 , cpBind
31 , cpSeq
22 , cpSeq
32− -- * Serialization dictionaries
23+ -- * Serialization dictionaries (and their static versions)
33− , serializableDictUnit
24+ , sdictUnit
25+ , sdictUnit__static
26+ , sdictProcessId
27+ , sdictProcessId__static
28+ -- * Runtime support
29+ , __remoteTable
34 ) where
30 ) where
35
31
36−import Prelude hiding (lookup)
32+import Data.Binary (encode, decode)
37−import Data.Typeable (Typeable)
33+import Data.ByteString.Lazy (ByteString, empty)
38−import Data.Tuple (swap)
34+import Data.Typeable (Typeable, typeOf, TypeRep, typeRepTyCon, TyCon)
39import Control.Applicative ((<$>))
35import Control.Applicative ((<$>))
40−import Control.Monad ((>=>))
36+import Control.Monad (join)
37+import Control.Distributed.Process.Serializable (Serializable)
41import Control.Distributed.Process.Internal.Types
38import Control.Distributed.Process.Internal.Types
42− ( Closure(..)
39+ ( Closure(Closure)
40+ , SerializableDict(SerializableDict)
41+ , Static(Static)
42+ , StaticLabel
43 , Process
43 , Process
44− , RemoteTable
44+ , staticApply
45− , SerializableDict(..)
46 , ProcessId
45 , ProcessId
46+ , LocalNode(remoteTable)
47+ , procMsg
47 )
48 )
48−import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
49−import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
50import Control.Distributed.Process.Internal.Primitives
49import Control.Distributed.Process.Internal.Primitives
51 ( link
50 ( link
52 , unlink
51 , unlink
52+ , send
53+ , expect
53 )
54 )
54−import Control.Distributed.Process.Internal.Closure.BuiltIn (closureApply)
55+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)
56+import Control.Distributed.Process.Internal.MessageT (getLocalNode)
57+import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)
58+import Control.Distributed.Process.Internal.Dynamic
59+ ( Dynamic(Dynamic)
60+ , dynBind
61+ , unsafeCoerce#
62+ , dynTypeRep
63+ )
64+import qualified Control.Arrow as Arrow (first, second, (***))
55
65
56--------------------------------------------------------------------------------
66--------------------------------------------------------------------------------
57−-- TH stuff --
67+-- Setup: A number of functions that we will pass to 'remotable' --
58--------------------------------------------------------------------------------
68--------------------------------------------------------------------------------
59
69
60−idUnit :: () -> ()
70+---- Type specializations of monadic operations on Processes -------------------
61−idUnit = id
62
71
63returnProcess :: a -> Process a
72returnProcess :: a -> Process a
64returnProcess = return
73returnProcess = return
65
74
66−kleisliComposeProcess :: (a -> Process b) -> (b -> Process c) -> a -> Process c
67−kleisliComposeProcess = (>=>)
68−
69−firstProcess :: (a -> Process b) -> (a, c) -> Process (b, c)
75+bindProcess :: Process a -> (a -> Process b) -> Process b
70−firstProcess f (a, c) = f a >>= \b -> return (b, c)
76+bindProcess = (>>=)
71−
77+
72−swapProcess :: (a, b) -> Process (b, a)
78+joinProcess :: Process (Process a) -> Process a
73−swapProcess = return . swap
79+joinProcess = join
74−
80+
75−copyProcess :: a -> Process (a, a)
81+---- Functionals ---------------------------------------------------------------
76−copyProcess x = return (x, x)
82+
77−
83+compose :: (b -> c) -> (a -> b) -> a -> c
78−leftProcess :: (a -> Process b) -> Either a c -> Process (Either b c)
84+compose = (.)
79−leftProcess f (Left a) = Left <$> f a
85+
80−leftProcess _ (Right b) = Right <$> return b
86+first :: (a -> b) -> (a, c) -> (b, c)
81−
87+first = Arrow.first
82−mirrorProcess :: Either a b -> Process (Either b a)
88+
83−mirrorProcess (Left x) = Right <$> return x
89+second :: (a -> b) -> (c, a) -> (c, b)
84−mirrorProcess (Right y) = Left <$> return y
90+second = Arrow.second
85−
91+
86−untagProcess :: Either a a -> Process a
92+split :: (a -> b) -> (a' -> b') -> (a, a') -> (b, b')
87−untagProcess (Left x) = return x
93+split = (Arrow.***)
88−untagProcess (Right y) = return y
94+
89−
95+---- Variations on standard or CH functions with an explicit dictionary arg ----
96+
97+decodeDict :: SerializableDict a -> ByteString -> a
98+decodeDict SerializableDict = decode
99+
100+sendDict :: SerializableDict a -> ProcessId -> a -> Process ()
101+sendDict SerializableDict = send
102+
90−serializableDictUnit :: SerializableDict ()
103+expectDict :: SerializableDict a -> Process a
104+expectDict SerializableDict = expect
105+
106+---- Serialization dictionaries ------------------------------------------------
107+
108+-- | Serialization dictionary for '()'
109+--
110+-- Use @$(mkStatic sdictUnit)@ (instead of 'sdictUnit__static')
111+-- to refer to the static dictionary.
112+sdictUnit :: SerializableDict ()
91−serializableDictUnit = SerializableDict
113+sdictUnit = SerializableDict
92−
114+
115+-- | Serialization dictionary for 'ProcessId'
116+--
117+-- Use @$(mkStatic sdictProcessId)@ (instead of 'sdictProcessId__static')
118+-- to refer to the static dictionary.
119+sdictProcessId :: SerializableDict ProcessId
120+sdictProcessId = SerializableDict
121+
122+-- | Specialized serialization dictionary required in 'cpBind'
123+sdictBind :: SerializableDict (((StaticLabel, ByteString), (StaticLabel, ByteString)), TypeRep)
124+sdictBind = SerializableDict
125+
126+---- Some specialised processes necessary to implement the combinators ---------
127+
128+unClosure :: (StaticLabel, ByteString) -> Process Dynamic
129+unClosure (label, env) = do
130+ rtable <- remoteTable <$> procMsg getLocalNode
131+ case resolveClosure rtable label env of
132+ Nothing -> fail "Derived.unClosure: resolveClosure failed"
133+ Just dyn -> return dyn
134+
135+unDynamic :: (Process Dynamic, TypeRep) -> Process a
136+unDynamic (pdyn, typ) = do
137+ Dynamic typ' val <- pdyn
138+ if typ == typ'
139+ then return (unsafeCoerce# val)
140+ else fail $ "unDynamic: cannot match "
141+ ++ show typ'
142+ ++ " against expected type "
143+ ++ show typ
144+
145+bindDyn :: (Process Dynamic, Process Dynamic) -> Process Dynamic
146+bindDyn (px, pf) = do
147+ x <- px
148+ f <- pf
149+ case dynBind tyConProcess bindProcess x f of
150+ Just dyn -> return dyn
151+ Nothing -> fail $ "bindDyn: could not bind "
152+ ++ show (dynTypeRep x)
153+ ++ " to "
154+ ++ show (dynTypeRep f)
155+ where
156+ tyConProcess :: TyCon
157+ tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))
158+
159+---- Finally, the call to remotable --------------------------------------------
160+
93−remotable [ 'const
161+remotable [ -- Monadic operations
94− , 'idUnit
95− , 'returnProcess
162+ 'returnProcess
96− , 'kleisliComposeProcess
163+ , 'bindProcess
97− , 'firstProcess
164+ , 'joinProcess
98− , 'swapProcess
165+ -- Functionals (predefined)
99− , 'copyProcess
166+ , 'const
100− , 'leftProcess
167+ -- Functionals (defined above)
101− , 'mirrorProcess
168+ , 'compose
102− , 'untagProcess
169+ , 'first
170+ , 'second
171+ , 'split
172+ -- CH primitives
103 , 'link
173 , 'link
104 , 'unlink
174 , 'unlink
105− , 'serializableDictUnit
175+ -- Explicit dictionaries
176+ , 'decodeDict
177+ , 'sendDict
178+ , 'expectDict
179+ -- Serialization dictionaries
180+ , 'sdictUnit
181+ , 'sdictProcessId
182+ , 'sdictBind
183+ -- Specialized processes
184+ , 'unClosure
185+ , 'unDynamic
186+ , 'bindDyn
106 ]
187 ]
107
188
108−remoteTable :: RemoteTable -> RemoteTable
109−remoteTable = __remoteTable
110−
111--------------------------------------------------------------------------------
189--------------------------------------------------------------------------------
112−-- Process primitives --
190+-- Static versions of the functionals --
191+-- (We give these explicit names because they are useful outside this module) --
113--------------------------------------------------------------------------------
192--------------------------------------------------------------------------------
114
193
115−-- | Closure version of 'link'
194+-- | Static version of 'const'
116−linkClosure :: ProcessId -> Closure (Process ())
195+staticConst :: (Typeable a, Typeable b) => Static (a -> b -> a)
117−linkClosure = $(mkClosure 'link)
196+staticConst = $(mkStatic 'const)
118
197
119−-- | Closure version of 'unlink'
198+-- | Static version of ('Prelude..')
120−unlinkClosure :: ProcessId -> Closure (Process ())
199+staticCompose :: (Typeable a, Typeable b, Typeable c)
121−unlinkClosure = $(mkClosure 'unlink)
200+ => Static (b -> c) -> Static (a -> b) -> Static (a -> c)
201+staticCompose f x = $(mkStatic 'compose) `staticApply` f `staticApply` x
122
202
123−--------------------------------------------------------------------------------
203+-- | Static version of 'Control.Arrow.first'
124−-- Generic closure combinators --
204+staticFirst :: (Typeable a, Typeable b, Typeable c)
125−--------------------------------------------------------------------------------
205+ => Static ((a -> b) -> (a, c) -> (b, c))
206+staticFirst = $(mkStatic 'first)
126
207
208+-- | Static version of 'Control.Arrow.second'
127−closureConst :: (Typeable a, Typeable b) => Closure (a -> b -> a)
209+staticSecond :: (Typeable a, Typeable b, Typeable c)
128−closureConst = $(mkClosure 'const)
210+ => Static ((a -> b) -> (c, a) -> (c, b))
211+staticSecond = $(mkStatic 'second)
129
212
130−closureUnit :: Closure ()
213+-- | Static version of ('Control.Arrow.***')
131−closureUnit = $(mkClosure 'idUnit) ()
214+staticSplit :: (Typeable a, Typeable b, Typeable c, Typeable d)
215+ => Static (a -> c) -> Static (b -> d) -> Static ((a, b) -> (c, d))
216+staticSplit f g = $(mkStatic 'split) `staticApply` f `staticApply` g
132
217
133--------------------------------------------------------------------------------
218--------------------------------------------------------------------------------
134−-- Arrow combinators for processes --
219+-- Creating closures --
135--------------------------------------------------------------------------------
220--------------------------------------------------------------------------------
136
221
137−type CP a b = Closure (a -> Process b)
222+staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)
138−
223+staticDecode dict = $(mkStatic 'decodeDict) `staticApply` dict
139−cpIntro :: (Typeable a, Typeable b)
140− => Closure (Process b) -> CP a b
141−cpIntro = closureApply closureConst
142
224
143−cpElim :: Typeable a
225+staticClosure :: forall a. Typeable a => Static a -> Closure a
144− => CP () a -> Closure (Process a)
226+staticClosure static = Closure decoder empty
145−cpElim = flip closureApply closureUnit
227+ where
228+ decoder :: Static (ByteString -> a)
229+ decoder = staticConst `staticApply` static
146
230
147−cpId :: Typeable a => CP a a
231+toClosure :: forall a. Serializable a
148−cpId = $(mkClosure 'returnProcess)
232+ => Static (SerializableDict a) -> a -> Closure a
233+toClosure dict x = Closure decoder (encode x)
234+ where
235+ decoder :: Static (ByteString -> a)
236+ decoder = $(mkStatic 'decodeDict) `staticApply` dict
149
237
150−cpComp :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP b c -> CP a c
238+--------------------------------------------------------------------------------
151−cpComp f g = $(mkClosure 'kleisliComposeProcess) `closureApply` f `closureApply` g
239+-- Closure versions of CH primitives --
152−
240+--------------------------------------------------------------------------------
153−cpFirst :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP (a, c) (b, c)
154−cpFirst = closureApply $(mkClosure 'firstProcess)
155−
156−cpSwap :: (Typeable a, Typeable b) => CP (a, b) (b, a)
157−cpSwap = $(mkClosure 'swapProcess)
158−
159−cpSecond :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP (c, a) (c, b)
160−cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap
161−
162−cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')
163− => CP a b -> CP a' b' -> CP (a, a') (b, b')
164−cpPair f g = cpFirst f `cpComp` cpSecond g
165−
166−cpCopy :: Typeable a => CP a (a, a)
167−cpCopy = $(mkClosure 'copyProcess)
168−
169−cpFanOut :: (Typeable a, Typeable b, Typeable c)
170− => CP a b -> CP a c -> CP a (b, c)
171−cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)
172−
173−cpLeft :: (Typeable a, Typeable b, Typeable c)
174− => CP a b -> CP (Either a c) (Either b c)
175−cpLeft = closureApply $(mkClosure 'leftProcess)
176−
177−cpMirror :: (Typeable a, Typeable b) => CP (Either a b) (Either b a)
178−cpMirror = $(mkClosure 'mirrorProcess)
179−
180−cpRight :: (Typeable a, Typeable b, Typeable c)
181− => CP a b -> CP (Either c a) (Either c b)
182−cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror
183−
184−cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')
185− => CP a b -> CP a' b' -> CP (Either a a') (Either b b')
186−cpEither f g = cpLeft f `cpComp` cpRight g
187
241
188−cpUntag :: Typeable a => CP (Either a a) a
242+-- | Closure version of 'link'
189−cpUntag = $(mkClosure 'untagProcess)
243+cpLink :: ProcessId -> Closure (Process ())
244+cpLink pid = Closure decoder (encode pid)
245+ where
246+ decoder :: Static (ByteString -> Process ())
247+ decoder = $(mkStatic 'link)
248+ `staticCompose`
249+ ($(mkStatic 'decodeDict) `staticApply` $(mkStatic 'sdictProcessId))
190
250
191−cpFanIn :: (Typeable a, Typeable b, Typeable c)
251+-- | Closure version of 'unlink'
192− => CP a c -> CP b c -> CP (Either a b) c
252+cpUnlink :: ProcessId -> Closure (Process ())
193−cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag
253+cpUnlink pid = Closure decoder (encode pid)
254+ where
255+ decoder :: Static (ByteString -> Process ())
256+ decoder = $(mkStatic 'unlink)
257+ `staticCompose`
258+ ($(mkStatic 'decodeDict) `staticApply` $(mkStatic 'sdictProcessId))
259+
260+-- | Closure version of 'send'
261+cpSend :: forall a. Typeable a
262+ => Static (SerializableDict a) -> ProcessId -> Closure (a -> Process ())
263+cpSend dict pid = Closure decoder (encode pid)
264+ where
265+ decoder :: Static (ByteString -> a -> Process ())
266+ decoder = ($(mkStatic 'sendDict) `staticApply` dict)
267+ `staticCompose`
268+ ($(mkStatic 'decodeDict) `staticApply` $(mkStatic 'sdictProcessId))
269+
270+-- | Closure version of 'expect'
271+cpExpect :: Typeable a => Static (SerializableDict a) -> Closure (Process a)
272+cpExpect dict = staticClosure ($(mkStatic 'expectDict) `staticApply` dict)
194
273
195--------------------------------------------------------------------------------
274--------------------------------------------------------------------------------
196−-- Some derived operators for processes --
275+-- (Closure . Process) as a not-quite-monad --
197--------------------------------------------------------------------------------
276--------------------------------------------------------------------------------
198
277
278+-- | Not-quite-monadic 'return'
279+cpReturn :: forall a. Serializable a
280+ => Static (SerializableDict a) -> a -> Closure (Process a)
281+cpReturn dict x = Closure decoder (encode x)
282+ where
283+ decoder :: Static (ByteString -> Process a)
284+ decoder = $(mkStatic 'returnProcess)
285+ `staticCompose`
286+ ($(mkStatic 'decodeDict) `staticApply` dict)
287+
288+-- | Not-quite-monadic bind ('>>=')
199−cpBind :: (Typeable a, Typeable b)
289+cpBind :: forall a b. Typeable b
200 => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
290 => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
201−cpBind x f = cpElim $ cpIntro x `cpComp` f
291+cpBind (Closure (Static xlabel) xenv) (Closure (Static flabel) fenv) =
202−
292+ let env :: (((StaticLabel, ByteString), (StaticLabel, ByteString)), TypeRep)
293+ env = (((xlabel, xenv), (flabel, fenv)), typeOf (undefined :: Process b))
294+ in Closure decoder (encode env)
295+ where
296+ decoder :: Static (ByteString -> Process b)
297+ decoder = aux8
298+ `staticCompose`
299+ aux7
300+ `staticCompose`
301+ aux6
302+ `staticCompose`
303+ aux1
304+
305+ aux1 :: Static (ByteString -> (((StaticLabel, ByteString), (StaticLabel, ByteString)), TypeRep))
306+ aux1 = $(mkStatic 'decodeDict) `staticApply` $(mkStatic 'sdictBind)
307+
308+ aux2 :: Static ((StaticLabel, ByteString) -> Process Dynamic)
309+ aux2 = $(mkStatic 'unClosure)
310+
311+ aux3 :: Static (((StaticLabel, ByteString), (StaticLabel, ByteString)) -> (Process Dynamic, Process Dynamic))
312+ aux3 = aux2 `staticSplit` aux2
313+
314+ aux4 :: Static ((Process Dynamic, Process Dynamic) -> Process Dynamic)
315+ aux4 = $(mkStatic 'bindDyn)
316+
317+ aux5 :: Static (((StaticLabel, ByteString), (StaticLabel, ByteString)) -> Process Dynamic)
318+ aux5 = aux4 `staticCompose` aux3
319+
320+ aux6 :: Static ((((StaticLabel, ByteString), (StaticLabel, ByteString)), TypeRep) -> (Process Dynamic, TypeRep))
321+ aux6 = staticFirst `staticApply` aux5
322+
323+ aux7 :: Static ((Process Dynamic, TypeRep) -> Process (Process b))
324+ aux7 = $(mkStatic 'unDynamic)
325+
326+ aux8 :: Static (Process (Process b) -> Process b)
327+ aux8 = $(mkStatic 'joinProcess)
328+
329+cpIntro :: forall a b. (Typeable a, Typeable b)
330+ => Closure (Process b) -> Closure (a -> Process b)
331+cpIntro (Closure static env) = Closure decoder env
332+ where
333+ decoder :: Static (ByteString -> a -> Process b)
334+ decoder = staticConst `staticCompose` static
335+
336+-- | Monadic sequencing ('>>')
203cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
337cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
204cpSeq p q = p `cpBind` cpIntro q
338cpSeq p q = p `cpBind` cpIntro q
distributed-process/src/Control/Distributed/Process/Internal/Closure/Resolution.hs+12−35
@@ -3,16 +3,10 @@ module Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)
3
3
4import Data.Accessor ((^.))
4import Data.Accessor ((^.))
5import Data.ByteString.Lazy (ByteString)
5import Data.ByteString.Lazy (ByteString)
6−import Data.Binary (decode)
7−import Data.Typeable (TypeRep)
8−import Control.Applicative ((<$>))
9import Control.Distributed.Process.Internal.Types
6import Control.Distributed.Process.Internal.Types
10 ( RemoteTable
7 ( RemoteTable
11 , remoteTableLabel
8 , remoteTableLabel
12− , remoteTableDict
13 , StaticLabel(..)
9 , StaticLabel(..)
14− , ProcessId
15− , RuntimeSerializableSupport(..)
16 )
10 )
17import Control.Distributed.Process.Internal.Dynamic
11import Control.Distributed.Process.Internal.Dynamic
18 ( Dynamic(Dynamic)
12 ( Dynamic(Dynamic)
@@ -21,33 +15,16 @@ import Control.Distributed.Process.Internal.Dynamic
21 )
15 )
22import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
16import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
23
17
18+resolveStatic :: RemoteTable -> StaticLabel -> Maybe Dynamic
19+resolveStatic rtable (StaticLabel string typ) = do
20+ Dynamic _ val <- rtable ^. remoteTableLabel string
21+ return (Dynamic typ val)
22+resolveStatic rtable (StaticApply static1 static2) = do
23+ f <- resolveStatic rtable static1
24+ x <- resolveStatic rtable static2
25+ f `dynApply` x
26+
24resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
27resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
25−-- Generic closure combinators
26−resolveClosure rtable ClosureApply env = do
28+resolveClosure rtable static env = do
27− f <- resolveClosure rtable labelf envf
29+ decoder <- resolveStatic rtable static
28− x <- resolveClosure rtable labelx envx
29− f `dynApply` x
30− where
31− (labelf, envf, labelx, envx) = decode env
32−-- Built-in closures
33−resolveClosure rtable ClosureSend env = do
34− rss <- rtable ^. remoteTableDict typ
35− rssSend rss `dynApply` toDyn pid
36− where
37− (typ, pid) = decode env :: (TypeRep, ProcessId)
38−resolveClosure rtable ClosureReturn env = do
39− rss <- rtable ^. remoteTableDict typ
40− rssReturn rss `dynApply` toDyn arg
30+ decoder `dynApply` toDyn env
41− where
42− (typ, arg) = decode env :: (TypeRep, ByteString)
43−resolveClosure rtable ClosureExpect env =
44− rssExpect <$> rtable ^. remoteTableDict typ
45− where
46− typ = decode env :: TypeRep
47−-- User defined closures
48−resolveClosure rtable (UserStatic label) env = do
49− val <- rtable ^. remoteTableLabel label
50− dynApply val (toDyn env)
51−resolveClosure rtable (PolyStatic label) env = do
52− Dynamic _ val <- rtable ^. remoteTableLabel label
53− return (Dynamic (decode env) val)
distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hs+45−16
@@ -5,12 +5,10 @@
5module Control.Distributed.Process.Internal.Closure.TH
5module Control.Distributed.Process.Internal.Closure.TH
6 ( -- * User-level API
6 ( -- * User-level API
7 remotable
7 remotable
8− , mkClosure
8+ , mkStatic
9 ) where
9 ) where
10
10
11import Prelude hiding (lookup)
11import Prelude hiding (lookup)
12−import Data.ByteString.Lazy (ByteString)
13−import Data.Binary (encode, decode)
14import Data.Accessor ((^=))
12import Data.Accessor ((^=))
15import Data.Typeable (typeOf)
13import Data.Typeable (typeOf)
16import Control.Applicative ((<$>))
14import Control.Applicative ((<$>))
@@ -25,7 +23,7 @@ import Language.Haskell.TH
25 -- Algebraic data types
23 -- Algebraic data types
26 , Dec
24 , Dec
27 , Exp
25 , Exp
28− , Type(AppT, ArrowT, ForallT, VarT)
26+ , Type(AppT, ForallT, VarT)
29 , Info(VarI)
27 , Info(VarI)
30 , TyVarBndr(PlainTV, KindedTV)
28 , TyVarBndr(PlainTV, KindedTV)
31 , Pred(ClassP)
29 , Pred(ClassP)
@@ -44,22 +42,14 @@ import Language.Haskell.TH
44 )
42 )
45import Control.Distributed.Process.Internal.Types
43import Control.Distributed.Process.Internal.Types
46 ( RemoteTable
44 ( RemoteTable
47− , Closure(..)
48 , Static(..)
45 , Static(..)
49 , StaticLabel(..)
46 , StaticLabel(..)
50− , Process
51− , ProcessId
52 , remoteTableLabel
47 , remoteTableLabel
53− , remoteTableDict
54− , SerializableDict(..)
55− , RuntimeSerializableSupport(..)
56 )
48 )
57import Control.Distributed.Process.Internal.Dynamic
49import Control.Distributed.Process.Internal.Dynamic
58 ( Dynamic(..)
50 ( Dynamic(..)
59− , toDyn
60 , unsafeCoerce#
51 , unsafeCoerce#
61 )
52 )
62−import Control.Distributed.Process.Internal.Primitives (send, expect)
63
53
64--------------------------------------------------------------------------------
54--------------------------------------------------------------------------------
65-- User-level API --
55-- User-level API --
@@ -73,12 +63,18 @@ remotable ns = do
73 rtable <- createMetaData inserts
63 rtable <- createMetaData inserts
74 return $ concat closures ++ rtable
64 return $ concat closures ++ rtable
75
65
66+-- | Construct a static value
67+mkStatic :: Name -> Q Exp
68+mkStatic = varE . staticName
69+
70+{-
76-- | Create a closure
71-- | Create a closure
77--
72--
78-- See module documentation header for "Control.Distributed.Process.Closure"
73-- See module documentation header for "Control.Distributed.Process.Closure"
79-- for a detailed explanation and examples.
74-- for a detailed explanation and examples.
80mkClosure :: Name -> Q Exp
75mkClosure :: Name -> Q Exp
81mkClosure = varE . closureName
76mkClosure = varE . closureName
77+-}
82
78
83--------------------------------------------------------------------------------
79--------------------------------------------------------------------------------
84-- Internal (Template Haskell) --
80-- Internal (Template Haskell) --
@@ -91,6 +87,42 @@ createMetaData is =
91 __remoteTable = $(compose is)
87 __remoteTable = $(compose is)
92 |]
88 |]
93
89
90+generateDefs :: Name -> Q ([Dec], Q Exp)
91+generateDefs n = do
92+ mType <- getType n
93+ case mType of
94+ Just (origName, typ) -> do
95+ let (typVars, typ') = case typ of ForallT vars [] mono -> (vars, mono)
96+ _ -> ([], typ)
97+ (static, label) <- generateStatic origName typVars typ'
98+ return (static, [| registerStatic $(stringE label) (Dynamic (error "Polymorphic value") (unsafeCoerce# $(varE origName))) |])
99+ _ ->
100+ fail $ "remotable: " ++ show n ++ " not found"
101+
102+registerStatic :: String -> Dynamic -> RemoteTable -> RemoteTable
103+registerStatic label dyn = remoteTableLabel label ^= Just dyn
104+
105+-- | Generate a static value
106+generateStatic :: Name -> [TyVarBndr] -> Type -> Q ([Dec], String)
107+generateStatic n xs typ = do
108+ staticTyp <- [t| Static |]
109+ closure <- sequence
110+ [ sigD (staticName n) (return (ForallT xs (map typeable xs) (staticTyp `AppT` typ)) )
111+ , sfnD (staticName n) [| Static (StaticLabel $(stringE label) (typeOf (undefined :: $(return typ)))) |]
112+ ]
113+ return (closure, label)
114+ where
115+ label :: String
116+ label = show n
117+
118+ typeable :: TyVarBndr -> Pred
119+ typeable (PlainTV v) = ClassP (mkName "Typeable") [VarT v]
120+ typeable (KindedTV v _) = ClassP (mkName "Typeable") [VarT v]
121+
122+staticName :: Name -> Name
123+staticName n = mkName $ nameBase n ++ "__static"
124+
125+{-
94-- | Generate the necessary definitions for one function
126-- | Generate the necessary definitions for one function
95--
127--
96-- Given an (f :: a -> b) in module M, create:
128-- Given an (f :: a -> b) in module M, create:
@@ -154,9 +186,6 @@ generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]
154closureName :: Name -> Name
186closureName :: Name -> Name
155closureName n = mkName $ nameBase n ++ "__closure"
187closureName n = mkName $ nameBase n ++ "__closure"
156
188
157−registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable
158−registerLabel label dyn = remoteTableLabel label ^= Just dyn
159−
160registerSerializableDict :: forall a. SerializableDict a -> RemoteTable -> RemoteTable
189registerSerializableDict :: forall a. SerializableDict a -> RemoteTable -> RemoteTable
161registerSerializableDict SerializableDict =
190registerSerializableDict SerializableDict =
162 let rss = RuntimeSerializableSupport {
191 let rss = RuntimeSerializableSupport {
@@ -165,6 +194,7 @@ registerSerializableDict SerializableDict =
165 , rssExpect = toDyn (expect :: Process a)
194 , rssExpect = toDyn (expect :: Process a)
166 }
195 }
167 in remoteTableDict (typeOf (undefined :: a)) ^= Just rss
196 in remoteTableDict (typeOf (undefined :: a)) ^= Just rss
197+-}
168
198
169--------------------------------------------------------------------------------
199--------------------------------------------------------------------------------
170-- Generic Template Haskell auxiliary functions --
200-- Generic Template Haskell auxiliary functions --
@@ -191,4 +221,3 @@ getType name = do
191-- | Variation on 'funD' which takes a single expression to define the function
221-- | Variation on 'funD' which takes a single expression to define the function
192sfnD :: Name -> Q Exp -> Q Dec
222sfnD :: Name -> Q Exp -> Q Dec
193sfnD n e = funD n [clause [] (normalB e) []]
223sfnD n e = funD n [clause [] (normalB e) []]
194−
distributed-process/src/Control/Distributed/Process/Internal/Dynamic.hs+25−1
@@ -13,14 +13,24 @@ module Control.Distributed.Process.Internal.Dynamic
13 , dynTypeRep
13 , dynTypeRep
14 , dynApply
14 , dynApply
15 , dynApp
15 , dynApp
16+ , dynBind
17+ , dynBind'
16 , GHC.unsafeCoerce#
18 , GHC.unsafeCoerce#
17 ) where
19 ) where
18
20
19−import Data.Typeable (Typeable, TypeRep, typeOf, funResultTy)
21+import Data.Typeable
22+ ( Typeable
23+ , TypeRep
24+ , typeOf
25+ , funResultTy
26+ , TyCon
27+ , splitTyConApp
28+ )
20import qualified GHC.Prim as GHC (Any, unsafeCoerce#)
29import qualified GHC.Prim as GHC (Any, unsafeCoerce#)
21import Data.Maybe (fromMaybe)
30import Data.Maybe (fromMaybe)
22
31
23data Dynamic = Dynamic TypeRep GHC.Any
32data Dynamic = Dynamic TypeRep GHC.Any
33+ deriving Typeable
24
34
25toDyn :: Typeable a => a -> Dynamic
35toDyn :: Typeable a => a -> Dynamic
26toDyn x = Dynamic (typeOf x) (GHC.unsafeCoerce# x)
36toDyn x = Dynamic (typeOf x) (GHC.unsafeCoerce# x)
@@ -55,3 +65,17 @@ dynApp f x = fromMaybe (error typeError) (dynApply f x)
55 typeError = "Type error in dynamic application.\n"
65 typeError = "Type error in dynamic application.\n"
56 ++ "Can't apply function " ++ show f
66 ++ "Can't apply function " ++ show f
57 ++ " to argument " ++ show x
67 ++ " to argument " ++ show x
68+
69+dynBind :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynamic -> Maybe Dynamic
70+dynBind m bind (Dynamic t1 x) (Dynamic t2 f) =
71+ case splitTyConApp t1 of
72+ (m', [a]) | m' == m ->
73+ case funResultTy t2 a of
74+ Just mb -> Just (Dynamic mb (GHC.unsafeCoerce# bind x f))
75+ _ -> Nothing
76+ _ -> Nothing
77+
78+dynBind' :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynamic -> Dynamic
79+dynBind' m bind x f = fromMaybe (error typeError) (dynBind m bind x f)
80+ where
81+ typeError = "Type error in dynamic bind.\nCan't bind " ++ show x ++ " to " ++ show f
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+17−50
@@ -27,10 +27,10 @@ module Control.Distributed.Process.Internal.Types
27 -- * Closures
27 -- * Closures
28 , StaticLabel(..)
28 , StaticLabel(..)
29 , Static(..)
29 , Static(..)
30+ , staticApply
30 , Closure(..)
31 , Closure(..)
31 , RemoteTable(..)
32 , RemoteTable(..)
32 , SerializableDict(..)
33 , SerializableDict(..)
33− , RuntimeSerializableSupport(..)
34 -- * Messages
34 -- * Messages
35 , Message(..)
35 , Message(..)
36 , createMessage
36 , createMessage
@@ -69,9 +69,7 @@ module Control.Distributed.Process.Internal.Types
69 , typedChannels
69 , typedChannels
70 , typedChannelWithId
70 , typedChannelWithId
71 , remoteTableLabels
71 , remoteTableLabels
72− , remoteTableDicts
73 , remoteTableLabel
72 , remoteTableLabel
74− , remoteTableDict
75 ) where
73 ) where
76
74
77import Data.Map (Map)
75import Data.Map (Map)
@@ -110,6 +108,7 @@ import Control.Distributed.Process.Serializable
110 )
108 )
111import Control.Distributed.Process.Internal.CQueue (CQueue)
109import Control.Distributed.Process.Internal.CQueue (CQueue)
112import Control.Distributed.Process.Internal.Dynamic (Dynamic)
110import Control.Distributed.Process.Internal.Dynamic (Dynamic)
111+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
113
112
114--------------------------------------------------------------------------------
113--------------------------------------------------------------------------------
115-- Node and process identifiers --
114-- Node and process identifiers --
@@ -259,33 +258,26 @@ data ReceivePort a =
259-- Closures --
258-- Closures --
260--------------------------------------------------------------------------------
259--------------------------------------------------------------------------------
261
260
262−data StaticLabel =
261+data StaticLabel =
263− UserStatic String
262+ StaticLabel String TypeRep
264− | PolyStatic String
263+ | StaticApply StaticLabel StaticLabel
265− -- Closure combinators
266− | ClosureApply
267− -- Built-in closures
268− | ClosureReturn
269− | ClosureSend
270− | ClosureExpect
271 deriving (Typeable, Show)
264 deriving (Typeable, Show)
272
265
273−-- | A static value is one that is bound at top-level.
266+-- | A static value is top-level bound or the application of two static values
274newtype Static a = Static StaticLabel
267newtype Static a = Static StaticLabel
275 deriving (Typeable, Show)
268 deriving (Typeable, Show)
276
269
270+-- | Apply two static values
271+staticApply :: Static (a -> b) -> Static a -> Static b
272+staticApply (Static f) (Static x) = Static (StaticApply f x)
273+
277-- | A closure is a static value and an encoded environment
274-- | A closure is a static value and an encoded environment
278data Closure a = Closure (Static (BSL.ByteString -> a)) BSL.ByteString
275data Closure a = Closure (Static (BSL.ByteString -> a)) BSL.ByteString
279 deriving (Typeable, Show)
276 deriving (Typeable, Show)
280
277
281−-- | Used to fake 'static' (see paper)
278+-- | Runtime dictionary for 'unstatic' lookups
282data RemoteTable = RemoteTable {
279data RemoteTable = RemoteTable {
283− -- | If the user creates a closure of type @a -> Closure b@ from a function
284− -- @f : a -> b@, then '_remoteTableLabels' should have an entry for "f"
285− -- of type @ByteString -> b@ (basically, @f . encode@)
286− _remoteTableLabels :: Map String Dynamic
280+ _remoteTableLabels :: Map String Dynamic
287− -- | Runtime counterpart to SerializableDict
288− , _remoteTableDicts :: Map TypeRep RuntimeSerializableSupport
289 }
281 }
290
282
291-- | Reification of 'Serializable' (see "Control.Distributed.Process.Closure")
283-- | Reification of 'Serializable' (see "Control.Distributed.Process.Closure")
@@ -293,17 +285,6 @@ data SerializableDict a where
293 SerializableDict :: Serializable a => SerializableDict a
285 SerializableDict :: Serializable a => SerializableDict a
294 deriving (Typeable)
286 deriving (Typeable)
295
287
296−-- | Runtime support for implementing "polymorphic" functions with a
297−-- Serializable qualifier (sendClosure, returnClosure, ..).
298−--
299−-- We don't attempt to keep this minimal, but instead just add functions as
300−-- convenient. This will be replaced anyway once 'static' has been implemented.
301−data RuntimeSerializableSupport = RuntimeSerializableSupport {
302− rssSend :: Dynamic
303− , rssReturn :: Dynamic
304− , rssExpect :: Dynamic
305− }
306−
307--------------------------------------------------------------------------------
288--------------------------------------------------------------------------------
308-- Messages --
289-- Messages --
309--------------------------------------------------------------------------------
290--------------------------------------------------------------------------------
@@ -544,22 +525,14 @@ instance Binary Identifier where
544 _ -> fail "Identifier.get: invalid"
525 _ -> fail "Identifier.get: invalid"
545
526
546instance Binary StaticLabel where
527instance Binary StaticLabel where
547− put (UserStatic string) = putWord8 0 >> put string
528+ put (StaticLabel string typ) = putWord8 0 >> put string >> put typ
548− put (PolyStatic string) = putWord8 1 >> put string
529+ put (StaticApply label1 label2) = putWord8 1 >> put label1 >> put label2
549− put ClosureApply = putWord8 2
550− put ClosureReturn = putWord8 3
551− put ClosureSend = putWord8 4
552− put ClosureExpect = putWord8 5
553 get = do
530 get = do
554 header <- getWord8
531 header <- getWord8
555 case header of
532 case header of
556− 0 -> UserStatic <$> get
533+ 0 -> StaticLabel <$> get <*> get
557− 1 -> PolyStatic <$> get
534+ 1 -> StaticApply <$> get <*> get
558− 2 -> return ClosureApply
559− 3 -> return ClosureReturn
560− 4 -> return ClosureSend
561− 5 -> return ClosureExpect
562− _ -> fail "StaticLabel.get: invalid"
535+ _ -> fail "StaticLabel.get: invalid"
563
536
564instance Binary WhereIsReply where
537instance Binary WhereIsReply where
565 put (WhereIsReply label mPid) = put label >> put mPid
538 put (WhereIsReply label mPid) = put label >> put mPid
@@ -599,15 +572,9 @@ typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid
599remoteTableLabels :: Accessor RemoteTable (Map String Dynamic)
572remoteTableLabels :: Accessor RemoteTable (Map String Dynamic)
600remoteTableLabels = accessor _remoteTableLabels (\ls tbl -> tbl { _remoteTableLabels = ls })
573remoteTableLabels = accessor _remoteTableLabels (\ls tbl -> tbl { _remoteTableLabels = ls })
601
574
602−remoteTableDicts :: Accessor RemoteTable (Map TypeRep RuntimeSerializableSupport)
603−remoteTableDicts = accessor _remoteTableDicts (\ds tbl -> tbl { _remoteTableDicts = ds })
604−
605remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic)
575remoteTableLabel :: String -> Accessor RemoteTable (Maybe Dynamic)
606remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label
576remoteTableLabel label = remoteTableLabels >>> DAC.mapMaybe label
607
577
608−remoteTableDict :: TypeRep -> Accessor RemoteTable (Maybe RuntimeSerializableSupport)
609−remoteTableDict label = remoteTableDicts >>> DAC.mapMaybe label
610−
611--------------------------------------------------------------------------------
578--------------------------------------------------------------------------------
612-- MessageT monad --
579-- MessageT monad --
613--------------------------------------------------------------------------------
580--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Node.hs+2−2
@@ -121,14 +121,14 @@ import Control.Distributed.Process.Internal.Dynamic (fromDynamic)
121import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)
121import Control.Distributed.Process.Internal.Closure.Resolution (resolveClosure)
122import Control.Distributed.Process.Internal.Node (runLocalProcess)
122import Control.Distributed.Process.Internal.Node (runLocalProcess)
123import Control.Distributed.Process.Internal.Primitives (expect, register)
123import Control.Distributed.Process.Internal.Primitives (expect, register)
124−import qualified Control.Distributed.Process.Internal.Closure.Derived as Derived (remoteTable)
124+import qualified Control.Distributed.Process.Internal.Closure.Derived as Derived (__remoteTable)
125
125
126--------------------------------------------------------------------------------
126--------------------------------------------------------------------------------
127-- Initialization --
127-- Initialization --
128--------------------------------------------------------------------------------
128--------------------------------------------------------------------------------
129
129
130initRemoteTable :: RemoteTable
130initRemoteTable :: RemoteTable
131−initRemoteTable = Derived.remoteTable $ RemoteTable Map.empty Map.empty
131+initRemoteTable = Derived.__remoteTable $ RemoteTable Map.empty
132
132
133-- | Initialize a new local node.
133-- | Initialize a new local node.
134--
134--
distributed-process/tests/TestClosure.hs+83−58
@@ -1,7 +1,8 @@
1module Main where
1module Main where
2
2
3−import qualified Data.ByteString.Lazy as BSL (empty)
3+import Data.ByteString.Lazy (ByteString, empty)
4−import Data.Typeable (Typeable)
4+import Data.Typeable (Typeable, typeOf)
5+import Data.Binary (encode)
5import Control.Monad (join, replicateM)
6import Control.Monad (join, replicateM)
6import Control.Exception (IOException, throw)
7import Control.Exception (IOException, throw)
7import Control.Concurrent (forkIO, threadDelay)
8import Control.Concurrent (forkIO, threadDelay)
@@ -15,8 +16,12 @@ import Control.Distributed.Process.Node
15import Control.Distributed.Process.Internal.Types (Closure(..), Static(..), StaticLabel(..))
16import Control.Distributed.Process.Internal.Types (Closure(..), Static(..), StaticLabel(..))
16import TestAuxiliary
17import TestAuxiliary
17
18
18−serializableDictInt :: SerializableDict Int
19+sdictInt :: SerializableDict Int
19−serializableDictInt = SerializableDict
20+sdictInt = SerializableDict
21+
22+factorial :: Int -> Process Int
23+factorial 0 = return 1
24+factorial n = (n *) <$> factorial (n - 1)
20
25
21addInt :: Int -> Int -> Int
26addInt :: Int -> Int -> Int
22addInt x y = x + y
27addInt x y = x + y
@@ -24,42 +29,76 @@ addInt x y = x + y
24putInt :: Int -> MVar Int -> IO ()
29putInt :: Int -> MVar Int -> IO ()
25putInt = flip putMVar
30putInt = flip putMVar
26
31
27−sendInt :: ProcessId -> Int -> Process ()
28−sendInt = send
29−
30sendPid :: ProcessId -> Process ()
32sendPid :: ProcessId -> Process ()
31sendPid toPid = do
33sendPid toPid = do
32 fromPid <- getSelfPid
34 fromPid <- getSelfPid
33 send toPid fromPid
35 send toPid fromPid
34
36
35−factorial :: Int -> Process Int
37+wait :: Int -> Process ()
36−factorial 0 = return 1
38+wait = liftIO . threadDelay
37−factorial n = (n *) <$> factorial (n - 1)
38
39
39−factorialOf :: () -> Int -> Process Int
40−factorialOf () = factorial
40+remotable [ 'factorial
41+ , 'addInt
42+ , 'putInt
43+ , 'sendPid
44+ , 'sdictInt
45+ , 'wait
46+ ]
41
47
42−returnForTestApply :: (Closure (Int -> Process Int), Int) -> (Closure (Int -> Process Int), Int)
48+factorialClosure :: Int -> Closure (Process Int)
43−returnForTestApply = id
49+factorialClosure n = Closure decoder (encode n)
50+ where
51+ decoder :: Static (ByteString -> Process Int)
52+ decoder = $(mkStatic 'factorial) `staticCompose` staticDecode $(mkStatic 'sdictInt)
44
53
45−wait :: Int -> Process ()
54+addIntClosure :: Int -> Closure (Int -> Int)
46−wait = liftIO . threadDelay
55+addIntClosure n = Closure decoder (encode n)
56+ where
57+ decoder :: Static (ByteString -> Int -> Int)
58+ decoder = $(mkStatic 'addInt) `staticCompose` staticDecode $(mkStatic 'sdictInt)
59+
60+putIntClosure :: Int -> Closure (MVar Int -> IO ())
61+putIntClosure n = Closure decoder (encode n)
62+ where
63+ decoder :: Static (ByteString -> MVar Int -> IO ())
64+ decoder = $(mkStatic 'putInt) `staticCompose` staticDecode $(mkStatic 'sdictInt)
65+
66+sendPidClosure :: ProcessId -> Closure (Process ())
67+sendPidClosure pid = Closure decoder (encode pid)
68+ where
69+ decoder :: Static (ByteString -> Process ())
70+ decoder = $(mkStatic 'sendPid) `staticCompose` staticDecode $(mkStatic 'sdictProcessId)
71+
72+sendFac :: Int -> ProcessId -> Closure (Process ())
73+sendFac n pid = factorialClosure n `cpBind` cpSend $(mkStatic 'sdictInt) pid
74+
75+factorialOf :: Closure (Int -> Process Int)
76+factorialOf = staticClosure $(mkStatic 'factorial)
77+
78+factorial' :: Int -> Closure (Process Int)
79+factorial' n = cpReturn $(mkStatic 'sdictInt) n `cpBind` factorialOf
80+
81+waitClosure :: Int -> Closure (Process ())
82+waitClosure n = Closure decoder (encode n)
83+ where
84+ decoder :: Static (ByteString -> Process ())
85+ decoder = $(mkStatic 'wait) `staticCompose` staticDecode $(mkStatic 'sdictInt)
86+
87+testUnclosure :: Transport -> RemoteTable -> IO ()
88+testUnclosure transport rtable = do
89+ node <- newLocalNode transport rtable
90+ runProcess node $ do
91+ 120 <- join . unClosure $ factorialClosure 5
92+ return ()
47
93
48−first :: (a, b) -> a
94+testBind :: Transport -> RemoteTable -> IO ()
49−first (x, _) = x
95+testBind transport rtable = do
50−
96+ node <- newLocalNode transport rtable
51−$(remotable
97+ runProcess node $ do
52− [ 'addInt
98+ us <- getSelfPid
53− , 'putInt
99+ join . unClosure $ sendFac 6 us
54− , 'sendInt
100+ (720 :: Int) <- expect
55− , 'sendPid
101+ return ()
56− , 'factorial
57− , 'factorialOf
58− , 'returnForTestApply
59− , 'wait
60− , 'serializableDictInt
61− , 'first
62− ])
63
102
64testSendPureClosure :: Transport -> RemoteTable -> IO ()
103testSendPureClosure :: Transport -> RemoteTable -> IO ()
65testSendPureClosure transport rtable = do
104testSendPureClosure transport rtable = do
@@ -71,14 +110,14 @@ testSendPureClosure transport rtable = do
71 addr <- forkProcess node $ do
110 addr <- forkProcess node $ do
72 cl <- expect
111 cl <- expect
73 fn <- unClosure cl :: Process (Int -> Int)
112 fn <- unClosure cl :: Process (Int -> Int)
74− 11 <- return $ fn 6
113+ 13 <- return $ fn 6
75 liftIO $ putMVar serverDone ()
114 liftIO $ putMVar serverDone ()
76 putMVar serverAddr addr
115 putMVar serverAddr addr
77
116
78 forkIO $ do
117 forkIO $ do
79 node <- newLocalNode transport rtable
118 node <- newLocalNode transport rtable
80 theirAddr <- readMVar serverAddr
119 theirAddr <- readMVar serverAddr
81− runProcess node $ send theirAddr ($(mkClosure 'addInt) 5)
120+ runProcess node $ send theirAddr (addIntClosure 7)
82
121
83 takeMVar serverDone
122 takeMVar serverDone
84
123
@@ -102,7 +141,7 @@ testSendIOClosure transport rtable = do
102 forkIO $ do
141 forkIO $ do
103 node <- newLocalNode transport rtable
142 node <- newLocalNode transport rtable
104 theirAddr <- readMVar serverAddr
143 theirAddr <- readMVar serverAddr
105− runProcess node $ send theirAddr ($(mkClosure 'putInt) 5)
144+ runProcess node $ send theirAddr (putIntClosure 5)
106
145
107 takeMVar serverDone
146 takeMVar serverDone
108
147
@@ -124,7 +163,7 @@ testSendProcClosure transport rtable = do
124 theirAddr <- readMVar serverAddr
163 theirAddr <- readMVar serverAddr
125 runProcess node $ do
164 runProcess node $ do
126 pid <- getSelfPid
165 pid <- getSelfPid
127− send theirAddr ($(mkClosure 'sendInt) pid)
166+ send theirAddr (cpSend $(mkStatic 'sdictInt) pid)
128 5 <- expect :: Process Int
167 5 <- expect :: Process Int
129 liftIO $ putMVar clientDone ()
168 liftIO $ putMVar clientDone ()
130
169
@@ -144,7 +183,7 @@ testSpawn transport rtable = do
144 nid <- readMVar serverNodeAddr
183 nid <- readMVar serverNodeAddr
145 runProcess node $ do
184 runProcess node $ do
146 pid <- getSelfPid
185 pid <- getSelfPid
147− pid' <- spawn nid ($(mkClosure 'sendPid) pid)
186+ pid' <- spawn nid (sendPidClosure pid)
148 pid'' <- expect
187 pid'' <- expect
149 True <- return $ pid' == pid''
188 True <- return $ pid' == pid''
150 liftIO $ putMVar clientDone ()
189 liftIO $ putMVar clientDone ()
@@ -164,26 +203,11 @@ testCall transport rtable = do
164 node <- newLocalNode transport rtable
203 node <- newLocalNode transport rtable
165 nid <- readMVar serverNodeAddr
204 nid <- readMVar serverNodeAddr
166 runProcess node $ do
205 runProcess node $ do
167− (120 :: Int) <- call serializableDictInt nid ($(mkClosure 'factorial) 5)
206+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
168 liftIO $ putMVar clientDone ()
207 liftIO $ putMVar clientDone ()
169
208
170 takeMVar clientDone
209 takeMVar clientDone
171
210
172−sendFac :: Int -> ProcessId -> Closure (Process ())
173−sendFac n pid = $(mkClosure 'factorial) n `cpBind` $(mkClosure 'sendInt) pid
174−
175−factorial' :: Int -> Closure (Process Int)
176−factorial' n = returnClosure serializableDictInt n `cpBind` $(mkClosure 'factorialOf) ()
177−
178−testBind :: Transport -> RemoteTable -> IO ()
179−testBind transport rtable = do
180− node <- newLocalNode transport rtable
181− runProcess node $ do
182− us <- getSelfPid
183− join . unClosure $ sendFac 6 us
184− (720 :: Int) <- expect
185− return ()
186−
187testCallBind :: Transport -> RemoteTable -> IO ()
211testCallBind :: Transport -> RemoteTable -> IO ()
188testCallBind transport rtable = do
212testCallBind transport rtable = do
189 serverNodeAddr <- newEmptyMVar
213 serverNodeAddr <- newEmptyMVar
@@ -197,7 +221,7 @@ testCallBind transport rtable = do
197 node <- newLocalNode transport rtable
221 node <- newLocalNode transport rtable
198 nid <- readMVar serverNodeAddr
222 nid <- readMVar serverNodeAddr
199 runProcess node $ do
223 runProcess node $ do
200− (120 :: Int) <- call serializableDictInt nid (factorial' 5)
224+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
201 liftIO $ putMVar clientDone ()
225 liftIO $ putMVar clientDone ()
202
226
203 takeMVar clientDone
227 takeMVar clientDone
@@ -227,7 +251,7 @@ testSpawnSupervised transport rtable = do
227 forkProcess node1 $ do
251 forkProcess node1 $ do
228 us <- getSelfPid
252 us <- getSelfPid
229 liftIO $ putMVar superPid us
253 liftIO $ putMVar superPid us
230− (child, _ref) <- spawnSupervised (localNodeId node2) ($(mkClosure 'wait) 1000000)
254+ (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)
231 liftIO $ do
255 liftIO $ do
232 putMVar childPid child
256 putMVar childPid child
233 threadDelay 500000 -- Give the child a chance to link to us
257 threadDelay 500000 -- Give the child a chance to link to us
@@ -251,7 +275,7 @@ testSpawnInvalid :: Transport -> RemoteTable -> IO ()
251testSpawnInvalid transport rtable = do
275testSpawnInvalid transport rtable = do
252 node <- newLocalNode transport rtable
276 node <- newLocalNode transport rtable
253 runProcess node $ do
277 runProcess node $ do
254− (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (UserStatic "ThisDoesNotExist")) BSL.empty)
278+ (pid, ref) <- spawnMonitor (localNodeId node) (Closure (Static (StaticLabel "ThisDoesNotExist" (typeOf ()))) empty)
255 ProcessMonitorNotification ref' pid' _reason <- expect
279 ProcessMonitorNotification ref' pid' _reason <- expect
256 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
280 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
257 True <- return $ ref' == ref && pid == pid'
281 True <- return $ ref' == ref && pid == pid'
@@ -263,7 +287,7 @@ testClosureExpect transport rtable = do
263 runProcess node $ do
287 runProcess node $ do
264 nodeId <- getSelfNode
288 nodeId <- getSelfNode
265 us <- getSelfPid
289 us <- getSelfPid
266− them <- spawn nodeId $ expectClosure serializableDictInt `cpBind` $(mkClosure 'sendInt) us
290+ them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `cpBind` cpSend $(mkStatic 'sdictInt) us
267 send them (1234 :: Int)
291 send them (1234 :: Int)
268 (1234 :: Int) <- expect
292 (1234 :: Int) <- expect
269 return ()
293 return ()
@@ -273,12 +297,13 @@ main = do
273 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
297 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
274 let rtable = __remoteTable initRemoteTable
298 let rtable = __remoteTable initRemoteTable
275 runTests
299 runTests
300+ [ ("Unclosure", testUnclosure transport rtable)
301+ , ("Bind", testBind transport rtable)
276− [ ("SendPureClosure", testSendPureClosure transport rtable)
302+ , ("SendPureClosure", testSendPureClosure transport rtable)
277 , ("SendIOClosure", testSendIOClosure transport rtable)
303 , ("SendIOClosure", testSendIOClosure transport rtable)
278 , ("SendProcClosure", testSendProcClosure transport rtable)
304 , ("SendProcClosure", testSendProcClosure transport rtable)
279 , ("Spawn", testSpawn transport rtable)
305 , ("Spawn", testSpawn transport rtable)
280 , ("Call", testCall transport rtable)
306 , ("Call", testCall transport rtable)
281− , ("Bind", testBind transport rtable)
282 , ("CallBind", testCallBind transport rtable)
307 , ("CallBind", testCallBind transport rtable)
283 , ("Seq", testSeq transport rtable)
308 , ("Seq", testSeq transport rtable)
284 , ("SpawnSupervised", testSpawnSupervised transport rtable)
309 , ("SpawnSupervised", testSpawnSupervised transport rtable)