dcaeb3d · distributed-process

distributed-process.cabal+31−1

@@ -168,6 +168,37 @@ Test-Suite TestStats

168168 ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

169169 HS-Source-Dirs: tests

170170

171+

172+Test-Suite TestMx

173+ Type: exitcode-stdio-1.0

174+ Main-Is: TestMx.hs

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

176+ random >= 1.0 && < 1.1,

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

178+ containers >= 0.4 && < 0.6,

179+ stm >= 2.4 && < 2.5,

180+ distributed-process,

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

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

183+ binary >= 0.5 && < 0.7,

184+ network >= 2.3 && < 2.5,

185+ HUnit >= 1.2 && < 1.3,

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

187+ test-framework-hunit >= 0.2.0 && < 0.4,

188+ unix >= 2.5.0.0

189+ Extensions: RankNTypes,

190+ ScopedTypeVariables,

191+ FlexibleInstances,

192+ UndecidableInstances,

193+ ExistentialQuantification,

194+ GADTs,

195+ GeneralizedNewtypeDeriving,

196+ DeriveDataTypeable,

197+ CPP,

198+ BangPatterns

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

200+ HS-Source-Dirs: tests

201+

171202Test-Suite TestTracing

172203 Type: exitcode-stdio-1.0

173204 Main-Is: TestTracing.hs

@@ -198,7 +229,6 @@ Test-Suite TestTracing

198229 ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

199230 HS-Source-Dirs: tests

200231

201−

202232Executable distributed-process-throughput

203233 if flag(benchmarks)

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

src/Control/Distributed/Process/Management.hs+2−1

@@ -44,7 +44,7 @@ module Control.Distributed.Process.Management

4444 , mxNotify

4545 -- * Constructing Mx Agents

4646 , MxAction()

47− , MxAgentId

47+ , MxAgentId(..)

4848 , MxAgent()

4949 , mxAgent

5050 , MxSink

@@ -74,6 +74,7 @@ import Control.Distributed.Process.Internal.Primitives

7474 , receiveWait

7575 , matchChan

7676 , unwrapMessage

77+ , whereis

7778 )

7879import Control.Distributed.Process.Internal.Types

7980 ( Process

src/Control/Distributed/Process/Management/Agent.hs+3−2

@@ -71,7 +71,6 @@ mxAgentController :: Fork

7171 -> MVar AgentConfig

7272 -> Process ()

7373mxAgentController forkProcess mv = do

74− -- node <- processNode <$> ask

7574 trc <- liftIO $ startTracing forkProcess

7675 sigbus <- liftIO $ newBroadcastTChanIO

7776 weakQueue <- processWeakQ <$> ask

@@ -112,6 +111,7 @@ mxAgentController forkProcess mv = do

112111 -> Message -> IO ()

113112 atomicBroadcast ch Nothing msg = liftIO $ atomically $ writeTChan ch msg

114113 atomicBroadcast ch (Just q) msg = do

114+ -- liftIO $ putStrLn $ "broadcasting " ++ (show msg)

115115 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg

116116

117117-- | Forks a new process in which an mxAgent is run.

@@ -121,7 +121,8 @@ mxStartAgent :: Fork

121121 -> IO ProcessId

122122mxStartAgent fork chan handler = do

123123 chan' <- atomically (dupTChan chan)

124− let proc = handler (chan, chan') in fork proc

124+ let proc = handler (chan, chan')

125+ fork proc

125126

126127startTracing :: Fork -> IO Tracer

127128startTracing forkProcess = do

src/Control/Distributed/Process/Management/Trace/Tracer.hs+22−18

@@ -134,11 +134,11 @@ nullTracer =

134134systemLoggerTracer :: Process ()

135135systemLoggerTracer = do

136136 node <- processNode <$> ask

137− let tr = sendTraceMsg node

137+ let tr = sendTraceLog node

138138 forever' $ receiveWait [ matchAny (\m -> handleMessage m tr) ]

139139 where

140− sendTraceMsg :: LocalNode -> MxEvent -> Process ()

141− sendTraceMsg node ev = do

140+ sendTraceLog :: LocalNode -> MxEvent -> Process ()

141+ sendTraceLog node ev = do

142142 now <- liftIO $ getCurrentTime

143143 msg <- return $ (formatTime defaultTimeLocale "%c" now, buildTxt ev)

144144 emptyPid <- return $ (nullProcessId (localNodeId node))

@@ -202,6 +202,7 @@ traceController mv = do

202202 where

203203 traceLoop :: TracerState -> Process ()

204204 traceLoop st = do

205+ let client' = client st

205206 -- Trace events are forwarded to the enabled trace target.

206207 -- At some point in the future, we're going to start writing these custom

207208 -- events to the ghc eventlog, at which point this design might change.

@@ -214,11 +215,11 @@ traceController mv = do

214215 case set of

215216 (TraceEnable pid) -> do

216217 -- notify the previous tracer it has been replaced

217− sendTrace st (createUnencodedMessage (MxTraceTakeover pid))

218+ sendTraceMsg client' (createUnencodedMessage (MxTraceTakeover pid))

218219 sendOk setResp

219220 return st { client = (Just pid) }

220221 TraceDisable -> do

221− sendTrace st (createUnencodedMessage MxTraceDisable)

222+ sendTraceMsg client' (createUnencodedMessage MxTraceDisable)

222223 sendOk setResp

223224 return st { client = Nothing })

224225 , match (\(confResp, flags') ->

@@ -294,13 +295,13 @@ handleTrace st msg ev = do

294295 case ev of

295296 (MxNodeDied _ _) ->

296297 case (traceNodes (flags st)) of

297− True -> sendTrace st msg

298+ True -> sendTrace st ev msg

298299 False -> return ()

299− (MxUser _) -> sendTrace st msg

300− (MxLog _) -> sendTrace st msg

300+ (MxUser _) -> sendTrace st ev msg

301+ (MxLog _) -> sendTrace st ev msg

301302 _ ->

302303 case (traceConnections (flags st)) of

303− True -> sendTrace st msg

304+ True -> sendTrace st ev msg

304305 False -> return ()

305306 return st

306307

@@ -310,14 +311,14 @@ traceEv :: MxEvent

310311 -> TracerState

311312 -> Process ()

312313traceEv _ _ Nothing _ = return ()

313−traceEv _ msg (Just TraceAll) st = sendTrace st msg

314+traceEv ev msg (Just TraceAll) st = sendTrace st ev msg

314315traceEv ev msg (Just (TraceProcs pids)) st = do

315316 node <- processNode <$> ask

316317 let p = case resolveToPid ev of

317318 Nothing -> (nullProcessId (localNodeId node))

318319 Just pid -> pid

319320 case (Set.member p pids) of

320− True -> sendTrace st msg

321+ True -> sendTrace st ev msg

321322 False -> return ()

322323traceEv ev msg (Just (TraceNames names)) st = do

323324 -- if we have recorded regnames for p, then we forward the trace iif

@@ -330,13 +331,16 @@ traceEv ev msg (Just (TraceNames names)) st = do

330331 Nothing -> return ()

331332 Just ns -> if (Set.null (Set.intersection ns names))

332333 then return ()

333− else sendTrace st msg

334+ else sendTrace st ev msg

334335

335−sendTrace :: TracerState -> Message -> Process ()

336−sendTrace st msg =

337− let pid = (client st) in do

338− case pid of

339− Just p -> (flip forward) p msg

340− Nothing -> return ()

336+sendTrace :: TracerState -> MxEvent -> Message -> Process ()

337+sendTrace st ev msg = do

338+ let c = client st

339+ if c == (resolveToPid ev) -- we do not send the tracer events about itself...

340+ then return ()

341+ else sendTraceMsg c msg

341342

343+sendTraceMsg :: Maybe ProcessId -> Message -> Process ()

344+sendTraceMsg Nothing _ = return ()

345+sendTraceMsg (Just p) msg = (flip forward) p msg

342346

src/Control/Distributed/Process/Management/Types.hs−1

@@ -41,7 +41,6 @@ import Network.Transport

4141 , EndPointAddress

4242 )

4343

44−

4544-- TODO: consider factoring out MxEvent's tracing specific constructors

4645

4746-- | This is the /default/ management event, fired for various internal

src/Control/Distributed/Process/Node.hs+22−12

@@ -158,6 +158,10 @@ import Control.Distributed.Process.Management.Agent

158158import qualified Control.Distributed.Process.Management.Bus as Mx

159159 ( publishEvent

160160 )

161+import qualified Control.Distributed.Process.Management.Table as Table

162+ ( mxTableCoordinator

163+ , startTableCoordinator

164+ )

161165import qualified Control.Distributed.Process.Management.Trace.Remote as Trace

162166 ( remoteTable

163167 )

@@ -272,6 +276,8 @@ startDefaultTracer node' = do

272276 runProcess node' $ register "tracer.initial" pid'

273277 _ -> return ()

274278

279+-- TODO: we need a better mechanism for defining and registering services

280+

275281-- | Start and register the service processes on a node

276282startServiceProcesses :: LocalNode -> IO ()

277283startServiceProcesses node = do

@@ -280,21 +286,25 @@ startServiceProcesses node = do

280286 -- before /that/ process has started - this is a totally harmless race

281287 -- however, so we deliberably ignore it

282288 startDefaultTracer node

289+ tableCoordinatorPid <- fork $ Table.startTableCoordinator fork

290+ runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid

283291 logger <- forkProcess node loop

284292 runProcess node $ register "logger" logger

285293 where

286− loop = do

287− receiveWait

288− [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

289− liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

290− loop

291− , match $ \((time, string) :: (String, String)) -> do

292− -- this is a 'trace' message from the local node tracer

293− liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

294− loop

295− , match $ \(ch :: SendPort ()) -> -- a shutdown request

296− sendChan ch ()

297− ]

294+ fork = forkProcess node

295+

296+ loop = do

297+ receiveWait

298+ [ match $ \((time, pid, string) ::(String, ProcessId, String)) -> do

299+ liftIO . hPutStrLn stderr $ time ++ " " ++ show pid ++ ": " ++ string

300+ loop

301+ , match $ \((time, string) :: (String, String)) -> do

302+ -- this is a 'trace' message from the local node tracer

303+ liftIO . hPutStrLn stderr $ time ++ " [trace] " ++ string

304+ loop

305+ , match $ \(ch :: SendPort ()) -> -- a shutdown request

306+ sendChan ch ()

307+ ]

298308

299309-- | Force-close a local node

300310--

tests/TestMx.hsnew file+107

@@ -0,0 +1,107 @@

1+module Main where

2+

3+import Control.Concurrent (threadDelay)

4+import Control.Concurrent.MVar

5+ ( newEmptyMVar

6+ , tryPutMVar

7+ , takeMVar

8+ )

9+import Control.Distributed.Process

10+import Control.Distributed.Process.Management

11+ ( MxEvent(..)

12+ , MxAgentId(..)

13+ , mxAgent

14+ , mxSink

15+ , mxReady

16+ , mxLift

17+ , mxGetLocal

18+ , mxSetLocal

19+ , mxNotify

20+ )

21+import Control.Distributed.Process.Node

22+ ( LocalNode

23+ )

24+import Data.List (find)

25+import Data.Maybe (isJust)

26+import qualified Network.Transport as NT

27+

28+#if ! MIN_VERSION_base(4,6,0)

29+import Prelude hiding (catch, log)

30+#endif

31+

32+import Test.Framework

33+ ( Test

34+ , testGroup

35+ )

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

37+import TestUtils

38+

39+testAgentEventHandling :: TestResult Bool -> Process ()

40+testAgentEventHandling result = do

41+ let initState = [] :: [MxEvent]

42+ agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [

43+ (mxSink $ \ev -> do

44+ st <- mxGetLocal

45+ let act = case ev of

46+ (MxSpawned _) -> mxSetLocal (ev:st)

47+ (MxProcessDied _ _) -> mxSetLocal (ev:st)

48+ _ -> return ()

49+ act >> mxReady),

50+ (mxSink $ \(ev, sp :: SendPort Bool) -> do

51+ st <- mxGetLocal

52+ let found = case ev of

53+ MxSpawned p ->

54+ isJust $ find (\ev' ->

55+ case ev' of

56+ (MxSpawned p') -> p' == p

57+ _ -> False) st

58+ MxProcessDied p r ->

59+ isJust $ find (\ev' ->

60+ case ev' of

61+ (MxProcessDied p' r') -> p' == p && r == r'

62+ _ -> False) st

63+ _ -> False

64+ mxLift $ sendChan sp found

65+ liftIO $ putStrLn $ "replied to " ++ (show sp)

66+ mxReady)

67+ ]

68+

69+ mref <- monitor agentPid

70+ (sp, rp) <- newChan

71+ pid <- spawnLocal $ sendChan sp ()

72+ () <- receiveChan rp

73+

74+ (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

75+ mxNotify (MxSpawned pid, replyTo)

76+ mxNotify (MxProcessDied pid DiedNormal, replyTo)

77+

78+ seenAlive <- receiveChan reply

79+ seenDead <- receiveChan reply

80+

81+ stash result $ seenAlive && seenDead

82+

83+ kill agentPid "finished"

84+ receiveWait [

85+ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)

86+ ((\_ -> return ()))

87+ ]

88+

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

90+tests node1 = do

91+ return [

92+ testGroup "Basic Agent Handling" [

93+ testCase "testAgentEventHandling"

94+ (delayedAssertion

95+ "expected "

96+ node1 True testAgentEventHandling)

97+ ] ]

98+

99+statsTests :: NT.Transport -> IO [Test]

100+statsTests _ = do

101+ mkNode "8080" >>= tests >>= return

102+

103+main :: IO ()

104+main = do

105+ testMain $ statsTests

106+ threadDelay 100000

107+

tests/TestStats.hs+9−34

@@ -18,9 +18,12 @@ import Control.Distributed.Process.Node

1818import Data.Binary()

1919import Data.Typeable()

2020import Network.Transport.TCP

21+import qualified Network.Transport as NT

22+

2123#if ! MIN_VERSION_base(4,6,0)

2224import Prelude hiding (catch)

2325#endif

26+

2427import Test.Framework

2528 ( Test

2629 , defaultMain

@@ -29,29 +32,7 @@ import Test.Framework

2932import Test.HUnit (Assertion)

3033import Test.HUnit.Base (assertBool)

3134import Test.Framework.Providers.HUnit (testCase)

32−

33−-- these utilities have been cribbed from distributed-process-platform

34−-- we should really find a way to share them...

35−

36−-- | A mutable cell containing a test result.

37−type TestResult a = MVar a

38−

39−delayedAssertion :: (Eq a) => String -> LocalNode -> a ->

40− (TestResult a -> Process ()) -> Assertion

41−delayedAssertion note localNode expected testProc = do

42− result <- newEmptyMVar

43− _ <- forkProcess localNode $ testProc result

44− assertComplete note result expected

45−

46−assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()

47−assertComplete msg mv a = do

48− b <- takeMVar mv

49− assertBool msg (a == b)

50−

51−stash :: TestResult a -> a -> Process ()

52−stash mvar x = liftIO $ putMVar mvar x

53−

54−------

35+import TestUtils

5536

5637testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()

5738testLocalDeadProcessInfo result = do

@@ -149,16 +130,10 @@ tests node1 = do

149130 (testRemoteLiveProcessInfo node1)

150131 ] ]

151132

152−mkNode :: String -> IO LocalNode

153−mkNode port = do

154− Right (transport1, _) <- createTransportExposeInternals

155− "127.0.0.1" port defaultTCPParameters

156− newLocalNode transport1 initRemoteTable

133+statsTests :: NT.Transport -> IO [Test]

134+statsTests transport = do

135+ mkNode "8080" >>= tests >>= return

157136

158137main :: IO ()

159−main = do

160− node1 <- mkNode "8081"

161− testData <- tests node1

162− defaultMain testData

163− closeLocalNode node1

164− return ()

138+main = testMain $ statsTests

139+

tests/TestTracing.hs+20−56

@@ -22,9 +22,12 @@ import Control.Distributed.Process.Node

2222 , closeLocalNode

2323 , LocalNode)

2424import Network.Transport.TCP

25+import qualified Network.Transport as NT

26+

2527#if ! MIN_VERSION_base(4,6,0)

2628import Prelude hiding (catch, log)

2729#endif

30+

2831import Test.Framework

2932 ( Test

3033 , defaultMain

@@ -33,39 +36,7 @@ import Test.Framework

3336import Test.HUnit (Assertion)

3437import Test.HUnit.Base (assertBool)

3538import Test.Framework.Providers.HUnit (testCase)

36−

37−-- these utilities have been cribbed from distributed-process-platform

38−-- we should really find a way to share them...

39−

40−-- | A mutable cell containing a test result.

41−type TestResult a = MVar a

42−

43−delayedAssertion :: (Eq a)

44− => String

45− -> LocalNode

46− -> a

47− -> (TestResult a -> Process ())

48− -> MVar ()

49− -> Assertion

50−delayedAssertion note localNode expected testProc lock = do

51− result <- newEmptyMVar

52− _ <- forkProcess localNode $ do

53− acquire lock

54− finally (testProc result)

55− (release lock)

56− assertComplete note result expected

57− where acquire lock' = liftIO $ takeMVar lock'

58− release lock' = liftIO $ putMVar lock' ()

59−

60−assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()

61−assertComplete msg mv a = do

62− b <- takeMVar mv

63− assertBool msg (a == b)

64−

65−stash :: TestResult a -> a -> Process ()

66−stash mvar x = liftIO $ putMVar mvar x

67−

68−------

39+import TestUtils

6940

7041testSpawnTracing :: TestResult Bool -> Process ()

7142testSpawnTracing result = do

@@ -328,50 +299,43 @@ tests node1 = do

328299 return [

329300 testGroup "Tracing" [

330301 testCase "Spawn Tracing"

331− (delayedAssertion

302+ (synchronisedAssertion

332303 "expected dead process-info to be ProcessInfoNone"

333304 node1 True testSpawnTracing lock)

334305 , testCase "Recv Tracing (Explicit Pid)"

335− (delayedAssertion

306+ (synchronisedAssertion

336307 "expected a recv trace for the supplied pid"

337308 node1 True testTraceRecvExplicitPid lock)

338309 , testCase "Recv Tracing (Named Pid)"

339− (delayedAssertion

310+ (synchronisedAssertion

340311 "expected a recv trace for the process registered as 'foobar'"

341312 node1 True testTraceRecvNamedPid lock)

342313 , testCase "Trace Send(er)"

343− (delayedAssertion

314+ (synchronisedAssertion

344315 "expected a 'send' trace with the requisite fields set"

345316 node1 True testTraceSending lock)

346317 , testCase "Trace Registration"

347− (delayedAssertion

318+ (synchronisedAssertion

348319 "expected a 'registered' trace"

349320 node1 True testTraceRegistration lock)

350321 , testCase "Trace Unregistration"

351− (delayedAssertion

322+ (synchronisedAssertion

352323 "expected an 'unregistered' trace"

353324 node1 True testTraceUnRegistration lock)

354325 , testCase "Trace Layering"

355− (delayedAssertion

356− "expected blah"

357− node1 () testTraceLayering lock)

326+ (synchronisedAssertion

327+ "expected blah"

328+ node1 () testTraceLayering lock)

358329 , testCase "Remote Trace Relay"

359− (delayedAssertion

360− "expected blah"

361− node1 True testRemoteTraceRelay lock)

330+ (synchronisedAssertion

331+ "expected blah"

332+ node1 True testRemoteTraceRelay lock)

362333 ] ]

363334

364−mkNode :: String -> IO LocalNode

365−mkNode port = do

366− Right (transport1, _) <- createTransportExposeInternals

367− "127.0.0.1" port defaultTCPParameters

368− newLocalNode transport1 initRemoteTable

335+timerTests :: NT.Transport -> IO [Test]

336+timerTests transport = do

337+ mkNode "8080" >>= tests >>= return

369338

370339main :: IO ()

371−main = do

372− node1 <- mkNode "8081"

373− testData <- tests node1

374− defaultMain testData

375− closeLocalNode node1

376− return ()

340+main = testMain $ timerTests

377341

tests/TestUtils.hsnew file+89

@@ -0,0 +1,89 @@

1+-- | General testing support. Just @import TestUtils@ in your test module.

2+module TestUtils where

3+

4+#if ! MIN_VERSION_base(4,6,0)

5+import Prelude hiding (catch)

6+#endif

7+import Control.Concurrent

8+ ( ThreadId

9+ , myThreadId

10+ , forkIO

11+ )

12+import Control.Concurrent.STM

13+ ( TQueue

14+ , newTQueueIO

15+ , readTQueue

16+ , writeTQueue

17+ )

18+import Control.Concurrent.MVar

19+ ( MVar

20+ , newEmptyMVar

21+ , takeMVar

22+ , putMVar

23+ )

24+

25+import Control.Distributed.Process

26+import Control.Distributed.Process.Node

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

28+import Control.Exception hiding (finally)

29+import Control.Monad (forever)

30+import Control.Monad.STM (atomically)

31+import Test.HUnit (Assertion, assertFailure)

32+import Test.HUnit.Base (assertBool)

33+import Test.Framework (Test, defaultMain)

34+

35+import Network.Transport.TCP

36+import qualified Network.Transport as NT

37+

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

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

40+

41+mkNode :: String -> IO LocalNode

42+mkNode port = do

43+ Right (transport1, _) <- createTransportExposeInternals

44+ "127.0.0.1" port defaultTCPParameters

45+ newLocalNode transport1 initRemoteTable

46+

47+-- | Given a @builder@ function, make and run a test suite on a single transport

48+testMain :: (NT.Transport -> IO [Test]) -> IO ()

49+testMain builder = do

50+ Right (transport, _) <- createTransportExposeInternals

51+ "127.0.0.1" "10001" defaultTCPParameters

52+ testData <- builder transport

53+ defaultMain testData

54+

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

56+type TestResult a = MVar a

57+

58+synchronisedAssertion :: (Eq a)

59+ => String

60+ -> LocalNode

61+ -> a

62+ -> (TestResult a -> Process ())

63+ -> MVar ()

64+ -> Assertion

65+synchronisedAssertion note localNode expected testProc lock = do

66+ result <- newEmptyMVar

67+ _ <- forkProcess localNode $ do

68+ acquire lock

69+ finally (testProc result)

70+ (release lock)

71+ assertComplete note result expected

72+ where acquire lock' = liftIO $ takeMVar lock'

73+ release lock' = liftIO $ putMVar lock' ()

74+

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

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

77+delayedAssertion note localNode expected testProc = do

78+ result <- newEmptyMVar

79+ _ <- forkProcess localNode $ testProc result

80+ assertComplete note result expected

81+

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

83+assertComplete msg mv a = do

84+ b <- takeMVar mv

85+ assertBool msg (a == b)

86+

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

88+stash mvar x = liftIO $ putMVar mvar x

89+