02494f2..master · distributed-process

git.iko.soy/distributed-process

on master help sign in

.travis.yml+4−5

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:

@@ -37,7 +37,6 @@ script:

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

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

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

40−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracingInTCP

42notifications:

41notifications:

43 slack:

42 slack:

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

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

@@ -120,12 +121,12 @@ Test-Suite TestMxInMemory

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

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

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

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

122 distributed-process-tests,

123 distributed-process-tests,

123− network >= 2.3 && < 2.7,

124+ network >= 2.3 && < 2.9,

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

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

125 network-transport-inmemory >= 0.5,

126 network-transport-inmemory >= 0.5,

126 test-framework >= 0.6 && < 0.9

127 test-framework >= 0.6 && < 0.9

127 Extensions: CPP

128 Extensions: CPP

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

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

129 HS-Source-Dirs: tests

130 HS-Source-Dirs: tests

130

131

131Test-Suite TestTracingInMemory

132Test-Suite TestTracingInMemory

@@ -134,12 +135,12 @@ Test-Suite TestTracingInMemory

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

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

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

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

136 distributed-process-tests,

137 distributed-process-tests,

137− network >= 2.3 && < 2.7,

138+ network >= 2.3 && < 2.9,

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

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

139 network-transport-inmemory >= 0.5,

140 network-transport-inmemory >= 0.5,

140 test-framework >= 0.6 && < 0.9

141 test-framework >= 0.6 && < 0.9

141 Extensions: CPP

142 Extensions: CPP

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

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

143 HS-Source-Dirs: tests

144 HS-Source-Dirs: tests

144

145

145Test-Suite TestMxInTCP

146Test-Suite TestMxInTCP

@@ -155,17 +156,3 @@ Test-Suite TestMxInTCP

155 Extensions: CPP

156 Extensions: CPP

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

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

157 HS-Source-Dirs: tests

158 HS-Source-Dirs: tests

158−

159−Test-Suite TestTracingInTCP

160− Type: exitcode-stdio-1.0

161− Main-Is: runInMemory.hs

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

163− Build-Depends: base >= 4.4 && < 5,

164− distributed-process-tests,

165− network >= 2.3 && < 2.9,

166− network-transport >= 0.4.1.0 && < 0.6,

167− network-transport-inmemory >= 0.5,

168− test-framework >= 0.6 && < 0.9

169− Extensions: CPP

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

171− HS-Source-Dirs: tests

distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+172−138

@@ -25,13 +25,18 @@ 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

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

39import Control.Distributed.Process.Node

44import Control.Distributed.Process.Node

40import Control.Distributed.Process.Debug

45import Control.Distributed.Process.Debug

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

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

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

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

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

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

44−

49+import Data.Maybe (isNothing, isJust)

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

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

46import Test.Framework (Test, testGroup)

51import Test.Framework (Test, testGroup)

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

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

48−import Control.Rematch hiding (match)

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

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

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

51newtype Ping = Ping ProcessId

56newtype Ping = Ping ProcessId

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

68forkTry :: IO () -> IO ThreadId

73forkTry :: IO () -> IO ThreadId

69forkTry p = do

74forkTry p = do

70 tid <- myThreadId

75 tid <- myThreadId

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

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

73-- | The ping server from the paper

78-- | The ping server from the paper

74ping :: Process ()

79ping :: Process ()

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

82verifyClient s b = takeMVar b >>= assertBool s

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

84expectPing :: MVar Bool -> Process ()

89expectPing :: MVar Bool -> Process ()

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

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

86 where

91 where

87 checkPing (Ping _) = True

92 checkPing (Ping _) = True

88− checkPing _ = False

90-- | Quick and dirty synchronous version of whereisRemoteAsync

94-- | Quick and dirty synchronous version of whereisRemoteAsync

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

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

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

235 pid <- getSelfPid

239 pid <- getSelfPid

236 replicateM_ numPings $ do

240 replicateM_ numPings $ do

237 send pingServer (Pong pid)

241 send pingServer (Pong pid)

238− p <- expect

242+ p <- expectTimeout 3000000

239 case p of

243 case p of

240− Ping _ -> return ()

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

241− _ -> die "Unexpected message"

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

242

246

243 putMVar clientDone ()

247 putMVar clientDone ()

244

248

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

555 ss <- liftIO $ mapM readMVar vs

559 ss <- liftIO $ mapM readMVar vs

556 case ss of

560 case ss of

557 [sa, sb, sc] ->

561 [sa, sb, sc] ->

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

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

559 [ -- a, b, c

563 [ -- a, b, c

560 (sa, 'a')

564 (sa, 'a')

561 , (sb, 'b')

565 , (sb, 'b')

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

731testRemoteRegistry TestTransport{..} = do

735testRemoteRegistry TestTransport{..} = do

732 node1 <- newLocalNode testTransport initRemoteTable

736 node1 <- newLocalNode testTransport initRemoteTable

733 node2 <- newLocalNode testTransport initRemoteTable

737 node2 <- newLocalNode testTransport initRemoteTable

734− done <- newEmptyMVar

735

738

736 pingServer <- forkProcess node1 ping

739 pingServer <- forkProcess node1 ping

737 deadProcess <- forkProcess node1 (return ())

740 deadProcess <- forkProcess node1 (return ())

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

746

749

747 pid <- verifyWhereIsRemote nid1 "ping"

750 pid <- verifyWhereIsRemote nid1 "ping"

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

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

752+

749 us <- getSelfPid

753 us <- getSelfPid

750 nsendRemote nid1 "ping" (Pong us)

754 nsendRemote nid1 "ping" (Pong us)

751 receiveWait [

755 receiveWait [

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

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

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

756 -- in owner field.

760 -- in owner field.

757 registerRemoteAsync nid1 "dead" deadProcess

761 registerRemoteAsync nid1 "dead" deadProcess

758− receiveWait [

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

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

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

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

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

765+

761 registerRemoteAsync nid1 "ping" deadProcess

766 registerRemoteAsync nid1 "ping" deadProcess

762 receiveWait [

767 receiveWait [

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

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

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

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

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

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

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

772+

766 unregisterRemoteAsync nid1 "dead"

773 unregisterRemoteAsync nid1 "dead"

767 receiveWait [

774 receiveWait [

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

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

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

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

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

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

771− liftIO $ putMVar done ()

772−

773− takeMVar done

774

778

775testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion

776testRemoteRegistryRemoteProcess TestTransport{..} = do

780testRemoteRegistryRemoteProcess TestTransport{..} = do

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

942 isSorted _ = True

946 isSorted _ = True

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

944 -- the last message.

948 -- the last message.

945− liftIO $ do

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

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

947− putMVar usendTestOk ()

948

950

949 forkProcess node2 $ do

951 forkProcess node2 $ do

950 them <- liftIO $ readMVar processA

952 them <- liftIO $ readMVar processA

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

955 usendPrim them i

957 usendPrim them i

956 liftIO (threadDelay 30000)

958 liftIO (threadDelay 30000)

957

959

958− takeMVar usendTestOk

960+ res <- takeMVar usendTestOk

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

959

962

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

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

961-- in between

964-- in between

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

1269

1272

1270 takeMVar done

1273 takeMVar done

1271

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+

1272testKillLocal :: TestTransport -> Assertion

1345testKillLocal :: TestTransport -> Assertion

1273testKillLocal TestTransport{..} = do

1346testKillLocal TestTransport{..} = do

1274 localNode <- newLocalNode testTransport initRemoteTable

1347 localNode <- newLocalNode testTransport initRemoteTable

@@ -1481,113 +1554,73 @@ testExitRemote TestTransport{..} = do

1481

1554

1482testRegistryMonitoring :: TestTransport -> Assertion

1555testRegistryMonitoring :: TestTransport -> Assertion

1483testRegistryMonitoring TestTransport{..} = do

1556testRegistryMonitoring TestTransport{..} = do

1484− localNode <- newLocalNode testTransport initRemoteTable

1557+ node1 <- newLocalNode testTransport initRemoteTable

1485− remoteNode <- newLocalNode testTransport initRemoteTable

1558+ node2 <- newLocalNode testTransport initRemoteTable

1486− return ()

1487

1559

1488− -- Local process. Test if local process will be removed from

1560+ let nid = localNodeId node2

1489− -- registry when it dies.

1561+ pid <- forkProcess node1 $ do

1490− box <- newEmptyMVar

1562+ self <- getSelfPid

1491− runProcess localNode $ do

1563+ runUntilRegistered nid self

1492− pid <- spawnLocal $ do

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

1493− expect

1565+ expect :: Process ()

1494− register "test" pid

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

1495− tpid <- whereis "test"

1496− if tpid == Just pid

1497− then do _ <- monitor pid

1498− send pid ()

1499− ProcessMonitorNotification{} <- expect

1500− tpid1 <- whereis "test"

1501− liftIO $ putMVar box (Nothing == tpid1)

1502− else liftIO $ putMVar box False

1503−

1504− takeMVar box >>= assertBool "expected local process to not be registered"

1505− return ()

1506−

1507− -- Remote process. Test if remote process entry is removed

1508− -- from registry when process dies.

1509− remote1 <- testRemote remoteNode

1510− runProcess localNode $

1511− let waitpoll = do

1512− w <- whereis "test" :: Process (Maybe ProcessId)

1513− forM_ w (const waitpoll)

1514− in do register "test" remote1

1515− -- [race-condition] invalid test if we force exit before waitpoll runs

1516− registered <- whereis "test"

1517− registered `shouldBe` (equalTo (Just remote1))

1518− send remote1 ()

1519− waitpoll

1520− return ()

1521− return ()

1522−

1523− -- Many labels. Test if all labels associated with process

1524− -- are removed from registry when it dies.

1525− remote2 <- testRemote remoteNode

1526− runProcess localNode $

1527− let waitpoll = do

1528− w1 <- whereis "test-3" :: Process (Maybe ProcessId)

1529− w2 <- whereis "test-4" :: Process (Maybe ProcessId)

1530− forM_ (w1 <|> w2) (const waitpoll)

1531− in do register "test-3" remote2

1532− register "test-4" remote2

1533− send remote2 ()

1534− waitpoll

1535− return ()

1536−

1537−{- XXX: waiting including patch for nsend for remote process

1538− remote3 <- testRemote remoteNode

1539− remote4 <- testRemote remoteNode

1540− -- test many labels

1541− runProcess localNode $ do

1542− register "test-3" remote3

1543− reregister "test-3" remote4

1544− send remote3 ()

1545− liftIO $ threadDelay 50000 -- XXX: racy

1546− monitor remote4

1547− nsend "test-3" ()

1548− ProcessMonitorNotification{} <- expect

1549− return ()

1550−-}

1551−

1552− -- Test registerRemoteAsync properties. Add a local process to

1553− -- remote registry and checks that it is removed

1554− -- when the process dies.

1555− remote5 <- testRemote remoteNode

1556− runProcess localNode $ do

1557− registerRemoteAsync (localNodeId remoteNode) "test" remote5

1558− receiveWait [

1559− match (\(RegisterReply _ True _) -> return ())

1560− ] >>= send remote5

1561− let waitpoll = do

1562− whereisRemoteAsync (localNodeId remoteNode) "test"

1563− receiveWait [

1564− match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))

1565− ]

1566− waitpoll

1567

1568− -- Add remote process to remote registry and checks if

1568+ runProcess node2 $ do

1569− -- entry is removed then process is dead.

1569+ register regName pid

1570− remote6 <- testRemote localNode

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

1571− runProcess localNode $ do

1571+ res <- whereis regName

1572− registerRemoteAsync (localNodeId remoteNode) "test" remote6

1572+ send pid ()

1573− receiveWait [

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

1574− match (\(RegisterReply _ True _) -> return ())

1574+ us <- getSelfPid

1575− ] >>= send remote6

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

1576− let waitpoll = do

1576+

1577− whereisRemoteAsync (localNodeId remoteNode) "test"

1577+

1578− receiveWait [

1578+ -- This delay isn't essential!

1579− match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))

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

1580− ]

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

1581− waitpoll

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

1582− where

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

1583− testRemote node = do

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

1584− -- test many labels

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

1585− pidBox <- newEmptyMVar

1585+ void $ receiveTimeout 4000000 [ matchAny return ]

1586− forkProcess node $ do

1586+

1587− us <- getSelfPid

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

1588− liftIO $ putMVar pidBox us

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

1589− expect :: Process ()

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

1590− takeMVar pidBox

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"

1591

1624

1592testUnsafeSend :: TestTransport -> Assertion

1625testUnsafeSend :: TestTransport -> Assertion

1593testUnsafeSend TestTransport{..} = do

1626testUnsafeSend TestTransport{..} = do

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

1781 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)

1782 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)

1783 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)

1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)

1784 , testCase "KillLocal" (testKillLocal testtrans)

1818 , testCase "KillLocal" (testKillLocal testtrans)

1785 , testCase "KillRemote" (testKillRemote testtrans)

1819 , testCase "KillRemote" (testKillRemote testtrans)

1786 , testCase "Die" (testDie testtrans)

1820 , testCase "Die" (testDie testtrans)

@@ -1790,7 +1824,7 @@ tests testtrans = return [

1790 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1824 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)

1791 , testCase "ExitLocal" (testExitLocal testtrans)

1825 , testCase "ExitLocal" (testExitLocal testtrans)

1792 , testCase "ExitRemote" (testExitRemote testtrans)

1826 , testCase "ExitRemote" (testExitRemote testtrans)

1793− , testCase "TestRegistryMonitor" (testRegistryMonitoring testtrans)

1827+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)

1794 , testCase "TextCallLocal" (testCallLocal testtrans)

1828 , testCase "TextCallLocal" (testCallLocal testtrans)

1795 -- Unsafe Primitives

1829 -- Unsafe Primitives

1796 , 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+5−2

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

74import Control.Concurrent.MVar

75 ( putMVar

76 )

77−import Control.Distributed.Process

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

78import Control.Distributed.Process.Node

79import Control.Distributed.Process.Serializable()

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

82import Control.Monad (forever, void)

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

83import Control.Monad.STM (atomically)

84import Control.Monad.STM (atomically)

84import Control.Rematch hiding (match)

85import Control.Rematch hiding (match)

85import Control.Rematch.Run

86import Control.Rematch.Run

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

90import Test.HUnit.Base (assertBool)

91import Test.HUnit.Base (assertBool)

92import GHC.Generics

93import GHC.Generics

94+import System.Timeout (timeout)

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

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

95type TestResult a = MVar a

97type TestResult a = MVar a

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

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

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

233tryRunProcess node p = do

235tryRunProcess node p = do

234 tid <- liftIO myThreadId

236 tid <- liftIO myThreadId

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

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

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

236

239

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

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

238tryForkProcess node p = do

241tryForkProcess node p = do

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

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

1−{-# LANGUAGE DeriveGeneric #-}

1+{-# LANGUAGE DeriveGeneric #-}

2+{-# LANGUAGE ParallelListComp #-}

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,15 +32,15 @@ import Control.Distributed.Process.Management

21 , mxUpdateLocal

32 , mxUpdateLocal

22 , mxNotify

33 , mxNotify

23 , mxBroadcast

34 , mxBroadcast

24− , mxGetId

25 )

35 )

26import Control.Monad (void, unless)

36import Control.Monad (void, unless)

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

27import Control.Rematch (equalTo)

38import Control.Rematch (equalTo)

28import Data.Binary

39import Data.Binary

29−import Data.List (find, sort)

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

30−import Data.Maybe (isJust, isNothing)

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

31import Data.Typeable

42import Data.Typeable

32−import GHC.Generics

43+import GHC.Generics hiding (from)

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

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

34import Prelude hiding (catch, log)

45import Prelude hiding (catch, log)

35#endif

46#endif

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

53 (\_ -> return ())

64 (\_ -> return ())

54 ]

65 ]

55 where

66 where

56− withMonitorRef pid = bracket (monitor pid) unmonitor

67+ withMonitorRef p = bracket (monitor p) unmonitor

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

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

59testAgentBroadcast result = do

70testAgentBroadcast result = do

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

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

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

70 mxNotify ()

81 mxNotify ()

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

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

73− stash result =<< receiveChan resultRP

84+ stash result =<< receiveChanTimeout 10000000 resultRP

75− kill publisher "finished" >> awaitExit publisher

86+ kill publisher "finished"

76− kill consumer "finished" >> awaitExit consumer

87+ kill consumer "finished"

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

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

79testAgentDualInput result = do

90testAgentDualInput result = do

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

88 else mxReady)

99 else mxReady)

89 ]

100 ]

101

102+ mRef <- monitor s

103+

91 mxNotify (1 :: Int)

104 mxNotify (1 :: Int)

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

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

93 mxNotify (2 :: Int)

106 mxNotify (2 :: Int)

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

95 mxNotify (5 :: Int)

108 mxNotify (5 :: Int)

109

97 stash result =<< receiveChanTimeout 10000000 rp

110 stash result =<< receiveChanTimeout 10000000 rp

98− awaitExit s

111+ died <- receiveTimeout 10000000 [

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

113+ ]

114+ died `shouldBe` equalTo (Just True)

115

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

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

101testAgentPrioritisation result = do

117testAgentPrioritisation result = do

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

109 (sp, rp) <- newChan

125 (sp, rp) <- newChan

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

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

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

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

112− mxUpdateLocal ((s:))

128+ mxUpdateLocal (s:)

113 st <- mxGetLocal

129 st <- mxGetLocal

114 case length st of

130 case length st of

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

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

116 mxDeactivate "finished"

132 mxDeactivate "finished"

117 _ -> mxReceive -- go to the mailbox

133 _ -> mxReceive -- go to the mailbox

118 )

134 )

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

123 mxNotify "fourth"

139 mxNotify "fourth"

124 nsend name "fifth"

140 nsend name "fifth"

125

141

126− stash result . sort =<< receiveChan rp

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

127 awaitExit s

143 awaitExit s

128

144

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

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

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

136 nsend "mailbox-agent" ()

152 nsend "mailbox-agent" ()

137

153

138 stash result =<< receiveChanTimeout 1000000 rp

154 stash result =<< receiveChanTimeout 1000000 rp

139− kill agent "finished" >> awaitExit agent

155+ kill agent "finished"

140

156

141testAgentEventHandling :: TestResult Bool -> Process ()

157testAgentEventHandling :: TestResult Bool -> Process ()

142testAgentEventHandling result = do

158testAgentEventHandling result = do

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

159+ us <- getSelfPid

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

161+ timer <- spawnLocal $ do

162+ pause (1000000 * 60 * 5)

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

164+ stash result False

165+ kill us "Test Timed Out"

144

166

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

145 (rc, rs) <- newChan

168 (rc, rs) <- newChan

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

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

147 (mxSink $ \ev -> do

170 (mxSink $ \ev -> do

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

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

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

175 faff 2000000 rs

198 faff 2000000 rs

176

199

177− _ <- monitor agentPid

178− (sp, rp) <- newChan

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

179

201

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

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

181

203

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

184 monitor pid

206 monitor pid

185

207

186 send pid ()

208 send pid ()

187− () <- receiveChan rp

209+ rct <- receiveChanTimeout 10000000 rp

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

188

211

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

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

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

190

214

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

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

192 faff 2000000 rs

216 faff 2000000 rs

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

195 mxNotify (MxSpawned pid, replyTo)

219 mxNotify (MxSpawned pid, replyTo)

196 mxNotify (MxProcessDied pid DiedNormal, replyTo)

220 mxNotify (MxProcessDied pid DiedNormal, replyTo)

197

221

198− seenAlive <- receiveChan reply

222+ seenAlive <- receiveChanTimeout 10000000 reply

199− seenDead <- receiveChan reply

223+ seenDead <- receiveChanTimeout 10000000 reply

200

224

201− stash result $ seenAlive && seenDead

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

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

226+ kill timer "test-complete"

227+ kill agentPid "test-complete"

203 where

228 where

204 faff delay port = do

229 faff delay port = do

205 res <- receiveChanTimeout delay port

230 res <- receiveChanTimeout delay port

206− unless (isNothing res) $ do

231+ unless (isNothing res) $ pause delay

207− pause delay

208− faff delay port

209

232

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

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

211testMxRegEvents result = do

234testMxRegEvents result = do

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

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

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

253

276

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

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

255− awaitExit agent

256

278

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

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

258testMxRegMon remoteNode result = do

280testMxRegMon remoteNode result = do

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

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

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

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

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

312

334

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

335+ kill agent "test-complete"

314

336

315ensure :: Process () -> Process () -> Process ()

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

316ensure = flip finally

338ensure = flip finally

317

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+

318tests :: TestTransport -> IO [Test]

467tests :: TestTransport -> IO [Test]

319tests TestTransport{..} = do

468tests TestTransport{..} = do

320 node1 <- newLocalNode testTransport initRemoteTable

469 node1 <- newLocalNode testTransport initRemoteTable

321 node2 <- newLocalNode testTransport initRemoteTable

470 node2 <- newLocalNode testTransport initRemoteTable

322 return [

471 return [

323− testGroup "Mx Agents" [

472+ testGroup "MxAgents" [

324− testCase "Event Handling"

473+ testCase "EventHandling"

325 (delayedAssertion

474 (delayedAssertion

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

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

327 node1 True testAgentEventHandling)

476 node1 True testAgentEventHandling)

328− , testCase "Inter-Agent Broadcast"

477+ , testCase "InterAgentBroadcast"

329 (delayedAssertion

478 (delayedAssertion

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

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

331− node1 () testAgentBroadcast)

480+ node1 (Just ()) testAgentBroadcast)

332− , testCase "Agent Mailbox Handling"

481+ , testCase "AgentMailboxHandling"

333 (delayedAssertion

482 (delayedAssertion

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

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

335 node1 (Just ()) testAgentMailboxHandling)

484 node1 (Just ()) testAgentMailboxHandling)

336− , testCase "Agent Dual Input Handling"

485+ , testCase "AgentDualInputHandling"

337 (delayedAssertion

486 (delayedAssertion

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

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

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

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

340− , testCase "Agent Input Prioritisation"

489+ , testCase "AgentInputPrioritisation"

341 (delayedAssertion

490 (delayedAssertion

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

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

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

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

344 "third", "fourth",

493 "third", "fourth",

345 "fifth"]) testAgentPrioritisation)

494 "fifth"]) testAgentPrioritisation)

346 ]

495 ]

347− , testGroup "Mx Events" [

496+ , testGroup "MxEvents" [

348− testCase "Name Registration Events"

497+ testCase "NameRegistrationEvents"

349 (delayedAssertion

498 (delayedAssertion

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

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

351 node1 () testMxRegEvents)

500 node1 () testMxRegEvents)

352− , testCase "Post Death Name UnRegistration Events"

501+ , testCase "PostDeathNameUnRegistrationEvents"

353 (delayedAssertion

502 (delayedAssertion

354 "expected process deaths to result in unregistration events"

503 "expected process deaths to result in unregistration events"

355 node1 () (testMxRegMon node2))

504 node1 () (testMxRegMon node2))

505+ , testGroup "SendEvents" $ buildTestCases node1 node2

356 ]

506 ]

357 ]

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

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

254 node = processNode proc

255 nodeId = localNodeId node

256 destNode = (processNodeId them) in do

257− case destNode == nodeId of

258− True -> sendLocal them msg

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

260− (ProcessIdentifier (processId proc))

261− (ProcessIdentifier them)

262− NoImplicitReconnect

263− msg

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

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

266 liftIO $ traceEvent (localEventBus node)

257 liftIO $ traceEvent (localEventBus node)

267 (MxSent them us (createUnencodedMessage msg))

258 (MxSent them us (createUnencodedMessage msg))

259+ if destNode == nodeId

260+ then sendLocal them msg

261+ else liftIO $ sendMessage (processNode proc)

262+ (ProcessIdentifier (processId proc))

263+ (ProcessIdentifier them)

264+ NoImplicitReconnect

265+ msg

268

266

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

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

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

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

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

284--

282--

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

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

286usend them msg = do

284usend them msg = do

287− here <- getSelfNode

285+ proc <- ask

288 let there = processNodeId them

286 let there = processNodeId them

289− if here == there

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

288+ let msg' = wrapMessage msg

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

290+ if localNodeId (processNode proc) == there

290 then sendLocal them msg

291 then sendLocal them msg

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

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

292 (createMessage msg)

293 (createMessage msg)

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

306-- Channels --

307-- Channels --

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

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

308

309

309−-- | Create a new typed channel

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

311+--

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

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

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

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

316+--

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

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

311newChan = do

318newChan = do

312 proc <- ask

319 proc <- ask

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

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

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

335sendChan (SendPort cid) msg = do

342sendChan (SendPort cid) msg = do

336 proc <- ask

343 proc <- ask

337− let node = localNodeId (processNode proc)

344+ let node = processNode proc

345+ pid = processId proc

346+ us = localNodeId node

338− destNode = processNodeId (sendPortProcessId cid) in do

347+ them = processNodeId (sendPortProcessId cid) in do

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

339− case destNode == node of

349+ case them == us of

340 True -> sendChanLocal cid msg

350 True -> sendChanLocal cid msg

341 False -> do

351 False -> do

342− liftIO $ sendBinary (processNode proc)

352+ liftIO $ sendBinary node

343− (ProcessIdentifier (processId proc))

353+ (ProcessIdentifier pid)

344 (SendPortIdentifier cid)

354 (SendPortIdentifier cid)

345 NoImplicitReconnect

355 NoImplicitReconnect

346 msg

356 msg

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

484 us = processId proc

494 us = processId proc

485 nid = localNodeId node

495 nid = localNodeId node

486 destNode = (processNodeId them) in do

496 destNode = (processNodeId them) in do

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

487− case destNode == nid of

498+ if destNode == nid

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

499+ then sendCtrlMsg Nothing (LocalSend them msg)

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

500+ else liftIO $ sendPayload (processNode proc)

490− (ProcessIdentifier (processId proc))

501+ (ProcessIdentifier (processId proc))

491− (ProcessIdentifier them)

502+ (ProcessIdentifier them)

492− NoImplicitReconnect

503+ NoImplicitReconnect

493− (messageToPayload msg)

504+ (messageToPayload msg)

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

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

496− liftIO $ traceEvent (localEventBus node)

497− (MxSent them us msg)

498

505

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

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

500--

507--

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

508 us = processId proc

515 us = processId proc

509 nid = localNodeId node

516 nid = localNodeId node

510 destNode = (processNodeId them) in do

517 destNode = (processNodeId them) in do

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

511− case destNode == nid of

519+ if destNode == nid

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

520+ then sendCtrlMsg Nothing (LocalSend them msg)

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

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

514− msg

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

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

517− liftIO $ traceEvent (localEventBus node)

518− (MxSent them us msg)

519

522

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

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

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

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

1224

1227

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

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

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

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

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

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

1235+ sendCtrlMsg Nothing (NamedSend label msg')

1229

1236

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

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

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

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

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

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

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

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

1239nsendRemote nid label msg = do

1246nsendRemote nid label msg = do

1240− here <- getSelfNode

1247+ proc <- ask

1248+ let us = processId proc

1249+ let node = processNode proc

1250+ if localNodeId node == nid

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

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

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

1243

1256

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

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

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

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

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

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

58 | MxSent ProcessId ProcessId Message

59 | MxSent ProcessId ProcessId Message

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

60 | MxReceived ProcessId Message

65 | MxReceived ProcessId Message

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

62 | MxConnected ConnectionId EndPointAddress

69 | MxConnected ConnectionId EndPointAddress

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

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

64 | MxDisconnected ConnectionId EndPointAddress

71 | MxDisconnected ConnectionId EndPointAddress

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

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

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

187 ( mxAgentController

188 )

189−import Control.Distributed.Process.Management

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

190− ( mxAgent

191− , mxSink

192− , liftMX

193− , mxUpdateLocal

194− , mxGetLocal

195− , mxReady

196− , MxEvent(..)

190+ ( MxEvent(..)

197− , MxAgentId(..)

198 )

191 )

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

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

200 ( remoteTable

193 ( remoteTable

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

323

316

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

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

325

318

326−{- note [registry monitoring agent]

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

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

329−

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

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

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

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

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

335− once unregistration occurs.

336− -}

337−

338−registryMonitorAgent :: Process ProcessId

339−registryMonitorAgent = do

340− nid <- getSelfNode

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

342− [ mxSink $ \ev -> do

343− case ev of

344− MxRegistered pid label

345− | processNodeId pid /= nid -> do

346− mref <- liftMX $ monitor pid

347− mxUpdateLocal (Map.insert label mref)

348− MxUnRegistered pid label

349− | processNodeId pid /= nid -> do

350− mrefs <- mxGetLocal

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

352− liftMX $ unmonitorAsync mref

353− mxUpdateLocal (Map.delete label)

354− _ -> return ()

355− mxReady

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

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

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

359− ]

360− where

361− registryMonitorAgentId :: MxAgentId

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

363−

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

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

365startServiceProcesses :: LocalNode -> IO ()

320startServiceProcesses :: LocalNode -> IO ()

366startServiceProcesses node = do

321startServiceProcesses node = do

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

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

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

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

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

378 register "trace.logger" logger

333 register "trace.logger" logger

379− void $ registryMonitorAgent

380 where

334 where

381 loop = do

335 loop = do

382 receiveWait

336 receiveWait

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

538data IncomingTarget =

492data IncomingTarget =

539 Uninit

493 Uninit

540 | ToProc ProcessId (Weak (CQueue Message))

494 | ToProc ProcessId (Weak (CQueue Message))

541− | ToChan TypedChannel

495+ | ToChan SendPortId TypedChannel

542 | ToNode

496 | ToNode

543

497

544data ConnectionState = ConnectionState {

498data ConnectionState = ConnectionState {

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

593 enqueue queue msg -- 'enqueue' is strict

547 enqueue queue msg -- 'enqueue' is strict

594 trace node (MxReceived pid msg)

548 trace node (MxReceived pid msg)

595 go st

549 go st

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

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

597 mChan <- deRefWeak chan'

551 mChan <- deRefWeak chan'

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

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

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

600− forM_ mChan $ \chan -> atomically $

554+ forM_ mChan $ \chan -> do

555+ msg' <- atomically $ do

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

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

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

558+ writeTQueue chan msg

559+ return msg

560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'

603 go st

561 go st

604 Just (_, ToNode) -> do

562 Just (_, ToNode) -> do

605 let ctrlMsg = decode . BSL.fromChunks $ payload

563 let ctrlMsg = decode . BSL.fromChunks $ payload

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

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

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

627 case mChannel of

585 case mChannel of

628 Just channel ->

586 Just channel ->

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

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

630 Nothing ->

588 Nothing ->

631 invalidRequest cid st $

589 invalidRequest cid st $

632 "incoming attempt to connect to unknown channel of"

590 "incoming attempt to connect to unknown channel of"

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

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

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

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

1098 ch <- deRefWeak chan'

1056 ch <- deRefWeak chan'

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

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

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

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

1101− deliverChan (UnencodedMessage _ raw) chan' =

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

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

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

1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)

1103− deliverChan (EncodedMessage _ _) _ =

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

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

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

1105 error "invalid local channel delivery"

1064 error "invalid local channel delivery"

1106

1065

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