0514f3c..master · distributed-process

git.iko.soy/distributed-process

on master help sign in

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

since splitsnapshots

unifiedsplit

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

6license: BSD3

7license-file: LICENSE

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

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

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

10copyright: Well-Typed LLP

11category: Control, Cloud Haskell

12build-type: Simple

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+110−32

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

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

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

49−

49+import Data.Maybe (isNothing, isJust)

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

51import Test.Framework (Test, testGroup)

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

53−import Control.Rematch hiding (match)

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

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

56newtype Ping = Ping ProcessId

@@ -87,10 +87,9 @@ 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

95-- | Quick and dirty synchronous version of whereisRemoteAsync

94-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

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

240 pid <- getSelfPid

239 pid <- getSelfPid

241 replicateM_ numPings $ do

240 replicateM_ numPings $ do

242 send pingServer (Pong pid)

241 send pingServer (Pong pid)

243− p <- expect

242+ p <- expectTimeout 3000000

244 case p of

243 case p of

245− Ping _ -> return ()

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

246− _ -> die "Unexpected message"

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

247

246

248 putMVar clientDone ()

247 putMVar clientDone ()

249

248

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

560 ss <- liftIO $ mapM readMVar vs

559 ss <- liftIO $ mapM readMVar vs

561 case ss of

560 case ss of

562 [sa, sb, sc] ->

561 [sa, sb, sc] ->

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

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

564 [ -- a, b, c

563 [ -- a, b, c

565 (sa, 'a')

564 (sa, 'a')

566 , (sb, 'b')

565 , (sb, 'b')

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

736testRemoteRegistry TestTransport{..} = do

735testRemoteRegistry TestTransport{..} = do

737 node1 <- newLocalNode testTransport initRemoteTable

736 node1 <- newLocalNode testTransport initRemoteTable

738 node2 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

739− done <- newEmptyMVar

740

738

741 pingServer <- forkProcess node1 ping

739 pingServer <- forkProcess node1 ping

742 deadProcess <- forkProcess node1 (return ())

740 deadProcess <- forkProcess node1 (return ())

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

751

749

752 pid <- verifyWhereIsRemote nid1 "ping"

750 pid <- verifyWhereIsRemote nid1 "ping"

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

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

752+

754 us <- getSelfPid

753 us <- getSelfPid

755 nsendRemote nid1 "ping" (Pong us)

754 nsendRemote nid1 "ping" (Pong us)

756 receiveWait [

755 receiveWait [

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

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

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

761 -- in owner field.

760 -- in owner field.

762 registerRemoteAsync nid1 "dead" deadProcess

761 registerRemoteAsync nid1 "dead" deadProcess

763− receiveWait [

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

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

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

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

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

765+

766 registerRemoteAsync nid1 "ping" deadProcess

767 receiveWait [

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

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

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

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

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

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

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

772+

771 unregisterRemoteAsync nid1 "dead"

773 unregisterRemoteAsync nid1 "dead"

772 receiveWait [

774 receiveWait [

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

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

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

779

778

780testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

781testRemoteRegistryRemoteProcess TestTransport{..} = do

780testRemoteRegistryRemoteProcess TestTransport{..} = do

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

1273

1272

1274 takeMVar done

1273 takeMVar done

1275

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+

1276testKillLocal :: TestTransport -> Assertion

1345testKillLocal :: TestTransport -> Assertion

1277testKillLocal TestTransport{..} = do

1346testKillLocal TestTransport{..} = do

1278 localNode <- newLocalNode testTransport initRemoteTable

1347 localNode <- newLocalNode testTransport initRemoteTable

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

1487testRegistryMonitoring TestTransport{..} = do

1556testRegistryMonitoring TestTransport{..} = do

1488 node1 <- newLocalNode testTransport initRemoteTable

1557 node1 <- newLocalNode testTransport initRemoteTable

1489 node2 <- newLocalNode testTransport initRemoteTable

1558 node2 <- newLocalNode testTransport initRemoteTable

1490− waitH <- newEmptyMVar

1491

1559

1492 let nid = localNodeId node2

1560 let nid = localNodeId node2

1493 pid <- forkProcess node1 $ do

1561 pid <- forkProcess node1 $ do

1562+ self <- getSelfPid

1494− getSelfPid >>= runUntilRegistered nid

1563+ runUntilRegistered nid self

1495− liftIO $ takeMVar waitH

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

1565+ expect :: Process ()

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

1496

1567

1497 runProcess node2 $ do

1568 runProcess node2 $ do

1498 register regName pid

1569 register regName pid

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

1499 res <- whereis regName

1571 res <- whereis regName

1572+ send pid ()

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

1500 us <- getSelfPid

1574 us <- getSelfPid

1501− liftIO $ do

1502− putMVar waitH ()

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

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

1576+

1504

1577

1505 -- This delay isn't essential!

1578 -- This delay isn't essential!

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

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

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

1525 regHere <- newEmptyMVar

1598 regHere <- newEmptyMVar

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

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

1527 res <- takeMVar regHere

1600 res <- takeMVar regHere

1601+ case res of

1602+ Nothing -> return ()

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

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

1529

1604

1530 where

1605 where

1531 runUntilRegistered nid us = do

1606 runUntilRegistered nid us = do

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

1537

1612

1538 delayUntilMaybeUnregistered nid p = do

1613 delayUntilMaybeUnregistered nid p = do

1539 whereisRemoteAsync nid regName

1614 whereisRemoteAsync nid regName

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

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

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

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

1542 (const $ return ())

1617 (const $ return ())

1543 ]

1618 ]

1619+ case res of

1544− return ()

1620+ Just () -> return ()

1621+ Nothing -> delayUntilMaybeUnregistered nid p

1545

1622

1546 regName = "testRegisterRemote"

1623 regName = "testRegisterRemote"

1547

1624

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

1737 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1738 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1739 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1740 , testCase "KillLocal" (testKillLocal testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1741 , testCase "KillRemote" (testKillRemote testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1742 , testCase "Die" (testDie testtrans)

1820 , testCase "Die" (testDie testtrans)

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

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

91import Test.HUnit.Base (assertBool)

93import GHC.Generics

94+import System.Timeout (timeout)

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

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

96type TestResult a = MVar a

97type TestResult a = MVar a

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

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

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

234tryRunProcess node p = do

235tryRunProcess node p = do

235 tid <- liftIO myThreadId

236 tid <- liftIO myThreadId

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

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

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

237

239

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

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

239tryForkProcess node p = do

241tryForkProcess node p = do

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+172−113

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

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2+{-# LANGUAGE ParallelListComp #-}

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

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

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

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

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

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

6−import Control.Exception (SomeException)

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

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

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

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

10+ ( ProcessExitException(..)

11+ , unsafeCreateUnencodedMessage

12+ )

9import Control.Distributed.Process.Node

13import Control.Distributed.Process.Node

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

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

11 ( send

15 ( send

12 , nsend

16 , nsend

13 , nsendRemote

17 , nsendRemote

14 , usend

18 , usend

19+ , sendChan

15 )

20 )

16import Control.Distributed.Process.Management

21import Control.Distributed.Process.Management

17 ( MxEvent(..)

22 ( MxEvent(..)

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

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

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

33import Control.Rematch (equalTo)

38import Control.Rematch (equalTo)

34import Data.Binary

39import Data.Binary

35−import Data.List (find, sort)

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

36−import Data.Maybe (isJust, isNothing)

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

37import Data.Typeable

42import Data.Typeable

38import GHC.Generics hiding (from)

43import GHC.Generics hiding (from)

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

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

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

61 where

66 where

62 withMonitorRef p = bracket (monitor p) unmonitor

67 withMonitorRef p = bracket (monitor p) unmonitor

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

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

65testAgentBroadcast result = do

70testAgentBroadcast result = do

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

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

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

76 mxNotify ()

81 mxNotify ()

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

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

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

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

79− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

81− kill publisher "finished" >> awaitExit publisher

86+ kill publisher "finished"

82− kill consumer "finished" >> awaitExit consumer

87+ kill consumer "finished"

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

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

85testAgentDualInput result = do

90testAgentDualInput result = do

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

94 else mxReady)

99 else mxReady)

95 ]

100 ]

101

102+ mRef <- monitor s

103+

97 mxNotify (1 :: Int)

104 mxNotify (1 :: Int)

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

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

99 mxNotify (2 :: Int)

106 mxNotify (2 :: Int)

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

101 mxNotify (5 :: Int)

108 mxNotify (5 :: Int)

102

109

103 stash result =<< receiveChanTimeout 10000000 rp

110 stash result =<< receiveChanTimeout 10000000 rp

104− awaitExit s

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

105

115

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

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

107testAgentPrioritisation result = do

117testAgentPrioritisation result = do

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

115 (sp, rp) <- newChan

125 (sp, rp) <- newChan

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

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

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

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

118− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

119 st <- mxGetLocal

129 st <- mxGetLocal

120 case length st of

130 case length st of

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

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

122 mxDeactivate "finished"

132 mxDeactivate "finished"

123 _ -> mxReceive -- go to the mailbox

133 _ -> mxReceive -- go to the mailbox

124 )

134 )

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

129 mxNotify "fourth"

139 mxNotify "fourth"

130 nsend name "fifth"

140 nsend name "fifth"

131

141

132− stash result . sort =<< receiveChan rp

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

133 awaitExit s

143 awaitExit s

134

144

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

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

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

142 nsend "mailbox-agent" ()

152 nsend "mailbox-agent" ()

143

153

144 stash result =<< receiveChanTimeout 1000000 rp

154 stash result =<< receiveChanTimeout 1000000 rp

145− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

146

156

147testAgentEventHandling :: TestResult Bool -> Process ()

157testAgentEventHandling :: TestResult Bool -> Process ()

148testAgentEventHandling result = do

158testAgentEventHandling result = do

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

159+ us <- getSelfPid

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

161+ timer <- spawnLocal $ do

162+ pause (1000000 * 60 * 5)

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

164+ stash result False

165+ kill us "Test Timed Out"

150

166

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

151 (rc, rs) <- newChan

168 (rc, rs) <- newChan

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

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

153 (mxSink $ \ev -> do

170 (mxSink $ \ev -> do

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

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

181 faff 2000000 rs

198 faff 2000000 rs

182

199

183− _ <- monitor agentPid

184− (sp, rp) <- newChan

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

185

201

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

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

187

203

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

190 monitor pid

206 monitor pid

191

207

192 send pid ()

208 send pid ()

193− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

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

194

211

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

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

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

196

214

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

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

198 faff 2000000 rs

216 faff 2000000 rs

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

201 mxNotify (MxSpawned pid, replyTo)

219 mxNotify (MxSpawned pid, replyTo)

202 mxNotify (MxProcessDied pid DiedNormal, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

203

221

204− seenAlive <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

205− seenDead <- receiveChan reply

223+ seenDead <- receiveChanTimeout 10000000 reply

206

224

207− stash result $ seenAlive && seenDead

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

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

209 where

228 where

210 faff delay port = do

229 faff delay port = do

211 res <- receiveChanTimeout delay port

230 res <- receiveChanTimeout delay port

212− unless (isNothing res) $ do

231+ unless (isNothing res) $ pause delay

213− pause delay

214− faff delay port

215

232

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

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

217testMxRegEvents result = do

234testMxRegEvents result = do

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

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

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

259

276

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

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

261− awaitExit agent

262

278

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

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

264testMxRegMon remoteNode result = do

280testMxRegMon remoteNode result = do

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

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

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

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

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

318

334

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

335+ kill agent "test-complete"

320

336

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

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

322ensure = flip finally

338ensure = flip finally

323

339

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

325−

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

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

341+ -> Maybe LocalNode

342+ -> Process ()

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

343+testNSend op n = do

344+ us <- getSelfPid

328 let delay = 5000000

345 let delay = 5000000

329− let label = "testMxSend"

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

347+ let isValid = isValidLabel n label

348+

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

350+ register label p1

351+ reg1 <- receiveChanTimeout delay sink

352+ case reg1 of

353+ Just (MxRegistered pd lb)

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

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

330

356

331− register label p1

332− reg1 <- receiveChanTimeout delay sink

333− case reg1 of

334− Just (MxRegistered pd lb)

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

357+ op label ()

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

337

358

338− op label ()

359+ sent <- receiveChanTimeout delay sink

360+ case sent of

361+ Just (MxSentToName lb by _)

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

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

339

364

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

340 us <- getSelfPid

375 us <- getSelfPid

341− sent <- receiveChanTimeout delay sink

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

342− case sent of

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

343− Just (MxSentToName lb by _)

378+ -- initiate a send

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

379+ op p1 ()

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

380+

381+ -- verify the management event

382+ sent <- receiveChanTimeout 5000000 sink

383+ case sent of

384+ Just (MxSent pidTo pidFrom _)

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

386+ _ -> return False

387+

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

389+ -> Maybe LocalNode

390+ -> Process ()

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

392+

393+ us <- getSelfPid

394+ send p1 us

395+

396+ cleared <- receiveChanTimeout 2000000 sink

397+ case cleared of

398+ Just (MxSent pidTo pidFrom _)

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

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

401+

402+ chan <- receiveTimeout 5000000 [ match return ]

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

346

404

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

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

349 -- initiate a send

405 -- initiate a send

350− op p1 ()

406+ op ch' ()

351

407

352 -- verify the management event

408 -- verify the management event

353− us <- getSelfPid

354 sent <- receiveChanTimeout 5000000 sink

409 sent <- receiveChanTimeout 5000000 sink

355 case sent of

410 case sent of

356− Just (MxSent pidTo pidFrom _)

411+ Just (MxSentToPort sId spId _)

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

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

358− _ -> return False

413+ _ -> return False

414+

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

359

416

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

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

361−testMxSend mNode result test = do

418+testMxSend mNode label test = do

362 us <- getSelfPid

419 us <- getSelfPid

420+ (sp, rp) <- newChan

363 (chan, sink) <- newChan

421 (chan, sink) <- newChan

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

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

365 mxSink $ \ev -> do

423 mxSink $ \ev -> do

366 case ev of

424 case ev of

425+ m@(MxSentToPort _ cid _)

426+ | cid /= sendPortId chan

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

367 m@(MxSent _ fromPid _)

428 m@(MxSent _ fromPid _)

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

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

369 m@(MxSentToName _ fromPid _)

430 m@(MxSentToName _ fromPid _)

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

374 mxReady

435 mxReady

375 ]

436 ]

376

437

377− (sp, rp) <- newChan

378 case mNode of

438 case mNode of

379 Nothing -> void $ spawnLocal (proc sp)

439 Nothing -> void $ spawnLocal (proc sp)

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

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

381

441

382− p1 <- receiveChan rp

442+ p1 <- receiveChanTimeout 2000000 rp

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

383− res <- try (test p1 sink)

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

384− case res of

445+

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

386− Right tr -> stash result tr

387 kill agent "bye"

446 kill agent "bye"

388− kill p1 "bye"

447+ kill (fromJust p1) "bye"

448+

449+ case res of

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

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

452+

389

453

390 where

454 where

391− label = "testMxSend"

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

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

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

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

457+

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

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

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

461+

462+ replyChannel p' = do

463+ (s, r) <- newChan

464+ send p' s

465+ return r

394

466

395tests :: TestTransport -> IO [Test]

467tests :: TestTransport -> IO [Test]

396tests TestTransport{..} = do

468tests TestTransport{..} = do

397 node1 <- newLocalNode testTransport initRemoteTable

469 node1 <- newLocalNode testTransport initRemoteTable

398 node2 <- newLocalNode testTransport initRemoteTable

470 node2 <- newLocalNode testTransport initRemoteTable

399− let nid = localNodeId node2

400 return [

471 return [

401 testGroup "MxAgents" [

472 testGroup "MxAgents" [

402 testCase "EventHandling"

473 testCase "EventHandling"

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

406 , testCase "InterAgentBroadcast"

477 , testCase "InterAgentBroadcast"

407 (delayedAssertion

478 (delayedAssertion

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

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

409− node1 () testAgentBroadcast)

480+ node1 (Just ()) testAgentBroadcast)

410 , testCase "AgentMailboxHandling"

481 , testCase "AgentMailboxHandling"

411 (delayedAssertion

482 (delayedAssertion

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

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

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

431 (delayedAssertion

502 (delayedAssertion

432 "expected process deaths to result in unregistration events"

503 "expected process deaths to result in unregistration events"

433 node1 () (testMxRegMon node2))

504 node1 () (testMxRegMon node2))

434− , testGroup "SentEvents" [

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

435− testGroup "RemoteTargets" [

506+ ]

436− testCase "Unsafe.nsend"

507+ ]

437− (delayedAssertion "expected mx events failed"

508+ where

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

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

439− , testCase "Unsafe.nsendRemote"

510+ ("NSend", [

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"

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

511+ ("nsend", testNSend nsend)

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

462− testCase "Unsafe.nsend"

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

463− (delayedAssertion "expected mx events failed"

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

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

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

465− , testCase "Unsafe.send"

515+ ])

466− (delayedAssertion "expected mx events failed"

516+ , ("Send", [

517+ ("send", testSend send)

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

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

468− , testCase "Unsafe.usend"

519+ , ("usend", testSend usend)

469− (delayedAssertion "expected mx events failed"

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

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

471− , testCase "nsend"

521+ , ("sendChan", testChan sendChan)

472− (delayedAssertion "expected mx events failed"

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

473− node1 True (testNSend nsend Nothing))

523+ ])

474− , testCase "send"

524+ , ("Forward", [

475− (delayedAssertion "expected mx events failed"

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

476− node1 True (testSend send Nothing))

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

477− , testCase "usend"

527+ ])

478− (delayedAssertion "expected mx events failed"

479− node1 True (testSend usend Nothing))

480 ]

528 ]

529+

530+ build :: LocalNode

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

481− ]

539+ ]

482− ]

540+ | (groupName, groupCases) <- specs

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

483 ]

542 ]

distributed-process.cabal+1−1

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

6License-File: LICENSE

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

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

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

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

10Stability: experimental

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

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

src/Control/Distributed/Process/Internal/Primitives.hs+54−39

@@ -254,17 +254,15 @@ 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.

266 liftIO $ traceEvent (localEventBus node)

257 liftIO $ traceEvent (localEventBus node)

267 (MxSent them us (createUnencodedMessage msg))

258 (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

268

266

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

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

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

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

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

284--

282--

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

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

286usend them msg = do

284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288 let there = processNodeId them

286 let there = processNodeId them

289− if here == there

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

288+ let msg' = wrapMessage msg

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

290+ if localNodeId (processNode proc) == there

290 then sendLocal them msg

291 then sendLocal them msg

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

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

292 (createMessage msg)

293 (createMessage msg)

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

306-- Channels --

307-- Channels --

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

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

308

309

309−-- | Create a new typed channel

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

311+--

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

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

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

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

316+--

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

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

311newChan = do

318newChan = do

312 proc <- ask

319 proc <- ask

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

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

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

335sendChan (SendPort cid) msg = do

342sendChan (SendPort cid) msg = do

336 proc <- ask

343 proc <- ask

337− let node = localNodeId (processNode proc)

344+ let node = processNode proc

345+ pid = processId proc

346+ us = localNodeId node

338− destNode = processNodeId (sendPortProcessId cid) in do

347+ them = processNodeId (sendPortProcessId cid) in do

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

339− case destNode == node of

349+ case them == us of

340 True -> sendChanLocal cid msg

350 True -> sendChanLocal cid msg

341 False -> do

351 False -> do

342− liftIO $ sendBinary (processNode proc)

352+ liftIO $ sendBinary node

343− (ProcessIdentifier (processId proc))

353+ (ProcessIdentifier pid)

344 (SendPortIdentifier cid)

354 (SendPortIdentifier cid)

345 NoImplicitReconnect

355 NoImplicitReconnect

346 msg

356 msg

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

484 us = processId proc

494 us = processId proc

485 nid = localNodeId node

495 nid = localNodeId node

486 destNode = (processNodeId them) in do

496 destNode = (processNodeId them) in do

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

487− case destNode == nid of

498+ if destNode == nid

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

499+ then sendCtrlMsg Nothing (LocalSend them msg)

489− False ->

500+ else liftIO $ sendPayload (processNode proc)

490− liftIO $ do

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

501+ (ProcessIdentifier (processId proc))

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

502+ (ProcessIdentifier them)

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

503+ NoImplicitReconnect

494− liftIO $ traceEvent (localEventBus node)

504+ (messageToPayload msg)

495− (MxSent them us msg)

496

505

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

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

498--

507--

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

506 us = processId proc

515 us = processId proc

507 nid = localNodeId node

516 nid = localNodeId node

508 destNode = (processNodeId them) in do

517 destNode = (processNodeId them) in do

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

509− case destNode == nid of

519+ if destNode == nid

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

520+ then sendCtrlMsg Nothing (LocalSend them msg)

511− False -> do

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

521+ else 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)

517

522

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

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

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

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

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

1222

1227

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

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

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

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

1225−nsend label msg =

1230+nsend label msg = do

1231+ proc <- ask

1232+ let msg' = createUnencodedMessage msg

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

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

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

1235+ sendCtrlMsg Nothing (NamedSend label msg')

1227

1236

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

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

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

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

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

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

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

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

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

1237nsendRemote nid label msg = do

1246nsendRemote nid label msg = do

1238− here <- getSelfNode

1247+ proc <- ask

1248+ let us = processId proc

1249+ let node = processNode proc

1250+ if localNodeId node == nid

1239− if here == nid then nsend label msg

1251+ then nsend label msg

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

1253+ liftIO $ traceEvent (localEventBus node)

1254+ (MxSentToName lbl us (wrapMessage msg))

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

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

1241

1256

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

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

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

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

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

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

76--

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

78--

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

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

81+--

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

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

84+--

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

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

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

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

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

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

91+-- one receiver.

92+--

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

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

95+--

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

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

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

99+-- implemented as such.

100+--

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

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

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

104+-- processes or agents.

105+--

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

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

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

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

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

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

112+--

79-- [Management Data API]

113-- [Management Data API]

80--

114--

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

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

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

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

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

145-- > monitorNames2 = getSelfPid >>= mxNotify

179-- > monitorNames2 = getSelfPid >>= mxNotify

146--

180--

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

148−--

149-- [Performance, Stablity and Scalability]

181-- [Performance, Stablity and Scalability]

150--

182--

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

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

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

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

21import Control.Distributed.Process.Internal.Types

22 ( Process

23 , ProcessId

24+ , SendPortId

24 , Message

25 , Message

25 , DiedReason

26 , DiedReason

26 , NodeId

27 , NodeId

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

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

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

60 | MxSentToName String ProcessId Message

61 | MxSentToName String ProcessId Message

61 -- ^ fired whenever a named send occurs

62 -- ^ fired whenever a named send occurs

63+ | MxSentToPort ProcessId SendPortId Message

64+ -- ^ fired whenever a sendChan occurs

62 | MxReceived ProcessId Message

65 | MxReceived ProcessId Message

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

66 -- ^ 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

64 | MxConnected ConnectionId EndPointAddress

69 | MxConnected ConnectionId EndPointAddress

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

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

66 | MxDisconnected ConnectionId EndPointAddress

71 | MxDisconnected ConnectionId EndPointAddress

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

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

492data IncomingTarget =

493 Uninit

494 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

496 | ToNode

497

498data ConnectionState = ConnectionState {

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

547 enqueue queue msg -- 'enqueue' is strict

548 trace node (MxReceived pid msg)

549 go st

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

550+ Just (_, ToChan chId (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 -> atomically $

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

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

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

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

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

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

557 go st

561 go st

558 Just (_, ToNode) -> do

562 Just (_, ToNode) -> do

559 let ctrlMsg = decode . BSL.fromChunks $ payload

563 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

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

581 case mChannel of

585 case mChannel of

582 Just channel ->

586 Just channel ->

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

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

584 Nothing ->

588 Nothing ->

585 invalidRequest cid st $

589 invalidRequest cid st $

586 "incoming attempt to connect to unknown channel of"

590 "incoming attempt to connect to unknown channel of"

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

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

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

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

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

1052 ch <- deRefWeak chan'

1056 ch <- deRefWeak chan'

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

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

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

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

1055− deliverChan (UnencodedMessage _ raw) chan' =

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

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

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

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1057− deliverChan (EncodedMessage _ _) _ =

1062+ deliverChan _ _ (EncodedMessage _ _) _ =

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

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

1059 error "invalid local channel delivery"

1064 error "invalid local channel delivery"

1060

1065

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

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

104nsend label msg = do

105 proc <- ask

106 let us = processId proc

107− let node = localNodeId (processNode proc)

108 let msg' = wrapMessage msg

107 let msg' = wrapMessage msg

109 -- see [note: tracing]

108 -- see [note: tracing]

110 liftIO $ traceEvent (localEventBus (processNode proc))

109 liftIO $ traceEvent (localEventBus (processNode proc))

111 (MxSentToName label us msg')

110 (MxSentToName label us msg')

112 sendCtrlMsg Nothing (NamedSend label msg')

111 sendCtrlMsg Nothing (NamedSend label msg')

113

112

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−

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

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

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

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

132nsendRemote nid label msg = do

115nsendRemote nid label msg = do

133 proc <- ask

116 proc <- ask

134 let us = processId proc

117 let us = processId proc

135 let node = processNode proc

118 let node = processNode proc

136− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

137 then nsend label msg

120 then nsend label msg

138 else

121 else

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

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

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

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

141 liftIO $ traceEvent (localEventBus node)

124 liftIO $ traceEvent (localEventBus node)

142− (MxSentToName label us (wrapMessage msg))

125+ (MxSentToName lbl us (wrapMessage msg))

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

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

144

127

145-- | Send a message

128-- | Send a message

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

176 let there = processNodeId them

159 let there = processNodeId them

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

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

178 let msg' = wrapMessage msg

161 let msg' = wrapMessage msg

162+ -- see [note: tracing]

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

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

180 if localNodeId (processNode proc) == there

164 if localNodeId (processNode proc) == there

181 then sendCtrlMsg Nothing $ LocalSend them msg'

165 then sendCtrlMsg Nothing $ LocalSend them msg'

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

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

183 (createMessage msg)

167 (createMessage msg)

184

168

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+

185-- | Send a message on a typed channel

177-- | Send a message on a typed channel

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

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

187sendChan (SendPort cid) msg = do

179sendChan (SendPort cid) msg = do

188 proc <- ask

180 proc <- ask

181+ let node = processNode proc

182+ pid = processId proc

189− let node = localNodeId (processNode proc)

183+ us = localNodeId node

190− destNode = processNodeId (sendPortProcessId cid) in do

184+ them = processNodeId (sendPortProcessId cid)

191− case destNode == node of

185+ msg' = wrapMessage msg in do

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

187+ if them == us

192− True -> unsafeSendChanLocal cid msg

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

193− False ->

194− liftIO $ sendBinary (processNode proc)

189+ else liftIO $ sendBinary node

195− (ProcessIdentifier (processId proc))

190+ (ProcessIdentifier pid)

196− (SendPortIdentifier cid)

191+ (SendPortIdentifier cid)

197− NoImplicitReconnect

192+ NoImplicitReconnect

198− msg

193+ msg

199 where

194 where

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

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

201− unsafeSendChanLocal spId msg' =

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

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

203

197

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

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

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

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