55b8cd8 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Reuse sockets for connections

55b8cd8Edsko de Vries15 years ago

network-transport-2/src/Network/Transport/TCP.hs+42−18

compare with master

unifiedsplit

@@ -33,7 +33,7 @@ import qualified Network.Socket as N ( HostName

33 )

34import Control.Concurrent (forkIO)

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

36−import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, readMVar)

36+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)

37import Control.Category ((>>>))

38import Control.Applicative ((<*>), (*>), (<$>))

39import Control.Monad (forever, forM_)

@@ -47,22 +47,32 @@ import qualified Data.ByteString.Char8 as BSC (pack, unpack)

47import Data.Int (Int16)

48import Data.IntMap (IntMap)

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

50+import Data.Map (Map)

51+import qualified Data.Map as Map (empty)

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

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

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

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

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

55+data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState)

56+ , _sockets :: Map EndPointAddress N.Socket

57+ }

54data EndPointState = EndPointState { _nextConnectionId :: ConnectionId }

58data EndPointState = EndPointState { _nextConnectionId :: ConnectionId }

55type EndPointId = Int16

59type EndPointId = Int16

57endPoints :: Lens TransportState (IntMap (Chan Event, MVar EndPointState))

61endPoints :: Lens TransportState (IntMap (Chan Event, MVar EndPointState))

58endPoints = lens _endPoints (\es st -> st { _endPoints = es })

62endPoints = lens _endPoints (\es st -> st { _endPoints = es })

60−nextConnectionId :: Lens EndPointState ConnectionId

64+sockets :: Lens TransportState (Map EndPointAddress N.Socket)

61−nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })

65+sockets = lens _sockets (\ss st -> st { _sockets = ss })

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

67endPointAt :: EndPointId -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))

64endPointAt ix = endPoints >>> intMapLens (fromIntegral ix)

68endPointAt ix = endPoints >>> intMapLens (fromIntegral ix)

70+socketAt :: EndPointAddress -> Lens TransportState (Maybe N.Socket)

71+socketAt addr = sockets >>> mapLens addr

72+

73+nextConnectionId :: Lens EndPointState ConnectionId

74+nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })

75+

66-- | Create a TCP transport

76-- | Create a TCP transport

67--

77--

68-- Hostname and service name should be "canonical" for the transport layer to

78-- Hostname and service name should be "canonical" for the transport layer to

@@ -75,7 +85,9 @@ endPointAt ix = endPoints >>> intMapLens (fromIntegral ix)

75-- * Deal with all exceptions that may occur

85-- * Deal with all exceptions that may occur

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

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

77createTransport host port = do

87createTransport host port = do

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

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

89+ , _sockets = Map.empty

90+ }

79 forkServer host port (transportServer state)

91 forkServer host port (transportServer state)

80 return Transport { newEndPoint = tcpNewEndPoint state host port }

92 return Transport { newEndPoint = tcpNewEndPoint state host port }

@@ -143,7 +155,7 @@ tcpNewEndPoint transportState host port = do

143 let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointId

155 let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointId

144 return . Right $ EndPoint { receive = readChan chan

156 return . Right $ EndPoint { receive = readChan chan

145 , address = addr

157 , address = addr

146− , connect = tcpConnect addr

158+ , connect = tcpConnect transportState addr

147 , newMulticastGroup = return . Left $ newMulticastGroupError

159 , newMulticastGroup = return . Left $ newMulticastGroupError

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

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

149 }

161 }

@@ -156,9 +168,13 @@ firstNonReservedConnectionId :: ConnectionId

156firstNonReservedConnectionId = 1024

168firstNonReservedConnectionId = 1024

157

169

158-- | Connnect to an endpoint

170-- | Connnect to an endpoint

171+tcpConnect :: MVar TransportState -- ^ Transport state

172+ -> EndPointAddress -- ^ Our address

173+ -> EndPointAddress -- ^ Their address

174+ -> Reliability -- ^ Connection reliability (ignored)

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

175+ -> IO (Either (FailedWith ConnectErrorCode) Connection)

160−tcpConnect myAddress theirAddress _ = runErrorT $ do

176+tcpConnect state ourAddress theirAddress _ = runErrorT $ do

161− sock <- socketForEndPoint myAddress theirAddress

177+ sock <- socketForEndPoint state ourAddress theirAddress

162 connId <- failWithT undefined $ requestNewConnection sock

178 connId <- failWithT undefined $ requestNewConnection sock

163 return Connection { send = runErrorT .

179 return Connection { send = runErrorT .

164 failWithT (FailedWith SendFailed "Send failed") .

180 failWithT (FailedWith SendFailed "Send failed") .

@@ -176,15 +192,23 @@ requestNewConnection sock = sendInt32 sock 0 >> BS.concat <$> recvExact sock 4

176-- TODOs:

192-- TODOs:

177-- * Reuse connections

193-- * Reuse connections

178-- * Hints?

194-- * Hints?

179−socketForEndPoint :: EndPointAddress -- ^ Our address

195+socketForEndPoint :: MVar TransportState -- ^ Transport state (for reusing sockets)

196+ -> EndPointAddress -- ^ Our address

180− -> EndPointAddress -- ^ Their address

197+ -> EndPointAddress -- ^ Their address

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

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

182−socketForEndPoint (EndPointAddress myAddress) theirAddress = do

199+socketForEndPoint state (EndPointAddress ourAddress) theirAddress = do

200+ msock <- liftIO $ (^. socketAt theirAddress) <$> readMVar state

201+ case msock of

202+ Just sock ->

203+ return sock

204+ Nothing -> do

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

183− (host, port, endPointId) <- failWith invalidAddress (decodeEndPointAddress theirAddress)

206+ (host, port, endPointId) <- failWith invalidAddress (decodeEndPointAddress theirAddress)

184− sock <- failWithT undefined (connectTo host port)

207+ sock <- failWithT undefined (connectTo host port)

185− endPointBS <- encodeInt16 endPointId

208+ endPointBS <- encodeInt16 endPointId

186− failWithT undefined $ sendWithLength sock (Just endPointBS) [myAddress]

209+ failWithT undefined $ sendWithLength sock (Just endPointBS) [ourAddress]

210+ liftIO $ modifyMVar_ state $ return . (socketAt theirAddress ^= Just sock)

187− return sock

211+ return sock

188 where

212 where

189 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

213 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"

190

214