c2f03d8 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Made Pipes use nonblocking mode and changed a number of things

c2f03d8Ryan Newton15 years ago

examples/DemoTransport.hs+6−10

compare with master

unifiedsplit

@@ -162,18 +162,16 @@ main = do

162

163 putStrLn "Demo0:"

164 runWAllTranports demo0 0

165−

166 putStrLn "Demo1:"

165 putStrLn "Demo1:"

167 runWAllTranports demo1 10

166 runWAllTranports demo1 10

168

167

169−-- putStrLn "Demo2:"

168+-- putStrLn "Demo2:"

170−-- runWAllTranports demo2 20

169+-- runWAllTranports demo2 20

171−

172−-- putStrLn "Demo3:"

173−-- runWAllTranports demo3 30

174

170

175−-- putStrLn "Demo4:"

171+ putStrLn "Demo3:"

172+ runWAllTranports demo3 30

173+ putStrLn "Demo4:"

176−-- runWAllTranports demo4 40

174+ runWAllTranports demo4 40

177

175

178 threadDelay (500 * 1000)

176 threadDelay (500 * 1000)

179

177

@@ -183,5 +181,3 @@ main = do

183

181

184-- trans1 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"

182-- trans1 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"

185-- trans2 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8081"

183-- trans2 <- mkTransport $ TCPConfig undefined "127.0.0.1" "8081"

186−

187−

network-transport/src/Network/Transport/Pipes.hs+105−19

@@ -5,7 +5,8 @@ module Network.Transport.Pipes

5 ) where

7import Control.Monad (when, unless)

8−import Control.Exception (evaluate)

8+import Control.Exception (evaluate, throw, handle, fromException,

9+ SomeException, IOException)

9import Control.Concurrent.MVar

10import Control.Concurrent.MVar

10import Control.Concurrent (threadDelay, forkOS)

11import Control.Concurrent (threadDelay, forkOS)

11import Data.IntMap (IntMap)

12import Data.IntMap (IntMap)

@@ -29,9 +30,10 @@ import Data.List (foldl')

29import Network.Transport

30import Network.Transport

30import System.Random (randomIO)

31import System.Random (randomIO)

31import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),

32import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),

32− openFile, hClose, hPutStrLn, hPutStr, stderr, stdout)

33+ openFile, hClose, hPutStrLn, hPutStr, stderr, stdout, hFlush)

33import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)

34import System.Posix.Files (createNamedPipe, unionFileModes, ownerReadMode, ownerWriteMode)

34import System.Posix.Types (Fd)

35import System.Posix.Types (Fd)

36+import System.Directory (doesFileExist)

36-- define DEBUG

38-- define DEBUG

37-- define USE_UNIX_BYTESTRING

39-- define USE_UNIX_BYTESTRING

@@ -55,6 +57,23 @@ readit fd n = do (s,_) <- PIO.fdRead fd n

55-- The msg header consists of just a length field represented as a Word32

57-- The msg header consists of just a length field represented as a Word32

56sizeof_header = 4

58sizeof_header = 4

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+

58mkTransport :: IO Transport

77mkTransport :: IO Transport

59mkTransport = do

78mkTransport = do

60 uid <- randomIO :: IO Word64

79 uid <- randomIO :: IO Word64

@@ -92,14 +111,17 @@ mkTransport = do

111

93 -- OPTION (1)

112 -- OPTION (1)

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

113 -- 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

114+-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags

115+ mv <- onOSThread$ tryUntilNoIOErr $

116+ PIO.openFd filename PIO.WriteOnly Nothing fileFlags

96−-- fd <- takeMVar mv

117+ fd <- takeMVar mv

118

98− -- OPTION (2)

119+ -- OPTION (2) / (3)

99−-- fd <- PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags

120+-- spinTillThere filename

100

121

101− -- OPTION (3)

122+ -- The reader must connect first, the writer here spins with backoff:

102− fd <- PIO.openFd filename PIO.WriteOnly Nothing PIO.defaultFileFlags

123+-- fd <- PIO.openFd filename PIO.WriteOnly Nothing fileFlags

124+-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags

103 dbgprint1$ "GOT WRITING END OPEN ... "

125 dbgprint1$ "GOT WRITING END OPEN ... "

104

126

105 return $

127 return $

@@ -138,22 +160,42 @@ mkTransport = do

138 -- This should only happen on a single process. But it may

160 -- This should only happen on a single process. But it may

139 -- happen on multiple threads so we grab a lock.

161 -- happen on multiple threads so we grab a lock.

140 takeMVar lock

162 takeMVar lock

163+ dbgprint2$ " (got lock)"

141

164

165+ spinTillThere filename

166+ fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags

167+-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags

142− mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing PIO.defaultFileFlags

168+-- mv <- onOSThread$ PIO.openFd filename PIO.ReadWrite Nothing fileFlags

143− fd <- takeMVar mv

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

144− let oneread :: Int64 -> IO (BS.ByteString, Int64)

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$

145− oneread n = do bs <- readit fd (fromIntegral n)

186+ readit fd (fromIntegral n)

146− return (bs, BS.length bs)

187+ return (bs, BS.length bs)

147 dbgprint2$ " Attempt read header..."

188 dbgprint2$ " Attempt read header..."

148

189

149 let spinread :: Int64 -> IO BS.ByteString

190 let spinread :: Int64 -> IO BS.ByteString

150 spinread desired = do

191 spinread desired = do

151#ifdef DEBUG

192#ifdef DEBUG

152− hPutStr stderr "."

193+-- hPutStr stderr "."

153−-- dbgprint2$ " SPINREAD "++ show desired

194+ dbgprint2$ " SPINREAD "++ show desired

154− BSS.hPutStr stdout (BSS.pack$ " "++show desired)

195+-- BSS.hPutStr stdout (BSS.pack$ " "++show desired)

155#endif

196#endif

197+-- fd <- readMVar mv

156− (bytes,len) <- oneread desired

198+ (bytes,len) <- oneread fd desired

157 case len of

199 case len of

158 n | n == desired -> return bytes

200 n | n == desired -> return bytes

159 0 -> do threadDelay (10*1000)

201 0 -> do threadDelay (10*1000)

@@ -180,9 +222,53 @@ mkTransport = do

180 return [payload]

222 return [payload]

181 }

223 }

182

224

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+

183#ifdef DEBUG

269#ifdef DEBUG

184−dbgprint1 s = BSS.hPutStrLn stderr (BSS.pack s)

270+dbgprint1 s = do BSS.hPutStrLn stderr (BSS.pack s); hFlush stderr;

185−dbgprint2 s = BSS.hPutStrLn stdout (BSS.pack s)

271+dbgprint2 s = do BSS.hPutStrLn stdout (BSS.pack s); hFlush stdout;

186#else

272#else

187dbgprint1 _ = return ()

273dbgprint1 _ = return ()

188dbgprint2 _ = return ()

274dbgprint2 _ = return ()

network-transport/src/Network/Transport/TCP.hs+7−3

@@ -223,9 +223,13 @@ recvExact sock n =

223

224

225interceptAllExn msg =

226− Control.Exception.handle $ \ e -> do

226+ Control.Exception.handle $ \ e ->

227+ case fromException e of

228+ Just ThreadKilled -> throw e

229+ Nothing -> do

227− BSS.hPutStrLn stderr $ BSS.pack$ "Exception inside "++msg++": "++show e

230+ BSS.hPutStrLn stderr $ BSS.pack$ "Exception inside "++msg++": "++show e

231+ throw e

228− throw (e :: SomeException)

232+-- throw (e :: SomeException)

229

233

230forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId

234forkWithExceptions :: (IO () -> IO ThreadId) -> String -> IO () -> IO ThreadId

231forkWithExceptions forkit descr action = do

235forkWithExceptions forkit descr action = do