37b0263 · distributed-process

git.iko.soy/distributed-process

Changed semantics of register/unregister to match Erlang

register will now throw an exception if the name is already registered unregister will throw an exception if the name isn't registered registerRemote and unregisterRemote have been replaced with Async versions, such that the user application is now responsible for checking the remote control process's response

commit37b02637a73735c5164fc7a585782534f9533736 authorJeff Epstein <jepst79@gmail.com> dateSun 18 Nov 2012, 05:52 (14 years ago) parentaf14826

5 files changed+76−22

compare with master

unifiedsplit

distributed-process/distributed-process.cabal+1−1 distributed-process/src/Control/Distributed/Process.hs+4−4 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+31−8 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+18 distributed-process/src/Control/Distributed/Process/Node.hs+22−9

distributed-process/distributed-process.cabal+1−1

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

11Name: distributed-process

2−Version: 0.4.0.2

2+Version: 0.4.0.3

33Cabal-Version: >=1.8

44Build-Type: Simple

55License: BSD3

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

@@ -80,8 +80,8 @@ module Control.Distributed.Process

8080 , unregister

8181 , whereis

8282 , nsend

83− , registerRemote

84− , unregisterRemote

83+ , registerRemoteAsync

84+ , unregisterRemoteAsync

8585 , whereisRemoteAsync

8686 , nsendRemote

8787 , WhereIsReply(..)

@@ -200,8 +200,8 @@ import Control.Distributed.Process.Internal.Primitives

200200 , unregister

201201 , whereis

202202 , nsend

203− , registerRemote

204− , unregisterRemote

203+ , registerRemoteAsync

204+ , unregisterRemoteAsync

205205 , whereisRemoteAsync

206206 , nsendRemote

207207 -- Closures

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+31−8

@@ -38,8 +38,8 @@ module Control.Distributed.Process.Internal.Primitives

3838 , unregister

3939 , whereis

4040 , nsend

41− , registerRemote

42− , unregisterRemote

41+ , registerRemoteAsync

42+ , unregisterRemoteAsync

4343 , whereisRemoteAsync

4444 , nsendRemote

4545 -- * Closures

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

7676import Data.Time.Format (formatTime)

7777import System.Locale (defaultTimeLocale)

7878import System.Timeout (timeout)

79+import Control.Monad (when)

7980import Control.Monad.Reader (ask)

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

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

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

127128 , DidUnlinkNode(..)

128129 , DidUnlinkPort(..)

129130 , WhereIsReply(..)

131+ , RegisterReply(..)

132+ , ProcessRegistrationException(..)

130133 , createMessage

131134 , runLocalProcess

132135 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

@@ -542,24 +545,44 @@ say string = do

542545--

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

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

545−register label pid =

548+register label pid = do

546549 sendCtrlMsg Nothing (Register label (Just pid))

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

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

552+ ]

547553

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

549555--

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

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

552−registerRemote nid label pid =

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

558+--

559+-- See comments in 'whereisRemoteAsync'

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

561+registerRemoteAsync nid label pid =

553562 sendCtrlMsg (Just nid) (Register label (Just pid))

554563

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

556565unregister :: String -> Process ()

557−unregister label =

566+unregister label = do

558567 sendCtrlMsg Nothing (Register label Nothing)

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

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

570+ ]

571+

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

573+-- by throwing an exception if necessary

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

575+handleRegistrationReply label ok =

576+ when (not ok) $

577+ liftIO $ throwIO $ ProcessRegistrationException label

559578

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

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

562−unregisterRemote nid label =

580+--

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

582+--

583+-- See comments in 'whereisRemoteAsync'

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

585+unregisterRemoteAsync nid label =

563586 sendCtrlMsg (Just nid) (Register label Nothing)

564587

565588-- | Query the local process registry

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

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

3939 , ProcessLinkException(..)

4040 , NodeLinkException(..)

4141 , PortLinkException(..)

42+ , ProcessRegistrationException(..)

4243 , DiedReason(..)

4344 , DidUnmonitor(..)

4445 , DidUnlinkProcess(..)

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

4748 , SpawnRef(..)

4849 , DidSpawn(..)

4950 , WhereIsReply(..)

51+ , RegisterReply(..)

5052 -- * Node controller internal data types

5153 , NCMsg(..)

5254 , ProcessSignal(..)

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

359361 PortLinkException !SendPortId !DiedReason

360362 deriving (Typeable, Show)

361363

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+

362371instance Exception ProcessLinkException

363372instance Exception NodeLinkException

364373instance Exception PortLinkException

374+instance Exception ProcessRegistrationException

365375

366376-- | Why did a process die?

367377data DiedReason =

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

406416data WhereIsReply = WhereIsReply String (Maybe ProcessId)

407417 deriving (Show, Typeable)

408418

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

420+data RegisterReply = RegisterReply String Bool

421+ deriving (Show, Typeable)

422+

409423--------------------------------------------------------------------------------

410424-- Node controller internal data types --

411425--------------------------------------------------------------------------------

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

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

527541 get = WhereIsReply <$> get <*> get

528542

543+instance Binary RegisterReply where

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

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

546+

529547--------------------------------------------------------------------------------

530548-- Accessors --

531549--------------------------------------------------------------------------------

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

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

3131import Data.Set (Set)

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

3333import Data.Foldable (forM_)

34−import Data.Maybe (isJust)

34+import Data.Maybe (isJust, isNothing)

3535import Data.Typeable (Typeable)

3636import Control.Category ((>>>))

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

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

118118 , nodeOf

119119 , SendPortId(..)

120120 , typedChannelWithId

121+ , RegisterReply(..)

121122 , WhereIsReply(..)

122123 , messageToPayload

123124 , payloadToMessage

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

485486 ncEffectDied ident reason

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

487488 ncEffectSpawn from proc ref

488− NCMsg _from (Register label pid) ->

489− ncEffectRegister label pid

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

490+ ncEffectRegister from label pid

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

491492 ncEffectWhereIs from label

492493 NCMsg from (NamedSend label msg') ->

@@ -587,12 +588,24 @@ ncEffectSpawn pid cProc ref = do

587588

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

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

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

591−ncEffectRegister label mPid =

592− modify' $ registryFor label ^= mPid

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

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

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

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

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

593+ncEffectRegister from label mPid = do

594+ node <- ask

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

596+ let isOk =

597+ case mPid of

598+ Nothing -> -- unregister request

599+ isJust currentVal

600+ Just _ -> -- register request

601+ isNothing currentVal

602+ when (isOk) $

603+ modify' $ registryFor label ^= mPid

604+ liftIO $ sendMessage node

605+ (NodeIdentifier (localNodeId node))

606+ (ProcessIdentifier from)

607+ WithImplicitReconnect

608+ (RegisterReply label isOk)

596609

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

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