@@ -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))