b901caa · distributed-process

git.iko.soy/distributed-process

on master help sign in

Race condition fixed.

b901caaEdsko de Vries14 years ago

network-transport/src/Network/Transport/Internal.hs−16

compare with master

unifiedsplit

@@ -9,7 +9,6 @@ module Network.Transport.Internal ( -- * Encoders/decoders

9 , mapIOException

10 , tryIO

11 , tryToEnum

12− , tryModifyMVar

13 -- * Replicated functionality from "base"

12 -- * Replicated functionality from "base"

14 , void

13 , void

15 , forkIOWithUnmask

14 , forkIOWithUnmask

@@ -32,11 +31,8 @@ import Control.Exception ( IOException

32 , catch

31 , catch

33 , try

32 , try

34 , throwIO

33 , throwIO

35− , mask

36− , onException

37 )

34 )

38import Control.Concurrent (ThreadId, forkIO)

35import Control.Concurrent (ThreadId, forkIO)

39−import Control.Concurrent (MVar, tryTakeMVar, putMVar)

40import GHC.IO (unsafeUnmask)

36import GHC.IO (unsafeUnmask)

41--import Control.Concurrent (myThreadId)

37--import Control.Concurrent (myThreadId)

@@ -110,15 +106,3 @@ tryToEnum = go minBound maxBound

110 where

106 where

111 go :: Enum b => b -> b -> Int -> Maybe b

107 go :: Enum b => b -> b -> Int -> Maybe b

112 go lo hi n = if fromEnum lo <= n && n <= fromEnum hi then Just (toEnum n) else Nothing

108 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

39 , mapIOException

40 , tryIO

41 , tryToEnum

42− , tryModifyMVar

43 , void

42 , void

44 )

43 )

45import qualified Network.Socket as N ( HostName

44import qualified Network.Socket as N ( HostName

@@ -108,6 +107,7 @@ import qualified Data.Map as Map (empty)

108import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

107import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))

109import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)

108import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)

110import Data.Foldable (forM_, mapM_)

109import Data.Foldable (forM_, mapM_)

110+import Data.Maybe (isNothing)

111

112-- $design

113--

@@ -292,6 +292,7 @@ data RemoteEndPoint = RemoteEndPoint

292

293data RemoteState =

294 RemoteEndPointInvalid (TransportError ConnectErrorCode)

295+ | RemoteEndPointInit (MVar ())

295 | RemoteEndPointValid ValidRemoteEndPointState

296 | RemoteEndPointValid ValidRemoteEndPointState

296 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState

297 | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState

297 | RemoteEndPointClosed

298 | RemoteEndPointClosed

@@ -487,7 +488,9 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =

487 liftM join . try . mapIOException sendFailed $

488 liftM join . try . mapIOException sendFailed $

488 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

489 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

489 { caseInvalid = \_ ->

490 { caseInvalid = \_ ->

491+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"

492+ , caseInit = \_ ->

490− relyViolation (ourEndPoint, theirEndPoint) "apiSend"

493+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"

491 , caseValid = \vst -> do

494 , caseValid = \vst -> do

492 alive <- readIORef connAlive

495 alive <- readIORef connAlive

493 if alive

496 if alive

@@ -540,6 +543,8 @@ apiCloseEndPoint transport evs ourEndPoint = do

540 case st of

543 case st of

541 RemoteEndPointInvalid _ ->

544 RemoteEndPointInvalid _ ->

542 return st

545 return st

546+ RemoteEndPointInit _ ->

547+ return st

543 RemoteEndPointValid conn -> do

548 RemoteEndPointValid conn -> do

544 -- Try to send a CloseSocket request

549 -- Try to send a CloseSocket request

545 tryIO $ sendOn conn [encodeInt32 CloseSocket]

550 tryIO $ sendOn conn [encodeInt32 CloseSocket]

@@ -621,6 +626,8 @@ internalSocketBetween transport ourAddress theirAddress = runErrorT $ do

621 Just ep -> return . Right $ ep

626 Just ep -> return . Right $ ep

622 ErrorT $

627 ErrorT $

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

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

629+ RemoteEndPointInit _ ->

630+ return . Left $ "Remote endpoint not yet initialized"

624 RemoteEndPointValid vst ->

631 RemoteEndPointValid vst ->

625 return . Right $ remoteSocket vst

632 return . Right $ remoteSocket vst

626 RemoteEndPointClosing _ vst ->

633 RemoteEndPointClosing _ vst ->

@@ -682,12 +689,15 @@ requestConnectionTo ourEndPoint theirAddress = go

682 where

689 where

683 go = do

690 go = do

684 (theirEndPoint, isNew) <- mapIOException connectFailed $

691 (theirEndPoint, isNew) <- mapIOException connectFailed $

685− findRemoteEndPoint ourEndPoint theirAddress

692+ findRemoteEndPoint ourEndPoint theirAddress True

686− (Just $ RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))

693+ (RemoteEndPointValid . (remoteOutgoing ^: (+ 1)))

687

694

695+ -- If this is a new endpoint, start the initialization process and wait

696+ forM_ isNew $ \resolved -> do

688− when isNew . void . forkEndPointThread ourEndPoint $ do

697+ void . forkEndPointThread ourEndPoint $ do

689− let outgoing = 1

698+ let outgoing = 1

690− setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing

699+ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing

700+ readMVar resolved

691

701

692 reply <- mapIOException connectFailed $

702 reply <- mapIOException connectFailed $

693 doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

703 doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId

@@ -718,7 +728,9 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do

718 , _remoteIncoming = IntSet.empty

728 , _remoteIncoming = IntSet.empty

719 , sendOn = sendMany sock

729 , sendOn = sendMany sock

720 }

730 }

731+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do

732+ putMVar resolved ()

721− putMVar theirState (RemoteEndPointValid vst)

733+ return (RemoteEndPointValid vst)

722 return True

734 return True

723 Right (sock, ConnectionRequestInvalid) -> do

735 Right (sock, ConnectionRequestInvalid) -> do

724 -- We remove the endpoint from our local state again because the next

736 -- We remove the endpoint from our local state again because the next

@@ -728,15 +740,21 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) outgoing = do

728 -- connection requests.

740 -- connection requests.

729 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

741 removeRemoteEndPoint (ourEndPoint, theirEndPoint)

730 let err = invalidAddress "Invalid endpoint"

742 let err = invalidAddress "Invalid endpoint"

743+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do

744+ putMVar resolved ()

731− putMVar theirState (RemoteEndPointInvalid err)

745+ return (RemoteEndPointInvalid err)

732 tryCloseSocket sock

746 tryCloseSocket sock

733 return False

747 return False

734 Right (sock, ConnectionRequestCrossed) -> do

748 Right (sock, ConnectionRequestCrossed) -> do

749+ -- We leave the endpoint in Init state, handleConnectionRequest will

750+ -- take care of it

735 tryCloseSocket sock

751 tryCloseSocket sock

736 return False

752 return False

737 Left err -> do

753 Left err -> do

738 removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above

754 removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above

755+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do

756+ putMVar resolved ()

739− putMVar theirState (RemoteEndPointInvalid err)

757+ return (RemoteEndPointInvalid err)

740 return False

758 return False

741

759

742 -- If we get to this point without an exception, then

760 -- If we get to this point without an exception, then

@@ -884,6 +902,7 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do

884 throwIO (userError "Local endpoint closed")

902 throwIO (userError "Local endpoint closed")

885 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

903 withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch

886 { caseInvalid = throwIO

904 { caseInvalid = throwIO

905+ , caseInit = \_ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"

887 , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]

906 , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]

888 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

907 , caseClosing = \_ _ -> relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"

889 , caseClosed = throwIO $ userError "Remote endpoint closed"

908 , caseClosed = throwIO $ userError "Remote endpoint closed"

@@ -954,6 +973,7 @@ forkEndPointThread ourEndPoint p =

954

973

955data RemoteStatePatternMatch a = RemoteStatePatternMatch

974data RemoteStatePatternMatch a = RemoteStatePatternMatch

956 { caseInvalid :: TransportError ConnectErrorCode -> IO a

975 { caseInvalid :: TransportError ConnectErrorCode -> IO a

976+ , caseInit :: MVar () -> IO a

957 , caseValid :: ValidRemoteEndPointState -> IO a

977 , caseValid :: ValidRemoteEndPointState -> IO a

958 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a

978 , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a

959 , caseClosed :: IO a

979 , caseClosed :: IO a

@@ -963,6 +983,7 @@ remoteStateIdentity :: RemoteStatePatternMatch RemoteState

963remoteStateIdentity =

983remoteStateIdentity =

964 RemoteStatePatternMatch

984 RemoteStatePatternMatch

965 { caseInvalid = return . RemoteEndPointInvalid

985 { caseInvalid = return . RemoteEndPointInvalid

986+ , caseInit = return . RemoteEndPointInit

966 , caseValid = return . RemoteEndPointValid

987 , caseValid = return . RemoteEndPointValid

967 , caseClosing = (return .) . RemoteEndPointClosing

988 , caseClosing = (return .) . RemoteEndPointClosing

968 , caseClosed = return RemoteEndPointClosed

989 , caseClosed = return RemoteEndPointClosed

@@ -990,6 +1011,11 @@ modifyRemoteState (ourEndPoint, theirEndPoint) match =

990 throwIO ex

1011 throwIO ex

991 -- The other cases are less interesting, because unless the endpoint is

1012 -- The other cases are less interesting, because unless the endpoint is

992 -- in Valid state we're not supposed to do any IO on it

1013 -- 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

993 RemoteEndPointClosing resolved vst -> do

1019 RemoteEndPointClosing resolved vst -> do

994 (st', a) <- onException (restore $ caseClosing match resolved vst)

1020 (st', a) <- onException (restore $ caseClosing match resolved vst)

995 (putMVar theirState st)

1021 (putMVar theirState st)

@@ -1027,6 +1053,7 @@ modifyRemoteState_ (ourEndPoint, theirEndPoint) match =

1027 modifyRemoteState (ourEndPoint, theirEndPoint)

1053 modifyRemoteState (ourEndPoint, theirEndPoint)

1028 RemoteStatePatternMatch

1054 RemoteStatePatternMatch

1029 { caseInvalid = u . caseInvalid match

1055 { caseInvalid = u . caseInvalid match

1056+ , caseInit = u . caseInit match

1030 , caseValid = u . caseValid match

1057 , caseValid = u . caseValid match

1031 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst

1058 , caseClosing = \resolved vst -> u $ caseClosing match resolved vst

1032 , caseClosed = u $ caseClosed match

1059 , caseClosed = u $ caseClosed match

@@ -1045,6 +1072,9 @@ withRemoteState (ourEndPoint, theirEndPoint) match =

1045 { caseInvalid = \err -> do

1072 { caseInvalid = \err -> do

1046 a <- caseInvalid match err

1073 a <- caseInvalid match err

1047 return (RemoteEndPointInvalid err, a)

1074 return (RemoteEndPointInvalid err, a)

1075+ , caseInit = \resolved -> do

1076+ a <- caseInit match resolved

1077+ return (RemoteEndPointInit resolved, a)

1048 , caseValid = \vst -> do

1078 , caseValid = \vst -> do

1049 a <- caseValid match vst

1079 a <- caseValid match vst

1050 return (RemoteEndPointValid vst, a)

1080 return (RemoteEndPointValid vst, a)

@@ -1094,9 +1124,13 @@ handleConnectionRequest transport sock = handle handleException $ do

1094 go ourEndPoint theirAddress = do

1124 go ourEndPoint theirAddress = do

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

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

1096 (theirEndPoint, isNew) <-

1126 (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

1098

1133

1099− let crossed = not isNew && localAddress ourEndPoint < theirAddress

1100 if crossed

1134 if crossed

1101 then do

1135 then do

1102 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]

1136 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]

@@ -1109,7 +1143,9 @@ handleConnectionRequest transport sock = handle handleException $ do

1109 , _remoteIncoming = IntSet.empty

1143 , _remoteIncoming = IntSet.empty

1110 , sendOn = sendMany sock

1144 , sendOn = sendMany sock

1111 }

1145 }

1146+ modifyMVar_ theirState $ \(RemoteEndPointInit resolved) -> do

1147+ putMVar resolved ()

1112− putMVar (remoteState theirEndPoint) (RemoteEndPointValid vst)

1148+ return (RemoteEndPointValid vst)

1113 sendMany sock [encodeInt32 ConnectionRequestAccepted]

1149 sendMany sock [encodeInt32 ConnectionRequestAccepted]

1114 return (Just theirEndPoint)

1150 return (Just theirEndPoint)

1115 -- If we left the scope of the exception handler with a return value of

1151 -- If we left the scope of the exception handler with a return value of

@@ -1127,71 +1163,64 @@ handleConnectionRequest transport sock = handle handleException $ do

1127 tryIO $ sendMany sock [encodeInt32 ConnectionRequestInvalid]

1163 tryIO $ sendMany sock [encodeInt32 ConnectionRequestInvalid]

1128 tryCloseSocket sock

1164 tryCloseSocket sock

1129

1165

1130−-- | Find a remote endpoint. Pass 'Nothing' for the third argument to find an

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

1131−-- uninitialized remote endpoint; pass 'Just' a function to find a Valid remote

1167+-- create it in Init state and return the corresponding 'resolved' MVar.

1132−-- endpoint (and modify its state).

1133findRemoteEndPoint

1168findRemoteEndPoint

1134− :: LocalEndPoint

1169+ :: LocalEndPoint -- ^ Local endpoint

1135− -> EndPointAddress

1170+ -> EndPointAddress -- ^ Remote address

1171+ -> Bool -- ^ Wait if remote endpoint is in Init state

1136− -> Maybe (ValidRemoteEndPointState -> RemoteState)

1172+ -> (ValidRemoteEndPointState -> RemoteState) -- ^ Apply if in Valid state

1137− -> IO (RemoteEndPoint, Bool)

1173+ -> IO (RemoteEndPoint, Maybe (MVar ()))

1138−findRemoteEndPoint ourEndPoint theirAddress mCaseValid = go

1174+findRemoteEndPoint ourEndPoint theirAddress waitOnInit whenValid = go

1139 where

1175 where

1140 go = do

1176 go = do

1141 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of

1177 (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of

1142− LocalEndPointValid vst ->

1143− case vst ^. localConnectionTo theirAddress of

1178+ LocalEndPointValid vst -> case vst ^. localConnectionTo theirAddress of

1144− Just theirEndPoint ->

1179+ Just theirEndPoint ->

1145− return (st, (theirEndPoint, False))

1180+ return (st, (theirEndPoint, Nothing))

1146− Nothing -> do

1181+ Nothing -> do

1147− theirState <- newEmptyMVar

1182+ resolved <- newEmptyMVar

1183+ theirState <- newMVar (RemoteEndPointInit resolved)

1148− let theirEndPoint = RemoteEndPoint

1184+ let theirEndPoint = RemoteEndPoint

1149− { remoteAddress = theirAddress

1185+ { remoteAddress = theirAddress

1150− , remoteState = theirState

1186+ , remoteState = theirState

1151− , remoteId = vst ^. nextRemoteId

1187+ , remoteId = vst ^. nextRemoteId

1152− }

1188+ }

1153− return ( LocalEndPointValid

1189+ return ( LocalEndPointValid

1154− . (localConnectionTo theirAddress ^= Just theirEndPoint)

1190+ . (localConnectionTo theirAddress ^= Just theirEndPoint)

1155− . (nextRemoteId ^: (+ 1))

1191+ . (nextRemoteId ^: (+ 1))

1156− $ vst

1192+ $ vst

1157− , (theirEndPoint, True)

1193+ , (theirEndPoint, Just resolved)

1158− )

1194+ )

1159 LocalEndPointClosed ->

1195 LocalEndPointClosed ->

1160 throwIO $ userError "Local endpoint closed"

1196 throwIO $ userError "Local endpoint closed"

1161−

1197+

1162− if isNew

1198+ case isNew of

1163− then

1199+ Just resolved ->

1164− return (theirEndPoint, True)

1200+ return (theirEndPoint, Just resolved)

1165− else do

1201+ Nothing -> do

1166 let theirState = remoteState theirEndPoint

1202 let theirState = remoteState theirEndPoint

1167− snapshot <- modify theirState $ \whenValid st -> case st of

1203+ snapshot <- modifyMVar theirState $ \st -> case st of

1168− RemoteEndPointInvalid _ ->

1169− return (st, st)

1170 RemoteEndPointValid vst -> do

1204 RemoteEndPointValid vst -> do

1171− let st' = whenValid vst

1205+ let st' = whenValid vst

1172 return (st', st')

1206 return (st', st')

1173− RemoteEndPointClosing _ _ ->

1207+ _ ->

1174− return (st, st)

1175− RemoteEndPointClosed ->

1176 return (st, st)

1208 return (st, st)

1177 case snapshot of

1209 case snapshot of

1178− Nothing ->

1179− return (theirEndPoint, False)

1180− Just (RemoteEndPointInvalid _) ->

1210+ RemoteEndPointInvalid err ->

1181− throwIO $ userError "Invalid endpoint"

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)

1182− Just (RemoteEndPointClosing resolved _) ->

1218+ RemoteEndPointClosing resolved _ ->

1183 readMVar resolved >> go

1219 readMVar resolved >> go

1184− Just RemoteEndPointClosed ->

1220+ RemoteEndPointClosed ->

1185 go

1221 go

1186− Just (RemoteEndPointValid _) ->

1222+

1187− return (theirEndPoint, False)

1188−

1189− ourState = localState ourEndPoint

1223+ 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)

1195

1224

1196-- | Handle requests from a remote endpoint.

1225-- | Handle requests from a remote endpoint.

1197--

1226--

@@ -1204,7 +1233,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1204 mSock <- withMVar theirState $ \st ->

1233 mSock <- withMVar theirState $ \st ->

1205 case st of

1234 case st of

1206 RemoteEndPointInvalid _ ->

1235 RemoteEndPointInvalid _ ->

1207− relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:initialization"

1236+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"

1237+ RemoteEndPointInit _ ->

1238+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages"

1208 RemoteEndPointValid ep ->

1239 RemoteEndPointValid ep ->

1209 return . Just $ remoteSocket ep

1240 return . Just $ remoteSocket ep

1210 RemoteEndPointClosing _ ep ->

1241 RemoteEndPointClosing _ ep ->

@@ -1264,6 +1295,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1264 vst <- case st of

1295 vst <- case st of

1265 RemoteEndPointInvalid _ ->

1296 RemoteEndPointInvalid _ ->

1266 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"

1297 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"

1298+ RemoteEndPointInit _ ->

1299+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:createNewConnection"

1267 RemoteEndPointValid vst ->

1300 RemoteEndPointValid vst ->

1268 return (remoteIncoming ^: IntSet.insert newId $ vst)

1301 return (remoteIncoming ^: IntSet.insert newId $ vst)

1269 RemoteEndPointClosing resolved vst -> do

1302 RemoteEndPointClosing resolved vst -> do

@@ -1314,6 +1347,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1314 modifyMVar_ theirState $ \st -> case st of

1347 modifyMVar_ theirState $ \st -> case st of

1315 RemoteEndPointInvalid _ ->

1348 RemoteEndPointInvalid _ ->

1316 relyViolation (ourEndPoint, theirEndPoint) "closeConnection"

1349 relyViolation (ourEndPoint, theirEndPoint) "closeConnection"

1350+ RemoteEndPointInit _ ->

1351+ relyViolation (ourEndPoint, theirEndPoint) "closeConnection"

1317 RemoteEndPointValid vst -> do

1352 RemoteEndPointValid vst -> do

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

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

1319 throwIO $ userError "Invalid CloseConnection"

1354 throwIO $ userError "Invalid CloseConnection"

@@ -1338,6 +1373,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1338 case st of

1373 case st of

1339 RemoteEndPointInvalid _ ->

1374 RemoteEndPointInvalid _ ->

1340 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"

1375 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"

1376+ RemoteEndPointInit _ ->

1377+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:closeSocket"

1341 RemoteEndPointValid vst -> do

1378 RemoteEndPointValid vst -> do

1342 -- We regard a CloseSocket message as an (optimized) way for the

1379 -- We regard a CloseSocket message as an (optimized) way for the

1343 -- remote endpoint to indicate that all its connections to us are

1380 -- remote endpoint to indicate that all its connections to us are

@@ -1385,6 +1422,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do

1385 case st of

1422 case st of

1386 RemoteEndPointInvalid _ ->

1423 RemoteEndPointInvalid _ ->

1387 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"

1424 relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"

1425+ RemoteEndPointInit _ ->

1426+ relyViolation (ourEndPoint, theirEndPoint) "handleIncomingMessages:prematureExit"

1388 RemoteEndPointValid vst -> do

1427 RemoteEndPointValid vst -> do

1389 let code = EventConnectionLost

1428 let code = EventConnectionLost

1390 (remoteAddress theirEndPoint)

1429 (remoteAddress theirEndPoint)

network-transport/tests/TestTransport.hs+1−1

@@ -35,7 +35,7 @@ echoServer endpoint = do

35 event <- receive endpoint

36 case event of

37 ConnectionOpened cid rel addr -> do

38− tlog $ "Opened new conncetion " ++ show cid

38+ tlog $ "Opened new connection " ++ show cid

39 Right conn <- connect endpoint addr rel

40 go (Map.insert cid conn cs)

41 Received cid payload -> do