478d769 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Continue work on error handling

478d769Edsko de Vries15 years ago

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

compare with master

unifiedsplit

@@ -47,7 +47,7 @@ data NewEndPointErrorCode

47-- | Connection failure

48data ConnectErrorCode =

49 ConnectInvalidAddress -- ^ Could not parse or resolve the address

50− | ConnectInsufficientResources -- ^ Insufficient resources (for instance, no more socketss available)

50+ | ConnectInsufficientResources -- ^ Insufficient resources (for instance, no more sockets available)

51 | ConnectFailed -- ^ Failed for other reasons

52 deriving Show

network-transport-2/src/Network/Transport/Internal.hs+10

@@ -8,6 +8,7 @@ module Network.Transport.Internal ( -- * Encoders/decoders

8 -- * Miscellaneous abstractions

9 , failWith

10 , failWithIO

11+ , failWithMaybeIO

11 ) where

12 ) where

13import Prelude hiding (catch)

14import Prelude hiding (catch)

@@ -20,6 +21,7 @@ import qualified Data.ByteString.Internal as BSI (unsafeCreate, toForeignPtr, in

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

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

21import Control.Monad.IO.Class (MonadIO, liftIO)

22import Control.Monad.IO.Class (MonadIO, liftIO)

22import Control.Monad.Error (MonadError, throwError)

23import Control.Monad.Error (MonadError, throwError)

24+import Control.Monad.Trans.Maybe (MaybeT, runMaybeT)

23import Control.Exception (IOException, catch)

25import Control.Exception (IOException, catch)

25foreign import ccall unsafe "htonl" htonl :: CInt -> CInt

27foreign import ccall unsafe "htonl" htonl :: CInt -> CInt

@@ -77,3 +79,11 @@ failWithIO f io = do

77 where

79 where

78 handleIOException :: IOException -> IO (Either IOException a)

80 handleIOException :: IOException -> IO (Either IOException a)

79 handleIOException = return . Left

81 handleIOException = return . Left

82+

83+-- | Like 'failWithIO' but throw an error on a return of 'Nothing' too

84+failWithMaybeIO :: (MonadIO m, MonadError e m) => (Maybe IOException -> e) -> MaybeT IO a -> m a

85+failWithMaybeIO f io = do

86+ ma <- failWithIO (f . Just) $ runMaybeT io

87+ case ma of

88+ Nothing -> throwError (f Nothing)

89+ Just a -> return a

network-transport-2/src/Network/Transport/Internal/TCP.hs+11−13

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

1-- | Utility functions for TCP sockets

2module Network.Transport.Internal.TCP ( forkServer

3 , recvWithLength

4− , sendMany

5 , recvExact

4 , recvExact

6 , recvInt32

5 , recvInt32

6+ , sendMany

7 ) where

9import Prelude hiding (catch)

10−import Network.Transport.Internal (decodeInt32, failWithIO)

10+import Network.Transport.Internal (decodeInt32)

11import qualified Network.Socket as N ( HostName

12 , ServiceName

13 , Socket

@@ -23,9 +23,9 @@ import qualified Network.Socket as N ( HostName

23 , defaultProtocol

24 , setSocketOption

25 )

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

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

27import Control.Concurrent (forkIO, ThreadId)

28−import Control.Monad (mzero, MonadPlus)

28+import Control.Monad (mzero, MonadPlus, liftM)

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

30import Control.Monad.Error (MonadError)

31import Control.Exception (catch, IOException)

@@ -36,7 +36,7 @@ import Data.Int (Int32)

36-- | Start a server at the specified address

37forkServer :: (MonadIO m, MonadError IOException m)

38 => N.HostName -> N.ServiceName -> (N.Socket -> IO ()) -> m ThreadId

39−forkServer host port server = failWithIO id $ do

39+forkServer host port server = liftIO $ do

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

41 -- return an empty list (but will throw an exception instead) and will return

42 -- the "best" address first, whatever that means

@@ -47,6 +47,10 @@ forkServer host port server = failWithIO id $ do

47 N.listen sock 5

48 forkIO $ server sock

50+-- | Lifted version of 'Network.Socket.ByteString.sendMany'

51+sendMany :: (MonadIO m) => N.Socket -> [ByteString] -> m ()

52+sendMany sock msg = liftIO $ NBS.sendMany sock msg

53+

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

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

51recvWithLength :: (MonadIO m, MonadPlus m) => N.Socket -> m [ByteString]

55recvWithLength :: (MonadIO m, MonadPlus m) => N.Socket -> m [ByteString]

52recvWithLength sock = recvInt32 sock >>= recvExact sock

56recvWithLength sock = recvInt32 sock >>= recvExact sock

@@ -54,20 +58,14 @@ recvWithLength sock = recvInt32 sock >>= recvExact sock

54-- | Receive a 32-bit integer

58-- | Receive a 32-bit integer

55recvInt32 :: (Enum a, MonadIO m, MonadPlus m) => N.Socket -> m a

59recvInt32 :: (Enum a, MonadIO m, MonadPlus m) => N.Socket -> m a

56recvInt32 sock = do

60recvInt32 sock = do

57− mi <- recvExact sock 4 >>= return . decodeInt32 . BS.concat

61+ mi <- liftM (decodeInt32 . BS.concat) $ recvExact sock 4

58 case mi of

62 case mi of

59 Nothing -> mzero

63 Nothing -> mzero

60 Just i -> return i

64 Just i -> return i

62−-- | Wrapper around 'Network.Socket.ByteString.sendMany'

63−--

64−-- Fails when an I/O exception is raised during the send

65−sendMany :: (MonadIO m, MonadError IOException m) => N.Socket -> [ByteString] -> m ()

66−sendMany sock msg = failWithIO id $ NBS.sendMany sock msg

67−

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

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

69--

67--

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

68+-- Fails if the socket closes prematurely or the length is non-positive

71recvExact :: (MonadIO m, MonadPlus m)

69recvExact :: (MonadIO m, MonadPlus m)

72 => N.Socket -- ^ Socket to read from

70 => N.Socket -- ^ Socket to read from

73 -> Int32 -- ^ Number of bytes to read

71 -> Int32 -- ^ Number of bytes to read

network-transport-2/src/Network/Transport/TCP.hs+36−24

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

9 , -- TCP specific functionality

10 EndPointId

11 , decodeEndPointAddress

12+ , ControlHeader(..)

12 ) where

13 ) where

14import Network.Transport

15import Network.Transport

@@ -22,6 +23,7 @@ import Network.Transport.Internal ( encodeInt32

22 , prependLength

23 , prependLength

23 , failWith

24 , failWith

24 , failWithIO

25 , failWithIO

26+ , failWithMaybeIO

25 )

27 )

26import qualified Network.Socket as N ( HostName

28import qualified Network.Socket as N ( HostName

27 , ServiceName

29 , ServiceName

@@ -104,12 +106,12 @@ createTransport host port = do

104 state <- newMVar TransportState { _localEndPoints = IntMap.empty

106 state <- newMVar TransportState { _localEndPoints = IntMap.empty

105 , _remoteEndPoints = Map.empty

107 , _remoteEndPoints = Map.empty

106 }

108 }

107− let transport = TCPTransport { _transportState = state

109+ let transport = TCPTransport { _transportState = state

108− , _transportHost = host

110+ , _transportHost = host

109− , _transportPort = port

111+ , _transportPort = port

110− }

112+ }

111 runErrorT $ do

113 runErrorT $ do

112− forkServer host port (transportServer transport)

114+ failWithIO id $ forkServer host port (transportServer transport)

113 return Transport { newEndPoint = tcpNewEndPoint transport }

115 return Transport { newEndPoint = tcpNewEndPoint transport }

114

116

115-- | The transport server handles incoming connections for all endpoints

117-- | The transport server handles incoming connections for all endpoints

@@ -129,13 +131,16 @@ handleConnectionRequest transport sock = do

129 ourEndPoint <- MaybeT $ (^. localEndPointAt ourEndPointId) <$> readMVar (transport ^. transportState)

131 ourEndPoint <- MaybeT $ (^. localEndPointAt ourEndPointId) <$> readMVar (transport ^. transportState)

130 return (ourEndPoint, theirAddress)

132 return (ourEndPoint, theirAddress)

131 case request of

133 case request of

132− Just (ourEndPoint, theirAddress) ->

134+ Just (ourEndPoint, theirAddress) -> do

135+ -- Send acknowledgement

136+ sendMany sock [encodeInt32 ValidEndPoint]

133 -- We ignore the return value of 'forkRemoteEndPoint' here because _they_ are

137 -- We ignore the return value of 'forkRemoteEndPoint' here because _they_ are

134 -- connecting to _us_. 'forkRemoteEndPoint' will have stored the remote endpoint

138 -- connecting to _us_. 'forkRemoteEndPoint' will have stored the remote endpoint

135 -- in the transport state, however, in case _we_ later connect to _them_.

139 -- in the transport state, however, in case _we_ later connect to _them_.

136 void $ forkRemoteEndPoint transport ourEndPoint theirAddress sock

140 void $ forkRemoteEndPoint transport ourEndPoint theirAddress sock

137 Nothing ->

141 Nothing ->

138− return ()

142+ -- Invalid request

143+ N.sClose sock

139

144

140-- | Set up a new remote endpoint (either because we connected to a

145-- | Set up a new remote endpoint (either because we connected to a

141-- new remote endpoint, or because a remote endpoint connected to us)

146-- new remote endpoint, or because a remote endpoint connected to us)

@@ -159,8 +164,9 @@ forkRemoteEndPoint transport ourEndPoint theirAddress sock = do

159 return theirEndPoint

164 return theirEndPoint

160

165

161-- Control headers

166-- Control headers

162−data ControlHeaders =

167+data ControlHeader =

168+ ValidEndPoint -- ^ Acknowledgement sent when setting up a new TCP connection

163− RequestConnectionId -- ^ Request a new connection ID from the remote endpoint

169+ | RequestConnectionId -- ^ Request a new connection ID from the remote endpoint

164 | ControlResponse -- ^ Respond to a control request _from_ the remote endpoint

170 | ControlResponse -- ^ Respond to a control request _from_ the remote endpoint

165 deriving Enum

171 deriving Enum

166

172

@@ -188,7 +194,7 @@ handleIncomingMessages ourEndPoint theirEndPoint = execWriterT . runMaybeT . for

188 createNewConnection :: ControlRequestId -> MaybeT (WriterT [ConnectionId] IO) ()

194 createNewConnection :: ControlRequestId -> MaybeT (WriterT [ConnectionId] IO) ()

189 createNewConnection reqId = do

195 createNewConnection reqId = do

190 newId <- liftIO $ getNextConnectionId ourEndPoint

196 newId <- liftIO $ getNextConnectionId ourEndPoint

191− withRemoteLock theirEndPoint $ do

197+ withRemoteLock theirEndPoint $

192 sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 newId])

198 sendMany sock (encodeInt32 ControlResponse : encodeInt32 reqId : prependLength [encodeInt32 newId])

193 liftIO $ writeChan ourChannel (ConnectionOpened newId ReliableOrdered (theirEndPoint ^. remoteAddress))

199 liftIO $ writeChan ourChannel (ConnectionOpened newId ReliableOrdered (theirEndPoint ^. remoteAddress))

194 -- We add the new connection ID to the list of open connections only once the

200 -- We add the new connection ID to the list of open connections only once the

@@ -258,18 +264,18 @@ tcpConnect :: TCPTransport -- ^ Transport

258 -> Reliability -- ^ Reliability (ignored)

264 -> Reliability -- ^ Reliability (ignored)

259 -> IO (Either (FailedWith ConnectErrorCode) Connection)

265 -> IO (Either (FailedWith ConnectErrorCode) Connection)

260tcpConnect transport ourEndPoint theirAddress _ = runErrorT $ do

266tcpConnect transport ourEndPoint theirAddress _ = runErrorT $ do

261− theirEndPoint <- remoteEndPoint transport ourEndPoint theirAddress

267+ theirEndPoint <- remoteEndPoint transport ourEndPoint theirAddress

262− connId <- requestNewConnection ourEndPoint theirEndPoint

268+ connId <- requestNewConnection ourEndPoint theirEndPoint

263− let sock = theirEndPoint ^. remoteSocket

269+ let sock = theirEndPoint ^. remoteSocket

264− return Connection { send = runErrorT .

270+ return Connection { send = runErrorT .

265− failWithIO (FailedWith SendFailed . show) .

271+ failWithIO (FailedWith SendFailed . show) .

266− withRemoteLock theirEndPoint .

272+ withRemoteLock theirEndPoint .

267− sendMany sock .

273+ sendMany sock .

268− (\payload -> encodeInt32 connId : prependLength payload)

274+ (\payload -> encodeInt32 connId : prependLength payload)

269− , close = -- TODO: we should not close the TCP connection until all

275+ , close = -- TODO: we should not close the TCP connection until all

270− -- lightweight connections have been closed

276+ -- lightweight connections have been closed

271− N.sClose sock

277+ N.sClose sock

272− }

278+ }

273

279

274-- | Request a new connection

280-- | Request a new connection

275requestNewConnection :: (MonadIO m, MonadError (FailedWith ConnectErrorCode) m)

281requestNewConnection :: (MonadIO m, MonadError (FailedWith ConnectErrorCode) m)

@@ -310,7 +316,7 @@ remoteEndPoint transport ourEndPoint theirAddress = do

310 -- Create a new connection

316 -- Create a new connection

311 -- liftIO $ putStrLn $ "Creating socket from " ++ show ourAddress ++ " to " ++ show theirAddress

317 -- liftIO $ putStrLn $ "Creating socket from " ++ show ourAddress ++ " to " ++ show theirAddress

312 (host, port, endPointId) <- failWith (invalidAddress "Could not parse") $

318 (host, port, endPointId) <- failWith (invalidAddress "Could not parse") $

313− (decodeEndPointAddress theirAddress)

319+ decodeEndPointAddress theirAddress

314 addr:_ <- failWithIO (invalidAddress . show) $

320 addr:_ <- failWithIO (invalidAddress . show) $

315 N.getAddrInfo Nothing (Just host) (Just port)

321 N.getAddrInfo Nothing (Just host) (Just port)

316 sock <- failWithIO (insufficientResources . show) $

322 sock <- failWithIO (insufficientResources . show) $

@@ -319,7 +325,11 @@ remoteEndPoint transport ourEndPoint theirAddress = do

319 failWithIO (failed . show) $ N.connect sock (N.addrAddress addr)

325 failWithIO (failed . show) $ N.connect sock (N.addrAddress addr)

320 -- Send the endpoint ID we're interested in and our own address

326 -- Send the endpoint ID we're interested in and our own address

321 failWithIO (failed . show) $ sendMany sock (encodeInt32 endPointId : prependLength [ourAddress])

327 failWithIO (failed . show) $ sendMany sock (encodeInt32 endPointId : prependLength [ourAddress])

328+ -- For the remote endpoint to acknowledge

329+ ack <- failWithMaybeIO (invalidAddress . show) $ recvInt32 sock

330+ case ack of

322− liftIO $ forkRemoteEndPoint transport ourEndPoint theirAddress sock

331+ ValidEndPoint -> liftIO $ forkRemoteEndPoint transport ourEndPoint theirAddress sock

332+ _ -> throwError (invalidAddress "Unexpected response")

323 where

333 where

324 invalidAddress = FailedWith ConnectInvalidAddress

334 invalidAddress = FailedWith ConnectInvalidAddress

325 insufficientResources = FailedWith ConnectInsufficientResources

335 insufficientResources = FailedWith ConnectInsufficientResources

@@ -399,3 +409,5 @@ remoteEndPointAt addr = remoteEndPoints >>> mapLens addr

399

409

400pendingCtrlRequestsAt :: ControlRequestId -> Lens EndPointState (Maybe (MVar [ByteString]))

410pendingCtrlRequestsAt :: ControlRequestId -> Lens EndPointState (Maybe (MVar [ByteString]))

401pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)

411pendingCtrlRequestsAt ix = pendingCtrlRequests >>> intMapLens (fromIntegral ix)

412+

413+

network-transport-2/tests/TestTCP.hs+11−14

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

1module Main where

3import Network.Transport

4−import Network.Transport.Internal (encodeInt32, decodeInt32, prependLength)

4+import Network.Transport.Internal (encodeInt32, prependLength)

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

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

6−import Network.Transport.TCP (createTransport, decodeEndPointAddress, EndPointId)

6+import Network.Transport.TCP (createTransport, decodeEndPointAddress, EndPointId, ControlHeader(..))

7import TestTransport (testTransport)

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

9import Control.Concurrent (forkIO)

10import Control.Monad.Trans.Maybe (runMaybeT)

11import Data.Maybe (fromJust)

12+import Data.Int (Int32)

12import qualified Network.Socket as N ( getAddrInfo

13import qualified Network.Socket as N ( getAddrInfo

13 , socket

14 , socket

14 , connect

15 , connect

@@ -25,8 +26,6 @@ import qualified Network.Socket as N ( getAddrInfo

26testEarlyDisconnect :: IO ()

27testEarlyDisconnect :: IO ()

27testEarlyDisconnect = do

28testEarlyDisconnect = do

28− return ()

29−{-

30 serverAddr <- newEmptyMVar

29 serverAddr <- newEmptyMVar

31 serverDone <- newEmptyMVar

30 serverDone <- newEmptyMVar

32 Right transport <- createTransport "127.0.0.1" "8081"

31 Right transport <- createTransport "127.0.0.1" "8081"

@@ -57,17 +56,17 @@ testEarlyDisconnect = do

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

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

59 runMaybeT $ sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])

58 runMaybeT $ sendMany sock (encodeInt32 endPointIx : prependLength [myAddress])

59+

60+ -- Wait for acknowledgement

61+ Just ValidEndPoint <- runMaybeT $ recvInt32 sock

61− -- Request a new connection

63+ -- Request a new connection, but don't wait for the response

62− runMaybeT $ do

64+ let reqId = 0 :: Int32

63− sendInt32 sock 0

65+ runMaybeT $ sendMany sock [encodeInt32 RequestConnectionId, encodeInt32 reqId]

64− [connBs] <- recvExact sock 2

65− decodeInt16 connBs

67 -- Close the socket without closing the connection explicitly

68 -- The server should still receive a ConnectionClosed message

69 N.sClose sock

70−-}

72testInvalidAddress :: IO ()

71testInvalidAddress :: IO ()

73testInvalidAddress = do

72testInvalidAddress = do

@@ -95,11 +94,9 @@ testInvalidConnect = do

95 putStrLn $ "Got expected error: " ++ show err3

94 putStrLn $ "Got expected error: " ++ show err3

97 -- Valid TCP address but invalid endpoint number

96 -- Valid TCP address but invalid endpoint number

98− {-

99− Left (FailedWith ConnectFailed err4) <-

97+ Left (FailedWith ConnectInvalidAddress err4) <-

100 connect endpoint (EndPointAddress "127.0.0.1:8083:1") ReliableOrdered

98 connect endpoint (EndPointAddress "127.0.0.1:8083:1") ReliableOrdered

101 putStrLn $ "Got expected error: " ++ show err4

99 putStrLn $ "Got expected error: " ++ show err4

102− -}

103

100

104main :: IO ()

101main :: IO ()

105main = do

102main = do