@@ -43,27 +43,29 @@ import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
4343 openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)
4444import System.IO.Error (ioeGetHandle)
4545import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
46−import System.Posix.Types (Fd)
46+import System.Posix.Types (Fd, ByteCount)
4747import System.Directory (doesFileExist, removeFile)
4848import System.Mem (performGC)
4949
50−-- define DEBUG
51−-- define USE_UNIX_BYTESTRING
52−
50+#define USE_UNIX_BYTESTRING
5351#ifdef USE_UNIX_BYTESTRING
5452import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO
55−import System.Posix.IO as PIO (openFd, closeFd, defaultFileFlags,
56− OpenMode(ReadWrite, WriteOnly))
57−(fromS,toS) = (BS.pack, BS.unpack)
58−(fromBS,toBS) = (id,id)
53+-- import qualified "unix-bytestring" System.Posix.IO.ByteString.Lazy as PIO
54+import System.Posix.IO as PIO (openFd, closeFd, -- append, exclusive, noctty, nonBlock, trunc,
55+ OpenFileFlags(..), OpenMode(ReadOnly, WriteOnly))
56+-- (fromS,toS) = (BS.pack, BS.unpack)
57+(fromS,toS) = (BSS.pack, BSS.unpack)
58+fromBS = id
5959readit fd n = PIO.fdRead fd n
6060#else
6161import qualified System.Posix.IO as PIO
6262(toS,fromS) = (id,id)
63−(fromBS,toBS) = (BS.unpack, BS.pack)
63+fromBS = BSS.unpack
6464readit fd n = do (s,_) <- PIO.fdRead fd n
65− return (BS.pack s)
65+ return (BSS.pack s)
6666#endif
67+-- readit :: Fd -> Int -> IO BS.ByteString
68+readit :: Fd -> ByteCount -> IO BSS.ByteString
6769
6870----------------------------------------------------------------------------------------------------
6971
@@ -125,21 +127,30 @@ mkTransport = do
125127 -- PIPE_BUF (4KB on linux) then this should be atomic, otherwise
126128 -- we have to do something more sophisticated.
127129 SourceEnd
128− { send = \bss -> do
130+ { send = \bsls -> do
129131 -- 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
131133 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
132134 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
133135 -- Otherwise it's just a simple write:
134136 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
137+
138+ let hdrbss :: BSS.ByteString
139+#ifdef CEREAL
140+ hdrbss = encode msgsize
141+#else
142+-- hdrbss = BS.fromChunks[encode msgsize]
143+ [hdrbss] = BS.toChunks (encode msgsize)
144+#endif
145+
135146-- 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)
137148
138149 fd <- readMVar mv -- Synchronize with file opening.
139150 ----------------------------------------
140151 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
141− unless (fromIntegral cnt == BS.length finalmsg) $
142− error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
152+ unless (fromIntegral cnt == BSS.length finalmsg) $
153+ error$ "Failed to write message in one go, length: "++ show (BSS.length finalmsg)
143154 ----------------------------------------
144155 return ()
145156 , closeSourceEnd = do
@@ -156,13 +167,13 @@ mkTransport = do
156167 -- happen on multiple threads so we grab a lock.
157168 takeMVar lock
158169
159− let spinread :: Int64 -> IO BS.ByteString
170+ let spinread :: Int -> IO BSS.ByteString
160171 spinread desired = do
161172
162173 bs <- tryUntilNoIOErr$
163174 readit fd (fromIntegral desired)
164175
165− case BS.length bs of
176+ case BSS.length bs of
166177 n | n == desired -> return bs
167178 0 -> do threadDelay (10*1000)
168179 spinread desired
@@ -170,16 +181,19 @@ mkTransport = do
170181 show desired ++" bytes) got "++ show l ++ " bytes"
171182
172183 hdr <- spinread sizeof_header
184+ let -- decoded :: BSS.ByteString
173185#ifdef CEREAL
174− let decoded = decode (BSS.concat$ BS.toChunks hdr)
186+-- decoded = decode (BSS.concat$ BS.toChunks hdr)
187+ decoded = decode hdr
175188#else
176− let decoded = decode hdr
189+-- decoded = decode hdr
190+ decoded = decode (BS.fromChunks [hdr])
177191#endif
178192 payload <- case decoded of
179193 Left err -> error$ "ERROR: "++ err
180194 Right size -> spinread (fromIntegral (size::Word32))
181195 putMVar lock ()
182− return [payload]
196+ return [BS.fromChunks [payload]] -- How terribly listy.
183197 , closeTargetEnd = do
184198 fd <- readMVar mv
185199 PIO.closeFd fd