git.iko.soy/distributed-process
on master help sign in
Race condition fixed.

We implicitly represented endpoint who were in "initialization state" as an empty mvar. This does not work, because handleConnectionRequest needs to know the state of the remote endpoint explicitly (this is the bug we fixed in the previous commit), but we can't test an mvar for being empty, because this might equally mean that another thread is in the process of modifying said MVar. Hence, we introduced an additional constructor RemoteEndPointInit for the RemoteState ADT, which carries a "resolved MVar (like RemoteEndPointClosing) which can be used by threads which want to block until the remote endpoint has been fully initialized.

commitb901caa83255bc75ba60ce5af2e7d866d1352da3 authorEdsko de Vries <edsko@well-typed.com> dateFri 25 May 2012, 20:12 (14 years ago) parentb0ba19e
3 files changed+106−83
compare with master
unifiedsplit
network-transport/src/Network/Transport/Internal.hs−16 network-transport/src/Network/Transport/TCP.hs+105−66 network-transport/tests/TestTransport.hs+1−1
network-transport/src/Network/Transport/Internal.hs−16
@@ -9,7 +9,6 @@ module Network.Transport.Internal ( -- * Encoders/decoders
99 , mapIOException
1010 , tryIO
1111 , tryToEnum
12− , tryModifyMVar
1312 -- * Replicated functionality from "base"
1413 , void
1514 , forkIOWithUnmask
@@ -32,11 +31,8 @@ import Control.Exception ( IOException
3231 , catch
3332 , try
3433 , throwIO
35− , mask
36− , onException
3734 )
3835import Control.Concurrent (ThreadId, forkIO)
39−import Control.Concurrent (MVar, tryTakeMVar, putMVar)
4036import GHC.IO (unsafeUnmask)
4137--import Control.Concurrent (myThreadId)
4238
@@ -110,15 +106,3 @@ tryToEnum = go minBound maxBound
110106 where
111107 go :: Enum b => b -> b -> Int -> Maybe b
112108 go lo hi n = if fromEnum lo <= n && n <= fromEnum hi then Just (toEnum n) else Nothing
113−
114−tryModifyMVar :: MVar a -> (a -> IO (a, b)) -> IO (Maybe b)
115−tryModifyMVar m io =
116− mask $ \restore -> do
117− ma <- tryTakeMVar m
118− case ma of
119− Nothing -> return Nothing
120− Just a -> do
121− (a', b) <- restore (io a) `onException` putMVar m a
122− putMVar m a'
123− return (Just b)
124−
network-transport/src/Network/Transport/TCP.hs+105−66
@@ -39,7 +39,6 @@ import Network.Transport.Internal ( encodeInt32
3939 , mapIOException
4040 , tryIO
4141 , tryToEnum
42− , tryModifyMVar
4342 , void
4443 )
4544import qualified Network.Socket as N ( HostName
@@ -108,6 +107,7 @@ import qualified Data.Map as Map (empty)
108107import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
109108import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
110109import Data.Foldable (forM_, mapM_)
110+import Data.Maybe (isNothing)
111111
112112-- $design
113113--
@@ -292,6 +292,7 @@ data RemoteEndPoint = RemoteEndPoint
292292
293293data RemoteState =
294294 RemoteEndPointInvalid (TransportError ConnectErrorCode)
295+ | RemoteEndPointInit (MVar ())
295296 | RemoteEndPointValid ValidRemoteEndPointState
296297 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState
297298 | RemoteEndPointClosed
@@ -487,7 +488,9 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
487488 liftM join . try . mapIOException sendFailed $
488489 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
489490 { caseInvalid = \_ ->
490− relyViolation (ourEndPoint, theirEndPoint) "apiSend"
491+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"
492+ , caseInit = \_ ->
493+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"
491494 , caseValid = \vst -> do
492495 alive <- readIORef connAlive
493496 if alive
@@ -540,6 +543,8 @@ apiCloseEndPoint transport evs ourEndPoint = do
540543 case st of
541544 RemoteEndPointInvalid _ ->
542545 return st
546+ RemoteEndPointInit _ ->
547+ return st
543548 RemoteEndPointValid conn -> do
544549 -- Try to send a CloseSocket request
545550 tryIO $ sendOn conn [encodeInt32 CloseSocket]
@@ -621,6 +626,8 @@ internalSocketBetween transport ourAddress theirAddress = runErrorT $ do
621626 Just ep -> return . Right $ ep
622627 ErrorT $
623628 withMVar (remoteState theirEndPoint) $ \st -> case st of
629+ RemoteEndPointInit _ ->
630+ return . Left $ "Remote endpoint not yet initialized"
624631 RemoteEndPointValid vst ->
625632 return . Right $ remoteSocket vst
626633 RemoteEndPointClosing _ vst ->
@@ -682,12 +689,15 @@ requestConnectionTo ourEndPoint theirAddress = go
682689 where
683690 go = do
684691 (theirEndPoint, isNew) <- mapIOException connectFailed $
685− findRemoteEndPoint ourEndPoint theirAddress
686− (Just $ RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))
692+ findRemoteEndPoint ourEndPoint theirAddress True
693+ (RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))
687694
688− when isNew . void . forkEndPointThread ourEndPoint $ do
689− let outgoing = 1
690− setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing
695+ -- If this is a new endpoint, start the initialization process and wait
696+ forM_ isNew $ \resolved -> do
697+ void . forkEndPointThread ourEndPoint $ do
698+ let outgoing = 1
699+ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing
700+ readMVar resolved
691701
692702 reply <- mapIOException connectFailed $
693703 doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
@@ -718,7 +728,9 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
718728 , _remoteIncoming = IntSet.empty
719729 , sendOn = sendMany sock
720730 }
721− putMVar theirState (RemoteEndPointValid vst)
731+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
732+ putMVar resolved ()
733+ return (RemoteEndPointValid vst)
722734 return True
723735 Right (sock, ConnectionRequestInvalid) -> do
724736 -- We remove the endpoint from our local state again because the next
@@ -728,15 +740,21 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do
728740 -- connection requests.
729741 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
730742 let err = invalidAddress "Invalid endpoint"
731− putMVar theirState (RemoteEndPointInvalid err)
743+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
744+ putMVar resolved ()
745+ return (RemoteEndPointInvalid err)
732746 tryCloseSocket sock
733747 return False
734748 Right (sock, ConnectionRequestCrossed) -> do
749+ -- We leave the endpoint in Init state, handleConnectionRequest will
750+ -- take care of it
735751 tryCloseSocket sock
736752 return False
737753 Left err -> do
738754 removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above
739− putMVar theirState (RemoteEndPointInvalid err)
755+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
756+ putMVar resolved ()
757+ return (RemoteEndPointInvalid err)
740758 return False
741759
742760 -- If we get to this point without an exception, then
@@ -884,6 +902,7 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do
884902 throwIO (userError "Local endpoint closed")
885903 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
886904 { caseInvalid = throwIO
905+ , caseInit = \_ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
887906 , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]
888907 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
889908 , caseClosed = throwIO $ userError "Remote endpoint closed"
@@ -954,6 +973,7 @@ forkEndPointThread ourEndPoint p =
954973
955974data RemoteStatePatternMatch a = RemoteStatePatternMatch
956975 { caseInvalid :: TransportError ConnectErrorCode -> IO a
976+ , caseInit :: MVar () -> IO a
957977 , caseValid :: ValidRemoteEndPointState -> IO a
958978 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
959979 , caseClosed :: IO a
@@ -963,6 +983,7 @@ remoteStateIdentity :: RemoteStatePatternMatch RemoteState
963983remoteStateIdentity =
964984 RemoteStatePatternMatch
965985 { caseInvalid = return . RemoteEndPointInvalid
986+ , caseInit = return . RemoteEndPointInit
966987 , caseValid = return . RemoteEndPointValid
967988 , caseClosing = (return .) . RemoteEndPointClosing
968989 , caseClosed = return RemoteEndPointClosed
@@ -990,6 +1011,11 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =
9901011 throwIO ex
9911012 -- The other cases are less interesting, because unless the endpoint is
9921013 -- in Valid state we're not supposed to do any IO on it
1014+ RemoteEndPointInit resolved -> do
1015+ (st', a) <- onException (restore $ caseInit match resolved)
1016+ (putMVar theirState st)
1017+ putMVar theirState st'
1018+ return a
9931019 RemoteEndPointClosing resolved vst -> do
9941020 (st', a) <- onException (restore $ caseClosing match resolved vst)
9951021 (putMVar theirState st)
@@ -1027,6 +1053,7 @@ modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
10271053 modifyRemoteState (ourEndPoint, theirEndPoint)
10281054 RemoteStatePatternMatch
10291055 { caseInvalid = u . caseInvalid match
1056+ , caseInit = u . caseInit match
10301057 , caseValid = u . caseValid match
10311058 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
10321059 , caseClosed = u $ caseClosed match
@@ -1045,6 +1072,9 @@ withRemoteState (ourEndPoint, theirEndPoint) match =
10451072 { caseInvalid = \err -> do
10461073 a <- caseInvalid match err
10471074 return (RemoteEndPointInvalid err, a)
1075+ , caseInit = \resolved -> do
1076+ a <- caseInit match resolved
1077+ return (RemoteEndPointInit resolved, a)
10481078 , caseValid = \vst -> do
10491079 a <- caseValid match vst
10501080 return (RemoteEndPointValid vst, a)
@@ -1094,9 +1124,13 @@ handleConnectionRequest transport sock = handle handleException $ do
10941124 go ourEndPoint theirAddress = do
10951125 mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do
10961126 (theirEndPoint, isNew) <-
1097− findRemoteEndPoint ourEndPoint theirAddress Nothing
1127+ findRemoteEndPoint ourEndPoint theirAddress False
1128+ (throw $ userError "Already connected")
1129+
1130+ let ourAddress = localAddress ourEndPoint
1131+ theirState = remoteState theirEndPoint
1132+ crossed = isNothing isNew && ourAddress < theirAddress
10981133
1099− let crossed = not isNew && localAddress ourEndPoint < theirAddress
11001134 if crossed
11011135 then do
11021136 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
@@ -1109,7 +1143,9 @@ handleConnectionRequest transport sock = handle handleException $ do
11091143 , _remoteIncoming = IntSet.empty
11101144 , sendOn = sendMany sock
11111145 }
1112− putMVar (remoteState theirEndPoint) (RemoteEndPointValid vst)
1146+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do
1147+ putMVar resolved ()
1148+ return (RemoteEndPointValid vst)
11131149 sendMany sock [encodeInt32 ConnectionRequestAccepted]
11141150 return (Just theirEndPoint)
11151151 -- If we left the scope of the exception handler with a return value of
@@ -1127,71 +1163,64 @@ handleConnectionRequest transport sock = handle handleException $ do
11271163 tryIO $ sendMany sock [encodeInt32 ConnectionRequestInvalid]
11281164 tryCloseSocket sock
11291165
1130−-- | Find a remote endpoint. Pass 'Nothing' for the third argument to find an
1131−-- uninitialized remote endpoint; pass 'Just' a function to find a Valid remote
1132−-- endpoint (and modify its state).
1166+-- | Find a remote endpoint. If the remote endpoint does not yet exist we
1167+-- create it in Init state and return the corresponding 'resolved' MVar.
11331168findRemoteEndPoint
1134− :: LocalEndPoint
1135− -> EndPointAddress
1136− -> Maybe (ValidRemoteEndPointState -> RemoteState)
1137− -> IO (RemoteEndPoint, Bool)
1138−findRemoteEndPoint ourEndPoint theirAddress mCaseValid = go
1169+ :: LocalEndPoint -- ^ Local endpoint
1170+ -> EndPointAddress -- ^ Remote address
1171+ -> Bool -- ^ Wait if remote endpoint is in Init state
1172+ -> (ValidRemoteEndPointState -> RemoteState) -- ^ Apply if in Valid state
1173+ -> IO (RemoteEndPoint, Maybe (MVar ()))
1174+findRemoteEndPoint ourEndPoint theirAddress waitOnInit whenValid = go
11391175 where
11401176 go = do
11411177 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
1142− LocalEndPointValid vst ->
1143− case vst ^. localConnectionTo theirAddress of
1144− Just theirEndPoint ->
1145− return (st, (theirEndPoint, False))
1146− Nothing -> do
1147− theirState <- newEmptyMVar
1148− let theirEndPoint = RemoteEndPoint
1149− { remoteAddress = theirAddress
1150− , remoteState = theirState
1151− , remoteId = vst ^. nextRemoteId
1152− }
1153− return ( LocalEndPointValid
1154− . (localConnectionTo theirAddress ^= Just theirEndPoint)
1155− . (nextRemoteId ^: (+ 1))
1156− $ vst
1157− , (theirEndPoint, True)
1158− )
1178+ LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of
1179+ Just theirEndPoint ->
1180+ return (st, (theirEndPoint, Nothing))
1181+ Nothing -> do
1182+ resolved <- newEmptyMVar
1183+ theirState <- newMVar (RemoteEndPointInit resolved)
1184+ let theirEndPoint = RemoteEndPoint
1185+ { remoteAddress = theirAddress
1186+ , remoteState = theirState
1187+ , remoteId = vst ^. nextRemoteId
1188+ }
1189+ return ( LocalEndPointValid
1190+ . (localConnectionTo theirAddress ^= Just theirEndPoint)
1191+ . (nextRemoteId ^: (+ 1))
1192+ $ vst
1193+ , (theirEndPoint, Just resolved)
1194+ )
11591195 LocalEndPointClosed ->
11601196 throwIO $ userError "Local endpoint closed"
1161−
1162− if isNew
1163− then
1164− return (theirEndPoint, True)
1165− else do
1197+
1198+ case isNew of
1199+ Just resolved ->
1200+ return (theirEndPoint, Just resolved)
1201+ Nothing -> do
11661202 let theirState = remoteState theirEndPoint
1167− snapshot <- modify theirState $ \whenValid st -> case st of
1168− RemoteEndPointInvalid _ ->
1169− return (st, st)
1203+ snapshot <- modifyMVar theirState $ \st -> case st of
11701204 RemoteEndPointValid vst -> do
1171− let st' = whenValid vst
1205+ let st' = whenValid vst
11721206 return (st', st')
1173− RemoteEndPointClosing _ _ ->
1174− return (st, st)
1175− RemoteEndPointClosed ->
1207+ _ ->
11761208 return (st, st)
11771209 case snapshot of
1178− Nothing ->
1179− return (theirEndPoint, False)
1180− Just (RemoteEndPointInvalid _) ->
1181− throwIO $ userError "Invalid endpoint"
1182− Just (RemoteEndPointClosing resolved _) ->
1210+ RemoteEndPointInvalid err ->
1211+ throwIO err
1212+ RemoteEndPointInit resolved ->
1213+ if waitOnInit
1214+ then readMVar resolved >> go
1215+ else return (theirEndPoint, Nothing)
1216+ RemoteEndPointValid _ ->
1217+ return (theirEndPoint, Nothing)
1218+ RemoteEndPointClosing resolved _ ->
11831219 readMVar resolved >> go
1184− Just RemoteEndPointClosed ->
1220+ RemoteEndPointClosed ->
11851221 go
1186− Just (RemoteEndPointValid _) ->
1187− return (theirEndPoint, False)
1188−
1189− ourState = localState ourEndPoint
1190− modify m io = case mCaseValid of
1191− Nothing ->
1192− tryModifyMVar m (io (throw $ userError "Unexpected valid state"))
1193− Just f ->
1194− Just <$> modifyMVar m (io f)
1222+
1223+ ourState = localState ourEndPoint
11951224
11961225-- | Handle requests from a remote endpoint.
11971226--
@@ -1204,7 +1233,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12041233 mSock <- withMVar theirState $ \st ->
12051234 case st of
12061235 RemoteEndPointInvalid _ ->
1207− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:initialization"
1236+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"
1237+ RemoteEndPointInit _ ->
1238+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"
12081239 RemoteEndPointValid ep ->
12091240 return . Just $ remoteSocket ep
12101241 RemoteEndPointClosing _ ep ->
@@ -1264,6 +1295,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12641295 vst <- case st of
12651296 RemoteEndPointInvalid _ ->
12661297 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"
1298+ RemoteEndPointInit _ ->
1299+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"
12671300 RemoteEndPointValid vst ->
12681301 return (remoteIncoming ^: IntSet.insert newId $ vst)
12691302 RemoteEndPointClosing resolved vst -> do
@@ -1314,6 +1347,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13141347 modifyMVar_ theirState $ \st -> case st of
13151348 RemoteEndPointInvalid _ ->
13161349 relyViolation (ourEndPoint, theirEndPoint) "closeConnection"
1350+ RemoteEndPointInit _ ->
1351+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection"
13171352 RemoteEndPointValid vst -> do
13181353 unless (IntSet.member cid (vst ^. remoteIncoming)) $
13191354 throwIO $ userError "Invalid CloseConnection"
@@ -1338,6 +1373,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13381373 case st of
13391374 RemoteEndPointInvalid _ ->
13401375 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"
1376+ RemoteEndPointInit _ ->
1377+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"
13411378 RemoteEndPointValid vst -> do
13421379 -- We regard a CloseSocket message as an (optimized) way for the
13431380 -- remote endpoint to indicate that all its connections to us are
@@ -1385,6 +1422,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
13851422 case st of
13861423 RemoteEndPointInvalid _ ->
13871424 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"
1425+ RemoteEndPointInit _ ->
1426+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"
13881427 RemoteEndPointValid vst -> do
13891428 let code = EventConnectionLost
13901429 (remoteAddress theirEndPoint)
network-transport/tests/TestTransport.hs+1−1
@@ -35,7 +35,7 @@ echoServer endpoint = do
3535 event <- receive endpoint
3636 case event of
3737 ConnectionOpened cid rel addr -> do
38− tlog $ "Opened new conncetion " ++ show cid
38+ tlog $ "Opened new connection " ++ show cid
3939 Right conn <- connect endpoint addr rel
4040 go (Map.insert cid conn cs)
4141 Received cid payload -> do