610efb9 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Start comparing the transport against base tests

610efb9Edsko de Vries15 years ago

benchmarks/Headers.gnuplot+4−4

compare with master

unifiedsplit

@@ -1,15 +1,15 @@

1set title "Roundtrip (us)"

2set yrange [0:150]

3−plot "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \

3+plot "JustPingHaskellNT.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \

4− "JustPingHaskell2.data" smooth bezier with lines title "JustPingHaskell", \

4+ "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell", \

5 "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \

6 "JustPingOneRecv.data" smooth bezier with lines title "JustPingOneRecv", \

7 "JustPingCacheHeader.data" smooth bezier with lines title "JustPingCacheHeader", \

8 "JustPingC.data" smooth bezier with lines title "JustPingC"

9set terminal postscript color

10set output "Headers.ps"

11−plot "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \

11+plot "JustPingHaskellNT.data" smooth bezier with lines title "JustPingHaskell (NOT -threaded)", \

12− "JustPingHaskell2.data" smooth bezier with lines title "JustPingHaskell", \

12+ "JustPingHaskell.data" smooth bezier with lines title "JustPingHaskell", \

13 "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \

14 "JustPingOneRecv.data" smooth bezier with lines title "JustPingOneRecv", \

15 "JustPingCacheHeader.data" smooth bezier with lines title "JustPingCacheHeader", \

benchmarks/Indirection.gnuplot+4−2

@@ -2,9 +2,11 @@ set title "Roundtrip (us)"

2set yrange [0:150]

3plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \

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+ "JustPingTransport.data" smooth bezier with lines title "JustPingTransport"

6set terminal postscript color

7set terminal postscript color

7set output "Headers.ps"

8set output "Headers.ps"

8plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \

9plot "JustPingWithHeader.data" smooth bezier with lines title "JustPingWithHeader", \

9 "JustPingThroughChan.data" smooth bezier with lines title "JustPingThroughChan", \

10 "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 #-}

3+module Main where

5+import Control.Monad

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

1# The "Just Ping" tests measure ping latency with as little overhead as possible

3NUMPINGS=100000

4+INCLUDES=-i../network-transport/src -i../network-transport-pipes/src

5Indirection.ps: JustPingWithHeader.data \

6Indirection.ps: JustPingWithHeader.data \

6 JustPingThroughChan.data \

7 JustPingThroughChan.data \

7− JustPingThroughMVar.data

8+ JustPingThroughMVar.data \

9+ JustPingTransport.data

8 gnuplot ./Indirection.gnuplot

10 gnuplot ./Indirection.gnuplot

10Headers.ps: JustPingC.data \

12Headers.ps: JustPingC.data \

13+ JustPingHaskellNT.data \

11 JustPingHaskell.data \

14 JustPingHaskell.data \

12− JustPingHaskell2.data \

13 JustPingWithHeader.data \

15 JustPingWithHeader.data \

14 JustPingOneRecv.data \

16 JustPingOneRecv.data \

15 JustPingCacheHeader.data

17 JustPingCacheHeader.data

@@ -18,14 +20,14 @@ Headers.ps: JustPingC.data \

18JustPingC.exe: JustPingC.c

20JustPingC.exe: JustPingC.c

19 gcc -o JustPingC.exe -O2 -std=c99 JustPingC.c

21 gcc -o JustPingC.exe -O2 -std=c99 JustPingC.c

21−JustPingHaskell.exe: JustPingHaskell.hs

23+JustPingHaskellNT.exe: JustPingHaskell.hs

22− ghc -o JustPingHaskell.exe -O2 JustPingHaskell.hs

24+ ghc -o JustPingHaskellNT.exe -O2 JustPingHaskell.hs

24%.data : %.exe

26%.data : %.exe

25 time ./$< $(NUMPINGS) 2>$@

27 time ./$< $(NUMPINGS) 2>$@

27%.exe :: %.hs

29%.exe :: %.hs

28− ghc -o $@ -O2 -threaded $<

30+ ghc -o $@ -O2 -threaded --make -rtsopts $< $(INCLUDES)

30.PHONY: clean

32.PHONY: clean

31clean:

33clean:

network-transport/src/Network/Transport/TCP.hs+36−17

@@ -39,6 +39,9 @@ import Text.Printf

40import Safe

41import System.IO (stderr, hPutStrLn)

42+import Foreign.Storable (pokeByteOff, peekByteOff)

43+import Foreign.C (CInt(..))

44+import Foreign.ForeignPtr (withForeignPtr)

43#ifndef LAZY

46#ifndef LAZY

44import Data.ByteString.Char8 (ByteString)

47import Data.ByteString.Char8 (ByteString)

@@ -60,6 +63,9 @@ decode = Ser.decodeLazy

60encode :: Ser.Serialize a => a -> ByteString

63encode :: Ser.Serialize a => a -> ByteString

61decode :: Ser.Serialize a => ByteString -> Either String a

64decode :: Ser.Serialize a => ByteString -> Either String a

66+foreign import ccall unsafe "htonl" htonl :: CInt -> CInt

67+foreign import ccall unsafe "ntohl" ntohl :: CInt -> CInt

68+

63type ChanId = Int

69type ChanId = Int

64type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))

70type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))

@@ -134,11 +140,12 @@ mkSourceEnd host service chanId = withSocketsDo $ do

134 N.connect sock (addrAddress serverAddr)

140 N.connect sock (addrAddress serverAddr)

135 NBS.sendAll sock $ encode (fromIntegral chanId :: Int32)

141 NBS.sendAll sock $ encode (fromIntegral chanId :: Int32)

136 return SourceEnd

142 return SourceEnd

137− { send = {-# SCC "send" #-} \bss -> do

143+ { send = {-# SCC "send" #-} \bss ->

138− let size = fromIntegral (sum . map BS.length $ bss) :: Int32

144+ let size = fromIntegral (sum . map BS.length $ bss) :: Int32 in

145+ if size < 255

139− sizeStr | size < 255 = BS.singleton (toEnum $ fromEnum size)

146+ then sendNBS sock (BS.singleton (toEnum $ fromEnum size) : bss)

147+ else do size' <- encodeLength size

140− | otherwise = BS.cons (toEnum 255) (encode size)

148+ sendNBS sock (BS.singleton (toEnum 255) : size' : bss)

141− sendNBS sock (sizeStr:bss)

142 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock

149 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock

143 }

150 }

144

151

@@ -193,15 +200,15 @@ procMessages chans chanId chan sock = do

193 esizeBS <- recvExact sock 1

200 esizeBS <- recvExact sock 1

194 case esizeBS of

201 case esizeBS of

195 Left _ -> closeSocket

202 Left _ -> closeSocket

196− Right sizeBS -> do

203+ Right sizeBS ->

197− case decode sizeBS :: Either String Word8 of

204+ let size = toEnum . fromEnum $ BS.index sizeBS 0 in

198− Right 255 -> do

205+ if size < 255

206+ then procMessage size

199− esizeBS' <- recvExact sock (fromIntegral $ sizeOf (0 :: Int32))

207+ else do esizeBS' <- recvExact sock (fromIntegral $ sizeOf (0 :: Int32))

200− case esizeBS' of

208+ case esizeBS' of

201− Left _ -> closeSocket

209+ Left _ -> closeSocket

202− Right sizeBS' -> procMessage $ decode sizeBS'

210+ Right sizeBS' -> do sizeDec <- decodeLength sizeBS'

203− esize@(Right _) -> procMessage (fromIntegral <$> esize)

211+ procMessage sizeDec

204− Left msg -> error msg

205 where

212 where

206 err m = error $ printf "procMessages: %s" m

213 err m = error $ printf "procMessages: %s" m

207 closeSocket :: IO ()

214 closeSocket :: IO ()

@@ -216,9 +223,8 @@ procMessages chans chanId chan sock = do

216 let chanMap' = IntMap.insert chanId (chan, socks') chanMap

223 let chanMap' = IntMap.insert chanId (chan, socks') chanMap

217 putMVar chans (chanId', chanMap')

224 putMVar chans (chanId', chanMap')

218 sClose sock

225 sClose sock

219− procMessage :: Either String Int32 -> IO ()

226+ procMessage :: Int32 -> IO ()

220− procMessage (Left msg) = err msg

227+ procMessage size = do

221− procMessage (Right size) = do

222 ebs <- recvExact sock size

228 ebs <- recvExact sock size

223 case ebs of

229 case ebs of

224 Left _ -> closeSocket

230 Left _ -> closeSocket

@@ -289,3 +295,16 @@ forkWithExceptions forkit descr action = do

289 throwTo parent (e::SomeException)

295 throwTo parent (e::SomeException)

290 )

296 )

291

297

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