a75cb28 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Monitor abnormal termination

a75cb28Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+35−21

compare with master

unifiedsplit

@@ -103,13 +103,13 @@ import Data.Int (Int32)

103import Data.Typeable (Typeable)

104import Data.Foldable (forM_)

105import Data.Maybe (isJust)

106−import Control.Monad (void, liftM2, liftM3, when, unless, forever)

106+import Control.Monad (void, when, unless, forever)

107import Control.Monad.Reader (MonadReader(..), ReaderT, runReaderT)

108import Control.Monad.IO.Class (MonadIO, liftIO)

109import Control.Monad.State (MonadState, StateT, evalStateT, gets, modify)

110−import Control.Applicative ((<$>))

110+import Control.Applicative ((<$>), (<*>))

111import Control.Category ((>>>))

112−import Control.Exception (Exception, throwIO, catch)

112+import Control.Exception (Exception, throwIO, catch, SomeException)

113import Control.Concurrent (forkIO)

114import Control.Concurrent.MVar ( MVar

115 , newMVar

@@ -294,6 +294,7 @@ data MonitorReply = ProcessDied MonitorRef ProcessId DiedReason

294

295data DiedReason =

296 DiedNormal

297+ | DiedException String -- TODO: would prefer SomeException instead of String, but exceptions don't implement Binary

297 | DiedDisconnect

298 | DiedDisconnect

298 | DiedNodeDown

299 | DiedNodeDown

299

300

@@ -527,13 +528,24 @@ forkProcess node proc = do

527 , lproc

528 , lproc

528 )

529 )

529 void . forkIO $ do

530 void . forkIO $ do

531+ catch (do

530− runLocalProcess lproc proc

532+ runLocalProcess lproc proc

531− -- [Unified: Table 4]

533+ -- [Unified: Table 4]

532− let pid = processId lproc

534+ let pid = processId lproc

533− writeChan (localCtrlChan node) $ NodeCtrlMsg

535+ writeChan (localCtrlChan node) $ NodeCtrlMsg

534− { ctrlMsgSender = Left pid

536+ { ctrlMsgSender = Left pid

535− , ctrlMsgSignal = Died (Left pid) DiedNormal

537+ , ctrlMsgSignal = Died (Left pid) DiedNormal

538+ }

539+ )

540+ (\ex -> do

541+ let pid = processId lproc

542+ err = show (ex :: SomeException)

543+ writeChan (localCtrlChan node) $ NodeCtrlMsg

544+ { ctrlMsgSender = Left pid

545+ , ctrlMsgSignal = Died (Left pid) (DiedException err)

536− }

546+ }

547+ )

548+

537 return (processId lproc)

549 return (processId lproc)

538

550

539handleIncomingMessages :: LocalNode -> IO ()

551handleIncomingMessages :: LocalNode -> IO ()

@@ -727,19 +739,19 @@ runLocalProcess lproc proc = runReaderT (unProcess proc) lproc

727

739

728instance Binary LocalProcessId where

740instance Binary LocalProcessId where

729 put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)

741 put lpid = put (lpidUnique lpid) >> put (lpidCounter lpid)

730− get = liftM2 LocalProcessId get get

742+ get = LocalProcessId <$> get <*> get

731

743

732instance Binary ProcessId where

744instance Binary ProcessId where

733 put pid = put (processNodeId pid) >> put (processLocalId pid)

745 put pid = put (processNodeId pid) >> put (processLocalId pid)

734− get = liftM2 ProcessId get get

746+ get = ProcessId <$> get <*> get

735

747

736instance Binary MonitorReply where

748instance Binary MonitorReply where

737 put (ProcessDied ref pid reason) = put ref >> put pid >> put reason

749 put (ProcessDied ref pid reason) = put ref >> put pid >> put reason

738− get = liftM3 ProcessDied get get get

750+ get = ProcessDied <$> get <*> get <*> get

739

751

740instance Binary NodeCtrlMsg where

752instance Binary NodeCtrlMsg where

741 put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)

753 put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)

742− get = liftM2 NodeCtrlMsg get get

754+ get = NodeCtrlMsg <$> get <*> get

743

755

744instance Binary ProcessSignal where

756instance Binary ProcessSignal where

745 put (Monitor pid ref) = putWord8 0 >> put pid >> put ref

757 put (Monitor pid ref) = putWord8 0 >> put pid >> put ref

@@ -747,20 +759,22 @@ instance Binary ProcessSignal where

747 get = do

759 get = do

748 header <- getWord8

760 header <- getWord8

749 case header of

761 case header of

750− 0 -> liftM2 Monitor get get

762+ 0 -> Monitor <$> get <*> get

751− 1 -> liftM2 Died get get

763+ 1 -> Died <$> get <*> get

752 _ -> fail "ProcessSignal.get: invalid"

764 _ -> fail "ProcessSignal.get: invalid"

753

765

754instance Binary DiedReason where

766instance Binary DiedReason where

755− put DiedNormal = putWord8 0

767+ put DiedNormal = putWord8 0

768+ put (DiedException e) = putWord8 1 >> put e

756− put DiedDisconnect = putWord8 1

769+ put DiedDisconnect = putWord8 2

757− put DiedNodeDown = putWord8 2

770+ put DiedNodeDown = putWord8 3

758 get = do

771 get = do

759 header <- getWord8

772 header <- getWord8

760 case header of

773 case header of

761 0 -> return DiedNormal

774 0 -> return DiedNormal

762− 1 -> return DiedDisconnect

775+ 1 -> DiedException <$> get

776+ 2 -> return DiedDisconnect

763− 2 -> return DiedNodeDown

777+ 3 -> return DiedNodeDown

764 _ -> fail "DiedReason.get: invalid"

778 _ -> fail "DiedReason.get: invalid"

765

779

766--------------------------------------------------------------------------------

780--------------------------------------------------------------------------------

distributed-process/tests/TestCH.hs+37−2

@@ -12,6 +12,7 @@ import Control.Concurrent.MVar ( newEmptyMVar

12import Control.Monad (replicateM_)

13import Control.Distributed.Process

14import Control.Monad.IO.Class (liftIO)

15+import Control.Exception (throwIO)

15import qualified Network.Transport as NT (Transport)

16import qualified Network.Transport as NT (Transport)

16import Network.Transport.TCP (createTransport, defaultTCPParameters)

17import Network.Transport.TCP (createTransport, defaultTCPParameters)

17import TestAuxiliary

18import TestAuxiliary

@@ -116,6 +117,39 @@ testMonitorNormalTermination transport = do

116

117

117 takeMVar done

118 takeMVar done

118

119

120+-- | Monitor a process which terminates abnormally

121+testMonitorAbnormalTermination :: NT.Transport -> IO ()

122+testMonitorAbnormalTermination transport = do

123+ monitorSetup <- newEmptyMVar

124+ monitoredProcess <- newEmptyMVar

125+ done <- newEmptyMVar

126+

127+ let err = userError "Abnormal termination"

128+

129+ forkIO $ do

130+ localNode <- newLocalNode transport

131+ addr <- forkProcess localNode . liftIO $ do

132+ readMVar monitorSetup

133+ throwIO err

134+ putMVar monitoredProcess addr

135+

136+ forkIO $ do

137+ localNode <- newLocalNode transport

138+ theirAddr <- readMVar monitoredProcess

139+ runProcess localNode $ do

140+ ref <- monitor theirAddr

141+ liftIO $ do

142+ -- Monitor is asynchronous, but we want to make sure the monitor has

143+ -- been fully created before allowing the remote process to terminate,

144+ -- otherwise we might get a different signal here

145+ threadDelay 100000

146+ putMVar monitorSetup ()

147+ ProcessDied ref' pid (DiedException err') <- expect

148+ True <- return $ ref' == ref && pid == theirAddr && err' == show err

149+ return ()

150+ putMVar done ()

151+

152+ takeMVar done

119

153

120

154

121

155

@@ -226,8 +260,9 @@ main = do

226 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

260 Right transport <- createTransport "127.0.0.1" "8080" defaultTCPParameters

227 runTests

261 runTests

228 [ --("Ping", testPing transport)

262 [ --("Ping", testPing transport)

229− ("MonitorUnreachable", testMonitorUnreachable transport)

263+ ("MonitorUnreachable", testMonitorUnreachable transport)

230− , ("MonitorNormalTermination", testMonitorNormalTermination transport)

264+ , ("MonitorNormalTermination", testMonitorNormalTermination transport)

265+ , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)

231 {-

266 {-

232 , ("Monitor2", testMonitor2 transport)

267 , ("Monitor2", testMonitor2 transport)

233 , ("Monitor3", testMonitor3 transport)

268 , ("Monitor3", testMonitor3 transport)