diff --git a/distributed-process-tests.cabal b/distributed-process-tests.cabal --- a/distributed-process-tests.cabal +++ b/distributed-process-tests.cabal @@ -1,5 +1,5 @@ name: distributed-process-tests -version: 0.4.3.3 +version: 0.4.4 synopsis: Tests and test support tools for distributed-process. homepage: http://github.com/haskell-distributed/distributed-process-tests description: Tests and test suite for Cloud Haskell libraries. @@ -12,6 +12,10 @@ build-type: Simple cabal-version: >=1.8 +flag tcp + Description: build and run TCP tests + Default: False + library exposed-modules: Network.Transport.Test Control.Distributed.Process.Tests.CH @@ -25,13 +29,14 @@ ansi-terminal >= 0.5 && < 0.7, binary >= 0.5 && < 0.8, bytestring >= 0.9 && < 0.11, - distributed-process >= 0.5.3 && < 0.6, + distributed-process >= 0.6.0 && < 0.7, distributed-static, HUnit >= 1.2 && < 1.4, network-transport >= 0.4.1.0 && < 0.5, network >= 2.5 && < 2.7, random >= 1.0 && < 1.2, 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 @@ -50,29 +55,47 @@ if impl(ghc <= 7.4.2) Build-Depends: ghc-prim == 0.2.0.0 -Test-Suite TestCH +Test-Suite TestCHInMemory Type: exitcode-stdio-1.0 - Main-Is: runTCP.hs + Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, - network-transport-tcp >= 0.3 && < 0.5, + 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 HS-Source-Dirs: tests -Test-Suite TestClosure +Test-Suite TestCHInTCP Type: exitcode-stdio-1.0 Main-Is: runTCP.hs + CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.CH + if flag(tcp) + Build-Depends: base >= 4.4 && < 5, + distributed-process-tests, + network >= 2.3 && < 2.7, + network-transport >= 0.4.1.0 && < 0.5, + network-transport-tcp >= 0.5 && < 0.6, + 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 + HS-Source-Dirs: tests + + +Test-Suite TestClosure + 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, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, - network-transport-tcp >= 0.3 && < 0.5, + 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 @@ -80,13 +103,13 @@ Test-Suite TestStats Type: exitcode-stdio-1.0 - Main-Is: runTCP.hs + Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Stats Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, - network-transport-tcp >= 0.3 && < 0.5, + 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 @@ -95,13 +118,13 @@ Test-Suite TestMx Type: exitcode-stdio-1.0 - Main-Is: runTCP.hs + Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Mx Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, - network-transport-tcp >= 0.3 && < 0.5, + 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 @@ -109,13 +132,13 @@ Test-Suite TestTracing Type: exitcode-stdio-1.0 - Main-Is: runTCP.hs + Main-Is: runInMemory.hs CPP-Options: -DTEST_SUITE_MODULE=Control.Distributed.Process.Tests.Tracing Build-Depends: base >= 4.4 && < 5, distributed-process-tests, network >= 2.3 && < 2.7, network-transport >= 0.4.1.0 && < 0.5, - network-transport-tcp >= 0.3 && < 0.5, + 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 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 @@ -9,7 +9,13 @@ import Data.Binary (Binary(..)) import Data.Typeable (Typeable) import Data.Foldable (forM_) -import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId) +import Data.Function (fix) +import Data.IORef + ( readIORef + , writeIORef + , newIORef + ) +import Control.Concurrent (forkIO, threadDelay, myThreadId, throwTo, ThreadId, yield) import Control.Concurrent.MVar ( MVar , newEmptyMVar @@ -17,17 +23,18 @@ , takeMVar , readMVar ) -import Control.Monad (replicateM_, replicateM, forever, void, unless) -import Control.Exception (SomeException, throwIO) +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 qualified Network.Transport as NT (closeEndPoint) +import qualified Network.Transport as NT (closeEndPoint, EndPointAddress) import Control.Distributed.Process import Control.Distributed.Process.Internal.Types ( NodeId(nodeAddress) , LocalNode(localEndPoint) , ProcessExitException(..) , nullProcessId + , createUnencodedMessage ) import Control.Distributed.Process.Node import Control.Distributed.Process.Serializable (Serializable) @@ -199,7 +206,7 @@ forkIO $ do localNode <- newLocalNode testTransport initRemoteTable - addr <- forkProcess localNode . liftIO $ threadDelay 1000000 + addr <- forkProcess localNode expect closeLocalNode localNode putMVar deadProcess addr @@ -302,19 +309,25 @@ testMonitorDisconnect :: TestTransport -> Bool -> Bool -> Assertion testMonitorDisconnect TestTransport{..} mOrL un = do processAddr <- newEmptyMVar + processAddr2 <- newEmptyMVar monitorSetup <- newEmptyMVar done <- newEmptyMVar forkIO $ do localNode <- newLocalNode testTransport initRemoteTable - addr <- forkProcess localNode . liftIO $ threadDelay 1000000 + addr <- forkProcess localNode $ expect + addr2 <- forkProcess localNode $ return () putMVar processAddr addr readMVar monitorSetup NT.closeEndPoint (localEndPoint localNode) + putMVar processAddr2 addr2 forkIO $ do localNode <- newLocalNode testTransport initRemoteTable theirAddr <- readMVar processAddr + forkProcess localNode $ do + lc <- liftIO $ readMVar processAddr2 + send lc () runProcess localNode $ do monitorTestProcess theirAddr mOrL un DiedDisconnect (Just monitorSetup) done @@ -398,16 +411,14 @@ testTimeout0 TestTransport{..} = do serverAddr <- newEmptyMVar clientDone <- newEmptyMVar - messagesSent <- newEmptyMVar forkIO $ do localNode <- newLocalNode testTransport initRemoteTable addr <- forkProcess localNode $ do - liftIO $ readMVar messagesSent >> threadDelay 1000000 -- Variation on the venerable ping server which uses a zero timeout - -- Since we wait for all messages to be sent before doing this receive, - -- we should nevertheless find the right message immediately - Just partner <- receiveTimeout 0 [match (\(Pong partner) -> return partner)] + partner <- fix $ \loop -> + receiveTimeout 0 [match (\(Pong partner) -> return partner)] + >>= maybe (liftIO (threadDelay 100000) >> loop) return self <- getSelfPid send partner (Ping self) putMVar serverAddr addr @@ -421,7 +432,6 @@ -- 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 () @@ -583,17 +593,22 @@ testMonitorLiveNode TestTransport{..} = do [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable ready <- newEmptyMVar + readyr <- newEmptyMVar done <- newEmptyMVar + p <- forkProcess node1 $ return () forkProcess node2 $ do ref <- monitorNode (localNodeId node1) liftIO $ putMVar ready () + liftIO $ takeMVar readyr + send p () NodeMonitorNotification ref' nid _ <- expect True <- return $ ref == ref' && nid == localNodeId node1 liftIO $ putMVar done () takeMVar ready closeLocalNode node1 + putMVar readyr () takeMVar done @@ -624,6 +639,7 @@ done <- newEmptyMVar pingServer <- forkProcess node ping + deadProcess <- forkProcess node (return ()) runProcess node $ do register "ping" pingServer @@ -633,10 +649,34 @@ 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" liftIO $ putMVar done () takeMVar done +testRegistryRemoteProcess :: TestTransport -> Assertion +testRegistryRemoteProcess TestTransport{..} = do + node1 <- newLocalNode testTransport initRemoteTable + node2 <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pingServer <- forkProcess node1 ping + + runProcess node2 $ 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 @@ -644,13 +684,15 @@ done <- newEmptyMVar pingServer <- forkProcess node1 ping + deadProcess <- forkProcess node1 (return ()) runProcess node2 $ do let nid1 = localNodeId node1 registerRemoteAsync nid1 "ping" pingServer receiveWait [ - matchIf (\(RegisterReply label' _) -> "ping" == label') - (\(RegisterReply _ _) -> return ()) ] + matchIf (\(RegisterReply label' _ (Just pid)) -> + "ping" == label' && pid == pingServer) + (\(RegisterReply _ _ _) -> return ()) ] Just pid <- whereisRemote nid1 "ping" True <- return $ pingServer == pid @@ -658,10 +700,51 @@ nsendRemote nid1 "ping" (Pong us) Ping pid' <- expect True <- return $ pingServer == pid' + + -- test that if process was not registered Nothing is returned + -- in owner field. + registerRemoteAsync nid1 "dead" deadProcess + receiveWait [ + matchIf (\(RegisterReply label' False Nothing) -> "dead" == label') + (\(RegisterReply _ _ _) -> return ()) ] + registerRemoteAsync nid1 "ping" deadProcess + receiveWait [ + matchIf (\(RegisterReply label' False (Just pid)) -> + "ping" == label' && pid == pingServer) + (\(RegisterReply _ _ _) -> return ()) ] + unregisterRemoteAsync nid1 "dead" + receiveWait [ + matchIf (\(RegisterReply label' False Nothing) -> + "dead" == label' && pid == pingServer) + (\(RegisterReply _ _ _) -> return ()) ] liftIO $ putMVar done () takeMVar done +testRemoteRegistryRemoteProcess :: TestTransport -> Assertion +testRemoteRegistryRemoteProcess TestTransport{..} = do + node1 <- newLocalNode testTransport initRemoteTable + node2 <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + pingServer <- forkProcess node2 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 @@ -684,11 +767,25 @@ takeMVar done +testSpawnAsyncStrictness :: TestTransport -> Assertion +testSpawnAsyncStrictness TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + done <- newEmptyMVar + + runProcess node $ do + here <-getSelfNode + + ev <- try $ spawnAsync here (error "boom") + liftIO $ case ev of + Right _ -> putMVar done (error "Exception didn't fire") + Left (_::SomeException) -> putMVar done (return ()) + + join $ 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 @@ -709,8 +806,9 @@ send them "message 1" >> liftIO (threadDelay 100000) -- Simulate network failure - liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) + liftIO $ syncBreakConnection testBreakConnection node1 node2 + -- Should not arrive send them "message 2" @@ -727,26 +825,26 @@ us <- getSelfPid registerRemoteAsync nid1 "a" us -- registerRemote is asynchronous receiveWait [ - matchIf (\(RegisterReply label' _) -> "a" == label') - (\(RegisterReply _ _) -> return ()) ] + matchIf (\(RegisterReply label' _ _) -> "a" == label') + (\(RegisterReply _ _ _) -> return ()) ] Just _ <- whereisRemote nid1 "a" -- Simulate network failure - liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) + liftIO $ syncBreakConnection testBreakConnection node1 node2 -- This will happen due to implicit reconnect registerRemoteAsync nid1 "b" us receiveWait [ - matchIf (\(RegisterReply label' _) -> "b" == label') - (\(RegisterReply _ _) -> return ()) ] + matchIf (\(RegisterReply label' _ _) -> "b" == label') + (\(RegisterReply _ _ _) -> return ()) ] -- Should happen registerRemoteAsync nid1 "c" us receiveWait [ - matchIf (\(RegisterReply label' _) -> "c" == label') - (\(RegisterReply _ _) -> return ()) ] + 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 @@ -757,6 +855,55 @@ takeMVar registerTestOk +-- | Tests that unreliable messages arrive sorted even when there are connection +-- failures. +testUSend :: (ProcessId -> Int -> Process ()) + -> TestTransport -> Int -> Assertion +testUSend usendPrim TestTransport{..} numMessages = do + [node1, node2] <- replicateM 2 $ newLocalNode testTransport initRemoteTable + let nid1 = localNodeId node1 + nid2 = localNodeId node2 + processA <- newEmptyMVar + usendTestOk <- newEmptyMVar + + forkProcess node1 $ flip catch (\e -> liftIO $ print (e :: SomeException) ) $ do + us <- getSelfPid + liftIO $ putMVar processA us + them <- expect + send them () + _ <- monitor them + let -- Collects messages from 'them' until the sender dies. + -- Disconnection notifications are ignored. + receiveMessages :: Process [Int] + receiveMessages = receiveWait + [ match $ \mn -> case mn of + ProcessMonitorNotification _ _ DiedDisconnect -> do + monitor them + receiveMessages + _ -> return [] + , match $ \i -> fmap (i :) receiveMessages + ] + msgs <- receiveMessages + let -- Checks that the input list is sorted. + isSorted :: [Int] -> Bool + isSorted (x : xs@(y : _)) = x <= y && isSorted xs + 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 () + + forkProcess node2 $ do + them <- liftIO $ readMVar processA + getSelfPid >>= send them + () <- expect + forM_ [1..numMessages] $ \i -> do + liftIO $ testBreakConnection (nodeAddress nid1) (nodeAddress nid2) + usendPrim them i + liftIO (threadDelay 30000) + + takeMVar usendTestOk + -- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server -- in between testMatchAny :: TestTransport -> Assertion @@ -1170,15 +1317,18 @@ localNode <- newLocalNode testTransport initRemoteTable supervisedDone <- newEmptyMVar supervisorDone <- newEmptyMVar + -- XXX: we guarantee that exception handler will be set up + -- regardless if forkProcess preserve masking state or not. + handlerSetUp <- newEmptyMVar pid <- forkProcess localNode $ do - (liftIO $ threadDelay 100000) - `catchExit` \_from reason -> do + (liftIO (putMVar handlerSetUp ()) >> expect) `catchExit` \_from reason -> do -- TODO: should verify that 'from' has the right value True <- return $ reason == "TestExit" liftIO $ putMVar supervisedDone () runProcess localNode $ do + liftIO $ takeMVar handlerSetUp ref <- monitor pid exit pid "TestExit" -- This time the client catches the exception, so it dies normally @@ -1233,6 +1383,26 @@ takeMVar clientDone +testUnsafeUSend :: TestTransport -> Assertion +testUnsafeUSend TestTransport{..} = do + serverAddr <- newEmptyMVar + clientDone <- newEmptyMVar + + localNode <- newLocalNode testTransport initRemoteTable + void $ forkProcess localNode $ do + self <- getSelfPid + liftIO $ putMVar serverAddr self + clientAddr <- expect + unsafeUSend clientAddr () + + void $ forkProcess localNode $ do + serverPid <- liftIO $ takeMVar serverAddr + getSelfPid >>= unsafeUSend serverPid + () <- expect + liftIO $ putMVar clientDone () + + takeMVar clientDone + testUnsafeNSend :: TestTransport -> Assertion testUnsafeNSend TestTransport{..} = do clientDone <- newEmptyMVar @@ -1249,6 +1419,25 @@ takeMVar clientDone +testUnsafeNSendRemote :: TestTransport -> Assertion +testUnsafeNSendRemote TestTransport{..} = do + clientDone <- newEmptyMVar + + localNode1 <- newLocalNode testTransport initRemoteTable + localNode2 <- newLocalNode testTransport initRemoteTable + + _ <- forkProcess localNode1 $ do + getSelfPid >>= register "foobar" + liftIO $ putMVar clientDone () + () <- expect + liftIO $ putMVar clientDone () + + takeMVar clientDone + void $ runProcess localNode2 $ do + unsafeNSendRemote (localNodeId localNode1) "foobar" () + + takeMVar clientDone + testUnsafeSendChan :: TestTransport -> Assertion testUnsafeSendChan TestTransport{..} = do serverAddr <- newEmptyMVar @@ -1270,20 +1459,103 @@ takeMVar clientDone +testCallLocal :: TestTransport -> Assertion +testCallLocal TestTransport{..} = do + node <- newLocalNode testTransport initRemoteTable + + -- Testing that (/=) <$> getSelfPid <*> callLocal getSelfPid. + result <- newEmptyMVar + runProcess node $ do + r <- (/=) <$> getSelfPid <*> callLocal getSelfPid + liftIO $ putMVar result r + True <- takeMVar result + return () + + -- Testing that when callLocal is interrupted, the worker is interrupted. + ibox <- newIORef False + runProcess node $ do + keeper <- getSelfPid + spawnLocal $ do + caller <- getSelfPid + send keeper caller + onException + (callLocal $ do + onException (do send keeper caller + expect) + (do liftIO $ writeIORef ibox True)) + (send keeper ()) + caller <- expect + exit caller "test" + () <- expect + return () + True <- readIORef ibox + return () + + -- Testing that when the worker raises an exception, the exception is propagated to the parent. + ibox2 <- newIORef False + runProcess node $ do + r <- try (callLocal $ error "e" >> return ()) + liftIO $ writeIORef ibox2 $ case r of + Left (ErrorCall "e") -> True + _ -> False + True <- readIORef ibox + return () + + -- Test that caller waits for the worker in correct situation + ibox3 <- newIORef False + result3 <- newEmptyMVar + runProcess node $ do + keeper <- getSelfPid + spawnLocal $ do + callLocal $ + (do us <- getSelfPid + send keeper us + () <- expect + liftIO yield) + `finally` (liftIO $ writeIORef ibox3 True) + liftIO $ putMVar result3 =<< readIORef ibox3 + worker <- expect + send worker () + True <- takeMVar result3 + return () + + -- Test that caller waits for the worker in case when caller gets an exception + ibox4 <- newIORef False + result4 <- newEmptyMVar + runProcess node $ do + keeper <- getSelfPid + spawnLocal $ do + caller <- getSelfPid + callLocal + ((do send keeper caller + expect) + `finally` (liftIO $ writeIORef ibox4 True)) + `finally` (liftIO $ putMVar result4 =<< readIORef ibox4) + caller <- expect + exit caller "hi!" + True <- takeMVar result4 + return () + -- XXX: Testing that when mask_ $ callLocal p runs p in masked state. + + + + + tests :: TestTransport -> IO [Test] tests testtrans = return [ - testGroup "Basic features" [ + 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 "RegistryRemoteProcess" (testRegistryRemoteProcess testtrans) , testCase "RemoteRegistry" (testRemoteRegistry testtrans) + , testCase "RemoteRegistryRemoteProcess" (testRemoteRegistryRemoteProcess testtrans) , testCase "SpawnLocal" (testSpawnLocal testtrans) + , testCase "SpawnAsyncStrictness" (testSpawnAsyncStrictness testtrans) , testCase "HandleMessageIf" (testHandleMessageIf testtrans) , testCase "MatchAny" (testMatchAny testtrans) , testCase "MatchAnyHandle" (testMatchAnyHandle testtrans) @@ -1301,19 +1573,27 @@ , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans) , testCase "ExitLocal" (testExitLocal testtrans) , testCase "ExitRemote" (testExitRemote testtrans) + , testCase "TextCallLocal" (testCallLocal testtrans) -- Unsafe Primitives , testCase "TestUnsafeSend" (testUnsafeSend testtrans) + , testCase "TestUnsafeUSend" (testUnsafeUSend testtrans) , testCase "TestUnsafeNSend" (testUnsafeNSend testtrans) + , testCase "TestUnsafeNSendRemote" (testUnsafeNSendRemote testtrans) , testCase "TestUnsafeSendChan" (testUnsafeSendChan testtrans) + -- usend + , testCase "USend" (testUSend usend testtrans 50) + , testCase "UForward" + (testUSend (\p m -> uforward (createUnencodedMessage m) p) + testtrans 50 + ) ] - , testGroup "Monitoring and Linking" [ + , 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 "MonitorNormalTermination" (testMonitorNormalTermination testtrans True False) , testCase "MonitorAbnormalTermination" (testMonitorAbnormalTermination testtrans True False) , testCase "MonitorLocalDeadProcess" (testMonitorLocalDeadProcess testtrans True False) , testCase "MonitorRemoteDeadProcess" (testMonitorRemoteDeadProcess testtrans True False) @@ -1335,6 +1615,27 @@ , testCase "MonitorLiveNode" (testMonitorLiveNode testtrans) , testCase "MonitorChannel" (testMonitorChannel testtrans) -- Reconnect - , testCase "Reconnect" (testReconnect testtrans) ] + + -- Tests that fail occasionally and should be revised + , testGroup "Flaky" [ + testCase "Reconnect" (testReconnect testtrans) + , testCase "Registry" (testRegistry testtrans) + , testCase "MergeChannels" (testMergeChannels testtrans) + , testCase "MonitorUnreachable" (testMonitorUnreachable testtrans True False) + ] ] + +syncBreakConnection :: (NT.EndPointAddress -> NT.EndPointAddress -> IO ()) -> LocalNode -> LocalNode -> IO () +syncBreakConnection breakConnection nid0 nid1 = do + m <- newEmptyMVar + _ <- forkProcess nid1 $ getSelfPid >>= liftIO . putMVar m + runProcess nid0 $ do + them <- liftIO $ takeMVar m + pinger <- spawnLocal $ forever $ send them () + _ <- monitorNode (localNodeId nid1) + liftIO $ breakConnection (nodeAddress $ localNodeId nid0) + (nodeAddress $ localNodeId nid1) + NodeMonitorNotification _ _ _ <- expect + kill pinger "finished" + return () 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 @@ -6,7 +6,8 @@ import Data.ByteString.Lazy (empty) import Data.IORef import Data.Typeable (Typeable) -import Control.Monad (join, replicateM, forever, replicateM_, void, when) +import Data.Maybe +import Control.Monad (join, replicateM, forever, replicateM_, void, when, unless) import Control.Exception (IOException, throw) import Control.Concurrent (forkIO, threadDelay) import Control.Concurrent.MVar @@ -63,6 +64,9 @@ wait :: Int -> Process () wait = liftIO . threadDelay +expectUnit :: Process () +expectUnit = expect + isPrime :: Integer -> Process Bool isPrime n = return . (n `elem`) . takeWhile (<= n) . sieve $ [2..] where @@ -85,6 +89,7 @@ , 'sendPid , 'sdictInt , 'wait + , 'expectUnit , 'typedPingServer , 'isPrime , 'quintuple @@ -141,11 +146,19 @@ waitClosure :: Int -> Closure (Process ()) waitClosure = $(mkClosure 'wait) -simulateNetworkFailure :: TestTransport -> NodeId -> NodeId -> Process () +simulateNetworkFailure :: TestTransport -> LocalNode -> LocalNode -> Process () simulateNetworkFailure TestTransport{..} from to = liftIO $ do - threadDelay 10000 - testBreakConnection (nodeAddress from) (nodeAddress to) - threadDelay 10000 + m <- newEmptyMVar + _ <- forkProcess to $ getSelfPid >>= liftIO . putMVar m + runProcess from $ do + them <- liftIO $ takeMVar m + pinger <- spawnLocal $ forever $ send them () + _ <- monitorNode (localNodeId to) + liftIO $ testBreakConnection (nodeAddress $ localNodeId from) + (nodeAddress $ localNodeId to) + NodeMonitorNotification _ _ _ <- expect + kill pinger "finished" + return () -------------------------------------------------------------------------------- -- The tests proper -- @@ -363,19 +376,35 @@ [node1, node2] <- replicateM 2 $ newLocalNode testTransport rtable [superPid, childPid] <- replicateM 2 $ newEmptyMVar thirdProcessDone <- newEmptyMVar + linkUp <- 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 + (child, _ref) <- spawnSupervised (localNodeId node2) + (sendPidClosure us `seqCP` $(mkStaticClosure 'expectUnit)) + _ <- expect :: Process ProcessId + liftIO $ do putMVar childPid child + -- Give the child a chance to link to us + takeMVar linkUp + 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 @@ -465,7 +494,7 @@ forkProcess node2 $ do _pid1 <- spawn nid1 ($(mkClosure 'signal) incr) - simulateNetworkFailure testtrans nid2 nid1 + simulateNetworkFailure testtrans node2 node1 _pid2 <- spawn nid1 ($(mkClosure 'signal) incr) _pid3 <- spawn nid1 ($(mkClosure 'signal) incr) 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 @@ -5,6 +5,7 @@ import Control.Distributed.Process.Tests.Internal.Utils import Network.Transport.Test (TestTransport(..)) +import Control.Applicative ((<*)) import Control.Concurrent (threadDelay) import Control.Concurrent.MVar ( MVar @@ -19,9 +20,13 @@ import Control.Distributed.Process.Management ( MxEvent(..) ) +import qualified Control.Exception as IO (bracket) +import Data.List (isPrefixOf, isSuffixOf) #if ! MIN_VERSION_base(4,6,0) import Prelude hiding (catch, log) +#else +import Prelude hiding ((<*)) #endif import Test.Framework @@ -29,7 +34,11 @@ , testGroup ) import Test.Framework.Providers.HUnit (testCase) +import System.Environment (getEnvironment) +-- These are available in System.Environment only since base 4.7 +import System.SetEnv (setEnv, unsetEnv) + testSpawnTracing :: TestResult Bool -> Process () testSpawnTracing result = do setTraceFlags defaultTraceFlags { @@ -282,6 +291,82 @@ -- and just to be polite... liftIO $ closeLocalNode node2 +-- | Sets the value of an environment variable while executing the given IO +-- computation and restores the preceeding value upon completion. +withEnv :: String -> String -> IO a -> IO a +withEnv var val = + IO.bracket (fmap (lookup var) getEnvironment <* setEnv var val) + (maybe (unsetEnv var) (setEnv var)) + . const + +-- | Tests that one and only one interesting trace message is produced when a +-- given action is performed. A message is considered interesting when the given +-- function return @True@. +testSystemLoggerMsg :: TestTransport + -> Process a + -> (a -> String -> Bool) + -> IO () +testSystemLoggerMsg t action interestingMessage = + withEnv "DISTRIBUTED_PROCESS_TRACE_CONSOLE" "yes" $ + withEnv "DISTRIBUTED_PROCESS_TRACE_FLAGS" "pdnusrl" $ do + n <- newLocalNode (testTransport t) initRemoteTable + + runProcess n $ do + self <- getSelfPid + reregister "trace.logger" self + a <- action + let interestingMessage' (_ :: String, msg) = interestingMessage a msg + -- Wait for the trace message. + receiveWait [ matchIf interestingMessage' $ const $ return () ] + -- Only one interesting message should arrive. + Nothing <- receiveTimeout 100000 + [ matchIf interestingMessage' $ const $ return () ] + return () + + +-- | Tests that one and only one trace message is produced when a message is +-- received. +testSystemLoggerMxReceive :: TestTransport -> IO () +testSystemLoggerMxReceive t = testSystemLoggerMsg t + (getSelfPid >>= flip send ()) + (\_ msg -> "MxReceived" `isPrefixOf` msg + -- discard traces of internal messages + && not (":: RegisterReply" `isSuffixOf` msg) + ) + +-- | Tests that one and only one trace message is produced when a message is +-- sent. +testSystemLoggerMxSent :: TestTransport -> IO () +testSystemLoggerMxSent t = testSystemLoggerMsg t + (getSelfPid >>= flip send ()) + (const $ isPrefixOf "MxSent") + +-- | Tests that one and only one trace message is produced when a process dies. +testSystemLoggerMxProcessDied :: TestTransport -> IO () +testSystemLoggerMxProcessDied t = testSystemLoggerMsg t + (spawnLocal $ return ()) + (\pid -> isPrefixOf $ "MxProcessDied " ++ show pid) + +-- | Tests that one and only one trace message appears when a process spawns. +testSystemLoggerMxSpawned :: TestTransport -> IO () +testSystemLoggerMxSpawned t = testSystemLoggerMsg t + (spawnLocal $ return ()) + (\pid -> isPrefixOf $ "MxSpawned " ++ show pid) + +-- | Tests that one and only one trace message appears when a process is +-- registered. +testSystemLoggerMxRegistered :: TestTransport -> IO () +testSystemLoggerMxRegistered t = testSystemLoggerMsg t + (getSelfPid >>= register "a" >> getSelfPid) + (\self -> isPrefixOf $ "MxRegistered " ++ show self ++ " " ++ show "a") + +-- | Tests that one and only one trace message appears when a process is +-- unregistered. +testSystemLoggerMxUnRegistered :: TestTransport -> IO () +testSystemLoggerMxUnRegistered t = testSystemLoggerMsg t + (getSelfPid >>= register "a" >> unregister "a" >> getSelfPid) + (\self -> isPrefixOf $ "MxUnRegistered " ++ show self ++ " " ++ show "a") + tests :: TestTransport -> IO [Test] tests testtrans@TestTransport{..} = do node1 <- newLocalNode testTransport initRemoteTable @@ -323,4 +408,12 @@ (synchronisedAssertion "expected blah" node1 True (testRemoteTraceRelay testtrans) lock) + , testGroup "SystemLoggerTracer" + [ testCase "MxReceive" $ testSystemLoggerMxReceive testtrans + , testCase "MxSent" $ testSystemLoggerMxSent testtrans + , testCase "MxProcessDied" $ testSystemLoggerMxProcessDied testtrans + , testCase "MxSpawned" $ testSystemLoggerMxSpawned testtrans + , testCase "MxRegistered" $ testSystemLoggerMxRegistered testtrans + , testCase "MxUnRegistered" $ testSystemLoggerMxUnRegistered testtrans + ] ] ] diff --git a/tests/runInMemory.hs b/tests/runInMemory.hs new file mode 100644 --- /dev/null +++ b/tests/runInMemory.hs @@ -0,0 +1,31 @@ +-- Run tests using the TCP transport. + +module Main where + +import TEST_SUITE_MODULE (tests) + +import Network.Transport.Test (TestTransport(..)) +import Network.Transport.InMemory +import Test.Framework (defaultMainWithArgs) + +import Control.Concurrent (threadDelay) +import System.Environment (getArgs) + +main :: IO () +main = do + (transport, internals) <- createTransportExposeInternals + ts <- tests TestTransport + { testTransport = transport + , testBreakConnection = \addr1 addr2 -> breakConnection internals addr1 addr2 "user error" + } + args <- getArgs + -- Tests are time sensitive. Running the tests concurrently can slow them + -- down enough that threads using threadDelay would wake up later than + -- expected, thus changing the order in which messages were expected. + -- Therefore we run the tests sequentially by passing "-j 1" to + -- test-framework. This does not solve the issue but makes it less likely. + -- + -- The problem was first detected with + -- 'Control.Distributed.Process.Tests.CH.testMergeChannels' + -- in particular. + defaultMainWithArgs ts ("-j" : "1" : args) diff --git a/tests/runTCP.hs b/tests/runTCP.hs --- a/tests/runTCP.hs +++ b/tests/runTCP.hs @@ -1,5 +1,5 @@ -- Run tests using the TCP transport. - +-- module Main where import TEST_SUITE_MODULE (tests) @@ -10,16 +10,21 @@ ( createTransportExposeInternals , TransportInternals(socketBetween) , defaultTCPParameters + , TCPParameters(..) ) import Test.Framework (defaultMainWithArgs) import Control.Concurrent (threadDelay) import System.Environment (getArgs) +import System.IO main :: IO () main = do + hSetBuffering stdout LineBuffering + hSetBuffering stderr LineBuffering Right (transport, internals) <- - createTransportExposeInternals "127.0.0.1" "8080" defaultTCPParameters + createTransportExposeInternals "127.0.0.1" "8080" + defaultTCPParameters { transportConnectTimeout = Just 3000000 } ts <- tests TestTransport { testTransport = transport , testBreakConnection = \addr1 addr2 -> do