git.iko.soy/distributed-process
on master help sign in
added switch between lazy and strict
bb31a08Adam Foltzer
benchmarks/PingTCP.hs+26−7
compare with master
unifiedsplit
@@ -1,9 +1,13 @@
1+{-# LANGUAGE CPP #-}
2+
1module Main where
3module Main where
2
4
3import Control.Monad
5import Control.Monad
4import Criterion.Main (Benchmark, bench, defaultMain, nfIO, bgroup)
6import Criterion.Main (Benchmark, bench, defaultMain, nfIO, bgroup)
5−import Data.Binary
7+
6import Data.Int
8import Data.Int
9+import qualified Data.Serialize as Ser
10+import Data.Word (Word8)
7import Network.Socket
11import Network.Socket
8 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
12 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
9 , SocketType (Stream), SocketOption (ReuseAddr)
13 , SocketType (Stream), SocketOption (ReuseAddr)
@@ -13,10 +17,25 @@ import Network.Socket
13import System.Environment (getArgs, withArgs)
17import System.Environment (getArgs, withArgs)
14
18
15import qualified Network.Socket as N
19import qualified Network.Socket as N
16−import qualified Network.Socket.ByteString.Lazy as NBS
17
20
18import Debug.Trace
21import Debug.Trace
19
22
23+#ifndef LAZY
24+import Data.ByteString (ByteString)
25+import qualified Network.Socket.ByteString as NBS
26+encode = Ser.encode
27+decode = Ser.decode
28+#else
29+import Data.ByteString.Lazy (ByteString)
30+import qualified Network.Socket.ByteString.Lazy as NBS
31+encode = Ser.encodeLazy
32+decode = Ser.decodeLazy
33+#endif
34+{-# INLINE encode #-}
35+{-# INLINE decode #-}
36+encode :: Ser.Serialize a => a -> ByteString
37+decode :: Ser.Serialize a => ByteString -> Either String a
38+
20-- | This performs a ping benchmark on a TCP connection created by
39-- | This performs a ping benchmark on a TCP connection created by
21-- Network.Socket. To compile this file, you might use:
40-- Network.Socket. To compile this file, you might use:
22--
41--
@@ -70,15 +89,15 @@ main = do
70
89
71-- | Each `ping` sends a single byte, and expects to receive one
90-- | Each `ping` sends a single byte, and expects to receive one
72-- back in return.
91-- back in return.
73−ping :: Socket -> IO Word8
92+ping :: Socket -> IO Int64
74ping sock = do
93ping sock = do
75− NBS.send sock $ encode (0 :: Word8)
94+ NBS.send sock $ encode (0 :: Int64)
76− bs <- NBS.recv sock 1
95+ bs <- NBS.recv sock 8
77− return $ decode bs
96+ either error return $ decode bs
78
97
79pong :: Socket -> IO ()
98pong :: Socket -> IO ()
80pong sock = do
99pong sock = do
81− bs <- NBS.recv sock 1
100+ bs <- NBS.recv sock 8
82 NBS.sendAll sock bs
101 NBS.sendAll sock bs
83 return ()
102 return ()
84
103
benchmarks/PingTransport.hs+19−1
@@ -1,3 +1,5 @@
1+{-# LANGUAGE CPP #-}
2+
1module Main where
3module Main where
2
4
3import Network.Transport
5import Network.Transport
@@ -5,12 +7,28 @@ import Network.Transport.TCP (mkTransport, TCPConfig (..))
5
7
6import Control.Monad (forever, replicateM_)
8import Control.Monad (forever, replicateM_)
7import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
9import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
8−import Data.Serialize
10+import qualified Data.Serialize as Ser
9import Data.Maybe (fromJust)
11import Data.Maybe (fromJust)
10import Data.Int
12import Data.Int
11import System.Environment (getArgs, withArgs)
13import System.Environment (getArgs, withArgs)
12
14
15+#ifndef LAZY
13import qualified Data.ByteString.Char8 as BS
16import qualified Data.ByteString.Char8 as BS
17+import qualified Network.Socket.ByteString as NBS
18+encode = Ser.encode
19+decode = Ser.decode
20+#else
21+import qualified Data.ByteString.Lazy.Char8 as BS
22+import qualified Network.Socket.ByteString.Lazy as NBS
23+encode = Ser.encodeLazy
24+decode = Ser.decodeLazy
25+#endif
26+{-# INLINE encode #-}
27+{-# INLINE decode #-}
28+encode :: Ser.Serialize a => a -> BS.ByteString
29+decode :: Ser.Serialize a => BS.ByteString -> Either String a
30+
31+
14
32
15-- | This performs a ping benchmark on the TCP transport. This can be
33-- | This performs a ping benchmark on the TCP transport. This can be
16-- compiled using:
34-- compiled using:
benchmarks/SendTCP.hs+23−5
@@ -1,10 +1,12 @@
1+{-# LANGUAGE CPP #-}
2+
1module Main where
3module Main where
2
4
3import Control.Monad
5import Control.Monad
4import Criterion.Main (Benchmark, bench, defaultMain, nfIO, bgroup)
6import Criterion.Main (Benchmark, bench, defaultMain, nfIO, bgroup)
5−import Data.Binary
6−import Data.ByteString.Lazy (ByteString)
7import Data.Int
7import Data.Int
8+import qualified Data.Serialize as Ser
9+import Data.Word (Word8)
8import Network.Socket
10import Network.Socket
9 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
11 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
10 , SocketType (Stream), SocketOption (ReuseAddr)
12 , SocketType (Stream), SocketOption (ReuseAddr)
@@ -13,9 +15,25 @@ import Network.Socket
13 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
15 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
14import System.Environment (getArgs, withArgs)
16import System.Environment (getArgs, withArgs)
15
17
16−import qualified Data.ByteString.Lazy as BS
17import qualified Network.Socket as N
18import qualified Network.Socket as N
19+
20+#ifdef LAZY
21+import Data.ByteString (ByteString)
22+import qualified Data.ByteString as BS
23+import qualified Network.Socket.ByteString as NBS
24+encode = Ser.encode
25+decode = Ser.decode
26+#else
27+import Data.ByteString.Lazy (ByteString)
28+import qualified Data.ByteString.Lazy as BS
18import qualified Network.Socket.ByteString.Lazy as NBS
29import qualified Network.Socket.ByteString.Lazy as NBS
30+encode = Ser.encodeLazy
31+decode = Ser.decodeLazy
32+#endif
33+{-# INLINE encode #-}
34+{-# INLINE decode #-}
35+encode :: Ser.Serialize a => a -> ByteString
36+decode :: Ser.Serialize a => ByteString -> Either String a
19
37
20-- | This performs a benchmark on a TCP connection to measure how long it takes
38-- | This performs a benchmark on a TCP connection to measure how long it takes
21-- to transfer a number of bytes.
39-- to transfer a number of bytes.
@@ -78,9 +96,9 @@ ping :: Socket -> ByteString -> IO Word8
78ping sock bs = do
96ping sock bs = do
79 NBS.send sock bs
97 NBS.send sock bs
80 bs' <- NBS.recv sock 1
98 bs' <- NBS.recv sock 1
81− return $ decode bs
99+ either error return $ decode bs
82
100
83−pong :: Socket -> Int64 -> IO ()
101+-- pong :: Socket -> Int64 -> IO ()
84pong sock size = do
102pong sock size = do
85 bs <- NBS.recv sock size
103 bs <- NBS.recv sock size
86 NBS.sendAll sock (encode (0 :: Word8))
104 NBS.sendAll sock (encode (0 :: Word8))
benchmarks/SendTransport.hs+22−6
@@ -1,19 +1,37 @@
1+{-# LANGUAGE CPP #-}
2+{-# LANGUAGE NoMonomorphismRestriction #-}
3+
1module Main where
4module Main where
2
5
3import Network.Transport
6import Network.Transport
4import Network.Transport.TCP (mkTransport, TCPConfig (..))
7import Network.Transport.TCP (mkTransport, TCPConfig (..))
5
8
6−import Control.DeepSeq (deepseq)
9+import Control.Applicative
7import Control.Monad (forever, replicateM, replicateM_)
10import Control.Monad (forever, replicateM, replicateM_)
8import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
11import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
9−import Data.Serialize
12+import qualified Data.Serialize as Ser
10−import Data.ByteString (ByteString)
11import Data.Maybe (fromJust)
13import Data.Maybe (fromJust)
12import Data.Int
14import Data.Int
13import System.Environment (getArgs, withArgs)
15import System.Environment (getArgs, withArgs)
14import System.Random
16import System.Random
15
17
18+#ifndef LAZY
19+import Data.ByteString (ByteString)
16import qualified Data.ByteString.Char8 as BS
20import qualified Data.ByteString.Char8 as BS
21+import qualified Network.Socket.ByteString as NBS
22+encode = Ser.encode
23+decode = Ser.decode
24+#else
25+import Data.ByteString.Lazy (ByteString)
26+import qualified Data.ByteString.Lazy.Char8 as BS
27+import qualified Network.Socket.ByteString.Lazy as NBS
28+encode = Ser.encodeLazy
29+decode = Ser.decodeLazy
30+#endif
31+{-# INLINE encode #-}
32+{-# INLINE decode #-}
33+encode :: Ser.Serialize a => a -> ByteString
34+decode :: Ser.Serialize a => ByteString -> Either String a
17
35
18-- | This performs a benchmark on the TCP transport to measure how long
36-- | This performs a benchmark on the TCP transport to measure how long
19-- it takes to transfer a number of bytes. This can be compiled using:
37-- it takes to transfer a number of bytes. This can be compiled using:
@@ -69,9 +87,7 @@ main = do
69 send sourceEndPing [serialize sourceAddrPong]
87 send sourceEndPing [serialize sourceAddrPong]
70
88
71 -- benchmark the data
89 -- benchmark the data
72− rands <- replicateM size randomIO
90+ bs <- BS.pack <$> replicateM size randomIO
73− let bs' = BS.pack (rands `deepseq` (rands :: [Char]))
74− bs = BS.unpack bs' `deepseq` bs'
75 withArgs args' $ defaultMain [ benchSend sourceEndPing targetEndPong bs ]
91 withArgs args' $ defaultMain [ benchSend sourceEndPing targetEndPong bs ]
76
92
77-- | The effect of `ping sourceEndPing targetEndPong bs` is to send the
93-- | The effect of `ping sourceEndPing targetEndPong bs` is to send the
network-transport/network-transport.cabal−2
@@ -14,10 +14,8 @@ Library
14 binary >= 0.5,
14 binary >= 0.5,
15 bytestring >= 0.9,
15 bytestring >= 0.9,
16 cereal >= 0.3,
16 cereal >= 0.3,
17− conduit >= 0.2,
18 containers >= 0.4,
17 containers >= 0.4,
19 network >= 2.3,
18 network >= 2.3,
20− network-conduit >= 0.2,
21 safe >= 0.3
19 safe >= 0.3
22 Exposed-modules: Network.Transport,
20 Exposed-modules: Network.Transport,
23 Network.Transport.MVar,
21 Network.Transport.MVar,
network-transport/src/Network/Transport.hs+6
@@ -1,3 +1,5 @@
1+{-# LANGUAGE CPP #-}
2+
1module Network.Transport
3module Network.Transport
2 ( Hints (..)
4 ( Hints (..)
3 , TargetEnd (..)
5 , TargetEnd (..)
@@ -12,7 +14,11 @@ module Network.Transport
12 , newMulticast
14 , newMulticast
13 ) where
15 ) where
14
16
17+#ifndef LAZY
15import Data.ByteString.Char8 (ByteString)
18import Data.ByteString.Char8 (ByteString)
19+#else
20+import Data.ByteString.Lazy.Char8 (ByteString)
21+#endif
16
22
17-- | The `Hints` and `SourceHints` provide hints to the underlying transport
23-- | The `Hints` and `SourceHints` provide hints to the underlying transport
18-- about the kind of connection that is required. This might include details
24-- about the kind of connection that is required. This might include details
network-transport/src/Network/Transport/MVar.hs+18−3
@@ -1,4 +1,5 @@
1{-# LANGUAGE BangPatterns #-}
1{-# LANGUAGE BangPatterns #-}
2+{-# LANGUAGE CPP #-}
2
3
3module Network.Transport.MVar
4module Network.Transport.MVar
4 ( mkTransport
5 ( mkTransport
@@ -6,14 +7,28 @@ module Network.Transport.MVar
6
7
7import Control.Concurrent.MVar
8import Control.Concurrent.MVar
8import Data.IntMap (IntMap)
9import Data.IntMap (IntMap)
9−import Data.ByteString (ByteString)
10−import Data.Serialize
10+import qualified Data.Serialize as Ser
11
11
12import qualified Data.IntMap as IntMap
12import qualified Data.IntMap as IntMap
13−import qualified Data.ByteString as BS
14
13
15import Network.Transport
14import Network.Transport
16
15
16+#ifndef LAZY
17+import Data.ByteString.Char8 (ByteString)
18+import qualified Data.ByteString.Char8 as BS
19+encode = Ser.encode
20+decode = Ser.decode
21+#else
22+import Data.ByteString.Lazy.Char8 (ByteString)
23+import qualified Data.ByteString.Lazy.Char8 as BS
24+encode = Ser.encodeLazy
25+decode = Ser.decodeLazy
26+#endif
27+{-# INLINE encode #-}
28+{-# INLINE decode #-}
29+encode :: Ser.Serialize a => a -> ByteString
30+decode :: Ser.Serialize a => ByteString -> Either String a
31+
17type Chans = MVar (Int, IntMap (MVar [ByteString]))
32type Chans = MVar (Int, IntMap (MVar [ByteString]))
18
33
19mkTransport :: IO Transport
34mkTransport :: IO Transport
network-transport/src/Network/Transport/TCP.hs+51−23
@@ -1,3 +1,5 @@
1+{-# LANGUAGE CPP #-}
2+
1module Network.Transport.TCP
3module Network.Transport.TCP
2 ( mkTransport
4 ( mkTransport
3 , TCPConfig (..)
5 , TCPConfig (..)
@@ -12,19 +14,18 @@ import Control.Concurrent.Chan
12import Control.Concurrent.MVar
14import Control.Concurrent.MVar
13import Control.Monad (forever, forM_)
15import Control.Monad (forever, forM_)
14
16
15−import Data.ByteString.Char8 (ByteString)
16−import qualified Data.ByteString.Char8 as BS
17−import Data.ByteString.Internal
17+import qualified Data.ByteString.Internal as BSI
18
18
19import Data.IntMap (IntMap)
19import Data.IntMap (IntMap)
20import qualified Data.IntMap as IntMap
20import qualified Data.IntMap as IntMap
21
21
22−import Data.Int
22+import Data.Int (Int32)
23−import Data.Serialize
23+import qualified Data.Serialize as Ser
24import Data.Word
24import Data.Word
25
25
26import Foreign.ForeignPtr (withForeignPtr)
26import Foreign.ForeignPtr (withForeignPtr)
27import Foreign.Ptr (Ptr, plusPtr)
27import Foreign.Ptr (Ptr, plusPtr)
28+import Foreign.Storable (sizeOf)
28
29
29import Network.Socket
30import Network.Socket
30 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
31 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
@@ -32,14 +33,30 @@ import Network.Socket
32 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
33 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
33 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
34 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
34import qualified Network.Socket as N
35import qualified Network.Socket as N
35−import qualified Network.Socket.ByteString as NBS
36
36
37import Text.Printf
37import Text.Printf
38
38
39import Safe
39import Safe
40
40
41−import qualified Data.Binary as B
41+#ifndef LAZY
42−
42+import Data.ByteString.Char8 (ByteString)
43+import qualified Data.ByteString.Char8 as BS
44+import qualified Network.Socket.ByteString as NBS
45+sendNBS = NBS.sendMany
46+encode = Ser.encode
47+decode = Ser.decode
48+#else
49+import Data.ByteString.Lazy.Char8 (ByteString)
50+import qualified Data.ByteString.Lazy.Char8 as BS
51+import qualified Network.Socket.ByteString.Lazy as NBS
52+sendNBS sock bs = NBS.sendAll sock $! head bs
53+encode = Ser.encodeLazy
54+decode = Ser.decodeLazy
55+#endif
56+{-# INLINE encode #-}
57+{-# INLINE decode #-}
58+encode :: Ser.Serialize a => a -> ByteString
59+decode :: Ser.Serialize a => ByteString -> Either String a
43
60
44type ChanId = Int
61type ChanId = Int
45type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
62type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
@@ -112,13 +129,13 @@ mkSourceEnd host service chanId = withSocketsDo $ do
112 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
129 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
113 setSocketOption sock ReuseAddr 1
130 setSocketOption sock ReuseAddr 1
114 N.connect sock (addrAddress serverAddr)
131 N.connect sock (addrAddress serverAddr)
115− NBS.sendAll sock $ encode (fromIntegral chanId :: Int64)
132+ NBS.sendAll sock $ encode (fromIntegral chanId :: Int32)
116 return SourceEnd
133 return SourceEnd
117 { send = {-# SCC "send" #-} \bss -> do
134 { send = {-# SCC "send" #-} \bss -> do
118− let size = fromIntegral (sum . map BS.length $ bss) :: Int64
135+ let size = fromIntegral (sum . map BS.length $ bss) :: Int32
119 sizeStr | size < 255 = encode (fromIntegral size :: Word8)
136 sizeStr | size < 255 = encode (fromIntegral size :: Word8)
120 | otherwise = BS.cons (toEnum 255) (encode size)
137 | otherwise = BS.cons (toEnum 255) (encode size)
121− NBS.sendMany sock (sizeStr:bss)
138+ sendNBS sock (sizeStr:bss)
122 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
139 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
123 }
140 }
124
141
@@ -145,11 +162,11 @@ procConnections chans sock = forever $ do
145 let err m = error $ printf "procConnections: %s" m
162 let err m = error $ printf "procConnections: %s" m
146 (clientSock, _clientAddr) <- accept sock
163 (clientSock, _clientAddr) <- accept sock
147 -- decode the first message to find the correct chanId
164 -- decode the first message to find the correct chanId
148− bss <- recvExact clientSock 8
165+ bss <- recvExact clientSock (fromIntegral $ sizeOf (0 :: Int32))
149 case bss of
166 case bss of
150 Left _ -> err "inbound chanId aborted"
167 Left _ -> err "inbound chanId aborted"
151 Right bs -> do
168 Right bs -> do
152− let chanId = either err fromIntegral (decode bs :: Either String Int64)
169+ let chanId = either err fromIntegral (decode bs :: Either String Int32)
153 (chanId', chanMap) <- takeMVar chans
170 (chanId', chanMap) <- takeMVar chans
154 case IntMap.lookup chanId chanMap of
171 case IntMap.lookup chanId chanMap of
155 Nothing -> do
172 Nothing -> do
@@ -175,7 +192,7 @@ procMessages chans chanId chan sock = do
175 Right sizeBS -> do
192 Right sizeBS -> do
176 case decode sizeBS :: Either String Word8 of
193 case decode sizeBS :: Either String Word8 of
177 Right 255 -> do
194 Right 255 -> do
178− esizeBS' <- recvExact sock 8
195+ esizeBS' <- recvExact sock (fromIntegral $ sizeOf (0 :: Int32))
179 case esizeBS' of
196 case esizeBS' of
180 Left _ -> closeSocket
197 Left _ -> closeSocket
181 Right sizeBS' -> procMessage $ decode sizeBS'
198 Right sizeBS' -> procMessage $ decode sizeBS'
@@ -194,7 +211,7 @@ procMessages chans chanId chan sock = do
194 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
211 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
195 putMVar chans (chanId', chanMap')
212 putMVar chans (chanId', chanMap')
196 sClose sock
213 sClose sock
197− procMessage :: Either String Int64 -> IO ()
214+ procMessage :: Either String Int32 -> IO ()
198 procMessage (Left msg) = err msg
215 procMessage (Left msg) = err msg
199 procMessage (Right size) = do
216 procMessage (Right size) = do
200 ebs <- recvExact sock size
217 ebs <- recvExact sock size
@@ -216,22 +233,33 @@ procMessages chans chanId chan sock = do
216-- a partial retrieval since the socket was closed, and `return Right bs`
233-- a partial retrieval since the socket was closed, and `return Right bs`
217-- indicates success. This hasn't been implemented, since a Transport `receive`
234-- indicates success. This hasn't been implemented, since a Transport `receive`
218-- represents an atomic receipt of a message.
235-- represents an atomic receipt of a message.
236+#ifndef LAZY
219−recvExact :: Socket -> Int64 -> IO (Either ByteString ByteString)
237+recvExact :: Socket -> Int32 -> IO (Either ByteString ByteString)
220recvExact sock l = do
238recvExact sock l = do
221− res <- createAndTrim (fromIntegral l) (go 0)
239+ res <- BSI.createAndTrim (fromIntegral l) (go 0)
222 case BS.length res of
240 case BS.length res of
223 n | n == (fromIntegral l) -> return $ Right res
241 n | n == (fromIntegral l) -> return $ Right res
224− n -> do
225− printf "recvExact: expected %dB, got %dB\n" l n
226− return $ Left res
242+ n -> return $ Left res
227 where
243 where
228 go :: Int -> Ptr Word8 -> IO Int
244 go :: Int -> Ptr Word8 -> IO Int
229 go n ptr | n == (fromIntegral l) = return n
245 go n ptr | n == (fromIntegral l) = return n
230 go n ptr = do
246 go n ptr = do
231− (p, off, len) <- toForeignPtr <$> NBS.recv sock (fromIntegral l)
247+ (p, off, len) <- BSI.toForeignPtr <$> NBS.recv sock (min (fromIntegral l-n) 4096)
232 if len == 0
248 if len == 0
233 then return n
249 then return n
234 else withForeignPtr p $ \p -> do
250 else withForeignPtr p $ \p -> do
235− memcpy (ptr `plusPtr` n) (p `plusPtr` off) (fromIntegral len)
251+ BSI.memcpy (ptr `plusPtr` n) (p `plusPtr` off) (fromIntegral len)
236 go (n+len) ptr
252 go (n+len) ptr
237−
253+#else
254+recvExact :: Socket -> Int32 -> IO (Either ByteString ByteString)
255+recvExact sock n =
256+ go [] sock n
257+ where
258+ go :: [ByteString] -> Socket -> Int32 -> IO (Either ByteString ByteString)
259+ go bss _ 0 = return (Right $ BS.concat (reverse bss))
260+ go bss sock n = do
261+ bs <- NBS.recv sock (min (fromIntegral n) 4096)
262+ if BS.null bs
263+ then return (Left $ BS.concat (reverse bss))
264+ else go (bs:bss) sock (n - (fromIntegral $ BS.length bs))
265+#endif