@@ -54,109 +54,40 @@ runNodeController :: LocalNode -> IO ()
5454runNodeController node =
5555 evalStateT (runReaderT (unNC nodeController) node) initNCState
5656
57+--------------------------------------------------------------------------------
58+-- Internal data types --
59+--------------------------------------------------------------------------------
5760
5861data NCState = NCState
5962 { -- Mapping from remote processes to linked local processes
60− _ncLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
63+ _links :: Map NodeId (Map LocalProcessId (Set ProcessId))
6164 -- Mapping from remote processes to monitoring local processes
62− , _ncMons :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
65+ , _monitors :: Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
6366 -- The node controller maintains its own set of connections
6467 -- TODO: still not convinced that this is correct
65− , _ncConns :: Map Identifier NT.Connection
66− }
67−
68−destNid :: ProcessSignal -> Maybe NodeId
69−destNid (Monitor pid _) = Just $ processNodeId pid
70−destNid (Died (Right _) _) = Nothing
71−destNid (Died (Left _pid) _) = fail "destNid: TODO"
72−
73−initNCState :: NCState
74−initNCState = NCState
75− { _ncLinks = Map.empty
76− , _ncMons = Map.empty
77− , _ncConns = Map.empty
68+ , _connections :: Map Identifier NT.Connection
7869 }
7970
8071newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }
8172 deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
8273
83−nodeController :: NC ()
84−nodeController = forever $ do
85− node <- ask
86− msg <- liftIO $ readChan (localCtrlChan node)
87−
88− -- Forward the message if appropriate
89− case destNid (ctrlMsgSignal msg) of
90− Just nid' | nid' /= localNodeId node -> ncSendCtrlMsg nid' msg
91− _ -> return ()
92−
93− ncEffect msg
94−
95−ctrlSendTo :: Identifier -> [BSS.ByteString] -> NC ()
96−ctrlSendTo them payload = do
97− mConn <- ncConnTo them
98− didSend <- case mConn of
99− Just conn -> do
100− didSend <- liftIO $ NT.send conn payload
101− case didSend of
102− Left _ -> return False
103− Right () -> return True
104− Nothing ->
105− return False
106− unless didSend $ do
107− -- [Unified: Table 9, rule node_disconnect]
108− node <- ask
109− liftIO . writeChan (localCtrlChan node) $ NCMsg
110− { ctrlMsgSender = them
111− , ctrlMsgSignal = Died them DiedDisconnect
112− }
113−
114−ncSendCtrlMsg :: NodeId -> NCMsg -> NC ()
115−ncSendCtrlMsg dest = ctrlSendTo (Right dest) . BSL.toChunks . encode
74+--------------------------------------------------------------------------------
75+-- Core functionality --
76+--------------------------------------------------------------------------------
11677
117−ncSendLocal :: LocalProcessId -> Message -> NC ()
118−ncSendLocal lpid msg = do
78+-- [Unified: Table 7]
79+nodeController :: NC ()
80+nodeController = do
11981 node <- ask
120− liftIO $ do
121− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
122− -- By [Unified: table 6, rule missing_process] messages to dead processes
123− -- can silently be dropped
124− forM_ mProc $ \proc -> enqueue (processQueue proc) msg
82+ forever $ do
83+ msg <- liftIO $ readChan (localCtrlChan node)
12584
126−ncConnTo :: Identifier -> NC (Maybe NT.Connection)
127−ncConnTo them = do
128− mConn <- gets (^. ncConnFor them)
129− case mConn of
130− Just conn -> return (Just conn)
131− Nothing -> ncCreateConnTo them
132−
133−ncCreateConnTo :: Identifier -> NC (Maybe NT.Connection)
134−ncCreateConnTo them = do
135− node <- ask
136− mConn <- liftIO $ NT.connect (localEndPoint node)
137− addr
138− NT.ReliableOrdered
139− NT.defaultConnectHints
140− case mConn of
141− Right conn -> do
142− didSend <- liftIO $ NT.send conn firstMsg
143− case didSend of
144− Left _ ->
145− return Nothing
146− Right () -> do
147− modify $ ncConnFor them ^= Just conn
148− return $ Just conn
149− Left _ ->
150− return Nothing
151− where
152− (addr, firstMsg) = case them of
153− Left pid -> ( nodeAddress (processNodeId pid)
154− , idToPayload (Just $ processLocalId pid)
155− )
156− Right nid -> ( nodeAddress nid
157− , idToPayload Nothing
158− )
85+ -- [Unified: Table 7, rule nc_forward]
86+ case destNid (ctrlMsgSignal msg) of
87+ Just nid' | nid' /= localNodeId node -> sendCtrlMsg nid' msg
88+ _ -> return ()
15989
90+ ncEffect msg
16091
16192-- [Unified: ncEffect]
16293ncEffect :: NCMsg -> NC ()
@@ -172,12 +103,12 @@ ncEffect (NCMsg (Left from) (Monitor them ref)) = do
172103 let localFrom = processNodeId from == localNodeId node
173104 case (shouldLink, localFrom) of
174105 (True, _) -> -- [Unified: first rule]
175− modify $ ncMonsFor them from ^: Set.insert ref
106+ modify $ monitorsFor them from ^: Set.insert ref
176107 (False, True) -> -- [Unified: second rule]
177− ncSendLocal (processLocalId from) . createMessage $
108+ sendLocal (processLocalId from) . createMessage $
178109 ProcessDied ref them DiedNoProc
179110 (False, False) -> -- [Unified: third rule]
180− ncSendCtrlMsg (processNodeId from) NCMsg
111+ sendCtrlMsg (processNodeId from) NCMsg
181112 { ctrlMsgSender = Right (localNodeId node)
182113 , ctrlMsgSignal = Died (Left them) DiedNoProc
183114 }
@@ -187,11 +118,11 @@ ncEffect (NCMsg (Right _) (Monitor _ _)) =
187118
188119-- [Unified: Table 12, bottom rule]
189120ncEffect (NCMsg _from (Died (Right nid) reason)) = do
190− node <- ask
191− links <- gets (^. ncLinksForNode nid)
192− mons <- gets (^. ncMonsForNode nid)
121+ node <- ask
122+ lnks <- gets (^. linksForNode nid)
123+ mons <- gets (^. monitorsForNode nid)
193124
194− forM_ (Map.toList links) $ \(_them, _uss) ->
125+ forM_ (Map.toList lnks) $ \(_them, _uss) ->
195126 fail "ncEffect: linking not implemented"
196127
197128 forM_ (Map.toList mons) $ \(them, uss) ->
@@ -201,18 +132,18 @@ ncEffect (NCMsg _from (Died (Right nid) reason)) = do
201132 let lpid = processLocalId us
202133 rpid = ProcessId nid them
203134 forM_ refs $ \ref ->
204− ncSendLocal lpid . createMessage $ ProcessDied ref rpid reason
135+ sendLocal lpid . createMessage $ ProcessDied ref rpid reason
205136
206− modify $ (ncLinksForNode nid ^= Map.empty)
207− . (ncMonsForNode nid ^= Map.empty)
137+ modify $ (linksForNode nid ^= Map.empty)
138+ . (monitorsForNode nid ^= Map.empty)
208139
209140-- [Unified: Table 12, top rule]
210141ncEffect (NCMsg _from (Died (Left pid) reason)) = do
211− node <- ask
212− links <- gets (^. ncLinksForProcess pid)
213− mons <- gets (^. ncMonsForProcess pid)
142+ node <- ask
143+ lnks <- gets (^. linksForProcess pid)
144+ mons <- gets (^. monitorsForProcess pid)
214145
215− forM_ links $ \_us ->
146+ forM_ lnks $ \_us ->
216147 fail "ncEffect: linking not implemented"
217148
218149 forM_ (Map.toList mons) $ \(us, refs) ->
@@ -220,43 +151,128 @@ ncEffect (NCMsg _from (Died (Left pid) reason)) = do
220151 let msg = createMessage $ ProcessDied ref pid reason
221152 if processNodeId us == localNodeId node
222153 then
223− ncSendLocal (processLocalId us) msg
154+ sendLocal (processLocalId us) msg
224155 else
225− ncSendCtrlMsg (processNodeId us) NCMsg
156+ sendCtrlMsg (processNodeId us) NCMsg
226157 { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
227158 , ctrlMsgSignal = Died (Left pid) reason
228159 }
229160
230− modify $ (ncLinksForProcess pid ^= Set.empty)
231− . (ncMonsForProcess pid ^= Map.empty)
161+ modify $ (linksForProcess pid ^= Set.empty)
162+ . (monitorsForProcess pid ^= Map.empty)
163+
164+--------------------------------------------------------------------------------
165+-- Connections --
166+--------------------------------------------------------------------------------
167+
168+sendTo :: Identifier -> [BSS.ByteString] -> NC ()
169+sendTo them payload = do
170+ mConn <- connTo them
171+ didSend <- case mConn of
172+ Just conn -> do
173+ didSend <- liftIO $ NT.send conn payload
174+ case didSend of
175+ Left _ -> return False
176+ Right () -> return True
177+ Nothing ->
178+ return False
179+ unless didSend $ do
180+ -- [Unified: Table 9, rule node_disconnect]
181+ node <- ask
182+ liftIO . writeChan (localCtrlChan node) $ NCMsg
183+ { ctrlMsgSender = them
184+ , ctrlMsgSignal = Died them DiedDisconnect
185+ }
186+
187+sendCtrlMsg :: NodeId -> NCMsg -> NC ()
188+sendCtrlMsg dest = sendTo (Right dest) . BSL.toChunks . encode
189+
190+sendLocal :: LocalProcessId -> Message -> NC ()
191+sendLocal lpid msg = do
192+ node <- ask
193+ liftIO $ do
194+ mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
195+ -- By [Unified: table 6, rule missing_process] messages to dead processes
196+ -- can silently be dropped
197+ forM_ mProc $ \proc -> enqueue (processQueue proc) msg
198+
199+connTo :: Identifier -> NC (Maybe NT.Connection)
200+connTo them = do
201+ mConn <- gets (^. connectionTo them)
202+ case mConn of
203+ Just conn -> return (Just conn)
204+ Nothing -> createConnTo them
205+
206+createConnTo :: Identifier -> NC (Maybe NT.Connection)
207+createConnTo them = do
208+ node <- ask
209+ mConn <- liftIO $ NT.connect (localEndPoint node)
210+ addr
211+ NT.ReliableOrdered
212+ NT.defaultConnectHints
213+ case mConn of
214+ Right conn -> do
215+ didSend <- liftIO $ NT.send conn firstMsg
216+ case didSend of
217+ Left _ ->
218+ return Nothing
219+ Right () -> do
220+ modify $ connectionTo them ^= Just conn
221+ return $ Just conn
222+ Left _ ->
223+ return Nothing
224+ where
225+ (addr, firstMsg) = case them of
226+ Left pid -> ( nodeAddress (processNodeId pid)
227+ , idToPayload (Just $ processLocalId pid)
228+ )
229+ Right nid -> ( nodeAddress nid
230+ , idToPayload Nothing
231+ )
232+
233+--------------------------------------------------------------------------------
234+-- Auxiliary --
235+--------------------------------------------------------------------------------
236+
237+destNid :: ProcessSignal -> Maybe NodeId
238+destNid (Monitor pid _) = Just $ processNodeId pid
239+destNid (Died (Right _) _) = Nothing
240+destNid (Died (Left _pid) _) = fail "destNid: TODO"
241+
242+initNCState :: NCState
243+initNCState = NCState
244+ { _links = Map.empty
245+ , _monitors = Map.empty
246+ , _connections = Map.empty
247+ }
232248
233249--------------------------------------------------------------------------------
234250-- Accessors --
235251--------------------------------------------------------------------------------
236252
237−ncLinks :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
238−ncLinks = accessor _ncLinks (\links st -> st { _ncLinks = links })
253+links :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
254+links = accessor _links (\lnks st -> st { _links = lnks })
239255
240−ncMons :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
241−ncMons = accessor _ncMons (\mons st -> st { _ncMons = mons })
256+monitors :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
257+monitors = accessor _monitors (\mons st -> st { _monitors = mons })
242258
243−ncConns :: Accessor NCState (Map Identifier NT.Connection)
244−ncConns = accessor _ncConns (\conns st -> st { _ncConns = conns })
259+connections :: Accessor NCState (Map Identifier NT.Connection)
260+connections = accessor _connections (\conns st -> st { _connections = conns })
245261
246−ncConnFor :: Identifier -> Accessor NCState (Maybe NT.Connection)
247−ncConnFor them = ncConns >>> DAC.mapMaybe them
262+connectionTo :: Identifier -> Accessor NCState (Maybe NT.Connection)
263+connectionTo them = connections >>> DAC.mapMaybe them
248264
249−ncLinksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
250−ncLinksForNode nid = ncLinks >>> DAC.mapDefault Map.empty nid
265+linksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
266+linksForNode nid = links >>> DAC.mapDefault Map.empty nid
251267
252−ncMonsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
253−ncMonsForNode nid = ncMons >>> DAC.mapDefault Map.empty nid
268+monitorsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
269+monitorsForNode nid = monitors >>> DAC.mapDefault Map.empty nid
254270
255−ncLinksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
256−ncLinksForProcess pid = ncLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
271+linksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
272+linksForProcess pid = linksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
257273
258−ncMonsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
259−ncMonsForProcess pid = ncMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
274+monitorsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
275+monitorsForProcess pid = monitorsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
260276
261−ncMonsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
262−ncMonsFor them us = ncMonsForProcess them >>> DAC.mapDefault Set.empty us
277+monitorsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
278+monitorsFor them us = monitorsForProcess them >>> DAC.mapDefault Set.empty us