git.iko.soy/distributed-process
on 1c2bc74 help sign in
distributed-process/src/Control/Distributed/Process/Internal/Spawn.hs
168 lines 4.8 kB history raw
1module Control.Distributed.Process.Internal.Spawn
2 ( spawn
3 , spawnLink
4 , spawnMonitor
5 , call
6 , spawnSupervised
7 , spawnChannel
8 ) where
10import Data.Typeable (Typeable)
11import Control.Monad.IO.Class (liftIO)
12import Control.Applicative ((<$>))
13import Control.Monad.Reader (ask)
14import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
15import Control.Distributed.Static
16 ( Static
17 , Closure
18 , closureCompose
19 , staticClosure
20 )
21import Control.Distributed.Process.Internal.Types
22 ( NodeId(..)
23 , ProcessId(..)
24 , Process(..)
25 , MonitorRef(..)
26 , ProcessMonitorNotification(..)
27 , NodeMonitorNotification(..)
28 , DidSpawn(..)
29 , SendPort(..)
30 , ReceivePort(..)
31 , LocalProcess(processNode)
32 , nullProcessId
33 )
34import Control.Distributed.Process.Serializable (Serializable, SerializableDict)
35import Control.Distributed.Process.Internal.Closure.BuiltIn
36 ( sdictSendPort
37 , sndStatic
38 , idCP
39 , seqCP
40 , bindCP
41 , splitCP
42 , cpLink
43 , cpSend
44 , cpNewChan
45 , cpDelay
46 )
47import Control.Distributed.Process.Internal.Primitives
48 ( -- Basic messaging
49 send
50 , expect
51 , receiveWait
52 , match
53 , matchIf
54 , link
55 , getSelfPid
56 , monitor
57 , monitorNode
58 , unmonitor
59 , spawnAsync
60 , reconnect
61 , newChan
62 )
64-- | Spawn a process
65--
66-- For more information about 'Closure', see
67-- "Control.Distributed.Process.Closure".
68--
69-- See also 'call'.
70spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
71spawn nid proc = do
72 us <- getSelfPid
73 mRef <- monitorNode nid
74 sRef <- spawnAsync nid (cpDelay us proc)
75 receiveWait [
76 matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
77 unmonitor mRef
78 send pid ()
79 return pid
80 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
81 return (nullProcessId nid)
82 ]
84-- | Spawn a process and link to it
85--
86-- Note that this is just the sequential composition of 'spawn' and 'link'.
87-- (The "Unified" semantics that underlies Cloud Haskell does not even support
88-- a synchronous link operation)
89spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId
90spawnLink nid proc = do
91 pid <- spawn nid proc
92 link pid
93 return pid
95-- | Like 'spawnLink', but monitor the spawned process
96spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)
97spawnMonitor nid proc = do
98 pid <- spawn nid proc
99 ref <- monitor pid
100 return (pid, ref)
102-- | Run a process remotely and wait for it to reply
103--
104-- We monitor the remote process: if it dies before it can send a reply, we die
105-- too.
106--
107-- For more information about 'Static', 'SerializableDict', and 'Closure', see
108-- "Control.Distributed.Process.Closure".
109--
110-- See also 'spawn'.
111call :: Serializable a
112 => Static (SerializableDict a)
113 -> NodeId
114 -> Closure (Process a)
115 -> Process a
116call dict nid proc = do
117 us <- getSelfPid
118 (pid, mRef) <- spawnMonitor nid (proc `bindCP` cpSend dict us)
119 -- We are guaranteed to receive the reply before the monitor notification
120 -- (if a reply is sent at all)
121 -- NOTE: This might not be true if we switch to unreliable delivery.
122 mResult <- receiveWait
123 [ match (return . Right)
124 , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
125 (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))
126 ]
127 case mResult of
128 Right a -> do
129 -- Wait for the monitor message so that we the mailbox doesn't grow
130 receiveWait
131 [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
132 (\(ProcessMonitorNotification {}) -> return ())
133 ]
134 -- Clean up connection to pid
135 reconnect pid
136 return a
137 Left err ->
138 fail $ "call: remote process died: " ++ show err
140-- | Spawn a child process, have the child link to the parent and the parent
141-- monitor the child
142spawnSupervised :: NodeId
143 -> Closure (Process ())
144 -> Process (ProcessId, MonitorRef)
145spawnSupervised nid proc = do
146 us <- getSelfPid
147 them <- spawn nid (cpLink us `seqCP` proc)
148 ref <- monitor them
149 return (them, ref)
151-- | Spawn a new process, supplying it with a new 'ReceivePort' and return
152-- the corresponding 'SendPort'.
153spawnChannel :: forall a. Typeable a => Static (SerializableDict a)
154 -> NodeId
155 -> Closure (ReceivePort a -> Process ())
156 -> Process (SendPort a)
157spawnChannel dict nid proc = do
158 us <- getSelfPid
159 _ <- spawn nid (go us)
160 expect
161 where
162 go :: ProcessId -> Closure (Process ())
163 go pid = cpNewChan dict
164 `bindCP`
165 (cpSend (sdictSendPort dict) pid `splitCP` proc)
166 `bindCP`
167 (idCP `closureCompose` staticClosure sndStatic)