@@ -43,27 +43,29 @@ import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
43 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)
43 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)
44import System.IO.Error (ioeGetHandle)
44import System.IO.Error (ioeGetHandle)
45import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
45import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
46−import System.Posix.Types (Fd)
46+import System.Posix.Types (Fd, ByteCount)
47import System.Directory (doesFileExist, removeFile)
47import System.Directory (doesFileExist, removeFile)
48import System.Mem (performGC)
48import System.Mem (performGC)
51−-- define USE_UNIX_BYTESTRING
50+#define USE_UNIX_BYTESTRING
53#ifdef USE_UNIX_BYTESTRING
51#ifdef USE_UNIX_BYTESTRING
54import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
52import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
53+-- import qualified "unix-bytestring" System.Posix.IO.ByteString.Lazy as PIO
55−import System.Posix.IO as PIO (openFd, closeFd, defaultFileFlags,
54+import System.Posix.IO as PIO (openFd, closeFd, -- append, exclusive, noctty, nonBlock, trunc,
56− OpenMode(ReadWrite, WriteOnly))
55+ OpenFileFlags(..), OpenMode(ReadOnly, WriteOnly))
57−(fromS,toS) = (BS.pack, BS.unpack)
56+-- (fromS,toS) = (BS.pack, BS.unpack)
57+(fromS,toS) = (BSS.pack, BSS.unpack)
58−(fromBS,toBS) = (id,id)
58+fromBS = id
59readit fd n = PIO.fdRead fd n
59readit fd n = PIO.fdRead fd n
61import qualified System.Posix.IO as PIO
61import qualified System.Posix.IO as PIO
62(toS,fromS) = (id,id)
62(toS,fromS) = (id,id)
63−(fromBS,toBS) = (BS.unpack, BS.pack)
63+fromBS = BSS.unpack
64readit fd n = do (s,_) <- PIO.fdRead fd n
64readit fd n = do (s,_) <- PIO.fdRead fd n
65− return (BS.pack s)
65+ return (BSS.pack s)
67+-- readit :: Fd -> Int -> IO BS.ByteString
68+readit :: Fd -> ByteCount -> IO BSS.ByteString
68----------------------------------------------------------------------------------------------------
70----------------------------------------------------------------------------------------------------
@@ -125,21 +127,30 @@ mkTransport = do
125 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
127 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
126 -- we have to do something more sophisticated.
128 -- we have to do something more sophisticated.
127 SourceEnd
129 SourceEnd
128− { send = \bss -> do
130+ { send = \bsls -> do
129 -- ThreadSafe: This may happen on multiple processes/threads:
131 -- ThreadSafe: This may happen on multiple processes/threads:
130− let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss
132+ let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bsls
131 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
133 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
132 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
134 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
133 -- Otherwise it's just a simple write:
135 -- Otherwise it's just a simple write:
134 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
136 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
138+ let hdrbss :: BSS.ByteString
140+ hdrbss = encode msgsize
142+-- hdrbss = BS.fromChunks[encode msgsize]
143+ [hdrbss] = BS.toChunks (encode msgsize)
135-- let finalmsg = BS.concat (encode msgsize : bss)
146-- let finalmsg = BS.concat (encode msgsize : bss)
136− let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)
147+ let finalmsg = BSS.concat (hdrbss : concatMap BS.toChunks bsls)
138 fd <- readMVar mv -- Synchronize with file opening.
149 fd <- readMVar mv -- Synchronize with file opening.
139 ----------------------------------------
150 ----------------------------------------
140 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
151 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
141− unless (fromIntegral cnt == BS.length finalmsg) $
152+ unless (fromIntegral cnt == BSS.length finalmsg) $
142− error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
153+ error$ "Failed to write message in one go, length: "++ show (BSS.length finalmsg)
143 ----------------------------------------
154 ----------------------------------------
144 return ()
155 return ()
145 , closeSourceEnd = do
156 , closeSourceEnd = do
@@ -156,13 +167,13 @@ mkTransport = do
156 -- happen on multiple threads so we grab a lock.
167 -- happen on multiple threads so we grab a lock.
157 takeMVar lock
168 takeMVar lock
159− let spinread :: Int64 -> IO BS.ByteString
170+ let spinread :: Int -> IO BSS.ByteString
160 spinread desired = do
171 spinread desired = do
162 bs <- tryUntilNoIOErr$
173 bs <- tryUntilNoIOErr$
163 readit fd (fromIntegral desired)
174 readit fd (fromIntegral desired)
165− case BS.length bs of
176+ case BSS.length bs of
166 n | n == desired -> return bs
177 n | n == desired -> return bs
167 0 -> do threadDelay (10*1000)
178 0 -> do threadDelay (10*1000)
168 spinread desired
179 spinread desired
@@ -170,16 +181,19 @@ mkTransport = do
170 show desired ++" bytes) got "++ show l ++ " bytes"
181 show desired ++" bytes) got "++ show l ++ " bytes"
172 hdr <- spinread sizeof_header
183 hdr <- spinread sizeof_header
184+ let -- decoded :: BSS.ByteString
173#ifdef CEREAL
185#ifdef CEREAL
174− let decoded = decode (BSS.concat$ BS.toChunks hdr)
186+-- decoded = decode (BSS.concat$ BS.toChunks hdr)
187+ decoded = decode hdr
176− let decoded = decode hdr
189+-- decoded = decode hdr
190+ decoded = decode (BS.fromChunks [hdr])
178 payload <- case decoded of
192 payload <- case decoded of
179 Left err -> error$ "ERROR: "++ err
193 Left err -> error$ "ERROR: "++ err
180 Right size -> spinread (fromIntegral (size::Word32))
194 Right size -> spinread (fromIntegral (size::Word32))
181 putMVar lock ()
195 putMVar lock ()
182− return [payload]
196+ return [BS.fromChunks [payload]] -- How terribly listy.
183 , closeTargetEnd = do
197 , closeTargetEnd = do
184 fd <- readMVar mv
198 fd <- readMVar mv
185 PIO.closeFd fd
199 PIO.closeFd fd