Replies: 1 comment 1 reply
-
Ruby is essentially a singe thread due to infamous GIL. def write(chunk)
path = extract_placeholders(@path_template, chunk)
FileUtils.mkdir_p File.dirname(path), mode: @dir_perm
# begin
log.info "before loop"
abc = 0
# the loop is to slow down flush_thread, the slower, the easier to see the problem.
while abc < 500000000
# seems like io-independent operation cannot reproduce the problem
# like sleep 10 or puts "something"
abc += 1
end
log.info "after loop"
# end... will block other threads in Fluentd from running. You need to use |
Beta Was this translation helpful? Give feedback.
-
|
Thanks @fujimotos . Here is my test code. require "ruby_thread_test/test/version"
require 'thread'
require 'ruby_thread_test/ext/mycext'
include MyCExt
module Ruby_thread_test
class TestClass
def testFunc
puts "hello in TestClass"
end
def testFunc2(a)
puts "hello in TestClass, a = #{a}"
end
end
a = 0
b = 0
c = 500000000
t = TestClass.new
t1 = Thread.new do
while a < c
a += 1
end
puts "[thread1] finish"
end
t2 = Thread.new do
# delay calling function in c extension
sleep 0.5
while b < 5
b += 1
puts "[thread2] out"
test4 3, t do |idx|
puts "idxf = #{idx}"
end
end
puts "[thread2] finish"
end
t1.join
t2.join
endC extension code, #include "ruby.h"
// Defining a space for information and references about the module to be stored internally
VALUE MyTest = Qnil;
// Prototype for the initialization method - Ruby calls this, not you
void Init_mycext();
VALUE method_test4(VALUE self, VALUE c, VALUE obj);
// The initialization method for this module
void Init_mycext() {
MyTest = rb_define_module("MyCExt");
rb_define_method(MyTest, "test4", method_test4, 2);
}
VALUE method_test4(VALUE self, VALUE c, VALUE obj)
{
int count = NUM2INT(c);
for (int i = 0; i < count; ++ i) {
rb_funcall(obj, rb_intern("testFunc"), 0);
rb_funcall(obj, rb_intern("testFunc2"), 1, INT2NUM(10));
rb_yield_values(1, INT2NUM(i));
}
return INT2NUM(0);
}To me, method_test4 should not be called until thread 1 is done. But the result is method_test4 can be called, which means thread 2 is not blocked. |
Beta Was this translation helpful? Give feedback.
-
Hello,
I use elasticsearch to collect logs. First I use
in_tailto collect logs andout_forwardto forward event to fluentd aggregator. In aggregator, I usein_forwardto accept events andout_elasticsearchto bulk logs.And I encounted a problem, in which the buffer was processed extremely slow so that buffer's size became bigger and bigger.
The version of fluentd I use is 1.11.5, which is inside td-agent 3.8.1.
I checkout fluentd's source code, and checkout tag
v1.11.5, head commit is 24fe4cb.By debugging, I found a simple way to reproduce this problem, but I was not sure it's properly. I just saw the similar
netstatresult(some connection got stuck at ESTABLISHED, and Recv-Q was large), and the buffer was slowly processed.Way to reproduce:
in_forward.rb, which should be at/opt/td-agent/embedded/lib/ruby/gems/2.4.0/gems/fluentd-1.11.5/lib/fluent/pluginif install fluentd through td-agent, or in sourcefluentd/lib/fluent/plugin/in_forward.rb
Lines 259 to 266 in 24fe4cb
Add
log.infoat probably line 263, inparser.feed_each's block.buffer.rb, which should be in the same directory ofin_forward.rb, or in sourcefluentd/lib/fluent/plugin/buffer.rb
Lines 288 to 294 in 24fe4cb
and add
log.on_infoat probably line 292.out_file.rb, which should be in the same directory ofin_forward.rb, or in sourcefluentd/lib/fluent/plugin/out_file.rb
Lines 195 to 199 in 24fe4cb
and add code at probably line 199.
Start fluentd aggretator.
Start the first fluentd.
My local result.
Log of first fluentd.
log of aggretator.
According to the aggregator's log, I think the flush_thread getting slowly actually affect event_loop, and then cause
in_forwardworks abnormally. There is another reason that I think the above way is similar toout_elasticsearch. Inout_elasticsearch, There is also a loop to map events in chunk to bulk request body, and I guess it's a io-independent operation.I am new to ruby, and I was tring to confirm whether ruby's thread switch machenism cause this problem. I also wrote simple tests(including pure ruby thread and interaction between ruby and c extension) to verify, but I failed.
Beta Was this translation helpful? Give feedback.
All reactions