9587c5d · distributed-process

25import Data.IORef

26import qualified Data.IntMap as IntMap

27import qualified Data.ByteString.Char8 as BS

28−import Data.Time.Clock (getCurrentTime, diffUTCTime) -- Not in 6.10

28+import Data.Time.Clock (getCurrentTime, diffUTCTime, UTCTime) -- Not in 6.10

29

30-- For some STRANGE reason this is not working with Data.Binary [2012.02.20]:

31#define CEREAL

100 { newConnectionWith = \ _ -> do

101 -- Here we protect from blocking other threads by running on a separate (OS) thread:

102 -- Opening the file on the reader side should always succeed:

103− mv <- onOSThread$ tryUntilNoIOErr $

103+ mv <- onOSThread$ tryUntilNoIOErr "open pipe for read" $

104-- spinTillThere filename

105 PIO.openFd filename PIO.ReadOnly Nothing fileFlags

106

120 mkSourceEnd :: String -> IO SourceEnd

121 mkSourceEnd filename = do

122 -- Initiate but do not block on file opening:

123− mv <- onOSThread$ tryUntilNoIOErr $

123+ mv <- onOSThread$ tryUntilNoIOErr "open pipe for write" $

124 -- The reader must connect first, the writer here spins with backoff.

125 PIO.openFd filename PIO.WriteOnly Nothing fileFlags

126

177 let spinread :: Int -> IO BS.ByteString

178 spinread desired = do

179

180− bs <- tryUntilNoIOErr$

180+ bs <- tryUntilNoIOErr "read pipe file descriptor"$

181 readit fd (fromIntegral desired)

182

183 case BS.length bs of

207 PIO.closeFd fd

208 }

209

210+-- | Wait untila file is present, with backoff.

210spinTillThere :: String -> IO ()

211spinTillThere :: String -> IO ()

211spinTillThere filename = mkBackoff >>= loop

212spinTillThere filename = mkBackoff >>= loop

212 where

213 where

213 loop bkoff = do b <- doesFileExist filename

214 loop bkoff = do b <- doesFileExist filename

214 unless b $ do bkoff; loop bkoff

215 unless b $ do bkoff; loop bkoff

215

216

216−-- Wait successively longer between attempts.

217+-- | Create a backoff action that waits successively longer between attempts.

217mkBackoff :: IO (IO ())

218mkBackoff :: IO (IO ())

218mkBackoff =

219mkBackoff =

219 do tref <- newIORef 1

220 do tref <- newIORef 1

223 where

224 where

224 maxwait = 50 * 1000

225 maxwait = 50 * 1000

225

226

227+-- | Try an IO action repeatedly (with backoff) until it succeeds without exception.

226−tryUntilNoIOErr :: IO a -> IO a

228+tryUntilNoIOErr :: String -> IO a -> IO a

227−tryUntilNoIOErr action =

229+tryUntilNoIOErr msg action =

228 do startTime <- getCurrentTime

230 do startTime <- getCurrentTime

229 bkoff <- mkBackoff

231 bkoff <- mkBackoff

230− loop startTime bkoff

232+ loop startTime startTime bkoff

231−

232 where

233 where

233− loop start bkoff =

234+ loop start lastWarned bkoff =

234 handle (\ (e :: IOException) ->

235 handle (\ (e :: IOException) ->

235− do maybeWarn start

236+ do lw <- maybeWarn start lastWarned

236 bkoff

237 bkoff

237-- BSS.hPutStr stderr$ BSS.pack$ " got IO err: " ++ show e

238-- BSS.hPutStr stderr$ BSS.pack$ " got IO err: " ++ show e

238 -- case ioeGetHandle e of

239 -- case ioeGetHandle e of

239 -- Nothing -> BSS.hPutStrLn stderr$ BSS.pack$ " no hndl io err."

240 -- Nothing -> BSS.hPutStrLn stderr$ BSS.pack$ " no hndl io err."

240 -- Just x -> BSS.hPutStrLn stderr$ BSS.pack$ " HNDL on io err!" ++ show x

241 -- Just x -> BSS.hPutStrLn stderr$ BSS.pack$ " HNDL on io err!" ++ show x

241− loop start bkoff) $

242+ loop start lw bkoff) $

242 action

243 action

243 -- After how many milleseconds should we start warning?

244 -- After how many milleseconds should we start warning?

244 warn_lvl = 3000

245 warn_lvl = 3000

245− maybeWarn start = do

246+ maybeWarn :: UTCTime -> UTCTime -> IO UTCTime

247+ maybeWarn start lastWarned = do

246 now <- getCurrentTime

248 now <- getCurrentTime

247− let seconds :: Double = fromRational $ toRational $ diffUTCTime now start

249+ let seconds :: Double = fromRational $ toRational $ diffUTCTime now start

250+ sinceWarned :: Double = fromRational $ toRational $ diffUTCTime now lastWarned

248− when (seconds * 1000 >= warn_lvl) $

251+ if (seconds * 1000 >= warn_lvl &&

252+ sinceWarned >= 1.0 ) then do

249− BS.hPutStrLn stderr$ BS.pack$ "WARNING: have been trying an IO action for "++show seconds++ " seconds without success"

253+ BS.hPutStrLn stderr$ BS.pack$ "WARNING: have been trying an IO action ("++msg++

250−

254+ ") for "++show seconds++ " seconds without success"

255+ return now

256+ else

257+ return lastWarned

251

258

252-- Execute an action on its own OS thread. Return an MVar to synchronize on.

259-- Execute an action on its own OS thread. Return an MVar to synchronize on.

253onOSThread :: IO a -> IO (MVar a)

260onOSThread :: IO a -> IO (MVar a)