@@ -187,7 +187,7 @@ data LocalNode = LocalNode
187187 { localNodeId :: NodeId
188188 , localEndPoint :: NT.EndPoint
189189 , localState :: MVar LocalNodeState
190− , localCtrlChan :: Chan NodeCtrlMsg
190+ , localCtrlChan :: Chan NCMsg
191191 }
192192
193193-- | Local node state
@@ -246,7 +246,7 @@ monitor them = do
246246 us <- getSelfPid
247247 monitorRef <- getMonitorRef
248248 ourCtrlChan <- localCtrlChan . processNode <$> ask
249− liftIO . writeChan ourCtrlChan $ NodeCtrlMsg
249+ liftIO . writeChan ourCtrlChan $ NCMsg
250250 { ctrlMsgSender = Left us
251251 , ctrlMsgSignal = Monitor them monitorRef
252252 }
@@ -302,19 +302,19 @@ data DiedReason =
302302
303303type Identifier = Either ProcessId NodeId
304304
305−data NodeCtrlMsg = NodeCtrlMsg
305+data NCMsg = NCMsg
306306 { ctrlMsgSender :: Identifier
307307 , ctrlMsgSignal :: ProcessSignal
308308 }
309309
310−data NodeCtrlState = NodeCtrlState
310+data NCState = NCState
311311 { -- Mapping from remote processes to linked local processes
312− _nodeCtrlLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
312+ _ncLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
313313 -- 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)))
315315 -- The node controller maintains its own set of connections
316316 -- TODO: still not convinced that this is correct
317− , _nodeCtrlConns :: Map Identifier NT.Connection
317+ , _ncConns :: Map Identifier NT.Connection
318318 }
319319
320320destNid :: ProcessSignal -> Maybe NodeId
@@ -322,17 +322,17 @@ destNid (Monitor pid _) = Just $ processNodeId pid
322322destNid (Died (Right _) _) = Nothing
323323destNid (Died (Left _pid) _) = fail "destNid: TODO"
324324
325−initNodeCtrlState :: NodeCtrlState
326−initNodeCtrlState = NodeCtrlState
327− { _nodeCtrlLinks = Map.empty
328− , _nodeCtrlMons = Map.empty
329− , _nodeCtrlConns = Map.empty
325+initNCState :: NCState
326+initNCState = NCState
327+ { _ncLinks = Map.empty
328+ , _ncMons = Map.empty
329+ , _ncConns = Map.empty
330330 }
331331
332−newtype NodeCtrl a = NodeCtrl { unNodeCtrl :: ReaderT LocalNode (StateT NodeCtrlState IO) a }
333− deriving (Functor, Monad, MonadIO, MonadState NodeCtrlState, MonadReader LocalNode)
332+newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }
333+ deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
334334
335−nodeController :: NodeCtrl ()
335+nodeController :: NC ()
336336nodeController = forever $ do
337337 node <- ask
338338 msg <- liftIO $ readChan (localCtrlChan node)
@@ -342,11 +342,11 @@ nodeController = forever $ do
342342 Just nid' | nid' /= localNodeId node -> ncSendCtrlMsg nid' msg
343343 _ -> return ()
344344
345− processCtrlMsg msg
345+ ncEffect msg
346346
347−ctrlSendTo :: Identifier -> [BSS.ByteString] -> NodeCtrl ()
347+ctrlSendTo :: Identifier -> [BSS.ByteString] -> NC ()
348348ctrlSendTo them payload = do
349− mConn <- ctrlConnTo them
349+ mConn <- ncConnTo them
350350 didSend <- case mConn of
351351 Just conn -> do
352352 didSend <- liftIO $ NT.send conn payload
@@ -358,30 +358,30 @@ ctrlSendTo them payload = do
358358 unless didSend $ do
359359 -- [Unified: Table 9, rule node_disconnect]
360360 node <- ask
361− liftIO . writeChan (localCtrlChan node) $ NodeCtrlMsg
361+ liftIO . writeChan (localCtrlChan node) $ NCMsg
362362 { ctrlMsgSender = them
363363 , ctrlMsgSignal = Died them DiedDisconnect
364364 }
365365
366−ncSendCtrlMsg :: NodeId -> NodeCtrlMsg -> NodeCtrl ()
366+ncSendCtrlMsg :: NodeId -> NCMsg -> NC ()
367367ncSendCtrlMsg dest = ctrlSendTo (Right dest) . BSL.toChunks . encode
368368
369−ctrlSendLocal :: LocalProcessId -> Message -> NodeCtrl ()
370−ctrlSendLocal lpid msg = do
369+ncSendLocal :: LocalProcessId -> Message -> NC ()
370+ncSendLocal lpid msg = do
371371 node <- ask
372372 liftIO $ do
373373 mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
374374 forM_ mProc $ \proc -> enqueue (processQueue proc) msg
375375
376−ctrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)
377−ctrlConnTo them = do
378− mConn <- gets (^. nodeCtrlConnFor them)
376+ncConnTo :: Identifier -> NC (Maybe NT.Connection)
377+ncConnTo them = do
378+ mConn <- gets (^. ncConnFor them)
379379 case mConn of
380380 Just conn -> return (Just conn)
381− Nothing -> createCtrlConnTo them
381+ Nothing -> ncCreateConnTo them
382382
383−createCtrlConnTo :: Identifier -> NodeCtrl (Maybe NT.Connection)
384−createCtrlConnTo them = do
383+ncCreateConnTo :: Identifier -> NC (Maybe NT.Connection)
384+ncCreateConnTo them = do
385385 node <- ask
386386 mConn <- liftIO $ NT.connect (localEndPoint node)
387387 addr
@@ -394,7 +394,7 @@ createCtrlConnTo them = do
394394 Left _ ->
395395 return Nothing
396396 Right () -> do
397− modify $ nodeCtrlConnFor them ^= Just conn
397+ modify $ ncConnFor them ^= Just conn
398398 return $ Just conn
399399 Left _ ->
400400 return Nothing
@@ -409,10 +409,10 @@ createCtrlConnTo them = do
409409
410410
411411-- [Unified: ncEffect]
412−processCtrlMsg :: NodeCtrlMsg -> NodeCtrl ()
412+ncEffect :: NCMsg -> NC ()
413413
414414-- [Unified: Table 10]
415−processCtrlMsg (NodeCtrlMsg (Left from) (Monitor them ref)) = do
415+ncEffect (NCMsg (Left from) (Monitor them ref)) = do
416416 node <- ask
417417 shouldLink <-
418418 if processNodeId them /= localNodeId node
@@ -422,63 +422,63 @@ processCtrlMsg (NodeCtrlMsg (Left from) (Monitor them ref)) = do
422422 let localFrom = processNodeId from == localNodeId node
423423 case (shouldLink, localFrom) of
424424 (True, _) -> -- [Unified: first rule]
425− modify $ nodeCtrlMonsFor them from ^: Set.insert ref
425+ modify $ ncMonsFor them from ^: Set.insert ref
426426 (False, True) -> -- [Unified: second rule]
427− ctrlSendLocal (processLocalId from) . createMessage $
427+ ncSendLocal (processLocalId from) . createMessage $
428428 ProcessDied ref them DiedNoProc
429429 (False, False) -> -- [Unified: third rule]
430− ncSendCtrlMsg (processNodeId from) $ NodeCtrlMsg
430+ ncSendCtrlMsg (processNodeId from) NCMsg
431431 { ctrlMsgSender = Right (localNodeId node)
432432 , ctrlMsgSignal = Died (Left them) DiedNoProc
433433 }
434434
435−processCtrlMsg (NodeCtrlMsg (Right _) (Monitor _ _)) =
435+ncEffect (NCMsg (Right _) (Monitor _ _)) =
436436 error "Monitor message from a node?"
437437
438438-- [Unified: Table 12, bottom rule]
439−processCtrlMsg (NodeCtrlMsg _from (Died (Right nid) reason)) = do
439+ncEffect (NCMsg _from (Died (Right nid) reason)) = do
440440 node <- ask
441− links <- gets (^. nodeCtrlLinksForNode nid)
442− mons <- gets (^. nodeCtrlMonsForNode nid)
441+ links <- gets (^. ncLinksForNode nid)
442+ mons <- gets (^. ncMonsForNode nid)
443443
444− forM_ (Map.toList links) $ \(_them, _uss) -> do
445− fail "processCtrlMsg: linking not implemented"
444+ forM_ (Map.toList links) $ \(_them, _uss) ->
445+ fail "ncEffect: linking not implemented"
446446
447− forM_ (Map.toList mons) $ \(them, uss) -> do
448− forM_ (Map.toList uss) $ \(us, refs) -> do
447+ forM_ (Map.toList mons) $ \(them, uss) ->
448+ forM_ (Map.toList uss) $ \(us, refs) ->
449449 -- We only need to notify local processes
450450 when (processNodeId us == localNodeId node) $ do
451451 let lpid = processLocalId us
452452 rpid = ProcessId nid them
453− forM_ refs $ \ref -> do
454− ctrlSendLocal lpid . createMessage $ ProcessDied ref rpid reason
453+ forM_ refs $ \ref ->
454+ ncSendLocal lpid . createMessage $ ProcessDied ref rpid reason
455455
456− modify $ (nodeCtrlLinksForNode nid ^= Map.empty)
457− . (nodeCtrlMonsForNode nid ^= Map.empty)
456+ modify $ (ncLinksForNode nid ^= Map.empty)
457+ . (ncMonsForNode nid ^= Map.empty)
458458
459459-- [Unified: Table 12, top rule]
460−processCtrlMsg (NodeCtrlMsg _from (Died (Left pid) reason)) = do
460+ncEffect (NCMsg _from (Died (Left pid) reason)) = do
461461 node <- ask
462− links <- gets (^. nodeCtrlLinksForProcess pid)
463− mons <- gets (^. nodeCtrlMonsForProcess pid)
462+ links <- gets (^. ncLinksForProcess pid)
463+ mons <- gets (^. ncMonsForProcess pid)
464464
465465 forM_ links $ \_us ->
466− fail "processCtrlMsg: linking not implemented"
466+ fail "ncEffect: linking not implemented"
467467
468468 forM_ (Map.toList mons) $ \(us, refs) ->
469469 forM_ refs $ \ref -> do
470470 let msg = createMessage $ ProcessDied ref pid reason
471471 if processNodeId us == localNodeId node
472472 then
473− ctrlSendLocal (processLocalId us) msg
473+ ncSendLocal (processLocalId us) msg
474474 else
475− ncSendCtrlMsg (processNodeId us) $ NodeCtrlMsg
475+ ncSendCtrlMsg (processNodeId us) NCMsg
476476 { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
477477 , ctrlMsgSignal = Died (Left pid) reason
478478 }
479479
480− modify $ (nodeCtrlLinksForProcess pid ^= Set.empty)
481− . (nodeCtrlMonsForProcess pid ^= Map.empty)
480+ modify $ (ncLinksForProcess pid ^= Set.empty)
481+ . (ncMonsForProcess pid ^= Map.empty)
482482
483483--------------------------------------------------------------------------------
484484-- Initialization --
@@ -504,8 +504,8 @@ newLocalNode transport = do
504504 , localState = state
505505 , localCtrlChan = ctrlChan
506506 }
507− void . forkIO $ evalStateT (runReaderT (unNodeCtrl nodeController) node)
508− initNodeCtrlState
507+ void . forkIO $ evalStateT (runReaderT (unNC nodeController) node)
508+ initNCState
509509 void . forkIO $ handleIncomingMessages node
510510 return node
511511
@@ -546,7 +546,7 @@ forkProcess node proc = do
546546 -- [Unified: Table 4, rules termination and exiting]
547547 modifyMVar_ (localState node) $
548548 return . (localProcessWithId lpid ^= Nothing)
549− writeChan (localCtrlChan node) $ NodeCtrlMsg
549+ writeChan (localCtrlChan node) NCMsg
550550 { ctrlMsgSender = Left pid
551551 , ctrlMsgSignal = Died (Left pid) reason
552552 }
@@ -570,24 +570,25 @@ handleIncomingMessages node = go [] Map.empty Set.empty
570570 let msg = payloadToMessage payload
571571 enqueue (processQueue proc) msg
572572 go uninitConns procConns ctrlConns
573− Nothing -> if cid `Set.member` ctrlConns
574− then writeChan ctrlChan (decode . BSL.fromChunks $ payload)
575− else if cid `elem` uninitConns
576− then case payloadToId payload of
577− Just lpid -> do
578− mProc <- withMVar state $ return . (^. localProcessWithId lpid)
579− case mProc of
580− Just proc ->
581− go (List.delete cid uninitConns)
582− (Map.insert cid proc procConns)
583− ctrlConns
584− Nothing ->
585− fail "handleIncomingMessages: TODO 1"
586− Nothing ->
587− go (List.delete cid uninitConns)
588− procConns
589− (Set.insert cid ctrlConns)
590− else fail "handleIncomingMessages: TODO 2"
573+ Nothing | cid `Set.member` ctrlConns ->
574+ writeChan ctrlChan (decode . BSL.fromChunks $ payload)
575+ Nothing | cid `elem` uninitConns ->
576+ case payloadToId payload of
577+ Just lpid -> do
578+ mProc <- withMVar state $ return . (^. localProcessWithId lpid)
579+ case mProc of
580+ Just proc ->
581+ go (List.delete cid uninitConns)
582+ (Map.insert cid proc procConns)
583+ ctrlConns
584+ Nothing ->
585+ fail "handleIncomingMessages: TODO 1"
586+ Nothing ->
587+ go (List.delete cid uninitConns)
588+ procConns
589+ (Set.insert cid ctrlConns)
590+ _ ->
591+ fail "handleIncomingMessages: TODO 2"
591592 NT.ConnectionClosed cid ->
592593 go (List.delete cid uninitConns)
593594 (Map.delete cid procConns)
@@ -691,15 +692,16 @@ payloadToId bss = let (bs1, bss') = BSS.splitAt 4 . BSS.concat $ bss
691692 (bs2, bs3) = BSS.splitAt 4 bss' in
692693 case NTI.decodeInt32 bs1 :: Int of
693694 0 -> Nothing
694− 1 -> Just $ LocalProcessId
695+ 1 -> Just LocalProcessId
695696 { lpidCounter = NTI.decodeInt32 bs2
696697 , lpidUnique = NTI.decodeInt32 bs3
697698 }
698699 _ -> fail "payloadToId"
699700
701+-- TODO: this was written before the change to the new semantics
700702remoteProcessFailed :: LocalNode -> ProcessId -> IO ()
701−remoteProcessFailed node them = do
702− liftIO $ modifyMVar_ (localState node) $ \st -> do
703+remoteProcessFailed node them = liftIO $
704+ modifyMVar_ (localState node) $ \st -> do
703705 sequence_ (st ^. monitorActionsFor them)
704706 return (monitorActionsFor them ^= [] $ st)
705707
@@ -753,9 +755,9 @@ instance Binary MonitorReply where
753755 put (ProcessDied ref pid reason) = put ref >> put pid >> put reason
754756 get = ProcessDied <$> get <*> get <*> get
755757
756−instance Binary NodeCtrlMsg where
758+instance Binary NCMsg where
757759 put msg = put (ctrlMsgSender msg) >> put (ctrlMsgSignal msg)
758− get = NodeCtrlMsg <$> get <*> get
760+ get = NCMsg <$> get <*> get
759761
760762instance Binary ProcessSignal where
761763 put (Monitor pid ref) = putWord8 0 >> put pid >> put ref
@@ -814,39 +816,29 @@ connections = accessor _connections (\conns st -> st { _connections = conns })
814816connectionTo :: ProcessId -> Accessor LocalProcessState (Maybe NT.Connection)
815817connectionTo pid = connections >>> DAC.mapMaybe pid
816818
817−nodeCtrlLinks :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Set ProcessId)))
818−nodeCtrlLinks = accessor _nodeCtrlLinks (\links st -> st { _nodeCtrlLinks = links })
819+ncLinks :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
820+ncLinks = accessor _ncLinks (\links st -> st { _ncLinks = links })
819821
820−nodeCtrlMons :: Accessor NodeCtrlState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
821−nodeCtrlMons = accessor _nodeCtrlMons (\mons st -> st { _nodeCtrlMons = mons })
822+ncMons :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
823+ncMons = accessor _ncMons (\mons st -> st { _ncMons = mons })
822824
823−nodeCtrlConns :: Accessor NodeCtrlState (Map Identifier NT.Connection)
824−nodeCtrlConns = accessor _nodeCtrlConns (\conns st -> st { _nodeCtrlConns = conns })
825+ncConns :: Accessor NCState (Map Identifier NT.Connection)
826+ncConns = accessor _ncConns (\conns st -> st { _ncConns = conns })
825827
826−nodeCtrlConnFor :: Identifier -> Accessor NodeCtrlState (Maybe NT.Connection)
827−nodeCtrlConnFor them = nodeCtrlConns >>> DAC.mapMaybe them
828+ncConnFor :: Identifier -> Accessor NCState (Maybe NT.Connection)
829+ncConnFor them = ncConns >>> DAC.mapMaybe them
828830
829−nodeCtrlLinksForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Set ProcessId))
830−nodeCtrlLinksForNode nid = nodeCtrlLinks >>> DAC.mapDefault Map.empty nid
831+ncLinksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
832+ncLinksForNode nid = ncLinks >>> DAC.mapDefault Map.empty nid
831833
832−nodeCtrlMonsForNode :: NodeId -> Accessor NodeCtrlState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
833−nodeCtrlMonsForNode nid = nodeCtrlMons >>> DAC.mapDefault Map.empty nid
834+ncMonsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
835+ncMonsForNode nid = ncMons >>> DAC.mapDefault Map.empty nid
834836
835−nodeCtrlLinksForProcess :: ProcessId -> Accessor NodeCtrlState (Set ProcessId)
836−nodeCtrlLinksForProcess pid = nodeCtrlLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
837+ncLinksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
838+ncLinksForProcess pid = ncLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
837839
838−nodeCtrlMonsForProcess :: ProcessId -> Accessor NodeCtrlState (Map ProcessId (Set MonitorRef))
839−nodeCtrlMonsForProcess pid = nodeCtrlMonsForNode (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"
840+ncMonsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
841+ncMonsForProcess pid = ncMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
852842
843+ncMonsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
844+ncMonsFor them us = ncMonsForProcess them >>> DAC.mapDefault Set.empty us