master..49fe7bd · distributed-process

git.iko.soy/distributed-process

on 49fe7bd help sign in

.travis.yml+6−11

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− - env: ARGS="--stack-yaml stack-ghc-8.2.2.yaml"

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

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

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

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

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

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

12+

13+ - env: ARGS=

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

16cache:

@@ -34,10 +34,5 @@ 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}TestMxInMemory

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

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+6−20

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

6license: BSD3

7license-file: LICENSE

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

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

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

10copyright: Well-Typed LLP

11category: Control, Cloud Haskell

12build-type: Simple

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

31 bytestring >= 0.9 && < 0.12,

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

33 distributed-static,

34− exceptions >= 0.5,

35 HUnit >= 1.2 && < 1.7,

34 HUnit >= 1.2 && < 1.7,

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

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

37 network >= 2.5 && < 2.9,

36 network >= 2.5 && < 2.9,

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

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

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

116 HS-Source-Dirs: tests

115 HS-Source-Dirs: tests

117

116

117+

118−Test-Suite TestMxInMemory

118+Test-Suite TestMx

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 TestMxInMemory

126 network-transport-inmemory >= 0.5,

127 test-framework >= 0.6 && < 0.9

128 Extensions: CPP

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

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

130 HS-Source-Dirs: tests

131

132−Test-Suite TestTracingInMemory

132+Test-Suite TestTracing

133 Type: exitcode-stdio-1.0

134 Main-Is: runInMemory.hs

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

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

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

143+ ghc-options: -Wall -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

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

158 HS-Source-Dirs: tests

144 HS-Source-Dirs: tests

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

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

25 )

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

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

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

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

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

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

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

31−import Control.Distributed.Process hiding

31+import Control.Distributed.Process

32− ( try

33− , catch

34− , finally

35− , mask

36− , onException

37− )

38import Control.Distributed.Process.Internal.Types

32import Control.Distributed.Process.Internal.Types

33+ ( NodeId(nodeAddress)

39− ( LocalNode(localEndPoint)

34+ , LocalNode(localEndPoint)

40 , ProcessExitException(..)

35 , ProcessExitException(..)

41 , nullProcessId

36 , nullProcessId

42 , createUnencodedMessage

37 , createUnencodedMessage

43 )

38 )

44import Control.Distributed.Process.Node

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

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

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

49−import Data.Maybe (isNothing, isJust)

41+

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

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

51import Test.Framework (Test, testGroup)

43import Test.Framework (Test, testGroup)

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

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

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

45+import Control.Rematch hiding (match)

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

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

56newtype Ping = Ping ProcessId

48newtype Ping = Ping ProcessId

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

73forkTry :: IO () -> IO ThreadId

65forkTry :: IO () -> IO ThreadId

74forkTry p = do

66forkTry p = do

75 tid <- myThreadId

67 tid <- myThreadId

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

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

78-- | The ping server from the paper

70-- | The ping server from the paper

79ping :: Process ()

71ping :: Process ()

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

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

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

89expectPing :: MVar Bool -> Process ()

81expectPing :: MVar Bool -> Process ()

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

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

91 where

83 where

92 checkPing (Ping _) = True

84 checkPing (Ping _) = True

85+ checkPing _ = False

94-- | Quick and dirty synchronous version of whereisRemoteAsync

87-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

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

239 pid <- getSelfPid

232 pid <- getSelfPid

240 replicateM_ numPings $ do

233 replicateM_ numPings $ do

241 send pingServer (Pong pid)

234 send pingServer (Pong pid)

242− p <- expectTimeout 3000000

235+ p <- expect

243 case p of

236 case p of

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

237+ Ping _ -> return ()

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

238+ _ -> die "Unexpected message"

246

239

247 putMVar clientDone ()

240 putMVar clientDone ()

248

241

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

559 ss <- liftIO $ mapM readMVar vs

552 ss <- liftIO $ mapM readMVar vs

560 case ss of

553 case ss of

561 [sa, sb, sc] ->

554 [sa, sb, sc] ->

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

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

563 [ -- a, b, c

556 [ -- a, b, c

564 (sa, 'a')

557 (sa, 'a')

565 , (sb, 'b')

558 , (sb, 'b')

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

735testRemoteRegistry TestTransport{..} = do

728testRemoteRegistry TestTransport{..} = do

736 node1 <- newLocalNode testTransport initRemoteTable

729 node1 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

730 node2 <- newLocalNode testTransport initRemoteTable

731+ done <- newEmptyMVar

738

732

739 pingServer <- forkProcess node1 ping

733 pingServer <- forkProcess node1 ping

740 deadProcess <- forkProcess node1 (return ())

734 deadProcess <- forkProcess node1 (return ())

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

749

743

750 pid <- verifyWhereIsRemote nid1 "ping"

744 pid <- verifyWhereIsRemote nid1 "ping"

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

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

752−

753 us <- getSelfPid

746 us <- getSelfPid

754 nsendRemote nid1 "ping" (Pong us)

747 nsendRemote nid1 "ping" (Pong us)

755 receiveWait [

748 receiveWait [

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

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

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

760 -- in owner field.

753 -- in owner field.

761 registerRemoteAsync nid1 "dead" deadProcess

754 registerRemoteAsync nid1 "dead" deadProcess

755+ receiveWait [

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

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

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

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

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

765−

766 registerRemoteAsync nid1 "ping" deadProcess

758 registerRemoteAsync nid1 "ping" deadProcess

767 receiveWait [

759 receiveWait [

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

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

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

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

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

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

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

772−

773 unregisterRemoteAsync nid1 "dead"

763 unregisterRemoteAsync nid1 "dead"

774 receiveWait [

764 receiveWait [

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

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

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

778

771

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

772testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

780testRemoteRegistryRemoteProcess TestTransport{..} = do

773testRemoteRegistryRemoteProcess TestTransport{..} = do

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

946 isSorted _ = True

939 isSorted _ = True

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

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

948 -- the last message.

941 -- the last message.

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

942+ liftIO $ do

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

944+ putMVar usendTestOk ()

950

945

951 forkProcess node2 $ do

946 forkProcess node2 $ do

952 them <- liftIO $ readMVar processA

947 them <- liftIO $ readMVar processA

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

957 usendPrim them i

952 usendPrim them i

958 liftIO (threadDelay 30000)

953 liftIO (threadDelay 30000)

959

954

960− res <- takeMVar usendTestOk

955+ takeMVar usendTestOk

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

962

956

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

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

964-- in between

958-- in between

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

1272

1266

1273 takeMVar done

1267 takeMVar done

1274

1268

1275−testChanLifecycle :: TestTransport -> Assertion

1276−testChanLifecycle TestTransport{..} = let delay = 3000000 in do

1277− result <- newEmptyMVar

1278− tchMV <- newEmptyMVar

1279− localNode <- newLocalNode testTransport initRemoteTable

1280− runProcess localNode $ do

1281−

1282− pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())

1283− liftIO $ putMVar tchMV tCh

1284− expect :: Process ()

1285− let (sp, _) = tCh

1286− sendChan sp ()

1287− expect :: Process ()

1288−

1289− mRefPid <- monitor pid

1290−

1291− cPid <- spawnLocal $ do

1292− (sp', rp) <- liftIO $ takeMVar tchMV

1293− -- say "picked up our test channel"

1294− send pid ()

1295− -- say "told pid to continue"

1296− res <- receiveChanTimeout delay rp

1297− case res of

1298− Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)

1299− Just () -> do mr <- monitor pid

1300− pause 10000

1301− -- say "sending pid a second () will cause it to exit"

1302− send pid ()

1303−

1304− -- say "make sure we see a DOWN notification for pid having stopped"

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

1306−

1307− -- now that pid has died, the send port should be useless...

1308− liftIO $ putMVar tchMV (sp', rp)

1309−

1310− -- let's verify that we do not see the message from our

1311− -- parent process on the channel, once pid has died...

1312− recv <- receiveChanTimeout delay rp

1313− -- say $ "finished waiting for second (), writing result" ++ (show recv)

1314− liftIO $ putMVar result $ isNothing recv

1315−

1316− mRefCPid <- monitor cPid

1317−

1318− receiveWait

1319− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)

1320− (\_ -> return ())

1321− ]

1322−

1323− -- say "seen first pid die..."

1324−

1325− (sendPort, _) <- liftIO $ takeMVar tchMV

1326− sendChan sendPort ()

1327− -- say "sent () after owning pid died"

1328−

1329− -- let cPid know we've written to the channel...

1330− send cPid ()

1331−

1332− receiveWait

1333− [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)

1334− (\_ -> return ())

1335− ]

1336−

1337− -- say "seen both pids die now..."

1338−

1339− -- and wait on the result back in IO land...

1340− testRes <- takeMVar result

1341− -- runProcess localNode $ say "got result..."

1342− assertBool "Expected sending on the channel to fail, but received data!" testRes

1343−

1344−

1345testKillLocal :: TestTransport -> Assertion

1269testKillLocal :: TestTransport -> Assertion

1346testKillLocal TestTransport{..} = do

1270testKillLocal TestTransport{..} = do

1347 localNode <- newLocalNode testTransport initRemoteTable

1271 localNode <- newLocalNode testTransport initRemoteTable

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

1552 takeMVar supervisedDone

1476 takeMVar supervisedDone

1553 takeMVar supervisorDone

1477 takeMVar supervisorDone

1554

1478

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−

1625testUnsafeSend :: TestTransport -> Assertion

1479testUnsafeSend :: TestTransport -> Assertion

1626testUnsafeSend TestTransport{..} = do

1480testUnsafeSend TestTransport{..} = do

1627 serverAddr <- newEmptyMVar

1481 serverAddr <- newEmptyMVar

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

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1668 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1669 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1670 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817− , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1671 , testCase "KillLocal" (testKillLocal testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1672 , testCase "KillRemote" (testKillRemote testtrans)

1820 , testCase "Die" (testDie testtrans)

1673 , testCase "Die" (testDie testtrans)

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

1824 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1677 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1825 , testCase "ExitLocal" (testExitLocal testtrans)

1678 , testCase "ExitLocal" (testExitLocal testtrans)

1826 , testCase "ExitRemote" (testExitRemote testtrans)

1679 , testCase "ExitRemote" (testExitRemote testtrans)

1827− , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)

1828 , testCase "TextCallLocal" (testCallLocal testtrans)

1680 , testCase "TextCallLocal" (testCallLocal testtrans)

1829 -- Unsafe Primitives

1681 -- Unsafe Primitives

1830 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)

1682 , 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+5−16

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

20 -- ping !

21 , Ping(Ping)

22 , ping

23− , pause

24 , shouldBe

23 , shouldBe

25 , shouldMatch

24 , shouldMatch

26 , shouldContain

25 , shouldContain

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

74import Control.Concurrent.MVar

73import Control.Concurrent.MVar

75 ( putMVar

74 ( putMVar

76 )

75 )

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

76+import Control.Distributed.Process

78import Control.Distributed.Process.Node

77import Control.Distributed.Process.Node

79import Control.Distributed.Process.Serializable()

78import Control.Distributed.Process.Serializable()

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

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

82−import Control.Monad (forever, void)

81+import Control.Monad (forever)

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

84import Control.Monad.STM (atomically)

82import Control.Monad.STM (atomically)

85import Control.Rematch hiding (match)

83import Control.Rematch hiding (match)

86−import Control.Rematch.Run

84+import Control.Rematch.Run

87import Data.Binary

85import Data.Binary

88import Data.Typeable (Typeable)

86import Data.Typeable (Typeable)

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

91import Test.HUnit.Base (assertBool)

89import Test.HUnit.Base (assertBool)

93import GHC.Generics

91import GHC.Generics

94−import System.Timeout (timeout)

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

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

97type TestResult a = MVar a

94type TestResult a = MVar a

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

110

107

111instance Binary TestProcessControl where

108instance Binary TestProcessControl where

112

109

113−data Private = Private

114− deriving (Typeable, Generic)

115−instance Binary Private where

116−

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

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

118noop :: Process ()

111noop :: Process ()

119noop = return ()

112noop = return ()

120

113

121−pause :: Int -> Process ()

122−pause delay =

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

124−

125synchronisedAssertion :: Eq a

114synchronisedAssertion :: Eq a

126 => String

115 => String

127 -> LocalNode

116 -> LocalNode

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

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

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

235tryRunProcess node p = do

224tryRunProcess node p = do

236 tid <- liftIO myThreadId

225 tid <- liftIO myThreadId

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

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

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

239

227

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

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

241tryForkProcess node p = do

229tryForkProcess node p = do

242 tid <- liftIO myThreadId

230 tid <- liftIO myThreadId

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

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

232+

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

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

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2−{-# LANGUAGE ParallelListComp #-}

3−

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

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

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

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

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

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

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

7+import Control.Concurrent (threadDelay)

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

8+import Control.Distributed.Process

10− ( ProcessExitException(..)

11− , unsafeCreateUnencodedMessage

12− )

13import Control.Distributed.Process.Node

9import Control.Distributed.Process.Node

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

15− ( send

16− , nsend

17− , nsendRemote

18− , usend

19− , sendChan

20− )

21import Control.Distributed.Process.Management

10import Control.Distributed.Process.Management

22 ( MxEvent(..)

11 ( MxEvent(..)

23 , MxAgentId(..)

12 , MxAgentId(..)

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

32 , mxUpdateLocal

21 , mxUpdateLocal

33 , mxNotify

22 , mxNotify

34 , mxBroadcast

23 , mxBroadcast

24+ , mxGetId

35 )

25 )

36−import Control.Monad (void, unless)

26+import Control.Monad (void)

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

38−import Control.Rematch (equalTo)

39import Data.Binary

27import Data.Binary

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

28+import Data.List (find, sort)

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

29+import Data.Maybe (isJust)

42import Data.Typeable

30import Data.Typeable

43−import GHC.Generics hiding (from)

31+import GHC.Generics

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

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

45import Prelude hiding (catch, log)

33import Prelude hiding (catch, log)

46#endif

34#endif

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

57instance Binary Publish where

45instance 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−

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

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

70testAgentBroadcast result = do

48testAgentBroadcast result = do

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

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

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

81 mxNotify ()

59 mxNotify ()

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

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

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

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

84− stash result =<< receiveChanTimeout 10000000 resultRP

62+ stash result =<< receiveChan resultRP

86 kill publisher "finished"

64 kill publisher "finished"

87 kill consumer "finished"

65 kill consumer "finished"

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

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

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

90testAgentDualInput result = do

68testAgentDualInput result = do

91 (sp, rp) <- newChan

69 (sp, rp) <- newChan

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

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

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

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

94 mxSetLocal . (+i) =<< mxGetLocal

72 mxSetLocal . (+i) =<< mxGetLocal

95 i' <- mxGetLocal

73 i' <- mxGetLocal

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

99 else mxReady)

77 else mxReady)

100 ]

78 ]

101

102− mRef <- monitor s

103−

104 mxNotify (1 :: Int)

80 mxNotify (1 :: Int)

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

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

106 mxNotify (2 :: Int)

82 mxNotify (2 :: Int)

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

108 mxNotify (5 :: Int)

84 mxNotify (5 :: Int)

109

110 stash result =<< receiveChanTimeout 10000000 rp

86 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

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

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

117testAgentPrioritisation result = do

89testAgentPrioritisation result = do

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

123

124 let name = "prioritising-agent"

96 let name = "prioritising-agent"

125 (sp, rp) <- newChan

97 (sp, rp) <- newChan

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

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

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

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

128− mxUpdateLocal (s:)

100+ mxUpdateLocal ((s:))

129 st <- mxGetLocal

101 st <- mxGetLocal

130 case length st of

102 case length st of

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

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

132 mxDeactivate "finished"

104 mxDeactivate "finished"

133 _ -> mxReceive -- go to the mailbox

105 _ -> mxReceive -- go to the mailbox

134 )

106 )

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

139 mxNotify "fourth"

111 mxNotify "fourth"

140 nsend name "fifth"

112 nsend name "fifth"

141

113

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

114+ stash result . sort =<< receiveChan rp

143− awaitExit s

144

115

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

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

146testAgentMailboxHandling result = do

117testAgentMailboxHandling result = do

147 (sp, rp) <- newChan

118 (sp, rp) <- newChan

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

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

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

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

150 ]

121 ]

151

122

152− nsend "mailbox-agent" ()

123+ nsend "listener-agent" ()

153

124

154 stash result =<< receiveChanTimeout 1000000 rp

125 stash result =<< receiveChanTimeout 1000000 rp

155 kill agent "finished"

126 kill agent "finished"

156

127

157testAgentEventHandling :: TestResult Bool -> Process ()

128testAgentEventHandling :: TestResult Bool -> Process ()

158testAgentEventHandling result = do

129testAgentEventHandling result = do

159− us <- getSelfPid

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

161− timer <- spawnLocal $ do

162− pause (1000000 * 60 * 5)

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

164− stash result False

165− kill us "Test Timed Out"

166−

167 let initState = [] :: [MxEvent]

130 let initState = [] :: [MxEvent]

168− (rc, rs) <- newChan

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

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

170 (mxSink $ \ev -> do

132 (mxSink $ \ev -> do

171 st <- mxGetLocal

133 st <- mxGetLocal

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

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

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

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

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

176 _ -> return ()

138 _ -> return ()

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

139+ act >> mxReady),

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

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

179 st <- mxGetLocal

141 st <- mxGetLocal

180 let found =

142 let found =

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

194 mxReady)

156 mxReady)

195 ]

157 ]

196

158

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

159+ _ <- monitor agentPid

198− faff 2000000 rs

199−

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

160+ (sp, rp) <- newChan

201−

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

161+ pid <- spawnLocal $ sendChan sp ()

162+ () <- receiveChan rp

203

163

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

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

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

165+ -- higher probably that the agent has seen

206 monitor pid

166 monitor pid

207−

208− send pid ()

209− rct <- receiveChanTimeout 10000000 rp

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

211−

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

167+ receiveWait [ 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

217

168

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

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

219 mxNotify (MxSpawned pid, replyTo)

170 mxNotify (MxSpawned pid, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

171 mxNotify (MxProcessDied pid DiedNormal, replyTo)

221

172

222− seenAlive <- receiveChanTimeout 10000000 reply

173+ seenAlive <- receiveChan reply

223− seenDead <- receiveChanTimeout 10000000 reply

174+ seenDead <- receiveChan 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"

324

175

325− unreg1 <- receiveChanTimeout delay unRegSink

176+ stash result $ seenAlive && seenDead

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

466

177

467tests :: TestTransport -> IO [Test]

178tests :: TestTransport -> IO [Test]

468tests TestTransport{..} = do

179tests TestTransport{..} = do

469 node1 <- newLocalNode testTransport initRemoteTable

180 node1 <- newLocalNode testTransport initRemoteTable

470− node2 <- newLocalNode testTransport initRemoteTable

471 return [

181 return [

472− testGroup "MxAgents" [

182+ testGroup "Mx Agents" [

473− testCase "EventHandling"

183+ testCase "Event Handling"

474− (delayedAssertion

184+ (delayedAssertion

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

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

476− node1 True testAgentEventHandling)

186+ node1 True testAgentEventHandling)

477− , testCase "InterAgentBroadcast"

187+ , testCase "Inter-Agent Broadcast"

478− (delayedAssertion

188+ (delayedAssertion

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

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

480− node1 (Just ()) testAgentBroadcast)

190+ node1 () testAgentBroadcast)

481− , testCase "AgentMailboxHandling"

191+ , testCase "Agent Mailbox Handling"

482− (delayedAssertion

192+ (delayedAssertion

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

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

484− node1 (Just ()) testAgentMailboxHandling)

194+ node1 (Just ()) testAgentMailboxHandling)

485− , testCase "AgentDualInputHandling"

195+ , testCase "Agent Dual Input Handling"

486− (delayedAssertion

196+ (delayedAssertion

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

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

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

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

489− , testCase "AgentInputPrioritisation"

199+ , testCase "Agent Input Prioritisation"

490− (delayedAssertion

200+ (delayedAssertion

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

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

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

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

493− "third", "fourth",

203+ "third", "fourth",

494− "fifth"]) testAgentPrioritisation)

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

506− ]

205+ ]]

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

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

11import Network.Transport (Transport)

12import Control.Distributed.Process

13+import Control.Distributed.Process.Closure

13import Control.Distributed.Process.Node

14import Control.Distributed.Process.Node

15import Control.Monad

16import Control.Monad

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

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

18−import Test.Framework (Test)

19+import Test.Framework (Test, defaultMain)

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

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

21-- Tests:

22-- Tests:

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

42 -> SendPort String

43 -> SendPort String

43 -> ReceivePort String -> ReceivePort String

44 -> ReceivePort String -> ReceivePort String

44 -> Process ()

45 -> Process ()

45−recTest2 wait sync r1 _ = do

46+recTest2 wait sync r1 r2 = do

46 forever $ do

47 forever $ do

47 receiveChan wait

48 receiveChan wait

48 r <- receiveWait

49 r <- receiveWait

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

55 -> SendPort String

56 -> SendPort String

56 -> ReceivePort String -> ReceivePort String

57 -> ReceivePort String -> ReceivePort String

57 -> Process ()

58 -> Process ()

58−recTest3 wait sync r1 _ = do

59+recTest3 wait sync r1 r2 = do

59 forever $ do

60 forever $ do

60 receiveChan wait

61 receiveChan wait

61 r <- receiveWait

62 r <- receiveWait

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

68 -> SendPort String

69 -> SendPort String

69 -> ReceivePort String -> ReceivePort String

70 -> ReceivePort String -> ReceivePort String

70 -> Process ()

71 -> Process ()

71−recTest4 wait sync r1 _ = do

72+recTest4 wait sync r1 r2 = do

72 forever $ do

73 forever $ do

73 receiveChan wait

74 receiveChan wait

74 r <- receiveWait

75 r <- receiveWait

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

82master = do

83master = do

83 (waits,waitr) <- newChan

84 (waits,waitr) <- newChan

84 (syncs,syncr) <- newChan

85 (syncs,syncr) <- newChan

85− let go expected = do

86+ let go expect = do

86 sendChan waits ()

87 sendChan waits ()

87 r <- receiveChan syncr

88 r <- receiveChan syncr

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

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

89− liftIO $ r @?= expected

90+ liftIO $ r @?= expect

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

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

92 (s1,r1) <- newChan

93 (s1,r1) <- newChan

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

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

8import Network.Transport.InMemory

9import Test.Framework (defaultMainWithArgs)

11+import Control.Concurrent (threadDelay)

11import System.Environment (getArgs)

12import System.Environment (getArgs)

13main :: IO ()

14main :: 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: Tim Watson <watson.timothy@gmail.com>

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

10Stability: experimental

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

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

@@ -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.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4

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

25Category: Control

26extra-source-files: ChangeLog

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

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

76 , unlink

77 , monitor

78 , unmonitor

79− , unmonitorAsync

80 , withMonitor

79 , withMonitor

81 , withMonitor_

80 , withMonitor_

82 -- * Logging

81 -- * Logging

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

254 node = processNode proc

253 node = processNode proc

255 nodeId = localNodeId node

254 nodeId = localNodeId node

256 destNode = (processNodeId them) in do

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

257 liftIO $ traceEvent (localEventBus node)

265 liftIO $ traceEvent (localEventBus node)

258 (MxSent them us (createUnencodedMessage msg))

266 (MxSent them us (createUnencodedMessage msg))

259− if destNode == nodeId

260− then sendLocal them msg

261− else liftIO $ sendMessage (processNode proc)

262− (ProcessIdentifier (processId proc))

263− (ProcessIdentifier them)

264− NoImplicitReconnect

265− msg

266

267

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

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

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

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

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

282--

283--

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

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

284usend them msg = do

285usend them msg = do

285− proc <- ask

286+ here <- getSelfNode

286 let there = processNodeId them

287 let there = processNodeId them

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

288+ if here == there

288− let msg' = wrapMessage msg

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

290− if localNodeId (processNode proc) == there

291 then sendLocal them msg

289 then sendLocal them msg

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

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

293 (createMessage msg)

291 (createMessage msg)

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

307-- Channels --

305-- Channels --

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

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

309

307

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

308+-- | Create a new typed channel

311−--

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

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

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

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

316−--

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

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

318newChan = do

310newChan = do

319 proc <- ask

311 proc <- ask

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

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

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

342sendChan (SendPort cid) msg = do

334sendChan (SendPort cid) msg = do

343 proc <- ask

335 proc <- ask

344− let node = processNode proc

345− pid = processId proc

346− us = localNodeId node

336+ let node = localNodeId (processNode proc)

347− them = processNodeId (sendPortProcessId cid) in do

337+ destNode = processNodeId (sendPortProcessId cid) in do

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

349− case them == us of

338+ case destNode == node of

350 True -> sendChanLocal cid msg

339 True -> sendChanLocal cid msg

351 False -> do

340 False -> do

352− liftIO $ sendBinary node

341+ liftIO $ sendBinary (processNode proc)

353− (ProcessIdentifier pid)

342+ (ProcessIdentifier (processId proc))

354 (SendPortIdentifier cid)

343 (SendPortIdentifier cid)

355 NoImplicitReconnect

344 NoImplicitReconnect

356 msg

345 msg

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

494 us = processId proc

483 us = processId proc

495 nid = localNodeId node

484 nid = localNodeId node

496 destNode = (processNodeId them) in do

485 destNode = (processNodeId them) in do

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

498− if destNode == nid

486+ case destNode == nid of

499− then sendCtrlMsg Nothing (LocalSend them msg)

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

500− else liftIO $ sendPayload (processNode proc)

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

501− (ProcessIdentifier (processId proc))

489+ (ProcessIdentifier (processId proc))

502− (ProcessIdentifier them)

490+ (ProcessIdentifier them)

503− NoImplicitReconnect

491+ NoImplicitReconnect

504− (messageToPayload msg)

492+ (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)

505

497

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

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

507--

499--

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

515 us = processId proc

507 us = processId proc

516 nid = localNodeId node

508 nid = localNodeId node

517 destNode = (processNodeId them) in do

509 destNode = (processNodeId them) in do

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

519− if destNode == nid

510+ case destNode == nid of

520− then sendCtrlMsg Nothing (LocalSend them msg)

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

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

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

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)

522

518

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

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

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

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

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

1227

1223

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

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

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

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

1230−nsend label msg = do

1226+nsend label msg =

1231− proc <- ask

1232− let msg' = createUnencodedMessage msg

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

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

1235− sendCtrlMsg Nothing (NamedSend label msg')

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

1236

1228

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

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

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

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

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

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

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

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

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

1246nsendRemote nid label msg = do

1238nsendRemote nid label msg = do

1247− proc <- ask

1239+ here <- getSelfNode

1248− let us = processId proc

1249− let node = processNode proc

1250− if localNodeId node == nid

1251− then nsend label msg

1240+ if here == nid then nsend label msg

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

1253− liftIO $ traceEvent (localEventBus node)

1254− (MxSentToName lbl us (wrapMessage msg))

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

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

1256

1242

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

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

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

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

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

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

76--

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

78--

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

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

81−--

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

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

84−--

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

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

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

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

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

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

91−-- one receiver.

92−--

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

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

95−--

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

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

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

99−-- implemented as such.

100−--

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

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

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

104−-- processes or agents.

105−--

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

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

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

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

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

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

112−--

113-- [Management Data API]

79-- [Management Data API]

114--

80--

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

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

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

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

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

179-- > monitorNames2 = getSelfPid >>= mxNotify

145-- > monitorNames2 = getSelfPid >>= mxNotify

180--

146--

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

148+--

181-- [Performance, Stablity and Scalability]

149-- [Performance, Stablity and Scalability]

182--

150--

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

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

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

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

21import Control.Distributed.Process.Internal.Types

22 ( Process

23 , ProcessId

24− , SendPortId

25 , Message

24 , Message

26 , DiedReason

25 , DiedReason

27 , NodeId

26 , NodeId

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

31 ( MonadState

30 ( MonadState

32 , StateT

31 , StateT

33 )

32 )

34−import Control.Monad.Fix (MonadFix)

35import Data.Binary

33import Data.Binary

36import Data.Typeable (Typeable)

34import Data.Typeable (Typeable)

37import GHC.Generics

35import GHC.Generics

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

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

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

59 | MxSent ProcessId ProcessId Message

57 | MxSent ProcessId ProcessId Message

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

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

65 | MxReceived ProcessId Message

59 | MxReceived ProcessId Message

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

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

67− | MxReceivedPort SendPortId Message

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

69 | MxConnected ConnectionId EndPointAddress

61 | MxConnected ConnectionId EndPointAddress

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

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

71 | MxDisconnected ConnectionId EndPointAddress

63 | MxDisconnected ConnectionId EndPointAddress

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

120 } deriving ( Functor

112 } deriving ( Functor

121 , Monad

113 , Monad

122 , MonadIO

114 , MonadIO

123− , MonadFix

124 , ST.MonadState (MxAgentState s)

115 , ST.MonadState (MxAgentState s)

125 , Typeable

116 , Typeable

126 , Applicative

117 , Applicative

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

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

28import Data.Map (Map)

29import qualified Data.Map as Map

30 ( empty

31− , delete

32 , toList

31 , toList

33 , fromList

32 , fromList

34− , insert

33+ , filter

35− , lookup

36− , partition

37 , partitionWithKey

34 , partitionWithKey

38 , elems

35 , elems

39 , size

36 , size

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

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

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

187 ( mxAgentController

184 ( mxAgentController

188 )

185 )

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

190− ( MxEvent(..)

191− )

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

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

193 ( remoteTable

187 ( remoteTable

194 )

188 )

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

201 , traceLogFmt

195 , traceLogFmt

202 , enableTrace

196 , enableTrace

203 )

197 )

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

199+ ( MxEvent(..)

200+ )

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

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

205import Control.Distributed.Process.Internal.Messaging

202import Control.Distributed.Process.Internal.Messaging

206 ( sendBinary

203 ( sendBinary

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

213 , match

210 , match

214 , sendChan

211 , sendChan

215 , unwrapMessage

212 , unwrapMessage

216− , monitor

217− , unmonitorAsync

218− , getSelfNode

219 , SayMessage(..)

213 , SayMessage(..)

220 )

214 )

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

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

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

492data IncomingTarget =

486data IncomingTarget =

493 Uninit

487 Uninit

494 | ToProc ProcessId (Weak (CQueue Message))

488 | ToProc ProcessId (Weak (CQueue Message))

495− | ToChan SendPortId TypedChannel

489+ | ToChan TypedChannel

496 | ToNode

490 | ToNode

497

491

498data ConnectionState = ConnectionState {

492data ConnectionState = ConnectionState {

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

547 enqueue queue msg -- 'enqueue' is strict

541 enqueue queue msg -- 'enqueue' is strict

548 trace node (MxReceived pid msg)

542 trace node (MxReceived pid msg)

549 go st

543 go st

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

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

551 mChan <- deRefWeak chan'

545 mChan <- deRefWeak chan'

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

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

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

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

554− forM_ mChan $ \chan -> do

548+ forM_ mChan $ \chan -> atomically $

555− msg' <- atomically $ do

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

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

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

558− writeTQueue chan msg

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

559− return msg

560− trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

561 go st

551 go st

562 Just (_, ToNode) -> do

552 Just (_, ToNode) -> do

563 let ctrlMsg = decode . BSL.fromChunks $ payload

553 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

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

585 case mChannel of

575 case mChannel of

586 Just channel ->

576 Just channel ->

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

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

588 Nothing ->

578 Nothing ->

589 invalidRequest cid st $

579 invalidRequest cid st $

590 "incoming attempt to connect to unknown channel of"

580 "incoming attempt to connect to unknown channel of"

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

913

903

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

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

915

905

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

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

906+ modify' $ registeredHere ^: Map.filter (\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

921

907

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

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

923 mapM (\(pid,nidlist) ->

909 mapM (\(pid,nidlist) ->

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

969 do modify' $ registeredHereFor label ^= mPid

955 do modify' $ registeredHereFor label ^= mPid

970 updateRemote node currentVal mPid

956 updateRemote node currentVal mPid

971 case mPid of

957 case mPid of

972− (Just p) -> do

973− if reregistration

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

975− else return ()

976− liftIO $ trace node (MxRegistered p label)

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

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

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

978 newVal <- gets (^. registeredHereFor label)

960 newVal <- gets (^. registeredHereFor label)

979 ncSendToProcess from $ unsafeCreateUnencodedMessage $

961 ncSendToProcess from $ unsafeCreateUnencodedMessage $

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

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

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

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

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

1056 ch <- deRefWeak chan'

1038 ch <- deRefWeak chan'

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

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

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

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

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

1041+ deliverChan (UnencodedMessage _ raw) chan' =

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

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

1061− trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

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

1043+ deliverChan (EncodedMessage _ _) _ =

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

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

1064 error "invalid local channel delivery"

1045 error "invalid local channel delivery"

1065

1046

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

1089 case mProc of

1070 case mProc of

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

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

1091 from (ProcessInfoNone DiedUnknownId)

1072 from (ProcessInfoNone DiedUnknownId)

1092− Just proc -> do

1073+ Just proc -> do

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

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

1094 gets (^. links)

1075 gets (^. links)

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

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

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

1103 infoNode = (processNodeId pid)

1084 infoNode = (processNodeId pid)

1104 , infoRegisteredNames = reg

1085 , infoRegisteredNames = reg

1105 , infoMessageQueueLength = size

1086 , infoMessageQueueLength = size

1106− , infoMonitors = Set.toList itsMons

1087+ , infoMonitors = Set.toList itsMons

1107− , infoLinks = Set.toList itsLinks

1088+ , infoLinks = Set.toList itsLinks

1108 }

1089 }

1109 where dispatch :: (Serializable a)

1090 where dispatch :: (Serializable a)

1110 => Bool

1091 => Bool

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

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

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

52−-- You have been warned!

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

53--

36--

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

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

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

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

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

72 , sendBinary

55 , sendBinary

73 , sendCtrlMsg

56 , sendCtrlMsg

74 )

57 )

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

58+

76− ( MxEvent(..)

77− )

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

79− ( traceEvent

80− )

81import Control.Distributed.Process.Internal.Types

59import Control.Distributed.Process.Internal.Types

82 ( ProcessId(..)

60 ( ProcessId(..)

83 , NodeId(..)

61 , NodeId(..)

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

101

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

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

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

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

104−nsend label msg = do

82+nsend label msg =

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

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

112

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

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

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

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

115nsendRemote nid label msg = do

87nsendRemote nid label msg = do

116 proc <- ask

88 proc <- ask

117− let us = processId proc

118− let node = processNode proc

119− if localNodeId node == nid

89+ if localNodeId (processNode proc) == nid

120 then nsend label msg

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

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

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

127

128-- | Send a message

93-- | Send a message

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

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

130send them msg = do

95send them msg = do

131 proc <- ask

96 proc <- ask

132 let node = localNodeId (processNode proc)

97 let node = localNodeId (processNode proc)

133− destNode = (processNodeId them)

98+ destNode = (processNodeId them) in do

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

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

140− then sendCtrlMsg Nothing $ LocalSend them msg'

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

141− else liftIO $ sendMessage (processNode proc)

142− (ProcessIdentifier (processId proc))

143− (ProcessIdentifier them)

144− NoImplicitReconnect

145− msg

146

110

147-- | Send a message unreliably.

111-- | Send a message unreliably.

148--

112--

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

157usend them msg = do

121usend them msg = do

158 proc <- ask

122 proc <- ask

159 let there = processNodeId them

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

164 if localNodeId (processNode proc) == there

124 if localNodeId (processNode proc) == there

165− then sendCtrlMsg Nothing $ LocalSend them msg'

125+ then sendCtrlMsg Nothing $

126+ LocalSend them (unsafeCreateUnencodedMessage msg)

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

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

167 (createMessage msg)

128 (createMessage msg)

168

129

169−-- [note: tracing]

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

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

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

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

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

175−--

176−

177-- | Send a message on a typed channel

130-- | Send a message on a typed channel

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

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

179sendChan (SendPort cid) msg = do

132sendChan (SendPort cid) msg = do

180 proc <- ask

133 proc <- ask

181− let node = processNode proc

182− pid = processId proc

183− us = localNodeId node

134+ let node = localNodeId (processNode proc)

184− them = processNodeId (sendPortProcessId cid)

135+ destNode = processNodeId (sendPortProcessId cid) in do

185− msg' = wrapMessage msg in do

136+ case destNode == node of

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

187− if them == us

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

137+ True -> unsafeSendChanLocal cid msg

138+ False -> do

189− else liftIO $ sendBinary node

139+ liftIO $ sendBinary (processNode proc)

190− (ProcessIdentifier pid)

140+ (ProcessIdentifier (processId proc))

191− (SendPortIdentifier cid)

141+ (SendPortIdentifier cid)

192− NoImplicitReconnect

142+ NoImplicitReconnect

193− msg

143+ msg

194 where

144 where

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

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

146+ unsafeSendChanLocal spId msg' =

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

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

197

148

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

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

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

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

stack-ghc-7.10.3.yaml+1−5

@@ -3,18 +3,14 @@ resolver: nightly-2016-03-08

3packages:

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

11extra-deps:

7extra-deps:

12- rank1dynamic-0.4.0

8- rank1dynamic-0.4.0

13- distributed-static-0.3.8

9- distributed-static-0.3.8

14- network-transport-0.5.2

10- network-transport-0.5.2

11+- network-transport-tcp-0.6.0

15- network-transport-inmemory-0.5.2

12- network-transport-inmemory-0.5.2

16- rematch-0.2.0.0

13- rematch-0.2.0.0

17−- exceptions-0.8.2.1 # test dependency

19flags:

15flags:

20 distributed-process-tests:

16 distributed-process-tests:

stack-ghc-8.0.2.yaml+1−1

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

1resolver: lts-9.21

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

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

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

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

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

16- network-transport-0.5.2

17- network-transport-0.5.2

17- network-transport-inmemory-0.5.2

18- network-transport-inmemory-0.5.2

18- rematch-0.2.0.0

19- rematch-0.2.0.0

19−# - exceptions-0.8.2.1 # test dependency

21flags:

22 distributed-process-tests:

stack-ghc-8.2.2.yamldeleted−21

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

1−resolver: lts-11.22

2−

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.yamldeleted−20

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

1−resolver: lts-12.18

2−

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

1+resolver: nightly-2018-11-10

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: d58250e2388b86a1865ac4139373cbdc106e72e1

12+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd

13 extra-dep: true

14- location:

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

16− commit: 0ce97924f15a8c1570b2355151834305c9bd2a28

16+ commit: b7a10ffc07a77f4fe6affbf8abe535b11933a01a

17 extra-dep: true

19extra-deps:

stack.yaml+1−3

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

1−resolver: lts-12.18 # Use GHC 8.4.4

1+resolver: nightly-2017-08-20

2−

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

5packages:

3packages:

6- '.'

4- '.'