master..0514f3c · distributed-process

git.iko.soy/distributed-process

on 0514f3c 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+32−110

@@ -44,13 +44,13 @@ import Control.Distributed.Process.Internal.Types

44import Control.Distributed.Process.Node

45import Control.Distributed.Process.Debug

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

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

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

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

@@ -87,9 +87,10 @@ verifyClient :: String -> MVar Bool -> IO ()

87verifyClient s b = takeMVar b >>= assertBool s

89expectPing :: MVar Bool -> Process ()

90−expectPing mv = expect >>= liftIO . putMVar mv . checkPing

90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing

91 where

92 checkPing (Ping _) = True

93+ checkPing _ = False

94-- | Quick and dirty synchronous version of whereisRemoteAsync

95-- | Quick and dirty synchronous version of whereisRemoteAsync

95whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

96whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)

@@ -239,10 +240,10 @@ testPing TestTransport{..} = do

239 pid <- getSelfPid

240 pid <- getSelfPid

240 replicateM_ numPings $ do

241 replicateM_ numPings $ do

241 send pingServer (Pong pid)

242 send pingServer (Pong pid)

242− p <- expectTimeout 3000000

243+ p <- expect

243 case p of

244 case p of

244− Just (Ping _) -> return ()

245+ Ping _ -> return ()

245− Nothing -> die "Failed to receive Ping"

246+ _ -> die "Unexpected message"

246

247

247 putMVar clientDone ()

248 putMVar clientDone ()

248

249

@@ -559,7 +560,7 @@ testMergeChannels TestTransport{..} = do

559 ss <- liftIO $ mapM readMVar vs

560 ss <- liftIO $ mapM readMVar vs

560 case ss of

561 case ss of

561 [sa, sb, sc] ->

562 [sa, sb, sc] ->

562− mapM_ ((>> pause 10000) . uncurry sendChan)

563+ mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)

563 [ -- a, b, c

564 [ -- a, b, c

564 (sa, 'a')

565 (sa, 'a')

565 , (sb, 'b')

566 , (sb, 'b')

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

735testRemoteRegistry TestTransport{..} = do

736testRemoteRegistry TestTransport{..} = do

736 node1 <- newLocalNode testTransport initRemoteTable

737 node1 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

738 node2 <- newLocalNode testTransport initRemoteTable

739+ done <- newEmptyMVar

738

740

739 pingServer <- forkProcess node1 ping

741 pingServer <- forkProcess node1 ping

740 deadProcess <- forkProcess node1 (return ())

742 deadProcess <- forkProcess node1 (return ())

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

749

751

750 pid <- verifyWhereIsRemote nid1 "ping"

752 pid <- verifyWhereIsRemote nid1 "ping"

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

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

752−

753 us <- getSelfPid

754 us <- getSelfPid

754 nsendRemote nid1 "ping" (Pong us)

755 nsendRemote nid1 "ping" (Pong us)

755 receiveWait [

756 receiveWait [

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

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

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

760 -- in owner field.

761 -- in owner field.

761 registerRemoteAsync nid1 "dead" deadProcess

762 registerRemoteAsync nid1 "dead" deadProcess

763+ receiveWait [

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

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

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

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

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

765−

766 registerRemoteAsync nid1 "ping" deadProcess

767 receiveWait [

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

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

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

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

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

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

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

772−

773 unregisterRemoteAsync nid1 "dead"

771 unregisterRemoteAsync nid1 "dead"

774 receiveWait [

772 receiveWait [

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

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

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

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

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

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

776+ liftIO $ putMVar done ()

777+

778+ takeMVar done

778

779

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

780testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

780testRemoteRegistryRemoteProcess TestTransport{..} = do

781testRemoteRegistryRemoteProcess TestTransport{..} = do

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

1272

1273

1273 takeMVar done

1274 takeMVar done

1274

1275

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

1276testKillLocal :: TestTransport -> Assertion

1346testKillLocal TestTransport{..} = do

1277testKillLocal TestTransport{..} = do

1347 localNode <- newLocalNode testTransport initRemoteTable

1278 localNode <- newLocalNode testTransport initRemoteTable

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

1556testRegistryMonitoring TestTransport{..} = do

1487testRegistryMonitoring TestTransport{..} = do

1557 node1 <- newLocalNode testTransport initRemoteTable

1488 node1 <- newLocalNode testTransport initRemoteTable

1558 node2 <- newLocalNode testTransport initRemoteTable

1489 node2 <- newLocalNode testTransport initRemoteTable

1490+ waitH <- newEmptyMVar

1559

1491

1560 let nid = localNodeId node2

1492 let nid = localNodeId node2

1561 pid <- forkProcess node1 $ do

1493 pid <- forkProcess node1 $ do

1562− self <- getSelfPid

1563− runUntilRegistered nid self

1494+ getSelfPid >>= runUntilRegistered nid

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

1495+ liftIO $ takeMVar waitH

1565− expect :: Process ()

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

1567

1496

1568 runProcess node2 $ do

1497 runProcess node2 $ do

1569 register regName pid

1498 register regName pid

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

1571 res <- whereis regName

1499 res <- whereis regName

1572− send pid ()

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

1574 us <- getSelfPid

1500 us <- getSelfPid

1501+ liftIO $ do

1502+ putMVar waitH ()

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

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

1576−

1577

1504

1578 -- This delay isn't essential!

1505 -- This delay isn't essential!

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

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

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

1598 regHere <- newEmptyMVar

1525 regHere <- newEmptyMVar

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

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

1600 res <- takeMVar regHere

1527 res <- takeMVar regHere

1601− case res of

1602− Nothing -> return ()

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

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

1604

1529

1605 where

1530 where

1606 runUntilRegistered nid us = do

1531 runUntilRegistered nid us = do

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

1612

1537

1613 delayUntilMaybeUnregistered nid p = do

1538 delayUntilMaybeUnregistered nid p = do

1614 whereisRemoteAsync nid regName

1539 whereisRemoteAsync nid regName

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

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

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

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

1617 (const $ return ())

1542 (const $ return ())

1618 ]

1543 ]

1619− case res of

1620− Just () -> return ()

1544+ return ()

1621− Nothing -> delayUntilMaybeUnregistered nid p

1622

1545

1623 regName = "testRegisterRemote"

1546 regName = "testRegisterRemote"

1624

1547

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

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1737 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1738 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1739 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817− , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1740 , testCase "KillLocal" (testKillLocal testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1741 , testCase "KillRemote" (testKillRemote testtrans)

1820 , testCase "Die" (testDie testtrans)

1742 , 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+113−172

@@ -1,22 +1,17 @@

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)

9−import Control.Distributed.Process.Internal.Types

8+import Control.Distributed.Process.Internal.Types (ProcessExitException(..))

10− ( ProcessExitException(..)

11− , unsafeCreateUnencodedMessage

12− )

13import Control.Distributed.Process.Node

9import Control.Distributed.Process.Node

14import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

10import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe

15 ( send

11 ( send

16 , nsend

12 , nsend

17 , nsendRemote

13 , nsendRemote

18 , usend

14 , usend

19− , sendChan

20 )

15 )

21import Control.Distributed.Process.Management

16import Control.Distributed.Process.Management

22 ( MxEvent(..)

17 ( MxEvent(..)

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

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

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

38import Control.Rematch (equalTo)

33import Control.Rematch (equalTo)

39import Data.Binary

34import Data.Binary

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

35+import Data.List (find, sort)

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

36+import Data.Maybe (isJust, isNothing)

42import Data.Typeable

37import Data.Typeable

43import GHC.Generics hiding (from)

38import GHC.Generics hiding (from)

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

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

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

66 where

61 where

67 withMonitorRef p = bracket (monitor p) unmonitor

62 withMonitorRef p = bracket (monitor p) unmonitor

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

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

70testAgentBroadcast result = do

65testAgentBroadcast result = do

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

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

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

81 mxNotify ()

76 mxNotify ()

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

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

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

84− stash result =<< receiveChanTimeout 10000000 resultRP

79+ stash result =<< receiveChan resultRP

86− kill publisher "finished"

81+ kill publisher "finished" >> awaitExit publisher

87− kill consumer "finished"

82+ kill consumer "finished" >> awaitExit consumer

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

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

90testAgentDualInput result = do

85testAgentDualInput result = do

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

99 else mxReady)

94 else mxReady)

100 ]

95 ]

101

102− mRef <- monitor s

103−

104 mxNotify (1 :: Int)

97 mxNotify (1 :: Int)

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

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

106 mxNotify (2 :: Int)

99 mxNotify (2 :: Int)

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

108 mxNotify (5 :: Int)

101 mxNotify (5 :: Int)

109

102

110 stash result =<< receiveChanTimeout 10000000 rp

103 stash result =<< receiveChanTimeout 10000000 rp

111− died <- receiveTimeout 10000000 [

104+ awaitExit s

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

113− ]

114− died `shouldBe` equalTo (Just True)

115

105

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

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

117testAgentPrioritisation result = do

107testAgentPrioritisation result = do

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

125 (sp, rp) <- newChan

115 (sp, rp) <- newChan

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

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

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

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

128− mxUpdateLocal (s:)

118+ mxUpdateLocal ((s:))

129 st <- mxGetLocal

119 st <- mxGetLocal

130 case length st of

120 case length st of

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

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

132 mxDeactivate "finished"

122 mxDeactivate "finished"

133 _ -> mxReceive -- go to the mailbox

123 _ -> mxReceive -- go to the mailbox

134 )

124 )

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

139 mxNotify "fourth"

129 mxNotify "fourth"

140 nsend name "fifth"

130 nsend name "fifth"

141

131

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

132+ stash result . sort =<< receiveChan rp

143 awaitExit s

133 awaitExit s

144

134

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

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

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

152 nsend "mailbox-agent" ()

142 nsend "mailbox-agent" ()

153

143

154 stash result =<< receiveChanTimeout 1000000 rp

144 stash result =<< receiveChanTimeout 1000000 rp

155− kill agent "finished"

145+ kill agent "finished" >> awaitExit agent

156

146

157testAgentEventHandling :: TestResult Bool -> Process ()

147testAgentEventHandling :: TestResult Bool -> Process ()

158testAgentEventHandling result = do

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

149 let initState = [] :: [MxEvent]

150+

168 (rc, rs) <- newChan

151 (rc, rs) <- newChan

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

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

170 (mxSink $ \ev -> do

153 (mxSink $ \ev -> do

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

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

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

198 faff 2000000 rs

181 faff 2000000 rs

199

182

183+ _ <- monitor agentPid

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

184+ (sp, rp) <- newChan

201

185

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

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

203

187

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

206 monitor pid

190 monitor pid

207

191

208 send pid ()

192 send pid ()

209− rct <- receiveChanTimeout 10000000 rp

193+ () <- receiveChan rp

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

211

194

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

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

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

214

196

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

216 faff 2000000 rs

198 faff 2000000 rs

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

219 mxNotify (MxSpawned pid, replyTo)

201 mxNotify (MxSpawned pid, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

202 mxNotify (MxProcessDied pid DiedNormal, replyTo)

221

203

222− seenAlive <- receiveChanTimeout 10000000 reply

204+ seenAlive <- receiveChan reply

223− seenDead <- receiveChanTimeout 10000000 reply

205+ seenDead <- receiveChan reply

224

206

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

207+ stash result $ seenAlive && seenDead

226− kill timer "test-complete"

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

227− kill agentPid "test-complete"

228 where

209 where

229 faff delay port = do

210 faff delay port = do

230 res <- receiveChanTimeout delay port

211 res <- receiveChanTimeout delay port

231− unless (isNothing res) $ pause delay

212+ unless (isNothing res) $ do

213+ pause delay

214+ faff delay port

232

215

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

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

234testMxRegEvents result = do

217testMxRegEvents result = do

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

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

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

276

259

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

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

261+ awaitExit agent

278

262

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

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

280testMxRegMon remoteNode result = do

264testMxRegMon remoteNode result = do

@@ -332,99 +316,54 @@ testMxRegMon remoteNode result = do

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

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

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

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

334

318

335− kill agent "test-complete"

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

336

320

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

321ensure :: Process () -> Process () -> Process ()

338ensure = flip finally

322ensure = flip finally

339

323

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

324+type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool

341− -> Maybe LocalNode

342− -> Process ()

343−testNSend op n = do

344− us <- getSelfPid

345− let delay = 5000000

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

347− let isValid = isValidLabel n label

348

325

326+testNSend :: (String -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()

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

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

350− register label p1

328+ let delay = 5000000

351− reg1 <- receiveChanTimeout delay sink

329+ let label = "testMxSend"

352− case reg1 of

353− Just (MxRegistered pd lb)

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

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

356

330

331+ register label p1

332+ reg1 <- receiveChanTimeout delay sink

333+ case reg1 of

334+ Just (MxRegistered pd lb)

357− op label ()

335+ | pd == p1 && lb == label -> return ()

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

358

337

359− sent <- receiveChanTimeout delay sink

338+ op label ()

360− case sent of

361− Just (MxSentToName lb by _)

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

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

364

339

365− where

366− isValidLabel nd l1 l2

367− | l2 == l2 = True

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

369− | otherwise = False

370−

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

372− -> Maybe LocalNode

373− -> Process ()

374−testSend op n = do

375 us <- getSelfPid

340 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

341+ sent <- receiveChanTimeout delay sink

383− case sent of

342+ case sent of

384− Just (MxSent pidTo pidFrom _)

343+ Just (MxSentToName lb by _)

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

344+ | by == us && lb == label -> return True

386− _ -> return False

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

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

346

347+testSend :: (ProcessId -> () -> Process ()) -> Maybe LocalNode -> TestResult Bool -> Process ()

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

405 -- initiate a send

349 -- initiate a send

406− op ch' ()

350+ op p1 ()

407

351

408 -- verify the management event

352 -- verify the management event

353+ us <- getSelfPid

409 sent <- receiveChanTimeout 5000000 sink

354 sent <- receiveChanTimeout 5000000 sink

410 case sent of

355 case sent of

411− Just (MxSentToPort sId spId _)

356+ Just (MxSent pidTo pidFrom _)

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

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

413− _ -> return False

358+ _ -> return False

414−

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

416

359

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

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

418−testMxSend mNode label test = do

361+testMxSend mNode result test = do

419 us <- getSelfPid

362 us <- getSelfPid

420− (sp, rp) <- newChan

421 (chan, sink) <- newChan

363 (chan, sink) <- newChan

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

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

423 mxSink $ \ev -> do

365 mxSink $ \ev -> do

424 case ev of

366 case ev of

425− m@(MxSentToPort _ cid _)

426− | cid /= sendPortId chan

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

428 m@(MxSent _ fromPid _)

367 m@(MxSent _ fromPid _)

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

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

430 m@(MxSentToName _ fromPid _)

369 m@(MxSentToName _ fromPid _)

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

435 mxReady

374 mxReady

436 ]

375 ]

437

376

377+ (sp, rp) <- newChan

438 case mNode of

378 case mNode of

439 Nothing -> void $ spawnLocal (proc sp)

379 Nothing -> void $ spawnLocal (proc sp)

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

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

441

381

442− p1 <- receiveChanTimeout 2000000 rp

382+ p1 <- receiveChan rp

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

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

383+ res <- try (test p1 sink)

445−

446− kill agent "bye"

447− kill (fromJust p1) "bye"

448−

449 case res of

384 case res of

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

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

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

386+ Right tr -> stash result tr

452−

387+ kill agent "bye"

388+ kill p1 "bye"

453

389

454 where

390 where

391+ label = "testMxSend"

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

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

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

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

394

467tests :: TestTransport -> IO [Test]

395tests :: TestTransport -> IO [Test]

468tests TestTransport{..} = do

396tests TestTransport{..} = do

469 node1 <- newLocalNode testTransport initRemoteTable

397 node1 <- newLocalNode testTransport initRemoteTable

470 node2 <- newLocalNode testTransport initRemoteTable

398 node2 <- newLocalNode testTransport initRemoteTable

399+ let nid = localNodeId node2

471 return [

400 return [

472 testGroup "MxAgents" [

401 testGroup "MxAgents" [

473 testCase "EventHandling"

402 testCase "EventHandling"

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

477 , testCase "InterAgentBroadcast"

406 , testCase "InterAgentBroadcast"

478 (delayedAssertion

407 (delayedAssertion

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

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

480− node1 (Just ()) testAgentBroadcast)

409+ node1 () testAgentBroadcast)

481 , testCase "AgentMailboxHandling"

410 , testCase "AgentMailboxHandling"

482 (delayedAssertion

411 (delayedAssertion

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

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

@@ -502,41 +431,53 @@ tests TestTransport{..} = do

502 (delayedAssertion

431 (delayedAssertion

503 "expected process deaths to result in unregistration events"

432 "expected process deaths to result in unregistration events"

504 node1 () (testMxRegMon node2))

433 node1 () (testMxRegMon node2))

505− , testGroup "SendEvents" $ buildTestCases node1 node2

434+ , testGroup "SentEvents" [

506− ]

435+ testGroup "RemoteTargets" [

507− ]

436+ testCase "Unsafe.nsend"

508− where

437+ (delayedAssertion "expected mx events failed"

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

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

510− ("NSend", [

439+ , testCase "Unsafe.nsendRemote"

440+ (delayedAssertion "expected mx events failed"

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

442+ , testCase "Unsafe.send"

443+ (delayedAssertion "expected mx events failed"

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

445+ , testCase "Unsafe.usend"

446+ (delayedAssertion "expected mx events failed"

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

448+ , testCase "nsend"

449+ (delayedAssertion "expected mx events failed"

511− ("nsend", testNSend nsend)

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

451+ , testCase "nsendRemote"

452+ (delayedAssertion "expected mx events failed"

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

454+ , testCase "send"

455+ (delayedAssertion "expected mx events failed"

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

457+ , testCase "usend"

458+ (delayedAssertion "expected mx events failed"

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

460+ ]

461+ , testGroup "LocalTargets" [

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

462+ testCase "Unsafe.nsend"

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

463+ (delayedAssertion "expected mx events failed"

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

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

515− ])

465+ , testCase "Unsafe.send"

516− , ("Send", [

466+ (delayedAssertion "expected mx events failed"

517− ("send", testSend send)

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

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

519− , ("usend", testSend usend)

468+ , testCase "Unsafe.usend"

469+ (delayedAssertion "expected mx events failed"

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

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

521− , ("sendChan", testChan sendChan)

471+ , testCase "nsend"

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

472+ (delayedAssertion "expected mx events failed"

523− ])

473+ node1 True (testNSend nsend Nothing))

524− , ("Forward", [

474+ , testCase "send"

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

475+ (delayedAssertion "expected mx events failed"

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

476+ node1 True (testSend send Nothing))

527− ])

477+ , testCase "usend"

478+ (delayedAssertion "expected mx events failed"

479+ node1 True (testSend usend Nothing))

528 ]

480 ]

529−

481+ ]

530− build :: LocalNode

482+ ]

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 ]

483 ]

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+39−54

@@ -254,15 +254,17 @@ send them msg = do

254 node = processNode proc

255 nodeId = localNodeId node

256 destNode = (processNodeId them) in do

257+ case destNode == nodeId of

258+ True -> sendLocal them msg

259+ False -> liftIO $ sendMessage (processNode proc)

260+ (ProcessIdentifier (processId proc))

261+ (ProcessIdentifier them)

262+ NoImplicitReconnect

263+ msg

264+ -- We do not fire the trace event until after the sending is complete;

265+ -- In the remote case, 'sendMessage' can block in the networking stack.

257 liftIO $ traceEvent (localEventBus node)

266 liftIO $ traceEvent (localEventBus node)

258 (MxSent them us (createUnencodedMessage msg))

267 (MxSent them us (createUnencodedMessage msg))

259− if destNode == nodeId

260− then sendLocal them msg

261− else liftIO $ sendMessage (processNode proc)

262− (ProcessIdentifier (processId proc))

263− (ProcessIdentifier them)

264− NoImplicitReconnect

265− msg

266

268

267-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

269-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize

268-- and (in the case when the destination process resides on the same local

270-- and (in the case when the destination process resides on the same local

@@ -282,12 +284,9 @@ unsafeSend = Unsafe.send

282--

284--

283usend :: Serializable a => ProcessId -> a -> Process ()

285usend :: Serializable a => ProcessId -> a -> Process ()

284usend them msg = do

286usend them msg = do

285− proc <- ask

287+ here <- getSelfNode

286 let there = processNodeId them

288 let there = processNodeId them

287− let (us, node) = (processId proc, processNode proc)

289+ if here == there

288− let msg' = wrapMessage msg

289− liftIO $ traceEvent (localEventBus node) (MxSent them us msg')

290− if localNodeId (processNode proc) == there

291 then sendLocal them msg

290 then sendLocal them msg

292 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

291 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

293 (createMessage msg)

292 (createMessage msg)

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

307-- Channels --

306-- Channels --

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

307--------------------------------------------------------------------------------

309

308

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

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

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

318newChan = do

311newChan = do

319 proc <- ask

312 proc <- ask

@@ -341,16 +334,13 @@ newChan = do

341sendChan :: Serializable a => SendPort a -> a -> Process ()

334sendChan :: Serializable a => SendPort a -> a -> Process ()

342sendChan (SendPort cid) msg = do

335sendChan (SendPort cid) msg = do

343 proc <- ask

336 proc <- ask

344− let node = processNode proc

345− pid = processId proc

346− us = localNodeId node

337+ let node = localNodeId (processNode proc)

347− them = processNodeId (sendPortProcessId cid) in do

338+ destNode = processNodeId (sendPortProcessId cid) in do

348− liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)

349− case them == us of

339+ case destNode == node of

350 True -> sendChanLocal cid msg

340 True -> sendChanLocal cid msg

351 False -> do

341 False -> do

352− liftIO $ sendBinary node

342+ liftIO $ sendBinary (processNode proc)

353− (ProcessIdentifier pid)

343+ (ProcessIdentifier (processId proc))

354 (SendPortIdentifier cid)

344 (SendPortIdentifier cid)

355 NoImplicitReconnect

345 NoImplicitReconnect

356 msg

346 msg

@@ -494,14 +484,15 @@ forward msg them = do

494 us = processId proc

484 us = processId proc

495 nid = localNodeId node

485 nid = localNodeId node

496 destNode = (processNodeId them) in do

486 destNode = (processNodeId them) in do

497− liftIO $ traceEvent (localEventBus node) (MxSent them us msg)

498− if destNode == nid

487+ case destNode == nid of

499− then sendCtrlMsg Nothing (LocalSend them msg)

488+ True -> sendCtrlMsg Nothing (LocalSend them msg)

500− else liftIO $ sendPayload (processNode proc)

489+ False ->

490+ liftIO $ do

501− (ProcessIdentifier (processId proc))

491+ sendPayload (processNode proc) (ProcessIdentifier (processId proc)) (ProcessIdentifier them) NoImplicitReconnect (messageToPayload msg)

502− (ProcessIdentifier them)

492+ -- We do not fire the trace event until after the sending is complete;

503− NoImplicitReconnect

493+ -- In this remote case, 'sendMessage' can block in the networking stack.

504− (messageToPayload msg)

494+ liftIO $ traceEvent (localEventBus node)

495+ (MxSent them us msg)

505

496

506-- | Forward a raw 'Message' to the given 'ProcessId'.

497-- | Forward a raw 'Message' to the given 'ProcessId'.

507--

498--

@@ -515,10 +506,14 @@ uforward msg them = do

515 us = processId proc

506 us = processId proc

516 nid = localNodeId node

507 nid = localNodeId node

517 destNode = (processNodeId them) in do

508 destNode = (processNodeId them) in do

518− liftIO $ traceEvent (localEventBus node) (MxSent them us msg)

519− if destNode == nid

509+ case destNode == nid of

520− then sendCtrlMsg Nothing (LocalSend them msg)

510+ True -> sendCtrlMsg Nothing (LocalSend them msg)

511+ False -> do

521− else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg

512+ sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg

513+ -- We do not fire the trace event until after the sending is complete;

514+ -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

515+ liftIO $ traceEvent (localEventBus node)

516+ (MxSent them us msg)

522

517

523-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are

518-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are

524-- 'Serializable' - like the datum they contain - but also note, deserialising

519-- 'Serializable' - like the datum they contain - but also note, deserialising

@@ -1227,12 +1222,8 @@ whereisRemoteAsync nid label = do

1227

1222

1228-- | Named send to a process in the local registry (asynchronous)

1223-- | Named send to a process in the local registry (asynchronous)

1229nsend :: Serializable a => String -> a -> Process ()

1224nsend :: Serializable a => String -> a -> Process ()

1230−nsend label msg = do

1225+nsend label msg =

1231− proc <- ask

1232− let msg' = createUnencodedMessage msg

1233− liftIO $ traceEvent (localEventBus (processNode proc))

1234− (MxSentToName label (processId proc) msg')

1235− sendCtrlMsg Nothing (NamedSend label msg')

1226+ sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))

1236

1227

1237-- | Named send to a process in the local registry (asynchronous).

1228-- | Named send to a process in the local registry (asynchronous).

1238-- This function makes /no/ attempt to serialize and (in the case when the

1229-- This function makes /no/ attempt to serialize and (in the case when the

@@ -1244,15 +1235,9 @@ unsafeNSend = Unsafe.nsend

1244-- | Named send to a process in a remote registry (asynchronous)

1235-- | Named send to a process in a remote registry (asynchronous)

1245nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

1236nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

1246nsendRemote nid label msg = do

1237nsendRemote nid label msg = do

1247− proc <- ask

1238+ here <- getSelfNode

1248− let us = processId proc

1249− let node = processNode proc

1250− if localNodeId node == nid

1251− then nsend label msg

1239+ if here == nid then nsend label msg

1252− else let lbl = label ++ "@" ++ show nid in do

1253− liftIO $ traceEvent (localEventBus node)

1254− (MxSentToName lbl us (wrapMessage msg))

1255− sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

1240+ else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

1256

1241

1257-- | Named send to a process in a remote registry (asynchronous)

1242-- | Named send to a process in a remote registry (asynchronous)

1258-- This function makes /no/ attempt to serialize and (in the case when the

1243-- This function makes /no/ attempt to serialize and (in the case when the

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:

src/Control/Distributed/Process/Management/Internal/Types.hs−5

@@ -21,7 +21,6 @@ import Control.Concurrent.STM

21import Control.Distributed.Process.Internal.Types

22 ( Process

23 , ProcessId

24− , SendPortId

25 , Message

24 , Message

26 , DiedReason

25 , DiedReason

27 , NodeId

26 , NodeId

@@ -60,12 +59,8 @@ data MxEvent =

60 -- ^ fired whenever a message is sent from a local process

59 -- ^ fired whenever a message is sent from a local process

61 | MxSentToName String ProcessId Message

60 | MxSentToName String ProcessId Message

62 -- ^ fired whenever a named send occurs

61 -- ^ fired whenever a named send occurs

63− | MxSentToPort ProcessId SendPortId Message

64− -- ^ fired whenever a sendChan occurs

65 | MxReceived ProcessId Message

62 | MxReceived ProcessId Message

66 -- ^ fired whenever a message is received by a local process

63 -- ^ fired whenever a message is received by a local process

67− | MxReceivedPort SendPortId Message

68− -- ^ fired whenever a message is received via a typed channel

69 | MxConnected ConnectionId EndPointAddress

64 | MxConnected ConnectionId EndPointAddress

70 -- ^ fired when a network-transport connection is first established

65 -- ^ fired when a network-transport connection is first established

71 | MxDisconnected ConnectionId EndPointAddress

66 | MxDisconnected ConnectionId EndPointAddress

src/Control/Distributed/Process/Node.hs+10−15

@@ -492,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)

492data IncomingTarget =

493 Uninit

494 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan SendPortId TypedChannel

495+ | ToChan TypedChannel

496 | ToNode

497

498data ConnectionState = ConnectionState {

@@ -547,17 +547,13 @@ handleIncomingMessages node = go initConnectionState

547 enqueue queue msg -- 'enqueue' is strict

548 trace node (MxReceived pid msg)

549 go st

550− Just (_, ToChan chId (TypedChannel chan')) -> do

550+ Just (_, ToChan (TypedChannel chan')) -> do

551 mChan <- deRefWeak chan'

552 -- If mChan is Nothing, the process has given up the read end of

553 -- the channel and we simply ignore the incoming message

554− forM_ mChan $ \chan -> do

554+ forM_ mChan $ \chan -> atomically $

555− msg' <- atomically $ do

556− msg <- return $! decode (BSL.fromChunks payload)

557− -- We make sure the message is fully decoded when it is enqueued

555+ -- We make sure the message is fully decoded when it is enqueued

558− writeTQueue chan msg

556+ writeTQueue chan $! decode (BSL.fromChunks payload)

559− return msg

560− trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

561 go st

557 go st

562 Just (_, ToNode) -> do

558 Just (_, ToNode) -> do

563 let ctrlMsg = decode . BSL.fromChunks $ payload

559 let ctrlMsg = decode . BSL.fromChunks $ payload

@@ -584,7 +580,7 @@ handleIncomingMessages node = go initConnectionState

584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

580 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)

585 case mChannel of

581 case mChannel of

586 Just channel ->

582 Just channel ->

587− go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)

583+ go (incomingAt cid ^= Just (src, ToChan channel) $ st)

588 Nothing ->

584 Nothing ->

589 invalidRequest cid st $

585 invalidRequest cid st $

590 "incoming attempt to connect to unknown channel of"

586 "incoming attempt to connect to unknown channel of"

@@ -1054,12 +1050,11 @@ ncEffectLocalPortSend from msg = do

1054 -- If ch is Nothing, the process has given up the read end of

1050 -- If ch is Nothing, the process has given up the read end of

1055 -- the channel and we simply ignore the incoming message - this

1051 -- the channel and we simply ignore the incoming message - this

1056 ch <- deRefWeak chan'

1052 ch <- deRefWeak chan'

1057− forM_ ch $ \chan -> deliverChan node from msg chan

1053+ forM_ ch $ \chan -> deliverChan msg chan

1058− where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()

1054+ where deliverChan :: forall a . Message -> TQueue a -> IO ()

1059− deliverChan n p (UnencodedMessage _ raw) chan' = do

1055+ deliverChan (UnencodedMessage _ raw) chan' =

1060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1056 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)

1061− trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1062− deliverChan _ _ (EncodedMessage _ _) _ =

1057+ deliverChan (EncodedMessage _ _) _ =

1063 -- this will not happen unless someone screws with Primitives.hs

1058 -- this will not happen unless someone screws with Primitives.hs

1064 error "invalid local channel delivery"

1059 error "invalid local channel delivery"

1065

1060

src/Control/Distributed/Process/UnsafePrimitives.hs+32−26

@@ -104,25 +104,42 @@ nsend :: Serializable a => String -> a -> Process ()

104nsend label msg = do

105 proc <- ask

106 let us = processId proc

107+ let node = localNodeId (processNode proc)

107 let msg' = wrapMessage msg

108 let msg' = wrapMessage msg

108 -- see [note: tracing]

109 -- see [note: tracing]

109 liftIO $ traceEvent (localEventBus (processNode proc))

110 liftIO $ traceEvent (localEventBus (processNode proc))

110 (MxSentToName label us msg')

111 (MxSentToName label us msg')

111 sendCtrlMsg Nothing (NamedSend label msg')

112 sendCtrlMsg Nothing (NamedSend label msg')

112

113

114+-- [note: tracing]

115+-- In the remote case, we do not fire a trace event until after sending is

116+-- complete, since 'sendMessage' can block in the networking stack. For locally

117+-- registered processes, we trace before sending the NC a control message, since

118+-- we cannot guarantee ordering otherwise.

119+--

120+-- In addition, MxSent trace messages are dispatched by the node controller's

121+-- thread for local sends, which also covers usend.

122+--

123+-- Also note that tracing writes to the local node's control channel, and this

124+-- module explicitly specifies to its clients that it does unsafe message

125+-- encoding. The same is true for the messages it puts onto the Management

126+-- event bus, however we do *not* want unevaluated thunks hitting the event

127+-- bus control thread. Hence the word /Unsafe/ in this module's name!

128+--

129+

113-- | Named send to a process in a remote registry (asynchronous)

130-- | Named send to a process in a remote registry (asynchronous)

114nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

131nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()

115nsendRemote nid label msg = do

132nsendRemote nid label msg = do

116 proc <- ask

133 proc <- ask

117 let us = processId proc

134 let us = processId proc

118 let node = processNode proc

135 let node = processNode proc

119− if localNodeId node == nid

136+ if localNodeId (processNode proc) == nid

120 then nsend label msg

137 then nsend label msg

121 else

138 else

122 let lbl = label ++ "@" ++ show nid in do

139 let lbl = label ++ "@" ++ show nid in do

123 -- see [note: tracing] NB: this is a remote call to another NC...

140 -- see [note: tracing] NB: this is a remote call to another NC...

124 liftIO $ traceEvent (localEventBus node)

141 liftIO $ traceEvent (localEventBus node)

125− (MxSentToName lbl us (wrapMessage msg))

142+ (MxSentToName label us (wrapMessage msg))

126 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

143 sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))

127

144

128-- | Send a message

145-- | Send a message

@@ -159,41 +176,30 @@ usend them msg = do

159 let there = processNodeId them

176 let there = processNodeId them

160 let (us, node) = (processId proc, processNode proc)

177 let (us, node) = (processId proc, processNode proc)

161 let msg' = wrapMessage msg

178 let msg' = wrapMessage msg

162− -- see [note: tracing]

163 liftIO $ traceEvent (localEventBus node) (MxSent them us msg')

179 liftIO $ traceEvent (localEventBus node) (MxSent them us msg')

164 if localNodeId (processNode proc) == there

180 if localNodeId (processNode proc) == there

165 then sendCtrlMsg Nothing $ LocalSend them msg'

181 then sendCtrlMsg Nothing $ LocalSend them msg'

166 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

182 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)

167 (createMessage msg)

183 (createMessage msg)

168

184

169−-- [note: tracing]

170−-- Note that tracing writes to the local node's control channel, and this

171−-- module explicitly specifies to its clients that it does unsafe message

172−-- encoding. The same is true for the messages it puts onto the Management

173−-- event bus, however we do *not* want unevaluated thunks hitting the event

174−-- bus control thread. Hence the word /Unsafe/ in this module's name!

175−--

176−

177-- | Send a message on a typed channel

185-- | Send a message on a typed channel

178sendChan :: Serializable a => SendPort a -> a -> Process ()

186sendChan :: Serializable a => SendPort a -> a -> Process ()

179sendChan (SendPort cid) msg = do

187sendChan (SendPort cid) msg = do

180 proc <- ask

188 proc <- ask

181− let node = processNode proc

182− pid = processId proc

183− us = localNodeId node

189+ let node = localNodeId (processNode proc)

184− them = processNodeId (sendPortProcessId cid)

190+ destNode = processNodeId (sendPortProcessId cid) in do

185− msg' = wrapMessage msg in do

191+ case destNode == node of

186− liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')

187− if them == us

188− then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!

192+ True -> unsafeSendChanLocal cid msg

193+ False ->

189− else liftIO $ sendBinary node

194+ liftIO $ sendBinary (processNode proc)

190− (ProcessIdentifier pid)

195+ (ProcessIdentifier (processId proc))

191− (SendPortIdentifier cid)

196+ (SendPortIdentifier cid)

192− NoImplicitReconnect

197+ NoImplicitReconnect

193− msg

198+ msg

194 where

199 where

195− unsafeSendChanLocal :: SendPortId -> Message -> Process ()

200+ unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()

201+ unsafeSendChanLocal spId msg' =

196− unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m

202+ sendCtrlMsg Nothing $ LocalPortSend spId (wrapMessage msg')

197

203

198-- | Create an unencoded @Message@ for any @Serializable@ type.

204-- | Create an unencoded @Message@ for any @Serializable@ type.

199wrapMessage :: Serializable a => a -> Message

205wrapMessage :: Serializable a => a -> Message