@@ -5,7 +5,8 @@ module Network.Transport.Pipes
55 ) where
66
77import Control.Monad (when, unless)
8−import Control.Exception (evaluate)
8+import Control.Exception (evaluate, throw, handle, fromException,
9+ SomeException, IOException)
910import Control.Concurrent.MVar
1011import Control.Concurrent (threadDelay, forkOS)
1112import Data.IntMap (IntMap)
@@ -29,9 +30,10 @@ import Data.List (foldl')
2930import Network.Transport
3031import System.Random (randomIO)
3132import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),
32− openFile, hClose, hPutStrLn, hPutStr, stderr, stdout)
33+ openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)
3334import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)
3435import System.Posix.Types (Fd)
36+import System.Directory (doesFileExist)
3537
3638-- define DEBUG
3739-- define USE_UNIX_BYTESTRING
@@ -55,6 +57,23 @@ readit fd n = do (s,_) <- PIO.fdRead fd n
5557-- The msg header consists of just a length field represented as a Word32
5658sizeof_header = 4
5759
60+-- fileFlags = PIO.defaultFileFlags
61+fileFlags =
62+ PIO.OpenFileFlags {
63+ PIO.append = False,
64+ PIO.exclusive = False,
65+ PIO.noctty = False,
66+-- PIO.nonBlock = False,
67+ PIO.nonBlock = True,
68+ -- In nonblocking mode opening for read will always succeed and
69+ -- opening for write must happen second.
70+ PIO.trunc = False
71+ }
72+-- NOTE:
73+-- "The only open file status flags that can be meaningfully applied
74+-- to a pipe or FIFO are O_NONBLOCK and O_ASYNC. "
75+
76+
5877mkTransport :: IO Transport
5978mkTransport = do
6079 uid <- randomIO :: IO Word64
@@ -92,14 +111,17 @@ mkTransport = do
92111
93112 -- OPTION (1)
94113 -- Here we protect from blocking other threads by running on a separate (OS) thread:
95−-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
96−-- fd <- takeMVar mv
114+-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags
115+ mv <- onOSThread$ tryUntilNoIOErr $
116+ PIO.openFd filename PIO.WriteOnly Nothing fileFlags
117+ fd <- takeMVar mv
97118
98− -- OPTION (2)
99−-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
119+ -- OPTION (2) / (3)
120+-- spinTillThere filename
100121
101− -- OPTION (3)
102− fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags
122+ -- The reader must connect first, the writer here spins with backoff:
123+-- fd <- PIO.openFd filename PIO.WriteOnly Nothing fileFlags
124+-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags
103125 dbgprint1$ "GOT WRITING END OPEN ... "
104126
105127 return $
@@ -138,22 +160,42 @@ mkTransport = do
138160 -- This should only happen on a single process. But it may
139161 -- happen on multiple threads so we grab a lock.
140162 takeMVar lock
163+ dbgprint2$ " (got lock)"
141164
142− mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags
143− fd <- takeMVar mv
144− let oneread :: Int64 -> IO (BS.ByteString, Int64)
145− oneread n = do bs <- readit fd (fromIntegral n)
146− return (bs, BS.length bs)
165+ spinTillThere filename
166+ fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags
167+-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags
168+-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags
169+ dbgprint2$ " spawned thread for open..."
170+-- let dummy 0 acc = acc
171+-- dummy n acc = dummy (n-1) (1.1 * acc)
172+-- dbgprint2$ " Silly work: " ++ show (dummy 1000000 0.0001)
173+
174+-- fd <- readMVar mv
175+ let -- oneread :: Int64 -> IO (BS.ByteString, Int64)
176+ oneread fd n =
177+-- handle (\e -> do
178+-- dbgprint2$ " Got Exception in read call! "++ show e
179+-- case fromException e of
180+-- Nothing -> dbgprint2$ "It's NOT an IO exception! "
181+-- Just e -> dbgprint2$ "It's an IO exception! "++ show (e :: IOException)
182+-- -- throw (e :: SomeException)
183+-- return undefined
184+-- ) $
185+ do bs <- tryUntilNoIOErr$
186+ readit fd (fromIntegral n)
187+ return (bs, BS.length bs)
147188 dbgprint2$ " Attempt read header..."
148189
149190 let spinread :: Int64 -> IO BS.ByteString
150191 spinread desired = do
151192#ifdef DEBUG
152− hPutStr stderr "."
153−-- dbgprint2$ " SPINREAD "++ show desired
154− BSS.hPutStr stdout (BSS.pack$ " "++show desired)
193+-- hPutStr stderr "."
194+ dbgprint2$ " SPINREAD "++ show desired
195+-- BSS.hPutStr stdout (BSS.pack$ " "++show desired)
155196#endif
156− (bytes,len) <- oneread desired
197+-- fd <- readMVar mv
198+ (bytes,len) <- oneread fd desired
157199 case len of
158200 n | n == desired -> return bytes
159201 0 -> do threadDelay (10*1000)
@@ -180,9 +222,53 @@ mkTransport = do
180222 return [payload]
181223 }
182224
225+spinTillThere :: String -> IO ()
226+spinTillThere filename =
227+ do dbgprint2$ " Spinning till file present: "++ filename
228+ loop 1
229+ where
230+ maxwait = 10 * 1000
231+ loop t | t > maxwait = loop maxwait
232+ loop t = do b <- doesFileExist filename
233+ unless b $ do
234+#ifdef DEBUG
235+ hPutStr stderr "?"
236+#endif
237+ threadDelay t
238+ loop (2 * t)
239+-- loop $ round (fromIntegral t * 1.5)
240+
241+data InfIO = InfIO (IO (InfIO))
242+
243+-- mkBackoff :: IO (IO ())
244+
245+mkBackoff =
246+ do tref <- newIORef 1
247+ return$ do t <- readIORef tref
248+ writeIORef tref (min (10 * 1000) (2 * t))
249+ threadDelay t
250+
251+tryUntilNoIOErr :: IO a -> IO a
252+tryUntilNoIOErr action = mkBackoff >>= loop
253+ where
254+ loop bkoff =
255+ handle (\ (e :: IOException) -> bkoff >> loop bkoff) $
256+ action
257+
258+
259+-- loop (InfIO bkoff) =
260+-- handle (\ (e :: IOException) ->
261+-- do InfIO nxt <- bkoff
262+-- loop nxt) $
263+-- action
264+
265+
266+
267+
268+
183269#ifdef DEBUG
184−dbgprint1 s = BSS.hPutStrLn stderr (BSS.pack s)
185−dbgprint2 s = BSS.hPutStrLn stdout (BSS.pack s)
270+dbgprint1 s = do BSS.hPutStrLn stderr (BSS.pack s); hFlush stderr;
271+dbgprint2 s = do BSS.hPutStrLn stdout (BSS.pack s); hFlush stdout;
186272#else
187273dbgprint1 _ = return ()
188274dbgprint2 _ = return ()