@@ -25,6 +25,7 @@ import qualified Data.Map as Map
2525 ( empty
2626 , toList
2727 , fromList
28+ , filter
2829 , partitionWithKey
2930 , elems
3031 , filterWithKey
@@ -440,7 +441,8 @@ data NCState = NCState
440441 -- Mapping from remote processes to monitoring local processes
441442 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
442443 -- 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)])
444446 }
445447
446448newtype 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 }
449451initNCState :: NCState
450452initNCState = NCState { _links = Map.empty
451453 , _monitors = Map.empty
452− , _registry = Map.empty
454+ , _registeredHere = Map.empty
455+ , _registeredOnNodes = Map.empty
453456 }
454457
455458--------------------------------------------------------------------------------
@@ -572,11 +575,20 @@ ncEffectDied ident reason = do
572575
573576 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
574577
575− remaining <- fmap Map.toList (gets (^. registry)) >>= mapM ( \whl@((_,nid),pid) ->
576− case ident `impliesDeathOf` ProcessIdentifier pid ||
577− ident `impliesDeathOf` NodeIdentifier nid of
578− True ->
579− do when (not $ isLocal node (NodeIdentifier nid)) $
578+ modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
579+
580+ remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
581+ mapM (\(pid,nidlist) ->
582+ case ident `impliesDeathOf` ProcessIdentifier pid of
583+ True ->
584+ do forM_ nidlist $ \(nid,_) ->
585+ when (not $ isLocal node (NodeIdentifier nid))
586+ (forwardNameDeath node nid)
587+ return Nothing
588+ False -> return $ Just (pid,nidlist) )
589+ modify' $ registeredOnNodes ^= (Map.fromList (catMaybes remaining))
590+ where
591+ forwardNameDeath node nid =
580592 liftIO $ sendBinary node
581593 (NodeIdentifier $ localNodeId node)
582594 (NodeIdentifier $ nid)
@@ -585,9 +597,7 @@ ncEffectDied ident reason = do
585597 { ctrlMsgSender = NodeIdentifier (localNodeId node)
586598 , ctrlMsgSignal = Died ident reason
587599 }
588− return Nothing
589− False -> return $ Just whl)
590− modify' $ registry ^= Map.fromList (catMaybes remaining)
600+
591601
592602-- [Unified: Table 13]
593603ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
@@ -612,7 +622,7 @@ ncEffectSpawn pid cProc ref = do
612622ncEffectRegister :: ProcessId -> String -> NodeId -> Maybe ProcessId -> Bool -> NC ()
613623ncEffectRegister from label atnode mPid reregistration = do
614624 node <- ask
615− currentVal <- gets (^. registryFor (label, localNodeId node))
625+ currentVal <- gets (^. registeredHereFor label)
616626 isOk <-
617627 case mPid of
618628 Nothing -> -- unregister request
@@ -623,30 +633,56 @@ ncEffectRegister from label atnode mPid reregistration = do
623633 (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
624634 if isLocal node (NodeIdentifier atnode)
625635 then when (isOk) $
626− do modify' $ registryFor (label, atnode) ^= mPid
627− let namedPid =
628− head $ catMaybes [mPid, currentVal]
629− when (not $ isLocal node (ProcessIdentifier namedPid)) $
636+ do modify' $ registeredHereFor label ^= mPid
637+ updateRemote node currentVal mPid
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
647+ in case mPid of
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
665+ decList [] _ = []
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 =
670+ when (not $ isLocal node (NodeIdentifier to)) $
630671 liftIO $ sendBinary node
631672 (ProcessIdentifier from)
632− (NodeIdentifier (processNodeId namedPid))
673+ (NodeIdentifier to)
633674 WithImplicitReconnect
634675 NCMsg
635676 { ctrlMsgSender = ProcessIdentifier from
636− , ctrlMsgSignal = Register label atnode mPid reregistration
677+ , ctrlMsgSignal = reg
637678 }
638− else modify' $ registryFor (label,atnode) ^= mPid
639− liftIO $ sendMessage node
640− (NodeIdentifier (localNodeId node))
641− (ProcessIdentifier from)
642− WithImplicitReconnect
643− (RegisterReply label isOk)
679+
644680
645681-- Unified semantics does not explicitly describe 'whereis'
646682ncEffectWhereIs :: ProcessId -> String -> NC ()
647683ncEffectWhereIs from label = do
648684 node <- ask
649− mPid <- gets (^. registryFor (label, localNodeId node))
685+ mPid <- gets (^. registeredHereFor label)
650686 liftIO $ sendMessage node
651687 (NodeIdentifier (localNodeId node))
652688 (ProcessIdentifier from)
@@ -657,7 +693,7 @@ ncEffectWhereIs from label = do
657693ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
658694ncEffectNamedSend from label msg = do
659695 node <- ask
660− mPid <- gets (^. registryFor (label, localNodeId node))
696+ mPid <- gets (^. registeredHereFor label)
661697 -- If mPid is Nothing, we just ignore the named send (as per Table 14)
662698 forM_ mPid $ \pid ->
663699 liftIO $ sendPayload node
@@ -780,8 +816,11 @@ links = accessor _links (\ls st -> st { _links = ls })
780816monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
781817monitors = accessor _monitors (\ms st -> st { _monitors = ms })
782818
783−registry :: Accessor NCState (Map (String,NodeId) ProcessId)
784−registry = accessor _registry (\ry st -> st { _registry = ry })
819+registeredHere :: Accessor NCState (Map String ProcessId)
820+registeredHere = accessor _registeredHere (\ry st -> st { _registeredHere = ry })
821+
822+registeredOnNodes :: Accessor NCState (Map ProcessId [(NodeId, Int)])
823+registeredOnNodes = accessor _registeredOnNodes (\ry st -> st { _registeredOnNodes = ry })
785824
786825linksFor :: Identifier -> Accessor NCState (Set ProcessId)
787826linksFor ident = links >>> DAC.mapDefault Set.empty ident
@@ -789,8 +828,11 @@ linksFor ident = links >>> DAC.mapDefault Set.empty ident
789828monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
790829monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
791830
792−registryFor :: (String, NodeId) -> Accessor NCState (Maybe ProcessId)
793−registryFor ident = registry >>> DAC.mapMaybe ident
831+registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)
832+registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident
833+
834+registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])
835+registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident
794836
795837-- | @splitNotif ident@ splits a notifications map into those
796838-- notifications that should trigger when 'ident' fails and those links that