distributed-process-tests 0.4.13 → 0.5.0
raw patch · 10 files changed
+552/−1276 lines, 10 filesdep +ghc-primdep +rematchdep −setenvdep ~ansi-terminaldep ~basedep ~binarysetup-changedPVP ok
version bump matches the API change (PVP)
Dependencies added: ghc-prim, rematch
Dependencies removed: setenv
Dependency ranges changed: ansi-terminal, base, binary, bytestring, distributed-process, exceptions, network, network-transport-tcp
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: expectThat :: a -> Matcher a -> Process ()
+ Control.Distributed.Process.Tests.Internal.Utils: shouldBe :: a -> Matcher a -> Process ()
+ Control.Distributed.Process.Tests.Internal.Utils: shouldMatch :: a -> Matcher a -> Process ()
Files
- Setup.hs +2/−0
- distributed-process-tests.cabal +53/−90
- src/Control/Distributed/Process/Tests/CH.hs +296/−514
- src/Control/Distributed/Process/Tests/Closure.hs +51/−91
- src/Control/Distributed/Process/Tests/Internal/Utils.hs +32/−23
- src/Control/Distributed/Process/Tests/Mx.hs +54/−390
- src/Control/Distributed/Process/Tests/Receive.hs +38/−34
- src/Control/Distributed/Process/Tests/Stats.hs +11/−11
- src/Control/Distributed/Process/Tests/Tracing.hs +11/−112
- tests/runTCP.hs +4/−11
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
distributed-process-tests.cabal view
@@ -1,38 +1,21 @@-cabal-version: 3.0 name: distributed-process-tests-version: 0.4.13+version: 0.5.0 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, specifically the core distributed-process library.-license: BSD-3-Clause+homepage: http://github.com/haskell-distributed/distributed-process-tests+license: BSD3 license-file: LICENSE Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-maintainer: The Distributed Haskell team+Maintainer: watson.timothy@gmail.com, edsko@well-typed.com, duncan@well-typed.com copyright: Well-Typed LLP category: Control, Cloud Haskell build-type: Simple--source-repository head- Type: git- Location: https://github.com/haskell-distributed/distributed-process- SubDir: packages/distributed-process-tests+cabal-version: >=1.10 flag tcp Description: build and run TCP tests Default: False -common warnings- ghc-options: -Wall- -Wcompat- -Widentities- -Wincomplete-uni-patterns- -Wincomplete-record-updates- -Wredundant-constraints- -fhide-source-paths- -Wpartial-fields- library- import: warnings exposed-modules: Network.Transport.Test Control.Distributed.Process.Tests.CH Control.Distributed.Process.Tests.Closure@@ -41,146 +24,126 @@ Control.Distributed.Process.Tests.Stats Control.Distributed.Process.Tests.Tracing Control.Distributed.Process.Tests.Internal.Utils- Build-Depends: base >= 4.14 && < 5,- ansi-terminal >= 0.5,- binary >= 0.8 && < 0.9,- bytestring >= 0.10 && < 0.13,- distributed-process >= 0.6.0 && < 0.8,+ Build-Depends: base >= 4.4 && < 5,+ ansi-terminal >= 0.5 && < 0.7,+ binary >= 0.5 && < 0.10,+ bytestring >= 0.9 && < 0.13,+ distributed-process >= 0.5.3 && < 0.8, distributed-static,- exceptions >= 0.10,+ exceptions >=0.10 && <0.11, HUnit >= 1.2 && < 1.7, network-transport >= 0.4.1.0 && < 0.6, network >= 2.5 && < 3.3, random >= 1.0 && < 1.3,- setenv >= 0.1.1.3,+ rematch >= 0.1.2.1, test-framework >= 0.6 && < 0.9, test-framework-hunit >= 0.2.0 && < 0.4, stm hs-source-dirs: src- default-language: Haskell98- ghc-options: -fno-warn-unused-do-bind- default-extensions: CPP,- ExistentialQuantification,- FlexibleInstances,- DeriveDataTypeable,- DeriveGeneric,- GeneralizedNewtypeDeriving,- RankNTypes,- RecordWildCards,- ScopedTypeVariables+ ghc-options: -Wall -fno-warn-unused-do-bind+ default-extensions:CPP,+ ExistentialQuantification,+ FlexibleInstances,+ DeriveDataTypeable,+ DeriveGeneric,+ GeneralizedNewtypeDeriving,+ RankNTypes,+ RecordWildCards,+ ScopedTypeVariables+ default-language: Haskell2010+ if impl(ghc <= 7.4.2)+ Build-Depends: ghc-prim == 0.2.0.0 Test-Suite TestCHInMemory- import: warnings Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH- Build-Depends: base >= 4.14 && < 5,+ Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 3.3, 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: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions:CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests+ default-language: Haskell2010 Test-Suite TestCHInTCP- import: warnings Type: exitcode-stdio-1.0 Main-Is: runTCP.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH if flag(tcp)- Build-Depends: base >= 4.14 && < 5,+ Build-Depends: base >= 4.4 && < 5, distributed-process-tests,- network >= 2.5 && < 3.2,+ network >= 2.3 && < 3.3, network-transport >= 0.4.1.0 && < 0.6,- network-transport-tcp >= 0.5 && < 0.9,+ network-transport-tcp >= 0.3 && < 0.5, test-framework >= 0.6 && < 0.9 else Buildable: False- default-extensions: CPP- default-language: Haskell98- ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions:CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests+ default-language: Haskell2010 Test-Suite TestClosure- import: warnings Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure- Build-Depends: base >= 4.14 && < 5,+ Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 3.3, 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: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions:CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests+ default-language: Haskell2010 Test-Suite TestStats- import: warnings Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats- Build-Depends: base >= 4.14 && < 5,+ Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 3.3, 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: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions:CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests+ default-language: Haskell2010 -Test-Suite TestMxInMemory- import: warnings+Test-Suite TestMx Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx- Build-Depends: base >= 4.14 && < 5,+ Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 3.3, 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: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions:CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests+ default-language: Haskell2010 -Test-Suite TestTracingInMemory- import: warnings+Test-Suite TestTracing Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing- Build-Depends: base >= 4.14 && < 5,- distributed-process-tests,- network >= 2.3 && < 3.3,- 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: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind- HS-Source-Dirs: tests--Test-Suite TestMxInTCP- import: warnings- Type: exitcode-stdio-1.0- Main-Is: runInMemory.hs- CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx- Build-Depends: base >= 4.14 && < 5,+ Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 3.3, 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: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions:CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests+ default-language: Haskell2010
src/Control/Distributed/Process/Tests/CH.hs view
@@ -1,12 +1,14 @@ module Control.Distributed.Process.Tests.CH (tests) where +#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif import Network.Transport.Test (TestTransport(..)) import Data.Binary (Binary(..)) import Data.Typeable (Typeable) import Data.Foldable (forM_)-import Data.Function (fix) import Data.IORef ( readIORef , writeIORef@@ -21,30 +23,26 @@ , readMVar ) import Control.Monad (replicateM_, replicateM, forever, void, unless, join)+import Control.Monad.Catch as Ex (catch, finally, mask, onException, try) import Control.Exception (SomeException, throwIO, ErrorCall(..))-import Control.Monad.Catch (try, catch, finally, mask, onException)-import Control.Applicative ((<|>))+import Control.Applicative ((<$>), (<*>), pure, (<|>)) import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)-import Control.Distributed.Process hiding- ( try- , catch- , finally- , mask- , onException- )+import Control.Distributed.Process import Control.Distributed.Process.Internal.Types- ( LocalNode(localEndPoint)+ ( NodeId(nodeAddress)+ , 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 Data.Maybe (isNothing, isJust)-import Test.HUnit (Assertion, assertBool, assertEqual, assertFailure)++import Test.HUnit (Assertion, assertFailure) import Test.Framework (Test, testGroup) import Test.Framework.Providers.HUnit (testCase)+import Control.Rematch hiding (match)+import Control.Rematch.Run (Match(..)) newtype Ping = Ping ProcessId deriving (Typeable, Binary, Show)@@ -56,11 +54,17 @@ -- Supporting definitions -- -------------------------------------------------------------------------------- +expectThat :: a -> Matcher a -> Assertion+expectThat a matcher = case res of+ MatchSuccess -> return ()+ (MatchFailure msg) -> assertFailure msg+ where res = runMatch matcher a+ -- | Like fork, but throw exceptions in the child thread to the parent forkTry :: IO () -> IO ThreadId forkTry p = do tid <- myThreadId- forkIO $ catch p (\e -> throwTo tid (e :: SomeException))+ forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException)) -- | The ping server from the paper ping :: Process ()@@ -70,38 +74,12 @@ 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- 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 ()+ WhereIsReply _ mPid <- expect+ return mPid data Add = Add ProcessId Double Double deriving (Typeable) data Divide = Divide ProcessId Double Double deriving (Typeable)@@ -167,27 +145,22 @@ -> MVar () -- Signal for successful termination -> Process () monitorTestProcess theirAddr mOrL un reason monitorSetup done =- catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup- case (un, mRef) of- (True, Nothing) -> do- unlink theirAddr- liftIO $ putMVar done ()- (True, Just ref) -> do- unmonitor ref- liftIO $ putMVar done ()- (False, ref) -> do- receiveWait [- match (\(ProcessMonitorNotification ref' pid reason') -> do- liftIO $ do- assertBool "Bad Monitor Signal"- (Just ref' == ref && pid == theirAddr &&- mOrL && reason == reason')- putMVar done ())- ]- )+ Ex.catch (do + mRef <- monitorOrLink mOrL theirAddr monitorSetup+ case (un, mRef) of+ (True, Nothing) -> do+ unlink theirAddr+ liftIO $ putMVar done ()+ (True, Just ref) -> do+ 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 ()+ ) (\(ProcessLinkException pid reason') -> do- (liftIO $ assertBool "link exception unmatched" $- pid == theirAddr && not mOrL && not un && reason == reason')+ True <- return $ pid == theirAddr && not mOrL && not un && reason == reason' liftIO $ putMVar done () ) @@ -218,10 +191,8 @@ pid <- getSelfPid replicateM_ numPings $ do send pingServer (Pong pid)- p <- expectTimeout 3000000- case p of- Just (Ping _) -> return ()- Nothing -> die "Failed to receive Ping"+ Ping _ <- expect+ return () putMVar clientDone () @@ -308,7 +279,7 @@ runProcess localNode $ do monitor theirAddr -- wait for the process to die- expect :: Process ProcessMonitorNotification+ ProcessMonitorNotification _ _ _ <- expect monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done takeMVar done@@ -382,17 +353,14 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- three <- expect :: Process Double+ 3 <- expect :: Process Double send mathServer (Divide pid 8 2)- four <- expect :: Process Double+ 4 <- expect :: Process Double send mathServer (Divide pid 8 0)- divByZ <- expect- liftIO $ putMVar clientDone (three, four, divByZ)+ DivByZero <- expect+ liftIO $ putMVar clientDone () - res <- takeMVar clientDone- case res of- (3, 4, DivByZero) -> return ()- _ -> error $ "Something went horribly wrong"+ takeMVar clientDone -- | Send first message (i.e. connect) to an already terminated process -- (without monitoring); then send another message to a second process on@@ -420,9 +388,11 @@ pid <- getSelfPid send server1 "Hi" send server2 (Pong pid)- expectPing clientDone+ Ping pid' <- expect+ True <- return $ pid' == server2+ liftIO $ putMVar clientDone () - verifyClient "Expected Ping from server" clientDone+ takeMVar clientDone -- | Test (non-zero) timeout testTimeout :: TestTransport -> Assertion@@ -431,24 +401,26 @@ done <- newEmptyMVar runProcess localNode $ do- res <- receiveTimeout 1000000 [match (\Add{} -> return ())]- liftIO $ putMVar done $ res == Nothing+ Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]+ liftIO $ putMVar done () - verifyClient "Expected receiveTimeout to timeout..." done+ takeMVar done -- | Test zero timeout testTimeout0 :: TestTransport -> Assertion testTimeout0 TestTransport{..} = do serverAddr <- newEmptyMVar clientDone <- newEmptyMVar+ messagesSent <- newEmptyMVar forkIO $ do localNode <- newLocalNode testTransport initRemoteTable addr <- forkProcess localNode $ do+ liftIO $ readMVar messagesSent >> threadDelay 1000000 -- Variation on the venerable ping server which uses a zero timeout- partner <- fix $ \loop ->- receiveTimeout 0 [match (\(Pong partner) -> return partner)]- >>= maybe (liftIO (threadDelay 100000) >> loop) return+ -- Since we wait for all messages to be sent before doing this receive,+ -- we should nevertheless find the right message immediately+ Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)] self <- getSelfPid send partner (Ping self) putMVar serverAddr addr@@ -462,9 +434,11 @@ -- is not interested in, and then a single message that it wants replicateM_ 10000 $ send server "Irrelevant message" send server (Pong pid)- expectPing clientDone+ liftIO $ putMVar messagesSent ()+ Ping _ <- expect+ liftIO $ putMVar clientDone () - verifyClient "Expected Ping from server" clientDone+ takeMVar clientDone -- | Test typed channels testTypedChannels :: TestTransport -> Assertion@@ -487,10 +461,10 @@ runProcess localNode $ do (clientSendPort, rport) <- newChan sendChan serverSendPort (clientSendPort, 5)- ch <- receiveChan rport- liftIO $ putMVar clientDone $ ch == False+ False <- receiveChan rport+ liftIO $ putMVar clientDone () - verifyClient "Expected channel to send 'False'" clientDone+ takeMVar clientDone -- | Test merging receive ports testMergeChannels :: TestTransport -> Assertion@@ -513,8 +487,9 @@ rs <- mapM charChannel "abc" m <- mergePorts biased rs xs <- replicateM 9 $ receiveChan m- liftIO $ putMVar done $ xs == expected- verifyClient "Expected single layer merge to match expected ordering" done+ True <- return $ xs == expected+ liftIO $ putMVar done ()+ takeMVar done -- Two layers of merging testNested :: LocalNode -> Bool -> Bool -> String -> IO ()@@ -525,8 +500,9 @@ ms <- mapM (mergePorts biasedInner) rss m <- mergePorts biasedOuter ms xs <- replicateM (9 * 3) $ receiveChan m- liftIO $ putMVar done $ xs == expected- verifyClient "Expected nested channels to match expeted ordering" done+ True <- return $ xs == expected+ liftIO $ putMVar done ()+ takeMVar done -- Test that if no messages are (immediately) available, the scheduler makes no difference testBlocked :: LocalNode -> Bool -> IO ()@@ -535,45 +511,43 @@ done <- newEmptyMVar forkProcess localNode $ do- 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"+ [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')+ ] 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- liftIO $ putMVar done $ xs == "abcacbbacbcacabcba"+ True <- return $ xs == "abcacbbacbcacabcba"+ liftIO $ putMVar done () - verifyClient "Expected merged ports to match expected ordering" done+ takeMVar done mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a) mergePorts True = mergePortsBiased@@ -589,12 +563,20 @@ testTerminate :: TestTransport -> Assertion testTerminate TestTransport{..} = do localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pid <- forkProcess localNode $ do+ liftIO $ threadDelay 100000+ terminate+ runProcess localNode $ do- e <- try terminate :: Process (Either ProcessTerminationException ())- if either show show e == show ProcessTerminationException- then return ()- else die "Unexpected result from terminate"+ ref <- monitor pid+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect+ True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException+ liftIO $ putMVar done () + takeMVar done+ testMonitorNode :: TestTransport -> Assertion testMonitorNode TestTransport{..} = do [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable@@ -604,12 +586,11 @@ runProcess node2 $ do ref <- monitorNode (localNodeId node1)- receiveWait [- match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->- return $ ref == ref' && nid == localNodeId node1)- ] >>= liftIO . putMVar done+ NodeMonitorNotification ref' nid DiedDisconnect <- expect+ True <- return $ ref == ref' && nid == localNodeId node1+ liftIO $ putMVar done () - verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done+ takeMVar done testMonitorLiveNode :: TestTransport -> Assertion testMonitorLiveNode TestTransport{..} = do@@ -624,16 +605,15 @@ liftIO $ putMVar ready () liftIO $ takeMVar readyr send p ()- receiveWait [- match (\(NodeMonitorNotification ref' nid _) ->- (return $ ref == ref' && nid == localNodeId node1))- ] >>= liftIO . putMVar done+ NodeMonitorNotification ref' nid _ <- expect+ True <- return $ ref == ref' && nid == localNodeId node1+ liftIO $ putMVar done () takeMVar ready closeLocalNode node1 putMVar readyr () - verifyClient "Expected NodeMonitorNotification for LIVE node" done+ takeMVar done testMonitorChannel :: TestTransport -> Assertion testMonitorChannel TestTransport{..} = do@@ -643,20 +623,18 @@ pid <- forkProcess node1 $ do sport <- expect :: Process (SendPort ()) ref <- monitorPort sport- 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+ 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 () runProcess node2 $ do (sport, _) <- newChan :: Process (SendPort (), ReceivePort ()) send pid sport liftIO $ threadDelay 100000 - verifyClient "Expected PortMonitorNotification" gotNotification+ takeMVar gotNotification testRegistry :: TestTransport -> Assertion testRegistry TestTransport{..} = do@@ -668,28 +646,20 @@ runProcess node $ do register "ping" pingServer- whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)+ Just pid <- whereis "ping"+ True <- return $ pingServer == pid us <- getSelfPid nsend "ping" (Pong us)- receiveWait [- matchIf (\(Ping pid') -> pingServer == pid') (const $ return ())- ]- checkRegException "dead" Nothing deadProcess- checkRegException "ping" (Just pingServer) deadProcess- try (unregister "dead") >>= checkReg "dead" Nothing+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ Left (ProcessRegistrationException "dead" Nothing) <- Ex.try $ register "dead" deadProcess+ Left (ProcessRegistrationException "ping" (Just x)) <- Ex.try $ register "ping" deadProcess+ True <- return $ x == pingServer+ Left (ProcessRegistrationException "dead" Nothing) <- Ex.try $ unregister "dead" 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@@ -700,13 +670,13 @@ runProcess node2 $ do register "ping" pingServer- whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer)+ Just pid <- whereis "ping"+ True <- return $ pingServer == pid us <- getSelfPid nsend "ping" (Pong us)- receiveWait [- matchIf (\(Ping pid') -> pingServer == pid')- (const $ liftIO $ putMVar done ())- ]+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ liftIO $ putMVar done () takeMVar done @@ -714,6 +684,7 @@ testRemoteRegistry TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable node2 <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar pingServer <- forkProcess node1 ping deadProcess <- forkProcess node1 (return ())@@ -726,35 +697,33 @@ "ping" == label' && pid == pingServer) (\(RegisterReply _ _ _) -> return ()) ] - pid <- verifyWhereIsRemote nid1 "ping"- liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid-+ Just pid <- whereisRemote nid1 "ping"+ True <- return $ pingServer == pid us <- getSelfPid nsendRemote nid1 "ping" (Pong us)- receiveWait [- match (\(Ping pid') -> return $ pingServer == pid')- ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId"+ Ping pid' <- expect+ True <- return $ pingServer == pid' -- test that if process was not registered Nothing is returned -- in owner field. registerRemoteAsync nid1 "dead" deadProcess- receiveWait [ matchIf (\(RegisterReply label' _ _) -> "dead" == label')- (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))- ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"-+ receiveWait [+ matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')+ (\(RegisterReply _ _ _) -> return ()) ] registerRemoteAsync nid1 "ping" deadProcess receiveWait [- 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"-+ matchIf (\(RegisterReply label' False (Just pid)) ->+ "ping" == label' && pid == pingServer)+ (\(RegisterReply _ _ _) -> return ()) ] unregisterRemoteAsync nid1 "dead" receiveWait [- matchIf (\(RegisterReply label' _ _) -> "dead" == label')- (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))- ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"+ matchIf (\(RegisterReply label' False Nothing) ->+ "dead" == label' && pid == pingServer)+ (\(RegisterReply _ _ _) -> return ()) ]+ liftIO $ putMVar done () + takeMVar done+ testRemoteRegistryRemoteProcess :: TestTransport -> Assertion testRemoteRegistryRemoteProcess TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable@@ -769,15 +738,15 @@ receiveWait [ matchIf (\(RegisterReply label' _ _) -> "ping" == label') (\(RegisterReply _ _ _) -> return ()) ]- pid <- verifyWhereIsRemote nid1 "ping"- liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid+ Just pid <- whereisRemote nid1 "ping"+ True <- return $ pingServer == pid us <- getSelfPid nsendRemote nid1 "ping" (Pong us)- receiveWait [- match (\(Ping pid') -> return $ pingServer == pid')- ] >>= liftIO . putMVar done+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ liftIO $ putMVar done () - verifyClient "Expected Ping with ping server's ProcessId" done+ takeMVar done testSpawnLocal :: TestTransport -> Assertion testSpawnLocal TestTransport{..} = do@@ -791,14 +760,15 @@ sport <- expect sendChan sport (1234 :: Int) - sport <- spawnChannelLocal $- \rport -> (receiveChan rport :: Process Int) >>= send us+ sport <- spawnChannelLocal $ \rport -> do+ (1234 :: Int) <- receiveChan rport+ send us () send pid sport- expect >>= liftIO . putMVar done+ () <- expect+ liftIO $ putMVar done () - res <- takeMVar done- assertBool "Expected 1234 :: Int" $ res == (1234 :: Int)+ takeMVar done testSpawnAsyncStrictness :: TestTransport -> Assertion testSpawnAsyncStrictness TestTransport{..} = do@@ -808,7 +778,7 @@ runProcess node $ do here <-getSelfNode - ev <- try $ spawnAsync here (error "boom")+ ev <- Ex.try $ spawnAsync here (error "boom") liftIO $ case ev of Right _ -> putMVar done (error "Exception didn't fire") Left (_::SomeException) -> putMVar done (return ())@@ -827,9 +797,8 @@ liftIO $ putMVar processA us msg1 <- expect msg2 <- expect- liftIO $ do- assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"- putMVar sendTestOk ()+ True <- return $ msg1 == "message 1" && msg2 == "message 3"+ liftIO $ putMVar sendTestOk () forkProcess node2 $ do {-@@ -862,7 +831,7 @@ matchIf (\(RegisterReply label' _ _) -> "a" == label') (\(RegisterReply _ _ _) -> return ()) ] - _ <- whereisRemote nid1 "a"+ Just _ <- whereisRemote nid1 "a" -- Simulate network failure@@ -881,10 +850,9 @@ (\(RegisterReply _ _ _) -> return ()) ] -- Check- 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"+ 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" liftIO $ putMVar registerTestOk () @@ -901,7 +869,7 @@ processA <- newEmptyMVar usendTestOk <- newEmptyMVar - forkProcess node1 $ flip catch (\e -> liftIO $ print (e :: SomeException) ) $ do+ forkProcess node1 $ flip Ex.catch (\e -> liftIO $ print (e :: SomeException) ) $ do us <- getSelfPid liftIO $ putMVar processA us them <- expect@@ -925,19 +893,19 @@ isSorted _ = True -- The list can't be null since there are no failures after sending -- the last message.- liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)+ True <- return $ isSorted msgs && not (null msgs)+ liftIO $ putMVar usendTestOk () forkProcess node2 $ do them <- liftIO $ readMVar processA getSelfPid >>= send them- expect :: Process ()+ () <- expect forM_ [1..numMessages] $ \i -> do liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) usendPrim them i liftIO (threadDelay 30000) - res <- takeMVar usendTestOk- assertBool "Unexpected failure after sending last msg" res+ takeMVar usendTestOk -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between@@ -963,17 +931,14 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- three <- expect :: Process Double+ 3 <- expect :: Process Double send mathServer (Divide pid 8 2)- four <- expect :: Process Double+ 4 <- expect :: Process Double send mathServer (Divide pid 8 0)- divByZ <- expect- liftIO $ putMVar clientDone (three, four, divByZ)+ DivByZero <- expect+ liftIO $ putMVar clientDone () - res <- takeMVar clientDone- case res of- (3, 4, DivByZero) -> return ()- _ -> error "Unexpected result"+ takeMVar clientDone -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between, however we block 'Divide' requests ....@@ -1000,13 +965,12 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- three <- expect :: Process Double+ 3 <- expect :: Process Double send mathServer (Divide pid 8 2)- res <- (expectTimeout 100000) :: Process (Maybe Double)- liftIO $ putMVar clientDone $ three == 3 && res == Nothing-- verifyClient "Expected Nothing (i.e. timeout)" clientDone+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)+ liftIO $ putMVar clientDone () + takeMVar clientDone where maybeForward :: ProcessId -> Message -> Process (Maybe ()) maybeForward s msg = handleMessage msg (\m@(Add _ _ _) -> send s m)@@ -1034,10 +998,8 @@ Just _ -> die "NONSENSE") ] -- we *must* have removed the message from our mailbox though!!!- res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]- liftIO $ do- assertBool "Expected timeout!" $ res == Nothing- putMVar serverDone ()+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]+ liftIO $ putMVar serverDone () putMVar addr server -- Client@@ -1080,21 +1042,14 @@ runProcess localNode $ do pid <- getSelfPid send server (pid, "foo")- foo <- expect- -- provoking what would be the wrong ordering is informative here...-- send server (pid, "bar")- bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !?-+ "foo" <- expect 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'+ "baz" <- expect+ send server (pid, "bar")+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)+ liftIO $ putMVar clientDone () + takeMVar clientDone where tryHandleMessage :: Message -> Process (Maybe ()) tryHandleMessage msg = handleMessage msg (\(pid :: ProcessId, (m :: String))@@ -1127,19 +1082,17 @@ 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 (foo, baz)+ "baz" <- expect+ liftIO $ putMVar clientDone () - res <- takeMVar clientDone- assertBool "Unexpected unwrapped results" $ res == ("foo", "baz")+ takeMVar clientDone -- Test 'receiveChanTimeout' testReceiveChanTimeout :: TestTransport -> Assertion testReceiveChanTimeout TestTransport{..} = do done <- newEmptyMVar- mvSender <- newEmptyMVar sendPort <- newEmptyMVar forkTry $ do@@ -1149,32 +1102,18 @@ (sp, rp) <- newChan :: Process (SendPort Bool, ReceivePort Bool) liftIO $ putMVar sendPort sp - -- Wait for a message with a delay. No message arrives, we should get- -- Nothing after the delay.- 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- res <- receiveChanTimeout 20000000 rp >>= maybe (die "Timeout") return- liftIO $ assertBool "Expected result to be 'True'" res+ -- Wait for a message with a delay. No message arrives, we should get Nothing after 1 second+ Nothing <- receiveChanTimeout 1000000 rp - -- Wait for a message with zero timeout: non-blocking check. No message is- -- available, we get Nothing- receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout")+ -- Wait for a message with a delay again. Now a message arrives after 0.5 seconds+ Just True <- receiveChanTimeout 1000000 rp - -- Let the sender know that it can send a message.- liftIO $ putMVar mvSender ()+ -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing+ Nothing <- receiveChanTimeout 0 rp -- Again, but now there is a message available- fix $ \loop -> do- liftIO $ threadDelay 100000- mb <- receiveChanTimeout 0 rp- case mb of- Just b -> do liftIO $ assertBool "Unexpected Message" $ not b- _ -> loop+ liftIO $ threadDelay 1000000+ Just False <- receiveChanTimeout 0 rp liftIO $ putMVar done () @@ -1183,11 +1122,10 @@ runProcess localNode $ do sp <- liftIO $ readMVar sendPort - liftIO $ takeMVar mvSender- liftIO $ threadDelay 100000+ liftIO $ threadDelay 1500000 sendChan sp True - liftIO $ takeMVar mvSender+ liftIO $ threadDelay 500000 sendChan sp False takeMVar done@@ -1210,8 +1148,8 @@ rp1 <- mergePortsBiased [even <$> rpInt, rpBool] - receiveChan rp1 >>= liftIO . assertBool "Expected True"- receiveChan rp1 >>= liftIO . assertBool "Expected False" . not+ True <- receiveChan rp1+ False <- receiveChan rp1 -- Test Applicative instance @@ -1220,7 +1158,7 @@ let rp2 = pure (+) <*> rpInt <*> rpInt - receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7)+ 7 <- receiveChan rp2 -- Test Alternative instance @@ -1229,8 +1167,8 @@ let rp3 = (even <$> rpInt) <|> rpBool - receiveChan rp3 >>= liftIO . assertBool "Expected False" . not- receiveChan rp3 >>= liftIO . assertBool "Expected True"+ False <- receiveChan rp3+ True <- receiveChan rp3 -- Test Monad instance @@ -1244,86 +1182,17 @@ then rpInt else return 7 - receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)- receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7)+ 5 <- receiveChan rp4+ 7 <- receiveChan rp4 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@@ -1332,18 +1201,17 @@ ref <- monitor pid us <- getSelfPid kill pid "TestKill"- 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"+ ProcessMonitorNotification ref' pid' (DiedException ex) <- expect+ True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"+ liftIO $ putMVar done () + takeMVar done+ testKillRemote :: TestTransport -> Assertion testKillRemote TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable node2 <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar pid <- forkProcess node1 $ do liftIO $ threadDelay 1000000@@ -1352,18 +1220,12 @@ ref <- monitor pid us <- getSelfPid kill pid "TestKill"- 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"+ ProcessMonitorNotification ref' pid' (DiedException reason) <- expect+ True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"+ liftIO $ putMVar done () + takeMVar done+ testCatchesExit :: TestTransport -> Assertion testCatchesExit TestTransport{..} = do localNode <- newLocalNode testTransport initRemoteTable@@ -1388,16 +1250,13 @@ self <- getSelfPid send self (5 :: Integer, 10 :: Integer) msg <- receiveWait [ matchMessage return ]- handleMessageIf msg- (\() -> True)- (\() -> die "whoops") >>= maybe (return ())- (const $ die "Expected Mismatch")+ Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops") handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10) (\input -> liftIO $ putMVar done input) return () result <- takeMVar done- assertEqual mempty (5, 10) result+ expectThat result $ equalTo (5, 10) testCatches :: TestTransport -> Assertion testCatches TestTransport{..} = do@@ -1419,13 +1278,13 @@ parentPid <- newEmptyMVar :: IO (MVar ProcessId) spawnedPid <- newEmptyMVar :: IO (MVar ProcessId) - void $ runProcess localNode $ mask $ \unmask -> do+ void $ runProcess localNode $ Ex.mask $ \unmask -> do getSelfPid >>= liftIO . putMVar parentPid void $ spawnLocal $ unmask (getSelfPid >>= liftIO . putMVar spawnedPid) parent <- liftIO $ takeMVar parentPid child <- liftIO $ takeMVar spawnedPid- assertBool mempty (not $ parent == child)+ expectThat parent $ isNot $ equalTo child testDie :: TestTransport -> Assertion testDie TestTransport{..} = do@@ -1436,11 +1295,8 @@ (die ("foobar", 123 :: Int)) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- let res = reason == ("foobar", 123 :: Int)+ True <- return $ reason == ("foobar", 123 :: Int) liftIO $ putMVar done ()- if res- then return ()- else die "Something went horribly wrong" takeMVar done @@ -1451,14 +1307,11 @@ _ <- forkProcess localNode $ do (die "timeout")- `catch` \ex@(ProcessExitException from _) ->+ `Ex.catch` \ex@(ProcessExitException from _) -> let expected = "exit-from=" ++ (show from) in do- let res = (show ex) == expected+ True <- return $ (show ex) == expected liftIO $ putMVar done ()- if res- then return ()- else die "Something went horribly wrong" takeMVar done @@ -1474,26 +1327,17 @@ pid <- forkProcess localNode $ do (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- res <- return $ reason == "TestExit"+ True <- 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- 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"+ ProcessMonitorNotification ref' pid' DiedNormal <- expect+ True <- return $ ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone () takeMVar supervisedDone takeMVar supervisorDone@@ -1509,98 +1353,19 @@ (receiveWait [] :: Process ()) -- block forever `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- res <- return $ reason == "TestExit"+ True <- 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"- 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"+ ProcessMonitorNotification ref' pid' DiedNormal <- expect+ True <- return $ ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone () 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@@ -1616,7 +1381,8 @@ void $ forkProcess localNode $ do serverPid <- liftIO $ takeMVar serverAddr getSelfPid >>= unsafeSend serverPid- expect >>= liftIO . putMVar clientDone+ () <- expect+ liftIO $ putMVar clientDone () takeMVar clientDone @@ -1635,7 +1401,8 @@ void $ forkProcess localNode $ do serverPid <- liftIO $ takeMVar serverAddr getSelfPid >>= unsafeUSend serverPid- expect >>= liftIO . putMVar clientDone+ () <- expect+ liftIO $ putMVar clientDone () takeMVar clientDone @@ -1646,7 +1413,8 @@ localNode <- newLocalNode testTransport initRemoteTable pid <- forkProcess localNode $ do- expect >>= liftIO . putMVar clientDone+ () <- expect+ liftIO $ putMVar clientDone () void $ runProcess localNode $ do register "foobar" pid@@ -1664,7 +1432,8 @@ _ <- forkProcess localNode1 $ do getSelfPid >>= register "foobar" liftIO $ putMVar clientDone ()- expect >>= liftIO . putMVar clientDone+ () <- expect+ liftIO $ putMVar clientDone () takeMVar clientDone void $ runProcess localNode2 $ do@@ -1688,7 +1457,7 @@ serverPid <- liftIO $ takeMVar serverAddr (sp, rp) <- newChan unsafeSend serverPid sp- receiveChan rp :: Process ()+ () <- receiveChan rp liftIO $ putMVar clientDone () takeMVar clientDone@@ -1702,8 +1471,8 @@ runProcess node $ do r <- (/=) <$> getSelfPid <*> callLocal getSelfPid liftIO $ putMVar result r-- takeMVar result >>= assertBool "Expected 'True'"+ True <- takeMVar result+ return () -- Testing that when callLocal is interrupted, the worker is interrupted. ibox <- newIORef False@@ -1712,27 +1481,28 @@ spawnLocal $ do caller <- getSelfPid send keeper caller- onException+ Ex.onException (callLocal $ do- onException (do send keeper caller- expect)- (do liftIO $ writeIORef ibox True))+ Ex.onException (do send keeper caller+ expect)+ (do liftIO $ writeIORef ibox True)) (send keeper ()) caller <- expect exit caller "test"- expect :: Process ()-- readIORef ibox >>= assertBool "Expected 'True'"+ () <- expect+ return ()+ True <- readIORef ibox+ return () -- Testing that when the worker raises an exception, the exception is propagated to the parent. ibox2 <- newIORef False runProcess node $ do- r <- try (callLocal $ error "e" >> return ())+ r <- Ex.try (callLocal $ error "e" >> return ()) liftIO $ writeIORef ibox2 $ case r of Left (ErrorCall "e") -> True _ -> False-- readIORef ibox >>= assertBool "Expected 'True'"+ True <- readIORef ibox+ return () -- Test that caller waits for the worker in correct situation ibox3 <- newIORef False@@ -1745,12 +1515,12 @@ send keeper us () <- expect liftIO yield)- `finally` (liftIO $ writeIORef ibox3 True)+ `Ex.finally` (liftIO $ writeIORef ibox3 True) liftIO $ putMVar result3 =<< readIORef ibox3 worker <- expect send worker ()-- takeMVar result3 >>= assertBool "Expected 'True'"+ True <- takeMVar result3+ return () -- Test that caller waits for the worker in case when caller gets an exception ibox4 <- newIORef False@@ -1762,14 +1532,18 @@ callLocal ((do send keeper caller expect)- `finally` (liftIO $ writeIORef ibox4 True))- `finally` (liftIO $ putMVar result4 =<< readIORef ibox4)+ `Ex.finally` (liftIO $ writeIORef ibox4 True))+ `Ex.finally` (liftIO $ putMVar result4 =<< readIORef ibox4) caller <- expect exit caller "hi!"-- takeMVar result4 >>= assertBool "Expected 'True'"+ True <- takeMVar result4+ return () -- XXX: Testing that when mask_ $ callLocal p runs p in masked state. ++++ tests :: TestTransport -> IO [Test] tests testtrans = return [ testGroup "Basic features" [@@ -1779,7 +1553,9 @@ , testCase "Timeout0" (testTimeout0 testtrans) , testCase "SendToTerminated" (testSendToTerminated testtrans) , testCase "TypedChannnels" (testTypedChannels testtrans)+ , testCase "MergeChannels" (testMergeChannels testtrans) , testCase "Terminate" (testTerminate testtrans)+ , testCase "Registry" (testRegistry testtrans) , testCase "RegistryRemoteProcess" (testRegistryRemoteProcess testtrans) , testCase "RemoteRegistry" (testRemoteRegistry testtrans) , testCase "RemoteRegistryRemoteProcess" (testRemoteRegistryRemoteProcess testtrans)@@ -1793,7 +1569,6 @@ , 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)@@ -1803,7 +1578,6 @@ , 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)@@ -1824,7 +1598,8 @@ -- The "missing" combinations in the list below don't make much sense, as -- we cannot guarantee that the monitor reply or link exception will not -- happen before the unmonitor or unlink- testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False)+ testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False)+ , testCase "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False) , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False) , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False) , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False)@@ -1846,13 +1621,20 @@ , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans) , testCase "MonitorChannel" (testMonitorChannel testtrans) -- Reconnect+ , testCase "Reconnect" (testReconnect testtrans) ]-- -- Tests that fail occasionally and should be revised- -- , testGroup "Flaky" [- -- testCase "Reconnect" (testReconnect testtrans)- -- , testCase "Registry" (testRegistry testtrans)- -- , testCase "MergeChannels" (testMergeChannels testtrans)- -- , 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,4 +1,3 @@-{-# OPTIONS_GHC -Wno-unused-top-binds #-} {-# LANGUAGE TemplateHaskell, KindSignatures #-} module Control.Distributed.Process.Tests.Closure (tests) where @@ -20,12 +19,14 @@ , 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- ( createMessage+ ( NodeId(nodeAddress)+ , createMessage , messageToPayload ) import Control.Distributed.Static (staticLabel, staticClosure)@@ -168,11 +169,8 @@ node <- newLocalNode testTransport rtable done <- newEmptyMVar forkProcess node $ do- i <- join . unClosure $ factorialClosure 5+ 120 <- join . unClosure $ factorialClosure 5 liftIO $ putMVar done ()- if i == 720- then return ()- else error "Something went horribly wrong" takeMVar done testBind :: TestTransport -> RemoteTable -> Assertion@@ -182,11 +180,8 @@ runProcess node $ do us <- getSelfPid join . unClosure $ sendFac 6 us- (i :: Int) <- expect+ (720 :: Int) <- expect liftIO $ putMVar done ()- if i == 720- then return ()- else error "Something went horribly wrong" takeMVar done testSendPureClosure :: TestTransport -> RemoteTable -> Assertion@@ -199,7 +194,7 @@ addr <- forkProcess node $ do cl <- expect fn <- unClosure cl :: Process (Int -> Int)- (_ :: Int) <- return $ fn 6+ 13 <- return $ fn 6 liftIO $ putMVar serverDone () putMVar serverAddr addr @@ -223,11 +218,8 @@ liftIO $ do someMVar <- newEmptyMVar io someMVar- i <- readMVar someMVar+ 5 <- readMVar someMVar putMVar serverDone ()- if i == 5- then return ()- else error "Something went horribly wrong" putMVar serverAddr addr forkIO $ do@@ -256,10 +248,8 @@ runProcess node $ do pid <- getSelfPid send theirAddr (cpSend $(mkStatic 'sdictInt) pid)- i <- expect :: Process Int- if i == 5- then liftIO $ putMVar clientDone ()- else error "Something went horribly wrong"+ 5 <- expect :: Process Int+ liftIO $ putMVar clientDone () takeMVar clientDone @@ -279,9 +269,8 @@ pid <- getSelfPid pid' <- spawn nid (sendPidClosure pid) pid'' <- expect- if pid' == pid''- then liftIO $ putMVar clientDone ()- else error "Something went horribly wrong"+ True <- return $ pid' == pid''+ liftIO $ putMVar clientDone () takeMVar clientDone @@ -305,9 +294,8 @@ spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid pid' <- expect :: Process ProcessId pid'' <- expect :: Process ProcessId- if pid' == pid''- then return ()- else error "Something went horribly wrong"+ True <- return $ pid' == pid''+ return () where @@ -317,16 +305,14 @@ wrapEP e = e { NT.connect = \x y z -> do healthy <- newIORef True- fmap (fmap $ wrapConnection healthy e x) $ NT.connect e x y z+ fmap (fmap $ wrapConnection healthy) $ NT.connect e x y z } - wrapConnection :: IORef Bool -> NT.EndPoint -> NT.EndPointAddress- -> NT.Connection -> NT.Connection- wrapConnection healthy e remoteAddr (NT.Connection s closeC) =+ wrapConnection :: IORef Bool -> NT.Connection -> NT.Connection+ wrapConnection healthy (NT.Connection s closeC) = flip NT.Connection closeC $ \msg -> do when (msg == messageToPayload (createMessage ())) $ do writeIORef healthy False- testBreakConnection (NT.address e) remoteAddr isHealthy <- readIORef healthy if isHealthy then s msg else return $ Left $ NT.TransportError NT.SendFailed ""@@ -344,10 +330,8 @@ node <- newLocalNode testTransport rtable nid <- readMVar serverNodeAddr runProcess node $ do- (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)- if a == 120- then liftIO $ putMVar clientDone ()- else error "something went horribly wrong"+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)+ liftIO $ putMVar clientDone () takeMVar clientDone @@ -364,10 +348,8 @@ node <- newLocalNode testTransport rtable nid <- readMVar serverNodeAddr runProcess node $ do- (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)- if a == 120- then liftIO $ putMVar clientDone ()- else error "Something went horribly wrong"+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)+ liftIO $ putMVar clientDone () takeMVar clientDone @@ -378,11 +360,9 @@ runProcess node $ do us <- getSelfPid join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us- a :: Int <- expect- b :: Int <- expect- if a == 120 && b == 720- then liftIO $ putMVar done ()- else error "Something went horribly wrong"+ 120 :: Int <- expect+ 720 :: Int <- expect+ liftIO $ putMVar done () takeMVar done -- Test 'spawnSupervised'@@ -411,33 +391,25 @@ throw supervisorDeath forkProcess node2 $ do- 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"+ [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 () takeMVar thirdProcessDone where@@ -452,10 +424,8 @@ (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- res <- return $ ref' == ref && pid == pid'- if res == True- then liftIO $ putMVar done ()- else error "Something went horribly wrong"+ True <- return $ ref' == ref && pid == pid'+ liftIO $ putMVar done () takeMVar done testClosureExpect :: TestTransport -> RemoteTable -> Assertion@@ -467,10 +437,8 @@ us <- getSelfPid them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us send them (1234 :: Int)- (res :: Int) <- expect- if res == 1234- then liftIO $ putMVar done ()- else error "Something went horribly wrong"+ (1234 :: Int) <- expect+ liftIO $ putMVar done () takeMVar done testSpawnChannel :: TestTransport -> RemoteTable -> Assertion@@ -495,10 +463,8 @@ done <- newEmptyMVar [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable forkProcess node1 $ do- res <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))- if res == True- then liftIO $ putMVar done ()- else error "Something went horribly wrong..."+ True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))+ liftIO $ putMVar done () takeMVar done testFib :: TestTransport -> RemoteTable -> Assertion@@ -509,11 +475,8 @@ forkProcess (head nodes) $ do (sport, rport) <- newChan spawnLocal $ dfib (map localNodeId nodes, sport, 10)- ff <- receiveChan rport :: Process Integer+ 55 <- receiveChan rport :: Process Integer liftIO $ putMVar done ()- if ff /= 55- then die $ "Something went horribly wrong"- else return () takeMVar done @@ -521,7 +484,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) @@ -538,12 +501,9 @@ liftIO $ threadDelay 100000 count <- liftIO $ takeMVar iv- res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'+ True <- 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,9 +20,11 @@ -- ping ! , Ping(Ping) , ping- , pause+ , shouldBe+ , shouldMatch , shouldContain , shouldNotContain+ , expectThat , synchronisedAssertion -- test process utilities , TestProcessControl@@ -45,6 +47,9 @@ , stash ) where +#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif import Control.Concurrent ( ThreadId , myThreadId@@ -68,14 +73,16 @@ import Control.Concurrent.MVar ( putMVar )-import Control.Distributed.Process hiding (finally, catch)+import Control.Distributed.Process import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable() import Control.Exception (AsyncException(ThreadKilled), SomeException)-import Control.Monad (forever, void)-import Control.Monad.Catch (finally, catch)+import Control.Monad (forever)+import Control.Monad.Catch as Ex (catch, finally) import Control.Monad.STM (atomically)+import Control.Rematch hiding (match)+import Control.Rematch.Run import Data.Binary import Data.Typeable (Typeable) @@ -83,7 +90,6 @@ import Test.HUnit.Base (assertBool) import GHC.Generics-import System.Timeout (timeout) -- | A mutable cell containing a test result. type TestResult a = MVar a@@ -102,18 +108,10 @@ 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@@ -125,8 +123,8 @@ result <- newEmptyMVar _ <- forkProcess localNode $ do acquire lock- finally (testProc result)- (release lock)+ Ex.finally (testProc result)+ (release lock) assertComplete note result expected where acquire lock' = liftIO $ takeMVar lock' release lock' = liftIO $ putMVar lock' ()@@ -134,13 +132,23 @@ stash :: TestResult a -> a -> Process () stash mvar x = liftIO $ putMVar mvar x +expectThat :: a -> Matcher a -> Process ()+expectThat a matcher = case res of+ MatchSuccess -> return ()+ (MatchFailure msg) -> liftIO $ assertFailure msg+ where res = runMatch matcher a +shouldBe :: a -> Matcher a -> Process ()+shouldBe = expectThat+ shouldContain :: (Show a, Eq a) => [a] -> a -> Process ()-shouldContain xs x = liftIO $ assertBool mempty (x `elem` xs)+shouldContain xs x = expectThat xs $ hasItem (equalTo x) shouldNotContain :: (Show a, Eq a) => [a] -> a -> Process ()-shouldNotContain xs x = liftIO $ assertBool mempty (not $ x `elem` xs)+shouldNotContain xs x = expectThat xs $ isNot (hasItem (equalTo x)) +shouldMatch :: a -> Matcher a -> Process ()+shouldMatch = expectThat -- | Run the supplied @testProc@ using an @MVar@ to collect and assert -- against its result. Uses the supplied @note@ if the assertion fails.@@ -169,9 +177,10 @@ q <- liftIO $ newTQueueIO _ <- forkIO $ logger q return $ Logger tid q- where logger q' = forever $ do- msg <- atomically $ readTQueue q'- putStrLn msg+ where + logger q' = forever $ do+ msg <- atomically $ readTQueue q'+ putStrLn msg -- | Send a message to the Logger putLogMsg :: Logger -> String -> Process ()@@ -216,10 +225,10 @@ tryRunProcess :: LocalNode -> Process () -> IO () tryRunProcess node p = do tid <- liftIO myThreadId- void $ timeout (1000000 * 60 * 5 :: Int) $- runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))+ runProcess node $ Ex.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))+ forkProcess node $ Ex.catch p (\e -> liftIO $ throwTo tid (e::SomeException))+
src/Control/Distributed/Process/Tests/Mx.hs view
@@ -1,23 +1,11 @@-{-# LANGUAGE DeriveGeneric #-}-{-# LANGUAGE ParallelListComp #-}-+{-# LANGUAGE DeriveGeneric #-} module Control.Distributed.Process.Tests.Mx (tests) where import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..))-import Control.Distributed.Process hiding (bracket, finally, try)-import Control.Distributed.Process.Internal.Types- ( ProcessExitException(..)- , unsafeCreateUnencodedMessage- )++import Control.Distributed.Process 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(..)@@ -33,37 +21,28 @@ , mxNotify , mxBroadcast )-import Control.Monad (void, unless)-import Control.Monad.Catch(finally, bracket, try)+import Control.Monad (void) import Data.Binary-import Data.List (find, sort, intercalate)-import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes)+import Data.List (find, sort)+import Data.Maybe (isJust) import Data.Typeable-import GHC.Generics hiding (from)+import GHC.Generics+#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, log)+#endif import Test.Framework ( Test , testGroup ) import Test.Framework.Providers.HUnit (testCase)-import Test.HUnit (assertBool, assertEqual) data Publish = Publish deriving (Typeable, Generic, Eq) instance Binary Publish where -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 :: TestResult () -> Process () testAgentBroadcast result = do (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ()) @@ -78,7 +57,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 =<< receiveChanTimeout 10000000 resultRP+ stash result =<< receiveChan resultRP kill publisher "finished" kill consumer "finished"@@ -86,7 +65,7 @@ testAgentDualInput :: TestResult (Maybe Int) -> Process () testAgentDualInput result = do (sp, rp) <- newChan- s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [+ _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [ mxSink $ (\(i :: Int) -> do mxSetLocal . (+i) =<< mxGetLocal i' <- mxGetLocal@@ -96,8 +75,6 @@ else mxReady) ] - mRef <- monitor s- mxNotify (1 :: Int) nsend "sum-agent" (3 :: Int) mxNotify (2 :: Int)@@ -105,10 +82,6 @@ mxNotify (5 :: Int) stash result =<< receiveChanTimeout 10000000 rp- died <- receiveTimeout 10000000 [- matchIf (\(ProcessMonitorNotification r _ _) -> r == mRef) (const $ return True)- ]- liftIO $ assertEqual mempty (Just True) died testAgentPrioritisation :: TestResult [String] -> Process () testAgentPrioritisation result = do@@ -120,12 +93,12 @@ let name = "prioritising-agent" (sp, rp) <- newChan- s <- mxAgent (MxAgentId name) ["first"] [- mxSink (\(s :: String) -> do- mxUpdateLocal (s:)+ void $ mxAgent (MxAgentId name) ["first"] [+ mxSink (\(s :: String) -> do+ mxUpdateLocal ((s:)) st <- mxGetLocal case length st of- n | n == 5 -> do liftMX $ sendChan sp (sort st)+ n | n == 5 -> do liftMX $ sendChan sp st mxDeactivate "finished" _ -> mxReceive -- go to the mailbox )@@ -136,33 +109,23 @@ mxNotify "fourth" nsend name "fifth" - stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp- awaitExit s+ stash result . sort =<< receiveChan rp testAgentMailboxHandling :: TestResult (Maybe ()) -> Process () testAgentMailboxHandling result = do (sp, rp) <- newChan- agent <- mxAgent (MxAgentId "mailbox-agent") () [+ agent <- mxAgent (MxAgentId "listener-agent") () [ mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady ] - nsend "mailbox-agent" ()+ nsend "listener-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@@ -171,7 +134,7 @@ (MxSpawned _) -> mxSetLocal (ev:st) (MxProcessDied _ _) -> mxSetLocal (ev:st) _ -> return ()- act >> (liftMX $ sendChan rc ()) >> mxReady),+ act >> mxReady), (mxSink $ \(ev, sp :: SendPort Bool) -> do st <- mxGetLocal let found =@@ -191,349 +154,50 @@ mxReady) ] - -- 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+ _ <- monitor agentPid+ (sp, rp) <- newChan+ pid <- spawnLocal $ sendChan sp ()+ () <- receiveChan rp -- By waiting for a monitor notification, we have a- -- higher probably that the agent has seen the spawn and died events+ -- higher probably that the agent has seen monitor pid-- 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+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ] (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool) mxNotify (MxSpawned pid, replyTo) mxNotify (MxProcessDied pid DiedNormal, replyTo) - seenAlive <- receiveChanTimeout 10000000 reply- seenDead <- receiveChanTimeout 10000000 reply-- 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- liftIO $ assertEqual mempty (Just (label, p1)) reg1-- unregister label- unreg1 <- receiveChanTimeout delay unRegSink- liftIO $ assertEqual mempty (Just (label, p1)) unreg1-- register label p2- reg2 <- receiveChanTimeout delay regSink- liftIO $ assertEqual mempty (Just (label, p2)) reg2-- reregister label p1- unreg2 <- receiveChanTimeout delay unRegSink- liftIO $ assertEqual mempty (Just (label, p2)) unreg2 -- reg3 <- receiveChanTimeout delay regSink- liftIO $ assertEqual mempty (Just (label, p1)) reg3-- 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- liftIO $ assertEqual mempty (Just (label1, p1)) reg1 -- register label2 p1- reg2 <- receiveChanTimeout delay regSink- liftIO $ assertEqual mempty (Just (label2, p1)) reg2-- n1 <- whereis label1- liftIO $ assertEqual mempty (Just p1) n1-- n2 <- whereis label2- liftIO $ assertEqual mempty (Just p1) n2-- 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 -> liftIO $ assertBool mempty tr--- 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+ seenAlive <- receiveChan reply+ seenDead <- receiveChan reply - replyChannel p' = do- (s, r) <- newChan- send p' s- return r+ stash result $ seenAlive && seenDead tests :: TestTransport -> IO [Test] tests TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable- node2 <- newLocalNode testTransport initRemoteTable return [- 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)]- ]+ 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)+ ]]
src/Control/Distributed/Process/Tests/Receive.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE TemplateHaskell #-}+{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-} {-# OPTIONS_GHC -Wall #-} -- | XXX test doesn't work, because failure exceptions don't get propagated. The@@ -10,12 +10,16 @@ import Network.Transport (Transport) import Control.Distributed.Process+import Control.Distributed.Process.Closure import Control.Distributed.Process.Node import Control.Monad+import Text.Printf+import Data.Binary+import Data.Typeable import Test.HUnit (Assertion, (@?=))-import Test.Framework (Test)+import Test.Framework (Test, defaultMain) import Test.Framework.Providers.HUnit (testCase) -- Tests:@@ -42,7 +46,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest2 wait sync r1 _ = do+recTest2 wait sync r1 r2 = do forever $ do receiveChan wait r <- receiveWait@@ -55,7 +59,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest3 wait sync r1 _ = do+recTest3 wait sync r1 r2 = do forever $ do receiveChan wait r <- receiveWait@@ -68,7 +72,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest4 wait sync r1 _ = do+recTest4 wait sync r1 r2 = do forever $ do receiveChan wait r <- receiveWait@@ -82,11 +86,11 @@ master = do (waits,waitr) <- newChan (syncs,syncr) <- newChan- let go expected = do+ let go expect = do sendChan waits () r <- receiveChan syncr- liftIO $ print (r, expected, r == expected)- liftIO $ r @?= expected+ liftIO $ print (r,expect, r == expect)+ liftIO $ r @?= expect liftIO $ putStrLn "---- Test 1 ----" (s1,r1) <- newChan@@ -101,44 +105,44 @@ kill p "BANG" liftIO $ putStrLn "\n---- Test 2 ----"- (s1',r1') <- newChan- (_ ,r2') <- newChan- p' <- spawnLocal (recTest2 waitr syncs r1' r2')+ (s1,r1) <- newChan+ (s2,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- (_ ,r2'') <- newChan- p'' <- spawnLocal (recTest3 waitr syncs r1'' r2'')+ (s1,r1) <- newChan+ (s2,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- (_ ,r2''') <- newChan- p''' <- spawnLocal (recTest4 waitr syncs r1''' r2''')+ (s1,r1) <- newChan+ (s2,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
@@ -16,6 +16,10 @@ import Data.Binary () import Data.Typeable () +#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif+ import Test.Framework ( Test , testGroup@@ -25,7 +29,7 @@ testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process () testLocalDeadProcessInfo result = do- pid <- spawnLocal $ do (_ :: String) <- expect; return ()+ pid <- spawnLocal $ do "finish" <- expect; return () mref <- monitor pid send pid "finish" _ <- receiveWait [@@ -48,10 +52,8 @@ link self mRef <- monitor self stash mon mRef- res <- expect- case res of- "die" -> return ()- _ -> die $ "unexpected message received: " ++ res+ "die" <- expect+ return () monRef <- liftIO $ takeMVar mon @@ -82,9 +84,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 $ pong == "pong" && pInfo /= Nothing+ stash result $ pInfo /= Nothing where launchRemote :: MVar ProcessId -> IO () launchRemote locMV = do@@ -95,10 +97,8 @@ _ <- receiveWait [ match (\(pid, "ping") -> send pid "pong") ]- res <- expect- case res of- "stop" -> return ()- _ -> die $ "unexpected message received: " ++ res+ "stop" <- expect+ return () return () withActiveRemote :: LocalNode
src/Control/Distributed/Process/Tests/Tracing.hs view
@@ -5,7 +5,6 @@ import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..)) -import Control.Applicative ((<*)) import Control.Concurrent (threadDelay) import Control.Concurrent.MVar ( MVar@@ -20,21 +19,17 @@ import Control.Distributed.Process.Management ( MxEvent(..) )-import qualified Control.Exception as IO (bracket)-import Data.List (isPrefixOf, isSuffixOf) -import Prelude hiding ((<*))+#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, log)+#endif import Test.Framework ( Test , testGroup ) import Test.Framework.Providers.HUnit (testCase)-import System.Environment (getEnvironment)--- These are available in System.Environment only since base 4.7-import System.SetEnv (setEnv, unsetEnv) - testSpawnTracing :: TestResult Bool -> Process () testSpawnTracing result = do setTraceFlags defaultTraceFlags {@@ -121,11 +116,9 @@ (\ev -> case ev of (MxSent to from msg) -> do- 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..."+ (Just s) <- unwrapMessage msg :: Process (Maybe String)+ stash res (to == pid && from == self && s == "hello there")+ stash res (to == pid && from == self) _ -> return ()) $ do send pid "hello there"@@ -244,8 +237,12 @@ -- 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@@ -285,96 +282,6 @@ -- 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-withEnv var val =- IO.bracket (fmap (lookup var) getEnvironment <* setEnv var val)- (maybe (unsetEnv var) (setEnv var))- . const---- | Tests that one and only one interesting trace message is produced when a--- given action is performed. A message is considered interesting when the given--- function return @True@.-testSystemLoggerMsg :: TestTransport- -> Process a- -> (a -> String -> Bool)- -> IO ()-testSystemLoggerMsg t action interestingMessage =- withEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" "yes" $- withEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" "pdnusrl" $ do- n <- newLocalNode (testTransport t) initRemoteTable-- runProcess n $ do- self <- getSelfPid- reregister "trace.logger" self- a <- action- let interestingMessage' (_ :: String, msg) = interestingMessage a msg- -- Wait for the trace message.- receiveWait [ matchIf interestingMessage' $ const $ return () ]- -- Only one interesting message should arrive.- 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--- received.-testSystemLoggerMxReceive :: TestTransport -> IO ()-testSystemLoggerMxReceive t = testSystemLoggerMsg t- (getSelfPid >>= flip send ())- (\_ msg -> "MxReceived" `isPrefixOf` msg- -- discard traces of internal messages- && not (":: RegisterReply" `isSuffixOf` msg)- )---- | Tests that one and only one trace message is produced when a message is--- sent.-testSystemLoggerMxSent :: TestTransport -> IO ()-testSystemLoggerMxSent t = testSystemLoggerMsg t- (getSelfPid >>= flip send ())- (const $ isPrefixOf "MxSent")---- | Tests that one and only one trace message is produced when a process dies.-testSystemLoggerMxProcessDied :: TestTransport -> IO ()-testSystemLoggerMxProcessDied t = testSystemLoggerMsg t- (spawnLocal $ return ())- (\pid -> isPrefixOf $ "MxProcessDied " ++ show pid)---- | Tests that one and only one trace message appears when a process spawns.-testSystemLoggerMxSpawned :: TestTransport -> IO ()-testSystemLoggerMxSpawned t = testSystemLoggerMsg t- (spawnLocal $ return ())- (\pid -> isPrefixOf $ "MxSpawned " ++ show pid)---- | Tests that one and only one trace message appears when a process is--- registered.-testSystemLoggerMxRegistered :: TestTransport -> IO ()-testSystemLoggerMxRegistered t = testSystemLoggerMsg t- (getSelfPid >>= register "a" >> getSelfPid)- (\self -> isPrefixOf $ "MxRegistered " ++ show self ++ " " ++ show "a")---- | Tests that one and only one trace message appears when a process is--- unregistered.-testSystemLoggerMxUnRegistered :: TestTransport -> IO ()-testSystemLoggerMxUnRegistered t = testSystemLoggerMsg t- (getSelfPid >>= register "a" >> unregister "a" >> getSelfPid)- (\self -> isPrefixOf $ "MxUnRegistered " ++ show self ++ " " ++ show "a")- tests :: TestTransport -> IO [Test] tests testtrans@TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable@@ -416,12 +323,4 @@ (synchronisedAssertion "expected blah" node1 True (testRemoteTraceRelay testtrans) lock)- , testGroup "SystemLoggerTracer"- [ testCase "MxReceive" $ testSystemLoggerMxReceive testtrans- , testCase "MxSent" $ testSystemLoggerMxSent testtrans- , testCase "MxProcessDied" $ testSystemLoggerMxProcessDied testtrans- , testCase "MxSpawned" $ testSystemLoggerMxSpawned testtrans- , testCase "MxRegistered" $ testSystemLoggerMxRegistered testtrans- , testCase "MxUnRegistered" $ testSystemLoggerMxUnRegistered testtrans- ] ] ]
tests/runTCP.hs view
@@ -5,33 +5,26 @@ import TEST_SUITE_MODULE (tests) import Network.Transport.Test (TestTransport(..))-import Network.Socket (close)+import Network.Socket (sClose) import Network.Transport.TCP ( createTransportExposeInternals , TransportInternals(socketBetween) , defaultTCPParameters- , defaultTCPAddr- , TCPParameters(..) ) import Test.Framework (defaultMainWithArgs) import Control.Concurrent (threadDelay)-import Control.Exception (IOException, try) import System.Environment (getArgs)-import System.IO main :: IO () main = do- hSetBuffering stdout LineBuffering- hSetBuffering stderr LineBuffering Right (transport, internals) <-- createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0")- defaultTCPParameters { transportConnectTimeout = Just 3000000 }+ createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters ts <- tests TestTransport { testTransport = transport , testBreakConnection = \addr1 addr2 -> do- esock <- try $ socketBetween internals addr1 addr2- either (\e -> const (return ()) (e :: IOException)) close esock+ sock <- socketBetween internals addr1 addr2+ sClose sock threadDelay 10000 } args <- getArgs