@@ -74,7 +74,7 @@ import Control.Concurrent.MVar ( MVar
74import Control.Category ((>>>))
74import Control.Category ((>>>))
75import Control.Applicative ((<$>))
75import Control.Applicative ((<$>))
76import Control.Monad (forM_, when, unless)
76import Control.Monad (forM_, when, unless)
77−import Control.Exception (IOException, SomeException, handle, throw, try, bracketOnError)
77+import Control.Exception (IOException, SomeException, handle, throw, try, bracketOnError, mask_)
78import Data.IORef (IORef, newIORef, writeIORef, readIORef)
78import Data.IORef (IORef, newIORef, writeIORef, readIORef)
79import Data.ByteString (ByteString)
79import Data.ByteString (ByteString)
80import qualified Data.ByteString as BS (concat)
80import qualified Data.ByteString as BS (concat)
@@ -628,8 +628,9 @@ requestConnectionTo ourEndPoint theirAddress = go
628 RemoteEndPointClosed -> do
628 RemoteEndPointClosed -> do
629 -- EndPointClosed indicates that a concurrent thread was in the
629 -- EndPointClosed indicates that a concurrent thread was in the
630 -- process of closing the TCP connection to the remote endpoint when
630 -- process of closing the TCP connection to the remote endpoint when
631− -- we obtained a reference to it. The remote endpoint will now have
631+ -- we obtained a reference to it. By INV-CLOSE we can assume that the
632− -- been removed from ourState, so we simply try again.
632+ -- remote endpoint will now have been removed from ourState, so we
633+ -- simply try again.
635 RemoteEndPointValid _ -> do
636 RemoteEndPointValid _ -> do
@@ -641,7 +642,6 @@ requestConnectionTo ourEndPoint theirAddress = go
641 failureHandler err = do
642 failureHandler err = do
642 modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
643 modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
643 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ ep)
644 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ ep)
644− -- TODO: should we call closeIfUnused here?
645 throw $ TransportError ConnectFailed (show err)
645 throw $ TransportError ConnectFailed (show err)
647 -- Do the actual connection request. This blocks until the remote
647 -- Do the actual connection request. This blocks until the remote
@@ -827,12 +827,23 @@ closeIfUnused theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ \st ->
827-- spawned). Returns whether or not a thread was spawned.
827-- spawned). Returns whether or not a thread was spawned.
828forkEndPointThread :: LocalEndPoint -> IO () -> IO Bool
828forkEndPointThread :: LocalEndPoint -> IO () -> IO Bool
829forkEndPointThread ourEndPoint p =
829forkEndPointThread ourEndPoint p =
830+ -- We use an explicit mask_ because we don't want to be interrupted until
831+ -- we have registered the thread. In particular, modifyMVar is not good
832+ -- enough because if we get an asynchronous exception after the fork but
833+ -- before the argument to modifyMVar returns we don't want to simply put
834+ -- the old value of the mvar back.
836+ st <- takeMVar ourState
830− modifyMVar ourState $ \st -> case st of
837+ case st of
831− LocalEndPointValid vst -> do
838+ LocalEndPointValid vst -> do
839+ threadRegistered <- newEmptyMVar
832− tid <- forkIO (p >> removeThread)
840+ tid <- forkIO (takeMVar threadRegistered >> p >> removeThread)
833− return (LocalEndPointValid . (internalThreads ^: (tid :)) $ vst, True)
841+ putMVar ourState $ LocalEndPointValid . (internalThreads ^: (tid :)) $ vst
842+ putMVar threadRegistered ()
834− LocalEndPointClosed ->
844+ LocalEndPointClosed -> do
835− return (LocalEndPointClosed, False)
846+ return False
837 removeThread :: IO ()
848 removeThread :: IO ()
838 removeThread = do
849 removeThread = do