git.iko.soy/distributed-process
on master help sign in
Continue cleanup, and add spawnSupervised
7f7e274Edsko de Vries
distributed-process/src/Control/Distributed/Process.hs+30−16
compare with master
unifiedsplit
@@ -58,6 +58,7 @@ module Control.Distributed.Process
58 , catch
58 , catch
59 , expectTimeout
59 , expectTimeout
60 , spawnAsync
60 , spawnAsync
61+ , spawnSupervised
61 ) where
62 ) where
62
63
63import Prelude hiding (catch)
64import Prelude hiding (catch)
@@ -65,7 +66,6 @@ import Data.Binary (decode, encode)
65import Data.Typeable (Typeable, typeOf)
66import Data.Typeable (Typeable, typeOf)
66import Control.Monad.Reader (ask)
67import Control.Monad.Reader (ask)
67import Control.Monad.IO.Class (MonadIO, liftIO)
68import Control.Monad.IO.Class (MonadIO, liftIO)
68−import qualified Control.Monad.Trans.Class as Trans (lift)
69import Control.Applicative ((<$>))
69import Control.Applicative ((<$>))
70import Control.Exception (Exception, throw)
70import Control.Exception (Exception, throw)
71import qualified Control.Exception as Exception (catch)
71import qualified Control.Exception as Exception (catch)
@@ -107,7 +107,6 @@ import Control.Distributed.Process.Internal.Types
107 , ProcessSignal(..)
107 , ProcessSignal(..)
108 , monitorCounter
108 , monitorCounter
109 , spawnCounter
109 , spawnCounter
110− , MessageT
111 , Closure(..)
110 , Closure(..)
112 , CallReply(..)
111 , CallReply(..)
113 , SendPort(..)
112 , SendPort(..)
@@ -117,6 +116,7 @@ import Control.Distributed.Process.Internal.Types
117 , TypedChannel(..)
116 , TypedChannel(..)
118 , ChannelId(..)
117 , ChannelId(..)
119 , Identifier(..)
118 , Identifier(..)
119+ , procMsg
120 )
120 )
121import Control.Distributed.Process.Internal.MessageT
121import Control.Distributed.Process.Internal.MessageT
122 ( sendMessage
122 ( sendMessage
@@ -126,6 +126,8 @@ import Control.Distributed.Process.Internal.MessageT
126import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
126import Control.Distributed.Process.Internal.Dynamic (fromDyn, dynTypeRep)
127import Control.Distributed.Process.Internal.Closure (resolveClosure)
127import Control.Distributed.Process.Internal.Closure (resolveClosure)
128import Control.Distributed.Process.Internal.Node (runLocalProcess)
128import Control.Distributed.Process.Internal.Node (runLocalProcess)
129+import {-# SOURCE #-} Control.Distributed.Process.Internal.Closure.BuiltIn (linkClosure)
130+import Control.Distributed.Process.Internal.Closure.Combinators (cpSeq)
129
131
130-- INTERNAL NOTES
132-- INTERNAL NOTES
131--
133--
@@ -194,7 +196,9 @@ import Control.Distributed.Process.Internal.Node (runLocalProcess)
194send :: Serializable a => ProcessId -> a -> Process ()
196send :: Serializable a => ProcessId -> a -> Process ()
195-- This requires a lookup on every send. If we want to avoid that we need to
197-- This requires a lookup on every send. If we want to avoid that we need to
196-- modify serializable to allow for stateful (IO) deserialization
198-- modify serializable to allow for stateful (IO) deserialization
199+-- Warning: if we change how 'send' is implemented, might also want to change
200+-- the implementation of 'Internal.Closure.TH.generateSender'
197−send them msg = lift $ sendMessage (ProcessIdentifier them) msg
201+send them msg = procMsg $ sendMessage (ProcessIdentifier them) msg
198
202
199-- | Wait for a message of a specific type
203-- | Wait for a message of a specific type
200expect :: forall a. Serializable a => Process a
204expect :: forall a. Serializable a => Process a
@@ -225,7 +229,7 @@ newChan = do
225
229
226-- | Send a message on a typed channel
230-- | Send a message on a typed channel
227sendChan :: Serializable a => SendPort a -> a -> Process ()
231sendChan :: Serializable a => SendPort a -> a -> Process ()
228−sendChan (SendPort them) msg = lift $ sendBinary (ChannelIdentifier them) msg
232+sendChan (SendPort them) msg = procMsg $ sendBinary (ChannelIdentifier them) msg
229
233
230-- | Wait for a message on a typed channel
234-- | Wait for a message on a typed channel
231receiveChan :: Serializable a => ReceivePort a -> Process a
235receiveChan :: Serializable a => ReceivePort a -> Process a
@@ -337,28 +341,30 @@ getSelfPid = processId <$> ask
337
341
338-- | Get the node ID of our local node
342-- | Get the node ID of our local node
339getSelfNode :: Process NodeId
343getSelfNode :: Process NodeId
340−getSelfNode = localNodeId <$> lift getLocalNode
344+getSelfNode = localNodeId <$> procMsg getLocalNode
341
345
342--------------------------------------------------------------------------------
346--------------------------------------------------------------------------------
343-- Monitoring and linking --
347-- Monitoring and linking --
344--------------------------------------------------------------------------------
348--------------------------------------------------------------------------------
345
349
346−-- | Link to a remote process
350+-- | Link to a remote process (asynchronous)
351+--
352+-- Note that 'link' provides unidirectional linking (see 'spawnSupervised').
347link :: ProcessId -> Process ()
353link :: ProcessId -> Process ()
348link = sendCtrlMsg Nothing . Link
354link = sendCtrlMsg Nothing . Link
349
355
350−-- | Remove a link
356+-- | Remove a link (asynchronous)
351unlink :: ProcessId -> Process ()
357unlink :: ProcessId -> Process ()
352unlink = sendCtrlMsg Nothing . Unlink
358unlink = sendCtrlMsg Nothing . Unlink
353
359
354−-- | Monitor another process
360+-- | Monitor another process (asynchronous)
355monitor :: ProcessId -> Process MonitorRef
361monitor :: ProcessId -> Process MonitorRef
356monitor them = do
362monitor them = do
357 monitorRef <- getMonitorRefFor them
363 monitorRef <- getMonitorRefFor them
358 sendCtrlMsg Nothing $ Monitor them monitorRef
364 sendCtrlMsg Nothing $ Monitor them monitorRef
359 return monitorRef
365 return monitorRef
360
366
361−-- | Remove a monitor
367+-- | Remove a monitor (asynchronous)
362unmonitor :: MonitorRef -> Process ()
368unmonitor :: MonitorRef -> Process ()
363unmonitor = sendCtrlMsg Nothing . Unmonitor
369unmonitor = sendCtrlMsg Nothing . Unmonitor
364
370
@@ -369,7 +375,7 @@ unmonitor = sendCtrlMsg Nothing . Unmonitor
369-- | Deserialize a closure
375-- | Deserialize a closure
370unClosure :: forall a. Typeable a => Closure a -> Process a
376unClosure :: forall a. Typeable a => Closure a -> Process a
371unClosure (Closure (Static label) env) = do
377unClosure (Closure (Static label) env) = do
372− rtable <- remoteTable <$> lift getLocalNode
378+ rtable <- remoteTable <$> procMsg getLocalNode
373 let Just dyn = resolveClosure rtable label env
379 let Just dyn = resolveClosure rtable label env
374 return $ fromDyn dyn (throw (typeError dyn))
380 return $ fromDyn dyn (throw (typeError dyn))
375 where
381 where
@@ -384,7 +390,7 @@ unClosure (Closure (Static label) env) = do
384-- | Catch exceptions within a process
390-- | Catch exceptions within a process
385catch :: Exception e => Process a -> (e -> Process a) -> Process a
391catch :: Exception e => Process a -> (e -> Process a) -> Process a
386catch p h = do
392catch p h = do
387− node <- lift getLocalNode
393+ node <- procMsg getLocalNode
388 lproc <- ask
394 lproc <- ask
389 let run :: Process a -> IO a
395 let run :: Process a -> IO a
390 run proc = runLocalProcess node proc lproc
396 run proc = runLocalProcess node proc lproc
@@ -403,6 +409,17 @@ spawnAsync nid proc = do
403 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
409 sendCtrlMsg (Just nid) $ Spawn proc spawnRef
404 return spawnRef
410 return spawnRef
405
411
412+-- | Spawn a child process, have the child link to the parent and the parent
413+-- monitor the child
414+spawnSupervised :: NodeId
415+ -> Closure (Process ())
416+ -> Process (ProcessId, MonitorRef)
417+spawnSupervised nid proc = do
418+ us <- getSelfPid
419+ them <- spawn nid (linkClosure us `cpSeq` proc)
420+ ref <- monitor them
421+ return (them, ref)
422+
406--------------------------------------------------------------------------------
423--------------------------------------------------------------------------------
407-- Auxiliary functions --
424-- Auxiliary functions --
408--------------------------------------------------------------------------------
425--------------------------------------------------------------------------------
@@ -436,10 +453,7 @@ sendCtrlMsg mNid signal = do
436 }
453 }
437 case mNid of
454 case mNid of
438 Nothing -> do
455 Nothing -> do
439− ctrlChan <- localCtrlChan <$> lift getLocalNode
456+ ctrlChan <- localCtrlChan <$> procMsg getLocalNode
440 liftIO $ writeChan ctrlChan msg
457 liftIO $ writeChan ctrlChan msg
441 Just nid ->
458 Just nid ->
442− lift $ sendBinary (NodeIdentifier nid) msg
459+ procMsg $ sendBinary (NodeIdentifier nid) msg
443−
444−lift :: MessageT IO a -> Process a
445−lift = Process . Trans.lift
distributed-process/src/Control/Distributed/Process/Closure.hs+53−156
@@ -1,13 +1,31 @@
1-- | Implementation of 'Closure' that works around the absence of 'static'.
1-- | Implementation of 'Closure' that works around the absence of 'static'.
2--
2--
3−-- [Example usage]
3+-- [Built-in closures]
4+--
5+-- We offer a number of standard commonly useful closures.
6+--
7+-- [Closure combinators]
8+--
9+-- Closures combinators allow to create closures from other closures. For
10+-- example, 'spawnSupervised' is defined as follows:
11+--
12+-- > spawnSupervised :: NodeId
13+-- > -> Closure (Process ())
14+-- > -> Process (ProcessId, MonitorRef)
15+-- > spawnSupervised nid proc = do
16+-- > us <- getSelfPid
17+-- > them <- spawn nid (linkClosure us `cpSeq` proc)
18+-- > ref <- monitor them
19+-- > return (them, ref)
20+--
21+-- [User-defined closures]
4--
22--
5-- Suppose we have a monomorphic function
23-- Suppose we have a monomorphic function
6--
24--
7-- > addInt :: Int -> Int -> Int
25-- > addInt :: Int -> Int -> Int
8-- > addInt x y = x + y
26-- > addInt x y = x + y
9--
27--
10−-- Then
28+-- Then the Template Haskell splice
11--
29--
12-- > remotable ['addInt]
30-- > remotable ['addInt]
13--
31--
@@ -20,7 +38,7 @@
20--
38--
21-- > unClosure :: Typeable a => Closure a -> Process a
39-- > unClosure :: Typeable a => Closure a -> Process a
22--
40--
23−-- In general, given a monomorphic function @f : a -> b@ the corresponding
41+-- In general, given a monomorphic function @f :: a -> b@ the corresponding
24-- function @$(mkClosure 'f)@ will have type @a -> Closure b@.
42-- function @$(mkClosure 'f)@ will have type @a -> Closure b@.
25--
43--
26-- The call to 'remotable' will also generate a function
44-- The call to 'remotable' will also generate a function
@@ -39,9 +57,11 @@
39--
57--
40-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.
58-- See Section 6, /Faking It/, of /Towards Haskell in the Cloud/ for more info.
41module Control.Distributed.Process.Closure
59module Control.Distributed.Process.Closure
42− ( -- * Primitive closures
60+ ( -- * User-defined closures
43 remotable
61 remotable
44 , mkClosure
62 , mkClosure
63+ -- * Built-in closures
64+ , linkClosure
45 -- * Generic closure combinators
65 -- * Generic closure combinators
46 , closureApply
66 , closureApply
47 , closureConst
67 , closureConst
@@ -70,156 +90,33 @@ module Control.Distributed.Process.Closure
70 , cpSeq
90 , cpSeq
71 ) where
91 ) where
72
92
73−import Prelude hiding (lookup)
74−import qualified Data.ByteString.Lazy as BS (empty)
75−import Data.Binary (encode)
76−import Data.Typeable (typeOf, Typeable)
77−import Control.Distributed.Process.Internal.Types
78− ( Closure(..)
79− , Static(..)
80− , StaticLabel(..)
81− , Process
82− )
83import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
93import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
84−
94+import Control.Distributed.Process.Internal.Closure.BuiltIn (linkClosure)
85−--------------------------------------------------------------------------------
95+import Control.Distributed.Process.Internal.Closure.Combinators
86−-- Generic closure combinators --
96+ ( -- Generic combinators
87−--------------------------------------------------------------------------------
88−
89−closureApply :: Closure (a -> b) -> Closure a -> Closure b
97+ closureApply
90−closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =
98+ , closureConst
91− Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)
92−
93−closureConst :: forall a b. (Typeable a, Typeable b)
94− => Closure (a -> b -> a)
95−closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)
96− where
97− aux :: a -> b -> a
98− aux = undefined
99−
100−closureUnit :: Closure ()
101−closureUnit = Closure (Static ClosureUnit) BS.empty
99+ , closureUnit
102−
103−--------------------------------------------------------------------------------
104−-- Arrow combinators for processes --
100+ -- Arrow combinators for processes
105−--------------------------------------------------------------------------------
101+ , CP
106−
102+ , cpIntro
107−type CP a b = Closure (a -> Process b)
103+ , cpElim
108−
104+ , cpId
109−cpIntro :: (Typeable a, Typeable b)
105+ , cpComp
110− => Closure (Process b) -> CP a b
106+ , cpFirst
111−cpIntro = closureApply closureConst
107+ , cpSwap
112−
108+ , cpSecond
113−cpElim :: Typeable a
109+ , cpPair
114− => CP () a -> Closure (Process a)
110+ , cpCopy
115−cpElim = flip closureApply closureUnit
111+ , cpFanOut
116−
112+ , cpLeft
117−cpId :: forall a. Typeable a
113+ , cpMirror
118− => CP a a
114+ , cpRight
119−cpId = Closure (Static CpId) (encode $ typeOf aux)
115+ , cpEither
120− where
116+ , cpUntag
121− aux :: a -> Process a
117+ , cpFanIn
122− aux = undefined
118+ , cpApply
123−
119+ -- Derived process operators
124−cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)
120+ , cpBind
125− => CP a b -> CP b c -> CP a c
121+ , cpSeq
126−cpComp f g = comp `closureApply` f `closureApply` g
122+ )
127− where
128− comp :: Closure ((a -> Process b) -> (b -> Process c) -> a -> Process c)
129− comp = Closure (Static CpComp) (encode $ typeOf aux)
130−
131− aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c
132− aux = undefined
133−
134−cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
135− => CP a b -> CP (a, c) (b, c)
136−cpFirst = closureApply first
137− where
138− first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))
139− first = Closure (Static CpFirst) (encode $ typeOf aux)
140−
141− aux :: (a -> Process b) -> (a, c) -> Process (b, c)
142− aux = undefined
143−
144−cpSwap :: forall a b. (Typeable a, Typeable b)
145− => CP (a, b) (b, a)
146−cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)
147− where
148− aux :: (a, b) -> Process (b, a)
149− aux = undefined
150−
151−cpSecond :: (Typeable a, Typeable b, Typeable c)
152− => CP a b -> CP (c, a) (c, b)
153−cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap
154−
155−cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')
156− => CP a b -> CP a' b' -> CP (a, a') (b, b')
157−cpPair f g = cpFirst f `cpComp` cpSecond g
158−
159−cpCopy :: forall a. Typeable a
160− => CP a (a, a)
161−cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)
162− where
163− aux :: a -> Process (a, a)
164− aux = undefined
165−
166−cpFanOut :: (Typeable a, Typeable b, Typeable c)
167− => CP a b -> CP a c -> CP a (b, c)
168−cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)
169−
170−cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)
171− => CP a b -> CP (Either a c) (Either b c)
172−cpLeft = closureApply left
173− where
174− left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c))
175− left = Closure (Static CpLeft) (encode $ typeOf aux)
176−
177− aux :: (a -> Process b) -> Either a c -> Process (Either b c)
178− aux = undefined
179−
180−cpMirror :: forall a b. (Typeable a, Typeable b)
181− => CP (Either a b) (Either b a)
182−cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)
183− where
184− aux :: Either a b -> Process (Either b a)
185− aux = undefined
186−
187−cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)
188− => CP a b -> CP (Either c a) (Either c b)
189−cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror
190−
191−cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')
192− => CP a b -> CP a' b' -> CP (Either a a') (Either b b')
193−cpEither f g = cpLeft f `cpComp` cpRight g
194−
195−cpUntag :: forall a. Typeable a
196− => CP (Either a a) a
197−cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)
198− where
199− aux :: Either a a -> Process a
200− aux = undefined
201−
202−cpFanIn :: (Typeable a, Typeable b, Typeable c)
203− => CP a c -> CP b c -> CP (Either a b) c
204−cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag
205−
206−cpApply :: forall a b. (Typeable a, Typeable b)
207− => CP (CP a b, a) b
208−cpApply = Closure (Static CpApply) $ encode ( typeOf aux
209− , typeOf (undefined :: a)
210− , typeOf (undefined :: Process b)
211− )
212− where
213− aux :: (Closure (a -> Process b), a) -> Process b
214− aux = undefined
215−
216−--------------------------------------------------------------------------------
217−-- Some derived operators for processes --
218−--------------------------------------------------------------------------------
219−
220−cpBind :: (Typeable a, Typeable b)
221− => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
222−cpBind x f = cpElim $ cpIntro x `cpComp` f
223−
224−cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
225−cpSeq p q = p `cpBind` cpIntro q
distributed-process/src/Control/Distributed/Process/Internal/Closure.hs+5−11
@@ -2,14 +2,12 @@
2module Control.Distributed.Process.Internal.Closure
2module Control.Distributed.Process.Internal.Closure
3 ( -- * Runtime support
3 ( -- * Runtime support
4 initRemoteTable
4 initRemoteTable
5− , registerLabel
6− , registerSender
7 , resolveClosure
5 , resolveClosure
8 ) where
6 ) where
9
7
10import qualified Data.Map as Map (empty)
8import qualified Data.Map as Map (empty)
11−import Data.Accessor ((^=), (^.))
9+import Data.Accessor ((^.))
12−import Data.Typeable (TypeRep, TyCon, typeRepTyCon, typeOf)
10+import Data.Typeable (TyCon, typeRepTyCon, typeOf)
13import Data.ByteString.Lazy (ByteString)
11import Data.ByteString.Lazy (ByteString)
14import Data.Binary (decode)
12import Data.Binary (decode)
15import Control.Distributed.Process.Internal.Types
13import Control.Distributed.Process.Internal.Types
@@ -32,6 +30,8 @@ import Control.Distributed.Process.Internal.Dynamic
32 , unsafeCoerce#
30 , unsafeCoerce#
33 )
31 )
34import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
32import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
33+import {-# SOURCE #-} qualified Control.Distributed.Process.Internal.Closure.BuiltIn as BuiltIn
34+ (remoteTable)
35
35
36--------------------------------------------------------------------------------
36--------------------------------------------------------------------------------
37-- Runtime support for closures --
37-- Runtime support for closures --
@@ -39,13 +39,7 @@ import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
39
39
40-- | Initial (empty) remote-call meta data
40-- | Initial (empty) remote-call meta data
41initRemoteTable :: RemoteTable
41initRemoteTable :: RemoteTable
42−initRemoteTable = RemoteTable Map.empty Map.empty
42+initRemoteTable = BuiltIn.remoteTable (RemoteTable Map.empty Map.empty)
43−
44−registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable
45−registerLabel label dyn = remoteTableLabel label ^= Just dyn
46−
47−registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable
48−registerSender typ dyn = remoteTableSender typ ^= Just dyn
49
43
50resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
44resolveClosure :: RemoteTable -> StaticLabel -> ByteString -> Maybe Dynamic
51-- Special support for call
45-- Special support for call
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hsnew file+23
@@ -0,0 +1,23 @@
1+{-# LANGUAGE TemplateHaskell #-}
2+module Control.Distributed.Process.Internal.Closure.BuiltIn
3+ ( remoteTable
4+ , linkClosure
5+ ) where
6+
7+import Control.Distributed.Process (link)
8+import Control.Distributed.Process.Internal.Types
9+ ( ProcessId
10+ , Closure
11+ , Process
12+ , RemoteTable
13+ )
14+import Control.Distributed.Process.Internal.Closure.TH (remotable, mkClosure)
15+
16+remotable ['link]
17+
18+remoteTable :: RemoteTable -> RemoteTable
19+remoteTable = __remoteTable
20+
21+-- | Closure version of 'link'
22+linkClosure :: ProcessId -> Closure (Process ())
23+linkClosure = $(mkClosure 'link)
distributed-process/src/Control/Distributed/Process/Internal/Closure/BuiltIn.hs-bootnew file+12
@@ -0,0 +1,12 @@
1+module Control.Distributed.Process.Internal.Closure.BuiltIn where
2+
3+import Control.Distributed.Process.Internal.Types
4+ ( ProcessId
5+ , Closure
6+ , Process
7+ , RemoteTable
8+ )
9+
10+remoteTable :: RemoteTable -> RemoteTable
11+
12+linkClosure :: ProcessId -> Closure (Process ())
distributed-process/src/Control/Distributed/Process/Internal/Closure/Combinators.hsnew file+182
@@ -0,0 +1,182 @@
1+module Control.Distributed.Process.Internal.Closure.Combinators
2+ ( -- * Generic combinators
3+ closureApply
4+ , closureConst
5+ , closureUnit
6+ -- * Arrow combinators for processes
7+ , CP
8+ , cpIntro
9+ , cpElim
10+ , cpId
11+ , cpComp
12+ , cpFirst
13+ , cpSwap
14+ , cpSecond
15+ , cpPair
16+ , cpCopy
17+ , cpFanOut
18+ , cpLeft
19+ , cpMirror
20+ , cpRight
21+ , cpEither
22+ , cpUntag
23+ , cpFanIn
24+ , cpApply
25+ -- * Derived process operators
26+ , cpBind
27+ , cpSeq
28+ ) where
29+
30+import Prelude hiding (lookup)
31+import qualified Data.ByteString.Lazy as BS (empty)
32+import Data.Binary (encode)
33+import Data.Typeable (typeOf, Typeable)
34+import Control.Distributed.Process.Internal.Types
35+ ( Closure(..)
36+ , Static(..)
37+ , StaticLabel(..)
38+ , Process
39+ )
40+import Control.Distributed.Process.Internal.TypeRep () -- Binary instances
41+
42+--------------------------------------------------------------------------------
43+-- Generic closure combinators --
44+--------------------------------------------------------------------------------
45+
46+closureApply :: Closure (a -> b) -> Closure a -> Closure b
47+closureApply (Closure (Static labelf) envf) (Closure (Static labelx) envx) =
48+ Closure (Static ClosureApply) $ encode (labelf, envf, labelx, envx)
49+
50+closureConst :: forall a b. (Typeable a, Typeable b)
51+ => Closure (a -> b -> a)
52+closureConst = Closure (Static ClosureConst) (encode $ typeOf aux)
53+ where
54+ aux :: a -> b -> a
55+ aux = undefined
56+
57+closureUnit :: Closure ()
58+closureUnit = Closure (Static ClosureUnit) BS.empty
59+
60+--------------------------------------------------------------------------------
61+-- Arrow combinators for processes --
62+--------------------------------------------------------------------------------
63+
64+type CP a b = Closure (a -> Process b)
65+
66+cpIntro :: (Typeable a, Typeable b)
67+ => Closure (Process b) -> CP a b
68+cpIntro = closureApply closureConst
69+
70+cpElim :: Typeable a
71+ => CP () a -> Closure (Process a)
72+cpElim = flip closureApply closureUnit
73+
74+cpId :: forall a. Typeable a
75+ => CP a a
76+cpId = Closure (Static CpId) (encode $ typeOf aux)
77+ where
78+ aux :: a -> Process a
79+ aux = undefined
80+
81+cpComp :: forall a b c. (Typeable a, Typeable b, Typeable c)
82+ => CP a b -> CP b c -> CP a c
83+cpComp f g = comp `closureApply` f `closureApply` g
84+ where
85+ comp :: Closure ((a -> Process b) -> (b -> Process c) -> a -> Process c)
86+ comp = Closure (Static CpComp) (encode $ typeOf aux)
87+
88+ aux :: (a -> Process b) -> (b -> Process c) -> a -> Process c
89+ aux = undefined
90+
91+cpFirst :: forall a b c. (Typeable a, Typeable b, Typeable c)
92+ => CP a b -> CP (a, c) (b, c)
93+cpFirst = closureApply first
94+ where
95+ first :: Closure ((a -> Process b) -> (a, c) -> Process (b, c))
96+ first = Closure (Static CpFirst) (encode $ typeOf aux)
97+
98+ aux :: (a -> Process b) -> (a, c) -> Process (b, c)
99+ aux = undefined
100+
101+cpSwap :: forall a b. (Typeable a, Typeable b)
102+ => CP (a, b) (b, a)
103+cpSwap = Closure (Static CpSwap) (encode $ typeOf aux)
104+ where
105+ aux :: (a, b) -> Process (b, a)
106+ aux = undefined
107+
108+cpSecond :: (Typeable a, Typeable b, Typeable c)
109+ => CP a b -> CP (c, a) (c, b)
110+cpSecond f = cpSwap `cpComp` cpFirst f `cpComp` cpSwap
111+
112+cpPair :: (Typeable a, Typeable a', Typeable b, Typeable b')
113+ => CP a b -> CP a' b' -> CP (a, a') (b, b')
114+cpPair f g = cpFirst f `cpComp` cpSecond g
115+
116+cpCopy :: forall a. Typeable a
117+ => CP a (a, a)
118+cpCopy = Closure (Static CpCopy) (encode $ typeOf aux)
119+ where
120+ aux :: a -> Process (a, a)
121+ aux = undefined
122+
123+cpFanOut :: (Typeable a, Typeable b, Typeable c)
124+ => CP a b -> CP a c -> CP a (b, c)
125+cpFanOut f g = cpCopy `cpComp` (f `cpPair` g)
126+
127+cpLeft :: forall a b c. (Typeable a, Typeable b, Typeable c)
128+ => CP a b -> CP (Either a c) (Either b c)
129+cpLeft = closureApply left
130+ where
131+ left :: Closure ((a -> Process b) -> Either a c -> Process (Either b c))
132+ left = Closure (Static CpLeft) (encode $ typeOf aux)
133+
134+ aux :: (a -> Process b) -> Either a c -> Process (Either b c)
135+ aux = undefined
136+
137+cpMirror :: forall a b. (Typeable a, Typeable b)
138+ => CP (Either a b) (Either b a)
139+cpMirror = Closure (Static CpMirror) (encode $ typeOf aux)
140+ where
141+ aux :: Either a b -> Process (Either b a)
142+ aux = undefined
143+
144+cpRight :: forall a b c. (Typeable a, Typeable b, Typeable c)
145+ => CP a b -> CP (Either c a) (Either c b)
146+cpRight f = cpMirror `cpComp` cpLeft f `cpComp` cpMirror
147+
148+cpEither :: (Typeable a, Typeable a', Typeable b, Typeable b')
149+ => CP a b -> CP a' b' -> CP (Either a a') (Either b b')
150+cpEither f g = cpLeft f `cpComp` cpRight g
151+
152+cpUntag :: forall a. Typeable a
153+ => CP (Either a a) a
154+cpUntag = Closure (Static CpUntag) (encode $ typeOf aux)
155+ where
156+ aux :: Either a a -> Process a
157+ aux = undefined
158+
159+cpFanIn :: (Typeable a, Typeable b, Typeable c)
160+ => CP a c -> CP b c -> CP (Either a b) c
161+cpFanIn f g = (f `cpEither` g) `cpComp` cpUntag
162+
163+cpApply :: forall a b. (Typeable a, Typeable b)
164+ => CP (CP a b, a) b
165+cpApply = Closure (Static CpApply) $ encode ( typeOf aux
166+ , typeOf (undefined :: a)
167+ , typeOf (undefined :: Process b)
168+ )
169+ where
170+ aux :: (Closure (a -> Process b), a) -> Process b
171+ aux = undefined
172+
173+--------------------------------------------------------------------------------
174+-- Some derived operators for processes --
175+--------------------------------------------------------------------------------
176+
177+cpBind :: (Typeable a, Typeable b)
178+ => Closure (Process a) -> Closure (a -> Process b) -> Closure (Process b)
179+cpBind x f = cpElim $ cpIntro x `cpComp` f
180+
181+cpSeq :: Closure (Process ()) -> Closure (Process ()) -> Closure (Process ())
182+cpSeq p q = p `cpBind` cpIntro q
distributed-process/src/Control/Distributed/Process/Internal/Closure/TH.hs+25−16
@@ -10,17 +10,9 @@ module Control.Distributed.Process.Internal.Closure.TH
10import Prelude hiding (lookup)
10import Prelude hiding (lookup)
11import Data.ByteString.Lazy (ByteString)
11import Data.ByteString.Lazy (ByteString)
12import Data.Binary (encode, decode)
12import Data.Binary (encode, decode)
13−import Data.Typeable (typeOf, Typeable)
13+import Data.Typeable (typeOf, Typeable, TypeRep)
14+import Data.Accessor ((^=))
14import Control.Applicative ((<$>))
15import Control.Applicative ((<$>))
15−import Control.Distributed.Process.Internal.Types
16− ( RemoteTable
17− , Closure(..)
18− , Static(..)
19− , StaticLabel(..)
20− , Process
21− , ProcessId
22− , CallReply(..)
23− )
24import Language.Haskell.TH
16import Language.Haskell.TH
25 ( -- Q monad and operations
17 ( -- Q monad and operations
26 Q
18 Q
@@ -47,12 +39,21 @@ import Language.Haskell.TH
47 , funD
39 , funD
48 , sigD
40 , sigD
49 )
41 )
50−import Control.Distributed.Process (send)
42+import Control.Distributed.Process.Internal.Types
51−import Control.Distributed.Process.Internal.Dynamic (toDyn)
43+ ( RemoteTable
52−import Control.Distributed.Process.Internal.Closure
44+ , Closure(..)
53− ( registerLabel
45+ , Static(..)
54− , registerSender
46+ , StaticLabel(..)
47+ , Process
48+ , ProcessId
49+ , CallReply(..)
50+ , procMsg
51+ , Identifier(ProcessIdentifier)
52+ , remoteTableLabel
53+ , remoteTableSender
55 )
54 )
55+import Control.Distributed.Process.Internal.Dynamic (Dynamic, toDyn)
56+import Control.Distributed.Process.Internal.MessageT (sendMessage)
56
57
57--------------------------------------------------------------------------------
58--------------------------------------------------------------------------------
58-- User-level API --
59-- User-level API --
@@ -139,12 +140,20 @@ generateDecoder n res = [| $(varE n) . decode :: ByteString -> $res |]
139-- | Generate the sender. This is necessary to support 'call'
140-- | Generate the sender. This is necessary to support 'call'
140generateSender :: Q Type -> Q Exp
141generateSender :: Q Type -> Q Exp
141generateSender tp =
142generateSender tp =
143+ -- This duplicates the functionality of 'Process.send', but gets the TH out
144+ -- of the mutually recursive module cycle
142− [| (\pid -> send pid . CallReply) :: ProcessId -> $tp -> Process () |]
145+ [| (\pid msg -> procMsg $ sendMessage (ProcessIdentifier pid) (CallReply msg)) :: ProcessId -> $tp -> Process () |]
143
146
144-- | The name for the function that generates the closure
147-- | The name for the function that generates the closure
145closureName :: Name -> Name
148closureName :: Name -> Name
146closureName n = mkName $ nameBase n ++ "__closure"
149closureName n = mkName $ nameBase n ++ "__closure"
147
150
151+registerLabel :: String -> Dynamic -> RemoteTable -> RemoteTable
152+registerLabel label dyn = remoteTableLabel label ^= Just dyn
153+
154+registerSender :: TypeRep -> Dynamic -> RemoteTable -> RemoteTable
155+registerSender typ dyn = remoteTableSender typ ^= Just dyn
156+
148--------------------------------------------------------------------------------
157--------------------------------------------------------------------------------
149-- Generic Template Haskell auxiliary functions --
158-- Generic Template Haskell auxiliary functions --
150--------------------------------------------------------------------------------
159--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+1−1
@@ -1,4 +1,4 @@
1−-- Add message sending capability to a monad
1+-- | Add message sending capability to a monad
2--
2--
3-- NOTE: Not thread-safe (you should not do concurrent sends within the same
3-- NOTE: Not thread-safe (you should not do concurrent sends within the same
4-- monad).
4-- monad).
distributed-process/src/Control/Distributed/Process/Internal/TypeRep.hs+1
@@ -1,3 +1,4 @@
1+-- | 'Binary' instances for 'TypeRep'
1{-# OPTIONS_GHC -fno-warn-orphans #-}
2{-# OPTIONS_GHC -fno-warn-orphans #-}
2module Control.Distributed.Process.Internal.TypeRep () where
3module Control.Distributed.Process.Internal.TypeRep () where
3
4
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+22−9
@@ -1,5 +1,8 @@
1+-- | Types used throughout the Cloud Haskell framework
2+--
1−-- | We collect all types used internally in a single module to avoid using
3+-- We collect all types used internally in a single module because
4+-- many of these data types are mutually recursive and cannot be split across
2−-- mutually recursive modules
5+-- modules.
3{-# LANGUAGE MagicHash #-}
6{-# LANGUAGE MagicHash #-}
4module Control.Distributed.Process.Internal.Types
7module Control.Distributed.Process.Internal.Types
5 ( -- * Node and process identifiers
8 ( -- * Node and process identifiers
@@ -13,6 +16,7 @@ module Control.Distributed.Process.Internal.Types
13 , LocalProcess(..)
16 , LocalProcess(..)
14 , LocalProcessState(..)
17 , LocalProcessState(..)
15 , Process(..)
18 , Process(..)
19+ , procMsg
16 -- * Typed channels
20 -- * Typed channels
17 , LocalChannelId
21 , LocalChannelId
18 , ChannelId(..)
22 , ChannelId(..)
@@ -77,6 +81,7 @@ import Control.Applicative (Applicative, (<$>), (<*>))
77import Control.Monad.Reader (MonadReader(..), ReaderT)
81import Control.Monad.Reader (MonadReader(..), ReaderT)
78import Control.Monad.IO.Class (MonadIO)
82import Control.Monad.IO.Class (MonadIO)
79import Control.Monad.State (MonadState, StateT)
83import Control.Monad.State (MonadState, StateT)
84+import qualified Control.Monad.Trans.Class as Trans (lift)
80import Control.Distributed.Process.Serializable (Fingerprint, Serializable)
85import Control.Distributed.Process.Serializable (Fingerprint, Serializable)
81import Control.Distributed.Process.Internal.CQueue (CQueue)
86import Control.Distributed.Process.Internal.CQueue (CQueue)
82import Control.Distributed.Process.Internal.Dynamic (Dynamic)
87import Control.Distributed.Process.Internal.Dynamic (Dynamic)
@@ -126,19 +131,24 @@ data Identifier =
126
131
127-- | Local nodes
132-- | Local nodes
128data LocalNode = LocalNode
133data LocalNode = LocalNode
134+ { -- | 'NodeId' of the node
129− { localNodeId :: NodeId
135+ localNodeId :: NodeId
136+ -- | The network endpoint associated with this node
130 , localEndPoint :: NT.EndPoint
137 , localEndPoint :: NT.EndPoint
138+ -- | Local node state
131− , localState :: MVar LocalNodeState
139+ , localState :: MVar LocalNodeState
140+ -- | Channel for the node controller
132 , localCtrlChan :: Chan NCMsg
141 , localCtrlChan :: Chan NCMsg
142+ -- | Runtime lookup table for supporting closures
133− -- TODO: this should be part of the CH state, not the local endpoint state
143+ -- TODO: this should be part of the CH state, not the local endpoint state
134− , remoteTable :: RemoteTable
144+ , remoteTable :: RemoteTable
135 }
145 }
136
146
137-- | Local node state
147-- | Local node state
138data LocalNodeState = LocalNodeState
148data LocalNodeState = LocalNodeState
139− { _localProcesses :: Map LocalProcessId LocalProcess
149+ { _localProcesses :: Map LocalProcessId LocalProcess
140− , _localPidCounter :: Int32
150+ , _localPidCounter :: Int32
141− , _localPidUnique :: Int32
151+ , _localPidUnique :: Int32
142 }
152 }
143
153
144-- | Processes running on our local node
154-- | Processes running on our local node
@@ -163,6 +173,9 @@ newtype Process a = Process {
163 }
173 }
164 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)
174 deriving (Functor, Monad, MonadIO, MonadReader LocalProcess, Typeable, Applicative)
165
175
176+procMsg :: MessageT IO a -> Process a
177+procMsg = Process . Trans.lift
178+
166--------------------------------------------------------------------------------
179--------------------------------------------------------------------------------
167-- Typed channels --
180-- Typed channels --
168--------------------------------------------------------------------------------
181--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Node.hs+15−15
@@ -306,8 +306,8 @@ data NCState = NCState
306newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }
306newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }
307 deriving (Functor, Monad, MonadIO, MonadState NCState)
307 deriving (Functor, Monad, MonadIO, MonadState NCState)
308
308
309−lift :: MessageT IO a -> NC a
309+ncMsg :: MessageT IO a -> NC a
310−lift = NC . Trans.lift
310+ncMsg = NC . Trans.lift
311
311
312initNCState :: NCState
312initNCState :: NCState
313initNCState = NCState { _links = Map.empty
313initNCState = NCState { _links = Map.empty
@@ -321,14 +321,14 @@ initNCState = NCState { _links = Map.empty
321-- [Unified: Table 7]
321-- [Unified: Table 7]
322nodeController :: NC ()
322nodeController :: NC ()
323nodeController = do
323nodeController = do
324− node <- lift getLocalNode
324+ node <- ncMsg getLocalNode
325 forever $ do
325 forever $ do
326 msg <- liftIO $ readChan (localCtrlChan node)
326 msg <- liftIO $ readChan (localCtrlChan node)
327
327
328 -- [Unified: Table 7, rule nc_forward]
328 -- [Unified: Table 7, rule nc_forward]
329 case destNid (ctrlMsgSignal msg) of
329 case destNid (ctrlMsgSignal msg) of
330 Just nid' | nid' /= localNodeId node ->
330 Just nid' | nid' /= localNodeId node ->
331− lift $ sendBinary (NodeIdentifier nid') msg
331+ ncMsg $ sendBinary (NodeIdentifier nid') msg
332 _ ->
332 _ ->
333 return ()
333 return ()
334
334
@@ -356,7 +356,7 @@ ncEffectMonitor :: ProcessId -- ^ Who's watching?
356 -> Maybe MonitorRef -- ^ 'Nothing' to link
356 -> Maybe MonitorRef -- ^ 'Nothing' to link
357 -> NC ()
357 -> NC ()
358ncEffectMonitor from them mRef = do
358ncEffectMonitor from them mRef = do
359− node <- lift getLocalNode
359+ node <- ncMsg getLocalNode
360 shouldLink <-
360 shouldLink <-
361 if not (isLocal node them)
361 if not (isLocal node them)
362 then return True
362 then return True
@@ -370,7 +370,7 @@ ncEffectMonitor from them mRef = do
370 (False, True) -> -- [Unified: second rule]
370 (False, True) -> -- [Unified: second rule]
371 notifyDied from them DiedNoProc mRef
371 notifyDied from them DiedNoProc mRef
372 (False, False) -> -- [Unified: third rule]
372 (False, False) -> -- [Unified: third rule]
373− lift $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg
373+ ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg
374 { ctrlMsgSender = NodeIdentifier (localNodeId node)
374 { ctrlMsgSender = NodeIdentifier (localNodeId node)
375 , ctrlMsgSignal = Died (ProcessIdentifier them) DiedNoProc
375 , ctrlMsgSignal = Died (ProcessIdentifier them) DiedNoProc
376 }
376 }
@@ -378,21 +378,21 @@ ncEffectMonitor from them mRef = do
378-- [Unified: Table 11]
378-- [Unified: Table 11]
379ncEffectUnlink :: ProcessId -> ProcessId -> NC ()
379ncEffectUnlink :: ProcessId -> ProcessId -> NC ()
380ncEffectUnlink from them = do
380ncEffectUnlink from them = do
381− node <- lift getLocalNode
381+ node <- ncMsg getLocalNode
382 when (isLocal node from) $ postMessage from $ DidUnlink them
382 when (isLocal node from) $ postMessage from $ DidUnlink them
383 modify $ linksForProcess them ^: Set.delete from
383 modify $ linksForProcess them ^: Set.delete from
384
384
385-- [Unified: Table 11]
385-- [Unified: Table 11]
386ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
386ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
387ncEffectUnmonitor from ref = do
387ncEffectUnmonitor from ref = do
388− node <- lift getLocalNode
388+ node <- ncMsg getLocalNode
389 when (isLocal node from) $ postMessage from $ DidUnmonitor ref
389 when (isLocal node from) $ postMessage from $ DidUnmonitor ref
390 modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref
390 modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref
391
391
392-- [Unified: Table 12, bottom rule]
392-- [Unified: Table 12, bottom rule]
393ncEffectNodeDied :: NodeId -> DiedReason -> NC ()
393ncEffectNodeDied :: NodeId -> DiedReason -> NC ()
394ncEffectNodeDied nid reason = do
394ncEffectNodeDied nid reason = do
395− node <- lift getLocalNode
395+ node <- ncMsg getLocalNode
396 lnks <- gets (^. linksForNode nid)
396 lnks <- gets (^. linksForNode nid)
397 mons <- gets (^. monitorsForNode nid)
397 mons <- gets (^. monitorsForNode nid)
398
398
@@ -434,9 +434,9 @@ ncEffectSpawn pid cProc ref = do
434 mProc <- unClosure cProc
434 mProc <- unClosure cProc
435 case mProc of
435 case mProc of
436 Just proc -> do
436 Just proc -> do
437− node <- lift getLocalNode
437+ node <- ncMsg getLocalNode
438 pid' <- liftIO $ forkProcess node proc
438 pid' <- liftIO $ forkProcess node proc
439− lift $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')
439+ ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')
440 Nothing ->
440 Nothing ->
441 -- TODO: what should we do when unClosure returns Nothing?
441 -- TODO: what should we do when unClosure returns Nothing?
442 liftIO . putStrLn $ "Error: closure " ++ show cProc ++ " not found"
442 liftIO . putStrLn $ "Error: closure " ++ show cProc ++ " not found"
@@ -452,7 +452,7 @@ notifyDied :: ProcessId -- ^ Who to notify?
452 -> Maybe MonitorRef -- ^ 'Nothing' for linking
452 -> Maybe MonitorRef -- ^ 'Nothing' for linking
453 -> NC ()
453 -> NC ()
454notifyDied dest src reason mRef = do
454notifyDied dest src reason mRef = do
455− node <- lift getLocalNode
455+ node <- ncMsg getLocalNode
456 case (isLocal node dest, mRef) of
456 case (isLocal node dest, mRef) of
457 (True, Just ref) ->
457 (True, Just ref) ->
458 postMessage dest $ MonitorNotification ref src reason
458 postMessage dest $ MonitorNotification ref src reason
@@ -461,7 +461,7 @@ notifyDied dest src reason mRef = do
461 (False, _) ->
461 (False, _) ->
462 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
462 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
463 -- (see [Unified: Table 10]
463 -- (see [Unified: Table 10]
464− lift $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg
464+ ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg
465 { ctrlMsgSender = NodeIdentifier (localNodeId node)
465 { ctrlMsgSender = NodeIdentifier (localNodeId node)
466 , ctrlMsgSignal = Died (ProcessIdentifier src) reason
466 , ctrlMsgSignal = Died (ProcessIdentifier src) reason
467 }
467 }
@@ -494,7 +494,7 @@ isLocal nid pid = processNodeId pid == localNodeId nid
494-- TODO: Duplication with onClosure in Process
494-- TODO: Duplication with onClosure in Process
495unClosure :: Typeable a => Closure a -> NC (Maybe a)
495unClosure :: Typeable a => Closure a -> NC (Maybe a)
496unClosure (Closure (Static label) env) = do
496unClosure (Closure (Static label) env) = do
497− rtable <- remoteTable <$> lift getLocalNode
497+ rtable <- remoteTable <$> ncMsg getLocalNode
498 case resolveClosure rtable label env of
498 case resolveClosure rtable label env of
499 Nothing -> return Nothing
499 Nothing -> return Nothing
500 Just dyn -> return (fromDynamic dyn)
500 Just dyn -> return (fromDynamic dyn)
@@ -513,7 +513,7 @@ throwException pid e = withLocalProc pid $ \p ->
513
513
514withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
514withLocalProc :: ProcessId -> (LocalProcess -> IO ()) -> NC ()
515withLocalProc pid p = do
515withLocalProc pid p = do
516− node <- lift getLocalNode
516+ node <- ncMsg getLocalNode
517 liftIO $ do
517 liftIO $ do
518 -- By [Unified: table 6, rule missing_process] messages to dead processes
518 -- By [Unified: table 6, rule missing_process] messages to dead processes
519 -- can silently be dropped
519 -- can silently be dropped
distributed-process/src/Control/Distributed/Process/Serializable.hs+1
@@ -21,6 +21,7 @@ import qualified Data.ByteString.Internal as BSI ( unsafeCreate
21import Foreign.Storable (pokeByteOff, peekByteOff, sizeOf)
21import Foreign.Storable (pokeByteOff, peekByteOff, sizeOf)
22import Foreign.ForeignPtr (withForeignPtr)
22import Foreign.ForeignPtr (withForeignPtr)
23
23
24+-- | Objects that can be sent across the network
24class (Binary a, Typeable a) => Serializable a
25class (Binary a, Typeable a) => Serializable a
25instance (Binary a, Typeable a) => Serializable a
26instance (Binary a, Typeable a) => Serializable a
26
27
distributed-process/tests/TestClosure.hs+43−3
@@ -1,8 +1,9 @@
1module Main where
1module Main where
2
2
3−import Control.Monad (join)
3+import Control.Monad (join, replicateM)
4import Control.Monad.IO.Class (liftIO)
4import Control.Monad.IO.Class (liftIO)
5+import Control.Exception (IOException, throw)
5−import Control.Concurrent (forkIO)
6+import Control.Concurrent (forkIO, threadDelay)
6import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)
7import Control.Concurrent.MVar (MVar, newEmptyMVar, readMVar, takeMVar, putMVar)
7import Control.Applicative ((<$>))
8import Control.Applicative ((<$>))
8import Network.Transport (Transport)
9import Network.Transport (Transport)
@@ -39,7 +40,10 @@ returnInt = return
39returnForTestApply :: (Closure (Int -> Process Int), Int) -> (Closure (Int -> Process Int), Int)
40returnForTestApply :: (Closure (Int -> Process Int), Int) -> (Closure (Int -> Process Int), Int)
40returnForTestApply = id
41returnForTestApply = id
41
42
43+wait :: Int -> Process ()
44+wait = liftIO . threadDelay
45+
42−$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid, 'factorial, 'factorialOf, 'returnInt, 'returnForTestApply])
46+$(remotable ['addInt, 'putInt, 'sendInt, 'sendPid, 'factorial, 'factorialOf, 'returnInt, 'returnForTestApply, 'wait])
43
47
44testSendPureClosure :: Transport -> RemoteTable -> IO ()
48testSendPureClosure :: Transport -> RemoteTable -> IO ()
45testSendPureClosure transport rtable = do
49testSendPureClosure transport rtable = do
@@ -205,6 +209,41 @@ testApply transport rtable = do
205 foo :: Closure (Closure (Int -> Process Int), Int)
209 foo :: Closure (Closure (Int -> Process Int), Int)
206 foo = $(mkClosure 'returnForTestApply) ($(mkClosure 'factorialOf) (), 5)
210 foo = $(mkClosure 'returnForTestApply) ($(mkClosure 'factorialOf) (), 5)
207
211
212+-- Test 'spawnSupervised'
213+--
214+-- Set up a supervisor, spawn a child, then have a third process monitor the
215+-- child. The supervisor then throws an exception, the child dies because it
216+-- was linked to the supervisor, and the third process notices that the child
217+-- dies.
218+testSpawnSupervised :: Transport -> RemoteTable -> IO ()
219+testSpawnSupervised transport rtable = do
220+ [node1, node2] <- replicateM 2 $ newLocalNode transport rtable
221+ [superPid, childPid] <- replicateM 2 $ newEmptyMVar
222+ thirdProcessDone <- newEmptyMVar
223+
224+ forkProcess node1 $ do
225+ us <- getSelfPid
226+ liftIO $ putMVar superPid us
227+ (child, _ref) <- spawnSupervised (localNodeId node2) ($(mkClosure 'wait) 1000000)
228+ liftIO $ do
229+ putMVar childPid child
230+ threadDelay 500000 -- Give the child a chance to link to us
231+ throw supervisorDeath
232+
233+ forkProcess node2 $ do
234+ [super, child] <- liftIO $ mapM readMVar [superPid, childPid]
235+ ref <- monitor child
236+ MonitorNotification ref' pid' (DiedException e) <- expect
237+ True <- return $ ref' == ref
238+ && pid' == child
239+ && e == show (LinkException super (DiedException (show supervisorDeath)))
240+ liftIO $ putMVar thirdProcessDone ()
241+
242+ takeMVar thirdProcessDone
243+ where
244+ supervisorDeath :: IOException
245+ supervisorDeath = userError "Supervisor died"
246+
208main :: IO ()
247main :: IO ()
209main = do
248main = do
210 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
249 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters
@@ -219,4 +258,5 @@ main = do
219 , ("CallBind", testCallBind transport rtable)
258 , ("CallBind", testCallBind transport rtable)
220 , ("Seq", testSeq transport rtable)
259 , ("Seq", testSeq transport rtable)
221 , ("Apply", testApply transport rtable)
260 , ("Apply", testApply transport rtable)
261+ , ("SpawnSupervised", testSpawnSupervised transport rtable)
222 ]
262 ]