git.iko.soy/distributed-process
on master help sign in
Initial commit
112921dNicolas Wu
README.mdnew file+16
compare with master
unifiedsplit
@@ -0,0 +1,16 @@
1+Haskell Distributed Project
2+===========================
3+
4+This repository holds an implementation of Cloud Haskell based on the work
5+of Jeff Epstein, Andrew Black, and Simon Peyton-Jones [1].
6+
7+At present, this repository hosts two packages:
8+
9+* distributed-process
10+* network-transport
11+
12+For more detailed information about the interfaces provided by these packages,
13+please refer to the [distributed-process repository wiki][2].
14+
15+[1]: http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
16+[2]: https://github.com/haskell-distributed/distributed-process/wiki
distributed-process/LICENSEnew file+31
@@ -0,0 +1,31 @@
1+Copyright Well-Typed LLP, 2011
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 the owner 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.
31+
distributed-process/Setup.hsnew file+2
@@ -0,0 +1,2 @@
1+import Distribution.Simple
2+main = defaultMain
distributed-process/distributed-process.cabalnew file+21
@@ -0,0 +1,21 @@
1+Name: distributed-process
2+Version: 0.0.1
3+Description: Distributed processes for message based concurrency.
4+License: BSD3
5+License-file: LICENSE
6+Author: Duncan Coutts, Nicolas Wu
7+Maintainer: duncan@well-typed.com
8+Homepage: http://github.com/haskell-distributed/
9+Build-Type: Simple
10+Cabal-Version: >=1.2
11+
12+Library
13+ Build-Depends: base >= 3 && < 5,
14+ bytestring >= 0.9,
15+ containers >= 0.4,
16+ mtl >= 2.0,
17+ network-transport >= 0.0
18+ Exposed-modules: Control.Distributed.Process
19+ Extensions: ExplicitForAll ScopedTypeVariables
20+ ghc-options: -Wall
21+ HS-Source-Dirs: src
distributed-process/src/Control/Distributed/Process.hsnew file+220
@@ -0,0 +1,220 @@
1+{-# LANGUAGE ExplicitForAll, ScopedTypeVariables #-}
2+module Control.Distributed.Process (
3+ -- * Processes
4+ Process,
5+ liftIO,
6+ ProcessId,
7+-- NodeId,
8+
9+ -- * Basic messaging
10+ send,
11+ expect,
12+
13+ -- * Process management
14+ spawnLocal,
15+ getSelfPid,
16+-- getSelfNode,
17+
18+ -- * Initialisation
19+ Transport,
20+ LocalNode,
21+ newLocalNode,
22+ runProcess,
23+
24+ ) where
25+
26+import qualified Network.Transport as Trans
27+import Network.Transport (Transport)
28+
29+import qualified Data.ByteString.Char8 as BS
30+import Data.ByteString.Char8 (ByteString)
31+import qualified Data.IntMap as IntMap
32+import Data.IntMap (IntMap)
33+import Data.IORef
34+import Control.Applicative
35+import Control.Monad
36+import Control.Monad.Trans
37+import Control.Concurrent.MVar
38+import Control.Concurrent.Chan
39+import Control.Concurrent
40+import Control.Exception
41+import Data.Typeable
42+
43+
44+------------------------
45+-- Cloud Haskell layer
46+--
47+
48+data NodeId = NodeId -- !Trans.SendEnd !Trans.SendEnd
49+
50+data ProcessId = ProcessId !Trans.SendEnd !NodeId !LocalProcessId
51+type LocalProcessId = Int
52+
53+newtype SendPort a = SendPort Trans.SendEnd
54+
55+newtype Process a = Process { unProcess :: ProcessState -> IO a }
56+
57+instance Functor Process where
58+ fmap f m = Process (\ps -> unProcess m ps >>= \x -> return (f x))
59+
60+instance Applicative Process where
61+ pure = return
62+ (<*>) = ap
63+
64+instance Monad Process where
65+ m >>= k = Process (\ps -> unProcess m ps >>= \x -> unProcess (k x) ps)
66+ return x = Process (\_ -> return x)
67+
68+instance MonadIO Process where
69+ liftIO io = Process (\_ -> io)
70+
71+getProcessState :: Process ProcessState
72+getProcessState = Process (\ps -> return ps)
73+
74+getLocalNode :: Process LocalNode
75+getLocalNode = prNode <$> getProcessState
76+
77+getSelfPid :: Process ProcessId
78+getSelfPid = prPid <$> getProcessState
79+
80+-- State a process carries around and has local access to
81+--
82+data ProcessState = ProcessState {
83+ prPid :: !ProcessId,
84+ prChan :: !Trans.ReceiveEnd,
85+ prQueue :: !(CQueue Message),
86+ prNode :: !LocalNode
87+ }
88+
89+data Message = Message String --string rep of TypeRep, sigh.
90+ String
91+
92+-- Context for a local node, all processes running on a node have direct access to this
93+--
94+data LocalNode = LocalNode {
95+ ndProcessTable :: !(MVar ProcessTable),
96+ ndTransport :: !Transport
97+ }
98+
99+data ProcessTable = ProcessTable !LocalProcessId !(IntMap ProcessTableEntry)
100+data ProcessTableEntry = ProcessTableEntry {
101+ pteThread :: !ThreadId
102+ }
103+
104+newLocalNode :: Transport -> IO LocalNode
105+newLocalNode trans = do
106+ processTableVar <- newMVar (ProcessTable 0 IntMap.empty)
107+ return LocalNode {
108+ ndProcessTable = processTableVar,
109+ ndTransport = trans
110+ }
111+
112+runProcess :: LocalNode -> Process () -> IO ()
113+runProcess node proc = do
114+ waitVar <- newEmptyMVar
115+ _ <- forkProcess node (proc >> liftIO (putMVar waitVar ()))
116+ --TODO: should use linking for waiting for the end
117+ takeMVar waitVar
118+
119+forkProcess :: LocalNode -> Process () -> IO ProcessId
120+forkProcess node (Process action) = do
121+
122+ (sendAddr, chan) <- Trans.newConnection (ndTransport node)
123+ sendEnd <- Trans.connectWith sendAddr Trans.SendHints
124+ processTable@(ProcessTable lpid _) <- takeMVar (ndProcessTable node)
125+ let pid = ProcessId sendEnd NodeId lpid
126+ forkIO $ do
127+ tid <- myThreadId
128+ lpid <- putMVar (ndProcessTable node) (insertProcess tid processTable)
129+ queue <- newCQueue
130+ forkIO $ receiverPump chan queue
131+
132+ let pstate = ProcessState {
133+ prPid = pid,
134+ prChan = chan,
135+ prQueue= queue,
136+ prNode = node
137+ }
138+
139+ action pstate
140+ return pid
141+
142+ where
143+ insertProcess :: ThreadId -> ProcessTable -> ProcessTable
144+ insertProcess tid (ProcessTable nextPid table) =
145+ let pte = ProcessTableEntry tid
146+ in ProcessTable (nextPid+1) (IntMap.insert nextPid pte table)
147+
148+ receiverPump chan queue = do
149+ msgBlobs <- Trans.receive chan
150+ let (typerep, body) = read (concatMap BS.unpack msgBlobs)
151+-- print ("receiverPump", typerep, body)
152+ enqueue queue (Message typerep body)
153+
154+send :: (Typeable a, Show a) => ProcessId -> a -> Process ()
155+send (ProcessId chan _ _) msg =
156+ liftIO (Trans.send chan msgBlobs)
157+ where
158+ msgBlobs :: [ByteString]
159+ msgBlobs = [BS.pack (show (show (typeOf msg), show msg))]
160+
161+expect :: forall a. (Typeable a, Read a) => Process a
162+expect = do
163+ ProcessState { prQueue = queue } <- getProcessState
164+ let typerepstr = show (typeOf (undefined :: a))
165+ Message _ body <- liftIO $
166+ dequeueMatching queue (\(Message typerepstr' _) -> typerepstr' == typerepstr)
167+ return (read body)
168+
169+spawnLocal :: Process () -> Process ProcessId
170+spawnLocal proc = do
171+ node <- getLocalNode
172+ liftIO $ forkProcess node proc
173+
174+{-
175+send :: Serializable a -> ProcessId -> a -> Process ()
176+expect :: Serializable a -> Process a
177+
178+newChan :: Serializable a => Process (SendPort a, ReceivePort a)
179+sendChan :: Serializable a => SendPort a -> a -> Process ()
180+receiveChan :: Serializable a => ReceivePort a -> Process a
181+
182+spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
183+terminate :: ProcessM a
184+getSelfPid :: ProcessM ProcessId
185+getSelfNode :: ProcessM NodeId
186+
187+monitorProcess :: ProcessId -> ProcessId -> MonitorAction -> Process ()
188+
189+-}
190+
191+-- Concurrent queue for single reader, single writer
192+--
193+data CQueue a = CQueue (MVar [a]) -- arrived
194+ (Chan a) -- incomming
195+
196+newCQueue :: IO (CQueue a)
197+newCQueue = do
198+ arrived <- newMVar []
199+ incomming <- newChan
200+ return (CQueue arrived incomming)
201+
202+enqueue :: CQueue a -> a -> IO ()
203+enqueue (CQueue _arrived incomming) a = writeChan incomming a
204+
205+dequeueMatching :: forall a. CQueue a -> (a -> Bool) -> IO a
206+dequeueMatching (CQueue arrived incomming) matches = do
207+ modifyMVar arrived (checkArrived [])
208+ where
209+ checkArrived :: [a] -> [a] -> IO ([a], a)
210+ checkArrived xs' [] = checkIncomming xs'
211+ checkArrived xs' (x:xs)
212+ | matches x = return (reverse xs' ++ xs, x)
213+ | otherwise = checkArrived (x:xs') xs
214+
215+ checkIncomming :: [a] -> IO ([a], a)
216+ checkIncomming xs' = do
217+ x <- readChan incomming
218+ if matches x
219+ then return (reverse xs', x)
220+ else checkIncomming (x:xs')
examples/DemoIP.hsnew file+61
@@ -0,0 +1,61 @@
1+module Main where
2+
3+import Network.Transport
4+import Network.Transport.IP
5+
6+import System.Environment
7+import Control.Monad
8+import qualified Data.ByteString.Char8 as BS
9+
10+
11+-- TODO:
12+-- * wikify duncan's email
13+-- * add current design to wiki
14+-- * link to CCI
15+-- * explain how to address peter's points
16+-- * package architecture:
17+-- * application layer
18+-- * cloud haskell layer
19+-- * transport layer
20+
21+main :: IO ()
22+main = do
23+ [mode, numberStr, addressStr, portStr] <- getArgs
24+
25+ let number = read numberStr
26+ let address = Address addressStr
27+ let port = read portStr
28+
29+ transport <- initTransport (Config undefined)
30+
31+ case mode of
32+ "master" -> do
33+ (clients, masterReceiveEnd) <- demoMaster number transport address port
34+ zipWithM_ (\clientSendEnd clientId -> send clientSendEnd [BS.pack . show $ clientId])
35+ clients [0 .. number-1]
36+ replicateM_ number $ do
37+ [clientId] <- receive masterReceiveEnd
38+ print clientId
39+ "slave" -> do
40+ (masterSendEnd, slaveReceiveEnd) <- demoSlave transport address port
41+ [clientId] <- receive slaveReceiveEnd
42+ let message = "Connected to slave: " ++ BS.unpack clientId
43+ send masterSendEnd [BS.pack message]
44+
45+demoMaster :: Int -> Transport -> Address -> Port -> IO ([SendEnd], ReceiveEnd)
46+demoMaster numSlaves transport address port = do
47+ receiveEnd <- newReceiveEnd address port
48+ sendEnds <- replicateM numSlaves $ do
49+ [bytes] <- receive receiveEnd
50+ case deserialize transport bytes of
51+ Nothing -> fail "Garbage message from slave"
52+ Just sendAddr -> connect sendAddr
53+ return (sendEnds, receiveEnd)
54+
55+demoSlave :: Transport -> Address -> Port -> IO (SendEnd, ReceiveEnd)
56+demoSlave transport address port = do
57+ (selfSendAddr, selfReceiveEnd) <- newConnection transport
58+ sendEnd <- connect (newSendAddr address port)
59+ send sendEnd [serialize selfSendAddr]
60+ return (sendEnd, selfReceiveEnd)
61+
examples/DemoProcess.hsnew file+35
@@ -0,0 +1,35 @@
1+module DemoProcess where
2+
3+import Control.Distributed.Process
4+import Control.Concurrent (threadDelay)
5+
6+import Network.Transport.MVar (mkTransport)
7+
8+demo1 = do
9+ trans <- mkTransport
10+ node <- newLocalNode trans
11+ print "foo"
12+ runProcess node $ do
13+ liftIO $ print "bar"
14+ lpid1 <- spawnLocal (logger 1)
15+-- liftIO $ print "baz"
16+-- lpid2 <- spawnLocal (logger 2)
17+
18+ spawnLocal (chatty "jim1" lpid1)
19+-- spawnLocal (chatty "jim2" lpid2)
20+ spawnLocal (chatty "bob1" lpid1)
21+-- spawnLocal (chatty "bob2" lpid2)
22+ liftIO $ threadDelay 500000
23+ return ()
24+
25+chatty :: String -> ProcessId -> Process ()
26+chatty name target = do
27+ send target (name ++ " says: hi there")
28+ liftIO $ threadDelay 500000
29+ send target (name ++ " says: bye")
30+
31+logger :: Int -> Process ()
32+logger n = do
33+ str <- expect
34+ liftIO $ putStrLn $ show n ++ ": " ++ str
35+ logger (n+1)
examples/DemoTransport.hsnew file+53
@@ -0,0 +1,53 @@
1+module DemoTransport where
2+
3+import Network.Transport
4+import Network.Transport.MVar (mkTransport)
5+
6+import Control.Concurrent
7+import qualified Data.ByteString.Char8 as BS
8+
9+-------------------------------------------
10+-- Example program using backend directly
11+--
12+
13+-- check the endpoint serialisation works
14+demo1 :: IO ()
15+demo1 = do
16+ trans <- mkTransport
17+
18+ (sendAddr, receiveEnd) <- newConnection trans
19+
20+ forkIO $ logServer receiveEnd
21+ threadDelay 100000
22+
23+ sendEnd <- connect sendAddr
24+
25+ -- This use of Just is slightly naughty
26+ let Just sendAddr' = deserialize trans (serialize sendAddr)
27+ sendEnd' <- connect sendAddr'
28+
29+ send sendEnd [BS.pack "hello 1"]
30+ send sendEnd' [BS.pack "hello 2"]
31+ threadDelay 100000
32+
33+-- check that we can send before any corresponding receive
34+demo2 :: IO ()
35+demo2 = do
36+ trans <- mkTransport
37+
38+ (sendAddr, receiveEnd) <- newConnection trans
39+
40+ sendEnd <- connect sendAddr
41+
42+ forkIO $ send sendEnd [BS.pack "hello 1"]
43+ threadDelay 100000
44+ forkIO $ logServer receiveEnd
45+ threadDelay 100000
46+ forkIO $ send sendEnd [BS.pack "hello 2"]
47+ threadDelay 100000
48+
49+logServer :: ReceiveEnd -> IO ()
50+logServer receiveEnd = do
51+ x <- receive receiveEnd
52+ print x
53+ logServer receiveEnd
network-transport/LICENSEnew file+31
@@ -0,0 +1,31 @@
1+Copyright Well-Typed LLP, 2011
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 the owner 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.
31+
network-transport/Setup.hsnew file+2
@@ -0,0 +1,2 @@
1+import Distribution.Simple
2+main = defaultMain
network-transport/network-transport.cabalnew file+21
@@ -0,0 +1,21 @@
1+Name: network-transport
2+Version: 0.0.1
3+Description: Network transport interface
4+License: BSD3
5+License-file: LICENSE
6+Author: Duncan Coutts, Nicolas Wu
7+Maintainer: duncan@well-typed.com
8+Homepage: http://github.com/haskell-distributed/
9+Build-Type: Simple
10+Cabal-Version: >=1.2
11+
12+Library
13+ Build-Depends: base >= 3 && < 5,
14+ bytestring >= 0.9,
15+ containers >= 0.4
16+ Exposed-modules: Network.Transport,
17+ Network.Transport.MVar,
18+ Network.Transport.TCP
19+ Extensions: BangPatterns
20+ ghc-options: -Wall
21+ HS-Source-Dirs: src
network-transport/src/Network/Transport.hsnew file+89
@@ -0,0 +1,89 @@
1+module Network.Transport
2+ ( Transport (..)
3+ , SendAddr (..)
4+ , SendEnd (..)
5+ , ReceiveEnd (..)
6+ , Hints (..)
7+ , defaultHints
8+ , SendHints (..)
9+ , defaultSendHints
10+ , newConnection
11+ , newMulticast
12+ , connect
13+ ) where
14+
15+import Data.ByteString.Char8 (ByteString)
16+
17+------------------------
18+-- Transport interface
19+--
20+
21+-- Buffer size
22+-- Sending: eager or buffered
23+-- Big record of defaults
24+data Hints = Hints
25+data SendHints = SendHints
26+
27+data Transport = Transport
28+ { newConnectionWith :: Hints -> IO (SendAddr, ReceiveEnd)
29+ , newMulticastWith :: Hints -> IO (MulticastSendEnd, MulticastReceiveAddr)
30+ , deserialize :: ByteString -> Maybe SendAddr
31+ }
32+
33+newConnection :: Transport -> IO (SendAddr, ReceiveEnd)
34+newConnection transport = newConnectionWith transport defaultHints
35+
36+newMulticast :: Transport -> IO (MulticastSendEnd, MulticastReceiveAddr)
37+newMulticast transport = newMulticastWith transport defaultHints
38+
39+defaultHints :: Hints
40+defaultHints = Hints
41+
42+data SendAddr = SendAddr
43+ { connectWith :: SendHints -> IO SendEnd
44+ , serialize :: ByteString
45+ }
46+
47+connect :: SendAddr -> IO SendEnd
48+connect sendAddr = connectWith sendAddr defaultSendHints
49+
50+defaultSendHints :: SendHints
51+defaultSendHints = SendHints
52+
53+-- Send and receive are vectored
54+data SendEnd = SendEnd
55+ { send :: [ByteString] -> IO ()
56+ -- , sendAddress :: SendAddr
57+ }
58+
59+newtype ReceiveEnd = ReceiveEnd
60+ { receive :: IO [ByteString]
61+ -- , receiveAddress :: SendAddr
62+ }
63+
64+data MulticastSendEnd = MulticastSendEnd
65+ { multicastSend :: ByteString -> IO ()
66+ }
67+
68+data MulticastReceiveAddr = MulticastReceiveAddr
69+ { multicastConnect :: IO MulticastReceiveEnd
70+ }
71+
72+data MulticastReceiveEnd = MulticastReceiveEnd
73+ { multicastReceive :: IO ByteString
74+ }
75+
76+-- data UnorderedSendEnd -- this is reliable
77+-- data UnreliableSendEnd -- this is also unordered
78+--
79+-- multicast is alwaysw unordered and unreliable
80+
81+-- TODO:
82+-- * Multicast
83+-- * Dual of the Transport: one to many, rather than many to one
84+-- * Optional: not supported by all transports
85+-- * Different transport types
86+-- * Unreliable
87+-- * Unordered
88+-- * Send / receive should be vectored
89+
network-transport/src/Network/Transport/MVar.hsnew file+59
@@ -0,0 +1,59 @@
1+{-# LANGUAGE BangPatterns #-}
2+
3+module Network.Transport.MVar
4+ ( mkTransport
5+ ) where
6+
7+import Control.Concurrent.MVar
8+import Data.IntMap (IntMap)
9+import Data.ByteString.Char8 (ByteString)
10+
11+import qualified Data.IntMap as IntMap
12+import qualified Data.ByteString.Char8 as BS
13+
14+import Network.Transport
15+
16+type Chans = MVar (Int, IntMap (MVar [ByteString]))
17+
18+mkTransport :: IO Transport
19+mkTransport = do
20+ channels <- newMVar (0, IntMap.empty)
21+ return Transport
22+ { newConnectionWith = \_ -> do
23+ (i, m) <- takeMVar channels
24+ receiveChan <- newEmptyMVar
25+ let sendAddr = i
26+ !i' = i+1
27+ !m' = IntMap.insert i receiveChan m
28+ putMVar channels (i', m')
29+ return (mkSendAddr channels sendAddr, mkReceiveEnd receiveChan)
30+ , newMulticastWith = undefined
31+ , deserialize = \bs ->
32+ case BS.readInt bs of
33+ Nothing -> error "dummyBackend.deserializeSendEnd: cannot parse"
34+ Just (n,_) -> Just . mkSendAddr channels $ n
35+ }
36+ where
37+ mkSendAddr :: Chans -> Int -> SendAddr
38+ mkSendAddr channels addr = SendAddr
39+ { connectWith = \_ -> mkSendEnd channels $ addr
40+ , serialize = BS.pack (show addr)
41+ }
42+ -- mkSendEnd channels sendAddr
43+
44+ mkSendEnd :: Chans -> Int -> IO SendEnd
45+ mkSendEnd channels addr = do
46+ (_, m) <- readMVar channels
47+ case IntMap.lookup addr m of
48+ Nothing -> fail "dummyBackend.send: bad send address"
49+ Just chan -> return SendEnd
50+ { send = realSend chan
51+ }
52+
53+ mkReceiveEnd :: MVar [ByteString] -> ReceiveEnd
54+ mkReceiveEnd chan = ReceiveEnd
55+ { receive = takeMVar chan
56+ }
57+
58+ realSend :: MVar [ByteString] -> [ByteString] -> IO ()
59+ realSend = putMVar
network-transport/src/Network/Transport/TCP.hsnew file+28
@@ -0,0 +1,28 @@
1+module Network.Transport.TCP
2+ ( mkTransport
3+ ) where
4+
5+import Network.Transport
6+import Data.Word
7+
8+data Address = Address String
9+type Port = Word16
10+
11+-- | This deals with several different configuration properties:
12+-- * Buffer size, specified in Hints
13+-- * LAN/WAN, since we can inspect the addresses
14+data TCPConfig = TCPConfig Hints Port Address Address
15+
16+mkTransport :: TCPConfig -> IO Transport
17+mkTransport tcpConfig =
18+ -- create buffer
19+ -- allocate listening socket
20+ -- fork thread on that socket
21+ return Transport
22+ { newConnectionWith = undefined
23+ -- create listening mailbox
24+ -- send back mailbox details
25+ , newMulticastWith = undefined
26+ , deserialize = undefined
27+ }
28+