git.iko.soy/distributed-process
on ab7368c help sign in
29 files changed+746−1299
since splitsnapshots
unifiedsplit
.travis.yml+6−11 benchmarks/Channels.hs+5−6 benchmarks/Latency.hs+5−7 benchmarks/ProcessRing.hs+2−2 benchmarks/Spawns.hs+5−6 benchmarks/Throughput.hs+5−6 distributed-process-tests/distributed-process-tests.cabal+25−12 distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+341−459 distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs+45−84 distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+5−16 distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+126−348 distributed-process-tests/src/Control/Distributed/Process/Tests/Receive.hs+8−7 distributed-process-tests/src/Control/Distributed/Process/Tests/Stats.hs+7−11 distributed-process-tests/src/Control/Distributed/Process/Tests/Tracing.hs+11−23 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+43−59 src/Control/Distributed/Process/Management.hs+2−34 src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs+2−8 src/Control/Distributed/Process/Management/Internal/Types.hs−7 src/Control/Distributed/Process/Node.hs+60−20 src/Control/Distributed/Process/UnsafePrimitives.hs+33−82 stack-ghc-7.10.3.yaml+1−5 stack-ghc-8.0.2.yaml+2−8 stack-ghc-8.2.2.yaml−21 stack-ghc-8.4.4.yaml−20 stack-nightly.yaml−26 stack.yaml+2−7
.travis.yml+6−11
@@ -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− - env: ARGS="--stack-yaml stack-ghc-8.2.2.yaml"
10− addons: {apt: {packages: [libgmp-dev]}}
11− - env: ARGS="--stack-yaml stack-ghc-8.4.4.yaml"
9+
10+ - env: ARGS="--stack-yaml stack-ghc-8.0.2.yaml"
1211 addons: {apt: {packages: [libgmp-dev]}}
13− - env: ARGS="--stack-yaml stack-nightly.yaml --resolver nightly"
12+
13+ - env: ARGS=
1414 addons: {apt: {packages: [libgmp-dev]}}
1515
1616cache:
@@ -34,10 +34,5 @@ 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}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=
37+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestMx
38+- stack ${ARGS} test $ARG='--plain' ${TEST_PACKAGE}TestTracing
benchmarks/Channels.hs+5−6
@@ -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, defaultTCPAddr, defaultTCPParameters)
7+import Network.Transport.TCP (createTransport, defaultTCPParameters)
88import Data.Binary (encode, decode)
99import qualified Data.ByteString.Lazy as BSL
1010
@@ -36,8 +36,7 @@ initialProcess "CLIENT" = do
3636main :: IO ()
3737main = do
3838 [role, host, port] <- getArgs
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
39+ Right transport <- createTransport
40+ host port (\sn -> (host, port)) defaultTCPParameters
41+ node <- newLocalNode transport initRemoteTable
42+ runProcess node $ initialProcess role
benchmarks/Latency.hs+5−7
@@ -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, defaultTCPAddr, defaultTCPParameters)
6+import Network.Transport.TCP (createTransport, defaultTCPParameters)
77import Data.Binary (encode, decode)
88import qualified Data.ByteString.Lazy as BSL
99
@@ -31,9 +31,7 @@ initialProcess "CLIENT" = do
3131main :: IO ()
3232main = do
3333 [role, host, port] <- getArgs
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
34+ Right transport <- createTransport
35+ host port (\sn -> (host, sn)) defaultTCPParameters
36+ node <- newLocalNode transport initRemoteTable
37+ runProcess node $ initialProcess role
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, defaultTCPAddr, defaultTCPParameters)
15+import Network.Transport.TCP (createTransport, 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− (defaultTCPAddr "127.0.0.1" "8090" ) defaultTCPParameters
114+ "127.0.0.1" "8090" (\sn -> ("127.0.0.1", sn)) defaultTCPParameters
115115 node <- newLocalNode transport initRemoteTable
116116 catch (void $ runProcess node $ initialProcess opt)
117117 (\(e :: SomeException) -> putStrLn $ "ERROR: " ++ (show e))
benchmarks/Spawns.hs+5−6
@@ -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, defaultTCPAddr, defaultTCPParameters)
10+import Network.Transport.TCP (createTransport, defaultTCPParameters)
1111import Data.Binary (encode, decode)
1212import qualified Data.ByteString.Lazy as BSL
1313
@@ -42,8 +42,7 @@ initialProcess "CLIENT" = do
4242main :: IO ()
4343main = do
4444 [role, host, port] <- getArgs
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
45+ Right transport <- createTransport
46+ host port (\sn -> (host, sn)) defaultTCPParameters
47+ node <- newLocalNode transport initRemoteTable
48+ runProcess node $ initialProcess role
benchmarks/Throughput.hs+5−6
@@ -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, defaultTCPAddr)
9+import Network.Transport.TCP (createTransport, defaultTCPParameters)
1010import Data.Binary
1111import qualified Data.ByteString.Lazy as BSL
1212import Data.Typeable
@@ -67,8 +67,7 @@ initialProcess "CLIENT" = do
6767main :: IO ()
6868main = do
6969 [role, host, port] <- getArgs
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
70+ Right transport <- createTransport
71+ host port (\sn -> (host, sn)) defaultTCPParameters
72+ node <- newLocalNode transport initRemoteTable
73+ runProcess node $ initialProcess role
distributed-process-tests/distributed-process-tests.cabal+25−12
@@ -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: Tim Watson <watson.timothy@gmail.com>
9+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>
1010copyright: Well-Typed LLP
1111category: Control, Cloud Haskell
1212build-type: Simple
@@ -31,10 +31,9 @@ library
3131 bytestring >= 0.9 && < 0.12,
3232 distributed-process >= 0.6.0 && < 0.8,
3333 distributed-static,
34− exceptions >= 0.5,
3534 HUnit >= 1.2 && < 1.7,
3635 network-transport >= 0.4.1.0 && < 0.6,
37− network >= 2.5 && < 2.9,
36+ network >= 2.5 && < 2.7,
3837 random >= 1.0 && < 1.2,
3938 rematch >= 0.1.2.1,
4039 setenv >= 0.1.1.3,
@@ -61,7 +60,7 @@ Test-Suite TestCHInMemory
6160 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH
6261 Build-Depends: base >= 4.4 && < 5,
6362 distributed-process-tests,
64− network >= 2.3 && < 2.9,
63+ network >= 2.3 && < 2.7,
6564 network-transport >= 0.4.1.0 && < 0.6,
6665 network-transport-inmemory >= 0.5,
6766 test-framework >= 0.6 && < 0.9
@@ -76,7 +75,7 @@ Test-Suite TestCHInTCP
7675 if flag(tcp)
7776 Build-Depends: base >= 4.4 && < 5,
7877 distributed-process-tests,
79− network >= 2.3 && < 2.9,
78+ network >= 2.3 && < 2.7,
8079 network-transport >= 0.4.1.0 && < 0.6,
8180 network-transport-tcp >= 0.5 && < 0.7,
8281 test-framework >= 0.6 && < 0.9
@@ -93,7 +92,7 @@ Test-Suite TestClosure
9392 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure
9493 Build-Depends: base >= 4.4 && < 5,
9594 distributed-process-tests,
96− network >= 2.3 && < 2.9,
95+ network >= 2.3 && < 2.7,
9796 network-transport >= 0.4.1.0 && < 0.6,
9897 network-transport-inmemory >= 0.5,
9998 test-framework >= 0.6 && < 0.9
@@ -107,7 +106,7 @@ Test-Suite TestStats
107106 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats
108107 Build-Depends: base >= 4.4 && < 5,
109108 distributed-process-tests,
110− network >= 2.3 && < 2.9,
109+ network >= 2.3 && < 2.7,
111110 network-transport >= 0.4.1.0 && < 0.6,
112111 network-transport-inmemory >= 0.5,
113112 test-framework >= 0.6 && < 0.9
@@ -121,12 +120,12 @@ Test-Suite TestMxInMemory
121120 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx
122121 Build-Depends: base >= 4.4 && < 5,
123122 distributed-process-tests,
124− network >= 2.3 && < 2.9,
123+ network >= 2.3 && < 2.7,
125124 network-transport >= 0.4.1.0 && < 0.6,
126125 network-transport-inmemory >= 0.5,
127126 test-framework >= 0.6 && < 0.9
128127 Extensions: CPP
129− ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
128+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
130129 HS-Source-Dirs: tests
131130
132131Test-Suite TestTracingInMemory
@@ -135,12 +134,12 @@ Test-Suite TestTracingInMemory
135134 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing
136135 Build-Depends: base >= 4.4 && < 5,
137136 distributed-process-tests,
138− network >= 2.3 && < 2.9,
137+ network >= 2.3 && < 2.7,
139138 network-transport >= 0.4.1.0 && < 0.6,
140139 network-transport-inmemory >= 0.5,
141140 test-framework >= 0.6 && < 0.9
142141 Extensions: CPP
143− ghc-options: -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
142+ ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
144143 HS-Source-Dirs: tests
145144
146145Test-Suite TestMxInTCP
@@ -149,10 +148,24 @@ Test-Suite TestMxInTCP
149148 CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx
150149 Build-Depends: base >= 4.4 && < 5,
151150 distributed-process-tests,
152− network >= 2.3 && < 2.9,
151+ network >= 2.3 && < 2.7,
153152 network-transport >= 0.4.1.0 && < 0.6,
154153 network-transport-inmemory >= 0.5,
155154 test-framework >= 0.6 && < 0.9
156155 Extensions: CPP
157156 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
158157 HS-Source-Dirs: tests
158+
159+Test-Suite TestTracingInTCP
160+ Type: exitcode-stdio-1.0
161+ Main-Is: runInMemory.hs
162+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing
163+ Build-Depends: base >= 4.4 && < 5,
164+ distributed-process-tests,
165+ network >= 2.3 && < 2.7,
166+ network-transport >= 0.4.1.0 && < 0.6,
167+ network-transport-inmemory >= 0.5,
168+ test-framework >= 0.6 && < 0.9
169+ Extensions: CPP
170+ ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind
171+ HS-Source-Dirs: tests
distributed-process-tests/src/Control/Distributed/Process/Tests/CH.hs+341−459
@@ -25,18 +25,13 @@ import Control.Concurrent.MVar
2525 )
2626import Control.Monad (replicateM_, replicateM, forever, void, unless, join)
2727import Control.Exception (SomeException, throwIO, ErrorCall(..))
28−import Control.Monad.Catch (try, catch, finally, mask, onException)
29−import Control.Applicative ((<|>))
28+import qualified Control.Exception as Ex (catch)
29+import Control.Applicative ((<$>), (<*>), pure, (<|>))
3030import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)
31−import Control.Distributed.Process hiding
32− ( try
33− , catch
34− , finally
35− , mask
36− , onException
37− )
31+import Control.Distributed.Process
3832import Control.Distributed.Process.Internal.Types
39− ( LocalNode(localEndPoint)
33+ ( NodeId(nodeAddress)
34+ , LocalNode(localEndPoint)
4035 , ProcessExitException(..)
4136 , nullProcessId
4237 , createUnencodedMessage
@@ -44,13 +39,13 @@ import Control.Distributed.Process.Internal.Types
4439import Control.Distributed.Process.Node
4540import Control.Distributed.Process.Debug
4641import Control.Distributed.Process.Management.Internal.Types
47−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)
42+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)
4843import Control.Distributed.Process.Serializable (Serializable)
49−import Data.Maybe (isNothing, isJust)
50−import Test.HUnit (Assertion, assertBool, assertFailure)
44+
45+import Test.HUnit (Assertion, assertFailure, assertBool)
5146import Test.Framework (Test, testGroup)
5247import Test.Framework.Providers.HUnit (testCase)
53−import Control.Rematch hiding (match, isNothing, isJust)
48+import Control.Rematch hiding (match)
5449import Control.Rematch.Run (Match(..))
5550
5651newtype Ping = Ping ProcessId
@@ -73,7 +68,7 @@ expectThat a matcher = case res of
7368forkTry :: IO () -> IO ThreadId
7469forkTry p = do
7570 tid <- myThreadId
76− forkIO $ catch p (\e -> throwTo tid (e :: SomeException))
71+ forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))
7772
7873-- | The ping server from the paper
7974ping :: Process ()
@@ -83,46 +78,12 @@ ping = do
8378 send partner (Ping self)
8479 ping
8580
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−
9481-- | Quick and dirty synchronous version of whereisRemoteAsync
9582whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
9683whereisRemote nid string = do
9784 whereisRemoteAsync nid string
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 ()
85+ WhereIsReply _ mPid <- expect
86+ return mPid
12687
12788data Add = Add ProcessId Double Double deriving (Typeable)
12889data Divide = Divide ProcessId Double Double deriving (Typeable)
@@ -197,18 +158,12 @@ monitorTestProcess theirAddr mOrL un reason monitorSetup done =
197158 unmonitor ref
198159 liftIO $ putMVar done ()
199160 (False, ref) -> do
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− ]
161+ ProcessMonitorNotification ref' pid reason' <- expect
162+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'
163+ liftIO $ putMVar done ()
208164 )
209165 (\(ProcessLinkException pid reason') -> do
210− (liftIO $ assertBool "link exception unmatched" $
211− pid == theirAddr && not mOrL && not un && reason == reason')
166+ True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'
212167 liftIO $ putMVar done ()
213168 )
214169
@@ -239,10 +194,8 @@ testPing TestTransport{..} = do
239194 pid <- getSelfPid
240195 replicateM_ numPings $ do
241196 send pingServer (Pong pid)
242− p <- expectTimeout 3000000
243− case p of
244− Just (Ping _) -> return ()
245− Nothing -> die "Failed to receive Ping"
197+ Ping _ <- expect
198+ return ()
246199
247200 putMVar clientDone ()
248201
@@ -329,7 +282,7 @@ testMonitorLocalDeadProcess TestTransport{..} mOrL un = do
329282 runProcess localNode $ do
330283 monitor theirAddr
331284 -- wait for the process to die
332− expect :: Process ProcessMonitorNotification
285+ ProcessMonitorNotification _ _ _ <- expect
333286 monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done
334287
335288 takeMVar done
@@ -403,17 +356,14 @@ testMath TestTransport{..} = do
403356 runProcess localNode $ do
404357 pid <- getSelfPid
405358 send mathServer (Add pid 1 2)
406− three <- expect :: Process Double
359+ 3 <- expect :: Process Double
407360 send mathServer (Divide pid 8 2)
408− four <- expect :: Process Double
361+ 4 <- expect :: Process Double
409362 send mathServer (Divide pid 8 0)
410− divByZ <- expect
411− liftIO $ putMVar clientDone (three, four, divByZ)
363+ DivByZero <- expect
364+ liftIO $ putMVar clientDone ()
412365
413− res <- takeMVar clientDone
414− case res of
415− (3, 4, DivByZero) -> return ()
416− _ -> error $ "Something went horribly wrong"
366+ takeMVar clientDone
417367
418368-- | Send first message (i.e. connect) to an already terminated process
419369-- (without monitoring); then send another message to a second process on
@@ -441,9 +391,11 @@ testSendToTerminated TestTransport{..} = do
441391 pid <- getSelfPid
442392 send server1 "Hi"
443393 send server2 (Pong pid)
444− expectPing clientDone
394+ Ping pid' <- expect
395+ True <- return $ pid' == server2
396+ liftIO $ putMVar clientDone ()
445397
446− verifyClient "Expected Ping from server" clientDone
398+ takeMVar clientDone
447399
448400-- | Test (non-zero) timeout
449401testTimeout :: TestTransport -> Assertion
@@ -452,10 +404,10 @@ testTimeout TestTransport{..} = do
452404 done <- newEmptyMVar
453405
454406 runProcess localNode $ do
455− res <- receiveTimeout 1000000 [match (\Add{} -> return ())]
456− liftIO $ putMVar done $ res == Nothing
407+ Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]
408+ liftIO $ putMVar done ()
457409
458− verifyClient "Expected receiveTimeout to timeout..." done
410+ takeMVar done
459411
460412-- | Test zero timeout
461413testTimeout0 :: TestTransport -> Assertion
@@ -483,9 +435,10 @@ testTimeout0 TestTransport{..} = do
483435 -- is not interested in, and then a single message that it wants
484436 replicateM_ 10000 $ send server "Irrelevant message"
485437 send server (Pong pid)
486− expectPing clientDone
438+ Ping _ <- expect
439+ liftIO $ putMVar clientDone ()
487440
488− verifyClient "Expected Ping from server" clientDone
441+ takeMVar clientDone
489442
490443-- | Test typed channels
491444testTypedChannels :: TestTransport -> Assertion
@@ -508,10 +461,10 @@ testTypedChannels TestTransport{..} = do
508461 runProcess localNode $ do
509462 (clientSendPort, rport) <- newChan
510463 sendChan serverSendPort (clientSendPort, 5)
511− ch <- receiveChan rport
512− liftIO $ putMVar clientDone $ ch == False
464+ False <- receiveChan rport
465+ liftIO $ putMVar clientDone ()
513466
514− verifyClient "Expected channel to send 'False'" clientDone
467+ takeMVar clientDone
515468
516469-- | Test merging receive ports
517470testMergeChannels :: TestTransport -> Assertion
@@ -534,8 +487,9 @@ testMergeChannels TestTransport{..} = do
534487 rs <- mapM charChannel "abc"
535488 m <- mergePorts biased rs
536489 xs <- replicateM 9 $ receiveChan m
537− liftIO $ putMVar done $ xs == expected
538− verifyClient "Expected single layer merge to match expected ordering" done
490+ True <- return $ xs == expected
491+ liftIO $ putMVar done ()
492+ takeMVar done
539493
540494 -- Two layers of merging
541495 testNested :: LocalNode -> Bool -> Bool -> String -> IO ()
@@ -546,8 +500,9 @@ testMergeChannels TestTransport{..} = do
546500 ms <- mapM (mergePorts biasedInner) rss
547501 m <- mergePorts biasedOuter ms
548502 xs <- replicateM (9 * 3) $ receiveChan m
549− liftIO $ putMVar done $ xs == expected
550− verifyClient "Expected nested channels to match expeted ordering" done
503+ True <- return $ xs == expected
504+ liftIO $ putMVar done ()
505+ takeMVar done
551506
552507 -- Test that if no messages are (immediately) available, the scheduler makes no difference
553508 testBlocked :: LocalNode -> Bool -> IO ()
@@ -556,45 +511,43 @@ testMergeChannels TestTransport{..} = do
556511 done <- newEmptyMVar
557512
558513 forkProcess localNode $ do
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"
514+ [sa, sb, sc] <- liftIO $ mapM readMVar vs
515+ mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
516+ [ -- a, b, c
517+ (sa, 'a')
518+ , (sb, 'b')
519+ , (sc, 'c')
520+ -- a, c, b
521+ , (sa, 'a')
522+ , (sc, 'c')
523+ , (sb, 'b')
524+ -- b, a, c
525+ , (sb, 'b')
526+ , (sa, 'a')
527+ , (sc, 'c')
528+ -- b, c, a
529+ , (sb, 'b')
530+ , (sc, 'c')
531+ , (sa, 'a')
532+ -- c, a, b
533+ , (sc, 'c')
534+ , (sa, 'a')
535+ , (sb, 'b')
536+ -- c, b, a
537+ , (sc, 'c')
538+ , (sb, 'b')
539+ , (sa, 'a')
540+ ]
589541
590542 forkProcess localNode $ do
591543 (ss, rs) <- unzip <$> replicateM 3 newChan
592544 liftIO $ mapM_ (uncurry putMVar) $ zip vs ss
593545 m <- mergePorts biased rs
594546 xs <- replicateM (6 * 3) $ receiveChan m
595− liftIO $ putMVar done $ xs == "abcacbbacbcacabcba"
547+ True <- return $ xs == "abcacbbacbcacabcba"
548+ liftIO $ putMVar done ()
596549
597− verifyClient "Expected merged ports to match expected ordering" done
550+ takeMVar done
598551
599552 mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)
600553 mergePorts True = mergePortsBiased
@@ -612,9 +565,8 @@ testTerminate TestTransport{..} = do
612565 localNode <- newLocalNode testTransport initRemoteTable
613566 runProcess localNode $ do
614567 e <- try terminate :: Process (Either ProcessTerminationException ())
615− if either show show e == show ProcessTerminationException
616− then return ()
617− else die "Unexpected result from terminate"
568+ True <- return $ either show show e == show ProcessTerminationException
569+ return ()
618570
619571testMonitorNode :: TestTransport -> Assertion
620572testMonitorNode TestTransport{..} = do
@@ -625,12 +577,11 @@ testMonitorNode TestTransport{..} = do
625577
626578 runProcess node2 $ do
627579 ref <- monitorNode (localNodeId node1)
628− receiveWait [
629− match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->
630− return $ ref == ref' && nid == localNodeId node1)
631− ] >>= liftIO . putMVar done
580+ NodeMonitorNotification ref' nid DiedDisconnect <- expect
581+ True <- return $ ref == ref' && nid == localNodeId node1
582+ liftIO $ putMVar done ()
632583
633− verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done
584+ takeMVar done
634585
635586testMonitorLiveNode :: TestTransport -> Assertion
636587testMonitorLiveNode TestTransport{..} = do
@@ -645,16 +596,15 @@ testMonitorLiveNode TestTransport{..} = do
645596 liftIO $ putMVar ready ()
646597 liftIO $ takeMVar readyr
647598 send p ()
648− receiveWait [
649− match (\(NodeMonitorNotification ref' nid _) ->
650− (return $ ref == ref' && nid == localNodeId node1))
651− ] >>= liftIO . putMVar done
599+ NodeMonitorNotification ref' nid _ <- expect
600+ True <- return $ ref == ref' && nid == localNodeId node1
601+ liftIO $ putMVar done ()
652602
653603 takeMVar ready
654604 closeLocalNode node1
655605 putMVar readyr ()
656606
657− verifyClient "Expected NodeMonitorNotification for LIVE node" done
607+ takeMVar done
658608
659609testMonitorChannel :: TestTransport -> Assertion
660610testMonitorChannel TestTransport{..} = do
@@ -664,20 +614,18 @@ testMonitorChannel TestTransport{..} = do
664614 pid <- forkProcess node1 $ do
665615 sport <- expect :: Process (SendPort ())
666616 ref <- monitorPort sport
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
617+ PortMonitorNotification ref' port' reason <- expect
618+ -- reason might be DiedUnknownId if the receive port is GCed before the
619+ -- monitor is established (TODO: not sure that this is reasonable)
620+ return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)
621+ liftIO $ putMVar gotNotification ()
674622
675623 runProcess node2 $ do
676624 (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())
677625 send pid sport
678626 liftIO $ threadDelay 100000
679627
680− verifyClient "Expected PortMonitorNotification" gotNotification
628+ takeMVar gotNotification
681629
682630testRegistry :: TestTransport -> Assertion
683631testRegistry TestTransport{..} = do
@@ -689,28 +637,20 @@ testRegistry TestTransport{..} = do
689637
690638 runProcess node $ do
691639 register "ping" pingServer
692− whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)
640+ Just pid <- whereis "ping"
641+ True <- return $ pingServer == pid
693642 us <- getSelfPid
694643 nsend "ping" (Pong us)
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
644+ Ping pid' <- expect
645+ True <- return $ pingServer == pid'
646+ Left (ProcessRegistrationException "dead" Nothing) <- try $ register "dead" deadProcess
647+ Left (ProcessRegistrationException "ping" (Just x)) <- try $ register "ping" deadProcess
648+ True <- return $ x == pingServer
649+ Left (ProcessRegistrationException "dead" Nothing) <- try $ unregister "dead"
701650 liftIO $ putMVar done ()
702651
703652 takeMVar done
704653
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−
714654testRegistryRemoteProcess :: TestTransport -> Assertion
715655testRegistryRemoteProcess TestTransport{..} = do
716656 node1 <- newLocalNode testTransport initRemoteTable
@@ -721,13 +661,13 @@ testRegistryRemoteProcess TestTransport{..} = do
721661
722662 runProcess node2 $ do
723663 register "ping" pingServer
724− whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)
664+ Just pid <- whereis "ping"
665+ True <- return $ pingServer == pid
725666 us <- getSelfPid
726667 nsend "ping" (Pong us)
727− receiveWait [
728− matchIf (\(Ping pid') -> pingServer == pid')
729− (const $ liftIO $ putMVar done ())
730− ]
668+ Ping pid' <- expect
669+ True <- return $ pingServer == pid'
670+ liftIO $ putMVar done ()
731671
732672 takeMVar done
733673
@@ -735,6 +675,7 @@ testRemoteRegistry :: TestTransport -> Assertion
735675testRemoteRegistry TestTransport{..} = do
736676 node1 <- newLocalNode testTransport initRemoteTable
737677 node2 <- newLocalNode testTransport initRemoteTable
678+ done <- newEmptyMVar
738679
739680 pingServer <- forkProcess node1 ping
740681 deadProcess <- forkProcess node1 (return ())
@@ -747,34 +688,32 @@ testRemoteRegistry TestTransport{..} = do
747688 "ping" == label' && pid == pingServer)
748689 (\(RegisterReply _ _ _) -> return ()) ]
749690
750− pid <- verifyWhereIsRemote nid1 "ping"
751− liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid
752−
691+ Just pid <- whereisRemote nid1 "ping"
692+ True <- return $ pingServer == pid
753693 us <- getSelfPid
754694 nsendRemote nid1 "ping" (Pong us)
755− receiveWait [
756− match (\(Ping pid') -> return $ pingServer == pid')
757− ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId"
695+ Ping pid' <- expect
696+ True <- return $ pingServer == pid'
758697
759698 -- test that if process was not registered Nothing is returned
760699 -- in owner field.
761700 registerRemoteAsync nid1 "dead" deadProcess
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−
701+ receiveWait [
702+ matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')
703+ (\(RegisterReply _ _ _) -> return ()) ]
766704 registerRemoteAsync nid1 "ping" deadProcess
767705 receiveWait [
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−
706+ matchIf (\(RegisterReply label' False (Just pid)) ->
707+ "ping" == label' && pid == pingServer)
708+ (\(RegisterReply _ _ _) -> return ()) ]
773709 unregisterRemoteAsync nid1 "dead"
774710 receiveWait [
775− matchIf (\(RegisterReply label' _ _) -> "dead" == label')
776− (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))
777− ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"
711+ matchIf (\(RegisterReply label' False Nothing) ->
712+ "dead" == label' && pid == pingServer)
713+ (\(RegisterReply _ _ _) -> return ()) ]
714+ liftIO $ putMVar done ()
715+
716+ takeMVar done
778717
779718testRemoteRegistryRemoteProcess :: TestTransport -> Assertion
780719testRemoteRegistryRemoteProcess TestTransport{..} = do
@@ -790,15 +729,15 @@ testRemoteRegistryRemoteProcess TestTransport{..} = do
790729 receiveWait [
791730 matchIf (\(RegisterReply label' _ _) -> "ping" == label')
792731 (\(RegisterReply _ _ _) -> return ()) ]
793− pid <- verifyWhereIsRemote nid1 "ping"
794− liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid
732+ Just pid <- whereisRemote nid1 "ping"
733+ True <- return $ pingServer == pid
795734 us <- getSelfPid
796735 nsendRemote nid1 "ping" (Pong us)
797− receiveWait [
798− match (\(Ping pid') -> return $ pingServer == pid')
799− ] >>= liftIO . putMVar done
736+ Ping pid' <- expect
737+ True <- return $ pingServer == pid'
738+ liftIO $ putMVar done ()
800739
801− verifyClient "Expected Ping with ping server's ProcessId" done
740+ takeMVar done
802741
803742testSpawnLocal :: TestTransport -> Assertion
804743testSpawnLocal TestTransport{..} = do
@@ -812,14 +751,15 @@ testSpawnLocal TestTransport{..} = do
812751 sport <- expect
813752 sendChan sport (1234 :: Int)
814753
815− sport <- spawnChannelLocal $
816− \rport -> (receiveChan rport :: Process Int) >>= send us
754+ sport <- spawnChannelLocal $ \rport -> do
755+ (1234 :: Int) <- receiveChan rport
756+ send us ()
817757
818758 send pid sport
819− expect >>= liftIO . putMVar done
759+ () <- expect
760+ liftIO $ putMVar done ()
820761
821− res <- takeMVar done
822− assertBool "Expected 1234 :: Int" $ res == (1234 :: Int)
762+ takeMVar done
823763
824764testSpawnAsyncStrictness :: TestTransport -> Assertion
825765testSpawnAsyncStrictness TestTransport{..} = do
@@ -848,9 +788,8 @@ testReconnect TestTransport{..} = do
848788 liftIO $ putMVar processA us
849789 msg1 <- expect
850790 msg2 <- expect
851− liftIO $ do
852− assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"
853− putMVar sendTestOk ()
791+ True <- return $ msg1 == "message 1" && msg2 == "message 3"
792+ liftIO $ putMVar sendTestOk ()
854793
855794 forkProcess node2 $ do
856795 {-
@@ -883,7 +822,7 @@ testReconnect TestTransport{..} = do
883822 matchIf (\(RegisterReply label' _ _) -> "a" == label')
884823 (\(RegisterReply _ _ _) -> return ()) ]
885824
886− _ <- whereisRemote nid1 "a"
825+ Just _ <- whereisRemote nid1 "a"
887826
888827
889828 -- Simulate network failure
@@ -902,10 +841,9 @@ testReconnect TestTransport{..} = do
902841 (\(RegisterReply _ _ _) -> return ()) ]
903842
904843 -- Check
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"
844+ Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected
845+ Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect
846+ Just _ <- whereisRemote nid1 "c"
909847
910848 liftIO $ putMVar registerTestOk ()
911849
@@ -946,19 +884,19 @@ testUSend usendPrim TestTransport{..} numMessages = do
946884 isSorted _ = True
947885 -- The list can't be null since there are no failures after sending
948886 -- the last message.
949− liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)
887+ True <- return $ isSorted msgs && not (null msgs)
888+ liftIO $ putMVar usendTestOk ()
950889
951890 forkProcess node2 $ do
952891 them <- liftIO $ readMVar processA
953892 getSelfPid >>= send them
954− expect :: Process ()
893+ () <- expect
955894 forM_ [1..numMessages] $ \i -> do
956895 liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)
957896 usendPrim them i
958897 liftIO (threadDelay 30000)
959898
960− res <- takeMVar usendTestOk
961− assertBool "Unexpected failure after sending last msg" res
899+ takeMVar usendTestOk
962900
963901-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
964902-- in between
@@ -984,17 +922,14 @@ testMatchAny TestTransport{..} = do
984922 runProcess localNode $ do
985923 pid <- getSelfPid
986924 send mathServer (Add pid 1 2)
987− three <- expect :: Process Double
925+ 3 <- expect :: Process Double
988926 send mathServer (Divide pid 8 2)
989− four <- expect :: Process Double
927+ 4 <- expect :: Process Double
990928 send mathServer (Divide pid 8 0)
991− divByZ <- expect
992− liftIO $ putMVar clientDone (three, four, divByZ)
929+ DivByZero <- expect
930+ liftIO $ putMVar clientDone ()
993931
994− res <- takeMVar clientDone
995− case res of
996− (3, 4, DivByZero) -> return ()
997− _ -> error "Unexpected result"
932+ takeMVar clientDone
998933
999934-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
1000935-- in between, however we block 'Divide' requests ....
@@ -1021,13 +956,12 @@ testMatchAnyHandle TestTransport{..} = do
1021956 runProcess localNode $ do
1022957 pid <- getSelfPid
1023958 send mathServer (Add pid 1 2)
1024− three <- expect :: Process Double
959+ 3 <- expect :: Process Double
1025960 send mathServer (Divide pid 8 2)
1026− res <- (expectTimeout 100000) :: Process (Maybe Double)
1027− liftIO $ putMVar clientDone $ three == 3 && res == Nothing
1028−
1029− verifyClient "Expected Nothing (i.e. timeout)" clientDone
961+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
962+ liftIO $ putMVar clientDone ()
1030963
964+ takeMVar clientDone
1031965 where maybeForward :: ProcessId -> Message -> Process (Maybe ())
1032966 maybeForward s msg =
1033967 handleMessage msg (\m@(Add _ _ _) -> send s m)
@@ -1055,10 +989,8 @@ testMatchAnyNoHandle TestTransport{..} = do
1055989 Just _ -> die "NONSENSE")
1056990 ]
1057991 -- we *must* have removed the message from our mailbox though!!!
1058− res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
1059− liftIO $ do
1060− assertBool "Expected timeout!" $ res == Nothing
1061− putMVar serverDone ()
992+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]
993+ liftIO $ putMVar serverDone ()
1062994 putMVar addr server
1063995
1064996 -- Client
@@ -1101,21 +1033,14 @@ testMatchAnyIf TestTransport{..} = do
11011033 runProcess localNode $ do
11021034 pid <- getSelfPid
11031035 send server (pid, "foo")
1104− foo <- expect
1105− -- provoking what would be the wrong ordering is informative here...
1106−
1107− send server (pid, "bar")
1108− bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !?
1109−
1036+ "foo" <- expect
11101037 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'
1038+ "baz" <- expect
1039+ send server (pid, "bar")
1040+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)
1041+ liftIO $ putMVar clientDone ()
11181042
1043+ takeMVar clientDone
11191044 where tryHandleMessage :: Message -> Process (Maybe ())
11201045 tryHandleMessage msg =
11211046 handleMessage msg (\(pid :: ProcessId, (m :: String))
@@ -1148,13 +1073,12 @@ testMatchMessageWithUnwrap TestTransport{..} = do
11481073 runProcess localNode $ do
11491074 pid <- getSelfPid
11501075 send server (pid, wrapMessage ("foo" :: String))
1151− foo <- expect
1076+ "foo" <- expect
11521077 send server (pid, wrapMessage ("baz" :: String))
1153− baz <- expect
1154− liftIO $ putMVar clientDone (foo, baz)
1078+ "baz" <- expect
1079+ liftIO $ putMVar clientDone ()
11551080
1156− res <- takeMVar clientDone
1157− assertBool "Unexpected unwrapped results" $ res == ("foo", "baz")
1081+ takeMVar clientDone
11581082
11591083-- Test 'receiveChanTimeout'
11601084testReceiveChanTimeout :: TestTransport -> Assertion
@@ -1172,19 +1096,18 @@ testReceiveChanTimeout TestTransport{..} = do
11721096
11731097 -- Wait for a message with a delay. No message arrives, we should get
11741098 -- Nothing after the delay.
1175− receiveChanTimeout 100000 rp >>= maybe (return ()) (const $ die "Expected Timeout")
1099+ Nothing <- receiveChanTimeout 100000 rp
11761100
11771101 -- Let the sender know that it can send a message.
11781102 liftIO $ putMVar mvSender ()
11791103
11801104 -- Wait for a message with a delay again. Now a message arrives after
11811105 -- 0.1 seconds
1182− res <- receiveChanTimeout 20000000 rp >>= maybe (die "Timeout") return
1183− liftIO $ assertBool "Expected result to be 'True'" res
1106+ Just True <- receiveChanTimeout 20000000 rp
11841107
11851108 -- Wait for a message with zero timeout: non-blocking check. No message is
11861109 -- available, we get Nothing
1187− receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout")
1110+ Nothing <- receiveChanTimeout 0 rp
11881111
11891112 -- Let the sender know that it can send a message.
11901113 liftIO $ putMVar mvSender ()
@@ -1194,7 +1117,7 @@ testReceiveChanTimeout TestTransport{..} = do
11941117 liftIO $ threadDelay 100000
11951118 mb <- receiveChanTimeout 0 rp
11961119 case mb of
1197− Just b -> do liftIO $ assertBool "Unexpected Message" $ not b
1120+ Just b -> do False <- return b; return ()
11981121 _ -> loop
11991122
12001123 liftIO $ putMVar done ()
@@ -1231,8 +1154,8 @@ testReceiveChanFeatures TestTransport{..} = do
12311154
12321155 rp1 <- mergePortsBiased [even <$> rpInt, rpBool]
12331156
1234− receiveChan rp1 >>= liftIO . assertBool "Expected True"
1235− receiveChan rp1 >>= liftIO . assertBool "Expected False" . not
1157+ True <- receiveChan rp1
1158+ False <- receiveChan rp1
12361159
12371160 -- Test Applicative instance
12381161
@@ -1241,7 +1164,7 @@ testReceiveChanFeatures TestTransport{..} = do
12411164
12421165 let rp2 = pure (+) <*> rpInt <*> rpInt
12431166
1244− receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7)
1167+ 7 <- receiveChan rp2
12451168
12461169 -- Test Alternative instance
12471170
@@ -1250,8 +1173,8 @@ testReceiveChanFeatures TestTransport{..} = do
12501173
12511174 let rp3 = (even <$> rpInt) <|> rpBool
12521175
1253− receiveChan rp3 >>= liftIO . assertBool "Expected False" . not
1254− receiveChan rp3 >>= liftIO . assertBool "Expected True"
1176+ False <- receiveChan rp3
1177+ True <- receiveChan rp3
12551178
12561179 -- Test Monad instance
12571180
@@ -1265,86 +1188,17 @@ testReceiveChanFeatures TestTransport{..} = do
12651188 then rpInt
12661189 else return 7
12671190
1268− receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)
1269− receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7)
1191+ 5 <- receiveChan rp4
1192+ 7 <- receiveChan rp4
12701193
12711194 liftIO $ putMVar done ()
12721195
12731196 takeMVar done
12741197
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−
13451198testKillLocal :: TestTransport -> Assertion
13461199testKillLocal TestTransport{..} = do
13471200 localNode <- newLocalNode testTransport initRemoteTable
1201+ done <- newEmptyMVar
13481202
13491203 pid <- forkProcess localNode $ do
13501204 liftIO $ threadDelay 1000000
@@ -1353,18 +1207,17 @@ testKillLocal TestTransport{..} = do
13531207 ref <- monitor pid
13541208 us <- getSelfPid
13551209 kill pid "TestKill"
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"
1210+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect
1211+ True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"
1212+ liftIO $ putMVar done ()
1213+
1214+ takeMVar done
13631215
13641216testKillRemote :: TestTransport -> Assertion
13651217testKillRemote TestTransport{..} = do
13661218 node1 <- newLocalNode testTransport initRemoteTable
13671219 node2 <- newLocalNode testTransport initRemoteTable
1220+ done <- newEmptyMVar
13681221
13691222 pid <- forkProcess node1 $ do
13701223 liftIO $ threadDelay 1000000
@@ -1373,17 +1226,11 @@ testKillRemote TestTransport{..} = do
13731226 ref <- monitor pid
13741227 us <- getSelfPid
13751228 kill pid "TestKill"
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"
1229+ ProcessMonitorNotification ref' pid' (DiedException reason) <- expect
1230+ True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"
1231+ liftIO $ putMVar done ()
1232+
1233+ takeMVar done
13871234
13881235testCatchesExit :: TestTransport -> Assertion
13891236testCatchesExit TestTransport{..} = do
@@ -1409,10 +1256,7 @@ testHandleMessageIf TestTransport{..} = do
14091256 self <- getSelfPid
14101257 send self (5 :: Integer, 10 :: Integer)
14111258 msg <- receiveWait [ matchMessage return ]
1412− handleMessageIf msg
1413− (\() -> True)
1414− (\() -> die "whoops") >>= maybe (return ())
1415− (const $ die "Expected Mismatch")
1259+ Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")
14161260 handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)
14171261 (\input -> liftIO $ putMVar done input)
14181262 return ()
@@ -1457,11 +1301,8 @@ testDie TestTransport{..} = do
14571301 (die ("foobar", 123 :: Int))
14581302 `catchExit` \_from reason -> do
14591303 -- TODO: should verify that 'from' has the right value
1460− let res = reason == ("foobar", 123 :: Int)
1304+ True <- return $ reason == ("foobar", 123 :: Int)
14611305 liftIO $ putMVar done ()
1462− if res
1463− then return ()
1464− else die "Something went horribly wrong"
14651306
14661307 takeMVar done
14671308
@@ -1475,11 +1316,8 @@ testPrettyExit TestTransport{..} = do
14751316 `catch` \ex@(ProcessExitException from _) ->
14761317 let expected = "exit-from=" ++ (show from)
14771318 in do
1478− let res = (show ex) == expected
1319+ True <- return $ (show ex) == expected
14791320 liftIO $ putMVar done ()
1480− if res
1481− then return ()
1482− else die "Something went horribly wrong"
14831321
14841322 takeMVar done
14851323
@@ -1495,26 +1333,17 @@ testExitLocal TestTransport{..} = do
14951333 pid <- forkProcess localNode $ do
14961334 (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do
14971335 -- TODO: should verify that 'from' has the right value
1498− res <- return $ reason == "TestExit"
1336+ True <- return $ reason == "TestExit"
14991337 liftIO $ putMVar supervisedDone ()
1500− if res
1501− then return ()
1502− else die "Something went horribly wrong"
15031338
15041339 runProcess localNode $ do
15051340 liftIO $ takeMVar handlerSetUp
15061341 ref <- monitor pid
15071342 exit pid "TestExit"
15081343 -- This time the client catches the exception, so it dies normally
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"
1344+ ProcessMonitorNotification ref' pid' DiedNormal <- expect
1345+ True <- return $ ref == ref' && pid == pid'
1346+ liftIO $ putMVar supervisorDone ()
15181347
15191348 takeMVar supervisedDone
15201349 takeMVar supervisorDone
@@ -1530,97 +1359,128 @@ testExitRemote TestTransport{..} = do
15301359 (receiveWait [] :: Process ()) -- block forever
15311360 `catchExit` \_from reason -> do
15321361 -- TODO: should verify that 'from' has the right value
1533− res <- return $ reason == "TestExit"
1362+ True <- return $ reason == "TestExit"
15341363 liftIO $ putMVar supervisedDone ()
1535− if res
1536− then return ()
1537− else die "Something went horribly wrong"
15381364
15391365 runProcess node2 $ do
15401366 ref <- monitor pid
15411367 exit pid "TestExit"
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"
1368+ ProcessMonitorNotification ref' pid' DiedNormal <- expect
1369+ True <- return $ ref == ref' && pid == pid'
1370+ liftIO $ putMVar supervisorDone ()
15511371
15521372 takeMVar supervisedDone
15531373 takeMVar supervisorDone
15541374
15551375testRegistryMonitoring :: TestTransport -> Assertion
15561376testRegistryMonitoring TestTransport{..} = do
1557− node1 <- newLocalNode testTransport initRemoteTable
1558− node2 <- newLocalNode testTransport initRemoteTable
1559−
1560− let nid = localNodeId node2
1561− pid <- forkProcess node1 $ do
1562− self <- getSelfPid
1563− runUntilRegistered nid self
1564− say $ (show self) ++ " registered as " ++ regName
1565− expect :: Process ()
1566− say $ (show self) ++ " exiting normally"
1377+ localNode <- newLocalNode testTransport initRemoteTable
1378+ remoteNode <- newLocalNode testTransport initRemoteTable
1379+ return ()
15671380
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
1381+ -- Local process. Test if local process will be removed from
1382+ -- registry when it dies.
1383+ box <- newEmptyMVar
1384+ runProcess localNode $ do
1385+ pid <- spawnLocal $ do
1386+ expect
1387+ register "test" pid
1388+ tpid <- whereis "test"
1389+ if tpid == Just pid
1390+ then do _ <- monitor pid
1391+ send pid ()
1392+ ProcessMonitorNotification{} <- expect
1393+ tpid1 <- whereis "test"
1394+ liftIO $ putMVar box (Nothing == tpid1)
1395+ else liftIO $ putMVar box False
1396+
1397+ takeMVar box >>= assertBool "expected local process to not be registered"
1398+ return ()
1399+
1400+ -- Remote process. Test if remote process entry is removed
1401+ -- from registry when process dies.
1402+ remote1 <- testRemote remoteNode
1403+ runProcess localNode $
1404+ let waitpoll = do
1405+ w <- whereis "test" :: Process (Maybe ProcessId)
1406+ forM_ w (const waitpoll)
1407+ in do register "test" remote1
1408+ -- [race-condition] invalid test if we force exit before waitpoll runs
1409+ registered <- whereis "test"
1410+ registered `shouldBe` (equalTo (Just remote1))
1411+ send remote1 ()
1412+ waitpoll
1413+ return ()
1414+ return ()
1415+
1416+ -- Many labels. Test if all labels associated with process
1417+ -- are removed from registry when it dies.
1418+ remote2 <- testRemote remoteNode
1419+ runProcess localNode $
1420+ let waitpoll = do
1421+ w1 <- whereis "test-3" :: Process (Maybe ProcessId)
1422+ w2 <- whereis "test-4" :: Process (Maybe ProcessId)
1423+ forM_ (w1 <|> w2) (const waitpoll)
1424+ in do register "test-3" remote2
1425+ register "test-4" remote2
1426+ send remote2 ()
1427+ waitpoll
1428+ return ()
1429+
1430+{- XXX: waiting including patch for nsend for remote process
1431+ remote3 <- testRemote remoteNode
1432+ remote4 <- testRemote remoteNode
1433+ -- test many labels
1434+ runProcess localNode $ do
1435+ register "test-3" remote3
1436+ reregister "test-3" remote4
1437+ send remote3 ()
1438+ liftIO $ threadDelay 50000 -- XXX: racy
1439+ monitor remote4
1440+ nsend "test-3" ()
1441+ ProcessMonitorNotification{} <- expect
1442+ return ()
1443+-}
1444+
1445+ -- Test registerRemoteAsync properties. Add a local process to
1446+ -- remote registry and checks that it is removed
1447+ -- when the process dies.
1448+ remote5 <- testRemote remoteNode
1449+ runProcess localNode $ do
1450+ registerRemoteAsync (localNodeId remoteNode) "test" remote5
1451+ receiveWait [
1452+ match (\(RegisterReply _ True _) -> return ())
1453+ ] >>= send remote5
1454+ let waitpoll = do
1455+ whereisRemoteAsync (localNodeId remoteNode) "test"
1456+ receiveWait [
1457+ match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))
1458+ ]
1459+ waitpoll
16041460
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"
1461+ -- Add remote process to remote registry and checks if
1462+ -- entry is removed then process is dead.
1463+ remote6 <- testRemote localNode
1464+ runProcess localNode $ do
1465+ registerRemoteAsync (localNodeId remoteNode) "test" remote6
1466+ receiveWait [
1467+ match (\(RegisterReply _ True _) -> return ())
1468+ ] >>= send remote6
1469+ let waitpoll = do
1470+ whereisRemoteAsync (localNodeId remoteNode) "test"
1471+ receiveWait [
1472+ match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))
1473+ ]
1474+ waitpoll
1475+ where
1476+ testRemote node = do
1477+ -- test many labels
1478+ pidBox <- newEmptyMVar
1479+ forkProcess node $ do
1480+ us <- getSelfPid
1481+ liftIO $ putMVar pidBox us
1482+ expect :: Process ()
1483+ takeMVar pidBox
16241484
16251485testUnsafeSend :: TestTransport -> Assertion
16261486testUnsafeSend TestTransport{..} = do
@@ -1637,7 +1497,8 @@ testUnsafeSend TestTransport{..} = do
16371497 void $ forkProcess localNode $ do
16381498 serverPid <- liftIO $ takeMVar serverAddr
16391499 getSelfPid >>= unsafeSend serverPid
1640− expect >>= liftIO . putMVar clientDone
1500+ () <- expect
1501+ liftIO $ putMVar clientDone ()
16411502
16421503 takeMVar clientDone
16431504
@@ -1656,7 +1517,8 @@ testUnsafeUSend TestTransport{..} = do
16561517 void $ forkProcess localNode $ do
16571518 serverPid <- liftIO $ takeMVar serverAddr
16581519 getSelfPid >>= unsafeUSend serverPid
1659− expect >>= liftIO . putMVar clientDone
1520+ () <- expect
1521+ liftIO $ putMVar clientDone ()
16601522
16611523 takeMVar clientDone
16621524
@@ -1667,7 +1529,8 @@ testUnsafeNSend TestTransport{..} = do
16671529 localNode <- newLocalNode testTransport initRemoteTable
16681530
16691531 pid <- forkProcess localNode $ do
1670− expect >>= liftIO . putMVar clientDone
1532+ () <- expect
1533+ liftIO $ putMVar clientDone ()
16711534
16721535 void $ runProcess localNode $ do
16731536 register "foobar" pid
@@ -1685,7 +1548,8 @@ testUnsafeNSendRemote TestTransport{..} = do
16851548 _ <- forkProcess localNode1 $ do
16861549 getSelfPid >>= register "foobar"
16871550 liftIO $ putMVar clientDone ()
1688− expect >>= liftIO . putMVar clientDone
1551+ () <- expect
1552+ liftIO $ putMVar clientDone ()
16891553
16901554 takeMVar clientDone
16911555 void $ runProcess localNode2 $ do
@@ -1709,7 +1573,7 @@ testUnsafeSendChan TestTransport{..} = do
17091573 serverPid <- liftIO $ takeMVar serverAddr
17101574 (sp, rp) <- newChan
17111575 unsafeSend serverPid sp
1712− receiveChan rp :: Process ()
1576+ () <- receiveChan rp
17131577 liftIO $ putMVar clientDone ()
17141578
17151579 takeMVar clientDone
@@ -1723,8 +1587,8 @@ testCallLocal TestTransport{..} = do
17231587 runProcess node $ do
17241588 r <- (/=) <$> getSelfPid <*> callLocal getSelfPid
17251589 liftIO $ putMVar result r
1726−
1727− takeMVar result >>= assertBool "Expected 'True'"
1590+ True <- takeMVar result
1591+ return ()
17281592
17291593 -- Testing that when callLocal is interrupted, the worker is interrupted.
17301594 ibox <- newIORef False
@@ -1741,9 +1605,10 @@ testCallLocal TestTransport{..} = do
17411605 (send keeper ())
17421606 caller <- expect
17431607 exit caller "test"
1744− expect :: Process ()
1745−
1746− readIORef ibox >>= assertBool "Expected 'True'"
1608+ () <- expect
1609+ return ()
1610+ True <- readIORef ibox
1611+ return ()
17471612
17481613 -- Testing that when the worker raises an exception, the exception is propagated to the parent.
17491614 ibox2 <- newIORef False
@@ -1752,8 +1617,8 @@ testCallLocal TestTransport{..} = do
17521617 liftIO $ writeIORef ibox2 $ case r of
17531618 Left (ErrorCall "e") -> True
17541619 _ -> False
1755−
1756− readIORef ibox >>= assertBool "Expected 'True'"
1620+ True <- readIORef ibox
1621+ return ()
17571622
17581623 -- Test that caller waits for the worker in correct situation
17591624 ibox3 <- newIORef False
@@ -1770,8 +1635,8 @@ testCallLocal TestTransport{..} = do
17701635 liftIO $ putMVar result3 =<< readIORef ibox3
17711636 worker <- expect
17721637 send worker ()
1773−
1774− takeMVar result3 >>= assertBool "Expected 'True'"
1638+ True <- takeMVar result3
1639+ return ()
17751640
17761641 -- Test that caller waits for the worker in case when caller gets an exception
17771642 ibox4 <- newIORef False
@@ -1787,10 +1652,14 @@ testCallLocal TestTransport{..} = do
17871652 `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)
17881653 caller <- expect
17891654 exit caller "hi!"
1790−
1791− takeMVar result4 >>= assertBool "Expected 'True'"
1655+ True <- takeMVar result4
1656+ return ()
17921657 -- XXX: Testing that when mask_ $ callLocal p runs p in masked state.
17931658
1659+
1660+
1661+
1662+
17941663tests :: TestTransport -> IO [Test]
17951664tests testtrans = return [
17961665 testGroup "Basic features" [
@@ -1814,7 +1683,6 @@ tests testtrans = return [
18141683 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)
18151684 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)
18161685 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)
1817− , testCase "ChanLifecycle" (testChanLifecycle testtrans)
18181686 , testCase "KillLocal" (testKillLocal testtrans)
18191687 , testCase "KillRemote" (testKillRemote testtrans)
18201688 , testCase "Die" (testDie testtrans)
@@ -1824,7 +1692,7 @@ tests testtrans = return [
18241692 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)
18251693 , testCase "ExitLocal" (testExitLocal testtrans)
18261694 , testCase "ExitRemote" (testExitRemote testtrans)
1827− , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)
1695+ , testCase "TestRegistryMonitor" (testRegistryMonitoring testtrans)
18281696 , testCase "TextCallLocal" (testCallLocal testtrans)
18291697 -- Unsafe Primitives
18301698 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)
@@ -1877,3 +1745,17 @@ tests testtrans = return [
18771745 , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)
18781746 ]
18791747 ]
1748+
1749+syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()
1750+syncBreakConnection breakConnection nid0 nid1 = do
1751+ m <- newEmptyMVar
1752+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m
1753+ runProcess nid0 $ do
1754+ them <- liftIO $ takeMVar m
1755+ pinger <- spawnLocal $ forever $ send them ()
1756+ _ <- monitorNode (localNodeId nid1)
1757+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)
1758+ (nodeAddress $ localNodeId nid1)
1759+ NodeMonitorNotification _ _ _ <- expect
1760+ kill pinger "finished"
1761+ return ()
distributed-process-tests/src/Control/Distributed/Process/Tests/Closure.hs+45−84
@@ -169,11 +169,8 @@ testUnclosure TestTransport{..} rtable = do
169169 node <- newLocalNode testTransport rtable
170170 done <- newEmptyMVar
171171 forkProcess node $ do
172− i <- join . unClosure $ factorialClosure 5
172+ 120 <- join . unClosure $ factorialClosure 5
173173 liftIO $ putMVar done ()
174− if i == 720
175− then return ()
176− else error "Something went horribly wrong"
177174 takeMVar done
178175
179176testBind :: TestTransport -> RemoteTable -> Assertion
@@ -183,11 +180,8 @@ testBind TestTransport{..} rtable = do
183180 runProcess node $ do
184181 us <- getSelfPid
185182 join . unClosure $ sendFac 6 us
186− (i :: Int) <- expect
183+ (720 :: Int) <- expect
187184 liftIO $ putMVar done ()
188− if i == 720
189− then return ()
190− else error "Something went horribly wrong"
191185 takeMVar done
192186
193187testSendPureClosure :: TestTransport -> RemoteTable -> Assertion
@@ -200,7 +194,7 @@ testSendPureClosure TestTransport{..} rtable = do
200194 addr <- forkProcess node $ do
201195 cl <- expect
202196 fn <- unClosure cl :: Process (Int -> Int)
203− (_ :: Int) <- return $ fn 6
197+ 13 <- return $ fn 6
204198 liftIO $ putMVar serverDone ()
205199 putMVar serverAddr addr
206200
@@ -224,11 +218,8 @@ testSendIOClosure TestTransport{..} rtable = do
224218 liftIO $ do
225219 someMVar <- newEmptyMVar
226220 io someMVar
227− i <- readMVar someMVar
221+ 5 <- readMVar someMVar
228222 putMVar serverDone ()
229− if i == 5
230− then return ()
231− else error "Something went horribly wrong"
232223 putMVar serverAddr addr
233224
234225 forkIO $ do
@@ -257,10 +248,8 @@ testSendProcClosure TestTransport{..} rtable = do
257248 runProcess node $ do
258249 pid <- getSelfPid
259250 send theirAddr (cpSend $(mkStatic 'sdictInt) pid)
260− i <- expect :: Process Int
261− if i == 5
262− then liftIO $ putMVar clientDone ()
263− else error "Something went horribly wrong"
251+ 5 <- expect :: Process Int
252+ liftIO $ putMVar clientDone ()
264253
265254 takeMVar clientDone
266255
@@ -280,9 +269,8 @@ testSpawn TestTransport{..} rtable = do
280269 pid <- getSelfPid
281270 pid' <- spawn nid (sendPidClosure pid)
282271 pid'' <- expect
283− if pid' == pid''
284− then liftIO $ putMVar clientDone ()
285− else error "Something went horribly wrong"
272+ True <- return $ pid' == pid''
273+ liftIO $ putMVar clientDone ()
286274
287275 takeMVar clientDone
288276
@@ -306,9 +294,8 @@ testSpawnRace TestTransport{..} rtable = do
306294 spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid
307295 pid' <- expect :: Process ProcessId
308296 pid'' <- expect :: Process ProcessId
309− if pid' == pid''
310− then return ()
311− else error "Something went horribly wrong"
297+ True <- return $ pid' == pid''
298+ return ()
312299
313300 where
314301
@@ -345,10 +332,8 @@ testCall TestTransport{..} rtable = do
345332 node <- newLocalNode testTransport rtable
346333 nid <- readMVar serverNodeAddr
347334 runProcess node $ do
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"
335+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)
336+ liftIO $ putMVar clientDone ()
352337
353338 takeMVar clientDone
354339
@@ -365,10 +350,8 @@ testCallBind TestTransport{..} rtable = do
365350 node <- newLocalNode testTransport rtable
366351 nid <- readMVar serverNodeAddr
367352 runProcess node $ do
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"
353+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)
354+ liftIO $ putMVar clientDone ()
372355
373356 takeMVar clientDone
374357
@@ -379,11 +362,9 @@ testSeq TestTransport{..} rtable = do
379362 runProcess node $ do
380363 us <- getSelfPid
381364 join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us
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"
365+ 120 :: Int <- expect
366+ 720 :: Int <- expect
367+ liftIO $ putMVar done ()
387368 takeMVar done
388369
389370-- Test 'spawnSupervised'
@@ -412,33 +393,25 @@ testSpawnSupervised TestTransport{..} rtable = do
412393 throw supervisorDeath
413394
414395 forkProcess node2 $ do
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"
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 ()
442415
443416 takeMVar thirdProcessDone
444417 where
@@ -453,10 +426,8 @@ testSpawnInvalid TestTransport{..} rtable = do
453426 (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty)
454427 ProcessMonitorNotification ref' pid' _reason <- expect
455428 -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure
456− res <- return $ ref' == ref && pid == pid'
457− if res == True
458− then liftIO $ putMVar done ()
459− else error "Something went horribly wrong"
429+ True <- return $ ref' == ref && pid == pid'
430+ liftIO $ putMVar done ()
460431 takeMVar done
461432
462433testClosureExpect :: TestTransport -> RemoteTable -> Assertion
@@ -468,10 +439,8 @@ testClosureExpect TestTransport{..} rtable = do
468439 us <- getSelfPid
469440 them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us
470441 send them (1234 :: Int)
471− (res :: Int) <- expect
472− if res == 1234
473− then liftIO $ putMVar done ()
474− else error "Something went horribly wrong"
442+ (1234 :: Int) <- expect
443+ liftIO $ putMVar done ()
475444 takeMVar done
476445
477446testSpawnChannel :: TestTransport -> RemoteTable -> Assertion
@@ -496,10 +465,8 @@ testTDict TestTransport{..} rtable = do
496465 done <- newEmptyMVar
497466 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
498467 forkProcess node1 $ do
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..."
468+ True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))
469+ liftIO $ putMVar done ()
503470 takeMVar done
504471
505472testFib :: TestTransport -> RemoteTable -> Assertion
@@ -510,11 +477,8 @@ testFib TestTransport{..} rtable = do
510477 forkProcess (head nodes) $ do
511478 (sport, rport) <- newChan
512479 spawnLocal $ dfib (map localNodeId nodes, sport, 10)
513− ff <- receiveChan rport :: Process Integer
480+ 55 <- receiveChan rport :: Process Integer
514481 liftIO $ putMVar done ()
515− if ff /= 55
516− then die $ "Something went horribly wrong"
517− else return ()
518482
519483 takeMVar done
520484
@@ -522,7 +486,7 @@ testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion
522486testSpawnReconnect testtrans@TestTransport{..} rtable = do
523487 [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable
524488 let nid1 = localNodeId node1
525− -- nid2 = localNodeId node2
489+ nid2 = localNodeId node2
526490 done <- newEmptyMVar
527491 iv <- newMVar (0 :: Int)
528492
@@ -539,12 +503,9 @@ testSpawnReconnect testtrans@TestTransport{..} rtable = do
539503 liftIO $ threadDelay 100000
540504
541505 count <- liftIO $ takeMVar iv
542− res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
506+ True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'
543507
544508 liftIO $ putMVar done ()
545− if res /= True
546− then error "Something went horribly wrong"
547− else return ()
548509
549510 takeMVar done
550511
distributed-process-tests/src/Control/Distributed/Process/Tests/Internal/Utils.hs+5−16
@@ -20,7 +20,6 @@ module Control.Distributed.Process.Tests.Internal.Utils
2020 -- ping !
2121 , Ping(Ping)
2222 , ping
23− , pause
2423 , shouldBe
2524 , shouldMatch
2625 , shouldContain
@@ -74,16 +73,15 @@ import Control.Concurrent
7473import Control.Concurrent.MVar
7574 ( putMVar
7675 )
77−import Control.Distributed.Process hiding (finally, catch)
76+import Control.Distributed.Process
7877import Control.Distributed.Process.Node
7978import Control.Distributed.Process.Serializable()
8079
8180import Control.Exception (AsyncException(ThreadKilled), SomeException)
82−import Control.Monad (forever, void)
83−import Control.Monad.Catch (finally, catch)
81+import Control.Monad (forever)
8482import Control.Monad.STM (atomically)
8583import Control.Rematch hiding (match)
86−import Control.Rematch.Run
84+import Control.Rematch.Run
8785import Data.Binary
8886import Data.Typeable (Typeable)
8987
@@ -91,7 +89,6 @@ import Test.HUnit (Assertion, assertFailure)
9189import Test.HUnit.Base (assertBool)
9290
9391import GHC.Generics
94−import System.Timeout (timeout)
9592
9693-- | A mutable cell containing a test result.
9794type TestResult a = MVar a
@@ -110,18 +107,10 @@ data TestProcessControl = Stop | Go | Report ProcessId
110107
111108instance Binary TestProcessControl where
112109
113−data Private = Private
114− deriving (Typeable, Generic)
115−instance Binary Private where
116−
117110-- | Does exactly what it says on the tin, doing so in the @Process@ monad.
118111noop :: Process ()
119112noop = return ()
120113
121−pause :: Int -> Process ()
122−pause delay =
123− void $ receiveTimeout delay [ match (\Private -> return ()) ]
124−
125114synchronisedAssertion :: Eq a
126115 => String
127116 -> LocalNode
@@ -234,10 +223,10 @@ testProcessReport pid = do
234223tryRunProcess :: LocalNode -> Process () -> IO ()
235224tryRunProcess node p = do
236225 tid <- liftIO myThreadId
237− void $ timeout (1000000 * 60 * 5 :: Int) $
238− runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
226+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
239227
240228tryForkProcess :: LocalNode -> Process () -> IO ProcessId
241229tryForkProcess node p = do
242230 tid <- liftIO myThreadId
243231 forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
232+
distributed-process-tests/src/Control/Distributed/Process/Tests/Mx.hs+126−348
@@ -1,23 +1,12 @@
1−{-# LANGUAGE DeriveGeneric #-}
2−{-# LANGUAGE ParallelListComp #-}
3−
1+{-# LANGUAGE DeriveGeneric #-}
42module Control.Distributed.Process.Tests.Mx (tests) where
53
64import Control.Distributed.Process.Tests.Internal.Utils
75import Network.Transport.Test (TestTransport(..))
8−import Control.Distributed.Process hiding (bracket, finally, try)
9−import Control.Distributed.Process.Internal.Types
10− ( ProcessExitException(..)
11− , unsafeCreateUnencodedMessage
12− )
6+
7+import Control.Concurrent (threadDelay)
8+import Control.Distributed.Process
139import Control.Distributed.Process.Node
14−import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe
15− ( send
16− , nsend
17− , nsendRemote
18− , usend
19− , sendChan
20− )
2110import Control.Distributed.Process.Management
2211 ( MxEvent(..)
2312 , MxAgentId(..)
@@ -32,15 +21,15 @@ import Control.Distributed.Process.Management
3221 , mxUpdateLocal
3322 , mxNotify
3423 , mxBroadcast
24+ , mxGetId
3525 )
36−import Control.Monad (void, unless)
37−import Control.Monad.Catch(finally, bracket, try)
26+import Control.Monad (void)
3827import Control.Rematch (equalTo)
3928import Data.Binary
40−import Data.List (find, sort, intercalate)
41−import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)
29+import Data.List (find, sort)
30+import Data.Maybe (isJust)
4231import Data.Typeable
43−import GHC.Generics hiding (from)
32+import GHC.Generics
4433#if ! MIN_VERSION_base(4,6,0)
4534import Prelude hiding (catch, log)
4635#endif
@@ -56,17 +45,7 @@ data Publish = Publish
5645
5746instance Binary Publish where
5847
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 ()
48+testAgentBroadcast :: TestResult () -> Process ()
7049testAgentBroadcast result = do
7150 (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())
7251
@@ -81,7 +60,7 @@ testAgentBroadcast result = do
8160 mxNotify ()
8261 -- Once the publisher has seen our message, it will broadcast the Publish
8362 -- and the consumer will see that and send the result to our typed channel.
84− stash result =<< receiveChanTimeout 10000000 resultRP
63+ stash result =<< receiveChan resultRP
8564
8665 kill publisher "finished"
8766 kill consumer "finished"
@@ -89,7 +68,7 @@ testAgentBroadcast result = do
8968testAgentDualInput :: TestResult (Maybe Int) -> Process ()
9069testAgentDualInput result = do
9170 (sp, rp) <- newChan
92− s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
71+ _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [
9372 mxSink $ (\(i :: Int) -> do
9473 mxSetLocal . (+i) =<< mxGetLocal
9574 i' <- mxGetLocal
@@ -99,8 +78,6 @@ testAgentDualInput result = do
9978 else mxReady)
10079 ]
10180
102− mRef <- monitor s
103−
10481 mxNotify (1 :: Int)
10582 nsend "sum-agent" (3 :: Int)
10683 mxNotify (2 :: Int)
@@ -108,10 +85,6 @@ testAgentDualInput result = do
10885 mxNotify (5 :: Int)
10986
11087 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)
11588
11689testAgentPrioritisation :: TestResult [String] -> Process ()
11790testAgentPrioritisation result = do
@@ -123,12 +96,12 @@ testAgentPrioritisation result = do
12396
12497 let name = "prioritising-agent"
12598 (sp, rp) <- newChan
126− s <- mxAgent (MxAgentId name) ["first"] [
127− mxSink (\(s :: String) -> do
128− mxUpdateLocal (s:)
99+ void $ mxAgent (MxAgentId name) ["first"] [
100+ mxSink (\(s :: String) -> do
101+ mxUpdateLocal ((s:))
129102 st <- mxGetLocal
130103 case length st of
131− n | n == 5 -> do liftMX $ sendChan sp (sort st)
104+ n | n == 5 -> do liftMX $ sendChan sp st
132105 mxDeactivate "finished"
133106 _ -> mxReceive -- go to the mailbox
134107 )
@@ -139,8 +112,7 @@ testAgentPrioritisation result = do
139112 mxNotify "fourth"
140113 nsend name "fifth"
141114
142− stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp
143− awaitExit s
115+ stash result . sort =<< receiveChan rp
144116
145117testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()
146118testAgentMailboxHandling result = do
@@ -156,16 +128,7 @@ testAgentMailboxHandling result = do
156128
157129testAgentEventHandling :: TestResult Bool -> Process ()
158130testAgentEventHandling 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−
167131 let initState = [] :: [MxEvent]
168− (rc, rs) <- newChan
169132 agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [
170133 (mxSink $ \ev -> do
171134 st <- mxGetLocal
@@ -174,7 +137,7 @@ testAgentEventHandling result = do
174137 (MxSpawned _) -> mxSetLocal (ev:st)
175138 (MxProcessDied _ _) -> mxSetLocal (ev:st)
176139 _ -> return ()
177− act >> (liftMX $ sendChan rc ()) >> mxReady),
140+ act >> mxReady),
178141 (mxSink $ \(ev, sp :: SendPort Bool) -> do
179142 st <- mxGetLocal
180143 let found =
@@ -194,349 +157,164 @@ testAgentEventHandling result = do
194157 mxReady)
195158 ]
196159
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
160+ _ <- monitor agentPid
161+ (sp, rp) <- newChan
162+ pid <- spawnLocal $ sendChan sp ()
163+ () <- receiveChan rp
203164
204165 -- By waiting for a monitor notification, we have a
205− -- higher probably that the agent has seen the spawn and died events
166+ -- higher probably that the agent has seen
206167 monitor pid
207−
208− send pid ()
209− rct <- receiveChanTimeout 10000000 rp
210− unless (isJust rct) $ die "No response on channel"
211−
212− pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]
213− unless (isJust pmn) $ die "No monitor notification arrived"
214−
215− -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...
216− faff 2000000 rs
168+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]
217169
218170 (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)
219171 mxNotify (MxSpawned pid, replyTo)
220172 mxNotify (MxProcessDied pid DiedNormal, replyTo)
221173
222− seenAlive <- receiveChanTimeout 10000000 reply
223− seenDead <- receiveChanTimeout 10000000 reply
174+ seenAlive <- receiveChan reply
175+ seenDead <- receiveChan reply
176+
177+ stash result $ seenAlive && seenDead
224178
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
179+testMxRegEvents :: Process ()
180+testMxRegEvents = do
232181
233−testMxRegEvents :: TestResult () -> Process ()
234−testMxRegEvents result = do
235182 {- This test only deals with the local case, to ensure that we are being
236183 notified in the expected order - the remote cases related to the
237184 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− ]
254185
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− ]
186+ let label = "testMxRegEvents"
187+ let agentLabel = "mxRegEvents-agent"
188+ let delay = 1000000
189+ (regChan, regSink) <- newChan
190+ (unRegChan, unRegSink) <- newChan
191+ agent <- mxAgent (MxAgentId agentLabel) () [
192+ mxSink $ \ev -> do
193+ 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 ()
199+ mxReady
200+ ]
301201
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) () [
202+ p1 <- spawnLocal expect
203+ p2 <- spawnLocal expect
204+
205+ register label p1
206+ reg1 <- receiveChanTimeout delay regSink
207+ reg1 `shouldBe` equalTo (Just (label, p1))
208+
209+ unregister label
210+ unreg1 <- receiveChanTimeout delay unRegSink
211+ unreg1 `shouldBe` equalTo (Just (label, p1))
212+
213+ register label p2
214+ reg2 <- receiveChanTimeout delay regSink
215+ reg2 `shouldBe` equalTo (Just (label, p2))
216+
217+ reregister label p1
218+ unreg2 <- receiveChanTimeout delay unRegSink
219+ unreg2 `shouldBe` equalTo (Just (label, p2))
220+
221+ reg3 <- receiveChanTimeout delay regSink
222+ reg3 `shouldBe` equalTo (Just (label, p1))
223+
224+ mapM_ (flip kill $ "test-complete") [agent, p1, p2]
225+
226+testMxRegMon :: LocalNode -> Process ()
227+testMxRegMon remoteNode = do
228+
229+ -- ensure that when a registered process dies, we get a notification that
230+ -- it has been unregistered as well as seeing the name get removed
231+
232+ let label1 = "aaaaa"
233+ let label2 = "bbbbb"
234+ let isValid l = l == label1 || l == label2
235+ let agentLabel = "mxRegMon-agent"
236+ let delay = 1000000
237+ (regChan, regSink) <- newChan
238+ (unRegChan, unRegSink) <- newChan
239+ agent <- mxAgent (MxAgentId agentLabel) () [
423240 mxSink $ \ev -> do
424241 case ev of
425− m@(MxSentToPort _ cid _)
426− | cid /= sendPortId chan
427− && cid /= sendPortId sp -> liftMX $ sendChan chan m
428− m@(MxSent _ fromPid _)
429− | fromPid == us -> liftMX $ sendChan chan m
430− m@(MxSentToName _ fromPid _)
431− | fromPid == us -> liftMX $ sendChan chan m
432− m@(MxRegistered _ name)
433− | name == label -> liftMX $ sendChan chan m
242+ MxRegistered pid label
243+ | isValid label -> liftMX $ sendChan regChan (label, pid)
244+ MxUnRegistered pid label
245+ | isValid label -> liftMX $ sendChan unRegChan (label, pid)
434246 _ -> return ()
435247 mxReady
436248 ]
437249
438− case mNode of
439− Nothing -> void $ spawnLocal (proc sp)
440− Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp
250+ (sp, rp) <- newChan
251+ liftIO $ forkProcess remoteNode $ do
252+ getSelfPid >>= sendChan sp
253+ expect :: Process ()
254+
255+ p1 <- receiveChan rp
441256
442− p1 <- receiveChanTimeout 2000000 rp
443− unless (isJust p1) $ die "Timed out waiting for ProcessId"
444− res <- try $ test (fromJust p1) sink
257+ register label1 p1
258+ reg1 <- receiveChanTimeout delay regSink
259+ reg1 `shouldBe` equalTo (Just (label1, p1))
445260
446− kill agent "bye"
447− kill (fromJust p1) "bye"
261+ register label2 p1
262+ reg2 <- receiveChanTimeout delay regSink
263+ reg2 `shouldBe` equalTo (Just (label2, p1))
448264
449− case res of
450− Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m
451− Right tr -> tr `shouldBe` equalTo True
265+ n1 <- whereis label1
266+ n1 `shouldBe` equalTo (Just p1)
452267
268+ n2 <- whereis label2
269+ n2 `shouldBe` equalTo (Just p1)
453270
454− where
455− agentLabel s = "mx-unsafe-check-agent-" ++ show s
456− proc sp' = getSelfPid >>= sendChan sp' >> go Nothing
271+ kill p1 "goodbye"
457272
458− go :: Maybe (ReceivePort ()) -> Process ()
459− go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go
460− go c@(Just c') = receiveWait [ matchChan c' return ] >> go c
273+ unreg1 <- receiveChanTimeout delay unRegSink
274+ unreg2 <- receiveChanTimeout delay unRegSink
461275
462− replyChannel p' = do
463− (s, r) <- newChan
464− send p' s
465− return r
276+ let evts = [unreg1, unreg2]
277+ -- we can't rely on the order of the values in the node controller's
278+ -- map (it's either racy to do so, or no such guarantee exists for Data.Map),
279+ -- so we simply verify that we received the un-registration events we expect
280+ evts `shouldContain` (Just (label1, p1))
281+ evts `shouldContain` (Just (label2, p1))
282+
283+ kill agent "test-complete"
466284
467285tests :: TestTransport -> IO [Test]
468286tests TestTransport{..} = do
469287 node1 <- newLocalNode testTransport initRemoteTable
470288 node2 <- newLocalNode testTransport initRemoteTable
471289 return [
472− testGroup "MxAgents" [
473− testCase "EventHandling"
290+ testGroup "Mx Agents" [
291+ testCase "Event Handling"
474292 (delayedAssertion
475293 "expected True, but events where not as expected"
476294 node1 True testAgentEventHandling)
477− , testCase "InterAgentBroadcast"
295+ , testCase "Inter-Agent Broadcast"
478296 (delayedAssertion
479297 "expected (), but no broadcast was received"
480− node1 (Just ()) testAgentBroadcast)
481− , testCase "AgentMailboxHandling"
298+ node1 () testAgentBroadcast)
299+ , testCase "Agent Mailbox Handling"
482300 (delayedAssertion
483301 "expected (Just ()), but no regular (mailbox) input was handled"
484302 node1 (Just ()) testAgentMailboxHandling)
485− , testCase "AgentDualInputHandling"
303+ , testCase "Agent Dual Input Handling"
486304 (delayedAssertion
487305 "expected sum = 15, but the result was Nothing"
488306 node1 (Just 15 :: Maybe Int) testAgentDualInput)
489− , testCase "AgentInputPrioritisation"
307+ , testCase "Agent Input Prioritisation"
490308 (delayedAssertion
491309 "expected [first, second, third, fourth, fifth], but result diverged"
492310 node1 (sort ["first", "second",
493311 "third", "fourth",
494312 "fifth"]) testAgentPrioritisation)
495313 ]
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
314+ , testGroup "Mx Events" [
315+ testCase "Name Registration Events"
316+ (runProcess node1 testMxRegEvents)
317+ , testCase "Post Death Name UnRegistration Events"
318+ (runProcess node1 (testMxRegMon node2))
506319 ]
507320 ]
508− where
509− buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [
510− ("NSend", [
511− ("nsend", testNSend nsend)
512− , ("Unsafe.nsend", testNSend Unsafe.nsend)
513− , ("nsendRemote", testNSend (nsendRemote nid))
514− , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))
515− ])
516− , ("Send", [
517− ("send", testSend send)
518− , ("Unsafe.send", testSend Unsafe.send)
519− , ("usend", testSend usend)
520− , ("Unsafe.usend", testSend Unsafe.usend)
521− , ("sendChan", testChan sendChan)
522− , ("Unsafe.sendChan", testChan Unsafe.sendChan)
523− ])
524− , ("Forward", [
525− ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))
526− , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))
527− ])
528− ]
529−
530− build :: LocalNode
531− -> LocalNode
532− -> [(String, [(String, (Maybe LocalNode -> Process ()))])]
533− -> [Test]
534− build n ln specs =
535− [ testGroup (intercalate "-" [groupName, caseSuffix]) [
536− testCase (intercalate "-" [caseName, caseSuffix])
537− (runProcess n (caseImpl caseNode))
538− | (caseName, caseImpl) <- groupCases
539− ]
540− | (groupName, groupCases) <- specs
541− , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]
542− ]
distributed-process-tests/src/Control/Distributed/Process/Tests/Receive.hs+8−7
@@ -10,12 +10,13 @@ import Network.Transport.Test (TestTransport(..))
1010
1111import Network.Transport (Transport)
1212import Control.Distributed.Process
13+import Control.Distributed.Process.Closure
1314import Control.Distributed.Process.Node
1415
1516import Control.Monad
1617
1718import Test.HUnit (Assertion, (@?=))
18−import Test.Framework (Test)
19+import Test.Framework (Test, defaultMain)
1920import Test.Framework.Providers.HUnit (testCase)
2021
2122-- Tests:
@@ -42,7 +43,7 @@ recTest2 :: ReceivePort ()
4243 -> SendPort String
4344 -> ReceivePort String -> ReceivePort String
4445 -> Process ()
45−recTest2 wait sync r1 _ = do
46+recTest2 wait sync r1 r2 = do
4647 forever $ do
4748 receiveChan wait
4849 r <- receiveWait
@@ -55,7 +56,7 @@ recTest3 :: ReceivePort ()
5556 -> SendPort String
5657 -> ReceivePort String -> ReceivePort String
5758 -> Process ()
58−recTest3 wait sync r1 _ = do
59+recTest3 wait sync r1 r2 = do
5960 forever $ do
6061 receiveChan wait
6162 r <- receiveWait
@@ -68,7 +69,7 @@ recTest4 :: ReceivePort ()
6869 -> SendPort String
6970 -> ReceivePort String -> ReceivePort String
7071 -> Process ()
71−recTest4 wait sync r1 _ = do
72+recTest4 wait sync r1 r2 = do
7273 forever $ do
7374 receiveChan wait
7475 r <- receiveWait
@@ -82,11 +83,11 @@ master :: Process ()
8283master = do
8384 (waits,waitr) <- newChan
8485 (syncs,syncr) <- newChan
85− let go expected = do
86+ let go expect = do
8687 sendChan waits ()
8788 r <- receiveChan syncr
88− liftIO $ print (r, expected, r == expected)
89− liftIO $ r @?= expected
89+ liftIO $ print (r,expect, r == expect)
90+ liftIO $ r @?= expect
9091
9192 liftIO $ putStrLn "---- Test 1 ----"
9293 (s1,r1) <- newChan
distributed-process-tests/src/Control/Distributed/Process/Tests/Stats.hs+7−11
@@ -29,7 +29,7 @@ import Test.Framework.Providers.HUnit (testCase)
2929
3030testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()
3131testLocalDeadProcessInfo result = do
32− pid <- spawnLocal $ do (_ :: String) <- expect; return ()
32+ pid <- spawnLocal $ do "finish" <- expect; return ()
3333 mref <- monitor pid
3434 send pid "finish"
3535 _ <- receiveWait [
@@ -52,10 +52,8 @@ testLocalLiveProcessInfo result = do
5252 link self
5353 mRef <- monitor self
5454 stash mon mRef
55− res <- expect
56− case res of
57− "die" -> return ()
58− _ -> die $ "unexpected message received: " ++ res
55+ "die" <- expect
56+ return ()
5957
6058 monRef <- liftIO $ takeMVar mon
6159
@@ -86,9 +84,9 @@ testRemoteLiveProcessInfo TestTransport{..} node1 = do
8684 -- our send op shouldn't overtake link or monitor requests AFAICT
8785 -- so a little table tennis should get us synchronised properly
8886 send serverPid (self, "ping")
89− pong <- expect
87+ "pong" <- expect
9088 pInfo <- getProcessInfo serverPid
91− stash result $ pong == "pong" && pInfo /= Nothing
89+ stash result $ pInfo /= Nothing
9290 where
9391 launchRemote :: MVar ProcessId -> IO ()
9492 launchRemote locMV = do
@@ -99,10 +97,8 @@ testRemoteLiveProcessInfo TestTransport{..} node1 = do
9997 _ <- receiveWait [
10098 match (\(pid, "ping") -> send pid "pong")
10199 ]
102− res <- expect
103− case res of
104− "stop" -> return ()
105− _ -> die $ "unexpected message received: " ++ res
100+ "stop" <- expect
101+ return ()
106102 return ()
107103
108104 withActiveRemote :: LocalNode
distributed-process-tests/src/Control/Distributed/Process/Tests/Tracing.hs+11−23
@@ -125,11 +125,9 @@ testTraceSending result = do
125125 (\ev ->
126126 case ev of
127127 (MxSent to from msg) -> do
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..."
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)
133131 _ ->
134132 return ()) $ do
135133 send pid "hello there"
@@ -248,8 +246,12 @@ testRemoteTraceRelay TestTransport{..} result =
248246 -- Here we set up that relay, and then wait for a signal
249247 -- that the tracer (on node1) has seen the expected
250248 -- 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 ())
251254
252− pid <- splinchLogger node2 mvNid
253255 nid <- liftIO $ takeMVar mvNid
254256 mref <- monitor pid
255257 observedPid <- liftIO $ newEmptyMVar
@@ -289,17 +291,6 @@ testRemoteTraceRelay TestTransport{..} result =
289291 -- and just to be polite...
290292 liftIO $ closeLocalNode node2
291293
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−
303294-- | Sets the value of an environment variable while executing the given IO
304295-- computation and restores the preceeding value upon completion.
305296withEnv :: String -> String -> IO a -> IO a
@@ -328,12 +319,9 @@ testSystemLoggerMsg t action interestingMessage =
328319 -- Wait for the trace message.
329320 receiveWait [ matchIf interestingMessage' $ const $ return () ]
330321 -- Only one interesting message should arrive.
331− expectedTimeout <- receiveTimeout
332− 100000
333− [ matchIf interestingMessage' $ const $ return () ]
334− case expectedTimeout of
335− Nothing -> return ()
336− Just _ -> die "Unexpected message arrived..."
322+ Nothing <- receiveTimeout 100000
323+ [ matchIf interestingMessage' $ const $ return () ]
324+ return ()
337325
338326
339327-- | Tests that one and only one trace message is produced when a message is
distributed-process-tests/tests/runInMemory.hs+1
@@ -8,6 +8,7 @@ import Network.Transport.Test (TestTransport(..))
88import Network.Transport.InMemory
99import Test.Framework (defaultMainWithArgs)
1010
11+import Control.Concurrent (threadDelay)
1112import System.Environment (getArgs)
1213
1314main :: IO ()
distributed-process-tests/tests/runTCP.hs+2−2
@@ -10,7 +10,6 @@ import Network.Transport.TCP
1010 ( createTransportExposeInternals
1111 , TransportInternals(socketBetween)
1212 , defaultTCPParameters
13− , defaultTCPAddr
1413 , TCPParameters(..)
1514 )
1615import Test.Framework (defaultMainWithArgs)
@@ -25,7 +24,8 @@ main = do
2524 hSetBuffering stdout LineBuffering
2625 hSetBuffering stderr LineBuffering
2726 Right (transport, internals) <-
28− createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "8080")
27+ createTransportExposeInternals "127.0.0.1" "8080"
28+ (\sn -> ("127.0.0.1", sn))
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: Tim Watson <watson.timothy@gmail.com>
9+Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>
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.10.3 GHC==8.0.2 GHC==8.2.2 GHC==8.4.4
24+Tested-With: GHC==7.6.3 GHC==7.8.4 GHC==7.10.3 GHC==8.0.1
2525Category: Control
2626extra-source-files: ChangeLog
2727
src/Control/Distributed/Process/Internal/Primitives.hs+43−59
@@ -254,15 +254,17 @@ 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.
257266 liftIO $ traceEvent (localEventBus node)
258267 (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
266268
267269-- | /Unsafe/ variant of 'send'. This function makes /no/ attempt to serialize
268270-- and (in the case when the destination process resides on the same local
@@ -282,12 +284,9 @@ unsafeSend = Unsafe.send
282284--
283285usend :: Serializable a => ProcessId -> a -> Process ()
284286usend them msg = do
285− proc <- ask
287+ here <- getSelfNode
286288 let there = processNodeId them
287− let (us, node) = (processId proc, processNode proc)
288− let msg' = wrapMessage msg
289− liftIO $ traceEvent (localEventBus node) (MxSent them us msg')
290− if localNodeId (processNode proc) == there
289+ if here == there
291290 then sendLocal them msg
292291 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
293292 (createMessage msg)
@@ -307,13 +306,7 @@ expect = receiveWait [match return]
307306-- Channels --
308307--------------------------------------------------------------------------------
309308
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−--
309+-- | Create a new typed channel
317310newChan :: Serializable a => Process (SendPort a, ReceivePort a)
318311newChan = do
319312 proc <- ask
@@ -341,16 +334,13 @@ newChan = do
341334sendChan :: Serializable a => SendPort a -> a -> Process ()
342335sendChan (SendPort cid) msg = do
343336 proc <- ask
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
337+ let node = localNodeId (processNode proc)
338+ destNode = processNodeId (sendPortProcessId cid) in do
339+ case destNode == node of
350340 True -> sendChanLocal cid msg
351341 False -> do
352− liftIO $ sendBinary node
353− (ProcessIdentifier pid)
342+ liftIO $ sendBinary (processNode proc)
343+ (ProcessIdentifier (processId proc))
354344 (SendPortIdentifier cid)
355345 NoImplicitReconnect
356346 msg
@@ -411,11 +401,8 @@ newtype Match b = Match { unMatch :: MatchOn Message (Process b) }
411401receiveWait :: [Match b] -> Process b
412402receiveWait ms = do
413403 queue <- processQueue <$> ask
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."
404+ Just proc <- liftIO $ dequeue queue Blocking (map unMatch ms)
405+ proc
419406
420407-- | Like 'receiveWait' but with a timeout.
421408--
@@ -494,14 +481,17 @@ forward msg them = do
494481 us = processId proc
495482 nid = localNodeId node
496483 destNode = (processNodeId them) in do
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)
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)
505495
506496-- | Forward a raw 'Message' to the given 'ProcessId'.
507497--
@@ -515,10 +505,14 @@ uforward msg them = do
515505 us = processId proc
516506 nid = localNodeId node
517507 destNode = (processNodeId them) in do
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
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)
522516
523517-- | Wrap a 'Serializable' value in a 'Message'. Note that 'Message's are
524518-- 'Serializable' - like the datum they contain - but also note, deserialising
@@ -1227,12 +1221,8 @@ whereisRemoteAsync nid label = do
12271221
12281222-- | Named send to a process in the local registry (asynchronous)
12291223nsend :: Serializable a => String -> a -> Process ()
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')
1224+nsend label msg =
1225+ sendCtrlMsg Nothing (NamedSend label (createUnencodedMessage msg))
12361226
12371227-- | Named send to a process in the local registry (asynchronous).
12381228-- This function makes /no/ attempt to serialize and (in the case when the
@@ -1244,15 +1234,9 @@ unsafeNSend = Unsafe.nsend
12441234-- | Named send to a process in a remote registry (asynchronous)
12451235nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
12461236nsendRemote nid label msg = do
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))
1237+ here <- getSelfNode
1238+ if here == nid then nsend label msg
1239+ else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
12561240
12571241-- | Named send to a process in a remote registry (asynchronous)
12581242-- This function makes /no/ attempt to serialize and (in the case when the
src/Control/Distributed/Process/Management.hs+2−34
@@ -76,40 +76,6 @@
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−--
11379-- [Management Data API]
11480--
11581-- Both management agents and clients of the API have access to a variety of
@@ -178,6 +144,8 @@
178144-- > monitorNames = getSelfPid >>= nsend "name-monitor"
179145-- > monitorNames2 = getSelfPid >>= mxNotify
180146--
147+-- For some real-world examples, see the distributed-process-platform package.
148+--
181149-- [Performance, Stablity and Scalability]
182150--
183151-- /Management Agents/ offer numerous advantages over regular processes:
src/Control/Distributed/Process/Management/Internal/Trace/Primitives.hs+2−8
@@ -33,7 +33,6 @@ import Control.Distributed.Process.Internal.Primitives
3333 ( whereis
3434 , newChan
3535 , receiveChan
36− , die
3736 )
3837import Control.Distributed.Process.Management.Internal.Trace.Types
3938 ( TraceArg(..)
@@ -169,11 +168,6 @@ withRegisteredTracer act = do
169168 withLocalTracer $ \t -> liftIO $ Tracer.getCurrentTraceClient t sp
170169 currentTracer <- receiveChan rp
171170 case currentTracer of
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)"
171+ Nothing -> do { (Just p') <- whereis "tracer.initial"; act p' }
179172 (Just p) -> act p
173+
src/Control/Distributed/Process/Management/Internal/Types.hs−7
@@ -21,7 +21,6 @@ import Control.Concurrent.STM
2121import Control.Distributed.Process.Internal.Types
2222 ( Process
2323 , ProcessId
24− , SendPortId
2524 , Message
2625 , DiedReason
2726 , NodeId
@@ -58,14 +57,8 @@ data MxEvent =
5857 -- ^ fired whenever a node /dies/ (i.e., the connection is broken/disconnected)
5958 | MxSent ProcessId ProcessId Message
6059 -- ^ 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
6560 | MxReceived ProcessId Message
6661 -- ^ 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
6962 | MxConnected ConnectionId EndPointAddress
7063 -- ^ fired when a network-transport connection is first established
7164 | MxDisconnected ConnectionId EndPointAddress
src/Control/Distributed/Process/Node.hs+60−20
@@ -186,8 +186,15 @@ import Control.Distributed.Process.Internal.Types
186186import Control.Distributed.Process.Management.Internal.Agent
187187 ( mxAgentController
188188 )
189−import Control.Distributed.Process.Management.Internal.Types
190− ( MxEvent(..)
189+import Control.Distributed.Process.Management
190+ ( mxAgent
191+ , mxSink
192+ , liftMX
193+ , mxUpdateLocal
194+ , mxGetLocal
195+ , mxReady
196+ , MxEvent(..)
197+ , MxAgentId(..)
191198 )
192199import qualified Control.Distributed.Process.Management.Internal.Trace.Remote as Trace
193200 ( remoteTable
@@ -316,6 +323,44 @@ startDefaultTracer node' = do
316323
317324-- TODO: we need a better mechanism for defining and registering services
318325
326+{- note [registry monitoring agent]
327+ This agent listens for 'MxRegistered' and 'MxUnRegistered' events and tracks
328+ all labels for remote 'ProcessId's that are stored in the registry.
329+
330+ When a remote process is registered, the agent starts monitoring it until it
331+ is unregistered. We can safely ignore ProcessMonitorNotification's, since the
332+ node controller is obligated to issue an MxUnRegistered event for any labels
333+ connected to a dead process. Thus, our only responsibility here is to ensure
334+ that we set up monitors for locally registered processes, and tear them down
335+ once unregistration occurs.
336+ -}
337+
338+registryMonitorAgent :: Process ProcessId
339+registryMonitorAgent = do
340+ nid <- getSelfNode
341+ mxAgent registryMonitorAgentId (Map.empty :: Map String MonitorRef)
342+ [ mxSink $ \ev -> do
343+ case ev of
344+ MxRegistered pid label
345+ | processNodeId pid /= nid -> do
346+ mref <- liftMX $ monitor pid
347+ mxUpdateLocal (Map.insert label mref)
348+ MxUnRegistered pid label
349+ | processNodeId pid /= nid -> do
350+ mrefs <- mxGetLocal
351+ forM_ (label `Map.lookup` mrefs) $ \mref -> do
352+ liftMX $ unmonitorAsync mref
353+ mxUpdateLocal (Map.delete label)
354+ _ -> return ()
355+ mxReady
356+ -- [note: no need to remove async answers from mailbox]
357+ -- The framework simply discards any input you don't have a handler for.
358+ -- See Management.hs `runAgent` for details.
359+ ]
360+ where
361+ registryMonitorAgentId :: MxAgentId
362+ registryMonitorAgentId = MxAgentId "service.registry.monitoring"
363+
319364-- | Start and register the service processes on a node
320365startServiceProcesses :: LocalNode -> IO ()
321366startServiceProcesses node = do
@@ -492,7 +537,7 @@ type IncomingConnection = (NT.EndPointAddress, IncomingTarget)
492537data IncomingTarget =
493538 Uninit
494539 | ToProc ProcessId (Weak (CQueue Message))
495− | ToChan SendPortId TypedChannel
540+ | ToChan TypedChannel
496541 | ToNode
497542
498543data ConnectionState = ConnectionState {
@@ -547,17 +592,13 @@ handleIncomingMessages node = go initConnectionState
547592 enqueue queue msg -- 'enqueue' is strict
548593 trace node (MxReceived pid msg)
549594 go st
550− Just (_, ToChan chId (TypedChannel chan')) -> do
595+ Just (_, ToChan (TypedChannel chan')) -> do
551596 mChan <- deRefWeak chan'
552597 -- If mChan is Nothing, the process has given up the read end of
553598 -- the channel and we simply ignore the incoming message
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'
599+ forM_ mChan $ \chan -> atomically $
600+ -- We make sure the message is fully decoded when it is enqueued
601+ writeTQueue chan $! decode (BSL.fromChunks payload)
561602 go st
562603 Just (_, ToNode) -> do
563604 let ctrlMsg = decode . BSL.fromChunks $ payload
@@ -584,7 +625,7 @@ handleIncomingMessages node = go initConnectionState
584625 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
585626 case mChannel of
586627 Just channel ->
587− go (incomingAt cid ^= Just (src, ToChan chId channel) $ st)
628+ go (incomingAt cid ^= Just (src, ToChan channel) $ st)
588629 Nothing ->
589630 invalidRequest cid st $
590631 "incoming attempt to connect to unknown channel of"
@@ -1054,12 +1095,11 @@ ncEffectLocalPortSend from msg = do
10541095 -- If ch is Nothing, the process has given up the read end of
10551096 -- the channel and we simply ignore the incoming message - this
10561097 ch <- deRefWeak 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
1098+ forM_ ch $ \chan -> deliverChan msg chan
1099+ where deliverChan :: forall a . Message -> TQueue a -> IO ()
1100+ deliverChan (UnencodedMessage _ raw) chan' =
10601101 atomically $ writeTQueue chan' ((unsafeCoerce raw) :: a)
1061− trace n (MxReceivedPort p $ unsafeCreateUnencodedMessage raw)
1062− deliverChan _ _ (EncodedMessage _ _) _ =
1102+ deliverChan (EncodedMessage _ _) _ =
10631103 -- this will not happen unless someone screws with Primitives.hs
10641104 error "invalid local channel delivery"
10651105
@@ -1089,7 +1129,7 @@ ncEffectGetInfo from pid =
10891129 case mProc of
10901130 Nothing -> dispatch (isLocal node (ProcessIdentifier from))
10911131 from (ProcessInfoNone DiedUnknownId)
1092− Just proc -> do
1132+ Just proc -> do
10931133 itsLinks <- Set.map fst . BiMultiMap.lookupBy1st them <$>
10941134 gets (^. links)
10951135 itsMons <- BiMultiMap.lookupBy1st them <$> gets (^. monitors)
@@ -1103,8 +1143,8 @@ ncEffectGetInfo from pid =
11031143 infoNode = (processNodeId pid)
11041144 , infoRegisteredNames = reg
11051145 , infoMessageQueueLength = size
1106− , infoMonitors = Set.toList itsMons
1107− , infoLinks = Set.toList itsLinks
1146+ , infoMonitors = Set.toList itsMons
1147+ , infoLinks = Set.toList itsLinks
11081148 }
11091149 where dispatch :: (Serializable a)
11101150 => Bool
src/Control/Distributed/Process/UnsafePrimitives.hs+33−82
@@ -32,24 +32,7 @@
3232-- the /normal/ strategy).
3333--
3434-- Use of the functions in this module can potentially change the runtime
35−-- behaviour of your application. In addition, messages passed between Cloud
36−-- Haskell processes are written to a tracing infrastructure on the local node,
37−-- to provide improved introspection and debugging facilities for complex actor
38−-- based systems. This module makes no attempt to force evaluation in these
39−-- cases either, thus evaluation problems in passed data structures could not
40−-- only crash your processes, but could also bring down critical internal
41−-- services on which the node relies to function correctly.
42−--
43−-- If you wish to repudiate such issues, you are advised to consider the use
44−-- of NFSerialisable in the distributed-process-extras package, which type
45−-- class brings NFData into scope along with Serializable, such that we can
46−-- force evaluation. Intended for use with modules such as this one, this
47−-- approach guarantees correct evaluatedness in terms of @NFData@. Please note
48−-- however, that we /cannot/ guarantee that an @NFData@ instance will behave the
49−-- same way as a @Binary@ one with regards evaluation, so it is still possible
50−-- to introduce unexpected behaviour by using /unsafe/ primitives in this way.
51−--
52−-- You have been warned!
35+-- behaviour of your application. You have been warned!
5336--
5437-- This module is exported so that you can replace the use of Cloud Haskell's
5538-- /safe/ messaging primitives. If you want to use both variants, then you can
@@ -72,12 +55,7 @@ import Control.Distributed.Process.Internal.Messaging
7255 , sendBinary
7356 , sendCtrlMsg
7457 )
75−import Control.Distributed.Process.Management.Internal.Types
76− ( MxEvent(..)
77− )
78−import Control.Distributed.Process.Management.Internal.Trace.Types
79− ( traceEvent
80− )
58+
8159import Control.Distributed.Process.Internal.Types
8260 ( ProcessId(..)
8361 , NodeId(..)
@@ -101,48 +79,34 @@ import Control.Monad.Reader (ask)
10179
10280-- | Named send to a process in the local registry (asynchronous)
10381nsend :: Serializable a => String -> a -> Process ()
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')
82+nsend label msg =
83+ sendCtrlMsg Nothing (NamedSend label (unsafeCreateUnencodedMessage msg))
11284
11385-- | Named send to a process in a remote registry (asynchronous)
11486nsendRemote :: Serializable a => NodeId -> String -> a -> Process ()
11587nsendRemote nid label msg = do
11688 proc <- ask
117− let us = processId proc
118− let node = processNode proc
119− if localNodeId node == nid
89+ if localNodeId (processNode proc) == nid
12090 then nsend label msg
121− else
122− let lbl = label ++ "@" ++ show nid in do
123− -- see [note: tracing] NB: this is a remote call to another NC...
124− liftIO $ traceEvent (localEventBus node)
125− (MxSentToName lbl us (wrapMessage msg))
126− sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
91+ else sendCtrlMsg (Just nid) (NamedSend label (createMessage msg))
12792
12893-- | Send a message
12994send :: Serializable a => ProcessId -> a -> Process ()
13095send them msg = do
13196 proc <- ask
13297 let node = localNodeId (processNode proc)
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
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')
146110
147111-- | Send a message unreliably.
148112--
@@ -157,43 +121,30 @@ usend :: Serializable a => ProcessId -> a -> Process ()
157121usend them msg = do
158122 proc <- ask
159123 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')
164124 if localNodeId (processNode proc) == there
165− then sendCtrlMsg Nothing $ LocalSend them msg'
125+ then sendCtrlMsg Nothing $
126+ LocalSend them (unsafeCreateUnencodedMessage msg)
166127 else sendCtrlMsg (Just there) $ UnreliableSend (processLocalId them)
167128 (createMessage msg)
168129
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−
177130-- | Send a message on a typed channel
178131sendChan :: Serializable a => SendPort a -> a -> Process ()
179132sendChan (SendPort cid) msg = do
180133 proc <- ask
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
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
194144 where
195− unsafeSendChanLocal :: SendPortId -> Message -> Process ()
196− unsafeSendChanLocal p m = sendCtrlMsg Nothing $ LocalPortSend p m
145+ unsafeSendChanLocal :: (Serializable a) => SendPortId -> a -> Process ()
146+ unsafeSendChanLocal spId msg' =
147+ sendCtrlMsg Nothing $ LocalPortSend spId (unsafeCreateUnencodedMessage msg')
197148
198149-- | Create an unencoded @Message@ for any @Serializable@ type.
199150wrapMessage :: Serializable a => a -> Message
stack-ghc-7.10.3.yaml+1−5
@@ -3,18 +3,14 @@ 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
106
117extra-deps:
128- rank1dynamic-0.4.0
139- distributed-static-0.3.8
1410- network-transport-0.5.2
11+- network-transport-tcp-0.6.0
1512- network-transport-inmemory-0.5.2
1613- rematch-0.2.0.0
17−- exceptions-0.8.2.1 # test dependency
1814
1915flags:
2016 distributed-process-tests:
stack-ghc-8.0.2.yaml+2−8
@@ -1,22 +1,16 @@
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 :/
1+resolver: lts-9.0
42
53packages:
64- '.'
75- distributed-process-tests/
8−- location:
9− git: https://github.com/haskell-distributed/network-transport-tcp.git
10− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
11− extra-dep: true
126
137extra-deps:
148- rank1dynamic-0.4.0
159- distributed-static-0.3.8
1610- network-transport-0.5.2
11+- network-transport-tcp-0.6.0
1712- network-transport-inmemory-0.5.2
1813- rematch-0.2.0.0
19−# - exceptions-0.8.2.1 # test dependency
2014
2115flags:
2216 distributed-process-tests:
stack-ghc-8.2.2.yamldeleted−21
@@ -1,21 +0,0 @@
1−resolver: lts-11.22
2−
3−packages:
4−- '.'
5−- distributed-process-tests/
6−- location:
7− git: https://github.com/haskell-distributed/network-transport-tcp.git
8− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
9− extra-dep: true
10−
11−extra-deps:
12−- rank1dynamic-0.4.0
13−- distributed-static-0.3.8
14−- network-transport-0.5.2
15−- network-transport-inmemory-0.5.2
16−- rematch-0.2.0.0
17−# - exceptions-0.8.2.1 # test dependency
18−
19−flags:
20− distributed-process-tests:
21− tcp: true
stack-ghc-8.4.4.yamldeleted−20
@@ -1,20 +0,0 @@
1−resolver: lts-12.18
2−
3−packages:
4−- '.'
5−- distributed-process-tests/
6−- location:
7− git: https://github.com/haskell-distributed/network-transport-tcp.git
8− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
9− extra-dep: true
10−
11−extra-deps:
12−- rank1dynamic-0.4.0
13−- distributed-static-0.3.8
14−- network-transport-0.5.2
15−- network-transport-inmemory-0.5.2
16−- rematch-0.2.0.0
17−
18−flags:
19− distributed-process-tests:
20− tcp: true
stack-nightly.yamldeleted−26
@@ -1,26 +0,0 @@
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+2−7
@@ -1,19 +1,14 @@
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
1+resolver: nightly-2017-08-20
42
53packages:
64- '.'
75- distributed-process-tests/
8−- location:
9− git: https://github.com/haskell-distributed/network-transport-tcp.git
10− commit: d87d6f55697a94a1fbf211ff9c1bb769a1e129cd
11− extra-dep: true
126
137extra-deps:
148- rank1dynamic-0.4.0
159- distributed-static-0.3.8
1610- network-transport-0.5.2
11+- network-transport-tcp-0.6.0
1712- network-transport-inmemory-0.5.2
1813- rematch-0.2.0.0
1914