git.iko.soy/distributed-process
on mx-fix help sign in
distributed-process/src/Control/Distributed/Process/Internal/Spawn.hs
165 lines 4.7 kB history raw
1{-# LANGUAGE RankNTypes #-}
2module Control.Distributed.Process.Internal.Spawn
3 ( spawn
4 , spawnLink
5 , spawnMonitor
6 , call
7 , spawnSupervised
8 , spawnChannel
9 ) where
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 , cpDelayed
41 , returnCP
42 , sdictUnit
43 )
44import Control.Distributed.Process.Internal.Primitives
45 ( -- Basic messaging
46 usend
47 , expect
48 , receiveWait
49 , match
50 , matchIf
51 , link
52 , getSelfPid
53 , monitor
54 , monitorNode
55 , unmonitor
56 , spawnAsync
57 , reconnect
58 )
60-- | Spawn a process
61--
62-- For more information about 'Closure', see
63-- "Control.Distributed.Process.Closure".
64--
65-- See also 'call'.
66spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
67spawn nid proc = do
68 us <- getSelfPid
69 mRef <- monitorNode nid
70 sRef <- spawnAsync nid (cpDelayed us proc)
71 receiveWait [
72 matchIf (\(DidSpawn ref _) -> ref == sRef) $ \(DidSpawn _ pid) -> do
73 unmonitor mRef
74 usend pid ()
75 return pid
76 , matchIf (\(NodeMonitorNotification ref _ _) -> ref == mRef) $ \_ ->
77 return (nullProcessId nid)
78 ]
80-- | Spawn a process and link to it
81--
82-- Note that this is just the sequential composition of 'spawn' and 'link'.
83-- (The "Unified" semantics that underlies Cloud Haskell does not even support
84-- a synchronous link operation)
85spawnLink :: NodeId -> Closure (Process ()) -> Process ProcessId
86spawnLink nid proc = do
87 pid <- spawn nid proc
88 link pid
89 return pid
91-- | Like 'spawnLink', but monitor the spawned process
92spawnMonitor :: NodeId -> Closure (Process ()) -> Process (ProcessId, MonitorRef)
93spawnMonitor nid proc = do
94 pid <- spawn nid proc
95 ref <- monitor pid
96 return (pid, ref)
98-- | Run a process remotely and wait for it to reply
99--
100-- We monitor the remote process: if it dies before it can send a reply, we die
101-- too.
102--
103-- For more information about 'Static', 'SerializableDict', and 'Closure', see
104-- "Control.Distributed.Process.Closure".
105--
106-- See also 'spawn'.
107call :: Serializable a
108 => Static (SerializableDict a)
109 -> NodeId
110 -> Closure (Process a)
111 -> Process a
112call dict nid proc = do
113 us <- getSelfPid
114 (pid, mRef) <- spawnMonitor nid (proc `bindCP`
115 cpSend dict us `seqCP`
116 -- Delay so the process does not terminate
117 -- before the response arrives.
118 cpDelayed us (returnCP sdictUnit ())
119 )
120 mResult <- receiveWait
121 [ match $ \a -> usend pid () >> return (Right a)
122 , matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
123 (\(ProcessMonitorNotification _ _ reason) -> return (Left reason))
124 ]
125 case mResult of
126 Right a -> do
127 -- Wait for the monitor message so that we the mailbox doesn't grow
128 receiveWait
129 [ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mRef)
130 (\(ProcessMonitorNotification {}) -> return ())
131 ]
132 -- Clean up connection to pid
133 reconnect pid
134 return a
135 Left err ->
136 fail $ "call: remote process died: " ++ show err
138-- | Spawn a child process, have the child link to the parent and the parent
139-- monitor the child
140spawnSupervised :: NodeId
141 -> Closure (Process ())
142 -> Process (ProcessId, MonitorRef)
143spawnSupervised nid proc = do
144 us <- getSelfPid
145 them <- spawn nid (cpLink us `seqCP` proc)
146 ref <- monitor them
147 return (them, ref)
149-- | Spawn a new process, supplying it with a new 'ReceivePort' and return
150-- the corresponding 'SendPort'.
151spawnChannel :: forall a. Serializable a => Static (SerializableDict a)
152 -> NodeId
153 -> Closure (ReceivePort a -> Process ())
154 -> Process (SendPort a)
155spawnChannel dict nid proc = do
156 us <- getSelfPid
157 _ <- spawn nid (go us)
158 expect
159 where
160 go :: ProcessId -> Closure (Process ())
161 go pid = cpNewChan dict
162 `bindCP`
163 (cpSend (sdictSendPort dict) pid `splitCP` proc)
164 `bindCP`
165 (idCP `closureCompose` staticClosure sndStatic)