git.iko.soy/distributed-process
on master help sign in
Put infrastructure in place for spawnChannel
00ef03fEdsko de Vries
distributed-process/src/Control/Distributed/Process.hs+20−10
compare with master
unifiedsplit
@@ -87,10 +87,12 @@ module Control.Distributed.Process
87 , spawnSupervised
87 , spawnSupervised
88 , spawnLink
88 , spawnLink
89 , spawnMonitor
89 , spawnMonitor
90+ , spawnChannel
90 , DidSpawn(..)
91 , DidSpawn(..)
91 ) where
92 ) where
92
93
93import Prelude hiding (catch)
94import Prelude hiding (catch)
95+import Data.Typeable (Typeable)
94import Control.Monad.IO.Class (liftIO)
96import Control.Monad.IO.Class (liftIO)
95import Control.Distributed.Process.Internal.Types
97import Control.Distributed.Process.Internal.Types
96 ( RemoteTable
98 ( RemoteTable
@@ -123,8 +125,14 @@ import Control.Distributed.Process.Internal.Closure.CP
123 , cpExpect
125 , cpExpect
124 , cpLink
126 , cpLink
125 , cpUnlink
127 , cpUnlink
128+ , cpNewChan
129+ , cpCancelL
130+ , cpSplit
131+ )
132+import Control.Distributed.Process.Internal.Closure.Static
133+ ( sdictUnit
134+ , sdictSendPort
126 )
135 )
127−import Control.Distributed.Process.Internal.Closure.Static (sdictUnit)
128import Control.Distributed.Process.Internal.Primitives
136import Control.Distributed.Process.Internal.Primitives
129 ( -- Basic messaging
137 ( -- Basic messaging
130 send
138 send
@@ -316,16 +324,18 @@ spawnSupervised nid proc = do
316 ref <- monitor them
324 ref <- monitor them
317 return (them, ref)
325 return (them, ref)
318
326
319−{-
320−spawnChannel :: forall a. Typeable a
327+spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
321− => NodeId
328+ -> NodeId
322 -> Closure (ReceivePort a -> Process ())
329 -> Closure (ReceivePort a -> Process ())
323 -> Process (SendPort a)
330 -> Process (SendPort a)
324−spawnChannel nid proc = do
331+spawnChannel dict nid proc = do
325 us <- getSelfPid
332 us <- getSelfPid
326− undefined
333+ spawn nid (go us)
334+ expect
327 where
335 where
328− rproc :: ProcessId -> Closure (Process ((), ReceivePort a))
336+ go :: ProcessId -> Closure (Process ())
329− rproc pid = newChanClosure
337+ go pid = cpNewChan dict
330− `cpBind` cpFirst (sendClosure undefined pid)
338+ `cpBind`
331−-}
339+ (cpSend (sdictSendPort dict) pid `cpSplit` proc)
340+ `cpBind`
341+ cpCancelL
distributed-process/src/Control/Distributed/Process/Closure.hs+6
@@ -102,6 +102,8 @@ module Control.Distributed.Process.Closure
102 , staticFirst
102 , staticFirst
103 , staticSecond
103 , staticSecond
104 , staticSplit
104 , staticSplit
105+ -- * Static constants
106+ , staticUnit
105 -- * Creating closures
107 -- * Creating closures
106 , staticDecode
108 , staticDecode
107 , staticClosure
109 , staticClosure
@@ -120,6 +122,7 @@ module Control.Distributed.Process.Closure
120 , SerializableDict(..)
122 , SerializableDict(..)
121 , sdictUnit
123 , sdictUnit
122 , sdictProcessId
124 , sdictProcessId
125+ , sdictSendPort
123 ) where
126 ) where
124
127
125import Control.Distributed.Process.Internal.Types
128import Control.Distributed.Process.Internal.Types
@@ -139,6 +142,8 @@ import Control.Distributed.Process.Internal.Closure.Static
139 , staticFirst
142 , staticFirst
140 , staticSecond
143 , staticSecond
141 , staticSplit
144 , staticSplit
145+ -- Constants
146+ , staticUnit
142 -- Creating closures
147 -- Creating closures
143 , staticDecode
148 , staticDecode
144 , staticClosure
149 , staticClosure
@@ -146,6 +151,7 @@ import Control.Distributed.Process.Internal.Closure.Static
146 -- Serialization dictionaries (and their static versions)
151 -- Serialization dictionaries (and their static versions)
147 , sdictUnit
152 , sdictUnit
148 , sdictProcessId
153 , sdictProcessId
154+ , sdictSendPort
149 )
155 )
150import Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure)
156import Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure)
151import Control.Distributed.Process.Internal.Closure.CP
157import Control.Distributed.Process.Internal.Closure.CP
distributed-process/src/Control/Distributed/Process/Internal/Closure/CP.hs+140−48
@@ -1,8 +1,19 @@
1-- | Combinator for process closures
1-- | Combinator for process closures
2{-# LANGUAGE MagicHash #-}
2{-# LANGUAGE MagicHash #-}
3module Control.Distributed.Process.Internal.Closure.CP
3module Control.Distributed.Process.Internal.Closure.CP
4+ ( -- * Definition of CP and the generalized arrow combinators
5+ CP
6+ , cpIntro
7+ , cpElim
8+ , cpId
9+ , cpComp
10+ , cpFirst
11+ , cpSecond
12+ , cpSplit
13+ , cpCancelL
14+ , cpCancelR
4− ( -- * Closure versions of CH primitives
15+ -- * Closure versions of CH primitives
5− cpLink
16+ , cpLink
6 , cpUnlink
17 , cpUnlink
7 , cpSend
18 , cpSend
8 , cpExpect
19 , cpExpect
@@ -18,8 +29,9 @@ module Control.Distributed.Process.Internal.Closure.CP
18import Data.Binary (encode)
29import Data.Binary (encode)
19import Data.ByteString.Lazy (ByteString)
30import Data.ByteString.Lazy (ByteString)
20import Data.Typeable (Typeable, typeOf, typeRepTyCon, TyCon)
31import Data.Typeable (Typeable, typeOf, typeRepTyCon, TyCon)
32+import Data.Typeable.Internal (TypeRep(..))
21import Control.Applicative ((<$>))
33import Control.Applicative ((<$>))
22−import Control.Monad (join)
34+import Control.Monad ((>=>))
23import Control.Distributed.Process.Serializable (Serializable)
35import Control.Distributed.Process.Serializable (Serializable)
24import Control.Distributed.Process.Internal.Types
36import Control.Distributed.Process.Internal.Types
25 ( Closure(Closure)
37 ( Closure(Closure)
@@ -53,14 +65,17 @@ import Control.Distributed.Process.Internal.Closure.Static
53 , staticSplit
65 , staticSplit
54 , staticConst
66 , staticConst
55 , staticUnit
67 , staticUnit
68+ , staticFlip
69+ , staticFst
70+ , staticSnd
56 , sdictProcessId
71 , sdictProcessId
57 )
72 )
58import Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure)
73import Control.Distributed.Process.Internal.Closure.MkClosure (mkClosure)
59import Control.Distributed.Process.Internal.Dynamic
74import Control.Distributed.Process.Internal.Dynamic
60 ( Dynamic(Dynamic)
75 ( Dynamic(Dynamic)
61− , dynBind
62 , unsafeCoerce#
76 , unsafeCoerce#
63 , dynTypeRep
77 , dynTypeRep
78+ , dynKleisli
64 )
79 )
65
80
66--------------------------------------------------------------------------------
81--------------------------------------------------------------------------------
@@ -75,8 +90,11 @@ returnProcess = return
75bindProcess :: Process a -> (a -> Process b) -> Process b
90bindProcess :: Process a -> (a -> Process b) -> Process b
76bindProcess = (>>=)
91bindProcess = (>>=)
77
92
78−joinProcess :: Process (Process a) -> Process a
93+kleisliProcess :: (a -> Process b) -> (b -> Process c) -> a -> Process c
79−joinProcess = join
94+kleisliProcess = (>=>)
95+
96+parJoinProcess :: Process (a -> Process b) -> a -> Process b
97+parJoinProcess proc a = proc >>= ($ a)
80
98
81---- Variations on standard or CH functions with an explicit dictionary arg ----
99---- Variations on standard or CH functions with an explicit dictionary arg ----
82
100
@@ -92,8 +110,8 @@ newChanDict SerializableDict = newChan
92---- Serialization dictionaries ------------------------------------------------
110---- Serialization dictionaries ------------------------------------------------
93
111
94-- | Specialized serialization dictionary required in 'cpBind'
112-- | Specialized serialization dictionary required in 'cpBind'
95−sdictBind :: SerializableDict (ByteString, ByteString)
113+sdictComp :: SerializableDict (ByteString, ByteString)
96−sdictBind = SerializableDict
114+sdictComp = SerializableDict
97
115
98---- Some specialised processes necessary to implement the combinators ---------
116---- Some specialised processes necessary to implement the combinators ---------
99
117
@@ -105,42 +123,54 @@ unClosure (Static label) env = do
105 Nothing -> fail "Derived.unClosure: resolveClosure failed"
123 Nothing -> fail "Derived.unClosure: resolveClosure failed"
106 Just dyn -> return dyn
124 Just dyn -> return dyn
107
125
126+-- | Work around a bug in Typeable
127+-- (http://hackage.haskell.org/trac/ghc/ticket/5692)
128+compareWithoutFingerprint :: TypeRep -> TypeRep -> Bool
129+compareWithoutFingerprint (TypeRep _ con ts) (TypeRep _ con' ts')
130+ = con == con' && all (uncurry compareWithoutFingerprint) (zip ts ts')
131+
108-- | Remove a 'Dynamic' constructor, provided that the recorded type matches the
132-- | Remove a 'Dynamic' constructor, provided that the recorded type matches the
109-- type of the first static argument (the value of that argument is not used)
133-- type of the first static argument (the value of that argument is not used)
110unDynamic :: Static a -> Process Dynamic -> Process a
134unDynamic :: Static a -> Process Dynamic -> Process a
111unDynamic (Static label) pdyn = do
135unDynamic (Static label) pdyn = do
112 Dynamic typ val <- pdyn
136 Dynamic typ val <- pdyn
113− if typ == typeOfStaticLabel label
137+ if compareWithoutFingerprint typ (typeOfStaticLabel label) -- typ == typeOfStaticLabel label
114 then return (unsafeCoerce# val)
138 then return (unsafeCoerce# val)
115 else fail $ "unDynamic: cannot match "
139 else fail $ "unDynamic: cannot match "
116 ++ show typ
140 ++ show typ
117 ++ " against expected type "
141 ++ " against expected type "
118 ++ show (typeOfStaticLabel label)
142 ++ show (typeOfStaticLabel label)
119
143
120−-- | Dynamic bind
144+-- | Dynamic kleisli composition
121--
145--
122-- The first argument stops remotable from trying to generate a SerializableDict
146-- The first argument stops remotable from trying to generate a SerializableDict
123-- for (Process Dynamic, Process Dynamic)
147-- for (Process Dynamic, Process Dynamic)
124−bindDyn :: () -> (Process Dynamic, Process Dynamic) -> Process Dynamic
148+kleisliCompDyn :: () -> (Process Dynamic, Process Dynamic) -> Process Dynamic
125−bindDyn () (px, pf) = do
149+kleisliCompDyn () (pf, pg) = do
126− x <- px
150+ f <- pf -- a -> Process b
127− f <- pf
151+ g <- pg -- b -> Process c
128− case dynBind tyConProcess bindProcess x f of
152+ case dynKleisli tyConProcess kleisliProcess f g of
129 Just dyn -> return dyn
153 Just dyn -> return dyn
130− Nothing -> fail $ "bindDyn: could not bind "
154+ Nothing -> fail $ "kleisliCompDyn: could not compose "
131− ++ show (dynTypeRep x)
132− ++ " to "
133 ++ show (dynTypeRep f)
155 ++ show (dynTypeRep f)
156+ ++ " with "
157+ ++ show (dynTypeRep g)
134 where
158 where
135 tyConProcess :: TyCon
159 tyConProcess :: TyCon
136 tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))
160 tyConProcess = typeRepTyCon (typeOf (undefined :: Process ()))
137
161
162+cpFirstAux :: (a -> Process b) -> (a, c) -> Process (b, c)
163+cpFirstAux f (a, c) = f a >>= \b -> return (b, c)
164+
165+cpSecondAux :: (a -> Process b) -> (c, a) -> Process (c, b)
166+cpSecondAux f (c, a) = f a >>= \b -> return (c, b)
167+
138---- Finally, the call to remotable --------------------------------------------
168---- Finally, the call to remotable --------------------------------------------
139
169
140remotable [ -- Monadic operations
170remotable [ -- Monadic operations
141 'returnProcess
171 'returnProcess
142 , 'bindProcess
172 , 'bindProcess
143− , 'joinProcess
173+ , 'parJoinProcess
144 -- CH primitives
174 -- CH primitives
145 , 'link
175 , 'link
146 , 'unlink
176 , 'unlink
@@ -149,13 +179,101 @@ remotable [ -- Monadic operations
149 , 'expectDict
179 , 'expectDict
150 , 'newChanDict
180 , 'newChanDict
151 -- Serialization dictionaries
181 -- Serialization dictionaries
152− , 'sdictBind
182+ , 'sdictComp
153 -- Specialized processes
183 -- Specialized processes
154 , 'unClosure
184 , 'unClosure
155 , 'unDynamic
185 , 'unDynamic
156− , 'bindDyn
186+ , 'kleisliCompDyn
187+ , 'cpFirstAux
188+ , 'cpSecondAux
157 ]
189 ]
158
190
191+--------------------------------------------------------------------------------
192+-- Some derived static functions --
193+--------------------------------------------------------------------------------
194+
195+staticUndynamic :: Typeable a => Static (Process Dynamic -> Process a)
196+staticUndynamic =
197+ $(mkStatic 'unDynamic) `staticApply` staticDuplicate (staticTypeOf (undefined :: a))
198+
199+staticKleisliCompDyn :: Static ((Process Dynamic, Process Dynamic) -> Process Dynamic)
200+staticKleisliCompDyn = $(mkStatic 'kleisliCompDyn) `staticApply` staticUnit
201+
202+--------------------------------------------------------------------------------
203+-- Definition of CP and the generalized arrow combinators --
204+--------------------------------------------------------------------------------
205+
206+-- | 'CP a b' represents the closure of a process parameterized by 'a' and
207+-- returning 'b'. 'CP a b' forms a (restricted) generalized arrow
208+-- (<http://www.cs.berkeley.edu/~megacz/garrows/>)
209+type CP a b = Closure (a -> Process b)
210+
211+-- | 'CP' introduction form
212+cpIntro :: forall a b. (Typeable a, Typeable b)
213+ => Closure (Process b) -> Closure (a -> Process b)
214+cpIntro (Closure static env) = Closure decoder env
215+ where
216+ decoder :: Static (ByteString -> a -> Process b)
217+ decoder = staticConst `staticCompose` static
218+
219+-- | 'CP' elimination form
220+cpElim :: forall a. Typeable a => CP () a -> Closure (Process a)
221+cpElim (Closure static env) = Closure decoder env
222+ where
223+ decoder :: Static (ByteString -> Process a)
224+ decoder = staticFlip static `staticApply` staticUnit
225+
226+-- | Identity ('Closure' version of 'return')
227+cpId :: Typeable a => CP a a
228+cpId = staticClosure $(mkStatic 'returnProcess)
229+
230+-- | Left-to-right composition ('Closure' version of '>=>')
231+cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)
232+ => CP a b -> CP b c -> CP a c
233+cpComp (Closure fstatic fenv) (Closure gstatic genv) =
234+ Closure decoder (encode (fenv, genv))
235+ where
236+ decoder :: Static (ByteString -> a -> Process c)
237+ decoder = $(mkStatic 'parJoinProcess)
238+ `staticCompose`
239+ staticUndynamic
240+ `staticCompose`
241+ staticKleisliCompDyn
242+ `staticCompose`
243+ (staticUnclosure fstatic `staticSplit` staticUnclosure gstatic)
244+ `staticCompose`
245+ staticDecode $(mkStatic 'sdictComp)
246+
247+-- | First
248+cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
249+ => CP a b -> CP (a, c) (b, c)
250+cpFirst (Closure static env) = Closure decoder env
251+ where
252+ decoder :: Static (ByteString -> (a, c) -> Process (b, c))
253+ decoder = $(mkStatic 'cpFirstAux) `staticCompose` static
254+
255+-- | Second
256+cpSecond :: forall a b c. (Typeable a, Typeable b, Typeable c)
257+ => CP a b -> CP (c, a) (c, b)
258+cpSecond (Closure static env) = Closure decoder env
259+ where
260+ decoder :: Static (ByteString -> (c, a) -> Process (c, b))
261+ decoder = $(mkStatic 'cpSecondAux) `staticCompose` static
262+
263+-- | Split (Like 'Control.Arrow.***')
264+cpSplit :: (Typeable a, Typeable b, Typeable c, Typeable d)
265+ => CP a c -> CP b d -> CP (a, b) (c, d)
266+cpSplit f g = cpFirst f `cpComp` cpSecond g
267+
268+-- | Left cancellation
269+cpCancelL :: Typeable a => CP ((), a) a
270+-- Closure (((), a) -> Process a)
271+cpCancelL = staticClosure ($(mkStatic 'returnProcess) `staticCompose` staticSnd)
272+
273+-- | Right cancellation
274+cpCancelR :: Typeable a => CP (a, ()) a
275+cpCancelR = staticClosure ($(mkStatic 'returnProcess) `staticCompose` staticFst)
276+
159--------------------------------------------------------------------------------
277--------------------------------------------------------------------------------
160-- Closure versions of CH primitives --
278-- Closure versions of CH primitives --
161--------------------------------------------------------------------------------
279--------------------------------------------------------------------------------
@@ -208,37 +326,11 @@ staticUnclosure :: Typeable a
208staticUnclosure s =
326staticUnclosure s =
209 $(mkStatic 'unClosure) `staticApply` staticDuplicate s
327 $(mkStatic 'unClosure) `staticApply` staticDuplicate s
210
328
211−staticUndynamic :: Typeable a => Static (Process Dynamic -> Process a)
212−staticUndynamic =
213− $(mkStatic 'unDynamic) `staticApply` staticDuplicate (staticTypeOf (undefined :: a))
214−
215−staticBindDyn :: Static ((Process Dynamic, Process Dynamic) -> Process Dynamic)
216−staticBindDyn = $(mkStatic 'bindDyn) `staticApply` staticUnit
217−
218-- | Not-quite-monadic bind ('>>=')
329-- | Not-quite-monadic bind ('>>=')
219cpBind :: forall a b. (Typeable a, Typeable b)
330cpBind :: forall a b. (Typeable a, Typeable b)
220 => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
331 => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
221−cpBind (Closure xstatic xenv) (Closure fstatic fenv) =
332+cpBind x f = cpElim (cpIntro x `cpComp` f)
222− Closure decoder (encode (xenv, fenv))
223− where
224− decoder :: Static (ByteString -> Process b)
225− decoder = $(mkStatic 'joinProcess)
226− `staticCompose`
227− staticUndynamic
228− `staticCompose`
229− staticBindDyn
230− `staticCompose`
231− (staticUnclosure xstatic `staticSplit` staticUnclosure fstatic)
232− `staticCompose`
233− staticDecode $(mkStatic 'sdictBind)
234
333
235−cpIntro :: forall a b. (Typeable a, Typeable b)
236− => Closure (Process b) -> Closure (a -> Process b)
237−cpIntro (Closure static env) = Closure decoder env
238− where
239− decoder :: Static (ByteString -> a -> Process b)
240− decoder = staticConst `staticCompose` static
241−
242-- | Monadic sequencing ('>>')
334-- | Monadic sequencing ('>>')
243cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
335cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
244cpSeq p q = p `cpBind` cpIntro q
336cpSeq p q = p `cpBind` cpIntro q
distributed-process/src/Control/Distributed/Process/Internal/Closure/Static.hs+32
@@ -3,6 +3,9 @@
3module Control.Distributed.Process.Internal.Closure.Static
3module Control.Distributed.Process.Internal.Closure.Static
4 ( -- * Static functionals
4 ( -- * Static functionals
5 staticConst
5 staticConst
6+ , staticFlip
7+ , staticFst
8+ , staticSnd
6 , staticCompose
9 , staticCompose
7 , staticFirst
10 , staticFirst
8 , staticSecond
11 , staticSecond
@@ -16,6 +19,7 @@ module Control.Distributed.Process.Internal.Closure.Static
16 -- * Serialization dictionaries (and their static versions)
19 -- * Serialization dictionaries (and their static versions)
17 , sdictUnit
20 , sdictUnit
18 , sdictProcessId
21 , sdictProcessId
22+ , sdictSendPort
19 -- * Runtime support
23 -- * Runtime support
20 , __remoteTable
24 , __remoteTable
21 ) where
25 ) where
@@ -30,6 +34,7 @@ import Control.Distributed.Process.Internal.Types
30 , Static
34 , Static
31 , staticApply
35 , staticApply
32 , ProcessId
36 , ProcessId
37+ , SendPort
33 )
38 )
34import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)
39import Control.Distributed.Process.Internal.Closure.TH (remotable, mkStatic)
35import qualified Control.Arrow as Arrow (first, second, (***))
40import qualified Control.Arrow as Arrow (first, second, (***))
@@ -70,10 +75,16 @@ sdictUnit_ = SerializableDict
70sdictProcessId_ :: SerializableDict ProcessId
75sdictProcessId_ :: SerializableDict ProcessId
71sdictProcessId_ = SerializableDict
76sdictProcessId_ = SerializableDict
72
77
78+sdictSendPort_ :: SerializableDict a -> SerializableDict (SendPort a)
79+sdictSendPort_ SerializableDict = SerializableDict
80+
73---- Finally, the call to remotable --------------------------------------------
81---- Finally, the call to remotable --------------------------------------------
74
82
75remotable [ -- Functionals (predefined)
83remotable [ -- Functionals (predefined)
76 'const
84 'const
85+ , 'flip
86+ , 'fst
87+ , 'snd
77 -- Functionals (defined above)
88 -- Functionals (defined above)
78 , 'compose
89 , 'compose
79 , 'first
90 , 'first
@@ -86,6 +97,7 @@ remotable [ -- Functionals (predefined)
86 -- Serialization dictionaries
97 -- Serialization dictionaries
87 , 'sdictUnit_
98 , 'sdictUnit_
88 , 'sdictProcessId_
99 , 'sdictProcessId_
100+ , 'sdictSendPort_
89 ]
101 ]
90
102
91--------------------------------------------------------------------------------
103--------------------------------------------------------------------------------
@@ -97,6 +109,21 @@ remotable [ -- Functionals (predefined)
97staticConst :: (Typeable a, Typeable b) => Static (a -> b -> a)
109staticConst :: (Typeable a, Typeable b) => Static (a -> b -> a)
98staticConst = $(mkStatic 'const)
110staticConst = $(mkStatic 'const)
99
111
112+-- | Static version of 'flip'
113+staticFlip :: (Typeable a, Typeable b, Typeable c)
114+ => Static (a -> b -> c) -> Static (b -> a -> c)
115+staticFlip f = $(mkStatic 'flip) `staticApply` f
116+
117+-- | Static version of 'fst'
118+staticFst :: (Typeable a, Typeable b)
119+ => Static ((a, b) -> a)
120+staticFst = $(mkStatic 'fst)
121+
122+-- | Static version of 'snd'
123+staticSnd :: (Typeable a, Typeable b)
124+ => Static ((a, b) -> b)
125+staticSnd = $(mkStatic 'snd)
126+
100-- | Static version of ('Prelude..')
127-- | Static version of ('Prelude..')
101staticCompose :: (Typeable a, Typeable b, Typeable c)
128staticCompose :: (Typeable a, Typeable b, Typeable c)
102 => Static (b -> c) -> Static (a -> b) -> Static (a -> c)
129 => Static (b -> c) -> Static (a -> b) -> Static (a -> c)
@@ -137,6 +164,11 @@ sdictUnit = $(mkStatic 'sdictUnit_)
137sdictProcessId :: Static (SerializableDict ProcessId)
164sdictProcessId :: Static (SerializableDict ProcessId)
138sdictProcessId = $(mkStatic 'sdictProcessId_)
165sdictProcessId = $(mkStatic 'sdictProcessId_)
139
166
167+-- | Serialization dictionary for 'SendPort'
168+sdictSendPort :: Typeable a
169+ => Static (SerializableDict a) -> Static (SerializableDict (SendPort a))
170+sdictSendPort = staticApply $(mkStatic 'sdictSendPort_)
171+
140--------------------------------------------------------------------------------
172--------------------------------------------------------------------------------
141-- Creating closures --
173-- Creating closures --
142--------------------------------------------------------------------------------
174--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Internal/Dynamic.hs+31
@@ -15,6 +15,7 @@ module Control.Distributed.Process.Internal.Dynamic
15 , dynApp
15 , dynApp
16 , dynBind
16 , dynBind
17 , dynBind'
17 , dynBind'
18+ , dynKleisli
18 , GHC.unsafeCoerce#
19 , GHC.unsafeCoerce#
19 ) where
20 ) where
20
21
@@ -25,7 +26,9 @@ import Data.Typeable
25 , funResultTy
26 , funResultTy
26 , TyCon
27 , TyCon
27 , splitTyConApp
28 , splitTyConApp
29+ , mkFunTy
28 )
30 )
31+import Data.Typeable.Internal (funTc)
29import qualified GHC.Prim as GHC (Any, unsafeCoerce#)
32import qualified GHC.Prim as GHC (Any, unsafeCoerce#)
30import Data.Maybe (fromMaybe)
33import Data.Maybe (fromMaybe)
31
34
@@ -79,3 +82,31 @@ dynBind' :: TyCon -> (forall a b. m a -> (a -> m b) -> m b) -> Dynamic -> Dynami
79dynBind' m bind x f = fromMaybe (error typeError) (dynBind m bind x f)
82dynBind' m bind x f = fromMaybe (error typeError) (dynBind m bind x f)
80 where
83 where
81 typeError = "Type error in dynamic bind.\nCan't bind " ++ show x ++ " to " ++ show f
84 typeError = "Type error in dynamic bind.\nCan't bind " ++ show x ++ " to " ++ show f
85+
86+
87+
88+
89+-- | Dynamically typed Kleisli composition
90+dynKleisli :: TyCon
91+ -> (forall a b c. (a -> m b) -> (b -> m c) -> a -> m c)
92+ -> Dynamic
93+ -> Dynamic
94+ -> Maybe Dynamic
95+dynKleisli m comp (Dynamic tf f) (Dynamic tg g) =
96+ case splitTyConApp tf of
97+ (arr1, [a, mb]) | arr1 == funTc ->
98+ case splitTyConApp tg of
99+ (arr2, [b, mc]) | arr2 == funTc ->
100+ case splitTyConApp mb of
101+ (m', [b']) | m' == m && b' == b ->
102+ case splitTyConApp mc of
103+ (m'', [_c]) | m'' == m ->
104+ Just (Dynamic (mkFunTy a mc) (GHC.unsafeCoerce# comp f g))
105+ _ ->
106+ Nothing
107+ _ ->
108+ Nothing
109+ _ ->
110+ Nothing
111+ _ ->
112+ Nothing
distributed-process/tests/TestClosure.hs+29−1
@@ -2,7 +2,7 @@ module Main where
2
2
3import Data.ByteString.Lazy (empty)
3import Data.ByteString.Lazy (empty)
4import Data.Typeable (Typeable, typeOf)
4import Data.Typeable (Typeable, typeOf)
5−import Control.Monad (join, replicateM)
5+import Control.Monad (join, replicateM, forever)
6import Control.Exception (IOException, throw)
6import Control.Exception (IOException, throw)
7import Control.Concurrent (forkIO, threadDelay)
7import Control.Concurrent (forkIO, threadDelay)
8import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)
8import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)
@@ -36,12 +36,22 @@ sendPid toPid = do
36wait :: Int -> Process ()
36wait :: Int -> Process ()
37wait = liftIO . threadDelay
37wait = liftIO . threadDelay
38
38
39+-- | First argument indicates empty closure environment
40+typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()
41+typedPingServer () rport = forever $ do
42+ us <- getSelfPid
43+ liftIO . putStrLn $ "Ping server " ++ show us ++ " waiting"
44+ sport <- receiveChan rport
45+ liftIO . putStrLn $ "Ping server " ++ show us ++ " replying on " ++ show (sendPortId sport)
46+ sendChan sport ()
47+
39remotable [ 'factorial
48remotable [ 'factorial
40 , 'addInt
49 , 'addInt
41 , 'putInt
50 , 'putInt
42 , 'sendPid
51 , 'sendPid
43 , 'sdictInt
52 , 'sdictInt
44 , 'wait
53 , 'wait
54+ , 'typedPingServer
45 ]
55 ]
46
56
47factorialClosure :: Int -> Closure (Process Int)
57factorialClosure :: Int -> Closure (Process Int)
@@ -276,6 +286,23 @@ testClosureExpect transport rtable = do
276 (1234 :: Int) <- expect
286 (1234 :: Int) <- expect
277 return ()
287 return ()
278
288
289+testSpawnChannel :: Transport -> RemoteTable -> IO ()
290+testSpawnChannel transport rtable = do
291+ [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
292+
293+ runProcess node1 $ do
294+ pingServer <- spawnChannel
295+ (sdictSendPort sdictUnit)
296+ (localNodeId node2)
297+ ($(mkClosure 'typedPingServer) ())
298+ (sendReply, receiveReply) <- newChan
299+ sendChan pingServer sendReply
300+ sendChan pingServer sendReply
301+ liftIO . putStrLn $ "client waiting for ping server to reply on " ++ show (sendPortId sendReply)
302+ () <- receiveChan receiveReply
303+ liftIO . putStrLn $ "client got reply"
304+ return ()
305+
279main :: IO ()
306main :: IO ()
280main = do
307main = do
281 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
308 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
@@ -293,4 +320,5 @@ main = do
293 , ("SpawnSupervised", testSpawnSupervised transport rtable)
320 , ("SpawnSupervised", testSpawnSupervised transport rtable)
294 , ("SpawnInvalid", testSpawnInvalid transport rtable)
321 , ("SpawnInvalid", testSpawnInvalid transport rtable)
295 , ("ClosureExpect", testClosureExpect transport rtable)
322 , ("ClosureExpect", testClosureExpect transport rtable)
323+ , ("SpawnChannel", testSpawnChannel transport rtable)
296 ]
324 ]