@@ -72,7 +72,6 @@ import Control.Concurrent.MVar ( MVar
7272 , putMVar
7373 , newEmptyMVar
7474 , withMVar
75− , tryPutMVar
7675 )
7776import Control.Category ((>>>))
7877import Control.Applicative ((<$>))
@@ -85,7 +84,6 @@ import Control.Exception ( IOException
8584 , try
8685 , bracketOnError
8786 , mask
88− , mask_
8987 , onException
9088 , fromException
9189 )
@@ -247,7 +245,6 @@ data LocalEndPointState =
247245data ValidLocalEndPointState = ValidLocalEndPointState
248246 { _nextConnectionId :: !ConnectionId
249247 , _localConnections :: Map EndPointAddress RemoteEndPoint
250− , _internalThreads :: [ThreadId]
251248 , _nextRemoteId :: !Int
252249 }
253250
@@ -529,7 +526,7 @@ apiConnect params ourEndPoint theirAddress _reliability hints =
529526 else try . asyncWhenCancelled close $ do
530527 resetIfBroken ourEndPoint theirAddress
531528 (theirEndPoint, connId) <-
532− requestConnectionTo params ourEndPoint theirAddress hints
529+ uRequestConnectionTo params ourEndPoint theirAddress hints
533530 -- connAlive can be an IORef rather than an MVar because it is protected
534531 -- by the remoteState MVar. We don't need the overhead of locking twice.
535532 connAlive <- newIORef True
@@ -611,9 +608,7 @@ apiCloseEndPoint transport evs ourEndPoint = do
611608 LocalEndPointClosed ->
612609 return (LocalEndPointClosed, Nothing)
613610 forM_ mOurState $ \vst -> do
614− -- Close all endpoints and kill all threads
615611 forM_ (vst ^. localConnections) tryCloseRemoteSocket
616− forM_ (vst ^. internalThreads) killThread
617612 forM_ evs $ writeChan (localChannel ourEndPoint)
618613 where
619614 -- Close the remote socket and return the set of all incoming connections
@@ -762,7 +757,6 @@ createLocalEndPoint transport = do
762757 state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
763758 { _nextConnectionId = firstNonReservedConnectionId
764759 , _localConnections = Map.empty
765− , _internalThreads = []
766760 , _nextRemoteId = 0
767761 }
768762 modifyMVar (transportState transport) $ \st -> case st of
@@ -784,104 +778,6 @@ createLocalEndPoint transport = do
784778 TransportClosed ->
785779 throwIO (TransportError NewEndPointFailed "Transport closed")
786780
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
885781
886782-- | Resolve an endpoint currently in 'Init' state
887783resolveInit :: EndPointPair -> RemoteState -> IO ()
@@ -1043,48 +939,6 @@ closeIfUnused (ourEndPoint, theirEndPoint) =
1043939 return $ RemoteEndPointValid vst
1044940 }
1045941
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−
1088942--------------------------------------------------------------------------------
1089943-- As soon as a remote connection fails, we want to put notify our endpoint --
1090944-- 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
12481102 return ourEndPoint
12491103 TransportClosed ->
12501104 throwIO $ userError "Transport closed"
1251− void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress
1105+ void . forkIO $ go ourEndPoint theirAddress
12521106 where
12531107 go :: LocalEndPoint -> EndPointAddress -> IO ()
12541108 go ourEndPoint theirAddress = do
@@ -1626,6 +1480,93 @@ getNextConnectionId ourEndpoint =
16261480 LocalEndPointClosed ->
16271481 throwIO $ userError "Local endpoint closed"
16281482
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+
16291570--------------------------------------------------------------------------------
16301571-- Constants --
16311572--------------------------------------------------------------------------------
@@ -1650,9 +1591,6 @@ nextConnectionId = accessor _nextConnectionId (\cix st -> st { _nextConnectionId
16501591localConnections :: Accessor ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
16511592localConnections = accessor _localConnections (\es st -> st { _localConnections = es })
16521593
1653−internalThreads :: Accessor ValidLocalEndPointState [ThreadId]
1654−internalThreads = accessor _internalThreads (\ts st -> st { _internalThreads = ts })
1655−
16561594nextRemoteId :: Accessor ValidLocalEndPointState Int
16571595nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })
16581596