git.iko.soy/distributed-process
on master help sign in
More checks for user errors

Also added missing file (Util.hs) -- whoops, sorry.

commita892422033012be985142d113e73a488df26a242 authorEdsko de Vries <edsko@edsko.net> dateWed 2 May 2012, 17:12 (14 years ago) parent230dee9
3 files changed+63−21
compare with master
unifiedsplit
network-transport-2/src/Network/Transport/TCP.hs+16−6 network-transport-2/src/Network/Transport/Util.hs+24 network-transport-2/tests/TestTransport.hs+23−15
network-transport-2/src/Network/Transport/TCP.hs+16−6
@@ -57,7 +57,7 @@ import Control.Concurrent.MVar ( MVar
5757 )
5858import Control.Category ((>>>))
5959import Control.Applicative ((<*>), (*>), (<$>))
60−import Control.Monad (forM_, void, unless)
60+import Control.Monad (forM_, void, unless, when)
6161import Control.Monad.Error (ErrorT(ErrorT), runErrorT)
6262import Control.Monad.IO.Class (liftIO)
6363import Control.Exception (IOException, bracket_)
@@ -266,17 +266,27 @@ apiConnect ourEndPoint theirAddress _ = runErrorT $ do
266266-- TODO: We ignore errors during a close. Is that right?
267267apiClose :: Conn -> ConnectionId -> MVar Bool -> IO ()
268268apiClose (ourEndPoint, theirEndPoint) connId connAlive = void . tryIO $ do
269− sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]
270− decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
269+ modifyMVar_ connAlive $ \alive -> do
270+ when alive $ do
271+ sendTo theirEndPoint [encodeInt32 CloseConnection, encodeInt32 connId]
272+ decreaseRemoteRefCount (ourEndPoint, theirEndPoint)
273+ return False
271274
272275-- | Send data across a connection
273276--
274277-- TODO: we should have some per-connection state so that the user cannot send
275278-- to the connection after they closed it, or close it more than once
276279apiSend :: RemoteEndPoint -> ConnectionId -> MVar Bool -> [ByteString] -> IO (Either (FailedWith SendErrorCode) ())
277−apiSend theirEndPoint connId connAlive payload = runErrorT $
278− failWithIO (FailedWith SendFailed . show) $
279− sendTo theirEndPoint (encodeInt32 connId : prependLength payload)
280+apiSend theirEndPoint connId connAlive payload =
281+ modifyMVar connAlive $ \alive -> do
282+ if alive
283+ then do
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")
280290
281291--------------------------------------------------------------------------------
282292-- 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 @@
11module TestTransport where
22
33import Prelude hiding (catch)
4−import Control.Concurrent (forkIO, threadDelay)
4+import Control.Concurrent (forkIO)
55import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar, readMVar)
66import Control.Monad (replicateM, replicateM_, when, guard, forM_)
77import Control.Monad.Reader (ReaderT, runReaderT, ask)
@@ -40,16 +40,20 @@ echoServer endpoint = do
4040 -- Ignore
4141 go cs
4242
43−expect :: EndPoint -> (Event -> Bool) -> IO ()
43+expect :: EndPoint -> (Event -> (Bool, a)) -> IO a
4444expect endpoint predicate = do
4545 event <- receive endpoint
4646 mbool <- catch (Right <$> evaluate (predicate event)) (return . Left)
4747 case mbool of
48− Left err -> do tlog $ "Unexpected event " ++ show event
49− throw (err :: SomeException)
50− Right False -> do tlog $ "Unexpected event " ++ show event
51− fail "Unexpected event"
52− _ -> return ()
48+ Left err -> do tlog $ "Unexpected event " ++ show event
49+ throw (err :: SomeException)
50+ Right (False, _) -> do tlog $ "Unexpected event " ++ show event
51+ fail "Unexpected event"
52+ Right (True, a) -> return a
53+
54+expect' :: EndPoint -> (Event -> Bool) -> IO ()
55+expect' endpoint predicate =
56+ expect endpoint (\event -> (predicate event, ()))
5357
5458ping :: EndPoint -> EndPointAddress -> Int -> ByteString -> IO ()
5559ping endpoint server numPings msg = do
@@ -59,13 +63,13 @@ ping endpoint server numPings msg = do
5963
6064 -- Wait for the server to open reply connection
6165 tlog "Wait for ConnectionOpened message"
62− expect endpoint (\(ConnectionOpened _ _ _) -> True)
66+ cid <- expect endpoint (\(ConnectionOpened cid _ _) -> (True, cid))
6367
6468 -- Send pings and wait for reply
6569 tlog "Send ping and wait for reply"
6670 replicateM_ numPings $ do
6771 send conn [msg]
68− expect endpoint (\(Received _ [reply]) -> reply == msg)
72+ expect' endpoint (\(Received cid' [reply]) -> cid == cid' && reply == msg)
6973
7074 -- Close the connection
7175 tlog "Close the connection"
@@ -73,7 +77,7 @@ ping endpoint server numPings msg = do
7377
7478 -- Wait for the server to close its connection to us
7579 tlog "Wait for ConnectionClosed message"
76− expect endpoint (\(ConnectionClosed _) -> True)
80+ expect' endpoint (\(ConnectionClosed cid') -> cid == cid')
7781
7882 -- Done
7983 tlog "Ping client done"
@@ -388,18 +392,19 @@ testSendAfterClose transport _ = do
388392 -- connection (i.e., on a closed connection while there is still another
389393 -- connection open)
390394 close conn2
391− send conn2 ["ping2"]
395+ Left (FailedWith SendConnectionClosed _) <- send conn2 ["ping2"]
392396
393397 -- Now close the first connection, and output on it (i.e., output while
394398 -- there are no lightweight connection at all anymore)
395399 close conn1
396− send conn1 ["ping1"]
400+ Left (FailedWith SendConnectionClosed _) <- send conn2 ["ping2"]
397401
398− threadDelay 100000
399402 putMVar clientDone ()
400403
401404 takeMVar clientDone
402405
406+
407+
403408-- Test that closing the same connection twice has no effect
404409testCloseTwice :: Transport -> Int -> IO ()
405410testCloseTwice transport _ = do
@@ -422,9 +427,12 @@ testCloseTwice transport _ = do
422427 close conn1
423428
424429 -- 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)
426435
427− threadDelay 1000000
428436 putMVar clientDone ()
429437
430438 takeMVar clientDone