distributed-process-tests (empty) → 0.4.1
raw patch · 12 files changed
+2939/−0 lines, 12 filesdep +HUnitdep +ansi-terminaldep +basesetup-changed
Dependencies added: HUnit, ansi-terminal, base, binary, bytestring, distributed-process, distributed-process-tests, distributed-static, network, network-transport, network-transport-tcp, random, rematch, test-framework, test-framework-hunit
Files
- LICENSE +31/−0
- Setup.hs +2/−0
- distributed-process-tests.cabal +118/−0
- src/Control/Distributed/Process/Tests/CH.hs +1324/−0
- src/Control/Distributed/Process/Tests/Closure.hs +469/−0
- src/Control/Distributed/Process/Tests/Internal/Utils.hs +60/−0
- src/Control/Distributed/Process/Tests/Mx.hs +286/−0
- src/Control/Distributed/Process/Tests/Receive.hs +158/−0
- src/Control/Distributed/Process/Tests/Stats.hs +125/−0
- src/Control/Distributed/Process/Tests/Tracing.hs +326/−0
- src/Network/Transport/Test.hs +12/−0
- tests/runTCP.hs +28/−0
+ LICENSE view
@@ -0,0 +1,31 @@+Copyright Well-Typed LLP, 2011-2012++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of the owner nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.+
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ distributed-process-tests.cabal view
@@ -0,0 +1,118 @@+name: distributed-process-tests+version: 0.4.1+synopsis: Tests for distributed-process.+homepage: http://github.com/haskell-distributed/distributed-process-tests+license: BSD3+license-file: LICENSE+Author: Duncan Coutts, Nicolas Wu, Edsko de Vries+Maintainer: watson.timothy@gmail.com, edsko@well-typed.com, duncan@well-typed.com+copyright: Well-Typed LLP+category: Control+build-type: Simple+cabal-version: >=1.8++library+ exposed-modules: Network.Transport.Test+ Control.Distributed.Process.Tests.CH+ Control.Distributed.Process.Tests.Closure+ Control.Distributed.Process.Tests.Mx+ Control.Distributed.Process.Tests.Receive+ 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.8,+ bytestring >= 0.9 && < 0.11,+ distributed-process,+ distributed-static,+ HUnit >= 1.2 && < 1.3,+ network-transport >= 0.4.1.0 && < 0.5,+ network >= 2.3 && < 2.7,+ random >= 1.0 && < 1.1,+ rematch >= 0.1.2.1,+ test-framework >= 0.6 && < 0.9,+ test-framework-hunit >= 0.2.0 && < 0.4+ hs-source-dirs: src+ ghc-options: -Wall -fno-warn-unused-do-bind+ extensions: CPP,+ ExistentialQuantification,+ FlexibleInstances,+ DeriveDataTypeable,+ DeriveGeneric,+ GeneralizedNewtypeDeriving,+ OverlappingInstances,+ RankNTypes,+ RecordWildCards,+ ScopedTypeVariables++Test-Suite TestCH+ Type: exitcode-stdio-1.0+ Main-Is: runTCP.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH+ Build-Depends: base >= 4.4 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 2.5,+ network-transport >= 0.4.1.0 && < 0.5,+ network-transport-tcp >= 0.3 && < 0.5,+ test-framework >= 0.6 && < 0.9+ Extensions: CPP+ ghc-options: -Wall -threaded -debug -eventlog -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests++Test-Suite TestClosure+ Type: exitcode-stdio-1.0+ Main-Is: runTCP.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Closure+ Build-Depends: base >= 4.4 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 2.5,+ network-transport >= 0.4.1.0 && < 0.5,+ network-transport-tcp >= 0.3 && < 0.5,+ test-framework >= 0.6 && < 0.9+ Extensions: CPP+ ghc-options: -Wall -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests++Test-Suite TestStats+ Type: exitcode-stdio-1.0+ Main-Is: runTCP.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats+ Build-Depends: base >= 4.4 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 2.5,+ network-transport >= 0.4.1.0 && < 0.5,+ network-transport-tcp >= 0.3 && < 0.5,+ test-framework >= 0.6 && < 0.9+ Extensions: CPP+ ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests+++Test-Suite TestMx+ Type: exitcode-stdio-1.0+ Main-Is: runTCP.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx+ Build-Depends: base >= 4.4 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 2.5,+ network-transport >= 0.4.1.0 && < 0.5,+ network-transport-tcp >= 0.3 && < 0.5,+ test-framework >= 0.6 && < 0.9+ Extensions: CPP+ ghc-options: -Wall -debug -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests++Test-Suite TestTracing+ Type: exitcode-stdio-1.0+ Main-Is: runTCP.hs+ CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing+ Build-Depends: base >= 4.4 && < 5,+ distributed-process-tests,+ network >= 2.3 && < 2.5,+ network-transport >= 0.4.1.0 && < 0.5,+ network-transport-tcp >= 0.3 && < 0.5,+ test-framework >= 0.6 && < 0.9+ Extensions: CPP+ ghc-options: -Wall -debug -eventlog -threaded -rtsopts -with-rtsopts=-N -fno-warn-unused-do-bind+ HS-Source-Dirs: tests
+ src/Control/Distributed/Process/Tests/CH.hs view
@@ -0,0 +1,1324 @@+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 Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId)+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , putMVar+ , takeMVar+ , readMVar+ )+import Control.Monad (replicateM_, replicateM, forever, void)+import Control.Exception (SomeException, throwIO)+import qualified Control.Exception as Ex (catch)+import Control.Applicative ((<$>), (<*>), pure, (<|>))+import qualified Network.Transport as NT (closeEndPoint)+import Control.Distributed.Process+import Control.Distributed.Process.Internal.Types+ ( NodeId(nodeAddress)+ , LocalNode(localEndPoint)+ , ProcessExitException(..)+ , nullProcessId+ )+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable (Serializable)++import Test.HUnit (Assertion, assertFailure)+import Test.Framework (Test, testGroup)+import Test.Framework.Providers.HUnit (testCase)+import Control.Rematch hiding (match)+import Control.Rematch.Run (Match(..))++newtype Ping = Ping ProcessId+ deriving (Typeable, Binary, Show)++newtype Pong = Pong ProcessId+ deriving (Typeable, Binary, Show)++--------------------------------------------------------------------------------+-- 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))++-- | The ping server from the paper+ping :: Process ()+ping = do+ Pong partner <- expect+ self <- getSelfPid+ send partner (Ping self)+ ping++-- | 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++data Add = Add ProcessId Double Double deriving (Typeable)+data Divide = Divide ProcessId Double Double deriving (Typeable)+data DivByZero = DivByZero deriving (Typeable)++instance Binary Add where+ put (Add pid x y) = put pid >> put x >> put y+ get = Add <$> get <*> get <*> get++instance Binary Divide where+ put (Divide pid x y) = put pid >> put x >> put y+ get = Divide <$> get <*> get <*> get++instance Binary DivByZero where+ put DivByZero = return ()+ get = return DivByZero++-- The math server from the paper+math :: Process ()+math = do+ receiveWait+ [ match (\(Add pid x y) -> send pid (x + y))+ , matchIf (\(Divide _ _ y) -> y /= 0)+ (\(Divide pid x y) -> send pid (x / y))+ , match (\(Divide pid _ _) -> send pid DivByZero)+ ]+ math++-- | Monitor or link to a remote node+monitorOrLink :: Bool -- ^ 'True' for monitor, 'False' for link+ -> ProcessId -- Process to monitor/link to+ -> Maybe (MVar ()) -- MVar to signal on once the monitor has been set up+ -> Process (Maybe MonitorRef)+monitorOrLink mOrL pid mSignal = do+ result <- if mOrL then Just <$> monitor pid+ else link pid >> return Nothing+ -- Monitor is asynchronous, which usually does not matter but if we want a+ -- *specific* signal then it does. Therefore we wait an arbitrary delay and+ -- hope that this means the monitor has been set up+ forM_ mSignal $ \signal -> liftIO . forkIO $ threadDelay 100000 >> putMVar signal ()+ return result++monitorTestProcess :: ProcessId -- Process to monitor/link to+ -> Bool -- 'True' for monitor, 'False' for link+ -> Bool -- Should we unmonitor?+ -> DiedReason -- Expected cause of death+ -> Maybe (MVar ()) -- Signal for 'monitor set up'+ -> MVar () -- Signal for successful termination+ -> Process ()+monitorTestProcess theirAddr mOrL un reason monitorSetup done =+ catch (do mRef <- monitorOrLink mOrL theirAddr monitorSetup+ case (un, mRef) of+ (True, Nothing) -> do+ unlink theirAddr+ liftIO $ putMVar done ()+ (True, Just ref) -> do+ unmonitor ref+ liftIO $ putMVar done ()+ (False, ref) -> do+ ProcessMonitorNotification ref' pid reason' <- expect+ True <- return $ Just ref' == ref && pid == theirAddr && mOrL && reason == reason'+ liftIO $ putMVar done ()+ )+ (\(ProcessLinkException pid reason') -> do+ True <- return $ pid == theirAddr && not mOrL && not un && reason == reason'+ liftIO $ putMVar done ()+ )++--------------------------------------------------------------------------------+-- The tests proper --+--------------------------------------------------------------------------------++-- | Basic ping test+testPing :: TestTransport -> Assertion+testPing TestTransport{..} = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ -- Server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode ping+ putMVar serverAddr addr++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ pingServer <- readMVar serverAddr++ let numPings = 10000++ runProcess localNode $ do+ pid <- getSelfPid+ replicateM_ numPings $ do+ send pingServer (Pong pid)+ Ping _ <- expect+ return ()++ putMVar clientDone ()++ takeMVar clientDone++-- | Monitor a process on an unreachable node+testMonitorUnreachable :: TestTransport -> Bool -> Bool -> Assertion+testMonitorUnreachable TestTransport{..} mOrL un = do+ deadProcess <- newEmptyMVar+ done <- newEmptyMVar++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode . liftIO $ threadDelay 1000000+ closeLocalNode localNode+ putMVar deadProcess addr++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ theirAddr <- readMVar deadProcess+ runProcess localNode $+ monitorTestProcess theirAddr mOrL un DiedDisconnect Nothing done++ takeMVar done++-- | Monitor a process which terminates normally+testMonitorNormalTermination :: TestTransport -> Bool -> Bool -> Assertion+testMonitorNormalTermination TestTransport{..} mOrL un = do+ monitorSetup <- newEmptyMVar+ monitoredProcess <- newEmptyMVar+ done <- newEmptyMVar++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode $+ liftIO $ readMVar monitorSetup+ putMVar monitoredProcess addr++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ theirAddr <- readMVar monitoredProcess+ runProcess localNode $+ monitorTestProcess theirAddr mOrL un DiedNormal (Just monitorSetup) done++ takeMVar done++-- | Monitor a process which terminates abnormally+testMonitorAbnormalTermination :: TestTransport -> Bool -> Bool -> Assertion+testMonitorAbnormalTermination TestTransport{..} mOrL un = do+ monitorSetup <- newEmptyMVar+ monitoredProcess <- newEmptyMVar+ done <- newEmptyMVar++ let err = userError "Abnormal termination"++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode . liftIO $ do+ readMVar monitorSetup+ throwIO err+ putMVar monitoredProcess addr++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ theirAddr <- readMVar monitoredProcess+ runProcess localNode $+ monitorTestProcess theirAddr mOrL un (DiedException (show err)) (Just monitorSetup) done++ takeMVar done++-- | Monitor a local process that is already dead+testMonitorLocalDeadProcess :: TestTransport -> Bool -> Bool -> Assertion+testMonitorLocalDeadProcess TestTransport{..} mOrL un = do+ processDead <- newEmptyMVar+ processAddr <- newEmptyMVar+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ forkIO $ do+ addr <- forkProcess localNode . liftIO $ putMVar processDead ()+ putMVar processAddr addr++ forkIO $ do+ theirAddr <- readMVar processAddr+ readMVar processDead+ runProcess localNode $ do+ monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done++ takeMVar done++-- | Monitor a remote process that is already dead+testMonitorRemoteDeadProcess :: TestTransport -> Bool -> Bool -> Assertion+testMonitorRemoteDeadProcess TestTransport{..} mOrL un = do+ processDead <- newEmptyMVar+ processAddr <- newEmptyMVar+ done <- newEmptyMVar++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode . liftIO $ putMVar processDead ()+ putMVar processAddr addr++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ theirAddr <- readMVar processAddr+ readMVar processDead+ runProcess localNode $ do+ monitorTestProcess theirAddr mOrL un DiedUnknownId Nothing done++ takeMVar done++-- | Monitor a process that becomes disconnected+testMonitorDisconnect :: TestTransport -> Bool -> Bool -> Assertion+testMonitorDisconnect TestTransport{..} mOrL un = do+ processAddr <- newEmptyMVar+ monitorSetup <- newEmptyMVar+ done <- newEmptyMVar++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode . liftIO $ threadDelay 1000000+ putMVar processAddr addr+ readMVar monitorSetup+ NT.closeEndPoint (localEndPoint localNode)++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ theirAddr <- readMVar processAddr+ runProcess localNode $ do+ monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done++ takeMVar done++-- | Test the math server (i.e., receiveWait)+testMath :: TestTransport -> Assertion+testMath TestTransport{..} = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ -- Server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode math+ putMVar serverAddr addr++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ mathServer <- readMVar serverAddr++ runProcess localNode $ do+ pid <- getSelfPid+ send mathServer (Add pid 1 2)+ 3 <- expect :: Process Double+ send mathServer (Divide pid 8 2)+ 4 <- expect :: Process Double+ send mathServer (Divide pid 8 0)+ DivByZero <- expect+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++-- | Send first message (i.e. connect) to an already terminated process+-- (without monitoring); then send another message to a second process on+-- the same remote node (we're checking that the remote node did not die)+testSendToTerminated :: TestTransport -> Assertion+testSendToTerminated TestTransport{..} = do+ serverAddr1 <- newEmptyMVar+ serverAddr2 <- newEmptyMVar+ clientDone <- newEmptyMVar++ forkIO $ do+ terminated <- newEmptyMVar+ localNode <- newLocalNode testTransport initRemoteTable+ addr1 <- forkProcess localNode $ liftIO $ putMVar terminated ()+ addr2 <- forkProcess localNode $ ping+ readMVar terminated+ putMVar serverAddr1 addr1+ putMVar serverAddr2 addr2++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ server1 <- readMVar serverAddr1+ server2 <- readMVar serverAddr2+ runProcess localNode $ do+ pid <- getSelfPid+ send server1 "Hi"+ send server2 (Pong pid)+ Ping pid' <- expect+ True <- return $ pid' == server2+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++-- | Test (non-zero) timeout+testTimeout :: TestTransport -> Assertion+testTimeout TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ runProcess localNode $ do+ Nothing <- receiveTimeout 1000000 [match (\(Add _ _ _) -> return ())]+ liftIO $ putMVar done ()++ takeMVar done++-- | Test zero timeout+testTimeout0 :: TestTransport -> Assertion+testTimeout0 TestTransport{..} = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar+ messagesSent <- newEmptyMVar++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ addr <- forkProcess localNode $ do+ liftIO $ readMVar messagesSent >> threadDelay 1000000+ -- Variation on the venerable ping server which uses a zero timeout+ -- Since we wait for all messages to be sent before doing this receive,+ -- we should nevertheless find the right message immediately+ Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)]+ self <- getSelfPid+ send partner (Ping self)+ putMVar serverAddr addr++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ server <- readMVar serverAddr+ runProcess localNode $ do+ pid <- getSelfPid+ -- Send a bunch of messages. A large number of messages that the server+ -- 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 ()++ takeMVar clientDone++-- | Test typed channels+testTypedChannels :: TestTransport -> Assertion+testTypedChannels TestTransport{..} = do+ serverChannel <- newEmptyMVar :: IO (MVar (SendPort (SendPort Bool, Int)))+ clientDone <- newEmptyMVar++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ forkProcess localNode $ do+ (serverSendPort, rport) <- newChan+ liftIO $ putMVar serverChannel serverSendPort+ (clientSendPort, i) <- receiveChan rport+ sendChan clientSendPort (even i)+ return ()++ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ serverSendPort <- readMVar serverChannel+ runProcess localNode $ do+ (clientSendPort, rport) <- newChan+ sendChan serverSendPort (clientSendPort, 5)+ False <- receiveChan rport+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++-- | Test merging receive ports+testMergeChannels :: TestTransport -> Assertion+testMergeChannels TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ testFlat localNode True "aaabbbccc"+ testFlat localNode False "abcabcabc"+ testNested localNode True True "aaabbbcccdddeeefffggghhhiii"+ testNested localNode True False "adgadgadgbehbehbehcficficfi"+ testNested localNode False True "abcabcabcdefdefdefghighighi"+ testNested localNode False False "adgbehcfiadgbehcfiadgbehcfi"+ testBlocked localNode True+ testBlocked localNode False+ where+ -- Single layer of merging+ testFlat :: LocalNode -> Bool -> String -> IO ()+ testFlat localNode biased expected = do+ done <- newEmptyMVar+ forkProcess localNode $ do+ rs <- mapM charChannel "abc"+ m <- mergePorts biased rs+ xs <- replicateM 9 $ receiveChan m+ True <- return $ xs == expected+ liftIO $ putMVar done ()+ takeMVar done++ -- Two layers of merging+ testNested :: LocalNode -> Bool -> Bool -> String -> IO ()+ testNested localNode biasedInner biasedOuter expected = do+ done <- newEmptyMVar+ forkProcess localNode $ do+ rss <- mapM (mapM charChannel) ["abc", "def", "ghi"]+ ms <- mapM (mergePorts biasedInner) rss+ m <- mergePorts biasedOuter ms+ xs <- replicateM (9 * 3) $ receiveChan m+ True <- return $ xs == expected+ liftIO $ putMVar done ()+ takeMVar done++ -- Test that if no messages are (immediately) available, the scheduler makes no difference+ testBlocked :: LocalNode -> Bool -> IO ()+ testBlocked localNode biased = do+ vs <- replicateM 3 newEmptyMVar+ 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')+ ]++ 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 ()++ takeMVar done++ mergePorts :: Serializable a => Bool -> [ReceivePort a] -> Process (ReceivePort a)+ mergePorts True = mergePortsBiased+ mergePorts False = mergePortsRR++ charChannel :: Char -> Process (ReceivePort Char)+ charChannel c = do+ (sport, rport) <- newChan+ replicateM_ 3 $ sendChan sport c+ liftIO $ threadDelay 10000 -- Make sure messages have been sent+ return rport++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++testMonitorNode :: TestTransport -> Assertion+testMonitorNode TestTransport{..} = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ closeLocalNode node1++ runProcess node2 $ do+ ref <- monitorNode (localNodeId node1)+ NodeMonitorNotification ref' nid DiedDisconnect <- expect+ True <- return $ ref == ref' && nid == localNodeId node1+ liftIO $ putMVar done ()++ takeMVar done++testMonitorLiveNode :: TestTransport -> Assertion+testMonitorLiveNode TestTransport{..} = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable+ ready <- newEmptyMVar+ done <- newEmptyMVar++ forkProcess node2 $ do+ ref <- monitorNode (localNodeId node1)+ liftIO $ putMVar ready ()+ NodeMonitorNotification ref' nid _ <- expect+ True <- return $ ref == ref' && nid == localNodeId node1+ liftIO $ putMVar done ()++ takeMVar ready+ closeLocalNode node1++ takeMVar done++testMonitorChannel :: TestTransport -> Assertion+testMonitorChannel TestTransport{..} = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable+ gotNotification <- newEmptyMVar++ 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 ()++ runProcess node2 $ do+ (sport, _) <- newChan :: Process (SendPort (), ReceivePort ())+ send pid sport+ liftIO $ threadDelay 100000++ takeMVar gotNotification++testRegistry :: TestTransport -> Assertion+testRegistry TestTransport{..} = do+ node <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pingServer <- forkProcess node ping++ runProcess node $ do+ register "ping" pingServer+ Just pid <- whereis "ping"+ True <- return $ pingServer == pid+ us <- getSelfPid+ nsend "ping" (Pong us)+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ liftIO $ putMVar done ()++ takeMVar done++testRemoteRegistry :: TestTransport -> Assertion+testRemoteRegistry TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pingServer <- forkProcess node1 ping++ runProcess node2 $ do+ let nid1 = localNodeId node1+ registerRemoteAsync nid1 "ping" pingServer+ receiveWait [+ matchIf (\(RegisterReply label' _) -> "ping" == label')+ (\(RegisterReply _ _) -> return ()) ]++ Just pid <- whereisRemote nid1 "ping"+ True <- return $ pingServer == pid+ us <- getSelfPid+ nsendRemote nid1 "ping" (Pong us)+ Ping pid' <- expect+ True <- return $ pingServer == pid'+ liftIO $ putMVar done ()++ takeMVar done++testSpawnLocal :: TestTransport -> Assertion+testSpawnLocal TestTransport{..} = do+ node <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ runProcess node $ do+ us <- getSelfPid++ pid <- spawnLocal $ do+ sport <- expect+ sendChan sport (1234 :: Int)++ sport <- spawnChannelLocal $ \rport -> do+ (1234 :: Int) <- receiveChan rport+ send us ()++ send pid sport+ () <- expect+ liftIO $ putMVar done ()++ takeMVar done++testReconnect :: TestTransport -> Assertion+testReconnect TestTransport{..} = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable+ let nid1 = localNodeId node1+ nid2 = localNodeId node2+ processA <- newEmptyMVar+ [sendTestOk, registerTestOk] <- replicateM 2 newEmptyMVar++ forkProcess node1 $ do+ us <- getSelfPid+ liftIO $ putMVar processA us+ msg1 <- expect+ msg2 <- expect+ True <- return $ msg1 == "message 1" && msg2 == "message 3"+ liftIO $ putMVar sendTestOk ()++ forkProcess node2 $ do+ {-+ - Make sure there is no implicit reconnect on normal message sending+ -}++ them <- liftIO $ readMVar processA+ send them "message 1" >> liftIO (threadDelay 100000)++ -- Simulate network failure+ liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)++ -- Should not arrive+ send them "message 2"++ -- Should arrive+ reconnect them+ send them "message 3"++ liftIO $ takeMVar sendTestOk++ {-+ - Test that there *is* implicit reconnect on node controller messages+ -}++ us <- getSelfPid+ registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous+ receiveWait [+ matchIf (\(RegisterReply label' _) -> "a" == label')+ (\(RegisterReply _ _) -> return ()) ]++ Just _ <- whereisRemote nid1 "a"+++ -- Simulate network failure+ liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2)++ -- This will happen due to implicit reconnect+ registerRemoteAsync nid1 "b" us+ receiveWait [+ matchIf (\(RegisterReply label' _) -> "b" == label')+ (\(RegisterReply _ _) -> return ()) ]++ -- Should happen+ registerRemoteAsync nid1 "c" us+ receiveWait [+ matchIf (\(RegisterReply label' _) -> "c" == label')+ (\(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"++ liftIO $ putMVar registerTestOk ()++ takeMVar registerTestOk++-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server+-- in between+testMatchAny :: TestTransport -> Assertion+testMatchAny TestTransport{..} = do+ proxyAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ -- Math server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ mathServer <- forkProcess localNode math+ proxyServer <- forkProcess localNode $ forever $ do+ msg <- receiveWait [ matchAny return ]+ forward msg mathServer+ putMVar proxyAddr proxyServer++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ mathServer <- readMVar proxyAddr++ runProcess localNode $ do+ pid <- getSelfPid+ send mathServer (Add pid 1 2)+ 3 <- expect :: Process Double+ send mathServer (Divide pid 8 2)+ 4 <- expect :: Process Double+ send mathServer (Divide pid 8 0)+ DivByZero <- expect+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server+-- in between, however we block 'Divide' requests ....+testMatchAnyHandle :: TestTransport -> Assertion+testMatchAnyHandle TestTransport{..} = do+ proxyAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ -- Math server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ mathServer <- forkProcess localNode math+ proxyServer <- forkProcess localNode $ forever $ do+ receiveWait [+ matchAny (maybeForward mathServer)+ ]+ putMVar proxyAddr proxyServer++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ mathServer <- readMVar proxyAddr++ runProcess localNode $ do+ pid <- getSelfPid+ send mathServer (Add pid 1 2)+ 3 <- expect :: Process Double+ send mathServer (Divide pid 8 2)+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)+ liftIO $ putMVar clientDone ()++ takeMVar clientDone+ where maybeForward :: ProcessId -> Message -> Process (Maybe ())+ maybeForward s msg =+ handleMessage msg (\m@(Add _ _ _) -> send s m)++testMatchAnyNoHandle :: TestTransport -> Assertion+testMatchAnyNoHandle TestTransport{..} = do+ addr <- newEmptyMVar+ clientDone <- newEmptyMVar+ serverDone <- newEmptyMVar++ -- Math server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ server <- forkProcess localNode $ forever $ do+ receiveWait [+ matchAnyIf+ -- the condition has type `Add -> Bool`+ (\(Add _ _ _) -> True)+ -- the match `AbstractMessage -> Process ()` will succeed!+ (\m -> do+ -- `String -> Process ()` does *not* match the input types however+ r <- (handleMessage m (\(_ :: String) -> die "NONSENSE" ))+ case r of+ Nothing -> return ()+ Just _ -> die "NONSENSE")+ ]+ -- we *must* have removed the message from our mailbox though!!!+ Nothing <- receiveTimeout 100000 [ match (\(Add _ _ _) -> return ()) ]+ liftIO $ putMVar serverDone ()+ putMVar addr server++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ server <- readMVar addr++ runProcess localNode $ do+ pid <- getSelfPid+ send server (Add pid 1 2)+ -- we only care about the client having sent a message, so we're done+ liftIO $ putMVar clientDone ()++ takeMVar clientDone+ takeMVar serverDone++-- | Test 'matchAnyIf'. We provide an /echo/ server, but it ignores requests+-- unless the text body @/= "bar"@ - this case should time out rather than+-- removing the message from the process mailbox.+testMatchAnyIf :: TestTransport -> Assertion+testMatchAnyIf TestTransport{..} = do+ echoAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ -- echo server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ echoServer <- forkProcess localNode $ forever $ do+ receiveWait [+ matchAnyIf (\(_ :: ProcessId, (s :: String)) -> s /= "bar")+ tryHandleMessage+ ]+ putMVar echoAddr echoServer++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ server <- readMVar echoAddr++ runProcess localNode $ do+ pid <- getSelfPid+ send server (pid, "foo")+ "foo" <- expect+ send server (pid, "baz")+ "baz" <- expect+ send server (pid, "bar")+ Nothing <- (expectTimeout 100000) :: Process (Maybe Double)+ liftIO $ putMVar clientDone ()++ takeMVar clientDone+ where tryHandleMessage :: Message -> Process (Maybe ())+ tryHandleMessage msg =+ handleMessage msg (\(pid :: ProcessId, (m :: String))+ -> do { send pid m; return () })++testMatchMessageWithUnwrap :: TestTransport -> Assertion+testMatchMessageWithUnwrap TestTransport{..} = do+ echoAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ -- echo server+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ echoServer <- forkProcess localNode $ forever $ do+ msg <- receiveWait [+ matchMessage (\(m :: Message) -> do+ return m)+ ]+ unwrapped <- unwrapMessage msg :: Process (Maybe (ProcessId, Message))+ case unwrapped of+ (Just (p, msg')) -> forward msg' p+ Nothing -> die "unable to unwrap the message"+ putMVar echoAddr echoServer++ -- Client+ forkIO $ do+ localNode <- newLocalNode testTransport initRemoteTable+ server <- readMVar echoAddr++ runProcess localNode $ do+ pid <- getSelfPid+ send server (pid, wrapMessage ("foo" :: String))+ "foo" <- expect+ send server (pid, wrapMessage ("baz" :: String))+ "baz" <- expect+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++-- Test 'receiveChanTimeout'+testReceiveChanTimeout :: TestTransport -> Assertion+testReceiveChanTimeout TestTransport{..} = do+ done <- newEmptyMVar+ sendPort <- newEmptyMVar++ forkTry $ do+ localNode <- newLocalNode testTransport initRemoteTable+ runProcess localNode $ do+ -- Create a typed channel+ (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 again. Now a message arrives after 0.5 seconds+ Just True <- receiveChanTimeout 1000000 rp++ -- Wait for a message with zero timeout: non-blocking check. No message is available, we get Nothing+ Nothing <- receiveChanTimeout 0 rp++ -- Again, but now there is a message available+ liftIO $ threadDelay 1000000+ Just False <- receiveChanTimeout 0 rp++ liftIO $ putMVar done ()++ forkTry $ do+ localNode <- newLocalNode testTransport initRemoteTable+ runProcess localNode $ do+ sp <- liftIO $ readMVar sendPort++ liftIO $ threadDelay 1500000+ sendChan sp True++ liftIO $ threadDelay 500000+ sendChan sp False++ takeMVar done++-- | Test Functor, Applicative, Alternative and Monad instances for ReceiveChan+testReceiveChanFeatures :: TestTransport -> Assertion+testReceiveChanFeatures TestTransport{..} = do+ done <- newEmptyMVar++ forkTry $ do+ localNode <- newLocalNode testTransport initRemoteTable+ runProcess localNode $ do+ (spInt, rpInt) <- newChan :: Process (SendPort Int, ReceivePort Int)+ (spBool, rpBool) <- newChan :: Process (SendPort Bool, ReceivePort Bool)++ -- Test Functor instance++ sendChan spInt 2+ sendChan spBool False++ rp1 <- mergePortsBiased [even <$> rpInt, rpBool]++ True <- receiveChan rp1+ False <- receiveChan rp1++ -- Test Applicative instance++ sendChan spInt 3+ sendChan spInt 4++ let rp2 = pure (+) <*> rpInt <*> rpInt++ 7 <- receiveChan rp2++ -- Test Alternative instance++ sendChan spInt 3+ sendChan spBool True++ let rp3 = (even <$> rpInt) <|> rpBool++ False <- receiveChan rp3+ True <- receiveChan rp3++ -- Test Monad instance++ sendChan spBool True+ sendChan spBool False+ sendChan spInt 5++ let rp4 :: ReceivePort Int+ rp4 = do b <- rpBool+ if b+ then rpInt+ else return 7++ 5 <- receiveChan rp4+ 7 <- receiveChan rp4++ liftIO $ putMVar done ()++ takeMVar done++testKillLocal :: TestTransport -> Assertion+testKillLocal TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pid <- forkProcess localNode $ do+ liftIO $ threadDelay 1000000++ runProcess localNode $ do+ 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++testKillRemote :: TestTransport -> Assertion+testKillRemote TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ pid <- forkProcess node1 $ do+ liftIO $ threadDelay 1000000++ runProcess node2 $ do+ 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++testCatchesExit :: TestTransport -> Assertion+testCatchesExit TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ _ <- forkProcess localNode $ do+ (die ("foobar", 123 :: Int))+ `catchesExit` [+ (\_ m -> handleMessage m (\(_ :: String) -> return ()))+ , (\_ m -> handleMessage m (\(_ :: Maybe Int) -> return ()))+ , (\_ m -> handleMessage m (\(_ :: String, _ :: Int)+ -> (liftIO $ putMVar done ()) >> return ()))+ ]++ takeMVar done++testHandleMessageIf :: TestTransport -> Assertion+testHandleMessageIf TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar+ _ <- forkProcess localNode $ do+ self <- getSelfPid+ send self (5 :: Integer, 10 :: Integer)+ msg <- receiveWait [ matchMessage return ]+ Nothing <- handleMessageIf msg (\() -> True) (\() -> die $ "whoops")+ handleMessageIf msg (\(x :: Integer, y :: Integer) -> x == 5 && y == 10)+ (\input -> liftIO $ putMVar done input)+ return ()++ result <- takeMVar done+ expectThat result $ equalTo (5, 10)++testCatches :: TestTransport -> Assertion+testCatches TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ _ <- forkProcess localNode $ do+ node <- getSelfNode+ (liftIO $ throwIO (ProcessLinkException (nullProcessId node) DiedNormal))+ `catches` [+ Handler (\(ProcessLinkException _ _) -> liftIO $ putMVar done ())+ ]++ takeMVar done++testMaskRestoreScope :: TestTransport -> Assertion+testMaskRestoreScope TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ parentPid <- newEmptyMVar :: IO (MVar ProcessId)+ spawnedPid <- newEmptyMVar :: IO (MVar ProcessId)++ 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++testDie :: TestTransport -> Assertion+testDie TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ _ <- forkProcess localNode $ do+ (die ("foobar", 123 :: Int))+ `catchExit` \_from reason -> do+ -- TODO: should verify that 'from' has the right value+ True <- return $ reason == ("foobar", 123 :: Int)+ liftIO $ putMVar done ()++ takeMVar done++testPrettyExit :: TestTransport -> Assertion+testPrettyExit TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ done <- newEmptyMVar++ _ <- forkProcess localNode $ do+ (die "timeout")+ `catch` \ex@(ProcessExitException from _) ->+ let expected = "exit-from=" ++ (show from)+ in do+ True <- return $ (show ex) == expected+ liftIO $ putMVar done ()++ takeMVar done++testExitLocal :: TestTransport -> Assertion+testExitLocal TestTransport{..} = do+ localNode <- newLocalNode testTransport initRemoteTable+ supervisedDone <- newEmptyMVar+ supervisorDone <- newEmptyMVar++ pid <- forkProcess localNode $ do+ (liftIO $ threadDelay 100000)+ `catchExit` \_from reason -> do+ -- TODO: should verify that 'from' has the right value+ True <- return $ reason == "TestExit"+ liftIO $ putMVar supervisedDone ()++ runProcess localNode $ do+ 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 ()++ takeMVar supervisedDone+ takeMVar supervisorDone++testExitRemote :: TestTransport -> Assertion+testExitRemote TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ node2 <- newLocalNode testTransport initRemoteTable+ supervisedDone <- newEmptyMVar+ supervisorDone <- newEmptyMVar++ pid <- forkProcess node1 $ do+ (liftIO $ threadDelay 100000)+ `catchExit` \_from reason -> do+ -- TODO: should verify that 'from' has the right value+ True <- return $ reason == "TestExit"+ liftIO $ putMVar supervisedDone ()++ runProcess node2 $ do+ ref <- monitor pid+ exit pid "TestExit"+ ProcessMonitorNotification ref' pid' DiedNormal <- expect+ True <- return $ ref == ref' && pid == pid'+ liftIO $ putMVar supervisorDone ()++ takeMVar supervisedDone+ takeMVar supervisorDone++testUnsafeSend :: TestTransport -> Assertion+testUnsafeSend TestTransport{..} = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ localNode <- newLocalNode testTransport initRemoteTable+ void $ forkProcess localNode $ do+ self <- getSelfPid+ liftIO $ putMVar serverAddr self+ clientAddr <- expect+ unsafeSend clientAddr ()++ void $ forkProcess localNode $ do+ serverPid <- liftIO $ takeMVar serverAddr+ getSelfPid >>= unsafeSend serverPid+ () <- expect+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++testUnsafeNSend :: TestTransport -> Assertion+testUnsafeNSend TestTransport{..} = do+ clientDone <- newEmptyMVar++ localNode <- newLocalNode testTransport initRemoteTable++ pid <- forkProcess localNode $ do+ () <- expect+ liftIO $ putMVar clientDone ()++ void $ runProcess localNode $ do+ register "foobar" pid+ unsafeNSend "foobar" ()++ takeMVar clientDone++testUnsafeSendChan :: TestTransport -> Assertion+testUnsafeSendChan TestTransport{..} = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ localNode <- newLocalNode testTransport initRemoteTable+ void $ forkProcess localNode $ do+ self <- getSelfPid+ liftIO $ putMVar serverAddr self+ sp <- expect+ unsafeSendChan sp ()++ void $ forkProcess localNode $ do+ serverPid <- liftIO $ takeMVar serverAddr+ (sp, rp) <- newChan+ unsafeSend serverPid sp+ () <- receiveChan rp+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++tests :: TestTransport -> IO [Test]+tests testtrans = return [+ testGroup "Basic features" [+ testCase "Ping" (testPing testtrans)+ , testCase "Math" (testMath testtrans)+ , testCase "Timeout" (testTimeout testtrans)+ , 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 "RemoteRegistry" (testRemoteRegistry testtrans)+ , testCase "SpawnLocal" (testSpawnLocal testtrans)+ , testCase "HandleMessageIf" (testHandleMessageIf testtrans)+ , testCase "MatchAny" (testMatchAny testtrans)+ , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans)+ , testCase "MatchAnyNoHandle" (testMatchAnyNoHandle testtrans)+ , testCase "MatchAnyIf" (testMatchAnyIf testtrans)+ , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)+ , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)+ , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)+ , testCase "KillLocal" (testKillLocal testtrans)+ , testCase "KillRemote" (testKillRemote testtrans)+ , testCase "Die" (testDie testtrans)+ , testCase "PrettyExit" (testPrettyExit testtrans)+ , testCase "CatchesExit" (testCatchesExit testtrans)+ , testCase "Catches" (testCatches testtrans)+ , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)+ , testCase "ExitLocal" (testExitLocal testtrans)+ , testCase "ExitRemote" (testExitRemote testtrans)+ -- Unsafe Primitives+ , testCase "TestUnsafeSend" (testUnsafeSend testtrans)+ , testCase "TestUnsafeNSend" (testUnsafeNSend testtrans)+ , testCase "TestUnsafeSendChan" (testUnsafeSendChan testtrans)+ ]+ , testGroup "Monitoring and Linking" [+ -- Monitoring processes+ --+ -- 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 "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False)+ , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False)+ , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False)+ , testCase "MonitorDisconnect" (testMonitorDisconnect testtrans True False)+ , testCase "LinkUnreachable" (testMonitorUnreachable testtrans False False)+ , testCase "LinkNormalTermination" (testMonitorNormalTermination testtrans False False)+ , testCase "LinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False False)+ , testCase "LinkLocalDeadProcess" (testMonitorLocalDeadProcess testtrans False False)+ , testCase "LinkRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans False False)+ , testCase "LinkDisconnect" (testMonitorDisconnect testtrans False False)+ , testCase "UnmonitorNormalTermination" (testMonitorNormalTermination testtrans True True)+ , testCase "UnmonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True True)+ , testCase "UnmonitorDisconnect" (testMonitorDisconnect testtrans True True)+ , testCase "UnlinkNormalTermination" (testMonitorNormalTermination testtrans False True)+ , testCase "UnlinkAbnormalTermination" (testMonitorAbnormalTermination testtrans False True)+ , testCase "UnlinkDisconnect" (testMonitorDisconnect testtrans False True)+ -- Monitoring nodes and channels+ , testCase "MonitorNode" (testMonitorNode testtrans)+ , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans)+ , testCase "MonitorChannel" (testMonitorChannel testtrans)+ -- Reconnect+ , testCase "Reconnect" (testReconnect testtrans)+ ]+ ]
+ src/Control/Distributed/Process/Tests/Closure.hs view
@@ -0,0 +1,469 @@+{-# LANGUAGE TemplateHaskell #-}+module Control.Distributed.Process.Tests.Closure (tests) where++import Network.Transport.Test (TestTransport(..))++import Data.ByteString.Lazy (empty)+import Data.Typeable (Typeable)+import Control.Monad (join, replicateM, forever, replicateM_, void)+import Control.Exception (IOException, throw)+import Control.Concurrent (forkIO, threadDelay)+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , readMVar+ , takeMVar+ , putMVar+ , 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))+import Control.Distributed.Static (staticLabel, staticClosure)++import Test.HUnit (Assertion)+import Test.Framework (Test)+import Test.Framework.Providers.HUnit (testCase)++--------------------------------------------------------------------------------+-- Supporting definitions --+--------------------------------------------------------------------------------++quintuple :: a -> b -> c -> d -> e -> (a, b, c, d, e)+quintuple a b c d e = (a, b, c, d, e)++sdictInt :: SerializableDict Int+sdictInt = SerializableDict++factorial :: Int -> Process Int+factorial 0 = return 1+factorial n = (n *) <$> factorial (n - 1)++addInt :: Int -> Int -> Int+addInt x y = x + y++putInt :: Int -> MVar Int -> IO ()+putInt = flip putMVar++sendPid :: ProcessId -> Process ()+sendPid toPid = do+ fromPid <- getSelfPid+ send toPid fromPid++wait :: Int -> Process ()+wait = liftIO . threadDelay++isPrime :: Integer -> Process Bool+isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..]+ where+ sieve :: [Integer] -> [Integer]+ sieve (p : xs) = p : sieve [x | x <- xs, x `mod` p > 0]+ sieve [] = error "Uh oh -- we've run out of primes"++-- | First argument indicates empty closure environment+typedPingServer :: () -> ReceivePort (SendPort ()) -> Process ()+typedPingServer () rport = forever $ do+ sport <- receiveChan rport+ sendChan sport ()++signal :: ProcessId -> Process ()+signal pid = send pid ()++remotable [ 'factorial+ , 'addInt+ , 'putInt+ , 'sendPid+ , 'sdictInt+ , 'wait+ , 'typedPingServer+ , 'isPrime+ , 'quintuple+ , 'signal+ ]++randomElement :: [a] -> IO a+randomElement xs = do+ ix <- randomIO+ return (xs !! (ix `mod` length xs))++remotableDecl [+ [d| dfib :: ([NodeId], SendPort Integer, Integer) -> Process () ;+ dfib (_, reply, 0) = sendChan reply 0+ dfib (_, reply, 1) = sendChan reply 1+ dfib (nids, reply, n) = do+ nid1 <- liftIO $ randomElement nids+ nid2 <- liftIO $ randomElement nids+ (sport, rport) <- newChan+ spawn nid1 $ $(mkClosure 'dfib) (nids, sport, n - 2)+ spawn nid2 $ $(mkClosure 'dfib) (nids, sport, n - 1)+ n1 <- receiveChan rport+ n2 <- receiveChan rport+ sendChan reply $ n1 + n2+ |]+ ]++-- Just try creating a static polymorphic value+staticQuintuple :: (Typeable a, Typeable b, Typeable c, Typeable d, Typeable e)+ => Static (a -> b -> c -> d -> e -> (a, b, c, d, e))+staticQuintuple = $(mkStatic 'quintuple)++factorialClosure :: Int -> Closure (Process Int)+factorialClosure = $(mkClosure 'factorial)++addIntClosure :: Int -> Closure (Int -> Int)+addIntClosure = $(mkClosure 'addInt)++putIntClosure :: Int -> Closure (MVar Int -> IO ())+putIntClosure = $(mkClosure 'putInt)++sendPidClosure :: ProcessId -> Closure (Process ())+sendPidClosure = $(mkClosure 'sendPid)++sendFac :: Int -> ProcessId -> Closure (Process ())+sendFac n pid = factorialClosure n `bindCP` cpSend $(mkStatic 'sdictInt) pid++factorialOf :: Closure (Int -> Process Int)+factorialOf = staticClosure $(mkStatic 'factorial)++factorial' :: Int -> Closure (Process Int)+factorial' n = returnCP $(mkStatic 'sdictInt) n `bindCP` factorialOf++waitClosure :: Int -> Closure (Process ())+waitClosure = $(mkClosure 'wait)++simulateNetworkFailure :: TestTransport -> NodeId -> NodeId -> Process ()+simulateNetworkFailure TestTransport{..} from to = liftIO $ do+ threadDelay 10000+ testBreakConnection (nodeAddress from) (nodeAddress to)+ threadDelay 10000++--------------------------------------------------------------------------------+-- The tests proper --+--------------------------------------------------------------------------------++testUnclosure :: TestTransport -> RemoteTable -> Assertion+testUnclosure TestTransport{..} rtable = do+ node <- newLocalNode testTransport rtable+ done <- newEmptyMVar+ forkProcess node $ do+ 120 <- join . unClosure $ factorialClosure 5+ liftIO $ putMVar done ()+ takeMVar done++testBind :: TestTransport -> RemoteTable -> Assertion+testBind TestTransport{..} rtable = do+ node <- newLocalNode testTransport rtable+ done <- newEmptyMVar+ runProcess node $ do+ us <- getSelfPid+ join . unClosure $ sendFac 6 us+ (720 :: Int) <- expect+ liftIO $ putMVar done ()+ takeMVar done++testSendPureClosure :: TestTransport -> RemoteTable -> Assertion+testSendPureClosure TestTransport{..} rtable = do+ serverAddr <- newEmptyMVar+ serverDone <- newEmptyMVar++ forkIO $ do+ node <- newLocalNode testTransport rtable+ addr <- forkProcess node $ do+ cl <- expect+ fn <- unClosure cl :: Process (Int -> Int)+ 13 <- return $ fn 6+ liftIO $ putMVar serverDone ()+ putMVar serverAddr addr++ forkIO $ do+ node <- newLocalNode testTransport rtable+ theirAddr <- readMVar serverAddr+ runProcess node $ send theirAddr (addIntClosure 7)++ takeMVar serverDone++testSendIOClosure :: TestTransport -> RemoteTable -> Assertion+testSendIOClosure TestTransport{..} rtable = do+ serverAddr <- newEmptyMVar+ serverDone <- newEmptyMVar++ forkIO $ do+ node <- newLocalNode testTransport rtable+ addr <- forkProcess node $ do+ cl <- expect+ io <- unClosure cl :: Process (MVar Int -> IO ())+ liftIO $ do+ someMVar <- newEmptyMVar+ io someMVar+ 5 <- readMVar someMVar+ putMVar serverDone ()+ putMVar serverAddr addr++ forkIO $ do+ node <- newLocalNode testTransport rtable+ theirAddr <- readMVar serverAddr+ runProcess node $ send theirAddr (putIntClosure 5)++ takeMVar serverDone++testSendProcClosure :: TestTransport -> RemoteTable -> Assertion+testSendProcClosure TestTransport{..} rtable = do+ serverAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ forkIO $ do+ node <- newLocalNode testTransport rtable+ addr <- forkProcess node $ do+ cl <- expect+ pr <- unClosure cl :: Process (Int -> Process ())+ pr 5+ putMVar serverAddr addr++ forkIO $ do+ node <- newLocalNode testTransport rtable+ theirAddr <- readMVar serverAddr+ runProcess node $ do+ pid <- getSelfPid+ send theirAddr (cpSend $(mkStatic 'sdictInt) pid)+ 5 <- expect :: Process Int+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++testSpawn :: TestTransport -> RemoteTable -> Assertion+testSpawn TestTransport{..} rtable = do+ serverNodeAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ forkIO $ do+ node <- newLocalNode testTransport rtable+ putMVar serverNodeAddr (localNodeId node)++ forkIO $ do+ node <- newLocalNode testTransport rtable+ nid <- readMVar serverNodeAddr+ runProcess node $ do+ pid <- getSelfPid+ pid' <- spawn nid (sendPidClosure pid)+ pid'' <- expect+ True <- return $ pid' == pid''+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++testCall :: TestTransport -> RemoteTable -> Assertion+testCall TestTransport{..} rtable = do+ serverNodeAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ forkIO $ do+ node <- newLocalNode testTransport rtable+ putMVar serverNodeAddr (localNodeId node)++ forkIO $ do+ node <- newLocalNode testTransport rtable+ nid <- readMVar serverNodeAddr+ runProcess node $ do+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorialClosure 5)+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++testCallBind :: TestTransport -> RemoteTable -> Assertion+testCallBind TestTransport{..} rtable = do+ serverNodeAddr <- newEmptyMVar+ clientDone <- newEmptyMVar++ forkIO $ do+ node <- newLocalNode testTransport rtable+ putMVar serverNodeAddr (localNodeId node)++ forkIO $ do+ node <- newLocalNode testTransport rtable+ nid <- readMVar serverNodeAddr+ runProcess node $ do+ (120 :: Int) <- call $(mkStatic 'sdictInt) nid (factorial' 5)+ liftIO $ putMVar clientDone ()++ takeMVar clientDone++testSeq :: TestTransport -> RemoteTable -> Assertion+testSeq TestTransport{..} rtable = do+ node <- newLocalNode testTransport rtable+ done <- newEmptyMVar+ runProcess node $ do+ us <- getSelfPid+ join . unClosure $ sendFac 5 us `seqCP` sendFac 6 us+ 120 :: Int <- expect+ 720 :: Int <- expect+ liftIO $ putMVar done ()+ takeMVar done++-- Test 'spawnSupervised'+--+-- Set up a supervisor, spawn a child, then have a third process monitor the+-- child. The supervisor then throws an exception, the child dies because it+-- was linked to the supervisor, and the third process notices that the child+-- dies.+testSpawnSupervised :: TestTransport -> RemoteTable -> Assertion+testSpawnSupervised TestTransport{..} rtable = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable+ [superPid, childPid] <- replicateM 2 $ newEmptyMVar+ thirdProcessDone <- newEmptyMVar++ forkProcess node1 $ do+ us <- getSelfPid+ liftIO $ putMVar superPid us+ (child, _ref) <- spawnSupervised (localNodeId node2) (waitClosure 1000000)+ liftIO $ do+ putMVar childPid child+ threadDelay 500000 -- Give the child a chance to link to us+ throw supervisorDeath++ forkProcess node2 $ do+ [super, child] <- liftIO $ mapM readMVar [superPid, childPid]+ ref <- monitor child+ ProcessMonitorNotification ref' pid' (DiedException e) <- expect+ True <- return $ ref' == ref+ && pid' == child+ && e == show (ProcessLinkException super (DiedException (show supervisorDeath)))+ liftIO $ putMVar thirdProcessDone ()++ takeMVar thirdProcessDone+ where+ supervisorDeath :: IOException+ supervisorDeath = userError "Supervisor died"++testSpawnInvalid :: TestTransport -> RemoteTable -> Assertion+testSpawnInvalid TestTransport{..} rtable = do+ node <- newLocalNode testTransport rtable+ done <- newEmptyMVar+ forkProcess node $ do+ (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 ()+ takeMVar done++testClosureExpect :: TestTransport -> RemoteTable -> Assertion+testClosureExpect TestTransport{..} rtable = do+ node <- newLocalNode testTransport rtable+ done <- newEmptyMVar+ runProcess node $ do+ nodeId <- getSelfNode+ us <- getSelfPid+ them <- spawn nodeId $ cpExpect $(mkStatic 'sdictInt) `bindCP` cpSend $(mkStatic 'sdictInt) us+ send them (1234 :: Int)+ (1234 :: Int) <- expect+ liftIO $ putMVar done ()+ takeMVar done++testSpawnChannel :: TestTransport -> RemoteTable -> Assertion+testSpawnChannel TestTransport{..} rtable = do+ done <- newEmptyMVar+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable++ forkProcess node1 $ do+ pingServer <- spawnChannel+ (sdictSendPort sdictUnit)+ (localNodeId node2)+ ($(mkClosure 'typedPingServer) ())+ (sendReply, receiveReply) <- newChan+ sendChan pingServer sendReply+ receiveChan receiveReply+ liftIO $ putMVar done ()++ takeMVar done++testTDict :: TestTransport -> RemoteTable -> Assertion+testTDict TestTransport{..} rtable = do+ 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 ()+ takeMVar done++testFib :: TestTransport -> RemoteTable -> Assertion+testFib TestTransport{..} rtable = do+ nodes <- replicateM 4 $ newLocalNode testTransport rtable+ done <- newEmptyMVar++ forkProcess (head nodes) $ do+ (sport, rport) <- newChan+ spawnLocal $ dfib (map localNodeId nodes, sport, 10)+ 55 <- receiveChan rport :: Process Integer+ liftIO $ putMVar done ()++ takeMVar done++testSpawnReconnect :: TestTransport -> RemoteTable -> Assertion+testSpawnReconnect testtrans@TestTransport{..} rtable = do+ [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable+ let nid1 = localNodeId node1+ nid2 = localNodeId node2+ done <- newEmptyMVar+ iv <- newMVar (0 :: Int)++ incr <- forkProcess node1 $ forever $ do+ () <- expect+ liftIO $ modifyMVar_ iv (return . (+ 1))++ forkProcess node2 $ do+ _pid1 <- spawn nid1 ($(mkClosure 'signal) incr)+ simulateNetworkFailure testtrans nid2 nid1+ _pid2 <- spawn nid1 ($(mkClosure 'signal) incr)+ _pid3 <- spawn nid1 ($(mkClosure 'signal) incr)++ liftIO $ threadDelay 100000++ count <- liftIO $ takeMVar iv+ True <- return $ count == 2 || count == 3 -- It depends on which message we get first in 'spawn'++ liftIO $ putMVar done ()++ takeMVar done++-- | 'spawn' used to ave a race condition which would be triggered if the+-- spawning process terminates immediately after spawning+testSpawnTerminate :: TestTransport -> RemoteTable -> Assertion+testSpawnTerminate TestTransport{..} rtable = do+ slave <- newLocalNode testTransport rtable+ master <- newLocalNode testTransport rtable+ masterDone <- newEmptyMVar++ runProcess master $ do+ us <- getSelfPid+ replicateM_ 1000 . spawnLocal . void . spawn (localNodeId slave) $ $(mkClosure 'signal) us+ replicateM_ 1000 $ (expect :: Process ())+ liftIO $ putMVar masterDone ()++ takeMVar masterDone++tests :: TestTransport -> IO [Test]+tests testtrans = do+ let rtable = __remoteTable . __remoteTableDecl $ initRemoteTable+ return+ [ testCase "Unclosure" (testUnclosure testtrans rtable)+ , testCase "Bind" (testBind testtrans rtable)+ , testCase "SendPureClosure" (testSendPureClosure testtrans rtable)+ , testCase "SendIOClosure" (testSendIOClosure testtrans rtable)+ , testCase "SendProcClosure" (testSendProcClosure testtrans rtable)+ , testCase "Spawn" (testSpawn testtrans rtable)+ , testCase "Call" (testCall testtrans rtable)+ , testCase "CallBind" (testCallBind testtrans rtable)+ , testCase "Seq" (testSeq testtrans rtable)+ , testCase "SpawnSupervised" (testSpawnSupervised testtrans rtable)+ , testCase "SpawnInvalid" (testSpawnInvalid testtrans rtable)+ , testCase "ClosureExpect" (testClosureExpect testtrans rtable)+ , testCase "SpawnChannel" (testSpawnChannel testtrans rtable)+ , testCase "TDict" (testTDict testtrans rtable)+ , testCase "Fib" (testFib testtrans rtable)+ , testCase "SpawnTerminate" (testSpawnTerminate testtrans rtable)+ , testCase "SpawnReconnect" (testSpawnReconnect testtrans rtable)+ ]
+ src/Control/Distributed/Process/Tests/Internal/Utils.hs view
@@ -0,0 +1,60 @@+-- | General testing support.+module Control.Distributed.Process.Tests.Internal.Utils where++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , takeMVar+ , putMVar+ )++import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Serializable ()+import Test.HUnit (Assertion)+import Test.HUnit.Base (assertBool)++-- these utilities have been cribbed from distributed-process-platform+-- we should really find a way to share them...++-- | A mutable cell containing a test result.+type TestResult a = MVar a++synchronisedAssertion :: Eq a+ => String+ -> LocalNode+ -> a+ -> (TestResult a -> Process ())+ -> MVar ()+ -> Assertion+synchronisedAssertion note localNode expected testProc lock = do+ result <- newEmptyMVar+ _ <- forkProcess localNode $ do+ acquire lock+ finally (testProc result)+ (release lock)+ assertComplete note result expected+ where acquire lock' = liftIO $ takeMVar lock'+ release lock' = liftIO $ putMVar lock' ()++delayedAssertion :: Eq a+ => String+ -> LocalNode+ -> a+ -> (TestResult a -> Process ())+ -> Assertion+delayedAssertion note localNode expected testProc = do+ result <- newEmptyMVar+ _ <- forkProcess localNode $ testProc result+ assertComplete note result expected++assertComplete :: Eq a => String -> MVar a -> a -> IO ()+assertComplete msg mv a = do+ b <- takeMVar mv+ assertBool msg (a == b)++stash :: TestResult a -> a -> Process ()+stash mvar x = liftIO $ putMVar mvar x
+ src/Control/Distributed/Process/Tests/Mx.hs view
@@ -0,0 +1,286 @@+{-# LANGUAGE DeriveGeneric #-}+module Control.Distributed.Process.Tests.Mx (tests) where++import Control.Distributed.Process.Tests.Internal.Utils+import Network.Transport.Test (TestTransport(..))++import Control.Concurrent (threadDelay)+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Management+ ( MxEvent(..)+ , MxAgentId(..)+ , mxAgent+ , mxSink+ , mxReady+ , mxReceive+ , mxDeactivate+ , liftMX+ , mxGetLocal+ , mxSetLocal+ , mxUpdateLocal+ , mxNotify+ , mxBroadcast+ , mxGetId+ , mxGet+ , mxSet+ , mxClear+ , mxPurgeTable+ , mxDropTable+ )+import Control.Monad (void)+import Data.Binary+import Data.List (find, sort)+import Data.Maybe (isJust)+import Data.Typeable+import GHC.Generics+#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, log)+#endif++import Test.Framework+ ( Test+ , testGroup+ )+import Test.Framework.Providers.HUnit (testCase)++data Publish = Publish+ deriving (Typeable, Generic, Eq)++instance Binary Publish where++testAgentBroadcast :: TestResult () -> Process ()+testAgentBroadcast result = do+ (resultSP, resultRP) <- newChan :: Process (SendPort (), ReceivePort ())++ publisher <- mxAgent (MxAgentId "publisher-agent") () [+ mxSink $ \() -> mxBroadcast Publish >> mxReady+ ]++ consumer <- mxAgent (MxAgentId "consumer-agent") () [+ mxSink $ \Publish -> (liftMX $ sendChan resultSP ()) >> mxReady+ ]++ 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++ kill publisher "finished"+ kill consumer "finished"++testAgentDualInput :: TestResult (Maybe Int) -> Process ()+testAgentDualInput result = do+ (sp, rp) <- newChan+ _ <- mxAgent (MxAgentId "sum-agent") (0 :: Int) [+ mxSink $ (\(i :: Int) -> do+ mxSetLocal . (+i) =<< mxGetLocal+ i' <- mxGetLocal+ if i' == 15+ then do mxGetLocal >>= liftMX . sendChan sp+ mxDeactivate "finished"+ else mxReady)+ ]++ mxNotify (1 :: Int)+ nsend "sum-agent" (3 :: Int)+ mxNotify (2 :: Int)+ nsend "sum-agent" (4 :: Int)+ mxNotify (5 :: Int)++ stash result =<< receiveChanTimeout 10000000 rp++testAgentPrioritisation :: TestResult [String] -> Process ()+testAgentPrioritisation result = do++ -- TODO: this isn't really testing how we /prioritise/ one source+ -- over another at all, but I've not yet figured out the right way+ -- to do so, since we're at the whim of the scheduler with regards+ -- the timeliness of nsend versus mxNotify anyway.++ let name = "prioritising-agent"+ (sp, rp) <- newChan+ void $ mxAgent (MxAgentId name) ["first"] [+ mxSink (\(s :: String) -> do+ mxUpdateLocal ((s:))+ st <- mxGetLocal+ case length st of+ n | n == 5 -> do liftMX $ sendChan sp st+ mxDeactivate "finished"+ _ -> mxReceive -- go to the mailbox+ )+ ]++ nsend name "second"+ mxNotify "third"+ mxNotify "fourth"+ nsend name "fifth"++ stash result . sort =<< receiveChan rp++testAgentMailboxHandling :: TestResult (Maybe ()) -> Process ()+testAgentMailboxHandling result = do+ (sp, rp) <- newChan+ agent <- mxAgent (MxAgentId "listener-agent") () [+ mxSink $ \() -> (liftMX $ sendChan sp ()) >> mxReady+ ]++ nsend "listener-agent" ()++ stash result =<< receiveChanTimeout 1000000 rp+ kill agent "finished"++testAgentEventHandling :: TestResult Bool -> Process ()+testAgentEventHandling result = do+ let initState = [] :: [MxEvent]+ agentPid <- mxAgent (MxAgentId "lifecycle-listener-agent") initState [+ (mxSink $ \ev -> do+ st <- mxGetLocal+ let act =+ case ev of+ (MxSpawned _) -> mxSetLocal (ev:st)+ (MxProcessDied _ _) -> mxSetLocal (ev:st)+ _ -> return ()+ act >> mxReady),+ (mxSink $ \(ev, sp :: SendPort Bool) -> do+ st <- mxGetLocal+ let found =+ case ev of+ MxSpawned p ->+ isJust $ find (\ev' ->+ case ev' of+ (MxSpawned p') -> p' == p+ _ -> False) st+ MxProcessDied p r ->+ isJust $ find (\ev' ->+ case ev' of+ (MxProcessDied p' r') -> p' == p && r == r'+ _ -> False) st+ _ -> False+ liftMX $ sendChan sp found+ mxReady)+ ]++ _ <- monitor agentPid+ (sp, rp) <- newChan+ pid <- spawnLocal $ sendChan sp ()+ () <- receiveChan rp++ -- By waiting for a monitor notification, we have a+ -- higher probably that the agent has seen+ monitor pid+ receiveWait [ match (\(ProcessMonitorNotification _ _ _) -> return ()) ]++ (replyTo, reply) <- newChan :: Process (SendPort Bool, ReceivePort Bool)+ mxNotify (MxSpawned pid, replyTo)+ mxNotify (MxProcessDied pid DiedNormal, replyTo)++ seenAlive <- receiveChan reply+ seenDead <- receiveChan reply++ stash result $ seenAlive && seenDead++testAgentPublication :: TestResult (Maybe Publish) -> Process ()+testAgentPublication result = do+ (syncChan, rp) <- newChan+ let ourId = MxAgentId "publication-agent"+ agentPid <- mxAgent ourId () [+ mxSink $ \() -> do+ selfId <- mxGetId+ liftMX $ mxSet selfId "publish" Publish >> sendChan syncChan ()+ mxReady+ ]++ mxNotify ()+ () <- receiveChan rp++ stash result =<< mxGet ourId "publish"++ mref <- monitor agentPid+ kill agentPid "finished"+ receiveWait [+ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)+ ((\_ -> return ()))+ ]++testAgentTableClear :: Int -> TestResult (Maybe Int, Maybe Int) -> Process ()+testAgentTableClear val result =+ let tId = (MxAgentId "agent-1")+ tKey = "key-1" in do+ mxSet tId tKey val+ get1 <- mxGet tId tKey+ mxClear tId tKey+ get2 <- mxGet tId tKey+ stash result (get1, get2)++testAgentTablePurge :: TestResult (Maybe Int) -> Process ()+testAgentTablePurge result =+ let tId = MxAgentId "agent-2"+ tKey = "key-2" in do+ mxSet tId tKey (12345 :: Int)+ mxPurgeTable tId+ stash result =<< mxGet tId tKey++testAgentTableDelete :: Int+ -> TestResult (Maybe Int, Maybe Int, Maybe Int)+ -> Process ()+testAgentTableDelete val result =+ let tId = (MxAgentId "agent-3")+ tKey = "key-3" in do+ mxSet tId tKey val+ get1 <- mxGet tId tKey+ mxDropTable tId+ get2 <- mxGet tId tKey+ mxSet tId tKey val+ get3 <- mxGet tId tKey+ stash result (get1, get2, get3)++tests :: TestTransport -> IO [Test]+tests TestTransport{..} = do+ node1 <- 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 "Mx Global Properties" [+ testCase "Global Property Publication"+ (delayedAssertion+ "expected (Just Publish), but no table entry was found"+ node1 (Just Publish) testAgentPublication)+ , testCase "Clearing Global Properties"+ (delayedAssertion+ "expected (Just 1024, Nothing): invalid table entry found!"+ node1 (Just 1024, Nothing) (testAgentTableClear 1024))+ , testCase "Purging Global Tables"+ (delayedAssertion+ "expected Nothing, but a table entry was found"+ node1 Nothing testAgentTablePurge)+ , testCase "Deleting and (Re)Creating Global Tables"+ (delayedAssertion+ "expected (Just 15, Nothing, Just 15): invalid table entry found!"+ node1 (Just 15, Nothing, Just 15) (testAgentTableDelete 15))+ -- Wait for other tests to finish.+ , testCase "Wait" $+ threadDelay 100000+ ]]
+ src/Control/Distributed/Process/Tests/Receive.hs view
@@ -0,0 +1,158 @@+{-# LANGUAGE TemplateHaskell, DeriveDataTypeable #-}+{-# OPTIONS_GHC -Wall #-}++-- | XXX test doesn't work, because failure exceptions don't get propagated. The+-- test always claims to succeed, even if it failed.++module Control.Distributed.Process.Tests.Receive (tests) where++import Network.Transport.Test (TestTransport(..))++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.Providers.HUnit (testCase)++-- Tests:++-- 1. 2 matchChans, receive on each one+-- 2. matchChan/matchIf, receive on each one+-- 3. matchIf/matchChan, receive on each one+-- 4. matchIf/matchChan/matchIf, receive on each one++recTest1 :: ReceivePort ()+ -> SendPort String+ -> ReceivePort String -> ReceivePort String+ -> Process ()+recTest1 wait sync r1 r2 = do+ forever $ do+ receiveChan wait+ r <- receiveWait+ [ matchChan r1 $ \s -> return ("received1 " ++ s)+ , matchChan r2 $ \s -> return ("received2 " ++ s)+ ]+ sendChan sync r++recTest2 :: ReceivePort ()+ -> SendPort String+ -> ReceivePort String -> ReceivePort String+ -> Process ()+recTest2 wait sync r1 r2 = do+ forever $ do+ receiveChan wait+ r <- receiveWait+ [ matchChan r1 $ \s -> return ("received1 " ++ s)+ , matchIf (== "foo") $ \s -> return ("received2 " ++ s)+ ]+ sendChan sync r++recTest3 :: ReceivePort ()+ -> SendPort String+ -> ReceivePort String -> ReceivePort String+ -> Process ()+recTest3 wait sync r1 r2 = do+ forever $ do+ receiveChan wait+ r <- receiveWait+ [ matchIf (== "foo") $ \s -> return ("received1 " ++ s)+ , matchChan r1 $ \s -> return ("received2 " ++ s)+ ]+ sendChan sync r++recTest4 :: ReceivePort ()+ -> SendPort String+ -> ReceivePort String -> ReceivePort String+ -> Process ()+recTest4 wait sync r1 r2 = do+ forever $ do+ receiveChan wait+ r <- receiveWait+ [ matchIf (== "foo") $ \s -> return ("received1 " ++ s)+ , matchChan r1 $ \s -> return ("received2 " ++ s)+ , matchIf (== "bar") $ \s -> return ("received3 " ++ s)+ ]+ sendChan sync r++master :: Process ()+master = do+ (waits,waitr) <- newChan+ (syncs,syncr) <- newChan+ let go expect = do+ sendChan waits ()+ r <- receiveChan syncr+ liftIO $ print (r,expect, r == expect)+ liftIO $ r @?= expect++ liftIO $ putStrLn "---- Test 1 ----"+ (s1,r1) <- newChan+ (s2,r2) <- newChan+ p <- spawnLocal (recTest1 waitr syncs r1 r2)++ sendChan s1 "a" >> go "received1 a"+ sendChan s2 "b" >> go "received2 b"+ sendChan s1 "a" >> sendChan s2 "b" >> go "received1 a"+ go "received2 b"++ kill p "BANG"++ liftIO $ putStrLn "\n---- Test 2 ----"+ (s1,r1) <- newChan+ (s2,r2) <- newChan+ p <- spawnLocal (recTest2 waitr syncs r1 r2)++ sendChan s1 "a" >> go "received1 a"+ send p "foo" >> go "received2 foo"+ sendChan s1 "a" >> send p "foo" >> go "received1 a"+ sendChan s1 "a" >> send p "bar" >> go "received1 a"+ go "received2 foo"++ kill p "BANG"++ liftIO $ putStrLn "\n---- Test 3 ----"+ (s1,r1) <- newChan+ (s2,r2) <- newChan+ p <- spawnLocal (recTest3 waitr syncs r1 r2)++ sendChan s1 "a" >> go "received2 a"+ send p "foo" >> go "received1 foo"+ sendChan s1 "a" >> send p "foo" >> go "received1 foo"+ sendChan s1 "a" >> send p "bar" >> go "received2 a"+ go "received2 a"++ kill p "BANG"++ liftIO $ putStrLn "\n---- Test 4 ----"+ (s1,r1) <- newChan+ (s2,r2) <- newChan+ p <- spawnLocal (recTest4 waitr syncs r1 r2)++ sendChan s1 "a" >> go "received2 a"+ send p "foo" >> go "received1 foo"+ send p "bar" >> go "received3 bar"+ sendChan s1 "a" >> send p "foo" >> go "received1 foo"+ send p "bar" >> go "received2 a"+ send p "foo" >> go "received1 foo" >> go "received3 bar"++ kill p "BANG"++ terminate++testReceive :: Transport -> RemoteTable -> Assertion+testReceive transport rtable = do+ node <- newLocalNode transport rtable+ runProcess node $ master++tests :: TestTransport -> IO [Test]+tests TestTransport{..} = do+ let rtable = initRemoteTable+ return+ [ testCase "testReceive" (testReceive testTransport rtable) ]
+ src/Control/Distributed/Process/Tests/Stats.hs view
@@ -0,0 +1,125 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Control.Distributed.Process.Tests.Stats (tests) where++import Control.Distributed.Process.Tests.Internal.Utils+import Network.Transport.Test (TestTransport(..))++import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , putMVar+ , takeMVar+ )+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Data.Binary ()+import Data.Typeable ()++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch)+#endif++import Test.Framework+ ( Test+ , testGroup+ )+import Test.HUnit (Assertion)+import Test.Framework.Providers.HUnit (testCase)++testLocalDeadProcessInfo :: TestResult (Maybe ProcessInfo) -> Process ()+testLocalDeadProcessInfo result = do+ pid <- spawnLocal $ do "finish" <- expect; return ()+ mref <- monitor pid+ send pid "finish"+ _ <- receiveWait [+ matchIf (\(ProcessMonitorNotification ref' pid' r) ->+ ref' == mref && pid' == pid && r == DiedNormal)+ (\p -> return p)+ ]+ getProcessInfo pid >>= stash result++testLocalLiveProcessInfo :: TestResult Bool -> Process ()+testLocalLiveProcessInfo result = do+ self <- getSelfPid+ node <- getSelfNode+ register "foobar" self++ mon <- liftIO $ newEmptyMVar+ -- TODO: we can't get the mailbox's length+ -- mapM (send self) ["hello", "there", "mr", "process"]+ pid <- spawnLocal $ do+ link self+ mRef <- monitor self+ stash mon mRef+ "die" <- expect+ return ()++ monRef <- liftIO $ takeMVar mon++ mpInfo <- getProcessInfo self+ case mpInfo of+ Nothing -> stash result False+ Just p -> verifyPInfo p pid monRef node+ where verifyPInfo :: ProcessInfo+ -> ProcessId+ -> MonitorRef+ -> NodeId+ -> Process ()+ verifyPInfo pInfo pid mref node =+ stash result $ infoNode pInfo == node &&+ infoLinks pInfo == [pid] &&+ infoMonitors pInfo == [(pid, mref)] &&+-- infoMessageQueueLength pInfo == Just 4 &&+ infoRegisteredNames pInfo == ["foobar"]++testRemoteLiveProcessInfo :: TestTransport -> LocalNode -> Assertion+testRemoteLiveProcessInfo TestTransport{..} node1 = do+ serverAddr <- liftIO $ newEmptyMVar :: IO (MVar ProcessId)+ liftIO $ launchRemote serverAddr+ serverPid <- liftIO $ takeMVar serverAddr+ withActiveRemote node1 $ \result -> do+ self <- getSelfPid+ link serverPid+ -- 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+ pInfo <- getProcessInfo serverPid+ stash result $ pInfo /= Nothing+ where+ launchRemote :: MVar ProcessId -> IO ()+ launchRemote locMV = do+ node2 <- liftIO $ newLocalNode testTransport initRemoteTable+ _ <- liftIO $ forkProcess node2 $ do+ self <- getSelfPid+ liftIO $ putMVar locMV self+ _ <- receiveWait [+ match (\(pid, "ping") -> send pid "pong")+ ]+ "stop" <- expect+ return ()+ return ()++ withActiveRemote :: LocalNode+ -> ((TestResult Bool -> Process ()) -> Assertion)+ withActiveRemote n = do+ a <- delayedAssertion "getProcessInfo remotePid failed" n True+ return a++tests :: TestTransport -> IO [Test]+tests testtrans@TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ return [+ testGroup "Process Info" [+ testCase "testLocalDeadProcessInfo"+ (delayedAssertion+ "expected dead process-info to be ProcessInfoNone"+ node1 (Nothing) testLocalDeadProcessInfo)+ , testCase "testLocalLiveProcessInfo"+ (delayedAssertion+ "expected process-info to be correctly populated"+ node1 True testLocalLiveProcessInfo)+ , testCase "testRemoveLiveProcessInfo"+ (testRemoteLiveProcessInfo testtrans node1)+ ] ]
+ src/Control/Distributed/Process/Tests/Tracing.hs view
@@ -0,0 +1,326 @@+{-# LANGUAGE DeriveDataTypeable #-}+{-# OPTIONS_GHC -fno-warn-orphans #-}+module Control.Distributed.Process.Tests.Tracing (tests) where++import Control.Distributed.Process.Tests.Internal.Utils+import Network.Transport.Test (TestTransport(..))++import Control.Concurrent (threadDelay)+import Control.Concurrent.MVar+ ( MVar+ , newEmptyMVar+ , newMVar+ , putMVar+ , takeMVar+ )+import Control.Distributed.Process+import Control.Distributed.Process.Node+import Control.Distributed.Process.Debug+import Control.Distributed.Process.Management+ ( MxEvent(..)+ )++#if ! MIN_VERSION_base(4,6,0)+import Prelude hiding (catch, log)+#endif++import Test.Framework+ ( Test+ , testGroup+ )+import Test.Framework.Providers.HUnit (testCase)++testSpawnTracing :: TestResult Bool -> Process ()+testSpawnTracing result = do+ setTraceFlags defaultTraceFlags {+ traceSpawned = (Just TraceAll)+ , traceDied = (Just TraceAll)+ }++ evSpawned <- liftIO $ newEmptyMVar+ evDied <- liftIO $ newEmptyMVar+ tracer <- startTracer $ \ev -> do+ case ev of+ (MxSpawned p) -> liftIO $ putMVar evSpawned p+ (MxProcessDied p r) -> liftIO $ putMVar evDied (p, r)+ _ -> return ()++ (sp, rp) <- newChan+ pid <- spawnLocal $ sendChan sp ()+ () <- receiveChan rp++ tracedAlive <- liftIO $ takeMVar evSpawned+ (tracedDead, tracedReason) <- liftIO $ takeMVar evDied++ mref <- monitor tracer+ stopTracer -- this is asynchronous, so we need to wait...+ receiveWait [+ matchIf (\(ProcessMonitorNotification ref _ _) -> ref == mref)+ ((\_ -> return ()))+ ]+ setTraceFlags defaultTraceFlags+ stash result (tracedAlive == pid &&+ tracedDead == pid &&+ tracedReason == DiedNormal)++testTraceRecvExplicitPid :: TestResult Bool -> Process ()+testTraceRecvExplicitPid result = do+ res <- liftIO $ newEmptyMVar+ pid <- spawnLocal $ do+ self <- getSelfPid+ expect >>= (flip sendChan) self+ withFlags defaultTraceFlags {+ traceRecv = traceOnly [pid]+ } $ do+ withTracer+ (\ev ->+ case ev of+ (MxReceived pid' _) -> stash res (pid == pid')+ _ -> return ()) $ do+ (sp, rp) <- newChan+ send pid sp+ p <- receiveChan rp+ res' <- liftIO $ takeMVar res+ stash result (res' && (p == pid))+ return ()++testTraceRecvNamedPid :: TestResult Bool -> Process ()+testTraceRecvNamedPid result = do+ res <- liftIO $ newEmptyMVar+ pid <- spawnLocal $ do+ self <- getSelfPid+ register "foobar" self+ expect >>= (flip sendChan) self+ withFlags defaultTraceFlags {+ traceRecv = traceOnly ["foobar"]+ } $ do+ withTracer+ (\ev ->+ case ev of+ (MxReceived pid' _) -> stash res (pid == pid')+ _ -> return ()) $ do+ (sp, rp) <- newChan+ send pid sp+ p <- receiveChan rp+ res' <- liftIO $ takeMVar res+ stash result (res' && (p == pid))+ return ()++testTraceSending :: TestResult Bool -> Process ()+testTraceSending result = do+ pid <- spawnLocal $ (expect :: Process String) >> return ()+ self <- getSelfPid+ res <- liftIO $ newEmptyMVar+ withFlags defaultTraceFlags { traceSend = traceOn } $ do+ withTracer+ (\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)+ _ ->+ return ()) $ do+ send pid "hello there"+ res' <- liftIO $ takeMVar res+ stash result res'++testTraceRegistration :: TestResult Bool -> Process ()+testTraceRegistration result = do+ (sp, rp) <- newChan+ pid <- spawnLocal $ do+ self <- getSelfPid+ () <- expect+ register "foobar" self+ sendChan sp ()+ () <- expect+ return ()+ res <- liftIO $ newEmptyMVar+ withFlags defaultTraceFlags { traceRegistered = traceOn } $ do+ withTracer+ (\ev ->+ case ev of+ MxRegistered p s ->+ stash res (p == pid && s == "foobar")+ _ ->+ return ()) $ do+ _ <- monitor pid+ send pid ()+ () <- receiveChan rp+ send pid ()+ receiveWait [+ match (\(ProcessMonitorNotification _ _ _) -> return ())+ ]+ res' <- liftIO $ takeMVar res+ stash result res'++testTraceUnRegistration :: TestResult Bool -> Process ()+testTraceUnRegistration result = do+ pid <- spawnLocal $ do+ () <- expect+ unregister "foobar"+ () <- expect+ return ()+ register "foobar" pid+ res <- liftIO $ newEmptyMVar+ withFlags defaultTraceFlags { traceUnregistered = traceOn } $ do+ withTracer+ (\ev ->+ case ev of+ MxUnRegistered p n -> do+ stash res (p == pid && n == "foobar")+ send pid ()+ _ ->+ return ()) $ do+ mref <- monitor pid+ send pid ()+ receiveWait [+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')+ (\_ -> return ())+ ]+ res' <- liftIO $ takeMVar res+ stash result res'++testTraceLayering :: TestResult () -> Process ()+testTraceLayering result = do+ pid <- spawnLocal $ do+ getSelfPid >>= register "foobar"+ () <- expect+ traceMessage ("traceMsg", 123 :: Int)+ return ()+ withFlags defaultTraceFlags {+ traceDied = traceOnly [pid]+ , traceRecv = traceOnly ["foobar"]+ } $ doTest pid result+ return ()+ where+ doTest :: ProcessId -> MVar () -> Process ()+ doTest pid result' = do+ -- TODO: this is pretty gross, even for a test case+ died <- liftIO $ newEmptyMVar+ withTracer+ (\ev ->+ case ev of+ MxProcessDied _ _ -> liftIO $ putMVar died ()+ _ -> return ())+ ( do {+ recv <- liftIO $ newEmptyMVar+ ; withTracer+ (\ev' ->+ case ev' of+ MxReceived _ _ -> liftIO $ putMVar recv ()+ _ -> return ())+ ( do {+ user <- liftIO $ newEmptyMVar+ ; withTracer+ (\ev'' ->+ case ev'' of+ MxUser _ -> liftIO $ putMVar user ()+ _ -> return ())+ (send pid () >> (liftIO $ takeMVar user))+ ; liftIO $ takeMVar recv+ })+ ; liftIO $ takeMVar died+ })+ liftIO $ putMVar result' ()++testRemoteTraceRelay :: TestTransport -> TestResult Bool -> Process ()+testRemoteTraceRelay TestTransport{..} result =+ let flags = defaultTraceFlags { traceSpawned = traceOn }+ in do+ node2 <- liftIO $ newLocalNode testTransport initRemoteTable+ mvNid <- liftIO $ newEmptyMVar++ -- As well as needing node2's NodeId, we want to+ -- redirect all its logs back here, to avoid generating+ -- garbage on stderr for the duration of the test run.+ -- 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 ())++ nid <- liftIO $ takeMVar mvNid+ mref <- monitor pid+ observedPid <- liftIO $ newEmptyMVar+ spawnedPid <- liftIO $ newEmptyMVar+ setTraceFlagsRemote flags nid++ withFlags defaultTraceFlags { traceSpawned = traceOn } $ do+ withTracer+ (\ev ->+ case ev of+ MxSpawned p -> stash observedPid p >> send pid ()+ _ -> return ()) $ do+ relayPid <- startTraceRelay nid+ liftIO $ threadDelay 1000000+ p <- liftIO $ forkProcess node2 $ do+ expectTimeout 1000000 :: Process (Maybe ())+ return ()+ stash spawnedPid p++ -- Now we wait for (the outer) pid to exit. This won't happen until+ -- our tracer has seen the trace event for `p' and sent `p' the+ -- message it's waiting for prior to exiting+ receiveWait [+ matchIf (\(ProcessMonitorNotification mref' _ _) -> mref == mref')+ (\_ -> return ())+ ]++ relayRef <- monitor relayPid+ kill relayPid "stop"+ receiveWait [+ matchIf (\(ProcessMonitorNotification rref' _ _) -> rref' == relayRef)+ (\_ -> return ())+ ]+ observed <- liftIO $ takeMVar observedPid+ expected <- liftIO $ takeMVar spawnedPid+ stash result (observed == expected)+ -- and just to be polite...+ liftIO $ closeLocalNode node2++tests :: TestTransport -> IO [Test]+tests testtrans@TestTransport{..} = do+ node1 <- newLocalNode testTransport initRemoteTable+ -- if we execute the test cases in parallel, the+ -- various tracers will race with one another and+ -- we'll get garbage results (or worse, deadlocks)+ lock <- liftIO $ newMVar ()+ return [+ testGroup "Tracing" [+ testCase "Spawn Tracing"+ (synchronisedAssertion+ "expected dead process-info to be ProcessInfoNone"+ node1 True testSpawnTracing lock)+ , testCase "Recv Tracing (Explicit Pid)"+ (synchronisedAssertion+ "expected a recv trace for the supplied pid"+ node1 True testTraceRecvExplicitPid lock)+ , testCase "Recv Tracing (Named Pid)"+ (synchronisedAssertion+ "expected a recv trace for the process registered as 'foobar'"+ node1 True testTraceRecvNamedPid lock)+ , testCase "Trace Send(er)"+ (synchronisedAssertion+ "expected a 'send' trace with the requisite fields set"+ node1 True testTraceSending lock)+ , testCase "Trace Registration"+ (synchronisedAssertion+ "expected a 'registered' trace"+ node1 True testTraceRegistration lock)+ , testCase "Trace Unregistration"+ (synchronisedAssertion+ "expected an 'unregistered' trace"+ node1 True testTraceUnRegistration lock)+ , testCase "Trace Layering"+ (synchronisedAssertion+ "expected blah"+ node1 () testTraceLayering lock)+ , testCase "Remote Trace Relay"+ (synchronisedAssertion+ "expected blah"+ node1 True (testRemoteTraceRelay testtrans) lock)+ ] ]
+ src/Network/Transport/Test.hs view
@@ -0,0 +1,12 @@+module Network.Transport.Test where++import qualified Network.Transport as NT+import Data.Typeable (Typeable)++-- | Extra operations required of transports for the purposes of testing.+data TestTransport = TestTransport+ { -- | The transport to use for testing.+ testTransport :: NT.Transport+ -- | IO action to perform to simulate losing a connection.+ , testBreakConnection :: NT.EndPointAddress -> NT.EndPointAddress -> IO ()+ } deriving (Typeable)
+ tests/runTCP.hs view
@@ -0,0 +1,28 @@+-- Run tests using the TCP transport.++module Main where++import TEST_SUITE_MODULE (tests)++import Network.Transport.Test (TestTransport(..))+import Network.Socket (sClose)+import Network.Transport.TCP+ ( createTransportExposeInternals+ , TransportInternals(socketBetween)+ , defaultTCPParameters+ )+import Test.Framework (defaultMain)++import Control.Concurrent (threadDelay)++main :: IO ()+main = do+ Right (transport, internals) <-+ createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters+ defaultMain =<< tests TestTransport+ { testTransport = transport+ , testBreakConnection = \addr1 addr2 -> do+ sock <- socketBetween internals addr1 addr2+ sClose sock+ threadDelay 10000+ }