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