a1cbb75 · distributed-process

git.iko.soy/distributed-process

Optimize handling of 'Crossed' connection requests

When A tries to connect (set up a heavyweight connection to) B, while B tries to connect to A at the same time, there will be two threads trying to set up a heavyweight connection for A (and similarly for B): the thread that deals with the apiConnect, and the thread that deals with the incoming request from B. In the case that "B" wins and sends a ConnectionRequestCrossed message to A, the thread that daels with the apiConenct should simply wait until the handleIncomingRequest thread completes. We did this already, but it is important that the handleIncomingRequest thread *in B* (which loses to B's apiConnect thread) waits until A receives the ConnectionRequestCrossed message (to avoid certain ordering problems). We had implemented this by regarding an incoming ConnectionRequestCrossed message as changing the heavyweight connection from Init to Closed state (and signalling on resolved), but that's wrong as it will cause A to try and connect again. Now instead when we receive a ConnectionRequestCrossed message we signal on a (new) MVar "crossed" (part of the Init state), but leave the EndPoint in Init state (it will be resolved by the handleIncmoingMessage thread).

commita1cbb75dd450c6035b4a7f9e8244e49ca3afd921 authorEdsko de Vries <edsko@edsko.net> dateMon 17 Sep 2012, 11:53 (14 years ago) parent0d60744

1 file changed+25−17

compare with master

unifiedsplit

network-transport-tcp/src/Network/Transport/TCP.hs+25−17

network-transport-tcp/src/Network/Transport/TCP.hs+25−17

@@ -357,7 +357,7 @@ data RemoteState =

357357 -- | Invalid remote endpoint (for example, invalid address)

358358 RemoteEndPointInvalid !(TransportError ConnectErrorCode)

359359 -- | The remote endpoint is being initialized

360− | RemoteEndPointInit !(MVar ()) !RequestedBy

360+ | RemoteEndPointInit !(MVar ()) !(MVar ()) !RequestedBy

361361 -- | "Normal" working endpoint

362362 | RemoteEndPointValid !ValidRemoteEndPointState

363363 -- | The remote endpoint is being closed (garbage collected)

@@ -604,7 +604,7 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

604604 act <- withMVar (remoteState theirEndPoint) $ \st -> case st of

605605 RemoteEndPointInvalid _ ->

606606 relyViolation (ourEndPoint, theirEndPoint) "apiSend"

607− RemoteEndPointInit _ _ ->

607+ RemoteEndPointInit _ _ _ ->

608608 relyViolation (ourEndPoint, theirEndPoint) "apiSend"

609609 RemoteEndPointValid vst -> do

610610 alive <- readIORef connAlive

@@ -660,7 +660,7 @@ apiCloseEndPoint transport evs ourEndPoint =

660660 case st of

661661 RemoteEndPointInvalid _ ->

662662 return (st, Nothing)

663− RemoteEndPointInit resolved _ -> do

663+ RemoteEndPointInit resolved _ _ -> do

664664 putMVar resolved ()

665665 return (closed, Nothing)

666666 RemoteEndPointValid vst -> do

@@ -764,7 +764,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

764764 RemoteEndPointInvalid _ ->

765765 relyViolation (ourEndPoint, theirEndPoint)

766766 "handleIncomingMessages (invalid)"

767− RemoteEndPointInit _ _ ->

767+ RemoteEndPointInit _ _ _ ->

768768 relyViolation (ourEndPoint, theirEndPoint)

769769 "handleIncomingMessages (init)"

770770 RemoteEndPointValid ep ->

@@ -820,7 +820,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

820820 RemoteEndPointInvalid _ ->

821821 relyViolation (ourEndPoint, theirEndPoint)

822822 "handleIncomingMessages:createNewConnection (invalid)"

823− RemoteEndPointInit _ _ ->

823+ RemoteEndPointInit _ _ _ ->

824824 relyViolation (ourEndPoint, theirEndPoint)

825825 "handleIncomingMessages:createNewConnection (init)"

826826 RemoteEndPointValid vst ->

@@ -854,7 +854,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

854854 RemoteEndPointInvalid _ ->

855855 relyViolation (ourEndPoint, theirEndPoint)

856856 "readControlResponse (invalid)"

857− RemoteEndPointInit _ _ ->

857+ RemoteEndPointInit _ _ _ ->

858858 relyViolation (ourEndPoint, theirEndPoint)

859859 "readControlResponse (init)"

860860 RemoteEndPointValid vst ->

@@ -884,7 +884,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

884884 modifyMVar_ theirState $ \st -> case st of

885885 RemoteEndPointInvalid _ ->

886886 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (invalid)"

887− RemoteEndPointInit _ _ ->

887+ RemoteEndPointInit _ _ _ ->

888888 relyViolation (ourEndPoint, theirEndPoint) "closeConnection (init)"

889889 RemoteEndPointValid vst -> do

890890 unless (IntSet.member cid (vst ^. remoteIncoming)) $

@@ -913,7 +913,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

913913 RemoteEndPointInvalid _ ->

914914 relyViolation (ourEndPoint, theirEndPoint)

915915 "handleIncomingMessages:closeSocket (invalid)"

916− RemoteEndPointInit _ _ ->

916+ RemoteEndPointInit _ _ _ ->

917917 relyViolation (ourEndPoint, theirEndPoint)

918918 "handleIncomingMessages:closeSocket (init)"

919919 RemoteEndPointValid vst -> do

@@ -971,7 +971,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

971971 RemoteEndPointInvalid _ ->

972972 relyViolation (ourEndPoint, theirEndPoint)

973973 "handleIncomingMessages:prematureExit"

974− RemoteEndPointInit _ _ ->

974+ RemoteEndPointInit _ _ _ ->

975975 relyViolation (ourEndPoint, theirEndPoint)

976976 "handleIncomingMessages:prematureExit"

977977 RemoteEndPointValid vst -> do

@@ -1055,7 +1055,13 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do

10551055 tryCloseSocket sock

10561056 return False

10571057 Right (sock, ConnectionRequestCrossed) -> do

1058− resolveInit (ourEndPoint, theirEndPoint) RemoteEndPointClosed

1058+ withMVar (remoteState theirEndPoint) $ \st -> case st of

1059+ RemoteEndPointInit _ crossed _ ->

1060+ putMVar crossed ()

1061+ RemoteEndPointFailed ex ->

1062+ throwIO ex

1063+ _ ->

1064+ relyViolation (ourEndPoint, theirEndPoint) "setupRemoteEndPoint: Crossed"

10591065 tryCloseSocket sock

10601066 return False

10611067 Left err -> do

@@ -1088,7 +1094,7 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do

10881094 -- Error cases

10891095 RemoteEndPointInvalid err ->

10901096 throwIO err

1091− RemoteEndPointInit _ _ ->

1097+ RemoteEndPointInit _ _ _ ->

10921098 relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

10931099 RemoteEndPointClosing _ _ ->

10941100 relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

@@ -1193,7 +1199,7 @@ connectToSelf ourEndPoint = do

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

11941200resolveInit (ourEndPoint, theirEndPoint) newState =

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

1196− RemoteEndPointInit resolved _ -> do

1202+ RemoteEndPointInit resolved _ _ -> do

11971203 putMVar resolved ()

11981204 case newState of

11991205 RemoteEndPointClosed ->

@@ -1307,7 +1313,8 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

13071313 return (st, (theirEndPoint, False))

13081314 Nothing -> do

13091315 resolved <- newEmptyMVar

1310− theirState <- newMVar (RemoteEndPointInit resolved findOrigin)

1316+ crossed <- newEmptyMVar

1317+ theirState <- newMVar (RemoteEndPointInit resolved crossed findOrigin)

13111318 scheduled <- newChan

13121319 let theirEndPoint = RemoteEndPoint

13131320 { remoteAddress = theirAddress

@@ -1347,7 +1354,7 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

13471354 case snapshot of

13481355 RemoteEndPointInvalid err ->

13491356 throwIO err

1350− RemoteEndPointInit resolved initOrigin ->

1357+ RemoteEndPointInit resolved crossed initOrigin ->

13511358 case (findOrigin, initOrigin) of

13521359 (RequestedByUs, RequestedByUs) ->

13531360 readMVar resolved >> go

@@ -1355,9 +1362,10 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go

13551362 readMVar resolved >> go

13561363 (RequestedByThem, RequestedByUs) ->

13571364 if ourAddress > theirAddress

1358− then

1365+ then do

13591366 -- Wait for the Crossed message

1360− readMVar resolved >> go

1367+ readMVar crossed

1368+ return (theirEndPoint, True)

13611369 else

13621370 return (theirEndPoint, False)

13631371 (RequestedByThem, RequestedByThem) ->

@@ -1531,7 +1539,7 @@ internalSocketBetween transport ourAddress theirAddress = do

15311539 Nothing -> throwIO $ userError "Remote endpoint not found"

15321540 Just ep -> return ep

15331541 withMVar (remoteState theirEndPoint) $ \st -> case st of

1534− RemoteEndPointInit _ _ ->

1542+ RemoteEndPointInit _ _ _ ->

15351543 throwIO $ userError "Remote endpoint not yet initialized"

15361544 RemoteEndPointValid vst ->

15371545 return $ remoteSocket vst