cc13eec · distributed-process

git.iko.soy/distributed-process

Add more exception handling. Pipes is still failing outright, the listener reads the header of the first message, then attempts to read the payload and gets stuck: 'too few bytes'.

cc13eecRyan Newton15 years ago

examples/DemoTransport.hs+5−4

compare with master

unifiedsplit

@@ -145,14 +145,14 @@ runWAllTranports demo offset = do

145 writeIORef cntr (cnt+1)

146 mkTransport (TCPConfig undefined "127.0.0.1" (show (8080 + offset + cnt)))

147

148−-- putStrLn "\n PIPES transport:"

148+ putStrLn "\n PIPES transport:"

149−-- demo Network.Transport.Pipes.mkTransport

149+ demo Network.Transport.Pipes.mkTransport

150 putStrLn "\n"

151

152main = do

153 putStrLn "Demo0:"

154 runWAllTranports demo0 0

155−

155+{-

156 putStrLn "Demo1:"

157 runWAllTranports demo1 10

158

@@ -164,7 +164,8 @@ main = do

164

165 putStrLn "Demo4:"

166 runWAllTranports demo4 40

167−

167+-}

168+ threadDelay (500 * 1000)

168 putStrLn "Done with all demos!"

169 putStrLn "Done with all demos!"

169

170

170-- trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"

171-- trans <- mkTransport $ TCPConfig undefined "127.0.0.1" "8080"

network-transport/src/Network/Transport/Pipes.hs+2

@@ -26,6 +26,8 @@ import System.IO (IOMode(ReadMode,AppendMode,WriteMode,ReadWriteMode),

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

27import System.Posix.Types (Fd)

29+-- define DEBUG

30+

29#ifdef USE_UNIX_BYTESTRING

31#ifdef USE_UNIX_BYTESTRING

30import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO

32import qualified "unix-bytestring" System.Posix.IO.ByteString as PIO

31import System.Posix.IO as PIO (openFd, defaultFileFlags, OpenMode(ReadWrite, WriteOnly))

33import System.Posix.IO as PIO (openFd, defaultFileFlags, OpenMode(ReadWrite, WriteOnly))

network-transport/src/Network/Transport/TCP.hs+17−4

@@ -8,7 +8,8 @@ import Network.Transport

8import Control.Concurrent (forkIO, ThreadId, killThread, myThreadId)

9import Control.Concurrent.Chan

10import Control.Concurrent.MVar

11−import Control.Exception (SomeException, throwTo, catch)

11+import Control.Exception (SomeException, IOException, AsyncException(ThreadKilled),

12+ fromException, throwTo, throw, catch, handle)

12import Control.Monad (forever, forM_)

13import Control.Monad (forever, forM_)

13import Data.ByteString.Lazy.Char8 (ByteString)

14import Data.ByteString.Lazy.Char8 (ByteString)

14import Data.IntMap (IntMap)

15import Data.IntMap (IntMap)

@@ -23,6 +24,7 @@ import Safe

23import System.IO (stderr, hPutStrLn)

24import System.IO (stderr, hPutStrLn)

25import qualified Data.Binary as B

26import qualified Data.Binary as B

27+import qualified Data.ByteString.Char8 as BSS

26import qualified Data.ByteString.Lazy.Char8 as BS

28import qualified Data.ByteString.Lazy.Char8 as BS

27import qualified Data.IntMap as IntMap

29import qualified Data.IntMap as IntMap

28import qualified Network.Socket as N

30import qualified Network.Socket as N

@@ -207,7 +209,9 @@ procMessages chans chanId chan sock = do

207-- indicates success. This hasn't been implemented, since a Transport `receive`

209-- indicates success. This hasn't been implemented, since a Transport `receive`

208-- represents an atomic receipt of a message.

210-- represents an atomic receipt of a message.

209recvExact :: Socket -> Int64 -> IO [ByteString]

211recvExact :: Socket -> Int64 -> IO [ByteString]

210−recvExact sock n = go [] sock n

212+recvExact sock n =

213+ interceptAllExn "recvExact" $

214+ go [] sock n

211 where

215 where

212 go :: [ByteString] -> Socket -> Int64 -> IO [ByteString]

216 go :: [ByteString] -> Socket -> Int64 -> IO [ByteString]

213 go bss _ 0 = return (reverse bss)

217 go bss _ 0 = return (reverse bss)

@@ -218,6 +222,10 @@ recvExact sock n = go [] sock n

218 else go (bs:bss) sock (n - BS.length bs)

222 else go (bs:bss) sock (n - BS.length bs)

219

223

220

224

225+interceptAllExn msg =

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

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

228+ throw (e :: SomeException)

221

229

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

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

223forkWithExceptions forkit descr action = do

231forkWithExceptions forkit descr action = do

@@ -225,7 +233,12 @@ forkWithExceptions forkit descr action = do

225 forkit $

233 forkit $

226 Control.Exception.catch action

234 Control.Exception.catch action

227 (\ e -> do

235 (\ e -> do

236+ case fromException e of

237+ Just ThreadKilled ->

238+-- BSS.hPutStrLn stderr $ BSS.pack$ "Note: Child thread killed: "++descr

239+ return ()

240+ _ -> do

228− hPutStrLn stderr $ "Exception inside child thread "++descr++": "++show e

241+ BSS.hPutStrLn stderr $ BSS.pack$ "Exception inside child thread "++descr++": "++show e

229− throwTo parent (e::SomeException)

242+ throwTo parent (e::SomeException)

230 )

243 )

231

244