39d3f1e · distributed-process

git.iko.soy/distributed-process

Deal with errors during sending

commit39d3f1e6798a93e3a0f10fa9d9a39c07c0db3247 authorEdsko de Vries <edsko@edsko.net> dateMon 23 Apr 2012, 17:23 (15 years ago) parent3938dfc

6 files changed+85−49

compare with master

unifiedsplit

network-transport-2/src/Network/Transport.hs+5−1 network-transport-2/src/Network/Transport/Chan.hs+2−1 network-transport-2/src/Network/Transport/Internal/TCP.hs+32−4 network-transport-2/src/Network/Transport/TCP.hs+23−32 network-transport-2/tests/TestTCP.hs+20−9 network-transport-2/tests/TestTransport.hs+3−2

network-transport-2/src/Network/Transport.hs+5−1

@@ -17,6 +17,7 @@ module Network.Transport ( -- * Types

1717 , ConnectErrorCode(..)

1818 , NewMulticastGroupErrorCode(..)

1919 , ResolveMulticastGroupErrorCode(..)

20+ , SendErrorCode(..)

2021 ) where

2122

2223import Data.ByteString (ByteString)

@@ -55,6 +56,9 @@ data ResolveMulticastGroupErrorCode =

5556 ResolveMulticastGroupNotFound

5657 | ResolveMulticastGroupUnsupported

5758

59+-- | Failure during sending a message

60+data SendErrorCode =

61+ SendFailed

5862

5963data FailedWith error = FailedWith error String

6064

@@ -88,7 +92,7 @@ type ConnectionId = Int

8892-- | Lightweight connection to an endpoint.

8993data Connection = Connection {

9094 -- | Send a message on this connection.

91− send :: [ByteString] -> IO ()

95+ send :: [ByteString] -> IO (Either (FailedWith SendErrorCode) ())

9296 -- | Close the connection.

9397 , close :: IO ()

9498 }

network-transport-2/src/Network/Transport/Chan.hs+2−1

@@ -73,7 +73,8 @@ chanConnect myAddress state theirAddress _ = do

7373 let conn = st ^. connectionAt theirAddress

7474 return (connectionAt theirAddress ^+= 1 $ st, (chan, conn))

7575 writeChan chan $ ConnectionOpened conn ReliableOrdered myAddress

76− return . Right $ Connection { send = writeChan chan . Received conn

76+ return . Right $ Connection { send = \msg -> do writeChan chan (Received conn msg)

77+ return (Right ())

7778 , close = writeChan chan $ ConnectionClosed conn

7879 }

7980

network-transport-2/src/Network/Transport/Internal/TCP.hs+32−4

@@ -1,10 +1,13 @@

11-- | Utility functions for TCP sockets

22module Network.Transport.Internal.TCP ( forkServer

3+ , connectTo

34 , sendWithLength

45 , recvWithLength

6+ , sendMany

57 , recvExact

68 ) where

10+import Prelude hiding (catch)

811import Network.Transport.Internal ( encodeInt32

912 , decodeInt32

1013 )

@@ -22,17 +25,22 @@ import qualified Network.Socket as N ( HostName

2225 , addrAddress

2326 , defaultProtocol

2427 , setSocketOption

28+ , connect

2529 )

2630import qualified Network.Socket.ByteString as NBS (sendMany, recv)

2731import Control.Concurrent (forkIO, ThreadId)

2832import Control.Monad (mzero)

2933import Control.Monad.IO.Class (liftIO)

30−import Control.Monad.Trans.Maybe (MaybeT)

34+import Control.Monad.Trans.Maybe (MaybeT(MaybeT))

35+import Control.Exception (catch, IOException)

36+import Control.Applicative ((<$>), pure)

3137import Data.ByteString (ByteString)

3238import qualified Data.ByteString as BS (length, concat, null)

3339import Data.Int (Int32)

3440

3541-- | Start a server at the specified address

42+--

43+-- TODO: deal with errors

3644forkServer :: N.HostName -> N.ServiceName -> (N.Socket -> IO ()) -> IO ThreadId

3745forkServer host port server = do

3846 -- Resolve the specified address. By specification, getAddrInfo will never

@@ -45,15 +53,26 @@ forkServer host port server = do

4553 N.listen sock 5

4654 forkIO $ server sock

4755

56+-- | Connect to another host

57+--

58+-- TODO: deal with errors

59+connectTo :: N.HostName -> N.ServiceName -> MaybeT IO N.Socket

60+connectTo host port = liftIO $ do

61+ addr:_ <- N.getAddrInfo Nothing (Just host) (Just port)

62+ sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol

63+ N.setSocketOption sock N.ReuseAddr 1

64+ N.connect sock (N.addrAddress addr)

65+ return sock

66+

4867-- | Send a bunch of bytestrings prepended with their length

4968sendWithLength :: N.Socket -- ^ Socket to send on

5069 -> Maybe ByteString -- ^ Optional header to send before the length

5170 -> [ByteString] -- ^ Payload

52− -> IO ()

71+ -> MaybeT IO ()

5372sendWithLength sock header payload = do

54− lengthBs <- encodeInt32 (fromIntegral . sum . map BS.length $ payload)

73+ lengthBs <- liftIO $ encodeInt32 (fromIntegral . sum . map BS.length $ payload)

5574 let msg = maybe id (:) header $ lengthBs : payload

56− NBS.sendMany sock msg

75+ sendMany sock msg

5776

5877-- | Read a length and then a payload of that length

5978recvWithLength :: N.Socket -> MaybeT IO [ByteString]

@@ -61,6 +80,15 @@ recvWithLength sock = do

6180 msgLengthBs <- recvExact sock 4

6281 decodeInt32 (BS.concat msgLengthBs) >>= recvExact sock

6382

83+-- | Wrapper around 'Network.Socket.ByteString.sendMany'

84+--

85+-- Returns Nothing when an I/O exception is raised during the send

86+sendMany :: N.Socket -> [ByteString] -> MaybeT IO ()

87+sendMany sock msg = MaybeT $ catch (pure <$> NBS.sendMany sock msg) (handleIOException)

88+ where

89+ handleIOException :: IOException -> IO (Maybe ())

90+ handleIOException _ = return mzero

91+

6492-- | Read an exact number of bytes from a socket

6593--

6694-- Returns 'Nothing' if the socket closes prematurely or the length is non-positive

network-transport-2/src/Network/Transport/TCP.hs+23−32

@@ -1,4 +1,9 @@

11-- | TCP transport

2+--

3+-- TODOs:

4+-- * Various errors are still left as "undefined"

5+-- * Many possible errors are not yet dealt with

6+-- * Connections are not yet lightweight

27module Network.Transport.TCP ( -- * Main API

38 createTransport

49 , -- TCP specific functionality

@@ -8,7 +13,9 @@ module Network.Transport.TCP ( -- * Main API

813

914import Network.Transport

1015import Network.Transport.Internal.TCP ( forkServer

16+ , connectTo

1117 , recvExact

18+ , sendMany

1219 , recvWithLength

1320 , sendWithLength

1421 )

@@ -20,19 +27,9 @@ import Network.Transport.Internal ( encodeInt16

2027import qualified Network.Socket as N ( HostName

2128 , ServiceName

2229 , Socket

23− , SocketType(Stream)

24− , SocketOption(ReuseAddr)

25− , getAddrInfo

26− , socket

27− , addrFamily

28− , addrAddress

29− , defaultProtocol

30− , setSocketOption

31− , connect

3230 , sClose

3331 , accept

3432 )

35−import qualified Network.Socket.ByteString as NBS (sendMany)

3633import Control.Concurrent (forkIO)

3734import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)

3835import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, readMVar)

@@ -109,16 +106,18 @@ endpointServer sock theirAddress chan state = forever $ do

109106 connBS <- recvExact sock 2

110107 connIx <- fromIntegral <$> decodeInt16 (BS.concat connBS)

111108 if connIx == 0

112− then liftIO $ createNewConnection

109+ then createNewConnection

113110 else readMessage connIx

114111 where

115112 -- Create a new connection

116− createNewConnection :: IO ()

113+ createNewConnection :: MaybeT IO ()

117114 createNewConnection = do

118− newIx <- modifyMVar state $ \st -> return (nextConnectionIx ^+= 1 $ st, st ^. nextConnectionIx)

119− newBs <- encodeInt16 (fromIntegral newIx)

120− NBS.sendMany sock [newBs]

121− writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

115+ (newIx, newBs) <- liftIO $ do

116+ newIx <- modifyMVar state $ \st -> return (nextConnectionIx ^+= 1 $ st, st ^. nextConnectionIx)

117+ newBs <- encodeInt16 (fromIntegral newIx)

118+ return (newIx, newBs)

119+ sendMany sock [newBs]

120+ liftIO $ writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

122121

123122 -- Read a message and output it on the endPoint's channel

124123 readMessage :: ConnectionIx -> MaybeT IO ()

@@ -149,16 +148,16 @@ tcpNewEndPoint transportState host port = do

149148tcpConnect :: EndPointAddress -> EndPointAddress -> Reliability -> IO (Either (FailedWith ConnectErrorCode) Connection)

150149tcpConnect myAddress theirAddress _ = runErrorT $ do

151150 sock <- socketForEndPoint myAddress theirAddress

152− -- TODO: test this failure

153− connIx <- maybeTToErrorT (FailedWith undefined undefined) $ requestNewConnection sock

154− return $ Connection { send = sendWithLength sock (Just connIx)

151+ connIx <- maybeTToErrorT undefined $ requestNewConnection sock

152+ return $ Connection { send = \msg -> runErrorT $ maybeTToErrorT (FailedWith SendFailed "Send failed") $

153+ sendWithLength sock (Just connIx) msg

155154 , close = N.sClose sock

156155 }

157156

158157-- | Request a new connection

159158requestNewConnection :: N.Socket -> MaybeT IO ByteString

160159requestNewConnection sock = do

161− liftIO $ pure <$> encodeInt16 0 >>= NBS.sendMany sock

160+ (liftIO $ pure <$> encodeInt16 0) >>= sendMany sock

162161 BS.concat <$> recvExact sock 2

163162

164163-- | Find a socket to the specified endpoint

@@ -171,17 +170,10 @@ socketForEndPoint :: EndPointAddress -- ^ Our address

171170 -> ErrorT (FailedWith ConnectErrorCode) IO N.Socket

172171socketForEndPoint (EndPointAddress myAddress) theirAddress = do

173172 (host, port, endPointIx) <- maybeToErrorT invalidAddress (decodeEndPointAddress theirAddress)

174− liftIO $ do

175− -- Connect to the destination transport

176− addr:_ <- N.getAddrInfo Nothing (Just host) (Just port)

177− sock <- N.socket (N.addrFamily addr) N.Stream N.defaultProtocol

178− N.setSocketOption sock N.ReuseAddr 1

179− N.connect sock (N.addrAddress addr)

180− -- Tell it what endpoint we're interested in and our own address

181− endPointBS <- encodeInt16 (fromIntegral endPointIx)

182− sendWithLength sock (Just endPointBS) [myAddress]

183− -- Connection is now established

184− return sock

173+ sock <- maybeTToErrorT undefined (connectTo host port)

174+ endPointBS <- liftIO $ encodeInt16 (fromIntegral endPointIx)

175+ maybeTToErrorT undefined $ sendWithLength sock (Just endPointBS) [myAddress]

176+ return sock

185177 where

186178 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

187179

@@ -192,4 +184,3 @@ decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack

192184 where

193185 endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointIx)

194186 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))

195−

network-transport-2/tests/TestTCP.hs+20−9

@@ -1,14 +1,15 @@

11module Main where

33import Network.Transport

4−import Network.Transport.Internal (encodeInt16)

5−import Network.Transport.Internal.TCP (sendWithLength)

4+import Network.Transport.Internal (encodeInt16, decodeInt16)

5+import Network.Transport.Internal.TCP (sendWithLength, recvExact, sendMany)

66import TestTransport (testTransport)

77import Network.Transport.TCP (createTransport, decodeEndPointAddress)

88import System.Exit (exitFailure, exitSuccess)

99import Control.Concurrent.MVar (newEmptyMVar, putMVar, takeMVar, readMVar)

1010import Control.Concurrent (forkIO)

11−import Control.Applicative ((<$>), pure)

11+-- import Control.Monad.IO.Class (liftIO)

12+import Control.Monad.Trans.Maybe (runMaybeT)

1213import Data.Maybe (fromJust)

1314import qualified Network.Socket as N ( getAddrInfo

1415 , socket

@@ -21,7 +22,6 @@ import qualified Network.Socket as N ( getAddrInfo

2122 , setSocketOption

2223 , sClose

2324 )

24−import qualified Network.Socket.ByteString as NBS (sendMany)

2525

2626testEarlyDisconnect :: IO ()

2727testEarlyDisconnect = do

@@ -34,12 +34,18 @@ testEarlyDisconnect = do

3434 Right endpoint <- newEndPoint transport

3535 putMVar serverAddr (fromJust . decodeEndPointAddress . address $ endpoint)

3636

37+ putStrLn "Server waiting for client to connect"

38+

3739 -- Wait for an incoming connection

3840 ConnectionOpened _ _ _ <- receive endpoint

3941

42+ putStrLn "Server waiting for client to disconnect"

43+

4044 -- Wait for the client to disconnect

4145 ConnectionClosed _ <- receive endpoint

4246

47+ putStrLn "Server exiting"

48+

4349 putMVar serverDone ()

4450

4551 -- Client

@@ -57,16 +63,21 @@ testEarlyDisconnect = do

5763 N.setSocketOption sock N.ReuseAddr 1

5864 N.connect sock (N.addrAddress addr)

5965

60− -- Send number of the destination endpoint and our own address

6166 (_, _, endPointIx) <- readMVar serverAddr

6267 endPointBS <- encodeInt16 (fromIntegral endPointIx)

63− sendWithLength sock (Just endPointBS) [myAddress]

68+ runMaybeT $ sendWithLength sock (Just endPointBS) [myAddress]

69+

70+ -- Request a new connection

71+ zero <- encodeInt16 0

72+ runMaybeT $ sendMany sock [zero]

6473

6574 -- Request a new connection

66− pure <$> encodeInt16 0 >>= NBS.sendMany sock

75+ connIx <- runMaybeT $ do

76+ [connBs] <- recvExact sock 2

77+ decodeInt16 connBs

78+ putStrLn $ "Client got connection ID " ++ show connIx

6779

68− -- At this point the server will try to reply with the connection ID,

69− -- but we close the connection instead

80+ -- Close the socket

7081 N.sClose sock

7182

7283 -- Wait for the server to finish

network-transport-2/tests/TestTransport.hs+3−2

@@ -1,6 +1,7 @@

11module TestTransport where

3−import Control.Concurrent (forkIO, myThreadId)

3+import Control.Concurrent (forkIO)

4+-- import Control.Concurrent (myThreadId)

45import Control.Monad (liftM2, replicateM, replicateM_, when)

56import Control.Applicative ((<$>))

67import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)

@@ -14,7 +15,7 @@ type Test = IO Bool

1415

1516-- Logging (for debugging)

1617tlog :: String -> IO ()

17−tlog msg = return ()

18+tlog _ = return ()

1819{-

1920tlog msg = do

2021 tid <- myThreadId