54129b0 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Another instance of the async exception bug

54129b0Edsko de Vries14 years ago

network-transport/network-transport.cabal+7−4

compare with master

unifiedsplit

@@ -37,9 +37,10 @@ Test-Suite TestTCP

37 network,

38 mtl,

39 transformers,

40− ansi-terminal

40+ ansi-terminal,

41+ random

41 extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, EmptyDataDecls, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

42 extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, EmptyDataDecls, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

42− ghc-options: -Wall -fno-warn-unused-do-bind

43+ ghc-options: -Wall -fno-warn-unused-do-bind -threaded -rtsopts -with-rtsopts=-N

43 HS-Source-Dirs: tests src

44 HS-Source-Dirs: tests src

45Test-Suite TestMulticastInMemory

46Test-Suite TestMulticastInMemory

@@ -51,7 +52,8 @@ Test-Suite TestMulticastInMemory

51 data-accessor,

52 data-accessor,

52 mtl,

53 mtl,

53 transformers,

54 transformers,

54− ansi-terminal

55+ ansi-terminal,

56+ random

55 extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, EmptyDataDecls, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

57 extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, EmptyDataDecls, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

56 ghc-options: -Wall -fno-warn-unused-do-bind

58 ghc-options: -Wall -fno-warn-unused-do-bind

57 HS-Source-Dirs: tests src

59 HS-Source-Dirs: tests src

@@ -65,7 +67,8 @@ Test-Suite TestInMemory

65 data-accessor,

67 data-accessor,

66 mtl,

68 mtl,

67 transformers,

69 transformers,

68− ansi-terminal

70+ ansi-terminal,

71+ random

69 extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, EmptyDataDecls, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

72 extensions: OverloadedStrings, ForeignFunctionInterface, DeriveDataTypeable, EmptyDataDecls, RankNTypes, ExistentialQuantification, FlexibleInstances, OverlappingInstances

70 ghc-options: -Wall -fno-warn-unused-do-bind

73 ghc-options: -Wall -fno-warn-unused-do-bind

71 HS-Source-Dirs: tests src

74 HS-Source-Dirs: tests src

network-transport/src/Network/Transport/TCP.hs+11−12

@@ -483,7 +483,7 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

483 -- instance if the connection is closed) with the outer exception (which is

484 -- returned by 'try' when an exception is thrown by 'sendOn', and handled

485 -- by 'withRemoteState')

486− liftM join . try . mapIOException sendFailed $

486+ liftM join . try . mapIOException sendFailed $

487 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

488 { caseInvalid = \_ ->

489 relyViolation (ourEndPoint, theirEndPoint) "apiSend"

@@ -679,7 +679,7 @@ requestConnectionTo ourEndPoint theirAddress = go

679 findRemoteEndPoint ourEndPoint theirAddress

680

681 -- If it was new, start listening for incoming messages, too

682− when isNew . void . forkEndPointThread ourEndPoint $

682+ when isNew . void . forkEndPointThread ourEndPoint $

683 setupRemoteEndPoint (ourEndPoint, theirEndPoint)

684

685 let theirState = remoteState theirEndPoint

@@ -703,15 +703,15 @@ requestConnectionTo ourEndPoint theirAddress = go

703 -- endpoint state, and might have changed in the meantime, these changes

704 -- won't matter.

705 case endPointStateSnapshot of

706− RemoteEndPointInvalid err ->

706+ RemoteEndPointInvalid err ->

707 throwIO err

708

709− RemoteEndPointClosing resolved _ ->

709+ RemoteEndPointClosing resolved _ ->

710 -- If the remote endpoint is closing, then we need to block until

711 -- this is resolved and we then try again

712 readMVar resolved >> go

713

714− RemoteEndPointClosed ->

714+ RemoteEndPointClosed ->

715 -- RemoteEndPointClosed indicates that a concurrent thread was in the

716 -- process of closing the TCP connection to the remote endpoint when

717 -- we obtained a reference to it. By INV-CLOSE we can assume that the

@@ -728,9 +728,9 @@ requestConnectionTo ourEndPoint theirAddress = go

728 reply <- mapIOException connectFailed $

729 doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

730 case decodeInt32 . BS.concat $ reply of

731− Nothing ->

731+ Nothing ->

732 throwIO (connectFailed $ userError "Invalid integer")

733− Just cid ->

733+ Just cid ->

734 return (theirEndPoint, cid)

735

736 connectFailed = TransportError ConnectFailed . show

@@ -772,15 +772,14 @@ findRemoteEndPoint ourEndPoint theirAddress =

772setupRemoteEndPoint :: EndPointPair -> IO ()

773setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do

774 didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress)

775− onError $ \result ->

775+ onError $ \result ->

776 case result of

777 Right (sock, ConnectionRequestAccepted) -> do

778 let vst = ValidRemoteEndPointState

779 { remoteSocket = sock

780 , _remoteOutgoing = 0

781 , _remoteIncoming = IntSet.empty

782− , sendOn = (`onException` tryCloseSocket sock)

783− . sendMany sock

782+ , sendOn = sendMany sock

784 }

783 }

785 putMVar theirState (RemoteEndPointValid vst)

784 putMVar theirState (RemoteEndPointValid vst)

786 return True

785 return True

@@ -871,7 +870,7 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do

871--

870--

872-- If the local endpoint is closed, do nothing

871-- If the local endpoint is closed, do nothing

873removeRemoteEndPoint :: EndPointPair -> IO ()

872removeRemoteEndPoint :: EndPointPair -> IO ()

874−removeRemoteEndPoint (ourEndPoint, theirEndPoint) =

873+removeRemoteEndPoint (ourEndPoint, theirEndPoint) =

875 modifyMVar_ ourState $ \st -> case st of

874 modifyMVar_ ourState $ \st -> case st of

876 LocalEndPointValid vst ->

875 LocalEndPointValid vst ->

877 case vst ^. localConnectionTo theirAddress of

876 case vst ^. localConnectionTo theirAddress of

@@ -956,7 +955,7 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do

956

955

957-- | Send a CloseSocket request if the remote endpoint is unused

956-- | Send a CloseSocket request if the remote endpoint is unused

958closeIfUnused :: EndPointPair -> IO ()

957closeIfUnused :: EndPointPair -> IO ()

959−closeIfUnused (ourEndPoint, theirEndPoint) =

958+closeIfUnused (ourEndPoint, theirEndPoint) =

960 modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

959 modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity

961 { caseValid = \vst ->

960 { caseValid = \vst ->

962 if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)

961 if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)

network-transport/tests/TestAuxiliary.hs+13−2

@@ -5,10 +5,11 @@ module TestAuxiliary ( -- Running tests

5 -- Writing tests

6 , forkTry

7 , trySome

8+ , randomThreadDelay

8 ) where

9 ) where

10import Prelude hiding (catch)

11import Prelude hiding (catch)

11−import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo)

12+import Control.Concurrent (myThreadId, forkIO, ThreadId, throwTo, threadDelay)

12import Control.Concurrent.Chan (Chan)

13import Control.Concurrent.Chan (Chan)

13import Control.Monad (liftM2, unless)

14import Control.Monad (liftM2, unless)

14import Control.Exception (SomeException, try, catch)

15import Control.Exception (SomeException, try, catch)

@@ -20,6 +21,7 @@ import System.Console.ANSI ( SGR(SetColor, Reset)

20 , ColorIntensity(Vivid)

21 , ColorIntensity(Vivid)

21 , setSGR

22 , setSGR

22 )

23 )

24+import System.Random (randomIO)

23import Network.Transport

25import Network.Transport

24import Traced (Traceable(..), traceShow)

26import Traced (Traceable(..), traceShow)

@@ -38,7 +40,7 @@ runTest :: String -> IO () -> IO Bool

38runTest description test = do

40runTest description test = do

39 putStr $ "Running " ++ show description ++ ": "

41 putStr $ "Running " ++ show description ++ ": "

40 hFlush stdout

42 hFlush stdout

41− done <- try . timeout 20000000 $ test -- 20 seconds

43+ done <- try . timeout 30000000 $ test -- 30 seconds

42 case done of

44 case done of

43 Left err -> failed $ "(exception: " ++ show (err :: SomeException) ++ ")"

45 Left err -> failed $ "(exception: " ++ show (err :: SomeException) ++ ")"

44 Right Nothing -> failed $ "(timeout)"

46 Right Nothing -> failed $ "(timeout)"

@@ -65,6 +67,12 @@ runTests tests = do

65 success <- foldr (liftM2 (&&) . uncurry runTest) (return True) $ tests

67 success <- foldr (liftM2 (&&) . uncurry runTest) (return True) $ tests

66 unless success $ fail "Some tests failed"

68 unless success $ fail "Some tests failed"

70+-- | Random thread delay between 0 and the specified max

71+randomThreadDelay :: Int -> IO ()

72+randomThreadDelay maxDelay = do

73+ delay <- randomIO :: IO Int

74+ threadDelay (delay `mod` maxDelay)

75+

68--------------------------------------------------------------------------------

76--------------------------------------------------------------------------------

69-- traceShow instances --

77-- traceShow instances --

70--------------------------------------------------------------------------------

78--------------------------------------------------------------------------------

@@ -95,3 +103,6 @@ instance Traceable ThreadId where

103

96instance Traceable (Chan a) where

104instance Traceable (Chan a) where

97 trace = const Nothing

105 trace = const Nothing

106+

107+instance Traceable Float where

108+ trace = traceShow

network-transport/tests/TestTCP.hs+1−1

@@ -588,7 +588,7 @@ testReconnect nextPort = do

588 Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints

589

590 -- But a send will fail because the server has closed the connection again

591− Left (TransportError SendClosed _) <- send conn1 ["ping"]

591+ Left (TransportError SendFailed _) <- send conn1 ["ping"]

592 ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint

593

594 -- But a subsequent call to connect should reestablish the connection

network-transport/tests/TestTransport.hs+54−3

@@ -2,20 +2,23 @@

2module TestTransport where

4import Prelude hiding (catch, (>>=), (>>), return, fail)

5−import TestAuxiliary (forkTry, runTests, trySome)

5+import TestAuxiliary (forkTry, runTests, trySome, randomThreadDelay)

6+import Control.Concurrent (forkIO, killThread)

6import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)

7import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)

7import Control.Exception (evaluate, throw)

8import Control.Exception (evaluate, throw)

8−import Control.Monad (replicateM, replicateM_, when, guard, forM_)

9+import Control.Monad (replicateM, replicateM_, when, guard, forM, forM_)

9import Control.Monad.Error ()

10import Control.Monad.Error ()

10import Network.Transport hiding (connect)

11import Network.Transport hiding (connect)

11import qualified Network.Transport as NT

12import qualified Network.Transport as NT

12import Network.Transport.Internal (tlog)

13import Network.Transport.Internal (tlog)

13import Network.Transport.Util (spawn)

14import Network.Transport.Util (spawn)

15+import System.Random (randomIO)

14import Data.ByteString (ByteString)

16import Data.ByteString (ByteString)

15import Data.ByteString.Char8 (pack)

17import Data.ByteString.Char8 (pack)

16import Data.Map (Map)

18import Data.Map (Map)

17import qualified Data.Map as Map (empty, insert, delete, findWithDefault, adjust, null, toList, map)

19import qualified Data.Map as Map (empty, insert, delete, findWithDefault, adjust, null, toList, map)

18import Data.String (fromString)

20import Data.String (fromString)

21+import Data.Maybe (catMaybes)

19import Traced

22import Traced

21-- | We overload connect to always pass the default hints

24-- | We overload connect to always pass the default hints

@@ -648,9 +651,9 @@ testCloseTransport newTransport = do

648

651

649 -- Client now closes down its transport. We should receive connection closed messages

652 -- Client now closes down its transport. We should receive connection closed messages

650 -- TODO: this assumes a certain ordering on the messages we receive; that's not guaranteed

653 -- TODO: this assumes a certain ordering on the messages we receive; that's not guaranteed

654+ ConnectionClosed cid1' <- receive endpoint ; True <- return $ cid1' == cid1

651 ConnectionClosed cid2'' <- receive endpoint ; True <- return $ cid2'' == cid2

655 ConnectionClosed cid2'' <- receive endpoint ; True <- return $ cid2'' == cid2

652 ErrorEvent (TransportError (EventConnectionLost addr'' []) _) <- receive endpoint ; True <- return $ addr'' == theirAddr2

656 ErrorEvent (TransportError (EventConnectionLost addr'' []) _) <- receive endpoint ; True <- return $ addr'' == theirAddr2

653− ConnectionClosed cid1' <- receive endpoint ; True <- return $ cid1' == cid1

654

657

655 -- An attempt to send to the endpoint should now fail

658 -- An attempt to send to the endpoint should now fail

656 Left (TransportError SendClosed _) <- send conn ["pong2"]

659 Left (TransportError SendClosed _) <- send conn ["pong2"]

@@ -755,6 +758,7 @@ testExceptionOnReceive newTransport = do

755

758

756 return ()

759 return ()

757

760

761+-- | Test what happens when the argument to 'send' is an exceptional value

758testSendException :: IO (Either String Transport) -> IO ()

762testSendException :: IO (Either String Transport) -> IO ()

759testSendException newTransport = do

763testSendException newTransport = do

760 Right transport <- newTransport

764 Right transport <- newTransport

@@ -785,6 +789,52 @@ testSendException newTransport = do

785

789

786 return ()

790 return ()

787

791

792+-- | If threads get killed while executing a 'connect', 'send', or 'close', this

793+-- should not affect other threads.

794+--

795+-- The intention of this test is to see what happens when a asynchronous

796+-- exception happes _while executing a send, connect, or close_. This is

797+-- exceedingly difficult to guarantee, however. Hence we run a large number of

798+-- tests and insert random thread delays -- and even then it might not happen.

799+-- Moreover, it will only happen when we run on multiple cores.

800+testKill :: IO (Either String Transport) -> Int -> IO ()

801+testKill newTransport numThreads = do

802+ Right transport1 <- newTransport

803+ Right transport2 <- newTransport

804+ Right endpoint1 <- newEndPoint transport1

805+ Right endpoint2 <- newEndPoint transport2

806+

807+ threads <- forM [1 .. numThreads] $ \_ -> do

808+ done <- newEmptyMVar

809+ tid <- forkIO $ do

810+ randomThreadDelay 10

811+ Right conn <- connect endpoint1 (address endpoint2) ReliableOrdered

812+ randomThreadDelay 10

813+ Right () <- send conn ["ping"]

814+ randomThreadDelay 10

815+ close conn

816+ putMVar done ()

817+

818+ return (tid, done)

819+

820+ -- Kill half of those threads, and wait on the rest

821+ killerDone <- newEmptyMVar

822+ forkIO $ do

823+ wait <- forM threads $ \(tid, done) -> do

824+ shouldKill <- randomIO

825+ if shouldKill

826+ then do

827+ randomThreadDelay 30

828+ killThread tid

829+ return Nothing

830+ else

831+ return (Just done)

832+

833+ mapM_ takeMVar (catMaybes wait)

834+ putMVar killerDone ()

835+

836+ takeMVar killerDone

837+

788-- Transport tests

838-- Transport tests

789testTransport :: IO (Either String Transport) -> IO ()

839testTransport :: IO (Either String Transport) -> IO ()

790testTransport newTransport = do

840testTransport newTransport = do

@@ -806,6 +856,7 @@ testTransport newTransport = do

806 , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)

856 , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)

807 , ("ExceptionOnReceive", testExceptionOnReceive newTransport)

857 , ("ExceptionOnReceive", testExceptionOnReceive newTransport)

808 , ("SendException", testSendException newTransport)

858 , ("SendException", testSendException newTransport)

859+ , ("Kill", testKill newTransport 100000)

809 ]

860 ]

810 where

861 where

811 numPings = 10000 :: Int

862 numPings = 10000 :: Int