a5191ce..master · distributed-process

git.iko.soy/distributed-process

on master 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: Facundo Domínguez <facundo.dominguez@tweag.io>

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

10copyright: Well-Typed LLP

11category: Control, Cloud Haskell

12build-type: Simple

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

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

49+import Data.Maybe (isNothing, isJust)

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

51import Test.Framework (Test, testGroup)

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

53−import Control.Rematch hiding (match)

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

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

56newtype Ping = Ping ProcessId

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

735testRemoteRegistry TestTransport{..} = do

736 node1 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

738− done <- newEmptyMVar

739

738

740 pingServer <- forkProcess node1 ping

739 pingServer <- forkProcess node1 ping

741 deadProcess <- forkProcess node1 (return ())

740 deadProcess <- forkProcess node1 (return ())

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

750

749

751 pid <- verifyWhereIsRemote nid1 "ping"

750 pid <- verifyWhereIsRemote nid1 "ping"

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

751 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 [

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

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

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

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

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

765+

765 registerRemoteAsync nid1 "ping" deadProcess

766 registerRemoteAsync nid1 "ping" deadProcess

766 receiveWait [

767 receiveWait [

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

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

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

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

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

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

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

772+

770 unregisterRemoteAsync nid1 "dead"

773 unregisterRemoteAsync nid1 "dead"

771 receiveWait [

774 receiveWait [

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

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

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,6 +1272,76 @@ 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+

1275testKillLocal :: TestTransport -> Assertion

1345testKillLocal :: TestTransport -> Assertion

1276testKillLocal TestTransport{..} = do

1346testKillLocal TestTransport{..} = do

1277 localNode <- newLocalNode testTransport initRemoteTable

1347 localNode <- newLocalNode testTransport initRemoteTable

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

1486testRegistryMonitoring TestTransport{..} = do

1556testRegistryMonitoring TestTransport{..} = do

1487 node1 <- newLocalNode testTransport initRemoteTable

1557 node1 <- newLocalNode testTransport initRemoteTable

1488 node2 <- newLocalNode testTransport initRemoteTable

1558 node2 <- newLocalNode testTransport initRemoteTable

1489− waitH <- newEmptyMVar

1490

1559

1491 let nid = localNodeId node2

1560 let nid = localNodeId node2

1492 pid <- forkProcess node1 $ do

1561 pid <- forkProcess node1 $ do

1562+ self <- getSelfPid

1493− getSelfPid >>= runUntilRegistered nid

1563+ runUntilRegistered nid self

1494− liftIO $ takeMVar waitH

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

1565+ expect :: Process ()

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

1495

1567

1496 runProcess node2 $ do

1568 runProcess node2 $ do

1497 register regName pid

1569 register regName pid

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

1498 res <- whereis regName

1571 res <- whereis regName

1572+ send pid ()

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

1499 us <- getSelfPid

1574 us <- getSelfPid

1500− liftIO $ do

1501− putMVar waitH ()

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

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

1576+

1503

1577

1504 -- This delay isn't essential!

1578 -- This delay isn't essential!

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

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

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

1524 regHere <- newEmptyMVar

1598 regHere <- newEmptyMVar

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

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

1526 res <- takeMVar regHere

1600 res <- takeMVar regHere

1601+ case res of

1602+ Nothing -> return ()

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

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

1528

1604

1529 where

1605 where

1530 runUntilRegistered nid us = do

1606 runUntilRegistered nid us = do

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

1536

1612

1537 delayUntilMaybeUnregistered nid p = do

1613 delayUntilMaybeUnregistered nid p = do

1538 whereisRemoteAsync nid regName

1614 whereisRemoteAsync nid regName

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

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

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

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

1541 (const $ return ())

1617 (const $ return ())

1542 ]

1618 ]

1619+ case res of

1543− return ()

1620+ Just () -> return ()

1621+ Nothing -> delayUntilMaybeUnregistered nid p

1544

1622

1545 regName = "testRegisterRemote"

1623 regName = "testRegisterRemote"

1546

1624

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

1736 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1737 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1738 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1739 , testCase "KillLocal" (testKillLocal testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1740 , testCase "KillRemote" (testKillRemote testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1741 , testCase "Die" (testDie testtrans)

1820 , testCase "Die" (testDie testtrans)

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

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

91import Test.HUnit.Base (assertBool)

93import GHC.Generics

94+import System.Timeout (timeout)

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

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

96type TestResult a = MVar a

97type TestResult a = MVar a

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

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

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

234tryRunProcess node p = do

235tryRunProcess node p = do

235 tid <- liftIO myThreadId

236 tid <- liftIO myThreadId

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

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

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

237

239

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

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

239tryForkProcess node p = do

241tryForkProcess node p = do

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

@@ -5,7 +5,6 @@ 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)

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

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

10import Control.Distributed.Process.Internal.Types

9import Control.Distributed.Process.Internal.Types

11 ( ProcessExitException(..)

10 ( ProcessExitException(..)

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

17 , nsend

16 , nsend

18 , nsendRemote

17 , nsendRemote

19 , usend

18 , 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)

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

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 () -> Process ()

69+testAgentBroadcast :: TestResult (Maybe ()) -> 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 =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

86− kill publisher "finished" >> awaitExit publisher

86+ kill publisher "finished"

87− kill consumer "finished" >> awaitExit consumer

87+ kill consumer "finished"

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

90testAgentDualInput result = do

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

99 else mxReady)

100 ]

101

102+ mRef <- monitor s

103+

102 mxNotify (1 :: Int)

104 mxNotify (1 :: Int)

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

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

104 mxNotify (2 :: Int)

106 mxNotify (2 :: Int)

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

106 mxNotify (5 :: Int)

108 mxNotify (5 :: Int)

107

109

108 stash result =<< receiveChanTimeout 10000000 rp

110 stash result =<< receiveChanTimeout 10000000 rp

109− awaitExit s

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

110

115

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

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

112testAgentPrioritisation result = do

117testAgentPrioritisation result = do

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

120 (sp, rp) <- newChan

125 (sp, rp) <- newChan

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

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

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

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

123− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

124 st <- mxGetLocal

129 st <- mxGetLocal

125 case length st of

130 case length st of

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

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

127 mxDeactivate "finished"

132 mxDeactivate "finished"

128 _ -> mxReceive -- go to the mailbox

133 _ -> mxReceive -- go to the mailbox

129 )

134 )

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

134 mxNotify "fourth"

139 mxNotify "fourth"

135 nsend name "fifth"

140 nsend name "fifth"

136

141

137− stash result . sort =<< receiveChan rp

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

138 awaitExit s

143 awaitExit s

139

144

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

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

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

147 nsend "mailbox-agent" ()

152 nsend "mailbox-agent" ()

148

153

149 stash result =<< receiveChanTimeout 1000000 rp

154 stash result =<< receiveChanTimeout 1000000 rp

150− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

151

156

152testAgentEventHandling :: TestResult Bool -> Process ()

157testAgentEventHandling :: TestResult Bool -> Process ()

153testAgentEventHandling result = do

158testAgentEventHandling result = do

154− let initState = [] :: [MxEvent]

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"

155

166

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

156 (rc, rs) <- newChan

168 (rc, rs) <- newChan

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

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

158 (mxSink $ \ev -> do

170 (mxSink $ \ev -> do

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

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

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

186 faff 2000000 rs

198 faff 2000000 rs

187

199

188− _ <- monitor agentPid

189− (sp, rp) <- newChan

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

190

201

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

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

192

203

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

195 monitor pid

206 monitor pid

196

207

197 send pid ()

208 send pid ()

198− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

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

199

211

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

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

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

201

214

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

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

203 faff 2000000 rs

216 faff 2000000 rs

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

206 mxNotify (MxSpawned pid, replyTo)

219 mxNotify (MxSpawned pid, replyTo)

207 mxNotify (MxProcessDied pid DiedNormal, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

208

221

209− seenAlive <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

210− seenDead <- receiveChan reply

223+ seenDead <- receiveChanTimeout 10000000 reply

211

224

212− stash result $ seenAlive && seenDead

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

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

214 where

228 where

215 faff delay port = do

229 faff delay port = do

216 res <- receiveChanTimeout delay port

230 res <- receiveChanTimeout delay port

217− unless (isNothing res) $ do

231+ unless (isNothing res) $ pause delay

218− pause delay

219− faff delay port

220

232

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

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

222testMxRegEvents result = do

234testMxRegEvents result = do

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

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

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

264

276

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

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

266− awaitExit agent

267

278

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

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

269testMxRegMon remoteNode result = do

280testMxRegMon remoteNode result = do

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

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

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

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

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

323

334

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

335+ kill agent "test-complete"

325

336

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

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

327ensure = flip finally

338ensure = flip finally

328

339

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

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

330 -> Maybe LocalNode

341 -> Maybe LocalNode

331− -> TestResult Bool

332 -> Process ()

342 -> Process ()

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

343+testNSend op n = do

344+ us <- getSelfPid

334 let delay = 5000000

345 let delay = 5000000

335− let label = "testMxSend"

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

336 let isValid = isValidLabel n label

347 let isValid = isValidLabel n label

337

348

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

338− register label p1

350+ register label p1

339− reg1 <- receiveChanTimeout delay sink

351+ reg1 <- receiveChanTimeout delay sink

340− case reg1 of

352+ case reg1 of

341− Just (MxRegistered pd lb)

353+ Just (MxRegistered pd lb)

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

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

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

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

344

356

345− op label ()

357+ op label ()

346

358

347− us <- getSelfPid

348− sent <- receiveChanTimeout delay sink

359+ sent <- receiveChanTimeout delay sink

349− case sent of

360+ case sent of

350− Just (MxSentToName lb by _)

361+ Just (MxSentToName lb by _)

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

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

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

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

353

364

354 where

365 where

355− isValidLabel n l1 l2

366+ isValidLabel nd l1 l2

356 | l2 == l2 = True

367 | l2 == l2 = True

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

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

358 | otherwise = False

369 | otherwise = False

359

370

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

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

361 -> Maybe LocalNode

372 -> Maybe LocalNode

362− -> TestResult Bool

363 -> Process ()

373 -> Process ()

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

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

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+

365 -- initiate a send

405 -- initiate a send

366− op p1 ()

406+ op ch' ()

367

407

368 -- verify the management event

408 -- verify the management event

369− us <- getSelfPid

370 sent <- receiveChanTimeout 5000000 sink

409 sent <- receiveChanTimeout 5000000 sink

371 case sent of

410 case sent of

372− Just (MxSent pidTo pidFrom _)

411+ Just (MxSentToPort sId spId _)

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

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

374− _ -> return False

413+ _ -> return False

375

414

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

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

377

416

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

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

379−testMxSend mNode result test = do

418+testMxSend mNode label test = do

380 us <- getSelfPid

419 us <- getSelfPid

420+ (sp, rp) <- newChan

381 (chan, sink) <- newChan

421 (chan, sink) <- newChan

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

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

383 mxSink $ \ev -> do

423 mxSink $ \ev -> do

384 case ev of

424 case ev of

425+ m@(MxSentToPort _ cid _)

426+ | cid /= sendPortId chan

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

385 m@(MxSent _ fromPid _)

428 m@(MxSent _ fromPid _)

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

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

387 m@(MxSentToName _ fromPid _)

430 m@(MxSentToName _ fromPid _)

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

392 mxReady

435 mxReady

393 ]

436 ]

394

437

395− (sp, rp) <- newChan

396 case mNode of

438 case mNode of

397 Nothing -> void $ spawnLocal (proc sp)

439 Nothing -> void $ spawnLocal (proc sp)

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

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

399

441

400− p1 <- receiveChan rp

442+ p1 <- receiveChanTimeout 2000000 rp

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

401− res <- try (test p1 sink)

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

402− case res of

445+

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

404− Right tr -> stash result tr

405 kill agent "bye"

446 kill agent "bye"

406− kill p1 "bye"

447+ kill (fromJust p1) "bye"

448+

449+ case res of

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

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

452+

407

453

408 where

454 where

409− label = "testMxSend"

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

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

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

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

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

412

466

413tests :: TestTransport -> IO [Test]

467tests :: TestTransport -> IO [Test]

414tests TestTransport{..} = do

468tests TestTransport{..} = do

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

423 , testCase "InterAgentBroadcast"

477 , testCase "InterAgentBroadcast"

424 (delayedAssertion

478 (delayedAssertion

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

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

426− node1 () testAgentBroadcast)

480+ node1 (Just ()) testAgentBroadcast)

427 , testCase "AgentMailboxHandling"

481 , testCase "AgentMailboxHandling"

428 (delayedAssertion

482 (delayedAssertion

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

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

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

452 ]

506 ]

453 ]

507 ]

454 where

508 where

455− buildTestCases n1 n2 =

456− let nid = localNodeId n2 in build n2 [

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

457 ("NSend", [

510 ("NSend", [

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

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

460− , ("nsend", testNSend nsend)

511+ ("nsend", testNSend nsend)

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

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

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

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

462 ])

515 ])

463 , ("Send", [

516 , ("Send", [

517+ ("send", testSend send)

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

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

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

519+ , ("usend", testSend usend)

466− , ("send", testSend send)

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

467− , ("usend", testSend usend)

521+ , ("sendChan", testChan sendChan)

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

468 ])

523 ])

469 , ("Forward", [

524 , ("Forward", [

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

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

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

473 ]

528 ]

474

529

475 build :: LocalNode

530 build :: LocalNode

531+ -> LocalNode

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

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

477 -> [Test]

533 -> [Test]

478− build ln specs =

534+ build n ln specs =

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

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

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

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

481− (delayedAssertion "expected mx events failed"

482− ln True (caseImpl caseNode))

537+ (runProcess n (caseImpl caseNode))

483 | (caseName, caseImpl) <- groupCases

538 | (caseName, caseImpl) <- groupCases

484 ]

539 ]

485 | (groupName, groupCases) <- specs

540 | (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: Facundo Domínguez <facundo.dominguez@tweag.io>

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

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

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

307-- Channels --

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

309

310−-- | Create a new typed channel

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

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

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

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

312newChan = do

318newChan = do

313 proc <- ask

319 proc <- ask

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

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

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

79-- [Management Data API]

113-- [Management Data API]

80--

114--

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

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

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

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

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

145-- > monitorNames2 = getSelfPid >>= mxNotify

179-- > monitorNames2 = getSelfPid >>= mxNotify

146--

180--

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

148−--

149-- [Performance, Stablity and Scalability]

181-- [Performance, Stablity and Scalability]

150--

182--

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

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