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