git.iko.soy/distributed-process
on master help sign in
Error handling
a2443a3Edsko de Vries
network-transport/src/Network/Transport/TCP.hs+55−26
compare with master
unifiedsplit
@@ -70,6 +70,7 @@ import Control.Concurrent.MVar ( MVar
70 , putMVar
70 , putMVar
71 , newEmptyMVar
71 , newEmptyMVar
72 , withMVar
72 , withMVar
73+ , tryPutMVar
73 )
74 )
74import Control.Category ((>>>))
75import Control.Category ((>>>))
75import Control.Applicative ((<$>))
76import Control.Applicative ((<$>))
@@ -691,33 +692,61 @@ requestConnectionTo ourEndPoint theirAddress = go
691-- RemoteEndPointInvalid.
692-- RemoteEndPointInvalid.
692setupRemoteEndPoint :: EndPointPair -> IO ()
693setupRemoteEndPoint :: EndPointPair -> IO ()
693setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
694setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
694− result <- socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint)
695+ didAccept <- bracketOnError (socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint))
696+ onError $ \result -> do
695− case result of
697+ case result of
696− Right (sock, ConnectionRequestAccepted) -> do
698+ Right (sock, ConnectionRequestAccepted) -> do
697− let vst = ValidRemoteEndPointState
699+ let vst = ValidRemoteEndPointState
698− { remoteSocket = sock
700+ { remoteSocket = sock
699− , _remoteOutgoing = 0
701+ , _remoteOutgoing = 0
700− , _remoteIncoming = IntSet.empty
702+ , _remoteIncoming = IntSet.empty
701− , sendOn = sendMany sock
703+ , sendOn = sendMany sock
702− }
704+ }
703− putMVar theirState (RemoteEndPointValid vst)
705+ putMVar theirState (RemoteEndPointValid vst)
704− handleIncomingMessages (ourEndPoint, theirEndPoint)
706+ return True
705− Right (sock, ConnectionRequestEndPointInvalid) -> do
707+ Right (sock, ConnectionRequestEndPointInvalid) -> do
706− -- We remove the endpoint from our local state again because the next
708+ -- We remove the endpoint from our local state again because the next
707− -- call to 'connect' might give a different result. Threads that were
709+ -- call to 'connect' might give a different result. Threads that were
708− -- waiting on the result of this call to connect will get the
710+ -- waiting on the result of this call to connect will get the
709− -- RemoteEndPointInvalid; subsequent threads will initiate a new
711+ -- RemoteEndPointInvalid; subsequent threads will initiate a new
710− -- connection requests.
712+ -- connection requests.
711− removeRemoteEndPoint (ourEndPoint, theirEndPoint)
713+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
712− putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
714+ putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
713− N.sClose sock
715+ N.sClose sock
716+ return False
714− Right (sock, ConnectionRequestCrossed) -> do
717+ Right (sock, ConnectionRequestCrossed) -> do
715− N.sClose sock
718+ N.sClose sock
719+ return False
716− Left err -> do
720+ Left err -> do
717− -- See comment above
721+ removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above
718− removeRemoteEndPoint (ourEndPoint, theirEndPoint)
722+ putMVar theirState (RemoteEndPointInvalid err)
719− putMVar theirState (RemoteEndPointInvalid err)
723+ return False
724+
725+ -- If we get to this point without an exception, then
726+ -- * if didAccept is False the socket has already been closed
727+ -- * if didAccept is True, the socket has been stored as part of the remote
728+ -- state so we no longer need to worry about closing it when an
729+ -- asynchronous exception occurs
730+ when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
720 where
731 where
732+ -- If an asynchronous exception occurs while we set up the remote endpoint
733+ -- we need to make sure to close the socket. It is also useful to
734+ -- initialize the remote state (to "invalid") so that concurrent threads
735+ -- that are blocked on reading the remote state are unblocked. It is
736+ -- possible, however, that the exception occurred after we already
737+ -- initialized the remote state, which is why we use tryPutMVar here.
738+ onError :: Either (TransportError ConnectErrorCode) (N.Socket, ConnectionRequestResponse) -> IO ()
739+ onError result = do
740+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
741+ case result of
742+ Left err -> do
743+ tryPutMVar theirState (RemoteEndPointInvalid (TransportError ConnectFailed (show err)))
744+ return ()
745+ Right (sock, _) -> do
746+ tryPutMVar theirState (RemoteEndPointInvalid (TransportError ConnectFailed "setupRemoteEndPoint failed"))
747+ tryIO $ N.sClose sock
748+ return ()
749+
721 theirState = remoteState theirEndPoint
750 theirState = remoteState theirEndPoint
722 invalidAddress = TransportError ConnectNotFound
751 invalidAddress = TransportError ConnectNotFound
723
752