git.iko.soy/distributed-process
on master help sign in
Error handling
commit2a0ef88ad18ec5b38f6aa11ca3a0d4b04abfbe97 authorEdsko de Vries <edsko@edsko.net> dateMon 21 May 2012, 13:10 (14 years ago) parente9af460
2 files changed+59−12
compare with master
unifiedsplit
network-transport/src/Network/Transport/TCP.hs+21−10 network-transport/tests/TestTCP.hs+38−2
network-transport/src/Network/Transport/TCP.hs+21−10
@@ -74,7 +74,7 @@ import Control.Concurrent.MVar ( MVar
7474import Control.Category ((>>>))
7575import Control.Applicative ((<$>))
7676import 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_)
7878import Data.IORef (IORef, newIORef, writeIORef, readIORef)
7979import Data.ByteString (ByteString)
8080import qualified Data.ByteString as BS (concat)
@@ -628,8 +628,9 @@ requestConnectionTo ourEndPoint theirAddress = go
628628 RemoteEndPointClosed -> do
629629 -- EndPointClosed indicates that a concurrent thread was in the
630630 -- process of closing the TCP connection to the remote endpoint when
631− -- we obtained a reference to it. The remote endpoint will now have
632− -- been removed from ourState, so we simply try again.
631+ -- we obtained a reference to it. By INV-CLOSE we can assume that the
632+ -- remote endpoint will now have been removed from ourState, so we
633+ -- simply try again.
633634 go
634635
635636 RemoteEndPointValid _ -> do
@@ -641,7 +642,6 @@ requestConnectionTo ourEndPoint theirAddress = go
641642 failureHandler err = do
642643 modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
643644 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ ep)
644− -- TODO: should we call closeIfUnused here?
645645 throw $ TransportError ConnectFailed (show err)
646646
647647 -- Do the actual connection request. This blocks until the remote
@@ -827,12 +827,23 @@ closeIfUnused theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ \st ->
827827-- spawned). Returns whether or not a thread was spawned.
828828forkEndPointThread :: LocalEndPoint -> IO () -> IO Bool
829829forkEndPointThread ourEndPoint p =
830− modifyMVar ourState $ \st -> case st of
831− LocalEndPointValid vst -> do
832− tid <- forkIO (p >> removeThread)
833− return (LocalEndPointValid . (internalThreads ^: (tid :)) $ vst, True)
834− LocalEndPointClosed ->
835− return (LocalEndPointClosed, False)
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.
835+ mask_ $ do
836+ st <- takeMVar ourState
837+ case st of
838+ LocalEndPointValid vst -> do
839+ threadRegistered <- newEmptyMVar
840+ tid <- forkIO (takeMVar threadRegistered >> p >> removeThread)
841+ putMVar ourState $ LocalEndPointValid . (internalThreads ^: (tid :)) $ vst
842+ putMVar threadRegistered ()
843+ return True
844+ LocalEndPointClosed -> do
845+ putMVar ourState st
846+ return False
836847 where
837848 removeThread :: IO ()
838849 removeThread = do
network-transport/tests/TestTCP.hs+38−2
@@ -507,6 +507,41 @@ testBreakTransport nextPort = do
507507
508508 return ()
509509
510+-- | Test that a second call to 'connect' might succeed even if the first
511+-- failed. This is a TCP specific test rather than an endpoint specific test
512+-- because we must manually create the endpoint address to match an endpoint we
513+-- have yet to set up
514+testReconnect :: IO N.ServiceName -> IO ()
515+testReconnect nextPort = do
516+ clientDone <- newEmptyMVar
517+ firstAttempt <- newEmptyMVar
518+ endpointCreated <- newEmptyMVar
519+ port <- nextPort
520+ Right transport <- createTransport "127.0.0.1" port
521+
522+ -- Server
523+ forkTry $ do
524+ takeMVar firstAttempt
525+ newEndPoint transport
526+ putMVar endpointCreated ()
527+
528+ -- Client
529+ forkTry $ do
530+ Right endpoint <- newEndPoint transport
531+ let theirAddr = encodeEndPointAddress "127.0.0.1" port 1
532+
533+ Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
534+ putMVar firstAttempt ()
535+
536+ takeMVar endpointCreated
537+ Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
538+
539+ putMVar clientDone ()
540+
541+ takeMVar clientDone
542+
543+
544+
510545main :: IO ()
511546main = do
512547 portMVar <- newEmptyMVar
@@ -520,7 +555,8 @@ main = do
520555 , ("TestUnnecessaryConnect", testUnnecessaryConnect nextPort)
521556 , ("InvalidAddress", testInvalidAddress nextPort)
522557 , ("InvalidConnect", testInvalidConnect nextPort)
523− , ("TestMany", testMany nextPort)
524− , ("TestBreakTransport", testBreakTransport nextPort)
558+ , ("Many", testMany nextPort)
559+ , ("BreakTransport", testBreakTransport nextPort)
560+ , ("Reconnect", testReconnect nextPort)
525561 ]
526562 testTransport (either (Left . show) (Right) <$> nextPort >>= createTransport "127.0.0.1")