git.iko.soy/distributed-process
on master help sign in
Compare one socket pair with two
a366d5eEdsko de Vries
benchmarks/Indirection.gnuplot+2
compare with master
unifiedsplit
@@ -3,10 +3,12 @@ set yrange [0:150]
3plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
3plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
4 "JustPingThroughChan.data" smooth bezier with lines title "JustPingThroughChan", \
4 "JustPingThroughChan.data" smooth bezier with lines title "JustPingThroughChan", \
5 "JustPingThroughMVar.data" smooth bezier with lines title "JustPingThroughMVar", \
5 "JustPingThroughMVar.data" smooth bezier with lines title "JustPingThroughMVar", \
6+ "JustPingTwoSocketPairs.data" smooth bezier with lines title "JustPingTwoSocketPairs", \
6 "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"
7 "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"
7set terminal postscript color
8set terminal postscript color
8set output "Indirection.ps"
9set output "Indirection.ps"
9plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
10plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
10 "JustPingThroughChan.data" smooth bezier with lines title "JustPingThroughChan", \
11 "JustPingThroughChan.data" smooth bezier with lines title "JustPingThroughChan", \
11 "JustPingThroughMVar.data" smooth bezier with lines title "JustPingThroughMVar", \
12 "JustPingThroughMVar.data" smooth bezier with lines title "JustPingThroughMVar", \
13+ "JustPingTwoSocketPairs.data" smooth bezier with lines title "JustPingTwoSocketPairs", \
12 "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"
14 "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"
benchmarks/JustPingThroughChan.hs+33−45
@@ -6,7 +6,7 @@ import Control.Monad
6
6
7import Data.Int
7import Data.Int
8import Network.Socket
8import Network.Socket
9− ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
9+ ( AddrInfo, AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
10 , SocketType (Stream), SocketOption (ReuseAddr)
10 , SocketType (Stream), SocketOption (ReuseAddr)
11 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
11 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
12 , defaultHints
12 , defaultHints
@@ -32,7 +32,9 @@ import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
32foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
32foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
33foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
33foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
34
34
35−main :: IO ()
35+passive :: Maybe AddrInfo
36+passive = Just (defaultHints { addrFlags = [AI_PASSIVE] })
37+
36main = do
38main = do
37 [pingsStr] <- getArgs
39 [pingsStr] <- getArgs
38 serverReady <- newEmptyMVar
40 serverReady <- newEmptyMVar
@@ -40,50 +42,44 @@ main = do
40
42
41 -- Start the server
43 -- Start the server
42 forkIO $ do
44 forkIO $ do
43− putStrLn "server: creating TCP connection"
45+ -- Initialize the server
44− serverAddrs <- getAddrInfo
46+ serverAddr:_ <- getAddrInfo passive Nothing (Just "8080")
45− (Just (defaultHints { addrFlags = [AI_PASSIVE] } ))
46− Nothing
47− (Just "8080")
48− let serverAddr = head serverAddrs
49− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
47+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
50 setSocketOption sock ReuseAddr 1
48 setSocketOption sock ReuseAddr 1
51 bindSocket sock (addrAddress serverAddr)
49 bindSocket sock (addrAddress serverAddr)
52−
53− putStrLn "server: awaiting client connection"
54− putMVar serverReady ()
55 listen sock 1
50 listen sock 1
56− (clientSock, clientAddr) <- accept sock
57
51
58− -- Set up channel that we will listen on and echo to the client
52+ -- Set up multiplexing channel
59− chan <- newChan
53+ multiplexChannel <- newChan
60− forkIO $ forever $ do
61− msg <- readChan chan
62− send clientSock msg
63
54
64− -- Read from the client and output on the channel
55+ -- Wait for incoming connections (pings from the client)
65− putStrLn "server: listening for pings"
56+ putMVar serverReady ()
66− pong clientSock chan
57+ (clientSock, pingAddr) <- accept sock
58+ forkIO $ socketToChan clientSock multiplexChannel
59+
60+ -- Reply to the client
61+ forever $ readChan multiplexChannel >>= send clientSock
67
62
68 -- Start the client
63 -- Start the client
69 forkIO $ do
64 forkIO $ do
70 takeMVar serverReady
65 takeMVar serverReady
71− let pings = read pingsStr
66+ serverAddr:_ <- getAddrInfo Nothing (Just "127.0.0.1") (Just "8080")
72− serverAddrs <- getAddrInfo
73− Nothing
74− (Just "127.0.0.1")
75− (Just "8080")
76− let serverAddr = head serverAddrs
77− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
67+ clientSock <- socket (addrFamily serverAddr) Stream defaultProtocol
78−
79− N.connect sock (addrAddress serverAddr)
68+ N.connect clientSock (addrAddress serverAddr)
80−
69+ ping clientSock (read pingsStr)
81− ping sock pings
82 putMVar clientDone ()
70 putMVar clientDone ()
83
71
84 -- Wait for the client to finish
72 -- Wait for the client to finish
85 takeMVar clientDone
73 takeMVar clientDone
86
74
75+socketToChan :: Socket -> Chan ByteString -> IO ()
76+socketToChan sock chan = go
77+ where
78+ go = do bs <- recv sock
79+ when (BS.length bs > 0) $ do
80+ writeChan chan bs
81+ go
82+
87pingMessage :: ByteString
83pingMessage :: ByteString
88pingMessage = pack "ping123"
84pingMessage = pack "ping123"
89
85
@@ -96,29 +92,21 @@ ping sock pings = go pings
96 go !i = do
92 go !i = do
97 before <- getCurrentTime
93 before <- getCurrentTime
98 send sock pingMessage
94 send sock pingMessage
99− bs <- recv sock 8
95+ bs <- recv sock
100 after <- getCurrentTime
96 after <- getCurrentTime
101 -- putStrLn $ "client received " ++ unpack bs
97 -- putStrLn $ "client received " ++ unpack bs
102 let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
98 let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
103 hPutStrLn stderr $ show i ++ " " ++ show latency
99 hPutStrLn stderr $ show i ++ " " ++ show latency
104 go (i - 1)
100 go (i - 1)
105
101
106−pong :: Socket -> Chan ByteString -> IO ()
102+-- | Receive a package
107−pong sock chan = do
108− bs <- recv sock 8
109− -- putStrLn $ "server received " ++ unpack bs
110− when (BS.length bs > 0) $ do
111− writeChan chan bs
112− pong sock chan
113−
114−-- | Wrapper around NBS.recv (for profiling)
115−recv :: Socket -> Int -> IO ByteString
103+recv :: Socket -> IO ByteString
116−recv sock _ = do
104+recv sock = do
117 header <- NBS.recv sock 4
105 header <- NBS.recv sock 4
118 length <- decodeLength header
106 length <- decodeLength header
119 NBS.recv sock (fromIntegral (length :: Int32))
107 NBS.recv sock (fromIntegral (length :: Int32))
120
108
121−-- | Wrapper around NBS.send (for profiling)
109+-- | Send a package
122send :: Socket -> ByteString -> IO ()
110send :: Socket -> ByteString -> IO ()
123send sock bs = do
111send sock bs = do
124 length <- encodeLength (fromIntegral (BS.length bs))
112 length <- encodeLength (fromIntegral (BS.length bs))
benchmarks/JustPingThroughMVar.hs+34−45
@@ -6,7 +6,7 @@ import Control.Monad
6
6
7import Data.Int
7import Data.Int
8import Network.Socket
8import Network.Socket
9− ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
9+ ( AddrInfo, AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
10 , SocketType (Stream), SocketOption (ReuseAddr)
10 , SocketType (Stream), SocketOption (ReuseAddr)
11 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
11 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
12 , defaultHints
12 , defaultHints
@@ -27,11 +27,14 @@ import Data.ByteString.Internal as BSI
27import Foreign.Storable (pokeByteOff, peekByteOff)
27import Foreign.Storable (pokeByteOff, peekByteOff)
28import Foreign.C (CInt(..))
28import Foreign.C (CInt(..))
29import Foreign.ForeignPtr (withForeignPtr)
29import Foreign.ForeignPtr (withForeignPtr)
30+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
30
31
31foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
32foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
32foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
33foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
33
34
34−main :: IO ()
35+passive :: Maybe AddrInfo
36+passive = Just (defaultHints { addrFlags = [AI_PASSIVE] })
37+
35main = do
38main = do
36 [pingsStr] <- getArgs
39 [pingsStr] <- getArgs
37 serverReady <- newEmptyMVar
40 serverReady <- newEmptyMVar
@@ -39,50 +42,44 @@ main = do
39
42
40 -- Start the server
43 -- Start the server
41 forkIO $ do
44 forkIO $ do
42− putStrLn "server: creating TCP connection"
45+ -- Initialize the server
43− serverAddrs <- getAddrInfo
46+ serverAddr:_ <- getAddrInfo passive Nothing (Just "8080")
44− (Just (defaultHints { addrFlags = [AI_PASSIVE] } ))
45− Nothing
46− (Just "8080")
47− let serverAddr = head serverAddrs
48− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
47+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
49 setSocketOption sock ReuseAddr 1
48 setSocketOption sock ReuseAddr 1
50 bindSocket sock (addrAddress serverAddr)
49 bindSocket sock (addrAddress serverAddr)
51−
52− putStrLn "server: awaiting client connection"
53− putMVar serverReady ()
54 listen sock 1
50 listen sock 1
55− (clientSock, clientAddr) <- accept sock
56
51
57− -- Set up an mvar that we will listen on and echo to the client
52+ -- Set up multiplexing channel
58− mvar <- newEmptyMVar
53+ multiplexMVar <- newEmptyMVar
59− forkIO $ forever $ do
60− msg <- takeMVar mvar
61− send clientSock msg
62
54
63− -- Read from the client and output on the channel
55+ -- Wait for incoming connections (pings from the client)
64− putStrLn "server: listening for pings"
56+ putMVar serverReady ()
65− pong clientSock mvar
57+ (clientSock, pingAddr) <- accept sock
58+ forkIO $ socketToMVar clientSock multiplexMVar
59+
60+ -- Reply to the client
61+ forever $ takeMVar multiplexMVar >>= send clientSock
66
62
67 -- Start the client
63 -- Start the client
68 forkIO $ do
64 forkIO $ do
69 takeMVar serverReady
65 takeMVar serverReady
70− let pings = read pingsStr
66+ serverAddr:_ <- getAddrInfo Nothing (Just "127.0.0.1") (Just "8080")
71− serverAddrs <- getAddrInfo
72− Nothing
73− (Just "127.0.0.1")
74− (Just "8080")
75− let serverAddr = head serverAddrs
76− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
67+ clientSock <- socket (addrFamily serverAddr) Stream defaultProtocol
77−
78− N.connect sock (addrAddress serverAddr)
68+ N.connect clientSock (addrAddress serverAddr)
79−
69+ ping clientSock (read pingsStr)
80− ping sock pings
81 putMVar clientDone ()
70 putMVar clientDone ()
82
71
83 -- Wait for the client to finish
72 -- Wait for the client to finish
84 takeMVar clientDone
73 takeMVar clientDone
85
74
75+socketToMVar :: Socket -> MVar ByteString -> IO ()
76+socketToMVar sock mvar = go
77+ where
78+ go = do bs <- recv sock
79+ when (BS.length bs > 0) $ do
80+ putMVar mvar bs
81+ go
82+
86pingMessage :: ByteString
83pingMessage :: ByteString
87pingMessage = pack "ping123"
84pingMessage = pack "ping123"
88
85
@@ -95,29 +92,21 @@ ping sock pings = go pings
95 go !i = do
92 go !i = do
96 before <- getCurrentTime
93 before <- getCurrentTime
97 send sock pingMessage
94 send sock pingMessage
98− bs <- recv sock 8
95+ bs <- recv sock
99 after <- getCurrentTime
96 after <- getCurrentTime
100 -- putStrLn $ "client received " ++ unpack bs
97 -- putStrLn $ "client received " ++ unpack bs
101 let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
98 let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
102 hPutStrLn stderr $ show i ++ " " ++ show latency
99 hPutStrLn stderr $ show i ++ " " ++ show latency
103 go (i - 1)
100 go (i - 1)
104
101
105−pong :: Socket -> MVar ByteString -> IO ()
102+-- | Receive a package
106−pong sock mvar = do
107− bs <- recv sock 8
108− -- putStrLn $ "server received " ++ unpack bs
109− when (BS.length bs > 0) $ do
110− putMVar mvar bs
111− pong sock mvar
112−
113−-- | Wrapper around NBS.recv (for profiling)
114−recv :: Socket -> Int -> IO ByteString
103+recv :: Socket -> IO ByteString
115−recv sock _ = do
104+recv sock = do
116 header <- NBS.recv sock 4
105 header <- NBS.recv sock 4
117 length <- decodeLength header
106 length <- decodeLength header
118 NBS.recv sock (fromIntegral (length :: Int32))
107 NBS.recv sock (fromIntegral (length :: Int32))
119
108
120−-- | Wrapper around NBS.send (for profiling)
109+-- | Send a package
121send :: Socket -> ByteString -> IO ()
110send :: Socket -> ByteString -> IO ()
122send sock bs = do
111send sock bs = do
123 length <- encodeLength (fromIntegral (BS.length bs))
112 length <- encodeLength (fromIntegral (BS.length bs))
benchmarks/JustPingTwoSocketPairs.hsnew file+149
@@ -0,0 +1,149 @@
1+{-# LANGUAGE CPP, BangPatterns #-}
2+
3+module Main where
4+
5+import Control.Monad
6+
7+import Data.Int
8+import Network.Socket
9+ ( AddrInfo, 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+import Data.Time (getCurrentTime, diffUTCTime, NominalDiffTime)
16+import System.IO (withFile, IOMode(..), hPutStrLn, Handle, stderr)
17+import Control.Concurrent (forkIO)
18+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
19+import qualified Network.Socket as N
20+import Debug.Trace
21+import Data.ByteString (ByteString)
22+import Data.ByteString.Char8 (pack, unpack)
23+import qualified Data.ByteString as BS
24+import qualified Network.Socket.ByteString as NBS
25+import Data.Time (getCurrentTime, diffUTCTime, NominalDiffTime)
26+import Data.ByteString.Internal as BSI
27+import Foreign.Storable (pokeByteOff, peekByteOff)
28+import Foreign.C (CInt(..))
29+import Foreign.ForeignPtr (withForeignPtr)
30+import Control.Concurrent.Chan (Chan, newChan, readChan, writeChan)
31+
32+foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
33+foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
34+
35+passive :: Maybe AddrInfo
36+passive = Just (defaultHints { addrFlags = [AI_PASSIVE] })
37+
38+main = do
39+ [pingsStr] <- getArgs
40+ serverReady <- newEmptyMVar
41+ clientReady <- newEmptyMVar
42+ clientDone <- newEmptyMVar
43+
44+ -- Start the server
45+ forkIO $ do
46+ -- Initialize the server
47+ serverAddr:_ <- getAddrInfo passive Nothing (Just "8080")
48+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
49+ setSocketOption sock ReuseAddr 1
50+ bindSocket sock (addrAddress serverAddr)
51+ listen sock 1
52+
53+ -- Set up multiplexing channel
54+ multiplexChannel <- newChan
55+
56+ -- Connect to the client (to reply)
57+ forkIO $ do
58+ takeMVar clientReady
59+ clientAddr:_ <- getAddrInfo Nothing (Just "127.0.0.1") (Just "8081")
60+ pongSock <- socket (addrFamily clientAddr) Stream defaultProtocol
61+ N.connect pongSock (addrAddress clientAddr)
62+ forever $ readChan multiplexChannel >>= send pongSock
63+
64+ -- Wait for incoming connections (pings from the client)
65+ putMVar serverReady ()
66+ (pingSock, pingAddr) <- accept sock
67+ socketToChan pingSock multiplexChannel
68+
69+ -- Start the client
70+ forkIO $ do
71+ clientAddr:_ <- getAddrInfo passive Nothing (Just "8081")
72+ sock <- socket (addrFamily clientAddr) Stream defaultProtocol
73+ setSocketOption sock ReuseAddr 1
74+ bindSocket sock (addrAddress clientAddr)
75+ listen sock 1
76+
77+ -- Set up multiplexing channel
78+ multiplexChannel <- newChan
79+
80+ -- Connect to the server (to send pings)
81+ forkIO $ do
82+ takeMVar serverReady
83+ serverAddr:_ <- getAddrInfo Nothing (Just "127.0.0.1") (Just "8080")
84+ pingSock <- socket (addrFamily serverAddr) Stream defaultProtocol
85+ N.connect pingSock (addrAddress serverAddr)
86+ ping pingSock multiplexChannel (read pingsStr)
87+ putMVar clientDone ()
88+
89+ -- Wait for incoming connections (pongs from the server)
90+ putMVar clientReady ()
91+ (pongSock, pongAddr) <- accept sock
92+ socketToChan pongSock multiplexChannel
93+
94+ -- Wait for the client to finish
95+ takeMVar clientDone
96+
97+socketToChan :: Socket -> Chan ByteString -> IO ()
98+socketToChan sock chan = go
99+ where
100+ go = do bs <- recv sock
101+ when (BS.length bs > 0) $ do
102+ writeChan chan bs
103+ go
104+
105+pingMessage :: ByteString
106+pingMessage = pack "ping123"
107+
108+ping :: Socket -> Chan ByteString -> Int -> IO ()
109+ping sock chan pings = go pings
110+ where
111+ go :: Int -> IO ()
112+ go 0 = do
113+ putStrLn $ "client did " ++ show pings ++ " pings"
114+ go !i = do
115+ before <- getCurrentTime
116+ send sock pingMessage
117+ bs <- readChan chan
118+ after <- getCurrentTime
119+ -- putStrLn $ "client received " ++ unpack bs
120+ let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
121+ hPutStrLn stderr $ show i ++ " " ++ show latency
122+ go (i - 1)
123+
124+-- | Receive a package
125+recv :: Socket -> IO ByteString
126+recv sock = do
127+ header <- NBS.recv sock 4
128+ length <- decodeLength header
129+ NBS.recv sock (fromIntegral (length :: Int32))
130+
131+-- | Send a package
132+send :: Socket -> ByteString -> IO ()
133+send sock bs = do
134+ length <- encodeLength (fromIntegral (BS.length bs))
135+ NBS.sendMany sock [length, bs]
136+
137+-- | Encode length (manual for now)
138+encodeLength :: Int32 -> IO ByteString
139+encodeLength i32 =
140+ BSI.create 4 $ \p ->
141+ pokeByteOff p 0 (htonl (fromIntegral i32))
142+
143+-- | Decode length (manual for now)
144+decodeLength :: ByteString -> IO Int32
145+decodeLength bs =
146+ let (fp, _, _) = BSI.toForeignPtr bs in
147+ withForeignPtr fp $ \p -> do
148+ w32 <- peekByteOff p 0
149+ return (fromIntegral (ntohl w32))
benchmarks/Makefile.JustPing+1
@@ -6,6 +6,7 @@ INCLUDES=-i../network-transport/src -i../network-transport-pipes/src
6Indirection.ps: JustPingWithHeader.data \
6Indirection.ps: JustPingWithHeader.data \
7 JustPingThroughChan.data \
7 JustPingThroughChan.data \
8 JustPingThroughMVar.data \
8 JustPingThroughMVar.data \
9+ JustPingTwoSocketPairs.data \
9 JustPingTransport.data
10 JustPingTransport.data
10 gnuplot ./Indirection.gnuplot
11 gnuplot ./Indirection.gnuplot
11
12