@@ -3,11 +3,13 @@
3module Control.Distributed.Process.Management.Agent where
3module Control.Distributed.Process.Management.Agent where
5import Control.Applicative ((<$>))
5import Control.Applicative ((<$>))
6+import Control.Concurrent (forkIO)
6import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
7import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)
7import Control.Concurrent.STM (atomically)
8import Control.Concurrent.STM (atomically)
8import Control.Concurrent.STM.TChan
9import Control.Concurrent.STM.TChan
10 , newBroadcastTChanIO
11 , newBroadcastTChanIO
11 , writeTChan
13 , writeTChan
12 , dupTChan
14 , dupTChan
@@ -73,6 +75,7 @@ mxAgentController :: Fork
73mxAgentController forkProcess mv = do
75mxAgentController forkProcess mv = do
74 trc <- liftIO $ startTracing forkProcess
76 trc <- liftIO $ startTracing forkProcess
75 sigbus <- liftIO $ newBroadcastTChanIO
77 sigbus <- liftIO $ newBroadcastTChanIO
78+ liftIO $ startDeadLetterQueue sigbus
76 weakQueue <- processWeakQ <$> ask
79 weakQueue <- processWeakQ <$> ask
77 liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)
80 liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)
78 go sigbus trc
81 go sigbus trc
@@ -115,6 +118,7 @@ mxAgentController forkProcess mv = do
115 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg
118 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg
117-- | Forks a new process in which an mxAgent is run.
120-- | Forks a new process in which an mxAgent is run.
118mxStartAgent :: Fork
122mxStartAgent :: Fork
119 -> TChan Message
123 -> TChan Message
120 -> ((TChan Message, TChan Message) -> Process ())
124 -> ((TChan Message, TChan Message) -> Process ())
@@ -124,6 +128,8 @@ mxStartAgent fork chan handler = do
124 let proc = handler (chan, chan')
128 let proc = handler (chan, chan')
125 fork proc
129 fork proc
131+-- | Start the tracer controller.
127startTracing :: Fork -> IO Tracer
133startTracing :: Fork -> IO Tracer
128startTracing forkProcess = do
134startTracing forkProcess = do
129 mv <- newEmptyMVar
135 mv <- newEmptyMVar
@@ -131,3 +137,20 @@ startTracing forkProcess = do
131 wQ <- liftIO $ takeMVar mv
137 wQ <- liftIO $ takeMVar mv
132 return $ Tracer pid wQ
138 return $ Tracer pid wQ
140+-- | Start a dead letter (agent) queue.
142+-- If no agents are registered on the system, the management
143+-- event bus will fill up and its data won't be GC'ed until someone
144+-- comes along and reads from the broadcast channel (via dupTChan
145+-- of course). This is effectively a leak, so to mitigate it, we
146+-- start a /dead letter queue/ that drains the event bus continuously,
147+-- thus ensuring if there are no other consumers that we won't use
148+-- up heap space unnecessarily.
150+startDeadLetterQueue :: TChan Message
152+startDeadLetterQueue sigbus = do
153+ chan' <- atomically (dupTChan sigbus)
154+ void $ forkIO $ forever' $ do
155+ void $ atomically $ readTChan chan'