41668d1 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Changes to the API (not yet implemented)

41668d1Edsko de Vries14 years ago

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

compare with master

unifiedsplit

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

11 , MulticastAddress(..)

12 -- * Error codes

13 , FailedWith(..)

14− , NewEndPointErrorCode

14+ , NewEndPointErrorCode(..)

15 , ConnectErrorCode(..)

16 , NewMulticastGroupErrorCode(..)

17 , ResolveMulticastGroupErrorCode(..)

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

30data Transport = Transport {

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

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

33+ -- | Shutdown the transport completely

34+ , closeTransport :: IO ()

33 }

35 }

35-- | Network endpoint.

37-- | Network endpoint.

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

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

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

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

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

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

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

49+ -- | Close the endpoint

50+ , closeEndPoint :: IO ()

47 }

51 }

49-- | Lightweight connection to an endpoint.

53-- | Lightweight connection to an endpoint.

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

57-- | Event on an endpoint.

61-- | Event on an endpoint.

58data Event =

62data Event =

63+ -- | Received a message

59 Received ConnectionId [ByteString]

64 Received ConnectionId [ByteString]

65+ -- | Connection closed

60 | ConnectionClosed ConnectionId

66 | ConnectionClosed ConnectionId

67+ -- | Connection opened

61 | ConnectionOpened ConnectionId Reliability EndPointAddress

68 | ConnectionOpened ConnectionId Reliability EndPointAddress

69+ -- | Received multicast

62 | ReceivedMulticast MulticastAddress [ByteString]

70 | ReceivedMulticast MulticastAddress [ByteString]

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

72+ | ErrorEvent ErrorEventErrorCode [ConnectionId]

63 deriving Show

73 deriving Show

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

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

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

109 strMsg = FailedWith undefined

119 strMsg = FailedWith undefined

110

120

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

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

112−data NewEndPointErrorCode

122+data NewEndPointErrorCode =

123+ NewEndPointTransportFailure -- ^ Transport-wide failure

113

124

114-- | Connection failure

125-- | Connection failure

115data ConnectErrorCode =

126data ConnectErrorCode =

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

131

142

132-- | Failure during sending a message

143-- | Failure during sending a message

133data SendErrorCode =

144data SendErrorCode =

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

134− SendFailed

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

135− | SendConnectionClosed

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

136 deriving Show

153 deriving Show

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

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

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

27import Control.Concurrent (forkIO, ThreadId)

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

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

29import Data.ByteString (ByteString)

30import Data.ByteString (ByteString)

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

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

31import Data.Int (Int32)

32import Data.Int (Int32)

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

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.

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

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

35−forkServer host port server = do

55+forkServer host port backlog requestHandler terminationHandler = do

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

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

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

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

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

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

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

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

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

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

41− N.setSocketOption sock N.ReuseAddr 1

61+ N.setSocketOption sock N.ReuseAddr 1

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

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

43− N.listen sock 5

63+ N.listen sock backlog

44− forkIO . forever $ do

64+ forkIO . handle terminationHandler . forever $ do

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

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

46− server clientSock

66+ requestHandler clientSock

47−

67+

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

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

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

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

50recvWithLength sock = recvInt32 sock >>= recvExact sock

70recvWithLength 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

46 , connect

47 )

48import Network.Socket.ByteString (sendMany)

49−import Control.Concurrent (forkIO, ThreadId)

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

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

51import Control.Concurrent.MVar ( MVar

52 , newMVar

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

58 , newEmptyMVar

59 , withMVar

60 )

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

62import Control.Category ((>>>))

61import Control.Category ((>>>))

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

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

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

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

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

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

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

65import 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)

68import Data.ByteString (ByteString)

69import qualified Data.ByteString as BS (concat)

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

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

183 }

184data RemoteEndPoint = RemoteEndPoint { remoteAddress :: EndPointAddress

185 , remoteState :: MVar RemoteState

186+ , remoteThread :: ThreadId

186 }

187 }

187

188

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

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

189

190

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

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

192+ , _acceptNewConnections :: Bool

193+ }

191−data LocalState = LocalState { _nextConnectionId :: ConnectionId

194+data LocalState = LocalState { _nextConnectionId :: ConnectionId

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

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

193− , _nextCtrlRequestId :: ControlRequestId

196+ , _nextCtrlRequestId :: ControlRequestId

194− , _localConnections :: Map EndPointAddress RemoteEndPoint

197+ , _localConnections :: Map EndPointAddress RemoteEndPoint

195 }

198 }

196

199

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

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

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

220-- the writes for control messages).

223-- the writes for control messages).

221--

224--

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

225-- 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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

231−-- actually attempt to close the socket.

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

232--

235--

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

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

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

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

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

270-- TODOs: deal with hints

273-- TODOs: deal with hints

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

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

272createTransport host port = do

275createTransport host port = do

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

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

277+ , _acceptNewConnections = True

278+ }

274− let transport = TCPTransport { transportState = state

279+ let transport = TCPTransport { transportState = state

275− , transportHost = host

280+ , transportHost = host

276− , transportPort = port

281+ , transportPort = port

277− }

282+ }

278− tryIO $ do

283+ tryIO $ do

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

279− forkServer host port $ void . handleConnectionRequest transport

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

286+ (terminationHandler transport)

280− return Transport { newEndPoint = apiNewEndPoint 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

281

302

282--------------------------------------------------------------------------------

303--------------------------------------------------------------------------------

283-- API functions --

304-- API functions --

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

593

614

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

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

595-- establish a TCP connection with us)

616-- 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 ()

599handleConnectionRequest transport sock = do

618handleConnectionRequest transport sock = do

619+ let tState = transportState transport

600 request <- tryIO $ do

620 request <- tryIO $ do

601 ourEndPointId <- recvInt32 sock

621 ourEndPointId <- recvInt32 sock

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

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

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

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

604 return (theirAddress, ourEndPoint)

624 return (theirAddress, ourEndPoint)

605 case request of

625 case request of

606 Right (_, Nothing) -> do

626 Right (_, Nothing) -> do

607 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]

627 sendMany sock [encodeInt32 ConnectionRequestEndPointInvalid]

608 N.sClose sock

628 N.sClose sock

609− return Nothing

610 Right (theirAddress, Just ourEndPoint) -> do

629 Right (theirAddress, Just ourEndPoint) -> do

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

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

612 case st ^. connectionTo theirAddress of

631 case st ^. connectionTo theirAddress of

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

622 then do

641 then do

623 sendMany sock [encodeInt32 ConnectionRequestCrossed]

642 sendMany sock [encodeInt32 ConnectionRequestCrossed]

624 N.sClose sock

643 N.sClose sock

625− return Nothing

626 else do

644 else do

627 let remoteConn = RemoteConnection { remoteSocket = sock

645 let remoteConn = RemoteConnection { remoteSocket = sock

628 , _remoteRefCount = 0

646 , _remoteRefCount = 0

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

630 }

648 }

631 putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)

649 putMVar (remoteState theirEndPoint) (RemoteEndPointValid remoteConn)

632 sendMany sock [encodeInt32 ConnectionRequestAccepted]

650 sendMany sock [encodeInt32 ConnectionRequestAccepted]

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

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

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

634− return (Just tid)

653+ return ()

635 Left _ -> do

654 Left _ -> do

636 -- Invalid request

655 -- Invalid request

637 N.sClose sock

656 N.sClose sock

638− return Nothing

639

657

640-- | Handle requests from a remote endpoint.

658-- | Handle requests from a remote endpoint.

641--

659--

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

794localEndPoints :: Lens TransportState (IntMap LocalEndPoint)

812localEndPoints :: Lens TransportState (IntMap LocalEndPoint)

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

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

796

814

815+acceptNewConnections :: Lens TransportState Bool

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

817+

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

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

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

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

799

820