@@ -25,6 +25,7 @@ import System.IO (fixIO, hPutStrLn, stderr)
25import System.Mem.Weak (Weak, deRefWeak)
25import System.Mem.Weak (Weak, deRefWeak)
26import qualified Data.ByteString.Lazy as BSL (fromChunks)
26import qualified Data.ByteString.Lazy as BSL (fromChunks)
27import Data.Binary (decode)
27import Data.Binary (decode)
28+import Data.Function (fix)
28import Data.Map (Map)
29import Data.Map (Map)
29import qualified Data.Map as Map
30import qualified Data.Map as Map
@@ -78,6 +79,8 @@ import qualified Control.Exception as Exception
82import Control.Concurrent (forkIO, killThread)
85import Control.Concurrent (forkIO, killThread)
83import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
86import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
@@ -98,7 +101,11 @@ import Control.Distributed.Process.Internal.ThreadPool
98 , ThreadPool
101 , ThreadPool
100import Control.Concurrent.Chan (newChan, writeChan, readChan)
103import Control.Concurrent.Chan (newChan, writeChan, readChan)
101−import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
104+import qualified Control.Concurrent.MVar as MVar
102import Control.Concurrent.STM
109import Control.Concurrent.STM
103 ( atomically
110 ( atomically
@@ -230,6 +237,7 @@ import Control.Monad.Catch (try)
230import GHC.IO (IO(..), unsafeUnmask)
237import GHC.IO (IO(..), unsafeUnmask)
231import GHC.Base ( maskAsyncExceptions# )
238import GHC.Base ( maskAsyncExceptions# )
240+import System.Timeout
233import Unsafe.Coerce
241import Unsafe.Coerce
234import Prelude
242import Prelude
@@ -520,24 +528,78 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
521 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
529 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
531+-- | And 'EventMonitor' monitors the handling of events reporting when an event
532+-- is not handled promptly. It monitors events which are handled in sequence.
533+data EventMonitor a = EventMonitor
534+ { -- | Notifies that an event was received. Takes the event as argument.
535+ eventMonitorGotEvent :: a -> IO ()
536+ -- | Notifies that the last received event was handled.
538+ -- Might be called even when there is no event, in which case it will have
540+ , eventMonitorHandledEvent :: IO ()
541+ -- | Stops the event monitor.
542+ , eventMonitorStop :: IO ()
545+-- | `eventMon <- monitorThread t action`
547+-- Creates an event monitor that executes `action` when an event hasn't been
548+-- handled withing `t` microseconds. `action` takes as argument the event that
549+-- wasn't handled timely.
551+monitorThread :: Int -> (a -> IO ()) -> IO (EventMonitor a)
552+monitorThread tmicrosecs action = do
553+ mv <- MVar.newEmptyMVar
554+ tid <- forkIO $ forever' $ do
555+ -- loop until some event is reported
556+ ev <- fix $ \loop -> MVar.takeMVar mv >>= maybe loop return
557+ -- wait until the event is handled
558+ Exception.mask_ (timeout tmicrosecs $ MVar.takeMVar mv) >>=
559+ maybe (do action ev -- report the unhandled event
560+ -- wait for the slow event handler
561+ void $ MVar.takeMVar mv
565+ { eventMonitorGotEvent = MVar.putMVar mv . Just
566+ , eventMonitorHandledEvent = MVar.putMVar mv Nothing
567+ , eventMonitorStop = killThread tid
523handleIncomingMessages :: LocalNode -> IO ()
570handleIncomingMessages :: LocalNode -> IO ()
524−handleIncomingMessages node = go initConnectionState
571+handleIncomingMessages node =
573+ (monitorThread 1000000 $ \ev ->
575+ "handleIncomingMessages blocked " ++ show (localNodeId node) ++
578+ eventMonitorStop $ \eventMon ->
579+ (go eventMon initConnectionState
525− `Exception.catch` \(NodeClosedException _) -> return ()
580+ `Exception.catch` \(NodeClosedException _) -> return ()
581+ ) `Exception.catch` \e -> do
583+ "handleIncomingMessages failed: " ++
584+ show (localNodeId node, e :: SomeException)
527− go :: ConnectionState -> IO ()
587+ go :: EventMonitor NT.Event -> ConnectionState -> IO ()
528− go !st = do
588+ go eventMon !st = do
589+ eventMonitorHandledEvent eventMon
529 event <- NT.receive endpoint
590 event <- NT.receive endpoint
591+ eventMonitorGotEvent eventMon event
530 case event of
592 case event of
531 NT.ConnectionOpened cid rel theirAddr ->
593 NT.ConnectionOpened cid rel theirAddr ->
532 if rel == NT.ReliableOrdered
594 if rel == NT.ReliableOrdered
534 trace node (MxConnected cid theirAddr)
596 trace node (MxConnected cid theirAddr)
535− >> go (
597+ >> go eventMon (
536 (incomingAt cid ^= Just (theirAddr, Uninit))
598 (incomingAt cid ^= Just (theirAddr, Uninit))
537 . (incomingFrom theirAddr ^: Set.insert cid)
599 . (incomingFrom theirAddr ^: Set.insert cid)
540− else invalidRequest cid st $
602+ else invalidRequest eventMon cid st $
541 "attempt to connect with unsupported reliability " ++ show rel
603 "attempt to connect with unsupported reliability " ++ show rel
542 NT.Received cid payload ->
604 NT.Received cid payload ->
543 case st ^. incomingAt cid of
605 case st ^. incomingAt cid of
@@ -549,7 +611,7 @@ handleIncomingMessages node = go initConnectionState
549 let msg = payloadToMessage payload
611 let msg = payloadToMessage payload
550 enqueue queue msg -- 'enqueue' is strict
612 enqueue queue msg -- 'enqueue' is strict
551 trace node (MxReceived pid msg)
613 trace node (MxReceived pid msg)
552− go st
614+ go eventMon st
553 Just (_, ToChan (TypedChannel chan')) -> do
615 Just (_, ToChan (TypedChannel chan')) -> do
554 mChan <- deRefWeak chan'
616 mChan <- deRefWeak chan'
555 -- If mChan is Nothing, the process has given up the read end of
617 -- If mChan is Nothing, the process has given up the read end of
@@ -557,11 +619,11 @@ handleIncomingMessages node = go initConnectionState
557 forM_ mChan $ \chan -> atomically $
619 forM_ mChan $ \chan -> atomically $
558 -- We make sure the message is fully decoded when it is enqueued
620 -- We make sure the message is fully decoded when it is enqueued
559 writeTQueue chan $! decode (BSL.fromChunks payload)
621 writeTQueue chan $! decode (BSL.fromChunks payload)
560− go st
622+ go eventMon st
561 Just (_, ToNode) -> do
623 Just (_, ToNode) -> do
562 let ctrlMsg = decode . BSL.fromChunks $ payload
624 let ctrlMsg = decode . BSL.fromChunks $ payload
563 writeChan ctrlChan $! ctrlMsg
625 writeChan ctrlChan $! ctrlMsg
564− go st
626+ go eventMon st
565 Just (src, Uninit) ->
627 Just (src, Uninit) ->
566 case decode (BSL.fromChunks payload) of
628 case decode (BSL.fromChunks payload) of
567 ProcessIdentifier pid -> do
629 ProcessIdentifier pid -> do
@@ -569,11 +631,11 @@ handleIncomingMessages node = go initConnectionState
569 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
631 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
570 case mProc of
632 case mProc of
571 Just proc ->
633 Just proc ->
572− go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
634+ go eventMon (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
573 Nothing ->
635 Nothing ->
574 -- incoming attempt to connect to unknown process - might
636 -- incoming attempt to connect to unknown process - might
575 -- be dead already
637 -- be dead already
576− go (incomingAt cid ^= Nothing $ st)
638+ go eventMon (incomingAt cid ^= Nothing $ st)
577 SendPortIdentifier chId -> do
639 SendPortIdentifier chId -> do
578 let lcid = sendPortLocalId chId
640 let lcid = sendPortLocalId chId
579 lpid = processLocalId (sendPortProcessId chId)
641 lpid = processLocalId (sendPortProcessId chId)
@@ -583,33 +645,34 @@ handleIncomingMessages node = go initConnectionState
583 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
645 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
584 case mChannel of
646 case mChannel of
585 Just channel ->
647 Just channel ->
586− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
649+ (incomingAt cid ^= Just (src, ToChan channel) $ st)
587 Nothing ->
650 Nothing ->
588− invalidRequest cid st $
651+ invalidRequest eventMon cid st $
589 "incoming attempt to connect to unknown channel of"
652 "incoming attempt to connect to unknown channel of"
590 ++ " process " ++ show (sendPortProcessId chId)
653 ++ " process " ++ show (sendPortProcessId chId)
591 Nothing ->
654 Nothing ->
592 -- incoming attempt to connect to channel of unknown
655 -- incoming attempt to connect to channel of unknown
593 -- process - might be dead already
656 -- process - might be dead already
594− go (incomingAt cid ^= Nothing $ st)
657+ go eventMon (incomingAt cid ^= Nothing $ st)
595 NodeIdentifier nid ->
658 NodeIdentifier nid ->
596 if nid == localNodeId node
659 if nid == localNodeId node
597− then go (incomingAt cid ^= Just (src, ToNode) $ st)
660+ then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)
598− else invalidRequest cid st $
661+ else invalidRequest eventMon cid st $
599 "incoming attempt to connect to a different node -"
662 "incoming attempt to connect to a different node -"
600 ++ " I'm " ++ show (localNodeId node)
663 ++ " I'm " ++ show (localNodeId node)
601 ++ " but the remote peer wants to connect to "
664 ++ " but the remote peer wants to connect to "
602 ++ show nid
665 ++ show nid
603 Nothing ->
666 Nothing ->
604− invalidRequest cid st
667+ invalidRequest eventMon cid st
605 "message received from an unknown connection"
668 "message received from an unknown connection"
606 NT.ConnectionClosed cid ->
669 NT.ConnectionClosed cid ->
607 case st ^. incomingAt cid of
670 case st ^. incomingAt cid of
608 Nothing ->
671 Nothing ->
609− invalidRequest cid st "closed unknown connection"
672+ invalidRequest eventMon cid st "closed unknown connection"
610 Just (src, _) -> do
673 Just (src, _) -> do
611 trace node (MxDisconnected cid src)
674 trace node (MxDisconnected cid src)
612− go ( (incomingAt cid ^= Nothing)
675+ go eventMon ( (incomingAt cid ^= Nothing)
613 . (incomingFrom src ^: Set.delete cid)
676 . (incomingFrom src ^: Set.delete cid)
@@ -624,7 +687,7 @@ handleIncomingMessages node = go initConnectionState
624 liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
687 liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
625 >>= maybe (return ()) killThread
688 >>= maybe (return ()) killThread
626 closeImplicitReconnections node nid
689 closeImplicitReconnections node nid
627− go ( (incomingFrom theirAddr ^= Set.empty)
690+ go eventMon ( (incomingFrom theirAddr ^= Set.empty)
628 . (incoming ^: Map.filterWithKey (const . notLost))
691 . (incoming ^: Map.filterWithKey (const . notLost))
@@ -639,8 +702,10 @@ handleIncomingMessages node = go initConnectionState
639 -- and we just give up
702 -- and we just give up
640 fail "Cloud Haskell fatal error: received unexpected multicast"
703 fail "Cloud Haskell fatal error: received unexpected multicast"
705+ invalidRequest :: EventMonitor NT.Event
642− invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
707+ -> ConnectionState -> String -> IO ()
643− invalidRequest cid st msg = do
708+ invalidRequest eventMon cid st msg = do
644 -- TODO: We should treat this as a fatal error on the part of the remote
709 -- TODO: We should treat this as a fatal error on the part of the remote
645 -- node. That is, we should report the remote node as having died, and we
710 -- node. That is, we should report the remote node as having died, and we
646 -- should close incoming connections (this requires a Transport layer
711 -- should close incoming connections (this requires a Transport layer
@@ -649,7 +714,7 @@ handleIncomingMessages node = go initConnectionState
649 ++ " (" ++ msg ++ "): "
714 ++ " (" ++ msg ++ "): "
650 , (Trace cid)
715 , (Trace cid)
652− go ( incomingAt cid ^= Nothing
717+ go eventMon ( incomingAt cid ^= Nothing
@@ -661,9 +726,21 @@ handleIncomingMessages node = go initConnectionState
661--------------------------------------------------------------------------------
726--------------------------------------------------------------------------------
663runNodeController :: LocalNode -> IO ()
728runNodeController :: LocalNode -> IO ()
664−runNodeController node =
729+runNodeController node = do
731+ (monitorThread 1000000 $ \ev ->
733+ "nodeController blocked " ++ show (localNodeId node) ++
736+ eventMonitorStop $ \eventMon ->
665− runReaderT (evalStateT (unNC nodeController) initNCState) node
737+ (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node
666− `Exception.catch` \(NodeClosedException _) -> return ()
738+ `Exception.catch` \(NodeClosedException _) -> return ()
739+ ) `Exception.catch` \e -> do
741+ "nodeController failed: " ++
742+ show (localNodeId node, e :: SomeException)
668--------------------------------------------------------------------------------
745--------------------------------------------------------------------------------
669-- Internal data types --
746-- Internal data types --
@@ -772,11 +849,13 @@ withLocalTracer node act = act (localEventBus node)
772--------------------------------------------------------------------------------
849--------------------------------------------------------------------------------
774-- [Unified: Table 7]
851-- [Unified: Table 7]
775−nodeController :: NC ()
852+nodeController :: EventMonitor NCMsg -> NC ()
776−nodeController = do
853+nodeController eventMon = do
777 node <- ask
854 node <- ask
778 forever' $ do
855 forever' $ do
856+ liftIO $ eventMonitorHandledEvent eventMon
779 msg <- liftIO $ readChan (localCtrlChan node)
857 msg <- liftIO $ readChan (localCtrlChan node)
858+ liftIO $ eventMonitorGotEvent eventMon msg
781 -- [Unified: Table 7, rule nc_forward]
860 -- [Unified: Table 7, rule nc_forward]
782 case destNid (ctrlMsgSignal msg) of
861 case destNid (ctrlMsgSignal msg) of