git.iko.soy/distributed-process
on master help sign in
Implement and test closeEndPoint
commit96793556c1cbe7dedc9f2545a740578c56fc2754 authorEdsko de Vries <edsko@edsko.net> dateSat 12 May 2012, 17:45 (14 years ago) parent43d2c27
3 files changed+96−21
compare with master
unifiedsplit
network-transport-2/src/Network/Transport/TCP.hs+18−19 network-transport-2/tests/TestTCP.hs−1 network-transport-2/tests/TestTransport.hs+78−1
network-transport-2/src/Network/Transport/TCP.hs+18−19
@@ -58,13 +58,11 @@ import Control.Concurrent.MVar ( MVar
5858 , putMVar
5959 , newEmptyMVar
6060 , withMVar
61− , tryTakeMVar
6261 )
6362import Control.Category ((>>>))
6463import Control.Applicative ((<*>), (*>), (<$>))
65−import Control.Monad (forM_, void, when, mzero, unless)
64+import Control.Monad (forM, forM_, void, when, unless)
6665import Control.Monad.Error (ErrorT(ErrorT), runErrorT, throwError)
67−import Control.Monad.Trans.Maybe (MaybeT, runMaybeT)
6866import Control.Monad.IO.Class (liftIO)
6967import Control.Exception (IOException, SomeException)
7068import Data.IORef (IORef, newIORef, writeIORef, readIORef)
@@ -75,7 +73,7 @@ import Data.Int (Int32)
7573import Data.IntMap (IntMap)
7674import qualified Data.IntMap as IntMap (empty)
7775import Data.IntSet (IntSet)
78−import qualified Data.IntSet as IntSet (empty, insert, elems, singleton, null, delete)
76+import qualified Data.IntSet as IntSet (empty, insert, elems, singleton, null, delete, unions)
7977import Data.Map (Map)
8078import qualified Data.Map as Map (empty, elems, size)
8179import Data.Lens.Lazy (Lens, lens, intMapLens, mapLens, (^.), (^=), (^+=), (^%=))
@@ -396,28 +394,30 @@ apiCloseEndPoint transport ourEndPoint = do
396394 -- We take the endpoint state and don't put it back
397395 -- (this may block some threads, but we don't care: we are killing them anyway)
398396 ourState <- takeMVar (localState ourEndPoint)
399− --incoming <- IntSet.elems <$> takeMVar (localIncoming ourEndPoint)
400− let incoming = undefined
401− -- Kill all threads associated with this endpoint and close all sockets
397+ -- Close all endpoints and kill all threads
398+ incoming <- IntSet.elems . IntSet.unions <$> (forM (Map.elems $ ourState ^. localConnections) $ tryCloseRemoteSocket)
402399 forM_ (ourState ^. internalThreads) killThread
403− forM_ (Map.elems $ ourState ^. localConnections) $ \theirEndPoint ->
404− runMaybeT $ remoteSocketOf theirEndPoint >>= liftIO . N.sClose
405400 -- We send a single message to the endpoint that it is closed. Subsequent
406401 -- calls will block. We could change this so that all subsequent calls to
407402 -- receive return an error, but this would mean checking for some state on
408403 -- every call to receive, which is an unnecessary overhead.
409404 writeChan (localChannel ourEndPoint) (ErrorEvent $ ErrorEventEndPointClosed incoming)
410405 where
411− remoteSocketOf :: RemoteEndPoint -> MaybeT IO N.Socket
412− remoteSocketOf theirEndPoint = do
406+ -- Close the remote socket and return the set of all incoming connections
407+ tryCloseRemoteSocket :: RemoteEndPoint -> IO IntSet
408+ tryCloseRemoteSocket theirEndPoint = do
413409 -- Like above, we take but don't put back
414− mRemoteState <- liftIO $ tryTakeMVar (remoteState theirEndPoint)
415− case mRemoteState of
416− Nothing -> mzero
417− Just (RemoteEndPointInvalid _) -> mzero
418− Just (RemoteEndPointValid conn) -> return $ remoteSocket conn
419− Just (RemoteEndPointClosing _ conn) -> return $ remoteSocket conn
420− Just RemoteEndPointClosed -> mzero
410+ st <- takeMVar (remoteState theirEndPoint)
411+ -- We make an attempt to close the connection nicely (by sending a CloseSocket first)
412+ case st of
413+ RemoteEndPointInvalid _ -> do return IntSet.empty
414+ RemoteEndPointValid conn -> do tryIO $ do
415+ sendOn conn [encodeInt32 CloseSocket]
416+ N.sClose (remoteSocket conn)
417+ return (conn ^. remoteIncoming)
418+ RemoteEndPointClosing _ conn -> do tryIO $ N.sClose (remoteSocket conn)
419+ return IntSet.empty
420+ RemoteEndPointClosed -> do return IntSet.empty
421421
422422-- | Special case of 'apiConnect': connect an endpoint to itself
423423connectToSelf :: LocalEndPoint -> IO (Either (FailedWith ConnectErrorCode) Connection)
@@ -906,7 +906,6 @@ handleIncomingMessages (ourEndPoint, theirEndPoint) = do
906906 Just unclosedConnections -> writeChan ourChannel . ErrorEvent $
907907 ErrorEventConnectionLost (remoteAddress theirEndPoint) (IntSet.elems unclosedConnections)
908908
909−
910909-- | Get the next connection ID
911910getNextConnectionId :: LocalEndPoint -> IO ConnectionId
912911getNextConnectionId ourEndpoint =
network-transport-2/tests/TestTCP.hs−1
@@ -498,4 +498,3 @@ main = do
498498 ]
499499 Right transport <- createTransport "127.0.0.1" "8080"
500500 testTransport transport
501− return ()
network-transport-2/tests/TestTransport.hs+78−1
@@ -512,6 +512,82 @@ testConnectToSelfTwice transport numPings = do
512512
513513 takeMVar done
514514
515+-- | Test various aspects of 'closeEndPoint'
516+testCloseEndPoint :: Transport -> Int -> IO ()
517+testCloseEndPoint transport _ = do
518+ serverDone <- newEmptyMVar
519+ clientDone <- newEmptyMVar
520+ clientAddr1 <- newEmptyMVar
521+ clientAddr2 <- newEmptyMVar
522+ serverAddr <- newEmptyMVar
523+
524+ -- Server
525+ forkTry $ do
526+ Right endpoint <- newEndPoint transport
527+ putMVar serverAddr (address endpoint)
528+
529+ -- First test (see client)
530+ do
531+ theirAddr <- readMVar clientAddr1
532+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
533+ ConnectionClosed cid' <- receive endpoint ; True <- return $ cid == cid'
534+ return ()
535+
536+ -- Second test
537+ do
538+ theirAddr <- readMVar clientAddr2
539+
540+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
541+ Received cid' ["ping"] <- receive endpoint ; True <- return $ cid == cid'
542+
543+ Right conn <- connect endpoint theirAddr ReliableOrdered
544+ send conn ["pong"]
545+
546+ ConnectionClosed cid'' <- receive endpoint ; True <- return $ cid == cid''
547+ ErrorEvent (ErrorEventConnectionLost addr' []) <- receive endpoint ; True <- return $ addr' == theirAddr
548+
549+ Left (FailedWith SendConnectionClosed _) <- send conn ["pong2"]
550+
551+ return ()
552+
553+ putMVar serverDone ()
554+
555+ -- Client
556+ forkTry $ do
557+ theirAddr <- readMVar serverAddr
558+
559+ -- First test: close endpoint with one outgoing but no incoming connections
560+ do
561+ Right endpoint <- newEndPoint transport
562+ putMVar clientAddr1 (address endpoint)
563+
564+ -- Connect to the server, then close the endpoint without disconnecting explicitly
565+ Right _ <- connect endpoint theirAddr ReliableOrdered
566+ closeEndPoint endpoint
567+ ErrorEvent (ErrorEventEndPointClosed []) <- receive endpoint
568+ return ()
569+
570+ -- Second test: close endpoint with one outgoing and one incoming connection
571+ do
572+ Right endpoint <- newEndPoint transport
573+ putMVar clientAddr2 (address endpoint)
574+
575+ Right conn <- connect endpoint theirAddr ReliableOrdered
576+ send conn ["ping"]
577+
578+ -- Reply from the server
579+ ConnectionOpened cid ReliableOrdered addr <- receive endpoint ; True <- return $ addr == theirAddr
580+ Received cid' ["pong"] <- receive endpoint ; True <- return $ cid == cid'
581+
582+ -- Close the endpoint
583+ closeEndPoint endpoint
584+ ErrorEvent (ErrorEventEndPointClosed [cid'']) <- receive endpoint; True <- return $ cid == cid''
585+ return ()
586+
587+ putMVar clientDone ()
588+
589+ mapM_ takeMVar [serverDone, clientDone]
590+
515591-- Transport tests
516592testTransport :: Transport -> IO ()
517593testTransport transport = runTests
@@ -525,7 +601,8 @@ testTransport transport = runTests
525601 , ("SendAfterClose", testSendAfterClose transport numPings)
526602 , ("CloseTwice", testCloseTwice transport numPings)
527603 , ("ConnectToSelf", testConnectToSelf transport numPings)
528− , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)
604+ , ("ConnectToSelfTwice", testConnectToSelfTwice transport numPings)
605+ , ("CloseEndPoint", testCloseEndPoint transport numPings)
529606 ]
530607 where
531608 numPings = 10000 :: Int