git.iko.soy/distributed-process
on master help sign in
Cleanup
24262d0Edsko de Vries
distributed-process/src/Control/Distributed/Process/Internal/NodeController.hs+140−124
compare with master
unifiedsplit
@@ -54,109 +54,40 @@ runNodeController :: LocalNode -> IO ()
54runNodeController node =
54runNodeController node =
55 evalStateT (runReaderT (unNC nodeController) node) initNCState
55 evalStateT (runReaderT (unNC nodeController) node) initNCState
56
56
57+--------------------------------------------------------------------------------
58+-- Internal data types --
59+--------------------------------------------------------------------------------
57
60
58data NCState = NCState
61data NCState = NCState
59 { -- Mapping from remote processes to linked local processes
62 { -- Mapping from remote processes to linked local processes
60− _ncLinks :: Map NodeId (Map LocalProcessId (Set ProcessId))
63+ _links :: Map NodeId (Map LocalProcessId (Set ProcessId))
61 -- Mapping from remote processes to monitoring local processes
64 -- 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)))
63 -- The node controller maintains its own set of connections
66 -- The node controller maintains its own set of connections
64 -- TODO: still not convinced that this is correct
67 -- TODO: still not convinced that this is correct
65− , _ncConns :: Map Identifier NT.Connection
68+ , _connections :: 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
78 }
69 }
79
70
80newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }
71newtype NC a = NC { unNC :: ReaderT LocalNode (StateT NCState IO) a }
81 deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
72 deriving (Functor, Monad, MonadIO, MonadState NCState, MonadReader LocalNode)
82
73
83−nodeController :: NC ()
74+--------------------------------------------------------------------------------
84−nodeController = forever $ do
75+-- Core functionality --
85− node <- ask
76+--------------------------------------------------------------------------------
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
116
77
117−ncSendLocal :: LocalProcessId -> Message -> NC ()
78+-- [Unified: Table 7]
118−ncSendLocal lpid msg = do
79+nodeController :: NC ()
80+nodeController = do
119 node <- ask
81 node <- ask
120− liftIO $ do
82+ forever $ do
121− mProc <- withMVar (localState node) $ return . (^. localProcessWithId lpid)
83+ msg <- liftIO $ readChan (localCtrlChan node)
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
125
84
126−ncConnTo :: Identifier -> NC (Maybe NT.Connection)
85+ -- [Unified: Table 7, rule nc_forward]
127−ncConnTo them = do
86+ case destNid (ctrlMsgSignal msg) of
128− mConn <- gets (^. ncConnFor them)
87+ Just nid' | nid' /= localNodeId node -> sendCtrlMsg nid' msg
129− case mConn of
88+ _ -> return ()
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− )
159
89
90+ ncEffect msg
160
91
161-- [Unified: ncEffect]
92-- [Unified: ncEffect]
162ncEffect :: NCMsg -> NC ()
93ncEffect :: NCMsg -> NC ()
@@ -172,12 +103,12 @@ ncEffect (NCMsg (Left from) (Monitor them ref)) = do
172 let localFrom = processNodeId from == localNodeId node
103 let localFrom = processNodeId from == localNodeId node
173 case (shouldLink, localFrom) of
104 case (shouldLink, localFrom) of
174 (True, _) -> -- [Unified: first rule]
105 (True, _) -> -- [Unified: first rule]
175− modify $ ncMonsFor them from ^: Set.insert ref
106+ modify $ monitorsFor them from ^: Set.insert ref
176 (False, True) -> -- [Unified: second rule]
107 (False, True) -> -- [Unified: second rule]
177− ncSendLocal (processLocalId from) . createMessage $
108+ sendLocal (processLocalId from) . createMessage $
178 ProcessDied ref them DiedNoProc
109 ProcessDied ref them DiedNoProc
179 (False, False) -> -- [Unified: third rule]
110 (False, False) -> -- [Unified: third rule]
180− ncSendCtrlMsg (processNodeId from) NCMsg
111+ sendCtrlMsg (processNodeId from) NCMsg
181 { ctrlMsgSender = Right (localNodeId node)
112 { ctrlMsgSender = Right (localNodeId node)
182 , ctrlMsgSignal = Died (Left them) DiedNoProc
113 , ctrlMsgSignal = Died (Left them) DiedNoProc
183 }
114 }
@@ -187,11 +118,11 @@ ncEffect (NCMsg (Right _) (Monitor _ _)) =
187
118
188-- [Unified: Table 12, bottom rule]
119-- [Unified: Table 12, bottom rule]
189ncEffect (NCMsg _from (Died (Right nid) reason)) = do
120ncEffect (NCMsg _from (Died (Right nid) reason)) = do
190− node <- ask
121+ node <- ask
191− links <- gets (^. ncLinksForNode nid)
122+ lnks <- gets (^. linksForNode nid)
192− mons <- gets (^. ncMonsForNode nid)
123+ mons <- gets (^. monitorsForNode nid)
193
124
194− forM_ (Map.toList links) $ \(_them, _uss) ->
125+ forM_ (Map.toList lnks) $ \(_them, _uss) ->
195 fail "ncEffect: linking not implemented"
126 fail "ncEffect: linking not implemented"
196
127
197 forM_ (Map.toList mons) $ \(them, uss) ->
128 forM_ (Map.toList mons) $ \(them, uss) ->
@@ -201,18 +132,18 @@ ncEffect (NCMsg _from (Died (Right nid) reason)) = do
201 let lpid = processLocalId us
132 let lpid = processLocalId us
202 rpid = ProcessId nid them
133 rpid = ProcessId nid them
203 forM_ refs $ \ref ->
134 forM_ refs $ \ref ->
204− ncSendLocal lpid . createMessage $ ProcessDied ref rpid reason
135+ sendLocal lpid . createMessage $ ProcessDied ref rpid reason
205
136
206− modify $ (ncLinksForNode nid ^= Map.empty)
137+ modify $ (linksForNode nid ^= Map.empty)
207− . (ncMonsForNode nid ^= Map.empty)
138+ . (monitorsForNode nid ^= Map.empty)
208
139
209-- [Unified: Table 12, top rule]
140-- [Unified: Table 12, top rule]
210ncEffect (NCMsg _from (Died (Left pid) reason)) = do
141ncEffect (NCMsg _from (Died (Left pid) reason)) = do
211− node <- ask
142+ node <- ask
212− links <- gets (^. ncLinksForProcess pid)
143+ lnks <- gets (^. linksForProcess pid)
213− mons <- gets (^. ncMonsForProcess pid)
144+ mons <- gets (^. monitorsForProcess pid)
214
145
215− forM_ links $ \_us ->
146+ forM_ lnks $ \_us ->
216 fail "ncEffect: linking not implemented"
147 fail "ncEffect: linking not implemented"
217
148
218 forM_ (Map.toList mons) $ \(us, refs) ->
149 forM_ (Map.toList mons) $ \(us, refs) ->
@@ -220,43 +151,128 @@ ncEffect (NCMsg _from (Died (Left pid) reason)) = do
220 let msg = createMessage $ ProcessDied ref pid reason
151 let msg = createMessage $ ProcessDied ref pid reason
221 if processNodeId us == localNodeId node
152 if processNodeId us == localNodeId node
222 then
153 then
223− ncSendLocal (processLocalId us) msg
154+ sendLocal (processLocalId us) msg
224 else
155 else
225− ncSendCtrlMsg (processNodeId us) NCMsg
156+ sendCtrlMsg (processNodeId us) NCMsg
226 { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
157 { ctrlMsgSender = Right (localNodeId node) -- TODO: why the change in sender? How does that affect 'reconnect' semantics?
227 , ctrlMsgSignal = Died (Left pid) reason
158 , ctrlMsgSignal = Died (Left pid) reason
228 }
159 }
229
160
230− modify $ (ncLinksForProcess pid ^= Set.empty)
161+ modify $ (linksForProcess pid ^= Set.empty)
231− . (ncMonsForProcess pid ^= Map.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+ }
232
248
233--------------------------------------------------------------------------------
249--------------------------------------------------------------------------------
234-- Accessors --
250-- Accessors --
235--------------------------------------------------------------------------------
251--------------------------------------------------------------------------------
236
252
237−ncLinks :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
253+links :: Accessor NCState (Map NodeId (Map LocalProcessId (Set ProcessId)))
238−ncLinks = accessor _ncLinks (\links st -> st { _ncLinks = links })
254+links = accessor _links (\lnks st -> st { _links = lnks })
239
255
240−ncMons :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
256+monitors :: Accessor NCState (Map NodeId (Map LocalProcessId (Map ProcessId (Set MonitorRef))))
241−ncMons = accessor _ncMons (\mons st -> st { _ncMons = mons })
257+monitors = accessor _monitors (\mons st -> st { _monitors = mons })
242
258
243−ncConns :: Accessor NCState (Map Identifier NT.Connection)
259+connections :: Accessor NCState (Map Identifier NT.Connection)
244−ncConns = accessor _ncConns (\conns st -> st { _ncConns = conns })
260+connections = accessor _connections (\conns st -> st { _connections = conns })
245
261
246−ncConnFor :: Identifier -> Accessor NCState (Maybe NT.Connection)
262+connectionTo :: Identifier -> Accessor NCState (Maybe NT.Connection)
247−ncConnFor them = ncConns >>> DAC.mapMaybe them
263+connectionTo them = connections >>> DAC.mapMaybe them
248
264
249−ncLinksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
265+linksForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Set ProcessId))
250−ncLinksForNode nid = ncLinks >>> DAC.mapDefault Map.empty nid
266+linksForNode nid = links >>> DAC.mapDefault Map.empty nid
251
267
252−ncMonsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
268+monitorsForNode :: NodeId -> Accessor NCState (Map LocalProcessId (Map ProcessId (Set MonitorRef)))
253−ncMonsForNode nid = ncMons >>> DAC.mapDefault Map.empty nid
269+monitorsForNode nid = monitors >>> DAC.mapDefault Map.empty nid
254
270
255−ncLinksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
271+linksForProcess :: ProcessId -> Accessor NCState (Set ProcessId)
256−ncLinksForProcess pid = ncLinksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
272+linksForProcess pid = linksForNode (processNodeId pid) >>> DAC.mapDefault Set.empty (processLocalId pid)
257
273
258−ncMonsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
274+monitorsForProcess :: ProcessId -> Accessor NCState (Map ProcessId (Set MonitorRef))
259−ncMonsForProcess pid = ncMonsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
275+monitorsForProcess pid = monitorsForNode (processNodeId pid) >>> DAC.mapDefault Map.empty (processLocalId pid)
260
276
261−ncMonsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
277+monitorsFor :: ProcessId -> ProcessId -> Accessor NCState (Set MonitorRef)
262−ncMonsFor them us = ncMonsForProcess them >>> DAC.mapDefault Set.empty us
278+monitorsFor them us = monitorsForProcess them >>> DAC.mapDefault Set.empty us
distributed-process/tests/TestCH.hs+4−4
@@ -10,11 +10,12 @@ import Control.Concurrent.MVar ( newEmptyMVar
10 , readMVar
10 , readMVar
11 )
11 )
12import Control.Monad (replicateM_)
12import Control.Monad (replicateM_)
13−import Control.Distributed.Process
14import Control.Monad.IO.Class (liftIO)
13import Control.Monad.IO.Class (liftIO)
15import Control.Exception (throwIO)
14import Control.Exception (throwIO)
16−import qualified Network.Transport as NT (Transport)
15+import qualified Network.Transport as NT (Transport, closeEndPoint)
17import Network.Transport.TCP (createTransport, defaultTCPParameters)
16import Network.Transport.TCP (createTransport, defaultTCPParameters)
17+import Control.Distributed.Process
18+import Control.Distributed.Process.Internal (LocalNode(localEndPoint))
18import TestAuxiliary
19import TestAuxiliary
19
20
20newtype Ping = Ping ProcessId
21newtype Ping = Ping ProcessId
@@ -212,8 +213,7 @@ testMonitorDisconnect transport = do
212 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
213 addr <- forkProcess localNode . liftIO $ threadDelay 1000000
213 putMVar processAddr addr
214 putMVar processAddr addr
214 readMVar monitorSetup
215 readMVar monitorSetup
215− -- TODO: closeLocalNode should eventually kill processes too, so it's not a good test of a network disconnect
216+ NT.closeEndPoint (localEndPoint localNode)
216− closeLocalNode localNode
217
217
218 forkIO $ do
218 forkIO $ do
219 localNode <- newLocalNode transport
219 localNode <- newLocalNode transport