89b2116 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Fixed remaining demo. Pipes works for all 5.

89b2116Ryan Newton15 years ago

examples/DemoTransport.hs+9−12

compare with master

unifiedsplit

@@ -66,11 +66,14 @@ demo2 mktrans = do

66 trans <- mktrans

68 (sourceAddr, targetEnd) <- newConnection trans

69−

69+ putStrLn " ================= "

70 sourceEnd <- connect sourceAddr

71+ putStrLn " FRK TO SEND"

71 forkIO $ send sourceEnd [BS.pack "hello 1"]

72 forkIO $ send sourceEnd [BS.pack "hello 1"]

72 threadDelay 100000

73 threadDelay 100000

75+ putStrLn " NEXT fork receiver (logServer):"

76+

74 forkIO $ logServer "logServer" targetEnd

77 forkIO $ logServer "logServer" targetEnd

75 threadDelay 100000

78 threadDelay 100000

@@ -132,7 +135,7 @@ logServer name targetEnd = forever $ do

132 x <- receive targetEnd

135 x <- receive targetEnd

133 trace (name ++ " rcvd: " ++ show x) $ return ()

136 trace (name ++ " rcvd: " ++ show x) $ return ()

134

137

135−

138+-- Run a demo above with one of these tranpsort shorthands as argument:

136

139

137-- do cnt <- readIORef cntr

140-- do cnt <- readIORef cntr

138-- writeIORef cntr (cnt+1)

141-- writeIORef cntr (cnt+1)

@@ -141,6 +144,8 @@ logServer name targetEnd = forever $ do

141--mkTCP = do cntr <- newIORef 0

144--mkTCP = do cntr <- newIORef 0

142-- mkTCPOff 8080

145-- mkTCPOff 8080

143

146

147+--------------------------------------------------------------------------------

148+

144runWAllTranports :: (IO Transport -> IO ()) -> Int -> IO ()

149runWAllTranports :: (IO Transport -> IO ()) -> Int -> IO ()

145runWAllTranports demo offset = do

150runWAllTranports demo offset = do

146 putStrLn "------------------------------------------------------------"

151 putStrLn "------------------------------------------------------------"

@@ -164,20 +169,12 @@ main = do

164 runWAllTranports demo0 0

169 runWAllTranports demo0 0

165 putStrLn "Demo1:"

170 putStrLn "Demo1:"

166 runWAllTranports demo1 10

171 runWAllTranports demo1 10

167−

168−-- putStrLn "Demo2:"

172+ putStrLn "Demo2:"

169−-- runWAllTranports demo2 20

173+ runWAllTranports demo2 20

170−

171 putStrLn "Demo3:"

174 putStrLn "Demo3:"

172 runWAllTranports demo3 30

175 runWAllTranports demo3 30

173 putStrLn "Demo4:"

176 putStrLn "Demo4:"

174 runWAllTranports demo4 40

177 runWAllTranports demo4 40

175

178

176 threadDelay (500 * 1000)

179 threadDelay (500 * 1000)

177−

178 putStrLn "Done with all demos!"

180 putStrLn "Done with all demos!"

179−

180−-- trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"

181−

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

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

network-transport/src/Network/Transport/Pipes.hs+22−37

@@ -112,9 +112,10 @@ mkTransport = do

112 -- OPTION (1)

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

115 mv <- onOSThread$ tryUntilNoIOErr $

116 mv <- onOSThread$ tryUntilNoIOErr $

116 PIO.openFd filename PIO.WriteOnly Nothing fileFlags

117 PIO.openFd filename PIO.WriteOnly Nothing fileFlags

117− fd <- takeMVar mv

118+-- fd <- takeMVar mv

118

119

119 -- OPTION (2) / (3)

120 -- OPTION (2) / (3)

120-- spinTillThere filename

121-- spinTillThere filename

@@ -143,7 +144,7 @@ mkTransport = do

143 dbgprint1$ " Final send msg: " ++ show finalmsg

144 dbgprint1$ " Final send msg: " ++ show finalmsg

144

145

145 -- OPTION 2: Speculative file opening, plus this synchronnization:

146 -- OPTION 2: Speculative file opening, plus this synchronnization:

146−-- fd <- readMVar mv

147+ fd <- readMVar mv

147 ----------------------------------------

148 ----------------------------------------

148 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!

149 cnt <- PIO.fdWrite fd (fromBS finalmsg) -- inefficient to use String here!

149 unless (fromIntegral cnt == BS.length finalmsg) $

150 unless (fromIntegral cnt == BS.length finalmsg) $

@@ -163,6 +164,7 @@ mkTransport = do

163 dbgprint2$ " (got lock)"

164 dbgprint2$ " (got lock)"

164

165

165 spinTillThere filename

166 spinTillThere filename

167+ -- Opening the file on the reader side should always succeed:

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

168 fd <- PIO.openFd filename PIO.ReadOnly Nothing fileFlags

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

169-- fd <- PIO.openFd filename PIO.ReadWrite Nothing fileFlags

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

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

@@ -224,47 +226,38 @@ mkTransport = do

224

226

225spinTillThere :: String -> IO ()

227spinTillThere :: String -> IO ()

226spinTillThere filename =

228spinTillThere filename =

227− do dbgprint2$ " Spinning till file present: "++ filename

229+ do dbgprint2$ " Spinning till file present: "++ filename

228− loop 1

230+ mkBackoff >>= loop

229 where

231 where

230− maxwait = 10 * 1000

231− loop t | t > maxwait = loop maxwait

232− loop t = do b <- doesFileExist filename

232+ loop bkoff = do b <- doesFileExist filename

233− unless b $ do

233+ unless b $ do bkoff; loop bkoff

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

234

243−-- mkBackoff :: IO (IO ())

244

235

236+mkBackoff :: IO (IO ())

245mkBackoff =

237mkBackoff =

246 do tref <- newIORef 1

238 do tref <- newIORef 1

247 return$ do t <- readIORef tref

239 return$ do t <- readIORef tref

248− writeIORef tref (min (10 * 1000) (2 * t))

240+ writeIORef tref (min maxwait (2 * t))

249 threadDelay t

241 threadDelay t

242+ where

243+ maxwait = 50 * 1000

250

244

251tryUntilNoIOErr :: IO a -> IO a

245tryUntilNoIOErr :: IO a -> IO a

252tryUntilNoIOErr action = mkBackoff >>= loop

246tryUntilNoIOErr action = mkBackoff >>= loop

253 where

247 where

254 loop bkoff =

248 loop bkoff =

255− handle (\ (e :: IOException) -> bkoff >> loop bkoff) $

249+ handle (\ (e :: IOException) ->

250+ do bkoff

251+ dbgprint2$ " got IO exn: " ++ show e

252+ loop bkoff) $

256 action

253 action

257

254

258−

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

259−-- loop (InfIO bkoff) =

256+onOSThread :: IO a -> IO (MVar a)

260−-- handle (\ (e :: IOException) ->

257+onOSThread action = do

261−-- do InfIO nxt <- bkoff

258+ mv <- newEmptyMVar

262−-- loop nxt) $

259+ forkOS (action >>= putMVar mv )

263−-- action

260+ return mv

264−

265−

266−

267−

268

261

269#ifdef DEBUG

262#ifdef DEBUG

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

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

@@ -274,11 +267,3 @@ dbgprint1 _ = return ()

274dbgprint2 _ = return ()

267dbgprint2 _ = return ()

275#endif

268#endif

276

269

277−

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

279−onOSThread :: IO a -> IO (MVar a)

280−onOSThread action = do

281− mv <- newEmptyMVar

282− forkOS (action >>= putMVar mv )

283− return mv

284−-- [2012.02.19] This didn't seem to help.