@@ -1,204 +1,338 @@
11-- | Derived (TH generated) closures
2−module Control.Distributed.Process.Internal.Closure.Derived
3− ( remoteTable
4− -- * Process primitives
5− , linkClosure
6− , unlinkClosure
7− -- * Generic combinators
8− , closureApply
9− , closureConst
10− , closureUnit
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
2+{-# LANGUAGE MagicHash #-}
3+module Control.Distributed.Process.Internal.Closure.Derived
4+ ( -- * Static functionals
5+ staticConst
6+ , staticCompose
7+ , staticFirst
8+ , staticSecond
9+ , staticSplit
10+ -- * Creating closures
11+ , staticDecode
12+ , staticClosure
13+ , toClosure
14+ -- * Closure versions of CH primitives
15+ , cpLink
16+ , cpUnlink
17+ , cpSend
18+ , cpExpect
19+ -- * @Closure (Process a)@ as a not-quite-monad
20+ , cpReturn
3021 , cpBind
3122 , cpSeq
32− -- * Serialization dictionaries
33− , serializableDictUnit
23+ -- * Serialization dictionaries (and their static versions)
24+ , sdictUnit
25+ , sdictUnit__static
26+ , sdictProcessId
27+ , sdictProcessId__static
28+ -- * Runtime support
29+ , __remoteTable
3430 ) where
3531
36−import Prelude hiding (lookup)
37−import Data.Typeable (Typeable)
38−import Data.Tuple (swap)
32+import Data.Binary (encode, decode)
33+import Data.ByteString.Lazy (ByteString, empty)
34+import Data.Typeable (Typeable, typeOf, TypeRep, typeRepTyCon, TyCon)
3935import Control.Applicative ((<$>))
40−import Control.Monad ((>=>))
36+import Control.Monad (join)
37+import Control.Distributed.Process.Serializable (Serializable)
4138import Control.Distributed.Process.Internal.Types
42− ( Closure(..)
39+ ( Closure(Closure)
40+ , SerializableDict(SerializableDict)
41+ , Static(Static)
42+ , StaticLabel
4343 , Process
44− , RemoteTable
45− , SerializableDict(..)
44+ , staticApply
4645 , ProcessId
46+ , LocalNode(remoteTable)
47+ , procMsg
4748 )
48−import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
49−import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
5049import Control.Distributed.Process.Internal.Primitives
5150 ( link
5251 , unlink
52+ , send
53+ , expect
5354 )
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, (***))
5565
5666--------------------------------------------------------------------------------
57−-- TH stuff --
67+-- Setup: A number of functions that we will pass to 'remotable' --
5868--------------------------------------------------------------------------------
5969
60−idUnit :: () -> ()
61−idUnit = id
70+---- Type specializations of monadic operations on Processes -------------------
6271
6372returnProcess :: a -> Process a
6473returnProcess = return
6574
66−kleisliComposeProcess :: (a -> Process b) -> (b -> Process c) -> a -> Process c
67−kleisliComposeProcess = (>=>)
68−
69−firstProcess :: (a -> Process b) -> (a, c) -> Process (b, c)
70−firstProcess f (a, c) = f a >>= \b -> return (b, c)
71−
72−swapProcess :: (a, b) -> Process (b, a)
73−swapProcess = return . swap
74−
75−copyProcess :: a -> Process (a, a)
76−copyProcess x = return (x, x)
77−
78−leftProcess :: (a -> Process b) -> Either a c -> Process (Either b c)
79−leftProcess f (Left a) = Left <$> f a
80−leftProcess _ (Right b) = Right <$> return b
81−
82−mirrorProcess :: Either a b -> Process (Either b a)
83−mirrorProcess (Left x) = Right <$> return x
84−mirrorProcess (Right y) = Left <$> return y
85−
86−untagProcess :: Either a a -> Process a
87−untagProcess (Left x) = return x
88−untagProcess (Right y) = return y
89−
90−serializableDictUnit :: SerializableDict ()
91−serializableDictUnit = SerializableDict
92−
93−remotable [ 'const
94− , 'idUnit
95− , 'returnProcess
96− , 'kleisliComposeProcess
97− , 'firstProcess
98− , 'swapProcess
99− , 'copyProcess
100− , 'leftProcess
101− , 'mirrorProcess
102− , 'untagProcess
75+bindProcess :: Process a -> (a -> Process b) -> Process b
76+bindProcess = (>>=)
77+
78+joinProcess :: Process (Process a) -> Process a
79+joinProcess = join
80+
81+---- Functionals ---------------------------------------------------------------
82+
83+compose :: (b -> c) -> (a -> b) -> a -> c
84+compose = (.)
85+
86+first :: (a -> b) -> (a, c) -> (b, c)
87+first = Arrow.first
88+
89+second :: (a -> b) -> (c, a) -> (c, b)
90+second = Arrow.second
91+
92+split :: (a -> b) -> (a' -> b') -> (a, a') -> (b, b')
93+split = (Arrow.***)
94+
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+
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 ()
113+sdictUnit = SerializableDict
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+
161+remotable [ -- Monadic operations
162+ 'returnProcess
163+ , 'bindProcess
164+ , 'joinProcess
165+ -- Functionals (predefined)
166+ , 'const
167+ -- Functionals (defined above)
168+ , 'compose
169+ , 'first
170+ , 'second
171+ , 'split
172+ -- CH primitives
103173 , 'link
104174 , '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
106187 ]
107188
108−remoteTable :: RemoteTable -> RemoteTable
109−remoteTable = __remoteTable
110−
111189--------------------------------------------------------------------------------
112−-- Process primitives --
190+-- Static versions of the functionals --
191+-- (We give these explicit names because they are useful outside this module) --
113192--------------------------------------------------------------------------------
114193
115−-- | Closure version of 'link'
116−linkClosure :: ProcessId -> Closure (Process ())
117−linkClosure = $(mkClosure 'link)
194+-- | Static version of 'const'
195+staticConst :: (Typeable a, Typeable b) => Static (a -> b -> a)
196+staticConst = $(mkStatic 'const)
118197
119−-- | Closure version of 'unlink'
120−unlinkClosure :: ProcessId -> Closure (Process ())
121−unlinkClosure = $(mkClosure 'unlink)
198+-- | Static version of ('Prelude..')
199+staticCompose :: (Typeable a, Typeable b, Typeable c)
200+ => Static (b -> c) -> Static (a -> b) -> Static (a -> c)
201+staticCompose f x = $(mkStatic 'compose) `staticApply` f `staticApply` x
122202
123−--------------------------------------------------------------------------------
124−-- Generic closure combinators --
125−--------------------------------------------------------------------------------
203+-- | Static version of 'Control.Arrow.first'
204+staticFirst :: (Typeable a, Typeable b, Typeable c)
205+ => Static ((a -> b) -> (a, c) -> (b, c))
206+staticFirst = $(mkStatic 'first)
126207
127−closureConst :: (Typeable a, Typeable b) => Closure (a -> b -> a)
128−closureConst = $(mkClosure 'const)
208+-- | Static version of 'Control.Arrow.second'
209+staticSecond :: (Typeable a, Typeable b, Typeable c)
210+ => Static ((a -> b) -> (c, a) -> (c, b))
211+staticSecond = $(mkStatic 'second)
129212
130−closureUnit :: Closure ()
131−closureUnit = $(mkClosure 'idUnit) ()
213+-- | Static version of ('Control.Arrow.***')
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
132217
133218--------------------------------------------------------------------------------
134−-- Arrow combinators for processes --
219+-- Creating closures --
135220--------------------------------------------------------------------------------
136221
137−type CP a b = Closure (a -> Process b)
138−
139−cpIntro :: (Typeable a, Typeable b)
140− => Closure (Process b) -> CP a b
141−cpIntro = closureApply closureConst
222+staticDecode :: Typeable a => Static (SerializableDict a) -> Static (ByteString -> a)
223+staticDecode dict = $(mkStatic 'decodeDict) `staticApply` dict
142224
143−cpElim :: Typeable a
144− => CP () a -> Closure (Process a)
145−cpElim = flip closureApply closureUnit
225+staticClosure :: forall a. Typeable a => Static a -> Closure a
226+staticClosure static = Closure decoder empty
227+ where
228+ decoder :: Static (ByteString -> a)
229+ decoder = staticConst `staticApply` static
146230
147−cpId :: Typeable a => CP a a
148−cpId = $(mkClosure 'returnProcess)
231+toClosure :: forall a. Serializable a
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
149237
150−cpComp :: (Typeable a, Typeable b, Typeable c) => CP a b -> CP b c -> CP a c
151−cpComp f g = $(mkClosure 'kleisliComposeProcess) `closureApply` f `closureApply` g
152−
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
238+--------------------------------------------------------------------------------
239+-- Closure versions of CH primitives --
240+--------------------------------------------------------------------------------
187241
188−cpUntag :: Typeable a => CP (Either a a) a
189−cpUntag = $(mkClosure 'untagProcess)
242+-- | Closure version of 'link'
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))
190250
191−cpFanIn :: (Typeable a, Typeable b, Typeable c)
192− => CP a c -> CP b c -> CP (Either a b) c
193−cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag
251+-- | Closure version of 'unlink'
252+cpUnlink :: ProcessId -> Closure (Process ())
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)
194273
195274--------------------------------------------------------------------------------
196−-- Some derived operators for processes --
275+-- (Closure . Process) as a not-quite-monad --
197276--------------------------------------------------------------------------------
198277
199−cpBind :: (Typeable a, Typeable b)
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 ('>>=')
289+cpBind :: forall a b. Typeable b
200290 => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
201−cpBind x f = cpElim $ cpIntro x `cpComp` f
202−
291+cpBind (Closure (Static xlabel) xenv) (Closure (Static flabel) fenv) =
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 ('>>')
203337cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
204338cpSeq p q = p `cpBind` cpIntro q