41668d1 · distributed-process

git.iko.soy/distributed-process

Changes to the API (not yet implemented)

commit41668d17f7d063160d642052a32b1122924c3db2 authorEdsko de Vries <edsko@edsko.net> dateTue 8 May 2012, 18:06 (14 years ago) parent8850548

3 files changed+110−52

compare with master

unifiedsplit

network-transport-2/src/Network/Transport.hs+21−4 network-transport-2/src/Network/Transport/Internal/TCP.hs+35−15 network-transport-2/src/Network/Transport/TCP.hs+54−33

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

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

1111 , MulticastAddress(..)

1212 -- * Error codes

1313 , FailedWith(..)

14− , NewEndPointErrorCode

14+ , NewEndPointErrorCode(..)

1515 , ConnectErrorCode(..)

1616 , NewMulticastGroupErrorCode(..)

1717 , ResolveMulticastGroupErrorCode(..)

@@ -30,6 +30,8 @@ import Control.Monad.Error (Error(..))

3030data Transport = Transport {

3131 -- | Create a new end point (heavyweight operation)

3232 newEndPoint :: IO (Either (FailedWith NewEndPointErrorCode) EndPoint)

33+ -- | Shutdown the transport completely

34+ , closeTransport :: IO ()

3335 }

3436

3537-- | Network endpoint.

@@ -44,6 +46,8 @@ data EndPoint = EndPoint {

4446 , newMulticastGroup :: IO (Either (FailedWith NewMulticastGroupErrorCode) MulticastGroup)

4547 -- | Resolve an address to a multicast group.

4648 , resolveMulticastGroup :: MulticastAddress -> IO (Either (FailedWith ResolveMulticastGroupErrorCode) MulticastGroup)

49+ -- | Close the endpoint

50+ , closeEndPoint :: IO ()

4751 }

4852

4953-- | Lightweight connection to an endpoint.

@@ -56,10 +60,16 @@ data Connection = Connection {

5660

5761-- | Event on an endpoint.

5862data Event =

63+ -- | Received a message

5964 Received ConnectionId [ByteString]

65+ -- | Connection closed

6066 | ConnectionClosed ConnectionId

67+ -- | Connection opened

6168 | ConnectionOpened ConnectionId Reliability EndPointAddress

69+ -- | Received multicast

6270 | ReceivedMulticast MulticastAddress [ByteString]

71+ -- | Error that caused a bunch of connections to close (possibly none)

72+ | ErrorEvent ErrorEventErrorCode [ConnectionId]

6373 deriving Show

6474

6575-- | Connection IDs enable receivers to distinguish one connection from another.

@@ -109,7 +119,8 @@ instance Error (FailedWith error) where

109119 strMsg = FailedWith undefined

110120

111121-- | Errors during the creation of an endpoint (currently, there are none)

112−data NewEndPointErrorCode

122+data NewEndPointErrorCode =

123+ NewEndPointTransportFailure -- ^ Transport-wide failure

113124

114125-- | Connection failure

115126data ConnectErrorCode =

@@ -131,6 +142,12 @@ data ResolveMulticastGroupErrorCode =

131142

132143-- | Failure during sending a message

133144data SendErrorCode =

134− SendFailed

135− | SendConnectionClosed

145+ SendConnectionClosed -- ^ Connection was closed manually or because of an error

146+ | SendFailed -- ^ Send failed for some other reason

147+ deriving Show

148+

149+-- | Error codes used when reporting errors to endpoints (through receive)

150+data ErrorEventErrorCode =

151+ ErrorEventEndPointClosed -- ^ Endpoint was closed (manually or because of an error)

152+ | TransportFatalError -- ^ Transport-wide fatal error

136153 deriving Show

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

@@ -26,25 +26,45 @@ import qualified Network.Socket as N ( HostName

2626import qualified Network.Socket.ByteString as NBS (recv)

2727import Control.Concurrent (forkIO, ThreadId)

2828import Control.Monad (mzero, MonadPlus, liftM, forever)

29+import Control.Exception (SomeException, try, handle)

2930import Data.ByteString (ByteString)

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

3132import Data.Int (Int32)

3233

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

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

35−forkServer host port server = do

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

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

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

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

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

41− N.setSocketOption sock N.ReuseAddr 1

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

43− N.listen sock 5

44− forkIO . forever $ do

45− (clientSock, _) <- N.accept sock

46− server clientSock

47−

34+-- | Start a server at the specified address.

35+--

36+-- This sets up a server socket for the specified host and port. Exceptions

37+-- thrown during setup are not caught.

38+--

39+-- Once the socket is created we spawn a new thread which repeatedly accepts

40+-- incoming connections and executes the given request handler. If any

41+-- exception occurs the thread terminates and calls the terminationHandler.

42+-- This exception may occur because of a call to 'N.accept', because the thread

43+-- was explicitly killed, or because of a synchronous exception thrown by the

44+-- request handler. Typically, you should avoid the last case by catching any

45+-- relevant exceptions in the request handler.

46+--

47+-- The request handler should spawn threads to handle each individual request

48+-- or the server will block.

49+forkServer :: N.HostName -- ^ Host

50+ -> N.ServiceName -- ^ Port

51+ -> Int -- ^ Backlog (maximum number of queued connections)

52+ -> (N.Socket -> IO ()) -- ^ Request handler

53+ -> (SomeException -> IO ()) -- ^ Termination handler

54+ -> IO ThreadId

55+forkServer host port backlog requestHandler terminationHandler = do

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

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

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

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

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

61+ N.setSocketOption sock N.ReuseAddr 1

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

63+ N.listen sock backlog

64+ forkIO . handle terminationHandler . forever $ do

65+ (clientSock, _) <- N.accept sock

66+ requestHandler clientSock

67+

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

4969recvWithLength :: N.Socket -> IO [ByteString]

5070recvWithLength sock = recvInt32 sock >>= recvExact sock

network-transport-2/src/Network/Transport/TCP.hs+54−33

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

4646 , connect

4747 )

4848import Network.Socket.ByteString (sendMany)

49−import Control.Concurrent (forkIO, ThreadId)

49+import Control.Concurrent (forkIO, ThreadId, killThread)

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

5151import Control.Concurrent.MVar ( MVar

5252 , newMVar

@@ -58,13 +58,13 @@ import Control.Concurrent.MVar ( MVar

5858 , newEmptyMVar

5959 , withMVar

6060 )

61−import Data.IORef (IORef, newIORef, writeIORef, readIORef, modifyIORef)

6261import Control.Category ((>>>))

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

6463import Control.Monad (forM_, void, when)

6564import Control.Monad.Error (ErrorT(ErrorT), runErrorT, throwError)

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

67−import Control.Exception (IOException)

66+import Control.Exception (IOException, SomeException)

67+import Data.IORef (IORef, newIORef, writeIORef, readIORef, modifyIORef)

6868import Data.ByteString (ByteString)

6969import qualified Data.ByteString as BS (concat)

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

@@ -183,15 +183,18 @@ data LocalEndPoint = LocalEndPoint { localAddress :: EndPointAddress

183183 }

184184data RemoteEndPoint = RemoteEndPoint { remoteAddress :: EndPointAddress

185185 , remoteState :: MVar RemoteState

186+ , remoteThread :: ThreadId

186187 }

187188

188189-- We use underscores for fields that we might update (using lenses).

189190

190−data TransportState = TransportState { _localEndPoints :: IntMap LocalEndPoint }

191−data LocalState = LocalState { _nextConnectionId :: ConnectionId

192− , _pendingCtrlRequests :: IntMap (MVar [ByteString])

193− , _nextCtrlRequestId :: ControlRequestId

194− , _localConnections :: Map EndPointAddress RemoteEndPoint

191+data TransportState = TransportState { _localEndPoints :: IntMap LocalEndPoint

192+ , _acceptNewConnections :: Bool

193+ }

194+data LocalState = LocalState { _nextConnectionId :: ConnectionId

195+ , _pendingCtrlRequests :: IntMap (MVar [ByteString])

196+ , _nextCtrlRequestId :: ControlRequestId

197+ , _localConnections :: Map EndPointAddress RemoteEndPoint

195198 }

196199

197200-- A remote endpoint has incoming and outgoing connections, and when the total

@@ -220,15 +223,15 @@ data LocalState = LocalState { _nextConnectionId :: ConnectionId

220223-- the writes for control messages).

221224--

222225-- Of course, "serialize" does not mean that we want for the remote endpoint to

223−-- reply. "Send" takes the mvar, sends to the endpoint, and then puts the mvar

224−-- back, without waiting for the endpoint to receive the message. Similarly,

225−-- when requesting a new connection, we take the mvar, tentatively increment the

226−-- reference count, send the control request, and then put the mvar back. When

227−-- the remote host responds to the new connection request we might have to do

228−-- another state change (reduce the refcount) if the connection request was

229−-- refused but we don't want to increment the ref count only after the remote

230−-- host acknowledges the request because then a concurrent 'close' might

231−-- actually attempt to close the socket.

226+-- reply. "Send" takes the mvar, sends to the endpoint (asynchronously), and

227+-- then puts the mvar back, without waiting for the endpoint to receive the

228+-- message. Similarly, when requesting a new connection, we take the mvar,

229+-- tentatively increment the reference count, send the control request, and

230+-- then put the mvar back. When the remote host responds to the new connection

231+-- request we might have to do another state change (reduce the refcount) if

232+-- the connection request was refused but we don't want to increment the ref

233+-- count only after the remote host acknowledges the request because then a

234+-- concurrent 'close' might actually attempt to close the socket.

232235--

233236-- Since we don't do concurrent reads from the same socket we don't need to

234237-- take the lock when reading from the socket.

@@ -270,14 +273,32 @@ data ConnectionRequestResponse =

270273-- TODOs: deal with hints

271274createTransport :: N.HostName -> N.ServiceName -> IO (Either IOException Transport)

272275createTransport host port = do

273− state <- newMVar TransportState { _localEndPoints = IntMap.empty }

274− let transport = TCPTransport { transportState = state

275− , transportHost = host

276− , transportPort = port

277− }

278− tryIO $ do

279− forkServer host port $ void . handleConnectionRequest transport

280− return Transport { newEndPoint = apiNewEndPoint transport }

276+ state <- newMVar TransportState { _localEndPoints = IntMap.empty

277+ , _acceptNewConnections = True

278+ }

279+ let transport = TCPTransport { transportState = state

280+ , transportHost = host

281+ , transportPort = port

282+ }

283+ tryIO $ do

284+ -- TODO: "5" is listed as "system dependent maximum backlog"?

285+ transportThread <- forkServer host port 5 (handleConnectionRequest transport)

286+ (terminationHandler transport)

287+ return Transport { newEndPoint = apiNewEndPoint transport

288+ , closeTransport = apiCloseTransport transportThread

289+ }

290+ where

291+ terminationHandler :: TCPTransport -> SomeException -> IO ()

292+ terminationHandler transport _ =

293+ modifyMVar_ (transportState transport) $ return . (acceptNewConnections ^= False)

294+

295+-- | Shut down the transport

296+apiCloseTransport :: ThreadId -> IO ()

297+apiCloseTransport transportThread = do

298+ -- Kill the thread that is listening for new incoming connections

299+ killThread transportThread

300+ -- Not yet implemented

301+ undefined

281302

282303--------------------------------------------------------------------------------

283304-- API functions --

@@ -593,20 +614,18 @@ decreaseRemoteRefCount theirEndPoint = do

593614

594615-- | Handle a connection request (that is, a remote endpoint that is trying to

595616-- establish a TCP connection with us)

596−--

597−-- Returns the thread ID of the thread that is handling incoming messages.

598−handleConnectionRequest :: TCPTransport -> N.Socket -> IO (Maybe ThreadId)

617+handleConnectionRequest :: TCPTransport -> N.Socket -> IO ()

599618handleConnectionRequest transport sock = do

619+ let tState = transportState transport

600620 request <- tryIO $ do

601621 ourEndPointId <- recvInt32 sock

602622 theirAddress <- EndPointAddress . BS.concat <$> recvWithLength sock

603− ourEndPoint <- (^. localEndPointAt ourEndPointId) <$> readMVar (transportState transport)

623+ ourEndPoint <- (^. localEndPointAt ourEndPointId) <$> readMVar tState

604624 return (theirAddress, ourEndPoint)

605625 case request of

606626 Right (_, Nothing) -> do

607627 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]

608628 N.sClose sock

609− return Nothing

610629 Right (theirAddress, Just ourEndPoint) -> do

611630 (crossed, theirEndPoint) <- modifyMVar (localState ourEndPoint) $ \st ->

612631 case st ^. connectionTo theirAddress of

@@ -622,7 +641,6 @@ handleConnectionRequest transport sock = do

622641 then do

623642 sendMany sock [encodeInt32 ConnectionRequestCrossed]

624643 N.sClose sock

625− return Nothing

626644 else do

627645 let remoteConn = RemoteConnection { remoteSocket = sock

628646 , _remoteRefCount = 0

@@ -630,12 +648,12 @@ handleConnectionRequest transport sock = do

630648 }

631649 putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)

632650 sendMany sock [encodeInt32 ConnectionRequestAccepted]

651+ -- TODO: we should store the tid as part of the endpoint state

633652 tid <- forkIO $ handleIncomingMessages (ourEndPoint, theirEndPoint)

634− return (Just tid)

653+ return ()

635654 Left _ -> do

636655 -- Invalid request

637656 N.sClose sock

638− return Nothing

639657

640658-- | Handle requests from a remote endpoint.

641659--

@@ -794,6 +812,9 @@ firstNonReservedConnectionId = 1024

794812localEndPoints :: Lens TransportState (IntMap LocalEndPoint)

795813localEndPoints = lens _localEndPoints (\es st -> st { _localEndPoints = es })

796814

815+acceptNewConnections :: Lens TransportState Bool

816+acceptNewConnections = lens _acceptNewConnections (\acc st -> st { _acceptNewConnections = acc })

817+

797818pendingCtrlRequests :: Lens LocalState (IntMap (MVar [ByteString]))

798819pendingCtrlRequests = lens _pendingCtrlRequests (\rep st -> st { _pendingCtrlRequests = rep })

799820