git.iko.soy/distributed-process
on master help sign in
Add data sending benchmark.
fb33053Nicolas Wu
benchmarks/SendTCP.hsnew file+91
compare with master
unifiedsplit
@@ -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+ let bs = BS.replicate size 0
58+
59+ -- establish transport
60+ transport <- mkTransport $ TCPConfig defaultHints host service
61+
62+ -- create ping end
63+ sourceAddrPingBS <- BS.readFile sourceAddrFilePath
64+ sourceEndPing <- connect . fromJust $ deserialize transport sourceAddrPingBS
65+
66+ -- create pong end
67+ (sourceAddrPong, targetEndPong) <- newConnection transport
68+ send sourceEndPing [serialize sourceAddrPong]
69+
70+ -- benchmark the pings
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 "Transport Ping" $
95+ nfIO (ping sourceEndPing targetEndPong bs)
96+
network-transport/src/Network/Transport/TCP.hs+3−3
@@ -26,8 +26,6 @@ import qualified Data.IntMap as IntMap
26import qualified Network.Socket as N
26import qualified Network.Socket as N
27import qualified Network.Socket.ByteString.Lazy as NBS
27import qualified Network.Socket.ByteString.Lazy as NBS
28
28
29−import Debug.Trace
30−
31type ChanId = Int
29type ChanId = Int
32type Chans = MVar (ChanId, IntMap (Chan ByteString, [(ThreadId, Socket)]))
30type Chans = MVar (ChanId, IntMap (Chan ByteString, [(ThreadId, Socket)]))
33
31
@@ -168,7 +166,9 @@ procMessages chans chanId chan sock = do
168 mSizeBS' <- recvExact sock 8
166 mSizeBS' <- recvExact sock 8
169 case mSizeBS' of
167 case mSizeBS' of
170 Nothing -> closeSocket
168 Nothing -> closeSocket
169+ Just sizeBS' -> do
171− Just sizeBS' -> procMessage (B.decode sizeBS' :: Int64)
170+ let size' = B.decode sizeBS' :: Int64
171+ procMessage size'
172 else procMessage size
172 else procMessage size
173 where
173 where
174 closeSocket :: IO ()
174 closeSocket :: IO ()