@@ -1,956 +0,0 @@
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