24163c2 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fixed de-registration of remote processes when the process terminates

24163c2Jeff Epstein14 years ago

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

compare with master

unifiedsplit

@@ -115,7 +115,7 @@ import Control.Distributed.Process

115 , WhereIsReply(..)

116 , whereis

117 , whereisRemoteAsync

118− , registerRemote

118+ , reregisterRemoteAsync

119 , getSelfPid

120 , register

121 , expect

@@ -244,7 +244,7 @@ apiRedirectLogsHere backend = do

244 mLogger <- whereis "logger"

245 forM_ mLogger $ \logger -> do

246 nids <- liftIO $ findPeers backend 1000000

247− forM_ nids $ \nid -> registerRemote nid "logger" logger

247+ forM_ nids $ \nid -> reregisterRemoteAsync nid "logger" logger -- ignore async response

248

249--------------------------------------------------------------------------------

250-- Slaves --

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

@@ -44,6 +44,7 @@ module Control.Distributed.Process

44 , call

45 , terminate

46 , ProcessTerminationException(..)

47+ , ProcessRegistrationException(..)

47 , SpawnRef

48 , SpawnRef

48 , getSelfPid

49 , getSelfPid

49 , getSelfNode

50 , getSelfNode

@@ -77,10 +78,12 @@ module Control.Distributed.Process

77 , say

78 , say

78 -- * Registry

79 -- * Registry

79 , register

80 , register

81+ , reregister

80 , unregister

82 , unregister

81 , whereis

83 , whereis

82 , nsend

84 , nsend

83 , registerRemoteAsync

85 , registerRemoteAsync

86+ , reregisterRemoteAsync

84 , unregisterRemoteAsync

87 , unregisterRemoteAsync

85 , whereisRemoteAsync

88 , whereisRemoteAsync

86 , nsendRemote

89 , nsendRemote

@@ -135,6 +138,7 @@ import Control.Distributed.Process.Internal.Types

135 , ProcessLinkException(..)

138 , ProcessLinkException(..)

136 , NodeLinkException(..)

139 , NodeLinkException(..)

137 , PortLinkException(..)

140 , PortLinkException(..)

141+ , ProcessRegistrationException(..)

138 , DiedReason(..)

142 , DiedReason(..)

139 , SpawnRef(..)

143 , SpawnRef(..)

140 , DidSpawn(..)

144 , DidSpawn(..)

@@ -197,10 +201,12 @@ import Control.Distributed.Process.Internal.Primitives

197 , say

201 , say

198 -- Registry

202 -- Registry

199 , register

203 , register

204+ , reregister

200 , unregister

205 , unregister

201 , whereis

206 , whereis

202 , nsend

207 , nsend

203 , registerRemoteAsync

208 , registerRemoteAsync

209+ , reregisterRemoteAsync

204 , unregisterRemoteAsync

210 , unregisterRemoteAsync

205 , whereisRemoteAsync

211 , whereisRemoteAsync

206 , nsendRemote

212 , nsendRemote

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+25−5

@@ -35,10 +35,12 @@ module Control.Distributed.Process.Internal.Primitives

35 , say

36 -- * Registry

37 , register

38+ , reregister

38 , unregister

39 , unregister

39 , whereis

40 , whereis

40 , nsend

41 , nsend

41 , registerRemoteAsync

42 , registerRemoteAsync

43+ , reregisterRemoteAsync

42 , unregisterRemoteAsync

44 , unregisterRemoteAsync

43 , whereisRemoteAsync

45 , whereisRemoteAsync

44 , nsendRemote

46 , nsendRemote

@@ -542,11 +544,23 @@ say string = do

542--------------------------------------------------------------------------------

544--------------------------------------------------------------------------------

543

545

544-- | Register a process with the local registry (asynchronous).

546-- | Register a process with the local registry (asynchronous).

547+-- This version will wait until a response is gotten from the

548+-- management process. The name must not already be registered.

549+-- The process need not be on this node.

550+-- A bad registration will result in a 'ProcessRegistrationException'

545--

551--

546-- The process to be registered does not have to be local itself.

552-- The process to be registered does not have to be local itself.

547register :: String -> ProcessId -> Process ()

553register :: String -> ProcessId -> Process ()

554+register = registerImpl False

555+

556+-- | Like 'register', but will replace an existing registration.

557+-- The name must already be registered.

558+reregister :: String -> ProcessId -> Process ()

559+reregister = registerImpl True

560+

561+registerImpl :: Bool -> String -> ProcessId -> Process ()

548−register label pid = do

562+registerImpl force label pid = do

549− sendCtrlMsg Nothing (Register label (Just pid))

563+ sendCtrlMsg Nothing (Register label (Just pid) force)

550 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')

564 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')

551 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)

565 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)

552 ]

566 ]

@@ -559,12 +573,18 @@ register label pid = do

559-- See comments in 'whereisRemoteAsync'

573-- See comments in 'whereisRemoteAsync'

560registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()

574registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()

561registerRemoteAsync nid label pid =

575registerRemoteAsync nid label pid =

562− sendCtrlMsg (Just nid) (Register label (Just pid))

576+ sendCtrlMsg (Just nid) (Register label (Just pid) False)

577+

578+reregisterRemoteAsync :: NodeId -> String -> ProcessId -> Process ()

579+reregisterRemoteAsync nid label pid =

580+ sendCtrlMsg (Just nid) (Register label (Just pid) True)

563

581

564-- | Remove a process from the local registry (asynchronous).

582-- | Remove a process from the local registry (asynchronous).

583+-- This version will wait until a response is gotten from the

584+-- management process. The name must already be registered.

565unregister :: String -> Process ()

585unregister :: String -> Process ()

566unregister label = do

586unregister label = do

567− sendCtrlMsg Nothing (Register label Nothing)

587+ sendCtrlMsg Nothing (Register label Nothing False)

568 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')

588 receiveWait [ matchIf (\(RegisterReply label' _) -> label == label')

569 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)

589 (\(RegisterReply _ ok) -> handleRegistrationReply label ok)

570 ]

590 ]

@@ -583,7 +603,7 @@ handleRegistrationReply label ok =

583-- See comments in 'whereisRemoteAsync'

603-- See comments in 'whereisRemoteAsync'

584unregisterRemoteAsync :: NodeId -> String -> Process ()

604unregisterRemoteAsync :: NodeId -> String -> Process ()

585unregisterRemoteAsync nid label =

605unregisterRemoteAsync nid label =

586− sendCtrlMsg (Just nid) (Register label Nothing)

606+ sendCtrlMsg (Just nid) (Register label Nothing False)

587

607

588-- | Query the local process registry

608-- | Query the local process registry

589whereis :: String -> Process (Maybe ProcessId)

609whereis :: String -> Process (Maybe ProcessId)

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

@@ -440,7 +440,7 @@ data ProcessSignal =

440 | Died Identifier !DiedReason

441 | Spawn !(Closure (Process ())) !SpawnRef

442 | WhereIs !String

443− | Register !String !(Maybe ProcessId) -- Use 'Nothing' to unregister

443+ | Register !String !(Maybe ProcessId) !Bool -- Use 'Nothing' to unregister, use True to force reregister

444 | NamedSend !String !Message

445 deriving Show

446

@@ -484,7 +484,7 @@ instance Binary ProcessSignal where

484 put (Died who reason) = putWord8 4 >> put who >> put reason

485 put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

486 put (WhereIs label) = putWord8 6 >> put label

487− put (Register label pid) = putWord8 7 >> put label >> put pid

487+ put (Register label pid force)= putWord8 7 >> put label >> put pid >> put force

488 put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

489 get = do

490 header <- getWord8

@@ -496,7 +496,7 @@ instance Binary ProcessSignal where

496 4 -> Died <$> get <*> get

497 5 -> Spawn <$> get <*> get

498 6 -> WhereIs <$> get

499− 7 -> Register <$> get <*> get

499+ 7 -> Register <$> get <*> get <*> get

500 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

501 _ -> fail "ProcessSignal.get: invalid"

502

distributed-process/src/Control/Distributed/Process/Node.hs+56−21

@@ -24,6 +24,7 @@ import Data.Map (Map)

24import qualified Data.Map as Map

25 ( empty

26 , toList

27+ , fromList

27 , partitionWithKey

28 , partitionWithKey

28 , elems

29 , elems

29 , filterWithKey

30 , filterWithKey

@@ -31,7 +32,7 @@ import qualified Data.Map as Map

31import Data.Set (Set)

32import Data.Set (Set)

32import qualified Data.Set as Set (empty, insert, delete, member)

33import qualified Data.Set as Set (empty, insert, delete, member)

33import Data.Foldable (forM_)

34import Data.Foldable (forM_)

34−import Data.Maybe (isJust, isNothing)

35+import Data.Maybe (isJust, isNothing, catMaybes)

35import Data.Typeable (Typeable)

36import Data.Typeable (Typeable)

36import Control.Category ((>>>))

37import Control.Category ((>>>))

37import Control.Applicative ((<$>))

38import Control.Applicative ((<$>))

@@ -438,8 +439,8 @@ data NCState = NCState

438 _links :: !(Map Identifier (Set ProcessId))

439 _links :: !(Map Identifier (Set ProcessId))

439 -- Mapping from remote processes to monitoring local processes

440 -- Mapping from remote processes to monitoring local processes

440 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))

441 , _monitors :: !(Map Identifier (Set (ProcessId, MonitorRef)))

441− -- Process registry

442+ -- Process registry: names and where they live, mapped to the PIDs

442− , _registry :: !(Map String ProcessId)

443+ , _registry :: !(Map (String,NodeId) ProcessId)

443 }

444 }

444

445

445newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

446newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

@@ -486,8 +487,8 @@ nodeController = do

486 ncEffectDied ident reason

487 ncEffectDied ident reason

487 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->

488 NCMsg (ProcessIdentifier from) (Spawn proc ref) ->

488 ncEffectSpawn from proc ref

489 ncEffectSpawn from proc ref

489− NCMsg (ProcessIdentifier from) (Register label pid) ->

490+ NCMsg (ProcessIdentifier from) (Register label pid force) ->

490− ncEffectRegister from label pid

491+ ncEffectRegister from label pid force

491 NCMsg (ProcessIdentifier from) (WhereIs label) ->

492 NCMsg (ProcessIdentifier from) (WhereIs label) ->

492 ncEffectWhereIs from label

493 ncEffectWhereIs from label

493 NCMsg from (NamedSend label msg') ->

494 NCMsg from (NamedSend label msg') ->

@@ -555,6 +556,8 @@ ncEffectDied ident reason = do

555 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))

556 (affectedLinks, unaffectedLinks) <- gets (splitNotif ident . (^. links))

556 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))

557 (affectedMons, unaffectedMons) <- gets (splitNotif ident . (^. monitors))

557

558

559+-- _registry :: !(Map (String,NodeId) ProcessId)

560+

558 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False

561 let localOnly = case ident of NodeIdentifier _ -> True ; _ -> False

559

562

560 forM_ (Map.toList affectedLinks) $ \(them, uss) ->

563 forM_ (Map.toList affectedLinks) $ \(them, uss) ->

@@ -565,10 +568,27 @@ ncEffectDied ident reason = do

565 forM_ (Map.toList affectedMons) $ \(them, refs) ->

568 forM_ (Map.toList affectedMons) $ \(them, refs) ->

566 forM_ refs $ \(us, ref) ->

569 forM_ refs $ \(us, ref) ->

567 when (localOnly <= isLocal node (ProcessIdentifier us)) $

570 when (localOnly <= isLocal node (ProcessIdentifier us)) $

568− notifyDied us them reason (Just ref)

571+ notifyDied us them reason (Just ref)

569

572

570 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)

573 modify' $ (links ^= unaffectedLinks) . (monitors ^= unaffectedMons)

571

574

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)) $

580+ liftIO $ sendBinary node

581+ (NodeIdentifier $ localNodeId node)

582+ (NodeIdentifier $ nid)

583+ WithImplicitReconnect

584+ NCMsg

585+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

586+ , ctrlMsgSignal = Died ident reason

587+ }

588+ return Nothing

589+ False -> return $ Just whl)

590+ modify' $ registry ^= Map.fromList (catMaybes remaining)

591+

572-- [Unified: Table 13]

592-- [Unified: Table 13]

573ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

593ncEffectSpawn :: ProcessId -> Closure (Process ()) -> SpawnRef -> NC ()

574ncEffectSpawn pid cProc ref = do

594ncEffectSpawn pid cProc ref = do

@@ -589,18 +609,33 @@ ncEffectSpawn pid cProc ref = do

589-- Unified semantics does not explicitly describe how to implement 'register',

609-- Unified semantics does not explicitly describe how to implement 'register',

590-- but mentions it's "very similar to nsend" (Table 14)

610-- but mentions it's "very similar to nsend" (Table 14)

591-- We send a response indicated if the operation is invalid

611-- We send a response indicated if the operation is invalid

592−ncEffectRegister :: ProcessId -> String -> Maybe ProcessId -> NC ()

612+ncEffectRegister :: ProcessId -> String -> Maybe ProcessId -> Bool -> NC ()

593−ncEffectRegister from label mPid = do

613+ncEffectRegister from label mPid reregistration = do

594 node <- ask

614 node <- ask

595− currentVal <- gets (^. registryFor label)

615+ currentVal <- gets (^. registryFor (label, localNodeId node))

596− let isOk =

616+ isOk <-

597 case mPid of

617 case mPid of

598 Nothing -> -- unregister request

618 Nothing -> -- unregister request

599− isJust currentVal

619+ return $ isJust currentVal

600− Just _ -> -- register request

620+ Just thepid -> -- register request

621+ do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)

601− isNothing currentVal

622+ return $ (isNothing currentVal /= reregistration) &&

623+ (not (isLocal node (ProcessIdentifier thepid) ) || isvalidlocal )

624+ if isLocal node (ProcessIdentifier from)

602− when (isOk) $

625+ then when (isOk) $

626+ do modify' $ registryFor (label, localNodeId node) ^= mPid

627+ let namedPid =

628+ head $ catMaybes [mPid, currentVal]

629+ when (not $ isLocal node (ProcessIdentifier namedPid)) $

630+ liftIO $ sendBinary node

631+ (ProcessIdentifier from)

632+ (NodeIdentifier (processNodeId namedPid))

633+ WithImplicitReconnect

634+ NCMsg

635+ { ctrlMsgSender = ProcessIdentifier from

636+ , ctrlMsgSignal = Register label mPid reregistration

637+ }

603− modify' $ registryFor label ^= mPid

638+ else modify' $ registryFor (label,processNodeId from) ^= mPid

604 liftIO $ sendMessage node

639 liftIO $ sendMessage node

605 (NodeIdentifier (localNodeId node))

640 (NodeIdentifier (localNodeId node))

606 (ProcessIdentifier from)

641 (ProcessIdentifier from)

@@ -611,7 +646,7 @@ ncEffectRegister from label mPid = do

611ncEffectWhereIs :: ProcessId -> String -> NC ()

646ncEffectWhereIs :: ProcessId -> String -> NC ()

612ncEffectWhereIs from label = do

647ncEffectWhereIs from label = do

613 node <- ask

648 node <- ask

614− mPid <- gets (^. registryFor label)

649+ mPid <- gets (^. registryFor (label, localNodeId node))

615 liftIO $ sendMessage node

650 liftIO $ sendMessage node

616 (NodeIdentifier (localNodeId node))

651 (NodeIdentifier (localNodeId node))

617 (ProcessIdentifier from)

652 (ProcessIdentifier from)

@@ -621,9 +656,9 @@ ncEffectWhereIs from label = do

621-- [Unified: Table 14]

656-- [Unified: Table 14]

622ncEffectNamedSend :: Identifier -> String -> Message -> NC ()

657ncEffectNamedSend :: Identifier -> String -> Message -> NC ()

623ncEffectNamedSend from label msg = do

658ncEffectNamedSend from label msg = do

659+ node <- ask

624− mPid <- gets (^. registryFor label)

660+ mPid <- gets (^. registryFor (label, localNodeId node))

625 -- If mPid is Nothing, we just ignore the named send (as per Table 14)

661 -- If mPid is Nothing, we just ignore the named send (as per Table 14)

626− node <- ask

627 forM_ mPid $ \pid ->

662 forM_ mPid $ \pid ->

628 liftIO $ sendPayload node

663 liftIO $ sendPayload node

629 from

664 from

@@ -673,7 +708,7 @@ destNid (Unlink ident) = Just $ nodeOf ident

673destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)

708destNid (Monitor ref) = Just $ nodeOf (monitorRefIdent ref)

674destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)

709destNid (Unmonitor ref) = Just $ nodeOf (monitorRefIdent ref)

675destNid (Spawn _ _) = Nothing

710destNid (Spawn _ _) = Nothing

676−destNid (Register _ _) = Nothing

711+destNid (Register _ _ _)= Nothing

677destNid (WhereIs _) = Nothing

712destNid (WhereIs _) = Nothing

678destNid (NamedSend _ _) = Nothing

713destNid (NamedSend _ _) = Nothing

679-- We don't need to forward 'Died' signals; if monitoring/linking is setup,

714-- 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 })

745monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))

780monitors :: Accessor NCState (Map Identifier (Set (ProcessId, MonitorRef)))

746monitors = accessor _monitors (\ms st -> st { _monitors = ms })

781monitors = accessor _monitors (\ms st -> st { _monitors = ms })

747

782

748−registry :: Accessor NCState (Map String ProcessId)

783+registry :: Accessor NCState (Map (String,NodeId) ProcessId)

749registry = accessor _registry (\ry st -> st { _registry = ry })

784registry = accessor _registry (\ry st -> st { _registry = ry })

750

785

751linksFor :: Identifier -> Accessor NCState (Set ProcessId)

786linksFor :: Identifier -> Accessor NCState (Set ProcessId)

@@ -754,7 +789,7 @@ linksFor ident = links >>> DAC.mapDefault Set.empty ident

754monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))

789monitorsFor :: Identifier -> Accessor NCState (Set (ProcessId, MonitorRef))

755monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident

790monitorsFor ident = monitors >>> DAC.mapDefault Set.empty ident

756

791

757−registryFor :: String -> Accessor NCState (Maybe ProcessId)

792+registryFor :: (String, NodeId) -> Accessor NCState (Maybe ProcessId)

758registryFor ident = registry >>> DAC.mapMaybe ident

793registryFor ident = registry >>> DAC.mapMaybe ident

759

794

760-- | @splitNotif ident@ splits a notifications map into those

795-- | @splitNotif ident@ splits a notifications map into those