git.iko.soy/distributed-process
on master help sign in
Deal with premature CloseSocket messages
43d2c27Edsko de Vries
network-transport-2/src/Network/Transport/TCP.hs+12−9
compare with master
unifiedsplit
@@ -842,18 +842,21 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
842 canClose <- modifyMVar theirState $ \st ->
842 canClose <- modifyMVar theirState $ \st ->
843 case st of
843 case st of
844 RemoteEndPointValid remoteConn -> do
844 RemoteEndPointValid remoteConn -> do
845− -- TODO: We check only for outgoing connections here. That is
845+ -- We regard a CloseSocket message as an (optimized) way for the
846− -- correct, but we need to make sure that we send the right event
847− -- to our endpoint for any incoming connections that are still live
846+ -- remote endpoint to indicate that all its connections to us are
848− -- at this point.
847+ -- now properly closed
848+ forM_ (IntSet.elems $ remoteConn ^. remoteIncoming) $ \cid ->
849+ writeChan ourChannel (ConnectionClosed cid)
850+ let remoteConn' = remoteIncoming ^= IntSet.empty $ remoteConn
851+ -- Check if we agree that the connection should be closed
849− if remoteConn ^. remoteOutgoing == 0
852+ if remoteConn' ^. remoteOutgoing == 0
850 then do
853 then do
851− -- TODO: code duplication with closeIfUnused
854+ -- Attempt to reply (but don't insist)
855+ tryIO $ sendOn remoteConn' [encodeInt32 CloseSocket]
852 resolved <- newEmptyMVar
856 resolved <- newEmptyMVar
853− sendOn remoteConn [encodeInt32 CloseSocket]
854− return (RemoteEndPointClosing resolved remoteConn, Just resolved)
857+ return (RemoteEndPointClosing resolved remoteConn', Just resolved)
855 else
858 else
856− return (st, Nothing)
859+ return (RemoteEndPointValid remoteConn', Nothing)
857 RemoteEndPointClosing resolved _ ->
860 RemoteEndPointClosing resolved _ ->
858 return (st, Just resolved)
861 return (st, Just resolved)
859 _ ->
862 _ ->
network-transport-2/tests/TestTCP.hs+158−38
@@ -10,8 +10,8 @@ import Network.Transport.TCP (createTransport, encodeEndPointAddress)
10import Data.Int (Int32)
10import Data.Int (Int32)
11import Data.Maybe (fromJust)
11import Data.Maybe (fromJust)
12import Control.Concurrent (threadDelay)
12import Control.Concurrent (threadDelay)
13−import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar)
13+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar, tryTakeMVar)
14−import Control.Monad (replicateM, guard)
14+import Control.Monad (replicateM, guard, forM_)
15import Control.Applicative ((<$>))
15import Control.Applicative ((<$>))
16import Control.Exception (throw)
16import Control.Exception (throw)
17import Network.Transport.TCP ( decodeEndPointAddress
17import Network.Transport.TCP ( decodeEndPointAddress
@@ -54,8 +54,8 @@ instance Traceable N.AddrInfo where
54
54
55-- Test that the server gets a ConnectionClosed message when the client closes
55-- 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
56-- the socket without sending an explicit control message to the server first
57−testEarlyDisconnect :: IO ()
57+testEarlyDisconnect :: IO N.ServiceName -> IO ()
58−testEarlyDisconnect = do
58+testEarlyDisconnect nextPort = do
59 clientAddr <- newEmptyMVar
59 clientAddr <- newEmptyMVar
60 serverAddr <- newEmptyMVar
60 serverAddr <- newEmptyMVar
61 serverDone <- newEmptyMVar
61 serverDone <- newEmptyMVar
@@ -66,12 +66,12 @@ testEarlyDisconnect = do
66
66
67 takeMVar serverDone
67 takeMVar serverDone
68 where
68 where
69− server :: MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> MVar () -> IO ()
69+ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()
70 server serverAddr clientAddr serverDone = do
70 server serverAddr clientAddr serverDone = do
71 tlog "Server"
71 tlog "Server"
72− Right transport <- createTransport "127.0.0.1" "8081"
72+ Right transport <- nextPort >>= createTransport "127.0.0.1"
73− Right endpoint <- newEndPoint transport
73+ Right endpoint <- newEndPoint transport
74− putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)
74+ putMVar serverAddr (address endpoint)
75 theirAddr <- readMVar clientAddr
75 theirAddr <- readMVar clientAddr
76
76
77 -- TEST 1: they connect to us, then drop the connection
77 -- TEST 1: they connect to us, then drop the connection
@@ -106,15 +106,16 @@ testEarlyDisconnect = do
106 -- *Pfew*
106 -- *Pfew*
107 putMVar serverDone ()
107 putMVar serverDone ()
108
108
109− client :: MVar (N.HostName, N.ServiceName, EndPointId) -> MVar EndPointAddress -> IO ()
109+ client :: MVar EndPointAddress -> MVar EndPointAddress -> IO ()
110 client serverAddr clientAddr = do
110 client serverAddr clientAddr = do
111 tlog "Client"
111 tlog "Client"
112+ clientPort <- nextPort
112− let ourAddress = encodeEndPointAddress "127.0.0.1" "8082" 0
113+ let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0
113 putMVar clientAddr ourAddress
114 putMVar clientAddr ourAddress
114− (_, _, endPointIx) <- readMVar serverAddr
115+ (serverHost, serverPort, endPointIx) <- fromJust . decodeEndPointAddress <$> readMVar serverAddr
115
116
116 -- Listen for incoming messages
117 -- Listen for incoming messages
117− forkServer "127.0.0.1" "8082" 5 throw $ \sock -> do
118+ forkServer "127.0.0.1" clientPort 5 throw $ \sock -> do
118 -- Initial setup
119 -- Initial setup
119 0 <- recvInt32 sock :: IO Int
120 0 <- recvInt32 sock :: IO Int
120 _ <- recvWithLength sock
121 _ <- recvWithLength sock
@@ -140,7 +141,7 @@ testEarlyDisconnect = do
140 N.sClose sock
141 N.sClose sock
141
142
142 -- Connect to the server
143 -- Connect to the server
143− addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")
144+ addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
144 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
145 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
145 N.setSocketOption sock N.ReuseAddr 1
146 N.setSocketOption sock N.ReuseAddr 1
146 N.connect sock (N.addrAddress addr)
147 N.connect sock (N.addrAddress addr)
@@ -155,20 +156,134 @@ testEarlyDisconnect = do
155 sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]
156 sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]
156
157
157 -- Close the socket without closing the connection explicitly
158 -- Close the socket without closing the connection explicitly
158− -- The server should still receive a ConnectionClosed message
159+ -- The server should receive an error event
159 N.sClose sock
160 N.sClose sock
160
161
162+-- | Test the behaviour of a premature CloseSocket request
163+testEarlyCloseSocket :: IO N.ServiceName -> IO ()
164+testEarlyCloseSocket nextPort = do
165+ clientAddr <- newEmptyMVar
166+ serverAddr <- newEmptyMVar
167+ serverDone <- newEmptyMVar
168+
169+ tlog "testEarlyDisconnect"
170+ forkTry $ server serverAddr clientAddr serverDone
171+ forkTry $ client serverAddr clientAddr
172+
173+ takeMVar serverDone
174+ where
175+ server :: MVar EndPointAddress -> MVar EndPointAddress -> MVar () -> IO ()
176+ server serverAddr clientAddr serverDone = do
177+ tlog "Server"
178+ Right transport <- nextPort >>= createTransport "127.0.0.1"
179+ Right endpoint <- newEndPoint transport
180+ putMVar serverAddr (address endpoint)
181+ theirAddr <- readMVar clientAddr
182+
183+ -- TEST 1: they connect to us, then send a CloseSocket. Since we don't
184+ -- have any outgoing connections, this means we will agree to close the
185+ -- socket
186+ do
187+ ConnectionOpened cid _ addr <- receive endpoint
188+ True <- return $ addr == theirAddr
189+
190+ ConnectionClosed cid' <- receive endpoint
191+ True <- return $ cid' == cid
192+
193+ return ()
194+
195+ -- TEST 2: after they dropped their connection to us, we now try to
196+ -- establish a connection to them. This should re-establish the broken
197+ -- TCP connection.
198+ tlog "Trying to connect to client"
199+ Right conn <- connect endpoint theirAddr ReliableOrdered
200+
201+ -- TEST 3: To test the connection, we do a simple ping test; as before,
202+ -- however, the remote client won't close the connection nicely but just
203+ -- sends a CloseSocket -- except that now we *do* have outgoing
204+ -- connections, so we won't agree and hence will receive an error when
205+ -- the socket gets closed
206+ do
207+ Right () <- send conn ["ping"]
208+ ConnectionOpened cid _ addr <- receive endpoint ; True <- return $ addr == theirAddr
209+ Received cid' ["pong"] <- receive endpoint ; True <- return $ cid' == cid
210+ ConnectionClosed cid'' <- receive endpoint ; True <- return $ cid'' == cid
211+ ErrorEvent (ErrorEventConnectionLost addr' []) <- receive endpoint ; True <- return $ addr' == theirAddr
212+ return ()
213+
214+ -- TEST 4: A subsequent send on an already-open connection will now break
215+ Left (FailedWith SendConnectionClosed _) <- send conn ["ping2"]
216+
217+ -- *Pfew*
218+ putMVar serverDone ()
219+
220+ client :: MVar EndPointAddress -> MVar EndPointAddress -> IO ()
221+ client serverAddr clientAddr = do
222+ tlog "Client"
223+ clientPort <- nextPort
224+ let ourAddress = encodeEndPointAddress "127.0.0.1" clientPort 0
225+ putMVar clientAddr ourAddress
226+ (serverHost, serverPort, endPointIx) <- fromJust . decodeEndPointAddress <$> readMVar serverAddr
227+
228+ -- Listen for incoming messages
229+ forkServer "127.0.0.1" clientPort 5 throw $ \sock -> do
230+ -- Initial setup
231+ 0 <- recvInt32 sock :: IO Int
232+ _ <- recvWithLength sock
233+ sendMany sock [encodeInt32 ConnectionRequestAccepted]
234+
235+ -- Server requests a logical connection
236+ RequestConnectionId <- toEnum <$> (recvInt32 sock :: IO Int)
237+ reqId <- recvInt32 sock :: IO Int
238+ sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 (10001 :: Int)])
239+
240+ -- Server sends a message
241+ 10001 <- recvInt32 sock :: IO Int
242+ ["ping"] <- recvWithLength sock
243+
244+ -- Reply
245+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 (10002 :: Int)]
246+ ControlResponse <- toEnum <$> (recvInt32 sock :: IO Int)
247+ 10002 <- recvInt32 sock :: IO Int
248+ [cid] <- recvWithLength sock
249+ sendMany sock (cid : prependLength ["pong"])
250+
251+ -- Send a CloseSocket even though there are still connections *in both
252+ -- directions*
253+ sendMany sock [encodeInt32 CloseSocket]
254+ N.sClose sock
255+
256+ -- 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
266+
267+ -- Request a new connection, but don't wait for the response
268+ let reqId = 0 :: Int32
269+ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]
270+
271+ -- Send a CloseSocket without sending a closeconnecton
272+ -- The server should still receive a ConnectionClosed message
273+ sendMany sock [encodeInt32 CloseSocket]
274+ N.sClose sock
161
275
162-- | Test the creation of a transport with an invalid address
276-- | Test the creation of a transport with an invalid address
163−testInvalidAddress :: IO ()
277+testInvalidAddress :: IO N.ServiceName -> IO ()
164−testInvalidAddress = do
278+testInvalidAddress nextPort = do
165− Left _ <- createTransport "invalidHostName" "8082"
279+ Left _ <- nextPort >>= createTransport "invalidHostName"
166 return ()
280 return ()
167
281
168-- | Test connecting to invalid or non-existing endpoints
282-- | Test connecting to invalid or non-existing endpoints
169−testInvalidConnect :: IO ()
283+testInvalidConnect :: IO N.ServiceName -> IO ()
170−testInvalidConnect = do
284+testInvalidConnect nextPort = do
285+ port <- nextPort
171− Right transport <- createTransport "127.0.0.1" "8083"
286+ Right transport <- createTransport "127.0.0.1" port
172 Right endpoint <- newEndPoint transport
287 Right endpoint <- newEndPoint transport
173
288
174 -- Syntax error in the endpoint address
289 -- Syntax error in the endpoint address
@@ -177,26 +292,26 @@ testInvalidConnect = do
177
292
178 -- Syntax connect, but invalid hostname (TCP address lookup failure)
293 -- Syntax connect, but invalid hostname (TCP address lookup failure)
179 Left (FailedWith ConnectInvalidAddress _) <-
294 Left (FailedWith ConnectInvalidAddress _) <-
180− connect endpoint (EndPointAddress "invalidHost:port:0") ReliableOrdered
295+ connect endpoint (encodeEndPointAddress "invalidHost" "port" 0) ReliableOrdered
181
296
182 -- TCP address correct, but nobody home at that address
297 -- TCP address correct, but nobody home at that address
183 Left (FailedWith ConnectFailed _) <-
298 Left (FailedWith ConnectFailed _) <-
184− connect endpoint (EndPointAddress "127.0.0.1:9000:0") ReliableOrdered
299+ connect endpoint (encodeEndPointAddress "127.0.0.1" "9000" 0) ReliableOrdered
185
300
186 -- Valid TCP address but invalid endpoint number
301 -- Valid TCP address but invalid endpoint number
187 Left (FailedWith ConnectInvalidAddress _) <-
302 Left (FailedWith ConnectInvalidAddress _) <-
188− connect endpoint (EndPointAddress "127.0.0.1:8083:1") ReliableOrdered
303+ connect endpoint (encodeEndPointAddress "127.0.0.1" port 1) ReliableOrdered
189
304
190 return ()
305 return ()
191
306
192-- | Test that an endpoint can ignore CloseSocket requests (in "reality" this
307-- | Test that an endpoint can ignore CloseSocket requests (in "reality" this
193-- would happen when the endpoint sends a new connection request before
308-- would happen when the endpoint sends a new connection request before
194-- receiving an (already underway) CloseSocket request)
309-- receiving an (already underway) CloseSocket request)
195−testIgnoreCloseSocket :: IO ()
310+testIgnoreCloseSocket :: IO N.ServiceName -> IO ()
196−testIgnoreCloseSocket = do
311+testIgnoreCloseSocket nextPort = do
197 serverAddr <- newEmptyMVar
312 serverAddr <- newEmptyMVar
198 clientDone <- newEmptyMVar
313 clientDone <- newEmptyMVar
199− Right transport <- createTransport "127.0.0.1" "8084"
314+ Right transport <- nextPort >>= createTransport "127.0.0.1"
200
315
201 forkTry $ server transport serverAddr
316 forkTry $ server transport serverAddr
202 forkTry $ client transport serverAddr clientDone
317 forkTry $ client transport serverAddr clientDone
@@ -232,15 +347,15 @@ testIgnoreCloseSocket = do
232 tlog "Client"
347 tlog "Client"
233 Right endpoint <- newEndPoint transport
348 Right endpoint <- newEndPoint transport
234 let EndPointAddress myAddress = address endpoint
349 let EndPointAddress myAddress = address endpoint
350+ (serverHost, serverPort, endPointIx) <- readMVar serverAddr
235
351
236 -- Connect to the server
352 -- Connect to the server
237− addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8084")
353+ addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
238 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
354 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
239 N.setSocketOption sock N.ReuseAddr 1
355 N.setSocketOption sock N.ReuseAddr 1
240 N.connect sock (N.addrAddress addr)
356 N.connect sock (N.addrAddress addr)
241
357
242 tlog "Connecting to endpoint"
358 tlog "Connecting to endpoint"
243− (_, _, endPointIx) <- readMVar serverAddr
244 sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])
359 sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])
245
360
246 -- Wait for acknowledgement
361 -- Wait for acknowledgement
@@ -283,12 +398,13 @@ testIgnoreCloseSocket = do
283-- | Like 'testIgnoreSocket', but now the server requests a connection after the
398-- | Like 'testIgnoreSocket', but now the server requests a connection after the
284-- client closed their connection. In the meantime, the server will have sent a
399-- client closed their connection. In the meantime, the server will have sent a
285-- CloseSocket request to the client, and must block until the client responds.
400-- CloseSocket request to the client, and must block until the client responds.
286−testBlockAfterCloseSocket :: IO ()
401+testBlockAfterCloseSocket :: IO N.ServiceName -> IO ()
287−testBlockAfterCloseSocket = do
402+testBlockAfterCloseSocket nextPort = do
288 serverAddr <- newEmptyMVar
403 serverAddr <- newEmptyMVar
289 clientAddr <- newEmptyMVar
404 clientAddr <- newEmptyMVar
290 clientDone <- newEmptyMVar
405 clientDone <- newEmptyMVar
406+ port <- nextPort
291− Right transport <- createTransport "127.0.0.1" "8085"
407+ Right transport <- createTransport "127.0.0.1" port
292
408
293 forkTry $ server transport serverAddr clientAddr
409 forkTry $ server transport serverAddr clientAddr
294 forkTry $ client transport serverAddr clientAddr clientDone
410 forkTry $ client transport serverAddr clientAddr clientDone
@@ -321,15 +437,15 @@ testBlockAfterCloseSocket = do
321 Right endpoint <- newEndPoint transport
437 Right endpoint <- newEndPoint transport
322 putMVar clientAddr (address endpoint)
438 putMVar clientAddr (address endpoint)
323 let EndPointAddress myAddress = address endpoint
439 let EndPointAddress myAddress = address endpoint
440+ (serverHost, serverPort, endPointIx) <- readMVar serverAddr
324
441
325 -- Connect to the server
442 -- Connect to the server
326− addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8084")
443+ addr:_ <- N.getAddrInfo Nothing (Just serverHost) (Just serverPort)
327 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
444 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
328 N.setSocketOption sock N.ReuseAddr 1
445 N.setSocketOption sock N.ReuseAddr 1
329 N.connect sock (N.addrAddress addr)
446 N.connect sock (N.addrAddress addr)
330
447
331 tlog "Connecting to endpoint"
448 tlog "Connecting to endpoint"
332− (_, _, endPointIx) <- readMVar serverAddr
333 sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])
449 sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])
334
450
335 -- Wait for acknowledgement
451 -- Wait for acknowledgement
@@ -369,12 +485,16 @@ testBlockAfterCloseSocket = do
369
485
370main :: IO ()
486main :: IO ()
371main = do
487main = do
488+ portMVar <- newEmptyMVar
489+ forkTry $ forM_ ([10080 .. 10087] :: [Int]) $ putMVar portMVar . show
490+ let nextPort = do Just port <- tryTakeMVar portMVar ; return port
372 tryIO $ runTests
491 tryIO $ runTests
373− [ ("EarlyDisconnect", testEarlyDisconnect)
492+ [ ("EarlyDisconnect", testEarlyDisconnect nextPort)
374− , ("InvalidAddress", testInvalidAddress)
493+ , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)
375− , ("InvalidConnect", testInvalidConnect)
376− , ("IgnoreCloseSocket", testIgnoreCloseSocket)
494+ , ("IgnoreCloseSocket", testIgnoreCloseSocket nextPort)
377− , ("BlockAfterCloseSocket", testBlockAfterCloseSocket)
495+ , ("BlockAfterCloseSocket", testBlockAfterCloseSocket nextPort)
496+ , ("InvalidAddress", testInvalidAddress nextPort)
497+ , ("InvalidConnect", testInvalidConnect nextPort)
378 ]
498 ]
379 Right transport <- createTransport "127.0.0.1" "8080"
499 Right transport <- createTransport "127.0.0.1" "8080"
380 testTransport transport
500 testTransport transport
network-transport-2/tests/Traced.hs+4
@@ -129,6 +129,10 @@ instance (Traceable a, Traceable b, Traceable c) => Traceable (a, b, c) where
129 (Nothing, Just t2, Just t3) -> traceShow (t2, t3)
129 (Nothing, Just t2, Just t3) -> traceShow (t2, t3)
130 (Just t1, Just t2, Just t3) -> traceShow (t1, t2, t3)
130 (Just t1, Just t2, Just t3) -> traceShow (t1, t2, t3)
131
131
132+instance Traceable a => Traceable (Maybe a) where
133+ trace Nothing = traceShow (Nothing :: Maybe ())
134+ trace (Just x) = mapShowable (Showable . Just) <$> trace x
135+
132instance Traceable a => Traceable [a] where
136instance Traceable a => Traceable [a] where
133 trace = traceShow . catMaybes . map trace
137 trace = traceShow . catMaybes . map trace
134
138