c094e66 · distributed-process

distributed-process/distributed-process.cabal+22

compare with master

unifiedsplit

@@ -123,8 +123,30 @@ Test-Suite TestClosure

123 Extensions: CPP,

124 ScopedTypeVariables

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+ distributed-process,

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

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

137+ binary >= 0.5 && < 0.7,

138+ network >= 2.3 && < 2.5,

139+ HUnit >= 1.2 && < 1.3,

140+ test-framework >= 0.6 && < 0.7,

141+ test-framework-hunit >= 0.2 && < 0.3

142+ Extensions: CPP,

143+ ScopedTypeVariables,

144+ DeriveDataTypeable,

145+ GeneralizedNewtypeDeriving

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

126 HS-Source-Dirs: tests

147 HS-Source-Dirs: tests

127

148

149+

128Executable distributed-process-throughput

150Executable distributed-process-throughput

129 if flag(benchmarks)

151 if flag(benchmarks)

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

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

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

@@ -199,6 +201,8 @@ import Control.Distributed.Process.Internal.Primitives

199 , kill

201 , kill

200 , getSelfPid

202 , getSelfPid

201 , getSelfNode

203 , getSelfNode

204+ , ProcessInfo(..)

205+ , getProcessInfo

202 -- Monitoring and linking

206 -- Monitoring and linking

203 , link

207 , link

204 , linkNode

208 , linkNode

distributed-process/src/Control/Distributed/Process/Internal/Primitives.hs+6−3

@@ -86,6 +86,7 @@ import Prelude hiding (catch)

86import Data.Binary (decode)

87import Data.Time.Clock (getCurrentTime)

88import Data.Time.Format (formatTime)

89+import System.IO (hPutStrLn, stderr)

89import System.Locale (defaultTimeLocale)

90import System.Locale (defaultTimeLocale)

90import System.Timeout (timeout)

91import System.Timeout (timeout)

91import Control.Monad (when)

92import Control.Monad (when)

@@ -407,14 +408,16 @@ getSelfNode = localNodeId . processNode <$> ask

407

408

408-- | Get information about the specified process

409-- | Get information about the specified process

409getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)

410getProcessInfo :: ProcessId -> Process (Maybe ProcessInfo)

410−getProcessInfo pid =

411+getProcessInfo pid =

411 let them = processNodeId pid in do

412 let them = processNodeId pid in do

412 us <- getSelfNode

413 us <- getSelfNode

413 dest <- mkNode them us

414 dest <- mkNode them us

415+ liftIO . putStrLn $ "sending getInfo req to " ++ (show dest)

416+

414 sendCtrlMsg dest $ GetInfo pid

417 sendCtrlMsg dest $ GetInfo pid

415 receiveWait [

418 receiveWait [

416 match (\x@(ProcessInfo _ _ _ _ _) -> return (Just x))

419 match (\x@(ProcessInfo _ _ _ _ _) -> return (Just x))

417− , match (\(ProcessInfoNone _) -> return Nothing)

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

418 ]

421 ]

419 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)

422 where mkNode :: NodeId -> NodeId -> Process (Maybe NodeId)

420 mkNode them us = case them == us of

423 mkNode them us = case them == us of

@@ -540,7 +543,7 @@ try p = do

540 lproc <- ask

543 lproc <- ask

541 liftIO $ Ex.try (runLocalProcess lproc p)

544 liftIO $ Ex.try (runLocalProcess lproc p)

542

545

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

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

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

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

545mask p = do

548mask p = do

546 lproc <- ask

549 lproc <- ask

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

@@ -452,7 +452,7 @@ toMarshalledPInfoKey ProcessInfoMessageQueueLength = SerializedProcessInfoKey 4

452toMarshalledPInfoKey ProcessInfoMonitors = SerializedProcessInfoKey 5

453toMarshalledPInfoKey ProcessInfoLinks = SerializedProcessInfoKey 6

454

455−-- | Convert 'SerializedProcessInfoKey' back into 'ProcessInfoKey'

455+-- | Convert 'SerializedProcessInfoKey' back into 'ProcessInfoKey'

456toPInfoKey :: SerializedProcessInfoKey -> ProcessInfoKey

457toPInfoKey (SerializedProcessInfoKey n)

458 | n == 1 = ProcessInfoAll

@@ -471,7 +471,7 @@ data ProcessInfo = ProcessInfo {

471 , infoMessageQueueLength :: Maybe Int

472 , infoMonitors :: Set (ProcessId, MonitorRef)

473 , infoLinks :: Set ProcessId

474− } deriving (Show, Typeable)

474+ } deriving (Show, Eq, Typeable)

475

476instance Binary ProcessInfo where

477 get = ProcessInfo <$> get <*> get <*> get <*> get <*> get

@@ -486,7 +486,7 @@ data ProcessInfoNone = ProcessInfoNone DiedReason

486

487instance Binary ProcessInfoNone where

488 get = ProcessInfoNone <$> get

489− put (ProcessInfoNone r) = put r

489+ put (ProcessInfoNone r) = put r

490

491--------------------------------------------------------------------------------

492-- Node controller internal data types --

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

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

124 , Identifier(..)

125 , nodeOf

126 , ProcessInfo(..)

127+ , ProcessInfoNone(..)

127 , SendPortId(..)

128 , SendPortId(..)

128 , typedChannelWithId

129 , typedChannelWithId

129 , RegisterReply(..)

130 , RegisterReply(..)

@@ -751,10 +752,13 @@ ncEffectGetInfo from pid =

751 let lpid = processLocalId pid

752 let lpid = processLocalId pid

752 them = (ProcessIdentifier pid)

753 them = (ProcessIdentifier pid)

753 in do

754 in do

755+ liftIO . putStrLn $ "processing get info"

754 node <- ask

756 node <- ask

755 mProc <- liftIO $ withMVar (localState node) $ return . (^. localProcessWithId lpid)

757 mProc <- liftIO $ withMVar (localState node) $ return . (^. localProcessWithId lpid)

756 case mProc of

758 case mProc of

757− Nothing -> return () -- send a response to the caller

759+ Nothing -> do

760+ dispatch (isLocal node (ProcessIdentifier from))

761+ from node (ProcessInfoNone DiedUnknownId)

758 Just proc -> do

762 Just proc -> do

759 itsLinks <- gets (^. linksFor them)

763 itsLinks <- gets (^. linksFor them)

760 itsMons <- gets (^. monitorsFor them)

764 itsMons <- gets (^. monitorsFor them)

@@ -781,10 +785,11 @@ ncEffectGetInfo from pid =

781 , infoLinks = l

785 , infoLinks = l

782 }

786 }

783

787

788+ dispatch :: (Serializable a)

784− dispatch :: Bool

789+ => Bool

785 -> ProcessId

790 -> ProcessId

786 -> LocalNode

791 -> LocalNode

787− -> ProcessInfo

792+ -> a

788 -> NC ()

793 -> NC ()

789 dispatch True dest _ pInfo = postAsMessage dest $ pInfo

794 dispatch True dest _ pInfo = postAsMessage dest $ pInfo

790 dispatch False dest node pInfo =

795 dispatch False dest node pInfo =

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

@@ -0,0 +1,88 @@

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

2+module Main where

3+

4+import Prelude hiding (catch)

5+import qualified Network.Transport as NT

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 Test.HUnit (Assertion)

26+import Test.HUnit.Base (assertBool)

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

28+

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

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

31+

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

33+type TestResult a = MVar a

34+

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

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

37+delayedAssertion note localNode expected testProc = do

38+ result <- newEmptyMVar

39+ _ <- forkProcess localNode $ testProc result

40+ assertComplete note result expected

41+

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

43+assertComplete msg mv a = do

44+ b <- takeMVar mv

45+ assertBool msg (a == b)

46+

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

48+stash mvar x = liftIO $ putMVar mvar x

49+

50+------

51+

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

53+testLocalDeadProcessInfo result = do

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

55+ mref <- monitor pid

56+ send pid "finish"

57+ _ <- receiveWait [

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

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

60+ (\p -> return p)

61+ ]

62+ say "waiting for process info...."

63+ getProcessInfo pid >>= stash result

64+

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

66+tests node1 = do

67+ return [

68+ testGroup "Process Info" [

69+ testCase "testLocalDeadProcessInfo"

70+ (delayedAssertion

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

72+ node1 (Nothing) testLocalDeadProcessInfo)

73+ ] ]

74+

75+mkNode :: String -> IO LocalNode

76+mkNode port = do

77+ Right (transport1, _) <- createTransportExposeInternals

78+ "127.0.0.1" port defaultTCPParameters

79+ newLocalNode transport1 initRemoteTable

80+

81+main :: IO ()

82+main = do

83+ node1 <- mkNode "8081"

84+ testData <- tests node1

85+ defaultMain testData

86+ closeLocalNode node1

87+ return ()

88+