git.iko.soy/distributed-process
on master help sign in
Some infrastructure for typed channels

(typed channels are not yet implemented though)

commit4d0f927b13a63f72b4111567dd0ac3277777f09e authorEdsko de Vries <edsko@edsko.net> dateWed 20 Jun 2012, 17:34 (14 years ago) parent2037164
5 files changed+193−97
compare with master
unifiedsplit
distributed-process/src/Control/Distributed/Process.hs+43−5 distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+1−1 distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+10−13 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+61−29 distributed-process/src/Control/Distributed/Process/Node.hs+78−49
distributed-process/src/Control/Distributed/Process.hs+43−5
@@ -15,6 +15,10 @@ module Control.Distributed.Process
1515 -- * Basic messaging
1616 , send
1717 , expect
18+ -- * Channels
19+ , ReceivePort
20+ , SendPort
21+ , newChan
1822 -- * Advanced messaging
1923 , Match
2024 , receiveWait
@@ -62,9 +66,13 @@ import Control.Exception (Exception, throw)
6266import qualified Control.Exception as Exception (catch)
6367import Control.Concurrent.MVar (modifyMVar)
6468import Control.Concurrent.Chan (writeChan)
65−import Control.Distributed.Process.Internal.CQueue (dequeue, BlockSpec(..))
69+import Control.Distributed.Process.Internal.CQueue
70+ ( newCQueue
71+ , dequeue
72+ , BlockSpec(..)
73+ )
6674import Control.Distributed.Process.Serializable (Serializable, fingerprint)
67−import Data.Accessor ((^.), (^:))
75+import Data.Accessor ((^.), (^:), (^=))
6876import Control.Distributed.Process.Internal.Types
6977 ( RemoteTable
7078 , NodeId(..)
@@ -90,6 +98,13 @@ import Control.Distributed.Process.Internal.Types
9098 , spawnCounter
9199 , MessageT
92100 , Closure(..)
101+ , SendPort(..)
102+ , ReceivePort(..)
103+ , channelCounter
104+ , typedChannelWithId
105+ , TypedChannel(..)
106+ , ChannelId(..)
107+ , Identifier(..)
93108 )
94109import Control.Distributed.Process.Internal.MessageT
95110 ( sendMessage
@@ -164,12 +179,35 @@ import Control.Distributed.Process.Internal.MessageT
164179send :: Serializable a => ProcessId -> a -> Process ()
165180-- This requires a lookup on every send. If we want to avoid that we need to
166181-- modify serializable to allow for stateful (IO) deserialization
167−send them msg = lift $ sendMessage (Left them) msg
182+send them msg = lift $ sendMessage (ProcessIdentifier them) msg
168183
169184-- | Wait for a message of a specific type
170185expect :: forall a. Serializable a => Process a
171186expect = receiveWait [match return]
172187
188+--------------------------------------------------------------------------------
189+-- Channels --
190+--------------------------------------------------------------------------------
191+
192+-- | Create a new typed channel (WORK IN PROGRESS)
193+newChan :: Serializable a => Process (SendPort a, ReceivePort a)
194+newChan = do
195+ proc <- ask
196+ liftIO . modifyMVar (processState proc) $ \st -> do
197+ queue <- liftIO newCQueue
198+ let lcid = st ^. channelCounter
199+ cid = ChannelId { channelProcessId = processId proc
200+ , channelLocalId = lcid
201+ }
202+ sport = SendPort cid
203+ rport = ReceivePort queue
204+ tch = TypedChannel queue
205+ return ( (channelCounter ^: (+ 1))
206+ . (typedChannelWithId lcid ^= Just tch)
207+ $ st
208+ , (sport, rport)
209+ )
210+
173211--------------------------------------------------------------------------------
174212-- Advanced messaging --
175213--------------------------------------------------------------------------------
@@ -320,7 +358,7 @@ sendCtrlMsg :: Maybe NodeId -- ^ Nothing for the local node
320358 -> Process ()
321359sendCtrlMsg mNid signal = do
322360 us <- getSelfPid
323− let msg = NCMsg { ctrlMsgSender = Left us
361+ let msg = NCMsg { ctrlMsgSender = ProcessIdentifier us
324362 , ctrlMsgSignal = signal
325363 }
326364 case mNid of
@@ -328,7 +366,7 @@ sendCtrlMsg mNid signal = do
328366 ctrlChan <- localCtrlChan <$> lift getLocalNode
329367 liftIO $ writeChan ctrlChan msg
330368 Just nid ->
331− lift $ sendBinary (Right nid) msg
369+ lift $ sendBinary (NodeIdentifier nid) msg
332370
333371lookupStatic :: Typeable a => String -> Process a
334372lookupStatic label = do
distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+1−1
@@ -39,7 +39,7 @@ data BlockSpec =
3939
4040dequeue :: forall a b.
4141 CQueue a -- ^ Queue
42− -> BlockSpec -- ^ Blocking behaviour
42+ -> BlockSpec -- ^ Blocking behaviour
4343 -> [a -> Maybe b] -- ^ List of matches
4444 -> IO (Maybe b) -- ^ 'Nothing' only on timeout
4545dequeue (CQueue arrived incoming) blockSpec matches =
distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+10−13
@@ -22,10 +22,10 @@ import Control.Monad.State (gets, modify)
2222import Control.Monad.IO.Class (MonadIO, liftIO)
2323import Control.Concurrent.Chan (writeChan)
2424import Control.Distributed.Process.Internal.Types
25− ( Identifier
25+ ( Identifier(..)
2626 , NodeId(nodeAddress)
27− , ProcessId(processNodeId, processLocalId)
28− , idToPayload
27+ , ProcessId(processNodeId)
28+ , ChannelId(channelProcessId)
2929 , LocalNode(localCtrlChan, localEndPoint)
3030 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)
3131 , DiedReason(DiedDisconnect)
@@ -75,12 +75,12 @@ setupConnTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
7575setupConnTo them = do
7676 endPoint <- localEndPoint `liftM` getLocalNode
7777 mConn <- liftIO $ NT.connect endPoint
78− addr
78+ (nodeAddress . identifierNode $ them)
7979 NT.ReliableOrdered
8080 NT.defaultConnectHints
8181 case mConn of
8282 Right conn -> do
83− didSend <- liftIO $ NT.send conn firstMsg
83+ didSend <- liftIO $ NT.send conn (BSL.toChunks . encode $ them)
8484 case didSend of
8585 Left _ ->
8686 return Nothing
@@ -89,14 +89,6 @@ setupConnTo them = do
8989 return $ Just conn
9090 Left _ ->
9191 return Nothing
92− where
93− (addr, firstMsg) = case them of
94− Left pid -> ( nodeAddress (processNodeId pid)
95− , idToPayload (Just $ processLocalId pid)
96− )
97− Right nid -> ( nodeAddress nid
98− , idToPayload Nothing
99− )
10092
10193connTo :: MonadIO m => Identifier -> MessageT m (Maybe NT.Connection)
10294connTo them = do
@@ -105,6 +97,11 @@ connTo them = do
10597 Just conn -> return $ Just conn
10698 Nothing -> setupConnTo them
10799
100+identifierNode :: Identifier -> NodeId
101+identifierNode (NodeIdentifier nid) = nid
102+identifierNode (ProcessIdentifier pid) = processNodeId pid
103+identifierNode (ChannelIdentifier cid) = processNodeId (channelProcessId cid)
104+
108105--------------------------------------------------------------------------------
109106-- Accessors --
110107--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Internal/Types.hs+61−29
@@ -8,7 +8,7 @@ module Control.Distributed.Process.Internal.Types
88 NodeId(..)
99 , LocalProcessId(..)
1010 , ProcessId(..)
11− , Identifier
11+ , Identifier(..)
1212 -- * Local nodes and processes
1313 , LocalNode(..)
1414 , LocalNodeState(..)
@@ -16,6 +16,12 @@ module Control.Distributed.Process.Internal.Types
1616 , LocalProcessState(..)
1717 , Process(..)
1818 , runLocalProcess
19+ -- * Typed channels
20+ , LocalChannelId
21+ , ChannelId(..)
22+ , TypedChannel(..)
23+ , SendPort(..)
24+ , ReceivePort(..)
1925 -- * Closures
2026 , Static(..)
2127 , Closure(..)
@@ -37,8 +43,6 @@ module Control.Distributed.Process.Internal.Types
3743 , createMessage
3844 , messageToPayload
3945 , payloadToMessage
40− , idToPayload
41− , payloadToId
4246 -- * Accessors
4347 , localProcesses
4448 , localPidCounter
@@ -46,6 +50,9 @@ module Control.Distributed.Process.Internal.Types
4650 , localProcessWithId
4751 , monitorCounter
4852 , spawnCounter
53+ , channelCounter
54+ , typedChannels
55+ , typedChannelWithId
4956 -- * MessageT monad
5057 , MessageT(..)
5158 , MessageState(..)
@@ -67,9 +74,8 @@ import Data.ByteString.Lazy (ByteString)
6774import Data.Accessor (Accessor, accessor)
6875import qualified Data.Accessor.Container as DAC (mapMaybe)
6976import qualified Data.ByteString.Lazy as BSL (ByteString, toChunks, fromChunks, splitAt)
70−import qualified Data.ByteString as BSS (ByteString, concat, splitAt)
77+import qualified Data.ByteString as BSS (ByteString, concat)
7178import qualified Network.Transport as NT (EndPoint, EndPointAddress, Connection)
72−import qualified Network.Transport.Internal as NTI (encodeInt32, decodeInt32)
7379import Control.Applicative ((<$>), (<*>))
7480import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)
7581import Control.Monad.IO.Class (MonadIO)
@@ -126,8 +132,12 @@ data ProcessId = ProcessId
126132instance Show ProcessId where
127133 show pid = show (processNodeId pid) ++ ":" ++ show (processLocalId pid)
128134
129−-- | Node or process identifier
130−type Identifier = Either ProcessId NodeId
135+-- | Union of all kinds of identifiers
136+data Identifier =
137+ ProcessIdentifier ProcessId
138+ | NodeIdentifier NodeId
139+ | ChannelIdentifier ChannelId
140+ deriving (Show, Eq, Ord)
131141
132142--------------------------------------------------------------------------------
133143-- Local nodes and processes --
@@ -162,6 +172,8 @@ data LocalProcess = LocalProcess
162172data LocalProcessState = LocalProcessState
163173 { _monitorCounter :: Int32
164174 , _spawnCounter :: Int32
175+ , _channelCounter :: Int32
176+ , _typedChannels :: Map LocalChannelId TypedChannel
165177 }
166178
167179-- | The Cloud Haskell 'Process' type
@@ -174,6 +186,23 @@ newtype Process a = Process {
174186runLocalProcess :: LocalNode -> Process a -> LocalProcess -> IO a
175187runLocalProcess node proc = runMessageT node . runReaderT (unProcess proc)
176188
189+--------------------------------------------------------------------------------
190+-- Typed channels --
191+--------------------------------------------------------------------------------
192+
193+type LocalChannelId = Int32
194+
195+data ChannelId = ChannelId {
196+ channelProcessId :: ProcessId
197+ , channelLocalId :: LocalChannelId
198+ }
199+ deriving (Show, Eq, Ord)
200+
201+data TypedChannel = forall a. Serializable a => TypedChannel (CQueue a)
202+
203+newtype SendPort a = SendPort ChannelId
204+newtype ReceivePort a = ReceivePort (CQueue a)
205+
177206--------------------------------------------------------------------------------
178207-- Closures --
179208--------------------------------------------------------------------------------
@@ -292,28 +321,6 @@ payloadToMessage payload = Message fp msg
292321 $ BSL.fromChunks payload
293322 fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp
294323
295−
296−-- | The first message we send across a connection to indicate the intended
297−-- recipient. Pass Nothing for the remote node controller
298−idToPayload :: Maybe LocalProcessId -> [BSS.ByteString]
299−idToPayload Nothing = [ NTI.encodeInt32 (0 :: Int) ]
300−idToPayload (Just lpid) = [ NTI.encodeInt32 (1 :: Int)
301− , NTI.encodeInt32 (lpidCounter lpid)
302− , NTI.encodeInt32 (lpidUnique lpid)
303− ]
304−
305−-- | Inverse of 'idToPayload'
306−payloadToId :: [BSS.ByteString] -> Maybe LocalProcessId
307−payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss
308− (bs2, bs3) = BSS.splitAt 4 bss' in
309− case NTI.decodeInt32 bs1 :: Int of
310− 0 -> Nothing
311− 1 -> Just LocalProcessId
312− { lpidCounter = NTI.decodeInt32 bs2
313− , lpidUnique = NTI.decodeInt32 bs3
314− }
315− _ -> fail "payloadToId"
316−
317324--------------------------------------------------------------------------------
318325-- Binary instances --
319326--------------------------------------------------------------------------------
@@ -380,6 +387,22 @@ instance Binary DidSpawn where
380387 put (DidSpawn ref pid) = put ref >> put pid
381388 get = DidSpawn <$> get <*> get
382389
390+instance Binary ChannelId where
391+ put cid = put (channelProcessId cid) >> put (channelLocalId cid)
392+ get = ChannelId <$> get <*> get
393+
394+instance Binary Identifier where
395+ put (ProcessIdentifier pid) = putWord8 0 >> put pid
396+ put (NodeIdentifier nid) = putWord8 1 >> put nid
397+ put (ChannelIdentifier cid) = putWord8 2 >> put cid
398+ get = do
399+ header <- getWord8
400+ case header of
401+ 0 -> ProcessIdentifier <$> get
402+ 1 -> NodeIdentifier <$> get
403+ 2 -> ChannelIdentifier <$> get
404+ _ -> fail "Identifier.get: invalid"
405+
383406--------------------------------------------------------------------------------
384407-- Accessors --
385408--------------------------------------------------------------------------------
@@ -402,6 +425,15 @@ monitorCounter = accessor _monitorCounter (\cnt st -> st { _monitorCounter = cnt
402425spawnCounter :: Accessor LocalProcessState Int32
403426spawnCounter = accessor _spawnCounter (\cnt st -> st { _spawnCounter = cnt })
404427
428+channelCounter :: Accessor LocalProcessState LocalChannelId
429+channelCounter = accessor _channelCounter (\cnt st -> st { _channelCounter = cnt })
430+
431+typedChannels :: Accessor LocalProcessState (Map LocalChannelId TypedChannel)
432+typedChannels = accessor _typedChannels (\cs st -> st { _typedChannels = cs })
433+
434+typedChannelWithId :: LocalChannelId -> Accessor LocalProcessState (Maybe TypedChannel)
435+typedChannelWithId cid = typedChannels >>> DAC.mapMaybe cid
436+
405437--------------------------------------------------------------------------------
406438-- MessageT monad --
407439--------------------------------------------------------------------------------
distributed-process/src/Control/Distributed/Process/Node.hs+78−49
@@ -71,7 +71,6 @@ import Control.Distributed.Process.Internal.Types
7171 , NCMsg(..)
7272 , ProcessSignal(..)
7373 , payloadToMessage
74− , payloadToId
7574 , localPidCounter
7675 , localPidUnique
7776 , localProcessWithId
@@ -88,6 +87,8 @@ import Control.Distributed.Process.Internal.Types
8887 , createMessage
8988 , MessageT
9089 , runMessageT
90+ , TypedChannel(..)
91+ , Identifier(..)
9192 )
9293import Control.Distributed.Process.Serializable (Serializable)
9394import Control.Distributed.Process.Internal.MessageT
@@ -152,6 +153,8 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do
152153 }
153154 pst <- newMVar LocalProcessState { _monitorCounter = 0
154155 , _spawnCounter = 0
156+ , _channelCounter = 0
157+ , _typedChannels = Map.empty
155158 }
156159 queue <- newCQueue
157160 (_, lproc) <- fixIO $ \ ~(tid, _) -> do
@@ -168,8 +171,8 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do
168171 modifyMVar_ (localState node) $
169172 return . (localProcessWithId lpid ^= Nothing)
170173 writeChan (localCtrlChan node) NCMsg
171− { ctrlMsgSender = Left pid
172− , ctrlMsgSignal = Died (Left pid) reason
174+ { ctrlMsgSender = ProcessIdentifier pid
175+ , ctrlMsgSignal = Died (ProcessIdentifier pid) reason
173176 }
174177 return (tid', lproc)
175178
@@ -181,35 +184,45 @@ forkProcess node proc = modifyMVar (localState node) $ \st -> do
181184 )
182185
183186handleIncomingMessages :: LocalNode -> IO ()
184−handleIncomingMessages node = go [] Map.empty Set.empty
187+handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
185188 where
186189 go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know
187190 -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes
191+ -> Map NT.ConnectionId TypedChannel -- ^ Connections to typed channels
188192 -> Set NT.ConnectionId -- ^ Connections to our controller
189193 -> IO ()
190− go uninitConns procConns ctrlConns = do
194+ go uninitConns procs chans ctrls = do
191195 event <- NT.receive endpoint
192196 case event of
193197 NT.ConnectionOpened cid _rel _theirAddr ->
194− go (cid : uninitConns) procConns ctrlConns
198+ go (cid : uninitConns) procs chans ctrls
195199 NT.Received cid payload ->
196− case Map.lookup cid procConns of
197− Just proc -> do
200+ case ( Map.lookup cid procs
201+ , Map.lookup cid chans
202+ , cid `Set.member` ctrls
203+ , cid `elem` uninitConns
204+ ) of
205+ (Just proc, _, _, _) -> do
198206 let msg = payloadToMessage payload
199207 enqueue (processQueue proc) msg
200− go uninitConns procConns ctrlConns
201− Nothing | cid `Set.member` ctrlConns -> do
208+ go uninitConns procs chans ctrls
209+ (_, Just (TypedChannel queue), _, _) -> do
210+ enqueue queue . decode . BSL.fromChunks $ payload
211+ go uninitConns procs chans ctrls
212+ (_, _, True, _) -> do
202213 writeChan ctrlChan (decode . BSL.fromChunks $ payload)
203− go uninitConns procConns ctrlConns
204− Nothing | cid `elem` uninitConns ->
205− case payloadToId payload of
206− Just lpid -> do
214+ go uninitConns procs chans ctrls
215+ (_, _, _, True) ->
216+ case decode (BSL.fromChunks payload) of
217+ ProcessIdentifier pid -> do
218+ let lpid = processLocalId pid
207219 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
208220 case mProc of
209221 Just proc ->
210222 go (List.delete cid uninitConns)
211− (Map.insert cid proc procConns)
212− ctrlConns
223+ (Map.insert cid proc procs)
224+ chans
225+ ctrls
213226 Nothing ->
214227 -- Request for an unknown process.
215228 -- TODO: We should close the incoming connection here, but
@@ -217,22 +230,27 @@ handleIncomingMessages node = go [] Map.empty Set.empty
217230 -- functionality. Not sure what the right approach is.
218231 -- For now, we just drop the incoming messages
219232 go (List.delete cid uninitConns)
220− procConns
221− ctrlConns
222− Nothing ->
233+ procs
234+ chans
235+ ctrls
236+ ChannelIdentifier _chId ->
237+ fail "Connecting to typed channels not yet implemented"
238+ NodeIdentifier _ ->
223239 go (List.delete cid uninitConns)
224− procConns
225− (Set.insert cid ctrlConns)
226− _ ->
240+ procs
241+ chans
242+ (Set.insert cid ctrls)
243+ _ ->
227244 -- Unexpected message. We just drop it.
228− go uninitConns procConns ctrlConns
245+ go uninitConns procs chans ctrls
229246 NT.ConnectionClosed cid ->
230247 go (List.delete cid uninitConns)
231− (Map.delete cid procConns)
232− (Set.delete cid ctrlConns)
248+ (Map.delete cid procs)
249+ (Map.delete cid chans)
250+ (Set.delete cid ctrls)
233251 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do
234252 -- [Unified table 9, rule node_disconnect]
235− let nid = Right $ NodeId theirAddr
253+ let nid = NodeIdentifier $ NodeId theirAddr
236254 writeChan ctrlChan NCMsg
237255 { ctrlMsgSender = nid
238256 , ctrlMsgSignal = Died nid DiedDisconnect
@@ -291,23 +309,25 @@ nodeController = do
291309
292310 -- [Unified: Table 7, rule nc_forward]
293311 case destNid (ctrlMsgSignal msg) of
294− Just nid' | nid' /= localNodeId node -> lift $ sendBinary (Right nid') msg
295− _ -> return ()
312+ Just nid' | nid' /= localNodeId node ->
313+ lift $ sendBinary (NodeIdentifier nid') msg
314+ _ ->
315+ return ()
296316
297317 case msg of
298− NCMsg (Left from) (Link them) ->
318+ NCMsg (ProcessIdentifier from) (Link them) ->
299319 ncEffectMonitor from them Nothing
300− NCMsg (Left from) (Monitor them ref) ->
320+ NCMsg (ProcessIdentifier from) (Monitor them ref) ->
301321 ncEffectMonitor from them (Just ref)
302− NCMsg (Left from) (Unlink them) ->
322+ NCMsg (ProcessIdentifier from) (Unlink them) ->
303323 ncEffectUnlink from them
304− NCMsg (Left from) (Unmonitor ref) ->
324+ NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
305325 ncEffectUnmonitor from ref
306− NCMsg _from (Died (Right nid) reason) ->
326+ NCMsg _from (Died (NodeIdentifier nid) reason) ->
307327 ncEffectNodeDied nid reason
308− NCMsg _from (Died (Left pid) reason) ->
328+ NCMsg _from (Died (ProcessIdentifier pid) reason) ->
309329 ncEffectProcessDied pid reason
310− NCMsg (Left from) (Spawn proc ref) ->
330+ NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
311331 ncEffectSpawn from proc ref
312332 unexpected ->
313333 error $ "Unexpected message " ++ show unexpected
@@ -332,9 +352,9 @@ ncEffectMonitor from them mRef = do
332352 (False, True) -> -- [Unified: second rule]
333353 notifyDied from them DiedNoProc mRef
334354 (False, False) -> -- [Unified: third rule]
335− lift $ sendBinary (Right $ processNodeId from) NCMsg
336− { ctrlMsgSender = Right (localNodeId node)
337− , ctrlMsgSignal = Died (Left them) DiedNoProc
355+ lift $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg
356+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
357+ , ctrlMsgSignal = Died (ProcessIdentifier them) DiedNoProc
338358 }
339359
340360-- [Unified: Table 11]
@@ -398,7 +418,7 @@ ncEffectSpawn pid cProc ref = do
398418 forM_ mProc $ \proc -> do
399419 node <- lift getLocalNode
400420 pid' <- liftIO $ forkProcess node proc
401− lift $ sendMessage (Left pid) (DidSpawn ref pid')
421+ lift $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')
402422
403423--------------------------------------------------------------------------------
404424-- Auxiliary --
@@ -419,20 +439,29 @@ notifyDied dest src reason mRef = do
419439 (False, _) ->
420440 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
421441 -- (see [Unified: Table 10]
422− lift $ sendBinary (Right $ processNodeId dest) NCMsg
423− { ctrlMsgSender = Right (localNodeId node)
424− , ctrlMsgSignal = Died (Left src) reason
442+ lift $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg
443+ { ctrlMsgSender = NodeIdentifier (localNodeId node)
444+ , ctrlMsgSignal = Died (ProcessIdentifier src) reason
425445 }
426446
427447-- | [Unified: Table 8]
428448destNid :: ProcessSignal -> Maybe NodeId
429−destNid (Link pid) = Just . processNodeId $ pid
430−destNid (Unlink pid) = Just . processNodeId $ pid
431−destNid (Monitor pid _) = Just . processNodeId $ pid
432−destNid (Unmonitor ref) = Just . processNodeId . monitorRefPid $ ref
433−destNid (Died (Right _) _) = Nothing
434−destNid (Died (Left _pid) _) = fail "destNid: TODO"
435−destNid (Spawn _ _) = Nothing
449+destNid (Link pid) =
450+ Just . processNodeId $ pid
451+destNid (Unlink pid) =
452+ Just . processNodeId $ pid
453+destNid (Monitor pid _) =
454+ Just . processNodeId $ pid
455+destNid (Unmonitor ref) =
456+ Just . processNodeId . monitorRefPid $ ref
457+destNid (Died (NodeIdentifier _) _) =
458+ Nothing
459+destNid (Died (ProcessIdentifier _pid) _) =
460+ fail "destNid: TODO"
461+destNid (Died (ChannelIdentifier _cid) _) =
462+ fail "destNid: TODO"
463+destNid (Spawn _ _) =
464+ Nothing
436465
437466-- | Check if a process is local to our own node
438467isLocal :: LocalNode -> ProcessId -> Bool