@@ -6,19 +6,28 @@
66-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
77-- @client1.local:http-alt@ then the transport layer will not realize that the
88-- TCP connection can be reused.
9+--
10+-- Applications that use the TCP transport should use
11+-- 'Network.Socket.withSocketsDo' in their main function for Windows
12+-- compatibility (see "Network.Socket").
13+--
14+-- TODOs:
15+-- * Output exception on channel after endpoint is closed
916module Network.Transport.TCP ( -- * Main API
1017 createTransport
11− , -- * TCP specific functionality
18+ , -- * TCP specific functionality (exported mostly for testing purposes_
1219 EndPointId
1320 , encodeEndPointAddress
1421 , decodeEndPointAddress
1522 , ControlHeader(..)
1623 , ConnectionRequestResponse(..)
1724 , firstNonReservedConnectionId
25+ , socketToEndPoint
1826 -- * Design notes
1927 -- $design
2028 ) where
2129
30+import Prelude hiding (catch)
2231import Network.Transport
2332import Network.Transport.Internal.TCP ( forkServer
2433 , recvWithLength
@@ -27,8 +36,7 @@ import Network.Transport.Internal.TCP ( forkServer
2736import Network.Transport.Internal ( encodeInt32
2837 , decodeInt32
2938 , prependLength
30− , failWith
31− , failWithIO
39+ , mapExceptionIO
3240 , tryIO
3341 , tryToEnum
3442 )
@@ -45,6 +53,8 @@ import qualified Network.Socket as N ( HostName
4553 , setSocketOption
4654 , SocketOption(ReuseAddr)
4755 , connect
56+ , sOMAXCONN
57+ , AddrInfo
4858 )
4959import Network.Socket.ByteString (sendMany)
5060import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
@@ -62,9 +72,7 @@ import Control.Concurrent.MVar ( MVar
6272import Control.Category ((>>>))
6373import Control.Applicative ((<*>), (*>), (<$>))
6474import Control.Monad (forM, forM_, void, when, unless)
65−import Control.Monad.Error (ErrorT(ErrorT), runErrorT, throwError)
66−import Control.Monad.IO.Class (liftIO)
67−import Control.Exception (IOException, SomeException)
75+import Control.Exception (IOException, SomeException, handle, throw, try, bracketOnError)
6876import Data.IORef (IORef, newIORef, writeIORef, readIORef)
6977import Data.ByteString (ByteString)
7078import qualified Data.ByteString as BS (concat)
@@ -174,29 +182,50 @@ import System.IO (hPutStrLn, stderr)
174182-- Internal datatypes --
175183--------------------------------------------------------------------------------
176184
177−data TCPTransport = TCPTransport { transportHost :: N.HostName
178− , transportPort :: N.ServiceName
179− , transportState :: MVar TransportState
180− }
181−data LocalEndPoint = LocalEndPoint { localAddress :: EndPointAddress
182− , localChannel :: Chan Event
183− , localState :: MVar LocalState
184− }
185−data RemoteEndPoint = RemoteEndPoint { remoteAddress :: EndPointAddress
186− , remoteState :: MVar RemoteState
187− }
188−
189−-- We use underscores for fields that we might update (using lenses).
190−
191−data TransportState = TransportState { _localEndPoints :: Map EndPointAddress LocalEndPoint
192− , _acceptNewConnections :: Bool
193− }
194−data LocalState = LocalState { _nextConnectionId :: ConnectionId
195− , _pendingCtrlRequests :: IntMap (MVar [ByteString])
196− , _nextCtrlRequestId :: ControlRequestId
197− , _localConnections :: Map EndPointAddress RemoteEndPoint
198− , _internalThreads :: [ThreadId]
199− }
185+-- We use underscores for fields that we might update (using lenses)
186+--
187+-- All data types follow the same structure:
188+--
189+-- * A top-level data type describing static properties (TCPTransport,
190+-- LocalEndPoint, RemoteEndPoint)
191+-- * The 'static' properties include an MVar containing a data structure for
192+-- the dynamic properties (TransportState, LocalEndPointState,
193+-- RemoteEndPointState). The state could be invalid/valid/closed,/etc.
194+-- * For the case of "valid" we use third data structure to give more details
195+-- about the state (ValidTransportState, ValidLocalEndPointState,
196+-- ValidRemoteEndPointState).
197+
198+data TCPTransport = TCPTransport
199+ { transportHost :: N.HostName
200+ , transportPort :: N.ServiceName
201+ , transportState :: MVar TransportState
202+ }
203+
204+data TransportState =
205+ TransportValid ValidTransportState
206+ | TransportClosed
207+
208+data ValidTransportState = ValidTransportState
209+ { _localEndPoints :: Map EndPointAddress LocalEndPoint
210+ }
211+
212+data LocalEndPoint = LocalEndPoint
213+ { localAddress :: EndPointAddress
214+ , localChannel :: Chan Event
215+ , localState :: MVar LocalEndPointState
216+ }
217+
218+data LocalEndPointState =
219+ LocalEndPointValid ValidLocalEndPointState
220+ | LocalEndPointClosed
221+
222+data ValidLocalEndPointState = ValidLocalEndPointState
223+ { _nextConnectionId :: ConnectionId
224+ , _pendingCtrlRequests :: IntMap (MVar [ByteString])
225+ , _nextCtrlRequestId :: ControlRequestId
226+ , _localConnections :: Map EndPointAddress RemoteEndPoint
227+ , _internalThreads :: [ThreadId]
228+ }
200229
201230-- A remote endpoint has incoming and outgoing connections, and when the total
202231-- number of connections (that is, the 'remoteRefCount') drops to zero we want
@@ -243,15 +272,23 @@ data LocalState = LocalState { _nextConnectionId :: ConnectionId
243272-- and then looks up the endpoint in localConnections it is guaranteed to
244273-- either find a different remote endpoint, or else none at all.
245274
246−data RemoteState = RemoteEndPointInvalid (FailedWith ConnectErrorCode)
247− | RemoteEndPointValid RemoteConnection
248− | RemoteEndPointClosing (MVar ()) RemoteConnection
249− | RemoteEndPointClosed
250−data RemoteConnection = RemoteConnection { _remoteOutgoing :: !Int
251− , _remoteIncoming :: IntSet
252− , remoteSocket :: N.Socket
253− , sendOn :: [ByteString] -> IO ()
254− }
275+data RemoteEndPoint = RemoteEndPoint
276+ { remoteAddress :: EndPointAddress
277+ , remoteState :: MVar RemoteState
278+ }
279+
280+data RemoteState =
281+ RemoteEndPointInvalid (FailedWith ConnectErrorCode)
282+ | RemoteEndPointValid ValidRemoteEndPointState
283+ | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState
284+ | RemoteEndPointClosed
285+
286+data ValidRemoteEndPointState = ValidRemoteEndPointState
287+ { _remoteOutgoing :: !Int
288+ , _remoteIncoming :: IntSet
289+ , remoteSocket :: N.Socket
290+ , sendOn :: [ByteString] -> IO ()
291+ }
255292
256293type EndPointId = Int32
257294type ControlRequestId = Int32
@@ -261,7 +298,7 @@ type EndPointPair = (LocalEndPoint, RemoteEndPoint)
261298data ControlHeader =
262299 RequestConnectionId -- ^ Request a new connection ID from the remote endpoint
263300 | CloseConnection -- ^ Tell the remote endpoint we will no longer be using a connection
264− | ControlResponse -- ^ Respond to a control request _from_ the remote endpoint
301+ | ControlResponse -- ^ Respond to a control request /from/ the remote endpoint
265302 | CloseSocket -- ^ Request to close the connection (see module description)
266303 deriving (Enum, Bounded, Show)
267304
@@ -281,23 +318,24 @@ data ConnectionRequestResponse =
281318-- TODOs: deal with hints
282319createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)
283320createTransport host port = do
284− state <- newMVar TransportState { _localEndPoints = Map.empty
285− , _acceptNewConnections = True
286− }
287− let transport = TCPTransport { transportState = state
288− , transportHost = host
289− , transportPort = port
290− }
321+ state <- newMVar . TransportValid $ ValidTransportState { _localEndPoints = Map.empty }
322+ let transport = TCPTransport { transportState = state
323+ , transportHost = host
324+ , transportPort = port
325+ }
291326 tryIO $ do
292− -- TODO: "5" is listed as "system dependent maximum backlog"?
293− transportThread <- forkServer host port 5 (terminationHandler transport) (handleConnectionRequest transport)
327+ -- For a discussion of the use of N.sOMAXCONN, see
328+ -- http://tangentsoft.net/wskfaq/advanced.html#backlog
329+ -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
330+ transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
294331 return Transport { newEndPoint = apiNewEndPoint transport
295332 , closeTransport = apiCloseTransport transportThread
296333 }
297334 where
298335 terminationHandler :: TCPTransport -> SomeException -> IO ()
299− terminationHandler transport _ =
300− modifyMVar_ (transportState transport) $ return . (acceptNewConnections ^= False)
336+ terminationHandler _ _ =
337+ -- TODO: implement
338+ return ()
301339
302340-- | Shut down the transport
303341apiCloseTransport :: ThreadId -> IO ()
@@ -313,15 +351,16 @@ apiCloseTransport transportThread = do
313351
314352-- | Create a new endpoint
315353apiNewEndPoint :: TCPTransport -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
316−apiNewEndPoint transport = do
354+apiNewEndPoint transport = try $ do
317355 ourEndPoint <- createLocalEndPoint transport
318− return . Right $ EndPoint { receive = readChan (localChannel ourEndPoint)
319− , address = localAddress ourEndPoint
320− , connect = apiConnect ourEndPoint
321− , closeEndPoint = apiCloseEndPoint transport ourEndPoint
322− , newMulticastGroup = return . Left $ newMulticastGroupError
323− , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
324− }
356+ return EndPoint
357+ { receive = readChan (localChannel ourEndPoint)
358+ , address = localAddress ourEndPoint
359+ , connect = apiConnect ourEndPoint
360+ , closeEndPoint = apiCloseEndPoint transport ourEndPoint
361+ , newMulticastGroup = return . Left $ newMulticastGroupError
362+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
363+ }
325364 where
326365 newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
327366 resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
@@ -333,19 +372,18 @@ apiConnect :: LocalEndPoint -- ^ Local end point
333372 -> IO (Either (FailedWith ConnectErrorCode) Connection)
334373apiConnect ourEndPoint theirAddress _ | localAddress ourEndPoint == theirAddress =
335374 connectToSelf ourEndPoint
336−apiConnect ourEndPoint theirAddress _ = runErrorT $ do
375+apiConnect ourEndPoint theirAddress _ = try $ do
337376 (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress
338377 -- connAlive can be an IORef rather than an MVar because it is protected by
339378 -- the remoteState MVar. We don't need the overhead of locking twice.
340− connAlive <- liftIO $ newIORef True
379+ connAlive <- newIORef True
341380 return $ Connection { send = apiSend theirEndPoint connId connAlive
342381 , close = apiClose theirEndPoint connId connAlive
343382 }
344383
345384-- | Close a connection
346385--
347−-- RELY: If the connection is alive on entry then the remote endpoint must
348−-- remain in the state 'RemoteEndPointValid'.
386+-- RELY: Remote endpoint must be in 'RemoteEndPointValid' or 'RemoteEndPointClosed'
349387-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
350388-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state
351389-- of the remote endpoint will not be changed.
@@ -353,15 +391,22 @@ apiConnect ourEndPoint theirAddress _ = runErrorT $ do
353391-- TODO: We ignore errors during a close. Is that right?
354392apiClose :: RemoteEndPoint -> ConnectionId -> IORef Bool -> IO ()
355393apiClose theirEndPoint connId connAlive = void . tryIO $ do
356− modifyMVar_ (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) -> do
357− alive <- readIORef connAlive
358− if alive
359− then do
360− sendOn remoteConn [encodeInt32 CloseConnection, encodeInt32 connId]
361− writeIORef connAlive False
362− return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ remoteConn)
363− else
364− return (RemoteEndPointValid remoteConn)
394+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
395+ RemoteEndPointValid vst -> do
396+ alive <- readIORef connAlive
397+ if alive
398+ then do
399+ writeIORef connAlive False
400+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
401+ return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ vst)
402+ else
403+ return (RemoteEndPointValid vst)
404+ RemoteEndPointClosed -> do
405+ return st
406+ RemoteEndPointClosing _ _ ->
407+ fail "apiClose RELY violation"
408+ RemoteEndPointInvalid _ ->
409+ fail "apiClose RELY violation"
365410 closeIfUnused theirEndPoint
366411
367412-- | Send data across a connection
@@ -370,38 +415,46 @@ apiClose theirEndPoint connId connAlive = void . tryIO $ do
370415-- GUARANTEE: The state of the remote endpoint will not be changed.
371416apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
372417apiSend theirEndPoint connId connAlive payload = do
373− withMVar (remoteState theirEndPoint) $ \st -> case st of
374− RemoteEndPointValid remoteConn -> do
375− alive <- readIORef connAlive
376− runErrorT $ if alive
377− then
378− failWithIO (FailedWith SendFailed . show) $
379− sendOn remoteConn (encodeInt32 connId : prependLength payload)
380− else
381− throwError $ FailedWith SendConnectionClosed "Connection closed"
382− RemoteEndPointClosed -> do
383− return (Left $ FailedWith SendConnectionClosed "Endpoint closed")
384− RemoteEndPointClosing _ _ ->
385− error "apiSend RELY violation"
386− RemoteEndPointInvalid _ ->
387− error "apiSend RELY violation"
418+ withMVar (remoteState theirEndPoint) $ \st -> case st of
419+ RemoteEndPointValid vst -> do
420+ alive <- readIORef connAlive
421+ try $ if alive
422+ then mapExceptionIO sendFailed $ sendOn vst (encodeInt32 connId : prependLength payload)
423+ else throw $ FailedWith SendConnectionClosed "Connection closed"
424+ RemoteEndPointClosed -> do
425+ return (Left $ FailedWith SendConnectionClosed "Endpoint closed")
426+ RemoteEndPointClosing _ _ ->
427+ error "apiSend RELY violation"
428+ RemoteEndPointInvalid _ ->
429+ error "apiSend RELY violation"
430+ where
431+ sendFailed :: IOException -> FailedWith SendErrorCode
432+ sendFailed = FailedWith SendFailed . show
388433
389434-- | Force-close the endpoint
390435apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
391436apiCloseEndPoint transport ourEndPoint = do
392437 -- Remove the reference from the transport state
393− modifyMVar_ (transportState transport) $ return . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
394− -- We take the endpoint state and don't put it back
395− -- (this may block some threads, but we don't care: we are killing them anyway)
396− ourState <- takeMVar (localState ourEndPoint)
397− -- Close all endpoints and kill all threads
398− incoming <- IntSet.elems . IntSet.unions <$> (forM (Map.elems $ ourState ^. localConnections) $ tryCloseRemoteSocket)
399− forM_ (ourState ^. internalThreads) killThread
400− -- We send a single message to the endpoint that it is closed. Subsequent
401− -- calls will block. We could change this so that all subsequent calls to
402− -- receive return an error, but this would mean checking for some state on
403− -- every call to receive, which is an unnecessary overhead.
404− writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)
438+ removeLocalEndPoint transport ourEndPoint
439+ -- Close the local endpoint
440+ mOurState <- modifyMVar (localState ourEndPoint) $ \st ->
441+ case st of
442+ LocalEndPointValid vst ->
443+ return (LocalEndPointClosed, Just vst)
444+ LocalEndPointClosed ->
445+ return (LocalEndPointClosed, Nothing)
446+ case mOurState of
447+ Nothing -> -- Already closed
448+ return ()
449+ Just vst -> do
450+ -- Close all endpoints and kill all threads
451+ incoming <- IntSet.elems . IntSet.unions <$> (forM (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket)
452+ forM_ (vst ^. internalThreads) killThread
453+ -- We send a single message to the endpoint that it is closed. Subsequent
454+ -- calls will block. We could change this so that all subsequent calls to
455+ -- receive return an error, but this would mean checking for some state on
456+ -- every call to receive, which is an unnecessary overhead.
457+ writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)
405458 where
406459 -- Close the remote socket and return the set of all incoming connections
407460 tryCloseRemoteSocket :: RemoteEndPoint -> IO IntSet
@@ -412,9 +465,11 @@ apiCloseEndPoint transport ourEndPoint = do
412465 RemoteEndPointInvalid _ ->
413466 return (st, IntSet.empty)
414467 RemoteEndPointValid conn -> do
415− tryIO $ do
416− sendOn conn [encodeInt32 CloseSocket]
417− N.sClose (remoteSocket conn)
468+ -- Try to send a CloseSocket request
469+ tryIO $ sendOn conn [encodeInt32 CloseSocket]
470+ -- .. but even if it fails, close the socket anyway
471+ -- (hence, two separate calls to tryIO)
472+ tryIO $ N.sClose (remoteSocket conn)
418473 return (RemoteEndPointClosed, conn ^. remoteIncoming)
419474 RemoteEndPointClosing _ conn -> do
420475 tryIO $ N.sClose (remoteSocket conn)
@@ -427,6 +482,7 @@ connectToSelf :: LocalEndPoint -> IO (Either (FailedWith ConnectErrorCode) Conne
427482connectToSelf ourEndPoint = do
428483 -- Here connAlive must an MVar because it is not protected by another lock
429484 connAlive <- newMVar True
485+ -- TODO: catch exception
430486 connId <- getNextConnectionId ourEndPoint
431487 writeChan ourChan (ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint))
432488 return . Right $ Connection { send = selfSend connAlive connId
@@ -457,176 +513,218 @@ connectToSelf ourEndPoint = do
457513--------------------------------------------------------------------------------
458514
459515-- | Create a new local endpoint
516+--
517+-- May throw a FailedWith NewEndPointErrorCode exception if the transport is closed.
460518createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
461519createLocalEndPoint transport = do
462− chan <- newChan
463− state <- newMVar LocalState { _nextConnectionId = firstNonReservedConnectionId
464− , _pendingCtrlRequests = IntMap.empty
465− , _nextCtrlRequestId = 0
466− , _localConnections = Map.empty
467− , _internalThreads = []
468− }
469− modifyMVar (transportState transport) $ \st -> do
470− let ix = fromIntegral $ Map.size (st ^. localEndPoints)
471− let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix
472− let localEndPoint = LocalEndPoint { localAddress = addr
473− , localChannel = chan
474− , localState = state
475− }
476− return ((localEndPointAt addr ^= Just localEndPoint) $ st, localEndPoint)
520+ chan <- newChan
521+ state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
522+ { _nextConnectionId = firstNonReservedConnectionId
523+ , _pendingCtrlRequests = IntMap.empty
524+ , _nextCtrlRequestId = 0
525+ , _localConnections = Map.empty
526+ , _internalThreads = []
527+ }
528+ modifyMVar (transportState transport) $ \st -> case st of
529+ TransportValid vst -> do
530+ let ix = fromIntegral $ Map.size (vst ^. localEndPoints)
531+ let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix
532+ let localEndPoint = LocalEndPoint { localAddress = addr
533+ , localChannel = chan
534+ , localState = state
535+ }
536+ return (TransportValid . (localEndPointAt addr ^= Just localEndPoint) $ vst, localEndPoint)
537+ TransportClosed ->
538+ throw (FailedWith NewEndPointTransportFailure "Transport closed")
477539
478540-- | Request a connection to a remote endpoint
479541--
480542-- This will block until we get a connection ID from the remote endpoint; if
481543-- the remote endpoint was in 'RemoteEndPointClosing' state then we will
482544-- additionally block until that is resolved.
545+--
546+-- May throw a FailedWith ConnectErrorCode exception.
483547requestConnectionTo :: LocalEndPoint
484548 -> EndPointAddress
485− -> ErrorT (FailedWith ConnectErrorCode) IO (RemoteEndPoint, ConnectionId)
486−requestConnectionTo ourEndPoint theirAddress = ErrorT go
549+ -> IO (RemoteEndPoint, ConnectionId)
550+requestConnectionTo ourEndPoint theirAddress = go
487551 where
488552 go = do
489− let ourState = localState ourEndPoint
490− (theirEndPoint, isNew) <- modifyMVar ourState $ \st ->
491− case st ^. localConnectionTo theirAddress of
492− Just theirEndPoint ->
493− return (st, (theirEndPoint, False))
494− Nothing -> do
495− theirState <- newEmptyMVar
496− let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
497− , remoteState = theirState
498− }
499− return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (theirEndPoint, True))
500−
501− -- If this is a new endpoint, fork a thread to listen for incoming
502− -- connections. We don't want to do this while we hold the lock, because
503− -- forkEndPointThread modifies the local state too (to record the thread ID)
504− when isNew . forkEndPointThread ourEndPoint $
505− connectToRemoteEndPoint (ourEndPoint, theirEndPoint)
506−
507− -- Before we initiate the new connection request we want to make sure
508− -- that refcount on the endpoint is incremented so that a concurrent
509− -- thread will not close the connection. Note that if IF we return
510− -- RemoteEndPointValid here then we can rely on the endpoint remaining in
511− -- that state.
553+ -- Find the remote endpoint (create it if it doesn't yet exist)
554+ theirEndPoint <- findTheirEndPoint
512555 let theirState = remoteState theirEndPoint
556+
557+ -- Before we initiate the new connection request we want to make sure that
558+ -- refcount on the endpoint is incremented so that a concurrent thread will
559+ -- not close the connection. Note that if IF we return RemoteEndPointValid
560+ -- here then we can rely on the endpoint remaining in that state.
513561 endPointStateSnapshot <- modifyMVar theirState $ \st ->
514562 case st of
515563 RemoteEndPointValid ep ->
516564 return (RemoteEndPointValid . (remoteOutgoing ^+= 1) $ ep, st)
517565 _ ->
518566 return (st, st)
519−
520− -- From this point on we are guaranteed the refcount is positive,
521− -- provided that the endpoint was valid. We still need to deal with the
522− -- case where it was not valid, however, which we didn't want to do that
523− -- while holding the endpoint lock.
524− -- Although 'endPointStateSnapshot' here refers to a snapshot of the endpoint
525− -- state, and might have changed in the meantime, these changes won't
526− -- matter.
567+
568+ -- From this point on we are guaranteed the refcount is positive, provided
569+ -- that the endpoint was valid. We still need to deal with the case where
570+ -- it was not valid, however, which we didn't want to do while holding the
571+ -- endpoint lock.
572+ --
573+ -- Although 'endPointStateSnapshot' here refers to a snapshot of the
574+ -- endpoint state, and might have changed in the meantime, these changes
575+ -- won't matter.
527576 case endPointStateSnapshot of
528577 RemoteEndPointInvalid err -> do
529− return . Left $ err
530−
531− RemoteEndPointClosing resolved _ -> do
578+ throw err
579+
580+ RemoteEndPointClosing resolved _ ->
532581 -- If the remote endpoint is closing, then we need to block until
533582 -- this is resolved and we then try again
534− readMVar resolved
535− go
536−
583+ readMVar resolved >> go
584+
537585 RemoteEndPointClosed -> do
538586 -- EndPointClosed indicates that a concurrent thread was in the
539587 -- process of closing the TCP connection to the remote endpoint when
540588 -- we obtained a reference to it. The remote endpoint will now have
541589 -- been removed from ourState, so we simply try again.
542590 go
543−
591+
544592 RemoteEndPointValid _ -> do
545− -- Do the actual connection request. This blocks until the remote
546− -- endpoint replies (but note that we don't hold any locks at this
547− -- point)
548− mcid <- tryIO $ do
549− reply <- doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
550− return (decodeInt32 . BS.concat $ reply)
551−
552593 -- On a failure we decrement the refcount again and return an error.
553594 -- The only state the remote endpoint can be in at this point is
554595 -- valid. As mentioned above, we can rely on the endpoint being in
555596 -- valid state at this point.
556− let failed err = do
557− modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
558− return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ ep)
559− -- TODO: should we call closeIfUnused here?
560− return . Left $ FailedWith ConnectFailed err
561−
562− case mcid of
563− Left err -> failed (show err)
564− Right Nothing -> failed "Invalid integer"
565− Right (Just cid) -> return . Right $ (theirEndPoint, cid)
597+ let failureHandler :: IOException -> IO b
598+ failureHandler err = do
599+ modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
600+ return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ ep)
601+ -- TODO: should we call closeIfUnused here?
602+ throw $ FailedWith ConnectFailed (show err)
603+
604+ -- Do the actual connection request. This blocks until the remote
605+ -- endpoint replies (but note that we don't hold any locks at this
606+ -- point)
607+ reply <- handle failureHandler $ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
608+ case decodeInt32 . BS.concat $ reply of
609+ Nothing -> failureHandler (userError "Invalid integer")
610+ Just cid -> return (theirEndPoint, cid)
566611
567−-- | Establish a connection to a remote endpoint
612+ -- If this is a new endpoint, fork a thread to listen for incoming
613+ -- connections. We don't want to do this while we hold the lock, because
614+ -- forkEndPointThread modifies the local state too (to record the thread
615+ -- ID)
616+ findTheirEndPoint :: IO RemoteEndPoint
617+ findTheirEndPoint = do
618+ (theirEndPoint, isNew) <- modifyMVar ourState $ \st -> case st of
619+ LocalEndPointClosed ->
620+ throw (FailedWith ConnectFailed "Local endpoint closed")
621+ LocalEndPointValid vst ->
622+ case vst ^. localConnectionTo theirAddress of
623+ Just theirEndPoint ->
624+ return (st, (theirEndPoint, False))
625+ Nothing -> do
626+ theirState <- newEmptyMVar
627+ let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
628+ , remoteState = theirState
629+ }
630+ return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) $ vst, (theirEndPoint, True))
631+
632+ -- The only way for forkEndPointThread to fail is if the local endpoint
633+ -- gets closed. This error will be caught elsewhere, so we ignore it
634+ -- here.
635+ when isNew . void . forkEndPointThread ourEndPoint $
636+ setupRemoteEndPoint (ourEndPoint, theirEndPoint)
637+ return theirEndPoint
638+
639+ ourState :: MVar LocalEndPointState
640+ ourState = localState ourEndPoint
641+
642+-- | Set up a remote endpoint
568643--
569644-- RELY: The state of the remote endpoint must be uninitialized.
570645-- GUARANTEE: Will only change the state to RemoteEndPointValid or
571646-- RemoteEndPointInvalid.
572−connectToRemoteEndPoint :: EndPointPair -> IO ()
573−connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do
574− let ourState = localState ourEndPoint
575− theirState = remoteState theirEndPoint
576− theirAddress = remoteAddress theirEndPoint
577− result <- runErrorT go
647+setupRemoteEndPoint :: EndPointPair -> IO ()
648+setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
649+ result <- socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint)
578650 case result of
579− Right (ConnectionRequestAccepted, sock) -> do
580− let remoteConn = RemoteConnection { remoteSocket = sock
581− , _remoteOutgoing = 0
582− , _remoteIncoming = IntSet.empty
583− , sendOn = sendMany sock
584− }
585− putMVar theirState (RemoteEndPointValid remoteConn)
651+ Right (sock, ConnectionRequestAccepted) -> do
652+ let vst = ValidRemoteEndPointState
653+ { remoteSocket = sock
654+ , _remoteOutgoing = 0
655+ , _remoteIncoming = IntSet.empty
656+ , sendOn = sendMany sock
657+ }
658+ putMVar theirState (RemoteEndPointValid vst)
586659 handleIncomingMessages (ourEndPoint, theirEndPoint)
587− Right (ConnectionRequestEndPointInvalid, sock) -> do
660+ Right (sock, ConnectionRequestEndPointInvalid) -> do
588661 -- We remove the endpoint from our local state again because the next
589662 -- call to 'connect' might give a different result. Threads that were
590663 -- waiting on the result of this call to connect will get the
591664 -- RemoteEndPointInvalid; subsequent threads will initiate a new
592665 -- connection requests.
593− modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)
666+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
594667 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
595668 N.sClose sock
596− Right (ConnectionRequestCrossed, sock) -> do
669+ Right (sock, ConnectionRequestCrossed) -> do
597670 N.sClose sock
598671 Left err -> do
599672 -- See comment above
600− modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)
673+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
601674 putMVar theirState (RemoteEndPointInvalid err)
602− -- TODO: this isn't quite right; socket *might* have been opened here,
603− -- in which case we should close it
604675 where
605− go :: ErrorT (FailedWith ConnectErrorCode) IO (ConnectionRequestResponse, N.Socket)
606− go = do
607− let theirAddress = remoteAddress theirEndPoint
608− (host, port, theirEndPointId) <- failWith (invalidAddress "Could not parse") $
609− decodeEndPointAddress theirAddress
610− addr:_ <- failWithIO (invalidAddress . show) $
611− N.getAddrInfo Nothing (Just host) (Just port)
612− sock <- failWithIO (insufficientResources . show) $
613− N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
614− response <- failWithIO (failed . show) $ do
676+ theirState = remoteState theirEndPoint
677+ invalidAddress = FailedWith ConnectInvalidAddress
678+
679+-- | Establish a connection to a remote endpoint
680+socketToEndPoint :: EndPointAddress -- ^ Our address
681+ -> EndPointAddress -- ^ Their address
682+ -> IO (Either (FailedWith ConnectErrorCode) (N.Socket, ConnectionRequestResponse))
683+socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
684+ (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
685+ Nothing -> throw (invalidAddress . userError $ "Could not parse")
686+ Just dec -> return dec
687+ addr:_ <- mapExceptionIO invalidAddress $ N.getAddrInfo Nothing (Just host) (Just port)
688+ bracketOnError (createSocket addr) N.sClose $ \sock -> do
689+ response <- mapExceptionIO failed $ do
615690 N.setSocketOption sock N.ReuseAddr 1
616691 N.connect sock (N.addrAddress addr)
617692 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
618693 recvInt32 sock
619694 case tryToEnum response of
620− Nothing -> throwError (failed "Unexpected response")
621− Just r -> return (r, sock)
695+ Nothing -> throw (failed . userError $ "Unexpected response")
696+ Just r -> return (sock, r)
697+ where
698+ createSocket :: N.AddrInfo -> IO N.Socket
699+ createSocket addr = mapExceptionIO insufficientResources $
700+ N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
622701
623− invalidAddress, insufficientResources, failed :: String -> FailedWith ConnectErrorCode
624− invalidAddress = FailedWith ConnectInvalidAddress
625− insufficientResources = FailedWith ConnectInsufficientResources
626− failed = FailedWith ConnectFailed
627−
628− ourAddress :: ByteString
629− EndPointAddress ourAddress = localAddress ourEndPoint
702+ invalidAddress, insufficientResources, failed :: IOException -> FailedWith ConnectErrorCode
703+ invalidAddress = FailedWith ConnectInvalidAddress . show
704+ insufficientResources = FailedWith ConnectInsufficientResources . show
705+ failed = FailedWith ConnectFailed . show
706+
707+-- | Remove reference to a remote endpoint from a local endpoint
708+--
709+-- If the local endpoint is closed, do nothing
710+removeRemoteEndPoint :: EndPointPair -> IO ()
711+removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do
712+ modifyMVar_ (localState ourEndPoint) $ \st -> case st of
713+ LocalEndPointValid vst ->
714+ return (LocalEndPointValid . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) $ vst)
715+ LocalEndPointClosed ->
716+ return LocalEndPointClosed
717+
718+-- | Remove reference to a local endpoint from the transport state
719+--
720+-- Does nothing if the transport is closed
721+removeLocalEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
722+removeLocalEndPoint transport ourEndPoint = do
723+ modifyMVar_ (transportState transport) $ \st -> case st of
724+ TransportValid vst ->
725+ return (TransportValid . (localEndPointAt (localAddress ourEndPoint) ^= Nothing) $ vst)
726+ TransportClosed ->
727+ return TransportClosed
630728
631729-- | Encode end point address
632730encodeEndPointAddress :: N.HostName -> N.ServiceName -> EndPointId -> EndPointAddress
@@ -645,42 +743,61 @@ decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack
645743--
646744-- RELY: Remote endpoint must be in valid state.
647745-- GUARANTEE: Will not change the state of the remote endpoint.
746+--
747+-- May throw IO (user) exception if the local endpoint is closed or if the send
748+-- fails.
648749doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
649−doRemoteRequest (ourEndPoint, theirEndPoint) header = do
750+doRemoteRequest (ourEndPoint, theirEndPoint) header = do
650751 reply <- newEmptyMVar
651− reqId <- modifyMVar (localState ourEndPoint) $ \st -> do
652− let reqId = st ^. nextCtrlRequestId
653− return ((nextCtrlRequestId ^+= 1) . (pendingCtrlRequestsAt reqId ^= Just reply) $ st, reqId)
654− withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) ->
655− sendOn remoteConn [encodeInt32 header, encodeInt32 reqId]
752+ reqId <- modifyMVar (localState ourEndPoint) $ \st -> case st of
753+ LocalEndPointValid vst -> do
754+ let reqId = vst ^. nextCtrlRequestId
755+ return (LocalEndPointValid . (nextCtrlRequestId ^+= 1) . (pendingCtrlRequestsAt reqId ^= Just reply) $ vst, reqId)
756+ LocalEndPointClosed ->
757+ throw (userError "Local endpoint closed")
758+ withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid vst) ->
759+ sendOn vst [encodeInt32 header, encodeInt32 reqId]
656760 takeMVar reply
657761
658762-- | Check if the remote endpoint is unused, and if so, send a CloseSocket request
659763closeIfUnused :: RemoteEndPoint -> IO ()
660764closeIfUnused theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ \st ->
661765 case st of
662− RemoteEndPointValid remoteConn -> do
663− if remoteConn ^. remoteOutgoing == 0 && IntSet.null (remoteConn ^. remoteIncoming)
766+ RemoteEndPointValid vst -> do
767+ if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
664768 then do
665769 resolved <- newEmptyMVar
666− sendOn remoteConn [encodeInt32 CloseSocket]
667− return (RemoteEndPointClosing resolved remoteConn)
770+ sendOn vst [encodeInt32 CloseSocket]
771+ return (RemoteEndPointClosing resolved vst)
668772 else
669773 return st
670774 _ -> do
671775 return st
672776
673777-- | Fork a new thread and store its ID as part of the transport state
674−forkEndPointThread :: LocalEndPoint -> IO () -> IO ()
675−forkEndPointThread ourEndPoint p = do
676− modifyMVar_ (localState ourEndPoint) $ \st -> do
677− tid <- forkIO (p >> removeThread)
678− return (internalThreads ^%= (tid :) $ st)
778+--
779+-- If the local end point is closed this function does nothing (no thread is
780+-- spawned). Returns whether or not a thread was spawned.
781+forkEndPointThread :: LocalEndPoint -> IO () -> IO Bool
782+forkEndPointThread ourEndPoint p =
783+ modifyMVar ourState $ \st -> case st of
784+ LocalEndPointValid vst -> do
785+ tid <- forkIO (p >> removeThread)
786+ return (LocalEndPointValid . (internalThreads ^%= (tid :)) $ vst, True)
787+ LocalEndPointClosed ->
788+ return (LocalEndPointClosed, False)
679789 where
680790 removeThread :: IO ()
681791 removeThread = do
682792 tid <- myThreadId
683− modifyMVar_ (localState ourEndPoint) $ return . (internalThreads ^%= filter (/= tid))
793+ modifyMVar_ ourState $ \st -> case st of
794+ LocalEndPointValid vst ->
795+ return (LocalEndPointValid . (internalThreads ^%= filter (/= tid)) $ vst)
796+ LocalEndPointClosed ->
797+ return LocalEndPointClosed
798+
799+ ourState :: MVar LocalEndPointState
800+ ourState = localState ourEndPoint
684801
685802--------------------------------------------------------------------------------
686803-- Incoming requests --
@@ -688,46 +805,84 @@ forkEndPointThread ourEndPoint p = do
688805
689806-- | Handle a connection request (that is, a remote endpoint that is trying to
690807-- establish a TCP connection with us)
808+--
809+-- 'handleConnectionRequest' runs in the context of the transport thread, which
810+-- can be killed asynchronously by 'closeTransport'. We fork a separate thread
811+-- as soon as we have located the lcoal endpoint that the remote endpoint is
812+-- interested in. We cannot fork any sooner because then we have no way of
813+-- storing the thread ID and hence no way of killing the thread when we take
814+-- the transport down. We must be careful to close the socket when a (possibly
815+-- asynchronous, ThreadKilled) exception occurs. (If an exception escapes from
816+-- handleConnectionRequest the transport will be shut down.)
691817handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
692−handleConnectionRequest transport sock = do
693− request <- tryIO $ do
818+handleConnectionRequest transport sock = handle tryCloseSocket $ do
694819 ourEndPointId <- recvInt32 sock
695820 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
696821 let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId
697− ourEndPoint <- (^. localEndPointAt ourAddress) <$> readMVar (transportState transport)
698− return (theirAddress, ourEndPoint)
699− case request of
700− Right (_, Nothing) -> do
701− sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
702− N.sClose sock
703− Right (theirAddress, Just ourEndPoint) -> forkEndPointThread ourEndPoint $ do
704− (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->
705− case st ^. localConnectionTo theirAddress of
822+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
823+ TransportValid vst ->
824+ case vst ^. localEndPointAt ourAddress of
706825 Nothing -> do
707− theirState <- newEmptyMVar
708− let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
709− , remoteState = theirState
710− }
711− return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (False, theirEndPoint))
712− Just theirEndPoint ->
713− return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
714− if crossed
715− then do
716− sendMany sock [encodeInt32 ConnectionRequestCrossed]
717− N.sClose sock
718− else do
719− let remoteConn = RemoteConnection { remoteSocket = sock
720− , _remoteOutgoing = 0
721− , _remoteIncoming = IntSet.empty
722− , sendOn = sendMany sock
723− }
724− -- TODO: this putMVar might block
725− putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)
726− sendMany sock [encodeInt32 ConnectionRequestAccepted]
826+ sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
827+ fail "Invalid endpoint"
828+ Just ourEndPoint ->
829+ return ourEndPoint
830+ TransportClosed -> do
831+ fail "Transport closed"
832+ void . forkEndPointThread ourEndPoint $ go ourEndPoint theirAddress
833+ where
834+ go :: LocalEndPoint -> EndPointAddress -> IO ()
835+ go ourEndPoint theirAddress = do
836+ mEndPoint <- handle (\e -> invalidEndPoint e >> return Nothing) $ do
837+ (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st -> case st of
838+ LocalEndPointValid vst ->
839+ case vst ^. localConnectionTo theirAddress of
840+ Nothing -> do
841+ theirState <- newEmptyMVar
842+ let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
843+ , remoteState = theirState
844+ }
845+ return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) $ vst, (False, theirEndPoint))
846+ Just theirEndPoint ->
847+ return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
848+ LocalEndPointClosed ->
849+ fail "Local endpoint closed"
850+ if crossed
851+ then do
852+ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
853+ tryIO $ N.sClose sock
854+ return Nothing
855+ else do
856+ let vst = ValidRemoteEndPointState
857+ { remoteSocket = sock
858+ , _remoteOutgoing = 0
859+ , _remoteIncoming = IntSet.empty
860+ , sendOn = sendMany sock
861+ }
862+ -- TODO: this putMVar might block if the remote endpoint sends a
863+ -- connection request for a local endpoint that it is already
864+ -- connected to
865+ putMVar (remoteState theirEndPoint) (RemoteEndPointValid vst)
866+ sendMany sock [encodeInt32 ConnectionRequestAccepted]
867+ return (Just theirEndPoint)
868+ -- If we left the scope of the exception handler with a return value of
869+ -- Nothing then the socket is already closed; otherwise, the socket has
870+ -- been recorded as part of the remote endpoint. Either way, we no longer
871+ -- have to worry about closing the socket on receiving an asynchronous
872+ -- exception from this point forward.
873+ case mEndPoint of
874+ Nothing ->
875+ return ()
876+ Just theirEndPoint ->
727877 handleIncomingMessages (ourEndPoint, theirEndPoint)
728− Left _ -> do
729− -- Invalid request
730− N.sClose sock
878+
879+ tryCloseSocket :: IOException -> IO ()
880+ tryCloseSocket _ = void . tryIO $ N.sClose sock
881+
882+ invalidEndPoint :: IOException -> IO ()
883+ invalidEndPoint ex = do
884+ tryIO $ sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]
885+ tryCloseSocket ex
731886
732887-- | Handle requests from a remote endpoint.
733888--
@@ -785,10 +940,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
785940 createNewConnection reqId = do
786941 newId <- getNextConnectionId ourEndPoint
787942 modifyMVar_ theirState $ \st -> do
788− remoteConn <- case st of
789− RemoteEndPointValid remoteConn ->
790− return (remoteIncoming ^%= IntSet.insert newId $ remoteConn)
791− RemoteEndPointClosing resolved remoteConn -> do
943+ vst <- case st of
944+ RemoteEndPointValid vst ->
945+ return (remoteIncoming ^%= IntSet.insert newId $ vst)
946+ RemoteEndPointClosing resolved vst -> do
792947 -- If the endpoint is in closing state that means we send a
793948 -- CloseSocket request to the remote endpoint. If the remote
794949 -- endpoint replies with the request to create a new connection, it
@@ -796,24 +951,27 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
796951 -- ours. Either way, at this point we simply restore the endpoint
797952 -- to RemoteEndPointValid
798953 putMVar resolved ()
799− return (remoteIncoming ^= IntSet.singleton newId $ remoteConn)
954+ return (remoteIncoming ^= IntSet.singleton newId $ vst)
800955 _ ->
801956 error "handleIncomingMessages RELY violation"
802− sendOn remoteConn ( encodeInt32 ControlResponse
957+ sendOn vst ( encodeInt32 ControlResponse
803958 : encodeInt32 reqId
804959 : prependLength [encodeInt32 newId]
805960 )
806961 -- We add the new connection ID to the list of open connections only once the
807962 -- endpoint has been notified of the new connection (sendOn may fail)
808− return (RemoteEndPointValid remoteConn)
963+ return (RemoteEndPointValid vst)
809964 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)
810965
811966 -- Read a control response
812967 readControlResponse :: N.Socket -> ControlRequestId -> IO ()
813968 readControlResponse sock reqId = do
814969 response <- recvWithLength sock
815− mmvar <- modifyMVar ourState $ \st ->
816− return (pendingCtrlRequestsAt reqId ^= Nothing $ st, st ^. pendingCtrlRequestsAt reqId)
970+ mmvar <- modifyMVar ourState $ \st -> case st of
971+ LocalEndPointValid vst ->
972+ return (LocalEndPointValid . (pendingCtrlRequestsAt reqId ^= Nothing) $ vst, vst ^. pendingCtrlRequestsAt reqId)
973+ LocalEndPointClosed ->
974+ fail "Local endpoint closed"
817975 case mmvar of
818976 Nothing -> do
819977 hPutStrLn stderr $ "Warning: Invalid request ID"
@@ -826,8 +984,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
826984 closeConnection cid = do
827985 -- TODO: we should check that this connection is in fact open
828986 writeChan ourChannel (ConnectionClosed cid)
829− modifyMVar_ theirState $ \(RemoteEndPointValid remoteConn) ->
830− return (RemoteEndPointValid . (remoteIncoming ^%= IntSet.delete cid) $ remoteConn)
987+ modifyMVar_ theirState $ \(RemoteEndPointValid vst) ->
988+ return (RemoteEndPointValid . (remoteIncoming ^%= IntSet.delete cid) $ vst)
831989 closeIfUnused theirEndPoint
832990
833991 -- Close the socket (if we don't have any outgoing connections)
@@ -839,27 +997,27 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
839997 -- endpoint in Closed state once we remove the endpoint from our local
840998 -- connections. But we can do that only once we are sure that we can
841999 -- close the endpoint. Catch-22. We can resolve the catch-22 by locking
842− -- *both* our local state and the endpoint state, but at the cost of
1000+ -- /both/ our local state and the endpoint state, but at the cost of
8431001 -- introducing a double lock and all the associated perils, or by putting
8441002 -- the remote endpoint in Closing state first. We opt for the latter.
8451003 canClose <- modifyMVar theirState $ \st ->
8461004 case st of
847− RemoteEndPointValid remoteConn -> do
1005+ RemoteEndPointValid vst -> do
8481006 -- We regard a CloseSocket message as an (optimized) way for the
8491007 -- remote endpoint to indicate that all its connections to us are
8501008 -- now properly closed
851− forM_ (IntSet.elems $ remoteConn ^. remoteIncoming) $ \cid ->
1009+ forM_ (IntSet.elems $ vst ^. remoteIncoming) $ \cid ->
8521010 writeChan ourChannel (ConnectionClosed cid)
853− let remoteConn' = remoteIncoming ^= IntSet.empty $ remoteConn
1011+ let vst' = remoteIncoming ^= IntSet.empty $ vst
8541012 -- Check if we agree that the connection should be closed
855− if remoteConn' ^. remoteOutgoing == 0
1013+ if vst' ^. remoteOutgoing == 0
8561014 then do
8571015 -- Attempt to reply (but don't insist)
858− tryIO $ sendOn remoteConn' [encodeInt32 CloseSocket]
1016+ tryIO $ sendOn vst' [encodeInt32 CloseSocket]
8591017 resolved <- newEmptyMVar
860− return (RemoteEndPointClosing resolved remoteConn', Just resolved)
1018+ return (RemoteEndPointClosing resolved vst', Just resolved)
8611019 else
862− return (RemoteEndPointValid remoteConn', Nothing)
1020+ return (RemoteEndPointValid vst', Nothing)
8631021 RemoteEndPointClosing resolved _ ->
8641022 return (st, Just resolved)
8651023 _ ->
@@ -869,7 +1027,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
8691027 Nothing ->
8701028 return False
8711029 Just resolved -> do
872− modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)
1030+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
8731031 modifyMVar_ theirState $ return . const RemoteEndPointClosed
8741032 N.sClose sock
8751033 putMVar resolved ()
@@ -889,13 +1047,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
8891047 prematureExit :: N.Socket -> IOException -> IO ()
8901048 prematureExit sock _ = do
8911049 N.sClose sock
892− modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)
1050+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
8931051 mUnclosedConnections <- modifyMVar theirState $ \st ->
8941052 case st of
8951053 RemoteEndPointInvalid _ ->
8961054 error "handleIncomingMessages RELY violation"
897− RemoteEndPointValid remoteConn ->
898− return (RemoteEndPointClosed, Just $ remoteConn ^. remoteIncoming)
1055+ RemoteEndPointValid vst ->
1056+ return (RemoteEndPointClosed, Just $ vst ^. remoteIncoming)
8991057 RemoteEndPointClosing _ _ ->
9001058 return (RemoteEndPointClosed, Nothing)
9011059 RemoteEndPointClosed ->
@@ -910,12 +1068,17 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
9101068 ErrorEventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)
9111069
9121070-- | Get the next connection ID
1071+--
1072+-- Throws an IO exception when the endpoint is closed.
9131073getNextConnectionId :: LocalEndPoint -> IO ConnectionId
9141074getNextConnectionId ourEndpoint =
915− modifyMVar (localState ourEndpoint) $ \st -> do
916− let connId = st ^. nextConnectionId
917− return (nextConnectionId ^= connId +1 $ st, connId)
918−
1075+ modifyMVar (localState ourEndpoint) $ \st -> case st of
1076+ LocalEndPointValid vst -> do
1077+ let connId = vst ^. nextConnectionId
1078+ return (LocalEndPointValid . (nextConnectionId ^= connId + 1) $ vst, connId)
1079+ LocalEndPointClosed ->
1080+ fail "Local endpoint closed"
1081+
9191082--------------------------------------------------------------------------------
9201083-- Constants --
9211084--------------------------------------------------------------------------------
@@ -928,38 +1091,35 @@ firstNonReservedConnectionId = 1024
9281091-- Lens definitions --
9291092--------------------------------------------------------------------------------
9301093
931−localEndPoints :: Lens TransportState (Map EndPointAddress LocalEndPoint)
1094+localEndPoints :: Lens ValidTransportState (Map EndPointAddress LocalEndPoint)
9321095localEndPoints = lens _localEndPoints (\es st -> st { _localEndPoints = es })
9331096
934−acceptNewConnections :: Lens TransportState Bool
935−acceptNewConnections = lens _acceptNewConnections (\acc st -> st { _acceptNewConnections = acc })
936−
937−pendingCtrlRequests :: Lens LocalState (IntMap (MVar [ByteString]))
1097+pendingCtrlRequests :: Lens ValidLocalEndPointState (IntMap (MVar [ByteString]))
9381098pendingCtrlRequests = lens _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })
9391099
940−nextCtrlRequestId :: Lens LocalState ControlRequestId
1100+nextCtrlRequestId :: Lens ValidLocalEndPointState ControlRequestId
9411101nextCtrlRequestId = lens _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid })
9421102
943−nextConnectionId :: Lens LocalState ConnectionId
1103+nextConnectionId :: Lens ValidLocalEndPointState ConnectionId
9441104nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
9451105
946−localConnections :: Lens LocalState (Map EndPointAddress RemoteEndPoint)
1106+localConnections :: Lens ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
9471107localConnections = lens _localConnections (\es st -> st { _localConnections = es })
9481108
949−internalThreads :: Lens LocalState [ThreadId]
1109+internalThreads :: Lens ValidLocalEndPointState [ThreadId]
9501110internalThreads = lens _internalThreads (\ts st -> st { _internalThreads = ts })
9511111
952−localEndPointAt :: EndPointAddress -> Lens TransportState (Maybe LocalEndPoint)
1112+localEndPointAt :: EndPointAddress -> Lens ValidTransportState (Maybe LocalEndPoint)
9531113localEndPointAt addr = localEndPoints >>> mapLens addr
9541114
955−pendingCtrlRequestsAt :: ControlRequestId -> Lens LocalState (Maybe (MVar [ByteString]))
1115+pendingCtrlRequestsAt :: ControlRequestId -> Lens ValidLocalEndPointState (Maybe (MVar [ByteString]))
9561116pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)
9571117
958−localConnectionTo :: EndPointAddress -> Lens LocalState (Maybe RemoteEndPoint)
1118+localConnectionTo :: EndPointAddress -> Lens ValidLocalEndPointState (Maybe RemoteEndPoint)
9591119localConnectionTo addr = localConnections >>> mapLens addr
9601120
961−remoteOutgoing :: Lens RemoteConnection Int
1121+remoteOutgoing :: Lens ValidRemoteEndPointState Int
9621122remoteOutgoing = lens _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })
9631123
964−remoteIncoming :: Lens RemoteConnection IntSet
1124+remoteIncoming :: Lens ValidRemoteEndPointState IntSet
9651125remoteIncoming = lens _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })