diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. + diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/distributed-process-tests.cabal b/distributed-process-tests.cabal new file mode 100644 --- /dev/null +++ b/distributed-process-tests.cabal @@ -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 diff --git a/src/Control/Distributed/Process/Tests/CH.hs b/src/Control/Distributed/Process/Tests/CH.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/CH.hs @@ -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) + ] + ] diff --git a/src/Control/Distributed/Process/Tests/Closure.hs b/src/Control/Distributed/Process/Tests/Closure.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/Closure.hs @@ -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) + ] diff --git a/src/Control/Distributed/Process/Tests/Internal/Utils.hs b/src/Control/Distributed/Process/Tests/Internal/Utils.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/Internal/Utils.hs @@ -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 diff --git a/src/Control/Distributed/Process/Tests/Mx.hs b/src/Control/Distributed/Process/Tests/Mx.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/Mx.hs @@ -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 + ]] diff --git a/src/Control/Distributed/Process/Tests/Receive.hs b/src/Control/Distributed/Process/Tests/Receive.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/Receive.hs @@ -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) ] diff --git a/src/Control/Distributed/Process/Tests/Stats.hs b/src/Control/Distributed/Process/Tests/Stats.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/Stats.hs @@ -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) + ] ] diff --git a/src/Control/Distributed/Process/Tests/Tracing.hs b/src/Control/Distributed/Process/Tests/Tracing.hs new file mode 100644 --- /dev/null +++ b/src/Control/Distributed/Process/Tests/Tracing.hs @@ -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) + ] ] diff --git a/src/Network/Transport/Test.hs b/src/Network/Transport/Test.hs new file mode 100644 --- /dev/null +++ b/src/Network/Transport/Test.hs @@ -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) diff --git a/tests/runTCP.hs b/tests/runTCP.hs new file mode 100644 --- /dev/null +++ b/tests/runTCP.hs @@ -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 + }