3083672...master · distributed-process

git.iko.soy/distributed-process

on master help sign in

.travis.yml+11−6

since splitsnapshots

unifiedsplit

@@ -6,11 +6,11 @@ matrix:

6 include:

7 - env: ARGS="--stack-yaml stack-ghc-7.10.3.yaml"

8 addons: {apt: {packages: [libgmp-dev]}}

9−

10− - env: ARGS="--stack-yaml stack-ghc-8.0.2.yaml"

9+ - env: ARGS="--stack-yaml stack-ghc-8.2.2.yaml"

11 addons: {apt: {packages: [libgmp-dev]}}

10 addons: {apt: {packages: [libgmp-dev]}}

12−

11+ - env: ARGS="--stack-yaml stack-ghc-8.4.4.yaml"

13− - env: ARGS=

12+ addons: {apt: {packages: [libgmp-dev]}}

13+ - env: ARGS="--stack-yaml stack-nightly.yaml --resolver nightly"

14 addons: {apt: {packages: [libgmp-dev]}}

16cache:

@@ -34,5 +34,10 @@ script:

34- stack ${ARGS} test $ARG='--plain -t "!Flaky"' ${TEST_PACKAGE}TestCHInTCP

35- stack ${ARGS} test $ARG='--plain -t "!SpawnReconnect"' ${TEST_PACKAGE}TestClosure

36- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestStats

37−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMx

37+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMxInMemory

38+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMxInTCP

38−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracing

39+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracingInMemory

40+

41+notifications:

42+ slack:

43+ secure: LbqbNjK1KiHNiQo/i/Za4vkMD9l9mhmN+PWYFBQkNvGFWffSRayRKFrTqt4znV2p7h4dY1XZFoC7T5RemBVQLq2ppZAUWxkITeu1OUlnmVLDfJLeVYGTR/fn90nP6Y5ITE7T3A07nKmWaRbKpIgFBAOjzgqaM0csscDx3z0WW18=

distributed-process-tests/distributed-process-tests.cabal+20−6

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

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

31 bytestring >= 0.9 && < 0.12,

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

33 distributed-static,

34+ exceptions >= 0.5,

34 HUnit >= 1.2 && < 1.7,

35 HUnit >= 1.2 && < 1.7,

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

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

36 network >= 2.5 && < 2.9,

37 network >= 2.5 && < 2.9,

@@ -114,8 +115,7 @@ Test-Suite TestStats

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

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

115 HS-Source-Dirs: tests

116 HS-Source-Dirs: tests

116

117

117−

118−Test-Suite TestMx

118+Test-Suite TestMxInMemory

119 Type: exitcode-stdio-1.0

120 Main-Is: runInMemory.hs

121 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

@@ -126,10 +126,10 @@ Test-Suite TestMx

126 network-transport-inmemory >= 0.5,

127 test-framework >= 0.6 && < 0.9

128 Extensions: CPP

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

130 HS-Source-Dirs: tests

131

132−Test-Suite TestTracing

132+Test-Suite TestTracingInMemory

133 Type: exitcode-stdio-1.0

134 Main-Is: runInMemory.hs

135 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing

@@ -140,5 +140,19 @@ Test-Suite TestTracing

140 network-transport-inmemory >= 0.5,

141 test-framework >= 0.6 && < 0.9

142 Extensions: CPP

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

144+ HS-Source-Dirs: tests

145+

146+Test-Suite TestMxInTCP

147+ Type: exitcode-stdio-1.0

148+ Main-Is: runInMemory.hs

149+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx

150+ Build-Depends: base >= 4.4 && < 5,

151+ distributed-process-tests,

152+ network >= 2.3 && < 2.9,

153+ network-transport >= 0.4.1.0 && < 0.6,

154+ network-transport-inmemory >= 0.5,

155+ test-framework >= 0.6 && < 0.9

156+ Extensions: CPP

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

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

144 HS-Source-Dirs: tests

158 HS-Source-Dirs: tests

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+179−31

@@ -25,24 +25,32 @@ import Control.Concurrent.MVar

25 )

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

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

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

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

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

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

30import 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+ )

32import Control.Distributed.Process.Internal.Types

38import Control.Distributed.Process.Internal.Types

33− ( NodeId(nodeAddress)

34− , LocalNode(localEndPoint)

39+ ( LocalNode(localEndPoint)

35 , ProcessExitException(..)

40 , ProcessExitException(..)

36 , nullProcessId

41 , nullProcessId

37 , createUnencodedMessage

42 , createUnencodedMessage

38 )

43 )

39import Control.Distributed.Process.Node

44import Control.Distributed.Process.Node

45+import Control.Distributed.Process.Debug

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

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

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

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

41−

49+import Data.Maybe (isNothing, isJust)

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

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

43import Test.Framework (Test, testGroup)

51import Test.Framework (Test, testGroup)

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

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

45−import Control.Rematch hiding (match)

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

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

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

48newtype Ping = Ping ProcessId

56newtype Ping = Ping ProcessId

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

65forkTry :: IO () -> IO ThreadId

73forkTry :: IO () -> IO ThreadId

66forkTry p = do

74forkTry p = do

67 tid <- myThreadId

75 tid <- myThreadId

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

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

70-- | The ping server from the paper

78-- | The ping server from the paper

71ping :: Process ()

79ping :: Process ()

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

79verifyClient s b = takeMVar b >>= assertBool s

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

81expectPing :: MVar Bool -> Process ()

89expectPing :: MVar Bool -> Process ()

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

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

83 where

91 where

84 checkPing (Ping _) = True

92 checkPing (Ping _) = True

85− checkPing _ = False

87-- | Quick and dirty synchronous version of whereisRemoteAsync

94-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

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

232 pid <- getSelfPid

239 pid <- getSelfPid

233 replicateM_ numPings $ do

240 replicateM_ numPings $ do

234 send pingServer (Pong pid)

241 send pingServer (Pong pid)

235− p <- expect

242+ p <- expectTimeout 3000000

236 case p of

243 case p of

237− Ping _ -> return ()

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

238− _ -> die "Unexpected message"

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

239

246

240 putMVar clientDone ()

247 putMVar clientDone ()

241

248

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

552 ss <- liftIO $ mapM readMVar vs

559 ss <- liftIO $ mapM readMVar vs

553 case ss of

560 case ss of

554 [sa, sb, sc] ->

561 [sa, sb, sc] ->

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

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

556 [ -- a, b, c

563 [ -- a, b, c

557 (sa, 'a')

564 (sa, 'a')

558 , (sb, 'b')

565 , (sb, 'b')

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

728testRemoteRegistry TestTransport{..} = do

735testRemoteRegistry TestTransport{..} = do

729 node1 <- newLocalNode testTransport initRemoteTable

736 node1 <- newLocalNode testTransport initRemoteTable

730 node2 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

731− done <- newEmptyMVar

732

738

733 pingServer <- forkProcess node1 ping

739 pingServer <- forkProcess node1 ping

734 deadProcess <- forkProcess node1 (return ())

740 deadProcess <- forkProcess node1 (return ())

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

743

749

744 pid <- verifyWhereIsRemote nid1 "ping"

750 pid <- verifyWhereIsRemote nid1 "ping"

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

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

752+

746 us <- getSelfPid

753 us <- getSelfPid

747 nsendRemote nid1 "ping" (Pong us)

754 nsendRemote nid1 "ping" (Pong us)

748 receiveWait [

755 receiveWait [

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

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

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

753 -- in owner field.

760 -- in owner field.

754 registerRemoteAsync nid1 "dead" deadProcess

761 registerRemoteAsync nid1 "dead" deadProcess

755− receiveWait [

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

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

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

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

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

765+

758 registerRemoteAsync nid1 "ping" deadProcess

766 registerRemoteAsync nid1 "ping" deadProcess

759 receiveWait [

767 receiveWait [

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

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

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

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

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

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

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

772+

763 unregisterRemoteAsync nid1 "dead"

773 unregisterRemoteAsync nid1 "dead"

764 receiveWait [

774 receiveWait [

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

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

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

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

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

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

768− liftIO $ putMVar done ()

769−

770− takeMVar done

771

778

772testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

773testRemoteRegistryRemoteProcess TestTransport{..} = do

780testRemoteRegistryRemoteProcess TestTransport{..} = do

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

939 isSorted _ = True

946 isSorted _ = True

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

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

941 -- the last message.

948 -- the last message.

942− liftIO $ do

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

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

944− putMVar usendTestOk ()

945

950

946 forkProcess node2 $ do

951 forkProcess node2 $ do

947 them <- liftIO $ readMVar processA

952 them <- liftIO $ readMVar processA

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

952 usendPrim them i

957 usendPrim them i

953 liftIO (threadDelay 30000)

958 liftIO (threadDelay 30000)

954

959

955− takeMVar usendTestOk

960+ res <- takeMVar usendTestOk

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

956

962

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

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

958-- in between

964-- in between

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

1266

1272

1267 takeMVar done

1273 takeMVar done

1268

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+

1269testKillLocal :: TestTransport -> Assertion

1345testKillLocal :: TestTransport -> Assertion

1270testKillLocal TestTransport{..} = do

1346testKillLocal TestTransport{..} = do

1271 localNode <- newLocalNode testTransport initRemoteTable

1347 localNode <- newLocalNode testTransport initRemoteTable

@@ -1476,6 +1552,76 @@ testExitRemote TestTransport{..} = do

1476 takeMVar supervisedDone

1552 takeMVar supervisedDone

1477 takeMVar supervisorDone

1553 takeMVar supervisorDone

1478

1554

1555+testRegistryMonitoring :: TestTransport -> Assertion

1556+testRegistryMonitoring TestTransport{..} = do

1557+ node1 <- newLocalNode testTransport initRemoteTable

1558+ node2 <- newLocalNode testTransport initRemoteTable

1559+

1560+ let nid = localNodeId node2

1561+ pid <- forkProcess node1 $ do

1562+ self <- getSelfPid

1563+ runUntilRegistered nid self

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

1565+ expect :: Process ()

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

1567+

1568+ runProcess node2 $ do

1569+ register regName pid

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

1571+ res <- whereis regName

1572+ send pid ()

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

1574+ us <- getSelfPid

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

1576+

1577+

1578+ -- This delay isn't essential!

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

1580+ -- and see), however waiting a few seconds here, makes it much more likely

1581+ -- that in delayUntilMaybeUnregistered we will hit the match case right

1582+ -- away, and thus not be subjected to a 20 second delay. The value of 4

1583+ -- seconds appears to work optimally on osx and across several linux distros

1584+ -- running in virtual machines (which is essentially what we do in CI)

1585+ void $ receiveTimeout 4000000 [ matchAny return ]

1586+

1587+ -- This delay doesn't serve much purpose in the happy path, however if some

1588+ -- future patch breaks the cooperative behaviour of node controllers viz

1589+ -- remote process registration and notification taking place via ncEffectDied,

1590+ -- there would be the possibility of a race in the test case should we attempt

1591+ -- to evaluate `whereis regName` on node2 right away. In case the name is still

1592+ -- erroneously registered, observing the 20 second delay (or lack of), could at

1593+ -- least give a hint that something is wrong, and we give up our time slice

1594+ -- so that there's a higher change the registrations have been cleaned up

1595+ -- in either case.

1596+ runProcess node2 $ delayUntilMaybeUnregistered nid pid

1597+

1598+ regHere <- newEmptyMVar

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

1600+ res <- takeMVar regHere

1601+ case res of

1602+ Nothing -> return ()

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

1604+

1605+ where

1606+ runUntilRegistered nid us = do

1607+ whereisRemoteAsync nid regName

1608+ receiveWait [

1609+ matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)

1610+ (const $ return ())

1611+ ]

1612+

1613+ delayUntilMaybeUnregistered nid p = do

1614+ whereisRemoteAsync nid regName

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

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

1617+ (const $ return ())

1618+ ]

1619+ case res of

1620+ Just () -> return ()

1621+ Nothing -> delayUntilMaybeUnregistered nid p

1622+

1623+ regName = "testRegisterRemote"

1624+

1479testUnsafeSend :: TestTransport -> Assertion

1625testUnsafeSend :: TestTransport -> Assertion

1480testUnsafeSend TestTransport{..} = do

1626testUnsafeSend TestTransport{..} = do

1481 serverAddr <- newEmptyMVar

1627 serverAddr <- newEmptyMVar

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

1668 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1669 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1670 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1671 , testCase "KillLocal" (testKillLocal testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1672 , testCase "KillRemote" (testKillRemote testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1673 , testCase "Die" (testDie testtrans)

1820 , testCase "Die" (testDie testtrans)

@@ -1677,6 +1824,7 @@ tests testtrans = return [

1677 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1824 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1678 , testCase "ExitLocal" (testExitLocal testtrans)

1825 , testCase "ExitLocal" (testExitLocal testtrans)

1679 , testCase "ExitRemote" (testExitRemote testtrans)

1826 , testCase "ExitRemote" (testExitRemote testtrans)

1827+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)

1680 , testCase "TextCallLocal" (testCallLocal testtrans)

1828 , testCase "TextCallLocal" (testCallLocal testtrans)

1681 -- Unsafe Primitives

1829 -- Unsafe Primitives

1682 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

1830 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

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

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

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

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

524 let nid1 = localNodeId node1

525− nid2 = localNodeId node2

525+ -- nid2 = localNodeId node2

526 done <- newEmptyMVar

527 iv <- newMVar (0 :: Int)

528

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

@@ -20,6 +20,7 @@ module Control.Distributed.Process.Tests.Internal.Utils

20 -- ping !

21 , Ping(Ping)

22 , ping

23+ , pause

23 , shouldBe

24 , shouldBe

24 , shouldMatch

25 , shouldMatch

25 , shouldContain

26 , shouldContain

@@ -73,15 +74,16 @@ import Control.Concurrent

73import Control.Concurrent.MVar

74import Control.Concurrent.MVar

74 ( putMVar

75 ( putMVar

75 )

76 )

76−import Control.Distributed.Process

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

77import Control.Distributed.Process.Node

78import Control.Distributed.Process.Node

78import Control.Distributed.Process.Serializable()

79import Control.Distributed.Process.Serializable()

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

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

81−import Control.Monad (forever)

82+import Control.Monad (forever, void)

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

82import Control.Monad.STM (atomically)

84import Control.Monad.STM (atomically)

83import Control.Rematch hiding (match)

85import Control.Rematch hiding (match)

84−import Control.Rematch.Run

86+import Control.Rematch.Run

85import Data.Binary

87import Data.Binary

86import Data.Typeable (Typeable)

88import Data.Typeable (Typeable)

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

89import Test.HUnit.Base (assertBool)

91import Test.HUnit.Base (assertBool)

91import GHC.Generics

93import GHC.Generics

94+import System.Timeout (timeout)

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

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

94type TestResult a = MVar a

97type TestResult a = MVar a

@@ -107,10 +110,18 @@ data TestProcessControl = Stop | Go | Report ProcessId

107

110

108instance Binary TestProcessControl where

111instance Binary TestProcessControl where

109

112

113+data Private = Private

114+ deriving (Typeable, Generic)

115+instance Binary Private where

116+

110-- | Does exactly what it says on the tin, doing so in the @Process@ monad.

117-- | Does exactly what it says on the tin, doing so in the @Process@ monad.

111noop :: Process ()

118noop :: Process ()

112noop = return ()

119noop = return ()

113

120

121+pause :: Int -> Process ()

122+pause delay =

123+ void $ receiveTimeout delay [ match (\Private -> return ()) ]

124+

114synchronisedAssertion :: Eq a

125synchronisedAssertion :: Eq a

115 => String

126 => String

116 -> LocalNode

127 -> LocalNode

@@ -223,10 +234,10 @@ testProcessReport pid = do

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

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

224tryRunProcess node p = do

235tryRunProcess node p = do

225 tid <- liftIO myThreadId

236 tid <- liftIO myThreadId

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

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

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

227

239

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

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

229tryForkProcess node p = do

241tryForkProcess node p = do

230 tid <- liftIO myThreadId

242 tid <- liftIO myThreadId

231 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

243 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))

232−

distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+390−53

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

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−

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

7−import Control.Concurrent (threadDelay)

8−import Control.Distributed.Process

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

10+ ( ProcessExitException(..)

11+ , unsafeCreateUnencodedMessage

12+ )

9import Control.Distributed.Process.Node

13import Control.Distributed.Process.Node

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

15+ ( send

16+ , nsend

17+ , nsendRemote

18+ , usend

19+ , sendChan

20+ )

10import Control.Distributed.Process.Management

21import Control.Distributed.Process.Management

11 ( MxEvent(..)

22 ( MxEvent(..)

12 , MxAgentId(..)

23 , MxAgentId(..)

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

21 , mxUpdateLocal

32 , mxUpdateLocal

22 , mxNotify

33 , mxNotify

23 , mxBroadcast

34 , mxBroadcast

24− , mxGetId

25 )

35 )

26−import Control.Monad (void)

36+import Control.Monad (void, unless)

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

38+import Control.Rematch (equalTo)

27import Data.Binary

39import Data.Binary

28−import Data.List (find, sort)

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

29−import Data.Maybe (isJust)

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

30import Data.Typeable

42import Data.Typeable

31−import GHC.Generics

43+import GHC.Generics hiding (from)

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

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

33import Prelude hiding (catch, log)

45import Prelude hiding (catch, log)

34#endif

46#endif

@@ -44,7 +56,17 @@ data Publish = Publish

45instance Binary Publish where

57instance Binary Publish where

59+awaitExit :: ProcessId -> Process ()

60+awaitExit pid =

61+ withMonitorRef pid $ \ref -> do

62+ receiveWait

63+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)

64+ (\_ -> return ())

65+ ]

66+ where

67+ withMonitorRef p = bracket (monitor p) unmonitor

68+

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

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

48testAgentBroadcast result = do

70testAgentBroadcast result = do

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

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

@@ -59,7 +81,7 @@ testAgentBroadcast result = do

59 mxNotify ()

81 mxNotify ()

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

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

62− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

64 kill publisher "finished"

86 kill publisher "finished"

65 kill consumer "finished"

87 kill consumer "finished"

@@ -67,7 +89,7 @@ testAgentBroadcast result = do

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

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

68testAgentDualInput result = do

90testAgentDualInput result = do

69 (sp, rp) <- newChan

91 (sp, rp) <- newChan

70− _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [

92+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [

71 mxSink $ (\(i :: Int) -> do

93 mxSink $ (\(i :: Int) -> do

72 mxSetLocal . (+i) =<< mxGetLocal

94 mxSetLocal . (+i) =<< mxGetLocal

73 i' <- mxGetLocal

95 i' <- mxGetLocal

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

77 else mxReady)

99 else mxReady)

78 ]

100 ]

101

102+ mRef <- monitor s

103+

80 mxNotify (1 :: Int)

104 mxNotify (1 :: Int)

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

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

82 mxNotify (2 :: Int)

106 mxNotify (2 :: Int)

@@ -84,6 +108,10 @@ testAgentDualInput result = do

84 mxNotify (5 :: Int)

108 mxNotify (5 :: Int)

109

86 stash result =<< receiveChanTimeout 10000000 rp

110 stash result =<< receiveChanTimeout 10000000 rp

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

115

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

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

89testAgentPrioritisation result = do

117testAgentPrioritisation result = do

@@ -95,12 +123,12 @@ testAgentPrioritisation result = do

123

96 let name = "prioritising-agent"

124 let name = "prioritising-agent"

97 (sp, rp) <- newChan

125 (sp, rp) <- newChan

98− void $ mxAgent (MxAgentId name) ["first"] [

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

99− mxSink (\(s :: String) -> do

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

100− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

101 st <- mxGetLocal

129 st <- mxGetLocal

102 case length st of

130 case length st of

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

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

104 mxDeactivate "finished"

132 mxDeactivate "finished"

105 _ -> mxReceive -- go to the mailbox

133 _ -> mxReceive -- go to the mailbox

106 )

134 )

@@ -111,23 +139,33 @@ testAgentPrioritisation result = do

111 mxNotify "fourth"

139 mxNotify "fourth"

112 nsend name "fifth"

140 nsend name "fifth"

113

141

114− stash result . sort =<< receiveChan rp

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

143+ awaitExit s

115

144

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

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

117testAgentMailboxHandling result = do

146testAgentMailboxHandling result = do

118 (sp, rp) <- newChan

147 (sp, rp) <- newChan

119− agent <- mxAgent (MxAgentId "listener-agent") () [

148+ agent <- mxAgent (MxAgentId "mailbox-agent") () [

120 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady

149 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady

121 ]

150 ]

122

151

123− nsend "listener-agent" ()

152+ nsend "mailbox-agent" ()

124

153

125 stash result =<< receiveChanTimeout 1000000 rp

154 stash result =<< receiveChanTimeout 1000000 rp

126 kill agent "finished"

155 kill agent "finished"

127

156

128testAgentEventHandling :: TestResult Bool -> Process ()

157testAgentEventHandling :: TestResult Bool -> Process ()

129testAgentEventHandling result = do

158testAgentEventHandling result = do

159+ us <- getSelfPid

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

161+ timer <- spawnLocal $ do

162+ pause (1000000 * 60 * 5)

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

164+ stash result False

165+ kill us "Test Timed Out"

166+

130 let initState = [] :: [MxEvent]

167 let initState = [] :: [MxEvent]

168+ (rc, rs) <- newChan

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

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

132 (mxSink $ \ev -> do

170 (mxSink $ \ev -> do

133 st <- mxGetLocal

171 st <- mxGetLocal

@@ -136,7 +174,7 @@ testAgentEventHandling result = do

136 (MxSpawned _) -> mxSetLocal (ev:st)

174 (MxSpawned _) -> mxSetLocal (ev:st)

137 (MxProcessDied _ _) -> mxSetLocal (ev:st)

175 (MxProcessDied _ _) -> mxSetLocal (ev:st)

138 _ -> return ()

176 _ -> return ()

139− act >> mxReady),

177+ act >> (liftMX $ sendChan rc ()) >> mxReady),

140 (mxSink $ \(ev, sp :: SendPort Bool) -> do

178 (mxSink $ \(ev, sp :: SendPort Bool) -> do

141 st <- mxGetLocal

179 st <- mxGetLocal

142 let found =

180 let found =

@@ -156,50 +194,349 @@ testAgentEventHandling result = do

156 mxReady)

194 mxReady)

157 ]

195 ]

158

196

159− _ <- monitor agentPid

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

198+ faff 2000000 rs

199+

160− (sp, rp) <- newChan

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

201+

161− pid <- spawnLocal $ sendChan sp ()

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

162− () <- receiveChan rp

163

203

164 -- By waiting for a monitor notification, we have a

204 -- By waiting for a monitor notification, we have a

165− -- higher probably that the agent has seen

205+ -- higher probably that the agent has seen the spawn and died events

166 monitor pid

206 monitor pid

207+

208+ send pid ()

209+ rct <- receiveChanTimeout 10000000 rp

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

211+

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

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

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

214+

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

216+ faff 2000000 rs

168

217

169 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

218 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)

170 mxNotify (MxSpawned pid, replyTo)

219 mxNotify (MxSpawned pid, replyTo)

171 mxNotify (MxProcessDied pid DiedNormal, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

172

221

173− seenAlive <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

174− seenDead <- receiveChan reply

223+ seenDead <- receiveChanTimeout 10000000 reply

224+

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

228+ where

229+ faff delay port = do

230+ res <- receiveChanTimeout delay port

231+ unless (isNothing res) $ pause delay

232+

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

234+testMxRegEvents result = do

235+ {- This test only deals with the local case, to ensure that we are being

236+ notified in the expected order - the remote cases related to the

237+ behaviour of the node controller are contained in the CH test suite. -}

238+ ensure (stash result ()) $ do

239+ let label = "testMxRegEvents"

240+ let agentLabel = "mxRegEvents-agent"

241+ let delay = 1000000

242+ (regChan, regSink) <- newChan

243+ (unRegChan, unRegSink) <- newChan

244+ agent <- mxAgent (MxAgentId agentLabel) () [

245+ mxSink $ \ev -> do

246+ case ev of

247+ MxRegistered pid label'

248+ | label' == label -> liftMX $ sendChan regChan (label', pid)

249+ MxUnRegistered pid label'

250+ | label' == label -> liftMX $ sendChan unRegChan (label', pid)

251+ _ -> return ()

252+ mxReady

253+ ]

254+

255+ p1 <- spawnLocal expect

256+ p2 <- spawnLocal expect

257+

258+ register label p1

259+ reg1 <- receiveChanTimeout delay regSink

260+ reg1 `shouldBe` equalTo (Just (label, p1))

261+

262+ unregister label

263+ unreg1 <- receiveChanTimeout delay unRegSink

264+ unreg1 `shouldBe` equalTo (Just (label, p1))

265+

266+ register label p2

267+ reg2 <- receiveChanTimeout delay regSink

268+ reg2 `shouldBe` equalTo (Just (label, p2))

269+

270+ reregister label p1

271+ unreg2 <- receiveChanTimeout delay unRegSink

272+ unreg2 `shouldBe` equalTo (Just (label, p2))

273+

274+ reg3 <- receiveChanTimeout delay regSink

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

276+

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

278+

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

280+testMxRegMon remoteNode result = do

281+ ensure (stash result ()) $ do

282+ -- ensure that when a registered process dies, we get a notification that

283+ -- it has been unregistered as well as seeing the name get removed

284+ let label1 = "aaaaa"

285+ let label2 = "bbbbb"

286+ let isValid l = l == label1 || l == label2

287+ let agentLabel = "mxRegMon-agent"

288+ let delay = 1000000

289+ (regChan, regSink) <- newChan

290+ (unRegChan, unRegSink) <- newChan

291+ agent <- mxAgent (MxAgentId agentLabel) () [

292+ mxSink $ \ev -> do

293+ case ev of

294+ MxRegistered pid label

295+ | isValid label -> liftMX $ sendChan regChan (label, pid)

296+ MxUnRegistered pid label

297+ | isValid label -> liftMX $ sendChan unRegChan (label, pid)

298+ _ -> return ()

299+ mxReady

300+ ]

301+

302+ (sp, rp) <- newChan

303+ liftIO $ forkProcess remoteNode $ do

304+ getSelfPid >>= sendChan sp

305+ expect :: Process ()

306+

307+ p1 <- receiveChan rp

308+

309+ register label1 p1

310+ reg1 <- receiveChanTimeout delay regSink

311+ reg1 `shouldBe` equalTo (Just (label1, p1))

312+

313+ register label2 p1

314+ reg2 <- receiveChanTimeout delay regSink

315+ reg2 `shouldBe` equalTo (Just (label2, p1))

316+

317+ n1 <- whereis label1

318+ n1 `shouldBe` equalTo (Just p1)

319+

320+ n2 <- whereis label2

321+ n2 `shouldBe` equalTo (Just p1)

322+

323+ kill p1 "goodbye"

175

324

176− stash result $ seenAlive && seenDead

325+ unreg1 <- receiveChanTimeout delay unRegSink

326+ unreg2 <- receiveChanTimeout delay unRegSink

327+

328+ let evts = [unreg1, unreg2]

329+ -- we can't rely on the order of the values in the node controller's

330+ -- map (it's either racy to do so, or no such guarantee exists for Data.Map),

331+ -- so we simply verify that we received the un-registration events we expect

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

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

334+

335+ kill agent "test-complete"

336+

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

338+ensure = flip finally

339+

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

177

466

178tests :: TestTransport -> IO [Test]

467tests :: TestTransport -> IO [Test]

179tests TestTransport{..} = do

468tests TestTransport{..} = do

180 node1 <- newLocalNode testTransport initRemoteTable

469 node1 <- newLocalNode testTransport initRemoteTable

470+ node2 <- newLocalNode testTransport initRemoteTable

181 return [

471 return [

182− testGroup "Mx Agents" [

472+ testGroup "MxAgents" [

183− testCase "Event Handling"

473+ testCase "EventHandling"

184− (delayedAssertion

474+ (delayedAssertion

185− "expected True, but events where not as expected"

475+ "expected True, but events where not as expected"

186− node1 True testAgentEventHandling)

476+ node1 True testAgentEventHandling)

187− , testCase "Inter-Agent Broadcast"

477+ , testCase "InterAgentBroadcast"

188− (delayedAssertion

478+ (delayedAssertion

189− "expected (), but no broadcast was received"

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

190− node1 () testAgentBroadcast)

480+ node1 (Just ()) testAgentBroadcast)

191− , testCase "Agent Mailbox Handling"

481+ , testCase "AgentMailboxHandling"

192− (delayedAssertion

482+ (delayedAssertion

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

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

194− node1 (Just ()) testAgentMailboxHandling)

484+ node1 (Just ()) testAgentMailboxHandling)

195− , testCase "Agent Dual Input Handling"

485+ , testCase "AgentDualInputHandling"

196− (delayedAssertion

486+ (delayedAssertion

197− "expected sum = 15, but the result was Nothing"

487+ "expected sum = 15, but the result was Nothing"

198− node1 (Just 15 :: Maybe Int) testAgentDualInput)

488+ node1 (Just 15 :: Maybe Int) testAgentDualInput)

199− , testCase "Agent Input Prioritisation"

489+ , testCase "AgentInputPrioritisation"

200− (delayedAssertion

490+ (delayedAssertion

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

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

202− node1 (sort ["first", "second",

492+ node1 (sort ["first", "second",

203− "third", "fourth",

493+ "third", "fourth",

204− "fifth"]) testAgentPrioritisation)

494+ "fifth"]) testAgentPrioritisation)

495+ ]

496+ , testGroup "MxEvents" [

497+ testCase "NameRegistrationEvents"

498+ (delayedAssertion

499+ "expected registration events to map to the correct ProcessId"

500+ node1 () testMxRegEvents)

501+ , testCase "PostDeathNameUnRegistrationEvents"

502+ (delayedAssertion

503+ "expected process deaths to result in unregistration events"

504+ node1 () (testMxRegMon node2))

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

205− ]]

506+ ]

507+ ]

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

11import Network.Transport (Transport)

12import Control.Distributed.Process

13−import Control.Distributed.Process.Closure

14import Control.Distributed.Process.Node

13import Control.Distributed.Process.Node

16import Control.Monad

15import Control.Monad

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

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

19−import Test.Framework (Test, defaultMain)

18+import Test.Framework (Test)

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

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

22-- Tests:

21-- Tests:

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

43 -> SendPort String

42 -> SendPort String

44 -> ReceivePort String -> ReceivePort String

43 -> ReceivePort String -> ReceivePort String

45 -> Process ()

44 -> Process ()

46−recTest2 wait sync r1 r2 = do

45+recTest2 wait sync r1 _ = do

47 forever $ do

46 forever $ do

48 receiveChan wait

47 receiveChan wait

49 r <- receiveWait

48 r <- receiveWait

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

56 -> SendPort String

55 -> SendPort String

57 -> ReceivePort String -> ReceivePort String

56 -> ReceivePort String -> ReceivePort String

58 -> Process ()

57 -> Process ()

59−recTest3 wait sync r1 r2 = do

58+recTest3 wait sync r1 _ = do

60 forever $ do

59 forever $ do

61 receiveChan wait

60 receiveChan wait

62 r <- receiveWait

61 r <- receiveWait

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

69 -> SendPort String

68 -> SendPort String

70 -> ReceivePort String -> ReceivePort String

69 -> ReceivePort String -> ReceivePort String

71 -> Process ()

70 -> Process ()

72−recTest4 wait sync r1 r2 = do

71+recTest4 wait sync r1 _ = do

73 forever $ do

72 forever $ do

74 receiveChan wait

73 receiveChan wait

75 r <- receiveWait

74 r <- receiveWait

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

83master = do

82master = do

84 (waits,waitr) <- newChan

83 (waits,waitr) <- newChan

85 (syncs,syncr) <- newChan

84 (syncs,syncr) <- newChan

86− let go expect = do

85+ let go expected = do

87 sendChan waits ()

86 sendChan waits ()

88 r <- receiveChan syncr

87 r <- receiveChan syncr

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

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

90− liftIO $ r @?= expect

89+ liftIO $ r @?= expected

92 liftIO $ putStrLn "---- Test 1 ----"

91 liftIO $ putStrLn "---- Test 1 ----"

93 (s1,r1) <- newChan

92 (s1,r1) <- newChan

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

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

8import Network.Transport.InMemory

9import Test.Framework (defaultMainWithArgs)

11−import Control.Concurrent (threadDelay)

12import System.Environment (getArgs)

11import System.Environment (getArgs)

14main :: IO ()

13main :: IO ()

distributed-process.cabal+2−2

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

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

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

23 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

25Category: Control

26extra-source-files: ChangeLog

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

@@ -76,6 +76,7 @@ module Control.Distributed.Process.Internal.Primitives

76 , unlink

77 , monitor

78 , unmonitor

79+ , unmonitorAsync

79 , withMonitor

80 , withMonitor

80 , withMonitor_

81 , withMonitor_

81 -- * Logging

82 -- * Logging

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

253 node = processNode proc

254 node = processNode proc

254 nodeId = localNodeId node

255 nodeId = localNodeId node

255 destNode = (processNodeId them) in do

256 destNode = (processNodeId them) in do

256− case destNode == nodeId of

257− True -> sendLocal them msg

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

259− (ProcessIdentifier (processId proc))

260− (ProcessIdentifier them)

261− NoImplicitReconnect

262− msg

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

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

265 liftIO $ traceEvent (localEventBus node)

257 liftIO $ traceEvent (localEventBus node)

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

267

266

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

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

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

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

283--

282--

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

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

285usend them msg = do

284usend them msg = do

286− here <- getSelfNode

285+ proc <- ask

287 let there = processNodeId them

286 let there = processNodeId them

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

289 then sendLocal them msg

291 then sendLocal them msg

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

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

291 (createMessage msg)

293 (createMessage msg)

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

305-- Channels --

307-- Channels --

306--------------------------------------------------------------------------------

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

307

309

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

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

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

310newChan = do

318newChan = do

311 proc <- ask

319 proc <- ask

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

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

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

334sendChan (SendPort cid) msg = do

342sendChan (SendPort cid) msg = do

335 proc <- ask

343 proc <- ask

336− let node = localNodeId (processNode proc)

344+ let node = processNode proc

345+ pid = processId proc

346+ us = localNodeId node

337− destNode = processNodeId (sendPortProcessId cid) in do

347+ them = processNodeId (sendPortProcessId cid) in do

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

338− case destNode == node of

349+ case them == us of

339 True -> sendChanLocal cid msg

350 True -> sendChanLocal cid msg

340 False -> do

351 False -> do

341− liftIO $ sendBinary (processNode proc)

352+ liftIO $ sendBinary node

342− (ProcessIdentifier (processId proc))

353+ (ProcessIdentifier pid)

343 (SendPortIdentifier cid)

354 (SendPortIdentifier cid)

344 NoImplicitReconnect

355 NoImplicitReconnect

345 msg

356 msg

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

483 us = processId proc

494 us = processId proc

484 nid = localNodeId node

495 nid = localNodeId node

485 destNode = (processNodeId them) in do

496 destNode = (processNodeId them) in do

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

486− case destNode == nid of

498+ if destNode == nid

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

499+ then sendCtrlMsg Nothing (LocalSend them msg)

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

500+ else liftIO $ sendPayload (processNode proc)

489− (ProcessIdentifier (processId proc))

501+ (ProcessIdentifier (processId proc))

490− (ProcessIdentifier them)

502+ (ProcessIdentifier them)

491− NoImplicitReconnect

503+ NoImplicitReconnect

492− (messageToPayload msg)

504+ (messageToPayload msg)

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

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

495− liftIO $ traceEvent (localEventBus node)

496− (MxSent them us msg)

497

505

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

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

499--

507--

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

507 us = processId proc

515 us = processId proc

508 nid = localNodeId node

516 nid = localNodeId node

509 destNode = (processNodeId them) in do

517 destNode = (processNodeId them) in do

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

510− case destNode == nid of

519+ if destNode == nid

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

520+ then sendCtrlMsg Nothing (LocalSend them msg)

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

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

513− msg

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

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

516− liftIO $ traceEvent (localEventBus node)

517− (MxSent them us msg)

518

522

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

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

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

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

1223

1227

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

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

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

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

1226−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')

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

1235+ sendCtrlMsg Nothing (NamedSend label msg')

1228

1236

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

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

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

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

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

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

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

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

1238nsendRemote nid label msg = do

1246nsendRemote nid label msg = do

1239− here <- getSelfNode

1247+ proc <- ask

1248+ let us = processId proc

1249+ let node = processNode proc

1250+ if localNodeId node == nid

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

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

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

1242

1256

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

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

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

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

@@ -30,6 +31,7 @@ import qualified Control.Monad.State as ST

30 ( MonadState

31 ( MonadState

31 , StateT

32 , StateT

32 )

33 )

34+import Control.Monad.Fix (MonadFix)

33import Data.Binary

35import Data.Binary

34import Data.Typeable (Typeable)

36import Data.Typeable (Typeable)

35import GHC.Generics

37import GHC.Generics

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

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

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

57 | MxSent ProcessId ProcessId Message

59 | MxSent ProcessId ProcessId Message

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

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

59 | MxReceived ProcessId Message

65 | MxReceived ProcessId Message

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

61 | MxConnected ConnectionId EndPointAddress

69 | MxConnected ConnectionId EndPointAddress

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

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

63 | MxDisconnected ConnectionId EndPointAddress

71 | MxDisconnected ConnectionId EndPointAddress

@@ -112,6 +120,7 @@ newtype MxAgent s a =

112 } deriving ( Functor

120 } deriving ( Functor

113 , Monad

121 , Monad

114 , MonadIO

122 , MonadIO

123+ , MonadFix

115 , ST.MonadState (MxAgentState s)

124 , ST.MonadState (MxAgentState s)

116 , Typeable

125 , Typeable

117 , Applicative

126 , Applicative

src/Control/Distributed/Process/Node.hs+38−19

@@ -28,9 +28,12 @@ import Data.Binary (decode)

28import Data.Map (Map)

29import qualified Data.Map as Map

30 ( empty

31+ , delete

31 , toList

32 , toList

32 , fromList

33 , fromList

33− , filter

34+ , insert

35+ , lookup

36+ , partition

34 , partitionWithKey

37 , partitionWithKey

35 , elems

38 , elems

36 , size

39 , size

@@ -183,6 +186,9 @@ import Control.Distributed.Process.Internal.Types

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

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

184 ( mxAgentController

187 ( mxAgentController

185 )

188 )

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

190+ ( MxEvent(..)

191+ )

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

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

187 ( remoteTable

193 ( remoteTable

188 )

194 )

@@ -195,9 +201,6 @@ import Control.Distributed.Process.Management.Internal.Trace.Types

195 , traceLogFmt

201 , traceLogFmt

196 , enableTrace

202 , enableTrace

197 )

203 )

198−import Control.Distributed.Process.Management.Internal.Types

199− ( MxEvent(..)

200− )

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

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

202import Control.Distributed.Process.Internal.Messaging

205import Control.Distributed.Process.Internal.Messaging

203 ( sendBinary

206 ( sendBinary

@@ -210,6 +213,9 @@ import Control.Distributed.Process.Internal.Primitives

210 , match

213 , match

211 , sendChan

214 , sendChan

212 , unwrapMessage

215 , unwrapMessage

216+ , monitor

217+ , unmonitorAsync

218+ , getSelfNode

213 , SayMessage(..)

219 , SayMessage(..)

214 )

220 )

215import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))

221import Control.Distributed.Process.Internal.Types (SendPort, Tracer(..))

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

486data IncomingTarget =

492data IncomingTarget =

487 Uninit

493 Uninit

488 | ToProc ProcessId (Weak (CQueue Message))

494 | ToProc ProcessId (Weak (CQueue Message))

489− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

490 | ToNode

496 | ToNode

491

497

492data ConnectionState = ConnectionState {

498data ConnectionState = ConnectionState {

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

541 enqueue queue msg -- 'enqueue' is strict

547 enqueue queue msg -- 'enqueue' is strict

542 trace node (MxReceived pid msg)

548 trace node (MxReceived pid msg)

543 go st

549 go st

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

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

545 mChan <- deRefWeak chan'

551 mChan <- deRefWeak chan'

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

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

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

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

548− forM_ mChan $ \chan -> atomically $

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

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

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

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

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

551 go st

561 go st

552 Just (_, ToNode) -> do

562 Just (_, ToNode) -> do

553 let ctrlMsg = decode . BSL.fromChunks $ payload

563 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

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

575 case mChannel of

585 case mChannel of

576 Just channel ->

586 Just channel ->

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

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

578 Nothing ->

588 Nothing ->

579 invalidRequest cid st $

589 invalidRequest cid st $

580 "incoming attempt to connect to unknown channel of"

590 "incoming attempt to connect to unknown channel of"

@@ -903,7 +913,11 @@ ncEffectDied ident reason = do

903

913

904 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

914 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')

905

915

916+ -- we now consider all labels for this identifier unregistered

906− modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)

917+ let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid

918+ (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)

919+ mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)

920+ modify' $ registeredHere ^= keepNames

907

921

908 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=

922 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=

909 mapM (\(pid,nidlist) ->

923 mapM (\(pid,nidlist) ->

@@ -955,7 +969,11 @@ ncEffectRegister from label atnode mPid reregistration = do

955 do modify' $ registeredHereFor label ^= mPid

969 do modify' $ registeredHereFor label ^= mPid

956 updateRemote node currentVal mPid

970 updateRemote node currentVal mPid

957 case mPid of

971 case mPid of

972+ (Just p) -> do

973+ if reregistration

974+ then liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

975+ else return ()

958− (Just p) -> liftIO $ trace node (MxRegistered p label)

976+ liftIO $ trace node (MxRegistered p label)

959 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

977 Nothing -> liftIO $ trace node (MxUnRegistered (fromJust currentVal) label)

960 newVal <- gets (^. registeredHereFor label)

978 newVal <- gets (^. registeredHereFor label)

961 ncSendToProcess from $ unsafeCreateUnencodedMessage $

979 ncSendToProcess from $ unsafeCreateUnencodedMessage $

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

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

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

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

1038 ch <- deRefWeak chan'

1056 ch <- deRefWeak chan'

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

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

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

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

1041− deliverChan (UnencodedMessage _ raw) chan' =

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

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

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

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1043− deliverChan (EncodedMessage _ _) _ =

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

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

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

1045 error "invalid local channel delivery"

1064 error "invalid local channel delivery"

1046

1065

@@ -1070,7 +1089,7 @@ ncEffectGetInfo from pid =

1070 case mProc of

1089 case mProc of

1071 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1090 Nothing -> dispatch (isLocal node (ProcessIdentifier from))

1072 from (ProcessInfoNone DiedUnknownId)

1091 from (ProcessInfoNone DiedUnknownId)

1073− Just proc -> do

1092+ Just proc -> do

1074 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>

1093 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>

1075 gets (^. links)

1094 gets (^. links)

1076 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)

1095 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)

@@ -1084,8 +1103,8 @@ ncEffectGetInfo from pid =

1084 infoNode = (processNodeId pid)

1103 infoNode = (processNodeId pid)

1085 , infoRegisteredNames = reg

1104 , infoRegisteredNames = reg

1086 , infoMessageQueueLength = size

1105 , infoMessageQueueLength = size

1087− , infoMonitors = Set.toList itsMons

1106+ , infoMonitors = Set.toList itsMons

1088− , infoLinks = Set.toList itsLinks

1107+ , infoLinks = Set.toList itsLinks

1089 }

1108 }

1090 where dispatch :: (Serializable a)

1109 where dispatch :: (Serializable a)

1091 => Bool

1110 => Bool

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

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

32-- the /normal/ strategy).

33--

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

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

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

52+-- You have been warned!

36--

53--

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

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

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

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

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

55 , sendBinary

72 , sendBinary

56 , sendCtrlMsg

73 , sendCtrlMsg

57 )

74 )

58−

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

76+ ( MxEvent(..)

77+ )

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

79+ ( traceEvent

80+ )

59import Control.Distributed.Process.Internal.Types

81import Control.Distributed.Process.Internal.Types

60 ( ProcessId(..)

82 ( ProcessId(..)

61 , NodeId(..)

83 , NodeId(..)

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

101

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

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

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

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

82−nsend label 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')

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

111+ sendCtrlMsg Nothing (NamedSend label msg')

112

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

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

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

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

87nsendRemote nid label msg = do

115nsendRemote nid label msg = do

88 proc <- ask

116 proc <- ask

117+ let us = processId proc

118+ let node = processNode proc

89− if localNodeId (processNode proc) == nid

119+ if localNodeId node == nid

90 then nsend label msg

120 then nsend label 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))

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

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

127

93-- | Send a message

128-- | Send a message

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

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

95send them msg = do

130send them msg = do

96 proc <- ask

131 proc <- ask

97 let node = localNodeId (processNode proc)

132 let node = localNodeId (processNode proc)

98− destNode = (processNodeId them) in do

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

99− case destNode == node of

139+ if destNode == node

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

140+ then sendCtrlMsg Nothing $ LocalSend them msg'

141+ else liftIO $ sendMessage (processNode proc)

142+ (ProcessIdentifier (processId proc))

143+ (ProcessIdentifier them)

144+ NoImplicitReconnect

145+ msg

110

146

111-- | Send a message unreliably.

147-- | Send a message unreliably.

112--

148--

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

121usend them msg = do

157usend them msg = do

122 proc <- ask

158 proc <- ask

123 let there = processNodeId them

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

124 if localNodeId (processNode proc) == there

164 if localNodeId (processNode proc) == there

125− then sendCtrlMsg Nothing $

165+ then sendCtrlMsg Nothing $ LocalSend them msg'

126− LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

128 (createMessage msg)

167 (createMessage msg)

129

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+

130-- | Send a message on a typed channel

177-- | Send a message on a typed channel

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

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

132sendChan (SendPort cid) msg = do

179sendChan (SendPort cid) msg = do

133 proc <- ask

180 proc <- ask

181+ let node = processNode proc

182+ pid = processId proc

134− let node = localNodeId (processNode proc)

183+ us = localNodeId node

135− destNode = processNodeId (sendPortProcessId cid) in do

184+ them = processNodeId (sendPortProcessId cid)

136− case destNode == node of

185+ msg' = wrapMessage msg in do

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

187+ if them == us

137− True -> unsafeSendChanLocal cid msg

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

138− False -> do

139− liftIO $ sendBinary (processNode proc)

189+ else liftIO $ sendBinary node

140− (ProcessIdentifier (processId proc))

190+ (ProcessIdentifier pid)

141− (SendPortIdentifier cid)

191+ (SendPortIdentifier cid)

142− NoImplicitReconnect

192+ NoImplicitReconnect

143− msg

193+ msg

144 where

194 where

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

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

146− unsafeSendChanLocal spId msg' =

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

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

148

197

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

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

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

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

stack-ghc-7.10.3.yaml+1

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

14- network-transport-0.5.2

15- network-transport-inmemory-0.5.2

16- rematch-0.2.0.0

17+- exceptions-0.8.2.1 # test dependency

18flags:

19flags:

19 distributed-process-tests:

20 distributed-process-tests:

stack-ghc-8.0.2.yaml+1−1

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

1resolver: lts-9.21

2−

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

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

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

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

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

17- network-transport-0.5.2

16- network-transport-0.5.2

18- network-transport-inmemory-0.5.2

17- network-transport-inmemory-0.5.2

19- rematch-0.2.0.0

18- rematch-0.2.0.0

19+# - exceptions-0.8.2.1 # test dependency

21flags:

22 distributed-process-tests:

stack-ghc-8.2.2.yamlnew file+21

@@ -0,0 +1,21 @@

1+resolver: lts-11.22

3+packages:

4+- '.'

5+- distributed-process-tests/

6+- location:

7+ git: https://github.com/haskell-distributed/network-transport-tcp.git

8+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9+ extra-dep: true

10+

11+extra-deps:

12+- rank1dynamic-0.4.0

13+- distributed-static-0.3.8

14+- network-transport-0.5.2

15+- network-transport-inmemory-0.5.2

16+- rematch-0.2.0.0

17+# - exceptions-0.8.2.1 # test dependency

18+

19+flags:

20+ distributed-process-tests:

21+ tcp: true

stack-ghc-8.4.4.yamlnew file+20

@@ -0,0 +1,20 @@

1+resolver: lts-12.18

3+packages:

4+- '.'

5+- distributed-process-tests/

6+- location:

7+ git: https://github.com/haskell-distributed/network-transport-tcp.git

8+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

9+ extra-dep: true

10+

11+extra-deps:

12+- rank1dynamic-0.4.0

13+- distributed-static-0.3.8

14+- network-transport-0.5.2

15+- network-transport-inmemory-0.5.2

16+- rematch-0.2.0.0

17+

18+flags:

19+ distributed-process-tests:

20+ tcp: true

stack-nightly.yaml+3−3

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

1−resolver: nightly-2018-11-10

1+resolver: nightly

3packages:

4- '.'

@@ -9,11 +9,11 @@ packages:

9 extra-dep: true

10- location:

11 git: https://github.com/haskell-distributed/network-transport-tcp.git

12− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

12+ commit: d58250e2388b86a1865ac4139373cbdc106e72e1

13 extra-dep: true

14- location:

15 git: https://github.com/haskell-distributed/network-transport-inmemory.git

16− commit: b7a10ffc07a77f4fe6affbf8abe535b11933a01a

16+ commit: 0ce97924f15a8c1570b2355151834305c9bd2a28

17 extra-dep: true

19extra-deps:

stack.yaml+3−1

@@ -1,4 +1,6 @@

1−resolver: nightly-2017-08-20

1+resolver: lts-12.18 # Use GHC 8.4.4

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

3packages:

5packages:

4- '.'

6- '.'