git.iko.soy/distributed-process
on master help sign in
Merge pull request #5 from acfoltzer/master
acbc141Nicolas Wu
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--
@@ -71,15 +90,15 @@ main = do
71
90
72-- | Each `ping` sends a single byte, and expects to receive one
91-- | Each `ping` sends a single byte, and expects to receive one
73-- back in return.
92-- back in return.
74−ping :: Socket -> IO Word8
93+ping :: Socket -> IO Int64
75ping sock = do
94ping sock = do
76− NBS.send sock $ encode (0 :: Word8)
95+ NBS.send sock $ encode (0 :: Int64)
77− bs <- NBS.recv sock 1
96+ bs <- NBS.recv sock 8
78− return $ decode bs
97+ either error return $ decode bs
79
98
80pong :: Socket -> IO ()
99pong :: Socket -> IO ()
81pong sock = do
100pong sock = do
82− bs <- NBS.recv sock 1
101+ bs <- NBS.recv sock 8
83 NBS.sendAll sock bs
102 NBS.sendAll sock bs
84 return ()
103 return ()
85
104
benchmarks/PingTransport.hs+27−8
@@ -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.Binary
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
16+import qualified Data.ByteString.Char8 as BS
17+import qualified Network.Socket.ByteString as NBS
18+encode = Ser.encode
19+decode = Ser.decode
20+#else
13import qualified Data.ByteString.Lazy.Char8 as BS
21import 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:
@@ -64,8 +82,8 @@ main = do
64 send sourceEndPing [serialize sourceAddrPong]
82 send sourceEndPing [serialize sourceAddrPong]
65
83
66 -- benchmark the pings
84 -- benchmark the pings
67−-- withArgs args' $ defaultMain [ benchPing sourceEndPing targetEndPong pings]
85+ withArgs args' $ defaultMain [ benchPing sourceEndPing targetEndPong pings]
68− replicateM_ pings (ping sourceEndPing targetEndPong)
86+-- replicateM_ pings (ping sourceEndPing targetEndPong)
69
87
70-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
88-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
71-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent
89-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent
@@ -78,16 +96,17 @@ pong targetEndPing sourceEndPong = do
78-- | The effect of `ping sourceEndPing targetEndPong n` is to send the number
96-- | The effect of `ping sourceEndPing targetEndPong n` is to send the number
79-- `n` using `sourceEndPing`, and to then receive the a number from
97-- `n` using `sourceEndPing`, and to then receive the a number from
80-- `targetEndPong`, which is then returned.
98-- `targetEndPong`, which is then returned.
81−ping :: SourceEnd -> TargetEnd -> IO Int64
99+ping :: SourceEnd -> TargetEnd -> Int64 -> IO Int64
82−ping sourceEndPing targetEndPong = do
100+ping sourceEndPing targetEndPong n = do
83− send sourceEndPing [BS.empty]
101+ send sourceEndPing [encode n]
84 [bs] <- receive targetEndPong
102 [bs] <- receive targetEndPong
103+ let (Right n) = decode bs
85− return $ decode bs
104+ return $! n
86
105
87-- | The effect of `benchPing sourceEndPing targetEndPong n` is to send
106-- | The effect of `benchPing sourceEndPing targetEndPong n` is to send
88-- `n` pings down `sourceEndPing` using the `ping` function. The time
107-- `n` pings down `sourceEndPing` using the `ping` function. The time
89-- taken is benchmarked.
108-- taken is benchmarked.
90benchPing :: SourceEnd -> TargetEnd -> Int64 -> Benchmark
109benchPing :: SourceEnd -> TargetEnd -> Int64 -> Benchmark
91benchPing sourceEndPing targetEndPong n = bench "PingTransport" $
110benchPing sourceEndPing targetEndPong n = bench "PingTransport" $
92− nfIO (replicateM_ (fromIntegral n) (ping sourceEndPing targetEndPong))
111+ nfIO (replicateM_ (fromIntegral n) (ping sourceEndPing targetEndPong 42))
93
112
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+31−11
@@ -1,17 +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
9+import Control.Applicative
6−import Control.Monad (forever, replicateM_)
10+import Control.Monad (forever, replicateM, replicateM_)
7import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
11import Criterion.Main (Benchmark, bench, defaultMain, nfIO)
8−import Data.Binary
12+import qualified Data.Serialize as Ser
9−import Data.ByteString.Lazy (ByteString)
10import Data.Maybe (fromJust)
13import Data.Maybe (fromJust)
11import Data.Int
14import Data.Int
12import System.Environment (getArgs, withArgs)
15import System.Environment (getArgs, withArgs)
13−
16+import System.Random
17+
18+#ifndef LAZY
19+import Data.ByteString (ByteString)
14−import qualified Data.ByteString.Lazy as BS
20+import 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
15
35
16-- | 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
17-- 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:
@@ -67,17 +87,17 @@ main = do
67 send sourceEndPing [serialize sourceAddrPong]
87 send sourceEndPing [serialize sourceAddrPong]
68
88
69 -- benchmark the data
89 -- benchmark the data
70− let bs = BS.replicate size 0
90+ bs <- BS.pack <$> replicateM size randomIO
71 withArgs args' $ defaultMain [ benchSend sourceEndPing targetEndPong bs ]
91 withArgs args' $ defaultMain [ benchSend sourceEndPing targetEndPong bs ]
72
92
73-- | The effect of `ping sourceEndPing targetEndPong bs` is to send the
93-- | The effect of `ping sourceEndPing targetEndPong bs` is to send the
74-- `ByteString` `bs` using `sourceEndPing`, and to then receive
94-- `ByteString` `bs` using `sourceEndPing`, and to then receive
75−-- a single byte back from `targetEndPong`.
95+-- that string back. Returns the length of the received message.
76−ping :: SourceEnd -> TargetEnd -> ByteString -> IO Word8
96+ping :: SourceEnd -> TargetEnd -> ByteString -> IO Int
77ping sourceEndPing targetEndPong bs = do
97ping sourceEndPing targetEndPong bs = do
78 send sourceEndPing [bs]
98 send sourceEndPing [bs]
79− [cs] <- receive targetEndPong
99+ cs <- receive targetEndPong
80− return $ decode cs
100+ return . fromIntegral $! sum (map BS.length cs)
81
101
82-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
102-- | This function takes a `TargetEnd` for the pings, and a `SourceEnd` for
83-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent
103-- pongs. Whenever a ping is received from the `TargetEnd`, a pong is sent
@@ -85,7 +105,7 @@ ping sourceEndPing targetEndPong bs = do
85pong :: TargetEnd -> SourceEnd -> IO ()
105pong :: TargetEnd -> SourceEnd -> IO ()
86pong targetEndPing sourceEndPong = do
106pong targetEndPing sourceEndPong = do
87 bs <- receive targetEndPing
107 bs <- receive targetEndPing
88− send sourceEndPong [encode (0 :: Word8)]
108+ send sourceEndPong bs
89
109
90-- | The effect of `benchSend sourceEndPing targetEndPong bs` is to send
110-- | The effect of `benchSend sourceEndPing targetEndPong bs` is to send
91-- `bs` pings down `sourceEndPing` using the `ping` function. The time
111-- `bs` pings down `sourceEndPing` using the `ping` function. The time
examples/DemoTCP.hs+6−3
@@ -4,10 +4,12 @@ import Network.Transport
4import Network.Transport.TCP
4import Network.Transport.TCP
5
5
6import Control.Monad
6import Control.Monad
7+import Data.Char
8+import Data.String
7import Network.Socket (HostName, ServiceName)
9import Network.Socket (HostName, ServiceName)
8import System.Environment
10import System.Environment
9
11
10−import qualified Data.ByteString.Lazy.Char8 as BS
12+import qualified Data.ByteString.Char8 as BS
11
13
12import Debug.Trace
14import Debug.Trace
13
15
@@ -39,7 +41,8 @@ main = do
39 transport <- mkTransport $ TCPConfig defaultHints host service
41 transport <- mkTransport $ TCPConfig defaultHints host service
40 (clientSourceEnds, masterTargetEnd) <- demoMaster transport sourceAddrFilePath numSlaves
42 (clientSourceEnds, masterTargetEnd) <- demoMaster transport sourceAddrFilePath numSlaves
41 zipWithM_
43 zipWithM_
44+ (\clientSourceEnd clientId ->
42− (\clientSourceEnd clientId -> send clientSourceEnd [BS.pack . show $ clientId])
45+ send clientSourceEnd [BS.pack . show $ clientId])
43 clientSourceEnds [0 .. numSlaves-1]
46 clientSourceEnds [0 .. numSlaves-1]
44 replicateM_ numSlaves $ do
47 replicateM_ numSlaves $ do
45 [clientMessage] <- receive masterTargetEnd
48 [clientMessage] <- receive masterTargetEnd
@@ -56,7 +59,7 @@ main = do
56 (masterSourceEnd, slaveTargetEnd) <- demoSlave transport sourceAddrFilePath
59 (masterSourceEnd, slaveTargetEnd) <- demoSlave transport sourceAddrFilePath
57 [clientId] <- receive slaveTargetEnd
60 [clientId] <- receive slaveTargetEnd
58 putStrLn $ "slave: received clientId: " ++ show clientId
61 putStrLn $ "slave: received clientId: " ++ show clientId
59− let message = "Connected to slave: " ++ BS.unpack clientId
62+ let message = "Connected to slave: " ++ (BS.unpack clientId)
60
63
61 send masterSourceEnd [BS.pack message]
64 send masterSourceEnd [BS.pack message]
62 putStrLn "slave: sent message to master"
65 putStrLn "slave: sent message to master"
network-transport-pipes/src/Network/Transport/Pipes.hs+12−13
@@ -25,8 +25,7 @@ import Data.Word
25import Data.Int
25import Data.Int
26import Data.IORef
26import Data.IORef
27import qualified Data.IntMap as IntMap
27import qualified Data.IntMap as IntMap
28−import qualified Data.ByteString.Char8 as BSS
28+import qualified Data.ByteString.Char8 as BS
29−import qualified Data.ByteString.Lazy.Char8 as BS
30
29
31-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:
30-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:
32#define CEREAL
31#define CEREAL
@@ -54,18 +53,18 @@ import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
54import System.Posix.IO as PIO (openFd, closeFd, -- append, exclusive, noctty, nonBlock, trunc,
53import System.Posix.IO as PIO (openFd, closeFd, -- append, exclusive, noctty, nonBlock, trunc,
55 OpenFileFlags(..), OpenMode(ReadOnly, WriteOnly))
54 OpenFileFlags(..), OpenMode(ReadOnly, WriteOnly))
56-- (fromS,toS) = (BS.pack, BS.unpack)
55-- (fromS,toS) = (BS.pack, BS.unpack)
57−(fromS,toS) = (BSS.pack, BSS.unpack)
56+(fromS,toS) = (BS.pack, BS.unpack)
58fromBS = id
57fromBS = id
59readit fd n = PIO.fdRead fd n
58readit fd n = PIO.fdRead fd n
60#else
59#else
61import qualified System.Posix.IO as PIO
60import qualified System.Posix.IO as PIO
62(toS,fromS) = (id,id)
61(toS,fromS) = (id,id)
63−fromBS = BSS.unpack
62+fromBS = BS.unpack
64readit fd n = do (s,_) <- PIO.fdRead fd n
63readit fd n = do (s,_) <- PIO.fdRead fd n
65− return (BSS.pack s)
64+ return (BS.pack s)
66#endif
65#endif
67-- readit :: Fd -> Int -> IO BS.ByteString
66-- readit :: Fd -> Int -> IO BS.ByteString
68−readit :: Fd -> ByteCount -> IO BSS.ByteString
67+readit :: Fd -> ByteCount -> IO BS.ByteString
69
68
70----------------------------------------------------------------------------------------------------
69----------------------------------------------------------------------------------------------------
71
70
@@ -135,7 +134,7 @@ mkTransport = do
135 -- Otherwise it's just a simple write:
134 -- Otherwise it's just a simple write:
136 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
135 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
137
136
138− let hdrbss :: BSS.ByteString
137+ let hdrbss :: BS.ByteString
139#ifdef CEREAL
138#ifdef CEREAL
140 hdrbss = encode msgsize
139 hdrbss = encode msgsize
141#else
140#else
@@ -144,13 +143,13 @@ mkTransport = do
144#endif
143#endif
145
144
146-- let finalmsg = BS.concat (encode msgsize : bss)
145-- let finalmsg = BS.concat (encode msgsize : bss)
147− let finalmsg = BSS.concat (hdrbss : concatMap BS.toChunks bsls)
146+ let finalmsg = BS.concat (hdrbss : bsls)
148
147
149 fd <- readMVar mv -- Synchronize with file opening.
148 fd <- readMVar mv -- Synchronize with file opening.
150 ----------------------------------------
149 ----------------------------------------
151 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
150 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
152− unless (fromIntegral cnt == BSS.length finalmsg) $
151+ unless (fromIntegral cnt == BS.length finalmsg) $
153− error$ "Failed to write message in one go, length: "++ show (BSS.length finalmsg)
152+ error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
154 ----------------------------------------
153 ----------------------------------------
155 return ()
154 return ()
156 , closeSourceEnd = do
155 , closeSourceEnd = do
@@ -167,13 +166,13 @@ mkTransport = do
167 -- happen on multiple threads so we grab a lock.
166 -- happen on multiple threads so we grab a lock.
168 takeMVar lock
167 takeMVar lock
169
168
170− let spinread :: Int -> IO BSS.ByteString
169+ let spinread :: Int -> IO BS.ByteString
171 spinread desired = do
170 spinread desired = do
172
171
173 bs <- tryUntilNoIOErr$
172 bs <- tryUntilNoIOErr$
174 readit fd (fromIntegral desired)
173 readit fd (fromIntegral desired)
175
174
176− case BSS.length bs of
175+ case BS.length bs of
177 n | n == desired -> return bs
176 n | n == desired -> return bs
178 0 -> do threadDelay (10*1000)
177 0 -> do threadDelay (10*1000)
179 spinread desired
178 spinread desired
@@ -193,7 +192,7 @@ mkTransport = do
193 Left err -> error$ "ERROR: "++ err
192 Left err -> error$ "ERROR: "++ err
194 Right size -> spinread (fromIntegral (size::Word32))
193 Right size -> spinread (fromIntegral (size::Word32))
195 putMVar lock ()
194 putMVar lock ()
196− return [BS.fromChunks [payload]] -- How terribly listy.
195+ return [payload] -- How terribly listy.
197 , closeTargetEnd = do
196 , closeTargetEnd = do
198 fd <- readMVar mv
197 fd <- readMVar mv
199 PIO.closeFd fd
198 PIO.closeFd fd
network-transport/network-transport.cabal+1
@@ -13,6 +13,7 @@ Library
13 Build-Depends: base >= 3 && < 5,
13 Build-Depends: base >= 3 && < 5,
14 binary >= 0.5,
14 binary >= 0.5,
15 bytestring >= 0.9,
15 bytestring >= 0.9,
16+ cereal >= 0.3,
16 containers >= 0.4,
17 containers >= 0.4,
17 network >= 2.3,
18 network >= 2.3,
18 safe >= 0.3,
19 safe >= 0.3,
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
18+import Data.ByteString.Char8 (ByteString)
19+#else
15import Data.ByteString.Lazy.Char8 (ByteString)
20import 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+22−6
@@ -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,13 +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.Lazy.Char8 (ByteString)
10+import qualified Data.Serialize as Ser
10
11
11import qualified Data.IntMap as IntMap
12import qualified Data.IntMap as IntMap
12−import qualified Data.ByteString.Lazy.Char8 as BS
13
13
14import Network.Transport
14import Network.Transport
15
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+
16type Chans = MVar (Int, IntMap (MVar [ByteString]))
32type Chans = MVar (Int, IntMap (MVar [ByteString]))
17
33
18mkTransport :: IO Transport
34mkTransport :: IO Transport
@@ -29,16 +45,16 @@ mkTransport = do
29 return (mkSourceAddr channels sourceAddr, mkTargetEnd receiveChan)
45 return (mkSourceAddr channels sourceAddr, mkTargetEnd receiveChan)
30 , newMulticastWith = undefined
46 , newMulticastWith = undefined
31 , deserialize = \bs ->
47 , deserialize = \bs ->
32− case BS.readInt bs of
33− Nothing -> error "dummyBackend.deserializeSourceEnd: cannot parse"
48+ either (error "dummyBackend.deserializeSourceEnd: cannot parse")
34− Just (n,_) -> Just . mkSourceAddr channels $ n
49+ (Just . mkSourceAddr channels)
50+ (decode bs)
35 , closeTransport = return ()
51 , closeTransport = return ()
36 }
52 }
37 where
53 where
38 mkSourceAddr :: Chans -> Int -> SourceAddr
54 mkSourceAddr :: Chans -> Int -> SourceAddr
39 mkSourceAddr channels addr = SourceAddr
55 mkSourceAddr channels addr = SourceAddr
40 { connectWith = \_ -> mkSourceEnd channels addr
56 { connectWith = \_ -> mkSourceEnd channels addr
41− , serialize = BS.pack (show addr)
57+ , serialize = encode addr
42 }
58 }
43
59
44 mkSourceEnd :: Chans -> Int -> IO SourceEnd
60 mkSourceEnd :: Chans -> Int -> IO SourceEnd
network-transport/src/Network/Transport/TCP.hs+112−69
@@ -1,3 +1,5 @@
1+{-# LANGUAGE CPP #-}
2+
1module Network.Transport.TCP
3module Network.Transport.TCP
2 ( mkTransport
4 ( mkTransport
3 , TCPConfig (..)
5 , TCPConfig (..)
@@ -5,30 +7,58 @@ module Network.Transport.TCP
5
7
6import Network.Transport
8import Network.Transport
7
9
10+import Control.Applicative
8import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
11import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)
9import Control.Concurrent.Chan
12import Control.Concurrent.Chan
10import Control.Concurrent.MVar
13import Control.Concurrent.MVar
11import Control.Exception (SomeException, IOException, AsyncException(ThreadKilled),
14import Control.Exception (SomeException, IOException, AsyncException(ThreadKilled),
12 fromException, throwTo, throw, catch, handle)
15 fromException, throwTo, throw, catch, handle)
13import Control.Monad (forever, forM_)
16import Control.Monad (forever, forM_)
17+
14−import Data.ByteString.Lazy.Char8 (ByteString)
18+import qualified Data.ByteString.Internal as BSI
19+
15import Data.IntMap (IntMap)
20import Data.IntMap (IntMap)
21+import qualified Data.IntMap as IntMap
22+
16−import Data.Int
23+import Data.Int (Int32)
24+import qualified Data.Serialize as Ser
17−import Data.Word
25+import Data.Word (Word8)
26+
27+import Foreign.ForeignPtr (withForeignPtr)
28+import Foreign.Ptr (Ptr, plusPtr)
29+import Foreign.Storable (sizeOf)
30+
18import Network.Socket
31import Network.Socket
19 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
32 ( AddrInfoFlag (AI_PASSIVE), HostName, ServiceName, Socket
20 , SocketType (Stream), SocketOption (ReuseAddr)
33 , SocketType (Stream), SocketOption (ReuseAddr)
21 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
34 , accept, addrAddress, addrFlags, addrFamily, bindSocket, defaultProtocol
22 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
35 , getAddrInfo, listen, setSocketOption, socket, sClose, withSocketsDo )
36+import qualified Network.Socket as N
37+
38+import Text.Printf
39+
23import Safe
40import Safe
24import System.IO (stderr, hPutStrLn)
41import System.IO (stderr, hPutStrLn)
25
42
43+#ifndef LAZY
44+import Data.ByteString.Char8 (ByteString)
26−import qualified Data.Binary as B
45+import qualified Data.ByteString.Char8 as BS
46+import qualified Network.Socket.ByteString as NBS
47+sendNBS = NBS.sendMany
48+encode = Ser.encode
49+decode = Ser.decode
50+#else
27−import qualified Data.ByteString.Char8 as BSS
51+import Data.ByteString.Lazy.Char8 (ByteString)
28import qualified Data.ByteString.Lazy.Char8 as BS
52import qualified Data.ByteString.Lazy.Char8 as BS
29−import qualified Data.IntMap as IntMap
30−import qualified Network.Socket as N
31import qualified Network.Socket.ByteString.Lazy as NBS
53import qualified Network.Socket.ByteString.Lazy as NBS
54+sendNBS sock = NBS.sendAll sock . BS.concat
55+encode = Ser.encodeLazy
56+decode = Ser.decodeLazy
57+#endif
58+{-# INLINE encode #-}
59+{-# INLINE decode #-}
60+encode :: Ser.Serialize a => a -> ByteString
61+decode :: Ser.Serialize a => ByteString -> Either String a
32
62
33type ChanId = Int
63type ChanId = Int
34type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
64type Chans = MVar (ChanId, IntMap (Chan [ByteString], [(ThreadId, Socket)]))
@@ -70,7 +100,9 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
70 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
100 return (mkSourceAddr host service chanId, mkTargetEnd chans chanId chan)
71 , newMulticastWith = error "newMulticastWith: not defined"
101 , newMulticastWith = error "newMulticastWith: not defined"
72 , deserialize = {-# SCC "deserialize" #-} \bs ->
102 , deserialize = {-# SCC "deserialize" #-} \bs ->
73− let (host, service, chanId) = B.decode bs in
103+ let (host, service, chanId) =
104+ either (\m -> error $ printf "deserialize: %s" m)
105+ id $ decode bs in
74 Just $ mkSourceAddr host service chanId
106 Just $ mkSourceAddr host service chanId
75 , closeTransport = {-# SCC "closeTransport" #-} do
107 , closeTransport = {-# SCC "closeTransport" #-} do
76 -- Kill the transport channel process
108 -- Kill the transport channel process
@@ -87,29 +119,26 @@ mkTransport (TCPConfig _hints host service) = withSocketsDo $ do
87mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
119mkSourceAddr :: HostName -> ServiceName -> ChanId -> SourceAddr
88mkSourceAddr host service chanId = SourceAddr
120mkSourceAddr host service chanId = SourceAddr
89 { connectWith = {-# SCC "connectWith" #-} \_ -> mkSourceEnd host service chanId
121 { connectWith = {-# SCC "connectWith" #-} \_ -> mkSourceEnd host service chanId
90− , serialize = {-# SCC "serialize" #-} B.encode (host, service, chanId)
122+ , serialize = {-# SCC "serialize" #-} encode (host, service, chanId)
91 }
123 }
92
124
93mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
125mkSourceEnd :: HostName -> ServiceName -> ChanId -> IO SourceEnd
94mkSourceEnd host service chanId = withSocketsDo $ do
126mkSourceEnd host service chanId = withSocketsDo $ do
127+ let err m = error $ printf "mkSourceEnd: %s" m
95 serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
128 serverAddrs <- getAddrInfo Nothing (Just host) (Just service)
96 let serverAddr = case serverAddrs of
129 let serverAddr = case serverAddrs of
97− [] -> error "mkSourceEnd: getAddrInfo returned []"
130+ [] -> err "getAddrInfo returned []"
98 as -> head as
131 as -> head as
99 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
132 sock <- socket (addrFamily serverAddr) Stream defaultProtocol
100 setSocketOption sock ReuseAddr 1
133 setSocketOption sock ReuseAddr 1
101 N.connect sock (addrAddress serverAddr)
134 N.connect sock (addrAddress serverAddr)
102− NBS.sendAll sock $ B.encode (fromIntegral chanId :: Int64)
135+ NBS.sendAll sock $ encode (fromIntegral chanId :: Int32)
103 return SourceEnd
136 return SourceEnd
104 { send = {-# SCC "send" #-} \bss -> do
137 { send = {-# SCC "send" #-} \bss -> do
105− let size = fromIntegral (sum . map BS.length $ bss) :: Int64
138+ let size = fromIntegral (sum . map BS.length $ bss) :: Int32
106− if size < 255
139+ sizeStr | size < 255 = BS.singleton (toEnum $ fromEnum size)
107− then
140+ | otherwise = BS.cons (toEnum 255) (encode size)
108− NBS.sendAll sock (B.encode (fromIntegral size :: Word8))
141+ sendNBS sock (sizeStr:bss)
109− else do
110− NBS.sendAll sock (B.encode (255 :: Word8))
111− NBS.sendAll sock (B.encode size)
112− mapM_ (NBS.sendAll sock) bss
113 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
142 , closeSourceEnd = {-# SCC "closeSourceEnd" #-} sClose sock
114 }
143 }
115
144
@@ -133,19 +162,19 @@ mkTargetEnd chans chanId chan = TargetEnd
133-- for some reason, an error is raised.
162-- for some reason, an error is raised.
134procConnections :: Chans -> Socket -> IO ()
163procConnections :: Chans -> Socket -> IO ()
135procConnections chans sock = forever $ do
164procConnections chans sock = forever $ do
165+ let err m = error $ printf "procConnections: %s" m
136 (clientSock, _clientAddr) <- accept sock
166 (clientSock, _clientAddr) <- accept sock
137 -- decode the first message to find the correct chanId
167 -- decode the first message to find the correct chanId
138− bss <- recvExact clientSock 8
168+ bss <- recvExact clientSock (fromIntegral $ sizeOf (0 :: Int32))
139 case bss of
169 case bss of
140− [] -> error "procConnections: inbound chanId aborted"
170+ Left _ -> err "inbound chanId aborted"
141− bss -> do
171+ Right bs -> do
142− let bs = BS.concat bss
143− let chanId = fromIntegral (B.decode bs :: Int64)
172+ let chanId = either err fromIntegral (decode bs :: Either String Int32)
144 (chanId', chanMap) <- takeMVar chans
173 (chanId', chanMap) <- takeMVar chans
145 case IntMap.lookup chanId chanMap of
174 case IntMap.lookup chanId chanMap of
146 Nothing -> do
175 Nothing -> do
147 putMVar chans (chanId', chanMap)
176 putMVar chans (chanId', chanMap)
148− error "procConnections: cannot find chanId"
177+ err "cannot find chanId"
149 Just (chan, socks) -> do
178 Just (chan, socks) -> do
150 threadId <- forkWithExceptions forkIO "Message Listener" $
179 threadId <- forkWithExceptions forkIO "Message Listener" $
151 procMessages chans chanId chan clientSock
180 procMessages chans chanId chan clientSock
@@ -154,80 +183,94 @@ procConnections chans sock = forever $ do
154
183
155-- | This function first extracts a header of type Word8, which determines
184-- | This function first extracts a header of type Word8, which determines
156-- the size of the ByteString that follows. If this size is 0, this indicates
185-- the size of the ByteString that follows. If this size is 0, this indicates
157−-- that the ByteString is large, so the next value is an Int64, which
186+-- that the ByteString is large, so the next value is an Int32, which
158-- determines the size of the ByteString that follows. The ByteString is then
187-- determines the size of the ByteString that follows. The ByteString is then
159-- extracted from the socket, and then written to the Chan only when
188-- extracted from the socket, and then written to the Chan only when
160-- complete. If either of the first header size is null this indicates the
189-- complete. If either of the first header size is null this indicates the
161-- socket has closed.
190-- socket has closed.
162procMessages :: Chans -> ChanId -> Chan [ByteString] -> Socket -> IO ()
191procMessages :: Chans -> ChanId -> Chan [ByteString] -> Socket -> IO ()
163procMessages chans chanId chan sock = do
192procMessages chans chanId chan sock = do
164− sizeBSs <- recvExact sock 1
193+ esizeBS <- recvExact sock 1
165− case sizeBSs of
194+ case esizeBS of
166− [] -> closeSocket
195+ Left _ -> closeSocket
167− [onebyte] -> do
196+ Right sizeBS -> do
168− let size = fromIntegral (B.decode onebyte :: Word8)
197+ case decode sizeBS :: Either String Word8 of
169− if size == 255
198+ Right 255 -> do
170− then do
171− sizeBSs' <- recvExact sock 8
199+ esizeBS' <- recvExact sock (fromIntegral $ sizeOf (0 :: Int32))
172− case sizeBSs' of
200+ case esizeBS' of
173− [] -> closeSocket
201+ Left _ -> closeSocket
174− _ -> procMessage (B.decode . BS.concat$ sizeBSs' :: Int64)
202+ Right sizeBS' -> procMessage $ decode sizeBS'
175− else procMessage size
203+ esize@(Right _) -> procMessage (fromIntegral <$> esize)
176− ls -> error "Shouldn't receive more than one bytestring when expecting a single byte!"
204+ Left msg -> error msg
177 where
205 where
206+ err m = error $ printf "procMessages: %s" m
178 closeSocket :: IO ()
207 closeSocket :: IO ()
179 closeSocket = do
208 closeSocket = do
180 (chanId', chanMap) <- takeMVar chans
209 (chanId', chanMap) <- takeMVar chans
181 case IntMap.lookup chanId chanMap of
210 case IntMap.lookup chanId chanMap of
182 Nothing -> do
211 Nothing -> do
183 putMVar chans (chanId', chanMap)
212 putMVar chans (chanId', chanMap)
184− error "procMessages: chanId not found."
213+ err "chanId not found."
185 Just (chan, socks) -> do
214 Just (chan, socks) -> do
186 let socks' = filter ((/= sock) . snd) socks
215 let socks' = filter ((/= sock) . snd) socks
187 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
216 let chanMap' = IntMap.insert chanId (chan, socks') chanMap
188 putMVar chans (chanId', chanMap')
217 putMVar chans (chanId', chanMap')
189 sClose sock
218 sClose sock
190− procMessage :: Int64 -> IO ()
219+ procMessage :: Either String Int32 -> IO ()
191− procMessage size = do
220+ procMessage (Left msg) = err msg
221+ procMessage (Right size) = do
192− bss <- recvExact sock size
222+ ebs <- recvExact sock size
193− case bss of
223+ case ebs of
194− [] -> closeSocket
224+ Left _ -> closeSocket
195− _ -> do
225+ Right bs -> do
196− writeChan chan bss
226+ writeChan chan [bs]
197 procMessages chans chanId chan sock
227 procMessages chans chanId chan sock
198
228
199−-- | The result of `recvExact sock n` is a `[ByteString]` whose concatenation
229+-- | The normal result of `recvExact sock n` is `Right ByteString`
200−-- is of length `n`, received from `sock`. No more bytes than necessary are
230+-- whose string of length `n`, received from `sock`. If fewer than `n`
231+-- bytes are read from `sock` before it closes, the result is `Left
232+-- ByteString` whose string is those bytes that were received. No more
201−-- read from the socket.
233+-- bytes than necessary are read from the socket. NB: This uses
202−-- NB: This uses Network.Socket.ByteString.recv, which may *discard*
234+-- Network.Socket.ByteString.recv, which may *discard* superfluous
203−-- superfluous input depending on the socket type. Also note that
235+-- input depending on the socket type.
204−-- if `recv` returns an empty `ByteString` then this means that the socket
236+#ifndef LAZY
205−-- was closed: in this case, we return an empty list.
237+recvExact :: Socket -> Int32 -> IO (Either ByteString ByteString)
206−-- NB: It may be appropriate to change the return type to
238+recvExact sock l = interceptAllExn "recvExact" $ do
207−-- IO (Either ByteString ByteString), where `return Left bs` indicates
239+ res <- BSI.createAndTrim (fromIntegral l) (go 0)
208−-- a partial retrieval since the socket was closed, and `return Right bs`
240+ case BS.length res of
209−-- indicates success. This hasn't been implemented, since a Transport `receive`
241+ n | n == (fromIntegral l) -> return $ Right res
210−-- represents an atomic receipt of a message.
242+ n -> return $ Left res
243+ where
244+ go :: Int -> Ptr Word8 -> IO Int
245+ go n ptr | n == (fromIntegral l) = return n
246+ go n ptr = do
247+ (p, off, len) <- BSI.toForeignPtr <$> NBS.recv sock (min (fromIntegral l-n) 4096)
248+ if len == 0
249+ then return n
250+ else withForeignPtr p $ \p -> do
251+ BSI.memcpy (ptr `plusPtr` n) (p `plusPtr` off) (fromIntegral len)
252+ go (n+len) ptr
253+#else
211−recvExact :: Socket -> Int64 -> IO [ByteString]
254+recvExact :: Socket -> Int32 -> IO (Either ByteString ByteString)
212recvExact sock n =
255recvExact sock n =
213− interceptAllExn "recvExact" $
256+ interceptAllExn "recvExact" $
214 go [] sock n
257 go [] sock n
215 where
258 where
216− go :: [ByteString] -> Socket -> Int64 -> IO [ByteString]
259+ go :: [ByteString] -> Socket -> Int32 -> IO (Either ByteString ByteString)
217− go bss _ 0 = return (reverse bss)
260+ go bss _ 0 = return (Right $ BS.concat (reverse bss))
218 go bss sock n = do
261 go bss sock n = do
219− bs <- NBS.recv sock n
262+ bs <- NBS.recv sock (min (fromIntegral n) 4096)
220 if BS.null bs
263 if BS.null bs
221− then return []
264+ then return (Left $ BS.concat (reverse bss))
222− else go (bs:bss) sock (n - BS.length bs)
265+ else go (bs:bss) sock (n - (fromIntegral $ BS.length bs))
223−
266+#endif
224
267
225interceptAllExn msg =
268interceptAllExn msg =
226 Control.Exception.handle $ \ e ->
269 Control.Exception.handle $ \ e ->
227 case fromException e of
270 case fromException e of
228 Just ThreadKilled -> throw e
271 Just ThreadKilled -> throw e
229 Nothing -> do
272 Nothing -> do
230− BSS.hPutStrLn stderr $ BSS.pack$ "Exception inside "++msg++": "++show e
273+ BS.hPutStrLn stderr $ BS.pack$ "Exception inside "++msg++": "++show e
231 throw e
274 throw e
232-- throw (e :: SomeException)
275-- throw (e :: SomeException)
233
276
@@ -242,7 +285,7 @@ forkWithExceptions forkit descr action = do
242-- BSS.hPutStrLn stderr $ BSS.pack$ "Note: Child thread killed: "++descr
285-- BSS.hPutStrLn stderr $ BSS.pack$ "Note: Child thread killed: "++descr
243 return ()
286 return ()
244 _ -> do
287 _ -> do
245− BSS.hPutStrLn stderr $ BSS.pack$ "Exception inside child thread "++descr++": "++show e
288+ BS.hPutStrLn stderr $ BS.pack$ "Exception inside child thread "++descr++": "++show e
246 throwTo parent (e::SomeException)
289 throwTo parent (e::SomeException)
247 )
290 )
248
291