Permalink
Browse files

Merge pull request #3394 from bdurand/buffered_logger_threads

Fix threading issues with BufferedLogger.
  • Loading branch information...
2 parents e2aaae1 + d565fda commit b838570bd69ff13d677fb43e79f10d6f3168c696 @josevalim josevalim committed Oct 21, 2011
Showing with 89 additions and 8 deletions.
  1. +36 −8 activesupport/lib/active_support/buffered_logger.rb
  2. +53 −0 activesupport/test/buffered_logger_test.rb
@@ -25,22 +25,28 @@ module Severity
# Silences the logger for the duration of the block.
def silence(temporary_level = ERROR)
if silencer
+ old_logger_level = @tmp_levels[Thread.current]
begin
- old_logger_level, self.level = level, temporary_level
+ @tmp_levels[Thread.current] = temporary_level
yield self
ensure
- self.level = old_logger_level
+ if old_logger_level
+ @tmp_levels[Thread.current] = old_logger_level
+ else
+ @tmp_levels.delete(Thread.current)
+ end
end
else
yield self
end
end
- attr_accessor :level
+ attr_writer :level
attr_reader :auto_flushing
def initialize(log, level = DEBUG)
@level = level
+ @tmp_levels = {}
@buffer = Hash.new { |h,k| h[k] = [] }
@auto_flushing = 1
@guard = Mutex.new
@@ -62,8 +68,12 @@ def open_log(log, mode)
end
end
+ def level
+ @tmp_levels[Thread.current] || @level
+ end
+
def add(severity, message = nil, progname = nil, &block)
- return if @level > severity
+ return if level > severity
message = (message || (block && block.call) || progname).to_s
# If a newline is necessary then create a new message ending with a newline.
# Ensures that the original message is not mutated.
@@ -84,7 +94,7 @@ def #{severity.downcase}(message = nil, progname = nil, &block) # def debug(mess
end # end
def #{severity.downcase}? # def debug?
- #{severity} >= @level # DEBUG >= @level
+ #{severity} >= level # DEBUG >= @level
end # end
EOT
end
@@ -105,13 +115,15 @@ def auto_flushing=(period)
def flush
@guard.synchronize do
- buffer.each do |content|
- @log.write(content)
- end
+ write_buffer(buffer)
# Important to do this even if buffer was empty or else @buffer will
# accumulate empty arrays for each request where nothing was logged.
clear_buffer
+
+ # Clear buffers associated with dead threads or else spawned threads
+ # that don't call flush will result in a memory leak.
+ flush_dead_buffers
end
end
@@ -133,5 +145,21 @@ def buffer
def clear_buffer
@buffer.delete(Thread.current)
end
+
+ # Find buffers created by threads that are no longer alive and flush them to the log
+ # in order to prevent memory leaks from spawned threads.
+ def flush_dead_buffers #:nodoc:
+ @buffer.keys.reject{|thread| thread.alive?}.each do |thread|
+ buffer = @buffer[thread]
+ write_buffer(buffer)
+ @buffer.delete(thread)
+ end
+ end
+
+ def write_buffer(buffer)
+ buffer.each do |content|
+ @log.write(content)
+ end
+ end
end
end
@@ -198,4 +198,57 @@ def test_buffer_multibyte
end
assert byte_string.include?(BYTE_STRING)
end
+
+ def test_silence_only_current_thread
+ @logger.auto_flushing = true
+ run_thread_a = false
+
+ a = Thread.new do
+ while !run_thread_a do
+ sleep(0.001)
+ end
+ @logger.info("x")
+ run_thread_a = false
+ end
+
+ @logger.silence do
+ run_thread_a = true
+ @logger.info("a")
+ while run_thread_a do
+ sleep(0.001)
+ end
+ end
+
+ a.join
+
+ assert @output.string.include?("x")
+ assert !@output.string.include?("a")
+ end
+
+ def test_flush_dead_buffers
+ @logger.auto_flushing = false
+
+ a = Thread.new do
+ @logger.info("a")
+ end
+
+ keep_running = true
+ b = Thread.new do
+ @logger.info("b")
+ while keep_running
+ sleep(0.001)
+ end
+ end
+
+ @logger.info("x")
+ a.join
+ @logger.flush
+
+
+ assert @output.string.include?("x")
+ assert @output.string.include?("a")
+ assert !@output.string.include?("b")
+
+ keep_running = false
+ end
end

0 comments on commit b838570

Please sign in to comment.