@@ -25,6 +25,7 @@ import qualified Data.Map as Map
27 , fromList
27 , fromList
28 , partitionWithKey
29 , partitionWithKey
30 , filterWithKey
31 , filterWithKey
@@ -440,7 +441,8 @@ data NCState = NCState
440 -- Mapping from remote processes to monitoring local processes
441 -- Mapping from remote processes to monitoring local processes
441 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
442 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
442 -- Process registry: names and where they live, mapped to the PIDs
443 -- Process registry: names and where they live, mapped to the PIDs
443− , _registry :: !(Map (String,NodeId) ProcessId)
444+ , _registeredHere :: !(Map String ProcessId)
445+ , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
446newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
448newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
@@ -449,7 +451,8 @@ newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
449initNCState :: NCState
451initNCState :: NCState
450initNCState = NCState { _links = Map.empty
452initNCState = NCState { _links = Map.empty
451 , _monitors = Map.empty
453 , _monitors = Map.empty
452− , _registry = Map.empty
454+ , _registeredHere = Map.empty
455+ , _registeredOnNodes = Map.empty
455--------------------------------------------------------------------------------
458--------------------------------------------------------------------------------
@@ -572,11 +575,20 @@ ncEffectDied ident reason = do
573 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
576 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
578+ modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
575− remaining <- fmap Map.toList (gets (^. registry)) >>= mapM ( \whl@((_,nid),pid) ->
580+ remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
581+ mapM (\(pid,nidlist) ->
576− case ident `impliesDeathOf` ProcessIdentifier pid ||
582+ case ident `impliesDeathOf` ProcessIdentifier pid of
577− ident `impliesDeathOf` NodeIdentifier nid of
578− True ->
583+ True ->
584+ do forM_ nidlist $ \(nid,_) ->
579− do when (not $ isLocal node (NodeIdentifier nid)) $
585+ when (not $ isLocal node (NodeIdentifier nid))
586+ (forwardNameDeath node nid)
588+ False -> return $ Just (pid,nidlist) )
589+ modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
591+ forwardNameDeath node nid =
580 liftIO $ sendBinary node
592 liftIO $ sendBinary node
581 (NodeIdentifier $ localNodeId node)
593 (NodeIdentifier $ localNodeId node)
582 (NodeIdentifier $ nid)
594 (NodeIdentifier $ nid)
@@ -585,9 +597,7 @@ ncEffectDied ident reason = do
585 { ctrlMsgSender = NodeIdentifier (localNodeId node)
597 { ctrlMsgSender = NodeIdentifier (localNodeId node)
586 , ctrlMsgSignal = Died ident reason
598 , ctrlMsgSignal = Died ident reason
589− False -> return $ Just whl)
590− modify' $ registry ^= Map.fromList (catMaybes remaining)
592-- [Unified: Table 13]
602-- [Unified: Table 13]
593ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
603ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
@@ -612,7 +622,7 @@ ncEffectSpawn pid cProc ref = do
612ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
622ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
613ncEffectRegister from label atnode mPid reregistration = do
623ncEffectRegister from label atnode mPid reregistration = do
614 node <- ask
624 node <- ask
615− currentVal <- gets (^. registryFor (label, localNodeId node))
625+ currentVal <- gets (^. registeredHereFor label)
617 case mPid of
627 case mPid of
618 Nothing -> -- unregister request
628 Nothing -> -- unregister request
@@ -623,30 +633,56 @@ ncEffectRegister from label atnode mPid reregistration = do
623 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
633 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
624 if isLocal node (NodeIdentifier atnode)
634 if isLocal node (NodeIdentifier atnode)
625 then when (isOk) $
635 then when (isOk) $
626− do modify' $ registryFor (label, atnode) ^= mPid
636+ do modify' $ registeredHereFor label ^= mPid
627− let namedPid =
637+ updateRemote node currentVal mPid
628− head $ catMaybes [mPid, currentVal]
638+ liftIO $ sendMessage node
639+ (NodeIdentifier (localNodeId node))
640+ (ProcessIdentifier from)
641+ WithImplicitReconnect
642+ (RegisterReply label isOk)
643+ else let operation =
644+ case reregistration of
645+ True -> flip decList
646+ False -> flip incList
648+ Nothing -> return ()
649+ Just pid -> modify' $ registeredOnNodesFor pid ^: (maybeify $ operation atnode)
650+ where updateRemote node (Just oldval) (Just newval) | processNodeId oldval /= processNodeId newval =
651+ do forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
652+ forward node (processNodeId newval) (Register label atnode (Just newval) False)
653+ updateRemote node Nothing (Just newval) =
654+ forward node (processNodeId newval) (Register label atnode (Just newval) False)
655+ updateRemote node (Just oldval) Nothing =
656+ forward node (processNodeId oldval) (Register label atnode (Just oldval) True)
657+ updateRemote _ _ _ = return ()
658+ maybeify f Nothing = unmaybeify $ f []
659+ maybeify f (Just x) = unmaybeify $ f x
660+ unmaybeify [] = Nothing
661+ unmaybeify x = Just x
662+ incList [] tag = [(tag,1)]
663+ incList ((atag,acount):xs) tag | tag==atag = (atag,acount+1) : xs
664+ incList (x:xs) tag = x : incList xs tag
666+ decList ((atag,1):xs) tag | atag == tag = xs
667+ decList ((atag,n):xs) tag | atag == tag = (atag,n-1):xs
668+ decList (x:xs) tag = x:decList xs tag
669+ forward node to reg =
629− when (not $ isLocal node (ProcessIdentifier namedPid)) $
670+ when (not $ isLocal node (NodeIdentifier to)) $
630 liftIO $ sendBinary node
671 liftIO $ sendBinary node
631 (ProcessIdentifier from)
672 (ProcessIdentifier from)
632− (NodeIdentifier (processNodeId namedPid))
673+ (NodeIdentifier to)
633 WithImplicitReconnect
674 WithImplicitReconnect
635 { ctrlMsgSender = ProcessIdentifier from
676 { ctrlMsgSender = ProcessIdentifier from
636− , ctrlMsgSignal = Register label atnode mPid reregistration
677+ , ctrlMsgSignal = reg
638− else modify' $ registryFor (label,atnode) ^= mPid
679+
639− liftIO $ sendMessage node
640− (NodeIdentifier (localNodeId node))
641− (ProcessIdentifier from)
642− WithImplicitReconnect
643− (RegisterReply label isOk)
645-- Unified semantics does not explicitly describe 'whereis'
681-- Unified semantics does not explicitly describe 'whereis'
646ncEffectWhereIs :: ProcessId -> String -> NC ()
682ncEffectWhereIs :: ProcessId -> String -> NC ()
647ncEffectWhereIs from label = do
683ncEffectWhereIs from label = do
648 node <- ask
684 node <- ask
649− mPid <- gets (^. registryFor (label, localNodeId node))
685+ mPid <- gets (^. registeredHereFor label)
650 liftIO $ sendMessage node
686 liftIO $ sendMessage node
651 (NodeIdentifier (localNodeId node))
687 (NodeIdentifier (localNodeId node))
652 (ProcessIdentifier from)
688 (ProcessIdentifier from)
@@ -657,7 +693,7 @@ ncEffectWhereIs from label = do
657ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
693ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
658ncEffectNamedSend from label msg = do
694ncEffectNamedSend from label msg = do
659 node <- ask
695 node <- ask
660− mPid <- gets (^. registryFor (label, localNodeId node))
696+ mPid <- gets (^. registeredHereFor label)
661 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
697 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
662 forM_ mPid $ \pid ->
698 forM_ mPid $ \pid ->
663 liftIO $ sendPayload node
699 liftIO $ sendPayload node
@@ -780,8 +816,11 @@ links = accessor _links (\ls st -> st { _links = ls })
780monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
816monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
781monitors = accessor _monitors (\ms st -> st { _monitors = ms })
817monitors = accessor _monitors (\ms st -> st { _monitors = ms })
783−registry :: Accessor NCState (Map (String,NodeId) ProcessId)
819+registeredHere :: Accessor NCState (Map String ProcessId)
784−registry = accessor _registry (\ry st -> st { _registry = ry })
820+registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
822+registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
823+registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
786linksFor :: Identifier -> Accessor NCState (Set ProcessId)
825linksFor :: Identifier -> Accessor NCState (Set ProcessId)
787linksFor ident = links >>> DAC.mapDefault Set.empty ident
826linksFor ident = links >>> DAC.mapDefault Set.empty ident
@@ -789,8 +828,11 @@ linksFor ident = links >>> DAC.mapDefault Set.empty ident
789monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
828monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
790monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
829monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
792−registryFor :: (String, NodeId) -> Accessor NCState (Maybe ProcessId)
831+registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
832+registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
834+registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
793−registryFor ident = registry >>> DAC.mapMaybe ident
835+registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
795-- | @splitNotif ident@ splits a notifications map into those
837-- | @splitNotif ident@ splits a notifications map into those
796-- notifications that should trigger when 'ident' fails and those links that
838-- notifications that should trigger when 'ident' fails and those links that