@@ -3,7 +3,8 @@
4-- * Various errors are still left as "undefined"
4-- * Various errors are still left as "undefined"
5-- * Many possible errors are not yet dealt with
5-- * Many possible errors are not yet dealt with
6−-- * Connections are not yet lightweight
6+-- * Multiple connections to the same endpoint (which reuse the same channel)
7+-- may be scrambed together
7module Network.Transport.TCP ( -- * Main API
8module Network.Transport.TCP ( -- * Main API
8 createTransport
9 createTransport
9 , -- TCP specific functionality
10 , -- TCP specific functionality
@@ -14,7 +15,6 @@ module Network.Transport.TCP ( -- * Main API
14import Network.Transport
15import Network.Transport
15import Network.Transport.Internal.TCP ( forkServer
16import Network.Transport.Internal.TCP ( forkServer
16 , connectTo
17 , connectTo
18 , recvWithLength
18 , recvWithLength
19 , sendWithLength
19 , sendWithLength
20 , recvInt16
20 , recvInt16
@@ -22,6 +22,7 @@ import Network.Transport.Internal.TCP ( forkServer
22 , sendInt32
22 , sendInt32
24import Network.Transport.Internal ( encodeInt16
24import Network.Transport.Internal ( encodeInt16
25 , failWith
26 , failWith
26 , failWithT
27 , failWithT
@@ -33,15 +34,15 @@ import qualified Network.Socket as N ( HostName
34import Control.Concurrent (forkIO)
35import Control.Concurrent (forkIO)
35import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
36import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
36−import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar)
37+import Control.Concurrent.MVar (MVar, newMVar, modifyMVar, modifyMVar_, readMVar, takeMVar, putMVar)
37import Control.Category ((>>>))
38import Control.Category ((>>>))
38import Control.Applicative ((<*>), (*>), (<$>))
39import Control.Applicative ((<*>), (*>), (<$>))
39−import Control.Monad (forever, forM_)
40+import Control.Monad (forever, forM_, void)
40import Control.Monad.Error (ErrorT, liftIO, runErrorT)
41import Control.Monad.Error (ErrorT, liftIO, runErrorT)
41import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)
42import Control.Monad.Trans.Maybe (MaybeT(MaybeT), runMaybeT)
42import Control.Monad.Trans.Writer (WriterT, execWriterT)
43import Control.Monad.Trans.Writer (WriterT, execWriterT)
43import Control.Monad.Writer.Class (tell)
44import Control.Monad.Writer.Class (tell)
44−import Data.ByteString (ByteString)
45+import Control.Monad.IO.Class (MonadIO)
45import qualified Data.ByteString as BS (concat)
46import qualified Data.ByteString as BS (concat)
46import qualified Data.ByteString.Char8 as BSC (pack, unpack)
47import qualified Data.ByteString.Char8 as BSC (pack, unpack)
47import Data.Int (Int16)
48import Data.Int (Int16)
@@ -52,26 +53,28 @@ import qualified Data.Map as Map (empty)
52import Data.Lens.Lazy (Lens, lens, intMapLens, mapLens, (^.), (^=), (^+=))
53import Data.Lens.Lazy (Lens, lens, intMapLens, mapLens, (^.), (^=), (^+=))
53import Text.Regex.Applicative (RE, few, anySym, sym, many, match)
54import Text.Regex.Applicative (RE, few, anySym, sym, many, match)
55−data TransportState = TransportState { _endPoints :: IntMap (Chan Event, MVar EndPointState)
56+data TCPTransport = TCPTransport { _transportHost :: N.HostName
57+ , _transportPort :: N.ServiceName
58+ , _transportState :: MVar TransportState
60+data TransportState = TransportState { _localEndPoints :: IntMap LocalEndPoint
56− , _sockets :: Map EndPointAddress N.Socket
61+ , _remoteEndPoints :: Map EndPointAddress RemoteEndPoint
63+data LocalEndPoint = LocalEndPoint { _localAddress :: EndPointAddress
64+ , _localChannel :: Chan Event
65+ , _localState :: MVar EndPointState
67+data RemoteEndPoint = RemoteEndPoint { _remoteAddress :: EndPointAddress
68+ , _remoteSocket :: N.Socket
69+ , _remoteConnectionIds :: Chan ConnectionId
70+ , _remoteLock :: MVar ()
72+data EndPointState = EndPointState { _nextConnectionId :: ConnectionId
58−data EndPointState = EndPointState { _nextConnectionId :: ConnectionId }
59type EndPointId = Int16
74type EndPointId = Int16
61−endPoints :: Lens TransportState (IntMap (Chan Event, MVar EndPointState))
76+#define REQUEST_CONNECTION_ID 0
62−endPoints = lens _endPoints (\es st -> st { _endPoints = es })
77+#define RESPOND_CONNECTION_ID 1
64−sockets :: Lens TransportState (Map EndPointAddress N.Socket)
65−sockets = lens _sockets (\ss st -> st { _sockets = ss })
67−endPointAt :: EndPointId -> Lens TransportState (Maybe (Chan Event, MVar EndPointState))
68−endPointAt ix = endPoints >>> intMapLens (fromIntegral ix)
70−socketAt :: EndPointAddress -> Lens TransportState (Maybe N.Socket)
71−socketAt addr = sockets >>> mapLens addr
73−nextConnectionId :: Lens EndPointState ConnectionId
74−nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
76-- | Create a TCP transport
79-- | Create a TCP transport
@@ -85,47 +88,69 @@ nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = c
85-- * Deal with all exceptions that may occur
88-- * Deal with all exceptions that may occur
86createTransport :: N.HostName -> N.ServiceName -> IO Transport
89createTransport :: N.HostName -> N.ServiceName -> IO Transport
87createTransport host port = do
90createTransport host port = do
88− state <- newMVar TransportState { _endPoints = IntMap.empty
91+ state <- newMVar TransportState { _localEndPoints = IntMap.empty
89− , _sockets = Map.empty
92+ , _remoteEndPoints = Map.empty
94+ let transport = TCPTransport { _transportState = state
95+ , _transportHost = host
96+ , _transportPort = port
91− forkServer host port (transportServer state)
98+ forkServer host port (transportServer transport)
92− return Transport { newEndPoint = tcpNewEndPoint state host port }
99+ return Transport { newEndPoint = tcpNewEndPoint transport }
94-- | The transport server handles incoming connections for all endpoints
101-- | The transport server handles incoming connections for all endpoints
96-- TODO: lots of error checking
103-- TODO: lots of error checking
97−transportServer :: MVar TransportState -> N.Socket -> IO ()
104+transportServer :: TCPTransport -> N.Socket -> IO ()
98−transportServer state sock = forever $ do
105+transportServer transport sock = forever $ do
99 (clientSock, _) <- N.accept sock
106 (clientSock, _) <- N.accept sock
100− forkIO $ handleClient state clientSock >> N.sClose clientSock
107+ handleConnectionRequest transport clientSock
102−-- | Handle a new incoming connection
109+handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()
103−--
110+handleConnectionRequest transport sock = do
104−-- Returns when the client closes the socket and on invalid input.
105−handleClient :: MVar TransportState -> N.Socket -> IO ()
106−handleClient state sock = do
107− mendpoint <- runMaybeT $ do
111+ request <- runMaybeT $ do
108 ourEndPointId <- recvInt16 sock
112 ourEndPointId <- recvInt16 sock
109− theirAddress <- BS.concat <$> recvWithLength sock
113+ theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock
110− ourEndPoint <- MaybeT $ (^. endPointAt ourEndPointId) <$> readMVar state
114+ ourEndPoint <- MaybeT $ (^. localEndPointAt ourEndPointId) <$> readMVar (transport ^. transportState)
111− return (ourEndPoint, EndPointAddress theirAddress)
115+ return (ourEndPoint, theirAddress)
112− case mendpoint of
116+ case request of
113− Just ((endPointCh, endPointSt), theirAddress) -> do
117+ Just (ourEndPoint, theirAddress) ->
114− unclosedConnections <- endpointServer sock theirAddress endPointCh endPointSt
118+ void $ forkRemoteEndPoint transport ourEndPoint theirAddress sock
122+forkRemoteEndPoint :: TCPTransport
126+ -> IO RemoteEndPoint
127+forkRemoteEndPoint transport ourEndPoint theirAddress theirSock = do
128+ theirConnectionIds <- newChan
129+ theirLock <- newMVar ()
130+ let theirEndPoint = RemoteEndPoint { _remoteAddress = theirAddress
131+ , _remoteSocket = theirSock
132+ , _remoteConnectionIds = theirConnectionIds
133+ , _remoteLock = theirLock
135+ modifyMVar_ (transport ^. transportState) $ return . (remoteEndPointAt theirAddress ^= Just theirEndPoint)
137+ unclosedConnections <- handleIncomingMessages ourEndPoint theirEndPoint
115− forM_ unclosedConnections $ \cix -> writeChan endPointCh (ConnectionClosed cix)
138+ forM_ unclosedConnections $ \cix -> writeChan (ourEndPoint ^. localChannel) (ConnectionClosed cix)
116− Nothing ->
139+ -- TODO: should we close the socket here? (forkRemoteEndPoint gets called from multiple places)
117− -- Invalid request
140+ return theirEndPoint
120−-- | Handle requests for one endpoint. Returns only if an error occurs,
142+-- | Handle requests from a remote endpoint.
144+-- Returns only if the remote party closes the socket or if an error occurs.
121−-- in which case it returns the connections that are still open (normally,
145+-- The return value is the list of connections that are still open (normally,
122-- this should be an empty list).
146-- this should be an empty list).
123−endpointServer :: N.Socket -> EndPointAddress -> Chan Event -> MVar EndPointState -> IO [ConnectionId]
147+handleIncomingMessages :: LocalEndPoint -> RemoteEndPoint -> IO [ConnectionId]
124−endpointServer sock theirAddress chan state = execWriterT . runMaybeT . forever $ do
148+handleIncomingMessages ourEndPoint theirEndPoint = execWriterT . runMaybeT . forever $ do
125 connId <- recvInt32 sock
149 connId <- recvInt32 sock
126 case connId of
150 case connId of
127− 0 -> createNewConnection
151+ REQUEST_CONNECTION_ID -> createNewConnection
152+ RESPOND_CONNECTION_ID -> readNewConnectionId
128− _ -> readMessage connId
153+ _ -> readMessage connId
130 -- A brief note on the types:
155 -- A brief note on the types:
131 -- These functions may fail (connection error): hence MaybeT
156 -- These functions may fail (connection error): hence MaybeT
@@ -134,28 +159,51 @@ endpointServer sock theirAddress chan state = execWriterT . runMaybeT . forever
134 -- Create a new connection
159 -- Create a new connection
135 createNewConnection :: MaybeT (WriterT [ConnectionId] IO) ()
160 createNewConnection :: MaybeT (WriterT [ConnectionId] IO) ()
136 createNewConnection = do
161 createNewConnection = do
137− newId <- liftIO $ modifyMVar state $ \st -> return (nextConnectionId ^+= 1 $ st, st ^. nextConnectionId)
162+ newId <- liftIO $ getNextConnectionId ourEndPoint
163+ withRemoteLock theirEndPoint $ do
164+ sendInt32 sock RESPOND_CONNECTION_ID
138− sendInt32 sock newId
165+ sendInt32 sock newId
139− liftIO $ writeChan chan (ConnectionOpened newId ReliableOrdered theirAddress)
166+ liftIO $ writeChan ourChannel (ConnectionOpened newId ReliableOrdered (theirEndPoint ^. remoteAddress))
140 -- We add the new connection ID to the list of open connections only once the
167 -- We add the new connection ID to the list of open connections only once the
141 -- endpoint has been notified of the new connection (sendInt32 may fail)
168 -- endpoint has been notified of the new connection (sendInt32 may fail)
142 tell [newId]
169 tell [newId]
144 -- Read a message and output it on the endPoint's channel
171 -- Read a message and output it on the endPoint's channel
145 readMessage :: ConnectionId -> MaybeT (WriterT [ConnectionId] IO) ()
172 readMessage :: ConnectionId -> MaybeT (WriterT [ConnectionId] IO) ()
146− readMessage connId = recvWithLength sock >>= liftIO . writeChan chan . Received connId
173+ readMessage connId = recvWithLength sock >>= liftIO . writeChan ourChannel . Received connId
175+ -- Read a new connection ID
176+ readNewConnectionId :: MaybeT (WriterT [ConnectionId] IO) ()
177+ readNewConnectionId = recvInt32 sock >>= liftIO . writeChan (theirEndPoint ^. remoteConnectionIds)
179+ -- Breakdown of the arguments
180+ ourChannel = ourEndPoint ^. localChannel
181+ sock = theirEndPoint ^. remoteSocket
183+-- | Get the next connection ID
184+getNextConnectionId :: LocalEndPoint -> IO ConnectionId
185+getNextConnectionId endpoint =
186+ modifyMVar (endpoint ^. localState) $ \st ->
187+ return (nextConnectionId ^+= 1 $ st, st ^. nextConnectionId)
148-- | Create a new endpoint
189-- | Create a new endpoint
149−tcpNewEndPoint :: MVar TransportState -> N.HostName -> N.ServiceName -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
190+tcpNewEndPoint :: TCPTransport -> IO (Either (FailedWith NewEndPointErrorCode) EndPoint)
150−tcpNewEndPoint transportState host port = do
191+tcpNewEndPoint transport = do
192+ endPoint <- modifyMVar (transport ^. transportState) $ \st -> do
151− chan <- newChan
193+ chan <- newChan
152− endPointState <- newMVar EndPointState { _nextConnectionId = firstNonReservedConnectionId }
194+ state <- newMVar EndPointState { _nextConnectionId = firstNonReservedConnectionId }
153− endPointId <- modifyMVar transportState $ \st -> do let ix = fromIntegral $ IntMap.size (st ^. endPoints)
195+ let ix = fromIntegral $ IntMap.size (st ^. localEndPoints)
154− return ((endPointAt ix ^= Just (chan, endPointState)) $ st, ix)
196+ let host = transport ^. transportHost
197+ let port = transport ^. transportPort
155− let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show endPointId
198+ let addr = EndPointAddress . BSC.pack $ host ++ ":" ++ port ++ ":" ++ show ix
199+ let localEndPoint = LocalEndPoint { _localAddress = addr
200+ , _localChannel = chan
201+ , _localState = state
203+ return ((localEndPointAt ix ^= Just localEndPoint) $ st, localEndPoint)
156− return . Right $ EndPoint { receive = readChan chan
204+ return . Right $ EndPoint { receive = readChan (endPoint ^. localChannel)
157− , address = addr
205+ , address = endPoint ^. localAddress
158− , connect = tcpConnect transportState addr
206+ , connect = tcpConnect transport endPoint
159 , newMulticastGroup = return . Left $ newMulticastGroupError
207 , newMulticastGroup = return . Left $ newMulticastGroupError
160 , resolveMulticastGroup = \_ -> return . Left $ resolveMulticastGroupError
208 , resolveMulticastGroup = \_ -> return . Left $ resolveMulticastGroupError
@@ -168,49 +216,62 @@ firstNonReservedConnectionId :: ConnectionId
168firstNonReservedConnectionId = 1024
216firstNonReservedConnectionId = 1024
170-- | Connnect to an endpoint
218-- | Connnect to an endpoint
171−tcpConnect :: MVar TransportState -- ^ Transport state
219+tcpConnect :: TCPTransport -- ^ Transport
172− -> EndPointAddress -- ^ Our address
220+ -> LocalEndPoint -- ^ Local end point
173− -> EndPointAddress -- ^ Their address
221+ -> EndPointAddress -- ^ Remote address
174− -> Reliability -- ^ Connection reliability (ignored)
222+ -> Reliability -- ^ Reliability (ignored)
175 -> IO (Either (FailedWith ConnectErrorCode) Connection)
223 -> IO (Either (FailedWith ConnectErrorCode) Connection)
176−tcpConnect state ourAddress theirAddress _ = runErrorT $ do
224+tcpConnect transport ourEndPoint theirAddress _ = runErrorT $ do
177− sock <- socketForEndPoint state ourAddress theirAddress
225+ theirEndPoint <- remoteEndPoint transport ourEndPoint theirAddress
178− connId <- failWithT undefined $ requestNewConnection sock
226+ connId <- failWithT undefined $ requestNewConnection theirEndPoint
227+ connBs <- encodeInt32 connId
228+ let sock = theirEndPoint ^. remoteSocket
179 return Connection { send = runErrorT .
229 return Connection { send = runErrorT .
180 failWithT (FailedWith SendFailed "Send failed") .
230 failWithT (FailedWith SendFailed "Send failed") .
231+ withRemoteLock theirEndPoint .
181− sendWithLength sock (Just connId)
232+ sendWithLength sock (Just connBs)
182 , close = N.sClose sock
233 , close = N.sClose sock
185-- | Request a new connection
236-- | Request a new connection
186−-- Returns the new connection ID in serialized form
187−requestNewConnection :: N.Socket -> MaybeT IO ByteString
237+requestNewConnection :: RemoteEndPoint -> MaybeT IO ConnectionId
188−requestNewConnection sock = sendInt32 sock 0 >> BS.concat <$> recvExact sock 4
238+requestNewConnection endPoint = do
239+ withRemoteLock endPoint $ sendInt32 (endPoint ^. remoteSocket) 0
240+ liftIO $ readChan (endPoint ^. remoteConnectionIds)
190−-- | Find a socket to the specified endpoint
242+-- | Find a remote endpoint
192−-- TODOs:
244+-- TODO: hints?
193−-- * Reuse connections
245+remoteEndPoint :: TCPTransport
194−-- * Hints?
246+ -> LocalEndPoint
195−socketForEndPoint :: MVar TransportState -- ^ Transport state (for reusing sockets)
196− -> EndPointAddress -- ^ Our address
197− -> EndPointAddress -- ^ Their address
247+ -> EndPointAddress
198− -> ErrorT (FailedWith ConnectErrorCode) IO N.Socket
248+ -> ErrorT (FailedWith ConnectErrorCode) IO RemoteEndPoint
199−socketForEndPoint state (EndPointAddress ourAddress) theirAddress = do
249+remoteEndPoint transport ourEndPoint theirAddress = do
200− msock <- liftIO $ (^. socketAt theirAddress) <$> readMVar state
250+ mendpoint <- liftIO $ (^. remoteEndPointAt theirAddress) <$> readMVar (transport ^. transportState)
201− case msock of
251+ case mendpoint of
202− Just sock ->
252+ Just endpoint ->
203− return sock
253+ return endpoint
204 Nothing -> do
254 Nothing -> do
205− liftIO $ putStrLn $ "Creating socket from " ++ show ourAddress ++ " to " ++ show theirAddress
255+ -- liftIO $ putStrLn $ "Creating socket from " ++ show ourAddress ++ " to " ++ show theirAddress
206 (host, port, endPointId) <- failWith invalidAddress (decodeEndPointAddress theirAddress)
256 (host, port, endPointId) <- failWith invalidAddress (decodeEndPointAddress theirAddress)
207 sock <- failWithT undefined (connectTo host port)
257 sock <- failWithT undefined (connectTo host port)
208 endPointBS <- encodeInt16 endPointId
258 endPointBS <- encodeInt16 endPointId
209 failWithT undefined $ sendWithLength sock (Just endPointBS) [ourAddress]
259 failWithT undefined $ sendWithLength sock (Just endPointBS) [ourAddress]
210− liftIO $ modifyMVar_ state $ return . (socketAt theirAddress ^= Just sock)
260+ liftIO $ forkRemoteEndPoint transport ourEndPoint theirAddress sock
213 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"
262 invalidAddress = FailedWith ConnectInvalidAddress "Invalid address"
263+ EndPointAddress ourAddress = ourEndPoint ^. localAddress
265+-- | We must avoid scrambling concurrent sends to the same endpoint
267+-- TODO: deal with exceptions
268+withRemoteLock :: MonadIO m => RemoteEndPoint -> m a -> m a
269+withRemoteLock theirEndPoint p = do
270+ let lock = theirEndPoint ^. remoteLock
271+ liftIO $ takeMVar lock
273+ liftIO $ putMVar lock ()
215-- | Decode end point address
276-- | Decode end point address
216-- TODO: This uses regular expression parsing, which is nice, but unnecessary
277-- TODO: This uses regular expression parsing, which is nice, but unnecessary
@@ -219,3 +280,52 @@ decodeEndPointAddress (EndPointAddress bs) = match endPointAddressRE (BSC.unpack
220 endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointId)
281 endPointAddressRE :: RE Char (N.HostName, N.ServiceName, EndPointId)
221 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))
282 endPointAddressRE = (,,) <$> few anySym <*> (sym ':' *> few anySym) <*> (sym ':' *> (read <$> many anySym))
284+--------------------------------------------------------------------------------
286+--------------------------------------------------------------------------------
288+transportState :: Lens TCPTransport (MVar TransportState)
289+transportState = lens _transportState (\st trans -> trans { _transportState = st })
291+transportHost :: Lens TCPTransport N.HostName
292+transportHost = lens _transportHost (\host trans -> trans { _transportHost = host })
294+transportPort :: Lens TCPTransport N.ServiceName
295+transportPort = lens _transportPort (\port trans -> trans { _transportPort = port })
297+localEndPoints :: Lens TransportState (IntMap LocalEndPoint)
298+localEndPoints = lens _localEndPoints (\es st -> st { _localEndPoints = es })
300+remoteEndPoints :: Lens TransportState (Map EndPointAddress RemoteEndPoint)
301+remoteEndPoints = lens _remoteEndPoints (\es st -> st { _remoteEndPoints = es })
303+localAddress :: Lens LocalEndPoint EndPointAddress
304+localAddress = lens _localAddress (\addr ep -> ep { _localAddress = addr })
306+localChannel :: Lens LocalEndPoint (Chan Event)
307+localChannel = lens _localChannel (\ch ep -> ep { _localChannel = ch })
309+localState :: Lens LocalEndPoint (MVar EndPointState)
310+localState = lens _localState (\st ep -> ep { _localState = st })
312+remoteAddress :: Lens RemoteEndPoint EndPointAddress
313+remoteAddress = lens _remoteAddress (\addr ep -> ep { _remoteAddress = addr })
315+remoteSocket :: Lens RemoteEndPoint N.Socket
316+remoteSocket = lens _remoteSocket (\sock ep -> ep { _remoteSocket = sock })
318+remoteConnectionIds :: Lens RemoteEndPoint (Chan ConnectionId)
319+remoteConnectionIds = lens _remoteConnectionIds (\ch ep -> ep { _remoteConnectionIds = ch })
321+remoteLock :: Lens RemoteEndPoint (MVar ())
322+remoteLock = lens _remoteLock (\lock ep -> ep { _remoteLock = lock })
324+nextConnectionId :: Lens EndPointState ConnectionId
325+nextConnectionId = lens _nextConnectionId (\cix st -> st { _nextConnectionId = cix })
327+localEndPointAt :: EndPointId -> Lens TransportState (Maybe LocalEndPoint)
328+localEndPointAt ix = localEndPoints >>> intMapLens (fromIntegral ix)
330+remoteEndPointAt :: EndPointAddress -> Lens TransportState (Maybe RemoteEndPoint)
331+remoteEndPointAt addr = remoteEndPoints >>> mapLens addr