git.iko.soy/distributed-process
on master help sign in
Factored Pipes.hs to work with either unix or unix-bytestring. Bytestring versions should be more efficient
7cc886fRyan Newton
network-transport/network-transport.cabal+1
compare with master
unifiedsplit
@@ -16,6 +16,7 @@ Library
16 network >= 2.3,
16 network >= 2.3,
17 safe >= 0.3,
17 safe >= 0.3,
18 unix >= 2.5.0.0, random, cereal
18 unix >= 2.5.0.0, random, cereal
19+-- , unix-bytestring
19 Exposed-modules: Network.Transport,
20 Exposed-modules: Network.Transport,
20 Network.Transport.MVar,
21 Network.Transport.MVar,
21 Network.Transport.TCP,
22 Network.Transport.TCP,
network-transport/src/Network/Transport/Pipes.hs+23−12
@@ -1,4 +1,4 @@
1−{-# LANGUAGE BangPatterns, CPP, ScopedTypeVariables #-}
1+{-# LANGUAGE BangPatterns, CPP, ScopedTypeVariables, PackageImports #-}
2
2
3module Network.Transport.Pipes
3module Network.Transport.Pipes
4 ( mkTransport
4 ( mkTransport
@@ -23,13 +23,24 @@ import System.Random (randomIO)
23import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
23import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
24 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout)
24 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout)
25import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
25import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
26−
27−-- Basing this on low level unix routines:
28−import qualified System.Posix.IO as PIO
29−-- Bytestring-version added in unix-2.5.1.0. Let's not depend on this yet:
30−-- import qualified System.Posix.IO.ByteString as PIO
31import System.Posix.Types (Fd)
26import System.Posix.Types (Fd)
32
27
28+#ifdef USE_UNIX_BYTESTRING
29+import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
30+import System.Posix.IO as PIO (openFd, defaultFileFlags, OpenMode(ReadWrite, WriteOnly))
31+(fromS,toS) = (BS.pack, BS.unpack)
32+(fromBS,toBS) = (id,id)
33+readit fd n = PIO.fdRead fd n
34+#else
35+import qualified System.Posix.IO as PIO
36+(toS,fromS) = (id,id)
37+(fromBS,toBS) = (BS.unpack, BS.pack)
38+readit fd n = do (s,_) <- PIO.fdRead fd n
39+ return (BS.pack s)
40+#endif
41+
42+----------------------------------------------------------------------------------------------------
43+
33-- The msg header consists of just a length field represented as a Word32
44-- The msg header consists of just a length field represented as a Word32
34sizeof_header = 4
45sizeof_header = 4
35
46
@@ -60,7 +71,9 @@ mkTransport = do
60 mkSourceEnd :: String -> IO SourceEnd
71 mkSourceEnd :: String -> IO SourceEnd
61 mkSourceEnd filename = do
72 mkSourceEnd filename = do
62 -- Initiate but do not block on file opening:
73 -- Initiate but do not block on file opening:
63− -- Note: Linux fifo semantics are NOT to block on open until
74+ -- Note: Linux fifo semantics are NOT to block on open-RW, but this is not Posix standard.
75+ --
76+ -- We may protect from blocking other threads by running on a separate (OS) thread:
64-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
77-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
65-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
78-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
66 fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags
79 fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags
@@ -84,10 +97,8 @@ mkTransport = do
84
97
85 -- OPTION 2: Speculative file opening, plus this synchronnization:
98 -- OPTION 2: Speculative file opening, plus this synchronnization:
86-- fd <- readMVar mv
99-- fd <- readMVar mv
87−-- dbgprint1$ " (sending... got file descriptor) "
88−
89 ----------------------------------------
100 ----------------------------------------
90− cnt <- PIO.fdWrite fd (BS.unpack finalmsg) -- inefficient!
101+ cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
91 unless (fromIntegral cnt == BS.length finalmsg) $
102 unless (fromIntegral cnt == BS.length finalmsg) $
92 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
103 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
93 ----------------------------------------
104 ----------------------------------------
@@ -105,8 +116,8 @@ mkTransport = do
105
116
106 mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
117 mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
107 fd <- takeMVar mv
118 fd <- takeMVar mv
108− let oneread n = do (s,cnt) <- PIO.fdRead fd (fromIntegral n)
119+ let oneread n = do bs <- readit fd (fromIntegral n)
109− return (BS.pack s, fromIntegral cnt)
120+ return (bs, BS.length bs)
110 dbgprint2$ " Attempt read header..."
121 dbgprint2$ " Attempt read header..."
111
122
112 let spinread :: Int -> IO BS.ByteString
123 let spinread :: Int -> IO BS.ByteString