distributed-process-tests 0.4.11 → 0.4.12
raw patch · 10 files changed
+1086/−442 lines, 10 filesdep +exceptionsdep ~basedep ~bytestringdep ~networknew-uploaderPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: exceptions
Dependency ranges changed: base, bytestring, network, network-transport-tcp, random
API changes (from Hackage documentation)
+ Control.Distributed.Process.Tests.Internal.Utils: instance Data.Binary.Class.Binary Control.Distributed.Process.Tests.Internal.Utils.Private
+ Control.Distributed.Process.Tests.Internal.Utils: instance GHC.Generics.Generic Control.Distributed.Process.Tests.Internal.Utils.Private
+ Control.Distributed.Process.Tests.Internal.Utils: pause :: Int -> Process ()
- Control.Distributed.Process.Tests.Internal.Utils: assertComplete :: (Eq a) => String -> MVar a -> a -> IO ()
+ Control.Distributed.Process.Tests.Internal.Utils: assertComplete :: Eq a => String -> MVar a -> a -> IO ()
- Control.Distributed.Process.Tests.Internal.Utils: delayedAssertion :: (Eq a) => String -> LocalNode -> a -> (TestResult a -> Process ()) -> Assertion
+ Control.Distributed.Process.Tests.Internal.Utils: delayedAssertion :: Eq a => String -> LocalNode -> a -> (TestResult a -> Process ()) -> Assertion
Files
- distributed-process-tests.cabal +65/−43
- src/Control/Distributed/Process/Tests/CH.hs +456/−234
- src/Control/Distributed/Process/Tests/Closure.hs +86/−48
- src/Control/Distributed/Process/Tests/Internal/Utils.hs +18/−7
- src/Control/Distributed/Process/Tests/Mx.hs +390/−53
- src/Control/Distributed/Process/Tests/Receive.hs +33/−34
- src/Control/Distributed/Process/Tests/Stats.hs +11/−7
- src/Control/Distributed/Process/Tests/Tracing.hs +23/−11
- tests/runInMemory.hs +0/−1
- tests/runTCP.hs +4/−4
distributed-process-tests.cabal view
@@ -1,16 +1,16 @@ name: distributed-process-tests-version: 0.4.11+version: 0.4.12 synopsis: Tests and test support tools for distributed-process. homepage: http://github.com/haskell-distributed/distributed-process/tree/master/distributed-process-tests description: Tests and test suite for Cloud Haskell libraries. license: BSD3 license-file: LICENSE Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer: Facundo Domínguez <facundo.dominguez@tweag.io>+Maintainer: Tim Watson <watson.timothy@gmail.com> copyright: Well-Typed LLP category: Control, Cloud Haskell build-type: Simple-cabal-version: >=1.8+cabal-version: >=1.10 flag tcp Description: build and run TCP tests@@ -25,32 +25,34 @@ Control.Distributed.Process.Tests.Stats Control.Distributed.Process.Tests.Tracing Control.Distributed.Process.Tests.Internal.Utils- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, ansi-terminal >= 0.5, binary >= 0.5 && < 0.9,- bytestring >= 0.9 && < 0.12,+ bytestring >= 0.9 && < 0.13, distributed-process >= 0.6.0 && < 0.8, distributed-static,+ exceptions >= 0.5, HUnit >= 1.2 && < 1.7, network-transport >= 0.4.1.0 && < 0.6,- network >= 2.5 && < 2.7,- random >= 1.0 && < 1.2,+ network >= 2.5 && < 3.2,+ random >= 1.0 && < 1.3, rematch >= 0.1.2.1, setenv >= 0.1.1.3, test-framework >= 0.6 && < 0.9, test-framework-hunit >= 0.2.0 && < 0.4, stm hs-source-dirs: src+ default-language: Haskell98 ghc-options: -Wall -fno-warn-unused-do-bind- extensions: CPP,- ExistentialQuantification,- FlexibleInstances,- DeriveDataTypeable,- DeriveGeneric,- GeneralizedNewtypeDeriving,- RankNTypes,- RecordWildCards,- ScopedTypeVariables+ default-extensions: CPP,+ ExistentialQuantification,+ FlexibleInstances,+ DeriveDataTypeable,+ DeriveGeneric,+ GeneralizedNewtypeDeriving,+ RankNTypes,+ RecordWildCards,+ ScopedTypeVariables if impl(ghc <= 7.4.2) Build-Depends: ghc-prim == 0.2.0.0 @@ -58,14 +60,15 @@ Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, distributed-process-tests,- network >= 2.3 && < 2.7,+ network >= 2.3 && < 3.2, network-transport >= 0.4.1.0 && < 0.6, network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9- Extensions: CPP- ghc-options: -Wall -threaded -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests Test-Suite TestCHInTCP@@ -73,16 +76,17 @@ Main-Is: runTCP.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH if flag(tcp)- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, distributed-process-tests,- network >= 2.3 && < 2.7,+ network >= 2.5 && < 3.2, network-transport >= 0.4.1.0 && < 0.6,- network-transport-tcp >= 0.5 && < 0.7,+ network-transport-tcp >= 0.5 && < 0.9, test-framework >= 0.6 && < 0.9 else Buildable: False- Extensions: CPP- ghc-options: -Wall -threaded -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests @@ -90,13 +94,14 @@ Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, distributed-process-tests,- network >= 2.3 && < 2.7,+ network >= 2.3 && < 3.2, network-transport >= 0.4.1.0 && < 0.6, network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9- Extensions: CPP+ default-extensions: CPP+ default-language: Haskell98 ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests @@ -104,41 +109,58 @@ Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, distributed-process-tests,- network >= 2.3 && < 2.7,+ network >= 2.3 && < 3.2, network-transport >= 0.4.1.0 && < 0.6, network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9- Extensions: CPP- ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests --Test-Suite TestMx+Test-Suite TestMxInMemory Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, distributed-process-tests,- network >= 2.3 && < 2.7,+ network >= 2.3 && < 3.2, network-transport >= 0.4.1.0 && < 0.6, network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9- Extensions: CPP- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests -Test-Suite TestTracing+Test-Suite TestTracingInMemory Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.9 && < 5, distributed-process-tests,- network >= 2.3 && < 2.7,+ network >= 2.3 && < 3.2, network-transport >= 0.4.1.0 && < 0.6, network-transport-inmemory >= 0.5, test-framework >= 0.6 && < 0.9- Extensions: CPP- ghc-options: -Wall -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests++Test-Suite TestMxInTCP+ Type: exitcode-stdio-1.0+ Main-Is: runInMemory.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx+ Build-Depends: base >= 4.9 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 3.2,+ network-transport >= 0.4.1.0 && < 0.6,+ network-transport-inmemory >= 0.5,+ test-framework >= 0.6 && < 0.9+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests
src/Control/Distributed/Process/Tests/CH.hs view
@@ -25,24 +25,30 @@ ) import Control.Monad (replicateM_, replicateM, forever, void, unless, join) import Control.Exception (SomeException, throwIO, ErrorCall(..))-import qualified Control.Exception as Ex (catch)-import Control.Applicative ((<$>), (<*>), pure, (<|>))+import Control.Monad.Catch (try, catch, finally, mask, onException)+import Control.Applicative ((<|>)) import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)-import Control.Distributed.Process+import Control.Distributed.Process hiding+ ( try+ , catch+ , finally+ , mask+ , onException+ ) import Control.Distributed.Process.Internal.Types- ( NodeId(nodeAddress)- , LocalNode(localEndPoint)+ ( LocalNode(localEndPoint) , ProcessExitException(..) , nullProcessId , createUnencodedMessage ) import Control.Distributed.Process.Node+import Control.Distributed.Process.Tests.Internal.Utils (pause) import Control.Distributed.Process.Serializable (Serializable)--import Test.HUnit (Assertion, assertFailure)+import Data.Maybe (isNothing, isJust)+import Test.HUnit (Assertion, assertBool, assertFailure) import Test.Framework (Test, testGroup) import Test.Framework.Providers.HUnit (testCase)-import Control.Rematch hiding (match)+import Control.Rematch hiding (match, isNothing, isJust) import Control.Rematch.Run (Match(..)) newtype Ping = Ping ProcessId@@ -65,7 +71,7 @@ forkTry :: IO () -> IO ThreadId forkTry p = do tid <- myThreadId- forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))+ forkIO $ catch p (\e -> throwTo tid (e :: SomeException)) -- | The ping server from the paper ping :: Process ()@@ -75,13 +81,39 @@ send partner (Ping self) ping +verifyClient :: String -> MVar Bool -> IO ()+verifyClient s b = takeMVar b >>= assertBool s++expectPing :: MVar Bool -> Process ()+expectPing mv = expect >>= liftIO . putMVar mv . checkPing+ where+ checkPing (Ping _) = True+ -- | Quick and dirty synchronous version of whereisRemoteAsync whereisRemote :: NodeId -> String -> Process (Maybe ProcessId) whereisRemote nid string = do whereisRemoteAsync nid string- WhereIsReply _ mPid <- expect- return mPid+ receiveWait [+ match (\(WhereIsReply _ mPid) -> return mPid)+ ] +verifyWhereIsRemote :: NodeId -> String -> Process ProcessId+verifyWhereIsRemote n s = whereisRemote n s >>= maybe (die "remote name not found") return++syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()+syncBreakConnection breakConnection nid0 nid1 = do+ m <- newEmptyMVar+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m+ runProcess nid0 $ do+ them <- liftIO $ takeMVar m+ pinger <- spawnLocal $ forever $ send them ()+ _ <- monitorNode (localNodeId nid1)+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)+ (nodeAddress $ localNodeId nid1)+ NodeMonitorNotification _ _ _ <- expect+ kill pinger "finished"+ return ()+ data Add = Add ProcessId Double Double deriving (Typeable) data Divide = Divide ProcessId Double Double deriving (Typeable) data DivByZero = DivByZero deriving (Typeable)@@ -155,12 +187,18 @@ unmonitor ref liftIO $ putMVar done () (False, ref) -> do- ProcessMonitorNotification ref' pid reason' <- expect- True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'- liftIO $ putMVar done ()+ receiveWait [+ match (\(ProcessMonitorNotification ref' pid reason') -> do+ liftIO $ do+ assertBool "Bad Monitor Signal"+ (Just ref' == ref && pid == theirAddr &&+ mOrL && reason == reason')+ putMVar done ())+ ] ) (\(ProcessLinkException pid reason') -> do- True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'+ (liftIO $ assertBool "link exception unmatched" $+ pid == theirAddr && not mOrL && not un && reason == reason') liftIO $ putMVar done () ) @@ -191,8 +229,10 @@ pid <- getSelfPid replicateM_ numPings $ do send pingServer (Pong pid)- Ping _ <- expect- return ()+ p <- expectTimeout 3000000+ case p of+ Just (Ping _) -> return ()+ Nothing -> die "Failed to receive Ping" putMVar clientDone () @@ -279,7 +319,7 @@ runProcess localNode $ do monitor theirAddr -- wait for the process to die- ProcessMonitorNotification _ _ _ <- expect+ expect :: Process ProcessMonitorNotification monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done takeMVar done@@ -353,14 +393,17 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- 3 <- expect :: Process Double+ three <- expect :: Process Double send mathServer (Divide pid 8 2)- 4 <- expect :: Process Double+ four <- expect :: Process Double send mathServer (Divide pid 8 0)- DivByZero <- expect- liftIO $ putMVar clientDone ()+ divByZ <- expect+ liftIO $ putMVar clientDone (three, four, divByZ) - takeMVar clientDone+ res <- takeMVar clientDone+ case res of+ (3, 4, DivByZero) -> return ()+ _ -> error $ "Something went horribly wrong" -- | Send first message (i.e. connect) to an already terminated process -- (without monitoring); then send another message to a second process on@@ -388,11 +431,9 @@ pid <- getSelfPid send server1 "Hi" send server2 (Pong pid)- Ping pid' <- expect- True <- return $ pid' == server2- liftIO $ putMVar clientDone ()+ expectPing clientDone - takeMVar clientDone+ verifyClient "Expected Ping from server" clientDone -- | Test (non-zero) timeout testTimeout :: TestTransport -> Assertion@@ -401,10 +442,10 @@ done <- newEmptyMVar runProcess localNode $ do- Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]- liftIO $ putMVar done ()+ res <- receiveTimeout 1000000 [match (\Add{} -> return ())]+ liftIO $ putMVar done $ res == Nothing - takeMVar done+ verifyClient "Expected receiveTimeout to timeout..." done -- | Test zero timeout testTimeout0 :: TestTransport -> Assertion@@ -432,10 +473,9 @@ -- is not interested in, and then a single message that it wants replicateM_ 10000 $ send server "Irrelevant message" send server (Pong pid)- Ping _ <- expect- liftIO $ putMVar clientDone ()+ expectPing clientDone - takeMVar clientDone+ verifyClient "Expected Ping from server" clientDone -- | Test typed channels testTypedChannels :: TestTransport -> Assertion@@ -458,10 +498,10 @@ runProcess localNode $ do (clientSendPort, rport) <- newChan sendChan serverSendPort (clientSendPort, 5)- False <- receiveChan rport- liftIO $ putMVar clientDone ()+ ch <- receiveChan rport+ liftIO $ putMVar clientDone $ ch == False - takeMVar clientDone+ verifyClient "Expected channel to send 'False'" clientDone -- | Test merging receive ports testMergeChannels :: TestTransport -> Assertion@@ -484,9 +524,8 @@ rs <- mapM charChannel "abc" m <- mergePorts biased rs xs <- replicateM 9 $ receiveChan m- True <- return $ xs == expected- liftIO $ putMVar done ()- takeMVar done+ liftIO $ putMVar done $ xs == expected+ verifyClient "Expected single layer merge to match expected ordering" done -- Two layers of merging testNested :: LocalNode -> Bool -> Bool -> String -> IO ()@@ -497,9 +536,8 @@ ms <- mapM (mergePorts biasedInner) rss m <- mergePorts biasedOuter ms xs <- replicateM (9 * 3) $ receiveChan m- True <- return $ xs == expected- liftIO $ putMVar done ()- takeMVar done+ liftIO $ putMVar done $ xs == expected+ verifyClient "Expected nested channels to match expeted ordering" done -- Test that if no messages are (immediately) available, the scheduler makes no difference testBlocked :: LocalNode -> Bool -> IO ()@@ -508,43 +546,45 @@ done <- newEmptyMVar forkProcess localNode $ do- [sa, sb, sc] <- liftIO $ mapM readMVar vs- mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)- [ -- a, b, c- (sa, 'a')- , (sb, 'b')- , (sc, 'c')- -- a, c, b- , (sa, 'a')- , (sc, 'c')- , (sb, 'b')- -- b, a, c- , (sb, 'b')- , (sa, 'a')- , (sc, 'c')- -- b, c, a- , (sb, 'b')- , (sc, 'c')- , (sa, 'a')- -- c, a, b- , (sc, 'c')- , (sa, 'a')- , (sb, 'b')- -- c, b, a- , (sc, 'c')- , (sb, 'b')- , (sa, 'a')- ]+ ss <- liftIO $ mapM readMVar vs+ case ss of+ [sa, sb, sc] ->+ mapM_ ((>> pause 10000) . uncurry sendChan)+ [ -- a, b, c+ (sa, 'a')+ , (sb, 'b')+ , (sc, 'c')+ -- a, c, b+ , (sa, 'a')+ , (sc, 'c')+ , (sb, 'b')+ -- b, a, c+ , (sb, 'b')+ , (sa, 'a')+ , (sc, 'c')+ -- b, c, a+ , (sb, 'b')+ , (sc, 'c')+ , (sa, 'a')+ -- c, a, b+ , (sc, 'c')+ , (sa, 'a')+ , (sb, 'b')+ -- c, b, a+ , (sc, 'c')+ , (sb, 'b')+ , (sa, 'a')+ ]+ _ -> die "Something went horribly wrong" forkProcess localNode $ do (ss, rs) <- unzip <$> replicateM 3 newChan liftIO $ mapM_ (uncurry putMVar) $ zip vs ss m <- mergePorts biased rs xs <- replicateM (6 * 3) $ receiveChan m- True <- return $ xs == "abcacbbacbcacabcba"- liftIO $ putMVar done ()+ liftIO $ putMVar done $ xs == "abcacbbacbcacabcba" - takeMVar done+ verifyClient "Expected merged ports to match expected ordering" done mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a) mergePorts True = mergePortsBiased@@ -562,8 +602,9 @@ localNode <- newLocalNode testTransport initRemoteTable runProcess localNode $ do e <- try terminate :: Process (Either ProcessTerminationException ())- True <- return $ either show show e == show ProcessTerminationException- return ()+ if either show show e == show ProcessTerminationException+ then return ()+ else die "Unexpected result from terminate" testMonitorNode :: TestTransport -> Assertion testMonitorNode TestTransport{..} = do@@ -574,11 +615,12 @@ runProcess node2 $ do ref <- monitorNode (localNodeId node1)- NodeMonitorNotification ref' nid DiedDisconnect <- expect- True <- return $ ref == ref' && nid == localNodeId node1- liftIO $ putMVar done ()+ receiveWait [+ match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->+ return $ ref == ref' && nid == localNodeId node1)+ ] >>= liftIO . putMVar done - takeMVar done+ verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done testMonitorLiveNode :: TestTransport -> Assertion testMonitorLiveNode TestTransport{..} = do@@ -593,15 +635,16 @@ liftIO $ putMVar ready () liftIO $ takeMVar readyr send p ()- NodeMonitorNotification ref' nid _ <- expect- True <- return $ ref == ref' && nid == localNodeId node1- liftIO $ putMVar done ()+ receiveWait [+ match (\(NodeMonitorNotification ref' nid _) ->+ (return $ ref == ref' && nid == localNodeId node1))+ ] >>= liftIO . putMVar done takeMVar ready closeLocalNode node1 putMVar readyr () - takeMVar done+ verifyClient "Expected NodeMonitorNotification for LIVE node" done testMonitorChannel :: TestTransport -> Assertion testMonitorChannel TestTransport{..} = do@@ -611,18 +654,20 @@ pid <- forkProcess node1 $ do sport <- expect :: Process (SendPort ()) ref <- monitorPort sport- PortMonitorNotification ref' port' reason <- expect- -- reason might be DiedUnknownId if the receive port is GCed before the- -- monitor is established (TODO: not sure that this is reasonable)- return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)- liftIO $ putMVar gotNotification ()+ receiveWait [+ -- reason might be DiedUnknownId if the receive port is GCed before the+ -- monitor is established (TODO: not sure that this is reasonable)+ match (\(PortMonitorNotification ref' port' reason) ->+ return $ ref' == ref && port' == sendPortId sport &&+ (reason == DiedNormal || reason == DiedUnknownId))+ ] >>= liftIO . putMVar gotNotification runProcess node2 $ do (sport, _) <- newChan :: Process (SendPort (), ReceivePort ()) send pid sport liftIO $ threadDelay 100000 - takeMVar gotNotification+ verifyClient "Expected PortMonitorNotification" gotNotification testRegistry :: TestTransport -> Assertion testRegistry TestTransport{..} = do@@ -634,20 +679,28 @@ runProcess node $ do register "ping" pingServer- Just pid <- whereis "ping"- True <- return $ pingServer == pid+ whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer) us <- getSelfPid nsend "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'- Left (ProcessRegistrationException "dead" Nothing) <- try $ register "dead" deadProcess- Left (ProcessRegistrationException "ping" (Just x)) <- try $ register "ping" deadProcess- True <- return $ x == pingServer- Left (ProcessRegistrationException "dead" Nothing) <- try $ unregister "dead"+ receiveWait [+ matchIf (\(Ping pid') -> pingServer == pid') (const $ return ())+ ]+ checkRegException "dead" Nothing deadProcess+ checkRegException "ping" (Just pingServer) deadProcess+ try (unregister "dead") >>= checkReg "dead" Nothing liftIO $ putMVar done () takeMVar done + where+ checkRegException name pid dead =+ try (register name dead) >>= checkReg name pid++ checkReg _ _ res =+ case res of+ Left (ProcessRegistrationException _ _) -> return ()+ _ -> die $ "Unexpected Registration" ++ show res+ testRegistryRemoteProcess :: TestTransport -> Assertion testRegistryRemoteProcess TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable@@ -658,13 +711,13 @@ runProcess node2 $ do register "ping" pingServer- Just pid <- whereis "ping"- True <- return $ pingServer == pid+ whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer) us <- getSelfPid nsend "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'- liftIO $ putMVar done ()+ receiveWait [+ matchIf (\(Ping pid') -> pingServer == pid')+ (const $ liftIO $ putMVar done ())+ ] takeMVar done @@ -672,7 +725,6 @@ testRemoteRegistry TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable node2 <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar pingServer <- forkProcess node1 ping deadProcess <- forkProcess node1 (return ())@@ -685,32 +737,34 @@ "ping" == label' && pid == pingServer) (\(RegisterReply _ _ _) -> return ()) ] - Just pid <- whereisRemote nid1 "ping"- True <- return $ pingServer == pid+ pid <- verifyWhereIsRemote nid1 "ping"+ liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid+ us <- getSelfPid nsendRemote nid1 "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'+ receiveWait [+ match (\(Ping pid') -> return $ pingServer == pid')+ ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId" -- test that if process was not registered Nothing is returned -- in owner field. registerRemoteAsync nid1 "dead" deadProcess- receiveWait [- matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')- (\(RegisterReply _ _ _) -> return ()) ]+ receiveWait [ matchIf (\(RegisterReply label' _ _) -> "dead" == label')+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))+ ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"+ registerRemoteAsync nid1 "ping" deadProcess receiveWait [- matchIf (\(RegisterReply label' False (Just pid)) ->- "ping" == label' && pid == pingServer)- (\(RegisterReply _ _ _) -> return ()) ]+ matchIf (\(RegisterReply label' False mPid) ->+ "ping" == label' && isJust mPid)+ (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))+ ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"+ unregisterRemoteAsync nid1 "dead" receiveWait [- matchIf (\(RegisterReply label' False Nothing) ->- "dead" == label' && pid == pingServer)- (\(RegisterReply _ _ _) -> return ()) ]- liftIO $ putMVar done ()-- takeMVar done+ matchIf (\(RegisterReply label' _ _) -> "dead" == label')+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))+ ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply" testRemoteRegistryRemoteProcess :: TestTransport -> Assertion testRemoteRegistryRemoteProcess TestTransport{..} = do@@ -726,15 +780,15 @@ receiveWait [ matchIf (\(RegisterReply label' _ _) -> "ping" == label') (\(RegisterReply _ _ _) -> return ()) ]- Just pid <- whereisRemote nid1 "ping"- True <- return $ pingServer == pid+ pid <- verifyWhereIsRemote nid1 "ping"+ liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid us <- getSelfPid nsendRemote nid1 "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'- liftIO $ putMVar done ()+ receiveWait [+ match (\(Ping pid') -> return $ pingServer == pid')+ ] >>= liftIO . putMVar done - takeMVar done+ verifyClient "Expected Ping with ping server's ProcessId" done testSpawnLocal :: TestTransport -> Assertion testSpawnLocal TestTransport{..} = do@@ -748,15 +802,14 @@ sport <- expect sendChan sport (1234 :: Int) - sport <- spawnChannelLocal $ \rport -> do- (1234 :: Int) <- receiveChan rport- send us ()+ sport <- spawnChannelLocal $+ \rport -> (receiveChan rport :: Process Int) >>= send us send pid sport- () <- expect- liftIO $ putMVar done ()+ expect >>= liftIO . putMVar done - takeMVar done+ res <- takeMVar done+ assertBool "Expected 1234 :: Int" $ res == (1234 :: Int) testSpawnAsyncStrictness :: TestTransport -> Assertion testSpawnAsyncStrictness TestTransport{..} = do@@ -785,8 +838,9 @@ liftIO $ putMVar processA us msg1 <- expect msg2 <- expect- True <- return $ msg1 == "message 1" && msg2 == "message 3"- liftIO $ putMVar sendTestOk ()+ liftIO $ do+ assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"+ putMVar sendTestOk () forkProcess node2 $ do {-@@ -819,7 +873,7 @@ matchIf (\(RegisterReply label' _ _) -> "a" == label') (\(RegisterReply _ _ _) -> return ()) ] - Just _ <- whereisRemote nid1 "a"+ _ <- whereisRemote nid1 "a" -- Simulate network failure@@ -838,9 +892,10 @@ (\(RegisterReply _ _ _) -> return ()) ] -- Check- Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected- Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect- Just _ <- whereisRemote nid1 "c"+ mPid <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected+ liftIO $ assertBool "Expected remote name to be lost" $ mPid == Nothing+ verifyWhereIsRemote nid1 "b" -- this will suceed because the value is set after thereconnect+ verifyWhereIsRemote nid1 "c" liftIO $ putMVar registerTestOk () @@ -881,19 +936,19 @@ isSorted _ = True -- The list can't be null since there are no failures after sending -- the last message.- True <- return $ isSorted msgs && not (null msgs)- liftIO $ putMVar usendTestOk ()+ liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs) forkProcess node2 $ do them <- liftIO $ readMVar processA getSelfPid >>= send them- () <- expect+ expect :: Process () forM_ [1..numMessages] $ \i -> do liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) usendPrim them i liftIO (threadDelay 30000) - takeMVar usendTestOk+ res <- takeMVar usendTestOk+ assertBool "Unexpected failure after sending last msg" res -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between@@ -919,14 +974,17 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- 3 <- expect :: Process Double+ three <- expect :: Process Double send mathServer (Divide pid 8 2)- 4 <- expect :: Process Double+ four <- expect :: Process Double send mathServer (Divide pid 8 0)- DivByZero <- expect- liftIO $ putMVar clientDone ()+ divByZ <- expect+ liftIO $ putMVar clientDone (three, four, divByZ) - takeMVar clientDone+ res <- takeMVar clientDone+ case res of+ (3, 4, DivByZero) -> return ()+ _ -> error "Unexpected result" -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between, however we block 'Divide' requests ....@@ -953,12 +1011,13 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- 3 <- expect :: Process Double+ three <- expect :: Process Double send mathServer (Divide pid 8 2)- Nothing <- (expectTimeout 100000) :: Process (Maybe Double)- liftIO $ putMVar clientDone ()+ res <- (expectTimeout 100000) :: Process (Maybe Double)+ liftIO $ putMVar clientDone $ three == 3 && res == Nothing - takeMVar clientDone+ verifyClient "Expected Nothing (i.e. timeout)" clientDone+ where maybeForward :: ProcessId -> Message -> Process (Maybe ()) maybeForward s msg = handleMessage msg (\m@(Add _ _ _) -> send s m)@@ -986,8 +1045,10 @@ Just _ -> die "NONSENSE") ] -- we *must* have removed the message from our mailbox though!!!- Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]- liftIO $ putMVar serverDone ()+ res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]+ liftIO $ do+ assertBool "Expected timeout!" $ res == Nothing+ putMVar serverDone () putMVar addr server -- Client@@ -1030,14 +1091,21 @@ runProcess localNode $ do pid <- getSelfPid send server (pid, "foo")- "foo" <- expect- send server (pid, "baz")- "baz" <- expect+ foo <- expect+ -- provoking what would be the wrong ordering is informative here...+ send server (pid, "bar")- Nothing <- (expectTimeout 100000) :: Process (Maybe Double)- liftIO $ putMVar clientDone ()+ bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !? - takeMVar clientDone+ send server (pid, "baz")+ baz <- expect++ liftIO $ putMVar clientDone (foo, bar, baz)++ res <- takeMVar clientDone+ let res' = res == ("foo", Nothing, "baz")+ assertBool "Expected timeout due to type mismatch" res'+ where tryHandleMessage :: Message -> Process (Maybe ()) tryHandleMessage msg = handleMessage msg (\(pid :: ProcessId, (m :: String))@@ -1070,12 +1138,13 @@ runProcess localNode $ do pid <- getSelfPid send server (pid, wrapMessage ("foo" :: String))- "foo" <- expect+ foo <- expect send server (pid, wrapMessage ("baz" :: String))- "baz" <- expect- liftIO $ putMVar clientDone ()+ baz <- expect+ liftIO $ putMVar clientDone (foo, baz) - takeMVar clientDone+ res <- takeMVar clientDone+ assertBool "Unexpected unwrapped results" $ res == ("foo", "baz") -- Test 'receiveChanTimeout' testReceiveChanTimeout :: TestTransport -> Assertion@@ -1093,18 +1162,19 @@ -- Wait for a message with a delay. No message arrives, we should get -- Nothing after the delay.- Nothing <- receiveChanTimeout 100000 rp+ receiveChanTimeout 100000 rp >>= maybe (return ()) (const $ die "Expected Timeout") -- Let the sender know that it can send a message. liftIO $ putMVar mvSender () -- Wait for a message with a delay again. Now a message arrives after -- 0.1 seconds- Just True <- receiveChanTimeout 20000000 rp+ res <- receiveChanTimeout 20000000 rp >>= maybe (die "Timeout") return+ liftIO $ assertBool "Expected result to be 'True'" res -- Wait for a message with zero timeout: non-blocking check. No message is -- available, we get Nothing- Nothing <- receiveChanTimeout 0 rp+ receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout") -- Let the sender know that it can send a message. liftIO $ putMVar mvSender ()@@ -1114,7 +1184,7 @@ liftIO $ threadDelay 100000 mb <- receiveChanTimeout 0 rp case mb of- Just b -> do False <- return b; return ()+ Just b -> do liftIO $ assertBool "Unexpected Message" $ not b _ -> loop liftIO $ putMVar done ()@@ -1151,8 +1221,8 @@ rp1 <- mergePortsBiased [even <$> rpInt, rpBool] - True <- receiveChan rp1- False <- receiveChan rp1+ receiveChan rp1 >>= liftIO . assertBool "Expected True"+ receiveChan rp1 >>= liftIO . assertBool "Expected False" . not -- Test Applicative instance @@ -1161,7 +1231,7 @@ let rp2 = pure (+) <*> rpInt <*> rpInt - 7 <- receiveChan rp2+ receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7) -- Test Alternative instance @@ -1170,8 +1240,8 @@ let rp3 = (even <$> rpInt) <|> rpBool - False <- receiveChan rp3- True <- receiveChan rp3+ receiveChan rp3 >>= liftIO . assertBool "Expected False" . not+ receiveChan rp3 >>= liftIO . assertBool "Expected True" -- Test Monad instance @@ -1185,17 +1255,86 @@ then rpInt else return 7 - 5 <- receiveChan rp4- 7 <- receiveChan rp4+ receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)+ receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7) liftIO $ putMVar done () takeMVar done +testChanLifecycle :: TestTransport -> Assertion+testChanLifecycle TestTransport{..} = let delay = 3000000 in do+ result <- newEmptyMVar+ tchMV <- newEmptyMVar+ localNode <- newLocalNode testTransport initRemoteTable+ runProcess localNode $ do++ pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())+ liftIO $ putMVar tchMV tCh+ expect :: Process ()+ let (sp, _) = tCh+ sendChan sp ()+ expect :: Process ()++ mRefPid <- monitor pid++ cPid <- spawnLocal $ do+ (sp', rp) <- liftIO $ takeMVar tchMV+ -- say "picked up our test channel"+ send pid ()+ -- say "told pid to continue"+ res <- receiveChanTimeout delay rp+ case res of+ Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)+ Just () -> do _ <- monitor pid+ pause 10000+ -- say "sending pid a second () will cause it to exit"+ send pid ()++ -- say "make sure we see a DOWN notification for pid having stopped"+ receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]++ -- now that pid has died, the send port should be useless...+ liftIO $ putMVar tchMV (sp', rp)++ -- let's verify that we do not see the message from our+ -- parent process on the channel, once pid has died...+ recv <- receiveChanTimeout delay rp+ -- say $ "finished waiting for second (), writing result" ++ (show recv)+ liftIO $ putMVar result $ isNothing recv++ mRefCPid <- monitor cPid++ receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)+ (\_ -> return ())+ ]++ -- say "seen first pid die..."++ (sendPort, _) <- liftIO $ takeMVar tchMV+ sendChan sendPort ()+ -- say "sent () after owning pid died"++ -- let cPid know we've written to the channel...+ send cPid ()++ receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)+ (\_ -> return ())+ ]++ -- say "seen both pids die now..."++ -- and wait on the result back in IO land...+ testRes <- takeMVar result+ -- runProcess localNode $ say "got result..."+ assertBool "Expected sending on the channel to fail, but received data!" testRes++ testKillLocal :: TestTransport -> Assertion testKillLocal TestTransport{..} = do localNode <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar pid <- forkProcess localNode $ do liftIO $ threadDelay 1000000@@ -1204,17 +1343,18 @@ ref <- monitor pid us <- getSelfPid kill pid "TestKill"- ProcessMonitorNotification ref' pid' (DiedException ex) <- expect- True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"- liftIO $ putMVar done ()-- takeMVar done+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' (DiedException ex) ->+ case ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill" of+ True -> return ()+ False -> die "Invalid ProcessMonitorNotification received"+ _ -> die "Something went horribly wrong" testKillRemote :: TestTransport -> Assertion testKillRemote TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable node2 <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar pid <- forkProcess node1 $ do liftIO $ threadDelay 1000000@@ -1223,11 +1363,17 @@ ref <- monitor pid us <- getSelfPid kill pid "TestKill"- ProcessMonitorNotification ref' pid' (DiedException reason) <- expect- True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"- liftIO $ putMVar done ()-- takeMVar done+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' (DiedException reason) ->+ case (ref == ref', pid == pid', reason == "killed-by=" ++ show us ++ ",reason=TestKill") of+ (True, True, True) -> return ()+ (a, b, c) -> do+ let a' = if a then "" else "Invalid ref"+ let b' = if b then "" else "Invalid pid"+ let c' = if c then "" else "Invalid message"+ die $ unwords [a', b', c']+ _ -> die "Received unexpected message" testCatchesExit :: TestTransport -> Assertion testCatchesExit TestTransport{..} = do@@ -1253,7 +1399,10 @@ self <- getSelfPid send self (5 :: Integer, 10 :: Integer) msg <- receiveWait [ matchMessage return ]- Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")+ handleMessageIf msg+ (\() -> True)+ (\() -> die "whoops") >>= maybe (return ())+ (const $ die "Expected Mismatch") handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10) (\input -> liftIO $ putMVar done input) return ()@@ -1298,8 +1447,11 @@ (die ("foobar", 123 :: Int)) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- True <- return $ reason == ("foobar", 123 :: Int)+ let res = reason == ("foobar", 123 :: Int) liftIO $ putMVar done ()+ if res+ then return ()+ else die "Something went horribly wrong" takeMVar done @@ -1313,8 +1465,11 @@ `catch` \ex@(ProcessExitException from _) -> let expected = "exit-from=" ++ (show from) in do- True <- return $ (show ex) == expected+ let res = (show ex) == expected liftIO $ putMVar done ()+ if res+ then return ()+ else die "Something went horribly wrong" takeMVar done @@ -1330,17 +1485,26 @@ pid <- forkProcess localNode $ do (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- True <- return $ reason == "TestExit"+ res <- return $ reason == "TestExit" liftIO $ putMVar supervisedDone ()+ if res+ then return ()+ else die "Something went horribly wrong" runProcess localNode $ do liftIO $ takeMVar handlerSetUp ref <- monitor pid exit pid "TestExit" -- This time the client catches the exception, so it dies normally- ProcessMonitorNotification ref' pid' DiedNormal <- expect- True <- return $ ref == ref' && pid == pid'- liftIO $ putMVar supervisorDone ()+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' DiedNormal -> do+ let res = ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone ()+ if res+ then return ()+ else die "Something went horribly wrong"+ _ -> die "Something went horribly wrong" takeMVar supervisedDone takeMVar supervisorDone@@ -1356,19 +1520,98 @@ (receiveWait [] :: Process ()) -- block forever `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- True <- return $ reason == "TestExit"+ res <- return $ reason == "TestExit" liftIO $ putMVar supervisedDone ()+ if res+ then return ()+ else die "Something went horribly wrong" runProcess node2 $ do ref <- monitor pid exit pid "TestExit"- ProcessMonitorNotification ref' pid' DiedNormal <- expect- True <- return $ ref == ref' && pid == pid'- liftIO $ putMVar supervisorDone ()+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' DiedNormal -> do+ res' <- return $ ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone ()+ if res'+ then return ()+ else die "Something went horribly wrong"+ _ -> die "Something went horribly wrong" takeMVar supervisedDone takeMVar supervisorDone +testRegistryMonitoring :: TestTransport -> Assertion+testRegistryMonitoring TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable++ let nid = localNodeId node2+ pid <- forkProcess node1 $ do+ self <- getSelfPid+ runUntilRegistered nid self+ say $ (show self) ++ " registered as " ++ regName+ expect :: Process ()+ say $ (show self) ++ " exiting normally"++ runProcess node2 $ do+ register regName pid+ say $ regName ++ " registered to " ++ show pid+ res <- whereis regName+ send pid ()+ say $ " sent finish signal to " ++ show pid+ _ <- getSelfPid+ liftIO $ assertBool "expected (Just pid)" $ res == (Just pid)+++ -- This delay isn't essential!+ -- The test case passes perfectly fine without it (feel free to comment out+ -- and see), however waiting a few seconds here, makes it much more likely+ -- that in delayUntilMaybeUnregistered we will hit the match case right+ -- away, and thus not be subjected to a 20 second delay. The value of 4+ -- seconds appears to work optimally on osx and across several linux distros+ -- running in virtual machines (which is essentially what we do in CI)+ void $ receiveTimeout 4000000 [ matchAny return ]++ -- This delay doesn't serve much purpose in the happy path, however if some+ -- future patch breaks the cooperative behaviour of node controllers viz+ -- remote process registration and notification taking place via ncEffectDied,+ -- there would be the possibility of a race in the test case should we attempt+ -- to evaluate `whereis regName` on node2 right away. In case the name is still+ -- erroneously registered, observing the 20 second delay (or lack of), could at+ -- least give a hint that something is wrong, and we give up our time slice+ -- so that there's a higher change the registrations have been cleaned up+ -- in either case.+ runProcess node2 $ delayUntilMaybeUnregistered nid pid++ regHere <- newEmptyMVar+ runProcess node2 $ whereis regName >>= liftIO . putMVar regHere+ res <- takeMVar regHere+ case res of+ Nothing -> return ()+ _ -> assertBool ("expected Nothing, but got " ++ show pid) False++ where+ runUntilRegistered nid us = do+ whereisRemoteAsync nid regName+ receiveWait [+ matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)+ (const $ return ())+ ]++ delayUntilMaybeUnregistered nid p = do+ whereisRemoteAsync nid regName+ res <- receiveTimeout 20000000 {- 20 sec delay -} [+ matchIf (\(WhereIsReply n p') -> n == regName && isNothing p')+ (const $ return ())+ ]+ case res of+ Just () -> return ()+ Nothing -> delayUntilMaybeUnregistered nid p++ regName = "testRegisterRemote"+ testUnsafeSend :: TestTransport -> Assertion testUnsafeSend TestTransport{..} = do serverAddr <- newEmptyMVar@@ -1384,8 +1627,7 @@ void $ forkProcess localNode $ do serverPid <- liftIO $ takeMVar serverAddr getSelfPid >>= unsafeSend serverPid- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone takeMVar clientDone @@ -1404,8 +1646,7 @@ void $ forkProcess localNode $ do serverPid <- liftIO $ takeMVar serverAddr getSelfPid >>= unsafeUSend serverPid- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone takeMVar clientDone @@ -1416,8 +1657,7 @@ localNode <- newLocalNode testTransport initRemoteTable pid <- forkProcess localNode $ do- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone void $ runProcess localNode $ do register "foobar" pid@@ -1435,8 +1675,7 @@ _ <- forkProcess localNode1 $ do getSelfPid >>= register "foobar" liftIO $ putMVar clientDone ()- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone takeMVar clientDone void $ runProcess localNode2 $ do@@ -1460,7 +1699,7 @@ serverPid <- liftIO $ takeMVar serverAddr (sp, rp) <- newChan unsafeSend serverPid sp- () <- receiveChan rp+ receiveChan rp :: Process () liftIO $ putMVar clientDone () takeMVar clientDone@@ -1474,9 +1713,9 @@ runProcess node $ do r <- (/=) <$> getSelfPid <*> callLocal getSelfPid liftIO $ putMVar result r- True <- takeMVar result- return () + takeMVar result >>= assertBool "Expected 'True'"+ -- Testing that when callLocal is interrupted, the worker is interrupted. ibox <- newIORef False runProcess node $ do@@ -1492,11 +1731,10 @@ (send keeper ()) caller <- expect exit caller "test"- () <- expect- return ()- True <- readIORef ibox- return ()+ expect :: Process () + readIORef ibox >>= assertBool "Expected 'True'"+ -- Testing that when the worker raises an exception, the exception is propagated to the parent. ibox2 <- newIORef False runProcess node $ do@@ -1504,9 +1742,9 @@ liftIO $ writeIORef ibox2 $ case r of Left (ErrorCall "e") -> True _ -> False- True <- readIORef ibox- return () + readIORef ibox >>= assertBool "Expected 'True'"+ -- Test that caller waits for the worker in correct situation ibox3 <- newIORef False result3 <- newEmptyMVar@@ -1522,9 +1760,9 @@ liftIO $ putMVar result3 =<< readIORef ibox3 worker <- expect send worker ()- True <- takeMVar result3- return () + takeMVar result3 >>= assertBool "Expected 'True'"+ -- Test that caller waits for the worker in case when caller gets an exception ibox4 <- newIORef False result4 <- newEmptyMVar@@ -1539,13 +1777,9 @@ `finally` (liftIO $ putMVar result4 =<< readIORef ibox4) caller <- expect exit caller "hi!"- True <- takeMVar result4- return ()- -- XXX: Testing that when mask_ $ callLocal p runs p in masked state. ---+ takeMVar result4 >>= assertBool "Expected 'True'"+ -- XXX: Testing that when mask_ $ callLocal p runs p in masked state. tests :: TestTransport -> IO [Test] tests testtrans = return [@@ -1570,6 +1804,7 @@ , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans) , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans) , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)+ , testCase "ChanLifecycle" (testChanLifecycle testtrans) , testCase "KillLocal" (testKillLocal testtrans) , testCase "KillRemote" (testKillRemote testtrans) , testCase "Die" (testDie testtrans)@@ -1579,6 +1814,7 @@ , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans) , testCase "ExitLocal" (testExitLocal testtrans) , testCase "ExitRemote" (testExitRemote testtrans)+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans) , testCase "TextCallLocal" (testCallLocal testtrans) -- Unsafe Primitives , testCase "TestUnsafeSend" (testUnsafeSend testtrans)@@ -1631,17 +1867,3 @@ , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False) ] ]--syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()-syncBreakConnection breakConnection nid0 nid1 = do- m <- newEmptyMVar- _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m- runProcess nid0 $ do- them <- liftIO $ takeMVar m- pinger <- spawnLocal $ forever $ send them ()- _ <- monitorNode (localNodeId nid1)- liftIO $ breakConnection (nodeAddress $ localNodeId nid0)- (nodeAddress $ localNodeId nid1)- NodeMonitorNotification _ _ _ <- expect- kill pinger "finished"- return ()
src/Control/Distributed/Process/Tests/Closure.hs view
@@ -1,3 +1,4 @@+{-# OPTIONS_GHC -Wno-unused-top-binds #-} {-# LANGUAGE TemplateHaskell, KindSignatures #-} module Control.Distributed.Process.Tests.Closure (tests) where @@ -19,14 +20,12 @@ , modifyMVar_ , newMVar )-import Control.Applicative ((<$>)) import System.Random (randomIO) import Control.Distributed.Process import Control.Distributed.Process.Closure import Control.Distributed.Process.Node import Control.Distributed.Process.Internal.Types- ( NodeId(nodeAddress)- , createMessage+ ( createMessage , messageToPayload ) import Control.Distributed.Static (staticLabel, staticClosure)@@ -169,8 +168,11 @@ node <- newLocalNode testTransport rtable done <- newEmptyMVar forkProcess node $ do- 120 <- join . unClosure $ factorialClosure 5+ i <- join . unClosure $ factorialClosure 5 liftIO $ putMVar done ()+ if i == 720+ then return ()+ else error "Something went horribly wrong" takeMVar done testBind :: TestTransport -> RemoteTable -> Assertion@@ -180,8 +182,11 @@ runProcess node $ do us <- getSelfPid join . unClosure $ sendFac 6 us- (720 :: Int) <- expect+ (i :: Int) <- expect liftIO $ putMVar done ()+ if i == 720+ then return ()+ else error "Something went horribly wrong" takeMVar done testSendPureClosure :: TestTransport -> RemoteTable -> Assertion@@ -194,7 +199,7 @@ addr <- forkProcess node $ do cl <- expect fn <- unClosure cl :: Process (Int -> Int)- 13 <- return $ fn 6+ (_ :: Int) <- return $ fn 6 liftIO $ putMVar serverDone () putMVar serverAddr addr @@ -218,8 +223,11 @@ liftIO $ do someMVar <- newEmptyMVar io someMVar- 5 <- readMVar someMVar+ i <- readMVar someMVar putMVar serverDone ()+ if i == 5+ then return ()+ else error "Something went horribly wrong" putMVar serverAddr addr forkIO $ do@@ -248,8 +256,10 @@ runProcess node $ do pid <- getSelfPid send theirAddr (cpSend $(mkStatic 'sdictInt) pid)- 5 <- expect :: Process Int- liftIO $ putMVar clientDone ()+ i <- expect :: Process Int+ if i == 5+ then liftIO $ putMVar clientDone ()+ else error "Something went horribly wrong" takeMVar clientDone @@ -269,8 +279,9 @@ pid <- getSelfPid pid' <- spawn nid (sendPidClosure pid) pid'' <- expect- True <- return $ pid' == pid''- liftIO $ putMVar clientDone ()+ if pid' == pid''+ then liftIO $ putMVar clientDone ()+ else error "Something went horribly wrong" takeMVar clientDone @@ -294,8 +305,9 @@ spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid pid' <- expect :: Process ProcessId pid'' <- expect :: Process ProcessId- True <- return $ pid' == pid''- return ()+ if pid' == pid''+ then return ()+ else error "Something went horribly wrong" where @@ -332,8 +344,10 @@ node <- newLocalNode testTransport rtable nid <- readMVar serverNodeAddr runProcess node $ do- (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)- liftIO $ putMVar clientDone ()+ (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)+ if a == 120+ then liftIO $ putMVar clientDone ()+ else error "something went horribly wrong" takeMVar clientDone @@ -350,8 +364,10 @@ node <- newLocalNode testTransport rtable nid <- readMVar serverNodeAddr runProcess node $ do- (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)- liftIO $ putMVar clientDone ()+ (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)+ if a == 120+ then liftIO $ putMVar clientDone ()+ else error "Something went horribly wrong" takeMVar clientDone @@ -362,9 +378,11 @@ runProcess node $ do us <- getSelfPid join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us- 120 :: Int <- expect- 720 :: Int <- expect- liftIO $ putMVar done ()+ a :: Int <- expect+ b :: Int <- expect+ if a == 120 && b == 720+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong" takeMVar done -- Test 'spawnSupervised'@@ -393,26 +411,34 @@ throw supervisorDeath forkProcess node2 $ do- [super, child] <- liftIO $ mapM readMVar [superPid, childPid]- ref <- monitor child- self <- getSelfPid- let waitForMOrL = do- liftIO $ threadDelay 10000- mpinfo <- getProcessInfo child- case mpinfo of- Nothing -> waitForMOrL- Just pinfo ->- unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL- waitForMOrL- liftIO $ putMVar linkUp ()- -- because monitor message was sent before message to process- -- we hope that it will be processed before- ProcessMonitorNotification ref' pid' (DiedException e) <- expect- True <- return $ ref' == ref- && pid' == child- && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))- liftIO $ putMVar thirdProcessDone ()+ res <- liftIO $ mapM readMVar [superPid, childPid]+ case res of+ [super, child] -> do+ ref <- monitor child+ self <- getSelfPid+ let waitForMOrL = do+ liftIO $ threadDelay 10000+ mpinfo <- getProcessInfo child+ case mpinfo of+ Nothing -> waitForMOrL+ Just pinfo ->+ unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL+ waitForMOrL+ liftIO $ putMVar linkUp ()+ -- because monitor message was sent before message to process+ -- we hope that it will be processed before+ res' <- expect+ case res' of+ (ProcessMonitorNotification ref' pid' (DiedException e)) ->+ if (ref' == ref && pid' == child &&+ e == show (ProcessLinkException super+ (DiedException (show supervisorDeath))))+ then liftIO $ putMVar thirdProcessDone ()+ else error "Something went horribly wrong"+ _ -> error "Something went horribly wrong" + _ -> die $ "Something went horribly wrong"+ takeMVar thirdProcessDone where supervisorDeath :: IOException@@ -426,8 +452,10 @@ (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty) ProcessMonitorNotification ref' pid' _reason <- expect -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure- True <- return $ ref' == ref && pid == pid'- liftIO $ putMVar done ()+ res <- return $ ref' == ref && pid == pid'+ if res == True+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong" takeMVar done testClosureExpect :: TestTransport -> RemoteTable -> Assertion@@ -439,8 +467,10 @@ us <- getSelfPid them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us send them (1234 :: Int)- (1234 :: Int) <- expect- liftIO $ putMVar done ()+ (res :: Int) <- expect+ if res == 1234+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong" takeMVar done testSpawnChannel :: TestTransport -> RemoteTable -> Assertion@@ -465,8 +495,10 @@ done <- newEmptyMVar [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable forkProcess node1 $ do- True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))- liftIO $ putMVar done ()+ res <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))+ if res == True+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong..." takeMVar done testFib :: TestTransport -> RemoteTable -> Assertion@@ -477,8 +509,11 @@ forkProcess (head nodes) $ do (sport, rport) <- newChan spawnLocal $ dfib (map localNodeId nodes, sport, 10)- 55 <- receiveChan rport :: Process Integer+ ff <- receiveChan rport :: Process Integer liftIO $ putMVar done ()+ if ff /= 55+ then die $ "Something went horribly wrong"+ else return () takeMVar done @@ -486,7 +521,7 @@ testSpawnReconnect testtrans@TestTransport{..} rtable = do [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable let nid1 = localNodeId node1- nid2 = localNodeId node2+ -- nid2 = localNodeId node2 done <- newEmptyMVar iv <- newMVar (0 :: Int) @@ -503,9 +538,12 @@ liftIO $ threadDelay 100000 count <- liftIO $ takeMVar iv- True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'+ res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn' liftIO $ putMVar done ()+ if res /= True+ then error "Something went horribly wrong"+ else return () takeMVar done
src/Control/Distributed/Process/Tests/Internal/Utils.hs view
@@ -20,6 +20,7 @@ -- ping ! , Ping(Ping) , ping+ , pause , shouldBe , shouldMatch , shouldContain@@ -73,15 +74,16 @@ import Control.Concurrent.MVar ( putMVar )-import Control.Distributed.Process+import Control.Distributed.Process hiding (finally, catch) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable() import Control.Exception (AsyncException(ThreadKilled), SomeException)-import Control.Monad (forever)+import Control.Monad (forever, void)+import Control.Monad.Catch (finally, catch) import Control.Monad.STM (atomically) import Control.Rematch hiding (match)-import Control.Rematch.Run +import Control.Rematch.Run import Data.Binary import Data.Typeable (Typeable) @@ -89,6 +91,7 @@ import Test.HUnit.Base (assertBool) import GHC.Generics+import System.Timeout (timeout) -- | A mutable cell containing a test result. type TestResult a = MVar a@@ -107,10 +110,18 @@ instance Binary TestProcessControl where +data Private = Private+ deriving (Typeable, Generic)+instance Binary Private where+ -- | Does exactly what it says on the tin, doing so in the @Process@ monad. noop :: Process () noop = return () +pause :: Int -> Process ()+pause delay =+ void $ receiveTimeout delay [ match (\Private -> return ()) ]+ synchronisedAssertion :: Eq a => String -> LocalNode@@ -177,8 +188,8 @@ _ <- forkIO $ logger q return $ Logger tid q where logger q' = forever $ do- msg <- atomically $ readTQueue q'- putStrLn msg+ msg <- atomically $ readTQueue q'+ putStrLn msg -- | Send a message to the Logger putLogMsg :: Logger -> String -> Process ()@@ -223,10 +234,10 @@ tryRunProcess :: LocalNode -> Process () -> IO () tryRunProcess node p = do tid <- liftIO myThreadId- runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))+ void $ timeout (1000000 * 60 * 5 :: Int) $+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException)) tryForkProcess :: LocalNode -> Process () -> IO ProcessId tryForkProcess node p = do tid <- liftIO myThreadId forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))-
src/Control/Distributed/Process/Tests/Mx.hs view
@@ -1,12 +1,23 @@-{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE ParallelListComp #-}+ module Control.Distributed.Process.Tests.Mx (tests) where import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..))--import Control.Concurrent (threadDelay)-import Control.Distributed.Process+import Control.Distributed.Process hiding (bracket, finally, try)+import Control.Distributed.Process.Internal.Types+ ( ProcessExitException(..)+ , unsafeCreateUnencodedMessage+ ) import Control.Distributed.Process.Node+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe+ ( send+ , nsend+ , nsendRemote+ , usend+ , sendChan+ ) import Control.Distributed.Process.Management ( MxEvent(..) , MxAgentId(..)@@ -21,14 +32,15 @@ , mxUpdateLocal , mxNotify , mxBroadcast- , mxGetId )-import Control.Monad (void)+import Control.Monad (void, unless)+import Control.Monad.Catch(finally, bracket, try)+import Control.Rematch (equalTo) import Data.Binary-import Data.List (find, sort)-import Data.Maybe (isJust)+import Data.List (find, sort, intercalate)+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes) import Data.Typeable-import GHC.Generics+import GHC.Generics hiding (from) #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch, log) #endif@@ -44,7 +56,17 @@ instance Binary Publish where -testAgentBroadcast :: TestResult () -> Process ()+awaitExit :: ProcessId -> Process ()+awaitExit pid =+ withMonitorRef pid $ \ref -> do+ receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)+ (\_ -> return ())+ ]+ where+ withMonitorRef p = bracket (monitor p) unmonitor++testAgentBroadcast :: TestResult (Maybe ()) -> Process () testAgentBroadcast result = do (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ()) @@ -59,7 +81,7 @@ mxNotify () -- Once the publisher has seen our message, it will broadcast the Publish -- and the consumer will see that and send the result to our typed channel.- stash result =<< receiveChan resultRP+ stash result =<< receiveChanTimeout 10000000 resultRP kill publisher "finished" kill consumer "finished"@@ -67,7 +89,7 @@ testAgentDualInput :: TestResult (Maybe Int) -> Process () testAgentDualInput result = do (sp, rp) <- newChan- _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [ mxSink $ (\(i :: Int) -> do mxSetLocal . (+i) =<< mxGetLocal i' <- mxGetLocal@@ -77,6 +99,8 @@ else mxReady) ] + mRef <- monitor s+ mxNotify (1 :: Int) nsend "sum-agent" (3 :: Int) mxNotify (2 :: Int)@@ -84,6 +108,10 @@ mxNotify (5 :: Int) stash result =<< receiveChanTimeout 10000000 rp+ died <- receiveTimeout 10000000 [+ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)+ ]+ died `shouldBe` equalTo (Just True) testAgentPrioritisation :: TestResult [String] -> Process () testAgentPrioritisation result = do@@ -95,12 +123,12 @@ let name = "prioritising-agent" (sp, rp) <- newChan- void $ mxAgent (MxAgentId name) ["first"] [- mxSink (\(s :: String) -> do- mxUpdateLocal ((s:))+ s <- mxAgent (MxAgentId name) ["first"] [+ mxSink (\(s :: String) -> do+ mxUpdateLocal (s:) st <- mxGetLocal case length st of- n | n == 5 -> do liftMX $ sendChan sp st+ n | n == 5 -> do liftMX $ sendChan sp (sort st) mxDeactivate "finished" _ -> mxReceive -- go to the mailbox )@@ -111,23 +139,33 @@ mxNotify "fourth" nsend name "fifth" - stash result . sort =<< receiveChan rp+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp+ awaitExit s testAgentMailboxHandling :: TestResult (Maybe ()) -> Process () testAgentMailboxHandling result = do (sp, rp) <- newChan- agent <- mxAgent (MxAgentId "listener-agent") () [+ agent <- mxAgent (MxAgentId "mailbox-agent") () [ mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady ] - nsend "listener-agent" ()+ nsend "mailbox-agent" () stash result =<< receiveChanTimeout 1000000 rp kill agent "finished" testAgentEventHandling :: TestResult Bool -> Process () testAgentEventHandling result = do+ us <- getSelfPid+ -- because this test is a bit racy, let's ensure it can't run indefinitely+ timer <- spawnLocal $ do+ pause (1000000 * 60 * 5)+ -- okay we've waited 5 mins, let's kill the test off if it's stuck...+ stash result False+ kill us "Test Timed Out"+ let initState = [] :: [MxEvent]+ (rc, rs) <- newChan agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [ (mxSink $ \ev -> do st <- mxGetLocal@@ -136,7 +174,7 @@ (MxSpawned _) -> mxSetLocal (ev:st) (MxProcessDied _ _) -> mxSetLocal (ev:st) _ -> return ()- act >> mxReady),+ act >> (liftMX $ sendChan rc ()) >> mxReady), (mxSink $ \(ev, sp :: SendPort Bool) -> do st <- mxGetLocal let found =@@ -156,50 +194,349 @@ mxReady) ] - _ <- monitor agentPid- (sp, rp) <- newChan- pid <- spawnLocal $ sendChan sp ()- () <- receiveChan rp+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...+ faff 2000000 rs + (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())++ pid <- spawnLocal $ expect >>= sendChan sp+ -- By waiting for a monitor notification, we have a- -- higher probably that the agent has seen+ -- higher probably that the agent has seen the spawn and died events monitor pid- receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ] + send pid ()+ rct <- receiveChanTimeout 10000000 rp+ unless (isJust rct) $ die "No response on channel"++ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]+ unless (isJust pmn) $ die "No monitor notification arrived"++ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...+ faff 2000000 rs+ (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool) mxNotify (MxSpawned pid, replyTo) mxNotify (MxProcessDied pid DiedNormal, replyTo) - seenAlive <- receiveChan reply- seenDead <- receiveChan reply+ seenAlive <- receiveChanTimeout 10000000 reply+ seenDead <- receiveChanTimeout 10000000 reply - stash result $ seenAlive && seenDead+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]+ kill timer "test-complete"+ kill agentPid "test-complete"+ where+ faff delay port = do+ res <- receiveChanTimeout delay port+ unless (isNothing res) $ pause delay +testMxRegEvents :: TestResult () -> Process ()+testMxRegEvents result = do+ {- This test only deals with the local case, to ensure that we are being+ notified in the expected order - the remote cases related to the+ behaviour of the node controller are contained in the CH test suite. -}+ ensure (stash result ()) $ do+ let label = "testMxRegEvents"+ let agentLabel = "mxRegEvents-agent"+ let delay = 1000000+ (regChan, regSink) <- newChan+ (unRegChan, unRegSink) <- newChan+ agent <- mxAgent (MxAgentId agentLabel) () [+ mxSink $ \ev -> do+ case ev of+ MxRegistered pid label'+ | label' == label -> liftMX $ sendChan regChan (label', pid)+ MxUnRegistered pid label'+ | label' == label -> liftMX $ sendChan unRegChan (label', pid)+ _ -> return ()+ mxReady+ ]++ p1 <- spawnLocal expect+ p2 <- spawnLocal expect++ register label p1+ reg1 <- receiveChanTimeout delay regSink+ reg1 `shouldBe` equalTo (Just (label, p1))++ unregister label+ unreg1 <- receiveChanTimeout delay unRegSink+ unreg1 `shouldBe` equalTo (Just (label, p1))++ register label p2+ reg2 <- receiveChanTimeout delay regSink+ reg2 `shouldBe` equalTo (Just (label, p2))++ reregister label p1+ unreg2 <- receiveChanTimeout delay unRegSink+ unreg2 `shouldBe` equalTo (Just (label, p2))++ reg3 <- receiveChanTimeout delay regSink+ reg3 `shouldBe` equalTo (Just (label, p1))++ mapM_ (flip kill $ "test-complete") [agent, p1, p2]++testMxRegMon :: LocalNode -> TestResult () -> Process ()+testMxRegMon remoteNode result = do+ ensure (stash result ()) $ do+ -- ensure that when a registered process dies, we get a notification that+ -- it has been unregistered as well as seeing the name get removed+ let label1 = "aaaaa"+ let label2 = "bbbbb"+ let isValid l = l == label1 || l == label2+ let agentLabel = "mxRegMon-agent"+ let delay = 1000000+ (regChan, regSink) <- newChan+ (unRegChan, unRegSink) <- newChan+ agent <- mxAgent (MxAgentId agentLabel) () [+ mxSink $ \ev -> do+ case ev of+ MxRegistered pid label+ | isValid label -> liftMX $ sendChan regChan (label, pid)+ MxUnRegistered pid label+ | isValid label -> liftMX $ sendChan unRegChan (label, pid)+ _ -> return ()+ mxReady+ ]++ (sp, rp) <- newChan+ liftIO $ forkProcess remoteNode $ do+ getSelfPid >>= sendChan sp+ expect :: Process ()++ p1 <- receiveChan rp++ register label1 p1+ reg1 <- receiveChanTimeout delay regSink+ reg1 `shouldBe` equalTo (Just (label1, p1))++ register label2 p1+ reg2 <- receiveChanTimeout delay regSink+ reg2 `shouldBe` equalTo (Just (label2, p1))++ n1 <- whereis label1+ n1 `shouldBe` equalTo (Just p1)++ n2 <- whereis label2+ n2 `shouldBe` equalTo (Just p1)++ kill p1 "goodbye"++ unreg1 <- receiveChanTimeout delay unRegSink+ unreg2 <- receiveChanTimeout delay unRegSink++ let evts = [unreg1, unreg2]+ -- we can't rely on the order of the values in the node controller's+ -- map (it's either racy to do so, or no such guarantee exists for Data.Map),+ -- so we simply verify that we received the un-registration events we expect+ evts `shouldContain` (Just (label1, p1))+ evts `shouldContain` (Just (label2, p1))++ kill agent "test-complete"++ensure :: Process () -> Process () -> Process ()+ensure = flip finally++testNSend :: (String -> () -> Process ())+ -> Maybe LocalNode+ -> Process ()+testNSend op n = do+ us <- getSelfPid+ let delay = 5000000+ let label = "testMxSend" ++ (show us)+ let isValid = isValidLabel n label++ testMxSend n label $ \p1 sink -> do+ register label p1+ reg1 <- receiveChanTimeout delay sink+ case reg1 of+ Just (MxRegistered pd lb)+ | pd == p1 && isValid lb -> return ()+ _ -> die $ "Reg-Failed: " ++ show reg1++ op label ()++ sent <- receiveChanTimeout delay sink+ case sent of+ Just (MxSentToName lb by _)+ | by == us && isValid lb -> return True+ _ -> die $ "Send-Failed: " ++ show sent++ where+ isValidLabel nd l1 l2+ | l2 == l2 = True+ | isJust nd = l2 == l1 ++ "@" ++ show (localNodeId $ fromJust nd)+ | otherwise = False++testSend :: (ProcessId -> () -> Process ())+ -> Maybe LocalNode+ -> Process ()+testSend op n = do+ us <- getSelfPid+ let label = "testMxSend" ++ (show us)+ testMxSend n label $ \p1 sink -> do+ -- initiate a send+ op p1 ()++ -- verify the management event+ sent <- receiveChanTimeout 5000000 sink+ case sent of+ Just (MxSent pidTo pidFrom _)+ | pidTo == p1 && pidFrom == us -> return True+ _ -> return False++testChan :: (SendPort () -> () -> Process ())+ -> Maybe LocalNode+ -> Process ()+testChan op n = testMxSend n "" $ \p1 sink -> do++ us <- getSelfPid+ send p1 us++ cleared <- receiveChanTimeout 2000000 sink+ case cleared of+ Just (MxSent pidTo pidFrom _)+ | pidTo == p1 && pidFrom == us -> return ()+ _ -> die "Received uncleared Mx Event"++ chan <- receiveTimeout 5000000 [ match return ]+ let ch' = fromMaybe (error "No reply chan received") chan++ -- initiate a send+ op ch' ()++ -- verify the management event+ sent <- receiveChanTimeout 5000000 sink+ case sent of+ Just (MxSentToPort sId spId _)+ | sId == us && spId == sendPortId ch' -> return True+ _ -> return False++type SendTest = ProcessId -> ReceivePort MxEvent -> Process Bool++testMxSend :: Maybe LocalNode -> String -> SendTest -> Process ()+testMxSend mNode label test = do+ us <- getSelfPid+ (sp, rp) <- newChan+ (chan, sink) <- newChan+ agent <- mxAgent (MxAgentId $ agentLabel us) () [+ mxSink $ \ev -> do+ case ev of+ m@(MxSentToPort _ cid _)+ | cid /= sendPortId chan+ && cid /= sendPortId sp -> liftMX $ sendChan chan m+ m@(MxSent _ fromPid _)+ | fromPid == us -> liftMX $ sendChan chan m+ m@(MxSentToName _ fromPid _)+ | fromPid == us -> liftMX $ sendChan chan m+ m@(MxRegistered _ name)+ | name == label -> liftMX $ sendChan chan m+ _ -> return ()+ mxReady+ ]++ case mNode of+ Nothing -> void $ spawnLocal (proc sp)+ Just remoteNode -> void $ liftIO $ forkProcess remoteNode $ proc sp++ p1 <- receiveChanTimeout 2000000 rp+ unless (isJust p1) $ die "Timed out waiting for ProcessId"+ res <- try $ test (fromJust p1) sink++ kill agent "bye"+ kill (fromJust p1) "bye"++ case res of+ Left (ProcessExitException _ m) -> (liftIO $ putStrLn $ "SomeException-" ++ show m) >> die m+ Right tr -> tr `shouldBe` equalTo True+++ where+ agentLabel s = "mx-unsafe-check-agent-" ++ show s+ proc sp' = getSelfPid >>= sendChan sp' >> go Nothing++ go :: Maybe (ReceivePort ()) -> Process ()+ go Nothing = receiveTimeout 5000000 [ match replyChannel ] >>= go+ go c@(Just c') = receiveWait [ matchChan c' return ] >> go c++ replyChannel p' = do+ (s, r) <- newChan+ send p' s+ return r+ tests :: TestTransport -> IO [Test] tests TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable return [- testGroup "Mx Agents" [- testCase "Event Handling"- (delayedAssertion- "expected True, but events where not as expected"- node1 True testAgentEventHandling)- , testCase "Inter-Agent Broadcast"- (delayedAssertion- "expected (), but no broadcast was received"- node1 () testAgentBroadcast)- , testCase "Agent Mailbox Handling"- (delayedAssertion- "expected (Just ()), but no regular (mailbox) input was handled"- node1 (Just ()) testAgentMailboxHandling)- , testCase "Agent Dual Input Handling"- (delayedAssertion- "expected sum = 15, but the result was Nothing"- node1 (Just 15 :: Maybe Int) testAgentDualInput)- , testCase "Agent Input Prioritisation"- (delayedAssertion- "expected [first, second, third, fourth, fifth], but result diverged"- node1 (sort ["first", "second",- "third", "fourth",- "fifth"]) testAgentPrioritisation)- ]]+ testGroup "MxAgents" [+ testCase "EventHandling"+ (delayedAssertion+ "expected True, but events where not as expected"+ node1 True testAgentEventHandling)+ , testCase "InterAgentBroadcast"+ (delayedAssertion+ "expected (), but no broadcast was received"+ node1 (Just ()) testAgentBroadcast)+ , testCase "AgentMailboxHandling"+ (delayedAssertion+ "expected (Just ()), but no regular (mailbox) input was handled"+ node1 (Just ()) testAgentMailboxHandling)+ , testCase "AgentDualInputHandling"+ (delayedAssertion+ "expected sum = 15, but the result was Nothing"+ node1 (Just 15 :: Maybe Int) testAgentDualInput)+ , testCase "AgentInputPrioritisation"+ (delayedAssertion+ "expected [first, second, third, fourth, fifth], but result diverged"+ node1 (sort ["first", "second",+ "third", "fourth",+ "fifth"]) testAgentPrioritisation)+ ]+ , testGroup "MxEvents" [+ testCase "NameRegistrationEvents"+ (delayedAssertion+ "expected registration events to map to the correct ProcessId"+ node1 () testMxRegEvents)+ , testCase "PostDeathNameUnRegistrationEvents"+ (delayedAssertion+ "expected process deaths to result in unregistration events"+ node1 () (testMxRegMon node2))+ , testGroup "SendEvents" $ buildTestCases node1 node2+ ]+ ]+ where+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [+ ("NSend", [+ ("nsend", testNSend nsend)+ , ("Unsafe.nsend", testNSend Unsafe.nsend)+ , ("nsendRemote", testNSend (nsendRemote nid))+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))+ ])+ , ("Send", [+ ("send", testSend send)+ , ("Unsafe.send", testSend Unsafe.send)+ , ("usend", testSend usend)+ , ("Unsafe.usend", testSend Unsafe.usend)+ , ("sendChan", testChan sendChan)+ , ("Unsafe.sendChan", testChan Unsafe.sendChan)+ ])+ , ("Forward", [+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))+ ])+ ]++ build :: LocalNode+ -> LocalNode+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]+ -> [Test]+ build n ln specs =+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [+ testCase (intercalate "-" [caseName, caseSuffix])+ (runProcess n (caseImpl caseNode))+ | (caseName, caseImpl) <- groupCases+ ]+ | (groupName, groupCases) <- specs+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]+ ]
src/Control/Distributed/Process/Tests/Receive.hs view
@@ -10,13 +10,12 @@ import Network.Transport (Transport) import Control.Distributed.Process-import Control.Distributed.Process.Closure import Control.Distributed.Process.Node import Control.Monad import Test.HUnit (Assertion, (@?=))-import Test.Framework (Test, defaultMain)+import Test.Framework (Test) import Test.Framework.Providers.HUnit (testCase) -- Tests:@@ -43,7 +42,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest2 wait sync r1 r2 = do+recTest2 wait sync r1 _ = do forever $ do receiveChan wait r <- receiveWait@@ -56,7 +55,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest3 wait sync r1 r2 = do+recTest3 wait sync r1 _ = do forever $ do receiveChan wait r <- receiveWait@@ -69,7 +68,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest4 wait sync r1 r2 = do+recTest4 wait sync r1 _ = do forever $ do receiveChan wait r <- receiveWait@@ -83,11 +82,11 @@ master = do (waits,waitr) <- newChan (syncs,syncr) <- newChan- let go expect = do+ let go expected = do sendChan waits () r <- receiveChan syncr- liftIO $ print (r,expect, r == expect)- liftIO $ r @?= expect+ liftIO $ print (r, expected, r == expected)+ liftIO $ r @?= expected liftIO $ putStrLn "---- Test 1 ----" (s1,r1) <- newChan@@ -102,44 +101,44 @@ kill p "BANG" liftIO $ putStrLn "\n---- Test 2 ----"- (s1,r1) <- newChan- (s2,r2) <- newChan- p <- spawnLocal (recTest2 waitr syncs r1 r2)+ (s1',r1') <- newChan+ (_ ,r2') <- newChan+ p' <- spawnLocal (recTest2 waitr syncs r1' r2') - sendChan s1 "a" >> go "received1 a"- send p "foo" >> go "received2 foo"- sendChan s1 "a" >> send p "foo" >> go "received1 a"- sendChan s1 "a" >> send p "bar" >> go "received1 a"+ sendChan s1' "a" >> go "received1 a"+ send p' "foo" >> go "received2 foo"+ sendChan s1' "a" >> send p "foo" >> go "received1 a"+ sendChan s1' "a" >> send p "bar" >> go "received1 a" go "received2 foo" - kill p "BANG"+ kill p' "BANG" liftIO $ putStrLn "\n---- Test 3 ----"- (s1,r1) <- newChan- (s2,r2) <- newChan- p <- spawnLocal (recTest3 waitr syncs r1 r2)+ (s1'',r1'') <- newChan+ (_ ,r2'') <- newChan+ p'' <- spawnLocal (recTest3 waitr syncs r1'' r2'') - sendChan s1 "a" >> go "received2 a"- send p "foo" >> go "received1 foo"- sendChan s1 "a" >> send p "foo" >> go "received1 foo"- sendChan s1 "a" >> send p "bar" >> go "received2 a"+ sendChan s1'' "a" >> go "received2 a"+ send p'' "foo" >> go "received1 foo"+ sendChan s1'' "a" >> send p "foo" >> go "received1 foo"+ sendChan s1'' "a" >> send p "bar" >> go "received2 a" go "received2 a" - kill p "BANG"+ kill p'' "BANG" liftIO $ putStrLn "\n---- Test 4 ----"- (s1,r1) <- newChan- (s2,r2) <- newChan- p <- spawnLocal (recTest4 waitr syncs r1 r2)+ (s1''',r1''') <- newChan+ (_ ,r2''') <- newChan+ p''' <- spawnLocal (recTest4 waitr syncs r1''' r2''') - sendChan s1 "a" >> go "received2 a"- send p "foo" >> go "received1 foo"- send p "bar" >> go "received3 bar"- sendChan s1 "a" >> send p "foo" >> go "received1 foo"- send p "bar" >> go "received2 a"- send p "foo" >> go "received1 foo" >> go "received3 bar"+ sendChan s1''' "a" >> go "received2 a"+ send p''' "foo" >> go "received1 foo"+ send p''' "bar" >> go "received3 bar"+ sendChan s1''' "a" >> send p''' "foo" >> go "received1 foo"+ send p''' "bar" >> go "received2 a"+ send p''' "foo" >> go "received1 foo" >> go "received3 bar" - kill p "BANG"+ kill p''' "BANG" terminate
src/Control/Distributed/Process/Tests/Stats.hs view
@@ -29,7 +29,7 @@ testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process () testLocalDeadProcessInfo result = do- pid <- spawnLocal $ do "finish" <- expect; return ()+ pid <- spawnLocal $ do (_ :: String) <- expect; return () mref <- monitor pid send pid "finish" _ <- receiveWait [@@ -52,8 +52,10 @@ link self mRef <- monitor self stash mon mRef- "die" <- expect- return ()+ res <- expect+ case res of+ "die" -> return ()+ _ -> die $ "unexpected message received: " ++ res monRef <- liftIO $ takeMVar mon @@ -84,9 +86,9 @@ -- our send op shouldn't overtake link or monitor requests AFAICT -- so a little table tennis should get us synchronised properly send serverPid (self, "ping")- "pong" <- expect+ pong <- expect pInfo <- getProcessInfo serverPid- stash result $ pInfo /= Nothing+ stash result $ pong == "pong" && pInfo /= Nothing where launchRemote :: MVar ProcessId -> IO () launchRemote locMV = do@@ -97,8 +99,10 @@ _ <- receiveWait [ match (\(pid, "ping") -> send pid "pong") ]- "stop" <- expect- return ()+ res <- expect+ case res of+ "stop" -> return ()+ _ -> die $ "unexpected message received: " ++ res return () withActiveRemote :: LocalNode
src/Control/Distributed/Process/Tests/Tracing.hs view
@@ -125,9 +125,11 @@ (\ev -> case ev of (MxSent to from msg) -> do- (Just s) <- unwrapMessage msg :: Process (Maybe String)- stash res (to == pid && from == self && s == "hello there")- stash res (to == pid && from == self)+ mS <- unwrapMessage msg :: Process (Maybe String)+ case mS of+ (Just s) -> do stash res (to == pid && from == self && s == "hello there")+ stash res (to == pid && from == self)+ _ -> die "failed state invariant, message type unmatched..." _ -> return ()) $ do send pid "hello there"@@ -246,12 +248,8 @@ -- Here we set up that relay, and then wait for a signal -- that the tracer (on node1) has seen the expected -- MxSpawned message, at which point we're finished- (Just log') <- whereis "logger"- pid <- liftIO $ forkProcess node2 $ do- logRelay <- spawnLocal $ relay log'- reregister "logger" logRelay- getSelfNode >>= stash mvNid >> (expect :: Process ()) + pid <- splinchLogger node2 mvNid nid <- liftIO $ takeMVar mvNid mref <- monitor pid observedPid <- liftIO $ newEmptyMVar@@ -291,6 +289,17 @@ -- and just to be polite... liftIO $ closeLocalNode node2 + where+ splinchLogger n2 mv = do+ mLog <- whereis "logger"+ case mLog of+ Nothing -> die "no logger registered"+ Just log' -> do liftIO $ forkProcess n2 $ do+ logRelay <- spawnLocal $ relay log'+ reregister "logger" logRelay+ getSelfNode >>= stash mv >> (expect :: Process ())++ -- | Sets the value of an environment variable while executing the given IO -- computation and restores the preceeding value upon completion. withEnv :: String -> String -> IO a -> IO a@@ -319,9 +328,12 @@ -- Wait for the trace message. receiveWait [ matchIf interestingMessage' $ const $ return () ] -- Only one interesting message should arrive.- Nothing <- receiveTimeout 100000- [ matchIf interestingMessage' $ const $ return () ]- return ()+ expectedTimeout <- receiveTimeout+ 100000+ [ matchIf interestingMessage' $ const $ return () ]+ case expectedTimeout of+ Nothing -> return ()+ Just _ -> die "Unexpected message arrived..." -- | Tests that one and only one trace message is produced when a message is
tests/runInMemory.hs view
@@ -8,7 +8,6 @@ import Network.Transport.InMemory import Test.Framework (defaultMainWithArgs) -import Control.Concurrent (threadDelay) import System.Environment (getArgs) main :: IO ()
tests/runTCP.hs view
@@ -5,11 +5,12 @@ import TEST_SUITE_MODULE (tests) import Network.Transport.Test (TestTransport(..))-import Network.Socket (sClose)+import Network.Socket (close) import Network.Transport.TCP ( createTransportExposeInternals , TransportInternals(socketBetween) , defaultTCPParameters+ , defaultTCPAddr , TCPParameters(..) ) import Test.Framework (defaultMainWithArgs)@@ -24,14 +25,13 @@ hSetBuffering stdout LineBuffering hSetBuffering stderr LineBuffering Right (transport, internals) <-- createTransportExposeInternals "127.0.0.1" "8080"- (\sn -> ("127.0.0.1", sn))+ createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0") defaultTCPParameters { transportConnectTimeout = Just 3000000 } ts <- tests TestTransport { testTransport = transport , testBreakConnection = \addr1 addr2 -> do esock <- try $ socketBetween internals addr1 addr2- either (\e -> const (return ()) (e :: IOException)) sClose esock+ either (\e -> const (return ()) (e :: IOException)) close esock threadDelay 10000 } args <- getArgs