distributed-process-tests 0.5.0 → 0.5.1
raw patch · 10 files changed
+1276/−552 lines, 10 filesdep +setenvdep −ghc-primdep −rematchdep ~ansi-terminaldep ~basedep ~binarysetup-changedPVP: major bump suggested
API removals or changes: PVP suggests a major version bump
Dependencies added: setenv
Dependencies removed: ghc-prim, rematch
Dependency ranges changed: ansi-terminal, base, binary, bytestring, distributed-process, exceptions, network, network-transport-tcp, random
API changes (from Hackage documentation)
- 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 ()
+ 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 ()
Files
- Setup.hs +0/−2
- distributed-process-tests.cabal +90/−53
- src/Control/Distributed/Process/Tests/CH.hs +514/−296
- src/Control/Distributed/Process/Tests/Closure.hs +91/−51
- src/Control/Distributed/Process/Tests/Internal/Utils.hs +23/−32
- src/Control/Distributed/Process/Tests/Mx.hs +390/−54
- src/Control/Distributed/Process/Tests/Receive.hs +34/−38
- src/Control/Distributed/Process/Tests/Stats.hs +11/−11
- src/Control/Distributed/Process/Tests/Tracing.hs +112/−11
- tests/runTCP.hs +11/−4
− Setup.hs
@@ -1,2 +0,0 @@-import Distribution.Simple-main = defaultMain
distributed-process-tests.cabal view
@@ -1,21 +1,38 @@+cabal-version: 3.0 name: distributed-process-tests-version: 0.5.0+version: 0.5.1 synopsis: Tests and test support tools for distributed-process.-homepage: http://github.com/haskell-distributed/distributed-process-tests-license: BSD3+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 license-file: LICENSE Author: Duncan Coutts, Nicolas Wu, Edsko de Vries-Maintainer: watson.timothy@gmail.com, edsko@well-typed.com, duncan@well-typed.com+maintainer: The Distributed Haskell team copyright: Well-Typed LLP category: Control, Cloud Haskell build-type: Simple-cabal-version: >=1.10 +source-repository head+ Type: git+ Location: https://github.com/haskell-distributed/distributed-process+ SubDir: packages/distributed-process-tests+ 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@@ -24,126 +41,146 @@ Control.Distributed.Process.Tests.Stats Control.Distributed.Process.Tests.Tracing Control.Distributed.Process.Tests.Internal.Utils- 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,+ 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, distributed-static,- exceptions >=0.10 && <0.11,+ exceptions >= 0.10, HUnit >= 1.2 && < 1.7, network-transport >= 0.4.1.0 && < 0.6, network >= 2.5 && < 3.3, random >= 1.0 && < 1.3,- rematch >= 0.1.2.1,+ setenv >= 0.1.1.3, test-framework >= 0.6 && < 0.9, test-framework-hunit >= 0.2.0 && < 0.4, stm hs-source-dirs: src- 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+ default-language: Haskell98+ ghc-options: -fno-warn-unused-do-bind+ default-extensions: CPP,+ ExistentialQuantification,+ FlexibleInstances,+ DeriveDataTypeable,+ DeriveGeneric,+ GeneralizedNewtypeDeriving,+ RankNTypes,+ RecordWildCards,+ ScopedTypeVariables 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.4 && < 5,+ 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- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests- 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.4 && < 5,+ Build-Depends: base >= 4.14 && < 5, distributed-process-tests,- network >= 2.3 && < 3.3,+ network >= 2.5 && < 3.2, network-transport >= 0.4.1.0 && < 0.6,- network-transport-tcp >= 0.3 && < 0.5,+ network-transport-tcp >= 0.5 && < 0.9, test-framework >= 0.6 && < 0.9 else Buildable: False- default-extensions:CPP- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests- 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.4 && < 5,+ 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- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests- 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.4 && < 5,+ 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- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests- default-language: Haskell2010 -Test-Suite TestMx+Test-Suite TestMxInMemory+ 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.4 && < 5,+ 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- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests- default-language: Haskell2010 -Test-Suite TestTracing+Test-Suite TestTracingInMemory+ import: warnings Type: exitcode-stdio-1.0 Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing- Build-Depends: base >= 4.4 && < 5,+ Build-Depends: base >= 4.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- ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ default-extensions: CPP+ default-language: Haskell98+ ghc-options: -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind HS-Source-Dirs: tests- default-language: Haskell2010++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,+ 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
src/Control/Distributed/Process/Tests/CH.hs view
@@ -1,14 +1,12 @@ 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@@ -23,26 +21,30 @@ , 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.Applicative ((<$>), (<*>), pure, (<|>))+import Control.Monad.Catch (try, catch, finally, mask, onException)+import Control.Applicative ((<|>)) import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)-import Control.Distributed.Process+import Control.Distributed.Process hiding+ ( try+ , catch+ , finally+ , mask+ , onException+ ) import Control.Distributed.Process.Internal.Types- ( NodeId(nodeAddress)- , LocalNode(localEndPoint)+ ( LocalNode(localEndPoint) , ProcessExitException(..) , nullProcessId , createUnencodedMessage ) import Control.Distributed.Process.Node+import Control.Distributed.Process.Tests.Internal.Utils (pause) import Control.Distributed.Process.Serializable (Serializable)--import Test.HUnit (Assertion, assertFailure)+import Data.Maybe (isNothing, isJust)+import Test.HUnit (Assertion, assertBool, assertEqual, 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)@@ -54,17 +56,11 @@ -- 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 $ Ex.catch p (\e -> throwTo tid (e :: SomeException))+ forkIO $ catch p (\e -> throwTo tid (e :: SomeException)) -- | The ping server from the paper ping :: Process ()@@ -74,13 +70,39 @@ send partner (Ping self) ping +verifyClient :: String -> MVar Bool -> IO ()+verifyClient s b = takeMVar b >>= assertBool s++expectPing :: MVar Bool -> Process ()+expectPing mv = expect >>= liftIO . putMVar mv . checkPing+ where+ checkPing (Ping _) = True+ -- | Quick and dirty synchronous version of whereisRemoteAsync whereisRemote :: NodeId -> String -> Process (Maybe ProcessId) whereisRemote nid string = do whereisRemoteAsync nid string- WhereIsReply _ mPid <- expect- return mPid+ receiveWait [+ match (\(WhereIsReply _ mPid) -> return mPid)+ ] +verifyWhereIsRemote :: NodeId -> String -> Process ProcessId+verifyWhereIsRemote n s = whereisRemote n s >>= maybe (die "remote name not found") return++syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO ()+syncBreakConnection breakConnection nid0 nid1 = do+ m <- newEmptyMVar+ _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m+ runProcess nid0 $ do+ them <- liftIO $ takeMVar m+ pinger <- spawnLocal $ forever $ send them ()+ _ <- monitorNode (localNodeId nid1)+ liftIO $ breakConnection (nodeAddress $ localNodeId nid0)+ (nodeAddress $ localNodeId nid1)+ NodeMonitorNotification _ _ _ <- expect+ kill pinger "finished"+ return ()+ data Add = Add ProcessId Double Double deriving (Typeable) data Divide = Divide ProcessId Double Double deriving (Typeable) data DivByZero = DivByZero deriving (Typeable)@@ -145,22 +167,27 @@ -> MVar () -- Signal for successful termination -> Process () monitorTestProcess theirAddr mOrL un reason monitorSetup 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 ()- )+ 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 ())+ ]+ ) (\(ProcessLinkException pid reason') -> do- True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'+ (liftIO $ assertBool "link exception unmatched" $+ pid == theirAddr && not mOrL && not un && reason == reason') liftIO $ putMVar done () ) @@ -191,8 +218,10 @@ pid <- getSelfPid replicateM_ numPings $ do send pingServer (Pong pid)- Ping _ <- expect- return ()+ p <- expectTimeout 3000000+ case p of+ Just (Ping _) -> return ()+ Nothing -> die "Failed to receive Ping" putMVar clientDone () @@ -279,7 +308,7 @@ runProcess localNode $ do monitor theirAddr -- wait for the process to die- ProcessMonitorNotification _ _ _ <- expect+ expect :: Process ProcessMonitorNotification monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done takeMVar done@@ -353,14 +382,17 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- 3 <- expect :: Process Double+ three <- expect :: Process Double send mathServer (Divide pid 8 2)- 4 <- expect :: Process Double+ four <- expect :: Process Double send mathServer (Divide pid 8 0)- DivByZero <- expect- liftIO $ putMVar clientDone ()+ divByZ <- expect+ liftIO $ putMVar clientDone (three, four, divByZ) - takeMVar clientDone+ res <- takeMVar clientDone+ case res of+ (3, 4, DivByZero) -> return ()+ _ -> error $ "Something went horribly wrong" -- | Send first message (i.e. connect) to an already terminated process -- (without monitoring); then send another message to a second process on@@ -388,11 +420,9 @@ pid <- getSelfPid send server1 "Hi" send server2 (Pong pid)- Ping pid' <- expect- True <- return $ pid' == server2- liftIO $ putMVar clientDone ()+ expectPing clientDone - takeMVar clientDone+ verifyClient "Expected Ping from server" clientDone -- | Test (non-zero) timeout testTimeout :: TestTransport -> Assertion@@ -401,26 +431,24 @@ done <- newEmptyMVar runProcess localNode $ do- Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]- liftIO $ putMVar done ()+ res <- receiveTimeout 1000000 [match (\Add{} -> return ())]+ liftIO $ putMVar done $ res == Nothing - takeMVar done+ verifyClient "Expected receiveTimeout to timeout..." done -- | Test zero timeout testTimeout0 :: TestTransport -> Assertion 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- -- 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)]+ partner <- fix $ \loop ->+ receiveTimeout 0 [match (\(Pong partner) -> return partner)]+ >>= maybe (liftIO (threadDelay 100000) >> loop) return self <- getSelfPid send partner (Ping self) putMVar serverAddr addr@@ -434,11 +462,9 @@ -- is not interested in, and then a single message that it wants replicateM_ 10000 $ send server "Irrelevant message" send server (Pong pid)- liftIO $ putMVar messagesSent ()- Ping _ <- expect- liftIO $ putMVar clientDone ()+ expectPing clientDone - takeMVar clientDone+ verifyClient "Expected Ping from server" clientDone -- | Test typed channels testTypedChannels :: TestTransport -> Assertion@@ -461,10 +487,10 @@ runProcess localNode $ do (clientSendPort, rport) <- newChan sendChan serverSendPort (clientSendPort, 5)- False <- receiveChan rport- liftIO $ putMVar clientDone ()+ ch <- receiveChan rport+ liftIO $ putMVar clientDone $ ch == False - takeMVar clientDone+ verifyClient "Expected channel to send 'False'" clientDone -- | Test merging receive ports testMergeChannels :: TestTransport -> Assertion@@ -487,9 +513,8 @@ rs <- mapM charChannel "abc" m <- mergePorts biased rs xs <- replicateM 9 $ receiveChan m- True <- return $ xs == expected- liftIO $ putMVar done ()- takeMVar done+ liftIO $ putMVar done $ xs == expected+ verifyClient "Expected single layer merge to match expected ordering" done -- Two layers of merging testNested :: LocalNode -> Bool -> Bool -> String -> IO ()@@ -500,9 +525,8 @@ ms <- mapM (mergePorts biasedInner) rss m <- mergePorts biasedOuter ms xs <- replicateM (9 * 3) $ receiveChan m- True <- return $ xs == expected- liftIO $ putMVar done ()- takeMVar done+ liftIO $ putMVar done $ xs == expected+ verifyClient "Expected nested channels to match expeted ordering" done -- Test that if no messages are (immediately) available, the scheduler makes no difference testBlocked :: LocalNode -> Bool -> IO ()@@ -511,43 +535,45 @@ done <- newEmptyMVar forkProcess localNode $ do- [sa, sb, sc] <- liftIO $ mapM readMVar vs- mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)- [ -- a, b, c- (sa, 'a')- , (sb, 'b')- , (sc, 'c')- -- a, c, b- , (sa, 'a')- , (sc, 'c')- , (sb, 'b')- -- b, a, c- , (sb, 'b')- , (sa, 'a')- , (sc, 'c')- -- b, c, a- , (sb, 'b')- , (sc, 'c')- , (sa, 'a')- -- c, a, b- , (sc, 'c')- , (sa, 'a')- , (sb, 'b')- -- c, b, a- , (sc, 'c')- , (sb, 'b')- , (sa, 'a')- ]+ ss <- liftIO $ mapM readMVar vs+ case ss of+ [sa, sb, sc] ->+ mapM_ ((>> pause 10000) . uncurry sendChan)+ [ -- a, b, c+ (sa, 'a')+ , (sb, 'b')+ , (sc, 'c')+ -- a, c, b+ , (sa, 'a')+ , (sc, 'c')+ , (sb, 'b')+ -- b, a, c+ , (sb, 'b')+ , (sa, 'a')+ , (sc, 'c')+ -- b, c, a+ , (sb, 'b')+ , (sc, 'c')+ , (sa, 'a')+ -- c, a, b+ , (sc, 'c')+ , (sa, 'a')+ , (sb, 'b')+ -- c, b, a+ , (sc, 'c')+ , (sb, 'b')+ , (sa, 'a')+ ]+ _ -> die "Something went horribly wrong" forkProcess localNode $ do (ss, rs) <- unzip <$> replicateM 3 newChan liftIO $ mapM_ (uncurry putMVar) $ zip vs ss m <- mergePorts biased rs xs <- replicateM (6 * 3) $ receiveChan m- True <- return $ xs == "abcacbbacbcacabcba"- liftIO $ putMVar done ()+ liftIO $ putMVar done $ xs == "abcacbbacbcacabcba" - takeMVar done+ verifyClient "Expected merged ports to match expected ordering" done mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a) mergePorts True = mergePortsBiased@@ -563,19 +589,11 @@ testTerminate :: TestTransport -> Assertion testTerminate TestTransport{..} = do localNode <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar-- pid <- forkProcess localNode $ do- liftIO $ threadDelay 100000- terminate- runProcess localNode $ do- ref <- monitor pid- ProcessMonitorNotification ref' pid' (DiedException ex) <- expect- True <- return $ ref == ref' && pid == pid' && ex == show ProcessTerminationException- liftIO $ putMVar done ()-- takeMVar done+ e <- try terminate :: Process (Either ProcessTerminationException ())+ if either show show e == show ProcessTerminationException+ then return ()+ else die "Unexpected result from terminate" testMonitorNode :: TestTransport -> Assertion testMonitorNode TestTransport{..} = do@@ -586,11 +604,12 @@ runProcess node2 $ do ref <- monitorNode (localNodeId node1)- NodeMonitorNotification ref' nid DiedDisconnect <- expect- True <- return $ ref == ref' && nid == localNodeId node1- liftIO $ putMVar done ()+ receiveWait [+ match (\(NodeMonitorNotification ref' nid DiedDisconnect) ->+ return $ ref == ref' && nid == localNodeId node1)+ ] >>= liftIO . putMVar done - takeMVar done+ verifyClient "Expected NodeMonitorNotification with matching ref & nodeId" done testMonitorLiveNode :: TestTransport -> Assertion testMonitorLiveNode TestTransport{..} = do@@ -605,15 +624,16 @@ liftIO $ putMVar ready () liftIO $ takeMVar readyr send p ()- NodeMonitorNotification ref' nid _ <- expect- True <- return $ ref == ref' && nid == localNodeId node1- liftIO $ putMVar done ()+ receiveWait [+ match (\(NodeMonitorNotification ref' nid _) ->+ (return $ ref == ref' && nid == localNodeId node1))+ ] >>= liftIO . putMVar done takeMVar ready closeLocalNode node1 putMVar readyr () - takeMVar done+ verifyClient "Expected NodeMonitorNotification for LIVE node" done testMonitorChannel :: TestTransport -> Assertion testMonitorChannel TestTransport{..} = do@@ -623,18 +643,20 @@ pid <- forkProcess node1 $ do sport <- expect :: Process (SendPort ()) ref <- monitorPort sport- PortMonitorNotification ref' port' reason <- expect- -- reason might be DiedUnknownId if the receive port is GCed before the- -- monitor is established (TODO: not sure that this is reasonable)- return $ ref' == ref && port' == sendPortId sport && (reason == DiedNormal || reason == DiedUnknownId)- liftIO $ putMVar gotNotification ()+ receiveWait [+ -- reason might be DiedUnknownId if the receive port is GCed before the+ -- monitor is established (TODO: not sure that this is reasonable)+ match (\(PortMonitorNotification ref' port' reason) ->+ return $ ref' == ref && port' == sendPortId sport &&+ (reason == DiedNormal || reason == DiedUnknownId))+ ] >>= liftIO . putMVar gotNotification runProcess node2 $ do (sport, _) <- newChan :: Process (SendPort (), ReceivePort ()) send pid sport liftIO $ threadDelay 100000 - takeMVar gotNotification+ verifyClient "Expected PortMonitorNotification" gotNotification testRegistry :: TestTransport -> Assertion testRegistry TestTransport{..} = do@@ -646,20 +668,28 @@ runProcess node $ do register "ping" pingServer- Just pid <- whereis "ping"- True <- return $ pingServer == pid+ whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer) us <- getSelfPid nsend "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'- Left (ProcessRegistrationException "dead" Nothing) <- 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"+ receiveWait [+ matchIf (\(Ping pid') -> pingServer == pid') (const $ return ())+ ]+ checkRegException "dead" Nothing deadProcess+ checkRegException "ping" (Just pingServer) deadProcess+ try (unregister "dead") >>= checkReg "dead" Nothing liftIO $ putMVar done () takeMVar done + where+ checkRegException name pid dead =+ try (register name dead) >>= checkReg name pid++ checkReg _ _ res =+ case res of+ Left (ProcessRegistrationException _ _) -> return ()+ _ -> die $ "Unexpected Registration" ++ show res+ testRegistryRemoteProcess :: TestTransport -> Assertion testRegistryRemoteProcess TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable@@ -670,13 +700,13 @@ runProcess node2 $ do register "ping" pingServer- Just pid <- whereis "ping"- True <- return $ pingServer == pid+ whereis "ping" >>= liftIO . assertBool "Unexpected ping" . (== Just pingServer) us <- getSelfPid nsend "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'- liftIO $ putMVar done ()+ receiveWait [+ matchIf (\(Ping pid') -> pingServer == pid')+ (const $ liftIO $ putMVar done ())+ ] takeMVar done @@ -684,7 +714,6 @@ testRemoteRegistry TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable node2 <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar pingServer <- forkProcess node1 ping deadProcess <- forkProcess node1 (return ())@@ -697,32 +726,34 @@ "ping" == label' && pid == pingServer) (\(RegisterReply _ _ _) -> return ()) ] - Just pid <- whereisRemote nid1 "ping"- True <- return $ pingServer == pid+ pid <- verifyWhereIsRemote nid1 "ping"+ liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid+ us <- getSelfPid nsendRemote nid1 "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'+ receiveWait [+ match (\(Ping pid') -> return $ pingServer == pid')+ ] >>= liftIO . assertBool "Expected Ping with ping server's ProcessId" -- test that if process was not registered Nothing is returned -- in owner field. registerRemoteAsync nid1 "dead" deadProcess- receiveWait [- matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')- (\(RegisterReply _ _ _) -> return ()) ]+ receiveWait [ matchIf (\(RegisterReply label' _ _) -> "dead" == label')+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))+ ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"+ registerRemoteAsync nid1 "ping" deadProcess receiveWait [- matchIf (\(RegisterReply label' False (Just pid)) ->- "ping" == label' && pid == pingServer)- (\(RegisterReply _ _ _) -> return ()) ]+ matchIf (\(RegisterReply label' False mPid) ->+ "ping" == label' && isJust mPid)+ (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))+ ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"+ unregisterRemoteAsync nid1 "dead" receiveWait [- matchIf (\(RegisterReply label' False Nothing) ->- "dead" == label' && pid == pingServer)- (\(RegisterReply _ _ _) -> return ()) ]- liftIO $ putMVar done ()-- takeMVar done+ matchIf (\(RegisterReply label' _ _) -> "dead" == label')+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))+ ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply" testRemoteRegistryRemoteProcess :: TestTransport -> Assertion testRemoteRegistryRemoteProcess TestTransport{..} = do@@ -738,15 +769,15 @@ receiveWait [ matchIf (\(RegisterReply label' _ _) -> "ping" == label') (\(RegisterReply _ _ _) -> return ()) ]- Just pid <- whereisRemote nid1 "ping"- True <- return $ pingServer == pid+ pid <- verifyWhereIsRemote nid1 "ping"+ liftIO $ assertBool "Expected pingServer to match remote name" $ pingServer == pid us <- getSelfPid nsendRemote nid1 "ping" (Pong us)- Ping pid' <- expect- True <- return $ pingServer == pid'- liftIO $ putMVar done ()+ receiveWait [+ match (\(Ping pid') -> return $ pingServer == pid')+ ] >>= liftIO . putMVar done - takeMVar done+ verifyClient "Expected Ping with ping server's ProcessId" done testSpawnLocal :: TestTransport -> Assertion testSpawnLocal TestTransport{..} = do@@ -760,15 +791,14 @@ sport <- expect sendChan sport (1234 :: Int) - sport <- spawnChannelLocal $ \rport -> do- (1234 :: Int) <- receiveChan rport- send us ()+ sport <- spawnChannelLocal $+ \rport -> (receiveChan rport :: Process Int) >>= send us send pid sport- () <- expect- liftIO $ putMVar done ()+ expect >>= liftIO . putMVar done - takeMVar done+ res <- takeMVar done+ assertBool "Expected 1234 :: Int" $ res == (1234 :: Int) testSpawnAsyncStrictness :: TestTransport -> Assertion testSpawnAsyncStrictness TestTransport{..} = do@@ -778,7 +808,7 @@ runProcess node $ do here <-getSelfNode - ev <- Ex.try $ spawnAsync here (error "boom")+ ev <- try $ spawnAsync here (error "boom") liftIO $ case ev of Right _ -> putMVar done (error "Exception didn't fire") Left (_::SomeException) -> putMVar done (return ())@@ -797,8 +827,9 @@ liftIO $ putMVar processA us msg1 <- expect msg2 <- expect- True <- return $ msg1 == "message 1" && msg2 == "message 3"- liftIO $ putMVar sendTestOk ()+ liftIO $ do+ assertBool "messages did not match" $ msg1 == "message 1" && msg2 == "message 3"+ putMVar sendTestOk () forkProcess node2 $ do {-@@ -831,7 +862,7 @@ matchIf (\(RegisterReply label' _ _) -> "a" == label') (\(RegisterReply _ _ _) -> return ()) ] - Just _ <- whereisRemote nid1 "a"+ _ <- whereisRemote nid1 "a" -- Simulate network failure@@ -850,9 +881,10 @@ (\(RegisterReply _ _ _) -> return ()) ] -- Check- Nothing <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected- Just _ <- whereisRemote nid1 "b" -- this will suceed because the value is set after thereconnect- Just _ <- whereisRemote nid1 "c"+ mPid <- whereisRemote nid1 "a" -- this will fail because the name is removed when the node is disconnected+ liftIO $ assertBool "Expected remote name to be lost" $ mPid == Nothing+ verifyWhereIsRemote nid1 "b" -- this will suceed because the value is set after thereconnect+ verifyWhereIsRemote nid1 "c" liftIO $ putMVar registerTestOk () @@ -869,7 +901,7 @@ processA <- newEmptyMVar usendTestOk <- newEmptyMVar - forkProcess node1 $ flip Ex.catch (\e -> liftIO $ print (e :: SomeException) ) $ do+ forkProcess node1 $ flip catch (\e -> liftIO $ print (e :: SomeException) ) $ do us <- getSelfPid liftIO $ putMVar processA us them <- expect@@ -893,19 +925,19 @@ isSorted _ = True -- The list can't be null since there are no failures after sending -- the last message.- True <- return $ isSorted msgs && not (null msgs)- liftIO $ putMVar usendTestOk ()+ liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs) forkProcess node2 $ do them <- liftIO $ readMVar processA getSelfPid >>= send them- () <- expect+ expect :: Process () forM_ [1..numMessages] $ \i -> do liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) usendPrim them i liftIO (threadDelay 30000) - takeMVar usendTestOk+ res <- takeMVar usendTestOk+ assertBool "Unexpected failure after sending last msg" res -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between@@ -931,14 +963,17 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- 3 <- expect :: Process Double+ three <- expect :: Process Double send mathServer (Divide pid 8 2)- 4 <- expect :: Process Double+ four <- expect :: Process Double send mathServer (Divide pid 8 0)- DivByZero <- expect- liftIO $ putMVar clientDone ()+ divByZ <- expect+ liftIO $ putMVar clientDone (three, four, divByZ) - takeMVar clientDone+ res <- takeMVar clientDone+ case res of+ (3, 4, DivByZero) -> return ()+ _ -> error "Unexpected result" -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between, however we block 'Divide' requests ....@@ -965,12 +1000,13 @@ runProcess localNode $ do pid <- getSelfPid send mathServer (Add pid 1 2)- 3 <- expect :: Process Double+ three <- expect :: Process Double send mathServer (Divide pid 8 2)- Nothing <- (expectTimeout 100000) :: Process (Maybe Double)- liftIO $ putMVar clientDone ()+ res <- (expectTimeout 100000) :: Process (Maybe Double)+ liftIO $ putMVar clientDone $ three == 3 && res == Nothing - takeMVar clientDone+ verifyClient "Expected Nothing (i.e. timeout)" clientDone+ where maybeForward :: ProcessId -> Message -> Process (Maybe ()) maybeForward s msg = handleMessage msg (\m@(Add _ _ _) -> send s m)@@ -998,8 +1034,10 @@ Just _ -> die "NONSENSE") ] -- we *must* have removed the message from our mailbox though!!!- Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]- liftIO $ putMVar serverDone ()+ res <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]+ liftIO $ do+ assertBool "Expected timeout!" $ res == Nothing+ putMVar serverDone () putMVar addr server -- Client@@ -1042,14 +1080,21 @@ runProcess localNode $ do pid <- getSelfPid send server (pid, "foo")- "foo" <- expect- send server (pid, "baz")- "baz" <- expect+ foo <- expect+ -- provoking what would be the wrong ordering is informative here...+ send server (pid, "bar")- Nothing <- (expectTimeout 100000) :: Process (Maybe Double)- liftIO $ putMVar clientDone ()+ bar <- (expectTimeout 100000) :: Process (Maybe String) -- was Double o_O !? - takeMVar clientDone+ send server (pid, "baz")+ baz <- expect++ liftIO $ putMVar clientDone (foo, bar, baz)++ res <- takeMVar clientDone+ let res' = res == ("foo", Nothing, "baz")+ assertBool "Expected timeout due to type mismatch" res'+ where tryHandleMessage :: Message -> Process (Maybe ()) tryHandleMessage msg = handleMessage msg (\(pid :: ProcessId, (m :: String))@@ -1082,17 +1127,19 @@ runProcess localNode $ do pid <- getSelfPid send server (pid, wrapMessage ("foo" :: String))- "foo" <- expect+ foo <- expect send server (pid, wrapMessage ("baz" :: String))- "baz" <- expect- liftIO $ putMVar clientDone ()+ baz <- expect+ liftIO $ putMVar clientDone (foo, baz) - takeMVar clientDone+ res <- takeMVar clientDone+ assertBool "Unexpected unwrapped results" $ res == ("foo", "baz") -- Test 'receiveChanTimeout' testReceiveChanTimeout :: TestTransport -> Assertion testReceiveChanTimeout TestTransport{..} = do done <- newEmptyMVar+ mvSender <- newEmptyMVar sendPort <- newEmptyMVar forkTry $ do@@ -1102,18 +1149,32 @@ (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 1 second- Nothing <- receiveChanTimeout 1000000 rp+ -- 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") - -- 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+ -- 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 zero timeout: non-blocking check. No message is+ -- available, we get Nothing+ receiveChanTimeout 0 rp >>= maybe (return ()) (const $ die "Expected Timeout")++ -- Let the sender know that it can send a message.+ liftIO $ putMVar mvSender ()+ -- Again, but now there is a message available- liftIO $ threadDelay 1000000- Just False <- receiveChanTimeout 0 rp+ fix $ \loop -> do+ liftIO $ threadDelay 100000+ mb <- receiveChanTimeout 0 rp+ case mb of+ Just b -> do liftIO $ assertBool "Unexpected Message" $ not b+ _ -> loop liftIO $ putMVar done () @@ -1122,10 +1183,11 @@ runProcess localNode $ do sp <- liftIO $ readMVar sendPort - liftIO $ threadDelay 1500000+ liftIO $ takeMVar mvSender+ liftIO $ threadDelay 100000 sendChan sp True - liftIO $ threadDelay 500000+ liftIO $ takeMVar mvSender sendChan sp False takeMVar done@@ -1148,8 +1210,8 @@ rp1 <- mergePortsBiased [even <$> rpInt, rpBool] - True <- receiveChan rp1- False <- receiveChan rp1+ receiveChan rp1 >>= liftIO . assertBool "Expected True"+ receiveChan rp1 >>= liftIO . assertBool "Expected False" . not -- Test Applicative instance @@ -1158,7 +1220,7 @@ let rp2 = pure (+) <*> rpInt <*> rpInt - 7 <- receiveChan rp2+ receiveChan rp2 >>= liftIO . assertBool "Expected 7" . (== 7) -- Test Alternative instance @@ -1167,8 +1229,8 @@ let rp3 = (even <$> rpInt) <|> rpBool - False <- receiveChan rp3- True <- receiveChan rp3+ receiveChan rp3 >>= liftIO . assertBool "Expected False" . not+ receiveChan rp3 >>= liftIO . assertBool "Expected True" -- Test Monad instance @@ -1182,17 +1244,86 @@ then rpInt else return 7 - 5 <- receiveChan rp4- 7 <- receiveChan rp4+ receiveChan rp4 >>= liftIO . assertBool "Expected 5" . (== 5)+ receiveChan rp4 >>= liftIO . assertBool "Expected 7" . (== 7) liftIO $ putMVar done () takeMVar done +testChanLifecycle :: TestTransport -> Assertion+testChanLifecycle TestTransport{..} = let delay = 3000000 in do+ result <- newEmptyMVar+ tchMV <- newEmptyMVar+ localNode <- newLocalNode testTransport initRemoteTable+ runProcess localNode $ do++ pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())+ liftIO $ putMVar tchMV tCh+ expect :: Process ()+ let (sp, _) = tCh+ sendChan sp ()+ expect :: Process ()++ mRefPid <- monitor pid++ cPid <- spawnLocal $ do+ (sp', rp) <- liftIO $ takeMVar tchMV+ -- say "picked up our test channel"+ send pid ()+ -- say "told pid to continue"+ res <- receiveChanTimeout delay rp+ case res of+ Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)+ Just () -> do _ <- monitor pid+ pause 10000+ -- say "sending pid a second () will cause it to exit"+ send pid ()++ -- say "make sure we see a DOWN notification for pid having stopped"+ receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]++ -- now that pid has died, the send port should be useless...+ liftIO $ putMVar tchMV (sp', rp)++ -- let's verify that we do not see the message from our+ -- parent process on the channel, once pid has died...+ recv <- receiveChanTimeout delay rp+ -- say $ "finished waiting for second (), writing result" ++ (show recv)+ liftIO $ putMVar result $ isNothing recv++ mRefCPid <- monitor cPid++ receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)+ (\_ -> return ())+ ]++ -- say "seen first pid die..."++ (sendPort, _) <- liftIO $ takeMVar tchMV+ sendChan sendPort ()+ -- say "sent () after owning pid died"++ -- let cPid know we've written to the channel...+ send cPid ()++ receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)+ (\_ -> return ())+ ]++ -- say "seen both pids die now..."++ -- and wait on the result back in IO land...+ testRes <- takeMVar result+ -- runProcess localNode $ say "got result..."+ assertBool "Expected sending on the channel to fail, but received data!" testRes++ testKillLocal :: TestTransport -> Assertion testKillLocal TestTransport{..} = do localNode <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar pid <- forkProcess localNode $ do liftIO $ threadDelay 1000000@@ -1201,17 +1332,18 @@ ref <- monitor pid us <- getSelfPid kill pid "TestKill"- ProcessMonitorNotification ref' pid' (DiedException ex) <- expect- True <- return $ ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill"- liftIO $ putMVar done ()-- takeMVar done+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' (DiedException ex) ->+ case ref == ref' && pid == pid' && ex == "killed-by=" ++ show us ++ ",reason=TestKill" of+ True -> return ()+ False -> die "Invalid ProcessMonitorNotification received"+ _ -> die "Something went horribly wrong" testKillRemote :: TestTransport -> Assertion testKillRemote TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable node2 <- newLocalNode testTransport initRemoteTable- done <- newEmptyMVar pid <- forkProcess node1 $ do liftIO $ threadDelay 1000000@@ -1220,11 +1352,17 @@ ref <- monitor pid us <- getSelfPid kill pid "TestKill"- ProcessMonitorNotification ref' pid' (DiedException reason) <- expect- True <- return $ ref == ref' && pid == pid' && reason == "killed-by=" ++ show us ++ ",reason=TestKill"- liftIO $ putMVar done ()-- takeMVar done+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' (DiedException reason) ->+ case (ref == ref', pid == pid', reason == "killed-by=" ++ show us ++ ",reason=TestKill") of+ (True, True, True) -> return ()+ (a, b, c) -> do+ let a' = if a then "" else "Invalid ref"+ let b' = if b then "" else "Invalid pid"+ let c' = if c then "" else "Invalid message"+ die $ unwords [a', b', c']+ _ -> die "Received unexpected message" testCatchesExit :: TestTransport -> Assertion testCatchesExit TestTransport{..} = do@@ -1250,13 +1388,16 @@ self <- getSelfPid send self (5 :: Integer, 10 :: Integer) msg <- receiveWait [ matchMessage return ]- Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")+ handleMessageIf msg+ (\() -> True)+ (\() -> die "whoops") >>= maybe (return ())+ (const $ die "Expected Mismatch") handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10) (\input -> liftIO $ putMVar done input) return () result <- takeMVar done- expectThat result $ equalTo (5, 10)+ assertEqual mempty (5, 10) result testCatches :: TestTransport -> Assertion testCatches TestTransport{..} = do@@ -1278,13 +1419,13 @@ parentPid <- newEmptyMVar :: IO (MVar ProcessId) spawnedPid <- newEmptyMVar :: IO (MVar ProcessId) - void $ runProcess localNode $ Ex.mask $ \unmask -> do+ void $ runProcess localNode $ mask $ \unmask -> do getSelfPid >>= liftIO . putMVar parentPid void $ spawnLocal $ unmask (getSelfPid >>= liftIO . putMVar spawnedPid) parent <- liftIO $ takeMVar parentPid child <- liftIO $ takeMVar spawnedPid- expectThat parent $ isNot $ equalTo child+ assertBool mempty (not $ parent == child) testDie :: TestTransport -> Assertion testDie TestTransport{..} = do@@ -1295,8 +1436,11 @@ (die ("foobar", 123 :: Int)) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- True <- return $ reason == ("foobar", 123 :: Int)+ let res = reason == ("foobar", 123 :: Int) liftIO $ putMVar done ()+ if res+ then return ()+ else die "Something went horribly wrong" takeMVar done @@ -1307,11 +1451,14 @@ _ <- forkProcess localNode $ do (die "timeout")- `Ex.catch` \ex@(ProcessExitException from _) ->+ `catch` \ex@(ProcessExitException from _) -> let expected = "exit-from=" ++ (show from) in do- True <- return $ (show ex) == expected+ let res = (show ex) == expected liftIO $ putMVar done ()+ if res+ then return ()+ else die "Something went horribly wrong" takeMVar done @@ -1327,17 +1474,26 @@ pid <- forkProcess localNode $ do (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- True <- return $ reason == "TestExit"+ res <- return $ reason == "TestExit" liftIO $ putMVar supervisedDone ()+ if res+ then return ()+ else die "Something went horribly wrong" runProcess localNode $ do liftIO $ takeMVar handlerSetUp ref <- monitor pid exit pid "TestExit" -- This time the client catches the exception, so it dies normally- ProcessMonitorNotification ref' pid' DiedNormal <- expect- True <- return $ ref == ref' && pid == pid'- liftIO $ putMVar supervisorDone ()+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' DiedNormal -> do+ let res = ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone ()+ if res+ then return ()+ else die "Something went horribly wrong"+ _ -> die "Something went horribly wrong" takeMVar supervisedDone takeMVar supervisorDone@@ -1353,19 +1509,98 @@ (receiveWait [] :: Process ()) -- block forever `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value- True <- return $ reason == "TestExit"+ res <- return $ reason == "TestExit" liftIO $ putMVar supervisedDone ()+ if res+ then return ()+ else die "Something went horribly wrong" runProcess node2 $ do ref <- monitor pid exit pid "TestExit"- ProcessMonitorNotification ref' pid' DiedNormal <- expect- True <- return $ ref == ref' && pid == pid'- liftIO $ putMVar supervisorDone ()+ mn <- expect+ case mn of+ ProcessMonitorNotification ref' pid' DiedNormal -> do+ res' <- return $ ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone ()+ if res'+ then return ()+ else die "Something went horribly wrong"+ _ -> die "Something went horribly wrong" takeMVar supervisedDone takeMVar supervisorDone +testRegistryMonitoring :: TestTransport -> Assertion+testRegistryMonitoring TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable++ let nid = localNodeId node2+ pid <- forkProcess node1 $ do+ self <- getSelfPid+ runUntilRegistered nid self+ say $ (show self) ++ " registered as " ++ regName+ expect :: Process ()+ say $ (show self) ++ " exiting normally"++ runProcess node2 $ do+ register regName pid+ say $ regName ++ " registered to " ++ show pid+ res <- whereis regName+ send pid ()+ say $ " sent finish signal to " ++ show pid+ _ <- getSelfPid+ liftIO $ assertBool "expected (Just pid)" $ res == (Just pid)+++ -- This delay isn't essential!+ -- The test case passes perfectly fine without it (feel free to comment out+ -- and see), however waiting a few seconds here, makes it much more likely+ -- that in delayUntilMaybeUnregistered we will hit the match case right+ -- away, and thus not be subjected to a 20 second delay. The value of 4+ -- seconds appears to work optimally on osx and across several linux distros+ -- running in virtual machines (which is essentially what we do in CI)+ void $ receiveTimeout 4000000 [ matchAny return ]++ -- This delay doesn't serve much purpose in the happy path, however if some+ -- future patch breaks the cooperative behaviour of node controllers viz+ -- remote process registration and notification taking place via ncEffectDied,+ -- there would be the possibility of a race in the test case should we attempt+ -- to evaluate `whereis regName` on node2 right away. In case the name is still+ -- erroneously registered, observing the 20 second delay (or lack of), could at+ -- least give a hint that something is wrong, and we give up our time slice+ -- so that there's a higher change the registrations have been cleaned up+ -- in either case.+ runProcess node2 $ delayUntilMaybeUnregistered nid pid++ regHere <- newEmptyMVar+ runProcess node2 $ whereis regName >>= liftIO . putMVar regHere+ res <- takeMVar regHere+ case res of+ Nothing -> return ()+ _ -> assertBool ("expected Nothing, but got " ++ show pid) False++ where+ runUntilRegistered nid us = do+ whereisRemoteAsync nid regName+ receiveWait [+ matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)+ (const $ return ())+ ]++ delayUntilMaybeUnregistered nid p = do+ whereisRemoteAsync nid regName+ res <- receiveTimeout 20000000 {- 20 sec delay -} [+ matchIf (\(WhereIsReply n p') -> n == regName && isNothing p')+ (const $ return ())+ ]+ case res of+ Just () -> return ()+ Nothing -> delayUntilMaybeUnregistered nid p++ regName = "testRegisterRemote"+ testUnsafeSend :: TestTransport -> Assertion testUnsafeSend TestTransport{..} = do serverAddr <- newEmptyMVar@@ -1381,8 +1616,7 @@ void $ forkProcess localNode $ do serverPid <- liftIO $ takeMVar serverAddr getSelfPid >>= unsafeSend serverPid- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone takeMVar clientDone @@ -1401,8 +1635,7 @@ void $ forkProcess localNode $ do serverPid <- liftIO $ takeMVar serverAddr getSelfPid >>= unsafeUSend serverPid- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone takeMVar clientDone @@ -1413,8 +1646,7 @@ localNode <- newLocalNode testTransport initRemoteTable pid <- forkProcess localNode $ do- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone void $ runProcess localNode $ do register "foobar" pid@@ -1432,8 +1664,7 @@ _ <- forkProcess localNode1 $ do getSelfPid >>= register "foobar" liftIO $ putMVar clientDone ()- () <- expect- liftIO $ putMVar clientDone ()+ expect >>= liftIO . putMVar clientDone takeMVar clientDone void $ runProcess localNode2 $ do@@ -1457,7 +1688,7 @@ serverPid <- liftIO $ takeMVar serverAddr (sp, rp) <- newChan unsafeSend serverPid sp- () <- receiveChan rp+ receiveChan rp :: Process () liftIO $ putMVar clientDone () takeMVar clientDone@@ -1471,9 +1702,9 @@ runProcess node $ do r <- (/=) <$> getSelfPid <*> callLocal getSelfPid liftIO $ putMVar result r- True <- takeMVar result- return () + takeMVar result >>= assertBool "Expected 'True'"+ -- Testing that when callLocal is interrupted, the worker is interrupted. ibox <- newIORef False runProcess node $ do@@ -1481,29 +1712,28 @@ spawnLocal $ do caller <- getSelfPid send keeper caller- Ex.onException+ onException (callLocal $ do- Ex.onException (do send keeper caller- expect)- (do liftIO $ writeIORef ibox True))+ onException (do send keeper caller+ expect)+ (do liftIO $ writeIORef ibox True)) (send keeper ()) caller <- expect exit caller "test"- () <- expect- return ()- True <- readIORef ibox- return ()+ expect :: Process () + readIORef ibox >>= assertBool "Expected 'True'"+ -- Testing that when the worker raises an exception, the exception is propagated to the parent. ibox2 <- newIORef False runProcess node $ do- r <- Ex.try (callLocal $ error "e" >> return ())+ r <- try (callLocal $ error "e" >> return ()) liftIO $ writeIORef ibox2 $ case r of Left (ErrorCall "e") -> True _ -> False- True <- readIORef ibox- return () + readIORef ibox >>= assertBool "Expected 'True'"+ -- Test that caller waits for the worker in correct situation ibox3 <- newIORef False result3 <- newEmptyMVar@@ -1515,13 +1745,13 @@ send keeper us () <- expect liftIO yield)- `Ex.finally` (liftIO $ writeIORef ibox3 True)+ `finally` (liftIO $ writeIORef ibox3 True) liftIO $ putMVar result3 =<< readIORef ibox3 worker <- expect send worker ()- True <- takeMVar result3- return () + takeMVar result3 >>= assertBool "Expected 'True'"+ -- Test that caller waits for the worker in case when caller gets an exception ibox4 <- newIORef False result4 <- newEmptyMVar@@ -1532,17 +1762,13 @@ callLocal ((do send keeper caller expect)- `Ex.finally` (liftIO $ writeIORef ibox4 True))- `Ex.finally` (liftIO $ putMVar result4 =<< readIORef ibox4)+ `finally` (liftIO $ writeIORef ibox4 True))+ `finally` (liftIO $ putMVar result4 =<< readIORef ibox4) caller <- expect exit caller "hi!"- True <- takeMVar result4- return ()- -- XXX: Testing that when mask_ $ callLocal p runs p in masked state. ---+ takeMVar result4 >>= assertBool "Expected 'True'"+ -- XXX: Testing that when mask_ $ callLocal p runs p in masked state. tests :: TestTransport -> IO [Test] tests testtrans = return [@@ -1553,9 +1779,7 @@ , 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)@@ -1569,6 +1793,7 @@ , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans) , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans) , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)+ , testCase "ChanLifecycle" (testChanLifecycle testtrans) , testCase "KillLocal" (testKillLocal testtrans) , testCase "KillRemote" (testKillRemote testtrans) , testCase "Die" (testDie testtrans)@@ -1578,6 +1803,7 @@ , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans) , testCase "ExitLocal" (testExitLocal testtrans) , testCase "ExitRemote" (testExitRemote testtrans)+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans) , testCase "TextCallLocal" (testCallLocal testtrans) -- Unsafe Primitives , testCase "TestUnsafeSend" (testUnsafeSend testtrans)@@ -1598,8 +1824,7 @@ -- 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 "MonitorUnreachable" (testMonitorUnreachable testtrans True False)- , testCase "MonitorNormalTermination" (testMonitorNormalTermination 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)@@ -1621,20 +1846,13 @@ , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans) , testCase "MonitorChannel" (testMonitorChannel testtrans) -- Reconnect- , testCase "Reconnect" (testReconnect testtrans) ]- ] -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 ()+ -- 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)+ -- ]+ ]
src/Control/Distributed/Process/Tests/Closure.hs view
@@ -1,3 +1,4 @@+{-# OPTIONS_GHC -Wno-unused-top-binds #-} {-# LANGUAGE TemplateHaskell, KindSignatures #-} module Control.Distributed.Process.Tests.Closure (tests) where @@ -19,14 +20,12 @@ , modifyMVar_ , newMVar )-import Control.Applicative ((<$>)) import System.Random (randomIO) import Control.Distributed.Process import Control.Distributed.Process.Closure import Control.Distributed.Process.Node import Control.Distributed.Process.Internal.Types- ( NodeId(nodeAddress)- , createMessage+ ( createMessage , messageToPayload ) import Control.Distributed.Static (staticLabel, staticClosure)@@ -169,8 +168,11 @@ node <- newLocalNode testTransport rtable done <- newEmptyMVar forkProcess node $ do- 120 <- join . unClosure $ factorialClosure 5+ i <- join . unClosure $ factorialClosure 5 liftIO $ putMVar done ()+ if i == 720+ then return ()+ else error "Something went horribly wrong" takeMVar done testBind :: TestTransport -> RemoteTable -> Assertion@@ -180,8 +182,11 @@ runProcess node $ do us <- getSelfPid join . unClosure $ sendFac 6 us- (720 :: Int) <- expect+ (i :: Int) <- expect liftIO $ putMVar done ()+ if i == 720+ then return ()+ else error "Something went horribly wrong" takeMVar done testSendPureClosure :: TestTransport -> RemoteTable -> Assertion@@ -194,7 +199,7 @@ addr <- forkProcess node $ do cl <- expect fn <- unClosure cl :: Process (Int -> Int)- 13 <- return $ fn 6+ (_ :: Int) <- return $ fn 6 liftIO $ putMVar serverDone () putMVar serverAddr addr @@ -218,8 +223,11 @@ liftIO $ do someMVar <- newEmptyMVar io someMVar- 5 <- readMVar someMVar+ i <- readMVar someMVar putMVar serverDone ()+ if i == 5+ then return ()+ else error "Something went horribly wrong" putMVar serverAddr addr forkIO $ do@@ -248,8 +256,10 @@ runProcess node $ do pid <- getSelfPid send theirAddr (cpSend $(mkStatic 'sdictInt) pid)- 5 <- expect :: Process Int- liftIO $ putMVar clientDone ()+ i <- expect :: Process Int+ if i == 5+ then liftIO $ putMVar clientDone ()+ else error "Something went horribly wrong" takeMVar clientDone @@ -269,8 +279,9 @@ pid <- getSelfPid pid' <- spawn nid (sendPidClosure pid) pid'' <- expect- True <- return $ pid' == pid''- liftIO $ putMVar clientDone ()+ if pid' == pid''+ then liftIO $ putMVar clientDone ()+ else error "Something went horribly wrong" takeMVar clientDone @@ -294,8 +305,9 @@ spawnLocal $ spawn (localNodeId node2) (sendPidClosure pid) >>= send pid pid' <- expect :: Process ProcessId pid'' <- expect :: Process ProcessId- True <- return $ pid' == pid''- return ()+ if pid' == pid''+ then return ()+ else error "Something went horribly wrong" where @@ -305,14 +317,16 @@ wrapEP e = e { NT.connect = \x y z -> do healthy <- newIORef True- fmap (fmap $ wrapConnection healthy) $ NT.connect e x y z+ fmap (fmap $ wrapConnection healthy e x) $ NT.connect e x y z } - wrapConnection :: IORef Bool -> NT.Connection -> NT.Connection- wrapConnection healthy (NT.Connection s closeC) =+ wrapConnection :: IORef Bool -> NT.EndPoint -> NT.EndPointAddress+ -> NT.Connection -> NT.Connection+ wrapConnection healthy e remoteAddr (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 ""@@ -330,8 +344,10 @@ node <- newLocalNode testTransport rtable nid <- readMVar serverNodeAddr runProcess node $ do- (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)- liftIO $ putMVar clientDone ()+ (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)+ if a == 120+ then liftIO $ putMVar clientDone ()+ else error "something went horribly wrong" takeMVar clientDone @@ -348,8 +364,10 @@ node <- newLocalNode testTransport rtable nid <- readMVar serverNodeAddr runProcess node $ do- (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)- liftIO $ putMVar clientDone ()+ (a :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)+ if a == 120+ then liftIO $ putMVar clientDone ()+ else error "Something went horribly wrong" takeMVar clientDone @@ -360,9 +378,11 @@ runProcess node $ do us <- getSelfPid join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us- 120 :: Int <- expect- 720 :: Int <- expect- liftIO $ putMVar done ()+ a :: Int <- expect+ b :: Int <- expect+ if a == 120 && b == 720+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong" takeMVar done -- Test 'spawnSupervised'@@ -391,26 +411,34 @@ throw supervisorDeath forkProcess node2 $ do- [super, child] <- liftIO $ mapM readMVar [superPid, childPid]- ref <- monitor child- self <- getSelfPid- let waitForMOrL = do- liftIO $ threadDelay 10000- mpinfo <- getProcessInfo child- case mpinfo of- Nothing -> waitForMOrL- Just pinfo ->- unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL- waitForMOrL- liftIO $ putMVar linkUp ()- -- because monitor message was sent before message to process- -- we hope that it will be processed before- ProcessMonitorNotification ref' pid' (DiedException e) <- expect- True <- return $ ref' == ref- && pid' == child- && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))- liftIO $ putMVar thirdProcessDone ()+ res <- liftIO $ mapM readMVar [superPid, childPid]+ case res of+ [super, child] -> do+ ref <- monitor child+ self <- getSelfPid+ let waitForMOrL = do+ liftIO $ threadDelay 10000+ mpinfo <- getProcessInfo child+ case mpinfo of+ Nothing -> waitForMOrL+ Just pinfo ->+ unless (isJust $ lookup self (infoMonitors pinfo)) waitForMOrL+ waitForMOrL+ liftIO $ putMVar linkUp ()+ -- because monitor message was sent before message to process+ -- we hope that it will be processed before+ res' <- expect+ case res' of+ (ProcessMonitorNotification ref' pid' (DiedException e)) ->+ if (ref' == ref && pid' == child &&+ e == show (ProcessLinkException super+ (DiedException (show supervisorDeath))))+ then liftIO $ putMVar thirdProcessDone ()+ else error "Something went horribly wrong"+ _ -> error "Something went horribly wrong" + _ -> die $ "Something went horribly wrong"+ takeMVar thirdProcessDone where supervisorDeath :: IOException@@ -424,8 +452,10 @@ (pid, ref) <- spawnMonitor (localNodeId node) (closure (staticLabel "ThisDoesNotExist") empty) ProcessMonitorNotification ref' pid' _reason <- expect -- Depending on the exact interleaving, reason might be NoProc or the exception thrown by the absence of the static closure- True <- return $ ref' == ref && pid == pid'- liftIO $ putMVar done ()+ res <- return $ ref' == ref && pid == pid'+ if res == True+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong" takeMVar done testClosureExpect :: TestTransport -> RemoteTable -> Assertion@@ -437,8 +467,10 @@ us <- getSelfPid them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us send them (1234 :: Int)- (1234 :: Int) <- expect- liftIO $ putMVar done ()+ (res :: Int) <- expect+ if res == 1234+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong" takeMVar done testSpawnChannel :: TestTransport -> RemoteTable -> Assertion@@ -463,8 +495,10 @@ done <- newEmptyMVar [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable forkProcess node1 $ do- True <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))- liftIO $ putMVar done ()+ res <- call $(functionTDict 'isPrime) (localNodeId node2) ($(mkClosure 'isPrime) (79 :: Integer))+ if res == True+ then liftIO $ putMVar done ()+ else error "Something went horribly wrong..." takeMVar done testFib :: TestTransport -> RemoteTable -> Assertion@@ -475,8 +509,11 @@ forkProcess (head nodes) $ do (sport, rport) <- newChan spawnLocal $ dfib (map localNodeId nodes, sport, 10)- 55 <- receiveChan rport :: Process Integer+ ff <- receiveChan rport :: Process Integer liftIO $ putMVar done ()+ if ff /= 55+ then die $ "Something went horribly wrong"+ else return () takeMVar done @@ -484,7 +521,7 @@ testSpawnReconnect testtrans@TestTransport{..} rtable = do [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable let nid1 = localNodeId node1- nid2 = localNodeId node2+ -- nid2 = localNodeId node2 done <- newEmptyMVar iv <- newMVar (0 :: Int) @@ -501,9 +538,12 @@ liftIO $ threadDelay 100000 count <- liftIO $ takeMVar iv- True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'+ res <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn' liftIO $ putMVar done ()+ if res /= True+ then error "Something went horribly wrong"+ else return () takeMVar done
src/Control/Distributed/Process/Tests/Internal/Utils.hs view
@@ -20,11 +20,9 @@ -- ping ! , Ping(Ping) , ping- , shouldBe- , shouldMatch+ , pause , shouldContain , shouldNotContain- , expectThat , synchronisedAssertion -- test process utilities , TestProcessControl@@ -47,9 +45,6 @@ , stash ) where -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif import Control.Concurrent ( ThreadId , myThreadId@@ -73,16 +68,14 @@ import Control.Concurrent.MVar ( putMVar )-import Control.Distributed.Process+import Control.Distributed.Process hiding (finally, catch) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable() import Control.Exception (AsyncException(ThreadKilled), SomeException)-import Control.Monad (forever)-import Control.Monad.Catch as Ex (catch, finally)+import Control.Monad (forever, void)+import Control.Monad.Catch (finally, catch) import Control.Monad.STM (atomically)-import Control.Rematch hiding (match)-import Control.Rematch.Run import Data.Binary import Data.Typeable (Typeable) @@ -90,6 +83,7 @@ import Test.HUnit.Base (assertBool) import GHC.Generics+import System.Timeout (timeout) -- | A mutable cell containing a test result. type TestResult a = MVar a@@ -108,10 +102,18 @@ instance Binary TestProcessControl where +data Private = Private+ deriving (Typeable, Generic)+instance Binary Private where+ -- | Does exactly what it says on the tin, doing so in the @Process@ monad. noop :: Process () noop = return () +pause :: Int -> Process ()+pause delay =+ void $ receiveTimeout delay [ match (\Private -> return ()) ]+ synchronisedAssertion :: Eq a => String -> LocalNode@@ -123,8 +125,8 @@ result <- newEmptyMVar _ <- forkProcess localNode $ do acquire lock- Ex.finally (testProc result)- (release lock)+ finally (testProc result)+ (release lock) assertComplete note result expected where acquire lock' = liftIO $ takeMVar lock' release lock' = liftIO $ putMVar lock' ()@@ -132,23 +134,13 @@ 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 = expectThat xs $ hasItem (equalTo x)+shouldContain xs x = liftIO $ assertBool mempty (x `elem` xs) shouldNotContain :: (Show a, Eq a) => [a] -> a -> Process ()-shouldNotContain xs x = expectThat xs $ isNot (hasItem (equalTo x))+shouldNotContain xs x = liftIO $ assertBool mempty (not $ x `elem` xs) -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.@@ -177,10 +169,9 @@ 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 ()@@ -225,10 +216,10 @@ tryRunProcess :: LocalNode -> Process () -> IO () tryRunProcess node p = do tid <- liftIO myThreadId- runProcess node $ Ex.catch p (\e -> liftIO $ throwTo tid (e::SomeException))+ void $ timeout (1000000 * 60 * 5 :: Int) $+ runProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException)) tryForkProcess :: LocalNode -> Process () -> IO ProcessId tryForkProcess node p = do tid <- liftIO myThreadId- forkProcess node $ Ex.catch p (\e -> liftIO $ throwTo tid (e::SomeException))-+ forkProcess node $ catch p (\e -> liftIO $ throwTo tid (e::SomeException))
src/Control/Distributed/Process/Tests/Mx.hs view
@@ -1,11 +1,23 @@-{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE DeriveGeneric #-}+{-# LANGUAGE ParallelListComp #-}+ module Control.Distributed.Process.Tests.Mx (tests) where import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..))--import Control.Distributed.Process+import Control.Distributed.Process hiding (bracket, finally, try)+import Control.Distributed.Process.Internal.Types+ ( ProcessExitException(..)+ , unsafeCreateUnencodedMessage+ ) import Control.Distributed.Process.Node+import qualified Control.Distributed.Process.UnsafePrimitives as Unsafe+ ( send+ , nsend+ , nsendRemote+ , usend+ , sendChan+ ) import Control.Distributed.Process.Management ( MxEvent(..) , MxAgentId(..)@@ -21,28 +33,37 @@ , mxNotify , mxBroadcast )-import Control.Monad (void)+import Control.Monad (void, unless)+import Control.Monad.Catch(finally, bracket, try) import Data.Binary-import Data.List (find, sort)-import Data.Maybe (isJust)+import Data.List (find, sort, intercalate)+import Data.Maybe (isJust, fromJust, isNothing, fromMaybe, catMaybes) import Data.Typeable-import GHC.Generics-#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch, log)-#endif+import GHC.Generics hiding (from) 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 -testAgentBroadcast :: TestResult () -> Process ()+awaitExit :: ProcessId -> Process ()+awaitExit pid =+ withMonitorRef pid $ \ref -> do+ receiveWait+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == ref)+ (\_ -> return ())+ ]+ where+ withMonitorRef p = bracket (monitor p) unmonitor++testAgentBroadcast :: TestResult (Maybe ()) -> Process () testAgentBroadcast result = do (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ()) @@ -57,7 +78,7 @@ mxNotify () -- Once the publisher has seen our message, it will broadcast the Publish -- and the consumer will see that and send the result to our typed channel.- stash result =<< receiveChan resultRP+ stash result =<< receiveChanTimeout 10000000 resultRP kill publisher "finished" kill consumer "finished"@@ -65,7 +86,7 @@ testAgentDualInput :: TestResult (Maybe Int) -> Process () testAgentDualInput result = do (sp, rp) <- newChan- _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [+ s <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [ mxSink $ (\(i :: Int) -> do mxSetLocal . (+i) =<< mxGetLocal i' <- mxGetLocal@@ -75,6 +96,8 @@ else mxReady) ] + mRef <- monitor s+ mxNotify (1 :: Int) nsend "sum-agent" (3 :: Int) mxNotify (2 :: Int)@@ -82,6 +105,10 @@ 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@@ -93,12 +120,12 @@ let name = "prioritising-agent" (sp, rp) <- newChan- void $ mxAgent (MxAgentId name) ["first"] [- mxSink (\(s :: String) -> do- mxUpdateLocal ((s:))+ s <- mxAgent (MxAgentId name) ["first"] [+ mxSink (\(s :: String) -> do+ mxUpdateLocal (s:) st <- mxGetLocal case length st of- n | n == 5 -> do liftMX $ sendChan sp st+ n | n == 5 -> do liftMX $ sendChan sp (sort st) mxDeactivate "finished" _ -> mxReceive -- go to the mailbox )@@ -109,23 +136,33 @@ mxNotify "fourth" nsend name "fifth" - stash result . sort =<< receiveChan rp+ stash result . fromMaybe [] =<< receiveChanTimeout 10000000 rp+ awaitExit s testAgentMailboxHandling :: TestResult (Maybe ()) -> Process () testAgentMailboxHandling result = do (sp, rp) <- newChan- agent <- mxAgent (MxAgentId "listener-agent") () [+ agent <- mxAgent (MxAgentId "mailbox-agent") () [ mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady ] - nsend "listener-agent" ()+ nsend "mailbox-agent" () stash result =<< receiveChanTimeout 1000000 rp kill agent "finished" testAgentEventHandling :: TestResult Bool -> Process () testAgentEventHandling result = do+ us <- getSelfPid+ -- because this test is a bit racy, let's ensure it can't run indefinitely+ timer <- spawnLocal $ do+ pause (1000000 * 60 * 5)+ -- okay we've waited 5 mins, let's kill the test off if it's stuck...+ stash result False+ kill us "Test Timed Out"+ let initState = [] :: [MxEvent]+ (rc, rs) <- newChan agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [ (mxSink $ \ev -> do st <- mxGetLocal@@ -134,7 +171,7 @@ (MxSpawned _) -> mxSetLocal (ev:st) (MxProcessDied _ _) -> mxSetLocal (ev:st) _ -> return ()- act >> mxReady),+ act >> (liftMX $ sendChan rc ()) >> mxReady), (mxSink $ \(ev, sp :: SendPort Bool) -> do st <- mxGetLocal let found =@@ -154,50 +191,349 @@ mxReady) ] - _ <- monitor agentPid- (sp, rp) <- newChan- pid <- spawnLocal $ sendChan sp ()- () <- receiveChan rp+ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...+ faff 2000000 rs + (sp, rp) <- newChan :: Process (SendPort (), ReceivePort ())++ pid <- spawnLocal $ expect >>= sendChan sp+ -- By waiting for a monitor notification, we have a- -- higher probably that the agent has seen+ -- higher probably that the agent has seen the spawn and died events monitor pid- receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ] + send pid ()+ rct <- receiveChanTimeout 10000000 rp+ unless (isJust rct) $ die "No response on channel"++ pmn <- receiveTimeout 2000000 [ match (\ProcessMonitorNotification{} -> return ()) ]+ unless (isJust pmn) $ die "No monitor notification arrived"++ -- TODO: yes, this is racy, but we're at the mercy of the scheduler here...+ faff 2000000 rs+ (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool) mxNotify (MxSpawned pid, replyTo) mxNotify (MxProcessDied pid DiedNormal, replyTo) - seenAlive <- receiveChan reply- seenDead <- receiveChan reply+ seenAlive <- receiveChanTimeout 10000000 reply+ seenDead <- receiveChanTimeout 10000000 reply - stash result $ seenAlive && seenDead+ stash result $ foldr (&&) True $ catMaybes [seenAlive, seenDead]+ kill timer "test-complete"+ kill agentPid "test-complete"+ where+ faff delay port = do+ res <- receiveChanTimeout delay port+ unless (isNothing res) $ pause delay +testMxRegEvents :: TestResult () -> Process ()+testMxRegEvents result = do+ {- This test only deals with the local case, to ensure that we are being+ notified in the expected order - the remote cases related to the+ behaviour of the node controller are contained in the CH test suite. -}+ ensure (stash result ()) $ do+ let label = "testMxRegEvents"+ let agentLabel = "mxRegEvents-agent"+ let delay = 1000000+ (regChan, regSink) <- newChan+ (unRegChan, unRegSink) <- newChan+ agent <- mxAgent (MxAgentId agentLabel) () [+ mxSink $ \ev -> do+ case ev of+ MxRegistered pid label'+ | label' == label -> liftMX $ sendChan regChan (label', pid)+ MxUnRegistered pid label'+ | label' == label -> liftMX $ sendChan unRegChan (label', pid)+ _ -> return ()+ mxReady+ ]++ p1 <- spawnLocal expect+ p2 <- spawnLocal expect++ register label p1+ reg1 <- receiveChanTimeout delay regSink+ 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++ replyChannel p' = do+ (s, r) <- newChan+ send p' s+ return r+ tests :: TestTransport -> IO [Test] tests TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable return [- testGroup "Mx Agents" [- testCase "Event Handling"- (delayedAssertion- "expected True, but events where not as expected"- node1 True testAgentEventHandling)- , testCase "Inter-Agent Broadcast"- (delayedAssertion- "expected (), but no broadcast was received"- node1 () testAgentBroadcast)- , testCase "Agent Mailbox Handling"- (delayedAssertion- "expected (Just ()), but no regular (mailbox) input was handled"- node1 (Just ()) testAgentMailboxHandling)- , testCase "Agent Dual Input Handling"- (delayedAssertion- "expected sum = 15, but the result was Nothing"- node1 (Just 15 :: Maybe Int) testAgentDualInput)- , testCase "Agent Input Prioritisation"- (delayedAssertion- "expected [first, second, third, fourth, fifth], but result diverged"- node1 (sort ["first", "second",- "third", "fourth",- "fifth"]) testAgentPrioritisation)- ]]+ testGroup "MxAgents" [+ testCase "EventHandling"+ (delayedAssertion+ "expected True, but events where not as expected"+ node1 True testAgentEventHandling)+ , testCase "InterAgentBroadcast"+ (delayedAssertion+ "expected (), but no broadcast was received"+ node1 (Just ()) testAgentBroadcast)+ , testCase "AgentMailboxHandling"+ (delayedAssertion+ "expected (Just ()), but no regular (mailbox) input was handled"+ node1 (Just ()) testAgentMailboxHandling)+ , testCase "AgentDualInputHandling"+ (delayedAssertion+ "expected sum = 15, but the result was Nothing"+ node1 (Just 15 :: Maybe Int) testAgentDualInput)+ , testCase "AgentInputPrioritisation"+ (delayedAssertion+ "expected [first, second, third, fourth, fifth], but result diverged"+ node1 (sort ["first", "second",+ "third", "fourth",+ "fifth"]) testAgentPrioritisation)+ ]+ , testGroup "MxEvents" [+ testCase "NameRegistrationEvents"+ (delayedAssertion+ "expected registration events to map to the correct ProcessId"+ node1 () testMxRegEvents)+ , testCase "PostDeathNameUnRegistrationEvents"+ (delayedAssertion+ "expected process deaths to result in unregistration events"+ node1 () (testMxRegMon node2))+ , testGroup "SendEvents" $ buildTestCases node1 node2+ ]+ ]+ where+ buildTestCases n1 n2 = let nid = localNodeId n2 in build n1 n2 [+ ("NSend", [+ ("nsend", testNSend nsend)+ , ("Unsafe.nsend", testNSend Unsafe.nsend)+ , ("nsendRemote", testNSend (nsendRemote nid))+ , ("Unsafe.nsendRemote", testNSend (Unsafe.nsendRemote nid))+ ])+ , ("Send", [+ ("send", testSend send)+ , ("Unsafe.send", testSend Unsafe.send)+ , ("usend", testSend usend)+ , ("Unsafe.usend", testSend Unsafe.usend)+ , ("sendChan", testChan sendChan)+ , ("Unsafe.sendChan", testChan Unsafe.sendChan)+ ])+ , ("Forward", [+ ("forward", testSend (\p m -> forward (unsafeCreateUnencodedMessage m) p))+ , ("uforward", testSend (\p m -> uforward (unsafeCreateUnencodedMessage m) p))+ ])+ ]++ build :: LocalNode+ -> LocalNode+ -> [(String, [(String, (Maybe LocalNode -> Process ()))])]+ -> [Test]+ build n ln specs =+ [ testGroup (intercalate "-" [groupName, caseSuffix]) [+ testCase (intercalate "-" [caseName, caseSuffix])+ (runProcess n (caseImpl caseNode))+ | (caseName, caseImpl) <- groupCases+ ]+ | (groupName, groupCases) <- specs+ , (caseSuffix, caseNode) <- [("RemotePid", Just ln), ("LocalPid", Nothing)]+ ]
src/Control/Distributed/Process/Tests/Receive.hs view
@@ -1,4 +1,4 @@-{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-}+{-# LANGUAGE TemplateHaskell #-} {-# OPTIONS_GHC -Wall #-} -- | XXX test doesn't work, because failure exceptions don't get propagated. The@@ -10,16 +10,12 @@ 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, defaultMain)+import Test.Framework (Test) import Test.Framework.Providers.HUnit (testCase) -- Tests:@@ -46,7 +42,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest2 wait sync r1 r2 = do+recTest2 wait sync r1 _ = do forever $ do receiveChan wait r <- receiveWait@@ -59,7 +55,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest3 wait sync r1 r2 = do+recTest3 wait sync r1 _ = do forever $ do receiveChan wait r <- receiveWait@@ -72,7 +68,7 @@ -> SendPort String -> ReceivePort String -> ReceivePort String -> Process ()-recTest4 wait sync r1 r2 = do+recTest4 wait sync r1 _ = do forever $ do receiveChan wait r <- receiveWait@@ -86,11 +82,11 @@ master = do (waits,waitr) <- newChan (syncs,syncr) <- newChan- let go expect = do+ let go expected = do sendChan waits () r <- receiveChan syncr- liftIO $ print (r,expect, r == expect)- liftIO $ r @?= expect+ liftIO $ print (r, expected, r == expected)+ liftIO $ r @?= expected liftIO $ putStrLn "---- Test 1 ----" (s1,r1) <- newChan@@ -105,44 +101,44 @@ kill p "BANG" liftIO $ putStrLn "\n---- Test 2 ----"- (s1,r1) <- newChan- (s2,r2) <- newChan- p <- spawnLocal (recTest2 waitr syncs r1 r2)+ (s1',r1') <- newChan+ (_ ,r2') <- newChan+ p' <- spawnLocal (recTest2 waitr syncs r1' r2') - sendChan s1 "a" >> go "received1 a"- send p "foo" >> go "received2 foo"- sendChan s1 "a" >> send p "foo" >> go "received1 a"- sendChan s1 "a" >> send p "bar" >> go "received1 a"+ sendChan s1' "a" >> go "received1 a"+ send p' "foo" >> go "received2 foo"+ sendChan s1' "a" >> send p "foo" >> go "received1 a"+ sendChan s1' "a" >> send p "bar" >> go "received1 a" go "received2 foo" - kill p "BANG"+ kill p' "BANG" liftIO $ putStrLn "\n---- Test 3 ----"- (s1,r1) <- newChan- (s2,r2) <- newChan- p <- spawnLocal (recTest3 waitr syncs r1 r2)+ (s1'',r1'') <- newChan+ (_ ,r2'') <- newChan+ p'' <- spawnLocal (recTest3 waitr syncs r1'' r2'') - sendChan s1 "a" >> go "received2 a"- send p "foo" >> go "received1 foo"- sendChan s1 "a" >> send p "foo" >> go "received1 foo"- sendChan s1 "a" >> send p "bar" >> go "received2 a"+ sendChan s1'' "a" >> go "received2 a"+ send p'' "foo" >> go "received1 foo"+ sendChan s1'' "a" >> send p "foo" >> go "received1 foo"+ sendChan s1'' "a" >> send p "bar" >> go "received2 a" go "received2 a" - kill p "BANG"+ kill p'' "BANG" liftIO $ putStrLn "\n---- Test 4 ----"- (s1,r1) <- newChan- (s2,r2) <- newChan- p <- spawnLocal (recTest4 waitr syncs r1 r2)+ (s1''',r1''') <- newChan+ (_ ,r2''') <- newChan+ p''' <- spawnLocal (recTest4 waitr syncs r1''' r2''') - sendChan s1 "a" >> go "received2 a"- send p "foo" >> go "received1 foo"- send p "bar" >> go "received3 bar"- sendChan s1 "a" >> send p "foo" >> go "received1 foo"- send p "bar" >> go "received2 a"- send p "foo" >> go "received1 foo" >> go "received3 bar"+ sendChan s1''' "a" >> go "received2 a"+ send p''' "foo" >> go "received1 foo"+ send p''' "bar" >> go "received3 bar"+ sendChan s1''' "a" >> send p''' "foo" >> go "received1 foo"+ send p''' "bar" >> go "received2 a"+ send p''' "foo" >> go "received1 foo" >> go "received3 bar" - kill p "BANG"+ kill p''' "BANG" terminate
src/Control/Distributed/Process/Tests/Stats.hs view
@@ -16,10 +16,6 @@ import Data.Binary () import Data.Typeable () -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch)-#endif- import Test.Framework ( Test , testGroup@@ -29,7 +25,7 @@ testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process () testLocalDeadProcessInfo result = do- pid <- spawnLocal $ do "finish" <- expect; return ()+ pid <- spawnLocal $ do (_ :: String) <- expect; return () mref <- monitor pid send pid "finish" _ <- receiveWait [@@ -52,8 +48,10 @@ link self mRef <- monitor self stash mon mRef- "die" <- expect- return ()+ res <- expect+ case res of+ "die" -> return ()+ _ -> die $ "unexpected message received: " ++ res monRef <- liftIO $ takeMVar mon @@ -84,9 +82,9 @@ -- our send op shouldn't overtake link or monitor requests AFAICT -- so a little table tennis should get us synchronised properly send serverPid (self, "ping")- "pong" <- expect+ pong <- expect pInfo <- getProcessInfo serverPid- stash result $ pInfo /= Nothing+ stash result $ pong == "pong" && pInfo /= Nothing where launchRemote :: MVar ProcessId -> IO () launchRemote locMV = do@@ -97,8 +95,10 @@ _ <- receiveWait [ match (\(pid, "ping") -> send pid "pong") ]- "stop" <- expect- return ()+ res <- expect+ case res of+ "stop" -> return ()+ _ -> die $ "unexpected message received: " ++ res return () withActiveRemote :: LocalNode
src/Control/Distributed/Process/Tests/Tracing.hs view
@@ -5,6 +5,7 @@ import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..)) +import Control.Applicative ((<*)) import Control.Concurrent (threadDelay) import Control.Concurrent.MVar ( MVar@@ -19,17 +20,21 @@ import Control.Distributed.Process.Management ( MxEvent(..) )+import qualified Control.Exception as IO (bracket)+import Data.List (isPrefixOf, isSuffixOf) -#if ! MIN_VERSION_base(4,6,0)-import Prelude hiding (catch, log)-#endif+import Prelude hiding ((<*)) 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 {@@ -116,9 +121,11 @@ (\ev -> case ev of (MxSent to from msg) -> do- (Just s) <- unwrapMessage msg :: Process (Maybe String)- stash res (to == pid && from == self && s == "hello there")- stash res (to == pid && from == self)+ mS <- unwrapMessage msg :: Process (Maybe String)+ case mS of+ (Just s) -> do stash res (to == pid && from == self && s == "hello there")+ stash res (to == pid && from == self)+ _ -> die "failed state invariant, message type unmatched..." _ -> return ()) $ do send pid "hello there"@@ -237,12 +244,8 @@ -- Here we set up that relay, and then wait for a signal -- that the tracer (on node1) has seen the expected -- MxSpawned message, at which point we're finished- (Just log') <- whereis "logger"- pid <- liftIO $ forkProcess node2 $ do- logRelay <- spawnLocal $ relay log'- reregister "logger" logRelay- getSelfNode >>= stash mvNid >> (expect :: Process ()) + pid <- splinchLogger node2 mvNid nid <- liftIO $ takeMVar mvNid mref <- monitor pid observedPid <- liftIO $ newEmptyMVar@@ -282,6 +285,96 @@ -- 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@@ -323,4 +416,12 @@ (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,26 +5,33 @@ import TEST_SUITE_MODULE (tests) import Network.Transport.Test (TestTransport(..))-import Network.Socket (sClose)+import Network.Socket (close) import Network.Transport.TCP ( createTransportExposeInternals , TransportInternals(socketBetween) , defaultTCPParameters+ , defaultTCPAddr+ , TCPParameters(..) ) import Test.Framework (defaultMainWithArgs) 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 "127.0.0.1" "8080" defaultTCPParameters+ createTransportExposeInternals (defaultTCPAddr "127.0.0.1" "0")+ defaultTCPParameters { transportConnectTimeout = Just 3000000 } ts <- tests TestTransport { testTransport = transport , testBreakConnection = \addr1 addr2 -> do- sock <- socketBetween internals addr1 addr2- sClose sock+ esock <- try $ socketBetween internals addr1 addr2+ either (\e -> const (return ()) (e :: IOException)) close esock threadDelay 10000 } args <- getArgs