@@ -0,0 +1,956 @@
1+{-# LANGUAGE RebindableSyntax #-}
2+module TestTransport where
4+import Prelude hiding (catch, (>>=), (>>), return, fail)
5+import TestAuxiliary (forkTry, runTests, trySome, randomThreadDelay)
6+import Control.Concurrent (forkIO, killThread, yield)
7+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar, tryTakeMVar, modifyMVar_, newMVar)
8+import Control.Exception (evaluate, throw, throwIO, bracket)
9+import Control.Monad (replicateM, replicateM_, when, guard, forM_, unless)
10+import Control.Monad.Error ()
11+import Control.Applicative ((<$>))
12+import Network.Transport
13+import Network.Transport.Internal (tlog, tryIO, timeoutMaybe)
14+import Network.Transport.Util (spawn)
15+import System.Random (randomIO)
16+import Data.ByteString (ByteString)
17+import Data.ByteString.Char8 (pack)
19+import qualified Data.Map as Map (empty, insert, delete, findWithDefault, adjust, null, toList, map)
20+import Data.String (fromString)
21+import Data.List (permutations)
24+-- | Server that echoes messages straight back to the origin endpoint.
25+echoServer :: EndPoint -> IO ()
26+echoServer endpoint = do
29+ go :: Map ConnectionId Connection -> IO ()
31+ event <- receive endpoint
33+ ConnectionOpened cid rel addr -> do
34+ tlog $ "Opened new connection " ++ show cid
35+ Right conn <- connect endpoint addr rel defaultConnectHints
36+ go (Map.insert cid conn cs)
37+ Received cid payload -> do
38+ send (Map.findWithDefault (error $ "Received: Invalid cid " ++ show cid) cid cs) payload
40+ ConnectionClosed cid -> do
41+ tlog $ "Close connection " ++ show cid
42+ close (Map.findWithDefault (error $ "ConnectionClosed: Invalid cid " ++ show cid) cid cs)
43+ go (Map.delete cid cs)
44+ ReceivedMulticast _ _ ->
48+ putStrLn $ "Echo server received error event: " ++ show event
52+-- | Ping client used in a few tests
53+ping :: EndPoint -> EndPointAddress -> Int -> ByteString -> IO ()
54+ping endpoint server numPings msg = do
55+ -- Open connection to the server
56+ tlog "Connect to echo server"
57+ Right conn <- connect endpoint server ReliableOrdered defaultConnectHints
59+ -- Wait for the server to open reply connection
60+ tlog "Wait for ConnectionOpened message"
61+ ConnectionOpened cid _ _ <- receive endpoint
63+ -- Send pings and wait for reply
64+ tlog "Send ping and wait for reply"
65+ replicateM_ numPings $ do
67+ Received cid' [reply] <- receive endpoint ; True <- return $ cid == cid' && reply == msg
70+ -- Close the connection
71+ tlog "Close the connection"
74+ -- Wait for the server to close its connection to us
75+ tlog "Wait for ConnectionClosed message"
76+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid'
79+ tlog "Ping client done"
82+testPingPong :: Transport -> Int -> IO ()
83+testPingPong transport numPings = do
84+ tlog "Starting ping pong test"
85+ server <- spawn transport echoServer
86+ result <- newEmptyMVar
91+ Right endpoint <- newEndPoint transport
92+ ping endpoint server numPings "ping"
97+-- | Test that endpoints don't get confused
98+testEndPoints :: Transport -> Int -> IO ()
99+testEndPoints transport numPings = do
100+ server <- spawn transport echoServer
101+ dones <- replicateM 2 newEmptyMVar
103+ forM_ (zip dones ['A'..]) $ \(done, name) -> forkTry $ do
104+ let name' :: ByteString
106+ Right endpoint <- newEndPoint transport
107+ tlog $ "Ping client " ++ show name' ++ ": " ++ show (address endpoint)
108+ ping endpoint server numPings name'
111+ forM_ dones takeMVar
113+-- Test that connections don't get confused
114+testConnections :: Transport -> Int -> IO ()
115+testConnections transport numPings = do
116+ server <- spawn transport echoServer
117+ result <- newEmptyMVar
121+ Right endpoint <- newEndPoint transport
123+ -- Open two connections to the server
124+ Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints
125+ ConnectionOpened serv1 _ _ <- receive endpoint
127+ Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints
128+ ConnectionOpened serv2 _ _ <- receive endpoint
130+ -- One thread to send "pingA" on the first connection
131+ forkTry $ replicateM_ numPings $ send conn1 ["pingA"]
133+ -- One thread to send "pingB" on the second connection
134+ forkTry $ replicateM_ numPings $ send conn2 ["pingB"]
136+ -- Verify server responses
137+ let verifyResponse 0 = putMVar result ()
138+ verifyResponse n = do
139+ event <- receive endpoint
141+ Received cid [payload] -> do
142+ when (cid == serv1 && payload /= "pingA") $ error "Wrong message"
143+ when (cid == serv2 && payload /= "pingB") $ error "Wrong message"
144+ verifyResponse (n - 1)
147+ verifyResponse (2 * numPings)
151+-- | Test that closing one connection does not close the other
152+testCloseOneConnection :: Transport -> Int -> IO ()
153+testCloseOneConnection transport numPings = do
154+ server <- spawn transport echoServer
155+ result <- newEmptyMVar
159+ Right endpoint <- newEndPoint transport
161+ -- Open two connections to the server
162+ Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints
163+ ConnectionOpened serv1 _ _ <- receive endpoint
165+ Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints
166+ ConnectionOpened serv2 _ _ <- receive endpoint
168+ -- One thread to send "pingA" on the first connection
170+ replicateM_ numPings $ send conn1 ["pingA"]
173+ -- One thread to send "pingB" on the second connection
174+ forkTry $ replicateM_ (numPings * 2) $ send conn2 ["pingB"]
176+ -- Verify server responses
177+ let verifyResponse 0 = putMVar result ()
178+ verifyResponse n = do
179+ event <- receive endpoint
181+ Received cid [payload] -> do
182+ when (cid == serv1 && payload /= "pingA") $ error "Wrong message"
183+ when (cid == serv2 && payload /= "pingB") $ error "Wrong message"
184+ verifyResponse (n - 1)
187+ verifyResponse (3 * numPings)
191+-- | Test that if A connects to B and B connects to A, B can still send to A after
192+-- A closes its connection to B (for instance, in the TCP transport, the socket pair
193+-- connecting A and B should not yet be closed).
194+testCloseOneDirection :: Transport -> Int -> IO ()
195+testCloseOneDirection transport numPings = do
196+ addrA <- newEmptyMVar
197+ addrB <- newEmptyMVar
198+ doneA <- newEmptyMVar
199+ doneB <- newEmptyMVar
204+ Right endpoint <- newEndPoint transport
205+ tlog (show (address endpoint))
206+ putMVar addrA (address endpoint)
210+ Right conn <- readMVar addrB >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
212+ -- Wait for B to connect to us
214+ ConnectionOpened cid _ _ <- receive endpoint
217+ tlog "Send pings to B"
218+ replicateM_ numPings $ send conn ["ping"]
220+ -- Close our connection to B
221+ tlog "Close connection"
224+ -- Wait for B's pongs
225+ tlog "Wait for pongs from B"
226+ replicateM_ numPings $ do Received _ _ <- receive endpoint ; return ()
228+ -- Wait for B to close it's connection to us
229+ tlog "Wait for B to close connection"
230+ ConnectionClosed cid' <- receive endpoint
240+ Right endpoint <- newEndPoint transport
241+ tlog (show (address endpoint))
242+ putMVar addrB (address endpoint)
244+ -- Wait for A to connect
245+ tlog "Wait for A to connect"
246+ ConnectionOpened cid _ _ <- receive endpoint
250+ Right conn <- readMVar addrA >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
252+ -- Wait for A's pings
253+ tlog "Wait for pings from A"
254+ replicateM_ numPings $ do Received _ _ <- receive endpoint ; return ()
256+ -- Wait for A to close it's connection to us
257+ tlog "Wait for A to close connection"
258+ ConnectionClosed cid' <- receive endpoint
262+ tlog "Send pongs to A"
263+ replicateM_ numPings $ send conn ["pong"]
265+ -- Close our connection to A
266+ tlog "Close connection to A"
273+ mapM_ takeMVar [doneA, doneB]
275+-- | Collect events and order them by connection ID
276+collect :: EndPoint -> Maybe Int -> Maybe Int -> IO [(ConnectionId, [[ByteString]])]
277+collect endPoint maxEvents timeout = go maxEvents Map.empty Map.empty
279+ -- TODO: for more serious use of this function we'd need to make these arguments strict
280+ go (Just 0) open closed = finish open closed
281+ go n open closed = do
282+ mEvent <- tryIO . timeoutMaybe timeout (userError "timeout") $ receive endPoint
284+ Left _ -> finish open closed
286+ let n' = (\x -> x - 1) <$> n
288+ ConnectionOpened cid _ _ ->
289+ go n' (Map.insert cid [] open) closed
290+ ConnectionClosed cid ->
291+ let list = Map.findWithDefault (error "Invalid ConnectionClosed") cid open in
292+ go n' (Map.delete cid open) (Map.insert cid list closed)
294+ go n' (Map.adjust (msg :) cid open) closed
295+ ReceivedMulticast _ _ ->
296+ fail "Unexpected multicast"
298+ fail "Unexpected error"
300+ fail "Unexpected endpoint closure"
302+ finish open closed =
304+ then return . Map.toList . Map.map reverse $ closed
305+ else fail $ "Open connections: " ++ show (map fst . Map.toList $ open)
307+-- | Open connection, close it, then reopen it
308+-- (In the TCP transport this means the socket will be closed, then reopened)
310+-- Note that B cannot expect to receive all of A's messages on the first connection
311+-- before receiving the messages on the second connection. What might (and sometimes
312+-- does) happen is that finishes sending all of its messages on the first connection
313+-- (in the TCP transport, the first socket pair) while B is behind on reading _from_
314+-- this connection (socket pair) -- the messages are "in transit" on the network
315+-- (these tests are done on localhost, so there are in some OS buffer). Then when
316+-- A opens the second connection (socket pair) B will spawn a new thread for this
317+-- connection, and hence might start interleaving messages from the first and second
320+-- This is correct behaviour, however: the transport API guarantees reliability and
321+-- ordering _per connection_, but not _across_ connections.
322+testCloseReopen :: Transport -> Int -> IO ()
323+testCloseReopen transport numPings = do
324+ addrB <- newEmptyMVar
325+ doneB <- newEmptyMVar
327+ let numRepeats = 2 :: Int
331+ Right endpoint <- newEndPoint transport
333+ forM_ [1 .. numRepeats] $ \i -> do
336+ Right conn <- readMVar addrB >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
340+ forM_ [1 .. numPings] $ \j -> send conn [pack $ "ping" ++ show i ++ "/" ++ show j]
350+ Right endpoint <- newEndPoint transport
351+ putMVar addrB (address endpoint)
353+ eventss <- collect endpoint (Just (numRepeats * (numPings + 2))) Nothing
355+ forM_ (zip [1 .. numRepeats] eventss) $ \(i, (_, events)) -> do
356+ forM_ (zip [1 .. numPings] events) $ \(j, event) -> do
357+ guard (event == [pack $ "ping" ++ show i ++ "/" ++ show j])
363+-- | Test lots of parallel connection attempts
364+testParallelConnects :: Transport -> Int -> IO ()
365+testParallelConnects transport numPings = do
366+ server <- spawn transport echoServer
367+ done <- newEmptyMVar
369+ Right endpoint <- newEndPoint transport
371+ -- Spawn lots of clients
372+ forM_ [1 .. numPings] $ \i -> forkTry $ do
373+ Right conn <- connect endpoint server ReliableOrdered defaultConnectHints
374+ send conn [pack $ "ping" ++ show i]
375+ send conn [pack $ "ping" ++ show i]
379+ eventss <- collect endpoint (Just (numPings * 4)) Nothing
380+ -- Check that no pings got sent to the wrong connection
381+ forM_ eventss $ \(_, [[ping1], [ping2]]) ->
382+ guard (ping1 == ping2)
387+-- | Test that sending on a closed connection gives an error
388+testSendAfterClose :: Transport -> Int -> IO ()
389+testSendAfterClose transport numRepeats = do
390+ server <- spawn transport echoServer
391+ clientDone <- newEmptyMVar
394+ Right endpoint <- newEndPoint transport
396+ -- We request two lightweight connections
397+ replicateM numRepeats $ do
398+ Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints
399+ Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints
401+ -- Close the second, but leave the first open; then output on the second
402+ -- connection (i.e., on a closed connection while there is still another
405+ Left (TransportError SendClosed _) <- send conn2 ["ping2"]
407+ -- Now close the first connection, and output on it (i.e., output while
408+ -- there are no lightweight connection at all anymore)
410+ Left (TransportError SendClosed _) <- send conn2 ["ping2"]
414+ putMVar clientDone ()
418+-- | Test that closing the same connection twice has no effect
419+testCloseTwice :: Transport -> Int -> IO ()
420+testCloseTwice transport numRepeats = do
421+ server <- spawn transport echoServer
422+ clientDone <- newEmptyMVar
425+ Right endpoint <- newEndPoint transport
427+ replicateM numRepeats $ do
428+ -- We request two lightweight connections
429+ Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints
430+ Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints
432+ -- Close the second one twice
436+ -- Then send a message on the first and close that twice too
440+ -- Verify expected response from the echo server
441+ ConnectionOpened cid1 _ _ <- receive endpoint
442+ ConnectionOpened cid2 _ _ <- receive endpoint
443+ ConnectionClosed cid2' <- receive endpoint ; True <- return $ cid2' == cid2
444+ Received cid1' ["ping"] <- receive endpoint ; True <- return $ cid1' == cid1
445+ ConnectionClosed cid1'' <- receive endpoint ; True <- return $ cid1'' == cid1
449+ putMVar clientDone ()
453+-- | Test that we can connect an endpoint to itself
454+testConnectToSelf :: Transport -> Int -> IO ()
455+testConnectToSelf transport numPings = do
456+ done <- newEmptyMVar
457+ Right endpoint <- newEndPoint transport
459+ tlog "Creating self-connection"
460+ Right conn <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints
462+ tlog "Talk to myself"
464+ -- One thread to write to the endpoint
468+ tlog $ "Sending ping"
469+ replicateM_ numPings $ send conn ["ping"]
471+ tlog $ "Closing connection"
474+ -- And one thread to read
478+ tlog "Waiting for ConnectionOpened"
479+ ConnectionOpened cid _ addr <- receive endpoint ; True <- return $ addr == address endpoint
481+ tlog "Waiting for Received"
482+ replicateM_ numPings $ do
483+ Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid'
486+ tlog "Waiting for ConnectionClosed"
487+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid'
494+-- | Test that we can connect an endpoint to itself multiple times
495+testConnectToSelfTwice :: Transport -> Int -> IO ()
496+testConnectToSelfTwice transport numPings = do
497+ done <- newEmptyMVar
498+ Right endpoint <- newEndPoint transport
500+ tlog "Creating self-connection"
501+ Right conn1 <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints
502+ Right conn2 <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints
504+ tlog "Talk to myself"
506+ -- One thread to write to the endpoint using the first connection
510+ tlog $ "Sending ping"
511+ replicateM_ numPings $ send conn1 ["pingA"]
513+ tlog $ "Closing connection"
516+ -- One thread to write to the endpoint using the second connection
520+ tlog $ "Sending ping"
521+ replicateM_ numPings $ send conn2 ["pingB"]
523+ tlog $ "Closing connection"
526+ -- And one thread to read
530+ [(_, events1), (_, events2)] <- collect endpoint (Just (2 * (numPings + 2))) Nothing
531+ True <- return $ events1 == replicate numPings ["pingA"]
532+ True <- return $ events2 == replicate numPings ["pingB"]
539+-- | Test that we self-connections no longer work once we close our endpoint
541+testCloseSelf :: IO (Either String Transport) -> IO ()
542+testCloseSelf newTransport = do
543+ Right transport <- newTransport
544+ Right endpoint1 <- newEndPoint transport
545+ Right endpoint2 <- newEndPoint transport
546+ Right conn1 <- connect endpoint1 (address endpoint1) ReliableOrdered defaultConnectHints
547+ Right conn2 <- connect endpoint1 (address endpoint1) ReliableOrdered defaultConnectHints
548+ Right conn3 <- connect endpoint2 (address endpoint2) ReliableOrdered defaultConnectHints
550+ -- Close the conneciton and try to send
552+ Left (TransportError SendClosed _) <- send conn1 ["ping"]
554+ -- Close the first endpoint. We should not be able to use the first
555+ -- connection anymore, or open more self connections, but the self connection
556+ -- to the second endpoint should still be fine
557+ closeEndPoint endpoint1
558+ Left (TransportError SendFailed _) <- send conn2 ["ping"]
559+ Left (TransportError ConnectFailed _) <- connect endpoint1 (address endpoint1) ReliableOrdered defaultConnectHints
560+ Right () <- send conn3 ["ping"]
562+ -- Close the transport; now the second should no longer work
563+ closeTransport transport
564+ Left (TransportError SendFailed _) <- send conn3 ["ping"]
565+ Left (TransportError ConnectFailed _) <- connect endpoint2 (address endpoint2) ReliableOrdered defaultConnectHints
569+-- | Test various aspects of 'closeEndPoint'
570+testCloseEndPoint :: Transport -> Int -> IO ()
571+testCloseEndPoint transport _ = do
572+ serverDone <- newEmptyMVar
573+ clientDone <- newEmptyMVar
574+ clientAddr1 <- newEmptyMVar
575+ clientAddr2 <- newEmptyMVar
576+ serverAddr <- newEmptyMVar
580+ Right endpoint <- newEndPoint transport
581+ putMVar serverAddr (address endpoint)
583+ -- First test (see client)
585+ theirAddr <- readMVar clientAddr1
586+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
587+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid'
592+ theirAddr <- readMVar clientAddr2
594+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
595+ Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid'
597+ Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
600+ ConnectionClosed cid'' <- receive endpoint ; True <- return $ cid == cid''
601+ ErrorEvent (TransportError (EventConnectionLost (Just addr') []) _) <- receive endpoint ; True <- return $ addr' == theirAddr
603+ Left (TransportError SendFailed _) <- send conn ["pong2"]
607+ putMVar serverDone ()
611+ theirAddr <- readMVar serverAddr
613+ -- First test: close endpoint with one outgoing but no incoming connections
615+ Right endpoint <- newEndPoint transport
616+ putMVar clientAddr1 (address endpoint)
618+ -- Connect to the server, then close the endpoint without disconnecting explicitly
619+ Right _ <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
620+ closeEndPoint endpoint
621+ EndPointClosed <- receive endpoint
624+ -- Second test: close endpoint with one outgoing and one incoming connection
626+ Right endpoint <- newEndPoint transport
627+ putMVar clientAddr2 (address endpoint)
629+ Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
632+ -- Reply from the server
633+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
634+ Received cid' ["pong"] <- receive endpoint ; True <- return $ cid == cid'
636+ -- Close the endpoint
637+ closeEndPoint endpoint
638+ EndPointClosed <- receive endpoint
640+ -- Attempt to send should fail with connection closed
641+ Left (TransportError SendFailed _) <- send conn ["ping2"]
643+ -- An attempt to close the already closed connection should just return
646+ -- And so should an attempt to connect
647+ Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
651+ putMVar clientDone ()
653+ mapM_ takeMVar [serverDone, clientDone]
655+-- Test closeTransport
657+-- This tests many of the same things that testEndPoint does, and some more
658+testCloseTransport :: IO (Either String Transport) -> IO ()
659+testCloseTransport newTransport = do
660+ serverDone <- newEmptyMVar
661+ clientDone <- newEmptyMVar
662+ clientAddr1 <- newEmptyMVar
663+ clientAddr2 <- newEmptyMVar
664+ serverAddr <- newEmptyMVar
668+ Right transport <- newTransport
669+ Right endpoint <- newEndPoint transport
670+ putMVar serverAddr (address endpoint)
672+ -- Client sets up first endpoint
673+ theirAddr1 <- readMVar clientAddr1
674+ ConnectionOpened cid1 ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr1
676+ -- Client sets up second endpoint
677+ theirAddr2 <- readMVar clientAddr2
679+ ConnectionOpened cid2 ReliableOrdered addr' <- receive endpoint ; True <- return $ addr' == theirAddr2
680+ Received cid2' ["ping"] <- receive endpoint ; True <- return $ cid2' == cid2
682+ Right conn <- connect endpoint theirAddr2 ReliableOrdered defaultConnectHints
685+ -- Client now closes down its transport. We should receive connection closed messages (we don't know the precise order, however)
686+ evs <- replicateM 3 $ receive endpoint
687+ let expected = [ ConnectionClosed cid1
688+ , ConnectionClosed cid2
689+ , ErrorEvent (TransportError (EventConnectionLost (Just theirAddr2) []) "")
691+ True <- return $ any (== expected) (permutations evs)
693+ -- An attempt to send to the endpoint should now fail
694+ Left (TransportError SendFailed _) <- send conn ["pong2"]
696+ putMVar serverDone ()
700+ Right transport <- newTransport
701+ theirAddr <- readMVar serverAddr
703+ -- Set up endpoint with one outgoing but no incoming connections
704+ Right endpoint1 <- newEndPoint transport
705+ putMVar clientAddr1 (address endpoint1)
707+ -- Connect to the server, then close the endpoint without disconnecting explicitly
708+ Right _ <- connect endpoint1 theirAddr ReliableOrdered defaultConnectHints
710+ -- Set up an endpoint with one outgoing and out incoming connection
711+ Right endpoint2 <- newEndPoint transport
712+ putMVar clientAddr2 (address endpoint2)
714+ Right conn <- connect endpoint2 theirAddr ReliableOrdered defaultConnectHints
717+ -- Reply from the server
718+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint2 ; True <- return $ addr == theirAddr
719+ Received cid' ["pong"] <- receive endpoint2 ; True <- return $ cid == cid'
721+ -- Now shut down the entire transport
722+ closeTransport transport
724+ -- Both endpoints should report that they have been closed
725+ EndPointClosed <- receive endpoint1
726+ EndPointClosed <- receive endpoint2
728+ -- Attempt to send should fail with connection closed
729+ Left (TransportError SendFailed _) <- send conn ["ping2"]
731+ -- An attempt to close the already closed connection should just return
734+ -- And so should an attempt to connect on either endpoint
735+ Left (TransportError ConnectFailed _) <- connect endpoint1 theirAddr ReliableOrdered defaultConnectHints
736+ Left (TransportError ConnectFailed _) <- connect endpoint2 theirAddr ReliableOrdered defaultConnectHints
738+ -- And finally, so should an attempt to create a new endpoint
739+ Left (TransportError NewEndPointFailed _) <- newEndPoint transport
741+ putMVar clientDone ()
743+ mapM_ takeMVar [serverDone, clientDone]
745+-- | Remote node attempts to connect to a closed local endpoint
746+testConnectClosedEndPoint :: Transport -> IO ()
747+testConnectClosedEndPoint transport = do
748+ serverAddr <- newEmptyMVar
749+ serverClosed <- newEmptyMVar
750+ clientDone <- newEmptyMVar
754+ Right endpoint <- newEndPoint transport
755+ putMVar serverAddr (address endpoint)
757+ closeEndPoint endpoint
758+ putMVar serverClosed ()
762+ Right endpoint <- newEndPoint transport
763+ readMVar serverClosed
765+ Left (TransportError ConnectNotFound _) <- readMVar serverAddr >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
767+ putMVar clientDone ()
771+-- | We should receive an exception when doing a 'receive' after we have been
772+-- notified that an endpoint has been closed
773+testExceptionOnReceive :: IO (Either String Transport) -> IO ()
774+testExceptionOnReceive newTransport = do
775+ Right transport <- newTransport
777+ -- Test one: when we close an endpoint specifically
778+ Right endpoint1 <- newEndPoint transport
779+ closeEndPoint endpoint1
780+ EndPointClosed <- receive endpoint1
781+ Left _ <- trySome (receive endpoint1 >>= evaluate)
783+ -- Test two: when we close the entire transport
784+ Right endpoint2 <- newEndPoint transport
785+ closeTransport transport
786+ EndPointClosed <- receive endpoint2
787+ Left _ <- trySome (receive endpoint2 >>= evaluate)
791+-- | Test what happens when the argument to 'send' is an exceptional value
792+testSendException :: IO (Either String Transport) -> IO ()
793+testSendException newTransport = do
794+ Right transport <- newTransport
795+ Right endpoint1 <- newEndPoint transport
796+ Right endpoint2 <- newEndPoint transport
798+ -- Connect endpoint1 to endpoint2
799+ Right conn <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints
800+ ConnectionOpened _ _ _ <- receive endpoint2
802+ -- Send an exceptional value
803+ Left (TransportError SendFailed _) <- send conn (throw $ userError "uhoh")
805+ -- This will have been as a failure to send by endpoint1, which will
806+ -- therefore have closed the socket. In turn this will have caused endpoint2
807+ -- to report that the connection was lost
808+ ErrorEvent (TransportError (EventConnectionLost _ []) _) <- receive endpoint1
809+ ErrorEvent (TransportError (EventConnectionLost _ [_]) _) <- receive endpoint2
811+ -- A new connection will re-establish the connection
812+ Right conn2 <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints
816+ ConnectionOpened _ _ _ <- receive endpoint2
817+ Received _ ["ping"] <- receive endpoint2
818+ ConnectionClosed _ <- receive endpoint2
822+-- | If threads get killed while executing a 'connect', 'send', or 'close', this
823+-- should not affect other threads.
825+-- The intention of this test is to see what happens when a asynchronous
826+-- exception happes _while executing a send_. This is exceedingly difficult to
827+-- guarantee, however. Hence we run a large number of tests and insert random
828+-- thread delays -- and even then it might not happen. Moreover, it will only
829+-- happen when we run on multiple cores.
830+testKill :: IO (Either String Transport) -> Int -> IO ()
831+testKill newTransport numThreads = do
832+ Right transport1 <- newTransport
833+ Right transport2 <- newTransport
834+ Right endpoint1 <- newEndPoint transport1
835+ Right endpoint2 <- newEndPoint transport2
837+ threads <- replicateM numThreads . forkIO $ do
838+ randomThreadDelay 100
839+ bracket (connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints)
840+ -- Note that we should not insert a randomThreadDelay into the
841+ -- exception handler itself as this means that the exception handler
842+ -- could be interrupted and we might not close
843+ (\(Right conn) -> close conn)
844+ (\(Right conn) -> do randomThreadDelay 100
845+ Right () <- send conn ["ping"]
846+ randomThreadDelay 100)
848+ numAlive <- newMVar (0 :: Int)
850+ -- Kill half of those threads
851+ forkIO . forM_ threads $ \tid -> do
852+ shouldKill <- randomIO
854+ then randomThreadDelay 600 >> killThread tid
855+ else modifyMVar_ numAlive (return . (+ 1))
857+ -- Since it is impossible to predict when the kill exactly happens, we don't
858+ -- know how many connects were opened and how many pings were sent. But we
859+ -- should not have any open connections (if we do, collect will throw an
860+ -- error) and we should have at least the number of pings equal to the number
861+ -- of threads we did *not* kill
862+ eventss <- collect endpoint2 Nothing (Just 1000000)
863+ let actualPings = sum . map (length . snd) $ eventss
864+ expectedPings <- takeMVar numAlive
865+ unless (actualPings >= expectedPings) $
866+ throwIO (userError "Missing pings")
868+-- print (actualPings, expectedPings)
871+-- | Set up conditions with a high likelyhood of "crossing" (for transports
872+-- that multiplex lightweight connections across heavyweight connections)
873+testCrossing :: Transport -> Int -> IO ()
874+testCrossing transport numRepeats = do
875+ [aAddr, bAddr] <- replicateM 2 newEmptyMVar
876+ [aDone, bDone] <- replicateM 2 newEmptyMVar
877+ [aTimeout, bTimeout] <- replicateM 2 newEmptyMVar
880+ let hints = defaultConnectHints {
881+ connectTimeout = Just 5000000
886+ Right endpoint <- newEndPoint transport
887+ putMVar aAddr (address endpoint)
888+ theirAddress <- readMVar bAddr
890+ replicateM_ numRepeats $ do
891+ takeMVar go >> yield
892+ -- Because we are creating lots of connections, it's possible that
893+ -- connect times out (for instance, in the TCP transport,
894+ -- Network.Socket.connect may time out). We shouldn't regard this as an
895+ -- error in the Transport, though.
896+ connectResult <- connect endpoint theirAddress ReliableOrdered hints
897+ case connectResult of
898+ Right conn -> close conn
899+ Left (TransportError ConnectTimeout _) -> putMVar aTimeout ()
900+ Left (TransportError ConnectFailed _) -> readMVar bTimeout
901+ Left err -> throwIO . userError $ "testCrossed: " ++ show err
906+ Right endpoint <- newEndPoint transport
907+ putMVar bAddr (address endpoint)
908+ theirAddress <- readMVar aAddr
910+ replicateM_ numRepeats $ do
911+ takeMVar go >> yield
912+ connectResult <- connect endpoint theirAddress ReliableOrdered hints
913+ case connectResult of
914+ Right conn -> close conn
915+ Left (TransportError ConnectTimeout _) -> putMVar bTimeout ()
916+ Left (TransportError ConnectFailed _) -> readMVar aTimeout
917+ Left err -> throwIO . userError $ "testCrossed: " ++ show err
921+ forM_ [1 .. numRepeats] $ \_i -> do
922+ -- putStrLn $ "Round " ++ show _i
923+ tryTakeMVar aTimeout
924+ tryTakeMVar bTimeout
931+testTransport :: IO (Either String Transport) -> IO ()
932+testTransport newTransport = do
933+ Right transport <- newTransport
935+ [ ("PingPong", testPingPong transport numPings)
936+ , ("EndPoints", testEndPoints transport numPings)
937+ , ("Connections", testConnections transport numPings)
938+ , ("CloseOneConnection", testCloseOneConnection transport numPings)
939+ , ("CloseOneDirection", testCloseOneDirection transport numPings)
940+ , ("CloseReopen", testCloseReopen transport numPings)
941+ , ("ParallelConnects", testParallelConnects transport numPings)
942+ , ("SendAfterClose", testSendAfterClose transport 1000)
943+ , ("Crossing", testCrossing transport 100)
944+ , ("CloseTwice", testCloseTwice transport 100)
945+ , ("ConnectToSelf", testConnectToSelf transport numPings)
946+ , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)
947+ , ("CloseSelf", testCloseSelf newTransport)
948+ , ("CloseEndPoint", testCloseEndPoint transport numPings)
949+ , ("CloseTransport", testCloseTransport newTransport)
950+ , ("ConnectClosedEndPoint", testConnectClosedEndPoint transport)
951+ , ("ExceptionOnReceive", testExceptionOnReceive newTransport)
952+ , ("SendException", testSendException newTransport)
953+ , ("Kill", testKill newTransport 10000)
956+ numPings = 10000 :: Int