0171bc6 · distributed-process

git.iko.soy/distributed-process

Fix a nasty memory leak in the management infrastructure

Previously, the management event bus could fill up with data that never got GC’ed, since no consumer (of the event bus) was guaranteed to exist. Introducing a “dead letter consumer” that continuously reads from the channel, ensures that even when no management agents are installed on a node, we still dequeue all MxEvents and thereby avoid leaking space.

commit0171bc6a82171a4fae4794763664740dd1b8cd09 authorTim Watson <watson.timothy@gmail.com> dateSat 4 Jan 2014, 03:44 (13 years ago) parent6b31b73

1 file changed+23

compare with master

unifiedsplit

src/Control/Distributed/Process/Management/Agent.hs+23

src/Control/Distributed/Process/Management/Agent.hs+23

@@ -3,11 +3,13 @@

33module Control.Distributed.Process.Management.Agent where

55import Control.Applicative ((<$>))

6+import Control.Concurrent (forkIO)

67import Control.Concurrent.MVar (MVar, newEmptyMVar, putMVar, takeMVar)

78import Control.Concurrent.STM (atomically)

89import Control.Concurrent.STM.TChan

910 ( TChan

1011 , newBroadcastTChanIO

12+ , readTChan

1113 , writeTChan

1214 , dupTChan

1315 )

@@ -73,6 +75,7 @@ mxAgentController :: Fork

7375mxAgentController forkProcess mv = do

7476 trc <- liftIO $ startTracing forkProcess

7577 sigbus <- liftIO $ newBroadcastTChanIO

78+ liftIO $ startDeadLetterQueue sigbus

7679 weakQueue <- processWeakQ <$> ask

7780 liftIO $ putMVar mv (trc, weakQueue, mxStartAgent forkProcess sigbus)

7881 go sigbus trc

@@ -115,6 +118,7 @@ mxAgentController forkProcess mv = do

115118 liftIO $ atomically $ enqueueSTM q msg >> writeTChan ch msg

116119

117120-- | Forks a new process in which an mxAgent is run.

121+--

118122mxStartAgent :: Fork

119123 -> TChan Message

120124 -> ((TChan Message, TChan Message) -> Process ())

@@ -124,6 +128,8 @@ mxStartAgent fork chan handler = do

124128 let proc = handler (chan, chan')

125129 fork proc

126130

131+-- | Start the tracer controller.

132+--

127133startTracing :: Fork -> IO Tracer

128134startTracing forkProcess = do

129135 mv <- newEmptyMVar

@@ -131,3 +137,20 @@ startTracing forkProcess = do

131137 wQ <- liftIO $ takeMVar mv

132138 return $ Tracer pid wQ

133139

140+-- | Start a dead letter (agent) queue.

141+--

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.

149+--

150+startDeadLetterQueue :: TChan Message

151+ -> IO ()

152+startDeadLetterQueue sigbus = do

153+ chan' <- atomically (dupTChan sigbus)

154+ void $ forkIO $ forever' $ do

155+ void $ atomically $ readTChan chan'

156+