cb444b2 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Monitor dead local process

cb444b2Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+39−38

compare with master

unifiedsplit

@@ -297,6 +297,8 @@ data DiedReason =

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

298 | DiedDisconnect

299 | DiedNodeDown

300+ | DiedNoProc

301+ deriving Show

300

302

301type Identifier = Either ProcessId NodeId

303type Identifier = Either ProcessId NodeId

302

304

@@ -363,6 +365,13 @@ ctrlSendTo them payload = do

363 , ctrlMsgSignal = Died them DiedDisconnect

365 , ctrlMsgSignal = Died them DiedDisconnect

364 }

366 }

365

367

368+ctrlSendLocal :: LocalProcessId -> Message -> NodeCtrl ()

369+ctrlSendLocal lpid msg = do

370+ node <- ask

371+ liftIO $ do

372+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)

373+ forM_ mProc $ \proc -> enqueue (processQueue proc) msg

374+

366ctrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)

375ctrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)

367ctrlConnTo them = do

376ctrlConnTo them = do

368 mConn <- gets (^. nodeCtrlConnFor them)

377 mConn <- gets (^. nodeCtrlConnFor them)

@@ -402,18 +411,23 @@ createCtrlConnTo them = do

402processCtrlMsg :: NodeCtrlMsg -> NodeCtrl ()

411processCtrlMsg :: NodeCtrlMsg -> NodeCtrl ()

403

412

404-- [Unified: Table 10]

413-- [Unified: Table 10]

405−processCtrlMsg (NodeCtrlMsg from (Monitor them ref)) = do

414+processCtrlMsg (NodeCtrlMsg (Left from) (Monitor them ref)) = do

406 node <- ask

415 node <- ask

407 shouldLink <-

416 shouldLink <-

408 if processNodeId them /= localNodeId node

417 if processNodeId them /= localNodeId node

409 then return True

418 then return True

410 else liftIO . withMVar (localState node) $

419 else liftIO . withMVar (localState node) $

411 return . isJust . (^. localProcessWithId (processLocalId them))

420 return . isJust . (^. localProcessWithId (processLocalId them))

421+ let localFrom = processNodeId from == localNodeId node

412− if shouldLink

422+ case (shouldLink, localFrom) of

413− then

423+ (True, _) -> -- [Unified: first rule]

414− modify $ nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref

424+ modify $ nodeCtrlMonsFor them from ^: Set.insert ref

415− else

425+ (False, True) -> -- [Unified: second rule]

426+ ctrlSendLocal (processLocalId from) . createMessage $

427+ ProcessDied ref them DiedNoProc

428+

416− fail "processCtrlMsg: TODO"

429+processCtrlMsg (NodeCtrlMsg (Right _) (Monitor _ _)) =

430+ error "Monitor message from a node?"

417

431

418-- [Unified: Table 12, bottom rule]

432-- [Unified: Table 12, bottom rule]

419processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do

433processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do

@@ -427,15 +441,11 @@ processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do

427 forM_ (Map.toList mons) $ \(them, uss) -> do

441 forM_ (Map.toList mons) $ \(them, uss) -> do

428 forM_ (Map.toList uss) $ \(us, refs) -> do

442 forM_ (Map.toList uss) $ \(us, refs) -> do

429 -- We only need to notify local processes

443 -- We only need to notify local processes

430− when (processNodeId us == localNodeId node) . liftIO $ do

444+ when (processNodeId us == localNodeId node) $ do

431 let lpid = processLocalId us

445 let lpid = processLocalId us

432 rpid = ProcessId nid them

446 rpid = ProcessId nid them

433 forM_ refs $ \ref -> do

447 forM_ refs $ \ref -> do

434− mProc <- withMVar (localState node) $

435− return . (^. localProcessWithId lpid)

436− forM_ mProc $ \proc ->

437− enqueue (processQueue proc) . createMessage $

438− ProcessDied ref rpid reason

448+ ctrlSendLocal lpid . createMessage $ ProcessDied ref rpid reason

439

449

440 modify $ (nodeCtrlLinksForNode nid ^= Map.empty)

450 modify $ (nodeCtrlLinksForNode nid ^= Map.empty)

441 . (nodeCtrlMonsForNode nid ^= Map.empty)

451 . (nodeCtrlMonsForNode nid ^= Map.empty)

@@ -453,19 +463,15 @@ processCtrlMsg (NodeCtrlMsg _from (Died (Left pid) reason)) = do

453 forM_ refs $ \ref -> do

463 forM_ refs $ \ref -> do

454 let msg = createMessage $ ProcessDied ref pid reason

464 let msg = createMessage $ ProcessDied ref pid reason

455 if processNodeId us == localNodeId node

465 if processNodeId us == localNodeId node

456− then liftIO $ do

466+ then

457− let lpid = processLocalId us

467+ ctrlSendLocal (processLocalId us) msg

458− mProc <- withMVar (localState node) $

459− return . (^. localProcessWithId lpid)

460− forM_ mProc $ \proc ->

461− enqueue (processQueue proc) msg

462− else do

468+ else

463 ctrlSendTo (Right . processNodeId $ us) . BSL.toChunks . encode $

469 ctrlSendTo (Right . processNodeId $ us) . BSL.toChunks . encode $

464 NodeCtrlMsg

470 NodeCtrlMsg

465 { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

471 { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?

466 , ctrlMsgSignal = Died (Left pid) reason

472 , ctrlMsgSignal = Died (Left pid) reason

467 }

473 }

468−

474+

469 modify $ (nodeCtrlLinksForProcess pid ^= Set.empty)

475 modify $ (nodeCtrlLinksForProcess pid ^= Set.empty)

470 . (nodeCtrlMonsForProcess pid ^= Map.empty)

476 . (nodeCtrlMonsForProcess pid ^= Map.empty)

471

477

@@ -528,24 +534,17 @@ forkProcess node proc = do

528 , lproc

534 , lproc

529 )

535 )

530 void . forkIO $ do

536 void . forkIO $ do

531− catch (do

537+ let pid = processId lproc

538+ lpid = processLocalId pid

532− runLocalProcess lproc proc

539+ reason <- catch (runLocalProcess lproc proc >> return DiedNormal)

540+ (return . DiedException . (show :: SomeException -> String))

533− -- [Unified: Table 4]

541+ -- [Unified: Table 4, rules termination and exiting]

534− let pid = processId lproc

542+ modifyMVar_ (localState node) $

535− writeChan (localCtrlChan node) $ NodeCtrlMsg

543+ return . (localProcessWithId lpid ^= Nothing)

536− { ctrlMsgSender = Left pid

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+ writeChan (localCtrlChan node) $ NodeCtrlMsg

544− { ctrlMsgSender = Left pid

545+ { ctrlMsgSender = Left pid

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

546+ , ctrlMsgSignal = Died (Left pid) reason

546− }

547+ }

547− )

548−

549 return (processId lproc)

548 return (processId lproc)

550

549

551handleIncomingMessages :: LocalNode -> IO ()

550handleIncomingMessages :: LocalNode -> IO ()

@@ -768,6 +767,7 @@ instance Binary DiedReason where

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

767 put (DiedException e) = putWord8 1 >> put e

769 put DiedDisconnect = putWord8 2

768 put DiedDisconnect = putWord8 2

770 put DiedNodeDown = putWord8 3

769 put DiedNodeDown = putWord8 3

770+ put DiedNoProc = putWord8 4

771 get = do

772 header <- getWord8

773 case header of

@@ -775,6 +775,7 @@ instance Binary DiedReason where

775 1 -> DiedException <$> get

776 2 -> return DiedDisconnect

777 3 -> return DiedNodeDown

778+ 4 -> return DiedNoProc

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

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

779

780

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

781--------------------------------------------------------------------------------

distributed-process/tests/TestCH.hs+23

@@ -151,7 +151,29 @@ testMonitorAbnormalTermination transport = do

151

152 takeMVar done

153

154+-- | Monitor a local process that is already dead

155+testMonitorLocalDeadProcess :: NT.Transport -> IO ()

156+testMonitorLocalDeadProcess transport = do

157+ processDead <- newEmptyMVar

158+ processAddr <- newEmptyMVar

159+ localNode <- newLocalNode transport

160+ done <- newEmptyMVar

161+

162+ forkIO $ do

163+ addr <- forkProcess localNode . liftIO $ putMVar processDead ()

164+ putMVar processAddr addr

154

165

166+ forkIO $ do

167+ theirAddr <- readMVar processAddr

168+ readMVar processDead

169+ runProcess localNode $ do

170+ ref <- monitor theirAddr

171+ ProcessDied ref' pid DiedNoProc <- expect

172+ True <- return $ ref' == ref && pid == theirAddr

173+ return ()

174+ putMVar done ()

175+

176+ takeMVar done

155

177

156{-

178{-

157-- Like 'testMonitor1', but throw an exception instead

179-- Like 'testMonitor1', but throw an exception instead

@@ -263,6 +285,7 @@ main = do

263 ("MonitorUnreachable", testMonitorUnreachable transport)

285 ("MonitorUnreachable", testMonitorUnreachable transport)

264 , ("MonitorNormalTermination", testMonitorNormalTermination transport)

286 , ("MonitorNormalTermination", testMonitorNormalTermination transport)

265 , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)

287 , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)

288+ , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport)

266 {-

289 {-

267 , ("Monitor2", testMonitor2 transport)

290 , ("Monitor2", testMonitor2 transport)

268 , ("Monitor3", testMonitor3 transport)

291 , ("Monitor3", testMonitor3 transport)