@@ -29,6 +29,7 @@ import System.IO (fixIO, hPutStrLn, stderr)
2929import System.Mem.Weak (Weak, deRefWeak)
3030import qualified Data.ByteString.Lazy as BSL (fromChunks)
3131import Data.Binary (decode)
32+import Data.Function (fix)
3233import Data.Map (Map)
3334import qualified Data.Map as Map
3435 ( empty
@@ -74,8 +75,16 @@ import qualified Control.Exception as Exception
7475 , catch
7576 , catches
7677 , finally
78+ , bracket
79+ , mask_
80+ )
81+import Control.Concurrent
82+ ( forkIO
83+ , forkIOWithUnmask
84+ , killThread
85+ , myThreadId
86+ , ThreadId
7787 )
78−import Control.Concurrent (forkIO, forkIOWithUnmask, killThread, myThreadId)
7988import Control.Distributed.Process.Internal.StrictMVar
8089 ( newMVar
8190 , withMVar
@@ -93,7 +102,11 @@ import Control.Distributed.Process.Internal.ThreadPool
93102 , ThreadPool
94103 )
95104import 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+ )
97110import Control.Concurrent.STM
98111 ( atomically
99112 )
@@ -230,6 +243,7 @@ import qualified Control.Distributed.Process.Internal.StrictContainerAccessors a
230243import GHC.IO (IO(..), unsafeUnmask)
231244import GHC.Base ( maskAsyncExceptions# )
232245
246+import System.Timeout
233247import Unsafe.Coerce
234248
235249-- Remove these when this is fixed:
@@ -521,24 +535,78 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
521535 where
522536 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
523537
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+
524577handleIncomingMessages :: LocalNode -> IO ()
525−handleIncomingMessages node = go initConnectionState
526− `Exception.catch` \(NodeClosedException _) -> return ()
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
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
527593 where
528− go :: ConnectionState -> IO ()
529− go !st = do
594+ go :: EventMonitor NT.Event -> ConnectionState -> IO ()
595+ go eventMon !st = do
596+ eventMonitorHandledEvent eventMon
530597 event <- NT.receive endpoint
598+ eventMonitorGotEvent eventMon event
531599 case event of
532600 NT.ConnectionOpened cid rel theirAddr ->
533601 if rel == NT.ReliableOrdered
534602 then
535603 trace node (MxConnected cid theirAddr)
536− >> go (
604+ >> go eventMon (
537605 (incomingAt cid ^= Just (theirAddr, Uninit))
538606 . (incomingFrom theirAddr ^: Set.insert cid)
539607 $ st
540608 )
541− else invalidRequest cid st
609+ else invalidRequest eventMon cid st
542610 NT.Received cid payload ->
543611 case st ^. incomingAt cid of
544612 Just (_, ToProc pid weakQueue) -> do
@@ -549,7 +617,7 @@ handleIncomingMessages node = go initConnectionState
549617 let msg = payloadToMessage payload
550618 enqueue queue msg -- 'enqueue' is strict
551619 trace node (MxReceived pid msg)
552− go st
620+ go eventMon st
553621 Just (_, ToChan (TypedChannel chan')) -> do
554622 mChan <- deRefWeak chan'
555623 -- If mChan is Nothing, the process has given up the read end of
@@ -557,11 +625,11 @@ handleIncomingMessages node = go initConnectionState
557625 forM_ mChan $ \chan -> atomically $
558626 -- We make sure the message is fully decoded when it is enqueued
559627 writeTQueue chan $! decode (BSL.fromChunks payload)
560− go st
628+ go eventMon st
561629 Just (_, ToNode) -> do
562630 let ctrlMsg = decode . BSL.fromChunks $ payload
563631 writeChan ctrlChan $! ctrlMsg
564− go st
632+ go eventMon st
565633 Just (src, Uninit) ->
566634 case decode (BSL.fromChunks payload) of
567635 ProcessIdentifier pid -> do
@@ -569,9 +637,9 @@ handleIncomingMessages node = go initConnectionState
569637 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
570638 case mProc of
571639 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)
573641 Nothing ->
574− invalidRequest cid st
642+ invalidRequest eventMon cid st
575643 SendPortIdentifier chId -> do
576644 let lcid = sendPortLocalId chId
577645 lpid = processLocalId (sendPortProcessId chId)
@@ -581,24 +649,25 @@ handleIncomingMessages node = go initConnectionState
581649 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
582650 case mChannel of
583651 Just channel ->
584− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
652+ go eventMon
653+ (incomingAt cid ^= Just (src, ToChan channel) $ st)
585654 Nothing ->
586− invalidRequest cid st
655+ invalidRequest eventMon cid st
587656 Nothing ->
588− invalidRequest cid st
657+ invalidRequest eventMon cid st
589658 NodeIdentifier nid ->
590659 if nid == localNodeId node
591− then go (incomingAt cid ^= Just (src, ToNode) $ st)
592− else invalidRequest cid st
660+ then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)
661+ else invalidRequest eventMon cid st
593662 Nothing ->
594− invalidRequest cid st
663+ invalidRequest eventMon cid st
595664 NT.ConnectionClosed cid ->
596665 case st ^. incomingAt cid of
597666 Nothing ->
598− invalidRequest cid st
667+ invalidRequest eventMon cid st
599668 Just (src, _) -> do
600669 trace node (MxDisconnected cid src)
601− go ( (incomingAt cid ^= Nothing)
670+ go eventMon ( (incomingAt cid ^= Nothing)
602671 . (incomingFrom src ^: Set.delete cid)
603672 $ st
604673 )
@@ -613,7 +682,7 @@ handleIncomingMessages node = go initConnectionState
613682 liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
614683 >>= maybe (return ()) killThread
615684 closeImplicitReconnections node nid
616− go ( (incomingFrom theirAddr ^= Set.empty)
685+ go eventMon ( (incomingFrom theirAddr ^= Set.empty)
617686 . (incoming ^: Map.filterWithKey (const . notLost))
618687 $ st
619688 )
@@ -628,15 +697,17 @@ handleIncomingMessages node = go initConnectionState
628697 -- and we just give up
629698 fail "Cloud Haskell fatal error: received unexpected multicast"
630699
631− invalidRequest :: NT.ConnectionId -> ConnectionState -> IO ()
632− invalidRequest cid st = do
700+ invalidRequest :: EventMonitor NT.Event
701+ -> NT.ConnectionId
702+ -> ConnectionState -> IO ()
703+ invalidRequest eventMon cid st = do
633704 -- TODO: We should treat this as a fatal error on the part of the remote
634705 -- node. That is, we should report the remote node as having died, and we
635706 -- should close incoming connections (this requires a Transport layer
636707 -- extension).
637708 traceEventFmtIO node "" [(TraceStr " [network] invalid request: "),
638709 (Trace cid)]
639− go ( incomingAt cid ^= Nothing
710+ go eventMon ( incomingAt cid ^= Nothing
640711 $ st
641712 )
642713
@@ -648,9 +719,21 @@ handleIncomingMessages node = go initConnectionState
648719--------------------------------------------------------------------------------
649720
650721runNodeController :: LocalNode -> IO ()
651−runNodeController node =
652− runReaderT (evalStateT (unNC nodeController) initNCState) node
653− `Exception.catch` \(NodeClosedException _) -> return ()
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 ->
730+ (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node
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
654737
655738--------------------------------------------------------------------------------
656739-- Internal data types --
@@ -757,11 +840,13 @@ withLocalTracer node act = act (localEventBus node)
757840--------------------------------------------------------------------------------
758841
759842-- [Unified: Table 7]
760−nodeController :: NC ()
761−nodeController = do
843+nodeController :: EventMonitor NCMsg -> NC ()
844+nodeController eventMon = do
762845 node <- ask
763846 forever' $ do
847+ liftIO $ eventMonitorHandledEvent eventMon
764848 msg <- liftIO $ readChan (localCtrlChan node)
849+ liftIO $ eventMonitorGotEvent eventMon msg
765850
766851 -- [Unified: Table 7, rule nc_forward]
767852 case destNid (ctrlMsgSignal msg) of