@@ -29,7 +29,7 @@ module Network.Transport.TCP ( -- * Main API
2929 -- $design
3030 ) where
3131
32−import Prelude hiding (catch)
32+import Prelude hiding (catch, mapM_)
3333import Network.Transport
3434import Network.Transport.Internal.TCP ( forkServer
3535 , recvWithLength
@@ -97,10 +97,10 @@ import qualified Data.IntMap as IntMap (empty)
9797import Data.IntSet (IntSet)
9898import qualified Data.IntSet as IntSet (empty, insert, elems, singleton, null, delete)
9999import Data.Map (Map)
100−import qualified Data.Map as Map (empty, elems)
100+import qualified Data.Map as Map (empty)
101101import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
102102import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
103−import Data.Foldable (forM_)
103+import Data.Foldable (forM_, mapM_)
104104import System.IO (hPutStrLn, stderr)
105105
106106-- $design
@@ -358,13 +358,8 @@ data TransportInternals = TransportInternals
358358--
359359-- TODOs: deal with hints
360360createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)
361−createTransport host port = do
362− result <- createTransportExposeInternals host port
363− case result of
364− Left err ->
365− return $ Left err
366− Right (transport, _) ->
367− return $ Right transport
361+createTransport host port =
362+ either Left (Right . fst) <$> createTransportExposeInternals host port
368363
369364-- | You should probably not use this function (used for unit testing only)
370365createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, TransportInternals))
@@ -382,12 +377,14 @@ createTransportExposeInternals host port = do
382377 -- http://tangentsoft.net/wskfaq/advanced.html#backlog
383378 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
384379 tid <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
385− return ( Transport { newEndPoint = apiNewEndPoint transport
386− , closeTransport = apiCloseTransport transport (Just tid) [EndPointClosed]
387− }
388− , TransportInternals { transportThread = tid
389− , socketBetween = internalSocketBetween transport
390− }
380+ return ( Transport
381+ { newEndPoint = apiNewEndPoint transport
382+ , closeTransport = apiCloseTransport transport (Just tid) [EndPointClosed]
383+ }
384+ , TransportInternals
385+ { transportThread = tid
386+ , socketBetween = internalSocketBetween transport
387+ }
391388 )
392389 where
393390 terminationHandler :: TCPTransport -> SomeException -> IO ()
@@ -405,13 +402,11 @@ apiCloseTransport transport mTransportThread evs = do
405402 mTSt <- modifyMVar (transportState transport) $ \st -> case st of
406403 TransportValid vst -> return (TransportClosed, Just vst)
407404 TransportClosed -> return (TransportClosed, Nothing)
408− forM_ mTSt $ \tSt -> do
409− mapM_ (apiCloseEndPoint transport evs) (Map.elems $ tSt ^. localEndPoints)
410− forM_ mTransportThread $ \tid -> do
411− -- This will invoke the termination handler, which in turn will call
412− -- apiCloseTransport again, but at that point the transport is
413− -- already in closed state so at that point we return immediately
414− killThread tid
405+ forM_ mTSt $ mapM_ (apiCloseEndPoint transport evs) . (^. localEndPoints)
406+ -- This will invoke the termination handler, which in turn will call
407+ -- apiCloseTransport again, but then the transport will already be closed and
408+ -- we won't be passed a transport thread, so we terminate immmediate
409+ forM_ mTransportThread killThread
415410
416411-- | Create a new endpoint
417412apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
@@ -442,9 +437,10 @@ apiConnect ourEndPoint theirAddress _reliability _hints = try $ do
442437 -- connAlive can be an IORef rather than an MVar because it is protected by
443438 -- the remoteState MVar. We don't need the overhead of locking twice.
444439 connAlive <- newIORef True
445− return $ Connection { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
446− , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
447− }
440+ return Connection
441+ { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
442+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
443+ }
448444
449445-- | Close a connection
450446--
@@ -468,9 +464,9 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
468464 else
469465 return (RemoteEndPointValid vst))
470466 -- RemoteEndPointClosing
471− (\(resolved, vst) -> return $ RemoteEndPointClosing resolved vst)
467+ (return . uncurry RemoteEndPointClosing)
472468 -- RemoteEndPoinClosed
473− (return $ RemoteEndPointClosed)
469+ (return RemoteEndPointClosed)
474470 closeIfUnused (ourEndPoint, theirEndPoint)
475471
476472-- | Send data across a connection
@@ -478,8 +474,8 @@ apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
478474-- RELY: The endpoint must not be in invalid state.
479475-- GUARANTEE: The state of the remote endpoint will not be changed.
480476apiSend :: EndPointPair -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
481−apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = do
482− join <$> (try . mapExceptionIO sendFailed $ modifyRemoteState (ourEndPoint, theirEndPoint)
477+apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
478+ join <$> (try . mapExceptionIO sendFailed $ withRemoteState (ourEndPoint, theirEndPoint)
483479 -- RemoteEndPointInvalid
484480 (\_ -> fail "apiSend RELY violation")
485481 -- RemoteEndPointValid
@@ -488,13 +484,13 @@ apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = do
488484 if alive
489485 then do
490486 sendOn vst (encodeInt32 connId : prependLength payload)
491− return (RemoteEndPointValid vst, Right $ ())
487+ return . Right $ ()
492488 else
493− return (RemoteEndPointValid vst, Left $ TransportError SendClosed "Connection closed"))
489+ return . Left $ TransportError SendClosed "Connection closed")
494490 -- RemoteEndPointClosing
495− (\(resolved, vst) -> return (RemoteEndPointClosing resolved vst, Left $ TransportError SendClosed "Connection lost"))
491+ (return . Left . const (TransportError SendClosed "Connection lost"))
496492 -- RemoteEndPointClosed
497− (return (RemoteEndPointClosed, Left $ TransportError SendClosed "Connection lost")))
493+ (return . Left $ TransportError SendClosed "Connection lost"))
498494 where
499495 sendFailed :: IOException -> TransportError SendErrorCode
500496 sendFailed = TransportError SendFailed . show
@@ -516,13 +512,13 @@ apiCloseEndPoint transport evs ourEndPoint = do
516512 return (LocalEndPointClosed, Nothing)
517513 forM_ mOurState $ \vst -> do
518514 -- Close all endpoints and kill all threads
519− forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket
515+ forM_ (vst ^. localConnections) tryCloseRemoteSocket
520516 forM_ (vst ^. internalThreads) killThread
521517 forM_ evs $ writeChan (localChannel ourEndPoint)
522518 where
523519 -- Close the remote socket and return the set of all incoming connections
524520 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
525− tryCloseRemoteSocket theirEndPoint = do
521+ tryCloseRemoteSocket theirEndPoint =
526522 -- We make an attempt to close the connection nicely (by sending a CloseSocket first)
527523 modifyMVar_ (remoteState theirEndPoint) $ \st ->
528524 case st of
@@ -560,7 +556,7 @@ connectToSelf ourEndPoint = do
560556 ourChan = localChannel ourEndPoint
561557
562558 selfSend :: MVar Bool -> ConnectionId -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
563− selfSend connAlive connId msg = do
559+ selfSend connAlive connId msg =
564560 modifyMVar connAlive $ \alive ->
565561 if alive
566562 then do
@@ -570,7 +566,7 @@ connectToSelf ourEndPoint = do
570566 return (alive, Left (TransportError SendFailed "Connection closed"))
571567
572568 selfClose :: MVar Bool -> ConnectionId -> IO ()
573− selfClose connAlive connId = do
569+ selfClose connAlive connId =
574570 modifyMVar_ connAlive $ \alive -> do
575571 when alive $ writeChan ourChan (ConnectionClosed connId)
576572 return False
@@ -585,23 +581,23 @@ internalSocketBetween :: TCPTransport -- ^ Transport
585581 -> EndPointAddress -- ^ Remote endpoint
586582 -> IO (Either String N.Socket)
587583internalSocketBetween transport ourAddress theirAddress = runErrorT $ do
588− ourEndPoint <- ErrorT $ do
584+ ourEndPoint <- ErrorT $
589585 withMVar (transportState transport) $ \st -> case st of
590586 TransportClosed ->
591587 return . Left $ "Transport closed"
592− TransportValid vst -> do
588+ TransportValid vst ->
593589 case vst ^. localEndPointAt ourAddress of
594590 Nothing -> return . Left $ "Local endpoint not found"
595591 Just ep -> return . Right $ ep
596− theirEndPoint <- ErrorT $ do
592+ theirEndPoint <- ErrorT $
597593 withMVar (localState ourEndPoint) $ \st -> case st of
598594 LocalEndPointClosed ->
599595 return . Left $ "Local endpoint closed"
600− LocalEndPointValid vst -> do
596+ LocalEndPointValid vst ->
601597 case vst ^. localConnectionTo theirAddress of
602598 Nothing -> return . Left $ "Remote endpoint not found"
603599 Just ep -> return . Right $ ep
604− ErrorT $ do
600+ ErrorT $
605601 withMVar (remoteState theirEndPoint) $ \st -> case st of
606602 RemoteEndPointValid vst ->
607603 return . Right $ remoteSocket vst
@@ -679,7 +675,7 @@ requestConnectionTo ourEndPoint theirAddress = go
679675 -- endpoint state, and might have changed in the meantime, these changes
680676 -- won't matter.
681677 case endPointStateSnapshot of
682− RemoteEndPointInvalid err -> do
678+ RemoteEndPointInvalid err ->
683679 throw err
684680
685681 RemoteEndPointClosing resolved _ ->
@@ -687,7 +683,7 @@ requestConnectionTo ourEndPoint theirAddress = go
687683 -- this is resolved and we then try again
688684 readMVar resolved >> go
689685
690− RemoteEndPointClosed -> do
686+ RemoteEndPointClosed ->
691687 -- EndPointClosed indicates that a concurrent thread was in the
692688 -- process of closing the TCP connection to the remote endpoint when
693689 -- we obtained a reference to it. By INV-CLOSE we can assume that the
@@ -705,7 +701,7 @@ requestConnectionTo ourEndPoint theirAddress = go
705701 case decodeInt32 . BS.concat $ reply of
706702 Nothing ->
707703 throw (connectFailed $ userError "Invalid integer")
708− Just cid -> do
704+ Just cid ->
709705 return (theirEndPoint, cid)
710706
711707 -- If this is a new endpoint, fork a thread to listen for incoming
@@ -750,14 +746,14 @@ requestConnectionTo ourEndPoint theirAddress = go
750746setupRemoteEndPoint :: EndPointPair -> IO ()
751747setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
752748 didAccept <- bracketOnError (socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint))
753− onError $ \result -> do
749+ onError $ \result ->
754750 case result of
755751 Right (sock, ConnectionRequestAccepted) -> do
756752 let vst = ValidRemoteEndPointState
757753 { remoteSocket = sock
758754 , _remoteOutgoing = 0
759755 , _remoteIncoming = IntSet.empty
760− , sendOn = \msg -> sendMany sock msg `onException` (tryIO $ N.sClose sock)
756+ , sendOn = \msg -> sendMany sock msg `onException` tryIO (N.sClose sock)
761757 }
762758 putMVar theirState (RemoteEndPointValid vst)
763759 return True
@@ -827,7 +823,7 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
827823 Just r -> return (sock, r)
828824 where
829825 createSocket :: N.AddrInfo -> IO N.Socket
830− createSocket addr = mapExceptionIO insufficientResources $ do
826+ createSocket addr = mapExceptionIO insufficientResources $
831827 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
832828
833829 invalidAddress, insufficientResources, failed :: IOException -> TransportError ConnectErrorCode
@@ -839,7 +835,7 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
839835--
840836-- If the local endpoint is closed, do nothing
841837removeRemoteEndPoint :: EndPointPair -> IO ()
842−removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do
838+removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
843839 modifyMVar_ ourState $ \st -> case st of
844840 LocalEndPointValid vst ->
845841 case vst ^. localConnectionTo theirAddress of
@@ -859,7 +855,7 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do
859855--
860856-- Does nothing if the transport is closed
861857removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
862−removeLocalEndPoint transport ourEndPoint = do
858+removeLocalEndPoint transport ourEndPoint =
863859 modifyMVar_ (transportState transport) $ \st -> case st of
864860 TransportValid vst ->
865861 return (TransportValid . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) $ vst)
@@ -971,7 +967,7 @@ modifyRemoteState :: EndPointPair
971967 -> ((MVar (), ValidRemoteEndPointState) -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointClosing
972968 -> IO (RemoteState, a) -- ^ Case for RemoteEndPointClosed
973969 -> IO a
974−modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed = do
970+modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =
975971 mask $ \restore -> do
976972 st <- takeMVar theirState
977973 case st of
@@ -981,7 +977,7 @@ modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing
981977 Right (st', a) -> do
982978 putMVar theirState st'
983979 return a
984− Left ex -> do
980+ Left ex ->
985981 handleException ex vst
986982 -- The other cases are less interesting, because unless the endpoint is
987983 -- in Valid state we're not supposed to do any IO on it
@@ -1058,13 +1054,12 @@ withRemoteState :: EndPointPair -- ^ Local and
10581054 -> ((MVar (), ValidRemoteEndPointState) -> IO a) -- ^ Case for RemoteEndPointClosing
10591055 -> IO a -- ^ Case for RemoteEndPointClosed
10601056 -> IO a
1061−withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed = do
1062− modifyRemoteState (ourEndPoint, theirEndPoint)
1063− (\err -> (,) (RemoteEndPointInvalid err) <$> caseInvalid err)
1064− (\vst -> (,) (RemoteEndPointValid vst) <$> caseValid vst)
1065− (\(resolved, vst) -> (,) (RemoteEndPointClosing resolved vst) <$> caseClosing (resolved, vst))
1066− ((,) RemoteEndPointClosed <$> caseClosed)
1067−
1057+withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =
1058+ modifyRemoteState (ourEndPoint, theirEndPoint)
1059+ (\err -> (,) (RemoteEndPointInvalid err) <$> caseInvalid err)
1060+ (\vst -> (,) (RemoteEndPointValid vst) <$> caseValid vst)
1061+ (\(resolved, vst) -> (,) (RemoteEndPointClosing resolved vst) <$> caseClosing (resolved, vst))
1062+ ((,) RemoteEndPointClosed <$> caseClosed)
10681063
10691064--------------------------------------------------------------------------------
10701065-- Incoming requests --
@@ -1094,7 +1089,7 @@ handleConnectionRequest transport sock = handle tryCloseSocket $ do
10941089 fail "Invalid endpoint"
10951090 Just ourEndPoint ->
10961091 return ourEndPoint
1097− TransportClosed -> do
1092+ TransportClosed ->
10981093 fail "Transport closed"
10991094 void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress
11001095 where
@@ -1182,7 +1177,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
11821177 then do
11831178 readMessage sock connId
11841179 go sock
1185− else do
1180+ else
11861181 case tryToEnum (fromIntegral connId) of
11871182 Just RequestConnectionId -> do
11881183 recvInt32 sock >>= createNewConnection
@@ -1241,7 +1236,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12411236 fail "Local endpoint closed"
12421237 case mmvar of
12431238 Nothing -> do
1244− hPutStrLn stderr $ "Warning: Invalid request ID"
1239+ hPutStrLn stderr "Warning: Invalid request ID"
12451240 return () -- Invalid request ID. TODO: We just ignore it?
12461241 Just mvar ->
12471242 putMVar mvar response
@@ -1273,8 +1268,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
12731268 -- We regard a CloseSocket message as an (optimized) way for the
12741269 -- remote endpoint to indicate that all its connections to us are
12751270 -- now properly closed
1276− forM_ (IntSet.elems $ vst ^. remoteIncoming) $ \cid ->
1277− writeChan ourChannel (ConnectionClosed cid)
1271+ forM_ (IntSet.elems $ vst ^. remoteIncoming) $
1272+ writeChan ourChannel . ConnectionClosed
12781273 let vst' = remoteIncoming ^= IntSet.empty $ vst
12791274 -- Check if we agree that the connection should be closed
12801275 if vst' ^. remoteOutgoing == 0