53d6312 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fixed bug for setting names at remote nodes

53d6312Jeff Epstein14 years ago

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

compare with master

unifiedsplit

@@ -560,7 +560,8 @@ reregister = registerImpl True

560

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

562registerImpl force label pid = do

563+ mynid <- getSelfNode

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

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

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

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

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

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

566 ]

567 ]

@@ -573,18 +574,19 @@ registerImpl force label pid = do

573-- See comments in 'whereisRemoteAsync'

574-- See comments in 'whereisRemoteAsync'

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

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

575registerRemoteAsync nid label pid =

576registerRemoteAsync nid label pid =

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

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

577

578

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

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

579reregisterRemoteAsync nid label pid =

580reregisterRemoteAsync nid label pid =

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

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

581

582

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

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

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

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

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

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

585unregister :: String -> Process ()

586unregister :: String -> Process ()

586unregister label = do

587unregister label = do

588+ mynid <- getSelfNode

587− sendCtrlMsg Nothing (Register label Nothing False)

589+ sendCtrlMsg Nothing (Register label mynid Nothing False)

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

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

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

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

590 ]

592 ]

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

603-- See comments in 'whereisRemoteAsync'

605-- See comments in 'whereisRemoteAsync'

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

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

605unregisterRemoteAsync nid label =

607unregisterRemoteAsync nid label =

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

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

607

609

608-- | Query the local process registry

610-- | Query the local process registry

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

611whereis :: 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) !Bool -- Use 'Nothing' to unregister, use True to force reregister

443+ | Register !String !NodeId !(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 force)= putWord8 7 >> put label >> put pid >> put force

487+ put (Register label nid pid force) = putWord8 7 >> put label >> put nid >> 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 <*> get

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

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

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

502

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

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

487 ncEffectDied ident reason

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

489 ncEffectSpawn from proc ref

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

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

491− ncEffectRegister from label pid force

491+ ncEffectRegister from label atnode pid force

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

493 ncEffectWhereIs from label

494 NCMsg from (NamedSend label msg') ->

@@ -609,8 +609,8 @@ ncEffectSpawn pid cProc ref = do

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

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

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

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

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

613−ncEffectRegister from label mPid reregistration = do

613+ncEffectRegister from label atnode mPid reregistration = do

614 node <- ask

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

616 isOk <-

@@ -621,9 +621,9 @@ ncEffectRegister from label mPid reregistration = do

621 do isvalidlocal <- isValidLocalIdentifier (ProcessIdentifier thepid)

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

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

624− if isLocal node (ProcessIdentifier from)

624+ if isLocal node (NodeIdentifier atnode)

625 then when (isOk) $

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

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

627 let namedPid =

628 head $ catMaybes [mPid, currentVal]

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

@@ -633,9 +633,9 @@ ncEffectRegister from label mPid reregistration = do

633 WithImplicitReconnect

634 NCMsg

635 { ctrlMsgSender = ProcessIdentifier from

636− , ctrlMsgSignal = Register label mPid reregistration

636+ , ctrlMsgSignal = Register label atnode mPid reregistration

637 }

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

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

639 liftIO $ sendMessage node

640 (NodeIdentifier (localNodeId node))

641 (ProcessIdentifier from)

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

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

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

710destNid (Spawn _ _) = Nothing

711−destNid (Register _ _ _)= Nothing

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

712destNid (WhereIs _) = Nothing

713destNid (NamedSend _ _) = Nothing

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

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