master..a5191ce · distributed-process

git.iko.soy/distributed-process

on a5191ce help sign in

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

since splitsnapshots

unifiedsplit

@@ -6,7 +6,7 @@ description: Tests and test suite for Cloud Haskell libraries.

6license: BSD3

7license-file: LICENSE

8Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

9−Maintainer: Tim Watson <watson.timothy@gmail.com>

9+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>

10copyright: Well-Typed LLP

11category: Control, Cloud Haskell

12build-type: Simple

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+25−104

@@ -46,11 +46,11 @@ import Control.Distributed.Process.Debug

46import Control.Distributed.Process.Management.Internal.Types

47import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)

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

49−import Data.Maybe (isNothing, isJust)

49+

50import Test.HUnit (Assertion, assertBool, assertFailure)

51import Test.Framework (Test, testGroup)

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

53−import Control.Rematch hiding (match, isNothing, isJust)

53+import Control.Rematch hiding (match)

54import Control.Rematch.Run (Match(..))

56newtype Ping = Ping ProcessId

@@ -735,6 +735,7 @@ testRemoteRegistry :: TestTransport -> Assertion

735testRemoteRegistry TestTransport{..} = do

736 node1 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

738+ done <- newEmptyMVar

738

739

739 pingServer <- forkProcess node1 ping

740 pingServer <- forkProcess node1 ping

740 deadProcess <- forkProcess node1 (return ())

741 deadProcess <- forkProcess node1 (return ())

@@ -749,7 +750,6 @@ testRemoteRegistry TestTransport{..} = do

749

750

750 pid <- verifyWhereIsRemote nid1 "ping"

751 pid <- verifyWhereIsRemote nid1 "ping"

751 liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid

752 liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid

752−

753 us <- getSelfPid

754 nsendRemote nid1 "ping" (Pong us)

755 receiveWait [

@@ -759,22 +759,22 @@ testRemoteRegistry TestTransport{..} = do

759 -- test that if process was not registered Nothing is returned

760 -- in owner field.

761 registerRemoteAsync nid1 "dead" deadProcess

762+ receiveWait [

762− receiveWait [ matchIf (\(RegisterReply label' f mPid) -> "dead" == label')

763+ matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')

763− (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))

764+ (\(RegisterReply _ _ _) -> return ()) ]

764− ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"

765−

766 registerRemoteAsync nid1 "ping" deadProcess

765 registerRemoteAsync nid1 "ping" deadProcess

767 receiveWait [

766 receiveWait [

768− matchIf (\(RegisterReply label' False mPid) ->

767+ matchIf (\(RegisterReply label' False (Just pid'')) ->

769− "ping" == label' && isJust mPid)

768+ "ping" == label' && pid'' == pingServer)

770− (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))

769+ (\(RegisterReply _ _ _) -> return ()) ]

771− ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"

772−

773 unregisterRemoteAsync nid1 "dead"

770 unregisterRemoteAsync nid1 "dead"

774 receiveWait [

771 receiveWait [

775− matchIf (\(RegisterReply label' _ _) -> "dead" == label')

772+ matchIf (\(RegisterReply label' False Nothing) ->

773+ "dead" == label' && pid == pingServer)

776− (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))

774+ (\(RegisterReply _ _ _) -> return ()) ]

777− ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"

775+ liftIO $ putMVar done ()

776+

777+ takeMVar done

778

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

780testRemoteRegistryRemoteProcess TestTransport{..} = do

@@ -1272,76 +1272,6 @@ testReceiveChanFeatures TestTransport{..} = do

1272

1273 takeMVar done

1274

1275−testChanLifecycle :: TestTransport -> Assertion

1276−testChanLifecycle TestTransport{..} = let delay = 3000000 in do

1277− result <- newEmptyMVar

1278− tchMV <- newEmptyMVar

1279− localNode <- newLocalNode testTransport initRemoteTable

1280− runProcess localNode $ do

1281−

1282− pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())

1283− liftIO $ putMVar tchMV tCh

1284− expect :: Process ()

1285− let (sp, _) = tCh

1286− sendChan sp ()

1287− expect :: Process ()

1288−

1289− mRefPid <- monitor pid

1290−

1291− cPid <- spawnLocal $ do

1292− (sp', rp) <- liftIO $ takeMVar tchMV

1293− -- say "picked up our test channel"

1294− send pid ()

1295− -- say "told pid to continue"

1296− res <- receiveChanTimeout delay rp

1297− case res of

1298− Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)

1299− Just () -> do mr <- monitor pid

1300− pause 10000

1301− -- say "sending pid a second () will cause it to exit"

1302− send pid ()

1303−

1304− -- say "make sure we see a DOWN notification for pid having stopped"

1305− receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]

1306−

1307− -- now that pid has died, the send port should be useless...

1308− liftIO $ putMVar tchMV (sp', rp)

1309−

1310− -- let's verify that we do not see the message from our

1311− -- parent process on the channel, once pid has died...

1312− recv <- receiveChanTimeout delay rp

1313− -- say $ "finished waiting for second (), writing result" ++ (show recv)

1314− liftIO $ putMVar result $ isNothing recv

1315−

1316− mRefCPid <- monitor cPid

1317−

1318− receiveWait

1319− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)

1320− (\_ -> return ())

1321− ]

1322−

1323− -- say "seen first pid die..."

1324−

1325− (sendPort, _) <- liftIO $ takeMVar tchMV

1326− sendChan sendPort ()

1327− -- say "sent () after owning pid died"

1328−

1329− -- let cPid know we've written to the channel...

1330− send cPid ()

1331−

1332− receiveWait

1333− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)

1334− (\_ -> return ())

1335− ]

1336−

1337− -- say "seen both pids die now..."

1338−

1339− -- and wait on the result back in IO land...

1340− testRes <- takeMVar result

1341− -- runProcess localNode $ say "got result..."

1342− assertBool "Expected sending on the channel to fail, but received data!" testRes

1343−

1344−

1345testKillLocal :: TestTransport -> Assertion

1275testKillLocal :: TestTransport -> Assertion

1346testKillLocal TestTransport{..} = do

1276testKillLocal TestTransport{..} = do

1347 localNode <- newLocalNode testTransport initRemoteTable

1277 localNode <- newLocalNode testTransport initRemoteTable

@@ -1556,24 +1486,20 @@ testRegistryMonitoring :: TestTransport -> Assertion

1556testRegistryMonitoring TestTransport{..} = do

1486testRegistryMonitoring TestTransport{..} = do

1557 node1 <- newLocalNode testTransport initRemoteTable

1487 node1 <- newLocalNode testTransport initRemoteTable

1558 node2 <- newLocalNode testTransport initRemoteTable

1488 node2 <- newLocalNode testTransport initRemoteTable

1489+ waitH <- newEmptyMVar

1559

1490

1560 let nid = localNodeId node2

1491 let nid = localNodeId node2

1561 pid <- forkProcess node1 $ do

1492 pid <- forkProcess node1 $ do

1562− self <- getSelfPid

1563− runUntilRegistered nid self

1493+ getSelfPid >>= runUntilRegistered nid

1564− say $ (show self) ++ " registered as " ++ regName

1494+ liftIO $ takeMVar waitH

1565− expect :: Process ()

1566− say $ (show self) ++ " exiting normally"

1567

1495

1568 runProcess node2 $ do

1496 runProcess node2 $ do

1569 register regName pid

1497 register regName pid

1570− say $ regName ++ " registered to " ++ show pid

1571 res <- whereis regName

1498 res <- whereis regName

1572− send pid ()

1573− say $ " sent finish signal to " ++ show pid

1574 us <- getSelfPid

1499 us <- getSelfPid

1500+ liftIO $ do

1501+ putMVar waitH ()

1575− liftIO $ assertBool "expected (Just pid)" $ res == (Just pid)

1502+ assertBool "expected (Just pid)" $ res == (Just pid)

1576−

1577

1503

1578 -- This delay isn't essential!

1504 -- This delay isn't essential!

1579 -- The test case passes perfectly fine without it (feel free to comment out

1505 -- The test case passes perfectly fine without it (feel free to comment out

@@ -1598,9 +1524,7 @@ testRegistryMonitoring TestTransport{..} = do

1598 regHere <- newEmptyMVar

1524 regHere <- newEmptyMVar

1599 runProcess node2 $ whereis regName >>= liftIO . putMVar regHere

1525 runProcess node2 $ whereis regName >>= liftIO . putMVar regHere

1600 res <- takeMVar regHere

1526 res <- takeMVar regHere

1601− case res of

1602− Nothing -> return ()

1603− Just pid -> assertBool ("expected Nothing, but got " ++ show pid) False

1527+ assertBool "expected Nothing, but process still registered" (res == Nothing)

1604

1528

1605 where

1529 where

1606 runUntilRegistered nid us = do

1530 runUntilRegistered nid us = do

@@ -1612,13 +1536,11 @@ testRegistryMonitoring TestTransport{..} = do

1612

1536

1613 delayUntilMaybeUnregistered nid p = do

1537 delayUntilMaybeUnregistered nid p = do

1614 whereisRemoteAsync nid regName

1538 whereisRemoteAsync nid regName

1615− res <- receiveTimeout 20000000 {- 20 sec delay -} [

1539+ receiveTimeout 20000000 {- 20 sec delay -} [

1616− matchIf (\(WhereIsReply n p) -> n == regName && isNothing p)

1540+ matchIf (\(WhereIsReply n p) -> n == regName && p == Nothing)

1617 (const $ return ())

1541 (const $ return ())

1618 ]

1542 ]

1619− case res of

1620− Just () -> return ()

1543+ return ()

1621− Nothing -> delayUntilMaybeUnregistered nid p

1622

1544

1623 regName = "testRegisterRemote"

1545 regName = "testRegisterRemote"

1624

1546

@@ -1814,7 +1736,6 @@ tests testtrans = return [

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1736 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1737 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1738 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817− , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1739 , testCase "KillLocal" (testKillLocal testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1740 , testCase "KillRemote" (testKillRemote testtrans)

1820 , testCase "Die" (testDie testtrans)

1741 , testCase "Die" (testDie testtrans)

distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+1−3

@@ -91,7 +91,6 @@ import Test.HUnit (Assertion, assertFailure)

91import Test.HUnit.Base (assertBool)

93import GHC.Generics

94−import System.Timeout (timeout)

96-- | A mutable cell containing a test result.

95-- | A mutable cell containing a test result.

97type TestResult a = MVar a

96type TestResult a = MVar a

@@ -234,8 +233,7 @@ testProcessReport pid = do

234tryRunProcess :: LocalNode -> Process () -> IO ()

233tryRunProcess :: LocalNode -> Process () -> IO ()

235tryRunProcess node p = do

234tryRunProcess node p = do

236 tid <- liftIO myThreadId

235 tid <- liftIO myThreadId

237− void $ timeout (1000000 * 60 * 5 :: Int) $

238− runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

236+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

239

237

240tryForkProcess :: LocalNode -> Process () -> IO ProcessId

238tryForkProcess :: LocalNode -> Process () -> IO ProcessId

241tryForkProcess node p = do

239tryForkProcess node p = do

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+75−130

@@ -5,6 +5,7 @@ module Control.Distributed.Process.Tests.Mx (tests) where

6import Control.Distributed.Process.Tests.Internal.Utils

7import Network.Transport.Test (TestTransport(..))

8+import Control.Exception (SomeException)

8import Control.Distributed.Process hiding (bracket, finally, try)

9import Control.Distributed.Process hiding (bracket, finally, try)

9import Control.Distributed.Process.Internal.Types

10import Control.Distributed.Process.Internal.Types

10 ( ProcessExitException(..)

11 ( ProcessExitException(..)

@@ -16,7 +17,6 @@ import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

16 , nsend

17 , nsend

17 , nsendRemote

18 , nsendRemote

18 , usend

19 , usend

19− , sendChan

20 )

21import Control.Distributed.Process.Management

22 ( MxEvent(..)

@@ -38,7 +38,7 @@ import Control.Monad.Catch(finally, bracket, try)

38import Control.Rematch (equalTo)

39import Data.Binary

40import Data.List (find, sort, intercalate)

41−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)

41+import Data.Maybe (isJust, fromJust, isNothing)

42import Data.Typeable

43import GHC.Generics hiding (from)

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

@@ -66,7 +66,7 @@ awaitExit pid =

66 where

67 withMonitorRef p = bracket (monitor p) unmonitor

69−testAgentBroadcast :: TestResult (Maybe ()) -> Process ()

69+testAgentBroadcast :: TestResult () -> Process ()

70testAgentBroadcast result = do

71 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())

@@ -81,10 +81,10 @@ testAgentBroadcast result = do

81 mxNotify ()

82 -- Once the publisher has seen our message, it will broadcast the Publish

83 -- and the consumer will see that and send the result to our typed channel.

84− stash result =<< receiveChanTimeout 10000000 resultRP

84+ stash result =<< receiveChan resultRP

86− kill publisher "finished"

86+ kill publisher "finished" >> awaitExit publisher

87− kill consumer "finished"

87+ kill consumer "finished" >> awaitExit consumer

89testAgentDualInput :: TestResult (Maybe Int) -> Process ()

90testAgentDualInput result = do

@@ -99,8 +99,6 @@ testAgentDualInput result = do

99 else mxReady)

100 ]

101

102− mRef <- monitor s

103−

104 mxNotify (1 :: Int)

102 mxNotify (1 :: Int)

105 nsend "sum-agent" (3 :: Int)

103 nsend "sum-agent" (3 :: Int)

106 mxNotify (2 :: Int)

104 mxNotify (2 :: Int)

@@ -108,10 +106,7 @@ testAgentDualInput result = do

108 mxNotify (5 :: Int)

106 mxNotify (5 :: Int)

109

107

110 stash result =<< receiveChanTimeout 10000000 rp

108 stash result =<< receiveChanTimeout 10000000 rp

111− died <- receiveTimeout 10000000 [

109+ awaitExit s

112− matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)

113− ]

114− died `shouldBe` equalTo (Just True)

115

110

116testAgentPrioritisation :: TestResult [String] -> Process ()

111testAgentPrioritisation :: TestResult [String] -> Process ()

117testAgentPrioritisation result = do

112testAgentPrioritisation result = do

@@ -125,10 +120,10 @@ testAgentPrioritisation result = do

125 (sp, rp) <- newChan

120 (sp, rp) <- newChan

126 s <- mxAgent (MxAgentId name) ["first"] [

121 s <- mxAgent (MxAgentId name) ["first"] [

127 mxSink (\(s :: String) -> do

122 mxSink (\(s :: String) -> do

128− mxUpdateLocal (s:)

123+ mxUpdateLocal ((s:))

129 st <- mxGetLocal

124 st <- mxGetLocal

130 case length st of

125 case length st of

131− n | n == 5 -> do liftMX $ sendChan sp (sort st)

126+ n | n == 5 -> do liftMX $ sendChan sp st

132 mxDeactivate "finished"

127 mxDeactivate "finished"

133 _ -> mxReceive -- go to the mailbox

128 _ -> mxReceive -- go to the mailbox

134 )

129 )

@@ -139,7 +134,7 @@ testAgentPrioritisation result = do

139 mxNotify "fourth"

134 mxNotify "fourth"

140 nsend name "fifth"

135 nsend name "fifth"

141

136

142− stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp

137+ stash result . sort =<< receiveChan rp

143 awaitExit s

138 awaitExit s

144

139

145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()

140testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()

@@ -152,19 +147,12 @@ testAgentMailboxHandling result = do

152 nsend "mailbox-agent" ()

147 nsend "mailbox-agent" ()

153

148

154 stash result =<< receiveChanTimeout 1000000 rp

149 stash result =<< receiveChanTimeout 1000000 rp

155− kill agent "finished"

150+ kill agent "finished" >> awaitExit agent

156

151

157testAgentEventHandling :: TestResult Bool -> Process ()

152testAgentEventHandling :: TestResult Bool -> Process ()

158testAgentEventHandling result = do

153testAgentEventHandling result = do

159− us <- getSelfPid

160− -- because this test is a bit racy, let's ensure it can't run indefinitely

161− timer <- spawnLocal $ do

162− pause (1000000 * 60 * 5)

163− -- okay we've waited 5 mins, let's kill the test off if it's stuck...

164− stash result False

165− kill us "Test Timed Out"

166−

167 let initState = [] :: [MxEvent]

154 let initState = [] :: [MxEvent]

155+

168 (rc, rs) <- newChan

156 (rc, rs) <- newChan

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

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

170 (mxSink $ \ev -> do

158 (mxSink $ \ev -> do

@@ -197,7 +185,8 @@ testAgentEventHandling result = do

197 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

185 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

198 faff 2000000 rs

186 faff 2000000 rs

199

187

188+ _ <- monitor agentPid

200− (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())

189+ (sp, rp) <- newChan

201

190

202 pid <- spawnLocal $ expect >>= sendChan sp

191 pid <- spawnLocal $ expect >>= sendChan sp

203

192

@@ -206,11 +195,9 @@ testAgentEventHandling result = do

206 monitor pid

195 monitor pid

207

196

208 send pid ()

197 send pid ()

209− rct <- receiveChanTimeout 10000000 rp

198+ () <- receiveChan rp

210− unless (isJust rct) $ die "No response on channel"

211

199

212− pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]

200+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]

213− unless (isJust pmn) $ die "No monitor notification arrived"

214

201

215 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

202 -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...

216 faff 2000000 rs

203 faff 2000000 rs

@@ -219,16 +206,17 @@ testAgentEventHandling result = do

219 mxNotify (MxSpawned pid, replyTo)

206 mxNotify (MxSpawned pid, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

207 mxNotify (MxProcessDied pid DiedNormal, replyTo)

221

208

222− seenAlive <- receiveChanTimeout 10000000 reply

209+ seenAlive <- receiveChan reply

223− seenDead <- receiveChanTimeout 10000000 reply

210+ seenDead <- receiveChan reply

224

211

225− stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]

212+ stash result $ seenAlive && seenDead

226− kill timer "test-complete"

213+ kill agentPid "test-complete" >> awaitExit agentPid

227− kill agentPid "test-complete"

228 where

214 where

229 faff delay port = do

215 faff delay port = do

230 res <- receiveChanTimeout delay port

216 res <- receiveChanTimeout delay port

231− unless (isNothing res) $ pause delay

217+ unless (isNothing res) $ do

218+ pause delay

219+ faff delay port

232

220

233testMxRegEvents :: TestResult () -> Process ()

221testMxRegEvents :: TestResult () -> Process ()

234testMxRegEvents result = do

222testMxRegEvents result = do

@@ -275,6 +263,7 @@ testMxRegEvents result = do

275 reg3 `shouldBe` equalTo (Just (label, p1))

263 reg3 `shouldBe` equalTo (Just (label, p1))

276

264

277 mapM_ (flip kill $ "test-complete") [agent, p1, p2]

265 mapM_ (flip kill $ "test-complete") [agent, p1, p2]

266+ awaitExit agent

278

267

279testMxRegMon :: LocalNode -> TestResult () -> Process ()

268testMxRegMon :: LocalNode -> TestResult () -> Process ()

280testMxRegMon remoteNode result = do

269testMxRegMon remoteNode result = do

@@ -332,99 +321,67 @@ testMxRegMon remoteNode result = do

332 evts `shouldContain` (Just (label1, p1))

321 evts `shouldContain` (Just (label1, p1))

333 evts `shouldContain` (Just (label2, p1))

322 evts `shouldContain` (Just (label2, p1))

334

323

335− kill agent "test-complete"

324+ kill agent "test-complete" >> awaitExit agent

336

325

337ensure :: Process () -> Process () -> Process ()

326ensure :: Process () -> Process () -> Process ()

338ensure = flip finally

327ensure = flip finally

339

328

340testNSend :: (String -> () -> Process ())

329testNSend :: (String -> () -> Process ())

341 -> Maybe LocalNode

330 -> Maybe LocalNode

331+ -> TestResult Bool

342 -> Process ()

332 -> Process ()

343−testNSend op n = do

333+testNSend op n r = testMxSend n r $ \p1 sink -> do

344− us <- getSelfPid

345 let delay = 5000000

334 let delay = 5000000

346− let label = "testMxSend" ++ (show us)

335+ let label = "testMxSend"

347 let isValid = isValidLabel n label

336 let isValid = isValidLabel n label

348

337

349− testMxSend n label $ \p1 sink -> do

350− register label p1

338+ register label p1

351− reg1 <- receiveChanTimeout delay sink

339+ reg1 <- receiveChanTimeout delay sink

352− case reg1 of

340+ case reg1 of

353− Just (MxRegistered pd lb)

341+ Just (MxRegistered pd lb)

354− | pd == p1 && isValid lb -> return ()

342+ | pd == p1 && isValid lb -> return ()

355− _ -> die $ "Reg-Failed: " ++ show reg1

343+ _ -> die $ "Reg-Failed: " ++ show reg1

356

344

357− op label ()

345+ op label ()

358

346

347+ us <- getSelfPid

359− sent <- receiveChanTimeout delay sink

348+ sent <- receiveChanTimeout delay sink

360− case sent of

349+ case sent of

361− Just (MxSentToName lb by _)

350+ Just (MxSentToName lb by _)

362− | by == us && isValid lb -> return True

351+ | by == us && isValid lb -> return True

363− _ -> die $ "Send-Failed: " ++ show sent

352+ _ -> die $ "Send-Failed: " ++ show sent

364

353

365 where

354 where

366− isValidLabel nd l1 l2

355+ isValidLabel n l1 l2

367 | l2 == l2 = True

356 | l2 == l2 = True

368− | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)

357+ | isJust n = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust n)

369 | otherwise = False

358 | otherwise = False

370

359

371testSend :: (ProcessId -> () -> Process ())

360testSend :: (ProcessId -> () -> Process ())

372 -> Maybe LocalNode

361 -> Maybe LocalNode

362+ -> TestResult Bool

373 -> Process ()

363 -> Process ()

374−testSend op n = do

375− us <- getSelfPid

376− let label = "testMxSend" ++ (show us)

377− testMxSend n label $ \p1 sink -> do

378− -- initiate a send

379− op p1 ()

380−

381− -- verify the management event

382− sent <- receiveChanTimeout 5000000 sink

383− case sent of

384− Just (MxSent pidTo pidFrom _)

385− | pidTo == p1 && pidFrom == us -> return True

386− _ -> return False

387−

388−testChan :: (SendPort () -> () -> Process ())

389− -> Maybe LocalNode

390− -> Process ()

391−testChan op n = testMxSend n "" $ \p1 sink -> do

364+testSend op n r = testMxSend n r $ \p1 sink -> do

392−

393− us <- getSelfPid

394− send p1 us

395−

396− cleared <- receiveChanTimeout 2000000 sink

397− case cleared of

398− Just (MxSent pidTo pidFrom _)

399− | pidTo == p1 && pidFrom == us -> return ()

400− _ -> die "Received uncleared Mx Event"

401−

402− chan <- receiveTimeout 5000000 [ match return ]

403− let ch' = fromMaybe (error "No reply chan received") chan

404−

405 -- initiate a send

365 -- initiate a send

406− op ch' ()

366+ op p1 ()

407

367

408 -- verify the management event

368 -- verify the management event

369+ us <- getSelfPid

409 sent <- receiveChanTimeout 5000000 sink

370 sent <- receiveChanTimeout 5000000 sink

410 case sent of

371 case sent of

411− Just (MxSentToPort sId spId _)

372+ Just (MxSent pidTo pidFrom _)

412− | sId == us && spId == sendPortId ch' -> return True

373+ | pidTo == p1 && pidFrom == us -> return True

413− _ -> return False

374+ _ -> return False

414

375

415type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool

376type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool

416

377

417−testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()

378+testMxSend :: Maybe LocalNode -> TestResult Bool -> SendTest -> Process ()

418−testMxSend mNode label test = do

379+testMxSend mNode result test = do

419 us <- getSelfPid

380 us <- getSelfPid

420− (sp, rp) <- newChan

421 (chan, sink) <- newChan

381 (chan, sink) <- newChan

422 agent <- mxAgent (MxAgentId $ agentLabel us) () [

382 agent <- mxAgent (MxAgentId $ agentLabel us) () [

423 mxSink $ \ev -> do

383 mxSink $ \ev -> do

424 case ev of

384 case ev of

425− m@(MxSentToPort _ cid _)

426− | cid /= sendPortId chan

427− && cid /= sendPortId sp -> liftMX $ sendChan chan m

428 m@(MxSent _ fromPid _)

385 m@(MxSent _ fromPid _)

429 | fromPid == us -> liftMX $ sendChan chan m

386 | fromPid == us -> liftMX $ sendChan chan m

430 m@(MxSentToName _ fromPid _)

387 m@(MxSentToName _ fromPid _)

@@ -435,34 +392,23 @@ testMxSend mNode label test = do

435 mxReady

392 mxReady

436 ]

393 ]

437

394

395+ (sp, rp) <- newChan

438 case mNode of

396 case mNode of

439 Nothing -> void $ spawnLocal (proc sp)

397 Nothing -> void $ spawnLocal (proc sp)

440 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp

398 Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp

441

399

442− p1 <- receiveChanTimeout 2000000 rp

400+ p1 <- receiveChan rp

443− unless (isJust p1) $ die "Timed out waiting for ProcessId"

444− res <- try $ test (fromJust p1) sink

401+ res <- try (test p1 sink)

445−

446− kill agent "bye"

447− kill (fromJust p1) "bye"

448−

449 case res of

402 case res of

450− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m

403+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> stash result False

451− Right tr -> tr `shouldBe` equalTo True

404+ Right tr -> stash result tr

452−

405+ kill agent "bye"

406+ kill p1 "bye"

453

407

454 where

408 where

409+ label = "testMxSend"

455 agentLabel s = "mx-unsafe-check-agent-" ++ show s

410 agentLabel s = "mx-unsafe-check-agent-" ++ show s

456− proc sp' = getSelfPid >>= sendChan sp' >> go Nothing

411+ proc sp' = getSelfPid >>= sendChan sp' >> expect :: Process ()

457−

458− go :: Maybe (ReceivePort ()) -> Process ()

459− go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go

460− go c@(Just c') = receiveWait [ matchChan c' return ] >> go c

461−

462− replyChannel p' = do

463− (s, r) <- newChan

464− send p' s

465− return r

466

412

467tests :: TestTransport -> IO [Test]

413tests :: TestTransport -> IO [Test]

468tests TestTransport{..} = do

414tests TestTransport{..} = do

@@ -477,7 +423,7 @@ tests TestTransport{..} = do

477 , testCase "InterAgentBroadcast"

423 , testCase "InterAgentBroadcast"

478 (delayedAssertion

424 (delayedAssertion

479 "expected (), but no broadcast was received"

425 "expected (), but no broadcast was received"

480− node1 (Just ()) testAgentBroadcast)

426+ node1 () testAgentBroadcast)

481 , testCase "AgentMailboxHandling"

427 , testCase "AgentMailboxHandling"

482 (delayedAssertion

428 (delayedAssertion

483 "expected (Just ()), but no regular (mailbox) input was handled"

429 "expected (Just ()), but no regular (mailbox) input was handled"

@@ -506,20 +452,19 @@ tests TestTransport{..} = do

506 ]

452 ]

507 ]

453 ]

508 where

454 where

455+ buildTestCases n1 n2 =

509− buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [

456+ let nid = localNodeId n2 in build n2 [

510 ("NSend", [

457 ("NSend", [

511− ("nsend", testNSend nsend)

512− , ("Unsafe.nsend", testNSend Unsafe.nsend)

458+ ("Unsafe.nsend", testNSend Unsafe.nsend)

513− , ("nsendRemote", testNSend (nsendRemote nid))

514 , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))

459 , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))

460+ , ("nsend", testNSend nsend)

461+ , ("nsendRemote", testNSend (nsendRemote nid))

515 ])

462 ])

516 , ("Send", [

463 , ("Send", [

517− ("send", testSend send)

518− , ("Unsafe.send", testSend Unsafe.send)

464+ ("Unsafe.send", testSend Unsafe.send)

519− , ("usend", testSend usend)

465+ , ("Unsafe.usend", testSend Unsafe.usend)

520− , ("Unsafe.usend", testSend Unsafe.usend)

466+ , ("send", testSend send)

521− , ("sendChan", testChan sendChan)

467+ , ("usend", testSend usend)

522− , ("Unsafe.sendChan", testChan Unsafe.sendChan)

523 ])

468 ])

524 , ("Forward", [

469 , ("Forward", [

525 ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))

470 ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))

@@ -528,13 +473,13 @@ tests TestTransport{..} = do

528 ]

473 ]

529

474

530 build :: LocalNode

475 build :: LocalNode

531− -> LocalNode

532− -> [(String, [(String, (Maybe LocalNode -> Process ()))])]

476+ -> [(String, [(String, (Maybe LocalNode -> TestResult Bool -> Process ()))])]

533 -> [Test]

477 -> [Test]

534− build n ln specs =

478+ build ln specs =

535 [ testGroup (intercalate "-" [groupName, caseSuffix]) [

479 [ testGroup (intercalate "-" [groupName, caseSuffix]) [

536 testCase (intercalate "-" [caseName, caseSuffix])

480 testCase (intercalate "-" [caseName, caseSuffix])

481+ (delayedAssertion "expected mx events failed"

537− (runProcess n (caseImpl caseNode))

482+ ln True (caseImpl caseNode))

538 | (caseName, caseImpl) <- groupCases

483 | (caseName, caseImpl) <- groupCases

539 ]

484 ]

540 | (groupName, groupCases) <- specs

485 | (groupName, groupCases) <- specs

distributed-process.cabal+1−1

@@ -6,7 +6,7 @@ License: BSD3

6License-File: LICENSE

7Copyright: Well-Typed LLP, Tweag I/O Limited

8Author: Duncan Coutts, Nicolas Wu, Edsko de Vries

9−Maintainer: Tim Watson <watson.timothy@gmail.com>

9+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>

10Stability: experimental

11Homepage: http://haskell-distributed.github.com/

12Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues

src/Control/Distributed/Process/Internal/Primitives.hs+1−7

@@ -307,13 +307,7 @@ expect = receiveWait [match return]

307-- Channels --

308--------------------------------------------------------------------------------

309

310−-- | Create a new typed channel, bound to the calling Process

310+-- | Create a new typed channel

311−--

312−-- Note that the channel is bound to the lifecycle of the process that evaluates

313−-- this function, such that when it dies/exits, the channel will no longer

314−-- function, but will remain accessible. Thus reading from the ReceivePort will

315−-- fail silently thereafter, blocking indefinitely (unless a timeout is used).

316−--

317newChan :: Serializable a => Process (SendPort a, ReceivePort a)

311newChan :: Serializable a => Process (SendPort a, ReceivePort a)

318newChan = do

312newChan = do

319 proc <- ask

313 proc <- ask

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

@@ -76,40 +76,6 @@

76--

77-- * Whether messages will be taken from the mailbox first, or the event bus.

78--

79−-- Since the event bus uses STM broadcast channels to communicate with agents,

80−-- no message written to the bus successfully can be lost.

81−--

82−-- Agents can also receive messages via their mailboxes - these are subject to

83−-- the same guarantees as all inter-process message sending.

84−--

85−-- Messages dispatched on an STM broadcast channel (i.e., management event bus)

86−-- are guaranteed to be delivered with the same FIFO ordering guarantees that

87−-- exist between two communicating processes, such that communication from the

88−-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but

89−-- messages dispatched to the event bus by other processes (including, but not

90−-- limited to agents) are only guaranteed to be ordered between one sender and

91−-- one receiver.

92−--

93−-- No guarantee exists for the ordering in which messages sent to an agent's

94−-- mailbox will be delivered, vs messages dispatched via the event bus.

95−--

96−-- Because of the above, there are no ordering guarantees for messages sent

97−-- between agents, or for processes to agents, except for those that apply to

98−-- messages sent between regular processes, since agents are

99−-- implemented as such.

100−--

101−-- The event bus is serial and single threaded. Anything that is published by

102−-- the node controller will be seen in FIFO order. There are no ordering

103−-- guarantees pertaining to entries published to the event bus by other

104−-- processes or agents.

105−--

106−-- It should not be possible to see, for example, an @MxReceived@ before the

107−-- corresponding @MxSent@ event, since the places where we issue the @MxSent@

108−-- write directly to the event bus (using STM) in the calling (green) thread,

109−-- before dispatching instructions to the node controller to perform the

110−-- necessary routing to deliver the message to a process (or registered name,

111−-- or typed channel) locally or remotely.

112−--

113-- [Management Data API]

79-- [Management Data API]

114--

80--

115-- Both management agents and clients of the API have access to a variety of

81-- Both management agents and clients of the API have access to a variety of

@@ -178,6 +144,8 @@

178-- > monitorNames = getSelfPid >>= nsend "name-monitor"

144-- > monitorNames = getSelfPid >>= nsend "name-monitor"

179-- > monitorNames2 = getSelfPid >>= mxNotify

145-- > monitorNames2 = getSelfPid >>= mxNotify

180--

146--

147+-- For some real-world examples, see the distributed-process-platform package.

148+--

181-- [Performance, Stablity and Scalability]

149-- [Performance, Stablity and Scalability]

182--

150--

183-- /Management Agents/ offer numerous advantages over regular processes:

151-- /Management Agents/ offer numerous advantages over regular processes: