master..5fc844f · distributed-process

git.iko.soy/distributed-process

on 5fc844f 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+125−156

@@ -1,10 +1,9 @@

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2−{-# LANGUAGE ParallelListComp #-}

3−

4module Control.Distributed.Process.Tests.Mx (tests) where

2module Control.Distributed.Process.Tests.Mx (tests) where

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

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

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

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

6+import Control.Exception (SomeException)

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

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

9import Control.Distributed.Process.Internal.Types

8import Control.Distributed.Process.Internal.Types

10 ( ProcessExitException(..)

9 ( ProcessExitException(..)

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

16 , nsend

15 , nsend

17 , nsendRemote

16 , nsendRemote

18 , usend

17 , usend

19− , sendChan

20 )

18 )

21import Control.Distributed.Process.Management

19import Control.Distributed.Process.Management

22 ( MxEvent(..)

20 ( MxEvent(..)

@@ -37,8 +35,8 @@ import Control.Monad (void, unless)

37import Control.Monad.Catch(finally, bracket, try)

35import Control.Monad.Catch(finally, bracket, try)

38import Control.Rematch (equalTo)

36import Control.Rematch (equalTo)

39import Data.Binary

37import Data.Binary

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

38+import Data.List (find, sort)

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

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

42import Data.Typeable

40import Data.Typeable

43import GHC.Generics hiding (from)

41import GHC.Generics hiding (from)

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

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

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

66 where

64 where

67 withMonitorRef p = bracket (monitor p) unmonitor

65 withMonitorRef p = bracket (monitor p) unmonitor

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

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

70testAgentBroadcast result = do

68testAgentBroadcast result = do

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

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

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

81 mxNotify ()

79 mxNotify ()

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

80 -- 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.

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

84− stash result =<< receiveChanTimeout 10000000 resultRP

82+ stash result =<< receiveChan resultRP

86− kill publisher "finished"

84+ kill publisher "finished" >> awaitExit publisher

87− kill consumer "finished"

85+ kill consumer "finished" >> awaitExit consumer

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

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

90testAgentDualInput result = do

88testAgentDualInput result = do

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

99 else mxReady)

97 else mxReady)

100 ]

98 ]

101

102− mRef <- monitor s

103−

104 mxNotify (1 :: Int)

100 mxNotify (1 :: Int)

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

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

106 mxNotify (2 :: Int)

102 mxNotify (2 :: Int)

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

108 mxNotify (5 :: Int)

104 mxNotify (5 :: Int)

109

105

110 stash result =<< receiveChanTimeout 10000000 rp

106 stash result =<< receiveChanTimeout 10000000 rp

111− died <- receiveTimeout 10000000 [

107+ awaitExit s

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

113− ]

114− died `shouldBe` equalTo (Just True)

115

108

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

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

117testAgentPrioritisation result = do

110testAgentPrioritisation result = do

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

125 (sp, rp) <- newChan

118 (sp, rp) <- newChan

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

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

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

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

128− mxUpdateLocal (s:)

121+ mxUpdateLocal ((s:))

129 st <- mxGetLocal

122 st <- mxGetLocal

130 case length st of

123 case length st of

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

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

132 mxDeactivate "finished"

125 mxDeactivate "finished"

133 _ -> mxReceive -- go to the mailbox

126 _ -> mxReceive -- go to the mailbox

134 )

127 )

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

139 mxNotify "fourth"

132 mxNotify "fourth"

140 nsend name "fifth"

133 nsend name "fifth"

141

134

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

135+ stash result . sort =<< receiveChan rp

143 awaitExit s

136 awaitExit s

144

137

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

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

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

152 nsend "mailbox-agent" ()

145 nsend "mailbox-agent" ()

153

146

154 stash result =<< receiveChanTimeout 1000000 rp

147 stash result =<< receiveChanTimeout 1000000 rp

155− kill agent "finished"

148+ kill agent "finished" >> awaitExit agent

156

149

157testAgentEventHandling :: TestResult Bool -> Process ()

150testAgentEventHandling :: TestResult Bool -> Process ()

158testAgentEventHandling result = do

151testAgentEventHandling 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]

152 let initState = [] :: [MxEvent]

153+

168 (rc, rs) <- newChan

154 (rc, rs) <- newChan

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

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

170 (mxSink $ \ev -> do

156 (mxSink $ \ev -> do

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

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

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

198 faff 2000000 rs

184 faff 2000000 rs

199

185

186+ _ <- monitor agentPid

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

187+ (sp, rp) <- newChan

201

188

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

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

203

190

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

206 monitor pid

193 monitor pid

207

194

208 send pid ()

195 send pid ()

209− rct <- receiveChanTimeout 10000000 rp

196+ () <- receiveChan rp

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

211

197

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

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

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

214

199

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

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

216 faff 2000000 rs

201 faff 2000000 rs

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

219 mxNotify (MxSpawned pid, replyTo)

204 mxNotify (MxSpawned pid, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

205 mxNotify (MxProcessDied pid DiedNormal, replyTo)

221

206

222− seenAlive <- receiveChanTimeout 10000000 reply

207+ seenAlive <- receiveChan reply

223− seenDead <- receiveChanTimeout 10000000 reply

208+ seenDead <- receiveChan reply

224

209

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

210+ stash result $ seenAlive && seenDead

226− kill timer "test-complete"

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

227− kill agentPid "test-complete"

228 where

212 where

229 faff delay port = do

213 faff delay port = do

230 res <- receiveChanTimeout delay port

214 res <- receiveChanTimeout delay port

231− unless (isNothing res) $ pause delay

215+ unless (isNothing res) $ do

216+ pause delay

217+ faff delay port

232

218

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

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

234testMxRegEvents result = do

220testMxRegEvents result = do

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

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

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

276

262

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

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

264+ awaitExit agent

278

265

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

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

280testMxRegMon remoteNode result = do

267testMxRegMon remoteNode result = do

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

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

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

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

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

334

321

335− kill agent "test-complete"

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

336

323

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

324ensure :: Process () -> Process () -> Process ()

338ensure = flip finally

325ensure = flip finally

339

326

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

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

341 -> Maybe LocalNode

328 -> Maybe LocalNode

329+ -> TestResult Bool

342 -> Process ()

330 -> Process ()

343−testNSend op n = do

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

344− us <- getSelfPid

345 let delay = 5000000

332 let delay = 5000000

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

333+ let label = "testMxSend"

347 let isValid = isValidLabel n label

334 let isValid = isValidLabel n label

348

335

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

350− register label p1

336+ register label p1

351− reg1 <- receiveChanTimeout delay sink

337+ reg1 <- receiveChanTimeout delay sink

352− case reg1 of

338+ case reg1 of

353− Just (MxRegistered pd lb)

339+ Just (MxRegistered pd lb)

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

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

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

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

356

342

357− op label ()

343+ op label ()

358

344

345+ us <- getSelfPid

359− sent <- receiveChanTimeout delay sink

346+ sent <- receiveChanTimeout delay sink

360− case sent of

347+ case sent of

361− Just (MxSentToName lb by _)

348+ Just (MxSentToName lb by _)

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

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

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

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

364

351

365 where

352 where

366− isValidLabel nd l1 l2

353+ isValidLabel n l1 l2

367 | l2 == l2 = True

354 | l2 == l2 = True

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

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

369 | otherwise = False

356 | otherwise = False

370

357

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

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

372 -> Maybe LocalNode

359 -> Maybe LocalNode

360+ -> TestResult Bool

373 -> Process ()

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

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

363 -- initiate a send

406− op ch' ()

364+ op p1 ()

407

365

408 -- verify the management event

366 -- verify the management event

367+ us <- getSelfPid

409 sent <- receiveChanTimeout 5000000 sink

368 sent <- receiveChanTimeout 5000000 sink

410 case sent of

369 case sent of

411− Just (MxSentToPort sId spId _)

370+ Just (MxSent pidTo pidFrom _)

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

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

413− _ -> return False

372+ _ -> return False

414

373

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

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

416

375

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

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

418−testMxSend mNode label test = do

377+testMxSend mNode result test = do

419 us <- getSelfPid

378 us <- getSelfPid

420− (sp, rp) <- newChan

421 (chan, sink) <- newChan

379 (chan, sink) <- newChan

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

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

423 mxSink $ \ev -> do

381 mxSink $ \ev -> do

424 case ev of

382 case ev of

425− m@(MxSentToPort _ cid _)

426− | cid /= sendPortId chan

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

428 m@(MxSent _ fromPid _)

383 m@(MxSent _ fromPid _)

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

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

430 m@(MxSentToName _ fromPid _)

385 m@(MxSentToName _ fromPid _)

@@ -435,39 +390,29 @@ testMxSend mNode label test = do

435 mxReady

390 mxReady

436 ]

391 ]

437

392

393+ (sp, rp) <- newChan

438 case mNode of

394 case mNode of

439 Nothing -> void $ spawnLocal (proc sp)

395 Nothing -> void $ spawnLocal (proc sp)

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

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

441

397

442− p1 <- receiveChanTimeout 2000000 rp

398+ p1 <- receiveChan rp

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

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

399+ res <- try (test p1 sink)

445−

446− kill agent "bye"

447− kill (fromJust p1) "bye"

448−

449 case res of

400 case res of

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

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

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

402+ Right tr -> stash result tr

452−

403+ kill agent "bye"

404+ kill p1 "bye"

453

405

454 where

406 where

407+ label = "testMxSend"

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

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

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

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

410

467tests :: TestTransport -> IO [Test]

411tests :: TestTransport -> IO [Test]

468tests TestTransport{..} = do

412tests TestTransport{..} = do

469 node1 <- newLocalNode testTransport initRemoteTable

413 node1 <- newLocalNode testTransport initRemoteTable

470 node2 <- newLocalNode testTransport initRemoteTable

414 node2 <- newLocalNode testTransport initRemoteTable

415+ let nid = localNodeId node2

471 return [

416 return [

472 testGroup "MxAgents" [

417 testGroup "MxAgents" [

473 testCase "EventHandling"

418 testCase "EventHandling"

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

477 , testCase "InterAgentBroadcast"

422 , testCase "InterAgentBroadcast"

478 (delayedAssertion

423 (delayedAssertion

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

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

480− node1 (Just ()) testAgentBroadcast)

425+ node1 () testAgentBroadcast)

481 , testCase "AgentMailboxHandling"

426 , testCase "AgentMailboxHandling"

482 (delayedAssertion

427 (delayedAssertion

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

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

@@ -502,41 +447,65 @@ tests TestTransport{..} = do

502 (delayedAssertion

447 (delayedAssertion

503 "expected process deaths to result in unregistration events"

448 "expected process deaths to result in unregistration events"

504 node1 () (testMxRegMon node2))

449 node1 () (testMxRegMon node2))

505− , testGroup "SendEvents" $ buildTestCases node1 node2

450+ , testGroup "SentEvents" [

506− ]

451+ testGroup "RemoteTargets" [

507− ]

452+ testCase "Unsafe.nsend"

508− where

453+ (delayedAssertion "expected mx events failed"

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

454+ node1 True (testNSend Unsafe.nsend $ Just node2))

510− ("NSend", [

455+ , testCase "Unsafe.nsendRemote"

511− ("nsend", testNSend nsend)

456+ (delayedAssertion "expected mx events failed"

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

457+ node1 True (testNSend (Unsafe.nsendRemote nid) $ Just node2))

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

458+ , testCase "Unsafe.send"

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

459+ (delayedAssertion "expected mx events failed"

515− ])

460+ node1 True (testSend Unsafe.send $ Just node2))

516− , ("Send", [

461+ , testCase "Unsafe.usend"

517− ("send", testSend send)

462+ (delayedAssertion "expected mx events failed"

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

463+ node1 True (testSend Unsafe.usend $ Just node2))

519− , ("usend", testSend usend)

464+ , testCase "nsend"

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

465+ (delayedAssertion "expected mx events failed"

521− , ("sendChan", testChan sendChan)

466+ node1 True (testNSend nsend $ Just node2))

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

467+ , testCase "nsendRemote"

523− ])

468+ (delayedAssertion "expected mx events failed"

524− , ("Forward", [

469+ node1 True (testNSend (nsendRemote nid) $ Just node2))

470+ , testCase "send"

471+ (delayedAssertion "expected mx events failed"

472+ node1 True (testSend send $ Just node2))

473+ , testCase "usend"

474+ (delayedAssertion "expected mx events failed"

475+ node1 True (testSend usend $ Just node2))

476+ , testCase "forward"

477+ (delayedAssertion "expected mx events failed"

478+ node1 True (testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p) $ Just node2))

479+ , testCase "uforward"

480+ (delayedAssertion "expected mx events failed"

481+ node1 True (testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p) $ Just node2))

482+ ]

483+ , testGroup "LocalTargets" [

484+ testCase "Unsafe.nsend"

485+ (delayedAssertion "expected mx events failed"

486+ node1 True (testNSend Unsafe.nsend Nothing))

487+ , testCase "Unsafe.send"

488+ (delayedAssertion "expected mx events failed"

489+ node1 True (testSend Unsafe.send Nothing))

490+ , testCase "Unsafe.usend"

491+ (delayedAssertion "expected mx events failed"

492+ node1 True (testSend Unsafe.usend Nothing))

493+ , testCase "nsend"

494+ (delayedAssertion "expected mx events failed"

495+ node1 True (testNSend nsend Nothing))

496+ , testCase "send"

497+ (delayedAssertion "expected mx events failed"

498+ node1 True (testSend send Nothing))

499+ , testCase "usend"

500+ (delayedAssertion "expected mx events failed"

501+ node1 True (testSend usend Nothing))

502+ , testCase "forward"

503+ (delayedAssertion "expected mx events failed"

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

504+ node1 True (testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p) Nothing))

505+ , testCase "uforward"

506+ (delayedAssertion "expected mx events failed"

526− , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))

507+ node1 True (testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p) Nothing))

527− ])

528 ]

508 ]

529−

509+ ]

530− build :: LocalNode

510+ ]

531− -> LocalNode

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

533− -> [Test]

534− build n ln specs =

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

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

537− (runProcess n (caseImpl caseNode))

538− | (caseName, caseImpl) <- groupCases

539− ]

540− | (groupName, groupCases) <- specs

541− , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]

542 ]

511 ]

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:

stack-ghc-7.10.3.yaml+1−1

@@ -14,7 +14,7 @@ extra-deps:

14- network-transport-0.5.2

15- network-transport-inmemory-0.5.2

16- rematch-0.2.0.0

17−- exceptions-0.8.2.1 # test dependency

17+- exceptions-0.8.2.1

19flags:

20 distributed-process-tests: