dcaeb3d · distributed-process

distributed-process.cabal+31−1

compare with master

unifiedsplit

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

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

169 HS-Source-Dirs: tests

170

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+

171Test-Suite TestTracing

202Test-Suite TestTracing

172 Type: exitcode-stdio-1.0

203 Type: exitcode-stdio-1.0

173 Main-Is: TestTracing.hs

204 Main-Is: TestTracing.hs

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

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

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

199 HS-Source-Dirs: tests

230 HS-Source-Dirs: tests

200

231

201−

202Executable distributed-process-throughput

232Executable distributed-process-throughput

203 if flag(benchmarks)

233 if flag(benchmarks)

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

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

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

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

44 , mxNotify

45 -- * Constructing Mx Agents

46 , MxAction()

47− , MxAgentId

47+ , MxAgentId(..)

48 , MxAgent()

49 , mxAgent

50 , MxSink

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

74 , receiveWait

75 , matchChan

76 , unwrapMessage

77+ , whereis

77 )

78 )

78import Control.Distributed.Process.Internal.Types

79import Control.Distributed.Process.Internal.Types

79 ( Process

80 ( Process

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

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

71 -> MVar AgentConfig

72 -> Process ()

73mxAgentController forkProcess mv = do

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

75 trc <- liftIO $ startTracing forkProcess

74 trc <- liftIO $ startTracing forkProcess

76 sigbus <- liftIO $ newBroadcastTChanIO

75 sigbus <- liftIO $ newBroadcastTChanIO

77 weakQueue <- processWeakQ <$> ask

76 weakQueue <- processWeakQ <$> ask

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

112 -> Message -> IO ()

111 -> Message -> IO ()

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

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

114 atomicBroadcast ch (Just q) msg = do

113 atomicBroadcast ch (Just q) msg = do

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

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

116

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

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

121 -> IO ProcessId

122mxStartAgent fork chan handler = do

123 chan' <- atomically (dupTChan chan)

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

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

125+ fork proc

125

126

126startTracing :: Fork -> IO Tracer

127startTracing :: Fork -> IO Tracer

127startTracing forkProcess = do

128startTracing forkProcess = do

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

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

134systemLoggerTracer :: Process ()

135systemLoggerTracer = do

136 node <- processNode <$> ask

137− let tr = sendTraceMsg node

137+ let tr = sendTraceLog node

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

139 where

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

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

141− sendTraceMsg node ev = do

141+ sendTraceLog node ev = do

142 now <- liftIO $ getCurrentTime

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

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

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

202 where

203 traceLoop :: TracerState -> Process ()

204 traceLoop st = do

205+ let client' = client st

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

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

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

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

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

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

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

214 case set of

215 case set of

215 (TraceEnable pid) -> do

216 (TraceEnable pid) -> do

216 -- notify the previous tracer it has been replaced

217 -- notify the previous tracer it has been replaced

217− sendTrace st (createUnencodedMessage (MxTraceTakeover pid))

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

218 sendOk setResp

219 sendOk setResp

219 return st { client = (Just pid) }

220 return st { client = (Just pid) }

220 TraceDisable -> do

221 TraceDisable -> do

221− sendTrace st (createUnencodedMessage MxTraceDisable)

222+ sendTraceMsg client' (createUnencodedMessage MxTraceDisable)

222 sendOk setResp

223 sendOk setResp

223 return st { client = Nothing })

224 return st { client = Nothing })

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

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

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

294 case ev of

295 case ev of

295 (MxNodeDied _ _) ->

296 (MxNodeDied _ _) ->

296 case (traceNodes (flags st)) of

297 case (traceNodes (flags st)) of

297− True -> sendTrace st msg

298+ True -> sendTrace st ev msg

298 False -> return ()

299 False -> return ()

299− (MxUser _) -> sendTrace st msg

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

300− (MxLog _) -> sendTrace st msg

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

301 _ ->

302 _ ->

302 case (traceConnections (flags st)) of

303 case (traceConnections (flags st)) of

303− True -> sendTrace st msg

304+ True -> sendTrace st ev msg

304 False -> return ()

305 False -> return ()

305 return st

306 return st

306

307

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

310 -> TracerState

311 -> TracerState

311 -> Process ()

312 -> Process ()

312traceEv _ _ Nothing _ = return ()

313traceEv _ _ Nothing _ = return ()

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

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

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

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

315 node <- processNode <$> ask

316 node <- processNode <$> ask

316 let p = case resolveToPid ev of

317 let p = case resolveToPid ev of

317 Nothing -> (nullProcessId (localNodeId node))

318 Nothing -> (nullProcessId (localNodeId node))

318 Just pid -> pid

319 Just pid -> pid

319 case (Set.member p pids) of

320 case (Set.member p pids) of

320− True -> sendTrace st msg

321+ True -> sendTrace st ev msg

321 False -> return ()

322 False -> return ()

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

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

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

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

330 Nothing -> return ()

331 Nothing -> return ()

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

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

332 then return ()

333 then return ()

333− else sendTrace st msg

334+ else sendTrace st ev msg

334

335

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

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

336−sendTrace st msg =

337+sendTrace st ev msg = do

337− let pid = (client st) in do

338+ let c = client st

338− case pid of

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

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

340+ then return ()

340− Nothing -> return ()

341+ else sendTraceMsg c msg

341

342

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

344+sendTraceMsg Nothing _ = return ()

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

342

346

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

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

41 , EndPointAddress

42 )

43

44−

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

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

46

45

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

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

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

159 ( publishEvent

160 )

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

162+ ( mxTableCoordinator

163+ , startTableCoordinator

164+ )

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

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

162 ( remoteTable

166 ( remoteTable

163 )

167 )

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

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

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

273 _ -> return ()

277 _ -> return ()

274

278

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

280+

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

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

276startServiceProcesses :: LocalNode -> IO ()

282startServiceProcesses :: LocalNode -> IO ()

277startServiceProcesses node = do

283startServiceProcesses node = do

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

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

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

281 -- however, so we deliberably ignore it

287 -- however, so we deliberably ignore it

282 startDefaultTracer node

288 startDefaultTracer node

289+ tableCoordinatorPid <- fork $ Table.startTableCoordinator fork

290+ runProcess node $ register Table.mxTableCoordinator tableCoordinatorPid

283 logger <- forkProcess node loop

291 logger <- forkProcess node loop

284 runProcess node $ register "logger" logger

292 runProcess node $ register "logger" logger

285 where

293 where

286− loop = do

294+ fork = forkProcess node

287− receiveWait

295+

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

296+ loop = do

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

297+ receiveWait

290− loop

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

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

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

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

300+ loop

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

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

294− loop

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

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

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

296− sendChan ch ()

304+ loop

297− ]

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

306+ sendChan ch ()

307+ ]

298

308

299-- | Force-close a local node

309-- | Force-close a local node

300--

310--

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

18import Data.Binary()

19import Data.Typeable()

20import Network.Transport.TCP

21+import qualified Network.Transport as NT

22+

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

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

22import Prelude hiding (catch)

24import Prelude hiding (catch)

23#endif

25#endif

26+

24import Test.Framework

27import Test.Framework

25 ( Test

28 ( Test

26 , defaultMain

29 , defaultMain

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

29import Test.HUnit (Assertion)

32import Test.HUnit (Assertion)

30import Test.HUnit.Base (assertBool)

33import Test.HUnit.Base (assertBool)

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

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

32−

35+import TestUtils

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

55

36

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

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

57testLocalDeadProcessInfo result = do

38testLocalDeadProcessInfo result = do

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

149 (testRemoteLiveProcessInfo node1)

130 (testRemoteLiveProcessInfo node1)

150 ] ]

131 ] ]

151

132

152−mkNode :: String -> IO LocalNode

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

153−mkNode port = do

134+statsTests transport = do

154− Right (transport1, _) <- createTransportExposeInternals

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

155− "127.0.0.1" port defaultTCPParameters

156− newLocalNode transport1 initRemoteTable

157

136

158main :: IO ()

137main :: IO ()

159−main = do

138+main = testMain $ statsTests

160− node1 <- mkNode "8081"

139+

161− testData <- tests node1

162− defaultMain testData

163− closeLocalNode node1

164− return ()

tests/TestTracing.hs+20−56

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

22 , closeLocalNode

23 , LocalNode)

24import Network.Transport.TCP

25+import qualified Network.Transport as NT

26+

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

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

26import Prelude hiding (catch, log)

28import Prelude hiding (catch, log)

27#endif

29#endif

30+

28import Test.Framework

31import Test.Framework

29 ( Test

32 ( Test

30 , defaultMain

33 , defaultMain

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

33import Test.HUnit (Assertion)

36import Test.HUnit (Assertion)

34import Test.HUnit.Base (assertBool)

37import Test.HUnit.Base (assertBool)

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

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

36−

39+import TestUtils

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

69

40

70testSpawnTracing :: TestResult Bool -> Process ()

41testSpawnTracing :: TestResult Bool -> Process ()

71testSpawnTracing result = do

42testSpawnTracing result = do

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

328 return [

299 return [

329 testGroup "Tracing" [

300 testGroup "Tracing" [

330 testCase "Spawn Tracing"

301 testCase "Spawn Tracing"

331− (delayedAssertion

302+ (synchronisedAssertion

332 "expected dead process-info to be ProcessInfoNone"

303 "expected dead process-info to be ProcessInfoNone"

333 node1 True testSpawnTracing lock)

304 node1 True testSpawnTracing lock)

334 , testCase "Recv Tracing (Explicit Pid)"

305 , testCase "Recv Tracing (Explicit Pid)"

335− (delayedAssertion

306+ (synchronisedAssertion

336 "expected a recv trace for the supplied pid"

307 "expected a recv trace for the supplied pid"

337 node1 True testTraceRecvExplicitPid lock)

308 node1 True testTraceRecvExplicitPid lock)

338 , testCase "Recv Tracing (Named Pid)"

309 , testCase "Recv Tracing (Named Pid)"

339− (delayedAssertion

310+ (synchronisedAssertion

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

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

341 node1 True testTraceRecvNamedPid lock)

312 node1 True testTraceRecvNamedPid lock)

342 , testCase "Trace Send(er)"

313 , testCase "Trace Send(er)"

343− (delayedAssertion

314+ (synchronisedAssertion

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

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

345 node1 True testTraceSending lock)

316 node1 True testTraceSending lock)

346 , testCase "Trace Registration"

317 , testCase "Trace Registration"

347− (delayedAssertion

318+ (synchronisedAssertion

348 "expected a 'registered' trace"

319 "expected a 'registered' trace"

349 node1 True testTraceRegistration lock)

320 node1 True testTraceRegistration lock)

350 , testCase "Trace Unregistration"

321 , testCase "Trace Unregistration"

351− (delayedAssertion

322+ (synchronisedAssertion

352 "expected an 'unregistered' trace"

323 "expected an 'unregistered' trace"

353 node1 True testTraceUnRegistration lock)

324 node1 True testTraceUnRegistration lock)

354 , testCase "Trace Layering"

325 , testCase "Trace Layering"

355− (delayedAssertion

326+ (synchronisedAssertion

356− "expected blah"

327+ "expected blah"

357− node1 () testTraceLayering lock)

328+ node1 () testTraceLayering lock)

358 , testCase "Remote Trace Relay"

329 , testCase "Remote Trace Relay"

359− (delayedAssertion

330+ (synchronisedAssertion

360− "expected blah"

331+ "expected blah"

361− node1 True testRemoteTraceRelay lock)

332+ node1 True testRemoteTraceRelay lock)

362 ] ]

333 ] ]

363

334

364−mkNode :: String -> IO LocalNode

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

365−mkNode port = do

336+timerTests transport = do

366− Right (transport1, _) <- createTransportExposeInternals

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

367− "127.0.0.1" port defaultTCPParameters

368− newLocalNode transport1 initRemoteTable

369

338

370main :: IO ()

339main :: IO ()

371−main = do

340+main = testMain $ timerTests

372− node1 <- mkNode "8081"

373− testData <- tests node1

374− defaultMain testData

375− closeLocalNode node1

376− return ()

377

341

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+