588e202 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Merge 'process-info' - fixes #111

588e202Tim Watson14 years ago

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

compare with master

unifiedsplit

@@ -125,7 +125,31 @@ Test-Suite TestClosure

125 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

126 HS-Source-Dirs: tests

127

128+Test-Suite TestStats

129+ Type: exitcode-stdio-1.0

130+ Main-Is: TestStats.hs

131+ Build-Depends: base >= 4.4 && < 5,

132+ random >= 1.0 && < 1.1,

133+ ansi-terminal >= 0.5 && < 0.6,

134+ containers >= 0.4 && < 0.6,

135+ stm >= 2.3 && < 2.5,

136+ distributed-process,

137+ network-transport >= 0.3 && < 0.4,

138+ network-transport-tcp >= 0.3 && < 0.4,

139+ binary >= 0.5 && < 0.7,

140+ network >= 2.3 && < 2.5,

141+ HUnit >= 1.2 && < 1.3,

142+ test-framework >= 0.6 && < 0.9,

143+ test-framework-hunit >= 0.2.0 && < 0.4

144+ Extensions: CPP,

145+ ScopedTypeVariables,

146+ DeriveDataTypeable,

147+ GeneralizedNewtypeDeriving

148+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

149+ HS-Source-Dirs: tests

150+

151+

128−Executable distributed-process-throughput

152+Executable distributed-process-throughput

129 if flag(benchmarks)

153 if flag(benchmarks)

130 Build-Depends: base >= 4.4 && < 5,

154 Build-Depends: base >= 4.4 && < 5,

131 distributed-process,

155 distributed-process,

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

@@ -52,6 +52,8 @@ module Control.Distributed.Process

52 , SpawnRef

53 , getSelfPid

54 , getSelfNode

55+ , ProcessInfo(..)

56+ , getProcessInfo

55 -- * Monitoring and linking

57 -- * Monitoring and linking

56 , link

58 , link

57 , linkNode

59 , linkNode

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

198 , kill

200 , kill

199 , getSelfPid

201 , getSelfPid

200 , getSelfNode

202 , getSelfNode

203+ , ProcessInfo(..)

204+ , getProcessInfo

201 -- Monitoring and linking

205 -- Monitoring and linking

202 , link

206 , link

203 , linkNode

207 , linkNode

distributed-process/src/Control/Distributed/Process/Internal/CQueue.hs+1−1

@@ -52,7 +52,7 @@ newCQueue = CQueue <$> newMVar Nil <*> atomically newTChan

52--

53-- Enqueue is strict.

54enqueue :: CQueue a -> a -> IO ()

55−enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a

55+enqueue (CQueue _arrived incoming) !a = atomically $ writeTChan incoming a

57data BlockSpec =

58 NonBlocking

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+21−1

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

32 , ProcessExitException(..)

33 , getSelfPid

34 , getSelfNode

35+ , ProcessInfo(..)

36+ , getProcessInfo

35 -- * Monitoring and linking

37 -- * Monitoring and linking

36 , link

38 , link

37 , unlink

39 , unlink

@@ -144,6 +146,8 @@ import Control.Distributed.Process.Internal.Types

144 , WhereIsReply(..)

146 , WhereIsReply(..)

145 , RegisterReply(..)

147 , RegisterReply(..)

146 , ProcessRegistrationException(..)

148 , ProcessRegistrationException(..)

149+ , ProcessInfo(..)

150+ , ProcessInfoNone(..)

147 , createMessage

151 , createMessage

148 , runLocalProcess

152 , runLocalProcess

149 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

153 , ImplicitReconnect(WithImplicitReconnect, NoImplicitReconnect)

@@ -400,6 +404,22 @@ getSelfPid = processId <$> ask

400getSelfNode :: Process NodeId

404getSelfNode :: Process NodeId

401getSelfNode = localNodeId . processNode <$> ask

405getSelfNode = localNodeId . processNode <$> ask

402

406

407+-- | Get information about the specified process

408+getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)

409+getProcessInfo pid =

410+ let them = processNodeId pid in do

411+ us <- getSelfNode

412+ dest <- mkNode them us

413+ sendCtrlMsg dest $ GetInfo pid

414+ receiveWait [

415+ match (\(p :: ProcessInfo) -> return $ Just p)

416+ , match (\(_ :: ProcessInfoNone) -> return Nothing)

417+ ]

418+ where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)

419+ mkNode them us = case them == us of

420+ True -> return Nothing

421+ _ -> return $ Just them

422+

403--------------------------------------------------------------------------------

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

404-- Monitoring and linking --

424-- Monitoring and linking --

405--------------------------------------------------------------------------------

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

@@ -519,7 +539,7 @@ try p = do

519 lproc <- ask

539 lproc <- ask

520 liftIO $ Ex.try (runLocalProcess lproc p)

540 liftIO $ Ex.try (runLocalProcess lproc p)

521

541

522−-- | Lift 'Control.Exception.mask'

542+-- | Lift 'Control.Exception.mask'

523mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b

543mask :: ((forall a. Process a -> Process a) -> Process b) -> Process b

524mask p = do

544mask p = do

525 lproc <- ask

545 lproc <- ask

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

@@ -49,6 +49,8 @@ module Control.Distributed.Process.Internal.Types

49 , DidSpawn(..)

50 , WhereIsReply(..)

51 , RegisterReply(..)

52+ , ProcessInfo(..)

53+ , ProcessInfoNone(..)

52 -- * Node controller internal data types

54 -- * Node controller internal data types

53 , NCMsg(..)

55 , NCMsg(..)

54 , ProcessSignal(..)

56 , ProcessSignal(..)

@@ -420,6 +422,29 @@ data WhereIsReply = WhereIsReply String (Maybe ProcessId)

420data RegisterReply = RegisterReply String Bool

422data RegisterReply = RegisterReply String Bool

421 deriving (Show, Typeable)

423 deriving (Show, Typeable)

422

424

425+data ProcessInfo = ProcessInfo {

426+ infoNode :: NodeId

427+ , infoRegisteredNames :: [String]

428+ , infoMessageQueueLength :: Maybe Int

429+ , infoMonitors :: [(ProcessId, MonitorRef)]

430+ , infoLinks :: [ProcessId]

431+ } deriving (Show, Eq, Typeable)

432+

433+instance Binary ProcessInfo where

434+ get = ProcessInfo <$> get <*> get <*> get <*> get <*> get

435+ put pInfo = put (infoNode pInfo)

436+ >> put (infoRegisteredNames pInfo)

437+ >> put (infoMessageQueueLength pInfo)

438+ >> put (infoMonitors pInfo)

439+ >> put (infoLinks pInfo)

440+

441+data ProcessInfoNone = ProcessInfoNone DiedReason

442+ deriving (Show, Typeable)

443+

444+instance Binary ProcessInfoNone where

445+ get = ProcessInfoNone <$> get

446+ put (ProcessInfoNone r) = put r

447+

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

448--------------------------------------------------------------------------------

424-- Node controller internal data types --

449-- Node controller internal data types --

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

450--------------------------------------------------------------------------------

@@ -444,6 +469,7 @@ data ProcessSignal =

444 | NamedSend !String !Message

469 | NamedSend !String !Message

445 | Kill !ProcessId !String

470 | Kill !ProcessId !String

446 | Exit !ProcessId !Message

471 | Exit !ProcessId !Message

472+ | GetInfo !ProcessId

447 deriving Show

473 deriving Show

448

474

449--------------------------------------------------------------------------------

475--------------------------------------------------------------------------------

@@ -490,6 +516,7 @@ instance Binary ProcessSignal where

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

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

491 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

517 put (Kill pid reason) = putWord8 9 >> put pid >> put reason

492 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

518 put (Exit pid reason) = putWord8 10 >> put pid >> put (messageToPayload reason)

519+ put (GetInfo about) = putWord8 30 >> put about

493 get = do

520 get = do

494 header <- getWord8

521 header <- getWord8

495 case header of

522 case header of

@@ -504,6 +531,7 @@ instance Binary ProcessSignal where

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

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

505 9 -> Kill <$> get <*> get

532 9 -> Kill <$> get <*> get

506 10 -> Exit <$> get <*> (payloadToMessage <$> get)

533 10 -> Exit <$> get <*> (payloadToMessage <$> get)

534+ 30 -> GetInfo <$> get

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

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

508

536

509instance Binary DiedReason where

537instance Binary DiedReason where

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

@@ -29,9 +29,16 @@ import qualified Data.Map as Map

29 , partitionWithKey

30 , elems

31 , filterWithKey

32+ , foldlWithKey

32 )

33 )

33import Data.Set (Set)

34import Data.Set (Set)

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

35+import qualified Data.Set as Set

36+ ( empty

37+ , insert

38+ , delete

39+ , member

40+ , toList

41+ )

35import Data.Foldable (forM_)

42import Data.Foldable (forM_)

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

43import Data.Maybe (isJust, isNothing, catMaybes)

37import Data.Typeable (Typeable)

44import Data.Typeable (Typeable)

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

118 , TypedChannel(..)

125 , TypedChannel(..)

119 , Identifier(..)

126 , Identifier(..)

120 , nodeOf

127 , nodeOf

128+ , ProcessInfo(..)

129+ , ProcessInfoNone(..)

121 , SendPortId(..)

130 , SendPortId(..)

122 , typedChannelWithId

131 , typedChannelWithId

123 , RegisterReply(..)

132 , RegisterReply(..)

@@ -517,6 +526,8 @@ nodeController = do

517 ncEffectKill from to reason

526 ncEffectKill from to reason

518 NCMsg (ProcessIdentifier from) (Exit to reason) ->

527 NCMsg (ProcessIdentifier from) (Exit to reason) ->

519 ncEffectExit from to reason

528 ncEffectExit from to reason

529+ NCMsg (ProcessIdentifier from) (GetInfo pid) ->

530+ ncEffectGetInfo from pid

520 unexpected ->

531 unexpected ->

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

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

522

533

@@ -737,6 +748,53 @@ ncEffectExit from to reason = do

737 when (isLocal node (ProcessIdentifier to)) $

748 when (isLocal node (ProcessIdentifier to)) $

738 throwException to $ ProcessExitException from reason

749 throwException to $ ProcessExitException from reason

739

750

751+-- [Issue #89]

752+ncEffectGetInfo :: ProcessId -> ProcessId -> NC ()

753+ncEffectGetInfo from pid =

754+ let lpid = processLocalId pid

755+ them = (ProcessIdentifier pid)

756+ in do

757+ node <- ask

758+ mProc <- liftIO $

759+ withMVar (localState node) $ return . (^. localProcessWithId lpid)

760+ case mProc of

761+ Nothing -> dispatch (isLocal node (ProcessIdentifier from))

762+ from node (ProcessInfoNone DiedUnknownId)

763+ Just _ -> do

764+ itsLinks <- gets (^. linksFor them)

765+ itsMons <- gets (^. monitorsFor them)

766+ registered <- gets (^. registeredHere)

767+

768+ let reg = registeredNames registered

769+ dispatch (isLocal node (ProcessIdentifier from))

770+ from

771+ node

772+ ProcessInfo {

773+ infoNode = (processNodeId pid)

774+ , infoRegisteredNames = reg

775+ -- we cannot populate this field

776+ , infoMessageQueueLength = Nothing

777+ , infoMonitors = Set.toList itsMons

778+ , infoLinks = Set.toList itsLinks

779+ }

780+ where dispatch :: (Serializable a, Show a)

781+ => Bool

782+ -> ProcessId

783+ -> LocalNode

784+ -> a

785+ -> NC ()

786+ dispatch True dest _ pInfo = postAsMessage dest $ pInfo

787+ dispatch False dest node pInfo = do

788+ liftIO $ sendMessage node

789+ (NodeIdentifier (localNodeId node))

790+ (ProcessIdentifier dest)

791+ WithImplicitReconnect

792+ pInfo

793+

794+ registeredNames = Map.foldlWithKey (\ks k v -> if v == pid

795+ then (k:ks)

796+ else ks) []

797+

740--------------------------------------------------------------------------------

798--------------------------------------------------------------------------------

741-- Auxiliary --

799-- Auxiliary --

742--------------------------------------------------------------------------------

800--------------------------------------------------------------------------------

@@ -788,6 +846,7 @@ destNid (NamedSend _ _) = Nothing

788destNid (Died _ _) = Nothing

846destNid (Died _ _) = Nothing

789destNid (Kill pid _) = Just $ processNodeId pid

847destNid (Kill pid _) = Just $ processNodeId pid

790destNid (Exit pid _) = Just $ processNodeId pid

848destNid (Exit pid _) = Just $ processNodeId pid

849+destNid (GetInfo pid) = Just $ processNodeId pid

791

850

792-- | Check if a process is local to our own node

851-- | Check if a process is local to our own node

793isLocal :: LocalNode -> Identifier -> Bool

852isLocal :: LocalNode -> Identifier -> Bool

@@ -827,7 +886,8 @@ postAsMessage :: Serializable a => ProcessId -> a -> NC ()

827postAsMessage pid = postMessage pid . createMessage

886postAsMessage pid = postMessage pid . createMessage

828

887

829postMessage :: ProcessId -> Message -> NC ()

888postMessage :: ProcessId -> Message -> NC ()

889+postMessage pid msg = do

830−postMessage pid msg = withLocalProc pid $ \p -> enqueue (processQueue p) msg

890+ withLocalProc pid $ \p -> enqueue (processQueue p) msg

831

891

832throwException :: Exception e => ProcessId -> e -> NC ()

892throwException :: Exception e => ProcessId -> e -> NC ()

833throwException pid e = withLocalProc pid $ \p ->

893throwException pid e = withLocalProc pid $ \p ->

distributed-process/tests/TestStats.hsnew file+162

@@ -0,0 +1,162 @@

1+{-# LANGUAGE DeriveDataTypeable #-}

2+{-# OPTIONS_GHC -fno-warn-orphans #-}

3+module Main where

5+import Prelude hiding (catch, log)

6+import Test.Framework

7+ ( Test

8+ , defaultMain

9+ , testGroup

10+ )

11+import Network.Transport.TCP

12+import Control.Distributed.Process

13+import Control.Distributed.Process.Node

14+ ( forkProcess

15+ , newLocalNode

16+ , initRemoteTable

17+ , closeLocalNode

18+ , LocalNode)

19+import Control.Concurrent.MVar

20+ ( MVar

21+ , newEmptyMVar

22+ , putMVar

23+ , takeMVar

24+ )

25+import Data.Binary()

26+import Data.Typeable()

27+import Test.HUnit (Assertion)

28+import Test.HUnit.Base (assertBool)

29+import Test.Framework.Providers.HUnit (testCase)

30+

31+-- these utilities have been cribbed from distributed-process-platform

32+-- we should really find a way to share them...

33+

34+-- | A mutable cell containing a test result.

35+type TestResult a = MVar a

36+

37+delayedAssertion :: (Eq a) => String -> LocalNode -> a ->

38+ (TestResult a -> Process ()) -> Assertion

39+delayedAssertion note localNode expected testProc = do

40+ result <- newEmptyMVar

41+ _ <- forkProcess localNode $ testProc result

42+ assertComplete note result expected

43+

44+assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()

45+assertComplete msg mv a = do

46+ b <- takeMVar mv

47+ assertBool msg (a == b)

48+

49+stash :: TestResult a -> a -> Process ()

50+stash mvar x = liftIO $ putMVar mvar x

51+

52+------

53+

54+testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()

55+testLocalDeadProcessInfo result = do

56+ pid <- spawnLocal $ do "finish" <- expect; return ()

57+ mref <- monitor pid

58+ send pid "finish"

59+ _ <- receiveWait [

60+ matchIf (\(ProcessMonitorNotification ref' pid' r) ->

61+ ref' == mref && pid' == pid && r == DiedNormal)

62+ (\p -> return p)

63+ ]

64+ getProcessInfo pid >>= stash result

65+

66+testLocalLiveProcessInfo :: TestResult Bool -> Process ()

67+testLocalLiveProcessInfo result = do

68+ self <- getSelfPid

69+ node <- getSelfNode

70+ register "foobar" self

71+

72+ mon <- liftIO $ newEmptyMVar

73+ -- TODO: we can't get the mailbox's length

74+ -- mapM (send self) ["hello", "there", "mr", "process"]

75+ pid <- spawnLocal $ do

76+ link self

77+ mRef <- monitor self

78+ stash mon mRef

79+ "die" <- expect

80+ return ()

81+

82+ monRef <- liftIO $ takeMVar mon

83+

84+ mpInfo <- getProcessInfo self

85+ case mpInfo of

86+ Nothing -> stash result False

87+ Just p -> verifyPInfo p pid monRef node

88+ where verifyPInfo :: ProcessInfo

89+ -> ProcessId

90+ -> MonitorRef

91+ -> NodeId

92+ -> Process ()

93+ verifyPInfo pInfo pid mref node =

94+ stash result $ infoNode pInfo == node &&

95+ infoLinks pInfo == [pid] &&

96+ infoMonitors pInfo == [(pid, mref)] &&

97+-- infoMessageQueueLength pInfo == Just 4 &&

98+ infoRegisteredNames pInfo == ["foobar"]

99+

100+testRemoteLiveProcessInfo :: LocalNode -> Assertion

101+testRemoteLiveProcessInfo node1 = do

102+ serverAddr <- liftIO $ newEmptyMVar :: IO (MVar ProcessId)

103+ liftIO $ launchRemote serverAddr

104+ serverPid <- liftIO $ takeMVar serverAddr

105+ withActiveRemote node1 $ \result -> do

106+ self <- getSelfPid

107+ link serverPid

108+ -- our send op shouldn't overtake link or monitor requests AFAICT

109+ -- so a little table tennis should get us synchronised properly

110+ send serverPid (self, "ping")

111+ "pong" <- expect

112+ pInfo <- getProcessInfo serverPid

113+ stash result $ pInfo /= Nothing

114+ where

115+ launchRemote :: MVar ProcessId -> IO ()

116+ launchRemote locMV = do

117+ node2 <- liftIO $ mkNode "8082"

118+ _ <- liftIO $ forkProcess node2 $ do

119+ self <- getSelfPid

120+ liftIO $ putMVar locMV self

121+ _ <- receiveWait [

122+ match (\(pid, "ping") -> send pid "pong")

123+ ]

124+ "stop" <- expect

125+ return ()

126+ return ()

127+

128+ withActiveRemote :: LocalNode

129+ -> ((TestResult Bool -> Process ()) -> Assertion)

130+ withActiveRemote n = do

131+ a <- delayedAssertion "getProcessInfo remotePid failed" n True

132+ return a

133+

134+tests :: LocalNode -> IO [Test]

135+tests node1 = do

136+ return [

137+ testGroup "Process Info" [

138+ testCase "testLocalDeadProcessInfo"

139+ (delayedAssertion

140+ "expected dead process-info to be ProcessInfoNone"

141+ node1 (Nothing) testLocalDeadProcessInfo)

142+ , testCase "testLocalLiveProcessInfo"

143+ (delayedAssertion

144+ "expected process-info to be correctly populated"

145+ node1 True testLocalLiveProcessInfo)

146+ , testCase "testRemoveLiveProcessInfo"

147+ (testRemoteLiveProcessInfo node1)

148+ ] ]

149+

150+mkNode :: String -> IO LocalNode

151+mkNode port = do

152+ Right (transport1, _) <- createTransportExposeInternals

153+ "127.0.0.1" port defaultTCPParameters

154+ newLocalNode transport1 initRemoteTable

155+

156+main :: IO ()

157+main = do

158+ node1 <- mkNode "8081"

159+ testData <- tests node1

160+ defaultMain testData

161+ closeLocalNode node1

162+ return ()