478d769 · distributed-process

git.iko.soy/distributed-process

Continue work on error handling

commit478d7694069ba992a82014a0894908ce60ba4ac4 authorEdsko de Vries <edsko@edsko.net> dateThu 26 Apr 2012, 14:47 (15 years ago) parentc067c47

5 files changed+69−52

compare with master

unifiedsplit

network-transport-2/src/Network/Transport.hs+1−1 network-transport-2/src/Network/Transport/Internal.hs+10 network-transport-2/src/Network/Transport/Internal/TCP.hs+11−13 network-transport-2/src/Network/Transport/TCP.hs+36−24 network-transport-2/tests/TestTCP.hs+11−14

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

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

4747-- | Connection failure

4848data ConnectErrorCode =

4949 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)

5151 | ConnectFailed -- ^ Failed for other reasons

5252 deriving Show

5353

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

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

88 -- * Miscellaneous abstractions

99 , failWith

1010 , failWithIO

11+ , failWithMaybeIO

1112 ) where

1213

1314import Prelude hiding (catch)

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

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

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

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

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

2325import Control.Exception (IOException, catch)

2426

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

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

7779 where

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

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

11-- | Utility functions for TCP sockets

22module Network.Transport.Internal.TCP ( forkServer

33 , recvWithLength

4− , sendMany

54 , recvExact

65 , recvInt32

6+ , sendMany

77 ) where

99import Prelude hiding (catch)

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

10+import Network.Transport.Internal (decodeInt32)

1111import qualified Network.Socket as N ( HostName

1212 , ServiceName

1313 , Socket

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

2323 , defaultProtocol

2424 , setSocketOption

2525 )

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

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

2727import Control.Concurrent (forkIO, ThreadId)

28−import Control.Monad (mzero, MonadPlus)

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

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

3030import Control.Monad.Error (MonadError)

3131import Control.Exception (catch, IOException)

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

3636-- | Start a server at the specified address

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

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

39−forkServer host port server = failWithIO id $ do

39+forkServer host port server = liftIO $ do

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

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

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

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

4747 N.listen sock 5

4848 forkIO $ server sock

4949

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+

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

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

5256recvWithLength sock = recvInt32 sock >>= recvExact sock

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

5458-- | Receive a 32-bit integer

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

5660recvInt32 sock = do

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

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

5862 case mi of

5963 Nothing -> mzero

6064 Just i -> return i

6165

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−

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

6967--

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

7169recvExact :: (MonadIO m, MonadPlus m)

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

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

99 , -- TCP specific functionality

1010 EndPointId

1111 , decodeEndPointAddress

12+ , ControlHeader(..)

1213 ) where

1314

1415import Network.Transport

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

2223 , prependLength

2324 , failWith

2425 , failWithIO

26+ , failWithMaybeIO

2527 )

2628import qualified Network.Socket as N ( HostName

2729 , ServiceName

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

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

105107 , _remoteEndPoints = Map.empty

106108 }

107− let transport = TCPTransport { _transportState = state

108− , _transportHost = host

109− , _transportPort = port

110− }

109+ let transport = TCPTransport { _transportState = state

110+ , _transportHost = host

111+ , _transportPort = port

112+ }

111113 runErrorT $ do

112− forkServer host port (transportServer transport)

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

113115 return Transport { newEndPoint = tcpNewEndPoint transport }

114116

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

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

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

130132 return (ourEndPoint, theirAddress)

131133 case request of

132− Just (ourEndPoint, theirAddress) ->

134+ Just (ourEndPoint, theirAddress) -> do

135+ -- Send acknowledgement

136+ sendMany sock [encodeInt32 ValidEndPoint]

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

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

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

136140 void $ forkRemoteEndPoint transport ourEndPoint theirAddress sock

137141 Nothing ->

138− return ()

142+ -- Invalid request

143+ N.sClose sock

139144

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

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

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

159164 return theirEndPoint

160165

161166-- Control headers

162−data ControlHeaders =

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

167+data ControlHeader =

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

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

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

165171 deriving Enum

166172

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

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

189195 createNewConnection reqId = do

190196 newId <- liftIO $ getNextConnectionId ourEndPoint

191− withRemoteLock theirEndPoint $ do

197+ withRemoteLock theirEndPoint $

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

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

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

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

258264 -> Reliability -- ^ Reliability (ignored)

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

260266tcpConnect transport ourEndPoint theirAddress _ = runErrorT $ do

261− theirEndPoint <- remoteEndPoint transport ourEndPoint theirAddress

262− connId <- requestNewConnection ourEndPoint theirEndPoint

263− let sock = theirEndPoint ^. remoteSocket

264− return Connection { send = runErrorT .

265− failWithIO (FailedWith SendFailed . show) .

266− withRemoteLock theirEndPoint .

267− sendMany sock .

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

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

270− -- lightweight connections have been closed

271− N.sClose sock

272− }

267+ theirEndPoint <- remoteEndPoint transport ourEndPoint theirAddress

268+ connId <- requestNewConnection ourEndPoint theirEndPoint

269+ let sock = theirEndPoint ^. remoteSocket

270+ return Connection { send = runErrorT .

271+ failWithIO (FailedWith SendFailed . show) .

272+ withRemoteLock theirEndPoint .

273+ sendMany sock .

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

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

276+ -- lightweight connections have been closed

277+ N.sClose sock

278+ }

273279

274280-- | Request a new connection

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

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

310316 -- Create a new connection

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

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

313− (decodeEndPointAddress theirAddress)

319+ decodeEndPointAddress theirAddress

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

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

316322 sock <- failWithIO (insufficientResources . show) $

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

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

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

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

322− liftIO $ forkRemoteEndPoint transport ourEndPoint theirAddress sock

328+ -- For the remote endpoint to acknowledge

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

330+ case ack of

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

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

323333 where

324334 invalidAddress = FailedWith ConnectInvalidAddress

325335 insufficientResources = FailedWith ConnectInsufficientResources

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

399409

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

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

412+

413+

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

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

11module Main where

33import Network.Transport

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

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

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

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

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

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

77import TestTransport (testTransport)

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

99import Control.Concurrent (forkIO)

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

1111import Data.Maybe (fromJust)

12+import Data.Int (Int32)

1213import qualified Network.Socket as N ( getAddrInfo

1314 , socket

1415 , connect

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

2526

2627testEarlyDisconnect :: IO ()

2728testEarlyDisconnect = do

28− return ()

29−{-

3029 serverAddr <- newEmptyMVar

3130 serverDone <- newEmptyMVar

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

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

5756

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

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

59+

60+ -- Wait for acknowledgement

61+ Just ValidEndPoint <- runMaybeT $ recvInt32 sock

6062

61− -- Request a new connection

62− runMaybeT $ do

63− sendInt32 sock 0

64− [connBs] <- recvExact sock 2

65− decodeInt16 connBs

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

64+ let reqId = 0 :: Int32

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

6666

6767 -- Close the socket without closing the connection explicitly

6868 -- The server should still receive a ConnectionClosed message

6969 N.sClose sock

70−-}

7170

7271testInvalidAddress :: IO ()

7372testInvalidAddress = do

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

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

9695

9796 -- Valid TCP address but invalid endpoint number

98− {-

99− Left (FailedWith ConnectFailed err4) <-

97+ Left (FailedWith ConnectInvalidAddress err4) <-

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

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

102− -}

103100

104101main :: IO ()

105102main = do