39d3f1e · distributed-process

git.iko.soy/distributed-process

on master help sign in

Deal with errors during sending

39d3f1eEdsko de Vries15 years ago

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

compare with master

unifiedsplit

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

17 , ConnectErrorCode(..)

18 , NewMulticastGroupErrorCode(..)

19 , ResolveMulticastGroupErrorCode(..)

20+ , SendErrorCode(..)

20 ) where

21 ) where

22import Data.ByteString (ByteString)

23import Data.ByteString (ByteString)

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

55 ResolveMulticastGroupNotFound

56 ResolveMulticastGroupNotFound

56 | ResolveMulticastGroupUnsupported

57 | ResolveMulticastGroupUnsupported

59+-- | Failure during sending a message

60+data SendErrorCode =

61+ SendFailed

59data FailedWith error = FailedWith error String

63data FailedWith error = FailedWith error String

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

88-- | Lightweight connection to an endpoint.

92-- | Lightweight connection to an endpoint.

89data Connection = Connection {

93data Connection = Connection {

90 -- | Send a message on this connection.

94 -- | Send a message on this connection.

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

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

92 -- | Close the connection.

96 -- | Close the connection.

93 , close :: IO ()

97 , close :: IO ()

94 }

98 }

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

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

73 let conn = st ^. connectionAt theirAddress

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

75 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 ())

77 , close = writeChan chan $ ConnectionClosed conn

78 , close = writeChan chan $ ConnectionClosed conn

78 }

79 }

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

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

1-- | Utility functions for TCP sockets

2module Network.Transport.Internal.TCP ( forkServer

3+ , connectTo

3 , sendWithLength

4 , sendWithLength

4 , recvWithLength

5 , recvWithLength

6+ , sendMany

5 , recvExact

7 , recvExact

6 ) where

8 ) where

10+import Prelude hiding (catch)

8import Network.Transport.Internal ( encodeInt32

11import Network.Transport.Internal ( encodeInt32

9 , decodeInt32

12 , decodeInt32

10 )

13 )

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

22 , addrAddress

25 , addrAddress

23 , defaultProtocol

26 , defaultProtocol

24 , setSocketOption

27 , setSocketOption

28+ , connect

25 )

29 )

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

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

27import Control.Concurrent (forkIO, ThreadId)

31import Control.Concurrent (forkIO, ThreadId)

28import Control.Monad (mzero)

32import Control.Monad (mzero)

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

33import 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)

31import Data.ByteString (ByteString)

37import Data.ByteString (ByteString)

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

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

33import Data.Int (Int32)

39import Data.Int (Int32)

35-- | Start a server at the specified address

41-- | Start a server at the specified address

42+--

43+-- TODO: deal with errors

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

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

37forkServer host port server = do

45forkServer host port server = do

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

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

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

45 N.listen sock 5

53 N.listen sock 5

46 forkIO $ server sock

54 forkIO $ server sock

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+

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

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

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

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

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

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

51 -> [ByteString] -- ^ Payload

70 -> [ByteString] -- ^ Payload

52− -> IO ()

71+ -> MaybeT IO ()

53sendWithLength sock header payload = do

72sendWithLength sock header payload = do

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

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

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

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

56− NBS.sendMany sock msg

75+ sendMany sock msg

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

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

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

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

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

61 msgLengthBs <- recvExact sock 4

80 msgLengthBs <- recvExact sock 4

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

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

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+

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

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

65--

93--

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

94-- 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 @@

1-- | 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

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

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

3 createTransport

8 createTransport

4 , -- TCP specific functionality

9 , -- TCP specific functionality

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

9import Network.Transport

14import Network.Transport

10import Network.Transport.Internal.TCP ( forkServer

15import Network.Transport.Internal.TCP ( forkServer

16+ , connectTo

11 , recvExact

17 , recvExact

18+ , sendMany

12 , recvWithLength

19 , recvWithLength

13 , sendWithLength

20 , sendWithLength

14 )

21 )

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

20import qualified Network.Socket as N ( HostName

27import qualified Network.Socket as N ( HostName

21 , ServiceName

28 , ServiceName

22 , Socket

29 , Socket

23− , SocketType(Stream)

24− , SocketOption(ReuseAddr)

25− , getAddrInfo

26− , socket

27− , addrFamily

28− , addrAddress

29− , defaultProtocol

30− , setSocketOption

31− , connect

32 , sClose

30 , sClose

33 , accept

31 , accept

34 )

32 )

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

36import Control.Concurrent (forkIO)

33import Control.Concurrent (forkIO)

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

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

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

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

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

109 connBS <- recvExact sock 2

106 connBS <- recvExact sock 2

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

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

111 if connIx == 0

108 if connIx == 0

112− then liftIO $ createNewConnection

109+ then createNewConnection

113 else readMessage connIx

110 else readMessage connIx

114 where

111 where

115 -- Create a new connection

112 -- Create a new connection

116− createNewConnection :: IO ()

113+ createNewConnection :: MaybeT IO ()

117 createNewConnection = do

114 createNewConnection = do

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

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

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

119− newBs <- encodeInt16 (fromIntegral newIx)

117+ newBs <- encodeInt16 (fromIntegral newIx)

118+ return (newIx, newBs)

120− NBS.sendMany sock [newBs]

119+ sendMany sock [newBs]

121− writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

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

122

121

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

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

124 readMessage :: ConnectionIx -> MaybeT IO ()

123 readMessage :: ConnectionIx -> MaybeT IO ()

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

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

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

150tcpConnect myAddress theirAddress _ = runErrorT $ do

149tcpConnect myAddress theirAddress _ = runErrorT $ do

151 sock <- socketForEndPoint myAddress theirAddress

150 sock <- socketForEndPoint myAddress theirAddress

152− -- TODO: test this failure

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

151+ connIx <- maybeTToErrorT undefined $ requestNewConnection sock

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

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

153+ sendWithLength sock (Just connIx) msg

155 , close = N.sClose sock

154 , close = N.sClose sock

156 }

155 }

157

156

158-- | Request a new connection

157-- | Request a new connection

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

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

160requestNewConnection sock = do

159requestNewConnection sock = do

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

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

162 BS.concat <$> recvExact sock 2

161 BS.concat <$> recvExact sock 2

163

162

164-- | Find a socket to the specified endpoint

163-- | Find a socket to the specified endpoint

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

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

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

172socketForEndPoint (EndPointAddress myAddress) theirAddress = do

171socketForEndPoint (EndPointAddress myAddress) theirAddress = do

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

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

174− liftIO $ do

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

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)

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

182− sendWithLength sock (Just endPointBS) [myAddress]

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

183− -- Connection is now established

184− return sock

176+ return sock

185 where

177 where

186 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

178 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

187

179

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

192 where

184 where

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

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

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

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

195−

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

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

1module Main where

3import Network.Transport

4−import Network.Transport.Internal (encodeInt16)

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

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

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

6import TestTransport (testTransport)

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

8import System.Exit (exitFailure, exitSuccess)

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

10import Control.Concurrent (forkIO)

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

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

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

12import Data.Maybe (fromJust)

13import Data.Maybe (fromJust)

13import qualified Network.Socket as N ( getAddrInfo

14import qualified Network.Socket as N ( getAddrInfo

14 , socket

15 , socket

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

21 , setSocketOption

22 , setSocketOption

22 , sClose

23 , sClose

23 )

24 )

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

26testEarlyDisconnect :: IO ()

27testEarlyDisconnect = do

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

34 Right endpoint <- newEndPoint transport

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

37+ putStrLn "Server waiting for client to connect"

38+

37 -- Wait for an incoming connection

39 -- Wait for an incoming connection

38 ConnectionOpened _ _ _ <- receive endpoint

40 ConnectionOpened _ _ _ <- receive endpoint

42+ putStrLn "Server waiting for client to disconnect"

43+

40 -- Wait for the client to disconnect

44 -- Wait for the client to disconnect

41 ConnectionClosed _ <- receive endpoint

45 ConnectionClosed _ <- receive endpoint

47+ putStrLn "Server exiting"

48+

43 putMVar serverDone ()

49 putMVar serverDone ()

45 -- Client

51 -- Client

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

57 N.setSocketOption sock N.ReuseAddr 1

63 N.setSocketOption sock N.ReuseAddr 1

58 N.connect sock (N.addrAddress addr)

64 N.connect sock (N.addrAddress addr)

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

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

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

62 endPointBS <- encodeInt16 (fromIntegral endPointIx)

67 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]

65 -- Request a new connection

74 -- 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

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

80+ -- Close the socket

69− -- but we close the connection instead

70 N.sClose sock

81 N.sClose sock

72 -- Wait for the server to finish

83 -- Wait for the server to finish

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

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

1module TestTransport where

3−import Control.Concurrent (forkIO, myThreadId)

3+import Control.Concurrent (forkIO)

4+-- import Control.Concurrent (myThreadId)

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

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

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

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

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

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

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

15-- Logging (for debugging)

16-- Logging (for debugging)

16tlog :: String -> IO ()

17tlog :: String -> IO ()

17−tlog msg = return ()

18+tlog _ = return ()

18{-

19{-

19tlog msg = do

20tlog msg = do

20 tid <- myThreadId

21 tid <- myThreadId