3938dfc · distributed-process

git.iko.soy/distributed-process

on master help sign in

Reorganize code a bit

3938dfcEdsko de Vries15 years ago

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

compare with master

unifiedsplit

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

15 , FailedWith(..)

16 , NewEndPointErrorCode

17 , ConnectErrorCode(..)

18− , NewMulticastGroupErrorCode

18+ , NewMulticastGroupErrorCode(..)

19 , ResolveMulticastGroupErrorCode(..)

20 ) where

@@ -47,11 +47,14 @@ data ConnectErrorCode =

47 ConnectInvalidAddress -- ^ Could not parse the address

49-- | Failure during the creation of a new multicast group

50−data NewMulticastGroupErrorCode

50+data NewMulticastGroupErrorCode =

51+ NewMulticastGroupUnsupported

52-- | Failure during the resolution of a multicast group

53-- | Failure during the resolution of a multicast group

53data ResolveMulticastGroupErrorCode =

54data ResolveMulticastGroupErrorCode =

54− MulticastGroupNotFound

55+ ResolveMulticastGroupNotFound

56+ | ResolveMulticastGroupUnsupported

57+

56data FailedWith error = FailedWith error String

59data FailedWith error = FailedWith error String

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

@@ -117,5 +117,5 @@ chanResolveMulticastGroup :: MVar TransportState

117chanResolveMulticastGroup state endpoint addr = do

118 group <- (^. (multigroups >>> mapLens addr)) <$> readMVar state

119 case group of

120− Nothing -> return . Left $ FailedWith MulticastGroupNotFound ("Group " ++ show addr ++ " not found")

120+ Nothing -> return . Left $ FailedWith ResolveMulticastGroupNotFound ("Group " ++ show addr ++ " not found")

121 Just mvar -> return . Right $ chanMulticastGroup state endpoint addr mvar

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

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

6 , decodeInt16

7 -- * Miscellaneous abstractions

8 , maybeToErrorT

9+ , maybeTToErrorT

9 ) where

10 ) where

11import Data.Int (Int16, Int32)

12import Data.Int (Int16, Int32)

@@ -17,8 +18,8 @@ import qualified Data.ByteString as BS (length)

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

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

18import Control.Monad (mzero)

19import Control.Monad (mzero)

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

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

20−import Control.Monad.Error (ErrorT, Error, throwError)

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

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

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

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

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

24foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt

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

@@ -55,7 +56,15 @@ decodeInt16 bs = liftIO $ do

55 w16 <- peekByteOff p 0

56 w16 <- peekByteOff p 0

56 return (fromIntegral (ntohs w16))

57 return (fromIntegral (ntohs w16))

58−-- | Convert maybe to an ErrorT value

59+-- | Convert 'Maybe' to an ErrorT value

59maybeToErrorT :: (Monad m, Error a) => a -> Maybe b -> ErrorT a m b

60maybeToErrorT :: (Monad m, Error a) => a -> Maybe b -> ErrorT a m b

60maybeToErrorT err Nothing = throwError err

61maybeToErrorT err Nothing = throwError err

61maybeToErrorT _ (Just x) = return x

62maybeToErrorT _ (Just x) = return x

63+

64+-- | Convert 'MaybeT' to an 'ErrorT'

65+maybeTToErrorT :: (Monad m, Error a) => a -> MaybeT m b -> ErrorT a m b

66+maybeTToErrorT err valueT = do

67+ mval <- lift $ runMaybeT valueT

68+ case mval of

69+ Nothing -> throwError err

70+ Just val -> return val

network-transport-2/src/Network/Transport/Internal/TCP.hsnew file+82

@@ -0,0 +1,82 @@

1+-- | Utility functions for TCP sockets

2+module Network.Transport.Internal.TCP ( forkServer

3+ , sendWithLength

4+ , recvWithLength

5+ , recvExact

6+ ) where

8+import Network.Transport.Internal ( encodeInt32

9+ , decodeInt32

10+ )

11+import qualified Network.Socket as N ( HostName

12+ , ServiceName

13+ , Socket

14+ , SocketType(Stream)

15+ , SocketOption(ReuseAddr)

16+ , getAddrInfo

17+ , defaultHints

18+ , socket

19+ , bindSocket

20+ , listen

21+ , addrFamily

22+ , addrAddress

23+ , defaultProtocol

24+ , setSocketOption

25+ )

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

27+import Control.Concurrent (forkIO, ThreadId)

28+import Control.Monad (mzero)

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

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

31+import Data.ByteString (ByteString)

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

33+import Data.Int (Int32)

34+

35+-- | Start a server at the specified address

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

37+forkServer host port server = do

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

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

40+ -- the "best" address first, whatever that means

41+ addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)

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

43+ N.setSocketOption sock N.ReuseAddr 1

44+ N.bindSocket sock (N.addrAddress addr)

45+ N.listen sock 5

46+ forkIO $ server sock

47+

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

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

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

51+ -> [ByteString] -- ^ Payload

52+ -> IO ()

53+sendWithLength sock header payload = do

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

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

56+ NBS.sendMany sock msg

57+

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

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

60+recvWithLength sock = do

61+ msgLengthBs <- recvExact sock 4

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

63+

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

65+--

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

67+recvExact :: N.Socket -- ^ Socket to read from

68+ -> Int32 -- ^ Number of bytes to read

69+ -> MaybeT IO [ByteString]

70+recvExact _ len | len <= 0 = mzero

71+recvExact sock len = do

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

73+ if socketClosed then mzero else return input

74+ where

75+ -- Returns input read and whether the socket closed prematurely

76+ go :: [ByteString] -> Int32 -> IO (Bool, [ByteString])

77+ go acc 0 = return (False, reverse acc)

78+ go acc l = do

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

80+ if BS.null bs

81+ then return (True, reverse acc)

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

network-transport-2/src/Network/Transport/TCP.hs+37−75

@@ -1,12 +1,21 @@

1-- | TCP transport

2−module Network.Transport.TCP where

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

3+ createTransport

4+ , -- TCP specific functionality

5+ EndPointIx

6+ , decodeEndPointAddress

7+ ) where

4import Network.Transport

9import Network.Transport

10+import Network.Transport.Internal.TCP ( forkServer

11+ , recvExact

12+ , recvWithLength

13+ , sendWithLength

14+ )

5import Network.Transport.Internal ( encodeInt16

15import Network.Transport.Internal ( encodeInt16

6 , decodeInt16

16 , decodeInt16

7− , encodeInt32

8− , decodeInt32

9 , maybeToErrorT

17 , maybeToErrorT

18+ , maybeTToErrorT

10 )

19 )

11import qualified Network.Socket as N ( HostName

20import qualified Network.Socket as N ( HostName

12 , ServiceName

21 , ServiceName

@@ -14,10 +23,7 @@ import qualified Network.Socket as N ( HostName

14 , SocketType(Stream)

23 , SocketType(Stream)

15 , SocketOption(ReuseAddr)

24 , SocketOption(ReuseAddr)

16 , getAddrInfo

25 , getAddrInfo

17− , defaultHints

18 , socket

26 , socket

19− , bindSocket

20− , listen

21 , addrFamily

27 , addrFamily

22 , addrAddress

28 , addrAddress

23 , defaultProtocol

29 , defaultProtocol

@@ -26,22 +32,21 @@ import qualified Network.Socket as N ( HostName

26 , sClose

32 , sClose

27 , accept

33 , accept

28 )

34 )

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

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

30−import Control.Concurrent (forkIO, ThreadId)

36+import Control.Concurrent (forkIO)

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

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

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

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

33import Control.Category ((>>>))

39import Control.Category ((>>>))

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

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

35−import Control.Monad (forever, mzero)

41+import Control.Monad (forever)

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

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

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

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

38import Data.ByteString (ByteString)

44import Data.ByteString (ByteString)

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

45+import qualified Data.ByteString as BS (concat)

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

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

41import Data.IntMap (IntMap)

47import Data.IntMap (IntMap)

42import qualified Data.IntMap as IntMap (empty, size)

48import qualified Data.IntMap as IntMap (empty, size)

43import Data.Lens.Lazy (Lens, lens, intMapLens, (^.), (^=), (^+=))

49import Data.Lens.Lazy (Lens, lens, intMapLens, (^.), (^=), (^+=))

44−import Data.Int (Int32)

45import Text.Regex.Applicative (RE, few, anySym, sym, many, match)

50import Text.Regex.Applicative (RE, few, anySym, sym, many, match)

47data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState) }

52data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState) }

@@ -55,7 +60,7 @@ endPoints = lens _endPoints (\es st -> st { _endPoints = es })

55nextConnectionIx :: Lens EndPointState ConnectionIx

60nextConnectionIx :: Lens EndPointState ConnectionIx

56nextConnectionIx = lens _nextConnectionIx (\cix st -> st { _nextConnectionIx = cix })

61nextConnectionIx = lens _nextConnectionIx (\cix st -> st { _nextConnectionIx = cix })

58−endPointAt :: Int -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))

63+endPointAt :: EndPointIx -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))

59endPointAt ix = endPoints >>> intMapLens ix

64endPointAt ix = endPoints >>> intMapLens ix

61-- | Create a TCP transport

66-- | Create a TCP transport

@@ -74,19 +79,6 @@ createTransport host port = do

74 forkServer host port (transportServer state)

79 forkServer host port (transportServer state)

75 return Transport { newEndPoint = tcpNewEndPoint state host port }

80 return Transport { newEndPoint = tcpNewEndPoint state host port }

77−-- | Start a server at the specified address

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

79−forkServer host port server = do

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

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

82− -- the "best" address first, whatever that means

83− addr:_ <- N.getAddrInfo (Just N.defaultHints) (Just host) (Just port)

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

85− N.setSocketOption sock N.ReuseAddr 1

86− N.bindSocket sock (N.addrAddress addr)

87− N.listen sock 5

88− forkIO $ server sock

89−

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

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

91--

83--

92-- TODO: lots of error checking

84-- TODO: lots of error checking

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

129 writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

121 writeChan chan (ConnectionOpened newIx ReliableOrdered theirAddress)

130

122

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

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

132− readMessage :: Int -> MaybeT IO ()

124+ readMessage :: ConnectionIx -> MaybeT IO ()

133 readMessage connIx = do

125 readMessage connIx = do

134 payload <- recvWithLength sock

126 payload <- recvWithLength sock

135 liftIO $ writeChan chan (Received connIx payload)

127 liftIO $ writeChan chan (Received connIx payload)

@@ -145,24 +137,29 @@ tcpNewEndPoint transportState host port = do

145 return . Right $ EndPoint { receive = readChan chan

137 return . Right $ EndPoint { receive = readChan chan

146 , address = addr

138 , address = addr

147 , connect = tcpConnect addr

139 , connect = tcpConnect addr

148− , newMulticastGroup = undefined

140+ , newMulticastGroup = return . Left $ newMulticastGroupError

149− , resolveMulticastGroup = undefined

141+ , resolveMulticastGroup = \_ -> return . Left $ resolveMulticastGroupError

150 }

142 }

143+ where

144+ newMulticastGroupError = FailedWith NewMulticastGroupUnsupported "TCP does not support multicast"

145+ resolveMulticastGroupError = FailedWith ResolveMulticastGroupUnsupported "TCP does not support multicast"

146+

151

147

152-- | Connnect to an endpoint

148-- | Connnect to an endpoint

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

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

154tcpConnect myAddress theirAddress _ = runErrorT $ do

150tcpConnect myAddress theirAddress _ = runErrorT $ do

155− sock <- socketForEndPoint myAddress theirAddress

151+ sock <- socketForEndPoint myAddress theirAddress

152+ -- TODO: test this failure

156− connIx <- liftIO $ requestNewConnection sock

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

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

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

158− , close = N.sClose sock

155+ , close = N.sClose sock

159− }

156+ }

160

157

161-- | Request a new connection

158-- | Request a new connection

162−requestNewConnection :: N.Socket -> IO ByteString

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

163requestNewConnection sock = do

160requestNewConnection sock = do

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

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

165− NBS.recv sock 2

162+ BS.concat <$> recvExact sock 2

166

163

167-- | Find a socket to the specified endpoint

164-- | Find a socket to the specified endpoint

168--

165--

@@ -172,7 +169,7 @@ requestNewConnection sock = do

172socketForEndPoint :: EndPointAddress -- ^ Our address

169socketForEndPoint :: EndPointAddress -- ^ Our address

173 -> EndPointAddress -- ^ Their address

170 -> EndPointAddress -- ^ Their address

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

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

175−socketForEndPoint (EndPointAddress myAddress) (EndPointAddress theirAddress) = do

172+socketForEndPoint (EndPointAddress myAddress) theirAddress = do

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

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

177 liftIO $ do

174 liftIO $ do

178 -- Connect to the destination transport

175 -- Connect to the destination transport

@@ -188,46 +185,11 @@ socketForEndPoint (EndPointAddress myAddress) (EndPointAddress theirAddress) = d

188 where

185 where

189 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

186 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

190

187

191−-- | Send a bunch of bytestrings prepended with their length

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

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

194− -> [ByteString] -- ^ Payload

195− -> IO ()

196−sendWithLength sock header payload = do

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

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

199− NBS.sendMany sock msg

200−

201−-- | Read a length and then a payload of that length

202−-- TODO: error handling

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

204−recvWithLength sock = do

205− msgLengthBs <- recvExact sock 4

206− decodeInt32 (BS.concat msgLengthBs) >>= recvExact sock

207−

208-- | Decode end point address

188-- | Decode end point address

209-- TODO: This uses regular expression parsing, which is nice, but unnecessary

189-- TODO: This uses regular expression parsing, which is nice, but unnecessary

210−decodeEndPointAddress :: ByteString -> Maybe (N.HostName, N.ServiceName, Int)

190+decodeEndPointAddress :: EndPointAddress -> Maybe (N.HostName, N.ServiceName, EndPointIx)

211−decodeEndPointAddress bs = match endPointAddressRE (BSC.unpack bs)

191+decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack bs)

212 where

192 where

213− endPointAddressRE :: RE Char (N.HostName, N.ServiceName, Int)

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

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

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

215

195

216−-- | Read an exact number of bytes from a socket

217−--

218−-- Returns 'Nothing' if the socket closes prematurely.

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

220− -> Int32 -- ^ Number of bytes to read

221− -> MaybeT IO [ByteString]

222−recvExact sock len = do

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

224− if socketClosed then mzero else return input

225− where

226− -- Returns input read and whether the socket closed prematurely

227− go :: [ByteString] -> Int32 -> IO (Bool, [ByteString])

228− go acc 0 = return (False, reverse acc)

229− go acc l = do

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

231− if BS.null bs

232− then return (True, reverse acc)

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

network-transport-2/tests/TestTCP.hs+71−2

@@ -1,10 +1,79 @@

1module Main where

3+import Network.Transport

4+import Network.Transport.Internal (encodeInt16)

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

3−import TestTransport

6+import TestTransport (testTransport)

4−import Network.Transport.TCP

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

5import System.Exit (exitFailure, exitSuccess)

8import System.Exit (exitFailure, exitSuccess)

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

10+import Control.Concurrent (forkIO)

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

12+import Data.Maybe (fromJust)

13+import qualified Network.Socket as N ( getAddrInfo

14+ , socket

15+ , connect

16+ , addrFamily

17+ , addrAddress

18+ , SocketType(Stream)

19+ , SocketOption(ReuseAddr)

20+ , defaultProtocol

21+ , setSocketOption

22+ , sClose

23+ )

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

25+

26+testEarlyDisconnect :: IO ()

27+testEarlyDisconnect = do

28+ serverAddr <- newEmptyMVar

29+ serverDone <- newEmptyMVar

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

32+ -- Server

33+ forkIO $ do

34+ Right endpoint <- newEndPoint transport

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

36+

37+ -- Wait for an incoming connection

38+ ConnectionOpened _ _ _ <- receive endpoint

39+

40+ -- Wait for the client to disconnect

41+ ConnectionClosed _ <- receive endpoint

42+

43+ putMVar serverDone ()

44+

45+ -- Client

46+ --

47+ -- We manually connect to the server to request a new connection,

48+ -- but then close the socket before the server gets the chance to

49+ -- respond

50+ forkIO $ do

51+ Right endpoint <- newEndPoint transport

52+ let EndPointAddress myAddress = address endpoint

53+

54+ -- Connect to the server

55+ addr:_ <- N.getAddrInfo Nothing (Just "127.0.0.1") (Just "8080")

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

57+ N.setSocketOption sock N.ReuseAddr 1

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

59+

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

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

62+ endPointBS <- encodeInt16 (fromIntegral endPointIx)

63+ sendWithLength sock (Just endPointBS) [myAddress]

64+

65+ -- Request a new connection

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

67+

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

69+ -- but we close the connection instead

70+ N.sClose sock

71+

72+ -- Wait for the server to finish

73+ takeMVar serverDone

74+

7main :: IO ()

75main :: IO ()

8main = do

76main = do

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

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

10 if success then exitSuccess else exitFailure

78 if success then exitSuccess else exitFailure

79+ -- testEarlyDisconnect