826dc6b · distributed-process

187 { localNodeId :: NodeId

188 , localEndPoint :: NT.EndPoint

189 , localState :: MVar LocalNodeState

190− , localCtrlChan :: Chan NodeCtrlMsg

190+ , localCtrlChan :: Chan NCMsg

191 }

192

193-- | Local node state

246 us <- getSelfPid

247 monitorRef <- getMonitorRef

248 ourCtrlChan <- localCtrlChan . processNode <$> ask

249− liftIO . writeChan ourCtrlChan $ NodeCtrlMsg

249+ liftIO . writeChan ourCtrlChan $ NCMsg

250 { ctrlMsgSender = Left us

251 , ctrlMsgSignal = Monitor them monitorRef

252 }

302

303type Identifier = Either ProcessId NodeId

304

305−data NodeCtrlMsg = NodeCtrlMsg

305+data NCMsg = NCMsg

306 { ctrlMsgSender :: Identifier

307 , ctrlMsgSignal :: ProcessSignal

308 }

309

310−data NodeCtrlState = NodeCtrlState

310+data NCState = NCState

311 { -- Mapping from remote processes to linked local processes

312− _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))

312+ _ncLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))

313 -- Mapping from remote processes to monitoring local processes

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

314+ , _ncMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

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

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

317− , _nodeCtrlConns :: Map Identifier NT.Connection

317+ , _ncConns :: Map Identifier NT.Connection

318 }

319

320destNid :: ProcessSignal -> Maybe NodeId

322destNid (Died (Right _) _) = Nothing

323destNid (Died (Left _pid) _) = fail "destNid: TODO"

324

325−initNodeCtrlState :: NodeCtrlState

325+initNCState :: NCState

326−initNodeCtrlState = NodeCtrlState

326+initNCState = NCState

327− { _nodeCtrlLinks = Map.empty

327+ { _ncLinks = Map.empty

328− , _nodeCtrlMons = Map.empty

328+ , _ncMons = Map.empty

329− , _nodeCtrlConns = Map.empty

329+ , _ncConns = Map.empty

330 }

331

332−newtype NodeCtrl a = NodeCtrl { unNodeCtrl :: ReaderT LocalNode (StateT NodeCtrlState IO) a }

332+newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }

333− deriving (Functor, Monad, MonadIO, MonadState NodeCtrlState, MonadReader LocalNode)

333+ deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)

334

335−nodeController :: NodeCtrl ()

335+nodeController :: NC ()

336nodeController = forever $ do

337 node <- ask

338 msg <- liftIO $ readChan (localCtrlChan node)

342 Just nid' | nid' /= localNodeId node -> ncSendCtrlMsg nid' msg

343 _ -> return ()

344

345− processCtrlMsg msg

345+ ncEffect msg

346

347−ctrlSendTo :: Identifier -> [BSS.ByteString] -> NodeCtrl ()

347+ctrlSendTo :: Identifier -> [BSS.ByteString] -> NC ()

348ctrlSendTo them payload = do

349− mConn <- ctrlConnTo them

349+ mConn <- ncConnTo them

350 didSend <- case mConn of

351 Just conn -> do

352 didSend <- liftIO $ NT.send conn payload

358 unless didSend $ do

359 -- [Unified: Table 9, rule node_disconnect]

360 node <- ask

361− liftIO . writeChan (localCtrlChan node) $ NodeCtrlMsg

361+ liftIO . writeChan (localCtrlChan node) $ NCMsg

362 { ctrlMsgSender = them

363 , ctrlMsgSignal = Died them DiedDisconnect

364 }

365

366−ncSendCtrlMsg :: NodeId -> NodeCtrlMsg -> NodeCtrl ()

366+ncSendCtrlMsg :: NodeId -> NCMsg -> NC ()

367ncSendCtrlMsg dest = ctrlSendTo (Right dest) . BSL.toChunks . encode

368

369−ctrlSendLocal :: LocalProcessId -> Message -> NodeCtrl ()

369+ncSendLocal :: LocalProcessId -> Message -> NC ()

370−ctrlSendLocal lpid msg = do

370+ncSendLocal lpid msg = do

371 node <- ask

372 liftIO $ do

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

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

375

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

376+ncConnTo :: Identifier -> NC (Maybe NT.Connection)

377−ctrlConnTo them = do

377+ncConnTo them = do

378− mConn <- gets (^. nodeCtrlConnFor them)

378+ mConn <- gets (^. ncConnFor them)

379 case mConn of

380 Just conn -> return (Just conn)

381− Nothing -> createCtrlConnTo them

381+ Nothing -> ncCreateConnTo them

382

383−createCtrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)

383+ncCreateConnTo :: Identifier -> NC (Maybe NT.Connection)

384−createCtrlConnTo them = do

384+ncCreateConnTo them = do

385 node <- ask

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

387 addr

394 Left _ ->

395 return Nothing

396 Right () -> do

397− modify $ nodeCtrlConnFor them ^= Just conn

397+ modify $ ncConnFor them ^= Just conn

398 return $ Just conn

399 Left _ ->

400 return Nothing

409

410

411-- [Unified: ncEffect]

412−processCtrlMsg :: NodeCtrlMsg -> NodeCtrl ()

412+ncEffect :: NCMsg -> NC ()

413

414-- [Unified: Table 10]

415−processCtrlMsg (NodeCtrlMsg (Left from) (Monitor them ref)) = do

415+ncEffect (NCMsg (Left from) (Monitor them ref)) = do

416 node <- ask

417 shouldLink <-

418 if processNodeId them /= localNodeId node

422 let localFrom = processNodeId from == localNodeId node

423 case (shouldLink, localFrom) of

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

425− modify $ nodeCtrlMonsFor them from ^: Set.insert ref

425+ modify $ ncMonsFor them from ^: Set.insert ref

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

427− ctrlSendLocal (processLocalId from) . createMessage $

427+ ncSendLocal (processLocalId from) . createMessage $

428 ProcessDied ref them DiedNoProc

429 (False, False) -> -- [Unified: third rule]

430− ncSendCtrlMsg (processNodeId from) $ NodeCtrlMsg

430+ ncSendCtrlMsg (processNodeId from) NCMsg

431 { ctrlMsgSender = Right (localNodeId node)

432 , ctrlMsgSignal = Died (Left them) DiedNoProc

433 }

434

435−processCtrlMsg (NodeCtrlMsg (Right _) (Monitor _ _)) =

435+ncEffect (NCMsg (Right _) (Monitor _ _)) =

436 error "Monitor message from a node?"

437

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

439−processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do

439+ncEffect (NCMsg _from (Died (Right nid) reason)) = do

440 node <- ask

441− links <- gets (^. nodeCtrlLinksForNode nid)

441+ links <- gets (^. ncLinksForNode nid)

442− mons <- gets (^. nodeCtrlMonsForNode nid)

442+ mons <- gets (^. ncMonsForNode nid)

443

444− forM_ (Map.toList links) $ \(_them, _uss) -> do

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

445− fail "processCtrlMsg: linking not implemented"

445+ fail "ncEffect: linking not implemented"

446

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

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

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

448+ forM_ (Map.toList uss) $ \(us, refs) ->

449 -- We only need to notify local processes

450 when (processNodeId us == localNodeId node) $ do

451 let lpid = processLocalId us

452 rpid = ProcessId nid them

453− forM_ refs $ \ref -> do

453+ forM_ refs $ \ref ->

454− ctrlSendLocal lpid . createMessage $ ProcessDied ref rpid reason

454+ ncSendLocal lpid . createMessage $ ProcessDied ref rpid reason

455

456− modify $ (nodeCtrlLinksForNode nid ^= Map.empty)

456+ modify $ (ncLinksForNode nid ^= Map.empty)

457− . (nodeCtrlMonsForNode nid ^= Map.empty)

457+ . (ncMonsForNode nid ^= Map.empty)

458

459-- [Unified: Table 12, top rule]

460−processCtrlMsg (NodeCtrlMsg _from (Died (Left pid) reason)) = do

460+ncEffect (NCMsg _from (Died (Left pid) reason)) = do

461 node <- ask

462− links <- gets (^. nodeCtrlLinksForProcess pid)

462+ links <- gets (^. ncLinksForProcess pid)

463− mons <- gets (^. nodeCtrlMonsForProcess pid)

463+ mons <- gets (^. ncMonsForProcess pid)

464

465 forM_ links $ \_us ->

466− fail "processCtrlMsg: linking not implemented"

466+ fail "ncEffect: linking not implemented"

467

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

469 forM_ refs $ \ref -> do

470 let msg = createMessage $ ProcessDied ref pid reason

471 if processNodeId us == localNodeId node

472 then

473− ctrlSendLocal (processLocalId us) msg

473+ ncSendLocal (processLocalId us) msg

474 else

475− ncSendCtrlMsg (processNodeId us) $ NodeCtrlMsg

475+ ncSendCtrlMsg (processNodeId us) NCMsg

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

477 , ctrlMsgSignal = Died (Left pid) reason

478 }

479

480− modify $ (nodeCtrlLinksForProcess pid ^= Set.empty)

480+ modify $ (ncLinksForProcess pid ^= Set.empty)

481− . (nodeCtrlMonsForProcess pid ^= Map.empty)

481+ . (ncMonsForProcess pid ^= Map.empty)

482

483--------------------------------------------------------------------------------

484-- Initialization --

504 , localState = state

505 , localCtrlChan = ctrlChan

506 }

507− void . forkIO $ evalStateT (runReaderT (unNodeCtrl nodeController) node)

507+ void . forkIO $ evalStateT (runReaderT (unNC nodeController) node)

508− initNodeCtrlState

508+ initNCState

509 void . forkIO $ handleIncomingMessages node

510 return node

511

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

547 modifyMVar_ (localState node) $

548 return . (localProcessWithId lpid ^= Nothing)

549− writeChan (localCtrlChan node) $ NodeCtrlMsg

549+ writeChan (localCtrlChan node) NCMsg

550 { ctrlMsgSender = Left pid

551 , ctrlMsgSignal = Died (Left pid) reason

552 }

570 let msg = payloadToMessage payload

571 enqueue (processQueue proc) msg

572 go uninitConns procConns ctrlConns

573− Nothing -> if cid `Set.member` ctrlConns

573+ Nothing | cid `Set.member` ctrlConns ->

574− then writeChan ctrlChan (decode . BSL.fromChunks $ payload)

574+ writeChan ctrlChan (decode . BSL.fromChunks $ payload)

575− else if cid `elem` uninitConns

575+ Nothing | cid `elem` uninitConns ->

576− then case payloadToId payload of

576+ case payloadToId payload of

577− Just lpid -> do

577+ Just lpid -> do

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

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

579− case mProc of

579+ case mProc of

580− Just proc ->

580+ Just proc ->

581− go (List.delete cid uninitConns)

581+ go (List.delete cid uninitConns)

582− (Map.insert cid proc procConns)

582+ (Map.insert cid proc procConns)

583− ctrlConns

583+ ctrlConns

584− Nothing ->

584+ Nothing ->

585− fail "handleIncomingMessages: TODO 1"

585+ fail "handleIncomingMessages: TODO 1"

586− Nothing ->

586+ Nothing ->

587− go (List.delete cid uninitConns)

587+ go (List.delete cid uninitConns)

588− procConns

588+ procConns

589− (Set.insert cid ctrlConns)

589+ (Set.insert cid ctrlConns)

590+ _ ->

590− else fail "handleIncomingMessages: TODO 2"

591+ fail "handleIncomingMessages: TODO 2"

591 NT.ConnectionClosed cid ->

592 NT.ConnectionClosed cid ->

592 go (List.delete cid uninitConns)

593 go (List.delete cid uninitConns)

593 (Map.delete cid procConns)

594 (Map.delete cid procConns)

691 (bs2, bs3) = BSS.splitAt 4 bss' in

692 (bs2, bs3) = BSS.splitAt 4 bss' in

692 case NTI.decodeInt32 bs1 :: Int of

693 case NTI.decodeInt32 bs1 :: Int of

693 0 -> Nothing

694 0 -> Nothing

694− 1 -> Just $ LocalProcessId

695+ 1 -> Just LocalProcessId

695 { lpidCounter = NTI.decodeInt32 bs2

696 { lpidCounter = NTI.decodeInt32 bs2

696 , lpidUnique = NTI.decodeInt32 bs3

697 , lpidUnique = NTI.decodeInt32 bs3

697 }

698 }

698 _ -> fail "payloadToId"

699 _ -> fail "payloadToId"

699

700

701+-- TODO: this was written before the change to the new semantics

700remoteProcessFailed :: LocalNode -> ProcessId -> IO ()

702remoteProcessFailed :: LocalNode -> ProcessId -> IO ()

701−remoteProcessFailed node them = do

703+remoteProcessFailed node them = liftIO $

702− liftIO $ modifyMVar_ (localState node) $ \st -> do

704+ modifyMVar_ (localState node) $ \st -> do

703 sequence_ (st ^. monitorActionsFor them)

705 sequence_ (st ^. monitorActionsFor them)

704 return (monitorActionsFor them ^= [] $ st)

706 return (monitorActionsFor them ^= [] $ st)

705

707

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

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

754 get = ProcessDied <$> get <*> get <*> get

756 get = ProcessDied <$> get <*> get <*> get

755

757

756−instance Binary NodeCtrlMsg where

758+instance Binary NCMsg where

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

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

758− get = NodeCtrlMsg <$> get <*> get

760+ get = NCMsg <$> get <*> get

759

761

760instance Binary ProcessSignal where

762instance Binary ProcessSignal where

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

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

814connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)

816connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)

815connectionTo pid = connections >>> DAC.mapMaybe pid

817connectionTo pid = connections >>> DAC.mapMaybe pid

816

818

817−nodeCtrlLinks :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Set ProcessId)))

819+ncLinks :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))

818−nodeCtrlLinks = accessor _nodeCtrlLinks (\links st -> st { _nodeCtrlLinks = links })

820+ncLinks = accessor _ncLinks (\links st -> st { _ncLinks = links })

819

821

820−nodeCtrlMons :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))

822+ncMons :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))

821−nodeCtrlMons = accessor _nodeCtrlMons (\mons st -> st { _nodeCtrlMons = mons })

823+ncMons = accessor _ncMons (\mons st -> st { _ncMons = mons })

822

824

823−nodeCtrlConns :: Accessor NodeCtrlState (Map Identifier NT.Connection)

825+ncConns :: Accessor NCState (Map Identifier NT.Connection)

824−nodeCtrlConns = accessor _nodeCtrlConns (\conns st -> st { _nodeCtrlConns = conns })

826+ncConns = accessor _ncConns (\conns st -> st { _ncConns = conns })

825

827

826−nodeCtrlConnFor :: Identifier -> Accessor NodeCtrlState (Maybe NT.Connection)

828+ncConnFor :: Identifier -> Accessor NCState (Maybe NT.Connection)

827−nodeCtrlConnFor them = nodeCtrlConns >>> DAC.mapMaybe them

829+ncConnFor them = ncConns >>> DAC.mapMaybe them

828

830

829−nodeCtrlLinksForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Set ProcessId))

831+ncLinksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))

830−nodeCtrlLinksForNode nid = nodeCtrlLinks >>> DAC.mapDefault Map.empty nid

832+ncLinksForNode nid = ncLinks >>> DAC.mapDefault Map.empty nid

831

833

832−nodeCtrlMonsForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

834+ncMonsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))

833−nodeCtrlMonsForNode nid = nodeCtrlMons >>> DAC.mapDefault Map.empty nid

835+ncMonsForNode nid = ncMons >>> DAC.mapDefault Map.empty nid

834

836

835−nodeCtrlLinksForProcess :: ProcessId -> Accessor NodeCtrlState (Set ProcessId)

837+ncLinksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)

836−nodeCtrlLinksForProcess pid = nodeCtrlLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)

838+ncLinksForProcess pid = ncLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)

837

839

838−nodeCtrlMonsForProcess :: ProcessId -> Accessor NodeCtrlState (Map ProcessId (Set MonitorRef))

840+ncMonsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))

839−nodeCtrlMonsForProcess pid = nodeCtrlMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)

841+ncMonsForProcess pid = ncMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)

840−

841−nodeCtrlMonsFor :: ProcessId -> ProcessId -> Accessor NodeCtrlState (Set MonitorRef)

842−nodeCtrlMonsFor them us = nodeCtrlMonsForProcess them >>> DAC.mapDefault Set.empty us

843−

844−

845−--------------------------------------------------------------------------------

846−-- Auxiliary --

847−--------------------------------------------------------------------------------

848−

849−fromLeft :: Either a b -> a

850−fromLeft (Left a) = a

851−fromLeft (Right _) = error "fromLeft"

852

842

843+ncMonsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)

844+ncMonsFor them us = ncMonsForProcess them >>> DAC.mapDefault Set.empty us