@@ -0,0 +1,956 @@
1+{-# LANGUAGE RebindableSyntax #-}
2+module TestTransport where
3+
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)
18+import Data.Map (Map)
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)
22+import Traced
23+
24+-- | Server that echoes messages straight back to the origin endpoint.
25+echoServer :: EndPoint -> IO ()
26+echoServer endpoint = do
27+ go Map.empty
28+ where
29+ go :: Map ConnectionId Connection -> IO ()
30+ go cs = do
31+ event <- receive endpoint
32+ case event of
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
39+ go cs
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 _ _ ->
45+ -- Ignore
46+ go cs
47+ ErrorEvent _ ->
48+ putStrLn $ "Echo server received error event: " ++ show event
49+ EndPointClosed ->
50+ return ()
51+
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
58+
59+ -- Wait for the server to open reply connection
60+ tlog "Wait for ConnectionOpened message"
61+ ConnectionOpened cid _ _ <- receive endpoint
62+
63+ -- Send pings and wait for reply
64+ tlog "Send ping and wait for reply"
65+ replicateM_ numPings $ do
66+ send conn [msg]
67+ Received cid' [reply] <- receive endpoint ; True <- return $ cid == cid' && reply == msg
68+ return ()
69+
70+ -- Close the connection
71+ tlog "Close the connection"
72+ close conn
73+
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'
77+
78+ -- Done
79+ tlog "Ping client done"
80+
81+-- | Basic ping test
82+testPingPong :: Transport -> Int -> IO ()
83+testPingPong transport numPings = do
84+ tlog "Starting ping pong test"
85+ server <- spawn transport echoServer
86+ result <- newEmptyMVar
87+
88+ -- Client
89+ forkTry $ do
90+ tlog "Ping client"
91+ Right endpoint <- newEndPoint transport
92+ ping endpoint server numPings "ping"
93+ putMVar result ()
94+
95+ takeMVar result
96+
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
102+
103+ forM_ (zip dones ['A'..]) $ \(done, name) -> forkTry $ do
104+ let name' :: ByteString
105+ name' = pack [name]
106+ Right endpoint <- newEndPoint transport
107+ tlog $ "Ping client " ++ show name' ++ ": " ++ show (address endpoint)
108+ ping endpoint server numPings name'
109+ putMVar done ()
110+
111+ forM_ dones takeMVar
112+
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
118+
119+ -- Client
120+ forkTry $ do
121+ Right endpoint <- newEndPoint transport
122+
123+ -- Open two connections to the server
124+ Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints
125+ ConnectionOpened serv1 _ _ <- receive endpoint
126+
127+ Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints
128+ ConnectionOpened serv2 _ _ <- receive endpoint
129+
130+ -- One thread to send "pingA" on the first connection
131+ forkTry $ replicateM_ numPings $ send conn1 ["pingA"]
132+
133+ -- One thread to send "pingB" on the second connection
134+ forkTry $ replicateM_ numPings $ send conn2 ["pingB"]
135+
136+ -- Verify server responses
137+ let verifyResponse 0 = putMVar result ()
138+ verifyResponse n = do
139+ event <- receive endpoint
140+ case event of
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)
145+ _ ->
146+ verifyResponse n
147+ verifyResponse (2 * numPings)
148+
149+ takeMVar result
150+
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
156+
157+ -- Client
158+ forkTry $ do
159+ Right endpoint <- newEndPoint transport
160+
161+ -- Open two connections to the server
162+ Right conn1 <- connect endpoint server ReliableOrdered defaultConnectHints
163+ ConnectionOpened serv1 _ _ <- receive endpoint
164+
165+ Right conn2 <- connect endpoint server ReliableOrdered defaultConnectHints
166+ ConnectionOpened serv2 _ _ <- receive endpoint
167+
168+ -- One thread to send "pingA" on the first connection
169+ forkTry $ do
170+ replicateM_ numPings $ send conn1 ["pingA"]
171+ close conn1
172+
173+ -- One thread to send "pingB" on the second connection
174+ forkTry $ replicateM_ (numPings * 2) $ send conn2 ["pingB"]
175+
176+ -- Verify server responses
177+ let verifyResponse 0 = putMVar result ()
178+ verifyResponse n = do
179+ event <- receive endpoint
180+ case event of
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)
185+ _ ->
186+ verifyResponse n
187+ verifyResponse (3 * numPings)
188+
189+ takeMVar result
190+
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
200+
201+ -- A
202+ forkTry $ do
203+ tlog "A"
204+ Right endpoint <- newEndPoint transport
205+ tlog (show (address endpoint))
206+ putMVar addrA (address endpoint)
207+
208+ -- Connect to B
209+ tlog "Connect to B"
210+ Right conn <- readMVar addrB >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
211+
212+ -- Wait for B to connect to us
213+ tlog "Wait for B"
214+ ConnectionOpened cid _ _ <- receive endpoint
215+
216+ -- Send pings to B
217+ tlog "Send pings to B"
218+ replicateM_ numPings $ send conn ["ping"]
219+
220+ -- Close our connection to B
221+ tlog "Close connection"
222+ close conn
223+
224+ -- Wait for B's pongs
225+ tlog "Wait for pongs from B"
226+ replicateM_ numPings $ do Received _ _ <- receive endpoint ; return ()
227+
228+ -- Wait for B to close it's connection to us
229+ tlog "Wait for B to close connection"
230+ ConnectionClosed cid' <- receive endpoint
231+ guard (cid == cid')
232+
233+ -- Done
234+ tlog "Done"
235+ putMVar doneA ()
236+
237+ -- B
238+ forkTry $ do
239+ tlog "B"
240+ Right endpoint <- newEndPoint transport
241+ tlog (show (address endpoint))
242+ putMVar addrB (address endpoint)
243+
244+ -- Wait for A to connect
245+ tlog "Wait for A to connect"
246+ ConnectionOpened cid _ _ <- receive endpoint
247+
248+ -- Connect to A
249+ tlog "Connect to A"
250+ Right conn <- readMVar addrA >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
251+
252+ -- Wait for A's pings
253+ tlog "Wait for pings from A"
254+ replicateM_ numPings $ do Received _ _ <- receive endpoint ; return ()
255+
256+ -- Wait for A to close it's connection to us
257+ tlog "Wait for A to close connection"
258+ ConnectionClosed cid' <- receive endpoint
259+ guard (cid == cid')
260+
261+ -- Send pongs to A
262+ tlog "Send pongs to A"
263+ replicateM_ numPings $ send conn ["pong"]
264+
265+ -- Close our connection to A
266+ tlog "Close connection to A"
267+ close conn
268+
269+ -- Done
270+ tlog "Done"
271+ putMVar doneB ()
272+
273+ mapM_ takeMVar [doneA, doneB]
274+
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
278+ where
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
283+ case mEvent of
284+ Left _ -> finish open closed
285+ Right event -> do
286+ let n' = (\x -> x - 1) <$> n
287+ case event of
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)
293+ Received cid msg ->
294+ go n' (Map.adjust (msg :) cid open) closed
295+ ReceivedMulticast _ _ ->
296+ fail "Unexpected multicast"
297+ ErrorEvent _ ->
298+ fail "Unexpected error"
299+ EndPointClosed ->
300+ fail "Unexpected endpoint closure"
301+
302+ finish open closed =
303+ if Map.null open
304+ then return . Map.toList . Map.map reverse $ closed
305+ else fail $ "Open connections: " ++ show (map fst . Map.toList $ open)
306+
307+-- | Open connection, close it, then reopen it
308+-- (In the TCP transport this means the socket will be closed, then reopened)
309+--
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
318+-- connection.
319+--
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
326+
327+ let numRepeats = 2 :: Int
328+
329+ -- A
330+ forkTry $ do
331+ Right endpoint <- newEndPoint transport
332+
333+ forM_ [1 .. numRepeats] $ \i -> do
334+ tlog "A connecting"
335+ -- Connect to B
336+ Right conn <- readMVar addrB >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
337+
338+ tlog "A pinging"
339+ -- Say hi
340+ forM_ [1 .. numPings] $ \j -> send conn [pack $ "ping" ++ show i ++ "/" ++ show j]
341+
342+ tlog "A closing"
343+ -- Disconnect again
344+ close conn
345+
346+ tlog "A finishing"
347+
348+ -- B
349+ forkTry $ do
350+ Right endpoint <- newEndPoint transport
351+ putMVar addrB (address endpoint)
352+
353+ eventss <- collect endpoint (Just (numRepeats * (numPings + 2))) Nothing
354+
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])
358+
359+ putMVar doneB ()
360+
361+ takeMVar doneB
362+
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
368+
369+ Right endpoint <- newEndPoint transport
370+
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]
376+ close conn
377+
378+ forkTry $ do
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)
383+ putMVar done ()
384+
385+ takeMVar done
386+
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
392+
393+ forkTry $ do
394+ Right endpoint <- newEndPoint transport
395+
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
400+
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
403+ -- connection open)
404+ close conn2
405+ Left (TransportError SendClosed _) <- send conn2 ["ping2"]
406+
407+ -- Now close the first connection, and output on it (i.e., output while
408+ -- there are no lightweight connection at all anymore)
409+ close conn1
410+ Left (TransportError SendClosed _) <- send conn2 ["ping2"]
411+
412+ return ()
413+
414+ putMVar clientDone ()
415+
416+ takeMVar clientDone
417+
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
423+
424+ forkTry $ do
425+ Right endpoint <- newEndPoint transport
426+
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
431+
432+ -- Close the second one twice
433+ close conn2
434+ close conn2
435+
436+ -- Then send a message on the first and close that twice too
437+ send conn1 ["ping"]
438+ close conn1
439+
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
446+
447+ return ()
448+
449+ putMVar clientDone ()
450+
451+ takeMVar clientDone
452+
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
458+
459+ tlog "Creating self-connection"
460+ Right conn <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints
461+
462+ tlog "Talk to myself"
463+
464+ -- One thread to write to the endpoint
465+ forkTry $ do
466+ tlog $ "writing"
467+
468+ tlog $ "Sending ping"
469+ replicateM_ numPings $ send conn ["ping"]
470+
471+ tlog $ "Closing connection"
472+ close conn
473+
474+ -- And one thread to read
475+ forkTry $ do
476+ tlog $ "reading"
477+
478+ tlog "Waiting for ConnectionOpened"
479+ ConnectionOpened cid _ addr <- receive endpoint ; True <- return $ addr == address endpoint
480+
481+ tlog "Waiting for Received"
482+ replicateM_ numPings $ do
483+ Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid'
484+ return ()
485+
486+ tlog "Waiting for ConnectionClosed"
487+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid'
488+
489+ tlog "Done"
490+ putMVar done ()
491+
492+ takeMVar done
493+
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
499+
500+ tlog "Creating self-connection"
501+ Right conn1 <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints
502+ Right conn2 <- connect endpoint (address endpoint) ReliableOrdered defaultConnectHints
503+
504+ tlog "Talk to myself"
505+
506+ -- One thread to write to the endpoint using the first connection
507+ forkTry $ do
508+ tlog $ "writing"
509+
510+ tlog $ "Sending ping"
511+ replicateM_ numPings $ send conn1 ["pingA"]
512+
513+ tlog $ "Closing connection"
514+ close conn1
515+
516+ -- One thread to write to the endpoint using the second connection
517+ forkTry $ do
518+ tlog $ "writing"
519+
520+ tlog $ "Sending ping"
521+ replicateM_ numPings $ send conn2 ["pingB"]
522+
523+ tlog $ "Closing connection"
524+ close conn2
525+
526+ -- And one thread to read
527+ forkTry $ do
528+ tlog $ "reading"
529+
530+ [(_, events1), (_, events2)] <- collect endpoint (Just (2 * (numPings + 2))) Nothing
531+ True <- return $ events1 == replicate numPings ["pingA"]
532+ True <- return $ events2 == replicate numPings ["pingB"]
533+
534+ tlog "Done"
535+ putMVar done ()
536+
537+ takeMVar done
538+
539+-- | Test that we self-connections no longer work once we close our endpoint
540+-- or our transport
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
549+
550+ -- Close the conneciton and try to send
551+ close conn1
552+ Left (TransportError SendClosed _) <- send conn1 ["ping"]
553+
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"]
561+
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
566+
567+ return ()
568+
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
577+
578+ -- Server
579+ forkTry $ do
580+ Right endpoint <- newEndPoint transport
581+ putMVar serverAddr (address endpoint)
582+
583+ -- First test (see client)
584+ do
585+ theirAddr <- readMVar clientAddr1
586+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
587+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid'
588+ return ()
589+
590+ -- Second test
591+ do
592+ theirAddr <- readMVar clientAddr2
593+
594+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
595+ Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid'
596+
597+ Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
598+ send conn ["pong"]
599+
600+ ConnectionClosed cid'' <- receive endpoint ; True <- return $ cid == cid''
601+ ErrorEvent (TransportError (EventConnectionLost (Just addr') []) _) <- receive endpoint ; True <- return $ addr' == theirAddr
602+
603+ Left (TransportError SendFailed _) <- send conn ["pong2"]
604+
605+ return ()
606+
607+ putMVar serverDone ()
608+
609+ -- Client
610+ forkTry $ do
611+ theirAddr <- readMVar serverAddr
612+
613+ -- First test: close endpoint with one outgoing but no incoming connections
614+ do
615+ Right endpoint <- newEndPoint transport
616+ putMVar clientAddr1 (address endpoint)
617+
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
622+ return ()
623+
624+ -- Second test: close endpoint with one outgoing and one incoming connection
625+ do
626+ Right endpoint <- newEndPoint transport
627+ putMVar clientAddr2 (address endpoint)
628+
629+ Right conn <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
630+ send conn ["ping"]
631+
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'
635+
636+ -- Close the endpoint
637+ closeEndPoint endpoint
638+ EndPointClosed <- receive endpoint
639+
640+ -- Attempt to send should fail with connection closed
641+ Left (TransportError SendFailed _) <- send conn ["ping2"]
642+
643+ -- An attempt to close the already closed connection should just return
644+ () <- close conn
645+
646+ -- And so should an attempt to connect
647+ Left (TransportError ConnectFailed _) <- connect endpoint theirAddr ReliableOrdered defaultConnectHints
648+
649+ return ()
650+
651+ putMVar clientDone ()
652+
653+ mapM_ takeMVar [serverDone, clientDone]
654+
655+-- Test closeTransport
656+--
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
665+
666+ -- Server
667+ forkTry $ do
668+ Right transport <- newTransport
669+ Right endpoint <- newEndPoint transport
670+ putMVar serverAddr (address endpoint)
671+
672+ -- Client sets up first endpoint
673+ theirAddr1 <- readMVar clientAddr1
674+ ConnectionOpened cid1 ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr1
675+
676+ -- Client sets up second endpoint
677+ theirAddr2 <- readMVar clientAddr2
678+
679+ ConnectionOpened cid2 ReliableOrdered addr' <- receive endpoint ; True <- return $ addr' == theirAddr2
680+ Received cid2' ["ping"] <- receive endpoint ; True <- return $ cid2' == cid2
681+
682+ Right conn <- connect endpoint theirAddr2 ReliableOrdered defaultConnectHints
683+ send conn ["pong"]
684+
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) []) "")
690+ ]
691+ True <- return $ any (== expected) (permutations evs)
692+
693+ -- An attempt to send to the endpoint should now fail
694+ Left (TransportError SendFailed _) <- send conn ["pong2"]
695+
696+ putMVar serverDone ()
697+
698+ -- Client
699+ forkTry $ do
700+ Right transport <- newTransport
701+ theirAddr <- readMVar serverAddr
702+
703+ -- Set up endpoint with one outgoing but no incoming connections
704+ Right endpoint1 <- newEndPoint transport
705+ putMVar clientAddr1 (address endpoint1)
706+
707+ -- Connect to the server, then close the endpoint without disconnecting explicitly
708+ Right _ <- connect endpoint1 theirAddr ReliableOrdered defaultConnectHints
709+
710+ -- Set up an endpoint with one outgoing and out incoming connection
711+ Right endpoint2 <- newEndPoint transport
712+ putMVar clientAddr2 (address endpoint2)
713+
714+ Right conn <- connect endpoint2 theirAddr ReliableOrdered defaultConnectHints
715+ send conn ["ping"]
716+
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'
720+
721+ -- Now shut down the entire transport
722+ closeTransport transport
723+
724+ -- Both endpoints should report that they have been closed
725+ EndPointClosed <- receive endpoint1
726+ EndPointClosed <- receive endpoint2
727+
728+ -- Attempt to send should fail with connection closed
729+ Left (TransportError SendFailed _) <- send conn ["ping2"]
730+
731+ -- An attempt to close the already closed connection should just return
732+ () <- close conn
733+
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
737+
738+ -- And finally, so should an attempt to create a new endpoint
739+ Left (TransportError NewEndPointFailed _) <- newEndPoint transport
740+
741+ putMVar clientDone ()
742+
743+ mapM_ takeMVar [serverDone, clientDone]
744+
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
751+
752+ -- Server
753+ forkTry $ do
754+ Right endpoint <- newEndPoint transport
755+ putMVar serverAddr (address endpoint)
756+
757+ closeEndPoint endpoint
758+ putMVar serverClosed ()
759+
760+ -- Client
761+ forkTry $ do
762+ Right endpoint <- newEndPoint transport
763+ readMVar serverClosed
764+
765+ Left (TransportError ConnectNotFound _) <- readMVar serverAddr >>= \addr -> connect endpoint addr ReliableOrdered defaultConnectHints
766+
767+ putMVar clientDone ()
768+
769+ takeMVar clientDone
770+
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
776+
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)
782+
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)
788+
789+ return ()
790+
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
797+
798+ -- Connect endpoint1 to endpoint2
799+ Right conn <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints
800+ ConnectionOpened _ _ _ <- receive endpoint2
801+
802+ -- Send an exceptional value
803+ Left (TransportError SendFailed _) <- send conn (throw $ userError "uhoh")
804+
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
810+
811+ -- A new connection will re-establish the connection
812+ Right conn2 <- connect endpoint1 (address endpoint2) ReliableOrdered defaultConnectHints
813+ send conn2 ["ping"]
814+ close conn2
815+
816+ ConnectionOpened _ _ _ <- receive endpoint2
817+ Received _ ["ping"] <- receive endpoint2
818+ ConnectionClosed _ <- receive endpoint2
819+
820+ return ()
821+
822+-- | If threads get killed while executing a 'connect', 'send', or 'close', this
823+-- should not affect other threads.
824+--
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
836+
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)
847+
848+ numAlive <- newMVar (0 :: Int)
849+
850+ -- Kill half of those threads
851+ forkIO . forM_ threads $ \tid -> do
852+ shouldKill <- randomIO
853+ if shouldKill
854+ then randomThreadDelay 600 >> killThread tid
855+ else modifyMVar_ numAlive (return . (+ 1))
856+
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")
867+
868+-- print (actualPings, expectedPings)
869+
870+
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
878+ go <- newEmptyMVar
879+
880+ let hints = defaultConnectHints {
881+ connectTimeout = Just 5000000
882+ }
883+
884+ -- A
885+ forkTry $ do
886+ Right endpoint <- newEndPoint transport
887+ putMVar aAddr (address endpoint)
888+ theirAddress <- readMVar bAddr
889+
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
902+ putMVar aDone ()
903+
904+ -- B
905+ forkTry $ do
906+ Right endpoint <- newEndPoint transport
907+ putMVar bAddr (address endpoint)
908+ theirAddress <- readMVar aAddr
909+
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
918+ putMVar bDone ()
919+
920+ -- Driver
921+ forM_ [1 .. numRepeats] $ \_i -> do
922+ -- putStrLn $ "Round " ++ show _i
923+ tryTakeMVar aTimeout
924+ tryTakeMVar bTimeout
925+ putMVar go ()
926+ putMVar go ()
927+ takeMVar aDone
928+ takeMVar bDone
929+
930+-- Transport tests
931+testTransport :: IO (Either String Transport) -> IO ()
932+testTransport newTransport = do
933+ Right transport <- newTransport
934+ runTests
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)
954+ ]
955+ where
956+ numPings = 10000 :: Int