@@ -25,13 +25,18 @@ import Control.Concurrent.MVar
2525 )
2626import Control.Monad (replicateM_, replicateM, forever, void, unless, join)
2727import Control.Exception (SomeException, throwIO, ErrorCall(..))
28−import qualified Control.Exception as Ex (catch)
29−import Control.Applicative ((<$>), (<*>), pure, (<|>))
28+import Control.Monad.Catch (try, catch, finally, mask, onException)
29+import Control.Applicative ((<|>))
3030import qualified Network.Transport as NT (closeEndPoint, EndPointAddress)
31−import Control.Distributed.Process
31+import Control.Distributed.Process hiding
32+ ( try
33+ , catch
34+ , finally
35+ , mask
36+ , onException
37+ )
3238import Control.Distributed.Process.Internal.Types
33− ( NodeId(nodeAddress)
34− , LocalNode(localEndPoint)
39+ ( LocalNode(localEndPoint)
3540 , ProcessExitException(..)
3641 , nullProcessId
3742 , createUnencodedMessage
@@ -39,13 +44,13 @@ import Control.Distributed.Process.Internal.Types
3944import Control.Distributed.Process.Node
4045import Control.Distributed.Process.Debug
4146import Control.Distributed.Process.Management.Internal.Types
42−import Control.Distributed.Process.Tests.Internal.Utils (shouldBe)
47+import Control.Distributed.Process.Tests.Internal.Utils (shouldBe, pause)
4348import Control.Distributed.Process.Serializable (Serializable)
44−
49+import Data.Maybe (isNothing, isJust)
4550import Test.HUnit (Assertion, assertBool, assertFailure)
4651import Test.Framework (Test, testGroup)
4752import Test.Framework.Providers.HUnit (testCase)
48−import Control.Rematch hiding (match)
53+import Control.Rematch hiding (match, isNothing, isJust)
4954import Control.Rematch.Run (Match(..))
5055
5156newtype Ping = Ping ProcessId
@@ -68,7 +73,7 @@ expectThat a matcher = case res of
6873forkTry :: IO () -> IO ThreadId
6974forkTry p = do
7075 tid <- myThreadId
71− forkIO $ Ex.catch p (\e -> throwTo tid (e :: SomeException))
76+ forkIO $ catch p (\e -> throwTo tid (e :: SomeException))
7277
7378-- | The ping server from the paper
7479ping :: Process ()
@@ -82,10 +87,9 @@ verifyClient :: String -> MVar Bool -> IO ()
8287verifyClient s b = takeMVar b >>= assertBool s
8388
8489expectPing :: MVar Bool -> Process ()
85−expectPing mv = expect >>= liftIO . putMVar mv . checkPing
90+expectPing mv = expect >>= liftIO . putMVar mv . checkPing
8691 where
8792 checkPing (Ping _) = True
88− checkPing _ = False
8993
9094-- | Quick and dirty synchronous version of whereisRemoteAsync
9195whereisRemote :: NodeId -> String -> Process (Maybe ProcessId)
@@ -235,10 +239,10 @@ testPing TestTransport{..} = do
235239 pid <- getSelfPid
236240 replicateM_ numPings $ do
237241 send pingServer (Pong pid)
238− p <- expect
242+ p <- expectTimeout 3000000
239243 case p of
240− Ping _ -> return ()
241− _ -> die "Unexpected message"
244+ Just (Ping _) -> return ()
245+ Nothing -> die "Failed to receive Ping"
242246
243247 putMVar clientDone ()
244248
@@ -555,7 +559,7 @@ testMergeChannels TestTransport{..} = do
555559 ss <- liftIO $ mapM readMVar vs
556560 case ss of
557561 [sa, sb, sc] ->
558− mapM_ ((>> liftIO (threadDelay 10000)) . uncurry sendChan)
562+ mapM_ ((>> pause 10000) . uncurry sendChan)
559563 [ -- a, b, c
560564 (sa, 'a')
561565 , (sb, 'b')
@@ -731,7 +735,6 @@ testRemoteRegistry :: TestTransport -> Assertion
731735testRemoteRegistry TestTransport{..} = do
732736 node1 <- newLocalNode testTransport initRemoteTable
733737 node2 <- newLocalNode testTransport initRemoteTable
734− done <- newEmptyMVar
735738
736739 pingServer <- forkProcess node1 ping
737740 deadProcess <- forkProcess node1 (return ())
@@ -746,6 +749,7 @@ testRemoteRegistry TestTransport{..} = do
746749
747750 pid <- verifyWhereIsRemote nid1 "ping"
748751 liftIO $ assertBool "Expected pindServer to match pid" $ pingServer == pid
752+
749753 us <- getSelfPid
750754 nsendRemote nid1 "ping" (Pong us)
751755 receiveWait [
@@ -755,22 +759,22 @@ testRemoteRegistry TestTransport{..} = do
755759 -- test that if process was not registered Nothing is returned
756760 -- in owner field.
757761 registerRemoteAsync nid1 "dead" deadProcess
758− receiveWait [
759− matchIf (\(RegisterReply label' False Nothing) -> "dead" == label')
760− (\(RegisterReply _ _ _) -> return ()) ]
762+ receiveWait [ matchIf (\(RegisterReply label' f mPid) -> "dead" == label')
763+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))
764+ ] >>= liftIO . assertBool "Expected False Nothing in RegisterReply"
765+
761766 registerRemoteAsync nid1 "ping" deadProcess
762767 receiveWait [
763− matchIf (\(RegisterReply label' False (Just pid)) ->
764− "ping" == label' && pid == pingServer)
765− (\(RegisterReply _ _ _) -> return ()) ]
768+ matchIf (\(RegisterReply label' False mPid) ->
769+ "ping" == label' && isJust mPid)
770+ (\(RegisterReply _ f (Just pid'')) -> return (not f && pid'' == pingServer))
771+ ] >>= liftIO . assertBool "Expected False and (Just alreadyRegisteredPid) in RegisterReply"
772+
766773 unregisterRemoteAsync nid1 "dead"
767774 receiveWait [
768− matchIf (\(RegisterReply label' False Nothing) ->
769− "dead" == label' && pid == pingServer)
770− (\(RegisterReply _ _ _) -> return ()) ]
771− liftIO $ putMVar done ()
772−
773− takeMVar done
775+ matchIf (\(RegisterReply label' _ _) -> "dead" == label')
776+ (\(RegisterReply _ f mPid) -> return (not f && isNothing mPid))
777+ ] >>= liftIO . assertBool "Expected False and Nothing in RegisterReply"
774778
775779testRemoteRegistryRemoteProcess :: TestTransport -> Assertion
776780testRemoteRegistryRemoteProcess TestTransport{..} = do
@@ -942,9 +946,7 @@ testUSend usendPrim TestTransport{..} numMessages = do
942946 isSorted _ = True
943947 -- The list can't be null since there are no failures after sending
944948 -- the last message.
945− liftIO $ do
946− assertBool "Unexpected failure after sending last msg" $ isSorted msgs && not (null msgs)
947− putMVar usendTestOk ()
949+ liftIO $ putMVar usendTestOk $ isSorted msgs && not (null msgs)
948950
949951 forkProcess node2 $ do
950952 them <- liftIO $ readMVar processA
@@ -955,7 +957,8 @@ testUSend usendPrim TestTransport{..} numMessages = do
955957 usendPrim them i
956958 liftIO (threadDelay 30000)
957959
958− takeMVar usendTestOk
960+ res <- takeMVar usendTestOk
961+ assertBool "Unexpected failure after sending last msg" res
959962
960963-- | Test 'matchAny'. This repeats the 'testMath' but with a proxy server
961964-- in between
@@ -1269,6 +1272,76 @@ testReceiveChanFeatures TestTransport{..} = do
12691272
12701273 takeMVar done
12711274
1275+testChanLifecycle :: TestTransport -> Assertion
1276+testChanLifecycle TestTransport{..} = let delay = 3000000 in do
1277+ result <- newEmptyMVar
1278+ tchMV <- newEmptyMVar
1279+ localNode <- newLocalNode testTransport initRemoteTable
1280+ runProcess localNode $ do
1281+
1282+ pid <- spawnLocal $ do tCh <- newChan :: Process (SendPort (), ReceivePort ())
1283+ liftIO $ putMVar tchMV tCh
1284+ expect :: Process ()
1285+ let (sp, _) = tCh
1286+ sendChan sp ()
1287+ expect :: Process ()
1288+
1289+ mRefPid <- monitor pid
1290+
1291+ cPid <- spawnLocal $ do
1292+ (sp', rp) <- liftIO $ takeMVar tchMV
1293+ -- say "picked up our test channel"
1294+ send pid ()
1295+ -- say "told pid to continue"
1296+ res <- receiveChanTimeout delay rp
1297+ case res of
1298+ Nothing -> say "initial chan () missing!" >> (liftIO $ putMVar result False)
1299+ Just () -> do mr <- monitor pid
1300+ pause 10000
1301+ -- say "sending pid a second () will cause it to exit"
1302+ send pid ()
1303+
1304+ -- say "make sure we see a DOWN notification for pid having stopped"
1305+ receiveWait [ match (\(_ :: ProcessMonitorNotification) -> return ()) ]
1306+
1307+ -- now that pid has died, the send port should be useless...
1308+ liftIO $ putMVar tchMV (sp', rp)
1309+
1310+ -- let's verify that we do not see the message from our
1311+ -- parent process on the channel, once pid has died...
1312+ recv <- receiveChanTimeout delay rp
1313+ -- say $ "finished waiting for second (), writing result" ++ (show recv)
1314+ liftIO $ putMVar result $ isNothing recv
1315+
1316+ mRefCPid <- monitor cPid
1317+
1318+ receiveWait
1319+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefPid)
1320+ (\_ -> return ())
1321+ ]
1322+
1323+ -- say "seen first pid die..."
1324+
1325+ (sendPort, _) <- liftIO $ takeMVar tchMV
1326+ sendChan sendPort ()
1327+ -- say "sent () after owning pid died"
1328+
1329+ -- let cPid know we've written to the channel...
1330+ send cPid ()
1331+
1332+ receiveWait
1333+ [ matchIf (\(ProcessMonitorNotification r _ _) -> r == mRefCPid)
1334+ (\_ -> return ())
1335+ ]
1336+
1337+ -- say "seen both pids die now..."
1338+
1339+ -- and wait on the result back in IO land...
1340+ testRes <- takeMVar result
1341+ -- runProcess localNode $ say "got result..."
1342+ assertBool "Expected sending on the channel to fail, but received data!" testRes
1343+
1344+
12721345testKillLocal :: TestTransport -> Assertion
12731346testKillLocal TestTransport{..} = do
12741347 localNode <- newLocalNode testTransport initRemoteTable
@@ -1481,113 +1554,73 @@ testExitRemote TestTransport{..} = do
14811554
14821555testRegistryMonitoring :: TestTransport -> Assertion
14831556testRegistryMonitoring TestTransport{..} = do
1484− localNode <- newLocalNode testTransport initRemoteTable
1485− remoteNode <- newLocalNode testTransport initRemoteTable
1486− return ()
1557+ node1 <- newLocalNode testTransport initRemoteTable
1558+ node2 <- newLocalNode testTransport initRemoteTable
14871559
1488− -- Local process. Test if local process will be removed from
1489− -- registry when it dies.
1490− box <- newEmptyMVar
1491− runProcess localNode $ do
1492− pid <- spawnLocal $ do
1493− expect
1494− register "test" pid
1495− tpid <- whereis "test"
1496− if tpid == Just pid
1497− then do _ <- monitor pid
1498− send pid ()
1499− ProcessMonitorNotification{} <- expect
1500− tpid1 <- whereis "test"
1501− liftIO $ putMVar box (Nothing == tpid1)
1502− else liftIO $ putMVar box False
1503−
1504− takeMVar box >>= assertBool "expected local process to not be registered"
1505− return ()
1506−
1507− -- Remote process. Test if remote process entry is removed
1508− -- from registry when process dies.
1509− remote1 <- testRemote remoteNode
1510− runProcess localNode $
1511− let waitpoll = do
1512− w <- whereis "test" :: Process (Maybe ProcessId)
1513− forM_ w (const waitpoll)
1514− in do register "test" remote1
1515− -- [race-condition] invalid test if we force exit before waitpoll runs
1516− registered <- whereis "test"
1517− registered `shouldBe` (equalTo (Just remote1))
1518− send remote1 ()
1519− waitpoll
1520− return ()
1521− return ()
1522−
1523− -- Many labels. Test if all labels associated with process
1524− -- are removed from registry when it dies.
1525− remote2 <- testRemote remoteNode
1526− runProcess localNode $
1527− let waitpoll = do
1528− w1 <- whereis "test-3" :: Process (Maybe ProcessId)
1529− w2 <- whereis "test-4" :: Process (Maybe ProcessId)
1530− forM_ (w1 <|> w2) (const waitpoll)
1531− in do register "test-3" remote2
1532− register "test-4" remote2
1533− send remote2 ()
1534− waitpoll
1535− return ()
1536−
1537−{- XXX: waiting including patch for nsend for remote process
1538− remote3 <- testRemote remoteNode
1539− remote4 <- testRemote remoteNode
1540− -- test many labels
1541− runProcess localNode $ do
1542− register "test-3" remote3
1543− reregister "test-3" remote4
1544− send remote3 ()
1545− liftIO $ threadDelay 50000 -- XXX: racy
1546− monitor remote4
1547− nsend "test-3" ()
1548− ProcessMonitorNotification{} <- expect
1549− return ()
1550−-}
1551−
1552− -- Test registerRemoteAsync properties. Add a local process to
1553− -- remote registry and checks that it is removed
1554− -- when the process dies.
1555− remote5 <- testRemote remoteNode
1556− runProcess localNode $ do
1557− registerRemoteAsync (localNodeId remoteNode) "test" remote5
1558− receiveWait [
1559− match (\(RegisterReply _ True _) -> return ())
1560− ] >>= send remote5
1561− let waitpoll = do
1562− whereisRemoteAsync (localNodeId remoteNode) "test"
1563− receiveWait [
1564− match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))
1565− ]
1566− waitpoll
1560+ let nid = localNodeId node2
1561+ pid <- forkProcess node1 $ do
1562+ self <- getSelfPid
1563+ runUntilRegistered nid self
1564+ say $ (show self) ++ " registered as " ++ regName
1565+ expect :: Process ()
1566+ say $ (show self) ++ " exiting normally"
15671567
1568− -- Add remote process to remote registry and checks if
1569− -- entry is removed then process is dead.
1570− remote6 <- testRemote localNode
1571− runProcess localNode $ do
1572− registerRemoteAsync (localNodeId remoteNode) "test" remote6
1573− receiveWait [
1574− match (\(RegisterReply _ True _) -> return ())
1575− ] >>= send remote6
1576− let waitpoll = do
1577− whereisRemoteAsync (localNodeId remoteNode) "test"
1578− receiveWait [
1579− match (\(WhereIsReply _ mr) -> forM_ mr (const waitpoll))
1580− ]
1581− waitpoll
1582− where
1583− testRemote node = do
1584− -- test many labels
1585− pidBox <- newEmptyMVar
1586− forkProcess node $ do
1587− us <- getSelfPid
1588− liftIO $ putMVar pidBox us
1589− expect :: Process ()
1590− takeMVar pidBox
1568+ runProcess node2 $ do
1569+ register regName pid
1570+ say $ regName ++ " registered to " ++ show pid
1571+ res <- whereis regName
1572+ send pid ()
1573+ say $ " sent finish signal to " ++ show pid
1574+ us <- getSelfPid
1575+ liftIO $ assertBool "expected (Just pid)" $ res == (Just pid)
1576+
1577+
1578+ -- This delay isn't essential!
1579+ -- The test case passes perfectly fine without it (feel free to comment out
1580+ -- and see), however waiting a few seconds here, makes it much more likely
1581+ -- that in delayUntilMaybeUnregistered we will hit the match case right
1582+ -- away, and thus not be subjected to a 20 second delay. The value of 4
1583+ -- seconds appears to work optimally on osx and across several linux distros
1584+ -- running in virtual machines (which is essentially what we do in CI)
1585+ void $ receiveTimeout 4000000 [ matchAny return ]
1586+
1587+ -- This delay doesn't serve much purpose in the happy path, however if some
1588+ -- future patch breaks the cooperative behaviour of node controllers viz
1589+ -- remote process registration and notification taking place via ncEffectDied,
1590+ -- there would be the possibility of a race in the test case should we attempt
1591+ -- to evaluate `whereis regName` on node2 right away. In case the name is still
1592+ -- erroneously registered, observing the 20 second delay (or lack of), could at
1593+ -- least give a hint that something is wrong, and we give up our time slice
1594+ -- so that there's a higher change the registrations have been cleaned up
1595+ -- in either case.
1596+ runProcess node2 $ delayUntilMaybeUnregistered nid pid
1597+
1598+ regHere <- newEmptyMVar
1599+ runProcess node2 $ whereis regName >>= liftIO . putMVar regHere
1600+ res <- takeMVar regHere
1601+ case res of
1602+ Nothing -> return ()
1603+ Just pid -> assertBool ("expected Nothing, but got " ++ show pid) False
1604+
1605+ where
1606+ runUntilRegistered nid us = do
1607+ whereisRemoteAsync nid regName
1608+ receiveWait [
1609+ matchIf (\(WhereIsReply n (Just p)) -> n == regName && p == us)
1610+ (const $ return ())
1611+ ]
1612+
1613+ delayUntilMaybeUnregistered nid p = do
1614+ whereisRemoteAsync nid regName
1615+ res <- receiveTimeout 20000000 {- 20 sec delay -} [
1616+ matchIf (\(WhereIsReply n p) -> n == regName && isNothing p)
1617+ (const $ return ())
1618+ ]
1619+ case res of
1620+ Just () -> return ()
1621+ Nothing -> delayUntilMaybeUnregistered nid p
1622+
1623+ regName = "testRegisterRemote"
15911624
15921625testUnsafeSend :: TestTransport -> Assertion
15931626testUnsafeSend TestTransport{..} = do
@@ -1781,6 +1814,7 @@ tests testtrans = return [
17811814 , testCase "MatchMessageUnwrap" (testMatchMessageWithUnwrap testtrans)
17821815 , testCase "ReceiveChanTimeout" (testReceiveChanTimeout testtrans)
17831816 , testCase "ReceiveChanFeatures" (testReceiveChanFeatures testtrans)
1817+ , testCase "ChanLifecycle" (testChanLifecycle testtrans)
17841818 , testCase "KillLocal" (testKillLocal testtrans)
17851819 , testCase "KillRemote" (testKillRemote testtrans)
17861820 , testCase "Die" (testDie testtrans)
@@ -1790,7 +1824,7 @@ tests testtrans = return [
17901824 , testCase "MaskRestoreScope" (testMaskRestoreScope testtrans)
17911825 , testCase "ExitLocal" (testExitLocal testtrans)
17921826 , testCase "ExitRemote" (testExitRemote testtrans)
1793− , testCase "TestRegistryMonitor" (testRegistryMonitoring testtrans)
1827+ , testCase "RegistryMonitoring" (testRegistryMonitoring testtrans)
17941828 , testCase "TextCallLocal" (testCallLocal testtrans)
17951829 -- Unsafe Primitives
17961830 , testCase "TestUnsafeSend" (testUnsafeSend testtrans)