506d4ba · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fixed the pipes transport so it opens the file only once on the reader side.

506d4baRyan Newton15 years ago

benchmarks/Makefile+6−4

compare with master

unifiedsplit

@@ -4,17 +4,19 @@ ifeq ($(GHC),)

4 GHC=ghc

5endif

7−all: tcp transport

7+all: pipes tcp transport

8tcp:

9− $(GHC) -threaded --make -i../network-transport/src PingTCP.hs -o PingTCP.exe

9+ $(GHC) -O2 -threaded --make -i../network-transport/src PingTCP.hs -o PingTCP.exe

10transport:

11− $(GHC) -threaded --make -i../network-transport/src PingTransport.hs -o PingTransport.exe

11+ $(GHC) -O2 -threaded --make -i../network-transport/src PingTransport.hs -o PingTransport.exe

12+pipes:

13+ $(GHC) -O2 -threaded --make -i../network-transport/src PingPipes.hs -o PingPipes.exe

13run: runBaseline runTransport

15run: runBaseline runTransport

14runBaseline:

16runBaseline:

15 ./PingTCP.exe server 8080 &

17 ./PingTCP.exe server 8080 &

16 sleep 1

18 sleep 1

17− ./PingTCP.exe client 0.0.0.0 8081 1000

19+ ./PingTCP.exe client 0.0.0.0 8080 1000

19runTransport:

21runTransport:

20 ./PingTransport.exe server 0.0.0.0 8080 sourceAddr &

22 ./PingTransport.exe server 0.0.0.0 8080 sourceAddr &

benchmarks/PingPipes.hsnew file+94

@@ -0,0 +1,94 @@

1+module Main where

3+import Network.Transport

4+import Network.Transport.Pipes (mkTransport)

6+import Control.Monad (forever, replicateM_)

7+import Criterion.Main (Benchmark, bench, defaultMain, nfIO)

8+import Data.Binary

9+import Data.Maybe (fromJust)

10+import Data.Int

11+import System.Environment (getArgs, withArgs)

12+

13+import qualified Data.ByteString.Lazy.Char8 as BS

14+

15+-- | This performs a ping benchmark on the TCP transport. This can be

16+-- compiled using:

17+--

18+-- ghc --make -idistributed-process/src -inetwork-transport/src -O2 benchmarks/PingTransport.hs

19+--

20+-- To use the compiled binary, first set up a server:

21+--

22+-- ./benchmarks/PingPipes server sourceAddr

23+--

24+-- Once this is established, launch a client to perform the benchmark. The

25+-- following command sends 1000 pings per mark.

26+--

27+-- ./benchmarks/PingPipes client sourceAddr 1000

28+--

29+-- The server must be restarted between benchmarks.

30+main :: IO ()

31+main = do

32+ args <- getArgs

33+ case args of

34+ "server" : sourceAddrFilePath : [] -> do

35+ -- establish transport

36+ transport <- mkTransport

37+

38+ -- create ping end

39+ putStrLn "server: creating ping end"

40+ (sourceAddrPing, targetEndPing) <- newConnection transport

41+ BS.writeFile sourceAddrFilePath $ serialize sourceAddrPing

42+

43+ -- create pong end

44+ putStrLn "server: creating pong end"

45+ [sourceAddrPongBS] <- receive targetEndPing

46+ sourceEndPong <- connect . fromJust $ deserialize transport sourceAddrPongBS

47+

48+ -- always respond to a ping with a pong

49+ putStrLn "server: awaiting pings"

50+ forever $ pong targetEndPing sourceEndPong

51+

52+

53+ "client" : sourceAddrFilePath : pingsStr : args' -> do

54+ let pings = read pingsStr

55+ -- establish transport

56+ transport <- mkTransport

57+

58+ -- create ping end

59+ sourceAddrPingBS <- BS.readFile sourceAddrFilePath

60+ sourceEndPing <- connect . fromJust $ deserialize transport sourceAddrPingBS

61+

62+ -- create pong end

63+ (sourceAddrPong, targetEndPong) <- newConnection transport

64+ send sourceEndPing [serialize sourceAddrPong]

65+

66+ -- benchmark the pings

67+-- withArgs args' $ defaultMain [ benchPing sourceEndPing targetEndPong pings]

68+ replicateM_ pings (ping sourceEndPing targetEndPong)

69+ putStrLn "Done with all ping/pongs."

70+

71+-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for

72+-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent

73+-- in reply down the `SourceEnd`, repeating whatever was sent.

74+pong :: TargetEnd -> SourceEnd -> IO ()

75+pong targetEndPing sourceEndPong = do

76+ bs <- receive targetEndPing

77+ send sourceEndPong bs

78+

79+-- | The effect of `ping sourceEndPing targetEndPong n` is to send the number

80+-- `n` using `sourceEndPing`, and to then receive the a number from

81+-- `targetEndPong`, which is then returned.

82+ping :: SourceEnd -> TargetEnd -> IO Int64

83+ping sourceEndPing targetEndPong = do

84+ send sourceEndPing [BS.empty]

85+ [bs] <- receive targetEndPong

86+ return $ decode bs

87+

88+-- | The effect of `benchPing sourceEndPing targetEndPong n` is to send

89+-- `n` pings down `sourceEndPing` using the `ping` function. The time

90+-- taken is benchmarked.

91+benchPing :: SourceEnd -> TargetEnd -> Int64 -> Benchmark

92+benchPing sourceEndPing targetEndPong n = bench "PingTransport" $

93+ nfIO (replicateM_ (fromIntegral n) (ping sourceEndPing targetEndPong))

94+

benchmarks/PingTCP.hs+3−1

@@ -67,6 +67,7 @@ main = do

68 -- benchmark the pings

69 withArgs args' $ defaultMain [ benchPing sock pings ]

70+-- replicateM_ pings (ping sock)

71-- | Each `ping` sends a single byte, and expects to receive one

72-- | Each `ping` sends a single byte, and expects to receive one

72-- back in return.

73-- back in return.

@@ -83,6 +84,7 @@ pong sock = do

83 return ()

84 return ()

85benchPing :: Socket -> Int64 -> Benchmark

86benchPing :: Socket -> Int64 -> Benchmark

86−benchPing sock n = bench "PingTCP" $

87+benchPing sock n =

88+ bench "PingTCP" $

87 nfIO (replicateM_ (fromIntegral n) (ping sock))

89 nfIO (replicateM_ (fromIntegral n) (ping sock))

benchmarks/PingTransport.hs+2−1

@@ -64,7 +64,8 @@ main = do

64 send sourceEndPing [serialize sourceAddrPong]

66 -- benchmark the pings

67− withArgs args' $ defaultMain [ benchPing sourceEndPing targetEndPong pings]

67+-- withArgs args' $ defaultMain [ benchPing sourceEndPing targetEndPong pings]

68+ replicateM_ pings (ping sourceEndPing targetEndPong)

69-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for

70-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for

70-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent

71-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent

network-transport/src/Network/Transport/Pipes.hs+27−19

@@ -41,16 +41,19 @@ import Network.Transport

41import System.Random (randomIO)

42import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),

43 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)

44+import System.IO.Error (ioeGetHandle)

44import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)

45import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)

45import System.Posix.Types (Fd)

46import System.Posix.Types (Fd)

46import System.Directory (doesFileExist, removeFile)

47import System.Directory (doesFileExist, removeFile)

48+import System.Mem (performGC)

48-- define DEBUG

50-- define DEBUG

49-- define USE_UNIX_BYTESTRING

51-- define USE_UNIX_BYTESTRING

51#ifdef USE_UNIX_BYTESTRING

53#ifdef USE_UNIX_BYTESTRING

52import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO

54import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO

53−import System.Posix.IO as PIO (openFd, defaultFileFlags, OpenMode(ReadWrite, WriteOnly))

55+import System.Posix.IO as PIO (openFd, closeFd, defaultFileFlags,

56+ OpenMode(ReadWrite, WriteOnly))

54(fromS,toS) = (BS.pack, BS.unpack)

57(fromS,toS) = (BS.pack, BS.unpack)

55(fromBS,toBS) = (id,id)

58(fromBS,toBS) = (id,id)

56readit fd n = PIO.fdRead fd n

59readit fd n = PIO.fdRead fd n

@@ -91,8 +94,14 @@ mkTransport = do

92 return Transport

95 return Transport

93 { newConnectionWith = \ _ -> do

96 { newConnectionWith = \ _ -> do

97+ -- Here we protect from blocking other threads by running on a separate (OS) thread:

98+ -- Opening the file on the reader side should always succeed:

99+ mv <- onOSThread$ tryUntilNoIOErr $

100+-- spinTillThere filename

101+ PIO.openFd filename PIO.ReadOnly Nothing fileFlags

102+

94 return (mkSourceAddr filename,

103 return (mkSourceAddr filename,

95− mkTargetEnd filename lock)

104+ mkTargetEnd mv lock)

96 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"

105 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"

97 , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)

106 , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)

98 , closeTransport = removeFile filename

107 , closeTransport = removeFile filename

@@ -107,11 +116,9 @@ mkTransport = do

107 mkSourceEnd :: String -> IO SourceEnd

116 mkSourceEnd :: String -> IO SourceEnd

108 mkSourceEnd filename = do

117 mkSourceEnd filename = do

109 -- Initiate but do not block on file opening:

118 -- Initiate but do not block on file opening:

110−

111− -- Here we protect from blocking other threads by running on a separate (OS) thread:

112 mv <- onOSThread$ tryUntilNoIOErr $

119 mv <- onOSThread$ tryUntilNoIOErr $

113 -- The reader must connect first, the writer here spins with backoff.

120 -- The reader must connect first, the writer here spins with backoff.

114− PIO.openFd filename PIO.WriteOnly Nothing fileFlags

121+ PIO.openFd filename PIO.WriteOnly Nothing fileFlags

115

122

116 return $

123 return $

117 -- Write to the named pipe. If the message is less than

124 -- Write to the named pipe. If the message is less than

@@ -128,31 +135,26 @@ mkTransport = do

128-- let finalmsg = BS.concat (encode msgsize : bss)

135-- let finalmsg = BS.concat (encode msgsize : bss)

129 let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)

136 let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)

130

137

131− fd <- readMVar mv -- Synchronnize with file opening.

138+ fd <- readMVar mv -- Synchronize with file opening.

132 ----------------------------------------

139 ----------------------------------------

133 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!

140 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!

134 unless (fromIntegral cnt == BS.length finalmsg) $

141 unless (fromIntegral cnt == BS.length finalmsg) $

135 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)

142 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)

136 ----------------------------------------

143 ----------------------------------------

137 return ()

144 return ()

138− , closeSourceEnd = error "Pipes.hs: closeSourceEnd not yet implemented"

145+ , closeSourceEnd = do

146+ fd <- readMVar mv

147+ PIO.closeFd fd

139 }

148 }

140

149

141− mkTargetEnd :: String -> MVar () -> TargetEnd

150+ mkTargetEnd :: MVar Fd -> MVar () -> TargetEnd

142− mkTargetEnd filename lock = TargetEnd

151+ mkTargetEnd mv lock = TargetEnd

143 { receive = do

152 { receive = do

153+ fd <- readMVar mv -- Make sure Fd is there before

154+

144 -- This should only happen on a single process. But it may

155 -- This should only happen on a single process. But it may

145 -- happen on multiple threads so we grab a lock.

156 -- happen on multiple threads so we grab a lock.

146 takeMVar lock

157 takeMVar lock

147−

148− spinTillThere filename

149− -- Opening the file on the reader side should always succeed:

150− fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags

151−

152− let oneread fd n =

153− do bs <- tryUntilNoIOErr$

154− readit fd (fromIntegral n)

155− return (bs, BS.length bs)

156

158

157 let spinread :: Int64 -> IO BS.ByteString

159 let spinread :: Int64 -> IO BS.ByteString

158 spinread desired = do

160 spinread desired = do

@@ -178,7 +180,9 @@ mkTransport = do

178 Right size -> spinread (fromIntegral (size::Word32))

180 Right size -> spinread (fromIntegral (size::Word32))

179 putMVar lock ()

181 putMVar lock ()

180 return [payload]

182 return [payload]

181− , closeTargetEnd = error "Pipes.hs: closeTargetEnd not yet implemented"

183+ , closeTargetEnd = do

184+ fd <- readMVar mv

185+ PIO.closeFd fd

182 }

186 }

183

187

184spinTillThere :: String -> IO ()

188spinTillThere :: String -> IO ()

@@ -202,6 +206,10 @@ tryUntilNoIOErr action = mkBackoff >>= loop

202 loop bkoff =

206 loop bkoff =

203 handle (\ (e :: IOException) ->

207 handle (\ (e :: IOException) ->

204 do bkoff

208 do bkoff

209+-- BSS.hPutStr stderr$ BSS.pack$ " got IO err: " ++ show e

210+ -- case ioeGetHandle e of

211+ -- Nothing -> BSS.hPutStrLn stderr$ BSS.pack$ " no hndl io err."

212+ -- Just x -> BSS.hPutStrLn stderr$ BSS.pack$ " HNDL on io err!" ++ show x

205 loop bkoff) $

213 loop bkoff) $

206 action

214 action

207

215