git.iko.soy/distributed-process
on master help sign in
Don't assume nodes are still alive in findSlaves
fd61d21Edsko de Vries
distributed-process-simplelocalnet/src/Control/Distributed/Process/Backend/SimpleLocalnet.hs+20−4
compare with master
unifiedsplit
@@ -97,7 +97,7 @@ import Data.Foldable (forM_)
97import Data.Typeable (Typeable)
97import Data.Typeable (Typeable)
98import Control.Applicative ((<$>))
98import Control.Applicative ((<$>))
99import Control.Exception (throw)
99import Control.Exception (throw)
100−import Control.Monad (forever, forM)
100+import Control.Monad (forever, forM, replicateM)
101import Control.Monad.IO.Class (liftIO)
101import Control.Monad.IO.Class (liftIO)
102import Control.Concurrent (forkIO, threadDelay, ThreadId)
102import Control.Concurrent (forkIO, threadDelay, ThreadId)
103import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
103import Control.Concurrent.MVar (MVar, newMVar, readMVar, modifyMVar_)
@@ -114,8 +114,12 @@ import Control.Distributed.Process
114 , expect
114 , expect
115 , nsendRemote
115 , nsendRemote
116 , receiveWait
116 , receiveWait
117+ , match
117 , matchIf
118 , matchIf
118 , processNodeId
119 , processNodeId
120+ , monitorNode
121+ , unmonitor
122+ , NodeMonitorNotification(..)
119 )
123 )
120import qualified Control.Distributed.Process.Node as Node
124import qualified Control.Distributed.Process.Node as Node
121 ( LocalNode
125 ( LocalNode
@@ -286,12 +290,24 @@ findSlaves :: Backend -> Process [NodeId]
286findSlaves backend = do
290findSlaves backend = do
287 nodes <- liftIO $ findPeers backend 1000000
291 nodes <- liftIO $ findPeers backend 1000000
288 -- Fire of asynchronous requests for the slave controller
292 -- Fire of asynchronous requests for the slave controller
293+ refs <- forM nodes $ \nid -> do
289− forM_ nodes $ \nid -> whereisRemoteAsync nid "slaveController"
294+ whereisRemoteAsync nid "slaveController"
295+ ref <- monitorNode nid
296+ return (nid, ref)
290 -- Wait for the replies
297 -- Wait for the replies
291− catMaybes <$> forM nodes (\_ ->
298+ catMaybes <$> replicateM (length nodes) (
292 receiveWait
299 receiveWait
293 [ matchIf (\(WhereIsReply label _) -> label == "slaveController")
300 [ matchIf (\(WhereIsReply label _) -> label == "slaveController")
294− (\(WhereIsReply _ mPid) -> return (processNodeId <$> mPid))
301+ (\(WhereIsReply _ mPid) ->
302+ case mPid of
303+ Nothing ->
304+ return Nothing
305+ Just pid -> do
306+ let nid = processNodeId pid
307+ Just ref = lookup nid refs
308+ unmonitor ref
309+ return (Just nid))
310+ , match (\(NodeMonitorNotification {}) -> return Nothing)
295 ])
311 ])
296
312
297-- | Terminate all slaves
313-- | Terminate all slaves