@@ -63,7 +63,6 @@ fileFlags =
6363 PIO.append = False,
6464 PIO.exclusive = False,
6565 PIO.noctty = False,
66−-- PIO.nonBlock = False,
6766 PIO.nonBlock = True,
6867 -- In nonblocking mode opening for read will always succeed and
6968 -- opening for write must happen second.
@@ -81,11 +80,8 @@ mkTransport = do
8180 let filename = "/tmp/pipe_"++show uid
8281 createNamedPipe filename $ unionFileModes ownerReadMode ownerWriteMode
8382
84− dbgprint1$ " Created pipe at location: "++ filename
85−
8683 return Transport
8784 { newConnectionWith = \ _ -> do
88− dbgprint1$ " Creating new connection"
8985 return (mkSourceAddr filename,
9086 mkTargetEnd filename lock)
9187 , newMulticastWith = error "Pipes.hs: newMulticastWith not implemented yet"
@@ -104,26 +100,11 @@ mkTransport = do
104100 mkSourceEnd :: String -> IO SourceEnd
105101 mkSourceEnd filename = do
106102 -- Initiate but do not block on file opening:
107− -- Note: Linux fifo semantics are NOT to block on open-RW, but this is not Posix standard.
108− --
109−
110− -- All THREE of the below options were observed to work on a simple demo:
111103
112− -- OPTION (1)
113104 -- Here we protect from blocking other threads by running on a separate (OS) thread:
114−-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags
115− dbgprint1$ "About to try opening writing end:"
116105 mv <- onOSThread$ tryUntilNoIOErr $
106+ -- The reader must connect first, the writer here spins with backoff.
117107 PIO.openFd filename PIO.WriteOnly Nothing fileFlags
118−-- fd <- takeMVar mv
119−
120− -- OPTION (2) / (3)
121−-- spinTillThere filename
122−
123− -- The reader must connect first, the writer here spins with backoff:
124−-- fd <- PIO.openFd filename PIO.WriteOnly Nothing fileFlags
125−-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags
126− dbgprint1$ "GOT WRITING END OPEN ... "
127108
128109 return $
129110 -- Write to the named pipe. If the message is less than
@@ -131,9 +112,7 @@ mkTransport = do
131112 -- we have to do something more sophisticated.
132113 SourceEnd
133114 { send = \bss -> do
134− dbgprint1$ "Sending.. "++ show bss
135−
136− -- This may happen on multiple processes/threads:
115+ -- ThreadSafe: This may happen on multiple processes/threads:
137116 let msgsize :: Word32 = fromIntegral$ foldl' (\n s -> n + BS.length s) 0 bss
138117 when (msgsize > 4096)$ -- TODO, look up PIPE_BUF in foreign code
139118 error "Message larger than blocksize written atomically to a named pipe. Unimplemented."
@@ -141,98 +120,64 @@ mkTransport = do
141120 -- We append the length as a header. TODO - REMOVE EXTRA COPY HERE:
142121-- let finalmsg = BS.concat (encode msgsize : bss)
143122 let finalmsg = BS.concat ((BS.fromChunks[encode msgsize]) : bss)
144− dbgprint1$ " Final send msg: " ++ show finalmsg
145−
146− -- OPTION 2: Speculative file opening, plus this synchronnization:
147− fd <- readMVar mv
123+
124+ fd <- readMVar mv -- Synchronnize with file opening.
148125 ----------------------------------------
149126 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!
150127 unless (fromIntegral cnt == BS.length finalmsg) $
151128 error$ "Failed to write message in one go, length: "++ show (BS.length finalmsg)
152129 ----------------------------------------
153−
154130 return ()
155131 }
156132
157133 mkTargetEnd :: String -> MVar () -> TargetEnd
158134 mkTargetEnd filename lock = TargetEnd
159135 { receive = do
160− dbgprint2$ "Begin receive action..."
161136 -- This should only happen on a single process. But it may
162137 -- happen on multiple threads so we grab a lock.
163138 takeMVar lock
164− dbgprint2$ " (got lock)"
165139
166140 spinTillThere filename
167141 -- Opening the file on the reader side should always succeed:
168142 fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags
169−-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags
170−-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags
171− dbgprint2$ " spawned thread for open..."
172−-- let dummy 0 acc = acc
173−-- dummy n acc = dummy (n-1) (1.1 * acc)
174−-- dbgprint2$ " Silly work: " ++ show (dummy 1000000 0.0001)
175143
176−-- fd <- readMVar mv
177− let -- oneread :: Int64 -> IO (BS.ByteString, Int64)
178− oneread fd n =
179−-- handle (\e -> do
180−-- dbgprint2$ " Got Exception in read call! "++ show e
181−-- case fromException e of
182−-- Nothing -> dbgprint2$ "It's NOT an IO exception! "
183−-- Just e -> dbgprint2$ "It's an IO exception! "++ show (e :: IOException)
184−-- -- throw (e :: SomeException)
185−-- return undefined
186−-- ) $
144+ let oneread fd n =
187145 do bs <- tryUntilNoIOErr$
188146 readit fd (fromIntegral n)
189147 return (bs, BS.length bs)
190− dbgprint2$ " Attempt read header..."
191148
192149 let spinread :: Int64 -> IO BS.ByteString
193150 spinread desired = do
194−#ifdef DEBUG
195−-- hPutStr stderr "."
196− dbgprint2$ " SPINREAD "++ show desired
197−-- BSS.hPutStr stdout (BSS.pack$ " "++show desired)
198−#endif
199−-- fd <- readMVar mv
200− (bytes,len) <- oneread fd desired
201− case len of
202− n | n == desired -> return bytes
151+
152+ bs <- tryUntilNoIOErr$
153+ readit fd (fromIntegral desired)
154+
155+ case BS.length bs of
156+ n | n == desired -> return bs
203157 0 -> do threadDelay (10*1000)
204158 spinread desired
205159 l -> error$ "Inclomplete read expected either 0 bytes or complete msg ("++
206160 show desired ++" bytes) got "++ show l ++ " bytes"
207161
208162 hdr <- spinread sizeof_header
209− dbgprint2$ " Got header "++ show hdr ++ ", next attempt to read payload:"
210163#ifdef CEREAL
211164 let decoded = decode (BSS.concat$ BS.toChunks hdr)
212−-- dbgprint2$ " DECODING HDR, bytes "++ show (BS.length hdr) ++ ": "++show hdr
213−-- evaluate decoded
214−-- dbgprint2$ " DONE DECODING HDR"
215165#else
216166 let decoded = decode hdr
217167#endif
218168 payload <- case decoded of
219169 Left err -> error$ "ERROR: "++ err
220170 Right size -> spinread (fromIntegral (size::Word32))
221− dbgprint2$ " Got payload "++ show payload
222−
223171 putMVar lock ()
224172 return [payload]
225173 }
226174
227175spinTillThere :: String -> IO ()
228−spinTillThere filename =
229− do dbgprint2$ " Spinning till file present: "++ filename
230− mkBackoff >>= loop
176+spinTillThere filename = mkBackoff >>= loop
231177 where
232178 loop bkoff = do b <- doesFileExist filename
233179 unless b $ do bkoff; loop bkoff
234180
235−
236181mkBackoff :: IO (IO ())
237182mkBackoff =
238183 do tref <- newIORef 1
@@ -248,7 +193,6 @@ tryUntilNoIOErr action = mkBackoff >>= loop
248193 loop bkoff =
249194 handle (\ (e :: IOException) ->
250195 do bkoff
251− dbgprint2$ " got IO exn: " ++ show e
252196 loop bkoff) $
253197 action
254198
@@ -258,12 +202,3 @@ onOSThread action = do
258202 mv <- newEmptyMVar
259203 forkOS (action >>= putMVar mv )
260204 return mv
261−
262−#ifdef DEBUG
263−dbgprint1 s = do BSS.hPutStrLn stderr (BSS.pack s); hFlush stderr;
264−dbgprint2 s = do BSS.hPutStrLn stdout (BSS.pack s); hFlush stdout;
265−#else
266−dbgprint1 _ = return ()
267−dbgprint2 _ = return ()
268−#endif
269−