git.iko.soy/distributed-process
on master help sign in
Start work on slightly alternative Transport API
583332dEdsko de Vries
.gitignore+1
compare with master
unifiedsplit
@@ -9,3 +9,4 @@ dist/
9*.o
9*.o
10*.prof
10*.prof
11*.ps
11*.ps
12+*.swp
doc/InternalsTransportTCP.txt+1−1
@@ -7,7 +7,7 @@ Overview
7When a TCP transport is created a new server is started which listens on a
7When a TCP transport is created a new server is started which listens on a
8given port number on the local host. In order to support multiple connections
8given port number on the local host. In order to support multiple connections
9the transport maintains a set of channels, one per connection, represented as a
9the transport maintains a set of channels, one per connection, represented as a
10−pair of a `Chan [ByteString]` and a list of pairs `(ThreadId, Socket)` of
10+pair of a `MVar [ByteString]` and a list of pairs `(ThreadId, Socket)` of
11threads that are listening on this channel. A source end then corresponds to a
11threads that are listening on this channel. A source end then corresponds to a
12hostname (the hostname used by clients to identity the local host), port
12hostname (the hostname used by clients to identity the local host), port
13number, and channel ID; a receive end simply corresponds to a channel ID.
13number, and channel ID; a receive end simply corresponds to a channel ID.
network-transport-2/Setup.hsnew file+2
@@ -0,0 +1,2 @@
1+import Distribution.Simple
2+main = defaultMain
network-transport-2/network-transport.cabalnew file+17
@@ -0,0 +1,17 @@
1+Name: network-transport
2+Version: 0.1.0
3+Description: Network Transport
4+Author: Duncan Coutts, Nicolas Wu, Edsko de Vries
5+Maintainer: dcoutts@well-typed.com
6+Homepage: http://github.com/haskell-distributed
7+Build-Type: Simple
8+Cabal-Version: >=1.9.2
9+
10+Library
11+ Build-Depends: base,
12+ bytestring,
13+ containers
14+ Exposed-modules: Network.Transport,
15+ Network.Transport.Chan
16+ extensions: EmptyDataDecls
17+ HS-Source-Dirs: src
network-transport-2/src/Network/Transport.hsnew file+63
@@ -0,0 +1,63 @@
1+-- | Network Transport
2+--
3+-- Connections are as light-weight as possible. That is, for a UDP
4+-- implementation they are practically free (only a single socket is
5+-- created per end point). For a TCP implementation the Transport guarantees
6+-- that only a single socket pair will be created for every pair of end
7+-- points.
8+module Network.Transport where
9+
10+import Data.ByteString (ByteString)
11+
12+-- | To create a network abstraction layer, use one of the
13+-- Network.Transport.* packages.
14+data Transport = Transport {
15+ newEndPoint :: Identity -> IO (Either Error EndPoint)
16+}
17+
18+-- | Every end point must have a unique identity.
19+type Identity = ByteString
20+
21+newtype Address = Address ByteString
22+
23+newtype MulticastAddress = MulticastAddress ByteString
24+
25+data ErrorCode
26+
27+data Error = Error ErrorCode String
28+
29+data EndPoint = EndPoint {
30+ receive :: IO Event
31+ , address :: Address
32+ , connect :: Address -> Reliability -> IO (Either Error Connection)
33+ , multicastNewGroup :: IO (Either Error MulticastGroup)
34+ , multicastConnect :: MulticastAddress -> IO MulticastGroup
35+ , multicastSubscribe :: MulticastAddress -> IO ()
36+ , multicastUnsubscribe :: MulticastAddress -> IO ()
37+}
38+
39+data Reliability =
40+ ReliableOrdered
41+ | ReliableUnordered
42+ | Unreliable
43+
44+type ConnectionId = Int
45+
46+data Connection = Connection {
47+ connId :: ConnectionId
48+ , send :: [ByteString] -> IO ()
49+ , close :: IO ()
50+ }
51+
52+data Event =
53+ Receive ConnectionId [ByteString]
54+ | ConnectionClosed ConnectionId
55+ | ConnectionOpened ConnectionId Reliability
56+ | MulticastReceive MulticastAddress [ByteString]
57+
58+data MulticastGroup = MulticastGroup {
59+ multicastSend :: [ByteString] -> IO ()
60+ , multicastAddress :: MulticastAddress
61+ , multicastMaxMsgSize :: Maybe Int
62+ , multicastGroupClose :: IO ()
63+ }
network-transport-2/src/Network/Transport/Chan.hsnew file+49
@@ -0,0 +1,49 @@
1+-- | Implementation of the Transport which creates a single Chan for every endpoint.
2+module Network.Transport.Chan (createTransport) where
3+
4+import Network.Transport
5+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
6+import Control.Concurrent.MVar (MVar, newMVar, takeMVar, putMVar)
7+import Data.Map (Map)
8+import qualified Data.Map as Map (empty, insert, (!), member)
9+
10+-- Global state: mapping from endpoint identities to the number of the next available connection
11+type TransportState = Map Identity Int
12+
13+-- | Create a new Transport
14+createTransport :: IO Transport
15+createTransport = do
16+ endpoints <- newMVar Map.empty
17+ return Transport { newEndPoint = chanNewEndPoint endpoints }
18+
19+-- Create a new end point
20+chanNewEndPoint :: MVar TransportState -> Identity -> IO (Either Error EndPoint)
21+chanNewEndPoint endpoints identity = do
22+ state <- takeMVar endpoints
23+ if identity `Map.member` state
24+ then do
25+ putMVar endpoints state
26+ return . Left $ Error undefined ("Identifier " ++ show identity ++ " already used")
27+ else do
28+ putMVar endpoints (Map.insert identity 0 state)
29+ chan <- newChan
30+ return . Right $ EndPoint { receive = readChan chan
31+ , address = Address identity
32+ , connect = chanConnect chan endpoints
33+ , multicastNewGroup = undefined
34+ , multicastConnect = undefined
35+ , multicastSubscribe = undefined
36+ , multicastUnsubscribe = undefined
37+ }
38+
39+-- Create a new connection
40+chanConnect :: Chan Event -> MVar TransportState -> Address -> Reliability -> IO (Either Error Connection)
41+chanConnect chan endpoints (Address identity) _ = do
42+ state <- takeMVar endpoints
43+ let conn = state Map.! identity
44+ putMVar endpoints (Map.insert identity (conn + 1) state)
45+ writeChan chan $ ConnectionOpened conn ReliableOrdered
46+ return . Right $ Connection { connId = conn
47+ , send = writeChan chan . Receive conn
48+ , close = writeChan chan $ ConnectionClosed conn
49+ }