git.iko.soy/distributed-process
on master help sign in
Text testEarlyDisconnect
955935dEdsko de Vries
network-transport-2/src/Network/Transport/Internal/TCP.hs+2−2
compare with master
unifiedsplit
@@ -49,10 +49,10 @@ import Data.Int (Int32)
49forkServer :: N.HostName -- ^ Host
49forkServer :: N.HostName -- ^ Host
50 -> N.ServiceName -- ^ Port
50 -> N.ServiceName -- ^ Port
51 -> Int -- ^ Backlog (maximum number of queued connections)
51 -> Int -- ^ Backlog (maximum number of queued connections)
52− -> (N.Socket -> IO ()) -- ^ Request handler
53 -> (SomeException -> IO ()) -- ^ Termination handler
52 -> (SomeException -> IO ()) -- ^ Termination handler
53+ -> (N.Socket -> IO ()) -- ^ Request handler
54 -> IO ThreadId
54 -> IO ThreadId
55−forkServer host port backlog requestHandler terminationHandler = do
55+forkServer host port backlog terminationHandler requestHandler = do
56 -- Resolve the specified address. By specification, getAddrInfo will never
56 -- Resolve the specified address. By specification, getAddrInfo will never
57 -- return an empty list (but will throw an exception instead) and will return
57 -- return an empty list (but will throw an exception instead) and will return
58 -- the "best" address first, whatever that means
58 -- the "best" address first, whatever that means
network-transport-2/src/Network/Transport/TCP.hs+19−13
@@ -292,8 +292,7 @@ createTransport host port = do
292 }
292 }
293 tryIO $ do
293 tryIO $ do
294 -- TODO: "5" is listed as "system dependent maximum backlog"?
294 -- TODO: "5" is listed as "system dependent maximum backlog"?
295− transportThread <- forkServer host port 5 (handleConnectionRequest transport)
295+ transportThread <- forkServer host port 5 (terminationHandler transport) (handleConnectionRequest transport)
296− (terminationHandler transport)
297 return Transport { newEndPoint = apiNewEndPoint transport
296 return Transport { newEndPoint = apiNewEndPoint transport
298 , closeTransport = apiCloseTransport transportThread
297 , closeTransport = apiCloseTransport transportThread
299 }
298 }
@@ -369,18 +368,25 @@ apiClose theirEndPoint connId connAlive = void . tryIO $ do
369
368
370-- | Send data across a connection
369-- | Send data across a connection
371--
370--
372−-- RELY: The remote endpoint must remain in state 'RemoteEndPointValid'.
371+-- RELY: The remote endpoint must be in state 'RemoteEndPointValid' or 'RemoteEndPointClosed'
373-- GUARANTEE: The state of the remote endpoint will not be changed.
372-- GUARANTEE: The state of the remote endpoint will not be changed.
374apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
373apiSend :: RemoteEndPoint -> ConnectionId -> IORef Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
375apiSend theirEndPoint connId connAlive payload = do
374apiSend theirEndPoint connId connAlive payload = do
375+ withMVar (remoteState theirEndPoint) $ \st -> case st of
376− withMVar (remoteState theirEndPoint) $ \(RemoteEndPointValid remoteConn) -> do
376+ RemoteEndPointValid remoteConn -> do
377− alive <- readIORef connAlive
377+ alive <- readIORef connAlive
378− runErrorT $ if alive
378+ runErrorT $ if alive
379− then
379+ then
380− failWithIO (FailedWith SendFailed . show) $
380+ failWithIO (FailedWith SendFailed . show) $
381− sendOn remoteConn (encodeInt32 connId : prependLength payload)
381+ sendOn remoteConn (encodeInt32 connId : prependLength payload)
382− else
382+ else
383− throwError $ FailedWith SendConnectionClosed "Connection closed"
383+ throwError $ FailedWith SendConnectionClosed "Connection closed"
384+ RemoteEndPointClosed -> do
385+ return (Left $ FailedWith SendConnectionClosed "Endpoint closed")
386+ RemoteEndPointClosing _ _ ->
387+ error "apiSend RELY violation"
388+ RemoteEndPointInvalid _ ->
389+ error "apiSend RELY violation"
384
390
385-- | Force-close the endpoint
391-- | Force-close the endpoint
386apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
392apiCloseEndPoint :: TCPTransport -> LocalEndPoint -> IO ()
@@ -837,7 +843,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
837 case st of
843 case st of
838 RemoteEndPointValid remoteConn -> do
844 RemoteEndPointValid remoteConn -> do
839 -- TODO: We check only for outgoing connections here. That is
845 -- TODO: We check only for outgoing connections here. That is
840− -- correct, but we need to makke sure that we send the right event
846+ -- correct, but we need to make sure that we send the right event
841 -- to our endpoint for any incoming connections that are still live
847 -- to our endpoint for any incoming connections that are still live
842 -- at this point.
848 -- at this point.
843 if remoteConn ^. remoteOutgoing == 0
849 if remoteConn ^. remoteOutgoing == 0
@@ -852,7 +858,7 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
852 return (st, Just resolved)
858 return (st, Just resolved)
853 _ ->
859 _ ->
854 error "handleIncomingConnections RELY violation"
860 error "handleIncomingConnections RELY violation"
855−
861+
856 case canClose of
862 case canClose of
857 Nothing ->
863 Nothing ->
858 return False
864 return False
network-transport-2/tests/TestTCP.hs+55−27
@@ -13,13 +13,14 @@ import Control.Concurrent (threadDelay)
13import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar)
13import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar)
14import Control.Monad (replicateM, guard)
14import Control.Monad (replicateM, guard)
15import Control.Applicative ((<$>))
15import Control.Applicative ((<$>))
16+import Control.Exception (throw)
16import Network.Transport.TCP ( decodeEndPointAddress
17import Network.Transport.TCP ( decodeEndPointAddress
17 , EndPointId
18 , EndPointId
18 , ControlHeader(..)
19 , ControlHeader(..)
19 , ConnectionRequestResponse(..)
20 , ConnectionRequestResponse(..)
20 )
21 )
21import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO)
22import Network.Transport.Internal (encodeInt32, prependLength, tlog, tryIO)
22−import Network.Transport.Internal.TCP (recvInt32)
23+import Network.Transport.Internal.TCP (recvInt32, forkServer, recvWithLength)
23import qualified Network.Socket as N ( getAddrInfo
24import qualified Network.Socket as N ( getAddrInfo
24 , socket
25 , socket
25 , connect
26 , connect
@@ -73,36 +74,37 @@ testEarlyDisconnect = do
73 putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
74 putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
74 theirAddr <- readMVar clientAddr
75 theirAddr <- readMVar clientAddr
75
76
76− -- First test: they connect to us, then drop the connection
77+ -- TEST 1: they connect to us, then drop the connection
78+ do
77− ConnectionOpened cid _ addr <- receive endpoint
79+ ConnectionOpened cid _ addr <- receive endpoint
78− True <- return $ addr == theirAddr
80+ True <- return $ addr == theirAddr
79
81
80− ErrorEvent (ErrorEventConnectionLost addr' [cid']) <- receive endpoint
82+ ErrorEvent (ErrorEventConnectionLost addr' [cid']) <- receive endpoint
81− True <- return $ addr' == theirAddr && cid' == cid
83+ True <- return $ addr' == theirAddr && cid' == cid
84+
85+ return ()
82
86
83− -- Second test: after they dropped their connection to us, we now try to
87+ -- TEST 2: after they dropped their connection to us, we now try to
84 -- establish a connection to them. This should re-establish the broken
88 -- establish a connection to them. This should re-establish the broken
85 -- TCP connection.
89 -- TCP connection.
86 tlog "Trying to connect to client"
90 tlog "Trying to connect to client"
87− Right _ <- connect endpoint theirAddr ReliableOrdered
91+ Right conn <- connect endpoint theirAddr ReliableOrdered
88
92
89− putMVar serverDone ()
90−
91−{-
92− -- To test the connection, we do a simple ping test; as before, however,
93+ -- TEST 3: To test the connection, we do a simple ping test; as before,
93− -- the remote client won't close the connection nicely but just closes
94+ -- however, the remote client won't close the connection nicely but just
94− -- the socket
95+ -- closes the socket
95 do
96 do
96− send conn ["ping"]
97+ Right () <- send conn ["ping"]
97− cid <- expect endpoint (\(ConnectionOpened cid _ addr) -> (addr == theirAddr, cid))
98+ ConnectionOpened cid _ addr <- receive endpoint ; True <- return $ addr == theirAddr
98− expect' endpoint (\(Received cid' [msg]) -> cid' == cid && msg == "pong")
99+ Received cid' ["pong"] <- receive endpoint ; True <- return $ cid == cid'
99− expect' endpoint (\(ErrorEvent (ErrorEventConnectionLost addr [cid'])) -> addr == theirAddr && cid' == cid)
100+ ErrorEvent (ErrorEventConnectionLost addr' [cid'']) <- receive endpoint ; True <- return $ addr' == theirAddr && cid'' == cid
101+ return ()
100
102
101− -- TODO: we now need to decide (and test) and should happen on a subsequent 'send'
103+ -- TEST 4: A subsequent send on an already-open connection will now break
102−
104+ Left (FailedWith SendConnectionClosed _) <- send conn ["ping2"]
103−-}
104−
105
105
106+ -- *Pfew*
107+ putMVar serverDone ()
106
108
107 client :: MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> IO ()
109 client :: MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> IO ()
108 client serverAddr clientAddr = do
110 client serverAddr clientAddr = do
@@ -111,7 +113,31 @@ testEarlyDisconnect = do
111 putMVar clientAddr ourAddress
113 putMVar clientAddr ourAddress
112 (_, _, endPointIx) <- readMVar serverAddr
114 (_, _, endPointIx) <- readMVar serverAddr
113
115
114−
116+ -- Listen for incoming messages
117+ forkServer "127.0.0.1" "8082" 5 throw $ \sock -> do
118+ -- Initial setup
119+ 0 <- recvInt32 sock :: IO Int
120+ _ <- recvWithLength sock
121+ sendMany sock [encodeInt32 ConnectionRequestAccepted]
122+
123+ -- Server requests a logical connection
124+ RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)
125+ reqId <- recvInt32 sock :: IO Int
126+ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])
127+
128+ -- Server sends a message
129+ 10001 <- recvInt32 sock :: IO Int
130+ ["ping"] <- recvWithLength sock
131+
132+ -- Reply
133+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)]
134+ ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int)
135+ 10002 <- recvInt32 sock :: IO Int
136+ [cid] <- recvWithLength sock
137+ sendMany sock (cid : prependLength ["pong"])
138+
139+ -- Close the socket
140+ N.sClose sock
115
141
116 -- Connect to the server
142 -- Connect to the server
117 addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")
143 addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")
@@ -132,6 +158,7 @@ testEarlyDisconnect = do
132 -- The server should still receive a ConnectionClosed message
158 -- The server should still receive a ConnectionClosed message
133 N.sClose sock
159 N.sClose sock
134
160
161+
135-- | Test the creation of a transport with an invalid address
162-- | Test the creation of a transport with an invalid address
136testInvalidAddress :: IO ()
163testInvalidAddress :: IO ()
137testInvalidAddress = do
164testInvalidAddress = do
@@ -344,10 +371,11 @@ main :: IO ()
344main = do
371main = do
345 tryIO $ runTests
372 tryIO $ runTests
346 [ ("EarlyDisconnect", testEarlyDisconnect)
373 [ ("EarlyDisconnect", testEarlyDisconnect)
347− , ("InvalidAddress", testInvalidAddress)
374+ , ("InvalidAddress", testInvalidAddress)
348− , ("InvalidConnect", testInvalidConnect)
375+ , ("InvalidConnect", testInvalidConnect)
349− , ("IgnoreCloseSocket", testIgnoreCloseSocket)
376+ , ("IgnoreCloseSocket", testIgnoreCloseSocket)
350− , ("BlockAfterCloseSocket", testBlockAfterCloseSocket)
377+ , ("BlockAfterCloseSocket", testBlockAfterCloseSocket)
351 ]
378 ]
352 Right transport <- createTransport "127.0.0.1" "8080"
379 Right transport <- createTransport "127.0.0.1" "8080"
353 testTransport transport
380 testTransport transport
381+ return ()