@@ -25,6 +25,7 @@ import System.IO (fixIO, hPutStrLn, stderr)
2525import System.Mem.Weak (Weak, deRefWeak)
2626import qualified Data.ByteString.Lazy as BSL (fromChunks)
2727import Data.Binary (decode)
28+import Data.Function (fix)
2829import Data.Map (Map)
2930import qualified Data.Map as Map
3031 ( empty
@@ -78,6 +79,8 @@ import qualified Control.Exception as Exception
7879 , catch
7980 , catches
8081 , finally
82+ , bracket
83+ , mask_
8184 )
8285import Control.Concurrent (forkIO, killThread)
8386import Control.Distributed.Process.Internal.BiMultiMap (BiMultiMap)
@@ -91,8 +94,18 @@ import Control.Distributed.Process.Internal.StrictMVar
9194 , putMVar
9295 , takeMVar
9396 )
97+import Control.Distributed.Process.Internal.ThreadPool
98+ ( newThreadPool
99+ , submitTask
100+ , lookupWorker
101+ , ThreadPool
102+ )
94103import Control.Concurrent.Chan (newChan, writeChan, readChan)
95−import qualified Control.Concurrent.MVar as MVar (newEmptyMVar, takeMVar)
104+import qualified Control.Concurrent.MVar as MVar
105+ ( newEmptyMVar
106+ , takeMVar
107+ , putMVar
108+ )
96109import Control.Concurrent.STM
97110 ( atomically
98111 )
@@ -224,6 +237,7 @@ import Control.Monad.Catch (try)
224237import GHC.IO (IO(..), unsafeUnmask)
225238import GHC.Base ( maskAsyncExceptions# )
226239
240+import System.Timeout
227241import Unsafe.Coerce
228242import Prelude
229243
@@ -264,12 +278,14 @@ createBareLocalNode endPoint rtable = do
264278 , _localConnections = Map.empty
265279 }
266280 ctrlChan <- newChan
281+ sendPool <- newThreadPool
267282 let node = LocalNode { localNodeId = NodeId $ NT.address endPoint
268283 , localEndPoint = endPoint
269284 , localState = state
270285 , localCtrlChan = ctrlChan
271286 , localEventBus = MxEventBusInitialising
272287 , remoteTable = rtable
288+ , localSendPool = sendPool
273289 }
274290 tracedNode <- startMxAgent node
275291
@@ -512,24 +528,78 @@ incomingFrom addr = aux >>> DAC.mapDefault Set.empty addr
512528 where
513529 aux = accessor _incomingFrom (\fr st -> st { _incomingFrom = fr })
514530
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.
537+ --
538+ -- Might be called even when there is no event, in which case it will have
539+ -- no effect.
540+ , eventMonitorHandledEvent :: IO ()
541+ -- | Stops the event monitor.
542+ , eventMonitorStop :: IO ()
543+ }
544+
545+-- | `eventMon <- monitorThread t action`
546+--
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.
550+--
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
562+ )
563+ (const $ return ())
564+ return EventMonitor
565+ { eventMonitorGotEvent = MVar.putMVar mv . Just
566+ , eventMonitorHandledEvent = MVar.putMVar mv Nothing
567+ , eventMonitorStop = killThread tid
568+ }
569+
515570handleIncomingMessages :: LocalNode -> IO ()
516−handleIncomingMessages node = go initConnectionState
517− `Exception.catch` \(NodeClosedException _) -> return ()
571+handleIncomingMessages node =
572+ Exception.bracket
573+ (monitorThread 1000000 $ \ev ->
574+ hPutStrLn stderr $
575+ "handleIncomingMessages blocked " ++ show (localNodeId node) ++
576+ " on " ++ show ev
577+ )
578+ eventMonitorStop $ \eventMon ->
579+ (go eventMon initConnectionState
580+ `Exception.catch` \(NodeClosedException _) -> return ()
581+ ) `Exception.catch` \e -> do
582+ hPutStrLn stderr $
583+ "handleIncomingMessages failed: " ++
584+ show (localNodeId node, e :: SomeException)
585+ throwIO e
518586 where
519− go :: ConnectionState -> IO ()
520− go !st = do
587+ go :: EventMonitor NT.Event -> ConnectionState -> IO ()
588+ go eventMon !st = do
589+ eventMonitorHandledEvent eventMon
521590 event <- NT.receive endpoint
591+ eventMonitorGotEvent eventMon event
522592 case event of
523593 NT.ConnectionOpened cid rel theirAddr ->
524594 if rel == NT.ReliableOrdered
525595 then
526596 trace node (MxConnected cid theirAddr)
527− >> go (
597+ >> go eventMon (
528598 (incomingAt cid ^= Just (theirAddr, Uninit))
529599 . (incomingFrom theirAddr ^: Set.insert cid)
530600 $ st
531601 )
532− else invalidRequest cid st $
602+ else invalidRequest eventMon cid st $
533603 "attempt to connect with unsupported reliability " ++ show rel
534604 NT.Received cid payload ->
535605 case st ^. incomingAt cid of
@@ -541,7 +611,7 @@ handleIncomingMessages node = go initConnectionState
541611 let msg = payloadToMessage payload
542612 enqueue queue msg -- 'enqueue' is strict
543613 trace node (MxReceived pid msg)
544− go st
614+ go eventMon st
545615 Just (_, ToChan (TypedChannel chan')) -> do
546616 mChan <- deRefWeak chan'
547617 -- If mChan is Nothing, the process has given up the read end of
@@ -549,11 +619,11 @@ handleIncomingMessages node = go initConnectionState
549619 forM_ mChan $ \chan -> atomically $
550620 -- We make sure the message is fully decoded when it is enqueued
551621 writeTQueue chan $! decode (BSL.fromChunks payload)
552− go st
622+ go eventMon st
553623 Just (_, ToNode) -> do
554624 let ctrlMsg = decode . BSL.fromChunks $ payload
555625 writeChan ctrlChan $! ctrlMsg
556− go st
626+ go eventMon st
557627 Just (src, Uninit) ->
558628 case decode (BSL.fromChunks payload) of
559629 ProcessIdentifier pid -> do
@@ -561,11 +631,11 @@ handleIncomingMessages node = go initConnectionState
561631 mProc <- withValidLocalState node $ return . (^. localProcessWithId lpid)
562632 case mProc of
563633 Just proc ->
564− go (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
634+ go eventMon (incomingAt cid ^= Just (src, ToProc pid (processWeakQ proc)) $ st)
565635 Nothing ->
566636 -- incoming attempt to connect to unknown process - might
567637 -- be dead already
568− go (incomingAt cid ^= Nothing $ st)
638+ go eventMon (incomingAt cid ^= Nothing $ st)
569639 SendPortIdentifier chId -> do
570640 let lcid = sendPortLocalId chId
571641 lpid = processLocalId (sendPortProcessId chId)
@@ -575,33 +645,34 @@ handleIncomingMessages node = go initConnectionState
575645 mChannel <- withMVar (processState proc) $ return . (^. typedChannelWithId lcid)
576646 case mChannel of
577647 Just channel ->
578− go (incomingAt cid ^= Just (src, ToChan channel) $ st)
648+ go eventMon
649+ (incomingAt cid ^= Just (src, ToChan channel) $ st)
579650 Nothing ->
580− invalidRequest cid st $
651+ invalidRequest eventMon cid st $
581652 "incoming attempt to connect to unknown channel of"
582653 ++ " process " ++ show (sendPortProcessId chId)
583654 Nothing ->
584655 -- incoming attempt to connect to channel of unknown
585656 -- process - might be dead already
586− go (incomingAt cid ^= Nothing $ st)
657+ go eventMon (incomingAt cid ^= Nothing $ st)
587658 NodeIdentifier nid ->
588659 if nid == localNodeId node
589− then go (incomingAt cid ^= Just (src, ToNode) $ st)
590− else invalidRequest cid st $
660+ then go eventMon (incomingAt cid ^= Just (src, ToNode) $ st)
661+ else invalidRequest eventMon cid st $
591662 "incoming attempt to connect to a different node -"
592663 ++ " I'm " ++ show (localNodeId node)
593664 ++ " but the remote peer wants to connect to "
594665 ++ show nid
595666 Nothing ->
596− invalidRequest cid st
667+ invalidRequest eventMon cid st
597668 "message received from an unknown connection"
598669 NT.ConnectionClosed cid ->
599670 case st ^. incomingAt cid of
600671 Nothing ->
601− invalidRequest cid st "closed unknown connection"
672+ invalidRequest eventMon cid st "closed unknown connection"
602673 Just (src, _) -> do
603674 trace node (MxDisconnected cid src)
604− go ( (incomingAt cid ^= Nothing)
675+ go eventMon ( (incomingAt cid ^= Nothing)
605676 . (incomingFrom src ^: Set.delete cid)
606677 $ st
607678 )
@@ -613,8 +684,10 @@ handleIncomingMessages node = go initConnectionState
613684 , ctrlMsgSignal = Died nid DiedDisconnect
614685 }
615686 let notLost k = not (k `Set.member` (st ^. incomingFrom theirAddr))
687+ liftIO $ lookupWorker (localSendPool node) (NodeId theirAddr)
688+ >>= maybe (return ()) killThread
616689 closeImplicitReconnections node nid
617− go ( (incomingFrom theirAddr ^= Set.empty)
690+ go eventMon ( (incomingFrom theirAddr ^= Set.empty)
618691 . (incoming ^: Map.filterWithKey (const . notLost))
619692 $ st
620693 )
@@ -629,8 +702,10 @@ handleIncomingMessages node = go initConnectionState
629702 -- and we just give up
630703 fail "Cloud Haskell fatal error: received unexpected multicast"
631704
632− invalidRequest :: NT.ConnectionId -> ConnectionState -> String -> IO ()
633− invalidRequest cid st msg = do
705+ invalidRequest :: EventMonitor NT.Event
706+ -> NT.ConnectionId
707+ -> ConnectionState -> String -> IO ()
708+ invalidRequest eventMon cid st msg = do
634709 -- TODO: We should treat this as a fatal error on the part of the remote
635710 -- node. That is, we should report the remote node as having died, and we
636711 -- should close incoming connections (this requires a Transport layer
@@ -639,7 +714,7 @@ handleIncomingMessages node = go initConnectionState
639714 ++ " (" ++ msg ++ "): "
640715 , (Trace cid)
641716 ]
642− go ( incomingAt cid ^= Nothing
717+ go eventMon ( incomingAt cid ^= Nothing
643718 $ st
644719 )
645720
@@ -651,9 +726,21 @@ handleIncomingMessages node = go initConnectionState
651726--------------------------------------------------------------------------------
652727
653728runNodeController :: LocalNode -> IO ()
654−runNodeController node =
655− runReaderT (evalStateT (unNC nodeController) initNCState) node
656− `Exception.catch` \(NodeClosedException _) -> return ()
729+runNodeController node = do
730+ Exception.bracket
731+ (monitorThread 1000000 $ \ev ->
732+ hPutStrLn stderr $
733+ "nodeController blocked " ++ show (localNodeId node) ++
734+ " on " ++ show ev
735+ )
736+ eventMonitorStop $ \eventMon ->
737+ (runReaderT (evalStateT (unNC (nodeController eventMon)) initNCState) node
738+ `Exception.catch` \(NodeClosedException _) -> return ()
739+ ) `Exception.catch` \e -> do
740+ hPutStrLn stderr $
741+ "nodeController failed: " ++
742+ show (localNodeId node, e :: SomeException)
743+ throwIO e
657744
658745--------------------------------------------------------------------------------
659746-- Internal data types --
@@ -669,6 +756,12 @@ data NCState = NCState
669756 , _registeredOnNodes :: !(Map ProcessId [(NodeId,Int)])
670757 }
671758
759+submitSendPool :: LocalNode -> NodeId -> IO () -> IO ()
760+submitSendPool node nid task = submitTask (localSendPool node) fork nid task
761+ where
762+ fork io =
763+ forkIO $ io `Exception.catch` \(NodeClosedException _) -> return ()
764+
672765newtype NC a = NC { unNC :: StateT NCState (ReaderT LocalNode IO) a }
673766 deriving ( Applicative
674767 , Functor
@@ -704,7 +797,7 @@ ncSendToProcessAndTrace shouldTrace pid msg = do
704797 node <- ask
705798 if processNodeId pid == localNodeId node
706799 then ncEffectLocalSendAndTrace shouldTrace node pid msg
707− else liftIO $ sendBinary node
800+ else liftIO $ submitSendPool node (processNodeId pid) $ sendBinary node
708801 (NodeIdentifier $ localNodeId node)
709802 (NodeIdentifier $ processNodeId pid)
710803 WithImplicitReconnect
@@ -717,7 +810,7 @@ ncSendToNode to msg = do
717810 node <- ask
718811 liftIO $ if to == localNodeId node
719812 then writeChan (localCtrlChan node) $! msg
720− else sendBinary node
813+ else submitSendPool node to $ sendBinary node
721814 (NodeIdentifier $ localNodeId node)
722815 (NodeIdentifier to)
723816 WithImplicitReconnect
@@ -756,11 +849,13 @@ withLocalTracer node act = act (localEventBus node)
756849--------------------------------------------------------------------------------
757850
758851-- [Unified: Table 7]
759−nodeController :: NC ()
760−nodeController = do
852+nodeController :: EventMonitor NCMsg -> NC ()
853+nodeController eventMon = do
761854 node <- ask
762855 forever' $ do
856+ liftIO $ eventMonitorHandledEvent eventMon
763857 msg <- liftIO $ readChan (localCtrlChan node)
858+ liftIO $ eventMonitorGotEvent eventMon msg
764859
765860 -- [Unified: Table 7, rule nc_forward]
766861 case destNid (ctrlMsgSignal msg) of