7bcba47 · distributed-process

git.iko.soy/distributed-process

on master help sign in

Monitor disconnect

7bcba47Edsko de Vries14 years ago

distributed-process/src/Control/Distributed/Process.hs+8−16

compare with master

unifiedsplit

@@ -86,7 +86,7 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString

86 )

87import Data.Binary (Binary, decode, encode, put, get, putWord8, getWord8)

88import Data.Map (Map)

89−import qualified Data.Map as Map (empty, lookup, insert, delete, keys, toList)

89+import qualified Data.Map as Map (empty, lookup, insert, delete, toList)

90import qualified Data.List as List (delete)

91import Data.Set (Set)

92import qualified Data.Set as Set (empty, insert, delete, member)

@@ -186,7 +186,6 @@ data LocalNodeState = LocalNodeState

186 { _localProcesses :: Map LocalProcessId LocalProcess

187 , _localPidCounter :: Int32

188 , _localPidUnique :: Int32

189− , _monitorActions :: Map ProcessId [IO ()]

190 , _localMonitorCounter :: Int32

189 , _localMonitorCounter :: Int32

191 }

190 }

192

191

@@ -232,6 +231,7 @@ send them msg = do

232getSelfPid :: Process ProcessId

231getSelfPid :: Process ProcessId

233getSelfPid = processId <$> ask

232getSelfPid = processId <$> ask

234

233

234+-- | Monitor another process

235monitor :: ProcessId -> Process MonitorRef

236monitor them = do

237 us <- getSelfPid

@@ -269,7 +269,6 @@ expectTimeout timeout = do

269 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)

270 return $ fmap (decode . messageEncoding) msg

271

272−

273--------------------------------------------------------------------------------

272--------------------------------------------------------------------------------

274-- The node controller --

273-- The node controller --

275--------------------------------------------------------------------------------

274--------------------------------------------------------------------------------

@@ -488,7 +487,6 @@ newLocalNode transport = do

488 { _localProcesses = Map.empty

487 { _localProcesses = Map.empty

489 , _localPidCounter = 0

488 , _localPidCounter = 0

490 , _localPidUnique = unq

489 , _localPidUnique = unq

491− , _monitorActions = Map.empty

492 , _localMonitorCounter = 0

490 , _localMonitorCounter = 0

493 }

491 }

494 ctrlChan <- newChan

492 ctrlChan <- newChan

@@ -587,12 +585,12 @@ handleIncomingMessages node = go [] Map.empty Set.empty

587 (Map.delete cid procConns)

585 (Map.delete cid procConns)

588 (Set.delete cid ctrlConns)

586 (Set.delete cid ctrlConns)

589 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do

587 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do

590− -- TODO: we could cache this information

588+ -- [Unified table 9, rule node_disconnect]

591− pids <- withMVar state $ \st ->

589+ let nid = Right $ NodeId theirAddr

592− return $ filter ((== theirAddr) . nodeAddress . processNodeId)

590+ writeChan ctrlChan NCMsg

593− . Map.keys

591+ { ctrlMsgSender = nid

594− $ st ^. monitorActions

592+ , ctrlMsgSignal = Died nid DiedDisconnect

595− forM_ pids $ remoteProcessFailed node

593+ }

596 NT.ErrorEvent _ ->

594 NT.ErrorEvent _ ->

597 fail "handleIncomingMessages: TODO 4"

595 fail "handleIncomingMessages: TODO 4"

598 NT.EndPointClosed ->

596 NT.EndPointClosed ->

@@ -774,18 +772,12 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =

774localPidUnique :: Accessor LocalNodeState Int32

772localPidUnique :: Accessor LocalNodeState Int32

775localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

773localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })

776

774

777−monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])

778−monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })

779−

780localMonitorCounter :: Accessor LocalNodeState Int32

775localMonitorCounter :: Accessor LocalNodeState Int32

781localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })

776localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })

782

777

783localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

778localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)

784localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

779localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid

785

780

786−monitorActionsFor :: ProcessId -> Accessor LocalNodeState [IO ()]

787−monitorActionsFor pid = monitorActions >>> DAC.mapDefault [] pid

788−

789connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)

781connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)

790connections = accessor _connections (\conns st -> st { _connections = conns })

782connections = accessor _connections (\conns st -> st { _connections = conns })

791

783

distributed-process/tests/TestCH.hs+31−1

@@ -70,7 +70,7 @@ testMonitorUnreachable transport = do

71 forkIO $ do

72 localNode <- newLocalNode transport

73− addr <- forkProcess localNode $ liftIO $ newEmptyMVar >>= readMVar -- deadlock

73+ addr <- forkProcess localNode . liftIO $ threadDelay 1000000

74 closeLocalNode localNode

75 putMVar deadProcess addr

@@ -200,6 +200,35 @@ testMonitorRemoteDeadProcess transport = do

200

201 takeMVar done

202

203+-- | Monitor a process that becomes disconnected

204+testMonitorDisconnect :: NT.Transport -> IO ()

205+testMonitorDisconnect transport = do

206+ processAddr <- newEmptyMVar

207+ monitorSetup <- newEmptyMVar

208+ done <- newEmptyMVar

209+

210+ forkIO $ do

211+ localNode <- newLocalNode transport

212+ addr <- forkProcess localNode . liftIO $ threadDelay 1000000

213+ putMVar processAddr addr

214+ readMVar monitorSetup

215+ -- TODO: closeLocalNode should eventually kill processes too, so it's not a good test of a network disconnect

216+ closeLocalNode localNode

217+

218+ forkIO $ do

219+ localNode <- newLocalNode transport

220+ theirAddr <- readMVar processAddr

221+ runProcess localNode $ do

222+ ref <- monitor theirAddr

223+ liftIO $ threadDelay 100000 >> putMVar monitorSetup ()

224+ ProcessDied ref' pid DiedDisconnect <- expect

225+ True <- return $ ref' == ref && pid == theirAddr

226+ return ()

227+ putMVar done ()

228+

229+ takeMVar done

230+

231+

203{-

232{-

204-- Like 'testMonitor1', but throw an exception instead

233-- Like 'testMonitor1', but throw an exception instead

205testMonitor2 :: NT.Transport -> IO ()

234testMonitor2 :: NT.Transport -> IO ()

@@ -312,6 +341,7 @@ main = do

312 , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)

341 , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)

313 , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport)

342 , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport)

314 , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport)

343 , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport)

344+ , ("MonitorDisconnect", testMonitorDisconnect transport)

315 {-

345 {-

316 , ("Monitor2", testMonitor2 transport)

346 , ("Monitor2", testMonitor2 transport)

317 , ("Monitor3", testMonitor3 transport)

347 , ("Monitor3", testMonitor3 transport)