37b0263 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Changed semantics of register/unregister to match Erlang

37b0263Jeff Epstein14 years ago

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

compare with master

unifiedsplit

@@ -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+4−4

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

80 , unregister

81 , whereis

82 , nsend

83− , registerRemote

83+ , registerRemoteAsync

84− , unregisterRemote

84+ , unregisterRemoteAsync

85 , whereisRemoteAsync

86 , nsendRemote

87 , WhereIsReply(..)

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

200 , unregister

201 , whereis

202 , nsend

203− , registerRemote

203+ , registerRemoteAsync

204− , unregisterRemote

204+ , unregisterRemoteAsync

205 , whereisRemoteAsync

206 , nsendRemote

207 -- Closures

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

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

38 , unregister

39 , whereis

40 , nsend

41− , registerRemote

41+ , registerRemoteAsync

42− , unregisterRemote

42+ , unregisterRemoteAsync

43 , whereisRemoteAsync

44 , nsendRemote

45 -- * Closures

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

76import Data.Time.Format (formatTime)

77import System.Locale (defaultTimeLocale)

78import System.Timeout (timeout)

79+import Control.Monad (when)

79import Control.Monad.Reader (ask)

80import Control.Monad.Reader (ask)

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

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

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

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

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

127 , DidUnlinkNode(..)

128 , DidUnlinkNode(..)

128 , DidUnlinkPort(..)

129 , DidUnlinkPort(..)

129 , WhereIsReply(..)

130 , WhereIsReply(..)

131+ , RegisterReply(..)

132+ , ProcessRegistrationException(..)

130 , createMessage

133 , createMessage

131 , runLocalProcess

134 , runLocalProcess

132 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

135 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

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

542--

545--

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

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

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

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

545−register label pid =

548+register label pid = do

546 sendCtrlMsg Nothing (Register label (Just pid))

549 sendCtrlMsg Nothing (Register label (Just pid))

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

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

552+ ]

547

553

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

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

549--

555--

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

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

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

558+--

559+-- See comments in 'whereisRemoteAsync'

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

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

552−registerRemote nid label pid =

561+registerRemoteAsync nid label pid =

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

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

554

563

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

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

556unregister :: String -> Process ()

565unregister :: String -> Process ()

557−unregister label =

566+unregister label = do

558 sendCtrlMsg Nothing (Register label Nothing)

567 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

559

578

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

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

580+--

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

582+--

583+-- See comments in 'whereisRemoteAsync'

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

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

562−unregisterRemote nid label =

585+unregisterRemoteAsync nid label =

563 sendCtrlMsg (Just nid) (Register label Nothing)

586 sendCtrlMsg (Just nid) (Register label Nothing)

564

587

565-- | Query the local process registry

588-- | 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

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

@@ -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+22−9

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

31import Data.Set (Set)

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

33import Data.Foldable (forM_)

34−import Data.Maybe (isJust)

34+import Data.Maybe (isJust, isNothing)

35import Data.Typeable (Typeable)

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

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

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

118 , nodeOf

119 , SendPortId(..)

120 , typedChannelWithId

121+ , RegisterReply(..)

121 , WhereIsReply(..)

122 , WhereIsReply(..)

122 , messageToPayload

123 , messageToPayload

123 , payloadToMessage

124 , payloadToMessage

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

485 ncEffectDied ident reason

486 ncEffectDied ident reason

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

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

487 ncEffectSpawn from proc ref

488 ncEffectSpawn from proc ref

488− NCMsg _from (Register label pid) ->

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

489− ncEffectRegister label pid

490+ ncEffectRegister from label pid

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

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

491 ncEffectWhereIs from label

492 ncEffectWhereIs from label

492 NCMsg from (NamedSend label msg') ->

493 NCMsg from (NamedSend label msg') ->

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

587

588

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

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

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

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

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

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

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

591−ncEffectRegister label mPid =

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

592− modify' $ registryFor label ^= mPid

603+ modify' $ registryFor label ^= mPid

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

604+ liftIO $ sendMessage node

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

605+ (NodeIdentifier (localNodeId node))

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

606+ (ProcessIdentifier from)

607+ WithImplicitReconnect

608+ (RegisterReply label isOk)

596

609

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

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

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

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