bf6c7a3 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Send ConnectionClosed when connection is broken

bf6c7a3Edsko de Vries15 years ago

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

compare with master

unifiedsplit

@@ -45,10 +45,10 @@ multigroupAt addr = multigroups >>> at addr

45-- (threads can, and should, create their own endpoints though).

46createTransport :: IO Transport

47createTransport = do

48− state <- newMVar $ State { _channels = Map.empty

48+ state <- newMVar State { _channels = Map.empty

49− , _connections = Map.empty

49+ , _connections = Map.empty

50− , _multigroups = Map.empty

50+ , _multigroups = Map.empty

51− }

51+ }

52 return Transport { newEndPoint = chanNewEndPoint state }

54-- Create a new end point

@@ -106,7 +106,7 @@ chanMulticastGroup state endpoint addr group =

106 cs <- readMVar group

107 forM_ cs $ \ch -> writeChan ch (ReceivedMulticast addr payload)

108 , multicastSubscribe = modifyMVar_ group $ return . (endpoint :)

109− , multicastUnsubscribe = modifyMVar_ group $ return . (filter (/= endpoint))

109+ , multicastUnsubscribe = modifyMVar_ group $ return . filter (/= endpoint)

110 , multicastClose = return ()

111 }

112

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

@@ -16,8 +16,8 @@ import Foreign.ForeignPtr (withForeignPtr)

16import Data.ByteString (ByteString)

17import qualified Data.ByteString as BS (length)

18import qualified Data.ByteString.Internal as BSI (create, toForeignPtr)

19−import Control.Monad (mzero)

19+import Control.Monad (mzero, MonadPlus)

20−import Control.Monad.IO.Class (liftIO)

20+import Control.Monad.IO.Class (MonadIO, liftIO)

21import Control.Monad.Error (ErrorT, Error, throwError, lift)

22import Control.Monad.Trans.Maybe (MaybeT, runMaybeT)

@@ -27,13 +27,13 @@ foreign import ccall unsafe "htons" htons :: CShort -> CShort

27foreign import ccall unsafe "ntohs" ntohs :: CShort -> CShort

29-- | Serialize 32-bit to network byte order

30−encodeInt32 :: Int32 -> IO ByteString

30+encodeInt32 :: (MonadIO m) => Int32 -> m ByteString

31−encodeInt32 i32 =

31+encodeInt32 i32 = liftIO $

32 BSI.create 4 $ \p ->

33 pokeByteOff p 0 (htonl (fromIntegral i32))

35-- | Deserialize 32-bit from network byte order

36−decodeInt32 :: ByteString -> MaybeT IO Int32

36+decodeInt32 :: (MonadIO m, MonadPlus m) => ByteString -> m Int32

37decodeInt32 bs | BS.length bs /= 4 = mzero

38decodeInt32 bs = liftIO $ do

39 let (fp, _, _) = BSI.toForeignPtr bs

@@ -42,13 +42,13 @@ decodeInt32 bs = liftIO $ do

42 return (fromIntegral (ntohl w32))

44-- | Serialize 16-bit to network byte order

45−encodeInt16 :: Int16 -> IO ByteString

45+encodeInt16 :: (MonadIO m) => Int16 -> m ByteString

46−encodeInt16 i16 =

46+encodeInt16 i16 = liftIO $

47 BSI.create 2 $ \p ->

48 pokeByteOff p 0 (htons (fromIntegral i16))

50-- | Deserialize 16-bit from network byte order

51−decodeInt16 :: ByteString -> MaybeT IO Int16

51+decodeInt16 :: (MonadIO m, MonadPlus m) => ByteString -> m Int16

52decodeInt16 bs | BS.length bs /= 2 = mzero

53decodeInt16 bs = liftIO $ do

54 let (fp, _, _) = BSI.toForeignPtr bs

network-transport-2/src/Network/Transport/Internal/TCP.hs+17−15

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

29 )

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

31import Control.Concurrent (forkIO, ThreadId)

32−import Control.Monad (mzero)

32+import Control.Monad (mzero, MonadPlus, unless)

33−import Control.Monad.IO.Class (liftIO)

33+import Control.Monad.IO.Class (MonadIO, liftIO)

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

35import Control.Exception (catch, IOException)

34import Control.Exception (catch, IOException)

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

37import Data.ByteString (ByteString)

35import Data.ByteString (ByteString)

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

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

39import Data.Int (Int32)

37import Data.Int (Int32)

@@ -56,7 +54,7 @@ forkServer host port server = do

56-- | Connect to another host

54-- | Connect to another host

57--

55--

58-- TODO: deal with errors

56-- TODO: deal with errors

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

57+connectTo :: (MonadIO m) => N.HostName -> N.ServiceName -> m N.Socket

60connectTo host port = liftIO $ do

58connectTo host port = liftIO $ do

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

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

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

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

@@ -65,17 +63,18 @@ connectTo host port = liftIO $ do

65 return sock

63 return sock

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

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

66+sendWithLength :: (MonadIO m, MonadPlus m)

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

67+ => N.Socket -- ^ Socket to send on

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

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

70 -> [ByteString] -- ^ Payload

69 -> [ByteString] -- ^ Payload

71− -> MaybeT IO ()

70+ -> m ()

72sendWithLength sock header payload = do

71sendWithLength sock header payload = do

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

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

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

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

75 sendMany sock msg

74 sendMany sock msg

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

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

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

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

79recvWithLength sock = do

78recvWithLength sock = do

80 msgLengthBs <- recvExact sock 4

79 msgLengthBs <- recvExact sock 4

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

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

@@ -83,18 +82,21 @@ recvWithLength sock = do

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

82-- | Wrapper around 'Network.Socket.ByteString.sendMany'

84--

83--

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

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

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

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

86+sendMany sock msg = do

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

87+ success <- liftIO $ catch (NBS.sendMany sock msg >> return True) handleIOException

88+ unless success mzero

88 where

89 where

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

90+ handleIOException :: IOException -> IO Bool

90− handleIOException _ = return mzero

91+ handleIOException _ = return False

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

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

93--

94--

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

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

96+recvExact :: (MonadIO m, MonadPlus m)

95−recvExact :: N.Socket -- ^ Socket to read from

97+ => N.Socket -- ^ Socket to read from

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

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

97− -> MaybeT IO [ByteString]

99+ -> m [ByteString]

98recvExact _ len | len <= 0 = mzero

100recvExact _ len | len <= 0 = mzero

99recvExact sock len = do

101recvExact sock len = do

100 (socketClosed, input) <- liftIO $ go [] len

102 (socketClosed, input) <- liftIO $ go [] len

@@ -107,4 +109,4 @@ recvExact sock len = do

107 bs <- NBS.recv sock (fromIntegral l `min` 4096)

109 bs <- NBS.recv sock (fromIntegral l `min` 4096)

108 if BS.null bs

110 if BS.null bs

109 then return (True, reverse acc)

111 then return (True, reverse acc)

110− else go (bs : acc) (l - (fromIntegral $ BS.length bs))

112+ else go (bs : acc) (l - fromIntegral (BS.length bs))

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

@@ -35,9 +35,11 @@ import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)

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

36import Control.Category ((>>>))

37import Control.Applicative ((<*>), (*>), (<$>), pure)

38−import Control.Monad (forever)

38+import Control.Monad (forever, forM_)

39import Control.Monad.Error (ErrorT, liftIO, runErrorT)

40import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)

41+import Control.Monad.Trans.Writer (WriterT, execWriterT)

42+import Control.Monad.Writer.Class (tell)

41import Data.ByteString (ByteString)

43import Data.ByteString (ByteString)

42import qualified Data.ByteString as BS (concat)

44import qualified Data.ByteString as BS (concat)

43import qualified Data.ByteString.Char8 as BSC (pack, unpack)

45import qualified Data.ByteString.Char8 as BSC (pack, unpack)

@@ -72,7 +74,7 @@ endPointAt ix = endPoints >>> intMapLens ix

72-- * Deal with all exceptions that may occur

74-- * Deal with all exceptions that may occur

73createTransport :: N.HostName -> N.ServiceName -> IO Transport

75createTransport :: N.HostName -> N.ServiceName -> IO Transport

74createTransport host port = do

76createTransport host port = do

75− state <- newMVar $ TransportState { _endPoints = IntMap.empty }

77+ state <- newMVar TransportState { _endPoints = IntMap.empty }

76 forkServer host port (transportServer state)

78 forkServer host port (transportServer state)

77 return Transport { newEndPoint = tcpNewEndPoint state host port }

79 return Transport { newEndPoint = tcpNewEndPoint state host port }

@@ -89,38 +91,49 @@ transportServer state sock = forever $ do

89-- Returns when the client closes the socket and on invalid input.

91-- Returns when the client closes the socket and on invalid input.

90handleClient :: MVar TransportState -> N.Socket -> IO ()

92handleClient :: MVar TransportState -> N.Socket -> IO ()

91handleClient state sock = do

93handleClient state sock = do

92− runMaybeT $ do

94+ mendpoint <- runMaybeT $ do

93 -- Our endpoint the other side is interested in

95 -- Our endpoint the other side is interested in

94 endPointBS <- recvExact sock 2

96 endPointBS <- recvExact sock 2

95− endPointIx <- decodeInt16 (BS.concat $ endPointBS)

97+ endPointIx <- decodeInt16 (BS.concat endPointBS)

96 -- Full endpoint address of the other side

98 -- Full endpoint address of the other side

97 theirAddress <- BS.concat <$> recvWithLength sock

99 theirAddress <- BS.concat <$> recvWithLength sock

98− -- Start handling incoming messages

100+ -- Find our endpoint

99 (endPointCh, endPointSt) <- MaybeT $ (^. endPointAt (fromIntegral endPointIx)) <$> readMVar state

101 (endPointCh, endPointSt) <- MaybeT $ (^. endPointAt (fromIntegral endPointIx)) <$> readMVar state

102+ return (endPointCh, endPointSt, theirAddress)

103+ case mendpoint of

104+ Just (endPointCh, endPointSt, theirAddress) -> do

100− endpointServer sock (EndPointAddress theirAddress) endPointCh endPointSt

105+ openConnections <- endpointServer sock (EndPointAddress theirAddress) endPointCh endPointSt

106+ forM_ openConnections $ \cix -> writeChan endPointCh (ConnectionClosed cix)

107+ Nothing ->

108+ -- Invalid request

101− return ()

109+ return ()

102

110

103−-- | Handle requests for one endpoint. Returns only if an error occurs

111+-- | Handle requests for one endpoint. Returns only if an error occurs,

112+-- in which case it returns the connections that are still open (normally,

113+-- this should be an empty list).

104−endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> MaybeT IO ()

114+endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> IO [ConnectionIx]

105−endpointServer sock theirAddress chan state = forever $ do

115+endpointServer sock theirAddress chan state = execWriterT . runMaybeT . forever $ do

106 connBS <- recvExact sock 2

116 connBS <- recvExact sock 2

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

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

108 if connIx == 0

118 if connIx == 0

109 then createNewConnection

119 then createNewConnection

110 else readMessage connIx

120 else readMessage connIx

111 where

121 where

122+ -- A brief note on the types:

123+ -- These functions may fail (connection error): hence MaybeT

124+ -- We also need to keep track of which connections are still open: hence WriterT

125+

112 -- Create a new connection

126 -- Create a new connection

113− createNewConnection :: MaybeT IO ()

127+ createNewConnection :: MaybeT (WriterT [ConnectionIx] IO) ()

114 createNewConnection = do

128 createNewConnection = do

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

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

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

117− newBs <- encodeInt16 (fromIntegral newIx)

130+ newBs <- encodeInt16 (fromIntegral newIx)

118− return (newIx, newBs)

119 sendMany sock [newBs]

131 sendMany sock [newBs]

120 liftIO $ writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

132 liftIO $ writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

133+ tell [newIx]

121

134

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

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

123− readMessage :: ConnectionIx -> MaybeT IO ()

136+ readMessage :: ConnectionIx -> MaybeT (WriterT [ConnectionIx] IO) ()

124 readMessage connIx = do

137 readMessage connIx = do

125 payload <- recvWithLength sock

138 payload <- recvWithLength sock

126 liftIO $ writeChan chan (Received connIx payload)

139 liftIO $ writeChan chan (Received connIx payload)

@@ -129,7 +142,7 @@ endpointServer sock theirAddress chan state = forever $ do

129tcpNewEndPoint :: MVar TransportState -> N.HostName -> N.ServiceName -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)

142tcpNewEndPoint :: MVar TransportState -> N.HostName -> N.ServiceName -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)

130tcpNewEndPoint transportState host port = do

143tcpNewEndPoint transportState host port = do

131 chan <- newChan

144 chan <- newChan

132− endPointState <- newMVar $ EndPointState { _nextConnectionIx = 1 } -- Connection id 0 is reserved for requesting new connections

145+ endPointState <- newMVar EndPointState { _nextConnectionIx = 1 } -- Connection id 0 is reserved for requesting new connections

133 endPointIx <- modifyMVar transportState $ \st -> do let ix = IntMap.size (st ^. endPoints)

146 endPointIx <- modifyMVar transportState $ \st -> do let ix = IntMap.size (st ^. endPoints)

134 return ((endPointAt ix ^= Just (chan, endPointState)) $ st, ix)

147 return ((endPointAt ix ^= Just (chan, endPointState)) $ st, ix)

135 let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointIx

148 let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointIx

@@ -149,15 +162,16 @@ tcpConnect :: EndPointAddress -> EndPointAddress -> Reliability -> IO (Either (F

149tcpConnect myAddress theirAddress _ = runErrorT $ do

162tcpConnect myAddress theirAddress _ = runErrorT $ do

150 sock <- socketForEndPoint myAddress theirAddress

163 sock <- socketForEndPoint myAddress theirAddress

151 connIx <- maybeTToErrorT undefined $ requestNewConnection sock

164 connIx <- maybeTToErrorT undefined $ requestNewConnection sock

165+ return Connection { send = runErrorT .

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

166+ maybeTToErrorT (FailedWith SendFailed "Send failed") .

153− sendWithLength sock (Just connIx) msg

167+ sendWithLength sock (Just connIx)

154− , close = N.sClose sock

168+ , close = N.sClose sock

155− }

169+ }

156

170

157-- | Request a new connection

171-- | Request a new connection

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

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

159requestNewConnection sock = do

173requestNewConnection sock = do

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

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

161 BS.concat <$> recvExact sock 2

175 BS.concat <$> recvExact sock 2

162

176

163-- | Find a socket to the specified endpoint

177-- | Find a socket to the specified endpoint

network-transport-2/tests/TestTCP.hs+5−4

@@ -27,7 +27,7 @@ testEarlyDisconnect :: IO ()

27testEarlyDisconnect = do

28 serverAddr <- newEmptyMVar

29 serverDone <- newEmptyMVar

30− transport <- createTransport "127.0.0.1" "8080"

30+ transport <- createTransport "127.0.0.1" "8081"

32 -- Server

33 forkIO $ do

@@ -58,7 +58,7 @@ testEarlyDisconnect = do

58 let EndPointAddress myAddress = address endpoint

60 -- Connect to the server

61− addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8080")

61+ addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")

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)

@@ -77,7 +77,8 @@ testEarlyDisconnect = do

77 decodeInt16 connBs

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

80− -- Close the socket

80+ -- Close the socket without closing the connection explicitly

81+ -- The server should still receive a ConnectionClosed message

81 N.sClose sock

82 N.sClose sock

83 -- Wait for the server to finish

84 -- Wait for the server to finish

@@ -85,6 +86,6 @@ testEarlyDisconnect = do

86main :: IO ()

87main :: IO ()

87main = do

88main = do

89+ testEarlyDisconnect

88 success <- createTransport "127.0.0.1" "8080" >>= testTransport

90 success <- createTransport "127.0.0.1" "8080" >>= testTransport

89 if success then exitSuccess else exitFailure

91 if success then exitSuccess else exitFailure

90− -- testEarlyDisconnect

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

@@ -64,7 +64,7 @@ ping endpoint server numPings msg = do

64 case event of

65 Received _ [reply] | reply == msg ->

66 return ()

67− _ -> do

67+ _ ->

68 error $ "Unexpected event " ++ show event

70 -- Close the connection