474909d · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge pull request #76 from jepst/master

474909dEdsko de Vries14 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/distributed-process.cabal+1−1

@@ -1,5 +1,5 @@

1Name: distributed-process

2−Version: 0.4.0.2

2+Version: 0.4.0.3

3Cabal-Version: >=1.8

4Build-Type: Simple

5License: BSD3

distributed-process/src/Control/Distributed/Process.hs+10−4

@@ -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,11 +78,13 @@ 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− , registerRemote

85+ , registerRemoteAsync

84− , unregisterRemote

86+ , reregisterRemoteAsync

87+ , unregisterRemoteAsync

85 , whereisRemoteAsync

88 , whereisRemoteAsync

86 , nsendRemote

89 , nsendRemote

87 , WhereIsReply(..)

90 , WhereIsReply(..)

@@ -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,11 +201,13 @@ 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− , registerRemote

208+ , registerRemoteAsync

204− , unregisterRemote

209+ , reregisterRemoteAsync

210+ , unregisterRemoteAsync

205 , whereisRemoteAsync

211 , whereisRemoteAsync

206 , nsendRemote

212 , nsendRemote

207 -- Closures

213 -- Closures

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+57−12

@@ -35,11 +35,13 @@ 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− , registerRemote

42+ , registerRemoteAsync

42− , unregisterRemote

43+ , reregisterRemoteAsync

44+ , unregisterRemoteAsync

43 , whereisRemoteAsync

45 , whereisRemoteAsync

44 , nsendRemote

46 , nsendRemote

45 -- * Closures

47 -- * Closures

@@ -76,6 +78,7 @@ import Data.Time.Clock (getCurrentTime)

76import Data.Time.Format (formatTime)

78import Data.Time.Format (formatTime)

77import System.Locale (defaultTimeLocale)

79import System.Locale (defaultTimeLocale)

78import System.Timeout (timeout)

80import System.Timeout (timeout)

81+import Control.Monad (when)

79import Control.Monad.Reader (ask)

82import Control.Monad.Reader (ask)

80import Control.Monad.IO.Class (MonadIO, liftIO)

83import Control.Monad.IO.Class (MonadIO, liftIO)

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

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

@@ -127,6 +130,8 @@ import Control.Distributed.Process.Internal.Types

127 , DidUnlinkNode(..)

130 , DidUnlinkNode(..)

128 , DidUnlinkPort(..)

131 , DidUnlinkPort(..)

129 , WhereIsReply(..)

132 , WhereIsReply(..)

133+ , RegisterReply(..)

134+ , ProcessRegistrationException(..)

130 , createMessage

135 , createMessage

131 , runLocalProcess

136 , runLocalProcess

132 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

137 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

@@ -539,28 +544,68 @@ say string = do

539--------------------------------------------------------------------------------

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

540

545

541-- | 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'

542--

551--

543-- 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.

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

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

545−register label pid =

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

562+registerImpl force label pid = do

563+ mynid <- getSelfNode

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

564+ sendCtrlMsg Nothing (Register label mynid (Just pid) force)

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

566+ (\(RegisterReply _ ok) -> handleRegistrationReply label ok)

567+ ]

547

568

548-- | Register a process with a remote registry (asynchronous).

569-- | Register a process with a remote registry (asynchronous).

549--

570--

550-- The process to be registered does not have to live on the same remote node.

571-- The process to be registered does not have to live on the same remote node.

572+-- Reply wil come in the form of a 'RegisterReply' message

573+--

574+-- See comments in 'whereisRemoteAsync'

551−registerRemote :: NodeId -> String -> ProcessId -> Process ()

575+registerRemoteAsync :: NodeId -> String -> ProcessId -> Process ()

552−registerRemote nid label pid =

576+registerRemoteAsync nid label pid =

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

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

578+

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

580+reregisterRemoteAsync nid label pid =

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

554

582

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

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

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

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

556unregister :: String -> Process ()

586unregister :: String -> Process ()

557−unregister label =

587+unregister label = do

588+ mynid <- getSelfNode

558− sendCtrlMsg Nothing (Register label Nothing)

589+ sendCtrlMsg Nothing (Register label mynid Nothing False)

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

591+ (\(RegisterReply _ ok) -> handleRegistrationReply label ok)

592+ ]

593+

594+-- | Deal with the result from an attempted registration or unregistration

595+-- by throwing an exception if necessary

596+handleRegistrationReply :: String -> Bool -> Process ()

597+handleRegistrationReply label ok =

598+ when (not ok) $

599+ liftIO $ throwIO $ ProcessRegistrationException label

559

600

560-- | Remove a process from a remote registry (asynchronous).

601-- | Remove a process from a remote registry (asynchronous).

602+--

603+-- Reply wil come in the form of a 'RegisterReply' message

604+--

605+-- See comments in 'whereisRemoteAsync'

561−unregisterRemote :: NodeId -> String -> Process ()

606+unregisterRemoteAsync :: NodeId -> String -> Process ()

562−unregisterRemote nid label =

607+unregisterRemoteAsync nid label =

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

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

564

609

565-- | Query the local process registry

610-- | Query the local process registry

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

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

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

@@ -39,6 +39,7 @@ module Control.Distributed.Process.Internal.Types

39 , ProcessLinkException(..)

40 , NodeLinkException(..)

41 , PortLinkException(..)

42+ , ProcessRegistrationException(..)

42 , DiedReason(..)

43 , DiedReason(..)

43 , DidUnmonitor(..)

44 , DidUnmonitor(..)

44 , DidUnlinkProcess(..)

45 , DidUnlinkProcess(..)

@@ -47,6 +48,7 @@ module Control.Distributed.Process.Internal.Types

47 , SpawnRef(..)

48 , SpawnRef(..)

48 , DidSpawn(..)

49 , DidSpawn(..)

49 , WhereIsReply(..)

50 , WhereIsReply(..)

51+ , RegisterReply(..)

50 -- * Node controller internal data types

52 -- * Node controller internal data types

51 , NCMsg(..)

53 , NCMsg(..)

52 , ProcessSignal(..)

54 , ProcessSignal(..)

@@ -359,9 +361,17 @@ data PortLinkException =

359 PortLinkException !SendPortId !DiedReason

361 PortLinkException !SendPortId !DiedReason

360 deriving (Typeable, Show)

362 deriving (Typeable, Show)

361

363

364+-- | Exception thrown when a process attempts to register

365+-- a process under an already-registered name or to

366+-- unregister a name that hasn't been registered

367+data ProcessRegistrationException =

368+ ProcessRegistrationException !String

369+ deriving (Typeable, Show)

370+

362instance Exception ProcessLinkException

371instance Exception ProcessLinkException

363instance Exception NodeLinkException

372instance Exception NodeLinkException

364instance Exception PortLinkException

373instance Exception PortLinkException

374+instance Exception ProcessRegistrationException

365

375

366-- | Why did a process die?

376-- | Why did a process die?

367data DiedReason =

377data DiedReason =

@@ -406,6 +416,10 @@ data DidSpawn = DidSpawn SpawnRef ProcessId

406data WhereIsReply = WhereIsReply String (Maybe ProcessId)

416data WhereIsReply = WhereIsReply String (Maybe ProcessId)

407 deriving (Show, Typeable)

417 deriving (Show, Typeable)

408

418

419+-- | (Asynchronous) reply from 'register' and 'unregister'

420+data RegisterReply = RegisterReply String Bool

421+ deriving (Show, Typeable)

422+

409--------------------------------------------------------------------------------

423--------------------------------------------------------------------------------

410-- Node controller internal data types --

424-- Node controller internal data types --

411--------------------------------------------------------------------------------

425--------------------------------------------------------------------------------

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

426 | Died Identifier !DiedReason

440 | Died Identifier !DiedReason

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

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

428 | WhereIs !String

442 | WhereIs !String

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

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

430 | NamedSend !String !Message

444 | NamedSend !String !Message

431 deriving Show

445 deriving Show

432

446

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

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

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

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

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

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

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

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

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

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

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

475 get = do

489 get = do

476 header <- getWord8

490 header <- getWord8

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

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

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

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

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

484 6 -> WhereIs <$> get

498 6 -> WhereIs <$> get

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

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

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

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

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

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

488

502

@@ -526,6 +540,10 @@ instance Binary WhereIsReply where

526 put (WhereIsReply label mPid) = put label >> put mPid

540 put (WhereIsReply label mPid) = put label >> put mPid

527 get = WhereIsReply <$> get <*> get

541 get = WhereIsReply <$> get <*> get

528

542

543+instance Binary RegisterReply where

544+ put (RegisterReply label ok) = put label >> put ok

545+ get = RegisterReply <$> get <*> get

546+

529--------------------------------------------------------------------------------

547--------------------------------------------------------------------------------

530-- Accessors --

548-- Accessors --

531--------------------------------------------------------------------------------

549--------------------------------------------------------------------------------

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

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

24import qualified Data.Map as Map

25 ( empty

26 , toList

27+ , fromList

28+ , filter

27 , partitionWithKey

29 , partitionWithKey

28 , elems

30 , elems

29 , filterWithKey

31 , filterWithKey

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

31import Data.Set (Set)

33import Data.Set (Set)

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

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

33import Data.Foldable (forM_)

35import Data.Foldable (forM_)

34−import Data.Maybe (isJust)

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

35import Data.Typeable (Typeable)

37import Data.Typeable (Typeable)

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

38import Control.Category ((>>>))

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

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

@@ -118,6 +120,7 @@ import Control.Distributed.Process.Internal.Types

118 , nodeOf

120 , nodeOf

119 , SendPortId(..)

121 , SendPortId(..)

120 , typedChannelWithId

122 , typedChannelWithId

123+ , RegisterReply(..)

121 , WhereIsReply(..)

124 , WhereIsReply(..)

122 , messageToPayload

125 , messageToPayload

123 , payloadToMessage

126 , payloadToMessage

@@ -437,8 +440,9 @@ data NCState = NCState

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

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

438 -- Mapping from remote processes to monitoring local processes

441 -- Mapping from remote processes to monitoring local processes

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

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

440− -- Process registry

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

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

444+ , _registeredHere :: !(Map String ProcessId)

445+ , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])

442 }

446 }

443

447

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

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

@@ -447,7 +451,8 @@ newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }

447initNCState :: NCState

451initNCState :: NCState

448initNCState = NCState { _links = Map.empty

452initNCState = NCState { _links = Map.empty

449 , _monitors = Map.empty

453 , _monitors = Map.empty

450− , _registry = Map.empty

454+ , _registeredHere = Map.empty

455+ , _registeredOnNodes = Map.empty

451 }

456 }

452

457

453--------------------------------------------------------------------------------

458--------------------------------------------------------------------------------

@@ -485,8 +490,8 @@ nodeController = do

485 ncEffectDied ident reason

490 ncEffectDied ident reason

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

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

487 ncEffectSpawn from proc ref

492 ncEffectSpawn from proc ref

488− NCMsg _from (Register label pid) ->

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

489− ncEffectRegister label pid

494+ ncEffectRegister from label atnode pid force

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

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

491 ncEffectWhereIs from label

496 ncEffectWhereIs from label

492 NCMsg from (NamedSend label msg') ->

497 NCMsg from (NamedSend label msg') ->

@@ -554,6 +559,8 @@ ncEffectDied ident reason = do

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

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

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

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

556

561

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

563+

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

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

558

565

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

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

@@ -564,10 +571,34 @@ ncEffectDied ident reason = do

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

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

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

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

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

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

567− notifyDied us them reason (Just ref)

574+ notifyDied us them reason (Just ref)

568

575

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

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

570

577

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 =

592+ liftIO $ sendBinary node

593+ (NodeIdentifier $ localNodeId node)

594+ (NodeIdentifier $ nid)

595+ WithImplicitReconnect

596+ NCMsg

597+ { ctrlMsgSender = NodeIdentifier (localNodeId node)

598+ , ctrlMsgSignal = Died ident reason

599+ }

600+

601+

571-- [Unified: Table 13]

602-- [Unified: Table 13]

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

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

573ncEffectSpawn pid cProc ref = do

604ncEffectSpawn pid cProc ref = do

@@ -587,18 +618,71 @@ ncEffectSpawn pid cProc ref = do

587

618

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

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

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

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

621+-- We send a response indicated if the operation is invalid

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

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

591−ncEffectRegister label mPid =

623+ncEffectRegister from label atnode mPid reregistration = do

624+ node <- ask

625+ currentVal <- gets (^. registeredHereFor label)

626+ isOk <-

627+ case mPid of

628+ Nothing -> -- unregister request

629+ return $ isJust currentVal

630+ Just thepid -> -- register request

631+ do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)

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

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

634+ if isLocal node (NodeIdentifier atnode)

635+ then do when (isOk) $

592− modify' $ registryFor label ^= mPid

636+ do modify' $ registeredHereFor label ^= mPid

593− -- An acknowledgement is not necessary. If we want a synchronous register,

637+ updateRemote node currentVal mPid

594− -- it suffices to send a whereis requiry immediately after the register

638+ liftIO $ sendMessage node

595− -- (that may not suffice if we do decide for unreliable messaging instead)

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

671+ liftIO $ sendBinary node

672+ (ProcessIdentifier from)

673+ (NodeIdentifier to)

674+ WithImplicitReconnect

675+ NCMsg

676+ { ctrlMsgSender = ProcessIdentifier from

677+ , ctrlMsgSignal = reg

678+ }

679+

596

680

597-- Unified semantics does not explicitly describe 'whereis'

681-- Unified semantics does not explicitly describe 'whereis'

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

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

599ncEffectWhereIs from label = do

683ncEffectWhereIs from label = do

600 node <- ask

684 node <- ask

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

685+ mPid <- gets (^. registeredHereFor label)

602 liftIO $ sendMessage node

686 liftIO $ sendMessage node

603 (NodeIdentifier (localNodeId node))

687 (NodeIdentifier (localNodeId node))

604 (ProcessIdentifier from)

688 (ProcessIdentifier from)

@@ -608,9 +692,9 @@ ncEffectWhereIs from label = do

608-- [Unified: Table 14]

692-- [Unified: Table 14]

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

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

610ncEffectNamedSend from label msg = do

694ncEffectNamedSend from label msg = do

695+ node <- ask

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

696+ mPid <- gets (^. registeredHereFor label)

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

613− node <- ask

614 forM_ mPid $ \pid ->

698 forM_ mPid $ \pid ->

615 liftIO $ sendPayload node

699 liftIO $ sendPayload node

616 from

700 from

@@ -660,7 +744,7 @@ destNid (Unlink ident) = Just $ nodeOf ident

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

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

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

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

662destNid (Spawn _ _) = Nothing

746destNid (Spawn _ _) = Nothing

663−destNid (Register _ _) = Nothing

747+destNid (Register _ _ _ _) = Nothing

664destNid (WhereIs _) = Nothing

748destNid (WhereIs _) = Nothing

665destNid (NamedSend _ _) = Nothing

749destNid (NamedSend _ _) = Nothing

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

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

@@ -732,8 +816,11 @@ links = accessor _links (\ls st -> st { _links = ls })

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

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

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

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

734

818

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

819+registeredHere :: Accessor NCState (Map String ProcessId)

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

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

737

824

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

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

739linksFor ident = links >>> DAC.mapDefault Set.empty ident

826linksFor ident = links >>> DAC.mapDefault Set.empty ident

@@ -741,8 +828,11 @@ linksFor ident = links >>> DAC.mapDefault Set.empty ident

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

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

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

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

743

830

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

831+registeredHereFor :: String -> Accessor NCState (Maybe ProcessId)

745−registryFor ident = registry >>> DAC.mapMaybe ident

832+registeredHereFor ident = registeredHere >>> DAC.mapMaybe ident

833+

834+registeredOnNodesFor :: ProcessId -> Accessor NCState (Maybe [(NodeId,Int)])

835+registeredOnNodesFor ident = registeredOnNodes >>> DAC.mapMaybe ident

746

836

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

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

748-- notifications that should trigger when 'ident' fails and those links that

838-- notifications that should trigger when 'ident' fails and those links that

distributed-process/tests/TestCH.hs+23−6

@@ -31,6 +31,7 @@ import Control.Distributed.Process

31import Control.Distributed.Process.Internal.Types

32 ( NodeId(nodeAddress)

33 , LocalNode(localEndPoint)

34+ , RegisterReply(..)

34 )

35 )

35import Control.Distributed.Process.Node

36import Control.Distributed.Process.Node

36import Control.Distributed.Process.Serializable (Serializable)

37import Control.Distributed.Process.Serializable (Serializable)

@@ -594,7 +595,11 @@ testRemoteRegistry transport = do

594

595

595 runProcess node2 $ do

596 runProcess node2 $ do

596 let nid1 = localNodeId node1

597 let nid1 = localNodeId node1

597− registerRemote nid1 "ping" pingServer

598+ registerRemoteAsync nid1 "ping" pingServer

599+ receiveWait [

600+ matchIf (\(RegisterReply label' _) -> "ping" == label')

601+ (\(RegisterReply _ _) -> return ()) ]

602+

598 Just pid <- whereisRemote nid1 "ping"

603 Just pid <- whereisRemote nid1 "ping"

599 True <- return $ pingServer == pid

604 True <- return $ pingServer == pid

600 us <- getSelfPid

605 us <- getSelfPid

@@ -665,7 +670,13 @@ testReconnect transport transportInternals = do

665 -}

670 -}

666

671

667 us <- getSelfPid

672 us <- getSelfPid

668− registerRemote nid1 "a" us >> liftIO (threadDelay 100000) -- registerRemote is asynchronous

673+ registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous

674+ receiveWait [

675+ matchIf (\(RegisterReply label' _) -> "a" == label')

676+ (\(RegisterReply _ _) -> return ()) ]

677+

678+ Just _ <- whereisRemote nid1 "a"

679+

669

680

670 -- Simulate network failure

681 -- Simulate network failure

671 liftIO $ do

682 liftIO $ do

@@ -674,14 +685,20 @@ testReconnect transport transportInternals = do

674 threadDelay 10000

685 threadDelay 10000

675

686

676 -- This will happen due to implicit reconnect

687 -- This will happen due to implicit reconnect

677− registerRemote nid1 "b" us

688+ registerRemoteAsync nid1 "b" us

689+ receiveWait [

690+ matchIf (\(RegisterReply label' _) -> "b" == label')

691+ (\(RegisterReply _ _) -> return ()) ]

678

692

679 -- Should happen

693 -- Should happen

680− registerRemote nid1 "c" us

694+ registerRemoteAsync nid1 "c" us

695+ receiveWait [

696+ matchIf (\(RegisterReply label' _) -> "c" == label')

697+ (\(RegisterReply _ _) -> return ()) ]

681

698

682 -- Check

699 -- Check

683− Just _ <- whereisRemote nid1 "a"

700+ Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected

684− Just _ <- whereisRemote nid1 "b"

701+ Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect

685 Just _ <- whereisRemote nid1 "c"

702 Just _ <- whereisRemote nid1 "c"

686

703

687 liftIO $ putMVar registerTestOk ()

704 liftIO $ putMVar registerTestOk ()