git.iko.soy/distributed-process
on master help sign in
More consistent error handling
b9ab49aEdsko de Vries
network-transport-2/network-transport.cabal+4−4
compare with master
unifiedsplit
@@ -20,7 +20,7 @@ Library
20 Network.Transport.Chan,
20 Network.Transport.Chan,
21 Network.Transport.TCP,
21 Network.Transport.TCP,
22 Network.Transport.Util
22 Network.Transport.Util
23− extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables
23+ extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables, DeriveDataTypeable
24 ghc-options: -Wall -fno-warn-unused-do-bind
24 ghc-options: -Wall -fno-warn-unused-do-bind
25 HS-Source-Dirs: src
25 HS-Source-Dirs: src
26
26
@@ -36,7 +36,7 @@ Test-Suite TestTCP
36 mtl,
36 mtl,
37 transformers,
37 transformers,
38 ansi-terminal
38 ansi-terminal
39− extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables
39+ extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables, DeriveDataTypeable
40 ghc-options: -Wall -fno-warn-unused-do-bind
40 ghc-options: -Wall -fno-warn-unused-do-bind
41 HS-Source-Dirs: tests src
41 HS-Source-Dirs: tests src
42
42
@@ -50,7 +50,7 @@ Test-Suite TestMulticastInMemory
50 mtl,
50 mtl,
51 transformers,
51 transformers,
52 ansi-terminal
52 ansi-terminal
53− extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables
53+ extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables, DeriveDataTypeable
54 ghc-options: -Wall -fno-warn-unused-do-bind
54 ghc-options: -Wall -fno-warn-unused-do-bind
55 HS-Source-Dirs: tests src
55 HS-Source-Dirs: tests src
56
56
@@ -64,6 +64,6 @@ Test-Suite TestInMemory
64 mtl,
64 mtl,
65 transformers,
65 transformers,
66 ansi-terminal
66 ansi-terminal
67− extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables
67+ extensions: EmptyDataDecls, OverloadedStrings, ForeignFunctionInterface, FlexibleContexts, BangPatterns, ScopedTypeVariables, DeriveDataTypeable
68 ghc-options: -Wall -fno-warn-unused-do-bind
68 ghc-options: -Wall -fno-warn-unused-do-bind
69 HS-Source-Dirs: tests src
69 HS-Source-Dirs: tests src
network-transport-2/src/Network/Transport.hs+11−6
@@ -21,6 +21,8 @@ module Network.Transport ( -- * Types
21
21
22import Data.ByteString (ByteString)
22import Data.ByteString (ByteString)
23import Control.Monad.Error (Error(..))
23import Control.Monad.Error (Error(..))
24+import Control.Exception (Exception)
25+import Data.Typeable (Typeable)
24
26
25--------------------------------------------------------------------------------
27--------------------------------------------------------------------------------
26-- Main API --
28-- Main API --
@@ -114,39 +116,42 @@ newtype MulticastAddress = MulticastAddress { multicastAddressToByteString :: By
114--------------------------------------------------------------------------------
116--------------------------------------------------------------------------------
115
117
116data FailedWith error = FailedWith error String
118data FailedWith error = FailedWith error String
117− deriving Show
119+ deriving (Show, Typeable)
118
120
119instance Error (FailedWith error) where
121instance Error (FailedWith error) where
120 strMsg = FailedWith undefined
122 strMsg = FailedWith undefined
121
123
124+-- | For internal convenience, FailedWith is declared as an exception type
125+instance (Typeable err, Show err) => Exception (FailedWith err)
126+
122-- | Errors during the creation of an endpoint (currently, there are none)
127-- | Errors during the creation of an endpoint (currently, there are none)
123data NewEndPointErrorCode =
128data NewEndPointErrorCode =
124 NewEndPointTransportFailure -- ^ Transport-wide failure
129 NewEndPointTransportFailure -- ^ Transport-wide failure
125− deriving Show
130+ deriving (Show, Typeable)
126
131
127-- | Connection failure
132-- | Connection failure
128data ConnectErrorCode =
133data ConnectErrorCode =
129 ConnectInvalidAddress -- ^ Could not parse or resolve the address
134 ConnectInvalidAddress -- ^ Could not parse or resolve the address
130 | ConnectInsufficientResources -- ^ Insufficient resources (for instance, no more sockets available)
135 | ConnectInsufficientResources -- ^ Insufficient resources (for instance, no more sockets available)
131 | ConnectFailed -- ^ Failed for other reasons
136 | ConnectFailed -- ^ Failed for other reasons
132− deriving Show
137+ deriving (Show, Typeable)
133
138
134-- | Failure during the creation of a new multicast group
139-- | Failure during the creation of a new multicast group
135data NewMulticastGroupErrorCode =
140data NewMulticastGroupErrorCode =
136 NewMulticastGroupUnsupported
141 NewMulticastGroupUnsupported
137− deriving Show
142+ deriving (Show, Typeable)
138
143
139-- | Failure during the resolution of a multicast group
144-- | Failure during the resolution of a multicast group
140data ResolveMulticastGroupErrorCode =
145data ResolveMulticastGroupErrorCode =
141 ResolveMulticastGroupNotFound
146 ResolveMulticastGroupNotFound
142 | ResolveMulticastGroupUnsupported
147 | ResolveMulticastGroupUnsupported
143− deriving Show
148+ deriving (Show, Typeable)
144
149
145-- | Failure during sending a message
150-- | Failure during sending a message
146data SendErrorCode =
151data SendErrorCode =
147 SendConnectionClosed -- ^ Connection was closed manually or because of an error
152 SendConnectionClosed -- ^ Connection was closed manually or because of an error
148 | SendFailed -- ^ Send failed for some other reason
153 | SendFailed -- ^ Send failed for some other reason
149− deriving Show
154+ deriving (Show, Typeable)
150
155
151-- | Error codes used when reporting errors to endpoints (through receive)
156-- | Error codes used when reporting errors to endpoints (through receive)
152data ErrorEventInfo =
157data ErrorEventInfo =
network-transport-2/src/Network/Transport/Internal.hs+8−39
@@ -6,14 +6,11 @@ module Network.Transport.Internal ( -- * Encoders/decoders
6 , decodeInt16
6 , decodeInt16
7 , prependLength
7 , prependLength
8 -- * Miscellaneous abstractions
8 -- * Miscellaneous abstractions
9− , failWith
9+ , mapExceptionIO
10− , failWithIO
11 , tryIO
10 , tryIO
12− , whileM
13 , tryToEnum
11 , tryToEnum
14 -- * Debugging
12 -- * Debugging
15 , tlog
13 , tlog
16− , loopInvariant
17 ) where
14 ) where
18
15
19import Prelude hiding (catch)
16import Prelude hiding (catch)
@@ -23,11 +20,8 @@ import Foreign.ForeignPtr (withForeignPtr)
23import Data.ByteString (ByteString)
20import Data.ByteString (ByteString)
24import qualified Data.ByteString as BS (length)
21import qualified Data.ByteString as BS (length)
25import qualified Data.ByteString.Internal as BSI (unsafeCreate, toForeignPtr, inlinePerformIO)
22import qualified Data.ByteString.Internal as BSI (unsafeCreate, toForeignPtr, inlinePerformIO)
26−import Control.Applicative ((<$>))
27−import Control.Monad (MonadPlus, guard, when)
28import Control.Monad.IO.Class (MonadIO, liftIO)
23import Control.Monad.IO.Class (MonadIO, liftIO)
29−import Control.Monad.Error (MonadError, throwError)
30−import Control.Exception (IOException, catch, try)
24+import Control.Exception (IOException, Exception, catch, try, throw)
31--import Control.Concurrent (myThreadId)
25--import Control.Concurrent (myThreadId)
32
26
33foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
27foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
@@ -69,26 +63,13 @@ decodeInt16 bs = Just . BSI.inlinePerformIO $ do
69prependLength :: [ByteString] -> [ByteString]
63prependLength :: [ByteString] -> [ByteString]
70prependLength bss = encodeInt32 (sum . map BS.length $ bss) : bss
64prependLength bss = encodeInt32 (sum . map BS.length $ bss) : bss
71
65
72−-- | Convert a "failing operation" into one that fails with a specific error
66+-- | Translate exceptions that arise in IO computations
73−failWith :: (MonadError e m) => e -> Maybe a -> m a
67+mapExceptionIO :: (Exception e1, Exception e2) => (e1 -> e2) -> IO a -> IO a
74−failWith err Nothing = throwError err
68+mapExceptionIO f p = catch p (throw . f)
75−failWith _ (Just x) = return x
76
69
77−-- | Try the specific I/O operation, and fail with the given error
70+-- | Like 'try', but lifted and specialized to IOExceptions
78−-- when an exception occurs
71+tryIO :: MonadIO m => IO a -> m (Either IOException a)
79−failWithIO :: (MonadIO m, MonadError e m) => (IOException -> e) -> IO a -> m a
72+tryIO = liftIO . try
80−failWithIO f io = do
81− ma <- liftIO $ catch (Right <$> io) handleIOException
82− case ma of
83− Left err -> throwError (f err)
84− Right a -> return a
85− where
86− handleIOException :: IOException -> IO (Either IOException a)
87− handleIOException = return . Left
88−
89−-- | Like 'try', but specialized to IOExceptions
90−tryIO :: IO a -> IO (Either IOException a)
91−tryIO = try
92
73
93-- | Logging (for debugging)
74-- | Logging (for debugging)
94tlog :: MonadIO m => String -> m ()
75tlog :: MonadIO m => String -> m ()
@@ -102,15 +83,3 @@ tlog msg = liftIO $ do
102-- | Safe version of 'toEnum'
83-- | Safe version of 'toEnum'
103tryToEnum :: forall a. (Enum a, Bounded a) => Int -> Maybe a
84tryToEnum :: forall a. (Enum a, Bounded a) => Int -> Maybe a
104tryToEnum n = if fromEnum (minBound :: a) <= n && n <= fromEnum (maxBound :: a) then Just (toEnum n) else Nothing
85tryToEnum n = if fromEnum (minBound :: a) <= n && n <= fromEnum (maxBound :: a) then Just (toEnum n) else Nothing
105−
106−-- | Loop invariant (like 'guard')
107−loopInvariant :: MonadPlus m => m Bool -> m ()
108−loopInvariant = (>>= guard)
109−
110−-- | While-loop
111−whileM :: Monad m => m Bool -> m a -> m ()
112−whileM cond body = go
113− where
114− go = do
115− b <- cond
116− when b $ body >> go
network-transport-2/src/Network/Transport/Internal/TCP.hs+13−9
@@ -22,11 +22,12 @@ import qualified Network.Socket as N ( HostName
22 , defaultProtocol
22 , defaultProtocol
23 , setSocketOption
23 , setSocketOption
24 , accept
24 , accept
25+ , sClose
25 )
26 )
26import qualified Network.Socket.ByteString as NBS (recv)
27import qualified Network.Socket.ByteString as NBS (recv)
27import Control.Concurrent (forkIO, ThreadId)
28import Control.Concurrent (forkIO, ThreadId)
28−import Control.Monad (mzero, MonadPlus, liftM, forever)
29+import Control.Monad (liftM, forever)
29−import Control.Exception (SomeException, handle)
30+import Control.Exception (SomeException, handle, bracketOnError, throw)
30import Data.ByteString (ByteString)
31import Data.ByteString (ByteString)
31import qualified Data.ByteString as BS (length, concat, null)
32import qualified Data.ByteString as BS (length, concat, null)
32import Data.Int (Int32)
33import Data.Int (Int32)
@@ -45,7 +46,9 @@ import Data.Int (Int32)
45-- relevant exceptions in the request handler.
46-- relevant exceptions in the request handler.
46--
47--
47-- The request handler should spawn threads to handle each individual request
48-- The request handler should spawn threads to handle each individual request
48−-- or the server will block.
49+-- or the server will block. Once a thread has been spawned it will be the
50+-- responsibility of the new thread to close the socket when an exception
51+-- occurs.
49forkServer :: N.HostName -- ^ Host
52forkServer :: N.HostName -- ^ Host
50 -> N.ServiceName -- ^ Port
53 -> N.ServiceName -- ^ Port
51 -> Int -- ^ Backlog (maximum number of queued connections)
54 -> Int -- ^ Backlog (maximum number of queued connections)
@@ -61,9 +64,10 @@ forkServer host port backlog terminationHandler requestHandler = do
61 N.setSocketOption sock N.ReuseAddr 1
64 N.setSocketOption sock N.ReuseAddr 1
62 N.bindSocket sock (N.addrAddress addr)
65 N.bindSocket sock (N.addrAddress addr)
63 N.listen sock backlog
66 N.listen sock backlog
64− forkIO . handle terminationHandler . forever $ do
67+ forkIO . handle terminationHandler . forever $
65− (clientSock, _) <- N.accept sock
68+ bracketOnError (N.accept sock)
69+ (N.sClose . fst)
66− requestHandler clientSock
70+ (requestHandler . fst)
67
71
68-- | Read a length and then a payload of that length
72-- | Read a length and then a payload of that length
69recvWithLength :: N.Socket -> IO [ByteString]
73recvWithLength :: N.Socket -> IO [ByteString]
@@ -74,7 +78,7 @@ recvInt32 :: Num a => N.Socket -> IO a
74recvInt32 sock = do
78recvInt32 sock = do
75 mi <- liftM (decodeInt32 . BS.concat) $ recvExact sock 4
79 mi <- liftM (decodeInt32 . BS.concat) $ recvExact sock 4
76 case mi of
80 case mi of
77− Nothing -> mzero
81+ Nothing -> throw (userError "Invalid integer")
78 Just i -> return i
82 Just i -> return i
79
83
80-- | Read an exact number of bytes from a socket
84-- | Read an exact number of bytes from a socket
@@ -84,7 +88,7 @@ recvInt32 sock = do
84recvExact :: N.Socket -- ^ Socket to read from
88recvExact :: N.Socket -- ^ Socket to read from
85 -> Int32 -- ^ Number of bytes to read
89 -> Int32 -- ^ Number of bytes to read
86 -> IO [ByteString]
90 -> IO [ByteString]
87−recvExact _ len | len <= 0 = mzero
91+recvExact _ len | len <= 0 = throw (userError "Negative length")
88recvExact sock len = go [] len
92recvExact sock len = go [] len
89 where
93 where
90 go :: [ByteString] -> Int32 -> IO [ByteString]
94 go :: [ByteString] -> Int32 -> IO [ByteString]
@@ -92,5 +96,5 @@ recvExact sock len = go [] len
92 go acc l = do
96 go acc l = do
93 bs <- NBS.recv sock (fromIntegral l `min` 4096)
97 bs <- NBS.recv sock (fromIntegral l `min` 4096)
94 if BS.null bs
98 if BS.null bs
95− then fail "Socket closed"
99+ then throw (userError "Socket closed")
96 else go (bs : acc) (l - fromIntegral (BS.length bs))
100 else go (bs : acc) (l - fromIntegral (BS.length bs))
network-transport-2/src/Network/Transport/TCP.hs+463−303
@@ -6,19 +6,28 @@
6-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
6-- @192.168.0.1:8080@ and /B/ subsequently connects tries to connect to /A/ as
7-- @client1.local:http-alt@ then the transport layer will not realize that the
7-- @client1.local:http-alt@ then the transport layer will not realize that the
8-- TCP connection can be reused.
8-- 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
9module Network.Transport.TCP ( -- * Main API
16module Network.Transport.TCP ( -- * Main API
10 createTransport
17 createTransport
11− , -- * TCP specific functionality
18+ , -- * TCP specific functionality (exported mostly for testing purposes_
12 EndPointId
19 EndPointId
13 , encodeEndPointAddress
20 , encodeEndPointAddress
14 , decodeEndPointAddress
21 , decodeEndPointAddress
15 , ControlHeader(..)
22 , ControlHeader(..)
16 , ConnectionRequestResponse(..)
23 , ConnectionRequestResponse(..)
17 , firstNonReservedConnectionId
24 , firstNonReservedConnectionId
25+ , socketToEndPoint
18 -- * Design notes
26 -- * Design notes
19 -- $design
27 -- $design
20 ) where
28 ) where
21
29
30+import Prelude hiding (catch)
22import Network.Transport
31import Network.Transport
23import Network.Transport.Internal.TCP ( forkServer
32import Network.Transport.Internal.TCP ( forkServer
24 , recvWithLength
33 , recvWithLength
@@ -27,8 +36,7 @@ import Network.Transport.Internal.TCP ( forkServer
27import Network.Transport.Internal ( encodeInt32
36import Network.Transport.Internal ( encodeInt32
28 , decodeInt32
37 , decodeInt32
29 , prependLength
38 , prependLength
30− , failWith
39+ , mapExceptionIO
31− , failWithIO
32 , tryIO
40 , tryIO
33 , tryToEnum
41 , tryToEnum
34 )
42 )
@@ -45,6 +53,8 @@ import qualified Network.Socket as N ( HostName
45 , setSocketOption
53 , setSocketOption
46 , SocketOption(ReuseAddr)
54 , SocketOption(ReuseAddr)
47 , connect
55 , connect
56+ , sOMAXCONN
57+ , AddrInfo
48 )
58 )
49import Network.Socket.ByteString (sendMany)
59import Network.Socket.ByteString (sendMany)
50import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
60import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
@@ -62,9 +72,7 @@ import Control.Concurrent.MVar ( MVar
62import Control.Category ((>>>))
72import Control.Category ((>>>))
63import Control.Applicative ((<*>), (*>), (<$>))
73import Control.Applicative ((<*>), (*>), (<$>))
64import Control.Monad (forM, forM_, void, when, unless)
74import 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)
68import Data.IORef (IORef, newIORef, writeIORef, readIORef)
76import Data.IORef (IORef, newIORef, writeIORef, readIORef)
69import Data.ByteString (ByteString)
77import Data.ByteString (ByteString)
70import qualified Data.ByteString as BS (concat)
78import qualified Data.ByteString as BS (concat)
@@ -174,29 +182,50 @@ import System.IO (hPutStrLn, stderr)
174-- Internal datatypes --
182-- Internal datatypes --
175--------------------------------------------------------------------------------
183--------------------------------------------------------------------------------
176
184
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
177−data TCPTransport = TCPTransport { transportHost :: N.HostName
199+ { transportHost :: N.HostName
178− , transportPort :: N.ServiceName
200+ , transportPort :: N.ServiceName
179− , transportState :: MVar TransportState
201+ , transportState :: MVar TransportState
180− }
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
181−data LocalEndPoint = LocalEndPoint { localAddress :: EndPointAddress
213+ { localAddress :: EndPointAddress
182− , localChannel :: Chan Event
214+ , localChannel :: Chan Event
183− , localState :: MVar LocalState
215+ , localState :: MVar LocalEndPointState
184− }
216+ }
185−data RemoteEndPoint = RemoteEndPoint { remoteAddress :: EndPointAddress
217+
186− , remoteState :: MVar RemoteState
218+data LocalEndPointState =
187− }
219+ LocalEndPointValid ValidLocalEndPointState
188−
220+ | LocalEndPointClosed
189−-- We use underscores for fields that we might update (using lenses).
221+
190−
222+data ValidLocalEndPointState = ValidLocalEndPointState
191−data TransportState = TransportState { _localEndPoints :: Map EndPointAddress LocalEndPoint
192− , _acceptNewConnections :: Bool
193− }
194−data LocalState = LocalState { _nextConnectionId :: ConnectionId
223+ { _nextConnectionId :: ConnectionId
195− , _pendingCtrlRequests :: IntMap (MVar [ByteString])
224+ , _pendingCtrlRequests :: IntMap (MVar [ByteString])
196− , _nextCtrlRequestId :: ControlRequestId
225+ , _nextCtrlRequestId :: ControlRequestId
197− , _localConnections :: Map EndPointAddress RemoteEndPoint
226+ , _localConnections :: Map EndPointAddress RemoteEndPoint
198− , _internalThreads :: [ThreadId]
227+ , _internalThreads :: [ThreadId]
199− }
228+ }
200
229
201-- A remote endpoint has incoming and outgoing connections, and when the total
230-- A remote endpoint has incoming and outgoing connections, and when the total
202-- number of connections (that is, the 'remoteRefCount') drops to zero we want
231-- number of connections (that is, the 'remoteRefCount') drops to zero we want
@@ -243,15 +272,23 @@ data LocalState = LocalState { _nextConnectionId :: ConnectionId
243-- and then looks up the endpoint in localConnections it is guaranteed to
272-- and then looks up the endpoint in localConnections it is guaranteed to
244-- either find a different remote endpoint, or else none at all.
273-- either find a different remote endpoint, or else none at all.
245
274
275+data RemoteEndPoint = RemoteEndPoint
276+ { remoteAddress :: EndPointAddress
277+ , remoteState :: MVar RemoteState
278+ }
279+
280+data RemoteState =
246−data RemoteState = RemoteEndPointInvalid (FailedWith ConnectErrorCode)
281+ RemoteEndPointInvalid (FailedWith ConnectErrorCode)
247− | RemoteEndPointValid RemoteConnection
282+ | RemoteEndPointValid ValidRemoteEndPointState
248− | RemoteEndPointClosing (MVar ()) RemoteConnection
283+ | RemoteEndPointClosing (MVar ()) ValidRemoteEndPointState
249− | RemoteEndPointClosed
284+ | RemoteEndPointClosed
285+
286+data ValidRemoteEndPointState = ValidRemoteEndPointState
250−data RemoteConnection = RemoteConnection { _remoteOutgoing :: !Int
287+ { _remoteOutgoing :: !Int
251− , _remoteIncoming :: IntSet
288+ , _remoteIncoming :: IntSet
252− , remoteSocket :: N.Socket
289+ , remoteSocket :: N.Socket
253− , sendOn :: [ByteString] -> IO ()
290+ , sendOn :: [ByteString] -> IO ()
254− }
291+ }
255
292
256type EndPointId = Int32
293type EndPointId = Int32
257type ControlRequestId = Int32
294type ControlRequestId = Int32
@@ -261,7 +298,7 @@ type EndPointPair = (LocalEndPoint, RemoteEndPoint)
261data ControlHeader =
298data ControlHeader =
262 RequestConnectionId -- ^ Request a new connection ID from the remote endpoint
299 RequestConnectionId -- ^ Request a new connection ID from the remote endpoint
263 | CloseConnection -- ^ Tell the remote endpoint we will no longer be using a connection
300 | 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
265 | CloseSocket -- ^ Request to close the connection (see module description)
302 | CloseSocket -- ^ Request to close the connection (see module description)
266 deriving (Enum, Bounded, Show)
303 deriving (Enum, Bounded, Show)
267
304
@@ -281,23 +318,24 @@ data ConnectionRequestResponse =
281-- TODOs: deal with hints
318-- TODOs: deal with hints
282createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)
319createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)
283createTransport host port = do
320createTransport host port = do
284− state <- newMVar TransportState { _localEndPoints = Map.empty
321+ state <- newMVar . TransportValid $ ValidTransportState { _localEndPoints = Map.empty }
285− , _acceptNewConnections = True
286− }
287− let transport = TCPTransport { transportState = state
322+ let transport = TCPTransport { transportState = state
288− , transportHost = host
323+ , transportHost = host
289− , transportPort = port
324+ , transportPort = port
290− }
325+ }
291 tryIO $ do
326 tryIO $ do
292− -- TODO: "5" is listed as "system dependent maximum backlog"?
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
293− transportThread <- forkServer host port 5 (terminationHandler transport) (handleConnectionRequest transport)
330+ transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
294 return Transport { newEndPoint = apiNewEndPoint transport
331 return Transport { newEndPoint = apiNewEndPoint transport
295 , closeTransport = apiCloseTransport transportThread
332 , closeTransport = apiCloseTransport transportThread
296 }
333 }
297 where
334 where
298 terminationHandler :: TCPTransport -> SomeException -> IO ()
335 terminationHandler :: TCPTransport -> SomeException -> IO ()
299− terminationHandler transport _ =
336+ terminationHandler _ _ =
300− modifyMVar_ (transportState transport) $ return . (acceptNewConnections ^= False)
337+ -- TODO: implement
338+ return ()
301
339
302-- | Shut down the transport
340-- | Shut down the transport
303apiCloseTransport :: ThreadId -> IO ()
341apiCloseTransport :: ThreadId -> IO ()
@@ -313,15 +351,16 @@ apiCloseTransport transportThread = do
313
351
314-- | Create a new endpoint
352-- | Create a new endpoint
315apiNewEndPoint :: TCPTransport -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
353apiNewEndPoint :: TCPTransport -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
316−apiNewEndPoint transport = do
354+apiNewEndPoint transport = try $ do
317 ourEndPoint <- createLocalEndPoint transport
355 ourEndPoint <- createLocalEndPoint transport
356+ return EndPoint
318− return . Right $ EndPoint { receive = readChan (localChannel ourEndPoint)
357+ { receive = readChan (localChannel ourEndPoint)
319− , address = localAddress ourEndPoint
358+ , address = localAddress ourEndPoint
320− , connect = apiConnect ourEndPoint
359+ , connect = apiConnect ourEndPoint
321− , closeEndPoint = apiCloseEndPoint transport ourEndPoint
360+ , closeEndPoint = apiCloseEndPoint transport ourEndPoint
322− , newMulticastGroup = return . Left $ newMulticastGroupError
361+ , newMulticastGroup = return . Left $ newMulticastGroupError
323− , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
362+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
324− }
363+ }
325 where
364 where
326 newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
365 newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"
327 resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
366 resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"
@@ -333,19 +372,18 @@ apiConnect :: LocalEndPoint -- ^ Local end point
333 -> IO (Either (FailedWith ConnectErrorCode) Connection)
372 -> IO (Either (FailedWith ConnectErrorCode) Connection)
334apiConnect ourEndPoint theirAddress _ | localAddress ourEndPoint == theirAddress =
373apiConnect ourEndPoint theirAddress _ | localAddress ourEndPoint == theirAddress =
335 connectToSelf ourEndPoint
374 connectToSelf ourEndPoint
336−apiConnect ourEndPoint theirAddress _ = runErrorT $ do
375+apiConnect ourEndPoint theirAddress _ = try $ do
337 (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress
376 (theirEndPoint, connId) <- requestConnectionTo ourEndPoint theirAddress
338 -- connAlive can be an IORef rather than an MVar because it is protected by
377 -- connAlive can be an IORef rather than an MVar because it is protected by
339 -- the remoteState MVar. We don't need the overhead of locking twice.
378 -- the remoteState MVar. We don't need the overhead of locking twice.
340− connAlive <- liftIO $ newIORef True
379+ connAlive <- newIORef True
341 return $ Connection { send = apiSend theirEndPoint connId connAlive
380 return $ Connection { send = apiSend theirEndPoint connId connAlive
342 , close = apiClose theirEndPoint connId connAlive
381 , close = apiClose theirEndPoint connId connAlive
343 }
382 }
344
383
345-- | Close a connection
384-- | Close a connection
346--
385--
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'
349-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
387-- GUARANTEE: If the connection is alive on entry then the remote endpoint will
350-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state
388-- either be RemoteEndPointValid or RemoteEndPointClosing. Otherwise, the state
351-- of the remote endpoint will not be changed.
389-- of the remote endpoint will not be changed.
@@ -353,15 +391,22 @@ apiConnect ourEndPoint theirAddress _ = runErrorT $ do
353-- TODO: We ignore errors during a close. Is that right?
391-- TODO: We ignore errors during a close. Is that right?
354apiClose :: RemoteEndPoint -> ConnectionId -> IORef Bool -> IO ()
392apiClose :: RemoteEndPoint -> ConnectionId -> IORef Bool -> IO ()
355apiClose theirEndPoint connId connAlive = void . tryIO $ do
393apiClose theirEndPoint connId connAlive = void . tryIO $ do
356− modifyMVar_ (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) -> do
394+ modifyMVar_ (remoteState theirEndPoint) $ \st -> case st of
395+ RemoteEndPointValid vst -> do
357− alive <- readIORef connAlive
396+ alive <- readIORef connAlive
358− if alive
397+ if alive
359− then do
398+ then do
399+ writeIORef connAlive False
360− sendOn remoteConn [encodeInt32 CloseConnection, encodeInt32 connId]
400+ sendOn vst [encodeInt32 CloseConnection, encodeInt32 connId]
361− writeIORef connAlive False
362− return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ remoteConn)
401+ return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ vst)
363− else
402+ else
364− return (RemoteEndPointValid remoteConn)
403+ return (RemoteEndPointValid vst)
404+ RemoteEndPointClosed -> do
405+ return st
406+ RemoteEndPointClosing _ _ ->
407+ fail "apiClose RELY violation"
408+ RemoteEndPointInvalid _ ->
409+ fail "apiClose RELY violation"
365 closeIfUnused theirEndPoint
410 closeIfUnused theirEndPoint
366
411
367-- | Send data across a connection
412-- | Send data across a connection
@@ -370,38 +415,46 @@ apiClose theirEndPoint connId connAlive = void . tryIO $ do
370-- GUARANTEE: The state of the remote endpoint will not be changed.
415-- GUARANTEE: The state of the remote endpoint will not be changed.
371apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
416apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
372apiSend theirEndPoint connId connAlive payload = do
417apiSend theirEndPoint connId connAlive payload = do
373− withMVar (remoteState theirEndPoint) $ \st -> case st of
418+ withMVar (remoteState theirEndPoint) $ \st -> case st of
374− RemoteEndPointValid remoteConn -> do
419+ RemoteEndPointValid vst -> do
375− alive <- readIORef connAlive
420+ alive <- readIORef connAlive
376− runErrorT $ if alive
421+ try $ if alive
377− then
378− failWithIO (FailedWith SendFailed . show) $
379− sendOn remoteConn (encodeInt32 connId : prependLength payload)
422+ then mapExceptionIO sendFailed $ sendOn vst (encodeInt32 connId : prependLength payload)
380− else
381− throwError $ FailedWith SendConnectionClosed "Connection closed"
423+ else throw $ FailedWith SendConnectionClosed "Connection closed"
382− RemoteEndPointClosed -> do
424+ RemoteEndPointClosed -> do
383− return (Left $ FailedWith SendConnectionClosed "Endpoint closed")
425+ return (Left $ FailedWith SendConnectionClosed "Endpoint closed")
384− RemoteEndPointClosing _ _ ->
426+ RemoteEndPointClosing _ _ ->
385− error "apiSend RELY violation"
427+ error "apiSend RELY violation"
386− RemoteEndPointInvalid _ ->
428+ RemoteEndPointInvalid _ ->
387− error "apiSend RELY violation"
429+ error "apiSend RELY violation"
430+ where
431+ sendFailed :: IOException -> FailedWith SendErrorCode
432+ sendFailed = FailedWith SendFailed . show
388
433
389-- | Force-close the endpoint
434-- | Force-close the endpoint
390apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
435apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
391apiCloseEndPoint transport ourEndPoint = do
436apiCloseEndPoint transport ourEndPoint = do
392 -- Remove the reference from the transport state
437 -- Remove the reference from the transport state
393− modifyMVar_ (transportState transport) $ return . (localEndPointAt (localAddress ourEndPoint) ^= Nothing)
438+ removeLocalEndPoint transport ourEndPoint
394− -- We take the endpoint state and don't put it back
439+ -- Close the local endpoint
395− -- (this may block some threads, but we don't care: we are killing them anyway)
396− ourState <- takeMVar (localState ourEndPoint)
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
397− -- Close all endpoints and kill all threads
450+ -- Close all endpoints and kill all threads
398− incoming <- IntSet.elems . IntSet.unions <$> (forM (Map.elems $ ourState ^. localConnections) $ tryCloseRemoteSocket)
451+ incoming <- IntSet.elems . IntSet.unions <$> (forM (Map.elems $ vst ^. localConnections) $ tryCloseRemoteSocket)
399− forM_ (ourState ^. internalThreads) killThread
452+ forM_ (vst ^. internalThreads) killThread
400− -- We send a single message to the endpoint that it is closed. Subsequent
453+ -- 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
454+ -- 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
455+ -- receive return an error, but this would mean checking for some state on
403− -- every call to receive, which is an unnecessary overhead.
456+ -- every call to receive, which is an unnecessary overhead.
404− writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)
457+ writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)
405 where
458 where
406 -- Close the remote socket and return the set of all incoming connections
459 -- Close the remote socket and return the set of all incoming connections
407 tryCloseRemoteSocket :: RemoteEndPoint -> IO IntSet
460 tryCloseRemoteSocket :: RemoteEndPoint -> IO IntSet
@@ -412,9 +465,11 @@ apiCloseEndPoint transport ourEndPoint = do
412 RemoteEndPointInvalid _ ->
465 RemoteEndPointInvalid _ ->
413 return (st, IntSet.empty)
466 return (st, IntSet.empty)
414 RemoteEndPointValid conn -> do
467 RemoteEndPointValid conn -> do
415− tryIO $ do
468+ -- Try to send a CloseSocket request
416− sendOn conn [encodeInt32 CloseSocket]
469+ tryIO $ sendOn conn [encodeInt32 CloseSocket]
470+ -- .. but even if it fails, close the socket anyway
471+ -- (hence, two separate calls to tryIO)
417− N.sClose (remoteSocket conn)
472+ tryIO $ N.sClose (remoteSocket conn)
418 return (RemoteEndPointClosed, conn ^. remoteIncoming)
473 return (RemoteEndPointClosed, conn ^. remoteIncoming)
419 RemoteEndPointClosing _ conn -> do
474 RemoteEndPointClosing _ conn -> do
420 tryIO $ N.sClose (remoteSocket conn)
475 tryIO $ N.sClose (remoteSocket conn)
@@ -427,6 +482,7 @@ connectToSelf :: LocalEndPoint -> IO (Either (FailedWith ConnectErrorCode) Conne
427connectToSelf ourEndPoint = do
482connectToSelf ourEndPoint = do
428 -- Here connAlive must an MVar because it is not protected by another lock
483 -- Here connAlive must an MVar because it is not protected by another lock
429 connAlive <- newMVar True
484 connAlive <- newMVar True
485+ -- TODO: catch exception
430 connId <- getNextConnectionId ourEndPoint
486 connId <- getNextConnectionId ourEndPoint
431 writeChan ourChan (ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint))
487 writeChan ourChan (ConnectionOpened connId ReliableOrdered (localAddress ourEndPoint))
432 return . Right $ Connection { send = selfSend connAlive connId
488 return . Right $ Connection { send = selfSend connAlive connId
@@ -457,176 +513,218 @@ connectToSelf ourEndPoint = do
457--------------------------------------------------------------------------------
513--------------------------------------------------------------------------------
458
514
459-- | Create a new local endpoint
515-- | Create a new local endpoint
516+--
517+-- May throw a FailedWith NewEndPointErrorCode exception if the transport is closed.
460createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
518createLocalEndPoint :: TCPTransport -> IO LocalEndPoint
461createLocalEndPoint transport = do
519createLocalEndPoint transport = do
462− chan <- newChan
520+ chan <- newChan
521+ state <- newMVar . LocalEndPointValid $ ValidLocalEndPointState
463− state <- newMVar LocalState { _nextConnectionId = firstNonReservedConnectionId
522+ { _nextConnectionId = firstNonReservedConnectionId
464− , _pendingCtrlRequests = IntMap.empty
523+ , _pendingCtrlRequests = IntMap.empty
465− , _nextCtrlRequestId = 0
524+ , _nextCtrlRequestId = 0
466− , _localConnections = Map.empty
525+ , _localConnections = Map.empty
467− , _internalThreads = []
526+ , _internalThreads = []
468− }
527+ }
469− modifyMVar (transportState transport) $ \st -> do
528+ modifyMVar (transportState transport) $ \st -> case st of
529+ TransportValid vst -> do
470− let ix = fromIntegral $ Map.size (st ^. localEndPoints)
530+ let ix = fromIntegral $ Map.size (vst ^. localEndPoints)
471− let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix
531+ let addr = encodeEndPointAddress (transportHost transport) (transportPort transport) ix
472− let localEndPoint = LocalEndPoint { localAddress = addr
532+ let localEndPoint = LocalEndPoint { localAddress = addr
473− , localChannel = chan
533+ , localChannel = chan
474− , localState = state
534+ , localState = state
475− }
535+ }
476− return ((localEndPointAt addr ^= Just localEndPoint) $ st, localEndPoint)
536+ return (TransportValid . (localEndPointAt addr ^= Just localEndPoint) $ vst, localEndPoint)
537+ TransportClosed ->
538+ throw (FailedWith NewEndPointTransportFailure "Transport closed")
477
539
478-- | Request a connection to a remote endpoint
540-- | Request a connection to a remote endpoint
479--
541--
480-- This will block until we get a connection ID from the remote endpoint; if
542-- This will block until we get a connection ID from the remote endpoint; if
481-- the remote endpoint was in 'RemoteEndPointClosing' state then we will
543-- the remote endpoint was in 'RemoteEndPointClosing' state then we will
482-- additionally block until that is resolved.
544-- additionally block until that is resolved.
545+--
546+-- May throw a FailedWith ConnectErrorCode exception.
483requestConnectionTo :: LocalEndPoint
547requestConnectionTo :: LocalEndPoint
484 -> EndPointAddress
548 -> EndPointAddress
485− -> ErrorT (FailedWith ConnectErrorCode) IO (RemoteEndPoint, ConnectionId)
549+ -> IO (RemoteEndPoint, ConnectionId)
486−requestConnectionTo ourEndPoint theirAddress = ErrorT go
550+requestConnectionTo ourEndPoint theirAddress = go
487 where
551 where
488 go = do
552 go = do
489− let ourState = localState ourEndPoint
553+ -- Find the remote endpoint (create it if it doesn't yet exist)
490− (theirEndPoint, isNew) <- modifyMVar ourState $ \st ->
554+ theirEndPoint <- findTheirEndPoint
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.
512 let theirState = remoteState theirEndPoint
555 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.
513 endPointStateSnapshot <- modifyMVar theirState $ \st ->
561 endPointStateSnapshot <- modifyMVar theirState $ \st ->
514 case st of
562 case st of
515 RemoteEndPointValid ep ->
563 RemoteEndPointValid ep ->
516 return (RemoteEndPointValid . (remoteOutgoing ^+= 1) $ ep, st)
564 return (RemoteEndPointValid . (remoteOutgoing ^+= 1) $ ep, st)
517 _ ->
565 _ ->
518 return (st, st)
566 return (st, st)
519−
567+
520− -- From this point on we are guaranteed the refcount is positive,
568+ -- From this point on we are guaranteed the refcount is positive, provided
521− -- provided that the endpoint was valid. We still need to deal with the
569+ -- that the endpoint was valid. We still need to deal with the case where
522− -- case where it was not valid, however, which we didn't want to do that
570+ -- it was not valid, however, which we didn't want to do while holding the
523− -- while holding the endpoint lock.
571+ -- endpoint lock.
572+ --
524− -- Although 'endPointStateSnapshot' here refers to a snapshot of the endpoint
573+ -- Although 'endPointStateSnapshot' here refers to a snapshot of the
525− -- state, and might have changed in the meantime, these changes won't
574+ -- endpoint state, and might have changed in the meantime, these changes
526− -- matter.
575+ -- won't matter.
527 case endPointStateSnapshot of
576 case endPointStateSnapshot of
528 RemoteEndPointInvalid err -> do
577 RemoteEndPointInvalid err -> do
529− return . Left $ err
578+ throw err
530−
579+
531− RemoteEndPointClosing resolved _ -> do
580+ RemoteEndPointClosing resolved _ ->
532 -- If the remote endpoint is closing, then we need to block until
581 -- If the remote endpoint is closing, then we need to block until
533 -- this is resolved and we then try again
582 -- this is resolved and we then try again
534− readMVar resolved
583+ readMVar resolved >> go
535− go
584+
536−
537 RemoteEndPointClosed -> do
585 RemoteEndPointClosed -> do
538 -- EndPointClosed indicates that a concurrent thread was in the
586 -- EndPointClosed indicates that a concurrent thread was in the
539 -- process of closing the TCP connection to the remote endpoint when
587 -- process of closing the TCP connection to the remote endpoint when
540 -- we obtained a reference to it. The remote endpoint will now have
588 -- we obtained a reference to it. The remote endpoint will now have
541 -- been removed from ourState, so we simply try again.
589 -- been removed from ourState, so we simply try again.
542 go
590 go
543−
591+
544 RemoteEndPointValid _ -> do
592 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−
552 -- On a failure we decrement the refcount again and return an error.
593 -- On a failure we decrement the refcount again and return an error.
553 -- The only state the remote endpoint can be in at this point is
594 -- The only state the remote endpoint can be in at this point is
554 -- valid. As mentioned above, we can rely on the endpoint being in
595 -- valid. As mentioned above, we can rely on the endpoint being in
555 -- valid state at this point.
596 -- valid state at this point.
556− let failed err = do
597+ let failureHandler :: IOException -> IO b
598+ failureHandler err = do
557− modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
599+ modifyMVar_ theirState $ \(RemoteEndPointValid ep) ->
558− return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ ep)
600+ return (RemoteEndPointValid . (remoteOutgoing ^+= (-1)) $ ep)
559− -- TODO: should we call closeIfUnused here?
601+ -- TODO: should we call closeIfUnused here?
560− return . Left $ FailedWith ConnectFailed err
602+ throw $ FailedWith ConnectFailed (show err)
561−
603+
562− case mcid of
604+ -- Do the actual connection request. This blocks until the remote
563− Left err -> failed (show err)
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
564− Right Nothing -> failed "Invalid integer"
609+ Nothing -> failureHandler (userError "Invalid integer")
565− Right (Just cid) -> return . Right $ (theirEndPoint, cid)
610+ Just cid -> return (theirEndPoint, cid)
566
611
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+
567−-- | Establish a connection to a remote endpoint
642+-- | Set up a remote endpoint
568--
643--
569-- RELY: The state of the remote endpoint must be uninitialized.
644-- RELY: The state of the remote endpoint must be uninitialized.
570-- GUARANTEE: Will only change the state to RemoteEndPointValid or
645-- GUARANTEE: Will only change the state to RemoteEndPointValid or
571-- RemoteEndPointInvalid.
646-- RemoteEndPointInvalid.
572−connectToRemoteEndPoint :: EndPointPair -> IO ()
647+setupRemoteEndPoint :: EndPointPair -> IO ()
573−connectToRemoteEndPoint (ourEndPoint, theirEndPoint) = do
648+setupRemoteEndPoint (ourEndPoint, theirEndPoint) = do
574− let ourState = localState ourEndPoint
649+ result <- socketToEndPoint (localAddress ourEndPoint) (remoteAddress theirEndPoint)
575− theirState = remoteState theirEndPoint
576− theirAddress = remoteAddress theirEndPoint
577− result <- runErrorT go
578 case result of
650 case result of
579− Right (ConnectionRequestAccepted, sock) -> do
651+ Right (sock, ConnectionRequestAccepted) -> do
652+ let vst = ValidRemoteEndPointState
580− let remoteConn = RemoteConnection { remoteSocket = sock
653+ { remoteSocket = sock
581− , _remoteOutgoing = 0
654+ , _remoteOutgoing = 0
582− , _remoteIncoming = IntSet.empty
655+ , _remoteIncoming = IntSet.empty
583− , sendOn = sendMany sock
656+ , sendOn = sendMany sock
584− }
657+ }
585− putMVar theirState (RemoteEndPointValid remoteConn)
658+ putMVar theirState (RemoteEndPointValid vst)
586 handleIncomingMessages (ourEndPoint, theirEndPoint)
659 handleIncomingMessages (ourEndPoint, theirEndPoint)
587− Right (ConnectionRequestEndPointInvalid, sock) -> do
660+ Right (sock, ConnectionRequestEndPointInvalid) -> do
588 -- We remove the endpoint from our local state again because the next
661 -- We remove the endpoint from our local state again because the next
589 -- call to 'connect' might give a different result. Threads that were
662 -- call to 'connect' might give a different result. Threads that were
590 -- waiting on the result of this call to connect will get the
663 -- waiting on the result of this call to connect will get the
591 -- RemoteEndPointInvalid; subsequent threads will initiate a new
664 -- RemoteEndPointInvalid; subsequent threads will initiate a new
592 -- connection requests.
665 -- connection requests.
593− modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)
666+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
594 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
667 putMVar theirState (RemoteEndPointInvalid (invalidAddress "Invalid endpoint"))
595 N.sClose sock
668 N.sClose sock
596− Right (ConnectionRequestCrossed, sock) -> do
669+ Right (sock, ConnectionRequestCrossed) -> do
597 N.sClose sock
670 N.sClose sock
598 Left err -> do
671 Left err -> do
599 -- See comment above
672 -- See comment above
600− modifyMVar_ ourState $ return . (localConnectionTo theirAddress ^= Nothing)
673+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
601 putMVar theirState (RemoteEndPointInvalid err)
674 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
604 where
675 where
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
605− go :: ErrorT (FailedWith ConnectErrorCode) IO (ConnectionRequestResponse, N.Socket)
682+ -> IO (Either (FailedWith ConnectErrorCode) (N.Socket, ConnectionRequestResponse))
606− go = do
683+socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
607− let theirAddress = remoteAddress theirEndPoint
684+ (host, port, theirEndPointId) <- case decodeEndPointAddress theirAddress of
608− (host, port, theirEndPointId) <- failWith (invalidAddress "Could not parse") $
685+ Nothing -> throw (invalidAddress . userError $ "Could not parse")
609− decodeEndPointAddress theirAddress
686+ Just dec -> return dec
610− addr:_ <- failWithIO (invalidAddress . show) $
611− N.getAddrInfo Nothing (Just host) (Just port)
687+ addr:_ <- mapExceptionIO invalidAddress $ N.getAddrInfo Nothing (Just host) (Just port)
612− sock <- failWithIO (insufficientResources . show) $
688+ bracketOnError (createSocket addr) N.sClose $ \sock -> do
613− N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
614− response <- failWithIO (failed . show) $ do
689+ response <- mapExceptionIO failed $ do
615 N.setSocketOption sock N.ReuseAddr 1
690 N.setSocketOption sock N.ReuseAddr 1
616 N.connect sock (N.addrAddress addr)
691 N.connect sock (N.addrAddress addr)
617 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
692 sendMany sock (encodeInt32 theirEndPointId : prependLength [ourAddress])
618 recvInt32 sock
693 recvInt32 sock
619 case tryToEnum response of
694 case tryToEnum response of
620− Nothing -> throwError (failed "Unexpected response")
695+ Nothing -> throw (failed . userError $ "Unexpected response")
621− Just r -> return (r, sock)
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
622
701
623− invalidAddress, insufficientResources, failed :: String -> FailedWith ConnectErrorCode
702+ invalidAddress, insufficientResources, failed :: IOException -> FailedWith ConnectErrorCode
624− invalidAddress = FailedWith ConnectInvalidAddress
703+ invalidAddress = FailedWith ConnectInvalidAddress . show
625− insufficientResources = FailedWith ConnectInsufficientResources
704+ insufficientResources = FailedWith ConnectInsufficientResources . show
626− failed = FailedWith ConnectFailed
705+ failed = FailedWith ConnectFailed . show
627−
706+
628− ourAddress :: ByteString
707+-- | Remove reference to a remote endpoint from a local endpoint
629− EndPointAddress ourAddress = localAddress ourEndPoint
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
630
728
631-- | Encode end point address
729-- | Encode end point address
632encodeEndPointAddress :: N.HostName -> N.ServiceName -> EndPointId -> EndPointAddress
730encodeEndPointAddress :: N.HostName -> N.ServiceName -> EndPointId -> EndPointAddress
@@ -645,42 +743,61 @@ decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack
645--
743--
646-- RELY: Remote endpoint must be in valid state.
744-- RELY: Remote endpoint must be in valid state.
647-- GUARANTEE: Will not change the state of the remote endpoint.
745-- 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.
648doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
749doRemoteRequest :: EndPointPair -> ControlHeader -> IO [ByteString]
649−doRemoteRequest (ourEndPoint, theirEndPoint) header = do
750+doRemoteRequest (ourEndPoint, theirEndPoint) header = do
650 reply <- newEmptyMVar
751 reply <- newEmptyMVar
651− reqId <- modifyMVar (localState ourEndPoint) $ \st -> do
752+ reqId <- modifyMVar (localState ourEndPoint) $ \st -> case st of
753+ LocalEndPointValid vst -> do
652− let reqId = st ^. nextCtrlRequestId
754+ let reqId = vst ^. nextCtrlRequestId
653− return ((nextCtrlRequestId ^+= 1) . (pendingCtrlRequestsAt reqId ^= Just reply) $ st, reqId)
755+ return (LocalEndPointValid . (nextCtrlRequestId ^+= 1) . (pendingCtrlRequestsAt reqId ^= Just reply) $ vst, reqId)
756+ LocalEndPointClosed ->
757+ throw (userError "Local endpoint closed")
654− withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) ->
758+ withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid vst) ->
655− sendOn remoteConn [encodeInt32 header, encodeInt32 reqId]
759+ sendOn vst [encodeInt32 header, encodeInt32 reqId]
656 takeMVar reply
760 takeMVar reply
657
761
658-- | Check if the remote endpoint is unused, and if so, send a CloseSocket request
762-- | Check if the remote endpoint is unused, and if so, send a CloseSocket request
659closeIfUnused :: RemoteEndPoint -> IO ()
763closeIfUnused :: RemoteEndPoint -> IO ()
660closeIfUnused theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ \st ->
764closeIfUnused theirEndPoint = modifyMVar_ (remoteState theirEndPoint) $ \st ->
661 case st of
765 case st of
662− RemoteEndPointValid remoteConn -> do
766+ RemoteEndPointValid vst -> do
663− if remoteConn ^. remoteOutgoing == 0 && IntSet.null (remoteConn ^. remoteIncoming)
767+ if vst ^. remoteOutgoing == 0 && IntSet.null (vst ^. remoteIncoming)
664 then do
768 then do
665 resolved <- newEmptyMVar
769 resolved <- newEmptyMVar
666− sendOn remoteConn [encodeInt32 CloseSocket]
770+ sendOn vst [encodeInt32 CloseSocket]
667− return (RemoteEndPointClosing resolved remoteConn)
771+ return (RemoteEndPointClosing resolved vst)
668 else
772 else
669 return st
773 return st
670 _ -> do
774 _ -> do
671 return st
775 return st
672
776
673-- | Fork a new thread and store its ID as part of the transport state
777-- | Fork a new thread and store its ID as part of the transport state
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.
674−forkEndPointThread :: LocalEndPoint -> IO () -> IO ()
781+forkEndPointThread :: LocalEndPoint -> IO () -> IO Bool
675−forkEndPointThread ourEndPoint p = do
782+forkEndPointThread ourEndPoint p =
676− modifyMVar_ (localState ourEndPoint) $ \st -> do
783+ modifyMVar ourState $ \st -> case st of
784+ LocalEndPointValid vst -> do
677− tid <- forkIO (p >> removeThread)
785+ tid <- forkIO (p >> removeThread)
678− return (internalThreads ^%= (tid :) $ st)
786+ return (LocalEndPointValid . (internalThreads ^%= (tid :)) $ vst, True)
787+ LocalEndPointClosed ->
788+ return (LocalEndPointClosed, False)
679 where
789 where
680 removeThread :: IO ()
790 removeThread :: IO ()
681 removeThread = do
791 removeThread = do
682 tid <- myThreadId
792 tid <- myThreadId
793+ modifyMVar_ ourState $ \st -> case st of
794+ LocalEndPointValid vst ->
683− modifyMVar_ (localState ourEndPoint) $ return . (internalThreads ^%= filter (/= tid))
795+ return (LocalEndPointValid . (internalThreads ^%= filter (/= tid)) $ vst)
796+ LocalEndPointClosed ->
797+ return LocalEndPointClosed
798+
799+ ourState :: MVar LocalEndPointState
800+ ourState = localState ourEndPoint
684
801
685--------------------------------------------------------------------------------
802--------------------------------------------------------------------------------
686-- Incoming requests --
803-- Incoming requests --
@@ -688,46 +805,84 @@ forkEndPointThread ourEndPoint p = do
688
805
689-- | Handle a connection request (that is, a remote endpoint that is trying to
806-- | Handle a connection request (that is, a remote endpoint that is trying to
690-- establish a TCP connection with us)
807-- 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.)
691handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
817handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
692−handleConnectionRequest transport sock = do
818+handleConnectionRequest transport sock = handle tryCloseSocket $ do
693− request <- tryIO $ do
694 ourEndPointId <- recvInt32 sock
819 ourEndPointId <- recvInt32 sock
695 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
820 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
696 let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId
821 let ourAddress = encodeEndPointAddress (transportHost transport) (transportPort transport) ourEndPointId
697− ourEndPoint <- (^. localEndPointAt ourAddress) <$> readMVar (transportState transport)
822+ ourEndPoint <- withMVar (transportState transport) $ \st -> case st of
698− return (theirAddress, ourEndPoint)
823+ TransportValid vst ->
699− case request of
824+ case vst ^. localEndPointAt ourAddress 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
706 Nothing -> do
825 Nothing -> do
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
707− theirState <- newEmptyMVar
841+ theirState <- newEmptyMVar
708− let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
842+ let theirEndPoint = RemoteEndPoint { remoteAddress = theirAddress
709− , remoteState = theirState
843+ , remoteState = theirState
710− }
844+ }
711− return (localConnectionTo theirAddress ^= Just theirEndPoint $ st, (False, theirEndPoint))
845+ return (LocalEndPointValid . (localConnectionTo theirAddress ^= Just theirEndPoint) $ vst, (False, theirEndPoint))
712− Just theirEndPoint ->
846+ Just theirEndPoint ->
713− return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
847+ return (st, (localAddress ourEndPoint < theirAddress, theirEndPoint))
848+ LocalEndPointClosed ->
849+ fail "Local endpoint closed"
714− if crossed
850+ if crossed
715− then do
851+ then do
716− sendMany sock [encodeInt32 ConnectionRequestCrossed]
852+ tryIO $ sendMany sock [encodeInt32 ConnectionRequestCrossed]
717− N.sClose sock
853+ tryIO $ N.sClose sock
854+ return Nothing
718− else do
855+ else do
856+ let vst = ValidRemoteEndPointState
719− let remoteConn = RemoteConnection { remoteSocket = sock
857+ { remoteSocket = sock
720− , _remoteOutgoing = 0
858+ , _remoteOutgoing = 0
721− , _remoteIncoming = IntSet.empty
859+ , _remoteIncoming = IntSet.empty
722− , sendOn = sendMany sock
860+ , sendOn = sendMany sock
723− }
861+ }
724− -- TODO: this putMVar might block
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
725− putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)
865+ putMVar (remoteState theirEndPoint) (RemoteEndPointValid vst)
726− sendMany sock [encodeInt32 ConnectionRequestAccepted]
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 ->
727 handleIncomingMessages (ourEndPoint, theirEndPoint)
877 handleIncomingMessages (ourEndPoint, theirEndPoint)
728− Left _ -> do
878+
729− -- Invalid request
879+ tryCloseSocket :: IOException -> IO ()
730− N.sClose sock
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
731
886
732-- | Handle requests from a remote endpoint.
887-- | Handle requests from a remote endpoint.
733--
888--
@@ -785,10 +940,10 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
785 createNewConnection reqId = do
940 createNewConnection reqId = do
786 newId <- getNextConnectionId ourEndPoint
941 newId <- getNextConnectionId ourEndPoint
787 modifyMVar_ theirState $ \st -> do
942 modifyMVar_ theirState $ \st -> do
788− remoteConn <- case st of
943+ vst <- case st of
789− RemoteEndPointValid remoteConn ->
944+ RemoteEndPointValid vst ->
790− return (remoteIncoming ^%= IntSet.insert newId $ remoteConn)
945+ return (remoteIncoming ^%= IntSet.insert newId $ vst)
791− RemoteEndPointClosing resolved remoteConn -> do
946+ RemoteEndPointClosing resolved vst -> do
792 -- If the endpoint is in closing state that means we send a
947 -- If the endpoint is in closing state that means we send a
793 -- CloseSocket request to the remote endpoint. If the remote
948 -- CloseSocket request to the remote endpoint. If the remote
794 -- endpoint replies with the request to create a new connection, it
949 -- endpoint replies with the request to create a new connection, it
@@ -796,24 +951,27 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
796 -- ours. Either way, at this point we simply restore the endpoint
951 -- ours. Either way, at this point we simply restore the endpoint
797 -- to RemoteEndPointValid
952 -- to RemoteEndPointValid
798 putMVar resolved ()
953 putMVar resolved ()
799− return (remoteIncoming ^= IntSet.singleton newId $ remoteConn)
954+ return (remoteIncoming ^= IntSet.singleton newId $ vst)
800 _ ->
955 _ ->
801 error "handleIncomingMessages RELY violation"
956 error "handleIncomingMessages RELY violation"
802− sendOn remoteConn ( encodeInt32 ControlResponse
957+ sendOn vst ( encodeInt32 ControlResponse
803 : encodeInt32 reqId
958 : encodeInt32 reqId
804 : prependLength [encodeInt32 newId]
959 : prependLength [encodeInt32 newId]
805 )
960 )
806 -- We add the new connection ID to the list of open connections only once the
961 -- We add the new connection ID to the list of open connections only once the
807 -- endpoint has been notified of the new connection (sendOn may fail)
962 -- endpoint has been notified of the new connection (sendOn may fail)
808− return (RemoteEndPointValid remoteConn)
963+ return (RemoteEndPointValid vst)
809 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)
964 writeChan ourChannel (ConnectionOpened newId ReliableOrdered theirAddr)
810
965
811 -- Read a control response
966 -- Read a control response
812 readControlResponse :: N.Socket -> ControlRequestId -> IO ()
967 readControlResponse :: N.Socket -> ControlRequestId -> IO ()
813 readControlResponse sock reqId = do
968 readControlResponse sock reqId = do
814 response <- recvWithLength sock
969 response <- recvWithLength sock
815− mmvar <- modifyMVar ourState $ \st ->
970+ mmvar <- modifyMVar ourState $ \st -> case st of
971+ LocalEndPointValid vst ->
816− return (pendingCtrlRequestsAt reqId ^= Nothing $ st, st ^. pendingCtrlRequestsAt reqId)
972+ return (LocalEndPointValid . (pendingCtrlRequestsAt reqId ^= Nothing) $ vst, vst ^. pendingCtrlRequestsAt reqId)
973+ LocalEndPointClosed ->
974+ fail "Local endpoint closed"
817 case mmvar of
975 case mmvar of
818 Nothing -> do
976 Nothing -> do
819 hPutStrLn stderr $ "Warning: Invalid request ID"
977 hPutStrLn stderr $ "Warning: Invalid request ID"
@@ -826,8 +984,8 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
826 closeConnection cid = do
984 closeConnection cid = do
827 -- TODO: we should check that this connection is in fact open
985 -- TODO: we should check that this connection is in fact open
828 writeChan ourChannel (ConnectionClosed cid)
986 writeChan ourChannel (ConnectionClosed cid)
829− modifyMVar_ theirState $ \(RemoteEndPointValid remoteConn) ->
987+ modifyMVar_ theirState $ \(RemoteEndPointValid vst) ->
830− return (RemoteEndPointValid . (remoteIncoming ^%= IntSet.delete cid) $ remoteConn)
988+ return (RemoteEndPointValid . (remoteIncoming ^%= IntSet.delete cid) $ vst)
831 closeIfUnused theirEndPoint
989 closeIfUnused theirEndPoint
832
990
833 -- Close the socket (if we don't have any outgoing connections)
991 -- Close the socket (if we don't have any outgoing connections)
@@ -839,27 +997,27 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
839 -- endpoint in Closed state once we remove the endpoint from our local
997 -- endpoint in Closed state once we remove the endpoint from our local
840 -- connections. But we can do that only once we are sure that we can
998 -- connections. But we can do that only once we are sure that we can
841 -- close the endpoint. Catch-22. We can resolve the catch-22 by locking
999 -- 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
843 -- introducing a double lock and all the associated perils, or by putting
1001 -- introducing a double lock and all the associated perils, or by putting
844 -- the remote endpoint in Closing state first. We opt for the latter.
1002 -- the remote endpoint in Closing state first. We opt for the latter.
845 canClose <- modifyMVar theirState $ \st ->
1003 canClose <- modifyMVar theirState $ \st ->
846 case st of
1004 case st of
847− RemoteEndPointValid remoteConn -> do
1005+ RemoteEndPointValid vst -> do
848 -- We regard a CloseSocket message as an (optimized) way for the
1006 -- We regard a CloseSocket message as an (optimized) way for the
849 -- remote endpoint to indicate that all its connections to us are
1007 -- remote endpoint to indicate that all its connections to us are
850 -- now properly closed
1008 -- now properly closed
851− forM_ (IntSet.elems $ remoteConn ^. remoteIncoming) $ \cid ->
1009+ forM_ (IntSet.elems $ vst ^. remoteIncoming) $ \cid ->
852 writeChan ourChannel (ConnectionClosed cid)
1010 writeChan ourChannel (ConnectionClosed cid)
853− let remoteConn' = remoteIncoming ^= IntSet.empty $ remoteConn
1011+ let vst' = remoteIncoming ^= IntSet.empty $ vst
854 -- Check if we agree that the connection should be closed
1012 -- Check if we agree that the connection should be closed
855− if remoteConn' ^. remoteOutgoing == 0
1013+ if vst' ^. remoteOutgoing == 0
856 then do
1014 then do
857 -- Attempt to reply (but don't insist)
1015 -- Attempt to reply (but don't insist)
858− tryIO $ sendOn remoteConn' [encodeInt32 CloseSocket]
1016+ tryIO $ sendOn vst' [encodeInt32 CloseSocket]
859 resolved <- newEmptyMVar
1017 resolved <- newEmptyMVar
860− return (RemoteEndPointClosing resolved remoteConn', Just resolved)
1018+ return (RemoteEndPointClosing resolved vst', Just resolved)
861 else
1019 else
862− return (RemoteEndPointValid remoteConn', Nothing)
1020+ return (RemoteEndPointValid vst', Nothing)
863 RemoteEndPointClosing resolved _ ->
1021 RemoteEndPointClosing resolved _ ->
864 return (st, Just resolved)
1022 return (st, Just resolved)
865 _ ->
1023 _ ->
@@ -869,7 +1027,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
869 Nothing ->
1027 Nothing ->
870 return False
1028 return False
871 Just resolved -> do
1029 Just resolved -> do
872− modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)
1030+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
873 modifyMVar_ theirState $ return . const RemoteEndPointClosed
1031 modifyMVar_ theirState $ return . const RemoteEndPointClosed
874 N.sClose sock
1032 N.sClose sock
875 putMVar resolved ()
1033 putMVar resolved ()
@@ -889,13 +1047,13 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
889 prematureExit :: N.Socket -> IOException -> IO ()
1047 prematureExit :: N.Socket -> IOException -> IO ()
890 prematureExit sock _ = do
1048 prematureExit sock _ = do
891 N.sClose sock
1049 N.sClose sock
892− modifyMVar_ ourState $ return . (localConnectionTo theirAddr ^= Nothing)
1050+ removeRemoteEndPoint (ourEndPoint, theirEndPoint)
893 mUnclosedConnections <- modifyMVar theirState $ \st ->
1051 mUnclosedConnections <- modifyMVar theirState $ \st ->
894 case st of
1052 case st of
895 RemoteEndPointInvalid _ ->
1053 RemoteEndPointInvalid _ ->
896 error "handleIncomingMessages RELY violation"
1054 error "handleIncomingMessages RELY violation"
897− RemoteEndPointValid remoteConn ->
1055+ RemoteEndPointValid vst ->
898− return (RemoteEndPointClosed, Just $ remoteConn ^. remoteIncoming)
1056+ return (RemoteEndPointClosed, Just $ vst ^. remoteIncoming)
899 RemoteEndPointClosing _ _ ->
1057 RemoteEndPointClosing _ _ ->
900 return (RemoteEndPointClosed, Nothing)
1058 return (RemoteEndPointClosed, Nothing)
901 RemoteEndPointClosed ->
1059 RemoteEndPointClosed ->
@@ -910,12 +1068,17 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
910 ErrorEventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)
1068 ErrorEventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)
911
1069
912-- | Get the next connection ID
1070-- | Get the next connection ID
1071+--
1072+-- Throws an IO exception when the endpoint is closed.
913getNextConnectionId :: LocalEndPoint -> IO ConnectionId
1073getNextConnectionId :: LocalEndPoint -> IO ConnectionId
914getNextConnectionId ourEndpoint =
1074getNextConnectionId ourEndpoint =
915− modifyMVar (localState ourEndpoint) $ \st -> do
1075+ modifyMVar (localState ourEndpoint) $ \st -> case st of
1076+ LocalEndPointValid vst -> do
916− let connId = st ^. nextConnectionId
1077+ let connId = vst ^. nextConnectionId
917− return (nextConnectionId ^= connId +1 $ st, connId)
1078+ return (LocalEndPointValid . (nextConnectionId ^= connId + 1) $ vst, connId)
918−
1079+ LocalEndPointClosed ->
1080+ fail "Local endpoint closed"
1081+
919--------------------------------------------------------------------------------
1082--------------------------------------------------------------------------------
920-- Constants --
1083-- Constants --
921--------------------------------------------------------------------------------
1084--------------------------------------------------------------------------------
@@ -928,38 +1091,35 @@ firstNonReservedConnectionId = 1024
928-- Lens definitions --
1091-- Lens definitions --
929--------------------------------------------------------------------------------
1092--------------------------------------------------------------------------------
930
1093
931−localEndPoints :: Lens TransportState (Map EndPointAddress LocalEndPoint)
1094+localEndPoints :: Lens ValidTransportState (Map EndPointAddress LocalEndPoint)
932localEndPoints = lens _localEndPoints (\es st -> st { _localEndPoints = es })
1095localEndPoints = lens _localEndPoints (\es st -> st { _localEndPoints = es })
933
1096
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]))
938pendingCtrlRequests = lens _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })
1098pendingCtrlRequests = lens _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })
939
1099
940−nextCtrlRequestId :: Lens LocalState ControlRequestId
1100+nextCtrlRequestId :: Lens ValidLocalEndPointState ControlRequestId
941nextCtrlRequestId = lens _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid })
1101nextCtrlRequestId = lens _nextCtrlRequestId (\cid st -> st { _nextCtrlRequestId = cid })
942
1102
943−nextConnectionId :: Lens LocalState ConnectionId
1103+nextConnectionId :: Lens ValidLocalEndPointState ConnectionId
944nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
1104nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
945
1105
946−localConnections :: Lens LocalState (Map EndPointAddress RemoteEndPoint)
1106+localConnections :: Lens ValidLocalEndPointState (Map EndPointAddress RemoteEndPoint)
947localConnections = lens _localConnections (\es st -> st { _localConnections = es })
1107localConnections = lens _localConnections (\es st -> st { _localConnections = es })
948
1108
949−internalThreads :: Lens LocalState [ThreadId]
1109+internalThreads :: Lens ValidLocalEndPointState [ThreadId]
950internalThreads = lens _internalThreads (\ts st -> st { _internalThreads = ts })
1110internalThreads = lens _internalThreads (\ts st -> st { _internalThreads = ts })
951
1111
952−localEndPointAt :: EndPointAddress -> Lens TransportState (Maybe LocalEndPoint)
1112+localEndPointAt :: EndPointAddress -> Lens ValidTransportState (Maybe LocalEndPoint)
953localEndPointAt addr = localEndPoints >>> mapLens addr
1113localEndPointAt addr = localEndPoints >>> mapLens addr
954
1114
955−pendingCtrlRequestsAt :: ControlRequestId -> Lens LocalState (Maybe (MVar [ByteString]))
1115+pendingCtrlRequestsAt :: ControlRequestId -> Lens ValidLocalEndPointState (Maybe (MVar [ByteString]))
956pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)
1116pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)
957
1117
958−localConnectionTo :: EndPointAddress -> Lens LocalState (Maybe RemoteEndPoint)
1118+localConnectionTo :: EndPointAddress -> Lens ValidLocalEndPointState (Maybe RemoteEndPoint)
959localConnectionTo addr = localConnections >>> mapLens addr
1119localConnectionTo addr = localConnections >>> mapLens addr
960
1120
961−remoteOutgoing :: Lens RemoteConnection Int
1121+remoteOutgoing :: Lens ValidRemoteEndPointState Int
962remoteOutgoing = lens _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })
1122remoteOutgoing = lens _remoteOutgoing (\cs conn -> conn { _remoteOutgoing = cs })
963
1123
964−remoteIncoming :: Lens RemoteConnection IntSet
1124+remoteIncoming :: Lens ValidRemoteEndPointState IntSet
965remoteIncoming = lens _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })
1125remoteIncoming = lens _remoteIncoming (\cs conn -> conn { _remoteIncoming = cs })
network-transport-2/tests/TestTCP.hs+44−72
@@ -8,30 +8,18 @@ import TestAuxiliary (forkTry, runTests)
8import Network.Transport
8import Network.Transport
9import Network.Transport.TCP (createTransport, encodeEndPointAddress)
9import Network.Transport.TCP (createTransport, encodeEndPointAddress)
10import Data.Int (Int32)
10import Data.Int (Int32)
11−import Data.Maybe (fromJust)
12import Control.Concurrent (threadDelay)
11import Control.Concurrent (threadDelay)
13import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar, tryTakeMVar)
12import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar, tryTakeMVar)
14import Control.Monad (replicateM, guard, forM_)
13import Control.Monad (replicateM, guard, forM_)
15import Control.Applicative ((<$>))
14import Control.Applicative ((<$>))
16import Control.Exception (throw)
15import Control.Exception (throw)
17−import Network.Transport.TCP ( decodeEndPointAddress
16+import Network.Transport.TCP ( ControlHeader(..)
18− , EndPointId
19− , ControlHeader(..)
20 , ConnectionRequestResponse(..)
17 , ConnectionRequestResponse(..)
18+ , socketToEndPoint
21 )
19 )
22import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO)
20import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO)
23import Network.Transport.Internal.TCP (recvInt32, forkServer, recvWithLength)
21import Network.Transport.Internal.TCP (recvInt32, forkServer, recvWithLength)
24−import qualified Network.Socket as N ( getAddrInfo
22+import qualified Network.Socket as N ( sClose
25− , socket
26− , connect
27− , addrFamily
28− , addrAddress
29− , SocketType(Stream)
30− , SocketOption(ReuseAddr)
31− , defaultProtocol
32− , setSocketOption
33− , sClose
34− , HostName
35 , ServiceName
23 , ServiceName
36 , Socket
24 , Socket
37 , AddrInfo
25 , AddrInfo
@@ -51,6 +39,7 @@ instance Traceable N.Socket where
51
39
52instance Traceable N.AddrInfo where
40instance Traceable N.AddrInfo where
53 trace = traceShow
41 trace = traceShow
42+
54
43
55-- Test that the server gets a ConnectionClosed message when the client closes
44-- Test that the server gets a ConnectionClosed message when the client closes
56-- the socket without sending an explicit control message to the server first
45-- the socket without sending an explicit control message to the server first
@@ -112,7 +101,6 @@ testEarlyDisconnect nextPort = do
112 clientPort <- nextPort
101 clientPort <- nextPort
113 let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0
102 let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0
114 putMVar clientAddr ourAddress
103 putMVar clientAddr ourAddress
115− (serverHost, serverPort, endPointIx) <- fromJust . decodeEndPointAddress <$> readMVar serverAddr
116
104
117 -- Listen for incoming messages
105 -- Listen for incoming messages
118 forkServer "127.0.0.1" clientPort 5 throw $ \sock -> do
106 forkServer "127.0.0.1" clientPort 5 throw $ \sock -> do
@@ -141,15 +129,7 @@ testEarlyDisconnect nextPort = do
141 N.sClose sock
129 N.sClose sock
142
130
143 -- Connect to the server
131 -- Connect to the server
144− addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
145− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
146− N.setSocketOption sock N.ReuseAddr 1
147− N.connect sock (N.addrAddress addr)
148−
149− sendMany sock (encodeInt32 endPointIx : prependLength [endPointAddressToByteString ourAddress])
150−
151− -- Wait for acknowledgement
152− ConnectionRequestAccepted <- toEnum <$> recvInt32 sock
132+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= socketToEndPoint ourAddress
153
133
154 -- Request a new connection, but don't wait for the response
134 -- Request a new connection, but don't wait for the response
155 let reqId = 0 :: Int32
135 let reqId = 0 :: Int32
@@ -223,7 +203,6 @@ testEarlyCloseSocket nextPort = do
223 clientPort <- nextPort
203 clientPort <- nextPort
224 let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0
204 let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0
225 putMVar clientAddr ourAddress
205 putMVar clientAddr ourAddress
226− (serverHost, serverPort, endPointIx) <- fromJust . decodeEndPointAddress <$> readMVar serverAddr
227
206
228 -- Listen for incoming messages
207 -- Listen for incoming messages
229 forkServer "127.0.0.1" clientPort 5 throw $ \sock -> do
208 forkServer "127.0.0.1" clientPort 5 throw $ \sock -> do
@@ -254,15 +233,7 @@ testEarlyCloseSocket nextPort = do
254 N.sClose sock
233 N.sClose sock
255
234
256 -- Connect to the server
235 -- Connect to the server
257− addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
258− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
259− N.setSocketOption sock N.ReuseAddr 1
260− N.connect sock (N.addrAddress addr)
261−
262− sendMany sock (encodeInt32 endPointIx : prependLength [endPointAddressToByteString ourAddress])
263−
264− -- Wait for acknowledgement
265− ConnectionRequestAccepted <- toEnum <$> recvInt32 sock
236+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= socketToEndPoint ourAddress
266
237
267 -- Request a new connection, but don't wait for the response
238 -- Request a new connection, but don't wait for the response
268 let reqId = 0 :: Int32
239 let reqId = 0 :: Int32
@@ -319,11 +290,11 @@ testIgnoreCloseSocket nextPort = do
319 takeMVar clientDone
290 takeMVar clientDone
320
291
321 where
292 where
322− server :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> IO ()
293+ server :: Transport -> MVar EndPointAddress -> IO ()
323 server transport serverAddr = do
294 server transport serverAddr = do
324 tlog "Server"
295 tlog "Server"
325 Right endpoint <- newEndPoint transport
296 Right endpoint <- newEndPoint transport
326− putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
297+ putMVar serverAddr (address endpoint)
327
298
328 -- Wait for the client to connect and disconnect
299 -- Wait for the client to connect and disconnect
329 tlog "Waiting for ConnectionOpened"
300 tlog "Waiting for ConnectionOpened"
@@ -342,24 +313,14 @@ testIgnoreCloseSocket nextPort = do
342
313
343 tlog "Server waiting.."
314 tlog "Server waiting.."
344
315
345− client :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> MVar () -> IO ()
316+ client :: Transport -> MVar EndPointAddress -> MVar () -> IO ()
346 client transport serverAddr clientDone = do
317 client transport serverAddr clientDone = do
347 tlog "Client"
318 tlog "Client"
348 Right endpoint <- newEndPoint transport
319 Right endpoint <- newEndPoint transport
349− let EndPointAddress myAddress = address endpoint
320+ let ourAddress = address endpoint
350− (serverHost, serverPort, endPointIx) <- readMVar serverAddr
351
321
352 -- Connect to the server
322 -- Connect to the server
353− addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
354− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
355− N.setSocketOption sock N.ReuseAddr 1
356− N.connect sock (N.addrAddress addr)
357−
358− tlog "Connecting to endpoint"
359− sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])
360−
361− -- Wait for acknowledgement
362− ConnectionRequestAccepted <- toEnum <$> recvInt32 sock
323+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= socketToEndPoint ourAddress
363
324
364 -- Request a new connection
325 -- Request a new connection
365 tlog "Requesting connection"
326 tlog "Requesting connection"
@@ -412,11 +373,11 @@ testBlockAfterCloseSocket nextPort = do
412 takeMVar clientDone
373 takeMVar clientDone
413
374
414 where
375 where
415− server :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> IO ()
376+ server :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> IO ()
416 server transport serverAddr clientAddr = do
377 server transport serverAddr clientAddr = do
417 tlog "Server"
378 tlog "Server"
418 Right endpoint <- newEndPoint transport
379 Right endpoint <- newEndPoint transport
419− putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
380+ putMVar serverAddr (address endpoint)
420
381
421 -- Wait for the client to connect and disconnect
382 -- Wait for the client to connect and disconnect
422 tlog "Waiting for ConnectionOpened"
383 tlog "Waiting for ConnectionOpened"
@@ -431,25 +392,15 @@ testBlockAfterCloseSocket nextPort = do
431
392
432 tlog "Server waiting.."
393 tlog "Server waiting.."
433
394
434− client :: Transport -> MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> MVar () -> IO ()
395+ client :: Transport -> MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()
435 client transport serverAddr clientAddr clientDone = do
396 client transport serverAddr clientAddr clientDone = do
436 tlog "Client"
397 tlog "Client"
437 Right endpoint <- newEndPoint transport
398 Right endpoint <- newEndPoint transport
438 putMVar clientAddr (address endpoint)
399 putMVar clientAddr (address endpoint)
439− let EndPointAddress myAddress = address endpoint
400+ let ourAddress = address endpoint
440− (serverHost, serverPort, endPointIx) <- readMVar serverAddr
441
401
442 -- Connect to the server
402 -- Connect to the server
443− addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
444− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
445− N.setSocketOption sock N.ReuseAddr 1
446− N.connect sock (N.addrAddress addr)
447−
448− tlog "Connecting to endpoint"
449− sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])
450−
451− -- Wait for acknowledgement
452− ConnectionRequestAccepted <- toEnum <$> recvInt32 sock
403+ Right (sock, ConnectionRequestAccepted) <- readMVar serverAddr >>= socketToEndPoint ourAddress
453
404
454 -- Request a new connection
405 -- Request a new connection
455 tlog "Requesting connection"
406 tlog "Requesting connection"
@@ -483,18 +434,39 @@ testBlockAfterCloseSocket nextPort = do
483 let reqId' = 1 :: Int32
434 let reqId' = 1 :: Int32
484 sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId']
435 sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId']
485
436
437+-- | Test what happens when a remote endpoint sends a connection request to our
438+-- transport for an endpoint it already has a connection to
439+testUnnecessaryConnect :: IO N.ServiceName -> IO ()
440+testUnnecessaryConnect nextPort = do
441+ clientDone <- newEmptyMVar
442+ serverAddr <- newEmptyMVar
443+
444+ forkTry $ do
445+ Right transport <- nextPort >>= createTransport "127.0.0.1"
446+ Right endpoint <- newEndPoint transport
447+ putMVar serverAddr (address endpoint)
448+
449+ forkTry $ do
450+ let ourAddress = EndPointAddress "ourAddress"
451+ Right (_, ConnectionRequestAccepted) <- readMVar serverAddr >>= socketToEndPoint ourAddress
452+ Right (_, ConnectionRequestEndPointInvalid) <- readMVar serverAddr >>= socketToEndPoint ourAddress
453+ putMVar clientDone ()
454+
455+ takeMVar clientDone
456+
486main :: IO ()
457main :: IO ()
487main = do
458main = do
488 portMVar <- newEmptyMVar
459 portMVar <- newEmptyMVar
489− forkTry $ forM_ ([10080 .. 10087] :: [Int]) $ putMVar portMVar . show
460+ forkTry $ forM_ ([10080 .. 10088] :: [Int]) $ putMVar portMVar . show
490 let nextPort = do Just port <- tryTakeMVar portMVar ; return port
461 let nextPort = do Just port <- tryTakeMVar portMVar ; return port
491 tryIO $ runTests
462 tryIO $ runTests
492− [ ("EarlyDisconnect", testEarlyDisconnect nextPort)
463+ [ ("EarlyDisconnect", testEarlyDisconnect nextPort)
493− , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)
464+ , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)
494− , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)
465+ , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)
495− , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)
466+ , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)
467+ , ("TestUnnecessaryConnect", testUnnecessaryConnect nextPort)
496− , ("InvalidAddress", testInvalidAddress nextPort)
468+ , ("InvalidAddress", testInvalidAddress nextPort)
497− , ("InvalidConnect", testInvalidConnect nextPort)
469+ , ("InvalidConnect", testInvalidConnect nextPort)
498 ]
470 ]
499 Right transport <- createTransport "127.0.0.1" "8080"
471 Right transport <- createTransport "127.0.0.1" "8080"
500 testTransport transport
472 testTransport transport
network-transport-2/tests/TestTransport.hs+6
@@ -586,6 +586,12 @@ testCloseEndPoint transport _ = do
586 -- Attempt to send should fail with connection closed
586 -- Attempt to send should fail with connection closed
587 Left (FailedWith SendConnectionClosed _) <- send conn ["ping2"]
587 Left (FailedWith SendConnectionClosed _) <- send conn ["ping2"]
588
588
589+ -- An attempt to close the already closed connection should just return
590+ () <- close conn
591+
592+ -- And so should an attempt to connect
593+ Left (FailedWith ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered
594+
589 return ()
595 return ()
590
596
591 putMVar clientDone ()
597 putMVar clientDone ()