@@ -13,7 +13,14 @@ import System.IO (fixIO)
1313import qualified Data.ByteString.Lazy as BSL (fromChunks)
1414import Data.Binary (decode)
1515import Data.Map (Map)
16−import qualified Data.Map as Map (empty, lookup, insert, delete, toList)
16+import qualified Data.Map as Map
17+ ( empty
18+ , lookup
19+ , insert
20+ , delete
21+ , toList
22+ , partitionWithKey
23+ )
1724import qualified Data.List as List (delete)
1825import Data.Set (Set)
1926import qualified Data.Set as Set (empty, insert, delete, member)
@@ -73,18 +80,26 @@ import Control.Distributed.Process.Internal.Types
7380 , localPidUnique
7481 , localProcessWithId
7582 , MonitorRef(..)
76− , MonitorNotification(..)
77− , LinkException(..)
83+ , ProcessMonitorNotification(..)
84+ , NodeMonitorNotification(..)
85+ , PortMonitorNotification(..)
86+ , ProcessLinkException(..)
87+ , NodeLinkException(..)
88+ , PortLinkException(..)
7889 , DidUnmonitor(..)
79− , DidUnlink(..)
90+ , DidUnlinkProcess(..)
91+ , DidUnlinkNode(..)
92+ , DidUnlinkPort(..)
8093 , SpawnRef
8194 , DidSpawn(..)
8295 , Closure(..)
8396 , Static(..)
97+ , Message
8498 , MessageT
8599 , TypedChannel(..)
86100 , Identifier(..)
87− , ChannelId(..)
101+ , nodeOf
102+ , SendPortId(..)
88103 , typedChannelWithId
89104 )
90105import Control.Distributed.Process.Serializable (Serializable)
@@ -249,9 +264,9 @@ handleIncomingMessages node = go [] Map.empty Map.empty Set.empty
249264 -- incoming connections (this requires a Transport layer
250265 -- extension). (**)
251266 go (List.delete cid uninitConns) procs chans ctrls
252− ChannelIdentifier chId -> do
253− let lcid = channelLocalId chId
254− lpid = processLocalId (channelProcessId chId)
267+ SendPortIdentifier chId -> do
268+ let lcid = sendPortLocalId chId
269+ lpid = processLocalId (sendPortProcessId chId)
255270 mProc <- withMVar state $ return . (^. localProcessWithId lpid)
256271 case mProc of
257272 Just proc -> do
@@ -325,9 +340,9 @@ runNodeController node =
325340
326341data NCState = NCState
327342 { -- Mapping from remote processes to linked local processes
328− _links :: Map NodeId (Map LocalProcessId (Set ProcessId))
343+ _links :: Map Identifier (Set ProcessId)
329344 -- Mapping from remote processes to monitoring local processes
330− , _monitors :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
345+ , _monitors :: Map Identifier (Set (ProcessId, MonitorRef))
331346 }
332347
333348newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }
@@ -362,24 +377,22 @@ nodeController = do
362377 case msg of
363378 NCMsg (ProcessIdentifier from) (Link them) ->
364379 ncEffectMonitor from them Nothing
365− NCMsg (ProcessIdentifier from) (Monitor them ref) ->
366− ncEffectMonitor from them (Just ref)
380+ NCMsg (ProcessIdentifier from) (Monitor ref) ->
381+ ncEffectMonitor from (monitorRefIdent ref) (Just ref)
367382 NCMsg (ProcessIdentifier from) (Unlink them) ->
368383 ncEffectUnlink from them
369384 NCMsg (ProcessIdentifier from) (Unmonitor ref) ->
370385 ncEffectUnmonitor from ref
371− NCMsg _from (Died (NodeIdentifier nid) reason) ->
372− ncEffectNodeDied nid reason
373− NCMsg _from (Died (ProcessIdentifier pid) reason) ->
374− ncEffectProcessDied pid reason
386+ NCMsg _from (Died ident reason) ->
387+ ncEffectDied ident reason
375388 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
376389 ncEffectSpawn from proc ref
377390 unexpected ->
378− error $ "Unexpected message " ++ show unexpected
391+ error $ "nodeController: unexpected message " ++ show unexpected
379392
380393-- [Unified: Table 10]
381394ncEffectMonitor :: ProcessId -- ^ Who's watching?
382− -> ProcessId -- ^ Who's being watched?
395+ -> Identifier -- ^ Who's being watched?
383396 -> Maybe MonitorRef -- ^ 'Nothing' to link
384397 -> NC ()
385398ncEffectMonitor from them mRef = do
@@ -387,73 +400,62 @@ ncEffectMonitor from them mRef = do
387400 shouldLink <-
388401 if not (isLocal node them)
389402 then return True
390− else liftIO . withMVar (localState node) $
391− return . isJust . (^. localProcessWithId (processLocalId them))
392− case (shouldLink, isLocal node from) of
403+ else isValidLocalIdentifier them
404+ case (shouldLink, isLocal node (ProcessIdentifier from)) of
393405 (True, _) -> -- [Unified: first rule]
394406 case mRef of
395− Just ref -> modify $ monitorsFor them from ^: Set.insert ref
396− Nothing -> modify $ linksForProcess them ^: Set.insert from
407+ Just ref -> modify $ monitorsFor them ^: Set.insert (from, ref)
408+ Nothing -> modify $ linksFor them ^: Set.insert from
397409 (False, True) -> -- [Unified: second rule]
398− notifyDied from them DiedNoProc mRef
410+ notifyDied from them DiedUnknownId mRef
399411 (False, False) -> -- [Unified: third rule]
400412 ncMsg $ sendBinary (NodeIdentifier $ processNodeId from) NCMsg
401413 { ctrlMsgSender = NodeIdentifier (localNodeId node)
402− , ctrlMsgSignal = Died (ProcessIdentifier them) DiedNoProc
414+ , ctrlMsgSignal = Died them DiedUnknownId
403415 }
404416
405417-- [Unified: Table 11]
406−ncEffectUnlink :: ProcessId -> ProcessId -> NC ()
418+ncEffectUnlink :: ProcessId -> Identifier -> NC ()
407419ncEffectUnlink from them = do
408420 node <- ncMsg getLocalNode
409− when (isLocal node from) $ postMessage from $ DidUnlink them
410− modify $ linksForProcess them ^: Set.delete from
421+ when (isLocal node (ProcessIdentifier from)) $
422+ case them of
423+ ProcessIdentifier pid ->
424+ postAsMessage from $ DidUnlinkProcess pid
425+ NodeIdentifier nid ->
426+ postAsMessage from $ DidUnlinkNode nid
427+ SendPortIdentifier cid ->
428+ postAsMessage from $ DidUnlinkPort cid
429+ modify $ linksFor them ^: Set.delete from
411430
412431-- [Unified: Table 11]
413432ncEffectUnmonitor :: ProcessId -> MonitorRef -> NC ()
414433ncEffectUnmonitor from ref = do
415434 node <- ncMsg getLocalNode
416− when (isLocal node from) $ postMessage from $ DidUnmonitor ref
417− modify $ monitorsFor (monitorRefPid ref) from ^: Set.delete ref
418−
419−-- [Unified: Table 12, bottom rule]
420−ncEffectNodeDied :: NodeId -> DiedReason -> NC ()
421−ncEffectNodeDied nid reason = do
422− node <- ncMsg getLocalNode
423− lnks <- gets (^. linksForNode nid)
424− mons <- gets (^. monitorsForNode nid)
425−
426− -- We only need to notify local processes
427− forM_ (Map.toList lnks) $ \(them, uss) -> do
428− let pid = ProcessId nid them
429− forM_ uss $ \us ->
430− when (isLocal node us) $
431− notifyDied us pid reason Nothing
435+ when (isLocal node (ProcessIdentifier from)) $
436+ postAsMessage from $ DidUnmonitor ref
437+ modify $ monitorsFor (monitorRefIdent ref) ^: Set.delete (from, ref)
432438
433− forM_ (Map.toList mons) $ \(them, uss) -> do
434− let pid = ProcessId nid them
435− forM_ (Map.toList uss) $ \(us, refs) ->
436− when (isLocal node us) $
437− forM_ refs $ notifyDied us pid reason . Just
438−
439− modify $ (linksForNode nid ^= Map.empty)
440− . (monitorsForNode nid ^= Map.empty)
439+-- [Unified: Table 12]
440+ncEffectDied :: Identifier -> DiedReason -> NC ()
441+ncEffectDied ident reason = do
442+ node <- ncMsg getLocalNode
443+ (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
444+ (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))
441445
442−-- [Unified: Table 12, top rule]
443−ncEffectProcessDied :: ProcessId -> DiedReason -> NC ()
444−ncEffectProcessDied pid reason = do
445− lnks <- gets (^. linksForProcess pid)
446− mons <- gets (^. monitorsForProcess pid)
446+ let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
447447
448− forM_ lnks $ \us ->
449− notifyDied us pid reason Nothing
448+ forM_ (Map.toList affectedLinks) $ \(them, uss) ->
449+ forM_ uss $ \us ->
450+ when (localOnly <= isLocal node (ProcessIdentifier us)) $
451+ notifyDied us them reason Nothing
450452
451− forM_ (Map.toList mons) $ \(us, refs) ->
452− forM_ refs $
453− notifyDied us pid reason . Just
453+ forM_ (Map.toList affectedMons) $ \(them, refs) ->
454+ forM_ refs $ \(us, ref) ->
455+ when (localOnly <= isLocal node (ProcessIdentifier us)) $
456+ notifyDied us them reason (Just ref)
454457
455− modify $ (linksForProcess pid ^= Set.empty)
456− . (monitorsForProcess pid ^= Map.empty)
458+ modify $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
457459
458460-- [Unified: Table 13]
459461ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
@@ -471,46 +473,48 @@ ncEffectSpawn pid cProc ref = do
471473--------------------------------------------------------------------------------
472474
473475notifyDied :: ProcessId -- ^ Who to notify?
474− -> ProcessId -- ^ Who died?
476+ -> Identifier -- ^ Who died?
475477 -> DiedReason -- ^ How did they die?
476478 -> Maybe MonitorRef -- ^ 'Nothing' for linking
477479 -> NC ()
478480notifyDied dest src reason mRef = do
479481 node <- ncMsg getLocalNode
480− case (isLocal node dest, mRef) of
481− (True, Just ref) ->
482− postMessage dest $ MonitorNotification ref src reason
483− (True, Nothing) ->
484− throwException dest $ LinkException src reason
485− (False, _) ->
482+ case (isLocal node (ProcessIdentifier dest), mRef, src) of
483+ (True, Just ref, ProcessIdentifier pid) ->
484+ postAsMessage dest $ ProcessMonitorNotification ref pid reason
485+ (True, Just ref, NodeIdentifier nid) ->
486+ postAsMessage dest $ NodeMonitorNotification ref nid reason
487+ (True, Just ref, SendPortIdentifier cid) ->
488+ postAsMessage dest $ PortMonitorNotification ref cid reason
489+ (True, Nothing, ProcessIdentifier pid) ->
490+ throwException dest $ ProcessLinkException pid reason
491+ (True, Nothing, NodeIdentifier pid) ->
492+ throwException dest $ NodeLinkException pid reason
493+ (True, Nothing, SendPortIdentifier pid) ->
494+ throwException dest $ PortLinkException pid reason
495+ (False, _, _) ->
486496 -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
487497 -- (see [Unified: Table 10]
488498 ncMsg $ sendBinary (NodeIdentifier $ processNodeId dest) NCMsg
489499 { ctrlMsgSender = NodeIdentifier (localNodeId node)
490− , ctrlMsgSignal = Died (ProcessIdentifier src) reason
500+ , ctrlMsgSignal = Died src reason
491501 }
492502
493503-- | [Unified: Table 8]
494504destNid :: ProcessSignal -> Maybe NodeId
495−destNid (Link pid) =
496− Just . processNodeId $ pid
497−destNid (Unlink pid) =
498− Just . processNodeId $ pid
499−destNid (Monitor pid _) =
500− Just . processNodeId $ pid
501−destNid (Unmonitor ref) =
502− Just . processNodeId . monitorRefPid $ ref
505+destNid (Link ident) = Just $ nodeOf ident
506+destNid (Unlink ident) = Just $ nodeOf ident
507+destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)
508+destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
509+destNid (Spawn _ _) = Nothing
503510-- We don't need to forward 'Died' signals; if monitoring/linking is setup,
504511-- then when a local process dies the monitoring/linking machinery will take
505512-- care of notifying remote nodes
506−destNid (Died _ _) =
507− Nothing
508−destNid (Spawn _ _) =
509− Nothing
513+destNid (Died _ _) = Nothing
510514
511515-- | Check if a process is local to our own node
512−isLocal :: LocalNode -> ProcessId -> Bool
513−isLocal nid pid = processNodeId pid == localNodeId nid
516+isLocal :: LocalNode -> Identifier -> Bool
517+isLocal nid ident = nodeOf ident == localNodeId nid
514518
515519-- | Lookup a local closure
516520unClosure :: Typeable a => Closure a -> NC (Maybe a)
@@ -518,13 +522,35 @@ unClosure (Closure (Static label) env) = do
518522 rtable <- remoteTable <$> ncMsg getLocalNode
519523 return (resolveClosure rtable label env >>= fromDynamic)
520524
525+-- | Check if an identifier refers to a valid local object
526+isValidLocalIdentifier :: Identifier -> NC Bool
527+isValidLocalIdentifier ident = do
528+ node <- ncMsg getLocalNode
529+ liftIO . withMVar (localState node) $ \nSt ->
530+ case ident of
531+ NodeIdentifier nid ->
532+ return $ nid == localNodeId node
533+ ProcessIdentifier pid -> do
534+ let mProc = nSt ^. localProcessWithId (processLocalId pid)
535+ return $ isJust mProc
536+ SendPortIdentifier cid -> do
537+ let pid = sendPortProcessId cid
538+ mProc = nSt ^. localProcessWithId (processLocalId pid)
539+ case mProc of
540+ Nothing -> return False
541+ Just proc -> withMVar (processState proc) $ \pSt -> do
542+ let mCh = pSt ^. typedChannelWithId (sendPortLocalId cid)
543+ return $ isJust mCh
544+
521545--------------------------------------------------------------------------------
522546-- Messages to local processes --
523547--------------------------------------------------------------------------------
524548
525−postMessage :: Serializable a => ProcessId -> a -> NC ()
526−postMessage pid msg = withLocalProc pid $ \p ->
527− enqueue (processQueue p) (createMessage msg)
549+postAsMessage :: Serializable a => ProcessId -> a -> NC ()
550+postAsMessage pid = postMessage pid . createMessage
551+
552+postMessage :: ProcessId -> Message -> NC ()
553+postMessage pid msg = withLocalProc pid $ \p -> enqueue (processQueue p) msg
528554
529555throwException :: Exception e => ProcessId -> e -> NC ()
530556throwException pid e = withLocalProc pid $ \p ->
@@ -544,23 +570,57 @@ withLocalProc pid p = do
544570-- Accessors --
545571--------------------------------------------------------------------------------
546572
547−links :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
548−links = accessor _links (\lnks st -> st { _links = lnks })
549−
550−monitors :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
551−monitors = accessor _monitors (\mons st -> st { _monitors = mons })
573+links :: Accessor NCState (Map Identifier (Set ProcessId))
574+links = accessor _links (\ls st -> st { _links = ls })
552575
553−linksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
554−linksForNode nid = links >>> DAC.mapDefault Map.empty nid
576+monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
577+monitors = accessor _monitors (\ms st -> st { _monitors = ms })
555578
556−monitorsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
557−monitorsForNode nid = monitors >>> DAC.mapDefault Map.empty nid
579+linksFor :: Identifier -> Accessor NCState (Set ProcessId)
580+linksFor ident = links >>> DAC.mapDefault Set.empty ident
558581
559−linksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
560−linksForProcess pid = linksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
582+monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
583+monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
561584
562−monitorsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
563−monitorsForProcess pid = monitorsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
564−
565−monitorsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
566−monitorsFor them us = monitorsForProcess them >>> DAC.mapDefault Set.empty us
585+-- | @splitNotif ident@ splits a notifications map into those
586+-- notifications that should trigger when 'ident' fails and those links that
587+-- should not.
588+--
589+-- There is a hierarchy between identifiers: failure of a node implies failure
590+-- of all processes on that node, and failure of a process implies failure of
591+-- all typed channels to that process. In other words, if 'ident' refers to a
592+-- node, then the /should trigger/ set will include
593+--
594+-- * the notifications for the node specifically
595+-- * the notifications for processes on that node, and
596+-- * the notifications for typed channels to processes on that node.
597+--
598+-- Similarly, if 'ident' refers to a process, the /should trigger/ set will
599+-- include
600+--
601+-- * the notifications for that process specifically and
602+-- * the notifications for typed channels to that process.
603+splitNotif :: Identifier
604+ -> Map Identifier a
605+ -> (Map Identifier a, Map Identifier a)
606+splitNotif ident = Map.partitionWithKey (const . impliesDeathOf ident)
607+ where
608+ -- | Does the death of one entity (node, project, channel) imply the death
609+ -- of another?
610+ impliesDeathOf :: Identifier -- ^ Who died
611+ -> Identifier -- ^ Who's being watched
612+ -> Bool -- ^ Does this death implies the death of the watchee?
613+ NodeIdentifier nid `impliesDeathOf` NodeIdentifier nid' =
614+ nid' == nid
615+ NodeIdentifier nid `impliesDeathOf` ProcessIdentifier pid =
616+ processNodeId pid == nid
617+ NodeIdentifier nid `impliesDeathOf` SendPortIdentifier cid =
618+ processNodeId (sendPortProcessId cid) == nid
619+ ProcessIdentifier pid `impliesDeathOf` ProcessIdentifier pid' =
620+ pid' == pid
621+ ProcessIdentifier pid `impliesDeathOf` SendPortIdentifier cid =
622+ sendPortProcessId cid == pid
623+ SendPortIdentifier cid `impliesDeathOf` SendPortIdentifier cid' =
624+ cid' == cid
625+ _ `impliesDeathOf` _ =
626+ False