@@ -0,0 +1,151 @@
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, NoDelay)
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:args <- 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+ when ("--NoDelay" `elem` args) $ setSocketOption pongSock NoDelay 1
63+ forever $ readChan multiplexChannel >>= send pongSock
64+
65+ -- Wait for incoming connections (pings from the client)
66+ putMVar serverReady ()
67+ (pingSock, pingAddr) <- accept sock
68+ socketToChan pingSock multiplexChannel
69+
70+ -- Start the client
71+ forkIO $ do
72+ clientAddr:_ <- getAddrInfo passive Nothing (Just "8081")
73+ sock <- socket (addrFamily clientAddr) Stream defaultProtocol
74+ setSocketOption sock ReuseAddr 1
75+ bindSocket sock (addrAddress clientAddr)
76+ listen sock 1
77+
78+ -- Set up multiplexing channel
79+ multiplexChannel <- newChan
80+
81+ -- Connect to the server (to send pings)
82+ forkIO $ do
83+ takeMVar serverReady
84+ serverAddr:_ <- getAddrInfo Nothing (Just "127.0.0.1") (Just "8080")
85+ pingSock <- socket (addrFamily serverAddr) Stream defaultProtocol
86+ N.connect pingSock (addrAddress serverAddr)
87+ when ("--NoDelay" `elem` args) $ setSocketOption pingSock NoDelay 1
88+ ping pingSock multiplexChannel (read pingsStr)
89+ putMVar clientDone ()
90+
91+ -- Wait for incoming connections (pongs from the server)
92+ putMVar clientReady ()
93+ (pongSock, pongAddr) <- accept sock
94+ socketToChan pongSock multiplexChannel
95+
96+ -- Wait for the client to finish
97+ takeMVar clientDone
98+
99+socketToChan :: Socket -> Chan ByteString -> IO ()
100+socketToChan sock chan = go
101+ where
102+ go = do bs <- recv sock
103+ when (BS.length bs > 0) $ do
104+ writeChan chan bs
105+ go
106+
107+pingMessage :: ByteString
108+pingMessage = pack "ping123"
109+
110+ping :: Socket -> Chan ByteString -> Int -> IO ()
111+ping sock chan pings = go pings
112+ where
113+ go :: Int -> IO ()
114+ go 0 = do
115+ putStrLn $ "client did " ++ show pings ++ " pings"
116+ go !i = do
117+ before <- getCurrentTime
118+ send sock pingMessage
119+ bs <- readChan chan
120+ after <- getCurrentTime
121+ -- putStrLn $ "client received " ++ unpack bs
122+ let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
123+ hPutStrLn stderr $ show i ++ " " ++ show latency
124+ go (i - 1)
125+
126+-- | Receive a package
127+recv :: Socket -> IO ByteString
128+recv sock = do
129+ header <- NBS.recv sock 4
130+ length <- decodeLength header
131+ NBS.recv sock (fromIntegral (length :: Int32))
132+
133+-- | Send a package
134+send :: Socket -> ByteString -> IO ()
135+send sock bs = do
136+ length <- encodeLength (fromIntegral (BS.length bs))
137+ NBS.sendMany sock [length, bs]
138+
139+-- | Encode length (manual for now)
140+encodeLength :: Int32 -> IO ByteString
141+encodeLength i32 =
142+ BSI.create 4 $ \p ->
143+ pokeByteOff p 0 (htonl (fromIntegral i32))
144+
145+-- | Decode length (manual for now)
146+decodeLength :: ByteString -> IO Int32
147+decodeLength bs =
148+ let (fp, _, _) = BSI.toForeignPtr bs in
149+ withForeignPtr fp $ \p -> do
150+ w32 <- peekByteOff p 0
151+ return (fromIntegral (ntohl w32))