git.iko.soy/distributed-process
on master help sign in
Merge branch 'master' of git://github.com/haskell-distributed/distributed-process
7041cc4Ryan Newton
.gitignore+4
compare with master
unifiedsplit
@@ -3,5 +3,9 @@ TAGS
3dist/
3dist/
4\#*#
4\#*#
5*~
5*~
6+*.aux
7+*.hp
6*.hi
8*.hi
7*.o
9*.o
10+*.prof
11+*.ps
README.md+98
@@ -12,5 +12,103 @@ At present, this repository hosts two packages:
12For more detailed information about the interfaces provided by these packages,
12For more detailed information about the interfaces provided by these packages,
13please refer to the [distributed-process repository wiki][2].
13please refer to the [distributed-process repository wiki][2].
14
14
15+Examples
16+--------
17+
18+Examples of the transport layer at work can be found in the `/examples` folder.
19+
20+### DemoTransport.hs
21+
22+The `DemoTransport.hs` file contains a number of different functions named
23+`demo#`. These functions demonstrate the use of the `Network.Transport` module.
24+
25+To try them out, launch `ghci`, including the appropriate sources:
26+
27+ ghci -idistributed-process/src -inetwork-transport/src examples/DemoTransport.hs
28+
29+The first demo, `demo0` yields the following output:
30+
31+ *DemoTransport> demo0
32+ logServer: [Chunk "hello 1" Empty]
33+ logServer: [Chunk "hello 2" Empty]
34+ logServer: [Chunk "hello 3" Empty]
35+ logServer: [Chunk "hello 4" Empty]
36+ logServer: [Chunk "hello 5" Empty]
37+ logServer: [Chunk "hello 6" Empty]
38+ logServer: [Chunk "hello 7" Empty]
39+ logServer: [Chunk "hello 8" Empty]
40+ logServer: [Chunk "hello 9" Empty]
41+ logServer: [Chunk "hello 10" Empty]
42+
43+See the source code for more details.
44+
45+### DemoProcess.hs
46+
47+The `DemoProcess.hs` file shows a very basic example of how to use the
48+`Control.Distributed.Process` module in conjunction with `Network.Transport.TCP`.
49+This is demonstrated using the `demo1` function:
50+
51+ ghci -idistributed-process/src -inetwork-transport/src examples/DemoProcess.hs
52+
53+The output of the example is as follows:
54+
55+ *DemoProcess> demo1
56+ "Starting node"
57+ "Starting process"
58+ 1: jim1 says: hi there
59+ 2: bob1 says: hi there
60+ 3: jim1 says: bye
61+ 4: bob1 says: bye
62+
63+### DemoTCP.hs
64+
65+The `DemoTCP.hs` file demonstrates how a master can be connected to slaves
66+using the transport layer over TCP.
67+This example should be compiled in the usual fashion:
68+
69+ ghc --make -idistributed-process/src -inetwork-transport/src examples/DemoTCP.hs
70+
71+After compiling, the following will initialize a new master that waits
72+for two clients to connect on the local machine:
73+
74+ ./examples/DemoTCP 127.0.0.1 8080 sourceAddrFile 2
75+
76+Following this, two slaves should be created that will connect to the master:
77+
78+ ./examples/DemoTCP 127.0.0.1 8081 sourceAddrFile
79+ ./examples/DemoTCP 127.0.0.1 8082 sourceAddrFile
80+
81+Once the connection is established, the slaves will provide the master
82+with their `SourceAddr`. The master then decodes this, and sends a message
83+back. After processing this message, the slaves respond to the master,
84+which then outputs some data.
85+
86+Benchmarks
87+----------
88+
89+The `Transport.TCP` backend uses `Network.ByteString.Lazy` as its underlying
90+protocol, so we produce a benchmark that compares it directly with this.
91+Furthermore, the `Data.Binary` package is used for converting values
92+to `ByteStrings` for transmission.
93+
94+The benchmarking is performed on the client side using criterion:
95+the server must be set up before benching is performed.
96+
97+### Ping
98+
99+This benchmark measures how quickly a sequence of ping/pong messages
100+are sent between a server and client. Full details can be found in:
101+
102+ ./benchmarks/PingTCP.hs
103+ ./benchmarks/PingTransport.hs
104+
105+### Send
106+
107+This benchmark measures how quickly a number of bytes can be sent from
108+a client to a server. Full details can be found in the following files:
109+
110+ ./benchmarks/SendTCP.hs
111+ ./benchmarks/SendTransport.hs
112+
15[1]: http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
113[1]: http://research.microsoft.com/en-us/um/people/simonpj/papers/parallel/remote.pdf
16[2]: https://github.com/haskell-distributed/distributed-process/wiki
114[2]: https://github.com/haskell-distributed/distributed-process/wiki
benchmarks/PingTCP.hsnew file+88
@@ -0,0 +1,88 @@
1+module Main where
2+
3+import Control.Monad
4+import Criterion.Main (Benchmark, bench, defaultMain, nfIO, bgroup)
5+import Data.Binary
6+import Data.Int
7+import Network.Socket
8+ ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
9+ , SocketType (Stream), SocketOption (ReuseAddr)
10+ , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
11+ , defaultHints
12+ , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
13+import System.Environment (getArgs, withArgs)
14+
15+import qualified Network.Socket as N
16+import qualified Network.Socket.ByteString.Lazy as NBS
17+
18+import Debug.Trace
19+
20+-- | This performs a ping benchmark on a TCP connection created by
21+-- Network.Socket. To compile this file, you might use:
22+--
23+-- ghc --make -idistributed-process/src -inetwork-transport/src -O2 benchmarks/PingTCP.hs
24+--
25+-- To use the compiled binary, first set up the server on the current machine:
26+--
27+-- ./benchmarks/PingTCP server 8080
28+--
29+-- Next, perform the benchmark on a client using the server address, where
30+-- each mark is 1000 pings:
31+--
32+-- ./benchmarks/PingTCP client 0.0.0.0 8080 1000
33+--
34+-- The server must be restarted between benchmarks.
35+main :: IO ()
36+main = do
37+ args <- getArgs
38+ case args of
39+ "server" : service : [] -> withSocketsDo $ do
40+ putStrLn "server: creating TCP connection"
41+ serverAddrs <- getAddrInfo
42+ (Just (defaultHints { addrFlags = [AI_PASSIVE] } ))
43+ Nothing
44+ (Just service)
45+ let serverAddr = head serverAddrs
46+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
47+ setSocketOption sock ReuseAddr 1
48+ bindSocket sock (addrAddress serverAddr)
49+
50+ putStrLn "server: awaiting client connection"
51+ listen sock 1
52+ (clientSock, clientAddr) <- accept sock
53+
54+ putStrLn "server: listening for pings"
55+ forever (pong clientSock)
56+
57+ "client": host : service : pingsStr : args' -> withSocketsDo $ do
58+ let pings = read pingsStr
59+ serverAddrs <- getAddrInfo
60+ Nothing
61+ (Just host)
62+ (Just service)
63+ let serverAddr = head serverAddrs
64+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
65+
66+ N.connect sock (addrAddress serverAddr)
67+
68+ -- benchmark the pings
69+ withArgs args' $ defaultMain [ benchPing sock pings ]
70+
71+-- | Each `ping` sends a single byte, and expects to receive one
72+-- back in return.
73+ping :: Socket -> IO Word8
74+ping sock = do
75+ NBS.send sock $ encode (0 :: Word8)
76+ bs <- NBS.recv sock 1
77+ return $ decode bs
78+
79+pong :: Socket -> IO ()
80+pong sock = do
81+ bs <- NBS.recv sock 1
82+ NBS.sendAll sock bs
83+ return ()
84+
85+benchPing :: Socket -> Int64 -> Benchmark
86+benchPing sock n = bench "PingTCP" $
87+ nfIO (replicateM_ (fromIntegral n) (ping sock))
88+
benchmarks/PingTransport.hsnew file+92
@@ -0,0 +1,92 @@
1+module Main where
2+
3+import Network.Transport
4+import Network.Transport.TCP (mkTransport, TCPConfig (..))
5+
6+import Control.Monad (forever, replicateM_)
7+import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
8+import Data.Binary
9+import Data.Maybe (fromJust)
10+import Data.Int
11+import System.Environment (getArgs, withArgs)
12+
13+import qualified Data.ByteString.Lazy.Char8 as BS
14+
15+-- | This performs a ping benchmark on the TCP transport. This can be
16+-- compiled using:
17+--
18+-- ghc --make -idistributed-process/src -inetwork-transport/src -O2 benchmarks/PingTransport.hs
19+--
20+-- To use the compiled binary, first set up a server:
21+--
22+-- ./benchmarks/PingTransport server 0.0.0.0 8080 sourceAddr
23+--
24+-- Once this is established, launch a client to perform the benchmark. The
25+-- following command sends 1000 pings per mark.
26+--
27+-- ./benchmarks/PingTransport client 0.0.0.0 8081 sourceAddr 1000
28+--
29+-- The server must be restarted between benchmarks.
30+main :: IO ()
31+main = do
32+ args <- getArgs
33+ case args of
34+ "server" : host : service : sourceAddrFilePath : [] -> do
35+ -- establish transport
36+ transport <- mkTransport $ TCPConfig defaultHints host service
37+
38+ -- create ping end
39+ putStrLn "server: creating ping end"
40+ (sourceAddrPing, targetEndPing) <- newConnection transport
41+ BS.writeFile sourceAddrFilePath $ serialize sourceAddrPing
42+
43+ -- create pong end
44+ putStrLn "server: creating pong end"
45+ [sourceAddrPongBS] <- receive targetEndPing
46+ sourceEndPong <- connect . fromJust $ deserialize transport sourceAddrPongBS
47+
48+ -- always respond to a ping with a pong
49+ putStrLn "server: awaiting pings"
50+ forever $ pong targetEndPing sourceEndPong
51+
52+
53+ "client" : host : service : sourceAddrFilePath : pingsStr : args' -> do
54+ let pings = read pingsStr
55+ -- establish transport
56+ transport <- mkTransport $ TCPConfig defaultHints host service
57+
58+ -- create ping end
59+ sourceAddrPingBS <- BS.readFile sourceAddrFilePath
60+ sourceEndPing <- connect . fromJust $ deserialize transport sourceAddrPingBS
61+
62+ -- create pong end
63+ (sourceAddrPong, targetEndPong) <- newConnection transport
64+ send sourceEndPing [serialize sourceAddrPong]
65+
66+ -- benchmark the pings
67+ withArgs args' $ defaultMain [ benchPing sourceEndPing targetEndPong pings]
68+
69+-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
70+-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent
71+-- in reply down the `SourceEnd`, repeating whatever was sent.
72+pong :: TargetEnd -> SourceEnd -> IO ()
73+pong targetEndPing sourceEndPong = do
74+ bs <- receive targetEndPing
75+ send sourceEndPong bs
76+
77+-- | The effect of `ping sourceEndPing targetEndPong n` is to send the number
78+-- `n` using `sourceEndPing`, and to then receive the a number from
79+-- `targetEndPong`, which is then returned.
80+ping :: SourceEnd -> TargetEnd -> IO Int64
81+ping sourceEndPing targetEndPong = do
82+ send sourceEndPing [BS.empty]
83+ [bs] <- receive targetEndPong
84+ return $ decode bs
85+
86+-- | The effect of `benchPing sourceEndPing targetEndPong n` is to send
87+-- `n` pings down `sourceEndPing` using the `ping` function. The time
88+-- taken is benchmarked.
89+benchPing :: SourceEnd -> TargetEnd -> Int64 -> Benchmark
90+benchPing sourceEndPing targetEndPong n = bench "PingTransport" $
91+ nfIO (replicateM_ (fromIntegral n) (ping sourceEndPing targetEndPong))
92+
benchmarks/SendTCP.hsnew file+91
@@ -0,0 +1,91 @@
1+module Main where
2+
3+import Control.Monad
4+import Criterion.Main (Benchmark, bench, defaultMain, nfIO, bgroup)
5+import Data.Binary
6+import Data.ByteString.Lazy (ByteString)
7+import Data.Int
8+import Network.Socket
9+ ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
10+ , SocketType (Stream), SocketOption (ReuseAddr)
11+ , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
12+ , defaultHints
13+ , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
14+import System.Environment (getArgs, withArgs)
15+
16+import qualified Data.ByteString.Lazy as BS
17+import qualified Network.Socket as N
18+import qualified Network.Socket.ByteString.Lazy as NBS
19+
20+-- | This performs a benchmark on a TCP connection to measure how long it takes
21+-- to transfer a number of bytes.
22+-- To compile this file, you might use:
23+--
24+-- ghc --make -idistributed-process/src -inetwork-transport/src -O2 benchmarks/SendTCP.hs
25+--
26+-- To use the compiled binary, first set up the server on the current machine,
27+-- such that it expects 1000 bytes:
28+--
29+-- ./benchmarks/SendTCP server 8080 1000
30+--
31+-- Next, perform the benchmark on a client using the server address, where
32+-- each mark is 1000 bytes:
33+--
34+-- ./benchmarks/SendTCP client 0.0.0.0 8080 1000
35+--
36+-- The server must be restarted between benchmarks.
37+main :: IO ()
38+main = do
39+ args <- getArgs
40+ case args of
41+ "server" : service : sizeStr : [] -> withSocketsDo $ do
42+ let size = read sizeStr
43+ putStrLn "server: creating TCP connection"
44+ serverAddrs <- getAddrInfo
45+ (Just (defaultHints { addrFlags = [AI_PASSIVE] } ))
46+ Nothing
47+ (Just service)
48+ let serverAddr = head serverAddrs
49+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
50+ setSocketOption sock ReuseAddr 1
51+ bindSocket sock (addrAddress serverAddr)
52+
53+ putStrLn "server: awaiting client connection"
54+ listen sock 1
55+ (clientSock, clientAddr) <- accept sock
56+
57+ putStrLn "server: listening for data"
58+
59+ forever $ pong clientSock size
60+
61+ "client": host : service : sizeStr : args' -> withSocketsDo $ do
62+ let size = read sizeStr
63+ serverAddrs <- getAddrInfo
64+ Nothing
65+ (Just host)
66+ (Just service)
67+ let serverAddr = head serverAddrs
68+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
69+
70+ N.connect sock (addrAddress serverAddr)
71+
72+ -- benchmark the data
73+ let bs = BS.replicate size 0
74+ withArgs args' $ defaultMain [ benchSend sock bs ]
75+
76+-- | Each `ping` sends a `ByteString` and expects a byte in return.
77+ping :: Socket -> ByteString -> IO Word8
78+ping sock bs = do
79+ NBS.send sock bs
80+ bs' <- NBS.recv sock 1
81+ return $ decode bs
82+
83+pong :: Socket -> Int64 -> IO ()
84+pong sock size = do
85+ bs <- NBS.recv sock size
86+ NBS.sendAll sock (encode (0 :: Word8))
87+ return ()
88+
89+benchSend :: Socket -> ByteString -> Benchmark
90+benchSend sock bs = bench "SendTCP" $
91+ nfIO (ping sock bs)
benchmarks/SendTransport.hsnew file+96
@@ -0,0 +1,96 @@
1+module Main where
2+
3+import Network.Transport
4+import Network.Transport.TCP (mkTransport, TCPConfig (..))
5+
6+import Control.Monad (forever, replicateM_)
7+import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
8+import Data.Binary
9+import Data.ByteString.Lazy (ByteString)
10+import Data.Maybe (fromJust)
11+import Data.Int
12+import System.Environment (getArgs, withArgs)
13+
14+import qualified Data.ByteString.Lazy as BS
15+
16+-- | This performs a benchmark on the TCP transport to measure how long
17+-- it takes to transfer a number of bytes. This can be compiled using:
18+--
19+-- ghc --make -idistributed-process/src -inetwork-transport/src -O2 benchmarks/SendTransport.hs
20+--
21+-- To use the compiled binary, first set up a server:
22+--
23+-- ./benchmarks/SendTransport server 0.0.0.0 8080 sourceAddr
24+--
25+-- Once this is established, launch a client to perform the benchmark. The
26+-- following command sends 1000 bytes per mark.
27+--
28+-- ./benchmarks/SendTransport client 0.0.0.0 8081 sourceAddr 1000
29+--
30+-- The server must be restarted between benchmarks.
31+main :: IO ()
32+main = do
33+ args <- getArgs
34+ case args of
35+ "server" : host : service : sourceAddrFilePath : [] -> do
36+
37+ -- establish transport
38+ transport <- mkTransport $ TCPConfig defaultHints host service
39+
40+ -- create ping end
41+ putStrLn "server: creating ping end"
42+ (sourceAddrPing, targetEndPing) <- newConnection transport
43+ BS.writeFile sourceAddrFilePath $ serialize sourceAddrPing
44+
45+ -- create pong end
46+ putStrLn "server: creating pong end"
47+ [sourceAddrPongBS] <- receive targetEndPing
48+ sourceEndPong <- connect . fromJust $ deserialize transport sourceAddrPongBS
49+
50+ -- always respond to a ping with a pong
51+ putStrLn "server: awaiting pings"
52+ forever $ pong targetEndPing sourceEndPong
53+
54+
55+ "client" : host : service : sourceAddrFilePath : sizeStr : args' -> do
56+ let size = read sizeStr
57+
58+ -- establish transport
59+ transport <- mkTransport $ TCPConfig defaultHints host service
60+
61+ -- create ping end
62+ sourceAddrPingBS <- BS.readFile sourceAddrFilePath
63+ sourceEndPing <- connect . fromJust $ deserialize transport sourceAddrPingBS
64+
65+ -- create pong end
66+ (sourceAddrPong, targetEndPong) <- newConnection transport
67+ send sourceEndPing [serialize sourceAddrPong]
68+
69+ -- benchmark the data
70+ let bs = BS.replicate size 0
71+ withArgs args' $ defaultMain [ benchSend sourceEndPing targetEndPong bs ]
72+
73+-- | The effect of `ping sourceEndPing targetEndPong bs` is to send the
74+-- `ByteString` `bs` using `sourceEndPing`, and to then receive
75+-- a single byte back from `targetEndPong`.
76+ping :: SourceEnd -> TargetEnd -> ByteString -> IO Word8
77+ping sourceEndPing targetEndPong bs = do
78+ send sourceEndPing [bs]
79+ [cs] <- receive targetEndPong
80+ return $ decode cs
81+
82+-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
83+-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent
84+-- in reply down the `SourceEnd`, repeating whatever was sent.
85+pong :: TargetEnd -> SourceEnd -> IO ()
86+pong targetEndPing sourceEndPong = do
87+ bs <- receive targetEndPing
88+ send sourceEndPong [encode (0 :: Word8)]
89+
90+-- | The effect of `benchSend sourceEndPing targetEndPong bs` is to send
91+-- `bs` pings down `sourceEndPing` using the `ping` function. The time
92+-- taken is benchmarked.
93+benchSend :: SourceEnd -> TargetEnd -> ByteString -> Benchmark
94+benchSend sourceEndPing targetEndPong bs = bench "SendTransport" $
95+ nfIO (ping sourceEndPing targetEndPong bs)
96+
distributed-process/src/Control/Distributed/Process.hs+2−2
@@ -27,8 +27,8 @@ module Control.Distributed.Process (
27import qualified Network.Transport as Trans
27import qualified Network.Transport as Trans
28import Network.Transport (Transport)
28import Network.Transport (Transport)
29
29
30−import qualified Data.ByteString.Char8 as BS
30+import qualified Data.ByteString.Lazy.Char8 as BS
31−import Data.ByteString.Char8 (ByteString)
31+import Data.ByteString.Lazy.Char8 (ByteString)
32import qualified Data.IntMap as IntMap
32import qualified Data.IntMap as IntMap
33import Data.IntMap (IntMap)
33import Data.IntMap (IntMap)
34import Control.Applicative
34import Control.Applicative
examples/DemoProcess.hs+3−1
@@ -3,6 +3,7 @@ module DemoProcess where
3import Control.Distributed.Process
3import Control.Distributed.Process
4import Control.Concurrent (threadDelay)
4import Control.Concurrent (threadDelay)
5
5
6+import Network.Transport (closeTransport)
6import Network.Transport.TCP (mkTransport, TCPConfig (..))
7import Network.Transport.TCP (mkTransport, TCPConfig (..))
7
8
8import Debug.Trace
9import Debug.Trace
@@ -16,8 +17,9 @@ demo1 = do
16 lpid1 <- spawnLocal (logger 1)
17 lpid1 <- spawnLocal (logger 1)
17 spawnLocal (chatty "jim1" lpid1)
18 spawnLocal (chatty "jim1" lpid1)
18 spawnLocal (chatty "bob1" lpid1)
19 spawnLocal (chatty "bob1" lpid1)
19− liftIO $ threadDelay 500000
20 return ()
20 return ()
21+ threadDelay 600000
22+ closeTransport trans
21
23
22chatty :: String -> ProcessId -> Process ()
24chatty :: String -> ProcessId -> Process ()
23chatty name target = do
25chatty name target = do
examples/DemoTCP.hs+7−1
@@ -7,7 +7,7 @@ import Control.Monad
7import Network.Socket (HostName, ServiceName)
7import Network.Socket (HostName, ServiceName)
8import System.Environment
8import System.Environment
9
9
10−import qualified Data.ByteString.Char8 as BS
10+import qualified Data.ByteString.Lazy.Char8 as BS
11
11
12import Debug.Trace
12import Debug.Trace
13
13
@@ -45,6 +45,9 @@ main = do
45 [clientMessage] <- receive masterTargetEnd
45 [clientMessage] <- receive masterTargetEnd
46 print clientMessage
46 print clientMessage
47
47
48+ closeTargetEnd masterTargetEnd
49+ putStrLn "master: close connections"
50+
48 -- Slave:
51 -- Slave:
49 -- Each slave provides its own host and service, as well as details of
52 -- Each slave provides its own host and service, as well as details of
50 -- the master it wishes to connect to.
53 -- the master it wishes to connect to.
@@ -58,6 +61,9 @@ main = do
58 send masterSourceEnd [BS.pack message]
61 send masterSourceEnd [BS.pack message]
59 putStrLn "slave: sent message to master"
62 putStrLn "slave: sent message to master"
60
63
64+ closeSourceEnd masterSourceEnd
65+ putStrLn "slave: close connection"
66+
61 _ -> error "Unexpected arguments"
67 _ -> error "Unexpected arguments"
62
68
63demoMaster :: Transport -- ^ Transport
69demoMaster :: Transport -- ^ Transport
examples/DemoTransport.hs+25−8
@@ -9,10 +9,13 @@ import Network.Transport.TCP (mkTransport, TCPConfig (..))
9import Control.Concurrent
9import Control.Concurrent
10import Control.Monad
10import Control.Monad
11
11
12+-- import qualified Data.ByteString.Lazy.Char8 as BS
12import qualified Data.ByteString.Char8 as BS
13import qualified Data.ByteString.Char8 as BS
13
14
14import Debug.Trace
15import Debug.Trace
15
16
17+closeTransport _ = return ()
18+
16-------------------------------------------
19-------------------------------------------
17-- Example program using backend directly
20-- Example program using backend directly
18--
21--
@@ -33,9 +36,7 @@ demo0 mktrans = do
33 mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
36 mapM_ (\n -> send sourceEnd [BS.pack ("hello " ++ show n)]) [1 .. 10]
34 threadDelay 100000
37 threadDelay 100000
35
38
36−-- closeSourceEnd sourceEnd
37−-- closeTransport trans
39+ closeTransport trans
38−
39
40
40-- | Check endpoint serialization and deserialization.
41-- | Check endpoint serialization and deserialization.
41demo1 :: IO Transport -> IO ()
42demo1 :: IO Transport -> IO ()
@@ -56,6 +57,7 @@ demo1 mktrans = do
56 send sourceEnd [BS.pack "hello 1"]
57 send sourceEnd [BS.pack "hello 1"]
57 send sourceEnd' [BS.pack "hello 2"]
58 send sourceEnd' [BS.pack "hello 2"]
58 threadDelay 100000
59 threadDelay 100000
60+ closeTransport trans
59
61
60-- | Check that messages can be sent before receive is set up.
62-- | Check that messages can be sent before receive is set up.
61demo2 :: IO ()
63demo2 :: IO ()
@@ -70,7 +72,9 @@ demo2 = do
70 threadDelay 100000
72 threadDelay 100000
71
73
72 forkIO $ logServer "logServer" targetEnd
74 forkIO $ logServer "logServer" targetEnd
73− return ()
75+ threadDelay 100000
76+
77+ closeTransport trans
74
78
75-- | Check that two different transports can be created.
79-- | Check that two different transports can be created.
76demo3 :: IO ()
80demo3 :: IO ()
@@ -81,8 +85,8 @@ demo3 = do
81 (sourceAddr1, targetEnd1) <- newConnection trans1
85 (sourceAddr1, targetEnd1) <- newConnection trans1
82 (sourceAddr2, targetEnd2) <- newConnection trans2
86 (sourceAddr2, targetEnd2) <- newConnection trans2
83
87
84− forkIO $ logServer "logServer1" targetEnd1
88+ threadId1 <- forkIO $ logServer "logServer1" targetEnd1
85− forkIO $ logServer "logServer2" targetEnd2
89+ threadId2 <- forkIO $ logServer "logServer2" targetEnd2
86
90
87 sourceEnd1 <- connect sourceAddr1
91 sourceEnd1 <- connect sourceAddr1
88 sourceEnd2 <- connect sourceAddr2
92 sourceEnd2 <- connect sourceAddr2
@@ -90,6 +94,13 @@ demo3 = do
90 send sourceEnd1 [BS.pack "hello1"]
94 send sourceEnd1 [BS.pack "hello1"]
91 send sourceEnd2 [BS.pack "hello2"]
95 send sourceEnd2 [BS.pack "hello2"]
92
96
97+ threadDelay 100000
98+
99+ killThread threadId1
100+ killThread threadId2
101+ closeTransport trans1
102+ closeTransport trans2
103+
93-- | Check that two different connections on the same transport can be created.
104-- | Check that two different connections on the same transport can be created.
94demo4 :: IO ()
105demo4 :: IO ()
95demo4 = do
106demo4 = do
@@ -99,8 +110,8 @@ demo4 = do
99 (sourceAddr1, targetEnd1) <- newConnection trans
110 (sourceAddr1, targetEnd1) <- newConnection trans
100 (sourceAddr2, targetEnd2) <- newConnection trans
111 (sourceAddr2, targetEnd2) <- newConnection trans
101
112
102− forkIO $ logServer "logServer1" targetEnd1
113+ threadId1 <- forkIO $ logServer "logServer1" targetEnd1
103− forkIO $ logServer "logServer2" targetEnd2
114+ threadId2 <- forkIO $ logServer "logServer2" targetEnd2
104
115
105 sourceEnd1 <- connect sourceAddr1
116 sourceEnd1 <- connect sourceAddr1
106 sourceEnd2 <- connect sourceAddr2
117 sourceEnd2 <- connect sourceAddr2
@@ -108,6 +119,12 @@ demo4 = do
108 send sourceEnd1 [BS.pack "hello1"]
119 send sourceEnd1 [BS.pack "hello1"]
109 send sourceEnd2 [BS.pack "hello2"]
120 send sourceEnd2 [BS.pack "hello2"]
110
121
122+ threadDelay 100000
123+
124+ killThread threadId1
125+ killThread threadId2
126+ closeTransport trans
127+
111logServer :: String -> TargetEnd -> IO ()
128logServer :: String -> TargetEnd -> IO ()
112logServer name targetEnd = forever $ do
129logServer name targetEnd = forever $ do
113 x <- receive targetEnd
130 x <- receive targetEnd
network-transport/network-transport.cabal+1
@@ -11,6 +11,7 @@ Cabal-Version: >=1.2
11
11
12Library
12Library
13 Build-Depends: base >= 3 && < 5,
13 Build-Depends: base >= 3 && < 5,
14+ binary >= 0.5,
14 bytestring >= 0.9,
15 bytestring >= 0.9,
15 containers >= 0.4,
16 containers >= 0.4,
16 network >= 2.3,
17 network >= 2.3,
network-transport/src/Network/Transport.hs+10−3
@@ -12,7 +12,7 @@ module Network.Transport
12 , newMulticast
12 , newMulticast
13 ) where
13 ) where
14
14
15−import Data.ByteString.Char8 (ByteString)
15+import Data.ByteString.Lazy.Char8 (ByteString)
16
16
17-- | The `Hints` and `SourceHints` 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
@@ -33,6 +33,7 @@ data Transport = Transport
33 { newConnectionWith :: Hints -> IO (SourceAddr, TargetEnd)
33 { newConnectionWith :: Hints -> IO (SourceAddr, TargetEnd)
34 , newMulticastWith :: Hints -> IO (MulticastSourceEnd, MulticastTargetAddr)
34 , newMulticastWith :: Hints -> IO (MulticastSourceEnd, MulticastTargetAddr)
35 , deserialize :: ByteString -> Maybe SourceAddr
35 , deserialize :: ByteString -> Maybe SourceAddr
36+ , closeTransport :: IO ()
36 }
37 }
37
38
38-- | This is a convenience function that creates a new connection on a transport
39-- | This is a convenience function that creates a new connection on a transport
@@ -68,14 +69,20 @@ defaultSourceHints = SourceHints
68
69
69-- | A `SourceEnd` provides a `send` function that allows vectored messages
70-- | A `SourceEnd` provides a `send` function that allows vectored messages
70-- to be sent to the corresponding `TargetEnd`.
71-- to be sent to the corresponding `TargetEnd`.
72+-- The `close` function closes the connection between this source and the target
73+-- end. Connections between other sources the target end remain unaffected
71−newtype SourceEnd = SourceEnd
74+data SourceEnd = SourceEnd
72 { send :: [ByteString] -> IO ()
75 { send :: [ByteString] -> IO ()
76+ , closeSourceEnd :: IO ()
73 }
77 }
74
78
75-- | A `TargetEnd` provides a `receive` function that allows messages
79-- | A `TargetEnd` provides a `receive` function that allows messages
76-- to be received from the corresponding `SourceEnd`s.
80-- to be received from the corresponding `SourceEnd`s.
81+-- The `closeTargetEnd` function closes all connections to this target,
82+-- and all new connections will be refused.
77−newtype TargetEnd = TargetEnd
83+data TargetEnd = TargetEnd
78 { receive :: IO [ByteString]
84 { receive :: IO [ByteString]
85+ , closeTargetEnd :: IO ()
79 }
86 }
80
87
81newtype MulticastSourceEnd = MulticastSourceEnd
88newtype MulticastSourceEnd = MulticastSourceEnd
network-transport/src/Network/Transport/MVar.hs+2−2
@@ -6,10 +6,10 @@ module Network.Transport.MVar
6
6
7import Control.Concurrent.MVar
7import Control.Concurrent.MVar
8import Data.IntMap (IntMap)
8import Data.IntMap (IntMap)
9−import Data.ByteString.Char8 (ByteString)
9+import Data.ByteString.Lazy.Char8 (ByteString)
10
10
11import qualified Data.IntMap as IntMap
11import qualified Data.IntMap as IntMap
12−import qualified Data.ByteString.Char8 as BS
12+import qualified Data.ByteString.Lazy.Char8 as BS
13
13
14import Network.Transport
14import Network.Transport
15
15
network-transport/src/Network/Transport/Pipes.hs+8−6
@@ -10,13 +10,14 @@ import Control.Concurrent (threadDelay, forkOS)
10import Data.IntMap (IntMap)
10import Data.IntMap (IntMap)
11-- import Data.ByteString.Char8 (ByteString)
11-- import Data.ByteString.Char8 (ByteString)
12import Data.Word
12import Data.Word
13+import Data.Int
13import Data.IORef
14import Data.IORef
14import qualified Data.IntMap as IntMap
15import qualified Data.IntMap as IntMap
15−import qualified Data.ByteString.Char8 as BS
16+-- import qualified Data.ByteString.Char8 as BS
16−-- import qualified Data.ByteString.Lazy.Char8 as BS
17+import qualified Data.ByteString.Lazy.Char8 as BS
17
18
18−-- import Data.Binary (decode)
19+import Data.Binary (encode,decode)
19−import Data.Serialize (encode,decode)
20+-- import Data.Serialize (encode,decode)
20import Data.List (foldl')
21import Data.List (foldl')
21import Network.Transport
22import Network.Transport
22import System.Random (randomIO)
23import System.Random (randomIO)
@@ -116,11 +117,12 @@ mkTransport = do
116
117
117 mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
118 mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
118 fd <- takeMVar mv
119 fd <- takeMVar mv
120+ let oneread :: Int64 -> IO (BS.ByteString, Int64)
119− let oneread n = do bs <- readit fd (fromIntegral n)
121+ oneread n = do bs <- readit fd (fromIntegral n)
120 return (bs, BS.length bs)
122 return (bs, BS.length bs)
121 dbgprint2$ " Attempt read header..."
123 dbgprint2$ " Attempt read header..."
122
124
123− let spinread :: Int -> IO BS.ByteString
125+ let spinread :: Int64 -> IO BS.ByteString
124 spinread desired = do
126 spinread desired = do
125-- hPutStr stderr "."
127-- hPutStr stderr "."
126 (bytes,len) <- oneread desired
128 (bytes,len) <- oneread desired
network-transport/src/Network/Transport/TCP.hs+162−62
@@ -5,23 +5,29 @@ module Network.Transport.TCP
5
5
6import Network.Transport
6import Network.Transport
7
7
8−import Control.Concurrent (forkIO)
8+import Control.Concurrent (forkIO, ThreadId, killThread)
9import Control.Concurrent.Chan
9import Control.Concurrent.Chan
10import Control.Concurrent.MVar
10import Control.Concurrent.MVar
11−import Control.Monad (forever, unless)
11+import Control.Monad (forever, forM_)
12−import Data.ByteString.Char8 (ByteString)
12+import Data.ByteString.Lazy.Char8 (ByteString)
13import Data.IntMap (IntMap)
13import Data.IntMap (IntMap)
14+import Data.Int
14import Data.Word
15import Data.Word
15import Network.Socket
16import Network.Socket
17+ ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
18+ , SocketType (Stream), SocketOption (ReuseAddr)
19+ , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
20+ , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
16import Safe
21import Safe
17
22
18−import qualified Data.ByteString.Char8 as BS
23+import qualified Data.Binary as B
24+import qualified Data.ByteString.Lazy.Char8 as BS
19import qualified Data.IntMap as IntMap
25import qualified Data.IntMap as IntMap
20import qualified Network.Socket as N
26import qualified Network.Socket as N
21−import qualified Network.Socket.ByteString as NBS
27+import qualified Network.Socket.ByteString.Lazy as NBS
22
28
23−type Chans = MVar (Int, IntMap (Chan ByteString))
24type ChanId = Int
29type ChanId = Int
30+type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
25
31
26-- | This deals with several different configuration properties:
32-- | This deals with several different configuration properties:
27-- * Buffer size, specified in Hints
33-- * Buffer size, specified in Hints
@@ -37,7 +43,7 @@ data TCPConfig = TCPConfig Hints HostName ServiceName
37-- Messages are all queued using an unbounded Chan.
43-- Messages are all queued using an unbounded Chan.
38mkTransport :: TCPConfig -> IO Transport
44mkTransport :: TCPConfig -> IO Transport
39mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
45mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
40− channels <- newMVar (0, IntMap.empty)
46+ chans <- newMVar (0, IntMap.empty)
41 serverAddrs <- getAddrInfo
47 serverAddrs <- getAddrInfo
42 (Just (N.defaultHints { addrFlags = [AI_PASSIVE] }))
48 (Just (N.defaultHints { addrFlags = [AI_PASSIVE] }))
43 Nothing
49 Nothing
@@ -46,69 +52,163 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
46 [] -> error "mkTransport: getAddrInfo returned []"
52 [] -> error "mkTransport: getAddrInfo returned []"
47 as -> head as
53 as -> head as
48 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
54 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
55+ setSocketOption sock ReuseAddr 1
49 bindSocket sock (addrAddress serverAddr)
56 bindSocket sock (addrAddress serverAddr)
50 listen sock 5
57 listen sock 5
51− forkIO $ procConnections channels sock
58+ threadId <- forkIO $ procConnections chans sock
52
59
53 return Transport
60 return Transport
54− { newConnectionWith = \_ -> do
61+ { newConnectionWith = {-# SCC "newConnectionWith" #-}\_ -> do
55− (chanId, chanMap) <- takeMVar channels
62+ (chanId, chanMap) <- takeMVar chans
56 chan <- newChan
63 chan <- newChan
57− putMVar channels (chanId + 1, IntMap.insert chanId chan chanMap)
64+ putMVar chans (chanId + 1, IntMap.insert chanId (chan, []) chanMap)
58− return (mkSourceAddr host service chanId, mkTargetEnd chan)
65+ return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
59− , newMulticastWith = undefined
66+ , newMulticastWith = error "newMulticastWith: not defined"
60− , deserialize = \bs ->
67+ , deserialize = {-# SCC "deserialize" #-} \bs ->
61− case readMay . BS.unpack $ bs of
68+ let (host, service, chanId) = B.decode bs in
62− Nothing -> error "deserialize: cannot parse"
63− Just (host, service, chanId) -> Just $ mkSourceAddr host service chanId
69+ Just $ mkSourceAddr host service chanId
70+ , closeTransport = {-# SCC "closeTransport" #-} do
71+ -- Kill the transport channel process
72+ killThread threadId
73+ sClose sock
74+ -- Kill all target end processes
75+ (chanId, chanMap) <- takeMVar chans
76+ forM_ (IntMap.elems chanMap) (\(chan, socks) ->
77+ forM_ socks (\(threadId', sock') -> do
78+ killThread threadId'
79+ sClose sock'))
64 }
80 }
65
81
66− where
67− mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
82+mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
68− mkSourceAddr host service chanId = SourceAddr
83+mkSourceAddr host service chanId = SourceAddr
69− { connectWith = \_ -> mkSourceEnd host service chanId
84+ { connectWith = {-# SCC "connectWith" #-} \_ -> mkSourceEnd host service chanId
70− , serialize = BS.pack . show $ (host, service, chanId)
85+ , serialize = {-# SCC "serialize" #-} B.encode (host, service, chanId)
71− }
86+ }
72
87
73− mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
88+mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
74− mkSourceEnd host service chanId = withSocketsDo $ do
89+mkSourceEnd host service chanId = withSocketsDo $ do
75− serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
90+ serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
76− let serverAddr = case serverAddrs of
91+ let serverAddr = case serverAddrs of
77− [] -> error "mkSourceEnd: getAddrInfo returned []"
92+ [] -> error "mkSourceEnd: getAddrInfo returned []"
78− as -> head as
93+ as -> head as
79− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
94+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
80− setSocketOption sock ReuseAddr 1
95+ setSocketOption sock ReuseAddr 1
81− N.connect sock (addrAddress serverAddr)
96+ N.connect sock (addrAddress serverAddr)
82− NBS.sendAll sock $ BS.pack . show $ chanId
97+ NBS.sendAll sock $ B.encode (fromIntegral chanId :: Int64)
83− return $ SourceEnd
98+ return SourceEnd
84− { Network.Transport.send = \bss -> NBS.sendMany sock bss
99+ { send = {-# SCC "send" #-} \bss -> do
100+ let size = fromIntegral (sum . map BS.length $ bss) :: Int64
101+ if size < 255
102+ then
103+ NBS.sendAll sock (B.encode (fromIntegral size :: Word8))
104+ else do
105+ NBS.sendAll sock (B.encode (255 :: Word8))
106+ NBS.sendAll sock (B.encode size)
107+ mapM_ (NBS.sendAll sock) bss
108+ , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
109+ }
110+
111+mkTargetEnd :: Chans -> ChanId -> Chan [ByteString] -> TargetEnd
112+mkTargetEnd chans chanId chan = TargetEnd
113+ { -- for now we will implement this as a Chan
114+ receive = {-# SCC "receive" #-} readChan chan
115+ , closeTargetEnd = {-# SCC "closeTargetEnd" #-} do
116+ (chanId', chanMap) <- takeMVar chans
117+ case IntMap.lookup chanId chanMap of
118+ Nothing -> putMVar chans (chanId', chanMap)
119+ Just (_, socks) -> do
120+ forM_ socks $ \(threadId, sock) -> do
121+ killThread threadId
122+ sClose sock
123+ let chanMap' = IntMap.delete chanId chanMap
124+ putMVar chans (chanId', chanMap')
85− }
125+ }
126+
127+-- | This function waits for inbound connections. If a connection fails
128+-- for some reason, an error is raised.
129+procConnections :: Chans -> Socket -> IO ()
130+procConnections chans sock = forever $ do
131+ (clientSock, _clientAddr) <- accept sock
132+ -- decode the first message to find the correct chanId
133+ bss <- recvExact clientSock 8
134+ case bss of
135+ [] -> error "procConnections: inbound chanId aborted"
136+ bss -> do
137+ let bs = BS.concat bss
138+ let chanId = fromIntegral (B.decode bs :: Int64)
139+ (chanId', chanMap) <- takeMVar chans
140+ case IntMap.lookup chanId chanMap of
141+ Nothing -> do
142+ putMVar chans (chanId', chanMap)
143+ error "procConnections: cannot find chanId"
144+ Just (chan, socks) -> do
145+ threadId <- forkIO $ procMessages chans chanId chan clientSock
146+ let chanMap' = IntMap.insert chanId (chan, (threadId, clientSock):socks) chanMap
147+ putMVar chans (chanId', chanMap')
86
148
87− mkTargetEnd :: Chan ByteString -> TargetEnd
149+-- | This function first extracts a header of type Word8, which determines
88− mkTargetEnd chan = TargetEnd
150+-- the size of the ByteString that follows. If this size is 0, this indicates
89− { -- for now we will implement this as a Chan
151+-- that the ByteString is large, so the next value is an Int64, which
90− receive = do
152+-- determines the size of the ByteString that follows. The ByteString is then
91− bs <- readChan chan
153+-- extracted from the socket, and then written to the Chan only when
92− return [bs]
154+-- complete. If either of the first header size is null this indicates the
93− }
155+-- socket has closed.
156+procMessages :: Chans -> ChanId -> Chan [ByteString] -> Socket -> IO ()
157+procMessages chans chanId chan sock = do
158+ sizeBSs <- recvExact sock 1
159+ case sizeBSs of
160+ [] -> closeSocket
161+ _ -> do
162+ let size = fromIntegral (B.decode . BS.concat $ sizeBSs :: Word8)
163+ if size == 255
164+ then do
165+ sizeBSs' <- recvExact sock 8
166+ case sizeBSs' of
167+ [] -> closeSocket
168+ _ -> procMessage (B.decode . BS.concat $ sizeBSs' :: Int64)
169+ else procMessage size
170+ where
171+ closeSocket :: IO ()
172+ closeSocket = do
173+ (chanId', chanMap) <- takeMVar chans
174+ case IntMap.lookup chanId chanMap of
175+ Nothing -> do
176+ putMVar chans (chanId', chanMap)
177+ error "procMessages: chanId not found."
178+ Just (chan, socks) -> do
179+ let socks' = filter ((/= sock) . snd) socks
180+ let chanMap' = IntMap.insert chanId (chan, socks') chanMap
181+ putMVar chans (chanId', chanMap')
182+ sClose sock
183+ procMessage :: Int64 -> IO ()
184+ procMessage size = do
185+ bss <- recvExact sock size
186+ case bss of
187+ [] -> closeSocket
188+ _ -> do
189+ writeChan chan bss
190+ procMessages chans chanId chan sock
94
191
95− procConnections :: Chans -> Socket -> IO ()
192+-- | The result of `recvExact sock n` is a `[ByteString]` whose concatenation
96− procConnections chans sock = forever $ do
193+-- is of length `n`, received from `sock`. No more bytes than necessary are
97− (clientSock, _clientAddr) <- accept sock
194+-- read from the socket.
98− -- decode the first message to find the correct chanId
195+-- NB: This uses Network.Socket.ByteString.recv, which may *discard*
196+-- superfluous input depending on the socket type. Also note that
197+-- if `recv` returns an empty `ByteString` then this means that the socket
198+-- was closed: in this case, we return an empty list.
199+-- NB: It may be appropriate to change the return type to
200+-- IO (Either ByteString ByteString), where `return Left bs` indicates
201+-- a partial retrieval since the socket was closed, and `return Right bs`
202+-- indicates success. This hasn't been implemented, since a Transport `receive`
203+-- represents an atomic receipt of a message.
204+recvExact :: Socket -> Int64 -> IO [ByteString]
205+recvExact sock n = go [] sock n
206+ where
207+ go :: [ByteString] -> Socket -> Int64 -> IO [ByteString]
208+ go bss _ 0 = return (reverse bss)
209+ go bss sock n = do
99− bs <- NBS.recv clientSock 4096
210+ bs <- NBS.recv sock n
100− case BS.readInt bs of
101− Nothing -> error "procConnections: cannot parse chanId"
102− Just (chanId, bs') -> do
103− -- lookup the channel
104− (_, chanMap) <- readMVar chans
105− case IntMap.lookup chanId chanMap of
106− Nothing -> error "procConnections: cannot find chanId"
107− Just chan -> forkIO $ do
108− unless (BS.null bs') $ writeChan chan bs'
211+ if BS.null bs
109− procMessages chan clientSock
212+ then return []
213+ else go (bs:bss) sock (n - BS.length bs)
110
214
111− procMessages :: Chan ByteString -> Socket -> IO ()
112− procMessages chan sock = forever $ do
113− bs <- NBS.recv sock 4096
114− writeChan chan bs