743e579 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Continue work on monitoring.

743e579Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+187−134

compare with master

unifiedsplit

@@ -98,14 +98,15 @@ import Data.Map (Map)

98import qualified Data.Map as Map (empty, lookup, insert, delete, keys, toList)

99import qualified Data.List as List (delete)

100import Data.Set (Set)

101−import qualified Data.Set as Set (empty, insert)

101+import qualified Data.Set as Set (empty, insert, delete, member)

102import Data.Int (Int32)

103import Data.Typeable (Typeable)

104import Data.Foldable (forM_)

105−import Data.Maybe (isNothing)

105+import Data.Maybe (isJust)

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

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

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

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

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

109import Control.Applicative ((<$>))

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

110import Control.Category ((>>>))

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

111import Control.Exception (Exception, throwIO, catch)

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

@@ -304,10 +305,10 @@ data NodeCtrlMsg = NodeCtrlMsg

304 }

305 }

305

306

306data NodeCtrlState = NodeCtrlState

307data NodeCtrlState = NodeCtrlState

307− { -- Mapping from remote processes to linked local processes

308+ { -- Mapping from remote processes to linked local processes

308 _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))

309 _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))

309 -- Mapping from remote processes to monitoring local processes

310 -- Mapping from remote processes to monitoring local processes

310− , _nodeCtrlMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set (MonitorRef))))

311+ , _nodeCtrlMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

311 -- The node controller maintains its own set of connections

312 -- The node controller maintains its own set of connections

312 -- TODO: still not convinced that this is correct

313 -- TODO: still not convinced that this is correct

313 , _nodeCtrlConns :: Map Identifier NT.Connection

314 , _nodeCtrlConns :: Map Identifier NT.Connection

@@ -325,118 +326,147 @@ initNodeCtrlState = NodeCtrlState

325 , _nodeCtrlConns = Map.empty

326 , _nodeCtrlConns = Map.empty

326 }

327 }

327

328

328−nodeController :: LocalNode -> IO ()

329+newtype NodeCtrl a = NodeCtrl { unNodeCtrl :: ReaderT LocalNode (StateT NodeCtrlState IO) a }

330+ deriving (Functor, Monad, MonadIO, MonadState NodeCtrlState, MonadReader LocalNode)

331+

329−nodeController node = go initNodeCtrlState

332+nodeController :: NodeCtrl ()

333+nodeController = forever $ do

334+ node <- ask

335+ msg <- liftIO $ readChan (localCtrlChan node)

336+

337+ -- Forward the message if appropriate

338+ case destNid (ctrlMsgSignal msg) of

339+ Just nid' | nid' /= localNodeId node ->

340+ ctrlSendTo (Right nid') . BSL.toChunks . encode $ msg

341+ _ ->

342+ return ()

343+

344+ processCtrlMsg msg

345+

346+ctrlSendTo :: Identifier -> [BSS.ByteString] -> NodeCtrl ()

347+ctrlSendTo them payload = do

348+ mConn <- ctrlConnTo them

349+ didSend <- case mConn of

350+ Just conn -> do

351+ didSend <- liftIO $ NT.send conn payload

352+ case didSend of

353+ Left _ -> return False

354+ Right () -> return True

355+ Nothing ->

356+ return False

357+ unless didSend $ do

358+ -- [Unified: Table 9, rule node_disconnect]

359+ node <- ask

360+ liftIO . writeChan (localCtrlChan node) $ NodeCtrlMsg

361+ { ctrlMsgSender = them

362+ , ctrlMsgSignal = Died them DiedDisconnect

363+ }

364+

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

366+ctrlConnTo them = do

367+ mConn <- gets (^. nodeCtrlConnFor them)

368+ case mConn of

369+ Just conn -> return (Just conn)

370+ Nothing -> createCtrlConnTo them

371+

372+createCtrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)

373+createCtrlConnTo them = do

374+ node <- ask

375+ mConn <- liftIO $ NT.connect (localEndPoint node)

376+ addr

377+ NT.ReliableOrdered

378+ NT.defaultConnectHints

379+ case mConn of

380+ Right conn -> do

381+ didSend <- liftIO $ NT.send conn firstMsg

382+ case didSend of

383+ Left _ ->

384+ return Nothing

385+ Right () -> do

386+ modify $ nodeCtrlConnFor them ^= Just conn

387+ return $ Just conn

388+ Left _ ->

389+ return Nothing

330 where

390 where

331− -- [Unified: Table 7]

332− go :: NodeCtrlState -> IO ()

333− go st = do

334− msg <- readChan ourChan

335− st' <- case destNid (ctrlMsgSignal msg) of

336− Just nid' | nid' /= localNodeId node -> do

337− (st', mConn) <- connTo st (Right nid')

338− didSend <- case mConn of

339− Just conn -> do

340− didSend <- NT.send conn (BSL.toChunks . encode $ msg)

341− case didSend of

342− Left _ -> return False

343− Right () -> return True

344− Nothing ->

345− return False

346− if didSend

347− then

348− return st'

349− else do

350− writeChan ourChan $ NodeCtrlMsg

351− { ctrlMsgSender = Right nid'

352− , ctrlMsgSignal = Died (Right nid') DiedDisconnect

353− }

354− return st'

355− _ ->

356− -- No need to forward

357− return st

358− st'' <- processCtrlMsg node st' msg

359− go st''

360−

361− ourChan = localCtrlChan node

362− ourEndPoint = localEndPoint node

363−

364− connTo :: NodeCtrlState

365− -> Identifier

366− -> IO (NodeCtrlState, Maybe NT.Connection)

367− connTo st them = do

368− case st ^. nodeCtrlConnFor them of

369− Nothing -> do

370− (addr, firstMsg) <- return $ case them of

391+ (addr, firstMsg) = case them of

371− Left pid -> ( nodeAddress (processNodeId pid)

392+ Left pid -> ( nodeAddress (processNodeId pid)

372− , idToPayload (Just $ processLocalId pid)

393+ , idToPayload (Just $ processLocalId pid)

373− )

394+ )

374− Right nid -> ( nodeAddress nid

395+ Right nid -> ( nodeAddress nid

375− , idToPayload Nothing

396+ , idToPayload Nothing

376− )

397+ )

377− mConn <- NT.connect ourEndPoint

398+

378− addr

379− NT.ReliableOrdered

380− NT.defaultConnectHints

381− case mConn of

382− Right conn -> do

383− didSend <- NT.send conn firstMsg

384− case didSend of

385− Left _ ->

386− return (st, Nothing)

387− Right () ->

388− return (nodeCtrlConnFor them ^= Just conn $ st, Just conn)

389− Left _ ->

390− return (st, Nothing)

391− Just conn ->

392− return (st, Just conn)

393

399

394-- [Unified: ncEffect]

400-- [Unified: ncEffect]

395−processCtrlMsg :: LocalNode

401+processCtrlMsg :: NodeCtrlMsg -> NodeCtrl ()

396− -> NodeCtrlState

402+

397− -> NodeCtrlMsg

398− -> IO (NodeCtrlState)

399-- [Unified: Table 10]

403-- [Unified: Table 10]

400−processCtrlMsg node st (NodeCtrlMsg from (Monitor them ref)) = do

404+processCtrlMsg (NodeCtrlMsg from (Monitor them ref)) = do

405+ node <- ask

401 shouldLink <-

406 shouldLink <-

402 if processNodeId them /= localNodeId node

407 if processNodeId them /= localNodeId node

403 then return True

408 then return True

404− else withMVar (localState node) $

409+ else liftIO . withMVar (localState node) $

405− return . isNothing . (^. localProcessWithId (processLocalId them))

410+ return . isJust . (^. localProcessWithId (processLocalId them))

406 if shouldLink

411 if shouldLink

407 then

412 then

408− return (nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref $ st)

413+ modify $ nodeCtrlMonsFor them (fromLeft from) ^: Set.insert ref

409 else

414 else

410 fail "processCtrlMsg: TODO"

415 fail "processCtrlMsg: TODO"

411

416

412−-- [Unified Table 9 and Table 12]

417+-- [Unified: Table 12, bottom rule]

413−processCtrlMsg node st (NodeCtrlMsg _from (Died (Right nid) reason)) = do

418+processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do

414− let localLinks :: Map LocalProcessId (Set ProcessId)

419+ node <- ask

415− localLinks = st ^. nodeCtrlLinksForNode nid

420+ links <- gets (^. nodeCtrlLinksForNode nid)

421+ mons <- gets (^. nodeCtrlMonsForNode nid)

416

422

417− localMons :: Map LocalProcessId (Map ProcessId (Set MonitorRef))

423+ forM_ (Map.toList links) $ \(_them, _uss) -> do

418− localMons = st ^. nodeCtrlMonsForNode nid

424+ fail "processCtrlMsg: linking not implemented"

419

425

420− forM_ (Map.toList localLinks) $ \(_them, _uss) -> do

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

421− fail "disconnected: linking not implemented"

422−

423− forM_ (Map.toList localMons) $ \(them, uss) -> do

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

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

425 -- We only need to notify local processes

428 -- We only need to notify local processes

426− when (processNodeId us == localNodeId node) $ do

429+ when (processNodeId us == localNodeId node) . liftIO $ do

427 let lpid = processLocalId us

430 let lpid = processLocalId us

428 rpid = ProcessId nid them

431 rpid = ProcessId nid them

429 forM_ refs $ \ref -> do

432 forM_ refs $ \ref -> do

430 mProc <- withMVar (localState node) $

433 mProc <- withMVar (localState node) $

431 return . (^. localProcessWithId lpid)

434 return . (^. localProcessWithId lpid)

432− forM_ mProc $ \proc ->

435+ forM_ mProc $ \proc ->

433 enqueue (processQueue proc) . createMessage $

436 enqueue (processQueue proc) . createMessage $

434 ProcessDied ref rpid reason

437 ProcessDied ref rpid reason

435

438

436− return . (nodeCtrlLinksForNode nid ^= Map.empty)

439+ modify $ (nodeCtrlLinksForNode nid ^= Map.empty)

437 . (nodeCtrlMonsForNode nid ^= Map.empty)

440 . (nodeCtrlMonsForNode nid ^= Map.empty)

438− $ st

439

441

442+-- [Unified: Table 12, top rule]

443+processCtrlMsg (NodeCtrlMsg _from (Died (Left pid) reason)) = do

444+ node <- ask

445+ links <- gets (^. nodeCtrlLinksForProcess pid)

446+ mons <- gets (^. nodeCtrlMonsForProcess pid)

447+

448+ forM_ links $ \_us ->

449+ fail "processCtrlMsg: linking not implemented"

450+

451+ forM_ (Map.toList mons) $ \(us, refs) ->

452+ forM_ refs $ \ref -> do

453+ let msg = createMessage $ ProcessDied ref pid reason

454+ if processNodeId us == localNodeId node

455+ then liftIO $ do

456+ let lpid = processLocalId us

457+ mProc <- withMVar (localState node) $

458+ return . (^. localProcessWithId lpid)

459+ forM_ mProc $ \proc ->

460+ enqueue (processQueue proc) msg

461+ else do

462+ ctrlSendTo (Right . processNodeId $ us) . BSL.toChunks . encode $

463+ NodeCtrlMsg

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

465+ , ctrlMsgSignal = Died (Left pid) reason

466+ }

467+

468+ modify $ (nodeCtrlLinksForProcess pid ^= Set.empty)

469+ . (nodeCtrlMonsForProcess pid ^= Map.empty)

440

470

441--------------------------------------------------------------------------------

471--------------------------------------------------------------------------------

442-- Initialization --

472-- Initialization --

@@ -462,7 +492,8 @@ newLocalNode transport = do

462 , localState = state

492 , localState = state

463 , localCtrlChan = ctrlChan

493 , localCtrlChan = ctrlChan

464 }

494 }

465− void . forkIO $ nodeController node

495+ void . forkIO $ evalStateT (runReaderT (unNodeCtrl nodeController) node)

496+ initNodeCtrlState

466 void . forkIO $ handleIncomingMessages node

497 void . forkIO $ handleIncomingMessages node

467 return node

498 return node

468

499

@@ -474,59 +505,77 @@ runProcess node proc = do

474

505

475forkProcess :: LocalNode -> Process () -> IO ProcessId

506forkProcess :: LocalNode -> Process () -> IO ProcessId

476forkProcess node proc = do

507forkProcess node proc = do

477− queue <- newCQueue

508+ queue <- newCQueue

478− lproc <- modifyMVar (localState node) $ \st -> do

509+ lproc <- modifyMVar (localState node) $ \st -> do

479− let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

510+ let lpid = LocalProcessId { lpidCounter = st ^. localPidCounter

480− , lpidUnique = st ^. localPidUnique

511+ , lpidUnique = st ^. localPidUnique

512+ }

513+ let pid = ProcessId { processNodeId = localNodeId node

514+ , processLocalId = lpid

515+ }

516+ pst <- newMVar LocalProcessState { _connections = Map.empty

517+ }

518+ let lproc = LocalProcess { processQueue = queue

519+ , processNode = node

520+ , processId = pid

521+ , processState = pst

481 }

522 }

482− let pid = ProcessId { processNodeId = localNodeId node

483− , processLocalId = lpid

484− }

485− pst <- newMVar LocalProcessState { _connections = Map.empty

486− }

487− let lproc = LocalProcess { processQueue = queue

488− , processNode = node

489− , processId = pid

490− , processState = pst

491− }

492− -- TODO: if the counter overflows we should pick a new unique

523+ -- TODO: if the counter overflows we should pick a new unique

493− return ( (localProcessWithId lpid ^= Just lproc)

524+ return ( (localProcessWithId lpid ^= Just lproc)

494− . (localPidCounter ^: (+ 1))

525+ . (localPidCounter ^: (+ 1))

495− $ st

526+ $ st

496− , lproc

527+ , lproc

497− )

528+ )

529+ void . forkIO $ do

498− void . forkIO $ runLocalProcess lproc proc

530+ runLocalProcess lproc proc

531+ -- [Unified: Table 4]

532+ let pid = processId lproc

533+ writeChan (localCtrlChan node) $ NodeCtrlMsg

534+ { ctrlMsgSender = Left pid

535+ , ctrlMsgSignal = Died (Left pid) DiedNormal

536+ }

499− return (processId lproc)

537+ return (processId lproc)

500

538

501handleIncomingMessages :: LocalNode -> IO ()

539handleIncomingMessages :: LocalNode -> IO ()

502−handleIncomingMessages node = go [] Map.empty

540+handleIncomingMessages node = go [] Map.empty Set.empty

503 where

541 where

542+ go :: [NT.ConnectionId] -- ^ Connections whose purpose we don't yet know

504− go :: [NT.ConnectionId] -> Map NT.ConnectionId LocalProcess -> IO ()

543+ -> Map NT.ConnectionId LocalProcess -- ^ Connections to local processes

505− go halfOpenConns openConns = do

544+ -> Set NT.ConnectionId -- ^ Connections to our controller

545+ -> IO ()

546+ go uninitConns procConns ctrlConns = do

506 event <- NT.receive endpoint

547 event <- NT.receive endpoint

507 case event of

548 case event of

508 NT.ConnectionOpened cid _rel _theirAddr ->

549 NT.ConnectionOpened cid _rel _theirAddr ->

509− go (cid : halfOpenConns) openConns

550+ go (cid : uninitConns) procConns ctrlConns

510 NT.Received cid payload ->

551 NT.Received cid payload ->

511− case Map.lookup cid openConns of

552+ case Map.lookup cid procConns of

512 Just proc -> do

553 Just proc -> do

513 let msg = payloadToMessage payload

554 let msg = payloadToMessage payload

514 enqueue (processQueue proc) msg

555 enqueue (processQueue proc) msg

515− go halfOpenConns openConns

556+ go uninitConns procConns ctrlConns

516− Nothing -> if cid `elem` halfOpenConns

557+ Nothing -> if cid `Set.member` ctrlConns

517− then do

558+ then writeChan ctrlChan (decode . BSL.fromChunks $ payload)

559+ else if cid `elem` uninitConns

518− let Just lpid = payloadToId payload

560+ then case payloadToId payload of

561+ Just lpid -> do

519− mProc <- withMVar state $ return . (^. localProcessWithId lpid)

562+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)

520− case mProc of

563+ case mProc of

521− Just proc ->

564+ Just proc ->

522− go (List.delete cid halfOpenConns)

565+ go (List.delete cid uninitConns)

523− (Map.insert cid proc openConns)

566+ (Map.insert cid proc procConns)

567+ ctrlConns

568+ Nothing ->

569+ fail "handleIncomingMessages: TODO 1"

524 Nothing ->

570 Nothing ->

525− fail "handleIncomingMessages: TODO 1"

571+ go (List.delete cid uninitConns)

526− else

572+ procConns

573+ (Set.insert cid ctrlConns)

527− fail "handleIncomingMessages: TODO 2"

574+ else fail "handleIncomingMessages: TODO 2"

528 NT.ConnectionClosed cid ->

575 NT.ConnectionClosed cid ->

529− go (List.delete cid halfOpenConns) (Map.delete cid openConns)

576+ go (List.delete cid uninitConns)

577+ (Map.delete cid procConns)

578+ (Set.delete cid ctrlConns)

530 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do

579 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost theirAddr _) _) -> do

531 -- TODO: we could cache this information

580 -- TODO: we could cache this information

532 pids <- withMVar state $ \st ->

581 pids <- withMVar state $ \st ->

@@ -543,6 +592,7 @@ handleIncomingMessages node = go [] Map.empty

543

592

544 state = localState node

593 state = localState node

545 endpoint = localEndPoint node

594 endpoint = localEndPoint node

595+ ctrlChan = localCtrlChan node

546

596

547--------------------------------------------------------------------------------

597--------------------------------------------------------------------------------

548-- Auxiliary functions --

598-- Auxiliary functions --

@@ -598,8 +648,10 @@ sendBinary them conn payload = do

598 liftIO $ remoteProcessFailed node them

648 liftIO $ remoteProcessFailed node them

599

649

600sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()

650sendMessage :: ProcessId -> NT.Connection -> Message -> Process ()

601−sendMessage them conn (Message fp enc) =

651+sendMessage them conn = sendBinary them conn . messageToPayload

652+

653+messageToPayload :: Message -> [BSS.ByteString]

602− sendBinary them conn $ encodeFingerprint fp : BSL.toChunks enc

654+messageToPayload (Message fp enc) = encodeFingerprint fp : BSL.toChunks enc

603

655

604payloadToMessage :: [BSS.ByteString] -> Message

656payloadToMessage :: [BSS.ByteString] -> Message

605payloadToMessage payload = Message fp msg

657payloadToMessage payload = Message fp msg

@@ -696,6 +748,7 @@ instance Binary ProcessSignal where

696 header <- getWord8

748 header <- getWord8

697 case header of

749 case header of

698 0 -> liftM2 Monitor get get

750 0 -> liftM2 Monitor get get

751+ 1 -> liftM2 Died get get

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

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

700

753

701instance Binary DiedReason where

754instance Binary DiedReason where

distributed-process/tests/TestCH.hs+41−6

@@ -61,15 +61,15 @@ testPing transport = do

61 takeMVar clientDone

64−-- | Set up monitoring, send first message to an unreachable node

64+-- | Monitor an unreachable node

65−testMonitor1 :: NT.Transport -> IO ()

65+testMonitorUnreachable :: NT.Transport -> IO ()

66−testMonitor1 transport = do

66+testMonitorUnreachable transport = do

67 deadProcess <- newEmptyMVar

68 done <- newEmptyMVar

70 forkIO $ do

71 localNode <- newLocalNode transport

72− addr <- forkProcess localNode $ return ()

72+ addr <- forkProcess localNode $ liftIO $ newEmptyMVar >>= readMVar -- deadlock

73 closeLocalNode localNode

74 putMVar deadProcess addr

@@ -85,6 +85,40 @@ testMonitor1 transport = do

86 takeMVar done

88+-- | Monitor a process which terminates normally

89+testMonitorNormalTermination :: NT.Transport -> IO ()

90+testMonitorNormalTermination transport = do

91+ monitorSetup <- newEmptyMVar

92+ monitoredProcess <- newEmptyMVar

93+ done <- newEmptyMVar

94+

95+ forkIO $ do

96+ localNode <- newLocalNode transport

97+ addr <- forkProcess localNode $

98+ liftIO $ readMVar monitorSetup

99+ putMVar monitoredProcess addr

100+

101+ forkIO $ do

102+ localNode <- newLocalNode transport

103+ theirAddr <- readMVar monitoredProcess

104+ runProcess localNode $ do

105+ ref <- monitor theirAddr

106+ liftIO $ do

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

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

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

110+ threadDelay 100000

111+ putMVar monitorSetup ()

112+ ProcessDied ref' pid DiedNormal <- expect

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

114+ return ()

115+ putMVar done ()

116+

117+ takeMVar done

118+

119+

120+

121+

88{-

122{-

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

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

90testMonitor2 :: NT.Transport -> IO ()

124testMonitor2 :: NT.Transport -> IO ()

@@ -191,8 +225,9 @@ main :: IO ()

191main = do

225main = do

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

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

193 runTests

227 runTests

194− [ ("Ping", testPing transport)

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

195− , ("Monitor1", testMonitor1 transport)

229+ ("MonitorUnreachable", testMonitorUnreachable transport)

230+ , ("MonitorNormalTermination", testMonitorNormalTermination transport)

196 {-

231 {-

197 , ("Monitor2", testMonitor2 transport)

232 , ("Monitor2", testMonitor2 transport)

198 , ("Monitor3", testMonitor3 transport)

233 , ("Monitor3", testMonitor3 transport)