git.iko.soy/distributed-process
on master help sign in
Partly implement closeTransport
b9e6b58Edsko de Vries
network-transport-2/src/Network/Transport/Internal/TCP.hs+2
compare with master
unifiedsplit
@@ -61,9 +61,11 @@ forkServer host port backlog terminationHandler requestHandler = do
61 -- the "best" address first, whatever that means
61 -- the "best" address first, whatever that means
62 addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)
62 addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)
63 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
63 sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
64+ -- putStrLn $ "Created server socket " ++ show sock ++ " for address " ++ host ++ ":" ++ port
64 N.setSocketOption sock N.ReuseAddr 1
65 N.setSocketOption sock N.ReuseAddr 1
65 N.bindSocket sock (N.addrAddress addr)
66 N.bindSocket sock (N.addrAddress addr)
66 N.listen sock backlog
67 N.listen sock backlog
68+ -- TODO: when is this socket closed?
67 forkIO . handle terminationHandler . forever $
69 forkIO . handle terminationHandler . forever $
68 bracketOnError (N.accept sock)
70 bracketOnError (N.accept sock)
69 (N.sClose . fst)
71 (N.sClose . fst)
network-transport-2/src/Network/Transport/TCP.hs+15−14
@@ -329,21 +329,20 @@ createTransport host port = do
329 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
329 -- http://www.linuxjournal.com/files/linuxjournal.com/linuxjournal/articles/023/2333/2333s2.html
330 transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
330 transportThread <- forkServer host port N.sOMAXCONN (terminationHandler transport) (handleConnectionRequest transport)
331 return Transport { newEndPoint = apiNewEndPoint transport
331 return Transport { newEndPoint = apiNewEndPoint transport
332− , closeTransport = apiCloseTransport transportThread
332+ , closeTransport = killThread transportThread -- This will invoke the termination handler
333 }
333 }
334 where
334 where
335 terminationHandler :: TCPTransport -> SomeException -> IO ()
335 terminationHandler :: TCPTransport -> SomeException -> IO ()
336− terminationHandler _ _ =
336+ terminationHandler transport _ = do
337− -- TODO: implement
337+ mTSt <- modifyMVar (transportState transport) $ \st -> case st of
338− return ()
338+ TransportValid vst -> return (TransportClosed, Just vst)
339−
339+ TransportClosed -> return (TransportClosed, Nothing)
340−-- | Shut down the transport
340+ case mTSt of
341−apiCloseTransport :: ThreadId -> IO ()
341+ Nothing ->
342−apiCloseTransport transportThread = do
342+ -- Transport already closed
343− -- Kill the thread that is listening for new incoming connections
343+ return ()
344− killThread transportThread
344+ Just tSt ->
345− -- Not yet implemented
345+ mapM_ (apiCloseEndPoint transport) (Map.elems $ tSt ^. localEndPoints)
346− undefined
347
346
348--------------------------------------------------------------------------------
347--------------------------------------------------------------------------------
349-- API functions --
348-- API functions --
@@ -696,8 +695,10 @@ socketToEndPoint (EndPointAddress ourAddress) theirAddress = try $ do
696 Just r -> return (sock, r)
695 Just r -> return (sock, r)
697 where
696 where
698 createSocket :: N.AddrInfo -> IO N.Socket
697 createSocket :: N.AddrInfo -> IO N.Socket
699− createSocket addr = mapExceptionIO insufficientResources $
698+ createSocket addr = mapExceptionIO insufficientResources $ do
700− N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
699+ sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol
700+ -- putStrLn $ "Created client socket " ++ show sock
701+ return sock
701
702
702 invalidAddress, insufficientResources, failed :: IOException -> FailedWith ConnectErrorCode
703 invalidAddress, insufficientResources, failed :: IOException -> FailedWith ConnectErrorCode
703 invalidAddress = FailedWith ConnectInvalidAddress . show
704 invalidAddress = FailedWith ConnectInvalidAddress . show
network-transport-2/tests/TestInMemory.hs+2−1
@@ -2,6 +2,7 @@ module Main where
2
2
3import TestTransport
3import TestTransport
4import Network.Transport.Chan
4import Network.Transport.Chan
5+import Control.Applicative ((<$>))
5
6
6main :: IO ()
7main :: IO ()
7−main = createTransport >>= testTransport
8+main = testTransport (Right <$> createTransport)
network-transport-2/tests/TestTCP.hs+4−5
@@ -9,7 +9,7 @@ import Network.Transport
9import Network.Transport.TCP (createTransport, encodeEndPointAddress)
9import Network.Transport.TCP (createTransport, encodeEndPointAddress)
10import Data.Int (Int32)
10import Data.Int (Int32)
11import Control.Concurrent (threadDelay)
11import Control.Concurrent (threadDelay)
12−import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar, tryTakeMVar)
12+import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar, readMVar, isEmptyMVar)
13import Control.Monad (replicateM, guard, forM_)
13import Control.Monad (replicateM, guard, forM_)
14import Control.Applicative ((<$>))
14import Control.Applicative ((<$>))
15import Control.Exception (throw)
15import Control.Exception (throw)
@@ -457,8 +457,8 @@ testUnnecessaryConnect nextPort = do
457main :: IO ()
457main :: IO ()
458main = do
458main = do
459 portMVar <- newEmptyMVar
459 portMVar <- newEmptyMVar
460− forkTry $ forM_ ([10080 .. 10088] :: [Int]) $ putMVar portMVar . show
460+ forkTry $ forM_ ([10080 ..] :: [Int]) $ putMVar portMVar . show
461− let nextPort = do Just port <- tryTakeMVar portMVar ; return port
461+ let nextPort = takeMVar portMVar
462 tryIO $ runTests
462 tryIO $ runTests
463 [ ("EarlyDisconnect", testEarlyDisconnect nextPort)
463 [ ("EarlyDisconnect", testEarlyDisconnect nextPort)
464 , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)
464 , ("EarlyCloseSocket", testEarlyCloseSocket nextPort)
@@ -468,5 +468,4 @@ main = do
468 , ("InvalidAddress", testInvalidAddress nextPort)
468 , ("InvalidAddress", testInvalidAddress nextPort)
469 , ("InvalidConnect", testInvalidConnect nextPort)
469 , ("InvalidConnect", testInvalidConnect nextPort)
470 ]
470 ]
471− Right transport <- createTransport "127.0.0.1" "8080"
471+ testTransport (either (Left . show) (Right) <$> nextPort >>= createTransport "127.0.0.1")
472− testTransport transport
network-transport-2/tests/TestTransport.hs+119−15
@@ -598,21 +598,125 @@ testCloseEndPoint transport _ = do
598
598
599 mapM_ takeMVar [serverDone, clientDone]
599 mapM_ takeMVar [serverDone, clientDone]
600
600
601+-- Test closeTransport
602+--
603+-- This tests many of the same things that testEndPoint does, and some more
604+testCloseTransport :: IO (Either String Transport) -> IO ()
605+testCloseTransport newTransport = do
606+ serverDone <- newEmptyMVar
607+ clientDone <- newEmptyMVar
608+ clientAddr1 <- newEmptyMVar
609+ clientAddr2 <- newEmptyMVar
610+ serverAddr <- newEmptyMVar
611+
612+ -- Server
613+ forkTry $ do
614+ Right transport <- newTransport
615+ Right endpoint <- newEndPoint transport
616+ putMVar serverAddr (address endpoint)
617+
618+ -- Client sets up first endpoint
619+ theirAddr1 <- readMVar clientAddr1
620+ ConnectionOpened cid1 ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr1
621+
622+ -- Client sets up second endpoint
623+ theirAddr2 <- readMVar clientAddr2
624+
625+ ConnectionOpened cid2 ReliableOrdered addr' <- receive endpoint ; True <- return $ addr' == theirAddr2
626+ Received cid2' ["ping"] <- receive endpoint ; True <- return $ cid2' == cid2
627+
628+ Right conn <- connect endpoint theirAddr2 ReliableOrdered
629+ send conn ["pong"]
630+
631+ -- Client now closes down its transport. We should receive connection closed messages
632+ -- TODO: this assumes a certain ordering on the messages we receive; that's not guaranteed
633+ ConnectionClosed cid2'' <- receive endpoint ; True <- return $ cid2'' == cid2
634+ ErrorEvent (ErrorEventConnectionLost addr'' []) <- receive endpoint ; True <- return $ addr'' == theirAddr2
635+ ConnectionClosed cid1' <- receive endpoint ; True <- return $ cid1' == cid1
636+
637+ -- An attempt to send to the endpoint should now fail
638+ Left (FailedWith SendConnectionClosed _) <- send conn ["pong2"]
639+
640+ putMVar serverDone ()
641+
642+ -- Client
643+ forkTry $ do
644+ Right transport <- newTransport
645+ theirAddr <- readMVar serverAddr
646+
647+ -- Set up endpoint with one outgoing but no incoming connections
648+ Right endpoint1 <- newEndPoint transport
649+ putMVar clientAddr1 (address endpoint1)
650+
651+ -- Connect to the server, then close the endpoint without disconnecting explicitly
652+ Right _ <- connect endpoint1 theirAddr ReliableOrdered
653+
654+ -- Set up an endpoint with one outgoing and out incoming connection
655+ Right endpoint2 <- newEndPoint transport
656+ putMVar clientAddr2 (address endpoint2)
657+
658+ Right conn <- connect endpoint2 theirAddr ReliableOrdered
659+ send conn ["ping"]
660+
661+ -- Reply from the server
662+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint2 ; True <- return $ addr == theirAddr
663+ Received cid' ["pong"] <- receive endpoint2 ; True <- return $ cid == cid'
664+
665+ -- Now shut down the entire transport
666+ closeTransport transport
667+
668+ -- Both endpoints should report an error
669+ ErrorEvent (ErrorEventEndPointClosed []) <- receive endpoint1
670+ ErrorEvent (ErrorEventEndPointClosed [cid'']) <- receive endpoint2; True <- return $ cid == cid''
671+
672+ -- Attempt to send should fail with connection closed
673+ Left (FailedWith SendConnectionClosed _) <- send conn ["ping2"]
674+
675+ -- An attempt to close the already closed connection should just return
676+ () <- close conn
677+
678+ -- And so should an attempt to connect on either endpoint
679+ Left (FailedWith ConnectFailed _) <- connect endpoint1 theirAddr ReliableOrdered
680+ Left (FailedWith ConnectFailed _) <- connect endpoint2 theirAddr ReliableOrdered
681+
682+ -- And finally, so should an attempt to create a new endpoint
683+ Left (FailedWith NewEndPointTransportFailure _) <- newEndPoint transport
684+
685+ putMVar clientDone ()
686+
687+ mapM_ takeMVar [serverDone, clientDone]
688+
689+testMany :: IO (Either String Transport) -> IO ()
690+testMany newTransport = do
691+ Right masterTransport <- newTransport
692+ Right masterEndPoint <- newEndPoint masterTransport
693+
694+ replicateM_ 20 $ do
695+ Right transport <- newTransport
696+ replicateM_ 2 $ do
697+ Right endpoint <- newEndPoint transport
698+ Right _ <- connect endpoint (address masterEndPoint) ReliableOrdered
699+ return ()
700+
601-- Transport tests
701-- Transport tests
602−testTransport :: Transport -> IO ()
702+testTransport :: IO (Either String Transport) -> IO ()
603−testTransport transport = runTests
703+testTransport newTransport = do
704+ Right transport <- newTransport
705+ runTests
706+ [ ("Many", testMany newTransport)
604− [ ("PingPong", testPingPong transport numPings)
707+ , ("PingPong", testPingPong transport numPings)
605− , ("EndPoints", testEndPoints transport numPings)
708+ , ("EndPoints", testEndPoints transport numPings)
606− , ("Connections", testConnections transport numPings)
709+ , ("Connections", testConnections transport numPings)
607− , ("CloseOneConnection", testCloseOneConnection transport numPings)
710+ , ("CloseOneConnection", testCloseOneConnection transport numPings)
608− , ("CloseOneDirection", testCloseOneDirection transport numPings)
711+ , ("CloseOneDirection", testCloseOneDirection transport numPings)
609− , ("CloseReopen", testCloseReopen transport numPings)
712+ , ("CloseReopen", testCloseReopen transport numPings)
610− , ("ParallelConnects", testParallelConnects transport numPings)
713+ , ("ParallelConnects", testParallelConnects transport numPings)
611− , ("SendAfterClose", testSendAfterClose transport numPings)
714+ , ("SendAfterClose", testSendAfterClose transport numPings)
612− , ("CloseTwice", testCloseTwice transport numPings)
715+ , ("CloseTwice", testCloseTwice transport numPings)
613− , ("ConnectToSelf", testConnectToSelf transport numPings)
716+ , ("ConnectToSelf", testConnectToSelf transport numPings)
614− , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)
717+ , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)
615− , ("CloseEndPoint", testCloseEndPoint transport numPings)
718+ , ("CloseEndPoint", testCloseEndPoint transport numPings)
719+ , ("CloseTransport", testCloseTransport newTransport)
616− ]
720+ ]
617 where
721 where
618 numPings = 10000 :: Int
722 numPings = 10000 :: Int