@@ -34,11 +34,12 @@ import Network.Transport
34import Network.Transport.Internal.TCP ( forkServer
34import Network.Transport.Internal.TCP ( forkServer
35 , recvWithLength
35 , recvWithLength
36 , recvInt32
36 , recvInt32
38import Network.Transport.Internal ( encodeInt32
39import Network.Transport.Internal ( encodeInt32
39 , decodeInt32
40 , decodeInt32
40 , prependLength
41 , prependLength
41− , mapExceptionIO
42+ , mapIOException
43 , tryToEnum
44 , tryToEnum
@@ -46,7 +47,6 @@ import Network.Transport.Internal ( encodeInt32
46import qualified Network.Socket as N ( HostName
47import qualified Network.Socket as N ( HostName
47 , ServiceName
48 , ServiceName
50 , getAddrInfo
50 , getAddrInfo
52 , addrFamily
52 , addrFamily
@@ -75,7 +75,7 @@ import Control.Concurrent.MVar ( MVar
76import Control.Category ((>>>))
76import Control.Category ((>>>))
77import Control.Applicative ((<$>))
77import Control.Applicative ((<$>))
78−import Control.Monad (when, unless, join)
78+import Control.Monad (when, unless, join, liftM)
79import Control.Monad.Error (ErrorT(..), runErrorT)
79import Control.Monad.Error (ErrorT(..), runErrorT)
80import Control.Exception ( IOException
80import Control.Exception ( IOException
81 , SomeException
81 , SomeException
@@ -95,13 +95,18 @@ import Data.Int (Int32)
95import Data.IntMap (IntMap)
95import Data.IntMap (IntMap)
96import qualified Data.IntMap as IntMap (empty)
96import qualified Data.IntMap as IntMap (empty)
97import Data.IntSet (IntSet)
97import Data.IntSet (IntSet)
98−import qualified Data.IntSet as IntSet (empty, insert, elems, singleton, null, delete)
98+import qualified Data.IntSet as IntSet ( empty
99import Data.Map (Map)
105import Data.Map (Map)
100import qualified Data.Map as Map (empty)
106import qualified Data.Map as Map (empty)
101import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
107import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
102import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
108import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
103import Data.Foldable (forM_, mapM_)
109import Data.Foldable (forM_, mapM_)
104−import System.IO (hPutStrLn, stderr)
106-- $design
111-- $design
@@ -236,12 +241,12 @@ data LocalEndPointState =
236 | LocalEndPointClosed
241 | LocalEndPointClosed
238data ValidLocalEndPointState = ValidLocalEndPointState
243data ValidLocalEndPointState = ValidLocalEndPointState
239− { _nextConnectionId :: ConnectionId
244+ { _nextConnectionId :: !ConnectionId
240 , _pendingCtrlRequests :: IntMap (MVar [ByteString])
245 , _pendingCtrlRequests :: IntMap (MVar [ByteString])
241− , _nextCtrlRequestId :: ControlRequestId
246+ , _nextCtrlRequestId :: !ControlRequestId
242 , _localConnections :: Map EndPointAddress RemoteEndPoint
247 , _localConnections :: Map EndPointAddress RemoteEndPoint
243 , _internalThreads :: [ThreadId]
248 , _internalThreads :: [ThreadId]
244− , _nextRemoteId :: Int
249+ , _nextRemoteId :: !Int
247-- A remote endpoint has incoming and outgoing connections, and when the total
252-- A remote endpoint has incoming and outgoing connections, and when the total
@@ -329,17 +334,24 @@ type EndPointPair = (LocalEndPoint, RemoteEndPoint)
330-- | Control headers
335-- | Control headers
331data ControlHeader =
336data ControlHeader =
332− RequestConnectionId -- ^ Request a new connection ID from the remote endpoint
337+ -- | Request a new connection ID from the remote endpoint
333− | CloseConnection -- ^ Tell the remote endpoint we will no longer be using a connection
339+ -- | Tell the remote endpoint we will no longer be using a connection
334− | ControlResponse -- ^ Respond to a control request /from/ the remote endpoint
341+ -- | Respond to a control request /from/ the remote endpoint
335− | CloseSocket -- ^ Request to close the connection (see module description)
343+ -- | Request to close the connection (see module description)
336 deriving (Enum, Bounded, Show)
345 deriving (Enum, Bounded, Show)
338-- Response sent by /B/ to /A/ when /A/ tries to connect
347-- Response sent by /B/ to /A/ when /A/ tries to connect
339data ConnectionRequestResponse =
348data ConnectionRequestResponse =
340− ConnectionRequestAccepted -- ^ /B/ accepts the connection
349+ -- | /B/ accepts the connection
350+ ConnectionRequestAccepted
341− | ConnectionRequestEndPointInvalid -- ^ /A/ requested an invalid endpoint
351+ -- | /A/ requested an invalid endpoint
352+ | ConnectionRequestEndPointInvalid
342− | ConnectionRequestCrossed -- ^ /A/s request crossed with a request from /B/ (see protocols)
353+ -- | /A/s request crossed with a request from /B/ (see protocols)
354+ | ConnectionRequestCrossed
343 deriving (Enum, Bounded, Show)
355 deriving (Enum, Bounded, Show)
345-- Internal functionality we expose for unit testing
357-- Internal functionality we expose for unit testing
@@ -347,7 +359,9 @@ data TransportInternals = TransportInternals
347 { -- | The ID of the thread that listens for new incoming connections
359 { -- | The ID of the thread that listens for new incoming connections
348 transportThread :: ThreadId
360 transportThread :: ThreadId
349 -- | Find the socket between a local and a remote endpoint
361 -- | Find the socket between a local and a remote endpoint
350− , socketBetween :: EndPointAddress -> EndPointAddress -> IO (Either String N.Socket)
362+ , socketBetween :: EndPointAddress
364+ -> IO (Either String N.Socket)
353--------------------------------------------------------------------------------
367--------------------------------------------------------------------------------
@@ -357,12 +371,17 @@ data TransportInternals = TransportInternals
357-- | Create a TCP transport
371-- | Create a TCP transport
359-- TODOs: deal with hints
373-- TODOs: deal with hints
374+createTransport :: N.HostName
360−createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)
376+ -> IO (Either IOException Transport)
361createTransport host port =
377createTransport host port =
362 either Left (Right . fst) <$> createTransportExposeInternals host port
378 either Left (Right . fst) <$> createTransportExposeInternals host port
364-- | You should probably not use this function (used for unit testing only)
380-- | You should probably not use this function (used for unit testing only)
381+createTransportExposeInternals
365−createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, TransportInternals))
384+ -> IO (Either IOException (Transport, TransportInternals))
366createTransportExposeInternals host port = do
385createTransportExposeInternals host port = do
367 state <- newMVar . TransportValid $ ValidTransportState
386 state <- newMVar . TransportValid $ ValidTransportState
368 { _localEndPoints = Map.empty
387 { _localEndPoints = Map.empty
@@ -373,19 +392,20 @@ createTransportExposeInternals host port = do
373 , transportPort = port
392 , transportPort = port
375 tryIO $ do
394 tryIO $ do
376− -- For a discussion of the use of N.sOMAXCONN, see
395+ tid <- forkServer host port N.sOMAXCONN
377− -- http://tangentsoft.net/wskfaq/advanced.html#backlog
396+ (terminationHandler transport)
378− -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
379− tid <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
397+ (handleConnectionRequest transport)
380− return ( Transport
399+ ( Transport
381− { newEndPoint = apiNewEndPoint transport
400+ { newEndPoint = apiNewEndPoint transport
382− , closeTransport = apiCloseTransport transport (Just tid) [EndPointClosed]
401+ , closeTransport = let evs = [EndPointClosed] in
402+ apiCloseTransport transport (Just tid) evs
384− , TransportInternals
404+ , TransportInternals
385− { transportThread = tid
405+ { transportThread = tid
386− , socketBetween = internalSocketBetween transport
406+ , socketBetween = internalSocketBetween transport
390 terminationHandler :: TCPTransport -> SomeException -> IO ()
410 terminationHandler :: TCPTransport -> SomeException -> IO ()
391 terminationHandler transport ex = do
411 terminationHandler transport ex = do
@@ -409,7 +429,8 @@ apiCloseTransport transport mTransportThread evs = do
409 forM_ mTransportThread killThread
429 forM_ mTransportThread killThread
411-- | Create a new endpoint
431-- | Create a new endpoint
432+apiNewEndPoint :: TCPTransport
412−apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
433+ -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
413apiNewEndPoint transport = try $ do
434apiNewEndPoint transport = try $ do
414 ourEndPoint <- createLocalEndPoint transport
435 ourEndPoint <- createLocalEndPoint transport
415 return EndPoint
436 return EndPoint
@@ -421,8 +442,10 @@ apiNewEndPoint transport = try $ do
421 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
442 , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
445+ newMulticastGroupError =
424− newMulticastGroupError = TransportError NewMulticastGroupUnsupported "TCP does not support multicast"
446+ TransportError NewMulticastGroupUnsupported "Multicast not supported"
447+ resolveMulticastGroupError =
425− resolveMulticastGroupError = TransportError ResolveMulticastGroupUnsupported "TCP does not support multicast"
448+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
427-- | Connnect to an endpoint
450-- | Connnect to an endpoint
428apiConnect :: LocalEndPoint -- ^ Local end point
451apiConnect :: LocalEndPoint -- ^ Local end point
@@ -430,78 +453,80 @@ apiConnect :: LocalEndPoint -- ^ Local end point
430 -> Reliability -- ^ Reliability (ignored)
453 -> Reliability -- ^ Reliability (ignored)
431 -> ConnectHints -- ^ Hints (ignored for now)
454 -> ConnectHints -- ^ Hints (ignored for now)
432 -> IO (Either (TransportError ConnectErrorCode) Connection)
455 -> IO (Either (TransportError ConnectErrorCode) Connection)
433−apiConnect ourEndPoint theirAddress _reliability _hints | localAddress ourEndPoint == theirAddress =
456+apiConnect ourEndPoint theirAddress _reliability _hints =
457+ if localAddress ourEndPoint == theirAddress
434− connectToSelf ourEndPoint
458+ then connectToSelf ourEndPoint
435−apiConnect ourEndPoint theirAddress _reliability _hints = try $ do
459+ else try $ do
436− (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress
460+ (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress
437− -- connAlive can be an IORef rather than an MVar because it is protected by
461+ -- connAlive can be an IORef rather than an MVar because it is protected
438− -- the remoteState MVar. We don't need the overhead of locking twice.
462+ -- by the remoteState MVar. We don't need the overhead of locking twice.
439− connAlive <- newIORef True
463+ connAlive <- newIORef True
440− return Connection
464+ return Connection
441− { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
465+ { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
442− , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
466+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
445-- | Close a connection
469-- | Close a connection
447-- RELY: The endpoint must not be invalid
471-- RELY: The endpoint must not be invalid
448-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
472-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
449−-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state
473+-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the
450−-- of the remote endpoint will not be changed.
474+-- state of the remote endpoint will not be changed.
451apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()
475apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()
452apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
476apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
453− modifyRemoteState_ (ourEndPoint, theirEndPoint)
477+ modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity
454− -- RemoteEndPointInvalid
455− (\_ -> fail "apiClose RELY violation")
456− -- RemoteEndPointValid
457− (\vst -> do
478+ { caseValid = \vst -> do
458− alive <- readIORef connAlive
479+ alive <- readIORef connAlive
459− if alive
480+ if alive
461− writeIORef connAlive False
482+ writeIORef connAlive False
462− sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
483+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
463− return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ vst)
484+ return ( RemoteEndPointValid
485+ . (remoteOutgoing ^: (\x -> x - 1))
465− return (RemoteEndPointValid vst))
489+ return (RemoteEndPointValid vst)
466− -- RemoteEndPointClosing
490+ }
467− (return . uncurry RemoteEndPointClosing)
468− -- RemoteEndPoinClosed
469− (return RemoteEndPointClosed)
470 closeIfUnused (ourEndPoint, theirEndPoint)
491 closeIfUnused (ourEndPoint, theirEndPoint)
472-- | Send data across a connection
493-- | Send data across a connection
474-- RELY: The endpoint must not be in invalid state.
495-- RELY: The endpoint must not be in invalid state.
475-- GUARANTEE: The state of the remote endpoint will not be changed.
496-- GUARANTEE: The state of the remote endpoint will not be changed.
497+apiSend :: EndPointPair -- ^ Local and remote endpoint
498+ -> ConnectionId -- ^ Connection ID (supplied by remote endpoint)
499+ -> IORef Bool -- ^ Is the connection still alive?
500+ -> [ByteString] -- ^ Payload
476−apiSend :: EndPointPair -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
501+ -> IO (Either (TransportError SendErrorCode) ())
477apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
502apiSend (ourEndPoint, theirEndPoint) connId connAlive payload =
478 -- The 'join' joins the inner exception that we explicitly return) for
503 -- The 'join' joins the inner exception that we explicitly return) for
479 -- instance if the connection is closed) with the outer exception (which is
504 -- instance if the connection is closed) with the outer exception (which is
480 -- returned by 'try' when an exception is thrown by 'sendOn', and handled
505 -- returned by 'try' when an exception is thrown by 'sendOn', and handled
481 -- by 'withRemoteState')
506 -- by 'withRemoteState')
507+ liftM join . try . mapIOException sendFailed $
482− join <$> (try . mapExceptionIO sendFailed $ withRemoteState (ourEndPoint, theirEndPoint)
508+ withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
483− -- RemoteEndPointInvalid
509+ { caseInvalid = \_ ->
484− (\_ -> fail "apiSend RELY violation")
510+ fail "apiSend RELY violation"
485− -- RemoteEndPointValid
486− (\vst -> do
511+ , caseValid = \vst -> do
487− alive <- readIORef connAlive
512+ alive <- readIORef connAlive
488− if alive
513+ if alive
489− then do
514+ then do
490− sendOn vst (encodeInt32 connId : prependLength payload)
515+ sendOn vst (encodeInt32 connId : prependLength payload)
491− return . Right $ ()
516+ return . Right $ ()
493− return . Left $ TransportError SendClosed "Connection closed")
518+ return . Left $ TransportError SendClosed "Connection closed"
494− -- RemoteEndPointClosing
519+ , caseClosing = \_ _ ->
495− (return . Left . const (TransportError SendClosed "Connection lost"))
520+ return . Left $ TransportError SendClosed "Connection lost"
496− -- RemoteEndPointClosed
521+ , caseClosed =
497− (return . Left $ TransportError SendClosed "Connection lost"))
522+ return . Left $ TransportError SendClosed "Connection lost"
499− sendFailed :: IOException -> TransportError SendErrorCode
500 sendFailed = TransportError SendFailed . show
525 sendFailed = TransportError SendFailed . show
502-- | Force-close the endpoint
527-- | Force-close the endpoint
503apiCloseEndPoint :: TCPTransport -- ^ Transport
528apiCloseEndPoint :: TCPTransport -- ^ Transport
504− -> [Event] -- ^ Events to output on the endpoint to indicate closure
529+ -> [Event] -- ^ Events used to report closure
505 -> LocalEndPoint -- ^ Local endpoint
530 -> LocalEndPoint -- ^ Local endpoint
507apiCloseEndPoint transport evs ourEndPoint = do
532apiCloseEndPoint transport evs ourEndPoint = do
@@ -523,7 +548,8 @@ apiCloseEndPoint transport evs ourEndPoint = do
523 -- Close the remote socket and return the set of all incoming connections
548 -- Close the remote socket and return the set of all incoming connections
524 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
549 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
525 tryCloseRemoteSocket theirEndPoint =
550 tryCloseRemoteSocket theirEndPoint =
526− -- We make an attempt to close the connection nicely (by sending a CloseSocket first)
551+ -- We make an attempt to close the connection nicely
552+ -- (by sending a CloseSocket first)
527 modifyMVar_ (remoteState theirEndPoint) $ \st ->
553 modifyMVar_ (remoteState theirEndPoint) $ \st ->
528 case st of
554 case st of
529 RemoteEndPointInvalid _ ->
555 RemoteEndPointInvalid _ ->
@@ -532,17 +558,17 @@ apiCloseEndPoint transport evs ourEndPoint = do
532 -- Try to send a CloseSocket request
558 -- Try to send a CloseSocket request
533 tryIO $ sendOn conn [encodeInt32 CloseSocket]
559 tryIO $ sendOn conn [encodeInt32 CloseSocket]
534 -- .. but even if it fails, close the socket anyway
560 -- .. but even if it fails, close the socket anyway
535− -- (hence, two separate calls to tryIO)
536− tryIO $ N.sClose (remoteSocket conn)
561+ tryCloseSocket (remoteSocket conn)
537 return RemoteEndPointClosed
562 return RemoteEndPointClosed
538 RemoteEndPointClosing _ conn -> do
563 RemoteEndPointClosing _ conn -> do
539− tryIO $ N.sClose (remoteSocket conn)
564+ tryCloseSocket (remoteSocket conn)
540 return RemoteEndPointClosed
565 return RemoteEndPointClosed
541 RemoteEndPointClosed ->
566 RemoteEndPointClosed ->
542 return RemoteEndPointClosed
567 return RemoteEndPointClosed
544-- | Special case of 'apiConnect': connect an endpoint to itself
569-- | Special case of 'apiConnect': connect an endpoint to itself
570+connectToSelf :: LocalEndPoint
545−connectToSelf :: LocalEndPoint -> IO (Either (TransportError ConnectErrorCode) Connection)
571+ -> IO (Either (TransportError ConnectErrorCode) Connection)
546connectToSelf ourEndPoint = do
572connectToSelf ourEndPoint = do
547 -- Here connAlive must an MVar because it is not protected by another lock
573 -- Here connAlive must an MVar because it is not protected by another lock
548 connAlive <- newMVar True
574 connAlive <- newMVar True
@@ -551,15 +577,20 @@ connectToSelf ourEndPoint = do
551 Left err ->
577 Left err ->
552 return . Left $ TransportError ConnectNotFound (show err)
578 return . Left $ TransportError ConnectNotFound (show err)
553 Right connId -> do
579 Right connId -> do
554− writeChan ourChan (ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint))
581+ ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint)
582+ return . Right $ Connection
555− return . Right $ Connection { send = selfSend connAlive connId
583+ { send = selfSend connAlive connId
556− , close = selfClose connAlive connId
584+ , close = selfClose connAlive connId
559 ourChan :: Chan Event
587 ourChan :: Chan Event
560 ourChan = localChannel ourEndPoint
588 ourChan = localChannel ourEndPoint
590+ selfSend :: MVar Bool
562− selfSend :: MVar Bool -> ConnectionId -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
593+ -> IO (Either (TransportError SendErrorCode) ())
563 selfSend connAlive connId msg =
594 selfSend connAlive connId msg =
564 modifyMVar connAlive $ \alive ->
595 modifyMVar connAlive $ \alive ->
@@ -605,8 +636,8 @@ internalSocketBetween transport ourAddress theirAddress = runErrorT $ do
605 withMVar (remoteState theirEndPoint) $ \st -> case st of
636 withMVar (remoteState theirEndPoint) $ \st -> case st of
606 RemoteEndPointValid vst ->
637 RemoteEndPointValid vst ->
607 return . Right $ remoteSocket vst
638 return . Right $ remoteSocket vst
608− RemoteEndPointClosing _ _ ->
639+ RemoteEndPointClosing _ vst ->
609− return . Left $ "Remote endpoint closing"
640+ return . Right $ remoteSocket vst
610 RemoteEndPointClosed ->
641 RemoteEndPointClosed ->
611 return . Left $ "Remote endpoint closed"
642 return . Left $ "Remote endpoint closed"
612 RemoteEndPointInvalid _ ->
643 RemoteEndPointInvalid _ ->
@@ -618,7 +649,8 @@ internalSocketBetween transport ourAddress theirAddress = runErrorT $ do
619-- | Create a new local endpoint
650-- | Create a new local endpoint
621−-- May throw a TransportError NewEndPointErrorCode exception if the transport is closed.
652+-- May throw a TransportError NewEndPointErrorCode exception if the transport
622createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
654createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
623createLocalEndPoint transport = do
655createLocalEndPoint transport = do
624 chan <- newChan
656 chan <- newChan
@@ -633,12 +665,19 @@ createLocalEndPoint transport = do
633 modifyMVar (transportState transport) $ \st -> case st of
665 modifyMVar (transportState transport) $ \st -> case st of
634 TransportValid vst -> do
666 TransportValid vst -> do
635 let ix = vst ^. nextEndPointId
667 let ix = vst ^. nextEndPointId
636− let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix
668+ let addr = encodeEndPointAddress (transportHost transport)
669+ (transportPort transport)
637 let localEndPoint = LocalEndPoint { localAddress = addr
671 let localEndPoint = LocalEndPoint { localAddress = addr
638 , localChannel = chan
672 , localChannel = chan
639 , localState = state
673 , localState = state
675+ return ( TransportValid
641− return (TransportValid . (localEndPointAt addr ^= Just localEndPoint) . (nextEndPointId ^= ix + 1) $ vst, localEndPoint)
676+ . (localEndPointAt addr ^= Just localEndPoint)
677+ . (nextEndPointId ^= ix + 1)
642 TransportClosed ->
681 TransportClosed ->
643 throwIO (TransportError NewEndPointFailed "Transport closed")
682 throwIO (TransportError NewEndPointFailed "Transport closed")
@@ -656,13 +695,18 @@ requestConnectionTo ourEndPoint theirAddress = go
658 -- Find the remote endpoint (create it if it doesn't yet exist)
697 -- Find the remote endpoint (create it if it doesn't yet exist)
659− theirEndPoint <- findTheirEndPoint
698+ (theirEndPoint, isNew) <- mapIOException connectFailed $
699+ findRemoteEndPoint ourEndPoint theirAddress
701+ -- If it was new, start listening for incoming messages, too
702+ when isNew . void . forkEndPointThread ourEndPoint $
703+ setupRemoteEndPoint (ourEndPoint, theirEndPoint)
660 let theirState = remoteState theirEndPoint
705 let theirState = remoteState theirEndPoint
662− -- Before we initiate the new connection request we want to make sure that
707+ -- Before we initiate the new connection request we want to make sure
663− -- refcount on the endpoint is incremented so that a concurrent thread will
708+ -- that refcount on the endpoint is incremented so that a concurrent
664− -- not close the connection. Note that if IF we return RemoteEndPointValid
709+ -- thread will not close the connection.
665− -- here then we can rely on the endpoint remaining in that state.
666 endPointStateSnapshot <- modifyMVar theirState $ \st ->
710 endPointStateSnapshot <- modifyMVar theirState $ \st ->
667 case st of
711 case st of
668 RemoteEndPointValid ep ->
712 RemoteEndPointValid ep ->
@@ -670,10 +714,10 @@ requestConnectionTo ourEndPoint theirAddress = go
671 return (st, st)
715 return (st, st)
673− -- From this point on we are guaranteed the refcount is positive, provided
717+ -- From this point on we are guaranteed the refcount is positive,
674− -- that the endpoint was valid. We still need to deal with the case where
718+ -- provided that the endpoint was valid. We still need to deal with the
675− -- it was not valid, however, which we didn't want to do while holding the
719+ -- case where it was not valid, however, which we didn't want to do while
676− -- endpoint lock.
720+ -- holding the endpoint lock.
678 -- Although 'endPointStateSnapshot' here refers to a snapshot of the
722 -- Although 'endPointStateSnapshot' here refers to a snapshot of the
679 -- endpoint state, and might have changed in the meantime, these changes
723 -- endpoint state, and might have changed in the meantime, these changes
@@ -701,47 +745,45 @@ requestConnectionTo ourEndPoint theirAddress = go
701 -- point). Note that doRemoteRequest may throw an error if the send
745 -- point). Note that doRemoteRequest may throw an error if the send
702 -- fails, and if it does, it will have put the remote endpoint in
746 -- fails, and if it does, it will have put the remote endpoint in
703 -- closed state.
747 -- closed state.
748+ reply <- mapIOException connectFailed $
704− reply <- mapExceptionIO connectFailed $ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
749+ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
705 case decodeInt32 . BS.concat $ reply of
750 case decodeInt32 . BS.concat $ reply of
706 Nothing ->
751 Nothing ->
707 throwIO (connectFailed $ userError "Invalid integer")
752 throwIO (connectFailed $ userError "Invalid integer")
708 Just cid ->
753 Just cid ->
709 return (theirEndPoint, cid)
754 return (theirEndPoint, cid)
711− -- If this is a new endpoint, fork a thread to listen for incoming
712− -- connections. We don't want to do this while we hold the lock, because
713− -- forkEndPointThread modifies the local state too (to record the thread
715− findTheirEndPoint :: IO RemoteEndPoint
716− findTheirEndPoint = do
717− (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
718− LocalEndPointClosed ->
719− throwIO (TransportError ConnectFailed "Local endpoint closed")
720− LocalEndPointValid vst ->
721− case vst ^. localConnectionTo theirAddress of
722− Just theirEndPoint ->
723− return (st, (theirEndPoint, False))
725− theirState <- newEmptyMVar
726− let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
727− , remoteState = theirState
728− , remoteId = vst ^. nextRemoteId
730− return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) . (nextRemoteId ^: (+ 1)) $ vst, (theirEndPoint, True))
732− -- The only way for forkEndPointThread to fail is if the local endpoint
733− -- gets closed. This error will be caught elsewhere, so we ignore it
735− when isNew . void . forkEndPointThread ourEndPoint $
736− setupRemoteEndPoint (ourEndPoint, theirEndPoint)
737− return theirEndPoint
739− ourState :: MVar LocalEndPointState
740− ourState = localState ourEndPoint
742− connectFailed :: IOException -> TransportError ConnectErrorCode
743 connectFailed = TransportError ConnectFailed . show
756 connectFailed = TransportError ConnectFailed . show
758+-- | Find a remote endpoint. Create an uninitialized remote endpoint if the
759+-- remote endpoint did not yet exist.
761+-- Throws an IOException if the local endpoint is closed.
762+findRemoteEndPoint :: LocalEndPoint -- ^ Our endpoint
763+ -> EndPointAddress -- ^ Their address
764+ -> IO (RemoteEndPoint, Bool) -- ^ Remote endpoint, new?
765+findRemoteEndPoint ourEndPoint theirAddress =
766+ modifyMVar (localState ourEndPoint) $ \st -> case st of
767+ LocalEndPointClosed ->
768+ throwIO $ userError "Local endpoint closed"
769+ LocalEndPointValid vst ->
770+ case vst ^. localConnectionTo theirAddress of
771+ Just theirEndPoint ->
772+ return (st, (theirEndPoint, False))
774+ theirState <- newEmptyMVar
775+ let theirEndPoint = RemoteEndPoint
776+ { remoteAddress = theirAddress
777+ , remoteState = theirState
778+ , remoteId = vst ^. nextRemoteId
780+ return ( LocalEndPointValid
781+ . (localConnectionTo theirAddress ^= Just theirEndPoint)
782+ . (nextRemoteId ^: (+ 1))
784+ , (theirEndPoint, True)
745-- | Set up a remote endpoint
787-- | Set up a remote endpoint
747-- RELY: The state of the remote endpoint must be uninitialized.
789-- RELY: The state of the remote endpoint must be uninitialized.
@@ -749,7 +791,7 @@ requestConnectionTo ourEndPoint theirAddress = go
749-- RemoteEndPointInvalid.
791-- RemoteEndPointInvalid.
750setupRemoteEndPoint :: EndPointPair -> IO ()
792setupRemoteEndPoint :: EndPointPair -> IO ()
751setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
793setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
752− didAccept <- bracketOnError (socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint))
794+ didAccept <- bracketOnError (socketToEndPoint ourAddress theirAddress)
753 onError $ \result ->
795 onError $ \result ->
754 case result of
796 case result of
755 Right (sock, ConnectionRequestAccepted) -> do
797 Right (sock, ConnectionRequestAccepted) -> do
@@ -757,7 +799,8 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
757 { remoteSocket = sock
799 { remoteSocket = sock
758 , _remoteOutgoing = 0
800 , _remoteOutgoing = 0
759 , _remoteIncoming = IntSet.empty
801 , _remoteIncoming = IntSet.empty
760− , sendOn = \msg -> sendMany sock msg `onException` tryIO (N.sClose sock)
802+ , sendOn = (`onException` tryCloseSocket sock)
762 putMVar theirState (RemoteEndPointValid vst)
805 putMVar theirState (RemoteEndPointValid vst)
763 return True
806 return True
@@ -768,11 +811,12 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
768 -- RemoteEndPointInvalid; subsequent threads will initiate a new
811 -- RemoteEndPointInvalid; subsequent threads will initiate a new
769 -- connection requests.
812 -- connection requests.
770 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
813 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
771− putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
814+ let err = invalidAddress "Invalid endpoint"
772− N.sClose sock
815+ putMVar theirState (RemoteEndPointInvalid err)
773 return False
817 return False
774 Right (sock, ConnectionRequestCrossed) -> do
818 Right (sock, ConnectionRequestCrossed) -> do
775− N.sClose sock
819+ tryCloseSocket sock
776 return False
820 return False
777 Left err -> do
821 Left err -> do
778 removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above
822 removeRemoteEndPoint (ourEndPoint, theirEndPoint) -- See comment above
@@ -780,10 +824,11 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
780 return False
824 return False
782 -- If we get to this point without an exception, then
826 -- If we get to this point without an exception, then
783− -- * if didAccept is False the socket has already been closed
828+ -- if didAccept is False the socket has already been closed
784− -- * if didAccept is True, the socket has been stored as part of the remote
829+ -- if didAccept is True, the socket has been stored as part of the remote
785− -- state so we no longer need to worry about closing it when an
830+ -- state so we no longer need to worry about closing it when an
786− -- asynchronous exception occurs
831+ -- asynchronous exception occurs
787 when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
832 when didAccept $ handleIncomingMessages (ourEndPoint, theirEndPoint)
789 -- If an asynchronous exception occurs while we set up the remote endpoint
834 -- If an asynchronous exception occurs while we set up the remote endpoint
@@ -792,34 +837,42 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
792 -- that are blocked on reading the remote state are unblocked. It is
837 -- that are blocked on reading the remote state are unblocked. It is
793 -- possible, however, that the exception occurred after we already
838 -- possible, however, that the exception occurred after we already
794 -- initialized the remote state, which is why we use tryPutMVar here.
839 -- initialized the remote state, which is why we use tryPutMVar here.
795− onError :: Either (TransportError ConnectErrorCode) (N.Socket, ConnectionRequestResponse) -> IO ()
840+ onError :: Either (TransportError ConnectErrorCode)
841+ (N.Socket, ConnectionRequestResponse)
796 onError result = do
843 onError result = do
797 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
844 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
798 case result of
845 case result of
799 Left err -> do
846 Left err -> do
800− tryPutMVar theirState (RemoteEndPointInvalid (TransportError ConnectFailed (show err)))
847+ tryPutMVar theirState (RemoteEndPointInvalid err)
801 return ()
848 return ()
802 Right (sock, _) -> do
849 Right (sock, _) -> do
850+ let err = failed "setupRemoteEndPoint failed"
803− tryPutMVar theirState (RemoteEndPointInvalid (TransportError ConnectFailed "setupRemoteEndPoint failed"))
851+ tryPutMVar theirState (RemoteEndPointInvalid err)
804− tryIO $ N.sClose sock
852+ tryCloseSocket sock
805 return ()
853 return ()
855+ failed = TransportError ConnectFailed
856+ ourAddress = localAddress ourEndPoint
857+ theirAddress = remoteAddress theirEndPoint
807 theirState = remoteState theirEndPoint
858 theirState = remoteState theirEndPoint
808 invalidAddress = TransportError ConnectNotFound
859 invalidAddress = TransportError ConnectNotFound
810-- | Establish a connection to a remote endpoint
861-- | Establish a connection to a remote endpoint
811socketToEndPoint :: EndPointAddress -- ^ Our address
862socketToEndPoint :: EndPointAddress -- ^ Our address
812 -> EndPointAddress -- ^ Their address
863 -> EndPointAddress -- ^ Their address
813− -> IO (Either (TransportError ConnectErrorCode) (N.Socket, ConnectionRequestResponse))
864+ -> IO (Either (TransportError ConnectErrorCode)
865+ (N.Socket, ConnectionRequestResponse))
814socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
866socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
815 (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
867 (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
816 Nothing -> throwIO (failed . userError $ "Could not parse")
868 Nothing -> throwIO (failed . userError $ "Could not parse")
817 Just dec -> return dec
869 Just dec -> return dec
870+ addr:_ <- mapIOException invalidAddress $
818− addr:_ <- mapExceptionIO invalidAddress $ N.getAddrInfo Nothing (Just host) (Just port)
871+ N.getAddrInfo Nothing (Just host) (Just port)
819− bracketOnError (createSocket addr) N.sClose $ \sock -> do
872+ bracketOnError (createSocket addr) tryCloseSocket $ \sock -> do
820− mapExceptionIO failed $ N.setSocketOption sock N.ReuseAddr 1
873+ mapIOException failed $ N.setSocketOption sock N.ReuseAddr 1
821− mapExceptionIO invalidAddress $ N.connect sock (N.addrAddress addr)
874+ mapIOException invalidAddress $ N.connect sock (N.addrAddress addr)
822− response <- mapExceptionIO failed $ do
875+ response <- mapIOException failed $ do
823 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
876 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
824 recvInt32 sock
877 recvInt32 sock
825 case tryToEnum response of
878 case tryToEnum response of
@@ -827,10 +880,9 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
827 Just r -> return (sock, r)
880 Just r -> return (sock, r)
829 createSocket :: N.AddrInfo -> IO N.Socket
882 createSocket :: N.AddrInfo -> IO N.Socket
830− createSocket addr = mapExceptionIO insufficientResources $
883+ createSocket addr = mapIOException insufficientResources $
831 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
884 N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
833− invalidAddress, insufficientResources, failed :: IOException -> TransportError ConnectErrorCode
834 invalidAddress = TransportError ConnectNotFound . show
886 invalidAddress = TransportError ConnectNotFound . show
835 insufficientResources = TransportError ConnectInsufficientResources . show
887 insufficientResources = TransportError ConnectInsufficientResources . show
836 failed = TransportError ConnectFailed . show
888 failed = TransportError ConnectFailed . show
@@ -847,7 +899,11 @@ removeRemoteEndPoint (ourEndPoint, theirEndPoint) =
847 return st
899 return st
848 Just remoteEndPoint' ->
900 Just remoteEndPoint' ->
849 if remoteId remoteEndPoint' == remoteId theirEndPoint
901 if remoteId remoteEndPoint' == remoteId theirEndPoint
903+ ( LocalEndPointValid
850− then return (LocalEndPointValid . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) $ vst)
904+ . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing)
851 else return st
907 else return st
852 LocalEndPointClosed ->
908 LocalEndPointClosed ->
853 return LocalEndPointClosed
909 return LocalEndPointClosed
@@ -862,18 +918,26 @@ removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
862removeLocalEndPoint transport ourEndPoint =
918removeLocalEndPoint transport ourEndPoint =
863 modifyMVar_ (transportState transport) $ \st -> case st of
919 modifyMVar_ (transportState transport) $ \st -> case st of
864 TransportValid vst ->
920 TransportValid vst ->
921+ return ( TransportValid
865− return (TransportValid . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) $ vst)
922+ . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
866 TransportClosed ->
925 TransportClosed ->
867 return TransportClosed
926 return TransportClosed
869-- | Encode end point address
928-- | Encode end point address
870−encodeEndPointAddress :: N.HostName -> N.ServiceName -> EndPointId -> EndPointAddress
929+encodeEndPointAddress :: N.HostName
871encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
933encodeEndPointAddress host port ix = EndPointAddress . BSC.pack $
872 host ++ ":" ++ port ++ ":" ++ show ix
934 host ++ ":" ++ port ++ ":" ++ show ix
874-- | Decode end point address
936-- | Decode end point address
875−decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointId)
937+decodeEndPointAddress :: EndPointAddress
938+ -> Maybe (N.HostName, N.ServiceName, EndPointId)
876−decodeEndPointAddress (EndPointAddress bs) = case map BSC.unpack $ BSC.split ':' bs of
939+decodeEndPointAddress (EndPointAddress bs) =
940+ case map BSC.unpack $ BSC.split ':' bs of
877 [host, port, endPointIdStr] ->
941 [host, port, endPointIdStr] ->
878 case reads endPointIdStr of
942 case reads endPointIdStr of
879 [(endPointId, "")] -> Just (host, port, endPointId)
943 [(endPointId, "")] -> Just (host, port, endPointId)
@@ -894,38 +958,35 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do
894 reqId <- modifyMVar (localState ourEndPoint) $ \st -> case st of
958 reqId <- modifyMVar (localState ourEndPoint) $ \st -> case st of
895 LocalEndPointValid vst -> do
959 LocalEndPointValid vst -> do
896 let reqId = vst ^. nextCtrlRequestId
960 let reqId = vst ^. nextCtrlRequestId
961+ return ( LocalEndPointValid
962+ . (nextCtrlRequestId ^: (+ 1))
897− return (LocalEndPointValid . (nextCtrlRequestId ^: (+ 1)) . (pendingCtrlRequestsAt reqId ^= Just reply) $ vst, reqId)
963+ . (pendingCtrlRequestsAt reqId ^= Just reply)
898 LocalEndPointClosed ->
967 LocalEndPointClosed ->
899 throwIO (userError "Local endpoint closed")
968 throwIO (userError "Local endpoint closed")
900− withRemoteState (ourEndPoint, theirEndPoint)
969+ withRemoteState (ourEndPoint, theirEndPoint) RemoteStatePatternMatch
901− -- RemoteEndPointInvalid
902− (\_ -> fail "doRemoteRequest RELY violation")
970+ { caseInvalid = \_ -> fail "doRemoteRequest RELY violation"
903− -- RemoteEndPointValid
904− (\vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId])
971+ , caseValid = \vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId]
905− -- RemoteEndPointClosing
906− (\_ -> fail "doRemoteRequest RELY violation")
972+ , caseClosing = \_ -> fail "doRemoteRequest RELY violation"
907− -- RemoteEndPointClosed
908− (throwIO (userError "Remote endpoint closed"))
973+ , caseClosed = throwIO (userError "Remote endpoint closed")
909 takeMVar reply
975 takeMVar reply
911−-- | Check if the remote endpoint is unused, and if so, send a CloseSocket request
977+-- | Send a CloseSocket request if the remote endpoint is unused
912closeIfUnused :: EndPointPair -> IO ()
978closeIfUnused :: EndPointPair -> IO ()
913closeIfUnused (ourEndPoint, theirEndPoint) =
979closeIfUnused (ourEndPoint, theirEndPoint) =
914− modifyRemoteState_ (ourEndPoint, theirEndPoint)
980+ modifyRemoteState_ (ourEndPoint, theirEndPoint) remoteStateIdentity
915− -- RemoteEndPointInvalid
981+ { caseValid = \vst ->
916− (return . RemoteEndPointInvalid)
917− -- RemoteEndPointValid
918− (\vst -> if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
982+ if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
919− then do
983+ then do
920− sendOn vst [encodeInt32 CloseSocket]
984+ sendOn vst [encodeInt32 CloseSocket]
921− resolved <- newEmptyMVar
985+ resolved <- newEmptyMVar
922− return $ RemoteEndPointClosing resolved vst
986+ return $ RemoteEndPointClosing resolved vst
924− return $ RemoteEndPointValid vst)
988+ return $ RemoteEndPointValid vst
925− -- RemoteEndPointClosing
989+ }
926− (return . uncurry RemoteEndPointClosing)
927− -- RemoteEndPointClosed
928− (return RemoteEndPointClosed)
930-- | Fork a new thread and store its ID as part of the transport state
991-- | Fork a new thread and store its ID as part of the transport state
@@ -944,7 +1005,10 @@ forkEndPointThread ourEndPoint p =
944 LocalEndPointValid vst -> do
1005 LocalEndPointValid vst -> do
945 threadRegistered <- newEmptyMVar
1006 threadRegistered <- newEmptyMVar
946 tid <- forkIO (takeMVar threadRegistered >> p >> removeThread)
1007 tid <- forkIO (takeMVar threadRegistered >> p >> removeThread)
947− putMVar ourState $ LocalEndPointValid . (internalThreads ^: (tid :)) $ vst
1008+ putMVar ourState ( LocalEndPointValid
1009+ . (internalThreads ^: (tid :))
948 putMVar threadRegistered ()
1012 putMVar threadRegistered ()
949 return True
1013 return True
950 LocalEndPointClosed -> do
1014 LocalEndPointClosed -> do
@@ -956,27 +1020,49 @@ forkEndPointThread ourEndPoint p =
956 tid <- myThreadId
1020 tid <- myThreadId
957 modifyMVar_ ourState $ \st -> case st of
1021 modifyMVar_ ourState $ \st -> case st of
958 LocalEndPointValid vst ->
1022 LocalEndPointValid vst ->
959− return (LocalEndPointValid . (internalThreads ^: filter (/= tid)) $ vst)
1023+ return ( LocalEndPointValid
1024+ . (internalThreads ^: filter (/= tid))
960 LocalEndPointClosed ->
1027 LocalEndPointClosed ->
961 return LocalEndPointClosed
1028 return LocalEndPointClosed
963 ourState :: MVar LocalEndPointState
1030 ourState :: MVar LocalEndPointState
964 ourState = localState ourEndPoint
1031 ourState = localState ourEndPoint
1033+--------------------------------------------------------------------------------
1034+-- As soon as a remote connection fails, we want to put notify our endpoint --
1035+-- and put it into a closed state. Since this may happen in many places, we --
1036+-- provide some abstractions. --
1037+--------------------------------------------------------------------------------
1039+data RemoteStatePatternMatch a = RemoteStatePatternMatch
1040+ { caseInvalid :: TransportError ConnectErrorCode -> IO a
1041+ , caseValid :: ValidRemoteEndPointState -> IO a
1042+ , caseClosing :: MVar () -> ValidRemoteEndPointState -> IO a
1043+ , caseClosed :: IO a
1046+remoteStateIdentity :: RemoteStatePatternMatch RemoteState
1047+remoteStateIdentity =
1048+ RemoteStatePatternMatch
1049+ { caseInvalid = return . RemoteEndPointInvalid
1050+ , caseValid = return . RemoteEndPointValid
1051+ , caseClosing = (return .) . RemoteEndPointClosing
1052+ , caseClosed = return RemoteEndPointClosed
966-- | Like modifyMVar, but if an exception occurs don't restore the remote
1055-- | Like modifyMVar, but if an exception occurs don't restore the remote
967-- endpoint to its original value but close it instead
1056-- endpoint to its original value but close it instead
968−modifyRemoteState :: EndPointPair -- ^ Local and remote endpoint
1057+modifyRemoteState :: EndPointPair
969− -> (TransportError ConnectErrorCode -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointInvalid
1058+ -> RemoteStatePatternMatch (RemoteState, a)
970− -> (ValidRemoteEndPointState -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointValid
971− -> ((MVar (), ValidRemoteEndPointState) -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointClosing
972− -> IO (RemoteState, a) -- ^ Case for RemoteEndPointClosed
974−modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =
1060+modifyRemoteState (ourEndPoint, theirEndPoint) match =
975 mask $ \restore -> do
1061 mask $ \restore -> do
976 st <- takeMVar theirState
1062 st <- takeMVar theirState
977 case st of
1063 case st of
978 RemoteEndPointValid vst -> do
1064 RemoteEndPointValid vst -> do
979− mResult <- try $ restore (caseValid vst)
1065+ mResult <- try $ restore (caseValid match vst)
980 case mResult of
1066 case mResult of
981 Right (st', a) -> do
1067 Right (st', a) -> do
982 putMVar theirState st'
1068 putMVar theirState st'
@@ -986,15 +1072,18 @@ modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing
986 -- The other cases are less interesting, because unless the endpoint is
1072 -- The other cases are less interesting, because unless the endpoint is
987 -- in Valid state we're not supposed to do any IO on it
1073 -- in Valid state we're not supposed to do any IO on it
988 RemoteEndPointClosing resolved vst -> do
1074 RemoteEndPointClosing resolved vst -> do
989− (st', a) <- restore (caseClosing (resolved, vst)) `onException` putMVar theirState st
1075+ (st', a) <- onException (restore $ caseClosing match resolved vst)
1076+ (putMVar theirState st)
990 putMVar theirState st'
1077 putMVar theirState st'
991 return a
1078 return a
992 RemoteEndPointInvalid err -> do
1079 RemoteEndPointInvalid err -> do
1080+ (st', a) <- onException (restore $ caseInvalid match err)
993− (st', a) <- restore (caseInvalid err) `onException` putMVar theirState st
1081+ (putMVar theirState st)
994 putMVar theirState st'
1082 putMVar theirState st'
995 return a
1083 return a
996 RemoteEndPointClosed -> do
1084 RemoteEndPointClosed -> do
1085+ (st', a) <- onException (restore $ caseClosed match)
997− (st', a) <- restore caseClosed `onException` putMVar theirState st
1086+ (putMVar theirState st)
998 putMVar theirState st'
1087 putMVar theirState st'
999 return a
1088 return a
@@ -1003,6 +1092,7 @@ modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing
1004 handleException :: SomeException -> ValidRemoteEndPointState -> IO a
1093 handleException :: SomeException -> ValidRemoteEndPointState -> IO a
1005 handleException ex vst = do
1094 handleException ex vst = do
1095+ tryCloseSocket (remoteSocket vst)
1006 -- We need to remove the remote endpoint from the local endpoint before
1096 -- We need to remove the remote endpoint from the local endpoint before
1007 -- putting it in Closed state (by INV-CLOSE), but we don't want to hold
1097 -- putting it in Closed state (by INV-CLOSE), but we don't want to hold
1008 -- two locks at the same time. So we put it in closing state first ..
1098 -- two locks at the same time. So we put it in closing state first ..
@@ -1013,7 +1103,8 @@ modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing
1013 -- .. close it and notify the local endpoint we lost the connection ..
1103 -- .. close it and notify the local endpoint we lost the connection ..
1014 mIncoming <- closeRemoteEndPoint
1104 mIncoming <- closeRemoteEndPoint
1015 forM_ mIncoming $ \incoming -> do
1105 forM_ mIncoming $ \incoming -> do
1016− let err = TransportError (EventConnectionLost (remoteAddress theirEndPoint) incoming) (show ex)
1106+ let code = EventConnectionLost (remoteAddress theirEndPoint) incoming
1107+ err = TransportError code (show ex)
1017 writeChan (localChannel ourEndPoint) $ ErrorEvent err
1108 writeChan (localChannel ourEndPoint) $ ErrorEvent err
1018 -- .. and finally rethrow the exception
1109 -- .. and finally rethrow the exception
1019 throwIO ex
1110 throwIO ex
@@ -1035,35 +1126,41 @@ modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing
1035 fail "the impossible happened"
1126 fail "the impossible happened"
1037-- | Like 'modifyRemoteState' but without a return value
1128-- | Like 'modifyRemoteState' but without a return value
1038−modifyRemoteState_ :: EndPointPair -- ^ Local and remote endpoint
1129+modifyRemoteState_ :: EndPointPair
1039− -> (TransportError ConnectErrorCode -> IO RemoteState) -- ^ Case for RemoteEndPointInvalid
1130+ -> RemoteStatePatternMatch RemoteState
1040− -> (ValidRemoteEndPointState -> IO RemoteState) -- ^ Case for RemoteEndPointValid
1041− -> ((MVar (), ValidRemoteEndPointState) -> IO RemoteState) -- ^ Case for RemoteEndPointClosing
1042− -> IO RemoteState -- ^ Case for RemoteEndPointClosed
1043 -> IO ()
1131 -> IO ()
1044−modifyRemoteState_ (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =
1132+modifyRemoteState_ (ourEndPoint, theirEndPoint) match =
1045 modifyRemoteState (ourEndPoint, theirEndPoint)
1133 modifyRemoteState (ourEndPoint, theirEndPoint)
1134+ RemoteStatePatternMatch
1046− (u . caseInvalid)
1135+ { caseInvalid = u . caseInvalid match
1047− (u . caseValid)
1136+ , caseValid = u . caseValid match
1048− (u . caseClosing)
1137+ , caseClosing = \resolved vst -> u $ caseClosing match resolved vst
1049− (u caseClosed)
1138+ , caseClosed = u $ caseClosed match
1051 u :: IO a -> IO (a, ())
1141 u :: IO a -> IO (a, ())
1052 u p = p >>= \a -> return (a, ())
1142 u p = p >>= \a -> return (a, ())
1054-- | Like 'modifyRemoteState' but without the ability to change the state
1144-- | Like 'modifyRemoteState' but without the ability to change the state
1055−withRemoteState :: EndPointPair -- ^ Local and remote endpoint
1145+withRemoteState :: EndPointPair
1056− -> (TransportError ConnectErrorCode -> IO a) -- ^ Case for RemoteEndPointInvalid
1146+ -> RemoteStatePatternMatch a
1057− -> (ValidRemoteEndPointState -> IO a) -- ^ Case for RemoteEndPointValid
1058− -> ((MVar (), ValidRemoteEndPointState) -> IO a) -- ^ Case for RemoteEndPointClosing
1059− -> IO a -- ^ Case for RemoteEndPointClosed
1060 -> IO a
1147 -> IO a
1061−withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =
1148+withRemoteState (ourEndPoint, theirEndPoint) match =
1062 modifyRemoteState (ourEndPoint, theirEndPoint)
1149 modifyRemoteState (ourEndPoint, theirEndPoint)
1150+ RemoteStatePatternMatch
1151+ { caseInvalid = \err -> do
1152+ a <- caseInvalid match err
1063− (\err -> (,) (RemoteEndPointInvalid err) <$> caseInvalid err)
1153+ return (RemoteEndPointInvalid err, a)
1154+ , caseValid = \vst -> do
1155+ a <- caseValid match vst
1064− (\vst -> (,) (RemoteEndPointValid vst) <$> caseValid vst)
1156+ return (RemoteEndPointValid vst, a)
1157+ , caseClosing = \resolved vst -> do
1158+ a <- caseClosing match resolved vst
1065− (\(resolved, vst) -> (,) (RemoteEndPointClosing resolved vst) <$> caseClosing (resolved, vst))
1159+ return (RemoteEndPointClosing resolved vst, a)
1161+ a <- caseClosed match
1066− ((,) RemoteEndPointClosed <$> caseClosed)
1162+ return (RemoteEndPointClosed, a)
1068--------------------------------------------------------------------------------
1165--------------------------------------------------------------------------------
1069-- Incoming requests --
1166-- Incoming requests --
@@ -1081,43 +1178,33 @@ withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing c
1081-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from
1178-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from
1082-- handleConnectionRequest the transport will be shut down.)
1179-- handleConnectionRequest the transport will be shut down.)
1083handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
1180handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
1084−handleConnectionRequest transport sock = handle tryCloseSocket $ do
1181+handleConnectionRequest transport sock = handle handleException $ do
1085 ourEndPointId <- recvInt32 sock
1182 ourEndPointId <- recvInt32 sock
1086 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
1183 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
1087− let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId
1184+ let ourAddress = encodeEndPointAddress (transportHost transport)
1185+ (transportPort transport)
1088 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
1187 ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
1089 TransportValid vst ->
1188 TransportValid vst ->
1090 case vst ^. localEndPointAt ourAddress of
1189 case vst ^. localEndPointAt ourAddress of
1091 Nothing -> do
1190 Nothing -> do
1092 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
1191 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
1093− fail "Invalid endpoint"
1192+ throwIO $ userError "Invalid endpoint"
1094 Just ourEndPoint ->
1193 Just ourEndPoint ->
1095 return ourEndPoint
1194 return ourEndPoint
1096 TransportClosed ->
1195 TransportClosed ->
1097− fail "Transport closed"
1196+ throwIO $ userError "Transport closed"
1098 void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress
1197 void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress
1100 go :: LocalEndPoint -> EndPointAddress -> IO ()
1199 go :: LocalEndPoint -> EndPointAddress -> IO ()
1101 go ourEndPoint theirAddress = do
1200 go ourEndPoint theirAddress = do
1102− mEndPoint <- handle (\e -> invalidEndPoint e >> return Nothing) $ do
1201+ mEndPoint <- handle ((>> return Nothing) . invalidEndPoint) $ do
1103− (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st -> case st of
1202+ (theirEndPoint, isNew) <- findRemoteEndPoint ourEndPoint theirAddress
1104− LocalEndPointValid vst ->
1105− case vst ^. localConnectionTo theirAddress of
1107− theirState <- newEmptyMVar
1108− let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
1109− , remoteState = theirState
1110− , remoteId = vst ^. nextRemoteId
1112− return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) . (nextRemoteId ^: (+ 1)) $ vst, (False, theirEndPoint))
1113− Just theirEndPoint ->
1114− return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
1203+ let crossed = not isNew && localAddress ourEndPoint < theirAddress
1115− LocalEndPointClosed ->
1116− fail "Local endpoint closed"
1117 if crossed
1204 if crossed
1119 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
1206 tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
1120− tryIO $ N.sClose sock
1207+ tryCloseSocket sock
1121 return Nothing
1208 return Nothing
1123 let vst = ValidRemoteEndPointState
1210 let vst = ValidRemoteEndPointState
@@ -1140,13 +1227,13 @@ handleConnectionRequest transport sock = handle tryCloseSocket $ do
1140 -- exception from this point forward.
1227 -- exception from this point forward.
1141 forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
1228 forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
1143− tryCloseSocket :: IOException -> IO ()
1230+ handleException :: IOException -> IO ()
1144− tryCloseSocket _ = void . tryIO $ N.sClose sock
1231+ handleException _ = tryCloseSocket sock
1146 invalidEndPoint :: IOException -> IO ()
1233 invalidEndPoint :: IOException -> IO ()
1147− invalidEndPoint ex = do
1234+ invalidEndPoint _ = do
1148 tryIO $ sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
1235 tryIO $ sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
1149− tryCloseSocket ex
1236+ tryCloseSocket sock
1151-- | Handle requests from a remote endpoint.
1238-- | Handle requests from a remote endpoint.
@@ -1168,10 +1255,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1169 error "handleIncomingMessages RELY violation"
1256 error "handleIncomingMessages RELY violation"
1171− mCleanExit <- tryIO $ go sock
1173− Right () -> return ()
1174− Left err -> prematureExit sock err
1258+ tryIO (go sock) >>= either (prematureExit sock) return
1176 -- Dispatch
1260 -- Dispatch
1177 go :: N.Socket -> IO ()
1261 go :: N.Socket -> IO ()
@@ -1196,8 +1280,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1196 didClose <- closeSocket sock
1280 didClose <- closeSocket sock
1197 unless didClose $ go sock
1281 unless didClose $ go sock
1198 Nothing ->
1282 Nothing ->
1199− -- Invalid control request, exit
1200− hPutStrLn stderr "Warning: invalid control request"
1283+ throwIO $ userError "Warning: invalid control request"
1202 -- Create a new connection
1285 -- Create a new connection
1203 createNewConnection :: ControlRequestId -> IO ()
1286 createNewConnection :: ControlRequestId -> IO ()
@@ -1224,8 +1307,9 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1224 : encodeInt32 reqId
1307 : encodeInt32 reqId
1225 : prependLength [encodeInt32 newId]
1308 : prependLength [encodeInt32 newId]
1227− -- We add the new connection ID to the list of open connections only once the
1310+ -- We add the new connection ID to the list of open connections only
1228− -- endpoint has been notified of the new connection (sendOn may fail)
1311+ -- once the endpoint has been notified of the new connection (sendOn
1229 return (RemoteEndPointValid vst)
1313 return (RemoteEndPointValid vst)
1230 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)
1314 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)
@@ -1235,13 +1319,16 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1235 response <- recvWithLength sock
1319 response <- recvWithLength sock
1236 mmvar <- modifyMVar ourState $ \st -> case st of
1320 mmvar <- modifyMVar ourState $ \st -> case st of
1237 LocalEndPointValid vst ->
1321 LocalEndPointValid vst ->
1322+ return ( LocalEndPointValid
1238− return (LocalEndPointValid . (pendingCtrlRequestsAt reqId ^= Nothing) $ vst, vst ^. pendingCtrlRequestsAt reqId)
1323+ . (pendingCtrlRequestsAt reqId ^= Nothing)
1325+ , vst ^. pendingCtrlRequestsAt reqId
1239 LocalEndPointClosed ->
1327 LocalEndPointClosed ->
1240− fail "Local endpoint closed"
1328+ throwIO $ userError "Local endpoint closed"
1241 case mmvar of
1329 case mmvar of
1242− Nothing -> do
1330+ Nothing ->
1243− hPutStrLn stderr "Warning: Invalid request ID"
1331+ throwIO $ userError "Warning: Invalid request ID"
1244− return () -- Invalid request ID. TODO: We just ignore it?
1245 Just mvar ->
1332 Just mvar ->
1246 putMVar mvar response
1333 putMVar mvar response
@@ -1251,21 +1338,17 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1251 -- TODO: we should check that this connection is in fact open
1338 -- TODO: we should check that this connection is in fact open
1252 writeChan ourChannel (ConnectionClosed cid)
1339 writeChan ourChannel (ConnectionClosed cid)
1253 modifyMVar_ theirState $ \(RemoteEndPointValid vst) ->
1340 modifyMVar_ theirState $ \(RemoteEndPointValid vst) ->
1341+ return ( RemoteEndPointValid
1254− return (RemoteEndPointValid . (remoteIncoming ^: IntSet.delete cid) $ vst)
1342+ . (remoteIncoming ^: IntSet.delete cid)
1255 closeIfUnused (ourEndPoint, theirEndPoint)
1345 closeIfUnused (ourEndPoint, theirEndPoint)
1257 -- Close the socket (if we don't have any outgoing connections)
1347 -- Close the socket (if we don't have any outgoing connections)
1258 closeSocket :: N.Socket -> IO Bool
1348 closeSocket :: N.Socket -> IO Bool
1259 closeSocket sock = do
1349 closeSocket sock = do
1260− -- We need to check if we can close the socket (that is, if we don't have
1261− -- any outgoing connections), and once we are sure that we can put the
1262− -- endpoint in Closed state. However, by INV-CLOSE we can only put the
1263− -- endpoint in Closed state once we remove the endpoint from our local
1264− -- connections. But we can do that only once we are sure that we can
1265− -- close the endpoint. Catch-22. We can resolve the catch-22 by locking
1266− -- /both/ our local state and the endpoint state, but at the cost of
1267− -- introducing a double lock and all the associated perils, or by putting
1268− -- the remote endpoint in Closing state first. We opt for the latter.
1350+ -- As in modifyRemoteState we put the remote state in Closing state
1351+ -- before putting it in Closed state.
1269 canClose <- modifyMVar theirState $ \st ->
1352 canClose <- modifyMVar theirState $ \st ->
1270 case st of
1353 case st of
1271 RemoteEndPointValid vst -> do
1354 RemoteEndPointValid vst -> do
@@ -1295,13 +1378,14 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1295 Just resolved -> do
1378 Just resolved -> do
1296 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1379 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1297 modifyMVar_ theirState $ return . const RemoteEndPointClosed
1380 modifyMVar_ theirState $ return . const RemoteEndPointClosed
1298− N.sClose sock
1381+ tryCloseSocket sock
1299 putMVar resolved ()
1382 putMVar resolved ()
1300 return True
1383 return True
1302 -- Read a message and output it on the endPoint's channel
1385 -- Read a message and output it on the endPoint's channel
1303 readMessage :: N.Socket -> ConnectionId -> IO ()
1386 readMessage :: N.Socket -> ConnectionId -> IO ()
1387+ readMessage sock connId =
1304− readMessage sock connId = recvWithLength sock >>= writeChan ourChannel . Received connId
1388+ recvWithLength sock >>= writeChan ourChannel . Received connId
1306 -- Arguments
1390 -- Arguments
1307 ourChannel = localChannel ourEndPoint
1391 ourChannel = localChannel ourEndPoint
@@ -1312,15 +1396,17 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1312 -- Deal with a premature exit
1396 -- Deal with a premature exit
1313 prematureExit :: N.Socket -> IOException -> IO ()
1397 prematureExit :: N.Socket -> IOException -> IO ()
1314 prematureExit sock err = do
1398 prematureExit sock err = do
1315− N.sClose sock
1399+ tryCloseSocket sock
1316 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1400 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1317 modifyMVar_ theirState $ \st ->
1401 modifyMVar_ theirState $ \st ->
1318 case st of
1402 case st of
1319 RemoteEndPointInvalid _ ->
1403 RemoteEndPointInvalid _ ->
1320 error "handleIncomingMessages RELY violation"
1404 error "handleIncomingMessages RELY violation"
1321 RemoteEndPointValid vst -> do
1405 RemoteEndPointValid vst -> do
1322− writeChan ourChannel . ErrorEvent $
1406+ let code = EventConnectionLost
1407+ (remoteAddress theirEndPoint)
1323− TransportError (EventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems $ vst ^. remoteIncoming)) (show err)
1408+ (IntSet.elems $ vst ^. remoteIncoming)
1409+ writeChan ourChannel . ErrorEvent $ TransportError code (show err)
1324 return RemoteEndPointClosed
1410 return RemoteEndPointClosed
1325 RemoteEndPointClosing _ _ ->
1411 RemoteEndPointClosing _ _ ->
1326 return RemoteEndPointClosed
1412 return RemoteEndPointClosed
@@ -1335,9 +1421,12 @@ getNextConnectionId ourEndpoint =
1335 modifyMVar (localState ourEndpoint) $ \st -> case st of
1421 modifyMVar (localState ourEndpoint) $ \st -> case st of
1336 LocalEndPointValid vst -> do
1422 LocalEndPointValid vst -> do
1337 let connId = vst ^. nextConnectionId
1423 let connId = vst ^. nextConnectionId
1338− return (LocalEndPointValid . (nextConnectionId ^= connId + 1) $ vst, connId)
1424+ return ( LocalEndPointValid
1425+ . (nextConnectionId ^= connId + 1)
1339 LocalEndPointClosed ->
1428 LocalEndPointClosed ->
1340− fail "Local endpoint closed"
1429+ throwIO $ userError "Local endpoint closed"
1342--------------------------------------------------------------------------------
1431--------------------------------------------------------------------------------
1343-- Constants --
1432-- Constants --
@@ -1348,7 +1437,7 @@ firstNonReservedConnectionId :: ConnectionId
1348firstNonReservedConnectionId = 1024
1437firstNonReservedConnectionId = 1024
1350--------------------------------------------------------------------------------
1439--------------------------------------------------------------------------------
1351−-- Accessor definitions --
1440+-- Accessor definitions --
1352--------------------------------------------------------------------------------
1441--------------------------------------------------------------------------------
1354localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint)
1443localEndPoints :: Accessor ValidTransportState (Map EndPointAddress LocalEndPoint)
@@ -1375,17 +1464,18 @@ internalThreads = accessor _internalThreads (\ts st -> st { _internalThreads = t
1375nextRemoteId :: Accessor ValidLocalEndPointState Int
1464nextRemoteId :: Accessor ValidLocalEndPointState Int
1376nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })
1465nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })
1467+remoteOutgoing :: Accessor ValidRemoteEndPointState Int
1468+remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })
1470+remoteIncoming :: Accessor ValidRemoteEndPointState IntSet
1471+remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })
1378localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1473localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1379localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1474localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1381pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidLocalEndPointState (Maybe (MVar [ByteString]))
1476pendingCtrlRequestsAt :: ControlRequestId -> Accessor ValidLocalEndPointState (Maybe (MVar [ByteString]))
1382pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)
1477pendingCtrlRequestsAt ix = pendingCtrlRequests >>> DAC.intMapMaybe (fromIntegral ix)
1479+localConnectionTo :: EndPointAddress
1384−localConnectionTo :: EndPointAddress -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)
1480+ -> Accessor ValidLocalEndPointState (Maybe RemoteEndPoint)
1385localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
1481localConnectionTo addr = localConnections >>> DAC.mapMaybe addr
1387−remoteOutgoing :: Accessor ValidRemoteEndPointState Int
1388−remoteOutgoing = accessor _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })
1390−remoteIncoming :: Accessor ValidRemoteEndPointState IntSet
1391−remoteIncoming = accessor _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })