a094396...master · distributed-process

git.iko.soy/distributed-process

17 files changed+569−236

since splitsnapshots

unifiedsplit

distributed-process-tests/distributed-process-tests.cabal+4−3

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

66license: BSD3

77license-file: LICENSE

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

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

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

1010copyright: Well-Typed LLP

1111category: Control, Cloud Haskell

1212build-type: Simple

@@ -31,6 +31,7 @@ library

3131 bytestring >= 0.9 && < 0.12,

3232 distributed-process >= 0.6.0 && < 0.8,

3333 distributed-static,

34+ exceptions >= 0.5,

3435 HUnit >= 1.2 && < 1.7,

3536 network-transport >= 0.4.1.0 && < 0.6,

3637 network >= 2.5 && < 2.9,

@@ -125,7 +126,7 @@ Test-Suite TestMxInMemory

125126 network-transport-inmemory >= 0.5,

126127 test-framework >= 0.6 && < 0.9

127128 Extensions: CPP

128− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

129+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

129130 HS-Source-Dirs: tests

130131

131132Test-Suite TestTracingInMemory

@@ -139,7 +140,7 @@ Test-Suite TestTracingInMemory

139140 network-transport-inmemory >= 0.5,

140141 test-framework >= 0.6 && < 0.9

141142 Extensions: CPP

142− ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

143+ ghc-options: -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind

143144 HS-Source-Dirs: tests

144145

145146Test-Suite TestMxInTCP

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+124−42

@@ -25,13 +25,18 @@ import Control.Concurrent.MVar

2525 )

2626import Control.Monad (replicateM_, replicateM, forever, void, unless, join)

2727import Control.Exception (SomeException, throwIO, ErrorCall(..))

28−import qualified Control.Exception as Ex (catch)

29−import Control.Applicative ((<$>), (<*>), pure, (<|>))

28+import Control.Monad.Catch (try, catch, finally, mask, onException)

29+import Control.Applicative ((<|>))

3030import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)

31−import Control.Distributed.Process

31+import Control.Distributed.Process hiding

32+ ( try

33+ , catch

34+ , finally

35+ , mask

36+ , onException

37+ )

3238import Control.Distributed.Process.Internal.Types

33− ( NodeId(nodeAddress)

34− , LocalNode(localEndPoint)

39+ ( LocalNode(localEndPoint)

3540 , ProcessExitException(..)

3641 , nullProcessId

3742 , createUnencodedMessage

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

3944import Control.Distributed.Process.Node

4045import Control.Distributed.Process.Debug

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

42−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)

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

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

44−

49+import Data.Maybe (isNothing, isJust)

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

4651import Test.Framework (Test, testGroup)

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

48−import Control.Rematch hiding (match)

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

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

5055

5156newtype Ping = Ping ProcessId

@@ -68,7 +73,7 @@ expectThat a matcher = case res of

6873forkTry :: IO () -> IO ThreadId

6974forkTry p = do

7075 tid <- myThreadId

71− forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))

76+ forkIO $ catch p (\e -> throwTo tid (e :: SomeException))

7277

7378-- | The ping server from the paper

7479ping :: Process ()

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

8287verifyClient s b = takeMVar b >>= assertBool s

8388

8489expectPing :: MVar Bool -> Process ()

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

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

8691 where

8792 checkPing (Ping _) = True

88− checkPing _ = False

8993

9094-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

235239 pid <- getSelfPid

236240 replicateM_ numPings $ do

237241 send pingServer (Pong pid)

238− p <- expect

242+ p <- expectTimeout 3000000

239243 case p of

240− Ping _ -> return ()

241− _ -> die "Unexpected message"

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

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

242246

243247 putMVar clientDone ()

244248

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

555559 ss <- liftIO $ mapM readMVar vs

556560 case ss of

557561 [sa, sb, sc] ->

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

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

559563 [ -- a, b, c

560564 (sa, 'a')

561565 , (sb, 'b')

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

731735testRemoteRegistry TestTransport{..} = do

732736 node1 <- newLocalNode testTransport initRemoteTable

733737 node2 <- newLocalNode testTransport initRemoteTable

734− done <- newEmptyMVar

735738

736739 pingServer <- forkProcess node1 ping

737740 deadProcess <- forkProcess node1 (return ())

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

746749

747750 pid <- verifyWhereIsRemote nid1 "ping"

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

752+

749753 us <- getSelfPid

750754 nsendRemote nid1 "ping" (Pong us)

751755 receiveWait [

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

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

756760 -- in owner field.

757761 registerRemoteAsync nid1 "dead" deadProcess

758− receiveWait [

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

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

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

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

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

765+

761766 registerRemoteAsync nid1 "ping" deadProcess

762767 receiveWait [

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

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

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

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

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

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

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

772+

766773 unregisterRemoteAsync nid1 "dead"

767774 receiveWait [

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

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

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

771− liftIO $ putMVar done ()

772−

773− takeMVar done

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

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

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

774778

775779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

776780testRemoteRegistryRemoteProcess TestTransport{..} = do

@@ -942,9 +946,7 @@ testUSend usendPrim TestTransport{..} numMessages = do

942946 isSorted _ = True

943947 -- The list can't be null since there are no failures after sending

944948 -- the last message.

945− liftIO $ do

946− assertBool "Unexpected failure after sending last msg" $ isSorted msgs && not (null msgs)

947− putMVar usendTestOk ()

949+ liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)

948950

949951 forkProcess node2 $ do

950952 them <- liftIO $ readMVar processA

@@ -955,7 +957,8 @@ testUSend usendPrim TestTransport{..} numMessages = do

955957 usendPrim them i

956958 liftIO (threadDelay 30000)

957959

958− takeMVar usendTestOk

960+ res <- takeMVar usendTestOk

961+ assertBool "Unexpected failure after sending last msg" res

959962

960963-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server

961964-- in between

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

12691272

12701273 takeMVar done

12711274

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+

12721345testKillLocal :: TestTransport -> Assertion

12731346testKillLocal TestTransport{..} = do

12741347 localNode <- newLocalNode testTransport initRemoteTable

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

14831556testRegistryMonitoring TestTransport{..} = do

14841557 node1 <- newLocalNode testTransport initRemoteTable

14851558 node2 <- newLocalNode testTransport initRemoteTable

1486− waitH <- newEmptyMVar

14871559

14881560 let nid = localNodeId node2

14891561 pid <- forkProcess node1 $ do

1490− getSelfPid >>= runUntilRegistered nid

1491− liftIO $ takeMVar waitH

1562+ self <- getSelfPid

1563+ runUntilRegistered nid self

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

1565+ expect :: Process ()

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

14921567

14931568 runProcess node2 $ do

14941569 register regName pid

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

14951571 res <- whereis regName

1572+ send pid ()

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

14961574 us <- getSelfPid

1497− liftIO $ do

1498− putMVar waitH ()

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

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

1576+

15001577

15011578 -- This delay isn't essential!

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

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

15211598 regHere <- newEmptyMVar

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

15231600 res <- takeMVar regHere

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

1601+ case res of

1602+ Nothing -> return ()

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

15251604

15261605 where

15271606 runUntilRegistered nid us = do

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

15331612

15341613 delayUntilMaybeUnregistered nid p = do

15351614 whereisRemoteAsync nid regName

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

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

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

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

15381617 (const $ return ())

15391618 ]

1540− return ()

1619+ case res of

1620+ Just () -> return ()

1621+ Nothing -> delayUntilMaybeUnregistered nid p

15411622

15421623 regName = "testRegisterRemote"

15431624

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

17331814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

17341815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

17351816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)

17361818 , testCase "KillLocal" (testKillLocal testtrans)

17371819 , testCase "KillRemote" (testKillRemote testtrans)

17381820 , testCase "Die" (testDie testtrans)

distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs+1−1

@@ -522,7 +522,7 @@ testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion

522522testSpawnReconnect testtrans@TestTransport{..} rtable = do

523523 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable

524524 let nid1 = localNodeId node1

525− nid2 = localNodeId node2

525+ -- nid2 = localNodeId node2

526526 done <- newEmptyMVar

527527 iv <- newMVar (0 :: Int)

528528

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

@@ -74,12 +74,13 @@ import Control.Concurrent

7474import Control.Concurrent.MVar

7575 ( putMVar

7676 )

77−import Control.Distributed.Process

77+import Control.Distributed.Process hiding (finally, catch)

7878import Control.Distributed.Process.Node

7979import Control.Distributed.Process.Serializable()

8080

8181import Control.Exception (AsyncException(ThreadKilled), SomeException)

8282import Control.Monad (forever, void)

83+import Control.Monad.Catch (finally, catch)

8384import Control.Monad.STM (atomically)

8485import Control.Rematch hiding (match)

8586import Control.Rematch.Run

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

9091import Test.HUnit.Base (assertBool)

9192

9293import GHC.Generics

94+import System.Timeout (timeout)

9395

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

9597type TestResult a = MVar a

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

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

233235tryRunProcess node p = do

234236 tid <- liftIO myThreadId

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

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

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

236239

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

238241tryForkProcess node p = do

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+227−42

@@ -1,12 +1,23 @@

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2+{-# LANGUAGE ParallelListComp #-}

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

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

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

6−

7−import Control.Concurrent (threadDelay)

8−import Control.Distributed.Process

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

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

10+ ( ProcessExitException(..)

11+ , unsafeCreateUnencodedMessage

12+ )

913import Control.Distributed.Process.Node

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

15+ ( send

16+ , nsend

17+ , nsendRemote

18+ , usend

19+ , sendChan

20+ )

1021import Control.Distributed.Process.Management

1122 ( MxEvent(..)

1223 , MxAgentId(..)

@@ -21,15 +32,15 @@ import Control.Distributed.Process.Management

2132 , mxUpdateLocal

2233 , mxNotify

2334 , mxBroadcast

24− , mxGetId

2535 )

2636import Control.Monad (void, unless)

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

2738import Control.Rematch (equalTo)

2839import Data.Binary

29−import Data.List (find, sort)

30−import Data.Maybe (isJust, isNothing)

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

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

3142import Data.Typeable

32−import GHC.Generics

43+import GHC.Generics hiding (from)

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

3445import Prelude hiding (catch, log)

3546#endif

@@ -53,9 +64,9 @@ awaitExit pid =

5364 (\_ -> return ())

5465 ]

5566 where

56− withMonitorRef pid = bracket (monitor pid) unmonitor

67+ withMonitorRef p = bracket (monitor p) unmonitor

5768

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

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

5970testAgentBroadcast result = do

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

6172

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

7081 mxNotify ()

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

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

73− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

7485

75− kill publisher "finished" >> awaitExit publisher

76− kill consumer "finished" >> awaitExit consumer

86+ kill publisher "finished"

87+ kill consumer "finished"

7788

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

7990testAgentDualInput result = do

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

8899 else mxReady)

89100 ]

90101

102+ mRef <- monitor s

103+

91104 mxNotify (1 :: Int)

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

93106 mxNotify (2 :: Int)

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

95108 mxNotify (5 :: Int)

96109

97110 stash result =<< receiveChanTimeout 10000000 rp

98− awaitExit s

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

99115

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

101117testAgentPrioritisation result = do

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

109125 (sp, rp) <- newChan

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

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

112− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

113129 st <- mxGetLocal

114130 case length st of

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

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

116132 mxDeactivate "finished"

117133 _ -> mxReceive -- go to the mailbox

118134 )

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

123139 mxNotify "fourth"

124140 nsend name "fifth"

125141

126− stash result . sort =<< receiveChan rp

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

127143 awaitExit s

128144

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

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

136152 nsend "mailbox-agent" ()

137153

138154 stash result =<< receiveChanTimeout 1000000 rp

139− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

140156

141157testAgentEventHandling :: TestResult Bool -> Process ()

142158testAgentEventHandling result = do

143− 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"

144166

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

145168 (rc, rs) <- newChan

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

147170 (mxSink $ \ev -> do

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

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

175198 faff 2000000 rs

176199

177− _ <- monitor agentPid

178− (sp, rp) <- newChan

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

179201

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

181203

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

184206 monitor pid

185207

186208 send pid ()

187− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

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

188211

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

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

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

190214

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

192216 faff 2000000 rs

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

195219 mxNotify (MxSpawned pid, replyTo)

196220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

197221

198− seenAlive <- receiveChan reply

199− seenDead <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

223+ seenDead <- receiveChanTimeout 10000000 reply

200224

201− stash result $ seenAlive && seenDead

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

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

203228 where

204229 faff delay port = do

205230 res <- receiveChanTimeout delay port

206− unless (isNothing res) $ do

207− pause delay

208− faff delay port

231+ unless (isNothing res) $ pause delay

209232

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

211234testMxRegEvents result = do

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

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

253276

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

255− awaitExit agent

256278

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

258280testMxRegMon remoteNode result = do

@@ -310,48 +332,211 @@ testMxRegMon remoteNode result = do

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

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

312334

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

335+ kill agent "test-complete"

314336

315337ensure :: Process () -> Process () -> Process ()

316338ensure = flip finally

317339

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

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+

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

356+

357+ op label ()

358+

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

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

375+ us <- getSelfPid

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

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

378+ -- initiate a send

379+ op p1 ()

380+

381+ -- verify the management event

382+ sent <- receiveChanTimeout 5000000 sink

383+ case sent of

384+ Just (MxSent pidTo pidFrom _)

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

386+ _ -> return False

387+

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

389+ -> Maybe LocalNode

390+ -> Process ()

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

392+

393+ us <- getSelfPid

394+ send p1 us

395+

396+ cleared <- receiveChanTimeout 2000000 sink

397+ case cleared of

398+ Just (MxSent pidTo pidFrom _)

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

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

401+

402+ chan <- receiveTimeout 5000000 [ match return ]

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

404+

405+ -- initiate a send

406+ op ch' ()

407+

408+ -- verify the management event

409+ sent <- receiveChanTimeout 5000000 sink

410+ case sent of

411+ Just (MxSentToPort sId spId _)

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

413+ _ -> return False

414+

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

416+

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

418+testMxSend mNode label test = do

419+ us <- getSelfPid

420+ (sp, rp) <- newChan

421+ (chan, sink) <- newChan

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

423+ mxSink $ \ev -> do

424+ case ev of

425+ m@(MxSentToPort _ cid _)

426+ | cid /= sendPortId chan

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

428+ m@(MxSent _ fromPid _)

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

430+ m@(MxSentToName _ fromPid _)

431+ | fromPid == us -> liftMX $ sendChan chan m

432+ m@(MxRegistered _ name)

433+ | name == label -> liftMX $ sendChan chan m

434+ _ -> return ()

435+ mxReady

436+ ]

437+

438+ case mNode of

439+ Nothing -> void $ spawnLocal (proc sp)

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

441+

442+ p1 <- receiveChanTimeout 2000000 rp

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

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

445+

446+ kill agent "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+

453+

454+ where

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

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

466+

318467tests :: TestTransport -> IO [Test]

319468tests TestTransport{..} = do

320469 node1 <- newLocalNode testTransport initRemoteTable

321470 node2 <- newLocalNode testTransport initRemoteTable

322471 return [

323− testGroup "Mx Agents" [

324− testCase "Event Handling"

472+ testGroup "MxAgents" [

473+ testCase "EventHandling"

325474 (delayedAssertion

326475 "expected True, but events where not as expected"

327476 node1 True testAgentEventHandling)

328− , testCase "Inter-Agent Broadcast"

477+ , testCase "InterAgentBroadcast"

329478 (delayedAssertion

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

331− node1 () testAgentBroadcast)

332− , testCase "Agent Mailbox Handling"

480+ node1 (Just ()) testAgentBroadcast)

481+ , testCase "AgentMailboxHandling"

333482 (delayedAssertion

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

335484 node1 (Just ()) testAgentMailboxHandling)

336− , testCase "Agent Dual Input Handling"

485+ , testCase "AgentDualInputHandling"

337486 (delayedAssertion

338487 "expected sum = 15, but the result was Nothing"

339488 node1 (Just 15 :: Maybe Int) testAgentDualInput)

340− , testCase "Agent Input Prioritisation"

489+ , testCase "AgentInputPrioritisation"

341490 (delayedAssertion

342491 "expected [first, second, third, fourth, fifth], but result diverged"

343492 node1 (sort ["first", "second",

344493 "third", "fourth",

345494 "fifth"]) testAgentPrioritisation)

346495 ]

347− , testGroup "Mx Events" [

348− testCase "Name Registration Events"

496+ , testGroup "MxEvents" [

497+ testCase "NameRegistrationEvents"

349498 (delayedAssertion

350499 "expected registration events to map to the correct ProcessId"

351500 node1 () testMxRegEvents)

352− , testCase "Post Death Name UnRegistration Events"

501+ , testCase "PostDeathNameUnRegistrationEvents"

353502 (delayedAssertion

354503 "expected process deaths to result in unregistration events"

355504 node1 () (testMxRegMon node2))

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

356506 ]

357507 ]

508+ where

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

510+ ("NSend", [

511+ ("nsend", testNSend nsend)

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

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

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

515+ ])

516+ , ("Send", [

517+ ("send", testSend send)

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

519+ , ("usend", testSend usend)

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

521+ , ("sendChan", testChan sendChan)

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

523+ ])

524+ , ("Forward", [

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

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

527+ ])

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

539+ ]

540+ | (groupName, groupCases) <- specs

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

542+ ]

distributed-process-tests/src/Control/Distributed/Process/Tests/Receive.hs+7−8

@@ -10,13 +10,12 @@ import Network.Transport.Test (TestTransport(..))

1010

1111import Network.Transport (Transport)

1212import Control.Distributed.Process

13−import Control.Distributed.Process.Closure

1413import Control.Distributed.Process.Node

1514

1615import Control.Monad

1716

1817import Test.HUnit (Assertion, (@?=))

19−import Test.Framework (Test, defaultMain)

18+import Test.Framework (Test)

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

2120

2221-- Tests:

@@ -43,7 +42,7 @@ recTest2 :: ReceivePort ()

4342 -> SendPort String

4443 -> ReceivePort String -> ReceivePort String

4544 -> Process ()

46−recTest2 wait sync r1 r2 = do

45+recTest2 wait sync r1 _ = do

4746 forever $ do

4847 receiveChan wait

4948 r <- receiveWait

@@ -56,7 +55,7 @@ recTest3 :: ReceivePort ()

5655 -> SendPort String

5756 -> ReceivePort String -> ReceivePort String

5857 -> Process ()

59−recTest3 wait sync r1 r2 = do

58+recTest3 wait sync r1 _ = do

6059 forever $ do

6160 receiveChan wait

6261 r <- receiveWait

@@ -69,7 +68,7 @@ recTest4 :: ReceivePort ()

6968 -> SendPort String

7069 -> ReceivePort String -> ReceivePort String

7170 -> Process ()

72−recTest4 wait sync r1 r2 = do

71+recTest4 wait sync r1 _ = do

7372 forever $ do

7473 receiveChan wait

7574 r <- receiveWait

@@ -83,11 +82,11 @@ master :: Process ()

8382master = do

8483 (waits,waitr) <- newChan

8584 (syncs,syncr) <- newChan

86− let go expect = do

85+ let go expected = do

8786 sendChan waits ()

8887 r <- receiveChan syncr

89− liftIO $ print (r,expect, r == expect)

90− liftIO $ r @?= expect

88+ liftIO $ print (r, expected, r == expected)

89+ liftIO $ r @?= expected

9190

9291 liftIO $ putStrLn "---- Test 1 ----"

9392 (s1,r1) <- newChan

distributed-process-tests/tests/runInMemory.hs−1

@@ -8,7 +8,6 @@ import Network.Transport.Test (TestTransport(..))

88import Network.Transport.InMemory

99import Test.Framework (defaultMainWithArgs)

1010

11−import Control.Concurrent (threadDelay)

1211import System.Environment (getArgs)

1312

1413main :: IO ()

distributed-process.cabal+2−2

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

66License-File: LICENSE

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

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

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

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

1010Stability: experimental

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

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

@@ -21,7 +21,7 @@ Description: This is an implementation of Cloud Haskell, as described in

2121

2222 You will probably also want to install a Cloud Haskell backend such

2323 as distributed-process-simplelocalnet.

24−Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1

24+Tested-With: GHC==7.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4

2525Category: Control

2626extra-source-files: ChangeLog

2727

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

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

254254 node = processNode proc

255255 nodeId = localNodeId node

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

266257 liftIO $ traceEvent (localEventBus node)

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

268266

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

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

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

284282--

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

286284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288286 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

290291 then sendLocal them msg

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

292293 (createMessage msg)

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

306307-- Channels --

307308--------------------------------------------------------------------------------

308309

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

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

311318newChan = do

312319 proc <- ask

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

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

335342sendChan (SendPort cid) msg = do

336343 proc <- ask

337− let node = localNodeId (processNode proc)

338− destNode = processNodeId (sendPortProcessId cid) in do

339− case destNode == node of

344+ let node = processNode proc

345+ pid = processId proc

346+ us = localNodeId node

347+ them = processNodeId (sendPortProcessId cid) in do

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

349+ case them == us of

340350 True -> sendChanLocal cid msg

341351 False -> do

342− liftIO $ sendBinary (processNode proc)

343− (ProcessIdentifier (processId proc))

352+ liftIO $ sendBinary node

353+ (ProcessIdentifier pid)

344354 (SendPortIdentifier cid)

345355 NoImplicitReconnect

346356 msg

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

484494 us = processId proc

485495 nid = localNodeId node

486496 destNode = (processNodeId them) in do

487− case destNode == nid of

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

489− False -> liftIO $ sendPayload (processNode proc)

490− (ProcessIdentifier (processId proc))

491− (ProcessIdentifier them)

492− NoImplicitReconnect

493− (messageToPayload msg)

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

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

496− liftIO $ traceEvent (localEventBus node)

497− (MxSent them us msg)

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

498+ if destNode == nid

499+ then sendCtrlMsg Nothing (LocalSend them msg)

500+ else liftIO $ sendPayload (processNode proc)

501+ (ProcessIdentifier (processId proc))

502+ (ProcessIdentifier them)

503+ NoImplicitReconnect

504+ (messageToPayload msg)

498505

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

500507--

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

508515 us = processId proc

509516 nid = localNodeId node

510517 destNode = (processNodeId them) in do

511− case destNode == nid of

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

513− False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)

514− msg

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

516− -- In the remote case, 'sendCtrlMsg' can block in the networking stack.

517− liftIO $ traceEvent (localEventBus node)

518− (MxSent them us msg)

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

519+ if destNode == nid

520+ then sendCtrlMsg Nothing (LocalSend them msg)

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

519522

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

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

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

12241227

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

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

1227−nsend label msg =

1228− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage 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')

1235+ sendCtrlMsg Nothing (NamedSend label msg')

12291236

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

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

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

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

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

12391246nsendRemote nid label msg = do

1240− here <- getSelfNode

1241− if here == nid then nsend label msg

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

1247+ proc <- ask

1248+ let us = processId proc

1249+ let node = processNode proc

1250+ if localNodeId node == nid

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

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

12431256

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

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

7676--

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

7878--

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

79113-- [Management Data API]

80114--

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

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

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

145179-- > monitorNames2 = getSelfPid >>= mxNotify

146180--

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

148−--

149181-- [Performance, Stablity and Scalability]

150182--

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

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

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

2121import Control.Distributed.Process.Internal.Types

2222 ( Process

2323 , ProcessId

24+ , SendPortId

2425 , Message

2526 , DiedReason

2627 , NodeId

@@ -57,8 +58,14 @@ data MxEvent =

5758 -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)

5859 | MxSent ProcessId ProcessId Message

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

61+ | MxSentToName String ProcessId Message

62+ -- ^ fired whenever a named send occurs

63+ | MxSentToPort ProcessId SendPortId Message

64+ -- ^ fired whenever a sendChan occurs

6065 | MxReceived ProcessId Message

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

6269 | MxConnected ConnectionId EndPointAddress

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

6471 | MxDisconnected ConnectionId EndPointAddress

src/Control/Distributed/Process/Node.hs+17−58

@@ -186,15 +186,8 @@ import Control.Distributed.Process.Internal.Types

186186import Control.Distributed.Process.Management.Internal.Agent

187187 ( mxAgentController

188188 )

189−import Control.Distributed.Process.Management

190− ( mxAgent

191− , mxSink

192− , liftMX

193− , mxUpdateLocal

194− , mxGetLocal

195− , mxReady

196− , MxEvent(..)

197− , MxAgentId(..)

189+import Control.Distributed.Process.Management.Internal.Types

190+ ( MxEvent(..)

198191 )

199192import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace

200193 ( remoteTable

@@ -323,44 +316,6 @@ startDefaultTracer node' = do

323316

324317-- TODO: we need a better mechanism for defining and registering services

325318

326−{- note [registry monitoring agent]

327− This agent listens for 'MxRegistered' and 'MxUnRegistered' events and tracks

328− all labels for remote 'ProcessId's that are stored in the registry.

329−

330− When a remote process is registered, the agent starts monitoring it until it

331− is unregistered. We can safely ignore ProcessMonitorNotification's, since the

332− node controller is obligated to issue an MxUnRegistered event for any labels

333− connected to a dead process. Thus, our only responsibility here is to ensure

334− that we set up monitors for locally registered processes, and tear them down

335− once unregistration occurs.

336− -}

337−

338−registryMonitorAgent :: Process ProcessId

339−registryMonitorAgent = do

340− nid <- getSelfNode

341− mxAgent registryMonitorAgentId (Map.empty :: Map String MonitorRef)

342− [ mxSink $ \ev -> do

343− case ev of

344− MxRegistered pid label

345− | processNodeId pid /= nid -> do

346− mref <- liftMX $ monitor pid

347− mxUpdateLocal (Map.insert label mref)

348− MxUnRegistered pid label

349− | processNodeId pid /= nid -> do

350− mrefs <- mxGetLocal

351− forM_ (label `Map.lookup` mrefs) $ \mref -> do

352− liftMX $ unmonitorAsync mref

353− mxUpdateLocal (Map.delete label)

354− _ -> return ()

355− mxReady

356− -- [note: no need to remove async answers from mailbox]

357− -- The framework simply discards any input you don't have a handler for.

358− -- See Management.hs `runAgent` for details.

359− ]

360− where

361− registryMonitorAgentId :: MxAgentId

362− registryMonitorAgentId = MxAgentId "service.registry.monitoring"

363−

364319-- | Start and register the service processes on a node

365320startServiceProcesses :: LocalNode -> IO ()

366321startServiceProcesses node = do

@@ -376,7 +331,6 @@ startServiceProcesses node = do

376331 -- loops during tracing if the user reregisters the "logger" with a custom

377332 -- process which uses 'send' or other primitives which are traced.

378333 register "trace.logger" logger

379− void $ registryMonitorAgent

380334 where

381335 loop = do

382336 receiveWait

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

538492data IncomingTarget =

539493 Uninit

540494 | ToProc ProcessId (Weak (CQueue Message))

541− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

542496 | ToNode

543497

544498data ConnectionState = ConnectionState {

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

593547 enqueue queue msg -- 'enqueue' is strict

594548 trace node (MxReceived pid msg)

595549 go st

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

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

597551 mChan <- deRefWeak chan'

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

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

600− forM_ mChan $ \chan -> atomically $

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

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

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

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

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

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

603561 go st

604562 Just (_, ToNode) -> do

605563 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

627585 case mChannel of

628586 Just channel ->

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

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

630588 Nothing ->

631589 invalidRequest cid st $

632590 "incoming attempt to connect to unknown channel of"

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

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

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

10981056 ch <- deRefWeak chan'

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

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

1101− deliverChan (UnencodedMessage _ raw) chan' =

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

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

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

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

1103− deliverChan (EncodedMessage _ _) _ =

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

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

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

11051064 error "invalid local channel delivery"

11061065

src/Control/Distributed/Process/UnsafePrimitives.hs+82−33

@@ -32,7 +32,24 @@

3232-- the /normal/ strategy).

3333--

3434-- Use of the functions in this module can potentially change the runtime

35−-- behaviour of your application. You have been warned!

35+-- behaviour of your application. In addition, messages passed between Cloud

36+-- Haskell processes are written to a tracing infrastructure on the local node,

37+-- to provide improved introspection and debugging facilities for complex actor

38+-- based systems. This module makes no attempt to force evaluation in these

39+-- cases either, thus evaluation problems in passed data structures could not

40+-- only crash your processes, but could also bring down critical internal

41+-- services on which the node relies to function correctly.

42+--

43+-- If you wish to repudiate such issues, you are advised to consider the use

44+-- of NFSerialisable in the distributed-process-extras package, which type

45+-- class brings NFData into scope along with Serializable, such that we can

46+-- force evaluation. Intended for use with modules such as this one, this

47+-- approach guarantees correct evaluatedness in terms of @NFData@. Please note

48+-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the

49+-- same way as a @Binary@ one with regards evaluation, so it is still possible

50+-- to introduce unexpected behaviour by using /unsafe/ primitives in this way.

51+--

52+-- You have been warned!

3653--

3754-- This module is exported so that you can replace the use of Cloud Haskell's

3855-- /safe/ messaging primitives. If you want to use both variants, then you can

@@ -55,7 +72,12 @@ import Control.Distributed.Process.Internal.Messaging

5572 , sendBinary

5673 , sendCtrlMsg

5774 )

58−

75+import Control.Distributed.Process.Management.Internal.Types

76+ ( MxEvent(..)

77+ )

78+import Control.Distributed.Process.Management.Internal.Trace.Types

79+ ( traceEvent

80+ )

5981import Control.Distributed.Process.Internal.Types

6082 ( ProcessId(..)

6183 , NodeId(..)

@@ -79,34 +101,48 @@ import Control.Monad.Reader (ask)

79101

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

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

82−nsend label msg =

83− sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))

104+nsend label msg = do

105+ proc <- ask

106+ let us = processId proc

107+ let msg' = wrapMessage msg

108+ -- see [note: tracing]

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

110+ (MxSentToName label us msg')

111+ sendCtrlMsg Nothing (NamedSend label msg')

84112

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

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

87115nsendRemote nid label msg = do

88116 proc <- ask

89− if localNodeId (processNode proc) == nid

117+ let us = processId proc

118+ let node = processNode proc

119+ if localNodeId node == nid

90120 then nsend label msg

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

121+ else

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

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

124+ liftIO $ traceEvent (localEventBus node)

125+ (MxSentToName lbl us (wrapMessage msg))

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

92127

93128-- | Send a message

94129send :: Serializable a => ProcessId -> a -> Process ()

95130send them msg = do

96131 proc <- ask

97132 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

99− case destNode == node of

100− True -> unsafeSendLocal them msg

101− False -> liftIO $ sendMessage (processNode proc)

102− (ProcessIdentifier (processId proc))

103− (ProcessIdentifier them)

104− NoImplicitReconnect

105− msg

106− where

107− unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()

108− unsafeSendLocal pid msg' =

109− sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')

133+ destNode = (processNodeId them)

134+ us = (processId proc)

135+ msg' = wrapMessage msg in do

136+ -- see [note: tracing]

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

138+ (MxSent them us msg')

139+ if destNode == node

140+ then sendCtrlMsg Nothing $ LocalSend them msg'

141+ else liftIO $ sendMessage (processNode proc)

142+ (ProcessIdentifier (processId proc))

143+ (ProcessIdentifier them)

144+ NoImplicitReconnect

145+ msg

110146

111147-- | Send a message unreliably.

112148--

@@ -121,30 +157,43 @@ usend :: Serializable a => ProcessId -> a -> Process ()

121157usend them msg = do

122158 proc <- ask

123159 let there = processNodeId them

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

161+ let msg' = wrapMessage msg

162+ -- see [note: tracing]

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

124164 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

126− LocalSend them (unsafeCreateUnencodedMessage msg)

165+ then sendCtrlMsg Nothing $ LocalSend them msg'

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

128167 (createMessage msg)

129168

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+

130177-- | Send a message on a typed channel

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

132179sendChan (SendPort cid) msg = do

133180 proc <- ask

134− let node = localNodeId (processNode proc)

135− destNode = processNodeId (sendPortProcessId cid) in do

136− case destNode == node of

137− True -> unsafeSendChanLocal cid msg

138− False -> do

139− liftIO $ sendBinary (processNode proc)

140− (ProcessIdentifier (processId proc))

141− (SendPortIdentifier cid)

142− NoImplicitReconnect

143− msg

181+ let node = processNode proc

182+ pid = processId proc

183+ us = localNodeId node

184+ them = processNodeId (sendPortProcessId cid)

185+ msg' = wrapMessage msg in do

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

187+ if them == us

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

189+ else liftIO $ sendBinary node

190+ (ProcessIdentifier pid)

191+ (SendPortIdentifier cid)

192+ NoImplicitReconnect

193+ msg

144194 where

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

146− unsafeSendChanLocal spId msg' =

147− sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')

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

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

148197

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

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

stack-ghc-7.10.3.yaml+1

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

1414- network-transport-0.5.2

1515- network-transport-inmemory-0.5.2

1616- rematch-0.2.0.0

17+- exceptions-0.8.2.1 # test dependency

1718

1819flags:

1920 distributed-process-tests:

stack-ghc-8.0.2.yaml+1−1

@@ -1,5 +1,4 @@

11resolver: lts-9.21

2−

32# added -location for network-transport-tcp-0.6.0

43# since c7fd79 broke the client facing API for createTransport :/

@@ -17,6 +16,7 @@ extra-deps:

1716- network-transport-0.5.2

1817- network-transport-inmemory-0.5.2

1918- rematch-0.2.0.0

19+# - exceptions-0.8.2.1 # test dependency

2020

2121flags:

2222 distributed-process-tests:

stack-ghc-8.2.2.yaml+1

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

1414- network-transport-0.5.2

1515- network-transport-inmemory-0.5.2

1616- rematch-0.2.0.0

17+# - exceptions-0.8.2.1 # test dependency

1718

1819flags:

1920 distributed-process-tests:

stack.yaml+2

@@ -1,5 +1,7 @@

11resolver: lts-12.18 # Use GHC 8.4.4

3+# TODO: we can enable nightly-2018-11-11 once PR #319 is merged

35packages:

46- '.'

57- distributed-process-tests/