Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

fine-gained locks while flushing #2

Merged
merged 2 commits into from Sep 18, 2018
Merged

fine-gained locks while flushing #2

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
b85e2c7
fine-gained locks
davies Sep 17, 2018
404c4b2
fix flushing
davies Sep 17, 2018
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
50 changes: 30 additions & 20 deletions lib/logstash/outputs/file.rb
View file
Open in desktop
Expand Up @@ -207,13 +207,14 @@ def extract_file_root
# the back-bone of @flusher, our periodic-flushing interval.
private
def flush_pending_files
@io_mutex.synchronize do
pending_files = @io_mutex.synchronize do
@logger.debug("Starting flush cycle")

@files.each do |path, fd|
@logger.debug("Flushing file", :path => path, :fd => fd)
fd.flush
end
# flushing all the files is slow, so we should do that without blocking
@files.clone
end
pending_files.each do |path, fd|
@logger.debug("Flushing file", :path => path, :fd => fd)
fd.flush
end
rescue => e
# squash exceptions caught while flushing after logging them
Expand Down Expand Up @@ -374,6 +375,7 @@ def run
class IOWriter
def initialize(io, buffer_size, recreate)
@io = io
@mutex = Mutex.new
@buffers = []
@size = 0
@limit = buffer_size
Expand All @@ -395,14 +397,16 @@ def reopen

public
def write(*args)
args.each do |arg|
@size += arg.bytesize
@buffers.push(arg)
end
if (@limit >= 0) and (@size > @limit)
do_write
@mutex.synchronize do
args.each do |arg|
@size += arg.bytesize
@buffers.push(arg)
end
if (@limit >= 0) and (@size > @limit)
do_write
end
@active = true
end
@active = true
end

private
Expand All @@ -417,23 +421,29 @@ def do_write

public
def flush
do_write
@io.flush
if @io.class == Zlib::GzipWriter
@io.to_io.flush
@mutex.synchronize do
do_write
@io.flush
if @io.class == Zlib::GzipWriter
@io.to_io.flush
end
end
end

public
def truncate(length)
reopen
@io.truncate(length)
@mutex.synchronize do
reopen
@io.truncate(length)
end
end

public
def close
flush
@io.close
@mutex.synchronize do
@io.close
end
end
def method_missing(method_name, *args, &block)
if @io.respond_to?(method_name)
Expand Down