4191d06 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Simplify some of the code using non-interruptable assumptions

4191d06Edsko de Vries14 years ago

network-transport/src/Network/Transport/TCP.hs+89−151

compare with master

unifiedsplit

@@ -72,7 +72,6 @@ import Control.Concurrent.MVar ( MVar

72 , putMVar

73 , newEmptyMVar

74 , withMVar

75− , tryPutMVar

76 )

75 )

77import Control.Category ((>>>))

76import Control.Category ((>>>))

78import Control.Applicative ((<$>))

77import Control.Applicative ((<$>))

@@ -85,7 +84,6 @@ import Control.Exception ( IOException

85 , try

84 , try

86 , bracketOnError

85 , bracketOnError

87 , mask

86 , mask

88− , mask_

89 , onException

87 , onException

90 , fromException

88 , fromException

91 )

89 )

@@ -247,7 +245,6 @@ data LocalEndPointState =

247data ValidLocalEndPointState = ValidLocalEndPointState

245data ValidLocalEndPointState = ValidLocalEndPointState

248 { _nextConnectionId :: !ConnectionId

246 { _nextConnectionId :: !ConnectionId

249 , _localConnections :: Map EndPointAddress RemoteEndPoint

247 , _localConnections :: Map EndPointAddress RemoteEndPoint

250− , _internalThreads :: [ThreadId]

251 , _nextRemoteId :: !Int

248 , _nextRemoteId :: !Int

252 }

249 }

253

250

@@ -529,7 +526,7 @@ apiConnect params ourEndPoint theirAddress _reliability hints =

529 else try . asyncWhenCancelled close $ do

526 else try . asyncWhenCancelled close $ do

530 resetIfBroken ourEndPoint theirAddress

527 resetIfBroken ourEndPoint theirAddress

531 (theirEndPoint, connId) <-

528 (theirEndPoint, connId) <-

532− requestConnectionTo params ourEndPoint theirAddress hints

529+ uRequestConnectionTo params ourEndPoint theirAddress hints

533 -- connAlive can be an IORef rather than an MVar because it is protected

530 -- connAlive can be an IORef rather than an MVar because it is protected

534 -- by the remoteState MVar. We don't need the overhead of locking twice.

531 -- by the remoteState MVar. We don't need the overhead of locking twice.

535 connAlive <- newIORef True

532 connAlive <- newIORef True

@@ -611,9 +608,7 @@ apiCloseEndPoint transport evs ourEndPoint = do

611 LocalEndPointClosed ->

608 LocalEndPointClosed ->

612 return (LocalEndPointClosed, Nothing)

609 return (LocalEndPointClosed, Nothing)

613 forM_ mOurState $ \vst -> do

610 forM_ mOurState $ \vst -> do

614− -- Close all endpoints and kill all threads

615 forM_ (vst ^. localConnections) tryCloseRemoteSocket

611 forM_ (vst ^. localConnections) tryCloseRemoteSocket

616− forM_ (vst ^. internalThreads) killThread

617 forM_ evs $ writeChan (localChannel ourEndPoint)

612 forM_ evs $ writeChan (localChannel ourEndPoint)

618 where

613 where

619 -- Close the remote socket and return the set of all incoming connections

614 -- Close the remote socket and return the set of all incoming connections

@@ -762,7 +757,6 @@ createLocalEndPoint transport = do

762 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

757 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState

763 { _nextConnectionId = firstNonReservedConnectionId

758 { _nextConnectionId = firstNonReservedConnectionId

764 , _localConnections = Map.empty

759 , _localConnections = Map.empty

765− , _internalThreads = []

766 , _nextRemoteId = 0

760 , _nextRemoteId = 0

767 }

761 }

768 modifyMVar (transportState transport) $ \st -> case st of

762 modifyMVar (transportState transport) $ \st -> case st of

@@ -784,104 +778,6 @@ createLocalEndPoint transport = do

784 TransportClosed ->

778 TransportClosed ->

785 throwIO (TransportError NewEndPointFailed "Transport closed")

779 throwIO (TransportError NewEndPointFailed "Transport closed")

786

780

787−-- | Request a connection to a remote endpoint

788−--

789−-- This will block until we get a connection ID from the remote endpoint; if

790−-- the remote endpoint was in 'RemoteEndPointClosing' state then we will

791−-- additionally block until that is resolved.

792−--

793−-- May throw a TransportError ConnectErrorCode exception.

794−requestConnectionTo :: TCPParameters

795− -> LocalEndPoint

796− -> EndPointAddress

797− -> ConnectHints

798− -> IO (RemoteEndPoint, ConnectionId)

799−requestConnectionTo params ourEndPoint theirAddress hints = go

800− where

801− go = do

802− (theirEndPoint, isNew) <- mapIOException connectFailed $

803− findRemoteEndPoint ourEndPoint theirAddress RequestedByUs

804−

805− if isNew

806− then do

807− void . forkEndPointThread ourEndPoint . handle absorbAllExceptions $

808− setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

809− go

810− else do

811− reply <- mapIOException connectFailed $

812− doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

813− return (theirEndPoint, decodeInt32 . BS.concat $ reply)

814−

815− connectFailed = TransportError ConnectFailed . show

816−

817− absorbAllExceptions :: SomeException -> IO ()

818− absorbAllExceptions _ex =

819− return ()

820−

821−-- | Set up a remote endpoint

822−setupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()

823−setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

824− didAccept <- bracketOnError (socketToEndPoint

825− ourAddress

826− theirAddress

827− (tcpReuseClientAddr params)

828− (connectTimeout hints))

829− onError $ \result -> case result of

830− Right (sock, ConnectionRequestAccepted) -> do

831− let vst = ValidRemoteEndPointState

832− { remoteSocket = sock

833− , _remoteOutgoing = 0

834− , _remoteIncoming = IntSet.empty

835− , sendOn = sendMany sock

836− , _pendingCtrlRequests = IntMap.empty

837− , _nextCtrlRequestId = 0

838− }

839− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

840− return True

841− Right (sock, ConnectionRequestInvalid) -> do

842− let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"

843− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

844− tryCloseSocket sock

845− return False

846− Right (sock, ConnectionRequestCrossed) -> do

847− resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed

848− tryCloseSocket sock

849− return False

850− Left err -> do

851− resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

852− return False

853−

854− -- If we get to this point without an exception, then

855− --

856− -- if didAccept is False the socket has already been closed

857− -- if didAccept is True, the socket has been stored as part of the remote

858− -- state so we no longer need to worry about closing it when an

859− -- asynchronous exception occurs

860− when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)

861− where

862− -- If an asynchronous exception occurs while we set up the remote endpoint

863− -- we need to make sure to close the socket. It is also useful to

864− -- initialize the remote state (to "invalid") so that concurrent threads

865− -- that are blocked on reading the remote state are unblocked. It is

866− -- possible, however, that the exception occurred after we already

867− -- initialized the remote state, which is why we use tryPutMVar here.

868− onError :: Either (TransportError ConnectErrorCode)

869− (N.Socket, ConnectionRequestResponse)

870− -> IO ()

871− onError result =

872− case result of

873− Left err ->

874− void $ tryPutMVar theirState (RemoteEndPointInvalid err)

875− Right (sock, _) -> do

876− let err = failed "setupRemoteEndPoint failed"

877− tryPutMVar theirState (RemoteEndPointInvalid err)

878− tryCloseSocket sock

879−

880− failed = TransportError ConnectFailed

881− ourAddress = localAddress ourEndPoint

882− theirAddress = remoteAddress theirEndPoint

883− theirState = remoteState theirEndPoint

884− invalidAddress = TransportError ConnectNotFound

885

781

886-- | Resolve an endpoint currently in 'Init' state

782-- | Resolve an endpoint currently in 'Init' state

887resolveInit :: EndPointPair -> RemoteState -> IO ()

783resolveInit :: EndPointPair -> RemoteState -> IO ()

@@ -1043,48 +939,6 @@ closeIfUnused (ourEndPoint, theirEndPoint) =

1043 return $ RemoteEndPointValid vst

939 return $ RemoteEndPointValid vst

1044 }

940 }

1045

941

1046−-- | Fork a new thread and store its ID as part of the transport state

1047−--

1048−-- If the local end point is closed this function does nothing (no thread is

1049−-- spawned). Returns whether or not a thread was spawned.

1050−forkEndPointThread :: LocalEndPoint -> IO () -> IO Bool

1051−forkEndPointThread ourEndPoint p =

1052− -- We use an explicit mask_ because we don't want to be interrupted until

1053− -- we have registered the thread. In particular, modifyMVar is not good

1054− -- enough because if we get an asynchronous exception after the fork but

1055− -- before the argument to modifyMVar returns we don't want to simply put

1056− -- the old value of the mvar back.

1057− mask_ $ do

1058− st <- takeMVar ourState

1059− case st of

1060− LocalEndPointValid vst -> do

1061− threadRegistered <- newEmptyMVar

1062− tid <- forkIO (takeMVar threadRegistered >> p >> removeThread)

1063− putMVar ourState ( LocalEndPointValid

1064− . (internalThreads ^: (tid :))

1065− $ vst

1066− )

1067− putMVar threadRegistered ()

1068− return True

1069− LocalEndPointClosed -> do

1070− putMVar ourState st

1071− return False

1072− where

1073− removeThread :: IO ()

1074− removeThread = do

1075− tid <- myThreadId

1076− modifyMVar_ ourState $ \st -> case st of

1077− LocalEndPointValid vst ->

1078− return ( LocalEndPointValid

1079− . (internalThreads ^: filter (/= tid))

1080− $ vst

1081− )

1082− LocalEndPointClosed ->

1083− return LocalEndPointClosed

1084−

1085− ourState :: MVar LocalEndPointState

1086− ourState = localState ourEndPoint

1087−

1088--------------------------------------------------------------------------------

942--------------------------------------------------------------------------------

1089-- As soon as a remote connection fails, we want to put notify our endpoint --

943-- As soon as a remote connection fails, we want to put notify our endpoint --

1090-- and put it into a closed state. Since this may happen in many places, we --

944-- and put it into a closed state. Since this may happen in many places, we --

@@ -1248,7 +1102,7 @@ handleConnectionRequest transport sock = handle handleException $ do

1248 return ourEndPoint

1102 return ourEndPoint

1249 TransportClosed ->

1103 TransportClosed ->

1250 throwIO $ userError "Transport closed"

1104 throwIO $ userError "Transport closed"

1251− void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress

1105+ void . forkIO $ go ourEndPoint theirAddress

1252 where

1106 where

1253 go :: LocalEndPoint -> EndPointAddress -> IO ()

1107 go :: LocalEndPoint -> EndPointAddress -> IO ()

1254 go ourEndPoint theirAddress = do

1108 go ourEndPoint theirAddress = do

@@ -1626,6 +1480,93 @@ getNextConnectionId ourEndpoint =

1626 LocalEndPointClosed ->

1480 LocalEndPointClosed ->

1627 throwIO $ userError "Local endpoint closed"

1481 throwIO $ userError "Local endpoint closed"

1628

1482

1483+--------------------------------------------------------------------------------

1484+-- Uninterruptable auxiliary functions --

1485+-- --

1486+-- All these functions assume they are running in a thread which will never --

1487+-- be killed, and are designed to be used in conjunction with --

1488+-- 'asyncWhenCancelled'. To be explicit about this assumption the function --

1489+-- names are prefixed with 'u'. --

1490+--------------------------------------------------------------------------------

1491+

1492+-- | Request a connection to a remote endpoint

1493+--

1494+-- This will block until we get a connection ID from the remote endpoint; if

1495+-- the remote endpoint was in 'RemoteEndPointClosing' state then we will

1496+-- additionally block until that is resolved.

1497+--

1498+-- May throw a TransportError ConnectErrorCode exception.

1499+uRequestConnectionTo :: TCPParameters

1500+ -> LocalEndPoint

1501+ -> EndPointAddress

1502+ -> ConnectHints

1503+ -> IO (RemoteEndPoint, ConnectionId)

1504+uRequestConnectionTo params ourEndPoint theirAddress hints = go

1505+ where

1506+ go = do

1507+ (theirEndPoint, isNew) <- mapIOException connectFailed $

1508+ findRemoteEndPoint ourEndPoint theirAddress RequestedByUs

1509+

1510+ if isNew

1511+ then do

1512+ forkIO . handle absorbAllExceptions $

1513+ uSetupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints

1514+ go

1515+ else do

1516+ reply <- mapIOException connectFailed $

1517+ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

1518+ return (theirEndPoint, decodeInt32 . BS.concat $ reply)

1519+

1520+ connectFailed = TransportError ConnectFailed . show

1521+

1522+ absorbAllExceptions :: SomeException -> IO ()

1523+ absorbAllExceptions _ex =

1524+ return ()

1525+

1526+-- | Set up a remote endpoint

1527+uSetupRemoteEndPoint :: TCPParameters -> EndPointPair -> ConnectHints -> IO ()

1528+uSetupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

1529+ result <- socketToEndPoint ourAddress

1530+ theirAddress

1531+ (tcpReuseClientAddr params)

1532+ (connectTimeout hints)

1533+ didAccept <- case result of

1534+ Right (sock, ConnectionRequestAccepted) -> do

1535+ let vst = ValidRemoteEndPointState

1536+ { remoteSocket = sock

1537+ , _remoteOutgoing = 0

1538+ , _remoteIncoming = IntSet.empty

1539+ , sendOn = sendMany sock

1540+ , _pendingCtrlRequests = IntMap.empty

1541+ , _nextCtrlRequestId = 0

1542+ }

1543+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointValid vst)

1544+ return True

1545+ Right (sock, ConnectionRequestInvalid) -> do

1546+ let err = invalidAddress "setupRemoteEndPoint: Invalid endpoint"

1547+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

1548+ tryCloseSocket sock

1549+ return False

1550+ Right (sock, ConnectionRequestCrossed) -> do

1551+ resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed

1552+ tryCloseSocket sock

1553+ return False

1554+ Left err -> do

1555+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointInvalid err)

1556+ return False

1557+

1558+ -- If we get to this point without an exception, then

1559+ --

1560+ -- if didAccept is False the socket has already been closed

1561+ -- if didAccept is True, the socket has been stored as part of the remote

1562+ -- state so we no longer need to worry about closing it when an

1563+ -- asynchronous exception occurs

1564+ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)

1565+ where

1566+ ourAddress = localAddress ourEndPoint

1567+ theirAddress = remoteAddress theirEndPoint

1568+ invalidAddress = TransportError ConnectNotFound

1569+

1629--------------------------------------------------------------------------------

1570--------------------------------------------------------------------------------

1630-- Constants --

1571-- Constants --

1631--------------------------------------------------------------------------------

1572--------------------------------------------------------------------------------

@@ -1650,9 +1591,6 @@ nextConnectionId = accessor _nextConnectionId (\cix st -> st { _nextConnectionId

1650localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)

1591localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)

1651localConnections = accessor _localConnections (\es st -> st { _localConnections = es })

1592localConnections = accessor _localConnections (\es st -> st { _localConnections = es })

1652

1593

1653−internalThreads :: Accessor ValidLocalEndPointState [ThreadId]

1654−internalThreads = accessor _internalThreads (\ts st -> st { _internalThreads = ts })

1655−

1656nextRemoteId :: Accessor ValidLocalEndPointState Int

1594nextRemoteId :: Accessor ValidLocalEndPointState Int

1657nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })

1595nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })

1658

1596

network-transport/tests/TestTransport.hs+9−4

@@ -302,7 +302,7 @@ collect endPoint maxEvents timeout = go maxEvents Map.empty Map.empty

302 finish open closed =

303 if Map.null open

304 then return . Map.toList . Map.map reverse $ closed

305− else fail "Open connections"

305+ else fail $ "Open connections: " ++ show (map fst . Map.toList $ open)

306

307-- | Open connection, close it, then reopen it

308-- (In the TCP transport this means the socket will be closed, then reopened)

@@ -837,8 +837,13 @@ testKill newTransport numThreads = do

837 threads <- replicateM numThreads . forkIO $ do

838 randomThreadDelay 100

839 bracket (connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints)

840+ -- Note that we should not insert a randomThreadDelay into the

841+ -- exception handler itself as this means that the exception handler

842+ -- could be interrupted and we might not close

840− (\(Right conn) -> randomThreadDelay 100 >> close conn)

843+ (\(Right conn) -> close conn)

844+ (\(Right conn) -> do randomThreadDelay 100

841− (\(Right conn) -> randomThreadDelay 100 >> do Right () <- send conn ["ping"] ; return ())

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

846+ randomThreadDelay 100)

842

847

843 numAlive <- newMVar (0 :: Int)

848 numAlive <- newMVar (0 :: Int)

844

849

@@ -935,7 +940,7 @@ testTransport newTransport = do

935 , ("CloseReopen", testCloseReopen transport numPings)

940 , ("CloseReopen", testCloseReopen transport numPings)

936 , ("ParallelConnects", testParallelConnects transport numPings)

941 , ("ParallelConnects", testParallelConnects transport numPings)

937 , ("SendAfterClose", testSendAfterClose transport 1000)

942 , ("SendAfterClose", testSendAfterClose transport 1000)

938− , ("Crossing", testCrossing transport 1000)

943+ , ("Crossing", testCrossing transport 100)

939 , ("CloseTwice", testCloseTwice transport 100)

944 , ("CloseTwice", testCloseTwice transport 100)

940 , ("ConnectToSelf", testConnectToSelf transport numPings)

945 , ("ConnectToSelf", testConnectToSelf transport numPings)

941 , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

946 , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)

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

@@ -176,7 +176,7 @@ instance MonadS IO where

176

177instance Show TracedException where

178 show (TracedException ts ex) =

179− show ex ++ "\nTrace:\n" ++ unlines (map (\(i, t) -> show i ++ "\t" ++ t) (zip ([0..] :: [Int]) (reverse ts)))

179+ show ex ++ "\nTrace:\n" ++ unlines (map (\(i, t) -> show i ++ "\t" ++ t) (zip ([0..] :: [Int]) (take 10 . reverse $ ts)))

180

181traceHandlers :: Traceable a => a -> [Handler b]

182traceHandlers a = case trace a of