git.iko.soy/distributed-process
on master help sign in
Replace SendEnd/SourceEnd ReceiveEnd/TargetEnd.
9c55ac5Nicolas Wu
distributed-process/src/Control/Distributed/Process.hs+11−11
compare with master
unifiedsplit
@@ -45,12 +45,12 @@ import Debug.Trace
45-- Cloud Haskell layer
45-- Cloud Haskell layer
46--
46--
47
47
48−data NodeId = NodeId -- !Trans.SendEnd !Trans.SendEnd
48+data NodeId = NodeId -- !Trans.SourceEnd !Trans.SourceEnd
49
49
50−data ProcessId = ProcessId !Trans.SendEnd !NodeId !LocalProcessId
50+data ProcessId = ProcessId !Trans.SourceEnd !NodeId !LocalProcessId
51type LocalProcessId = Int
51type LocalProcessId = Int
52
52
53−newtype SendPort a = SendPort Trans.SendEnd
53+newtype SourcePort a = SourcePort Trans.SourceEnd
54
54
55newtype Process a = Process { unProcess :: ProcessState -> IO a }
55newtype Process a = Process { unProcess :: ProcessState -> IO a }
56
56
@@ -81,7 +81,7 @@ getSelfPid = prPid <$> getProcessState
81--
81--
82data ProcessState = ProcessState {
82data ProcessState = ProcessState {
83 prPid :: !ProcessId,
83 prPid :: !ProcessId,
84− prChan :: !Trans.ReceiveEnd,
84+ prChan :: !Trans.TargetEnd,
85 prQueue :: !(CQueue Message),
85 prQueue :: !(CQueue Message),
86 prNode :: !LocalNode
86 prNode :: !LocalNode
87 }
87 }
@@ -124,10 +124,10 @@ runProcess node proc = do
124-- returns the ProcessId when the new process has been created.
124-- returns the ProcessId when the new process has been created.
125forkProcess :: LocalNode -> Process () -> IO ProcessId
125forkProcess :: LocalNode -> Process () -> IO ProcessId
126forkProcess node (Process action) = do
126forkProcess node (Process action) = do
127− (sendAddr, chan) <- Trans.newConnection (ndTransport node)
127+ (sourceAddr, chan) <- Trans.newConnection (ndTransport node)
128− sendEnd <- Trans.connect sendAddr
128+ sourceEnd <- Trans.connect sourceAddr
129 processTable@(ProcessTable lpid _) <- takeMVar (ndProcessTable node)
129 processTable@(ProcessTable lpid _) <- takeMVar (ndProcessTable node)
130− let pid = ProcessId sendEnd NodeId lpid
130+ let pid = ProcessId sourceEnd NodeId lpid
131 _ <- forkIO $ do
131 _ <- forkIO $ do
132 tid <- myThreadId
132 tid <- myThreadId
133 putMVar (ndProcessTable node) (insertProcess tid processTable)
133 putMVar (ndProcessTable node) (insertProcess tid processTable)
@@ -142,7 +142,7 @@ forkProcess node (Process action) = do
142 let pte = ProcessTableEntry tid
142 let pte = ProcessTableEntry tid
143 in ProcessTable (nextPid+1) (IntMap.insert nextPid pte table)
143 in ProcessTable (nextPid+1) (IntMap.insert nextPid pte table)
144
144
145− receiverPump :: Trans.ReceiveEnd -> CQueue Message -> IO ()
145+ receiverPump :: Trans.TargetEnd -> CQueue Message -> IO ()
146 receiverPump chan queue = forever $ do
146 receiverPump chan queue = forever $ do
147 msgBlobs <- Trans.receive chan
147 msgBlobs <- Trans.receive chan
148 readBlobs (concatMap BS.unpack msgBlobs)
148 readBlobs (concatMap BS.unpack msgBlobs)
@@ -179,9 +179,9 @@ spawnLocal proc = do
179send :: Serializable a -> ProcessId -> a -> Process ()
179send :: Serializable a -> ProcessId -> a -> Process ()
180expect :: Serializable a -> Process a
180expect :: Serializable a -> Process a
181
181
182−newChan :: Serializable a => Process (SendPort a, ReceivePort a)
182+newChan :: Serializable a => Process (SourcePort a, TargetPort a)
183−sendChan :: Serializable a => SendPort a -> a -> Process ()
183+sendChan :: Serializable a => SourcePort a -> a -> Process ()
184−receiveChan :: Serializable a => ReceivePort a -> Process a
184+receiveChan :: Serializable a => TargetPort a -> Process a
185
185
186spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
186spawn :: NodeId -> Closure (Process ()) -> Process ProcessId
187terminate :: ProcessM a
187terminate :: ProcessM a
examples/DemoTCP.hs+35−35
@@ -25,7 +25,7 @@ type ChanId = Int
25-- ./DemoTCP 127.0.0.1 8082
25-- ./DemoTCP 127.0.0.1 8082
26--
26--
27-- Once the connection is established, the slaves will provide the master
27-- Once the connection is established, the slaves will provide the master
28−-- with their SendAddr. The master then decodes this, and sends a message
28+-- with their SourceAddr. The master then decodes this, and sends a message
29-- back. After processing this message, the slaves respond to the master,
29-- back. After processing this message, the slaves respond to the master,
30-- which then outputs some data.
30-- which then outputs some data.
31main :: IO ()
31main :: IO ()
@@ -34,71 +34,71 @@ main = do
34 case args of
34 case args of
35 -- Master:
35 -- Master:
36 -- The master only provides the host, service, and slaves to expect.
36 -- The master only provides the host, service, and slaves to expect.
37− [host, service, sendAddrFilePath, number] -> do
37+ [host, service, sourceAddrFilePath, number] -> do
38 let numSlaves = read number
38 let numSlaves = read number
39 transport <- mkTransport $ TCPConfig defaultHints host service
39 transport <- mkTransport $ TCPConfig defaultHints host service
40− (clientSendEnds, masterReceiveEnd) <- demoMaster transport sendAddrFilePath numSlaves
40+ (clientSourceEnds, masterTargetEnd) <- demoMaster transport sourceAddrFilePath numSlaves
41 zipWithM_
41 zipWithM_
42− (\clientSendEnd clientId -> send clientSendEnd [BS.pack . show $ clientId])
42+ (\clientSourceEnd clientId -> send clientSourceEnd [BS.pack . show $ clientId])
43− clientSendEnds [0 .. numSlaves-1]
43+ clientSourceEnds [0 .. numSlaves-1]
44 replicateM_ numSlaves $ do
44 replicateM_ numSlaves $ do
45− [clientMessage] <- receive masterReceiveEnd
45+ [clientMessage] <- receive masterTargetEnd
46 print clientMessage
46 print clientMessage
47
47
48 -- Slave:
48 -- Slave:
49 -- Each slave provides its own host and service, as well as details of
49 -- Each slave provides its own host and service, as well as details of
50 -- the master it wishes to connect to.
50 -- the master it wishes to connect to.
51− [host, service, sendAddrFilePath] -> do
51+ [host, service, sourceAddrFilePath] -> do
52 transport <- mkTransport $ TCPConfig defaultHints host service
52 transport <- mkTransport $ TCPConfig defaultHints host service
53− (masterSendEnd, slaveReceiveEnd) <- demoSlave transport sendAddrFilePath
53+ (masterSourceEnd, slaveTargetEnd) <- demoSlave transport sourceAddrFilePath
54− [clientId] <- receive slaveReceiveEnd
54+ [clientId] <- receive slaveTargetEnd
55 putStrLn $ "slave: received clientId: " ++ show clientId
55 putStrLn $ "slave: received clientId: " ++ show clientId
56 let message = "Connected to slave: " ++ BS.unpack clientId
56 let message = "Connected to slave: " ++ BS.unpack clientId
57
57
58− send masterSendEnd [BS.pack message]
58+ send masterSourceEnd [BS.pack message]
59 putStrLn "slave: sent message to master"
59 putStrLn "slave: sent message to master"
60
60
61 _ -> error "Unexpected arguments"
61 _ -> error "Unexpected arguments"
62
62
63demoMaster :: Transport -- ^ Transport
63demoMaster :: Transport -- ^ Transport
64− -> FilePath -- ^ File to write SendAddr
64+ -> FilePath -- ^ File to write SourceAddr
65 -> Int -- ^ Number of slaves to expect
65 -> Int -- ^ Number of slaves to expect
66− -> IO ([SendEnd], ReceiveEnd)
66+ -> IO ([SourceEnd], TargetEnd)
67−demoMaster transport sendAddrFilePath numSlaves = do
67+demoMaster transport sourceAddrFilePath numSlaves = do
68 -- Establish an initial transport
68 -- Establish an initial transport
69− (sendAddr, receiveEnd) <- newConnection transport
69+ (sourceAddr, targetEnd) <- newConnection transport
70
70
71− -- Write a file containing the sendAddr
71+ -- Write a file containing the sourceAddr
72− putStrLn $ "master: writing sendAddr file:" ++ (show . serialize $ sendAddr)
72+ putStrLn $ "master: writing sourceAddr file:" ++ (show . serialize $ sourceAddr)
73− BS.writeFile sendAddrFilePath $ serialize sendAddr
73+ BS.writeFile sourceAddrFilePath $ serialize sourceAddr
74
74
75− -- Deserialize each of the sendEnds received by the slaves
75+ -- Deserialize each of the sourceEnds received by the slaves
76− sendEnds <- replicateM numSlaves $ do
76+ sourceEnds <- replicateM numSlaves $ do
77− [bytes] <- receive receiveEnd
77+ [bytes] <- receive targetEnd
78 putStrLn "master: received connection."
78 putStrLn "master: received connection."
79 case deserialize transport bytes of
79 case deserialize transport bytes of
80 Nothing -> fail "Garbage message from slave!"
80 Nothing -> fail "Garbage message from slave!"
81− Just sendAddr -> connect sendAddr
81+ Just sourceAddr -> connect sourceAddr
82− return (sendEnds, receiveEnd)
82+ return (sourceEnds, targetEnd)
83
83
84demoSlave :: Transport -- ^ Transport
84demoSlave :: Transport -- ^ Transport
85− -> FilePath -- ^ File to read SendAddr
85+ -> FilePath -- ^ File to read SourceAddr
86− -> IO (SendEnd, ReceiveEnd)
86+ -> IO (SourceEnd, TargetEnd)
87−demoSlave transport masterSendAddrFilePath = do
87+demoSlave transport masterSourceAddrFilePath = do
88 -- Create a new connection on the transport
88 -- Create a new connection on the transport
89− (sendAddr, receiveEnd) <- newConnection transport
89+ (sourceAddr, targetEnd) <- newConnection transport
90
90
91 -- Establish connection with the master
91 -- Establish connection with the master
92− bytes <- BS.readFile masterSendAddrFilePath
92+ bytes <- BS.readFile masterSourceAddrFilePath
93 case deserialize transport bytes of
93 case deserialize transport bytes of
94− Just masterSendAddr -> do
94+ Just masterSourceAddr -> do
95− putStrLn $ "slave: connecting to sendAddr: " ++ (show . serialize $ masterSendAddr)
95+ putStrLn $ "slave: connecting to sourceAddr: " ++ (show . serialize $ masterSourceAddr)
96− masterSendEnd <- connect masterSendAddr
96+ masterSourceEnd <- connect masterSourceAddr
97
97
98− -- Send the serialized address of this slave
98+ -- Source the serialized address of this slave
99− send masterSendEnd [serialize sendAddr]
99+ send masterSourceEnd [serialize sourceAddr]
100− putStrLn "slave: sent sendAddr"
100+ putStrLn "slave: sent sourceAddr"
101− return (masterSendEnd, receiveEnd)
101+ return (masterSourceEnd, targetEnd)
102
102
103− Nothing -> error "slave: Unable to deserialize master SendAddr!"
103+ Nothing -> error "slave: Unable to deserialize master SourceAddr!"
104
104
examples/DemoTransport.hs+35−35
@@ -20,14 +20,14 @@ demo0 :: IO ()
20demo0 = do
20demo0 = do
21 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
21 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
22
22
23− (sendAddr, receiveEnd) <- newConnection trans
23+ (sourceAddr, targetEnd) <- newConnection trans
24
24
25− forkIO $ logServer "logServer" receiveEnd
25+ forkIO $ logServer "logServer" targetEnd
26 threadDelay 100000
26 threadDelay 100000
27
27
28− sendEnd <- connect sendAddr
28+ sourceEnd <- connect sourceAddr
29
29
30− mapM_ (\n -> send sendEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
30+ mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
31 threadDelay 100000
31 threadDelay 100000
32
32
33-- | Check endpoint serialization and deserialization.
33-- | Check endpoint serialization and deserialization.
@@ -36,19 +36,19 @@ demo1 = do
36 -- trans <- mkTransport
36 -- trans <- mkTransport
37 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
37 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
38
38
39− (sendAddr, receiveEnd) <- newConnection trans
39+ (sourceAddr, targetEnd) <- newConnection trans
40
40
41− forkIO $ logServer "logServer" receiveEnd
41+ forkIO $ logServer "logServer" targetEnd
42 threadDelay 100000
42 threadDelay 100000
43
43
44− sendEnd <- connect sendAddr
44+ sourceEnd <- connect sourceAddr
45
45
46 -- This use of Just is slightly naughty
46 -- This use of Just is slightly naughty
47− let Just sendAddr' = deserialize trans (serialize sendAddr)
47+ let Just sourceAddr' = deserialize trans (serialize sourceAddr)
48− sendEnd' <- connect sendAddr'
48+ sourceEnd' <- connect sourceAddr'
49
49
50− send sendEnd [BS.pack "hello 1"]
50+ send sourceEnd [BS.pack "hello 1"]
51− send sendEnd' [BS.pack "hello 2"]
51+ send sourceEnd' [BS.pack "hello 2"]
52 threadDelay 100000
52 threadDelay 100000
53
53
54-- | Check that messages can be sent before receive is set up.
54-- | Check that messages can be sent before receive is set up.
@@ -57,13 +57,13 @@ demo2 = do
57 -- trans <- mkTransport
57 -- trans <- mkTransport
58 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
58 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
59
59
60− (sendAddr, receiveEnd) <- newConnection trans
60+ (sourceAddr, targetEnd) <- newConnection trans
61
61
62− sendEnd <- connect sendAddr
62+ sourceEnd <- connect sourceAddr
63− forkIO $ send sendEnd [BS.pack "hello 1"]
63+ forkIO $ send sourceEnd [BS.pack "hello 1"]
64 threadDelay 100000
64 threadDelay 100000
65
65
66− forkIO $ logServer "logServer" receiveEnd
66+ forkIO $ logServer "logServer" targetEnd
67 return ()
67 return ()
68
68
69-- | Check that two different transports can be created.
69-- | Check that two different transports can be created.
@@ -72,17 +72,17 @@ demo3 = do
72 trans1 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
72 trans1 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
73 trans2 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8081"
73 trans2 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8081"
74
74
75− (sendAddr1, receiveEnd1) <- newConnection trans1
75+ (sourceAddr1, targetEnd1) <- newConnection trans1
76− (sendAddr2, receiveEnd2) <- newConnection trans2
76+ (sourceAddr2, targetEnd2) <- newConnection trans2
77
77
78− forkIO $ logServer "logServer1" receiveEnd1
78+ forkIO $ logServer "logServer1" targetEnd1
79− forkIO $ logServer "logServer2" receiveEnd2
79+ forkIO $ logServer "logServer2" targetEnd2
80
80
81− sendEnd1 <- connect sendAddr1
81+ sourceEnd1 <- connect sourceAddr1
82− sendEnd2 <- connect sendAddr2
82+ sourceEnd2 <- connect sourceAddr2
83
83
84− send sendEnd1 [BS.pack "hello1"]
84+ send sourceEnd1 [BS.pack "hello1"]
85− send sendEnd2 [BS.pack "hello2"]
85+ send sourceEnd2 [BS.pack "hello2"]
86
86
87-- | Check that two different connections on the same transport can be created.
87-- | Check that two different connections on the same transport can be created.
88demo4 :: IO ()
88demo4 :: IO ()
@@ -90,19 +90,19 @@ demo4 = do
90 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
90 trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"
91 -- trans <- Network.Transport.MVar.mkTransport
91 -- trans <- Network.Transport.MVar.mkTransport
92
92
93− (sendAddr1, receiveEnd1) <- newConnection trans
93+ (sourceAddr1, targetEnd1) <- newConnection trans
94− (sendAddr2, receiveEnd2) <- newConnection trans
94+ (sourceAddr2, targetEnd2) <- newConnection trans
95
95
96− forkIO $ logServer "logServer1" receiveEnd1
96+ forkIO $ logServer "logServer1" targetEnd1
97− forkIO $ logServer "logServer2" receiveEnd2
97+ forkIO $ logServer "logServer2" targetEnd2
98
98
99− sendEnd1 <- connect sendAddr1
99+ sourceEnd1 <- connect sourceAddr1
100− sendEnd2 <- connect sendAddr2
100+ sourceEnd2 <- connect sourceAddr2
101
101
102− send sendEnd1 [BS.pack "hello1"]
102+ send sourceEnd1 [BS.pack "hello1"]
103− send sendEnd2 [BS.pack "hello2"]
103+ send sourceEnd2 [BS.pack "hello2"]
104
104
105−logServer :: String -> ReceiveEnd -> IO ()
105+logServer :: String -> TargetEnd -> IO ()
106−logServer name receiveEnd = forever $ do
106+logServer name targetEnd = forever $ do
107− x <- receive receiveEnd
107+ x <- receive targetEnd
108− trace (name ++ ": " ++ show x) $ return ()
108+ trace (name ++ ": " ++ show x) $ return ()
network-transport/src/Network/Transport.hs+41−41
@@ -1,96 +1,96 @@
1module Network.Transport
1module Network.Transport
2 ( Hints (..)
2 ( Hints (..)
3− , ReceiveEnd (..)
3+ , TargetEnd (..)
4− , SendAddr (..)
4+ , SourceAddr (..)
5− , SendEnd (..)
5+ , SourceEnd (..)
6− , SendHints (..)
6+ , SourceHints (..)
7 , Transport (..)
7 , Transport (..)
8 , connect
8 , connect
9 , defaultHints
9 , defaultHints
10− , defaultSendHints
10+ , defaultSourceHints
11 , newConnection
11 , newConnection
12 , newMulticast
12 , newMulticast
13 ) where
13 ) where
14
14
15import Data.ByteString.Char8 (ByteString)
15import Data.ByteString.Char8 (ByteString)
16
16
17−-- | The `Hints` and `SendHints` provide hints to the underlying transport
17+-- | The `Hints` and `SourceHints` provide hints to the underlying transport
18-- about the kind of connection that is required. This might include details
18-- about the kind of connection that is required. This might include details
19-- such as whether the connection is eager or buffered, and the buffer size.
19-- such as whether the connection is eager or buffered, and the buffer size.
20data Hints = Hints
20data Hints = Hints
21−data SendHints = SendHints
21+data SourceHints = SourceHints
22
22
23-- | A `Transport` encapsulates the functions required to establish many-to-one
23-- | A `Transport` encapsulates the functions required to establish many-to-one
24-- and one-to-many connections between clients and servers.
24-- and one-to-many connections between clients and servers.
25−-- The `newConnectionWith` function creates a `ReceiveEnd` that listens to
25+-- The `newConnectionWith` function creates a `TargetEnd` that listens to
26−-- messages sent using the corresponding `SendAddr`. This connection is
26+-- messages sent using the corresponding `SourceAddr`. This connection is
27-- established using a `Hints` value, which provides information about the
27-- established using a `Hints` value, which provides information about the
28-- connection topology.
28-- connection topology.
29−-- Each `SendAddr` can be serialised into a `ByteString`, and the `deserialize`
29+-- Each `SourceAddr` can be serialised into a `ByteString`, and the `deserialize`
30−-- function converts this back into a `SendAddr`.
30+-- function converts this back into a `SourceAddr`.
31-- Note that these connections provide reliable and ordered messages.
31-- Note that these connections provide reliable and ordered messages.
32data Transport = Transport
32data Transport = Transport
33− { newConnectionWith :: Hints -> IO (SendAddr, ReceiveEnd)
33+ { newConnectionWith :: Hints -> IO (SourceAddr, TargetEnd)
34− , newMulticastWith :: Hints -> IO (MulticastSendEnd, MulticastReceiveAddr)
34+ , newMulticastWith :: Hints -> IO (MulticastSourceEnd, MulticastTargetAddr)
35− , deserialize :: ByteString -> Maybe SendAddr
35+ , deserialize :: ByteString -> Maybe SourceAddr
36 }
36 }
37
37
38-- | This is a convenience function that creates a new connection on a transport
38-- | This is a convenience function that creates a new connection on a transport
39-- using the default hints.
39-- using the default hints.
40−newConnection :: Transport -> IO (SendAddr, ReceiveEnd)
40+newConnection :: Transport -> IO (SourceAddr, TargetEnd)
41newConnection transport = newConnectionWith transport defaultHints
41newConnection transport = newConnectionWith transport defaultHints
42
42
43−newMulticast :: Transport -> IO (MulticastSendEnd, MulticastReceiveAddr)
43+newMulticast :: Transport -> IO (MulticastSourceEnd, MulticastTargetAddr)
44newMulticast transport = newMulticastWith transport defaultHints
44newMulticast transport = newMulticastWith transport defaultHints
45
45
46-- | The default `Hints` for establishing a new transport connection.
46-- | The default `Hints` for establishing a new transport connection.
47defaultHints :: Hints
47defaultHints :: Hints
48defaultHints = Hints
48defaultHints = Hints
49
49
50−-- | A `SendAddr` is an address that corresponds to a listening `ReceiveEnd`
50+-- | A `SourceAddr` is an address that corresponds to a listening `TargetEnd`
51−-- initially created using `newConnection`. A `SendAddr` can be shared between
51+-- initially created using `newConnection`. A `SourceAddr` can be shared between
52-- clients by using `serialize`, and passing the resulting `ByteString`.
52-- clients by using `serialize`, and passing the resulting `ByteString`.
53−-- Given a `SendAddr`, the `connectWith` function creates a `SendEnd` which
53+-- Given a `SourceAddr`, the `connectWith` function creates a `SourceEnd` which
54-- can be used to send messages.
54-- can be used to send messages.
55−data SendAddr = SendAddr
55+data SourceAddr = SourceAddr
56− { connectWith :: SendHints -> IO SendEnd
56+ { connectWith :: SourceHints -> IO SourceEnd
57 , serialize :: ByteString
57 , serialize :: ByteString
58 }
58 }
59
59
60−-- | This is a convenience function that connects with a given `SendAddr` using
60+-- | This is a convenience function that connects with a given `SourceAddr` using
61-- the default hints for sending.
61-- the default hints for sending.
62−connect :: SendAddr -> IO SendEnd
62+connect :: SourceAddr -> IO SourceEnd
63−connect sendAddr = connectWith sendAddr defaultSendHints
63+connect sourceAddr = connectWith sourceAddr defaultSourceHints
64
64
65−-- | The default `SendHints` for establishing a `SendEnd`.
65+-- | The default `SourceHints` for establishing a `SourceEnd`.
66−defaultSendHints :: SendHints
66+defaultSourceHints :: SourceHints
67−defaultSendHints = SendHints
67+defaultSourceHints = SourceHints
68
68
69−-- | A `SendEnd` provides a `send` function that allows vectored messages
69+-- | A `SourceEnd` provides a `send` function that allows vectored messages
70−-- to be sent to the corresponding `ReceiveEnd`.
70+-- to be sent to the corresponding `TargetEnd`.
71−newtype SendEnd = SendEnd
71+newtype SourceEnd = SourceEnd
72 { send :: [ByteString] -> IO ()
72 { send :: [ByteString] -> IO ()
73 }
73 }
74
74
75−-- | A `ReceiveEnd` provides a `receive` function that allows messages
75+-- | A `TargetEnd` provides a `receive` function that allows messages
76−-- to be received from the corresponding `SendEnd`s.
76+-- to be received from the corresponding `SourceEnd`s.
77−newtype ReceiveEnd = ReceiveEnd
77+newtype TargetEnd = TargetEnd
78 { receive :: IO [ByteString]
78 { receive :: IO [ByteString]
79 }
79 }
80
80
81−newtype MulticastSendEnd = MulticastSendEnd
81+newtype MulticastSourceEnd = MulticastSourceEnd
82− { multicastSend :: ByteString -> IO ()
82+ { multicastSource :: ByteString -> IO ()
83 }
83 }
84
84
85−newtype MulticastReceiveAddr = MulticastReceiveAddr
85+newtype MulticastTargetAddr = MulticastTargetAddr
86− { multicastConnect :: IO MulticastReceiveEnd
86+ { multicastConnect :: IO MulticastTargetEnd
87 }
87 }
88
88
89−newtype MulticastReceiveEnd = MulticastReceiveEnd
89+newtype MulticastTargetEnd = MulticastTargetEnd
90 { multicastReceive :: IO ByteString
90 { multicastReceive :: IO ByteString
91 }
91 }
92
92
93−-- TODO: Other SendEnds that might be of use:
93+-- TODO: Other SourceEnds that might be of use:
94−-- data UnorderedSendEnd -- reliable, unordered
94+-- data UnorderedSourceEnd -- reliable, unordered
95−-- data UnreliableSendEnd -- unreliable, unordered
95+-- data UnreliableSourceEnd -- unreliable, unordered
96
96
network-transport/src/Network/Transport/MVar.hs+12−12
@@ -22,35 +22,35 @@ mkTransport = do
22 { newConnectionWith = \_ -> do
22 { newConnectionWith = \_ -> do
23 (i, m) <- takeMVar channels
23 (i, m) <- takeMVar channels
24 receiveChan <- newEmptyMVar
24 receiveChan <- newEmptyMVar
25− let sendAddr = i
25+ let sourceAddr = i
26 !i' = i+1
26 !i' = i+1
27 !m' = IntMap.insert i receiveChan m
27 !m' = IntMap.insert i receiveChan m
28 putMVar channels (i', m')
28 putMVar channels (i', m')
29− return (mkSendAddr channels sendAddr, mkReceiveEnd receiveChan)
29+ return (mkSourceAddr channels sourceAddr, mkTargetEnd receiveChan)
30 , newMulticastWith = undefined
30 , newMulticastWith = undefined
31 , deserialize = \bs ->
31 , deserialize = \bs ->
32 case BS.readInt bs of
32 case BS.readInt bs of
33− Nothing -> error "dummyBackend.deserializeSendEnd: cannot parse"
33+ Nothing -> error "dummyBackend.deserializeSourceEnd: cannot parse"
34− Just (n,_) -> Just . mkSendAddr channels $ n
34+ Just (n,_) -> Just . mkSourceAddr channels $ n
35 }
35 }
36 where
36 where
37− mkSendAddr :: Chans -> Int -> SendAddr
37+ mkSourceAddr :: Chans -> Int -> SourceAddr
38− mkSendAddr channels addr = SendAddr
38+ mkSourceAddr channels addr = SourceAddr
39− { connectWith = \_ -> mkSendEnd channels addr
39+ { connectWith = \_ -> mkSourceEnd channels addr
40 , serialize = BS.pack (show addr)
40 , serialize = BS.pack (show addr)
41 }
41 }
42
42
43− mkSendEnd :: Chans -> Int -> IO SendEnd
43+ mkSourceEnd :: Chans -> Int -> IO SourceEnd
44− mkSendEnd channels addr = do
44+ mkSourceEnd channels addr = do
45 (_, m) <- readMVar channels
45 (_, m) <- readMVar channels
46 case IntMap.lookup addr m of
46 case IntMap.lookup addr m of
47 Nothing -> fail "dummyBackend.send: bad send address"
47 Nothing -> fail "dummyBackend.send: bad send address"
48− Just chan -> return SendEnd
48+ Just chan -> return SourceEnd
49 { send = realSend chan
49 { send = realSend chan
50 }
50 }
51
51
52− mkReceiveEnd :: MVar [ByteString] -> ReceiveEnd
52+ mkTargetEnd :: MVar [ByteString] -> TargetEnd
53− mkReceiveEnd chan = ReceiveEnd
53+ mkTargetEnd chan = TargetEnd
54 { receive = takeMVar chan
54 { receive = takeMVar chan
55 }
55 }
56
56
network-transport/src/Network/Transport/TCP.hs+14−14
@@ -31,8 +31,8 @@ type ChanId = Int
31data TCPConfig = TCPConfig Hints HostName ServiceName
31data TCPConfig = TCPConfig Hints HostName ServiceName
32
32
33-- | This creates a TCP connection between a server and a number of
33-- | This creates a TCP connection between a server and a number of
34−-- clients. Behind the scenes, the server hostname is passed as the SendAddr
34+-- clients. Behind the scenes, the server hostname is passed as the SourceAddr
35−-- and when a connection is made, messages sent down the SendEnd go
35+-- and when a connection is made, messages sent down the SourceEnd go
36-- via a socket made for the client that connected.
36-- via a socket made for the client that connected.
37-- Messages are all queued using an unbounded Chan.
37-- Messages are all queued using an unbounded Chan.
38mkTransport :: TCPConfig -> IO Transport
38mkTransport :: TCPConfig -> IO Transport
@@ -55,37 +55,37 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
55 (chanId, chanMap) <- takeMVar channels
55 (chanId, chanMap) <- takeMVar channels
56 chan <- newChan
56 chan <- newChan
57 putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
57 putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
58− return (mkSendAddr host service chanId, mkReceiveEnd chan)
58+ return (mkSourceAddr host service chanId, mkTargetEnd chan)
59 , newMulticastWith = undefined
59 , newMulticastWith = undefined
60 , deserialize = \bs ->
60 , deserialize = \bs ->
61 case readMay . BS.unpack $ bs of
61 case readMay . BS.unpack $ bs of
62 Nothing -> error "deserialize: cannot parse"
62 Nothing -> error "deserialize: cannot parse"
63− Just (host, service, chanId) -> Just $ mkSendAddr host service chanId
63+ Just (host, service, chanId) -> Just $ mkSourceAddr host service chanId
64 }
64 }
65
65
66 where
66 where
67− mkSendAddr :: HostName -> ServiceName -> ChanId -> SendAddr
67+ mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
68− mkSendAddr host service chanId = SendAddr
68+ mkSourceAddr host service chanId = SourceAddr
69− { connectWith = \_ -> mkSendEnd host service chanId
69+ { connectWith = \_ -> mkSourceEnd host service chanId
70 , serialize = BS.pack . show $ (host, service, chanId)
70 , serialize = BS.pack . show $ (host, service, chanId)
71 }
71 }
72
72
73− mkSendEnd :: HostName -> ServiceName -> ChanId -> IO SendEnd
73+ mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
74− mkSendEnd host service chanId = withSocketsDo $ do
74+ mkSourceEnd host service chanId = withSocketsDo $ do
75 serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
75 serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
76 let serverAddr = case serverAddrs of
76 let serverAddr = case serverAddrs of
77− [] -> error "mkSendEnd: getAddrInfo returned []"
77+ [] -> error "mkSourceEnd: getAddrInfo returned []"
78 as -> head as
78 as -> head as
79 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
79 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
80− setSocketOption sock KeepAlive 1
80+ setSocketOption sock ReuseAddr 1
81 N.connect sock (addrAddress serverAddr)
81 N.connect sock (addrAddress serverAddr)
82 NBS.sendAll sock $ BS.pack . show $ chanId
82 NBS.sendAll sock $ BS.pack . show $ chanId
83− return $ SendEnd
83+ return $ SourceEnd
84 { Network.Transport.send = \bss -> NBS.sendMany sock bss
84 { Network.Transport.send = \bss -> NBS.sendMany sock bss
85 }
85 }
86
86
87− mkReceiveEnd :: Chan ByteString -> ReceiveEnd
87+ mkTargetEnd :: Chan ByteString -> TargetEnd
88− mkReceiveEnd chan = ReceiveEnd
88+ mkTargetEnd chan = TargetEnd
89 { -- for now we will implement this as a Chan
89 { -- for now we will implement this as a Chan
90 receive = do
90 receive = do
91 bs <- readChan chan
91 bs <- readChan chan