@@ -6,7 +6,6 @@
66
77-- * Add support for messages greater than 4096 bytes.
88-- * debug ODD problem with CEREAL below
9−-- * switch to unix-bytestring after that package is updated for 7.4.1
109
1110----------------------------------------------------------------------------------------------------
1211
@@ -88,9 +87,12 @@ fileFlags =
8887
8988mkTransport :: IO Transport
9089mkTransport = do
91− uid <- randomIO :: IO Word64
90+-- uid <- randomIO :: IO Word64
91+ -- For backwards compatibility we shouldn't assume a Random instance for Word64:
92+ uid <- randomIO :: IO Int
93+
9294 lock <- newMVar ()
93− let filename = "/tmp/pipe_"++show uid
95+ let filename = "/tmp/pipe_"++show (abs uid)
9496 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
9597
9698 return Transport
@@ -128,6 +130,7 @@ mkTransport = do
128130 SourceEnd
129131 { send = \bsls -> do
130132 -- ThreadSafe: This may happen on multiple processes/threads:
133+ -- We don't need to lock because the write is atomic (for <4096):
131134 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bsls
132135 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
133136 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
@@ -162,10 +165,14 @@ mkTransport = do
162165 { receive = do
163166 fd <- readMVar mv -- Make sure Fd is there before
164167
165− -- This should only happen on a single process. But it may
168+ -- This should only happen on a *single* process. But it may
166169 -- happen on multiple threads so we grab a lock.
170+
171+ -- If we don't lock, we cannot atomically read the length
172+ -- and THEN read the payload. (A single read could be
173+ -- atomic, but not two reads.)
167174 takeMVar lock
168−
175+
169176 let spinread :: Int -> IO BS.ByteString
170177 spinread desired = do
171178
@@ -174,23 +181,24 @@ mkTransport = do
174181
175182 case BS.length bs of
176183 n | n == desired -> return bs
177− 0 -> do threadDelay (10*1000)
178− spinread desired
184+ -- Because we're in non-blocking mode we deal with failed reads thusly:
185+-- 0 -> do threadDelay (10*1000)
186+-- spinread desired
187+ 0 -> error$ "receive: read zero bytes from pipe. Even in non-blocking mode should have had an EAGAIN error instead."
179188 l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++
180189 show desired ++" bytes) got "++ show l ++ " bytes"
181190
182191 hdr <- spinread sizeof_header
183− let -- decoded :: BSS.ByteString
192+ let
184193#ifdef CEREAL
185−-- decoded = decode (BSS.concat$ BS.toChunks hdr)
186194 decoded = decode hdr
187195#else
188−-- decoded = decode hdr
189196 decoded = decode (BS.fromChunks [hdr])
190197#endif
191198 payload <- case decoded of
192199 Left err -> error$ "ERROR: "++ err
193200 Right size -> spinread (fromIntegral (size::Word32))
201+
194202 putMVar lock ()
195203 return [payload] -- How terribly listy.
196204 , closeTargetEnd = do