git.iko.soy/distributed-process
on master help sign in
Monitor disconnect
commit7bcba47f0f97ebf4af704316412e4d6a2f9465e4 authorEdsko de Vries <edsko@edsko.net> dateWed 13 Jun 2012, 18:05 (14 years ago) parent5ffbd26
2 files changed+39−17
distributed-process/src/Control/Distributed/Process.hs+8−16 distributed-process/tests/TestCH.hs+31−1
distributed-process/src/Control/Distributed/Process.hs+8−16
@@ -86,7 +86,7 @@ import qualified Data.ByteString.Lazy as BSL ( ByteString
8686 )
8787import Data.Binary (Binary, decode, encode, put, get, putWord8, getWord8)
8888import 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)
9090import qualified Data.List as List (delete)
9191import Data.Set (Set)
9292import qualified Data.Set as Set (empty, insert, delete, member)
@@ -186,7 +186,6 @@ data LocalNodeState = LocalNodeState
186186 { _localProcesses :: Map LocalProcessId LocalProcess
187187 , _localPidCounter :: Int32
188188 , _localPidUnique :: Int32
189− , _monitorActions :: Map ProcessId [IO ()]
190189 , _localMonitorCounter :: Int32
191190 }
192191
@@ -232,6 +231,7 @@ send them msg = do
232231getSelfPid :: Process ProcessId
233232getSelfPid = processId <$> ask
234233
234+-- | Monitor another process
235235monitor :: ProcessId -> Process MonitorRef
236236monitor them = do
237237 us <- getSelfPid
@@ -269,7 +269,6 @@ expectTimeout timeout = do
269269 dequeueMatching queue (Just timeout) ((== fp) . messageFingerprint)
270270 return $ fmap (decode . messageEncoding) msg
271271
272−
273272--------------------------------------------------------------------------------
274273-- The node controller --
275274--------------------------------------------------------------------------------
@@ -488,7 +487,6 @@ newLocalNode transport = do
488487 { _localProcesses = Map.empty
489488 , _localPidCounter = 0
490489 , _localPidUnique = unq
491− , _monitorActions = Map.empty
492490 , _localMonitorCounter = 0
493491 }
494492 ctrlChan <- newChan
@@ -587,12 +585,12 @@ handleIncomingMessages node = go [] Map.empty Set.empty
587585 (Map.delete cid procConns)
588586 (Set.delete cid ctrlConns)
589587 NT.ErrorEvent (NT.TransportError (NT.EventConnectionLost (Just theirAddr) _) _) -> do
590− -- TODO: we could cache this information
591− pids <- withMVar state $ \st ->
592− return $ filter ((== theirAddr) . nodeAddress . processNodeId)
593− . Map.keys
594− $ st ^. monitorActions
595− forM_ pids $ remoteProcessFailed node
588+ -- [Unified table 9, rule node_disconnect]
589+ let nid = Right $ NodeId theirAddr
590+ writeChan ctrlChan NCMsg
591+ { ctrlMsgSender = nid
592+ , ctrlMsgSignal = Died nid DiedDisconnect
593+ }
596594 NT.ErrorEvent _ ->
597595 fail "handleIncomingMessages: TODO 4"
598596 NT.EndPointClosed ->
@@ -774,18 +772,12 @@ localPidCounter = accessor _localPidCounter (\ctr st -> st { _localPidCounter =
774772localPidUnique :: Accessor LocalNodeState Int32
775773localPidUnique = accessor _localPidUnique (\unq st -> st { _localPidUnique = unq })
776774
777−monitorActions :: Accessor LocalNodeState (Map ProcessId [IO ()])
778−monitorActions = accessor _monitorActions (\as st -> st { _monitorActions = as })
779−
780775localMonitorCounter :: Accessor LocalNodeState Int32
781776localMonitorCounter = accessor _localMonitorCounter (\ctr st -> st { _localMonitorCounter = ctr })
782777
783778localProcessWithId :: LocalProcessId -> Accessor LocalNodeState (Maybe LocalProcess)
784779localProcessWithId lpid = localProcesses >>> DAC.mapMaybe lpid
785780
786−monitorActionsFor :: ProcessId -> Accessor LocalNodeState [IO ()]
787−monitorActionsFor pid = monitorActions >>> DAC.mapDefault [] pid
788−
789781connections :: Accessor LocalProcessState (Map ProcessId NT.Connection)
790782connections = accessor _connections (\conns st -> st { _connections = conns })
791783
distributed-process/tests/TestCH.hs+31−1
@@ -70,7 +70,7 @@ testMonitorUnreachable transport = do
7070
7171 forkIO $ do
7272 localNode <- newLocalNode transport
73− addr <- forkProcess localNode $ liftIO $ newEmptyMVar >>= readMVar -- deadlock
73+ addr <- forkProcess localNode . liftIO $ threadDelay 1000000
7474 closeLocalNode localNode
7575 putMVar deadProcess addr
7676
@@ -200,6 +200,35 @@ testMonitorRemoteDeadProcess transport = do
200200
201201 takeMVar done
202202
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+
203232{-
204233-- Like 'testMonitor1', but throw an exception instead
205234testMonitor2 :: NT.Transport -> IO ()
@@ -312,6 +341,7 @@ main = do
312341 , ("MonitorAbnormalTermination", testMonitorAbnormalTermination transport)
313342 , ("MonitorLocalDeadProcess", testMonitorLocalDeadProcess transport)
314343 , ("MonitorRemoteDeadProcess", testMonitorRemoteDeadProcess transport)
344+ , ("MonitorDisconnect", testMonitorDisconnect transport)
315345 {-
316346 , ("Monitor2", testMonitor2 transport)
317347 , ("Monitor3", testMonitor3 transport)