git.iko.soy/distributed-process
on master help sign in
29 files changed+1326−683
since splitsnapshots
unifiedsplit
.travis.yml+11−6 benchmarks/Channels.hs+6−5 benchmarks/Latency.hs+7−5 benchmarks/ProcessRing.hs+2−2 benchmarks/Spawns.hs+6−5 benchmarks/Throughput.hs+6−5 distributed-process-tests/distributed-process-tests.cabal+27−13 distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+459−337 distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs+84−45 distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+16−5 distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+354−70 distributed-process-tests/src/Control/Distributed/Process/Tests/Receive.hs+7−8 distributed-process-tests/src/Control/Distributed/Process/Tests/Stats.hs+11−7 distributed-process-tests/src/Control/Distributed/Process/Tests/Tracing.hs+23−11 distributed-process-tests/tests/runInMemory.hs−1 distributed-process-tests/tests/runTCP.hs+2−2 distributed-process.cabal+2−2 src/Control/Distributed/Process/Internal/Primitives.hs+59−43 src/Control/Distributed/Process/Management.hs+34−2 src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs+8−2 src/Control/Distributed/Process/Management/Internal/Types.hs+7 src/Control/Distributed/Process/Node.hs+26−69 src/Control/Distributed/Process/UnsafePrimitives.hs+82−33 stack-ghc-7.10.3.yaml+5−1 stack-ghc-8.0.2.yaml+8−2 stack-ghc-8.2.2.yaml+21 stack-ghc-8.4.4.yaml+20 stack-nightly.yaml+26 stack.yaml+7−2
.travis.yml+11−6
@@ -6,11 +6,11 @@ matrix:
66 include:
77 - env: ARGS="--stack-yaml stack-ghc-7.10.3.yaml"
88 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"
1110 addons: {apt: {packages: [libgmp-dev]}}
12−
13− - env: ARGS=
11+ - env: ARGS="--stack-yaml stack-ghc-8.4.4.yaml"
12+ addons: {apt: {packages: [libgmp-dev]}}
13+ - env: ARGS="--stack-yaml stack-nightly.yaml --resolver nightly"
1414 addons: {apt: {packages: [libgmp-dev]}}
1515
1616cache:
@@ -34,5 +34,10 @@ script:
3434- stack ${ARGS} test $ARG='--plain -t "!Flaky"' ${TEST_PACKAGE}TestCHInTCP
3535- stack ${ARGS} test $ARG='--plain -t "!SpawnReconnect"' ${TEST_PACKAGE}TestClosure
3636- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestStats
37−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMx
38−- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracing
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+
41+notifications:
42+ slack:
43+ secure: LbqbNjK1KiHNiQo/i/Za4vkMD9l9mhmN+PWYFBQkNvGFWffSRayRKFrTqt4znV2p7h4dY1XZFoC7T5RemBVQLq2ppZAUWxkITeu1OUlnmVLDfJLeVYGTR/fn90nP6Y5ITE7T3A07nKmWaRbKpIgFBAOjzgqaM0csscDx3z0WW18=
benchmarks/Channels.hs+6−5
@@ -4,7 +4,7 @@ import Control.Monad
44import Control.Applicative
55import Control.Distributed.Process
66import Control.Distributed.Process.Node
7−import Network.Transport.TCP (createTransport, defaultTCPParameters)
7+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)
88import Data.Binary (encode, decode)
99import qualified Data.ByteString.Lazy as BSL
1010
@@ -36,7 +36,8 @@ initialProcess "CLIENT" = do
3636main :: IO ()
3737main = do
3838 [role, host, port] <- getArgs
39− Right transport <- createTransport
40− host port (\sn -> (host, port)) defaultTCPParameters
41− node <- newLocalNode transport initRemoteTable
42− runProcess node $ initialProcess role
39+ trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters
40+ case trans of
41+ Right transport -> do node <- newLocalNode transport initRemoteTable
42+ runProcess node $ initialProcess role
43+ Left other -> error $ show other
benchmarks/Latency.hs+7−5
@@ -3,7 +3,7 @@ import Control.Monad
33import Control.Applicative
44import Control.Distributed.Process
55import Control.Distributed.Process.Node
6−import Network.Transport.TCP (createTransport, defaultTCPParameters)
6+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)
77import Data.Binary (encode, decode)
88import qualified Data.ByteString.Lazy as BSL
99
@@ -31,7 +31,9 @@ initialProcess "CLIENT" = do
3131main :: IO ()
3232main = do
3333 [role, host, port] <- getArgs
34− Right transport <- createTransport
35− host port (\sn -> (host, sn)) defaultTCPParameters
36− node <- newLocalNode transport initRemoteTable
37− runProcess node $ initialProcess role
34+ trans <- createTransport
35+ (defaultTCPAddr host port) defaultTCPParameters
36+ case trans of
37+ Right transport -> do node <- newLocalNode transport initRemoteTable
38+ runProcess node $ initialProcess role
39+ Left other -> error $ show other
benchmarks/ProcessRing.hs+2−2
@@ -12,7 +12,7 @@ import Control.Monad
1212import Control.Distributed.Process hiding (catch)
1313import Control.Distributed.Process.Node
1414import Control.Exception (catch, SomeException)
15−import Network.Transport.TCP (createTransport, defaultTCPParameters)
15+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)
1616import System.Environment
1717import System.Console.GetOpt
1818
@@ -111,7 +111,7 @@ main = do
111111 (opt, _) <- parseArgv argv
112112 putStrLn $ "options: " ++ (show opt)
113113 Right transport <- createTransport
114− "127.0.0.1" "8090" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters
114+ (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters
115115 node <- newLocalNode transport initRemoteTable
116116 catch (void $ runProcess node $ initialProcess opt)
117117 (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))
benchmarks/Spawns.hs+6−5
@@ -7,7 +7,7 @@ import Control.Monad
77import Control.Applicative
88import Control.Distributed.Process
99import Control.Distributed.Process.Node
10−import Network.Transport.TCP (createTransport, defaultTCPParameters)
10+import Network.Transport.TCP (createTransport, defaultTCPAddr, defaultTCPParameters)
1111import Data.Binary (encode, decode)
1212import qualified Data.ByteString.Lazy as BSL
1313
@@ -42,7 +42,8 @@ initialProcess "CLIENT" = do
4242main :: IO ()
4343main = do
4444 [role, host, port] <- getArgs
45− Right transport <- createTransport
46− host port (\sn -> (host, sn)) defaultTCPParameters
47− node <- newLocalNode transport initRemoteTable
48− runProcess node $ initialProcess role
45+ trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters
46+ case trans of
47+ Right transport -> do node <- newLocalNode transport initRemoteTable
48+ runProcess node $ initialProcess role
49+ Left other -> error $ show other
benchmarks/Throughput.hs+6−5
@@ -6,7 +6,7 @@ import Control.Monad
66import Control.Applicative
77import Control.Distributed.Process
88import Control.Distributed.Process.Node
9−import Network.Transport.TCP (createTransport, defaultTCPParameters)
9+import Network.Transport.TCP (createTransport, defaultTCPParameters, defaultTCPAddr)
1010import Data.Binary
1111import qualified Data.ByteString.Lazy as BSL
1212import Data.Typeable
@@ -67,7 +67,8 @@ initialProcess "CLIENT" = do
6767main :: IO ()
6868main = do
6969 [role, host, port] <- getArgs
70− Right transport <- createTransport
71− host port (\sn -> (host, sn)) defaultTCPParameters
72− node <- newLocalNode transport initRemoteTable
73− runProcess node $ initialProcess role
70+ trans <- createTransport (defaultTCPAddr host port) defaultTCPParameters
71+ case trans of
72+ Right transport -> do node <- newLocalNode transport initRemoteTable
73+ runProcess node $ initialProcess role
74+ Left other -> error $ show other
distributed-process-tests/distributed-process-tests.cabal+27−13
@@ -6,7 +6,7 @@ description: Tests and test suite for Cloud Haskell libraries.
66license: BSD3
77license-file: LICENSE
88Author: Duncan Coutts, Nicolas Wu, Edsko de Vries
9−Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>
9+Maintainer: Tim Watson <watson.timothy@gmail.com>
1010copyright: Well-Typed LLP
1111category: Control, Cloud Haskell
1212build-type: Simple
@@ -31,9 +31,10 @@ library
3131 bytestring >= 0.9 && < 0.12,
3232 distributed-process >= 0.6.0 && < 0.8,
3333 distributed-static,
34+ exceptions >= 0.5,
3435 HUnit >= 1.2 && < 1.7,
3536 network-transport >= 0.4.1.0 && < 0.6,
36− network >= 2.5 && < 2.7,
37+ network >= 2.5 && < 2.9,
3738 random >= 1.0 && < 1.2,
3839 rematch >= 0.1.2.1,
3940 setenv >= 0.1.1.3,
@@ -60,7 +61,7 @@ Test-Suite TestCHInMemory
6061 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH
6162 Build-Depends: base >= 4.4 && < 5,
6263 distributed-process-tests,
63− network >= 2.3 && < 2.7,
64+ network >= 2.3 && < 2.9,
6465 network-transport >= 0.4.1.0 && < 0.6,
6566 network-transport-inmemory >= 0.5,
6667 test-framework >= 0.6 && < 0.9
@@ -75,7 +76,7 @@ Test-Suite TestCHInTCP
7576 if flag(tcp)
7677 Build-Depends: base >= 4.4 && < 5,
7778 distributed-process-tests,
78− network >= 2.3 && < 2.7,
79+ network >= 2.3 && < 2.9,
7980 network-transport >= 0.4.1.0 && < 0.6,
8081 network-transport-tcp >= 0.5 && < 0.7,
8182 test-framework >= 0.6 && < 0.9
@@ -92,7 +93,7 @@ Test-Suite TestClosure
9293 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure
9394 Build-Depends: base >= 4.4 && < 5,
9495 distributed-process-tests,
95− network >= 2.3 && < 2.7,
96+ network >= 2.3 && < 2.9,
9697 network-transport >= 0.4.1.0 && < 0.6,
9798 network-transport-inmemory >= 0.5,
9899 test-framework >= 0.6 && < 0.9
@@ -106,7 +107,7 @@ Test-Suite TestStats
106107 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats
107108 Build-Depends: base >= 4.4 && < 5,
108109 distributed-process-tests,
109− network >= 2.3 && < 2.7,
110+ network >= 2.3 && < 2.9,
110111 network-transport >= 0.4.1.0 && < 0.6,
111112 network-transport-inmemory >= 0.5,
112113 test-framework >= 0.6 && < 0.9
@@ -114,31 +115,44 @@ Test-Suite TestStats
114115 ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
115116 HS-Source-Dirs: tests
116117
117−
118−Test-Suite TestMx
118+Test-Suite TestMxInMemory
119119 Type: exitcode-stdio-1.0
120120 Main-Is: runInMemory.hs
121121 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx
122122 Build-Depends: base >= 4.4 && < 5,
123123 distributed-process-tests,
124− network >= 2.3 && < 2.7,
124+ network >= 2.3 && < 2.9,
125125 network-transport >= 0.4.1.0 && < 0.6,
126126 network-transport-inmemory >= 0.5,
127127 test-framework >= 0.6 && < 0.9
128128 Extensions: CPP
129− ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
129+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
130130 HS-Source-Dirs: tests
131131
132−Test-Suite TestTracing
132+Test-Suite TestTracingInMemory
133133 Type: exitcode-stdio-1.0
134134 Main-Is: runInMemory.hs
135135 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing
136136 Build-Depends: base >= 4.4 && < 5,
137137 distributed-process-tests,
138− network >= 2.3 && < 2.7,
138+ network >= 2.3 && < 2.9,
139139 network-transport >= 0.4.1.0 && < 0.6,
140140 network-transport-inmemory >= 0.5,
141141 test-framework >= 0.6 && < 0.9
142142 Extensions: CPP
143− 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
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
144158 HS-Source-Dirs: tests
distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+459−337
@@ -25,13 +25,18 @@ import Control.Concurrent.MVar
2525 )
2626import Control.Monad (replicateM_, replicateM, forever, void, unless, join)
2727import Control.Exception (SomeException, throwIO, ErrorCall(..))
28−import qualified Control.Exception as Ex (catch)
29−import Control.Applicative ((<$>), (<*>), pure, (<|>))
28+import Control.Monad.Catch (try, catch, finally, mask, onException)
29+import Control.Applicative ((<|>))
3030import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)
31−import Control.Distributed.Process
31+import Control.Distributed.Process hiding
32+ ( try
33+ , catch
34+ , finally
35+ , mask
36+ , onException
37+ )
3238import Control.Distributed.Process.Internal.Types
33− ( NodeId(nodeAddress)
34− , LocalNode(localEndPoint)
39+ ( LocalNode(localEndPoint)
3540 , ProcessExitException(..)
3641 , nullProcessId
3742 , createUnencodedMessage
@@ -39,12 +44,13 @@ import Control.Distributed.Process.Internal.Types
3944import Control.Distributed.Process.Node
4045import Control.Distributed.Process.Debug
4146import Control.Distributed.Process.Management.Internal.Types
47+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)
4248import Control.Distributed.Process.Serializable (Serializable)
43−
44−import Test.HUnit (Assertion, assertFailure, assertBool)
49+import Data.Maybe (isNothing, isJust)
50+import Test.HUnit (Assertion, assertBool, assertFailure)
4551import Test.Framework (Test, testGroup)
4652import Test.Framework.Providers.HUnit (testCase)
47−import Control.Rematch hiding (match)
53+import Control.Rematch hiding (match, isNothing, isJust)
4854import Control.Rematch.Run (Match(..))
4955
5056newtype Ping = Ping ProcessId
@@ -67,7 +73,7 @@ expectThat a matcher = case res of
6773forkTry :: IO () -> IO ThreadId
6874forkTry p = do
6975 tid <- myThreadId
70− forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))
76+ forkIO $ catch p (\e -> throwTo tid (e :: SomeException))
7177
7278-- | The ping server from the paper
7379ping :: Process ()
@@ -77,12 +83,46 @@ ping = do
7783 send partner (Ping self)
7884 ping
7985
86+verifyClient :: String -> MVar Bool -> IO ()
87+verifyClient s b = takeMVar b >>= assertBool s
88+
89+expectPing :: MVar Bool -> Process ()
90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing
91+ where
92+ checkPing (Ping _) = True
93+
8094-- | Quick and dirty synchronous version of whereisRemoteAsync
8195whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
8296whereisRemote nid string = do
8397 whereisRemoteAsync nid string
84− WhereIsReply _ mPid <- expect
85− return mPid
98+ receiveWait [
99+ match (\(WhereIsReply _ mPid) -> return mPid)
100+ ]
101+
102+verifyWhereIsRemote :: NodeId -> String -> Process ProcessId
103+verifyWhereIsRemote n s = whereisRemote n s >>= maybe (die "remote name not found") return
104+
105+waitForExit :: ProcessId -> Process DiedReason
106+waitForExit pid = monitor pid >>= waitForDown
107+
108+waitForDown :: MonitorRef -> Process DiedReason
109+waitForDown ref =
110+ receiveWait [ matchIf (\(ProcessMonitorNotification ref' _ _) -> ref == ref')
111+ (\(ProcessMonitorNotification _ _ dr) -> return dr) ]
112+
113+syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()
114+syncBreakConnection breakConnection nid0 nid1 = do
115+ m <- newEmptyMVar
116+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m
117+ runProcess nid0 $ do
118+ them <- liftIO $ takeMVar m
119+ pinger <- spawnLocal $ forever $ send them ()
120+ _ <- monitorNode (localNodeId nid1)
121+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)
122+ (nodeAddress $ localNodeId nid1)
123+ NodeMonitorNotification _ _ _ <- expect
124+ kill pinger "finished"
125+ return ()
86126
87127data Add = Add ProcessId Double Double deriving (Typeable)
88128data Divide = Divide ProcessId Double Double deriving (Typeable)
@@ -157,12 +197,18 @@ monitorTestProcess theirAddr mOrL un reason monitorSetup done =
157197 unmonitor ref
158198 liftIO $ putMVar done ()
159199 (False, ref) -> do
160− ProcessMonitorNotification ref' pid reason' <- expect
161− True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'
162− liftIO $ putMVar done ()
200+ receiveWait [
201+ match (\(ProcessMonitorNotification ref' pid reason') -> do
202+ liftIO $ do
203+ assertBool "Bad Monitor Signal"
204+ (Just ref' == ref && pid == theirAddr &&
205+ mOrL && reason == reason')
206+ putMVar done ())
207+ ]
163208 )
164209 (\(ProcessLinkException pid reason') -> do
165− True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'
210+ (liftIO $ assertBool "link exception unmatched" $
211+ pid == theirAddr && not mOrL && not un && reason == reason')
166212 liftIO $ putMVar done ()
167213 )
168214
@@ -193,8 +239,10 @@ testPing TestTransport{..} = do
193239 pid <- getSelfPid
194240 replicateM_ numPings $ do
195241 send pingServer (Pong pid)
196− Ping _ <- expect
197− return ()
242+ p <- expectTimeout 3000000
243+ case p of
244+ Just (Ping _) -> return ()
245+ Nothing -> die "Failed to receive Ping"
198246
199247 putMVar clientDone ()
200248
@@ -281,7 +329,7 @@ testMonitorLocalDeadProcess TestTransport{..} mOrL un = do
281329 runProcess localNode $ do
282330 monitor theirAddr
283331 -- wait for the process to die
284− ProcessMonitorNotification _ _ _ <- expect
332+ expect :: Process ProcessMonitorNotification
285333 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
286334
287335 takeMVar done
@@ -355,14 +403,17 @@ testMath TestTransport{..} = do
355403 runProcess localNode $ do
356404 pid <- getSelfPid
357405 send mathServer (Add pid 1 2)
358− 3 <- expect :: Process Double
406+ three <- expect :: Process Double
359407 send mathServer (Divide pid 8 2)
360− 4 <- expect :: Process Double
408+ four <- expect :: Process Double
361409 send mathServer (Divide pid 8 0)
362− DivByZero <- expect
363− liftIO $ putMVar clientDone ()
410+ divByZ <- expect
411+ liftIO $ putMVar clientDone (three, four, divByZ)
364412
365− takeMVar clientDone
413+ res <- takeMVar clientDone
414+ case res of
415+ (3, 4, DivByZero) -> return ()
416+ _ -> error $ "Something went horribly wrong"
366417
367418-- | Send first message (i.e. connect) to an already terminated process
368419-- (without monitoring); then send another message to a second process on
@@ -390,11 +441,9 @@ testSendToTerminated TestTransport{..} = do
390441 pid <- getSelfPid
391442 send server1 "Hi"
392443 send server2 (Pong pid)
393− Ping pid' <- expect
394− True <- return $ pid' == server2
395− liftIO $ putMVar clientDone ()
444+ expectPing clientDone
396445
397− takeMVar clientDone
446+ verifyClient "Expected Ping from server" clientDone
398447
399448-- | Test (non-zero) timeout
400449testTimeout :: TestTransport -> Assertion
@@ -403,10 +452,10 @@ testTimeout TestTransport{..} = do
403452 done <- newEmptyMVar
404453
405454 runProcess localNode $ do
406− Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]
407− liftIO $ putMVar done ()
455+ res <- receiveTimeout 1000000 [match (\Add{} -> return ())]
456+ liftIO $ putMVar done $ res == Nothing
408457
409− takeMVar done
458+ verifyClient "Expected receiveTimeout to timeout..." done
410459
411460-- | Test zero timeout
412461testTimeout0 :: TestTransport -> Assertion
@@ -434,10 +483,9 @@ testTimeout0 TestTransport{..} = do
434483 -- is not interested in, and then a single message that it wants
435484 replicateM_ 10000 $ send server "Irrelevant message"
436485 send server (Pong pid)
437− Ping _ <- expect
438− liftIO $ putMVar clientDone ()
486+ expectPing clientDone
439487
440− takeMVar clientDone
488+ verifyClient "Expected Ping from server" clientDone
441489
442490-- | Test typed channels
443491testTypedChannels :: TestTransport -> Assertion
@@ -460,10 +508,10 @@ testTypedChannels TestTransport{..} = do
460508 runProcess localNode $ do
461509 (clientSendPort, rport) <- newChan
462510 sendChan serverSendPort (clientSendPort, 5)
463− False <- receiveChan rport
464− liftIO $ putMVar clientDone ()
511+ ch <- receiveChan rport
512+ liftIO $ putMVar clientDone $ ch == False
465513
466− takeMVar clientDone
514+ verifyClient "Expected channel to send 'False'" clientDone
467515
468516-- | Test merging receive ports
469517testMergeChannels :: TestTransport -> Assertion
@@ -486,9 +534,8 @@ testMergeChannels TestTransport{..} = do
486534 rs <- mapM charChannel "abc"
487535 m <- mergePorts biased rs
488536 xs <- replicateM 9 $ receiveChan m
489− True <- return $ xs == expected
490− liftIO $ putMVar done ()
491− takeMVar done
537+ liftIO $ putMVar done $ xs == expected
538+ verifyClient "Expected single layer merge to match expected ordering" done
492539
493540 -- Two layers of merging
494541 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()
@@ -499,9 +546,8 @@ testMergeChannels TestTransport{..} = do
499546 ms <- mapM (mergePorts biasedInner) rss
500547 m <- mergePorts biasedOuter ms
501548 xs <- replicateM (9 * 3) $ receiveChan m
502− True <- return $ xs == expected
503− liftIO $ putMVar done ()
504− takeMVar done
549+ liftIO $ putMVar done $ xs == expected
550+ verifyClient "Expected nested channels to match expeted ordering" done
505551
506552 -- Test that if no messages are (immediately) available, the scheduler makes no difference
507553 testBlocked :: LocalNode -> Bool -> IO ()
@@ -510,43 +556,45 @@ testMergeChannels TestTransport{..} = do
510556 done <- newEmptyMVar
511557
512558 forkProcess localNode $ do
513− [sa, sb, sc] <- liftIO $ mapM readMVar vs
514− mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
515− [ -- a, b, c
516− (sa, 'a')
517− , (sb, 'b')
518− , (sc, 'c')
519− -- a, c, b
520− , (sa, 'a')
521− , (sc, 'c')
522− , (sb, 'b')
523− -- b, a, c
524− , (sb, 'b')
525− , (sa, 'a')
526− , (sc, 'c')
527− -- b, c, a
528− , (sb, 'b')
529− , (sc, 'c')
530− , (sa, 'a')
531− -- c, a, b
532− , (sc, 'c')
533− , (sa, 'a')
534− , (sb, 'b')
535− -- c, b, a
536− , (sc, 'c')
537− , (sb, 'b')
538− , (sa, 'a')
539− ]
559+ ss <- liftIO $ mapM readMVar vs
560+ case ss of
561+ [sa, sb, sc] ->
562+ mapM_ ((>> pause 10000) . uncurry sendChan)
563+ [ -- a, b, c
564+ (sa, 'a')
565+ , (sb, 'b')
566+ , (sc, 'c')
567+ -- a, c, b
568+ , (sa, 'a')
569+ , (sc, 'c')
570+ , (sb, 'b')
571+ -- b, a, c
572+ , (sb, 'b')
573+ , (sa, 'a')
574+ , (sc, 'c')
575+ -- b, c, a
576+ , (sb, 'b')
577+ , (sc, 'c')
578+ , (sa, 'a')
579+ -- c, a, b
580+ , (sc, 'c')
581+ , (sa, 'a')
582+ , (sb, 'b')
583+ -- c, b, a
584+ , (sc, 'c')
585+ , (sb, 'b')
586+ , (sa, 'a')
587+ ]
588+ _ -> die "Something went horribly wrong"
540589
541590 forkProcess localNode $ do
542591 (ss, rs) <- unzip <$> replicateM 3 newChan
543592 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss
544593 m <- mergePorts biased rs
545594 xs <- replicateM (6 * 3) $ receiveChan m
546− True <- return $ xs == "abcacbbacbcacabcba"
547− liftIO $ putMVar done ()
595+ liftIO $ putMVar done $ xs == "abcacbbacbcacabcba"
548596
549− takeMVar done
597+ verifyClient "Expected merged ports to match expected ordering" done
550598
551599 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)
552600 mergePorts True = mergePortsBiased
@@ -564,8 +612,9 @@ testTerminate TestTransport{..} = do
564612 localNode <- newLocalNode testTransport initRemoteTable
565613 runProcess localNode $ do
566614 e <- try terminate :: Process (Either ProcessTerminationException ())
567− True <- return $ either show show e == show ProcessTerminationException
568− return ()
615+ if either show show e == show ProcessTerminationException
616+ then return ()
617+ else die "Unexpected result from terminate"
569618
570619testMonitorNode :: TestTransport -> Assertion
571620testMonitorNode TestTransport{..} = do
@@ -576,11 +625,12 @@ testMonitorNode TestTransport{..} = do
576625
577626 runProcess node2 $ do
578627 ref <- monitorNode (localNodeId node1)
579− NodeMonitorNotification ref' nid DiedDisconnect <- expect
580− True <- return $ ref == ref' && nid == localNodeId node1
581− liftIO $ putMVar done ()
628+ receiveWait [
629+ match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->
630+ return $ ref == ref' && nid == localNodeId node1)
631+ ] >>= liftIO . putMVar done
582632
583− takeMVar done
633+ verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done
584634
585635testMonitorLiveNode :: TestTransport -> Assertion
586636testMonitorLiveNode TestTransport{..} = do
@@ -595,15 +645,16 @@ testMonitorLiveNode TestTransport{..} = do
595645 liftIO $ putMVar ready ()
596646 liftIO $ takeMVar readyr
597647 send p ()
598− NodeMonitorNotification ref' nid _ <- expect
599− True <- return $ ref == ref' && nid == localNodeId node1
600− liftIO $ putMVar done ()
648+ receiveWait [
649+ match (\(NodeMonitorNotification ref' nid _) ->
650+ (return $ ref == ref' && nid == localNodeId node1))
651+ ] >>= liftIO . putMVar done
601652
602653 takeMVar ready
603654 closeLocalNode node1
604655 putMVar readyr ()
605656
606− takeMVar done
657+ verifyClient "Expected NodeMonitorNotification for LIVE node" done
607658
608659testMonitorChannel :: TestTransport -> Assertion
609660testMonitorChannel TestTransport{..} = do
@@ -613,18 +664,20 @@ testMonitorChannel TestTransport{..} = do
613664 pid <- forkProcess node1 $ do
614665 sport <- expect :: Process (SendPort ())
615666 ref <- monitorPort sport
616− PortMonitorNotification ref' port' reason <- expect
617− -- reason might be DiedUnknownId if the receive port is GCed before the
618− -- monitor is established (TODO: not sure that this is reasonable)
619− return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)
620− liftIO $ putMVar gotNotification ()
667+ receiveWait [
668+ -- reason might be DiedUnknownId if the receive port is GCed before the
669+ -- monitor is established (TODO: not sure that this is reasonable)
670+ match (\(PortMonitorNotification ref' port' reason) ->
671+ return $ ref' == ref && port' == sendPortId sport &&
672+ (reason == DiedNormal || reason == DiedUnknownId))
673+ ] >>= liftIO . putMVar gotNotification
621674
622675 runProcess node2 $ do
623676 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())
624677 send pid sport
625678 liftIO $ threadDelay 100000
626679
627− takeMVar gotNotification
680+ verifyClient "Expected PortMonitorNotification" gotNotification
628681
629682testRegistry :: TestTransport -> Assertion
630683testRegistry TestTransport{..} = do
@@ -636,20 +689,28 @@ testRegistry TestTransport{..} = do
636689
637690 runProcess node $ do
638691 register "ping" pingServer
639− Just pid <- whereis "ping"
640− True <- return $ pingServer == pid
692+ whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)
641693 us <- getSelfPid
642694 nsend "ping" (Pong us)
643− Ping pid' <- expect
644− True <- return $ pingServer == pid'
645− Left (ProcessRegistrationException "dead" Nothing) <- try $ register "dead" deadProcess
646− Left (ProcessRegistrationException "ping" (Just x)) <- try $ register "ping" deadProcess
647− True <- return $ x == pingServer
648− Left (ProcessRegistrationException "dead" Nothing) <- try $ unregister "dead"
695+ receiveWait [
696+ matchIf (\(Ping pid') -> pingServer == pid') (const $ return ())
697+ ]
698+ checkRegException "dead" Nothing deadProcess
699+ checkRegException "ping" (Just pingServer) deadProcess
700+ try (unregister "dead") >>= checkReg "dead" Nothing
649701 liftIO $ putMVar done ()
650702
651703 takeMVar done
652704
705+ where
706+ checkRegException name pid dead =
707+ try (register name dead) >>= checkReg name pid
708+
709+ checkReg name pid res =
710+ case res of
711+ Left (ProcessRegistrationException name pid) -> return ()
712+ _ -> die $ "Unexpected Registration" ++ show res
713+
653714testRegistryRemoteProcess :: TestTransport -> Assertion
654715testRegistryRemoteProcess TestTransport{..} = do
655716 node1 <- newLocalNode testTransport initRemoteTable
@@ -660,13 +721,13 @@ testRegistryRemoteProcess TestTransport{..} = do
660721
661722 runProcess node2 $ do
662723 register "ping" pingServer
663− Just pid <- whereis "ping"
664− True <- return $ pingServer == pid
724+ whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)
665725 us <- getSelfPid
666726 nsend "ping" (Pong us)
667− Ping pid' <- expect
668− True <- return $ pingServer == pid'
669− liftIO $ putMVar done ()
727+ receiveWait [
728+ matchIf (\(Ping pid') -> pingServer == pid')
729+ (const $ liftIO $ putMVar done ())
730+ ]
670731
671732 takeMVar done
672733
@@ -674,7 +735,6 @@ testRemoteRegistry :: TestTransport -> Assertion
674735testRemoteRegistry TestTransport{..} = do
675736 node1 <- newLocalNode testTransport initRemoteTable
676737 node2 <- newLocalNode testTransport initRemoteTable
677− done <- newEmptyMVar
678738
679739 pingServer <- forkProcess node1 ping
680740 deadProcess <- forkProcess node1 (return ())
@@ -687,32 +747,34 @@ testRemoteRegistry TestTransport{..} = do
687747 "ping" == label' && pid == pingServer)
688748 (\(RegisterReply _ _ _) -> return ()) ]
689749
690− Just pid <- whereisRemote nid1 "ping"
691− True <- return $ pingServer == pid
750+ pid <- verifyWhereIsRemote nid1 "ping"
751+ liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid
752+
692753 us <- getSelfPid
693754 nsendRemote nid1 "ping" (Pong us)
694− Ping pid' <- expect
695− True <- return $ pingServer == pid'
755+ receiveWait [
756+ match (\(Ping pid') -> return $ pingServer == pid')
757+ ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId"
696758
697759 -- test that if process was not registered Nothing is returned
698760 -- in owner field.
699761 registerRemoteAsync nid1 "dead" deadProcess
700− receiveWait [
701− matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')
702− (\(RegisterReply _ _ _) -> return ()) ]
762+ receiveWait [ matchIf (\(RegisterReply label' f mPid) -> "dead" == label')
763+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))
764+ ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"
765+
703766 registerRemoteAsync nid1 "ping" deadProcess
704767 receiveWait [
705− matchIf (\(RegisterReply label' False (Just pid)) ->
706− "ping" == label' && pid == pingServer)
707− (\(RegisterReply _ _ _) -> return ()) ]
768+ matchIf (\(RegisterReply label' False mPid) ->
769+ "ping" == label' && isJust mPid)
770+ (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))
771+ ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"
772+
708773 unregisterRemoteAsync nid1 "dead"
709774 receiveWait [
710− matchIf (\(RegisterReply label' False Nothing) ->
711− "dead" == label' && pid == pingServer)
712− (\(RegisterReply _ _ _) -> return ()) ]
713− liftIO $ putMVar done ()
714−
715− takeMVar done
775+ matchIf (\(RegisterReply label' _ _) -> "dead" == label')
776+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))
777+ ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"
716778
717779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion
718780testRemoteRegistryRemoteProcess TestTransport{..} = do
@@ -728,15 +790,15 @@ testRemoteRegistryRemoteProcess TestTransport{..} = do
728790 receiveWait [
729791 matchIf (\(RegisterReply label' _ _) -> "ping" == label')
730792 (\(RegisterReply _ _ _) -> return ()) ]
731− Just pid <- whereisRemote nid1 "ping"
732− True <- return $ pingServer == pid
793+ pid <- verifyWhereIsRemote nid1 "ping"
794+ liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid
733795 us <- getSelfPid
734796 nsendRemote nid1 "ping" (Pong us)
735− Ping pid' <- expect
736− True <- return $ pingServer == pid'
737− liftIO $ putMVar done ()
797+ receiveWait [
798+ match (\(Ping pid') -> return $ pingServer == pid')
799+ ] >>= liftIO . putMVar done
738800
739− takeMVar done
801+ verifyClient "Expected Ping with ping server's ProcessId" done
740802
741803testSpawnLocal :: TestTransport -> Assertion
742804testSpawnLocal TestTransport{..} = do
@@ -750,15 +812,14 @@ testSpawnLocal TestTransport{..} = do
750812 sport <- expect
751813 sendChan sport (1234 :: Int)
752814
753− sport <- spawnChannelLocal $ \rport -> do
754− (1234 :: Int) <- receiveChan rport
755− send us ()
815+ sport <- spawnChannelLocal $
816+ \rport -> (receiveChan rport :: Process Int) >>= send us
756817
757818 send pid sport
758− () <- expect
759− liftIO $ putMVar done ()
819+ expect >>= liftIO . putMVar done
760820
761− takeMVar done
821+ res <- takeMVar done
822+ assertBool "Expected 1234 :: Int" $ res == (1234 :: Int)
762823
763824testSpawnAsyncStrictness :: TestTransport -> Assertion
764825testSpawnAsyncStrictness TestTransport{..} = do
@@ -787,8 +848,9 @@ testReconnect TestTransport{..} = do
787848 liftIO $ putMVar processA us
788849 msg1 <- expect
789850 msg2 <- expect
790− True <- return $ msg1 == "message 1" && msg2 == "message 3"
791− liftIO $ putMVar sendTestOk ()
851+ liftIO $ do
852+ assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"
853+ putMVar sendTestOk ()
792854
793855 forkProcess node2 $ do
794856 {-
@@ -821,7 +883,7 @@ testReconnect TestTransport{..} = do
821883 matchIf (\(RegisterReply label' _ _) -> "a" == label')
822884 (\(RegisterReply _ _ _) -> return ()) ]
823885
824− Just _ <- whereisRemote nid1 "a"
886+ _ <- whereisRemote nid1 "a"
825887
826888
827889 -- Simulate network failure
@@ -840,9 +902,10 @@ testReconnect TestTransport{..} = do
840902 (\(RegisterReply _ _ _) -> return ()) ]
841903
842904 -- Check
843− Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
844− Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect
845− Just _ <- whereisRemote nid1 "c"
905+ mPid <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
906+ liftIO $ assertBool "Expected remote name to be lost" $ mPid == Nothing
907+ verifyWhereIsRemote nid1 "b" -- this will suceed because the value is set after thereconnect
908+ verifyWhereIsRemote nid1 "c"
846909
847910 liftIO $ putMVar registerTestOk ()
848911
@@ -883,19 +946,19 @@ testUSend usendPrim TestTransport{..} numMessages = do
883946 isSorted _ = True
884947 -- The list can't be null since there are no failures after sending
885948 -- the last message.
886− True <- return $ isSorted msgs && not (null msgs)
887− liftIO $ putMVar usendTestOk ()
949+ liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)
888950
889951 forkProcess node2 $ do
890952 them <- liftIO $ readMVar processA
891953 getSelfPid >>= send them
892− () <- expect
954+ expect :: Process ()
893955 forM_ [1..numMessages] $ \i -> do
894956 liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)
895957 usendPrim them i
896958 liftIO (threadDelay 30000)
897959
898− takeMVar usendTestOk
960+ res <- takeMVar usendTestOk
961+ assertBool "Unexpected failure after sending last msg" res
899962
900963-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
901964-- in between
@@ -921,14 +984,17 @@ testMatchAny TestTransport{..} = do
921984 runProcess localNode $ do
922985 pid <- getSelfPid
923986 send mathServer (Add pid 1 2)
924− 3 <- expect :: Process Double
987+ three <- expect :: Process Double
925988 send mathServer (Divide pid 8 2)
926− 4 <- expect :: Process Double
989+ four <- expect :: Process Double
927990 send mathServer (Divide pid 8 0)
928− DivByZero <- expect
929− liftIO $ putMVar clientDone ()
991+ divByZ <- expect
992+ liftIO $ putMVar clientDone (three, four, divByZ)
930993
931− takeMVar clientDone
994+ res <- takeMVar clientDone
995+ case res of
996+ (3, 4, DivByZero) -> return ()
997+ _ -> error "Unexpected result"
932998
933999-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
9341000-- in between, however we block 'Divide' requests ....
@@ -955,12 +1021,13 @@ testMatchAnyHandle TestTransport{..} = do
9551021 runProcess localNode $ do
9561022 pid <- getSelfPid
9571023 send mathServer (Add pid 1 2)
958− 3 <- expect :: Process Double
1024+ three <- expect :: Process Double
9591025 send mathServer (Divide pid 8 2)
960− Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
961− liftIO $ putMVar clientDone ()
1026+ res <- (expectTimeout 100000) :: Process (Maybe Double)
1027+ liftIO $ putMVar clientDone $ three == 3 && res == Nothing
1028+
1029+ verifyClient "Expected Nothing (i.e. timeout)" clientDone
9621030
963− takeMVar clientDone
9641031 where maybeForward :: ProcessId -> Message -> Process (Maybe ())
9651032 maybeForward s msg =
9661033 handleMessage msg (\m@(Add _ _ _) -> send s m)
@@ -988,8 +1055,10 @@ testMatchAnyNoHandle TestTransport{..} = do
9881055 Just _ -> die "NONSENSE")
9891056 ]
9901057 -- we *must* have removed the message from our mailbox though!!!
991− Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
992− liftIO $ putMVar serverDone ()
1058+ res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
1059+ liftIO $ do
1060+ assertBool "Expected timeout!" $ res == Nothing
1061+ putMVar serverDone ()
9931062 putMVar addr server
9941063
9951064 -- Client
@@ -1032,14 +1101,21 @@ testMatchAnyIf TestTransport{..} = do
10321101 runProcess localNode $ do
10331102 pid <- getSelfPid
10341103 send server (pid, "foo")
1035− "foo" <- expect
1036− send server (pid, "baz")
1037− "baz" <- expect
1104+ foo <- expect
1105+ -- provoking what would be the wrong ordering is informative here...
1106+
10381107 send server (pid, "bar")
1039− Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
1040− liftIO $ putMVar clientDone ()
1108+ bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !?
1109+
1110+ send server (pid, "baz")
1111+ baz <- expect
1112+
1113+ liftIO $ putMVar clientDone (foo, bar, baz)
1114+
1115+ res <- takeMVar clientDone
1116+ let res' = res == ("foo", Nothing, "baz")
1117+ assertBool "Expected timeout due to type mismatch" res'
10411118
1042− takeMVar clientDone
10431119 where tryHandleMessage :: Message -> Process (Maybe ())
10441120 tryHandleMessage msg =
10451121 handleMessage msg (\(pid :: ProcessId, (m :: String))
@@ -1072,12 +1148,13 @@ testMatchMessageWithUnwrap TestTransport{..} = do
10721148 runProcess localNode $ do
10731149 pid <- getSelfPid
10741150 send server (pid, wrapMessage ("foo" :: String))
1075− "foo" <- expect
1151+ foo <- expect
10761152 send server (pid, wrapMessage ("baz" :: String))
1077− "baz" <- expect
1078− liftIO $ putMVar clientDone ()
1153+ baz <- expect
1154+ liftIO $ putMVar clientDone (foo, baz)
10791155
1080− takeMVar clientDone
1156+ res <- takeMVar clientDone
1157+ assertBool "Unexpected unwrapped results" $ res == ("foo", "baz")
10811158
10821159-- Test 'receiveChanTimeout'
10831160testReceiveChanTimeout :: TestTransport -> Assertion
@@ -1095,18 +1172,19 @@ testReceiveChanTimeout TestTransport{..} = do
10951172
10961173 -- Wait for a message with a delay. No message arrives, we should get
10971174 -- Nothing after the delay.
1098− Nothing <- receiveChanTimeout 100000 rp
1175+ receiveChanTimeout 100000 rp >>= maybe (return ()) (const $ die "Expected Timeout")
10991176
11001177 -- Let the sender know that it can send a message.
11011178 liftIO $ putMVar mvSender ()
11021179
11031180 -- Wait for a message with a delay again. Now a message arrives after
11041181 -- 0.1 seconds
1105− Just True <- receiveChanTimeout 20000000 rp
1182+ res <- receiveChanTimeout 20000000 rp >>= maybe (die "Timeout") return
1183+ liftIO $ assertBool "Expected result to be 'True'" res
11061184
11071185 -- Wait for a message with zero timeout: non-blocking check. No message is
11081186 -- available, we get Nothing
1109− Nothing <- receiveChanTimeout 0 rp
1187+ receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout")
11101188
11111189 -- Let the sender know that it can send a message.
11121190 liftIO $ putMVar mvSender ()
@@ -1116,7 +1194,7 @@ testReceiveChanTimeout TestTransport{..} = do
11161194 liftIO $ threadDelay 100000
11171195 mb <- receiveChanTimeout 0 rp
11181196 case mb of
1119− Just b -> do False <- return b; return ()
1197+ Just b -> do liftIO $ assertBool "Unexpected Message" $ not b
11201198 _ -> loop
11211199
11221200 liftIO $ putMVar done ()
@@ -1153,8 +1231,8 @@ testReceiveChanFeatures TestTransport{..} = do
11531231
11541232 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]
11551233
1156− True <- receiveChan rp1
1157− False <- receiveChan rp1
1234+ receiveChan rp1 >>= liftIO . assertBool "Expected True"
1235+ receiveChan rp1 >>= liftIO . assertBool "Expected False" . not
11581236
11591237 -- Test Applicative instance
11601238
@@ -1163,7 +1241,7 @@ testReceiveChanFeatures TestTransport{..} = do
11631241
11641242 let rp2 = pure (+) <*> rpInt <*> rpInt
11651243
1166− 7 <- receiveChan rp2
1244+ receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7)
11671245
11681246 -- Test Alternative instance
11691247
@@ -1172,8 +1250,8 @@ testReceiveChanFeatures TestTransport{..} = do
11721250
11731251 let rp3 = (even <$> rpInt) <|> rpBool
11741252
1175− False <- receiveChan rp3
1176− True <- receiveChan rp3
1253+ receiveChan rp3 >>= liftIO . assertBool "Expected False" . not
1254+ receiveChan rp3 >>= liftIO . assertBool "Expected True"
11771255
11781256 -- Test Monad instance
11791257
@@ -1187,17 +1265,86 @@ testReceiveChanFeatures TestTransport{..} = do
11871265 then rpInt
11881266 else return 7
11891267
1190− 5 <- receiveChan rp4
1191− 7 <- receiveChan rp4
1268+ receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)
1269+ receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7)
11921270
11931271 liftIO $ putMVar done ()
11941272
11951273 takeMVar done
11961274
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+
11971345testKillLocal :: TestTransport -> Assertion
11981346testKillLocal TestTransport{..} = do
11991347 localNode <- newLocalNode testTransport initRemoteTable
1200− done <- newEmptyMVar
12011348
12021349 pid <- forkProcess localNode $ do
12031350 liftIO $ threadDelay 1000000
@@ -1206,17 +1353,18 @@ testKillLocal TestTransport{..} = do
12061353 ref <- monitor pid
12071354 us <- getSelfPid
12081355 kill pid "TestKill"
1209− ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
1210− True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"
1211− liftIO $ putMVar done ()
1212−
1213− takeMVar done
1356+ mn <- expect
1357+ case mn of
1358+ ProcessMonitorNotification ref' pid' (DiedException ex) ->
1359+ case ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill" of
1360+ True -> return ()
1361+ False -> die "Invalid ProcessMonitorNotification received"
1362+ _ -> die "Something went horribly wrong"
12141363
12151364testKillRemote :: TestTransport -> Assertion
12161365testKillRemote TestTransport{..} = do
12171366 node1 <- newLocalNode testTransport initRemoteTable
12181367 node2 <- newLocalNode testTransport initRemoteTable
1219− done <- newEmptyMVar
12201368
12211369 pid <- forkProcess node1 $ do
12221370 liftIO $ threadDelay 1000000
@@ -1225,11 +1373,17 @@ testKillRemote TestTransport{..} = do
12251373 ref <- monitor pid
12261374 us <- getSelfPid
12271375 kill pid "TestKill"
1228− ProcessMonitorNotification ref' pid' (DiedException reason) <- expect
1229− True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"
1230− liftIO $ putMVar done ()
1231−
1232− takeMVar done
1376+ mn <- expect
1377+ case mn of
1378+ ProcessMonitorNotification ref' pid' (DiedException reason) ->
1379+ case (ref == ref', pid == pid', reason == "killed-by=" ++ show us ++ ",reason=TestKill") of
1380+ (True, True, True) -> return ()
1381+ (a, b, c) -> do
1382+ let a' = if a then "" else "Invalid ref"
1383+ let b' = if b then "" else "Invalid pid"
1384+ let c' = if c then "" else "Invalid message"
1385+ die $ unwords [a', b', c']
1386+ _ -> die "Received unexpected message"
12331387
12341388testCatchesExit :: TestTransport -> Assertion
12351389testCatchesExit TestTransport{..} = do
@@ -1255,7 +1409,10 @@ testHandleMessageIf TestTransport{..} = do
12551409 self <- getSelfPid
12561410 send self (5 :: Integer, 10 :: Integer)
12571411 msg <- receiveWait [ matchMessage return ]
1258− Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")
1412+ handleMessageIf msg
1413+ (\() -> True)
1414+ (\() -> die "whoops") >>= maybe (return ())
1415+ (const $ die "Expected Mismatch")
12591416 handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)
12601417 (\input -> liftIO $ putMVar done input)
12611418 return ()
@@ -1300,8 +1457,11 @@ testDie TestTransport{..} = do
13001457 (die ("foobar", 123 :: Int))
13011458 `catchExit` \_from reason -> do
13021459 -- TODO: should verify that 'from' has the right value
1303− True <- return $ reason == ("foobar", 123 :: Int)
1460+ let res = reason == ("foobar", 123 :: Int)
13041461 liftIO $ putMVar done ()
1462+ if res
1463+ then return ()
1464+ else die "Something went horribly wrong"
13051465
13061466 takeMVar done
13071467
@@ -1315,8 +1475,11 @@ testPrettyExit TestTransport{..} = do
13151475 `catch` \ex@(ProcessExitException from _) ->
13161476 let expected = "exit-from=" ++ (show from)
13171477 in do
1318− True <- return $ (show ex) == expected
1478+ let res = (show ex) == expected
13191479 liftIO $ putMVar done ()
1480+ if res
1481+ then return ()
1482+ else die "Something went horribly wrong"
13201483
13211484 takeMVar done
13221485
@@ -1332,17 +1495,26 @@ testExitLocal TestTransport{..} = do
13321495 pid <- forkProcess localNode $ do
13331496 (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do
13341497 -- TODO: should verify that 'from' has the right value
1335− True <- return $ reason == "TestExit"
1498+ res <- return $ reason == "TestExit"
13361499 liftIO $ putMVar supervisedDone ()
1500+ if res
1501+ then return ()
1502+ else die "Something went horribly wrong"
13371503
13381504 runProcess localNode $ do
13391505 liftIO $ takeMVar handlerSetUp
13401506 ref <- monitor pid
13411507 exit pid "TestExit"
13421508 -- This time the client catches the exception, so it dies normally
1343− ProcessMonitorNotification ref' pid' DiedNormal <- expect
1344− True <- return $ ref == ref' && pid == pid'
1345− liftIO $ putMVar supervisorDone ()
1509+ mn <- expect
1510+ case mn of
1511+ ProcessMonitorNotification ref' pid' DiedNormal -> do
1512+ let res = ref == ref' && pid == pid'
1513+ liftIO $ putMVar supervisorDone ()
1514+ if res
1515+ then return ()
1516+ else die "Something went horribly wrong"
1517+ _ -> die "Something went horribly wrong"
13461518
13471519 takeMVar supervisedDone
13481520 takeMVar supervisorDone
@@ -1358,125 +1530,97 @@ testExitRemote TestTransport{..} = do
13581530 (receiveWait [] :: Process ()) -- block forever
13591531 `catchExit` \_from reason -> do
13601532 -- TODO: should verify that 'from' has the right value
1361− True <- return $ reason == "TestExit"
1533+ res <- return $ reason == "TestExit"
13621534 liftIO $ putMVar supervisedDone ()
1535+ if res
1536+ then return ()
1537+ else die "Something went horribly wrong"
13631538
13641539 runProcess node2 $ do
13651540 ref <- monitor pid
13661541 exit pid "TestExit"
1367− ProcessMonitorNotification ref' pid' DiedNormal <- expect
1368− True <- return $ ref == ref' && pid == pid'
1369− liftIO $ putMVar supervisorDone ()
1542+ mn <- expect
1543+ case mn of
1544+ ProcessMonitorNotification ref' pid' DiedNormal -> do
1545+ res' <- return $ ref == ref' && pid == pid'
1546+ liftIO $ putMVar supervisorDone ()
1547+ if res'
1548+ then return ()
1549+ else die "Something went horribly wrong"
1550+ _ -> die "Something went horribly wrong"
13701551
13711552 takeMVar supervisedDone
13721553 takeMVar supervisorDone
13731554
13741555testRegistryMonitoring :: TestTransport -> Assertion
13751556testRegistryMonitoring TestTransport{..} = do
1376− localNode <- newLocalNode testTransport initRemoteTable
1377− remoteNode <- newLocalNode testTransport initRemoteTable
1378− return ()
1557+ node1 <- newLocalNode testTransport initRemoteTable
1558+ node2 <- newLocalNode testTransport initRemoteTable
13791559
1380− -- Local process. Test if local process will be removed from
1381− -- registry when it dies.
1382− box <- newEmptyMVar
1383− runProcess localNode $ do
1384− pid <- spawnLocal $ do
1385− expect
1386− register "test" pid
1387− tpid <- whereis "test"
1388− if tpid == Just pid
1389− then do _ <- monitor pid
1390− send pid ()
1391− ProcessMonitorNotification{} <- expect
1392− tpid1 <- whereis "test"
1393− liftIO $ putMVar box (Nothing == tpid1)
1394− else liftIO $ putMVar box False
1395−
1396− takeMVar box >>= assertBool "expected local process to not be registered"
1397− return ()
1398−
1399− -- Remote process. Test if remote process entry is removed
1400− -- from registry when process dies.
1401− remote1 <- testRemote remoteNode
1402− runProcess localNode $
1403− let waitpoll = do
1404− w <- whereis "test" :: Process (Maybe ProcessId)
1405− forM_ w (const waitpoll)
1406− in do register "test" remote1
1407− send remote1 ()
1408− waitpoll
1409− return ()
1410− return ()
1411−
1412− -- Many labels. Test if all labels associated with process
1413− -- are removed from registry when it dies.
1414− remote2 <- testRemote remoteNode
1415− runProcess localNode $
1416− let waitpoll = do
1417− w1 <- whereis "test-3" :: Process (Maybe ProcessId)
1418− w2 <- whereis "test-4" :: Process (Maybe ProcessId)
1419− forM_ (w1 <|> w2) (const waitpoll)
1420− in do register "test-3" remote2
1421− register "test-4" remote2
1422− send remote2 ()
1423− waitpoll
1424− return ()
1425−
1426−{- XXX: waiting including patch for nsend for remote process
1427− remote3 <- testRemote remoteNode
1428− remote4 <- testRemote remoteNode
1429− -- test many labels
1430− runProcess localNode $ do
1431− register "test-3" remote3
1432− reregister "test-3" remote4
1433− send remote3 ()
1434− liftIO $ threadDelay 50000 -- XXX: racy
1435− monitor remote4
1436− nsend "test-3" ()
1437− ProcessMonitorNotification{} <- expect
1438− return ()
1439−-}
1440−
1441− -- Test registerRemoteAsync properties. Add a local process to
1442− -- remote registry and checks that it is removed
1443− -- when the process dies.
1444− remote5 <- testRemote remoteNode
1445− runProcess localNode $ do
1446− registerRemoteAsync (localNodeId remoteNode) "test" remote5
1447− receiveWait [
1448− match (\(RegisterReply _ True _) -> return ())
1449− ] >>= send remote5
1450− let waitpoll = do
1451− whereisRemoteAsync (localNodeId remoteNode) "test"
1452− receiveWait [
1453− match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))
1454− ]
1455− waitpoll
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"
14561567
1457− -- Add remote process to remote registry and checks if
1458− -- entry is removed then process is dead.
1459− remote6 <- testRemote localNode
1460− runProcess localNode $ do
1461− registerRemoteAsync (localNodeId remoteNode) "test" remote6
1462− receiveWait [
1463− match (\(RegisterReply _ True _) -> return ())
1464− ] >>= send remote6
1465− let waitpoll = do
1466− whereisRemoteAsync (localNodeId remoteNode) "test"
1467− receiveWait [
1468− match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))
1469− ]
1470− waitpoll
1471− where
1472− testRemote node = do
1473− -- test many labels
1474− pidBox <- newEmptyMVar
1475− forkProcess node $ do
1476− us <- getSelfPid
1477− liftIO $ putMVar pidBox us
1478− expect :: Process ()
1479− takeMVar pidBox
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"
14801624
14811625testUnsafeSend :: TestTransport -> Assertion
14821626testUnsafeSend TestTransport{..} = do
@@ -1493,8 +1637,7 @@ testUnsafeSend TestTransport{..} = do
14931637 void $ forkProcess localNode $ do
14941638 serverPid <- liftIO $ takeMVar serverAddr
14951639 getSelfPid >>= unsafeSend serverPid
1496− () <- expect
1497− liftIO $ putMVar clientDone ()
1640+ expect >>= liftIO . putMVar clientDone
14981641
14991642 takeMVar clientDone
15001643
@@ -1513,8 +1656,7 @@ testUnsafeUSend TestTransport{..} = do
15131656 void $ forkProcess localNode $ do
15141657 serverPid <- liftIO $ takeMVar serverAddr
15151658 getSelfPid >>= unsafeUSend serverPid
1516− () <- expect
1517− liftIO $ putMVar clientDone ()
1659+ expect >>= liftIO . putMVar clientDone
15181660
15191661 takeMVar clientDone
15201662
@@ -1525,8 +1667,7 @@ testUnsafeNSend TestTransport{..} = do
15251667 localNode <- newLocalNode testTransport initRemoteTable
15261668
15271669 pid <- forkProcess localNode $ do
1528− () <- expect
1529− liftIO $ putMVar clientDone ()
1670+ expect >>= liftIO . putMVar clientDone
15301671
15311672 void $ runProcess localNode $ do
15321673 register "foobar" pid
@@ -1544,8 +1685,7 @@ testUnsafeNSendRemote TestTransport{..} = do
15441685 _ <- forkProcess localNode1 $ do
15451686 getSelfPid >>= register "foobar"
15461687 liftIO $ putMVar clientDone ()
1547− () <- expect
1548− liftIO $ putMVar clientDone ()
1688+ expect >>= liftIO . putMVar clientDone
15491689
15501690 takeMVar clientDone
15511691 void $ runProcess localNode2 $ do
@@ -1569,7 +1709,7 @@ testUnsafeSendChan TestTransport{..} = do
15691709 serverPid <- liftIO $ takeMVar serverAddr
15701710 (sp, rp) <- newChan
15711711 unsafeSend serverPid sp
1572− () <- receiveChan rp
1712+ receiveChan rp :: Process ()
15731713 liftIO $ putMVar clientDone ()
15741714
15751715 takeMVar clientDone
@@ -1583,8 +1723,8 @@ testCallLocal TestTransport{..} = do
15831723 runProcess node $ do
15841724 r <- (/=) <$> getSelfPid <*> callLocal getSelfPid
15851725 liftIO $ putMVar result r
1586− True <- takeMVar result
1587− return ()
1726+
1727+ takeMVar result >>= assertBool "Expected 'True'"
15881728
15891729 -- Testing that when callLocal is interrupted, the worker is interrupted.
15901730 ibox <- newIORef False
@@ -1601,10 +1741,9 @@ testCallLocal TestTransport{..} = do
16011741 (send keeper ())
16021742 caller <- expect
16031743 exit caller "test"
1604− () <- expect
1605− return ()
1606− True <- readIORef ibox
1607− return ()
1744+ expect :: Process ()
1745+
1746+ readIORef ibox >>= assertBool "Expected 'True'"
16081747
16091748 -- Testing that when the worker raises an exception, the exception is propagated to the parent.
16101749 ibox2 <- newIORef False
@@ -1613,8 +1752,8 @@ testCallLocal TestTransport{..} = do
16131752 liftIO $ writeIORef ibox2 $ case r of
16141753 Left (ErrorCall "e") -> True
16151754 _ -> False
1616− True <- readIORef ibox
1617− return ()
1755+
1756+ readIORef ibox >>= assertBool "Expected 'True'"
16181757
16191758 -- Test that caller waits for the worker in correct situation
16201759 ibox3 <- newIORef False
@@ -1631,8 +1770,8 @@ testCallLocal TestTransport{..} = do
16311770 liftIO $ putMVar result3 =<< readIORef ibox3
16321771 worker <- expect
16331772 send worker ()
1634− True <- takeMVar result3
1635− return ()
1773+
1774+ takeMVar result3 >>= assertBool "Expected 'True'"
16361775
16371776 -- Test that caller waits for the worker in case when caller gets an exception
16381777 ibox4 <- newIORef False
@@ -1648,13 +1787,9 @@ testCallLocal TestTransport{..} = do
16481787 `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)
16491788 caller <- expect
16501789 exit caller "hi!"
1651− True <- takeMVar result4
1652− return ()
1653− -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.
1654−
1655−
1656−
16571790
1791+ takeMVar result4 >>= assertBool "Expected 'True'"
1792+ -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.
16581793
16591794tests :: TestTransport -> IO [Test]
16601795tests testtrans = return [
@@ -1679,6 +1814,7 @@ tests testtrans = return [
16791814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)
16801815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)
16811816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)
1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)
16821818 , testCase "KillLocal" (testKillLocal testtrans)
16831819 , testCase "KillRemote" (testKillRemote testtrans)
16841820 , testCase "Die" (testDie testtrans)
@@ -1688,7 +1824,7 @@ tests testtrans = return [
16881824 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)
16891825 , testCase "ExitLocal" (testExitLocal testtrans)
16901826 , testCase "ExitRemote" (testExitRemote testtrans)
1691− , testCase "TestRegistryMonitor" (testRegistryMonitoring testtrans)
1827+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)
16921828 , testCase "TextCallLocal" (testCallLocal testtrans)
16931829 -- Unsafe Primitives
16941830 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)
@@ -1741,17 +1877,3 @@ tests testtrans = return [
17411877 , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)
17421878 ]
17431879 ]
1744−
1745−syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()
1746−syncBreakConnection breakConnection nid0 nid1 = do
1747− m <- newEmptyMVar
1748− _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m
1749− runProcess nid0 $ do
1750− them <- liftIO $ takeMVar m
1751− pinger <- spawnLocal $ forever $ send them ()
1752− _ <- monitorNode (localNodeId nid1)
1753− liftIO $ breakConnection (nodeAddress $ localNodeId nid0)
1754− (nodeAddress $ localNodeId nid1)
1755− NodeMonitorNotification _ _ _ <- expect
1756− kill pinger "finished"
1757− return ()
distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs+84−45
@@ -169,8 +169,11 @@ testUnclosure TestTransport{..} rtable = do
169169 node <- newLocalNode testTransport rtable
170170 done <- newEmptyMVar
171171 forkProcess node $ do
172− 120 <- join . unClosure $ factorialClosure 5
172+ i <- join . unClosure $ factorialClosure 5
173173 liftIO $ putMVar done ()
174+ if i == 720
175+ then return ()
176+ else error "Something went horribly wrong"
174177 takeMVar done
175178
176179testBind :: TestTransport -> RemoteTable -> Assertion
@@ -180,8 +183,11 @@ testBind TestTransport{..} rtable = do
180183 runProcess node $ do
181184 us <- getSelfPid
182185 join . unClosure $ sendFac 6 us
183− (720 :: Int) <- expect
186+ (i :: Int) <- expect
184187 liftIO $ putMVar done ()
188+ if i == 720
189+ then return ()
190+ else error "Something went horribly wrong"
185191 takeMVar done
186192
187193testSendPureClosure :: TestTransport -> RemoteTable -> Assertion
@@ -194,7 +200,7 @@ testSendPureClosure TestTransport{..} rtable = do
194200 addr <- forkProcess node $ do
195201 cl <- expect
196202 fn <- unClosure cl :: Process (Int -> Int)
197− 13 <- return $ fn 6
203+ (_ :: Int) <- return $ fn 6
198204 liftIO $ putMVar serverDone ()
199205 putMVar serverAddr addr
200206
@@ -218,8 +224,11 @@ testSendIOClosure TestTransport{..} rtable = do
218224 liftIO $ do
219225 someMVar <- newEmptyMVar
220226 io someMVar
221− 5 <- readMVar someMVar
227+ i <- readMVar someMVar
222228 putMVar serverDone ()
229+ if i == 5
230+ then return ()
231+ else error "Something went horribly wrong"
223232 putMVar serverAddr addr
224233
225234 forkIO $ do
@@ -248,8 +257,10 @@ testSendProcClosure TestTransport{..} rtable = do
248257 runProcess node $ do
249258 pid <- getSelfPid
250259 send theirAddr (cpSend $(mkStatic 'sdictInt) pid)
251− 5 <- expect :: Process Int
252− liftIO $ putMVar clientDone ()
260+ i <- expect :: Process Int
261+ if i == 5
262+ then liftIO $ putMVar clientDone ()
263+ else error "Something went horribly wrong"
253264
254265 takeMVar clientDone
255266
@@ -269,8 +280,9 @@ testSpawn TestTransport{..} rtable = do
269280 pid <- getSelfPid
270281 pid' <- spawn nid (sendPidClosure pid)
271282 pid'' <- expect
272− True <- return $ pid' == pid''
273− liftIO $ putMVar clientDone ()
283+ if pid' == pid''
284+ then liftIO $ putMVar clientDone ()
285+ else error "Something went horribly wrong"
274286
275287 takeMVar clientDone
276288
@@ -294,8 +306,9 @@ testSpawnRace TestTransport{..} rtable = do
294306 spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid
295307 pid' <- expect :: Process ProcessId
296308 pid'' <- expect :: Process ProcessId
297− True <- return $ pid' == pid''
298− return ()
309+ if pid' == pid''
310+ then return ()
311+ else error "Something went horribly wrong"
299312
300313 where
301314
@@ -332,8 +345,10 @@ testCall TestTransport{..} rtable = do
332345 node <- newLocalNode testTransport rtable
333346 nid <- readMVar serverNodeAddr
334347 runProcess node $ do
335− (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
336− liftIO $ putMVar clientDone ()
348+ (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
349+ if a == 120
350+ then liftIO $ putMVar clientDone ()
351+ else error "something went horribly wrong"
337352
338353 takeMVar clientDone
339354
@@ -350,8 +365,10 @@ testCallBind TestTransport{..} rtable = do
350365 node <- newLocalNode testTransport rtable
351366 nid <- readMVar serverNodeAddr
352367 runProcess node $ do
353− (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
354− liftIO $ putMVar clientDone ()
368+ (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
369+ if a == 120
370+ then liftIO $ putMVar clientDone ()
371+ else error "Something went horribly wrong"
355372
356373 takeMVar clientDone
357374
@@ -362,9 +379,11 @@ testSeq TestTransport{..} rtable = do
362379 runProcess node $ do
363380 us <- getSelfPid
364381 join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
365− 120 :: Int <- expect
366− 720 :: Int <- expect
367− liftIO $ putMVar done ()
382+ a :: Int <- expect
383+ b :: Int <- expect
384+ if a == 120 && b == 720
385+ then liftIO $ putMVar done ()
386+ else error "Something went horribly wrong"
368387 takeMVar done
369388
370389-- Test 'spawnSupervised'
@@ -393,25 +412,33 @@ testSpawnSupervised TestTransport{..} rtable = do
393412 throw supervisorDeath
394413
395414 forkProcess node2 $ do
396− [super, child] <- liftIO $ mapM readMVar [superPid, childPid]
397− ref <- monitor child
398− self <- getSelfPid
399− let waitForMOrL = do
400− liftIO $ threadDelay 10000
401− mpinfo <- getProcessInfo child
402− case mpinfo of
403− Nothing -> waitForMOrL
404− Just pinfo ->
405− unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL
406− waitForMOrL
407− liftIO $ putMVar linkUp ()
408− -- because monitor message was sent before message to process
409− -- we hope that it will be processed before
410− ProcessMonitorNotification ref' pid' (DiedException e) <- expect
411− True <- return $ ref' == ref
412− && pid' == child
413− && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))
414− liftIO $ putMVar thirdProcessDone ()
415+ res <- liftIO $ mapM readMVar [superPid, childPid]
416+ case res of
417+ [super, child] -> do
418+ ref <- monitor child
419+ self <- getSelfPid
420+ let waitForMOrL = do
421+ liftIO $ threadDelay 10000
422+ mpinfo <- getProcessInfo child
423+ case mpinfo of
424+ Nothing -> waitForMOrL
425+ Just pinfo ->
426+ unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL
427+ waitForMOrL
428+ liftIO $ putMVar linkUp ()
429+ -- because monitor message was sent before message to process
430+ -- we hope that it will be processed before
431+ res' <- expect
432+ case res' of
433+ (ProcessMonitorNotification ref' pid' (DiedException e)) ->
434+ if (ref' == ref && pid' == child &&
435+ e == show (ProcessLinkException super
436+ (DiedException (show supervisorDeath))))
437+ then liftIO $ putMVar thirdProcessDone ()
438+ else error "Something went horribly wrong"
439+ _ -> error "Something went horribly wrong"
440+
441+ _ -> die $ "Something went horribly wrong"
415442
416443 takeMVar thirdProcessDone
417444 where
@@ -426,8 +453,10 @@ testSpawnInvalid TestTransport{..} rtable = do
426453 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
427454 ProcessMonitorNotification ref' pid' _reason <- expect
428455 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
429− True <- return $ ref' == ref && pid == pid'
430− liftIO $ putMVar done ()
456+ res <- return $ ref' == ref && pid == pid'
457+ if res == True
458+ then liftIO $ putMVar done ()
459+ else error "Something went horribly wrong"
431460 takeMVar done
432461
433462testClosureExpect :: TestTransport -> RemoteTable -> Assertion
@@ -439,8 +468,10 @@ testClosureExpect TestTransport{..} rtable = do
439468 us <- getSelfPid
440469 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
441470 send them (1234 :: Int)
442− (1234 :: Int) <- expect
443− liftIO $ putMVar done ()
471+ (res :: Int) <- expect
472+ if res == 1234
473+ then liftIO $ putMVar done ()
474+ else error "Something went horribly wrong"
444475 takeMVar done
445476
446477testSpawnChannel :: TestTransport -> RemoteTable -> Assertion
@@ -465,8 +496,10 @@ testTDict TestTransport{..} rtable = do
465496 done <- newEmptyMVar
466497 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
467498 forkProcess node1 $ do
468− True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
469− liftIO $ putMVar done ()
499+ res <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
500+ if res == True
501+ then liftIO $ putMVar done ()
502+ else error "Something went horribly wrong..."
470503 takeMVar done
471504
472505testFib :: TestTransport -> RemoteTable -> Assertion
@@ -477,8 +510,11 @@ testFib TestTransport{..} rtable = do
477510 forkProcess (head nodes) $ do
478511 (sport, rport) <- newChan
479512 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
480− 55 <- receiveChan rport :: Process Integer
513+ ff <- receiveChan rport :: Process Integer
481514 liftIO $ putMVar done ()
515+ if ff /= 55
516+ then die $ "Something went horribly wrong"
517+ else return ()
482518
483519 takeMVar done
484520
@@ -486,7 +522,7 @@ testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion
486522testSpawnReconnect testtrans@TestTransport{..} rtable = do
487523 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
488524 let nid1 = localNodeId node1
489− nid2 = localNodeId node2
525+ -- nid2 = localNodeId node2
490526 done <- newEmptyMVar
491527 iv <- newMVar (0 :: Int)
492528
@@ -503,9 +539,12 @@ testSpawnReconnect testtrans@TestTransport{..} rtable = do
503539 liftIO $ threadDelay 100000
504540
505541 count <- liftIO $ takeMVar iv
506− True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
542+ res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
507543
508544 liftIO $ putMVar done ()
545+ if res /= True
546+ then error "Something went horribly wrong"
547+ else return ()
509548
510549 takeMVar done
511550
distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+16−5
@@ -20,6 +20,7 @@ module Control.Distributed.Process.Tests.Internal.Utils
2020 -- ping !
2121 , Ping(Ping)
2222 , ping
23+ , pause
2324 , shouldBe
2425 , shouldMatch
2526 , shouldContain
@@ -73,15 +74,16 @@ import Control.Concurrent
7374import Control.Concurrent.MVar
7475 ( putMVar
7576 )
76−import Control.Distributed.Process
77+import Control.Distributed.Process hiding (finally, catch)
7778import Control.Distributed.Process.Node
7879import Control.Distributed.Process.Serializable()
7980
8081import Control.Exception (AsyncException(ThreadKilled), SomeException)
81−import Control.Monad (forever)
82+import Control.Monad (forever, void)
83+import Control.Monad.Catch (finally, catch)
8284import Control.Monad.STM (atomically)
8385import Control.Rematch hiding (match)
84−import Control.Rematch.Run
86+import Control.Rematch.Run
8587import Data.Binary
8688import Data.Typeable (Typeable)
8789
@@ -89,6 +91,7 @@ import Test.HUnit (Assertion, assertFailure)
8991import Test.HUnit.Base (assertBool)
9092
9193import GHC.Generics
94+import System.Timeout (timeout)
9295
9396-- | A mutable cell containing a test result.
9497type TestResult a = MVar a
@@ -107,10 +110,18 @@ data TestProcessControl = Stop | Go | Report ProcessId
107110
108111instance Binary TestProcessControl where
109112
113+data Private = Private
114+ deriving (Typeable, Generic)
115+instance Binary Private where
116+
110117-- | Does exactly what it says on the tin, doing so in the @Process@ monad.
111118noop :: Process ()
112119noop = return ()
113120
121+pause :: Int -> Process ()
122+pause delay =
123+ void $ receiveTimeout delay [ match (\Private -> return ()) ]
124+
114125synchronisedAssertion :: Eq a
115126 => String
116127 -> LocalNode
@@ -223,10 +234,10 @@ testProcessReport pid = do
223234tryRunProcess :: LocalNode -> Process () -> IO ()
224235tryRunProcess node p = do
225236 tid <- liftIO myThreadId
226− runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
237+ void $ timeout (1000000 * 60 * 5 :: Int) $
238+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
227239
228240tryForkProcess :: LocalNode -> Process () -> IO ProcessId
229241tryForkProcess node p = do
230242 tid <- liftIO myThreadId
231243 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
232−
distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+354−70
@@ -1,12 +1,23 @@
1−{-# LANGUAGE DeriveGeneric #-}
1+{-# LANGUAGE DeriveGeneric #-}
2+{-# LANGUAGE ParallelListComp #-}
3+
24module Control.Distributed.Process.Tests.Mx (tests) where
35
46import Control.Distributed.Process.Tests.Internal.Utils
57import Network.Transport.Test (TestTransport(..))
6−
7−import Control.Concurrent (threadDelay)
8−import Control.Distributed.Process
8+import Control.Distributed.Process hiding (bracket, finally, try)
9+import Control.Distributed.Process.Internal.Types
10+ ( ProcessExitException(..)
11+ , unsafeCreateUnencodedMessage
12+ )
913import Control.Distributed.Process.Node
14+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
15+ ( send
16+ , nsend
17+ , nsendRemote
18+ , usend
19+ , sendChan
20+ )
1021import Control.Distributed.Process.Management
1122 ( MxEvent(..)
1223 , MxAgentId(..)
@@ -21,15 +32,15 @@ import Control.Distributed.Process.Management
2132 , mxUpdateLocal
2233 , mxNotify
2334 , mxBroadcast
24− , mxGetId
2535 )
26−import Control.Monad (void)
36+import Control.Monad (void, unless)
37+import Control.Monad.Catch(finally, bracket, try)
2738import Control.Rematch (equalTo)
2839import Data.Binary
29−import Data.List (find, sort)
30−import Data.Maybe (isJust)
40+import Data.List (find, sort, intercalate)
41+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
3142import Data.Typeable
32−import GHC.Generics
43+import GHC.Generics hiding (from)
3344#if ! MIN_VERSION_base(4,6,0)
3445import Prelude hiding (catch, log)
3546#endif
@@ -45,7 +56,17 @@ data Publish = Publish
4556
4657instance Binary Publish where
4758
48−testAgentBroadcast :: TestResult () -> Process ()
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 ()
4970testAgentBroadcast result = do
5071 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
5172
@@ -60,7 +81,7 @@ testAgentBroadcast result = do
6081 mxNotify ()
6182 -- Once the publisher has seen our message, it will broadcast the Publish
6283 -- and the consumer will see that and send the result to our typed channel.
63− stash result =<< receiveChan resultRP
84+ stash result =<< receiveChanTimeout 10000000 resultRP
6485
6586 kill publisher "finished"
6687 kill consumer "finished"
@@ -68,7 +89,7 @@ testAgentBroadcast result = do
6889testAgentDualInput :: TestResult (Maybe Int) -> Process ()
6990testAgentDualInput result = do
7091 (sp, rp) <- newChan
71− _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
92+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
7293 mxSink $ (\(i :: Int) -> do
7394 mxSetLocal . (+i) =<< mxGetLocal
7495 i' <- mxGetLocal
@@ -78,6 +99,8 @@ testAgentDualInput result = do
7899 else mxReady)
79100 ]
80101
102+ mRef <- monitor s
103+
81104 mxNotify (1 :: Int)
82105 nsend "sum-agent" (3 :: Int)
83106 mxNotify (2 :: Int)
@@ -85,6 +108,10 @@ testAgentDualInput result = do
85108 mxNotify (5 :: Int)
86109
87110 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)
88115
89116testAgentPrioritisation :: TestResult [String] -> Process ()
90117testAgentPrioritisation result = do
@@ -96,12 +123,12 @@ testAgentPrioritisation result = do
96123
97124 let name = "prioritising-agent"
98125 (sp, rp) <- newChan
99− void $ mxAgent (MxAgentId name) ["first"] [
100− mxSink (\(s :: String) -> do
101− mxUpdateLocal ((s:))
126+ s <- mxAgent (MxAgentId name) ["first"] [
127+ mxSink (\(s :: String) -> do
128+ mxUpdateLocal (s:)
102129 st <- mxGetLocal
103130 case length st of
104− n | n == 5 -> do liftMX $ sendChan sp st
131+ n | n == 5 -> do liftMX $ sendChan sp (sort st)
105132 mxDeactivate "finished"
106133 _ -> mxReceive -- go to the mailbox
107134 )
@@ -112,23 +139,33 @@ testAgentPrioritisation result = do
112139 mxNotify "fourth"
113140 nsend name "fifth"
114141
115− stash result . sort =<< receiveChan rp
142+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
143+ awaitExit s
116144
117145testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
118146testAgentMailboxHandling result = do
119147 (sp, rp) <- newChan
120− agent <- mxAgent (MxAgentId "listener-agent") () [
148+ agent <- mxAgent (MxAgentId "mailbox-agent") () [
121149 mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady
122150 ]
123151
124− nsend "listener-agent" ()
152+ nsend "mailbox-agent" ()
125153
126154 stash result =<< receiveChanTimeout 1000000 rp
127155 kill agent "finished"
128156
129157testAgentEventHandling :: TestResult Bool -> Process ()
130158testAgentEventHandling 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+
131167 let initState = [] :: [MxEvent]
168+ (rc, rs) <- newChan
132169 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
133170 (mxSink $ \ev -> do
134171 st <- mxGetLocal
@@ -137,7 +174,7 @@ testAgentEventHandling result = do
137174 (MxSpawned _) -> mxSetLocal (ev:st)
138175 (MxProcessDied _ _) -> mxSetLocal (ev:st)
139176 _ -> return ()
140− act >> mxReady),
177+ act >> (liftMX $ sendChan rc ()) >> mxReady),
141178 (mxSink $ \(ev, sp :: SendPort Bool) -> do
142179 st <- mxGetLocal
143180 let found =
@@ -157,102 +194,349 @@ testAgentEventHandling result = do
157194 mxReady)
158195 ]
159196
160− _ <- monitor agentPid
161− (sp, rp) <- newChan
162− pid <- spawnLocal $ sendChan sp ()
163− () <- receiveChan rp
197+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
198+ faff 2000000 rs
199+
200+ (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())
201+
202+ pid <- spawnLocal $ expect >>= sendChan sp
164203
165204 -- By waiting for a monitor notification, we have a
166− -- higher probably that the agent has seen
205+ -- higher probably that the agent has seen the spawn and died events
167206 monitor pid
168− receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
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 ()) ]
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
169217
170218 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
171219 mxNotify (MxSpawned pid, replyTo)
172220 mxNotify (MxProcessDied pid DiedNormal, replyTo)
173221
174− seenAlive <- receiveChan reply
175− seenDead <- receiveChan reply
176−
177− stash result $ seenAlive && seenDead
222+ seenAlive <- receiveChanTimeout 10000000 reply
223+ seenDead <- receiveChanTimeout 10000000 reply
178224
179−testMxMonitorEvents :: Process ()
180−testMxMonitorEvents = do
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
181232
233+testMxRegEvents :: TestResult () -> Process ()
234+testMxRegEvents result = do
182235 {- This test only deals with the local case, to ensure that we are being
183236 notified in the expected order - the remote cases related to the
184237 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+ ]
185254
186− let label = "testMxMonitorEvents"
187− let agentLabel = "listener-agent"
188− let delay = 1000000
189− (regChan, regSink) <- newChan
190− (unRegChan, unRegSink) <- newChan
191− agent <- mxAgent (MxAgentId agentLabel) () [
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+
325+ unreg1 <- receiveChanTimeout delay unRegSink
326+ unreg2 <- receiveChanTimeout delay unRegSink
327+
328+ let evts = [unreg1, unreg2]
329+ -- we can't rely on the order of the values in the node controller's
330+ -- map (it's either racy to do so, or no such guarantee exists for Data.Map),
331+ -- so we simply verify that we received the un-registration events we expect
332+ evts `shouldContain` (Just (label1, p1))
333+ evts `shouldContain` (Just (label2, p1))
334+
335+ kill agent "test-complete"
336+
337+ensure :: Process () -> Process () -> Process ()
338+ensure = flip finally
339+
340+testNSend :: (String -> () -> Process ())
341+ -> Maybe LocalNode
342+ -> Process ()
343+testNSend op n = do
344+ us <- getSelfPid
345+ let delay = 5000000
346+ let label = "testMxSend" ++ (show us)
347+ let isValid = isValidLabel n label
348+
349+ testMxSend n label $ \p1 sink -> do
350+ register label p1
351+ reg1 <- receiveChanTimeout delay sink
352+ case reg1 of
353+ Just (MxRegistered pd lb)
354+ | pd == p1 && isValid lb -> return ()
355+ _ -> die $ "Reg-Failed: " ++ show reg1
356+
357+ op label ()
358+
359+ sent <- receiveChanTimeout delay sink
360+ case sent of
361+ Just (MxSentToName lb by _)
362+ | by == us && isValid lb -> return True
363+ _ -> die $ "Send-Failed: " ++ show sent
364+
365+ where
366+ isValidLabel nd l1 l2
367+ | l2 == l2 = True
368+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)
369+ | otherwise = False
370+
371+testSend :: (ProcessId -> () -> Process ())
372+ -> Maybe LocalNode
373+ -> Process ()
374+testSend op n = do
375+ us <- getSelfPid
376+ let label = "testMxSend" ++ (show us)
377+ testMxSend n label $ \p1 sink -> do
378+ -- initiate a send
379+ op p1 ()
380+
381+ -- verify the management event
382+ sent <- receiveChanTimeout 5000000 sink
383+ case sent of
384+ Just (MxSent pidTo pidFrom _)
385+ | pidTo == p1 && pidFrom == us -> return True
386+ _ -> return False
387+
388+testChan :: (SendPort () -> () -> Process ())
389+ -> Maybe LocalNode
390+ -> Process ()
391+testChan op n = testMxSend n "" $ \p1 sink -> do
392+
393+ us <- getSelfPid
394+ send p1 us
395+
396+ cleared <- receiveChanTimeout 2000000 sink
397+ case cleared of
398+ Just (MxSent pidTo pidFrom _)
399+ | pidTo == p1 && pidFrom == us -> return ()
400+ _ -> die "Received uncleared Mx Event"
401+
402+ chan <- receiveTimeout 5000000 [ match return ]
403+ let ch' = fromMaybe (error "No reply chan received") chan
404+
405+ -- initiate a send
406+ op ch' ()
407+
408+ -- verify the management event
409+ sent <- receiveChanTimeout 5000000 sink
410+ case sent of
411+ Just (MxSentToPort sId spId _)
412+ | sId == us && spId == sendPortId ch' -> return True
413+ _ -> return False
414+
415+type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool
416+
417+testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()
418+testMxSend mNode label test = do
419+ us <- getSelfPid
420+ (sp, rp) <- newChan
421+ (chan, sink) <- newChan
422+ agent <- mxAgent (MxAgentId $ agentLabel us) () [
192423 mxSink $ \ev -> do
193424 case ev of
194− MxRegistered pid label
195− | label /= agentLabel -> liftMX $ sendChan regChan (label, pid)
196− MxUnRegistered pid label
197− | label /= agentLabel -> liftMX $ sendChan unRegChan (label, pid)
198− _ -> return ()
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 ()
199435 mxReady
200436 ]
201437
202− p1 <- spawnLocal expect
203− p2 <- spawnLocal expect
438+ case mNode of
439+ Nothing -> void $ spawnLocal (proc sp)
440+ Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
204441
205− register label p1
206− reg1 <- receiveChanTimeout delay regSink
207− reg1 `shouldBe` equalTo (Just (label, p1))
442+ p1 <- receiveChanTimeout 2000000 rp
443+ unless (isJust p1) $ die "Timed out waiting for ProcessId"
444+ res <- try $ test (fromJust p1) sink
208445
209− unregister label
210− unreg1 <- receiveChanTimeout delay unRegSink
211− unreg1 `shouldBe` equalTo (Just (label, p1))
446+ kill agent "bye"
447+ kill (fromJust p1) "bye"
212448
213− register label p2
214− reg2 <- receiveChanTimeout delay regSink
215− reg2 `shouldBe` equalTo (Just (label, p2))
449+ case res of
450+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
451+ Right tr -> tr `shouldBe` equalTo True
216452
217− reregister label p1
218− unreg2 <- receiveChanTimeout delay unRegSink
219− unreg2 `shouldBe` equalTo (Just (label, p2))
220453
221− reg3 <- receiveChanTimeout delay regSink
222− reg3 `shouldBe` equalTo (Just (label, p1))
454+ where
455+ agentLabel s = "mx-unsafe-check-agent-" ++ show s
456+ proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
223457
224− mapM_ (flip kill $ "test-complete") [agent, p1, p2]
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
225466
226467tests :: TestTransport -> IO [Test]
227468tests TestTransport{..} = do
228469 node1 <- newLocalNode testTransport initRemoteTable
470+ node2 <- newLocalNode testTransport initRemoteTable
229471 return [
230− testGroup "Mx Agents" [
231− testCase "Event Handling"
472+ testGroup "MxAgents" [
473+ testCase "EventHandling"
232474 (delayedAssertion
233475 "expected True, but events where not as expected"
234476 node1 True testAgentEventHandling)
235− , testCase "Inter-Agent Broadcast"
477+ , testCase "InterAgentBroadcast"
236478 (delayedAssertion
237479 "expected (), but no broadcast was received"
238− node1 () testAgentBroadcast)
239− , testCase "Agent Mailbox Handling"
480+ node1 (Just ()) testAgentBroadcast)
481+ , testCase "AgentMailboxHandling"
240482 (delayedAssertion
241483 "expected (Just ()), but no regular (mailbox) input was handled"
242484 node1 (Just ()) testAgentMailboxHandling)
243− , testCase "Agent Dual Input Handling"
485+ , testCase "AgentDualInputHandling"
244486 (delayedAssertion
245487 "expected sum = 15, but the result was Nothing"
246488 node1 (Just 15 :: Maybe Int) testAgentDualInput)
247− , testCase "Agent Input Prioritisation"
489+ , testCase "AgentInputPrioritisation"
248490 (delayedAssertion
249491 "expected [first, second, third, fourth, fifth], but result diverged"
250492 node1 (sort ["first", "second",
251493 "third", "fourth",
252494 "fifth"]) testAgentPrioritisation)
253495 ]
254− , testGroup "Mx Events" [
255− testCase "Monitor Events"
256− (runProcess node1 testMxMonitorEvents)
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
257506 ]
258507 ]
508+ where
509+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
510+ ("NSend", [
511+ ("nsend", testNSend nsend)
512+ , ("Unsafe.nsend", testNSend Unsafe.nsend)
513+ , ("nsendRemote", testNSend (nsendRemote nid))
514+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
515+ ])
516+ , ("Send", [
517+ ("send", testSend send)
518+ , ("Unsafe.send", testSend Unsafe.send)
519+ , ("usend", testSend usend)
520+ , ("Unsafe.usend", testSend Unsafe.usend)
521+ , ("sendChan", testChan sendChan)
522+ , ("Unsafe.sendChan", testChan Unsafe.sendChan)
523+ ])
524+ , ("Forward", [
525+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
526+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))
527+ ])
528+ ]
529+
530+ build :: LocalNode
531+ -> LocalNode
532+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
533+ -> [Test]
534+ build n ln specs =
535+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [
536+ testCase (intercalate "-" [caseName, caseSuffix])
537+ (runProcess n (caseImpl caseNode))
538+ | (caseName, caseImpl) <- groupCases
539+ ]
540+ | (groupName, groupCases) <- specs
541+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]
542+ ]
distributed-process-tests/src/Control/Distributed/Process/Tests/Receive.hs+7−8
@@ -10,13 +10,12 @@ import Network.Transport.Test (TestTransport(..))
1010
1111import Network.Transport (Transport)
1212import Control.Distributed.Process
13−import Control.Distributed.Process.Closure
1413import Control.Distributed.Process.Node
1514
1615import Control.Monad
1716
1817import Test.HUnit (Assertion, (@?=))
19−import Test.Framework (Test, defaultMain)
18+import Test.Framework (Test)
2019import Test.Framework.Providers.HUnit (testCase)
2120
2221-- Tests:
@@ -43,7 +42,7 @@ recTest2 :: ReceivePort ()
4342 -> SendPort String
4443 -> ReceivePort String -> ReceivePort String
4544 -> Process ()
46−recTest2 wait sync r1 r2 = do
45+recTest2 wait sync r1 _ = do
4746 forever $ do
4847 receiveChan wait
4948 r <- receiveWait
@@ -56,7 +55,7 @@ recTest3 :: ReceivePort ()
5655 -> SendPort String
5756 -> ReceivePort String -> ReceivePort String
5857 -> Process ()
59−recTest3 wait sync r1 r2 = do
58+recTest3 wait sync r1 _ = do
6059 forever $ do
6160 receiveChan wait
6261 r <- receiveWait
@@ -69,7 +68,7 @@ recTest4 :: ReceivePort ()
6968 -> SendPort String
7069 -> ReceivePort String -> ReceivePort String
7170 -> Process ()
72−recTest4 wait sync r1 r2 = do
71+recTest4 wait sync r1 _ = do
7372 forever $ do
7473 receiveChan wait
7574 r <- receiveWait
@@ -83,11 +82,11 @@ master :: Process ()
8382master = do
8483 (waits,waitr) <- newChan
8584 (syncs,syncr) <- newChan
86− let go expect = do
85+ let go expected = do
8786 sendChan waits ()
8887 r <- receiveChan syncr
89− liftIO $ print (r,expect, r == expect)
90− liftIO $ r @?= expect
88+ liftIO $ print (r, expected, r == expected)
89+ liftIO $ r @?= expected
9190
9291 liftIO $ putStrLn "---- Test 1 ----"
9392 (s1,r1) <- newChan
distributed-process-tests/src/Control/Distributed/Process/Tests/Stats.hs+11−7
@@ -29,7 +29,7 @@ import Test.Framework.Providers.HUnit (testCase)
2929
3030testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()
3131testLocalDeadProcessInfo result = do
32− pid <- spawnLocal $ do "finish" <- expect; return ()
32+ pid <- spawnLocal $ do (_ :: String) <- expect; return ()
3333 mref <- monitor pid
3434 send pid "finish"
3535 _ <- receiveWait [
@@ -52,8 +52,10 @@ testLocalLiveProcessInfo result = do
5252 link self
5353 mRef <- monitor self
5454 stash mon mRef
55− "die" <- expect
56− return ()
55+ res <- expect
56+ case res of
57+ "die" -> return ()
58+ _ -> die $ "unexpected message received: " ++ res
5759
5860 monRef <- liftIO $ takeMVar mon
5961
@@ -84,9 +86,9 @@ testRemoteLiveProcessInfo TestTransport{..} node1 = do
8486 -- our send op shouldn't overtake link or monitor requests AFAICT
8587 -- so a little table tennis should get us synchronised properly
8688 send serverPid (self, "ping")
87− "pong" <- expect
89+ pong <- expect
8890 pInfo <- getProcessInfo serverPid
89− stash result $ pInfo /= Nothing
91+ stash result $ pong == "pong" && pInfo /= Nothing
9092 where
9193 launchRemote :: MVar ProcessId -> IO ()
9294 launchRemote locMV = do
@@ -97,8 +99,10 @@ testRemoteLiveProcessInfo TestTransport{..} node1 = do
9799 _ <- receiveWait [
98100 match (\(pid, "ping") -> send pid "pong")
99101 ]
100− "stop" <- expect
101− return ()
102+ res <- expect
103+ case res of
104+ "stop" -> return ()
105+ _ -> die $ "unexpected message received: " ++ res
102106 return ()
103107
104108 withActiveRemote :: LocalNode
distributed-process-tests/src/Control/Distributed/Process/Tests/Tracing.hs+23−11
@@ -125,9 +125,11 @@ testTraceSending result = do
125125 (\ev ->
126126 case ev of
127127 (MxSent to from msg) -> do
128− (Just s) <- unwrapMessage msg :: Process (Maybe String)
129− stash res (to == pid && from == self && s == "hello there")
130− stash res (to == pid && from == self)
128+ mS <- unwrapMessage msg :: Process (Maybe String)
129+ case mS of
130+ (Just s) -> do stash res (to == pid && from == self && s == "hello there")
131+ stash res (to == pid && from == self)
132+ _ -> die "failed state invariant, message type unmatched..."
131133 _ ->
132134 return ()) $ do
133135 send pid "hello there"
@@ -246,12 +248,8 @@ testRemoteTraceRelay TestTransport{..} result =
246248 -- Here we set up that relay, and then wait for a signal
247249 -- that the tracer (on node1) has seen the expected
248250 -- MxSpawned message, at which point we're finished
249− (Just log') <- whereis "logger"
250− pid <- liftIO $ forkProcess node2 $ do
251− logRelay <- spawnLocal $ relay log'
252− reregister "logger" logRelay
253− getSelfNode >>= stash mvNid >> (expect :: Process ())
254251
252+ pid <- splinchLogger node2 mvNid
255253 nid <- liftIO $ takeMVar mvNid
256254 mref <- monitor pid
257255 observedPid <- liftIO $ newEmptyMVar
@@ -291,6 +289,17 @@ testRemoteTraceRelay TestTransport{..} result =
291289 -- and just to be polite...
292290 liftIO $ closeLocalNode node2
293291
292+ where
293+ splinchLogger n2 mv = do
294+ mLog <- whereis "logger"
295+ case mLog of
296+ Nothing -> die "no logger registered"
297+ Just log' -> do liftIO $ forkProcess n2 $ do
298+ logRelay <- spawnLocal $ relay log'
299+ reregister "logger" logRelay
300+ getSelfNode >>= stash mv >> (expect :: Process ())
301+
302+
294303-- | Sets the value of an environment variable while executing the given IO
295304-- computation and restores the preceeding value upon completion.
296305withEnv :: String -> String -> IO a -> IO a
@@ -319,9 +328,12 @@ testSystemLoggerMsg t action interestingMessage =
319328 -- Wait for the trace message.
320329 receiveWait [ matchIf interestingMessage' $ const $ return () ]
321330 -- Only one interesting message should arrive.
322− Nothing <- receiveTimeout 100000
323− [ matchIf interestingMessage' $ const $ return () ]
324− return ()
331+ expectedTimeout <- receiveTimeout
332+ 100000
333+ [ matchIf interestingMessage' $ const $ return () ]
334+ case expectedTimeout of
335+ Nothing -> return ()
336+ Just _ -> die "Unexpected message arrived..."
325337
326338
327339-- | Tests that one and only one trace message is produced when a message is
distributed-process-tests/tests/runInMemory.hs−1
@@ -8,7 +8,6 @@ import Network.Transport.Test (TestTransport(..))
88import Network.Transport.InMemory
99import Test.Framework (defaultMainWithArgs)
1010
11−import Control.Concurrent (threadDelay)
1211import System.Environment (getArgs)
1312
1413main :: IO ()
distributed-process-tests/tests/runTCP.hs+2−2
@@ -10,6 +10,7 @@ import Network.Transport.TCP
1010 ( createTransportExposeInternals
1111 , TransportInternals(socketBetween)
1212 , defaultTCPParameters
13+ , defaultTCPAddr
1314 , TCPParameters(..)
1415 )
1516import Test.Framework (defaultMainWithArgs)
@@ -24,8 +25,7 @@ main = do
2425 hSetBuffering stdout LineBuffering
2526 hSetBuffering stderr LineBuffering
2627 Right (transport, internals) <-
27− createTransportExposeInternals "127.0.0.1" "8080"
28− (\sn -> ("127.0.0.1", sn))
28+ createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "8080")
2929 defaultTCPParameters { transportConnectTimeout = Just 3000000 }
3030 ts <- tests TestTransport
3131 { testTransport = transport
distributed-process.cabal+2−2
@@ -6,7 +6,7 @@ License: BSD3
66License-File: LICENSE
77Copyright: Well-Typed LLP, Tweag I/O Limited
88Author: Duncan Coutts, Nicolas Wu, Edsko de Vries
9−Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>
9+Maintainer: Tim Watson <watson.timothy@gmail.com>
1010Stability: experimental
1111Homepage: http://haskell-distributed.github.com/
1212Bug-Reports: https://github.com/haskell-distributed/distributed-process/issues
@@ -21,7 +21,7 @@ Description: This is an implementation of Cloud Haskell, as described in
2121
2222 You will probably also want to install a Cloud Haskell backend such
2323 as distributed-process-simplelocalnet.
24−Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1
24+Tested-With: GHC==7.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4
2525Category: Control
2626extra-source-files: ChangeLog
2727
src/Control/Distributed/Process/Internal/Primitives.hs+59−43
@@ -254,17 +254,15 @@ send them msg = do
254254 node = processNode proc
255255 nodeId = localNodeId node
256256 destNode = (processNodeId them) in do
257− case destNode == nodeId of
258− True -> sendLocal them msg
259− False -> liftIO $ sendMessage (processNode proc)
260− (ProcessIdentifier (processId proc))
261− (ProcessIdentifier them)
262− NoImplicitReconnect
263− msg
264− -- We do not fire the trace event until after the sending is complete;
265− -- In the remote case, 'sendMessage' can block in the networking stack.
266257 liftIO $ traceEvent (localEventBus node)
267258 (MxSent them us (createUnencodedMessage msg))
259+ if destNode == nodeId
260+ then sendLocal them msg
261+ else liftIO $ sendMessage (processNode proc)
262+ (ProcessIdentifier (processId proc))
263+ (ProcessIdentifier them)
264+ NoImplicitReconnect
265+ msg
268266
269267-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
270268-- and (in the case when the destination process resides on the same local
@@ -284,9 +282,12 @@ unsafeSend = Unsafe.send
284282--
285283usend :: Serializable a => ProcessId -> a -> Process ()
286284usend them msg = do
287− here <- getSelfNode
285+ proc <- ask
288286 let there = processNodeId them
289− if here == there
287+ let (us, node) = (processId proc, processNode proc)
288+ let msg' = wrapMessage msg
289+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg')
290+ if localNodeId (processNode proc) == there
290291 then sendLocal them msg
291292 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
292293 (createMessage msg)
@@ -306,7 +307,13 @@ expect = receiveWait [match return]
306307-- Channels --
307308--------------------------------------------------------------------------------
308309
309−-- | Create a new typed channel
310+-- | Create a new typed channel, bound to the calling Process
311+--
312+-- Note that the channel is bound to the lifecycle of the process that evaluates
313+-- this function, such that when it dies/exits, the channel will no longer
314+-- function, but will remain accessible. Thus reading from the ReceivePort will
315+-- fail silently thereafter, blocking indefinitely (unless a timeout is used).
316+--
310317newChan :: Serializable a => Process (SendPort a, ReceivePort a)
311318newChan = do
312319 proc <- ask
@@ -334,13 +341,16 @@ newChan = do
334341sendChan :: Serializable a => SendPort a -> a -> Process ()
335342sendChan (SendPort cid) msg = do
336343 proc <- ask
337− let node = localNodeId (processNode proc)
338− destNode = processNodeId (sendPortProcessId cid) in do
339− case destNode == node of
344+ let node = processNode proc
345+ pid = processId proc
346+ us = localNodeId node
347+ them = processNodeId (sendPortProcessId cid) in do
348+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid $ wrapMessage msg)
349+ case them == us of
340350 True -> sendChanLocal cid msg
341351 False -> do
342− liftIO $ sendBinary (processNode proc)
343− (ProcessIdentifier (processId proc))
352+ liftIO $ sendBinary node
353+ (ProcessIdentifier pid)
344354 (SendPortIdentifier cid)
345355 NoImplicitReconnect
346356 msg
@@ -401,8 +411,11 @@ newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
401411receiveWait :: [Match b] -> Process b
402412receiveWait ms = do
403413 queue <- processQueue <$> ask
404− Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
405− proc
414+ mProc <- liftIO $ dequeue queue Blocking (map unMatch ms)
415+ case mProc of
416+ Just proc' -> proc'
417+ Nothing -> die $ "System Invariant Violation: CQueue.hs returned `Nothing` "
418+ ++ "in the absence of a timeout value."
406419
407420-- | Like 'receiveWait' but with a timeout.
408421--
@@ -481,17 +494,14 @@ forward msg them = do
481494 us = processId proc
482495 nid = localNodeId node
483496 destNode = (processNodeId them) in do
484− case destNode == nid of
485− True -> sendCtrlMsg Nothing (LocalSend them msg)
486− False -> liftIO $ sendPayload (processNode proc)
487− (ProcessIdentifier (processId proc))
488− (ProcessIdentifier them)
489− NoImplicitReconnect
490− (messageToPayload msg)
491− -- We do not fire the trace event until after the sending is complete;
492− -- In the remote case, 'sendMessage' can block in the networking stack.
493− liftIO $ traceEvent (localEventBus node)
494− (MxSent them us msg)
497+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
498+ if destNode == nid
499+ then sendCtrlMsg Nothing (LocalSend them msg)
500+ else liftIO $ sendPayload (processNode proc)
501+ (ProcessIdentifier (processId proc))
502+ (ProcessIdentifier them)
503+ NoImplicitReconnect
504+ (messageToPayload msg)
495505
496506-- | Forward a raw 'Message' to the given 'ProcessId'.
497507--
@@ -505,14 +515,10 @@ uforward msg them = do
505515 us = processId proc
506516 nid = localNodeId node
507517 destNode = (processNodeId them) in do
508− case destNode == nid of
509− True -> sendCtrlMsg Nothing (LocalSend them msg)
510− False -> sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them)
511− msg
512− -- We do not fire the trace event until after the sending is complete;
513− -- In the remote case, 'sendCtrlMsg' can block in the networking stack.
514− liftIO $ traceEvent (localEventBus node)
515− (MxSent them us msg)
518+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg)
519+ if destNode == nid
520+ then sendCtrlMsg Nothing (LocalSend them msg)
521+ else sendCtrlMsg (Just destNode) $ UnreliableSend (processLocalId them) msg
516522
517523-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
518524-- 'Serializable' - like the datum they contain - but also note, deserialising
@@ -1221,8 +1227,12 @@ whereisRemoteAsync nid label = do
12211227
12221228-- | Named send to a process in the local registry (asynchronous)
12231229nsend :: Serializable a => String -> a -> Process ()
1224−nsend label msg =
1225− sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
1230+nsend label msg = do
1231+ proc <- ask
1232+ let msg' = createUnencodedMessage msg
1233+ liftIO $ traceEvent (localEventBus (processNode proc))
1234+ (MxSentToName label (processId proc) msg')
1235+ sendCtrlMsg Nothing (NamedSend label msg')
12261236
12271237-- | Named send to a process in the local registry (asynchronous).
12281238-- This function makes /no/ attempt to serialize and (in the case when the
@@ -1234,9 +1244,15 @@ unsafeNSend = Unsafe.nsend
12341244-- | Named send to a process in a remote registry (asynchronous)
12351245nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
12361246nsendRemote nid label msg = do
1237− here <- getSelfNode
1238− if here == nid then nsend label msg
1239− else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
1247+ proc <- ask
1248+ let us = processId proc
1249+ let node = processNode proc
1250+ if localNodeId node == nid
1251+ then nsend label msg
1252+ else let lbl = label ++ "@" ++ show nid in do
1253+ liftIO $ traceEvent (localEventBus node)
1254+ (MxSentToName lbl us (wrapMessage msg))
1255+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
12401256
12411257-- | Named send to a process in a remote registry (asynchronous)
12421258-- This function makes /no/ attempt to serialize and (in the case when the
src/Control/Distributed/Process/Management.hs+34−2
@@ -76,6 +76,40 @@
7676--
7777-- * Whether messages will be taken from the mailbox first, or the event bus.
7878--
79+-- Since the event bus uses STM broadcast channels to communicate with agents,
80+-- no message written to the bus successfully can be lost.
81+--
82+-- Agents can also receive messages via their mailboxes - these are subject to
83+-- the same guarantees as all inter-process message sending.
84+--
85+-- Messages dispatched on an STM broadcast channel (i.e., management event bus)
86+-- are guaranteed to be delivered with the same FIFO ordering guarantees that
87+-- exist between two communicating processes, such that communication from the
88+-- node controller's threads (i.e., MxEvent's) will never be re-ordered, but
89+-- messages dispatched to the event bus by other processes (including, but not
90+-- limited to agents) are only guaranteed to be ordered between one sender and
91+-- one receiver.
92+--
93+-- No guarantee exists for the ordering in which messages sent to an agent's
94+-- mailbox will be delivered, vs messages dispatched via the event bus.
95+--
96+-- Because of the above, there are no ordering guarantees for messages sent
97+-- between agents, or for processes to agents, except for those that apply to
98+-- messages sent between regular processes, since agents are
99+-- implemented as such.
100+--
101+-- The event bus is serial and single threaded. Anything that is published by
102+-- the node controller will be seen in FIFO order. There are no ordering
103+-- guarantees pertaining to entries published to the event bus by other
104+-- processes or agents.
105+--
106+-- It should not be possible to see, for example, an @MxReceived@ before the
107+-- corresponding @MxSent@ event, since the places where we issue the @MxSent@
108+-- write directly to the event bus (using STM) in the calling (green) thread,
109+-- before dispatching instructions to the node controller to perform the
110+-- necessary routing to deliver the message to a process (or registered name,
111+-- or typed channel) locally or remotely.
112+--
79113-- [Management Data API]
80114--
81115-- Both management agents and clients of the API have access to a variety of
@@ -144,8 +178,6 @@
144178-- > monitorNames = getSelfPid >>= nsend "name-monitor"
145179-- > monitorNames2 = getSelfPid >>= mxNotify
146180--
147−-- For some real-world examples, see the distributed-process-platform package.
148−--
149181-- [Performance, Stablity and Scalability]
150182--
151183-- /Management Agents/ offer numerous advantages over regular processes:
src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs+8−2
@@ -33,6 +33,7 @@ import Control.Distributed.Process.Internal.Primitives
3333 ( whereis
3434 , newChan
3535 , receiveChan
36+ , die
3637 )
3738import Control.Distributed.Process.Management.Internal.Trace.Types
3839 ( TraceArg(..)
@@ -168,6 +169,11 @@ withRegisteredTracer act = do
168169 withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp
169170 currentTracer <- receiveChan rp
170171 case currentTracer of
171− Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' }
172+ Nothing -> do mTP <- whereis "tracer.initial"
173+ -- NB: this should NOT ever happen, but forcing pattern matches
174+ -- is not considered cool in later versions of MonadFail
175+ case mTP of
176+ Just p' -> act p'
177+ Nothing -> die $ "System Invariant Violation: Tracer Process "
178+ ++ "Name Not Found (whereis tracer.initial)"
172179 (Just p) -> act p
173−
src/Control/Distributed/Process/Management/Internal/Types.hs+7
@@ -21,6 +21,7 @@ import Control.Concurrent.STM
2121import Control.Distributed.Process.Internal.Types
2222 ( Process
2323 , ProcessId
24+ , SendPortId
2425 , Message
2526 , DiedReason
2627 , NodeId
@@ -57,8 +58,14 @@ data MxEvent =
5758 -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)
5859 | MxSent ProcessId ProcessId Message
5960 -- ^ fired whenever a message is sent from a local process
61+ | MxSentToName String ProcessId Message
62+ -- ^ fired whenever a named send occurs
63+ | MxSentToPort ProcessId SendPortId Message
64+ -- ^ fired whenever a sendChan occurs
6065 | MxReceived ProcessId Message
6166 -- ^ fired whenever a message is received by a local process
67+ | MxReceivedPort SendPortId Message
68+ -- ^ fired whenever a message is received via a typed channel
6269 | MxConnected ConnectionId EndPointAddress
6370 -- ^ fired when a network-transport connection is first established
6471 | MxDisconnected ConnectionId EndPointAddress
src/Control/Distributed/Process/Node.hs+26−69
@@ -31,10 +31,9 @@ import qualified Data.Map as Map
3131 , delete
3232 , toList
3333 , fromList
34− , filter
3534 , insert
3635 , lookup
37− , insertLookupWithKey
36+ , partition
3837 , partitionWithKey
3938 , elems
4039 , size
@@ -187,16 +186,8 @@ import Control.Distributed.Process.Internal.Types
187186import Control.Distributed.Process.Management.Internal.Agent
188187 ( mxAgentController
189188 )
190−import Control.Distributed.Process.Management
191− ( mxAgent
192− , mxSink
193− , liftMX
194− , mxUpdateLocal
195− , mxGetLocal
196− , mxSetLocal
197− , mxReady
198− , MxEvent(..)
199− , MxAgentId(..)
189+import Control.Distributed.Process.Management.Internal.Types
190+ ( MxEvent(..)
200191 )
201192import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
202193 ( remoteTable
@@ -325,49 +316,6 @@ startDefaultTracer node' = do
325316
326317-- TODO: we need a better mechanism for defining and registering services
327318
328−registryMonitorAgentId :: MxAgentId
329−registryMonitorAgentId = MxAgentId "service.registry.monitoring"
330−
331−-- note [registry monitoring agent]
332−-- this agent listens for 'MxRegistered' and 'MxUnRegistered' events and tracks
333−-- all remote 'ProcessId's that are stored in the registry.
334−--
335−-- When a remote process is registered, the agent starts monitoring it until it
336−-- is unregistered or the monitor notification arrives.
337−--
338−-- The agent keeps the amount of labels associated with each registered remote
339−-- process. This is necessary so the process is unmonitored only when it has
340−-- been unregistered from all of the labels.
341−--
342−
343−registryMonitorAgent :: Process ProcessId
344−registryMonitorAgent = do
345− nid <- getSelfNode
346− -- For each process the map associates the 'MonitorRef' used to monitor it and
347− -- the amount of labels associated with it.
348− mxAgent registryMonitorAgentId (Map.empty :: Map String MonitorRef)
349− [ mxSink $ \(ProcessMonitorNotification mref _ _) -> do
350− mxUpdateLocal $ Map.filter (== mref)
351− mxReady
352− , mxSink $ \ev -> do
353− case ev of
354− MxRegistered pid label
355− | processNodeId pid /= nid -> do
356− mref <- liftMX $ monitor pid
357− mxUpdateLocal (Map.insert label mref)
358− MxUnRegistered pid label
359− | processNodeId pid /= nid -> do
360− mrefs <- mxGetLocal
361− forM_ (label `Map.lookup` mrefs) $ \mref -> do
362− liftMX $ unmonitorAsync mref
363− mxUpdateLocal (Map.delete label)
364− _ -> return ()
365− mxReady
366− -- remove async answers from mailbox
367− , mxSink $ \RegisterReply{} -> mxReady
368− , mxSink $ \DidUnmonitor{} -> mxReady
369− ]
370−
371319-- | Start and register the service processes on a node
372320startServiceProcesses :: LocalNode -> IO ()
373321startServiceProcesses node = do
@@ -544,7 +492,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
544492data IncomingTarget =
545493 Uninit
546494 | ToProc ProcessId (Weak (CQueue Message))
547− | ToChan TypedChannel
495+ | ToChan SendPortId TypedChannel
548496 | ToNode
549497
550498data ConnectionState = ConnectionState {
@@ -599,13 +547,17 @@ handleIncomingMessages node = go initConnectionState
599547 enqueue queue msg -- 'enqueue' is strict
600548 trace node (MxReceived pid msg)
601549 go st
602− Just (_, ToChan (TypedChannel chan')) -> do
550+ Just (_, ToChan chId (TypedChannel chan')) -> do
603551 mChan <- deRefWeak chan'
604552 -- If mChan is Nothing, the process has given up the read end of
605553 -- the channel and we simply ignore the incoming message
606− forM_ mChan $ \chan -> atomically $
607− -- We make sure the message is fully decoded when it is enqueued
608− writeTQueue chan $! decode (BSL.fromChunks payload)
554+ forM_ mChan $ \chan -> do
555+ msg' <- atomically $ do
556+ msg <- return $! decode (BSL.fromChunks payload)
557+ -- We make sure the message is fully decoded when it is enqueued
558+ writeTQueue chan msg
559+ return msg
560+ trace node $ MxReceivedPort chId $ unsafeCreateUnencodedMessage msg'
609561 go st
610562 Just (_, ToNode) -> do
611563 let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -632,7 +584,7 @@ handleIncomingMessages node = go initConnectionState
632584 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
633585 case mChannel of
634586 Just channel ->
635− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
587+ go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
636588 Nothing ->
637589 invalidRequest cid st $
638590 "incoming attempt to connect to unknown channel of"
@@ -961,7 +913,11 @@ ncEffectDied ident reason = do
961913
962914 modify' $ (links ^= unaffectedLinks') . (monitors ^= unaffectedMons')
963915
964− modify' $ registeredHere ^: Map.filter (\pid -> not $ ident `impliesDeathOf` ProcessIdentifier pid)
916+ -- we now consider all labels for this identifier unregistered
917+ let toDrop pid = not $ ident `impliesDeathOf` ProcessIdentifier pid
918+ (keepNames, dropNames) <- Map.partition toDrop <$> gets (^. registeredHere)
919+ mapM_ (\(p, l) -> liftIO $ trace node (MxUnRegistered l p)) (Map.toList dropNames)
920+ modify' $ registeredHere ^= keepNames
965921
966922 remaining <- fmap Map.toList (gets (^. registeredOnNodes)) >>=
967923 mapM (\(pid,nidlist) ->
@@ -1098,11 +1054,12 @@ ncEffectLocalPortSend from msg = do
10981054 -- If ch is Nothing, the process has given up the read end of
10991055 -- the channel and we simply ignore the incoming message - this
11001056 ch <- deRefWeak chan'
1101− forM_ ch $ \chan -> deliverChan msg chan
1102− where deliverChan :: forall a . Message -> TQueue a -> IO ()
1103− deliverChan (UnencodedMessage _ raw) chan' =
1057+ forM_ ch $ \chan -> deliverChan node from msg chan
1058+ where deliverChan :: forall a . LocalNode -> SendPortId -> Message -> TQueue a -> IO ()
1059+ deliverChan n p (UnencodedMessage _ raw) chan' = do
11041060 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1105− deliverChan (EncodedMessage _ _) _ =
1061+ trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
1062+ deliverChan _ _ (EncodedMessage _ _) _ =
11061063 -- this will not happen unless someone screws with Primitives.hs
11071064 error "invalid local channel delivery"
11081065
@@ -1132,7 +1089,7 @@ ncEffectGetInfo from pid =
11321089 case mProc of
11331090 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
11341091 from (ProcessInfoNone DiedUnknownId)
1135− Just proc -> do
1092+ Just proc -> do
11361093 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
11371094 gets (^. links)
11381095 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
@@ -1146,8 +1103,8 @@ ncEffectGetInfo from pid =
11461103 infoNode = (processNodeId pid)
11471104 , infoRegisteredNames = reg
11481105 , infoMessageQueueLength = size
1149− , infoMonitors = Set.toList itsMons
1150− , infoLinks = Set.toList itsLinks
1106+ , infoMonitors = Set.toList itsMons
1107+ , infoLinks = Set.toList itsLinks
11511108 }
11521109 where dispatch :: (Serializable a)
11531110 => Bool
src/Control/Distributed/Process/UnsafePrimitives.hs+82−33
@@ -32,7 +32,24 @@
3232-- the /normal/ strategy).
3333--
3434-- Use of the functions in this module can potentially change the runtime
35−-- behaviour of your application. You have been warned!
35+-- behaviour of your application. In addition, messages passed between Cloud
36+-- Haskell processes are written to a tracing infrastructure on the local node,
37+-- to provide improved introspection and debugging facilities for complex actor
38+-- based systems. This module makes no attempt to force evaluation in these
39+-- cases either, thus evaluation problems in passed data structures could not
40+-- only crash your processes, but could also bring down critical internal
41+-- services on which the node relies to function correctly.
42+--
43+-- If you wish to repudiate such issues, you are advised to consider the use
44+-- of NFSerialisable in the distributed-process-extras package, which type
45+-- class brings NFData into scope along with Serializable, such that we can
46+-- force evaluation. Intended for use with modules such as this one, this
47+-- approach guarantees correct evaluatedness in terms of @NFData@. Please note
48+-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the
49+-- same way as a @Binary@ one with regards evaluation, so it is still possible
50+-- to introduce unexpected behaviour by using /unsafe/ primitives in this way.
51+--
52+-- You have been warned!
3653--
3754-- This module is exported so that you can replace the use of Cloud Haskell's
3855-- /safe/ messaging primitives. If you want to use both variants, then you can
@@ -55,7 +72,12 @@ import Control.Distributed.Process.Internal.Messaging
5572 , sendBinary
5673 , sendCtrlMsg
5774 )
58−
75+import Control.Distributed.Process.Management.Internal.Types
76+ ( MxEvent(..)
77+ )
78+import Control.Distributed.Process.Management.Internal.Trace.Types
79+ ( traceEvent
80+ )
5981import Control.Distributed.Process.Internal.Types
6082 ( ProcessId(..)
6183 , NodeId(..)
@@ -79,34 +101,48 @@ import Control.Monad.Reader (ask)
79101
80102-- | Named send to a process in the local registry (asynchronous)
81103nsend :: Serializable a => String -> a -> Process ()
82−nsend label msg =
83− sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))
104+nsend label msg = do
105+ proc <- ask
106+ let us = processId proc
107+ let msg' = wrapMessage msg
108+ -- see [note: tracing]
109+ liftIO $ traceEvent (localEventBus (processNode proc))
110+ (MxSentToName label us msg')
111+ sendCtrlMsg Nothing (NamedSend label msg')
84112
85113-- | Named send to a process in a remote registry (asynchronous)
86114nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
87115nsendRemote nid label msg = do
88116 proc <- ask
89− if localNodeId (processNode proc) == nid
117+ let us = processId proc
118+ let node = processNode proc
119+ if localNodeId node == nid
90120 then nsend label msg
91− else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
121+ else
122+ let lbl = label ++ "@" ++ show nid in do
123+ -- see [note: tracing] NB: this is a remote call to another NC...
124+ liftIO $ traceEvent (localEventBus node)
125+ (MxSentToName lbl us (wrapMessage msg))
126+ sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
92127
93128-- | Send a message
94129send :: Serializable a => ProcessId -> a -> Process ()
95130send them msg = do
96131 proc <- ask
97132 let node = localNodeId (processNode proc)
98− destNode = (processNodeId them) in do
99− case destNode == node of
100− True -> unsafeSendLocal them msg
101− False -> liftIO $ sendMessage (processNode proc)
102− (ProcessIdentifier (processId proc))
103− (ProcessIdentifier them)
104− NoImplicitReconnect
105− msg
106− where
107− unsafeSendLocal :: (Serializable a) => ProcessId -> a -> Process ()
108− unsafeSendLocal pid msg' =
109− sendCtrlMsg Nothing $ LocalSend pid (unsafeCreateUnencodedMessage msg')
133+ destNode = (processNodeId them)
134+ us = (processId proc)
135+ msg' = wrapMessage msg in do
136+ -- see [note: tracing]
137+ liftIO $ traceEvent (localEventBus (processNode proc))
138+ (MxSent them us msg')
139+ if destNode == node
140+ then sendCtrlMsg Nothing $ LocalSend them msg'
141+ else liftIO $ sendMessage (processNode proc)
142+ (ProcessIdentifier (processId proc))
143+ (ProcessIdentifier them)
144+ NoImplicitReconnect
145+ msg
110146
111147-- | Send a message unreliably.
112148--
@@ -121,30 +157,43 @@ usend :: Serializable a => ProcessId -> a -> Process ()
121157usend them msg = do
122158 proc <- ask
123159 let there = processNodeId them
160+ let (us, node) = (processId proc, processNode proc)
161+ let msg' = wrapMessage msg
162+ -- see [note: tracing]
163+ liftIO $ traceEvent (localEventBus node) (MxSent them us msg')
124164 if localNodeId (processNode proc) == there
125− then sendCtrlMsg Nothing $
126− LocalSend them (unsafeCreateUnencodedMessage msg)
165+ then sendCtrlMsg Nothing $ LocalSend them msg'
127166 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
128167 (createMessage msg)
129168
169+-- [note: tracing]
170+-- Note that tracing writes to the local node's control channel, and this
171+-- module explicitly specifies to its clients that it does unsafe message
172+-- encoding. The same is true for the messages it puts onto the Management
173+-- event bus, however we do *not* want unevaluated thunks hitting the event
174+-- bus control thread. Hence the word /Unsafe/ in this module's name!
175+--
176+
130177-- | Send a message on a typed channel
131178sendChan :: Serializable a => SendPort a -> a -> Process ()
132179sendChan (SendPort cid) msg = do
133180 proc <- ask
134− let node = localNodeId (processNode proc)
135− destNode = processNodeId (sendPortProcessId cid) in do
136− case destNode == node of
137− True -> unsafeSendChanLocal cid msg
138− False -> do
139− liftIO $ sendBinary (processNode proc)
140− (ProcessIdentifier (processId proc))
141− (SendPortIdentifier cid)
142− NoImplicitReconnect
143− msg
181+ let node = processNode proc
182+ pid = processId proc
183+ us = localNodeId node
184+ them = processNodeId (sendPortProcessId cid)
185+ msg' = wrapMessage msg in do
186+ liftIO $ traceEvent (localEventBus node) (MxSentToPort pid cid msg')
187+ if them == us
188+ then unsafeSendChanLocal cid msg' -- NB: we wrap to P.Message !!!
189+ else liftIO $ sendBinary node
190+ (ProcessIdentifier pid)
191+ (SendPortIdentifier cid)
192+ NoImplicitReconnect
193+ msg
144194 where
145− unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
146− unsafeSendChanLocal spId msg' =
147− sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')
195+ unsafeSendChanLocal :: SendPortId -> Message -> Process ()
196+ unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m
148197
149198-- | Create an unencoded @Message@ for any @Serializable@ type.
150199wrapMessage :: Serializable a => a -> Message
stack-ghc-7.10.3.yaml+5−1
@@ -3,14 +3,18 @@ resolver: nightly-2016-03-08
33packages:
44- '.'
55- distributed-process-tests/
6+- location:
7+ git: https://github.com/haskell-distributed/network-transport-tcp.git
8+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
9+ extra-dep: true
610
711extra-deps:
812- rank1dynamic-0.4.0
913- distributed-static-0.3.8
1014- network-transport-0.5.2
11−- network-transport-tcp-0.6.0
1215- network-transport-inmemory-0.5.2
1316- rematch-0.2.0.0
17+- exceptions-0.8.2.1 # test dependency
1418
1519flags:
1620 distributed-process-tests:
stack-ghc-8.0.2.yaml+8−2
@@ -1,16 +1,22 @@
1−resolver: lts-9.0
1+resolver: lts-9.21
2+# added -location for network-transport-tcp-0.6.0
3+# since c7fd79 broke the client facing API for createTransport :/
24
35packages:
46- '.'
57- distributed-process-tests/
8+- location:
9+ git: https://github.com/haskell-distributed/network-transport-tcp.git
10+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
11+ extra-dep: true
612
713extra-deps:
814- rank1dynamic-0.4.0
915- distributed-static-0.3.8
1016- network-transport-0.5.2
11−- network-transport-tcp-0.6.0
1217- network-transport-inmemory-0.5.2
1318- rematch-0.2.0.0
19+# - exceptions-0.8.2.1 # test dependency
1420
1521flags:
1622 distributed-process-tests:
stack-ghc-8.2.2.yamlnew file+21
@@ -0,0 +1,21 @@
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.yamlnew file+20
@@ -0,0 +1,20 @@
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.yamlnew file+26
@@ -0,0 +1,26 @@
1+resolver: nightly
2+
3+packages:
4+- '.'
5+- distributed-process-tests/
6+- location:
7+ git: https://github.com/haskell-distributed/distributed-static.git
8+ commit: c51e6aadd3a44c044e96a8e986c11d3a94b45b11
9+ extra-dep: true
10+- location:
11+ git: https://github.com/haskell-distributed/network-transport-tcp.git
12+ commit: d58250e2388b86a1865ac4139373cbdc106e72e1
13+ extra-dep: true
14+- location:
15+ git: https://github.com/haskell-distributed/network-transport-inmemory.git
16+ commit: 0ce97924f15a8c1570b2355151834305c9bd2a28
17+ extra-dep: true
18+
19+extra-deps:
20+- rank1dynamic-0.4.0
21+- network-transport-0.5.2
22+- rematch-0.2.0.0
23+
24+flags:
25+ distributed-process-tests:
26+ tcp: true
stack.yaml+7−2
@@ -1,14 +1,19 @@
1−resolver: nightly-2017-08-20
1+resolver: lts-12.18 # Use GHC 8.4.4
2+
3+# TODO: we can enable nightly-2018-11-11 once PR #319 is merged
24
35packages:
46- '.'
57- distributed-process-tests/
8+- location:
9+ git: https://github.com/haskell-distributed/network-transport-tcp.git
10+ commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
11+ extra-dep: true
612
713extra-deps:
814- rank1dynamic-0.4.0
915- distributed-static-0.3.8
1016- network-transport-0.5.2
11−- network-transport-tcp-0.6.0
1217- network-transport-inmemory-0.5.2
1318- rematch-0.2.0.0
1419