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