git.iko.soy/distributed-process
on master help sign in
Run server and client inside a single process
90cb954Edsko de Vries
benchmarks/PingTCP.hs+38−32
compare with master
unifiedsplit
@@ -16,6 +16,8 @@ import Network.Socket
16import System.Environment (getArgs, withArgs)
16import System.Environment (getArgs, withArgs)
17import Data.Time (getCurrentTime, diffUTCTime, NominalDiffTime)
17import Data.Time (getCurrentTime, diffUTCTime, NominalDiffTime)
18import System.IO (withFile, IOMode(..), hPutStrLn)
18import System.IO (withFile, IOMode(..), hPutStrLn)
19+import Control.Concurrent (forkIO)
20+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
19
21
20import qualified Network.Socket as N
22import qualified Network.Socket as N
21
23
@@ -56,38 +58,42 @@ decode :: Ser.Serialize a => ByteString -> Either String a
56-- The server must be restarted between benchmarks.
58-- The server must be restarted between benchmarks.
57main :: IO ()
59main :: IO ()
58main = do
60main = do
59− args <- getArgs
61+ [pingsStr] <- getArgs
60− case args of
62+ serverReady <- newEmptyMVar
61− "server" : service : [] -> withSocketsDo $ do
63+
64+ -- Start the server
65+ forkIO $ do
62− putStrLn "server: creating TCP connection"
66+ putStrLn "server: creating TCP connection"
63− serverAddrs <- getAddrInfo
67+ serverAddrs <- getAddrInfo
64− (Just (defaultHints { addrFlags = [AI_PASSIVE] } ))
68+ (Just (defaultHints { addrFlags = [AI_PASSIVE] } ))
65− Nothing
69+ Nothing
66− (Just service)
70+ (Just "8080")
67− let serverAddr = head serverAddrs
71+ let serverAddr = head serverAddrs
68− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
72+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
69− setSocketOption sock ReuseAddr 1
73+ setSocketOption sock ReuseAddr 1
70− bindSocket sock (addrAddress serverAddr)
74+ bindSocket sock (addrAddress serverAddr)
71−
75+
72− putStrLn "server: awaiting client connection"
76+ putStrLn "server: awaiting client connection"
77+ putMVar serverReady ()
73− listen sock 1
78+ listen sock 1
74− (clientSock, clientAddr) <- accept sock
79+ (clientSock, clientAddr) <- accept sock
75−
80+
76− putStrLn "server: listening for pings"
81+ putStrLn "server: listening for pings"
77− pong clientSock
82+ pong clientSock
78−
83+
79− "client": host : service : pingsStr : [] -> withSocketsDo $ do
84+ -- Client
85+ takeMVar serverReady
80− let pings = read pingsStr
86+ let pings = read pingsStr
81− serverAddrs <- getAddrInfo
87+ serverAddrs <- getAddrInfo
82− Nothing
88+ Nothing
83− (Just host)
89+ (Just "127.0.0.1")
84− (Just service)
90+ (Just "8080")
85− let serverAddr = head serverAddrs
91+ let serverAddr = head serverAddrs
86− sock <- socket (addrFamily serverAddr) Stream defaultProtocol
92+ sock <- socket (addrFamily serverAddr) Stream defaultProtocol
87−
93+
88− N.connect sock (addrAddress serverAddr)
94+ N.connect sock (addrAddress serverAddr)
89−
95+
90− ping sock pings
96+ ping sock pings
91
97
92pingMessage :: ByteString
98pingMessage :: ByteString
93pingMessage = pack "ping123"
99pingMessage = pack "ping123"
benchmarks/PingTCPTransport.hs+39−36
@@ -14,6 +14,8 @@ import Data.Int
14import System.Environment (getArgs, withArgs)
14import System.Environment (getArgs, withArgs)
15import Data.Time (getCurrentTime, diffUTCTime)
15import Data.Time (getCurrentTime, diffUTCTime)
16import System.IO (withFile, IOMode(..), hPutStrLn)
16import System.IO (withFile, IOMode(..), hPutStrLn)
17+import Control.Concurrent (forkIO)
18+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
17
19
18#ifndef LAZY
20#ifndef LAZY
19import qualified Data.ByteString.Char8 as BS
21import qualified Data.ByteString.Char8 as BS
@@ -50,42 +52,43 @@ decode :: Ser.Serialize a => BS.ByteString -> Either String a
50-- The server must be restarted between benchmarks.
52-- The server must be restarted between benchmarks.
51main :: IO ()
53main :: IO ()
52main = do
54main = do
53− args <- getArgs
55+ [pingsStr] <- getArgs
54− case args of
56+ serverReady <- newEmptyMVar
55− "server" : host : service : sourceAddrFilePath : [] -> do
57+
58+ -- Start the server
59+ forkIO $ do
56− -- establish transport
60+ -- establish transport
57− transport <- mkTransport $ TCPConfig defaultHints host service
61+ transport <- mkTransport $ TCPConfig defaultHints "127.0.0.1" "8080"
58−
62+
59− -- create ping end
63+ -- create ping end
60− putStrLn "server: creating ping end"
64+ putStrLn "server: creating ping end"
61− (sourceAddrPing, targetEndPing) <- newConnection transport
65+ (sourceAddrPing, targetEndPing) <- newConnection transport
62− BS.writeFile sourceAddrFilePath $ serialize sourceAddrPing
66+ putMVar serverReady sourceAddrPing
63−
67+
64− -- create pong end
68+ -- create pong end
65− putStrLn "server: creating pong end"
69+ putStrLn "server: creating pong end"
66− [sourceAddrPongBS] <- receive targetEndPing
70+ [sourceAddrPongBS] <- receive targetEndPing
67− sourceEndPong <- connect . fromJust $ deserialize transport sourceAddrPongBS
71+ sourceEndPong <- connect . fromJust $ deserialize transport sourceAddrPongBS
68−
72+
69− -- always respond to a ping with a pong
73+ -- always respond to a ping with a pong
70− putStrLn "server: awaiting pings"
74+ putStrLn "server: awaiting pings"
71− pong targetEndPing sourceEndPong
75+ pong targetEndPing sourceEndPong
72−
76+
73−
77+ -- Client
74− "client" : host : service : sourceAddrFilePath : pingsStr : [] -> do
78+ sourceAddr <- takeMVar serverReady
75− let pings = read pingsStr
79+ let pings = read pingsStr
76− -- establish transport
80+ -- establish transport
77− transport <- mkTransport $ TCPConfig defaultHints host service
81+ transport <- mkTransport $ TCPConfig defaultHints "127.0.0.1" "8081"
78−
82+
79− -- create ping end
83+ -- create ping end
80− sourceAddrPingBS <- BS.readFile sourceAddrFilePath
81− sourceEndPing <- connect . fromJust $ deserialize transport sourceAddrPingBS
84+ sourceEndPing <- connect sourceAddr
82−
85+
83− -- create pong end
86+ -- create pong end
84− (sourceAddrPong, targetEndPong) <- newConnection transport
87+ (sourceAddrPong, targetEndPong) <- newConnection transport
85− send sourceEndPing [serialize sourceAddrPong]
88+ send sourceEndPing [serialize sourceAddrPong]
86−
89+
87− ping sourceEndPing targetEndPong pings
90+ ping sourceEndPing targetEndPong pings
88− putStrLn "Done with all ping/pongs."
91+ putStrLn "Done with all ping/pongs."
89
92
90-- | The message we use to ping (CAF so that we don't keep encoding it)
93-- | The message we use to ping (CAF so that we don't keep encoding it)
91pingMessage :: BS.ByteString
94pingMessage :: BS.ByteString
benchmarks/pingtest.c+11−6
@@ -39,6 +39,8 @@ double timestamp() {
39}
39}
40
40
41int server() {
41int server() {
42+ printf("starting server\n");
43+
42 struct addrinfo hints, *res;
44 struct addrinfo hints, *res;
43 int error, server_socket;
45 int error, server_socket;
44
46
@@ -161,18 +163,21 @@ int client(int pings) {
161}
163}
162
164
163int usage(int argc, char** argv) {
165int usage(int argc, char** argv) {
164− printf("usage: %s server or %s client <number of pings>\n", argv[0], argv[0]);
166+ printf("usage: %s <number of pings>\n", argv[0]);
165 return -1;
167 return -1;
166}
168}
167
169
168int main(int argc, char** argv) {
170int main(int argc, char** argv) {
169− if(argc == 2 && !strcmp(argv[1], "server")) {
171+ if(argc != 2) {
170− return server();
172+ return usage(argc, argv);
171− } else if(argc == 3 && !strcmp(argv[1], "client")) {
173+ }
174+
175+ if(fork() == 0) {
176+ // TODO: we should wait until we know the server is ready
172 int pings = 0;
177 int pings = 0;
173− sscanf(argv[2], "%d", &pings);
178+ sscanf(argv[1], "%d", &pings);
174 return client(pings);
179 return client(pings);
175 } else {
180 } else {
176− return usage(argc, argv);
181+ return server();
177 }
182 }
178}
183}