git.iko.soy/distributed-process
on master help sign in
More error handling
8fbebc7Edsko de Vries
network-transport/src/Network/Transport/TCP.hs+303−129
compare with master
unifiedsplit
@@ -17,6 +17,7 @@ module Network.Transport.TCP ( -- * Main API
17 createTransport
17 createTransport
18 -- * Internals (exposed for unit tests)
18 -- * Internals (exposed for unit tests)
19 , createTransportExposeInternals
19 , createTransportExposeInternals
20+ , TransportInternals(..)
20 , EndPointId
21 , EndPointId
21 , encodeEndPointAddress
22 , encodeEndPointAddress
22 , decodeEndPointAddress
23 , decodeEndPointAddress
@@ -74,8 +75,18 @@ import Control.Concurrent.MVar ( MVar
74 )
75 )
75import Control.Category ((>>>))
76import Control.Category ((>>>))
76import Control.Applicative ((<$>))
77import Control.Applicative ((<$>))
77−import Control.Monad (forM_, when, unless)
78+import Control.Monad (when, unless, join)
79+import Control.Monad.Error (ErrorT(..), runErrorT)
78−import Control.Exception (IOException, SomeException, handle, throw, try, bracketOnError, mask_)
80+import Control.Exception ( IOException
81+ , SomeException
82+ , handle
83+ , throw
84+ , try
85+ , bracketOnError
86+ , mask
87+ , mask_
88+ , onException
89+ )
79import Data.IORef (IORef, newIORef, writeIORef, readIORef)
90import Data.IORef (IORef, newIORef, writeIORef, readIORef)
80import Data.ByteString (ByteString)
91import Data.ByteString (ByteString)
81import qualified Data.ByteString as BS (concat)
92import qualified Data.ByteString as BS (concat)
@@ -89,6 +100,7 @@ import Data.Map (Map)
89import qualified Data.Map as Map (empty, elems)
100import qualified Data.Map as Map (empty, elems)
90import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
101import Data.Accessor (Accessor, accessor, (^.), (^=), (^:))
91import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
102import qualified Data.Accessor.Container as DAC (mapMaybe, intMapMaybe)
103+import Data.Foldable (forM_)
92import System.IO (hPutStrLn, stderr)
104import System.IO (hPutStrLn, stderr)
93
105
94-- $design
106-- $design
@@ -229,6 +241,7 @@ data ValidLocalEndPointState = ValidLocalEndPointState
229 , _nextCtrlRequestId :: ControlRequestId
241 , _nextCtrlRequestId :: ControlRequestId
230 , _localConnections :: Map EndPointAddress RemoteEndPoint
242 , _localConnections :: Map EndPointAddress RemoteEndPoint
231 , _internalThreads :: [ThreadId]
243 , _internalThreads :: [ThreadId]
244+ , _nextRemoteId :: Int
232 }
245 }
233
246
234-- A remote endpoint has incoming and outgoing connections, and when the total
247-- A remote endpoint has incoming and outgoing connections, and when the total
@@ -270,15 +283,31 @@ data ValidLocalEndPointState = ValidLocalEndPointState
270-- Since we don't do concurrent reads from the same socket we don't need to
283-- Since we don't do concurrent reads from the same socket we don't need to
271-- take the lock when reading from the socket.
284-- take the lock when reading from the socket.
272--
285--
273−-- Moreover, we insist on the invariant (INV-CLOSE) that whenever we put an
286+-- Invariants:
287+--
274−-- endpoint in closed state we remove that endpoint from localConnections
288+-- INV-CLOSE: Whenever we put an endpoint in closed state we remove that
275−-- first, so that if a concurrent thread reads the mvar, finds EndPointClosed,
289+-- endpoint from localConnections first, so that if a concurrent thread reads
290+-- the mvar, finds EndPointClosed, and then looks up the endpoint in
276−-- and then looks up the endpoint in localConnections it is guaranteed to
291+-- localConnections it is guaranteed to either find a different remote
277−-- either find a different remote endpoint, or else none at all.
292+-- endpoint, or else none at all.
293+-- INV-RESOLVE: Whenever we move a endpoint from Closing to Closed state, we
294+-- signal on the corresponding MVar only *after* the endpoint has been put in
295+-- Closed state. This way when to threads try to resolve they don't both
296+-- attempt to write to the "resolved" MVar. TODO: Make sure that this
297+-- invariant is adhered too.
298+-- INV-LOST: If a send or recv fails, or a socket is closed unexpectedly, we
299+-- first put the remote endpoint in Closing or Closed state, and then send a
300+-- EventConnectionLost event. This guarantees that we only send this event
301+-- once.
302+-- INV-CLOSING: An endpoint in closing state is for all intents and purposes
303+-- closed; that is, we shouldn't do any 'send's on it (although 'recv' is
304+-- acceptable, of course -- as we are waiting for the remote endpoint to
305+-- confirm or deny the request).
278
306
279data RemoteEndPoint = RemoteEndPoint
307data RemoteEndPoint = RemoteEndPoint
280 { remoteAddress :: EndPointAddress
308 { remoteAddress :: EndPointAddress
281 , remoteState :: MVar RemoteState
309 , remoteState :: MVar RemoteState
310+ , remoteId :: Int
282 }
311 }
283
312
284data RemoteState =
313data RemoteState =
@@ -313,6 +342,14 @@ data ConnectionRequestResponse =
313 | ConnectionRequestCrossed -- ^ /A/s request crossed with a request from /B/ (see protocols)
342 | ConnectionRequestCrossed -- ^ /A/s request crossed with a request from /B/ (see protocols)
314 deriving (Enum, Bounded, Show)
343 deriving (Enum, Bounded, Show)
315
344
345+-- Internal functionality we expose for unit testing
346+data TransportInternals = TransportInternals
347+ { -- | The ID of the thread that listens for new incoming connections
348+ transportThread :: ThreadId
349+ -- | Find the socket between a local and a remote endpoint
350+ , socketBetween :: EndPointAddress -> EndPointAddress -> IO (Either String N.Socket)
351+ }
352+
316--------------------------------------------------------------------------------
353--------------------------------------------------------------------------------
317-- Top-level functionality --
354-- Top-level functionality --
318--------------------------------------------------------------------------------
355--------------------------------------------------------------------------------
@@ -330,7 +367,7 @@ createTransport host port = do
330 return $ Right transport
367 return $ Right transport
331
368
332-- | You should probably not use this function (used for unit testing only)
369-- | You should probably not use this function (used for unit testing only)
333−createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, ThreadId))
370+createTransportExposeInternals :: N.HostName -> N.ServiceName -> IO (Either IOException (Transport, TransportInternals))
334createTransportExposeInternals host port = do
371createTransportExposeInternals host port = do
335 state <- newMVar . TransportValid $ ValidTransportState
372 state <- newMVar . TransportValid $ ValidTransportState
336 { _localEndPoints = Map.empty
373 { _localEndPoints = Map.empty
@@ -344,11 +381,13 @@ createTransportExposeInternals host port = do
344 -- For a discussion of the use of N.sOMAXCONN, see
381 -- For a discussion of the use of N.sOMAXCONN, see
345 -- http://tangentsoft.net/wskfaq/advanced.html#backlog
382 -- http://tangentsoft.net/wskfaq/advanced.html#backlog
346 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
383 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
347− transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
384+ tid <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
348 return ( Transport { newEndPoint = apiNewEndPoint transport
385 return ( Transport { newEndPoint = apiNewEndPoint transport
349− , closeTransport = apiCloseTransport transport (Just transportThread) [EndPointClosed]
386+ , closeTransport = apiCloseTransport transport (Just tid) [EndPointClosed]
350 }
387 }
351− , transportThread
388+ , TransportInternals { transportThread = tid
389+ , socketBetween = internalSocketBetween transport
390+ }
352 )
391 )
353 where
392 where
354 terminationHandler :: TCPTransport -> SomeException -> IO ()
393 terminationHandler :: TCPTransport -> SomeException -> IO ()
@@ -366,22 +405,13 @@ apiCloseTransport transport mTransportThread evs = do
366 mTSt <- modifyMVar (transportState transport) $ \st -> case st of
405 mTSt <- modifyMVar (transportState transport) $ \st -> case st of
367 TransportValid vst -> return (TransportClosed, Just vst)
406 TransportValid vst -> return (TransportClosed, Just vst)
368 TransportClosed -> return (TransportClosed, Nothing)
407 TransportClosed -> return (TransportClosed, Nothing)
369− case mTSt of
370− Nothing ->
371− -- Transport already closed
372− return ()
373− Just tSt -> do
408+ forM_ mTSt $ \tSt -> do
374− mapM_ (apiCloseEndPoint transport evs) (Map.elems $ tSt ^. localEndPoints)
409+ mapM_ (apiCloseEndPoint transport evs) (Map.elems $ tSt ^. localEndPoints)
375− case mTransportThread of
410+ forM_ mTransportThread $ \tid -> do
376− Just transportThread ->
377− -- This will invoke the termination handler, which in turn will call
411+ -- This will invoke the termination handler, which in turn will call
378− -- apiCloseTransport again, but at that point the transport is
412+ -- apiCloseTransport again, but at that point the transport is
379− -- already in closed state so at that point we return immediately
413+ -- already in closed state so at that point we return immediately
380− killThread transportThread
414+ killThread tid
381− Nothing ->
382− -- This will happen is apiCloseTransport was invoked *because* the
383− -- transport thread died (in which case we don't need to kill it)
384− return ()
385
415
386-- | Create a new endpoint
416-- | Create a new endpoint
387apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
417apiNewEndPoint :: TCPTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
@@ -412,8 +442,8 @@ apiConnect ourEndPoint theirAddress _reliability _hints = try $ do
412 -- connAlive can be an IORef rather than an MVar because it is protected by
442 -- connAlive can be an IORef rather than an MVar because it is protected by
413 -- the remoteState MVar. We don't need the overhead of locking twice.
443 -- the remoteState MVar. We don't need the overhead of locking twice.
414 connAlive <- newIORef True
444 connAlive <- newIORef True
415− return $ Connection { send = apiSend theirEndPoint connId connAlive
445+ return $ Connection { send = apiSend (ourEndPoint, theirEndPoint) connId connAlive
416− , close = apiClose theirEndPoint connId connAlive
446+ , close = apiClose (ourEndPoint, theirEndPoint) connId connAlive
417 }
447 }
418
448
419-- | Close a connection
449-- | Close a connection
@@ -422,53 +452,49 @@ apiConnect ourEndPoint theirAddress _reliability _hints = try $ do
422-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
452-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
423-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state
453-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state
424-- of the remote endpoint will not be changed.
454-- of the remote endpoint will not be changed.
425−apiClose :: RemoteEndPoint -> ConnectionId -> IORef Bool -> IO ()
455+apiClose :: EndPointPair -> ConnectionId -> IORef Bool -> IO ()
426−apiClose theirEndPoint connId connAlive = void . tryIO $ do
456+apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
427− modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
457+ modifyRemoteState_ (ourEndPoint, theirEndPoint)
458+ -- RemoteEndPointInvalid
459+ (\_ -> fail "apiClose RELY violation")
428− RemoteEndPointValid vst -> do
460+ -- RemoteEndPointValid
461+ (\vst -> do
429 alive <- readIORef connAlive
462 alive <- readIORef connAlive
430 if alive
463 if alive
431 then do
464 then do
432 writeIORef connAlive False
465 writeIORef connAlive False
433− -- Possible exception caught top-level
434 sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
466 sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
435 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ vst)
467 return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ vst)
436 else
468 else
437− return (RemoteEndPointValid vst)
469+ return (RemoteEndPointValid vst))
470+ -- RemoteEndPointClosing
471+ (\(resolved, vst) -> return $ RemoteEndPointClosing resolved vst)
472+ -- RemoteEndPoinClosed
438− RemoteEndPointClosed -> do
473+ (return $ RemoteEndPointClosed)
439− return st
440− RemoteEndPointClosing _ _ ->
441− -- See discussion in apiSend why we must allow for this case
442− return st
443− RemoteEndPointInvalid _ ->
444− fail "apiClose RELY violation"
445− closeIfUnused theirEndPoint
474+ closeIfUnused (ourEndPoint, theirEndPoint)
446
475
447-- | Send data across a connection
476-- | Send data across a connection
448--
477--
449-- RELY: The endpoint must not be in invalid state.
478-- RELY: The endpoint must not be in invalid state.
450-- GUARANTEE: The state of the remote endpoint will not be changed.
479-- GUARANTEE: The state of the remote endpoint will not be changed.
451−apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
480+apiSend :: EndPointPair -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (TransportError SendErrorCode) ())
452−apiSend theirEndPoint connId connAlive payload = do
481+apiSend (ourEndPoint, theirEndPoint) connId connAlive payload = do
453− withMVar (remoteState theirEndPoint) $ \st -> case st of
482+ join <$> (try . mapExceptionIO sendFailed $ modifyRemoteState (ourEndPoint, theirEndPoint)
483+ -- RemoteEndPointInvalid
484+ (\_ -> fail "apiSend RELY violation")
454− RemoteEndPointValid vst -> do
485+ -- RemoteEndPointValid
486+ (\vst -> do
455 alive <- readIORef connAlive
487 alive <- readIORef connAlive
456− try $ if alive
488+ if alive
489+ then do
457− then mapExceptionIO sendFailed $ sendOn vst (encodeInt32 connId : prependLength payload)
490+ sendOn vst (encodeInt32 connId : prependLength payload)
458− else throw $ TransportError SendClosed "Connection closed"
491+ return (RemoteEndPointValid vst, Right $ ())
459− RemoteEndPointClosed -> do
492+ else
460− return . Left $ TransportError SendClosed "Connection lost"
493+ return (RemoteEndPointValid vst, Left $ TransportError SendClosed "Connection closed"))
461− RemoteEndPointClosing _ _ ->
494+ -- RemoteEndPointClosing
462− -- The only way for the endpoint to be in closing state, while a
495+ (\(resolved, vst) -> return (RemoteEndPointClosing resolved vst, Left $ TransportError SendClosed "Connection lost"))
463− -- connection is still active, is for the application to call 'close'
496+ -- RemoteEndPointClosed
464− -- on the connection triggering "garbage collection" of the TCP channel
465− -- (CloseSocket request to the remote endpoint) which hasn't completed
466− -- yet. Even if the remote endpoint comes back with "please don't
467− -- close", this still means that our *outgoing* connection has been
468− -- closed
469− return . Left $ TransportError SendClosed "Connection lost"
497+ (return (RemoteEndPointClosed, Left $ TransportError SendClosed "Connection lost")))
470− RemoteEndPointInvalid _ ->
471− error "apiSend RELY violation"
472 where
498 where
473 sendFailed :: IOException -> TransportError SendErrorCode
499 sendFailed :: IOException -> TransportError SendErrorCode
474 sendFailed = TransportError SendFailed . show
500 sendFailed = TransportError SendFailed . show
@@ -488,14 +514,11 @@ apiCloseEndPoint transport evs ourEndPoint = do
488 return (LocalEndPointClosed, Just vst)
514 return (LocalEndPointClosed, Just vst)
489 LocalEndPointClosed ->
515 LocalEndPointClosed ->
490 return (LocalEndPointClosed, Nothing)
516 return (LocalEndPointClosed, Nothing)
491− case mOurState of
517+ forM_ mOurState $ \vst -> do
492− Nothing -> -- Already closed
493− return ()
494− Just vst -> do
495− -- Close all endpoints and kill all threads
518+ -- Close all endpoints and kill all threads
496− forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket
519+ forM_ (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket
497− forM_ (vst ^. internalThreads) killThread
520+ forM_ (vst ^. internalThreads) killThread
498− forM_ evs $ writeChan (localChannel ourEndPoint)
521+ forM_ evs $ writeChan (localChannel ourEndPoint)
499 where
522 where
500 -- Close the remote socket and return the set of all incoming connections
523 -- Close the remote socket and return the set of all incoming connections
501 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
524 tryCloseRemoteSocket :: RemoteEndPoint -> IO ()
@@ -552,6 +575,43 @@ connectToSelf ourEndPoint = do
552 when alive $ writeChan ourChan (ConnectionClosed connId)
575 when alive $ writeChan ourChan (ConnectionClosed connId)
553 return False
576 return False
554
577
578+--------------------------------------------------------------------------------
579+-- Functions from TransportInternals --
580+--------------------------------------------------------------------------------
581+
582+-- Find a socket between two endpoints
583+internalSocketBetween :: TCPTransport -- ^ Transport
584+ -> EndPointAddress -- ^ Local endpoint
585+ -> EndPointAddress -- ^ Remote endpoint
586+ -> IO (Either String N.Socket)
587+internalSocketBetween transport ourAddress theirAddress = runErrorT $ do
588+ ourEndPoint <- ErrorT $ do
589+ withMVar (transportState transport) $ \st -> case st of
590+ TransportClosed ->
591+ return . Left $ "Transport closed"
592+ TransportValid vst -> do
593+ case vst ^. localEndPointAt ourAddress of
594+ Nothing -> return . Left $ "Local endpoint not found"
595+ Just ep -> return . Right $ ep
596+ theirEndPoint <- ErrorT $ do
597+ withMVar (localState ourEndPoint) $ \st -> case st of
598+ LocalEndPointClosed ->
599+ return . Left $ "Local endpoint closed"
600+ LocalEndPointValid vst -> do
601+ case vst ^. localConnectionTo theirAddress of
602+ Nothing -> return . Left $ "Remote endpoint not found"
603+ Just ep -> return . Right $ ep
604+ ErrorT $ do
605+ withMVar (remoteState theirEndPoint) $ \st -> case st of
606+ RemoteEndPointValid vst ->
607+ return . Right $ remoteSocket vst
608+ RemoteEndPointClosing _ _ ->
609+ return . Left $ "Remote endpoint closing"
610+ RemoteEndPointClosed ->
611+ return . Left $ "Remote endpoint closed"
612+ RemoteEndPointInvalid _ ->
613+ return . Left $ "Remote endpoint invalid"
614+
555--------------------------------------------------------------------------------
615--------------------------------------------------------------------------------
556-- Lower level functionality --
616-- Lower level functionality --
557--------------------------------------------------------------------------------
617--------------------------------------------------------------------------------
@@ -568,6 +628,7 @@ createLocalEndPoint transport = do
568 , _nextCtrlRequestId = 0
628 , _nextCtrlRequestId = 0
569 , _localConnections = Map.empty
629 , _localConnections = Map.empty
570 , _internalThreads = []
630 , _internalThreads = []
631+ , _nextRemoteId = 0
571 }
632 }
572 modifyMVar (transportState transport) $ \st -> case st of
633 modifyMVar (transportState transport) $ \st -> case st of
573 TransportValid vst -> do
634 TransportValid vst -> do
@@ -635,23 +696,15 @@ requestConnectionTo ourEndPoint theirAddress = go
635 go
696 go
636
697
637 RemoteEndPointValid _ -> do
698 RemoteEndPointValid _ -> do
638− -- On a failure we decrement the refcount again and return an error.
639− -- The only state the remote endpoint can be in at this point is
640− -- valid. As mentioned above, we can rely on the endpoint being in
641− -- valid state at this point.
642− let failureHandler :: IOException -> IO b
643− failureHandler err = do
644− modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
645− return (RemoteEndPointValid . (remoteOutgoing ^: (\x -> x - 1)) $ ep)
646− throw $ TransportError ConnectFailed (show err)
647−
648 -- Do the actual connection request. This blocks until the remote
699 -- Do the actual connection request. This blocks until the remote
649 -- endpoint replies (but note that we don't hold any locks at this
700 -- endpoint replies (but note that we don't hold any locks at this
650− -- point)
701+ -- 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
703+ -- closed state.
651− reply <- handle failureHandler $ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
704+ reply <- mapExceptionIO connectFailed $ doRemoteRequest (ourEndPoint, theirEndPoint) RequestConnectionId
652 case decodeInt32 . BS.concat $ reply of
705 case decodeInt32 . BS.concat $ reply of
653 Nothing ->
706 Nothing ->
654− failureHandler (userError "Invalid integer")
707+ throw (connectFailed $ userError "Invalid integer")
655 Just cid -> do
708 Just cid -> do
656 return (theirEndPoint, cid)
709 return (theirEndPoint, cid)
657
710
@@ -672,8 +725,9 @@ requestConnectionTo ourEndPoint theirAddress = go
672 theirState <- newEmptyMVar
725 theirState <- newEmptyMVar
673 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
726 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
674 , remoteState = theirState
727 , remoteState = theirState
728+ , remoteId = vst ^. nextRemoteId
675 }
729 }
676− return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) $ vst, (theirEndPoint, True))
730+ return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) . (nextRemoteId ^: (+ 1)) $ vst, (theirEndPoint, True))
677
731
678 -- The only way for forkEndPointThread to fail is if the local endpoint
732 -- The only way for forkEndPointThread to fail is if the local endpoint
679 -- gets closed. This error will be caught elsewhere, so we ignore it
733 -- gets closed. This error will be caught elsewhere, so we ignore it
@@ -685,6 +739,9 @@ requestConnectionTo ourEndPoint theirAddress = go
685 ourState :: MVar LocalEndPointState
739 ourState :: MVar LocalEndPointState
686 ourState = localState ourEndPoint
740 ourState = localState ourEndPoint
687
741
742+ connectFailed :: IOException -> TransportError ConnectErrorCode
743+ connectFailed = TransportError ConnectFailed . show
744+
688-- | Set up a remote endpoint
745-- | Set up a remote endpoint
689--
746--
690-- RELY: The state of the remote endpoint must be uninitialized.
747-- RELY: The state of the remote endpoint must be uninitialized.
@@ -700,7 +757,7 @@ setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
700 { remoteSocket = sock
757 { remoteSocket = sock
701 , _remoteOutgoing = 0
758 , _remoteOutgoing = 0
702 , _remoteIncoming = IntSet.empty
759 , _remoteIncoming = IntSet.empty
703− , sendOn = sendMany sock
760+ , sendOn = \msg -> sendMany sock msg `onException` (tryIO $ N.sClose sock)
704 }
761 }
705 putMVar theirState (RemoteEndPointValid vst)
762 putMVar theirState (RemoteEndPointValid vst)
706 return True
763 return True
@@ -783,11 +840,20 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
783-- If the local endpoint is closed, do nothing
840-- If the local endpoint is closed, do nothing
784removeRemoteEndPoint :: EndPointPair -> IO ()
841removeRemoteEndPoint :: EndPointPair -> IO ()
785removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do
842removeRemoteEndPoint (ourEndPoint, theirEndPoint) = do
786− modifyMVar_ (localState ourEndPoint) $ \st -> case st of
843+ modifyMVar_ ourState $ \st -> case st of
787− LocalEndPointValid vst ->
844+ LocalEndPointValid vst ->
845+ case vst ^. localConnectionTo theirAddress of
846+ Nothing ->
847+ return st
848+ Just remoteEndPoint' ->
849+ if remoteId remoteEndPoint' == remoteId theirEndPoint
788− return (LocalEndPointValid . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) $ vst)
850+ then return (LocalEndPointValid . (localConnectionTo (remoteAddress theirEndPoint) ^= Nothing) $ vst)
851+ else return st
789− LocalEndPointClosed ->
852+ LocalEndPointClosed ->
790− return LocalEndPointClosed
853+ return LocalEndPointClosed
854+ where
855+ ourState = localState ourEndPoint
856+ theirAddress = remoteAddress theirEndPoint
791
857
792-- | Remove reference to a local endpoint from the transport state
858-- | Remove reference to a local endpoint from the transport state
793--
859--
@@ -817,11 +883,11 @@ decodeEndPointAddress (EndPointAddress bs) = case map BSC.unpack $ BSC.split ':'
817
883
818-- | Do a (blocking) remote request
884-- | Do a (blocking) remote request
819--
885--
820−-- RELY: Remote endpoint must be in valid state.
886+-- RELY: Remote endpoint must be in valid or closed state.
821-- GUARANTEE: Will not change the state of the remote endpoint.
887-- GUARANTEE: Will not change the state of the remote endpoint.
822--
888--
823−-- May throw IO (user) exception if the local endpoint is closed or if the send
889+-- May throw IO (user) exception if the local or the remote endpoint is closed,
824−-- fails.
890+-- or if the send fails.
825doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
891doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
826doRemoteRequest (ourEndPoint, theirEndPoint) header = do
892doRemoteRequest (ourEndPoint, theirEndPoint) header = do
827 reply <- newEmptyMVar
893 reply <- newEmptyMVar
@@ -831,24 +897,35 @@ doRemoteRequest (ourEndPoint, theirEndPoint) header = do
831 return (LocalEndPointValid . (nextCtrlRequestId ^: (+ 1)) . (pendingCtrlRequestsAt reqId ^= Just reply) $ vst, reqId)
897 return (LocalEndPointValid . (nextCtrlRequestId ^: (+ 1)) . (pendingCtrlRequestsAt reqId ^= Just reply) $ vst, reqId)
832 LocalEndPointClosed ->
898 LocalEndPointClosed ->
833 throw (userError "Local endpoint closed")
899 throw (userError "Local endpoint closed")
900+ withRemoteState (ourEndPoint, theirEndPoint)
901+ -- RemoteEndPointInvalid
902+ (\_ -> fail "doRemoteRequest RELY violation")
834− withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid vst) ->
903+ -- RemoteEndPointValid
835− sendOn vst [encodeInt32 header, encodeInt32 reqId]
904+ (\vst -> sendOn vst [encodeInt32 header, encodeInt32 reqId])
905+ -- RemoteEndPointClosing
906+ (\_ -> fail "doRemoteRequest RELY violation")
907+ -- RemoteEndPointClosed
908+ (throw (userError "Remote endpoint closed"))
836 takeMVar reply
909 takeMVar reply
837
910
838-- | Check if the remote endpoint is unused, and if so, send a CloseSocket request
911-- | Check if the remote endpoint is unused, and if so, send a CloseSocket request
839−closeIfUnused :: RemoteEndPoint -> IO ()
912+closeIfUnused :: EndPointPair -> IO ()
840−closeIfUnused theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ \st ->
913+closeIfUnused (ourEndPoint, theirEndPoint) =
841− case st of
914+ modifyRemoteState_ (ourEndPoint, theirEndPoint)
842− RemoteEndPointValid vst -> do
915+ -- RemoteEndPointInvalid
916+ (return . RemoteEndPointInvalid)
917+ -- RemoteEndPointValid
843− if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
918+ (\vst -> if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
844− then do
919+ then do
845− resolved <- newEmptyMVar
846− sendOn vst [encodeInt32 CloseSocket]
920+ sendOn vst [encodeInt32 CloseSocket]
921+ resolved <- newEmptyMVar
847− return (RemoteEndPointClosing resolved vst)
922+ return $ RemoteEndPointClosing resolved vst
848− else
923+ else
849− return st
924+ return $ RemoteEndPointValid vst)
850− _ -> do
925+ -- RemoteEndPointClosing
851− return st
926+ (return . uncurry RemoteEndPointClosing)
927+ -- RemoteEndPointClosed
928+ (return RemoteEndPointClosed)
852
929
853-- | Fork a new thread and store its ID as part of the transport state
930-- | Fork a new thread and store its ID as part of the transport state
854--
931--
@@ -886,6 +963,109 @@ forkEndPointThread ourEndPoint p =
886 ourState :: MVar LocalEndPointState
963 ourState :: MVar LocalEndPointState
887 ourState = localState ourEndPoint
964 ourState = localState ourEndPoint
888
965
966+-- | Like modifyMVar, but if an exception occurs don't restore the remote
967+-- endpoint to its original value but close it instead
968+modifyRemoteState :: EndPointPair -- ^ Local and remote endpoint
969+ -> (TransportError ConnectErrorCode -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointInvalid
970+ -> (ValidRemoteEndPointState -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointValid
971+ -> ((MVar (), ValidRemoteEndPointState) -> IO (RemoteState, a)) -- ^ Case for RemoteEndPointClosing
972+ -> IO (RemoteState, a) -- ^ Case for RemoteEndPointClosed
973+ -> IO a
974+modifyRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed = do
975+ mask $ \restore -> do
976+ st <- takeMVar theirState
977+ case st of
978+ RemoteEndPointValid vst -> do
979+ mResult <- try $ restore (caseValid vst)
980+ case mResult of
981+ Right (st', a) -> do
982+ putMVar theirState st'
983+ return a
984+ Left ex -> do
985+ handleException ex vst
986+ -- 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
988+ RemoteEndPointClosing resolved vst -> do
989+ (st', a) <- restore (caseClosing (resolved, vst)) `onException` putMVar theirState st
990+ putMVar theirState st'
991+ return a
992+ RemoteEndPointInvalid err -> do
993+ (st', a) <- restore (caseInvalid err) `onException` putMVar theirState st
994+ putMVar theirState st'
995+ return a
996+ RemoteEndPointClosed -> do
997+ (st', a) <- restore caseClosed `onException` putMVar theirState st
998+ putMVar theirState st'
999+ return a
1000+ where
1001+ theirState :: MVar RemoteState
1002+ theirState = remoteState theirEndPoint
1003+
1004+ handleException :: SomeException -> ValidRemoteEndPointState -> IO a
1005+ handleException ex vst = do
1006+ -- 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
1008+ -- two locks at the same time. So we put it in closing state first ..
1009+ resolved <- newEmptyMVar
1010+ putMVar theirState (RemoteEndPointClosing resolved vst)
1011+ -- .. then remove it from the local endpoint ..
1012+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1013+ -- .. close it and notify the local endpoint we lost the connection ..
1014+ mIncoming <- closeRemoteEndPoint
1015+ forM_ mIncoming $ \incoming -> do
1016+ let err = TransportError (EventConnectionLost (remoteAddress theirEndPoint) incoming) (show ex)
1017+ writeChan (localChannel ourEndPoint) $ ErrorEvent err
1018+ -- .. and finally rethrow the exception
1019+ throw ex
1020+
1021+ -- Returns the set of incoming connections if we closed the connection, or
1022+ -- 'Nothing' if the endpoint was already closed
1023+ closeRemoteEndPoint :: IO (Maybe [ConnectionId])
1024+ closeRemoteEndPoint = do
1025+ st <- takeMVar theirState
1026+ case st of
1027+ RemoteEndPointClosing resolved vst -> do
1028+ putMVar theirState RemoteEndPointClosed
1029+ putMVar resolved ()
1030+ return . Just . IntSet.elems $ vst ^. remoteIncoming
1031+ RemoteEndPointClosed -> do
1032+ putMVar theirState RemoteEndPointClosed
1033+ return Nothing
1034+ _ ->
1035+ fail "the impossible happened"
1036+
1037+-- | Like 'modifyRemoteState' but without a return value
1038+modifyRemoteState_ :: EndPointPair -- ^ Local and remote endpoint
1039+ -> (TransportError ConnectErrorCode -> IO RemoteState) -- ^ Case for RemoteEndPointInvalid
1040+ -> (ValidRemoteEndPointState -> IO RemoteState) -- ^ Case for RemoteEndPointValid
1041+ -> ((MVar (), ValidRemoteEndPointState) -> IO RemoteState) -- ^ Case for RemoteEndPointClosing
1042+ -> IO RemoteState -- ^ Case for RemoteEndPointClosed
1043+ -> IO ()
1044+modifyRemoteState_ (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed =
1045+ modifyRemoteState (ourEndPoint, theirEndPoint)
1046+ (u . caseInvalid)
1047+ (u . caseValid)
1048+ (u . caseClosing)
1049+ (u caseClosed)
1050+ where
1051+ u :: IO a -> IO (a, ())
1052+ u p = p >>= \a -> return (a, ())
1053+
1054+-- | Like 'modifyRemoteState' but without the ability to change the state
1055+withRemoteState :: EndPointPair -- ^ Local and remote endpoint
1056+ -> (TransportError ConnectErrorCode -> IO a) -- ^ Case for RemoteEndPointInvalid
1057+ -> (ValidRemoteEndPointState -> IO a) -- ^ Case for RemoteEndPointValid
1058+ -> ((MVar (), ValidRemoteEndPointState) -> IO a) -- ^ Case for RemoteEndPointClosing
1059+ -> IO a -- ^ Case for RemoteEndPointClosed
1060+ -> IO a
1061+withRemoteState (ourEndPoint, theirEndPoint) caseInvalid caseValid caseClosing caseClosed = do
1062+ modifyRemoteState (ourEndPoint, theirEndPoint)
1063+ (\err -> (,) (RemoteEndPointInvalid err) <$> caseInvalid err)
1064+ (\vst -> (,) (RemoteEndPointValid vst) <$> caseValid vst)
1065+ (\(resolved, vst) -> (,) (RemoteEndPointClosing resolved vst) <$> caseClosing (resolved, vst))
1066+ ((,) RemoteEndPointClosed <$> caseClosed)
1067+
1068+
889--------------------------------------------------------------------------------
1069--------------------------------------------------------------------------------
890-- Incoming requests --
1070-- Incoming requests --
891--------------------------------------------------------------------------------
1071--------------------------------------------------------------------------------
@@ -928,8 +1108,9 @@ handleConnectionRequest transport sock = handle tryCloseSocket $ do
928 theirState <- newEmptyMVar
1108 theirState <- newEmptyMVar
929 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
1109 let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
930 , remoteState = theirState
1110 , remoteState = theirState
1111+ , remoteId = vst ^. nextRemoteId
931 }
1112 }
932− return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) $ vst, (False, theirEndPoint))
1113+ return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) . (nextRemoteId ^: (+ 1)) $ vst, (False, theirEndPoint))
933 Just theirEndPoint ->
1114 Just theirEndPoint ->
934 return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
1115 return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
935 LocalEndPointClosed ->
1116 LocalEndPointClosed ->
@@ -958,11 +1139,7 @@ handleConnectionRequest transport sock = handle tryCloseSocket $ do
958 -- been recorded as part of the remote endpoint. Either way, we no longer
1139 -- been recorded as part of the remote endpoint. Either way, we no longer
959 -- have to worry about closing the socket on receiving an asynchronous
1140 -- have to worry about closing the socket on receiving an asynchronous
960 -- exception from this point forward.
1141 -- exception from this point forward.
961− case mEndPoint of
962− Nothing ->
963− return ()
964− Just theirEndPoint ->
965− handleIncomingMessages (ourEndPoint, theirEndPoint)
1142+ forM_ mEndPoint $ handleIncomingMessages . (,) ourEndPoint
966
1143
967 tryCloseSocket :: IOException -> IO ()
1144 tryCloseSocket :: IOException -> IO ()
968 tryCloseSocket _ = void . tryIO $ N.sClose sock
1145 tryCloseSocket _ = void . tryIO $ N.sClose sock
@@ -1076,7 +1253,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1076 writeChan ourChannel (ConnectionClosed cid)
1253 writeChan ourChannel (ConnectionClosed cid)
1077 modifyMVar_ theirState $ \(RemoteEndPointValid vst) ->
1254 modifyMVar_ theirState $ \(RemoteEndPointValid vst) ->
1078 return (RemoteEndPointValid . (remoteIncoming ^: IntSet.delete cid) $ vst)
1255 return (RemoteEndPointValid . (remoteIncoming ^: IntSet.delete cid) $ vst)
1079− closeIfUnused theirEndPoint
1256+ closeIfUnused (ourEndPoint, theirEndPoint)
1080
1257
1081 -- Close the socket (if we don't have any outgoing connections)
1258 -- Close the socket (if we don't have any outgoing connections)
1082 closeSocket :: N.Socket -> IO Bool
1259 closeSocket :: N.Socket -> IO Bool
@@ -1138,24 +1315,18 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
1138 prematureExit sock err = do
1315 prematureExit sock err = do
1139 N.sClose sock
1316 N.sClose sock
1140 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1317 removeRemoteEndPoint (ourEndPoint, theirEndPoint)
1141− mUnclosedConnections <- modifyMVar theirState $ \st ->
1318+ modifyMVar_ theirState $ \st ->
1142 case st of
1319 case st of
1143 RemoteEndPointInvalid _ ->
1320 RemoteEndPointInvalid _ ->
1144 error "handleIncomingMessages RELY violation"
1321 error "handleIncomingMessages RELY violation"
1145− RemoteEndPointValid vst ->
1322+ RemoteEndPointValid vst -> do
1146− return (RemoteEndPointClosed, Just $ vst ^. remoteIncoming)
1323+ writeChan ourChannel . ErrorEvent $
1324+ TransportError (EventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems $ vst ^. remoteIncoming)) (show err)
1325+ return RemoteEndPointClosed
1147 RemoteEndPointClosing _ _ ->
1326 RemoteEndPointClosing _ _ ->
1148− return (RemoteEndPointClosed, Nothing)
1327+ return RemoteEndPointClosed
1149 RemoteEndPointClosed ->
1328 RemoteEndPointClosed ->
1150− return (st, Nothing)
1329+ return RemoteEndPointClosed
1151−
1152− -- We send the connection lost message even if unclosedConnections is the
1153− -- empty set, because *outgoing* connections will be broken too now
1154− case mUnclosedConnections of
1155− Nothing ->
1156− return ()
1157− Just unclosedConnections -> writeChan ourChannel . ErrorEvent $
1158− TransportError (EventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)) (show err)
1159
1330
1160-- | Get the next connection ID
1331-- | Get the next connection ID
1161--
1332--
@@ -1202,6 +1373,9 @@ localConnections = accessor _localConnections (\es st -> st { _localConnections
1202internalThreads :: Accessor ValidLocalEndPointState [ThreadId]
1373internalThreads :: Accessor ValidLocalEndPointState [ThreadId]
1203internalThreads = accessor _internalThreads (\ts st -> st { _internalThreads = ts })
1374internalThreads = accessor _internalThreads (\ts st -> st { _internalThreads = ts })
1204
1375
1376+nextRemoteId :: Accessor ValidLocalEndPointState Int
1377+nextRemoteId = accessor _nextRemoteId (\rid st -> st { _nextRemoteId = rid })
1378+
1205localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1379localEndPointAt :: EndPointAddress -> Accessor ValidTransportState (Maybe LocalEndPoint)
1206localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1380localEndPointAt addr = localEndPoints >>> DAC.mapMaybe addr
1207
1381
network-transport/tests/TestTCP.hs+169−19
@@ -6,29 +6,44 @@ import Prelude hiding (catch, (>>=), (>>), return, fail)
6import TestTransport (testTransport)
6import TestTransport (testTransport)
7import TestAuxiliary (forkTry, runTests)
7import TestAuxiliary (forkTry, runTests)
8import Network.Transport
8import Network.Transport
9−import Network.Transport.TCP (createTransport, createTransportExposeInternals, encodeEndPointAddress)
9+import Network.Transport.TCP ( createTransport
10+ , createTransportExposeInternals
11+ , TransportInternals(..)
12+ , encodeEndPointAddress
13+ )
10import Data.Int (Int32)
14import Data.Int (Int32)
11import Control.Concurrent (threadDelay, ThreadId, killThread)
15import Control.Concurrent (threadDelay, ThreadId, killThread)
12−import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar)
16+import Control.Concurrent.MVar ( MVar
17+ , newEmptyMVar
18+ , putMVar
19+ , takeMVar
20+ , readMVar
21+ , isEmptyMVar
22+ , newMVar
23+ , modifyMVar
24+ )
13−import Control.Monad (replicateM, guard, forM_, replicateM_)
25+import Control.Monad (replicateM, guard, forM_, replicateM_, when)
14import Control.Applicative ((<$>))
26import Control.Applicative ((<$>))
15−import Control.Exception (throw)
27+import Control.Exception (throw, try, SomeException)
16import Network.Transport.TCP ( ControlHeader(..)
28import Network.Transport.TCP ( ControlHeader(..)
17 , ConnectionRequestResponse(..)
29 , ConnectionRequestResponse(..)
18 , socketToEndPoint
30 , socketToEndPoint
19 )
31 )
20−import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO)
32+import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO, void)
21import Network.Transport.Internal.TCP (recvInt32, forkServer, recvWithLength)
33import Network.Transport.Internal.TCP (recvInt32, forkServer, recvWithLength)
22import qualified Network.Socket as N ( sClose
34import qualified Network.Socket as N ( sClose
23 , ServiceName
35 , ServiceName
24 , Socket
36 , Socket
25 , AddrInfo
37 , AddrInfo
38+ , shutdown
39+ , ShutdownCmd(ShutdownSend)
26 )
40 )
27import Network.Socket.ByteString (sendMany)
41import Network.Socket.ByteString (sendMany)
28import Data.String (fromString)
42import Data.String (fromString)
29import Traced
43import Traced
30import GHC.IO.Exception (ioe_errno)
44import GHC.IO.Exception (ioe_errno)
31import Foreign.C.Error (Errno(..), eADDRNOTAVAIL)
45import Foreign.C.Error (Errno(..), eADDRNOTAVAIL)
46+import System.Timeout (timeout)
32
47
33instance Traceable ControlHeader where
48instance Traceable ControlHeader where
34 trace = traceShow
49 trace = traceShow
@@ -37,7 +52,7 @@ instance Traceable ConnectionRequestResponse where
37 trace = traceShow
52 trace = traceShow
38
53
39instance Traceable N.Socket where
54instance Traceable N.Socket where
40− trace = const Nothing
55+ trace = traceShow
41
56
42instance Traceable N.AddrInfo where
57instance Traceable N.AddrInfo where
43 trace = traceShow
58 trace = traceShow
@@ -45,6 +60,12 @@ instance Traceable N.AddrInfo where
45instance Traceable ThreadId where
60instance Traceable ThreadId where
46 trace = const Nothing
61 trace = const Nothing
47
62
63+instance Traceable TransportInternals where
64+ trace = const Nothing
65+
66+instance Traceable SomeException where
67+ trace = traceShow
68+
48-- Test that the server gets a ConnectionClosed message when the client closes
69-- Test that the server gets a ConnectionClosed message when the client closes
49-- the socket without sending an explicit control message to the server first
70-- the socket without sending an explicit control message to the server first
50testEarlyDisconnect :: IO N.ServiceName -> IO ()
71testEarlyDisconnect :: IO N.ServiceName -> IO ()
@@ -498,10 +519,10 @@ testMany nextPort = do
498-- | Test what happens when the transport breaks completely
519-- | Test what happens when the transport breaks completely
499testBreakTransport :: IO N.ServiceName -> IO ()
520testBreakTransport :: IO N.ServiceName -> IO ()
500testBreakTransport nextPort = do
521testBreakTransport nextPort = do
501− Right (transport, transportThread) <- nextPort >>= createTransportExposeInternals "127.0.0.1"
522+ Right (transport, internals) <- nextPort >>= createTransportExposeInternals "127.0.0.1"
502 Right endpoint <- newEndPoint transport
523 Right endpoint <- newEndPoint transport
503
524
504− killThread transportThread -- Uh oh
525+ killThread (transportThread internals) -- Uh oh
505
526
506 ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint
527 ErrorEvent (TransportError EventTransportFailed _) <- receive endpoint
507
528
@@ -510,37 +531,165 @@ testBreakTransport nextPort = do
510-- | Test that a second call to 'connect' might succeed even if the first
531-- | Test that a second call to 'connect' might succeed even if the first
511-- failed. This is a TCP specific test rather than an endpoint specific test
532-- failed. This is a TCP specific test rather than an endpoint specific test
512-- because we must manually create the endpoint address to match an endpoint we
533-- because we must manually create the endpoint address to match an endpoint we
513−-- have yet to set up
534+-- have yet to set up.
535+-- Then test that we get a connection lost message after the remote endpoint
536+-- suddenly closes the socket, and that a subsequent 'connect' allows us to
537+-- re-establish a connection to the same endpoint
514testReconnect :: IO N.ServiceName -> IO ()
538testReconnect :: IO N.ServiceName -> IO ()
515testReconnect nextPort = do
539testReconnect nextPort = do
540+ serverPort <- nextPort
516− clientDone <- newEmptyMVar
541+ serverDone <- newEmptyMVar
517 firstAttempt <- newEmptyMVar
542 firstAttempt <- newEmptyMVar
518 endpointCreated <- newEmptyMVar
543 endpointCreated <- newEmptyMVar
519− port <- nextPort
520− Right transport <- createTransport "127.0.0.1" port
521
544
522 -- Server
545 -- Server
523 forkTry $ do
546 forkTry $ do
547+ -- Wait for the client to do its first attempt
524− takeMVar firstAttempt
548+ readMVar firstAttempt
525− newEndPoint transport
549+
550+ counter <- newMVar (0 :: Int)
551+
552+ forkServer "127.0.0.1" serverPort 5 throw $ \sock -> do
553+ -- Accept the connection
554+ Right 0 <- tryIO $ (recvInt32 sock :: IO Int)
555+ Right _ <- tryIO $ recvWithLength sock
556+ Right () <- tryIO $ sendMany sock [encodeInt32 ConnectionRequestAccepted]
557+
558+ -- The first time we close the socket before accepting the logical connection
559+ count <- modifyMVar counter $ \i -> return (i + 1, i)
560+
561+ when (count > 0) $ do
562+ -- Client requests a logical connection
563+ Right RequestConnectionId <- tryIO $ toEnum <$> (recvInt32 sock :: IO Int)
564+ Right reqId <- tryIO $ (recvInt32 sock :: IO Int)
565+ Right () <- tryIO $ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])
566+ return ()
567+
568+ when (count > 1) $ do
569+ -- Client sends a message
570+ Right 10001 <- tryIO $ (recvInt32 sock :: IO Int)
571+ Right ["ping"] <- tryIO $ recvWithLength sock
572+ putMVar serverDone ()
573+
574+ Right () <- tryIO $ N.sClose sock
575+ return ()
576+
526 putMVar endpointCreated ()
577 putMVar endpointCreated ()
527
578
528 -- Client
579 -- Client
529 forkTry $ do
580 forkTry $ do
581+ Right transport <- nextPort >>= createTransport "127.0.0.1"
530− Right endpoint <- newEndPoint transport
582+ Right endpoint <- newEndPoint transport
531− let theirAddr = encodeEndPointAddress "127.0.0.1" port 1
583+ let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0
532
584
585+ -- The first attempt will fail because no endpoint is yet set up
533 Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
586 Left (TransportError ConnectNotFound _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
534 putMVar firstAttempt ()
587 putMVar firstAttempt ()
535
588
589+ -- The second attempt will fail because the server closes the socket before we can request a connection
536 takeMVar endpointCreated
590 takeMVar endpointCreated
537− Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
591+ Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
538
592
539− putMVar clientDone ()
593+ -- The third attempt succeeds
594+ Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
595+
596+ -- But a send will fail because the server has closed the connection again
597+ Left (TransportError SendClosed _) <- send conn1 ["ping"]
598+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint
540
599
541− takeMVar clientDone
600+ -- But a subsequent call to connect should reestablish the connection
601+ Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
542
602
603+ -- Send should now succeed
604+ Right () <- send conn2 ["ping"]
605+ return ()
543
606
607+ takeMVar serverDone
608+
609+-- Test what happens if we close the socket one way only. This means that the
610+-- 'recv' in 'handleIncomingMessages' will not fail, but a 'send' or 'connect'
611+-- *will* fail. We are testing that error handling everywhere does the right
612+-- thing.
613+testUnidirectionalError :: IO N.ServiceName -> IO ()
614+testUnidirectionalError nextPort = do
615+ clientDone <- newEmptyMVar
616+ serverPort <- nextPort
617+ serverGotPing <- newEmptyMVar
618+
619+ -- Server
620+ forkServer "127.0.0.1" serverPort 5 throw $ \sock -> do
621+ -- We accept connections, but when an exception occurs we don't do
622+ -- anything (in particular, we don't close the socket). This is important
623+ -- because when we shutdown one direction of the socket a recv here will
624+ -- fail, but we don't want to close that socket at that point (which
625+ -- would shutdown the socket in the other direction)
626+ void . (try :: IO () -> IO (Either SomeException ())) $ do
627+ 0 <- recvInt32 sock :: IO Int
628+ _ <- recvWithLength sock
629+ () <- sendMany sock [encodeInt32 ConnectionRequestAccepted]
630+
631+ RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)
632+ reqId <- recvInt32 sock :: IO Int
633+ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])
634+
635+ 10001 <- recvInt32 sock :: IO Int
636+ ["ping"] <- recvWithLength sock
637+ putMVar serverGotPing ()
638+
639+ -- Client
640+ forkTry $ do
641+ Right (transport, internals) <- nextPort >>= createTransportExposeInternals "127.0.0.1"
642+ Right endpoint <- newEndPoint transport
643+ let theirAddr = encodeEndPointAddress "127.0.0.1" serverPort 0
644+
645+ -- Establish a connection to the server
646+ Right conn1 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
647+ send conn1 ["ping"]
648+ takeMVar serverGotPing
649+
650+ -- Close the *outgoing* part of the socket only
651+ Right sock <- socketBetween internals (address endpoint) theirAddr
652+ N.shutdown sock N.ShutdownSend
653+
654+ -- At this point we cannot notice the problem yet so we shouldn't receive an event yet
655+ Nothing <- timeout 500000 $ receive endpoint
656+
657+ -- But when we send we find the error
658+ Left (TransportError SendFailed _) <- send conn1 ["ping"]
659+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint
660+
661+ -- A call to connect should now re-establish the connection
662+ Right conn2 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
663+ send conn2 ["ping"]
664+ takeMVar serverGotPing
665+
666+ -- Again, close the outgoing part of the socket
667+ Right sock' <- socketBetween internals (address endpoint) theirAddr
668+ N.shutdown sock' N.ShutdownSend
669+
670+ -- We now find the error when we attempt to close the connection
671+ Nothing <- timeout 500000 $ receive endpoint
672+ close conn2
673+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint
674+ Right conn3 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
675+ send conn3 ["ping"]
676+ takeMVar serverGotPing
677+
678+ -- We repeat once more.
679+ Right sock'' <- socketBetween internals (address endpoint) theirAddr
680+ N.shutdown sock'' N.ShutdownSend
681+
682+ -- Now we notice the problem when we try to connect
683+ Nothing <- timeout 500000 $ receive endpoint
684+ Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
685+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint
686+ Right conn4 <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
687+ send conn4 ["ping"]
688+ takeMVar serverGotPing
689+
690+ putMVar clientDone ()
691+
692+ takeMVar clientDone
544
693
545main :: IO ()
694main :: IO ()
546main = do
695main = do
@@ -558,5 +707,6 @@ main = do
558 , ("Many", testMany nextPort)
707 , ("Many", testMany nextPort)
559 , ("BreakTransport", testBreakTransport nextPort)
708 , ("BreakTransport", testBreakTransport nextPort)
560 , ("Reconnect", testReconnect nextPort)
709 , ("Reconnect", testReconnect nextPort)
710+ , ("UnidirectionalError", testUnidirectionalError nextPort)
561 ]
711 ]
562 testTransport (either (Left . show) (Right) <$> nextPort >>= createTransport "127.0.0.1")
712 testTransport (either (Left . show) (Right) <$> nextPort >>= createTransport "127.0.0.1")