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.

I'm observing a 72 us latency for PingTCP (vs. 28us for the equivalent C program). I'm observing a 99 us latency for PingPipes, and it goes up to 100% CPU. That's with the *String* based Pipes, however!

commit506d4bafdf5a101bf62c8dccbd550851877da1a0 authorRyan Newton <rrnewton@gmail.com> dateTue 21 Feb 2012, 16:54 (15 years ago) parent6995d1f
5 files changed+132−25
compare with master
unifiedsplit
benchmarks/Makefile+6−4 benchmarks/PingPipes.hs+94 benchmarks/PingTCP.hs+3−1 benchmarks/PingTransport.hs+2−1 network-transport/src/Network/Transport/Pipes.hs+27−19
benchmarks/Makefile+6−4
@@ -4,17 +4,19 @@ ifeq ($(GHC),)
44 GHC=ghc
55endif
66
7−all: tcp transport
7+all: pipes tcp transport
88tcp:
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
1010transport:
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
1214
1315run: runBaseline runTransport
1416runBaseline:
1517 ./PingTCP.exe server 8080 &
1618 sleep 1
17− ./PingTCP.exe client 0.0.0.0 8081 1000
19+ ./PingTCP.exe client 0.0.0.0 8080 1000
1820
1921runTransport:
2022 ./PingTransport.exe server 0.0.0.0 8080 sourceAddr &
benchmarks/PingPipes.hsnew file+94
@@ -0,0 +1,94 @@
1+module Main where
2+
3+import Network.Transport
4+import Network.Transport.Pipes (mkTransport)
5+
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
6767
6868 -- benchmark the pings
6969 withArgs args' $ defaultMain [ benchPing sock pings ]
70+-- replicateM_ pings (ping sock)
7071
7172-- | Each `ping` sends a single byte, and expects to receive one
7273-- back in return.
@@ -83,6 +84,7 @@ pong sock = do
8384 return ()
8485
8586benchPing :: Socket -> Int64 -> Benchmark
86−benchPing sock n = bench "PingTCP" $
87+benchPing sock n =
88+ bench "PingTCP" $
8789 nfIO (replicateM_ (fromIntegral n) (ping sock))
8890
benchmarks/PingTransport.hs+2−1
@@ -64,7 +64,8 @@ main = do
6464 send sourceEndPing [serialize sourceAddrPong]
6565
6666 -- 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)
6869
6970-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
7071-- 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
4141import System.Random (randomIO)
4242import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
4343 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)
44+import System.IO.Error (ioeGetHandle)
4445import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
4546import System.Posix.Types (Fd)
4647import System.Directory (doesFileExist, removeFile)
48+import System.Mem (performGC)
4749
4850-- define DEBUG
4951-- define USE_UNIX_BYTESTRING
5052
5153#ifdef USE_UNIX_BYTESTRING
5254import 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))
5457(fromS,toS) = (BS.pack, BS.unpack)
5558(fromBS,toBS) = (id,id)
5659readit fd n = PIO.fdRead fd n
@@ -91,8 +94,14 @@ mkTransport = do
9194
9295 return Transport
9396 { 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+
94103 return (mkSourceAddr filename,
95− mkTargetEnd filename lock)
104+ mkTargetEnd mv lock)
96105 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
97106 , deserialize = \bs -> return$ mkSourceAddr (BS.unpack bs)
98107 , closeTransport = removeFile filename
@@ -107,11 +116,9 @@ mkTransport = do
107116 mkSourceEnd :: String -> IO SourceEnd
108117 mkSourceEnd filename = do
109118 -- Initiate but do not block on file opening:
110−
111− -- Here we protect from blocking other threads by running on a separate (OS) thread:
112119 mv <- onOSThread$ tryUntilNoIOErr $
113120 -- 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
115122
116123 return $
117124 -- Write to the named pipe. If the message is less than
@@ -128,31 +135,26 @@ mkTransport = do
128135-- let finalmsg = BS.concat (encode msgsize : bss)
129136 let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)
130137
131− fd <- readMVar mv -- Synchronnize with file opening.
138+ fd <- readMVar mv -- Synchronize with file opening.
132139 ----------------------------------------
133140 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
134141 unless (fromIntegral cnt == BS.length finalmsg) $
135142 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
136143 ----------------------------------------
137144 return ()
138− , closeSourceEnd = error "Pipes.hs: closeSourceEnd not yet implemented"
145+ , closeSourceEnd = do
146+ fd <- readMVar mv
147+ PIO.closeFd fd
139148 }
140149
141− mkTargetEnd :: String -> MVar () -> TargetEnd
142− mkTargetEnd filename lock = TargetEnd
150+ mkTargetEnd :: MVar Fd -> MVar () -> TargetEnd
151+ mkTargetEnd mv lock = TargetEnd
143152 { receive = do
153+ fd <- readMVar mv -- Make sure Fd is there before
154+
144155 -- This should only happen on a single process. But it may
145156 -- happen on multiple threads so we grab a lock.
146157 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)
156158
157159 let spinread :: Int64 -> IO BS.ByteString
158160 spinread desired = do
@@ -178,7 +180,9 @@ mkTransport = do
178180 Right size -> spinread (fromIntegral (size::Word32))
179181 putMVar lock ()
180182 return [payload]
181− , closeTargetEnd = error "Pipes.hs: closeTargetEnd not yet implemented"
183+ , closeTargetEnd = do
184+ fd <- readMVar mv
185+ PIO.closeFd fd
182186 }
183187
184188spinTillThere :: String -> IO ()
@@ -202,6 +206,10 @@ tryUntilNoIOErr action = mkBackoff >>= loop
202206 loop bkoff =
203207 handle (\ (e :: IOException) ->
204208 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
205213 loop bkoff) $
206214 action
207215