08ac22b · distributed-process

733requestConnectionTo ourEndPoint theirAddress = go

734 where

735 go = do

736− (theirEndPoint, isNew, _) <- mapIOException connectFailed $

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

737 findRemoteEndPoint ourEndPoint theirAddress Local

738

739 if isNew

774 Right (sock, ConnectionRequestCrossed) -> do

775 -- We leave the endpoint in Init state, handleConnectionRequest will

776 -- take care of it

777+ resolveInit (ourEndPoint, theirEndPoint) (RemoteEndPointClosed Nothing)

777 tryCloseSocket sock

778 tryCloseSocket sock

778 return False

779 return False

779 Left err -> do

780 Left err -> do

812 theirState = remoteState theirEndPoint

813 theirState = remoteState theirEndPoint

813 invalidAddress = TransportError ConnectNotFound

814 invalidAddress = TransportError ConnectNotFound

814

815

815−-- TODO: Document

816+-- | Resolve an endpoint currently in 'Init' state

816−-- TODO: is this what we want? modifyRemoteState only does something special

817−-- when an exception is thrown in the Valid case.

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

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

819resolveInit (ourEndPoint, theirEndPoint) newState =

818resolveInit (ourEndPoint, theirEndPoint) newState =

820− modifyRemoteState_ (ourEndPoint, theirEndPoint) $ remoteStateIdentity

819+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of

821− { caseInit = \resolved _ -> do

820+ RemoteEndPointInit resolved _ -> do

822− putMVar resolved ()

821+ putMVar resolved ()

823− return newState

822+ case newState of

824− , caseInvalid = \_ -> throwException

823+ RemoteEndPointClosed _ ->

825− , caseValid = \_ -> throwException

824+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)

826− , caseClosing = \_ _ -> throwException

825+ _ ->

827− , caseClosed = \_ -> throwException

826+ return ()

828− }

827+ return newState

829− where

828+ RemoteEndPointClosed (Just ex) ->

830− throwException = throwIO $ userError "resolveInit: not in Init state"

829+ throwIO ex

830+ _ ->

831+ relyViolation (ourEndPoint, theirEndPoint) "resolveInit"

831

832

832-- | Establish a connection to a remote endpoint

833-- | Establish a connection to a remote endpoint

833socketToEndPoint :: EndPointAddress -- ^ Our address

834socketToEndPoint :: EndPointAddress -- ^ Our address

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

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

1158 go ourEndPoint theirAddress = do

1159 go ourEndPoint theirAddress = do

1159 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do

1160 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do

1160− (theirEndPoint, _, crossed) <-

1161+ (theirEndPoint, isNew) <-

1161 findRemoteEndPoint ourEndPoint theirAddress Remote

1162 findRemoteEndPoint ourEndPoint theirAddress Remote

1162

1163

1163− if crossed

1164+ if not isNew

1164 then do

1165 then do

1165 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]

1166 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]

1166 tryCloseSocket sock

1167 tryCloseSocket sock

1193 tryCloseSocket sock

1194 tryCloseSocket sock

1194

1195

1195-- | Find a remote endpoint. If the remote endpoint does not yet exist we

1196-- | Find a remote endpoint. If the remote endpoint does not yet exist we

1196−-- create it in Init state. Returns if the endpoint was new and, for requests

1197+-- create it in Init state. Returns if the endpoint was new.

1197−-- of Remote origin, whether the request crossed.

1198findRemoteEndPoint

1199 :: LocalEndPoint

1200 -> EndPointAddress

1201 -> Origin

1202− -> IO (RemoteEndPoint, Bool, Bool)

1202+ -> IO (RemoteEndPoint, Bool)

1203findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

1204 where

1205 go = do

1226

1227 if isNew

1228 then

1229− return (theirEndPoint, True, False)

1229+ return (theirEndPoint, True)

1230 else do

1231 let theirState = remoteState theirEndPoint

1232 snapshot <- modifyMVar theirState $ \st -> case st of

1251 case (findOrigin, initOrigin) of

1252 (Local, Local) -> readMVar resolved >> go

1253 (Local, Remote) -> readMVar resolved >> go

1254+ (Remote, Local) -> if ourAddress > theirAddress

1255+ then

1256+ -- Wait for the Crossed message

1257+ readMVar resolved >> go

1258+ else

1254− (Remote, Local) -> return (theirEndPoint, False, ourAddress < theirAddress)

1259+ return (theirEndPoint, False)

1255 (Remote, Remote) -> throwIO $ userError "Already connected (B)"

1260 (Remote, Remote) -> throwIO $ userError "Already connected (B)"

1256 RemoteEndPointValid _ ->

1261 RemoteEndPointValid _ ->

1257 -- We assume that the request crossed if we find the endpoint in

1262 -- We assume that the request crossed if we find the endpoint in

1258 -- Valid state. It is possible that this is really an invalid

1263 -- Valid state. It is possible that this is really an invalid

1259 -- request, but only in the case of a broken client (we don't

1264 -- request, but only in the case of a broken client (we don't

1260 -- maintain enough history to be able to tell the difference).

1265 -- maintain enough history to be able to tell the difference).

1261− return (theirEndPoint, False, True)

1266+ return (theirEndPoint, False)

1262 RemoteEndPointClosing resolved _ ->

1267 RemoteEndPointClosing resolved _ ->

1263 readMVar resolved >> go

1268 readMVar resolved >> go

1264 RemoteEndPointClosed Nothing ->

1269 RemoteEndPointClosed Nothing ->