git.iko.soy/distributed-process
on master help sign in
Start comparing the transport against base tests

Modified the TCP transport not to use serialization for lengths, but since the ping messages are only short this has no effect on the tests. The transport is still significnatly slower though. Not yet sure what's causing this.

commit610efb900c4532fd3fc152cba97ead8b0e74830c authorEdsko de Vries <edsko@well-typed.com> dateFri 13 Apr 2012, 16:14 (15 years ago) parent842cc13
5 files changed+146−28
compare with master
unifiedsplit
benchmarks/Headers.gnuplot+4−4 benchmarks/Indirection.gnuplot+4−2 benchmarks/JustPingTransport.hs+95 benchmarks/Makefile.JustPing+7−5 network-transport/src/Network/Transport/TCP.hs+36−17
benchmarks/Headers.gnuplot+4−4
@@ -1,15 +1,15 @@
11set title "Roundtrip (us)"
22set yrange [0:150]
3−plot "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \
4− "JustPingHaskell2.data" smooth bezier with lines title "JustPingHaskell", \
3+plot "JustPingHaskellNT.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \
4+ "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell", \
55 "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
66 "JustPingOneRecv.data" smooth bezier with lines title "JustPingOneRecv", \
77 "JustPingCacheHeader.data" smooth bezier with lines title "JustPingCacheHeader", \
88 "JustPingC.data" smooth bezier with lines title "JustPingC"
99set terminal postscript color
1010set output "Headers.ps"
11−plot "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \
12− "JustPingHaskell2.data" smooth bezier with lines title "JustPingHaskell", \
11+plot "JustPingHaskellNT.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \
12+ "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell", \
1313 "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
1414 "JustPingOneRecv.data" smooth bezier with lines title "JustPingOneRecv", \
1515 "JustPingCacheHeader.data" smooth bezier with lines title "JustPingCacheHeader", \
benchmarks/Indirection.gnuplot+4−2
@@ -2,9 +2,11 @@ set title "Roundtrip (us)"
22set yrange [0:150]
33plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
44 "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+ "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"
67set terminal postscript color
78set output "Headers.ps"
89plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \
910 "JustPingThroughChan.data" smooth bezier with lines title "JustPingThroughChan", \
10− "JustPingThroughMVar.data" smooth bezier with lines title "JustPingThroughMVar"
11+ "JustPingThroughMVar.data" smooth bezier with lines title "JustPingThroughMVar", \
12+ "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"
benchmarks/JustPingTransport.hsnew file+95
@@ -0,0 +1,95 @@
1+{-# LANGUAGE CPP, BangPatterns #-}
2+
3+module Main where
4+
5+import Control.Monad
6+
7+import Data.Int
8+import System.Environment (getArgs, withArgs)
9+import Data.Time (getCurrentTime, diffUTCTime, NominalDiffTime)
10+import System.IO (withFile, IOMode(..), hPutStrLn, Handle, stderr)
11+import Control.Concurrent (forkIO)
12+import Control.Concurrent.MVar (newEmptyMVar, takeMVar, putMVar)
13+import qualified Network.Socket as N
14+import Debug.Trace
15+import Data.ByteString (ByteString)
16+import Data.ByteString.Char8 (pack, unpack)
17+import qualified Data.ByteString as BS
18+import qualified Network.Socket.ByteString as NBS
19+import Data.Time (getCurrentTime, diffUTCTime, NominalDiffTime)
20+import Network.Transport (newConnection, receive, connect, send, defaultHints,
21+ serialize, deserialize, SourceEnd, TargetEnd)
22+import Network.Transport.TCP (mkTransport, TCPConfig (..))
23+
24+main :: IO ()
25+main = do
26+ [pingsStr] <- getArgs
27+ serverAddr <- newEmptyMVar
28+ clientAddr <- newEmptyMVar
29+ clientDone <- newEmptyMVar
30+
31+ -- Start the server
32+ forkIO $ do
33+ -- establish transport
34+ putStrLn "server: creating TCP connection"
35+ transport <- mkTransport $ TCPConfig defaultHints "127.0.0.1" "8080"
36+
37+ -- create ping end
38+ (sourceAddrPing, targetEndPing) <- newConnection transport
39+ putMVar serverAddr sourceAddrPing
40+
41+ -- create pong end
42+ sourceAddrPong <- takeMVar clientAddr
43+ sourceEndPong <- connect sourceAddrPong
44+
45+ -- reply to pings with pongs
46+ putStrLn "server: awaiting client connection"
47+ pong targetEndPing sourceEndPong
48+
49+ -- Start the client
50+ forkIO $ do
51+ sourceAddr <- takeMVar serverAddr
52+ let pings = read pingsStr
53+
54+ -- establish transport
55+ transport <- mkTransport $ TCPConfig defaultHints "127.0.0.1" "8081"
56+
57+ -- Create ping end
58+ sourceEndPing <- connect sourceAddr
59+
60+ -- Create pong end
61+ (sourceAddrPong, targetEndPong) <- newConnection transport
62+ putMVar clientAddr sourceAddrPong
63+
64+ ping sourceEndPing targetEndPong pings
65+ putMVar clientDone ()
66+
67+ -- Wait for the client to finish
68+ takeMVar clientDone
69+
70+pingMessage :: ByteString
71+pingMessage = pack "ping123"
72+
73+ping :: SourceEnd -> TargetEnd -> Int -> IO ()
74+ping sourceEndPing targetEndPong pings = go pings
75+ where
76+ go :: Int -> IO ()
77+ go 0 = do
78+ putStrLn $ "client did " ++ show pings ++ " pings"
79+ go !i = do
80+ before <- getCurrentTime
81+ send sourceEndPing (replicate i pingMessage)
82+ [bs] <- receive targetEndPong
83+ after <- getCurrentTime
84+ putStrLn $ "client received " ++ unpack bs
85+ let latency = (1e6 :: Double) * realToFrac (diffUTCTime after before)
86+ hPutStrLn stderr $ show i ++ " " ++ show latency
87+ go (i - 1)
88+
89+pong :: TargetEnd -> SourceEnd -> IO ()
90+pong targetEndPing sourceEndPong = do
91+ [bs] <- receive targetEndPing
92+ putStrLn $ "server received " ++ unpack bs
93+ when (BS.length bs > 0) $ do
94+ send sourceEndPong [bs]
95+ pong targetEndPing sourceEndPong
benchmarks/Makefile.JustPing+7−5
@@ -1,15 +1,17 @@
11# The "Just Ping" tests measure ping latency with as little overhead as possible
22
33NUMPINGS=100000
4+INCLUDES=-i../network-transport/src -i../network-transport-pipes/src
45
56Indirection.ps: JustPingWithHeader.data \
67 JustPingThroughChan.data \
7− JustPingThroughMVar.data
8+ JustPingThroughMVar.data \
9+ JustPingTransport.data
810 gnuplot ./Indirection.gnuplot
911
1012Headers.ps: JustPingC.data \
13+ JustPingHaskellNT.data \
1114 JustPingHaskell.data \
12− JustPingHaskell2.data \
1315 JustPingWithHeader.data \
1416 JustPingOneRecv.data \
1517 JustPingCacheHeader.data
@@ -18,14 +20,14 @@ Headers.ps: JustPingC.data \
1820JustPingC.exe: JustPingC.c
1921 gcc -o JustPingC.exe -O2 -std=c99 JustPingC.c
2022
21−JustPingHaskell.exe: JustPingHaskell.hs
22− ghc -o JustPingHaskell.exe -O2 JustPingHaskell.hs
23+JustPingHaskellNT.exe: JustPingHaskell.hs
24+ ghc -o JustPingHaskellNT.exe -O2 JustPingHaskell.hs
2325
2426%.data : %.exe
2527 time ./$< $(NUMPINGS) 2>$@
2628
2729%.exe :: %.hs
28− ghc -o $@ -O2 -threaded $<
30+ ghc -o $@ -O2 -threaded --make -rtsopts $< $(INCLUDES)
2931
3032.PHONY: clean
3133clean:
network-transport/src/Network/Transport/TCP.hs+36−17
@@ -39,6 +39,9 @@ import Text.Printf
3939
4040import Safe
4141import System.IO (stderr, hPutStrLn)
42+import Foreign.Storable (pokeByteOff, peekByteOff)
43+import Foreign.C (CInt(..))
44+import Foreign.ForeignPtr (withForeignPtr)
4245
4346#ifndef LAZY
4447import Data.ByteString.Char8 (ByteString)
@@ -60,6 +63,9 @@ decode = Ser.decodeLazy
6063encode :: Ser.Serialize a => a -> ByteString
6164decode :: Ser.Serialize a => ByteString -> Either String a
6265
66+foreign import ccall unsafe "htonl" htonl :: CInt -> CInt
67+foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt
68+
6369type ChanId = Int
6470type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
6571
@@ -134,11 +140,12 @@ mkSourceEnd host service chanId = withSocketsDo $ do
134140 N.connect sock (addrAddress serverAddr)
135141 NBS.sendAll sock $ encode (fromIntegral chanId :: Int32)
136142 return SourceEnd
137− { send = {-# SCC "send" #-} \bss -> do
138− let size = fromIntegral (sum . map BS.length $ bss) :: Int32
139− sizeStr | size < 255 = BS.singleton (toEnum $ fromEnum size)
140− | otherwise = BS.cons (toEnum 255) (encode size)
141− sendNBS sock (sizeStr:bss)
143+ { send = {-# SCC "send" #-} \bss ->
144+ let size = fromIntegral (sum . map BS.length $ bss) :: Int32 in
145+ if size < 255
146+ then sendNBS sock (BS.singleton (toEnum $ fromEnum size) : bss)
147+ else do size' <- encodeLength size
148+ sendNBS sock (BS.singleton (toEnum 255) : size' : bss)
142149 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
143150 }
144151
@@ -193,15 +200,15 @@ procMessages chans chanId chan sock = do
193200 esizeBS <- recvExact sock 1
194201 case esizeBS of
195202 Left _ -> closeSocket
196− Right sizeBS -> do
197− case decode sizeBS :: Either String Word8 of
198− Right 255 -> do
199− esizeBS' <- recvExact sock (fromIntegral $ sizeOf (0 :: Int32))
200− case esizeBS' of
201− Left _ -> closeSocket
202− Right sizeBS' -> procMessage $ decode sizeBS'
203− esize@(Right _) -> procMessage (fromIntegral <$> esize)
204− Left msg -> error msg
203+ Right sizeBS ->
204+ let size = toEnum . fromEnum $ BS.index sizeBS 0 in
205+ if size < 255
206+ then procMessage size
207+ else do esizeBS' <- recvExact sock (fromIntegral $ sizeOf (0 :: Int32))
208+ case esizeBS' of
209+ Left _ -> closeSocket
210+ Right sizeBS' -> do sizeDec <- decodeLength sizeBS'
211+ procMessage sizeDec
205212 where
206213 err m = error $ printf "procMessages: %s" m
207214 closeSocket :: IO ()
@@ -216,9 +223,8 @@ procMessages chans chanId chan sock = do
216223 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
217224 putMVar chans (chanId', chanMap')
218225 sClose sock
219− procMessage :: Either String Int32 -> IO ()
220− procMessage (Left msg) = err msg
221− procMessage (Right size) = do
226+ procMessage :: Int32 -> IO ()
227+ procMessage size = do
222228 ebs <- recvExact sock size
223229 case ebs of
224230 Left _ -> closeSocket
@@ -289,3 +295,16 @@ forkWithExceptions forkit descr action = do
289295 throwTo parent (e::SomeException)
290296 )
291297
298+-- | Encode length (manual for now)
299+encodeLength :: Int32 -> IO ByteString
300+encodeLength i32 =
301+ BSI.create 4 $ \p ->
302+ pokeByteOff p 0 (htonl (fromIntegral i32))
303+
304+-- | Decode length (manual for now)
305+decodeLength :: ByteString -> IO Int32
306+decodeLength bs =
307+ let (fp, _, _) = BSI.toForeignPtr bs in
308+ withForeignPtr fp $ \p -> do
309+ w32 <- peekByteOff p 0
310+ return (fromIntegral (ntohl w32))