e73c666 · distributed-process

git.iko.soy/distributed-process

Implement process registry

commite73c666c4063b41578e57036ca09ef4690dddae7 authorEdsko de Vries <edsko@edsko.net> dateThu 5 Jul 2012, 14:56 (14 years ago) parent2b096d2

6 files changed+219−45

compare with master

unifiedsplit

distributed-process/src/Control/Distributed/Process.hs+18 distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+6−30 distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+66 distributed-process/src/Control/Distributed/Process/Internal/Types.hs+63−13 distributed-process/src/Control/Distributed/Process/Node.hs+47−1 distributed-process/tests/TestCH.hs+19−1

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

@@ -64,6 +64,15 @@ module Control.Distributed.Process

6464 , Static

6565 , unClosure

6666 , RemoteTable

67+ -- * Registry

68+ , register

69+ , unregister

70+ , whereis

71+ , nsend

72+ , registerRemote

73+ , unregisterRemote

74+ , whereisRemote

75+ , nsendRemote

6776 -- * Auxiliary API

6877 , catch

6978 , expectTimeout

@@ -146,6 +155,15 @@ import Control.Distributed.Process.Internal.Primitives

146155 , monitorNode

147156 , monitorPort

148157 , unmonitor

158+ -- Registry

159+ , register

160+ , unregister

161+ , whereis

162+ , nsend

163+ , registerRemote

164+ , unregisterRemote

165+ , whereisRemote

166+ , nsendRemote

149167 -- Auxiliary API

150168 , catch

151169 , expectTimeout

distributed-process/src/Control/Distributed/Process/Internal/MessageT.hs+6−30

@@ -14,8 +14,8 @@ module Control.Distributed.Process.Internal.MessageT

1414 ) where

1515

1616import Data.Binary (Binary, encode)

17−import qualified Data.ByteString as BSS (ByteString, concat)

18−import qualified Data.ByteString.Lazy as BSL (toChunks, fromChunks, splitAt)

17+import qualified Data.ByteString as BSS (ByteString)

18+import qualified Data.ByteString.Lazy as BSL (toChunks)

1919import Data.Map (Map)

2020import qualified Data.Map as Map (empty)

2121import Data.Accessor (Accessor, accessor, (^=), (^.))

@@ -33,17 +33,13 @@ import Control.Distributed.Process.Internal.Types

3333 , NCMsg(NCMsg, ctrlMsgSender, ctrlMsgSignal)

3434 , DiedReason(DiedDisconnect)

3535 , ProcessSignal(Died)

36− , Message(..)

3736 , MessageT(..)

3837 , MessageState(..)

38+ , createMessage

39+ , messageToPayload

40+ , payloadToMessage

3941 )

40−import Control.Distributed.Process.Serializable

41− ( Serializable

42− , fingerprint

43− , encodeFingerprint

44− , decodeFingerprint

45− , sizeOfFingerprint

46− )

42+import Control.Distributed.Process.Serializable (Serializable)

4743import qualified Network.Transport as NT

4844 ( EndPoint

4945 , Connection

@@ -87,26 +83,6 @@ sendBinary them = sendPayload them . BSL.toChunks . encode

8783sendMessage :: (MonadIO m, Serializable a) => Identifier -> a -> MessageT m ()

8884sendMessage them = sendPayload them . messageToPayload . createMessage

8985

90−--------------------------------------------------------------------------------

91−-- Serialization/deserialization --

92−--------------------------------------------------------------------------------

93−

94−-- | Turn any serialiable term into a message

95−createMessage :: Serializable a => a -> Message

96−createMessage a = Message (fingerprint a) (encode a)

97−

98−-- | Serialize a message

99−messageToPayload :: Message -> [BSS.ByteString]

100−messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc

101−

102−-- | Deserialize a message

103−payloadToMessage :: [BSS.ByteString] -> Message

104−payloadToMessage payload = Message fp msg

105− where

106− (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)

107− $ BSL.fromChunks payload

108− fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp

109−

11086--------------------------------------------------------------------------------

11187-- Internal --

11288--------------------------------------------------------------------------------

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+66

@@ -29,6 +29,15 @@ module Control.Distributed.Process.Internal.Primitives

2929 , unlink

3030 , monitor

3131 , unmonitor

32+ -- * Registry

33+ , register

34+ , unregister

35+ , whereis

36+ , nsend

37+ , registerRemote

38+ , unregisterRemote

39+ , whereisRemote

40+ , nsendRemote

3241 -- * Auxiliary API

3342 , catch

3443 , expectTimeout

@@ -91,6 +100,8 @@ import Control.Distributed.Process.Internal.Types

91100 , DidUnlinkProcess(..)

92101 , DidUnlinkNode(..)

93102 , DidUnlinkPort(..)

103+ , WhereIsReply(..)

104+ , createMessage

94105 )

95106import Control.Distributed.Process.Internal.MessageT

96107 ( sendMessage

@@ -351,6 +362,60 @@ unlinkPortAsync :: SendPort a -> Process ()

351362unlinkPortAsync (SendPort cid) =

352363 sendCtrlMsg Nothing . Unlink $ SendPortIdentifier cid

353364

365+--------------------------------------------------------------------------------

366+-- Registry --

367+--------------------------------------------------------------------------------

368+

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

370+--

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

372+register :: String -> ProcessId -> Process ()

373+register label pid =

374+ sendCtrlMsg Nothing (Register label (Just pid))

375+

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

377+--

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

379+registerRemote :: NodeId -> String -> ProcessId -> Process ()

380+registerRemote nid label pid =

381+ sendCtrlMsg (Just nid) (Register label (Just pid))

382+

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

384+unregister :: String -> Process ()

385+unregister label =

386+ sendCtrlMsg Nothing (Register label Nothing)

387+

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

389+unregisterRemote :: NodeId -> String -> Process ()

390+unregisterRemote nid label =

391+ sendCtrlMsg (Just nid) (Register label Nothing)

392+

393+-- | Query the local process registry (synchronous).

394+whereis :: String -> Process (Maybe ProcessId)

395+whereis label = do

396+ sendCtrlMsg Nothing (WhereIs label)

397+ receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')

398+ (\(WhereIsReply _ mPid) -> return mPid)

399+ ]

400+

401+-- | Query a remote process registry (synchronous)

402+whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

403+whereisRemote nid label = do

404+ sendCtrlMsg (Just nid) (WhereIs label)

405+ receiveWait [ matchIf (\(WhereIsReply label' _) -> label == label')

406+ (\(WhereIsReply _ mPid) -> return mPid)

407+ ]

408+

409+-- | Named send to a process in the local registry (asynchronous)

410+nsend :: Serializable a => String -> a -> Process ()

411+nsend label msg =

412+ sendCtrlMsg Nothing (NamedSend label (createMessage msg))

413+

414+-- | Named send to a process in a remote registry (asynchronous)

415+nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

416+nsendRemote nid label msg =

417+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

418+

354419--------------------------------------------------------------------------------

355420-- Auxiliary functions --

356421--------------------------------------------------------------------------------

@@ -399,3 +464,4 @@ sendCtrlMsg mNid signal = do

399464 liftIO $ writeChan ctrlChan msg

400465 Just nid ->

401466 procMsg $ sendBinary (NodeIdentifier nid) msg

467+

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

@@ -33,6 +33,9 @@ module Control.Distributed.Process.Internal.Types

3333 , RuntimeSerializableSupport(..)

3434 -- * Messages

3535 , Message(..)

36+ , createMessage

37+ , messageToPayload

38+ , payloadToMessage

3639 -- * Node controller user-visible data types

3740 , MonitorRef(..)

3841 , ProcessMonitorNotification(..)

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

4851 , DidUnlinkPort(..)

4952 , SpawnRef(..)

5053 , DidSpawn(..)

54+ , WhereIsReply(..)

5155 -- * Node controller internal data types

5256 , NCMsg(..)

5357 , ProcessSignal(..)

@@ -73,11 +77,16 @@ module Control.Distributed.Process.Internal.Types

7377import Data.Map (Map)

7478import Data.Int (Int32)

7579import Data.Typeable (Typeable, TypeRep)

76−import Data.Binary (Binary(put, get), putWord8, getWord8)

77−import Data.ByteString.Lazy (ByteString)

80+import Data.Binary (Binary(put, get), putWord8, getWord8, encode)

81+import qualified Data.ByteString as BSS (ByteString, concat)

82+import qualified Data.ByteString.Lazy as BSL

83+ ( ByteString

84+ , toChunks

85+ , splitAt

86+ , fromChunks

87+ )

7888import Data.Accessor (Accessor, accessor)

7989import qualified Data.Accessor.Container as DAC (mapMaybe)

80−import qualified Data.ByteString.Lazy as BSL (ByteString)

8190import Control.Category ((>>>))

8291import Control.Exception (Exception)

8392import Control.Concurrent (ThreadId)

@@ -90,7 +99,14 @@ import Control.Monad.Reader (MonadReader(..), ReaderT)

9099import Control.Monad.IO.Class (MonadIO)

91100import Control.Monad.State (MonadState, StateT)

92101import qualified Control.Monad.Trans.Class as Trans (lift)

93−import Control.Distributed.Process.Serializable (Fingerprint, Serializable)

102+import Control.Distributed.Process.Serializable

103+ ( Fingerprint

104+ , Serializable

105+ , fingerprint

106+ , encodeFingerprint

107+ , sizeOfFingerprint

108+ , decodeFingerprint

109+ )

94110import Control.Distributed.Process.Internal.CQueue (CQueue)

95111import Control.Distributed.Process.Internal.Dynamic (Dynamic)

96112

@@ -255,7 +271,7 @@ newtype Static a = Static StaticLabel

255271 deriving (Typeable, Show)

256272

257273-- | A closure is a static value and an encoded environment

258−data Closure a = Closure (Static (ByteString -> a)) ByteString

274+data Closure a = Closure (Static (BSL.ByteString -> a)) BSL.ByteString

259275 deriving (Typeable, Show)

260276

261277-- | Used to fake 'static' (see paper)

@@ -294,6 +310,25 @@ data Message = Message

294310 , messageEncoding :: BSL.ByteString

295311 }

296312

313+instance Show Message where

314+ show = show . messageEncoding

315+

316+-- | Turn any serialiable term into a message

317+createMessage :: Serializable a => a -> Message

318+createMessage a = Message (fingerprint a) (encode a)

319+

320+-- | Serialize a message

321+messageToPayload :: Message -> [BSS.ByteString]

322+messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc

323+

324+-- | Deserialize a message

325+payloadToMessage :: [BSS.ByteString] -> Message

326+payloadToMessage payload = Message fp msg

327+ where

328+ (encFp, msg) = BSL.splitAt (fromIntegral sizeOfFingerprint)

329+ $ BSL.fromChunks payload

330+ fp = decodeFingerprint . BSS.concat . BSL.toChunks $ encFp

331+

297332--------------------------------------------------------------------------------

298333-- Node controller user-visible data types --

299334--------------------------------------------------------------------------------

@@ -376,10 +411,14 @@ newtype DidUnlinkPort = DidUnlinkPort SendPortId

376411newtype SpawnRef = SpawnRef Int32

377412 deriving (Show, Binary, Typeable, Eq)

378413

379−-- | (Asynchronius) reply from spawn

414+-- | (Asynchronius) reply from 'spawn'

380415data DidSpawn = DidSpawn SpawnRef ProcessId

381416 deriving (Show, Typeable)

382417

418+-- | (Asynchronous) reply from 'whereis'

419+data WhereIsReply = WhereIsReply String (Maybe ProcessId)

420+ deriving (Show, Typeable)

421+

383422--------------------------------------------------------------------------------

384423-- Node controller internal data types --

385424--------------------------------------------------------------------------------

@@ -399,6 +438,9 @@ data ProcessSignal =

399438 | Unmonitor MonitorRef

400439 | Died Identifier DiedReason

401440 | Spawn (Closure (Process ())) SpawnRef

441+ | WhereIs String

442+ | Register String (Maybe ProcessId) -- Nothing to unregister

443+ | NamedSend String Message

402444 deriving Show

403445

404446--------------------------------------------------------------------------------

@@ -434,12 +476,15 @@ instance Binary MonitorRef where

434476 get = MonitorRef <$> get <*> get

435477

436478instance Binary ProcessSignal where

437− put (Link pid) = putWord8 0 >> put pid

438− put (Unlink pid) = putWord8 1 >> put pid

439− put (Monitor ref) = putWord8 2 >> put ref

440− put (Unmonitor ref) = putWord8 3 >> put ref

441− put (Died who reason) = putWord8 4 >> put who >> put reason

442− put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

479+ put (Link pid) = putWord8 0 >> put pid

480+ put (Unlink pid) = putWord8 1 >> put pid

481+ put (Monitor ref) = putWord8 2 >> put ref

482+ put (Unmonitor ref) = putWord8 3 >> put ref

483+ put (Died who reason) = putWord8 4 >> put who >> put reason

484+ put (Spawn proc ref) = putWord8 5 >> put proc >> put ref

485+ put (WhereIs label) = putWord8 6 >> put label

486+ put (Register label pid) = putWord8 7 >> put label >> put pid

487+ put (NamedSend label msg) = putWord8 8 >> put label >> put (messageToPayload msg)

443488 get = do

444489 header <- getWord8

445490 case header of

@@ -449,6 +494,9 @@ instance Binary ProcessSignal where

449494 3 -> Unmonitor <$> get

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

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

497+ 6 -> WhereIs <$> get

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

499+ 8 -> NamedSend <$> get <*> (payloadToMessage <$> get)

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

453501

454502instance Binary DiedReason where

@@ -529,7 +577,9 @@ instance Binary StaticLabel where

529577 15 -> return CpApply

530578 _ -> fail "StaticLabel.get: invalid"

531579

532−

580+instance Binary WhereIsReply where

581+ put (WhereIsReply label mPid) = put label >> put mPid

582+ get = WhereIsReply <$> get <*> get

533583

534584--------------------------------------------------------------------------------

535585-- Accessors --

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

@@ -61,7 +61,7 @@ import qualified Network.Transport as NT

6161 , ConnectionId

6262 )

6363import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

64−import qualified Data.Accessor.Container as DAC (mapDefault)

64+import qualified Data.Accessor.Container as DAC (mapDefault, mapMaybe)

6565import System.Random (randomIO)

6666import Control.Distributed.Process.Internal.Types

6767 ( RemoteTable

@@ -101,11 +101,14 @@ import Control.Distributed.Process.Internal.Types

101101 , nodeOf

102102 , SendPortId(..)

103103 , typedChannelWithId

104+ , WhereIsReply(..)

105+ , messageToPayload

104106 )

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

106108import Control.Distributed.Process.Internal.MessageT

107109 ( sendBinary

108110 , sendMessage

111+ , sendPayload

109112 , getLocalNode

110113 , runMessageT

111114 , payloadToMessage

@@ -343,6 +346,8 @@ data NCState = NCState

343346 _links :: Map Identifier (Set ProcessId)

344347 -- Mapping from remote processes to monitoring local processes

345348 , _monitors :: Map Identifier (Set (ProcessId, MonitorRef))

349+ -- Process registry

350+ , _registry :: Map String ProcessId

346351 }

347352

348353newtype NC a = NC { unNC :: StateT NCState (MessageT IO) a }

@@ -354,6 +359,7 @@ ncMsg = NC . Trans.lift

354359initNCState :: NCState

355360initNCState = NCState { _links = Map.empty

356361 , _monitors = Map.empty

362+ , _registry = Map.empty

357363 }

358364

359365--------------------------------------------------------------------------------

@@ -387,6 +393,12 @@ nodeController = do

387393 ncEffectDied ident reason

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

389395 ncEffectSpawn from proc ref

396+ NCMsg _from (Register label pid) ->

397+ ncEffectRegister label pid

398+ NCMsg (ProcessIdentifier from) (WhereIs label) ->

399+ ncEffectWhereIs from label

400+ NCMsg from (NamedSend label msg') ->

401+ ncEffectNamedSend from label msg'

390402 unexpected ->

391403 error $ "nodeController: unexpected message " ++ show unexpected

392404

@@ -468,6 +480,31 @@ ncEffectSpawn pid cProc ref = do

468480 pid' <- liftIO $ forkProcess node proc

469481 ncMsg $ sendMessage (ProcessIdentifier pid) (DidSpawn ref pid')

470482

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

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

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

486+ncEffectRegister label mPid =

487+ modify $ registryFor label ^= mPid

488+ -- An acknowledgement is not necessary. If we want a synchronous register,

489+ -- it suffices to send a whereis requiry immediately after the register

490+ -- (that may not suffice if we do decide for unreliable messaging instead)

491+

492+-- Unified semantics does not explicitly describe 'whereis'

493+ncEffectWhereIs :: ProcessId -> String -> NC ()

494+ncEffectWhereIs from label = do

495+ mPid <- gets (^. registryFor label)

496+ ncMsg $ sendMessage (ProcessIdentifier from) (WhereIsReply label mPid)

497+

498+-- [Unified: Table 14]

499+ncEffectNamedSend :: Identifier -> String -> Message -> NC ()

500+ncEffectNamedSend _from label msg = do

501+ mPid <- gets (^. registryFor label)

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

503+ -- TODO: messages don't carry a "from", but if they do, we should set it

504+ -- to be the 'from' of the original named send, not this node controller

505+ forM_ mPid $ \pid ->

506+ ncMsg $ sendPayload (ProcessIdentifier pid) (messageToPayload msg)

507+

471508--------------------------------------------------------------------------------

472509-- Auxiliary --

473510--------------------------------------------------------------------------------

@@ -507,6 +544,9 @@ destNid (Unlink ident) = Just $ nodeOf ident

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

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

509546destNid (Spawn _ _) = Nothing

547+destNid (Register _ _) = Nothing

548+destNid (WhereIs _) = Nothing

549+destNid (NamedSend _ _) = Nothing

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

511551-- then when a local process dies the monitoring/linking machinery will take

512552-- care of notifying remote nodes

@@ -576,12 +616,18 @@ links = accessor _links (\ls st -> st { _links = ls })

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

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

578618

619+registry :: Accessor NCState (Map String ProcessId)

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

621+

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

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

581624

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

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

584627

628+registryFor :: String -> Accessor NCState (Maybe ProcessId)

629+registryFor ident = registry >>> DAC.mapMaybe ident

630+

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

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

587633-- should not.

distributed-process/tests/TestCH.hs+19−1

@@ -524,6 +524,23 @@ testMonitorChannel transport = do

524524

525525 takeMVar gotNotification

526526

527+testRegistry :: NT.Transport -> IO ()

528+testRegistry transport = do

529+ node <- newLocalNode transport initRemoteTable

530+

531+ pingServer <- forkProcess node ping

532+

533+ runProcess node $ do

534+ register "ping" pingServer

535+ Just pid <- whereis "ping"

536+ True <- return $ pingServer == pid

537+ us <- getSelfPid

538+ nsend "ping" (Pong us)

539+ Ping pid' <- expect

540+ True <- return $ pingServer == pid'

541+ return ()

542+

543+

527544main :: IO ()

528545main = do

529546 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

@@ -535,7 +552,8 @@ main = do

535552 , ("SendToTerminated", testSendToTerminated transport)

536553 , ("TypedChannnels", testTypedChannels transport)

537554 , ("MergeChannels", testMergeChannels transport)

538− , ("TestTerminate", testTerminate transport)

555+ , ("Terminate", testTerminate transport)

556+ , ("Registry", testRegistry transport)

539557 -- Monitoring processes

540558 --

541559 -- The "missing" combinations in the list below don't make much sense, as