git.iko.soy/distributed-process
on master help sign in
More checks for user errors
a892422Edsko de Vries
network-transport-2/src/Network/Transport/TCP.hs+16−6
compare with master
unifiedsplit
@@ -57,7 +57,7 @@ import Control.Concurrent.MVar ( MVar
57 )
57 )
58import Control.Category ((>>>))
58import Control.Category ((>>>))
59import Control.Applicative ((<*>), (*>), (<$>))
59import Control.Applicative ((<*>), (*>), (<$>))
60−import Control.Monad (forM_, void, unless)
60+import Control.Monad (forM_, void, unless, when)
61import Control.Monad.Error (ErrorT(ErrorT), runErrorT)
61import Control.Monad.Error (ErrorT(ErrorT), runErrorT)
62import Control.Monad.IO.Class (liftIO)
62import Control.Monad.IO.Class (liftIO)
63import Control.Exception (IOException, bracket_)
63import Control.Exception (IOException, bracket_)
@@ -266,17 +266,27 @@ apiConnect ourEndPoint theirAddress _ = runErrorT $ do
266-- TODO: We ignore errors during a close. Is that right?
266-- TODO: We ignore errors during a close. Is that right?
267apiClose :: Conn -> ConnectionId -> MVar Bool -> IO ()
267apiClose :: Conn -> ConnectionId -> MVar Bool -> IO ()
268apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
268apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
269+ modifyMVar_ connAlive $ \alive -> do
270+ when alive $ do
269− sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]
271+ sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]
270− decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
272+ decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
273+ return False
271
274
272-- | Send data across a connection
275-- | Send data across a connection
273--
276--
274-- TODO: we should have some per-connection state so that the user cannot send
277-- TODO: we should have some per-connection state so that the user cannot send
275-- to the connection after they closed it, or close it more than once
278-- to the connection after they closed it, or close it more than once
276apiSend :: RemoteEndPoint -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
279apiSend :: RemoteEndPoint -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
277−apiSend theirEndPoint connId connAlive payload = runErrorT $
280+apiSend theirEndPoint connId connAlive payload =
278− failWithIO (FailedWith SendFailed . show) $
281+ modifyMVar connAlive $ \alive -> do
282+ if alive
283+ then do
279− sendTo theirEndPoint (encodeInt32 connId : prependLength payload)
284+ result <- tryIO $ sendTo theirEndPoint (encodeInt32 connId : prependLength payload)
285+ case result of
286+ Left err -> return (alive, Left $ FailedWith SendFailed (show err))
287+ Right _ -> return (alive, Right $ ())
288+ else do
289+ return (alive, Left $ FailedWith SendConnectionClosed "Connection closed")
280
290
281--------------------------------------------------------------------------------
291--------------------------------------------------------------------------------
282-- Lower level functionality --
292-- Lower level functionality --
network-transport-2/src/Network/Transport/Util.hsnew file+24
@@ -0,0 +1,24 @@
1+-- | Utility functions
2+--
3+-- Note: this module is bound to change even more than the rest of the API :)
4+module Network.Transport.Util (spawn) where
5+
6+import Network.Transport ( Transport
7+ , EndPoint(..)
8+ , EndPointAddress
9+ , newEndPoint
10+ )
11+import Control.Concurrent (forkIO)
12+import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar)
13+
14+-- | Fork a new thread, create a new end point on that thread, and run the specified IO operation on that thread.
15+--
16+-- Returns the address of the new end point.
17+spawn :: Transport -> (EndPoint -> IO ()) -> IO EndPointAddress
18+spawn transport proc = do
19+ addr <- newEmptyMVar
20+ forkIO $ do
21+ Right endpoint <- newEndPoint transport
22+ putMVar addr (address endpoint)
23+ proc endpoint
24+ takeMVar addr
network-transport-2/tests/TestTransport.hs+23−15
@@ -1,7 +1,7 @@
1module TestTransport where
1module TestTransport where
2
2
3import Prelude hiding (catch)
3import Prelude hiding (catch)
4−import Control.Concurrent (forkIO, threadDelay)
4+import Control.Concurrent (forkIO)
5import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)
5import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)
6import Control.Monad (replicateM, replicateM_, when, guard, forM_)
6import Control.Monad (replicateM, replicateM_, when, guard, forM_)
7import Control.Monad.Reader (ReaderT, runReaderT, ask)
7import Control.Monad.Reader (ReaderT, runReaderT, ask)
@@ -40,16 +40,20 @@ echoServer endpoint = do
40 -- Ignore
40 -- Ignore
41 go cs
41 go cs
42
42
43−expect :: EndPoint -> (Event -> Bool) -> IO ()
43+expect :: EndPoint -> (Event -> (Bool, a)) -> IO a
44expect endpoint predicate = do
44expect endpoint predicate = do
45 event <- receive endpoint
45 event <- receive endpoint
46 mbool <- catch (Right <$> evaluate (predicate event)) (return . Left)
46 mbool <- catch (Right <$> evaluate (predicate event)) (return . Left)
47 case mbool of
47 case mbool of
48− Left err -> do tlog $ "Unexpected event " ++ show event
48+ Left err -> do tlog $ "Unexpected event " ++ show event
49− throw (err :: SomeException)
49+ throw (err :: SomeException)
50− Right False -> do tlog $ "Unexpected event " ++ show event
50+ Right (False, _) -> do tlog $ "Unexpected event " ++ show event
51− fail "Unexpected event"
51+ fail "Unexpected event"
52− _ -> return ()
52+ Right (True, a) -> return a
53+
54+expect' :: EndPoint -> (Event -> Bool) -> IO ()
55+expect' endpoint predicate =
56+ expect endpoint (\event -> (predicate event, ()))
53
57
54ping :: EndPoint -> EndPointAddress -> Int -> ByteString -> IO ()
58ping :: EndPoint -> EndPointAddress -> Int -> ByteString -> IO ()
55ping endpoint server numPings msg = do
59ping endpoint server numPings msg = do
@@ -59,13 +63,13 @@ ping endpoint server numPings msg = do
59
63
60 -- Wait for the server to open reply connection
64 -- Wait for the server to open reply connection
61 tlog "Wait for ConnectionOpened message"
65 tlog "Wait for ConnectionOpened message"
62− expect endpoint (\(ConnectionOpened _ _ _) -> True)
66+ cid <- expect endpoint (\(ConnectionOpened cid _ _) -> (True, cid))
63
67
64 -- Send pings and wait for reply
68 -- Send pings and wait for reply
65 tlog "Send ping and wait for reply"
69 tlog "Send ping and wait for reply"
66 replicateM_ numPings $ do
70 replicateM_ numPings $ do
67 send conn [msg]
71 send conn [msg]
68− expect endpoint (\(Received _ [reply]) -> reply == msg)
72+ expect' endpoint (\(Received cid' [reply]) -> cid == cid' && reply == msg)
69
73
70 -- Close the connection
74 -- Close the connection
71 tlog "Close the connection"
75 tlog "Close the connection"
@@ -73,7 +77,7 @@ ping endpoint server numPings msg = do
73
77
74 -- Wait for the server to close its connection to us
78 -- Wait for the server to close its connection to us
75 tlog "Wait for ConnectionClosed message"
79 tlog "Wait for ConnectionClosed message"
76− expect endpoint (\(ConnectionClosed _) -> True)
80+ expect' endpoint (\(ConnectionClosed cid') -> cid == cid')
77
81
78 -- Done
82 -- Done
79 tlog "Ping client done"
83 tlog "Ping client done"
@@ -388,18 +392,19 @@ testSendAfterClose transport _ = do
388 -- connection (i.e., on a closed connection while there is still another
392 -- connection (i.e., on a closed connection while there is still another
389 -- connection open)
393 -- connection open)
390 close conn2
394 close conn2
391− send conn2 ["ping2"]
395+ Left (FailedWith SendConnectionClosed _) <- send conn2 ["ping2"]
392
396
393 -- Now close the first connection, and output on it (i.e., output while
397 -- Now close the first connection, and output on it (i.e., output while
394 -- there are no lightweight connection at all anymore)
398 -- there are no lightweight connection at all anymore)
395 close conn1
399 close conn1
396− send conn1 ["ping1"]
400+ Left (FailedWith SendConnectionClosed _) <- send conn2 ["ping2"]
397
401
398− threadDelay 100000
399 putMVar clientDone ()
402 putMVar clientDone ()
400
403
401 takeMVar clientDone
404 takeMVar clientDone
402
405
406+
407+
403-- Test that closing the same connection twice has no effect
408-- Test that closing the same connection twice has no effect
404testCloseTwice :: Transport -> Int -> IO ()
409testCloseTwice :: Transport -> Int -> IO ()
405testCloseTwice transport _ = do
410testCloseTwice transport _ = do
@@ -422,9 +427,12 @@ testCloseTwice transport _ = do
422 close conn1
427 close conn1
423
428
424 -- Verify expected response from the echo server
429 -- Verify expected response from the echo server
425− -- TODO
430+ cid1 <- expect endpoint (\(ConnectionOpened cid _ _) -> (True, cid))
431+ cid2 <- expect endpoint (\(ConnectionOpened cid _ _) -> (True, cid))
432+ expect' endpoint (\(ConnectionClosed cid) -> cid == cid2)
433+ expect' endpoint (\(Received cid msg) -> cid == cid1 && msg == ["ping"])
434+ expect' endpoint (\(ConnectionClosed cid) -> cid == cid1)
426
435
427− threadDelay 1000000
428 putMVar clientDone ()
436 putMVar clientDone ()
429
437
430 takeMVar clientDone
438 takeMVar clientDone