git.iko.soy/distributed-process
on master help sign in
Monitor the NC and handleIncomingMessages.
e219567Facundo Domínguez
src/Control/Distributed/Process/Node.hs+115−30
compare with master
unifiedsplit
@@ -29,6 +29,7 @@ import System.IO (fixIO, hPutStrLn, stderr)
29import System.Mem.Weak (Weak, deRefWeak)
29import System.Mem.Weak (Weak, deRefWeak)
30import qualified Data.ByteString.Lazy as BSL (fromChunks)
30import qualified Data.ByteString.Lazy as BSL (fromChunks)
31import Data.Binary (decode)
31import Data.Binary (decode)
32+import Data.Function (fix)
32import Data.Map (Map)
33import Data.Map (Map)
33import qualified Data.Map as Map
34import qualified Data.Map as Map
34 ( empty
35 ( empty
@@ -74,8 +75,16 @@ import qualified Control.Exception as Exception
74 , catch
75 , catch
75 , catches
76 , catches
76 , finally
77 , finally
78+ , bracket
79+ , mask_
80+ )
81+import Control.Concurrent
82+ ( forkIO
83+ , forkIOWithUnmask
84+ , killThread
85+ , myThreadId
86+ , ThreadId
77 )
87 )
78−import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)
79import Control.Distributed.Process.Internal.StrictMVar
88import Control.Distributed.Process.Internal.StrictMVar
80 ( newMVar
89 ( newMVar
81 , withMVar
90 , withMVar
@@ -93,7 +102,11 @@ import Control.Distributed.Process.Internal.ThreadPool
93 , ThreadPool
102 , ThreadPool
94 )
103 )
95import Control.Concurrent.Chan (newChan, writeChan, readChan)
104import Control.Concurrent.Chan (newChan, writeChan, readChan)
96−import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
105+import qualified Control.Concurrent.MVar as MVar
106+ ( newEmptyMVar
107+ , takeMVar
108+ , putMVar
109+ )
97import Control.Concurrent.STM
110import Control.Concurrent.STM
98 ( atomically
111 ( atomically
99 )
112 )
@@ -230,6 +243,7 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a
230import GHC.IO (IO(..), unsafeUnmask)
243import GHC.IO (IO(..), unsafeUnmask)
231import GHC.Base ( maskAsyncExceptions# )
244import GHC.Base ( maskAsyncExceptions# )
232
245
246+import System.Timeout
233import Unsafe.Coerce
247import Unsafe.Coerce
234
248
235-- Remove these when this is fixed:
249-- Remove these when this is fixed:
@@ -521,24 +535,78 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
521 where
535 where
522 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
536 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
523
537
538+-- | And 'EventMonitor' monitors the handling of events reporting when an event
539+-- is not handled promptly. It monitors events which are handled in sequence.
540+data EventMonitor a = EventMonitor
541+ { -- | Notifies that an event was received. Takes the event as argument.
542+ eventMonitorGotEvent :: a -> IO ()
543+ -- | Notifies that the last received event was handled.
544+ --
545+ -- Might be called even when there is no event, in which case it will have
546+ -- no effect.
547+ , eventMonitorHandledEvent :: IO ()
548+ -- | Stops the event monitor.
549+ , eventMonitorStop :: IO ()
550+ }
551+
552+-- | `eventMon <- monitorThread t action`
553+--
554+-- Creates an event monitor that executes `action` when an event hasn't been
555+-- handled withing `t` microseconds. `action` takes as argument the event that
556+-- wasn't handled timely.
557+--
558+monitorThread :: Int -> (a -> IO ()) -> IO (EventMonitor a)
559+monitorThread tmicrosecs action = do
560+ mv <- MVar.newEmptyMVar
561+ tid <- forkIO $ forever' $ do
562+ -- loop until some event is reported
563+ ev <- fix $ \loop -> MVar.takeMVar mv >>= maybe loop return
564+ -- wait until the event is handled
565+ Exception.mask_ (timeout tmicrosecs $ MVar.takeMVar mv) >>=
566+ maybe (do action ev -- report the unhandled event
567+ -- wait for the slow event handler
568+ void $ MVar.takeMVar mv
569+ )
570+ (const $ return ())
571+ return EventMonitor
572+ { eventMonitorGotEvent = MVar.putMVar mv . Just
573+ , eventMonitorHandledEvent = MVar.putMVar mv Nothing
574+ , eventMonitorStop = killThread tid
575+ }
576+
524handleIncomingMessages :: LocalNode -> IO ()
577handleIncomingMessages :: LocalNode -> IO ()
525−handleIncomingMessages node = go initConnectionState
578+handleIncomingMessages node =
579+ Exception.bracket
580+ (monitorThread 1000000 $ \ev ->
581+ hPutStrLn stderr $
582+ "handleIncomingMessages blocked" ++ show (localNodeId node) ++
583+ " on " ++ show ev
584+ )
585+ eventMonitorStop $ \eventMon ->
586+ (go eventMon initConnectionState
526− `Exception.catch` \(NodeClosedException _) -> return ()
587+ `Exception.catch` \(NodeClosedException _) -> return ()
588+ ) `Exception.catch` \e -> do
589+ hPutStrLn stderr $
590+ "handleIncomingMessages failed: " ++
591+ show (localNodeId node, e :: SomeException)
592+ throwIO e
527 where
593 where
528− go :: ConnectionState -> IO ()
594+ go :: EventMonitor NT.Event -> ConnectionState -> IO ()
529− go !st = do
595+ go eventMon !st = do
596+ eventMonitorHandledEvent eventMon
530 event <- NT.receive endpoint
597 event <- NT.receive endpoint
598+ eventMonitorGotEvent eventMon event
531 case event of
599 case event of
532 NT.ConnectionOpened cid rel theirAddr ->
600 NT.ConnectionOpened cid rel theirAddr ->
533 if rel == NT.ReliableOrdered
601 if rel == NT.ReliableOrdered
534 then
602 then
535 trace node (MxConnected cid theirAddr)
603 trace node (MxConnected cid theirAddr)
536− >> go (
604+ >> go eventMon (
537 (incomingAt cid ^= Just (theirAddr, Uninit))
605 (incomingAt cid ^= Just (theirAddr, Uninit))
538 . (incomingFrom theirAddr ^: Set.insert cid)
606 . (incomingFrom theirAddr ^: Set.insert cid)
539 $ st
607 $ st
540 )
608 )
541− else invalidRequest cid st
609+ else invalidRequest eventMon cid st
542 NT.Received cid payload ->
610 NT.Received cid payload ->
543 case st ^. incomingAt cid of
611 case st ^. incomingAt cid of
544 Just (_, ToProc pid weakQueue) -> do
612 Just (_, ToProc pid weakQueue) -> do
@@ -549,7 +617,7 @@ handleIncomingMessages node = go initConnectionState
549 let msg = payloadToMessage payload
617 let msg = payloadToMessage payload
550 enqueue queue msg -- 'enqueue' is strict
618 enqueue queue msg -- 'enqueue' is strict
551 trace node (MxReceived pid msg)
619 trace node (MxReceived pid msg)
552− go st
620+ go eventMon st
553 Just (_, ToChan (TypedChannel chan')) -> do
621 Just (_, ToChan (TypedChannel chan')) -> do
554 mChan <- deRefWeak chan'
622 mChan <- deRefWeak chan'
555 -- If mChan is Nothing, the process has given up the read end of
623 -- If mChan is Nothing, the process has given up the read end of
@@ -557,11 +625,11 @@ handleIncomingMessages node = go initConnectionState
557 forM_ mChan $ \chan -> atomically $
625 forM_ mChan $ \chan -> atomically $
558 -- We make sure the message is fully decoded when it is enqueued
626 -- We make sure the message is fully decoded when it is enqueued
559 writeTQueue chan $! decode (BSL.fromChunks payload)
627 writeTQueue chan $! decode (BSL.fromChunks payload)
560− go st
628+ go eventMon st
561 Just (_, ToNode) -> do
629 Just (_, ToNode) -> do
562 let ctrlMsg = decode . BSL.fromChunks $ payload
630 let ctrlMsg = decode . BSL.fromChunks $ payload
563 writeChan ctrlChan $! ctrlMsg
631 writeChan ctrlChan $! ctrlMsg
564− go st
632+ go eventMon st
565 Just (src, Uninit) ->
633 Just (src, Uninit) ->
566 case decode (BSL.fromChunks payload) of
634 case decode (BSL.fromChunks payload) of
567 ProcessIdentifier pid -> do
635 ProcessIdentifier pid -> do
@@ -569,9 +637,9 @@ handleIncomingMessages node = go initConnectionState
569 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
637 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
570 case mProc of
638 case mProc of
571 Just proc ->
639 Just proc ->
572− go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
640+ go eventMon (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
573 Nothing ->
641 Nothing ->
574− invalidRequest cid st
642+ invalidRequest eventMon cid st
575 SendPortIdentifier chId -> do
643 SendPortIdentifier chId -> do
576 let lcid = sendPortLocalId chId
644 let lcid = sendPortLocalId chId
577 lpid = processLocalId (sendPortProcessId chId)
645 lpid = processLocalId (sendPortProcessId chId)
@@ -581,24 +649,25 @@ handleIncomingMessages node = go initConnectionState
581 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
649 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
582 case mChannel of
650 case mChannel of
583 Just channel ->
651 Just channel ->
652+ go eventMon
584− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
653+ (incomingAt cid ^= Just (src, ToChan channel) $ st)
585 Nothing ->
654 Nothing ->
586− invalidRequest cid st
655+ invalidRequest eventMon cid st
587 Nothing ->
656 Nothing ->
588− invalidRequest cid st
657+ invalidRequest eventMon cid st
589 NodeIdentifier nid ->
658 NodeIdentifier nid ->
590 if nid == localNodeId node
659 if nid == localNodeId node
591− then go (incomingAt cid ^= Just (src, ToNode) $ st)
660+ then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)
592− else invalidRequest cid st
661+ else invalidRequest eventMon cid st
593 Nothing ->
662 Nothing ->
594− invalidRequest cid st
663+ invalidRequest eventMon cid st
595 NT.ConnectionClosed cid ->
664 NT.ConnectionClosed cid ->
596 case st ^. incomingAt cid of
665 case st ^. incomingAt cid of
597 Nothing ->
666 Nothing ->
598− invalidRequest cid st
667+ invalidRequest eventMon cid st
599 Just (src, _) -> do
668 Just (src, _) -> do
600 trace node (MxDisconnected cid src)
669 trace node (MxDisconnected cid src)
601− go ( (incomingAt cid ^= Nothing)
670+ go eventMon ( (incomingAt cid ^= Nothing)
602 . (incomingFrom src ^: Set.delete cid)
671 . (incomingFrom src ^: Set.delete cid)
603 $ st
672 $ st
604 )
673 )
@@ -613,7 +682,7 @@ handleIncomingMessages node = go initConnectionState
613 liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
682 liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
614 >>= maybe (return ()) killThread
683 >>= maybe (return ()) killThread
615 closeImplicitReconnections node nid
684 closeImplicitReconnections node nid
616− go ( (incomingFrom theirAddr ^= Set.empty)
685+ go eventMon ( (incomingFrom theirAddr ^= Set.empty)
617 . (incoming ^: Map.filterWithKey (const . notLost))
686 . (incoming ^: Map.filterWithKey (const . notLost))
618 $ st
687 $ st
619 )
688 )
@@ -628,15 +697,17 @@ handleIncomingMessages node = go initConnectionState
628 -- and we just give up
697 -- and we just give up
629 fail "Cloud Haskell fatal error: received unexpected multicast"
698 fail "Cloud Haskell fatal error: received unexpected multicast"
630
699
700+ invalidRequest :: EventMonitor NT.Event
701+ -> NT.ConnectionId
631− invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
702+ -> ConnectionState -> IO ()
632− invalidRequest cid st = do
703+ invalidRequest eventMon cid st = do
633 -- TODO: We should treat this as a fatal error on the part of the remote
704 -- TODO: We should treat this as a fatal error on the part of the remote
634 -- node. That is, we should report the remote node as having died, and we
705 -- node. That is, we should report the remote node as having died, and we
635 -- should close incoming connections (this requires a Transport layer
706 -- should close incoming connections (this requires a Transport layer
636 -- extension).
707 -- extension).
637 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
708 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
638 (Trace cid)]
709 (Trace cid)]
639− go ( incomingAt cid ^= Nothing
710+ go eventMon ( incomingAt cid ^= Nothing
640 $ st
711 $ st
641 )
712 )
642
713
@@ -648,9 +719,21 @@ handleIncomingMessages node = go initConnectionState
648--------------------------------------------------------------------------------
719--------------------------------------------------------------------------------
649
720
650runNodeController :: LocalNode -> IO ()
721runNodeController :: LocalNode -> IO ()
651−runNodeController node =
722+runNodeController node = do
723+ Exception.bracket
724+ (monitorThread 1000000 $ \ev ->
725+ hPutStrLn stderr $
726+ "nodeController blocked" ++ show (localNodeId node) ++
727+ " on " ++ show ev
728+ )
729+ eventMonitorStop $ \eventMon ->
652− runReaderT (evalStateT (unNC nodeController) initNCState) node
730+ (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node
653− `Exception.catch` \(NodeClosedException _) -> return ()
731+ `Exception.catch` \(NodeClosedException _) -> return ()
732+ ) `Exception.catch` \e -> do
733+ hPutStrLn stderr $
734+ "nodeController failed: " ++
735+ show (localNodeId node, e :: SomeException)
736+ throwIO e
654
737
655--------------------------------------------------------------------------------
738--------------------------------------------------------------------------------
656-- Internal data types --
739-- Internal data types --
@@ -757,11 +840,13 @@ withLocalTracer node act = act (localEventBus node)
757--------------------------------------------------------------------------------
840--------------------------------------------------------------------------------
758
841
759-- [Unified: Table 7]
842-- [Unified: Table 7]
760−nodeController :: NC ()
843+nodeController :: EventMonitor NCMsg -> NC ()
761−nodeController = do
844+nodeController eventMon = do
762 node <- ask
845 node <- ask
763 forever' $ do
846 forever' $ do
847+ liftIO $ eventMonitorHandledEvent eventMon
764 msg <- liftIO $ readChan (localCtrlChan node)
848 msg <- liftIO $ readChan (localCtrlChan node)
849+ liftIO $ eventMonitorGotEvent eventMon msg
765
850
766 -- [Unified: Table 7, rule nc_forward]
851 -- [Unified: Table 7, rule nc_forward]
767 case destNid (ctrlMsgSignal msg) of
852 case destNid (ctrlMsgSignal msg) of