git.iko.soy/distributed-process
on master help sign in
Start work on combined transport
b6f32f0Edsko de Vries
network-transport-composed/LICENSEnew file+30
compare with master
unifiedsplit
@@ -0,0 +1,30 @@
1+Copyright (c) 2012, Edsko de Vries
2+
3+All rights reserved.
4+
5+Redistribution and use in source and binary forms, with or without
6+modification, are permitted provided that the following conditions are met:
7+
8+ * Redistributions of source code must retain the above copyright
9+ notice, this list of conditions and the following disclaimer.
10+
11+ * Redistributions in binary form must reproduce the above
12+ copyright notice, this list of conditions and the following
13+ disclaimer in the documentation and/or other materials provided
14+ with the distribution.
15+
16+ * Neither the name of Edsko de Vries nor the names of other
17+ contributors may be used to endorse or promote products derived
18+ from this software without specific prior written permission.
19+
20+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
21+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
22+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
23+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
24+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
25+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
26+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
27+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
28+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
29+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
30+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
network-transport-composed/Setup.hsnew file+2
@@ -0,0 +1,2 @@
1+import Distribution.Simple
2+main = defaultMain
network-transport-composed/network-transport-composed.cabalnew file+22
@@ -0,0 +1,22 @@
1+name: network-transport-composed
2+version: 0.1.0.0
3+synopsis: Compose network transports
4+-- description:
5+homepage: http://github.com/haskell-distributed/distributed-process
6+license: BSD3
7+license-file: LICENSE
8+author: Edsko de Vries
9+maintainer: edsko@well-typed.com
10+copyright: Well-Typed LLP
11+category: Network
12+build-type: Simple
13+cabal-version: >=1.8
14+
15+library
16+ exposed-modules: Network.Transport.Composed
17+ -- other-modules:
18+ build-depends: base ==4.5.*,
19+ network-transport >= 0.3 && < 0.4
20+ hs-source-dirs: src
21+ ghc-options: -Wall
22+ extensions: BangPatterns
network-transport-composed/src/Network/Transport/Composed.hsnew file+177
@@ -0,0 +1,177 @@
1+-- | Combine two transports into a single new transport
2+--
3+-- Given two transports A and B, endpoints in the combined transport are
4+-- pairs of endpoints for the underlying transports. Consequently, the address
5+-- of an endpoint in the combined transport is a pair of addresses of the
6+-- two underlying endpoints. Events from both underlying endpoints are posted
7+-- to the combined endpoint.
8+--
9+-- When the address of combined endpoint C1 is sent to combined endpoint C2,
10+-- and C2 attempts to make a connection to C1, it needs to choice an underlying
11+-- endpoint to make the connection. The choice is made by a parameter
12+-- 'resolveAddress' of the composed transport.
13+--
14+-- When C2 connects to C1 then it is assumed that whatever transport C2 uses
15+-- to connect to C1, this is the same transport that C1 needs to use to connect
16+-- back to C2.
17+module Network.Transport.Composed
18+ ( ComposedTransport(..)
19+ , createTransport
20+ ) where
21+
22+import qualified Data.ByteString as BSS
23+ ( concat
24+ , length
25+ , splitAt
26+ , singleton
27+ , uncons
28+ )
29+import Control.Monad (void)
30+import Control.Concurrent.Chan (Chan, newChan, writeChan, readChan)
31+import Control.Concurrent (forkIO)
32+import Network.Transport
33+ ( Transport(..)
34+ , EndPointAddress(EndPointAddress)
35+ , TransportError(TransportError)
36+ , NewEndPointErrorCode
37+ , EndPoint(..)
38+ , NewMulticastGroupErrorCode(NewMulticastGroupUnsupported)
39+ , ResolveMulticastGroupErrorCode(ResolveMulticastGroupUnsupported)
40+ , Event(EndPointClosed, ConnectionOpened, ErrorEvent)
41+ , EventErrorCode(EventConnectionLost)
42+ , Reliability
43+ , ConnectHints
44+ , ConnectErrorCode
45+ , Connection
46+ )
47+import Network.Transport.Internal (encodeInt32, decodeInt32)
48+
49+data ComposedTransport = ComposedTransport {
50+ -- | The left transport
51+ transportA :: Transport
52+ -- | The right transport
53+ , transportB :: Transport
54+ -- | Pick a suitable transport for an address
55+ , resolveAddress :: (EndPoint, EndPoint)
56+ -> (EndPointAddress, EndPointAddress)
57+ -> IO (EndPoint, EndPointAddress)
58+ }
59+
60+data ComposedAddress =
61+ LeftAddress EndPointAddress
62+ | RightAddress EndPointAddress
63+ | EitherAddress EndPointAddress EndPointAddress
64+
65+serializeComposedAddress :: ComposedAddress -> EndPointAddress
66+serializeComposedAddress (LeftAddress (EndPointAddress addr)) =
67+ EndPointAddress $ BSS.concat [BSS.singleton 0, addr]
68+serializeComposedAddress (RightAddress (EndPointAddress addr)) =
69+ EndPointAddress $ BSS.concat [BSS.singleton 1, addr]
70+serializeComposedAddress (EitherAddress (EndPointAddress addr1) (EndPointAddress addr2)) =
71+ EndPointAddress $ BSS.concat [BSS.singleton 2, encodeInt32 (BSS.length addr1), addr1, addr2]
72+
73+deserializeComposedAddress :: EndPointAddress -> ComposedAddress
74+deserializeComposedAddress (EndPointAddress addr) =
75+ let (header, remainder) = case BSS.uncons addr of
76+ Just (x, xs) -> (x, xs)
77+ Nothing -> error "deserializeComposedAddress"
78+ in case header of
79+ 0 -> LeftAddress (EndPointAddress remainder)
80+ 1 -> RightAddress (EndPointAddress remainder)
81+ 2 -> let (len, remainder') = BSS.splitAt 4 remainder
82+ (a, b) = BSS.splitAt (decodeInt32 len) remainder'
83+ in EitherAddress (EndPointAddress a) (EndPointAddress b)
84+ _ -> error "deserializeComposedAddress"
85+
86+createTransport :: ComposedTransport -> IO Transport
87+createTransport composed =
88+ return Transport {
89+ newEndPoint = apiNewEndPoint composed
90+ , closeTransport = apiCloseTransport composed
91+ }
92+
93+apiNewEndPoint :: ComposedTransport -> IO (Either (TransportError NewEndPointErrorCode) EndPoint)
94+apiNewEndPoint composed = do
95+ mEndPoints <- createEndPoints
96+ case mEndPoints of
97+ Left err ->
98+ return $ Left err
99+ Right (endPointA, endPointB) -> do
100+ let addr = serializeComposedAddress (EitherAddress (address endPointA) (address endPointB))
101+ chan <- newChan
102+ void . forkIO $ forwardEvents endPointA chan LeftAddress
103+ void . forkIO $ forwardEvents endPointB chan RightAddress
104+ return . Right $ EndPoint {
105+ receive = readChan chan
106+ , address = addr
107+ , connect = apiConnect composed endPointA endPointB
108+ , closeEndPoint = apiCloseEndPoint endPointA endPointB
109+ , newMulticastGroup = return . Left $ newMulticastGroupError
110+ , resolveMulticastGroup = return . Left . const resolveMulticastGroupError
111+ }
112+ where
113+ newMulticastGroupError =
114+ TransportError NewMulticastGroupUnsupported "Multicast not supported"
115+ resolveMulticastGroupError =
116+ TransportError ResolveMulticastGroupUnsupported "Multicast not supported"
117+
118+ createEndPoints = do
119+ mEndPointA <- newEndPoint (transportA composed)
120+ case mEndPointA of
121+ Left err -> return $ Left err
122+ Right endPointA -> do
123+ mEndPointB <- newEndPoint (transportB composed)
124+ case mEndPointB of
125+ Left err -> do
126+ closeEndPoint endPointA
127+ return $ Left err
128+ Right endPointB ->
129+ return $ Right (endPointA, endPointB)
130+
131+ forwardEvents :: EndPoint -> Chan Event -> (EndPointAddress -> ComposedAddress) -> IO ()
132+ forwardEvents endPoint chan inj = go
133+ where
134+ go = do
135+ event <- receive endPoint
136+ case event of
137+ ConnectionOpened cid rel addr -> do
138+ let addr' = serializeComposedAddress (inj addr)
139+ writeChan chan (ConnectionOpened cid rel addr')
140+ go
141+ ErrorEvent (TransportError (EventConnectionLost addr) msg) -> do
142+ let addr' = serializeComposedAddress (inj addr)
143+ writeChan chan $ ErrorEvent (TransportError (EventConnectionLost addr') msg)
144+ go
145+ EndPointClosed -> do
146+ writeChan chan event
147+ return ()
148+ _ -> do
149+ writeChan chan event
150+ go
151+
152+apiCloseTransport :: ComposedTransport -> IO ()
153+apiCloseTransport composed = do
154+ closeTransport (transportA composed)
155+ closeTransport (transportB composed)
156+
157+apiConnect :: ComposedTransport
158+ -> EndPoint
159+ -> EndPoint
160+ -> EndPointAddress
161+ -> Reliability
162+ -> ConnectHints
163+ -> IO (Either (TransportError ConnectErrorCode) Connection)
164+apiConnect composed endPointA endPointB theirComposedAddr rel hints =
165+ case deserializeComposedAddress theirComposedAddr of
166+ LeftAddress addr ->
167+ connect endPointA addr rel hints
168+ RightAddress addr ->
169+ connect endPointB addr rel hints
170+ EitherAddress addrA addrB -> do
171+ (endPoint, addr) <- resolveAddress composed (endPointA, endPointB) (addrA, addrB)
172+ connect endPoint addr rel hints
173+
174+apiCloseEndPoint :: EndPoint -> EndPoint -> IO ()
175+apiCloseEndPoint endPointA endPointB = do
176+ closeEndPoint endPointA
177+ closeEndPoint endPointB