Fixed de-registration of remote processes when the process terminates

Added reregister and reregisterRemoteAsync (keeping in line with Erlang's approach)
Changed SimpleLocalNet to use reregisterRemoteAsync to adjust logging (currently without waiting for result)

Author: jepst

distributed-process-simplelocalnet/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs

@@ -115,7 +115,7 @@ import Control.Distributed.Process
   , WhereIsReply(..)
   , whereis
   , whereisRemoteAsync
-  , registerRemote
+  , reregisterRemoteAsync
   , getSelfPid
   , register
   , expect
@@ -244,7 +244,7 @@ apiRedirectLogsHere backend = do
   mLogger <- whereis "logger"
   forM_ mLogger $ \logger -> do
     nids <- liftIO $ findPeers backend 1000000 
-    forM_ nids $ \nid -> registerRemote nid "logger" logger
+    forM_ nids $ \nid -> reregisterRemoteAsync nid "logger" logger -- ignore async response
 
 --------------------------------------------------------------------------------
 -- Slaves                                                                     --

distributed-process/src/Control/Distributed/Process.hs

@@ -44,6 +44,7 @@ module Control.Distributed.Process
   , call
   , terminate
   , ProcessTerminationException(..)
+  , ProcessRegistrationException(..)
   , SpawnRef
   , getSelfPid
   , getSelfNode
@@ -77,10 +78,12 @@ module Control.Distributed.Process
   , say
     -- * Registry
   , register
+  , reregister
   , unregister
   , whereis
   , nsend
   , registerRemoteAsync
+  , reregisterRemoteAsync
   , unregisterRemoteAsync
   , whereisRemoteAsync
   , nsendRemote
@@ -135,6 +138,7 @@ import Control.Distributed.Process.Internal.Types
   , ProcessLinkException(..)
   , NodeLinkException(..)
   , PortLinkException(..)
+  , ProcessRegistrationException(..)
   , DiedReason(..)
   , SpawnRef(..)
   , DidSpawn(..)
@@ -197,10 +201,12 @@ import Control.Distributed.Process.Internal.Primitives
   , say
     -- Registry
   , register
+  , reregister
   , unregister
   , whereis
   , nsend
   , registerRemoteAsync
+  , reregisterRemoteAsync
   , unregisterRemoteAsync
   , whereisRemoteAsync
   , nsendRemote

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs

@@ -35,10 +35,12 @@ module Control.Distributed.Process.Internal.Primitives
   , say
     -- * Registry
   , register
+  , reregister
   , unregister
   , whereis
   , nsend
   , registerRemoteAsync
+  , reregisterRemoteAsync
   , unregisterRemoteAsync
   , whereisRemoteAsync
   , nsendRemote
@@ -542,11 +544,23 @@ say string = do
 --------------------------------------------------------------------------------
 
 -- | Register a process with the local registry (asynchronous).
+-- This version will wait until a response is gotten from the
+-- management process. The name must not already be registered.
+-- The process need not be on this node.
+-- A bad registration will result in a 'ProcessRegistrationException'
 --
 -- The process to be registered does not have to be local itself.
 register :: String -> ProcessId -> Process ()
-register label pid = do
-  sendCtrlMsg Nothing (Register label (Just pid))
+register = registerImpl False
+
+-- | Like 'register', but will replace an existing registration.
+-- The name must already be registered.
+reregister :: String -> ProcessId -> Process ()
+reregister = registerImpl True
+
+registerImpl :: Bool -> String -> ProcessId -> Process ()
+registerImpl force label pid = do
+  sendCtrlMsg Nothing (Register label (Just pid) force)
   receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
                         (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
               ]
@@ -559,12 +573,18 @@ register label pid = do
 -- See comments in 'whereisRemoteAsync'
 registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
 registerRemoteAsync nid label pid = 
-  sendCtrlMsg (Just nid) (Register label (Just pid)) 
+  sendCtrlMsg (Just nid) (Register label (Just pid) False) 
+
+reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()
+reregisterRemoteAsync nid label pid = 
+  sendCtrlMsg (Just nid) (Register label (Just pid) True) 
 
 -- | Remove a process from the local registry (asynchronous).
+-- This version will wait until a response is gotten from the
+-- management process. The name must already be registered.
 unregister :: String -> Process ()
 unregister label = do
-  sendCtrlMsg Nothing (Register label Nothing)
+  sendCtrlMsg Nothing (Register label Nothing False)
   receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')
                         (\(RegisterReply _ ok) -> handleRegistrationReply label ok)
               ]
@@ -583,7 +603,7 @@ handleRegistrationReply label ok =
 -- See comments in 'whereisRemoteAsync'
 unregisterRemoteAsync :: NodeId -> String -> Process ()
 unregisterRemoteAsync nid label =
-  sendCtrlMsg (Just nid) (Register label Nothing)
+  sendCtrlMsg (Just nid) (Register label Nothing False)
 
 -- | Query the local process registry
 whereis :: String -> Process (Maybe ProcessId)

distributed-process/src/Control/Distributed/Process/Internal/Types.hs

@@ -440,7 +440,7 @@ data ProcessSignal =
   | Died Identifier !DiedReason
   | Spawn !(Closure (Process ())) !SpawnRef 
   | WhereIs !String
-  | Register !String !(Maybe ProcessId) -- Use 'Nothing' to unregister
+  | Register !String !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister
   | NamedSend !String !Message
   deriving Show
 
@@ -484,7 +484,7 @@ instance Binary ProcessSignal where
   put (Died who reason)     = putWord8 4 >> put who >> put reason
   put (Spawn proc ref)      = putWord8 5 >> put proc >> put ref
   put (WhereIs label)       = putWord8 6 >> put label
-  put (Register label pid)  = putWord8 7 >> put label >> put pid
+  put (Register label pid force)= putWord8 7 >> put label >> put pid >> put force
   put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg) 
   get = do
     header <- getWord8
@@ -496,7 +496,7 @@ instance Binary ProcessSignal where
       4 -> Died <$> get <*> get
       5 -> Spawn <$> get <*> get
       6 -> WhereIs <$> get
-      7 -> Register <$> get <*> get
+      7 -> Register <$> get <*> get <*> get
       8 -> NamedSend <$> get <*> (payloadToMessage <$> get)
       _ -> fail "ProcessSignal.get: invalid"
 

distributed-process/src/Control/Distributed/Process/Node.hs

@@ -24,14 +24,15 @@ import Data.Map (Map)
 import qualified Data.Map as Map 
   ( empty
   , toList
+  , fromList
   , partitionWithKey
   , elems
   , filterWithKey
   )
 import Data.Set (Set)
 import qualified Data.Set as Set (empty, insert, delete, member)
 import Data.Foldable (forM_)
-import Data.Maybe (isJust, isNothing)
+import Data.Maybe (isJust, isNothing, catMaybes)
 import Data.Typeable (Typeable)
 import Control.Category ((>>>))
 import Control.Applicative ((<$>))
@@ -438,8 +439,8 @@ data NCState = NCState
     _links    :: !(Map Identifier (Set ProcessId))
      -- Mapping from remote processes to monitoring local processes
   , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))
-     -- Process registry
-  , _registry :: !(Map String ProcessId)
+     -- Process registry: names and where they live, mapped to the PIDs
+  , _registry :: !(Map (String,NodeId) ProcessId)
   }
 
 newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
@@ -486,8 +487,8 @@ nodeController = do
         ncEffectDied ident reason
       NCMsg (ProcessIdentifier from) (Spawn proc ref) ->
         ncEffectSpawn from proc ref
-      NCMsg (ProcessIdentifier from) (Register label pid) ->
-        ncEffectRegister from label pid
+      NCMsg (ProcessIdentifier from) (Register label pid force) ->
+        ncEffectRegister from label pid force
       NCMsg (ProcessIdentifier from) (WhereIs label) ->
         ncEffectWhereIs from label
       NCMsg from (NamedSend label msg') ->
@@ -555,6 +556,8 @@ ncEffectDied ident reason = do
   (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))
   (affectedMons,  unaffectedMons)  <- gets (splitNotif ident . (^. monitors))
 
+--  _registry :: !(Map (String,NodeId) ProcessId)
+
   let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False
 
   forM_ (Map.toList affectedLinks) $ \(them, uss) -> 
@@ -565,10 +568,27 @@ ncEffectDied ident reason = do
   forM_ (Map.toList affectedMons) $ \(them, refs) ->
     forM_ refs $ \(us, ref) ->
       when (localOnly <= isLocal node (ProcessIdentifier us)) $
-        notifyDied us them reason (Just ref)
+        notifyDied us them reason (Just ref)   
 
   modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)
 
+  remaining <- fmap Map.toList (gets (^. registry)) >>= mapM ( \whl@((_,nid),pid) ->
+     case ident `impliesDeathOf` ProcessIdentifier pid ||
+          ident `impliesDeathOf` NodeIdentifier nid of
+       True -> 
+         do when (not $ isLocal node (NodeIdentifier nid)) $
+                   liftIO $ sendBinary node
+                             (NodeIdentifier $ localNodeId node)
+                             (NodeIdentifier $ nid)
+                             WithImplicitReconnect
+                             NCMsg
+                             { ctrlMsgSender = NodeIdentifier (localNodeId node) 
+                             , ctrlMsgSignal = Died ident reason
+                             }              
+            return Nothing
+       False -> return $ Just whl)
+  modify' $ registry ^= Map.fromList (catMaybes remaining)
+
 -- [Unified: Table 13]
 ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()
 ncEffectSpawn pid cProc ref = do
@@ -589,18 +609,33 @@ ncEffectSpawn pid cProc ref = do
 -- Unified semantics does not explicitly describe how to implement 'register',
 -- but mentions it's "very similar to nsend" (Table 14)
 -- We send a response indicated if the operation is invalid
-ncEffectRegister :: ProcessId -> String -> Maybe ProcessId -> NC ()
-ncEffectRegister from label mPid = do
+ncEffectRegister :: ProcessId -> String -> Maybe ProcessId -> Bool -> NC ()
+ncEffectRegister from label mPid reregistration = do
   node <- ask
-  currentVal <- gets (^. registryFor label)
-  let isOk = 
+  currentVal <- gets (^. registryFor (label, localNodeId node))
+  isOk <-
        case mPid of
          Nothing -> -- unregister request
-           isJust currentVal
-         Just _ -> -- register request
-           isNothing currentVal
-  when (isOk) $
-     modify' $ registryFor label ^= mPid
+           return $ isJust currentVal
+         Just thepid -> -- register request
+           do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)
+              return $ (isNothing currentVal /= reregistration) && 
+                (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )
+  if isLocal node (ProcessIdentifier from)
+     then when (isOk) $
+              do modify' $ registryFor (label, localNodeId node) ^= mPid
+                 let namedPid = 
+                       head $ catMaybes [mPid, currentVal]
+                 when (not $ isLocal node (ProcessIdentifier namedPid)) $
+                    liftIO $ sendBinary node
+                                        (ProcessIdentifier from)
+                                        (NodeIdentifier (processNodeId namedPid))
+                                        WithImplicitReconnect
+                                        NCMsg
+                                         { ctrlMsgSender = ProcessIdentifier from
+                                         , ctrlMsgSignal = Register label mPid reregistration
+                                         }
+     else modify' $ registryFor (label,processNodeId from) ^= mPid
   liftIO $ sendMessage node
                        (NodeIdentifier (localNodeId node))
                        (ProcessIdentifier from) 
@@ -611,7 +646,7 @@ ncEffectRegister from label mPid = do
 ncEffectWhereIs :: ProcessId -> String -> NC ()
 ncEffectWhereIs from label = do
   node <- ask
-  mPid <- gets (^. registryFor label)
+  mPid <- gets (^. registryFor (label, localNodeId node))
   liftIO $ sendMessage node
                        (NodeIdentifier (localNodeId node))
                        (ProcessIdentifier from) 
@@ -621,9 +656,9 @@ ncEffectWhereIs from label = do
 -- [Unified: Table 14]
 ncEffectNamedSend :: Identifier -> String -> Message -> NC ()
 ncEffectNamedSend from label msg = do
-  mPid <- gets (^. registryFor label)
+  node <- ask  
+  mPid <- gets (^. registryFor (label, localNodeId node))
   -- If mPid is Nothing, we just ignore the named send (as per Table 14)
-  node <- ask
   forM_ mPid $ \pid -> 
     liftIO $ sendPayload node
                          from
@@ -673,7 +708,7 @@ destNid (Unlink ident)  = Just $ nodeOf ident
 destNid (Monitor ref)   = Just $ nodeOf (monitorRefIdent ref)
 destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)
 destNid (Spawn _ _)     = Nothing 
-destNid (Register _ _)  = Nothing
+destNid (Register _ _ _)= Nothing
 destNid (WhereIs _)     = Nothing
 destNid (NamedSend _ _) = Nothing
 -- We don't need to forward 'Died' signals; if monitoring/linking is setup,
@@ -745,7 +780,7 @@ links = accessor _links (\ls st -> st { _links = ls })
 monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))
 monitors = accessor _monitors (\ms st -> st { _monitors = ms })
 
-registry :: Accessor NCState (Map String ProcessId)
+registry :: Accessor NCState (Map (String,NodeId) ProcessId)
 registry = accessor _registry (\ry st -> st { _registry = ry })
 
 linksFor :: Identifier -> Accessor NCState (Set ProcessId)
@@ -754,7 +789,7 @@ linksFor ident = links >>> DAC.mapDefault Set.empty ident
 monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))
 monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident
 
-registryFor :: String -> Accessor NCState (Maybe ProcessId)
+registryFor :: (String, NodeId) -> Accessor NCState (Maybe ProcessId)
 registryFor ident = registry >>> DAC.mapMaybe ident
 
 -- | @splitNotif ident@ splits a notifications map into those