@@ -86,7 +86,7 @@ import Control.Concurrent.MVar
8686 )
8787import Control.Category ((>>>))
8888import Control.Applicative ((<$>))
89−import Control.Monad (when, unless)
89+import Control.Monad (when, unless, join)
9090import Control.Exception
9191 ( IOException
9292 , SomeException
@@ -96,9 +96,8 @@ import Control.Exception
9696 , throwIO
9797 , try
9898 , bracketOnError
99− , mask
100− , onException
10199 , fromException
100+ , catch
102101 )
103102import Data.IORef (IORef, newIORef, writeIORef, readIORef)
104103import Data.ByteString (ByteString)
@@ -345,9 +344,10 @@ data ValidLocalEndPointState = ValidLocalEndPointState
345344-- modifyRemoteState.
346345
347346data RemoteEndPoint = RemoteEndPoint
348− { remoteAddress :: !EndPointAddress
349− , remoteState :: !(MVar RemoteState)
350− , remoteId :: !Int
347+ { remoteAddress :: !EndPointAddress
348+ , remoteState :: !(MVar RemoteState)
349+ , remoteId :: !Int
350+ , remoteScheduled :: !(Chan (IO ()))
351351 }
352352
353353data RequestedBy = RequestedByUs | RequestedByThem
@@ -571,22 +571,27 @@ apiConnect params ourEndPoint theirAddress _reliability hints =
571571apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()
572572apiClose (ourEndPoint, theirEndPoint) connId connAlive =
573573 void . tryIO . asyncWhenCancelled return $ do
574− modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity
575− { caseValid = \vst -> do
576− alive <- readIORef connAlive
577− if alive
578− then do
579− writeIORef connAlive False
580− sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
581− return ( RemoteEndPointValid
582− . (remoteOutgoing ^: (\x -> x - 1))
583− $ vst
584− )
585− else
586− return (RemoteEndPointValid vst)
587− }
574+ mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
575+ RemoteEndPointValid vst -> do
576+ alive <- readIORef connAlive
577+ if alive
578+ then do
579+ writeIORef connAlive False
580+ act <- schedule theirEndPoint $
581+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
582+ return ( RemoteEndPointValid
583+ . (remoteOutgoing ^: (\x -> x - 1))
584+ $ vst
585+ , Just act
586+ )
587+ else
588+ return (RemoteEndPointValid vst, Nothing)
589+ _ ->
590+ return (st, Nothing)
591+ forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
588592 closeIfUnused (ourEndPoint, theirEndPoint)
589593
594+
590595-- | Send data across a connection
591596apiSend :: EndPointPair -- ^ Local and remote endpoint
592597 -> ConnectionId -- ^ Connection ID (supplied by remote endpoint)
@@ -595,33 +600,33 @@ apiSend :: EndPointPair -- ^ Local and remote endpoint
595600 -> IO (Either (TransportError SendErrorCode) ())
596601apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
597602 -- We don't need the overhead of asyncWhenCancelled here
598− try . mapIOException sendFailed $
599− withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
600− { caseInvalid = \_ ->
601− relyViolation (ourEndPoint, theirEndPoint) "apiSend"
602− , caseInit = \_ _ ->
603− relyViolation (ourEndPoint, theirEndPoint) "apiSend"
604− , caseValid = \vst -> do
605− alive <- readIORef connAlive
606− if alive
607− then sendOn vst (encodeInt32 connId : prependLength payload)
608− else throwIO $ TransportError SendClosed "Connection closed"
609− , caseClosing = \_ _ -> do
610− alive <- readIORef connAlive
611− if alive
612− then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
613− else throwIO $ TransportError SendClosed "Connection closed"
614− , caseClosed = do
615− alive <- readIORef connAlive
616− if alive
617− then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
618− else throwIO $ TransportError SendClosed "Connection closed"
619− , caseFailed = \err -> do
620− alive <- readIORef connAlive
621− if alive
622− then throwIO $ TransportError SendFailed (show err)
623− else throwIO $ TransportError SendClosed "Connection closed"
624− }
603+ try . mapIOException sendFailed $ do
604+ act <- withMVar (remoteState theirEndPoint) $ \st -> case st of
605+ RemoteEndPointInvalid _ ->
606+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"
607+ RemoteEndPointInit _ _ ->
608+ relyViolation (ourEndPoint, theirEndPoint) "apiSend"
609+ RemoteEndPointValid vst -> do
610+ alive <- readIORef connAlive
611+ if alive
612+ then schedule theirEndPoint $ sendOn vst (encodeInt32 connId : prependLength payload)
613+ else throwIO $ TransportError SendClosed "Connection closed"
614+ RemoteEndPointClosing _ _ -> do
615+ alive <- readIORef connAlive
616+ if alive
617+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
618+ else throwIO $ TransportError SendClosed "Connection closed"
619+ RemoteEndPointClosed -> do
620+ alive <- readIORef connAlive
621+ if alive
622+ then relyViolation (ourEndPoint, theirEndPoint) "apiSend"
623+ else throwIO $ TransportError SendClosed "Connection closed"
624+ RemoteEndPointFailed err -> do
625+ alive <- readIORef connAlive
626+ if alive
627+ then throwIO $ TransportError SendFailed (show err)
628+ else throwIO $ TransportError SendClosed "Connection closed"
629+ runScheduledAction (ourEndPoint, theirEndPoint) act
625630 where
626631 sendFailed = TransportError SendFailed . show
627632
@@ -651,155 +656,28 @@ apiCloseEndPoint transport evs ourEndPoint =
651656 -- We make an attempt to close the connection nicely
652657 -- (by sending a CloseSocket first)
653658 let closed = RemoteEndPointFailed . userError $ "apiCloseEndPoint"
654− modifyMVar_ (remoteState theirEndPoint) $ \st ->
659+ mAct <- modifyMVar (remoteState theirEndPoint) $ \st ->
655660 case st of
656661 RemoteEndPointInvalid _ ->
657− return st
662+ return (st, Nothing)
658663 RemoteEndPointInit resolved _ -> do
659664 putMVar resolved ()
660− return closed
661− RemoteEndPointValid conn -> do
662− tryIO $ sendOn conn [encodeInt32 CloseSocket]
663− tryCloseSocket (remoteSocket conn)
664− return closed
665− RemoteEndPointClosing resolved conn -> do
665+ return (closed, Nothing)
666+ RemoteEndPointValid vst -> do
667+ act <- schedule theirEndPoint $ do
668+ tryIO $ sendOn vst [encodeInt32 CloseSocket]
669+ tryCloseSocket (remoteSocket vst)
670+ return (closed, Just act)
671+ RemoteEndPointClosing resolved vst -> do
666672 putMVar resolved ()
667− tryCloseSocket (remoteSocket conn)
668− return closed
673+ act <- schedule theirEndPoint $ tryCloseSocket (remoteSocket vst)
674+ return (closed, Just act)
669675 RemoteEndPointClosed ->
670− return st
676+ return (st, Nothing)
671677 RemoteEndPointFailed err ->
672− return $ RemoteEndPointFailed err
673−
674−--------------------------------------------------------------------------------
675−-- As soon as a remote connection fails, we want to put notify our endpoint --
676−-- and put it into a closed state. Since this may happen in many places, we --
677−-- provide some abstractions. --
678−--------------------------------------------------------------------------------
679−
680−data RemoteStatePatternMatch a = RemoteStatePatternMatch
681− { caseInvalid :: TransportError ConnectErrorCode -> IO a
682− , caseInit :: MVar () -> RequestedBy -> IO a
683− , caseValid :: ValidRemoteEndPointState -> IO a
684− , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
685− , caseClosed :: IO a
686− , caseFailed :: IOException -> IO a
687− }
688−
689−remoteStateIdentity :: RemoteStatePatternMatch RemoteState
690−remoteStateIdentity =
691− RemoteStatePatternMatch
692− { caseInvalid = return . RemoteEndPointInvalid
693− , caseInit = (return .) . RemoteEndPointInit
694− , caseValid = return . RemoteEndPointValid
695− , caseClosing = (return .) . RemoteEndPointClosing
696− , caseClosed = return RemoteEndPointClosed
697− , caseFailed = return . RemoteEndPointFailed
698− }
699−
700−-- | Like modifyMVar, but if an I/O exception occurs don't restore the remote
701−-- endpoint to its original value but close it instead
702−modifyRemoteState :: EndPointPair
703− -> RemoteStatePatternMatch (RemoteState, a)
704− -> IO a
705−modifyRemoteState (ourEndPoint, theirEndPoint) match =
706− mask $ \restore -> do
707− st <- takeMVar theirState
708− case st of
709− RemoteEndPointValid vst -> do
710− mResult <- try $ restore (caseValid match vst)
711− case mResult of
712− Right (st', a) -> do
713− putMVar theirState st'
714− return a
715− Left ex -> do
716− case fromException ex of
717− Just ioEx -> handleIOException ioEx vst
718− Nothing -> putMVar theirState st
719− throwIO ex
720− -- The other cases are less interesting, because unless the endpoint is
721− -- in Valid state we're not supposed to do any IO on it
722− RemoteEndPointInit resolved origin -> do
723− (st', a) <- onException (restore $ caseInit match resolved origin)
724− (putMVar theirState st)
725− putMVar theirState st'
726− return a
727− RemoteEndPointClosing resolved vst -> do
728− (st', a) <- onException (restore $ caseClosing match resolved vst)
729− (putMVar theirState st)
730− putMVar theirState st'
731− return a
732− RemoteEndPointInvalid err -> do
733− (st', a) <- onException (restore $ caseInvalid match err)
734− (putMVar theirState st)
735− putMVar theirState st'
736− return a
737− RemoteEndPointClosed -> do
738− (st', a) <- onException (restore $ caseClosed match)
739− (putMVar theirState st)
740− putMVar theirState st'
741− return a
742− RemoteEndPointFailed err -> do
743− (st', a) <- onException (restore $ caseFailed match err)
744− (putMVar theirState st)
745− putMVar theirState st'
746− return a
747− where
748− theirState :: MVar RemoteState
749− theirState = remoteState theirEndPoint
750−
751− handleIOException :: IOException -> ValidRemoteEndPointState -> IO ()
752− handleIOException ex vst = do
753− tryCloseSocket (remoteSocket vst)
754− putMVar theirState (RemoteEndPointFailed ex)
755− let code = EventConnectionLost (remoteAddress theirEndPoint)
756− err = TransportError code (show ex)
757− writeChan (localChannel ourEndPoint) $ ErrorEvent err
758−
759−-- | Like 'modifyRemoteState' but without a return value
760−modifyRemoteState_ :: EndPointPair
761− -> RemoteStatePatternMatch RemoteState
762− -> IO ()
763−modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
764− modifyRemoteState (ourEndPoint, theirEndPoint)
765− RemoteStatePatternMatch
766− { caseInvalid = u . caseInvalid match
767− , caseInit = \resolved origin -> u $ caseInit match resolved origin
768− , caseValid = u . caseValid match
769− , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
770− , caseClosed = u $ caseClosed match
771− , caseFailed = u . caseFailed match
772− }
773− where
774− u :: IO a -> IO (a, ())
775− u p = p >>= \a -> return (a, ())
776−
777−-- | Like 'modifyRemoteState' but without the ability to change the state
778−withRemoteState :: EndPointPair
779− -> RemoteStatePatternMatch a
780− -> IO a
781−withRemoteState (ourEndPoint, theirEndPoint) match =
782− modifyRemoteState (ourEndPoint, theirEndPoint)
783− RemoteStatePatternMatch
784− { caseInvalid = \err -> do
785− a <- caseInvalid match err
786− return (RemoteEndPointInvalid err, a)
787− , caseInit = \resolved origin -> do
788− a <- caseInit match resolved origin
789− return (RemoteEndPointInit resolved origin, a)
790− , caseValid = \vst -> do
791− a <- caseValid match vst
792− return (RemoteEndPointValid vst, a)
793− , caseClosing = \resolved vst -> do
794− a <- caseClosing match resolved vst
795− return (RemoteEndPointClosing resolved vst, a)
796− , caseClosed = do
797− a <- caseClosed match
798− return (RemoteEndPointClosed, a)
799− , caseFailed = \err -> do
800− a <- caseFailed match err
801− return (RemoteEndPointFailed err, a)
802− }
678+ return (RemoteEndPointFailed err, Nothing)
679+ forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
680+
803681
804682--------------------------------------------------------------------------------
805683-- Incoming requests --
@@ -1029,8 +907,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1029907
1030908 -- Close the socket (if we don't have any outgoing connections)
1031909 closeSocket :: N.Socket -> IO Bool
1032− closeSocket sock =
1033− modifyMVar theirState $ \st ->
910+ closeSocket sock = do
911+ mAct <- modifyMVar theirState $ \st ->
1034912 case st of
1035913 RemoteEndPointInvalid _ ->
1036914 relyViolation (ourEndPoint, theirEndPoint)
@@ -1050,22 +928,28 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1050928 then do
1051929 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1052930 -- Attempt to reply (but don't insist)
1053− tryIO $ sendOn vst' [encodeInt32 CloseSocket]
1054− tryCloseSocket sock
1055− return (RemoteEndPointClosed, True)
931+ act <- schedule theirEndPoint $ do
932+ tryIO $ sendOn vst' [encodeInt32 CloseSocket]
933+ tryCloseSocket sock
934+ return (RemoteEndPointClosed, Just act)
1056935 else
1057− return (RemoteEndPointValid vst', False)
936+ return (RemoteEndPointValid vst', Nothing)
1058937 RemoteEndPointClosing resolved _ -> do
1059938 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1060− tryCloseSocket sock
939+ act <- schedule theirEndPoint $ tryCloseSocket sock
1061940 putMVar resolved ()
1062− return (RemoteEndPointClosed, True)
941+ return (RemoteEndPointClosed, Just act)
1063942 RemoteEndPointFailed err ->
1064943 throwIO err
1065944 RemoteEndPointClosed ->
1066945 relyViolation (ourEndPoint, theirEndPoint)
1067946 "handleIncomingMessages:closeSocket (closed)"
1068−
947+ case mAct of
948+ Nothing -> return False
949+ Just act -> do
950+ runScheduledAction (ourEndPoint, theirEndPoint) act
951+ return True
952+
1069953 -- Read a message and output it on the endPoint's channel. By rights we
1070954 -- should verify that the connection ID is valid, but this is unnecessary
1071955 -- overhead
@@ -1191,27 +1075,28 @@ setupRemoteEndPoint params (ourEndPoint, theirEndPoint) hints = do
11911075doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
11921076doRemoteRequest (ourEndPoint, theirEndPoint) header = do
11931077 replyMVar <- newEmptyMVar
1194− modifyRemoteState_ (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
1195− { caseValid = \vst -> do
1196− let reqId = vst ^. nextCtrlRequestId
1197− sendOn vst [encodeInt32 header, encodeInt32 reqId]
1198− return ( RemoteEndPointValid
1199− . (nextCtrlRequestId ^: (+ 1))
1200− . (pendingCtrlRequestsAt reqId ^= Just replyMVar)
1201− $ vst
1202− )
1078+ act <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1079+ RemoteEndPointValid vst -> do
1080+ let reqId = vst ^. nextCtrlRequestId
1081+ act <- schedule theirEndPoint $ sendOn vst [encodeInt32 header, encodeInt32 reqId]
1082+ return ( RemoteEndPointValid
1083+ . (nextCtrlRequestId ^: (+ 1))
1084+ . (pendingCtrlRequestsAt reqId ^= Just replyMVar)
1085+ $ vst
1086+ , act
1087+ )
12031088 -- Error cases
1204− , caseInvalid =
1205− throwIO
1206− , caseInit = \_ _ ->
1207− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
1208− , caseClosing = \_ _ ->
1209− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
1210− , caseClosed =
1211− relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)"
1212− , caseFailed =
1213− throwIO
1214− }
1089+ RemoteEndPointInvalid err ->
1090+ throwIO err
1091+ RemoteEndPointInit _ _ ->
1092+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (init)"
1093+ RemoteEndPointClosing _ _ ->
1094+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closing)"
1095+ RemoteEndPointClosed ->
1096+ relyViolation (ourEndPoint, theirEndPoint) "doRemoteRequest (closed)"
1097+ RemoteEndPointFailed err ->
1098+ throwIO err
1099+ runScheduledAction (ourEndPoint, theirEndPoint) act
12151100 mReply <- takeMVar replyMVar
12161101 case mReply of
12171102 Left err -> throwIO err
@@ -1219,17 +1104,19 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do
12191104
12201105-- | Send a CloseSocket request if the remote endpoint is unused
12211106closeIfUnused :: EndPointPair -> IO ()
1222−closeIfUnused (ourEndPoint, theirEndPoint) =
1223− modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity
1224− { caseValid = \vst ->
1225− if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
1226− then do
1227− sendOn vst [encodeInt32 CloseSocket]
1228− resolved <- newEmptyMVar
1229− return $ RemoteEndPointClosing resolved vst
1230− else
1231− return $ RemoteEndPointValid vst
1232− }
1107+closeIfUnused (ourEndPoint, theirEndPoint) = do
1108+ mAct <- modifyMVar (remoteState theirEndPoint) $ \st -> case st of
1109+ RemoteEndPointValid vst ->
1110+ if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
1111+ then do
1112+ resolved <- newEmptyMVar
1113+ act <- schedule theirEndPoint $ sendOn vst [encodeInt32 CloseSocket]
1114+ return (RemoteEndPointClosing resolved vst, Just act)
1115+ else
1116+ return (RemoteEndPointValid vst, Nothing)
1117+ _ ->
1118+ return (st, Nothing)
1119+ forM_ mAct $ runScheduledAction (ourEndPoint, theirEndPoint)
12331120
12341121-- | Reset a remote endpoint if it is in Invalid mode
12351122--
@@ -1419,12 +1306,14 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
14191306 Just theirEndPoint ->
14201307 return (st, (theirEndPoint, False))
14211308 Nothing -> do
1422− resolved <- newEmptyMVar
1309+ resolved <- newEmptyMVar
14231310 theirState <- newMVar (RemoteEndPointInit resolved findOrigin)
1311+ scheduled <- newChan
14241312 let theirEndPoint = RemoteEndPoint
1425− { remoteAddress = theirAddress
1426− , remoteState = theirState
1427− , remoteId = vst ^. nextRemoteId
1313+ { remoteAddress = theirAddress
1314+ , remoteState = theirState
1315+ , remoteId = vst ^. nextRemoteId
1316+ , remoteScheduled = scheduled
14281317 }
14291318 return ( LocalEndPointValid
14301319 . (localConnectionTo theirAddress ^= Just theirEndPoint)
@@ -1489,6 +1378,55 @@ findRemoteEndPoint ourEndPoint theirAddress findOrigin = go
14891378 ourState = localState ourEndPoint
14901379 ourAddress = localAddress ourEndPoint
14911380
1381+--------------------------------------------------------------------------------
1382+-- Scheduling actions --
1383+--------------------------------------------------------------------------------
1384+
1385+-- | See 'schedule'/'runScheduledAction'
1386+type Action a = MVar (Either SomeException a)
1387+
1388+-- | Schedule an action to be executed (see also 'runScheduledAction')
1389+schedule :: RemoteEndPoint -> IO a -> IO (Action a)
1390+schedule theirEndPoint act = do
1391+ mvar <- newEmptyMVar
1392+ writeChan (remoteScheduled theirEndPoint) $
1393+ catch (act >>= putMVar mvar . Right) (putMVar mvar . Left)
1394+ return mvar
1395+
1396+-- | Run a scheduled action. Every call to 'schedule' should be paired with a
1397+-- call to 'runScheduledAction' so that every scheduled action is run. Note
1398+-- however that the there is no guarantee that in
1399+--
1400+-- > do act <- schedule p
1401+-- > runScheduledAction
1402+--
1403+-- 'runScheduledAction' will run @p@ (it might run some other scheduled action).
1404+-- However, it will then wait until @p@ is executed (by this call to
1405+-- 'runScheduledAction' or by another).
1406+runScheduledAction :: EndPointPair -> Action a -> IO a
1407+runScheduledAction (ourEndPoint, theirEndPoint) mvar = do
1408+ join $ readChan (remoteScheduled theirEndPoint)
1409+ ma <- readMVar mvar
1410+ case ma of
1411+ Right a -> return a
1412+ Left e -> do
1413+ forM_ (fromException e) $ \ioe ->
1414+ modifyMVar_ (remoteState theirEndPoint) $ \st ->
1415+ case st of
1416+ RemoteEndPointValid vst -> handleIOException ioe vst
1417+ _ -> return (RemoteEndPointFailed ioe)
1418+ throwIO e
1419+ where
1420+ handleIOException :: IOException
1421+ -> ValidRemoteEndPointState
1422+ -> IO RemoteState
1423+ handleIOException ex vst = do
1424+ tryCloseSocket (remoteSocket vst)
1425+ let code = EventConnectionLost (remoteAddress theirEndPoint)
1426+ err = TransportError code (show ex)
1427+ writeChan (localChannel ourEndPoint) $ ErrorEvent err
1428+ return (RemoteEndPointFailed ex)
1429+
14921430--------------------------------------------------------------------------------
14931431-- "Stateless" (MVar free) functions --
14941432--------------------------------------------------------------------------------